feat: merge source registry at startup

2026-06-04 00:29:44 +07:00
parent 4e8726f36d
commit 2433d58988
10 changed files with 273 additions and 24 deletions
@@ -120,15 +120,15 @@ 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.
+   - First source registry implementation slices complete: validated registry
+      models, `FileSourceRegistryStore`, generic `wf_api` registry mechanics,
+      MCP entry conversion, and startup merge are implemented.
+   - Source registry startup merge is implemented: absent registry preserves
+      config-only behavior, registry-only entries hydrate as dynamic connections,
+      and config entries shadow same-id registry entries with an event.
   - Next source registry slices are planned in
      [2026-06-03 source registry next slices](./superpowers/plans/2026-06-03-source-registry-next-slices.md):
-      generic registry mechanics now live in `wf_api`, MCP entry conversion is
-      explicit and tested, and the next executable slice is
-      [startup merge](./superpowers/plans/2026-06-03-source-registry-startup-merge.md).
-      Desired-registry admin reads and safe mutation commands remain later
+      desired-registry admin reads and safe mutation commands remain later
      slices.
   - 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
@@ -53,7 +53,7 @@ The next executable slice is startup merge:
   - Keep config merge out of scope.

 3. **Slice 3: Startup Merge**
-   - **Status: planned.**
+   - **Status: complete.**
   - Load registry at broker/server startup.
   - Merge config and registry with config precedence.
   - Emit events/diagnostics for shadowed registry entries.
@@ -262,13 +262,10 @@ profile, transport details, enabled state, and source-registry origin.

 ### Slice 3: Startup Merge

-Implementation plan:
-[2026-06-03 source registry startup merge](../plans/2026-06-03-source-registry-startup-merge.md).
-
- Load registry during server/broker construction.
- Merge config + registry deterministically.
- Emit diagnostics/events for ignored shadowed entries.
- Preserve existing config-only behavior when registry file is absent.
+Status: complete. Broker/service construction now loads `source_registry.json`,
+merges config-defined connections with dynamic registry entries, preserves config
+precedence, and emits `source_registry_ignored_config_shadow` for shadowed
+registry entries.

 ### Slice 4: Read Registry Through Admin

@@ -9,6 +9,7 @@ from ..control import BrokerConfigFile
 from ..models import BrokerConfig
 from ..runtime import McpRuntimePool, PersistentSessionFactory
 from ..sdk import McpSdkAdapter
+from ..source_registry import FileSourceRegistryStore
 from ..storage import FileStore
 from .service import WfMcpService

@@ -34,8 +35,12 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
        # across sequential workflow nodes.
        tool_executor=McpRuntimePool(runtime_factory.create),
    )
-    for connection in config.connections:
-        service.register_connection(connection)
+    source_registry_store = FileSourceRegistryStore(config.store_root)
+    service.sync_connections_from_config(
+        config,
+        source_registry_store=source_registry_store,
+    )
+    for connection in service.connections.list_all():
        if connection.server not in service.adapters:
            service.register_adapter(connection.server, McpSdkAdapter())
    return service
@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
 from ...connections import ConnectionRegistry, parse_connection_id
 from ...models import BrokerConfig, ConnectionConfig
 from ...shared.names import RESERVED_CONNECTION_IDS
+from ...source_registry import SourceRegistryStore, registry_entry_to_connection_config
 from .events import BrokerEventRecorder
 from .source_catalog import SourceCatalogService

@@ -53,10 +54,35 @@ class ConnectionService:
            payload={"server": connection.server, "account": connection.account},
        )

-    def sync_connections_from_config(self, config: BrokerConfig) -> None:
+    def sync_connections_from_config(
+        self,
+        config: BrokerConfig,
+        *,
+        source_registry_store: SourceRegistryStore | None = None,
+    ) -> None:
        """Reconcile registry/source state after the public server reloads config."""
