feat(server): add aclose() to drain ActiveTask background tasks (#1101)

src/a2a/server/agent_execution/active_task.py

@@ -729,6 +729,48 @@ async def cancel(self, call_context: ServerCallContext) -> Task:
             raise RuntimeError('Task should have been created')
         return task
 
+    async def aclose(self) -> None:
+        """Force-closes the task's queues and drains its background tasks.
+
+        Provides a bounded, public teardown for the producer and consumer
+        ``asyncio.Task``s spawned in ``start()``. Without it, a producer
+        wedged in its ``finally`` closing an abandoned subscriber sink can
+        survive until event-loop shutdown and surface as
+        ``Task was destroyed but it is pending!``.
+
+        Always forces: the queues are closed with ``immediate=True`` and the
+        background tasks are cancelled, so teardown is bounded even when a
+        subscriber sink was never drained. It is safe to call multiple times.
+        """
+        # Shut down the request queue first, mirroring the producer's
+        # ``finally``. If `start()` was never called, the producer/consumer
+        # ``finally`` blocks never run, and without this a caller parked in
+        # `enqueue_request()` would wait forever.
+        self._request_queue.shutdown(immediate=True)
+        await self._event_queue_agent.close(immediate=True)
+        await self._event_queue_subscribers.close(immediate=True)
+        # Set `_is_finished` and collect the background tasks under `_lock` so
+        # this is mutually exclusive with `start()`, which refuses to spawn
+        # once `_is_finished` is set. The lock is released before awaiting the
+        # tasks, because their teardown re-acquires it.
+        async with self._lock:
+            self._is_finished.set()
+            background_tasks = [
+                task
+                for task in (self._producer_task, self._consumer_task)
+                if task is not None
+            ]
+            for task in background_tasks:
+                task.cancel()
+        if background_tasks:
+            results = await asyncio.gather(
+                *background_tasks, return_exceptions=True
+            )
+            for result in results:
+                # CancelledError is a BaseException, so it is excluded here.
+                if isinstance(result, Exception):
+                    logger.error('Error during aclose', exc_info=result)
+
     async def _maybe_cleanup(self) -> None:
         """Triggers cleanup if task is finished and has no subscribers.
 

src/a2a/server/agent_execution/active_task_registry.py

@@ -34,6 +34,7 @@ def __init__(
         self._active_tasks: dict[str, ActiveTask] = {}
         self._lock = asyncio.Lock()
         self._cleanup_tasks: set[asyncio.Task[None]] = set()
+        self._closed = False
 
     async def get_or_create(
         self,
@@ -44,6 +45,8 @@ async def get_or_create(
     ) -> ActiveTask:
         """Retrieves an existing ActiveTask or creates a new one."""
         async with self._lock:
+            if self._closed:
+                raise RuntimeError('ActiveTaskRegistry is closed')
             if task_id in self._active_tasks:
                 return self._active_tasks[task_id]
 
@@ -86,3 +89,39 @@ async def get(self, task_id: str) -> ActiveTask | None:
         """Retrieves an existing task."""
         async with self._lock:
             return self._active_tasks.get(task_id)
+
+    async def aclose(self) -> None:
+        """Closes the registry and drains all active tasks.
+
+        Marks the registry closed so ``get_or_create`` refuses new work, then
+        force-closes every registered ``ActiveTask`` and awaits the in-flight
+        ``_remove_task`` cleanup tasks they schedule, so no SDK-owned
+        ``asyncio.Task`` is left pending at event-loop shutdown. Safe to call
+        multiple times.
+
+        The close flag is set and the active-task snapshot is taken under
+        ``_lock``, and the lock is then released before awaiting, because
+        ``_remove_task`` re-acquires ``_lock``; holding it while draining
+        would deadlock. Marking closed under the same lock prevents a
+        concurrent ``get_or_create`` from registering a task that the drain
+        would miss.
+        """
+        async with self._lock:
+            self._closed = True
+            active_tasks = list(self._active_tasks.values())
+
+        if active_tasks:
+            results = await asyncio.gather(
+                *(task.aclose() for task in active_tasks),
+                return_exceptions=True,
+            )
+            for result in results:
+                if isinstance(result, Exception):
+                    logger.error('Error draining active task', exc_info=result)
+
+        cleanup_tasks = list(self._cleanup_tasks)
+        if cleanup_tasks:
+            await asyncio.gather(*cleanup_tasks, return_exceptions=True)
+
+        async with self._lock:
+            self._active_tasks.clear()

src/a2a/server/events/event_queue_v2.py

@@ -239,6 +239,21 @@ async def close(self, immediate: bool = False) -> None:
                 *(sink.close(immediate=False) for sink in sinks_to_close)
             )
 
+        # Both branches above cancel the dispatcher task, but the sink gather
+        # only awaits the sinks. Await the cancelled dispatcher here so it is
+        # never left pending when close() returns -- e.g. a source created with
+        # create_default_sink=False and no taps has no sinks for the gather to
+        # await, which would otherwise leave the cancelled dispatcher pending
+        # and trigger "Task was destroyed but it is pending!".
+        #
+        # Suppress the expected CancelledError and any error the dispatcher
+        # raised while shutting down: close() is teardown (reachable from
+        # __aexit__) and must not raise, and _dispatch_loop already logs its own
+        # exceptions, so a dispatcher failure stays observable without
+        # propagating out of close().
+        with contextlib.suppress(asyncio.CancelledError, Exception):
+            await self._dispatcher_task
+
     def is_closed(self) -> bool:
         """[DEPRECATED] Checks if the queue is closed.
 

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -112,6 +112,15 @@ def __init__(  # noqa: PLR0913
         )
         self._background_tasks = set()
 
+    async def aclose(self) -> None:
+        """Shuts down the handler, draining all active tasks.
+
+        Drains the ``ActiveTaskRegistry`` so a server shutdown leaves no
+        pending ``asyncio.Task``. Intended to be wired into an ASGI
+        ``lifespan`` / ``on_shutdown`` hook. Safe to call multiple times.
+        """
+        await self._active_task_registry.aclose()
+
     @validate_request_params
     async def on_get_task(  # noqa: D102
         self,

tests/server/agent_execution/test_active_task.py

@@ -20,6 +20,7 @@
     TaskStatus,
     TaskStatusUpdateEvent,
 )
+from a2a.utils._async_queue_compat import QueueShutDown, create_async_queue
 from a2a.utils.errors import InvalidParamsError
 
 
@@ -895,3 +896,125 @@ async def execute_mock(req, q):
     assert len(events) == 0
 
     await active_task.cancel(request_context)
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_active_task_aclose_reaps_background_tasks():
+    """aclose() drains a live producer and consumer."""
+    agent_executor = Mock()
+    task_manager = Mock()
+    request_context = Mock(spec=RequestContext)
+
+    active_task = ActiveTask(
+        agent_executor=agent_executor,
+        task_id='test-task-id',
+        task_manager=task_manager,
+        push_sender=Mock(),
+    )
+
+    async def slow_execute(req, q):
+        await asyncio.sleep(10)
+
+    agent_executor.execute = AsyncMock(side_effect=slow_execute)
+    task_manager.get_task = AsyncMock(
+        return_value=Task(
+            id='test-task-id',
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+        )
+    )
+
+    await active_task.enqueue_request(request_context)
+    await active_task.start(
+        call_context=ServerCallContext(), create_task_if_missing=True
+    )
+
+    await active_task.aclose()
+
+    assert active_task._producer_task is not None
+    assert active_task._producer_task.done()
+    assert active_task._consumer_task is not None
+    assert active_task._consumer_task.done()
+    assert active_task._is_finished.is_set()
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_active_task_aclose_force_closes_undrained_subscriber():
+    """aclose() unblocks past an undrained subscriber sink.
+
+    Reproduces issue #1101: a graceful close(immediate=False) would block
+    forever on the leaked sink's join().
+    """
+    agent_executor = Mock()
+    task_manager = Mock()
+    request_context = Mock(spec=RequestContext)
+
+    active_task = ActiveTask(
+        agent_executor=agent_executor,
+        task_id='test-task-id',
+        task_manager=task_manager,
+        push_sender=Mock(),
+    )
+
+    async def slow_execute(req, q):
+        await asyncio.sleep(10)
+
+    agent_executor.execute = AsyncMock(side_effect=slow_execute)
+    task_manager.get_task = AsyncMock(
+        return_value=Task(
+            id='test-task-id',
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+        )
+    )
+
+    await active_task.enqueue_request(request_context)
+    await active_task.start(
+        call_context=ServerCallContext(), create_task_if_missing=True
+    )
+
+    # Leak a subscriber sink and push an event into it without draining it.
+    leaked = await active_task._event_queue_subscribers.tap()
+    await active_task._event_queue_subscribers.enqueue_event(Message())
+    await asyncio.sleep(0.05)
+
+    await active_task.aclose()
+
+    assert active_task._producer_task is not None
+    assert active_task._producer_task.done()
+    assert leaked.is_closed()
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_active_task_aclose_never_started_shuts_down_request_queue():
+    """aclose() on a never-started task shuts down the request queue.
+
+    If start() was never called, the producer and consumer `finally` blocks
+    that normally shut down `_request_queue` never run, so without this a
+    caller parked in `enqueue_request()` would never be released.
+    """
+    active_task = ActiveTask(
+        agent_executor=Mock(),
+        task_id='test-task-id',
+        task_manager=Mock(),
+        push_sender=Mock(),
+    )
+
+    # Swap in a bounded queue and fill it so the next enqueue_request parks.
+    active_task._request_queue = create_async_queue(maxsize=1)
+    await active_task.enqueue_request(Mock(spec=RequestContext))
+    waiter = asyncio.create_task(
+        active_task.enqueue_request(Mock(spec=RequestContext))
+    )
+    await asyncio.sleep(0.05)
+    assert not waiter.done()
+
+    await active_task.aclose()
+
+    # The parked waiter is released with the shutdown error, and any
+    # subsequent enqueue fails fast instead of hanging.
+    with pytest.raises(QueueShutDown):
+        await waiter
+    with pytest.raises(QueueShutDown):
+        await active_task.enqueue_request(Mock(spec=RequestContext))

tests/server/agent_execution/test_active_task_registry.py

@@ -0,0 +1,107 @@
+import asyncio
+import logging
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from a2a.server.agent_execution.active_task_registry import ActiveTaskRegistry
+from a2a.server.agent_execution.agent_executor import AgentExecutor
+from a2a.server.agent_execution.context import RequestContext
+from a2a.server.context import ServerCallContext
+from a2a.server.events.event_queue_v2 import EventQueue
+from a2a.server.tasks import InMemoryTaskStore
+
+
+class _SlowExecutor(AgentExecutor):
+    """An executor whose execute() blocks until cancelled."""
+
+    async def execute(
+        self, context: RequestContext, event_queue: EventQueue
+    ) -> None:
+        await asyncio.sleep(10)
+
+    async def cancel(
+        self, context: RequestContext, event_queue: EventQueue
+    ) -> None:
+        return None
+
+
+def _make_registry() -> ActiveTaskRegistry:
+    return ActiveTaskRegistry(
+        agent_executor=_SlowExecutor(),
+        task_store=InMemoryTaskStore(),
+    )
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_aclose_reaps_active_tasks_and_empties_registry():
+    """aclose() reaps background tasks and removes them."""
+    registry = _make_registry()
+    active = await registry.get_or_create(
+        'task-1',
+        call_context=ServerCallContext(),
+        create_task_if_missing=True,
+    )
+
+    await registry.aclose()
+
+    assert active._producer_task is not None
+    assert active._producer_task.done()
+    assert active._consumer_task is not None
+    assert active._consumer_task.done()
+    assert await registry.get('task-1') is None
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_aclose_is_idempotent():
+    """Calling aclose() repeatedly is a safe no-op."""
+    registry = _make_registry()
+    await registry.get_or_create(
+        'task-1',
+        call_context=ServerCallContext(),
+        create_task_if_missing=True,
+    )
+
+    await registry.aclose()
+    await registry.aclose()
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_aclose_on_empty_registry():
+    """aclose() with no active tasks returns immediately."""
+    registry = _make_registry()
+    await registry.aclose()
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_get_or_create_rejected_after_aclose():
+    """A closed registry refuses to create new tasks (no orphan race)."""
+    registry = _make_registry()
+    await registry.aclose()
+
+    with pytest.raises(RuntimeError):
+        await registry.get_or_create(
+            'task-1',
+            call_context=ServerCallContext(),
+            create_task_if_missing=True,
+        )
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_aclose_logs_and_swallows_task_errors(caplog):
+    """A failing ActiveTask.aclose is logged, not propagated."""
+    registry = _make_registry()
+    failing = AsyncMock()
+    failing.aclose = AsyncMock(side_effect=ValueError('boom'))
+    registry._active_tasks['bad'] = failing
+
+    with caplog.at_level(logging.ERROR):
+        await registry.aclose()
+
+    assert 'Error draining active task' in caplog.text