fix(mcp): guard against duplicate spawns and stale connecting entries (#58862)

Three fixes for concurrent MCP server spawn races in register_mcp_servers()
and discover_mcp_tools():

1. register_mcp_servers: add k not in _server_connecting guard to the
   new_servers filter. Without this, a concurrent second call sees the
   same servers as 'new' and spawns duplicate stdio subprocesses.

2. discover_mcp_tools: same _server_connecting guard in the
   new_server_names filter. This entry point is called from CLI, TUI,
   gateway, and cron — any two racing would double-spawn.

3. Stale _server_connecting cleanup on TimeoutError/InterruptedError.
   When _run_on_mcp_loop times out or is interrupted, _discover_all's
   gather may not have finished, leaving entries stranded in
   _server_connecting that block future reconnection attempts. The
   cleanup clears only entries added by this call (not external ones),
   logs a warning, and records connect errors.

Salvage of #58879 by @nanami7777777 (superset of #58867 by @liuhao1024).
Adapted to current main which has evolved significantly since July 5.

Closes #58862
Closes #58867
Closes #58879

tests/tools/test_mcp_tool.py

@@ -2454,6 +2454,79 @@ class TestRegisterMcpServers:
         assert "mcp__my_server__tool1" in result
         _servers.pop("my_server", None)
 
+    def test_skips_servers_already_connecting(self):
+        """Servers in _server_connecting must not be spawned again (#58862)."""
+        from tools.mcp_tool import (
+            register_mcp_servers, _servers, _server_connecting, _ensure_mcp_loop,
+        )
+
+        fake_config = {"my_srv": {"command": "npx", "args": ["test"]}}
+
+        # Simulate a prior call that started connecting but hasn't finished
+        _server_connecting.add("my_srv")
+        connect_calls = []
+
+        async def fake_register(name, cfg):
+            connect_calls.append(name)
+            server = _make_mock_server(name)
+            server._registered_tool_names = [f"mcp_{name}_tool"]
+            _servers[name] = server
+            return [f"mcp_{name}_tool"]
+
+        try:
+            with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
+                 patch("tools.mcp_tool._discover_and_register_server", side_effect=fake_register), \
+                 patch("tools.mcp_tool._existing_tool_names", return_value=[]), \
+                 patch("tools.mcp_tool._connect_cooldown_active", return_value=False):
+                _ensure_mcp_loop()
+                result = register_mcp_servers(fake_config)
+
+            # Should NOT have attempted to connect my_srv again
+            assert connect_calls == [], (
+                f"Server already in _server_connecting should be skipped, "
+                f"but connect was called for: {connect_calls}"
+            )
+            assert result == []
+        finally:
+            _server_connecting.discard("my_srv")
+            _servers.pop("my_srv", None)
+
+    def test_clears_stale_connecting_on_timeout(self):
+        """Stale entries in _server_connecting are cleaned up after timeout (#58862)."""
+        from tools.mcp_tool import (
+            register_mcp_servers, _servers, _server_connecting,
+            _server_connect_errors, _ensure_mcp_loop,
+        )
+
+        fake_config = {
+            "srv_a": {"command": "npx", "args": ["a"]},
+            "srv_b": {"command": "npx", "args": ["b"]},
+        }
+
+        # Simulate that srv_a is already connecting from another call
+        _server_connecting.add("srv_a")
+
+        with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
+             patch("tools.mcp_tool._run_on_mcp_loop", side_effect=TimeoutError("timed out")), \
+             patch("tools.mcp_tool._existing_tool_names", return_value=[]), \
+             patch("tools.mcp_tool._connect_cooldown_active", return_value=False):
+            _ensure_mcp_loop()
+
+            with pytest.raises(TimeoutError):
+                register_mcp_servers(fake_config)
+
+        # After timeout, srv_b (which was in new_servers and added to _server_connecting)
+        # should have been cleaned up from _server_connecting.
+        # srv_a should remain since it was added externally and not part of new_servers.
+        assert "srv_b" not in _server_connecting, (
+            "Stale server added during this call should have been removed from "
+            "_server_connecting after timeout"
+        )
+        # Cleanup
+        _server_connecting.discard("srv_a")
+        _servers.pop("srv_a", None)
+        _servers.pop("srv_b", None)
+
 # ---------------------------------------------------------------------------
 # Tests for parallel tool call support (port from openai/codex#17667)
 # ---------------------------------------------------------------------------

tools/mcp_tool.py

@@ -5969,13 +5969,17 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]:
         logger.debug("No explicit MCP servers provided")
         return []
 
-    # Only attempt servers that aren't already connected and are enabled
-    # (enabled: false skips the server entirely without removing its config)
+    # Only attempt servers that aren't already connected (or currently
+    # connecting) and are enabled.  Checking ``_server_connecting`` prevents
+    # duplicate subprocess spawns when ``discover_mcp_tools()`` is called
+    # from multiple entry-points before the first batch finishes (#58862).
     with _lock:
+        connecting = set(_server_connecting)
         new_servers = {
             k: v
             for k, v in servers.items()
             if k not in _servers
+            and k not in connecting
             and _parse_boolish(v.get("enabled", True), default=True)
             # Skip a server still serving its post-failure backoff. Without
             # this, a server that fails to connect (and is therefore never
@@ -6061,6 +6065,28 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]:
         _set_interrupt(False)
     try:
         _run_on_mcp_loop(_discover_all, timeout=120)
+    except (TimeoutError, InterruptedError) as _e:
+        # When the outer timeout fires or the user interrupts,
+        # _discover_all's gather may not have finished, leaving
+        # entries stranded in _server_connecting.  Those stale
+        # entries would block future reconnection attempts (#58862).
+        with _lock:
+            stale = [n for n in new_servers if n in _server_connecting]
+            if stale:
+                logger.warning(
+                    "MCP discovery %s while %d server(s) were still "
+                    "connecting; clearing stale connecting set: %s",
+                    "timed out" if isinstance(_e, TimeoutError) else "interrupted",
+                    len(stale),
+                    ", ".join(stale),
+                )
+                _server_connecting.difference_update(stale)
+                for _sn in stale:
+                    _server_connect_errors.setdefault(
+                        _sn,
+                        f"Connection attempt {'timed out' if isinstance(_e, TimeoutError) else 'interrupted'} during discovery",
+                    )
+        raise
     finally:
         if _was_interrupted:
             _set_interrupt(True)
@@ -6133,10 +6159,13 @@ def discover_mcp_tools() -> List[str]:
 
     try:
         with _lock:
+            connecting = set(_server_connecting)
             new_server_names = [
                 name
                 for name, cfg in servers.items()
-                if name not in _servers and _parse_boolish(cfg.get("enabled", True), default=True)
+                if name not in _servers
+                and name not in connecting
+                and _parse_boolish(cfg.get("enabled", True), default=True)
             ]
 
         tool_names = register_mcp_servers(servers)