Source catalog! wfmcpservice thinning

docs/current_roadmap.md

@@ -118,6 +118,9 @@ implementation state.
   - Double-delegation has been removed: CLI and MCP workflow tools construct
     `WorkflowApi(context_from_service(service))` directly. `WorkflowSurfaceHandlers`
     remains only as a temporary compatibility shim for older imports.
+  - `WfMcpService` is being reduced into injected implementation services. Source
+    registry and catalog projection now live in `SourceCatalogService`; the old
+    service methods remain as compatibility delegates for MCP broker callers.
 
 Frame stress points remaining for native subgraphs and future fork/gather:
 

docs/superpowers/research/2026-06-01-wf-api-extraction-map.md

@@ -112,6 +112,10 @@ The extraction seam is clean: `WorkflowSurfaceHandlers` touches ~6 distinct capa
 | `adapters` | `dict[str, BackendAdapter]` | `self.service.adapters` | 1 | **MCP-specific** (live source check) |
 | `load_auth(connection_id)` | `AuthRecord \| None` | `self.service.load_auth(...)` | 1 | **MCP-specific** (live source check) |
 
+Source/catalog ownership is now split: `WfMcpService` coordinates broker runtime
+state, while `SourceCatalogService` owns capability source maps, planner catalog
+projection, snapshot hydration, and local docs lookup.
+
 ### WfMcpService Members NOT Used by WorkflowSurfaceHandlers
 
 These members of `WfMcpService` (`src/wf_mcp/broker/service/core.py`) are NOT accessed by `WorkflowSurfaceHandlers`:

src/wf_mcp/broker/service/core.py

@@ -4,8 +4,6 @@ import time
 from dataclasses import dataclass, field
 from typing import Any
 
