feat: add semantic draft authoring operations

docs/current_roadmap.md

@@ -63,16 +63,17 @@ clear operator feedback before adding more architecture.
 - Completed: `wf schema` now lists workflow document/component models, emits
   compact JSON outlines for agent discovery, and emits valid self-contained
   JSON Schema with `--verbose`.
-- Completed: `wf draft add-state-from-output` projects capability output
-  property schemas into draft state schemas, preserving `$defs` / `definitions`
-  for schema refs and reducing brittle whole-`state_schema` patches.
 - Completed: `wf draft bind-output-to-state` composes state schema projection
   with output binding merge, reducing manual draft patch repairs in agent
   challenge runs.
 - Completed: `wf draft add-step-from-capability` inserts one explicit
   capability-backed step with route, input, and output-to-state schema/binding
   wiring in a single revision, reducing brittle JSON Patch authoring for
-  multi-step workflows.
+  multi-step workflows. Accepts `--route OUTCOME=TARGET` for multi-outcome steps.
+- Completed: `wf draft branch` and `wf draft handle` provide atomic route
+  editing for existing draft steps without rewriting the full routes object.
+- Completed: `wf draft compile` returns the compiled raw plan plus required
+  capabilities without mutating or saving the draft workspace.
 - Completed: draft validation now preserves structured core validation issues
   and adds exact `bind-output-to-state` repair hints for missing state fields.
 - Keep status read-only; do not mutate registry, auth, config, or stores.

docs/wf_cli.md

@@ -297,7 +297,9 @@ wf draft set-route concat_ws --revision 2 --step call --outcome ok --to __end__
 wf draft set-input concat_ws --revision 3 --step call --map input.items=items --map input.separator=separator
 wf draft set-output concat_ws --revision 4 --step call --map value=state.value
 wf draft set-input concat_ws --revision 5 --step call --merge --map input.limit=limit
-wf draft add-state-from-output concat_ws --revision 5 --step call --output value --state state.value
+wf draft branch concat_ws --revision 6 --step call --route ok=__end__ --route error=tool_error
+wf draft handle concat_ws --revision 7 --to fail --branch lookup:error --branch transform:error
+wf draft compile concat_ws
 ```
 
 `set-input` maps graph source paths to node-local input fields:
@@ -311,13 +313,6 @@ Use repeated `--map` flags in one command when you know the complete map. Use
 `--merge` when adding or updating one entry across a later revision while
 preserving existing bindings.
 
-Before mapping a step output to a new state field, the state schema must declare
-that root field. `add-state-from-output` copies the selected step capability's
-top-level output property schema into `state_schema.properties`, including local
-`$defs` / `definitions` blocks needed by `$ref` schemas. It only declares the
-state field; still run `set-output` or `draft patch` to write values into that
-field, then run `wf draft validate`.
-
 ### Bind A Step Output To State
 
 Use `bind-output-to-state` when a step output should become workflow state and
@@ -351,8 +346,8 @@ wf draft add-step-from-capability report_ws \
   --capability local.report.render_markdown_report \
   --from-step extract \
   --from-outcome ok \
-  --outcome ok \
-  --to __end__ \
+  --route ok=__end__ \
+  --route error=tool_error \
   --input state.title=title \
   --bind-output markdown=state.markdown
 ```
@@ -360,6 +355,32 @@ wf draft add-step-from-capability report_ws \
 Run `wf draft validate report_ws` after adding the step. If validation returns
 a `repair_hint`, prefer the focused helper in that hint before JSON Patch.
 
+### Branch And Handle Existing Steps
+
+Use `wf draft branch` to update routes for an existing step in one revision:
+
+```bash
+wf draft branch concat_ws --revision 6 --step call --route ok=__end__ --route error=tool_error
+```
+
+Use `wf draft handle` to route multiple source step outcomes to a common target:
+
+```bash
+wf draft handle concat_ws --revision 7 --to fail --branch lookup:error --branch transform:error
+```
+
+### Compile A Draft Workspace
+
+Use `wf draft compile` to return the compiled raw plan plus required
+capabilities without mutating or saving the draft:
+
+```bash
+wf draft compile concat_ws
+```
+
+On success, prints the compiled plan to stdout. On invalid draft status, prints
+the structured diagnostic envelope to stderr and exits nonzero.
+
 Validate:
 
 ```bash

docs/workflow_drafts.md

@@ -614,6 +614,9 @@ The workflow MCP surface exposes these draft tools:
 | `wf.workflow.compile_draft` | Return the compiled raw plan plus dependency summaries. |
 | `wf.workflow.patch_draft` | Apply RFC 6902 JSON Patch and validate the patched draft. |
 | `wf.workflow.create_artifact_from_draft` | Compile, normalize, and save a workflow artifact. |
+| `wf.workflow.branch_draft` | Update routes for an existing step in one revision. |
+| `wf.workflow.handle_draft` | Route multiple source step outcomes to a common target. |
+| `wf.workflow.compile_draft_workspace` | Return compiled plan plus capabilities without mutation. |
 
 Use `validate_draft` before saving. Use `patch_draft` when an LLM client needs a
 small targeted correction instead of rewriting the whole workflow.
@@ -665,6 +668,9 @@ wf draft set-name <workspace_id> --revision <n> --name <name>
 wf draft set-route <workspace_id> --revision <n> --step <step_id> --outcome ok --to <target_step_or___end__>
 wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.text=text
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
+wf draft branch <workspace_id> --revision <n> --step <step_id> --route ok=__end__ --route error=tool_error
+wf draft handle <workspace_id> --revision <n> --to fail --branch lookup:error --branch transform:error
+wf draft compile <workspace_id>
 ```
 
 Use `draft patch` when these focused commands do not cover the structural edit.

skills/wf-cli/SKILL.md

@@ -44,9 +44,11 @@ wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.te
 wf draft set-input <workspace_id> --revision <n> --step <step_id> --merge --map input.other=other
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
-wf draft add-state-from-output <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft branch <workspace_id> --revision <n> --step <step_id> --route ok=__end__ --route error=fail
+wf draft handle <workspace_id> --revision <n> --to fail --branch lookup:error --branch transform:error
+wf draft compile <workspace_id>
 wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
-wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --outcome ok --to <next-or-__end__> --input input.text=text --bind-output result=state.result
+wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --route ok=__end__ --route error=fail --input input.text=text --bind-output result=state.result
 wf draft validate <workspace_id>
 wf draft save <workspace_id> --artifact <artifact_id> --version <n> --title <title>
 
@@ -80,14 +82,9 @@ For `draft set-input` and `draft set-output`, repeated `--map` flags in one
 command define the complete replacement map. If you split map edits across
 multiple commands, pass `--merge` or the later command replaces the earlier map.
 
-If mapping `LOCAL_SOURCE=state.new_field`, declare the state field first with
-`draft add-state-from-output` when the schema should match a capability output
-field. Do not hand-copy `$defs` unless the helper cannot express the shape.
-
 Prefer `draft bind-output-to-state` when a step output should write to a new
 root state field. It declares the matching state schema and merges the output
-binding in one revision-checked edit. Use `draft add-state-from-output` only
-when you need the schema declaration without changing bindings.
+binding in one revision-checked edit.
 `bind-output-to-state` requires a capability-backed step with `use`; use JSON
 Patch for non-capability/control draft steps.
 

skills/wf-workflow/references/draft-workspaces.md

@@ -73,9 +73,11 @@ Prefer focused helpers over JSON Patch for common edits:
 - `set_draft_route`
 - `set_step_input_map`
 - `set_step_output_map`
-- `add_state_schema_from_output`
 - `bind_output_to_state`
 - `add_step_from_capability`
+- `branch_draft`
+- `handle_draft`
+- `compile_draft_workspace`
 
 CLI equivalents:
 
@@ -86,9 +88,11 @@ wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.te
 wf draft set-input <workspace_id> --revision <n> --step <step_id> --merge --map input.other=other
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
-wf draft add-state-from-output <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft branch <workspace_id> --revision <n> --step <step_id> --route ok=__end__ --route error=fail
+wf draft handle <workspace_id> --revision <n> --to fail --branch lookup:error --branch transform:error
+wf draft compile <workspace_id>
 wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
-wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --outcome ok --to <next-or-__end__> --input input.text=text --bind-output result=state.result
+wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --route ok=__end__ --route error=fail --input input.text=text --bind-output result=state.result
 ```
 
 `set-input` direction: `input.text=text` means graph source `input.text` maps to
@@ -101,10 +105,6 @@ Without `--merge`, `set-input` and `set-output` replace the whole map for that
 step. Use repeated `--map` flags in one command for a complete replacement. Use
 `--merge` only when adding/updating entries over multiple revisions.
 
-Use `add-state-from-output` when the target state field should reuse a capability
-output schema. This prevents dangling `$ref` values by copying local `$defs` /
-`definitions` with the selected property schema.
-
 - `bind_output_to_state`
 
   Declares one root state field from a step capability output schema and merges
@@ -124,15 +124,35 @@ wf draft validate <workspace_id>
 
   Adds a new capability-backed step with explicit route, input bindings, and
   output-to-state schema/binding wiring in one revision. It can set the incoming
-  edge, outgoing edge, input map, and output-to-state schema/binding. It still
+  edge, outgoing edges, input map, and output-to-state schema/binding. Use
+  `--route OUTCOME=TARGET` for each outcome; when omitted and the capability
+  declares a single outcome, that outcome routes to `__end__`. It still
   requires explicit choices; if you do not know a map, inspect the capability or
   run validation rather than guessing.
 
 ```bash
-wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --outcome ok --to <next-or-__end__> --input input.text=text --bind-output result=state.result
+wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --route ok=__end__ --route error=fail --input input.text=text --bind-output result=state.result
 wf draft validate <workspace_id>
 ```
 
+- `branch_draft`
+
+  Updates routes for an existing step in one revision without rewriting the
+  full routes object. Supply `--route OUTCOME=TARGET` for each outcome to
+  set or update.
+
+- `handle_draft`
+
+  Routes multiple source step outcomes to a common target. Supply
+  `--branch STEP:OUTCOME` for each source outcome and `--to TARGET` for the
+  shared destination.
+
+- `compile_draft_workspace`
+
+  Returns the compiled raw plan plus required capabilities without mutating
+  or saving the draft workspace. On invalid draft status, returns structured
+  diagnostics without a `compiled_plan`.
+
 Validation repair hints are product guidance. If a diagnostic suggests
 `bind-output-to-state`, use it before hand-editing `state_schema` or step output
 bindings.

skills/wf-workflow/references/workflow-lifecycle.md

@@ -18,11 +18,10 @@ validated, runnable deployment.
      for common edits.
    - `set-input` and `set-output` replace full maps by default; pass `--merge`
      only when adding or updating one entry across a later revision.
-   - Before output-mapping into a new state field, declare it with
-     `add-state-from-output` when it should mirror a capability output property.
-   - Use `bind-output-to-state` when a capability output should become state;
-     it declares the matching state schema and merges the output binding in one
-     revision-checked edit.
+    - Before output-mapping into a new state field, declare it with
+      `bind-output-to-state` when it should mirror a capability output property.
+      It declares the matching state schema and merges the output binding in one
+      revision-checked edit.
    - When adding a new capability-backed step, prefer:
      ```bash
      wf draft add-step-from-capability ...

src/wf_api/capabilities.py

