fix(mcp): review round 1 — supersede retired-lock refreshes, per-kind debounce, complete gather pairs

- _refresh_server: post-acquire lock-identity + state-existence recheck;
  a pass superseded by remove (or remove + re-add) returns None and
  writes NO status — it must not run its list calls as a second,
  unserialized publisher against the re-add's discovery wiring,
  resurrect status rows for a removed server, or stamp a false "ok"
  over a generation it never refreshed. _refresh_all treats None as a
  deliberate skip (no breaker success record).
- Debounce stamps are per (server, kind) on BOTH paths: refreshes are
  kind-scoped, so a server-scoped stamp dropped a different-kind
  notification inside the window outright — a tools push swallowed the
  prompts push 100ms behind it, and nothing observed the prompt change
  until the server pushed that kind again. Teardown pops loop the
  kinds; remove_server_sync also discards the server's coalesce
  markers so a parked old-generation runner's marker cannot coalesce
  away a re-added server's first push.
- Resource refreshers (static + pool) gather with
  return_exceptions=True: fail-fast gather left the surviving list
  call running detached — outside the timeout scope and the lock
  serialization — as an unbounded in-flight request on the shared
  session.
- Spawned post-reconnect refreshes route through _refresh_server_logged:
  the re-raise escaped into _spawn_background's done-callback, whose
  exc_info log serializes the chained httpx.Request carrying the
  configured bearer for auth_type=static servers; _refresh_all's
  except drops exc_info for the same reason. Failure diagnostics widen
  to "Type: message" in logs and the error pill — the message text is
  header-free; only the serialized chain leaks.
- Accepted + documented: connect-lock contention on dispatch
  reconnects is bounded to one in-flight list call (parked runners
  bail instantly post-eviction); the error pill persists until the
  next COMPLETED refresh (a notification's arrival proves nothing
  about whether the failure resolved).
- Tests: per-kind debounce independence, superseded-pass writes
  nothing, gather-sibling completion, logged-wrapper swallow with the
  exc_info channel asserted SILENT, remove clears markers;
  _run_on_loop/_drain_background hoisted to conftest (4 drifted
  copies); proc.kill() portability in the live push test.

Runner-twin dedup (static/pool protocol duplication) deferred to #842.

Refs #839

CHANGELOG.md

@@ -173,6 +173,14 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   that same lock, so a slower publisher can no longer land a staler catalog
   over a fresher one. Every teardown path now also clears the notification
   debounce stamp, so a reconnected server's first push refreshes immediately.
+  Push-refresh debouncing is now per (server, kind) on BOTH the static and
+  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.
 
 - **OpenAI Responses streaming: truncated and refused responses no longer
   vanish.** A response that hit `max_output_tokens` terminates the stream

tests/conftest.py

@@ -216,6 +216,33 @@ def _seed_static_state(mgr: MCPClientManager, name: str, **overrides: Any) -> St
     return state
 
 
+def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any, timeout: float = 10) -> Any:
+    """Submit *coro* to *loop*, wait for the result.
+
+    The ONE copy shared by the MCP test files — four hand-synced copies
+    had already drifted on the timeout (5s hardcoded vs a 10s default).
+    The timeout is an upper bound on waiting, not a behavior assertion,
+    so the most generous variant won the merge.
+    """
+    fut = asyncio.run_coroutine_threadsafe(coro, loop)
+    return fut.result(timeout=timeout)
+
+
+def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) -> None:
+    """Deterministically await ``mgr``'s tracked background tasks.
+
+    Replaces fixed sleeps for synchronizing with scheduled dead-grant
+    drops / spawned refreshes: exact, and immune to slow-runner flake.
+    """
+
+    async def _drain() -> None:
+        tasks = [t for t in list(mgr._background_tasks) if not t.done()]
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    _run_on_loop(loop, _drain())
+
+
 def make_oidc_test_config(**overrides: Any) -> OIDCConfig:
     """Build a test ``OIDCConfig`` with sensible defaults.
 

tests/test_mcp_client.py

@@ -16,7 +16,7 @@ import anyio
 import mcp.types as mcp_types
 import pytest
 
-from tests.conftest import _seed_static_state
+from tests.conftest import _drain_background, _run_on_loop, _seed_static_state
 from turnstone.core.mcp_client import (
     MCPClientManager,
     _db_servers_to_config,
@@ -2851,63 +2851,49 @@ class TestSafeTransportStreams:
 
 
 class TestNotificationDebounce:
-    """Verify notification-triggered refreshes are debounced."""
+    """Verify notification-triggered refreshes are debounced per (server, kind)."""
 
     def test_debounce_within_window(self):
         mgr = MCPClientManager({})
-        mgr._last_notification_refresh["srv"] = time.monotonic()
-        # We can't easily call _on_notification (it's a closure), so test
-        # the debounce logic directly via the timestamp check
+        mgr._last_notification_refresh[("srv", "tools")] = time.monotonic()
+        # Stamp math only — the handler-level behavior (including the
+        # per-kind independence) is covered by TestStaticNotificationRefresh.
         now = time.monotonic()
-        last = mgr._last_notification_refresh.get("srv", 0.0)
+        last = mgr._last_notification_refresh.get(("srv", "tools"), 0.0)
         assert now - last < mgr._NOTIFICATION_DEBOUNCE
 
     def test_debounce_passes_after_window(self):
         mgr = MCPClientManager({})
         # Set timestamp well in the past
-        mgr._last_notification_refresh["srv"] = time.monotonic() - 10
+        mgr._last_notification_refresh[("srv", "tools")] = time.monotonic() - 10
         now = time.monotonic()
-        last = mgr._last_notification_refresh.get("srv", 0.0)
+        last = mgr._last_notification_refresh.get(("srv", "tools"), 0.0)
         assert now - last >= mgr._NOTIFICATION_DEBOUNCE
 
     def test_debounce_is_per_server(self):
         mgr = MCPClientManager({})
-        mgr._last_notification_refresh["srv_a"] = time.monotonic()
+        mgr._last_notification_refresh[("srv_a", "tools")] = time.monotonic()
         # srv_b has no timestamp — should pass debounce
         now = time.monotonic()
-        last_b = mgr._last_notification_refresh.get("srv_b", 0.0)
+        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
+
 
 # ---------------------------------------------------------------------------
 # Static-path list_changed refresh — spawned runner protocol (#839)
 # ---------------------------------------------------------------------------
 
 
-def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:
-    """Submit *coro* to *loop*, wait for the result with a 5s timeout.
-
-    Mirrors ``test_mcp_user_pool.py``'s helper of the same name.
-    """
-    fut = asyncio.run_coroutine_threadsafe(coro, loop)
-    return fut.result(timeout=5)
-
-
-def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) -> None:
-    """Deterministically await the manager's tracked background tasks.
-
-    Replaces fixed sleeps for synchronizing with spawned refreshes:
-    exact, and immune to slow-runner flake.
-    """
-
-    async def _drain() -> None:
-        tasks = [t for t in list(mgr._background_tasks) if not t.done()]
-        if tasks:
-            await asyncio.gather(*tasks, return_exceptions=True)
-
-    _run_on_loop(loop, _drain())
-
-
 class TestStaticNotificationRefresh:
     """The static ``*/list_changed`` protocol: spawn / serialize / coalesce.
 
