second slice: drafts system

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

@@ -297,13 +297,17 @@ 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.
 
+Leave `create_draft_workspace_from_capability` in the MCP-backed handler during
+4B. It depends on `inspect_capability`, wrapper hints, and capability source
+inspection, so it should move with either a small follow-up capability bootstrap
+slice or Slice 4E.
+
 #### Slice 4C: Artifacts And Deployments
 
 Move saved artifact and deployment operations next:

docs/superpowers/plans/2026-06-01-wf-api-slice-4b-draft-service.md

@@ -0,0 +1,502 @@
+# wf_api Slice 4B: Draft Service Extraction Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Move draft validation, draft workspace editing, and minimal draft bootstrapping out of `WorkflowSurfaceHandlers` into a protocol-neutral `wf_api.drafts.WorkflowDraftApi`.
+
+**Architecture:** `WorkflowDraftApi` depends on `WorkflowOperationContext`, not `WfMcpService`. `WorkflowSurfaceHandlers` keeps the public method names and delegates the draft subset to `WorkflowDraftApi`. This is the first real method-body extraction, so the slice intentionally excludes methods that save artifacts, record events, or call `inspect_capability`.
+
+**Tech Stack:** Python 3.14+, `wf_api.operation_context`, `wf_artifacts` draft helpers, `wf_core` path/binding models, pytest, ruff, basedpyright.
+
+---
+
+## Scope
+
+### Move In This Slice
+
+Move these methods from `WorkflowSurfaceHandlers` to `wf_api.drafts.WorkflowDraftApi`:
+
+```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
+```
+
+Move these draft-only helper functions to `wf_api.drafts`:
+
+```text
+_required_capabilities_for_plan
+_required_capability_payloads
+_observed_node_specs
+_draft_input_maps
+_draft_output_map
+_draft_input_bindings_payload
+_draft_output_bindings_payload
+_graph_path_payload
+_local_path_payload
+_state_path_payload
+_escape_json_pointer
+```
+
+### Do Not Move In This Slice
+
+Do not move:
+
+```text
+create_draft_workspace_from_capability
+create_artifact_from_draft
+create_artifact_from_workspace
+create_wrapper_from_workspace
+```
+
+Reasons:
+
+- `create_draft_workspace_from_capability` depends on `inspect_capability`, which belongs to the capability domain. Move it later after the capability-inspection seam is explicit.
+- artifact-from-draft/workspace methods save artifacts and record events. Move them with artifacts/deployments in Slice 4C.
+
+Temporary duplication is allowed for private helper functions used by both
+draft preview and artifact creation. If a helper still has live callers in
+`WorkflowSurfaceHandlers` after draft delegation, keep the old copy until Slice
+4C moves the artifact/deployment methods. Do not make `wf_api` import
+`wf_mcp` just to avoid duplication.
+
+### Invariants
+
+- No public payload changes.
+- No MCP tool schema changes.
+- `WorkflowSurfaceHandlers` still exposes the same methods.
+- `wf_api` imports no `wf_mcp`.
+- Draft methods delegate through `WorkflowDraftApi`.
+
+---
+
+## Task 1: Create `wf_api.drafts`
+
+**Files:**
+- Create: `src/wf_api/drafts.py`
+
+- [ ] **Step 1: Create `WorkflowDraftApi` skeleton**
+
+Create `src/wf_api/drafts.py` with imports and class skeleton:
+
+```python
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from wf_artifacts import (
+    DraftWorkspaceStore,
+    RequiredCapability,
+    build_workflow_artifact_from_plan,
+    compile_workflow_draft,
+    create_draft_workspace as create_draft_workspace_record,
+    get_draft_workspace as get_draft_workspace_record,
+    patch_draft_workspace as patch_draft_workspace_record,
+    patch_workflow_draft,
+    validate_workflow_draft,
+)
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    OutputBinding,
+)
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+from wf_platform import CapabilityRef, NodeSpecInventory
+
+from .constants import (
+    DEFAULT_CALL_STEP_ID,
+    DEFAULT_ERROR_OUTCOME,
+    DEFAULT_ERROR_STEP_ID,
+    DEFAULT_OK_OUTCOME,
+    RUNTIME_ERROR_CAPABILITY,
+)
+from .operation_context import WorkflowOperationContext
+
+
+class WorkflowDraftApi:
+    """Draft validation and workspace editing operations.
+
+    This service deliberately excludes artifact persistence and capability
+    inspection. Those domains still live in the MCP-backed handler until later
+    extraction slices.
+    """
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+```
+
+- [ ] **Step 2: Export the draft service**
+
+Add `WorkflowDraftApi` to `src/wf_api/__init__.py` so future adapters can use
+the canonical import path:
+
+```python
+from .drafts import WorkflowDraftApi
+```
+
+Also add `"WorkflowDraftApi"` to `__all__`.
+
+- [ ] **Step 3: Add store helper**
+
+Add:
+
+```python
+    def _draft_store(self) -> DraftWorkspaceStore:
+        if self.context.draft_workspace_store is None:
+            raise KeyError("draft workspace store is not configured")
+        return self.context.draft_workspace_store
+```
+
+- [ ] **Step 4: Add outcome lookup helper**
+
+Add:
+
+```python
+    def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
+        try:
+            spec = self.context.specs.get_qualified_spec(qualified_name)
+        except KeyError:
+            return None
+        outcomes = getattr(spec, "outcomes", None)
+        return tuple(outcomes) if outcomes is not None else None
+```
+
+---
+
+## Task 2: Move Stateless Draft Methods
+
+**Files:**
+- Modify: `src/wf_api/drafts.py`
+
+- [ ] **Step 1: Add `validate_draft`**
+
+```python
+    async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+        return validate_workflow_draft(
+            draft,
+            outcome_lookup=self._outcomes_for_capability,
+        )
+```
+
+- [ ] **Step 2: Add `compile_draft`**
+
+```python
+    async def compile_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+        plan = compile_workflow_draft(draft)
+        return {
+            "compiled_plan": plan,
+            "required_capabilities": _required_capability_payloads(
+                _required_capabilities_for_plan(
+                    plan,
+                    source_bindings=None,
+                    context=self.context,
+                )
+            ),
+        }
+```
+
+- [ ] **Step 3: Add `patch_draft`**
+
+```python
+    async def patch_draft(
+        self,
+        *,
+        draft: dict[str, Any],
+        patch: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        return patch_workflow_draft(draft, patch)
+```
+
+---
+
+## Task 3: Move Draft Workspace Methods
+
+**Files:**
+- Modify: `src/wf_api/drafts.py`
+
+- [ ] **Step 1: Add workspace CRUD and validation methods**
+
+Move these method bodies exactly from `WorkflowSurfaceHandlers`, replacing
+`self._draft_store()` with the new `WorkflowDraftApi._draft_store()`:
+
+```text
+list_draft_workspaces
+create_draft_workspace
+get_draft_workspace
+delete_draft_workspace
+validate_draft_workspace
+patch_draft_workspace
+```
+
+Keep behavior and return payloads identical.
+
+- [ ] **Step 2: Add patch convenience methods**
+
+Move these method bodies exactly:
+
+```text
+set_draft_name
+set_draft_route
+set_step_input_map
+set_step_output_map
+```
+
+They should call `self.patch_draft_workspace(...)` inside `WorkflowDraftApi`.
+
+---
+
+## Task 4: Move Minimal Draft Bootstrap
+
+**Files:**
+- Modify: `src/wf_api/drafts.py`
+
+- [ ] **Step 1: Add `create_minimal_draft_workspace`**
+
+Move `WorkflowSurfaceHandlers.create_minimal_draft_workspace` into
+`WorkflowDraftApi` unchanged except:
+
+- use `self._outcomes_for_capability(...)`
+- use `self.create_draft_workspace(...)`
+- keep the existing comments about provider-specific error envelopes
+
+- [ ] **Step 2: Add helper functions**
+
+Move these helper functions from `handlers.py` to the bottom of `wf_api.drafts`:
+
+```text
+_draft_input_maps
+_draft_output_map
+_draft_input_bindings_payload
+_draft_output_bindings_payload
+_graph_path_payload
+_local_path_payload
+_state_path_payload
+_escape_json_pointer
+```
+
+Do not change their behavior.
+
+---
+
+## Task 5: Move Required-Capability Draft Helpers
+
+**Files:**
+- Modify: `src/wf_api/drafts.py`
+
+- [ ] **Step 1: Move `_required_capabilities_for_plan`**
+
+Move the helper from `handlers.py` and change its signature from:
+
+```python
+def _required_capabilities_for_plan(
+    plan: dict[str, Any],
+    *,
+    source_bindings: dict[str, str] | None,
+    service: WfMcpService,
+) -> dict[str, RequiredCapability]:
+```
+
+to:
+
+```python
+def _required_capabilities_for_plan(
+    plan: dict[str, Any],
+    *,
+    source_bindings: dict[str, str] | None,
+    context: WorkflowOperationContext,
+) -> dict[str, RequiredCapability]:
+```
+
+Inside it, call `_observed_node_specs(context)` instead of
+`_observed_node_specs(service)`.
+
+- [ ] **Step 2: Move `_observed_node_specs`**
+
+Change its signature from:
+
+```python
+def _observed_node_specs(service: WfMcpService) -> dict[str, NodeSpecInventory]:
+```
+
+to:
+
+```python
+def _observed_node_specs(
+    context: WorkflowOperationContext,
+) -> dict[str, NodeSpecInventory]:
+```
+
+Loop over `context.capability_sources.values()`.
+
+- [ ] **Step 3: Move `_required_capability_payloads`**
+
+Move it unchanged.
+
+---
+
+## Task 6: Wire `WorkflowSurfaceHandlers` To Delegate Draft Methods
+
+**Files:**
+- Modify: `src/wf_mcp/workflow_surface/handlers.py`
+
+- [ ] **Step 1: Add imports**
+
+Add:
+
+```python
+from wf_api.drafts import WorkflowDraftApi
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
+```
+
+- [ ] **Step 2: Instantiate draft service**
+
+In `WorkflowSurfaceHandlers.__init__`, add:
+
+```python
+self._drafts = WorkflowDraftApi(context_from_service(service))
+```
+
+- [ ] **Step 3: Replace moved method bodies with delegates**
+
+For each moved method, keep the same signature and replace the body with a call
+to `self._drafts`.
+
+Example:
+
+```python
+async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+    return await self._drafts.validate_draft(draft=draft)
+```
+
+Apply this pattern to every method in the Slice 4B move list.
+
+- [ ] **Step 4: Remove moved helper functions from `handlers.py`**
+
+Delete only helper functions that are no longer used in `handlers.py`:
+
+```text
+_draft_input_maps
+_draft_output_map
+_draft_input_bindings_payload
+_draft_output_bindings_payload
+_graph_path_payload
+_local_path_payload
+_state_path_payload
+```
+
+Keep `_escape_json_pointer` if `set_draft_route` delegation still needs no local use. Remove it only if no remaining references exist.
+
+Do not remove from `handlers.py` yet unless `rg` proves there are no remaining
+callers:
+
+```text
+_required_capabilities_for_plan
+_required_capability_payloads
+_observed_node_specs
+```
+
+Artifact creation currently still uses these helpers. It is acceptable for
+`wf_api.drafts` and `handlers.py` to each have a copy until Slice 4C moves the
+artifact/deployment methods. If the implementor can safely share them from
+`wf_api` without creating an MCP import or changing behavior, that is allowed,
+but not required for this slice.
+
+---
+
+## Task 7: Add Focused Tests
+
+**Files:**
+- Create: `tests/wf_api/test_drafts_service.py`
+
+- [ ] **Step 1: Write direct service tests**
+
+Create tests that build a `WfMcpService` through existing test helpers or
+`load_cli_context`, adapt it with `context_from_service`, then instantiate
+`WorkflowDraftApi`.
+
+Cover:
+
+- `patch_draft` applies a JSON patch.
+- `create_draft_workspace` creates a workspace.
+- `patch_draft_workspace` updates revision.
+- `validate_draft_workspace` refreshes status.
+- `create_minimal_draft_workspace` returns the same shape as before for a simple `wf.std` capability or a registered test spec.
+
+- [ ] **Step 2: Add delegation smoke test**
+
+In an existing workflow-surface draft test file or a new focused test, assert
+that `WorkflowSurfaceHandlers.validate_draft(...)` and
+`WorkflowDraftApi.validate_draft(...)` return equivalent status/diagnostics for
+the same draft.
+
+Do not assert entire dict equality; compare stable fields individually.
+
+---
+
+## Task 8: Verification
+
+- [ ] **Step 1: Run draft tests**
+
+```powershell
+uv run pytest tests/wf_api/test_drafts_service.py tests/wf_mcp/workflow_surface/test_drafts.py -q
+```
+
+Expected: all pass.
+
+- [ ] **Step 2: Run MCP schema/config tests**
+
+```powershell
+uv run pytest tests/wf_mcp/server/test_config.py tests/wf_mcp/workflow_surface -q
+```
+
+Expected: all pass.
+
+- [ ] **Step 3: Run ruff on touched files**
+
+```powershell
+uv run ruff check src/wf_api/drafts.py src/wf_mcp/workflow_surface/handlers.py tests/wf_api/test_drafts_service.py
+```
+
+Expected: all checks pass.
+
+- [ ] **Step 4: Run basedpyright on touched files**
+
+```powershell
+uv run basedpyright --level error src/wf_api/drafts.py src/wf_mcp/workflow_surface/handlers.py tests/wf_api/test_drafts_service.py
+```
+
+Expected: `0 errors`.
+
+- [ ] **Step 5: Optional full suite**
+
+```powershell
+uv run pytest -q
+```
+
+Expected: full suite passes with the project’s existing skipped/xfailed counts.
+
+---
+
+## Self-Review Checklist
+
+- `wf_api.drafts` imports no `wf_mcp`.
+- `WorkflowSurfaceHandlers` public draft method signatures are unchanged.
+- Moved methods delegate through `WorkflowDraftApi`.
+- `create_draft_workspace_from_capability` remains in handlers.
+- artifact-saving methods remain in handlers.
+- No public payload shape changed.
+- No MCP schema changed.
+- No new dependency on the whole `WfMcpService` inside `wf_api`.

