diff --git a/docs/mcp-oauth.md b/docs/mcp-oauth.md new file mode 100644 index 00000000..8146d226 --- /dev/null +++ b/docs/mcp-oauth.md @@ -0,0 +1,117 @@ +# MCP OAuth — per-user authorization for MCP servers + +Turnstone supports **per-(user, MCP server) OAuth 2.1 + PKCE** delegation so each Turnstone user authorizes a remote MCP server with their own identity, rather than sharing a single bearer token across the deployment. This is the right shape for MCP servers that expose user-specific data (a personal CRM, an email inbox, a calendar) and for MCP servers that want per-user audit attribution. + +Per-user OAuth is opt-in per `mcp_servers` row. Local-auth Turnstone installs with no `oauth_user` rows exercise zero new code paths — the entire feature is dark by default. + +> **Note**: This is a separate authorization layer from Turnstone's own user authentication. A user who logs into Turnstone with a local username + password can still authorize a per-server OAuth MCP server. OIDC SSO and per-server OAuth are orthogonal. + +--- + +## When to use which `auth_type` + +The MCP server admin form exposes three authorization modes ("Multitenant Authorization"): + +| `auth_type` | What it means | When to use | +|---|---|---| +| `none` | No headers attached. Open MCP server (or one gated by network policy only). | Internal MCP servers on a trusted network. | +| `static` | One static bearer token, configured per server, sent on every request from every user. | Service-to-service MCP servers where per-user attribution doesn't matter, or single-tenant deployments. | +| `oauth_user` *(recommended for user-data servers)* | Each user authorizes separately via OAuth 2.1 + PKCE; Turnstone stores per-user tokens encrypted at rest. | MCP servers that expose user-specific data or that want per-user audit attribution. | + +Switching `auth_type` away from `oauth_user` orphans existing per-user tokens. Use the admin **bulk-revoke** affordance on the server row (Phase 9) to clear them, or let them expire naturally — they're inert without the matching `auth_type` value. + +--- + +## Prerequisites for `auth_type=oauth_user` + +1. **Encryption key**. Tokens are stored encrypted with Fernet. Set `[security] mcp_token_encryption_key` in `config.toml` (Turnstone won't start with an `oauth_user` row configured but no key installed). Rotate via `MultiFernet` — add the new key first, then later remove the old one once all rows have been re-encrypted. + +2. **MCP server publishes RFC 9728 PRM and RFC 8414 AS metadata** *or* you configure the AS URL override on the server row. PKCE S256 is mandatory; Turnstone refuses to connect to authorization servers that don't advertise `code_challenge_methods_supported: ["S256"]`. + +3. **OAuth client registration**. Two paths: + - **Pre-registered** (most common): you create an OAuth client at the authorization server (manually, via admin console, or via Terraform), then paste the `client_id` / `client_secret` into the Turnstone admin form. + - **Dynamic client registration** (RFC 7591): if the AS supports it and you select that mode in the admin form, Turnstone registers a client at first use and persists the `client_id` automatically. + +4. **Redirect URI** registered at the authorization server: `https://your-turnstone-host/v1/api/mcp/oauth/callback`. + +--- + +## Configuration + +### Per-server fields (admin UI) + +| Field | Required | Description | +|---|---|---| +| Server URL | Yes | The MCP server's `streamable-http` base URL. | +| Multitenant Authorization | Yes | `none` / `static` / `oauth_user` (recommended). | +| Authorization Server URL | No | Override for RFC 9728 PRM discovery. Set when your AS endpoint differs from the MCP server URL (e.g., corporate AS protecting a third-party MCP). When unset, Turnstone falls back to PRM discovery against the MCP server itself. | +| Client Registration | Yes (oauth_user) | `preregistered` or `dynamic`. | +| Client ID | Yes (preregistered) | OAuth 2.0 client ID. Stored unencrypted. | +| Client Secret | Optional (write-only) | OAuth 2.0 client secret (confidential client). Encrypted at rest. Written but never re-read by the API; field stays masked. | +| Scopes | No | Space-separated default scope set requested at the authorize endpoint. Per-tool step-up may union additional scopes from a server's `insufficient_scope` response. | +| Audience | No | RFC 8707 `resource=` parameter sent on every authorize and token request. Defaults to the MCP server URL when unset. Validate against the `aud` claim in returned JWT tokens. | + +### Encryption key + +```toml +[security] +mcp_token_encryption_key = "base64-fernet-key" +# For rotation, list the keys in priority order — first is used for new +# writes, all are tried for reads. +# mcp_token_encryption_keys = ["new-key", "old-key"] +``` + +Keep this in `config.toml` rather than environment variables. An in-process LLM with shell-tool access can read the server's environment via `env` / `os.environ` and exfiltrate any secret stored there; secrets in `config.toml` are only loaded into the server at startup and never re-read on a tool-driven path, so a prompt-injection attack against the agent cannot reach them. + +--- + +## Lifecycle + +1. **First tool call** for a user against an `oauth_user` MCP server: pool dispatch finds no stored token, returns `mcp_consent_required` to the agent. Dashboard renders an inline "Connect" action card. + +2. **User clicks Connect**: opens `/v1/api/mcp/oauth/start?server=` in a popup. Browser redirects through the AS authorize endpoint, user grants consent, AS redirects back to `/v1/api/mcp/oauth/callback`. Turnstone exchanges code → tokens via PKCE, validates audience, encrypts, persists in `mcp_user_tokens`, redirects user back to the originating URL. + +3. **Subsequent tool calls** by the same user against the same server reuse the persisted token via the per-(user, server) session pool. Tokens auto-refresh via the refresh-token grant when expired; failed refresh emits `mcp_consent_required` to drive re-consent. + +4. **Step-up scope**: when a tool call hits `403` with `WWW-Authenticate: error="insufficient_scope"`, Turnstone emits `mcp_insufficient_scope` with the parsed scope set; the dashboard offers a "Connect with additional scopes" affordance that opens `/v1/api/mcp/oauth/start?server=&scopes=` so the union of original + new scopes flows into the AS authorize request. + +5. **User revoke** (settings modal): `DELETE /v1/api/mcp/oauth/connections/{server_name}` runs the authoritative local delete + best-effort RFC 7009 upstream revoke (fire-and-forget, capped at 256 concurrent in-flight tasks). + +6. **Admin bulk-revoke** (Phase 9): `POST /v1/api/admin/mcp-servers/{name}/bulk-revoke` drops every user's token for the server. Upstream RFC 7009 revoke is intentionally **not** attempted in bulk (avoids N upstream HTTP calls per admin click); tokens at the AS expire naturally. Use the per-user revoke endpoint if you need guaranteed upstream invalidation. + +--- + +## Admin status indicators + +The MCP Servers admin tab shows per-server status pills (Phase 9): + +- **Consented users count** — distinct users with a non-expired token for this server. Surfaced as a `bulk-revoke (N)` button when ≥1; clicking it opens a confirmation dialog. Hidden when 0. +- **Last refresh** — timestamp + outcome (`ok` / `error:ClassName`) of the most recent manual or auto-reconnect refresh. Per node. Absent until at least one refresh has occurred (renders as "never" in the admin UI). + +Additional indicators (circuit-breaker state, encryption-key mismatch) are exposed via `get_server_status` on the API but do not yet have a dedicated admin pill — operators see them today via the per-server status text + error tooltip and in audit logs. A future phase may surface these as discrete pills. + +--- + +## Auth-type transitions + +| From | To | What happens | +|---|---|---| +| `none` / `static` → `oauth_user` | — | New code path activates for this server. Existing static headers (if any) are no longer sent. Users must authorize on first use. | +| `oauth_user` → `none` / `static` | — | Existing `mcp_user_tokens` rows are **orphaned** — inert without a matching `auth_type`. Use admin bulk-revoke to drop them, or let them expire. Switching back to `oauth_user` later re-activates the orphaned rows if they haven't been deleted. | +| OAuth `client_id` or `client_secret` rotated | — | Existing tokens may stop refreshing if the AS treats them as bound to the previous client. Bulk-revoke after rotation. | + +The orphan-by-default behavior is chosen so switching back to `oauth_user` is non-destructive. Bulk-revoke is the explicit cleanup path. + +--- + +## Troubleshooting + +| Symptom | Likely cause | Action | +|---|---|---| +| `mcp_consent_required` even after consenting | Token persistence failed, or refresh-token rejected by AS | Check audit log for `mcp_server.oauth.persist_failed` or `mcp_server.oauth.token_revoked`. Re-consent via settings modal. | +| `mcp_token_undecryptable_key_unknown` | Encryption key rotated without keeping the previous key in the keyring | Add the previous key back to `mcp_token_encryption_keys` until all rows have been re-encrypted, then drop. | +| `mcp_oauth_url_insecure` | MCP server URL is `http://` (not `https://`) on a non-loopback host | Use `https://`. Per-user bearers must not transit cleartext. | +| Tools fail in scheduled / Discord / Slack runs | OAuth-MCP requires browser-based consent | Users must pre-consent via the web UI. Phase 9 dashboard badge surfaces deferred consents from these runs on next login. | +| Circuit breaker open repeatedly | Transport-level errors on the MCP server (DNS, TLS, 5xx) | Check the per-server error pill; auth errors do not trip the breaker. | + +See also: `docs/operations/mcp-oauth-headless.md` for the cron / channel-driven run caveat. diff --git a/docs/operations/mcp-oauth-headless.md b/docs/operations/mcp-oauth-headless.md new file mode 100644 index 00000000..f340cebe --- /dev/null +++ b/docs/operations/mcp-oauth-headless.md @@ -0,0 +1,29 @@ +# MCP OAuth in headless / scheduled / channel-driven runs + +**Constraint**: OAuth-MCP servers (`auth_type=oauth_user`) require browser-based user consent. Users must pre-consent via the web UI before any run that cannot drive a browser redirect. + +**Affected surfaces**: + +- Scheduled workstreams (`turnstone-console` task scheduler). +- Discord adapter runs. +- Slack adapter runs. +- Any future channel adapter without an interactive browser session. + +**What happens when consent is missing**: + +A tool call against an `oauth_user` server returns a structured `mcp_consent_required` error to the agent. The agent surfaces the deferred work in its output. Turnstone persists a record to `mcp_pending_consent` so the dashboard badge surfaces the deferred consent need to the user on next login. + +**Recovery**: + +The user opens the dashboard, sees the gear-icon badge counting pending consents, opens the settings modal, clicks Connect for each affected server, and completes the OAuth dance. The pending-consent record is cleared by the OAuth callback handler on success. Subsequent scheduled / channel runs use the freshly-stored token. + +**Pre-consent recipe**: + +Before scheduling a workstream that depends on an `oauth_user` MCP server, the user should: + +1. Open the dashboard. +2. Open the settings modal (gear icon). +3. Click Connect on each MCP server the schedule will use. +4. Confirm consent in the popup. + +This stores tokens that the scheduled run will reuse. Refresh-token rotation is handled transparently on the run side; only the first consent requires browser interaction. diff --git a/tests/test_chat_session_interactivity_flag.py b/tests/test_chat_session_interactivity_flag.py new file mode 100644 index 00000000..83b49a0c --- /dev/null +++ b/tests/test_chat_session_interactivity_flag.py @@ -0,0 +1,66 @@ +"""ChatSession interactivity flag tests (Phase 9). + +Validates that ``ChatSession._is_interactive_for_consent`` is computed +correctly from ``client_type`` on construction. This is the front of +the Phase 9 plumb-through: the flag flows from here to +``_dispatch_pool_sync`` to the structured-error → pending-consent +write path. +""" + +from __future__ import annotations + +from tests._session_helpers import make_session +from turnstone.prompts import INTERACTIVE_CONSENT_CLIENT_TYPES, ClientType + + +def test_web_is_interactive() -> None: + s = make_session(client_type=ClientType.WEB) + assert s._is_interactive_for_consent is True + + +def test_cli_is_interactive() -> None: + s = make_session(client_type=ClientType.CLI) + assert s._is_interactive_for_consent is True + + +def test_chat_is_not_interactive() -> None: + # Discord / Slack adapters cannot drive a browser redirect from + # inside the channel — consent prompts must be deferred to the + # dashboard badge. + s = make_session(client_type=ClientType.CHAT) + assert s._is_interactive_for_consent is False + + +def test_scheduled_is_not_interactive() -> None: + # The scheduler runs autonomously; the user isn't online to + # complete the OAuth redirect. + s = make_session(client_type=ClientType.SCHEDULED) + assert s._is_interactive_for_consent is False + + +def test_interactive_set_matches_module_constant() -> None: + # Pin the module-level frozenset against the flag computation — + # a future reorganisation that drifts the set vs the per-session + # logic would silently break the gating. + for ct in ClientType: + s = make_session(client_type=ct) + assert s._is_interactive_for_consent == (ct in INTERACTIVE_CONSENT_CLIENT_TYPES), ct + + +def test_default_client_type_is_cli_interactive() -> None: + # Defaults preserved — make_session uses ChatSession's default + # which is CLI. Sanity check that the default user experience + # stays interactive-for-consent. + s = make_session() + assert s._client_type == ClientType.CLI + assert s._is_interactive_for_consent is True + + +def test_scheduled_env_file_exists() -> None: + """The SCHEDULED env module must exist; otherwise + ``compose_system_message`` for a scheduled session would 500.""" + from turnstone.prompts import _load + + text = _load("env/scheduled.md") + assert "Output Environment" in text + assert "consent" in text.lower() diff --git a/tests/test_mcp_admin_bulk_revoke.py b/tests/test_mcp_admin_bulk_revoke.py new file mode 100644 index 00000000..4b2a36f7 --- /dev/null +++ b/tests/test_mcp_admin_bulk_revoke.py @@ -0,0 +1,211 @@ +"""Integration tests for the Phase 9 admin bulk-revoke endpoint. + +POST /v1/api/admin/mcp-servers/{name}/bulk-revoke clears every user's +OAuth token for a server (admin-side counterpart to the per-user +DELETE /v1/api/mcp/oauth/connections/{server_name} that shipped in +Phase 8). + +Coverage: +- requires ``admin.mcp`` permission (401/403 without). +- 404 when the named server is missing. +- 400 when the server's ``auth_type`` is not ``oauth_user``. +- 200 + ``rows_deleted`` + ``consented_users_before`` on success. +- Audit row written with + ``upstream_revoke_outcome="bulk_admin_no_upstream"``. +- Token rows are gone from ``mcp_user_tokens`` post-call. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest +from starlette.applications import Starlette +from starlette.middleware import Middleware +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.routing import Mount, Route +from starlette.testclient import TestClient + +from turnstone.console.server import admin_mcp_bulk_revoke +from turnstone.core.auth import AuthResult +from turnstone.core.storage._sqlite import SQLiteBackend + +if TYPE_CHECKING: + from starlette.requests import Request + from starlette.responses import Response + + +class _InjectAdminMcp(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next: Any) -> Response: + request.state.auth_result = AuthResult( + user_id="admin-user", + scopes=frozenset({"approve"}), + token_source="config", + permissions=frozenset({"read", "write", "approve", "admin.mcp"}), + ) + return await call_next(request) + + +class _InjectNoAdminMcp(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next: Any) -> Response: + request.state.auth_result = AuthResult( + user_id="regular-user", + scopes=frozenset({"approve"}), + token_source="jwt", + permissions=frozenset({"read", "write", "approve"}), + ) + return await call_next(request) + + +def _build_app(storage: SQLiteBackend, *, with_admin_mcp: bool = True) -> Starlette: + mw = _InjectAdminMcp if with_admin_mcp else _InjectNoAdminMcp + app = Starlette( + routes=[ + Mount( + "/v1", + routes=[ + Route( + "/api/admin/mcp-servers/{name}/bulk-revoke", + admin_mcp_bulk_revoke, + methods=["POST"], + ), + ], + ), + ], + middleware=[Middleware(mw)], + ) + app.state.auth_storage = storage + return app + + +@pytest.fixture +def storage(tmp_path: Any) -> SQLiteBackend: + return SQLiteBackend(str(tmp_path / "test.db")) + + +def _seed_oauth_server( + backend: SQLiteBackend, + *, + name: str = "srv-oauth", + server_id: str = "srv-oauth-id", +) -> None: + backend.create_mcp_server( + server_id=server_id, + name=name, + transport="streamable-http", + url="https://example.com/mcp", + auth_type="oauth_user", + ) + + +def _seed_static_server( + backend: SQLiteBackend, + *, + name: str = "srv-static", + server_id: str = "srv-static-id", +) -> None: + backend.create_mcp_server( + server_id=server_id, + name=name, + transport="streamable-http", + url="https://example.com/mcp", + auth_type="static", + ) + + +def _seed_user_tokens(backend: SQLiteBackend, server_name: str, users: int) -> None: + for i in range(users): + backend.create_mcp_user_token( + f"user-{i}", + server_name, + access_token_ct=b"ct", + refresh_token_ct=None, + expires_at=None, + scopes=None, + as_issuer="https://as.example.com", + audience="https://example.com/mcp", + ) + + +def test_requires_admin_mcp_permission(storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + client = TestClient(_build_app(storage, with_admin_mcp=False)) + resp = client.post("/v1/api/admin/mcp-servers/srv-oauth/bulk-revoke") + assert resp.status_code == 403 + + +def test_404_on_missing_server(storage: SQLiteBackend) -> None: + client = TestClient(_build_app(storage)) + resp = client.post("/v1/api/admin/mcp-servers/never-existed/bulk-revoke") + assert resp.status_code == 404 + assert resp.json() == {"error": "No such server"} + + +def test_400_on_static_server(storage: SQLiteBackend) -> None: + _seed_static_server(storage) + client = TestClient(_build_app(storage)) + resp = client.post("/v1/api/admin/mcp-servers/srv-static/bulk-revoke") + assert resp.status_code == 400 + body = resp.json() + assert "oauth_user" in body["error"] + + +def test_400_on_invalid_server_name(storage: SQLiteBackend) -> None: + # double-underscore is reserved for the prefixed-tool-name encoding. + client = TestClient(_build_app(storage)) + resp = client.post("/v1/api/admin/mcp-servers/bad__name/bulk-revoke") + assert resp.status_code == 400 + + +def test_200_on_success_with_no_consented_users(storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + client = TestClient(_build_app(storage)) + resp = client.post("/v1/api/admin/mcp-servers/srv-oauth/bulk-revoke") + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "ok" + assert body["rows_deleted"] == 0 + assert body["consented_users_before"] == 0 + + +def test_200_clears_all_user_tokens(storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_user_tokens(storage, "srv-oauth", users=3) + # Token for another server must survive the bulk-revoke. + _seed_oauth_server(storage, name="srv-other", server_id="srv-other-id") + _seed_user_tokens(storage, "srv-other", users=2) + + client = TestClient(_build_app(storage)) + resp = client.post("/v1/api/admin/mcp-servers/srv-oauth/bulk-revoke") + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "ok" + assert body["rows_deleted"] == 3 + assert body["consented_users_before"] == 3 + + # Target server's tokens are gone; bystander's tokens survive. + assert storage.count_mcp_consented_users_by_server("srv-oauth") == 0 + assert storage.count_mcp_consented_users_by_server("srv-other") == 2 + + +def test_audits_with_bulk_admin_no_upstream(storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_user_tokens(storage, "srv-oauth", users=2) + client = TestClient(_build_app(storage)) + resp = client.post("/v1/api/admin/mcp-servers/srv-oauth/bulk-revoke") + assert resp.status_code == 200 + + # Pull the most-recent audit row for the bulk_revoked action and + # verify it carries the deferral marker. + events = storage.list_audit_events(limit=10) + bulk_rows = [e for e in events if e.get("action") == "mcp_server.oauth.bulk_revoked"] + assert len(bulk_rows) == 1 + detail = bulk_rows[0].get("detail") + if isinstance(detail, str): + import json as _json + + detail = _json.loads(detail) + assert detail.get("upstream_revoke_outcome") == "bulk_admin_no_upstream" + assert detail.get("rows_deleted") == 2 + assert detail.get("consented_users_before") == 2 + assert detail.get("name") == "srv-oauth" diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py index 9b9cc85c..023910a5 100644 --- a/tests/test_mcp_client.py +++ b/tests/test_mcp_client.py @@ -788,7 +788,11 @@ class TestSessionIntegration: assert call_id == "call_789" assert output == "result text" mock_mcp.call_tool_sync.assert_called_once_with( - "mcp__test__search", {"query": "hello"}, user_id=None, timeout=30 + "mcp__test__search", + {"query": "hello"}, + user_id=None, + timeout=30, + is_interactive_for_consent=True, ) def test_exec_mcp_tool_error(self, tmp_db): @@ -1056,6 +1060,147 @@ class TestRefreshServer: asyncio.run(_run()) +class TestLastRefreshTracking: + """Phase 9 admin status pill — ``_last_refresh`` is written on every + refresh path so the admin UI reflects manual-refresh AND auto- + reconnect outcomes uniformly. This test class pins the contract. + """ + + @staticmethod + def _seed_minimal(mgr: MCPClientManager, name: str = "srv") -> MagicMock: + mock_session = MagicMock() + mock_session.list_tools = AsyncMock(return_value=MagicMock(tools=[])) + mock_session.list_resources = AsyncMock(return_value=MagicMock(resources=[])) + mock_session.list_resource_templates = AsyncMock( + return_value=MagicMock(resourceTemplates=[]) + ) + mock_session.list_prompts = AsyncMock(return_value=MagicMock(prompts=[])) + _seed_static_state( + mgr, + name, + session=mock_session, + tools=[], + supports_resources=True, + supports_prompts=True, + ) + return mock_session + + def test_last_refresh_written_on_success(self) -> None: + async def _run() -> None: + mgr = MCPClientManager({}) + self._seed_minimal(mgr) + assert "srv" not in mgr._last_refresh + + await mgr._refresh_server("srv") + + entry = mgr._last_refresh.get("srv") + assert entry is not None + ts, outcome = entry + assert outcome == "ok" + assert isinstance(ts, float) and ts > 0 + + asyncio.run(_run()) + + def test_last_refresh_written_on_tool_refresh_failure(self) -> None: + """When ``_refresh_server_tools`` raises, the outcome reflects + the exception class and the exception still propagates.""" + + async def _run() -> None: + mgr = MCPClientManager({}) + mock_session = self._seed_minimal(mgr) + mock_session.list_tools = AsyncMock(side_effect=RuntimeError("upstream down")) + + with pytest.raises(RuntimeError, match="upstream down"): + await mgr._refresh_server("srv") + + entry = mgr._last_refresh.get("srv") + assert entry is not None + _, outcome = entry + assert outcome == "error:RuntimeError" + + asyncio.run(_run()) + + def test_last_refresh_records_first_exception_when_multiple_fail( + self, + ) -> None: + """``return_exceptions=True`` lets sibling tasks complete; the + outcome reflects the FIRST exception encountered.""" + + async def _run() -> None: + mgr = MCPClientManager({}) + mock_session = self._seed_minimal(mgr) + # Tools succeeds; resources raises first (gather preserves + # argument order in its results list, so resources is the + # first failure regardless of which awaitable finished first + # in wall-clock terms). + mock_session.list_resources = AsyncMock(side_effect=ValueError("res boom")) + mock_session.list_prompts = AsyncMock(side_effect=KeyError("prompts boom")) + + with pytest.raises((ValueError, KeyError)): + await mgr._refresh_server("srv") + + entry = mgr._last_refresh.get("srv") + assert entry is not None + _, outcome = entry + # Either of the two failing tasks could be "first" in + # gather's results list ordering — the order is positional + # so resources (arg #2) comes before prompts (arg #3). + assert outcome == "error:ValueError" + + asyncio.run(_run()) + + def test_refresh_all_overwrites_stale_ok_on_reconnect_failure( + self, + ) -> None: + """The chokepoint bug-1 fix: a prior successful refresh's ``'ok'`` + entry MUST be overwritten when a subsequent reconnect fails — + otherwise the admin pill shows misleading "ok" while the server + is in fact broken.""" + + async def _run() -> None: + mgr = MCPClientManager({}) + # Server is configured but has no live session — _refresh_all + # routes to the reconnect branch. + mgr._server_configs["srv"] = {"type": "stdio", "command": "x"} + # Pre-seed a stale "ok" from an earlier successful refresh. + mgr._last_refresh["srv"] = (1000.0, "ok") + + async def _raise(*_a: object, **_kw: object) -> None: + raise ConnectionError("reconnect failed") + + mgr._connect_one = _raise # type: ignore[assignment] + + await mgr._refresh_all("srv") + + entry = mgr._last_refresh.get("srv") + assert entry is not None + ts, outcome = entry + # Outcome reflects the new failure, not the stale ok. + assert outcome == "error:ConnectionError" + assert ts > 1000.0 + + asyncio.run(_run()) + + def test_get_server_status_surfaces_last_refresh_fields(self) -> None: + """``get_server_status`` surfaces ``last_refresh_at`` and + ``last_refresh_outcome`` for the admin pill — null when no + refresh has occurred yet, populated after one.""" + mgr = MCPClientManager({}) + mgr._server_configs["srv"] = {"type": "stdio", "command": "x"} + + # No refresh yet — fields must be present and null so the JS + # renderer can branch on absence cleanly. + status = mgr.get_server_status("srv") + assert status["last_refresh_at"] is None + assert status["last_refresh_outcome"] is None + + # Populate the tuple directly and re-read. + mgr._last_refresh["srv"] = (12345.5, "ok") + status = mgr.get_server_status("srv") + assert status["last_refresh_at"] == 12345.5 + assert status["last_refresh_outcome"] == "ok" + + class TestListeners: def test_add_and_notify(self): mgr = MCPClientManager({}) diff --git a/tests/test_mcp_oauth_handlers.py b/tests/test_mcp_oauth_handlers.py index 4379183f..a1680a21 100644 --- a/tests/test_mcp_oauth_handlers.py +++ b/tests/test_mcp_oauth_handlers.py @@ -612,6 +612,97 @@ class TestCallback: assert plain is not None assert plain["refresh_token"] is None + def test_callback_clears_pending_consent_on_success( + self, storage: SQLiteBackend, http_client_mock: MagicMock + ) -> None: + """Successful callback must drop any ``mcp_pending_consent`` rows + for the just-consented ``(user, server)`` (Phase 9 lifecycle + contract). Regression guard for the dashboard-stays-stale-after- + consent invariant. + """ + _seed_oauth_user_server(storage) + self._seed_pending(storage) + # Seed a deferred-consent record that a prior non-interactive run + # would have left behind. Plus a cross-tenant record that must + # NOT be touched. + storage.upsert_mcp_pending_consent( + user_id="user-1", + server_name="srv-oauth", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id="ws-1", + last_tool_call_id="tool-1", + now_iso="2026-05-11T12:00:00", + ) + storage.upsert_mcp_pending_consent( + user_id="other-user", + server_name="srv-oauth", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso="2026-05-11T12:00:00", + ) + token_store = _make_token_store(storage) + http_client_mock.get.return_value = _mk_response(200, _good_as_metadata_doc()) + http_client_mock.post.return_value = _mk_response( + 200, + {"access_token": "opaque-aaa", "expires_in": 3600}, + ) + app = _build_app(storage=storage, http_client=http_client_mock, token_store=token_store) + client = TestClient(app, raise_server_exceptions=False) + + with _public_addr_patch(): + resp = client.get( + "/v1/api/mcp/oauth/callback?code=c&state=valid-state", + follow_redirects=False, + ) + assert resp.status_code == 302 + # Callback completed → user-1's deferred-consent row was cleared. + assert storage.list_mcp_pending_consent_by_user("user-1") == [] + # Cross-tenant row survives — clear is per-(user, server). + other = storage.list_mcp_pending_consent_by_user("other-user") + assert len(other) == 1 + assert other[0]["server_name"] == "srv-oauth" + + def test_callback_storage_failure_does_not_block_redirect( + self, storage: SQLiteBackend, http_client_mock: MagicMock + ) -> None: + """If the post-persist ``delete_mcp_pending_consent`` raises, the + callback's redirect still completes (best-effort contract). The + stale badge is preferred over a broken consent flow. + """ + _seed_oauth_user_server(storage) + self._seed_pending(storage) + token_store = _make_token_store(storage) + http_client_mock.get.return_value = _mk_response(200, _good_as_metadata_doc()) + http_client_mock.post.return_value = _mk_response( + 200, + {"access_token": "opaque-aaa", "expires_in": 3600}, + ) + app = _build_app(storage=storage, http_client=http_client_mock, token_store=token_store) + client = TestClient(app, raise_server_exceptions=False) + + original_delete = storage.delete_mcp_pending_consent + + def _raise(*_a: Any, **_kw: Any) -> bool: + raise RuntimeError("storage offline") + + storage.delete_mcp_pending_consent = _raise # type: ignore[method-assign] + try: + with _public_addr_patch(): + resp = client.get( + "/v1/api/mcp/oauth/callback?code=c&state=valid-state", + follow_redirects=False, + ) + finally: + storage.delete_mcp_pending_consent = original_delete # type: ignore[method-assign] + + assert resp.status_code == 302 + # Token persistence still succeeded — the user-visible contract. + plain = token_store.get_user_token("user-1", "srv-oauth") + assert plain is not None + # --------------------------------------------------------------------------- # 503 paths when mcp_token_store is None diff --git a/tests/test_mcp_pending_consent_dispatch.py b/tests/test_mcp_pending_consent_dispatch.py new file mode 100644 index 00000000..a4d68cfe --- /dev/null +++ b/tests/test_mcp_pending_consent_dispatch.py @@ -0,0 +1,304 @@ +"""Boundary tests for the Phase 9 pending-consent write path. + +Drives ``MCPClientManager._dispatch_pool_sync`` (and the helper it +calls, ``_record_pending_consent_best_effort``) and asserts that +deferred-consent records reach storage only on non-interactive callers. + +Per ``feedback_tests_through_boundaries.md``, at least one test must +drive the real sync dispatcher → real ``_is_structured_error`` → +real ``_record_pending_consent_best_effort`` plumb-through; the +``_helpers`` unit tests below cover the classifier in isolation, but +the end-to-end test is the structural gate that catches +plumb-through regressions. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import threading +from typing import Any +from unittest.mock import patch + +import pytest + +from tests.conftest import make_mcp_token_cipher +from turnstone.core.mcp_client import ( + _PENDING_CONSENT_PERSIST_CODES, + MCPClientManager, + _parse_pending_consent_envelope, +) +from turnstone.core.mcp_crypto import MCPTokenStore +from turnstone.core.mcp_oauth import TokenLookupResult + +# --------------------------------------------------------------------------- +# Helper-level unit tests (cheap, no event loop) +# --------------------------------------------------------------------------- + + +class TestParseEnvelope: + def test_consent_required_no_scopes(self) -> None: + env = json.dumps({"error": {"code": "mcp_consent_required", "server": "x", "detail": "d"}}) + assert _parse_pending_consent_envelope(env) == ("mcp_consent_required", None) + + def test_insufficient_scope_with_scopes(self) -> None: + env = json.dumps( + { + "error": { + "code": "mcp_insufficient_scope", + "server": "x", + "detail": "d", + "scopes_required": ["read", "write"], + } + } + ) + assert _parse_pending_consent_envelope(env) == ( + "mcp_insufficient_scope", + ["read", "write"], + ) + + def test_operator_codes_filtered(self) -> None: + # Key-unknown / url-insecure / *_forbidden are operator-actionable, + # NOT user-consent-shaped. They must not produce pending-consent + # rows, regardless of whether the caller is interactive. + for code in ( + "mcp_token_undecryptable_key_unknown", + "mcp_oauth_url_insecure", + "mcp_tool_call_forbidden", + "mcp_resource_read_forbidden", + "mcp_prompt_get_forbidden", + ): + env = json.dumps({"error": {"code": code, "server": "x", "detail": "d"}}) + assert _parse_pending_consent_envelope(env) is None, code + + def test_malformed_json_returns_none(self) -> None: + assert _parse_pending_consent_envelope("not json") is None + assert _parse_pending_consent_envelope("") is None + + def test_persist_codes_set_is_expected(self) -> None: + # Pin the contract — adding a new persistable code here is a + # deliberate design decision and should require a test update. + assert { + "mcp_consent_required", + "mcp_insufficient_scope", + } == _PENDING_CONSENT_PERSIST_CODES + + +# --------------------------------------------------------------------------- +# End-to-end plumb-through (drives _dispatch_pool_sync) +# --------------------------------------------------------------------------- + + +def _seed_oauth_server(backend: Any, *, name: str = "pool-srv") -> None: + backend.create_mcp_server( + server_id="srv-" + name, + name=name, + transport="streamable-http", + command="", + args="[]", + url="https://example.com/mcp", + headers="{}", + env="{}", + auto_approve=False, + enabled=True, + created_by="admin", + ) + backend.update_mcp_server("srv-" + name, auth_type="oauth_user") + + +@pytest.fixture +def running_loop_mgr(): + cfg: dict[str, Any] = {} + mgr = MCPClientManager(cfg) + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever, daemon=True, name="phase9-test-loop") + thread.start() + mgr._loop = loop + try: + yield mgr, loop, thread + finally: + + async def _drain(m: MCPClientManager) -> None: + task = m._user_pool_eviction_task + if task is not None: + task.cancel() + with contextlib.suppress(BaseException): + await task + m._user_pool_eviction_task = None + + with contextlib.suppress(Exception): + asyncio.run_coroutine_threadsafe(_drain(mgr), loop).result(timeout=2) + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=2) + + +def _wire_mgr(mgr: MCPClientManager, backend: Any) -> None: + cipher = make_mcp_token_cipher() + from types import SimpleNamespace + from unittest.mock import MagicMock + + app_state = SimpleNamespace( + auth_storage=backend, + mcp_token_store=MCPTokenStore(backend, cipher, node_id="test"), + mcp_oauth_http_client=MagicMock(), + mcp_oauth_refresh_locks={}, + mcp_oauth_metadata_cache={}, + ) + mgr.set_storage(backend) + mgr.set_app_state(app_state) + + +def test_dispatch_persists_pending_for_non_interactive_caller( + running_loop_mgr: Any, backend: Any +) -> None: + """Non-interactive caller hits ``mcp_consent_required`` → a + ``mcp_pending_consent`` row appears for ``(user_id, server_name)``.""" + mgr, _loop, _ = running_loop_mgr + _seed_oauth_server(backend) + _wire_mgr(mgr, backend) + + async def _missing_token(**kwargs: Any) -> TokenLookupResult: + return TokenLookupResult(kind="missing") + + with ( + patch( + "turnstone.core.mcp_client.get_user_access_token_classified", + side_effect=_missing_token, + ), + pytest.raises(RuntimeError) as exc_info, + ): + mgr.call_tool_sync( + "mcp__pool-srv__echo", + {"payload": "hi"}, + user_id="user-a", + timeout=10, + is_interactive_for_consent=False, + ) + + # Structured error envelope surfaces as RuntimeError to the caller. + payload = json.loads(str(exc_info.value)).get("error", {}) + assert payload.get("code") == "mcp_consent_required" + + # Persistent row written for the dashboard badge. + rows = backend.list_mcp_pending_consent_by_user("user-a") + assert len(rows) == 1 + r = rows[0] + assert r["user_id"] == "user-a" + assert r["server_name"] == "pool-srv" + assert r["error_code"] == "mcp_consent_required" + assert r["occurrence_count"] == 1 + + +def test_dispatch_does_not_persist_for_interactive_caller( + running_loop_mgr: Any, backend: Any +) -> None: + """Interactive caller hits the same error path → NO row written. + + Interactive (WEB / CLI) sessions surface the consent prompt in-flight + via the Phase 8 SSE renderer; persisting would just produce + immediately-stale dashboard badges. + """ + mgr, _loop, _ = running_loop_mgr + _seed_oauth_server(backend) + _wire_mgr(mgr, backend) + + async def _missing_token(**kwargs: Any) -> TokenLookupResult: + return TokenLookupResult(kind="missing") + + with ( + patch( + "turnstone.core.mcp_client.get_user_access_token_classified", + side_effect=_missing_token, + ), + pytest.raises(RuntimeError), + ): + mgr.call_tool_sync( + "mcp__pool-srv__echo", + {"payload": "hi"}, + user_id="user-a", + timeout=10, + is_interactive_for_consent=True, + ) + + assert backend.list_mcp_pending_consent_by_user("user-a") == [] + + +def test_dispatch_returns_envelope_unchanged_on_storage_failure( + running_loop_mgr: Any, backend: Any +) -> None: + """When ``upsert_mcp_pending_consent`` raises, the agent-observable + contract is unchanged: the structured-error ``RuntimeError`` still + surfaces with the original ``mcp_consent_required`` code. The doc- + string promises best-effort persistence; this test pins that + promise so a regression that propagates the storage exception would + fail visibly. + """ + mgr, _loop, _ = running_loop_mgr + _seed_oauth_server(backend) + _wire_mgr(mgr, backend) + + async def _missing_token(**kwargs: Any) -> TokenLookupResult: + return TokenLookupResult(kind="missing") + + original_upsert = backend.upsert_mcp_pending_consent + + def _raise(*_a: Any, **_kw: Any) -> None: + raise RuntimeError("storage offline") + + backend.upsert_mcp_pending_consent = _raise # type: ignore[method-assign] + try: + with ( + patch( + "turnstone.core.mcp_client.get_user_access_token_classified", + side_effect=_missing_token, + ), + pytest.raises(RuntimeError) as exc_info, + ): + mgr.call_tool_sync( + "mcp__pool-srv__echo", + {"payload": "hi"}, + user_id="user-a", + timeout=10, + is_interactive_for_consent=False, + ) + finally: + backend.upsert_mcp_pending_consent = original_upsert # type: ignore[method-assign] + + payload = json.loads(str(exc_info.value)).get("error", {}) + assert payload.get("code") == "mcp_consent_required" + + +def test_dispatch_does_not_persist_for_operator_actionable_code( + running_loop_mgr: Any, backend: Any +) -> None: + """Decrypt-failure → operator-actionable; even non-interactive callers + must NOT produce a user-facing pending-consent record (the user can't + resolve this by re-consenting). + """ + mgr, _loop, _ = running_loop_mgr + _seed_oauth_server(backend) + _wire_mgr(mgr, backend) + + async def _decrypt_failure(**kwargs: Any) -> TokenLookupResult: + return TokenLookupResult(kind="decrypt_failure") + + with ( + patch( + "turnstone.core.mcp_client.get_user_access_token_classified", + side_effect=_decrypt_failure, + ), + pytest.raises(RuntimeError) as exc_info, + ): + mgr.call_tool_sync( + "mcp__pool-srv__echo", + {"payload": "hi"}, + user_id="user-a", + timeout=10, + is_interactive_for_consent=False, + ) + + payload = json.loads(str(exc_info.value)).get("error", {}) + assert payload.get("code") == "mcp_token_undecryptable_key_unknown" + # The operator-actionable code does NOT produce a pending-consent row. + assert backend.list_mcp_pending_consent_by_user("user-a") == [] diff --git a/tests/test_mcp_pending_consent_endpoints.py b/tests/test_mcp_pending_consent_endpoints.py new file mode 100644 index 00000000..3c002558 --- /dev/null +++ b/tests/test_mcp_pending_consent_endpoints.py @@ -0,0 +1,259 @@ +"""HTTP tests for the Phase 9 pending-consent endpoints. + +Covers: +- ``GET /v1/api/mcp/oauth/pending`` (install gate + read path) +- ``DELETE /v1/api/mcp/oauth/pending/{server_name}`` (single clear) +- ``DELETE /v1/api/mcp/oauth/pending`` (bulk clear) +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest +from starlette.applications import Starlette +from starlette.middleware import Middleware +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.routing import Mount, Route +from starlette.testclient import TestClient + +from turnstone.core.auth import AuthResult +from turnstone.core.mcp_oauth import ( + handle_mcp_oauth_clear_all_pending, + handle_mcp_oauth_clear_pending, + handle_mcp_oauth_list_pending, +) +from turnstone.core.storage._sqlite import SQLiteBackend + +if TYPE_CHECKING: + from starlette.requests import Request + from starlette.responses import Response + + +class _InjectAuthMiddleware(BaseHTTPMiddleware): + """Stamp a fixed authenticated user on every request.""" + + def __init__(self, app: Any, user_id: str = "user-1") -> None: + super().__init__(app) + self._user_id = user_id + + async def dispatch(self, request: Request, call_next: Any) -> Response: + request.state.auth_result = AuthResult( + user_id=self._user_id, + scopes=frozenset({"write"}), + token_source="config", + permissions=frozenset({"read", "write"}), + ) + return await call_next(request) + + +def _build_app(storage: SQLiteBackend, *, user_id: str = "user-1") -> Starlette: + class _Mw(_InjectAuthMiddleware): + def __init__(self, app: Any) -> None: + super().__init__(app, user_id=user_id) + + app = Starlette( + routes=[ + Mount( + "/v1", + routes=[ + Route("/api/mcp/oauth/pending", handle_mcp_oauth_list_pending), + Route( + "/api/mcp/oauth/pending", + handle_mcp_oauth_clear_all_pending, + methods=["DELETE"], + ), + Route( + "/api/mcp/oauth/pending/{server_name}", + handle_mcp_oauth_clear_pending, + methods=["DELETE"], + ), + ], + ), + ], + middleware=[Middleware(_Mw)], + ) + app.state.auth_storage = storage + return app + + +@pytest.fixture +def storage(tmp_path: Any) -> SQLiteBackend: + backend = SQLiteBackend(str(tmp_path / "test.db")) + backend.create_user("user-1", "user1", "User One", "hash") + backend.create_user("user-2", "user2", "User Two", "hash") + return backend + + +def _seed_oauth_server(backend: SQLiteBackend, *, name: str = "srv-x") -> None: + backend.create_mcp_server( + server_id="srv-id-" + name, + name=name, + transport="streamable-http", + url="https://example.com/mcp", + auth_type="oauth_user", + ) + + +def _seed_pending( + backend: SQLiteBackend, + *, + user_id: str = "user-1", + server_name: str = "srv-x", + error_code: str = "mcp_consent_required", + now_iso: str = "2026-05-11T12:00:00", +) -> None: + backend.upsert_mcp_pending_consent( + user_id=user_id, + server_name=server_name, + error_code=error_code, + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso=now_iso, + ) + + +class TestListPending: + def test_install_gate_short_circuits_on_no_oauth_servers(self, storage: SQLiteBackend) -> None: + # Seed a pending row but NO oauth_user MCP server — the gate + # must short-circuit to {pending: 0} regardless. + _seed_pending(storage) + client = TestClient(_build_app(storage)) + resp = client.get("/v1/api/mcp/oauth/pending") + assert resp.status_code == 200 + assert resp.json() == {"pending": 0, "servers": []} + + def test_lists_pending_records_for_authenticated_user(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_pending(storage) + client = TestClient(_build_app(storage)) + resp = client.get("/v1/api/mcp/oauth/pending") + assert resp.status_code == 200 + body = resp.json() + assert body["pending"] == 1 + assert len(body["servers"]) == 1 + assert body["servers"][0]["server_name"] == "srv-x" + assert body["servers"][0]["error_code"] == "mcp_consent_required" + + def test_does_not_leak_cross_user_records(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_pending(storage, user_id="user-2") + client = TestClient(_build_app(storage, user_id="user-1")) + resp = client.get("/v1/api/mcp/oauth/pending") + assert resp.status_code == 200 + assert resp.json() == {"pending": 0, "servers": []} + + +class TestClearPending: + def test_delete_single(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_pending(storage) + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending/srv-x") + assert resp.status_code == 204 + assert storage.list_mcp_pending_consent_by_user("user-1") == [] + + def test_delete_missing_still_returns_204(self, storage: SQLiteBackend) -> None: + # Idempotent — must not leak cross-user existence info via 404. + _seed_oauth_server(storage) + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending/never-existed") + assert resp.status_code == 204 + + def test_delete_does_not_touch_cross_user_rows(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_pending(storage, user_id="user-1") + _seed_pending(storage, user_id="user-2") + client = TestClient(_build_app(storage, user_id="user-1")) + resp = client.delete("/v1/api/mcp/oauth/pending/srv-x") + assert resp.status_code == 204 + # User-2's row survives. + assert len(storage.list_mcp_pending_consent_by_user("user-2")) == 1 + + +class TestAuditTrail: + def test_single_dismiss_audits(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_pending(storage) + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending/srv-x") + assert resp.status_code == 204 + + events = storage.list_audit_events(limit=10) + rows = [ + e for e in events if e.get("action") == "mcp_server.oauth.pending_consent_dismissed" + ] + assert len(rows) == 1 + detail = rows[0].get("detail") + if isinstance(detail, str): + import json as _json + + detail = _json.loads(detail) + assert detail.get("mode") == "single" + assert detail.get("cleared") == 1 + + def test_single_dismiss_audits_even_when_no_row_existed(self, storage: SQLiteBackend) -> None: + # Cross-tenant non-observability requires a 204 in the never-existed + # case — the audit row distinguishes a real dismiss from a stuffed + # attempt by recording ``cleared=0``. + _seed_oauth_server(storage) + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending/never-existed") + assert resp.status_code == 204 + + events = storage.list_audit_events(limit=10) + rows = [ + e for e in events if e.get("action") == "mcp_server.oauth.pending_consent_dismissed" + ] + assert len(rows) == 1 + detail = rows[0].get("detail") + if isinstance(detail, str): + import json as _json + + detail = _json.loads(detail) + assert detail.get("mode") == "single" + assert detail.get("cleared") == 0 + + def test_bulk_dismiss_audits(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_oauth_server(storage, name="srv-y") + _seed_pending(storage, server_name="srv-x") + _seed_pending(storage, server_name="srv-y") + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending") + assert resp.status_code == 200 + assert resp.json() == {"cleared": 2} + + events = storage.list_audit_events(limit=10) + rows = [ + e for e in events if e.get("action") == "mcp_server.oauth.pending_consent_dismissed" + ] + assert len(rows) == 1 + detail = rows[0].get("detail") + if isinstance(detail, str): + import json as _json + + detail = _json.loads(detail) + assert detail.get("mode") == "bulk" + assert detail.get("cleared") == 2 + + +class TestClearAllPending: + def test_bulk_clear(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + _seed_oauth_server(storage, name="srv-y") + _seed_pending(storage, server_name="srv-x") + _seed_pending(storage, server_name="srv-y") + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending") + assert resp.status_code == 200 + assert resp.json() == {"cleared": 2} + assert storage.list_mcp_pending_consent_by_user("user-1") == [] + + def test_bulk_clear_zero_when_empty(self, storage: SQLiteBackend) -> None: + _seed_oauth_server(storage) + client = TestClient(_build_app(storage)) + resp = client.delete("/v1/api/mcp/oauth/pending") + assert resp.status_code == 200 + assert resp.json() == {"cleared": 0} diff --git a/tests/test_mcp_pending_consent_storage.py b/tests/test_mcp_pending_consent_storage.py new file mode 100644 index 00000000..f225674c --- /dev/null +++ b/tests/test_mcp_pending_consent_storage.py @@ -0,0 +1,263 @@ +"""Storage CRUD tests for the Phase 9 ``mcp_pending_consent`` table. + +Validates protocol additions backing the dashboard pending-consent badge: + +- ``upsert_mcp_pending_consent`` — insert + on-conflict refresh +- ``list_mcp_pending_consent_by_user`` — read path +- ``delete_mcp_pending_consent`` — single-row clear +- ``delete_all_mcp_pending_consent_by_user`` — bulk clear +- ``count_mcp_consented_users_by_server`` — admin status pill +- ``any_oauth_user_mcp_servers`` — install-level gate +""" + +from __future__ import annotations + + +def _iso(ts: str = "2026-05-11T12:00:00") -> str: + return ts + + +class TestUpsertAndList: + def test_insert_round_trip(self, backend) -> None: + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv-x", + error_code="mcp_consent_required", + scopes_required="read write", + last_ws_id="ws-1", + last_tool_call_id="tool-1", + now_iso=_iso(), + ) + rows = backend.list_mcp_pending_consent_by_user("user-a") + assert len(rows) == 1 + r = rows[0] + assert r["user_id"] == "user-a" + assert r["server_name"] == "srv-x" + assert r["error_code"] == "mcp_consent_required" + assert r["scopes_required"] == "read write" + assert r["last_ws_id"] == "ws-1" + assert r["last_tool_call_id"] == "tool-1" + assert r["occurrence_count"] == 1 + assert r["first_seen_at"] == r["last_seen_at"] + + def test_upsert_bumps_count_and_refreshes_recency(self, backend) -> None: + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv-x", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso="2026-05-11T12:00:00", + ) + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv-x", + error_code="mcp_insufficient_scope", + scopes_required="read", + last_ws_id="ws-2", + last_tool_call_id="tool-2", + now_iso="2026-05-11T13:00:00", + ) + rows = backend.list_mcp_pending_consent_by_user("user-a") + assert len(rows) == 1 + r = rows[0] + # Recency fields refreshed to the second call's values; count bumped. + assert r["occurrence_count"] == 2 + assert r["error_code"] == "mcp_insufficient_scope" + assert r["scopes_required"] == "read" + assert r["last_ws_id"] == "ws-2" + assert r["last_tool_call_id"] == "tool-2" + assert r["last_seen_at"] == "2026-05-11T13:00:00" + # first_seen_at preserved — that's the load-bearing audit value. + assert r["first_seen_at"] == "2026-05-11T12:00:00" + + def test_list_orders_by_last_seen_desc(self, backend) -> None: + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv-old", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso="2026-05-11T10:00:00", + ) + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv-new", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso="2026-05-11T11:00:00", + ) + rows = backend.list_mcp_pending_consent_by_user("user-a") + assert [r["server_name"] for r in rows] == ["srv-new", "srv-old"] + + def test_per_user_isolation(self, backend) -> None: + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso=_iso(), + ) + assert backend.list_mcp_pending_consent_by_user("user-b") == [] + + +class TestDelete: + def test_delete_single(self, backend) -> None: + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name="srv-x", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso=_iso(), + ) + assert backend.delete_mcp_pending_consent("user-a", "srv-x") is True + assert backend.list_mcp_pending_consent_by_user("user-a") == [] + # Second delete returns False (no row). + assert backend.delete_mcp_pending_consent("user-a", "srv-x") is False + + def test_delete_missing_returns_false(self, backend) -> None: + assert backend.delete_mcp_pending_consent("never", "missing") is False + + def test_delete_all_by_user(self, backend) -> None: + for name in ("srv-a", "srv-b", "srv-c"): + backend.upsert_mcp_pending_consent( + user_id="user-a", + server_name=name, + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso=_iso(), + ) + # Cross-user row that must NOT be touched. + backend.upsert_mcp_pending_consent( + user_id="user-b", + server_name="srv-z", + error_code="mcp_consent_required", + scopes_required=None, + last_ws_id=None, + last_tool_call_id=None, + now_iso=_iso(), + ) + assert backend.delete_all_mcp_pending_consent_by_user("user-a") == 3 + assert backend.list_mcp_pending_consent_by_user("user-a") == [] + assert len(backend.list_mcp_pending_consent_by_user("user-b")) == 1 + + +class TestCountConsentedUsersByServer: + def _seed_server(self, backend, name: str = "srv-x") -> None: + backend.create_mcp_server( + server_id="srv-id-" + name, + name=name, + transport="streamable-http", + command="", + args="[]", + url="https://example.com/mcp", + headers="{}", + env="{}", + auto_approve=False, + enabled=True, + created_by="admin", + ) + backend.update_mcp_server("srv-id-" + name, auth_type="oauth_user") + + def test_counts_distinct_non_expired_users(self, backend) -> None: + self._seed_server(backend) + future = "2099-01-01T00:00:00" + backend.create_mcp_user_token( + "alice", + "srv-x", + access_token_ct=b"ct", + refresh_token_ct=None, + expires_at=future, + scopes=None, + as_issuer="https://as.example.com", + audience="https://example.com/mcp", + ) + backend.create_mcp_user_token( + "bob", + "srv-x", + access_token_ct=b"ct", + refresh_token_ct=None, + expires_at=None, # null treated as non-expired + scopes=None, + as_issuer="https://as.example.com", + audience="https://example.com/mcp", + ) + # Different server — must not count. + self._seed_server(backend, name="srv-y") + backend.create_mcp_user_token( + "carol", + "srv-y", + access_token_ct=b"ct", + refresh_token_ct=None, + expires_at=future, + scopes=None, + as_issuer="https://as.example.com", + audience="https://example.com/mcp", + ) + assert backend.count_mcp_consented_users_by_server("srv-x") == 2 + assert backend.count_mcp_consented_users_by_server("srv-y") == 1 + + def test_excludes_expired(self, backend) -> None: + self._seed_server(backend) + backend.create_mcp_user_token( + "alice", + "srv-x", + access_token_ct=b"ct", + refresh_token_ct=None, + expires_at="2020-01-01T00:00:00", # well in the past + scopes=None, + as_issuer="https://as.example.com", + audience="https://example.com/mcp", + ) + assert backend.count_mcp_consented_users_by_server("srv-x") == 0 + + def test_zero_when_no_rows(self, backend) -> None: + assert backend.count_mcp_consented_users_by_server("missing") == 0 + + +class TestInstallGate: + def test_any_oauth_user_returns_false_on_empty(self, backend) -> None: + assert backend.any_oauth_user_mcp_servers() is False + + def test_any_oauth_user_ignores_static_rows(self, backend) -> None: + backend.create_mcp_server( + server_id="srv-1", + name="static-only", + transport="streamable-http", + command="", + args="[]", + url="https://example.com", + headers='{"Authorization": "Bearer x"}', + env="{}", + auto_approve=False, + enabled=True, + created_by="admin", + ) + assert backend.any_oauth_user_mcp_servers() is False + + def test_any_oauth_user_returns_true_when_one_exists(self, backend) -> None: + backend.create_mcp_server( + server_id="srv-2", + name="oauth-srv", + transport="streamable-http", + command="", + args="[]", + url="https://example.com", + headers="{}", + env="{}", + auto_approve=False, + enabled=True, + created_by="admin", + ) + backend.update_mcp_server("srv-2", auth_type="oauth_user") + assert backend.any_oauth_user_mcp_servers() is True diff --git a/turnstone/console/scheduler.py b/turnstone/console/scheduler.py index 12392614..7767dacb 100644 --- a/turnstone/console/scheduler.py +++ b/turnstone/console/scheduler.py @@ -319,6 +319,12 @@ class TaskScheduler: user_id=task.get("created_by", ""), skill=task.get("skill", ""), notify_targets=task.get("notify_targets", "[]"), + # Mark the resulting ChatSession as non-interactive-for- + # consent so OAuth-MCP errors get persisted to + # ``mcp_pending_consent`` for later dashboard surfacing, + # rather than relying on an in-flight SSE redirect the + # absent user can't complete. + client_type="scheduled", ) ws_id = resp.ws_id except Exception: diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 0875b43a..4a12b07d 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -1652,6 +1652,27 @@ async def mcp_oauth_revoke_connection(request: Request) -> Response: return await handle_mcp_oauth_revoke_connection(request) +async def mcp_oauth_list_pending(request: Request) -> Response: + """GET /v1/api/mcp/oauth/pending — list deferred-consent records (Phase 9).""" + from turnstone.core.mcp_oauth import handle_mcp_oauth_list_pending + + return await handle_mcp_oauth_list_pending(request) + + +async def mcp_oauth_clear_pending(request: Request) -> Response: + """DELETE /v1/api/mcp/oauth/pending/{server_name} — dismiss a deferred-consent record.""" + from turnstone.core.mcp_oauth import handle_mcp_oauth_clear_pending + + return await handle_mcp_oauth_clear_pending(request) + + +async def mcp_oauth_clear_all_pending(request: Request) -> Response: + """DELETE /v1/api/mcp/oauth/pending — bulk-dismiss deferred-consent records.""" + from turnstone.core.mcp_oauth import handle_mcp_oauth_clear_all_pending + + return await handle_mcp_oauth_clear_all_pending(request) + + # --------------------------------------------------------------------------- # Route handlers — available models (lightweight, no admin permission) # --------------------------------------------------------------------------- @@ -8818,10 +8839,19 @@ def _mask_mcp_secrets(server: dict[str, Any], reveal: bool = False) -> dict[str, def _mcp_server_to_detail( server: dict[str, Any], node_statuses: dict[str, dict[str, Any]] | None = None, + consented_users_count: int | None = None, ) -> dict[str, Any]: - """Convert a storage dict to a McpServerDetail-shaped dict.""" + """Convert a storage dict to a McpServerDetail-shaped dict. + + *consented_users_count* is the Phase 9 admin pill data — distinct + non-expired tokens issued for this ``(server_name)``. Omitted + (``None``) when the row's ``auth_type`` is not ``oauth_user``, so + static / none rows don't carry an irrelevant ``0``. + """ d = dict(server) d["status"] = node_statuses or {} + if consented_users_count is not None: + d["consented_users_count"] = consented_users_count return d @@ -8872,8 +8902,31 @@ async def admin_list_mcp_servers(request: Request) -> JSONResponse: reveal = str(request.query_params.get("reveal", "")).lower() in ("true", "1") servers = storage.list_mcp_servers() - # Collect live status from all nodes - node_statuses = await _collect_mcp_status(request) + # Phase 9: bulk-aggregate consented-users-count across all oauth_user + # rows in a single GROUP BY query (rather than N per-row sync DB + # round-trips inside this async handler). Run in parallel with the + # cross-node HTTP status fan-out below — neither has a data + # dependency on the other, so awaiting them sequentially would + # stack the DB latency on top of the fan-out latency. Skipped + # entirely when no row is oauth_user so static-only installs + # exercise zero new storage queries. + has_oauth_user = any(s.get("auth_type") == "oauth_user" for s in servers) + status_task: asyncio.Task[dict[str, dict[str, dict[str, Any]]]] = asyncio.create_task( + _collect_mcp_status(request) + ) + count_task: asyncio.Task[dict[str, int]] | None = ( + asyncio.create_task(asyncio.to_thread(storage.count_mcp_consented_users_grouped_by_server)) + if has_oauth_user + else None + ) + + node_statuses = await status_task + consent_counts: dict[str, int] = {} + if count_task is not None: + try: + consent_counts = await count_task + except Exception: + log.debug("admin.mcp_consented_users_bulk_count_failed", exc_info=True) db_names: set[str] = set() result = [] @@ -8885,8 +8938,14 @@ async def admin_list_mcp_servers(request: Request) -> JSONResponse: status = node_servers.get(s["name"]) if status: per_node[node_id] = status + # Phase 9: surface the consented-users-count pill for + # oauth_user rows. Aggregate was pre-computed above with a + # single bulk GROUP BY query; we just look up here. + consent_count: int | None = None + if s.get("auth_type") == "oauth_user": + consent_count = consent_counts.get(s["name"], 0) s = _mask_mcp_secrets(s, reveal) - result.append(_mcp_server_to_detail(s, per_node)) + result.append(_mcp_server_to_detail(s, per_node, consent_count)) # Merge config-sourced servers visible on nodes but not in DB config_names: set[str] = set() @@ -9527,6 +9586,92 @@ async def admin_mcp_reconnect_one(request: Request) -> JSONResponse: return await _admin_mcp_action(request, "reconnect") +async def admin_mcp_bulk_revoke(request: Request) -> JSONResponse: + """POST /v1/api/admin/mcp-servers/{name}/bulk-revoke — clear every user's token (Phase 9). + + Admin-side counterpart to the per-user + ``DELETE /v1/api/mcp/oauth/connections/{server_name}`` revoke that + shipped in Phase 8. Used to drop orphaned tokens after an + ``auth_type`` transition (oauth_user → static) or after rotating + the configured OAuth client. + + Authoritative local delete via + :meth:`StorageBackend.delete_mcp_oauth_rows_by_server_name` — + purges both ``mcp_user_tokens`` and ``mcp_oauth_pending`` rows for + the named server. Upstream RFC 7009 revoke is intentionally NOT + attempted in bulk (would require per-row decrypt + N upstream HTTP + calls); operators who need upstream cleanup should use the per- + user revoke endpoint or let tokens expire naturally. The audit + detail records ``upstream_revoke_outcome="bulk_admin_no_upstream"`` + so the deferral is visible. + + Pool eviction is NOT performed by this handler. Per-user revoke + has a per-(user, server) eviction primitive + (``MCPClientManager.evict_user_session``); bulk-revoke would need a + per-server iteration over consented users that no current primitive + supports. Stale in-flight sessions surface as a per-user 401 on + the next dispatch, which refreshes through the now-empty token row + and emits ``mcp_consent_required`` — the documented v1 fallback. + See :func:`turnstone.core.mcp_client._dispatch_pool` retry path. + """ + 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.get("name", "").strip() + if not name or "__" in name: + return JSONResponse({"error": "invalid server name"}, status_code=400) + + existing = storage.get_mcp_server_by_name(name) + if existing is None: + return JSONResponse({"error": "No such server"}, status_code=404) + if existing.get("auth_type") != "oauth_user": + return JSONResponse( + {"error": "bulk-revoke is only valid for auth_type=oauth_user servers"}, + status_code=400, + ) + + target_id = existing.get("server_id", name) + consented_before = 0 + try: + consented_before = storage.count_mcp_consented_users_by_server(name) + except Exception: + log.debug("admin.mcp_bulk_revoke_pre_count_failed server=%s", name, exc_info=True) + + deleted = storage.delete_mcp_oauth_rows_by_server_name(name) + + audit_uid, ip = _audit_context(request) + record_audit( + storage, + audit_uid, + "mcp_server.oauth.bulk_revoked", + "mcp_server", + target_id, + { + "name": name, + "rows_deleted": deleted, + "consented_users_before": consented_before, + "upstream_revoke_outcome": "bulk_admin_no_upstream", + }, + ip, + ) + + return JSONResponse( + { + "status": "ok", + "rows_deleted": deleted, + "consented_users_before": consented_before, + } + ) + + async def admin_import_mcp_config(request: Request) -> JSONResponse: """POST /v1/api/admin/mcp-servers/import — import from pasted JSON config.""" import uuid @@ -12247,6 +12392,17 @@ def create_app( mcp_oauth_revoke_connection, methods=["DELETE"], ), + Route("/api/mcp/oauth/pending", mcp_oauth_list_pending), + Route( + "/api/mcp/oauth/pending", + mcp_oauth_clear_all_pending, + methods=["DELETE"], + ), + Route( + "/api/mcp/oauth/pending/{server_name}", + mcp_oauth_clear_pending, + methods=["DELETE"], + ), Route("/api/admin/users", admin_list_users), Route("/api/admin/users", admin_create_user, methods=["POST"]), Route("/api/admin/users/{user_id}", admin_delete_user, methods=["DELETE"]), @@ -12432,6 +12588,11 @@ def create_app( admin_mcp_reconnect_one, methods=["POST"], ), + Route( + "/api/admin/mcp-servers/{name}/bulk-revoke", + admin_mcp_bulk_revoke, + 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 9bc6562a..7c0bb667 100644 --- a/turnstone/console/static/admin.js +++ b/turnstone/console/static/admin.js @@ -3314,6 +3314,11 @@ function _renderMcpServers(items) { var totalTools = 0, totalRes = 0, totalPrompts = 0; + // Phase 9: aggregate the most-recent refresh entry across nodes + // so the admin pill reflects "the freshest known state" rather + // than picking an arbitrary node. + var newestRefreshAt = null; + var newestRefreshOutcome = null; for (var j = 0; j < nodeIds.length; j++) { var ns = statusEntries[nodeIds[j]]; if (ns.connected) { @@ -3326,6 +3331,13 @@ function _renderMcpServers(items) { anyError = true; if (!firstError) firstError = ns.error; } + if ( + typeof ns.last_refresh_at === "number" && + (newestRefreshAt === null || ns.last_refresh_at > newestRefreshAt) + ) { + newestRefreshAt = ns.last_refresh_at; + newestRefreshOutcome = ns.last_refresh_outcome || null; + } } var dotClass = "mcp-status-dot disabled"; @@ -3351,6 +3363,38 @@ function _renderMcpServers(items) { statusText = "idle"; } + // Phase 9: refresh pill shows the short-relative age (e.g. "12m") with + // outcome-tinted color (ok vs err) and the full ISO timestamp + outcome + // in the tooltip. Pill is omitted (and the cell stays unchanged from + // its pre-Phase-9 shape) when no node has yet recorded a refresh + // outcome for this server. + var refreshPill = ""; + if (newestRefreshAt !== null) { + var ageSeconds = Math.max(0, Math.floor(Date.now() / 1000 - newestRefreshAt)); + var ageShort; + if (ageSeconds < 60) ageShort = ageSeconds + "s"; + else if (ageSeconds < 3600) ageShort = Math.floor(ageSeconds / 60) + "m"; + else if (ageSeconds < 86400) ageShort = Math.floor(ageSeconds / 3600) + "h"; + else ageShort = Math.floor(ageSeconds / 86400) + "d"; + var outcomeText = newestRefreshOutcome || "unknown"; + var pillCls = + outcomeText === "ok" ? "mcp-refresh-pill-ok" : "mcp-refresh-pill-err"; + var pillTitle = + "Last refresh " + + new Date(newestRefreshAt * 1000).toISOString() + + " (" + + outcomeText + + ")"; + refreshPill = + ' ' + + escapeHtml(ageShort) + + ""; + } + var transportCls = s.transport === "stdio" ? "mcp-transport-stdio" : "mcp-transport-http"; var toolsVal = anyConnected @@ -3385,6 +3429,24 @@ function _renderMcpServers(items) { ''; + // Phase 9: surface the consented-users count + bulk-revoke + // affordance only when at least one user has consented. + var consentCount = + typeof s.consented_users_count === "number" + ? s.consented_users_count + : 0; + if (consentCount > 0) { + actionBtns += + '"; + } } var actions = isConfig ? actionBtns @@ -3429,6 +3491,7 @@ function _renderMcpServers(items) { dotClass + '" aria-hidden="true">' + escapeHtml(statusText) + + refreshPill + "" + '' + actions + @@ -3507,6 +3570,42 @@ function _renderMcpServers(items) { window.open(url, "_blank", "noopener"); }); }); + el.querySelectorAll("[data-mcp-bulk-revoke]").forEach(function (btn) { + btn.addEventListener("click", function () { + var name = this.getAttribute("data-mcp-bulk-revoke"); + var count = this.getAttribute("data-mcp-consent-count") || "?"; + showConfirmModal( + "Bulk-revoke MCP consents", + "Drop all " + + count + + ' user consents for server "' + + name + + '"? Users will need to re-consent on next use. Upstream revoke is not attempted in bulk; tokens at the authorization server will expire naturally.', + "Bulk-revoke", + function () { + authFetch( + "/v1/api/admin/mcp-servers/" + + encodeURIComponent(name) + + "/bulk-revoke", + { method: "POST" }, + ) + .then(function (r) { + if (!r.ok) throw new Error(); + return r.json(); + }) + .then(function (j) { + showToast( + "Bulk-revoked " + (j.rows_deleted || 0) + " row(s) for " + name, + ); + loadAdminMcp(); + }) + .catch(function () { + showToast("Failed to bulk-revoke " + name); + }); + }, + ); + }); + }); el.querySelectorAll("[data-mcp-delete]").forEach(function (btn) { btn.addEventListener("click", function () { var sid = this.getAttribute("data-mcp-delete"); diff --git a/turnstone/console/static/style.css b/turnstone/console/static/style.css index 7e0a6dcf..2b072f7e 100644 --- a/turnstone/console/static/style.css +++ b/turnstone/console/static/style.css @@ -3602,6 +3602,28 @@ textarea.skill-content-area { opacity: 0.4; } +/* Phase 9 — last-refresh pill in the MCP status cell. Compact age + + outcome indicator inline with the existing status text. Uses the + semantic theme tokens (--bg-highlight, --fg-dim, --warn) defined in + shared_static/base.css so the pill follows dark/light theme swaps. */ +.mcp-refresh-pill { + display: inline-block; + margin-left: 6px; + padding: 0 4px; + border-radius: var(--radius-sm); + font-size: 0.8em; + font-variant-numeric: tabular-nums; + opacity: 0.85; +} +.mcp-refresh-pill-ok { + background: var(--bg-highlight); + color: var(--fg-dim); +} +.mcp-refresh-pill-err { + background: color-mix(in srgb, var(--warn) 15%, transparent); + color: var(--warn); +} + .mcp-detail-modal::before { background: linear-gradient( 90deg, diff --git a/turnstone/core/mcp_client.py b/turnstone/core/mcp_client.py index 13d1de03..86573473 100644 --- a/turnstone/core/mcp_client.py +++ b/turnstone/core/mcp_client.py @@ -27,6 +27,7 @@ import urllib.parse import uuid from contextlib import AsyncExitStack from dataclasses import dataclass, field +from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any, Literal @@ -45,6 +46,7 @@ from turnstone.core.config import load_config from turnstone.core.log import get_logger from turnstone.core.mcp_http_parsers import ( MAX_INSUFFICIENT_SCOPE_REPORTED, + is_valid_scope_token, parse_www_authenticate_error, parse_www_authenticate_scope, ) @@ -510,6 +512,20 @@ class MCPClientManager: # Notification debounce (per-server) self._last_notification_refresh: dict[str, float] = {} + # Last refresh outcome (Phase 9 — admin status indicator). Per- + # server tuple of ``(unix_ts, outcome)`` where outcome is one of + # ``ok`` or ``error:``. Populated by + # ``_refresh_server`` on every call (success and failure paths), + # which means manual operator-driven refresh (``refresh_sync``) + # AND the ``_cb_auto_reconnect`` follow-up that schedules + # ``_refresh_server`` directly both populate the field — adding + # a future schedule site only needs to call ``_refresh_server`` + # to participate. Read by the admin status endpoint to render + # the per-server "last refresh" pill. No initial entry is + # created at server-register time — absence surfaces as ``null`` + # in the admin JSON, which the UI renders as "never". + self._last_refresh: dict[str, tuple[float, str]] = {} + # Per-(user, server) state for auth_type=oauth_user. Loop-bound: # mutated only on the mcp-loop. Sync threads interact via # ``asyncio.run_coroutine_threadsafe``. @@ -2131,14 +2147,51 @@ class MCPClientManager: Returns ``(added_tools, removed_tools)`` names (tool diff only, for backward compatibility with ``/mcp refresh`` output). + + Writes the ``_last_refresh`` entry on every call so the Phase 9 + admin status pill reflects every refresh path — manual + operator-driven ``refresh_sync`` AND the ``_cb_auto_reconnect`` + follow-up that schedules ``_refresh_server`` directly. + Centralising the write here means future schedule sites + automatically populate the field. + + Uses ``asyncio.gather(return_exceptions=True)`` so that a failure + in one of the three concurrent sub-refreshes does NOT orphan the + others mid-mutation: every sibling reaches completion (success + or per-task failure) before the outcome is computed. The + ``_last_refresh`` write is ``"ok"`` iff all three succeeded; on + any failure the outcome is ``f"error:{type(first_exc).__name__}"`` + and the first exception is re-raised so the outer caller's error + path (``_refresh_all``'s except, or the manual-refresh sync + wrapper) sees the same shape it did before this rework. + Partial-success mutations of ``state.tools`` / ``state.resources`` + / ``state.prompts`` are bounded to whichever sub-refresh + succeeded — the documented trade-off vs leaving orphan tasks + running after the error is observed. """ - tool_diff, _, _ = await asyncio.gather( + results = await asyncio.gather( self._refresh_server_tools(name), self._refresh_server_resources(name), self._refresh_server_prompts(name), + return_exceptions=True, ) + first_exc: BaseException | None = next( + (r for r in results if isinstance(r, BaseException)), None + ) + if first_exc is not None: + self._last_refresh[name] = ( + time.time(), + f"error:{type(first_exc).__name__}", + ) + raise first_exc + tool_diff = results[0] + # ``return_exceptions=True`` widens the static type; on the all- + # success path each entry is the awaited result. We narrow the + # tool-diff entry to the documented ``(added, removed)`` shape. + assert isinstance(tool_diff, tuple) added, removed = tool_diff self._last_error.pop(name, None) + self._last_refresh[name] = (time.time(), "ok") return added, removed async def _refresh_all( @@ -2167,6 +2220,7 @@ class MCPClientManager: [t["function"]["name"] for t in post.tools] if post is not None else [] ) results[name] = (new_names, []) + self._last_refresh[name] = (time.time(), "ok") continue added, removed = await self._refresh_server(name) self._cb_record_success(name) @@ -2175,6 +2229,20 @@ class MCPClientManager: log.warning("Refresh failed for MCP server '%s'", name, exc_info=True) self._set_error(name, f"Refresh failed: {exc}") results[name] = ([], []) + # Overwrite unconditionally with the freshest observed + # outcome. Two cases produce the write: + # (1) Reconnect branch: ``_connect_one`` raised before + # ``_refresh_server`` could write — no prior entry + # from this iteration exists yet, so the write is + # the only fresh signal. + # (2) ``_refresh_server`` branch: it already wrote a + # fresh ``error:`` before re-raising, so + # the outer overwrite is a no-op for the value. + # Using ``setdefault`` here would preserve a stale prior + # ``"ok"`` from the previous successful refresh when the + # current attempt fails — the admin pill would show + # "ok" for a broken server. + self._last_refresh[name] = (time.time(), f"error:{type(exc).__name__}") # Final sync to clean up templates from servers that are no longer connected try: @@ -2952,6 +3020,7 @@ class MCPClientManager: transport = cfg.get("type", "stdio") cb_deadline = self._circuit_open_until.get(name) cb_open = cb_deadline is not None and time.monotonic() < cb_deadline + last_refresh = self._last_refresh.get(name) # Inline predicate (instead of reusing ``connected``) so mypy narrows # ``state`` for the attribute reads — a separate boolean wouldn't. return { @@ -2967,6 +3036,10 @@ class MCPClientManager: "url": cfg.get("url", "") if transport != "stdio" else "", "circuit_open": cb_open, "consecutive_failures": self._consecutive_failures.get(name, 0), + # Phase 9 admin status: last manual / auto-reconnect refresh. + # ``null`` when no refresh has occurred since process start. + "last_refresh_at": last_refresh[0] if last_refresh is not None else None, + "last_refresh_outcome": last_refresh[1] if last_refresh is not None else None, } def get_all_server_status(self) -> dict[str, dict[str, Any]]: @@ -3303,6 +3376,7 @@ class MCPClientManager: *, user_id: str | None = None, timeout: int = 120, + is_interactive_for_consent: bool = True, ) -> str: """Execute an MCP tool call synchronously (blocks the calling thread). @@ -3348,6 +3422,7 @@ class MCPClientManager: arguments=arguments, server_row=pool_target[2], timeout=timeout, + is_interactive_for_consent=is_interactive_for_consent, ) if mapping is None or server_name is None or original_name is None: @@ -3500,6 +3575,54 @@ class MCPClientManager: return None return server_name, original, row + def _record_pending_consent_best_effort( + self, + *, + user_id: str, + server_name: str, + result: str, + ) -> None: + """Persist a deferred-consent row for non-interactive callers. + + Called from the three sync dispatchers when the dispatch returns + a structured-error envelope AND the caller is not interactive + (CHAT / SCHEDULED). Filters out structured-error codes that + aren't user-consent-shaped (key-unknown, url-insecure, + *_forbidden) — those are operator-actionable and outside the + scope of the dashboard pending-consent badge. + + Best-effort. A storage exception is logged but never raised: + the structured-error envelope must reach the agent unchanged + regardless of whether the pending-consent row was persisted, so + a transient DB failure doesn't change the agent-observable + contract. + """ + if self._storage is None: + return + parsed = _parse_pending_consent_envelope(result) + if parsed is None: + return + code, scopes = parsed + scopes_str = " ".join(scopes) if scopes else None + try: + self._storage.upsert_mcp_pending_consent( + user_id=user_id, + server_name=server_name, + error_code=code, + scopes_required=scopes_str, + last_ws_id=None, + last_tool_call_id=None, + now_iso=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + ) + except Exception: + log.warning( + "mcp_pool.pending_consent_persist_failed user=%s server=%s code=%s", + user_id, + server_name, + code, + exc_info=True, + ) + def _dispatch_pool_sync( self, *, @@ -3509,6 +3632,7 @@ class MCPClientManager: arguments: dict[str, Any], server_row: dict[str, Any], timeout: int, + is_interactive_for_consent: bool = True, ) -> str: """Synchronous wrapper for pool dispatch. @@ -3581,6 +3705,10 @@ class MCPClientManager: server_row=server_row, ) if _is_structured_error(result): + if not is_interactive_for_consent: + self._record_pending_consent_best_effort( + user_id=user_id, server_name=server_name, result=result + ) raise RuntimeError(result) return result @@ -3636,6 +3764,7 @@ class MCPClientManager: uri: str, server_row: dict[str, Any], timeout: int, + is_interactive_for_consent: bool = True, ) -> str: """Synchronous wrapper for pool resource read. @@ -3679,6 +3808,10 @@ class MCPClientManager: server_row=server_row, ) if _is_structured_error(result): + if not is_interactive_for_consent: + self._record_pending_consent_best_effort( + user_id=user_id, server_name=server_name, result=result + ) raise RuntimeError(result) return result @@ -3722,6 +3855,7 @@ class MCPClientManager: arguments: dict[str, str] | None, server_row: dict[str, Any], timeout: int, + is_interactive_for_consent: bool = True, ) -> list[dict[str, Any]]: """Synchronous wrapper for pool prompt invocation. @@ -3765,6 +3899,10 @@ class MCPClientManager: # Structured-error path — surface as RuntimeError so the # agent-loop renders the JSON via its except-Exception # handler. + if not is_interactive_for_consent: + self._record_pending_consent_best_effort( + user_id=user_id, server_name=server_name, result=result + ) raise RuntimeError(result) return result @@ -4604,7 +4742,12 @@ class MCPClientManager: return best def read_resource_sync( - self, uri: str, *, user_id: str | None = None, timeout: int = 120 + self, + uri: str, + *, + user_id: str | None = None, + timeout: int = 120, + is_interactive_for_consent: bool = True, ) -> str: """Read a resource by URI synchronously (blocks the calling thread). @@ -4627,6 +4770,7 @@ class MCPClientManager: uri=pool_target[1], server_row=pool_target[2], timeout=timeout, + is_interactive_for_consent=is_interactive_for_consent, ) mapping = self._resource_map.get(uri) @@ -4689,6 +4833,7 @@ class MCPClientManager: *, user_id: str | None = None, timeout: int = 30, + is_interactive_for_consent: bool = True, ) -> list[dict[str, Any]]: """Invoke an MCP prompt synchronously and return expanded messages. @@ -4727,6 +4872,7 @@ class MCPClientManager: arguments=arguments, server_row=pool_target[2], timeout=timeout, + is_interactive_for_consent=is_interactive_for_consent, ) if static_mapping is None: @@ -4912,6 +5058,64 @@ def _structured_error( return json.dumps({"error": err}) +# Structured-error codes that represent a deferred-consent need. When +# encountered on a non-interactive call (chat / scheduled), the sync +# dispatcher persists a row to ``mcp_pending_consent`` so the dashboard +# badge can surface the deferred work later. Operator-actionable codes +# (key-unknown, url-insecure, *_forbidden) are intentionally excluded — +# the user cannot resolve them by completing a consent flow. +_PENDING_CONSENT_PERSIST_CODES: frozenset[str] = frozenset( + {"mcp_consent_required", "mcp_insufficient_scope"} +) + + +def _parse_pending_consent_envelope( + result: str, +) -> tuple[str, list[str] | None] | None: + """Extract ``(error_code, scopes_required)`` from a structured-error JSON. + + Returns ``None`` when the envelope's ``code`` is not in + :data:`_PENDING_CONSENT_PERSIST_CODES`. Callers should already have + gated on :func:`_is_structured_error`; this helper deliberately + re-parses (cheap on the failure path) rather than threading the + decoded dict through the sync-dispatcher hot path. + + Defends against non-dict JSON values (``null``, strings, numbers) + via the same ``isinstance(decoded, dict)`` guard + :func:`_is_structured_error` uses, so a misuse from a future caller + that bypasses the structured-error contract surfaces as a clean + ``None`` rather than an ``AttributeError`` propagating out of the + sync dispatcher's hot path. + """ + try: + decoded = json.loads(result) + except (json.JSONDecodeError, ValueError): + return None + if not isinstance(decoded, dict): + return None + err = decoded.get("error") + if not isinstance(err, dict): + return None + code = err.get("code", "") + if code not in _PENDING_CONSENT_PERSIST_CODES: + return None + scopes = err.get("scopes_required") + if isinstance(scopes, list): + # Defense-in-depth scope filter — production paths construct + # this list via ``parse_www_authenticate_scope`` which already + # validates and caps, but the helper is reusable; re-applying + # the predicate here forecloses any future caller that bypasses + # the upstream filter from landing attacker-controlled bytes in + # ``mcp_pending_consent.scopes_required``. Type-filter BEFORE + # ``is_valid_scope_token`` so non-string entries (``None``, + # ints) don't slip through as their ``str()`` repr (e.g. + # ``None`` → ``"None"`` passes the ASCII grammar). Cap mirrors + # ``MAX_INSUFFICIENT_SCOPE_REPORTED`` semantics. + cleaned = [s for s in scopes if isinstance(s, str) and is_valid_scope_token(s)] + return code, cleaned[:MAX_INSUFFICIENT_SCOPE_REPORTED] + return code, None + + def _is_structured_error(result: str) -> bool: """Return True if *result* parses as a :func:`_structured_error` envelope. diff --git a/turnstone/core/mcp_oauth.py b/turnstone/core/mcp_oauth.py index cc962dc9..8e378642 100644 --- a/turnstone/core/mcp_oauth.py +++ b/turnstone/core/mcp_oauth.py @@ -2350,6 +2350,22 @@ async def _handle_mcp_oauth_callback_inner(request: Request) -> Response: }, ) + # Phase 9 — clear any deferred-consent records for this (user, + # server) now that consent has completed. Best-effort: a storage + # failure here doesn't change the user-observable callback success; + # the worst case is a stale badge that the user can dismiss + # manually. ``delete_mcp_pending_consent`` returns False on + # no-such-row (the common case for interactive consent flows that + # never deferred), which is fine. + try: + await asyncio.to_thread(storage.delete_mcp_pending_consent, user_id, server_name) + except Exception: + log.debug( + "mcp_server.oauth.pending_consent_clear_failed", + server_name=server_name, + exc_info=True, + ) + return RedirectResponse(pending["return_url"] or "/", status_code=302) @@ -2618,6 +2634,153 @@ async def _handle_mcp_oauth_revoke_connection_inner(request: Request) -> Respons return Response(status_code=204) +# --------------------------------------------------------------------------- +# Pending-consent endpoints (Phase 9) +# --------------------------------------------------------------------------- + + +async def handle_mcp_oauth_list_pending(request: Request) -> Response: + """``GET /v1/api/mcp/oauth/pending``. + + Returns the authenticated user's deferred-consent records — populated + by the pool dispatchers when a non-interactive run (scheduled / + channel) hits ``mcp_consent_required`` or ``mcp_insufficient_scope``. + Used by the dashboard badge to surface deferred consent needs on + next login. + + Install-level gate: when no ``mcp_servers`` row has + ``auth_type='oauth_user'``, the entire feature is dark — we + short-circuit to ``{pending: 0, servers: []}`` without querying the + pending table at all. This keeps local-auth installs on a + zero-new-storage-query path. + """ + return _apply_security_headers(await _handle_mcp_oauth_list_pending_inner(request)) + + +_INSTALL_GATE_CACHE_TTL_S = 60.0 + + +async def _install_gate_passes(app_state: Any, storage: Any) -> bool: + """Cached install-level gate for OAuth-MCP features. + + Returns True iff at least one ``mcp_servers`` row has + ``auth_type='oauth_user'``. Result is cached on ``app_state`` for + :data:`_INSTALL_GATE_CACHE_TTL_S` seconds — admin-rare transitions + don't justify a per-request DB round-trip on every dashboard load. + + Reset semantics: cache is invalidated by time only. Operators who + just enabled an ``oauth_user`` row see the gate flip within the TTL + window. False positives (cache says True but the row was just + deleted) are bounded by the same window — the downstream list + query already filters by user, so the cost is at most one cheap + user-scoped read. + """ + now = time.monotonic() + cached = getattr(app_state, "_mcp_install_gate_cache", None) + if cached is not None: + cached_value, cached_at = cached + if (now - cached_at) < _INSTALL_GATE_CACHE_TTL_S: + return bool(cached_value) + value = bool(await asyncio.to_thread(storage.any_oauth_user_mcp_servers)) + app_state._mcp_install_gate_cache = (value, now) + return value + + +async def _handle_mcp_oauth_list_pending_inner(request: Request) -> Response: + from starlette.responses import JSONResponse + + user_id = _require_user_id(request) + if user_id is None: + return JSONResponse({"error": "Authentication required"}, status_code=401) + + storage = _get_storage(request.app.state) + if storage is None: + return JSONResponse({"pending": 0, "servers": []}) + + if not await _install_gate_passes(request.app.state, storage): + return JSONResponse({"pending": 0, "servers": []}) + + rows = await asyncio.to_thread(storage.list_mcp_pending_consent_by_user, user_id) + return JSONResponse({"pending": len(rows), "servers": list(rows)}) + + +async def handle_mcp_oauth_clear_pending(request: Request) -> Response: + """``DELETE /v1/api/mcp/oauth/pending/{server_name}``. + + Manual user-initiated dismissal of a single deferred-consent record. + Called from the dashboard settings modal when the user opts to clear + the entry without completing consent (e.g., the underlying + auth_type was changed and the deferred record is now stale). + + Returns 204 in both the existed-and-deleted and never-existed cases + to keep cross-tenant existence non-observable. + """ + return _apply_security_headers(await _handle_mcp_oauth_clear_pending_inner(request)) + + +async def _handle_mcp_oauth_clear_pending_inner(request: Request) -> Response: + from starlette.responses import JSONResponse, Response + + user_id = _require_user_id(request) + if user_id is None: + return JSONResponse({"error": "Authentication required"}, status_code=401) + + server_name = request.path_params.get("server_name", "").strip() + if not server_name: + return JSONResponse({"error": "Missing server_name"}, status_code=400) + + storage = _get_storage(request.app.state) + if storage is None: + return JSONResponse({"error": "Storage unavailable"}, status_code=503) + + cleared = bool( + await asyncio.to_thread(storage.delete_mcp_pending_consent, user_id, server_name) + ) + # Audit even on no-op deletes (returns 204 either way for cross-tenant + # non-observability) so an attacker who tries to scrub deferred-consent + # breadcrumbs leaves an audit trail of the attempts. + await _audit_event( + request.app.state, + user_id=user_id, + action="mcp_server.oauth.pending_consent_dismissed", + server_name=server_name, + detail={"mode": "single", "cleared": 1 if cleared else 0}, + ) + return Response(status_code=204) + + +async def handle_mcp_oauth_clear_all_pending(request: Request) -> Response: + """``DELETE /v1/api/mcp/oauth/pending``. + + Bulk dismiss of every deferred-consent record for the authenticated + user. Returns the count cleared so the dashboard can update its + badge in one round-trip. + """ + return _apply_security_headers(await _handle_mcp_oauth_clear_all_pending_inner(request)) + + +async def _handle_mcp_oauth_clear_all_pending_inner(request: Request) -> Response: + from starlette.responses import JSONResponse + + user_id = _require_user_id(request) + if user_id is None: + return JSONResponse({"error": "Authentication required"}, status_code=401) + + storage = _get_storage(request.app.state) + if storage is None: + return JSONResponse({"error": "Storage unavailable"}, status_code=503) + + cleared = await asyncio.to_thread(storage.delete_all_mcp_pending_consent_by_user, user_id) + await _audit_event( + request.app.state, + user_id=user_id, + action="mcp_server.oauth.pending_consent_dismissed", + server_name="(bulk)", + detail={"mode": "bulk", "cleared": cleared}, + ) + return JSONResponse({"cleared": cleared}) + + # --------------------------------------------------------------------------- # Lifespan integration # --------------------------------------------------------------------------- @@ -2674,7 +2837,10 @@ __all__ = [ "get_user_access_token_classified", "handle_mcp_oauth_authorize", "handle_mcp_oauth_callback", + "handle_mcp_oauth_clear_all_pending", + "handle_mcp_oauth_clear_pending", "handle_mcp_oauth_list_connections", + "handle_mcp_oauth_list_pending", "handle_mcp_oauth_revoke_connection", "initialize_mcp_oauth_state", "pop_pending_state", diff --git a/turnstone/core/session.py b/turnstone/core/session.py index fc6cd86a..cae05088 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -112,7 +112,12 @@ from turnstone.core.tools import ( from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS from turnstone.core.web import check_ssrf, strip_html from turnstone.core.workstream import WorkstreamKind -from turnstone.prompts import ClientType, SessionContext, compose_system_message +from turnstone.prompts import ( + INTERACTIVE_CONSENT_CLIENT_TYPES, + ClientType, + SessionContext, + compose_system_message, +) from turnstone.ui.colors import DIM, GRAY, GREEN, RED, RESET, YELLOW, bold, cyan, dim log = get_logger(__name__) @@ -851,6 +856,15 @@ class ChatSession: self._mcp_user_id: str | None = user_id or None self._username = username self._client_type = client_type + # Whether the user is online to complete an in-flight OAuth + # consent redirect. WEB and CLI users are; CHAT (Discord / + # Slack) and SCHEDULED (autonomous runs) are not — their + # consent-required errors must be persisted to + # ``mcp_pending_consent`` by the pool dispatchers for later + # surfacing on the dashboard badge, rather than relying on the + # in-flight SSE rendering path that Phase 8 ships for + # interactive surfaces. + self._is_interactive_for_consent: bool = client_type in INTERACTIVE_CONSENT_CLIENT_TYPES self._config_store = config_store # Initialize rule registry for configurable judge rules self._rule_registry = None @@ -8595,6 +8609,7 @@ class ChatSession: args, user_id=self._mcp_user_id, timeout=self.tool_timeout, + is_interactive_for_consent=self._is_interactive_for_consent, ) except TimeoutError: output = f"MCP tool timed out after {self.tool_timeout}s" @@ -8677,7 +8692,10 @@ class ChatSession: # 401 / 403 / consent-required handling. Otherwise the # static path runs byte-identical (invariant 1). output = self._mcp_client.read_resource_sync( - uri, user_id=self._mcp_user_id, timeout=self.tool_timeout + uri, + user_id=self._mcp_user_id, + timeout=self.tool_timeout, + is_interactive_for_consent=self._is_interactive_for_consent, ) except TimeoutError: output = f"MCP resource read timed out after {self.tool_timeout}s" @@ -8774,6 +8792,7 @@ class ChatSession: arguments or None, user_id=self._mcp_user_id, timeout=self.tool_timeout, + is_interactive_for_consent=self._is_interactive_for_consent, ) output = "\n\n".join(f"[{m['role']}]: {m['content']}" for m in messages) except TimeoutError: diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 23b59101..af9e992a 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -19,6 +19,7 @@ import sqlalchemy as sa from turnstone.core.log import get_logger from turnstone.core.storage._protocol import ( MCPOAuthPendingState, + MCPPendingConsentRow, MCPUserToken, MCPUserTokenMetadataRow, OIDCIdentity, @@ -33,6 +34,7 @@ from turnstone.core.storage._schema import ( heuristic_rules, intent_verdicts, mcp_oauth_pending, + mcp_pending_consent, mcp_servers, mcp_user_tokens, metadata, @@ -4308,6 +4310,136 @@ class PostgreSQLBackend: conn.commit() return result.rowcount + # -- MCP pending-consent (Phase 9) ---------------------------------------- + + def upsert_mcp_pending_consent( + self, + user_id: str, + server_name: str, + error_code: str, + scopes_required: str | None, + last_ws_id: str | None, + last_tool_call_id: str | None, + now_iso: str, + ) -> None: + from sqlalchemy.dialects import postgresql + + stmt = postgresql.insert(mcp_pending_consent).values( + user_id=user_id, + server_name=server_name, + error_code=error_code, + scopes_required=scopes_required, + last_ws_id=last_ws_id, + last_tool_call_id=last_tool_call_id, + first_seen_at=now_iso, + last_seen_at=now_iso, + occurrence_count=1, + ) + stmt = stmt.on_conflict_do_update( + index_elements=["user_id", "server_name"], + set_={ + "error_code": stmt.excluded.error_code, + "scopes_required": stmt.excluded.scopes_required, + "last_ws_id": stmt.excluded.last_ws_id, + "last_tool_call_id": stmt.excluded.last_tool_call_id, + "last_seen_at": stmt.excluded.last_seen_at, + "occurrence_count": mcp_pending_consent.c.occurrence_count + 1, + }, + ) + with self._conn() as conn: + conn.execute(stmt) + conn.commit() + + def list_mcp_pending_consent_by_user(self, user_id: str) -> list[MCPPendingConsentRow]: + with self._conn() as conn: + rows = conn.execute( + sa.select(mcp_pending_consent) + .where(mcp_pending_consent.c.user_id == user_id) + .order_by(mcp_pending_consent.c.last_seen_at.desc()) + ).fetchall() + out: list[MCPPendingConsentRow] = [] + for r in rows: + m = r._mapping + out.append( + MCPPendingConsentRow( + user_id=m["user_id"], + server_name=m["server_name"], + error_code=m["error_code"], + scopes_required=m["scopes_required"], + last_ws_id=m["last_ws_id"], + last_tool_call_id=m["last_tool_call_id"], + first_seen_at=m["first_seen_at"], + last_seen_at=m["last_seen_at"], + occurrence_count=m["occurrence_count"], + ) + ) + return out + + def delete_mcp_pending_consent(self, user_id: str, server_name: str) -> bool: + with self._conn() as conn: + result = conn.execute( + sa.delete(mcp_pending_consent).where( + (mcp_pending_consent.c.user_id == user_id) + & (mcp_pending_consent.c.server_name == server_name) + ) + ) + conn.commit() + return bool(result.rowcount) + + def delete_all_mcp_pending_consent_by_user(self, user_id: str) -> int: + with self._conn() as conn: + result = conn.execute( + sa.delete(mcp_pending_consent).where(mcp_pending_consent.c.user_id == user_id) + ) + conn.commit() + return int(result.rowcount or 0) + + def count_mcp_consented_users_by_server(self, server_name: str) -> int: + # ``expires_at IS NULL`` => non-expired (refresh-only tokens with no + # advertised expiry). Compare lexically against ISO-8601 strings, + # mirroring the convention in ``mcp_user_tokens.expires_at``. + now_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._conn() as conn: + result = conn.execute( + sa.select(sa.func.count(sa.distinct(mcp_user_tokens.c.user_id))) + .where(mcp_user_tokens.c.server_name == server_name) + .where( + sa.or_( + mcp_user_tokens.c.expires_at.is_(None), + mcp_user_tokens.c.expires_at > now_iso, + ) + ) + ).scalar() + return int(result or 0) + + def count_mcp_consented_users_grouped_by_server(self) -> dict[str, int]: + now_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._conn() as conn: + rows = conn.execute( + sa.select( + mcp_user_tokens.c.server_name, + sa.func.count(sa.distinct(mcp_user_tokens.c.user_id)), + ) + .where( + sa.or_( + mcp_user_tokens.c.expires_at.is_(None), + mcp_user_tokens.c.expires_at > now_iso, + ) + ) + .group_by(mcp_user_tokens.c.server_name) + ).fetchall() + return {row[0]: int(row[1] or 0) for row in rows} + + def any_oauth_user_mcp_servers(self) -> bool: + with self._conn() as conn: + result = conn.execute( + sa.select(sa.literal(1)) + .select_from(mcp_servers) + .where(mcp_servers.c.auth_type == "oauth_user") + .limit(1) + ).scalar() + return result is not None + # -- Model definitions ----------------------------------------------------- def create_model_definition( diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index dfef0147..3e9ed598 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -93,6 +93,27 @@ class MCPOAuthPendingState(TypedDict): created_at: str +class MCPPendingConsentRow(TypedDict): + """Row shape for deferred-consent records. + + Emitted by the pool dispatchers (Phase 5+) when a non-interactive + run (scheduled / channel) hits ``mcp_consent_required`` or + ``mcp_insufficient_scope`` and the user can't be prompted in the + moment. Composite PK ``(user_id, server_name)`` collapses repeat + occurrences for the same server into one row. + """ + + user_id: str + server_name: str + error_code: str + scopes_required: str | None + last_ws_id: str | None + last_tool_call_id: str | None + first_seen_at: str + last_seen_at: str + occurrence_count: int + + @runtime_checkable class StorageBackend(Protocol): """Protocol that every storage backend adapter must implement. @@ -1865,6 +1886,89 @@ class StorageBackend(Protocol): """Bulk-delete expired pending MCP OAuth rows. Returns count deleted.""" ... + # -- MCP pending-consent (Phase 9; deferred-consent persistence) ---------- + + def upsert_mcp_pending_consent( + self, + user_id: str, + server_name: str, + error_code: str, + scopes_required: str | None, + last_ws_id: str | None, + last_tool_call_id: str | None, + now_iso: str, + ) -> None: + """Insert or refresh a deferred-consent record for ``(user, server)``. + + On insert: ``first_seen_at = last_seen_at = now_iso``, + ``occurrence_count = 1``. On conflict (existing row for the + same composite PK): rewrites ``error_code``, ``scopes_required``, + ``last_ws_id``, ``last_tool_call_id``, ``last_seen_at`` to the + current values; bumps ``occurrence_count`` by 1. Preserves + ``first_seen_at`` so the dashboard can show how long the + deferred-consent need has been pending. + """ + ... + + def list_mcp_pending_consent_by_user(self, user_id: str) -> list[MCPPendingConsentRow]: + """Return all deferred-consent records for ``user_id``. + + Ordered by ``last_seen_at`` DESC. Empty list when the user has + none. Used by the dashboard badge endpoint to render the + servers-need-consent list. + """ + ... + + def delete_mcp_pending_consent(self, user_id: str, server_name: str) -> bool: + """Delete the pending-consent row for ``(user, server)``. Returns True if existed. + + Called automatically by the OAuth callback handler when consent + completes, and manually via the user-facing DELETE endpoint. + """ + ... + + def delete_all_mcp_pending_consent_by_user(self, user_id: str) -> int: + """Bulk-delete every pending-consent row for ``user_id``. Returns count. + + Used by the manual "dismiss all" endpoint from the settings + modal. + """ + ... + + def count_mcp_consented_users_by_server(self, server_name: str) -> int: + """Distinct-user count of non-expired tokens for ``server_name``. + + ``expires_at IS NULL`` is treated as non-expired (refresh-only + tokens with no advertised expiry). Used by the admin status + indicator to show "N users consented" per MCP server row. + """ + ... + + def count_mcp_consented_users_grouped_by_server(self) -> dict[str, int]: + """Bulk distinct-user count of non-expired tokens, grouped by server. + + Single round-trip variant of + :meth:`count_mcp_consented_users_by_server` for the admin list + handler — replaces the N-call loop that issued one query per + server with one ``GROUP BY`` query returning ``{server_name: + count}`` for every server that has at least one non-expired + token. Servers with zero consented users are absent from the + result; callers should ``dict.get(name, 0)`` rather than + indexing. + """ + ... + + def any_oauth_user_mcp_servers(self) -> bool: + """Install-level gate for OAuth-MCP features. + + Returns True iff at least one ``mcp_servers`` row has + ``auth_type='oauth_user'``. Used to short-circuit the pending- + consent badge endpoint to ``{pending: 0}`` on local-auth installs + with no OAuth MCP servers, so those code paths exercise zero new + storage queries. + """ + ... + # -- Model definitions ----------------------------------------------------- def create_model_definition( diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index 18ef1410..bf60a26c 100644 --- a/turnstone/core/storage/_schema.py +++ b/turnstone/core/storage/_schema.py @@ -807,6 +807,15 @@ mcp_user_tokens = sa.Table( sa.Column("last_refreshed", sa.Text, nullable=True), sa.PrimaryKeyConstraint("user_id", "server_name"), ) +# Phase 9: covers the ``WHERE server_name = ? AND (expires_at IS NULL +# OR expires_at > now)`` shape used by ``count_mcp_consented_users_*`` +# for the admin status pill. The composite PK can't satisfy filters +# that don't lead with ``user_id``. +sa.Index( + "idx_mcp_user_tokens_server", + mcp_user_tokens.c.server_name, + mcp_user_tokens.c.expires_at, +) mcp_oauth_pending = sa.Table( "mcp_oauth_pending", @@ -821,6 +830,32 @@ mcp_oauth_pending = sa.Table( sa.Index("idx_mcp_pending_created", mcp_oauth_pending.c.created_at) +# Per-(user, server) pending-consent state for non-interactive contexts. +# Populated by the pool dispatchers when a scheduled / channel-driven run +# hits ``mcp_consent_required`` or ``mcp_insufficient_scope`` and the user +# can't be prompted in the moment. Read on dashboard load to render the +# "N MCP servers need consent" badge. Cleared by the OAuth callback when +# the matching ``(user, server)`` completes consent. +# +# Composite PK ``(user_id, server_name)`` collapses repeat occurrences for +# the same server into one row; ``occurrence_count`` + ``last_*`` fields +# carry recency metadata for the dashboard without inflating row count. +mcp_pending_consent = sa.Table( + "mcp_pending_consent", + metadata, + sa.Column("user_id", sa.Text, nullable=False), + sa.Column("server_name", sa.Text, nullable=False), + sa.Column("error_code", sa.Text, nullable=False), + sa.Column("scopes_required", sa.Text, nullable=True), + sa.Column("last_ws_id", sa.Text, nullable=True), + sa.Column("last_tool_call_id", sa.Text, nullable=True), + sa.Column("first_seen_at", sa.Text, nullable=False), + sa.Column("last_seen_at", sa.Text, nullable=False), + sa.Column("occurrence_count", sa.Integer, nullable=False, server_default="1"), + sa.PrimaryKeyConstraint("user_id", "server_name"), +) +sa.Index("idx_mcp_pending_consent_user", mcp_pending_consent.c.user_id) + # ── TLS / ACME (lacme integration) ────────────────────────────────────────── tls_account_keys = sa.Table( diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index d81f02c0..a392d3c2 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -19,6 +19,7 @@ if TYPE_CHECKING: from turnstone.core.log import get_logger from turnstone.core.storage._protocol import ( MCPOAuthPendingState, + MCPPendingConsentRow, MCPUserToken, MCPUserTokenMetadataRow, OIDCIdentity, @@ -33,6 +34,7 @@ from turnstone.core.storage._schema import ( heuristic_rules, intent_verdicts, mcp_oauth_pending, + mcp_pending_consent, mcp_servers, mcp_user_tokens, metadata, @@ -4462,6 +4464,133 @@ class SQLiteBackend: conn.commit() return result.rowcount + # -- MCP pending-consent (Phase 9) ---------------------------------------- + + def upsert_mcp_pending_consent( + self, + user_id: str, + server_name: str, + error_code: str, + scopes_required: str | None, + last_ws_id: str | None, + last_tool_call_id: str | None, + now_iso: str, + ) -> None: + from sqlalchemy.dialects import sqlite as sa_sqlite + + stmt = sa_sqlite.insert(mcp_pending_consent).values( + user_id=user_id, + server_name=server_name, + error_code=error_code, + scopes_required=scopes_required, + last_ws_id=last_ws_id, + last_tool_call_id=last_tool_call_id, + first_seen_at=now_iso, + last_seen_at=now_iso, + occurrence_count=1, + ) + stmt = stmt.on_conflict_do_update( + index_elements=["user_id", "server_name"], + set_={ + "error_code": stmt.excluded.error_code, + "scopes_required": stmt.excluded.scopes_required, + "last_ws_id": stmt.excluded.last_ws_id, + "last_tool_call_id": stmt.excluded.last_tool_call_id, + "last_seen_at": stmt.excluded.last_seen_at, + "occurrence_count": mcp_pending_consent.c.occurrence_count + 1, + }, + ) + with self._conn() as conn: + conn.execute(stmt) + conn.commit() + + def list_mcp_pending_consent_by_user(self, user_id: str) -> list[MCPPendingConsentRow]: + with self._conn() as conn: + rows = conn.execute( + sa.select(mcp_pending_consent) + .where(mcp_pending_consent.c.user_id == user_id) + .order_by(mcp_pending_consent.c.last_seen_at.desc()) + ).fetchall() + out: list[MCPPendingConsentRow] = [] + for r in rows: + m = r._mapping + out.append( + MCPPendingConsentRow( + user_id=m["user_id"], + server_name=m["server_name"], + error_code=m["error_code"], + scopes_required=m["scopes_required"], + last_ws_id=m["last_ws_id"], + last_tool_call_id=m["last_tool_call_id"], + first_seen_at=m["first_seen_at"], + last_seen_at=m["last_seen_at"], + occurrence_count=m["occurrence_count"], + ) + ) + return out + + def delete_mcp_pending_consent(self, user_id: str, server_name: str) -> bool: + with self._conn() as conn: + result = conn.execute( + sa.delete(mcp_pending_consent).where( + (mcp_pending_consent.c.user_id == user_id) + & (mcp_pending_consent.c.server_name == server_name) + ) + ) + conn.commit() + return bool(result.rowcount) + + def delete_all_mcp_pending_consent_by_user(self, user_id: str) -> int: + with self._conn() as conn: + result = conn.execute( + sa.delete(mcp_pending_consent).where(mcp_pending_consent.c.user_id == user_id) + ) + conn.commit() + return int(result.rowcount or 0) + + def count_mcp_consented_users_by_server(self, server_name: str) -> int: + now_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._conn() as conn: + result = conn.execute( + sa.select(sa.func.count(sa.distinct(mcp_user_tokens.c.user_id))) + .where(mcp_user_tokens.c.server_name == server_name) + .where( + sa.or_( + mcp_user_tokens.c.expires_at.is_(None), + mcp_user_tokens.c.expires_at > now_iso, + ) + ) + ).scalar() + return int(result or 0) + + def count_mcp_consented_users_grouped_by_server(self) -> dict[str, int]: + now_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + with self._conn() as conn: + rows = conn.execute( + sa.select( + mcp_user_tokens.c.server_name, + sa.func.count(sa.distinct(mcp_user_tokens.c.user_id)), + ) + .where( + sa.or_( + mcp_user_tokens.c.expires_at.is_(None), + mcp_user_tokens.c.expires_at > now_iso, + ) + ) + .group_by(mcp_user_tokens.c.server_name) + ).fetchall() + return {row[0]: int(row[1] or 0) for row in rows} + + def any_oauth_user_mcp_servers(self) -> bool: + with self._conn() as conn: + result = conn.execute( + sa.select(sa.literal(1)) + .select_from(mcp_servers) + .where(mcp_servers.c.auth_type == "oauth_user") + .limit(1) + ).scalar() + return result is not None + # -- Model definitions ----------------------------------------------------- def create_model_definition( diff --git a/turnstone/core/storage/migrations/versions/054_mcp_pending_consent.py b/turnstone/core/storage/migrations/versions/054_mcp_pending_consent.py new file mode 100644 index 00000000..86647f1d --- /dev/null +++ b/turnstone/core/storage/migrations/versions/054_mcp_pending_consent.py @@ -0,0 +1,50 @@ +"""Add mcp_pending_consent table. + +Stores per-(user, server) deferred-consent records emitted by the pool +dispatchers when a non-interactive run (scheduled / channel) hits +``mcp_consent_required`` or ``mcp_insufficient_scope``. Read on +dashboard load to render the "N MCP servers need consent" badge; cleared +by the OAuth callback handler when consent completes. + +Composite PK ``(user_id, server_name)`` collapses repeat occurrences for +the same server into one row. No FKs (matches the rest of the +oauth_user schema in migration 049). + +Revision ID: 054 +Revises: 053 +Create Date: 2026-05-11 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "054" +down_revision = "053" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "mcp_pending_consent", + sa.Column("user_id", sa.Text, nullable=False), + sa.Column("server_name", sa.Text, nullable=False), + sa.Column("error_code", sa.Text, nullable=False), + sa.Column("scopes_required", sa.Text, nullable=True), + sa.Column("last_ws_id", sa.Text, nullable=True), + sa.Column("last_tool_call_id", sa.Text, nullable=True), + sa.Column("first_seen_at", sa.Text, nullable=False), + sa.Column("last_seen_at", sa.Text, nullable=False), + sa.Column("occurrence_count", sa.Integer, nullable=False, server_default="1"), + sa.PrimaryKeyConstraint("user_id", "server_name"), + ) + op.create_index( + "idx_mcp_pending_consent_user", + "mcp_pending_consent", + ["user_id"], + ) + + +def downgrade() -> None: + op.drop_index("idx_mcp_pending_consent_user", table_name="mcp_pending_consent") + op.drop_table("mcp_pending_consent") diff --git a/turnstone/core/storage/migrations/versions/055_mcp_user_tokens_server_index.py b/turnstone/core/storage/migrations/versions/055_mcp_user_tokens_server_index.py new file mode 100644 index 00000000..1c9c3172 --- /dev/null +++ b/turnstone/core/storage/migrations/versions/055_mcp_user_tokens_server_index.py @@ -0,0 +1,59 @@ +"""Index mcp_user_tokens by (server_name, expires_at). + +Phase 9 admin pill (``count_mcp_consented_users_*``) filters +``mcp_user_tokens`` by ``server_name`` and ``expires_at``. The table's +only existing index is the composite PK ``(user_id, server_name)`` — +``user_id`` is the leading column, so a filter on ``server_name`` alone +must full-scan the table. The bulk ``GROUP BY server_name`` variant in +the admin list handler benefits from the same index. + +PostgreSQL uses ``CREATE INDEX CONCURRENTLY`` inside an +``autocommit_block`` so the build is non-blocking on a live system — +``mcp_user_tokens`` is on the token-refresh hot path and an +ACCESS EXCLUSIVE lock during build would stall refresh writers on +installs with non-trivial row counts. SQLite has no concurrent build +concept and the table-level write lock already serializes, so a plain +``op.create_index`` is fine. Pattern mirrors migration 048 +(``idx_workstreams_reaper``). + +Revision ID: 055 +Revises: 054 +Create Date: 2026-05-11 +""" + +from alembic import op + +revision = "055" +down_revision = "054" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + bind = op.get_bind() + dialect = bind.dialect.name + + if dialect == "postgresql": + with op.get_context().autocommit_block(): + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS " + "idx_mcp_user_tokens_server ON mcp_user_tokens " + "(server_name, expires_at)" + ) + else: + op.create_index( + "idx_mcp_user_tokens_server", + "mcp_user_tokens", + ["server_name", "expires_at"], + ) + + +def downgrade() -> None: + bind = op.get_bind() + dialect = bind.dialect.name + + if dialect == "postgresql": + with op.get_context().autocommit_block(): + op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_mcp_user_tokens_server") + else: + op.drop_index("idx_mcp_user_tokens_server", table_name="mcp_user_tokens") diff --git a/turnstone/prompts/__init__.py b/turnstone/prompts/__init__.py index 9cb2d929..dda24385 100644 --- a/turnstone/prompts/__init__.py +++ b/turnstone/prompts/__init__.py @@ -37,6 +37,19 @@ class ClientType(enum.StrEnum): WEB = "web" CLI = "cli" CHAT = "chat" + SCHEDULED = "scheduled" + + +# Subset of ``ClientType`` values where the user is present to complete +# an in-flight OAuth consent flow (browser redirect + return). CHAT and +# SCHEDULED users cannot drive a browser redirect from inside their +# delivery surface, so consent-required errors must be persisted to +# ``mcp_pending_consent`` for later surfacing rather than relying on the +# in-flight SSE rendering path. Used by ``ChatSession`` to set +# ``_is_interactive_for_consent`` at construction time. +INTERACTIVE_CONSENT_CLIENT_TYPES: frozenset[ClientType] = frozenset( + {ClientType.WEB, ClientType.CLI} +) @dataclasses.dataclass @@ -56,6 +69,7 @@ _ENV_MAP: dict[ClientType, str] = { ClientType.WEB: "env/web.md", ClientType.CLI: "env/cli.md", ClientType.CHAT: "env/chat.md", + ClientType.SCHEDULED: "env/scheduled.md", } diff --git a/turnstone/prompts/env/scheduled.md b/turnstone/prompts/env/scheduled.md new file mode 100644 index 00000000..043b0d14 --- /dev/null +++ b/turnstone/prompts/env/scheduled.md @@ -0,0 +1,20 @@ +## Output Environment + +Your response is generated by a scheduled or autonomous run — no human is watching the output as it streams. The result is delivered to the user later (Discord notification, dashboard badge, or persisted workstream history) where they will see it as a static block of markdown. + +**Implications:** + +- The user is not online to answer mid-task clarifying questions. Make the reasonable judgment and continue; mention the assumption you made in the final output so the user can correct course on the next run if needed. +- Tool calls that require interactive user consent (e.g., MCP servers gated on OAuth user authorization that the user has not yet completed) will return a deferred-consent error rather than block the run. Surface the deferred work in your final summary so the user knows what was skipped. +- Optimize for a clear, scannable final summary over conversational back-and-forth — the user reads the whole transcript at once, not turn-by-turn. + +**Available rendering:** + +- Standard GitHub-flavored markdown is supported in the dashboard surface. Discord delivery uses the same constraints as `chat.md` (no tables, no headings beyond bold text, no Mermaid/KaTeX). +- Default to chat-portable formatting (bold/italic/inline-code/bullets/code-blocks) unless you know the destination is the web dashboard. + +**Formatting principles:** + +- Lead with the outcome in one line: what was accomplished, what was skipped, and why. +- For multi-step work, end with a concise checklist of what ran and what remains. +- Cite specific identifiers (workstream IDs, tool names, MCP server names) so the user can resume the work without re-reading the trace. diff --git a/turnstone/server.py b/turnstone/server.py index ccd882f2..7d60ca10 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -2822,6 +2822,27 @@ async def mcp_oauth_revoke_connection(request: Request) -> Response: return await handle_mcp_oauth_revoke_connection(request) +async def mcp_oauth_list_pending(request: Request) -> Response: + """GET /v1/api/mcp/oauth/pending — list deferred-consent records (Phase 9).""" + from turnstone.core.mcp_oauth import handle_mcp_oauth_list_pending + + return await handle_mcp_oauth_list_pending(request) + + +async def mcp_oauth_clear_pending(request: Request) -> Response: + """DELETE /v1/api/mcp/oauth/pending/{server_name} — dismiss a deferred-consent record.""" + from turnstone.core.mcp_oauth import handle_mcp_oauth_clear_pending + + return await handle_mcp_oauth_clear_pending(request) + + +async def mcp_oauth_clear_all_pending(request: Request) -> Response: + """DELETE /v1/api/mcp/oauth/pending — bulk-dismiss deferred-consent records.""" + from turnstone.core.mcp_oauth import handle_mcp_oauth_clear_all_pending + + return await handle_mcp_oauth_clear_all_pending(request) + + def list_interface_settings(request: Request) -> JSONResponse: """GET /v1/api/admin/settings — return interface settings from ConfigStore. @@ -4064,6 +4085,17 @@ def create_app( mcp_oauth_revoke_connection, methods=["DELETE"], ), + Route("/api/mcp/oauth/pending", mcp_oauth_list_pending), + Route( + "/api/mcp/oauth/pending", + mcp_oauth_clear_all_pending, + methods=["DELETE"], + ), + Route( + "/api/mcp/oauth/pending/{server_name}", + mcp_oauth_clear_pending, + methods=["DELETE"], + ), Route("/api/admin/settings", list_interface_settings), Route( "/api/admin/settings/{key:path}", diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js index bd82daab..453dba14 100644 --- a/turnstone/ui/static/app.js +++ b/turnstone/ui/static/app.js @@ -5230,6 +5230,35 @@ function _clearConsentBadge() { _refreshConsentBadge(); } +// Hydrate the pending-consent badge from the Phase 9 persistence endpoint +// on dashboard load. Closes the gap that pre-Phase-9 left open: a +// scheduled / channel-driven run that hit ``mcp_consent_required`` while +// the user wasn't online produced an in-flight SSE event that nobody saw. +// The endpoint short-circuits to ``{pending: 0}`` on installs with no +// ``auth_type=oauth_user`` MCP servers, so the call is cheap on local- +// auth deployments. Failures are silent — the badge will be re-driven +// by the next in-flight tool error if any. +function loadPendingConsents() { + authFetch("/v1/api/mcp/oauth/pending") + .then(function (r) { + if (!r.ok) return null; + return r.json(); + }) + .then(function (data) { + if (!data || !Array.isArray(data.servers)) return; + for (var i = 0; i < data.servers.length; i++) { + var row = data.servers[i]; + if (row && typeof row.server_name === "string") { + _pendingConsentServers.add(row.server_name); + } + } + _refreshConsentBadge(); + }) + .catch(function () { + // Endpoint failures must not block dashboard init. + }); +} + function _refreshConsentBadge() { var btn = document.getElementById("settings-btn"); if (!btn) return; @@ -6154,6 +6183,12 @@ function loadMcpConnections() { // failed fetch keeps the pending-consent signal until the user // gets confirmation that consents are in fact reachable. _clearConsentBadge(); + // Phase 9: re-hydrate the badge from the persistent pending- + // consent table. Phase 8 cleared in-memory state on settings- + // panel open (signal-acknowledged); Phase 9 records are + // DB-backed, so we re-pull them now to keep the badge in sync + // with what's actually pending across page lifetimes. + loadPendingConsents(); }) .catch(function (err) { loadingEl.style.display = "none"; @@ -6583,6 +6618,7 @@ initLogin(); pollHealth(); loadInterfaceSettings(); initWorkstreams(); +loadPendingConsents(); function loadInterfaceSettings() { authFetch("/v1/api/admin/settings")