draft 2

docs/superpowers/specs/2026-05-18-typed-workflow-drafts-design.md

@@ -0,0 +1,100 @@
+# Typed Workflow Drafts Design
+
+## Goal
+
+Keep `WorkflowDraft` as the LLM-friendly authoring format without letting it
+become a second workflow system.
+
+Drafts should:
+
+- have concrete Pydantic models at the authoring seam
+- compile into the existing raw/core workflow models
+- reuse existing workflow validation rather than reimplementing graph rules
+- return structured draft diagnostics without parsing exception strings
+
+## Model Roles
+
+### `WorkflowDraft`
+
+Authoring model.
+
+It owns the intentionally nicer JSON shape:
+
+- `steps` instead of raw `nodes`
+- `kind="use"` plus `capability`
+- `in` / `out`
+- `interrupt_kind`, `request`, `resume`
+
+This model exists because authoring wants a more legible surface than the raw
+runtime plan.
+
+### `RawWorkflowPlan`
+
+Low-level transport model for MCP callers that already have a normalized plan.
+
+It reuses core `Step` and `Edge`, but currently keeps boundary schemas as plain
+JSON objects. It is not a second workflow language. Converting it into
+`Workflow` is nearly a validation/coercion step:
+
+```text
+RawWorkflowPlan.model_dump(...) -> Workflow.model_validate(...)
+```
+
+### `Workflow`
+
+Core runtime model.
+
+It remains the only normalized graph model consumed by runtime validation and
+execution.
+
+## Chosen Design
+
+Add concrete draft models in `wf_artifacts.drafts`:
+
+- `WorkflowDraft`
+- discriminated `DraftStep` variants
+- `DraftEdge`
+- `DraftDiagnostic`
+
+Compilation flow:
+
+```text
+dict payload
+  -> WorkflowDraft.model_validate(...)
+  -> RawWorkflowPlan built from draft models
+  -> Workflow.model_validate(...)
+  -> existing artifact factory validation on save
+```
+
+The compiler only translates the genuine authoring differences. It does not own
+workflow graph semantics.
+
+## Error Handling
+
+Use Pydantic error locations and explicit draft errors. Do not infer paths by
+parsing exception message text.
+
+Expected diagnostic fields:
+
+- `code`
+- `path`
+- `step_id` when the failing step is known
+- `message`
+
+Patch failures remain their own `patch_invalid` diagnostic.
+
+## Non-Goals
+
+- changing workflow semantics
+- changing the raw plan escape hatch
+- adding new step kinds
+- replacing existing core validation
+
+## Testing
+
+Add tests that prove:
+
+- draft payloads validate into concrete models
+- compilation still returns the current raw plan shape
+- structured diagnostics identify the failing path and step id
+- graph validation remains delegated to the existing normalized workflow path

docs/wf_mcp_end_to_end_runbook.md

