Great question. Assuming Prefect 3.x: timeouts are enforced per run and do not cascade across parent/child boundaries. Key behaviors - Flow timeouts are local to that flow run. Task timeouts are local to that task run. - Timeouts do not automatically cancel anything else. When a parent times out, Prefect does not cancel running tasks or subflows. - Retries get a fresh timeout window on each attempt. Scenarios - Parent timeout < Child timeout - The parent flow hits its timeout and fails with a TimedOut state. - The child flow continues running independently and will finish (or time out) on its own. - Child timeout < Parent timeout - The child flow hits its timeout and fails with a TimedOut state. - The parent flow receives that failed state as the child’s result and can continue or handle it. Tasks vs subflows - Task timeouts only affect that task; they don’t change the flow’s timeout. The flow just receives a failed task state. - If a flow times out, Prefect does not proactively cancel running tasks in that flow (no blanket cancel). Those tasks may continue running in the executor while the flow run is marked TimedOut. - Subflows are separate flow runs; a parent timing out does not cancel them. Minimal example

from prefect import flow
import time

@flow(timeout_seconds=5)
def parent():
    state = child(return_state=True)  # waits for child
    return state

@flow(timeout_seconds=30)
def child():
    time.sleep(20)
    return "done"

# Result:
# - parent: Failed (TimedOut) after ~5s
# - child: Continues and completes after ~20s

If you need the parent timing out to also stop children, you’ll need to explicitly cancel those child flow runs (e.g., via the UI/API). Let me know your desired behavior and I can suggest a pattern.

Absolutely — here are direct docs and code references that back up each claim. Docs (Prefect 3.x) - How to write and run a workflow (includes timeout usage) - Write and run a workflow - Flows concept (flow settings incl. timeout) - Flows concept - Tasks concept (task settings incl. timeout) - Tasks concept - Retries guide (retries are per-attempt; pairs with code references below for timeout+retry) - Retries guide - API refs - Flow decorator (shows

timeout_seconds

) - prefect.flows API - Task decorator (shows

timeout_seconds

) - prefect.tasks API - Timeout utilities used by the engines - prefect.utilities.timeout Source code (Prefect 3.x) - Flow timeouts are enforced locally in the flow engine via a timeout context; on timeout, the flow is marked TimedOut — no blanket cancellation is performed. - Flow run context and timeout handling - src/prefect/flow_engine.py - Look for: -

run_context(...)

which wraps execution in `timeout()`/`timeout_async()` -

handle_timeout(...)

which sets state to Failed(name="TimedOut") -

cancel_all_tasks(...)

exists, but note it is NOT called from

handle_timeout

- Task timeouts are enforced locally in the task engine and are retryable; each retry attempt gets a fresh timeout window. - Task timeout and retry interaction - src/prefect/task_engine.py - Look for: -

handle_timeout(...)

calling

handle_retry(...)

when a task times out - Cancellation propagation: the engine calls

cancel_all_tasks()

on external termination signals, not on timeouts. - External termination handling (where

cancel_all_tasks()

is used) - src/prefect/flow_engine.py - Look for a

try/except

catching

TerminationSignal

and then calling

cancel_all_tasks()

- Termination signal definition - src/prefect/exceptions.py (search for

class TerminationSignal

) - Task runner cancellation implementation (what

cancel_all()

actually does) - src/prefect/task_runners.py (search for

cancel_all

in thread/process task runners) - Timeout utilities used by the engines (local enforcement, not server-side) - src/prefect/utilities/timeout.py - Cancellation scopes used under the hood - src/prefect/_internal/concurrency/cancellation.py How these prove the behavior you asked about - Parent vs child (subflow) timeouts are independent: - Each flow run is wrapped in its own timeout context (see

run_context

in

flow_engine.py

), so the parent’s timeout does not propagate into the child’s execution. - When a parent flow times out: -

handle_timeout(...)

marks the parent as Failed(TimedOut). There’s no call to

cancel_all_tasks()

here, and there’s no logic to cancel subflow runs. Subflows continue to run independently. - When a child times out: - The child’s engine marks it Failed(TimedOut). The parent just receives that state and can continue logic accordingly. - Retries + timeouts (per-attempt):

- In

task_engine.py

,

handle_timeout(...)

invokes

handle_retry(...)

, confirming timeouts are retryable and each retry gets its own timeout window. If you want, I can paste the exact snippets from those files showing

handle_timeout

, the

TerminationSignal

catch, and the

cancel_all_tasks

usage.

