Great question — here’s a ready-to-run Prefect 3.x flow that finds all “failed or unfinished” flow runs older than a specified time window. Assumptions (please confirm/adjust): - “Failed” = FAILED, CRASHED, CANCELLED - “Unfinished”: - Not started yet: SCHEDULED or PENDING where expected_start_time is older than the cutoff - In progress: RUNNING, PAUSED, CANCELLING where start_time is older than the cutoff - “Older than” is measured against: - expected_start_time for SCHEDULED/PENDING - start_time for RUNNING/PAUSED/CANCELLING - end_time for FAILED/CRASHED/CANCELLED Code (sync flow using the Prefect client): ``` from datetime import datetime, timedelta, timezone from typing import List, Dict, Any from prefect import flow, get_run_logger from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType, FlowRunFilterStartTime, FlowRunFilterExpectedStartTime, FlowRunFilterEndTime, ) from prefect.client.schemas.objects import StateType def _page_flow_runs(client, flow_run_filter, page_size: int = 200): """Utility to page through read_flow_runs results.""" results = [] offset = 0 while True: runs = client.read_flow_runs( flow_run_filter=flow_run_filter, limit=page_size, offset=offset, ) if not runs: break results.extend(runs) offset += len(runs) if len(runs) < page_size: break return results @flow(name="find-failed-or-unfinished-runs") def find_failed_or_unfinished_runs(age_hours: int = 24) -> List[Dict[str, Any]]: """ Returns flow runs that are: - Failed (FAILED, CRASHED, CANCELLED) with end_time older than cutoff - Unfinished and in-progress (RUNNING, PAUSED, CANCELLING) with start_time older than cutoff - Unfinished and not-started (SCHEDULED, PENDING) with expected_start_time older than cutoff """ logger = get_run_logger() now = datetime.now(timezone.utc) cutoff = now - timedelta(hours=age_hours) failed_types = [StateType.FAILED, StateType.CRASHED, StateType.CANCELLED] in_progress_types = [StateType.RUNNING, StateType.PAUSED, StateType.CANCELLING] not_started_types = [StateType.SCHEDULED, StateType.PENDING] with get_client() as client: # Failed runs: end_time before cutoff failed_filter = FlowRunFilter( state=FlowRunFilterState( type=FlowRunFilterStateType(any_=failed_types) ), end_time=FlowRunFilterEndTime(before_=cutoff), ) failed_runs = _page_flow_runs(client, failed_filter) # In-progress unfinished runs: start_time before cutoff in_progress_filter = FlowRunFilter( state=FlowRunFilterState( type=FlowRunFilterStateType(any_=in_progress_types) ), start_time=FlowRunFilterStartTime(before_=cutoff), ) in_progress_runs = _page_flow_runs(client, in_progress_filter) # Not-started unfinished runs: expected_start_time before cutoff not_started_filter = FlowRunFilter( state=FlowRunFilterState( type=FlowRunFilterStateType(any_=not_started_types) ), expected_start_time=FlowRunFilterExpectedStartTime(before_=cutoff), ) not_started_runs = _page_flow_runs(client, not_started_filter) # Merge and deduplicate by id runs_by_id = {} for r in failed_runs + in_progress_runs + not_started_runs: runs_by_id[r.id] = r # Build a simple serializable summary result = [] for r in runs_by_id.values(): result.append( { "id": str(r.id), "name": r.name, "state_type": getattr(r.state_type, "value", str(r.state_type)),

"state_name": getattr(r.state_name, "value", str(r.state_name)) if hasattr(r, "state_name") else None, "start_time": getattr(r, "start_time", None), "end_time": getattr(r, "end_time", None), "expected_start_time": getattr(r, "expected_start_time", None), "deployment_id": str(getattr(r, "deployment_id", "")) if getattr(r, "deployment_id", None) else None, } ) logger.info( f"Found {len(result)} runs older than {age_hours}h " f"(failed: {len(failed_runs)}, in_progress: {len(in_progress_runs)}, not_started: {len(not_started_runs)})" ) return result if name == "__main__": # Example local run runs = find_failed_or_unfinished_runs(age_hours=24) print(f"Total runs found: {len(runs)}") for r in runs[10] print(r) ``` Notes: - This uses your active profile/workspace. Ensure your PREFECT_API_URL and, if using Cloud, PREFECT_API_KEY are set. - If you’d prefer a different definition of “unfinished” or want to measure age differently (e.g., created time), I can adjust the filters accordingly. - You can turn this into a deployment and schedule it to run periodically with