@@ -2953,7 +2939,7 @@ class TestStaticNotificationRefresh:
         assert refreshed == ["srv"]
         # Debounce stamp consumed at schedule time (storm dedupe); the
         # runner released its coalesce marker at lock acquire.
-        assert "srv" in mgr._last_notification_refresh
+        assert ("srv", "tools") in mgr._last_notification_refresh
         assert not mgr._static_refresh_pending
 
     def test_handler_ignores_non_list_changed_messages(self, running_loop_mgr) -> None:
@@ -2968,7 +2954,7 @@ class TestStaticNotificationRefresh:
             return len(mgr._background_tasks) - before
 
         assert _run_on_loop(loop, _fire()) == 0
-        assert "srv" not in mgr._last_notification_refresh
+        assert ("srv", "tools") not in mgr._last_notification_refresh
         assert not mgr._static_refresh_pending
 
     def test_notification_coalesces_when_refresh_already_queued(self, running_loop_mgr) -> None:
@@ -3001,7 +2987,7 @@ class TestStaticNotificationRefresh:
             await handler(note)
             # Age the stamp past the debounce window: the MARKER (not
             # the stamp) must do the suppression for the second fire.
-            mgr._last_notification_refresh["srv"] = time.monotonic() - 10.0
+            mgr._last_notification_refresh[("srv", "tools")] = time.monotonic() - 10.0
             await handler(note)
             spawned = len(mgr._background_tasks) - before
             lock.release()
@@ -3031,21 +3017,24 @@ class TestStaticNotificationRefresh:
             _seed_static_state(mgr, "srv", session=MagicMock())
 
         _run_on_loop(loop, _seed())
-        mgr._last_notification_refresh["srv"] = 123.0
+        mgr._last_notification_refresh[("srv", "tools")] = 123.0
 
         async def _boom(_name: str) -> tuple[list[str], list[str]]:
             raise TimeoutError("slow server")
 
         _run_on_loop(loop, mgr._run_static_notification_refresh("srv", "tools", _boom))
-        assert mgr._last_notification_refresh["srv"] == 123.0
-        assert mgr._last_error["srv"] == "Refresh failed: TimeoutError"
+        assert mgr._last_notification_refresh[("srv", "tools")] == 123.0
+        # type + message, never the serialized chain: the message text is
+        # diagnostic and header-free; exc_info's chained request carries
+        # the configured bearer.
+        assert mgr._last_error["srv"] == "Refresh failed: TimeoutError: slow server"
 
         async def _ok(_name: str) -> tuple[list[str], list[str]]:
             return [], []
 
-        mgr._last_notification_refresh["srv"] = 456.0
+        mgr._last_notification_refresh[("srv", "tools")] = 456.0
         _run_on_loop(loop, mgr._run_static_notification_refresh("srv", "tools", _ok))
-        assert mgr._last_notification_refresh["srv"] == 456.0
+        assert mgr._last_notification_refresh[("srv", "tools")] == 456.0
         assert "srv" not in mgr._last_error  # success clears the error
 
     def test_notification_refresh_catches_exception_group(self, running_loop_mgr) -> None:
@@ -3065,14 +3054,16 @@ class TestStaticNotificationRefresh:
             _seed_static_state(mgr, "srv", session=MagicMock())
 
         _run_on_loop(loop, _seed())
-        mgr._last_notification_refresh["srv"] = 123.0
+        mgr._last_notification_refresh[("srv", "tools")] = 123.0
 
         async def _wedge(_name: str) -> tuple[list[str], list[str]]:
             raise BaseExceptionGroup("wedged transport", [asyncio.CancelledError()])
 
         _run_on_loop(loop, mgr._run_static_notification_refresh("srv", "tools", _wedge))
