fix(mcp): review round 4 — removal/reconcile lifecycle, honest skip reporting, same-kind debounce recovery

reconcile_sync no longer abandons a DB-driven removal that timed out:
both the removal loop and the config-update loop keep the name in
_db_managed (and skip the follow-on add) when remove_server_sync
returns its mutated-nothing False, so the next pass retries instead of
the deleted/reconfigured server serving stale tools until restart.

remove_server_sync is now cancel-safe end to end: it FORCE-drops the
session before queueing (parked push runners bail at their session
gate instead of serializing ≤30s list calls ahead of the removal —
the noisy #839 server was exactly the one whose runners could starve
its own removal), and wraps the post-lock cleanup in try/finally so a
caller-timeout cancel landing mid-teardown still completes the state
pop, catalog rebuild, and lock retirement rather than stranding a
config-gone ghost catalog. Config survives a park-cancel, so the
health loop recovers it.

_refresh_all reports None (not a fake ([], [])) for a busy-skip or
supersede, stamps a 'skipped' status row, and /mcp refresh renders it
distinctly — the operator is no longer told a never-refreshed server
is current. A same-kind push lost to the debounce window (the prior
runner already finished; the server won't re-announce) arms the
health-tick retry, closing the one staleness hole the per-kind
debounce still had; a push covered by a queued runner does not arm
(no lost change). Static resource/prompt catalogs are capped at
connect discovery and every refresh. _list_resource_pair's reap is
bounded so a future SDK cancel-regression can't wedge the lock.

Cleanups: _arm_refresh_retry (retry-arm gate, ×3), _spawn_full_refresh
(discard+spawn, ×3), _popen_mcp_server (live-server spawn, ×2), the
tautological stamp-arithmetic TestNotificationDebounce deleted. Suite
9395 green.

Refs #839

CHANGELOG.md

@@ -169,36 +169,40 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   transport down (which was also the only way the changed catalog ever
   landed). Push refreshes now run as spawned tasks — debounced, coalesced per
   (server, kind), bounded by the connect timeout, and serialized on the
-  per-server connect lock — and the manual/periodic refresh publishes under
-  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 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
+  per-server connect lock — and the manual and post-reconnect refreshes
+  publish under 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. A change genuinely lost to the
+  debounce window (a same-kind push landing after the prior refresh finished,
+  which the server will never re-announce) is recovered by an automatic
+  health-tick retry rather than staying invisible until an unrelated push or
+  a reconnect. 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 and reaped, under a bounded grace, inside the scope. A
+  push refresh that fails while the connection stays up is likewise retried 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 on both the
-  connected and disconnected branches, the skip arms the automatic retry so
-  the operator's request isn't silently dropped, and a force-reconnect
-  drops the session up front so queued push refreshes can't starve it.
-  Static-path resource and prompt catalogs are now size-capped like the
-  pool path's (and like static tools) at discovery and on every refresh,
-  so a misbehaving server's push can't balloon the node's merged catalogs.
-  Deleting a server can no longer leave it half-removed: a removal that
-  times out behind a busy connect lock now mutates nothing and is cleanly
-  retryable (previously the config was popped up front, stranding a live
-  session and its published catalog with no driver able to reach them).
+  until an operator intervened. 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 on both the connected and
+  disconnected branches, reported distinctly as "skipped" rather than as a
+  false "no changes", the skip arms the automatic retry, and a
+  force-reconnect drops the session up front so queued push refreshes can't
+  starve it. Static-path resource and prompt catalogs are now size-capped
+  like the pool path's (and like static tools) at discovery and on every
+  refresh, so a misbehaving server's push can't balloon the node's merged
+  catalogs. Deleting or reconfiguring a server can no longer leave it
+  half-removed: the config removal and all cleanup are serialized under the
+  connect lock (a cancelled removal completes its cleanup rather than
+  stranding a live session and published catalog with the config already
+  gone), and `reconcile_sync` retries a removal that timed out instead of
+  marking it done — previously a DB-driven delete of a busy server could be a
+  silent, permanent no-op until process restart.
 
 - **OpenAI Responses streaming: truncated and refused responses no longer
   vanish.** A response that hit `max_output_tokens` terminates the stream

tests/conftest.py

@@ -5,6 +5,8 @@ import contextlib
 import logging
 import os
 import socket
+import subprocess
+import sys
 import threading
 import time
 from typing import TYPE_CHECKING, Any
@@ -293,6 +295,22 @@ def _wait_session_live(mgr: MCPClientManager, name: str, timeout: float) -> bool
     return _poll_until(_live, timeout)
 
 
+def _popen_mcp_server(script_path: Any, port: int) -> subprocess.Popen[bytes]:
+    """Start a FastMCP live-server subprocess, streams to DEVNULL.
+
+    The shared spawn primitive for the live MCP smoke tests
+    (flaky-server flap loop, push-refresh) — the readiness wait and the
+    skip-vs-raise-on-failure policy legitimately differ per test and
+    stay at the call sites. ``sys.executable`` runs the same interpreter,
+    so a server-side import gap surfaces as a failed TCP wait, not here.
+    """
+    return subprocess.Popen(
+        [sys.executable, str(script_path), str(port)],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
 def make_oidc_test_config(**overrides: Any) -> OIDCConfig:
     """Build a test ``OIDCConfig`` with sensible defaults.
 

tests/test_mcp_client.py

@@ -7,6 +7,7 @@ import concurrent.futures
 import contextlib
 import inspect
 import json
+import threading
 import time
 from contextlib import AsyncExitStack, suppress
 from typing import Any
@@ -2847,44 +2848,14 @@ class TestSafeTransportStreams:
 
 
 # ---------------------------------------------------------------------------
-# Fix 4: Notification debounce
+# Notification debounce is covered BEHAVIORALLY (through the real handler) by
+# TestStaticNotificationRefresh — per-kind independence, coalescing, and the
+# debounce-drop retry arm. The old stamp-arithmetic TestNotificationDebounce
+# was deleted: it seeded a dict and asserted time math on the same dict,
+# exercising no product code path.
 # ---------------------------------------------------------------------------
 
 
-class TestNotificationDebounce:
-    """Verify notification-triggered refreshes are debounced per (server, kind)."""
-
-    def test_debounce_within_window(self):
-        mgr = MCPClientManager({})
-        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", "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", "tools")] = time.monotonic() - 10
-        now = time.monotonic()
-        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", "tools")] = time.monotonic()
-        # srv_b has no timestamp — should pass debounce
-        now = time.monotonic()
-        last_b = mgr._last_notification_refresh.get(("srv_b", "tools"), 0.0)
-        assert now - last_b >= 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.
-
-
 # ---------------------------------------------------------------------------
 # Static-path list_changed refresh — spawned runner protocol (#839)
 # ---------------------------------------------------------------------------
@@ -3387,6 +3358,73 @@ class TestStaticNotificationRefresh:
             "a same-window different-kind notification must spawn its own refresh"
         )
 
+    def test_same_kind_debounce_drop_arms_retry(self, running_loop_mgr) -> None:
+        """A SAME-kind push landing in the debounce window AFTER the
+        previous runner completed (no runner queued) is genuinely lost to
+        the window — MCP servers announce a change once. It must arm the
+        health-tick retry so the change still converges; otherwise the
+        catalog stays stale for every user until an unrelated push or a
+        reconnect."""
+        mgr, loop, _thread = running_loop_mgr
+
+        async def _rec(_name: str) -> tuple[list[str], list[str]]:
+            return [], []
+
+        mgr._refresh_server_tools = _rec  # type: ignore[method-assign]
+        mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"}
+        handler = mgr._make_static_notification_handler("srv")
+
+        async def _fire_then_redrive() -> int:
+            mgr._static_connect_lock_for("srv")
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            note = mcp_types.ServerNotification(
+                mcp_types.ToolListChangedNotification(method="notifications/tools/list_changed")
+            )
+            await handler(note)  # admitted, runner spawns
+            return len(mgr._background_tasks)
+
+        _run_on_loop(loop, _fire_then_redrive())
+        _drain_background(mgr, loop)  # first runner completes, clears marker
+        # Second same-kind push, still inside the 5s window, NO runner queued.
+        mgr._static_refresh_retry.discard("srv")
+
+        async def _fire_again() -> int:
+            note = mcp_types.ServerNotification(
+                mcp_types.ToolListChangedNotification(method="notifications/tools/list_changed")
+            )
+            before = len(mgr._background_tasks)
+            await handler(note)
+            return len(mgr._background_tasks) - before
+
+        spawned = _run_on_loop(loop, _fire_again())
+        assert spawned == 0, "debounce must still suppress the spawn (throttle intact)"
+        assert "srv" in mgr._static_refresh_retry, "but the lost change must arm the retry"
+
+    def test_debounce_drop_behind_running_runner_does_not_arm(self, running_loop_mgr) -> None:
+        """A same-kind push arriving while a runner is STILL queued (marker
+        present) is already covered — that runner spawns a successor — so
+        the debounce drop must NOT arm the retry (no lost change to
+        recover). Guards against the arm firing on every chatty push."""
+        mgr, loop, _thread = running_loop_mgr
+        mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"}
+        handler = mgr._make_static_notification_handler("srv")
+
+        async def _fire_with_marker_present() -> None:
+            mgr._static_connect_lock_for("srv")
+            _seed_static_state(mgr, "srv", session=MagicMock())
+            # Stamp fresh (inside window) AND a runner marker present.
+            mgr._last_notification_refresh[("srv", "tools")] = time.monotonic()
+            mgr._static_refresh_pending.add(("srv", "tools"))
+            note = mcp_types.ServerNotification(
+                mcp_types.ToolListChangedNotification(method="notifications/tools/list_changed")
+            )
+            await handler(note)
+
+        _run_on_loop(loop, _fire_with_marker_present())
+        assert "srv" not in mgr._static_refresh_retry, (
+            "a push covered by a queued runner must not arm a redundant retry"
+        )
+
     def test_refresh_server_superseded_by_remove_returns_none(self, running_loop_mgr) -> None:
         """A ``_refresh_server`` pass whose server was removed before it
         ran must publish nothing and write NO status: the removal
@@ -3540,14 +3578,15 @@ class TestStaticNotificationRefresh:
         assert "srv" not in mgr._static_refresh_retry
 
     def test_refresh_all_arms_retry_on_busy_skip(self, running_loop_mgr) -> None:
-        """An operator-requested pass that busy-skips a server must arm
-        the health-tick retry: the lock holder may be a single-kind push
-        runner, not the full pass the operator asked for, and without
-        the arm the request is silently dropped — output
-        indistinguishable from 'refreshed, no changes'."""
+        """An operator-requested pass that busy-skips a server must report
+        None (distinct from a real ``([], [])`` no-change) AND arm the
+        health-tick retry: the lock holder may be a single-kind push
+        runner, not the full pass the operator asked for. A ``None`` +
+        ``skipped`` status is what keeps the operator from being told a
+        stale server is current."""
         mgr, loop, _thread = running_loop_mgr
 
-        async def _scenario() -> dict[str, tuple[list[str], list[str]]]:
+        async def _scenario() -> dict[str, tuple[list[str], list[str]] | None]:
             lock = mgr._static_connect_lock_for("srv")
             _seed_static_state(mgr, "srv", session=MagicMock())
             await lock.acquire()
@@ -3557,8 +3596,9 @@ class TestStaticNotificationRefresh:
                 lock.release()
 
         results = _run_on_loop(loop, _scenario())
-        assert results == {"srv": ([], [])}
+        assert results == {"srv": None}, "a busy-skip must be None, not a fake no-change"
         assert "srv" in mgr._static_refresh_retry
+        assert mgr._last_refresh["srv"][1] == "skipped"
 
     def test_refresh_all_skips_disconnected_server_with_reconnect_in_flight(
         self, running_loop_mgr
@@ -3567,7 +3607,8 @@ class TestStaticNotificationRefresh:
         attempt holds the per-name lock for up to 45s, and parking there
         burned the whole 30s ``refresh_sync`` budget on one server,
         starving every healthy server behind it. The in-flight reconnect
-        finishes the job (full rediscovery)."""
+        finishes the job (full rediscovery); the pass reports the skip
+        as None with a ``skipped`` pill."""
         mgr, loop, _thread = running_loop_mgr
         ensure_calls: list[str] = []
 
@@ -3576,8 +3617,9 @@ class TestStaticNotificationRefresh:
             return MagicMock()
 
         mgr._ensure_static_connected = _ensure  # type: ignore[method-assign]
+        mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"}
 
-        async def _scenario() -> dict[str, tuple[list[str], list[str]]]:
+        async def _scenario() -> dict[str, tuple[list[str], list[str]] | None]:
             lock = mgr._static_connect_lock_for("srv")
             _seed_static_state(mgr, "srv", session=None)  # disconnected
             await lock.acquire()  # a reconnect driver holds the lock
@@ -3587,16 +3629,18 @@ class TestStaticNotificationRefresh:
                 lock.release()
 
         results = _run_on_loop(loop, _scenario())
-        assert results == {"srv": ([], [])}
+        assert results == {"srv": None}
         assert ensure_calls == [], "the pass must not park inside _ensure_static_connected"
+        assert mgr._last_refresh["srv"][1] == "skipped"
 
-    def test_remove_server_timeout_leaves_no_half_state(self, running_loop_mgr) -> None:
+    def test_remove_server_timeout_leaves_retryable_state(self, running_loop_mgr) -> None:
         """A removal cancelled while PARKED behind a lock holder must
-        mutate NOTHING — pre-fix it popped the config up front, so a
-        timeout left a half-removed server (config gone, session and
-        catalogs still published, no driver able to reconnect or cleanly
-        re-remove). Every mutation now sits under the lock, making a
-        timed-out removal honestly retryable."""
+        leave RETRYABLE state — config still present (pre-fix it popped
+        the config up front, so a timeout stranded a live session and its
+        published catalog with no config behind them). The FORCE session
+        pre-drop is recoverable precisely because the config survives: the
+        health loop reconnects it. The config pop and all cleanup now sit
+        under the lock, so a re-remove works."""
         mgr, loop, _thread = running_loop_mgr
         mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"}
 
@@ -3607,17 +3651,86 @@ class TestStaticNotificationRefresh:
 
         _run_on_loop(loop, _hold())
         assert mgr.remove_server_sync("srv", timeout=0.3) is False
-        # Nothing was mutated: the removal is retryable.
+        # Config survives (the pop is under the lock the timeout never
+        # reached), so the health loop can reconnect and a retry works —
+        # no config-gone ghost. The session pre-drop is the one recoverable
+        # mutation.
         assert "srv" in mgr._server_configs
-        assert "srv" in mgr._static_servers
 
         async def _release() -> None:
             mgr._static_connect_locks["srv"].release()
 
         _run_on_loop(loop, _release())
-        assert mgr.remove_server_sync("srv", timeout=5) is True
+        # The retry completes the removal. Its bool return is ``was_connected``
+        # (connected AND removed) — False here because the first attempt's
+        # FORCE pre-drop already disconnected the session; the EFFECT is what
+        # matters, and every trace of the server is now gone.
+        mgr.remove_server_sync("srv", timeout=5)
         assert "srv" not in mgr._server_configs
         assert "srv" not in mgr._static_servers
+        assert "srv" not in mgr._static_connect_locks
+
+    def test_remove_cancel_mid_teardown_completes_cleanup(self, running_loop_mgr) -> None:
+        """If the caller timeout cancels _remove AFTER the config pop,
+        while it is awaiting inside _teardown_static_session, the SYNC
+        cleanup in the finally must still run — else config-gone +
+        published ghost catalogs strand with no driver to reach them.
+        We drive the exact interleave: teardown blocks, the caller times
+        out and cancels mid-await, and we assert full cleanup ran."""
+        mgr, loop, _thread = running_loop_mgr
+        mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"}
+        _seed_static_state(
+            mgr, "srv", session=MagicMock(), tools=[_fake_openai_tool("mcp__srv__t")]
+        )
+        mgr._rebuild_tools()
+
+        release = threading.Event()
+
+        async def _blocking_teardown(_name: str) -> None:
+            # Simulate a teardown that blocks past the caller budget, then
+            # gets cancelled at this await.
+            await asyncio.get_running_loop().run_in_executor(None, release.wait)
+
+        mgr._teardown_static_session = _blocking_teardown  # type: ignore[method-assign]
+
+        # Short caller timeout: cancels _remove while it's parked in the
+        # blocking teardown (config already popped).
+        result = mgr.remove_server_sync("srv", timeout=0.4)
+        release.set()  # let the executor wait return so the loop drains
+        assert result is False
+        # The finally completed the removal despite the mid-teardown cancel:
+        # no ghost catalog, no orphaned state.
+        deadline = time.time() + 5
+        while "srv" in mgr._static_servers and time.time() < deadline:
+            time.sleep(0.02)
+        assert "srv" not in mgr._static_servers, "finally must complete cleanup on cancel"
+        assert "srv" not in mgr._server_configs
+        assert "mcp__srv__t" not in mgr._tool_map, "ghost tool must not survive"
+
+    def test_reconcile_keeps_managed_when_removal_times_out(self, running_loop_mgr) -> None:
+        """reconcile_sync must NOT discard a name from _db_managed when
+        remove_server_sync times out (returns False having mutated
+        nothing): discarding it made a DB-driven removal a permanent
+        no-op — the removal loop iterates ``_db_managed - desired``, so a
+        discarded name can never be retried while the health loop keeps
+        the server alive off ``_server_configs``."""
+        mgr, _loop, _thread = running_loop_mgr
+        mgr._db_managed = {"gone"}
+        mgr._server_configs["gone"] = {"type": "http", "url": "http://x/mcp"}
+
+        # remove_server_sync reports timeout (False) WITHOUT clearing config
+        # — the mutated-nothing path.
+        def _timeout_remove(name: str, timeout: float = 30) -> bool:
+            return False
+
+        mgr.remove_server_sync = _timeout_remove  # type: ignore[method-assign]
+        # DB now has NO servers, so 'gone' is in the removal set.
+        storage = MagicMock()
+        storage.list_mcp_servers.return_value = []
+        mgr.reconcile_sync(storage)
+        # Name RETAINED for the next reconcile to retry (config still present,
+        # so the False means "timed out", not "removed").
+        assert "gone" in mgr._db_managed, "a timed-out removal must be retried, not abandoned"
 
     def test_reconnect_sync_drops_session_before_queueing(self, running_loop_mgr) -> None:
         """Force-reconnect drops the session up front so parked

tests/test_mcp_live_flaky_server.py

@@ -19,8 +19,6 @@ from __future__ import annotations
 import asyncio
 import gc
 import signal
-import subprocess
-import sys
 import textwrap
 import time
 from typing import TYPE_CHECKING
@@ -28,10 +26,11 @@ from unittest.mock import patch
 
 import pytest
 
-from tests.conftest import _free_port, _wait_session_live, _wait_tcp_ready
+from tests.conftest import _free_port, _popen_mcp_server, _wait_session_live, _wait_tcp_ready
 from turnstone.core.mcp_client import MCPClientManager
 
 if TYPE_CHECKING:
+    import subprocess
     from pathlib import Path
 
 SERVER_SRC = textwrap.dedent(
@@ -108,11 +107,7 @@ class TestFlakyServerNoSpin:
         monkeypatch.setattr(MCPClientManager, "_STATIC_HEALTH_PING_TIMEOUT_S", 1.5)
 
         def _spawn_server(*, initial: bool = False) -> subprocess.Popen[bytes]:
-            proc = subprocess.Popen(
-                [sys.executable, str(script), str(port)],
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-            )
+            proc = _popen_mcp_server(script, port)
             if not _wait_tcp_ready(port, 10.0):
                 proc.kill()
                 proc.wait(timeout=5)

tests/test_mcp_live_push_refresh.py

@@ -30,18 +30,23 @@ Self-contained (spawns its own server; no LLM backend, no network beyond
 
 from __future__ import annotations
 
-import subprocess
-import sys
 import textwrap
 from typing import TYPE_CHECKING
 from unittest.mock import patch
 
 import pytest
 
-from tests.conftest import _free_port, _poll_until, _wait_session_live, _wait_tcp_ready
+from tests.conftest import (
+    _free_port,
+    _poll_until,
+    _popen_mcp_server,
+    _wait_session_live,
+    _wait_tcp_ready,
+)
 from turnstone.core.mcp_client import MCPClientManager
 
 if TYPE_CHECKING:
+    import subprocess
     from pathlib import Path
 
 SERVER_SRC = textwrap.dedent(
@@ -111,11 +116,7 @@ class TestPushRefreshNoDeadlock:
         proc: subprocess.Popen[bytes] | None = None
         mgr: MCPClientManager | None = None
         try:
-            proc = subprocess.Popen(
-                [sys.executable, str(script), str(port)],
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-            )
+            proc = _popen_mcp_server(script, port)
             if not _wait_tcp_ready(port, 10.0):
                 pytest.skip("push-refresh server subprocess did not come up")
             with patch(

turnstone/core/mcp_client.py

@@ -1595,6 +1595,7 @@ class MCPClientManager:
         origin: str,
         label_prefix: str,
         label_target: str,
+        on_debounce_drop: Callable[[], None] | None = None,
     ) -> None:
         """Admission half of the ``*/list_changed`` protocol — ONE copy.
 
@@ -1620,6 +1621,20 @@ class MCPClientManager:
         notifying-but-slow server from accreting waiters that starve the
         lock's other users (reconnect drivers, dispatches, eviction).
 
+        *on_debounce_drop* recovers the ONE change the debounce genuinely
+        loses: a same-kind push landing in the window AFTER the previous
+        runner already completed (its marker cleared at lock-acquire, its
+        list finished) has no queued runner to observe it and, since MCP
+        servers announce a change once, would leave the catalog stale
+        indefinitely. Static passes an arm of the health-tick retry
+        (idempotent — a set — so a chatty server yields at most one retry
+        per tick, honouring the debounce's throttle while guaranteeing
+        eventual convergence). Pool passes ``None``: it has no health
+        tick, and its residual (a same-kind change debounced away
+        converges only on the entry's next reconnect or a later
+        cross-kind push) is the pre-existing #840 behaviour, unchanged
+        here.
+
         The refresh is SPAWNED, never awaited: the SDK awaits message
         handlers inline in its receive loop, so an in-handler request on
         the same session can never receive its response — only the
@@ -1640,6 +1655,12 @@ class MCPClientManager:
                 origin,
                 now - last,
             )
+            # A push queued behind a still-running runner (marker present)
+            # is already covered — that runner spawns a successor. Only a
+            # push with NO runner in flight is genuinely lost to the
+            # window; hand it to the recovery hook.
+            if on_debounce_drop is not None and marker not in pending:
+                on_debounce_drop()
             return
         if marker in pending:
             log.debug("Coalescing %s notification from %s (refresh queued)", kind, origin)
@@ -1818,6 +1839,7 @@ class MCPClientManager:
                 origin=f"'{name}'",
                 label_prefix="static",
                 label_target=name,
+                on_debounce_drop=lambda: self._arm_refresh_retry(name),
             )
 
         return _on_notification
@@ -4053,6 +4075,16 @@ class MCPClientManager:
         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``.
+
+        The reap is BOUNDED by its own grace deadline, NOT left inside the
+        outer ``asyncio.timeout`` (which fires exactly once — after it has
+        fired, awaiting the cancelled children there is unbounded). The
+        pinned SDK (mcp>=1.27,<2) honours cancellation promptly, so this
+        is belt-and-braces against a future regression: a child that
+        ignores ``cancel()`` must not wedge this runner while it holds the
+        per-server connect lock — the #839 deadlock class. A child still
+        pending after the grace is logged and abandoned (a cancelled local
+        task, GC-reaped) rather than held onto.
         """
         async with asyncio.timeout(self._CONNECT_TIMEOUT):
             res_task = asyncio.create_task(session.list_resources())
@@ -4062,14 +4094,40 @@ class MCPClientManager:
             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.
+                # and REAP within a bounded grace 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)
+                await self._reap_bounded((res_task, tmpl_task))
                 raise
             return res_result, tmpl_result
 
+    async def _reap_bounded(self, tasks: tuple[asyncio.Task[Any], ...]) -> None:
+        """Await already-cancelled *tasks* under a grace deadline.
+
+        Uses ``asyncio.shield`` so THIS await is immune to the caller's
+        own in-flight cancellation (an expired ``asyncio.timeout`` scope
+        has already delivered its single cancel), and ``asyncio.wait``
+        with a timeout so a child that refuses to honour ``cancel()``
+        cannot block indefinitely. Exceptions on done tasks are retrieved
+        (marking them handled); any straggler is logged and abandoned.
+        """
+        try:
+            done, pending = await asyncio.shield(
+                asyncio.wait(set(tasks), timeout=self._OWNER_CANCEL_GRACE_S)
+            )
+        except asyncio.CancelledError:
+            done, pending = {t for t in tasks if t.done()}, {t for t in tasks if not t.done()}
+        for task in done:
+            with contextlib.suppress(BaseException):
+                task.exception()
+        if pending:
+            log.warning(
+                "MCP resource-list reap: %d task(s) ignored cancellation; abandoning",
+                len(pending),
+            )
+
     async def _refresh_pool_server_resources(
         self, key: tuple[str, str]
     ) -> tuple[list[str], list[str]]:
@@ -4232,9 +4290,30 @@ class MCPClientManager:
             exc,
         )
         self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}")
+        self._arm_refresh_retry(name)
+
+    def _arm_refresh_retry(self, name: str) -> None:
+        """Arm the health-tick refresh retry for *name* — the ONE gate copy.
+
+        Config-gated: a failure/skip/drop observed for a just-removed
+        server must not park a retry flag nothing will ever drain (the
+        tick iterates ``_server_configs`` only).
+        """
         if name in self._server_configs:
             self._static_refresh_retry.add(name)
 
+    def _spawn_full_refresh(self, name: str, label: str) -> None:
+        """Spawn a full logged catalog refresh; consume any pending retry.
+
+        The ONE copy of the discard-then-spawn idiom shared by the three
+        drivers that run a full pass (health-tick retry drain,
+        post-health-reconnect, post-dispatch-reconnect): the spawned pass
+        IS the refresh a pending health-tick retry wants, so clearing the
+        flag first prevents the next tick running a redundant second pass.
+        """
+        self._static_refresh_retry.discard(name)
+        self._spawn_background(self._refresh_server_logged(name), label)
+
     async def _refresh_server(self, name: str) -> tuple[list[str], list[str]] | None:
         """Re-fetch tools, resources, and prompts for one server.
 
@@ -4392,18 +4471,25 @@ class MCPClientManager:
         except (Exception, BaseExceptionGroup) as 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)
+            if refreshed is None:
+                self._arm_refresh_retry(name)
 
     async def _refresh_all(
         self, server_name: str | None = None
-    ) -> dict[str, tuple[list[str], list[str]]]:
+    ) -> dict[str, tuple[list[str], list[str]] | None]:
         """Refresh tools, resources, and prompts for one or all servers.
 
         For disconnected servers (in config but not connected), attempts
-        reconnect.  Returns ``{server: (added, removed)}`` per server.
+        reconnect.  Returns ``{server: (added, removed)}`` per server —
+        or ``None`` for a server whose refresh was SKIPPED (another
+        operation held its lock, or it was removed/evicted mid-pass).
+        ``None`` is a distinct shape on purpose: rendering a skip as
+        ``([], [])`` told the operator "refreshed, no changes" for a
+        server that was never refreshed at all. Skips on live servers
+        also stamp a ``skipped`` ``_last_refresh`` row so the admin pill
+        tells the same story.
         """
-        results: dict[str, tuple[list[str], list[str]]] = {}
+        results: dict[str, tuple[list[str], list[str]] | None] = {}
         targets = [server_name] if server_name else list(self._server_configs.keys())
 
         for name in targets:
@@ -4417,12 +4503,16 @@ class MCPClientManager:
                         # starve every healthy server behind it (the same
                         # reason ``_refresh_server`` busy-skips). The
                         # holder finishes the job: a reconnect ends in
-                        # full rediscovery.
+                        # full rediscovery. No retry arm: the flag drains
+                        # only on live sessions, and a successful reconnect
+                        # already spawns its own full refresh.
                         log.info(
                             "Refresh pass for '%s' skipped: reconnect in flight",
                             name,
                         )
-                        results[name] = ([], [])
+                        if name in self._server_configs:
+                            self._last_refresh[name] = (time.time(), "skipped")
+                        results[name] = None
                         continue
                     # Attempt reconnect via the shared lazy primitive — it owns
                     # the per-name lock, live-session reuse, the in_flight
@@ -4437,8 +4527,9 @@ class MCPClientManager:
                             # Deliberate skip: removed concurrently, or a
                             # sibling call is still in flight on the old
                             # stack. Not a failure — the next refresh or
-                            # health tick retries.
-                            results[name] = ([], [])
+                            # health tick retries; report it as the skip
+                            # it is.
+                            results[name] = None
                             continue
                         post = self._static_servers.get(name)
                         new_names = (
@@ -4452,15 +4543,17 @@ class MCPClientManager:
                     # Skipped (lock busy) or superseded (removed /
                     # evicted) — a deliberate non-outcome: record neither
                     # success nor failure for whatever generation lives
-                    # now. BUT the operator asked for a refresh and a
-                    # busy-skip's lock holder may be a single-kind push
-                    # runner, not the full pass — arm the health-tick
-                    # retry so the request isn't silently dropped
-                    # (mirrors ``_refresh_server_logged``; config-gated,
-                    # so removed servers arm nothing).
+                    # now, and report None so the operator sees "skipped",
+                    # not a fake "no changes". BUT the operator asked for
+                    # a refresh and a busy-skip's lock holder may be a
+                    # single-kind push runner, not the full pass — arm the
+                    # health-tick retry so the request isn't silently
+                    # dropped (mirrors ``_refresh_server_logged``;
+                    # config-gated, so removed servers arm nothing).
                     if name in self._server_configs:
-                        self._static_refresh_retry.add(name)
-                    results[name] = ([], [])
+                        self._arm_refresh_retry(name)
+                        self._last_refresh[name] = (time.time(), "skipped")
+                    results[name] = None
                     continue
                 added, removed = refreshed
                 self._cb_record_success(name)
@@ -4509,10 +4602,16 @@ class MCPClientManager:
 
     def refresh_sync(
         self, server_name: str | None = None, timeout: int = 30
-    ) -> dict[str, tuple[list[str], list[str]]]:
+    ) -> dict[str, tuple[list[str], list[str]] | None]:
         """Refresh tools synchronously (blocks the calling thread).
 
-        Returns ``{server: (added_names, removed_names)}`` per server.
+        Returns ``{server: (added_names, removed_names)}`` per server, or
+        ``{server: None}`` when that server's refresh was SKIPPED (its
+        lock was held by a concurrent reconnect/refresh, or it was
+        removed mid-pass). Callers MUST distinguish ``None`` from
+        ``([], [])``: the latter is a refresh that ran and found no
+        changes; ``None`` is a refresh that never ran, deferred to the
+        health-tick retry.
         """
         assert self._loop is not None
         future = asyncio.run_coroutine_threadsafe(self._refresh_all(server_name), self._loop)
@@ -5390,59 +5489,90 @@ class MCPClientManager:
         if self._loop is not None:
 
             async def _remove() -> None:
+                # FORCE-drop the session before queueing (mirrors
+                # ``reconnect_sync``): parked push-refresh runners bail at
+                # their session gate instead of serializing up to one ≤30s
+                # list call per kind ahead of this removal — the noisy slow
+                # server most in need of removal was exactly the one whose
+                # runners could starve it past every caller budget.
+                # Timeout-safe: the config is still present at this point,
+                # so if the removal is then cancelled while parked, the
+                # health loop simply reconnects — recoverable, never
+                # half-removed.
+                pre = self._static_servers.get(name)
+                if pre is not None:
+                    self._drop_static_session_and_stamp(name, pre)
                 # Hold the per-name connect lock across the WHOLE removal —
                 # including the config pop. Popping the config before
                 # queueing (the old shape) meant a timed-out ``_remove``
-                # (cancelled while parked behind a slow reconnect or the
-                # push-refresh runners this lock now serializes) left a
-                # HALF-REMOVED server: config gone, but session + published
-                # catalogs alive with no driver able to reconnect or
-                # re-remove them. With every mutation under the lock, a
-                # removal cancelled while parked has touched NOTHING — the
-                # caller's False is honest and a retry works. The cost: a
-                # reconnect driver that wins the lock first can rebuild the
-                # session moments before removal — this teardown then
-                # closes it anyway.
+                # (cancelled while parked) left a HALF-REMOVED server:
+                # config gone, but session + published catalogs alive with
+                # no driver able to reconnect or re-remove them. With every
+                # mutation under the lock, a removal cancelled while parked
+                # has touched NOTHING (beyond the recoverable session drop
+                # above) — the caller's False is honest and a retry works.
+                # The cost: a reconnect driver that wins the lock first can
+                # rebuild the session moments before removal — this teardown
+                # then closes it anyway.
                 async with self._static_connect_lock_for(name):
                     self._server_configs.pop(name, None)
-                    # 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 push-state clear backs up the teardown call above:
-                    # ``_teardown_static_session`` early-returns (no pop) when
-                    # 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)
-                    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.
-                    self._static_reconnect_attempt.pop(name, None)
-                    self._static_reconnect_next.pop(name, None)
-                    self._static_next_ping.pop(name, None)
-                    # Rebuild merged state (serialized with notification handlers)
-                    self._rebuild_tools()
-                    self._rebuild_resources()
-                    self._rebuild_prompts()
-                # Drop the now-orphaned per-name lock AFTER releasing it (the
-                # server is gone; a re-add re-creates it lazily).
-                self._static_connect_locks.pop(name, None)
+                    try:
+                        # Close session + transport via the owner close protocol
+                        await self._teardown_static_session(name)
+                    finally:
+                        # SYNC cleanup — all of it, no awaits: it must
+                        # complete even when the caller's timeout cancels us
+                        # inside the teardown awaits above, or a cancel
+                        # landing after the config pop would strand
+                        # published ghost catalogs with no config behind
+                        # them (the abandoned owner unwinds solo — the
+                        # close protocol tolerates that). A removal
+                        # cancelled HERE therefore still completes; the
+                        # caller's False is then stale, but a retry no-ops
+                        # and ``reconcile_sync``'s config post-condition
+                        # sees the truth.
+                        #
+                        # The push-state clear backs up the teardown call:
+                        # ``_teardown_static_session`` early-returns (no
+                        # pop) when 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)
+                        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.
+                        self._static_reconnect_attempt.pop(name, None)
+                        self._static_reconnect_next.pop(name, None)
+                        self._static_next_ping.pop(name, None)
+                        # Rebuild merged state (serialized with notification
+                        # handlers on the loop thread).
+                        self._rebuild_tools()
+                        self._rebuild_resources()
+                        self._rebuild_prompts()
+                        # Retire the lock (a re-add re-creates it lazily);
+                        # parked waiters acquire the retired object and bail
+                        # at their identity checks.
+                        self._static_connect_locks.pop(name, None)
 
             future = asyncio.run_coroutine_threadsafe(_remove(), self._loop)
             try:
                 future.result(timeout=timeout)
             except concurrent.futures.TimeoutError:
-                # A slow reconnect (or the push-refresh runners that now
-                # share this lock) held it past our wait. CANCEL the pending
-                # ``_remove`` so it can't later pop a re-added entry (or its
-                # new lock) and corrupt state; report failure rather than a
-                # false "removed". Cancelled-while-parked has mutated
-                # NOTHING (every mutation, including the config pop, sits
-                # under the lock), so the removal is cleanly retryable.
+                # A slow reconnect held the lock past our wait (the session
+                # pre-drop already drained parked push runners). CANCEL the
+                # pending ``_remove`` so it can't later pop a re-added entry
+                # (or its new lock) and corrupt state; report failure rather
+                # than a false "removed". Cancelled while PARKED → nothing
+                # mutated (retry works); cancelled inside the teardown →
+                # the ``finally`` completes the removal anyway (a retry
+                # no-ops, and reconcile's config post-condition sees the
+                # truth either way).
                 future.cancel()
                 log.warning("MCP server '%s' removal timed out; cancelled (retryable)", name)
                 return False
@@ -5665,6 +5795,16 @@ class MCPClientManager:
         # Config-file servers (not in _db_managed) are left untouched.
         for name in list(self._db_managed - desired_names):
             self.remove_server_sync(name, timeout=timeout)
+            if name in self._server_configs:
+                # Removal timed out while parked (it mutates nothing in that
+                # case) — KEEP the name in ``_db_managed`` so the next
+                # reconcile pass retries. Discarding here made a DB-driven
+                # removal a silent, permanent no-op: the removal loop
+                # iterates ``_db_managed - desired``, so a discarded name
+                # can never be removed again while the health loop keeps
+                # the server alive off ``_server_configs``.
+                log.warning("reconcile_sync: removal of '%s' timed out; retrying next pass", name)
+                continue
             self._db_managed.discard(name)
             removed.append(name)
 
@@ -5685,6 +5825,19 @@ class MCPClientManager:
             if desired[name] != self._server_configs.get(name):
                 log.info("Config changed for MCP server '%s', reconnecting", name)
                 self.remove_server_sync(name, timeout=timeout)
+                if name in self._server_configs:
+                    # Removal timed out (nothing mutated) — defer the update
+                    # rather than layering the new config via add: a
+                    # subsequently timed-out add pops the config outright,
+                    # stranding the still-live OLD session as exactly the
+                    # half-removed ghost the locked removal exists to
+                    # prevent. The DB row is unchanged, so the next pass
+                    # sees the same drift and retries.
+                    log.warning(
+                        "reconcile_sync: config update for '%s' deferred (removal timed out)",
+                        name,
+                    )
+                    continue
                 result = self.add_server_sync(name, desired[name], timeout=timeout)
                 if result.get("connected"):
                     updated.append(name)
@@ -6128,11 +6281,7 @@ class MCPClientManager:
         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}'",
-                )
+                self._spawn_full_refresh(name, f"catalog refresh retry for '{name}'")
             return await self._static_ping_one(name, now)
         return await self._static_reconnect_one(name)
 
@@ -6210,15 +6359,7 @@ 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}'",
-        )
+        self._spawn_full_refresh(name, f"catalog refresh after static health reconnect '{name}'")
         log.info("MCP static health: reconnected '%s'", name)
         return next_ping
 
@@ -6381,13 +6522,8 @@ 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}'",
+                self._spawn_full_refresh(
+                    server_name, f"catalog refresh after reconnect for '{server_name}'"
                 )
             except Exception:
                 log.warning(

turnstone/core/session.py

@@ -3057,7 +3057,16 @@ class ChatSession:
             return
 
         lines: list[str] = []
-        for srv, (added, removed) in sorted(results.items()):
+        for srv, diff in sorted(results.items()):
+            if diff is None:
+                # Skipped, NOT refreshed: another operation held the
+                # server's lock (a reconnect / a push refresh), or it was
+                # removed mid-pass. Reported distinctly from "no changes"
+                # so the operator isn't told a stale server is current;
+                # the health-tick retry runs the real refresh shortly.
+                lines.append(f"  {srv}: {dim('skipped (busy — retry scheduled)')}")
+                continue
+            added, removed = diff
             if added or removed:
                 summary: list[str] = []
                 if added: