fix(oidc/mcp): make runtime OIDC rediscovery actually work; preserve oauth_user paths

Round-7 review follow-up.

- The runtime OIDC re-discovery feature was dead code: discover_oidc
  PRESERVES the input config's `enabled` flag on success (only
  load_oidc_config ever sets it True), and maybe_rediscover_oidc always
  probed from the disabled boot config, so a successful rediscovery still
  returned enabled=False and the config swap was unreachable — the whole
  boot-outage auto-heal never worked. It now probes with enabled forced on
  so the flag is a reliable success signal. The unit test that "covered"
  this was mocking discover_oidc to return enabled=True, masking the bug;
  it now drives the real discover_oidc through a mocked HTTP discovery GET.

- The console never runs runtime rediscovery, so a transient discovery
  failure at console boot made every oauth_obo server un-editable and
  un-disable-able. The write gate now accepts a discovery_retryable config
  (OIDC configured, discovery transiently down) and rejects only a
  genuinely absent OIDC.

- The first rediscovery probe was suppressed for ~60s after host boot
  because the "last probe" timestamp defaulted to 0.0; it now uses a None
  sentinel for "never probed".

- Two behavior-preservation fixes for the pre-existing oauth_user path:
  the shared hardened token-POST no longer escalates oauth_user oversized
  error bodies (that status-based classification is opt-in for the obo
  legs only), and the token_revoked audit fires unconditionally for
  oauth_user again (a refresh failure means a real grant died) while
  staying delete-gated for obo to avoid revocation rows for tokens that
  never existed.

Cleanups: drop a throwaway set allocation in the pool-emptiness check,
compute the create handler's cleaned OAuth text once, remove a dead
no-op pop with a false comment, and simplify the cleared-map prune to two
non-overlapping passes.

tests/test_mcp_admin_api.py

@@ -864,6 +864,53 @@ class TestUpdateMcpServer:
         assert r.status_code == 400, r.text
         assert "OIDC" in r.json()["error"]
 
