You’re running into expected behavior:

run_deployment

will create a new flow run every time you call it unless you make the call idempotent. Result persistence controls how results are stored/retrieved, but it does not deduplicate

run_deployment

calls. What you want is to pass a stable

idempotency_key

to

run_deployment

. With the same key, Prefect 3 will return the existing child FlowRun instead of creating a new one on parent retries. The scope of idempotency is per flow (i.e., keyed by

(flow_id, idempotency_key)

), and there’s no TTL—so it will continue to resolve to the same run as long as that flow run remains in the database. Example pattern:

from prefect import flow, get_run_context
from prefect.deployments import run_deployment
import hashlib, json

def params_fingerprint(d: dict) -> str:
    return hashlib.sha1(json.dumps(d, sort_keys=True).encode()).hexdigest()

@flow
def parent_flow(x: int):
    # Stable identifier for this child call across parent retries:
    ctx = get_run_context()
    child_params = {"x": x}
    # Include params in the key if you want different inputs to produce new runs
    idem_key = f"{ctx.flow_run.id}:child-foo:{params_fingerprint(child_params)}"

    child_run = run_deployment(
        name="child-flow/production",
        parameters=child_params,
        idempotency_key=idem_key,
        as_subflow=True,  # optional; helps with UI lineage
    )

    # Get the child's returned value from persisted results
    # (this does not re-run the child; it reads from result storage)
    if child_run.state:
        result = child_run.state.result()
        # use `result` as needed

Notes and gotchas: - Keep the

idempotency_key

identical across the parent’s retries. Don’t use attempt counters or timestamps in the key. - If you redeploy in a way that creates a new underlying flow (different

flow_id

), the same

idempotency_key

will not dedupe against the previous run. Try to keep the same flow identity across versions if you rely on idempotency. -

run_deployment

returns a FlowRun object, not the child’s return value. Use

flow_run.state.result()

(or

aresult()

async) to read the persisted result. - If you’d like to skip even the lookup on retries, another pattern is to wrap the

run_deployment

call in a task with a cache policy; but using

idempotency_key

is the simplest, robust fix for your case. If you’re already passing an

idempotency_key

and still seeing a new child run on retries, share how you construct the key and whether the child’s flow/deployment changed between attempts—I can help you make it stable.

Great question — you’ve run into a real constraint of Prefect’s idempotency model. Short answer: there is no built-in “idempotent only-if-completed” mode for

run_deployment

. Idempotency is state-blind: if you pass the same

idempotency_key

, Prefect will always return the existing child run regardless of whether it succeeded or failed. To get “reuse on success, rerun on failure” you need a bit of logic around the

run_deployment

call. You do not have to wrap it in a task; you can do it at the flow level with a small helper. Two solid patterns 1) Prefer: State-aware lookup with tags (no idempotency_key) - Tag each child invocation so you can find it later. - Before launching, look for an existing child run for this parent + tag that is Completed. - If found, read its result; if not, create a new child run (with a fresh key or no key). Example

from prefect import flow, get_run_context
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FilterSet, FlowRunFilter, FlowRunFilterParentFlowRunId, FlowRunFilterTags, FlowRunFilterState, FlowRunFilterStateType
from prefect.client.schemas.sorting import FlowRunSort
from prefect.states import StateType

async def _find_completed_child(parent_id: str, tag: str):
    async with get_client() as client:
        runs = await client.read_flow_runs(
            filters=FilterSet(
                flow_runs=FlowRunFilter(
                    parent_flow_run_id=FlowRunFilterParentFlowRunId(any_=[parent_id]),
                    tags=FlowRunFilterTags(all_=[tag]),
                    state=FlowRunFilterState(type=FlowRunFilterStateType(any_=[StateType.COMPLETED])),
                )
            ),
            sort=FlowRunSort.ID_DESC,
            limit=1,
        )
        return runs[0] if runs else None

@flow
def parent_flow(x: int):
    ctx = get_run_context()
    subcall_tag = f"child:foo"  # or include params if needed, e.g. f"child:foo:{x}"

    # Look for an already-completed child for this parent + tag
    import anyio
    existing = anyio.run(_find_completed_child, ctx.flow_run.id, subcall_tag)

    if existing:
        # Read the persisted result from the completed run
        result = existing.state.result()
        return result

    # No completed child yet -> launch a new run
    child_run = run_deployment(
        name="child-flow/production",
        parameters={"x": x},
        tags=[subcall_tag],         # so we can find it next time
        as_subflow=True
    )
    return child_run.state.result()

Why this works - On parent retries: if a previous child run succeeded, you’ll find it and reuse the result; no new run is created. - If a previous child failed, there is no completed child to find — so you’ll launch a new one (exactly what you want). 2) Alternative: State-aware idempotency_key - Use a stable “base key” per subcall (e.g., parent_run_id + subcall_id [+ params]). - Before launching, look up the latest child with that base key. - If Completed: reuse it (don’t launch). - If Failed/Cancelled/Crashed: call

run_deployment

again with a new key (e.g., base_key + “:retry-<n>” or a fresh UUID) so it creates a new run. This is functionally similar to the tag approach, but you manage keys instead of tags. You still need the lookup to decide whether to reuse or create a new run. Notes - Result persistence env vars you set are perfect — they ensure

state.result()

reads from storage. -

run_deployment

always returns a FlowRun object, not the function’s return value. Use

flow_run.state.result()

(or

aresult()

in async) to load the child’s returned value. - If your parent flow is synchronous, the example uses

anyio.run(...)

to make the one async lookup. If you prefer, I can share a fully-async parent-flow variant or a version using

SyncPrefectClient

.

If you share how you uniquely identify each child invocation (e.g., one spot in the flow vs many, and which params matter), I can tailor the filter and tagging/keying so it’s a drop-in for your codebase.