prefect deploy

(let me know your work pool type and I’ll share the exact command and yaml). Would you like me to: - Include/exclude certain states (e.g., treat CANCELLED as “failed” or not)? - Use a different time basis (e.g., age since created vs start/end/expected start)? - Add filters by flow name, deployment, or tags?

Got it — here’s an updated Prefect 3.x flow that: - Filters by tag (matches any of the provided tags) - Uses expected_start_time as the sole time basis

from datetime import datetime, timedelta, timezone
from typing import List, Dict, Any, Optional

from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateType,
    FlowRunFilterExpectedStartTime,
    FlowRunFilterTags,
)
from prefect.client.schemas.objects import StateType


def _page_flow_runs(client, flow_run_filter, page_size: int = 200):
    """Utility to page through read_flow_runs results."""
    results = []
    offset = 0
    while True:
        runs = client.read_flow_runs(
            flow_run_filter=flow_run_filter,
            limit=page_size,
            offset=offset,
        )
        if not runs:
            break
        results.extend(runs)
        offset += len(runs)
        if len(runs) < page_size:
            break
    return results


@flow(name="find-failed-or-unfinished-runs-by-tag-expected-start")
def find_failed_or_unfinished_runs(
    age_hours: int = 24,
    tags: Optional[List[str]] = None,  # match runs having ANY of these tags
) -> List[Dict[str, Any]]:
    """
    Returns flow runs that are:
      - Failed (FAILED, CRASHED, CANCELLED), or
      - Unfinished (SCHEDULED, PENDING, RUNNING, PAUSED, CANCELLING)
    And whose expected_start_time is older than the cutoff (now - age_hours).
    If tags are provided, only runs containing ANY of the tags are returned.
    """
    logger = get_run_logger()
    now = datetime.now(timezone.utc)
    cutoff = now - timedelta(hours=age_hours)

    # Failed + unfinished states
    target_states = [
        StateType.FAILED,
        StateType.CRASHED,
        StateType.CANCELLED,
        StateType.SCHEDULED,
        StateType.PENDING,
        StateType.RUNNING,
        StateType.PAUSED,
        StateType.CANCELLING,
    ]

    with get_client() as client:
        flow_run_filter = FlowRunFilter(
            state=FlowRunFilterState(
                type=FlowRunFilterStateType(any_=target_states)
            ),
            expected_start_time=FlowRunFilterExpectedStartTime(before_=cutoff),
            tags=FlowRunFilterTags(any_=tags) if tags else None,
        )

        runs = _page_flow_runs(client, flow_run_filter)

    # Build a simple serializable summary
    result = []
    for r in runs:
        result.append(
            {
                "id": str(r.id),
                "name": r.name,
                "state_type": getattr(r.state_type, "value", str(r.state_type)),
                "start_time": getattr(r, "start_time", None),
                "end_time": getattr(r, "end_time", None),
                "expected_start_time": getattr(r, "expected_start_time", None),
                "deployment_id": str(getattr(r, "deployment_id", "")) if getattr(r, "deployment_id", None) else None,
                "tags": getattr(r, "tags", []),
            }
        )

    <http://logger.info|logger.info>(
        f"Found {len(result)} runs older than {age_hours}h by expected_start_time"
        + (f" with tags any of {tags}" if tags else "")
    )
    return result


if __name__ == "__main__":
    # Example local run
    runs = find_failed_or_unfinished_runs(age_hours=24, tags=["priority:high", "team:data"])
    print(f"Total runs found: {len(runs)}")
    for r in runs[:10]:
        print(r)

Notes: - Tags filter uses “any” semantics. If you need “all tags must match,” I can switch to

FlowRunFilterTags(all_=...)

. - Runs without an expected_start_time will not be returned. If you want a fallback (e.g., to start_time), say the word and I’ll add it.

