fix(mcp): review round 2 — busy-skip the refresh pass, fail-fast list pairs, health-tick refresh retry

- _refresh_server never parks on a held connect lock: the holder is
  itself a catalog publisher whose publish supersedes the pass, and
  parking burned refresh_sync's whole 30s budget on ONE busy server (a
  reconnect attempt holds the lock up to 45s), failing the operator
  pass for every healthy server queued behind it. Busy → skip (None),
  no publish, no status writes; the identity/state recheck stays as
  belt-and-braces for the one-tick check→acquire race.
- _list_resource_pair: the ONE copy of the paired resources/templates
  list protocol (both twins). Fail-fast — a fast real error (auth /
  method rejection) surfaces as ITSELF instead of being masked behind
  a hung sibling's eventual 30s TimeoutError — with the survivor
  CANCELLED and REAPED inside the timeout scope, never left detached
  on the shared session.
- Health-tick refresh retry: there is NO periodic refresh pass
  (removed in eb2a119d; the docs still claimed the 4h tier — fixed),
  so a push refresh that failed while the transport stayed up had no
  automatic recovery and the shared catalog stayed stale for every
  user until an operator intervened. Failures and busy-skips arm
  _static_refresh_retry via the shared recorder; the health tick
  drains it with one bounded, lock-serialized full pass per tick;
  success, session drops, removal, and the post-reconnect spawns
  clear it. This also un-latches the error pill: the retry's
  completion clears it within a tick.
- _record_refresh_failure: the bearer-redaction policy (type +
  message, never exc_info) lives exactly once; all three
  refresh-failure sites route through it.
- Static runner discards its coalesce marker only AFTER the
  lock-identity check: on the superseded path a marker present in the
  set belongs to the re-added generation's parked runner, and
  discarding it would mint duplicates past the one-parked-runner
  bound (the pool runner deliberately differs — nothing else clears
  pool markers, so its marker is its own to release).
- _clear_static_push_state: the ONE (server, kind) keyspace walk for
  stamps + retry flag (+ markers on removal).
- Tests: busy-skip, superseded-no-status, fail-fast + reap (<5s
  bound), retry arm/drain/re-arm/clear quartet, logged-wrapper
  contract updated to the shared recorder's arg shape; vacuous
  stamp-math test deleted (behavioral per-kind coverage retained);
  _free_port/_wait_tcp_ready/_wait_session_live hoisted to conftest
  for both live tests.

Refs #839

CHANGELOG.md

@@ -177,10 +177,20 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   per-user pool paths — a tools push no longer swallows a prompts push
   arriving in the same 5-second window (previously the second push was
   dropped outright, and the change stayed invisible until the server pushed
-  that kind again). The resource-refresh fan-out on both paths also no
-  longer orphans its sibling list call when one of the pair fails fast —
-  both calls now complete inside the timeout scope before the failure is
-  re-raised.
+  that kind again). The resource-refresh fan-out on both paths no longer
+  orphans its sibling list call when one of the pair fails fast — the real
+  error surfaces immediately (not masked as a 30-second timeout) and the
+  surviving sibling is cancelled inside the timeout scope. A push refresh
+  that fails while the connection stays up is now retried automatically on
+  the next health-loop tick until one completes — previously a single
+  transient blip left the shared catalog stale for every user on the node
+  until an operator intervened, since a server that already announced its
+  change never announces it again. And an operator `/mcp refresh` no longer
+  parks behind a busy per-server connect lock (a slow reconnect attempt
+  could eat the whole 30-second refresh budget and fail the pass for every
+  healthy server behind it) — the busy server is skipped, since whatever
+  holds the lock publishes a fresher catalog than the skipped pass would
+  have.
 
 - **OpenAI Responses streaming: truncated and refused responses no longer
   vanish.** A response that hit `max_output_tokens` terminates the stream

docs/tools.md

@@ -779,7 +779,10 @@ MCP tool lists stay up-to-date without restart through two mechanisms:
 1. **Push notifications** -- MCP servers that declare `tools.listChanged: true` in
    their capabilities send `notifications/tools/list_changed` when their tool list
    changes. `MCPClientManager` registers a `message_handler` on each `ClientSession`
-   that triggers an immediate refresh for that server.
+   that triggers an immediate refresh for that server (debounced per server and
+   notification kind, and run off the receive loop). A refresh that fails while
+   the connection stays up is retried automatically on the next health-loop tick
+   until one completes.
 
 2. **Manual** -- `/mcp refresh` re-fetches tools from all servers immediately.
    `/mcp refresh <server>` targets a single server. If a server has disconnected,
@@ -787,6 +790,10 @@ MCP tool lists stay up-to-date without restart through two mechanisms:
    same controls (refresh / reconnect buttons per server) for cluster-wide
    fan-out.
 
+Reconnects (health-loop, dispatch-driven, or operator-forced) always end in a
+full catalog rediscovery, so a server that changed its tools while disconnected
+comes back current.
+
 When tools change, `MCPClientManager` rebuilds its merged tool list using copy-on-write
 (new list/dict objects assigned atomically) and notifies all active `ChatSession`
 instances via registered listener callbacks. Each session rebuilds its `_tools`,
@@ -857,13 +864,16 @@ catalog.
 
 ### Refresh
 
-Resource lists stay current through the same three-tier mechanism as tool lists:
+Resource lists stay current through the same mechanisms as tool lists:
 
 1. **Push** -- Servers declaring `resources.listChanged: true` send
-   `notifications/resources/list_changed`, triggering an immediate refresh.
-2. **Periodic** -- Servers without push are polled on the configured refresh
-   interval (default 4 hours, same timer as tools).
-3. **Manual** -- `/mcp refresh` re-fetches resources alongside tools.
+   `notifications/resources/list_changed`, triggering an immediate refresh
+   (with the same failed-refresh retry on the health-loop tick).
+2. **Manual** -- `/mcp refresh` re-fetches resources alongside tools.
+
+Servers without push support are refreshed whenever they reconnect (every
+reconnect ends in full rediscovery) or when an operator refreshes manually;
+there is no periodic polling.
 
 ---
 

tests/conftest.py

