feat: complete draft step model parity

src/wf_artifacts/drafts/__init__.py

@@ -14,6 +14,7 @@ from .models import (
     DraftJoinStep,
     DraftMatchCase,
     DraftMatchStep,
+    DraftSubgraphStep,
     DraftUseStep,
     DraftWhenStep,
     WorkflowDraft,
@@ -29,6 +30,7 @@ __all__ = [
     "DraftJoinStep",
     "DraftMatchCase",
     "DraftMatchStep",
+    "DraftSubgraphStep",
     "DraftWhenStep",
     "DraftUseStep",
     "WorkflowDraft",

src/wf_artifacts/drafts/adapter.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+from typing import Any
+
 from wf_authoring import WorkflowBuilder
 from wf_authoring.dsl import PathExpr
-from wf_core import JoinNode, Workflow
+from wf_core import JoinNode, SubgraphNode, Workflow
 from wf_core.paths import GraphSourcePath
 
 from .models import (
@@ -13,6 +15,7 @@ from .models import (
     DraftJoinStep,
     DraftMatchStep,
     DraftStep,
+    DraftSubgraphStep,
     DraftUseStep,
     DraftWhenStep,
     WorkflowDraft,
@@ -63,12 +66,23 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
             concurrent=step.foreach.concurrent,
         )
     if isinstance(step, DraftInterruptStep):
+        interrupt_kwargs: dict[str, Any] = {
+            "id": step_id,
+            "kind": step.interrupt.kind,
+            "request": step.interrupt.request,
+            "resume": step.interrupt.resume,
+            "outcomes": step.interrupt.outcomes,
+        }
+        if step.interrupt.request_schema is not None:
+            interrupt_kwargs["request_schema"] = step.interrupt.request_schema.model_dump(
+                mode="json", exclude_none=True
+            )
+        if step.interrupt.resume_schema is not None:
+            interrupt_kwargs["resume_schema"] = step.interrupt.resume_schema.model_dump(
+                mode="json", exclude_none=True
+            )
         return builder.interrupt(
-            id=step_id,
+            **interrupt_kwargs,
-            kind=step.interrupt.kind,
-            request=step.interrupt.request,
-            resume=step.interrupt.resume,
-            outcomes=step.interrupt.outcomes,
         )
     if isinstance(step, DraftJoinStep):
         node = JoinNode(id=step_id, type="join")
@@ -96,4 +110,12 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
             id=step_id,
             default=step.match.default,
         ).entry
+    if isinstance(step, DraftSubgraphStep):
+        node = SubgraphNode(
+            id=step_id,
+            type="subgraph",
+            **step.subgraph.model_dump(),
+        )
+        builder.nodes.append(node)
+        return node
     raise TypeError(f"unsupported draft step {type(step)!r}")

src/wf_artifacts/drafts/models.py

@@ -2,15 +2,17 @@ from __future__ import annotations
 
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from wf_core.models.conditions import Condition
+from wf_core.models.schemas import SchemaRef
 from wf_core.models.steps import (
     ForeachConcurrentPolicy,
     ForeachItemErrorPolicy,
     InputBinding,
     OutputBinding,
 )
+from wf_core.models.workflow_refs import WorkflowRef
 from wf_core.paths import GraphSourcePath
 
 JsonObject = dict[str, Any]
@@ -24,6 +26,7 @@ STEP_KIND_KEYS = frozenset(
         "when",
         "choose",
         "match",
+        "subgraph",
     }
 )
 
@@ -153,8 +156,18 @@ class DraftInterruptPayload(BaseModel):
     kind: str
     request: list[InputBinding] = Field(default_factory=list)
     resume: list[OutputBinding] = Field(default_factory=list)
+    request_schema: SchemaRef | None = None
+    resume_schema: SchemaRef | None = None
     outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
 
+    @field_validator("request_schema", "resume_schema")
+    @classmethod
+    def _require_object_schema(cls, value: SchemaRef | None) -> SchemaRef | None:
+        """Keep explicit interrupt contracts distinct from untyped interrupts."""
+        if value is not None and value.type != "object":
+            raise ValueError("interrupt schema must describe a JSON object")
+        return value
+
     @model_validator(mode="before")
     @classmethod
     def _coerce_legacy_maps(cls, data: object) -> object:
@@ -184,6 +197,30 @@ class DraftInterruptStep(BaseModel):
     interrupt: DraftInterruptPayload
 
 
+class DraftSubgraphPayload(BaseModel):
+    """Declarative boundary contract for a referenced child workflow."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    workflow: WorkflowRef
+    desc: str | None = None
+    input_schema: SchemaRef = Field(default_factory=lambda: SchemaRef(type="object"))
+    output_schema: SchemaRef = Field(
+        default_factory=lambda: SchemaRef(type="object")
+    )
+    input: list[InputBinding] = Field(default_factory=list)
+    output: list[OutputBinding] = Field(default_factory=list)
+    outcomes: list[str] = Field(default_factory=lambda: ["ok"], min_length=1)
+
+
+class DraftSubgraphStep(BaseModel):
+    """Draft step that lowers to a native `SubgraphNode` boundary."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    subgraph: DraftSubgraphPayload
+
+
 class DraftJoinStep(BaseModel):
     """Draft step that emits the current core join node."""
 
@@ -292,6 +329,7 @@ DraftStep = (
     | DraftWhenStep
     | DraftChooseStep
     | DraftMatchStep
+    | DraftSubgraphStep
 )
 
 

tests/artifacts/test_draft_adapter.py

@@ -9,9 +9,11 @@ from wf_core import (
     ConditionNode,
     EndNode,
     ForeachNode,
+    InterruptNode,
     NodeDef,
     NodeUse,
     SchemaRef,
+    SubgraphNode,
     execute_workflow,
 )
 from wf_core.models.steps import InputValueBinding
@@ -480,6 +482,99 @@ def test_adapter_lowers_foreach_policy_through_builder() -> None:
     assert str(foreach.item_error.collect_to) == "state.item_errors"
 
 
+def test_adapter_lowers_typed_and_untyped_interrupt_steps() -> None:
+    request_schema = {
+        "type": "object",
+        "properties": {"issues": {"type": "array"}},
+        "required": ["issues"],
+    }
+    resume_schema = {
+        "type": "object",
+        "properties": {"selected": {"type": "array"}},
+        "required": ["selected"],
+    }
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "review",
+            "input_schema": {},
+            "state_schema": {"type": "object"},
+            "output_schema": {},
+            "start": "review",
+            "steps": {
+                "review": {
+                    "interrupt": {
+                        "kind": "issue_review",
+                        "request_schema": request_schema,
+                        "resume_schema": resume_schema,
+                        "outcomes": ["submitted", "cancelled"],
+                    }
+                },
+                "legacy": {"interrupt": {"kind": "legacy"}},
+            },
+            "routes": {
+                "review": {"submitted": "__end__", "cancelled": "__end__"},
+                "legacy": {"submitted": "__end__"},
+            },
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+
+    review = workflow.nodes[0]
+    legacy = workflow.nodes[1]
+    assert isinstance(review, InterruptNode)
+    assert review.request_schema == request_schema
+    assert review.resume_schema == resume_schema
+    assert review.has_explicit_contract is True
+    assert isinstance(legacy, InterruptNode)
+    assert legacy.has_explicit_contract is False
+
+
+def test_adapter_lowers_subgraph_step_without_resolving_artifact() -> None:
+    input_schema = {
+        "type": "object",
+        "properties": {"topic": {"type": "string"}},
+    }
+    output_schema = {
+        "type": "object",
+        "properties": {"report": {"type": "string"}},
+    }
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "parent",
+            "input_schema": {},
+            "state_schema": {"type": "object"},
+            "output_schema": {},
+            "start": "child",
+            "steps": {
+                "child": {
+                    "subgraph": {
+                        "workflow": {"artifact_id": "child_report", "version": 2},
+                        "input_schema": input_schema,
+                        "output_schema": output_schema,
+                        "input": [{"target": "topic", "path": "state.topic"}],
+                        "output": [
+                            {"source": "report", "target": "state.report"}
+                        ],
+                        "outcomes": ["ok", "error"],
+                    }
+                }
+            },
+            "routes": {"child": {"ok": "__end__", "error": "__end__"}},
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+    child = workflow.nodes[0]
+
+    assert isinstance(child, SubgraphNode)
+    assert child.workflow.artifact_id == "child_report"
+    assert child.workflow.version == 2
+    assert child.input_schema == SchemaRef.model_validate(input_schema)
+    assert child.output_schema == SchemaRef.model_validate(output_schema)
+    assert child.outcomes == ["ok", "error"]
+
+
 def test_validate_workflow_draft_reports_structured_output_destination_issue() -> None:
     draft = {
         "name": "missing_state_field",

tests/artifacts/test_draft_models.py

@@ -9,7 +9,9 @@ from wf_artifacts.drafts import (
     DraftChooseStep,
     DraftEndStep,
     DraftForeachStep,
+    DraftInterruptStep,
     DraftMatchStep,
+    DraftSubgraphStep,
     DraftUseStep,
     DraftWhenStep,
     WorkflowDraft,
@@ -81,6 +83,115 @@ def test_workflow_draft_accepts_legacy_interrupt_maps_but_dumps_canonical_bindin
     )
 
 
+def test_workflow_draft_preserves_typed_interrupt_contracts() -> None:
+    request_schema = {
+        "type": "object",
+        "properties": {"issues": {"type": "array"}},
+        "required": ["issues"],
+    }
+    resume_schema = {
+        "type": "object",
+        "properties": {"selected": {"type": "array"}},
+        "required": ["selected"],
+    }
+    draft = WorkflowDraft.model_validate(
+        {
+            **_keyed_echo_draft(),
+            "start": "review",
+            "steps": {
+                "review": {
+                    "interrupt": {
+                        "kind": "issue_review",
+                        "request_schema": request_schema,
+                        "resume_schema": resume_schema,
+                        "outcomes": ["submitted", "cancelled"],
+                    }
+                }
+            },
+        }
+    )
+
+    step = draft.steps["review"]
+    dumped = draft.model_dump(mode="json", by_alias=True)
+
+    assert isinstance(step, DraftInterruptStep)
+    assert step.interrupt.request_schema is not None
+    assert step.interrupt.request_schema.type == "object"
+    assert step.interrupt.request_schema.properties == {"issues": {"type": "array"}}
+    assert step.interrupt.resume_schema is not None
+    assert step.interrupt.resume_schema.required == ["selected"]
+    assert dumped["steps"]["review"]["interrupt"]["request_schema"] == request_schema
+    assert dumped["steps"]["review"]["interrupt"]["resume_schema"] == resume_schema
+
+
+def test_workflow_draft_rejects_non_object_interrupt_contracts() -> None:
+    with pytest.raises(ValidationError, match="interrupt schema must describe"):
+        WorkflowDraft.model_validate(
+            {
+                **_keyed_echo_draft(),
+                "start": "review",
+                "steps": {
+                    "review": {
+                        "interrupt": {
+                            "kind": "issue_review",
+                            "request_schema": {"type": "array"},
+                        }
+                    }
+                },
+            }
+        )
+
+
+def test_workflow_draft_preserves_subgraph_workflow_boundaries() -> None:
+    child_report = {
+        "workflow": {"artifact_id": "child_report", "version": 2},
+        "input_schema": {
+            "type": "object",
+            "properties": {"topic": {"type": "string"}},
+        },
+        "output_schema": {
+            "type": "object",
+            "properties": {"report": {"type": "string"}},
+        },
+        "input": [{"target": "topic", "path": "state.topic"}],
+        "output": [{"source": "report", "target": "state.report"}],
+        "outcomes": ["ok", "error"],
+    }
+    for step_id, subgraph in [
+        (
+            "child",
+            {"workflow": {"name": "child"}, "outcomes": ["ok"]},
+        ),
+        ("child_report", child_report),
+    ]:
+        draft = WorkflowDraft.model_validate(
+            {
+                **_keyed_echo_draft(),
+                "start": step_id,
+                "steps": {step_id: {"subgraph": subgraph}},
+            }
+        )
+
+        step = draft.steps[step_id]
+        dumped = draft.model_dump(mode="json", by_alias=True)
+
+        assert isinstance(step, DraftSubgraphStep)
+        dumped_subgraph = dumped["steps"][step_id]["subgraph"]
+        assert dumped_subgraph["workflow"] == subgraph["workflow"]
+        assert dumped_subgraph["outcomes"] == subgraph["outcomes"]
+        if step_id == "child_report":
+            assert dumped_subgraph["input_schema"]["type"] == "object"
+            assert dumped_subgraph["input_schema"]["properties"] == {
+                "topic": {"type": "string"}
+            }
+            assert dumped_subgraph["output_schema"]["type"] == "object"
+            assert dumped_subgraph["output_schema"]["properties"] == {
+                "report": {"type": "string"}
+            }
+            assert dumped_subgraph["input"] == child_report["input"]
+            assert dumped_subgraph["output"] == child_report["output"]
+
+
 def test_draft_step_requires_exactly_one_kind_key() -> None:
     draft = _keyed_echo_draft()
     steps = draft["steps"]