@@ -219,7 +219,7 @@ Create a one-node workflow that:
 - ends on the node's `ok` outcome
 
 ```yaml
-tool: wf.workflow.create_artifact_from_plan
+tool: wf.workflow.create_artifact_from_draft
 arguments:
 {
   "artifact_id": "echo",
@@ -229,7 +229,7 @@ arguments:
   "source_bindings": {
     "demo": "demo.personal"
   },
-  "plan": {
+  "draft": {
     "name": "echo",
     "input_schema": {
       "type": "object",
@@ -257,15 +257,15 @@ arguments:
       "required": ["echoed"]
     },
     "start": "echo",
-    "nodes": [
+    "steps": [
       {
         "id": "echo",
-        "type": "node",
+        "kind": "use",
-        "node": "demo.personal.echo_tool",
+        "capability": "demo.personal.echo_tool",
-        "in_map": {
+        "in": {
           "input.text": "text"
         },
-        "out_map": {
+        "out": {
           "echoed": "state.echoed"
         }
       }
@@ -287,6 +287,7 @@ Important behavior:
   is normalized to logical node ref `demo.echo_tool`
 - the saved dependency contract records what was observed from
   `demo.personal.echo_tool` at creation time
+- the draft is compiled into a raw workflow plan and validated before saving
 
 Inspect the saved result if needed:
 
@@ -421,7 +422,8 @@ wf.admin.list_sources
 wf.workflow.list_capabilities
 wf.workflow.inspect_capability
 wf.workflow.call_capability
-wf.workflow.create_artifact_from_plan
+wf.workflow.validate_draft
+wf.workflow.create_artifact_from_draft
 wf.workflow.save_deployment
 wf.workflow.validate_deployment
 wf.workflow.run_deployment
@@ -450,6 +452,11 @@ A tools-only server should still refresh successfully.
 That is the raw-tool versus workflow-capability distinction. Use or build a
 workflow-facing wrapper when the raw tool's shape is provider-centric.
 
+### The draft is close but has one wrong field
+
+Use `wf.workflow.patch_draft` instead of asking the client to rewrite the whole
+workflow. Draft patching uses JSON Patch and revalidates the patched result.
+
 ### The deployment used to run but now fails validation
 
 Likely causes:
@@ -474,5 +481,6 @@ wf.workflow.validate_deployment
   discovery and deployment failures
 - [`workflow_capabilities.md`](workflow_capabilities.md) for raw tool versus
   workflow capability
+- [`workflow_drafts.md`](workflow_drafts.md) for the preferred authoring format
 - [`workflow_artifacts.md`](workflow_artifacts.md) for immutable artifacts,
   deployments, and dependency contracts

docs/wf_mcp_operator_manual.md

@@ -75,7 +75,8 @@ Use it for:
 
 - discovering workflow-ready capabilities
 - inspecting and directly test-calling one capability
-- creating saved artifacts
+- validating, patching, and compiling workflow drafts
+- creating saved artifacts from drafts or raw plans
 - listing and inspecting saved artifacts
 - saving deployments
 - validating and running deployments
@@ -85,6 +86,10 @@ Typical tools:
 - `wf.workflow.list_capabilities`
 - `wf.workflow.inspect_capability`
 - `wf.workflow.call_capability`
+- `wf.workflow.validate_draft`
+- `wf.workflow.compile_draft`
+- `wf.workflow.patch_draft`
+- `wf.workflow.create_artifact_from_draft`
 - `wf.workflow.create_artifact_from_plan`
 - `wf.workflow.list_artifacts`
 - `wf.workflow.inspect_artifact`
@@ -212,12 +217,18 @@ graph would consume.
 ### 4. Build And Save
 
 ```text
-wf.workflow.create_artifact_from_plan
+wf.workflow.validate_draft
+wf.workflow.create_artifact_from_draft
 wf.workflow.save_deployment
 ```
 
-Artifacts should prefer logical source aliases in saved plans. Deployments bind
+Drafts are the preferred interactive authoring format. They compile into raw
-those logical aliases to concrete sources such as `context7.default`.
+workflow plans before saving. Use `create_artifact_from_plan` only when a caller
+already has a compiled raw plan or intentionally wants the low-level escape
+hatch.
+
+Artifacts should prefer logical source aliases in saved workflows. Deployments
+bind those logical aliases to concrete sources such as `context7.default`.
 
 ### 5. Validate And Run
 
@@ -244,7 +255,11 @@ schemas mid-session.
 | Find one workflow-ready node | `wf.workflow.list_capabilities` |
 | Read one node contract in full | `wf.workflow.inspect_capability` |
 | Test one node directly | `wf.workflow.call_capability` |
-| Save a workflow definition | `wf.workflow.create_artifact_from_plan` |
+| Validate an authored workflow draft | `wf.workflow.validate_draft` |
+| Apply a targeted fix to a draft | `wf.workflow.patch_draft` |
+| Compile a draft without saving | `wf.workflow.compile_draft` |
+| Save a workflow definition from a draft | `wf.workflow.create_artifact_from_draft` |
+| Save a compiled raw workflow definition | `wf.workflow.create_artifact_from_plan` |
 | List saved workflows/wrappers | `wf.workflow.list_artifacts` |
 | Bind a saved workflow to concrete sources | `wf.workflow.save_deployment` |
 | Check whether a deployment can run | `wf.workflow.validate_deployment` |
@@ -310,5 +325,6 @@ accounts.
   model
 - [`workflow_capabilities.md`](workflow_capabilities.md) for raw versus
   workflow-facing capability design
+- [`workflow_drafts.md`](workflow_drafts.md) for the preferred authoring format
 - [`workflow_artifacts.md`](workflow_artifacts.md) for immutable artifacts,
   deployments, and saved workflows as future nodes

docs/wf_mcp_troubleshooting.md

@@ -158,6 +158,9 @@ System-source self-bindings look redundant, but they mean "use the local
 standard source with the same id." Current deployments bind local and external
 sources through the same field.
 
+If the artifact was created from a draft, inspect the
+`create_artifact_from_draft` response for suggested local bindings.
+
 Use:
 
 ```text
