refactor: add static source provider seam

docs/current_roadmap.md

@@ -96,6 +96,9 @@ auth admin are implemented. The next work is polish, not new broad surfaces.
   `ops.py` source config, capability call, draft artifact creation, deployment,
   and workflow run. Runbook:
   [`Python source`](runbooks/python-source.md).
+- Completed: static source inventory providers now have an explicit
+  `WorkflowSourceProvider.load_sources()` seam in `wf_server`, and Python
+  source loading is behind `PythonSourceProvider`.
 - Completed: server startup policy moved to `wf_server.cli`; JSON-RPC HTTP
   remains in `wf_transport_rpc_http`:
   [`server CLI and transport boundary`](superpowers/specs/2026-06-10-server-cli-transport-boundary.md).

docs/source_architecture.md

@@ -65,6 +65,18 @@ wf_config.server.sources[]
   -> transport or CLI
 ```
 
+The first shared provider seam is intentionally static:
+
+```python
+class WorkflowSourceProvider(Protocol):
+    def load_sources(self) -> Mapping[str, CapabilitySource]: ...
+```
+
+This covers source families that can project configured inventory into
+workflow-facing `CapabilitySource` objects. Provider-specific runtime pools,
+admin/apply hooks, auth, catalog caches, and live health checks stay outside
+this narrow seam until a source family needs them.
+
 For MCP, the provider also owns stateful upstream sessions:
 
 ```text
