feat(mcp): admin status, deferred-consent persistence, operator docs (Phase 9) (#516)

* feat(mcp): admin status, deferred-consent persistence, operator docs (Phase 9)

Completes the OAuth-MCP build-out (Phases 0-8 shipped) by closing the
operator + deferred-consent gaps:

1. **Per-(user, server) deferred-consent persistence** — when a
   non-interactive run (scheduled / channel) hits ``mcp_consent_required``
   or ``mcp_insufficient_scope``, the sync pool dispatchers now upsert a
   row into a new ``mcp_pending_consent`` table.  The dashboard hydrates
   the gear-icon badge from this table on load, so users who weren't
   online to see the in-flight SSE prompt still surface the deferred
   work on next login.  Cleared automatically by the OAuth callback
   handler on consent completion; manual user dismiss via new DELETE
   endpoints.  Composite PK ``(user_id, server_name)`` collapses repeat
   occurrences for the same server — no NULLs-not-distinct trap.

2. **Admin status pill + bulk-revoke** — the MCP Servers admin row now
   shows ``consented_users_count`` for ``auth_type=oauth_user`` rows
   when ≥1, with a two-step-confirm ``bulk-revoke`` button that drops
   every user's token for the server via the existing
   ``delete_mcp_oauth_rows_by_server_name`` primitive.  Upstream RFC
   7009 revoke is intentionally NOT attempted in bulk (avoids N
   upstream HTTP calls per admin click); audit detail records
   ``upstream_revoke_outcome=bulk_admin_no_upstream``.  A "last
   refresh" pill (age + outcome) renders on each row, sourced from a
   new ``_last_refresh`` dict populated by ``_refresh_server`` on every
   call (both manual ``refresh_sync`` and the ``_cb_auto_reconnect``
   follow-up).

3. **ClientType.SCHEDULED** added to the prompts module + scheduler
   passes it through to ``create_workstream``.  ``ChatSession`` now
   computes ``_is_interactive_for_consent`` at construction (WEB / CLI
   are interactive; CHAT / SCHEDULED are not) and plumbs the flag
   through ``call_tool_sync`` / ``read_resource_sync`` /
   ``get_prompt_sync`` to the three sync dispatchers.  The wrap at the
   ``_is_structured_error`` gate routes consent codes to the new
   ``_record_pending_consent_best_effort`` helper for non-interactive
   callers only; interactive sessions stay on the in-flight SSE path
   Phase 8 ships unchanged.

4. **Operator docs** — ``docs/mcp-oauth.md`` (operator guide, parallel
   to ``docs/oidc.md``: ``auth_type`` choice, OAuth client setup,
   encryption-key rotation, troubleshooting matrix) and
   ``docs/operations/mcp-oauth-headless.md`` (one-paragraph runbook
   per ``feedback_runbook_trust_llm.md``: pre-consent recipe for
   scheduled / channel-driven runs).

Schema
- Migration 054_mcp_pending_consent.py — composite PK
  ``(user_id, server_name)``, ``occurrence_count`` + ``first_seen_at`` /
  ``last_seen_at`` for recency metadata, ``idx_mcp_pending_consent_user``
  for the badge-load query.  No FKs (matches the rest of the
  oauth_user schema).
- Migration 055_mcp_user_tokens_server_index.py — adds
  ``idx_mcp_user_tokens_server`` on ``(server_name, expires_at)`` so
  the admin pill's ``count_mcp_consented_users_*`` queries don't
  full-scan against the leading-``user_id`` composite PK.
- Cross-backend: works on SQLite + PostgreSQL via dialect-specific
  ``on_conflict_do_update`` (PG ``postgresql.insert`` / SQLite
  ``sqlalchemy.dialects.sqlite.insert``).  No ``NULLS NOT DISTINCT``
  needed — the simplified PK eliminates the cross-version trap.

Endpoints
- ``GET /v1/api/mcp/oauth/pending`` — list deferred-consent records for
  the authenticated user.  Install-level gate via cached
  ``any_oauth_user_mcp_servers`` short-circuits to ``{pending: 0}`` on
  installs with no oauth_user MCP servers — local-auth deployments
  exercise zero new storage queries on this path.  The gate result is
  cached on ``app.state`` with a 60s TTL to spare repeat dashboard
  loads.
- ``DELETE /v1/api/mcp/oauth/pending/{server_name}`` — single dismiss.
  Returns 204 in both existed-and-deleted and never-existed cases
  (no cross-tenant existence leak); audits
  ``mcp_server.oauth.pending_consent_dismissed`` with
  ``mode=single`` + ``cleared=0|1`` so a session-hijack attacker
  scrubbing breadcrumbs leaves an audit trail.
- ``DELETE /v1/api/mcp/oauth/pending`` — bulk dismiss; audits
  ``mode=bulk`` + ``cleared=N``.
- ``POST /v1/api/admin/mcp-servers/{name}/bulk-revoke`` — admin
  bulk-revoke for the named server's per-user tokens.  Requires
  ``admin.mcp`` permission + 400s when the row isn't ``oauth_user``.

All four registered on both ``turnstone-server`` and
``turnstone-console`` (mirrors the Phase 8 ``/connections`` endpoint
shape).

Performance
- Admin list handler now uses a single ``GROUP BY`` bulk-count query
  (``count_mcp_consented_users_grouped_by_server``) wrapped in
  ``asyncio.to_thread`` rather than N per-row sync DB round-trips
  inside the async handler.  Skipped entirely when no row is
  oauth_user.

Frontend
- ``ui/static/app.js``: ``loadPendingConsents()`` hydrates the
  existing ``_pendingConsentServers`` set on dashboard init + after
  the user opens the settings modal.  Endpoint failures stay silent
  — the badge will be re-driven by the next in-flight tool error.
- ``console/static/admin.js``: ``consented_users_count`` pill +
  ``bulk-revoke`` button on each MCP row (only when ≥1 consented),
  two-step confirm matching the existing delete pattern.  ``last-
  refresh`` age + outcome pill in the per-row status cell, sourced
  from the freshest per-node entry in ``status[*].last_refresh_at`` /
  ``last_refresh_outcome``.  CSS for the pills in ``style.css``.

Tests
- ``test_mcp_pending_consent_storage`` — 13 tests covering upsert
  idempotency, list ordering, per-user isolation, single/bulk delete,
  count-by-server + grouped variant, install-level gate.
- ``test_mcp_pending_consent_dispatch`` — 9 tests, including the
  boundary-cross gate per ``feedback_tests_through_boundaries.md``:
  drives the real ``call_tool_sync`` → ``_dispatch_pool_sync`` →
  ``_is_structured_error`` → ``_record_pending_consent_best_effort``
  with a mocked classified-lookup so the structural plumb-through is
  verified end-to-end.  Includes a storage-failure test that pins
  the docstring's "envelope unchanged on storage failure" promise.
- ``test_mcp_pending_consent_endpoints`` — 11 tests: install gate,
  list-for-self, no-cross-user-leak, single/bulk delete, idempotent
  not-found, audit emission on single + bulk + cross-tenant dismiss.
- ``test_chat_session_interactivity_flag`` — 7 tests pinning the
  ``ClientType`` → ``_is_interactive_for_consent`` mapping against
  the module-level ``INTERACTIVE_CONSENT_CLIENT_TYPES`` frozenset.
- ``test_mcp_admin_bulk_revoke`` — 7 tests covering admin.mcp
  permission gate, 404 on missing, 400 on non-oauth_user, 200 with
  ``rows_deleted`` + ``consented_users_before``, audit row with
  ``upstream_revoke_outcome=bulk_admin_no_upstream``, cross-server
  isolation.
- ``test_mcp_oauth_handlers`` — 2 new callback tests pin the post-
  callback ``delete_mcp_pending_consent`` invocation: success-clears
  + storage-failure-still-redirects.
- 636 tests pass on the impacted surface (47 new + Phase 0-8 OAuth-MCP
  + session + prompts + storage admin).  ruff + mypy clean.

Hard invariants honored
- Static path byte-identical for ``auth_type ∈ {none, static}`` — the
  flag flows only through the pool dispatchers, which only fire when
  the row resolves to ``oauth_user``.
- ``asyncio.timeout`` (not ``asyncio.wait_for``) preserved on every
  AS / SDK / pool-loop await — no new awaits added to the hot path.
- Install-level gate on the badge endpoint: cached
  ``any_oauth_user_mcp_servers`` returns False on a row-less
  deployment → endpoint short-circuits without touching the pending-
  consent table; 60s TTL bounds the staleness window after admin
  flips ``auth_type``.
- Operator-actionable codes (key-unknown, url-insecure, *_forbidden)
  explicitly filtered out of persistence — they're outside the
  user-facing consent badge scope.
- Best-effort write: the structured-error envelope returned to the
  agent is identical whether the persistence write succeeds or fails
  (storage exception is logged with type name only — no chained
  context that could carry an ``httpx.Request`` bearer header).
- No ``exc_info=True`` on any new path that can chain a bearer-bearing
  ``httpx.Request``.
- Defensive parsing: ``_parse_pending_consent_envelope`` mirrors
  ``_is_structured_error``'s ``isinstance(decoded, dict)`` guard plus
  filters scope tokens through ``is_valid_scope_token`` capped at
  ``MAX_INSUFFICIENT_SCOPE_REPORTED`` — defense-in-depth even though
  production callers already validate upstream.
- Audit events on every dismiss endpoint so a session-control attacker
  scrubbing dashboard breadcrumbs still leaves a trail.

Cross-backend
- Tested on SQLite via the conftest backend fixture.
- PostgreSQL path uses ``postgresql.insert(...).on_conflict_do_update``
  parallel to the existing ``mcp_user_tokens`` upsert in Phase 3.

Deferred (not Phase 9 blockers)
- Multi-node pool eviction on bulk-revoke: only local-node sessions
  would be evicted if we built it, and there's no bulk-by-server
  primitive on MCPClientManager today; remote nodes will surface as
  a 401 on next dispatch which refreshes through the (now empty)
  token row.
- RFC 8693 / Azure OBO ``auth_type=oauth_token_exchange`` — captured
  in the design doc as a future architectural direction (~600 LOC +
  IdP-side admin work); requires OIDC token capture and per-MCP-server
  resource-trust configuration that v1 does not ship.

* docs(mcp): address Copilot review feedback on Phase 9

- Fix misleading admin.js comment that claimed the refresh pill rendered
  "<short-relative> <outcome>" — the pill actually renders only the short
  age, with outcome reflected via CSS class and tooltip.
- Replace broken feedback_secrets_not_in_env.md repo-root link in
  mcp-oauth.md with the inlined rationale (env-borne secrets reachable
  via shell tools / os.environ; TOML secrets are not).

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=<name>` 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=<name>&scopes=<extra>` 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.

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.

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()

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"

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({})

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

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") == []

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}

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

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:

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,

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 =
+        ' <span class="mcp-refresh-pill ' +
+        pillCls +
+        '" title="' +
+        escapeHtml(pillTitle) +
+        '">' +
+        escapeHtml(ageShort) +
+        "</span>";
+    }
+
     var transportCls =
       s.transport === "stdio" ? "mcp-transport-stdio" : "mcp-transport-http";
     var toolsVal = anyConnected
@@ -3385,6 +3429,24 @@ function _renderMcpServers(items) {
         '<button class="admin-btn-action" data-mcp-oauth-connect="' +
         escapeHtml(s.name) +
         '">connect</button>';
+      // 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 +=
+          '<button class="admin-btn-danger" data-mcp-bulk-revoke="' +
+          escapeHtml(s.name) +
+          '" data-mcp-consent-count="' +
+          consentCount +
+          '" title="Drop all ' +
+          consentCount +
+          " user consents for this server">bulk-revoke (" +
+          consentCount +
+          ")</button>";
+      }
     }
     var actions = isConfig
       ? actionBtns
@@ -3429,6 +3491,7 @@ function _renderMcpServers(items) {
       dotClass +
       '" aria-hidden="true"></span>' +
       escapeHtml(statusText) +
+      refreshPill +
       "</span>" +
       '<span class="admin-col admin-col-mactions">' +
       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");

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,

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:<ExceptionClassName>``.  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:<ClassName>`` 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.
 

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",

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:

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(

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(

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(

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(

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")

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")

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",
 }
 
 

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.

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}",

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")