MCP client reconnect retry limits and resilience

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #53599 fix(mcp): never permanently wedge the circuit breaker on a dead transport

  merged to main

  files: tests/tools/test_mcp_circuit_breaker.py, tools/mcp_tool.py

  Show the diff

  • tests/tools/test_mcp_circuit_breaker.py

    @@ -179,6 +179,107 @@ def _fake_monotonic():
             _cleanup(mcp_tool, "srv")
     
     
    +def test_half_open_probe_on_dead_session_requests_reconnect(monkeypatch, tmp_path):
    +    """A half-open probe against a server with no live session must request
    +    a transport reconnect and return a clean error — NOT write into a dead
    +    pipe or permanently re-arm the breaker.
    +
    +    This is the #16788 wedge: a dead stdio subprocess leaves ``session=None``
    +    (the run loop parked after exhausting retries). The old handler bumped
    +    the breaker every cooldown forever; the fix signals ``_reconnect_event``
    +    so the parked task revives and rebuilds the transport.
    +    """
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import _make_tool_handler
    +
    +    server = _install_stub_server(mcp_tool, "srv", None)
    +    # Simulate a dead/parked transport: no live session.
    +    server.session = None
    +    # Drive _signal_reconnect down its direct .set() path (no live loop).
    +    monkeypatch.setattr(mcp_tool, "_mcp_loop", None)
    +
    +    try:
    +        mcp_tool._server_error_counts["srv"] = mcp_tool._CIRCUIT_BREAKER_THRESHOLD
    +        fake_now = [1000.0]
    +
    +        def _fake_monotonic():
    +            return fake_now[0]
    +
    +        monkeypatch.setattr(mcp_tool.time, "monotonic", _fake_monotonic)
    +        mcp_tool._server_breaker_opened_at["srv"] = fake_now[0]
    +        cooldown = getattr(mcp_tool, "_CIRCUIT_BREAKER_COOLDOWN_SEC", 60.0)
    +
    +        # Advance past cooldown → next call is a half-open probe.
    +        fake_now[0] += cooldown + 1.0
    +
    +        handler = _make_tool_handler("srv", "tool1", 10.0)
    +        result = handler({})
    +        parsed = json.loads(result)
    +
    +        # Clean "reconnecting" error, and a reconnect was actually signalled.
    +        assert "reconnect" in parsed.get("error", "").lower(), parsed
    +        server._reconnect_event.set.assert_called_once()
    +    finally:
    +        _cleanup(mcp_tool, "srv")
    +
    +
    +def test_half_open_dead_session_recovers_after_reconnect(monkeypatch, tmp_path):
    +    """Once the transport comes back (session repopulated + breaker reset by
    +    the run loop), the next call must go straight through — proving the wedge
    +    is escapable, not just deferred.
    +    """
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import _make_tool_handler
    +
    +    async def _call_tool_success(*a, **kw):
    +        result = MagicMock()
    +        result.isError = False
    +        block = MagicMock()
    +        block.text = "ok"
    +        result.content = [block]
    +        result.structuredContent = None
    +        return result
    +
    +    server = _install_stub_server(mcp_tool, "srv", _call_tool_success)
    +    server.session = None  # transport down at first
    +    monkeypatch.setattr(mcp_tool, "_mcp_loop", None)
    +    mcp_tool._ensure_mcp_loop()
    +
    +    try:
    +        mcp_tool._server_error_counts["srv"] = mcp_tool._CIRCUIT_BREAKER_THRESHOLD
    +        fake_now = [1000.0]
    +        monkeypatch.setattr(mcp_tool.time, "monotonic", lambda: fake_now[0])
    +        mcp_tool._server_breaker_opened_at["srv"] = fake_now[0]
    +        cooldown = getattr(mcp_tool, "_CIRCUIT_BREAKER_COOLDOWN_SEC", 60.0)
    +        fake_now[0] += cooldown + 1.0
    +
    +        handler = _make_tool_handler("srv", "tool1", 10.0)
    +
    +        # Probe 1: transport down → reconnect requested, clean error.
    +        parsed = json.loads(handler({}))
    +        assert "reconnect" in parsed.get("error", "").lower(), parsed
    +
    +        # Simulate the run loop rebuilding the session + resetting the breaker
    +        # (what _run_stdio does on successful re-init).
    +        live = MagicMock()
    +        live.call_tool = _call_tool_success
    +        server.session = live
    +        mcp_tool._reset_server_error("srv")
    +
    +        # Advance past the re-armed cooldown so the next call is a fresh probe.
    +        fake_now[0] += cooldown + 1.0
    +
    +        # Next call goes straight through.
    +        parsed = json.loads(handler({}))
    +        assert parsed.get("result") == "ok", parsed
    +    finally:
    +        _cleanup(mcp_tool, "srv")
    +
    +
     def test_circuit_breaker_cleared_on_reconnect(monkeypatch, tmp_path):
         """When the auth-recovery path successfully reconnects the server,
         the breaker should be cleared so subsequent calls aren't gated on a
    @@ -250,3 +351,98 @@ def _retry_call():
             )
         finally:
             _cleanup(mcp_tool, "srv")
    +
    +
    +def test_run_loop_parks_instead_of_exiting_then_revives(monkeypatch, tmp_path):
    +    """The run loop must NOT exit when the reconnect budget is exhausted.
    +
    +    It deregisters tools and parks as a dormant listener; a later
    +    ``_reconnect_event`` revives it and re-enters the transport. This is the
    +    structural fix for #16788 — without a live task, no half-open probe could
    +    ever bring a dead stdio server back.
    +    """
    +    import asyncio
    +
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import MCPServerTask
    +
    +    # Shrink the budget and collapse backoff sleeps (but still yield control
    +    # to the loop) so the test runs fast without starving the scheduler.
    +    monkeypatch.setattr(mcp_tool, "_MAX_RECONNECT_RETRIES", 2)
    +
    +    _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, "deregistered": 0, "revived": False}
    +
    +    async def _scenario():
    +        class _Task(MCPServerTask):
    +            def _is_http(self):
    +                return False
    +
    +            def _deregister_tools(self):
    +                state["deregistered"] += 1
    +                self._registered_tool_names = []
    +
    +            async def _run_stdio(self, config):
    +                state["transport_calls"] += 1
    +                # First connect succeeds (sets _ready) then immediately
    +                # fails, as if the subprocess died — the post-ready failure
    +                # path that counts toward the reconnect budget.
    +                if state["transport_calls"] == 1:
    +                    self.session = object()
    +                    self._ready.set()
    +                    self.session = None
    +                    raise RuntimeError("subprocess died")
    +                # Keep failing until the budget is exhausted and the loop
    +                # parks, UNLESS we've been revived after parking.
    +                if state["revived"]:
    +                    self.session = object()
    +                    self._ready.set()
    +                    await self._wait_for_lifecycle_event()
    +                    return
    +                raise RuntimeError("still down")
    +
    +        task = _Task("srv")
    +        task._registered_tool_names = ["srv__tool"]
    +
    +        run_task = asyncio.ensure_future(task.run({"command": "x"}))
    +
    +        # Wait until the loop has parked (it deregisters tools right before
    +        # blocking on _wait_for_reconnect_or_shutdown).
    +        for _ in range(500):
    +            await _real_sleep(0)
    +            if state["deregistered"] >= 1:
    +                break
    +        # Give the loop one more tick to settle into the park wait.
    +        await _real_sleep(0)
    +        assert not run_task.done(), "run loop exited instead of parking"
    +        assert state["deregistered"] >= 1, "tools not deregistered on park"
    +
    +        # Revive it: a reconnect signal must wake the parked task.
    +        state["revived"] = True
    +        before = state["transport_calls"]
    +        task._reconnect_event.set()
    +        for _ in range(500):
    +            await _real_sleep(0)
    +            if state["transport_calls"] > before:
    +                break
    +        assert state["transport_calls"] > before, (
    +            "parked task did not re-enter transport on reconnect signal"
    +        )
    +
    +        # Clean shutdown.
    +        task._shutdown_event.set()
    +        task._reconnect_event.set()
    +        try:
    +            await asyncio.wait_for(run_task, timeout=2)
    +        except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
    +            run_task.cancel()
    +
    +    asyncio.run(_scenario())

  • tools/mcp_tool.py

    @@ -1759,6 +1759,41 @@ async def _wait_for_lifecycle_event(self) -> str:
             self._reconnect_event.clear()
             return "reconnect"
     
    +    async def _wait_for_reconnect_or_shutdown(self) -> str:
    +        """Block until a reconnect or shutdown is requested while parked.
    +
    +        Used by :meth:`run` after the reconnect budget is exhausted. The
    +        task stays alive (so ``_reconnect_event`` always has a listener) but
    +        does no work until something explicitly asks it to come back —
    +        the circuit-breaker half-open probe, OAuth recovery, or a manual
    +        ``/mcp`` refresh.
    +
    +        Returns:
    +            ``"shutdown"`` if the server should exit the run loop entirely,
    +            ``"reconnect"`` if it should rebuild the transport. The reconnect
    +            event is cleared before returning so the next park cycle starts
    +            from a fresh signal. Shutdown takes precedence.
    +        """
    +        shutdown_task = asyncio.ensure_future(self._shutdown_event.wait())
    +        reconnect_task = asyncio.ensure_future(self._reconnect_event.wait())
    +        try:
    +            await asyncio.wait(
    +                {shutdown_task, reconnect_task},
    +                return_when=asyncio.FIRST_COMPLETED,
    +            )
    +        finally:
    +            for t in (shutdown_task, reconnect_task):
    +                if not t.done():
    +                    t.cancel()
    +                    try:
    +                        await t
    +                    except (asyncio.CancelledError, Exception):
    +                        pass
    +        if self._shutdown_event.is_set():
    +            return "shutdown"
    +        self._reconnect_event.clear()
    +        return "reconnect"
    +
         async def _run_stdio(self, config: dict):
             """Run the server using stdio transport."""
             if not _MCP_AVAILABLE:
    @@ -1857,6 +1892,10 @@ async def _run_stdio(self, config: dict):
                         self.session = session
                         await self._discover_tools()
                         self._ready.set()
    +                    # Session is live again: clear any breaker state from a
    +                    # prior outage so the first call after recovery isn't
    +                    # gated on a stale consecutive-failure count (#16788).
    +                    _reset_server_error(self.name)
                         # stdio transport does not use OAuth, but we still honor
                         # _reconnect_event (e.g. future manual /mcp refresh) for
                         # consistency with _run_http.
    @@ -2088,6 +2127,10 @@ def _mcp_http_client_factory(
                         self.session = session
                         await self._discover_tools()
                         self._ready.set()
    +                    # Session is live again: clear any breaker state from a
    +                    # prior outage so the first call after recovery isn't
    +                    # gated on a stale consecutive-failure count (#16788).
    +                    _reset_server_error(self.name)
                         reason = await self._wait_for_lifecycle_event()
                         if reason == "reconnect":
                             logger.info(
    @@ -2137,6 +2180,10 @@ async def _strip_auth_on_cross_origin_redirect(response):
                             self.session = session
                             await self._discover_tools()
                             self._ready.set()
    +                        # Session is live again: clear any breaker state from
    +                        # a prior outage so the first call after recovery
    +                        # isn't gated on a stale failure count (#16788).
    +                        _reset_server_error(self.name)
                             reason = await self._wait_for_lifecycle_event()
                             if reason == "reconnect":
                                 logger.info(
    @@ -2160,6 +2207,10 @@ async def _strip_auth_on_cross_origin_redirect(response):
                         self.session = session
                         await self._discover_tools()
                         self._ready.set()
    +                    # Session is live again: clear any breaker state from a
    +                    # prior outage so the first call after recovery isn't
    +                    # gated on a stale consecutive-failure count (#16788).
    +                    _reset_server_error(self.name)
                         reason = await self._wait_for_lifecycle_event()
                         if reason == "reconnect":
                             logger.info(
    @@ -2298,6 +2349,15 @@ async def run(self, config: dict):
                         "manual refresh)",
                         self.name,
                     )
    +                # A clean transport return only happens after a session was
    +                # successfully established and then asked to rebuild (auth
    +                # recovery / manual refresh / breaker-driven reconnect). That
    +                # is proof the server is reachable, so clear the consecutive-
    +                # failure budget — otherwise transient drops accumulated over
    +                # a long-lived session would eventually exhaust it and
    +                # permanently kill an otherwise-healthy server.
    +                retries = 0
    +                backoff = 1.0
                     # Reset the session reference; _run_http/_run_stdio will
                     # repopulate it on successful re-entry.
                     self.session = None
    @@ -2374,10 +2434,32 @@ async def run(self, config: dict):
                     if retries > _MAX_RECONNECT_RETRIES:
                         logger.warning(
                             "MCP server '%s' failed after %d reconnection attempts, "
    -                        "giving up: %s",
    +                        "parking until a reconnect is requested: %s",
                             self.name, _MAX_RECONNECT_RETRIES, exc,
                         )
    -                    return
    +                    # Do NOT return — exiting the task orphans the server:
    +                    # nothing would ever listen for _reconnect_event again,
    +                    # so a half-open circuit-breaker probe could never revive
    +                    # it and the server would be permanently wedged for the
    +                    # life of the process (#16788). Instead, drop the phantom
    +                    # tools from the registry and park as a dormant listener.
    +                    # A future _reconnect_event.set() — from the breaker's
    +                    # half-open probe, OAuth recovery, or a manual /mcp
    +                    # refresh — wakes us to rebuild the transport (respawning
    +                    # a dead stdio subprocess in the process).
    +                    self._deregister_tools()
    +                    self._reconnect_event.clear()
    +                    parked = await self._wait_for_reconnect_or_shutdown()
    +                    if parked == "shutdown":
    +                        return
    +                    logger.info(
    +                        "MCP server '%s': reconnect requested while parked; "
    +                        "rebuilding transport.",
    +                        self.name,
    +                    )
    +                    retries = 0
    +                    backoff = 1.0
    +                    continue
     
                     logger.warning(
                         "MCP server '%s' connection lost (attempt %d/%d), "
    @@ -2403,8 +2485,6 @@ async def start(self, config: dict):
     
         async def shutdown(self):
             """Signal the Task to exit and wait for clean resource teardown."""
    -        from tools.registry import registry
    -
             self._shutdown_event.set()
             # Defensive: if _wait_for_lifecycle_event is blocking, we need ANY
             # event to unblock it. _shutdown_event alone is sufficient (the
    @@ -2430,11 +2510,24 @@ async def shutdown(self):
                     task.cancel()
                 await asyncio.gather(*self._pending_refresh_tasks, return_exceptions=True)
                 self._pending_refresh_tasks.clear()
    +        self._deregister_tools()
    +        self.session = None
    +
    +    def _deregister_tools(self) -> None:
    +        """Drop this server's tools from the global registry (idempotent).
    +
    +        Pulls the server's tool schemas out of the registry so the agent
    +        stops advertising them to the model. Called on shutdown AND when the
    +        reconnect budget is exhausted, so a dead server never leaves phantom
    +        tool definitions bloating the prompt cache and producing "not
    +        connected" errors on every turn.
    +        """
    +        from tools.registry import registry
    +
             for tool_name in list(getattr(self, "_registered_tool_names", [])):
                 registry.deregister(tool_name)
                 _forget_mcp_tool_server(tool_name)
             self._registered_tool_names = []
    -        self.session = None
     
     
     # ---------------------------------------------------------------------------
    @@ -2491,6 +2584,31 @@ def _reset_server_error(server_name: str) -> None:
         _server_error_counts[server_name] = 0
         _server_breaker_opened_at.pop(server_name, None)
     
    +
    +def _signal_reconnect(server: Any) -> bool:
    +    """Ask a server task to rebuild its transport, thread-safely.
    +
    +    The tool handlers run on caller threads, while the server task and its
    +    ``_reconnect_event`` live on the background MCP loop. Setting an
    +    asyncio.Event from another thread must go through
    +    ``loop.call_soon_threadsafe``; only fall back to a direct ``.set()``
    +    when the loop isn't running (e.g. unit tests that drive the handler
    +    synchronously).
    +
    +    Returns True if a reconnect signal was delivered, False if the server
    +    has no reconnect machinery (nothing to revive).
    +    """
    +    event = getattr(server, "_reconnect_event", None)
    +    if event is None:
    +        return False
    +    loop = _mcp_loop
    +    if loop is not None and loop.is_running():
    +        loop.call_soon_threadsafe(event.set)
    +    else:
    +        event.set()
    +    return True
    +
    +
     # ---------------------------------------------------------------------------
     # Auth-failure detection helpers (Task 6 of MCP OAuth consolidation)
     # ---------------------------------------------------------------------------
    @@ -3176,12 +3294,35 @@ def _handler(args: dict, **kwargs) -> str:
     
             with _lock:
                 server = _servers.get(server_name)
    -        if not server or not server.session:
    +        if not server:
                 _bump_server_error(server_name)
                 return json.dumps({
                     "error": f"MCP server '{server_name}' is not connected"
                 }, ensure_ascii=False)
     
    +        if not server.session:
    +            # No live session — the server task is reconnecting, or it has
    +            # exhausted its retry budget and parked (e.g. a dead stdio
    +            # subprocess). Probing here would write into a dead/absent
    +            # transport and re-arm the breaker forever (#16788). Instead,
    +            # ask the (always-present) server task to rebuild the transport
    +            # — which respawns a dead stdio subprocess — and return a clean
    +            # "reconnecting" error so the model backs off without burning
    +            # iterations. The breaker resets once the fresh session
    +            # initializes (_run_stdio/_run_http call _reset_server_error).
    +            _bump_server_error(server_name)
    +            if _signal_reconnect(server):
    +                return json.dumps({
    +                    "error": (
    +                        f"MCP server '{server_name}' transport is down; "
    +                        f"reconnect requested. Do NOT retry this tool "
    +                        f"immediately — give it a few seconds to come back."
    +                    )
    +                }, ensure_ascii=False)
    +            return json.dumps({
    +                "error": f"MCP server '{server_name}' is not connected"
    +            }, ensure_ascii=False)
    +
             async def _call():
                 async with server._rpc_lock:
                     # Snapshot the agent's context so an elicitation callback

• #59222 fix(mcp): reconnect resilience — reset budget per healthy session, self-probe parked servers, re-register tools on revival (salvage #57615, #54139, #57477)

  merged to main

  files: scripts/release.py, tests/tools/test_mcp_circuit_breaker.py, tests/tools/test_mcp_parked_self_probe.py, tests/tools/test_mcp_reconnect_retry_reset.py, tests/tools/test_mcp_stability.py, tests/tools/test_mcp_tool.py, tools/mcp_tool.py

  Show the diff

  • scripts/release.py

    @@ -1198,6 +1198,7 @@
         "[redacted:email]": "LeonSGP43",
         "[redacted:email]": "Lubrsy706",
         "[redacted:email]": "spniyant",
    +    "[redacted:email]": "nicha16",  # PR #54139 partial salvage (re-register tools after park revival)
         "[redacted:email]": "olafthiele",
         "[redacted:email]": "sprmn24",
         "[redacted:email]": "olafthiele",

  • tests/tools/test_mcp_circuit_breaker.py

    @@ -446,3 +446,84 @@ async def _run_stdio(self, config):
                 run_task.cancel()
     
         asyncio.run(_scenario())
    +
    +
    +def test_initial_connect_budget_parks_instead_of_exiting_then_revives(monkeypatch, tmp_path):
    +    """Initial connection failures must park, not permanently exit the task.
    +
    +    Regression for #57129's remaining live case: a slow HTTP/SSE server or
    +    late-starting stdio server could exhaust the initial-connect budget before
    +    it ever registered tools. The run loop returned, leaving no task alive to
    +    hear a later manual /mcp refresh.
    +    """
    +    import asyncio
    +
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import MCPServerTask
    +
    +    monkeypatch.setattr(mcp_tool, "_MAX_INITIAL_CONNECT_RETRIES", 2)
    +
    +    _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, "deregistered": 0, "revived": False}
    +
    +    async def _scenario():
    +        class _Task(MCPServerTask):
    +            def _is_http(self):
    +                return False
    +
    +            def _deregister_tools(self):
    +                state["deregistered"] += 1
    +                self._registered_tool_names = []
    +
    +            async def _run_stdio(self, config):
    +                state["transport_calls"] += 1
    +                if not state["revived"]:
    +                    raise RuntimeError("server still booting")
    +                self.session = object()
    +                self._ready.set()
    +                await self._wait_for_lifecycle_event()
    +                return
    +
    +        task = _Task("srv")
    +        run_task = asyncio.ensure_future(task.run({"command": "x"}))
    +
    +        for _ in range(500):
    +            await _real_sleep(0)
    +            if state["deregistered"] >= 1:
    +                break
    +
    +        await _real_sleep(0)
    +        assert state["transport_calls"] == 3
    +        assert state["deregistered"] >= 1
    +        assert task._ready.is_set()
    +        assert task._error is not None
    +        assert not run_task.done(), "initial failure exited instead of parking"
    +
    +        state["revived"] = True
    +        before = state["transport_calls"]
    +        task._reconnect_event.set()
    +        for _ in range(500):
    +            await _real_sleep(0)
    +            if state["transport_calls"] > before and task.session is not None:
    +                break
    +
    +        assert state["transport_calls"] > before
    +        assert task.session is not None
    +        assert task._error is None
    +
    +        task._shutdown_event.set()
    +        task._reconnect_event.set()
    +        try:
    +            await asyncio.wait_for(run_task, timeout=2)
    +        except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
    +            run_task.cancel()
    +
    +    asyncio.run(_scenario())

  • tests/tools/test_mcp_parked_self_probe.py

    @@ -0,0 +1,108 @@
    +"""Tests for the parked-server self-probe revival path (#57129).
    +
    +Parking deregisters a server's tools, so no tool call can reach the
    +circuit-breaker half-open probe or ``_signal_reconnect`` — the only
    +things that set ``_reconnect_event``. The parked wait must therefore be
    +timed: the run task wakes on ``_PARKED_RETRY_INTERVAL`` and attempts one
    +revival probe on its own.
    +"""
    +
    +import asyncio
    +
    +import pytest
    +
    +
    [redacted:email]_isolate
    +def test_parked_server_self_probes_and_revives(monkeypatch, tmp_path):
    +    """A parked server must revive on its own once the backend recovers,
    +    without any explicit _reconnect_event.set()."""
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import MCPServerTask
    +
    +    monkeypatch.setattr(mcp_tool, "_MAX_RECONNECT_RETRIES", 1)
    +    # Keep the self-probe cadence tiny so the test is fast.
    +    monkeypatch.setattr(mcp_tool, "_PARKED_RETRY_INTERVAL", 0.05)
    +
    +    _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,
    +        "deregistered": 0,
    +        "backend_up": False,
    +        "revived_registration": 0,
    +    }
    +
    +    async def _scenario():
    +        class _Task(MCPServerTask):
    +            def _is_http(self):
    +                return False
    +
    +            def _deregister_tools(self):
    +                state["deregistered"] += 1
    +                self._registered_tool_names = []
    +
    +            def _register_discovered_tools_if_needed(self):
    +                if self._ready.is_set() and not self._registered_tool_names:
    +                    state["revived_registration"] += 1
    +                    self._registered_tool_names = ["srv__tool"]
    +
    +            async def _run_stdio(self, config):
    +                state["transport_calls"] += 1
    +                if state["transport_calls"] == 1:
    +                    # First connect succeeds (sets _ready), then dies.
    +                    self.session = object()
    +                    self._ready.set()
    +                    self.session = None
    +                    raise RuntimeError("backend outage begins")
    +                if not state["backend_up"]:
    +                    raise RuntimeError("backend still down")
    +                # Backend recovered: establish a session and park in the
    +                # lifecycle wait like the real transport does.
    +                self.session = object()
    +                self._register_discovered_tools_if_needed()
    +                await self._wait_for_lifecycle_event()
    +
    +        task = _Task("srv")
    +        task._registered_tool_names = ["srv__tool"]
    +
    +        run_task = asyncio.ensure_future(task.run({"command": "x"}))
    +
    +        # Let it exhaust the budget (1 retry) and park.
    +        for _ in range(2000):
    +            await _real_sleep(0)
    +            if state["deregistered"] >= 1:
    +                break
    +        assert state["deregistered"] >= 1, "server never parked"
    +        assert not run_task.done(), "run task exited instead of parking"
    +
    +        # The backend comes back. NOTHING sets _reconnect_event — revival
    +        # must come from the timed self-probe alone.
    +        state["backend_up"] = True
    +        for _ in range(200):
    +            await _real_sleep(0.01)
    +            if task.session is not None:
    +                break
    +
    +        assert task.session is not None, (
    +            "parked server never self-probed back to life "
    +            f"(transport_calls={state['transport_calls']})"
    +        )
    +        assert state["revived_registration"] >= 1, (
    +            "revived server did not re-register its tools"
    +        )
    +
    +        task._shutdown_event.set()
    +        task._reconnect_event.set()
    +        try:
    +            await asyncio.wait_for(run_task, timeout=2)
    +        except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
    +            run_task.cancel()
    +
    +    asyncio.run(_scenario())

  • tests/tools/test_mcp_reconnect_retry_reset.py

    @@ -0,0 +1,196 @@
    +"""Tests for MCP reconnect retry counter reset (#57604).
    +
    +Verifies that the reconnect retry counter resets after each successful
    +reconnection, so transient blips do not accumulate toward permanent parking.
    +"""
    +
    +import asyncio
    +
    +import pytest
    +
    +
    [redacted:email]_isolate
    +def test_reconnect_counter_resets_after_successful_session(monkeypatch, tmp_path):
    +    """Transient disconnections must not accumulate toward permanent parking.
    +
    +    Before the fix, ``retries`` was a local variable in ``run()`` that only
    +    reset on clean transport return (line 2367) or park-wake (line 2468).
    +    Each exception from ``_run_stdio`` incremented it without reset, so 5
    +    transient blips over a long-uptime gateway would permanently park the
    +    server.
    +
    +    After the fix, ``_reconnect_retries`` is an instance variable that resets
    +    to 0 whenever a session is successfully established (``_reset_server_error``
    +    call sites in ``_run_stdio`` / ``_run_http``).
    +    """
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import MCPServerTask
    +
    +    # Shrink budget so the test can exhaust it quickly if the counter
    +    # does NOT reset (the bug scenario).
    +    monkeypatch.setattr(mcp_tool, "_MAX_RECONNECT_RETRIES", 3)
    +
    +    _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,
    +        "max_retries_seen": 0,
    +    }
    +
    +    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
    +                call = state["transport_calls"]
    +
    +                if call == 1:
    +                    # First connect: succeed (sets _ready), then fail.
    +                    self.session = object()
    +                    self._ready.set()
    +                    self.session = None
    +                    raise RuntimeError("blip 1")
    +
    +                # Subsequent calls: succeed (session established, which
    +                # triggers _reset_server_error and should reset retries),
    +                # then immediately fail again — simulating a new transient
    +                # blip.  If retries accumulate, call 4 would exceed the
    +                # budget of 3 and park.  If retries reset correctly,
    +                # this loop can continue indefinitely.
    +                if call <= 8:
    +                    self.session = object()
    +                    # _run_stdio calls _reset_server_error and sets
    +                    # _reconnect_retries = 0 after session establishment.
    +                    # We simulate that by calling the real method.
    +                    mcp_tool._reset_server_error(self.name)
    +                    self._reconnect_retries = 0
    +                    self.session = None
    +                    raise RuntimeError(f"blip {call}")
    +
    +                # If we reach here without parking, the fix works.
    +                self.session = object()
    +                mcp_tool._reset_server_error(self.name)
    +                self._reconnect_retries = 0
    +                await self._wait_for_lifecycle_event()
    +                return
    +
    +        task = _Task("srv")
    +        task._registered_tool_names = ["srv__tool"]
    +
    +        run_task = asyncio.ensure_future(task.run({"command": "x"}))
    +
    +        # Let the scenario run.
    +        for _ in range(2000):
    +            await _real_sleep(0)
    +            if state["transport_calls"] >= 8 or state["parked"]:
    +                break
    +
    +        # The fix: the server should NOT have parked despite 8 transient
    +        # disconnections, because each successful reconnection reset the
    +        # retry counter.
    +        assert not state["parked"], (
    +            f"server parked after {state['transport_calls']} transport calls "
    +            f"— retry counter accumulated instead of resetting"
    +        )
    +        assert state["transport_calls"] >= 8, (
    +            f"only {state['transport_calls']} transport calls reached "
    +            f"(expected >= 8)"
    +        )
    +
    +        # Verify the counter is an instance variable, not a local.
    +        assert hasattr(task, "_reconnect_retries"), (
    +            "_reconnect_retries should be an instance variable"
    +        )
    +
    +        # Clean shutdown.
    +        task._shutdown_event.set()
    +        task._reconnect_event.set()
    +        try:
    +            await asyncio.wait_for(run_task, timeout=2)
    +        except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
    +            run_task.cancel()
    +
    +    asyncio.run(_scenario())
    +
    +
    [redacted:email]_isolate
    +def test_reconnect_counter_still_parks_on_consecutive_failures(monkeypatch, tmp_path):
    +    """The server must still park when failures are genuinely consecutive
    +    (no successful reconnection in between).
    +    """
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +
    +    from tools import mcp_tool
    +    from tools.mcp_tool import MCPServerTask
    +
    +    monkeypatch.setattr(mcp_tool, "_MAX_RECONNECT_RETRIES", 2)
    +
    +    _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
    +                call = state["transport_calls"]
    +
    +                if call == 1:
    +                    self.session = object()
    +                    self._ready.set()
    +                    self.session = None
    +                    raise RuntimeError("first failure")
    +
    +                # All subsequent calls fail WITHOUT establishing a session
    +                # (no _reset_server_error, no retry reset). This simulates
    +                # genuinely consecutive failures.
    +                raise RuntimeError(f"failure {call}")
    +
    +        task = _Task("srv")
    +        task._registered_tool_names = ["srv__tool"]
    +
    +        run_task = asyncio.ensure_future(task.run({"command": "x"}))
    +
    +        for _ in range(500):
    +            await _real_sleep(0)
    +            if state["parked"] or run_task.done():
    +                break
    +
    +        assert state["parked"], (
    +            "server should park on consecutive failures without successful reconnect"
    +        )
    +
    +        task._shutdown_event.set()
    +        task._reconnect_event.set()
    +        try:
    +            await asyncio.wait_for(run_task, timeout=2)
    +        except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
    +            run_task.cancel()
    +
    +    asyncio.run(_scenario())

  • tests/tools/test_mcp_stability.py

    @@ -560,7 +560,7 @@ async def fake_run_stdio(self_inner, config):
             asyncio.get_event_loop().run_until_complete(_run())
     
         def test_initial_connect_gives_up_after_max_retries(self):
    -        """Server gives up after _MAX_INITIAL_CONNECT_RETRIES failures."""
    +        """Server parks (does not exit) after _MAX_INITIAL_CONNECT_RETRIES failures."""
             from tools.mcp_tool import MCPServerTask, _MAX_INITIAL_CONNECT_RETRIES
     
             call_count = 0
    @@ -583,8 +583,13 @@ async def fake_run_stdio(self_inner, config):
                     assert "DNS resolution failed" in str(server._error)
                     # 1 initial + N retries = _MAX_INITIAL_CONNECT_RETRIES + 1 total attempts
                     assert call_count == _MAX_INITIAL_CONNECT_RETRIES + 1
    +                # The task parks for later revival instead of exiting.
    +                await asyncio.sleep(0)
    +                assert not task.done(), "run task should park, not exit"
     
    -                await task
    +                server._shutdown_event.set()
    +                server._reconnect_event.set()
    +                await asyncio.wait_for(task, timeout=5)
     
             asyncio.get_event_loop().run_until_complete(_run())
     

  • tests/tools/test_mcp_tool.py

    @@ -1876,12 +1876,19 @@ async def _test():
     
                 with patch.object(MCPServerTask, "_run_stdio", patched_run_stdio), \
                      patch("asyncio.sleep", new_callable=AsyncMock):
    -                await server.run({"command": "test"})
    +                task = asyncio.ensure_future(server.run({"command": "test"}))
    +                await server._ready.wait()
     
    -            # Now retries up to _MAX_INITIAL_CONNECT_RETRIES before giving up
    -            assert run_count == _MAX_INITIAL_CONNECT_RETRIES + 1
    -            assert server._error is not None
    -            assert "cannot connect" in str(server._error)
    +                # Now retries up to _MAX_INITIAL_CONNECT_RETRIES, then PARKS
    +                # (keeps the task alive for later revival) instead of exiting.
    +                assert run_count == _MAX_INITIAL_CONNECT_RETRIES + 1
    +                assert server._error is not None
    +                assert "cannot connect" in str(server._error)
    +                assert not task.done(), "run task should park, not exit"
    +
    +                server._shutdown_event.set()
    +                server._reconnect_event.set()
    +                await asyncio.wait_for(task, timeout=5)
     
             asyncio.run(_test())
     

  • tools/mcp_tool.py

    @@ -321,6 +321,11 @@ def _check_logging_callback_support() -> bool:
     _MAX_RECONNECT_RETRIES = 5
     _MAX_INITIAL_CONNECT_RETRIES = 3 # retries for the very first connection attempt
     _MAX_BACKOFF_SECONDS = 60
    +# While parked (reconnect budget exhausted, tools deregistered) the run task
    +# wakes on this cadence and attempts one revival probe. Without it a parked
    +# server is unrevivable: its tools are out of the registry, so no tool call
    +# can ever reach the circuit-breaker half-open probe or _signal_reconnect.
    +_PARKED_RETRY_INTERVAL = 300     # seconds between parked self-probes
     
     # Keepalive cadence for HTTP/SSE sessions. The MCP spec lets a server expire
     # idle sessions on any TTL it chooses (Streamable HTTP "Session Management"),
    @@ -1475,6 +1480,7 @@ class MCPServerTask:
             "_rpc_lock", "_pending_refresh_tasks",
             "_pending_call_context",
             "initialize_result", "_ping_unsupported",
    +        "_reconnect_retries",
         )
     
         def __init__(self, name: str):
    @@ -1496,6 +1502,7 @@ def __init__(self, name: str):
             self._sampling: Optional[SamplingHandler] = None
             self._elicitation: Optional[ElicitationHandler] = None
             self._registered_tool_names: list[str] = []
    +        self._reconnect_retries: int = 0
             self._auth_type: str = ""
             self._refresh_lock = asyncio.Lock()
             # MCP stdio sessions are a single JSON-RPC stream. Some servers emit
    @@ -1837,20 +1844,27 @@ async def _wait_for_lifecycle_event(self) -> str:
             self._reconnect_event.clear()
             return "reconnect"
     
    -    async def _wait_for_reconnect_or_shutdown(self) -> str:
    +    async def _wait_for_reconnect_or_shutdown(
    +        self, timeout: Optional[float] = None
    +    ) -> str:
             """Block until a reconnect or shutdown is requested while parked.
     
             Used by :meth:`run` after the reconnect budget is exhausted. The
             task stays alive (so ``_reconnect_event`` always has a listener) but
             does no work until something explicitly asks it to come back —
    -        the circuit-breaker half-open probe, OAuth recovery, or a manual
    -        ``/mcp`` refresh.
    +        OAuth recovery, a manual ``/mcp`` refresh — or, when ``timeout`` is
    +        given, until the timeout elapses (a periodic self-probe). The timed
    +        wake matters because parking deregisters this server's tools, so
    +        no tool call can ever reach the circuit-breaker's half-open probe
    +        or ``_signal_reconnect`` — without a self-probe a parked server
    +        would be unrevivable short of a full reload.
     
             Returns:
                 ``"shutdown"`` if the server should exit the run loop entirely,
    -            ``"reconnect"`` if it should rebuild the transport. The reconnect
    -            event is cleared before returning so the next park cycle starts
    -            from a fresh signal. Shutdown takes precedence.
    +            ``"reconnect"`` if it should rebuild the transport (explicit
    +            request or self-probe timeout). The reconnect event is cleared
    +            before returning so the next park cycle starts from a fresh
    +            signal. Shutdown takes precedence.
             """
             shutdown_task = asyncio.ensure_future(self._shutdown_event.wait())
             reconnect_task = asyncio.ensure_future(self._reconnect_event.wait())
    @@ -1858,6 +1872,7 @@ async def _wait_for_reconnect_or_shutdown(self) -> str:
                 await asyncio.wait(
                     {shutdown_task, reconnect_task},
                     return_when=asyncio.FIRST_COMPLETED,
    +                timeout=timeout,
                 )
             finally:
                 for t in (shutdown_task, reconnect_task):
    @@ -1984,6 +1999,10 @@ async def _run_stdio(self, config: dict):
                         # prior outage so the first call after recovery isn't
                         # gated on a stale consecutive-failure count (#16788).
                         _reset_server_error(self.name)
    +                    # This session is live: reset the reconnect retry counter
    +                    # so transient prior failures do not accumulate toward
    +                    # permanent parking (#57604).
    +                    self._reconnect_retries = 0
                         # stdio transport does not use OAuth, but we still honor
                         # _reconnect_event (e.g. future manual /mcp refresh) for
                         # consistency with _run_http.
    @@ -2221,6 +2240,7 @@ def _mcp_http_client_factory(
                         # prior outage so the first call after recovery isn't
                         # gated on a stale consecutive-failure count (#16788).
                         _reset_server_error(self.name)
    +                    self._reconnect_retries = 0
                         reason = await self._wait_for_lifecycle_event()
                         if reason == "reconnect":
                             logger.info(
    @@ -2274,6 +2294,7 @@ async def _strip_auth_on_cross_origin_redirect(response):
                             # a prior outage so the first call after recovery
                             # isn't gated on a stale failure count (#16788).
                             _reset_server_error(self.name)
    +                        self._reconnect_retries = 0
                             reason = await self._wait_for_lifecycle_event()
                             if reason == "reconnect":
                                 logger.info(
    @@ -2301,6 +2322,7 @@ async def _strip_auth_on_cross_origin_redirect(response):
                         # prior outage so the first call after recovery isn't
                         # gated on a stale consecutive-failure count (#16788).
                         _reset_server_error(self.name)
    +                    self._reconnect_retries = 0
                         reason = await self._wait_for_lifecycle_event()
                         if reason == "reconnect":
                             logger.info(
    @@ -2331,6 +2353,7 @@ async def _discover_tools(self):
                     self.name,
                 )
                 self._tools = []
    +            self._register_discovered_tools_if_needed()
                 return
             async with self._rpc_lock:
                 tools_result = await self.session.list_tools()
    @@ -2339,6 +2362,23 @@ async def _discover_tools(self):
                 if hasattr(tools_result, "tools")
                 else []
             )
    +        self._register_discovered_tools_if_needed()
    +
    +    def _register_discovered_tools_if_needed(self) -> None:
    +        """Re-register tools after a post-ready reconnect if needed.
    +
    +        Initial registration is performed by ``_discover_and_register_server``
    +        after ``start()`` completes. During a later reconnect, however,
    +        ``_ready`` remains set; if outage handling previously deregistered
    +        stale tools (parking calls ``_deregister_tools``), a successful
    +        revival must publish the freshly discovered tools again — otherwise
    +        the transport comes back alive with zero registered tools.
    +        """
    +        if not self._ready.is_set() or self._registered_tool_names:
    +            return
    +        self._registered_tool_names = _register_server_tools(
    +            self.name, self, self._config
    +        )
     
         async def run(self, config: dict):
             """Long-lived coroutine: connect, discover tools, wait, disconnect.
    @@ -2417,7 +2457,7 @@ async def run(self, config: dict):
                         self._ready.set()
                         return
     
    -        retries = 0
    +        self._reconnect_retries = 0
             initial_retries = 0
             backoff = 1.0
     
    @@ -2446,7 +2486,7 @@ async def run(self, config: dict):
                     # failure budget — otherwise transient drops accumulated over
                     # a long-lived session would eventually exhaust it and
                     # permanently kill an otherwise-healthy server.
    -                retries = 0
    +                self._reconnect_retries = 0
                     backoff = 1.0
                     # Reset the session reference; _run_http/_run_stdio will
                     # repopulate it on successful re-entry.
    @@ -2489,12 +2529,30 @@ async def run(self, config: dict):
                         if initial_retries > _MAX_INITIAL_CONNECT_RETRIES:
                             logger.warning(
                                 "MCP server '%s' failed initial connection after "
    -                            "%d attempts, giving up: %s",
    +                            "%d attempts, parking until a reconnect is requested: %s",
                                 self.name, _MAX_INITIAL_CONNECT_RETRIES, exc,
                             )
                             self._error = exc
                             self._ready.set()
    -                        return
    +                        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 initial "
    +                            "connection failures (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
     
                         logger.warning(
                             "MCP server '%s' initial connection failed "
    @@ -2520,41 +2578,49 @@ async def run(self, config: dict):
                         )
                         return
     
    -                retries += 1
    -                if retries > _MAX_RECONNECT_RETRIES:
    +                self._reconnect_retries += 1
    +                if self._reconnect_retries > _MAX_RECONNECT_RETRIES:
                         logger.warning(
                             "MCP server '%s' failed after %d reconnection attempts, "
    -                        "parking until a reconnect is requested: %s",
    -                        self.name, _MAX_RECONNECT_RETRIES, exc,
    +                        "parking; will self-probe every %ds until it recovers: %s",
    +                        self.name, _MAX_RECONNECT_RETRIES,
    +                        _PARKED_RETRY_INTERVAL, exc,
                         )
                         # Do NOT return — exiting the task orphans the server:
    -                    # nothing would ever listen for _reconnect_event again,
    -                    # so a half-open circuit-breaker probe could never revive
    -                    # it and the server would be permanently wedged for the
    +                    # nothing would ever listen for _reconnect_event again
    +                    # and the server would be permanently wedged for the
                         # life of the process (#16788). Instead, drop the phantom
    -                    # tools from the registry and park as a dormant listener.
    -                    # A future _reconnect_event.set() — from the breaker's
    -                    # half-open probe, OAuth recovery, or a manual /mcp
    -                    # refresh — wakes us to rebuild the transport (respawning
    -                    # a dead stdio subprocess in the process).
    +                    # tools from the registry and park. Because parking
    +                    # deregisters the tools, no tool call can reach the
    +                    # circuit-breaker half-open probe or _signal_reconnect —
    +                    # so the park is a TIMED wait: every _PARKED_RETRY_INTERVAL
    +                    # we wake and attempt one reconnect ourselves (#57129).
    +                    # An explicit _reconnect_event.set() (OAuth recovery,
    +                    # manual /mcp refresh) still wakes us immediately.
                         self._deregister_tools()
                         self._reconnect_event.clear()
    -                    parked = await self._wait_for_reconnect_or_shutdown()
    +                    parked = await self._wait_for_reconnect_or_shutdown(
    +                        timeout=_PARKED_RETRY_INTERVAL
    +                    )
                         if parked == "shutdown":
                             return
                         logger.info(
    -                        "MCP server '%s': reconnect requested while parked; "
    +                        "MCP server '%s': attempting revival from parked state "
    +                        "(self-probe or explicit reconnect request); "
                             "rebuilding transport.",
                             self.name,
                         )
    -                    retries = 0
    +                    # One probe attempt per wake: budget of 1 so a still-dead
    +                    # server parks again for another interval instead of
    +                    # burning 5 rapid retries each cycle.
    +                    self._reconnect_retries = _MAX_RECONNECT_RETRIES
                         backoff = 1.0
                         continue
     
                     logger.warning(
                         "MCP server '%s' connection lost (attempt %d/%d), "
                         "reconnecting in %.0fs: %s",
    -                    self.name, retries, _MAX_RECONNECT_RETRIES,
    +                    self.name, self._reconnect_retries, _MAX_RECONNECT_RETRIES,
                         backoff, exc,
                     )
                     await asyncio.sleep(backoff)