+        # Config-defined connections win over registry entries with the same id;
+        # registry entries fill ids not present in config.
+        connections = list(config.connections)
+        config_ids = {connection.id for connection in connections}
+        if source_registry_store is not None:
+            registry = source_registry_store.load_registry()
+            for entry in registry.sources:
+                if entry.id in config_ids:
+                    self.events.record_kind(
+                        "source_registry_ignored_config_shadow",
+                        connection_id=entry.id,
+                        payload={
+                            "server": entry.provider,
+                            "account": entry.account,
+                            "reason": "config_connection_takes_precedence",
+                        },
+                    )
+                    continue
+                connections.append(registry_entry_to_connection_config(entry))
+
        source_catalog = self._source_catalog()
-        next_ids = {connection.id for connection in config.connections}
+        next_ids = {connection.id for connection in connections}
        previous_ids = set(self.connections.connections)
        for connection_id in previous_ids - next_ids:
            previous = self.connections.connections[connection_id]
@@ -71,7 +97,7 @@ class ConnectionService:
                payload={"server": previous.server, "account": previous.account},
            )

-        for connection in config.connections:
+        for connection in connections:
            self._validate_connection_id(connection.id)
            previous = self.connections.connections.get(connection.id)
            self.connections.register(connection)
@@ -32,6 +32,7 @@ from ...models import (
 )
 from ...sdk import BackendAdapter
 from ...runtime import ToolExecutor
+from ...source_registry import SourceRegistryStore
 from .connection_service import ConnectionService
 from .content_access import ContentAccessService
 from ...storage import Store
@@ -138,8 +139,16 @@ class WfMcpService:
    def register_connection(self, connection: ConnectionConfig) -> None:
        self.connection_service.register_connection(connection)

-    def sync_connections_from_config(self, config: BrokerConfig) -> None:
-        self.connection_service.sync_connections_from_config(config)
+    def sync_connections_from_config(
+        self,
+        config: BrokerConfig,
+        *,
+        source_registry_store: SourceRegistryStore | None = None,
+    ) -> None:
+        self.connection_service.sync_connections_from_config(
+            config,
+            source_registry_store=source_registry_store,
+        )

    def register_adapter(self, server: str, adapter: BackendAdapter) -> None:
        self.upstream.register_adapter(server, adapter)
@@ -13,6 +13,7 @@ from ..broker.transport import normalize_transport
 from ..documentation import build_local_documentation_source
 from ..models import BrokerConfig
 from ..sdk import McpSdkAdapter
+from ..source_registry import FileSourceRegistryStore
 from ..proxy.runtime import ProxyRuntime
 from ..workflow_surface import register_workflow_tools
 from .prompts import register_documentation_prompts
