Thanks for the detail. That message means Prefect saw an OS/runtime-level cancellation (anyio/asyncio CancelledError) during your task run and treated it as a crash. After macOS upgrades, we commonly see this when the OS pauses or terminates the Python process (sleep/App Nap/memory pressure) or when a signal (SIGTERM/SIGKILL) reaches the process. With

serve()

running locally, these interruptions are more likely to propagate to your flow/tasks. Likely culprits on macOS after an upgrade - System sleep/App Nap pauses the event loop, then Prefect cancels in-flight work - Memory pressure causes the OS to SIGKILL the process - Terminal/iTerm session changes or shutdown send SIGTERM - A paused event loop can also cause internal lease renewals to fail, which cancels runs (see this related report: Concurrency lease renewal failed → cancellation) Quick things to try 1) Keep the machine awake during `serve()`: - Run your script under caffeinate to prevent sleep/App Nap:

caffeinate -dimsu python your_script.py

- Also make sure “Prevent App Nap” is enabled for your terminal app (Get Info → Prevent App Nap) and disable system sleep while testing. 2) Turn on debug logging to see the exact cause:

export PREFECT_LOGGING_LEVEL=DEBUG
   python your_script.py

Then look for lines mentioning SIGTERM, “lease renewal failed”, exit codes, or memory issues above the crash. 3) Watch for memory pressure: - Keep Activity Monitor open (Memory tab) while the flow runs. - If you see spikes, reduce concurrency or data size in the task, or try a sequential runner to isolate:

from prefect import flow
     from prefect.task_runners import SequentialTaskRunner

     @flow(task_runner=SequentialTaskRunner())
     def my_flow():
         ...

4) If you’re using any explicit timeouts or concurrency keys, temporarily remove them to rule out cancellation from those features. More robust local alternative to serve() - Instead of

serve()

(which co-locates everything in a single process), run a worker that polls your deployment. This decouples orchestration from your terminal session and is less sensitive to accidental signals: - Create/update a deployment as usual. - Start a worker:

prefect worker start -p <your-work-pool>

(Verified command structure.) - Then trigger your flow runs from the UI/CLI. Upgrade Prefect - If you’re not on the latest 3.x, upgrade—there have been fixes and improvements around engine robustness: - Latest notes: 3.6.7 release and changelog If you can, please share: - Output of

prefect version

and your Python version - How you’re launching

serve()

(Terminal/iTerm/VS Code), and whether the machine sleeps or the lid closes during runs - Any lines above the error when running with

PREFECT_LOGGING_LEVEL=DEBUG

- Whether you’ve configured timeouts or concurrency for the flow/tasks With that, I can pinpoint whether this is a signal (SIGTERM/SIGKILL), a lease/timeout cancellation, or something else and suggest a more targeted fix.

Thanks — those logs are super helpful. Key signals in your output - RuntimeWarning: coroutine 'run_task_async' was never awaited - Task run crashed as “Execution was cancelled by the runtime environment” - Flow finished in state Completed() That combination almost always means: your flow returned while one or more submitted tasks were still in-flight. When the flow ends, the task runner shuts down its executor and cancels any outstanding work; Prefect then logs the crash for each cancelled task and you see the “was never awaited” warning during executor shutdown. The flow can still end as Completed if you never awaited/collected those task futures. Most likely cause in your code - You’re calling

.submit(...)

for tasks like

process_rss_item(...)

but not calling

.result()

or otherwise waiting for them before the flow function returns. How to fix - Ensure you wait for all submitted tasks to finish before the flow exits. Any of these patterns will work: 1) Collect results explicitly

from prefect.futures import as_completed

futs = [process_rss_item.submit(item) for item in items]
for f in as_completed(futs):
    f.result()  # raises if a task failed

2) Wait for completion, then resolve

from prefect.futures import wait

futs = [process_rss_item.submit(item) for item in items]
done, not_done = wait(futs)
for f in done:
    f.result()

3) If you don’t need the values but want to ensure completion

_ = [f.result() for f in futs]

Why you saw the two warnings - They’re emitted when the executor is shutting down (

self._executor.shutdown(cancel_futures=True)

in

prefect/task_runners.py

) while a coroutine representing task work (

run_task_async

) was still pending. That’s consistent with cancellation during teardown because the flow ended early. macOS angle - The macOS upgrade may also increase the chance of cancellations if the machine sleeps or the terminal session sends signals, but in your logs the “flow Completed + task crashed” pattern strongly suggests un-awaited futures rather than an OS-level signal as the primary cause. - Still, for stability during local dev with

serve()

, you can run under caffeinate to avoid sleep:

caffeinate -dimsu python your_script.py

Optional next steps - If you need “fire-and-forget” behavior, Prefect isn’t designed to detach tasks from a flow run; instead, kick off a subflow (or a deployment via an automation) and let it run independently. Otherwise, always wait for futures within the flow. - Upgrade to the latest Prefect 3.x for improved robustness: 3.6.7 release notes - For a more resilient local setup than

serve()

, consider running a worker and submitting runs to it:

prefect worker start -p <your-work-pool>

Then trigger runs from the UI/CLI. If you can share the portion of your flow where you submit

process_rss_item

, I can suggest the most concise wait/collection pattern for your case. Also, if you intended to run those tasks in the background and not block the flow, we can outline an approach with subflows/deployments.

Short answer: because in Prefect, tasks live under the supervision of the flow. When your flow returns, the task runner shuts down and cancels any still-running tasks. If you don’t wait for those futures, they’ll be cancelled during teardown—hence your “coroutine was never awaited” warnings and “Execution was cancelled by the runtime environment” crashes. Waiting isn’t about retrieving values; it keeps the runner alive until tasks reach a terminal state. What to do if you don’t need the return values - Ensure completion without using results:

from prefect.futures import wait

  futs = [process_rss_item.submit(item) for item in items]
  done, _ = wait(futs)  # blocks until all are finished

- Keep the flow green even if some tasks fail:

from prefect.futures import wait

  futs = [process_rss_item.submit(item) for item in items]
  done, _ = wait(futs)
  for f in done:
      _ = f.result(raise_on_failure=False)  # observe completion without raising

If you truly want “fire-and-forget” Tasks can’t be detached from the parent flow; they’ll be cancelled when the flow ends. Instead, fan out separate flow runs via a deployment so they continue independently after the parent finishes:

from prefect.deployments import run_deployment

for item in items:
    run_deployment(
        "your-project/your-child-flow-deployment-name",
        parameters={"item": item},
        as_subflow=False,  # run as its own flow run, not tied to the parent
    )
# parent flow can return immediately; children keep running in the work pool

This preserves the “don’t block the parent” behavior without cancelling work. If you want, I can help sketch the child flow + deployment pattern for your RSS items.