Repository navigation
Conversation
ActiveTask.subscribe caught asyncio.CancelledError next to QueueShutDown and broke out of its loop. When the task iterating the subscription was cancelled, the generator ended normally instead of re-raising, so the cancellation was lost. For DefaultRequestHandlerV2.on_message_send this meant asyncio.wait_for never raised TimeoutError and Task.cancel() was ignored. The call returned the task in whatever non-terminal state it had at that moment as if it were the result. A cancelled consumer of on_message_send_stream or on_subscribe_to_task saw its stream end quietly instead of being cancelled. Only QueueShutDown is caught now. The producer and consumer only enqueue Exception instances and queue shutdown raises QueueShutDown, so a CancelledError at this point always comes from the caller. The finally block still detaches the tap and drops the reference count, so cleanup is unchanged.
🧪 Code Coverage (vs
|
| Base | PR | Delta | |
|---|---|---|---|
| src/a2a/server/cluster/database_event_stream.py | 92.86% | 96.94% | 🟢 +4.08% |
| src/a2a/server/request_handlers/default_request_handler_v2.py | 88.68% | 87.87% | 🔴 -0.81% |
| Total | 93.37% | 93.38% | 🟢 +0.01% |
Generated by coverage-comment.yml
This branch has not been deployed
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
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.
Description
ActiveTask.subscribecatchesasyncio.CancelledErrortogether withQueueShutDownand breaks out of its loop. So when the task that iterates a subscription gets cancelled, the generator just finishes and the cancellation is lost.It shows up with a timeout around a blocking send. With an executor that moves the task to WORKING and keeps going,
asyncio.wait_for(handler.on_message_send(...), timeout=0.3)returns the task in TASK_STATE_WORKING instead of raisingTimeoutError. Callingcancel()on a task runningon_message_sendbehaves the same way. It ends normally withcancelled() == Falseand returns whatever non-terminal state the task had at that moment. A cancelled consumer ofon_message_send_streamoron_subscribe_to_tasksees its stream end as if the task had finished.The fix catches only
QueueShutDown. Thefinallyblock already closes the tap, drops the reference count and runs_maybe_cleanup(), so cleanup happens the same way when the cancellation propagates. I couldn't find anything other than the caller that would raiseCancelledErrorthere. The producer and the consumer only putExceptioninstances on the queue (the producer swallows its own cancellation in_run_producer), and shutting the queue down raisesQueueShutDownfor bothimmediate=Trueandimmediate=False.The existing
test_active_task_subscribe_cancelled_during_waitaccepted eitherCancelledErrororStopAsyncIteration, with a comment putting the second one down to a Python 3.10 generator quirk. It was really thisexceptswallowing the cancel, and the test getsStopAsyncIterationon 3.13 as well. With the fix both versions raiseCancelledError, so the test now requires it.I also added three handler tests using the existing
_HangingAgent. One wrapson_message_sendinwait_forand expectsTimeoutError, one cancels a task runningon_message_send, and one cancels a task consumingon_message_send_stream. With thesrcchange stashed all four fail on both 3.10 and 3.13 (the three new ones with "DID NOT RAISE", the old one withStopAsyncIteration), and with it they pass.On 3.10 at 43d1eff the full suite gives 2494 passed, 203 skipped, 4 xfailed, 1 xpassed, and the coverage run clears the 88% gate.
./scripts/lint.shpasses.One related thing I left alone. The unary wrapper in
_plugin_hooks.pyonly callsafter_handler_runfromexcept Exception, so a cancelledsend_messagenow getsbefore_handler_runwith no matching after call. That already happened for a cancel at any other await, but the subscribe wait is where a blocking send usually sits when a timeout or disconnect hits, and before this change the after hook ran there with the stale task. The stream wrapper calls it fromfinally. I'm happy to make the unary side match in a separate PR if you want that.CONTRIBUTINGGuide.bash scripts/format.shfrom the repository root to format)