drafts system to get ts too

docs/workflow_drafts.md

@@ -286,7 +286,8 @@ the generated `error` outcome dangling.
 
 ### `foreach`
 
-Runs a child body over items.
+Runs a child body over items. Draft foreach mirrors the core foreach policy
+model: use `item_error` and `concurrent`, not draft-only field names.
 
 ```json
 {
@@ -294,13 +295,35 @@ Runs a child body over items.
     "over": {"root": "state", "parts": ["items"]},
     "as": "item",
     "mode": "serial",
-    "on_item_error": "fail"
+    "item_error": "fail"
   }
 }
 ```
 
-Use `serial` unless the runtime explicitly supports a parallel async path for
+Concurrent foreach uses the same canonical policy shape as core:
-the target workflow.
+
+```json
+{
+  "foreach": {
+    "over": {"root": "state", "parts": ["items"]},
+    "as": "item",
+    "mode": "concurrent",
+    "concurrent": {
+      "max_active": 2,
+      "max_outstanding": 4
+    },
+    "item_error": {
+      "action": "collect",
+      "collect_to": {"root": "state", "parts": ["item_errors"]}
+    }
+  }
+}
+```
+
+`item_error` accepts `"fail"` and `"skip"` as shorthand. `collect` needs an
+explicit destination, so `item_error: "collect"` is invalid; use the object
+shape and provide `collect_to`. Deprecated `on_item_error` and `parallel` are
+accepted only as parse-only compatibility and dump back to canonical fields.
 
 ### `interrupt`
 

src/wf_artifacts/drafts/adapter.py

@@ -52,7 +52,7 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
             over=step.foreach.over,
             as_=step.foreach.as_,
             mode=step.foreach.mode,
-            on_item_error=step.foreach.on_item_error,
+            item_error=step.foreach.item_error,
             concurrent=step.foreach.concurrent,
         )
     if isinstance(step, DraftInterruptStep):

src/wf_artifacts/drafts/models.py

@@ -5,7 +5,12 @@ from typing import Any, Literal
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from wf_core.models.conditions import Condition
-from wf_core.models.steps import InputBinding, OutputBinding
+from wf_core.models.steps import (
+    ForeachConcurrentPolicy,
+    ForeachItemErrorPolicy,
+    InputBinding,
+    OutputBinding,
+)
 from wf_core.paths import GraphSourcePath
 
 JsonObject = dict[str, Any]
@@ -98,13 +103,18 @@ class DraftForeachPayload(BaseModel):
     over: GraphSourcePath
     as_: str = Field(alias="as")
     mode: Literal["serial", "concurrent"] = "serial"
-    on_item_error: Literal["fail", "collect", "skip"] = "fail"
+    item_error: ForeachItemErrorPolicy = Field(default_factory=ForeachItemErrorPolicy)
-    concurrent: JsonObject | None = None
+    concurrent: ForeachConcurrentPolicy | None = None
+    on_item_error: Literal["fail", "collect", "skip"] | None = Field(
+        default=None,
+        exclude=True,
+        description="Deprecated parse-only shorthand; use item_error.",
+    )
 
     @model_validator(mode="before")
     @classmethod
     def _coerce_legacy_parallel_policy(cls, data: object) -> object:
-        """Accept old draft foreach parallel names as parse-only compatibility."""
+        """Accept old foreach names while saving canonical policy fields."""
         if not isinstance(data, dict):
             return data
         normalized = dict(data)
@@ -114,6 +124,11 @@ class DraftForeachPayload(BaseModel):
             if "concurrent" in normalized:
                 raise ValueError("cannot mix deprecated parallel with concurrent")
             normalized["concurrent"] = normalized.pop("parallel")
+        old_item_error = normalized.pop("on_item_error", None)
+        if old_item_error is not None:
+            if "item_error" in normalized:
+                raise ValueError("cannot mix deprecated on_item_error with item_error")
+            normalized["item_error"] = old_item_error
         return normalized
 
 

tests/artifacts/test_draft_adapter.py

@@ -5,7 +5,7 @@ from pydantic import ValidationError
 from wf_artifacts.drafts import WorkflowDraft
 from wf_artifacts.drafts.api import compile_workflow_draft, validate_workflow_draft
 from wf_artifacts.drafts.adapter import build_workflow_from_draft