@@ -18,6 +18,7 @@ from wf_platform import CapabilitySource
 from .artifact_plans import raw_plan_from_artifact
 from .artifact_refs import artifact_capability_id
 from .capability_requirements import required_capability_payloads
+from .draft_authoring import WorkflowDraftAuthoringApi
 from .drafts import WorkflowDraftApi
 from .listing import matches_query, paged_list_payload
 from .next_actions import NextActions
@@ -64,6 +65,7 @@ class WorkflowCapabilityApi:
     def __init__(self, context: WorkflowOperationContext) -> None:
         self.context = context
         self.drafts = WorkflowDraftApi(context)
+        self.draft_authoring = WorkflowDraftAuthoringApi(context, self.drafts)
 
     async def list_capabilities(
         self,
@@ -353,7 +355,7 @@ class WorkflowCapabilityApi:
         """Create a patchable draft workspace from inspect_capability hints."""
         capability = await self.inspect_capability(qualified_name=capability_name)
         hints = capability["wrapper_hints"]
-        result = await self.drafts.create_minimal_draft_workspace(
+        result = await self.draft_authoring.create_minimal_draft_workspace(
             workspace_id=workspace_id,
             name=name or _draft_name_from_capability(capability_name),
             capability_name=capability_name,

src/wf_api/draft_authoring.py

@@ -0,0 +1,392 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from wf_core.models.steps import (
+    InputBinding,
+    OutputBinding,
+)
+from wf_core.paths import GraphSourcePath
+
+from .constants import (
+    DEFAULT_CALL_STEP_ID,
+    DEFAULT_ERROR_OUTCOME,
+    DEFAULT_ERROR_STEP_ID,
+    DEFAULT_OK_OUTCOME,
+    RUNTIME_ERROR_CAPABILITY,
+)
+from .draft_payloads import (
+    _graph_path_payload,
+    draft_step,
+    escape_json_pointer,
+    input_bindings_payload,
+    output_bindings_payload,
+    state_root_field,
+)
+from .drafts import (
+    WorkflowDraftApi,
+    _draft_input_maps,
+    _draft_output_map,
+)
+from .operation_context import WorkflowOperationContext
+from .schema_projection import project_output_property_to_state_schema
+
+
+class WorkflowDraftAuthoringApi:
+    """Capability-aware semantic edits over revisioned workflow drafts."""
+
+    def __init__(
+        self,
+        context: WorkflowOperationContext,
+        drafts: WorkflowDraftApi,
+    ) -> None:
+        self.context = context
+        self.drafts = drafts
+
+    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 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": input_bindings_payload(draft_input, draft_with),
+                "output": 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:
+            error_input: dict[str, Any] = {
+                "target": "message",
+                "value": "Capability call failed",
+            }
+            if error_message_source is not None:
+                error_input = {
+                    "target": "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.drafts.create_draft_workspace(
+            workspace_id=workspace_id,
+            title=title,
+            draft=draft,
+        )
+
+    async def bind_output_to_state(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        output_field: str,
+        state_path: str,
+    ) -> dict[str, Any]:
+        """Declare a state field from a step output and bind that output to it.
+
+        This is the common draft-authoring repair for validation errors where a
+        step writes to ``state.x`` before ``state_schema.properties.x`` exists.
+        It deliberately edits only one root state field and one step output map.
+        Route changes remain explicit through ``set_draft_route``.
+        """
+        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        step = draft_step(workspace.draft, step_id)
+        capability_name = step.get("use")
+        if not isinstance(capability_name, str) or not capability_name:
+            raise ValueError(
+                f"draft step {step_id!r} does not declare a capability use"
+            )
+
+        state_field = state_root_field(state_path)
+        spec = self.context.specs.get_qualified_spec(capability_name)
+        output_schema = (
+            spec.output_schema_contract or spec.output_model.model_json_schema()
+        )
+        state_schema = workspace.draft.get("state_schema", {})
+        if not isinstance(state_schema, dict):
+            raise ValueError("draft state_schema must be an object")
+        projected = project_output_property_to_state_schema(
+            state_schema=state_schema,
+            output_schema=output_schema,
+            output_field=output_field,
+            state_field=state_field,
+        )
+        output_map = {
+            **self.drafts._step_output_map(workspace_id=workspace_id, step_id=step_id),
+            output_field: state_path,
+        }
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {
+                    "op": "replace",
+                    "path": "/state_schema",
+                    "value": projected,
+                },
+                {
+                    "op": "replace",
+                    "path": f"/steps/{escape_json_pointer(step_id)}/output",
+                    "value": output_bindings_payload(output_map),
+                },
+            ],
+        )
+
+    async def add_step_from_capability(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        capability_name: str,
+        route_from_step: str | None = None,
+        route_from_outcome: str = DEFAULT_OK_OUTCOME,
+        routes: dict[str, str] | None = None,
+        input_map: dict[str, str] | None = None,
+        bind_outputs: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        """Add one capability step plus explicit route/map/schema wiring.
+
+        This is a composed authoring helper for agents.  It edits the draft in
+        one revision so callers do not have to interleave add-step, route,
+        input-map, state-schema, and output-map operations by hand.
+        """
+        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        steps = workspace.draft.get("steps")
+        if not isinstance(steps, dict):
+            raise ValueError("draft steps must be an object")
+        if step_id in steps:
+            raise ValueError(f"draft step {step_id!r} already exists")
+
+        spec = self.context.specs.get_qualified_spec(capability_name)
+        output_schema = (
+            spec.output_schema_contract or spec.output_model.model_json_schema()
+        )
+        state_schema = workspace.draft.get("state_schema", {})
+        if not isinstance(state_schema, dict):
+            raise ValueError("draft state_schema must be an object")
+
+        declared_outcomes = self._outcomes_for_capability(capability_name)
+        if declared_outcomes is None:
+            declared_outcomes = (DEFAULT_OK_OUTCOME,)
+
+        if routes is not None:
+            missing_outcomes = set(declared_outcomes) - set(routes.keys())
+            unknown_outcomes = set(routes.keys()) - set(declared_outcomes)
+            if missing_outcomes or unknown_outcomes:
+                details = [
+                    f"declared_outcomes={declared_outcomes!r}",
+                    f"missing_outcomes={sorted(missing_outcomes)!r}",
+                    f"unknown_outcomes={sorted(unknown_outcomes)!r}",
+                ]
+                raise ValueError(
+                    f"capability {capability_name!r} declares outcomes "
+                    f"{declared_outcomes}, but routes has "
+                    f"missing routes {sorted(missing_outcomes)} and unknown "
+                    f"routes {sorted(unknown_outcomes)}; " + ", ".join(details)
+                )
+            step_routes = dict(routes)
+        else:
+            if len(declared_outcomes) == 1:
+                step_routes = {declared_outcomes[0]: "__end__"}
+            else:
+                missing_outcomes = sorted(declared_outcomes)
+                raise ValueError(
+                    f"capability {capability_name!r} declares outcomes "
+                    f"{declared_outcomes} with no routes supplied; missing "
+                    f"routes for {missing_outcomes}"
+                )
+
+        input_map = input_map or {}
+        bind_outputs = bind_outputs or {}
+        projected_state_schema = state_schema
+        for output_field, path in bind_outputs.items():
+            sf = state_root_field(path)
+            projected_state_schema = project_output_property_to_state_schema(
+                state_schema=projected_state_schema,
+                output_schema=output_schema,
+                output_field=output_field,
+                state_field=sf,
+            )
+
+        patch: list[dict[str, Any]] = [
+            {
+                "op": "add",
+                "path": f"/steps/{escape_json_pointer(step_id)}",
+                "value": {
+                    "use": capability_name,
+                    "input": input_bindings_payload(input_map, {}),
+                    "output": output_bindings_payload(bind_outputs),
+                },
+            },
+            {
+                "op": "add",
+                "path": f"/routes/{escape_json_pointer(step_id)}",
+                "value": step_routes,
+            },
+        ]
+        if projected_state_schema != state_schema:
+            patch.insert(
+                0,
+                {
+                    "op": "replace",
+                    "path": "/state_schema",
+                    "value": projected_state_schema,
+                },
+            )
+        if route_from_step is not None:
+            patch.append(
+                {
+                    "op": "add",
+                    "path": (
+                        f"/routes/{escape_json_pointer(route_from_step)}/"
+                        f"{escape_json_pointer(route_from_outcome)}"
+                    ),
+                    "value": step_id,
+                }
+            )
+
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
+    async def branch_draft(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        routes: dict[str, str],
+    ) -> dict[str, Any]:
+        """Atomically set routes for one step, preserving unspecified outcomes."""
+        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        draft_routes = workspace.draft.get("routes", {})
+        if not isinstance(draft_routes, dict):
+            raise ValueError("draft routes must be an object")
+        existing = draft_routes.get(step_id, {})
+        if not isinstance(existing, dict):
+            raise ValueError(f"routes for step {step_id!r} must be an object")
+        merged = {**existing, **routes}
+        if merged == existing:
+            return await self.drafts.get_draft_workspace(
+                workspace_id=workspace_id,
+            )
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {
+                    "op": "replace",
+                    "path": f"/routes/{escape_json_pointer(step_id)}",
+                    "value": merged,
+                }
+            ],
+        )
+
+    async def handle_draft(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        branches: Sequence[DraftOutcomeRef],
+        target: str,
+    ) -> dict[str, Any]:
+        """Update the target for multiple (step, outcome) pairs atomically."""
+        if not branches:
+            return await self.drafts.get_draft_workspace(
+                workspace_id=workspace_id,
+            )
+        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        draft_routes = workspace.draft.get("routes", {})
+        if not isinstance(draft_routes, dict):
+            raise ValueError("draft routes must be an object")
+        patch: list[dict[str, Any]] = []
+        seen: set[tuple[str, str]] = set()
+        for ref in branches:
+            key = (ref.step_id, ref.outcome)
+            if key in seen:
+                continue
+            seen.add(key)
+            step_routes = draft_routes.get(ref.step_id, {})
+            if not isinstance(step_routes, dict):
+                continue
+            if ref.outcome not in step_routes:
+                continue
+            if step_routes[ref.outcome] == target:
+                continue
+            patch.append(
+                {
+                    "op": "replace",
+                    "path": (
+                        f"/routes/{escape_json_pointer(ref.step_id)}/"
+                        f"{escape_json_pointer(ref.outcome)}"
+                    ),
+                    "value": target,
+                }
+            )
+        if not patch:
+            return await self.drafts.get_draft_workspace(
+                workspace_id=workspace_id,
+            )
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
+
+@dataclass(frozen=True)
+class DraftOutcomeRef:
+    """A reference to a specific outcome of a draft step."""
+
+    step_id: str
+    outcome: str

