draft: REALLY ergonomic, not full impl

things that stop it from being complete: workflow builder route needs work

is this the call for normal branch/PR styled work?

docs/superpowers/plans/2026-05-18-workflow-draft-surface.md

@@ -0,0 +1,692 @@
+# Workflow Draft Surface Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace the disposable draft prototype with the first real MCP-facing workflow draft surface: keyed, patchable JSON that adapts into `wf_authoring.WorkflowBuilder` instead of rebuilding graph semantics itself.
+
+**Architecture:** The new draft layer remains a typed JSON seam in `wf_artifacts`, but delegates graph construction to `wf_authoring`. Draft parsing owns keyed presentation, validation, and stable patch paths; `WorkflowBuilder` owns graph construction. The MCP workflow surface keeps the same tool family while accepting the new draft shape.
+
+**Tech Stack:** Python 3.14, Pydantic v2, `wf_authoring`, `wf_core`, `jsonpatch`, pytest, basedpyright, Ruff.
+
+---
+
+## File Structure
+
+### Create
+
+- `src/wf_artifacts/drafts/models.py`
+  - concrete Pydantic draft document models
+- `src/wf_artifacts/drafts/adapter.py`
+  - thin JSON-draft-to-`WorkflowBuilder` adapter
+- `src/wf_artifacts/drafts/api.py`
+  - public compile/validate/patch functions and diagnostics
+- `tests/artifacts/test_draft_models.py`
+  - draft document validation
+- `tests/artifacts/test_draft_adapter.py`
+  - keyed step/routes lowering through `WorkflowBuilder`
+- `tests/artifacts/test_draft_api.py`
+  - public compile/validate/patch behavior
+
+### Modify
+
+- `src/wf_artifacts/drafts.py`
+  - replace module body with compatibility re-exports or remove once imports are updated
+- `src/wf_artifacts/__init__.py`
+  - export public draft API/models from the package
+- `src/wf_authoring/builder/core.py`
+  - add `use_ref(...)` for named external capabilities without local `NodeSpec`s
+- `src/wf_authoring/ops/*`
+  - only if route helpers need a reusable public lowering entrypoint
+- `src/wf_mcp/workflow_surface/handlers.py`
+  - keep using public draft API; no semantic duplication
+- `tests/wf_mcp/test_workflow_surface.py`
+  - update draft fixtures to the new keyed document shape
+- `tests/wf_mcp/test_server.py`
+  - confirm MCP schemas remain plain-object friendly
+- `docs/workflow_drafts.md`
+  - replace prototype examples with the first real draft surface
+- `docs/wf_mcp_end_to_end_runbook.md`
+  - update draft example
+- `docs/wf_mcp_operator_manual.md`
+  - keep draft-first guidance accurate
+- `docs/wf_mcp_troubleshooting.md`
+  - update patch-path examples
+
+## Task 1: Split Draft Code Into Focused Modules
+
+**Files:**
+- Create: `src/wf_artifacts/drafts/models.py`
+- Create: `src/wf_artifacts/drafts/api.py`
+- Create: `src/wf_artifacts/drafts/adapter.py`
+- Modify: `src/wf_artifacts/drafts.py`
+- Modify: `src/wf_artifacts/__init__.py`
+- Test: `tests/artifacts/test_draft_models.py`
+
+- [ ] **Step 1: Write the failing model tests**
+
+```python
+from wf_artifacts.drafts import DraftUseStep, WorkflowDraft
+
+
+def test_workflow_draft_uses_keyed_steps() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "echo",
+            "input_schema": {},
+            "state_schema": {"fields": {}},
+            "output_schema": {},
+            "start": "echo",
+            "steps": {
+                "echo": {
+                    "use": "demo.echo",
+                    "in": {"input.text": "text"},
+                    "out": {"echoed": "state.echoed"},
+                }
+            },
+            "routes": {"echo": {"ok": "__end__"}},
+        }
+    )
+
+    assert isinstance(draft.steps["echo"], DraftUseStep)
+    assert draft.steps["echo"].use == "demo.echo"
+
+
+def test_draft_step_requires_exactly_one_kind_key() -> None:
+    result = WorkflowDraft.model_validate(
+        {
+            "name": "bad",
+            "input_schema": {},
+            "state_schema": {"fields": {}},
+            "output_schema": {},
+            "start": "bad",
+            "steps": {
+                "bad": {
+                    "use": "demo.echo",
+                    "join": {},
+                }
+            },
+            "routes": {},
+        }
+    )
+```
+
+The second test should be written with `pytest.raises(ValidationError)` and assert the authoring path identifies `steps.bad`.
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/artifacts/test_draft_models.py -q
+```
+
+Expected: import errors or validation failures because keyed draft models do not exist yet.
+
+- [ ] **Step 3: Implement minimal concrete draft models**
+
+Create:
+
+```python
+# src/wf_artifacts/drafts/models.py
+from __future__ import annotations
+
+from typing import Annotated, Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+JsonObject = dict[str, Any]
+STEP_KIND_KEYS = frozenset({"use", "foreach", "interrupt", "join"})
+
+
+class DraftUseStep(BaseModel):
+    use: str
+    in_: dict[str, str] = Field(default_factory=dict, alias="in")
+    out: dict[str, str] = Field(default_factory=dict)
+    desc: str | None = None
+    retry: int | None = Field(default=None, ge=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
+
+
+class DraftForeachPayload(BaseModel):
+    over: str
+    as_: str = Field(alias="as")
+    mode: Literal["serial", "parallel"] = "serial"
+    on_item_error: Literal["fail", "collect", "skip"] = "fail"
+
+
+class DraftForeachStep(BaseModel):
+    foreach: DraftForeachPayload
+
+
+class DraftInterruptPayload(BaseModel):
+    kind: str
+    request: dict[str, str] = Field(default_factory=dict)
+    resume: dict[str, str] = Field(default_factory=dict)
+    outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
+
+
+class DraftInterruptStep(BaseModel):
+    interrupt: DraftInterruptPayload
+
+
+class DraftJoinStep(BaseModel):
+    join: JsonObject = Field(default_factory=dict)
+
+
+DraftStep = Annotated[
+    DraftUseStep
+    | DraftForeachStep
+    | DraftInterruptStep
+    | DraftJoinStep,
+    Field(discriminator=None),
+]
+
+
+class WorkflowDraft(BaseModel):
+    name: str
+    input_schema: JsonObject
+    state_schema: JsonObject
+    output_schema: JsonObject
+    start: str
+    steps: dict[str, DraftStep]
+    routes: dict[str, dict[str, str]] = Field(default_factory=dict)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _validate_step_kinds(cls, value: object) -> object:
+        if not isinstance(value, dict):
+            return value
+        steps = value.get("steps")
+        if not isinstance(steps, dict):
+            return value
+        for step_id, payload in steps.items():
+            if not isinstance(payload, dict):
+                continue
+            present = STEP_KIND_KEYS.intersection(payload)
+            if len(present) != 1:
+                raise ValueError(
+                    f"steps.{step_id} must contain exactly one step kind key"
+                )
+        return value
+```
+
+Keep the public import path stable by re-exporting through `src/wf_artifacts/drafts.py` during the transition.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/artifacts/test_draft_models.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_artifacts tests/artifacts/test_draft_models.py
+git commit -m "refactor: add keyed workflow draft models"
+```
+
+## Task 2: Add `use_ref` And Thin Adapter Over `WorkflowBuilder`
+
+**Files:**
+- Create: `src/wf_artifacts/drafts/adapter.py`
+- Modify only if needed: `src/wf_authoring/builder/core.py`
+- Modify only if needed: `src/wf_authoring/ops/*`
+- Test: `tests/artifacts/test_draft_adapter.py`
+
+- [ ] **Step 1: Write failing adapter tests**
+
+```python
+from wf_artifacts.drafts import WorkflowDraft
+from wf_artifacts.drafts.adapter import build_workflow_from_draft
+
+
+def test_adapter_lowers_keyed_use_steps_and_routes_through_builder() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "echo",
+            "input_schema": {},
+            "state_schema": {"fields": {}},
+            "output_schema": {},
+            "start": "echo",
+            "steps": {"echo": {"use": "demo.echo"}},
+            "routes": {"echo": {"ok": "__end__"}},
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+
+    assert workflow.nodes[0].id == "echo"
+    assert workflow.nodes[0].node == "demo.echo"
+    assert workflow.edges[0].from_ == "echo"
+    assert workflow.edges[0].outcome == "ok"
+    assert workflow.edges[0].to == "__end__"
+
+
+def test_builder_use_ref_creates_external_node_use_without_node_def() -> None:
+    builder = WorkflowBuilder(
+        "echo",
+        input_schema={},
+        state_schema={"fields": {}},
+        output_schema={},
+    )
+
+    step = builder.use_ref("demo.echo", id="echo")
+    builder.set_entry_point(step)
+    builder.connect(step, "ok", "__end__")
+    workflow = builder.compile()
+
+    assert step.node == "demo.echo"
+    assert workflow.node_defs == []
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/artifacts/test_draft_adapter.py -q
+```
+
+Expected: import error because `build_workflow_from_draft` does not exist.
+
+- [ ] **Step 3: Implement the thin adapter**
+
+First add:
+
+```python
+def use_ref(
+    self,
+    name: str,
+    *,
+    id: str | None = None,
+    in_map: MapArg | None = None,
+    out_map: MapArg | None = None,
+    desc: str | None = None,
+) -> NodeUse:
+    ...
+```
+
+`use_ref` creates a `NodeUse` for an already named external capability and does
+not add a local `NodeDef`.
+
+Then implement `build_workflow_from_draft(draft: WorkflowDraft) -> Workflow` so
+it:
+
+1. constructs a `WorkflowBuilder`
+2. registers each draft step by stable id
+3. uses existing `WorkflowBuilder` public methods for:
+   - `use_ref`
+   - `foreach`
+   - `interrupt`
+   - `join`
+4. applies `routes`
+5. calls explicit `start(...)`
+6. returns `builder.build(...)`
+
+Do **not** invent draft route sugar in this pass.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/artifacts/test_draft_adapter.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_artifacts src/wf_authoring tests/artifacts/test_draft_adapter.py
+git commit -m "feat: adapt workflow drafts through workflow builder"
+```
+
+## Task 3: Replace Prototype Public API
+
+**Files:**
+- Create: `src/wf_artifacts/drafts/api.py`
+- Modify: `src/wf_artifacts/drafts.py`
+- Modify: `src/wf_artifacts/__init__.py`
+- Test: `tests/artifacts/test_draft_api.py`
+
+- [ ] **Step 1: Write failing API tests**
+
+```python
+from wf_artifacts.drafts import compile_workflow_draft, patch_workflow_draft
+
+
+def test_compile_workflow_draft_returns_raw_core_shape() -> None:
+    plan = compile_workflow_draft(_keyed_echo_draft())
+
+    assert plan["nodes"][0]["id"] == "echo"
+    assert plan["nodes"][0]["node"] == "demo.echo"
+    assert plan["edges"][0]["outcome"] == "ok"
+
+
+def test_patch_workflow_draft_uses_stable_step_paths() -> None:
+    result = patch_workflow_draft(
+        _keyed_echo_draft(),
+        [
+            {
+                "op": "replace",
+                "path": "/steps/echo/in/input.text",
+                "value": "message",
+            }
+        ],
+    )
+
+    assert result["status"] == "valid"
+    assert result["draft"]["steps"]["echo"]["in"]["input.text"] == "message"
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/artifacts/test_draft_api.py -q
+```
+
+Expected: failures because the old prototype API still expects array `steps`.
+
+- [ ] **Step 3: Implement the API**
+
+Create:
+
+```python
+# src/wf_artifacts/drafts/api.py
+def compile_workflow_draft(draft: JsonObject) -> JsonObject:
+    parsed = WorkflowDraft.model_validate(draft)
+    workflow = build_workflow_from_draft(parsed)
+    return workflow.model_dump(mode="json", by_alias=True, exclude={"node_defs"})
+```
+
+Keep:
+
+- `validate_workflow_draft`
+- `patch_workflow_draft`
+- structured `DraftDiagnostic`
+
+Update diagnostics to use keyed paths such as:
+
+```text
+steps.echo.in
+routes.echo.error
+```
+
+Delete the old array-step prototype code after public tests are green.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/artifacts/test_draft_api.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_artifacts tests/artifacts/test_draft_api.py
+git commit -m "feat: replace draft prototype with keyed public api"
+```
+
+## Task 4: Update MCP Workflow Surface
+
+**Files:**
+- Modify: `tests/wf_mcp/test_workflow_surface.py`
+- Modify: `tests/wf_mcp/test_server.py`
+- Modify only if needed: `src/wf_mcp/workflow_surface/handlers.py`
+
+- [ ] **Step 1: Update the failing MCP tests**
+
+Replace old fixtures like:
+
+```python
+"steps": [{"id": "echo", "kind": "use", ...}]
+```
+
+with:
+
+```python
+"steps": {"echo": {"use": "demo.echo", ...}},
+"routes": {"echo": {"ok": "__end__"}},
+```
+
+Keep assertions that:
+
+- draft tools still expose plain object schemas to MCP clients
+- `create_artifact_from_draft` still saves artifacts
+- source binding normalization still works
+- missing `wf.std` self-binding diagnostics still work
+
+- [ ] **Step 2: Run tests to verify failures**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/wf_mcp/test_workflow_surface.py tests/wf_mcp/test_server.py -q
+```
+
+Expected: failures wherever MCP handlers still assume the old prototype shape.
+
+- [ ] **Step 3: Make minimal MCP adjustments**
+
+Keep handlers thin:
+
+```python
+plan = compile_workflow_draft(draft)
+```
+
+No duplicate draft interpretation should appear in `wf_mcp`.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/wf_mcp/test_workflow_surface.py tests/wf_mcp/test_server.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_mcp tests/wf_mcp
+git commit -m "feat: accept keyed workflow drafts over mcp"
+```
+
+## Task 5: Add Outcome Validation When Capability Contracts Are Available
+
+**Files:**
+- Modify: `src/wf_artifacts/drafts/api.py`
+- Modify: `src/wf_mcp/workflow_surface/handlers.py`
+- Test: `tests/wf_mcp/test_workflow_surface.py`
+
+- [ ] **Step 1: Write failing outcome validation test**
+
+```python
+def test_draft_validation_rejects_unknown_capability_outcome_when_spec_is_known() -> None:
+    handlers = _handlers_with_demo_echo_spec()
+    draft = _keyed_echo_draft()
+    draft["routes"]["echo"] = {"typo": "__end__"}
+
+    result = asyncio.run(handlers.validate_draft(draft=draft))
+
+    assert result["status"] == "invalid"
+    assert result["diagnostics"][0]["path"] == "routes.echo.typo"
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/wf_mcp/test_workflow_surface.py -q
+```
+
+Expected: validation currently accepts the typo.
+
+- [ ] **Step 3: Implement capability-aware outcome validation**
+
+Pass an optional capability lookup into draft validation from MCP handlers.
+
+Rules:
+
+- validate outcome keys for `use` steps only when the capability is resolvable
+- if the capability is unknown/unavailable, leave dependency validation to the
+  later artifact/deployment stages
+- diagnostic path must identify the keyed route entry
+
+Do not make `wf_artifacts` depend on `wf_mcp`; define a tiny callable/protocol
+interface for lookup instead.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run:
+
+```bash
+uv run --with pytest pytest tests/wf_mcp/test_workflow_surface.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/wf_artifacts src/wf_mcp tests/wf_mcp/test_workflow_surface.py
+git commit -m "feat: validate draft routes against known outcomes"
+```
+
+## Task 6: Update Documentation
+
+**Files:**
+- Modify: `docs/workflow_drafts.md`
+- Modify: `docs/wf_mcp_end_to_end_runbook.md`
+- Modify: `docs/wf_mcp_operator_manual.md`
+- Modify: `docs/wf_mcp_troubleshooting.md`
+
+- [ ] **Step 1: Update docs to the real draft surface**
+
+Replace prototype array examples with keyed examples:
+
+```json
+"steps": {
+  "echo": {
+    "use": "demo.echo_tool",
+    "in": {"input.text": "text"},
+    "out": {"echoed": "state.echoed"}
+  }
+},
+"routes": {
+  "echo": {
+    "ok": "__end__"
+  }
+}
+```
+
+Document:
+
+- exactly-one-kind-key rule
+- stable keyed patch paths
+- `route` as repeated condition-chain sugar
+- `WorkflowBuilder` as the semantic owner beneath the JSON adapter
+- outcome strings validated against capability contracts when available
+
+- [ ] **Step 2: Run a targeted docs scan**
+
+Run:
+
+```bash
+rg -n '\"steps\": \\[|\"kind\": \"use\"|/steps/0|create_artifact_from_draft' docs
+```
+
+Expected:
+
+- no stale prototype examples in current docs
+- `create_artifact_from_draft` still documented as the preferred path
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add docs
+git commit -m "docs: describe keyed workflow draft surface"
+```
+
+## Task 7: Full Verification
+
+**Files:**
+- No new files
+
+- [ ] **Step 1: Run focused verification**
+
+```bash
+uv run --with pytest pytest tests/artifacts tests/wf_mcp/test_workflow_surface.py tests/wf_mcp/test_server.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 2: Run full project tests**
+
+```bash
+uv run --with pytest pytest -q
+```
+
+Expected: pass.
+
+- [ ] **Step 3: Run type checking**
+
+```bash
+uv run basedpyright --level error
+```
+
+Expected: `0 errors`.
+
+- [ ] **Step 4: Run lint**
+
+```bash
+uvx ruff check src tests
+```
+
+Expected: pass.
+
+- [ ] **Step 5: Commit final cleanup**
+
+```bash
+git add .
+git commit -m "feat: ship keyed workflow draft authoring surface"
+```
+
+## Self-Review
+
+### Spec Coverage
+
+- keyed `steps`: Tasks 1-4
+- compact `routes`: Tasks 1-4
+- verb-keyed explicit step kinds: Task 1
+- saved capability/workflow refs in `use`: Task 2, existing capability refs pass through unchanged
+- stable patch paths: Tasks 3 and 6
+- `wf_authoring` as semantic owner: Tasks 2 and 6
+- outcome validation against declared contracts: Task 5
+- prototype replacement rather than migration: Tasks 3, 4, 6
+
+### Placeholder Scan
+
+- no `TBD`
+- no unspecified "add validation" placeholders
+- every task has exact files, tests, commands, and expected behavior
+
+### Type Consistency
+
+- `WorkflowDraft`, `DraftUseStep`, `build_workflow_from_draft`, and public API
+  names stay consistent across all tasks
+- patch examples use keyed paths consistently
+- `routes` stays the only authored outcome-routing section

docs/superpowers/specs/2026-05-18-workflow-draft-surface-design.md

@@ -22,7 +22,7 @@ This pass covers:
 
 1. keyed steps
 2. compact outcome routes
-3. verb-keyed step shapes such as `use` and `route`
+3. verb-keyed step shapes such as `use`
 4. saved capability/workflow references in `use`
 5. stable JSON Patch paths
 6. parity documentation against current `wf_authoring`
@@ -30,6 +30,7 @@ This pass covers:
 This pass does **not** cover:
 
 - reverse-branch / shared outcome handlers
+- draft `route` sugar
 - a new `wf_authoring` fluent API
 - true subgraph support
 - new core graph semantics
@@ -85,7 +86,7 @@ The draft layer owns:
 
 It should not own:
 
-- route expansion
+- graph construction rules
 - graph construction rules
 - duplicate edge-building machinery
 - alternate workflow semantics
@@ -109,7 +110,6 @@ but no step kind is inferred.
 Exactly one step-kind key must be present. Allowed step-kind keys are:
 
 - `use`
-- `route`
 - `foreach`
 - `interrupt`
 - `join`
@@ -145,22 +145,8 @@ Zero kind keys or multiple kind keys are validation errors.
       }
     }
   },