-from wf_core import ConditionNode, NodeUse
+from wf_core import ConditionNode, ForeachNode, NodeUse
 from wf_core.models.steps import InputValueBinding
 
 
@@ -279,3 +279,53 @@ def test_adapter_lowers_match_step_through_builder() -> None:
         ("match_status_2", "true", "waiting"),
         ("match_status_2", "false", "__end__"),
     ]
+
+
+def test_adapter_lowers_foreach_policy_through_builder() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "foreach_policy",
+            "input_schema": {},
+            "state_schema": {
+                "type": "object",
+                "properties": {
+                    "items": {"type": "array"},
+                    "item_errors": {"type": "array"},
+                },
+            },
+            "output_schema": {},
+            "start": "each_item",
+            "steps": {
+                "each_item": {
+                    "foreach": {
+                        "over": "state.items",
+                        "as": "item",
+                        "mode": "concurrent",
+                        "concurrent": {"max_active": 2, "max_outstanding": 4},
+                        "item_error": {
+                            "action": "collect",
+                            "collect_to": "state.item_errors",
+                        },
+                    }
+                }
+            },
+            "routes": {
+                "each_item": {
+                    "loop": "__end__",
+                    "done": "__end__",
+                    "completed_with_errors": "__end__",
+                }
+            },
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+    foreach = workflow.nodes[0]
+
+    assert isinstance(foreach, ForeachNode)
+    assert foreach.mode == "concurrent"
+    assert foreach.concurrent is not None
+    assert foreach.concurrent.max_active == 2
+    assert foreach.concurrent.max_outstanding == 4
+    assert foreach.item_error.action == "collect"
+    assert str(foreach.item_error.collect_to) == "state.item_errors"

tests/artifacts/test_draft_models.py

@@ -7,6 +7,7 @@ from pydantic import ValidationError
 
 from wf_artifacts.drafts import (
     DraftChooseStep,
+    DraftForeachStep,
     DraftMatchStep,
     DraftUseStep,
     DraftWhenStep,
@@ -191,6 +192,82 @@ def test_workflow_draft_foreach_over_dumps_structural_path() -> None:
     }
 
 
+def test_workflow_draft_foreach_accepts_canonical_item_error_policy() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            **_keyed_echo_draft(),
+            "start": "each_item",
+            "steps": {
+                "each_item": {
+                    "foreach": {
+                        "over": "state.items",
+                        "as": "item",
+                        "mode": "concurrent",
+                        "concurrent": {"max_active": 2, "max_outstanding": 3},
+                        "item_error": {
+                            "action": "collect",
+                            "collect_to": "state.item_errors",
+                        },
+                    }
+                }
+            },
+            "routes": {"each_item": {"loop": "__end__", "done": "__end__"}},
+        }
+    )
+
+    step = draft.steps["each_item"]
+    dumped = draft.model_dump(mode="json")
+
+    assert isinstance(step, DraftForeachStep)
+    assert dumped["steps"]["each_item"]["foreach"]["item_error"] == {
+        "action": "collect",
+        "collect_to": {"root": "state", "parts": ["item_errors"]},
+    }
+    assert "on_item_error" not in dumped["steps"]["each_item"]["foreach"]
+
+
+def test_workflow_draft_foreach_accepts_item_error_action_string() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            **_keyed_echo_draft(),
+            "start": "each_item",
+            "steps": {
+                "each_item": {
+                    "foreach": {
+                        "over": "state.items",
+                        "as": "item",
+                        "item_error": "skip",
+                    }
+                }
+            },
+            "routes": {"each_item": {"loop": "__end__", "done": "__end__"}},
+        }
+    )
+
+    dumped = draft.model_dump(mode="json")
+
+    assert dumped["steps"]["each_item"]["foreach"]["item_error"]["action"] == "skip"
+
+
+def test_workflow_draft_foreach_collect_string_requires_destination() -> None:
+    with pytest.raises(ValidationError, match="collect_to"):
+        WorkflowDraft.model_validate(
+            {
+                **_keyed_echo_draft(),
+                "start": "each_item",
+                "steps": {
+                    "each_item": {
+                        "foreach": {
+                            "over": "state.items",
+                            "as": "item",
+                            "item_error": "collect",
+                        }
+                    }
+                },
+            }
+        )
+
+
 def test_workflow_draft_accepts_match_step() -> None:
     draft = WorkflowDraft.model_validate(
         {