@@ -323,6 +326,33 @@ wrapper when it:
 
 See [`workflow_capabilities.md`](workflow_capabilities.md).
 
+## A Draft Is Invalid Or Almost Correct
+
+Use the draft tools before saving:
+
+```text
+wf.workflow.validate_draft
+wf.workflow.patch_draft
+wf.workflow.compile_draft
+```
+
+`validate_draft` checks the authored draft and the compiled raw workflow plan.
+`patch_draft` applies RFC 6902 JSON Patch and validates the patched result.
+
+Prefer patching a draft over patching a raw plan. The draft is the authoring
+surface; the raw plan is the execution boundary.
+
+If a draft compiles but the deployment later fails with `binding_missing`, add
+the reported source binding to the deployment. Local system-source bindings can
+be explicit, for example:
+
+```json
+{
+  "wf.std": "wf.std",
+  "wf.mcp": "wf.mcp"
+}
+```
+
 ## MCP Resources Or Prompts Are Missing
 
 First ask whether the upstream server actually supports them.
@@ -370,3 +400,4 @@ whether the LLM harness can call it in the same session
   path
 - [`workflow_artifacts.md`](workflow_artifacts.md) for dependency diagnostics
   and drift policy
+- [`workflow_drafts.md`](workflow_drafts.md) for draft validation and patching

docs/workflow_capabilities.md

@@ -224,7 +224,13 @@ A client authoring workflows, including an LLM client, should be able to:
 3. inspect existing workflow capabilities and saved wrappers
 4. call a workflow capability directly once
 5. inspect the normalized output and outcome
-6. reuse that capability inside a graph
+6. author a workflow draft
+7. validate or patch the draft
+8. save the compiled workflow artifact
+
+Drafts are the preferred authoring format for this loop. See
+[`workflow_drafts.md`](workflow_drafts.md). Raw workflow plans remain an escape
+hatch for advanced clients and compiler outputs.
 
 Saved wrapper artifacts can be called with a deployment id when they use logical
 source names. The deployment supplies the concrete source bindings for that

docs/workflow_drafts.md

