overlays

src/wf_core/runtime/foreach_state.py

@@ -239,17 +239,28 @@ class ForeachBarrierState:
     def add_success_patch(
         self, *, index: int, frame_id: str, patch: StatePatch
     ) -> None:
-        """Buffer one successful item patch by item index."""
-        if index in self.pending_results:
-            raise WorkflowExecutionError(
-                f"foreach item result for index {index!r} already recorded"
+        """Buffer or extend successful item patches by item index.
+
+        A multi-step item body can produce multiple node patches. They are
+        accumulated for the same item lineage and replayed by the barrier in
+        item index order. Do not merge `_prepared_writes` here: the barrier
+        intentionally replays public changes against one staged parent state.
+        """
+        existing = self.pending_results.get(index)
+        if existing is None:
+            self.pending_results[index] = PendingItemResult(
+                index=index,
+                frame_id=frame_id,
+                status="succeeded",
+                patch=patch,
             )
-        self.pending_results[index] = PendingItemResult(
-            index=index,
-            frame_id=frame_id,
-            status="succeeded",
-            patch=patch,
-        )
+            return
+        if existing.frame_id != frame_id:
+            raise WorkflowExecutionError(
+                f"foreach item result for index {index!r} belongs to frame "
+                f"{existing.frame_id!r}, got {frame_id!r}"
+            )
+        existing.patch.changes.update(patch.changes)
 
 
 def item_frame_owner(frame: ExecutionFrame) -> tuple[str, str, int] | None:

src/wf_core/runtime/ops/foreach.py

@@ -4,7 +4,7 @@ from collections.abc import Mapping
 
 from wf_core.conditions import safe_resolve_path
 from wf_core.errors import WorkflowExecutionError
-from wf_core.models.steps import ForeachNode, NodeUse
+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
@@ -18,7 +18,6 @@ from wf_core.runtime.scheduler import (
     add_frame,
     block_frame_on_children,
 )
-from wf_core.tokens import END
 
 
 def step_foreach(
@@ -127,8 +126,6 @@ def _step_foreach_concurrent(
         )
     if step.concurrent is None:
         raise WorkflowExecutionError("concurrent foreach requires concurrent policy")
-    _validate_single_node_loop_body(index, step)
-
     frame = run.current_frame()
     barrier = ForeachBarrierState.from_frame(frame, step.id)
     if barrier is None:
@@ -182,27 +179,6 @@ def _resolve_foreach_iterable(
     return iterable
 
 
-def _validate_single_node_loop_body(index: WorkflowIndex, step: ForeachNode) -> None:
-    """Reject multi-step concurrent item bodies until item overlays are real.
-
-    The current slice has a no-op item-state overlay seam. Without a real overlay,
-    multi-node item bodies would read stale parent state after earlier item-local
-    writes, so V1 only allows loop -> one node -> END.
-    """
-    loop_start = index.next_node_id(step.id, "loop")
-    loop_step = index.nodes_by_id.get(loop_start)
-    if not isinstance(loop_step, NodeUse):
-        raise WorkflowExecutionError(
-            "concurrent foreach v1 only supports loop bodies with one node"
-        )
-    node_def = index.node_defs[loop_step.node]
-    for outcome in node_def.outcomes:
-        if index.next_node_id(loop_step.id, outcome) != END:
-            raise WorkflowExecutionError(
-                "concurrent foreach v1 only supports loop bodies with one node"
-            )
-
-
 def _finish_completed_children(run: RunState, barrier: ForeachBarrierState) -> None:
     for child_id in tuple(barrier.outstanding_frame_ids):
         child = run.frames[child_id]

src/wf_core/runtime/ops/nodes.py

@@ -31,7 +31,7 @@ def _resolve_node_execution(
     run: RunState,
     node: NodeUse,
     node_def: NodeDef,
-) -> tuple[dict[str, Any], RuntimeContext]:
+) -> tuple[dict[str, Any], RuntimeContext, dict[str, Any]]:
     frame = run.current_frame()
     context_values = frame_context_values(frame)
     state_view = state_view_for_frame(run, frame)
@@ -65,7 +65,7 @@ def _resolve_node_execution(
         activated_incoming_edge=frame.activated_incoming_edge,
         metadata=dict(frame.metadata),
     )
-    return resolved_input, context
+    return resolved_input, context, state_view
 
 
 def _finalize_node_execution(
@@ -76,6 +76,7 @@ def _finalize_node_execution(
     node_def: NodeDef,
     resolved_input: dict[str, Any],
     raw_result: NodeResult | dict[str, Any],
+    state_view: dict[str, Any],
     reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> StepExecutionResult:
     result = coerce_node_result(raw_result)
@@ -92,7 +93,7 @@ def _finalize_node_execution(
         workflow,
         node.output,
         result.output,
-        run.state,
+        state_view,
         reducers=reducers,
     )
     owner = item_frame_owner(run.current_frame())
@@ -134,7 +135,7 @@ def execute_node_use(
             f"no handler registered for node def {node.node!r}"
         )
 
-    resolved_input, context = _resolve_node_execution(
+    resolved_input, context, state_view = _resolve_node_execution(
         workflow=workflow,
         run=run,
         node=node,
@@ -148,6 +149,7 @@ def execute_node_use(
         node_def=node_def,
         resolved_input=resolved_input,
         raw_result=raw_result,
+        state_view=state_view,
         reducers=reducers,
     )
 
@@ -166,7 +168,7 @@ async def execute_node_use_async(
             f"no handler registered for node def {node.node!r}"
         )
 
-    resolved_input, context = _resolve_node_execution(
+    resolved_input, context, state_view = _resolve_node_execution(
         workflow=workflow,
         run=run,
         node=node,
@@ -184,6 +186,7 @@ async def execute_node_use_async(
         node_def=node_def,
         resolved_input=resolved_input,
         raw_result=cast(NodeResult | dict[str, Any], raw_result),
+        state_view=state_view,
         reducers=reducers,
     )
 

src/wf_core/runtime/ops/overlays.py

@@ -1,16 +1,37 @@
 from __future__ import annotations
 
+from copy import deepcopy
 from typing import Any
 
+from wf_core.paths import StatePath
 from wf_core.run_state import ExecutionFrame, RunState
+from wf_core.runtime.foreach_state import ForeachBarrierState, item_frame_owner
+from wf_core.runtime.ops.state import safe_set_nested_value
 
 
 def state_view_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]:
-    """Return the state view visible to one execution frame.
+    """Return committed parent state plus this frame's item-local overlay.
 
-    This is intentionally a no-op seam for concurrent foreach V1. The first
-    sync-concurrent slice only supports single-node item bodies, so item frames
-    do not need to read their own prior buffered writes yet. The overlay slice
-    should replace this with parent-state plus item-local staged writes.
+    Concurrent foreach item frames buffer writes in the parent barrier until the
+    foreach barrier commits. Later nodes in the same item must read those
+    earlier writes, while sibling item frames must not see them.
     """
-    return run.state
+    owner = item_frame_owner(frame)
+    if owner is None:
+        return run.state
+
+    parent_frame_id, foreach_node_id, item_index = owner
+    parent_frame = run.frames[parent_frame_id]
+    barrier = ForeachBarrierState.from_frame(parent_frame, foreach_node_id)
+    if barrier is None or barrier.mode != "concurrent":
+        return run.state
+
+    pending = barrier.pending_results.get(item_index)
+    if pending is None:
+        return run.state
+
+    state_view = deepcopy(run.state)
+    for destination, value in pending.patch.changes.items():
+        path = StatePath.parse(destination)
+        safe_set_nested_value(state_view, list(path.parts), value)
+    return state_view