src/wf_api/draft_payloads.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+
+
+def draft_step(draft: Mapping[str, Any], step_id: str) -> Mapping[str, Any]:
+    """Return one step mapping from a draft, raising on missing or non-object."""
+    steps = draft.get("steps", {})
+    if not isinstance(steps, Mapping):
+        raise KeyError("draft steps are not available")
+    step = steps[step_id]
+    if not isinstance(step, Mapping):
+        raise KeyError(f"draft step {step_id!r} is not an object")
+    return step
+
+
+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 input_bindings_payload(
+    input_map: dict[str, str],
+    input_values: dict[str, Any],
+) -> list[dict[str, Any]]:
+    """Serialize draft input maps into canonical string-path 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 output_bindings_payload(output_map: dict[str, str]) -> list[dict[str, Any]]:
+    """Serialize draft output maps into canonical string-path binding payloads."""
+    return [
+        {"source": _local_path_payload(source), "target": _state_path_payload(target)}
+        for source, target in output_map.items()
+    ]
+
+
+def state_root_field(value: str) -> str:
+    """Return the single root field name from a state path, or raise."""
+    path = StatePath.parse(value)
+    if len(path.parts) != 1:
+        raise ValueError("state_path must name one root field, such as state.after")
+    return path.parts[0]
+
+
+def _local_path_payload(value: str) -> str:
+    return LocalPath._serialize(LocalPath.parse(value))
+
+
+def _graph_path_payload(value: str | GraphSourcePath) -> str:
+    path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
+    return GraphSourcePath._serialize(path)
+
+
+def _state_path_payload(value: str) -> str:
+    return StatePath._serialize(StatePath.parse(value))

src/wf_api/drafts.py

@@ -25,21 +25,24 @@ from wf_core.models.steps import (
     InputValueBinding,
     OutputBinding,
 )
-from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
 from .capability_requirements import (
     required_capabilities_for_plan,
     required_capability_payloads,
 )
-from .constants import (
-    DEFAULT_CALL_STEP_ID,
-    DEFAULT_ERROR_OUTCOME,
-    DEFAULT_ERROR_STEP_ID,
-    DEFAULT_OK_OUTCOME,
-    RUNTIME_ERROR_CAPABILITY,
+from .draft_payloads import (
+    draft_step as _draft_step,
+)
+from .draft_payloads import (
+    escape_json_pointer as _escape_json_pointer,
+)
+from .draft_payloads import (
+    input_bindings_payload as _draft_input_bindings_payload,
+)
+from .draft_payloads import (
+    output_bindings_payload as _draft_output_bindings_payload,
 )
 from .operation_context import WorkflowOperationContext
-from .schema_projection import project_output_property_to_state_schema
 
 
 class WorkflowDraftApi:
@@ -181,6 +184,14 @@ class WorkflowDraftApi:
         store.save_workspace(refreshed)
         return get_draft_workspace_record(store, workspace_id=workspace_id)
 
+    async def compile_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
+        """Compile a stored draft workspace without mutating it."""
+        workspace = self._draft_store().get_workspace(workspace_id)
+        validation = await self.validate_draft(draft=workspace.draft)
+        if validation["status"] != "valid":
+            return validation
+        return await self.compile_draft(draft=workspace.draft)
+
     async def patch_draft_workspace(
         self,
         *,
@@ -288,197 +299,6 @@ class WorkflowDraftApi:
             ],
         )
 
-    async def add_state_schema_from_output(
-        self,
-        *,
-        workspace_id: str,
-        revision: int,
-        step_id: str,
-        output_field: str,
-        state_path: str,
-    ) -> dict[str, Any]:
-        workspace = self._draft_store().get_workspace(workspace_id)
-        step = _draft_step(workspace.draft, step_id)
-        capability_name = step.get("use")
-        if not isinstance(capability_name, str) or not capability_name:
-            raise ValueError(
-                f"draft step {step_id!r} does not declare a capability use"
-            )
-        state_field = _state_root_field(state_path)
-        spec = self.context.specs.get_qualified_spec(capability_name)
-        output_schema = (
-            spec.output_schema_contract or spec.output_model.model_json_schema()
-        )
-        state_schema = workspace.draft.get("state_schema", {})
-        if not isinstance(state_schema, dict):
-            raise ValueError("draft state_schema must be an object")
-        projected = project_output_property_to_state_schema(
-            state_schema=state_schema,
-            output_schema=output_schema,
-            output_field=output_field,
-            state_field=state_field,
-        )
-        return await self.patch_draft_workspace(
-            workspace_id=workspace_id,
-            revision=revision,
-            patch=[
-                {
-                    "op": "replace",
-                    "path": "/state_schema",
-                    "value": projected,
-                }
-            ],
-        )
-
-    async def bind_output_to_state(
-        self,
-        *,
-        workspace_id: str,
-        revision: int,
-        step_id: str,
-        output_field: str,
-        state_path: str,
-    ) -> dict[str, Any]:
-        """Declare a state field from a step output and bind that output to it.
-
-        This is the common draft-authoring repair for validation errors where a
-        step writes to ``state.x`` before ``state_schema.properties.x`` exists.
-        It deliberately edits only one root state field and one step output map.
-        Route changes remain explicit through ``set_draft_route``.
-        """
-        workspace = self._draft_store().get_workspace(workspace_id)
-        step = _draft_step(workspace.draft, step_id)
-        capability_name = step.get("use")
-        if not isinstance(capability_name, str) or not capability_name:
-            raise ValueError(
-                f"draft step {step_id!r} does not declare a capability use"
-            )
-
-        state_field = _state_root_field(state_path)
-        spec = self.context.specs.get_qualified_spec(capability_name)
-        output_schema = (
-            spec.output_schema_contract or spec.output_model.model_json_schema()
-        )
-        state_schema = workspace.draft.get("state_schema", {})
-        if not isinstance(state_schema, dict):
-            raise ValueError("draft state_schema must be an object")
-        projected = project_output_property_to_state_schema(
-            state_schema=state_schema,
-            output_schema=output_schema,
-            output_field=output_field,
-            state_field=state_field,
-        )
-        output_map = {
-            **self._step_output_map(workspace_id=workspace_id, step_id=step_id),
-            output_field: state_path,
-        }
-        return await self.patch_draft_workspace(
-            workspace_id=workspace_id,
-            revision=revision,
-            patch=[
-                {
-                    "op": "replace",
-                    "path": "/state_schema",
-                    "value": projected,
-                },
-                {
-                    "op": "replace",
-                    "path": f"/steps/{_escape_json_pointer(step_id)}/output",
-                    "value": _draft_output_bindings_payload(output_map),
-                },
-            ],
-        )
-
-    async def add_step_from_capability(
-        self,
-        *,
-        workspace_id: str,
-        revision: int,
-        step_id: str,
-        capability_name: str,
-        route_from_step: str | None = None,
-        route_from_outcome: str = DEFAULT_OK_OUTCOME,
-        route_outcome: str = DEFAULT_OK_OUTCOME,
-        route_to: str = "__end__",
-        input_map: dict[str, str] | None = None,
-        bind_outputs: dict[str, str] | None = None,
-    ) -> dict[str, Any]:
-        """Add one capability step plus explicit route/map/schema wiring.
-
-        This is a composed authoring helper for agents.  It edits the draft in
-        one revision so callers do not have to interleave add-step, route,
-        input-map, state-schema, and output-map operations by hand.
-        """
-        workspace = self._draft_store().get_workspace(workspace_id)
-        steps = workspace.draft.get("steps")
-        if not isinstance(steps, dict):
-            raise ValueError("draft steps must be an object")
-        if step_id in steps:
-            raise ValueError(f"draft step {step_id!r} already exists")
-
-        spec = self.context.specs.get_qualified_spec(capability_name)
-        output_schema = (
-            spec.output_schema_contract or spec.output_model.model_json_schema()
-        )
-        state_schema = workspace.draft.get("state_schema", {})
-        if not isinstance(state_schema, dict):
-            raise ValueError("draft state_schema must be an object")
-
-        input_map = input_map or {}
-        bind_outputs = bind_outputs or {}
-        projected_state_schema = state_schema
-        for output_field, state_path in bind_outputs.items():
-            state_field = _state_root_field(state_path)
-            projected_state_schema = project_output_property_to_state_schema(
-                state_schema=projected_state_schema,
-                output_schema=output_schema,
-                output_field=output_field,
-                state_field=state_field,
-            )
-
-        patch: list[dict[str, Any]] = [
-            {
-                "op": "add",
-                "path": f"/steps/{_escape_json_pointer(step_id)}",
-                "value": {
-                    "use": capability_name,
-                    "input": _draft_input_bindings_payload(input_map, {}),
-                    "output": _draft_output_bindings_payload(bind_outputs),
-                },
-            },
-            {
-                "op": "add",
-                "path": f"/routes/{_escape_json_pointer(step_id)}",
-                "value": {route_outcome: route_to},
-            },
-        ]
-        if projected_state_schema != state_schema:
-            patch.insert(
-                0,
-                {
-                    "op": "replace",
-                    "path": "/state_schema",
-                    "value": projected_state_schema,
-                },
-            )
-        if route_from_step is not None:
-            patch.append(
-                {
-                    "op": "add",
-                    "path": (
-                        f"/routes/{_escape_json_pointer(route_from_step)}/"
-                        f"{_escape_json_pointer(route_from_outcome)}"
-                    ),
-                    "value": step_id,
-                }
-            )
-
-        return await self.patch_draft_workspace(
-            workspace_id=workspace_id,
-            revision=revision,
-            patch=patch,
-        )
-
     def _step_input_maps(
         self,
         *,
@@ -494,76 +314,6 @@ class WorkflowDraftApi:
         step = _draft_step(workspace.draft, step_id)
         return _output_map_from_payload(step.get("output", []))
 
-    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 _draft_input_maps(
     *,
@@ -606,38 +356,6 @@ def _draft_output_map(
     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 _draft_step(draft: Mapping[str, Any], step_id: str) -> Mapping[str, Any]:
-    steps = draft.get("steps", {})
-    if not isinstance(steps, Mapping):
-        raise KeyError("draft steps are not available")
-    step = steps[step_id]
-    if not isinstance(step, Mapping):
-        raise KeyError(f"draft step {step_id!r} is not an object")
-    return step
-
-
 def _input_maps_from_payload(
     payload: Any,
 ) -> tuple[dict[str, str], dict[str, Any]]:
@@ -694,24 +412,6 @@ def _path_text(value: Any, *, expected_root: str | None = None) -> str:
     return root if not raw_parts else f"{root}.{'.'.join(raw_parts)}"
 
 
-def _graph_path_payload(value: str | GraphSourcePath) -> str:
-    path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
-    return GraphSourcePath._serialize(path)
-
-
-def _local_path_payload(value: str) -> str:
-    return LocalPath._serialize(LocalPath.parse(value))
-
-
-def _state_path_payload(value: str) -> 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 _with_workspace_repair_hints(
     payload: dict[str, Any],
     *,
@@ -762,10 +462,3 @@ def _draft_repair_hint(
         f"wf draft bind-output-to-state {workspace_id} --revision {revision} "
         f"--step {step_id} --output {output_field} --state {state_path}"
     )
-
-
-def _state_root_field(value: str) -> str:
-    path = StatePath.parse(value)
-    if len(path.parts) != 1:
-        raise ValueError("state_path must name one root field, such as state.after")
-    return path.parts[0]

src/wf_api/service.py

@@ -8,6 +8,7 @@ from wf_artifacts import ArtifactKind
 from .artifacts import WorkflowArtifactApi
 from .capabilities import WorkflowCapabilityApi
 from .deployments import WorkflowDeploymentApi
+from .draft_authoring import WorkflowDraftAuthoringApi
 from .drafts import WorkflowDraftApi
 from .models import RawWorkflowPlan
 from .operation_context import WorkflowOperationContext
@@ -26,6 +27,7 @@ class WorkflowApi:
         self.context = context
         self.capabilities = WorkflowCapabilityApi(context)
         self.drafts = WorkflowDraftApi(context)
+        self.draft_authoring = WorkflowDraftAuthoringApi(context, self.drafts)
         self.artifacts = WorkflowArtifactApi(context)
         self.deployments = WorkflowDeploymentApi(context)
         self.runs = WorkflowRunApi(context)
@@ -285,6 +287,13 @@ class WorkflowApi:
     ) -> dict[str, Any]:
         return await self.drafts.validate_draft_workspace(workspace_id=workspace_id)
 
+    async def compile_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+    ) -> dict[str, Any]:
+        return await self.drafts.compile_draft_workspace(workspace_id=workspace_id)
+
     async def patch_draft_workspace(
         self,
         *,
@@ -362,23 +371,6 @@ class WorkflowApi:
             merge=merge,
         )
 
-    async def add_state_schema_from_output(
-        self,
-        *,
-        workspace_id: str,
-        revision: int,
-        step_id: str,
-        output_field: str,
-        state_path: str,
-    ) -> dict[str, Any]:
-        return await self.drafts.add_state_schema_from_output(
-            workspace_id=workspace_id,
-            revision=revision,
-            step_id=step_id,
-            output_field=output_field,
-            state_path=state_path,
-        )
-
     async def bind_output_to_state(
         self,
         *,
@@ -388,7 +380,7 @@ class WorkflowApi:
         output_field: str,
         state_path: str,
     ) -> dict[str, Any]:
-        return await self.drafts.bind_output_to_state(
+        return await self.draft_authoring.bind_output_to_state(
             workspace_id=workspace_id,
             revision=revision,
             step_id=step_id,
@@ -405,24 +397,58 @@ class WorkflowApi:
         capability_name: str,
         route_from_step: str | None = None,
         route_from_outcome: str = "ok",
-        route_outcome: str = "ok",
-        route_to: str = "__end__",
+        routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
         bind_outputs: dict[str, str] | None = None,
     ) -> dict[str, Any]:
-        return await self.drafts.add_step_from_capability(
+        return await self.draft_authoring.add_step_from_capability(
             workspace_id=workspace_id,
             revision=revision,
             step_id=step_id,
             capability_name=capability_name,
             route_from_step=route_from_step,
             route_from_outcome=route_from_outcome,
-            route_outcome=route_outcome,
-            route_to=route_to,
+            routes=routes,
             input_map=input_map,
             bind_outputs=bind_outputs,
         )
 
+    async def branch_draft(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        routes: dict[str, str],
+    ) -> dict[str, Any]:
+        return await self.draft_authoring.branch_draft(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            routes=routes,
+        )
+
+    async def handle_draft(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        branches: list[dict[str, str]],
+        target: str,
+    ) -> dict[str, Any]:
+        from .draft_authoring import DraftOutcomeRef
+
+        refs = [
+            DraftOutcomeRef(step_id=b["step_id"], outcome=b["outcome"])
+            for b in branches
+        ]
+        return await self.draft_authoring.handle_draft(
+            workspace_id=workspace_id,
+            revision=revision,
+            branches=refs,
+            target=target,
+        )
+
     async def create_minimal_draft_workspace(
         self,
         *,
@@ -439,7 +465,7 @@ class WorkflowApi:
         error_message_source: Any | None = None,
         title: str | None = None,
     ) -> dict[str, Any]:
-        return await self.drafts.create_minimal_draft_workspace(
+        return await self.draft_authoring.create_minimal_draft_workspace(
             workspace_id=workspace_id,
             name=name,
             capability_name=capability_name,

src/wf_api/surface.py

@@ -114,16 +114,6 @@ class WorkflowDraftSurface(Protocol):
         merge: bool = False,
     ) -> dict[str, Any]: ...
 
-    async def add_state_schema_from_output(
-        self,
-        *,
-        workspace_id: str,
-        revision: int,
-        step_id: str,
-        output_field: str,
-        state_path: str,
-    ) -> dict[str, Any]: ...
-
     async def bind_output_to_state(
         self,
         *,
@@ -143,18 +133,41 @@ class WorkflowDraftSurface(Protocol):
         capability_name: str,
         route_from_step: str | None = None,
         route_from_outcome: str = "ok",
-        route_outcome: str = "ok",
-        route_to: str = "__end__",
+        routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
         bind_outputs: dict[str, str] | None = None,
     ) -> dict[str, Any]: ...
 
+    async def branch_draft(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        routes: dict[str, str],
+    ) -> dict[str, Any]: ...
+
+    async def handle_draft(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        branches: list[dict[str, str]],
+        target: str,
+    ) -> dict[str, Any]: ...
+
     async def validate_draft_workspace(
         self,
         *,
         workspace_id: str,
     ) -> dict[str, Any]: ...
 
+    async def compile_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+    ) -> dict[str, Any]: ...
+
     async def delete_draft_workspace(
         self,
         *,

src/wf_cli/commands/drafts.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from pathlib import Path
 from typing import Annotated, Literal
 
@@ -277,47 +278,6 @@ def set_step_output_map(
     )
 
 
-@app.command("add-state-from-output")
-def add_state_from_output(
-    ctx: typer.Context,
-    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
-    revision: Annotated[
-        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
-    ],
-    step_id: Annotated[str, typer.Option("--step", help="Draft step id.")],
-    output_field: Annotated[
-        str,
-        typer.Option("--output", help="Top-level capability output field."),
-    ],
-    state_path: Annotated[
-        str,
-        typer.Option("--state", help="Root state path, for example state.after."),
-    ],
-) -> None:
-    """Copy one capability output field schema into draft state_schema.
-
-    Use this before mapping a step output into a new state field. The command
-    reads the selected draft step's capability output schema, copies the
-    requested output property schema, and preserves local $defs/definitions so
-    JSON Schema refs remain valid.
-
-    Run `wf draft validate <workspace_id>` after adding state schema fields.
-    """
-    context = load_cli_context(ctx)
-    emit_json(
-        run_cli_operation(
-            context,
-            context.handlers.add_state_schema_from_output(
-                workspace_id=workspace_id,
-                revision=revision,
-                step_id=step_id,
-                output_field=output_field,
-                state_path=state_path,
-            ),
-        )
-    )
-
-
 @app.command("bind-output-to-state")
 def bind_output_to_state(
     ctx: typer.Context,
@@ -381,14 +341,13 @@ def add_step_from_capability(
         str,
         typer.Option("--from-outcome", help="Outcome on --from-step."),
     ] = "ok",
-    route_outcome: Annotated[
-        str,
-        typer.Option("--outcome", help="Outcome emitted by the new step."),
-    ] = "ok",
-    route_to: Annotated[
-        str,
-        typer.Option("--to", help="Target step id or __end__ for the new step."),
-    ] = "__end__",
+    route: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--route",
+            help="Route mapping OUTCOME=TARGET. Repeat for multiple outcomes.",
+        ),
+    ] = None,
     input_mapping: Annotated[
         list[str] | None,
         typer.Option(
@@ -414,6 +373,15 @@ def add_step_from_capability(
     """
     input_map = _parse_map_flags(input_mapping)
     bind_outputs = _parse_map_flags(output_mapping)
