feat: add server store role overrides

docs/current_roadmap.md

@@ -252,6 +252,13 @@ implementation state.
     `server.store` as the fallback for every missing role. First implementation
     should stay filesystem-only; secret managers, SQL, and object stores are
     later backend implementations.
+    First role-specific store slice complete: neutral config now accepts optional
+    `server.stores.workflow`, `server.stores.auth`,
+    `server.stores.source_registry`, and `server.stores.catalog_cache`
+    filesystem overrides. Missing roles still fall back to `server.store`.
+    MCP compatibility still uses one `FileStore` class for auth and catalog
+    snapshots internally; the separate catalog root is carried for the future
+    store split.
   - Completed: `wf run watch` starts run progress UX with polling over existing
     `inspect_run` and optional bounded `read_run_trace`. SSE/WebSocket/MCP
     progress remains deferred until polling UX proves insufficient.

docs/superpowers/specs/2026-06-03-workflow-config-targets-and-sources.md

@@ -252,6 +252,9 @@ optional role-specific overrides without breaking existing files:
 }
 ```
 
+Implementation status: first filesystem-only slice implemented. Role overrides
+are optional and fall back to `server.store`.
+
 Resolution rule:
 
 ```text

docs/wf_cli.md

@@ -55,9 +55,10 @@ The old `store_root` field maps to
 
 `server.store` is currently the default root for all file-backed server state:
 workflow artifacts/deployments/runs, source registry entries, catalog cache, and
-local/dev auth records. Future configs may add role-specific store overrides
-such as `server.stores.auth` or `server.stores.source_registry`; missing roles
-should continue to fall back to `server.store`.
+local/dev auth records. Role-specific store overrides are now supported via
+`server.stores.*` (e.g. `server.stores.workflow`, `server.stores.auth`,
+`server.stores.source_registry`, `server.stores.catalog_cache`); missing roles
+continue to fall back to `server.store`.
 
 Start a JSON-RPC server backed by MCP broker config and MCP-capable sources:
 

src/wf_cli/context.py

