diff --git a/CHANGELOG.md b/CHANGELOG.md index 84f874a6..187249ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -173,6 +173,14 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. that same lock, so a slower publisher can no longer land a staler catalog over a fresher one. Every teardown path now also clears the notification debounce stamp, so a reconnected server's first push refreshes immediately. + Push-refresh debouncing is now per (server, kind) on BOTH the static and + per-user pool paths — a tools push no longer swallows a prompts push + arriving in the same 5-second window (previously the second push was + dropped outright, and the change stayed invisible until the server pushed + that kind again). The resource-refresh fan-out on both paths also no + longer orphans its sibling list call when one of the pair fails fast — + both calls now complete inside the timeout scope before the failure is + re-raised. - **OpenAI Responses streaming: truncated and refused responses no longer vanish.** A response that hit `max_output_tokens` terminates the stream diff --git a/tests/conftest.py b/tests/conftest.py index d4f9a04a..f7ff1975 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -216,6 +216,33 @@ def _seed_static_state(mgr: MCPClientManager, name: str, **overrides: Any) -> St return state +def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any, timeout: float = 10) -> Any: + """Submit *coro* to *loop*, wait for the result. + + The ONE copy shared by the MCP test files — four hand-synced copies + had already drifted on the timeout (5s hardcoded vs a 10s default). + The timeout is an upper bound on waiting, not a behavior assertion, + so the most generous variant won the merge. + """ + fut = asyncio.run_coroutine_threadsafe(coro, loop) + return fut.result(timeout=timeout) + + +def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) -> None: + """Deterministically await ``mgr``'s tracked background tasks. + + Replaces fixed sleeps for synchronizing with scheduled dead-grant + drops / spawned refreshes: exact, and immune to slow-runner flake. + """ + + async def _drain() -> None: + tasks = [t for t in list(mgr._background_tasks) if not t.done()] + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + _run_on_loop(loop, _drain()) + + def make_oidc_test_config(**overrides: Any) -> OIDCConfig: """Build a test ``OIDCConfig`` with sensible defaults. diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py index e423afa6..a4e20cfa 100644 --- a/tests/test_mcp_client.py +++ b/tests/test_mcp_client.py @@ -16,7 +16,7 @@ import anyio import mcp.types as mcp_types import pytest -from tests.conftest import _seed_static_state +from tests.conftest import _drain_background, _run_on_loop, _seed_static_state from turnstone.core.mcp_client import ( MCPClientManager, _db_servers_to_config, @@ -2851,63 +2851,49 @@ class TestSafeTransportStreams: class TestNotificationDebounce: - """Verify notification-triggered refreshes are debounced.""" + """Verify notification-triggered refreshes are debounced per (server, kind).""" def test_debounce_within_window(self): mgr = MCPClientManager({}) - mgr._last_notification_refresh["srv"] = time.monotonic() - # We can't easily call _on_notification (it's a closure), so test - # the debounce logic directly via the timestamp check + mgr._last_notification_refresh[("srv", "tools")] = time.monotonic() + # Stamp math only — the handler-level behavior (including the + # per-kind independence) is covered by TestStaticNotificationRefresh. now = time.monotonic() - last = mgr._last_notification_refresh.get("srv", 0.0) + last = mgr._last_notification_refresh.get(("srv", "tools"), 0.0) assert now - last < mgr._NOTIFICATION_DEBOUNCE def test_debounce_passes_after_window(self): mgr = MCPClientManager({}) # Set timestamp well in the past - mgr._last_notification_refresh["srv"] = time.monotonic() - 10 + mgr._last_notification_refresh[("srv", "tools")] = time.monotonic() - 10 now = time.monotonic() - last = mgr._last_notification_refresh.get("srv", 0.0) + last = mgr._last_notification_refresh.get(("srv", "tools"), 0.0) assert now - last >= mgr._NOTIFICATION_DEBOUNCE def test_debounce_is_per_server(self): mgr = MCPClientManager({}) - mgr._last_notification_refresh["srv_a"] = time.monotonic() + mgr._last_notification_refresh[("srv_a", "tools")] = time.monotonic() # srv_b has no timestamp — should pass debounce now = time.monotonic() - last_b = mgr._last_notification_refresh.get("srv_b", 0.0) + last_b = mgr._last_notification_refresh.get(("srv_b", "tools"), 0.0) assert now - last_b >= mgr._NOTIFICATION_DEBOUNCE + def test_debounce_is_per_kind(self): + mgr = MCPClientManager({}) + mgr._last_notification_refresh[("srv", "tools")] = time.monotonic() + # A prompts notification for the SAME server must pass: refreshes + # are kind-scoped, so a server-scoped stamp would drop it outright + # with no parked runner to observe the change. + now = time.monotonic() + last_prompts = mgr._last_notification_refresh.get(("srv", "prompts"), 0.0) + assert now - last_prompts >= mgr._NOTIFICATION_DEBOUNCE + # --------------------------------------------------------------------------- # Static-path list_changed refresh — spawned runner protocol (#839) # --------------------------------------------------------------------------- -def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any: - """Submit *coro* to *loop*, wait for the result with a 5s timeout. - - Mirrors ``test_mcp_user_pool.py``'s helper of the same name. - """ - fut = asyncio.run_coroutine_threadsafe(coro, loop) - return fut.result(timeout=5) - - -def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) -> None: - """Deterministically await the manager's tracked background tasks. - - Replaces fixed sleeps for synchronizing with spawned refreshes: - exact, and immune to slow-runner flake. - """ - - async def _drain() -> None: - tasks = [t for t in list(mgr._background_tasks) if not t.done()] - if tasks: - await asyncio.gather(*tasks, return_exceptions=True) - - _run_on_loop(loop, _drain()) - - class TestStaticNotificationRefresh: """The static ``*/list_changed`` protocol: spawn / serialize / coalesce. @@ -2953,7 +2939,7 @@ class TestStaticNotificationRefresh: assert refreshed == ["srv"] # Debounce stamp consumed at schedule time (storm dedupe); the # runner released its coalesce marker at lock acquire. - assert "srv" in mgr._last_notification_refresh + assert ("srv", "tools") in mgr._last_notification_refresh assert not mgr._static_refresh_pending def test_handler_ignores_non_list_changed_messages(self, running_loop_mgr) -> None: @@ -2968,7 +2954,7 @@ class TestStaticNotificationRefresh: return len(mgr._background_tasks) - before assert _run_on_loop(loop, _fire()) == 0 - assert "srv" not in mgr._last_notification_refresh + assert ("srv", "tools") not in mgr._last_notification_refresh assert not mgr._static_refresh_pending def test_notification_coalesces_when_refresh_already_queued(self, running_loop_mgr) -> None: @@ -3001,7 +2987,7 @@ class TestStaticNotificationRefresh: await handler(note) # Age the stamp past the debounce window: the MARKER (not # the stamp) must do the suppression for the second fire. - mgr._last_notification_refresh["srv"] = time.monotonic() - 10.0 + mgr._last_notification_refresh[("srv", "tools")] = time.monotonic() - 10.0 await handler(note) spawned = len(mgr._background_tasks) - before lock.release() @@ -3031,21 +3017,24 @@ class TestStaticNotificationRefresh: _seed_static_state(mgr, "srv", session=MagicMock()) _run_on_loop(loop, _seed()) - mgr._last_notification_refresh["srv"] = 123.0 + mgr._last_notification_refresh[("srv", "tools")] = 123.0 async def _boom(_name: str) -> tuple[list[str], list[str]]: raise TimeoutError("slow server") _run_on_loop(loop, mgr._run_static_notification_refresh("srv", "tools", _boom)) - assert mgr._last_notification_refresh["srv"] == 123.0 - assert mgr._last_error["srv"] == "Refresh failed: TimeoutError" + assert mgr._last_notification_refresh[("srv", "tools")] == 123.0 + # type + message, never the serialized chain: the message text is + # diagnostic and header-free; exc_info's chained request carries + # the configured bearer. + assert mgr._last_error["srv"] == "Refresh failed: TimeoutError: slow server" async def _ok(_name: str) -> tuple[list[str], list[str]]: return [], [] - mgr._last_notification_refresh["srv"] = 456.0 + mgr._last_notification_refresh[("srv", "tools")] = 456.0 _run_on_loop(loop, mgr._run_static_notification_refresh("srv", "tools", _ok)) - assert mgr._last_notification_refresh["srv"] == 456.0 + assert mgr._last_notification_refresh[("srv", "tools")] == 456.0 assert "srv" not in mgr._last_error # success clears the error def test_notification_refresh_catches_exception_group(self, running_loop_mgr) -> None: @@ -3065,14 +3054,16 @@ class TestStaticNotificationRefresh: _seed_static_state(mgr, "srv", session=MagicMock()) _run_on_loop(loop, _seed()) - mgr._last_notification_refresh["srv"] = 123.0 + mgr._last_notification_refresh[("srv", "tools")] = 123.0 async def _wedge(_name: str) -> tuple[list[str], list[str]]: raise BaseExceptionGroup("wedged transport", [asyncio.CancelledError()]) _run_on_loop(loop, mgr._run_static_notification_refresh("srv", "tools", _wedge)) - assert mgr._last_notification_refresh["srv"] == 123.0 - assert mgr._last_error["srv"] == "Refresh failed: BaseExceptionGroup" + assert mgr._last_notification_refresh[("srv", "tools")] == 123.0 + assert mgr._last_error["srv"] == ( + "Refresh failed: BaseExceptionGroup: wedged transport (1 sub-exception)" + ) def test_notification_refresh_serializes_on_connect_lock(self, running_loop_mgr) -> None: """The runner must take the per-name connect lock before @@ -3239,11 +3230,13 @@ class TestStaticNotificationRefresh: async def _scenario() -> None: _seed_static_state(mgr, "srv", session=MagicMock()) - mgr._last_notification_refresh["srv"] = 123.0 + mgr._last_notification_refresh[("srv", "tools")] = 123.0 + mgr._last_notification_refresh[("srv", "prompts")] = 123.0 await mgr._teardown_static_session("srv") _run_on_loop(loop, _scenario()) - assert "srv" not in mgr._last_notification_refresh + assert ("srv", "tools") not in mgr._last_notification_refresh + assert ("srv", "prompts") not in mgr._last_notification_refresh def test_owner_death_pops_debounce_stamp(self, running_loop_mgr) -> None: """The unrequested-collapse path (owner done-callback) is a @@ -3259,21 +3252,21 @@ class TestStaticNotificationRefresh: task = asyncio.ensure_future(_noop()) await task state.owner_task = task - mgr._last_notification_refresh["srv"] = 123.0 + mgr._last_notification_refresh[("srv", "tools")] = 123.0 mgr._on_static_owner_death("srv", task) _run_on_loop(loop, _scenario()) - assert "srv" not in mgr._last_notification_refresh + assert ("srv", "tools") not in mgr._last_notification_refresh def test_dead_transport_eviction_pops_debounce_stamp(self) -> None: """The dispatch-observed transport-failure eviction is a teardown too — the stamp must not survive the session it throttled.""" mgr = MCPClientManager({}) _seed_static_state(mgr, "srv", session=MagicMock()) - mgr._last_notification_refresh["srv"] = 123.0 + mgr._last_notification_refresh[("srv", "tools")] = 123.0 mgr._record_and_evict_on_dead_transport("srv", anyio.ClosedResourceError()) assert mgr._static_servers["srv"].session is None - assert "srv" not in mgr._last_notification_refresh + assert ("srv", "tools") not in mgr._last_notification_refresh def test_refresh_server_tools_bounded_by_timeout(self) -> None: """A wedged server's list call must not hang a spawned refresh @@ -3322,6 +3315,158 @@ class TestStaticNotificationRefresh: assert mgr._static_servers["srv"].tools == [] # new entry untouched assert old_state.tools == [] # stale result not published + def test_different_kind_notification_not_debounced(self, running_loop_mgr) -> None: + """A tools push must not swallow a prompts push arriving inside + the same debounce window: refreshes are kind-scoped, so a + server-scoped stamp would drop the prompts notification outright + — no runner exists or is spawned for prompts, and the new prompt + never appears until the server pushes again (the same staleness + class #839 was opened to fix).""" + mgr, loop, _thread = running_loop_mgr + refreshed: list[str] = [] + + async def _rec_tools(_name: str) -> tuple[list[str], list[str]]: + refreshed.append("tools") + return [], [] + + async def _rec_prompts(_name: str) -> None: + refreshed.append("prompts") + + mgr._refresh_server_tools = _rec_tools # type: ignore[method-assign] + mgr._refresh_server_prompts = _rec_prompts # type: ignore[method-assign] + handler = mgr._make_static_notification_handler("srv") + + async def _fire_both() -> None: + mgr._static_connect_lock_for("srv") + _seed_static_state(mgr, "srv", session=MagicMock()) + tools_note = mcp_types.ServerNotification( + mcp_types.ToolListChangedNotification(method="notifications/tools/list_changed") + ) + prompts_note = mcp_types.ServerNotification( + mcp_types.PromptListChangedNotification(method="notifications/prompts/list_changed") + ) + await handler(tools_note) + # Immediately inside the tools stamp's window. + await handler(prompts_note) + + _run_on_loop(loop, _fire_both()) + _drain_background(mgr, loop) + assert sorted(refreshed) == ["prompts", "tools"], ( + "a same-window different-kind notification must spawn its own refresh" + ) + + def test_refresh_server_superseded_by_remove_returns_none(self, running_loop_mgr) -> None: + """A ``_refresh_server`` pass that parked on a lock retired by + ``remove_server_sync`` must publish nothing and write NO status: + the re-add's discovery owns the new generation, and this pass + running its list calls under the retired lock would be a second, + unserialized catalog publisher — the exact race the lock exists + to close. It must also not resurrect ``_last_refresh`` / + ``_last_error`` rows for the (possibly gone) server.""" + mgr, loop, _thread = running_loop_mgr + refreshed: list[str] = [] + + async def _rec(_name: str) -> tuple[list[str], list[str]]: + refreshed.append("ran") + return [], [] + + mgr._refresh_server_tools = _rec # type: ignore[method-assign] + mgr._refresh_server_resources = _rec # type: ignore[method-assign] + mgr._refresh_server_prompts = _rec # type: ignore[method-assign] + + async def _scenario() -> Any: + old_lock = mgr._static_connect_lock_for("srv") + _seed_static_state(mgr, "srv", session=MagicMock()) + await old_lock.acquire() + task = asyncio.ensure_future(mgr._refresh_server("srv")) + for _ in range(3): + await asyncio.sleep(0) + # Simulate remove → re-add while parked: retire the lock. + mgr._static_connect_locks.pop("srv", None) + mgr._static_connect_lock_for("srv") + old_lock.release() + return await task + + result = _run_on_loop(loop, _scenario()) + assert result is None + assert refreshed == [] + assert "srv" not in mgr._last_refresh, "superseded pass must not write a status row" + assert "srv" not in mgr._last_error + + def test_refresh_server_logged_swallows_failure(self, running_loop_mgr) -> None: + """The spawned post-reconnect pass has no caller to observe a + re-raise; an escaping exception would reach ``_spawn_background``'s + ``exc_info`` failure log, which serializes the bearer-carrying + request chain for ``auth_type=static`` servers. The wrapper must + swallow, record the pill, and leave the error row written by + ``_refresh_server``.""" + mgr, loop, _thread = running_loop_mgr + + async def _boom(_name: str) -> tuple[list[str], list[str]]: + raise TimeoutError("slow server") + + mgr._refresh_server_tools = _boom # type: ignore[method-assign] + mgr._refresh_server_resources = _boom # type: ignore[method-assign] + mgr._refresh_server_prompts = _boom # type: ignore[method-assign] + + async def _scenario() -> None: + mgr._static_connect_lock_for("srv") + _seed_static_state(mgr, "srv", session=MagicMock()) + await mgr._refresh_server_logged("srv") # must not raise + + _run_on_loop(loop, _scenario()) + assert mgr._last_refresh["srv"][1] == "error:TimeoutError" + assert mgr._last_error["srv"] == "Refresh failed: TimeoutError: slow server" + + def test_resources_refresh_does_not_orphan_sibling_on_fast_failure(self) -> None: + """Fail-fast gather leaves the surviving list call running + DETACHED — outside the timeout scope and the lock serialization + — as an unbounded request on the shared session. With + ``return_exceptions=True`` both calls complete before the first + failure is re-raised.""" + mgr = MCPClientManager({}) + sibling_completed: list[bool] = [] + + async def _slow_resources() -> Any: + await asyncio.sleep(0.05) + sibling_completed.append(True) + result = MagicMock() + result.resources = [] + return result + + async def _fast_fail_templates() -> Any: + raise RuntimeError("method not found") + + session = MagicMock() + session.list_resources = _slow_resources + session.list_resource_templates = _fast_fail_templates + _seed_static_state(mgr, "srv", session=session, supports_resources=True) + + async def _run() -> None: + await mgr._refresh_server_resources("srv") + + with pytest.raises(RuntimeError, match="method not found"): + asyncio.run(_run()) + assert sibling_completed == [True], ( + "the sibling list call must complete inside the scope, not be orphaned" + ) + + def test_remove_server_clears_markers_and_stamps(self) -> None: + """``remove_server_sync`` must clear the server's coalesce + markers: a parked old-generation runner's marker would otherwise + coalesce AWAY a re-added server's first push, and that runner + bails at its lock-identity check without refreshing — the pushed + change would be silently dropped.""" + mgr = MCPClientManager({}) # no loop → direct-mutation branch + _seed_static_state(mgr, "srv", session=None) + mgr._server_configs["srv"] = {"type": "http", "url": "http://x/mcp"} + mgr._static_refresh_pending.add(("srv", "tools")) + mgr._static_refresh_pending.add(("srv", "prompts")) + mgr._last_notification_refresh[("srv", "tools")] = 123.0 + mgr.remove_server_sync("srv") + assert not mgr._static_refresh_pending + assert ("srv", "tools") not in mgr._last_notification_refresh + def test_refresh_server_serializes_on_connect_lock(self, running_loop_mgr) -> None: """The manual/periodic refresh is a catalog publisher too: it must queue behind the same per-name lock as connect wiring and @@ -3583,10 +3728,12 @@ class TestCBAutoReconnectRefresh: assert session is new_session def test_auto_reconnect_retrieves_and_logs_refresh_failure(self, running_loop_mgr): - """A refresh failure must be RETRIEVED and logged by the task's - done-callback — not abandoned for asyncio to report as "Task exception - was never retrieved" at GC time (which lands on whatever stream pytest - has attached by then: the closed-file CI spew).""" + """A refresh failure must be logged INSIDE ``_refresh_server_logged`` + (type + message, no ``exc_info``) — not by the done-callback, whose + ``exc_info`` log serializes the chained ``httpx.Request`` carrying the + configured bearer for ``auth_type=static`` servers, and not abandoned + for asyncio to report as "Task exception was never retrieved" at GC + time (the closed-file CI spew).""" import threading as _threading mgr, _loop, _thread = running_loop_mgr @@ -3616,7 +3763,9 @@ class TestCBAutoReconnectRefresh: warn_calls = [] while not warn_calls and time.time() < deadline: warn_calls = [ - c for c in mock_log.warning.call_args_list if "MCP background" in str(c.args[0]) + c + for c in mock_log.warning.call_args_list + if "Post-reconnect catalog refresh failed" in str(c.args[0]) ] time.sleep(0.02) # The tracked task must also fully drain (emptiness now implies @@ -3625,14 +3774,20 @@ class TestCBAutoReconnectRefresh: while mgr._background_tasks and time.time() < deadline: time.sleep(0.02) assert not mgr._background_tasks, "background refresh task never drained" + # The done-callback's exc_info channel must have stayed silent: + # the wrapper swallowed the failure before it could escape. + assert not [ + c for c in mock_log.warning.call_args_list if "MCP background" in str(c.args[0]) + ], "failure escaped to _spawn_background's exc_info log (bearer-leak channel)" assert session is new_session assert warn_calls, ( - "the refresh failure must be logged by the done-callback, not left " - "for GC-time reporting" + "the refresh failure must be logged by _refresh_server_logged, not " + "left for GC-time reporting" ) - exc = warn_calls[0].kwargs.get("exc_info") - assert isinstance(exc, RuntimeError) - assert "catalog fetch broke" in str(exc) + # type + message as structured args; exc_info must NOT be passed. + assert warn_calls[0].kwargs.get("exc_info") is None + assert warn_calls[0].args[2] == "RuntimeError" + assert "catalog fetch broke" in str(warn_calls[0].args[3]) def _run_hl(loop: asyncio.AbstractEventLoop, coro: Any) -> Any: diff --git a/tests/test_mcp_live_push_refresh.py b/tests/test_mcp_live_push_refresh.py index b846a970..d2670a01 100644 --- a/tests/test_mcp_live_push_refresh.py +++ b/tests/test_mcp_live_push_refresh.py @@ -30,7 +30,6 @@ Self-contained (spawns its own server; no LLM backend, no network beyond from __future__ import annotations -import signal import socket import subprocess import sys @@ -186,5 +185,5 @@ class TestPushRefreshNoDeadlock: if mgr is not None: mgr.shutdown() if proc is not None: - proc.send_signal(signal.SIGKILL) + proc.kill() proc.wait(timeout=5) diff --git a/tests/test_mcp_pool_auth_introspection.py b/tests/test_mcp_pool_auth_introspection.py index e61b08ad..3fb753c8 100644 --- a/tests/test_mcp_pool_auth_introspection.py +++ b/tests/test_mcp_pool_auth_introspection.py @@ -34,7 +34,7 @@ from unittest.mock import AsyncMock, MagicMock import httpx import pytest -from tests.conftest import make_mcp_token_cipher, stop_loop_thread +from tests.conftest import _run_on_loop, make_mcp_token_cipher, stop_loop_thread from turnstone.core.mcp_client import ( MCPClientManager, _AuthCapture, @@ -134,11 +134,6 @@ def running_loop_mgr(): stop_loop_thread(loop, thread) -def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any: - fut = asyncio.run_coroutine_threadsafe(coro, loop) - return fut.result(timeout=5) - - # --------------------------------------------------------------------------- # _classify_failure with capture vs legacy fallback # --------------------------------------------------------------------------- diff --git a/tests/test_mcp_user_catalog.py b/tests/test_mcp_user_catalog.py index 04844df2..dbe89af4 100644 --- a/tests/test_mcp_user_catalog.py +++ b/tests/test_mcp_user_catalog.py @@ -25,6 +25,7 @@ from unittest.mock import MagicMock import httpx import pytest +from tests.conftest import _run_on_loop from turnstone.core.mcp_client import ( MCPClientManager, PoolEntryState, @@ -64,11 +65,6 @@ def running_loop_mgr() -> Any: thread.join(timeout=2) -def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any, timeout: float = 10) -> Any: - fut = asyncio.run_coroutine_threadsafe(coro, loop) - return fut.result(timeout=timeout) - - def _build_mock_transport_factory( mgr: MCPClientManager, monkeypatch: pytest.MonkeyPatch, @@ -577,7 +573,7 @@ def test_close_pool_entry_if_idle_cools_entry_for_live_session_user( assert mgr.is_mcp_tool("mcp__pool-srv__do_thing", user_id="user-1") is True assert mgr._user_pool_entries[key].in_flight == 0 # Seed the debounce dict so the prune-on-close is observable. - mgr._last_pool_notification_refresh[key] = 0.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 0.0 user_calls = [0] admin_calls = [0] @@ -606,7 +602,7 @@ def test_close_pool_entry_if_idle_cools_entry_for_live_session_user( # dispatcher's next acquire needs the same lock. assert key in mgr._user_pool_locks # Debounce stamp pruned with the transport. - assert key not in mgr._last_pool_notification_refresh + assert (key, "tools") not in mgr._last_pool_notification_refresh # Per-user catalog view untouched. assert "user-1" in mgr._user_tool_map assert "user-1" in mgr._user_tools @@ -655,7 +651,7 @@ def test_close_pool_entry_if_idle_drops_entry_without_live_listener( assert mgr.is_mcp_tool("mcp__pool-srv__do_thing", user_id="user-1") is True assert mgr._user_pool_entries[key].in_flight == 0 # Seed the debounce dict so the perf-1 prune is observable. - mgr._last_pool_notification_refresh[key] = 0.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 0.0 admin_calls = [0] other_calls = [0] @@ -678,7 +674,7 @@ def test_close_pool_entry_if_idle_drops_entry_without_live_listener( assert key not in mgr._user_pool_last_used assert key not in mgr._user_pool_locks # perf-1 prune: debounce dict no longer carries the key. - assert key not in mgr._last_pool_notification_refresh + assert (key, "tools") not in mgr._last_pool_notification_refresh # Catalog cleanup ran in BOTH dicts (bug-1 sibling + bug-2 cleanup). assert "user-1" not in mgr._user_tool_map assert "user-1" not in mgr._user_tools diff --git a/tests/test_mcp_user_pool.py b/tests/test_mcp_user_pool.py index 9e2ea489..186c4709 100644 --- a/tests/test_mcp_user_pool.py +++ b/tests/test_mcp_user_pool.py @@ -26,7 +26,12 @@ from unittest.mock import AsyncMock, MagicMock, patch import mcp.types as mcp_types import pytest -from tests.conftest import make_mcp_token_cipher, stop_loop_thread +from tests.conftest import ( + _drain_background, + _run_on_loop, + make_mcp_token_cipher, + stop_loop_thread, +) from turnstone.core.mcp_client import MCPClientManager, PoolEntryState from turnstone.core.mcp_crypto import MCPTokenStore from turnstone.core.storage._sqlite import SQLiteBackend @@ -146,12 +151,6 @@ def running_loop_mgr(): stop_loop_thread(loop, thread) -def _run_on_loop(loop: asyncio.AbstractEventLoop, coro: Any) -> Any: - """Submit *coro* to *loop*, wait for the result with a 5s timeout.""" - fut = asyncio.run_coroutine_threadsafe(coro, loop) - return fut.result(timeout=5) - - def _fake_pool_tools(server_name: str, tool_name: str) -> list[dict[str, Any]]: """One OpenAI-format tool entry, shaped as the pool catalog seeds expect. @@ -171,21 +170,6 @@ def _fake_pool_tools(server_name: str, tool_name: str) -> list[dict[str, Any]]: ] -def _drain_background(mgr: MCPClientManager, loop: asyncio.AbstractEventLoop) -> None: - """Deterministically await the manager's tracked background tasks. - - Replaces fixed sleeps for synchronizing with scheduled dead-grant - drops / spawned refreshes: exact, and immune to slow-runner flake. - """ - - async def _drain() -> None: - tasks = [t for t in list(mgr._background_tasks) if not t.done()] - if tasks: - await asyncio.gather(*tasks, return_exceptions=True) - - _run_on_loop(loop, _drain()) - - # --------------------------------------------------------------------------- # Pool data structures # --------------------------------------------------------------------------- @@ -713,20 +697,20 @@ class TestEviction: entry.session = MagicMock() # runner refreshes only live sessions _run_on_loop(loop, _seed()) - mgr._last_pool_notification_refresh[key] = 123.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 123.0 async def _boom(_key: tuple[str, str]) -> tuple[list[str], list[str]]: raise TimeoutError("slow server") _run_on_loop(loop, mgr._run_notification_refresh(key, "tools", _boom)) - assert mgr._last_pool_notification_refresh[key] == 123.0 + assert mgr._last_pool_notification_refresh[(key, "tools")] == 123.0 async def _ok(_key: tuple[str, str]) -> tuple[list[str], list[str]]: return [], [] - mgr._last_pool_notification_refresh[key] = 456.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 456.0 _run_on_loop(loop, mgr._run_notification_refresh(key, "tools", _ok)) - assert mgr._last_pool_notification_refresh[key] == 456.0 + assert mgr._last_pool_notification_refresh[(key, "tools")] == 456.0 def test_notification_refresh_catches_exception_group(self, running_loop_mgr) -> None: """A wedged anyio transport surfaces session-op failures as @@ -746,13 +730,13 @@ class TestEviction: entry.session = MagicMock() # runner refreshes only live sessions _run_on_loop(loop, _seed()) - mgr._last_pool_notification_refresh[key] = 123.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 123.0 async def _wedge(_key: tuple[str, str]) -> tuple[list[str], list[str]]: raise BaseExceptionGroup("wedged transport", [asyncio.CancelledError()]) _run_on_loop(loop, mgr._run_notification_refresh(key, "tools", _wedge)) - assert mgr._last_pool_notification_refresh[key] == 123.0 + assert mgr._last_pool_notification_refresh[(key, "tools")] == 123.0 def test_notification_refresh_serializes_on_open_lock(self, running_loop_mgr) -> None: """The runner must take ``open_lock`` before refreshing: an @@ -845,7 +829,7 @@ class TestEviction: await handler(note) # Age the stamp past the debounce window: the MARKER (not # the stamp) must do the suppression for the second fire. - mgr._last_pool_notification_refresh[key] = time.monotonic() - 10.0 + mgr._last_pool_notification_refresh[(key, "tools")] = time.monotonic() - 10.0 await handler(note) spawned = len(mgr._background_tasks) - before entry.open_lock.release() @@ -926,9 +910,9 @@ class TestEviction: entry.session = MagicMock() _run_on_loop(loop, _seed()) - mgr._last_pool_notification_refresh[key] = 123.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 123.0 _run_on_loop(loop, mgr._teardown_pool_entry(key)) - assert key not in mgr._last_pool_notification_refresh + assert (key, "tools") not in mgr._last_pool_notification_refresh def test_owner_death_pops_debounce_stamp(self, running_loop_mgr) -> None: """The unrequested-collapse path (owner done-callback) is a @@ -946,11 +930,11 @@ class TestEviction: task = asyncio.ensure_future(_noop()) await task entry.owner_task = task - mgr._last_pool_notification_refresh[key] = 123.0 + mgr._last_pool_notification_refresh[(key, "tools")] = 123.0 mgr._on_pool_owner_death(key, task) _run_on_loop(loop, _scenario()) - assert key not in mgr._last_pool_notification_refresh + assert (key, "tools") not in mgr._last_pool_notification_refresh def test_list_changed_handler_spawns_refresh_off_receive_loop(self, running_loop_mgr) -> None: """The notification handler must SPAWN the refresh, not await it: @@ -987,7 +971,7 @@ class TestEviction: _drain_background(mgr, loop) assert refreshed == [key] # Debounce stamp consumed at schedule time (storm dedupe). - assert key in mgr._last_pool_notification_refresh + assert (key, "tools") in mgr._last_pool_notification_refresh def test_lookup_grant_dead_requires_wired_infrastructure(self, running_loop_mgr) -> None: """kind='missing' is authoritative only when the stores that diff --git a/turnstone/core/mcp_client.py b/turnstone/core/mcp_client.py index da3a3612..618aef5a 100644 --- a/turnstone/core/mcp_client.py +++ b/turnstone/core/mcp_client.py @@ -717,8 +717,11 @@ class MCPClientManager: self._circuit_open_until: dict[str, float] = {} # monotonic timestamp self._circuit_trip_count: dict[str, int] = {} # backoff exponent - # Notification debounce (per-server) - self._last_notification_refresh: dict[str, float] = {} + # Notification debounce, keyed ``(server, kind)`` — kind-scoped to + # match the kind-scoped refreshes: a server-scoped stamp would drop + # a different-kind notification inside the window outright, with no + # parked runner to observe the change. + self._last_notification_refresh: dict[tuple[str, str], float] = {} # Coalescing markers for spawned static ``list_changed`` refreshes, # keyed ``(server, kind)``: set at spawn, cleared the moment the # runner acquires the per-name connect lock (before its list call). @@ -796,11 +799,14 @@ class MCPClientManager: self._user_prompt_map: dict[str, dict[str, tuple[str, str]]] = {} self._user_prompts: dict[str, list[dict[str, Any]]] = {} - # Notification debounce for pool sessions, keyed ``(user_id, server)``. - # Mirrors ``_last_notification_refresh`` (static) but per-pool-key - # so a noisy server in one user's pool doesn't suppress a refresh - # in another user's pool of the same server. - self._last_pool_notification_refresh: dict[tuple[str, str], float] = {} + # Notification debounce for pool sessions, keyed + # ``((user_id, server), kind)``. Mirrors ``_last_notification_refresh`` + # (static) but per-pool-key so a noisy server in one user's pool + # doesn't suppress a refresh in another user's pool of the same + # server — and kind-scoped for the same reason as the static stamp: + # refreshes are kind-scoped, so a key-scoped stamp would drop a + # different-kind notification inside the window outright. + self._last_pool_notification_refresh: dict[tuple[tuple[str, str], str], float] = {} # Coalescing markers for spawned ``list_changed`` refreshes, keyed # ``((user_id, server), kind)``: set at spawn, cleared the moment # the runner acquires ``open_lock`` (before its list call). While @@ -1345,7 +1351,8 @@ class MCPClientManager: halves are idempotent. """ state.session = None - self._last_notification_refresh.pop(name, None) + for kind in _LIST_CHANGED_KINDS.values(): + self._last_notification_refresh.pop((name, kind), None) async def _teardown_static_session(self, name: str) -> None: """Tear down a static server's session/transport (the ONE canonical order). @@ -1556,10 +1563,19 @@ class MCPClientManager: response can only be routed by the receive loop that is parked awaiting this handler, and while it is parked EVERY in-flight and subsequent call on this shared per-node session stalls with it. - List-change refreshes are debounced, coalesced per (server, kind), - and SPAWNED as tracked tasks + List-change refreshes are debounced per (server, kind), coalesced + per (server, kind), and SPAWNED as tracked tasks (:meth:`_run_static_notification_refresh`), mirroring the pool handler (:meth:`_make_pool_notification_handler`). + + Deliberate change from the pre-#839 handler: the error pill + (``_last_error``) is no longer cleared on ANY incoming + notification — only a COMPLETED refresh (push, periodic, or + reconnect-driven) clears it, because a notification's arrival + proves nothing about whether the previous refresh failure + resolved. A transient push-refresh failure can therefore show in + the pill until the next refresh attempt (worst case: the periodic + pass), where the old handler cleared it on the next notification. """ async def _on_notification( @@ -1570,17 +1586,22 @@ class MCPClientManager: kind = _LIST_CHANGED_KINDS.get(type(msg.root)) if kind is None: return - # Debounce: skip if we refreshed this server very recently + marker = (name, kind) + # Debounce per (server, kind): refreshes are kind-scoped, so a + # server-scoped stamp would DROP a different-kind notification + # landing inside the window (tools push swallowing the prompts + # push 100ms behind it) with no parked runner to observe it — + # nothing would refresh prompts until the server pushed again. now = time.monotonic() - last = self._last_notification_refresh.get(name, 0.0) + last = self._last_notification_refresh.get(marker, 0.0) if now - last < self._NOTIFICATION_DEBOUNCE: log.debug( - "Debouncing notification from '%s' (%.1fs since last refresh)", + "Debouncing %s notification from '%s' (%.1fs since last refresh)", + kind, name, now - last, ) return - marker = (name, kind) if marker in self._static_refresh_pending: # A runner for this server+kind is queued but has not yet # issued its list call — it will observe this change when @@ -1608,7 +1629,7 @@ class MCPClientManager: "prompts": self._refresh_server_prompts, } log.info("Received %s/list_changed from '%s'", kind, name) - self._last_notification_refresh[name] = now + self._last_notification_refresh[marker] = now self._static_refresh_pending.add(marker) self._spawn_background( self._run_static_notification_refresh(name, kind, refreshers[kind]), @@ -1653,11 +1674,26 @@ class MCPClientManager: entry identity): ``remove_server_sync`` retires the lock object after teardown, so a runner that parked on the OLD lock must not touch state now owned by a NEW-lock holder. - * ``(Exception, BaseExceptionGroup)`` is caught type-name-only for - the same hygiene reason as the pool runner, but the credential at - stake is the CONFIGURED bearer: for ``auth_type=static`` servers - the chained ``httpx.Request`` headers carry it, and an escaping + * ``(Exception, BaseExceptionGroup)`` is caught for the same + hygiene reason as the pool runner, but the credential at stake + is the CONFIGURED bearer: for ``auth_type=static`` servers the + chained ``httpx.Request`` headers carry it, and an escaping exception would reach ``_spawn_background``'s ``exc_info`` log. + The log line and error pill carry ``type: str(exc)`` — the + message text (server error / URL) is diagnostic and header-free; + it is ``exc_info``'s serialized request CHAIN that leaks, so + that is the only thing withheld. + + Bounded contention residual, accepted: this runner (and the + lock-serialized :meth:`_refresh_server`) holds the connect lock + for up to one list timeout (``_CONNECT_TIMEOUT``), and a + dispatch-driven reconnect queues behind it — but a reconnect only + runs after the session was EVICTED, and every parked runner bails + instantly on the session check below once that happens, so the + added wait is at most the single in-flight list call. A caller + timeout expiring on that wait is deliberately not a breaker + record (see :meth:`_cb_auto_reconnect`); the next dispatch + retries. """ marker = (name, kind) lock = self._static_connect_locks.get(name) @@ -1688,16 +1724,19 @@ class MCPClientManager: await refresh(name) self._last_error.pop(name, None) except (Exception, BaseExceptionGroup) as exc: - # Structured fields only — ``exc_info`` would serialize the - # chained ``httpx.Request`` whose headers carry the configured - # bearer for ``auth_type=static`` servers. + # ``type: str(exc)``, never ``exc_info`` — the message text is + # diagnostic and header-free; it is the serialized exception + # CHAIN (chained ``httpx.Request`` whose headers carry the + # configured bearer for ``auth_type=static``) that must never + # reach the log. log.warning( - "Static %s refresh after notification failed for '%s' exc=%s", + "Static %s refresh after notification failed for '%s' exc=%s: %s", kind, name, type(exc).__name__, + exc, ) - self._set_error(name, f"Refresh failed: {type(exc).__name__}") + self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}") finally: if not acquired: # Cancelled (or failed) while PARKED — the marker still in @@ -2055,17 +2094,21 @@ class MCPClientManager: kind = _LIST_CHANGED_KINDS.get(type(msg.root)) if kind is None: return + marker = (key, kind) + # Debounce per (key, kind) — see the static handler's rationale: + # refreshes are kind-scoped, so a key-scoped stamp would drop a + # different-kind notification inside the window outright. now = time.monotonic() - last = self._last_pool_notification_refresh.get(key, 0.0) + last = self._last_pool_notification_refresh.get(marker, 0.0) if now - last < self._NOTIFICATION_DEBOUNCE: log.debug( - "Debouncing pool notification user=%s server=%s (%.1fs since last refresh)", + "Debouncing pool %s notification user=%s server=%s (%.1fs since last refresh)", + kind, user_id, server_name, now - last, ) return - marker = (key, kind) if marker in self._pool_refresh_pending: # A runner for this key+kind is queued but has not yet # issued its list call — it will observe this change @@ -2106,7 +2149,7 @@ class MCPClientManager: user_id, server_name, ) - self._last_pool_notification_refresh[key] = now + self._last_pool_notification_refresh[marker] = now self._pool_refresh_pending.add(marker) self._spawn_background( self._run_notification_refresh(key, kind, refreshers[kind]), @@ -3975,10 +4018,26 @@ class MCPClientManager: # 1-RTT (gather) instead of 2 sequential RTTs — both calls # share the same timeout budget and target disjoint catalogs # (resources vs. templates), so ordering is irrelevant. - res_result, tmpl_result = await asyncio.gather( + # ``return_exceptions=True`` so a fast failure on one call + # cannot orphan the sibling: fail-fast gather leaves the + # survivor running DETACHED — outside this timeout scope and + # outside the lock serialization — as an unbounded in-flight + # request on the shared session. Both awaitables complete + # here (the timeout cancels them together on expiry), then + # the first failure is re-raised. + pool_res_pair: tuple[Any, Any] = await asyncio.gather( session.list_resources(), session.list_resource_templates(), + return_exceptions=True, ) + res_result, tmpl_result = pool_res_pair + pool_res_exc: BaseException | None = next( + (r for r in (res_result, tmpl_result) if isinstance(r, BaseException)), None + ) + if pool_res_exc is not None: + raise pool_res_exc + assert not isinstance(res_result, BaseException) + assert not isinstance(tmpl_result, BaseException) if self._user_pool_entries.get(key) is not entry: # Entry replaced mid-flight — stale result, discard. return [], [] @@ -4091,11 +4150,19 @@ class MCPClientManager: ) return added, removed - async def _refresh_server(self, name: str) -> tuple[list[str], list[str]]: + async def _refresh_server(self, name: str) -> tuple[list[str], list[str]] | None: """Re-fetch tools, resources, and prompts for one server. Returns ``(added_tools, removed_tools)`` names (tool diff only, - for backward compatibility with ``/mcp refresh`` output). + for backward compatibility with ``/mcp refresh`` output), or + ``None`` when the pass was SUPERSEDED — the server was removed + (or removed and re-added) while this pass waited on the lock, so + its mandate is gone and it must publish nothing and write no + status: the removal cleaned the status maps, and a re-add's + connect discovery owns the new generation's status. Writing + anything here would either resurrect rows for a nonexistent + server or stamp a false ``ok`` over a generation this pass never + actually refreshed. Writes the ``_last_refresh`` entry on every call so the Phase 9 admin status pill reflects every refresh path — manual @@ -4129,9 +4196,31 @@ class MCPClientManager: newer one, with no convergence until the next change. No caller holds the lock coming in: ``_refresh_all`` takes it per-branch (never nested), and the two post-reconnect schedule sites spawn - this as its own task. + this as its own task (through + :meth:`_refresh_server_logged` — the raise below must never + reach ``_spawn_background``'s ``exc_info`` failure log). + + The post-acquire recheck closes the remove → re-add race the + same way :meth:`_ensure_static_connected` does — by LOCK + IDENTITY: ``remove_server_sync`` retires the lock object after + teardown, so a pass that parked on the OLD lock must not run + its list calls concurrently with a re-add publishing under the + NEW lock (two unserialized publishers — the exact race this + lock exists to close). The get-or-create acquire mirrors + ``_ensure_static_connected``; a mint for a just-removed name is + bounded (a re-add reuses it, shutdown clears it) and the + recheck below keeps it inert. """ - async with self._static_connect_lock_for(name): + lock = self._static_connect_lock_for(name) + async with lock: + if ( + self._static_connect_locks.get(name) is not lock + or self._static_servers.get(name) is None + ): + # Superseded while parked: removed (state gone), or + # removed + re-added (lock retired — the new generation + # publishes its own discovery under the NEW lock). + return None results = await asyncio.gather( self._refresh_server_tools(name), self._refresh_server_resources(name), @@ -4157,6 +4246,29 @@ class MCPClientManager: self._last_refresh[name] = (time.time(), "ok") return added, removed + async def _refresh_server_logged(self, name: str) -> None: + """:meth:`_refresh_server` for ``_spawn_background`` schedule sites. + + The spawned pass has no caller to observe the re-raise, so an + escaping exception would land in ``_spawn_background``'s + done-callback, whose ``exc_info`` log serializes the chained + ``httpx.Request`` — headers carrying the configured bearer for + ``auth_type=static`` servers. Swallow here with the same + ``type: str(exc)`` shape as :meth:`_refresh_all`'s except; + ``_refresh_server`` already wrote the ``_last_refresh`` error + row before re-raising, so no outcome is lost — only the leak. + """ + try: + await self._refresh_server(name) + except (Exception, BaseExceptionGroup) as exc: + log.warning( + "Post-reconnect catalog refresh failed for '%s' exc=%s: %s", + name, + type(exc).__name__, + exc, + ) + self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}") + async def _refresh_all( self, server_name: str | None = None ) -> dict[str, tuple[list[str], list[str]]]: @@ -4195,15 +4307,34 @@ class MCPClientManager: results[name] = (new_names, []) self._last_refresh[name] = (time.time(), "ok") continue - added, removed = await self._refresh_server(name) + refreshed = await self._refresh_server(name) + if refreshed is None: + # Superseded (removed / removed+re-added while parked + # on the lock) — a deliberate skip, not an outcome: + # nothing was refreshed, so record neither success + # nor failure for whatever generation lives now. + results[name] = ([], []) + continue + added, removed = refreshed self._cb_record_success(name) results[name] = (added, removed) except (Exception, BaseExceptionGroup) as exc: # BaseExceptionGroup: a transport task-group failure from the # reconnect/refresh must stay isolated to this server, not # abort the whole refresh pass. - log.warning("Refresh failed for MCP server '%s'", name, exc_info=True) - self._set_error(name, f"Refresh failed: {exc}") + # + # ``type: str(exc)``, never ``exc_info`` — the serialized + # exception CHAIN carries the chained ``httpx.Request`` + # whose headers hold the configured bearer for + # ``auth_type=static`` servers; the message text is + # diagnostic and header-free. + log.warning( + "Refresh failed for MCP server '%s' exc=%s: %s", + name, + type(exc).__name__, + exc, + ) + self._set_error(name, f"Refresh failed: {type(exc).__name__}: {exc}") results[name] = ([], []) # A dead transport leaves a non-None but unusable session, so # the reconnect branch at the top of this loop (gated on @@ -4331,11 +4462,23 @@ class MCPClientManager: # 1-RTT (gather) instead of 2 sequential RTTs — both calls # share the same timeout budget and target disjoint catalogs # (resources vs. templates), so ordering is irrelevant. - # Mirrors :meth:`_refresh_pool_server_resources`. - res_result, tmpl_result = await asyncio.gather( + # ``return_exceptions=True`` so a fast failure on one call + # cannot orphan the sibling as a detached, unbounded request + # on the shared session — see + # :meth:`_refresh_pool_server_resources`. + static_res_pair: tuple[Any, Any] = await asyncio.gather( session.list_resources(), session.list_resource_templates(), + return_exceptions=True, ) + res_result, tmpl_result = static_res_pair + static_res_exc: BaseException | None = next( + (r for r in (res_result, tmpl_result) if isinstance(r, BaseException)), None + ) + if static_res_exc is not None: + raise static_res_exc + assert not isinstance(res_result, BaseException) + assert not isinstance(tmpl_result, BaseException) if self._static_servers.get(name) is not state: # Entry replaced (remove + re-add) mid-flight — stale result. return @@ -5121,10 +5264,19 @@ class MCPClientManager: async with self._static_connect_lock_for(name): # Close session + transport via the owner close protocol await self._teardown_static_session(name) - # Clean up per-server state (on the event loop thread) + # Clean up per-server state (on the event loop thread). + # The direct stamp pops back up the teardown call above: + # ``_teardown_static_session`` early-returns (no pop) when + # the state entry is already gone. The marker discards + # keep a parked old-generation runner's marker from + # coalescing AWAY a re-added server's first push — that + # runner bails at its lock-identity check without + # refreshing, so nothing would cover the dropped change. self._static_servers.pop(name, None) self._last_error.pop(name, None) - self._last_notification_refresh.pop(name, None) + for kind in _LIST_CHANGED_KINDS.values(): + self._last_notification_refresh.pop((name, kind), None) + self._static_refresh_pending.discard((name, kind)) self._cb_clear(name) # Clear health-loop backoff/ping state so a later re-add of # the same name doesn't inherit stale ``due`` deadlines. @@ -5158,7 +5310,9 @@ class MCPClientManager: # No event loop (tests / pre-start) — mutate directly self._static_servers.pop(name, None) self._last_error.pop(name, None) - self._last_notification_refresh.pop(name, None) + for kind in _LIST_CHANGED_KINDS.values(): + self._last_notification_refresh.pop((name, kind), None) + self._static_refresh_pending.discard((name, kind)) self._cb_clear(name) self._rebuild_tools() self._rebuild_resources() @@ -5899,7 +6053,7 @@ class MCPClientManager: next_ping = time.monotonic() + self._static_health_check_s self._static_next_ping[name] = next_ping self._spawn_background( - self._refresh_server(name), + self._refresh_server_logged(name), f"catalog refresh after static health reconnect '{name}'", ) log.info("MCP static health: reconnected '%s'", name) @@ -6058,12 +6212,14 @@ class MCPClientManager: # Schedule catalog refresh on the loop without blocking the caller. # The reconnected session is valid for the imminent dispatch; catalog # drift will be reconciled on the loop in the background. The task is - # tracked: a refresh FAILURE is logged by the done-callback — this - # except only covers the scheduling itself. + # tracked; a refresh FAILURE is logged inside ``_refresh_server_logged`` + # (type + message, never the done-callback's ``exc_info`` — the chain + # carries the configured bearer) — this except only covers the + # scheduling itself. def _schedule_refresh() -> None: try: self._spawn_background( - self._refresh_server(server_name), + self._refresh_server_logged(server_name), f"catalog refresh after reconnect for '{server_name}'", ) except Exception: @@ -7216,7 +7372,8 @@ class MCPClientManager: and the pop is a no-op. """ entry.drop_session() - self._last_pool_notification_refresh.pop(key, None) + for kind in _LIST_CHANGED_KINDS.values(): + self._last_pool_notification_refresh.pop((key, kind), None) def _evict_session(self, key: tuple[str, str]) -> None: """Drop the cached session on a pool entry; KEEP the catalog.