@@ -4,6 +4,7 @@ import asyncio
 import contextlib
 import logging
 import os
+import socket
 import threading
 import time
 from typing import TYPE_CHECKING, Any
@@ -243,6 +244,41 @@ def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) ->
     _run_on_loop(loop, _drain())
 
 
+def _free_port() -> int:
+    """Grab an ephemeral localhost port for a live-server subprocess.
+
+    Shared by the live MCP smoke tests (flaky-server, push-refresh) so
+    the socket-probe helpers stay in one place instead of drifting per
+    file.
+    """
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return int(s.getsockname()[1])
+
+
+def _wait_tcp_ready(port: int, timeout: float) -> bool:
+    """Poll until something accepts TCP on 127.0.0.1:*port* (live tests)."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
+                return True
+        except OSError:
+            time.sleep(0.05)
+    return False
+
+
+def _wait_session_live(mgr: MCPClientManager, name: str, timeout: float) -> bool:
+    """Poll until static server *name* has a live session (live tests)."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        state = mgr._static_servers.get(name)
+        if state is not None and state.session is not None:
+            return True
+        time.sleep(0.05)
+    return False
+
+
 def make_oidc_test_config(**overrides: Any) -> OIDCConfig:
     """Build a test ``OIDCConfig`` with sensible defaults.
 

tests/test_mcp_client.py

@@ -2878,15 +2878,10 @@ class TestNotificationDebounce:
         last_b = mgr._last_notification_refresh.get(("srv_b", "tools"), 0.0)
         assert now - last_b >= mgr._NOTIFICATION_DEBOUNCE
 
-    def test_debounce_is_per_kind(self):
-        mgr = MCPClientManager({})
-        mgr._last_notification_refresh[("srv", "tools")] = time.monotonic()
-        # A prompts notification for the SAME server must pass: refreshes
-        # are kind-scoped, so a server-scoped stamp would drop it outright
-        # with no parked runner to observe the change.
-        now = time.monotonic()
-        last_prompts = mgr._last_notification_refresh.get(("srv", "prompts"), 0.0)
-        assert now - last_prompts >= mgr._NOTIFICATION_DEBOUNCE
+    # Per-kind independence is covered behaviorally (through the real
+    # handler) by TestStaticNotificationRefresh.
+    # test_different_kind_notification_not_debounced — a stamp-math
+    # variant here would exercise nothing but dict arithmetic.
 
 
 # ---------------------------------------------------------------------------
@@ -3005,11 +3000,13 @@ class TestStaticNotificationRefresh:
         it re-arms the handler on every notification, so a fast-failing
         server spawns unthrottled refresh tasks at its notification rate.
         The bounded cost — a change announced in the remainder of the
-        failed window waits for the server's next ``list_changed`` or the
-        next reconnect — is the lesser failure (every teardown pops the
+        failed window waits for the health tick's retry (armed by this
+        failure), the server's next ``list_changed``, or the next
+        reconnect — is the lesser failure (every teardown pops the
         stamp, so a reconnect's first notification refreshes immediately).
-        The recorded operator error is type-name-only: the full exception
-        chain can carry the configured bearer for ``auth_type=static``."""
+        The recorded operator error is ``type: message`` — never the
+        serialized exception chain, which can carry the configured
+        bearer for ``auth_type=static``."""
         mgr, loop, _thread = running_loop_mgr
 
         async def _seed() -> None:
@@ -3028,6 +3025,10 @@ class TestStaticNotificationRefresh:
         # diagnostic and header-free; exc_info's chained request carries
         # the configured bearer.
         assert mgr._last_error["srv"] == "Refresh failed: TimeoutError: slow server"
+        # The failure armed the health-tick retry — the only automatic
+        # driver left (no periodic pass; the server may never re-push).
+        assert "srv" in mgr._static_refresh_retry
+        mgr._static_refresh_retry.discard("srv")
 
         async def _ok(_name: str) -> tuple[list[str], list[str]]:
             return [], []
@@ -3356,13 +3357,12 @@ class TestStaticNotificationRefresh:
         )
 
     def test_refresh_server_superseded_by_remove_returns_none(self, running_loop_mgr) -> None:
-        """A ``_refresh_server`` pass that parked on a lock retired by
-        ``remove_server_sync`` must publish nothing and write NO status:
-        the re-add's discovery owns the new generation, and this pass
-        running its list calls under the retired lock would be a second,
-        unserialized catalog publisher — the exact race the lock exists
-        to close. It must also not resurrect ``_last_refresh`` /
-        ``_last_error`` rows for the (possibly gone) server."""
+        """A ``_refresh_server`` pass whose server was removed before it
+        ran must publish nothing and write NO status: the removal
+        cleaned the status maps, and running the list calls would
+        resurrect ``_last_refresh`` / ``_last_error`` rows for a server
+        that no longer exists (or stamp a false ``ok`` over a re-added
+        generation this pass never refreshed)."""
         mgr, loop, _thread = running_loop_mgr
         refreshed: list[str] = []
 
@@ -3375,17 +3375,11 @@ class TestStaticNotificationRefresh:
         mgr._refresh_server_prompts = _rec  # type: ignore[method-assign]
 
         async def _scenario() -> Any:
-            old_lock = mgr._static_connect_lock_for("srv")
-            _seed_static_state(mgr, "srv", session=MagicMock())
-            await old_lock.acquire()
-            task = asyncio.ensure_future(mgr._refresh_server("srv"))
-            for _ in range(3):
-                await asyncio.sleep(0)
-            # Simulate remove → re-add while parked: retire the lock.
-            mgr._static_connect_locks.pop("srv", None)
             mgr._static_connect_lock_for("srv")
-            old_lock.release()
-            return await task
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            # Remove completed before the pass ran: state gone, lock free.
+            mgr._static_servers.pop("srv")
+            return await mgr._refresh_server("srv")
 
         result = _run_on_loop(loop, _scenario())
         assert result is None
@@ -3393,13 +3387,49 @@ class TestStaticNotificationRefresh:
         assert "srv" not in mgr._last_refresh, "superseded pass must not write a status row"
         assert "srv" not in mgr._last_error
 
+    def test_refresh_server_skips_when_lock_busy(self, running_loop_mgr) -> None:
+        """A manual/periodic-style pass must NOT park on a held connect
+        lock: the holder is itself a catalog publisher whose publish
+        supersedes the pass, and parking would burn ``refresh_sync``'s
+        whole 30s budget on one busy server (a reconnect attempt holds
+        the lock up to 45s), failing the pass for every healthy server
+        queued behind it. Skip → ``None``, no publish, no status row."""
+        mgr, loop, _thread = running_loop_mgr
+        refreshed: list[str] = []
+
+        async def _rec(_name: str) -> tuple[list[str], list[str]]:
+            refreshed.append("ran")
+            return [], []
+
+        mgr._refresh_server_tools = _rec  # type: ignore[method-assign]
+        mgr._refresh_server_resources = _rec  # type: ignore[method-assign]
+        mgr._refresh_server_prompts = _rec  # type: ignore[method-assign]
+
+        async def _scenario() -> Any:
+            lock = mgr._static_connect_lock_for("srv")
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            await lock.acquire()
+            try:
+                # Returns IMMEDIATELY (no parking) even though the lock
+                # is held — a parked pass would deadlock this scenario.
+                return await mgr._refresh_server("srv")
+            finally:
+                lock.release()
+
+        result = _run_on_loop(loop, _scenario())
+        assert result is None
+        assert refreshed == []
+        assert "srv" not in mgr._last_refresh
+        assert "srv" not in mgr._last_error
+
     def test_refresh_server_logged_swallows_failure(self, running_loop_mgr) -> None:
         """The spawned post-reconnect pass has no caller to observe a
         re-raise; an escaping exception would reach ``_spawn_background``'s
         ``exc_info`` failure log, which serializes the bearer-carrying
         request chain for ``auth_type=static`` servers. The wrapper must
