fix(mcp): unwrap exception groups and park permanent failures immediately (#65673)

- Add module-level _unwrap_exception_group (ported from
  hermes_cli/mcp_config.py, adapted): handles nested groups, prefers
  non-cancellation leaves over the CancelledErrors anyio sprays across
  sibling tasks, and re-raises KeyboardInterrupt/SystemExit leaves.

- Add _classify_mcp_failure(exc) -> 'permanent'|'transient'.
  Permanent: auth 401/403, NonMcpEndpointError, InvalidMcpUrlError,
  FileNotFoundError/ENOENT on the stdio command. Transient: everything
  else (network/EOF/ClosedResource/TaskGroup drops). Permanent failures
  park immediately without burning the retry ladder — every retry
  against a missing binary or a revoked credential hits the same wall.

- Every log site in run() and the keepalive failure log now logs the
  unwrapped root cause as 'TypeName: message', so a dead stdio pipe
  says 'BrokenPipeError' instead of the opaque
  'unhandled errors in a TaskGroup (1 sub-exception)' (and empty
  str(exc) dead-pipe errors are no longer blank).

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()

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)