Absolutely! In Prefect 3.x, runs that were scheduled but not picked up after their scheduled time are marked with the “Late” state (state_type=SCHEDULED). You can query those from the last week with the Prefect client inside a flow. Here’s a complete, ready-to-run flow that returns flow runs that were scheduled in the last week but did not run:

from datetime import datetime, timedelta, timezone
from typing import Sequence
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterExpectedStartTime,
    FlowRunFilterStateType,
    FlowRunFilterStateName,
    FlowRunFilterDeploymentId,
    FlowRunFilterWorkQueueName,
)
from prefect.client.schemas.objects import StateType


@flow
def find_missed_runs(
    days: int = 7,
    deployment_ids: Sequence[str] | None = None,
    work_queue_names: Sequence[str] | None = None,
    only_late: bool = True,  # if False, returns all scheduled states in the window
    limit: int = 1000,
) -> list[str]:
    now = datetime.now(timezone.utc)
    window_start = now - timedelta(days=days)

    # Build filters
    expected_time_filter = FlowRunFilterExpectedStartTime(after_=window_start, before_=now)

    # Option A: specifically “Late” runs (recommended for “missed”)
    state_filter = FlowRunFilterStateName(any_=["Late"]) if only_late else None

    # Option B: any scheduled runs in the window (includes Scheduled/Late/AwaitingRetry/etc.)
    state_type_filter = None if only_late else FlowRunFilterStateType(any_=[StateType.SCHEDULED])

    dep_filter = FlowRunFilterDeploymentId(any_=list(deployment_ids)) if deployment_ids else None
    wq_filter = FlowRunFilterWorkQueueName(any_=list(work_queue_names)) if work_queue_names else None

    filters = FlowRunFilter(
        expected_start_time=expected_time_filter,
        state_name=state_filter,
        state_type=state_type_filter,
        deployment_id=dep_filter,
        work_queue_name=wq_filter,
    )

    with get_client(sync_client=True) as client:
        runs = client.read_flow_runs(flow_run_filter=filters, limit=limit)

    # Log a quick summary
    for r in runs:
        print(
            f"{r.id} | {r.name} | expected_start={r.expected_start_time} | "
            f"state={r.state_name} | deployment_id={r.deployment_id}"
        )

    # Return the IDs (adjust to return the whole objects if you prefer)
    return [r.id for r in runs]


if __name__ == "__main__":
    find_missed_runs()

Notes: - Set

only_late=True

to capture just the missed runs. If you set it to

False

, you’ll get all scheduled runs in the time window (Scheduled, Late, AwaitingRetry, etc.). - You can scope to specific deployments (

deployment_ids=[...]

) or work queues (

work_queue_names=[...]

) if you don’t want to search everything. Docs for reference: - States in Prefect 3.x (what “Late” means): States - Python client usage: API Client - Filters API reference: Filters API Would you like this scoped to a particular deployment, flow, or work queue, or should it scan your entire workspace? Also, what would you like to do with the results—just list them, or requeue/reschedule them automatically?

Great question. Short answer: - If a run never started because the machine/worker that should pick it up was off, it will be “Late” (state type SCHEDULED) once its scheduled time has passed. - “Crashed” is used when the run started (or was being submitted) and something abnormal interrupted it. Prefect won’t automatically mark a run as Crashed just because a machine powered off mid-run unless you have heartbeat-based detection enabled via an Automation. What triggers “Crashed” in Prefect 3.x: - Infrastructure/submit failures while moving from Scheduled → Pending → Running - Example: bad image, missing credentials, cluster not reachable. The worker can set the run to Crashed during submission. - Fatal exceptions outside your flow code while executing - Example: KeyboardInterrupt, termination signals, SystemExit. The engine converts these to a Crashed state. - Heartbeat-based zombie detection (recommended) - When a run is truly “stuck” (e.g., the process died or the machine was powered off mid-run), Prefect won’t see a final state. With a “no heartbeat for N seconds/minutes” Automation, you can automatically transition such runs to Crashed. Runners emit heartbeats by default; the Automation is what acts on missing heartbeats. How this applies if the computer was turned off: - Before the run is picked up: it will stay Scheduled, then become Late after the scheduled time if no worker picks it up. - While the worker is submitting infrastructure: the run is typically Pending; failures here commonly mark it Crashed. - After the run has started: it will remain Running if the machine dies abruptly. With a heartbeat-based Automation in place, Prefect will mark it Crashed after your configured “no heartbeat” window. Helpful docs: - States: States in Prefect 3.x - Detecting/handling zombie runs: Detect zombie flows - Worker health and related considerations: Worker health checks If you’d like, I can help you: - Set up the “no heartbeat → mark Crashed” Automation in your workspace - Tune the heartbeat window - Add alerting when runs become Late or Crashed Do you want to crash-mark zombies after a few minutes without heartbeats, and should we scope this to specific work pools or deployments?

