Cancelling a non-streaming on_message_send while the agent is running never completes (CancelledError swallowed, then hangs in ActiveTask.get_task) #1448

Versions

• a2a-sdk[http-server] 1.3.0 (DefaultRequestHandlerV2)
• Python 3.13
• macOS / Linux

What happens

If the asyncio task awaiting DefaultRequestHandlerV2.on_message_send (non-streaming message/send) is cancelled while the AgentExecutor is still running and has not emitted any event yet, the cancellation is swallowed and the task never finishes: it stays parked forever in ActiveTask.get_task() on self._task_created.wait().

A visible consequence: asyncio.run(...) hangs at shutdown in _cancel_all_tasks() whenever a message/send call is in flight, because the cancelled handler task never completes. We first hit it through the SDK client's send_message over HTTP to an in-process server; the repro below calls the request handler directly.

Expected

Cancellation propagates: the task awaiting on_message_send ends with asyncio.CancelledError promptly (after the subscriber cleanup has run).

Minimal reproduction

No network, no server; only the SDK. The executor blocks before emitting anything; the send is started as a task, cancelled, and awaited with a 3 s bound so the script reports instead of hanging.

"""Cancelling a non-streaming on_message_send while the agent runs never finishes."""

import asyncio
import importlib.metadata
import os
import re
import sys

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.context import ServerCallContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers.default_request_handler_v2 import (
    DefaultRequestHandlerV2,
)
from a2a.server.tasks import InMemoryTaskStore
from a2a.types.a2a_pb2 import (
    AgentCapabilities,
    AgentCard,
    Message,
    Part,
    Role,
    SendMessageRequest,
)

started = asyncio.Event()


class SlowExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        started.set()
        await asyncio.Event().wait()  # long-running work; emits nothing yet

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        pass


async def main() -> int:
    print("a2a-sdk", importlib.metadata.version("a2a-sdk"), "| Python", sys.version.split()[0])
    handler = DefaultRequestHandlerV2(
        agent_executor=SlowExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=AgentCard(
            name="repro",
            description="repro",
            version="1.0",
            capabilities=AgentCapabilities(),
            default_input_modes=["text/plain"],
            default_output_modes=["text/plain"],
        ),
    )
    request = SendMessageRequest(
        message=Message(
            message_id="m1", role=Role.ROLE_USER, parts=[Part(text="hi")]
        )
    )
    send = asyncio.create_task(handler.on_message_send(request, ServerCallContext()))
    await asyncio.wait_for(started.wait(), 3)

    send.cancel()
    try:
        await asyncio.wait_for(asyncio.shield(send), 3)
    except asyncio.CancelledError:
        print("OK: on_message_send ended with CancelledError")
        return 0
    except TimeoutError:
        print("HANG: on_message_send still pending 3 s after cancel()")
        stack, coro = "", send.get_coro()
        while coro is not None and hasattr(coro, "cr_frame"):  # walk the await chain
            f = coro.cr_frame
            if f is not None:
                name = re.sub(r".*/(site-packages|lib/python3\.\d+)/", "", f.f_code.co_filename)
                stack += f"  {name}:{f.f_lineno} {f.f_code.co_name}\n"
            coro = coro.cr_await
        print("pending at:\n" + stack, end="")
        return 1
    print("UNEXPECTED: returned", send.result())
    return 2


if __name__ == "__main__":
    # asyncio.run() would itself hang in _cancel_all_tasks() on the stuck task,
    # so run the loop directly and exit without cancelling leftovers.
    code = asyncio.new_event_loop().run_until_complete(main())
    sys.stdout.flush()
    os._exit(code)

Output on 1.3.0:

a2a-sdk 1.3.0 | Python 3.13.15
HANG: on_message_send still pending 3 s after cancel()
pending at:
  a2a/utils/telemetry.py:250 async_wrapper
  a2a/server/request_handlers/request_handler.py:285 async_wrapper
  a2a/server/request_handlers/default_request_handler_v2.py:464 on_message_send
  a2a/server/agent_execution/active_task.py:1096 get_task
  asyncio/locks.py:213 wait

The asyncio.run shutdown variant: start on_message_send as a task, wait until the executor has started, and return from the main coroutine. asyncio.run() cancels the leftover tasks (the handler task and the SDK's producer/consumer tasks) and then waits for them; it never returns.

Root cause (line numbers from 1.3.0)

1. a2a/server/agent_execution/active_task.py:851, in ActiveTask.subscribe():

   except (QueueShutDown, asyncio.CancelledError):
       break

   A CancelledError delivered to the subscriber while it waits in tapped_queue.dequeue_event() is caught and turned into a break. The finally (lines 853-860) cleans up and the generator then ends normally, so the cancellation does not reach the caller.

2. a2a/server/request_handlers/default_request_handler_v2.py:424-464, in on_message_send(): the async for over subscribe() simply ends, result is still None, so the handler calls:

   result = await active_task.get_task()   # line 464

3. a2a/server/agent_execution/active_task.py:1094-1096, ActiveTask.get_task():

   await self._task_created.wait()

   _task_created is set only when a task already exists at start() (line 607), when the consumer stores the first task event emitted by the agent (line 273), or when the consumer records a producer failure (line 188). None of these happens here:

   • On a plain cancel of the request, the producer task is a separate background task and is not cancelled; it keeps running the executor, which has not emitted anything yet. _task_created stays clear for as long as the agent is silent (forever in the repro).
   • At asyncio.run shutdown, the producer is cancelled too; _run_producer catches CancelledError (line 705) and just closes the queues; nothing sets _task_created on that path either.

   Since the one CancelledError was already consumed in step 1, nothing interrupts the wait.

Side effect even when it does not hang: if the agent later emits a task, the cancelled on_message_send returns a normal result instead of raising CancelledError, which breaks structured-concurrency expectations (e.g. asyncio.timeout, TaskGroup).

Suggested fix direction

• In ActiveTask.subscribe(), do not swallow asyncio.CancelledError: handle only QueueShutDown with break, and let CancelledError propagate after the finally cleanup (or re-raise it explicitly). The existing finally already performs the unsubscribe and cleanup.
• Independently, on_message_send / get_task() should not wait unboundedly for _task_created once the subscription has ended without a result; e.g. wait on _task_created and _is_finished together, or return/raise when the task has finished without a stored task.