-        assert mgr._last_notification_refresh["srv"] == 123.0
-        assert mgr._last_error["srv"] == "Refresh failed: BaseExceptionGroup"
+        assert mgr._last_notification_refresh[("srv", "tools")] == 123.0
+        assert mgr._last_error["srv"] == (
+            "Refresh failed: BaseExceptionGroup: wedged transport (1 sub-exception)"
+        )
 
     def test_notification_refresh_serializes_on_connect_lock(self, running_loop_mgr) -> None:
         """The runner must take the per-name connect lock before
@@ -3239,11 +3230,13 @@ class TestStaticNotificationRefresh:
 
         async def _scenario() -> None:
             _seed_static_state(mgr, "srv", session=MagicMock())
-            mgr._last_notification_refresh["srv"] = 123.0
+            mgr._last_notification_refresh[("srv", "tools")] = 123.0
+            mgr._last_notification_refresh[("srv", "prompts")] = 123.0
             await mgr._teardown_static_session("srv")
 
         _run_on_loop(loop, _scenario())
-        assert "srv" not in mgr._last_notification_refresh
+        assert ("srv", "tools") not in mgr._last_notification_refresh
+        assert ("srv", "prompts") not in mgr._last_notification_refresh
 
     def test_owner_death_pops_debounce_stamp(self, running_loop_mgr) -> None:
         """The unrequested-collapse path (owner done-callback) is a
@@ -3259,21 +3252,21 @@ class TestStaticNotificationRefresh:
             task = asyncio.ensure_future(_noop())
             await task
             state.owner_task = task
-            mgr._last_notification_refresh["srv"] = 123.0
+            mgr._last_notification_refresh[("srv", "tools")] = 123.0
             mgr._on_static_owner_death("srv", task)
 
         _run_on_loop(loop, _scenario())
-        assert "srv" not in mgr._last_notification_refresh
+        assert ("srv", "tools") not in mgr._last_notification_refresh
 
     def test_dead_transport_eviction_pops_debounce_stamp(self) -> None:
         """The dispatch-observed transport-failure eviction is a teardown
         too — the stamp must not survive the session it throttled."""
         mgr = MCPClientManager({})
         _seed_static_state(mgr, "srv", session=MagicMock())
-        mgr._last_notification_refresh["srv"] = 123.0
+        mgr._last_notification_refresh[("srv", "tools")] = 123.0
         mgr._record_and_evict_on_dead_transport("srv", anyio.ClosedResourceError())
         assert mgr._static_servers["srv"].session is None
