fix: Prevent client disconnect from stopping task execution

src/a2a/server/request_handlers/default_request_handler.py

@@ -67,6 +67,7 @@ class DefaultRequestHandler(RequestHandler):
     """
 
     _running_agents: dict[str, asyncio.Task]
+    _background_tasks: set[asyncio.Task]
 
     def __init__(  # noqa: PLR0913
         self,
@@ -102,6 +103,9 @@ def __init__(  # noqa: PLR0913
         # TODO: Likely want an interface for managing this, like AgentExecutionManager.
         self._running_agents = {}
         self._running_agents_lock = asyncio.Lock()
+        # Tracks background tasks (e.g., deferred cleanups) to avoid orphaning
+        # asyncio tasks and to surface unexpected exceptions.
+        self._background_tasks = set()
 
     async def on_get_task(
         self,
@@ -355,10 +359,11 @@ async def push_notification_callback() -> None:
             raise
         finally:
             if interrupted_or_non_blocking:
-                # TODO: Track this disconnected cleanup task.
-                asyncio.create_task(  # noqa: RUF006
+                cleanup_task = asyncio.create_task(
                     self._cleanup_producer(producer_task, task_id)
                 )
+                cleanup_task.set_name(f'cleanup_producer:{task_id}')
+                self._track_background_task(cleanup_task)
             else:
                 await self._cleanup_producer(producer_task, task_id)
 
@@ -394,7 +399,11 @@ async def on_message_send_stream(
                 )
                 yield event
         finally:
-            await self._cleanup_producer(producer_task, task_id)
+            cleanup_task = asyncio.create_task(
+                self._cleanup_producer(producer_task, task_id)
+            )
+            cleanup_task.set_name(f'cleanup_producer:{task_id}')
+            self._track_background_task(cleanup_task)
 
     async def _register_producer(
         self, task_id: str, producer_task: asyncio.Task
@@ -403,6 +412,29 @@ async def _register_producer(
         async with self._running_agents_lock:
             self._running_agents[task_id] = producer_task
 
+    def _track_background_task(self, task: asyncio.Task) -> None:
+        """Tracks a background task and logs exceptions on completion.
+
+        This avoids unreferenced tasks (and associated lint warnings) while
+        ensuring any exceptions are surfaced in logs.
+        """
+        self._background_tasks.add(task)
+
+        def _on_done(completed: asyncio.Task) -> None:
+            try:
+                # Retrieve result to raise exceptions, if any
+                completed.result()
+            except asyncio.CancelledError:
+                name = completed.get_name()
+                logger.debug('Background task %s cancelled', name)
+            except Exception:
+                name = completed.get_name()
+                logger.exception('Background task %s failed', name)
+            finally:
+                self._background_tasks.discard(completed)
+
+        task.add_done_callback(_on_done)
+
     async def _cleanup_producer(
         self,
         producer_task: asyncio.Task,