Short answer That message is emitted when Prefect catches an async cancellation from the runtime/infrastructure (e.g., worker shutdown, pod/container termination, task group cancellation). Prefect converts that cancellation into a CRASHED state with the message “Execution was cancelled by the runtime environment.” Evidence in code - Message definition (async and sync) - src/prefect/states.py → function

exception_to_crashed_state

sets: - “Execution was cancelled by the runtime environment.” when

isinstance(exc, anyio.get_cancelled_exc_class())

- Link: src/prefect/states.py - src/prefect/_states.py → sync equivalent

exception_to_crashed_state_sync

- Link: src/prefect/_states.py - Where it’s used (flows) - Flow engines catch BaseException and call

handle_crash(...)

, which uses the above converter and sets a Crashed state - Sync flow engine: src/prefect/flow_engine.py (see try/except around execution;

handle_crash

path) - Async flow engine: same file, async section; uses

CancelScope(shield=True)

to ensure the crash state is sent to the API even during cancellation - Where it’s used (tasks) - Task engines do the same for BaseException: call

handle_crash(...)

which uses the same converter - src/prefect/task_engine.py - Test that proves it - Cancelling an anyio task group running a flow produces this exact message in the Crashed state: - tests/public/flows/test_flow_crashes.py →

test_anyio_cancellation_crashes_flow

- Link: tests/public/flows/test_flow_crashes.py When does it happen in practice? - Worker or infrastructure cancels your run’s async scope (e.g., Kubernetes pod eviction, Docker container stop, worker shutdown) - An anyio/asyncio cancellation (e.g.,

CancelScope.cancel()

or

asyncio.CancelledError

) propagates out of user code - Prefect catches that cancellation, treats it as an infrastructure-level interruption, and sets the run to Crashed with that message Related docs - States and “Crashed” definition (terminal, infrastructure issue) - States concept - Notes indicate CRASHED is typically due to infra issues like OOM, evicted pod, termination signals, etc. If you share which runtime you’re on (Kubernetes, Docker, ECS, etc.), I can point to the most common cancellation sources and how to prevent them.

Short answer: It depends what you mean by “parent flow gets run.” If the parent flow is simply executing normally (or even timing out), then no — child tasks will not get the “Execution was cancelled by the runtime environment.” message. That message appears when the runtime/infrastructure cancels execution (e.g., pod/container/worker shutdown), not from normal execution or a flow-level timeout. Two common cases with evidence: 1) Parent flow times out (flow timeout_seconds exceeded) - Behavior: Parent is marked Failed(TimedOut). Prefect does not blanket-cancel running tasks; they may continue. - Child tasks will not get the “cancelled by the runtime environment” message from a parent timeout. - Evidence:

handle_timeout()

sets a TimedOut state and does not call `cancel_all_tasks()`: - flow timeout handling: src/prefect/flow_engine.py (see the timeout try/except and

handle_timeout

) - no task cancellation on timeout: same file;

cancel_all_tasks()

is not called from

handle_timeout

2) Parent flow is cancelled by the runtime/infrastructure (e.g., worker shuts down, pod is terminated) - Behavior: An async cancellation (e.g.,

asyncio.CancelledError

) propagates. The engine converts this to a Crashed state with the message “Execution was cancelled by the runtime environment.” - Child tasks may also be cancelled if they share the same async cancellation scope (common for async tasks). Thread/process pool tasks may not be interrupted mid-flight, but pending ones can be cancelled. - Evidence for the message and when it’s emitted: - Message string and condition (anyio cancellation class) → “Execution was cancelled by the runtime environment.”: - async: src/prefect/states.py (

exception_to_crashed_state

) - sync: src/prefect/_states.py (

exception_to_crashed_state_sync

) - Flow engine crash path (catches BaseException outside user code and sets Crashed): - src/prefect/flow_engine.py (see try/except that calls `handle_crash(...)`; async uses

CancelScope(shield=True)

to ensure state is set) - Task engine crash path: - src/prefect/task_engine.py (BaseException →

handle_crash(...)

) - Test proving cancellation produces this exact message: - tests/public/flows/test_flow_crashes.py (

test_anyio_cancellation_crashes_flow

) - States docs: CRASHED is a terminal state for infrastructure interruptions: - States concept Extra nuance: A termination signal (e.g., SIGTERM) path explicitly calls

cancel_all_tasks()

to try to cancel in-flight tasks: - See the TerminationSignal catch and

cancel_all_tasks()

usage in the flow engine: - src/prefect/flow_engine.py (look for

except TerminationSignal

calling

cancel_all_tasks()

) - Thread/process task runners best-effort cancel pending futures; running threads/processes may continue until the process terminates: - src/prefect/task_runners.py (search for

cancel_all

) If you meant a different scenario (e.g., “when the parent flow is explicitly cancelled via API/UI”), say the word and I’ll tailor the behavior and code pointers for that path too.