-        assert "srv" not in mgr._last_notification_refresh
+        assert ("srv", "tools") not in mgr._last_notification_refresh
 
     def test_refresh_server_tools_bounded_by_timeout(self) -> None:
         """A wedged server's list call must not hang a spawned refresh
@@ -3322,6 +3315,158 @@ class TestStaticNotificationRefresh:
         assert mgr._static_servers["srv"].tools == []  # new entry untouched
         assert old_state.tools == []  # stale result not published
 
+    def test_different_kind_notification_not_debounced(self, running_loop_mgr) -> None:
+        """A tools push must not swallow a prompts push arriving inside
+        the same debounce window: refreshes are kind-scoped, so a
+        server-scoped stamp would drop the prompts notification outright
+        — no runner exists or is spawned for prompts, and the new prompt
+        never appears until the server pushes again (the same staleness
+        class #839 was opened to fix)."""
+        mgr, loop, _thread = running_loop_mgr
+        refreshed: list[str] = []
+
+        async def _rec_tools(_name: str) -> tuple[list[str], list[str]]:
+            refreshed.append("tools")
+            return [], []
+
+        async def _rec_prompts(_name: str) -> None:
+            refreshed.append("prompts")
+
+        mgr._refresh_server_tools = _rec_tools  # type: ignore[method-assign]
+        mgr._refresh_server_prompts = _rec_prompts  # type: ignore[method-assign]
+        handler = mgr._make_static_notification_handler("srv")
+
+        async def _fire_both() -> None:
+            mgr._static_connect_lock_for("srv")
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            tools_note = mcp_types.ServerNotification(
+                mcp_types.ToolListChangedNotification(method="notifications/tools/list_changed")
+            )
+            prompts_note = mcp_types.ServerNotification(
+                mcp_types.PromptListChangedNotification(method="notifications/prompts/list_changed")
+            )
+            await handler(tools_note)
+            # Immediately inside the tools stamp's window.
+            await handler(prompts_note)
+
+        _run_on_loop(loop, _fire_both())
+        _drain_background(mgr, loop)
+        assert sorted(refreshed) == ["prompts", "tools"], (
+            "a same-window different-kind notification must spawn its own refresh"
+        )
+
+    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."""
+        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:
+            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
+
+        result = _run_on_loop(loop, _scenario())
+        assert result is None
+        assert refreshed == []
+        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_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``."""
+        mgr, loop, _thread = running_loop_mgr
+
+        async def _boom(_name: str) -> tuple[list[str], list[str]]:
+            raise TimeoutError("slow server")
+
+        mgr._refresh_server_tools = _boom  # type: ignore[method-assign]
+        mgr._refresh_server_resources = _boom  # type: ignore[method-assign]
+        mgr._refresh_server_prompts = _boom  # type: ignore[method-assign]
+
+        async def _scenario() -> None:
+            mgr._static_connect_lock_for("srv")
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            await mgr._refresh_server_logged("srv")  # must not raise
+
+        _run_on_loop(loop, _scenario())
+        assert mgr._last_refresh["srv"][1] == "error:TimeoutError"
+        assert mgr._last_error["srv"] == "Refresh failed: TimeoutError: slow server"
+
+    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."""
+        mgr = MCPClientManager({})
+        sibling_completed: list[bool] = []
+
+        async def _slow_resources() -> Any:
+            await asyncio.sleep(0.05)
+            sibling_completed.append(True)
+            result = MagicMock()
+            result.resources = []
+            return result
+
+        async def _fast_fail_templates() -> Any:
+            raise RuntimeError("method not found")
+
+        session = MagicMock()
+        session.list_resources = _slow_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")
+
+        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"
+        )
+
+    def test_remove_server_clears_markers_and_stamps(self) -> None:
+        """``remove_server_sync`` must clear the server's coalesce
+        markers: a parked old-generation runner's marker would otherwise
+        coalesce AWAY a re-added server's first push, and that runner
+        bails at its lock-identity check without refreshing — the pushed
+        change would be silently dropped."""
+        mgr = MCPClientManager({})  # no loop → direct-mutation branch
+        _seed_static_state(mgr, "srv", session=None)
+        mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"}
+        mgr._static_refresh_pending.add(("srv", "tools"))
+        mgr._static_refresh_pending.add(("srv", "prompts"))
+        mgr._last_notification_refresh[("srv", "tools")] = 123.0
+        mgr.remove_server_sync("srv")
+        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
@@ -3583,10 +3728,12 @@ class TestCBAutoReconnectRefresh:
         assert session is new_session
 
     def test_auto_reconnect_retrieves_and_logs_refresh_failure(self, running_loop_mgr):
-        """A refresh failure must be RETRIEVED and logged by the task's
-        done-callback — not abandoned for asyncio to report as "Task exception
-        was never retrieved" at GC time (which lands on whatever stream pytest
-        has attached by then: the closed-file CI spew)."""
+        """A refresh failure must be logged INSIDE ``_refresh_server_logged``
+        (type + message, no ``exc_info``) — not by the done-callback, whose
+        ``exc_info`` log serializes the chained ``httpx.Request`` carrying the
+        configured bearer for ``auth_type=static`` servers, and not abandoned
+        for asyncio to report as "Task exception was never retrieved" at GC
+        time (the closed-file CI spew)."""
         import threading as _threading
 
         mgr, _loop, _thread = running_loop_mgr
@@ -3616,7 +3763,9 @@ class TestCBAutoReconnectRefresh:
             warn_calls = []
             while not warn_calls and time.time() < deadline:
                 warn_calls = [
-                    c for c in mock_log.warning.call_args_list if "MCP background" in str(c.args[0])
+                    c
+                    for c in mock_log.warning.call_args_list
+                    if "Post-reconnect catalog refresh failed" in str(c.args[0])
                 ]
                 time.sleep(0.02)
             # The tracked task must also fully drain (emptiness now implies
@@ -3625,14 +3774,20 @@ class TestCBAutoReconnectRefresh:
             while mgr._background_tasks and time.time() < deadline:
                 time.sleep(0.02)
             assert not mgr._background_tasks, "background refresh task never drained"
+            # The done-callback's exc_info channel must have stayed silent:
+            # the wrapper swallowed the failure before it could escape.
+            assert not [
+                c for c in mock_log.warning.call_args_list if "MCP background" in str(c.args[0])
+            ], "failure escaped to _spawn_background's exc_info log (bearer-leak channel)"
         assert session is new_session
         assert warn_calls, (
-            "the refresh failure must be logged by the done-callback, not left "
-            "for GC-time reporting"
+            "the refresh failure must be logged by _refresh_server_logged, not "
+            "left for GC-time reporting"
         )
-        exc = warn_calls[0].kwargs.get("exc_info")
-        assert isinstance(exc, RuntimeError)
-        assert "catalog fetch broke" in str(exc)
+        # type + message as structured args; 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])
 
 
 def _run_hl(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:

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 signal
 import socket
 import subprocess
 import sys
@@ -186,5 +185,5 @@ class TestPushRefreshNoDeadlock:
             if mgr is not None:
                 mgr.shutdown()
             if proc is not None:
-                proc.send_signal(signal.SIGKILL)
+                proc.kill()
                 proc.wait(timeout=5)

tests/test_mcp_pool_auth_introspection.py

@@ -34,7 +34,7 @@ from unittest.mock import AsyncMock, MagicMock
 import httpx
 import pytest
 
-from tests.conftest import make_mcp_token_cipher, stop_loop_thread
+from tests.conftest import _run_on_loop, make_mcp_token_cipher, stop_loop_thread
 from turnstone.core.mcp_client import (
     MCPClientManager,
     _AuthCapture,
@@ -134,11 +134,6 @@ def running_loop_mgr():
         stop_loop_thread(loop, thread)
 
 
-def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:
-    fut = asyncio.run_coroutine_threadsafe(coro, loop)
-    return fut.result(timeout=5)
-
-
 # ---------------------------------------------------------------------------
 # _classify_failure with capture vs legacy fallback
 # ---------------------------------------------------------------------------

tests/test_mcp_user_catalog.py

@@ -25,6 +25,7 @@ from unittest.mock import MagicMock
 import httpx
 import pytest
 
+from tests.conftest import _run_on_loop
 from turnstone.core.mcp_client import (
     MCPClientManager,
     PoolEntryState,
@@ -64,11 +65,6 @@ def running_loop_mgr() -> Any:
         thread.join(timeout=2)
 
 
-def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any, timeout: float = 10) -> Any:
-    fut = asyncio.run_coroutine_threadsafe(coro, loop)
-    return fut.result(timeout=timeout)
-
-
 def _build_mock_transport_factory(
     mgr: MCPClientManager,
     monkeypatch: pytest.MonkeyPatch,
@@ -577,7 +573,7 @@ def test_close_pool_entry_if_idle_cools_entry_for_live_session_user(
     assert mgr.is_mcp_tool("mcp__pool-srv__do_thing", user_id="user-1") is True
     assert mgr._user_pool_entries[key].in_flight == 0
     # Seed the debounce dict so the prune-on-close is observable.
-    mgr._last_pool_notification_refresh[key] = 0.0
+    mgr._last_pool_notification_refresh[(key, "tools")] = 0.0
 
     user_calls = [0]
     admin_calls = [0]
@@ -606,7 +602,7 @@ def test_close_pool_entry_if_idle_cools_entry_for_live_session_user(
     # dispatcher's next acquire needs the same lock.
     assert key in mgr._user_pool_locks
     # Debounce stamp pruned with the transport.
-    assert key not in mgr._last_pool_notification_refresh
+    assert (key, "tools") not in mgr._last_pool_notification_refresh
     # Per-user catalog view untouched.
     assert "user-1" in mgr._user_tool_map
     assert "user-1" in mgr._user_tools
@@ -655,7 +651,7 @@ def test_close_pool_entry_if_idle_drops_entry_without_live_listener(
     assert mgr.is_mcp_tool("mcp__pool-srv__do_thing", user_id="user-1") is True
     assert mgr._user_pool_entries[key].in_flight == 0
     # Seed the debounce dict so the perf-1 prune is observable.
-    mgr._last_pool_notification_refresh[key] = 0.0
+    mgr._last_pool_notification_refresh[(key, "tools")] = 0.0
 
     admin_calls = [0]
     other_calls = [0]
@@ -678,7 +674,7 @@ def test_close_pool_entry_if_idle_drops_entry_without_live_listener(
     assert key not in mgr._user_pool_last_used
     assert key not in mgr._user_pool_locks
     # perf-1 prune: debounce dict no longer carries the key.
-    assert key not in mgr._last_pool_notification_refresh
+    assert (key, "tools") not in mgr._last_pool_notification_refresh
     # Catalog cleanup ran in BOTH dicts (bug-1 sibling + bug-2 cleanup).
     assert "user-1" not in mgr._user_tool_map
     assert "user-1" not in mgr._user_tools

tests/test_mcp_user_pool.py

@@ -26,7 +26,12 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import mcp.types as mcp_types
 import pytest
 
-from tests.conftest import make_mcp_token_cipher, stop_loop_thread
+from tests.conftest import (
+    _drain_background,
+    _run_on_loop,
+    make_mcp_token_cipher,
+    stop_loop_thread,
+)
 from turnstone.core.mcp_client import MCPClientManager, PoolEntryState
 from turnstone.core.mcp_crypto import MCPTokenStore
 from turnstone.core.storage._sqlite import SQLiteBackend
@@ -146,12 +151,6 @@ def running_loop_mgr():
         stop_loop_thread(loop, thread)
 
 
-def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any:
-    """Submit *coro* to *loop*, wait for the result with a 5s timeout."""
-    fut = asyncio.run_coroutine_threadsafe(coro, loop)
-    return fut.result(timeout=5)
-
-
 def _fake_pool_tools(server_name: str, tool_name: str) -> list[dict[str, Any]]:
     """One OpenAI-format tool entry, shaped as the pool catalog seeds expect.
 