@@ -0,0 +1,281 @@
+# Workflow Drafts
+
+Workflow drafts are the preferred authoring format for LLM and human clients.
+
+They sit above the raw `Workflow` model:
+
+```text
+WorkflowDraft -> RawWorkflowPlan -> WorkflowArtifact -> Deployment
+```
+
+The draft format is intentionally explicit and patchable. It avoids asking an
+LLM client to write the full core model directly, while still compiling into the
+same validated workflow plan used by the runtime.
+
+Raw plans still exist as an escape hatch for advanced clients and compiler
+outputs. New authoring flows should normally start with drafts.
+
+## Why Drafts Exist
+
+The raw workflow model is normalized for execution. That makes it precise, but
+not always pleasant as an interactive authoring target.
+
+Drafts optimize for:
+
+- stable JSON shapes that are easy to inspect
+- explicit step ids instead of implicit Python object references
+- targeted fixes through JSON Patch
+- a clear place to validate before saving
+- preserving the existing raw workflow/runtime boundary
+
+Drafts do not change workflow semantics. They compile into the same core plan
+shape and then run through normal validation.
+
+## Draft Shape
+
+A minimal draft looks like this:
+
+```json
+{
+  "name": "echo",
+  "input_schema": {
+    "type": "object",
+    "properties": {
+      "text": {
+        "type": "string"
+      }
+    },
+    "required": ["text"]
+  },
+  "state_schema": {
+    "fields": {
+      "echoed": {
+        "type": "string"
+      }
+    }
+  },
+  "output_schema": {
+    "type": "object",
+    "properties": {
+      "echoed": {
+        "type": "string"
+      }
+    },
+    "required": ["echoed"]
+  },
+  "start": "echo",
+  "steps": [
+    {
+      "id": "echo",
+      "kind": "use",
+      "capability": "demo.personal.echo_tool",
+      "in": {
+        "input.text": "text"
+      },
+      "out": {
+        "echoed": "state.echoed"
+      }
+    }
+  ],
+  "edges": [
+    {
+      "from": "echo",
+      "outcome": "ok",
+      "to": "__end__"
+    }
+  ]
+}
+```
+
+Important details:
+
+- `steps[].id` is required because JSON has no Python object identity.
+- `start` names a step id.
+- `edges[].to` can name 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
+  refs such as `demo.echo_tool`.
+
+## Step Kinds
+
+### `use`
+
+Calls a workflow capability.
+
+```json
+{
+  "id": "echo",
+  "kind": "use",
+  "capability": "demo.personal.echo_tool",
+  "in": {
+    "input.text": "text"
+  },
+  "out": {
+    "echoed": "state.echoed"
+  }
+}
+```
+
+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",
+  "kind": "foreach",
+  "over": "state.items",
+  "as": "item",
+  "mode": "serial",
+  "on_item_error": "fail"
+}
+```
+
+Use `serial` unless the runtime explicitly supports a parallel async path for
+the target workflow.
+
+### `interrupt`
+
+Declares an interrupting step.
+
+```json
+{
+  "id": "ask_user",
+  "kind": "interrupt",
+  "interrupt_kind": "input",
+  "request": {
+    "state.question": "question"
+  },
+  "resume": {
+    "answer": "state.answer"
+  },
+  "outcomes": ["resumed", "cancelled"]
+}
+```
+
+Saved interrupting artifacts are still limited in the current execution
+surface. If a deployment reports `interrupting_artifact_unsupported`, that is a
+known platform limitation rather than a draft bug.
+
+### `join`
+
+Joins control flow.
+
+```json
+{
+  "id": "join_results",
+  "kind": "join"
+}
+```
+
+## Draft Tools
+
+The workflow MCP surface exposes these draft tools:
+
+| Tool | Purpose |
+| --- | --- |
+| `wf.workflow.validate_draft` | Validate draft shape and compiled workflow without saving. |
+| `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. |
+
+Use `validate_draft` before saving. Use `patch_draft` when an LLM client needs a
+small targeted correction instead of rewriting the whole workflow.
+
+## Patching Drafts
+
+`patch_draft` accepts JSON Patch operations.
+
+Example:
+
+```json
+[
+  {
+    "op": "replace",
+    "path": "/steps/0/in/input.text",
+    "value": "message"
+  },
+  {
+    "op": "add",
+    "path": "/edges/-",
+    "value": {
+      "from": "echo",
+      "outcome": "error",
+      "to": "__end__"
+    }
+  }
+]
+```
+
+Patch the draft, not the compiled raw plan. The raw plan is an implementation
+boundary and may be harder for an LLM client to repair correctly.
+
+## Saving And Running
+
+The normal path is:
+
+```text
+wf.workflow.validate_draft
+wf.workflow.create_artifact_from_draft
+wf.workflow.save_deployment
+wf.workflow.validate_deployment
+wf.workflow.run_deployment
+```
+
+`create_artifact_from_draft` may return suggested bindings for local system
+sources such as:
+
+```json
+{
+  "wf.std": "wf.std",
+  "wf.mcp": "wf.mcp"
+}
+```
+
+Keep those explicit when deployment validation reports `binding_missing`.
+
+## Raw Plan Escape Hatch
+
+Use `wf.workflow.create_artifact_from_plan` only when the caller already has a
+compiled raw workflow plan or is intentionally bypassing the draft layer.
+
+For normal interactive authoring, prefer:
+
+```text
+draft -> validate -> patch if needed -> create artifact from draft
+```
+
+That keeps errors local and gives the author a smaller object to reason about.
+
+## Read Next
+
+- [`wf_mcp_operator_manual.md`](wf_mcp_operator_manual.md) for the MCP-facing
+  tool family map
+- [`wf_mcp_end_to_end_runbook.md`](wf_mcp_end_to_end_runbook.md) for a complete
+  connection-to-run example
+- [`workflow_capabilities.md`](workflow_capabilities.md) for raw capability
+  versus workflow capability
+- [`workflow_artifacts.md`](workflow_artifacts.md) for artifacts, deployments,
+  and dependency contracts

