Short answer: because the flow ran in a different process/worker, Prefect won’t have your return value in-memory unless you persist it. If result persistence isn’t enabled or the storage isn’t reachable/serializable, resolving the result will fail even though the state is Completed. Most common causes - Result persistence not enabled (default is off) - No result storage configured that both the worker and your orchestrator can access (e.g., trying to read a result written to the worker’s local disk) - Return value not serializable by your configured serializer - Eventual consistency or storage credential issues prevented the write How to fix 1) Enable persistence - Globally:

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

- Or on the flow:

from prefect import flow

@flow(persist_result=True)
def my_flow(...):
    return some_value

2) Configure result storage both sides can access - Prefer a remote filesystem block (S3/GCS/Azure). Example:

# Create and save an S3Bucket block (once)
from prefect_aws.s3 import S3Bucket
S3Bucket(bucket_name="my-results").save("my-s3-results", overwrite=True)

# Use it on your flow
@flow(persist_result=True, result_storage="s3-bucket/my-s3-results")
def my_flow(...):
    return some_value

- Or set a default:

prefect config set PREFECT_DEFAULT_RESULT_STORAGE_BLOCK='s3-bucket/my-s3-results'

3) Retrieve the result from your orchestrator after the run completes

from prefect.deployments import run_deployment

flow_run = await run_deployment("my-flow/prod", timeout=None)  # wait to complete
value = await flow_run.state.aresult()  # raises if failed; returns data if persisted

If you still see failures, please share: - How you’re launching the flow (deployment name, infra/worker type) - The exact code you use to “extract worker result” - Your result settings (persist_result on the flow? any PREFECT_* config?) and which storage block you’re using - The exception message/trace when extraction fails Docs and references - Results and persistence: https://docs.prefect.io/v3/advanced/results/ - States and resolving results: https://docs.prefect.io/v3/concepts/states/ and API https://docs.prefect.io/v3/api-ref/python/prefect-states/ - Running deployments and awaiting completion: https://docs.prefect.io/v3/how-to-guides/deployments/run-deployments/ and API https://docs.prefect.io/v3/api-ref/python/prefect-deployments-flow_runs/ Tip: If your flow returns large/complex objects, ensure they’re compatible with your serializer (default is pickle). For remote storage, confirm the worker had credentials to write and your orchestrator has credentials to read.

Great questions. Short answer - Yes, you can store the final aggregate metrics on the server’s local filesystem. - They’ll persist on disk indefinitely unless you delete them; Prefect does not auto-clean local results. Details and trade-offs - Local result storage - If you enable result persistence and do not set a result storage block, Prefect will write results to the worker’s local filesystem at

PREFECT_LOCAL_STORAGE_PATH

(defaults to something like

~/.prefect/storage

on that machine). - This is fine if both “writer” (the worker) and “reader” (your orchestrator process) are on the same machine. - If you try to fetch a result from a different machine, it won’t be accessible unless the path points to a shared mount (e.g., NFS) or you use a remote storage block (S3/GCS/Azure). - Persistence/retention - Prefect doesn’t delete local result files for you. They stay until you remove them. - You can set cache expiration on tasks for cache validity, but that does not delete files; it only controls whether a cached result should be reused. Recommended patterns for your use case (final aggregate metrics) - Single-machine: Use local storage - Enable persistence and leave storage unset, or explicitly configure the local path:

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
  prefect config set PREFECT_LOCAL_STORAGE_PATH='/path/on/server/prefect-results'

- Return your aggregate metrics from the flow. From the same server/process (or anything with access to that path), resolve the result:

# after awaiting completion
  value = await flow_run.state.aresult()

- Multi-machine or future-proofing: Use shared storage - Point results to a shared filesystem or a remote block (e.g., S3) so you can fetch from anywhere:

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
  prefect config set PREFECT_DEFAULT_RESULT_STORAGE_BLOCK='s3-bucket/my-results-block'

- Or on the flow:

from prefect import flow

  @flow(persist_result=True, result_storage='s3-bucket/my-results-block')
  def my_flow():
      return aggregate_metrics

Optional: Artifacts for human-readable summaries - If you mainly want a human-friendly summary in the UI, consider creating an artifact (markdown/table) alongside or instead of persisting the full object result. Artifacts live in Prefect Cloud/Server and have plan-based retention; they are great for quick dashboards but aren’t a general file store. - Docs: https://docs.prefect.io/v3/advanced/results/ and https://docs.prefect.io/v3/concepts/states/ If you can share whether the reader of these metrics runs on the same host as the worker (and your current

PREFECT_*

result settings), I can suggest an exact configuration.