docs: plan typed auth and oauth login

2026-06-13 01:50:36 +07:00
parent 49a78a2808
commit c3e931e30c
4 changed files with 1604 additions and 16 deletions
@@ -0,0 +1,676 @@
+# OAuth Login Auth Records Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add a local/dev OAuth login command that uses provider profiles to create typed `oauth_refresh_token` auth records for sources such as Google Drive MCP.
+
+**Architecture:** OAuth login is admin/CLI setup, not runtime execution. Provider profiles describe how to start authorization-code + PKCE login. The resulting refresh token is saved as a typed auth record. Runtime source calls continue to consume auth records through source-owned binders from the previous slice.
+
+**Tech Stack:** Python 3.14, Typer, Authlib (new dependency if accepted), pytest, basedpyright, ruff, local callback or pasted redirect URL flow.
+
+---
+
+## Dependency On Previous Slice
+
+This plan assumes `docs/superpowers/plans/2026-06-13-typed-auth-records-and-mcp-binder.md` is complete:
+
+- `wf_api.auth.StoredAuthRecord`
+- `wf_api.auth.OAuthRefreshTokenAuth`
+- auth store can save typed records
+- MCP runtime can use typed records through `McpAuthBinder`
+
+Do not implement this plan first.
+
+## File Structure
+
+- Modify `src/wf_config/models.py`: add OAuth provider profile config models under a top-level auth config section.
+- Modify `src/wf_config/loader.py` only if config-relative loading needs provider profile defaults.
+- Create `src/wf_cli/oauth.py`: OAuth login flow helpers and provider profile DTOs for CLI use.
+- Modify `src/wf_cli/commands/auth_admin.py`: add `oauth-login` command.
+- Modify `src/wf_cli/context.py` only if CLI context must expose loaded workflow config auth providers.
+- Add tests:
+  - `tests/wf_config/test_config_models.py`
+  - `tests/wf_cli/test_auth_oauth_login.py`
+- Docs:
+  - `docs/wf_cli.md`
+  - `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
+  - `docs/current_roadmap.md`
+
+## Task 1: OAuth Provider Profile Config
+
+**Files:**
+- Modify: `src/wf_config/models.py`
+- Modify: `src/wf_config/__init__.py`
+- Test: `tests/wf_config/test_config_models.py`
+
+- [ ] **Step 1: Add failing config model test**
+
+Append:
+
+```python
+from wf_config import WorkflowConfigFile
+
+
+def test_workflow_config_parses_oauth_provider_profile() -> None:
+    config = WorkflowConfigFile.model_validate(
+        {
+            "auth": {
+                "providers": {
+                    "google": {
+                        "kind": "oauth_authorization_code_pkce",
+                        "auth_url": "https://accounts.google.com/o/oauth2/v2/auth",
+                        "token_url": "https://oauth2.googleapis.com/token",
+                        "client_id_env": "GOOGLE_OAUTH_CLIENT_ID",
+                        "client_secret_env": "GOOGLE_OAUTH_CLIENT_SECRET",
+                        "scopes": [
+                            "https://www.googleapis.com/auth/drive.readonly",
+                        ],
+                    }
+                }
+            }
+        }
+    )
+
+    provider = config.auth.providers["google"]
+    assert provider.kind == "oauth_authorization_code_pkce"
+    assert provider.client_id_env == "GOOGLE_OAUTH_CLIENT_ID"
+    assert provider.scopes == ("https://www.googleapis.com/auth/drive.readonly",)
+```
+
+- [ ] **Step 2: Run test and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_config/test_config_models.py::test_workflow_config_parses_oauth_provider_profile -q
+```
+
+Expected: fails because `auth` config does not exist.
+
+- [ ] **Step 3: Implement config models**
+
+In `src/wf_config/models.py`, add:
+
+```python
+class OAuthProviderConfig(BaseModel):
+    kind: Literal["oauth_authorization_code_pkce"]
+    auth_url: AnyUrl
+    token_url: AnyUrl
+    client_id_env: str
+    client_secret_env: str | None = None
+    scopes: tuple[str, ...] = ()
+    redirect_uri: str = "http://127.0.0.1:0/oauth/callback"
+
+
+class AuthConfig(BaseModel):
+    providers: dict[str, OAuthProviderConfig] = Field(default_factory=dict)
+```
+
+Add `auth: AuthConfig = Field(default_factory=AuthConfig)` to `WorkflowConfigFile`.
+
+Export `AuthConfig` and `OAuthProviderConfig` in `src/wf_config/__init__.py`.
+
+- [ ] **Step 4: Run config tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_config/test_config_models.py -q
+uv run basedpyright --level error src/wf_config tests/wf_config/test_config_models.py
+uv run ruff check src/wf_config tests/wf_config/test_config_models.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_config/models.py src/wf_config/__init__.py tests/wf_config/test_config_models.py
+git commit -m "feat: add oauth provider config"
+```
+
+## Task 2: OAuth Login Flow Helper
+
+**Files:**
+- Create: `src/wf_cli/oauth.py`
+- Test: `tests/wf_cli/test_auth_oauth_login.py`
+
+- [ ] **Step 1: Add failing helper tests**
+
+Create `tests/wf_cli/test_auth_oauth_login.py`:
+
+```python
+from __future__ import annotations
+
+import pytest
+
+from wf_cli.oauth import OAuthLoginResult, build_oauth_record
+from wf_config import OAuthProviderConfig
+from wf_api.auth import OAuthRefreshTokenAuth
+
+
+def test_build_oauth_record_creates_refresh_token_auth() -> None:
+    provider = OAuthProviderConfig(
+        kind="oauth_authorization_code_pkce",
+        auth_url="https://accounts.google.com/o/oauth2/v2/auth",
+        token_url="https://oauth2.googleapis.com/token",
+        client_id_env="GOOGLE_OAUTH_CLIENT_ID",
+        client_secret_env="GOOGLE_OAUTH_CLIENT_SECRET",
+        scopes=("https://www.googleapis.com/auth/drive.readonly",),
+    )
+    result = OAuthLoginResult(
+        refresh_token="refresh",
+        subject="[email protected]",
+        scopes=("https://www.googleapis.com/auth/drive.readonly",),
+    )
+
+    record = build_oauth_record(
+        auth_ref="google.drive.personal",
+        provider_name="google",
+        provider=provider,
+        client_id="client",
+        client_secret="secret",
+        result=result,
+    )
+
+    assert record.id == "google.drive.personal"
+    assert isinstance(record.auth, OAuthRefreshTokenAuth)
+    assert record.auth.client_id == "client"
+    assert record.auth.client_secret == "secret"
+    assert record.auth.refresh_token == "refresh"
+    assert record.metadata["provider"] == "google"
+    assert record.metadata["subject"] == "[email protected]"
+
+
+def test_build_oauth_record_rejects_missing_refresh_token() -> None:
+    provider = OAuthProviderConfig(
+        kind="oauth_authorization_code_pkce",
+        auth_url="https://accounts.google.com/o/oauth2/v2/auth",
+        token_url="https://oauth2.googleapis.com/token",
+        client_id_env="GOOGLE_OAUTH_CLIENT_ID",
+    )
+
+    with pytest.raises(ValueError, match="refresh token"):
+        build_oauth_record(
+            auth_ref="google.drive.personal",
+            provider_name="google",
+            provider=provider,
+            client_id="client",
+            client_secret=None,
+            result=OAuthLoginResult(refresh_token=None),
+        )
+```
+
+- [ ] **Step 2: Run tests and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py -q
+```
+
+Expected: fails because `wf_cli.oauth` does not exist.
+
+- [ ] **Step 3: Implement minimal helper**
+
+Create `src/wf_cli/oauth.py`:
+
+```python
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from wf_api.auth import OAuthRefreshTokenAuth, StoredAuthRecord
+from wf_config import OAuthProviderConfig
+
+
+@dataclass(frozen=True, slots=True)
+class OAuthLoginResult:
+    refresh_token: str | None
+    subject: str | None = None
+    scopes: tuple[str, ...] = ()
+
+
+def build_oauth_record(
+    *,
+    auth_ref: str,
+    provider_name: str,
+    provider: OAuthProviderConfig,
+    client_id: str,
+    client_secret: str | None,
+    result: OAuthLoginResult,
+) -> StoredAuthRecord:
+    if not result.refresh_token:
+        raise ValueError("OAuth login did not return a refresh token")
+    metadata: dict[str, object] = {"provider": provider_name}
+    if result.subject:
+        metadata["subject"] = result.subject
+    if result.scopes:
+        metadata["scopes"] = list(result.scopes)
+    return StoredAuthRecord(
+        id=auth_ref,
+        auth=OAuthRefreshTokenAuth(
+            client_id=client_id,
+            client_secret=client_secret or "",
+            refresh_token=result.refresh_token,
+            token_url=provider.token_url,
+            scopes=tuple(result.scopes or provider.scopes),
+        ),
+        metadata=metadata,
+    )
+```
+
+- [ ] **Step 4: Run helper tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py -q
+uv run basedpyright --level error src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+uv run ruff check src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+git commit -m "feat: build oauth auth records"
+```
+
+## Task 3: Interactive OAuth Exchange Abstraction
+
+**Files:**
+- Modify: `src/wf_cli/oauth.py`
+- Test: `tests/wf_cli/test_auth_oauth_login.py`
+
+- [ ] **Step 1: Add failing test with fake OAuth client**
+
+Append:
+
+```python
+from wf_cli.oauth import OAuthCodeLoginFlow
+
+
+class _FakeOAuthClient:
+    def __init__(self) -> None:
+        self.authorization_url = "https://auth.example/authorize?state=abc"
+        self.fetch_calls: list[str] = []
+
+    def create_authorization_url(self, auth_url: str, **kwargs: object) -> tuple[str, str]:
+        assert auth_url == "https://accounts.google.com/o/oauth2/v2/auth"
+        return self.authorization_url, "state-123"
+
+    async def fetch_token(self, token_url: str, authorization_response: str) -> dict[str, object]:
+        self.fetch_calls.append(authorization_response)
+        return {
+            "refresh_token": "refresh",
+            "scope": "https://www.googleapis.com/auth/drive.readonly",
+        }
+
+
+async def test_oauth_code_login_flow_uses_injected_client() -> None:
+    provider = OAuthProviderConfig(
+        kind="oauth_authorization_code_pkce",
+        auth_url="https://accounts.google.com/o/oauth2/v2/auth",
+        token_url="https://oauth2.googleapis.com/token",
+        client_id_env="GOOGLE_OAUTH_CLIENT_ID",
+        scopes=("https://www.googleapis.com/auth/drive.readonly",),
+    )
+    client = _FakeOAuthClient()
+    flow = OAuthCodeLoginFlow(client_factory=lambda **kwargs: client)
+
+    result = await flow.login_with_authorization_response(
+        provider=provider,
+        client_id="client",
+        client_secret=None,
+        authorization_response="http://127.0.0.1/callback?code=abc&state=state-123",
+    )
+
+    assert result.refresh_token == "refresh"
+    assert result.scopes == ("https://www.googleapis.com/auth/drive.readonly",)
+    assert client.fetch_calls == ["http://127.0.0.1/callback?code=abc&state=state-123"]
+```
+
+- [ ] **Step 2: Run test and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py::test_oauth_code_login_flow_uses_injected_client -q
+```
+
+Expected: fails because `OAuthCodeLoginFlow` does not exist.
+
+- [ ] **Step 3: Implement injectable OAuth flow**
+
+In `src/wf_cli/oauth.py`, add:
+
+```python
+from collections.abc import Callable
+from typing import Any, Protocol
+
+
+class OAuthClientLike(Protocol):
+    def create_authorization_url(self, auth_url: str, **kwargs: object) -> tuple[str, str]: ...
+    async def fetch_token(self, token_url: str, authorization_response: str) -> dict[str, object]: ...
+
+
+OAuthClientFactory = Callable[..., OAuthClientLike]
+
+
+class OAuthCodeLoginFlow:
+    def __init__(self, client_factory: OAuthClientFactory) -> None:
+        self._client_factory = client_factory
+
+    async def login_with_authorization_response(
+        self,
+        *,
+        provider: OAuthProviderConfig,
+        client_id: str,
+        client_secret: str | None,
+        authorization_response: str,
+    ) -> OAuthLoginResult:
+        client = self._client_factory(
+            client_id=client_id,
+            client_secret=client_secret,
+            scope=" ".join(provider.scopes),
+            code_challenge_method="S256",
+        )
+        client.create_authorization_url(str(provider.auth_url))
+        token = await client.fetch_token(
+            str(provider.token_url),
+            authorization_response=authorization_response,
+        )
+        refresh_token = token.get("refresh_token")
+        if refresh_token is not None and not isinstance(refresh_token, str):
+            raise ValueError("OAuth refresh_token must be a string")
+        raw_scope = token.get("scope")
+        scopes = tuple(str(raw_scope).split()) if raw_scope else provider.scopes
+        return OAuthLoginResult(refresh_token=refresh_token, scopes=scopes)
+```
+
+This helper supports pasted authorization response first. Browser callback can be a later refinement.
+
+- [ ] **Step 4: Run helper tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py -q
+uv run basedpyright --level error src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+uv run ruff check src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+git commit -m "feat: add oauth code login helper"
+```
+
+## Task 4: CLI Command `wf admin auth oauth-login`
+
+**Files:**
+- Modify: `src/wf_cli/commands/auth_admin.py`
+- Test: `tests/wf_cli/test_auth_oauth_login.py`
+
+- [ ] **Step 1: Add failing CLI test with fake flow**
+
+Append:
+
+```python
+from typer.testing import CliRunner
+
+from wf_cli.app import app
+
+
+def test_auth_oauth_login_saves_record_from_provider_profile(monkeypatch, tmp_path) -> None:
+    saved: list[object] = []
+
+    class _FakeAdmin:
+        async def save_auth_record(self, **kwargs: object) -> dict[str, object]:
+            saved.append(kwargs)
+            return {"id": kwargs["auth_ref"], "scheme": "oauth_refresh_token"}
+
+    class _FakeContext:
+        admin = _FakeAdmin()
+
+    async def _fake_login(*args: object, **kwargs: object) -> OAuthLoginResult:
+        return OAuthLoginResult(refresh_token="refresh")
+
+    monkeypatch.setattr(
+        "wf_cli.commands.auth_admin.load_cli_context_from_typer",
+        lambda ctx: _FakeContext(),
+    )
+    monkeypatch.setattr(
+        "wf_cli.commands.auth_admin._login_with_pasted_response",
+        _fake_login,
+    )
+
+    config_path = tmp_path / "wf.config.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "auth": {
+                    "providers": {
+                        "google": {
+                            "kind": "oauth_authorization_code_pkce",
+                            "auth_url": "https://accounts.google.com/o/oauth2/v2/auth",
+                            "token_url": "https://oauth2.googleapis.com/token",
+                            "client_id_env": "GOOGLE_OAUTH_CLIENT_ID",
+                            "scopes": ["https://www.googleapis.com/auth/drive.readonly"],
+                        }
+                    }
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("GOOGLE_OAUTH_CLIENT_ID", "client")
+
+    result = CliRunner().invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "admin",
+            "auth",
+            "oauth-login",
+            "google",
+            "--id",
+            "google.drive.personal",
+            "--authorization-response",
+            "http://127.0.0.1/callback?code=abc&state=state",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert saved
+    assert saved[0]["auth_ref"] == "google.drive.personal"
+    assert saved[0]["scheme"] == "oauth_refresh_token"
+```
+
+- [ ] **Step 2: Run test and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py::test_auth_oauth_login_saves_record_from_provider_profile -q
+```
+
+Expected: fails because command does not exist.
+
+- [ ] **Step 3: Implement CLI command**
+
+In `src/wf_cli/commands/auth_admin.py`, add command `oauth-login`.
+
+Implementation outline:
+
+```python
+@app.command("oauth-login")
+def oauth_login(
+    ctx: typer.Context,
+    provider_name: Annotated[str, typer.Argument(help="Auth provider profile name.")],
+    auth_ref: Annotated[str, typer.Option("--id", help="Auth record id/ref to save.")],
+    authorization_response: Annotated[
+        str,
+        typer.Option("--authorization-response", help="Full redirected callback URL after login."),
+    ],
+) -> None:
+    workflow_config = load_workflow_config_from_typer(ctx)
+    provider = workflow_config.auth.providers.get(provider_name)
+    if provider is None:
+        raise typer.BadParameter(f"unknown auth provider {provider_name!r}")
+    client_id = os.environ.get(provider.client_id_env)
+    if not client_id:
+        raise typer.BadParameter(f"missing env var {provider.client_id_env}")
+    client_secret = (
+        os.environ.get(provider.client_secret_env)
+        if provider.client_secret_env is not None
+        else None
+    )
+    result = run_cli_operation(
+        load_cli_context_from_typer(ctx),
+        _login_with_pasted_response(
+            provider=provider,
+            client_id=client_id,
+            client_secret=client_secret,
+            authorization_response=authorization_response,
+        ),
+    )
+    record = build_oauth_record(...)
+    context = load_cli_context_from_typer(ctx)
+    saved = run_cli_operation(
+        context,
+        context.admin.save_auth_record(
+            auth_ref=record.id,
+            scheme="oauth_refresh_token",
+            payload=record.auth.model_dump(mode="json", exclude={"kind"}),
+            metadata=record.metadata,
+        ),
+    )
+    emit_json(saved)
+```
+
+Use existing config loading helpers from `wf_cli.context` if available. If not available, add a small helper in this command module that reads `ctx.params["config"]` or the existing config option path used by `load_cli_context_from_typer`.
+
+Add `_login_with_pasted_response()` wrapper so tests can monkeypatch it:
+
+```python
+async def _login_with_pasted_response(...) -> OAuthLoginResult:
+    from authlib.integrations.httpx_client import AsyncOAuth2Client
+
+    flow = OAuthCodeLoginFlow(client_factory=AsyncOAuth2Client)
+    return await flow.login_with_authorization_response(...)
+```
+
+If `authlib` is not yet a dependency, add it with `uv add authlib` in a separate commit or update `pyproject.toml` manually according to project style.
+
+- [ ] **Step 4: Run CLI tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py tests/wf_cli/test_auth_admin.py -q
+uv run basedpyright --level error src/wf_cli/commands/auth_admin.py src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+uv run ruff check src/wf_cli/commands/auth_admin.py src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_cli/commands/auth_admin.py src/wf_cli/oauth.py tests/wf_cli/test_auth_oauth_login.py pyproject.toml uv.lock
+git commit -m "feat: add oauth login auth command"
+```
+
+## Task 5: Google Drive MCP Docs And Smoke Instructions
+
+**Files:**
+- Modify: `docs/wf_cli.md`
+- Modify: `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
+- Modify: `docs/current_roadmap.md`
+
+- [ ] **Step 1: Document config**
+
+Add a section to `docs/wf_cli.md`:
+
+```md
+### Google Drive MCP OAuth Setup
+
+Google Drive MCP is a remote HTTP MCP source:
+
+```json
+{
+  "sources": [
+    {
+      "id": "google.drive",
+      "kind": "mcp",
+      "transport": {
+        "kind": "http",
+        "url": "https://drivemcp.googleapis.com/mcp/v1"
+      },
+      "auth_ref": "google.drive.personal"
+    }
+  ],
+  "auth": {
+    "providers": {
+      "google": {
+        "kind": "oauth_authorization_code_pkce",
+        "auth_url": "https://accounts.google.com/o/oauth2/v2/auth",
+        "token_url": "https://oauth2.googleapis.com/token",
+        "client_id_env": "GOOGLE_OAUTH_CLIENT_ID",
+        "client_secret_env": "GOOGLE_OAUTH_CLIENT_SECRET",
+        "scopes": [
+          "https://www.googleapis.com/auth/drive.readonly"
+        ]
+      }
+    }
+  }
+}
+```
+
+Run OAuth login:
+
+```bash
+wf --config wf.config.json admin auth oauth-login google --id google.drive.personal --authorization-response "<redirected URL>"
+```
+```
+
+Mention that refresh tokens are sensitive and file store is plaintext local/dev only.
+
+- [ ] **Step 2: Update spec/roadmap status**
+
+In the auth spec, mark OAuth login as implemented if this slice is done. In roadmap, update auth line so Google Drive smoke remains optional/manual if credentials are local-only.
+
+- [ ] **Step 3: Run docs-adjacent verification**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_auth_oauth_login.py tests/wf_config/test_config_models.py -q
+uv run ruff check src/wf_cli src/wf_config tests/wf_cli/test_auth_oauth_login.py
+uv run basedpyright --level error src/wf_cli/oauth.py src/wf_cli/commands/auth_admin.py src/wf_config
+git diff --check
+```
+
+Expected: tests pass, lint/typecheck clean, no whitespace errors except acceptable CRLF warnings on Windows.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add docs/wf_cli.md docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md docs/current_roadmap.md
+git commit -m "docs: document oauth auth login"
+```
+
+## Self-Review Checklist
+
+- Spec coverage: provider profiles, typed refresh-token records, CLI login, and Drive MCP setup are covered.
+- Placeholder scan: no TODO/TBD placeholders.
+- Type consistency: plan consistently uses `OAuthProviderConfig`, `OAuthCodeLoginFlow`, `OAuthLoginResult`, and `StoredAuthRecord`.
+- Scope check: browser-opening/local callback UX is intentionally not required in this slice; pasted authorization response is enough for a verifiable first pass.
@@ -0,0 +1,778 @@
+# Typed Auth Records And MCP Binder Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace stringly runtime MCP auth interpretation with typed auth records plus a source-owned MCP auth binder, while preserving current auth file compatibility.
+
+**Architecture:** `wf_api.auth` owns durable typed auth record models and compatibility parsing. `wf_sources_mcp.auth` owns MCP-specific binding from typed records to HTTP headers/`httpx.Auth` or stdio env. `open_mcp_session()` performs the small MCP client glue from bound auth into `httpx.AsyncClient` or `StdioServerParameters`.
+
+**Tech Stack:** Python 3.14, dataclasses/Pydantic already in repo, pytest, basedpyright, ruff, httpx, MCP Python SDK.
+
+---
+
+## File Structure
+
+- Modify `src/wf_api/auth.py`: add typed auth variants, stored record wrapper, compatibility parse/serialize helpers.
+- Modify `src/wf_sources_mcp/auth.py`: add `BoundMcpHttpAuth`, `BoundMcpStdioAuth`, `McpAuthBinder`, and compatibility bridge from neutral records.
+- Modify `src/wf_sources_mcp/client/transport.py`: use `McpAuthBinder` in `open_mcp_session()`.
+- Modify `src/wf_sources_mcp/storage/store.py`: read old and new auth JSON shapes; write new shape through neutral record save path if practical.
+- Modify `src/wf_mcp/broker/service/auth_admin.py` only if summary shape needs `kind` aliasing.
+- Add/modify tests:
+  - `tests/wf_api/test_auth.py`
+  - `tests/wf_sources_mcp/test_auth.py`
+  - `tests/wf_sources_mcp/test_client_transport.py`
+  - `tests/wf_mcp/service/test_auth_admin.py`
+
+## Task 1: Typed Neutral Auth Models
+
+**Files:**
+- Modify: `src/wf_api/auth.py`
+- Test: `tests/wf_api/test_auth.py`
+
+- [ ] **Step 1: Add failing tests for typed auth parsing and compatibility**
+
+Append tests:
+
+```python
+import pytest
+
+from wf_api.auth import (
+    BearerAuth,
+    EnvAuth,
+    HeaderAuth,
+    OAuthRefreshTokenAuth,
+    OpaqueAuth,
+    StoredAuthRecord,
+    auth_record_from_compat,
+)
+
+
+def test_stored_auth_record_accepts_bearer_variant() -> None:
+    record = StoredAuthRecord(
+        id="google.drive.personal",
+        auth=BearerAuth(access_token="access-token"),
+        metadata={"provider": "google"},
+    )
+
+    assert record.id == "google.drive.personal"
+    assert record.auth.kind == "bearer"
+    assert record.metadata["provider"] == "google"
+
+
+def test_stored_auth_record_accepts_oauth_refresh_token_variant() -> None:
+    record = StoredAuthRecord(
+        id="google.drive.personal",
+        auth=OAuthRefreshTokenAuth(
+            client_id="client-id",
+            client_secret="client-secret",
+            refresh_token="refresh-token",
+            token_url="https://oauth2.googleapis.com/token",
+            scopes=("https://www.googleapis.com/auth/drive.readonly",),
+        ),
+    )
+
+    assert record.auth.kind == "oauth_refresh_token"
+    assert str(record.auth.token_url) == "https://oauth2.googleapis.com/token"
+    assert record.auth.scopes == ("https://www.googleapis.com/auth/drive.readonly",)
+
+
+def test_auth_record_from_compat_maps_existing_scheme_payload_shape() -> None:
+    record = auth_record_from_compat(
+        id="demo.default",
+        scheme="bearer",
+        payload={"token": "abc"},
+        metadata={"source": "test"},
+    )
+
+    assert isinstance(record.auth, BearerAuth)
+    assert record.auth.access_token == "abc"
+    assert record.metadata["source"] == "test"
+
+
+def test_auth_record_from_compat_preserves_unknown_as_opaque() -> None:
+    record = auth_record_from_compat(
+        id="demo.default",
+        scheme="custom",
+        payload={"x": "y"},
+        metadata={},
+    )
+
+    assert isinstance(record.auth, OpaqueAuth)
+    assert record.auth.scheme == "custom"
+    assert record.auth.payload == {"x": "y"}
+
+
+def test_typed_auth_rejects_missing_bearer_token() -> None:
+    with pytest.raises(ValueError, match="bearer token"):
+        auth_record_from_compat(
+            id="demo.default",
+            scheme="bearer",
+            payload={},
+            metadata={},
+        )
+```
+
+- [ ] **Step 2: Run the tests and verify they fail**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_auth.py -q
+```
+
+Expected: fails because typed auth classes/helpers do not exist.
+
+- [ ] **Step 3: Implement typed auth models and compatibility parser**
+
+In `src/wf_api/auth.py`, keep existing `AuthRecord` for compatibility but add:
+
+```python
+from typing import Annotated, Any, Literal
+
+from pydantic import AnyUrl, BaseModel, Field
+
+
+class BearerAuth(BaseModel):
+    kind: Literal["bearer"] = "bearer"
+    access_token: str
+
+
+class HeaderAuth(BaseModel):
+    kind: Literal["headers"] = "headers"
+    headers: dict[str, str]
+
+
+class EnvAuth(BaseModel):
+    kind: Literal["env"] = "env"
+    env: dict[str, str]
+
+
+class OAuthRefreshTokenAuth(BaseModel):
+    kind: Literal["oauth_refresh_token"] = "oauth_refresh_token"
+    client_id: str
+    client_secret: str
+    refresh_token: str
+    token_url: AnyUrl
+    scopes: tuple[str, ...] = ()
+
+
+class OpaqueAuth(BaseModel):
+    kind: Literal["opaque"] = "opaque"
+    scheme: str
+    payload: dict[str, object] = Field(default_factory=dict)
+
+
+AuthVariant = Annotated[
+    BearerAuth | HeaderAuth | EnvAuth | OAuthRefreshTokenAuth | OpaqueAuth,
+    Field(discriminator="kind"),
+]
+
+
+class StoredAuthRecord(BaseModel):
+    id: str
+    auth: AuthVariant
+    metadata: dict[str, object] = Field(default_factory=dict)
+
+    def model_post_init(self, __context: Any) -> None:
+        validate_auth_id(self.id)
+
+
+def auth_record_from_compat(
+    *,
+    id: str,
+    scheme: str,
+    payload: Mapping[str, object],
+    metadata: Mapping[str, object] | None = None,
+) -> StoredAuthRecord:
+    payload_dict = dict(payload)
+    metadata_dict = dict(metadata or {})
+    match scheme:
+        case "bearer":
+            token = payload_dict.get("token") or payload_dict.get("access_token")
+            if not isinstance(token, str) or not token:
+                raise ValueError("bearer token is required")
+            auth: AuthVariant = BearerAuth(access_token=token)
+        case "headers":
+            raw_headers = payload_dict.get("headers", {})
+            headers = (
+                {
+                    str(key): str(value)
+                    for key, value in raw_headers.items()
+                    if isinstance(key, str) and isinstance(value, str)
+                }
+                if isinstance(raw_headers, dict)
+                else {}
+            )
+            auth = HeaderAuth(headers=headers)
+        case "env":
+            raw_env = payload_dict.get("env", {})
+            env = (
+                {
+                    str(key): str(value)
+                    for key, value in raw_env.items()
+                    if isinstance(key, str) and isinstance(value, str)
+                }
+                if isinstance(raw_env, dict)
+                else {}
+            )
+            auth = EnvAuth(env=env)
+        case _:
+            auth = OpaqueAuth(scheme=scheme, payload=payload_dict)
+    return StoredAuthRecord(id=id, auth=auth, metadata=metadata_dict)
+```
+
+Update `__all__` to export all new symbols.
+
+- [ ] **Step 4: Run tests and typecheck**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_auth.py -q
+uv run basedpyright --level error src/wf_api/auth.py tests/wf_api/test_auth.py
+uv run ruff check src/wf_api/auth.py tests/wf_api/test_auth.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_api/auth.py tests/wf_api/test_auth.py
+git commit -m "feat: add typed auth record variants"
+```
+
+## Task 2: MCP Auth Binder
+
+**Files:**
+- Modify: `src/wf_sources_mcp/auth.py`
+- Test: `tests/wf_sources_mcp/test_auth.py`
+
+- [ ] **Step 1: Add failing MCP binder tests**
+
+Append tests:
+
+```python
+import httpx
+import pytest
+
+from wf_api.auth import (
+    BearerAuth,
+    EnvAuth,
+    HeaderAuth,
+    OAuthRefreshTokenAuth,
+    StoredAuthRecord,
+)
+from wf_sources_mcp.auth import McpAuthBinder, OAuthAccessToken
+
+
+class _FakeRefresher:
+    def __init__(self) -> None:
+        self.calls: list[OAuthRefreshTokenAuth] = []
+
+    async def refresh(self, auth: OAuthRefreshTokenAuth) -> OAuthAccessToken:
+        self.calls.append(auth)
+        return OAuthAccessToken(access_token="fresh-token", expires_in=3600)
+
+
+async def test_mcp_binder_binds_bearer_for_http() -> None:
+    binder = McpAuthBinder()
+    record = StoredAuthRecord(
+        id="demo.auth",
+        auth=BearerAuth(access_token="token"),
+    )
+
+    bound = await binder.bind_http_auth(record)
+
+    assert bound.headers == {"Authorization": "Bearer token"}
+    assert bound.auth is None
+
+
+async def test_mcp_binder_binds_headers_for_http() -> None:
+    binder = McpAuthBinder()
+    record = StoredAuthRecord(
+        id="demo.auth",
+        auth=HeaderAuth(headers={"X-Test": "yes"}),
+    )
+
+    bound = await binder.bind_http_auth(record)
+
+    assert bound.headers == {"X-Test": "yes"}
+
+
+async def test_mcp_binder_binds_env_for_stdio() -> None:
+    binder = McpAuthBinder()
+    record = StoredAuthRecord(
+        id="demo.auth",
+        auth=EnvAuth(env={"TOKEN": "abc"}),
+    )
+
+    bound = await binder.bind_stdio_auth(record)
+
+    assert bound.env == {"TOKEN": "abc"}
+
+
+async def test_mcp_binder_refreshes_oauth_for_http() -> None:
+    refresher = _FakeRefresher()
+    binder = McpAuthBinder(oauth_refresher=refresher)
+    record = StoredAuthRecord(
+        id="google.drive.personal",
+        auth=OAuthRefreshTokenAuth(
+            client_id="client",
+            client_secret="secret",
+            refresh_token="refresh",
+            token_url="https://oauth2.googleapis.com/token",
+        ),
+    )
+
+    bound = await binder.bind_http_auth(record)
+
+    assert bound.headers == {"Authorization": "Bearer fresh-token"}
+    assert len(refresher.calls) == 1
+
+
+async def test_mcp_binder_rejects_env_for_http() -> None:
+    binder = McpAuthBinder()
+    record = StoredAuthRecord(id="demo.auth", auth=EnvAuth(env={"TOKEN": "abc"}))
+
+    with pytest.raises(ValueError, match="not supported for MCP HTTP"):
+        await binder.bind_http_auth(record)
+
+
+async def test_mcp_binder_rejects_headers_for_stdio() -> None:
+    binder = McpAuthBinder()
+    record = StoredAuthRecord(
+        id="demo.auth",
+        auth=HeaderAuth(headers={"Authorization": "Bearer token"}),
+    )
+
+    with pytest.raises(ValueError, match="not supported for MCP stdio"):
+        await binder.bind_stdio_auth(record)
+```
+
+- [ ] **Step 2: Run tests and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_auth.py -q
+```
+
+Expected: fails because `McpAuthBinder` and `OAuthAccessToken` do not exist.
+
+- [ ] **Step 3: Implement binder**
+
+In `src/wf_sources_mcp/auth.py`, import typed variants from `wf_api.auth` and add:
+
+```python
+from dataclasses import dataclass, field
+from typing import Protocol
+
+import httpx
+from wf_api.auth import (
+    BearerAuth,
+    EnvAuth,
+    HeaderAuth,
+    OAuthRefreshTokenAuth,
+    OpaqueAuth,
+    StoredAuthRecord,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class OAuthAccessToken:
+    access_token: str
+    expires_in: int | None = None
+
+
+class OAuthTokenRefresher(Protocol):
+    async def refresh(self, auth: OAuthRefreshTokenAuth) -> OAuthAccessToken: ...
+
+
+@dataclass(frozen=True, slots=True)
+class BoundMcpHttpAuth:
+    headers: dict[str, str] = field(default_factory=dict)
+    auth: httpx.Auth | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class BoundMcpStdioAuth:
+    env: dict[str, str] = field(default_factory=dict)
+
+
+class McpAuthBinder:
+    def __init__(self, oauth_refresher: OAuthTokenRefresher | None = None) -> None:
+        self._oauth_refresher = oauth_refresher
+
+    async def bind_http_auth(
+        self,
+        record: StoredAuthRecord | None,
+    ) -> BoundMcpHttpAuth:
+        if record is None:
+            return BoundMcpHttpAuth()
+        auth = record.auth
+        if isinstance(auth, BearerAuth):
+            return BoundMcpHttpAuth(
+                headers={"Authorization": f"Bearer {auth.access_token}"}
+            )
+        if isinstance(auth, HeaderAuth):
+            return BoundMcpHttpAuth(headers=dict(auth.headers))
+        if isinstance(auth, OAuthRefreshTokenAuth):
+            if self._oauth_refresher is None:
+                raise ValueError("oauth_refresh_token requires an OAuthTokenRefresher")
+            token = await self._oauth_refresher.refresh(auth)
+            return BoundMcpHttpAuth(
+                headers={"Authorization": f"Bearer {token.access_token}"}
+            )
+        if isinstance(auth, EnvAuth):
+            raise ValueError("env auth is not supported for MCP HTTP")
+        if isinstance(auth, OpaqueAuth):
+            raise ValueError(f"opaque auth scheme {auth.scheme!r} is not supported for MCP HTTP")
+        raise TypeError(f"unsupported auth variant {type(auth).__name__}")
+
+    async def bind_stdio_auth(
+        self,
+        record: StoredAuthRecord | None,
+    ) -> BoundMcpStdioAuth:
+        if record is None:
+            return BoundMcpStdioAuth()
+        auth = record.auth
+        if isinstance(auth, EnvAuth):
+            return BoundMcpStdioAuth(env=dict(auth.env))
+        if isinstance(auth, BearerAuth | HeaderAuth | OAuthRefreshTokenAuth | OpaqueAuth):
+            raise ValueError(f"{auth.kind} auth is not supported for MCP stdio")
+        raise TypeError(f"unsupported auth variant {type(auth).__name__}")
+```
+
+Keep existing `mcp_auth_headers()` and `mcp_auth_env()` compatibility helpers for old callers in this slice.
+
+Update `__all__` with binder symbols.
+
+- [ ] **Step 4: Run tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_auth.py -q
+uv run basedpyright --level error src/wf_sources_mcp/auth.py tests/wf_sources_mcp/test_auth.py
+uv run ruff check src/wf_sources_mcp/auth.py tests/wf_sources_mcp/test_auth.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_sources_mcp/auth.py tests/wf_sources_mcp/test_auth.py
+git commit -m "feat: add mcp auth binder"
+```
+
+## Task 3: Store Compatibility For New Auth Shape
+
+**Files:**
+- Modify: `src/wf_sources_mcp/storage/store.py`
+- Test: `tests/wf_sources_mcp/test_auth_storage_exports.py` or `tests/wf_mcp/test_store.py`
+
+- [ ] **Step 1: Add failing store tests**
+
+Add tests near existing auth store tests:
+
+```python
+from wf_api.auth import BearerAuth, StoredAuthRecord
+from wf_sources_mcp.storage import FileAuthStore
+
+
+def test_file_auth_store_loads_new_stored_auth_record_shape(tmp_path: Path) -> None:
+    store = FileAuthStore(tmp_path)
+    path = tmp_path / "auth" / "google.drive.personal.json"
+    path.write_text(
+        json.dumps(
+            {
+                "id": "google.drive.personal",
+                "auth": {"kind": "bearer", "access_token": "token"},
+                "metadata": {"provider": "google"},
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    record = store.load_auth_record("google.drive.personal")
+
+    assert isinstance(record, StoredAuthRecord)
+    assert isinstance(record.auth, BearerAuth)
+    assert record.auth.access_token == "token"
+    assert record.metadata["provider"] == "google"
+
+
+def test_file_auth_store_writes_new_stored_auth_record_shape(tmp_path: Path) -> None:
+    store = FileAuthStore(tmp_path)
+    record = StoredAuthRecord(
+        id="google.drive.personal",
+        auth=BearerAuth(access_token="token"),
+        metadata={"provider": "google"},
+    )
+
+    store.save_auth_record(record)
+
+    data = json.loads((tmp_path / "auth" / "google.drive.personal.json").read_text())
+    assert data["id"] == "google.drive.personal"
+    assert data["auth"]["kind"] == "bearer"
+    assert data["auth"]["access_token"] == "token"
+    assert data["metadata"]["provider"] == "google"
+```
+
+Adjust import path if the chosen existing test file already imports `Path`/`json`.
+
+- [ ] **Step 2: Run tests and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_auth_storage_exports.py -q
+```
+
+Expected: fails because store returns old neutral `AuthRecord` or writes old shape.
+
+- [ ] **Step 3: Update store read/write**
+
+In `src/wf_sources_mcp/storage/store.py`:
+
+- Import `StoredAuthRecord` and `auth_record_from_compat`.
+- Change `save_auth_record()` to accept `NeutralAuthRecord | StoredAuthRecord`; if `StoredAuthRecord`, write `record.model_dump(mode="json", indent=2)` shape.
+- Change `load_auth_record()` to return `StoredAuthRecord | NeutralAuthRecord` only if too many callers break; preferred return is `StoredAuthRecord`.
+- Keep `load_auth()` returning legacy `wf_sources_mcp.auth.AuthRecord` for compatibility by converting the stored typed record back to legacy where possible.
+
+Minimal conversion helper:
+
+```python
+def _stored_to_legacy(record: StoredAuthRecord) -> AuthRecord:
+    auth = record.auth
+    if isinstance(auth, BearerAuth):
+        return AuthRecord(record.id, "bearer", {"token": auth.access_token})
+    if isinstance(auth, HeaderAuth):
+        return AuthRecord(record.id, "headers", {"headers": dict(auth.headers)})
+    if isinstance(auth, EnvAuth):
+        return AuthRecord(record.id, "env", {"env": dict(auth.env)})
+    if isinstance(auth, OAuthRefreshTokenAuth):
+        return AuthRecord(
+            record.id,
+            "oauth_refresh_token",
+            auth.model_dump(mode="json", exclude={"kind"}),
+        )
+    if isinstance(auth, OpaqueAuth):
+        return AuthRecord(record.id, auth.scheme, dict(auth.payload))
+    raise TypeError(f"unsupported auth variant {type(auth).__name__}")
+```
+
+- [ ] **Step 4: Run focused auth/store tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_auth.py tests/wf_sources_mcp/test_auth.py tests/wf_sources_mcp/test_auth_storage_exports.py tests/wf_mcp/test_store.py -q
+uv run basedpyright --level error src/wf_api/auth.py src/wf_sources_mcp/auth.py src/wf_sources_mcp/storage/store.py
+uv run ruff check src/wf_api/auth.py src/wf_sources_mcp/auth.py src/wf_sources_mcp/storage/store.py tests/wf_api/test_auth.py tests/wf_sources_mcp/test_auth.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_sources_mcp/storage/store.py tests/wf_sources_mcp/test_auth_storage_exports.py tests/wf_mcp/test_store.py
+git commit -m "feat: read and write typed auth records"
+```
+
+## Task 4: Wire Binder Into MCP Session Opening
+
+**Files:**
+- Modify: `src/wf_sources_mcp/client/transport.py`
+- Test: `tests/wf_sources_mcp/test_client_transport.py`
+
+- [ ] **Step 1: Add failing client transport tests**
+
+Add tests that capture `httpx.AsyncClient` kwargs and stdio env using existing monkeypatch style in `test_client_transport.py`:
+
+```python
+from wf_api.auth import BearerAuth, EnvAuth, StoredAuthRecord
+
+
+async def test_open_mcp_session_uses_binder_for_http_headers(monkeypatch) -> None:
+    captured: dict[str, object] = {}
+
+    class _AsyncClient:
+        def __init__(self, **kwargs: object) -> None:
+            captured.update(kwargs)
+
+        async def __aenter__(self) -> "_AsyncClient":
+            return self
+
+        async def __aexit__(self, *args: object) -> None:
+            return None
+
+    monkeypatch.setattr("wf_sources_mcp.client.transport.httpx.AsyncClient", _AsyncClient)
+    # Reuse existing fake streamable_http_client/ClientSession helpers in this file.
+
+    connection = McpSourceConnection(
+        id="google.drive",
+        provider="google",
+        transport=HttpSourceTransport(url="https://drivemcp.googleapis.com/mcp/v1"),
+    )
+    auth = StoredAuthRecord(
+        id="google.drive.personal",
+        auth=BearerAuth(access_token="token"),
+    )
+
+    async with open_mcp_session(connection, auth):
+        pass
+
+    assert captured["headers"] == {"Authorization": "Bearer token"}
+
+
+async def test_open_mcp_session_uses_binder_for_stdio_env(monkeypatch) -> None:
+    # Follow existing stdio capture pattern in this file.
+    connection = McpSourceConnection(
+        id="demo.default",
+        provider="demo",
+        transport=StdioSourceTransport(command="demo", env={"BASE": "1"}),
+    )
+    auth = StoredAuthRecord(id="demo.auth", auth=EnvAuth(env={"TOKEN": "abc"}))
+
+    async with open_mcp_session(connection, auth):
+        pass
+
+    assert captured_stdio_params.env == {"BASE": "1", "TOKEN": "abc"}
+```
+
+Use the existing fake helpers in `tests/wf_sources_mcp/test_client_transport.py`; do not duplicate a second fake session stack if the file already has one.
+
+- [ ] **Step 2: Run tests and verify failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_client_transport.py -q
+```
+
+Expected: fails because `open_mcp_session()` still calls `mcp_auth_headers()` / `mcp_auth_env()`.
+
+- [ ] **Step 3: Update `open_mcp_session()`**
+
+Change signature to accept both old MCP auth and new stored auth during transition:
+
+```python
+async def open_mcp_session(
+    connection: McpSourceConnection,
+    auth: AuthRecord | StoredAuthRecord | None,
+    *,
+    auth_binder: McpAuthBinder | None = None,
+) -> AsyncIterator[ClientSession]:
+```
+
+Inside, normalize old `AuthRecord` through compatibility if needed, or keep a small helper:
+
+```python
+def _as_stored_auth(auth: AuthRecord | StoredAuthRecord | None) -> StoredAuthRecord | None:
+    if auth is None or isinstance(auth, StoredAuthRecord):
+        return auth
+    return auth_record_from_compat(
+        id=auth.connection_id,
+        scheme=auth.scheme,
+        payload=auth.payload,
+        metadata={},
+    )
+```
+
+Use:
+
+```python
+binder = auth_binder or McpAuthBinder()
+stored_auth = _as_stored_auth(auth)
+```
+
+For stdio:
+
+```python
+bound = await binder.bind_stdio_auth(stored_auth)
+env = {**transport.env, **bound.env}
+```
+
+For HTTP:
+
+```python
+bound = await binder.bind_http_auth(stored_auth)
+http_client = httpx.AsyncClient(
+    headers=bound.headers or None,
+    auth=bound.auth,
+)
+```
+
+- [ ] **Step 4: Run MCP source tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp tests/wf_mcp/test_stateful_runtime.py tests/wf_mcp/test_compat_imports.py -q
+uv run basedpyright --level error src/wf_sources_mcp
+uv run ruff check src/wf_sources_mcp tests/wf_sources_mcp
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_sources_mcp/client/transport.py tests/wf_sources_mcp/test_client_transport.py
+git commit -m "feat: bind auth when opening mcp sessions"
+```
+
+## Task 5: Docs And Final Verification
+
+**Files:**
+- Modify: `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
+- Modify: `docs/current_roadmap.md`
+
+- [ ] **Step 1: Update status**
+
+In the auth spec `## Status`, add:
+
+```md
+Slice 5 introduces typed stored auth records and MCP auth binding while preserving old `scheme + payload` compatibility input.
+```
+
+In `docs/current_roadmap.md`, mark typed auth records and MCP binder as completed, leaving OAuth refresh-token and Drive smoke as next.
+
+- [ ] **Step 2: Run focused verification**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_auth.py tests/wf_sources_mcp tests/wf_mcp/service/test_auth_admin.py tests/wf_mcp/test_auth.py -q
+uv run ruff check src/wf_api/auth.py src/wf_sources_mcp tests/wf_api/test_auth.py tests/wf_sources_mcp
+uv run basedpyright --level error src/wf_api/auth.py src/wf_sources_mcp
+git diff --check
+```
+
+Expected: tests pass, lint clean, typecheck clean, no whitespace errors except acceptable CRLF warnings on Windows.
+
+- [ ] **Step 3: Final review**
+
+Check:
+
+- `wf_api` still imports no `wf_mcp`.
+- No secret payload values are returned by auth admin summaries.
+- Existing `scheme + payload` auth files can still be read.
+- `open_mcp_session()` accepts old auth records and new typed records.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md docs/current_roadmap.md
+git commit -m "docs: record typed mcp auth binding"
+```
+
+## Self-Review Checklist
+
+- Spec coverage: typed records, source-owned binder, Drive-compatible bearer output, and compatibility parsing are covered.
+- Placeholder scan: no TODO/TBD placeholders.
+- Type consistency: plan consistently uses `StoredAuthRecord`, `AuthVariant`, `McpAuthBinder`, `BoundMcpHttpAuth`, and `BoundMcpStdioAuth`.