Short answer: you can’t. In Prefect 3.x,

prefect.deployments.run_deployment

does not accept an

empirical_policy

argument. Retries for a subflow started via

run_deployment

must be configured on the called flow/deployment or applied to the created flow run after the fact. Your options: - Preferable: set retries on the called flow

from prefect import flow

  @flow(retries=3, retry_delay_seconds=30)
  def child_flow(...):
      ...

This will apply no matter how the deployment is triggered (including as a subflow on ECS). - Programmatically override right after creating the flow run

from prefect.deployments import run_deployment
  from prefect.client.orchestration import get_client
  from prefect.client.schemas.objects import FlowRunPolicy

  # Create the subflow run (return immediately so we can update it)
  flow_run = await run_deployment(
      name=flow_name,
      parameters=kwargs,
      as_subflow=True,
      job_variables=job_variables,
      flow_run_name=flow_run_name,
      idempotency_key=idempotency_key,
      timeout=0,
  )

  # Set/override retry policy on the flow run
  policy = FlowRunPolicy(max_retries=3, retry_delay_seconds=30, retry_type="reschedule")
  async with get_client() as client:
      await client.update_flow_run(flow_run.id, empirical_policy=policy)

  # now wait for completion if you like
  state = await wait_for_state(flow_run, raise_exception_on_failure=raise_exception_on_failure)

Tip: using

timeout=0

or scheduling the run a few seconds in the future can reduce race conditions so the policy is set before the first attempt. For ECS,

retry_type="reschedule"

is typically what you want so each retry is a new scheduled attempt picked up by a worker. - As a separate pattern: use an Automation to “retry” by running the deployment again on failure. This is event-driven and outside the flow engine’s in-process retry loop. References: - Flow retries how-to: Retries guide - Retrying flow runs (manually/automation): Retry flow runs - API reference for `run_deployment`: run_deployment API - FlowRunPolicy schema: FlowRunPolicy

@Marvin so what happens if I set the timeout to 5 seconds, for example

If you set timeout=5 on run_deployment: - Prefect will create the flow run immediately, then poll for up to 5 seconds for it to reach a final state. - If the flow finishes within 5 seconds, run_deployment returns the FlowRun in its final state (e.g., Completed/Failed). - If it’s still Pending/Running after 5 seconds, run_deployment returns the FlowRun as-is with its current non-final state. Nothing is cancelled; the run keeps going on the worker. - poll_interval controls how often Prefect checks during those 5 seconds (default 5s). With a 5s timeout and default poll_interval, you’ll usually get 1 poll. Common patterns: - timeout=0: return immediately with the created FlowRun (usually Scheduled). Useful if you want to update the empirical_policy right after creation via client.update_flow_run before the first attempt starts. - timeout=None (default): block until the run finishes, polling every poll_interval seconds. This behavior is independent of as_subflow; that flag only controls parent/child linkage, not timeout/polling.

Short answer: you don’t set

retry_type

on `@flow`—and if you leave it “blank,” its default is

None

. Details (Prefect 3.x): -

@flow

does not accept a

retry_type

argument; you configure

retries

,

retry_delay_seconds

, and

retry_jitter_factor

instead. -

retry_type

is an internal flow-run field. Its default is

None

when the run is not being retried. - When Prefect automatically retries a failed flow (because

retries

> 0), orchestration sets

retry_type

to

in_process

. - If you manually rerun/reschedule a terminal flow run from the UI/API, orchestration sets

retry_type

to

reschedule

. Links (source): - FlowRunPolicy schema showing

retry_type

default `None`: https://github.com/PrefectHQ/prefect/blob/75a10ad4aab57304d21bf261f2eb174a137ff56e/src/prefect/server/schemas/core.py#L130-L132 - Orchestration setting

retry_type="in_process"

on automatic retries: https://github.com/PrefectHQ/prefect/blob/75a10ad4aab57304d21bf261f2eb174a137ff56e/src/prefect/server/orchestration/core_policy.py#L1098-L1105 - Orchestration setting

retry_type="reschedule"

on manual reruns: https://github.com/PrefectHQ/prefect/blob/75a10ad4aab57304d21bf261f2eb174a137ff56e/src/prefect/server/orchestration/core_policy.py#L1618-L1625 FYI: tasks don’t have a

retry_type

field—only flows track this.