@@ -33,8 +34,11 @@ def create_server(
    service = build_service_from_config(config)

    def sync_service(config: BrokerConfig) -> None:
-        service.sync_connections_from_config(config)
-        for connection in config.connections:
+        service.sync_connections_from_config(
+            config,
+            source_registry_store=FileSourceRegistryStore(config.store_root),
+        )
+        for connection in service.connections.list_all():
            if connection.server not in service.adapters:
                service.register_adapter(connection.server, McpSdkAdapter())

@@ -9,6 +9,12 @@ from mcp import types as mcp_types
 from wf_mcp.broker.config import load_broker_config
 from wf_mcp.models import BrokerConfig
 from wf_mcp.server import create_server_client
+from wf_mcp.source_registry import (
+    FileSourceRegistryStore,
+    McpSourceRegistryEntry,
+    SourceRegistryFile,
+    StdioSourceTransport,
+)

 from ..test_support import fixture_server_path, local_temp_root
 from .conftest import structured
@@ -178,3 +184,45 @@ def test_server_reload_syncs_service_connection_source_enabled_state() -> None:
            assert "fixture.personal.echo_tool" in names

    asyncio.run(run_proxy())
+
+
+def test_server_reload_preserves_source_registry_connections() -> None:
+    tmp_path = local_temp_root() / "unified_reload_registry_source_store"
+    tmp_path.mkdir(parents=True, exist_ok=True)
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(
+        json.dumps({"store_root": ".wf_mcp_store", "connections": []}),
+        encoding="utf-8",
+    )
+    config = load_broker_config(config_path)
+    FileSourceRegistryStore(config.store_root).save_registry(
+        SourceRegistryFile(
+            sources=[
+                McpSourceRegistryEntry(
+                    id="fixture.registry",
+                    kind="mcp",
+                    enabled=True,
+                    provider="fixture",
+                    account="registry",
+                    transport=StdioSourceTransport(command=sys.executable),
+                )
+            ]
+        )
+    )
+
+    async def run_proxy() -> None:
+        client = create_server_client(config, config_path=config_path)
+        async with client:
+            before = await client.call_tool("wf.admin.list_sources", {"limit": 100})
+            before_ids = {
+                source["id"] for source in structured(before)["sources"]
+            }
+            assert "fixture.registry" in before_ids
+
+            await client.call_tool("wf.admin.reload_config")
+
+            after = await client.call_tool("wf.admin.list_sources", {"limit": 100})
+            after_ids = {source["id"] for source in structured(after)["sources"]}
+            assert "fixture.registry" in after_ids
+
+    asyncio.run(run_proxy())
@@ -7,6 +7,12 @@ from wf_mcp.broker.service.source_catalog import SourceCatalogService
 from wf_mcp.events import EventBus
 from wf_mcp.models import BrokerConfig, ConnectionConfig
 from wf_mcp.runtime import ToolExecutor
+from wf_mcp.source_registry import (
+    FileSourceRegistryStore,
+    McpSourceRegistryEntry,
+    SourceRegistryFile,
+    StdioSourceTransport,
+)
 from wf_mcp.storage import FileStore

 from ..test_support import local_temp_root
@@ -174,3 +180,93 @@ def test_wfmcpservice_sync_connections_delegates_to_connection_service() -> None
    ]
    assert "demo.personal" not in service.capability_sources
    assert "demo.work" in service.capability_sources
+
+
+# ---------------------------------------------------------------------------
+# Source registry merge helpers and tests
+# ---------------------------------------------------------------------------
+
+
+def _registry_entry(
+    source_id: str = "demo.registry",
+    *,
+    enabled: bool = True,
+) -> McpSourceRegistryEntry:
+    return McpSourceRegistryEntry(
+        id=source_id,
+        kind="mcp",
+        enabled=enabled,
+        provider="demo",
+        account=source_id.rsplit(".", 1)[-1],
+        transport=StdioSourceTransport(command="demo-server"),
+    )
+
+
+def test_connection_service_sync_merges_registry_entries() -> None:
+    service = ConnectionService(events=BrokerEventRecorder(EventBus()))
+    catalog = _source_catalog(service)
+    store = FileSourceRegistryStore(local_temp_root() / "registry_merge")
+    store.save_registry(SourceRegistryFile(sources=[_registry_entry()]))
+
+    service.sync_connections_from_config(
+        BrokerConfig(store_root=local_temp_root(), connections=[]),
+        source_registry_store=store,
+    )
+
+    assert [connection.id for connection in service.list_all()] == ["demo.registry"]
+    assert "demo.registry" in catalog.capability_sources
+
+
+def test_connection_service_sync_config_shadows_registry_entry() -> None:
+    service = ConnectionService(events=BrokerEventRecorder(EventBus()))
+    _source_catalog(service)
+    store = FileSourceRegistryStore(local_temp_root() / "registry_shadow")
+    store.save_registry(SourceRegistryFile(sources=[_registry_entry("demo.same")]))
+
+    service.sync_connections_from_config(
+        BrokerConfig(
+            store_root=local_temp_root(),
+            connections=[
+                ConnectionConfig(id="demo.same", server="demo", account="config"),
+            ],
+        ),
+        source_registry_store=store,
+    )
+
+    assert service.get("demo.same").account == "config"
+    assert any(
+        event.kind == "source_registry_ignored_config_shadow"
+        and event.connection_id == "demo.same"
+        for event in service.events.list_events()
+    )
+
+
+def test_connection_service_sync_registry_disabled_entry_hydrates_disabled_source() -> None:
+    service = ConnectionService(events=BrokerEventRecorder(EventBus()))
+    catalog = _source_catalog(service)
+    store = FileSourceRegistryStore(local_temp_root() / "registry_disabled")
+    store.save_registry(SourceRegistryFile(sources=[_registry_entry(enabled=False)]))
+
+    service.sync_connections_from_config(
+        BrokerConfig(store_root=local_temp_root(), connections=[]),
+        source_registry_store=store,
+    )
+
+    assert service.get("demo.registry").enabled is False
+    assert catalog.capability_sources["demo.registry"].enabled is False
+
+
+def test_wfmcpservice_sync_connections_delegates_registry_store() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "facade_registry"))
+    store = FileSourceRegistryStore(local_temp_root() / "facade_registry_store")
+    store.save_registry(SourceRegistryFile(sources=[_registry_entry()]))
+
+    service.sync_connections_from_config(
+        BrokerConfig(store_root=local_temp_root(), connections=[]),
+        source_registry_store=store,
+    )
+
+    assert [connection.id for connection in service.connections.list_all()] == [
+        "demo.registry"
+    ]
+    assert "demo.registry" in service.capability_sources
@@ -19,6 +19,12 @@ from wf_mcp.broker import (
    load_broker_config,
 )
 from wf_mcp.models import BrokerConfig, ConnectionConfig
