fix(worker): evict cached workflows during shutdown

[cystema]

Problem

Core does not evict idle cached workflows when a worker shuts down. In sdk-core, shutdown_done (workflow_stream.rs) only waits for runs that has_any_pending_work: an outstanding WFT, activation, buffered task, or an eviction that was already requested. An idle cached run has none of those, so Core reports shutdown complete and the run stays in the cache without ever receiving a remove_from_cache job.

On the Python side those workflows are then left as suspended coroutines. Python later closes them during garbage collection, which throws GeneratorExit into the workflow on whatever thread or event loop happens to be running GC. Any cleanup that needs to await in a finally raises RuntimeError: coroutine ignored GeneratorExit, and any commands issued there can cross workflow contexts (the failure mode documented in #494).

A workflow that waits on a condition and awaits asyncio.sleep(0) in finally reproduces this with no framework integration. The practical trigger is any deployment that stops a worker with open workflows; the workflows resume fine on the replacement worker, but the stopped worker fails its own teardown.

Fix

After in-flight activations finish and before the workflow executor stops, the workflow worker builds a local eviction activation for every run still in _running_workflows and drives it through the existing safe-eviction path. The activation has the same shape as Core's create_evict_activation: no timestamp, is_replaying false, a single remove_from_cache job. A new report_to_core keyword on _handle_cache_eviction skips the completion acknowledgment, since polling has stopped and Core never issued the job.

This mirrors what sdk-typescript already does: when the poller leaves the POLLING state it injects a synthetic eviction activation (SELF_INDUCED_SHUTDOWN_EVICTION) into every cached workflow and does not send its completion to Core. This PR brings the Python worker to parity.

Evictions run concurrently, honor the explicit disable_safe_workflow_eviction opt-out, do not cancel workflows on the server, and do not close a caller-owned executor. The reason is LANG_REQUESTED, which the replayer's eviction hook already treats as a non-failure.

Behavior to be aware of

Shutdown now waits for these evictions the same way it waits for Core-driven ones. A workflow that swallows BaseException and keeps waiting will make eviction retry forever and hold shutdown open, exactly as test_workflow_eviction_swallow documents for the Core-driven case. Previously such a workflow was silently garbage-collected instead. The existing "Timed out running eviction job" log fires so the cause is visible.

Tests

test_worker_shutdown_cleans_cached_workflows starts three cached workflows, shuts the worker down, and asserts each finally ran under eviction (is_replaying() is set to true on eviction, so the assertion proves the cleanup ran on the workflow's own loop rather than in GC). It is parametrized over the default and a caller-owned executor; both fail before the patch and pass after. The server-side workflows are asserted still RUNNING.

Validation

• poe test -n 0 tests/worker/test_workflow.py -k 'shutdown_cleans_cached or cache_eviction_tear_down or workflow_eviction_exception or workflow_eviction_swallow' on Python 3.12.8: all selected tests pass.
• poe lint passes (Ruff, Pyright, mypy, BasedPyright, docstyle).
• Python-only change; tested against the released 1.32.0 native bridge. No Rust changes.

[CLAassistant]

All committers have signed the CLA.

[tconley1428]

Can you elaborate on what the user impact is here. What are you trying to prevent or accomplish with this change?

[cystema]

The user impact is incomplete cleanup when a worker shuts down with running workflows—for example, during a deployment.

In our worker-replacement test, the workflow resumes successfully on the replacement worker, but the stopped worker leaves pending workflow coroutines behind. Python later tries to finalize them through garbage collection. Cleanup that needs to await then raises RuntimeError: coroutine ignored GeneratorExit, causing our tests to fail.

This also reproduces without the Agents SDK: a workflow waits indefinitely and its finally block contains await asyncio.sleep(0).

The change uses the existing safe-eviction path to clean up cached workflow instances before shutting down the executor. Server-side workflows remain running and can resume on another worker. The added regression tests cover both default and caller-owned executors.

[cystema]

@tconley1428 Adding two points my earlier answer was missing.

What Core does today. sdk-core never evicts idle cached runs at shutdown. shutdown_done in workflow_stream.rs only waits on runs that has_any_pending_work (an outstanding WFT, activation, buffered task, or an eviction that was already requested). An idle cached run has none of those, so Core reports shutdown complete with the run still cached and no remove_from_cache job is ever issued. The ignore_evicts_on_shutdown flag only covers evictions that are already pending.

What the other SDK does. sdk-typescript already handles this: when the poller leaves POLLING, worker.ts injects a synthetic eviction (SELF_INDUCED_SHUTDOWN_EVICTION) into every cached workflow and does not report its completion to Core. This PR is the same design for Python, with report_to_core=False playing the role of TS's synthetic: true.

User impact, concretely. Any deployment that stops a worker with open workflows leaves those workflows to be closed by GC. Cleanup that awaits in a finally then raises RuntimeError: coroutine ignored GeneratorExit, and commands issued there can land on another workflow's loop (#494). With this change the workflow is closed on its own loop before the executor stops; the server-side execution is untouched and resumes on the next worker.

Trade-off. Shutdown now waits for these evictions the same way it waits for Core-driven ones, so a workflow that swallows BaseException will hold shutdown open (as test_workflow_eviction_swallow already documents) instead of being silently GC'd. TS accepts the same trade-off. I've rewritten the PR description to spell all of this out.