Got it — you want to make sure overdue “Late” runs don’t execute. In Prefect 3.x, the simplest and most reliable way is to auto-cancel Late runs with an Automation. Workers can pick up Late runs because they’re still state_type=SCHEDULED, so canceling them on entry to Late prevents execution. Do this: 1) Create an automation that cancels Late runs Save this as cancel-late.yaml:

name: Cancel Late Flow Runs
description: Automatically cancel flow runs that enter the Late state
enabled: true
trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Late
  threshold: 1
actions:
  - type: cancel-flow-run

Apply it:

prefect automation create --from-file cancel-late.yaml

Docs: - Creating automations - Event triggers and filters 2) (Optional) Scope it to a deployment or work pool For a specific deployment by name:

trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Late
  threshold: 1
  match_related:
    prefect.resource.role: deployment
    prefect.resource.name: my-deployment-name

For a specific work pool:

trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Late
  threshold: 1
  match_related:
    prefect.resource.role: work-pool
    prefect.resource.name: my-pool

3) Clean up any Late runs that already exist (one-time) Automations trigger on new events only. If you already have Late runs, cancel them in bulk with a short script:

import asyncio
from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName

async def main():
    async with get_client() as client:
        runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                state=FlowRunFilterState(
                    name=FlowRunFilterStateName(any_=["Late"])
                )
            ),
            limit=1000,
        )
        for r in runs:
            try:
                await client.cancel_flow_run(r.id)
                print(f"Cancelled {r.id} ({r.name})")
            except Exception as e:
                print(f"Failed to cancel {r.id}: {e}")

asyncio.run(main())

Helpful CLI for inspection: - List Late runs:

prefect flow-run ls --state LATE

- Cancel a single run:

prefect flow-run cancel <flow_run_id>

Tips to prevent Late runs in the first place: - Pause a work pool during maintenance:

prefect work-pool pause <pool>

and resume after. - Note: In Prefect 3.1+, schedules don’t “catch up” missed times by default; scheduling resumes from now forward. Late runs arise when a generated run misses its start time before it’s picked up — the automation above handles those. Would you like this automation scoped to specific deployments or to apply org-wide?