diff --git a/tests/tools/test_mcp_failure_classification.py b/tests/tools/test_mcp_failure_classification.py new file mode 100644 index 000000000000..2d6f029102ff --- /dev/null +++ b/tests/tools/test_mcp_failure_classification.py @@ -0,0 +1,232 @@ +"""Tests for exception-group unwrapping and failure classification in +``tools/mcp_tool.py`` (#65673, #66092). + +The MCP SDK's anyio TaskGroups wrap real errors in ``BaseExceptionGroup``, +whose ``str()`` is "unhandled errors in a TaskGroup (N sub-exceptions)" — +useless in logs. ``_unwrap_exception_group`` digs out the root cause; +``_classify_mcp_failure`` decides whether a failure is worth retrying. +""" + +import asyncio +import errno +import logging + +import pytest + +from tools.mcp_tool import ( + InvalidMcpUrlError, + MCPServerTask, + NonMcpEndpointError, + _classify_mcp_failure, + _unwrap_exception_group, +) + + +def _group(*excs, msg="unhandled errors in a TaskGroup") -> BaseExceptionGroup: + return BaseExceptionGroup(msg, list(excs)) + + +# ── _unwrap_exception_group ────────────────────────────────────────────────── + +class TestUnwrapExceptionGroup: + def test_plain_exception_passes_through(self): + exc = ConnectionError("boom") + assert _unwrap_exception_group(exc) is exc + + def test_single_level_group(self): + inner = BrokenPipeError() + assert _unwrap_exception_group(_group(inner)) is inner + + def test_nested_groups(self): + inner = ConnectionResetError("reset by peer") + nested = _group(_group(_group(inner))) + assert _unwrap_exception_group(nested) is inner + + def test_root_cause_name_visible_for_empty_message(self): + # Dead stdio pipes raise BrokenPipeError with an EMPTY str() — + # the log format must rely on type(exc).__name__, and unwrap must + # hand back the BrokenPipeError, not the opaque group. + root = _unwrap_exception_group(_group(BrokenPipeError())) + assert type(root).__name__ == "BrokenPipeError" + + def test_prefers_non_cancellation_leaf(self): + # anyio cancellation sprays CancelledError across sibling tasks; + # the real error must win. + real = ConnectionError("server hung up") + g = _group(asyncio.CancelledError(), real, asyncio.CancelledError()) + assert _unwrap_exception_group(g) is real + + def test_prefers_non_cancellation_leaf_nested(self): + real = TimeoutError("read timed out") + g = _group(_group(asyncio.CancelledError()), _group(real)) + assert _unwrap_exception_group(g) is real + + def test_all_cancellation_returns_cancellation(self): + g = _group(asyncio.CancelledError()) + assert isinstance(_unwrap_exception_group(g), asyncio.CancelledError) + + def test_keyboard_interrupt_reraises(self): + with pytest.raises(KeyboardInterrupt): + _unwrap_exception_group(_group(KeyboardInterrupt())) + + def test_nested_keyboard_interrupt_reraises(self): + with pytest.raises(KeyboardInterrupt): + _unwrap_exception_group( + _group(ConnectionError("x"), _group(KeyboardInterrupt())) + ) + + def test_system_exit_reraises(self): + with pytest.raises(SystemExit): + _unwrap_exception_group(_group(SystemExit(2))) + + +# ── _classify_mcp_failure ──────────────────────────────────────────────────── + +class TestClassifyMcpFailure: + @pytest.mark.parametrize("exc", [ + ConnectionError("connection refused"), + ConnectionResetError("reset by peer"), + BrokenPipeError(), + EOFError(), + TimeoutError("read timeout"), + OSError(errno.ECONNRESET, "reset"), + RuntimeError("something odd"), + ]) + def test_transient_failures(self, exc): + assert _classify_mcp_failure(exc) == "transient" + + def test_transient_taskgroup_drop(self): + g = _group(ConnectionError("sse stream dropped")) + assert _classify_mcp_failure(g) == "transient" + + def test_closed_resource_transient(self): + anyio = pytest.importorskip("anyio") + assert _classify_mcp_failure(anyio.ClosedResourceError()) == "transient" + + @pytest.mark.parametrize("exc_factory", [ + lambda: FileNotFoundError("no such file: nonexistent-mcp-cmd"), + lambda: OSError(errno.ENOENT, "No such file or directory"), + lambda: NonMcpEndpointError("url serves text/html"), + lambda: InvalidMcpUrlError("bad scheme"), + ]) + def test_permanent_failures(self, exc_factory): + assert _classify_mcp_failure(exc_factory()) == "permanent" + + @pytest.mark.parametrize("status", [401, 403]) + def test_http_auth_status_permanent(self, status): + httpx = pytest.importorskip("httpx") + req = httpx.Request("POST", "http://x/mcp") + resp = httpx.Response(status, request=req) + exc = httpx.HTTPStatusError("auth", request=req, response=resp) + assert _classify_mcp_failure(exc) == "permanent" + + def test_http_5xx_transient(self): + httpx = pytest.importorskip("httpx") + req = httpx.Request("POST", "http://x/mcp") + resp = httpx.Response(503, request=req) + exc = httpx.HTTPStatusError("unavailable", request=req, response=resp) + assert _classify_mcp_failure(exc) == "transient" + + def test_permanent_inside_taskgroup(self): + # Classification must apply to the UNWRAPPED root cause. + g = _group(_group(FileNotFoundError("cmd not found"))) + assert _classify_mcp_failure(g) == "permanent" + + +# ── Keepalive failure log surfaces the root cause ──────────────────────────── + +@pytest.mark.no_isolate +def test_keepalive_failure_logs_root_cause(monkeypatch, tmp_path, caplog): + """A keepalive that dies with a TaskGroup-wrapped BrokenPipeError (empty + str) must log 'BrokenPipeError', not 'unhandled errors in a TaskGroup'.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + + class _Task(MCPServerTask): + async def _keepalive_probe(self): + raise _group(BrokenPipeError()) + + task = _Task("pipey") + task._config = {"keepalive_interval": 0.01} + task.session = object() + + monkeypatch.setattr(mcp_tool, "_MIN_KEEPALIVE_INTERVAL", 0.01) + + async def _scenario(): + with caplog.at_level(logging.WARNING, logger="tools.mcp_tool"): + reason = await task._wait_for_lifecycle_event() + assert reason == "reconnect" + + asyncio.run(_scenario()) + + keepalive_logs = [ + r.getMessage() for r in caplog.records if "keepalive failed" in r.getMessage() + ] + assert keepalive_logs, "keepalive failure was not logged" + assert any("BrokenPipeError" in m for m in keepalive_logs), keepalive_logs + assert not any("unhandled errors in a TaskGroup" in m for m in keepalive_logs) + + +# ── run() parks permanent failures immediately ─────────────────────────────── + +@pytest.mark.no_isolate +def test_permanent_failure_parks_without_retry_ladder(monkeypatch, tmp_path, caplog): + """A stdio command that doesn't exist (FileNotFoundError) must park after + ONE attempt — not burn _MAX_INITIAL_CONNECT_RETRIES identical warnings.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + + _real_sleep = asyncio.sleep + + async def _fast_sleep(_delay, *a, **kw): + await _real_sleep(0) + + monkeypatch.setattr(mcp_tool.asyncio, "sleep", _fast_sleep) + + state = {"transport_calls": 0, "parked": False} + + async def _scenario(): + class _Task(MCPServerTask): + def _is_http(self): + return False + + def _deregister_tools(self): + state["parked"] = True + self._registered_tool_names = [] + + async def _run_stdio(self, config): + state["transport_calls"] += 1 + raise FileNotFoundError("nonexistent-mcp-command") + + task = _Task("missing-cmd") + + with caplog.at_level(logging.DEBUG, logger="tools.mcp_tool"): + run_task = asyncio.ensure_future(task.run({"command": "nope"})) + for _ in range(500): + await _real_sleep(0) + if state["parked"]: + break + + assert state["parked"], "permanent failure never parked" + assert state["transport_calls"] == 1, ( + f"permanent failure burned {state['transport_calls']} attempts — " + "should park immediately" + ) + + task._shutdown_event.set() + task._reconnect_event.set() + try: + await asyncio.wait_for(run_task, timeout=15) + except (asyncio.TimeoutError, asyncio.CancelledError, Exception): + run_task.cancel() + + asyncio.run(_scenario()) + + park_warnings = [ + r for r in caplog.records + if r.levelno == logging.WARNING and "permanent error" in r.getMessage() + ] + assert len(park_warnings) == 1 + assert "FileNotFoundError" in park_warnings[0].getMessage() diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index 58074668c6dd..7eea74d5f6f4 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -92,6 +92,7 @@ Thread safety: import asyncio import contextvars import concurrent.futures +import errno import inspect import json import logging @@ -947,6 +948,85 @@ class NonMcpEndpointError(ConnectionError): """ +def _unwrap_exception_group(exc: BaseException) -> BaseException: + """Extract the root-cause exception from anyio TaskGroup wrappers. + + The MCP SDK uses anyio task groups, which wrap errors in + ``BaseExceptionGroup`` / ``ExceptionGroup``. Their ``str()`` is opaque — + "unhandled errors in a TaskGroup (1 sub-exception)" — so log sites must + unwrap to surface the real cause (e.g. ``BrokenPipeError`` on a dead + stdio pipe, "401 Unauthorized" on an auth failure). + + Adapted from :func:`hermes_cli.mcp_config._unwrap_exception_group` with + two extra behaviours needed on the runtime path: + + - **Fatal leaves re-raise.** A ``KeyboardInterrupt`` / ``SystemExit`` + anywhere in the (possibly nested) group must propagate to the + interpreter, never be flattened into a loggable error. + - **Prefer non-cancellation leaves.** When a group carries both a real + error and the ``CancelledError``s that anyio cancellation sprays across + sibling tasks, the real error is the root cause worth logging. + """ + while isinstance(exc, BaseExceptionGroup) and exc.exceptions: + fatal, _rest = exc.split((KeyboardInterrupt, SystemExit)) + if fatal is not None: + # Surface the fatal signal itself, not the wrapper. + leaf: BaseException = fatal + while isinstance(leaf, BaseExceptionGroup) and leaf.exceptions: + leaf = leaf.exceptions[0] + raise leaf + # Prefer a non-cancellation leaf when one exists: cancellation + # noise from sibling tasks should not mask the real error. + chosen = exc.exceptions[0] + for sub in exc.exceptions: + if not _contains_only_cancellation(sub): + chosen = sub + break + exc = chosen + return exc + + +def _contains_only_cancellation(exc: BaseException) -> bool: + """True if ``exc`` is (or a group containing only) CancelledError.""" + if isinstance(exc, BaseExceptionGroup): + return all(_contains_only_cancellation(sub) for sub in exc.exceptions) + return isinstance(exc, asyncio.CancelledError) + + +def _classify_mcp_failure(exc: BaseException) -> str: + """Classify an MCP connection failure as ``'permanent'`` or ``'transient'``. + + Permanent failures are deterministic — every retry hits the same wall, so + burning the retry ladder (and log lines) on them is pure noise; ``run()`` + parks them immediately: + + - auth failures (401/403) — need new credentials, not a retry; + - :class:`NonMcpEndpointError` — the URL serves a web page, not MCP; + - :class:`InvalidMcpUrlError` — unusable config; + - ``FileNotFoundError`` / ``ENOENT`` — the stdio command doesn't exist. + + Everything else (network blips, EOF, ``ClosedResourceError``, transport + TaskGroup drops, timeouts) is transient and keeps the normal + retry-with-backoff ladder. + """ + root = _unwrap_exception_group(exc) + if _is_auth_error(root): + return "permanent" + if isinstance(root, (NonMcpEndpointError, InvalidMcpUrlError)): + return "permanent" + # Stdio command missing: FileNotFoundError, or an OSError carrying ENOENT. + if isinstance(root, FileNotFoundError): + return "permanent" + if isinstance(root, OSError) and getattr(root, "errno", None) == errno.ENOENT: + return "permanent" + # httpx.HTTPStatusError with 401/403 that _is_auth_error's type-gate + # missed (e.g. auth types not importable in this environment). + status = getattr(getattr(root, "response", None), "status_code", None) + if status in (401, 403): + return "permanent" + return "transient" + + def _validate_remote_mcp_url(server_name: str, url: Any) -> str: """Return the URL as a string if it's a valid http(s) remote MCP URL. @@ -2178,10 +2258,11 @@ class MCPServerTask: try: await self._keepalive_probe() except Exception as exc: + root = _unwrap_exception_group(exc) logger.warning( "MCP server '%s' keepalive failed, " - "triggering reconnect: %s", - self.name, exc, + "triggering reconnect: %s: %s", + self.name, type(root).__name__, root, ) self._reconnect_event.set() break @@ -3052,7 +3133,7 @@ class MCPServerTask: ) if parked == "shutdown": break - logger.debug( + logger.info( "MCP server '%s': attempting revival from parked " "state (self-probe or explicit reconnect request); " "rebuilding transport.", @@ -3083,11 +3164,19 @@ class MCPServerTask: raise except Exception as exc: self.session = None + # Unwrap anyio TaskGroup wrappers first: str(exc) on a + # BaseExceptionGroup is "unhandled errors in a TaskGroup + # (N sub-exceptions)" — useless in logs, and it hides the + # root cause from the auth/permanence classification below. + # Empty dead-pipe errors still get a name this way + # (e.g. "BrokenPipeError: "). + root = _unwrap_exception_group(exc) + failure_class = _classify_mcp_failure(root) if self._is_recycled_stdio(): logger.warning( "MCP server '%s': lazy reconnect after stdio recycle " - "failed, marking unavailable while retrying: %s", - self.name, exc, + "failed, marking unavailable while retrying: %s: %s", + self.name, type(root).__name__, root, ) self._recycled_reason = None @@ -3096,25 +3185,62 @@ class MCPServerTask: # should not permanently kill the server. # (Ported from Kilo Code's MCP resilience fix.) if not self._ready.is_set(): - if _is_auth_error(exc): + if _is_auth_error(root): logger.warning( "MCP server '%s' failed initial OAuth authentication, " - "not retrying automatically: %s", - self.name, exc, + "not retrying automatically: %s: %s", + self.name, type(root).__name__, root, ) self._error = exc self._ready.set() return + if failure_class == "permanent": + # Deterministic failure (bad command, non-MCP URL, + # 401/403): every retry hits the same wall. Park + # immediately instead of burning the retry ladder + # and spamming N identical warnings (#65673). + logger.warning( + "MCP server '%s' failed initial connection with a " + "permanent error, parking without retries " + "(state: connecting → parked): %s: %s", + self.name, type(root).__name__, root, + ) + self._error = exc + self._ready.set() + self._was_parked = True + self._deregister_tools() + self._reconnect_event.clear() + parked = await self._wait_for_reconnect_or_shutdown( + timeout=_PARKED_RETRY_INTERVAL + ) + if parked == "shutdown": + return + logger.info( + "MCP server '%s': attempting revival after " + "permanent initial failure (self-probe or explicit " + "reconnect request); rebuilding transport.", + self.name, + ) + initial_retries = 0 + self._reconnect_retries = 0 + backoff = 1.0 + self._error = None + self._ready.clear() + continue + initial_retries += 1 if initial_retries > _MAX_INITIAL_CONNECT_RETRIES: logger.warning( "MCP server '%s' failed initial connection after " - "%d attempts, parking until a reconnect is requested: %s", - self.name, _MAX_INITIAL_CONNECT_RETRIES, exc, + "%d attempts, parking until a reconnect is " + "requested (state: connecting → parked): %s: %s", + self.name, _MAX_INITIAL_CONNECT_RETRIES, + type(root).__name__, root, ) self._error = exc self._ready.set() + self._was_parked = True self._deregister_tools() self._reconnect_event.clear() parked = await self._wait_for_reconnect_or_shutdown( @@ -3137,9 +3263,10 @@ class MCPServerTask: logger.warning( "MCP server '%s' initial connection failed " - "(attempt %d/%d), retrying in %.0fs: %s", + "(attempt %d/%d), retrying in %.0fs: %s: %s", self.name, initial_retries, - _MAX_INITIAL_CONNECT_RETRIES, backoff, exc, + _MAX_INITIAL_CONNECT_RETRIES, backoff, + type(root).__name__, root, ) await asyncio.sleep(backoff) backoff = min(backoff * 2, _MAX_BACKOFF_SECONDS) @@ -3154,18 +3281,50 @@ class MCPServerTask: # If shutdown was requested, don't reconnect if self._shutdown_event.is_set(): logger.debug( - "MCP server '%s' disconnected during shutdown: %s", - self.name, exc, + "MCP server '%s' disconnected during shutdown: %s: %s", + self.name, type(root).__name__, root, ) return + if failure_class == "permanent": + # A previously-working server now fails deterministically + # (revoked credentials, URL now serving a web page, stdio + # binary uninstalled). Retrying can't help — park + # immediately without burning the retry ladder. + logger.warning( + "MCP server '%s' hit a permanent error, parking " + "without retries; will self-probe every %ds " + "(state: connected → parked): %s: %s", + self.name, _PARKED_RETRY_INTERVAL, + type(root).__name__, root, + ) + self._was_parked = True + self._deregister_tools() + self._reconnect_event.clear() + parked = await self._wait_for_reconnect_or_shutdown( + timeout=_PARKED_RETRY_INTERVAL + ) + if parked == "shutdown": + return + logger.info( + "MCP server '%s': attempting revival from parked state " + "(permanent error; self-probe or explicit reconnect " + "request); rebuilding transport.", + self.name, + ) + self._reconnect_retries = _MAX_RECONNECT_RETRIES + backoff = 1.0 + continue + self._reconnect_retries += 1 if self._reconnect_retries > _MAX_RECONNECT_RETRIES: logger.warning( "MCP server '%s' failed after %d reconnection attempts, " - "parking; will self-probe every %ds until it recovers: %s", + "parking; will self-probe every %ds until it recovers " + "(state: degraded → parked): %s: %s", self.name, _MAX_RECONNECT_RETRIES, - _PARKED_RETRY_INTERVAL, exc, + _PARKED_RETRY_INTERVAL, + type(root).__name__, root, ) # Do NOT return — exiting the task orphans the server: # nothing would ever listen for _reconnect_event again @@ -3178,6 +3337,7 @@ class MCPServerTask: # we wake and attempt one reconnect ourselves (#57129). # An explicit _reconnect_event.set() (OAuth recovery, # manual /mcp refresh) still wakes us immediately. + self._was_parked = True self._deregister_tools() self._reconnect_event.clear() parked = await self._wait_for_reconnect_or_shutdown( @@ -3200,9 +3360,9 @@ class MCPServerTask: logger.warning( "MCP server '%s' connection lost (attempt %d/%d), " - "reconnecting in %.0fs: %s", + "reconnecting in %.0fs: %s: %s", self.name, self._reconnect_retries, _MAX_RECONNECT_RETRIES, - backoff, exc, + backoff, type(root).__name__, root, ) await asyncio.sleep(backoff) backoff = min(backoff * 2, _MAX_BACKOFF_SECONDS)