Chu
11/14/2025, 6:33 PMfoo labelled with result_storage=X. I deploy it and call run_deployment from another flow bar with result_storage=Y, I see that get_run_context().result_store.result_storage inside foo actually returns Y and not X as expected. I've verified that foo.result_storage=X. What's wrong?Marvin
11/14/2025, 6:33 PMMarvin
11/14/2025, 6:45 PMrun_deployment(...) should use the child flow’s own result storage (its @flow(result_storage=...) or its work pool default), not the caller’s. Seeing Y inside foo suggests you’re not actually looking at the deployed child run’s context, or the deployed code/config doesn’t have X applied at runtime.
A few common causes for what you’re seeing:
- You’re not actually inside the deployed child run. If foo() is ever called directly (as a subflow), it will inherit the parent’s result store Y. Note that run_deployment(..., as_subflow=True) only affects the UI lineage; it does not share execution context or result storage.
- The deployed code for foo doesn’t have @flow(result_storage=X) (e.g., the worker pulled a different branch/commit than your local verification). In that case, the work pool default or global default could be Y.
- The work pool used by `foo`’s deployment has a default result storage set to Y, and foo isn’t actually setting result_storage at runtime (e.g., an override elsewhere like foo.with_options(...) removed it or a conditional is leaving it unset).
- You passed job_variables or environment that sets a default result storage to Y and foo isn’t explicitly setting one at runtime.
What should happen
- Flow-level result_storage (X) has the highest precedence when the run actually executes your deployed foo. The parent’s result store does not override it.
- run_deployment creates a separate flow run (executed by a worker), which resolves its own result storage independently.
- Docs: Run a deployment, Results and result storage, Work pools
How to verify quickly
1) Inside foo, log these at the top of the run:
from prefect.flow_engine import get_run_context
ctx = get_run_context()
print("child ctx.result_store.result_storage =", ctx.result_store.result_storage)
print("flow-level result_storage on this run =", getattr(ctx.flow, "result_storage", None))
print("child flow run id =", ctx.flow_run.id)
2) In bar, capture and log the FlowRun returned by `run_deployment`:
from prefect.deployments import run_deployment
fr = run_deployment("path/to/foo-deployment", parameters={...})
print("created child flow run id =", fr.id)
Ensure the IDs match; otherwise you’re not looking at the child run’s logs.
3) Double-check the deployment and pool:
- Confirm `foo`’s deployment references the code version that contains @flow(result_storage=X).
- Check the work pool used by `foo`’s deployment for a default result storage set to Y.
If this still reproduces, can you share:
- Your Prefect version
- How foo is defined (the @flow(...) line) and how it’s deployed
- The exact run_deployment(...) call from bar
- Whether bar itself is a deployment or run locally
- The work pool used by foo and whether it has a default result storage configured
With that, I can pinpoint whether this is a configuration mismatch or a bug.Chu
11/14/2025, 6:51 PMrun_deployment. Does that change anything?Marvin
11/14/2025, 6:51 PMMarvin
11/14/2025, 6:51 PMfoo() directly from bar, foo runs as a subflow in the same process and inherits the parent flow’s current execution context, including its ResultStore. That’s why get_run_context().result_store.result_storage shows Y. This is expected for subflows.
If you want foo to use its own result storage X, you have a few options:
- Run it as a separate flow run via a deployment (use run_deployment(...)). This creates an independent context that resolves `foo`’s own result_storage.
- Force it at call time: foo.with_options(result_storage=X)(...)
- Or set task-level storage in foo with @task(result_storage=X) for the tasks whose results you want stored in X.
References:
- Run a deployment
- Results and result storageChu
11/14/2025, 6:52 PMresult_storage?Marvin
11/14/2025, 6:53 PMMarvin
11/14/2025, 6:59 PMY.
Authoritative references you can use:
- Subflows run in the same process as their parent flow (so they share the parent’s execution context):
- Workflow Design Patterns — “A subflow has the same relationship to its parent flow as a task does. It runs in the same process as its parent flow.”
- Flow API notes about inheriting defaults from the parent when called as a subflow:
- Flow API reference — the flow parameters (e.g., result serializer) note that when called as a subflow, defaults are loaded from the parent flow. Result storage follows the same context-inheritance mechanism.
- Result-store resolution logic:
- Results and result storage — describes how Prefect resolves result storage.
- Source code: ResultStore.update_for_flow keeps the current context’s result store unless the child flow explicitly sets one, which is why subflows inherit the parent’s store by default:
- prefect/results.py
- The current context store is accessed via `get_run_context().result_store`:
- prefect/flow_engine.py
Quick repro you can run locally to see the inheritance:
from prefect import flow
from prefect.flow_engine import get_run_context
@flow(result_storage="/tmp/X")
def foo():
ctx = get_run_context()
print("foo result_storage:", ctx.result_store.result_storage)
@flow(result_storage="/tmp/Y")
def bar():
ctx = get_run_context()
print("bar result_storage:", ctx.result_store.result_storage)
foo() # subflow call; inherits bar's context unless foo overrides
foo.with_options(result_storage="/tmp/X")() # explicitly override if desired
if __name__ == "__main__":
bar()
What to remember:
- Subflow (direct call) → inherits parent’s context/result store unless the child flow explicitly sets its own result_storage.
- Deployment (run_deployment) → separate flow run with its own context; the child resolves its own result_storage based on its flow definition/work pool defaults.
If you want, share a minimal snippet of your `foo`/`bar` definitions and I can confirm exactly how Prefect will resolve result storage in your setup.Chu
11/14/2025, 7:01 PMfoo() without with_options, what is the result_storage?Marvin
11/14/2025, 7:02 PMMarvin
11/14/2025, 7:02 PMfoo() directly as a subflow:
- The subflow inherits the parent’s current run context, including its ResultStore.
- So if bar has result_storage=Y and you call foo() directly, get_run_context().result_store.result_storage inside foo will be Y (the parent’s), even if foo was decorated with result_storage=X.
If you want foo to use X while running as a subflow, you must override at call time:
foo.with_options(result_storage=X)(...)
Or run it as an independent flow run via a deployment (run_deployment(...)) so it resolves its own storage.
References you can cite:
- Subflows run in the same process as their parent, i.e., they share the parent’s execution context:
- Workflow Design Patterns (see the “Subflows” section)
- Result storage is a part of the run context:
- Results and result storage (explains how result storage is resolved from the current context)
- The current context (including result_store) is accessed via `get_run_context()`:
- Context API reference
- Deployments (vs. subflows) create separate flow runs with their own context and result storage resolution:
- Run a deployment
Note: There isn’t a one-line statement in the docs that says “subflows inherit result_storage” verbatim, but since subflows run in the same process and share the parent run context, and the result store is part of that context, the inheritance follows from those documented behaviors.