You’ve got two good sources for failure details in Prefect 3.x: - The flow run’s state: quick error summary via

state.message

, and a full exception/traceback if the result is available - The run’s logs: reliable way to reconstruct the traceback even when the exception object isn’t persisted Below is a drop-in helper that fetches both. It prefers the real exception/traceback from the state; if not available, it falls back to parsing ERROR logs for a traceback.

import logging
import traceback
from typing import Optional, Dict, Any, Tuple, List
from uuid import UUID

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    LogFilter, LogFilterFlowRunId, LogFilterLevel
)
from prefect.client.schemas.sorting import LogSort
from prefect.client.schemas.objects import StateType
from prefect.states import get_state_exception  # use aget_state_exception in async code


def _extract_traceback_from_logs(messages: List[str]) -> str:
    # Try to capture from the canonical traceback start marker
    tb_start = "Traceback (most recent call last):"
    captured: List[str] = []
    capturing = False
    for m in messages:
        if tb_start in m:
            capturing = True
        if capturing:
            captured.append(m)
    # Fallback: if we never saw the canonical marker, just join all ERROR messages
    return "\n".join(captured) if captured else "\n".join(messages)


def get_failure_details_for_run(flow_run_id: UUID) -> Dict[str, Any]:
    """
    Returns:
      {
        'flow_run_id': str,
        'state_type': 'failed'|'crashed'|'cancelled'|...,
        'error_message': Optional[str],
        'exception_type': Optional[str],
        'traceback': Optional[str],   # full traceback if available
        'source': 'state'|'logs'|'none'
      }
    """
    with get_client() as client:
        run = client.read_flow_run(flow_run_id)

        if not run or not run.state:
            return {
                "flow_run_id": str(flow_run_id),
                "state_type": None,
                "error_message": None,
                "exception_type": None,
                "traceback": None,
                "source": "none",
            }

        stype = run.state.type
        msg = run.state.message

        # Try to load the real exception from state data first (best fidelity)
        # This may not be available if results/exceptions were not persisted.
        try:
            exc = get_state_exception(run.state)
            if exc is not None:
                tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
                return {
                    "flow_run_id": str(flow_run_id),
                    "state_type": stype.value if hasattr(stype, "value") else str(stype),
                    "error_message": msg,
                    "exception_type": type(exc).__name__,
                    "traceback": tb,
                    "source": "state",
                }
        except Exception:
            pass  # Fall back to logs

        # Fallback: pull ERROR+ logs and reconstruct a traceback
        log_filter = LogFilter(
            flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]),
            level=LogFilterLevel(ge_=logging.ERROR),
        )
        logs = client.read_logs(
            log_filter=log_filter,
            limit=1000,
            offset=0,
            sort=LogSort.TIMESTAMP_ASC,
        )
        error_messages = [l.message or "" for l in logs]
        tb_from_logs = _extract_traceback_from_logs(error_messages)

        return {
            "flow_run_id": str(flow_run_id),
            "state_type": stype.value if hasattr(stype, "value") else str(stype),
            "error_message": msg,  # concise summary from state
            "exception_type": None,
            "traceback": tb_from_logs or None,
            "source": "logs" if tb_from_logs else "none",
        }

How to use it with your finder flow results:

- If you already have a list of flow_run_ids, just loop over them and call

get_failure_details_for_run(run_id)

. - If you’re enriching inside your existing flow, only call it for runs whose state_type is FAILED/CRASHED (CANCELLED generally won’t have a stack). Example enrichment loop:

from prefect.client.schemas.objects import StateType

failed_details = []
for r in runs:  # runs from your finder flow
    if r.get("state_type") in {StateType.FAILED.value, StateType.CRASHED.value}:
        details = get_failure_details_for_run(UUID(r["id"]))
        failed_details.append(details)

# failed_details now contains error_message + traceback per failed run

Notes: -

state.message

is the most reliable quick summary of the error. -

get_state_exception

returns the actual exception with traceback when the state’s result is retrievable; otherwise, use logs. - If you prefer async usage, use

async with get_client()

and `await client.read_flow_run(...)`/`await client.read_logs(...)`, plus

aget_state_exception

.