A couple things to clarify first, because “timeout=0” can mean two different knobs: - If you mean `@flow(timeout_seconds=0)`: in Prefect 3.x,

0

is treated as “no timeout” (it becomes

None

internally), so it won’t force a stop. It won’t cause a Late state either. - If you mean `run_deployment(..., timeout=0, as_subflow=True)`:

timeout=0

just makes the parent return immediately after creating the subflow run; it does not affect scheduling or execution. It’ll rapidly queue a lot of subflow runs. What “Late” means here: - “Late” is a Scheduled state that the server sets when a run has not started by its scheduled time plus a small grace period (15s by default). It’s not a terminal/hanging state; it’s “scheduled but hasn’t started yet.” Code reference: https://github.com/PrefectHQ/prefect/blob/75a10ad4aab57304d21bf261f2eb174a137ff56e/src/prefect/server/orchestration/core_policy.py#L1704-L1723 Why you’re likely seeing “Late” with concurrent subflows on ECS: - With

timeout=0

on

run_deployment

, your parent flow will enqueue many subflow runs very quickly. - If your ECS work pool/cluster doesn’t have enough capacity (or the worker isn’t polling the right pool/queue), those runs won’t start promptly and will flip to Late after the grace period. - ECS startup overhead (registering task definitions, pulling images, capacity constraints) can add seconds to startup; if there’s a backlog, runs can remain Late until capacity frees up. Checks to run: - Verify the worker is online and polling the correct pool/queue:

prefect work-pool inspect "<your-ecs-pool>"

Look for “paused: false”, current concurrency limit, and assigned queues. - Confirm there’s an ECS worker running and pointed at that pool:

prefect worker start -p "<your-ecs-pool>"

- Check deployment settings for the subflows: they must target the ECS work pool you expect (and the right work queue). - Verify ECS capacity (CPU/memory), task definition registration logs, VPC/subnet/security groups. If capacity is tight, scale the cluster or reduce each task’s CPU/memory request. Options to mitigate: - Scale up ECS capacity and/or run more ECS workers. - Throttle how many subflows you launch at once (e.g., batch calls to

run_deployment

and wait for some to start/finish before launching more). - If you just don’t want to see “Late” for short startup delays, increase the late-run threshold on the server:

export PREFECT_SERVER_SERVICES_LATE_RUNS_AFTER_SECONDS=30

(Default is 15 seconds.) - If you intended an actual execution timeout for the subflow, set a positive

timeout_seconds

on the subflow’s flow decorator (e.g.,

@flow(timeout_seconds=300)

). Don’t use `0`—that disables timeouts. If runs remain stuck in Late and never transition to Running, that usually means the worker isn’t able to start them. Check the ECS worker logs for task creation/registration retries and ensure the pool/queue names match. Happy to dig in further—could you share: - How you’re launching subflows (a small snippet), especially whether you’re using

run_deployment(..., as_subflow=True, timeout=0)

. - The ECS work pool name and

prefect work-pool inspect

output. - One affected flow run ID and the ECS worker logs around the time it should have started.

Got it — you’re using run_deployment(..., timeout=0, as_subflow=True) and want two things: 1) wait up to a small window (e.g., 5s) for the subflow to finish, and 2) update the FlowRunPolicy (e.g., retries/retry_delay) for that subflow. What timeout does here - The run_deployment timeout only controls how long the parent waits for the child run to reach a final state before returning. It doesn’t influence execution or scheduling. - timeout=0 means “return immediately after creating the subflow run.” - So set timeout=5 to wait up to 5 seconds; if the subflow doesn’t complete in that window, run_deployment will return the latest state (likely Scheduled or Running) and the subflow will continue in the background. About FlowRunPolicy - In Prefect 3.x, FlowRunPolicy supports user-settable fields like retries and retry_delay. - retry_type is orchestration-managed only; you can’t set it directly. It will be set to "in_process" or "reschedule" by the server during retries as appropriate. - You can update a flow run’s empirical_policy with retries/retry_delay, either at creation (not exposed by run_deployment) or right after creation using the client’s update_flow_run. Pattern you can use with run_deployment as a subflow - Call run_deployment(..., as_subflow=True, timeout=5) to get a FlowRun back quickly. - Immediately update that run’s policy via the client to set retries/retry_delay. - This does not block scheduling; it just tweaks retry behavior for that subflow. Example