src/wf_artifacts/drafts.py

@@ -1,34 +1,173 @@
 from __future__ import annotations
 
 from copy import deepcopy
-from typing import Any
+from typing import Annotated, Any, Literal
 
 import jsonpatch
-from pydantic import ValidationError
+from pydantic import BaseModel, Field, ValidationError
 
-from wf_core import Workflow
+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 LLM-friendly draft shape into the normalized workflow plan."""
+    """Compile the authoring draft into the normalized raw workflow plan."""
-    plan = _compile_unvalidated_draft(draft)
+    try:
-    _validate_plan_model(plan)
+        parsed = WorkflowDraft.model_validate(draft)
-    _validate_graph_references(plan)
+    except ValidationError as exc:
-    return plan
+        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 Exception as exc:
+    except DraftValidationError as exc:
-        return {
+        return _invalid_result(exc.diagnostic)
-            "status": "invalid",
+    except DraftReferenceError as exc:
-            "diagnostics": [_diagnostic_from_exception(exc)],
+        return _invalid_result(
-        }
+            DraftDiagnostic(
+                code="draft_invalid",
+                path=exc.path,
+                step_id=exc.step_id,
+                message=str(exc),
+            )
+        )
     return {
         "status": "valid",
         "diagnostics": [],
@@ -45,225 +184,154 @@ def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
     try:
         patched = jsonpatch.JsonPatch(patch).apply(deepcopy(draft), in_place=False)
     except Exception as exc:
-        return {
+        return _invalid_result(
-            "status": "invalid",
+            DraftDiagnostic(
-            "diagnostics": [
+                code="patch_invalid",
-                {
+                path="patch",
-                    "code": "patch_invalid",
+                message=str(exc),
-                    "path": "patch",
+            )
-                    "message": str(exc),
+        )
-                }
-            ],
-        }
     if not isinstance(patched, dict):
-        return {
+        return _invalid_result(
-            "status": "invalid",
+            DraftDiagnostic(
-            "diagnostics": [
+                code="draft_not_object",
-                {
+                path="",
-                    "code": "draft_not_object",
+                message="patched draft must be a JSON object",
-                    "path": "",
+            )
-                    "message": "patched draft must be a JSON object",
+        )
-                }
-            ],
-        }
     result = validate_workflow_draft(patched)
     return {"draft": patched, **result}
 
 
-def _compile_unvalidated_draft(draft: JsonObject) -> JsonObject:
+def _compile_steps(
-    if not isinstance(draft, dict):
+    steps: list[DraftStep],
-        raise ValueError("draft must be a JSON object")
+) -> list[NodeUse | ConditionNode | ForeachNode | InterruptNode | JoinNode]:
-
+    return [_compile_step(step) for step in steps]
-    plan: JsonObject = {
-        "name": _required_str(draft, "name"),
-        "input_schema": _required_object(draft, "input_schema"),
-        "state_schema": _required_object(draft, "state_schema"),
-        "output_schema": _required_object(draft, "output_schema"),
-        "start": _required_str(draft, "start"),
-        "nodes": [
-            _compile_step(step, index) for index, step in enumerate(_steps(draft))
-        ],
-        "edges": [
-            _compile_edge(edge, index) for index, edge in enumerate(_edges(draft))
-        ],
-    }
-    return plan
 
 
-def _compile_step(step: object, index: int) -> JsonObject:
+def _compile_step(
-    path = f"steps[{index}]"
+    step: DraftStep,
-    if not isinstance(step, dict):
+) -> NodeUse | ConditionNode | ForeachNode | InterruptNode | JoinNode:
-        raise ValueError(f"{path} must be a JSON object")
+    if isinstance(step, DraftNodeUse):
-
+        return NodeUse.model_validate(
-    step_id = _required_str(step, "id", path=path)
+            {
-    kind = _required_str(step, "kind", path=path)
+                "id": step.id,
-    if kind == "use":
+                "type": "node",
-        node: JsonObject = {
+                "node": step.capability,
-            "id": step_id,
+                "desc": step.desc,
-            "type": "node",
+                "in_map": deepcopy(step.in_),
-            "node": _required_str(step, "capability", path=path),
+                "out_map": deepcopy(step.out),
-            "in_map": _optional_object(step, "in", default={}),
+                "retry": step.retry,
-            "out_map": _optional_object(step, "out", default={}),
+                "timeout_seconds": step.timeout_seconds,
-        }
+            }
-        _copy_optional(step, node, "desc")
+        )
-        _copy_optional(step, node, "retry")
+    if isinstance(step, DraftConditionNode):
-        _copy_optional(step, node, "timeout_seconds")
+        return ConditionNode.model_validate(
-        return node
+            {
-    if kind == "condition":
+                "id": step.id,
-        return {
+                "type": "condition",
-            "id": step_id,
+                "check": deepcopy(step.check),
-            "type": "condition",
+            }
-            "check": _required_object(step, "check", path=path),
+        )
-        }
+    if isinstance(step, DraftForeachNode):
-    if kind == "foreach":
+        return ForeachNode.model_validate(
-        node = {
+            {
-            "id": step_id,
+                "id": step.id,
-            "type": "foreach",
+                "type": "foreach",
-            "over": _required_str(step, "over", path=path),
+                "over": step.over,
-            "as": _required_str(step, "as", path=path),
+                "as": step.as_,
-        }
+                "mode": step.mode,
-        _copy_optional(step, node, "mode")
+                "on_item_error": step.on_item_error,
-        _copy_optional(step, node, "on_item_error")
+            }
-        return node
+        )
-    if kind == "interrupt":
+    if isinstance(step, DraftInterruptNode):
-        node = {
+        return InterruptNode.model_validate(
-            "id": step_id,
+            {
-            "type": "interrupt",
+                "id": step.id,
-            "kind": _required_str(step, "interrupt_kind", path=path),
+                "type": "interrupt",
-            "request_map": _optional_object(step, "request", default={}),
+                "kind": step.interrupt_kind,
-            "out_map": _optional_object(step, "resume", default={}),
+                "request_map": deepcopy(step.request),
-        }
+                "out_map": deepcopy(step.resume),
-        _copy_optional(step, node, "outcomes")
+                "outcomes": deepcopy(step.outcomes),
-        return node
+            }
-    if kind == "join":
+        )
-        return {"id": step_id, "type": "join"}
+    return JoinNode.model_validate({"id": step.id, "type": "join"})
-    raise ValueError(f"{path}.kind has unsupported value {kind!r}")
 
 
-def _compile_edge(edge: object, index: int) -> JsonObject:
+def _validate_graph_references(draft: WorkflowDraft) -> None:
-    path = f"edges[{index}]"
+    step_ids = [step.id for step in draft.steps]
-    if not isinstance(edge, dict):
+    step_id_set = set(step_ids)
-        raise ValueError(f"{path} must be a JSON object")
+    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 {
-        "from": _required_str(edge, "from", path=path),
+        "status": "invalid",
-        "outcome": _required_str(edge, "outcome", path=path),
+        "diagnostics": [diagnostic.model_dump(mode="json")],
-        "to": _required_str(edge, "to", path=path),
     }
 
 
-def _validate_plan_model(plan: JsonObject) -> None:
+def _diagnostic_from_validation_error(
-    try:
+    exc: ValidationError,
-        Workflow.model_validate(plan)
+    draft: JsonObject,
-    except ValidationError as exc:
+) -> DraftDiagnostic:
-        raise ValueError(_validation_error_message(exc)) from exc
-
-
-def _validate_graph_references(plan: JsonObject) -> None:
-    node_ids = [node["id"] for node in plan["nodes"] if isinstance(node, dict)]
-    node_id_set = set(node_ids)
-    if len(node_ids) != len(node_id_set):
-        raise ValueError("steps contain duplicate ids")
-    if plan["start"] not in node_id_set:
-        raise ValueError(f"start references unknown step id {plan['start']!r}")
-    for index, edge in enumerate(plan["edges"]):
-        if edge["from"] not in node_id_set:
-            raise ValueError(
-                f"edges[{index}].from references unknown step id {edge['from']!r}"
-            )
-        if edge["to"] != "__end__" and edge["to"] not in node_id_set:
-            raise ValueError(
-                f"edges[{index}].to references unknown step id {edge['to']!r}"
-            )
-
-
-def _steps(draft: JsonObject) -> list[object]:
-    steps = draft.get("steps")
-    if not isinstance(steps, list):
-        raise ValueError("steps must be an array")
-    return steps
-
-
-def _edges(draft: JsonObject) -> list[object]:
-    edges = draft.get("edges")
-    if not isinstance(edges, list):
-        raise ValueError("edges must be an array")
-    return edges
-
-
-def _required_object(payload: JsonObject, key: str, *, path: str = "") -> JsonObject:
-    value = payload.get(key)
-    if not isinstance(value, dict):
-        raise ValueError(f"{_join_path(path, key)} must be an object")
-    return deepcopy(value)
-
-
-def _optional_object(
-    payload: JsonObject,
-    key: str,
-    *,
-    default: JsonObject,
-) -> JsonObject:
-    value = payload.get(key, default)
-    if not isinstance(value, dict):
-        raise ValueError(f"{key} must be an object")
-    return deepcopy(value)
-
-
-def _required_str(payload: JsonObject, key: str, *, path: str = "") -> str:
-    value = payload.get(key)
-    if not isinstance(value, str) or not value:
-        raise ValueError(f"{_join_path(path, key)} must be a non-empty string")
-    return value
-
-
-def _copy_optional(source: JsonObject, target: JsonObject, key: str) -> None:
-    if key in source:
-        target[key] = deepcopy(source[key])
-
-
-def _join_path(path: str, key: str) -> str:
-    return f"{path}.{key}" if path else key
-
-
-def _validation_error_message(exc: ValidationError) -> str:
     first_error = exc.errors()[0]
-    location = ".".join(str(part) for part in first_error["loc"])
+    path = _format_error_path(first_error["loc"])
-    return f"{_draft_path(location)}: {first_error['msg']}"
+    return DraftDiagnostic(
+        code="draft_invalid",
+        path=path,
+        step_id=_step_id_for_path(draft, path),
+        message=first_error["msg"],
+    )
 
 
-def _diagnostic_from_exception(exc: Exception) -> JsonObject:
+def _format_error_path(location: tuple[object, ...]) -> str:
-    message = str(exc)
+    parts: list[str] = []
-    path = _path_from_message(message)
+    for part in location:
-    return {
+        if isinstance(part, int):
-        "code": "draft_invalid",
+            parts[-1] = f"{parts[-1]}[{part}]"
-        "path": path,
+            continue
-        "step_id": _step_id_from_path(path),
+        if part == "in_":
-        "message": message,
+            part = "in"
-    }
+        elif part == "as_":
+            part = "as"
+        if part in {"use", "condition", "foreach", "interrupt", "join"}:
+            continue
+        parts.append(str(part))
+    return ".".join(parts)
 
 
-def _path_from_message(message: str) -> str:
+def _step_id_for_path(draft: JsonObject, path: str) -> str | None:
-    if ":" in message and message.split(":", 1)[0]:
-        return _draft_path(message.split(":", 1)[0])
-    token = message.split(" ", 1)[0]
-    if token.startswith(("steps[", "edges[")) or token in {
-        "name",
-        "input_schema",
-        "state_schema",
-        "output_schema",
-        "start",
-        "steps",
-        "edges",
-    }:
-        return _draft_path(token)
-    return ""
-
-
-def _draft_path(path: str) -> str:
-    return path.replace("nodes[", "steps[").replace(".type", ".kind")
-
-
-def _step_id_from_path(path: str) -> str | None:
     if not path.startswith("steps["):
         return None
-    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

tests/artifacts/test_drafts.py

@@ -2,7 +2,13 @@ from __future__ import annotations
 
 import pytest
 
-from wf_artifacts.drafts import compile_workflow_draft, patch_workflow_draft
+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:
@@ -18,6 +24,13 @@ def test_compile_draft_maps_use_step_to_raw_node_use() -> None:
     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(
@@ -77,6 +90,16 @@ def test_compile_draft_requires_explicit_step_ids() -> None:
     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()]),