src/wf_api/__init__.py

@@ -8,6 +8,7 @@ from .constants import (
     DEFAULT_OK_OUTCOME,
     RUNTIME_ERROR_CAPABILITY,
 )
+from .drafts import WorkflowDraftApi
 from .next_actions import NextActionPatchExample, NextActionTool, NextActions
 from .refs import WorkflowSurfaceCapabilityId, parse_workflow_surface_capability_id
 from .service import WorkflowApi
@@ -52,6 +53,7 @@ __all__ = [
     "WorkflowApi",
     "WorkflowApiBackend",
     "WorkflowArtifactCataloger",
+    "WorkflowDraftApi",
     "WorkflowEventRecorder",
     "WorkflowLiveSourceChecker",
     "WorkflowOperationContext",

src/wf_api/drafts.py

@@ -0,0 +1,445 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from wf_artifacts import (
+    DraftWorkspaceStore,
+    RequiredCapability,
+    WorkflowArtifact,
+    compile_workflow_draft,
+    create_draft_workspace as create_draft_workspace_record,
+    create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
+    get_draft_workspace as get_draft_workspace_record,
+    patch_draft_workspace as patch_draft_workspace_record,
+    patch_workflow_draft,
+    validate_workflow_draft,
+)
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    OutputBinding,
+)
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+from wf_platform import CapabilityRef, NodeSpecInventory
+
+from .constants import (
+    DEFAULT_CALL_STEP_ID,
+    DEFAULT_ERROR_OUTCOME,
+    DEFAULT_ERROR_STEP_ID,
+    DEFAULT_OK_OUTCOME,
+    RUNTIME_ERROR_CAPABILITY,
+)
+from .operation_context import WorkflowOperationContext
+
+
+class WorkflowDraftApi:
+    """Draft validation and workspace editing operations.
+
+    This service deliberately excludes artifact persistence and capability
+    inspection. Those domains still live in the MCP-backed handler until later
+    extraction slices.
+    """
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+
+    def _draft_store(self) -> DraftWorkspaceStore:
+        if self.context.draft_workspace_store is None:
+            raise KeyError("draft workspace store is not configured")
+        return self.context.draft_workspace_store
+
+    def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
+        try:
+            spec = self.context.specs.get_qualified_spec(qualified_name)
+        except KeyError:
+            return None
+        outcomes = getattr(spec, "outcomes", None)
+        return tuple(outcomes) if outcomes is not None else None
+
+    async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+        return validate_workflow_draft(
+            draft,
+            outcome_lookup=self._outcomes_for_capability,
+        )
+
+    async def compile_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+        plan = compile_workflow_draft(draft)
+        return {
+            "compiled_plan": plan,
+            "required_capabilities": _required_capability_payloads(
+                _required_capabilities_for_plan(
+                    plan,
+                    source_bindings=None,
+                    context=self.context,
+                )
+            ),
+        }
+
+    async def patch_draft(
+        self,
+        *,
+        draft: dict[str, Any],
+        patch: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        return patch_workflow_draft(draft, patch)
+
+    async def list_draft_workspaces(self) -> dict[str, Any]:
+        """Return compact summaries for stored draft workspaces."""
+        store = self._draft_store()
+        return {
+            "workspaces": [
+                get_draft_workspace_record(store, workspace_id=workspace.id)
+                for workspace in store.list_workspaces()
+            ]
+        }
+
+    async def create_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        draft: dict[str, Any],
+        title: str | None = None,
+    ) -> dict[str, Any]:
+        return create_draft_workspace_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            draft=draft,
+            title=title,
+        )
+
+    async def get_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        include_draft: bool = False,
+    ) -> dict[str, Any]:
+        return get_draft_workspace_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            include_draft=include_draft,
+        )
+
+    async def delete_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
+        deleted = self._draft_store().delete_workspace(workspace_id)
+        return {
+            "workspace_id": workspace_id,
+            "deleted": deleted,
+            "status": "deleted" if deleted else "not_found",
+        }
+
+    async def validate_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
+        """Refresh stored validation status without changing draft revision."""
+        store = self._draft_store()
+        workspace = store.get_workspace(workspace_id)
+        validation = await self.validate_draft(draft=workspace.draft)
+        refreshed = workspace.model_copy(
+            update={
+                "status": validation["status"],
+                "diagnostics": validation["diagnostics"],
+            }
+        )
+        store.save_workspace(refreshed)
+        return get_draft_workspace_record(store, workspace_id=workspace_id)
+
+    async def patch_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        patch: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        return patch_draft_workspace_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
+    async def set_draft_name(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        name: str,
+    ) -> dict[str, Any]:
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[{"op": "replace", "path": "/name", "value": name}],
+        )
+
+    async def set_draft_route(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        outcome: str,
+        target: str,
+    ) -> dict[str, Any]:
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {
+                    "op": "add",
+                    "path": (
+                        f"/routes/{_escape_json_pointer(step_id)}/"
+                        f"{_escape_json_pointer(outcome)}"
+                    ),
+                    "value": target,
+                }
+            ],
+        )
+
+    async def set_step_input_map(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        input_map: dict[str, str],
+    ) -> dict[str, Any]:
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {
+                    "op": "replace",
+                    "path": f"/steps/{_escape_json_pointer(step_id)}/input",
+                    "value": _draft_input_bindings_payload(input_map, {}),
+                }
+            ],
+        )
+
+    async def set_step_output_map(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        output_map: dict[str, str],
+    ) -> dict[str, Any]:
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {
+                    "op": "replace",
+                    "path": f"/steps/{_escape_json_pointer(step_id)}/output",
+                    "value": _draft_output_bindings_payload(output_map),
+                }
+            ],
+        )
+
+    async def create_minimal_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        name: str,
+        capability_name: str,
+        input_schema: dict[str, Any],
+        state_schema: dict[str, Any],
+        output_schema: dict[str, Any],
+        input: Sequence[InputBinding] | None = None,
+        output: Sequence[OutputBinding] | None = None,
+        input_map: dict[str, str] | None = None,
+        output_map: dict[str, str] | None = None,
+        error_message_source: str | GraphSourcePath | None = None,
+        title: str | None = None,
+    ) -> dict[str, Any]:
+        """Bootstrap the smallest patchable draft around one workflow capability."""
+        draft_input, draft_with = _draft_input_maps(
+            input=input,
+            input_map=input_map,
+        )
+        draft_output = _draft_output_map(output=output, output_map=output_map)
+        outcomes = self._outcomes_for_capability(capability_name) or (
+            DEFAULT_OK_OUTCOME,
+        )
+        steps: dict[str, Any] = {
+            DEFAULT_CALL_STEP_ID: {
+                "use": capability_name,
+                "input": _draft_input_bindings_payload(draft_input, draft_with),
+                "output": _draft_output_bindings_payload(draft_output),
+            }
+        }
+        routes: dict[str, dict[str, str]] = {
+            DEFAULT_CALL_STEP_ID: {DEFAULT_OK_OUTCOME: "__end__"}
+        }
+        if DEFAULT_ERROR_OUTCOME in outcomes:
+            # The bootstrapper cannot infer provider-specific error envelopes.
+            # Use a static default unless the caller explicitly supplies the
+            # state path containing a better provider error message.
+            error_input: dict[str, Any] = {
+                "target": {"root": "local", "parts": ["message"]},
+                "value": "Capability call failed",
+            }
+            if error_message_source is not None:
+                error_input = {
+                    "target": {"root": "local", "parts": ["message"]},
+                    "path": _graph_path_payload(error_message_source),
+                }
+            steps[DEFAULT_ERROR_STEP_ID] = {
+                "use": RUNTIME_ERROR_CAPABILITY,
+                "input": [error_input],
+                "output": [],
+            }
+            routes[DEFAULT_CALL_STEP_ID][DEFAULT_ERROR_OUTCOME] = DEFAULT_ERROR_STEP_ID
+            routes[DEFAULT_ERROR_STEP_ID] = {DEFAULT_OK_OUTCOME: "__end__"}
+        draft = {
+            "name": name,
+            "input_schema": input_schema,
+            "state_schema": state_schema,
+            "output_schema": output_schema,
+            "start": DEFAULT_CALL_STEP_ID,
+            "steps": steps,
+            "routes": routes,
+        }
+        return await self.create_draft_workspace(
+            workspace_id=workspace_id,
+            title=title,
+            draft=draft,
+        )
+
+
+def _required_capabilities_for_plan(
+    plan: dict[str, Any],
+    *,
+    source_bindings: dict[str, str] | None,
+    context: WorkflowOperationContext,
+) -> dict[str, RequiredCapability]:
+    """Infer a draft dependency summary without persisting an artifact."""
+    artifact = build_workflow_artifact_from_plan(
+        artifact_id="draft_preview",
+        version=1,
+        title="Draft Preview",
+        plan=plan,
+        outcomes=("completed",),
+        source_bindings=source_bindings,
+        observed_node_specs=_observed_node_specs(context),
+    )
+    requirements = artifact.required_capability_map()
+    for node in _plan_nodes(artifact):
+        raw_ref = node.get("node")
+        if not isinstance(raw_ref, str) or raw_ref in requirements:
+            continue
+        try:
+            parsed = CapabilityRef.parse(raw_ref)
+        except ValueError:
+            continue
+        requirements[raw_ref] = RequiredCapability(
+            ref=parsed,
+            kind="node_spec",
+        )
+    return requirements
+
+
+def _required_capability_payloads(
+    requirements: dict[str, RequiredCapability],
+) -> dict[str, dict[str, Any]]:
+    return {
+        name: capability.model_dump(mode="json")
+        for name, capability in sorted(requirements.items())
+    }
+
+
+def _observed_node_specs(
+    context: WorkflowOperationContext,
+) -> dict[str, NodeSpecInventory]:
+    """Project current executable specs into serializable observed contracts."""
+    observed: dict[str, NodeSpecInventory] = {}
+    for source in context.capability_sources.values():
+        inventory = source.as_inventory()
+        observed.update(
+            {detail.name: detail for detail in inventory.capabilities.node_spec_details}
+        )
+    return observed
+
+
+def _plan_nodes(artifact: WorkflowArtifact) -> list[dict[str, Any]]:
+    nodes = artifact.plan.get("nodes", [])
+    return [node for node in nodes if isinstance(node, dict)]
+
+
+def _draft_input_maps(
+    *,
+    input: Sequence[InputBinding] | None,
+    input_map: dict[str, str] | None,
+) -> tuple[dict[str, str], dict[str, Any]]:
+    """Convert canonical MCP input bindings into draft `in` and `with` maps.
+
+    Draft workspaces intentionally keep compact maps as patch targets, while
+    MCP-facing request models prefer the canonical core binding structs. This
+    helper keeps that translation explicit at the frontend boundary.
+    """
+    if input is not None and input_map is not None:
+        raise ValueError("cannot mix canonical input bindings with input_map")
+    if input is None:
+        return dict(input_map or {}), {}
+
+    mapped_inputs: dict[str, str] = {}
+    literal_inputs: dict[str, Any] = {}
+    for binding in input:
+        if isinstance(binding, InputPathBinding):
+            mapped_inputs[str(binding.path)] = str(binding.target)
+        elif isinstance(binding, InputValueBinding):
+            literal_inputs[str(binding.target)] = binding.value
+        else:  # pragma: no cover - defensive against future input binding variants.
+            raise TypeError(f"unsupported input binding {binding!r}")
+    return mapped_inputs, literal_inputs
+
+
+def _draft_output_map(
+    *,
+    output: Sequence[OutputBinding] | None,
+    output_map: dict[str, str] | None,
+) -> dict[str, str]:
+    """Convert canonical MCP output bindings into the draft `out` map."""
+    if output is not None and output_map is not None:
+        raise ValueError("cannot mix canonical output bindings with output_map")
+    if output is None:
+        return dict(output_map or {})
+    return {str(binding.source): str(binding.target) for binding in output}
+
+
+def _draft_input_bindings_payload(
+    input_map: dict[str, str],
+    input_values: dict[str, Any],
+) -> list[dict[str, Any]]:
+    """Serialize draft input maps into canonical structural binding payloads."""
+    return [
+        {"target": _local_path_payload(target), "value": value}
+        for target, value in input_values.items()
+    ] + [
+        {"target": _local_path_payload(target), "path": _graph_path_payload(source)}
+        for source, target in input_map.items()
+    ]
+
+
+def _draft_output_bindings_payload(output_map: dict[str, str]) -> list[dict[str, Any]]:
+    """Serialize draft output maps into canonical structural binding payloads."""
+    return [
+        {"source": _local_path_payload(source), "target": _state_path_payload(target)}
+        for source, target in output_map.items()
+    ]
+
+
+def _graph_path_payload(value: str | GraphSourcePath) -> dict[str, str | list[str]]:
+    path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
+    return GraphSourcePath._serialize(path)
+
+
+def _local_path_payload(value: str) -> dict[str, str | list[str]]:
+    return LocalPath._serialize(LocalPath.parse(value))
+
+
+def _state_path_payload(value: str) -> dict[str, str | list[str]]:
+    return StatePath._serialize(StatePath.parse(value))
+
+
+def _escape_json_pointer(value: str) -> str:
+    """Escape one JSON Pointer path segment for generated JSON Patch helpers."""
+    return value.replace("~", "~0").replace("/", "~1")