Short answer: you don’t need to kick it yourself. When your worker comes back online, it will automatically pick up eligible “Late” runs and move them to Pending → Running, as long as the worker is polling the correct work pool/queue and there aren’t blocking constraints (e.g., concurrency limits). What to expect: - Late = scheduled time has passed but the run hasn’t been picked up yet. When a worker resumes, it will submit Late runs it’s responsible for. - You’ll only need to intervene if: - The run was canceled/paused. - The deployment or its schedules were paused/disabled. - The worker is polling the wrong work pool/queue or isn’t healthy. - Concurrency limits or tags prevent submission. If you want to skip “catching up” on Late runs: - Add an Automation to cancel or reschedule Late runs after some delay (e.g., “when a flow run enters Late for > X minutes → CancelFlowRun”). - Add a guard at the start of your flow to exit if it’s too old:

from datetime import datetime, timezone, timedelta
from prefect import flow
from prefect.runtime import flow_run

@flow
def my_flow(max_lateness_minutes: int = 30):
    sched = flow_run.scheduled_start_time()
    if sched and datetime.now(timezone.utc) - sched > timedelta(minutes=max_lateness_minutes):
        print("Skipping run because it’s too late")
        return
    # normal work here

References: - States: States in Prefect 3.x - Workers/polling: Workers - Automations: Creating automations

Exactly. Late-run alerts are mainly belt-and-suspenders—useful for visibility and to enforce your policy—but workers will automatically pick up Late runs once they’re back online. Good automations to consider: - Notify when a run has been Late for > X minutes - Auto-cancel or reschedule runs that are Late beyond your SLA - Alert (or mark Crashed) when a Running run has no heartbeat for N minutes (“zombie” detection) - Alert when a run is Pending for too long (infra/submit issues) - Alert when there are 0 healthy workers or a large queue backlog in a work pool If you don’t want to “catch up” on old runs, pair the alerts with: - A flow guard to skip if too old, or - An automation that cancels Late runs past a threshold References: - Automations: Creating automations - Zombie detection: Detect zombie flows - Workers: Workers

You don’t have to choose permanently—I showed A/B to illustrate two common filters. You can make it configurable and even include multiple state names. If your goal is “missed runs,” use Late. If you want all scheduled-but-not-started runs in the window, use the broader SCHEDULED type. Or pass a list of specific scheduled state names. Here’s a small tweak that supports all three patterns:

from datetime import datetime, timedelta, timezone
from typing import Sequence
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterExpectedStartTime,
    FlowRunFilterStateType,
    FlowRunFilterStateName,
)
from prefect.client.schemas.objects import StateType

@flow
def find_missed_runs(
    days: int = 7,
    state_names: Sequence[str] | None = ("Late",),  # None => all SCHEDULED states
    limit: int = 1000,
):
    now = datetime.now(timezone.utc)
    window_start = now - timedelta(days=days)

    expected_time = FlowRunFilterExpectedStartTime(after_=window_start, before_=now)

    # If state_names is provided, filter by those names; otherwise include all SCHEDULED states
    state_name_filter = FlowRunFilterStateName(any_=list(state_names)) if state_names else None
    state_type_filter = None if state_names else FlowRunFilterStateType(any_=[StateType.SCHEDULED])

    flt = FlowRunFilter(
        expected_start_time=expected_time,
        state_name=state_name_filter,
        state_type=state_type_filter,
    )

    with get_client(sync_client=True) as client:
        runs = client.read_flow_runs(flow_run_filter=flt, limit=limit)

    for r in runs:
        print(f"{r.id} | {r.name} | expected={r.expected_start_time} | state={r.state_name}")

    return runs

Examples: - Only missed runs: keep default

state_names=("Late",)

- All scheduled states in the window:

state_names=None

- A custom subset (e.g., Late and AwaitingRetry):

state_names=("Late", "AwaitingRetry")

State reference: States in Prefect 3.x Which behavior do you prefer: only Late, or a broader view of all scheduled-but-not-started runs?