from prefect import flow, get_client
from prefect.deployments import run_deployment

@flow
def parent():
    # Launch child as a subflow and wait up to 5s for completion
    child_run = run_deployment(
        name="my-child-flow/my-child-deployment",
        parameters={"x": 1},
        as_subflow=True,
        timeout=5,           # wait up to 5 seconds
        poll_interval=1,     # optional, faster polling
        # job_variables=...  # if you need ECS-specific overrides
    )

    # Update the subflow’s empirical policy (retries/retry_delay)
    # Note: retry_type cannot be set by users.
    with get_client() as client:
        client.update_flow_run(
            flow_run_id=child_run.id,
            empirical_policy={
                "retries": 2,         # number of retries
                "retry_delay": 30,    # seconds between retries
            },
        )

    return child_run.id

Notes and caveats - If the child finishes within 5s, great — you’ll get its final state back. If not, you still get the FlowRun object; it continues running on ECS. - Updating empirical_policy right after creation is safe; Prefect will respect it for subsequent retries. - You cannot set retry_type directly; it’s set by orchestration depending on whether it’s an automatic retry (in_process) or a reschedule/reenqueue (reschedule). If you’d prefer to ensure the policy is applied before the child starts running, we can switch to using the client to create the run and set the policy up front, then return to the parent — but run_deployment’s convenience function doesn’t expose empirical_policy at creation time. If you want that route, I can share a snippet that mirrors run_deployment behavior manually and sets empirical_policy during creation.

Here’s an alternative that mirrors run_deployment but lets you set empirical_policy at creation time by using the client directly. Notes: - create_flow_run_from_deployment does not expose empirical_policy as a parameter, so we’ll set it right after creation with update_flow_run. - retry_type cannot be set by users; orchestration manages it. - timeout=5 behavior from run_deployment is replicated by polling the created run for up to 5 seconds. Async version

import anyio
from prefect import get_client
from prefect.client.schemas.objects import FlowRunPolicy
from uuid import UUID

async def launch_subflow_with_policy(
    deployment_id: UUID,
    *,
    parameters: dict | None = None,
    tags: list[str] | None = None,
    job_variables: dict | None = None,
    name: str | None = None,
    # empirical policy you want to enforce
    retries: int | None = None,
    retry_delay: int | None = None,  # seconds
    # emulate run_deployment timeout/polling behavior
    timeout: float | None = 5,
    poll_interval: float = 1.0,
):
    async with get_client() as client:
        # 1) Create the flow run from the deployment
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters=parameters,
            tags=tags,
            name=name,
            job_variables=job_variables,
            # you can pass work_queue_name if you need to target a specific queue
        )

        # 2) Immediately update the run’s empirical policy (retries/retry_delay)
        #    retry_type is not settable by users.
        if retries is not None or retry_delay is not None:
            await client.update_flow_run(
                flow_run_id=flow_run.id,
                empirical_policy=FlowRunPolicy(
                    retries=retries,
                    retry_delay=retry_delay,
                ),
            )

        # 3) Optional: poll like run_deployment(timeout=...)
        if timeout == 0:
            return flow_run

        if timeout is None:
            # poll indefinitely until final state
            while True:
                fr = await client.read_flow_run(flow_run.id)
                if fr.state and fr.state.is_final():
                    return fr
                await anyio.sleep(poll_interval)
        else:
            # poll up to timeout seconds
            with anyio.move_on_after(timeout):
                while True:
                    fr = await client.read_flow_run(flow_run.id)
                    if fr.state and fr.state.is_final():
                        return fr
                    await anyio.sleep(poll_interval)

            # timed out waiting; return latest snapshot
            return await client.read_flow_run(flow_run.id)