src/wf_api/service.py

@@ -31,7 +31,10 @@ class WorkflowApi:
         limit: int = 50,
     ) -> dict[str, Any]:
         return await self.backend.list_capabilities(
-            query=query, source_id=source_id, cursor=cursor, limit=limit,
+            query=query,
+            source_id=source_id,
+            cursor=cursor,
+            limit=limit,
         )
 
     async def inspect_capability(
@@ -49,7 +52,9 @@ class WorkflowApi:
         deployment_id: str | None = None,
     ) -> dict[str, Any]:
         return await self.backend.call_capability(
-            qualified_name=qualified_name, payload=payload, deployment_id=deployment_id,
+            qualified_name=qualified_name,
+            payload=payload,
+            deployment_id=deployment_id,
         )
 
     # -- artifacts --
@@ -63,7 +68,10 @@ class WorkflowApi:
         limit: int = 50,
     ) -> dict[str, Any]:
         return await self.backend.list_artifacts(
-            query=query, kind=kind, cursor=cursor, limit=limit,
+            query=query,
+            kind=kind,
+            cursor=cursor,
+            limit=limit,
         )
 
     async def inspect_artifact(
@@ -73,7 +81,8 @@ class WorkflowApi:
         version: int,
     ) -> dict[str, Any]:
         return await self.backend.inspect_artifact(
-            artifact_id=artifact_id, version=version,
+            artifact_id=artifact_id,
+            version=version,
         )
 
     async def save_artifact(
@@ -225,7 +234,9 @@ class WorkflowApi:
         title: str | None = None,
     ) -> dict[str, Any]:
         return await self.backend.create_draft_workspace(
-            workspace_id=workspace_id, draft=draft, title=title,
+            workspace_id=workspace_id,
+            draft=draft,
+            title=title,
         )
 
     async def get_draft_workspace(
@@ -235,7 +246,8 @@ class WorkflowApi:
         include_draft: bool = False,
     ) -> dict[str, Any]:
         return await self.backend.get_draft_workspace(
-            workspace_id=workspace_id, include_draft=include_draft,
+            workspace_id=workspace_id,
+            include_draft=include_draft,
         )
 
     async def delete_draft_workspace(
@@ -260,7 +272,9 @@ class WorkflowApi:
         patch: list[dict[str, Any]],
     ) -> dict[str, Any]:
         return await self.backend.patch_draft_workspace(
-            workspace_id=workspace_id, revision=revision, patch=patch,
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
         )
 
     async def set_draft_name(
@@ -271,7 +285,9 @@ class WorkflowApi:
         name: str,
     ) -> dict[str, Any]:
         return await self.backend.set_draft_name(
-            workspace_id=workspace_id, revision=revision, name=name,
+            workspace_id=workspace_id,
+            revision=revision,
+            name=name,
         )
 
     async def set_draft_route(
@@ -415,7 +431,8 @@ class WorkflowApi:
         live_check: bool = False,
     ) -> dict[str, Any]:
         return await self.backend.validate_deployment(
-            deployment_id=deployment_id, live_check=live_check,
+            deployment_id=deployment_id,
+            live_check=live_check,
         )
 
     # -- runs --