+    def test_obo_editable_when_oidc_discovery_transiently_failed(self, client, storage):
+        """Review finding: the console never runs runtime OIDC rediscovery, so a
+        transient discovery failure at console boot (enabled=False,
+        discovery_retryable=True) must NOT make oauth_obo servers un-editable /
+        un-disable-able. OIDC is still CONFIGURED (issuer set) — the write gate
+        accepts a discovery_retryable config; it rejects only a genuinely absent
+        OIDC (neither flag set)."""
+        # Seed an obo row (created while OIDC was healthy).
+        storage.create_mcp_server(
+            server_id="obo-retry-id",
+            name="obo-retry",
+            transport="streamable-http",
+            url="https://mcp.example.com/sse",
+            auth_type="oauth_obo",
+            oauth_audience="api://mcp-a",
+        )
+        # Console process booted while the IdP was briefly unreachable.
+        client.app.state.oidc_config = SimpleNamespace(
+            enabled=False,
+            issuer="https://idp.example.com",
+            obo_grant_profile="entra",
+            capture_user_credential=True,
+            discovery_retryable=True,
+        )
+        # Disabling the misbehaving obo server must succeed, not 400.
+        r = client.put(
+            "/v1/api/admin/mcp-servers/obo-retry-id",
+            json={"enabled": False},
+        )
+        assert r.status_code == 200, r.text
+        assert r.json()["enabled"] is False
+
+        # A genuinely-absent OIDC (neither enabled nor retryable) still rejects.
+        client.app.state.oidc_config = SimpleNamespace(
+            enabled=False,
+            issuer="",
+            obo_grant_profile="entra",
+            capture_user_credential=True,
+            discovery_retryable=False,
+        )
+        r2 = client.put(
+            "/v1/api/admin/mcp-servers/obo-retry-id",
+            json={"enabled": True},
+        )
+        assert r2.status_code == 400
+        assert "OIDC" in r2.json()["error"]
+
     def test_create_obo_rejected_on_invalid_grant_profile(self, client):
         """Review finding: a typo'd deployment obo_grant_profile leaves the mint
         leg unresolved (obo_misconfigured per dispatch), so reject it at the

tests/test_mcp_oauth_refresh.py

@@ -232,6 +232,35 @@ class TestRefreshFailureClassification:
         # Token survives a transient failure — no cluster-wide revoke; self-heals.
         assert state.mcp_token_store.get_user_token("user-1", "srv-oauth") is not None
 
+    def test_oversized_error_body_stays_transient_for_oauth_user(
+        self, storage: SQLiteBackend
+    ) -> None:
+        """Review finding: routing refresh_token() through the shared
+        _hardened_token_post must NOT change the oauth_user oversized-error
+        behavior. On main an over-cap error body was TRANSIENT (default class,
+        token kept, retryable forever); the OBO-only status-based escalation must
+        not leak onto oauth_user, or a large upstream error could escalate a
+        pre-existing consent to an unexpected re-consent. oauth_user keeps
+        TRANSIENT; the token survives."""
+        _seed_server(storage)
+        client = MagicMock(spec=httpx.AsyncClient)
+        client.get = AsyncMock(return_value=_mk_response(200, _good_as_metadata_doc()))
+        # 401 with an error body over the 64KB cap.
+        client.post = AsyncMock(
+            return_value=_mk_response(401, {"error": "invalid_grant", "pad": "x" * (70 * 1024)})
+        )
+        state = _make_app_state(storage, http_client=client)
+        _seed_token(state, expires_in_seconds=-1000)
+
+        result = self._lookup(state)
+
+        assert result.kind == "refresh_failed_transient"
+        # Kept (TRANSIENT), not revoked — and the ambiguous streak did not advance.
+        assert state.mcp_token_store.get_user_token("user-1", "srv-oauth") is not None
+        from turnstone.core.mcp_oauth import _refresh_backoff_state
+
+        assert _refresh_backoff_state(state, "user-1", "srv-oauth").ambiguous_streak == 0
+
     def test_permanent_invalid_grant_revokes(self, storage: SQLiteBackend) -> None:
         """Contrast: 400 invalid_grant IS permanent — deletion is correct and the
         eventual fix MUST preserve it."""
@@ -247,6 +276,32 @@ class TestRefreshFailureClassification:
         assert result.kind == "refresh_failed"
         assert state.mcp_token_store.get_user_token("user-1", "srv-oauth") is None
 
+    def test_permanent_revoke_audits_even_when_row_concurrently_deleted(
+        self, storage: SQLiteBackend
+    ) -> None:
+        """Review finding: gating the token_revoked audit on delete-returned-True
+        (the obo spam fix) must NOT suppress the oauth_user audit when a
+        concurrent admin/user revoke deletes the row first. For oauth_user a
+        refresh failure means a grant EXISTED (a real revocation), so the audit
+        fires even on an empty delete — an operator's SIEM must not miss it."""
+        from unittest.mock import patch
+
+        _seed_server(storage)
+        client = MagicMock(spec=httpx.AsyncClient)
+        client.get = AsyncMock(return_value=_mk_response(200, _good_as_metadata_doc()))
+        client.post = AsyncMock(return_value=_mk_response(400, {"error": "invalid_grant"}))
+        state = _make_app_state(storage, http_client=client)
+        _seed_token(state, expires_in_seconds=-1000)
+
+        # Simulate a concurrent external revoke: the dispatch-side delete finds
+        # the row already gone (returns False).
+        with patch.object(state.mcp_token_store, "delete_user_token", return_value=False):
+            result = self._lookup(state)
+
+        assert result.kind == "refresh_failed"
+        events = storage.list_audit_events(action="mcp_server.oauth.token_revoked")
+        assert len(events) == 1  # audited despite the empty delete
+
     def test_400_invalid_client_keeps_token(self, storage: SQLiteBackend) -> None:
         """A 400 ``invalid_client`` is operator-fixable, NOT a dead grant: keep
         the token. Pins the discriminator on the *error code*, not the 4xx

tests/test_oidc.py

@@ -2524,21 +2524,37 @@ class TestRuntimeRediscovery:
         return SimpleNamespace(oidc_config=cfg)
 
     def test_rediscover_swaps_enabled_config_on_success(self):
+        """Drives the REAL discover_oidc through a mocked HTTP discovery GET
+        (NOT a mock of discover_oidc itself): discover_oidc preserves the input
+        config's ``enabled`` on success and only clears it on failure, so a
+        probe started from the disabled boot config must first force enabled=True
+        or the recovered config never installs. An earlier version of this test
+        mocked discover_oidc to return enabled=True and so masked exactly that
+        dead-code bug."""
         from turnstone.core.oidc import maybe_rediscover_oidc
 
-        state = self._disabled_retryable_state()
-        recovered = dataclasses.replace(
-            state.oidc_config,
-            enabled=True,
-            token_endpoint="https://idp.example.com/token",
-            authorization_endpoint="https://idp.example.com/authorize",
-            jwks_uri="https://idp.example.com/jwks",
-        )
-        with patch(
-            "turnstone.core.oidc.discover_oidc", new=AsyncMock(return_value=recovered)
-        ) as disc:
-            asyncio.run(maybe_rediscover_oidc(state))
-        disc.assert_awaited_once()
+        state = self._disabled_retryable_state()  # issuer=https://idp.example.com
+        discovery_doc = {
+            "authorization_endpoint": "https://idp.example.com/authorize",
+            "token_endpoint": "https://idp.example.com/token",
+            "userinfo_endpoint": "https://idp.example.com/userinfo",
+            "jwks_uri": "https://idp.example.com/.well-known/jwks.json",
+        }
+        mock_response = MagicMock()
+        mock_response.json.return_value = discovery_doc
+        mock_response.raise_for_status = MagicMock()
+
+        async def _run():
+            client = _mock_async_client(lambda url: _async_return(mock_response))
+            with (
+                patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]),
+                patch("httpx.AsyncClient", return_value=client),
+            ):
+                await maybe_rediscover_oidc(state)
+
+        asyncio.run(_run())
+
+        # The real discover_oidc succeeded and the recovered config was installed.
         assert state.oidc_config.enabled is True
         assert state.oidc_config.token_endpoint == "https://idp.example.com/token"
         # The healed config no longer advertises a retryable failure.

turnstone/console/server.py

@@ -9875,7 +9875,17 @@ def _enforce_oauth_obo_requirements(
     if getattr(request.app.state, "mcp_token_store", None) is None:
         return JSONResponse({"error": _OAUTH_TOKEN_STORE_503_MSG}, status_code=503)
     oidc_config = getattr(request.app.state, "oidc_config", None)
-    if oidc_config is None or not getattr(oidc_config, "enabled", False):
+    # Accept a config that is enabled OR merely transiently un-discovered
+    # (``discovery_retryable`` — the IdP was unreachable at this process's boot).
+    # OIDC is still CONFIGURED there (issuer set); rejecting it would make every
+    # oauth_obo server un-editable/un-disable-able on the console — which never
+    # runs the runtime rediscovery that heals server nodes — until a manual
+    # restart. Reject only a genuinely absent / operator-disabled OIDC (neither
+    # flag set: no issuer configured).
+    oidc_configured = oidc_config is not None and (
+        getattr(oidc_config, "enabled", False) or getattr(oidc_config, "discovery_retryable", False)
+    )
+    if not oidc_configured:
         return JSONResponse(
             {
                 "error": (
@@ -10315,6 +10325,12 @@ async def admin_create_mcp_server(request: Request) -> JSONResponse:
     # Helper returns None when key is absent — fall back to the default.
     auth_type = auth_type_value if auth_type_value is not None else "static"
 
+    # Clean the OAuth text fields ONCE — reused by both the write-time validator
+    # and the persisted ``oauth_cols`` below, so the validated and persisted
+    # values (and the 2048 max_length) can't silently diverge.
+    clean_audience = _clean_oauth_text(body.get("oauth_audience"), max_length=2048)
+    clean_scopes = _clean_oauth_text(body.get("oauth_scopes"))
+
     # sec-1: oauth_user/oauth_obo must use https:// (loopback http allowed for dev).
     err_resp = _enforce_oauth_user_https(auth_type, str(body.get("url", "")).strip())
     if err_resp is not None:
@@ -10324,8 +10340,8 @@ async def admin_create_mcp_server(request: Request) -> JSONResponse:
     err_resp = _enforce_oauth_obo_requirements(
         request,
         auth_type,
-        audience=_clean_oauth_text(body.get("oauth_audience"), max_length=2048),
-        scopes=_clean_oauth_text(body.get("oauth_scopes")),
+        audience=clean_audience,
+        scopes=clean_scopes,
     )
     if err_resp is not None:
         return err_resp
@@ -10363,19 +10379,19 @@ async def admin_create_mcp_server(request: Request) -> JSONResponse:
     # handler applies on a flip: persist only the OAuth columns the target
     # auth_type actually uses, so a row created as oauth_obo can't carry a
     # client_id / registration_mode / AS-URL straight into a later oauth_user
-    # flip. (``oauth_as_issuer_cached`` is discovery-owned and not a create
-    # parameter, so it is dropped from the cleared set here.)
+    # flip. ``oauth_as_issuer_cached`` (nulled by the cleared set for a
+    # non-oauth_user type) IS a valid create parameter, so it passes through
+    # to ``create_mcp_server`` as its default None.
     oauth_cols: dict[str, Any] = {
         "oauth_client_id": _clean_oauth_text(body.get("oauth_client_id")),
-        "oauth_scopes": _clean_oauth_text(body.get("oauth_scopes")),
-        "oauth_audience": _clean_oauth_text(body.get("oauth_audience"), max_length=2048),
+        "oauth_scopes": clean_scopes,
+        "oauth_audience": clean_audience,
         "oauth_registration_mode": _clean_oauth_text(body.get("oauth_registration_mode")),
         "oauth_authorization_server_url": _clean_oauth_text(
             body.get("oauth_authorization_server_url"), max_length=2048
         ),
     }
     oauth_cols.update(_oauth_columns_to_clear(auth_type))
-    oauth_cols.pop("oauth_as_issuer_cached", None)
 
     storage.create_mcp_server(
         server_id=server_id,

turnstone/core/mcp_client.py

@@ -2535,8 +2535,8 @@ class MCPClientManager:
         """
         if not user_id or self._loop is None:
             return
