foreach container arg to get the path treatment too!

This commit is contained in:
lda
2026-05-21 14:31:52 +07:00 Verified
parent 3be1593276
commit 23ffddf9b3
8 changed files with 54 additions and 8 deletions
+1 -1
View File
@@ -277,7 +277,7 @@ Runs a child body over items.
```json
{
"foreach": {
"over": "state.items",
"over": {"root": "state", "parts": ["items"]},
"as": "item",
"mode": "serial",
"on_item_error": "fail"
+1 -1
View File
@@ -137,7 +137,7 @@ def build_demo_workflow() -> Workflow:
{
"id": "summarize_each",
"type": "foreach",
"over": "state.documents",
"over": {"root": "state", "parts": ["documents"]},
"as": "document",
"mode": "serial",
"on_item_error": "fail",
+2 -1
View File
@@ -6,6 +6,7 @@ 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.paths import GraphSourcePath
JsonObject = dict[str, Any]
STEP_KIND_KEYS = frozenset(
@@ -94,7 +95,7 @@ class DraftForeachPayload(BaseModel):
model_config = ConfigDict(extra="forbid", populate_by_name=True)
over: str
over: GraphSourcePath
as_: str = Field(alias="as")
mode: Literal["serial", "parallel"] = "serial"
on_item_error: Literal["fail", "collect", "skip"] = "fail"
-2
View File
@@ -397,8 +397,6 @@ class WorkflowBuilder:
mode: Literal["serial", "parallel"] = "serial",
on_item_error: Literal["fail", "collect", "skip"] = "fail",
) -> ForeachNode:
# Core foreach still stores `over` as a string. Keep this compatibility
# path isolated until ForeachNode grows a typed GraphSourcePath field.
node = ForeachNode.model_validate(
{
"id": id or self._next_step_id(f"foreach_{slug_id(as_)}"),
+1 -1
View File
@@ -124,7 +124,7 @@ class ForeachNode(BaseModel):
id: str
type: Literal["foreach"]
over: str
over: GraphSourcePath
as_: str = Field(alias="as")
mode: Literal["serial", "parallel"] = "serial"
on_item_error: Literal["fail", "collect", "skip"] = "fail"
+2 -2
View File
@@ -26,14 +26,14 @@ def step_foreach(
progress = progress_map.setdefault(step.id, {"index": 0})
iterable = safe_resolve_path(
step.over,
str(step.over),
state=run.state,
workflow_input=run.workflow_input,
context=frame_context_values(frame),
)
if not isinstance(iterable, list):
raise WorkflowExecutionError(
f"foreach source {step.over!r} must resolve to a list"
f"foreach source {str(step.over)!r} must resolve to a list"
)
loop_index = progress["index"]
+29
View File
@@ -162,6 +162,35 @@ def test_workflow_draft_accepts_choose_step() -> None:
assert isinstance(draft.steps["choose_next"], DraftChooseStep)
def test_workflow_draft_foreach_over_dumps_structural_path() -> None:
draft = WorkflowDraft.model_validate(
{
**_keyed_echo_draft(),
"start": "each_item",
"steps": {
**_keyed_echo_draft()["steps"],
"each_item": {
"foreach": {
"over": "state.items",
"as": "item",
}
},
},
"routes": {
"each_item": {"loop": "echo", "done": "__end__"},
"echo": {"ok": "__end__"},
},
}
)
dumped = draft.model_dump(mode="json")
assert dumped["steps"]["each_item"]["foreach"]["over"] == {
"root": "state",
"parts": ["items"],
}
def test_workflow_draft_accepts_match_step() -> None:
draft = WorkflowDraft.model_validate(
{
@@ -4,6 +4,7 @@ from pydantic import ValidationError
from wf_core.models.steps import (
InputPathBinding,
InputValueBinding,
ForeachNode,
InterruptNode,
NodeUse,
)
@@ -276,3 +277,20 @@ def test_interrupt_node_rejects_mixed_old_and_new_binding_styles():
"request_map": {"input.other": "other"},
}
)
def test_foreach_node_serializes_over_path_as_structural_json():
node = ForeachNode.model_validate(
{
"id": "each_item",
"type": "foreach",
"over": "state.items",
"as": "item",
}
)
assert node.over == GraphSourcePath.state("items")
assert node.model_dump(mode="json")["over"] == {
"root": "state",
"parts": ["items"],
}