diff --git a/docs/architecture.md b/docs/architecture.md index a17c8be1..7362a0f6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -546,11 +546,9 @@ adds, removes, or reconnects servers as needed. 6. `_exec_mcp_tool()` calls `call_tool_sync()` which dispatches to the async loop via `asyncio.run_coroutine_threadsafe()` -**Tool refresh:** Three mechanisms keep tools up-to-date without restart: +**Tool refresh:** Two mechanisms keep tools up-to-date without restart: - **Push:** Servers declaring `tools.listChanged` send `ToolListChangedNotification`; the registered `message_handler` triggers immediate single-server refresh. -- **Periodic:** Servers without push support are polled on a staggered interval - (default 4 h, configurable via `[mcp] refresh_interval` or `--mcp-refresh-interval`). - **Manual:** `/mcp refresh [server]` calls `refresh_sync()` for on-demand refresh (also attempts reconnection for disconnected servers). @@ -572,10 +570,11 @@ from a healthy connection do not trip the breaker. When the cooldown expires (`call_tool_sync`, `read_resource_sync`, `get_prompt_sync`, `refresh_sync`) cancel orphaned futures on timeout to prevent coroutine accumulation on the background event loop. Push notification refreshes are debounced (5 s per -server) to protect against notification storms. The periodic refresh loop -attempts reconnection for disconnected servers with exponential backoff -(60 s–1 h). Transport stream references are pre-closed before stack teardown to -work around the MCP SDK's anyio cancel-scope CPU busy-loop (SDK #2147). +server) to protect against notification storms. Operators can force a +catalog refresh or full reconnect from the admin panel; reconnects clear +the circuit breaker and run a fresh handshake. Transport stream references +are pre-closed before stack teardown to work around the MCP SDK's anyio +cancel-scope CPU busy-loop (SDK #2147). **Error isolation:** Per-server connection/refresh failures are caught and logged; other servers are unaffected. Tool execution errors return error strings to the LLM diff --git a/docs/diagrams/02-package-structure.puml b/docs/diagrams/02-package-structure.puml index 61ec8512..2597fdc8 100644 --- a/docs/diagrams/02-package-structure.puml +++ b/docs/diagrams/02-package-structure.puml @@ -40,7 +40,7 @@ package "turnstone/core/" <> { component [auth.py\nAuthentication] as auth <> component [healthcheck.py\nBackendHealthMonitor] as healthcheck <> component [ratelimit.py\nRateLimiter] as ratelimit <> - component [mcp_client.py\nMCPClientManager\n(push + periodic refresh)] as mcp <> + component [mcp_client.py\nMCPClientManager\n(push + manual refresh)] as mcp <> component [tool_search.py\nToolSearchManager, BM25] as toolsearch <> component [model_registry.py\nModelRegistry] as registry <> } diff --git a/docs/diagrams/03-core-engine-classes.puml b/docs/diagrams/03-core-engine-classes.puml index 9be18e34..b026da23 100644 --- a/docs/diagrams/03-core-engine-classes.puml +++ b/docs/diagrams/03-core-engine-classes.puml @@ -253,7 +253,7 @@ class "MCPClientManager" as MCPMgr { Background asyncio event loop bridges async MCP SDK to sync ChatSession dispatch. - Push + periodic + manual refresh. + Push + manual refresh. Resources + prompts discovered alongside tools at startup. -- diff --git a/docs/diagrams/20-mcp-architecture.puml b/docs/diagrams/20-mcp-architecture.puml index ca39d9d9..261c8e01 100644 --- a/docs/diagrams/20-mcp-architecture.puml +++ b/docs/diagrams/20-mcp-architecture.puml @@ -190,21 +190,25 @@ group Push Notifications (debounced 5s per server) MCPMgr -> Storage : sync_prompts_to_storage() end -group Periodic Polling (default 4h) - MCPMgr -> MCPMgr : _periodic_refresh() +group Manual Refresh + Session -> MCPMgr : refresh_sync() note right - Only polls capabilities - without push support. - Staggered per-server. - Disconnected servers get - reconnect attempts with - exponential backoff (60s-1h). + /mcp refresh [server] — + re-fetches catalog and + attempts reconnect for + disconnected servers. end note end -group Manual Refresh - Session -> MCPMgr : refresh_sync() - note right: /mcp refresh [server] +group Manual Reconnect + Session -> MCPMgr : reconnect_sync(name) + note right + Operator-driven via the + console admin panel — + tears down session, clears + circuit breaker, runs a + fresh handshake. + end note end == Policy Evaluation == diff --git a/docs/diagrams/png/02-package-structure.png b/docs/diagrams/png/02-package-structure.png index d31111ce..c6053131 100644 --- a/docs/diagrams/png/02-package-structure.png +++ b/docs/diagrams/png/02-package-structure.png @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:a3b5c59403a6febd81667fc8fd2a7d22bc59da6130eba0dea5449c42668d0ede -size 387044 +oid sha256:95dd5ebc899a1261d516686a5aa3319a7f45015d411302825fa28afbfc82e1ce +size 326766 diff --git a/docs/diagrams/png/03-core-engine-classes.png b/docs/diagrams/png/03-core-engine-classes.png index 3dd5e533..b6d07b97 100644 --- a/docs/diagrams/png/03-core-engine-classes.png +++ b/docs/diagrams/png/03-core-engine-classes.png @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:474b900448ec04d1117b48a2b55614524721b2f04ac4bda66170bd0a06aae0f2 -size 624573 +oid sha256:25b5448bbb7da8ddafe4f65c6c5e6cbcaa9cb9f31746ca46d3a2241bc47b1956 +size 259687 diff --git a/docs/diagrams/png/20-mcp-architecture.png b/docs/diagrams/png/20-mcp-architecture.png index 34cdf4df..5530fac5 100644 --- a/docs/diagrams/png/20-mcp-architecture.png +++ b/docs/diagrams/png/20-mcp-architecture.png @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:7623df33be9baf7647ca1c2450640df57e1cd73e8be1f8168aae16e546ad683c -size 459941 +oid sha256:d6aff446a062aa08f316985d00c2183148694f786d7f22172bc50b30046c728b +size 379259 diff --git a/docs/settings.md b/docs/settings.md index 501102dd..bc3a063a 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -100,7 +100,7 @@ initialization: | `tools` | timeout, truncation, agent_max_turns, skip_permissions, search, search_threshold, search_max_results | | `server` | workstream_idle_timeout, max_workstreams | | `cluster` | node_fan_out_limit, mcp_max_servers | -| `mcp` | config_path, refresh_interval, registry_url | +| `mcp` | config_path, registry_url | | `ratelimit` | enabled, requests_per_second, burst, trusted_proxies | | `health` | backend_probe_interval, backend_probe_timeout, circuit_breaker_threshold, circuit_breaker_cooldown | | `judge` | enabled, model, provider, base_url, api_key, confidence_threshold, max_context_ratio, timeout, read_only_tools, output_guard, redact_secrets, cancel_on_approval | diff --git a/docs/tools.md b/docs/tools.md index 7aafb4ab..ef778ac6 100644 --- a/docs/tools.md +++ b/docs/tools.md @@ -758,22 +758,18 @@ MCP tools (3): ### Dynamic tool refresh -MCP tool lists stay up-to-date without restart through three mechanisms: +MCP tool lists stay up-to-date without restart through two mechanisms: 1. **Push notifications** -- MCP servers that declare `tools.listChanged: true` in their capabilities send `notifications/tools/list_changed` when their tool list changes. `MCPClientManager` registers a `message_handler` on each `ClientSession` that triggers an immediate refresh for that server. -2. **Periodic timer** -- Servers that do *not* support push notifications are polled - on a configurable interval (default 4 hours). The timer is staggered using a - launch-time seed (`monotonic_ns ^ pid`) so cluster nodes don't all hit MCP - servers simultaneously. Configure via `[mcp] refresh_interval` in `config.toml` - or `--mcp-refresh-interval SECONDS` on the CLI. Set to `0` to disable. - -3. **Manual** -- `/mcp refresh` re-fetches tools from all servers immediately. +2. **Manual** -- `/mcp refresh` re-fetches tools from all servers immediately. `/mcp refresh ` targets a single server. If a server has disconnected, - manual refresh attempts reconnection. + manual refresh attempts reconnection. The console admin panel exposes the + same controls (refresh / reconnect buttons per server) for cluster-wide + fan-out. When tools change, `MCPClientManager` rebuilds its merged tool list using copy-on-write (new list/dict objects assigned atomically) and notifies all active `ChatSession` @@ -781,11 +777,6 @@ instances via registered listener callbacks. Each session rebuilds its `_tools`, `_task_tools`, `_agent_tools`, and reconstructs its `ToolSearchManager` (if active), preserving the set of previously expanded (discovered) tools. -```toml -[mcp] -refresh_interval = 14400 # seconds (default 4h), 0 to disable -``` - ``` /mcp refresh MCP refresh complete: diff --git a/tests/test_auth.py b/tests/test_auth.py index 826d4d7b..eee1163a 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -207,6 +207,30 @@ class TestRequiredScope: """Only POST is elevated — GET falls through to read.""" assert required_scope("GET", "/api/_internal/mcp-reload") == "read" + def test_internal_mcp_refresh_one_needs_approve(self): + assert required_scope("POST", "/api/_internal/mcp-refresh/srv") == "approve" + + def test_v1_internal_mcp_refresh_one_needs_approve(self): + assert required_scope("POST", "/v1/api/_internal/mcp-refresh/srv") == "approve" + + def test_proxy_internal_mcp_refresh_one_needs_approve(self): + assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-refresh/srv") == "approve" + + def test_proxy_no_v1_internal_mcp_refresh_one_needs_approve(self): + assert required_scope("POST", "/node/n1/api/_internal/mcp-refresh/srv") == "approve" + + def test_internal_mcp_reconnect_one_needs_approve(self): + assert required_scope("POST", "/api/_internal/mcp-reconnect/srv") == "approve" + + def test_v1_internal_mcp_reconnect_one_needs_approve(self): + assert required_scope("POST", "/v1/api/_internal/mcp-reconnect/srv") == "approve" + + def test_proxy_internal_mcp_reconnect_one_needs_approve(self): + assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-reconnect/srv") == "approve" + + def test_proxy_no_v1_internal_mcp_reconnect_one_needs_approve(self): + assert required_scope("POST", "/node/n1/api/_internal/mcp-reconnect/srv") == "approve" + # Workstream sub-resource mutations (parametric paths) def test_ws_delete_needs_write(self): assert required_scope("POST", "/api/workstreams/abc123/delete") == "write" diff --git a/tests/test_mcp_admin_api.py b/tests/test_mcp_admin_api.py index e20dc751..f9b8703d 100644 --- a/tests/test_mcp_admin_api.py +++ b/tests/test_mcp_admin_api.py @@ -20,12 +20,16 @@ if TYPE_CHECKING: from turnstone.console.server import ( _collect_mcp_status, + _notify_nodes_mcp_reconnect_one, + _notify_nodes_mcp_refresh_one, _notify_nodes_mcp_reload, admin_create_mcp_server, admin_delete_mcp_server, admin_get_mcp_server, admin_import_mcp_config, admin_list_mcp_servers, + admin_mcp_reconnect_one, + admin_mcp_refresh_one, admin_mcp_reload, admin_update_mcp_server, ) @@ -102,6 +106,16 @@ _ROUTES = [ admin_mcp_reload, methods=["POST"], ), + Route( + "/api/admin/mcp-servers/{name}/refresh", + admin_mcp_refresh_one, + methods=["POST"], + ), + Route( + "/api/admin/mcp-servers/{name}/reconnect", + admin_mcp_reconnect_one, + methods=["POST"], + ), Route( "/api/admin/mcp-servers/{server_id}", admin_get_mcp_server, @@ -122,8 +136,12 @@ _ROUTES = [ def _routes_with_internal() -> list[Mount]: - """Routes including the node-side internal endpoint (lazy-imported).""" - from turnstone.server import internal_mcp_reload + """Routes including the node-side internal endpoints (lazy-imported).""" + from turnstone.server import ( + internal_mcp_reconnect_one, + internal_mcp_refresh_one, + internal_mcp_reload, + ) return [ Mount( @@ -131,6 +149,16 @@ def _routes_with_internal() -> list[Mount]: routes=[ *_ROUTES[0].routes, # type: ignore[union-attr] Route("/api/_internal/mcp-reload", internal_mcp_reload, methods=["POST"]), + Route( + "/api/_internal/mcp-refresh/{name}", + internal_mcp_refresh_one, + methods=["POST"], + ), + Route( + "/api/_internal/mcp-reconnect/{name}", + internal_mcp_reconnect_one, + methods=["POST"], + ), ], ), ] @@ -889,3 +917,414 @@ class TestInternalMcpReloadEndpoint: assert data["added"] == ["a"] assert data["removed"] == ["b"] assert data["updated"] == ["c"] + + +# --------------------------------------------------------------------------- +# _notify_nodes_mcp_refresh_one / _notify_nodes_mcp_reconnect_one +# --------------------------------------------------------------------------- + + +class TestNotifyNodesMcpRefreshOne: + @pytest.mark.anyio + async def test_returns_json_on_success(self): + client = AsyncMock() + client.post.return_value = _mock_resp(200, {"status": "ok"}) + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_refresh_one(req, "srv") + assert result == {"n1": {"status": "ok"}} + # Verify the URL used the safe-encoded name segment + call_args = client.post.call_args + assert call_args[0][0].endswith("/v1/api/_internal/mcp-refresh/srv") + + @pytest.mark.anyio + async def test_skips_nodes_without_url(self): + client = AsyncMock() + req = _fake_request( + {"node_id": "n1", "server_url": ""}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_refresh_one(req, "srv") + assert result == {} + client.post.assert_not_called() + + @pytest.mark.anyio + async def test_records_error_on_exception(self): + client = AsyncMock() + client.post.side_effect = ConnectionError("refused") + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_refresh_one(req, "srv") + assert "n1" in result + assert "error" in result["n1"] + assert "refused" in result["n1"]["error"] + + @pytest.mark.anyio + async def test_multiple_nodes_mixed(self): + client = AsyncMock() + client.post.side_effect = [ + _mock_resp(200, {"status": "ok"}), + TimeoutError("timeout"), + ] + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + {"node_id": "n2", "server_url": "http://n2:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_refresh_one(req, "srv") + assert result["n1"] == {"status": "ok"} + assert "error" in result["n2"] + + @pytest.mark.anyio + async def test_empty_cluster(self): + req = _fake_request() + result = await _notify_nodes_mcp_refresh_one(req, "srv") + assert result == {} + + +class TestNotifyNodesMcpReconnectOne: + @pytest.mark.anyio + async def test_returns_json_on_success(self): + client = AsyncMock() + client.post.return_value = _mock_resp(200, {"status": "ok"}) + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_reconnect_one(req, "srv") + assert result == {"n1": {"status": "ok"}} + call_args = client.post.call_args + assert call_args[0][0].endswith("/v1/api/_internal/mcp-reconnect/srv") + + @pytest.mark.anyio + async def test_skips_nodes_without_url(self): + client = AsyncMock() + req = _fake_request( + {"node_id": "n1", "server_url": ""}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_reconnect_one(req, "srv") + assert result == {} + client.post.assert_not_called() + + @pytest.mark.anyio + async def test_records_error_on_exception(self): + client = AsyncMock() + client.post.side_effect = ConnectionError("refused") + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_reconnect_one(req, "srv") + assert "n1" in result + assert "error" in result["n1"] + assert "refused" in result["n1"]["error"] + + @pytest.mark.anyio + async def test_multiple_nodes_mixed(self): + client = AsyncMock() + client.post.side_effect = [ + _mock_resp(200, {"status": "ok"}), + TimeoutError("timeout"), + ] + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + {"node_id": "n2", "server_url": "http://n2:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_reconnect_one(req, "srv") + assert result["n1"] == {"status": "ok"} + assert "error" in result["n2"] + + @pytest.mark.anyio + async def test_empty_cluster(self): + req = _fake_request() + result = await _notify_nodes_mcp_reconnect_one(req, "srv") + assert result == {} + + +# --------------------------------------------------------------------------- +# Console refresh / reconnect endpoints +# --------------------------------------------------------------------------- + + +class TestAdminMcpRefreshOneEndpoint: + """HTTP-level tests for the console refresh-one endpoint.""" + + def test_refresh_one_success(self, client: TestClient) -> None: + with patch( + "turnstone.console.server._notify_nodes_mcp_action", + new_callable=AsyncMock, + return_value={"n1": {"status": "ok"}}, + ) as mock_notify: + r = client.post("/v1/api/admin/mcp-servers/srv/refresh") + assert r.status_code == 200 + data = r.json() + assert data["status"] == "ok" + assert data["results"] == {"n1": {"status": "ok"}} + # The shared helper is called with the action verb. + mock_notify.assert_awaited_once() + args = mock_notify.await_args.args + assert args[1] == "refresh" + assert args[2] == "srv" + + def test_refresh_one_permission_denied(self, client_no_perm: TestClient) -> None: + r = client_no_perm.post("/v1/api/admin/mcp-servers/srv/refresh") + assert r.status_code == 403 + assert "admin.mcp" in r.json()["error"] + + def test_refresh_one_invalid_name(self, client: TestClient) -> None: + # Names with '__' (reserved delimiter) are rejected. + r = client.post("/v1/api/admin/mcp-servers/bad__name/refresh") + assert r.status_code == 400 + assert "invalid" in r.json()["error"].lower() + + +class TestAdminMcpReconnectOneEndpoint: + """HTTP-level tests for the console reconnect-one endpoint.""" + + def test_reconnect_one_success(self, client: TestClient) -> None: + with patch( + "turnstone.console.server._notify_nodes_mcp_action", + new_callable=AsyncMock, + return_value={"n1": {"status": "ok"}}, + ) as mock_notify: + r = client.post("/v1/api/admin/mcp-servers/srv/reconnect") + assert r.status_code == 200 + data = r.json() + assert data["status"] == "ok" + assert data["results"] == {"n1": {"status": "ok"}} + mock_notify.assert_awaited_once() + args = mock_notify.await_args.args + assert args[1] == "reconnect" + assert args[2] == "srv" + + def test_reconnect_one_permission_denied(self, client_no_perm: TestClient) -> None: + r = client_no_perm.post("/v1/api/admin/mcp-servers/srv/reconnect") + assert r.status_code == 403 + assert "admin.mcp" in r.json()["error"] + + def test_reconnect_one_invalid_name(self, client: TestClient) -> None: + r = client.post("/v1/api/admin/mcp-servers/bad__name/reconnect") + assert r.status_code == 400 + assert "invalid" in r.json()["error"].lower() + + +# --------------------------------------------------------------------------- +# Node refresh-one endpoint: POST /v1/api/_internal/mcp-refresh/{name} +# --------------------------------------------------------------------------- + + +class TestInternalMcpRefreshOneEndpoint: + """HTTP-level tests for the node-side per-server refresh endpoint.""" + + @pytest.fixture() + def node_app_factory(self, storage: SQLiteBackend): + """Build a TestClient with an MCP client manager on app.state.""" + + def _make(mgr: Any) -> TestClient: + app = Starlette( + routes=_routes_with_internal(), + middleware=[Middleware(_InjectAuthMiddleware)], + ) + app.state.auth_storage = storage + if mgr is not None: + app.state.mcp_client = mgr + return TestClient(app, raise_server_exceptions=False) + + return _make + + def test_refresh_one_success(self, node_app_factory) -> None: + mgr = MagicMock() + mgr.refresh_sync.return_value = None + mgr.get_server_status.return_value = { + "connected": True, + "tools": 3, + "resources": 0, + "prompts": 1, + "error": "", + "transport": "stdio", + "command": "/usr/bin/secret-stdio", + "url": "", + "circuit_open": False, + "consecutive_failures": 0, + } + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-refresh/srv") + assert r.status_code == 200 + data = r.json() + assert data["status"] == "ok" + # sec-3: command/url stripped from response. + assert "command" not in data["server"] + assert "url" not in data["server"] + assert data["server"]["tools"] == 3 + mgr.refresh_sync.assert_called_once_with(server_name="srv") + + def test_refresh_one_no_mcp_client_returns_503(self, node_app_factory) -> None: + c = node_app_factory(None) + r = c.post("/v1/api/_internal/mcp-refresh/srv") + assert r.status_code == 503 + assert r.json()["status"] == "error" + + def test_refresh_one_raises_returns_500(self, node_app_factory) -> None: + mgr = MagicMock() + mgr.refresh_sync.side_effect = RuntimeError("internal stdio path /etc/shadow blew up") + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-refresh/srv") + assert r.status_code == 500 + # sec-2: raw exception detail must not leak to the caller. + body = r.json() + assert body["error"] == "refresh failed" + assert "shadow" not in body["error"] + + def test_refresh_one_per_server_error_returns_500(self, node_app_factory) -> None: + # q-3 / bug-3: refresh_sync swallows per-server errors into _last_error, + # so a 200 from refresh_sync is not enough — get_server_status reports. + mgr = MagicMock() + mgr.refresh_sync.return_value = None + mgr.get_server_status.return_value = { + "connected": False, + "tools": 0, + "resources": 0, + "prompts": 0, + "error": "Refresh failed: connection refused", + "transport": "stdio", + "command": "secret", + "url": "", + "circuit_open": True, + "consecutive_failures": 5, + } + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-refresh/srv") + assert r.status_code == 500 + data = r.json() + assert data["status"] == "error" + assert data["error"] == "refresh failed" + # Public status echoed but command/url stripped. + assert "command" not in data["server"] + assert "url" not in data["server"] + assert data["server"]["circuit_open"] is True + + def test_refresh_one_invalid_name_returns_400(self, node_app_factory) -> None: + # sec-4: name validation symmetric with console side. + mgr = MagicMock() + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-refresh/bad__name") + assert r.status_code == 400 + assert "invalid" in r.json()["error"].lower() + mgr.refresh_sync.assert_not_called() + + +# --------------------------------------------------------------------------- +# Node reconnect-one endpoint: POST /v1/api/_internal/mcp-reconnect/{name} +# --------------------------------------------------------------------------- + + +class TestInternalMcpReconnectOneEndpoint: + """HTTP-level tests for the node-side per-server reconnect endpoint.""" + + @pytest.fixture() + def node_app_factory(self, storage: SQLiteBackend): + """Build a TestClient with an MCP client manager on app.state.""" + + def _make(mgr: Any) -> TestClient: + app = Starlette( + routes=_routes_with_internal(), + middleware=[Middleware(_InjectAuthMiddleware)], + ) + app.state.auth_storage = storage + if mgr is not None: + app.state.mcp_client = mgr + return TestClient(app, raise_server_exceptions=False) + + return _make + + def test_reconnect_one_success(self, node_app_factory) -> None: + mgr = MagicMock() + mgr.reconnect_sync.return_value = { + "connected": True, + "tools": 2, + "resources": 0, + "prompts": 0, + "error": "", + } + mgr.get_server_status.return_value = { + "connected": True, + "tools": 2, + "resources": 0, + "prompts": 0, + "error": "", + "transport": "stdio", + "command": "secret-cmd", + "url": "", + "circuit_open": False, + "consecutive_failures": 0, + } + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-reconnect/srv") + assert r.status_code == 200 + data = r.json() + assert data["status"] == "ok" + # sec-3: command/url stripped from response. + assert "command" not in data["server"] + assert "url" not in data["server"] + mgr.reconnect_sync.assert_called_once_with("srv") + + def test_reconnect_one_no_mcp_client_returns_503(self, node_app_factory) -> None: + c = node_app_factory(None) + r = c.post("/v1/api/_internal/mcp-reconnect/srv") + assert r.status_code == 503 + + def test_reconnect_one_returns_error_dict_500(self, node_app_factory) -> None: + mgr = MagicMock() + mgr.reconnect_sync.return_value = { + "connected": False, + "tools": 0, + "resources": 0, + "prompts": 0, + "error": "secret stdio at /etc/shadow timed out", + } + mgr.get_server_status.return_value = { + "connected": False, + "tools": 0, + "resources": 0, + "prompts": 0, + "error": "secret stdio at /etc/shadow timed out", + "transport": "stdio", + "command": "secret", + "url": "", + "circuit_open": False, + "consecutive_failures": 1, + } + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-reconnect/srv") + assert r.status_code == 500 + body = r.json() + # The top-level `error` is generic; the inner `server.error` echoes + # whatever ``get_server_status`` returned (still admin-facing). + assert body["error"] == "reconnect failed" + assert "command" not in body["server"] + assert "url" not in body["server"] + + def test_reconnect_one_raises_returns_500(self, node_app_factory) -> None: + mgr = MagicMock() + mgr.reconnect_sync.side_effect = RuntimeError("internal stdio /etc/shadow blew up") + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-reconnect/srv") + assert r.status_code == 500 + body = r.json() + assert body["error"] == "reconnect failed" + assert "shadow" not in body["error"] + + def test_reconnect_one_invalid_name_returns_400(self, node_app_factory) -> None: + # sec-4: name validation symmetric with console side. + mgr = MagicMock() + c = node_app_factory(mgr) + r = c.post("/v1/api/_internal/mcp-reconnect/bad__name") + assert r.status_code == 400 + assert "invalid" in r.json()["error"].lower() + mgr.reconnect_sync.assert_not_called() diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py index 7581e875..5e607ead 100644 --- a/tests/test_mcp_client.py +++ b/tests/test_mcp_client.py @@ -130,6 +130,31 @@ def _fake_prompt_dict( } +@pytest.fixture +def running_loop_mgr(): + """Yield a (mgr, loop, thread) triple with a background loop already running. + + Spawns an MCPClientManager with a default config of {"srv": stdio echo} + and a fresh asyncio loop driven by a daemon thread. Tests that need a + different config can mutate ``mgr._server_configs`` directly. The + fixture stops the loop and joins the thread on teardown so each test + leaves a clean slate. + """ + import threading as _threading + + cfg = {"srv": {"type": "stdio", "command": "echo"}} + mgr = MCPClientManager(cfg) + loop = asyncio.new_event_loop() + thread = _threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + mgr._loop = loop + try: + yield mgr, loop, thread + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=2) + + # --------------------------------------------------------------------------- # Schema conversion # --------------------------------------------------------------------------- @@ -1903,60 +1928,179 @@ class TestNotificationDebounce: # --------------------------------------------------------------------------- -# Fix 5: Periodic refresh backoff +# reconnect_sync — operator-driven full reconnect # --------------------------------------------------------------------------- -class TestPeriodicRefreshBackoff: - """Verify periodic refresh backoff and auto-reconnect.""" +class TestReconnectSync: + """Verify reconnect_sync tears down old session, clears CB, calls _connect_one.""" - def test_backoff_set_on_failure(self): + def test_reconnect_unknown_server_returns_error(self): mgr = MCPClientManager({}) - mgr._refresh_failures["srv"] = 1 - # Simulate what _periodic_refresh does on failure - failures = mgr._refresh_failures.get("srv", 0) + 1 - mgr._refresh_failures["srv"] = failures - backoff = min(mgr._REFRESH_BACKOFF_BASE * (2 ** (failures - 1)), mgr._REFRESH_BACKOFF_MAX) - mgr._refresh_backoff_until["srv"] = time.monotonic() + backoff - assert mgr._refresh_backoff_until["srv"] > time.monotonic() - assert failures == 2 + result = mgr.reconnect_sync("missing") + assert result == { + "connected": False, + "tools": 0, + "resources": 0, + "prompts": 0, + "error": "unknown server", + } - def test_backoff_doubles(self): - mgr = MCPClientManager({}) - b1 = min(mgr._REFRESH_BACKOFF_BASE * (2**0), mgr._REFRESH_BACKOFF_MAX) - b2 = min(mgr._REFRESH_BACKOFF_BASE * (2**1), mgr._REFRESH_BACKOFF_MAX) - b3 = min(mgr._REFRESH_BACKOFF_BASE * (2**2), mgr._REFRESH_BACKOFF_MAX) - assert b1 == 60 - assert b2 == 120 - assert b3 == 240 + def test_reconnect_clears_circuit_breaker(self, running_loop_mgr): + mgr, _loop, _thread = running_loop_mgr - def test_backoff_capped(self): - mgr = MCPClientManager({}) - b = min(mgr._REFRESH_BACKOFF_BASE * (2**20), mgr._REFRESH_BACKOFF_MAX) - assert b == mgr._REFRESH_BACKOFF_MAX + async def _fake_connect_one(name: str, _cfg: dict[str, Any]) -> None: + mgr._sessions[name] = MagicMock() - def test_backoff_clears_on_success(self): - mgr = MCPClientManager({}) - mgr._refresh_failures["srv"] = 3 - mgr._refresh_backoff_until["srv"] = time.monotonic() + 1000 - # Simulate success - mgr._refresh_failures.pop("srv", None) - mgr._refresh_backoff_until.pop("srv", None) - assert "srv" not in mgr._refresh_failures - assert "srv" not in mgr._refresh_backoff_until - - def test_server_status_includes_circuit_info(self): - mgr = MCPClientManager({"srv": {"type": "stdio", "command": "echo"}}) - status = mgr.get_server_status("srv") - assert "circuit_open" in status - assert "consecutive_failures" in status - assert status["circuit_open"] is False - assert status["consecutive_failures"] == 0 - - def test_server_status_shows_open_circuit(self): - mgr = MCPClientManager({"srv": {"type": "stdio", "command": "echo"}}) + # Pre-trip the breaker for _ in range(3): mgr._cb_record_failure("srv") - status = mgr.get_server_status("srv") - assert status["circuit_open"] is True - assert status["consecutive_failures"] == 3 + assert "srv" in mgr._circuit_open_until + + with ( + patch.object(mgr, "_connect_one", side_effect=_fake_connect_one), + patch.object(mgr, "_pre_close_streams", new=AsyncMock()), + ): + result = mgr.reconnect_sync("srv") + assert result["connected"] is True + assert result["error"] == "" + assert "srv" not in mgr._circuit_open_until + assert "srv" not in mgr._consecutive_failures + + def test_reconnect_closes_old_session_then_calls_connect_one(self, running_loop_mgr): + mgr, _loop, _thread = running_loop_mgr + + order: list[str] = [] + old_stack = MagicMock(spec=AsyncExitStack) + + async def _pre_close(name: str) -> None: + order.append("pre_close") + + async def _safe_close(stack: Any) -> None: + order.append("safe_close") + assert stack is old_stack + + async def _connect_one(name: str, _cfg: dict[str, Any]) -> None: + order.append("connect_one") + mgr._sessions[name] = MagicMock() + + mgr._sessions["srv"] = MagicMock() # populated old session + mgr._per_server_stacks["srv"] = old_stack + mgr._server_streams["srv"] = (MagicMock(), MagicMock()) + + with ( + patch.object(mgr, "_pre_close_streams", side_effect=_pre_close), + patch.object(mgr, "_safe_close_stack", side_effect=_safe_close), + patch.object(mgr, "_connect_one", side_effect=_connect_one), + ): + result = mgr.reconnect_sync("srv") + assert result["connected"] is True + assert order == ["pre_close", "safe_close", "connect_one"] + assert "srv" not in mgr._per_server_stacks + + def test_reconnect_failure_returns_error_dict(self, running_loop_mgr): + mgr, _loop, _thread = running_loop_mgr + + async def _connect_one(name: str, _cfg: dict[str, Any]) -> None: + raise RuntimeError("handshake failed") + + with ( + patch.object(mgr, "_connect_one", side_effect=_connect_one), + patch.object(mgr, "_pre_close_streams", new=AsyncMock()), + ): + result = mgr.reconnect_sync("srv") + assert result["connected"] is False + assert "handshake failed" in result["error"] + + def test_reconnect_failure_clears_stale_catalog(self, running_loop_mgr): + # bug-2: when _connect_one fails mid-reconnect, the per-server + # catalog must be dropped so the merged tool/resource/prompt maps + # don't keep advertising entries with no live session. + mgr, _loop, _thread = running_loop_mgr + + # Seed catalog state from a previous successful connect. + mgr._per_server_tools["srv"] = [_fake_openai_tool("mcp__srv__t")] + mgr._per_server_resources["srv"] = [_fake_resource_dict(server="srv")] + mgr._per_server_prompts["srv"] = [_fake_prompt_dict(server="srv")] + mgr._rebuild_tools() + mgr._rebuild_resources() + mgr._rebuild_prompts() + + async def _connect_one(name: str, _cfg: dict[str, Any]) -> None: + raise RuntimeError("handshake failed") + + with ( + patch.object(mgr, "_connect_one", side_effect=_connect_one), + patch.object(mgr, "_pre_close_streams", new=AsyncMock()), + ): + result = mgr.reconnect_sync("srv") + assert result["connected"] is False + # Per-server catalog and merged maps should both be empty for srv. + assert "srv" not in mgr._per_server_tools + assert "srv" not in mgr._per_server_resources + assert "srv" not in mgr._per_server_prompts + assert "mcp__srv__t" not in mgr._tool_map + + +# --------------------------------------------------------------------------- +# _cb_auto_reconnect — refresh-on-reconnect +# --------------------------------------------------------------------------- + + +class TestCBAutoReconnectRefresh: + """Verify _cb_auto_reconnect schedules catalog refresh after a successful reconnect. + + The refresh runs as a fire-and-forget background task on the loop so it + doesn't block the caller (perf-1). Tests wait briefly for the scheduled + task to run and observe its effect. + """ + + def test_auto_reconnect_schedules_refresh_server_on_success(self, running_loop_mgr): + import threading as _threading + + mgr, _loop, _thread = running_loop_mgr + + new_session = MagicMock() + refresh_event = _threading.Event() + + async def _connect_one(name: str, _cfg: dict[str, Any]) -> None: + mgr._sessions[name] = new_session + + async def _refresh(name: str) -> tuple[list[str], list[str]]: + refresh_event.set() + return [], [] + + with ( + patch.object(mgr, "_connect_one", side_effect=_connect_one), + patch.object(mgr, "_refresh_server", side_effect=_refresh), + ): + session = mgr._cb_auto_reconnect("srv") + # Wait for the scheduled refresh task to actually run on the loop. + assert refresh_event.wait(timeout=5), "refresh task was not scheduled" + assert session is new_session + + def test_auto_reconnect_swallows_refresh_failure(self, running_loop_mgr): + import threading as _threading + + mgr, _loop, _thread = running_loop_mgr + + new_session = MagicMock() + refresh_started = _threading.Event() + + async def _connect_one(name: str, _cfg: dict[str, Any]) -> None: + mgr._sessions[name] = new_session + + async def _refresh_failing(name: str) -> tuple[list[str], list[str]]: + refresh_started.set() + raise RuntimeError("catalog fetch broke") + + with ( + patch.object(mgr, "_connect_one", side_effect=_connect_one), + patch.object(mgr, "_refresh_server", side_effect=_refresh_failing), + ): + # Must not raise — refresh failures are non-fatal. + session = mgr._cb_auto_reconnect("srv") + # Background refresh actually started and exception was swallowed + # by the task without affecting the synchronous caller. + assert refresh_started.wait(timeout=5), "refresh task was not scheduled" + assert session is new_session diff --git a/turnstone.example.toml b/turnstone.example.toml index 7efd52e1..d3072914 100644 --- a/turnstone.example.toml +++ b/turnstone.example.toml @@ -117,7 +117,6 @@ [mcp] # config_path = "" # Path to MCP servers config file (JSON) -# refresh_interval = 14400 # Refresh interval in seconds (default: 4h) # --- Server (node, console) --- diff --git a/turnstone/cli.py b/turnstone/cli.py index a8107ca8..79df67f0 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -1015,15 +1015,6 @@ def main() -> None: help="Path to MCP server config file (standard mcpServers JSON format)", ) - from turnstone.core.config import nonneg_float - - parser.add_argument( - "--mcp-refresh-interval", - type=nonneg_float, - default=14400, - metavar="SECONDS", - help="Periodic MCP tool refresh interval for servers without push notifications (default: 14400 = 4h, 0 to disable)", - ) judge_group = parser.add_argument_group("Judge options") judge_group.add_argument( "--judge", @@ -1136,7 +1127,6 @@ def main() -> None: mcp_client = create_mcp_client( getattr(args, "mcp_config", None), - refresh_interval=getattr(args, "mcp_refresh_interval", 14400), storage=_get_storage(), ) diff --git a/turnstone/console/server.py b/turnstone/console/server.py index da37fbcd..f88cd500 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -8388,6 +8388,57 @@ async def _notify_nodes_mcp_reload(request: Request) -> dict[str, Any]: return {nid: data for nid, data in results if data is not None} +async def _notify_nodes_mcp_action(request: Request, action: str, name: str) -> dict[str, Any]: + """Tell all nodes to perform a per-server MCP action. + + *action* is the suffix of the internal endpoint — currently + ``"refresh"`` or ``"reconnect"``. Each node is hit at + ``/v1/api/_internal/mcp-{action}/{name}``. + """ + collector: ClusterCollector = request.app.state.collector + nodes = collector.get_all_nodes() + client: httpx.AsyncClient = request.app.state.proxy_client + headers = _proxy_auth_headers(request) + sem = asyncio.Semaphore(_get_fan_out_limit(request)) + safe_name = urllib.parse.quote(name, safe="") + + async def _notify(node: dict[str, Any]) -> tuple[str, Any]: + node_id = node.get("node_id", "") + url = node.get("server_url", "") + if not url: + return node_id, None + async with sem: + try: + resp = await client.post( + f"{url.rstrip('/')}/v1/api/_internal/mcp-{action}/{safe_name}", + headers=headers, + timeout=30, + ) + return node_id, resp.json() + except Exception as exc: + log.debug( + "Failed to notify node %s for MCP %s of %s", + node_id, + action, + name, + exc_info=True, + ) + return node_id, {"error": str(exc)} + + results = await asyncio.gather(*[_notify(n) for n in nodes]) + return {nid: data for nid, data in results if data is not None} + + +async def _notify_nodes_mcp_refresh_one(request: Request, name: str) -> dict[str, Any]: + """Tell all nodes to refresh a single MCP server's catalog.""" + return await _notify_nodes_mcp_action(request, "refresh", name) + + +async def _notify_nodes_mcp_reconnect_one(request: Request, name: str) -> dict[str, Any]: + """Tell all nodes to force-reconnect a single MCP server.""" + return await _notify_nodes_mcp_action(request, "reconnect", name) + + async def admin_mcp_reload(request: Request) -> JSONResponse: """POST /v1/api/admin/mcp-servers/reload — tell nodes to re-read DB.""" from turnstone.core.auth import require_permission @@ -8404,6 +8455,57 @@ async def admin_mcp_reload(request: Request) -> JSONResponse: return JSONResponse({"status": "ok", "results": results}) +async def _admin_mcp_action(request: Request, action: str) -> JSONResponse: + """Shared body for the per-server MCP admin actions. + + *action* is the verb suffix — ``"refresh"`` or ``"reconnect"``. + Auth, name validation, audit, and per-node fan-out are identical + across the two; only the audit action string and the notify call + vary, both keyed on *action*. + """ + from turnstone.core.audit import record_audit + from turnstone.core.auth import require_permission + from turnstone.core.web_helpers import require_storage_or_503 + + storage, err = require_storage_or_503(request) + if err: + return err + err = require_permission(request, "admin.mcp") + if err: + return err + + name = request.path_params["name"] + if "__" in name: + return JSONResponse({"error": "invalid server name"}, status_code=400) + + existing = storage.get_mcp_server_by_name(name) + target_id = existing.get("id", name) if existing else name + + audit_uid, ip = _audit_context(request) + record_audit( + storage, + audit_uid, + f"mcp_server.{action}", + "mcp_server", + target_id, + {"name": name}, + ip, + ) + + results = await _notify_nodes_mcp_action(request, action, name) + return JSONResponse({"status": "ok", "results": results}) + + +async def admin_mcp_refresh_one(request: Request) -> JSONResponse: + """POST /v1/api/admin/mcp-servers/{name}/refresh — refresh one server's catalog.""" + return await _admin_mcp_action(request, "refresh") + + +async def admin_mcp_reconnect_one(request: Request) -> JSONResponse: + """POST /v1/api/admin/mcp-servers/{name}/reconnect — force-reconnect one server.""" + return await _admin_mcp_action(request, "reconnect") + + async def admin_import_mcp_config(request: Request) -> JSONResponse: """POST /v1/api/admin/mcp-servers/import — import from pasted JSON config.""" import uuid @@ -11198,6 +11300,16 @@ def create_app( admin_mcp_reload, methods=["POST"], ), + Route( + "/api/admin/mcp-servers/{name}/refresh", + admin_mcp_refresh_one, + methods=["POST"], + ), + Route( + "/api/admin/mcp-servers/{name}/reconnect", + admin_mcp_reconnect_one, + methods=["POST"], + ), Route( "/api/admin/mcp-servers/{server_id}", admin_get_mcp_server, diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js index f1bff652..7bc705ba 100644 --- a/turnstone/console/static/admin.js +++ b/turnstone/console/static/admin.js @@ -3316,9 +3316,17 @@ function _renderMcpServers(items) { var detailAttr = isConfig ? 'data-mcp-detail-name="' + escapeHtml(s.name) + '"' : 'data-mcp-detail="' + escapeHtml(s.server_id) + '"'; + var actionBtns = + '' + + ''; var actions = isConfig - ? "" - : '' + '