fix(mcp): drain pending tasks before closing the MCP loop

_stop_mcp_loop() stopped and closed the background loop without reaping
the tasks still on it. A task left suspended is resumed later by the GC,
whose finalizer drives its cleanup against the now-closed loop:

    Exception ignored in: <coroutine object MCPServerTask.run ...>
      File "tools/mcp_tool.py", line 2947, in run
        parked = await self._wait_for_reconnect_or_shutdown(
      File "tools/mcp_tool.py", line 2161, in _wait_for_reconnect_or_shutdown
        t.cancel()
    RuntimeError: Event loop is closed

shutdown_mcp_servers() only reaps servers held in _servers, so a server
that parked after exhausting its initial-connect budget — never inserted
there, because start() raises _error before the caller registers it — has
no owner to signal it and stays suspended until the loop is gone.

Drain the loop the way asyncio.run() does: cancel the remaining tasks and
gather them while the loop is still open, so each runs its own finally.
Cancel alone is not enough — Task.cancel() only schedules the throw.

This resolves the reported traceback, but not the ownership bug that
strands the task in the first place; that needs a follow-up. Deliberately
not using "Fixes" so #60197 stays open for it.

Addresses #60197
Addresses #66113

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/tools/test_mcp_stability.py

@@ -736,3 +736,75 @@ class TestMCPInitialConnectionRetry:
                 await task
 
         asyncio.get_event_loop().run_until_complete(_run())
+
+
+# ---------------------------------------------------------------------------
+# Fix: drain pending tasks before closing the MCP loop
+# ---------------------------------------------------------------------------
+
+class TestMCPLoopDrainOnStop:
+    """_stop_mcp_loop reaps pending tasks while the loop is still open."""
+
+    def test_pending_task_cleanup_runs_before_close(self):
+        """A task left on the loop must finish its cleanup before close.
+
+        Regression for #60197: a task still suspended when the loop closes is
+        resumed later by the GC, and its ``finally`` then drives ``cancel()``
+        -> ``call_soon()`` against the closed loop, surfacing as an ignored
+        ``RuntimeError: Event loop is closed``. Servers absent from
+        ``_servers`` (e.g. parked after an initial-connect failure) never get
+        ``shutdown()``, so this drain is their only reaper.
+        """
+        import time
+        import tools.mcp_tool as mcp_mod
+
+        state = {
+            "started": False,
+            "cleanup_ran": False,
+            "cleanup_error": None,
+            "task": None,
+        }
+
+        async def _parked():
+            state["task"] = asyncio.current_task()
+            state["started"] = True
+            try:
+                await asyncio.sleep(3600)
+            finally:
+                # Mirrors _wait_for_reconnect_or_shutdown's finally: cancelling
+                # a helper task needs call_soon(), which a closed loop rejects.
+                try:
+                    helper = asyncio.ensure_future(asyncio.sleep(0))
+                    helper.cancel()
+                    state["cleanup_ran"] = True
+                except BaseException as exc:
+                    state["cleanup_error"] = exc
+
+        with mcp_mod._lock:
+            mcp_mod._servers.clear()
+            mcp_mod._server_connecting.clear()
+        try:
+            mcp_mod._ensure_mcp_loop()
+            with mcp_mod._lock:
+                loop = mcp_mod._mcp_loop
+            assert loop is not None
+            asyncio.run_coroutine_threadsafe(_parked(), loop)
+
+            deadline = time.monotonic() + 5
+            while not state["started"] and time.monotonic() < deadline:
+                time.sleep(0.01)
+            assert state["started"], "task never started on the MCP loop"
+
+            mcp_mod._stop_mcp_loop()
+
+            assert state["task"] is not None
+            assert state["task"].done(), "task left pending when the loop closed"
+            assert state["cleanup_error"] is None, (
+                f"cleanup ran against a closed loop: {state['cleanup_error']!r}"
+            )
+            assert state["cleanup_ran"], "task cleanup never ran"
+        finally:
+            with mcp_mod._lock:
+                mcp_mod._servers.clear()
+                mcp_mod._server_connecting.clear()
+            mcp_mod._stop_mcp_loop()

tools/mcp_tool.py

@@ -6618,6 +6618,23 @@ def _stop_mcp_loop_if_idle() -> bool:
     return _stop_mcp_loop(only_if_idle=True)
 
 
+async def _drain_mcp_loop_tasks() -> None:
+    """Cancel every task still pending on the MCP loop and reap it.
+
+    Cancelling is not enough on its own: ``Task.cancel()`` only schedules the
+    throw, so the task must be awaited before the loop goes away. Gather them
+    here — while the loop is still open — so each one runs its own ``finally``.
+    """
+    current = asyncio.current_task()
+    pending = [t for t in asyncio.all_tasks() if t is not current and not t.done()]
+    if not pending:
+        return
+    logger.debug("Draining %d pending task(s) from the MCP loop", len(pending))
+    for task in pending:
+        task.cancel()
+    await asyncio.gather(*pending, return_exceptions=True)
+
+
 def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool:
     """Stop the background event loop and join its thread."""
     global _mcp_loop, _mcp_thread
@@ -6630,6 +6647,24 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool:
         _mcp_loop = None
         _mcp_thread = None
     if loop is not None:
+        # Drain before stopping: closing the loop with tasks still suspended
+        # leaves their coroutines for the GC, whose finalizer then resumes them
+        # to run cleanup against a loop that is already closed -> "Event loop
+        # is closed" (#60197). ``shutdown_mcp_servers`` only reaps servers held
+        # in ``_servers``, so anything else left on this loop ends up here.
+        if loop.is_running():
+            from agent.async_utils import safe_schedule_threadsafe
+
+            future = safe_schedule_threadsafe(
+                _drain_mcp_loop_tasks(), loop,
+                logger=logger,
+                log_message="MCP loop drain: failed to schedule",
+            )
+            if future is not None:
+                try:
+                    future.result(timeout=5)
+                except BaseException as exc:
+                    logger.debug("Error draining MCP loop tasks: %s", exc)
         loop.call_soon_threadsafe(loop.stop)
         if thread is not None:
             thread.join(timeout=5)