Repository navigation
fix(server): mark internal lifecycle events done so a cancelled stream frees its producer - #1336
Merged
Iwaniukooo11 merged 10 commits intoOct 14, 2026
Merged
Conversation
…m frees its producer ActiveTask.subscribe consumed _RequestStarted and _RequestCompleted with a bare `continue`, never calling tapped_queue.task_done() for them. When a task finished normally the _RequestCompleted event made the subscriber return before the producer's cleanup, so it did no harm. But a cancel stops the agent before _RequestCompleted is sent: the producer then runs _event_queue_subscribers.close(immediate=False), which waits until every tap has marked all of its items done. The stream's tap still holds the unmarked _RequestStarted, so the wait never ends -- the producer and consumer tasks and the ActiveTask registry entry leak until the whole handler is aclose()-d. Mark the two internal lifecycle events done before skipping them. Adds a regression test that fails (TimeoutError) without the fix and passes with it. Fixes a2aproject#1322 Signed-off-by: Carsten Piepel <cpiepel@esri.com>
🧪 Code Coverage (vs
|
| Base | PR | Delta | |
|---|---|---|---|
| src/a2a/server/agent_execution/active_task.py | 95.14% | 95.40% | 🟢 +0.26% |
| Total | 93.12% | 93.13% | 🟢 +0.01% |
Generated by coverage-comment.yml
…leep Address review feedback on the a2aproject#1322 regression test: - Stop swallowing the producer's exception. The previous _settle helper suppressed all exceptions, so the test could pass if teardown crashed rather than completed. Now assert the producer finishes with no exception; the CancelledError from the explicit cancel is the only one allowed, closing a false-pass hole. - Replace the fixed sleep with an explicit synchronization event set when the subscriber enters its loop, removing a timing guess that could flake on a slow runner. Re-verified: the test still passes with the fix and still fails with TimeoutError when the source fix is reverted. Signed-off-by: Carsten Piepel <cpiepel@esri.com>
carstenpiepel
marked this pull request as draft
October 12, 2026 14:08
…ment Second review pass on the a2aproject#1322 regression test: - _run_producer catches asyncio.CancelledError and returns normally, so a cancelled producer never surfaces CancelledError from await and producer.exception() is always None here. Drop the dead CancelledError handling and the redundant done()/exception() checks; just await the producer. If teardown raises, await re-raises and the test fails -- the stronger property. Verified by injecting a RuntimeError into teardown. - Correct the synchronization comment: setting the event before the first await does not prove _RequestStarted was consumed. Describe the actual scheduling handoff instead of overstating a guarantee. Re-verified: passes with the fix, fails with TimeoutError when the fix is reverted, and fails by propagating the exception when teardown raises. Signed-off-by: Carsten Piepel <cpiepel@esri.com>
Instrumentation showed _RequestStarted is not dequeued before producer.cancel(); producer teardown drains it through the tap instead. Correct the comments so they no longer claim the handoff consumes the event before cancellation. Comment-only change. Signed-off-by: Carsten Piepel <cpiepel@esri.com>
Producer teardown drains _RequestStarted through the subscriber tap, so the reader-started event and sleep(0) were not needed. Remove them. The test still needs one synchronization point: cancelling the producer before it runs skips _run_producer's CancelledError handler, so the await raises CancelledError and the cancel path under test is never exercised. Wait on an event set from the mocked execute() instead. Verified 10/10 passes with the fix, 10/10 TimeoutError failures without it, and an injected teardown RuntimeError still propagates. Signed-off-by: Carsten Piepel <cpiepel@esri.com>
Iwaniukooo11
requested changes
Oct 13, 2026
Wrap each dequeued item in one try/finally so task_done() runs exactly once on every path (raise, lifecycle skip, return, yield). A re-raised error was never acked, which hangs a graceful close already joining the tap. Add a regression test and trim the a2aproject#1322 test to essentials.
Iwaniukooo11
requested changes
Oct 14, 2026
Member
|
@carstenpiepel I've added just one final comment. Once that is addressed, we should be all set and ready to merge:) |
Iwaniukooo11
approved these changes
Oct 14, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #1322.
Problem
When a streamed task is cancelled, its background producer never finishes and the
producer:<task_id>entry is never pruned from the registry.In
ActiveTask.subscribe, the two internal lifecycle events_RequestStartedand_RequestCompletedare dequeued from the per-subscriber tapped queue and then skipped withcontinue(orreturn) without callingtapped_queue.task_done(). Because those items are never marked done, the producer's teardownawait self._event_queue_subscribers.close(immediate=False)waits forever on a tap that still has unfinished items, so_run_producernever completes.The maintainer confirmed the root cause and approved this fix in #1322.
Fix
Call
tapped_queue.task_done()on the_RequestStarted/_RequestCompletedbranches before they are skipped, so the tapped queue's unfinished count returns to zero andclose(immediate=False)can complete. One-line change insrc/a2a/server/agent_execution/active_task.py.Regression test
tests/server/agent_execution/test_active_task.py::TestActiveTask::test_active_task_cancel_while_subscribed_frees_producer, built on the module's existing fixtures. It drives a subscriber past_RequestStarted, cancels the producer (the streamed-cancel path), and asserts the producer task completes rather than wedging.Verification
TimeoutErrorwhen the source change is temporarily reverted (confirming it catches the bug rather than being green by construction).active_taskmodule: 33 passed.agent_execution+ cluster cancel/subscribe + default request handler): 172 passed.ruff checkandruff format --check: clean.