@@ -462,5 +479,6 @@ class WorkflowApi:
         trace_range: TraceRange,
     ) -> dict[str, Any]:
         return await self.backend.read_run_trace(
-            run_id=run_id, trace_range=trace_range,
+            run_id=run_id,
+            trace_range=trace_range,
         )

src/wf_mcp/broker/service/workflow_api_backend.py

@@ -32,7 +32,10 @@ class WfMcpWorkflowApiBackend:
         limit: int = 50,
     ) -> dict[str, Any]:
         return await self._handlers.list_capabilities(
-            query=query, source_id=source_id, cursor=cursor, limit=limit,
+            query=query,
+            source_id=source_id,
+            cursor=cursor,
+            limit=limit,
         )
 
     async def inspect_capability(
@@ -50,7 +53,9 @@ class WfMcpWorkflowApiBackend:
         deployment_id: str | None = None,
     ) -> dict[str, Any]:
         return await self._handlers.call_capability(
-            qualified_name=qualified_name, payload=payload, deployment_id=deployment_id,
+            qualified_name=qualified_name,
+            payload=payload,
+            deployment_id=deployment_id,
         )
 
     # -- artifacts --
@@ -64,7 +69,10 @@ class WfMcpWorkflowApiBackend:
         limit: int = 50,
     ) -> dict[str, Any]:
         return await self._handlers.list_artifacts(
-            query=query, kind=kind, cursor=cursor, limit=limit,
+            query=query,
+            kind=kind,
+            cursor=cursor,
+            limit=limit,
         )
 
     async def inspect_artifact(
@@ -74,7 +82,8 @@ class WfMcpWorkflowApiBackend:
         version: int,
     ) -> dict[str, Any]:
         return await self._handlers.inspect_artifact(
-            artifact_id=artifact_id, version=version,
+            artifact_id=artifact_id,
+            version=version,
         )
 
     async def save_artifact(
@@ -226,7 +235,9 @@ class WfMcpWorkflowApiBackend:
         title: str | None = None,
     ) -> dict[str, Any]:
         return await self._handlers.create_draft_workspace(
-            workspace_id=workspace_id, draft=draft, title=title,
+            workspace_id=workspace_id,
+            draft=draft,
+            title=title,
         )
 
     async def get_draft_workspace(
@@ -236,7 +247,8 @@ class WfMcpWorkflowApiBackend:
         include_draft: bool = False,
     ) -> dict[str, Any]:
         return await self._handlers.get_draft_workspace(
-            workspace_id=workspace_id, include_draft=include_draft,
+            workspace_id=workspace_id,
+            include_draft=include_draft,
         )
 
     async def delete_draft_workspace(
@@ -261,7 +273,9 @@ class WfMcpWorkflowApiBackend:
         patch: list[dict[str, Any]],
     ) -> dict[str, Any]:
         return await self._handlers.patch_draft_workspace(
-            workspace_id=workspace_id, revision=revision, patch=patch,
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
         )
 
     async def set_draft_name(
@@ -272,7 +286,9 @@ class WfMcpWorkflowApiBackend:
         name: str,
     ) -> dict[str, Any]:
         return await self._handlers.set_draft_name(
-            workspace_id=workspace_id, revision=revision, name=name,
+            workspace_id=workspace_id,
+            revision=revision,
+            name=name,
         )
 
     async def set_draft_route(
@@ -416,7 +432,8 @@ class WfMcpWorkflowApiBackend:
         live_check: bool = False,
     ) -> dict[str, Any]:
         return await self._handlers.validate_deployment(
-            deployment_id=deployment_id, live_check=live_check,
+            deployment_id=deployment_id,
+            live_check=live_check,
         )
 
     # -- runs --