+    routes: dict[str, str] = {}
+    if route:
+        for r in route:
+            key, _, value = r.partition("=")
+            if not key or not value:
+                raise typer.BadParameter(
+                    f"invalid route: {r!r} (expected OUTCOME=TARGET)"
+                )
+            routes[key] = value
     context = load_cli_context(ctx)
     emit_json(
         run_cli_operation(
@@ -425,8 +393,7 @@ def add_step_from_capability(
                 capability_name=capability_name,
                 route_from_step=route_from_step,
                 route_from_outcome=route_from_outcome,
-                route_outcome=route_outcome,
-                route_to=route_to,
+                routes=routes or None,
                 input_map=input_map,
                 bind_outputs=bind_outputs,
             ),
@@ -434,6 +401,86 @@ def add_step_from_capability(
     )
 
 
+@app.command("branch")
+def branch_draft(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+    revision: Annotated[
+        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
+    ],
+    step: Annotated[str, typer.Option("--step", help="Draft step id.")],
+    route: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--route",
+            help="Route mapping OUTCOME=TARGET. Repeat for multiple outcomes.",
+        ),
+    ] = None,
+) -> None:
+    """Branch multiple outcome routes on a single step atomically."""
+    routes: dict[str, str] = {}
+    if route:
+        for r in route:
+            key, _, value = r.partition("=")
+            if not key or not value:
+                raise typer.BadParameter(
+                    f"invalid route: {r!r} (expected OUTCOME=TARGET)"
+                )
+            routes[key] = value
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.branch_draft(
+                workspace_id=workspace_id,
+                revision=revision,
+                step_id=step,
+                routes=routes,
+            ),
+        )
+    )
+
+
+@app.command("handle")
+def handle_draft(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+    revision: Annotated[
+        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
+    ],
+    to: Annotated[str, typer.Option("--to", help="Target step id or __end__.")],
+    branch: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--branch",
+            help="Branch mapping STEP:OUTCOME. Repeat for multiple branches.",
+        ),
+    ] = None,
+) -> None:
+    """Set a common target for multiple step/outcome pairs atomically."""
+    branches: list[dict[str, str]] = []
+    if branch:
+        for b in branch:
+            parts = b.rsplit(":", 1)
+            if len(parts) != 2 or not parts[0] or not parts[1]:
+                raise typer.BadParameter(
+                    f"invalid branch: {b!r} (expected STEP:OUTCOME)"
+                )
+            branches.append({"step_id": parts[0], "outcome": parts[1]})
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.handle_draft(
+                workspace_id=workspace_id,
+                revision=revision,
+                branches=branches,
+                target=to,
+            ),
+        )
+    )
+
+
 @app.command("validate")
 def validate_draft(
     ctx: typer.Context,
@@ -449,6 +496,24 @@ def validate_draft(
     )
 
 
+@app.command("compile")
+def compile_draft(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+) -> None:
+    """Compile a stored draft workspace without mutating it."""
+    context = load_cli_context(ctx)
+    result = run_cli_operation(
+        context,
+        context.handlers.compile_draft_workspace(workspace_id=workspace_id),
+    )
+    if "compiled_plan" in result:
+        emit_json(result["compiled_plan"])
+        return
+    typer.echo(json.dumps(result, indent=2, sort_keys=True), err=True)
+    raise typer.Exit(1)
+
+
 @app.command("delete")
 def delete_draft(
     ctx: typer.Context,

src/wf_mcp/workflow_surface/models.py

@@ -187,6 +187,12 @@ class ValidateDraftWorkspaceRequest(BaseModel):
     workspace_id: WorkspaceId
 
 
+class CompileDraftWorkspaceRequest(BaseModel):
+    """Typed MCP request for compiling one workspace draft without mutation."""
+
+    workspace_id: WorkspaceId
+
+
 class SetDraftNameRequest(BaseModel):
     """Typed MCP request for changing the workflow draft name."""
 
@@ -237,20 +243,6 @@ class SetStepOutputMapRequest(BaseModel):
     )
 
 
-class AddStateFromOutputRequest(BaseModel):
-    """Typed MCP request for declaring a state field from a step output schema."""
-
-    workspace_id: WorkspaceId
-    revision: int = Field(ge=1, description="Expected current workspace revision.")
-    step_id: str = Field(description="Draft step id whose capability output is used.")
-    output_field: str = Field(
-        description="Top-level output field to copy, for example after."
-    )
-    state_path: str = Field(
-        description="Root state path to declare, for example state.after."
-    )
-
-
 class BindOutputToStateRequest(BaseModel):
     """Typed MCP request for binding one step output to one root state field."""
 
@@ -280,13 +272,13 @@ class AddStepFromCapabilityRequest(BaseModel):
         default="ok",
         description="Outcome on route_from_step that should route to the new step.",
     )