+from wf_mcp.source_registry import (
+    FileSourceRegistryStore,
+    McpSourceRegistryEntry,
+    SourceRegistryFile,
+    StdioSourceTransport,
+)
 from wf_mcp.storage import FileStore

 from .test_support import (
@@ -485,6 +491,64 @@ def test_build_service_from_config_uses_store_root_for_workflow_stores() -> None
    assert service.run_store.root == store_root


+# ---------------------------------------------------------------------------
+# Source registry integration tests
+# ---------------------------------------------------------------------------
+
+
+def _registry_entry(
+    source_id: str = "demo.registry",
+    *,
+    enabled: bool = True,
+) -> McpSourceRegistryEntry:
+    return McpSourceRegistryEntry(
+        id=source_id,
+        kind="mcp",
+        enabled=enabled,
+        provider="demo",
+        account=source_id.rsplit(".", 1)[-1],
+        transport=StdioSourceTransport(command="demo-server"),
+    )
+
+
+def test_build_service_from_config_loads_source_registry_entries() -> None:
+    tmp_path = local_temp_root() / "broker_config_registry_load"
+    tmp_path.mkdir(parents=True, exist_ok=True)
+    config = BrokerConfig(store_root=tmp_path, connections=[])
+    FileSourceRegistryStore(tmp_path).save_registry(
+        SourceRegistryFile(sources=[_registry_entry("fixture.registry")])
+    )
+
+    service = build_service_from_config(config)
+
+    assert service.connections.get("fixture.registry").server == "demo"
+    assert "demo" in service.adapters
+    assert "fixture.registry" in service.capability_sources
+
+
+def test_build_service_from_config_config_shadows_registry() -> None:
+    tmp_path = local_temp_root() / "broker_config_registry_shadow"
+    tmp_path.mkdir(parents=True, exist_ok=True)
+    config = BrokerConfig(
+        store_root=tmp_path,
+        connections=[
+            ConnectionConfig(id="fixture.same", server="fixture", account="config"),
+        ],
+    )
+    FileSourceRegistryStore(tmp_path).save_registry(
+        SourceRegistryFile(sources=[_registry_entry("fixture.same")])
+    )
+
+    service = build_service_from_config(config)
+
+    assert service.connections.get("fixture.same").account == "config"
+    assert any(
+        event.kind == "source_registry_ignored_config_shadow"
+        and event.connection_id == "fixture.same"
+        for event in service.list_events()
+    )
+
+
 def _artifact() -> WorkflowArtifact:
    return WorkflowArtifact(
        id="summarize_docs",