That specific traceback comes from AnyIO/asyncio when a cancellation is triggered on a cancel scope. In Prefect, that typically means the runtime/infrastructure cancelled the run (worker shutdown, pod/container termination, task group cancellation, etc.). Prefect will convert that into a CRASHED state with a message like “Execution was cancelled by the runtime environment.” How to debug it effectively 1) Check the run’s final state and message - Look in the UI, or fetch the state history to see exactly what state was set and when. - Code to fetch the full state history:

from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.objects import StateType

async def show_states(flow_run_id: str):
    async with get_client() as client:
        states = await client.read_flow_run_states(UUID(flow_run_id))
        for s in states:
            print(s.timestamp, s.type.name, s.name, s.message)

- Evidence that Prefect generates the “Execution was cancelled by the runtime environment.” message on cancellation: - State conversion code: src/prefect/states.py and src/prefect/_states.py - The engines call this on BaseException cancellations: - Flow engine: src/prefect/flow_engine.py - Task engine: src/prefect/task_engine.py - Test proving cancellation produces that exact message: tests/public/flows/test_flow_crashes.py (see

test_anyio_cancellation_crashes_flow

) - States docs (CRASHED = infra interruption): States concept 2) Identify who/what initiated the cancellation - Use the Events API to find the actor (user, automation, worker) and the event sequence around the cancellation.

from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.events.filters import EventFilter, EventResourceFilter, EventIDFilter, EventNameFilter

async def show_cancellation_events(flow_run_id: str):
    async with get_client() as client:
        f = EventFilter(
            resource=EventResourceFilter(
                id=EventIDFilter(id=[f"prefect.flow-run.{flow_run_id}"])
            ),
            event=EventNameFilter(name=["prefect.flow-run-state-changed"]),
        )
        page = await client.read_events(filter=f, limit=100)
        for ev in page.events:
            print(ev.occurred, ev.event)
            print("resource:", ev.resource)
            print("payload:", ev.payload)
            for rel in ev.related or []:
                print("related:", rel)

- Look for related resources indicating a user, automation, or worker as the source. 3) Correlate with worker and infrastructure logs - Worker logs around the timestamp: look for messages indicating shutdown, stop signals, or job cancellation. - Infrastructure: - Kubernetes:

kubectl describe pod <pod>

(check Events for eviction, preemption, node drain), OOMKilled, terminationGracePeriod. - Docker:

docker logs

, container stop/kill events. - ECS/Batch/etc.: service stop reasons, scaling events. - Common real causes: scaling down workers, pod eviction/preemption, OOM kill, node drain, explicit cancel via UI/automation. 4) Turn on more verbosity to capture breadcrumbs - Prefect debug logging:

prefect config set logging.level=DEBUG
# or env var:
# PREFECT_LOGGING_LEVEL=DEBUG

- asyncio debug:

# In your environment
PYTHONASYNCIODEBUG=1

- Add breadcrumbs in your flow/task: ``` import anyio, asyncio, logging from prefect import flow, task logger = logging.getLogger(name) @task async def work(): try: logger.info("task starting")

await anyio.sleep(300) logger.info("task finished") except asyncio.CancelledError: logger.exception("task received CancelledError") raise @flow async def my_flow(): try: await work() except asyncio.CancelledError: logger.exception("flow received CancelledError") raise ``` 5) Check for parent run interactions - If this is a subflow or a task within a parent flow, see if the parent was cancelled or terminated around the same time (use the same state/event queries for the parent). - Note: A parent flow timing out does NOT automatically cancel children; but an infrastructure/worker cancellation can propagate to both. 6) Verify no automations or policies are cancelling runs - Review automations that might cancel “long-running” flows or react to other events. - Check concurrency settings; some orgs implement custom automations to cancel when limits are exceeded. What the error means technically - The “Cancelled by cancel scope …” string is from AnyIO; a cancel scope was triggered (e.g., worker shutdown, task group cancellation). Prefect detects this via

anyio.get_cancelled_exc_class()

and converts it into a Crashed state with the message “Execution was cancelled by the runtime environment.” - Evidence: states.py and test at test_flow_crashes.py If you can share which infrastructure you’re running on (K8s/Docker/ECS/etc.) and a flow run ID/timestamp window, I can suggest the most likely root causes and where to look next.

