diff --git a/tests/tools/test_mcp_discovery_cross_process.py b/tests/tools/test_mcp_discovery_cross_process.py new file mode 100644 index 00000000000..f5be704ea7a --- /dev/null +++ b/tests/tools/test_mcp_discovery_cross_process.py @@ -0,0 +1,187 @@ +"""Cross-process regression coverage for MCP discovery serialization. + +Two independent Hermes processes can start MCP discovery at the same time +(dashboard and gateway startup). The losing process must wait for the shared +lock and then perform its own local discovery; another process's registry is +not usable because ``_servers`` is process-local. +""" + +import json +import os +from pathlib import Path +import subprocess +import sys +import textwrap +import time + + +_REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _wait_for_file(path: Path, *, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if path.exists(): + return + time.sleep(0.01) + raise AssertionError(f"timed out waiting for {path}") + + +def test_two_processes_each_complete_local_mcp_discovery(tmp_path): + """A lock loser waits, acquires the lock, and builds its own registry.""" + hermes_home = tmp_path / "hermes-home" + hermes_home.mkdir() + + holder_ready = tmp_path / "holder-ready" + release_holder = tmp_path / "release-holder" + loser_started = tmp_path / "loser-started" + holder_output = tmp_path / "holder.json" + loser_output = tmp_path / "loser.json" + child_script = tmp_path / "mcp-discovery-child.py" + + child_script.write_text( + textwrap.dedent( + """ + import json + import os + from pathlib import Path + import sys + import time + from types import SimpleNamespace + + repo_root, role, ready_arg, release_arg, started_arg, output_arg = sys.argv[1:] + sys.path.insert(0, repo_root) + + import tools.mcp_tool as mcp_tool + + ready = Path(ready_arg) + release = Path(release_arg) + started = Path(started_arg) + output = Path(output_arg) + + mcp_tool._MCP_AVAILABLE = True + mcp_tool._MCP_DISCOVERY_LOCK_PATH = None + mcp_tool._MCP_DISCOVERY_LOCK_MAX_RETRIES = 200 + mcp_tool._MCP_DISCOVERY_LOCK_RETRY_DELAY_S = 0.01 + mcp_tool._servers.clear() + + config = { + "test_srv": { + "command": "fake", + "enabled": True, + } + } + mcp_tool._load_mcp_config = lambda: config + + def fake_register_mcp_servers(servers): + tool_name = "mcp__test_srv__ping" + mcp_tool._servers["test_srv"] = SimpleNamespace( + _registered_tool_names=[tool_name], + ) + + if role == "holder": + ready.write_text("1", encoding="utf-8") + deadline = time.monotonic() + 10.0 + while not release.exists(): + if time.monotonic() >= deadline: + raise RuntimeError("holder release signal timed out") + time.sleep(0.01) + + return [tool_name] + + mcp_tool.register_mcp_servers = fake_register_mcp_servers + started.write_text("1", encoding="utf-8") + + result = mcp_tool.discover_mcp_tools() + server = mcp_tool._servers.get("test_srv") + output.write_text( + json.dumps( + { + "pid": os.getpid(), + "result": result, + "server_present": server is not None, + "registered_tools": ( + list(server._registered_tool_names) + if server is not None + else [] + ), + } + ), + encoding="utf-8", + ) + """ + ), + encoding="utf-8", + ) + + env = os.environ.copy() + env["HERMES_HOME"] = str(hermes_home) + + holder_started = tmp_path / "holder-started" + holder = subprocess.Popen( + [ + sys.executable, + str(child_script), + str(_REPO_ROOT), + "holder", + str(holder_ready), + str(release_holder), + str(holder_started), + str(holder_output), + ], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + loser = None + + try: + _wait_for_file(holder_ready) + + loser = subprocess.Popen( + [ + sys.executable, + str(child_script), + str(_REPO_ROOT), + "loser", + str(holder_ready), + str(release_holder), + str(loser_started), + str(loser_output), + ], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + _wait_for_file(loser_started) + + # The loser must not finish from an empty process-local registry while + # the holder owns the lock. + time.sleep(0.1) + assert loser.poll() is None + assert not loser_output.exists() + + release_holder.write_text("1", encoding="utf-8") + + holder_stdout, holder_stderr = holder.communicate(timeout=15) + loser_stdout, loser_stderr = loser.communicate(timeout=15) + assert holder.returncode == 0, holder_stdout + holder_stderr + assert loser.returncode == 0, loser_stdout + loser_stderr + finally: + release_holder.touch(exist_ok=True) + for process in (holder, loser): + if process is not None and process.poll() is None: + process.terminate() + process.wait(timeout=5) + + expected_tools = ["mcp__test_srv__ping"] + holder_result = json.loads(holder_output.read_text(encoding="utf-8")) + loser_result = json.loads(loser_output.read_text(encoding="utf-8")) + + assert holder_result["pid"] != loser_result["pid"] + for result in (holder_result, loser_result): + assert result["result"] == expected_tools + assert result["server_present"] is True + assert result["registered_tools"] == expected_tools diff --git a/tests/tools/test_mcp_tool.py b/tests/tools/test_mcp_tool.py index 7e5b40c1cc3..dc8e532c0b7 100644 --- a/tests/tools/test_mcp_tool.py +++ b/tests/tools/test_mcp_tool.py @@ -6,6 +6,7 @@ All tests use mocks -- no real MCP servers or subprocesses are started. import asyncio import concurrent.futures import json +import os import threading import time from types import SimpleNamespace @@ -4608,3 +4609,290 @@ class TestMcpParallelToolCalls: register_mcp_servers(config_off) with _lock: assert sanitize_mcp_name_component("toggle_srv") not in _parallel_safe_servers + + +# --------------------------------------------------------------------------- +# Cross-process MCP discovery lock (issue #62771) +# --------------------------------------------------------------------------- + + +class TestMCPDiscoveryCrossProcessLock: + """Tests for the cross-process MCP discovery guard in discover_mcp_tools().""" + + @pytest.fixture(autouse=True) + def _fast_retries(self): + """Override retry constants so tests are fast.""" + import tools.mcp_tool as mcp_tool + orig_max = mcp_tool._MCP_DISCOVERY_LOCK_MAX_RETRIES + orig_delay = mcp_tool._MCP_DISCOVERY_LOCK_RETRY_DELAY_S + mcp_tool._MCP_DISCOVERY_LOCK_MAX_RETRIES = 3 + mcp_tool._MCP_DISCOVERY_LOCK_RETRY_DELAY_S = 0.01 + yield + mcp_tool._MCP_DISCOVERY_LOCK_MAX_RETRIES = orig_max + mcp_tool._MCP_DISCOVERY_LOCK_RETRY_DELAY_S = orig_delay + + def test_lock_acquired_path(self, tmp_path): + """Lock acquired -> discovery runs normally, lock released at end.""" + from tools.mcp_tool import ( + _LockCookie, + discover_mcp_tools, + ) + + lock_file = tmp_path / ".mcp-discovery.lock" + fh = open(lock_file, "w", encoding="utf-8") + cookie = _LockCookie(fh) + + def mock_acquire(): + return cookie + + mock_config = {"test_srv": {"command": "echo", "enabled": True}} + with patch.object(cookie, "release", wraps=cookie.release) as release_spy: + with patch("tools.mcp_tool._try_acquire_mcp_discovery_lock", mock_acquire), \ + patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._load_mcp_config", return_value=mock_config), \ + patch("tools.mcp_tool.register_mcp_servers", return_value=["mcp__test_srv__ping"]) as reg_spy: + result = discover_mcp_tools() + assert result == ["mcp__test_srv__ping"] + release_spy.assert_called_once() + + def test_lock_held_retries_then_acquires(self): + """First attempt sees lock held; retry succeeds; then discovery runs.""" + from tools.mcp_tool import ( + _LOCK_UNAVAILABLE, + discover_mcp_tools, + ) + + import tempfile + import portalocker + from tools.mcp_tool import _LockCookie + + lock_path = [None] + call_count = [0] + + def mock_acquire(): + call_count[0] += 1 + if call_count[0] <= 1: + return None # first call: lock held + # build a real cookie so release() works + tf = tempfile.NamedTemporaryFile( + prefix="mcp-lock-", suffix=".tmp", delete=False + ) + lock_path[0] = tf.name + portalocker.lock(tf, portalocker.LOCK_EX | portalocker.LOCK_NB) + return _LockCookie(tf) + + mock_config = {"test_srv": {"command": "echo", "enabled": True}} + try: + with patch("tools.mcp_tool._try_acquire_mcp_discovery_lock", mock_acquire), \ + patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._load_mcp_config", return_value=mock_config), \ + patch("tools.mcp_tool.register_mcp_servers", return_value=["mcp__test_srv__ping"]) as reg_spy: + result = discover_mcp_tools() + assert result == ["mcp__test_srv__ping"] + # register_mcp_servers must be called (local discovery ran) + reg_spy.assert_called_once_with(mock_config) + finally: + if lock_path[0]: + try: + os.unlink(lock_path[0]) + except Exception: + pass + + def test_lock_held_retries_exhausted_fallback(self): + """All retry attempts see lock held -> runs discovery unguarded.""" + from tools.mcp_tool import ( + _LOCK_UNAVAILABLE, + discover_mcp_tools, + _MCP_DISCOVERY_LOCK_MAX_RETRIES, + ) + + mock_config = {"test_srv": {"command": "echo", "enabled": True}} + # Every attempt returns None (lock held) + with patch("tools.mcp_tool._try_acquire_mcp_discovery_lock", return_value=None), \ + patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._load_mcp_config", return_value=mock_config), \ + patch("tools.mcp_tool.register_mcp_servers") as reg_spy, \ + patch("tools.mcp_tool._existing_tool_names", return_value=[]): + result = discover_mcp_tools() + # Must still run local discovery + reg_spy.assert_called_once_with(mock_config) + + def test_lock_unavailable_fallback(self): + """Lock unavailable/broken -> run discovery unguarded (no retry).""" + from tools.mcp_tool import ( + _LOCK_UNAVAILABLE, + discover_mcp_tools, + ) + + mock_config = {"test_srv": {"command": "echo", "enabled": True}} + with patch("tools.mcp_tool._try_acquire_mcp_discovery_lock", return_value=_LOCK_UNAVAILABLE), \ + patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._load_mcp_config", return_value=mock_config), \ + patch("tools.mcp_tool.register_mcp_servers") as reg_spy, \ + patch("tools.mcp_tool._existing_tool_names", return_value=[]): + result = discover_mcp_tools() + reg_spy.assert_called_once_with(mock_config) + + def test_windows_portalocker_handle_lifetime(self): + """_LockCookie keeps file handle alive until release().""" + import tempfile + + import portalocker + from tools.mcp_tool import _LockCookie + + with tempfile.NamedTemporaryFile(prefix="mcp-lock-", suffix=".tmp", delete=False) as tf: + lock_path = tf.name + + try: + fh = open(lock_path, "w", encoding="utf-8") + portalocker.lock(fh, portalocker.LOCK_EX | portalocker.LOCK_NB) + cookie = _LockCookie(fh) + assert not fh.closed + fno = fh.fileno() + assert fno > 0 + cookie.release() + assert fh.closed + finally: + try: + os.unlink(lock_path) + except Exception: + pass + + def test_double_release_safety(self): + """Calling release() twice is safe (no exception).""" + import tempfile + + import portalocker + from tools.mcp_tool import _LockCookie + + with tempfile.NamedTemporaryFile(prefix="mcp-lock-", suffix=".tmp", delete=False) as tf: + lock_path = tf.name + + try: + fh = open(lock_path, "w", encoding="utf-8") + portalocker.lock(fh, portalocker.LOCK_EX | portalocker.LOCK_NB) + cookie = _LockCookie(fh) + cookie.release() + assert fh.closed + # Second release -- must not raise + cookie.release() + finally: + try: + os.unlink(lock_path) + except Exception: + pass + + def test_posix_flock_acquire_and_release(self): + """_acquire_lock_on_fh uses fcntl.flock on POSIX.""" + import sys + import tempfile + from unittest.mock import MagicMock + + mock_fcntl = MagicMock() + mock_fcntl.LOCK_EX = 2 + mock_fcntl.LOCK_NB = 4 + + with tempfile.NamedTemporaryFile(prefix="mcp-lock-", suffix=".tmp", delete=False) as tf: + lock_path = tf.name + + try: + fh = open(lock_path, "w", encoding="utf-8") + with patch.dict("sys.modules", {"fcntl": mock_fcntl}), \ + patch("tools.mcp_tool.os.name", "posix"): + from tools.mcp_tool import _acquire_lock_on_fh + result = _acquire_lock_on_fh(fh) + assert result is True + mock_fcntl.flock.assert_called_once_with( + fh.fileno(), mock_fcntl.LOCK_EX | mock_fcntl.LOCK_NB + ) + fh.close() + finally: + try: + os.unlink(lock_path) + except Exception: + pass + + def test_posix_flock_oserror_eagain_returns_false(self): + """POSIX fcntl.flock raising OSError(EAGAIN) -> return False (lock held).""" + import errno + import tempfile + from unittest.mock import MagicMock, patch + + mock_fcntl = MagicMock() + mock_fcntl.LOCK_EX = 2 + mock_fcntl.LOCK_NB = 4 + mock_fcntl.flock.side_effect = OSError(errno.EAGAIN, "Resource temporarily unavailable") + + with tempfile.NamedTemporaryFile(prefix="mcp-lock-", suffix=".tmp", delete=False) as tf: + lock_path = tf.name + + try: + fh = open(lock_path, "w", encoding="utf-8") + with patch.dict("sys.modules", {"fcntl": mock_fcntl}), \ + patch("tools.mcp_tool.os.name", "posix"): + from tools.mcp_tool import _acquire_lock_on_fh + result = _acquire_lock_on_fh(fh) + assert result is False + fh.close() + finally: + try: + os.unlink(lock_path) + except Exception: + pass + + def test_two_concurrent_discovery_attempts(self): + """Two sequential calls both end up with a non-empty registry + (no early empty return). Each call gets its own cookie directly + on first acquire attempt.""" + from tools.mcp_tool import ( + _LOCK_UNAVAILABLE, + _LockCookie, + discover_mcp_tools, + ) + + mock_config = {"test_srv": {"command": "echo", "enabled": True}} + + # Build two real cookie handles so release() works + import tempfile + import portalocker + tf1 = tempfile.NamedTemporaryFile( + prefix="mcp-lock-1-", suffix=".tmp", delete=False + ) + tf2 = tempfile.NamedTemporaryFile( + prefix="mcp-lock-2-", suffix=".tmp", delete=False + ) + lock_path1 = tf1.name + lock_path2 = tf2.name + cookie1 = _LockCookie(tf1) + cookie2 = _LockCookie(tf2) + portalocker.lock(tf1, portalocker.LOCK_EX | portalocker.LOCK_NB) + portalocker.lock(tf2, portalocker.LOCK_EX | portalocker.LOCK_NB) + + def make_sequencer(): + state = {"call": 0, "cookie1": cookie1, "cookie2": cookie2} + def seq(): + state["call"] += 1 + if state["call"] == 1: + return state["cookie1"] + else: + return state["cookie2"] + return seq + + seq_fn = make_sequencer() + + try: + with patch("tools.mcp_tool._try_acquire_mcp_discovery_lock", side_effect=seq_fn), \ + patch("tools.mcp_tool._MCP_AVAILABLE", True), \ + patch("tools.mcp_tool._load_mcp_config", return_value=mock_config), \ + patch("tools.mcp_tool.register_mcp_servers", return_value=["mcp__test_srv__ping"]): + r1 = discover_mcp_tools() + r2 = discover_mcp_tools() + + assert r1 == ["mcp__test_srv__ping"] + assert r2 == ["mcp__test_srv__ping"] + finally: + for path in (lock_path1, lock_path2): + try: + os.unlink(path) + except Exception: + pass diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index 28dd5e91741..dea31592469 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -4096,6 +4096,127 @@ _mcp_thread: Optional[threading.Thread] = None # _parallel_safe_servers, _mcp_tool_server_names, and _stdio_pids. _lock = threading.Lock() +# --------------------------------------------------------------------------- +# Cross-process MCP discovery guard +# --------------------------------------------------------------------------- +# Advisory file lock that prevents N concurrent Hermes processes (e.g. +# gateway + CLI + TUI) from all running MCP discovery simultaneously. +# See issue #62771. +_LOCK_UNAVAILABLE: Any = object() # sentinel: locking broken/unavailable +_MCP_DISCOVERY_LOCK_PATH: Optional[str] = None # resolved lazily + +# Retry constants for the bounded wait when another process holds the lock. +_MCP_DISCOVERY_LOCK_MAX_RETRIES: int = 10 +_MCP_DISCOVERY_LOCK_RETRY_DELAY_S: float = 0.2 + + +class _LockCookie: + """Holds a cross-process file lock; release() drops it. + + On Windows the underlying file handle MUST stay alive while the lock is + held (portalocker keeps the kernel lock on the fd). On POSIX the fcntl + lockdown is similarly tied to the file-descriptor lifetime. We keep the + file object in ``_fh`` and close it on release. + """ + + def __init__(self, fh: Any) -> None: + self._fh = fh + + def release(self) -> None: + if self._fh is not None: + try: + fd = self._fh.fileno() + if os.name == "posix": + import fcntl + try: + fcntl.flock(fd, fcntl.LOCK_UN) + except Exception: + pass + else: + import portalocker + try: + portalocker.unlock(self._fh) + except Exception: + pass + except Exception: + pass + try: + self._fh.close() + except Exception: + pass + self._fh = None + + +def _acquire_lock_on_fh(fh: Any) -> bool: + """Acquire a non-blocking exclusive lock on an open file handle. + + Uses ``fcntl.flock`` on POSIX and ``portalocker.lock`` on Windows. + + Returns ``True`` if the lock was acquired, ``False`` if another process + holds it (non-blocking refusal). Raises ``RuntimeError`` on unexpected + errors so the caller can treat lock acquisition as unavailable. + """ + fd = fh.fileno() + if os.name == "posix": + import fcntl + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError as e: + if e.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK): + return False + raise + else: + import portalocker + try: + portalocker.lock(fh, portalocker.LOCK_EX | portalocker.LOCK_NB) + return True + except portalocker.LockException: + return False + + +def _try_acquire_mcp_discovery_lock() -> Any: + """Try to acquire an exclusive cross-process lock for MCP discovery. + + Returns + ------- + _LockCookie + Lock acquired successfully. + None + Another process holds the lock (non-blocking refusal). + _LOCK_UNAVAILABLE + Locking mechanism is broken or unavailable -- caller should run + discovery unguarded. + """ + global _MCP_DISCOVERY_LOCK_PATH + try: + from hermes_constants import get_hermes_home + if _MCP_DISCOVERY_LOCK_PATH is None: + _MCP_DISCOVERY_LOCK_PATH = str( + get_hermes_home() / ".mcp-discovery.lock" + ) + lock_path = _MCP_DISCOVERY_LOCK_PATH + except Exception: + return _LOCK_UNAVAILABLE + + try: + fh = open(lock_path, "w", encoding="utf-8") + except Exception: + return _LOCK_UNAVAILABLE + + try: + acquired = _acquire_lock_on_fh(fh) + except Exception: + fh.close() + return _LOCK_UNAVAILABLE + + if acquired: + return _LockCookie(fh) + else: + fh.close() + return None + + # PIDs of stdio MCP server subprocesses. Tracked so we can force-kill # them on shutdown if the graceful cleanup (SDK context-manager teardown) # fails or times out. PIDs are added after connection and removed on @@ -5785,33 +5906,61 @@ def discover_mcp_tools() -> List[str]: logger.debug("No MCP servers configured") return [] - with _lock: - new_server_names = [ - name - for name, cfg in servers.items() - if name not in _servers and _parse_boolish(cfg.get("enabled", True), default=True) - ] + # Cross-process discovery guard (#62771). A lock loser waits for + # the holder, then performs its own process-local discovery. If locking is + # unavailable or the bounded wait expires, preserve the previous + # fail-soft behavior by running discovery unguarded. + cookie = _try_acquire_mcp_discovery_lock() + if cookie is None: + logger.debug( + "Another process holds MCP discovery lock -- retrying with backoff" + ) + for _ in range(_MCP_DISCOVERY_LOCK_MAX_RETRIES): + time.sleep(_MCP_DISCOVERY_LOCK_RETRY_DELAY_S) + cookie = _try_acquire_mcp_discovery_lock() + if cookie is not None: + break + + if cookie is None: + logger.warning( + "MCP discovery lock still held after %d retries -- " + "running discovery unguarded", + _MCP_DISCOVERY_LOCK_MAX_RETRIES, + ) + elif cookie is not _LOCK_UNAVAILABLE: + logger.debug("Retry succeeded -- acquired MCP discovery lock") + + try: + with _lock: + new_server_names = [ + name + for name, cfg in servers.items() + if name not in _servers and _parse_boolish(cfg.get("enabled", True), default=True) + ] + + tool_names = register_mcp_servers(servers) + if not new_server_names: + return tool_names + + with _lock: + connected_server_names = [name for name in new_server_names if name in _servers] + new_tool_count = sum( + len(getattr(_servers[name], "_registered_tool_names", [])) + for name in connected_server_names + ) + + failed_count = len(new_server_names) - len(connected_server_names) + if new_tool_count or failed_count: + summary = f" MCP: {new_tool_count} tool(s) from {len(connected_server_names)} server(s)" + if failed_count: + summary += f" ({failed_count} failed)" + logger.info(summary) - tool_names = register_mcp_servers(servers) - if not new_server_names: return tool_names - with _lock: - connected_server_names = [name for name in new_server_names if name in _servers] - new_tool_count = sum( - len(getattr(_servers[name], "_registered_tool_names", [])) - for name in connected_server_names - ) - - failed_count = len(new_server_names) - len(connected_server_names) - if new_tool_count or failed_count: - summary = f" MCP: {new_tool_count} tool(s) from {len(connected_server_names)} server(s)" - if failed_count: - summary += f" ({failed_count} failed)" - logger.info(summary) - - return tool_names - + finally: + if cookie not in (None, _LOCK_UNAVAILABLE): + cookie.release() def is_mcp_tool_parallel_safe(tool_name: str) -> bool: """Check if an MCP tool belongs to a server that supports parallel tool calls.