Short answer: Prefect 3 does not “resume” a failed flow run.

run_deployment(..., idempotency_key=...)

only deduplicates flow-run creation; it won’t restart a failed run. The fix is to make tasks in Flow B (and any subflows) persist results with a cache policy that works across runs so a new run will reuse succeeded task results instead of re-executing them. Why your cache was missed - By default, tasks use the DEFAULT cache policy which includes

RUN_ID

. That means any new flow run of B will miss cache even if inputs/code are unchanged. - In a CI environment like CodeBuild, local result storage is ephemeral; even if cache keys matched, results may not be there unless you use durable storage. What to do 1) Use a durable result store shared by all runs (e.g., S3) - In CodeBuild, prefer S3 via Prefect’s RemoteFileSystem (it uses s3fs/boto and picks up IAM role creds). 2) Make tasks cache across runs - Use a cache policy that does NOT include

RUN_ID

, typically

INPUTS + TASK_SOURCE

. - Ensure persistence is enabled (it is auto-enabled when you set a cache policy, but set it explicitly for clarity). - Optionally set a cache_expiration. Concrete example for Flow B (applies to any subflow too)

from datetime import timedelta
from prefect import flow, task
from prefect.filesystems import RemoteFileSystem
from prefect.cache_policies import INPUTS, TASK_SOURCE

# Durable, shared result storage (works with your CodeBuild IAM role)
s3_store = RemoteFileSystem(basepath="<s3://my-prefect-results/my-project/>")

@task(
    cache_policy=INPUTS + TASK_SOURCE,   # exclude RUN_ID to reuse across runs
    cache_expiration=timedelta(days=7),
    persist_result=True                  # ensure results are written
)
def expensive_step(x: int) -> int:
    # do heavy work
    return x * 2

@flow(
    persist_result=True,                 # flow result persistence (optional)
    result_storage=s3_store,             # all tasks inherit storage unless overridden
    # result_serializer can be left default; pickle is typical for Python objects
)
def flow_b(x: int) -> int:
    a = expensive_step(x)                # will be reused on retry if inputs/code unchanged
    # ... other steps, give them the same cache_policy if you want them reused ...
    return a

Then from Flow A, when you retry: - Do NOT reuse the same idempotency key if you want a new run. Either omit it or generate a new key. The new run of B will find cached results in S3 for all previously completed tasks and only re-run the failed ones. Example call pattern from Flow A

from prefect import flow
from prefect.deployments import run_deployment

@flow
def flow_a(x: int):
    # first attempt
    run = run_deployment(
        "flow-b/my-deployment",
        parameters={"x": x},
        # no idempotency_key here so a real new run is created
    )
    # you can optionally wait for it if you need the result/state
    # from prefect.flow_runs import wait_for_flow_run
    # wait_for_flow_run(run, raise_on_failure=False)

    # on manual retry of A, call run_deployment again with the same parameters
    # a new flow run of B will be created, and cached tasks will be reused

Custom cache keys when inputs include non-deterministic values If your tasks’ inputs contain timestamps, UUIDs, etc., those will change the

INPUTS

hash and cause misses. In that case, provide a

cache_key_fn

that only uses stable business keys:

@task(
    persist_result=True,
    cache_key_fn=lambda ctx, inputs: f"expensive-step:v1:{inputs['job_id']}",
    cache_expiration=timedelta(days=7),
)
def expensive_step(job_id: str, ts: str) -> ...:
    ...

Checklist - Configure a durable ResultStore accessible to all workers (S3/RemoteFileSystem). - For every task you want to reuse on retry: set

persist_result=True

and

cache_policy=INPUTS + TASK_SOURCE

(or a custom

cache_key_fn

). - Keep parameters stable between retries (or use a custom

cache_key_fn

).

- On retry of Flow A, create a new run of Flow B (don’t reuse the same

idempotency_key

). - Ensure caching is not globally disabled (PREFECT_TASKS_DISABLE_CACHING should be false). Happy to help you wire this up to your CodeBuild/IAM setup and test with a minimal B flow to confirm you see skipped tasks on the second run.

Got it — you want retries of the same root run of A to reuse all succeeded task results (even inside a separately-deployed B), but starting a brand-new run of A should force all work to run from scratch. Key idea: propagate the root flow run’s ID from A into B and include it in the cache key for tasks in B. Then: - Retrying the same run of A keeps the same root_run_id, so B’s task caches are reused. - Starting a brand-new run of A yields a new root_run_id, so B’s task caches are naturally bypassed. Two clean ways to do this A) Use FLOW_PARAMETERS in B’s task cache policy (minimal wiring) - Pass A’s run id into B as a parameter (e.g., root_run_id). - In B, set tasks’ cache_policy to include FLOW_PARAMETERS so the root_run_id becomes part of the cache key. B) Pass root_run_id to each expensive task in B - Keep tasks’ cache_policy as INPUTS + TASK_SOURCE. - Add root_run_id as an argument to those tasks (even if the function doesn’t use it), so it’s included in INPUTS. I recommend A when you have many tasks to scope — it keeps task signatures clean. B is fine if you want explicit, per-task control. Concrete example Flow A: capture the root run id and pass it into B’s deployment call

