cancel() and producer-failure write a terminal state to the store but not to active subscriber streams

[chopmob-cloud]

Follow-up from the review of #1172 (raised by @astrogilda).

What happens

ActiveTask.cancel() now closes a task out to a terminal CANCELED state when
the executor's cancel() is cleanup-only or the task was parked in a
non-terminal state (e.g. input-required). That terminal state is written
directly to the task store via save_task_event, because by that point both
the agent and subscriber event queues are already closed.

The consequence is that the terminal state reaches the store but not an
active subscriber stream. A client that was streaming the task
(SubscribeToTask / message/stream) receives the initial Task in WORKING
and then the stream ends; it never sees a CANCELED event and has to re-read
the task to learn the outcome. The push path does not fire either, since
_update_task_state only notifies on PushNotificationEvent.

Why it was left out of #1172

At the point of the direct store write, both queues are already closed, so a
store write is the only option left. The same pattern already exists for the
producer-failure path (active_task.py, the FAILED write in
_run_producer's except Exception arm), under a comment that says so. #1172
kept scope to the authorization fix (#1159) and the "cancel leaves an actionable
terminal state" fix (#1170); this stream-visibility gap is orthogonal.

Proposed follow-up

Give a streaming or push-subscribed client a way to observe the terminal state
without polling. Options to weigh:

• Emit a synthetic terminal TaskStatusUpdateEvent to _event_queue_subscribers
  before the queues are torn down, so an active subscriber sees the transition.
• Route the terminal store write through the push path so push_sender fires.

Both the cancel() CANCELED write and the producer-failure FAILED write share
this gap and should get the same treatment.

Repro

Subscribe to a live task whose executor's cancel() is cleanup-only, cancel it,
and collect the events the subscriber receives: the stream yields the initial
WORKING task and ends, while task_store.get afterwards returns CANCELED.

[anthonyrs06]

I'll take the follow-up that #1172 scoped out.

Long-lived A2A clients that subscribe (message/stream / SubscribeToTask) need to observe a terminal outcome without polling. Today cancel() and the producer-failure path write CANCELED/FAILED to the store after both queues are closed, so a subscriber sees WORKING and then a hung-up stream.

I'll emit a synthetic terminal TaskStatusUpdateEvent to _event_queue_subscribers before teardown so the store write and the stream stay aligned, and check whether that same emit can feed push_sender. Tests for both cancel() and the producer-failure arm, with a subscriber attached.

[anthonyrs06]

Opened #1191 for this.

cancel() emits CANCELED to the live subscriber stream and push before teardown. Producer-failure persists FAILED and fires push before close; the stream still gets the producer exception (blocking on_message_send would otherwise treat a FAILED Task as success).

[rohityan]

Hi @chopmob-cloud , thanks for separating the issue. PR #1191 seems to fix the issue. Thanks @anthonyrs06 for working on this.
As a workaround till then you can wrap your streaming loop with a terminal event gaurd, such that when the stream disconnects without a terminal status event ,the client can perform a fallback get_task() query to read the final state from the database.
Also , If you control the server and want your active clients to see cancellations immediately, you can emit the terminal event inside your custom AgentExecutor.cancel() method before returning: