[v0ropaev]

Description

ActiveTask.subscribe catches asyncio.CancelledError together with QueueShutDown and 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 raising TimeoutError. Calling cancel() on a task running on_message_send behaves the same way. It ends normally with cancelled() == False and returns whatever non-terminal state the task had at that moment. A cancelled consumer of on_message_send_stream or on_subscribe_to_task sees its stream end as if the task had finished.

The fix catches only QueueShutDown. The finally block 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 raise CancelledError there. The producer and the consumer only put Exception instances on the queue (the producer swallows its own cancellation in _run_producer), and shutting the queue down raises QueueShutDown for both immediate=True and immediate=False.

The existing test_active_task_subscribe_cancelled_during_wait accepted either CancelledError or StopAsyncIteration, with a comment putting the second one down to a Python 3.10 generator quirk. It was really this except swallowing the cancel, and the test gets StopAsyncIteration on 3.13 as well. With the fix both versions raise CancelledError, so the test now requires it.

I also added three handler tests using the existing _HangingAgent. One wraps on_message_send in wait_for and expects TimeoutError, one cancels a task running on_message_send, and one cancels a task consuming on_message_send_stream. With the src change stashed all four fail on both 3.10 and 3.13 (the three new ones with "DID NOT RAISE", the old one with StopAsyncIteration), 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.sh passes.

One related thing I left alone. The unary wrapper in _plugin_hooks.py only calls after_handler_run from except Exception, so a cancelled send_message now gets before_handler_run with 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 from finally. I'm happy to make the unary side match in a separate PR if you want that.

• Follow the CONTRIBUTING Guide.
• Make your Pull Request title in the https://www.conventionalcommits.org/ specification.
• Ensure the tests and linter pass (Run bash scripts/format.sh from the repository root to format)
• Appropriate docs were updated (if necessary)

[github-actions]

🧪 Code Coverage (vs main)

⬇️ Download Full Report

	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