-        swallow, record the pill, and leave the error row written by
-        ``_refresh_server``."""
+        swallow, record the pill, leave the error row written by
+        ``_refresh_server`` — and ARM the health-tick retry, the only
+        automatic driver that can converge the catalog afterwards."""
         mgr, loop, _thread = running_loop_mgr
 
         async def _boom(_name: str) -> tuple[list[str], list[str]]:
@@ -3417,38 +3447,101 @@ class TestStaticNotificationRefresh:
         _run_on_loop(loop, _scenario())
         assert mgr._last_refresh["srv"][1] == "error:TimeoutError"
         assert mgr._last_error["srv"] == "Refresh failed: TimeoutError: slow server"
+        assert "srv" in mgr._static_refresh_retry
 
-    def test_resources_refresh_does_not_orphan_sibling_on_fast_failure(self) -> None:
-        """Fail-fast gather leaves the surviving list call running
-        DETACHED — outside the timeout scope and the lock serialization
-        — as an unbounded request on the shared session. With
-        ``return_exceptions=True`` both calls complete before the first
-        failure is re-raised."""
+    def test_refresh_server_logged_rearms_retry_on_busy_skip(self, running_loop_mgr) -> None:
+        """A skipped pass (connect lock busy) must RE-ARM the health-tick
+        retry: the lock holder may be a single-kind notification refresh,
+        not the full pass the retry wanted, so a skip is not convergence."""
+        mgr, loop, _thread = running_loop_mgr
+
+        async def _scenario() -> None:
+            lock = mgr._static_connect_lock_for("srv")
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            await lock.acquire()
+            try:
+                await mgr._refresh_server_logged("srv")
+            finally:
+                lock.release()
+
+        _run_on_loop(loop, _scenario())
+        assert "srv" in mgr._static_refresh_retry
+
+    def test_health_tick_drains_refresh_retry(self, running_loop_mgr) -> None:
+        """An armed retry flag + a live session → the health pass spawns
+        one full lock-serialized refresh and clears the flag (a failure
+        inside that refresh re-arms it for the next tick)."""
+        mgr, loop, _thread = running_loop_mgr
+        refreshed: list[str] = []
+
+        async def _rec(name: str) -> tuple[list[str], list[str]]:
+            refreshed.append(name)
+            return [], []
+
+        mgr._refresh_server_tools = _rec  # type: ignore[method-assign]
+        mgr._refresh_server_resources = _rec  # type: ignore[method-assign]
+        mgr._refresh_server_prompts = _rec  # type: ignore[method-assign]
+
+        async def _scenario() -> None:
+            mgr._static_connect_lock_for("srv")
+            session = MagicMock()
+            session.send_ping = AsyncMock()
+            _seed_static_state(mgr, "srv", session=session)
+            mgr._static_refresh_retry.add("srv")
+            await mgr._static_health_one("srv", time.monotonic())
+
+        _run_on_loop(loop, _scenario())
+        _drain_background(mgr, loop)
+        assert refreshed == ["srv", "srv", "srv"], "retry must run the FULL three-kind pass"
+        assert "srv" not in mgr._static_refresh_retry
+
+    def test_session_drop_clears_refresh_retry(self, running_loop_mgr) -> None:
+        """Every session drop clears the retry flag — the reconnect's
+        full rediscovery supersedes the pending retry."""
+        mgr, loop, _thread = running_loop_mgr
+
+        async def _scenario() -> None:
+            state = _seed_static_state(mgr, "srv", session=MagicMock())
+            mgr._static_refresh_retry.add("srv")
+            mgr._drop_static_session_and_stamp("srv", state)
+
+        _run_on_loop(loop, _scenario())
+        assert "srv" not in mgr._static_refresh_retry
+
+    def test_resources_refresh_fails_fast_and_reaps_sibling(self) -> None:
+        """A fast real error must surface as ITSELF — not be masked
+        behind a hung sibling's eventual 30s ``TimeoutError`` — and the
+        surviving sibling must be CANCELLED and REAPED inside the scope,
+        never left running detached (outside the timeout scope and the
+        lock serialization) on the shared session."""
         mgr = MCPClientManager({})
-        sibling_completed: list[bool] = []
+        sibling_events: list[str] = []
 