Sync version ``` import time from prefect import get_client from prefect.client.schemas.objects import FlowRunPolicy from uuid import UUID def launch_subflow_with_policy_sync( deployment_id: UUID, *, parameters: dict | None = None, tags: list[str] | None = None, job_variables: dict | None = None, name: str | None = None, retries: int | None = None, retry_delay: int | None = None, # seconds timeout: float | None = 5, poll_interval: float = 1.0, ): with get_client(sync_client=True) as client: flow_run = client.create_flow_run_from_deployment( deployment_id=deployment_id, parameters=parameters, tags=tags, name=name, job_variables=job_variables, ) if retries is not None or retry_delay is not None: client.update_flow_run( flow_run_id=flow_run.id, empirical_policy=FlowRunPolicy( retries=retries, retry_delay=retry_delay, ), ) if timeout == 0: return flow_run start = time.monotonic() while True: fr = client.read_flow_run(flow_run.id)

if fr.state and fr.state.is_final(): return fr if timeout is not None and (time.monotonic() - start) >= timeout: return fr time.sleep(poll_interval) ``` Why this works: - You still get the quick-return behavior (timeout=0 or small timeout) like run_deployment. - You can immediately apply retries/retry_delay on the subflow run via update_flow_run. - retry_type remains orchestration-managed and will be set automatically during retries. If you want to set policy before the run is ever picked up by a worker, this pattern does so as early as possible after creation. If you need to guarantee the update lands before scheduling begins, consider setting a scheduled_time slightly in the future when creating the run, apply the policy, then let it start; I can share that variant too.

You’re right — my prior snippet didn’t link as a subflow. Here are two working patterns that DO create a real subflow (i.e., with a parent task in the current flow) and let you control empirical_policy. Background links: - How subflow linking works in run_deployment (creates a parent task then passes parent_task_run_id): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/deployments/flow_runs.py - create_flow_run_from_deployment accepts parent_task_run_id: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/client/orchestration/_deployments/client.py#L653-L730 - DeploymentFlowRunCreate includes empirical_policy: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/client/schemas/actions.py#L525-L589 Option A — Official client methods, then update policy immediately after creation - Creates the parent placeholder task run (so it’s a real subflow), creates the child flow run, then updates its empirical_policy. You can still apply timeout-style polling like run_deployment. ``` import time import uuid from prefect import get_client from prefect.context import FlowRunContext, TaskRunContext from prefect.tasks import Task from prefect.states import Pending from prefect.client.schemas.objects import FlowRunPolicy # Optional: collect upstream deps for UI graph from prefect.engine import collect_task_run_inputs_sync def launch_subflow_as_subflow_with_policy( *, deployment_id, # UUID (recommended) or resolve via read_deployment_by_name parameters: dict | None = None, tags: list[str] | None = None, job_variables: dict | None = None, name: str | None = None, work_queue_name: str | None = None, # policy to apply retries: int | None = None, retry_delay: int | None = None, # seconds # emulate run_deployment polling timeout: float | None = 5, poll_interval: float = 1.0, ): parameters = parameters or {} # 1) Ensure we are in a flow/task context so this can be linked as a subflow flow_ctx = FlowRunContext.get() task_ctx = TaskRunContext.get() if not (flow_ctx or task_ctx): raise RuntimeError("Must be called from within a flow or task to create a subflow.") # 2) Create a synthetic parent task run (placeholder in the parent’s graph) flow_run_id = flow_ctx.flow_run.id if flow_ctx else task_ctx.task_run.flow_run_id dummy_task = Task(name=name or "subflow", fn=lambda: None, version=None) dynamic_key = f"subflow-{uuid.uuid4().hex[:8]}" task_inputs = {k: collect_task_run_inputs_sync(v) for k, v in parameters.items()} if task_ctx and (not flow_ctx or task_ctx.task_run.flow_run_id == getattr(flow_ctx.flow_run, "id", None)): # Track parent task if calling from within a task from prefect.client.schemas.objects import TaskRunResult task_inputs["__parents__"] = [TaskRunResult(id=task_ctx.task_run.id)] with get_client(sync_client=True) as client: parent_task = client.create_task_run( task=dummy_task, flow_run_id=flow_run_id, dynamic_key=dynamic_key, task_inputs=task_inputs, state=Pending(), ) # 3) Create the child flow run and link with parent_task_run_id child = client.create_flow_run_from_deployment( deployment_id=deployment_id, parameters=parameters, tags=tags, name=name, parent_task_run_id=parent_task.id, work_queue_name=work_queue_name, job_variables=job_variables, ) # 4) Immediately update the child’s empirical policy if retries is not None or retry_delay is not None: client.update_flow_run( flow_run_id=child.id, empirical_policy=FlowRunPolicy(retries=retries, retry_delay=retry_delay), ) # 5) Optional: poll like run_deployment(timeout=...) if timeout == 0: return child