-  "start": "has_text",
+  "start": "echo",
   "steps": {
-    "has_text": {
-      "route": [
-        {
-          "when": {
-            "op": "exists",
-            "path": "input.text"
-          },
-          "to": "echo"
-        },
-        {
-          "otherwise": "missing_text"
-        }
-      ]
-    },
     "echo": {
       "use": "demo.personal.echo_tool",
       "in": {
@@ -217,49 +203,9 @@ Call one workflow capability.
 - saved wrapper capability
 - saved workflow capability once graph-as-node is available
 
-The compiler lowers this to a core `NodeUse`.
+The adapter lowers this through `WorkflowBuilder.use_ref(...)`, which exists for
-
+named external capabilities that do not have a local Python callable-backed
-### `route`
+`NodeSpec`.
-
-Route based on input/state conditions.
-
-```json
-{
-  "route": [
-    {
-      "when": {
-        "op": "exists",
-        "path": "state.hit"
-      },
-      "to": "found"
-    },
-    {
-      "otherwise": "missing"
-    }
-  ]
-}
-```
-
-The draft adapter asks `wf_authoring` to lower a route step using the same
-repeated condition-chain sugar that current Python authoring uses:
-
-```text
-if cond_1 -> target_1 else
-if cond_2 -> target_2 else
-...
-otherwise -> fallback
-```
-
-That means traces show the expanded condition steps. No new core runtime
-semantics are required.
-
-Rules:
-
-- clauses are ordered
-- zero or more `when` clauses may appear
-- at most one `otherwise` clause may appear
-- `otherwise` must be last
-- every clause must declare a target
 
 ### `foreach`
 
@@ -323,7 +269,7 @@ Most ordinary edges should be authored through `routes`:
 }
 ```
 
-The adapter passes these through the same authoring route/connection path that
+The adapter passes these through the same authoring connection path that
 produces normal core edges.
 
 Outcome keys remain strings in JSON because they are wire values, but they
@@ -334,7 +280,7 @@ declared outcomes.
 `routes` is intended for:
 
 - outcome routing from `use`
-- outgoing edges from lowered route/condition machinery
+- outgoing edges from authored graph steps
 - ordinary terminal edges
 
 The raw edge list remains compiler output, not the normal authoring surface.
@@ -367,7 +313,7 @@ This is a core reason for keyed `steps`.
 | `g.use(spec, ...)` | `steps[id].use` | direct conceptual match |
 | `g.connect(step, outcome, target)` | `routes[id][outcome] = target` | same graph meaning, better JSON |
 | `g.branch(...)` | `routes[...]` | outcome routing is already compact in JSON |
-| `route(...)` node | `steps[id].route` | state/input routing |
+| `route(...)` node | not in the first draft surface | defer until both front doors agree |
 | explicit `start(...)` | `start` | direct match |
 | `END` | `"__end__"` | keep wire token explicit |
 | `NodeSpec` object | capability ref string | MCP cannot carry Python callable identity |
@@ -403,7 +349,7 @@ Possible later surfaces:
 The MCP draft surface exposes friction in current Python authoring too:
 
 - no fluent shared-handler helper
-- no full parity with route sugar
+- no draft parity for route sugar yet
 - likely room for better grouped declarations
 - outcome names still travel as bare strings even though `NodeSpec` already
   declares them
@@ -423,6 +369,13 @@ Potential later core work:
 - meaningful join semantics
 - future START-edge support if `Workflow.start` changes
 
+### Draft `route` Sugar
+
+Current `wf_authoring.route()` already has a specific equality/boolean routing
+surface. The draft layer should not invent a richer JSON route language ahead of
+Python authoring. Add draft route sugar later, after the shape is chosen
+deliberately for both front doors.
+
 ## Prototype Replacement
 
 The current draft prototype should be deleted or replaced directly.
@@ -441,8 +394,7 @@ Tests should cover:
 1. verb-key validation: exactly one step-kind key
 2. keyed-step compilation into core nodes
 3. `routes` compilation into core edges
-4. `route` lowering into current condition nodes/edges
+4. saved capability refs passing through `use`
-5. saved capability refs passing through `use`
+5. patching by stable ids
-6. patching by stable ids
+6. artifact creation from the new draft surface
-7. artifact creation from the new draft surface
+7. diagnostics with stable draft paths
-8. diagnostics with stable draft paths

docs/wf_mcp_end_to_end_runbook.md

@@ -257,11 +257,9 @@ arguments:
       "required": ["echoed"]
     },
     "start": "echo",
-    "steps": [
+    "steps": {
-      {
+      "echo": {
-        "id": "echo",
+        "use": "demo.personal.echo_tool",
-        "kind": "use",
-        "capability": "demo.personal.echo_tool",
         "in": {
           "input.text": "text"
         },
@@ -269,14 +267,12 @@ arguments:
           "echoed": "state.echoed"
         }
       }
-    ],
+    },
-    "edges": [
+    "routes": {
-      {
+      "echo": {
-        "from": "echo",
+        "ok": "__end__"
-        "outcome": "ok",
-        "to": "__end__"
       }
-    ]
+    }
   }
 }
 ```

docs/workflow_drafts.md

@@ -64,11 +64,9 @@ A minimal draft looks like this:
     "required": ["echoed"]
   },
   "start": "echo",
-  "steps": [
+  "steps": {
-    {
+    "echo": {
-      "id": "echo",
+      "use": "demo.personal.echo_tool",
-      "kind": "use",
-      "capability": "demo.personal.echo_tool",
       "in": {
         "input.text": "text"
       },
@@ -76,22 +74,20 @@ A minimal draft looks like this:
         "echoed": "state.echoed"
       }
     }
-  ],
+  },
-  "edges": [
+  "routes": {
-    {
+    "echo": {
-      "from": "echo",
+      "ok": "__end__"
-      "outcome": "ok",
-      "to": "__end__"
     }
-  ]
+  }
 }
 ```
 
 Important details:
 
-- `steps[].id` is required because JSON has no Python object identity.
+- `steps` are keyed by stable ids so patches do not depend on array positions.
-- `start` names a step id.
+- `start` names one step id.
-- `edges[].to` can name another step id or `__end__`.
+- `routes` map step outcomes to another step id or `__end__`.
 - `capability` may be concrete during exploration, such as
   `demo.personal.echo_tool`.
 - When saved with source bindings, concrete refs can be normalized to logical
@@ -105,9 +101,7 @@ Calls a workflow capability.
 
 ```json
 {
-  "id": "echo",
+  "use": "demo.personal.echo_tool",
-  "kind": "use",
-  "capability": "demo.personal.echo_tool",
   "in": {
     "input.text": "text"
   },
@@ -120,36 +114,18 @@ Calls a workflow capability.
 Use this for normal node calls, including generated workflow wrappers around
 MCP tools and local `wf.std` capabilities.
 
-### `condition`
-
-Evaluates a condition and routes by outcome.
-
-```json
-{
-  "id": "has_text",
-  "kind": "condition",
-  "check": {
-    "op": "exists",
-    "args": ["input.text"]
-  }
-}
-```
-
-Condition nodes compile to graph nodes with condition semantics. Their outgoing
-edges should use condition outcomes such as `true` and `false`.
-
 ### `foreach`
 
 Runs a child body over items.
 
 ```json
 {
-  "id": "each_item",
+  "foreach": {
-  "kind": "foreach",
+    "over": "state.items",
-  "over": "state.items",
+    "as": "item",
-  "as": "item",
+    "mode": "serial",
-  "mode": "serial",
+    "on_item_error": "fail"
-  "on_item_error": "fail"
+  }
 }
 ```
 
@@ -162,16 +138,16 @@ Declares an interrupting step.
 
 ```json
 {
-  "id": "ask_user",
+  "interrupt": {
-  "kind": "interrupt",
+    "kind": "input",
-  "interrupt_kind": "input",
+    "request": {
-  "request": {
+      "state.question": "question"
-    "state.question": "question"
+    },
-  },
+    "resume": {
-  "resume": {
+      "answer": "state.answer"
-    "answer": "state.answer"
+    },
-  },
+    "outcomes": ["resumed", "cancelled"]
-  "outcomes": ["resumed", "cancelled"]
+  }
 }
 ```
 
@@ -185,8 +161,7 @@ Joins control flow.
 
 ```json
 {
-  "id": "join_results",
+  "join": {}
-  "kind": "join"
 }
 ```
 
@@ -214,17 +189,13 @@ Example:
 [
   {
     "op": "replace",
-    "path": "/steps/0/in/input.text",
+    "path": "/steps/echo/in/input.text",
     "value": "message"
   },
   {
     "op": "add",
-    "path": "/edges/-",
+    "path": "/routes/echo/error",
-    "value": {
+    "value": "__end__"
-      "from": "echo",
-      "outcome": "error",
-      "to": "__end__"
-    }
   }
 ]
 ```

src/wf_artifacts/drafts.py

@@ -1,337 +0,0 @@
-from __future__ import annotations
-
-from copy import deepcopy
-from typing import Annotated, Any, Literal
-
-import jsonpatch
-from pydantic import BaseModel, Field, ValidationError
-
-from wf_core import (
-    END,
-    ConditionNode,
-    ForeachNode,
-    InterruptNode,
-    JoinNode,
-    NodeUse,
-    Workflow,
-)
-
-JsonObject = dict[str, Any]
-JsonPatch = list[dict[str, Any]]
-
-
-class DraftNodeUse(BaseModel):
-    """Authoring-friendly use of one workflow capability."""
-
-    id: str
-    kind: Literal["use"]
-    capability: str
-    desc: str | None = None
-    in_: dict[str, str] = Field(default_factory=dict, alias="in")
-    out: dict[str, str] = Field(default_factory=dict)
-    retry: int | None = Field(default=None, ge=0)
-    timeout_seconds: int | None = Field(default=None, gt=0)
-
-
-class DraftConditionNode(BaseModel):
-    """Authoring-friendly condition step."""
-
-    id: str
-    kind: Literal["condition"]
-    check: JsonObject
-
-
-class DraftForeachNode(BaseModel):
-    """Authoring-friendly foreach step."""
-
-    id: str
-    kind: Literal["foreach"]
-    over: str
-    as_: str = Field(alias="as")
-    mode: Literal["serial", "parallel"] = "serial"
-    on_item_error: Literal["fail", "collect", "skip"] = "fail"
-
-
-class DraftInterruptNode(BaseModel):
-    """Authoring-friendly interrupt step."""
-
-    id: str
-    kind: Literal["interrupt"]
-    interrupt_kind: str
-    request: dict[str, str] = Field(default_factory=dict)
-    resume: dict[str, str] = Field(default_factory=dict)
-    outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
-
-
-class DraftJoinNode(BaseModel):
-    """Authoring-friendly join step."""
-
-    id: str
-    kind: Literal["join"]
-
-
-DraftStep = Annotated[
-    DraftNodeUse
-    | DraftConditionNode
-    | DraftForeachNode
-    | DraftInterruptNode
-    | DraftJoinNode,
-    Field(discriminator="kind"),
-]
-"""Discriminated union of workflow draft steps."""
-
-
-class DraftEdge(BaseModel):
-    """Outcome-specific transition between draft steps."""
-
-    from_: str = Field(alias="from")
-    outcome: str
-    to: str
-
-
-class WorkflowDraft(BaseModel):
-    """LLM-friendly authoring shape that compiles into one raw workflow plan."""
-
-    name: str
-    input_schema: JsonObject
-    state_schema: JsonObject
-    output_schema: JsonObject
-    start: str
-    steps: list[DraftStep]
-    edges: list[DraftEdge]
-
-
-class DraftDiagnostic(BaseModel):
-    """Machine-readable reason a draft could not be compiled."""
-
-    code: str
-    path: str
-    step_id: str | None = None
-    message: str
-
-
-class DraftReferenceError(ValueError):
-    """Reference failure discovered after the draft shape itself is valid."""
-
-    def __init__(self, *, path: str, message: str, step_id: str | None = None) -> None:
-        super().__init__(message)
-        self.path = path
-        self.step_id = step_id
-
-
-class DraftValidationError(ValueError):
-    """Typed draft-shape failure with already-normalized authoring diagnostics."""
-
-    def __init__(self, diagnostic: DraftDiagnostic) -> None:
-        super().__init__(f"{diagnostic.path}: {diagnostic.message}")
-        self.diagnostic = diagnostic
-
-
-def compile_workflow_draft(draft: JsonObject) -> JsonObject:
-    """Compile the authoring draft into the normalized raw workflow plan."""
-    try:
-        parsed = WorkflowDraft.model_validate(draft)
-    except ValidationError as exc:
-        raise DraftValidationError(
-            _diagnostic_from_validation_error(exc, draft)
-        ) from exc
-    workflow = Workflow.model_validate(
-        {
-            "name": parsed.name,
-            "input_schema": deepcopy(parsed.input_schema),
-            "state_schema": deepcopy(parsed.state_schema),
-            "output_schema": deepcopy(parsed.output_schema),
-            "start": parsed.start,
-            "nodes": [
-                step.model_dump(mode="json", by_alias=True)
-                for step in _compile_steps(parsed.steps)
-            ],
-            "edges": [edge.model_dump(by_alias=True) for edge in parsed.edges],
-        }
-    )
-    _validate_graph_references(parsed)
-    return workflow.model_dump(mode="json", by_alias=True, exclude={"node_defs"})
-
-
-def validate_workflow_draft(draft: JsonObject) -> JsonObject:
-    """Return structured draft diagnostics instead of raising on bad input."""
-    try:
-        compiled_plan = compile_workflow_draft(draft)
-    except DraftValidationError as exc:
-        return _invalid_result(exc.diagnostic)
-    except DraftReferenceError as exc:
-        return _invalid_result(
-            DraftDiagnostic(
-                code="draft_invalid",
-                path=exc.path,
-                step_id=exc.step_id,
-                message=str(exc),
-            )
-        )
-    return {
-        "status": "valid",
-        "diagnostics": [],
-        "compiled_plan": compiled_plan,
-    }
-
-
-def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
-    """Apply RFC 6902 JSON Patch to a draft, then validate the patched draft.
-
-    Patch authoring is intentionally draft-first. Compiled raw plans are compiler
-    output, so callers should patch the readable source document and recompile it.
-    """
-    try:
-        patched = jsonpatch.JsonPatch(patch).apply(deepcopy(draft), in_place=False)
-    except Exception as exc:
-        return _invalid_result(
-            DraftDiagnostic(
-                code="patch_invalid",
-                path="patch",
-                message=str(exc),
-            )
-        )
-    if not isinstance(patched, dict):
-        return _invalid_result(
-            DraftDiagnostic(
-                code="draft_not_object",
-                path="",
-                message="patched draft must be a JSON object",
-            )
-        )
-    result = validate_workflow_draft(patched)
-    return {"draft": patched, **result}
-
-
-def _compile_steps(
-    steps: list[DraftStep],
-) -> list[NodeUse | ConditionNode | ForeachNode | InterruptNode | JoinNode]:
-    return [_compile_step(step) for step in steps]
-
-
-def _compile_step(
-    step: DraftStep,
-) -> NodeUse | ConditionNode | ForeachNode | InterruptNode | JoinNode:
-    if isinstance(step, DraftNodeUse):
-        return NodeUse.model_validate(
-            {
-                "id": step.id,
-                "type": "node",
-                "node": step.capability,
-                "desc": step.desc,
-                "in_map": deepcopy(step.in_),
-                "out_map": deepcopy(step.out),
-                "retry": step.retry,
-                "timeout_seconds": step.timeout_seconds,
-            }
-        )
-    if isinstance(step, DraftConditionNode):
-        return ConditionNode.model_validate(
-            {
-                "id": step.id,
-                "type": "condition",
-                "check": deepcopy(step.check),
-            }
-        )
-    if isinstance(step, DraftForeachNode):
-        return ForeachNode.model_validate(
-            {
-                "id": step.id,
-                "type": "foreach",
-                "over": step.over,
-                "as": step.as_,
-                "mode": step.mode,
-                "on_item_error": step.on_item_error,
-            }
-        )
-    if isinstance(step, DraftInterruptNode):
-        return InterruptNode.model_validate(
-            {
-                "id": step.id,
-                "type": "interrupt",
-                "kind": step.interrupt_kind,
-                "request_map": deepcopy(step.request),
-                "out_map": deepcopy(step.resume),
-                "outcomes": deepcopy(step.outcomes),
-            }
-        )
-    return JoinNode.model_validate({"id": step.id, "type": "join"})
-
-
-def _validate_graph_references(draft: WorkflowDraft) -> None:
-    step_ids = [step.id for step in draft.steps]
-    step_id_set = set(step_ids)
-    if len(step_ids) != len(step_id_set):
-        raise DraftReferenceError(
-            path="steps",
-            message="steps contain duplicate ids",
-        )
-    if draft.start not in step_id_set:
-        raise DraftReferenceError(
-            path="start",
-            message=f"start references unknown step id {draft.start!r}",
-        )
-    for index, edge in enumerate(draft.edges):
-        if edge.from_ not in step_id_set:
-            raise DraftReferenceError(
-                path=f"edges[{index}].from",
-                message=f"edges[{index}].from references unknown step id {edge.from_!r}",
-            )
-        if edge.to != END and edge.to not in step_id_set:
-            raise DraftReferenceError(
-                path=f"edges[{index}].to",
-                message=f"edges[{index}].to references unknown step id {edge.to!r}",
-            )
-
-
-def _invalid_result(diagnostic: DraftDiagnostic) -> JsonObject:
-    return {
-        "status": "invalid",
-        "diagnostics": [diagnostic.model_dump(mode="json")],
-    }
-
-
-def _diagnostic_from_validation_error(
-    exc: ValidationError,
-    draft: JsonObject,
-) -> DraftDiagnostic:
-    first_error = exc.errors()[0]
-    path = _format_error_path(first_error["loc"])
-    return DraftDiagnostic(
-        code="draft_invalid",
-        path=path,
-        step_id=_step_id_for_path(draft, path),
-        message=first_error["msg"],
-    )
-
-
-def _format_error_path(location: tuple[object, ...]) -> str:
-    parts: list[str] = []
-    for part in location:
-        if isinstance(part, int):
-            parts[-1] = f"{parts[-1]}[{part}]"
-            continue
-        if part == "in_":
-            part = "in"
-        elif part == "as_":
-            part = "as"
-        if part in {"use", "condition", "foreach", "interrupt", "join"}:
-            continue
-        parts.append(str(part))
-    return ".".join(parts)
-
-
-def _step_id_for_path(draft: JsonObject, path: str) -> str | None:
-    if not path.startswith("steps["):
-        return None
-    index_text = path.removeprefix("steps[").split("]", 1)[0]
-    if not index_text.isdecimal():
-        return None
-    steps = draft.get("steps")
-    if not isinstance(steps, list):
-        return None
-    index = int(index_text)
-    if index >= len(steps) or not isinstance(steps[index], dict):
-        return None
-    step_id = steps[index].get("id")
-    return step_id if isinstance(step_id, str) else None

src/wf_artifacts/drafts/__init__.py

@@ -0,0 +1,27 @@
+from .adapter import build_workflow_from_draft
+from .api import (
+    DraftDiagnostic,
+    compile_workflow_draft,
+    patch_workflow_draft,
+    validate_workflow_draft,
+)
+from .models import (
+    DraftForeachStep,
+    DraftInterruptStep,
+    DraftJoinStep,
+    DraftUseStep,
+    WorkflowDraft,
+)
+
+__all__ = [
+    "DraftDiagnostic",
+    "DraftForeachStep",
+    "DraftInterruptStep",
+    "DraftJoinStep",
+    "DraftUseStep",
+    "WorkflowDraft",
+    "build_workflow_from_draft",
+    "compile_workflow_draft",
+    "patch_workflow_draft",
+    "validate_workflow_draft",
+]

src/wf_artifacts/drafts/adapter.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from wf_authoring import WorkflowBuilder
+from wf_core import JoinNode, Workflow
+
+from .models import (
+    DraftForeachStep,
+    DraftInterruptStep,
+    DraftJoinStep,
+    DraftStep,
+    DraftUseStep,
+    WorkflowDraft,
+)
+
+
+def build_workflow_from_draft(draft: WorkflowDraft) -> Workflow:
+    """Adapt one typed draft through `WorkflowBuilder` into a core workflow."""
+    builder = WorkflowBuilder(
+        name=draft.name,
+        input_schema=draft.input_schema,
+        state_schema=draft.state_schema,
+        output_schema=draft.output_schema,
+    )
+    step_refs = {
+        step_id: _add_step(builder, step_id, step)
+        for step_id, step in draft.steps.items()
+    }
+    builder.set_entry_point(step_refs[draft.start])
+    for source_id, routes in draft.routes.items():
+        for outcome, target in routes.items():
+            builder.connect(step_refs[source_id], outcome, target)
+    return builder.compile()
+
+
+def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
+    if isinstance(step, DraftUseStep):
+        return builder.use_ref(
+            step.use,
+            id=step_id,
+            in_map=step.in_,
+            out_map=step.out,
+            desc=step.desc,
+        )
+    if isinstance(step, DraftForeachStep):
+        return builder.foreach(
+            id=step_id,
+            over=step.foreach.over,
+            as_=step.foreach.as_,
+            mode=step.foreach.mode,
+            on_item_error=step.foreach.on_item_error,
+        )
+    if isinstance(step, DraftInterruptStep):
+        return builder.interrupt(
+            id=step_id,
+            kind=step.interrupt.kind,
+            request_map=step.interrupt.request,
+            out_map=step.interrupt.resume,
+            outcomes=step.interrupt.outcomes,
+        )
+    if isinstance(step, DraftJoinStep):
+        node = JoinNode(id=step_id, type="join")
+        builder.nodes.append(node)
+        return node
+    raise TypeError(f"unsupported draft step {type(step)!r}")

src/wf_artifacts/drafts/api.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from collections.abc import Callable
+from typing import Any
+
+import jsonpatch
+from pydantic import BaseModel, ValidationError
+
+from .adapter import build_workflow_from_draft
+from .models import WorkflowDraft
+
+JsonObject = dict[str, Any]
+JsonPatch = list[dict[str, Any]]
+OutcomeLookup = Callable[[str], tuple[str, ...] | None]
+
+
+class DraftDiagnostic(BaseModel):
+    """Machine-readable reason a keyed draft could not be compiled."""
+
+    code: str
+    path: str
+    step_id: str | None = None
+    message: str
+
+
+def compile_workflow_draft(draft: JsonObject) -> JsonObject:
+    """Compile a keyed draft through `WorkflowBuilder` into raw workflow JSON."""
+    parsed = WorkflowDraft.model_validate(draft)
+    workflow = build_workflow_from_draft(parsed)
+    return workflow.model_dump(mode="json", by_alias=True, exclude={"node_defs"})
+
+
+def validate_workflow_draft(
+    draft: JsonObject,
+    *,
+    outcome_lookup: OutcomeLookup | None = None,
+) -> JsonObject:
+    """Return structured diagnostics instead of raising on a bad keyed draft."""
+    try:
+        compiled_plan = compile_workflow_draft(draft)
+    except (ValidationError, KeyError, ValueError) as exc:
+        return _invalid_result(_diagnostic_from_exception(exc))
+    if outcome_lookup is not None:
+        diagnostic = _validate_known_outcomes(draft, outcome_lookup)
+        if diagnostic is not None:
+            return _invalid_result(diagnostic)
+    return {
+        "status": "valid",
+        "diagnostics": [],
+        "compiled_plan": compiled_plan,
+    }
+
+
+def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
+    """Patch the draft source document, then validate the patched result."""
+    try:
+        patched = jsonpatch.JsonPatch(patch).apply(deepcopy(draft), in_place=False)
+    except Exception as exc:
+        return _invalid_result(
+            DraftDiagnostic(
+                code="patch_invalid",
+                path="patch",
+                message=str(exc),
+            )
+        )
+    if not isinstance(patched, dict):
+        return _invalid_result(
+            DraftDiagnostic(
+                code="draft_not_object",
+                path="",
+                message="patched draft must be a JSON object",
+            )
+        )
+    result = validate_workflow_draft(patched)
+    return {"draft": patched, **result}
+
+
+def _invalid_result(diagnostic: DraftDiagnostic) -> JsonObject:
+    return {
+        "status": "invalid",
+        "diagnostics": [diagnostic.model_dump(mode="json")],
+    }
+
+
+def _diagnostic_from_exception(exc: Exception) -> DraftDiagnostic:
+    if isinstance(exc, ValidationError):
+        error = exc.errors()[0]
+        return DraftDiagnostic(
+            code="draft_invalid",
+            path=_format_location(error["loc"]),
+            message=error["msg"],
+        )
+    return DraftDiagnostic(
+        code="draft_invalid",
+        path="",
+        message=str(exc),
+    )
+
+
+def _format_location(location: tuple[object, ...]) -> str:
+    return ".".join(str(part) for part in location)
+
+
+def _validate_known_outcomes(
+    draft: JsonObject,
+    outcome_lookup: OutcomeLookup,
+) -> DraftDiagnostic | None:
+    steps = draft.get("steps")
+    routes = draft.get("routes")
+    if not isinstance(steps, dict) or not isinstance(routes, dict):
+        return None
+    for step_id, route_map in routes.items():
+        step = steps.get(step_id)
+        if not isinstance(step, dict) or not isinstance(route_map, dict):
+            continue
+        capability = step.get("use")
+        if not isinstance(capability, str):
+            continue
+        outcomes = outcome_lookup(capability)
+        if outcomes is None:
+            continue
+        known_outcomes = set(outcomes)
+        for outcome in route_map:
+            if outcome not in known_outcomes:
+                return DraftDiagnostic(
+                    code="unknown_outcome",
+                    path=f"routes.{step_id}.{outcome}",
+                    step_id=step_id,
+                    message=(
+                        f"step {step_id!r} routes unknown outcome {outcome!r}; "
+                        f"expected one of {sorted(known_outcomes)!r}"
+                    ),
+                )
+    return None

src/wf_artifacts/drafts/models.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+JsonObject = dict[str, Any]
+STEP_KIND_KEYS = frozenset({"use", "foreach", "interrupt", "join"})
+
+
+class DraftUseStep(BaseModel):
+    """Draft step that calls one externally resolvable workflow capability."""
+
+    use: str
+    in_: dict[str, str] = Field(default_factory=dict, alias="in")
+    out: dict[str, str] = Field(default_factory=dict)
+    desc: str | None = None
+    retry: int | None = Field(default=None, ge=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
+
+
+class DraftForeachPayload(BaseModel):
+    """Payload for one draft foreach step."""
+
+    over: str
+    as_: str = Field(alias="as")
+    mode: Literal["serial", "parallel"] = "serial"
+    on_item_error: Literal["fail", "collect", "skip"] = "fail"
+
+
+class DraftForeachStep(BaseModel):
+    """Draft step that delegates foreach construction to `WorkflowBuilder`."""
+
+    foreach: DraftForeachPayload
+
+
+class DraftInterruptPayload(BaseModel):
+    """Payload for one draft interrupt step."""
+
+    kind: str
+    request: dict[str, str] = Field(default_factory=dict)
+    resume: dict[str, str] = Field(default_factory=dict)
+    outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
+
+
+class DraftInterruptStep(BaseModel):
+    """Draft step that pauses execution and waits for resume input."""
+
+    interrupt: DraftInterruptPayload
+
+
+class DraftJoinStep(BaseModel):
+    """Draft step that emits the current core join node."""
+
+    join: JsonObject = Field(default_factory=dict)
+
+
+DraftStep = DraftUseStep | DraftForeachStep | DraftInterruptStep | DraftJoinStep
+
+
+class WorkflowDraft(BaseModel):
+    """Patch-friendly JSON authoring document for one workflow graph."""
+
+    name: str
+    input_schema: JsonObject
+    state_schema: JsonObject
+    output_schema: JsonObject
+    start: str
+    steps: dict[str, DraftStep]
+    routes: dict[str, dict[str, str]] = Field(default_factory=dict)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _validate_step_kinds(cls, value: object) -> object:
+        if not isinstance(value, dict):
+            return value
+        steps = value.get("steps")
+        if not isinstance(steps, dict):
+            return value
+        for step_id, payload in steps.items():
+            if not isinstance(payload, dict):
+                continue
+            present = STEP_KIND_KEYS.intersection(payload)
+            if len(present) != 1:
+                raise ValueError(
+                    f"steps.{step_id} must contain exactly one step kind key"
+                )
+        return value

src/wf_authoring/builder/core.py

@@ -92,6 +92,33 @@ class WorkflowBuilder:
         self.nodes.append(node)
         return node
 
+    def use_ref(
+        self,
+        name: str,
+        *,
+        id: str | None = None,
+        in_map: MapArg | None = None,
+        out_map: MapArg | None = None,
+        desc: str | None = None,
+    ) -> NodeUse:
+        """Use an already-named external capability without a local `NodeSpec`.
+
+        `use()` is for callable-backed Python specs that can contribute a local
+        node definition and registry handler. `use_ref()` is the matching escape
+        hatch for MCP/saved-workflow capability refs that are resolved later by
+        the environment runner into node definitions and registry handlers.
+        """
+        node = NodeUse(
+            id=id or self._next_step_id(slug_id(name)),
+            type="node",
+            node=name,
+            desc=desc,
+            in_map=normalize_mapping(in_map),
+            out_map=normalize_mapping(out_map),
+        )
+        self.nodes.append(node)
+        return node
+
     def _next_step_id(self, base: str) -> str:
         """Return a stable unused step id based on the requested base name."""
         return next_step_id(base, cast(list[StepRef], self.nodes))

src/wf_authoring/builder/refs.py

@@ -3,11 +3,13 @@ from __future__ import annotations
 from dataclasses import dataclass
 from typing import Any, TypeAlias, TypeGuard
 
-from wf_core import ConditionNode, ForeachNode, InterruptNode, NodeUse
+from wf_core import ConditionNode, ForeachNode, InterruptNode, JoinNode, NodeUse
 
 from ..nodes import NodeSpec
 
-StepRef: TypeAlias = str | NodeUse | ConditionNode | ForeachNode | InterruptNode
+StepRef: TypeAlias = (
+    str | NodeUse | ConditionNode | ForeachNode | InterruptNode | JoinNode
+)
 BranchRef: TypeAlias = StepRef | NodeSpec[Any, Any]
 
 

src/wf_mcp/workflow_surface/handlers.py

@@ -254,7 +254,10 @@ class WorkflowSurfaceHandlers:
         }
 
     async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
-        return validate_workflow_draft(draft)
+        return validate_workflow_draft(
+            draft,
+            outcome_lookup=self._outcomes_for_capability,
+        )
 
     async def compile_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
         plan = compile_workflow_draft(draft)
@@ -336,6 +339,12 @@ class WorkflowSurfaceHandlers:
     ) -> dict[str, Any]:
         return patch_workflow_draft(draft, patch)
 
+    def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
+        try:
+            return self.service._get_qualified_spec(qualified_name).outcomes
+        except KeyError:
+            return None
+
     async def inspect_artifact(
         self, *, artifact_id: str, version: int
     ) -> dict[str, Any]:

tests/artifacts/test_draft_adapter.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from wf_artifacts.drafts import WorkflowDraft
+from wf_artifacts.drafts.adapter import build_workflow_from_draft
+from wf_core import NodeUse
+
+
+def test_adapter_lowers_keyed_use_steps_and_routes_through_builder() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "echo",
+            "input_schema": {},
+            "state_schema": {"fields": {}},
+            "output_schema": {},
+            "start": "echo",
+            "steps": {"echo": {"use": "demo.echo"}},
+            "routes": {"echo": {"ok": "__end__"}},
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+
+    node = workflow.nodes[0]
+    assert isinstance(node, NodeUse)
+    assert node.id == "echo"
+    assert node.node == "demo.echo"
+    assert workflow.edges[0].from_ == "echo"
+    assert workflow.edges[0].outcome == "ok"
+    assert workflow.edges[0].to == "__end__"

tests/artifacts/test_draft_api.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from wf_artifacts.drafts import compile_workflow_draft, patch_workflow_draft
+
+
+def test_compile_workflow_draft_returns_raw_core_shape() -> None:
+    plan = compile_workflow_draft(_keyed_echo_draft())
+
+    assert plan["nodes"][0]["id"] == "echo"
+    assert plan["nodes"][0]["node"] == "demo.echo"
+    assert plan["edges"][0]["outcome"] == "ok"
+
+
+def test_patch_workflow_draft_uses_stable_step_paths() -> None:
+    result = patch_workflow_draft(
+        _keyed_echo_draft(),
+        [
+            {
+                "op": "replace",
+                "path": "/steps/echo/in/input.text",
+                "value": "message",
+            }
+        ],
+    )
+
+    assert result["status"] == "valid"
+    assert result["draft"]["steps"]["echo"]["in"]["input.text"] == "message"
+
+
+def _keyed_echo_draft() -> dict[str, object]:
+    return {
+        "name": "echo",
+        "input_schema": {},
+        "state_schema": {"fields": {}},
+        "output_schema": {},
+        "start": "echo",
+        "steps": {
+            "echo": {
+                "use": "demo.echo",
+                "in": {"input.text": "text"},
+                "out": {"echoed": "state.echoed"},
+            }
+        },
+        "routes": {"echo": {"ok": "__end__"}},
+    }

tests/artifacts/test_draft_models.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from wf_artifacts.drafts import DraftUseStep, WorkflowDraft
+
+
+def test_workflow_draft_uses_keyed_steps() -> None:
+    draft = WorkflowDraft.model_validate(_keyed_echo_draft())
+
+    assert isinstance(draft.steps["echo"], DraftUseStep)
+    assert draft.steps["echo"].use == "demo.echo"
+
+
+def test_draft_step_requires_exactly_one_kind_key() -> None:
+    draft = _keyed_echo_draft()
+    steps = draft["steps"]
+    assert isinstance(steps, dict)
+    echo = steps["echo"]
+    assert isinstance(echo, dict)
+    echo["join"] = {}
+
+    with pytest.raises(ValidationError) as exc_info:
+        WorkflowDraft.model_validate(draft)
+
+    assert "steps.echo" in str(exc_info.value)
+
+
+def _keyed_echo_draft() -> dict[str, object]:
+    return {
+        "name": "echo",
+        "input_schema": {},
+        "state_schema": {"fields": {}},
+        "output_schema": {},
+        "start": "echo",
+        "steps": {
+            "echo": {
+                "use": "demo.echo",
+                "in": {"input.text": "text"},
+                "out": {"echoed": "state.echoed"},
+            }
+        },
+        "routes": {"echo": {"ok": "__end__"}},
+    }

tests/artifacts/test_drafts.py

@@ -1,160 +0,0 @@
-from __future__ import annotations
-
-import pytest
-
-from wf_artifacts.drafts import (
-    DraftNodeUse,
-    WorkflowDraft,
-    compile_workflow_draft,
-    patch_workflow_draft,
-    validate_workflow_draft,
-)
-
-
-def test_compile_draft_maps_use_step_to_raw_node_use() -> None:
-    plan = compile_workflow_draft(_draft_with_steps([_use_step()]))
-
-    assert plan["name"] == "echo_probe"
-    assert plan["start"] == "echo"
-    assert plan["nodes"][0]["id"] == "echo"
-    assert plan["nodes"][0]["type"] == "node"
-    assert plan["nodes"][0]["node"] == "everything.echo"
-    assert plan["nodes"][0]["in_map"]["input.message"] == "message"
-    assert plan["nodes"][0]["out_map"]["content"] == "state.content"
-    assert plan["edges"][0]["to"] == "__end__"
-
-
-def test_workflow_draft_uses_concrete_authoring_models() -> None:
-    draft = WorkflowDraft.model_validate(_draft_with_steps([_use_step()]))
-
-    assert isinstance(draft.steps[0], DraftNodeUse)
-    assert draft.steps[0].capability == "everything.echo"
-
-
-def test_compile_draft_maps_condition_foreach_interrupt_and_join() -> None:
-    plan = compile_workflow_draft(
-        _draft_with_steps(
-            [
-                {
-                    "id": "route",
-                    "kind": "condition",
-                    "check": {"op": "exists", "path": "input.items"},
-                },
-                {
-                    "id": "each",
-                    "kind": "foreach",
-                    "over": "input.items",
-                    "as": "item",
-                    "mode": "serial",
-                    "on_item_error": "collect",
-                },
-                {
-                    "id": "approval",
-                    "kind": "interrupt",
-                    "interrupt_kind": "approval",
-                    "request": {"input.message": "message"},
-                    "resume": {"approved": "state.approved"},
-                    "outcomes": ["submitted"],
-                },
-                {"id": "joined", "kind": "join"},
-            ],
-            edges=[
-                {"from": "route", "outcome": "true", "to": "each"},
-                {"from": "each", "outcome": "done", "to": "approval"},
-                {"from": "approval", "outcome": "submitted", "to": "joined"},
-                {"from": "joined", "outcome": "done", "to": "__end__"},
-            ],
-            start="route",
-        )
-    )
-
-    assert plan["nodes"][0]["type"] == "condition"
-    assert plan["nodes"][0]["check"]["op"] == "exists"
-    assert plan["nodes"][1]["type"] == "foreach"
-    assert plan["nodes"][1]["as"] == "item"
-    assert plan["nodes"][1]["on_item_error"] == "collect"
-    assert plan["nodes"][2]["type"] == "interrupt"
-    assert plan["nodes"][2]["kind"] == "approval"
-    assert plan["nodes"][2]["request_map"]["input.message"] == "message"
-    assert plan["nodes"][2]["out_map"]["approved"] == "state.approved"
-    assert plan["nodes"][3]["type"] == "join"
-    assert plan["edges"][3]["from"] == "joined"
-
-
-def test_compile_draft_requires_explicit_step_ids() -> None:
-    draft = _draft_with_steps([{"kind": "join"}])
-
-    with pytest.raises(ValueError) as exc_info:
-        compile_workflow_draft(draft)
-
-    assert "steps[0].id" in str(exc_info.value)
-
-
-def test_validate_draft_returns_structured_step_diagnostic() -> None:
-    draft = _draft_with_steps([{"id": "each", "kind": "foreach", "over": "items"}])
-
-    result = validate_workflow_draft(draft)
-
-    assert result["status"] == "invalid"
-    assert result["diagnostics"][0]["path"] == "steps[0].as"
-    assert result["diagnostics"][0]["step_id"] == "each"
-
-
-def test_patch_workflow_draft_applies_json_patch_and_validates_result() -> None:
-    patched = patch_workflow_draft(
-        _draft_with_steps([_use_step()]),
-        [
-            {
-                "op": "replace",
-                "path": "/steps/0/in/input.message",
-                "value": "text",
-            },
-            {
-                "op": "add",
-                "path": "/edges/-",
-                "value": {"from": "echo", "outcome": "error", "to": "__end__"},
-            },
-        ],
-    )
-
-    assert patched["status"] == "valid"
-    assert patched["draft"]["steps"][0]["in"]["input.message"] == "text"
-    assert patched["compiled_plan"]["edges"][1]["outcome"] == "error"
-
-
-def test_patch_workflow_draft_reports_invalid_patch_without_partial_result() -> None:
-    patched = patch_workflow_draft(
-        _draft_with_steps([_use_step()]),
-        [{"op": "replace", "path": "/steps/99/id", "value": "missing"}],
-    )
-
-    assert patched["status"] == "invalid"
-    assert patched["diagnostics"][0]["code"] == "patch_invalid"
-    assert "draft" not in patched
-
-
-def _draft_with_steps(
-    steps: list[dict[str, object]],
-    *,
-    edges: list[dict[str, str]] | None = None,
-    start: str = "echo",
-) -> dict[str, object]:
-    return {
-        "name": "echo_probe",
-        "input_schema": {"type": "object", "properties": {}},
-        "state_schema": {"fields": {"content": {"type": "string"}}},
-        "output_schema": {"type": "object", "properties": {}},
-        "start": start,
-        "steps": steps,
-        "edges": edges or [{"from": "echo", "outcome": "ok", "to": "__end__"}],
-    }
-
-
-def _use_step() -> dict[str, object]:
-    return {
-        "id": "echo",
-        "kind": "use",
-        "capability": "everything.echo",
-        "in": {"input.message": "message"},
-        "out": {"content": "state.content"},
-    }

tests/authoring/test_builder.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import pytest
 
 from wf_authoring import WorkflowBuilder, state
-from wf_core import RunStatus, WorkflowExecutionError
+from wf_core import END, RunStatus, WorkflowExecutionError
 
 from tests.authoring.helpers import (
     AutoBindInput,
@@ -187,3 +187,27 @@ def test_builder_connect_can_use_node_specs_and_returns_resolved_refs() -> None:
     assert builder.edges[0].from_ == "test_auto_bind"
     assert builder.edges[0].outcome == "ok"
     assert builder.edges[0].to == "test_auto_bind_2"
+
+
+def test_builder_use_ref_creates_external_node_use_without_node_def() -> None:
+    builder = WorkflowBuilder(
+        name="external_ref_demo",
+        input_schema={},
+        state_schema={"fields": {}},
+        output_schema={},
+    )
+
+    step = builder.use_ref(
+        "demo.echo",
+        id="echo",
+        in_map={"input.text": "text"},
+        out_map={"echoed": "state.echoed"},
+    )
+    builder.set_entry_point(step)
+    builder.connect(step, "ok", END)
+    workflow = builder.compile()
+
+    assert step.node == "demo.echo"
+    assert step.in_map["input.text"] == "text"
+    assert step.out_map["echoed"] == "state.echoed"
+    assert workflow.node_defs == []

tests/wf_mcp/test_workflow_surface.py

@@ -224,6 +224,30 @@ def test_workflow_surface_validates_draft_without_saving() -> None:
     assert artifact_store.list_artifacts() == []
 
 
+def test_workflow_surface_rejects_unknown_draft_route_outcome_when_spec_is_known() -> (
+    None
+):
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_draft_bad_outcome"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_draft_bad_outcome_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+    draft = _echo_draft()
+    draft["routes"]["echo"] = {"typo": "__end__"}
+
+    payload = asyncio.run(handlers.validate_draft(draft=draft))
+
+    assert payload["status"] == "invalid"
+    assert payload["diagnostics"][0]["path"] == "routes.echo.typo"
+
+
 def test_workflow_surface_creates_artifact_from_draft_with_binding_suggestions() -> (
     None
 ):
@@ -240,7 +264,7 @@ def test_workflow_surface_creates_artifact_from_draft_with_binding_suggestions()
     service.register_specs("demo.personal", echo_tool)
     handlers = WorkflowSurfaceHandlers(service)
     draft = _echo_draft()
-    draft["steps"][0]["capability"] = "demo.personal.echo_tool"
+    draft["steps"]["echo"]["use"] = "demo.personal.echo_tool"
 
     payload = asyncio.run(
         handlers.create_artifact_from_draft(
@@ -307,7 +331,7 @@ def test_workflow_surface_patches_draft_without_saving() -> None:
             patch=[
                 {
                     "op": "replace",
-                    "path": "/steps/0/in/input.text",
+                    "path": "/steps/echo/in/input.text",
                     "value": "message",
                 }
             ],
@@ -315,7 +339,7 @@ def test_workflow_surface_patches_draft_without_saving() -> None:
     )
 
     assert payload["status"] == "valid"
-    assert payload["draft"]["steps"][0]["in"]["input.text"] == "message"
+    assert payload["draft"]["steps"]["echo"]["in"]["input.text"] == "message"
     assert artifact_store.list_artifacts() == []
 
 
@@ -719,16 +743,14 @@ def _echo_draft() -> dict[str, Any]:
             "required": ["echoed"],
         },
         "start": "echo",
-        "steps": [
+        "steps": {
-            {
+            "echo": {
-                "id": "echo",
+                "use": "demo.personal.echo_tool",
-                "kind": "use",
-                "capability": "demo.personal.echo_tool",
                 "in": {"input.text": "text"},
                 "out": {"echoed": "state.echoed"},
             }
-        ],
+        },
-        "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
+        "routes": {"echo": {"ok": "__end__"}},
     }