From 9118512b18d3b8cc30c6207d6222cc524f7cdf4b Mon Sep 17 00:00:00 2001
From: Carlos Chinchilla Corbacho
 <188046461+cchinchilla-dev@users.noreply.github.com>
Date: Tue, 3 Nov 2026 18:37:01 +0100
Subject: [PATCH] fix(server): close the event queues when ActiveTask.start()
 fails

---
 src/a2a/server/agent_execution/active_task.py |  2 +
 .../agent_execution/test_active_task.py       | 45 ++++++++++++++++++-
 2 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py
index 121900fa2..513ea7bdc 100644
--- a/src/a2a/server/agent_execution/active_task.py
+++ b/src/a2a/server/agent_execution/active_task.py
@@ -636,6 +636,8 @@ async def start(
                     self._task_id,
                 )
                 self._is_finished.set()
+                await self._event_queue_agent.close(immediate=True)
+                await self._event_queue_subscribers.close(immediate=True)
                 if self._reference_count == 0 and self._on_cleanup:
                     self._on_cleanup(self)
                 raise
diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py
index 9f82d1685..ac9770475 100644
--- a/tests/server/agent_execution/test_active_task.py
+++ b/tests/server/agent_execution/test_active_task.py
@@ -22,7 +22,7 @@
     TaskStatusUpdateEvent,
 )
 from a2a.utils._async_queue_compat import QueueShutDown, create_async_queue
-from a2a.utils.errors import UnsupportedOperationError
+from a2a.utils.errors import TaskNotFoundError, UnsupportedOperationError
 
 
 logger = logging.getLogger(__name__)
@@ -1445,3 +1445,46 @@ async def slow_execute(req, q):
     for t in background:
         t.cancel()
     await asyncio.gather(*background, return_exceptions=True)
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('stored_task', 'expected_error'),
+    [
+        (
+            Task(
+                id='test-task-id',
+                status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+            ),
+            UnsupportedOperationError,
+        ),
+        (None, TaskNotFoundError),
+    ],
+)
+async def test_active_task_start_setup_failure_closes_event_queues(
+    stored_task, expected_error
+):
+    """A failed start() closes both event queues.
+
+    Each EventQueueSource starts its dispatcher task when it is created, so
+    without this the dispatchers stay pending until garbage collection and
+    surface as ``Task was destroyed but it is pending!``.
+    """
+    task_manager = Mock()
+    task_manager.get_task = AsyncMock(return_value=stored_task)
+
+    active_task = ActiveTask(
+        agent_executor=Mock(),
+        task_id='test-task-id',
+        task_manager=task_manager,
+        push_sender=Mock(),
+    )
+
+    with pytest.raises(expected_error):
+        await active_task.start(
+            call_context=ServerCallContext(), create_task_if_missing=False
+        )
+
+    assert active_task._event_queue_agent._dispatcher_task.done()
+    assert active_task._event_queue_subscribers._dispatcher_task.done()