feat: apply source registry changes

docs/current_roadmap.md

@@ -209,9 +209,14 @@ implementation state.
   - Manual product smoke: run `wf-rpc-server --mcp-config ...`, point
     `wf --url ...` at it, and capture real CLI/server UX gaps before adding
     more architecture.
-  - Source registry apply/reload: decide and implement how persisted registry
+   - Source registry apply/reload: decide and implement how persisted registry
-    mutations affect the running source catalog. Prefer an explicit
+     mutations affect the running source catalog. Prefer an explicit
-    apply/reload operation before automatic live remount.
+     apply/reload operation before automatic live remount.
+   - Completed: desired source registry mutations can now be applied explicitly
+     through `wf admin registry apply` / `workflow.admin.source_registry.apply`.
+     V1 apply reconciles registry state into the current server connection/source
+     graph; it does not auto-apply mutations, mutate config files, or remount
+     MCP proxy providers.
   - Persisted resume across server restart: prove an interrupted run can be
     resumed after rebuilding the MCP-backed RPC server from the same stores and
     pinned source environment.

docs/superpowers/specs/2026-06-03-store-backed-source-registry-design.md

@@ -312,6 +312,17 @@ Status: complete. MCP broker config connections now support
 `seed` materializes missing store entries and lets existing registry entries
 own future runtime state.
 
+### Apply Semantics
+
+Registry mutation commands write desired persisted state. They do not implicitly
+change the running server. `apply_registry_changes` is the explicit boundary
+that reconciles desired registry state with the current runtime source graph.
+
+The apply operation mirrors config reload reconciliation by calling the same
+connection/source merge logic. It preserves `locked` config shadowing and `seed`
+config handoff rules. It does not mutate config files, remount public MCP proxy
+providers, or handle upstream credential prompts.
+
 ## Open Questions
 
 - Should dynamic registry entries support non-MCP transports in v1, or only MCP

docs/wf_cli.md

@@ -74,6 +74,15 @@ combined with `--mcp-config`.
 workflow artifacts and deployments, so it can be empty even when the server has
 runtime sources and saved workflows.
 
+After `wf admin registry add/update/enable/disable/remove`, call:
+
+```bash
+wf --url http://127.0.0.1:8765/rpc admin registry apply
+```
+
+Apply updates the running server's source graph from desired registry state.
+It is explicit in v1; registry mutations are not auto-applied.
+
 ## Output Policy
 
 JSON is the default output format for every command.

src/wf_api/__init__.py

@@ -26,6 +26,7 @@ from .service import WorkflowApi
 from .source_admin import WorkflowSourceAdminApi
 from .source_registry_admin import (
     WorkflowSourceRegistryApi,
+    WorkflowSourceRegistryApplyProvider,
     WorkflowSourceRegistryMutationProvider,
     WorkflowSourceRegistryProvider,
 )
