fix(mcp): oauth_obo credential lifecycle + pending-badge coverage (P1 per review)

- credential revocation (440): admin OIDC identity-unlink now deletes the
  captured IdP credential too (via delete_oidc_credential, previously
  zero callers), so a deprovisioned user stops minting — audited with
  obo_credential_revoked
- pending-consent badge gate (3539): new any_user_scoped_mcp_servers
  (oauth_user OR oauth_obo) replaces the oauth_user-only gate, so an
  obo-only install no longer short-circuits the badge to {pending: 0}
- pending-consent clear (5966): dispatch SUCCESS now clears the pending
  row (auth-blind _clear_pending_consent_sync) — the only clear path that
  covers obo, whose rows the token sweep (skips obo) and consent callback
  (obo never runs) would otherwise never clear
- test:50: strengthened the created-preservation assertion to plant a
  distinctly-past created via SQL so a reset is actually detectable

+4 tests (obo/user-scoped gate). NOTE: finding 1992 (orphan cache row on
concurrent delete-during-mint) accepted as bounded residual — the orphan
is a short-lived access-token cache row with NO refresh token, useless
without the deleted credential and self-expiring; a full fix needs FKs or
a delete-spanning lock. Tracked for follow-up.

Refs #551.

tests/test_mcp_pending_consent_storage.py

@@ -239,3 +239,41 @@ class TestInstallGate:
         )
         backend.update_mcp_server("srv-2", auth_type="oauth_user")
         assert backend.any_oauth_user_mcp_servers() is True
+
+    def test_any_user_scoped_true_for_obo_only_install(self, backend) -> None:
+        """#551: the pending-consent badge gate must fire for an oauth_obo-only
+        install — obo dispatch writes pending rows, so short-circuiting to
+        {pending: 0} would hide the re-login affordance. The oauth_user-only
+        gate stays False (obo is not oauth_user)."""
+        backend.create_mcp_server(
+            server_id="srv-obo",
+            name="obo-only",
+            transport="streamable-http",
+            command="",
+            args="[]",
+            url="https://example.com",
+            headers="{}",
+            env="{}",
+            auto_approve=False,
+            enabled=True,
+            created_by="admin",
+        )
+        backend.update_mcp_server("srv-obo", auth_type="oauth_obo")
+        assert backend.any_user_scoped_mcp_servers() is True
+        assert backend.any_oauth_user_mcp_servers() is False
+
+    def test_any_user_scoped_false_on_static_only(self, backend) -> None:
+        backend.create_mcp_server(
+            server_id="srv-s",
+            name="static-only",
+            transport="streamable-http",
+            command="",
+            args="[]",
+            url="https://example.com",
+            headers="{}",
+            env="{}",
+            auto_approve=False,
+            enabled=True,
+            created_by="admin",
+        )
+        assert backend.any_user_scoped_mcp_servers() is False

tests/test_oidc_credential_storage.py

@@ -39,15 +39,37 @@ class TestUpsertAndGet:
         assert backend.get_oidc_user_credential("u2", ISS) is None
 
     def test_upsert_replaces_on_conflict(self, backend) -> None:
-        """A fresh login must overwrite a stale credential; ``created`` survives."""
+        """A fresh login must overwrite a stale credential; ``created`` survives.
+
+        Plants a distinctly-past ``created`` via direct SQL so the assertion
+        actually detects a reset (comparing two upserts milliseconds apart would
+        pass at second granularity even if the on-conflict clause reset created).
+        """
+        import sqlalchemy as sa
+
+        from turnstone.core.storage._schema import oidc_user_credentials
+
         backend.upsert_oidc_user_credential("u1", ISS, refresh_token_ct=b"ct-old")
-        first = backend.get_oidc_user_credential("u1", ISS)
-        assert first is not None
+        planted = "2020-01-01T00:00:00"
+        with backend._engine.connect() as conn:
+            conn.execute(
+                sa.update(oidc_user_credentials)
+                .where(
+                    (oidc_user_credentials.c.user_id == "u1")
+                    & (oidc_user_credentials.c.issuer == ISS)
+                )
+                .values(created=planted)
+            )
+            conn.commit()
+
         backend.upsert_oidc_user_credential("u1", ISS, refresh_token_ct=b"ct-new")
         second = backend.get_oidc_user_credential("u1", ISS)
         assert second is not None
         assert second["refresh_token_ct"] == b"ct-new"
-        assert second["created"] == first["created"]
+        # created is PRESERVED across the replace (not reset to now).
+        assert second["created"] == planted
+        # last_refreshed, by contrast, advances off the planted-past value.
+        assert second["last_refreshed"] != planted
 
 
 class TestRotationWriteBack:

turnstone/console/server.py

@@ -6002,6 +6002,25 @@ async def admin_delete_oidc_identity(request: Request) -> JSONResponse:
 
     storage.delete_oidc_identity(issuer, subject)
 