-        async def _slow_resources() -> Any:
-            await asyncio.sleep(0.05)
-            sibling_completed.append(True)
-            result = MagicMock()
-            result.resources = []
-            return result
+        async def _hanging_resources() -> Any:
+            try:
+                await asyncio.sleep(30)  # would mask the real error as TimeoutError
+            except asyncio.CancelledError:
+                sibling_events.append("cancelled")
+                raise
+            sibling_events.append("completed")
 
         async def _fast_fail_templates() -> Any:
             raise RuntimeError("method not found")
 
         session = MagicMock()
-        session.list_resources = _slow_resources
+        session.list_resources = _hanging_resources
         session.list_resource_templates = _fast_fail_templates
         _seed_static_state(mgr, "srv", session=session, supports_resources=True)
 
         async def _run() -> None:
             await mgr._refresh_server_resources("srv")
 
+        start = time.monotonic()
         with pytest.raises(RuntimeError, match="method not found"):
             asyncio.run(_run())
-        assert sibling_completed == [True], (
-            "the sibling list call must complete inside the scope, not be orphaned"
+        assert time.monotonic() - start < 5, "real error must surface fast, not at timeout"
+        assert sibling_events == ["cancelled"], (
+            "the hung sibling must be cancelled and reaped inside the scope"
         )
 
     def test_remove_server_clears_markers_and_stamps(self) -> None:
@@ -3467,11 +3560,11 @@ class TestStaticNotificationRefresh:
         assert not mgr._static_refresh_pending
         assert ("srv", "tools") not in mgr._last_notification_refresh
 
-    def test_refresh_server_serializes_on_connect_lock(self, running_loop_mgr) -> None:
-        """The manual/periodic refresh is a catalog publisher too: it
-        must queue behind the same per-name lock as connect wiring and
-        the notification runner, or its older snapshot can land over a
-        notification refresh's newer one."""
+    def test_refresh_server_runs_and_publishes_when_lock_free(self, running_loop_mgr) -> None:
+        """The happy path: lock free → the pass acquires it, runs all
+        three kind refreshes under it, and records the ``ok`` row.
+        (Serialization against a BUSY lock is the skip test above — the
+        pass never runs concurrently with another publisher.)"""
         mgr, loop, _thread = running_loop_mgr
         refreshed: list[str] = []
 
@@ -3486,21 +3579,15 @@ class TestStaticNotificationRefresh:
         mgr._refresh_server_resources = _rec_none  # type: ignore[method-assign]
         mgr._refresh_server_prompts = _rec_none  # type: ignore[method-assign]
 
-        async def _scenario() -> bool:
-            lock = mgr._static_connect_lock_for("srv")
+        async def _scenario() -> Any:
+            mgr._static_connect_lock_for("srv")
             _seed_static_state(mgr, "srv", session=MagicMock())
-            await lock.acquire()
-            task = asyncio.ensure_future(mgr._refresh_server("srv"))
-            for _ in range(3):
-                await asyncio.sleep(0)
-            held_back = len(refreshed) == 0
-            lock.release()
-            await task
-            return held_back
+            return await mgr._refresh_server("srv")
 
-        held_back = _run_on_loop(loop, _scenario())
-        assert held_back, "manual refresh must wait for the connect lock"
+        result = _run_on_loop(loop, _scenario())
+        assert result == ([], [])
         assert refreshed == ["srv"]
+        assert mgr._last_refresh["srv"][1] == "ok"
 
 
 # ---------------------------------------------------------------------------
@@ -3765,7 +3852,7 @@ class TestCBAutoReconnectRefresh:
                 warn_calls = [
                     c
                     for c in mock_log.warning.call_args_list
-                    if "Post-reconnect catalog refresh failed" in str(c.args[0])
+                    if "Background catalog refresh" in str(c.args)
                 ]
                 time.sleep(0.02)
             # The tracked task must also fully drain (emptiness now implies
@@ -3784,10 +3871,12 @@ class TestCBAutoReconnectRefresh:
             "the refresh failure must be logged by _refresh_server_logged, not "
             "left for GC-time reporting"
         )
-        # type + message as structured args; exc_info must NOT be passed.
+        # type + message as structured args (via _record_refresh_failure:
+        # fmt, context, name, type-name, message); exc_info must NOT be
+        # passed.
         assert warn_calls[0].kwargs.get("exc_info") is None