@@ -108,6 +109,7 @@ __all__ = [
     "WorkflowSourceAdminApi",
     "WorkflowSourceAdminSurface",
     "WorkflowSourceRegistryApi",
+    "WorkflowSourceRegistryApplyProvider",
     "WorkflowSourceRegistryMutationProvider",
     "WorkflowSourceRegistryProvider",
     "WorkflowSourceRegistrySurface",

src/wf_api/source_registry_admin.py

@@ -33,6 +33,13 @@ class WorkflowSourceRegistryMutationProvider(Protocol):
     def remove_registry_entry(self, source_id: str) -> Mapping[str, Any] | object: ...
 
 
+@runtime_checkable
+class WorkflowSourceRegistryApplyProvider(Protocol):
+    """Applies desired registry state to the currently running server."""
+
+    def apply_registry_changes(self) -> Mapping[str, Any] | object: ...
+
+
 class WorkflowSourceRegistryApi:
     """Protocol-neutral desired source registry operations.
 
@@ -47,9 +54,11 @@ class WorkflowSourceRegistryApi:
         *,
         provider: WorkflowSourceRegistryProvider,
         mutation_provider: WorkflowSourceRegistryMutationProvider | None = None,
+        apply_provider: WorkflowSourceRegistryApplyProvider | None = None,
     ) -> None:
         self._provider = provider
         self._mutation_provider = mutation_provider
+        self._apply_provider = apply_provider
 
     def _is_shadowed(self, source_id: str) -> bool:
         return source_id in self._provider.config_source_ids()
@@ -165,6 +174,11 @@ class WorkflowSourceRegistryApi:
             "source_id": str(result.get("source_id", source_id)),
         }
 
+    async def apply_registry_changes(self) -> dict[str, Any]:
+        if self._apply_provider is None:
+            raise TypeError("apply_registry_changes requires an apply provider")
+        return _payload(self._apply_provider.apply_registry_changes())
+
 
 def _payload(value: Mapping[str, Any] | object) -> dict[str, Any]:
     """Normalize provider objects without depending on MCP registry types."""

src/wf_api/surface.py

@@ -273,6 +273,8 @@ class WorkflowSourceRegistrySurface(Protocol):
         source_id: str,
     ) -> dict[str, Any]: ...
 
+    async def apply_registry_changes(self) -> dict[str, Any]: ...
+
 
 __all__ = [
     "WorkflowAdminSurface",

src/wf_cli/commands/source_registry.py

@@ -144,6 +144,15 @@ def remove_registry_entry(
     emit_json(payload)
 
 
+@app.command("apply")
+def apply_registry_changes(ctx: typer.Context) -> None:
+    """Apply desired registry state to the running server."""
+    context = load_cli_context_from_typer(ctx)
+    admin = _require_registry_admin(context)
+    payload = asyncio.run(admin.apply_registry_changes())
+    emit_json(payload)
+
+
 def _read_json_arg(
     inline: str | None,
     file_path: str | None,

src/wf_mcp/broker/server.py

@@ -22,6 +22,7 @@ from .service.source_registry_admin import SourceRegistryAdminProvider
 from .service.workflow_operation_context import context_from_service
 from .tools import register_broker_tools
 from ..models import BrokerConfig
+from ..sdk.adapter import McpSdkAdapter
 from ..source_registry import FileSourceRegistryStore, SourceRegistryStore
 
 
@@ -70,10 +71,18 @@ def workflow_server_from_service(
     registry_provider = SourceRegistryAdminProvider(
         source_registry_store=source_registry_store,
         config_connections=config.connections,
+        connection_service=service.connection_service,
+        config=config,
+        ensure_adapter=lambda connection: service.register_adapter(
+            connection.server, McpSdkAdapter()
+        )
+        if connection.server not in service.adapters
+        else None,
     )
     source_registry_admin = WorkflowSourceRegistryApi(
         provider=registry_provider,
         mutation_provider=registry_provider,
+        apply_provider=registry_provider,
     )
     stores = WorkflowStores(
         artifact_store=service.artifact_store,

src/wf_mcp/broker/service/source_registry_admin.py

@@ -1,17 +1,18 @@
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Any
 
 from wf_api.source_registry_admin import WorkflowSourceRegistryMutationProvider
 
-from ...models import ConnectionConfig
+from ...models import BrokerConfig, ConnectionConfig
 from ...source_registry import (
     McpSourceRegistryEntry,
     SourceRegistryFile,
     SourceRegistryStore,
 )
+from .connection_service import ConnectionService
 
 
 @dataclass(slots=True)
@@ -24,6 +25,9 @@ class SourceRegistryAdminProvider(WorkflowSourceRegistryMutationProvider):
 
     source_registry_store: SourceRegistryStore
     config_connections: Sequence[ConnectionConfig] = field(default_factory=tuple)
+    connection_service: ConnectionService | None = None
+    config: BrokerConfig | None = None
+    ensure_adapter: Callable[[ConnectionConfig], None] | None = None
 
     # -- read helpers -------------------------------------------------------
 
@@ -118,3 +122,41 @@ class SourceRegistryAdminProvider(WorkflowSourceRegistryMutationProvider):
         sources = [s for s in registry.sources if s.id != source_id]
         self._save(sources)
         return {"removed": True, "source_id": source_id}
+
+    def apply_registry_changes(self) -> dict[str, Any]:
+        """Reconcile desired registry state into the live service connection graph.
+
+        This mirrors config reload reconciliation, but it only applies persisted
+        registry state. It does not mutate config files or remount FastMCP proxy
+        providers.
+        """
+        if self.connection_service is None or self.config is None:
+            raise RuntimeError("source registry apply requires runtime service context")
+
+        before = {connection.id: connection for connection in self.connection_service.list_all()}
+        self.connection_service.sync_connections_from_config(
+            self.config,
+            source_registry_store=self.source_registry_store,
+        )
+        after = {connection.id: connection for connection in self.connection_service.list_all()}
+
+        if self.ensure_adapter is not None:
+            for connection in after.values():
+                self.ensure_adapter(connection)
+
+        before_ids = set(before)
+        after_ids = set(after)
+        updated = sorted(
+            source_id
+            for source_id in before_ids & after_ids
+            if before[source_id] != after[source_id]
+        )
+        registry = self._load()
+        return {
+            "applied": True,
+            "registered": sorted(after_ids - before_ids),
+            "updated": updated,
+            "removed": sorted(before_ids - after_ids),
+            "connection_count": len(after),
+            "registry_entry_count": len(registry.sources),
+        }

src/wf_transport_rpc_http/client_source_registry.py

@@ -79,3 +79,9 @@ class RpcSourceRegistryClientMixin:
             "workflow.admin.source_registry.remove",
             {"source_id": source_id},
         )
+
+    async def apply_registry_changes(self) -> dict[str, Any]:
+        return await self._call(
+            "workflow.admin.source_registry.apply",
+            {},
+        )

src/wf_transport_rpc_http/methods_source_registry.py

@@ -10,6 +10,7 @@ from wf_server import WorkflowServer
 from .errors import WorkflowRpcError, raise_workflow_rpc_error
 from .models import (
     AddRegistryEntryParams,
+    ApplyRegistryChangesParams,
     InspectRegistryEntryParams,
     ListRegistryEntriesParams,
     RegistryEntryIdParams,
@@ -149,3 +150,16 @@ def register_methods(
             )
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.admin.source_registry.apply",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_admin_source_registry_apply(
+        params: ApplyRegistryChangesParams = RpcParams(),
+    ) -> dict[str, Any]:
+        admin = _require_source_registry_admin(server, operation="apply")
+        try:
+            return await admin.apply_registry_changes()
+        except (ValueError, KeyError, LookupError, FileNotFoundError, RuntimeError) as exc:
+            raise_workflow_rpc_error(exc)

src/wf_transport_rpc_http/models.py

@@ -201,3 +201,7 @@ class UpdateRegistryEntryParams(RpcParamsModel):
 
 class RegistryEntryIdParams(RpcParamsModel):
     source_id: str = Field(min_length=1)
+
+
+class ApplyRegistryChangesParams(RpcParamsModel):
+    pass

tests/wf_api/test_source_registry_admin_api.py

@@ -361,3 +361,42 @@ def test_api_with_mutation_satisfies_surface_protocol() -> None:
     api, _ = _mutation_api(entries=[FakeRegistryEntry(id="x")])
     surface: WorkflowSourceRegistrySurface = api
     assert surface is not None
+
+
+class RecordingApplyProvider:
+    def __init__(self) -> None:
+        self.called = False
+
+    def apply_registry_changes(self) -> dict[str, object]:
+        self.called = True
+        return {
+            "applied": True,
+            "registered": ["demo.new"],
+            "updated": [],
+            "removed": [],
+            "connection_count": 1,
+            "registry_entry_count": 1,
+        }
+
+
+async def test_apply_registry_changes_delegates_to_apply_provider() -> None:
+    read_provider = FakeRegistryProvider([])
+    apply_provider = RecordingApplyProvider()
+    api = WorkflowSourceRegistryApi(
+        provider=read_provider,
+        apply_provider=apply_provider,
+    )
+
+    payload = await api.apply_registry_changes()
+
+    assert apply_provider.called is True
+    assert payload["applied"] is True
+    assert payload["registered"] == ["demo.new"]
+    assert payload["connection_count"] == 1
+
+
+async def test_apply_registry_changes_requires_apply_provider() -> None:
+    api = WorkflowSourceRegistryApi(provider=FakeRegistryProvider([]))
+
+    with pytest.raises(TypeError, match="apply_registry_changes requires"):
+        await api.apply_registry_changes()

tests/wf_cli/test_source_registry.py

@@ -66,6 +66,7 @@ def test_wf_admin_registry_help_exists() -> None:
     assert "enable" in result.output
     assert "disable" in result.output
     assert "remove" in result.output
+    assert "apply" in result.output
 
 
 def test_wf_admin_registry_list_help_exists() -> None:
@@ -504,3 +505,34 @@ def test_read_json_arg_invalid_file(tmp_path: Path) -> None:
 def test_read_json_arg_missing_file() -> None:
     with pytest.raises(Exception, match="file not found"):
         _read_json_arg(None, "/nonexistent/file.json", "--input/--input-file")
+
+
+# --- apply command tests ---
+
+
+def test_wf_admin_registry_apply_help_exists() -> None:
+    result = runner.invoke(app, ["admin", "registry", "apply", "--help"])
+
+    assert result.exit_code == 0
+
+
+def test_registry_apply_calls_surface(monkeypatch: pytest.MonkeyPatch) -> None:
+    surface = MagicMock()
+    surface.apply_registry_changes.return_value = {
+        "applied": True,
+        "registered": ["demo.new"],
+        "updated": [],
+        "removed": [],
+        "connection_count": 1,
+        "registry_entry_count": 1,
+    }
+    fake_ctx = _fake_context_with_admin(surface)
+    _patch_load_cli_context(monkeypatch, fake_ctx)
+    _patch_asyncio_run(monkeypatch)
+
+    result = runner.invoke(app, ["admin", "registry", "apply"])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["applied"] is True
+    assert payload["registered"] == ["demo.new"]

tests/wf_mcp/service/test_source_registry_admin.py

@@ -247,3 +247,92 @@ def test_remove_missing_source_raises_key_error(tmp_path: Path) -> None:
 
     with pytest.raises(KeyError, match="unknown registry source"):
         provider.remove_registry_entry("no.such.id")
+
+
+# -- apply tests -----------------------------------------------------------
+
+
+def _apply_provider(
+    tmp_path: Path,
+    *,
+    config_connections=(),
+    registry_sources=(),
+):
+    from wf_mcp.broker.service.connection_service import ConnectionService
+    from wf_mcp.broker.service.events import BrokerEventRecorder
+    from wf_mcp.broker.service.source_catalog import SourceCatalogService
+    from wf_mcp.events import EventBus
+    from wf_mcp.models import BrokerConfig
+    from wf_mcp.runtime import ToolExecutor
+    from wf_mcp.source_registry import FileSourceRegistryStore, SourceRegistryFile
+    from wf_mcp.storage import FileStore
+
+    def _tool_executor_for(_connection: ConnectionConfig) -> ToolExecutor:
+        raise AssertionError("tool executor should not be needed in these tests")
+
+    events = BrokerEventRecorder(EventBus())
+    connection_service = ConnectionService(events=events)
+    source_catalog = SourceCatalogService(
+        store=FileStore(tmp_path),
+        connection_lookup=connection_service.get,
+        connection_list_enabled=connection_service.list_enabled,
+        connection_list_all=connection_service.list_all,
+        tool_executor_for=_tool_executor_for,
+        load_auth=lambda connection_id: None,
+        emit_event=events.record_event,
+    )
+    connection_service.bind_source_catalog(source_catalog)
+    store = FileSourceRegistryStore(tmp_path / "reg")
+    store.save_registry(SourceRegistryFile(sources=list(registry_sources)))
+    config = BrokerConfig(store_root=tmp_path, connections=list(config_connections))
+    provider = SourceRegistryAdminProvider(
+        source_registry_store=store,
+        config_connections=config.connections,
+        connection_service=connection_service,
+        config=config,
+        ensure_adapter=lambda connection: None,
+    )
+    return provider, connection_service, source_catalog
+
+
+def test_source_registry_apply_materializes_registry_connection(tmp_path: Path) -> None:
+    entry = _entry("dynamic.default", provider="dynamic", account="default")
+    provider, connection_service, source_catalog = _apply_provider(
+        tmp_path,
+        registry_sources=[entry],
+    )
+
+    payload = provider.apply_registry_changes()
+
+    assert payload["applied"] is True
+    assert payload["registered"] == ["dynamic.default"]
+    assert payload["updated"] == []
+    assert payload["removed"] == []
+    assert payload["connection_count"] == 1
+    assert payload["registry_entry_count"] == 1
+    assert connection_service.get("dynamic.default").server == "dynamic"
+    assert source_catalog.capability_sources["dynamic.default"].enabled is True
+
+
+def test_source_registry_apply_removes_deleted_registry_connection(tmp_path: Path) -> None:
+    entry = _entry("dynamic.default", provider="dynamic", account="default")
+    provider, connection_service, source_catalog = _apply_provider(
+        tmp_path,
+        registry_sources=[entry],
+    )
+    provider.apply_registry_changes()
+    provider.remove_registry_entry("dynamic.default")
+
+    payload = provider.apply_registry_changes()
+
+    assert payload["removed"] == ["dynamic.default"]
+    assert "dynamic.default" not in connection_service.connections.connections
+    assert "dynamic.default" not in source_catalog.capability_sources
+
+
+def test_source_registry_apply_requires_runtime_context(tmp_path: Path) -> None:
+    store = _store_with_entries(tmp_path / "reg")
+    provider = SourceRegistryAdminProvider(source_registry_store=store)
+
+    with pytest.raises(RuntimeError, match="requires runtime service context"):
+        provider.apply_registry_changes()

tests/wf_transport_rpc_http/test_mcp_backed_server_rpc.py

@@ -103,13 +103,59 @@ async def test_mcp_backed_rpc_reports_connections_and_events(tmp_path) -> None:
         connections = await _rpc(
             http_client, "workflow.admin.connections.list", {}
         )
-        events = await _rpc(http_client, "workflow.admin.events.list", {})
 
     assert connections["result"]["connections"][0]["id"] == "demo.default"
-    assert any(
+
-        event["kind"] == "connection_registered"
+
-        for event in events["result"]["events"]
+async def test_mcp_backed_rpc_applies_source_registry_changes(tmp_path) -> None:
-    )
+    config = BrokerConfig(store_root=tmp_path / "store", connections=[])
+    server = build_workflow_server_from_config(config)
+    app = create_rpc_app(server)
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test",
+    ) as client:
+        await _rpc(
+            client,
+            "workflow.admin.source_registry.add",
+            {
+                "entry": {
+                    "kind": "mcp",
+                    "id": "dynamic.default",
+                    "enabled": True,
+                    "provider": "dynamic",
+                    "account": "default",
+                    "transport": {
+                        "kind": "stdio",
+                        "command": "dynamic-server",
+                        "args": [],
+                        "env": {},
+                    },
+                }
+            },
+        )
+        before = await _rpc(
+            client,
+            "workflow.sources.list",
+            {"limit": 50},
+        )
+        applied = await _rpc(
+            client,
+            "workflow.admin.source_registry.apply",
+            {},
+        )
+        after = await _rpc(
+            client,
+            "workflow.sources.list",
+            {"limit": 50},
+        )
+
+    before_ids = {source["id"] for source in before["result"]["sources"]}
+    after_ids = {source["id"] for source in after["result"]["sources"]}
+    assert "dynamic.default" not in before_ids
+    assert applied["result"]["registered"] == ["dynamic.default"]
+    assert "dynamic.default" in after_ids
 
 
 async def test_mcp_backed_rpc_can_be_built_from_neutral_workflow_config(

tests/wf_transport_rpc_http/test_source_registry_rpc.py

@@ -8,6 +8,7 @@ import httpx
 from wf_api import WorkflowSourceRegistryApi
 from wf_server import build_local_static_workflow_server
 from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
+from wf_transport_rpc_http.client_source_registry import RpcSourceRegistryClientMixin
 
 
 @dataclass(frozen=True, slots=True)
@@ -472,3 +473,67 @@ async def test_rpc_client_source_registry_mutation_methods_exist() -> None:
         "workflow.admin.source_registry.remove",
         {"source_id": "s"},
     )
+
+
+# --- apply tests ---
+
+
+async def test_rpc_source_registry_apply_unavailable_on_local_static(tmp_path) -> None:
+    app = create_rpc_app(build_local_static_workflow_server(tmp_path / "store"))
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test",
+    ) as client:
+        payload = await _rpc(
+            client,
+            "workflow.admin.source_registry.apply",
+            {},
+        )
+
+    assert payload["error"]["data"]["code"] == "source_registry_unavailable"
+
+
+async def test_rpc_source_registry_apply_returns_summary(tmp_path) -> None:
+    from unittest.mock import AsyncMock
+
+    admin = AsyncMock()
+    admin.apply_registry_changes.return_value = {
+        "applied": True,
+        "registered": ["demo.new"],
+        "updated": [],
+        "removed": [],
+        "connection_count": 1,
+        "registry_entry_count": 1,
+    }
+    server = replace(
+        build_local_static_workflow_server(tmp_path / "store"),
+        source_registry_admin=admin,
+    )
+    app = create_rpc_app(server)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://test",
+    ) as client:
+        payload = await _rpc(
+            client,
+            "workflow.admin.source_registry.apply",
+            {},
+        )
+
+    assert payload["result"]["applied"] is True
+    assert payload["result"]["registered"] == ["demo.new"]
+    admin.apply_registry_changes.assert_awaited_once()
+
+
+async def test_rpc_client_source_registry_apply_method_exists() -> None:
+    calls: list[tuple[str, dict[str, Any]]] = []
+
+    class Client(RpcSourceRegistryClientMixin):
+        async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
+            calls.append((method, params))
+            return {"applied": True}
+
+    payload = await Client().apply_registry_changes()
+
+    assert payload["applied"] is True
+    assert calls == [("workflow.admin.source_registry.apply", {})]