fix(mcp): drain connecting tasks before loop shutdown

hermes_cli/main.py

@@ -126,7 +126,7 @@ def _cleanup_oneshot_runtime() -> None:
         pass
     try:
         from tools.mcp_tool import shutdown_mcp_servers
-        shutdown_mcp_servers()
+        shutdown_mcp_servers(final=True)
     except BaseException:
         pass
     try:

tests/hermes_cli/test_mcp_security.py

@@ -207,7 +207,7 @@ def test_explicit_registration_skips_dangerous_entry_before_connect(monkeypatch)
     import tools.mcp_tool as mcp_tool
 
     monkeypatch.setattr(mcp_tool, "_MCP_AVAILABLE", True)
-    monkeypatch.setattr(mcp_tool, "_ensure_mcp_loop", lambda: None)
+    monkeypatch.setattr(mcp_tool, "_ensure_mcp_loop", lambda: True)
 
     connected = []
 

tests/hermes_cli/test_tui_resume_flow.py

@@ -890,7 +890,10 @@ def _mod(name, **attrs):
     monkeypatch.setitem(
         sys.modules,
         "tools.mcp_tool",
-        _mod("tools.mcp_tool", shutdown_mcp_servers=lambda: events.append("mcp")),
+        _mod(
+            "tools.mcp_tool",
+            shutdown_mcp_servers=lambda *, final=False: events.append(f"mcp:{final}"),
+        ),
     )
     monkeypatch.setitem(
         sys.modules,
@@ -903,15 +906,17 @@ def _mod(name, **attrs):
 
     main_mod._run_and_exit_oneshot("hello")
 
-    assert events == ["run", "terminal", "delegation", "browser", "mcp", "aux", "exit:0"]
+    assert events == [
+        "run", "terminal", "delegation", "browser", "mcp:True", "aux", "exit:0",
+    ]
 
 
 def test_run_and_exit_oneshot_still_exits_when_global_cleanup_raises(
     monkeypatch, main_mod
 ):
     events = []
 
-    def _raise_mcp():
+    def _raise_mcp(*, final=False):
         raise RuntimeError("mcp boom")
 
     monkeypatch.setitem(

tests/tools/test_mcp_tool.py

@@ -1552,12 +1552,35 @@ def test_no_servers_returns_empty(self):
 # ---------------------------------------------------------------------------
 
 class TestShutdown:
+    def test_final_shutdown_prevents_late_loop_restart(self):
+        """One-shot teardown must not let stale discovery recreate the loop."""
+        import tools.mcp_tool as mcp_mod
+
+        mcp_mod._servers.clear()
+        mcp_mod._mcp_final_shutdown = False
+        try:
+            mcp_mod.shutdown_mcp_servers(final=True)
+
+            assert mcp_mod._ensure_mcp_loop() is False
+            assert mcp_mod.register_mcp_servers(
+                {"late": {"command": "late-discovery"}}
+            ) == []
+            assert mcp_mod._mcp_loop is None
+            assert not mcp_mod._server_connecting
+        finally:
+            mcp_mod._mcp_final_shutdown = False
+
     def test_no_servers_safe(self):
         """shutdown_mcp_servers with no servers does nothing."""
+        import tools.mcp_tool as mcp_mod
         from tools.mcp_tool import shutdown_mcp_servers, _servers
 
         _servers.clear()
         shutdown_mcp_servers()  # Should not raise
+        try:
+            assert mcp_mod._ensure_mcp_loop() is True
+        finally:
+            mcp_mod._stop_mcp_loop()
 
     def test_shutdown_clears_servers(self):
         """shutdown_mcp_servers calls shutdown() on each server and clears dict."""
@@ -1666,6 +1689,103 @@ async def slow_shutdown():
         # Parallel: ~1s, not ~3s. Allow some margin.
         assert elapsed < 2.5, f"Shutdown took {elapsed:.1f}s, expected ~1s (parallel)"
 
+    def test_run_on_loop_schedules_atomically_with_shutdown_state(self):
+        """The lifecycle lock must cover validation and task submission."""
+        import concurrent.futures
+        import threading
+
+        import agent.async_utils as async_utils
+        import tools.mcp_tool as mcp_mod
+
+        loop = MagicMock()
+        loop.is_running.return_value = True
+        mcp_mod._mcp_loop = loop
+        probe_acquired_lock = []
+
+        async def operation():
+            return "ok"
+
+        def observe_schedule_lock(coro, scheduled_loop, **kwargs):
+            del kwargs
+            assert scheduled_loop is loop
+
+            def probe_lock():
+                acquired = mcp_mod._lock.acquire(blocking=False)
+                probe_acquired_lock.append(acquired)
+                if acquired:
+                    mcp_mod._lock.release()
+
+            probe = threading.Thread(target=probe_lock)
+            probe.start()
+            probe.join(timeout=1)
+            coro.close()
+            future = concurrent.futures.Future()
+            future.set_result("ok")
+            return future
+
+        try:
+            with patch.object(
+                async_utils,
+                "safe_schedule_threadsafe",
+                side_effect=observe_schedule_lock,
+            ):
+                assert mcp_mod._run_on_mcp_loop(operation) == "ok"
+        finally:
+            mcp_mod._mcp_loop = None
+
+        assert probe_acquired_lock == [False]
+
+    def test_shutdown_cancels_connecting_server_task_before_loop_close(self):
+        """A one-shot exit must drain an MCP task still connecting in background.
+
+        CLI startup discovers MCP servers on a daemon thread. A fast one-shot
+        can finish before that thread publishes its server into ``_servers``.
+        Closing the loop while ``MCPServerTask.run`` is still pending leaves the
+        coroutine to be finalized after loop close, producing the observed
+        ``RuntimeError: Event loop is closed`` cleanup traceback.
+        """
+        import concurrent.futures
+        import threading
+
+        import tools.mcp_tool as mcp_mod
+        from tools.mcp_tool import MCPServerTask, shutdown_mcp_servers, _servers
+
+        _servers.clear()
+        entered = threading.Event()
+        cleaned_up = threading.Event()
+        server = MCPServerTask("still-connecting")
+        mcp_mod._server_connecting.add(server.name)
+
+        async def wait_forever(self, config):
+            entered.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                # This must execute while the owning loop is still alive.
+                asyncio.get_running_loop().call_soon(cleaned_up.set)
+
+        mcp_mod._ensure_mcp_loop()
+        loop = mcp_mod._mcp_loop
+        assert loop is not None
+        with patch.object(MCPServerTask, "_run_stdio", wait_forever):
+            future = asyncio.run_coroutine_threadsafe(
+                server.run({"command": "test"}), loop
+            )
+            assert entered.wait(timeout=2)
+
+            try:
+                shutdown_mcp_servers()
+            finally:
+                mcp_mod._stop_mcp_loop()
+                mcp_mod._mcp_loop = None
+                mcp_mod._mcp_thread = None
+
+        assert cleaned_up.wait(timeout=1)
+        assert future.done()
+        with pytest.raises(concurrent.futures.CancelledError):
+            future.result()
+        assert not mcp_mod._server_connecting
+
 
 # ---------------------------------------------------------------------------
 # _build_safe_env