-        if not self._pool_server_names:
-            return
+        if not self._oauth_user_server_names and not self._obo_server_names:
+            return  # no pool-backed servers — nothing to prime (no set allocation)
         if self._app_state is None or self._storage is None:
             return
         # run_coroutine_threadsafe keeps the task referenced by the loop while
@@ -5705,6 +5705,14 @@ class MCPClientManager:
         an obo pending row — written when the credential was missing — would
         persist forever after the user re-logs in.
 
+        Deliberately NOT gated to oauth_obo: for oauth_user it also clears the
+        dispatch-time transient badges written by
+        :meth:`_record_pending_consent_best_effort` (non-interactive runs), which
+        the sweep's ``_token_sweep_warned``-keyed clear never touches for a pair
+        it didn't proactively warn. The TTL map below bounds the cost to one
+        DELETE per pair per window, so the extra oauth_user coverage is close to
+        free rather than redundant.
+
         Deduped via the ``_pending_consent_cleared`` TTL map so it is NOT an
         unconditional per-call SQL write: after a DB-confirmed DELETE the pair
         is skipped for ``_PENDING_CONSENT_CLEAR_TTL_SECONDS``, then the DELETE
@@ -5755,13 +5763,17 @@ class MCPClientManager:
         """
         if len(self._pending_consent_cleared) < _PENDING_CONSENT_CLEARED_MAX:
             return
-        entries = list(self._pending_consent_cleared.items())
-        expired = [k for k, t in entries if now - t >= _PENDING_CONSENT_CLEAR_TTL_SECONDS]
-        for k in expired:
-            self._pending_consent_cleared.pop(k, None)
+        # Pass 1: drop expired entries (snapshot the items so concurrent mutation
+        # can't raise mid-iteration; ``pop`` tolerates an already-removed key).
+        for k, t in list(self._pending_consent_cleared.items()):
+            if now - t >= _PENDING_CONSENT_CLEAR_TTL_SECONDS:
+                self._pending_consent_cleared.pop(k, None)
+        # Pass 2: only if still over the cap, drop the oldest half of what
+        # REMAINS. Re-reading the map here means we sort just the live survivors
+        # — never re-targeting keys pass 1 already removed.
         if len(self._pending_consent_cleared) >= _PENDING_CONSENT_CLEARED_MAX:
-            by_age = sorted(entries, key=lambda kv: kv[1])
-            for k, _ in by_age[: len(by_age) // 2]:
+            survivors = sorted(self._pending_consent_cleared.items(), key=lambda kv: kv[1])
+            for k, _ in survivors[: len(survivors) // 2]:
                 self._pending_consent_cleared.pop(k, None)
 
     def _dispatch_pool_sync(

turnstone/core/mcp_oauth.py

@@ -901,6 +901,7 @@ async def _hardened_token_post(
     http_client: httpx.AsyncClient | None,
     request_label: str,
     endpoint_label: str,
+    classify_oversized_by_status: bool = False,
 ) -> dict[str, Any]:
     """POST one token-grant request with the shared hardening skeleton.
 
