feat: add source registry file store

docs/current_roadmap.md

@@ -120,6 +120,9 @@ implementation state.
       they are safe.
    - The store-backed source registry design is recorded in
       [2026-06-03 store-backed source registry](./superpowers/specs/2026-06-03-store-backed-source-registry-design.md).
+   - First source registry implementation slice complete: validated registry
+      models plus `FileSourceRegistryStore` exist, but startup merge and mutation
+      commands are still deferred.
    - Longer term: make the MCP frontend an adapter over these neutral workflow,
       source-admin, and config-admin surfaces so the old `wf_mcp` server entry
       point can shrink or retire.

src/wf_mcp/source_registry.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Annotated, Literal, Protocol
+from uuid import uuid4
+
+from pydantic import (
+    AnyHttpUrl,
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+    model_validator,
+)
+
+from .connections import parse_connection_id
+from .shared.names import RESERVED_CONNECTION_IDS
+
+
+class SourceRegistryModel(BaseModel):
+    """Base model for persisted source registry state; reject misspelled fields."""
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class StdioSourceTransport(SourceRegistryModel):
+    kind: Literal["stdio"] = "stdio"
+    command: str = Field(min_length=1)
+    args: tuple[str, ...] = ()
+    env: dict[str, str] = Field(default_factory=dict)
+
+
+class HttpSourceTransport(SourceRegistryModel):
+    kind: Literal["http"] = "http"
+    url: AnyHttpUrl
+    headers: dict[str, str] = Field(default_factory=dict)
+
+
+SourceTransport = Annotated[
+    StdioSourceTransport | HttpSourceTransport,
+    Field(discriminator="kind"),
+]
+
+
+class McpSourceRegistryEntry(SourceRegistryModel):
+    """Desired MCP source configuration persisted by server-owned mutation."""
+
+    id: str
+    kind: Literal["mcp"] = "mcp"
+    enabled: bool = True
+    provider: str = Field(min_length=1)
+    account: str = Field(min_length=1)
+    profile: str | None = None
+    transport: SourceTransport
+    auth_ref: str | None = None
+    metadata: dict[str, object] = Field(default_factory=dict)
+
+    @field_validator("id")
+    @classmethod
+    def validate_id(cls, value: str) -> str:
+        parse_connection_id(value)
+        if value in RESERVED_CONNECTION_IDS:
+            raise ValueError(f"source id {value!r} is reserved")
+        return value
+
+
+class SourceRegistryFile(SourceRegistryModel):
+    version: Literal[1] = 1
+    sources: list[McpSourceRegistryEntry] = Field(default_factory=list)
+
+    @model_validator(mode="after")
+    def validate_unique_source_ids(self) -> SourceRegistryFile:
+        seen: set[str] = set()
+        for source in self.sources:
+            if source.id in seen:
+                raise ValueError(f"duplicate source id {source.id!r}")
+            seen.add(source.id)
+        return self
+
+    def source_map(self) -> dict[str, McpSourceRegistryEntry]:
+        return {source.id: source for source in self.sources}
+
+
+class SourceRegistryStore(Protocol):
+    """Persistence boundary for desired server-owned source configuration."""
+
+    def load_registry(self) -> SourceRegistryFile:
+        """Return the stored registry, or an empty registry when absent."""
+        ...
+
+    def save_registry(self, registry: SourceRegistryFile) -> None:
+        """Persist one validated registry atomically."""
+        ...
+
+
+class FileSourceRegistryStore:
+    """Filesystem implementation for desired source registry state."""
+
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    @property
+    def path(self) -> Path:
+        return self.root / "source_registry.json"
+
+    def load_registry(self) -> SourceRegistryFile:
+        if not self.path.exists():
+            return SourceRegistryFile()
+        try:
+            data = json.loads(self.path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"source registry file is corrupted: {self.path}") from exc
+        return SourceRegistryFile.model_validate(data)
+
+    def save_registry(self, registry: SourceRegistryFile) -> None:
+        validated = SourceRegistryFile.model_validate(registry.model_dump(mode="json"))
+        payload = json.dumps(validated.model_dump(mode="json"), indent=2)
+        # Use a unique temp file so multiple store objects pointing at the same
+        # root do not trample each other's pending writes before replacement.
+        tmp_path = self.path.with_name(f"{self.path.name}.{uuid4().hex}.tmp")
+        tmp_path.write_text(payload, encoding="utf-8")
+        tmp_path.replace(self.path)
+
+
+__all__ = [
+    "FileSourceRegistryStore",
+    "HttpSourceTransport",
+    "McpSourceRegistryEntry",
+    "SourceRegistryFile",
+    "SourceRegistryStore",
+    "SourceTransport",
+    "StdioSourceTransport",
+]

