Great question. Here are the practical ways to clear or cancel all scheduled or late runs in Prefect 3.x. What do you want to do? - Stop future runs from being created: pause or clear the deployment’s schedules - Cancel or delete runs that already exist in SCHEDULED or LATE state: use a small Python script (bulk), or cancel individually via CLI Stop future scheduled runs - Pause a deployment’s schedules (can be resumed later):

prefect deployment schedule pause "<flow-name>/<deployment-name>"

- Remove all schedule definitions from a deployment:

prefect deployment schedule clear "<flow-name>/<deployment-name>" -y

Note: clearing schedules removes the schedule definitions and prevents future runs; it does not cancel runs that already exist. - CLI help: Deployment schedule CLI reference - Concepts: Schedules in Prefect 3 Cancel existing SCHEDULED or LATE runs (bulk via Python) There isn’t a CLI bulk-cancel, but this short async script will cancel all runs in SCHEDULED or LATE states. Run it in an environment with Prefect configured to your workspace.

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FilterSet, FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName
from prefect.states import Cancelled

# Optional: narrow to a specific deployment by name
# from prefect.client.schemas.filters import DeploymentFilterName, DeploymentFilter
# deployment_filter = DeploymentFilter(name=DeploymentFilterName(any_=["<flow-name>/<deployment-name>"]))

async def cancel_scheduled_and_late(limit=10000):
    async with get_client() as client:
        filters = FilterSet(
            flow_runs=FlowRunFilter(
                state=FlowRunFilterState(
                    name=FlowRunFilterStateName(any_=["SCHEDULED", "LATE"])
                ),
                # deployment=deployment_filter,  # uncomment to scope to a single deployment
            )
        )
        runs = await client.read_flow_runs(limit=limit, filters=filters)
        print(f"Found {len(runs)} runs to cancel")
        for fr in runs:
            await client.set_flow_run_state(fr.id, state=Cancelled(message="Bulk cancelled"))
        print("Done")

if __name__ == "__main__":
    asyncio.run(cancel_scheduled_and_late())

If you prefer to completely remove them from history, you can delete instead of cancel:

await client.delete_flow_run(fr.id)

Cancel a single run via CLI - List and then cancel by ID:

prefect flow-run ls --state LATE --limit 100
prefect flow-run cancel <FLOW_RUN_ID>

CLI help: - Flow run commands: run

prefect flow-run --help

(subcommands include

ls

,

cancel

,

delete

) Docs for reference - Cancelling workflows: Cancel workflows - Schedules: Schedules in Prefect 3 Quick check - Do you want this across your entire workspace or just specific deployments? - Do you want to stop future runs as well, or only cancel existing ones?