from prefect import flow
from prefect.context import get_run_context
from prefect.deployments import run_deployment

@flow
def flow_a(x: int):
    root_run_id = str(get_run_context().flow_run.id)

    # First attempt (new run of B)
    run_deployment(
        "flow-b/my-deployment",
        parameters={"x": x, "root_run_id": root_run_id},
        # do NOT reuse the same idempotency_key here if you want a new run
    )

    # On retry of A (same flow run), call run_deployment again with the same root_run_id
    # A brand-new run of A will generate a different root_run_id and thus bypass caches

Flow B: scope task caches by the root_run_id Option A (FLOW_PARAMETERS)

from datetime import timedelta
from prefect import flow, task
from prefect.filesystems import RemoteFileSystem
from prefect.cache_policies import INPUTS, TASK_SOURCE, FLOW_PARAMETERS

# Durable shared result store (e.g., S3) so caches survive worker restarts
s3_store = RemoteFileSystem(basepath="<s3://my-prefect-results/my-project/>")

@task(
    cache_policy=INPUTS + TASK_SOURCE + FLOW_PARAMETERS,  # includes root_run_id as part of flow params
    cache_expiration=timedelta(days=7),
    persist_result=True
)
def expensive_step(x: int) -> int:
    # heavy work
    return x * 2

@flow(
    persist_result=True,
    result_storage=s3_store,
)
def flow_b(x: int, root_run_id: str):
    # Any tasks using the above cache_policy will be scoped by root_run_id
    y = expensive_step(x)
    return y

Option B (make root_run_id an explicit task input)

@task(
    cache_policy=INPUTS + TASK_SOURCE,   # root_run_id is part of INPUTS
    cache_expiration=timedelta(days=7),
    persist_result=True
)
def expensive_step(x: int, root_run_id: str) -> int:
    return x * 2

@flow(persist_result=True, result_storage=s3_store)
def flow_b(x: int, root_run_id: str):
    y = expensive_step(x, root_run_id)   # pass through explicitly
    return y

Important notes - Don’t reuse the same idempotency_key if you actually want a new run of B; idempotency only deduplicates creation and will hand you the same (failed) run. - Ensure result storage is durable/shared (e.g., S3 via RemoteFileSystem) so caches from the first attempt are visible to subsequent runs in your CodeBuild/worker environment. - If B’s flow parameters include volatile values (timestamps, UUIDs), FLOW_PARAMETERS will cause cache misses. In that case, either: - Keep volatile params out of B’s parameters and only include root_run_id, or - Use Option B so only root_run_id affects the task’s INPUTS-based key, or - Write a custom cache_key_fn that hashes just the stable business keys + root_run_id + code version.

If you’d like, I can help you wire this up with an S3 block and a quick end-to-end test to confirm tasks are skipped on A’s retry and recomputed on a new A run.

Yes — that’s a great approach. You can implement a

cache_key_fn

that mimics INPUTS + TASK_SOURCE and appends the root flow run id, so: - Retries of the same Flow A run reuse results (same root id) - A brand-new Flow A run forces fresh work (new root id) The easiest way is to reuse Prefect’s built-in

task_input_hash

(which already covers INPUTS + TASK_SOURCE) and tack on the root id from

get_root_flow_run_id()

. Example

from prefect import task
from prefect.tasks import task_input_hash
from prefect.runtime.flow_run import get_root_flow_run_id

def inputs_source_plus_root(context, arguments) -> str:
    base = task_input_hash(context, arguments)  # INPUTS + TASK_SOURCE
    root_id = get_root_flow_run_id()            # the top-most parent (Flow A)
    return f"{base}:root:{root_id}"

@task(
    cache_key_fn=inputs_source_plus_root,
    persist_result=True,            # ensure results are written so future runs can reuse
    # optional: cache_expiration=...
)
def expensive_step(x: int):
    ...

Notes - This works across deployed subflows: a B run started by A will have A as the root, so tasks in B will share the same root id with A’s run. - If B is run “standalone,” its root will just be the B run id — still deterministic. - Make sure you have durable/shared result storage (e.g., S3 via RemoteFileSystem) so the cache is visible across retries in CodeBuild. - Keep idempotency_key usage aligned with your intent: use a new run of B on retries (so caches are consulted) rather than returning the same failed run. - Ensure caching isn’t globally disabled (e.g., PREFECT_TASKS_DISABLE_CACHING). Docs - How to cache workflow steps - Access runtime info (root flow run id) - prefect.runtime.flow_run API