From 1a8046636943ff17337c39e6f93e20110226b372 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Tue, 14 Jul 2026 09:01:32 -0700 Subject: [PATCH] =?UTF-8?q?fix(mcp):=20review=20round=204=20=E2=80=94=20re?= =?UTF-8?q?moval/reconcile=20lifecycle,=20honest=20skip=20reporting,=20sam?= =?UTF-8?q?e-kind=20debounce=20recovery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 60 +++--- tests/conftest.py | 18 ++ tests/test_mcp_client.py | 221 +++++++++++++++----- tests/test_mcp_live_flaky_server.py | 11 +- tests/test_mcp_live_push_refresh.py | 17 +- turnstone/core/mcp_client.py | 308 ++++++++++++++++++++-------- turnstone/core/session.py | 11 +- 7 files changed, 461 insertions(+), 185 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ff09354..ff1a8053 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py index 3ab9de36..62aa979e 100644 --- a/tests/conftest.py +++ b/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. diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py index cb0f66e0..849e5bab 100644 --- a/tests/test_mcp_client.py +++ b/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 diff --git a/tests/test_mcp_live_flaky_server.py b/tests/test_mcp_live_flaky_server.py index 519c3923..f5d08eab 100644 --- a/tests/test_mcp_live_flaky_server.py +++ b/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) diff --git a/tests/test_mcp_live_push_refresh.py b/tests/test_mcp_live_push_refresh.py index 14eaf69a..97692e24 100644 --- a/tests/test_mcp_live_push_refresh.py +++ b/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( diff --git a/turnstone/core/mcp_client.py b/turnstone/core/mcp_client.py index 1cc2750d..4a583ce1 100644 --- a/turnstone/core/mcp_client.py +++ b/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( diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 477f1e15..add7b520 100644 --- a/turnstone/core/session.py +++ b/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: