test(e2e): cover MCP OAuth SSO and cold restart persistence

tests/e2e/CLAUDE.md

@@ -14,7 +14,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
 - `quota_management/` - quota enforcement and accounting, one subfolder per behavior: `ratelimit/` (rpm/tpm blocks, window reset, pacing headers on live traffic), `budgets/` (budget definition, enforcement, and reset windows: key, team, tag, soft, multi-window), and `spend_tracking/` (spend logging and cost attribution on `/spend/*`)
 - `management/` - key/team/user/organization management routes: create/update/delete persistence via the info routes, team membership, and llm-only-key route denials (API surface; not Playwright)
 - `a2a/` - the A2A (agent-to-agent) surface: admin registration via `/v1/agents`, proxy-fronted card discovery at `/.well-known/agent-card.json`, and JSON-RPC `message/send` invocation, driving agents backed by the litellm completion bridge (a real provider) and asserting protocol-version normalization (0.3 vs 1.0)
-- `mcp/` - the MCP server surface over api_key auth against the real Datadog remote MCP server (see "MCP suite: real Datadog only" below); plus the gateway-managed OAuth (authorization_code) path exercised through `/chat/completions`, the one behavior Datadog's static-header auth cannot reach, seeding the per-user upstream token via the interactive authorize dance driven with the mcp SDK's own OAuth client (headless-browser consent from a saved session) and asserting the completion lists and executes the server's tools with the stored per-user token
+- `mcp/` - the MCP server surface over api_key auth against the real Datadog remote MCP server (see "MCP suite: real Datadog only" below); plus the gateway-managed OAuth (authorization_code) path exercised through `/chat/completions` in `test_mcp_chat_completion_oauth_e2e.py` and direct MCP protocol operations in `test_mcp_oauth_happy_path_e2e.py`, the one behavior Datadog's static-header auth cannot reach, seeding the per-user upstream token via the interactive authorize dance driven with the mcp SDK's own OAuth client (headless-browser consent from a saved session) and asserting the completion or protocol call lists and executes the server's tools with the stored per-user token
 - `logging/` - logging-integration delivery (datadog and friends)
 - `security/` - secret handling and log-leak protection
 - `router/` - routing and reliability behavior (fallbacks, cooldowns) plus the memory regression test (`test_reliability_memory_e2e.py`: a few hundred failing requests with retries and fallbacks must not grow proxy RSS past a fixed budget nor store a request snapshot past a fixed size, the release-gate check for the v1.100.0 retry-breadcrumb leak)
@@ -26,14 +26,14 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
 
 ## MCP suite: real Datadog only
 
-Every test under `tests/e2e/mcp/` must exercise the proxy against the real Datadog remote MCP server. Do not add a compose service, FastMCP fixture, mock upstream, or any other fake MCP host for this suite
+Every test under `tests/e2e/mcp/` must exercise the proxy against the real Datadog remote MCP server, except the two Linear OAuth tests `test_mcp_chat_completion_oauth_e2e.py` and `test_mcp_oauth_happy_path_e2e.py`. Do not add a compose service, FastMCP fixture, mock upstream, or any other fake MCP host for this suite
 
 - Register via `register_datadog_mcp` in `tests/e2e/mcp/datadog_mcp.py` (or extend that helper if you need a different `toolsets=` / `allowed_tools` slice of the same Datadog endpoint). That posts `/v1/mcp/server` with `url=datadog_mcp_url(...)` and static headers `DD-API-KEY` / `DD-APPLICATION-KEY` from the process env
 - Auth is Datadog's documented CI/header path, not a browser OAuth authorize/token dance. Hard-fail when `DD_API_KEY` or `DD_APP_KEY` is missing (`assert_dd_mcp_creds`); never skip for a missing fake upstream
 - Prefer calling real Datadog tools that prove the product path (e.g. `search_datadog_logs` for list/call and permission denials). Seed a unique marker (`e2e-datadog-mcp-*`) in a chat completion when you need a log the tool can find; dual-read with `dd_logs` from conftest when delivery matters
 - Delete the MCP server (and any keys) through `resources.defer` the same way every other suite tears down
 - If a new MCP behavior cannot be covered with Datadog's tool surface, say so in the PR and get agreement before inventing another upstream; the default is always Datadog