-        assert warn_calls[0].args[2] == "RuntimeError"
-        assert "catalog fetch broke" in str(warn_calls[0].args[3])
+        assert warn_calls[0].args[3] == "RuntimeError"
+        assert "catalog fetch broke" in str(warn_calls[0].args[4])
 
 
 def _run_hl(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:

tests/test_mcp_live_flaky_server.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 import asyncio
 import gc
 import signal
-import socket
 import subprocess
 import sys
 import textwrap
@@ -29,6 +28,7 @@ from unittest.mock import patch
 
 import pytest
 
+from tests.conftest import _free_port, _wait_session_live, _wait_tcp_ready
 from turnstone.core.mcp_client import MCPClientManager
 
 if TYPE_CHECKING:
@@ -57,33 +57,6 @@ SERVER_SRC = textwrap.dedent(
 ).lstrip()
 
 
-def _free_port() -> int:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("127.0.0.1", 0))
-        return int(s.getsockname()[1])
-
-
-def _wait_tcp_ready(port: int, timeout: float) -> bool:
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        try:
-            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
-                return True
-        except OSError:
-            time.sleep(0.05)
-    return False
-
-
-def _wait_session_live(mgr: MCPClientManager, name: str, timeout: float) -> bool:
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        state = mgr._static_servers.get(name)
-        if state is not None and state.session is not None:
-            return True
-        time.sleep(0.05)
-    return False
-
-
 async def _armed_scope_count() -> int:
     """Armed scopes hosted on THIS (the mcp) loop — mirrors the production
     disarm sweep's scoping, and keeps an unrelated scope on another loop that

tests/test_mcp_live_push_refresh.py

@@ -30,7 +30,6 @@ Self-contained (spawns its own server; no LLM backend, no network beyond
 
 from __future__ import annotations
 
-import socket
 import subprocess
 import sys
 import textwrap
@@ -40,6 +39,7 @@ from unittest.mock import patch
 
 import pytest
 
+from tests.conftest import _free_port, _wait_session_live, _wait_tcp_ready
 from turnstone.core.mcp_client import MCPClientManager
 
 if TYPE_CHECKING:
@@ -81,33 +81,6 @@ SERVER_SRC = textwrap.dedent(
 ).lstrip()
 
 
-def _free_port() -> int:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("127.0.0.1", 0))
-        return int(s.getsockname()[1])
-
-
-def _wait_tcp_ready(port: int, timeout: float) -> bool:
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        try:
-            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
-                return True
-        except OSError:
-            time.sleep(0.05)
-    return False
-
-
-def _wait_session_live(mgr: MCPClientManager, name: str, timeout: float) -> bool:
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        state = mgr._static_servers.get(name)
-        if state is not None and state.session is not None:
-            return True
-        time.sleep(0.05)
-    return False
-
-
 def _wait_tool_visible(mgr: MCPClientManager, server: str, tool: str, timeout: float) -> bool:
     """Poll the per-server catalog for *tool* — the push refresh landing."""
     deadline = time.monotonic() + timeout

turnstone/core/mcp_client.py

@@ -730,6 +730,16 @@ class MCPClientManager:
         # the connect lock's waiter queue at ONE parked runner per
         # server+kind. Mirrors ``_pool_refresh_pending``.
         self._static_refresh_pending: set[tuple[str, str]] = set()
+        # Servers whose last static-path refresh FAILED (or was skipped as
+        # lock-busy) while their transport stayed up. There is NO periodic
+        # refresh pass — push + manual + post-reconnect are the only
+        # drivers — so without this a catalog whose one ``list_changed``
+        # push failed on a transient blip stays stale for every user on
+        # the node until an operator intervenes. The health tick drains
+        # this set: one bounded, lock-serialized full refresh per tick
+        # until a pass completes. Cleared by every session drop / removal
+        # (the reconnect's rediscovery supersedes the retry).
+        self._static_refresh_retry: set[str] = set()
 
         # Last refresh outcome (Phase 9 — admin status indicator).  Per-
         # server tuple of ``(unix_ts, outcome)`` where outcome is one of
@@ -1338,21 +1348,40 @@ class MCPClientManager:
             )
         return disarmed
 
+    def _clear_static_push_state(self, name: str, *, markers: bool) -> None:
+        """Clear per-server push-refresh bookkeeping — the ONE keyspace walk.
+
+        Covers the per-(server, kind) debounce stamps, the health-tick
+        refresh-retry flag, and (``markers=True``) the coalesce markers.
+        ``markers=False`` for session drops: a parked runner still owns
+        its marker and releases it itself (at-acquire or via its gated
+        ``finally``), and clobbering it would let the handler mint a
+        duplicate runner. ``markers=True`` for REMOVAL: retired-lock
+        runners bail at the identity check WITHOUT touching the set, so
+        a stale marker left behind would coalesce away a re-added
+        server's first push with no runner covering it.
+        """
+        self._static_refresh_retry.discard(name)
+        for kind in _LIST_CHANGED_KINDS.values():
+            self._last_notification_refresh.pop((name, kind), None)
+            if markers:
+                self._static_refresh_pending.discard((name, kind))
+
     def _drop_static_session_and_stamp(self, name: str, state: StaticServerState) -> None:
-        """Null the session AND pop its notification debounce stamp (paired).
+        """Null the session AND its push bookkeeping (stamps + retry flag).
 
         The static twin of :meth:`_drop_session_and_stamp` (pool). The
         stamp must not outlive the transport: the keep-stamp-on-failure
         design leans on every teardown popping it, so a reconnected
         transport's first ``list_changed`` refreshes immediately instead
-        of being debounced against a pre-collapse stamp. An eviction site
-        that nulled the session directly would silently re-open that
-        stale-stamp hole. Safe when the session is already gone: both
-        halves are idempotent.
+        of being debounced against a pre-collapse stamp — and the
+        reconnect's full rediscovery supersedes any pending refresh
+        retry. An eviction site that nulled the session directly would
+        silently re-open that stale-stamp hole. Safe when the session is
+        already gone: all parts are idempotent.
         """
         state.session = None
-        for kind in _LIST_CHANGED_KINDS.values():
-            self._last_notification_refresh.pop((name, kind), None)
+        self._clear_static_push_state(name, markers=False)
 
     async def _teardown_static_session(self, name: str) -> None:
         """Tear down a static server's session/transport (the ONE canonical order).
@@ -1570,12 +1599,14 @@ class MCPClientManager:
 
         Deliberate change from the pre-#839 handler: the error pill
         (``_last_error``) is no longer cleared on ANY incoming
-        notification — only a COMPLETED refresh (push, periodic, or
-        reconnect-driven) clears it, because a notification's arrival
-        proves nothing about whether the previous refresh failure
-        resolved. A transient push-refresh failure can therefore show in
-        the pill until the next refresh attempt (worst case: the periodic
-        pass), where the old handler cleared it on the next notification.
+        notification — only a COMPLETED refresh clears it, because a
+        notification's arrival proves nothing about whether the previous
+        refresh failure resolved. A transient push-refresh failure shows
+        in the pill until the health tick's refresh RETRY completes
+        (:meth:`_static_health_one` — there is NO periodic refresh pass;
+        push, manual, post-reconnect, and that retry are the only
+        drivers), where the old handler cleared it, dishonestly, on the
+        next notification of any kind.
         """
 
         async def _on_notification(
@@ -1707,13 +1738,22 @@ class MCPClientManager:
         acquired = False
         try:
             async with lock:
-                self._static_refresh_pending.discard(marker)
-                acquired = True
                 if self._static_connect_locks.get(name) is not lock:
                     # Removed (and possibly re-added) while we were parked:
                     # the notification belonged to the old transport, and a
                     # re-add publishes its own discovery under the NEW lock.
+                    # Do NOT touch the pending set here — OUR marker was
+                    # discarded by ``remove_server_sync``, and a marker
+                    # present NOW belongs to the new generation's parked
+                    # runner; discarding it would let the handler mint a
+                    # duplicate past the one-parked-runner bound. (The pool
+                    # runner deliberately differs: nothing else clears pool
+                    # markers, so on ITS superseded path the marker is
+                    # still its own to release.)
+                    acquired = True  # marker is not ours — finally must not touch it
                     return
+                self._static_refresh_pending.discard(marker)
+                acquired = True
                 state = self._static_servers.get(name)
                 if state is None or state.session is None:
                     # Torn down / evicted while we were parked. The
@@ -1724,19 +1764,9 @@ class MCPClientManager:
                 await refresh(name)
                 self._last_error.pop(name, None)
         except (Exception, BaseExceptionGroup) as exc:
-            # ``type: str(exc)``, never ``exc_info`` — the message text is
-            # diagnostic and header-free; it is the serialized exception
-            # CHAIN (chained ``httpx.Request`` whose headers carry the
-            # configured bearer for ``auth_type=static``) that must never
-            # reach the log.
-            log.warning(
-                "Static %s refresh after notification failed for '%s' exc=%s: %s",
-                kind,
-                name,
-                type(exc).__name__,
-                exc,
+            self._record_refresh_failure(
+                name, exc, context=f"Static {kind} refresh after notification"
             )
-            self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}")
         finally:
             if not acquired:
                 # Cancelled (or failed) while PARKED — the marker still in
@@ -3990,6 +4020,41 @@ class MCPClientManager:
             )
         return added, removed
 