-    route_outcome: str = Field(
-        default="ok",
-        description="Outcome emitted by the new step.",
-    )
-    route_to: str = Field(
-        default="__end__",
-        description="Target step id or __end__ for the new step outcome.",
+    routes: dict[str, str] | None = Field(
+        default=None,
+        description=(
+            "Outcome to target route map for the new step. If omitted, a single "
+            "declared outcome routes to __end__; multiple declared outcomes "
+            "require explicit routes."
+        ),
     )
     input_map: dict[str, str] = Field(
         default_factory=dict,
@@ -298,6 +290,31 @@ class AddStepFromCapabilityRequest(BaseModel):
     )
 
 
+class BranchDraftRequest(BaseModel):
+    """Typed MCP request for branching routes on a draft step."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected workspace revision.")
+    step_id: str = Field(description="Draft step id whose routes should be branched.")
+    routes: dict[str, str] = Field(description="Outcome to target step map.")
+
+
+class HandleDraftBranchItem(BaseModel):
+    """Single branch item for handle_draft."""
+
+    step_id: str = Field(description="Draft step id.")
+    outcome: str = Field(description="Outcome label.")
+
+
+class HandleDraftRequest(BaseModel):
+    """Typed MCP request for handling draft branches."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected workspace revision.")
+    branches: list[HandleDraftBranchItem] = Field(description="Branch items to handle.")
+    target: str = Field(description="Target step id or __end__.")
+
+
 class DeleteDraftWorkspaceRequest(BaseModel):
     """Typed MCP request payload for deleting one draft workspace."""
 

src/wf_mcp/workflow_surface/tools.py

@@ -12,10 +12,11 @@ from wf_mcp.broker.service import WfMcpService
 from wf_mcp.broker.service.workflow_operation_context import context_from_service
 
 from .models import (
-    AddStateFromOutputRequest,
     AddStepFromCapabilityRequest,
     BindOutputToStateRequest,
+    BranchDraftRequest,
     CallCapabilityResult,
+    CompileDraftWorkspaceRequest,
     CreateArtifactFromWorkspaceRequest,
     CreateDraftWorkspaceFromCapabilityRequest,
     CreateDraftWorkspaceFromCapabilityResult,
@@ -26,6 +27,7 @@ from .models import (
     DeleteDraftWorkspaceResult,
     DraftWorkspaceListResult,
     DraftWorkspaceResult,
+    HandleDraftRequest,
     PatchDraftWorkspaceRequest,
     RunDeploymentResult,
     SetDraftNameRequest,
@@ -368,6 +370,22 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             await handlers.validate_draft_workspace(workspace_id=request.workspace_id)
         )
 
+    @server.tool(
+        name="wf.workflow.compile_draft_workspace",
+        title="Compile Draft Workspace",
+        description=(
+            "Compile a stored draft workspace without mutating it. Returns "
+            "compiled_plan and required_capabilities when valid, or diagnostics "
+            "when invalid."
+        ),
+    )
+    async def compile_draft_workspace(
+        request: CompileDraftWorkspaceRequest,
+    ) -> dict[str, Any]:
+        return await handlers.compile_draft_workspace(
+            workspace_id=request.workspace_id,
+        )
+
     @server.tool(
         name="wf.workflow.set_draft_name",
         title="Set Draft Name",
@@ -442,27 +460,6 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
-    @server.tool(
-        name="wf.workflow.add_state_from_output",
-        title="Add State From Output",
-        description=(
-            "Declare one root state field by copying a draft step capability output "
-            "field schema, including local $defs/definitions when present."
-        ),
-    )
-    async def add_state_from_output(
-        request: AddStateFromOutputRequest,
-    ) -> DraftWorkspaceResult:
-        return DraftWorkspaceResult.model_validate(
-            await handlers.add_state_schema_from_output(
-                workspace_id=request.workspace_id,
-                revision=request.revision,
-                step_id=request.step_id,
-                output_field=request.output_field,
-                state_path=request.state_path,
-            )
-        )
-
     @server.tool(
         name="wf.workflow.bind_output_to_state",
         title="Bind Output To State",
@@ -503,13 +500,48 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 capability_name=request.capability_name,
                 route_from_step=request.route_from_step,
                 route_from_outcome=request.route_from_outcome,
-                route_outcome=request.route_outcome,
-                route_to=request.route_to,
+                routes=request.routes,
                 input_map=request.input_map,
                 bind_outputs=request.bind_outputs,
             )
         )
 
+    @server.tool(
+        name="wf.workflow.branch_draft",
+        title="Branch Draft",
+        description=(
+            "Branch multiple outcome routes on a single draft step atomically. "
+            "Provided outcomes are updated while unspecified outcomes are preserved."
+        ),
+    )
+    async def branch_draft(request: BranchDraftRequest) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.branch_draft(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                step_id=request.step_id,
+                routes=request.routes,
+            )
+        )
+
+    @server.tool(
+        name="wf.workflow.handle_draft",
+        title="Handle Draft",
+        description=("Set a common target for multiple step/outcome pairs atomically."),
+    )
+    async def handle_draft(request: HandleDraftRequest) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.handle_draft(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                branches=[
+                    {"step_id": b.step_id, "outcome": b.outcome}
+                    for b in request.branches
+                ],
+                target=request.target,
+            )
+        )
+
     @server.tool(
         name="wf.workflow.create_minimal_draft_workspace",
         title="Create Minimal Draft Workspace",

src/wf_transport_rpc_http/__init__.py

@@ -6,13 +6,17 @@ from .errors import WorkflowRpcError
 from .models import (
     AddStepFromCapabilityParams,
     AdminEmptyParams,
+    BranchDraftParams,
     CallCapabilityParams,
+    CompileDraftWorkspaceParams,
     CreateArtifactFromPlanParams,
     CreateArtifactFromWorkspaceParams,
     CreateDraftFromCapabilityParams,
     CreateWrapperFromWorkspaceParams,
     DeleteDeploymentParams,
     GetDraftWorkspaceParams,
+    HandleDraftBranch,
+    HandleDraftParams,
     HealthParams,
     InspectArtifactParams,
     InspectCapabilityParams,
@@ -42,15 +46,19 @@ from .models import (
 )
 
 __all__ = [
-    "CreateArtifactFromPlanParams",
-    "CreateArtifactFromWorkspaceParams",
     "AddStepFromCapabilityParams",
     "AdminEmptyParams",
+    "BranchDraftParams",
     "CallCapabilityParams",
+    "CompileDraftWorkspaceParams",
+    "CreateArtifactFromPlanParams",
+    "CreateArtifactFromWorkspaceParams",
     "CreateDraftFromCapabilityParams",
     "CreateWrapperFromWorkspaceParams",
     "DeleteDeploymentParams",
     "GetDraftWorkspaceParams",
+    "HandleDraftBranch",
+    "HandleDraftParams",
     "HealthParams",
     "InspectArtifactParams",
     "InspectCapabilityParams",

src/wf_transport_rpc_http/client/drafts.py

@@ -141,26 +141,6 @@ class RpcDraftClientMixin:
             },
         )
 
-    async def add_state_schema_from_output(
-        self: RpcCaller,
-        *,
-        workspace_id: str,
-        revision: int,
-        step_id: str,
-        output_field: str,
-        state_path: str,
-    ) -> dict[str, Any]:
-        return await self._call(
-            "workflow.draft_workspaces.add_state_from_output",
-            {
-                "workspace_id": workspace_id,
-                "revision": revision,
-                "step_id": step_id,
-                "output_field": output_field,
-                "state_path": state_path,
-            },
-        )
-
     async def bind_output_to_state(
         self: RpcCaller,
         *,
@@ -190,8 +170,7 @@ class RpcDraftClientMixin:
         capability_name: str,
         route_from_step: str | None = None,
         route_from_outcome: str = "ok",
-        route_outcome: str = "ok",
-        route_to: str = "__end__",
+        routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
         bind_outputs: dict[str, str] | None = None,
     ) -> dict[str, Any]:
@@ -204,13 +183,48 @@ class RpcDraftClientMixin:
                 "capability_name": capability_name,
                 "route_from_step": route_from_step,
                 "route_from_outcome": route_from_outcome,
-                "route_outcome": route_outcome,
-                "route_to": route_to,
+                "routes": routes,
                 "input_map": input_map or {},
                 "bind_outputs": bind_outputs or {},
             },
         )
 
+    async def branch_draft(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        routes: dict[str, str],
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.branch",
+            {
+                "workspace_id": workspace_id,
+                "revision": revision,
+                "step_id": step_id,
+                "routes": routes,
+            },
+        )
+
+    async def handle_draft(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        branches: list[dict[str, str]],
+        target: str,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.handle",
+            {
+                "workspace_id": workspace_id,
+                "revision": revision,
+                "branches": branches,
+                "target": target,
+            },
+        )
+
     async def validate_draft_workspace(
         self: RpcCaller,
         *,
@@ -221,6 +235,16 @@ class RpcDraftClientMixin:
             {"workspace_id": workspace_id},
         )
 