@@ -78,6 +90,7 @@ For Python, the provider is simpler:
 
 ```text
 PythonSourceConfig(path, module, registry)
+  -> PythonSourceProvider
   -> import module
   -> load NodeSpec registry
   -> qualify specs under source id

docs/superpowers/specs/2026-06-11-python-source-provider.md

@@ -121,7 +121,8 @@ Implemented:
 - `wf_config` accepts `server.sources[]` entries with `kind: "python"`.
 - `wf_sources_python` loads trusted in-process `NodeSpec` registries from
   `path` plus `module:registry`.
-- `wf_server.config` composes Python sources into local/static servers.
+- `wf_sources_python.PythonSourceProvider` implements the static
+  `WorkflowSourceProvider.load_sources()` seam used by `wf_server.config`.
 - `wf config validate` imports configured trusted Python sources and reports
   missing modules, missing registries, invalid registry shapes, and duplicate
   specs before server startup.

src/wf_server/config.py

@@ -3,9 +3,9 @@ from __future__ import annotations
 from pathlib import Path
 
 from wf_config import FilesystemStoreConfig, PythonSourceConfig, WorkflowConfigFile
-from wf_platform import CapabilitySource
 
 from .context import WorkflowServer, build_local_static_workflow_server
+from .sources import WorkflowSourceProvider, collect_static_sources
 
 
 def _has_mcp_sources(config: WorkflowConfigFile) -> bool:
@@ -14,20 +14,20 @@ def _has_mcp_sources(config: WorkflowConfigFile) -> bool:
     )
 
 
-def _python_sources(config: WorkflowConfigFile) -> dict[str, CapabilitySource]:
-    from wf_sources_python import load_python_source
-
-    return {
-        source.id: load_python_source(
-            source_id=source.id,
-            path=source.path,
-            module=source.module,
-            registry=source.registry,
-            enabled=source.enabled,
-        )
+def _static_source_providers(
+    config: WorkflowConfigFile,
+) -> list[WorkflowSourceProvider]:
+    providers: list[WorkflowSourceProvider] = []
+    python_configs = [
+        source
         for source in config.server.sources
         if isinstance(source, PythonSourceConfig)
-    }
+    ]
+    if python_configs:
+        from wf_sources_python import PythonSourceProvider
+
+        providers.append(PythonSourceProvider(python_configs))
+    return providers
 
 
 def _build_mcp_workflow_server_from_workflow_config(
@@ -68,7 +68,7 @@ def build_workflow_server_from_workflow_config(
         raise ValueError("wf-rpc-server currently requires filesystem store")
     return build_local_static_workflow_server(
         store.root,
-        extra_sources=_python_sources(config),
+        extra_sources=collect_static_sources(_static_source_providers(config)),
     )
 
 

src/wf_server/sources.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from typing import Protocol
+
+from wf_platform import CapabilitySource
+
+
+class WorkflowSourceProvider(Protocol):
+    """Static source inventory provider for server composition.
+
+    This seam is intentionally narrow: providers return workflow-facing
+    `CapabilitySource` objects. Runtime pools, admin/apply hooks, and live
+    health checks remain separate provider-specific concerns.
+    """
+
+    def load_sources(self) -> Mapping[str, CapabilitySource]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class StaticSourceProvider:
+    """Adapter for already-materialized capability sources."""
+
+    sources: Mapping[str, CapabilitySource]
+
+    def load_sources(self) -> Mapping[str, CapabilitySource]:
+        return dict(self.sources)
+
+
+def collect_static_sources(
+    providers: Sequence[WorkflowSourceProvider],
+) -> dict[str, CapabilitySource]:
+    """Merge static provider inventories while rejecting ambiguous source ids."""
+    collected: dict[str, CapabilitySource] = {}
+    for provider in providers:
+        for source_id, source in provider.load_sources().items():
+            if source_id in collected:
+                raise ValueError(f"duplicate workflow source ids: {[source_id]}")
+            collected[source_id] = source
+    return collected
+
+
+__all__ = [
+    "StaticSourceProvider",
+    "WorkflowSourceProvider",
+    "collect_static_sources",
+]

src/wf_sources_python/__init__.py

@@ -1,9 +1,15 @@
 from __future__ import annotations
 
-from .loader import PythonSourceConfigLike, load_python_source, python_capability_source
+from .loader import (
+    PythonSourceConfigLike,
+    PythonSourceProvider,
+    load_python_source,
+    python_capability_source,
+)
 
 __all__ = [
     "PythonSourceConfigLike",
+    "PythonSourceProvider",
     "load_python_source",
     "python_capability_source",
 ]

src/wf_sources_python/loader.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import sys
 from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
 from importlib import import_module
 from pathlib import Path
 from typing import Any, Protocol
@@ -23,6 +24,16 @@ class PythonSourceConfigLike(Protocol):
     enabled: bool
 
 
+@dataclass(frozen=True, slots=True)
+class PythonSourceProvider:
+    """Static source provider for trusted Python source config entries."""
+
+    configs: Sequence[PythonSourceConfigLike]
+
+    def load_sources(self) -> Mapping[str, CapabilitySource]:
+        return {config.id: python_capability_source(config) for config in self.configs}
+
+
 def load_python_source(
     *,
     source_id: str,

tests/wf_server/test_config_composition.py

@@ -5,11 +5,24 @@ from pathlib import Path
 import pytest
 
 from wf_config import WorkflowConfigFile
+from wf_platform import CapabilityBuckets, CapabilitySource
 from wf_server.config import (
     build_workflow_server_from_legacy_mcp_config,
     build_workflow_server_from_workflow_config,
 )
 from wf_server.context import WorkflowServer
+from wf_server.sources import StaticSourceProvider, collect_static_sources
+
+
+class FakeSourceProvider:
+    def load_sources(self):
+        return {
+            "fake.ops": CapabilitySource(
+                id="fake.ops",
+                kind="python",
+                capabilities=CapabilityBuckets(),
+            )
+        }
 
 
 def test_workflow_config_with_python_source_exposes_capability(tmp_path: Path) -> None:
@@ -39,6 +52,27 @@ def test_workflow_config_with_python_source_exposes_capability(tmp_path: Path) -
     )
 
 
+def test_static_source_provider_protocol_collects_sources() -> None:
+    sources = collect_static_sources([FakeSourceProvider()])
+
+    assert set(sources) == {"fake.ops"}
+
+
+def test_static_source_provider_rejects_duplicate_source_ids() -> None:
+    provider = StaticSourceProvider(
+        {
+            "fake.ops": CapabilitySource(
+                id="fake.ops",
+                kind="python",
+                capabilities=CapabilityBuckets(),
+            )
+        }
+    )
+
+    with pytest.raises(ValueError, match="duplicate workflow source ids"):
+        collect_static_sources([provider, FakeSourceProvider()])
+
+
 def test_build_workflow_server_from_workflow_config_uses_local_static_for_no_mcp_sources(
     tmp_path: Path,
 ) -> None:

tests/wf_sources_python/test_loader.py

@@ -4,7 +4,15 @@ from pathlib import Path
 
 import pytest
 
-from wf_sources_python import load_python_source
+from wf_sources_python import PythonSourceProvider, load_python_source
+
+
+class PythonSourceConfigFixture:
+    id = "local.ops"
+    path = Path(".")
+    module = "tests.fixtures.python_source_ops"
+    registry = "registry"
+    enabled = True
 
 
 def test_load_python_source_from_sequence_registry() -> None:
@@ -23,6 +31,13 @@ def test_load_python_source_from_sequence_registry() -> None:
     assert source.permissions.safe_for_workflow is True
 
 
+def test_python_source_provider_loads_configured_sources() -> None:
+    sources = PythonSourceProvider([PythonSourceConfigFixture()]).load_sources()
+
+    assert set(sources) == {"local.ops"}
+    assert "local.ops.echo" in sources["local.ops"].capabilities.node_specs
+
+
 def test_load_python_source_from_callable_registry() -> None:
     source = load_python_source(
         source_id="local.ops",