start = time.monotonic() while True: fr = client.read_flow_run(child.id) if fr.state and fr.state.is_final(): return fr if timeout is not None and (time.monotonic() - start) >= timeout: return fr time.sleep(poll_interval)

Option B — Set empirical_policy at creation (advanced)
- Same subflow linkage as above, but we POST our own DeploymentFlowRunCreate (including empirical_policy) to the API directly. This applies the policy “during creation.”

import time import uuid from prefect import get_client from prefect.context import FlowRunContext, TaskRunContext from prefect.tasks import Task from prefect.states import Pending from prefect.client.schemas.objects import FlowRunPolicy, FlowRun from prefect.client.schemas.actions import DeploymentFlowRunCreate, StateCreate from prefect.states import Scheduled from datetime import datetime, timezone from prefect.engine import collect_task_run_inputs_sync def launch_subflow_with_policy_during_creation( *, deployment_id, # UUID parameters: dict | None = None, tags: list[str] | None = None, job_variables: dict | None = None, name: str | None = None, work_queue_name: str | None = None, # policy to apply at creation time retries: int | None = None, retry_delay: int | None = None, # seconds timeout: float | None = 5, poll_interval: float = 1.0, ): parameters = parameters or {} flow_ctx = FlowRunContext.get() task_ctx = TaskRunContext.get() if not (flow_ctx or task_ctx): raise RuntimeError("Must be called from within a flow or task to create a subflow.") flow_run_id = flow_ctx.flow_run.id if flow_ctx else task_ctx.task_run.flow_run_id dummy_task = Task(name=name or "subflow", fn=lambda: None, version=None) dynamic_key = f"subflow-{uuid.uuid4().hex[:8]}" task_inputs = {k: collect_task_run_inputs_sync(v) for k, v in parameters.items()} if task_ctx and (not flow_ctx or task_ctx.task_run.flow_run_id == getattr(flow_ctx.flow_run, "id", None)): from prefect.client.schemas.objects import TaskRunResult task_inputs["__parents__"] = [TaskRunResult(id=task_ctx.task_run.id)] with get_client(sync_client=True) as client: parent_task = client.create_task_run( task=dummy_task, flow_run_id=flow_run_id, dynamic_key=dynamic_key, task_inputs=task_inputs, state=Pending(), ) # Build the body we’ll POST directly to the deployments endpoint body = DeploymentFlowRunCreate( parameters=parameters, tags=tags or [], name=name, parent_task_run_id=parent_task.id, work_queue_name=work_queue_name, job_variables=job_variables, # set empirical_policy at creation empirical_policy=FlowRunPolicy(retries=retries, retry_delay=retry_delay), # replicate run_deployment's immediate scheduling state=StateCreate.from_state(Scheduled(scheduled_time=datetime.now(timezone.utc))), ).model_dump(mode="json", exclude_unset=True) # POST: /deployments/{id}/create_flow_run resp = client.request( "POST", "/deployments/{id}/create_flow_run", path_params={"id": deployment_id}, json=body, ) child = FlowRun.model_validate(resp.json()) if timeout == 0: return child start = time.monotonic() while True: fr = client.read_flow_run(child.id) if fr.state and fr.state.is_final(): return fr if timeout is not None and (time.monotonic() - start) >= timeout: return fr time.sleep(poll_interval) ``` Key points

- Both approaches create a real subflow by first creating a parent task run and then passing parent_task_run_id into the child’s create call (this is exactly how run_deployment does it). - Option A is the “official client” route; Option B is an advanced direct-API approach that sets empirical_policy at creation time. - You cannot set retry_type; it’s orchestration-managed. - If you prefer using a deployment name like "flow-name/deployment-name", you can resolve the deployment_id with client.read_deployment_by_name(...) before calling either function. If you share how you prefer to reference the deployment (name vs id) and whether you want async variants, I can tailor the snippet further.

Thanks for the details. A few important points about your snippet and the “Late” runs you’re seeing: 1) timeout=5 does not affect scheduling - In run_deployment(..., timeout=5, as_subflow=True), the timeout only controls how long the caller waits before returning. It does not change scheduling or startup behavior. If ECS capacity is briefly constrained, some runs may not start within the 15s Late threshold and will flip to Late, even though they can still start later. Source: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/deployments/flow_runs.py 2) The policy update you’re sending uses deprecated fields and a non-user-settable field - In Prefect 3.x, set retries and retry_delay on FlowRunPolicy. The older max_retries and retry_delay_seconds fields are deprecated; retry_type is orchestration-managed and will be ignored/overwritten. - Use:

from prefect.client.schemas.objects import FlowRunPolicy

policy = FlowRunPolicy(retries=2, retry_delay=30)
async with get_client() as client:
    await client.update_flow_run(flow_run.id, empirical_policy=policy)

Schema reference showing current fields (and presence of empirical_policy during creation): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/client/schemas/actions.py#L525-L589 3) Why 4 start and 2 remain Late - “Late” means “still Scheduled and missed its scheduled_start_time by the grace period” (15s by default). It’s typically due to: - ECS capacity/scale-up delay (common when launching several tasks simultaneously) - Work pool or deployment concurrency limits - Slower task definition registration or image pulls for some of the runs - They should still transition out of Late once a worker starts them, unless something is blocking those two from being picked up. Late state rule: https://github.com/PrefectHQ/prefect/blob/75a10ad4aab57304d21bf261f2eb174a137ff56e/src/prefect/server/orchestration/core_policy.py#L1704-L1723 - Late threshold settings (default after_seconds=15): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/settings/models/server/services.py (ServerServicesLateRunsSettings) 4) Make the policy stick before pickup (and reduce Late) If you want to guarantee the empirical_policy is applied at creation time (and avoid the race), create the subflow with parent_task_run_id and set empirical_policy during creation. You can also push the scheduled_time a few seconds into the future to avoid Late while ECS spins up: ``` from datetime import datetime, timezone, timedelta from prefect import get_client from prefect.context import FlowRunContext, TaskRunContext from prefect.tasks import Task from prefect.states import Pending, Scheduled from prefect.client.schemas.objects import FlowRunPolicy, FlowRun from prefect.client.schemas.actions import DeploymentFlowRunCreate, StateCreate async def launch_subflow_with_policy_now( deployment_id, parameters=None, name=None, tags=None, job_variables=None, work_queue_name=None, retries=2, retry_delay=30, cushion_seconds=10, # schedule a few seconds in the future to avoid Late ): flow_ctx = FlowRunContext.get() task_ctx = TaskRunContext.get() if not (flow_ctx or task_ctx): raise RuntimeError("Call from within a flow/task to create a true subflow.") flow_run_id = flow_ctx.flow_run.id if flow_ctx else task_ctx.task_run.flow_run_id # 1) Create a parent placeholder task run (subflow linkage) dummy_task = Task(name=name or "subflow", fn=lambda: None, version=None) async with get_client() as client: parent_task = await client.create_task_run( task=dummy_task, flow_run_id=flow_run_id, dynamic_key=f"subflow-{datetime.now().timestamp()}", task_inputs={}, state=Pending(), ) # 2) Create the child with empirical_policy set at creation, and a slight schedule cushion body = DeploymentFlowRunCreate(

parameters=parameters or {}, tags=tags or [], name=name, parent_task_run_id=parent_task.id, work_queue_name=work_queue_name, job_variables=job_variables, empirical_policy=FlowRunPolicy(retries=retries, retry_delay=retry_delay), state=StateCreate.from_state( Scheduled(scheduled_time=datetime.now(timezone.utc) + timedelta(seconds=cushion_seconds)) ), ).model_dump(mode="json", exclude_unset=True) resp = await client.request( "POST", "/deployments/{id}/create_flow_run", path_params={"id": deployment_id}, json=body, ) return FlowRun.model_validate(resp.json())

5) Troubleshooting checklist for the two Late runs
- Check pool and queue:
  - Is the ECS work pool paused or at its concurrency limit?
  - Are there enough workers polling that pool/queue?

prefect work-pool inspect "<your-ecs-pool>"

- Check deployment concurrency limit and strategy (may gate starts).

prefect deployment inspect "<flow-name>/<deployment-name>"

- Check ECS capacity/logs:
  - Are corresponding ECS tasks being created for those two runs?
  - Any capacity/provider constraints, task-definition registration retries, or image pull delays?
- If Late is purely cosmetic for short start delays, consider increasing the Late threshold:

export PREFECT_SERVER_SERVICES_LATE_RUNS_AFTER_SECONDS=30 ``` TL;DR - Use FlowRunPolicy(retries=..., retry_delay=...) and don’t set retry_type. - timeout=5 just controls how long you wait; it doesn’t influence scheduling. - Two “Late” runs usually indicate ECS capacity or concurrency constraints; they should start once capacity frees up. - If you want the policy applied before pickup (and to reduce Late), use the “create as subflow + set empirical_policy during creation” pattern above and optionally schedule a few seconds in the future.

Thanks for the straight ask — here’s how to achieve both goals and why those two runs are stuck. What you can do safely - You can update empirical_policy immediately after run_deployment returns; that’s fine and supported. Just use the current fields:

from prefect.client.schemas.objects import FlowRunPolicy
  policy = FlowRunPolicy(retries=2, retry_delay=30)  # do not set retry_type
  async with get_client() as client:
      await client.update_flow_run(flow_run.id, empirical_policy=policy)

This works while the run is Scheduled (or any state). It does not affect scheduling, but it configures retries when failures occur. Why two subflows are stuck in Late and never start - Late is a Scheduled variant set when a run hasn’t started by its scheduled time plus a grace period (default ~15s). It will only move out of Late when a worker picks it up. - If some runs never leave Late, there’s almost always a pickup/blocking issue, not a policy issue. Common blockers: - Pool or queue paused - No worker polling that pool/queue - Deployment concurrency limit is full - With CANCEL_NEW collision strategy: Late runs can be proactively Cancelled instead of staying Late - With ENQUEUE: Late runs may remain Late until a slot frees up - ECS capacity constraints (insufficient CPU/memory, scaling delays), task-def registration/image pulls slower for some tasks - Worker query filters exclude certain runs, e.g., runs with empirical_policy.retry_type='in_process' are excluded from queue selection Concrete checks to run - Verify the pool is healthy and not paused:

prefect work-pool inspect "<your-ecs-pool>"

Look for paused=false, a non-zero concurrency limit if set, and that workers are connected. - Inspect the deployment for a concurrency limit and collision strategy:

prefect deployment inspect "<flow-name>/<deployment-name>"

If the limit is small and the strategy is CANCEL_NEW, Late runs can be cancelled at mark-late time; with ENQUEUE, they can remain Late until a slot frees. - Confirm a worker is polling the right pool/queue and is healthy:

prefect worker start -p "<your-ecs-pool>"

Then check worker logs for: - Task definition registration retries or failures - Image pull delays - Capacity provider issues (Fargate/EC2) - Whether those specific run IDs are ever requested - If you rely on work queues, ensure you target the correct queue in run_deployment (work_queue_name) and that the queue is not paused. - If Late is caused by brief ECS startup delays and is only cosmetic, consider increasing the Late grace threshold:

export PREFECT_SERVER_SERVICES_LATE_RUNS_AFTER_SECONDS=30

This won’t fix “never picked up” runs, but reduces false Late during bursts. Key point about empirical_policy vs Late - Updating empirical_policy does not move a run out of Late; it’s just metadata for retries once the run actually starts. - If runs never leave Late, a worker/pool/queue/concurrency/capacity condition is preventing pickup. Fixing that is the path forward. If you can share: - The ECS work pool name and output of prefect work-pool inspect - The deployment’s concurrency settings and collision strategy - One affected flow run ID I can identify the exact blocker.

Short answer: set it on your Prefect Server only, not on workers. Details: - The late-runs check is a background service that runs in the Prefect Server process. Workers do not run this service. - Accepted env var names (both work; the API-prefixed one is the canonical form): - PREFECT_API_SERVICES_LATE_RUNS_AFTER_SECONDS - PREFECT_SERVER_SERVICES_LATE_RUNS_AFTER_SECONDS Source: - Setting definition and accepted aliases: https://github.com/PrefectHQ/prefect/blob/54aca27ce0/src/prefect/settings/models/server/services.py#L344-L354 - Late-runs service reads the setting server-side: https://github.com/PrefectHQ/prefect/blob/54aca27ce0/src/prefect/server/services/late_runs.py#L53-L63 How to apply (self-hosted): - Export the env var in the environment where you run prefect server start (or add it to the server container/pod env):

export PREFECT_API_SERVICES_LATE_RUNS_AFTER_SECONDS=30
  prefect server start

- No change is needed on ECS workers.