fix: publish terminal task state to subscriber streams (#1175)

[anthonyrs06]

Summary

Follow-up to #1172 / #1175. cancel() and the producer-failure path wrote CANCELED / FAILED to the task store after both event queues were already closed, so a live SubscribeToTask / message/stream client saw WORKING and then a hung-up stream. Push did not fire either.

This persists a copy of the current task (the shared get_task() object is not mutated) and notifies observers before teardown:

• cancel() — if the executor's cancel() is cleanup-only (or the task is still non-terminal), emit a synthetic TaskStatusUpdateEvent(CANCELED) to _event_queue_subscribers and push_sender while the queues are still open. executor.cancel() runs before producer.cancel() so that write can still land on an open stream.
• Producer failure — persist FAILED to the store and fire push_sender before the finally closes the queues. A FAILED status event is not enqueued on the subscriber stream: blocking on_message_send treats a terminal Task as a successful result, so the crash must still surface as the producer exception (see scenario 9 / 12). The exception is enqueued before close so the stream does not hang up.

Overlaps #1172 on cancel ordering; this PR is the stream-visibility leftover that #1172 scoped out. Happy to rebase if #1172 lands first.

Test plan

• ./scripts/lint.sh
• uv run pytest (1787 passed locally)
• uv run pytest --cov=src --cov-report=term-missing (93%)
• New: live subscriber sees CANCELED on cancel(); shared task object is not mutated; push is notified
• New: producer crash persists FAILED + push, subscriber receives the exception (no hang)
• Existing blocking error scenarios 9 and 12 still raise A2AClientError

Made with Cursor

[github-actions]

🧪 Code Coverage (vs main)

⬇️ Download Full Report

	Base	PR	Delta
src/a2a/server/agent_execution/active_task.py	95.09%	94.65%	🔴 -0.43%
Total	92.97%	92.96%	🔴 -0.01%

Generated by coverage-comment.yml

[astrogilda]

Reviewed this because it carries the third point from my #1172 review, and #1175 credits it. Read of the diff, not a run of the branch.

The cancel() arm is right and the ordering is the part I'd have got wrong. Emitting inside finally before self._producer_task.cancel() means the _event_queue_subscribers queue is still open when the TaskStatusUpdateEvent goes in, because teardown lives in _run_producer's own finally and that can't run until the producer task is actually cancelled. Putting the cancel back at the end of the finally rather than before the try also keeps the BaseException property from #1172 intact: if executor.cancel raises CancelledError, the except Exception arm doesn't run but finally does, so the producer still gets cancelled. Both of those are easy to lose in a reorder and neither is lost here.

The publish_to_stream=False choice on the producer-failure arm is correct for the reason in the comment, and I want to be explicit that I agree, because it looks like an inconsistency and isn't. A blocking on_message_send that receives a FAILED Task treats it as a successful terminal result, so emitting a FAILED status event there would convert a crash into a success for every non-streaming caller. Surfacing the producer exception instead is the right call.

The consequence is worth stating in #1175 rather than leaving implicit, because it's a scope statement about what the issue closes. #1175 asks for a terminal outcome without polling. After this PR that holds for cancel() and does not hold for producer failure: a streaming client that survives the exception still cannot distinguish "the producer crashed" from "the connection dropped" without a get_task(), which is the workaround @rohityan gave above as a stopgap. So either the issue records that the producer-failure half stays polling-dependent, or the actual root cause gets its own issue. The blocking path conflating a FAILED terminal Task with success is a defect in on_message_send, not a constraint this PR has to design around. I'd rather see that filed than have the workaround become the contract. Happy to write it up if you'd rather not.

Three smaller things.

The _persist_and_publish_terminal helper guards only the store write with if task.status.state not in TERMINAL_TASK_STATES. The TaskStatusUpdateEvent construction, the stream publish and the push_sender.send_notification call all run unconditionally below it. Both cancel() call sites check terminality before calling, so they're fine, but the producer-failure call site doesn't. If the task is already terminal there, from a cancel that landed first, the store write is correctly skipped and push then fires anyway, carrying task.status.state, which is whatever the earlier terminal state was rather than FAILED. That's a duplicate notification with a state the caller didn't ask to send. Moving the guard inside the helper and returning early makes all three call sites safe by construction.

The else branch, which covers a producer that already finished or never started, calls _persist_and_publish_terminal(CANCELED) with the default publish_to_stream=True. If the producer has finished, its finally has already closed the subscriber queue, and EventQueue.enqueue_event on a closed queue logs "Queue is closed. Event will not be enqueued." and returns rather than raising (event_queue.py:119-122). So that branch is store plus push, never stream. Benign, since no live subscriber remains, but the code and the docstring both read as publishing there. One line either way.

This call puts a tuple on a queue whose declared element type is Event, with the cast doing the silencing:

await self._event_queue_subscribers.enqueue_event(cast('Any', (event, updated_task_copy)))

If (event, task) is genuinely the subscriber queue's contract, the type wants widening rather than a cast at the call site, because a cast('Any', ...) here is the thing that lets the next wrong shape through without a diagnostic. If it isn't the contract, this is the bug. Which is it?