audit: walk serial owners in write routing, harden result identity

docs/superpowers/specs/2026-09-04-foreach-back-edge-design.md

@@ -434,7 +434,14 @@ may independently return and complete the item.
 Back-edge return changes control representation, not state semantics.
 Iteration writes remain buffered in the item lineage. Serial behavior and the
 concurrent barrier continue to commit or merge those writes according to the
-accepted concurrent-foreach ADR and declared reducers. The completed item is
+accepted concurrent-foreach ADR and declared reducers. One shared helper
+routes every item write: it climbs through each serial owner to the scope
+root, where it commits, or stops at the first concurrent item boundary,
+where it buffers for that barrier to merge (the concurrent barrier finish
+routes its combined patch through the same helper, so nested serial owners
+cannot strand it). Parent cycles, missing parents, and orphaned item frames
+fail closed. Buffered failure records must carry an error whose index and
+frame match the enclosing result. The completed item is
 registered with its barrier at the owner back-edge, keyed by the returning
 frame rather than by whichever operation ran last, so node, subgraph, and
 nested-control endings all count. A return naming a closed or superseded

src/wf_core/runtime/foreach_state.py

@@ -138,16 +138,21 @@ class PendingItemResult:
                 raise WorkflowExecutionError(
                     "malformed pending foreach result lineage id"
                 )
+        error = (
+            ItemErrorRecord.from_metadata(raw_error) if raw_error is not None else None
+        )
+        if error is not None and (error.index != index or error.frame_id != frame_id):
+            raise WorkflowExecutionError(
+                "malformed pending foreach result: error identity "
+                f"(index {error.index!r}, frame {error.frame_id!r}) does not "
+                f"match enclosing result (index {index!r}, frame {frame_id!r})"
+            )
         return cls(
             index=index,
             frame_id=frame_id,
             status=status,
             lineage_id=lineage_id,
-            error=(
-                ItemErrorRecord.from_metadata(raw_error)
-                if raw_error is not None
-                else None
-            ),
+            error=error,
         )
 
     def to_metadata(self) -> dict[str, Any]:
@@ -443,10 +448,16 @@ def item_frame_owner(frame: ExecutionFrame) -> ForeachItemOwner | None:
     """Return the named foreach ownership record for item frames.
 
     Malformed item metadata fails closed via ``ForeachIterationMetadata``;
-    only non-item frames return ``None``.
+    only genuinely non-item frames return ``None``. An item frame without
+    a parent is corrupt state and raises rather than masquerading as an
+    ordinary frame.
     """
-    if frame.kind != "foreach_iteration" or frame.parent_frame_id is None:
+    if frame.kind != "foreach_iteration":
         return None
+    if frame.parent_frame_id is None:
+        raise WorkflowExecutionError(
+            f"foreach item frame {frame.id!r} is missing its parent frame"
+        )
     metadata = ForeachIterationMetadata.from_frame(frame)
     if metadata is None:
         return None

src/wf_core/runtime/lineage.py

@@ -94,38 +94,47 @@ def commit_foreach_aware_patch(
 ) -> dict[str, Any]:
     """Commit one write patch with foreach-aware routing.
 
-    Ordinary frames commit (or buffer) through their own lineage. Serial
-    item writes commit through the parent scope so they land in root state;
-    concurrent item writes stay buffered in the item lineage for the barrier
-    to merge. Malformed ownership, missing parents, and closed or
-    superseded activations fail closed.
+    Ordinary frames commit (or buffer) through their own lineage. The walk
+    climbs through every serial item owner until it reaches either the
+    workflow/subgraph scope root, where it commits, or a concurrent item
+    boundary, where it buffers in that item lineage for the barrier to
+    merge. Malformed ownership, missing parents, parent cycles, and closed
+    or superseded activations fail closed.
     """
     from wf_core.runtime.foreach_state import (
         item_frame_owner,
         require_foreach_activation,
     )
 
-    owner = item_frame_owner(frame)
-    if owner is None:
-        return commit_patch_for_frame(run, frame, patch)
-    parent_frame = run.frames.get(owner.parent_frame_id)
-    if parent_frame is None:
-        raise WorkflowExecutionError(
-            "foreach item state references missing parent frame "
-            f"{owner.parent_frame_id!r} for child frame {frame.id!r}"
+    current = frame
+    seen: set[str] = set()
+    while True:
+        owner = item_frame_owner(current)
+        if owner is None:
+            return commit_patch_for_frame(run, current, patch)
+        if current.id in seen:
+            raise WorkflowExecutionError(
+                f"cycle detected in foreach parent chain at frame {current.id!r}"
+            )
+        seen.add(current.id)
+        parent_frame = run.frames.get(owner.parent_frame_id)
+        if parent_frame is None:
+            raise WorkflowExecutionError(
+                "foreach item state references missing parent frame "
+                f"{owner.parent_frame_id!r} for child frame {current.id!r}"
+            )
+        activation = require_foreach_activation(
+            parent_frame, owner.foreach_node_id, owner.activation_id
         )
-    activation = require_foreach_activation(
-        parent_frame, owner.foreach_node_id, owner.activation_id
-    )
-    if activation.barrier.mode == "concurrent":
-        append_lineage_writes(
-            run,
-            scope_id=frame.scope_id,
-            lineage_id=frame.lineage_id,
-            writes=patch.writes,
-        )
-        return {}
-    return commit_patch_for_frame(run, parent_frame, patch)
+        if activation.barrier.mode == "concurrent":
+            append_lineage_writes(
+                run,
+                scope_id=current.scope_id,
+                lineage_id=current.lineage_id,
+                writes=patch.writes,
+            )
+            return {}
+        current = parent_frame
 
 
 def scope_state_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]:

src/wf_core/runtime/ops/foreach.py

@@ -18,7 +18,7 @@ from wf_core.runtime.foreach_state import (
 )
 from wf_core.runtime.lineage import (
     add_lineage,
-    commit_patch_for_frame,
+    commit_foreach_aware_patch,
     lineage_patch,
     scope_input_for_frame,
 )
@@ -379,7 +379,7 @@ def _finish_concurrent_foreach(
         state_view_for_frame(run, frame),
         reducers=reducers,
     )
-    state_changes = commit_patch_for_frame(run, frame, combined)
+    state_changes = commit_foreach_aware_patch(run, frame, combined)
     append_step_result_trace(
         run,
         frame_id=frame.id,

tests/core/test_foreach_back_edges.py

@@ -427,6 +427,174 @@ def test_nested_foreach_returns_inner_then_outer() -> None:
     assert run.state["seen"][:3] == [1, 2, "a"]
 
 
+def _nested_mode_workflow(*, outer_mode: str, inner_mode: str) -> Workflow:
+    def _foreach(node_id: str, *, over: str, alias: str, mode: str) -> ForeachNode:
+        payload: dict[str, Any] = {
+            "id": node_id,
+            "type": "foreach",
+            "over": over,
+            "as": alias,
+            "mode": mode,
+        }
+        if mode == "concurrent":
+            payload["concurrent"] = {"max_active": 2, "max_outstanding": 2}
+        return ForeachNode.model_validate(payload)
+
+    return Workflow(
+        name="nested_foreach_modes",
+        input_schema=SchemaRef(type="object", properties={}),
+        state_schema=StateSchema.from_field_map(
+            {
+                "items": StateField(type="array"),
+                "inner_items": StateField(type="array"),
+                "seen": StateField(
+                    type="array", reducer=ReducerRef(name="wf.std.append")
+                ),
+            }
+        ),
+        output_schema=SchemaRef(type="object", properties={"seen": {"type": "array"}}),
+        node_defs=[
+            NodeDef(
+                name="work",
+                input_schema=SchemaRef(
+                    type="object", properties={"value": {}}, required=["value"]
+                ),
+                output_schema=SchemaRef(
+                    type="object", properties={"seen": {}}, required=["seen"]
+                ),
+                outcomes=["ok"],
+            )
+        ],
+        start="outer",
+        nodes=[
+            _foreach("outer", over="state.items", alias="outer_item", mode=outer_mode),
+            _foreach(
+                "inner",
+                over="state.inner_items",
+                alias="inner_item",
+                mode=inner_mode,
+            ),
+            NodeUse.model_validate(
+                {
+                    "id": "work",
+                    "type": "node",
+                    "node": "work",
+                    "input": [{"target": "value", "path": "context.inner_item"}],
+                    "output": [{"source": "seen", "target": "state.seen"}],
+                }
+            ),
+        ],
+        edges=[
+            Edge.model_validate({"from": "outer", "outcome": "loop", "to": "inner"}),
+            Edge.model_validate({"from": "inner", "outcome": "loop", "to": "work"}),
+            Edge.model_validate({"from": "work", "outcome": "ok", "to": "inner"}),
+            # No intermediate writer: the inner barrier (or serial return)
+            # must route inner writes to the scope root on its own.
+            Edge.model_validate({"from": "inner", "outcome": "done", "to": "outer"}),
+            Edge.model_validate({"from": "outer", "outcome": "done", "to": END}),
+        ],
+    )
+
+
+@pytest.mark.parametrize(
+    ("outer_mode", "inner_mode"),
+    [
+        ("serial", "serial"),
+        ("serial", "concurrent"),
+        ("concurrent", "serial"),
+        ("concurrent", "concurrent"),
+    ],
+)
+def test_nested_foreach_preserves_inner_writes_in_all_modes(
+    outer_mode: str, inner_mode: str
+) -> None:
+    """Inner writes must reach root state whatever the nesting modes are.
+
+    Serial owners commit through the scope root; concurrent owners buffer
+    for their barrier. Every inner write (1, 2 per outer item) must survive
+    even with no intermediate writer to replay-rescue stranded lineages.
+    """
+    workflow = _nested_mode_workflow(outer_mode=outer_mode, inner_mode=inner_mode)
+
+    run = execute_workflow(
+        workflow,
+        {"items": ["a", "b"], "inner_items": [1, 2]},
+        {
+            "work": lambda payload, _ctx: {
+                "outcome": "ok",
+                "output": {"seen": payload["value"]},
+            }
+        },
+    )
+
+    assert run.status == RunStatus.COMPLETED
+    assert sorted(run.state.get("seen") or [], key=repr) == sorted(
+        [1, 2, 1, 2], key=repr
+    )
+
+
+def test_item_frame_owner_rejects_missing_parent_frame() -> None:
+    """A foreach_iteration frame without a parent is malformed, not ordinary."""
+    frame = ExecutionFrame(
+        id="orphan",
+        kind="foreach_iteration",
+        node_id="work",
+        parent_frame_id=None,
+        metadata={
+            "foreach_node_id": "each",
+            "activation_id": "root:each#0",
+            "loop_index": 0,
+            "loop_item": "a",
+            "loop_alias": "item",
+        },
+    )
+
+    with pytest.raises(WorkflowExecutionError, match="parent"):
+        item_frame_owner(frame)
+
+
+def test_foreach_aware_patch_rejects_parent_cycle() -> None:
+    """A cyclic item-parent chain fails closed instead of looping forever."""
+    from wf_core.runtime.foreach_state import load_or_begin_foreach_activation
+    from wf_core.runtime.lineage import commit_foreach_aware_patch
+    from wf_core.runtime.ops.state import StatePatch
+
+    frame_a = ExecutionFrame(id="frame-a", kind="foreach_iteration", node_id="work")
+    frame_b = ExecutionFrame(id="frame-b", kind="foreach_iteration", node_id="work")
+    activation_on_b = load_or_begin_foreach_activation(frame_b, "each", mode="serial")
+    activation_on_a = load_or_begin_foreach_activation(frame_a, "each", mode="serial")
+    frame_a.parent_frame_id = "frame-b"
+    frame_a.metadata.update(
+        {
+            "foreach_node_id": "each",
+            "activation_id": activation_on_b.id,
+            "loop_index": 0,
+            "loop_item": "a",
+            "loop_alias": "item",
+        }
+    )
+    frame_b.parent_frame_id = "frame-a"
+    frame_b.metadata.update(
+        {
+            "foreach_node_id": "each",
+            "activation_id": activation_on_a.id,
+            "loop_index": 0,
+            "loop_item": "a",
+            "loop_alias": "item",
+        }
+    )
+    run = RunState(
+        workflow_name="parent_cycle",
+        status=RunStatus.RUNNING,
+        workflow_input={},
+        state={},
+        frames={"frame-a": frame_a, "frame-b": frame_b},
+    )
+
+    with pytest.raises(WorkflowExecutionError, match="cycle"):
+        commit_foreach_aware_patch(run, frame_a, StatePatch(changes={}))
+
+
 def test_reentering_foreach_uses_fresh_activation_and_item_frames() -> None:
     workflow = Workflow(
         name="foreach_reentry",

tests/core/test_foreach_barrier_state.py

@@ -340,3 +340,51 @@ def test_pending_item_result_rejects_index_key_mismatch() -> None:
 
     with pytest.raises(WorkflowExecutionError, match="index mismatch"):
         ForeachBarrierState.from_metadata(raw["each"]["active"]["barrier"])
+
+
+def _failed_result(
+    *, index: int, frame_id: str, error_index: int, error_frame: str
+) -> dict:
+    return {
+        "index": index,
+        "frame_id": frame_id,
+        "status": "failed",
+        "lineage_id": None,
+        "error": {
+            "index": error_index,
+            "frame_id": error_frame,
+            "node_id": "work",
+            "error_type": "ValueError",
+            "message": "bad",
+        },
+    }
+
+
+def test_pending_item_result_rejects_error_index_mismatch() -> None:
+    with pytest.raises(WorkflowExecutionError, match="error.*index|index.*error"):
+        PendingItemResult.from_metadata(
+            _failed_result(
+                index=0, frame_id="child-0", error_index=7, error_frame="child-0"
+            )
+        )
+
+
+def test_pending_item_result_rejects_error_frame_mismatch() -> None:
+    with pytest.raises(WorkflowExecutionError, match="error.*frame|frame.*error"):
+        PendingItemResult.from_metadata(
+            _failed_result(
+                index=0, frame_id="child-0", error_index=0, error_frame="other"
+            )
+        )
+
+
+def test_pending_item_result_accepts_matching_error_identity() -> None:
+    result = PendingItemResult.from_metadata(
+        _failed_result(
+            index=0, frame_id="child-0", error_index=0, error_frame="child-0"
+        )
+    )
+
+    assert result.error is not None
+    assert result.error.index == 0
+    assert result.error.frame_id == "child-0"