@@ -171,21 +170,6 @@ def _fake_pool_tools(server_name: str, tool_name: str) -> list[dict[str, Any]]:
     ]
 
 
-def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) -> None:
-    """Deterministically await the manager's tracked background tasks.
-
-    Replaces fixed sleeps for synchronizing with scheduled dead-grant
-    drops / spawned refreshes: exact, and immune to slow-runner flake.
-    """
-
-    async def _drain() -> None:
-        tasks = [t for t in list(mgr._background_tasks) if not t.done()]
-        if tasks:
-            await asyncio.gather(*tasks, return_exceptions=True)
-
-    _run_on_loop(loop, _drain())
-
-
 # ---------------------------------------------------------------------------
 # Pool data structures
 # ---------------------------------------------------------------------------
@@ -713,20 +697,20 @@ class TestEviction:
             entry.session = MagicMock()  # runner refreshes only live sessions
 
         _run_on_loop(loop, _seed())
-        mgr._last_pool_notification_refresh[key] = 123.0
+        mgr._last_pool_notification_refresh[(key, "tools")] = 123.0
 
         async def _boom(_key: tuple[str, str]) -> tuple[list[str], list[str]]:
             raise TimeoutError("slow server")
 
         _run_on_loop(loop, mgr._run_notification_refresh(key, "tools", _boom))
-        assert mgr._last_pool_notification_refresh[key] == 123.0
+        assert mgr._last_pool_notification_refresh[(key, "tools")] == 123.0
 
         async def _ok(_key: tuple[str, str]) -> tuple[list[str], list[str]]:
             return [], []
 
-        mgr._last_pool_notification_refresh[key] = 456.0
+        mgr._last_pool_notification_refresh[(key, "tools")] = 456.0
         _run_on_loop(loop, mgr._run_notification_refresh(key, "tools", _ok))
-        assert mgr._last_pool_notification_refresh[key] == 456.0
+        assert mgr._last_pool_notification_refresh[(key, "tools")] == 456.0
 
     def test_notification_refresh_catches_exception_group(self, running_loop_mgr) -> None:
         """A wedged anyio transport surfaces session-op failures as
@@ -746,13 +730,13 @@ class TestEviction:
             entry.session = MagicMock()  # runner refreshes only live sessions
 
         _run_on_loop(loop, _seed())
-        mgr._last_pool_notification_refresh[key] = 123.0
+        mgr._last_pool_notification_refresh[(key, "tools")] = 123.0
 
         async def _wedge(_key: tuple[str, str]) -> tuple[list[str], list[str]]:
             raise BaseExceptionGroup("wedged transport", [asyncio.CancelledError()])
 
         _run_on_loop(loop, mgr._run_notification_refresh(key, "tools", _wedge))
-        assert mgr._last_pool_notification_refresh[key] == 123.0
+        assert mgr._last_pool_notification_refresh[(key, "tools")] == 123.0
 
     def test_notification_refresh_serializes_on_open_lock(self, running_loop_mgr) -> None:
         """The runner must take ``open_lock`` before refreshing: an
@@ -845,7 +829,7 @@ class TestEviction:
             await handler(note)
             # Age the stamp past the debounce window: the MARKER (not
             # the stamp) must do the suppression for the second fire.
-            mgr._last_pool_notification_refresh[key] = time.monotonic() - 10.0
+            mgr._last_pool_notification_refresh[(key, "tools")] = time.monotonic() - 10.0
             await handler(note)
             spawned = len(mgr._background_tasks) - before
             entry.open_lock.release()