tests/wf_mcp/test_source_registry.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from wf_mcp.source_registry import (
+    FileSourceRegistryStore,
+    HttpSourceTransport,
+    McpSourceRegistryEntry,
+    SourceRegistryFile,
+    StdioSourceTransport,
+)
+
+
+def _entry(source_id: str = "github.work") -> McpSourceRegistryEntry:
+    return McpSourceRegistryEntry(
+        id=source_id,
+        provider="github",
+        account="work",
+        transport=StdioSourceTransport(
+            command="npx",
+            args=("-y", "@modelcontextprotocol/server-github"),
+            env={"GITHUB_TOKEN": "${GITHUB_TOKEN}"},
+        ),
+        auth_ref=source_id,
+        metadata={"purpose": "tests"},
+    )
+
+
+def test_source_registry_entry_keeps_identity_and_transport_structural() -> None:
+    entry = _entry()
+
+    assert entry.id == "github.work"
+    assert entry.provider == "github"
+    assert entry.account == "work"
+    assert entry.profile is None
+    assert entry.transport.kind == "stdio"
+    assert entry.transport.command == "npx"
+    assert entry.auth_ref == "github.work"
+
+
+def test_source_registry_accepts_http_transport() -> None:
+    entry = McpSourceRegistryEntry(
+        id="github.http",
+        provider="github",
+        account="work",
+        transport=HttpSourceTransport(url="https://example.test/mcp"),  # type: ignore[arg-type]
+    )
+
+    assert entry.transport.kind == "http"
+    assert str(entry.transport.url) == "https://example.test/mcp"
+
+
+def test_source_registry_rejects_duplicate_ids() -> None:
+    with pytest.raises(ValueError, match="duplicate source id 'github.work'"):
+        SourceRegistryFile(sources=[_entry("github.work"), _entry("github.work")])
+
+
+def test_source_registry_rejects_reserved_ids() -> None:
+    with pytest.raises(ValueError, match="reserved"):
+        _entry("wf.admin")
+
+
+def test_source_registry_rejects_unsafe_ids() -> None:
+    with pytest.raises(ValueError, match="connection id"):
+        _entry("../bad")
+
+
+def test_file_source_registry_store_loads_empty_registry_when_missing(
+    tmp_path: Path,
+) -> None:
+    store = FileSourceRegistryStore(tmp_path)
+
+    registry = store.load_registry()
+
+    assert registry.version == 1
+    assert registry.sources == []
+    assert store.path == tmp_path / "source_registry.json"
+
+
+def test_file_source_registry_store_round_trips_registry(tmp_path: Path) -> None:
+    store = FileSourceRegistryStore(tmp_path)
+    registry = SourceRegistryFile(sources=[_entry("github.work")])
+
+    store.save_registry(registry)
+    loaded = store.load_registry()
+
+    assert loaded.source_map()["github.work"].provider == "github"
+    assert loaded.source_map()["github.work"].transport.kind == "stdio"
+
+
+def test_file_source_registry_store_validates_loaded_registry(tmp_path: Path) -> None:
+    store = FileSourceRegistryStore(tmp_path)
+    store.path.write_text(
+        '{"version": 1, "sources": [{"id": "wf.admin", "provider": "wf", '
+        '"account": "admin", "transport": {"kind": "stdio", "command": "x"}}]}',
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="reserved"):
+        store.load_registry()
+
+
+def test_file_source_registry_store_rejects_corrupted_json(tmp_path: Path) -> None:
+    store = FileSourceRegistryStore(tmp_path)
+    store.path.write_text("not json{{{", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="corrupted"):
+        store.load_registry()