-from pydantic import BaseModel
-
 from wf_artifacts import (
     DraftWorkspaceStore,
     RunStore,
@@ -15,7 +13,7 @@ from wf_artifacts import (
     WorkflowDeployment,
     artifact_catalog_entry,
 )
-from wf_authoring import NodeReturn, NodeSpec
+from wf_authoring import NodeSpec
 from wf_core import (
     NodeUse,
     RunState,
@@ -26,15 +24,9 @@ from wf_core import (
 from wf_api.models import RawWorkflowPlan
 from wf_api.runtime_dependencies import resolve_runtime_dependencies
 from wf_platform import (
-    CapabilityBuckets,
     CapabilitySource,
-    DocumentationPrompt,
-    DocumentationResource,
-    SourcePermissions,
-    SourceVisibility,
-    page_items,
 )
-from ...connections import ConnectionRegistry, parse_connection_id, qualify_node_name
+from ...connections import ConnectionRegistry, parse_connection_id
 from ...events import EventBus, McpEvent, make_event
 from ...models import (
     AuthRecord,
@@ -50,7 +42,6 @@ from ...runtime import ToolExecutor
 from ...shared.errors import error_payload
 from ...shared.names import RESERVED_CONNECTION_IDS
 from ...storage import Store
-from ...workflow.wrappers import _model_from_schema
 from wf_api.saved_subgraphs import (
     SavedSubgraphTree,
     prepare_saved_subgraphs,
@@ -61,7 +52,7 @@ from ..catalog import CombinedCatalog, snapshot_from_specs
 from ..discovery import discover_connection_capabilities, specs_from_discovered_tools
 from .adapters import require_adapter
 from .builtins import builtin_sources
-from .specs import get_qualified_spec, qualify_spec
+from .source_catalog import SourceCatalogService
 
 
 @dataclass(slots=True)
@@ -70,13 +61,13 @@ class WfMcpService:
     default_catalog_max_age_seconds: int = 300
     connections: ConnectionRegistry = field(default_factory=ConnectionRegistry)
     adapters: dict[str, BackendAdapter] = field(default_factory=dict)
-    capability_sources: dict[str, CapabilitySource] = field(default_factory=dict)
     event_bus: EventBus = field(default_factory=EventBus)
     include_builtin_specs: bool = True
     artifact_store: WorkflowArtifactStore | None = None
     draft_workspace_store: DraftWorkspaceStore | None = None
     run_store: RunStore | None = None
     tool_executor: ToolExecutor | None = None
+    source_catalog: SourceCatalogService = field(init=False)
 
     def __post_init__(self) -> None:
         """Install broker-local system specs when enabled.
@@ -85,17 +76,36 @@ class WfMcpService:
         must not guess workflow persistence from the MCP catalog/auth store because
         CLI, MCP, and future HTTP frontends may share or swap those stores.
         """
+        self.source_catalog = SourceCatalogService(
+            store=self.store,
+            connection_lookup=self.connections.get,
+            connection_list_enabled=self.connections.list_enabled,
+            connection_list_all=self.connections.list_all,
+            tool_executor_for=self._tool_executor_for,
+            load_auth=self.load_auth,
+            emit_event=self._record_event,
+            default_catalog_max_age_seconds=self.default_catalog_max_age_seconds,
+        )
         if self.include_builtin_specs:
             for source in builtin_sources().values():
                 self.register_capability_source(source)
         self.register_capability_source(admin_source())
 
+    @property
+    def capability_sources(self) -> dict[str, CapabilitySource]:
+        """Compatibility view of source catalog state.
+
+        Source ownership is moving into SourceCatalogService. Keep this property
+        because workflow APIs and existing tests still consume the service facade.
+        """
+        return self.source_catalog.capability_sources
+
     def register_connection(self, connection: ConnectionConfig) -> None:
         parse_connection_id(connection.id)
         if connection.id in RESERVED_CONNECTION_IDS:
             raise ValueError(f"connection id {connection.id!r} is reserved by wf-mcp")
         self.connections.register(connection)
-        self._hydrate_connection_source_from_snapshot(connection)
+        self.source_catalog.hydrate_connection_source_from_snapshot(connection)
         self._record_event(
             make_event(
                 "connection_registered",
@@ -127,7 +137,7 @@ class WfMcpService:
             self.connections.register(connection)
             source = self.capability_sources.get(connection.id)
             if source is None:
-                self._hydrate_connection_source_from_snapshot(connection)
+                self.source_catalog.hydrate_connection_source_from_snapshot(connection)
             else:
                 source.enabled = connection.enabled
 
@@ -165,113 +175,28 @@ class WfMcpService:
         max_age_seconds: int | None = None,
         emit_change_events: bool = True,
     ) -> None:
-        self.connections.get(connection_id)
+        self.source_catalog.register_specs(
-        qualified_specs = {
-            qualify_node_name(connection_id, spec.name): qualify_spec(
-                connection_id, spec
-            )
-            for spec in specs
-        }
-        existing_source = self.capability_sources.get(connection_id)
-        if existing_source is not None:
-            # Catalog refreshes replace discovered specs, not operator policy.
-            existing_source.capabilities.node_specs = qualified_specs
-        else:
-            self.register_capability_source(
-                CapabilitySource(
-                    id=connection_id,
-                    kind="connection",
-                    capabilities=CapabilityBuckets(node_specs=qualified_specs),
-                    enabled=self.connections.get(connection_id).enabled,
-                    visibility=SourceVisibility(
-                        planner=True,
-                        mcp_client=True,
-                        admin_dashboard=True,
-                    ),
-                    permissions=SourcePermissions(calls_upstream=True),
-                    description=(
-                        f"Specs discovered or registered for {connection_id}."
-                    ),
-                )
-            )
-        snapshot = snapshot_from_specs(
             connection_id,
-            specs=qualified_specs,
+            *specs,
-            fetched_at_epoch_ms=int(time.time() * 1000),
+            max_age_seconds=max_age_seconds,
-            max_age_seconds=max_age_seconds or self.default_catalog_max_age_seconds,
+            emit_change_events=emit_change_events,
+            record_catalog_change_events=lambda source_id, snapshot, reason: (
+                self._record_catalog_change_events(
+                    source_id,
+                    snapshot,
+                    reason=reason,
+                )
+            ),
         )
-        self.store.save_catalog(snapshot)
-        self._record_event(
-            make_event(
-                "specs_registered",
-                connection_id=connection_id,
-                payload={"node_count": len(qualified_specs)},
-            )
-        )
-        if emit_change_events:
-            self._record_catalog_change_events(
-                connection_id,
-                snapshot,
-                reason="specs_registered",
-            )
 
     def get_catalog(self) -> CombinedCatalog:
-        snapshots: dict[str, CatalogSnapshot] = {}
+        return self.source_catalog.get_catalog()
-        for connection in self.connections.list_enabled():
-            snapshot = self.store.load_catalog(connection.id)
-            if snapshot is not None:
-                snapshots[connection.id] = snapshot
-        return CombinedCatalog(snapshots=snapshots)
 
     def get_planner_catalog(self) -> CombinedCatalog:
-        """Return all planner-visible specs, including broker-local sources."""
+        return self.source_catalog.get_planner_catalog()
-        snapshots: dict[str, CatalogSnapshot] = {}
-        fetched_at_epoch_ms = int(time.time() * 1000)
-        for source in self.capability_sources.values():
-            if not source.enabled or not source.visibility.planner:
-                continue
-            stored_snapshot = self.store.load_catalog(source.id)
-            snapshots[source.id] = snapshot_from_specs(
-                source.id,
-                specs=source.capabilities.node_specs,
-                tool_display_names={
-                    entry.local_name: entry.title for entry in stored_snapshot.nodes
-                }
-                if stored_snapshot is not None
-                else None,
-                metadata={
-                    "kind": source.kind,
-                    "description": source.description,
-                }
-                if stored_snapshot is None
-                else stored_snapshot.metadata,
-                fetched_at_epoch_ms=(
-                    stored_snapshot.fetched_at_epoch_ms
-                    if stored_snapshot is not None
-                    else fetched_at_epoch_ms
-                ),
-                max_age_seconds=(
-                    stored_snapshot.max_age_seconds
-                    if stored_snapshot is not None
-                    else self.default_catalog_max_age_seconds
-                ),
-            )
-            if stored_snapshot is not None:
-                # Connection resources/prompts are discovered by the backend catalog,
-                # while planner node visibility is governed by capability sources.
-                snapshots[source.id].resources = list(stored_snapshot.resources)
-                snapshots[source.id].prompts = list(stored_snapshot.prompts)
-        return CombinedCatalog(snapshots=snapshots)
 
     def list_sources(self) -> list[dict[str, Any]]:
-        """Return every capability source with the names it currently owns."""
+        return self.source_catalog.list_sources()
-        return [
-            source.as_inventory().model_dump(mode="json")
-            for source in sorted(
-                self.capability_sources.values(),
-                key=lambda source: source.id,
-            )
-        ]
 
     def list_source_summaries(
         self,
@@ -279,32 +204,13 @@ class WfMcpService:
         cursor: str | None = None,
         limit: int = 50,
     ) -> dict[str, Any]:
-        """Return compact paged source summaries for progressive discovery."""
+        return self.source_catalog.list_source_summaries(cursor=cursor, limit=limit)
-        summaries = [
-            source.as_status().model_dump(mode="json")
-            for source in sorted(
-                self.capability_sources.values(),
-                key=lambda source: source.id,
-            )
-        ]
-        page = page_items(summaries, cursor=cursor, limit=limit)
-        return {
-            "sources": list(page.items),
-            "next_cursor": page.next_cursor,
-            "total": page.total,
-        }
 
     def inspect_source(self, source_id: str) -> dict[str, Any]:
-        """Return the full source inventory for one exact source id."""
+        return self.source_catalog.inspect_source(source_id)
-        try:
-            source = self.capability_sources[source_id]
-        except KeyError as exc:
-            raise KeyError(f"unknown source {source_id!r}") from exc
-        return source.as_inventory().model_dump(mode="json")
 
     def list_available_specs(self) -> list[CatalogNodeEntry]:
-        """Return planner-visible node catalog entries from every visible source."""
+        return self.source_catalog.list_available_specs()
-        return self.get_planner_catalog().entries()
 
     def workflow_artifact_catalog_entry(
         self,
@@ -314,73 +220,35 @@ class WfMcpService:
         return artifact_catalog_entry(artifact)
 
     def get_connection_snapshot(self, connection_id: str) -> CatalogSnapshot | None:
-        self.connections.get(connection_id)
+        return self.source_catalog.get_connection_snapshot(connection_id)
-        return self.store.load_catalog(connection_id)
 
     def connection_statuses(self) -> list[dict[str, Any]]:
-        statuses: list[dict[str, Any]] = []
+        return self.source_catalog.connection_statuses()
-        for connection in self.connections.list_all():
-            snapshot = self.store.load_catalog(connection.id)
-            statuses.append(
-                {
-                    "connection_id": connection.id,
-                    "server": connection.server,
-                    "account": connection.account,
-                    "enabled": connection.enabled,
-                    "has_snapshot": snapshot is not None,
-                    "fetched_at_epoch_ms": None
-                    if snapshot is None
-                    else snapshot.fetched_at_epoch_ms,
-                    "max_age_seconds": None
-                    if snapshot is None
-                    else snapshot.max_age_seconds,
-                    "node_count": 0 if snapshot is None else len(snapshot.nodes),
-                    "resource_count": 0
-                    if snapshot is None
-                    else len(snapshot.resources),
-                    "prompt_count": 0 if snapshot is None else len(snapshot.prompts),
-                }
-            )
-        return statuses
 
     def list_resources(
         self,
         *,
         connection_id: str | None = None,
     ) -> list[CatalogResourceEntry]:
-        if connection_id is None:
+        return self.source_catalog.list_resources(connection_id=connection_id)
-            return self.get_catalog().resource_entries()
-        snapshot = self.get_connection_snapshot(connection_id)
-        if snapshot is None:
-            return []
-        return sorted(snapshot.resources, key=lambda entry: entry.qualified_name)
 
     def list_prompts(
         self,
         *,
         connection_id: str | None = None,
     ) -> list[CatalogPromptEntry]:
-        if connection_id is None:
+        return self.source_catalog.list_prompts(connection_id=connection_id)
-            return self.get_catalog().prompt_entries()
-        snapshot = self.get_connection_snapshot(connection_id)
-        if snapshot is None:
-            return []
-        return sorted(snapshot.prompts, key=lambda entry: entry.qualified_name)
 
     def get_resource(self, qualified_name: str) -> CatalogResourceEntry:
-        entry = self.get_catalog().find_resource(qualified_name)
+        return self.source_catalog.get_resource(qualified_name)
-        if entry is None:
-            raise KeyError(f"unknown resource {qualified_name!r}")
-        return entry
 
     def get_prompt(self, qualified_name: str) -> CatalogPromptEntry:
-        entry = self.get_catalog().find_prompt(qualified_name)
+        return self.source_catalog.get_prompt(qualified_name)
-        if entry is None:
-            raise KeyError(f"unknown prompt {qualified_name!r}")
-        return entry
 
     async def read_resource(self, qualified_name: str) -> dict[str, Any]:
-        local_resource = self._local_documentation_resource(qualified_name)
+        local_resource = self.source_catalog.local_documentation_resource(
+            qualified_name
+        )
         if local_resource is not None:
             self._record_event(
                 make_event(
@@ -485,7 +353,7 @@ class WfMcpService:
         *,
         arguments: dict[str, str] | None = None,
     ) -> dict[str, Any]:
-        local_prompt = self._local_documentation_prompt(qualified_name)
+        local_prompt = self.source_catalog.local_documentation_prompt(qualified_name)
         if local_prompt is not None:
             self._record_event(
                 make_event(
@@ -538,28 +406,6 @@ class WfMcpService:
         )
         return result
 
-    def _local_documentation_resource(
-        self,
-        qualified_name: str,
-    ) -> DocumentationResource | None:
-        """Return a local docs resource from capability sources by qualified name."""
-        for source in self.capability_sources.values():
-            resource = source.capabilities.resources.get(qualified_name)
-            if isinstance(resource, DocumentationResource):
-                return resource
-        return None
-
-    def _local_documentation_prompt(
-        self,
-        qualified_name: str,
-    ) -> DocumentationPrompt | None:
-        """Return a local docs prompt from capability sources by qualified name."""
-        for source in self.capability_sources.values():
-            prompt = source.capabilities.prompts.get(qualified_name)
-            if isinstance(prompt, DocumentationPrompt):
-                return prompt
-        return None
-
     async def refresh_connection_catalog(
         self,
         connection_id: str,
@@ -814,85 +660,8 @@ class WfMcpService:
         """Register a capability source as canonical service state."""
         self.capability_sources[source.id] = source
 
-    def _hydrate_connection_source_from_snapshot(
-        self,
-        connection: ConnectionConfig,
-    ) -> None:
-        """Register one connection source, hydrating specs from snapshot if present."""
-        if connection.id in self.capability_sources:
-            return
-
-        snapshot = self.store.load_catalog(connection.id)
-        specs = {
-            entry.qualified_name: self._spec_from_snapshot_entry(entry)
-            for entry in (() if snapshot is None else snapshot.nodes)
-        }
-        description = (
-            f"Specs restored from catalog for {connection.id}."
-            if specs
-            else f"No catalog loaded for {connection.id}."
-        )
-        self.register_capability_source(
-            CapabilitySource(
-                id=connection.id,
-                kind="connection",
-                enabled=connection.enabled,
-                capabilities=CapabilityBuckets(node_specs=specs),
-                visibility=SourceVisibility(
-                    planner=True,
-                    mcp_client=True,
-                    admin_dashboard=True,
-                ),
-                permissions=SourcePermissions(calls_upstream=True),
-                description=description,
-            )
-        )
-
-    def _spec_from_snapshot_entry(
-        self,
-        entry: CatalogNodeEntry,
-    ) -> NodeSpec[Any, Any]:
-        """Rebuild an executable tool wrapper from a stored catalog node entry.
-
-        Snapshot entries store schema/name metadata, not Python functions. This
-        helper reconstructs the same generated NodeSpec shape and routes calls
-        through `_tool_executor_for()`, so hydrated specs use the persistent MCP
-        runtime when the service has one configured.
-        """
-        model_prefix = entry.qualified_name.replace(".", "_").replace("-", "_")
-        input_model = _model_from_schema(f"{model_prefix}_Input", entry.input_schema)
-        output_schema = entry.output_schema
-        output_model = _model_from_schema(f"{model_prefix}_Output", output_schema)
-
-        async def invoke_tool(payload: BaseModel) -> NodeReturn[BaseModel]:
-            connection = self.connections.get(entry.connection_id)
-            auth = self.load_auth(entry.connection_id)
-            result = await self._tool_executor_for(connection).call_tool(
-                connection,
-                auth,
-                entry.local_name,
-                payload.model_dump(exclude_unset=True),
-            )
-            return NodeReturn(
-                outcome=result.outcome,
-                output=output_model.model_validate(result.output),
-            )
-
-        return NodeSpec(
-            name=entry.qualified_name,
-            input_model=input_model,
-            output_model=output_model,
-            outcomes=entry.outcomes,
-            fn=invoke_tool,
-            description=entry.description,
-            is_async=True,
-            accepts_context=False,
-            input_schema_contract=entry.input_schema,
-            output_schema_contract=output_schema,
-        )
-
     def _get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
-        return get_qualified_spec(self.capability_sources, qualified_name)
+        return self.source_catalog.get_qualified_spec(qualified_name)
 
     def _record_event(self, event: McpEvent) -> None:
         self.event_bus.publish(event)

src/wf_mcp/broker/service/source_catalog.py

@@ -0,0 +1,379 @@
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+from pydantic import BaseModel
+from wf_authoring import NodeReturn, NodeSpec
+from wf_platform import (
+    CapabilityBuckets,
+    CapabilitySource,
+    DocumentationPrompt,
+    DocumentationResource,
+    SourcePermissions,
+    SourceVisibility,
+    page_items,
+)
+
+from ...connections import ConnectionConfig, qualify_node_name
+from ...events import McpEvent, make_event
+from ...models import (
+    AuthRecord,
+    CatalogNodeEntry,
+    CatalogPromptEntry,
+    CatalogResourceEntry,
+    CatalogSnapshot,
+)
+from ...runtime import ToolExecutor
+from ...storage import Store
+from ...workflow.wrappers import _model_from_schema
+from ..catalog import CombinedCatalog, snapshot_from_specs
+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]
+EventEmitter = Callable[[McpEvent], None]
+
+
+@dataclass(slots=True)
+class SourceCatalogService:
+    """Own service-local capability sources and catalog projections.
+
+    This is deliberately still MCP-broker-internal. It knows about stored MCP
+    catalog snapshots because hydrated workflow NodeSpecs must call back through
+    the broker's configured tool executor.
+    """
+
+    store: Store
+    connection_lookup: ConnectionLookup
+    connection_list_enabled: ConnectionList
+    connection_list_all: ConnectionList
+    tool_executor_for: ToolExecutorLookup
+    load_auth: AuthLoader
+    emit_event: EventEmitter
+    default_catalog_max_age_seconds: int = 300
+    capability_sources: dict[str, CapabilitySource] = field(default_factory=dict)
+
+    def register_capability_source(self, source: CapabilitySource) -> None:
+        """Register one source as canonical planner/runtime source state."""
+        self.capability_sources[source.id] = source
+
+    def get_catalog(self) -> CombinedCatalog:
+        snapshots: dict[str, CatalogSnapshot] = {}
+        for connection in self.connection_list_enabled():
+            snapshot = self.store.load_catalog(connection.id)
+            if snapshot is not None:
+                snapshots[connection.id] = snapshot
+        return CombinedCatalog(snapshots=snapshots)
+
+    def get_planner_catalog(self) -> CombinedCatalog:
+        """Return all planner-visible specs, including broker-local sources."""
+        snapshots: dict[str, CatalogSnapshot] = {}
+        fetched_at_epoch_ms = int(time.time() * 1000)
+        for source in self.capability_sources.values():
+            if not source.enabled or not source.visibility.planner:
+                continue
+            stored_snapshot = self.store.load_catalog(source.id)
+            snapshots[source.id] = snapshot_from_specs(
+                source.id,
+                specs=source.capabilities.node_specs,
+                tool_display_names={
+                    entry.local_name: entry.title for entry in stored_snapshot.nodes
+                }
+                if stored_snapshot is not None
+                else None,
+                metadata={
+                    "kind": source.kind,
+                    "description": source.description,
+                }
+                if stored_snapshot is None
+                else stored_snapshot.metadata,
+                fetched_at_epoch_ms=(
+                    stored_snapshot.fetched_at_epoch_ms
+                    if stored_snapshot is not None
+                    else fetched_at_epoch_ms
+                ),
+                max_age_seconds=(
+                    stored_snapshot.max_age_seconds
+                    if stored_snapshot is not None
+                    else self.default_catalog_max_age_seconds
+                ),
+            )
+            if stored_snapshot is not None:
+                snapshots[source.id].resources = list(stored_snapshot.resources)
+                snapshots[source.id].prompts = list(stored_snapshot.prompts)
+        return CombinedCatalog(snapshots=snapshots)
+
+    def list_sources(self) -> list[dict[str, Any]]:
+        """Return every capability source with the names it currently owns."""
+        return [
+            source.as_inventory().model_dump(mode="json")
+            for source in sorted(
+                self.capability_sources.values(),
+                key=lambda source: source.id,
+            )
+        ]
+
+    def list_source_summaries(
+        self,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        """Return compact paged source summaries for progressive discovery."""
+        summaries = [
+            source.as_status().model_dump(mode="json")
+            for source in sorted(
+                self.capability_sources.values(),
+                key=lambda source: source.id,
+            )
+        ]
+        page = page_items(summaries, cursor=cursor, limit=limit)
+        return {
+            "sources": list(page.items),
+            "next_cursor": page.next_cursor,
+            "total": page.total,
+        }
+
+    def inspect_source(self, source_id: str) -> dict[str, Any]:
+        """Return the full source inventory for one exact source id."""
+        try:
+            source = self.capability_sources[source_id]
+        except KeyError as exc:
+            raise KeyError(f"unknown source {source_id!r}") from exc
+        return source.as_inventory().model_dump(mode="json")
+
+    def list_available_specs(self) -> list[CatalogNodeEntry]:
+        """Return planner-visible node catalog entries from every visible source."""
+        return self.get_planner_catalog().entries()
+
+    def get_connection_snapshot(self, connection_id: str) -> CatalogSnapshot | None:
+        self.connection_lookup(connection_id)
+        return self.store.load_catalog(connection_id)
+
+    def connection_statuses(self) -> list[dict[str, Any]]:
+        statuses: list[dict[str, Any]] = []
+        for connection in self.connection_list_all():
+            snapshot = self.store.load_catalog(connection.id)
+            statuses.append(
+                {
+                    "connection_id": connection.id,
+                    "server": connection.server,
+                    "account": connection.account,
+                    "enabled": connection.enabled,
+                    "has_snapshot": snapshot is not None,
+                    "fetched_at_epoch_ms": None
+                    if snapshot is None
+                    else snapshot.fetched_at_epoch_ms,
+                    "max_age_seconds": None
+                    if snapshot is None
+                    else snapshot.max_age_seconds,
+                    "node_count": 0 if snapshot is None else len(snapshot.nodes),
+                    "resource_count": 0
+                    if snapshot is None
+                    else len(snapshot.resources),
+                    "prompt_count": 0 if snapshot is None else len(snapshot.prompts),
+                }
+            )
+        return statuses
+
+    def list_resources(
+        self,
+        *,
+        connection_id: str | None = None,
+    ) -> list[CatalogResourceEntry]:
+        if connection_id is None:
+            return self.get_catalog().resource_entries()
+        snapshot = self.get_connection_snapshot(connection_id)
+        if snapshot is None:
+            return []
+        return sorted(snapshot.resources, key=lambda entry: entry.qualified_name)
+
+    def list_prompts(
+        self,
+        *,
+        connection_id: str | None = None,
+    ) -> list[CatalogPromptEntry]:
+        if connection_id is None:
+            return self.get_catalog().prompt_entries()
+        snapshot = self.get_connection_snapshot(connection_id)
+        if snapshot is None:
+            return []
+        return sorted(snapshot.prompts, key=lambda entry: entry.qualified_name)
+
+    def get_resource(self, qualified_name: str) -> CatalogResourceEntry:
+        entry = self.get_catalog().find_resource(qualified_name)
+        if entry is None:
+            raise KeyError(f"unknown resource {qualified_name!r}")
+        return entry
+
+    def get_prompt(self, qualified_name: str) -> CatalogPromptEntry:
+        entry = self.get_catalog().find_prompt(qualified_name)
+        if entry is None:
+            raise KeyError(f"unknown prompt {qualified_name!r}")
+        return entry
+
+    def hydrate_connection_source_from_snapshot(
+        self,
+        connection: ConnectionConfig,
+    ) -> None:
+        """Register one connection source, hydrating specs from snapshot if present."""
+        if connection.id in self.capability_sources:
+            return
+
+        snapshot = self.store.load_catalog(connection.id)
+        specs = {
+            entry.qualified_name: self.spec_from_snapshot_entry(entry)
+            for entry in (() if snapshot is None else snapshot.nodes)
+        }
+        description = (
+            f"Specs restored from catalog for {connection.id}."
+            if specs
+            else f"No catalog loaded for {connection.id}."
+        )
+        self.register_capability_source(
+            CapabilitySource(
+                id=connection.id,
+                kind="connection",
+                enabled=connection.enabled,
+                capabilities=CapabilityBuckets(node_specs=specs),
+                visibility=SourceVisibility(
+                    planner=True,
+                    mcp_client=True,
+                    admin_dashboard=True,
+                ),
+                permissions=SourcePermissions(calls_upstream=True),
+                description=description,
+            )
+        )
+
+    def spec_from_snapshot_entry(
+        self,
+        entry: CatalogNodeEntry,
+    ) -> NodeSpec[Any, Any]:
+        """Rebuild an executable tool wrapper from a stored catalog node entry.
+
+        Snapshot entries store schema/name metadata, not Python functions. This
+        helper reconstructs the same generated NodeSpec shape and routes calls
+        through `tool_executor_for()`, so hydrated specs use the persistent MCP
+        runtime when the service has one configured.
+        """
+        model_prefix = entry.qualified_name.replace(".", "_").replace("-", "_")
+        input_model = _model_from_schema(f"{model_prefix}_Input", entry.input_schema)
+        output_schema = entry.output_schema
+        output_model = _model_from_schema(f"{model_prefix}_Output", output_schema)
+
+        async def invoke_tool(payload: BaseModel) -> NodeReturn[BaseModel]:
+            connection = self.connection_lookup(entry.connection_id)
+            auth = self.load_auth(entry.connection_id)
+            result = await self.tool_executor_for(connection).call_tool(
+                connection,
+                auth,
+                entry.local_name,
+                payload.model_dump(exclude_unset=True),
+            )
+            return NodeReturn(
+                outcome=result.outcome,
+                output=output_model.model_validate(result.output),
+            )
+
+        return NodeSpec(
+            name=entry.qualified_name,
+            input_model=input_model,
+            output_model=output_model,
+            outcomes=entry.outcomes,
+            fn=invoke_tool,
+            description=entry.description,
+            is_async=True,
+            accepts_context=False,
+            input_schema_contract=entry.input_schema,
+            output_schema_contract=output_schema,
+        )
+
+    def get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
+        return get_qualified_spec(self.capability_sources, qualified_name)
+
+    def register_specs(
+        self,
+        connection_id: str,
+        *specs: NodeSpec[Any, Any],
+        max_age_seconds: int | None = None,
+        emit_change_events: bool = True,
+        record_catalog_change_events: Callable[
+            [str, CatalogSnapshot, str],
+            None,
+        ]
+        | None = None,
+    ) -> CatalogSnapshot:
+        self.connection_lookup(connection_id)
+        qualified_specs = {
+            qualify_node_name(connection_id, spec.name): qualify_spec(
+                connection_id, spec
+            )
+            for spec in specs
+        }
+        existing_source = self.capability_sources.get(connection_id)
+        if existing_source is not None:
+            existing_source.capabilities.node_specs = qualified_specs
+        else:
+            self.register_capability_source(
+                CapabilitySource(
+                    id=connection_id,
+                    kind="connection",
+                    capabilities=CapabilityBuckets(node_specs=qualified_specs),
+                    enabled=self.connection_lookup(connection_id).enabled,
+                    visibility=SourceVisibility(
+                        planner=True,
+                        mcp_client=True,
+                        admin_dashboard=True,
+                    ),
+                    permissions=SourcePermissions(calls_upstream=True),
+                    description=(
+                        f"Specs discovered or registered for {connection_id}."
+                    ),
+                )
+            )
+        snapshot = snapshot_from_specs(
+            connection_id,
+            specs=qualified_specs,
+            fetched_at_epoch_ms=int(time.time() * 1000),
+            max_age_seconds=max_age_seconds or self.default_catalog_max_age_seconds,
+        )
+        self.store.save_catalog(snapshot)
+        self.emit_event(
+            make_event(
+                "specs_registered",
+                connection_id=connection_id,
+                payload={"node_count": len(qualified_specs)},
+            )
+        )
+        if emit_change_events and record_catalog_change_events is not None:
+            record_catalog_change_events(connection_id, snapshot, "specs_registered")
+        return snapshot
+
+    def local_documentation_resource(
+        self,
+        qualified_name: str,
+    ) -> DocumentationResource | None:
+        """Return a local docs resource from capability sources by qualified name."""
+        for source in self.capability_sources.values():
+            resource = source.capabilities.resources.get(qualified_name)
+            if isinstance(resource, DocumentationResource):
+                return resource
+        return None
+
+    def local_documentation_prompt(
+        self,
+        qualified_name: str,
+    ) -> DocumentationPrompt | None:
+        """Return a local docs prompt from capability sources by qualified name."""
+        for source in self.capability_sources.values():
+            prompt = source.capabilities.prompts.get(qualified_name)
+            if isinstance(prompt, DocumentationPrompt):
+                return prompt
+        return None

src/wf_mcp/broker/service/workflow_live_checks.py

@@ -70,7 +70,7 @@ async def live_source_diagnostics(
     """
     diagnostics: list[DependencyDiagnostic] = []
     for source_id, logical_ref in _required_live_sources(deployment, artifacts).items():
-        source = service.capability_sources.get(source_id)
+        source = service.source_catalog.capability_sources.get(source_id)
         if (
             source is None
             or not source.enabled

src/wf_mcp/broker/service/workflow_operation_context.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass
 from typing import Any
 
 from wf_artifacts import DependencyDiagnostic, WorkflowArtifact, WorkflowDeployment
+from wf_authoring import NodeSpec
 from wf_api.operation_context import (
     WorkflowArtifactCataloger,
     WorkflowEventRecorder,
@@ -48,10 +49,10 @@ class WfMcpWorkflowSpecProvider(WorkflowSpecProvider):
 
     @property
     def capability_sources(self):
-        return self.service.capability_sources
+        return self.service.source_catalog.capability_sources
 
-    def get_qualified_spec(self, qualified_name: str) -> object:
+    def get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
-        return self.service._get_qualified_spec(qualified_name)  # noqa: SLF001
+        return self.service.source_catalog.get_qualified_spec(qualified_name)
 
 
 @dataclass(frozen=True, slots=True)

tests/wf_api/test_operation_context.py

@@ -2,11 +2,25 @@ from __future__ import annotations
 
 import ast
 import json
+import tempfile
 from pathlib import Path
 
 from wf_api.operation_context import WorkflowOperationContext
 from wf_cli.context import load_cli_context
+from wf_mcp.broker import WfMcpService
 from wf_mcp.broker.service.workflow_operation_context import context_from_service
+from wf_mcp.storage import FileStore
+
+
+def _local_temp_root() -> Path:
+    return Path(tempfile.mkdtemp())
+
+
+def test_context_uses_source_catalog_mapping() -> None:
+    service = WfMcpService(store=FileStore(_local_temp_root() / "context_sources"))
+    context = context_from_service(service)
+
+    assert context.capability_sources is service.source_catalog.capability_sources
 
 
 def test_wf_api_operation_context_imports_no_wf_mcp() -> None:

tests/wf_mcp/service/test_catalog.py

@@ -6,11 +6,13 @@ import shutil
 from wf_authoring import NodeSpec
 from wf_core import RunStatus
 from wf_mcp.broker import WfMcpService
+from wf_mcp.broker.service.source_catalog import SourceCatalogService
 from wf_mcp.models import ConnectionConfig
 from wf_mcp.storage import FileStore
 from wf_platform import (
     CapabilityBuckets,
     CapabilitySource,
+    DocumentationResource,
     SourceVisibility,
 )
 
@@ -296,3 +298,206 @@ def test_service_hydrates_planner_specs_from_stored_catalog() -> None:
     assert "demo.personal.echo_tool" in planner_names
     assert run.status == RunStatus.COMPLETED
     assert run.output["echoed"] == "hello"
+
+
+def test_source_catalog_service_registers_and_lists_sources_directly() -> None:
+    store = FileStore(local_temp_root() / "source_catalog_direct")
+
+    def unused_tool_executor(connection: ConnectionConfig):
+        raise AssertionError("tool executor should not be used by source listing")
+
+    catalog = SourceCatalogService(
+        store=store,
+        connection_lookup=lambda connection_id: ConnectionConfig(
+            id=connection_id,
+            server="demo",
+            account="personal",
+        ),
+        connection_list_enabled=lambda: [],
+        connection_list_all=lambda: [],
+        tool_executor_for=unused_tool_executor,
+        load_auth=lambda connection_id: None,
+        emit_event=lambda event: None,
+    )
+
+    catalog.register_capability_source(
+        CapabilitySource(
+            id="demo.personal",
+            kind="connection",
+            capabilities=CapabilityBuckets(),
+            visibility=SourceVisibility(planner=True),
+        )
+    )
+
+    payload = catalog.list_source_summaries(limit=10)
+
+    assert payload["total"] == 1
+    assert payload["sources"][0]["id"] == "demo.personal"
+
+
+def test_wfmcpservice_capability_sources_proxy_source_catalog() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "source_catalog_proxy"))
+
+    assert service.capability_sources is service.source_catalog.capability_sources
+    assert "wf.std" in service.source_catalog.capability_sources
+
+
+def test_source_catalog_service_excludes_hidden_sources_from_planner_catalog() -> None:
+    def unused_tool_executor(connection: ConnectionConfig):
+        raise AssertionError("tool executor should not be used by planner listing")
+
+    catalog = SourceCatalogService(
+        store=FileStore(local_temp_root() / "source_catalog_hidden"),
+        connection_lookup=lambda connection_id: ConnectionConfig(
+            id=connection_id,
+            server="demo",
+            account="personal",
+        ),
+        connection_list_enabled=lambda: [],
+        connection_list_all=lambda: [],
+        tool_executor_for=unused_tool_executor,
+        load_auth=lambda connection_id: None,
+        emit_event=lambda event: None,
+    )
+    visible_tool = NodeSpec(
+        name="visible.source.echo_tool",
+        input_model=echo_tool.input_model,
+        output_model=echo_tool.output_model,
+        outcomes=echo_tool.outcomes,
+        fn=echo_tool.fn,
+        description=echo_tool.description,
+        is_async=echo_tool.is_async,
+        accepts_context=echo_tool.accepts_context,
+        input_schema_contract=echo_tool.input_schema_contract,
+        output_schema_contract=echo_tool.output_schema_contract,
+    )
+    hidden_tool = NodeSpec(
+        name="hidden.source.echo_tool",
+        input_model=echo_tool.input_model,
+        output_model=echo_tool.output_model,
+        outcomes=echo_tool.outcomes,
+        fn=echo_tool.fn,
+        description=echo_tool.description,
+        is_async=echo_tool.is_async,
+        accepts_context=echo_tool.accepts_context,
+        input_schema_contract=echo_tool.input_schema_contract,
+        output_schema_contract=echo_tool.output_schema_contract,
+    )
+    catalog.register_capability_source(
+        CapabilitySource(
+            id="visible.source",
+            kind="system",
+            capabilities=CapabilityBuckets(
+                node_specs={"visible.source.echo_tool": visible_tool}
+            ),
+            visibility=SourceVisibility(planner=True),
+        )
+    )
+    catalog.register_capability_source(
+        CapabilitySource(
+            id="hidden.source",
+            kind="system",
+            capabilities=CapabilityBuckets(
+                node_specs={"hidden.source.echo_tool": hidden_tool}
+            ),
+            visibility=SourceVisibility(planner=False, admin_dashboard=False),
+        )
+    )
+
+    planner_names = {
+        entry.qualified_name for entry in catalog.get_planner_catalog().entries()
+    }
+
+    assert "visible.source.echo_tool" in planner_names
+    assert "hidden.source.echo_tool" not in planner_names
+
+
+def test_source_catalog_hydrates_connection_source_from_snapshot_directly() -> None:
+    root = local_temp_root() / "source_catalog_hydrate_direct"
+    shutil.rmtree(root, ignore_errors=True)
+    first_service = WfMcpService(store=FileStore(root))
+    first_service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    first_service.register_adapter("demo", FakeAdapter())
+    asyncio.run(first_service.refresh_connection_catalog("demo.personal"))
+
+    second_service = WfMcpService(store=FileStore(root))
+    second_service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+
+    specs = second_service.source_catalog.capability_sources[
+        "demo.personal"
+    ].capabilities.node_specs
+
+    assert "demo.personal.echo_tool" in specs
+
+
+def test_source_catalog_register_specs_replaces_discovered_specs_directly() -> None:
+    connection = ConnectionConfig(
+        id="demo.personal",
+        server="demo",
+        account="personal",
+    )
+
+    def unused_tool_executor(connection: ConnectionConfig):
+        raise AssertionError("tool executor should not be used by spec registration")
+
+    catalog = SourceCatalogService(
+        store=FileStore(local_temp_root() / "source_catalog_register_specs"),
+        connection_lookup=lambda connection_id: connection,
+        connection_list_enabled=lambda: [connection],
+        connection_list_all=lambda: [connection],
+        tool_executor_for=unused_tool_executor,
+        load_auth=lambda connection_id: None,
+        emit_event=lambda event: None,
+    )
+
+    catalog.register_capability_source(
+        CapabilitySource(
+            id="demo.personal",
+            kind="connection",
+            capabilities=CapabilityBuckets(
+                node_specs={"demo.personal.finalize_tool": finalize_tool}
+            ),
+            visibility=SourceVisibility(planner=True),
+        )
+    )
+    assert "demo.personal.finalize_tool" in (
+        catalog.capability_sources["demo.personal"].capabilities.node_specs
+    )
+
+    catalog.register_specs("demo.personal", echo_tool)
+
+    specs = catalog.capability_sources["demo.personal"].capabilities.node_specs
+
+    assert set(specs) == {"demo.personal.echo_tool"}
+    assert catalog.store.load_catalog("demo.personal") is not None
+
+
+def test_source_catalog_finds_local_documentation_resource_directly() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "source_local_docs"))
+    test_resource = DocumentationResource(
+        name="test.docs.example",
+        uri="wf://docs/example",
+        title="Example Doc",
+        description="Test documentation resource.",
+        mime_type="text/markdown",
+        text="# Example",
+    )
+    service.register_capability_source(
+        CapabilitySource(
+            id="test.docs",
+            kind="system",
+            capabilities=CapabilityBuckets(
+                resources={"test.docs.example": test_resource}
+            ),
+            visibility=SourceVisibility(planner=True),
+        )
+    )
+
+    result = service.source_catalog.local_documentation_resource("test.docs.example")
+
+    assert result is not None
+    assert result.uri == test_resource.uri