@@ -926,9 +910,9 @@ class TestEviction:
             entry.session = MagicMock()
 
         _run_on_loop(loop, _seed())
-        mgr._last_pool_notification_refresh[key] = 123.0
+        mgr._last_pool_notification_refresh[(key, "tools")] = 123.0
         _run_on_loop(loop, mgr._teardown_pool_entry(key))
-        assert key not in mgr._last_pool_notification_refresh
+        assert (key, "tools") not in mgr._last_pool_notification_refresh
 
     def test_owner_death_pops_debounce_stamp(self, running_loop_mgr) -> None:
         """The unrequested-collapse path (owner done-callback) is a
@@ -946,11 +930,11 @@ class TestEviction:
             task = asyncio.ensure_future(_noop())
             await task
             entry.owner_task = task
-            mgr._last_pool_notification_refresh[key] = 123.0
+            mgr._last_pool_notification_refresh[(key, "tools")] = 123.0
             mgr._on_pool_owner_death(key, task)
 
         _run_on_loop(loop, _scenario())
-        assert key not in mgr._last_pool_notification_refresh
+        assert (key, "tools") not in mgr._last_pool_notification_refresh
 
     def test_list_changed_handler_spawns_refresh_off_receive_loop(self, running_loop_mgr) -> None:
         """The notification handler must SPAWN the refresh, not await it:
@@ -987,7 +971,7 @@ class TestEviction:
         _drain_background(mgr, loop)
         assert refreshed == [key]
         # Debounce stamp consumed at schedule time (storm dedupe).
-        assert key in mgr._last_pool_notification_refresh
+        assert (key, "tools") in mgr._last_pool_notification_refresh
 
     def test_lookup_grant_dead_requires_wired_infrastructure(self, running_loop_mgr) -> None:
         """kind='missing' is authoritative only when the stores that

turnstone/core/mcp_client.py

@@ -717,8 +717,11 @@ class MCPClientManager:
         self._circuit_open_until: dict[str, float] = {}  # monotonic timestamp
         self._circuit_trip_count: dict[str, int] = {}  # backoff exponent
 
-        # Notification debounce (per-server)
-        self._last_notification_refresh: dict[str, float] = {}
+        # Notification debounce, keyed ``(server, kind)`` — kind-scoped to
+        # match the kind-scoped refreshes: a server-scoped stamp would drop
+        # a different-kind notification inside the window outright, with no
+        # parked runner to observe the change.
+        self._last_notification_refresh: dict[tuple[str, str], float] = {}
         # Coalescing markers for spawned static ``list_changed`` refreshes,
         # keyed ``(server, kind)``: set at spawn, cleared the moment the
         # runner acquires the per-name connect lock (before its list call).
@@ -796,11 +799,14 @@ class MCPClientManager:
         self._user_prompt_map: dict[str, dict[str, tuple[str, str]]] = {}
         self._user_prompts: dict[str, list[dict[str, Any]]] = {}
 
-        # Notification debounce for pool sessions, keyed ``(user_id, server)``.
-        # Mirrors ``_last_notification_refresh`` (static) but per-pool-key
-        # so a noisy server in one user's pool doesn't suppress a refresh
-        # in another user's pool of the same server.
-        self._last_pool_notification_refresh: dict[tuple[str, str], float] = {}
+        # Notification debounce for pool sessions, keyed
+        # ``((user_id, server), kind)``. Mirrors ``_last_notification_refresh``
+        # (static) but per-pool-key so a noisy server in one user's pool
+        # doesn't suppress a refresh in another user's pool of the same
+        # server — and kind-scoped for the same reason as the static stamp:
+        # refreshes are kind-scoped, so a key-scoped stamp would drop a
+        # different-kind notification inside the window outright.
+        self._last_pool_notification_refresh: dict[tuple[tuple[str, str], str], float] = {}
         # Coalescing markers for spawned ``list_changed`` refreshes, keyed
         # ``((user_id, server), kind)``: set at spawn, cleared the moment
         # the runner acquires ``open_lock`` (before its list call). While
@@ -1345,7 +1351,8 @@ class MCPClientManager:
         halves are idempotent.
         """
         state.session = None
-        self._last_notification_refresh.pop(name, None)
+        for kind in _LIST_CHANGED_KINDS.values():
+            self._last_notification_refresh.pop((name, kind), None)
 
     async def _teardown_static_session(self, name: str) -> None:
         """Tear down a static server's session/transport (the ONE canonical order).
@@ -1556,10 +1563,19 @@ class MCPClientManager:
         response can only be routed by the receive loop that is parked
         awaiting this handler, and while it is parked EVERY in-flight and
         subsequent call on this shared per-node session stalls with it.
-        List-change refreshes are debounced, coalesced per (server, kind),
-        and SPAWNED as tracked tasks
+        List-change refreshes are debounced per (server, kind), coalesced
+        per (server, kind), and SPAWNED as tracked tasks
         (:meth:`_run_static_notification_refresh`), mirroring the pool
         handler (:meth:`_make_pool_notification_handler`).
+
+        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.
         """
 
         async def _on_notification(
@@ -1570,17 +1586,22 @@ class MCPClientManager:
             kind = _LIST_CHANGED_KINDS.get(type(msg.root))
             if kind is None:
                 return
-            # Debounce: skip if we refreshed this server very recently
+            marker = (name, kind)
+            # Debounce per (server, kind): refreshes are kind-scoped, so a
+            # server-scoped stamp would DROP a different-kind notification
+            # landing inside the window (tools push swallowing the prompts
+            # push 100ms behind it) with no parked runner to observe it —
+            # nothing would refresh prompts until the server pushed again.
             now = time.monotonic()
-            last = self._last_notification_refresh.get(name, 0.0)
+            last = self._last_notification_refresh.get(marker, 0.0)
             if now - last < self._NOTIFICATION_DEBOUNCE:
                 log.debug(
-                    "Debouncing notification from '%s' (%.1fs since last refresh)",
+                    "Debouncing %s notification from '%s' (%.1fs since last refresh)",
+                    kind,
                     name,
                     now - last,
                 )
                 return
-            marker = (name, kind)
             if marker in self._static_refresh_pending:
                 # A runner for this server+kind is queued but has not yet
                 # issued its list call — it will observe this change when
@@ -1608,7 +1629,7 @@ class MCPClientManager:
                     "prompts": self._refresh_server_prompts,
                 }
                 log.info("Received %s/list_changed from '%s'", kind, name)
