fix(gateway): run MCP shutdown off-loop with a bounded wait on the shutdown path

shutdown_mcp_servers() blocks on future.result(timeout=15) which, called
from the gateway event-loop thread during SIGTERM teardown, freezes the
loop for up to 15s when the MCP loop and its stdio children are torn down
concurrently. Supervisors with a shorter kill grace (s6-overlay: 3s)
SIGKILL the gateway before lifecycle_ledger.mark_exited() runs, producing
phantom 'exited UNCLEANLY' reports on every subsequent boot.

Run the sync shutdown on a daemon thread and poll via _await_thread_exit
with a 5s budget; proceed with teardown if it wedges. Fixes #82874;
completes the shutdown half of #64155.

gateway/run.py

@@ -32322,6 +32322,45 @@ async def _await_thread_exit(
     return not thread.is_alive()
 
 
+async def _shutdown_mcp_servers_nonblocking(timeout: float = 5.0) -> bool:
+    """Close MCP servers off-loop with a bounded wait (#82874).
+
+    ``shutdown_mcp_servers()`` is synchronous and can block for its full
+    internal 15s future wait when the MCP loop and its stdio children are
+    torn down concurrently (every process in the tree gets SIGTERM at once
+    under a container/supervisor stop). Calling it directly from the gateway
+    event-loop thread freezes the loop for that whole window, so supervisors
+    with a shorter kill grace (s6-overlay defaults to 3s) SIGKILL the gateway
+    before ``lifecycle_ledger.mark_exited()`` runs and every subsequent boot
+    reports a phantom unclean death.
+
+    Run it on a daemon thread instead and poll with ``_await_thread_exit`` so
+    the loop keeps servicing teardown. If it does not finish within
+    ``timeout`` we proceed with shutdown; the daemon thread is left to finish
+    (or die with the process) in the background. Returns True when the MCP
+    shutdown completed within the budget.
+    """
+
+    def _do() -> None:
+        try:
+            from tools.mcp_tool import shutdown_mcp_servers
+
+            shutdown_mcp_servers()
+        except Exception:
+            logger.debug("MCP shutdown raised", exc_info=True)
+
+    thread = threading.Thread(target=_do, name="mcp-shutdown", daemon=True)
+    thread.start()
+    done = await _await_thread_exit(thread, timeout=timeout)
+    if not done:
+        logger.warning(
+            "MCP shutdown did not finish within %.1fs; continuing gateway "
+            "teardown (background thread will be reaped at process exit)",
+            timeout,
+        )
+    return done
+
+
 def _shutdown_gateway_health_export(runner: Any) -> None:
     """Idempotently drain and detach Gateway Health OTLP export."""
     runtime = getattr(runner, "_gateway_health_export_runtime", None)
@@ -33115,8 +33154,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
                     logger.error("Gateway exiting with failure: %s", runner.exit_reason)
                 return False
             try:
-                from tools.mcp_tool import shutdown_mcp_servers
-                shutdown_mcp_servers()
+                await _shutdown_mcp_servers_nonblocking()
             except Exception:
                 pass
             if runner.exit_code is not None:
@@ -33296,10 +33334,9 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
     _planned_stop_watcher_stop.set()
     _planned_stop_watcher_thread.join(timeout=2)
 
-    # Close MCP server connections
+    # Close MCP server connections (off-loop, bounded — #82874)
     try:
-        from tools.mcp_tool import shutdown_mcp_servers
-        shutdown_mcp_servers()
+        await _shutdown_mcp_servers_nonblocking()
     except Exception:
         pass
 

tests/gateway/test_gateway_shutdown.py

@@ -395,3 +395,50 @@ def test_pid_exists_zombie_via_psutil_returns_false(monkeypatch):
     monkeypatch.setitem(sys.modules, "psutil", fake_psutil)
 
     assert status._pid_exists(4242) is False
+
+
+
+
+@pytest.mark.asyncio
+async def test_shutdown_mcp_servers_nonblocking_keeps_loop_responsive():
+    """A wedged MCP shutdown must not freeze the gateway event loop (#82874)."""
+    started = asyncio.Event()
+    loop = asyncio.get_running_loop()
+
+    def wedged_shutdown():
+        loop.call_soon_threadsafe(started.set)
+        import time as _time
+
+        _time.sleep(30)
+
+    heartbeats = 0
+
+    async def heartbeat():
+        nonlocal heartbeats
+        while True:
+            heartbeats += 1
+            await asyncio.sleep(0.05)
+
+    hb = asyncio.create_task(heartbeat())
+    try:
+        with patch("tools.mcp_tool.shutdown_mcp_servers", wedged_shutdown):
+            done = await asyncio.wait_for(
+                gateway_run._shutdown_mcp_servers_nonblocking(timeout=0.5),
+                timeout=5,
+            )
+    finally:
+        hb.cancel()
+
+    assert started.is_set()
+    assert done is False  # wedged shutdown exceeded the budget
+    # The loop kept running while the shutdown thread was wedged.
+    assert heartbeats >= 5
+
+
+@pytest.mark.asyncio
+async def test_shutdown_mcp_servers_nonblocking_completes_fast_path():
+    calls = []
+    with patch("tools.mcp_tool.shutdown_mcp_servers", lambda: calls.append(1)):
+        done = await gateway_run._shutdown_mcp_servers_nonblocking(timeout=5)
+    assert done is True
+    assert calls == [1]