Thanks for sharing the stack traces — they point to two different things happening at the very end of your run. Here’s what they mean and how to fix them so your results persist cleanly. What’s going on - PicklingError: “Pickling client objects is explicitly not supported” - Prefect is trying to persist the return value of your flow (or a subflow/task). That return value contains a “client” object (e.g., PrefectClient, httpx.Client, boto3 client, DB engine/connection, etc.). These are intentionally not serializable and will crash result persistence. - OSError: [Errno 5] Input/output error while monitoring - This occurs while the worker streams process output/logs from the run. It’s a transient I/O issue (e.g., stdout/stderr/file descriptor closed) during monitoring. Prefect warns you but does not mark the run failed. It’s usually harmless noise, but we can make it less likely. How to stop the crash and persist results safely 1) Do not return clients or other unpickleable objects - Make sure your flow, subflows, and tasks only return JSON-serializable values (dict/list/str/int/float/bool/None, datetimes/UUIDs are handled). - If you have a client, extract serializable data from it before returning, or persist your artifacts yourself (e.g., write to S3/DB) and return references/paths. Example patterns:

# BAD: returns a client
   @task
   def get_client():
       return httpx.Client()

   # GOOD: use the client internally; return data
   @task
   def fetch_data(url):
       with httpx.Client() as client:
           r = client.get(url)
           return r.json()

   # GOOD: for Pydantic v2 models
   @task
   def make_model():
       model = MyModel(...)
       return model.model_dump()  # or model_dump_json()

2) Explicitly configure result persistence on flows/subflows with a JSON serializer and a storage location - In Prefect 3.x, you can set these on each flow/task so only serializable data is written.

from prefect import flow, task
   from prefect.serializers import JSONSerializer
   from prefect.filesystems import LocalFileSystem

   result_store = LocalFileSystem(basepath="/opt/prefect/results")  # or S3/GCS/etc.

   @flow(
       persist_result=True,
       result_serializer=JSONSerializer(),
       result_storage=result_store,
   )
   def parent_flow():
       # call subflows/tasks and return serializable data
       sub_result = child_flow()
       return {"child_summary": sub_result, "status": "ok"}

   @flow(
       persist_result=True,
       result_serializer=JSONSerializer(),
       result_storage=result_store,
   )
   def child_flow():
       data = compute_something()  # return only JSON-serializable values here
       return {"rows": len(data)}

Notes: - If you have complex objects, you can use a custom object encoder with JSONSerializer to convert them to JSON-safe forms, but you still shouldn’t try to persist live client connections. - Alternatively, if you truly don’t need Prefect to store the return values, set

persist_result=False

and manage persistence yourself inside the task/flow. But since you asked to keep results, prefer the JSON approach above. 3) Reduce the “monitoring” OSError noise - Upgrade to the latest Prefect 3.x worker/runtime — we’ve improved robustness in log streaming and process monitoring over time. See Prefect release notes. - Ensure the worker’s stdout/stderr are not being closed or redirected to an ephemeral/rotating sink while the run is still streaming logs (common with some container logging setups). - If the error persists frequently, capturing worker logs and environment details can help us pinpoint the logging backend/source. The message itself does not mark the run failed. Helpful docs - Results and persistence: Prefect 3.x Results guide

- Flow API (persist_result/result_serializer/result_storage): Flow API - Task API (persist_result/result_serializer/result_storage): Task API - Serializers (JSON/Compressed/Custom encoders): Serializers API - Process worker reference: Process worker A couple quick questions to make sure we give you the best fix: - Which Prefect version are you on? - What does your flow return at the end (roughly)? Is there any client/engine/connection object in that return value or in subflow returns? - Where are your workers running (local, Docker, Kubernetes) and do you have any log collectors/sidecars that might close stdout/stderr? If you can share a small snippet of the flow’s return statement and the subflow/task outputs, I can suggest the exact changes to make them safely persisted.