first slice: operation context scaffolding

splitting WorkflowSurfaceHandlers (because it is BIG)

docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md

@@ -273,6 +273,91 @@ async def list_capabilities(...):
     return await self.capabilities.list_capabilities(...)
 ```
 
+### Planned Domain Split Order
+
+After Slice 4A proves the operation-context seam, split method groups in small
+behavior-preserving slices:
+
+#### Slice 4B: Drafts First
+
+Move stateless draft and draft workspace operations first:
+
+```text
+validate_draft
+compile_draft
+patch_draft
+list_draft_workspaces
+create_draft_workspace
+get_draft_workspace
+delete_draft_workspace
+validate_draft_workspace
+patch_draft_workspace
+set_draft_name
+set_draft_route
+set_step_input_map
+set_step_output_map
+create_minimal_draft_workspace
+create_draft_workspace_from_capability
+```
+
+Reason: drafts mostly use the draft workspace store, workflow draft compiler,
+wrapper hints, and deterministic patch helpers. They have the lowest live-source
+and durable-runtime coupling.
+
+#### Slice 4C: Artifacts And Deployments
+
+Move saved artifact and deployment operations next:
+
+```text
+list_artifacts
+save_artifact
+create_artifact_from_plan
+create_artifact_from_draft
+create_artifact_from_workspace
+create_wrapper_from_workspace
+inspect_artifact
+list_deployments
+inspect_deployment
+save_deployment
+delete_deployment
+validate_deployment
+```
+
+Reason: this group is store-heavy and introduces dependency validation, saved
+subgraph tree resolution, and event recording. It should move only after drafts
+prove the context seam.
+
+#### Slice 4D: Runs
+
+Move run lifecycle operations after artifacts/deployments:
+
+```text
+run_deployment
+resume_run
+inspect_run
+read_run_trace
+```
+
+Reason: runs are runtime-sensitive. They touch durable checkpoints, pinned
+dependency environments, resume readiness, prepared saved subgraphs, trace
+slicing, and compact next-action guidance. This should not be the first method
+move.
+
+#### Slice 4E: Capabilities Last
+
+Move workflow capability operations last:
+
+```text
+list_capabilities
+inspect_capability
+call_capability
+```
+
+Reason: capabilities look simple but are the messiest boundary. They combine
+planner-visible source inventory, wrapper artifacts, direct wrapper calls,
+external live source calls, source visibility, and schema/wrapper hints. Keep
+them in the MCP-backed implementation until the other domain services are stable.
+
 ### If/Then
 
 - If the context starts mirroring all of `WfMcpService`, stop and split it into

src/wf_api/__init__.py

@@ -23,6 +23,15 @@ from .wrapper_hints import (
     wrapper_hints_for_capability,
 )
 
+from .operation_context import (
+    WorkflowArtifactCataloger,
+    WorkflowEventRecorder,
+    WorkflowLiveSourceChecker,
+    WorkflowOperationContext,
+    WorkflowRuntimeRunner,
+    WorkflowSpecProvider,
+)
+
 from .runtime_dependencies import RuntimeDependencies, resolve_runtime_dependencies
 
 __all__ = [
@@ -42,6 +51,12 @@ __all__ = [
     "TraceRange",
     "WorkflowApi",
     "WorkflowApiBackend",
+    "WorkflowArtifactCataloger",
+    "WorkflowEventRecorder",
+    "WorkflowLiveSourceChecker",
+    "WorkflowOperationContext",
+    "WorkflowRuntimeRunner",
+    "WorkflowSpecProvider",
     "WorkflowSurfaceCapabilityId",
     "WrapperAuthoringHints",
     "WrapperHintConfidence",

src/wf_api/operation_context.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from wf_artifacts import (
+    DraftWorkspaceStore,
+    RunStore,
+    WorkflowArtifact,
+    WorkflowArtifactCatalogEntry,
+    WorkflowArtifactStore,
+)
+from wf_authoring import AsyncRegistryHandler
+from wf_core import RunState
+from wf_core.runtime.ops.merges import ReducerDefinition
+from wf_platform import CapabilitySource
+
+from .models import RawWorkflowPlan
+
+
+class WorkflowEventRecorder(Protocol):
+    """Records workflow lifecycle events without exposing MCP event types."""
+
+    def record_event(self, event: object) -> None:
+        """Record one event object supplied by an adapter-owned event factory."""
+        ...
+
+
+class WorkflowSpecProvider(Protocol):
+    """Provides planner-visible capability sources and qualified node specs."""
+
+    @property
+    def capability_sources(self) -> Mapping[str, CapabilitySource]:
+        """Planner-visible capability sources keyed by source id."""
+        ...
+
+    def get_qualified_spec(self, qualified_name: str) -> object:
+        """Return the node spec for one fully qualified capability name."""
+        ...
+
+
+class WorkflowArtifactCataloger(Protocol):
+    """Formats saved workflow artifacts for list/detail surfaces."""
+
+    def workflow_artifact_catalog_entry(
+        self, artifact: WorkflowArtifact
+    ) -> WorkflowArtifactCatalogEntry:
+        """Return the catalog entry representation for one saved artifact."""
+        ...
+
+
+class WorkflowRuntimeRunner(Protocol):
+    """Runs and resumes workflow plans using an adapter-owned runtime backend."""
+
+    async def run_workflow_from_plan(
+        self,
+        plan: RawWorkflowPlan,
+        *,
+        workflow_input: dict[str, Any],
+        node_name_bindings: dict[str, str] | None = None,
+        registry: dict[str, AsyncRegistryHandler] | None = None,
+        reducers: dict[str, ReducerDefinition] | None = None,
+        prepared_subgraphs: dict[str, object] | None = None,
+    ) -> RunState:
+        """Execute one raw workflow plan and return its run state."""
+        ...
+
+    async def resume_workflow_from_plan(
+        self,
+        plan: RawWorkflowPlan,
+        *,
+        run: RunState,
+        resume_payload: dict[str, Any],
+        resume_outcome: str,
+        node_name_bindings: dict[str, str] | None = None,
+        registry: dict[str, AsyncRegistryHandler] | None = None,
+        reducers: dict[str, ReducerDefinition] | None = None,
+        prepared_subgraphs: dict[str, object] | None = None,
+    ) -> RunState:
+        """Resume one interrupted raw workflow plan and return its run state."""
+        ...
+
+
+class WorkflowLiveSourceChecker(Protocol):
+    """Optional hook for validating live external source availability."""
+
+    async def available_sources(self) -> list[object]:
+        """Return source availability records understood by the caller."""
+        ...
+
+
+@dataclass(frozen=True, slots=True)
+class WorkflowOperationContext:
+    """Protocol-neutral dependencies needed by workflow API operations.
+
+    This is scaffolding for splitting the large MCP-backed handler into domain
+    services. Keep this shape explicit; do not add arbitrary access to the whole
+    MCP service.
+    """
+
+    artifact_store: WorkflowArtifactStore | None
+    draft_workspace_store: DraftWorkspaceStore | None
+    run_store: RunStore | None
+    capability_sources: Mapping[str, CapabilitySource]
+    events: WorkflowEventRecorder
+    specs: WorkflowSpecProvider
+    artifacts: WorkflowArtifactCataloger
+    runtime: WorkflowRuntimeRunner
+    live_sources: WorkflowLiveSourceChecker | None = None
+
+
+__all__ = [
+    "WorkflowArtifactCataloger",
+    "WorkflowEventRecorder",
+    "WorkflowLiveSourceChecker",
+    "WorkflowOperationContext",
+    "WorkflowRuntimeRunner",
+    "WorkflowSpecProvider",
+]

src/wf_mcp/broker/service/workflow_operation_context.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from wf_api.operation_context import (
+    WorkflowArtifactCataloger,
+    WorkflowEventRecorder,
+    WorkflowLiveSourceChecker,
+    WorkflowOperationContext,
+    WorkflowRuntimeRunner,
+    WorkflowSpecProvider,
+)
+
+from .core import WfMcpService
+
+
+@dataclass(frozen=True, slots=True)
+class WfMcpWorkflowEventRecorder(WorkflowEventRecorder):
+    """Adapter-owned event recorder backed by WfMcpService."""
+
+    service: WfMcpService
+
+    def record_event(self, event: Any) -> None:
+        self.service._record_event(event)  # noqa: SLF001
+
+
+@dataclass(frozen=True, slots=True)
+class WfMcpWorkflowSpecProvider(WorkflowSpecProvider):
+    """Adapter-owned spec provider backed by WfMcpService."""
+
+    service: WfMcpService
+
+    @property
+    def capability_sources(self):
+        return self.service.capability_sources
+
+    def get_qualified_spec(self, qualified_name: str) -> object:
+        return self.service._get_qualified_spec(qualified_name)  # noqa: SLF001
+
+
+@dataclass(frozen=True, slots=True)
+class WfMcpWorkflowArtifactCataloger(WorkflowArtifactCataloger):
+    """Adapter-owned artifact catalog formatter backed by WfMcpService."""
+
+    service: WfMcpService
+
+    def workflow_artifact_catalog_entry(self, artifact):
+        return self.service.workflow_artifact_catalog_entry(artifact)
+
+
+@dataclass(frozen=True, slots=True)
+class WfMcpWorkflowRuntimeRunner(WorkflowRuntimeRunner):
+    """Adapter-owned runtime runner backed by WfMcpService."""
+
+    service: WfMcpService
+
+    async def run_workflow_from_plan(self, plan, **kwargs):
+        return await self.service.run_workflow_from_plan(plan, **kwargs)
+
+    async def resume_workflow_from_plan(self, plan, **kwargs):
+        return await self.service.resume_workflow_from_plan(plan, **kwargs)
+
+
+@dataclass(frozen=True, slots=True)
+class WfMcpWorkflowLiveSourceChecker(WorkflowLiveSourceChecker):
+    """Placeholder live source checker while Slice 4A only defines the seam.
+
+    See `docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md`.
+    Real live source checks still live in workflow handlers until a later
+    capability-domain extraction can move them without dragging MCP adapters
+    into `wf_api`.
+    """
+
+    service: WfMcpService
+
+    async def available_sources(self) -> list[object]:
+        # Existing live source availability logic still lives near handlers.
+        # Slice 4A only creates the seam; it does not move live-check behavior.
+        return []
+
+
+def context_from_service(service: WfMcpService) -> WorkflowOperationContext:
+    """Adapt the current MCP service stack into a protocol-neutral context."""
+    specs = WfMcpWorkflowSpecProvider(service)
+    return WorkflowOperationContext(
+        artifact_store=service.artifact_store,
+        draft_workspace_store=service.draft_workspace_store,
+        run_store=service.run_store,
+        capability_sources=specs.capability_sources,
+        events=WfMcpWorkflowEventRecorder(service),
+        specs=specs,
+        artifacts=WfMcpWorkflowArtifactCataloger(service),
+        runtime=WfMcpWorkflowRuntimeRunner(service),
+        live_sources=WfMcpWorkflowLiveSourceChecker(service),
+    )
+
+
+__all__ = [
+    "WfMcpWorkflowArtifactCataloger",
+    "WfMcpWorkflowEventRecorder",
+    "WfMcpWorkflowLiveSourceChecker",
+    "WfMcpWorkflowRuntimeRunner",
+    "WfMcpWorkflowSpecProvider",
+    "context_from_service",
+]

tests/wf_api/test_operation_context.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import ast
+import json
+from pathlib import Path
+
+from wf_api.operation_context import WorkflowOperationContext
+from wf_cli.context import load_cli_context
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
+
+
+def test_wf_api_operation_context_imports_no_wf_mcp() -> None:
+    path = Path(__file__).resolve().parents[2] / "src" / "wf_api" / "operation_context.py"
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+
+    violations: list[str] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and node.module is not None:
+            if node.module.startswith("wf_mcp"):
+                violations.append(f"{node.lineno}: from {node.module} import ...")
+        elif isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name.startswith("wf_mcp"):
+                    violations.append(f"{node.lineno}: import {alias.name}")
+
+    assert violations == []
+
+
+def test_context_from_service_exposes_existing_store_objects(tmp_path: Path) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "store_root": ".wf_mcp_store",
+                "connections": [
+                    {
+                        "id": "demo.personal",
+                        "server": "demo",
+                        "account": "personal",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    cli_context = load_cli_context(config_path)
+
+    operation_context = context_from_service(cli_context.service)
+
+    assert isinstance(operation_context, WorkflowOperationContext)
+    assert operation_context.artifact_store is cli_context.service.artifact_store
+    assert operation_context.draft_workspace_store is cli_context.service.draft_workspace_store
+    assert operation_context.run_store is cli_context.service.run_store
+    assert operation_context.capability_sources is cli_context.service.capability_sources
+
+
+def test_context_from_service_delegates_specs_and_events(tmp_path: Path) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(
+        json.dumps({"store_root": ".wf_mcp_store", "connections": []}),
+        encoding="utf-8",
+    )
+    cli_context = load_cli_context(config_path)
+    operation_context = context_from_service(cli_context.service)
+
+    event = object()
+    operation_context.events.record_event(event)
+
+    assert cli_context.service.list_events()[-1] is event