@@ -431,7 +448,9 @@ class WfMcpWorkflowApiBackend:
         return await self._handlers.run_deployment(
             deployment_id=deployment_id,
             workflow_input=workflow_input,
-            trace_range=_to_handler_trace_range(trace_range) if trace_range is not None else None,
+            trace_range=_to_handler_trace_range(trace_range)
+            if trace_range is not None
+            else None,
         )
 
     async def resume_run(
@@ -446,7 +465,9 @@ class WfMcpWorkflowApiBackend:
             run_id=run_id,
             resume_payload=resume_payload,
             resume_outcome=resume_outcome,
-            trace_range=_to_handler_trace_range(trace_range) if trace_range is not None else None,
+            trace_range=_to_handler_trace_range(trace_range)
+            if trace_range is not None
+            else None,
         )
 
     async def inspect_run(
@@ -463,5 +484,6 @@ class WfMcpWorkflowApiBackend:
         trace_range: ApiTraceRange,
     ) -> dict[str, Any]:
         return await self._handlers.read_run_trace(
-            run_id=run_id, trace_range=_to_handler_trace_range(trace_range),
+            run_id=run_id,
+            trace_range=_to_handler_trace_range(trace_range),
         )

src/wf_mcp/workflow_surface/handlers.py

@@ -23,12 +23,7 @@ from wf_artifacts import (
     WorkflowCapabilityRef,
     WorkflowDeployment,
     compile_workflow_draft,
-    create_draft_workspace as create_draft_workspace_record,
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
-    get_draft_workspace as get_draft_workspace_record,
-    patch_draft_workspace as patch_draft_workspace_record,
-    patch_workflow_draft,
-    validate_workflow_draft,
     validate_deployment_dependencies,
 )
 from wf_platform import (
@@ -41,19 +36,11 @@ from wf_authoring import build_async_registry
 from wf_core import RuntimeContext
 from wf_core.models.steps import (
     InputBinding,
-    InputPathBinding,
-    InputValueBinding,
     OutputBinding,
 )
-from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+from wf_core.paths import GraphSourcePath
 
-from wf_api.constants import (
-    DEFAULT_CALL_STEP_ID,
-    DEFAULT_ERROR_OUTCOME,
-    DEFAULT_ERROR_STEP_ID,
-    DEFAULT_OK_OUTCOME,
-    RUNTIME_ERROR_CAPABILITY,
-)
+from wf_api.drafts import WorkflowDraftApi
 from wf_api.models import RawWorkflowPlan
 from wf_api.next_actions import NextActions
 from wf_api.refs import parse_workflow_surface_capability_id
@@ -70,6 +57,7 @@ from wf_api.wrapper_hints import (
 )
 
 from ..broker.service.adapters import require_adapter
+from ..broker.service.workflow_operation_context import context_from_service
 from ..events import make_event
 from ..shared import matches_query, paged_list_payload
 from .models import TraceRange
@@ -107,6 +95,7 @@ class WorkflowSurfaceHandlers:
 
     def __init__(self, service: WfMcpService) -> None:
         self.service = service
+        self._drafts = WorkflowDraftApi(context_from_service(service))
 
     async def list_artifacts(
         self,
@@ -491,23 +480,10 @@ class WorkflowSurfaceHandlers:
         }
 
     async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
-        return validate_workflow_draft(
-            draft,
-            outcome_lookup=self._outcomes_for_capability,
-        )
+        return await self._drafts.validate_draft(draft=draft)
 
     async def compile_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
-        plan = compile_workflow_draft(draft)
-        return {
-            "compiled_plan": plan,
-            "required_capabilities": _required_capability_payloads(
-                _required_capabilities_for_plan(
-                    plan,
-                    source_bindings=None,
-                    service=self.service,
-                )
-            ),
-        }
+        return await self._drafts.compile_draft(draft=draft)
 
     async def create_artifact_from_draft(
         self,
@@ -574,7 +550,7 @@ class WorkflowSurfaceHandlers:
         draft: dict[str, Any],
         patch: list[dict[str, Any]],
     ) -> dict[str, Any]:
-        return patch_workflow_draft(draft, patch)
+        return await self._drafts.patch_draft(draft=draft, patch=patch)
 
     def _draft_store(self) -> DraftWorkspaceStore:
         if self.service.draft_workspace_store is None:
@@ -583,13 +559,7 @@ class WorkflowSurfaceHandlers:
 
     async def list_draft_workspaces(self) -> dict[str, Any]:
         """Return compact summaries for stored draft workspaces."""
-        store = self._draft_store()
-        return {
-            "workspaces": [
-                get_draft_workspace_record(store, workspace_id=workspace.id)
-                for workspace in store.list_workspaces()
-            ]
-        }
+        return await self._drafts.list_draft_workspaces()
 
     async def create_draft_workspace(
         self,
@@ -598,8 +568,7 @@ class WorkflowSurfaceHandlers:
         draft: dict[str, Any],
         title: str | None = None,
     ) -> dict[str, Any]:
-        return create_draft_workspace_record(
-            self._draft_store(),
+        return await self._drafts.create_draft_workspace(
             workspace_id=workspace_id,
             draft=draft,
             title=title,
@@ -611,33 +580,17 @@ class WorkflowSurfaceHandlers:
         workspace_id: str,
         include_draft: bool = False,
     ) -> dict[str, Any]:
-        return get_draft_workspace_record(
-            self._draft_store(),
+        return await self._drafts.get_draft_workspace(
             workspace_id=workspace_id,
             include_draft=include_draft,
         )
 
     async def delete_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
-        deleted = self._draft_store().delete_workspace(workspace_id)
-        return {
-            "workspace_id": workspace_id,
-            "deleted": deleted,
-            "status": "deleted" if deleted else "not_found",
-        }
+        return await self._drafts.delete_draft_workspace(workspace_id=workspace_id)
 
     async def validate_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
         """Refresh stored validation status without changing draft revision."""
-        store = self._draft_store()
-        workspace = store.get_workspace(workspace_id)
-        validation = await self.validate_draft(draft=workspace.draft)
-        refreshed = workspace.model_copy(
-            update={
-                "status": validation["status"],
-                "diagnostics": validation["diagnostics"],
-            }
-        )
-        store.save_workspace(refreshed)
-        return get_draft_workspace_record(store, workspace_id=workspace_id)
+        return await self._drafts.validate_draft_workspace(workspace_id=workspace_id)
 
     async def patch_draft_workspace(
         self,
@@ -646,8 +599,7 @@ class WorkflowSurfaceHandlers:
         revision: int,
         patch: list[dict[str, Any]],
     ) -> dict[str, Any]:
-        return patch_draft_workspace_record(
-            self._draft_store(),
+        return await self._drafts.patch_draft_workspace(
             workspace_id=workspace_id,
             revision=revision,
             patch=patch,
@@ -660,10 +612,10 @@ class WorkflowSurfaceHandlers:
         revision: int,
         name: str,
     ) -> dict[str, Any]:
-        return await self.patch_draft_workspace(
+        return await self._drafts.set_draft_name(
             workspace_id=workspace_id,
             revision=revision,
-            patch=[{"op": "replace", "path": "/name", "value": name}],
+            name=name,
         )
 
     async def set_draft_route(
@@ -675,19 +627,12 @@ class WorkflowSurfaceHandlers:
         outcome: str,
         target: str,
     ) -> dict[str, Any]:
-        return await self.patch_draft_workspace(
+        return await self._drafts.set_draft_route(
             workspace_id=workspace_id,
             revision=revision,
-            patch=[
-                {
-                    "op": "add",
-                    "path": (
-                        f"/routes/{_escape_json_pointer(step_id)}/"
-                        f"{_escape_json_pointer(outcome)}"
-                    ),
-                    "value": target,
-                }
-            ],
+            step_id=step_id,
+            outcome=outcome,
+            target=target,
         )
 
     async def set_step_input_map(
@@ -698,16 +643,11 @@ class WorkflowSurfaceHandlers:
         step_id: str,
         input_map: dict[str, str],
     ) -> dict[str, Any]:
-        return await self.patch_draft_workspace(
+        return await self._drafts.set_step_input_map(
             workspace_id=workspace_id,
             revision=revision,
-            patch=[
-                {
-                    "op": "replace",
-                    "path": f"/steps/{_escape_json_pointer(step_id)}/input",
-                    "value": _draft_input_bindings_payload(input_map, {}),
-                }
-            ],
+            step_id=step_id,
+            input_map=input_map,
         )
 
     async def set_step_output_map(
@@ -718,16 +658,11 @@ class WorkflowSurfaceHandlers:
         step_id: str,
         output_map: dict[str, str],
     ) -> dict[str, Any]:
-        return await self.patch_draft_workspace(
+        return await self._drafts.set_step_output_map(
             workspace_id=workspace_id,
             revision=revision,
-            patch=[
-                {
-                    "op": "replace",
-                    "path": f"/steps/{_escape_json_pointer(step_id)}/output",
-                    "value": _draft_output_bindings_payload(output_map),
-                }
-            ],
+            step_id=step_id,
+            output_map=output_map,
         )
 
     async def create_minimal_draft_workspace(
@@ -747,57 +682,19 @@ class WorkflowSurfaceHandlers:
         title: str | None = None,
     ) -> dict[str, Any]:
         """Bootstrap the smallest patchable draft around one workflow capability."""
-        draft_input, draft_with = _draft_input_maps(
-            input=input,
-            input_map=input_map,
-        )
-        draft_output = _draft_output_map(output=output, output_map=output_map)
-        outcomes = self._outcomes_for_capability(capability_name) or (
-            DEFAULT_OK_OUTCOME,
-        )
-        steps: dict[str, Any] = {
-            DEFAULT_CALL_STEP_ID: {
-                "use": capability_name,
-                "input": _draft_input_bindings_payload(draft_input, draft_with),
-                "output": _draft_output_bindings_payload(draft_output),
-            }
-        }
-        routes: dict[str, dict[str, str]] = {
-            DEFAULT_CALL_STEP_ID: {DEFAULT_OK_OUTCOME: "__end__"}
-        }
-        if DEFAULT_ERROR_OUTCOME in outcomes:
-            # The bootstrapper cannot infer provider-specific error envelopes.
-            # Use a static default unless the caller explicitly supplies the
-            # state path containing a better provider error message.
-            error_input: dict[str, Any] = {
-                "target": {"root": "local", "parts": ["message"]},
-                "value": "Capability call failed",
-            }
-            if error_message_source is not None:
-                error_input = {
-                    "target": {"root": "local", "parts": ["message"]},
-                    "path": _graph_path_payload(error_message_source),
-                }
-            steps[DEFAULT_ERROR_STEP_ID] = {
-                "use": RUNTIME_ERROR_CAPABILITY,
-                "input": [error_input],
-                "output": [],
-            }
-            routes[DEFAULT_CALL_STEP_ID][DEFAULT_ERROR_OUTCOME] = DEFAULT_ERROR_STEP_ID
-            routes[DEFAULT_ERROR_STEP_ID] = {DEFAULT_OK_OUTCOME: "__end__"}
-        draft = {
-            "name": name,
-            "input_schema": input_schema,
-            "state_schema": state_schema,
-            "output_schema": output_schema,
-            "start": DEFAULT_CALL_STEP_ID,
-            "steps": steps,
-            "routes": routes,
-        }
-        return await self.create_draft_workspace(
+        return await self._drafts.create_minimal_draft_workspace(
             workspace_id=workspace_id,
+            name=name,
+            capability_name=capability_name,
+            input_schema=input_schema,
+            state_schema=state_schema,
+            output_schema=output_schema,
+            input=input,
+            output=output,
+            input_map=input_map,
+            output_map=output_map,
+            error_message_source=error_message_source,
             title=title,
-            draft=draft,
         )
 
     async def create_draft_workspace_from_capability(
@@ -909,12 +806,6 @@ class WorkflowSurfaceHandlers:
             created_from_catalog_version=created_from_catalog_version,
         )
 
-    def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
-        try:
-            return self.service._get_qualified_spec(qualified_name).outcomes
-        except KeyError:
-            return None
-
     async def inspect_artifact(
         self, *, artifact_id: str, version: int
     ) -> dict[str, Any]:
@@ -1406,87 +1297,6 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
-def _draft_input_maps(
-    *,
-    input: Sequence[InputBinding] | None,
-    input_map: dict[str, str] | None,
-) -> tuple[dict[str, str], dict[str, Any]]:
-    """Convert canonical MCP input bindings into draft `in` and `with` maps.
-
-    Draft workspaces intentionally keep compact maps as patch targets, while
-    MCP-facing request models prefer the canonical core binding structs. This
-    helper keeps that translation explicit at the frontend boundary.
-    """
-    if input is not None and input_map is not None:
-        raise ValueError("cannot mix canonical input bindings with input_map")
-    if input is None:
-        return dict(input_map or {}), {}
-
-    mapped_inputs: dict[str, str] = {}
-    literal_inputs: dict[str, Any] = {}
-    for binding in input:
-        if isinstance(binding, InputPathBinding):
-            mapped_inputs[str(binding.path)] = str(binding.target)
-        elif isinstance(binding, InputValueBinding):
-            literal_inputs[str(binding.target)] = binding.value
-        else:  # pragma: no cover - defensive against future input binding variants.
-            raise TypeError(f"unsupported input binding {binding!r}")
-    return mapped_inputs, literal_inputs
-
-
-def _draft_output_map(
-    *,
-    output: Sequence[OutputBinding] | None,
-    output_map: dict[str, str] | None,
-) -> dict[str, str]:
-    """Convert canonical MCP output bindings into the draft `out` map."""
-    if output is not None and output_map is not None:
-        raise ValueError("cannot mix canonical output bindings with output_map")
-    if output is None:
-        return dict(output_map or {})
-    return {str(binding.source): str(binding.target) for binding in output}
-
-
-def _draft_input_bindings_payload(
-    input_map: dict[str, str],
-    input_values: dict[str, Any],
-) -> list[dict[str, Any]]:
-    """Serialize draft input maps into canonical structural binding payloads."""
-    return [
-        {"target": _local_path_payload(target), "value": value}
-        for target, value in input_values.items()
-    ] + [
-        {"target": _local_path_payload(target), "path": _graph_path_payload(source)}
-        for source, target in input_map.items()
-    ]
-
-
-def _draft_output_bindings_payload(output_map: dict[str, str]) -> list[dict[str, Any]]:
-    """Serialize draft output maps into canonical structural binding payloads."""
-    return [
-        {"source": _local_path_payload(source), "target": _state_path_payload(target)}
-        for source, target in output_map.items()
-    ]
-
-
-def _graph_path_payload(value: str | GraphSourcePath) -> dict[str, str | list[str]]:
-    path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
-    return GraphSourcePath._serialize(path)
-
-
-def _local_path_payload(value: str) -> dict[str, str | list[str]]:
-    return LocalPath._serialize(LocalPath.parse(value))
-
-
-def _state_path_payload(value: str) -> dict[str, str | list[str]]:
-    return StatePath._serialize(StatePath.parse(value))
-
-
-def _escape_json_pointer(value: str) -> str:
-    """Escape one JSON Pointer path segment for generated JSON Patch helpers."""
-    return value.replace("~", "~0").replace("/", "~1")
-
-
 def _draft_name_from_capability(capability_name: str) -> str:
     """Return a stable draft name when caller does not provide one."""
     return capability_name.replace(".", "_").replace("-", "_")

src/wf_mcp/workflow_surface/models.py

@@ -399,7 +399,9 @@ class RunDeploymentResult(BaseModel):
     deployment_id: str = Field(description="Deployment that was run.")
     artifact_id: str = Field(description="Artifact targeted by the deployment.")
     artifact_version: int = Field(description="Artifact version.")
-    status: str = Field(description="Run status, such as completed, failed, or interrupted.")
+    status: str = Field(
+        description="Run status, such as completed, failed, or interrupted."
+    )
     run_id: str | None = Field(default=None, description="Durable run identifier.")
     resume_readiness: str | None = Field(
         default=None, description="Resume readiness state."

src/wf_mcp/workflow_surface/run_lifecycle.py

@@ -5,6 +5,7 @@ This module re-exports every public symbol so that existing
 continue to work unchanged. New code should import from ``wf_api.run_lifecycle``
 directly.
 """
+
 from __future__ import annotations
 
 from wf_api.run_lifecycle import (

src/wf_mcp/workflow_surface/tools.py

@@ -669,7 +669,9 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             await handlers.run_deployment(
                 deployment_id=deployment_id,
                 workflow_input=workflow_input,
-                trace_range=_to_api_trace_range(trace_range) if trace_range is not None else None,
+                trace_range=_to_api_trace_range(trace_range)
+                if trace_range is not None
+                else None,
             )
         )
 
@@ -701,7 +703,9 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 run_id=run_id,
                 resume_payload=resume_payload,
                 resume_outcome=resume_outcome,
-                trace_range=_to_api_trace_range(trace_range) if trace_range is not None else None,
+                trace_range=_to_api_trace_range(trace_range)
+                if trace_range is not None
+                else None,
             )
         )
 

tests/wf_api/test_drafts_service.py

@@ -0,0 +1,221 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from wf_artifacts import FileWorkflowArtifactStore
+from wf_api.drafts import WorkflowDraftApi
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
+
+from tests.wf_mcp.test_support import echo_tool, local_temp_root
+
+
+def _echo_draft() -> dict[str, Any]:
+    return {
+        "name": "echo",
+        "input_schema": {
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        "state_schema": {"fields": {"echoed": {"type": "string"}}},
+        "output_schema": {
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        "start": "echo",
+        "steps": {
+            "echo": {
+                "use": "demo.personal.echo_tool",
+                "input": [
+                    {
+                        "target": {"root": "local", "parts": ["text"]},
+                        "path": {"root": "input", "parts": ["text"]},
+                    }
+                ],
+                "output": [
+                    {
+                        "source": {"root": "local", "parts": ["echoed"]},
+                        "target": {"root": "state", "parts": ["echoed"]},
+                    }
+                ],
+            }
+        },
+        "routes": {"echo": {"ok": "__end__"}},
+    }
+
+
+def _draft_api(
+    artifact_store: FileWorkflowArtifactStore,
+    *,
+    register_echo: bool = False,
+) -> tuple[WorkflowDraftApi, WfMcpService]:
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "drafts_mcp" / str(id(artifact_store))),
+        artifact_store=artifact_store,
+    )
+    if register_echo:
+        service.register_connection(
+            ConnectionConfig(id="demo.personal", server="demo", account="personal")
+        )
+        service.register_specs("demo.personal", echo_tool)
+    context = context_from_service(service)
+    return WorkflowDraftApi(context), service
+
+
+def test_patch_draft_applies_json_patch() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "drafts_patch")
+    api, _service = _draft_api(artifact_store)
+
+    result = asyncio.run(
+        api.patch_draft(
+            draft=_echo_draft(),
+            patch=[
+                {
+                    "op": "replace",
+                    "path": "/steps/echo/input/0/target/parts/0",
+                    "value": "message",
+                }
+            ],
+        )
+    )
+
+    assert result["status"] == "valid"
+    assert result["draft"]["steps"]["echo"]["input"][0]["target"] == {
+        "root": "local",
+        "parts": ["message"],
+    }
+
+
+def test_create_draft_workspace_creates_workspace() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "drafts_create_workspace"
+    )
+    api, _service = _draft_api(artifact_store)
+
+    result = asyncio.run(
+        api.create_draft_workspace(
+            workspace_id="echo_ws",
+            title="Echo Workspace",
+            draft=_echo_draft(),
+        )
+    )
+
+    assert result["workspace_id"] == "echo_ws"
+    assert result["revision"] == 1
+
+
+def test_patch_draft_workspace_updates_revision() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "drafts_patch_workspace"
+    )
+    api, _service = _draft_api(artifact_store)
+    asyncio.run(
+        api.create_draft_workspace(
+            workspace_id="echo_ws",
+            draft=_echo_draft(),
+        )
+    )
+
+    patched = asyncio.run(
+        api.patch_draft_workspace(
+            workspace_id="echo_ws",
+            revision=1,
+            patch=[{"op": "replace", "path": "/name", "value": "echo_v2"}],
+        )
+    )
+
+    assert patched["revision"] == 2
+    assert patched["status"] == "valid"
+
+
+def test_validate_draft_workspace_refreshes_status() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "drafts_validate_workspace"
+    )
+    api, service = _draft_api(artifact_store, register_echo=True)
+    draft = _echo_draft()
+    draft["routes"]["echo"] = {"typo": "__end__"}
+    asyncio.run(
+        api.create_draft_workspace(
+            workspace_id="echo_ws",
+            draft=draft,
+        )
+    )
+
+    payload = asyncio.run(api.validate_draft_workspace(workspace_id="echo_ws"))
+    fetched = asyncio.run(api.get_draft_workspace(workspace_id="echo_ws"))
+
+    assert payload["revision"] == 1
+    assert payload["status"] == "invalid"
+    assert payload["diagnostics"][0]["code"] == "unknown_outcome"
+    assert fetched["status"] == "invalid"
+
+
+def test_create_minimal_draft_workspace_with_error_route() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "drafts_minimal_workspace"
+    )
+    api, _service = _draft_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(
+        api.create_minimal_draft_workspace(
+            workspace_id="echo_minimal",
+            name="echo",
+            capability_name="demo.personal.echo_tool",
+            input_schema={
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            },
+            state_schema={"fields": {"echoed": {"type": "string"}}},
+            output_schema={
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+                "required": ["echoed"],
+            },
+            input_map={"input.text": "text"},
+            output_map={"echoed": "state.echoed"},
+        )
+    )
+
+    assert result["workspace_id"] == "echo_minimal"
+    fetched = asyncio.run(
+        api.get_draft_workspace(workspace_id="echo_minimal", include_draft=True)
+    )
+    assert fetched["draft"]["routes"]["call"]["ok"] == "__end__"
+    assert fetched["draft"]["steps"]["call"]["use"] == "demo.personal.echo_tool"
+
+
+def test_delegation_smoke_validate_draft_equivalence() -> None:
+    """WorkflowSurfaceHandlers.validate_draft delegates to WorkflowDraftApi."""
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "drafts_delegation_smoke"
+    )
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "delegation_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    h = WorkflowSurfaceHandlers(service)
+    context = context_from_service(service)
+    api = WorkflowDraftApi(context)
+    draft = _echo_draft()
+
+    handler_result = asyncio.run(h.validate_draft(draft=draft))
+    api_result = asyncio.run(api.validate_draft(draft=draft))
+
+    assert handler_result["status"] == api_result["status"]
+    assert handler_result["diagnostics"] == api_result["diagnostics"]
+    assert (
+        handler_result["compiled_plan"]["nodes"] == api_result["compiled_plan"]["nodes"]
+    )

tests/wf_api/test_import_direction.py

@@ -15,14 +15,18 @@ def test_wf_api_has_no_wf_mcp_imports() -> None:
         tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
         for node in ast.walk(tree):
             if isinstance(node, ast.ImportFrom) and node.module is not None:
-                if node.module.startswith("wf_mcp") or node.module.startswith("wf_mcp."):
+                if node.module.startswith("wf_mcp") or node.module.startswith(
+                    "wf_mcp."
+                ):
                     violations.append(
                         f"{module}:{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"{module}:{node.lineno}: import {alias.name}")
+                        violations.append(
+                            f"{module}:{node.lineno}: import {alias.name}"
+                        )
 
     assert violations == [], (
         "wf_api imports wf_mcp — this breaks the dependency direction rule:\n"

tests/wf_api/test_operation_context.py

@@ -10,7 +10,9 @@ from wf_mcp.broker.service.workflow_operation_context import context_from_servic
 
 
 def test_wf_api_operation_context_imports_no_wf_mcp() -> None:
-    path = Path(__file__).resolve().parents[2] / "src" / "wf_api" / "operation_context.py"
+    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] = []
@@ -49,9 +51,14 @@ def test_context_from_service_exposes_existing_store_objects(tmp_path: Path) ->
 
     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.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
+    assert (
+        operation_context.capability_sources is cli_context.service.capability_sources
+    )
 
 
 def test_context_from_service_delegates_specs_and_events(tmp_path: Path) -> None:

tests/wf_mcp/test_workflow_surface_refs.py

@@ -52,7 +52,10 @@ def test_workflow_surface_capability_id_parses_structural_saved_wrapper_ref() ->
 
 
 def test_workflow_surface_refs_shim_reexports_canonical_parser() -> None:
-    assert parse_workflow_surface_capability_id_shim is parse_workflow_surface_capability_id
+    assert (
+        parse_workflow_surface_capability_id_shim
+        is parse_workflow_surface_capability_id
+    )
 
 
 def test_workflow_surface_constants_shim_reexports_canonical_literals() -> None:
@@ -62,4 +65,3 @@ def test_workflow_surface_constants_shim_reexports_canonical_literals() -> None:
     )
 
     assert DEFAULT_CALL_STEP_ID_SHIM == DEFAULT_CALL_STEP_ID
-