@@ -919,6 +920,14 @@ async def _hardened_token_post(
     ``None`` a transient per-request client is used — the OBO mint path runs
     on the MCP loop and deliberately passes ``None`` (see
     :func:`get_obo_access_token_classified`).
+
+    ``classify_oversized_by_status`` controls how an OVER-sized error body is
+    classified. The oauth_user refresh path keeps the default (``False`` →
+    TRANSIENT), byte-identical to the pre-refactor behavior, so a large upstream
+    error can never escalate a pre-existing consent to re-consent. The OBO legs
+    pass ``True`` so an over-sized client-error body is AMBIGUOUS (it can't read
+    the body to pin PERMANENT without defeating the guard) and still escalates
+    to the honest re-login/admin remedy instead of looping "please retry".
     """
     try:
         if http_client is not None:
@@ -930,15 +939,9 @@ async def _hardened_token_post(
         raise MCPOAuthRefreshFailed(f"{request_label} request failed: {exc}") from exc
 
     if len(resp.content) > _MAX_TOKEN_BODY_BYTES:
-        # Classify by STATUS without reading the over-sized body (reading it to
-        # pin PERMANENT would defeat the guard). A client-error status — a likely
-        # dead grant — becomes AMBIGUOUS so it still escalates to re-consent after
-        # a streak, rather than the default TRANSIENT that would loop "please
-        # retry" forever on a permanently-dead grant whose error body happened to
-        # exceed the cap.
         oversized_class = (
             _RefreshFailureClass.AMBIGUOUS
-            if resp.status_code in (400, 401, 403)
+            if classify_oversized_by_status and resp.status_code in (400, 401, 403)
             else _RefreshFailureClass.TRANSIENT
         )
         raise MCPOAuthRefreshFailed(
@@ -1544,6 +1547,7 @@ async def _revoke_after_refresh_failure(
     server_id_for_audit: str,
     *,
     reason: str,
+    audit_when_absent: bool = True,
 ) -> TokenLookupResult:
     """Delete the stored token, emit a ``token_revoked`` audit, and drop locks.
 
@@ -1553,19 +1557,24 @@ async def _revoke_after_refresh_failure(
     entry exists — and returns ``refresh_failed`` so the dispatcher surfaces
     re-consent.
 
-    The ``token_revoked`` audit is emitted ONLY when a row was actually deleted.
-    This matters for oauth_obo: its shared credential survives a per-server
-    revoke, so after the first permanent rejection deletes the cache row, every
-    later dispatch/prime past the cooldown re-runs the doomed redemption and
-    lands here again with nothing to delete — auditing unconditionally would
-    append a fresh "token_revoked" row for a token that no longer exists on
-    every cycle, forever (misleading forensics + alert fatigue for operators
-    who alert on token_revoked). oauth_user cannot hit the empty-delete case (a
-    revoke there also removes the only refresh source, so no recurrence), so
-    this gate is a no-op for it.
+    ``audit_when_absent`` controls the audit when ``delete_user_token`` finds no
+    row:
+
+    - oauth_user (default ``True``): reaching a refresh-grant failure means a
+      grant EXISTED (you cannot refresh without a token), so a real grant died —
+      audit it even if a concurrent admin/user revoke deleted the row first, so
+      an operator's SIEM never misses the AS-rejected-the-grant signal. This is
+      the pre-refactor behavior (the audit was unconditional on main).
+    - oauth_obo (``False``): a mint needs no pre-existing cache row, so a
+      permanent rejection on a server the user never successfully minted for
+      lands here with nothing to delete; its shared credential also survives, so
+      every later dispatch/prime past the cooldown re-runs the doomed redemption
+      and returns here again. Auditing those would append a "token_revoked" row
+      for a token that never existed, forever — so obo audits only a real
+      deletion.
     """
     deleted = await asyncio.to_thread(token_store.delete_user_token, user_id, server_name)
-    if deleted:
+    if deleted or audit_when_absent:
         await _audit_event(
             app_state,
             server_id=server_id_for_audit,
@@ -1728,6 +1737,11 @@ async def _handle_refresh_failure(
             server_name,
             server_id_for_audit,
             reason=permanent_reason,
+            # ``arm_cooldown_on_permanent`` marks the obo path (shared credential
+            # survives the revoke); there, audit only a real deletion to avoid
+            # revocation rows for tokens that never existed. oauth_user audits
+            # unconditionally (a refresh failure means a grant existed).
+            audit_when_absent=not arm_cooldown_on_permanent,
         )
         if arm_cooldown_on_permanent:
             # _revoke_after_refresh_failure cleared the backoff; re-arm the
@@ -1763,6 +1777,7 @@ async def _handle_refresh_failure(
                     server_name,
                     server_id_for_audit,
                     reason=escalation_reason,
+                    audit_when_absent=not arm_cooldown_on_permanent,
                 )
                 if arm_cooldown_on_permanent:
                     # Same shared-credential backstop as the PERMANENT branch: an
@@ -2067,6 +2082,10 @@ async def _obo_token_post(
         http_client=http_client,
         request_label=label,
         endpoint_label=label,
+        # OBO: an over-sized client-error body escalates (AMBIGUOUS) rather than
+        # looping "please retry" — see _hardened_token_post. (oauth_user keeps
+        # the TRANSIENT default.)
+        classify_oversized_by_status=True,
     )
 
 

turnstone/core/oidc.py

@@ -730,11 +730,21 @@ async def maybe_rediscover_oidc(app_state: Any) -> None:
         return
     try:
         now = time.monotonic()
-        last = float(getattr(app_state, "oidc_rediscover_last", 0.0))
-        if now - last < _REDISCOVER_COOLDOWN_SECONDS:
+        # ``None`` (not 0.0) means "never probed" — otherwise the first probe
+        # within ~60s of the monotonic reference (host boot) would be suppressed
+        # by the cooldown against a phantom probe at time 0.
+        last = getattr(app_state, "oidc_rediscover_last", None)
+        if last is not None and now - float(last) < _REDISCOVER_COOLDOWN_SECONDS:
             return
         app_state.oidc_rediscover_last = now
-        fresh = await discover_oidc(cfg)
+        # Probe with enabled=True FORCED ON: discover_oidc PRESERVES the input's
+        # ``enabled`` on success (only ``load_oidc_config`` ever sets it True) and
+        # sets it False on any failure. Passing the disabled boot config verbatim
+        # would make a SUCCESSFUL rediscovery still return enabled=False, so the
+        # swap below would be unreachable and the feature inert. Forcing it True
+        # up front makes ``fresh.enabled`` a reliable success signal (a real
+        # failure clears it back to False).
+        fresh = await discover_oidc(dataclasses.replace(cfg, enabled=True))
         if not fresh.enabled:
             return  # still failing — next probe after the cooldown lapses
         app_state.oidc_config = dataclasses.replace(fresh, discovery_retryable=False)