+    async def _list_resource_pair(self, session: Any) -> tuple[Any, Any]:
+        """``list_resources`` + ``list_resource_templates`` in one bounded RTT.
+
+        The ONE copy of the paired-list protocol for both refresh twins
+        (:meth:`_refresh_server_resources` /
+        :meth:`_refresh_pool_server_resources`). Both calls share the
+        timeout budget and target disjoint catalogs (resources vs.
+        templates), so ordering is irrelevant.
+
+        Fail-FAST on the first real error — a fast, meaningful failure
+        (an auth or method rejection) must surface as ITSELF, not be
+        masked behind a hung sibling's eventual ``TimeoutError`` — but
+        with the surviving sibling CANCELLED and REAPED inside this
+        scope before the error re-raises: bare fail-fast ``gather``
+        leaves the survivor running detached — outside the timeout scope
+        and outside the lock serialization — as an unbounded in-flight
+        request on the shared session. On timeout expiry both tasks are
+        cancelled together and the caller sees ``TimeoutError``.
+        """
+        async with asyncio.timeout(self._CONNECT_TIMEOUT):
+            res_task = asyncio.create_task(session.list_resources())
+            tmpl_task = asyncio.create_task(session.list_resource_templates())
+            try:
+                res_result, tmpl_result = await asyncio.gather(res_task, tmpl_task)
+            except BaseException:
+                # First failure (or our own cancellation, incl. the
+                # timeout's): cancel the pair — a done task ignores it —
+                # and REAP both so nothing survives detached and no
+                # exception goes unretrieved, then surface the original.
+                for task in (res_task, tmpl_task):
+                    task.cancel()
+                await asyncio.gather(res_task, tmpl_task, return_exceptions=True)
+                raise
+            return res_result, tmpl_result
+
     async def _refresh_pool_server_resources(
         self, key: tuple[str, str]
     ) -> tuple[list[str], list[str]]:
@@ -4014,30 +4079,7 @@ class MCPClientManager:
         user_id, server_name = key
         old_uris = {r["uri"] for r in (entry.resources or []) if not r.get("template")}
 
-        async with asyncio.timeout(self._CONNECT_TIMEOUT):
-            # 1-RTT (gather) instead of 2 sequential RTTs — both calls
-            # share the same timeout budget and target disjoint catalogs
-            # (resources vs. templates), so ordering is irrelevant.
-            # ``return_exceptions=True`` so a fast failure on one call
-            # cannot orphan the sibling: fail-fast gather leaves the
-            # survivor running DETACHED — outside this timeout scope and
-            # outside the lock serialization — as an unbounded in-flight
-            # request on the shared session. Both awaitables complete
-            # here (the timeout cancels them together on expiry), then
-            # the first failure is re-raised.
-            pool_res_pair: tuple[Any, Any] = await asyncio.gather(
-                session.list_resources(),
-                session.list_resource_templates(),
-                return_exceptions=True,
-            )
-            res_result, tmpl_result = pool_res_pair
-        pool_res_exc: BaseException | None = next(
-            (r for r in (res_result, tmpl_result) if isinstance(r, BaseException)), None
-        )
-        if pool_res_exc is not None:
-            raise pool_res_exc
-        assert not isinstance(res_result, BaseException)
-        assert not isinstance(tmpl_result, BaseException)
+        res_result, tmpl_result = await self._list_resource_pair(session)
         if self._user_pool_entries.get(key) is not entry:
             # Entry replaced mid-flight — stale result, discard.
             return [], []
@@ -4150,6 +4192,34 @@ class MCPClientManager:
             )
         return added, removed
 