Thanks — since you’re running a served, auto-scheduled flow on ECS, that asyncio CancelledError is almost always the container/task being stopped or an async cancel scope being triggered by the runtime (deployment rollouts, service restarts, healthcheck failures, scaling down, spot interruptions, or explicit stop). In Prefect terms, that typically yields a CRASHED run with the message “Execution was cancelled by the runtime environment.” What to check first (with evidence/links) - Confirm the Prefect run state and message - The “cancelled by the runtime environment” message is set when an AnyIO/async cancellation is raised: - Code: src/prefect/states.py and src/prefect/_states.py (see where it checks

anyio.get_cancelled_exc_class()

and sets that exact message) - Engines convert this into a Crashed state: - Flow engine: flow_engine.py (BaseException →

handle_crash(...)

) - Task engine: task_engine.py (BaseException →

handle_crash(...)

) - Test proving cancellation yields that message: tests/public/flows/test_flow_crashes.py - State types and meaning (CRASHED = infra interruption): States concept - ECS service/task events and CloudWatch logs - In the ECS console, open the service and the specific task: - Check stoppedReason, lastStatus/desiredStatus, deployment events, and container exit codes - Common causes: rolling deployment replacing tasks, health check failures, memory/CPU pressure, ALB target health flapping, Fargate spot interruptions Prefect-side debugging steps 1) Inspect state history and messages for the run

from uuid import UUID
from prefect.client.orchestration import get_client

async def show_states(flow_run_id: str):
    async with get_client() as client:
        states = await client.read_flow_run_states(UUID(flow_run_id))
        for s in states:
            print(s.timestamp, s.type.name, s.name, s.message)

2) See who/what initiated the state change (events API)

from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.events.filters import EventFilter, EventResourceFilter, EventIDFilter, EventNameFilter

async def show_cancellation_events(flow_run_id: str):
    async with get_client() as client:
        f = EventFilter(
            resource=EventResourceFilter(id=EventIDFilter(id=[f"prefect.flow-run.{flow_run_id}"])),
            event=EventNameFilter(name=["prefect.flow-run-state-changed"]),
        )
        page = await client.read_events(filter=f, limit=100)
        for ev in page.events:
            print(ev.occurred, ev.event)
            print("resource:", ev.resource)
            print("payload:", ev.payload)
            for rel in ev.related or []:
                print("related:", rel)

3) Add targeted logging to catch cancellations in your code

import asyncio
from prefect import flow, task
from prefect import get_run_logger

@task
async def long_task():
    log = get_run_logger()
    try:
        <http://log.info|log.info>("task starting")
        await asyncio.sleep(600)
        <http://log.info|log.info>("task finished")
    except asyncio.CancelledError:
        log.exception("task received CancelledError")
        raise

@flow
async def served_flow():
    try:
        await long_task()
    except asyncio.CancelledError:
        get_run_logger().exception("flow received CancelledError")
        raise

ECS-specific checklist to stop the churn - Rolling deployments/updates - Confirm the service isn’t constantly redeploying (task definition changes, image tags, autoscaling events)

- If it is, those new deployments stop old tasks → CancelledError in the old tasks - Health checks (ALB or container) - If you’ve set a container health check for served flows, make sure the Prefect runner health endpoint is enabled and reachable: -

PREFECT_RUNNER_SERVER_ENABLE=true

, port 8080 exposed - Docs: Serve flows in a long‑lived container - Flapping health checks cause ECS to kill and replace tasks - Resource constraints - Check CloudWatch/container stats for CPU/memory; OOM or throttling often precede termination - For Fargate, also check ephemeral storage limits - Spot interruptions / service scaling down - If using Fargate Spot or aggressive scaling, tasks may be stopped; you’ll see this in the task stoppedReason - Shutdown grace period - ECS stopTimeout (task definition) controls how long the container has to shut down gracefully; increasing it can let Prefect finish state updates/logging during shutdown Relevant Prefect docs for your setup - ECS worker overview and deployment (even if you’re using serve, these pages cover ECS-specific patterns and troubleshooting) - ECS worker overview - Manual ECS worker deployment - Serve flows in long-lived containers (health endpoint, platform notes including ECS) - Serve flows in long‑lived Docker container - Worker concepts and healthchecks (useful for long-running services) - Workers concept - Worker healthchecks Why this produces asyncio CancelledError (evidence) - AnyIO cancellation is converted by Prefect into a Crashed state with message “Execution was cancelled by the runtime environment.” - Source: states.py, _states.py - Engines catch BaseException and call `handle_crash(...)`: - flow_engine.py, task_engine.py - Verified by tests: test_flow_crashes.py If you can share an example task ARN or flow run ID + timestamp window, I can help correlate the ECS stoppedReason with the Prefect run state/event timeline and suggest a concrete fix (e.g., health check config, deployment settings, or resource tuning).