@@ -148,7 +148,7 @@ def load_cli_context(
     config = load_workflow_config(resolved_config_path)
     target = config.client.target
     if force_local or isinstance(target, LocalTargetConfig):
-        store = config.server.store
+        store = config.server.workflow_store
         if not isinstance(store, FilesystemStoreConfig):
             raise ValueError("local CLI target currently requires filesystem store")
         server = build_local_static_workflow_server(store.root)

src/wf_config/__init__.py

@@ -10,6 +10,7 @@ from .models import (
     RpcHttpTargetConfig,
     RpcHttpTransportConfig,
     ServerConfig,
+    ServerStoresConfig,
     SourceConfigOwnership,
     SourceTransportConfig,
     StdioSourceTransportConfig,
@@ -27,6 +28,7 @@ __all__ = [
     "RpcHttpTargetConfig",
     "RpcHttpTransportConfig",
     "ServerConfig",
+    "ServerStoresConfig",
     "SourceConfigOwnership",
     "SourceTransportConfig",
     "StdioSourceTransportConfig",

src/wf_config/loader.py

@@ -3,7 +3,12 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
-from .models import FilesystemStoreConfig, WorkflowConfigFile
+from .models import (
+    FilesystemStoreConfig,
+    ServerStoresConfig,
+    StoreConfig,
+    WorkflowConfigFile,
+)
 
 
 def load_workflow_config(path: str | Path) -> WorkflowConfigFile:
@@ -11,22 +16,50 @@ def load_workflow_config(path: str | Path) -> WorkflowConfigFile:
 
     Relative filesystem store roots are config-file relative so `wf --config`
     behaves the same regardless of the caller's current working directory.
+    Role-specific store overrides follow the same rule.
     """
 
     config_path = Path(path)
     data = json.loads(config_path.read_text(encoding="utf-8"))
     config = WorkflowConfigFile.model_validate(data)
-    store = config.server.store
+    return _resolve_store_paths(config, base_dir=config_path.parent)
+
+
+def _resolve_store_paths(
+    config: WorkflowConfigFile,
+    *,
+    base_dir: Path,
+) -> WorkflowConfigFile:
+    server = config.server
+    resolved_stores = ServerStoresConfig(
+        workflow=_resolve_store(server.stores.workflow, base_dir=base_dir),
+        auth=_resolve_store(server.stores.auth, base_dir=base_dir),
+        source_registry=_resolve_store(
+            server.stores.source_registry,
+            base_dir=base_dir,
+        ),
+        catalog_cache=_resolve_store(
+            server.stores.catalog_cache,
+            base_dir=base_dir,
+        ),
+    )
+    return config.model_copy(
+        update={
+            "server": server.model_copy(
+                update={
+                    "store": _resolve_store(server.store, base_dir=base_dir),
+                    "stores": resolved_stores,
+                }
+            )
+        }
+    )
+
+
+def _resolve_store(
+    store: StoreConfig | None,
+    *,
+    base_dir: Path,
+) -> StoreConfig | None:
     if isinstance(store, FilesystemStoreConfig) and not store.root.is_absolute():
-        config = config.model_copy(
-            update={
-                "server": config.server.model_copy(
-                    update={
-                        "store": store.model_copy(
-                            update={"root": (config_path.parent / store.root).resolve()}
-                        )
-                    }
-                )
-            }
-        )
-    return config
+        return store.model_copy(update={"root": (base_dir / store.root).resolve()})
+    return store

src/wf_config/models.py

@@ -139,8 +139,23 @@ SourceConfig = Annotated[
 ]
 
 
+class ServerStoresConfig(WorkflowConfigModel):
+    """Optional role-specific store overrides.
+
+    Missing roles fall back to `ServerConfig.store`. Keep this config role-based
+    so future backends can split workflow records, auth, desired sources, and
+    cache storage independently.
+    """
+
+    workflow: StoreConfig | None = None
+    auth: StoreConfig | None = None
+    source_registry: StoreConfig | None = None
+    catalog_cache: StoreConfig | None = None
+
+
 class ServerConfig(WorkflowConfigModel):
     store: StoreConfig = Field(default_factory=FilesystemStoreConfig)
+    stores: ServerStoresConfig = Field(default_factory=ServerStoresConfig)
     transports: list[ServerTransportConfig] = Field(default_factory=list)
     sources: list[SourceConfig] = Field(default_factory=list)
 
@@ -153,6 +168,22 @@ class ServerConfig(WorkflowConfigModel):
             seen.add(source.id)
         return self
 
+    @property
+    def workflow_store(self) -> StoreConfig:
+        return self.stores.workflow or self.store
+
+    @property
+    def auth_store(self) -> StoreConfig:
+        return self.stores.auth or self.store
+
+    @property
+    def source_registry_store(self) -> StoreConfig:
+        return self.stores.source_registry or self.store
+
+    @property
+    def catalog_cache_store(self) -> StoreConfig:
+        return self.stores.catalog_cache or self.store
+
 
 class WorkflowConfigFile(WorkflowConfigModel):
     version: Literal[1] = 1

src/wf_mcp/broker/config.py

@@ -9,6 +9,7 @@ from wf_config.models import FilesystemStoreConfig, McpSourceConfig, ServerConfi
 
 from ..control import BrokerConfigFile, ConnectionConfigFile
 from ..models import BrokerConfig
+from .models import BrokerStoreRoots
 from ..runtime import McpRuntimePool, PersistentSessionFactory
 from ..sdk import McpSdkAdapter
 from ..source_registry import (
@@ -104,10 +105,32 @@ def load_broker_config(path: str | Path) -> BrokerConfig:
     return BrokerConfigFile.model_validate(data).to_runtime(config_path=config_path)
 
 
+def _filesystem_store_root(store: object, *, role: str) -> Path:
+    if not isinstance(store, FilesystemStoreConfig):
+        raise ValueError(f"MCP-backed workflow server requires filesystem {role} store")
+    return store.root
+
+
 def broker_config_from_workflow_config(config: WorkflowConfigFile) -> BrokerConfig:
     """Create MCP broker runtime config from neutral workflow server config."""
     return BrokerConfig(
         store_root=config.server.store.root,
+        store_roots=BrokerStoreRoots(
+            default_root=config.server.store.root,
+            workflow_root=_filesystem_store_root(
+                config.server.workflow_store,
+                role="workflow",
+            ),
+            auth_root=_filesystem_store_root(config.server.auth_store, role="auth"),
+            source_registry_root=_filesystem_store_root(
+                config.server.source_registry_store,
+                role="source_registry",
+            ),
+            catalog_cache_root=_filesystem_store_root(
+                config.server.catalog_cache_store,
+                role="catalog_cache",
+            ),
+        ),
         connections=[
             workflow_mcp_source_to_connection_config(source)
             for source in config.server.sources
@@ -139,9 +162,13 @@ def migrate_broker_config_file(path: str | Path) -> WorkflowConfigFile:
 def build_service_from_config(config: BrokerConfig) -> WfMcpService:
     """Create a broker service with SDK adapters for configured connections."""
     runtime_factory = PersistentSessionFactory()
-    workflow_stores = file_workflow_stores(config.store_root)
+    store_roots = config.store_roots or BrokerStoreRoots.from_default(config.store_root)
+    workflow_stores = file_workflow_stores(store_roots.workflow_root)
+    # FileStore still owns both auth files and catalog snapshots. Role roots are
+    # carried separately so a later FileStore split can move catalog_cache without a
+    # config migration.
     service = WfMcpService(
-        store=FileStore(config.store_root),
+        store=FileStore(store_roots.auth_root),
         artifact_store=workflow_stores.artifact_store,
         draft_workspace_store=workflow_stores.draft_workspace_store,
         run_store=workflow_stores.run_store,
@@ -150,7 +177,7 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
         # across sequential workflow nodes.
         tool_executor=McpRuntimePool(runtime_factory.create),
     )
-    source_registry_store = FileSourceRegistryStore(config.store_root)
+    source_registry_store = FileSourceRegistryStore(store_roots.source_registry_root)
     service.sync_connections_from_config(
         config,
         source_registry_store=source_registry_store,

src/wf_mcp/broker/models.py

@@ -7,6 +7,32 @@ from typing import Any, Literal
 SourceConfigOwnership = Literal["locked", "seed"]
 
 
+@dataclass(frozen=True, slots=True)
+class BrokerStoreRoots:
+    """Resolved filesystem roots for MCP compatibility stores.
+
+    `default_root` preserves legacy `store_root` behavior. Role roots let
+    neutral config split workflow records, auth, desired source registry, and
+    catalog/cache storage without changing legacy config files.
+    """
+
+    default_root: Path
+    workflow_root: Path
+    auth_root: Path
+    source_registry_root: Path
+    catalog_cache_root: Path
+
+    @classmethod
+    def from_default(cls, root: Path) -> BrokerStoreRoots:
+        return cls(
+            default_root=root,
+            workflow_root=root,
+            auth_root=root,
+            source_registry_root=root,
+            catalog_cache_root=root,
+        )
+
+
 @dataclass(slots=True)
 class ConnectionConfig:
     id: str
@@ -21,10 +47,16 @@ class ConnectionConfig:
 class BrokerConfig:
     store_root: Path
     connections: list[ConnectionConfig] = field(default_factory=list)
+    store_roots: BrokerStoreRoots | None = None
+
+    def __post_init__(self) -> None:
+        if self.store_roots is None:
+            self.store_roots = BrokerStoreRoots.from_default(self.store_root)
 
 
 __all__ = [
     "BrokerConfig",
+    "BrokerStoreRoots",
     "ConnectionConfig",
     "SourceConfigOwnership",
 ]

src/wf_mcp/broker/server.py

@@ -107,10 +107,13 @@ def workflow_server_from_service(
 def build_workflow_server_from_config(config: BrokerConfig) -> WorkflowServer:
     """Build a neutral WorkflowServer backed by MCP broker runtime services."""
     service = build_service_from_config(config)
+    assert config.store_roots is not None
     return workflow_server_from_service(
         service,
         config=config,
-        source_registry_store=FileSourceRegistryStore(config.store_root),
+        source_registry_store=FileSourceRegistryStore(
+            config.store_roots.source_registry_root
+        ),
     )
 
 

src/wf_mcp/models.py

@@ -2,12 +2,18 @@ from __future__ import annotations
 
 from wf_api.models import RawWorkflowPlan
 from wf_mcp.auth import AuthRecord
-from wf_mcp.broker.models import BrokerConfig, ConnectionConfig, SourceConfigOwnership
+from wf_mcp.broker.models import (
+    BrokerConfig,
+    BrokerStoreRoots,
+    ConnectionConfig,
+    SourceConfigOwnership,
+)
 from wf_mcp.catalog.models import CatalogSnapshot, dump_catalog_snapshot
 
 __all__ = [
     "AuthRecord",
     "BrokerConfig",
+    "BrokerStoreRoots",
     "CatalogSnapshot",
     "ConnectionConfig",
     "RawWorkflowPlan",

src/wf_server/config.py

@@ -44,7 +44,7 @@ def build_workflow_server_from_workflow_config(
     """
     if _has_mcp_sources(config):
         return _build_mcp_workflow_server_from_workflow_config(config)
-    store = config.server.store
+    store = config.server.workflow_store
     if not isinstance(store, FilesystemStoreConfig):
         # Roadmap: SQL/transactional stores are deferred until the remote server
         # storage boundary is proven with file-backed stores.

src/wf_transport_rpc_http/cli.py

@@ -66,7 +66,7 @@ def serve(
 
     if config is not None:
         workflow_config = load_workflow_config(config)
-        store = workflow_config.server.store
+        store = workflow_config.server.workflow_store
         if server is None and store_root is None:
             server = build_workflow_server_from_workflow_config(workflow_config)
         elif server is None:

tests/wf_cli/test_context.py

@@ -15,6 +15,7 @@ from wf_cli.context import (
     rpc_timeout_from_context,
     rpc_url_from_context,
 )
+from wf_server import build_local_static_workflow_server
 
 
 def _typer_context(obj: object | None) -> typer.Context:
@@ -96,3 +97,43 @@ def test_load_cli_context_builds_service_and_handlers(tmp_path: Path) -> None:
         context.handlers.context.draft_workspace_store is service.draft_workspace_store
     )
     assert context.handlers.context.run_store is service.run_store
+
+
+def test_load_cli_context_local_uses_workflow_store_override(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    config_path = tmp_path / "wf.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "client": {"target": {"kind": "local"}},
+                "server": {
+                    "store": {"kind": "filesystem", "root": ".default"},
+                    "stores": {
+                        "workflow": {
+                            "kind": "filesystem",
+                            "root": ".workflow",
+                        }
+                    },
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    captured: dict[str, object] = {}
+
+    def fake_build_local_static_workflow_server(root):
+        captured["store_root"] = root
+        return build_local_static_workflow_server(tmp_path / "actual")
+
+    monkeypatch.setattr(
+        "wf_cli.context.build_local_static_workflow_server",
+        fake_build_local_static_workflow_server,
+    )
+
+    context = load_cli_context(config_path)
+
+    assert context.service is None
+    assert captured["store_root"] == (tmp_path / ".workflow").resolve()

tests/wf_config/test_config_models.py

@@ -274,6 +274,38 @@ def test_workflow_config_rejects_unsafe_mcp_source_id() -> None:
         )
 
 
+def test_workflow_config_parses_role_specific_store_overrides() -> None:
+    config = WorkflowConfigFile.model_validate(
+        {
+            "version": 1,
+            "server": {
+                "store": {"kind": "filesystem", "root": ".wf_store"},
+                "stores": {
+                    "workflow": {"kind": "filesystem", "root": ".wf_workflow"},
+                    "auth": {"kind": "filesystem", "root": ".wf_auth"},
+                    "source_registry": {
+                        "kind": "filesystem",
+                        "root": ".wf_sources",
+                    },
+                    "catalog_cache": {
+                        "kind": "filesystem",
+                        "root": ".wf_catalog",
+                    },
+                },
+            },
+        }
+    )
+
+    assert isinstance(config.server.stores.workflow, FilesystemStoreConfig)
+    assert config.server.stores.workflow.root.as_posix() == ".wf_workflow"
+    assert isinstance(config.server.stores.auth, FilesystemStoreConfig)
+    assert config.server.stores.auth.root.as_posix() == ".wf_auth"
+    assert isinstance(config.server.stores.source_registry, FilesystemStoreConfig)
+    assert config.server.stores.source_registry.root.as_posix() == ".wf_sources"
+    assert isinstance(config.server.stores.catalog_cache, FilesystemStoreConfig)
+    assert config.server.stores.catalog_cache.root.as_posix() == ".wf_catalog"
+
+
 def test_workflow_config_rejects_duplicate_source_ids_across_kinds() -> None:
     with pytest.raises(ValidationError, match="duplicate source id"):
         WorkflowConfigFile.model_validate(
@@ -293,3 +325,80 @@ def test_workflow_config_rejects_duplicate_source_ids_across_kinds() -> None:
                 },
             }
         )
+
+
+def test_load_workflow_config_resolves_role_store_paths_relative_to_config(
+    tmp_path: Path,
+) -> None:
+    config_path = tmp_path / "nested" / "wf.json"
+    config_path.parent.mkdir()
+    config_path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "server": {
+                    "store": {"kind": "filesystem", "root": ".default_store"},
+                    "stores": {
+                        "workflow": {
+                            "kind": "filesystem",
+                            "root": ".workflow_store",
+                        },
+                        "auth": {
+                            "kind": "filesystem",
+                            "root": ".auth_store",
+                        },
+                        "source_registry": {
+                            "kind": "filesystem",
+                            "root": ".source_store",
+                        },
+                        "catalog_cache": {
+                            "kind": "filesystem",
+                            "root": ".catalog_store",
+                        },
+                    },
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    config = load_workflow_config(config_path)
+
+    assert config.server.store.root == (
+        config_path.parent / ".default_store"
+    ).resolve()
+    assert config.server.stores.workflow is not None
+    assert config.server.stores.workflow.root == (
+        config_path.parent / ".workflow_store"
+    ).resolve()
+    assert config.server.stores.auth is not None
+    assert config.server.stores.auth.root == (
+        config_path.parent / ".auth_store"
+    ).resolve()
+    assert config.server.stores.source_registry is not None
+    assert config.server.stores.source_registry.root == (
+        config_path.parent / ".source_store"
+    ).resolve()
+    assert config.server.stores.catalog_cache is not None
+    assert config.server.stores.catalog_cache.root == (
+        config_path.parent / ".catalog_store"
+    ).resolve()
+
+
+def test_server_config_resolves_missing_role_stores_to_default_store() -> None:
+    config = WorkflowConfigFile.model_validate(
+        {
+            "version": 1,
+            "server": {
+                "store": {"kind": "filesystem", "root": ".default"},
+                "stores": {
+                    "auth": {"kind": "filesystem", "root": ".auth"},
+                },
+            },
+        }
+    )
+
+    assert config.server.workflow_store.root.as_posix() == ".default"
+    assert config.server.auth_store.root.as_posix() == ".auth"
+    assert config.server.source_registry_store.root.as_posix() == ".default"
+    assert config.server.catalog_cache_store.root.as_posix() == ".default"

tests/wf_mcp/test_workflow_config_bridge.py

@@ -106,3 +106,88 @@ def test_broker_config_from_workflow_config_ignores_non_mcp_sources(
 
     assert broker_config.store_root == tmp_path / "store"
     assert broker_config.connections == []
+
+
+def test_broker_config_from_workflow_config_carries_role_store_roots() -> None:
+    config = WorkflowConfigFile.model_validate(
+        {
+            "version": 1,
+            "server": {
+                "store": {"kind": "filesystem", "root": ".default"},
+                "stores": {
+                    "workflow": {"kind": "filesystem", "root": ".workflow"},
+                    "auth": {"kind": "filesystem", "root": ".auth"},
+                    "source_registry": {
+                        "kind": "filesystem",
+                        "root": ".sources",
+                    },
+                    "catalog_cache": {
+                        "kind": "filesystem",
+                        "root": ".catalog",
+                    },
+                },
+            },
+        }
+    )
+
+    broker = broker_config_from_workflow_config(config)
+
+    assert broker.store_roots.workflow_root == Path(".workflow")
+    assert broker.store_roots.auth_root == Path(".auth")
+    assert broker.store_roots.source_registry_root == Path(".sources")
+    assert broker.store_roots.catalog_cache_root == Path(".catalog")
+
+
+def test_build_service_from_neutral_config_uses_role_store_roots(
+    tmp_path: Path,
+) -> None:
+    config = WorkflowConfigFile.model_validate(
+        {
+            "version": 1,
+            "server": {
+                "store": {"kind": "filesystem", "root": str(tmp_path / "default")},
+                "stores": {
+                    "workflow": {
+                        "kind": "filesystem",
+                        "root": str(tmp_path / "workflow"),
+                    },
+                    "auth": {
+                        "kind": "filesystem",
+                        "root": str(tmp_path / "auth"),
+                    },
+                    "source_registry": {
+                        "kind": "filesystem",
+                        "root": str(tmp_path / "sources"),
+                    },
+                    "catalog_cache": {
+                        "kind": "filesystem",
+                        "root": str(tmp_path / "catalog"),
+                    },
+                },
+                "sources": [
+                    {
+                        "kind": "mcp",
+                        "id": "everything.default",
+                        "provider": "everything",
+                        "account": "default",
+                        "transport": {
+                            "kind": "stdio",
+                            "command": "uvx",
+                            "args": ["mcp-server-everything"],
+                        },
+                    }
+                ],
+            },
+        }
+    )
+
+    from wf_mcp.broker.config import build_service_from_config
+
+    broker = broker_config_from_workflow_config(config)
+    service = build_service_from_config(broker)
+
+    assert service.store.root == tmp_path / "auth"
+    assert service.artifact_store.root == tmp_path / "workflow"
+    assert service.draft_workspace_store.root == tmp_path / "workflow"
+    assert service.run_store.root == tmp_path / "workflow"
+    assert (tmp_path / "sources").exists()

tests/wf_transport_rpc_http/test_cli.py

@@ -374,3 +374,52 @@ def test_rpc_server_cli_rejects_store_root_with_mcp_source_config(tmp_path) -> N
 
     assert result.exit_code != 0
     assert "--store-root cannot override MCP-source config" in result.output
+
+
+def test_rpc_server_cli_config_uses_workflow_store_override(
+    tmp_path,
+    monkeypatch,
+) -> None:
+    config_path = tmp_path / "wf.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "server": {
+                    "store": {"kind": "filesystem", "root": ".default"},
+                    "stores": {
+                        "workflow": {
+                            "kind": "filesystem",
+                            "root": ".workflow",
+                        }
+                    },
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    captured: dict[str, object] = {}
+
+    def fake_build_server(config):
+        captured["workflow_store_root"] = config.server.workflow_store.root
+        return object()
+
+    def fake_create_rpc_app(server, *, rpc_path="/rpc"):
+        captured["server"] = server
+        captured["rpc_path"] = rpc_path
+        return object()
+
+    def fake_uvicorn_run(app_obj, *, host, port, access_log):
+        captured["app"] = app_obj
+
+    monkeypatch.setattr(
+        "wf_transport_rpc_http.cli.build_workflow_server_from_workflow_config",
+        fake_build_server,
+    )
+    monkeypatch.setattr("wf_transport_rpc_http.cli.create_rpc_app", fake_create_rpc_app)
+    monkeypatch.setattr("wf_transport_rpc_http.cli.uvicorn.run", fake_uvicorn_run)
+
+    result = CliRunner().invoke(app, ["--config", str(config_path)])
+
+    assert result.exit_code == 0, result.output
+    assert captured["workflow_store_root"] == (tmp_path / ".workflow").resolve()