+    def _record_refresh_failure(self, name: str, exc: BaseException, *, context: str) -> None:
+        """Record a static-path refresh failure — the ONE redaction policy copy.
+
+        ``type: str(exc)`` and NEVER ``exc_info``: the message text (server
+        error / URL) is diagnostic and header-free, while the serialized
+        exception CHAIN carries the chained ``httpx.Request`` whose headers
+        hold the configured bearer for ``auth_type=static`` servers. Every
+        static refresh-failure path routes through here so a drive-by
+        "improve logging" edit cannot silently reintroduce the leak at one
+        hand-synced copy.
+
+        Also arms the health-tick refresh retry: there is no periodic
+        refresh pass, so this is what converges a catalog whose one
+        ``list_changed`` push failed on a transient blip. Config-gated —
+        a failure observed for a just-removed server must not park a
+        retry flag nothing will ever drain.
+        """
+        log.warning(
+            "%s failed for MCP server '%s' exc=%s: %s",
+            context,
+            name,
+            type(exc).__name__,
+            exc,
+        )
+        self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}")
+        if name in self._server_configs:
+            self._static_refresh_retry.add(name)
+
     async def _refresh_server(self, name: str) -> tuple[list[str], list[str]] | None:
         """Re-fetch tools, resources, and prompts for one server.
 
@@ -4189,16 +4259,16 @@ class MCPClientManager:
         static per-server catalog — a connect's discovery wiring, a
         spawned notification refresh
         (:meth:`_run_static_notification_refresh`), and this
-        manual/periodic pass — writes under the same lock, so each list
+        manual/retry pass — writes under the same lock, so each list
         call is issued only after the previous publisher finished and
         the last publish is always the freshest. An unserialized pass
         could land its older snapshot over a notification refresh's
         newer one, with no convergence until the next change. No caller
         holds the lock coming in: ``_refresh_all`` takes it per-branch
-        (never nested), and the two post-reconnect schedule sites spawn
-        this as its own task (through
-        :meth:`_refresh_server_logged` — the raise below must never
-        reach ``_spawn_background``'s ``exc_info`` failure log).
+        (never nested), and the spawn sites (post-reconnect, health-tick
+        retry) go through :meth:`_refresh_server_logged` — the raise
+        below must never reach ``_spawn_background``'s ``exc_info``
+        failure log.
 
         The post-acquire recheck closes the remove → re-add race the
         same way :meth:`_ensure_static_connected` does — by LOCK
@@ -4212,14 +4282,35 @@ class MCPClientManager:
         recheck below keeps it inert.
         """
         lock = self._static_connect_lock_for(name)
+        if lock.locked():
+            # Contended: the holder IS a catalog publisher — a connect's
+            # discovery wiring, a spawned notification refresh, or a
+            # sibling pass — whose publish supersedes this one. Parking
+            # here would burn the operator's whole ``refresh_sync``
+            # budget (default 30s) on ONE busy server (a reconnect
+            # attempt holds this lock for up to 45s) and fail the pass
+            # for every healthy server queued behind it, where the
+            # pre-serialization code never waited at all. The one-tick
+            # check→acquire race window below is accepted: the sync
+            # boundary cancels a parked pass at its own budget, and
+            # ``_refresh_server_logged`` re-arms the health-tick retry
+            # on a skip.
+            log.info(
+                "Refresh pass for '%s' skipped: connect lock busy "
+                "(the holder's publish supersedes this pass)",
+                name,
+            )
+            return None
         async with lock:
             if (
                 self._static_connect_locks.get(name) is not lock
                 or self._static_servers.get(name) is None
             ):
