feat: add neutral auth store boundary

docs/current_roadmap.md

@@ -227,6 +227,10 @@ implementation state.
     [2026-06-06 auth/source secrets boundary](./superpowers/specs/2026-06-06-auth-source-secrets-boundary.md):
     sources carry `auth_ref`, runtime resolves through an auth store interface,
     and the current filesystem auth files are only one adapter.
+    First implementation slice complete: neutral auth records/store protocol
+    exist in `wf_api`, MCP runtime auth resolution prefers explicit `auth_ref`
+    with legacy connection-id fallback, and MCP payload interpretation is
+    isolated in provider-specific adapter helpers.
   - Completed: `wf run watch` starts run progress UX with polling over existing
     `inspect_run` and optional bounded `read_run_trace`. SSE/WebSocket/MCP
     progress remains deferred until polling UX proves insufficient.

docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md

@@ -11,6 +11,12 @@ This spec defines the next boundary before expanding auth behavior. The
 important choice is interface first: file-backed auth is one implementation, not
 the architecture.
 
+## Status
+
+Slice 1 implements the neutral auth record/store protocol and MCP compatibility
+bridge. Diagnostics, auth admin surfaces, and provider-specific auth unions are
+future slices.
+
 ## Current State
 
 Existing MCP runtime auth is connection-id keyed:

src/wf_api/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from .auth import AUTH_ID_PATTERN, AuthRecord, AuthStore, validate_auth_id
 from .listing import matches_query, paged_list_payload
 from .admin import (
     WorkflowAdminApi,
@@ -66,7 +67,10 @@ from .stores import WorkflowStores, file_workflow_stores
 from .durable_context import durable_workflow_api, require_workflow_stores
 
 __all__ = [
+    "AUTH_ID_PATTERN",
     "DEFAULT_CALL_STEP_ID",
+    "AuthRecord",
+    "AuthStore",
     "builtin_sources",
     "get_qualified_spec",
     "matches_query",
@@ -119,6 +123,7 @@ __all__ = [
     "WrapperHintConfidence",
     "WrapperOutcomePolicy",
     "parse_workflow_surface_capability_id",
+    "validate_auth_id",
     "workflow_output_schema_for_authoring",
     "wrapper_hints_for_capability",
     "resolve_runtime_dependencies",

src/wf_api/auth.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from typing import Protocol
+
+AUTH_ID_PATTERN = r"^[A-Za-z0-9_][A-Za-z0-9_.-]*$"
+
+
+def validate_auth_id(value: str) -> str:
+    """Validate auth refs that are safe as store keys and path segments.
+
+    Auth refs deliberately carry no provider semantics. Source providers decide
+    how a resolved auth record is interpreted.
+    """
+
+    if not re.fullmatch(AUTH_ID_PATTERN, value):
+        raise ValueError(
+            "auth id must start with alphanumeric or underscore and contain "
+            "only [A-Za-z0-9_.-]"
+        )
+    return value
+
+
+@dataclass(frozen=True, slots=True)
+class AuthRecord:
+    """Neutral credential record resolved by auth ref.
+
+    `scheme + payload` is a compatibility bridge, not the long-term taxonomy.
+    Keep payload interpretation inside provider adapters so a future
+    discriminated union can replace this without touching workflow/config code.
+    """
+
+    id: str
+    scheme: str
+    payload: Mapping[str, object]
+    metadata: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        validate_auth_id(self.id)
+        if not self.scheme:
+            raise ValueError("auth scheme must be non-empty")
+
+
+class AuthStore(Protocol):
+    """Read-only runtime credential lookup by auth ref."""
+
+    def load_auth(self, auth_ref: str) -> AuthRecord | None: ...
+
+
+__all__ = [
+    "AUTH_ID_PATTERN",
+    "AuthRecord",
+    "AuthStore",
+    "validate_auth_id",
+]

src/wf_mcp/auth.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from wf_api.auth import AuthRecord as NeutralAuthRecord
+
+from .models import AuthRecord as McpAuthRecord
+
+
+def mcp_auth_from_neutral(record: NeutralAuthRecord) -> McpAuthRecord:
+    """Adapt neutral auth to the current MCP compatibility record."""
+
+    return McpAuthRecord(
+        connection_id=record.id,
+        scheme=record.scheme,
+        payload=dict(record.payload),
+    )
+
+
+def neutral_auth_from_mcp(record: McpAuthRecord) -> NeutralAuthRecord:
+    """Adapt legacy MCP auth into the neutral record shape."""
+
+    return NeutralAuthRecord(
+        id=record.connection_id,
+        scheme=record.scheme,
+        payload=dict(record.payload),
+    )
+
+
+def mcp_auth_headers(auth: McpAuthRecord | None) -> dict[str, str]:
+    """Return HTTP headers understood by MCP HTTP transports.
+
+    This is intentionally MCP-specific. Neutral code must not inspect payload
+    keys such as `headers` or `token`.
+    """
+
+    if auth is None:
+        return {}
+    raw_headers = auth.payload.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 {}
+    token = auth.payload.get("token")
+    if isinstance(token, str) and "Authorization" not in headers:
+        headers["Authorization"] = f"Bearer {token}"
+    return headers
+
+
+def mcp_auth_env(auth: McpAuthRecord | None) -> dict[str, str]:
+    """Return environment variables understood by MCP stdio transports."""
+
+    if auth is None:
+        return {}
+    raw_env = auth.payload.get("env", {})
+    if not isinstance(raw_env, dict):
+        return {}
+    return {
+        str(key): str(value)
+        for key, value in raw_env.items()
+        if isinstance(key, str) and isinstance(value, str)
+    }
+
+
+__all__ = [
+    "mcp_auth_env",
+    "mcp_auth_from_neutral",
+    "mcp_auth_headers",
+    "neutral_auth_from_mcp",
+]

src/wf_mcp/broker/service/core.py

@@ -91,7 +91,7 @@ class WfMcpService:
             connection_list_enabled=self.connection_service.list_enabled,
             connection_list_all=self.connection_service.list_all,
             tool_executor_for=self.upstream.tool_executor_for,
-            load_auth=self.upstream.load_auth,
+            load_auth=self.upstream.load_connection_auth,
             emit_event=self.events.record_event,
             default_catalog_max_age_seconds=self.default_catalog_max_age_seconds,
         )

src/wf_mcp/broker/service/source_catalog.py

@@ -36,7 +36,7 @@ from .specs import get_qualified_spec, qualify_spec
 ConnectionLookup = Callable[[str], ConnectionConfig]
 ConnectionList = Callable[[], list[ConnectionConfig]]
 ToolExecutorLookup = Callable[[ConnectionConfig], ToolExecutor]
-AuthLoader = Callable[[str], AuthRecord | None]
+AuthLoader = Callable[[ConnectionConfig], AuthRecord | None]
 EventEmitter = Callable[[McpEvent], None]
 
 
@@ -270,7 +270,7 @@ class SourceCatalogService:
 
         async def invoke_tool(payload: BaseModel) -> NodeReturn[BaseModel]:
             connection = self.connection_lookup(entry.connection_id)
-            auth = self.load_auth(entry.connection_id)
+            auth = self.load_auth(connection)
             result = await self.tool_executor_for(connection).call_tool(
                 connection,
                 auth,

src/wf_mcp/broker/service/upstream_transport.py

@@ -63,6 +63,19 @@ class UpstreamTransportService:
     def load_auth(self, connection_id: str) -> AuthRecord | None:
         return self.store.load_auth(connection_id)
 
+    def load_connection_auth(self, connection: ConnectionConfig) -> AuthRecord | None:
+        """Resolve auth for a connection, preferring explicit source auth_ref.
+
+        Legacy MCP auth records are keyed by connection id. New source registry
+        and neutral config entries carry `auth_ref`; keep both paths until the
+        old compatibility surface has no callers.
+        """
+
+        auth_ref = connection.metadata.get("auth_ref")
+        if isinstance(auth_ref, str):
+            return self.load_auth(auth_ref)
+        return self.load_auth(connection.id)
+
     def tool_executor_for(self, connection: ConnectionConfig) -> ToolExecutor:
         """Return the executor used by generated workflow NodeSpecs.
 
@@ -81,7 +94,7 @@ class UpstreamTransportService:
         uri: str,
     ) -> dict[str, Any]:
         adapter = require_adapter(connection, self.adapters)
-        auth = self.load_auth(connection.id)
+        auth = self.load_connection_auth(connection)
         self.event_sink(
             make_event(
                 "resource_read_started",
@@ -109,7 +122,7 @@ class UpstreamTransportService:
         arguments: dict[str, str] | None = None,
     ) -> dict[str, Any]:
         adapter = require_adapter(connection, self.adapters)
-        auth = self.load_auth(connection.id)
+        auth = self.load_connection_auth(connection)
         self.event_sink(
             make_event(
                 "prompt_get_started",
@@ -137,7 +150,7 @@ class UpstreamTransportService:
         params: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         adapter = require_adapter(connection, self.adapters)
-        auth = self.load_auth(connection.id)
+        auth = self.load_connection_auth(connection)
         self.event_sink(
             make_event(
                 "raw_method_started",
@@ -165,7 +178,7 @@ class UpstreamTransportService:
         params: dict[str, Any] | None = None,
     ) -> None:
         adapter = require_adapter(connection, self.adapters)
-        auth = self.load_auth(connection.id)
+        auth = self.load_connection_auth(connection)
         self.event_sink(
             make_event(
                 "raw_notification_started",
@@ -193,7 +206,7 @@ class UpstreamTransportService:
         default_catalog_max_age_seconds: int = 300,
         record_catalog_change_events: Callable[[str, CatalogSnapshot, str], None],
     ) -> None:
-        auth = self.load_auth(connection.id)
+        auth = self.load_connection_auth(connection)
         self.event_sink(
             make_event(
                 "catalog_refresh_started",
@@ -295,7 +308,7 @@ class UpstreamTransportService:
                 continue
             try:
                 adapter = require_adapter(connection, self.adapters)
-                auth = self.load_auth(source_id)
+                auth = self.load_connection_auth(connection)
                 await asyncio.wait_for(
                     adapter.list_tools(connection, auth),
                     timeout=LIVE_SOURCE_CHECK_TIMEOUT_SECONDS,

src/wf_mcp/runtime/factory.py

@@ -10,20 +10,11 @@ from mcp.client.stdio import StdioServerParameters, stdio_client
 from mcp.client.streamable_http import streamable_http_client
 from mcp.types import CallToolResult
 
+from ..auth import mcp_auth_env, mcp_auth_headers
 from ..models import AuthRecord, ConnectionConfig
 from .session import PersistentMcpSession
 
 
-def _auth_headers(auth: AuthRecord | None) -> dict[str, str]:
-    if auth is None:
-        return {}
-    headers = dict(auth.payload.get("headers", {}))
-    token = auth.payload.get("token")
-    if isinstance(token, str) and "Authorization" not in headers:
-        headers["Authorization"] = f"Bearer {token}"
-    return headers
-
-
 @dataclass(slots=True)
 class PersistentSessionFactory:
     """Create initialized persistent MCP sessions for configured connections.
@@ -57,10 +48,9 @@ class PersistentSessionFactory:
         transport = connection.metadata.get("transport", "stdio")
         if transport == "stdio":
             env = connection.metadata.get("env")
-            if auth is not None:
+            auth_env = mcp_auth_env(auth)
-                auth_env = auth.payload.get("env")
+            if auth_env:
-                if isinstance(auth_env, dict):
+                env = {**(env or {}), **auth_env}
-                    env = {**(env or {}), **auth_env}
             params = StdioServerParameters(
                 command=connection.metadata["command"],
                 args=list(connection.metadata.get("args", [])),
@@ -78,7 +68,7 @@ class PersistentSessionFactory:
 
         if transport == "streamable_http":
             http_client = await stack.enter_async_context(
-                httpx.AsyncClient(headers=_auth_headers(auth) or None)
+                httpx.AsyncClient(headers=mcp_auth_headers(auth) or None)
             )
             (
                 read_stream,

src/wf_mcp/sdk/adapter.py

@@ -17,6 +17,7 @@ from mcp.types import (
 )
 from pydantic import AnyUrl
 
+from ..auth import mcp_auth_env, mcp_auth_headers
 from ..capabilities import DiscoveredPrompt, DiscoveredResource, DiscoveredTool
 from ..models import AuthRecord, ConnectionConfig
 from .base import BackendAdapter, ToolCallResult
@@ -28,16 +29,6 @@ from .converters import (
 )
 
 
-def _auth_headers(auth: AuthRecord | None) -> dict[str, str]:
-    if auth is None:
-        return {}
-    headers = dict(auth.payload.get("headers", {}))
-    token = auth.payload.get("token")
-    if isinstance(token, str) and "Authorization" not in headers:
-        headers["Authorization"] = f"Bearer {token}"
-    return headers
-
-
 class McpSdkAdapter(BackendAdapter):
     @asynccontextmanager
     async def _session(
@@ -51,10 +42,9 @@ class McpSdkAdapter(BackendAdapter):
             args = list(connection.metadata.get("args", []))
             env = connection.metadata.get("env")
             cwd = connection.metadata.get("cwd")
-            if auth is not None:
+            auth_env = mcp_auth_env(auth)
-                auth_env = auth.payload.get("env")
+            if auth_env:
-                if isinstance(auth_env, dict):
+                env = {**(env or {}), **auth_env}
-                    env = {**(env or {}), **auth_env}
             params = StdioServerParameters(
                 command=command,
                 args=args,
@@ -69,7 +59,7 @@ class McpSdkAdapter(BackendAdapter):
 
         if transport == "streamable_http":
             url = connection.metadata["url"]
-            headers = _auth_headers(auth)
+            headers = mcp_auth_headers(auth)
             http_client = httpx.AsyncClient(headers=headers or None)
             async with http_client:
                 async with streamable_http_client(

src/wf_mcp/storage/store.py

@@ -3,6 +3,9 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
+from wf_api.auth import AuthRecord as NeutralAuthRecord
+
+from ..auth import mcp_auth_from_neutral, neutral_auth_from_mcp
 from ..connections import parse_connection_id
 from ..models import (
     AuthRecord,
@@ -21,6 +24,12 @@ class Store:
     def load_auth(self, connection_id: str) -> AuthRecord | None:
         raise NotImplementedError
 
+    def save_auth_record(self, record: NeutralAuthRecord) -> None:
+        raise NotImplementedError
+
+    def load_auth_record(self, auth_ref: str) -> NeutralAuthRecord | None:
+        raise NotImplementedError
+
     def save_catalog(self, snapshot: CatalogSnapshot) -> None:
         raise NotImplementedError
 
@@ -81,6 +90,19 @@ class FileStore(Store):
         data = json.loads(path.read_text(encoding="utf-8"))
         return AuthRecord(**data)
 
+    def save_auth_record(self, record: NeutralAuthRecord) -> None:
+        """Save neutral auth through the legacy MCP file shape."""
+
+        self.save_auth(mcp_auth_from_neutral(record))
+
+    def load_auth_record(self, auth_ref: str) -> NeutralAuthRecord | None:
+        """Load neutral auth from the legacy MCP file shape."""
+
+        record = self.load_auth(auth_ref)
+        if record is None:
+            return None
+        return neutral_auth_from_mcp(record)
+
     def save_catalog(self, snapshot: CatalogSnapshot) -> None:
         self._catalog_path(snapshot.connection_id).write_text(
             json.dumps(dump_catalog_snapshot(snapshot), indent=2),

tests/wf_api/test_auth.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import assert_type
+
+import pytest
+
+from wf_api.auth import AuthRecord, AuthStore, validate_auth_id
+
+
+def test_validate_auth_id_accepts_safe_dotted_ids() -> None:
+    assert validate_auth_id("github.work") == "github.work"
+    assert validate_auth_id("api_ci-1") == "api_ci-1"
+
+
+@pytest.mark.parametrize("auth_id", ["", ".hidden", "../secret", "bad/id"])
+def test_validate_auth_id_rejects_unsafe_ids(auth_id: str) -> None:
+    with pytest.raises(ValueError, match="auth id must start"):
+        validate_auth_id(auth_id)
+
+
+def test_auth_record_is_immutable_and_mapping_typed() -> None:
+    record = AuthRecord(
+        id="github.work",
+        scheme="bearer",
+        payload={"token": "secret"},
+        metadata={"owner": "test"},
+    )
+
+    assert record.id == "github.work"
+    assert record.scheme == "bearer"
+    assert record.payload["token"] == "secret"
+    assert_type(record.payload, Mapping[str, object])
+
+    with pytest.raises(AttributeError):
+        record.scheme = "headers"  # type: ignore[misc]
+
+
+class MemoryAuthStore:
+    def __init__(self, records: dict[str, AuthRecord]) -> None:
+        self.records = records
+
+    def load_auth(self, auth_ref: str) -> AuthRecord | None:
+        return self.records.get(auth_ref)
+
+
+def test_auth_store_protocol_is_read_only_lookup() -> None:
+    record = AuthRecord(id="github.work", scheme="opaque", payload={"x": 1})
+    store: AuthStore = MemoryAuthStore({"github.work": record})
+
+    assert store.load_auth("github.work") is record
+    assert store.load_auth("missing") is None

tests/wf_mcp/service/test_upstream_transport.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 from wf_artifacts import WorkflowDeployment
 from wf_platform import CapabilityBuckets, CapabilitySource, SourcePermissions
 
@@ -16,6 +18,14 @@ from ..test_support import FakeAdapter, local_temp_root
 from ..workflow_surface.conftest import echo_artifact
 
 
+def _transport(root: Path) -> UpstreamTransportService:
+    events: list[McpEvent] = []
+    return UpstreamTransportService(
+        store=FileStore(root),
+        event_sink=events.append,
+    )
+
+
 def test_upstream_transport_registers_adapter() -> None:
     events: list[McpEvent] = []
     transport = UpstreamTransportService(
@@ -98,7 +108,7 @@ async def test_upstream_transport_refreshes_catalog_directly() -> None:
         connection_list_enabled=connections.list_enabled,
         connection_list_all=connections.list_all,
         tool_executor_for=transport.tool_executor_for,
-        load_auth=transport.load_auth,
+        load_auth=transport.load_connection_auth,
         emit_event=events.append,
     )
     source_catalog.hydrate_connection_source_from_snapshot(connection)
@@ -131,7 +141,7 @@ async def test_upstream_transport_live_diagnostics_report_missing_connection() -
         connection_list_enabled=lambda: [],
         connection_list_all=lambda: [],
         tool_executor_for=transport.tool_executor_for,
-        load_auth=transport.load_auth,
+        load_auth=transport.load_connection_auth,
         emit_event=lambda event: None,
     )
     source_catalog.register_capability_source(
@@ -158,3 +168,82 @@ async def test_upstream_transport_live_diagnostics_report_missing_connection() -
 
     assert diagnostics[0].code == "source_unreachable"
     assert diagnostics[0].bound_source == "demo.personal"
+
+
+def test_upstream_load_connection_auth_prefers_auth_ref(tmp_path: Path) -> None:
+    service = _transport(tmp_path)
+    service.save_auth(
+        AuthRecord(
+            connection_id="github.creds",
+            scheme="bearer",
+            payload={"token": "secret"},
+        )
+    )
+    service.save_auth(
+        AuthRecord(
+            connection_id="github.work",
+            scheme="bearer",
+            payload={"token": "wrong"},
+        )
+    )
+    connection = ConnectionConfig(
+        id="github.work",
+        server="github",
+        account="work",
+        metadata={"auth_ref": "github.creds"},
+    )
+
+    assert service.load_connection_auth(connection) == AuthRecord(
+        connection_id="github.creds",
+        scheme="bearer",
+        payload={"token": "secret"},
+    )
+
+
+def test_upstream_load_connection_auth_falls_back_to_connection_id(
+    tmp_path: Path,
+) -> None:
+    service = _transport(tmp_path)
+    service.save_auth(
+        AuthRecord(
+            connection_id="github.work",
+            scheme="bearer",
+            payload={"token": "legacy"},
+        )
+    )
+    connection = ConnectionConfig(
+        id="github.work",
+        server="github",
+        account="work",
+    )
+
+    assert service.load_connection_auth(connection) == AuthRecord(
+        connection_id="github.work",
+        scheme="bearer",
+        payload={"token": "legacy"},
+    )
+
+
+def test_upstream_load_connection_auth_ignores_non_string_auth_ref(
+    tmp_path: Path,
+) -> None:
+    service = _transport(tmp_path)
+    service.save_auth(
+        AuthRecord(
+            connection_id="github.work",
+            scheme="bearer",
+            payload={"token": "legacy"},
+        )
+    )
+    connection = ConnectionConfig(
+        id="github.work",
+        server="github",
+        account="work",
+        metadata={"auth_ref": 123},
+    )
+
+    assert service.load_connection_auth(connection) == AuthRecord(
+        connection_id="github.work",
+        scheme="bearer",
+        payload={"token": "legacy"},
+    )

tests/wf_mcp/test_auth.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from wf_api.auth import AuthRecord as NeutralAuthRecord
+from wf_mcp.auth import (
+    mcp_auth_env,
+    mcp_auth_headers,
+    mcp_auth_from_neutral,
+    neutral_auth_from_mcp,
+)
+from wf_mcp.models import AuthRecord as McpAuthRecord
+from wf_mcp.storage import FileStore
+
+
+def test_mcp_auth_from_neutral_preserves_scheme_and_payload() -> None:
+    neutral = NeutralAuthRecord(
+        id="github.work",
+        scheme="bearer",
+        payload={"token": "secret"},
+        metadata={"owner": "test"},
+    )
+
+    mcp = mcp_auth_from_neutral(neutral)
+
+    assert mcp == McpAuthRecord(
+        connection_id="github.work",
+        scheme="bearer",
+        payload={"token": "secret"},
+    )
+
+
+def test_neutral_auth_from_mcp_preserves_payload() -> None:
+    mcp = McpAuthRecord(
+        connection_id="github.work",
+        scheme="headers",
+        payload={"headers": {"X-Test": "yes"}},
+    )
+
+    neutral = neutral_auth_from_mcp(mcp)
+
+    assert neutral.id == "github.work"
+    assert neutral.scheme == "headers"
+    assert neutral.payload == {"headers": {"X-Test": "yes"}}
+
+
+def test_mcp_auth_headers_extracts_explicit_headers_and_bearer_token() -> None:
+    auth = McpAuthRecord(
+        connection_id="api.work",
+        scheme="bearer",
+        payload={"headers": {"X-Test": "yes"}, "token": "secret"},
+    )
+
+    assert mcp_auth_headers(auth) == {
+        "X-Test": "yes",
+        "Authorization": "Bearer secret",
+    }
+
+
+def test_mcp_auth_headers_does_not_override_authorization_header() -> None:
+    auth = McpAuthRecord(
+        connection_id="api.work",
+        scheme="bearer",
+        payload={
+            "headers": {"Authorization": "Basic already"},
+            "token": "secret",
+        },
+    )
+
+    assert mcp_auth_headers(auth) == {"Authorization": "Basic already"}
+
+
+def test_mcp_auth_env_returns_string_map_only() -> None:
+    auth = McpAuthRecord(
+        connection_id="mcp.local",
+        scheme="env",
+        payload={"env": {"TOKEN": "secret", "BAD": 123}},
+    )
+
+    assert mcp_auth_env(auth) == {"TOKEN": "secret"}
+
+
+def test_file_store_saves_and_loads_neutral_auth_record(tmp_path: Path) -> None:
+    store = FileStore(tmp_path)
+    record = NeutralAuthRecord(
+        id="github.work",
+        scheme="bearer",
+        payload={"token": "secret"},
+        metadata={"owner": "test"},
+    )
+
+    store.save_auth_record(record)
+
+    loaded = store.load_auth_record("github.work")
+    assert loaded is not None
+    assert loaded.id == record.id
+    assert loaded.scheme == record.scheme
+    assert loaded.payload == record.payload
+    # Legacy file format does not persist metadata
+    assert loaded.metadata == {}
+
+
+def test_file_store_legacy_auth_methods_still_work(tmp_path: Path) -> None:
+    store = FileStore(tmp_path)
+    legacy = McpAuthRecord(
+        connection_id="github.work",
+        scheme="bearer",
+        payload={"token": "secret"},
+    )
+
+    store.save_auth(legacy)
+
+    assert store.load_auth("github.work") == legacy
+    assert store.load_auth_record("github.work") == NeutralAuthRecord(
+        id="github.work",
+        scheme="bearer",
+        payload={"token": "secret"},
+    )