add "when" and "choose" as a draft step

docs/workflow_drafts.md

@@ -183,6 +183,69 @@ Joins control flow.
 }
 ```
 
+### `when`
+
+Creates one boolean decision step. The condition uses the same JSON shape as
+`wf_core.models.conditions.Condition`.
+
+```json
+{
+  "when": {
+    "if": {
+      "op": "ge",
+      "left": {
+        "path": "state.count"
+      },
+      "right": {
+        "value": 1
+      }
+    },
+    "then": "positive",
+    "otherwise": "zero"
+  }
+}
+```
+
+The draft adapter lowers this through `WorkflowBuilder.when()`. The draft step
+id becomes the generated condition entry id, so other routes can target it.
+
+### `choose`
+
+Creates an ordered first-true decision chain.
+
+```json
+{
+  "choose": {
+    "clauses": [
+      {
+        "if": {
+          "op": "gt",
+          "left": {
+            "path": "state.score"
+          },
+          "right": {
+            "value": 80
+          }
+        },
+        "then": "high"
+      },
+      {
+        "if": {
+          "op": "exists",
+          "path": "state.fallback"
+        },
+        "then": "fallback"
+      }
+    ],
+    "default": "__end__"
+  }
+}
+```
+
+`choose` lowers through `WorkflowBuilder.choose()` and expands to generated
+condition nodes. It replaces the deprecated `route()` concept for draft JSON;
+there is intentionally no draft `route` step kind.
+
 ## Draft Tools
 
 The workflow MCP surface exposes these draft tools:

src/wf_artifacts/drafts/__init__.py

@@ -6,18 +6,24 @@ from .api import (
     validate_workflow_draft,
 )
 from .models import (
+    DraftChooseClause,
+    DraftChooseStep,
     DraftForeachStep,
     DraftInterruptStep,
     DraftJoinStep,
+    DraftWhenStep,
     DraftUseStep,
     WorkflowDraft,
 )
 
 __all__ = [
     "DraftDiagnostic",
+    "DraftChooseClause",
+    "DraftChooseStep",
     "DraftForeachStep",
     "DraftInterruptStep",
     "DraftJoinStep",
+    "DraftWhenStep",
     "DraftUseStep",
     "WorkflowDraft",
     "build_workflow_from_draft",

src/wf_artifacts/drafts/adapter.py

@@ -9,6 +9,8 @@ from .models import (
     DraftJoinStep,
     DraftStep,
     DraftUseStep,
+    DraftWhenStep,
+    DraftChooseStep,
     WorkflowDraft,
 )
 
@@ -61,4 +63,20 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
         node = JoinNode(id=step_id, type="join")
         builder.nodes.append(node)
         return node
+    if isinstance(step, DraftWhenStep):
+        return builder.when(
+            step.when.if_,
+            id=step_id,
+            then=step.when.then,
+            otherwise=step.when.otherwise,
+        ).entry
+    if isinstance(step, DraftChooseStep):
+        return builder.choose(
+            *[
+                (clause.if_, clause.then)
+                for clause in step.choose.clauses
+            ],
+            id=step_id,
+            default=step.choose.default,
+        ).entry
     raise TypeError(f"unsupported draft step {type(step)!r}")

src/wf_artifacts/drafts/models.py

@@ -4,8 +4,10 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, Field, model_validator
 
+from wf_core.models.conditions import Condition
+
 JsonObject = dict[str, Any]
-STEP_KIND_KEYS = frozenset({"use", "foreach", "interrupt", "join"})
+STEP_KIND_KEYS = frozenset({"use", "foreach", "interrupt", "join", "when", "choose"})
 
 
 class DraftUseStep(BaseModel):
@@ -55,7 +57,48 @@ class DraftJoinStep(BaseModel):
     join: JsonObject = Field(default_factory=dict)
 
 
-DraftStep = DraftUseStep | DraftForeachStep | DraftInterruptStep | DraftJoinStep
+class DraftWhenPayload(BaseModel):
+    """Payload for one boolean draft decision."""
+
+    if_: Condition = Field(alias="if")
+    then: str
+    otherwise: str = "__end__"
+
+
+class DraftWhenStep(BaseModel):
+    """Draft step that delegates one boolean decision to `WorkflowBuilder.when`."""
+
+    when: DraftWhenPayload
+
+
+class DraftChooseClause(BaseModel):
+    """One ordered boolean clause in a draft choose decision."""
+
+    if_: Condition = Field(alias="if")
+    then: str
+
+
+class DraftChoosePayload(BaseModel):
+    """Payload for an ordered first-true draft decision."""
+
+    clauses: list[DraftChooseClause] = Field(min_length=1)
+    default: str = "__end__"
+
+
+class DraftChooseStep(BaseModel):
+    """Draft step that delegates ordered decisions to `WorkflowBuilder.choose`."""
+
+    choose: DraftChoosePayload
+
+
+DraftStep = (
+    DraftUseStep
+    | DraftForeachStep
+    | DraftInterruptStep
+    | DraftJoinStep
+    | DraftWhenStep
+    | DraftChooseStep
+)
 
 
 class WorkflowDraft(BaseModel):

src/wf_authoring/builder/core.py

@@ -346,7 +346,7 @@ class WorkflowBuilder:
 
     def when(
         self,
-        condition: Expr,
+        condition: CoreCondition | Expr,
         *,
         then: BranchRef,
         otherwise: BranchRef = runtime_error,
@@ -369,7 +369,7 @@ class WorkflowBuilder:
 
     def choose(
         self,
-        *clauses: tuple[Expr, BranchRef],
+        *clauses: tuple[CoreCondition | Expr, BranchRef],
         default: BranchRef = runtime_error,
         id: str | None = None,
     ) -> DecisionResult:

tests/artifacts/test_draft_adapter.py

@@ -2,7 +2,7 @@ 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
+from wf_core import ConditionNode, NodeUse
 
 
 def test_adapter_lowers_keyed_use_steps_and_routes_through_builder() -> None:
@@ -27,3 +27,97 @@ def test_adapter_lowers_keyed_use_steps_and_routes_through_builder() -> None:
     assert workflow.edges[0].from_ == "echo"
     assert workflow.edges[0].outcome == "ok"
     assert workflow.edges[0].to == "__end__"
+
+
+def test_adapter_lowers_when_step_through_builder() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "when_example",
+            "input_schema": {},
+            "state_schema": {"fields": {}},
+            "output_schema": {},
+            "start": "decide",
+            "steps": {
+                "decide": {
+                    "when": {
+                        "if": {
+                            "op": "ge",
+                            "left": {"path": "state.count"},
+                            "right": {"value": 1},
+                        },
+                        "then": "echo",
+                        "otherwise": "__end__",
+                    }
+                },
+                "echo": {"use": "demo.echo"},
+            },
+            "routes": {"echo": {"ok": "__end__"}},
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+    condition = workflow.nodes[0]
+
+    assert isinstance(condition, ConditionNode)
+    assert condition.id == "decide"
+    assert workflow.start == "decide"
+    assert [(edge.from_, edge.outcome, edge.to) for edge in workflow.edges[:2]] == [
+        ("decide", "true", "echo"),
+        ("decide", "false", "__end__"),
+    ]
+
+
+def test_adapter_lowers_choose_step_through_builder() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "choose_example",
+            "input_schema": {},
+            "state_schema": {"fields": {}},
+            "output_schema": {},
+            "start": "pick",
+            "steps": {
+                "pick": {
+                    "choose": {
+                        "clauses": [
+                            {
+                                "if": {
+                                    "op": "gt",
+                                    "left": {"path": "state.score"},
+                                    "right": {"value": 80},
+                                },
+                                "then": "high",
+                            },
+                            {
+                                "if": {
+                                    "op": "exists",
+                                    "path": "state.fallback",
+                                },
+                                "then": "fallback",
+                            },
+                        ],
+                        "default": "__end__",
+                    }
+                },
+                "high": {"use": "demo.high"},
+                "fallback": {"use": "demo.fallback"},
+            },
+            "routes": {
+                "high": {"ok": "__end__"},
+                "fallback": {"ok": "__end__"},
+            },
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+    condition_ids = [
+        node.id for node in workflow.nodes if isinstance(node, ConditionNode)
+    ]
+
+    assert condition_ids == ["pick", "pick_2"]
+    assert workflow.start == "pick"
+    assert [(edge.from_, edge.outcome, edge.to) for edge in workflow.edges[:4]] == [
+        ("pick", "true", "high"),
+        ("pick", "false", "pick_2"),
+        ("pick_2", "true", "fallback"),
+        ("pick_2", "false", "__end__"),
+    ]

tests/artifacts/test_draft_models.py

@@ -1,9 +1,16 @@
 from __future__ import annotations
 
+from typing import Any
+
 import pytest
 from pydantic import ValidationError
 
-from wf_artifacts.drafts import DraftUseStep, WorkflowDraft
+from wf_artifacts.drafts import (
+    DraftChooseStep,
+    DraftUseStep,
+    DraftWhenStep,
+    WorkflowDraft,
+)
 
 
 def test_workflow_draft_uses_keyed_steps() -> None:
@@ -27,7 +34,60 @@ def test_draft_step_requires_exactly_one_kind_key() -> None:
     assert "steps.echo" in str(exc_info.value)
 
 
-def _keyed_echo_draft() -> dict[str, object]:
+def test_workflow_draft_accepts_when_step() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            **_keyed_echo_draft(),
+            "start": "decide",
+            "steps": {
+                **_keyed_echo_draft()["steps"],
+                "decide": {
+                    "when": {
+                        "if": {
+                            "op": "ge",
+                            "left": {"path": "state.count"},
+                            "right": {"value": 1},
+                        },
+                        "then": "echo",
+                        "otherwise": "__end__",
+                    }
+                },
+            },
+        }
+    )
+
+    assert isinstance(draft.steps["decide"], DraftWhenStep)
+
+
+def test_workflow_draft_accepts_choose_step() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            **_keyed_echo_draft(),
+            "start": "choose_next",
+            "steps": {
+                **_keyed_echo_draft()["steps"],
+                "choose_next": {
+                    "choose": {
+                        "clauses": [
+                            {
+                                "if": {
+                                    "op": "exists",
+                                    "path": "state.text",
+                                },
+                                "then": "echo",
+                            }
+                        ],
+                        "default": "__end__",
+                    }
+                },
+            },
+        }
+    )
+
+    assert isinstance(draft.steps["choose_next"], DraftChooseStep)
+
+
+def _keyed_echo_draft() -> dict[str, Any]:
     return {
         "name": "echo",
         "input_schema": {},