-- The one standing exception is `test_mcp_chat_completion_oauth_e2e.py`. Datadog authenticates with the static `DD-API-KEY` / `DD-APPLICATION-KEY` headers and exposes no authorize/token dance at all, so it cannot exercise gateway-managed OAuth or per-user token seeding in any form. That test drives a real Linear MCP server instead; it is still a real remote upstream, so the no-mock, no-fixture rule above holds unchanged
+- The two standing exceptions are `test_mcp_chat_completion_oauth_e2e.py` and `test_mcp_oauth_happy_path_e2e.py`. Datadog authenticates with the static `DD-API-KEY` / `DD-APPLICATION-KEY` headers and exposes no authorize/token dance at all, so these tests drive a real Linear MCP server instead; they are still real remote upstreams, so the no-mock, no-fixture rule above holds unchanged
 
 ## Lay the pattern down in a class
 
@@ -152,7 +152,7 @@ MCPs - endpoint features with the protocol op as the variant
 mcp.<operation>.<auth_family>.<assertion>
   operation   : list_tools | call_tool | list_resources | read_resource | list_prompts | get_prompt
   auth_family : none | api_key | bearer | oauth
-  assertion   : succeeds | denied_without_permission
+  assertion   : succeeds | denied_without_permission | persists_across_processes
   e.g.  mcp.call_tool.oauth.succeeds
 ```
 

tests/e2e/conftest.py

@@ -28,6 +28,7 @@
     FIXTURE_DIR,
     FIXTURE_MODE_RAW,
     MANAGED_FILES_OPT_IN_ENV,
+    MCP_OAUTH_LIVE_OPT_IN_ENV,
     PROMPT_CACHING_OPT_IN_ENV,
     PROXY_BASE_URL,
     REDIS_CHAOS_OPT_IN_ENV,
@@ -56,6 +57,7 @@
         "prompt_caching_stack": PROMPT_CACHING_OPT_IN_ENV,
         "redis_chaos": REDIS_CHAOS_OPT_IN_ENV,
         "cli_determinism": CLI_DETERMINISM_OPT_IN_ENV,
+        "mcp_oauth_live": MCP_OAUTH_LIVE_OPT_IN_ENV,
     }
 )
 
@@ -132,6 +134,11 @@ def pytest_configure(config: pytest.Config) -> None:
         "redis_chaos: load test that pauses the proxy's Redis outright mid-run; needs a proxy booted from "
         "gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set",
     )
+    config.addinivalue_line(
+        "markers",
+        "mcp_oauth_live: real Linear OAuth consent via a captured browser session; deselected unless "
+        "E2E_MCP_OAUTH_LIVE is set",
+    )
 
 
 def pytest_sessionstart(session: pytest.Session) -> None:

tests/e2e/coverage_registry/mcp.yaml

@@ -71,6 +71,14 @@
   assertions: [succeeds]
   source: "db.py user_oauth_credential lookup"
   rationale: OAuth2 token passthrough; per-user credential storage
+- id: mcp.call_tool.oauth.persists_across_processes
+  module: mcp
+  tier: P1
+  operation: call_tool
+  auth_family: oauth
+  assertions: [persists_across_processes]
+  source: "outbound_credentials/per_user_oauth_store.py V2PerUserTokenStore"
+  rationale: Stored per-user token is resolved by a gateway process that did not run the consent
 - id: mcp.list_tools.none.succeeds
   module: mcp
   tier: P1

tests/e2e/e2e_config.py

@@ -52,6 +52,7 @@ def parse_replica_urls(raw: str, fallback: str) -> tuple[str, ...]:
 
 LINEAR_MCP_URL = os.environ.get("E2E_LINEAR_MCP_URL", "https://mcp.linear.app/mcp")
 LINEAR_STORAGE_STATE = os.environ.get("E2E_LINEAR_STORAGE_STATE", "")
+LINEAR_READONLY_TOOL: Final = "list_teams"
 
 # Jaeger query API of the compose stack's OTEL trace destination (the `jaeger`
 # service in docker-compose.yml maps it to host 16686). Trace-completeness tests
@@ -144,6 +145,7 @@ def parse_replica_urls(raw: str, fallback: str) -> tuple[str, ...]:
 PROMPT_CACHING_OPT_IN_ENV = "E2E_PROMPT_CACHING_STACK"
 REDIS_CHAOS_OPT_IN_ENV = "E2E_REDIS_CHAOS"
 CLI_DETERMINISM_OPT_IN_ENV = "E2E_CLI_DETERMINISM"
+MCP_OAUTH_LIVE_OPT_IN_ENV: Final = "E2E_MCP_OAUTH_LIVE"
 ANOMALY_SESSIONS = int(os.environ.get("E2E_ANOMALY_SESSIONS", "6"))
 ANOMALY_TURNS_PER_SESSION = int(os.environ.get("E2E_ANOMALY_TURNS_PER_SESSION", "6"))
 ANOMALY_TURN_ATTEMPTS = int(os.environ.get("E2E_ANOMALY_TURN_ATTEMPTS", "3"))

tests/e2e/mcp/oauth_chat_client.py

@@ -18,20 +18,28 @@
 import re
 import time
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
 from urllib.parse import parse_qsl
 
 import httpx
 import pytest
+from e2e_config import PROXY_BASE_URL, REQUEST_TIMEOUT
+from e2e_http import AuthHeaders, NoBody, unwrap
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider
 from mcp.client.streamable_http import streamable_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
-
-from e2e_config import PROXY_BASE_URL, REQUEST_TIMEOUT
+from mcp.types import TextContent
+from models import (
+    ChatBody,
+    ChatResponse,
+    McpOauthUserCredentialStatus,
+    McpServerCreateBody,
+    McpServerInfo,
+    McpServerUserCredentialListResponse,
+    McpServerUserCredentialRow,
+)
 from proxy_client import ProxyClient
-from e2e_http import AuthHeaders, NoBody, unwrap
-from models import ChatBody, ChatResponse, McpServerCreateBody, McpServerInfo
 
 if TYPE_CHECKING:
     from playwright.async_api import Route
@@ -44,8 +52,8 @@
 BROWSER_CONSENT_TIMEOUT = 60.0
 
 
-def _mcp_url(alias: str) -> str:
-    return f"{PROXY_BASE_URL}/{alias}/mcp"
+def _mcp_url(alias: str, base_url: str = PROXY_BASE_URL) -> str:
+    return f"{base_url.rstrip('/')}/{alias}/mcp"
 
 
 class InMemoryTokenStorage:
@@ -88,7 +96,7 @@ def _note_request(request: object) -> None:
         if url.startswith(OAUTH_CLIENT_REDIRECT_URI) and "url" not in captured:
             captured["url"] = url
 
-    async def _swallow_redirect(route: "Route") -> None:
+    async def _swallow_redirect(route: Route) -> None:
         await route.fulfill(status=200, content_type="text/plain", body="ok")
 
     async with async_playwright() as playwright:
@@ -128,17 +136,23 @@ async def _swallow_redirect(route: "Route") -> None:
     return params["code"], params.get("state")
 
 
-def _oauth_provider(url: str, storage: InMemoryTokenStorage, storage_state_path: str) -> OAuthClientProvider:
+def _oauth_provider(url: str, storage: InMemoryTokenStorage, storage_state_path: str | None) -> OAuthClientProvider:
     """The SDK's real OAuth machinery (RFC 9728/8414 discovery, RFC 7591 DCR,
     PKCE, token exchange) with the browser leg driven by Playwright against the
     upstream's consent screen."""
     code_holder: dict[str, str | None] = {}  # mutable-ok: hand-off between the two SDK callbacks
 
-    async def redirect_handler(authorize_url: str) -> None:
+    async def _reject_redirect(_: str) -> None:
+        raise AssertionError("gateway demanded a fresh upstream consent; stored per-user token was not reused")
+
+    async def _follow_redirect(authorize_url: str) -> None:
+        assert storage_state_path is not None
         code, state = await _browser_follow_authorize(authorize_url, storage_state_path)
         code_holder["code"] = code
         code_holder["state"] = state
 
+    redirect_handler: Final = _reject_redirect if storage_state_path is None else _follow_redirect
+
     async def callback_handler() -> tuple[str, str | None]:
         code = code_holder.get("code")
         assert code is not None, "callback_handler ran before the authorize redirect completed"
@@ -167,24 +181,38 @@ class _HeaderInjectingTransport(httpx.AsyncBaseTransport):
     store the upstream token for from the key on the token exchange, exactly
     like a production MCP host configured with a LiteLLM key header."""
 
-    def __init__(self, inner: httpx.AsyncBaseTransport, headers: dict[str, str]) -> None:
+    def __init__(self, inner: httpx.AsyncBaseTransport, headers: dict[str, str], gateway_url: str) -> None:
         self._inner = inner
         self._headers = headers
+        self._gateway_url = httpx.URL(gateway_url)
+
+    @staticmethod
+    def _port(url: httpx.URL) -> int | None:
+        if url.port is not None:
+            return url.port
+        return {"http": 80, "https": 443}.get(url.scheme)
 
     async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
-        for name, value in self._headers.items():
-            if name not in request.headers:
-                request.headers[name] = value
+        same_origin: Final = (
+            request.url.scheme == self._gateway_url.scheme
+            and request.url.host == self._gateway_url.host
+            and self._port(request.url) == self._port(self._gateway_url)
+        )
+        if same_origin:
+            for name, value in self._headers.items():
+                if name not in request.headers:
+                    request.headers[name] = value
         return await self._inner.handle_async_request(request)
 
 
-def _oauth_http_client(headers: dict[str, str], auth: OAuthClientProvider) -> httpx.AsyncClient:
+def _oauth_http_client(
+    headers: dict[str, str], auth: OAuthClientProvider, gateway_url: str = PROXY_BASE_URL
+) -> httpx.AsyncClient:
     return httpx.AsyncClient(
-        headers=headers,
         auth=auth,
         timeout=httpx.Timeout(REQUEST_TIMEOUT),
         follow_redirects=True,
-        transport=_HeaderInjectingTransport(httpx.AsyncHTTPTransport(), headers),
+        transport=_HeaderInjectingTransport(httpx.AsyncHTTPTransport(), headers, gateway_url),
     )
 
 
@@ -199,6 +227,38 @@ async def _seed_via_dance(
                 return tuple(sorted(tool.name for tool in listed.tools))
 
 
+@dataclass(frozen=True, slots=True)
+class OauthToolRun:
+    tools: tuple[str, ...]
+    is_error: bool
+    text: str
+
+
+async def _list_and_call(
+    url: str,
+    headers: dict[str, str],
+    storage: InMemoryTokenStorage,
+    storage_state_path: str | None,
+    tool: str,
+    arguments: dict[str, str],
+    gateway_url: str = PROXY_BASE_URL,
+) -> OauthToolRun:
+    async with _oauth_http_client(
+        headers, _oauth_provider(url, storage, storage_state_path), gateway_url
+    ) as http_client:
+        async with streamable_http_client(url, http_client=http_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                listed: Final = await session.list_tools()
+                result: Final = await session.call_tool(tool, arguments)
+                text: Final = "".join(content.text for content in result.content if isinstance(content, TextContent))
+                return OauthToolRun(
+                    tools=tuple(sorted(tool_item.name for tool_item in listed.tools)),
+                    is_error=result.isError,
+                    text=text,
+                )
+
+
 @dataclass(frozen=True, slots=True)
 class ChatMcpClient:
     proxy: ProxyClient
@@ -252,6 +312,60 @@ def seed_user_token(self, alias: str, key: str, storage_state_path: str) -> tupl
             f"last error: {last_error!r}"
         )
 
+    def list_and_call(
+        self,
+        alias: str,
+        headers: dict[str, str],
+        storage: InMemoryTokenStorage,
+        storage_state_path: str | None,
+        tool: str,
+        arguments: dict[str, str],
+        base_url: str = PROXY_BASE_URL,
+    ) -> OauthToolRun:
+        deadline: Final = time.monotonic() + self.proxy.poll_timeout
+        last_error: Exception | None = None
+        while time.monotonic() < deadline:
+            try:
+                return asyncio.run(
+                    _list_and_call(
+                        _mcp_url(alias, base_url),
+                        headers,
+                        storage,
+                        storage_state_path,
+                        tool,
+                        arguments,
+                        base_url,
+                    )
+                )
+            except AssertionError:
+                raise
+            except Exception as exc:  # noqa: BLE001 - retried to the deadline; the last error surfaces below
+                last_error = exc
+                time.sleep(self.proxy.poll_interval)
+        pytest.fail(
+            f"list and call for {alias!r} never completed within {self.proxy.poll_timeout}s; last error: {last_error!r}"
+        )
+
+    def server_user_credentials(self, server_id: str) -> tuple[McpServerUserCredentialRow, ...]:
+        return unwrap(
+            self.proxy.transport.get(
+                f"/v1/mcp/server/{server_id}/user-credentials",
+                headers=self.proxy.transport.master,
+                params=NoBody(),
+                response_type=McpServerUserCredentialListResponse,
+            )
+        ).root
+
+    def revoke_user_token(self, server_id: str, headers: AuthHeaders) -> None:
+        _ = unwrap(
+            self.proxy.transport.delete(
+                f"/v1/mcp/server/{server_id}/oauth-user-credential",
+                headers=headers,
+                json=NoBody(),
+                response_type=McpOauthUserCredentialStatus,
+            )
+        )
+
     def chat_with_mcp(self, headers: AuthHeaders, body: ChatBody) -> ChatResponse:
         """POST /chat/completions carrying the LiteLLM key in `headers` (either
         ingress form) with an MCP server attached in `body.tools`. The gateway

tests/e2e/mcp/test_mcp_chat_completion_oauth_e2e.py

@@ -27,8 +27,13 @@
 import os
 
 import pytest
-
-from e2e_config import CHEAP_ANTHROPIC_MODEL, LINEAR_MCP_URL, LINEAR_STORAGE_STATE, unique_marker
+from e2e_config import (
+    CHEAP_ANTHROPIC_MODEL,
+    LINEAR_MCP_URL,
+    LINEAR_READONLY_TOOL,
+    LINEAR_STORAGE_STATE,
+    unique_marker,
+)
 from e2e_http import AuthHeaders
 from lifecycle import ResourceManager
 from models import ChatBody, ChatMessage, KeyGenerateBody, McpChatTool, McpServerCreateBody, ObjectPermission
@@ -50,10 +55,6 @@
     ),
 ]
 
-# Pinned from a live dance during verification (never guessed); the gateway
-# prefixes every upstream tool name with the server alias. list_teams is a
-# read-only Linear tool that takes no arguments and returns the caller's teams.
-LINEAR_READONLY_TOOL = "list_teams"
 LINEAR_PROMPT = "Use the list_teams tool to list my Linear teams, then reply with the name of one of them."