-                self._last_notification_refresh[name] = now
+                self._last_notification_refresh[marker] = now
                 self._static_refresh_pending.add(marker)
                 self._spawn_background(
                     self._run_static_notification_refresh(name, kind, refreshers[kind]),
@@ -1653,11 +1674,26 @@ class MCPClientManager:
           entry identity): ``remove_server_sync`` retires the lock object
           after teardown, so a runner that parked on the OLD lock must not
           touch state now owned by a NEW-lock holder.
-        * ``(Exception, BaseExceptionGroup)`` is caught type-name-only for
-          the same hygiene reason as the pool runner, but the credential at
-          stake is the CONFIGURED bearer: for ``auth_type=static`` servers
-          the chained ``httpx.Request`` headers carry it, and an escaping
+        * ``(Exception, BaseExceptionGroup)`` is caught for the same
+          hygiene reason as the pool runner, but the credential at stake
+          is the CONFIGURED bearer: for ``auth_type=static`` servers the
+          chained ``httpx.Request`` headers carry it, and an escaping
           exception would reach ``_spawn_background``'s ``exc_info`` log.
+          The log line and error pill carry ``type: str(exc)`` — the
+          message text (server error / URL) is diagnostic and header-free;
+          it is ``exc_info``'s serialized request CHAIN that leaks, so
+          that is the only thing withheld.
+
+        Bounded contention residual, accepted: this runner (and the
+        lock-serialized :meth:`_refresh_server`) holds the connect lock
+        for up to one list timeout (``_CONNECT_TIMEOUT``), and a
+        dispatch-driven reconnect queues behind it — but a reconnect only
+        runs after the session was EVICTED, and every parked runner bails
+        instantly on the session check below once that happens, so the
+        added wait is at most the single in-flight list call. A caller
+        timeout expiring on that wait is deliberately not a breaker
+        record (see :meth:`_cb_auto_reconnect`); the next dispatch
+        retries.
         """
         marker = (name, kind)
         lock = self._static_connect_locks.get(name)
@@ -1688,16 +1724,19 @@ class MCPClientManager:
                 await refresh(name)
                 self._last_error.pop(name, None)
         except (Exception, BaseExceptionGroup) as exc:
-            # Structured fields only — ``exc_info`` would serialize the
-            # chained ``httpx.Request`` whose headers carry the configured
-            # bearer for ``auth_type=static`` servers.
+            # ``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",
+                "Static %s refresh after notification failed for '%s' exc=%s: %s",
                 kind,
                 name,
                 type(exc).__name__,
+                exc,
             )
-            self._set_error(name, f"Refresh failed: {type(exc).__name__}")
+            self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}")
         finally:
             if not acquired:
                 # Cancelled (or failed) while PARKED — the marker still in
@@ -2055,17 +2094,21 @@ class MCPClientManager:
             kind = _LIST_CHANGED_KINDS.get(type(msg.root))
             if kind is None:
                 return
+            marker = (key, kind)
+            # Debounce per (key, kind) — see the static handler's rationale:
+            # refreshes are kind-scoped, so a key-scoped stamp would drop a
+            # different-kind notification inside the window outright.
             now = time.monotonic()
-            last = self._last_pool_notification_refresh.get(key, 0.0)
+            last = self._last_pool_notification_refresh.get(marker, 0.0)
             if now - last < self._NOTIFICATION_DEBOUNCE:
                 log.debug(
-                    "Debouncing pool notification user=%s server=%s (%.1fs since last refresh)",
+                    "Debouncing pool %s notification user=%s server=%s (%.1fs since last refresh)",
+                    kind,
                     user_id,
                     server_name,
                     now - last,
                 )
                 return
-            marker = (key, kind)
             if marker in self._pool_refresh_pending:
                 # A runner for this key+kind is queued but has not yet
                 # issued its list call — it will observe this change
@@ -2106,7 +2149,7 @@ class MCPClientManager:
                     user_id,
                     server_name,
                 )
-                self._last_pool_notification_refresh[key] = now
+                self._last_pool_notification_refresh[marker] = now
                 self._pool_refresh_pending.add(marker)
                 self._spawn_background(
                     self._run_notification_refresh(key, kind, refreshers[kind]),
@@ -3975,10 +4018,26 @@ class MCPClientManager:
             # 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.
-            res_result, tmpl_result = await asyncio.gather(
+            # ``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)
         if self._user_pool_entries.get(key) is not entry:
             # Entry replaced mid-flight — stale result, discard.
             return [], []
@@ -4091,11 +4150,19 @@ class MCPClientManager:
             )
         return added, removed
 
-    async def _refresh_server(self, name: str) -> tuple[list[str], list[str]]:
+    async def _refresh_server(self, name: str) -> tuple[list[str], list[str]] | None:
         """Re-fetch tools, resources, and prompts for one server.
 
         Returns ``(added_tools, removed_tools)`` names (tool diff only,
-        for backward compatibility with ``/mcp refresh`` output).
+        for backward compatibility with ``/mcp refresh`` output), or
+        ``None`` when the pass was SUPERSEDED — the server was removed
+        (or removed and re-added) while this pass waited on the lock, so
+        its mandate is gone and it must publish nothing and write no
+        status: the removal cleaned the status maps, and a re-add's
+        connect discovery owns the new generation's status. Writing
+        anything here would either resurrect rows for a nonexistent
+        server or stamp a false ``ok`` over a generation this pass never
+        actually refreshed.
 
         Writes the ``_last_refresh`` entry on every call so the Phase 9
         admin status pill reflects every refresh path — manual
