[Bug]: ActiveTask.start() leaves the event queue dispatchers pending when setup fails #1421

What happened?

ActiveTask (src/a2a/server/agent_execution/active_task.py) creates its two EventQueueSource queues in __init__, and each EventQueueSource starts its dispatcher task as soon as it is created (src/a2a/server/events/event_queue_v2.py):

self._dispatcher_task = asyncio.create_task(self._dispatch_loop())

When the setup in start() fails, the except block marks the task as finished and re-raises, but never closes the queues:

except Exception:
    logger.debug('ActiveTask[%s]: Setup failed, cleaning up', self._task_id)
    self._is_finished.set()
    if self._reference_count == 0 and self._on_cleanup:
        self._on_cleanup(self)
    raise  # _event_queue_agent and _event_queue_subscribers stay open

The setup fails when the stored task is in a terminal state (UnsupportedOperationError) or is missing and create_task_if_missing is false (TaskNotFoundError). With DefaultRequestHandler, a SendMessage to a completed task takes the first path, so each rejected request leaves two _dispatch_loop tasks pending. Once they are garbage collected, asyncio logs:

Task was destroyed but it is pending!
task: <Task pending name='Task-28' coro=<EventQueueSource._dispatch_loop() ...>>

On a server built with DefaultRequestHandler, InMemoryTaskStore and the JSON-RPC and REST routes, six SendMessage requests to completed tasks log this twelve times.

Suggested fix: close both queues in the except block, before _on_cleanup:

await self._event_queue_agent.close(immediate=True)
await self._event_queue_subscribers.close(immediate=True)

aclose() already closes the queues the same way, but it can't be called here because it takes _lock, which start() holds.

I have a fix ready and I'm opening a PR alongside this issue.

Code of Conduct

• I agree to follow this project's Code of Conduct