concurrent foreach preparation, types, validation, refactors

src/wf_core/__init__.py

@@ -1,6 +1,8 @@
 from .models import (
     ConditionNode,
     Edge,
+    ForeachConcurrentPolicy,
+    ForeachItemErrorPolicy,
     ForeachNode,
     InterruptNode,
     JoinNode,
@@ -47,6 +49,8 @@ from .validation import (
 __all__ = [
     "ConditionNode",
     "Edge",
+    "ForeachConcurrentPolicy",
+    "ForeachItemErrorPolicy",
     "ForeachNode",
     "InterruptNode",
     "JoinNode",

src/wf_core/models/__init__.py

@@ -13,6 +13,8 @@ from wf_core.models.reducers import ReducerRef, ReducerSpec
 from wf_core.models.schemas import NodeDef, SchemaRef, StateField, StateSchema
 from wf_core.models.steps import (
     ConditionNode,
+    ForeachConcurrentPolicy,
+    ForeachItemErrorPolicy,
     ForeachNode,
     InterruptNode,
     JoinNode,
@@ -27,6 +29,8 @@ __all__ = [
     "ConditionNode",
     "Edge",
     "ExistsCondition",
+    "ForeachConcurrentPolicy",
+    "ForeachItemErrorPolicy",
     "ForeachNode",
     "InterruptNode",
     "JoinNode",

src/wf_core/models/steps.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import Annotated, Literal
+from typing import Annotated, Literal, Self
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
@@ -163,6 +163,44 @@ class ConditionNode(BaseModel):
     check: Condition
 
 
+class ForeachItemErrorPolicy(BaseModel):
+    """Policy for runtime failures inside one foreach item lineage."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    action: Literal["fail", "skip", "collect"] = "fail"
+    collect_to: StatePath | None = None
+
+    @model_validator(mode="after")
+    def _validate_collect_to(self) -> Self:
+        if self.action == "collect" and self.collect_to is None:
+            raise ValueError("collect item error policy requires collect_to")
+        if self.action != "collect" and self.collect_to is not None:
+            raise ValueError("collect_to is only valid when action='collect'")
+        return self
+
+
+class ForeachConcurrentPolicy(BaseModel):
+    """Concurrency policy for foreach frame admission.
+
+    This controls workflow-level child frame admission. It does not imply
+    thread/process execution for sync node handlers; async runtime can use the
+    same policy to admit simultaneous async handler calls later.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    max_active: int = Field(default=4, ge=1)
+    max_outstanding: int = Field(default=20, ge=1)
+    interrupt: Literal["quiesce"] = "quiesce"
+
+    @model_validator(mode="after")
+    def _validate_capacity(self) -> Self:
+        if self.max_outstanding < self.max_active:
+            raise ValueError("max_outstanding must be >= max_active")
+        return self
+
+
 class ForeachNode(BaseModel):
     """Control-flow step that iterates over an input or state list."""
 
@@ -174,8 +212,50 @@ class ForeachNode(BaseModel):
         description="Workflow input/state/context path that must resolve to a list."
     )
     as_: str = Field(alias="as", description="Context key for the current item.")
-    mode: Literal["serial", "parallel"] = "serial"
-    on_item_error: Literal["fail", "collect", "skip"] = "fail"
+    mode: Literal["serial", "concurrent"] = "serial"
+    item_error: ForeachItemErrorPolicy = Field(
+        default_factory=ForeachItemErrorPolicy
+    )
+    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.action.",
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_policy_shape(cls, data: object) -> object:
+        """Accept old foreach policy names while saving the new canonical shape."""
+        if not isinstance(data, Mapping):
+            return data
+
+        normalized = dict(data)
+        if normalized.get("mode") == "parallel":
+            normalized["mode"] = "concurrent"
+
+        if "parallel" in normalized:
+            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"] = {"action": old_item_error}
+
+        return normalized
+
+    @model_validator(mode="after")
+    def _validate_concurrent_policy(self) -> Self:
+        if self.mode == "concurrent" and self.concurrent is None:
+            raise ValueError("concurrent foreach requires concurrent policy")
+        if self.mode == "serial" and self.concurrent is not None:
+            raise ValueError(
+                "concurrent policy is only valid when mode='concurrent'"
+            )
+        return self
 
 
 class JoinNode(BaseModel):

src/wf_core/runtime/foreach_state.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from wf_core.errors import WorkflowExecutionError
+from wf_core.run_state import ExecutionFrame
+from wf_core.runtime.ops.state import StatePatch
+
+_BARRIER_METADATA_KEY = "foreach_barriers"
+
+
+@dataclass(slots=True)
+class ItemErrorRecord:
+    """Structured runtime failure record for one foreach item."""
+
+    index: int
+    frame_id: str
+    node_id: str
+    error_type: str
+    message: str
+    item: Any = None
+
+    @classmethod
+    def from_metadata(cls, raw: object) -> ItemErrorRecord:
+        if not isinstance(raw, dict):
+            raise WorkflowExecutionError("malformed foreach item error record")
+        try:
+            index = raw["index"]
+            frame_id = raw["frame_id"]
+            node_id = raw["node_id"]
+            error_type = raw["error_type"]
+            message = raw["message"]
+        except KeyError as exc:
+            raise WorkflowExecutionError(
+                f"malformed foreach item error record missing {exc.args[0]!r}"
+            ) from exc
+        if not isinstance(index, int):
+            raise WorkflowExecutionError("malformed foreach item error index")
+        if not all(
+            isinstance(value, str)
+            for value in (frame_id, node_id, error_type, message)
+        ):
+            raise WorkflowExecutionError("malformed foreach item error text fields")
+        return cls(
+            index=index,
+            frame_id=frame_id,
+            node_id=node_id,
+            error_type=error_type,
+            message=message,
+            item=raw.get("item"),
+        )
+
+    def to_metadata(self) -> dict[str, Any]:
+        return {
+            "index": self.index,
+            "frame_id": self.frame_id,
+            "node_id": self.node_id,
+            "error_type": self.error_type,
+            "message": self.message,
+            "item": self.item,
+        }
+
+
+@dataclass(slots=True)
+class PendingItemResult:
+    """Buffered item result waiting for a future foreach barrier commit."""
+
+    index: int
+    frame_id: str
+    status: Literal["succeeded", "failed"]
+    patch: StatePatch = field(default_factory=StatePatch)
+    error: ItemErrorRecord | None = None
+
+    @classmethod
+    def from_metadata(cls, raw: object) -> PendingItemResult:
+        if not isinstance(raw, dict):
+            raise WorkflowExecutionError("malformed pending foreach result")
+        index = raw.get("index")
+        frame_id = raw.get("frame_id")
+        status = raw.get("status")
+        patch_changes = raw.get("patch_changes", {})
+        if not isinstance(index, int):
+            raise WorkflowExecutionError("malformed pending foreach result index")
+        if not isinstance(frame_id, str):
+            raise WorkflowExecutionError("malformed pending foreach result frame id")
+        if status not in {"succeeded", "failed"}:
+            raise WorkflowExecutionError("malformed pending foreach result status")
+        if not isinstance(patch_changes, dict):
+            raise WorkflowExecutionError("malformed pending foreach result patch")
+        raw_error = raw.get("error")
+        return cls(
+            index=index,
+            frame_id=frame_id,
+            status=status,
+            patch=StatePatch(changes=dict(patch_changes)),
+            error=ItemErrorRecord.from_metadata(raw_error)
+            if raw_error is not None
+            else None,
+        )
+
+    def to_metadata(self) -> dict[str, Any]:
+        return {
+            "index": self.index,
+            "frame_id": self.frame_id,
+            "status": self.status,
+            "patch_changes": dict(self.patch.changes),
+            "error": self.error.to_metadata() if self.error is not None else None,
+        }
+
+
+@dataclass(slots=True)
+class ForeachBarrierState:
+    """Resumable state owned by one foreach parent frame."""
+
+    next_index: int = 0
+    active_frame_ids: tuple[str, ...] = ()
+    outstanding_frame_ids: tuple[str, ...] = ()
+    pending_results: dict[int, PendingItemResult] = field(default_factory=dict)
+
+    @classmethod
+    def from_frame(
+        cls,
+        frame: ExecutionFrame,
+        foreach_node_id: str,
+    ) -> ForeachBarrierState | None:
+        """Load one foreach barrier state from frame metadata.
+
+        Missing metadata means the foreach has not started on this frame yet.
+        Malformed metadata means runtime state is corrupt and should fail fast.
+        """
+        all_barriers = frame.metadata.get(_BARRIER_METADATA_KEY)
+        if all_barriers is None:
+            return None
+        if not isinstance(all_barriers, dict):
+            raise WorkflowExecutionError(
+                f"malformed foreach barrier table for frame {frame.id!r}"
+            )
+        raw = all_barriers.get(foreach_node_id)
+        if raw is None:
+            return None
+        if not isinstance(raw, dict):
+            raise WorkflowExecutionError(
+                f"malformed foreach barrier state for frame {frame.id!r}"
+            )
+        return cls.from_metadata(raw)
+
+    @classmethod
+    def from_metadata(cls, raw: object) -> ForeachBarrierState:
+        if not isinstance(raw, dict):
+            raise WorkflowExecutionError("malformed foreach barrier state")
+        next_index = raw.get("next_index")
+        active_frame_ids = _string_tuple(raw.get("active_frame_ids", ()))
+        outstanding_frame_ids = _string_tuple(raw.get("outstanding_frame_ids", ()))
+        pending_results = raw.get("pending_results", {})
+        if not isinstance(next_index, int):
+            raise WorkflowExecutionError("malformed foreach barrier next_index")
+        if not isinstance(pending_results, dict):
+            raise WorkflowExecutionError("malformed foreach barrier pending results")
+        parsed_results: dict[int, PendingItemResult] = {}
+        for raw_index, raw_result in pending_results.items():
+            try:
+                index = int(raw_index)
+            except (TypeError, ValueError) as exc:
+                raise WorkflowExecutionError(
+                    "malformed foreach barrier pending result index"
+                ) from exc
+            parsed_results[index] = PendingItemResult.from_metadata(raw_result)
+        return cls(
+            next_index=next_index,
+            active_frame_ids=active_frame_ids,
+            outstanding_frame_ids=outstanding_frame_ids,
+            pending_results=parsed_results,
+        )
+
+    def save_to_frame(self, frame: ExecutionFrame, foreach_node_id: str) -> None:
+        """Store this barrier state in frame metadata under its foreach node id."""
+        raw = frame.metadata.setdefault(_BARRIER_METADATA_KEY, {})
+        if not isinstance(raw, dict):
+            raise WorkflowExecutionError(
+                f"malformed foreach barrier table for frame {frame.id!r}"
+            )
+        raw[foreach_node_id] = self.to_metadata()
+
+    def to_metadata(self) -> dict[str, Any]:
+        return {
+            "next_index": self.next_index,
+            "active_frame_ids": list(self.active_frame_ids),
+            "outstanding_frame_ids": list(self.outstanding_frame_ids),
+            "pending_results": {
+                str(index): result.to_metadata()
+                for index, result in self.pending_results.items()
+            },
+        }
+
+
+def _string_tuple(raw: object) -> tuple[str, ...]:
+    if isinstance(raw, tuple) and all(isinstance(item, str) for item in raw):
+        return raw
+    if isinstance(raw, list) and all(isinstance(item, str) for item in raw):
+        return tuple(raw)
+    raise WorkflowExecutionError("malformed foreach barrier frame id list")

src/wf_core/runtime/ops/foreach.py

@@ -5,6 +5,7 @@ from wf_core.errors import WorkflowExecutionError
 from wf_core.models.steps import ForeachNode
 from wf_core.models.workflow import Workflow
 from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, StepExecutionResult
+from wf_core.runtime.foreach_state import ForeachBarrierState
 from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
 from wf_core.runtime.ops.frames import frame_context_values
 from wf_core.runtime.ops.index import WorkflowIndex
@@ -23,12 +24,11 @@ def step_foreach(
 ) -> RunState:
     if step.mode != "serial":
         raise WorkflowExecutionError(
-            "parallel foreach execution is not implemented yet"
+            "concurrent foreach execution is not implemented yet"
         )
 
     frame = run.current_frame()
-    progress_map = frame.metadata.setdefault("foreach_progress", {})
-    progress = progress_map.setdefault(step.id, {"index": 0})
+    barrier = ForeachBarrierState.from_frame(frame, step.id) or ForeachBarrierState()
 
     iterable = safe_resolve_path(
         str(step.over),
@@ -41,7 +41,7 @@ def step_foreach(
             f"foreach source {str(step.over)!r} must resolve to a list"
         )
 
-    loop_index = progress["index"]
+    loop_index = barrier.next_index
     if loop_index >= len(iterable):
         outcome = "done"
         next_node_id = index.next_node_id(frame.node_id, outcome)
@@ -64,7 +64,8 @@ def step_foreach(
     loop_start = index.next_node_id(frame.node_id, "loop")
 
     item = iterable[loop_index]
-    progress["index"] = loop_index + 1
+    barrier.next_index = loop_index + 1
+    barrier.save_to_frame(frame, step.id)
     child_id = f"{frame.id}:{step.id}:{loop_index}"
     child_metadata = ForeachIterationMetadata(
         foreach_node_id=step.id,

src/wf_core/runtime/ops/state.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
+from dataclasses import dataclass, field as dataclass_field
 from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
@@ -23,6 +24,24 @@ from wf_core.runtime.ops.schemas import validate_payload_against_schema
 _MISSING = object()
 
 
+@dataclass(slots=True)
+class StatePatch:
+    """Validated state writes produced by one step before commit.
+
+    `changes` is the public trace-facing view: the incoming values keyed by
+    state path. `_prepared_writes` and `_staged_state` are the executor internals
+    needed to commit reducer-aware values atomically without recomputing the
+    patch.
+    """
+
+    changes: dict[str, Any] = dataclass_field(default_factory=dict)
+    _prepared_writes: dict[StatePath, tuple[list[str], Any]] = dataclass_field(
+        default_factory=dict,
+        repr=False,
+    )
+    _staged_state: dict[str, Any] = dataclass_field(default_factory=dict, repr=False)
+
+
 def apply_output_map(
     workflow: Workflow,
     node: NodeUse,
@@ -58,6 +77,27 @@ def apply_output_bindings(
     missing_field_message: str = "node output did not include required field {field}",
 ) -> dict[str, Any]:
     """Prepare and commit one atomic state patch from canonical output bindings."""
+    patch = build_output_patch(
+        workflow,
+        bindings,
+        node_output,
+        state,
+        reducers=reducers,
+        missing_field_message=missing_field_message,
+    )
+    return commit_state_patch(state, patch)
+
+
+def build_output_patch(
+    workflow: Workflow,
+    bindings: Sequence[OutputBinding],
+    node_output: Mapping[str, Any],
+    state: dict[str, Any],
+    *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
+    missing_field_message: str = "node output did not include required field {field}",
+) -> StatePatch:
+    """Build and validate one reducer-aware state patch without mutating state."""
     if has_overlapping_paths(str(binding.target) for binding in bindings):
         raise WorkflowExecutionError(
             "mapped state patch has overlapping destination paths"
@@ -91,9 +131,18 @@ def apply_output_bindings(
     for _destination_path, (key_path, merged_value) in prepared_patch.items():
         safe_set_nested_value(staged_state, key_path, merged_value)
     validate_staged_state_patch(staged_state, prepared_patch, state_fields)
+    return StatePatch(
+        changes={str(path): value for path, value in resolved_patch.items()},
+        _prepared_writes=prepared_patch,
+        _staged_state=staged_state,
+    )
+
+
+def commit_state_patch(state: dict[str, Any], patch: StatePatch) -> dict[str, Any]:
+    """Commit a prevalidated patch to state and return trace-facing changes."""
     state.clear()
-    state.update(staged_state)
-    return {str(path): value for path, value in resolved_patch.items()}
+    state.update(patch._staged_state)
+    return dict(patch.changes)
 
 
 def apply_mapped_state(

src/wf_core/validation/core.py

@@ -75,7 +75,12 @@ def _validate_nodes(
             )
         elif isinstance(node, ForeachNode):
             validate_foreach_node(
-                node, index, report, state_root_fields, input_root_fields
+                node,
+                index,
+                report,
+                state_root_fields,
+                input_root_fields,
+                workflow,
             )
         elif isinstance(node, InterruptNode):
             validate_interrupt_node(

src/wf_core/validation/issues.py

@@ -21,6 +21,7 @@ class ValidationIssueCode(StrEnum):
     EMPTY_CONDITION_ARGS = "empty_condition_args"
     INVALID_CONDITION_PATH = "invalid_condition_path"
     INVALID_FOREACH_SOURCE = "invalid_foreach_source"
+    INVALID_FOREACH_COLLECT_DESTINATION = "invalid_foreach_collect_destination"
     INVALID_INTERRUPT_SOURCE = "invalid_interrupt_source"
     INVALID_INTERRUPT_DESTINATION = "invalid_interrupt_destination"
 

src/wf_core/validation/outcomes.py

@@ -13,7 +13,10 @@ def declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> set
     if step.type == "condition":
         return {"true", "false"}
     if step.type == "foreach":
-        return {"loop", "done"}
+        outcomes = {"loop", "done"}
+        if step.item_error.action in {"skip", "collect"}:
+            outcomes.add("completed_with_errors")
+        return outcomes
     if step.type == "join":
         return {"done"}
     if isinstance(step, InterruptNode):

src/wf_core/validation/steps.py

@@ -149,6 +149,7 @@ def validate_foreach_node(
     report: ValidationReport,
     state_root_fields: set[str],
     input_root_fields: set[str],
+    workflow: Workflow,
 ) -> None:
     if not is_valid_source_path(node.over, state_root_fields, input_root_fields):
         report.add(
@@ -156,6 +157,27 @@ def validate_foreach_node(
             f"nodes[{index}].over",
             "foreach source path must start with input. or state. and reference a declared root field",
         )
+    if node.item_error.action != "collect":
+        return
+    collect_to = node.item_error.collect_to
+    if collect_to is None:
+        return
+    destination_root = _state_destination_root(collect_to)
+    state_fields = workflow.state_schema.field_index()
+    field = state_fields.get(collect_to)
+    if destination_root is None or destination_root not in state_root_fields:
+        report.add(
+            ValidationIssueCode.INVALID_FOREACH_COLLECT_DESTINATION,
+            f"nodes[{index}].item_error.collect_to",
+            "collect_to must start with state. and reference a declared state field",
+        )
+        return
+    if field is None or field.type != "array":
+        report.add(
+            ValidationIssueCode.INVALID_FOREACH_COLLECT_DESTINATION,
+            f"nodes[{index}].item_error.collect_to",
+            "collect_to must reference a declared array state field",
+        )
 
 
 def validate_interrupt_node(