+    # #551: unlinking the identity must also revoke the captured IdP refresh
+    # credential (keyed on (user_id, issuer)) — otherwise a deprovisioned user's
+    # scheduled/autonomous runs keep minting oauth_obo access tokens
+    # indefinitely (rotation write-back keeps the RT alive). Best-effort: a
+    # failure here must not leave the identity un-deleted.
+    credential_revoked = False
+    token_store = getattr(request.app.state, "mcp_token_store", None)
+    if token_store is not None:
+        try:
+            credential_revoked = bool(
+                token_store.delete_oidc_credential(identity["user_id"], issuer)
+            )
+        except Exception:
+            log.warning(
+                "admin.oidc_identity.credential_revoke_failed user=%s",
+                identity["user_id"],
+                exc_info=True,
+            )
+
     audit_uid, ip = _audit_context(request)
     record_audit(
         storage,
@@ -6009,7 +6028,7 @@ async def admin_delete_oidc_identity(request: Request) -> JSONResponse:
         "oidc_identity.delete",
         "oidc_identity",
         f"{issuer}:{subject}",
-        {"user_id": identity["user_id"]},
+        {"user_id": identity["user_id"], "obo_credential_revoked": credential_revoked},
         ip,
     )
 

turnstone/core/mcp_client.py

@@ -5609,6 +5609,29 @@ class MCPClientManager:
                 exc_info=True,
             )
 
+    def _clear_pending_consent_sync(self, user_id: str, server_name: str) -> None:
+        """Best-effort synchronous clear of a deferred-consent row on dispatch success.
+
+        Called from the sync dispatchers when a dispatch SUCCEEDS, so a stale
+        badge self-heals. This is the only clear path that covers oauth_obo: the
+        oauth_user clears live in the token sweep (which skips obo) and the
+        consent callback (which obo never runs), so without a success-side clear
+        an obo pending row — written when the credential was missing — would
+        persist forever after the user re-logs in. Delete is a no-op when no row
+        exists, so this is safe to call on every successful dispatch.
+        """
+        if self._storage is None:
+            return
+        try:
+            self._storage.delete_mcp_pending_consent(user_id, server_name)
+        except Exception:
+            log.debug(
+                "mcp_pool.pending_consent_clear_failed user=%s server=%s",
+                user_id,
+                server_name,
+                exc_info=True,
+            )
+
     def _dispatch_pool_sync(
         self,
         *,
@@ -5696,6 +5719,8 @@ class MCPClientManager:
                     user_id=user_id, server_name=server_name, result=result
                 )
             raise RuntimeError(result)
+        # Success clears any stale pending-consent badge (the obo self-heal path).
+        self._clear_pending_consent_sync(user_id, server_name)
         return result
 
     def _run_pool_dispatch_attempt(
@@ -5799,6 +5824,7 @@ class MCPClientManager:
                     user_id=user_id, server_name=server_name, result=result
                 )
             raise RuntimeError(result)
+        self._clear_pending_consent_sync(user_id, server_name)
         return result
 
     def _run_pool_dispatch_resource_attempt(
@@ -5890,6 +5916,7 @@ class MCPClientManager:
                     user_id=user_id, server_name=server_name, result=result
                 )
             raise RuntimeError(result)
+        self._clear_pending_consent_sync(user_id, server_name)
         return result
 
     def _run_pool_dispatch_prompt_attempt(

turnstone/core/mcp_oauth.py

@@ -3668,7 +3668,7 @@ async def _install_gate_passes(app_state: Any, storage: Any) -> bool:
         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))
+    value = bool(await asyncio.to_thread(storage.any_user_scoped_mcp_servers))
     app_state._mcp_install_gate_cache = (value, now)
     return value
 

turnstone/core/storage/_postgresql.py

@@ -5044,6 +5044,16 @@ class PostgreSQLBackend:
             ).scalar()
         return result is not None
 
+    def any_user_scoped_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.in_(("oauth_user", "oauth_obo")))
+                .limit(1)
+            ).scalar()
+        return result is not None
+
     # -- Model definitions -----------------------------------------------------
 
     def create_model_definition(

turnstone/core/storage/_protocol.py

@@ -2362,6 +2362,17 @@ class StorageBackend(Protocol):
         """
         ...
 
+    def any_user_scoped_mcp_servers(self) -> bool:
+        """Install-level gate for the pending-consent badge (issue #551).
+
+        Returns True iff at least one ``mcp_servers`` row is pool-backed
+        (``auth_type`` in ``oauth_user`` / ``oauth_obo``) — both write
+        ``mcp_pending_consent`` rows on a non-interactive dispatch failure, so
+        an oauth_obo-only install must NOT short-circuit the badge to
+        ``{pending: 0}`` (that would hide the re-login affordance).
+        """
+        ...
+
     def any_oauth_user_mcp_servers(self) -> bool:
         """Install-level gate for OAuth-MCP features.
 

turnstone/core/storage/_sqlite.py

@@ -5198,6 +5198,16 @@ class SQLiteBackend:
             ).scalar()
         return result is not None
 
+    def any_user_scoped_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.in_(("oauth_user", "oauth_obo")))
+                .limit(1)
+            ).scalar()
+        return result is not None
+
     # -- Model definitions -----------------------------------------------------
 
     def create_model_definition(