+    async def compile_draft_workspace(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.compile",
+            {"workspace_id": workspace_id},
+        )
+
     async def delete_draft_workspace(
         self: RpcCaller,
         *,

src/wf_transport_rpc_http/methods/drafts.py

@@ -8,14 +8,16 @@ from wf_server import WorkflowServer
 
 from ..errors import WorkflowRpcError, raise_workflow_rpc_error
 from ..models import (
-    AddStateFromOutputParams,
     AddStepFromCapabilityParams,
     BindOutputToStateParams,
+    BranchDraftParams,
+    CompileDraftWorkspaceParams,
     CreateArtifactFromWorkspaceParams,
     CreateDraftFromCapabilityParams,
     CreateWrapperFromWorkspaceParams,
     DeleteDraftWorkspaceParams,
     GetDraftWorkspaceParams,
+    HandleDraftParams,
     ListDraftWorkspacesParams,
     PatchDraftParams,
     PatchDraftWorkspaceParams,
@@ -180,24 +182,6 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
-    @entrypoint.method(
-        name="workflow.draft_workspaces.add_state_from_output",
-        errors=[WorkflowRpcError],
-    )
-    async def workflow_draft_workspaces_add_state_from_output(
-        params: AddStateFromOutputParams = RpcParams(),
-    ) -> dict[str, Any]:
-        try:
-            return await server.api.add_state_schema_from_output(
-                workspace_id=params.workspace_id,
-                revision=params.revision,
-                step_id=params.step_id,
-                output_field=params.output_field,
-                state_path=params.state_path,
-            )
-        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
-            raise_workflow_rpc_error(exc)
-
     @entrypoint.method(
         name="workflow.draft_workspaces.bind_output_to_state",
         errors=[WorkflowRpcError],
@@ -231,14 +215,48 @@ def register_methods(
                 capability_name=params.capability_name,
                 route_from_step=params.route_from_step,
                 route_from_outcome=params.route_from_outcome,
-                route_outcome=params.route_outcome,
-                route_to=params.route_to,
+                routes=params.routes,
                 input_map=params.input_map,
                 bind_outputs=params.bind_outputs,
             )
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.branch", errors=[WorkflowRpcError]
+    )
+    async def workflow_draft_workspaces_branch(
+        params: BranchDraftParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.branch_draft(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                step_id=params.step_id,
+                routes=params.routes,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(
+        name="workflow.draft_workspaces.handle", errors=[WorkflowRpcError]
+    )
+    async def workflow_draft_workspaces_handle(
+        params: HandleDraftParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.handle_draft(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                branches=[
+                    {"step_id": b.step_id, "outcome": b.outcome}
+                    for b in params.branches
+                ],
+                target=params.target,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.validate", errors=[WorkflowRpcError]
     )
@@ -252,6 +270,19 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.compile", errors=[WorkflowRpcError]
+    )
+    async def workflow_draft_workspaces_compile(
+        params: CompileDraftWorkspaceParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.compile_draft_workspace(
+                workspace_id=params.workspace_id,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.delete", errors=[WorkflowRpcError]
     )

src/wf_transport_rpc_http/models.py

@@ -141,14 +141,6 @@ class SetStepOutputMapParams(RpcParamsModel):
     merge: bool = False
 
 
-class AddStateFromOutputParams(RpcParamsModel):
-    workspace_id: str = Field(min_length=1)
-    revision: int = Field(ge=1)
-    step_id: str = Field(min_length=1)
-    output_field: str = Field(min_length=1)
-    state_path: str = Field(min_length=1)
-
-
 class BindOutputToStateParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     revision: int = Field(ge=1)
@@ -164,16 +156,38 @@ class AddStepFromCapabilityParams(RpcParamsModel):
     capability_name: str = Field(min_length=1)
     route_from_step: str | None = None
     route_from_outcome: str = Field(default="ok", min_length=1)
-    route_outcome: str = Field(default="ok", min_length=1)
-    route_to: str = Field(default="__end__", min_length=1)
+    routes: dict[str, str] | None = None
     input_map: dict[str, str] = Field(default_factory=dict)
     bind_outputs: dict[str, str] = Field(default_factory=dict)
 
 
+class BranchDraftParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    step_id: str = Field(min_length=1)
+    routes: dict[str, str]
+
+
+class HandleDraftBranch(RpcParamsModel):
+    step_id: str = Field(min_length=1)
+    outcome: str = Field(min_length=1)
+
+
+class HandleDraftParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    branches: list[HandleDraftBranch]
+    target: str = Field(min_length=1)
+
+
 class ValidateDraftWorkspaceParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
 
 
+class CompileDraftWorkspaceParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+
+
 class DeleteDraftWorkspaceParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
 

tests/wf_api/test_drafts_service.py

@@ -7,7 +7,9 @@ import pytest
 from pydantic import BaseModel
 
 from tests.wf_mcp.test_support import echo_tool
+from wf_api.draft_authoring import DraftOutcomeRef, WorkflowDraftAuthoringApi
 from wf_api.drafts import WorkflowDraftApi
+from wf_api.service import WorkflowApi
 from wf_artifacts import FileDraftWorkspaceStore, FileWorkflowArtifactStore
 from wf_authoring import node
 from wf_mcp.broker import WfMcpService
@@ -65,7 +67,7 @@ class _SnapshotInput(BaseModel):
     pass
 
 
-@node(name="snapshot_tool")
+@node(name="snapshot_tool", outcomes=("ok", "skipped"))
 def _snapshot_tool(payload: _SnapshotInput) -> _SnapshotOutput:
     return _SnapshotOutput(after=_Snapshot(clicked=True))
 
@@ -74,7 +76,7 @@ def _draft_api(
     artifact_store: FileWorkflowArtifactStore,
     *,
     register_echo: bool = False,
-) -> tuple[WorkflowDraftApi, WfMcpService]:
+) -> tuple[WorkflowDraftApi, WfMcpService, WorkflowDraftAuthoringApi]:
     mcp_root = artifact_store.root / "drafts_mcp" / str(id(artifact_store))
     service = WfMcpService(
         store=FileStore(mcp_root),
@@ -87,13 +89,17 @@ def _draft_api(
         )
         service.register_specs("demo.personal", echo_tool)
     context = context_from_service(service)
-    return WorkflowDraftApi(context), service
+    return (
+        WorkflowDraftApi(context),
+        service,
+        WorkflowDraftAuthoringApi(context, WorkflowDraftApi(context)),
+    )
 
 
 @pytest.mark.asyncio
 async def test_patch_draft_applies_json_patch(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch")
-    api, _service = _draft_api(artifact_store, register_echo=True)
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
 
     result = await api.patch_draft(
         draft=_echo_draft(),
@@ -116,7 +122,7 @@ async def test_patch_draft_applies_json_patch(tmp_path: Path) -> None:
 @pytest.mark.asyncio
 async def test_create_draft_workspace_creates_workspace(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_create_workspace")
-    api, _service = _draft_api(artifact_store)
+    api, _service, _authoring = _draft_api(artifact_store)
 
     result = await api.create_draft_workspace(
         workspace_id="echo_ws",
@@ -138,7 +144,7 @@ async def test_list_draft_workspaces_returns_sorted_summaries_without_drafts(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_list_workspaces")
-    api, _service = _draft_api(artifact_store)
+    api, _service, _authoring = _draft_api(artifact_store)
     await api.create_draft_workspace(
         workspace_id="b_draft",
         title="B Draft",
@@ -164,7 +170,7 @@ async def test_list_draft_workspaces_returns_sorted_summaries_without_drafts(
 @pytest.mark.asyncio
 async def test_delete_draft_workspace_is_idempotent(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_delete_workspace")
-    api, _service = _draft_api(artifact_store)
+    api, _service, _authoring = _draft_api(artifact_store)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
         draft=_echo_draft(),
@@ -186,7 +192,7 @@ async def test_delete_draft_workspace_is_idempotent(tmp_path: Path) -> None:
 @pytest.mark.asyncio
 async def test_patch_draft_workspace_updates_revision(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_workspace")
-    api, _service = _draft_api(artifact_store, register_echo=True)
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
         draft=_echo_draft(),
@@ -207,7 +213,7 @@ async def test_draft_workspace_patch_helpers_update_revision_and_bindings(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_helpers")
-    api, _service = _draft_api(artifact_store)
+    api, _service, _authoring = _draft_api(artifact_store)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
         draft=_echo_draft(),
@@ -263,7 +269,7 @@ async def test_draft_workspace_patch_helpers_update_revision_and_bindings(
 @pytest.mark.asyncio
 async def test_step_map_helpers_merge_with_existing_bindings(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_helper_merge")
-    api, _service = _draft_api(artifact_store)
+    api, _service, _authoring = _draft_api(artifact_store)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
         draft=_echo_draft(),
@@ -315,7 +321,7 @@ async def test_step_map_helpers_merge_with_existing_bindings(tmp_path: Path) ->
 @pytest.mark.asyncio
 async def test_validate_draft_workspace_refreshes_status(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_validate_workspace")
-    api, service = _draft_api(artifact_store, register_echo=True)
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
     draft = _echo_draft()
     draft["routes"]["echo"] = {"typo": "__end__"}
     await api.create_draft_workspace(
@@ -340,7 +346,7 @@ async def test_validate_draft_workspace_suggests_bind_output_to_state(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_repair_hint")
-    api, service = _draft_api(artifact_store)
+    api, service, authoring = _draft_api(artifact_store)
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
@@ -385,7 +391,7 @@ async def test_patch_draft_workspace_validates_new_use_step_with_context_specs(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_new_use")
-    api, service = _draft_api(artifact_store, register_echo=True)
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
     service.register_specs("demo.personal", echo_tool, _snapshot_tool)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
@@ -438,9 +444,9 @@ async def test_create_minimal_draft_workspace_minimal_success_path(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_minimal_workspace")
-    api, _service = _draft_api(artifact_store, register_echo=True)
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
 
-    result = await api.create_minimal_draft_workspace(
+    result = await authoring.create_minimal_draft_workspace(
         workspace_id="echo_minimal",
         name="echo",
         capability_name="demo.personal.echo_tool",
@@ -498,92 +504,40 @@ async def test_delegation_smoke_validate_draft_equivalence(tmp_path: Path) -> No
 
 
 @pytest.mark.asyncio
-async def test_add_state_schema_from_output_copies_output_property_defs(
+async def test_facade_delegates_semantic_authoring_to_authoring_service(
     tmp_path: Path,
 ) -> None:
-    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_state_from_output")
-    api, service = _draft_api(artifact_store)
+    """WorkflowApi constructs a sibling WorkflowDraftAuthoringApi."""
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_facade_delegation")
+    mcp_root = artifact_store.root / "facade_mcp"
+    service = WfMcpService(
+        store=FileStore(mcp_root),
+        artifact_store=artifact_store,
+        draft_workspace_store=FileDraftWorkspaceStore(mcp_root),
+    )
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
-    service.register_specs("demo.personal", _snapshot_tool)
-    await api.create_draft_workspace(
-        workspace_id="snapshot_ws",
-        draft={
-            "name": "snapshot",
-            "input_schema": {"type": "object", "properties": {}},
-            "state_schema": {"type": "object", "properties": {}},
-            "output_schema": {"type": "object", "properties": {}},
-            "start": "snap",
-            "steps": {
-                "snap": {
-                    "use": "demo.personal.snapshot_tool",
-                    "input": [],
-                    "output": [],
-                }
-            },
-            "routes": {"snap": {"ok": "__end__"}},
-        },
-    )
+    service.register_specs("demo.personal", echo_tool, _snapshot_tool)
 
-    updated = await api.add_state_schema_from_output(
-        workspace_id="snapshot_ws",
+    context = context_from_service(service)
+    facade = WorkflowApi(context)
+
+    assert facade.draft_authoring is not None
+    assert isinstance(facade.draft_authoring, WorkflowDraftAuthoringApi)
+
+    await facade.create_draft_workspace(
+        workspace_id="ws1",
+        draft=_echo_draft(),
+    )
+    result = await facade.bind_output_to_state(
+        workspace_id="ws1",
         revision=1,
-        step_id="snap",
-        output_field="after",
-        state_path="state.after",
+        step_id="echo",
+        output_field="echoed",
+        state_path="state.echoed",
     )
-    fetched = await api.get_draft_workspace(
-        workspace_id="snapshot_ws",
-        include_draft=True,
-    )
-
-    state_schema = fetched["draft"]["state_schema"]
-    assert updated["revision"] == 2
-    assert state_schema["properties"]["after"]["$ref"] == "#/$defs/_Snapshot"
-    assert state_schema["$defs"]["_Snapshot"]["properties"]["clicked"] == {
-        "title": "Clicked",
-        "type": "boolean",
-    }
-
-
-@pytest.mark.asyncio
-async def test_add_state_schema_from_output_rejects_nested_state_path(
-    tmp_path: Path,
-) -> None:
-    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_nested_state_output")
-    api, service = _draft_api(artifact_store)
-    service.register_connection(
-        ConnectionConfig(id="demo.personal", server="demo", account="personal")
-    )
-    service.register_specs("demo.personal", _snapshot_tool)
-    await api.create_draft_workspace(
-        workspace_id="snapshot_ws",
-        draft={
-            "name": "snapshot",
-            "input_schema": {"type": "object", "properties": {}},
-            "state_schema": {"type": "object", "properties": {}},
-            "output_schema": {"type": "object", "properties": {}},
-            "start": "snap",
-            "steps": {
-                "snap": {
-                    "use": "demo.personal.snapshot_tool",
-                    "input": [],
-                    "output": [],
-                }
-            },
-            "routes": {"snap": {"ok": "__end__"}},
-        },
-    )
-
-    with pytest.raises(ValueError, match="state_path must name one root field"):
-        await api.add_state_schema_from_output(
-            workspace_id="snapshot_ws",
-            revision=1,
-            step_id="snap",
-            output_field="after",
-            state_path="state.after.clicked",
-        )
+    assert result["revision"] == 2
 
 
 @pytest.mark.asyncio
@@ -591,7 +545,7 @@ async def test_bind_output_to_state_projects_schema_and_merges_output_map(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_output_state")
-    api, service = _draft_api(artifact_store)
+    api, service, authoring = _draft_api(artifact_store)
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
@@ -623,7 +577,7 @@ async def test_bind_output_to_state_projects_schema_and_merges_output_map(
         },
     )
 
-    updated = await api.bind_output_to_state(
+    updated = await authoring.bind_output_to_state(
         workspace_id="snapshot_ws",
         revision=1,
         step_id="snap",
@@ -657,7 +611,7 @@ async def test_bind_output_to_state_projects_schema_and_merges_output_map(
 @pytest.mark.asyncio
 async def test_bind_output_to_state_rejects_nested_state_path(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_nested_state")
-    api, service = _draft_api(artifact_store)
+    api, service, authoring = _draft_api(artifact_store)
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
@@ -682,7 +636,7 @@ async def test_bind_output_to_state_rejects_nested_state_path(tmp_path: Path) ->
     )
 
     with pytest.raises(ValueError, match="state_path must name one root field"):
-        await api.bind_output_to_state(
+        await authoring.bind_output_to_state(
             workspace_id="snapshot_ws",
             revision=1,
             step_id="snap",
@@ -696,7 +650,7 @@ async def test_bind_output_to_state_rejects_missing_output_field(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_missing_output")
-    api, service = _draft_api(artifact_store)
+    api, service, authoring = _draft_api(artifact_store)
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
@@ -721,7 +675,7 @@ async def test_bind_output_to_state_rejects_missing_output_field(
     )
 
     with pytest.raises(ValueError, match="output field 'missing'"):
-        await api.bind_output_to_state(
+        await authoring.bind_output_to_state(
             workspace_id="snapshot_ws",
             revision=1,
             step_id="snap",
@@ -735,7 +689,7 @@ async def test_bind_output_to_state_rejects_step_without_capability_use(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_no_use")
-    api, _service = _draft_api(artifact_store)
+    api, _service, authoring = _draft_api(artifact_store)
     await api.create_draft_workspace(
         workspace_id="snapshot_ws",
         draft={
@@ -755,7 +709,7 @@ async def test_bind_output_to_state_rejects_step_without_capability_use(
     )
 
     with pytest.raises(ValueError, match="does not declare a capability use"):
-        await api.bind_output_to_state(
+        await authoring.bind_output_to_state(
             workspace_id="snapshot_ws",
             revision=1,
             step_id="snap",
@@ -769,22 +723,21 @@ async def test_add_step_from_capability_wires_route_inputs_and_state_outputs(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_add_step")
-    api, service = _draft_api(artifact_store, register_echo=True)
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
     service.register_specs("demo.personal", echo_tool, _snapshot_tool)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
         draft=_echo_draft(),
     )
 
-    result = await api.add_step_from_capability(
+    result = await authoring.add_step_from_capability(
         workspace_id="echo_ws",
         revision=1,
         step_id="snap",
         capability_name="demo.personal.snapshot_tool",
         route_from_step="echo",
         route_from_outcome="ok",
-        route_outcome="ok",
-        route_to="__end__",
+        routes={"ok": "__end__", "skipped": "__end__"},
         input_map={},
         bind_outputs={"after": "state.after"},
     )
@@ -814,22 +767,266 @@ async def test_add_step_from_capability_rejects_existing_step_id(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_add_step_duplicate")
-    api, _service = _draft_api(artifact_store, register_echo=True)
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
     await api.create_draft_workspace(
         workspace_id="echo_ws",
         draft=_echo_draft(),
     )
 
     with pytest.raises(ValueError, match="draft step 'echo' already exists"):
-        await api.add_step_from_capability(
+        await authoring.add_step_from_capability(
             workspace_id="echo_ws",
             revision=1,
             step_id="echo",
             capability_name="demo.personal.echo_tool",
             route_from_step=None,
             route_from_outcome="ok",
-            route_outcome="ok",
-            route_to="__end__",
+            routes={"ok": "__end__"},
             input_map={},
             bind_outputs={},
         )
+
+
+@pytest.mark.asyncio
+async def test_branch_draft_updates_routes_atomically(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_branch")
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="branching",
+        draft={
+            "name": "branching",
+            "input_schema": {"type": "object", "properties": {}},
+            "state_schema": {"type": "object", "properties": {}},
+            "output_schema": {"type": "object", "properties": {}},
+            "start": "classify",
+            "steps": {
+                "classify": {
+                    "use": "demo.personal.echo_tool",
+                    "input": [],
+                    "output": [],
+                },
+                "tool_error": {
+                    "use": "demo.personal.echo_tool",
+                    "input": [],
+                    "output": [],
+                },
+            },
+            "routes": {
+                "classify": {"ok": "classify"},
+                "tool_error": {"ok": "__end__"},
+            },
+        },
+    )
+
+    result = await authoring.branch_draft(
+        workspace_id="branching",
+        revision=1,
+        step_id="classify",
+        routes={"ok": "classify", "error": "tool_error"},
+    )
+    assert result["revision"] == 2
+    workspace = await api.get_draft_workspace(
+        workspace_id="branching", include_draft=True
+    )
+    assert workspace["draft"]["routes"]["classify"] == {
+        "ok": "classify",
+        "error": "tool_error",
+    }
+    assert "tool_error" in workspace["draft"]["steps"]
+
+
+@pytest.mark.asyncio
+async def test_handle_draft_updates_multiple_source_outcomes(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_handle")
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="handling",
+        draft={
+            "name": "handling",
+            "input_schema": {"type": "object", "properties": {}},
+            "state_schema": {"type": "object", "properties": {}},
+            "output_schema": {"type": "object", "properties": {}},
+            "start": "lookup",
+            "steps": {
+                "lookup": {
+                    "use": "demo.personal.echo_tool",
+                    "input": [],
+                    "output": [],
+                },
+                "transform": {
+                    "use": "demo.personal.echo_tool",
+                    "input": [],
+                    "output": [],
+                },
+            },
+            "routes": {
+                "lookup": {"ok": "transform", "error": "lookup"},
+                "transform": {"ok": "__end__", "error": "transform"},
+            },
+        },
+    )
+
+    result = await authoring.handle_draft(
+        workspace_id="handling",
+        revision=1,
+        branches=[
+            DraftOutcomeRef(step_id="lookup", outcome="error"),
+            DraftOutcomeRef(step_id="transform", outcome="error"),
+        ],
+        target="__end__",
+    )
+    assert result["revision"] == 2
+    workspace = await api.get_draft_workspace(
+        workspace_id="handling", include_draft=True
+    )
+    assert workspace["draft"]["routes"]["lookup"]["error"] == "__end__"
+    assert workspace["draft"]["routes"]["transform"]["error"] == "__end__"
+    assert workspace["draft"]["routes"]["lookup"]["ok"] == "transform"
+    assert workspace["draft"]["routes"]["transform"]["ok"] == "__end__"
+
+
+@pytest.mark.asyncio
+async def test_branch_draft_no_change_when_routes_unchanged(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_branch_noop")
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="noop_ws",
+        draft=_echo_draft(),
+    )
+    before = await api.get_draft_workspace(workspace_id="noop_ws", include_draft=True)
+
+    result = await authoring.branch_draft(
+        workspace_id="noop_ws",
+        revision=1,
+        step_id="echo",
+        routes={"ok": "__end__"},
+    )
+    after = await api.get_draft_workspace(workspace_id="noop_ws", include_draft=True)
+    assert result["revision"] == 1
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_handle_draft_empty_branches_noop(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_handle_noop")
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="noop_ws",
+        draft=_echo_draft(),
+    )
+    before = await api.get_draft_workspace(workspace_id="noop_ws", include_draft=True)
+
+    result = await authoring.handle_draft(
+        workspace_id="noop_ws",
+        revision=1,
+        branches=[],
+        target="fail",
+    )
+    after = await api.get_draft_workspace(workspace_id="noop_ws", include_draft=True)
+    assert result["revision"] == 1
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_infers_single_outcome_route(
+    tmp_path: Path,
+) -> None:
+    """One declared outcome named 'done', no routes supplied -> routes to __end__."""
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_single_outcome")
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
+    service.register_specs("demo.personal", echo_tool)
+    await api.create_draft_workspace(
+        workspace_id="single",
+        draft=_echo_draft(),
+    )
+
+    result = await authoring.add_step_from_capability(
+        workspace_id="single",
+        revision=1,
+        step_id="done_step",
+        capability_name="demo.personal.echo_tool",
+    )
+    assert result["revision"] == 2
+    workspace = await api.get_draft_workspace(workspace_id="single", include_draft=True)
+    assert workspace["draft"]["routes"]["done_step"] == {"ok": "__end__"}
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_requires_complete_routes_for_multi_outcome(
+    tmp_path: Path,
+) -> None:
+    """Multiple declared outcomes with incomplete explicit routes raises ValueError."""
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_multi_outcome")
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
+    service.register_specs("demo.personal", echo_tool, _snapshot_tool)
+    await api.create_draft_workspace(
+        workspace_id="multi",
+        draft=_echo_draft(),
+    )
+
+    with pytest.raises(ValueError, match="missing routes"):
+        await authoring.add_step_from_capability(
+            workspace_id="multi",
+            revision=1,
+            step_id="snap",
+            capability_name="demo.personal.snapshot_tool",
+            routes={"ok": "__end__"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_rejects_unknown_routes_for_multi_outcome(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_unknown_outcome")
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
+    service.register_specs("demo.personal", _snapshot_tool)
+    await api.create_draft_workspace(
+        workspace_id="unknown_multi",
+        draft=_echo_draft(),
+    )
+
+    with pytest.raises(ValueError, match="unknown routes"):
+        await authoring.add_step_from_capability(
+            workspace_id="unknown_multi",
+            revision=1,
+            step_id="snap",
+            capability_name="demo.personal.snapshot_tool",
+            routes={"ok": "__end__", "skipped": "__end__", "typo": "__end__"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_compile_draft_workspace_returns_compiled_plan(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_compile")
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="compile_me",
+        draft=_echo_draft(),
+    )
+    before = await api.get_draft_workspace(
+        workspace_id="compile_me", include_draft=True
+    )
+    result = await api.compile_draft_workspace(workspace_id="compile_me")
+    after = await api.get_draft_workspace(workspace_id="compile_me", include_draft=True)
+    assert result["compiled_plan"]["name"] == "echo"
+    assert result["required_capabilities"]
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_compile_draft_workspace_invalid_returns_diagnostics(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_compile_invalid")
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    draft = _echo_draft()
+    draft["routes"]["echo"] = {"typo": "__end__"}
+    await api.create_draft_workspace(
+        workspace_id="invalid_ws",
+        draft=draft,
+    )
+    result = await api.compile_draft_workspace(workspace_id="invalid_ws")
+    assert result["status"] == "invalid"
+    assert "compiled_plan" not in result
+    assert result["diagnostics"]

tests/wf_cli/test_app.py

@@ -144,16 +144,6 @@ def test_wf_draft_map_help_explains_replace_merge_and_validate() -> None:
     assert "draft validate" in output_help
 
 
-def test_wf_draft_add_state_from_output_help_explains_schema_copy() -> None:
-    result = runner.invoke(app, ["draft", "add-state-from-output", "--help"])
-
-    assert result.exit_code == 0
-    help_text = " ".join(result.output.split())
-    assert "capability output field schema" in help_text
-    assert "$defs" in help_text
-    assert "draft validate" in help_text
-
-
 def test_wf_draft_bind_output_to_state_help_explains_composed_edit() -> None:
     result = runner.invoke(app, ["draft", "bind-output-to-state", "--help"])
 

tests/wf_cli/test_remote_target.py

@@ -1101,23 +1101,6 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
             "--merge",
         ],
     )
-    state_added = runner.invoke(
-        app,
-        [
-            *base_args,
-            "draft",
-            "add-state-from-output",
-            "focused_ws",
-            "--revision",
-            "7",
-            "--step",
-            "call",
-            "--output",
-            "value",
-            "--state",
-            "state.extra_value",
-        ],
-    )
     inspected = runner.invoke(
         app,
         [*base_args, "draft", "inspect", "focused_ws", "--include-draft"],
@@ -1129,7 +1112,6 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
     assert output_mapped.exit_code == 0, output_mapped.output
     assert input_merged.exit_code == 0, input_merged.output
     assert output_merged.exit_code == 0, output_merged.output
-    assert state_added.exit_code == 0, state_added.output
     assert inspected.exit_code == 0, inspected.output
     payload = json.loads(inspected.output)
     draft = payload["draft"]
@@ -1137,25 +1119,24 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
     assert draft["routes"]["call"]["ok"] == "__end__"
     assert draft["steps"]["call"]["input"] == [
         {
-            "target": {"root": "local", "parts": ["value"]},
-            "path": {"root": "input", "parts": ["value"]},
+            "target": "value",
+            "path": "input.value",
         },
         {
-            "target": {"root": "local", "parts": ["extra"]},
-            "path": {"root": "input", "parts": ["extra"]},
+            "target": "extra",
+            "path": "input.extra",
         },
     ]
     assert draft["steps"]["call"]["output"] == [
         {
-            "source": {"root": "local", "parts": ["value"]},
-            "target": {"root": "state", "parts": ["value"]},
+            "source": "value",
+            "target": "state.value",
         },
         {
-            "source": {"root": "local", "parts": ["extra"]},
-            "target": {"root": "state", "parts": ["extra"]},
+            "source": "extra",
+            "target": "state.extra",
         },
     ]
-    assert "extra_value" in draft["state_schema"]["properties"]
 
 
 def test_wf_draft_bind_output_to_state_uses_rpc_target(monkeypatch, tmp_path) -> None:
@@ -1244,10 +1225,8 @@ def test_wf_draft_add_step_from_capability_uses_rpc_target(
             "call",
             "--from-outcome",
             "ok",
-            "--outcome",
-            "ok",
-            "--to",
-            "__end__",
+            "--route",
+            "ok=__end__",
             "--input",
             "input.value=value",
             "--bind-output",
@@ -1261,6 +1240,85 @@ def test_wf_draft_add_step_from_capability_uses_rpc_target(
     assert payload["status"] == "valid"
 
 
+def test_wf_draft_compile_prints_compiled_plan(monkeypatch, tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    _patch_rpc_client_to_server(monkeypatch, server)
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+
+    created = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "create-from-capability",
+            "compile_ws",
+            "wf.std.constant",
+            "--name",
+            "compile_me",
+        ],
+    )
+    assert created.exit_code == 0, created.output
+
+    result = runner.invoke(app, [*base_args, "draft", "compile", "compile_ws"])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["name"] == "compile_me"
+    assert "compiled_plan" not in payload
+
+
+def test_wf_draft_compile_invalid_prints_diagnostics_to_stderr(
+    monkeypatch, tmp_path
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    asyncio.run(
+        server.api.create_draft_workspace(
+            workspace_id="invalid_compile_ws",
+            draft={
+                "name": "invalid_compile",
+                "input_schema": {"type": "object"},
+                "state_schema": {"type": "object", "properties": {}},
+                "output_schema": {"type": "object", "properties": {}},
+                "start": "call",
+                "steps": {
+                    "call": {
+                        "use": "wf.std.constant",
+                        "input": [],
+                        "output": [],
+                    }
+                },
+                "routes": {"call": {"typo": "__end__"}},
+            },
+        )
+    )
+    _patch_rpc_client_to_server(monkeypatch, server)
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+
+    result = runner.invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "--url",
+            "http://test/rpc",
+            "draft",
+            "compile",
+            "invalid_compile_ws",
+        ],
+    )
+
+    assert result.exit_code == 1
+    # This Typer test runner mixes stderr into output; the command implementation
+    # writes invalid compile diagnostics with err=True for real terminals.
+    assert '"status": "invalid"' in result.output
+    assert "compiled_plan" not in result.output
+
+
 def test_wf_deploy_create_alias_saves_deployment(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     asyncio.run(

tests/wf_mcp/server/test_config.py

@@ -56,7 +56,6 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_draft_route" in names
             assert "wf.workflow.set_step_input_map" in names
             assert "wf.workflow.set_step_output_map" in names
-            assert "wf.workflow.add_state_from_output" in names
             assert "wf.workflow.bind_output_to_state" in names
             assert "wf.workflow.add_step_from_capability" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
@@ -115,12 +114,6 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             ].inputSchema
             set_input_request = set_input_schema["properties"]["request"]
             assert "merge" in set_input_request["properties"]
-            state_from_output_schema = tools_by_name[
-                "wf.workflow.add_state_from_output"
-            ].inputSchema
-            state_request = state_from_output_schema["properties"]["request"]
-            assert "output_field" in state_request["properties"]
-            assert "state_path" in state_request["properties"]
             add_step_schema = tools_by_name[
                 "wf.workflow.add_step_from_capability"
             ].inputSchema

tests/wf_mcp/workflow_surface/test_drafts.py

@@ -44,7 +44,7 @@ def test_workflow_surface_rejects_unknown_draft_route_outcome_when_spec_is_known
     payload = asyncio.run(h.validate_draft(draft=draft))
 
     assert payload["status"] == "invalid"
-    assert payload["diagnostics"][0]["path"] == "routes.echo.typo"
+    assert payload["diagnostics"][0]["path"] == "edges[0].outcome"
 
 
 def test_workflow_surface_creates_artifact_from_draft_with_binding_suggestions(
@@ -150,7 +150,7 @@ def test_workflow_surface_validates_draft_workspace_with_live_outcomes(
 
     assert payload["revision"] == 1
     assert payload["status"] == "invalid"
-    assert payload["diagnostics"][0]["code"] == "unknown_outcome"
+    assert payload["diagnostics"][0]["code"] == "undeclared_edge_outcome"
     assert fetched["status"] == "invalid"
 
 

tests/wf_transport_rpc_http/test_app.py

@@ -266,7 +266,7 @@ async def test_rpc_draft_workspace_methods(tmp_path) -> None:
                 "name": "remote_constant",
                 "title": "Remote Constant",
                 "input_map": {},
-                "output_map": {"value": "state.result"},
+                "output_map": {},
             },
         )
         listed = await _rpc(client, "workflow.draft_workspaces.list", {})
@@ -722,23 +722,12 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
                 "merge": True,
             },
         )
-        state_added = await _rpc(
-            client,
-            "workflow.draft_workspaces.add_state_from_output",
-            {
-                "workspace_id": "focused_ws",
-                "revision": 7,
-                "step_id": "call",
-                "output_field": "value",
-                "state_path": "state.extra_value",
-            },
-        )
         state_bound = await _rpc(
             client,
             "workflow.draft_workspaces.bind_output_to_state",
             {
                 "workspace_id": "focused_ws",
-                "revision": state_added["result"]["revision"],
+                "revision": 7,
                 "step_id": "call",
                 "output_field": "value",
                 "state_path": "state.extra_value",
@@ -756,8 +745,7 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
     assert output_mapped["result"]["revision"] == 5
     assert input_merged["result"]["revision"] == 6
     assert output_merged["result"]["revision"] == 7
-    assert state_added["result"]["revision"] == 8
-    assert state_bound["result"]["revision"] == 9
+    assert state_bound["result"]["revision"] == 8
     draft = fetched["result"]["draft"]
     assert draft["name"] == "focused_renamed"
     assert draft["routes"]["call"]["ok"] == "__end__"
@@ -812,8 +800,7 @@ async def test_rpc_draft_workspace_add_step_from_capability(tmp_path) -> None:
                 "capability_name": "wf.std.constant",
                 "route_from_step": "call",
                 "route_from_outcome": "ok",
-                "route_outcome": "ok",
-                "route_to": "__end__",
+                "routes": {"ok": "__end__"},
                 "input_map": {"input.value": "value"},
                 "bind_outputs": {"value": "state.second_value"},
             },

tests/wf_transport_rpc_http/test_client.py

@@ -293,7 +293,7 @@ async def test_rpc_workflow_client_draft_workspace_lifecycle(tmp_path) -> None:
             name="client_constant",
             title="Client Constant",
             input_map={},
-            output_map={"value": "state.result"},
+            output_map={},
         )
         listed = await client.list_draft_workspaces()
         fetched = await client.get_draft_workspace(workspace_id="client_ws")
@@ -501,16 +501,9 @@ async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None
             output_map={"extra": "state.extra"},
             merge=True,
         )
-        state_added = await client.add_state_schema_from_output(
-            workspace_id="client_focused_ws",
-            revision=7,
-            step_id="call",
-            output_field="value",
-            state_path="state.extra_value",
-        )
         state_bound = await client.bind_output_to_state(
             workspace_id="client_focused_ws",
-            revision=state_added["revision"],
+            revision=7,
             step_id="call",
             output_field="value",
             state_path="state.extra_value",
@@ -522,8 +515,7 @@ async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None
     assert output_mapped["revision"] == 5
     assert input_merged["revision"] == 6
     assert output_merged["revision"] == 7
-    assert state_added["revision"] == 8
-    assert state_bound["revision"] == 9
+    assert state_bound["revision"] == 8
 
 
 async def test_rpc_client_draft_workspace_add_step_from_capability(tmp_path) -> None:
@@ -550,8 +542,7 @@ async def test_rpc_client_draft_workspace_add_step_from_capability(tmp_path) ->
             capability_name="wf.std.constant",
             route_from_step="call",
             route_from_outcome="ok",
-            route_outcome="ok",
-            route_to="__end__",
+            routes={"ok": "__end__"},
             input_map={"input.value": "value"},
             bind_outputs={"value": "state.second_value"},
         )