-                # Superseded while parked: removed (state gone), or
-                # removed + re-added (lock retired — the new generation
-                # publishes its own discovery under the NEW lock).
+                # Superseded: removed (state gone), or removed + re-added
+                # (lock retired — the new generation publishes its own
+                # discovery under the NEW lock). With the busy-skip above
+                # this is belt-and-braces: it fires only when the lock
+                # was won through the one-tick check→acquire race.
                 return None
             results = await asyncio.gather(
                 self._refresh_server_tools(name),
@@ -4253,21 +4344,24 @@ class MCPClientManager:
         escaping exception would land in ``_spawn_background``'s
         done-callback, whose ``exc_info`` log serializes the chained
         ``httpx.Request`` — headers carrying the configured bearer for
-        ``auth_type=static`` servers. Swallow here with the same
-        ``type: str(exc)`` shape as :meth:`_refresh_all`'s except;
-        ``_refresh_server`` already wrote the ``_last_refresh`` error
-        row before re-raising, so no outcome is lost — only the leak.
+        ``auth_type=static`` servers. Failures route through
+        :meth:`_record_refresh_failure` (redacted log + pill + retry
+        arm); ``_refresh_server`` already wrote the ``_last_refresh``
+        error row before re-raising, so no outcome is lost — only the
+        leak.
+
+        A ``None`` (skipped: lock busy / superseded) RE-ARMS the
+        health-tick retry when the server still exists: the lock holder
+        may be a single-kind notification refresh, not the full pass a
+        retry wanted, so a skip must not count as convergence.
         """
         try:
-            await self._refresh_server(name)
+            refreshed = await self._refresh_server(name)
         except (Exception, BaseExceptionGroup) as exc:
-            log.warning(
-                "Post-reconnect catalog refresh failed for '%s' exc=%s: %s",
-                name,
-                type(exc).__name__,
-                exc,
-            )
-            self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}")
+            self._record_refresh_failure(name, exc, context="Background catalog refresh")
+        else:
+            if refreshed is None and name in self._server_configs:
+                self._static_refresh_retry.add(name)
 
     async def _refresh_all(
         self, server_name: str | None = None
@@ -4321,28 +4415,18 @@ class MCPClientManager:
             except (Exception, BaseExceptionGroup) as exc:
                 # BaseExceptionGroup: a transport task-group failure from the
                 # reconnect/refresh must stay isolated to this server, not
-                # abort the whole refresh pass.
-                #
-                # ``type: str(exc)``, never ``exc_info`` — the serialized
-                # exception CHAIN carries the chained ``httpx.Request``
-                # whose headers hold the configured bearer for
-                # ``auth_type=static`` servers; the message text is
-                # diagnostic and header-free.
-                log.warning(
-                    "Refresh failed for MCP server '%s' exc=%s: %s",
-                    name,
-                    type(exc).__name__,
-                    exc,
-                )
-                self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}")
+                # abort the whole refresh pass. Redaction + retry-arm live
+                # in the shared helper.
+                self._record_refresh_failure(name, exc, context="Refresh")
                 results[name] = ([], [])
                 # A dead transport leaves a non-None but unusable session, so
                 # the reconnect branch at the top of this loop (gated on
-                # ``session is None``) would never fire and we would re-probe
-                # the corpse on every tick forever — the exact failure that
+                # ``session is None``) would never fire and every later pass
+                # would re-probe the corpse forever — the exact failure that
                 # required a full process restart to clear. Evicting the
-                # session here makes the NEXT refresh tick reconnect, turning
-                # this periodic refresh into a self-healing liveness probe.
+                # session here hands recovery to the health loop (and makes
+                # the next manual pass reconnect), turning any refresh pass
+                # into a self-healing liveness probe.
                 if _is_dead_transport(exc):
                     dead_state = self._static_servers.get(name)
                     if dead_state is not None:
@@ -4456,29 +4540,7 @@ class MCPClientManager:
         # turn the second list_resource_templates() call into AttributeError.
         session = state.session
 
-        # ``asyncio.timeout`` mandatory — a wedged server must not hang a
-        # spawned refresh (and the connect lock it holds) forever.
-        async with asyncio.timeout(self._CONNECT_TIMEOUT):
-            # 1-RTT (gather) instead of 2 sequential RTTs — both calls
-            # share the same timeout budget and target disjoint catalogs
-            # (resources vs. templates), so ordering is irrelevant.
-            # ``return_exceptions=True`` so a fast failure on one call
-            # cannot orphan the sibling as a detached, unbounded request
-            # on the shared session — see
-            # :meth:`_refresh_pool_server_resources`.
-            static_res_pair: tuple[Any, Any] = await asyncio.gather(
-                session.list_resources(),
-                session.list_resource_templates(),
-                return_exceptions=True,
-            )
-            res_result, tmpl_result = static_res_pair
-        static_res_exc: BaseException | None = next(
-            (r for r in (res_result, tmpl_result) if isinstance(r, BaseException)), None
-        )
-        if static_res_exc is not None:
-            raise static_res_exc
-        assert not isinstance(res_result, BaseException)
-        assert not isinstance(tmpl_result, BaseException)
+        res_result, tmpl_result = await self._list_resource_pair(session)
         if self._static_servers.get(name) is not state:
             # Entry replaced (remove + re-add) mid-flight — stale result.
             return
@@ -5052,6 +5114,7 @@ class MCPClientManager:
         self._circuit_trip_count.clear()
         self._last_notification_refresh.clear()
         self._static_refresh_pending.clear()
+        self._static_refresh_retry.clear()
         self._last_pool_notification_refresh.clear()
         self._pool_refresh_pending.clear()
         # Pool state already cleared above when the loop was alive; this
@@ -5265,18 +5328,16 @@ class MCPClientManager:
                     # Close session + transport via the owner close protocol
                     await self._teardown_static_session(name)
                     # Clean up per-server state (on the event loop thread).
-                    # The direct stamp pops back up the teardown call above:
+                    # The push-state clear backs up the teardown call above:
                     # ``_teardown_static_session`` early-returns (no pop) when
-                    # the state entry is already gone. The marker discards
-                    # keep a parked old-generation runner's marker from
-                    # coalescing AWAY a re-added server's first push — that
-                    # runner bails at its lock-identity check without
-                    # refreshing, so nothing would cover the dropped change.
+                    # the state entry is already gone. ``markers=True`` keeps
+                    # a parked old-generation runner's marker from coalescing
+                    # AWAY a re-added server's first push — that runner bails
+                    # at its lock-identity check without refreshing, so
+                    # nothing would cover the dropped change.
                     self._static_servers.pop(name, None)
                     self._last_error.pop(name, None)
-                    for kind in _LIST_CHANGED_KINDS.values():
-                        self._last_notification_refresh.pop((name, kind), None)
-                        self._static_refresh_pending.discard((name, kind))
+                    self._clear_static_push_state(name, markers=True)
                     self._cb_clear(name)
                     # Clear health-loop backoff/ping state so a later re-add of
                     # the same name doesn't inherit stale ``due`` deadlines.
@@ -5310,9 +5371,7 @@ class MCPClientManager:
             # No event loop (tests / pre-start) — mutate directly
             self._static_servers.pop(name, None)
             self._last_error.pop(name, None)
-            for kind in _LIST_CHANGED_KINDS.values():
-                self._last_notification_refresh.pop((name, kind), None)
-                self._static_refresh_pending.discard((name, kind))
+            self._clear_static_push_state(name, markers=True)
             self._cb_clear(name)
             self._rebuild_tools()
             self._rebuild_resources()
@@ -5972,9 +6031,25 @@ class MCPClientManager:
 
     async def _static_health_one(self, name: str, now: float) -> float:
         """Ping a connected server or reconnect a disconnected one; return its
-        next-due monotonic deadline. Runs concurrently per server in the tick."""
+        next-due monotonic deadline. Runs concurrently per server in the tick.
+
+        Also the refresh-RETRY driver: a push-driven (or manual) refresh
+        that failed while the transport stayed up has no other automatic
+        recovery — there is no periodic refresh pass, and the server may
+        never push that ``list_changed`` again — so the catalog would
+        stay stale for every user on the node until an operator
+        intervened. One bounded, lock-serialized full pass is spawned
+        per tick while the flag stays armed (a failure or busy-skip
+        re-arms it; success, session drop, or removal clears it).
+        """
         state = self._static_servers.get(name)
         if state is not None and state.session is not None:
+            if name in self._static_refresh_retry:
+                self._static_refresh_retry.discard(name)
+                self._spawn_background(
+                    self._refresh_server_logged(name),
+                    f"catalog refresh retry for '{name}'",
+                )
             return await self._static_ping_one(name, now)
         return await self._static_reconnect_one(name)
 
@@ -6052,6 +6127,11 @@ class MCPClientManager:
         self._static_reconnect_next.pop(name, None)
         next_ping = time.monotonic() + self._static_health_check_s
         self._static_next_ping[name] = next_ping
+        # The full pass spawned below IS the refresh a pending retry flag
+        # wants (a reconnect-branch failure arms the flag without a session
+        # drop to clear it) — discard it so the next tick doesn't run a
+        # redundant second pass.
+        self._static_refresh_retry.discard(name)
         self._spawn_background(
             self._refresh_server_logged(name),
             f"catalog refresh after static health reconnect '{name}'",
@@ -6218,6 +6298,10 @@ class MCPClientManager:
         # scheduling itself.
         def _schedule_refresh() -> None:
             try:
+                # This full pass IS the refresh a pending retry flag wants —
+                # discard it so the next health tick doesn't run a redundant
+                # second pass (mirrors ``_static_reconnect_one``).
+                self._static_refresh_retry.discard(server_name)
                 self._spawn_background(
                     self._refresh_server_logged(server_name),
                     f"catalog refresh after reconnect for '{server_name}'",