@@ -4129,9 +4196,31 @@ class MCPClientManager:
         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.
+        this as its own task (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
+        IDENTITY: ``remove_server_sync`` retires the lock object after
+        teardown, so a pass that parked on the OLD lock must not run
+        its list calls concurrently with a re-add publishing under the
+        NEW lock (two unserialized publishers — the exact race this
+        lock exists to close). The get-or-create acquire mirrors
+        ``_ensure_static_connected``; a mint for a just-removed name is
+        bounded (a re-add reuses it, shutdown clears it) and the
+        recheck below keeps it inert.
         """
-        async with self._static_connect_lock_for(name):
+        lock = self._static_connect_lock_for(name)
+        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).
+                return None
             results = await asyncio.gather(
                 self._refresh_server_tools(name),
                 self._refresh_server_resources(name),
@@ -4157,6 +4246,29 @@ class MCPClientManager:
             self._last_refresh[name] = (time.time(), "ok")
             return added, removed
 
+    async def _refresh_server_logged(self, name: str) -> None:
+        """:meth:`_refresh_server` for ``_spawn_background`` schedule sites.
+
+        The spawned pass has no caller to observe the re-raise, so an
+        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.
+        """
+        try:
+            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}")
+
     async def _refresh_all(
         self, server_name: str | None = None
     ) -> dict[str, tuple[list[str], list[str]]]:
@@ -4195,15 +4307,34 @@ class MCPClientManager:
                         results[name] = (new_names, [])
                         self._last_refresh[name] = (time.time(), "ok")
                     continue
-                added, removed = await self._refresh_server(name)
+                refreshed = await self._refresh_server(name)
+                if refreshed is None:
+                    # Superseded (removed / removed+re-added while parked
+                    # on the lock) — a deliberate skip, not an outcome:
+                    # nothing was refreshed, so record neither success
+                    # nor failure for whatever generation lives now.
+                    results[name] = ([], [])
+                    continue
+                added, removed = refreshed
                 self._cb_record_success(name)
                 results[name] = (added, removed)
             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.
-                log.warning("Refresh failed for MCP server '%s'", name, exc_info=True)
-                self._set_error(name, f"Refresh failed: {exc}")
+                #
+                # ``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}")
                 results[name] = ([], [])
                 # A dead transport leaves a non-None but unusable session, so
                 # the reconnect branch at the top of this loop (gated on
@@ -4331,11 +4462,23 @@ class MCPClientManager:
             # 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.
-            # Mirrors :meth:`_refresh_pool_server_resources`.
-            res_result, tmpl_result = await asyncio.gather(
+            # ``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)
         if self._static_servers.get(name) is not state:
             # Entry replaced (remove + re-add) mid-flight — stale result.
             return
@@ -5121,10 +5264,19 @@ class MCPClientManager:
                 async with self._static_connect_lock_for(name):
                     # Close session + transport via the owner close protocol
                     await self._teardown_static_session(name)
-                    # Clean up per-server state (on the event loop thread)
+                    # Clean up per-server state (on the event loop thread).
+                    # The direct stamp pops back 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.
                     self._static_servers.pop(name, None)
                     self._last_error.pop(name, None)
-                    self._last_notification_refresh.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._cb_clear(name)
                     # Clear health-loop backoff/ping state so a later re-add of
                     # the same name doesn't inherit stale ``due`` deadlines.
@@ -5158,7 +5310,9 @@ class MCPClientManager:
             # No event loop (tests / pre-start) — mutate directly
             self._static_servers.pop(name, None)
             self._last_error.pop(name, None)
-            self._last_notification_refresh.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._cb_clear(name)
             self._rebuild_tools()
             self._rebuild_resources()
@@ -5899,7 +6053,7 @@ class MCPClientManager:
         next_ping = time.monotonic() + self._static_health_check_s
         self._static_next_ping[name] = next_ping
         self._spawn_background(
-            self._refresh_server(name),
+            self._refresh_server_logged(name),
             f"catalog refresh after static health reconnect '{name}'",
         )
         log.info("MCP static health: reconnected '%s'", name)
@@ -6058,12 +6212,14 @@ class MCPClientManager:
         # Schedule catalog refresh on the loop without blocking the caller.
         # The reconnected session is valid for the imminent dispatch; catalog
         # drift will be reconciled on the loop in the background.  The task is
-        # tracked: a refresh FAILURE is logged by the done-callback — this
-        # except only covers the scheduling itself.
+        # tracked; a refresh FAILURE is logged inside ``_refresh_server_logged``
+        # (type + message, never the done-callback's ``exc_info`` — the chain
+        # carries the configured bearer) — this except only covers the
+        # scheduling itself.
         def _schedule_refresh() -> None:
             try:
                 self._spawn_background(
-                    self._refresh_server(server_name),
+                    self._refresh_server_logged(server_name),
                     f"catalog refresh after reconnect for '{server_name}'",
                 )
             except Exception:
@@ -7216,7 +7372,8 @@ class MCPClientManager:
         and the pop is a no-op.
         """
         entry.drop_session()
-        self._last_pool_notification_refresh.pop(key, None)
+        for kind in _LIST_CHANGED_KINDS.values():
+            self._last_pool_notification_refresh.pop((key, kind), None)
 
     def _evict_session(self, key: tuple[str, str]) -> None:
         """Drop the cached session on a pool entry; KEEP the catalog.