overlay reads + reducer write metadata

according to chat it is so i myself idk

src/wf_core/runtime/foreach_state.py

@@ -4,7 +4,9 @@ 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.models.reducers import ReducerRef
+from wf_core.paths import StatePath
+from wf_core.run_state import ExecutionFrame, StateWrite
 from wf_core.runtime.ops.state import StatePatch
 from wf_core.runtime.scheduler import ForeachIterationMetadata
 
@@ -85,6 +87,7 @@ class PendingItemResult:
                 f"malformed pending foreach result missing {exc.args[0]!r}"
             ) from exc
         patch_changes = raw.get("patch_changes", {})
+        patch_writes = raw.get("patch_writes")
         if not isinstance(index, int) or index < 0:
             raise WorkflowExecutionError("malformed pending foreach result index")
         if not isinstance(frame_id, str):
@@ -93,12 +96,20 @@ class PendingItemResult:
             raise WorkflowExecutionError("malformed pending foreach result status")
         if not isinstance(patch_changes, dict):
             raise WorkflowExecutionError("malformed pending foreach result patch")
+        if patch_writes is not None and not isinstance(patch_writes, list):
+            raise WorkflowExecutionError("malformed pending foreach result writes")
         raw_error = raw.get("error")
         return cls(
             index=index,
             frame_id=frame_id,
             status=status,
-            patch=StatePatch(changes=patch_changes),
+            patch=(
+                StatePatch(
+                    writes=[_state_write_from_metadata(item) for item in patch_writes]
+                )
+                if patch_writes is not None
+                else StatePatch(changes=patch_changes)
+            ),
             error=(
                 ItemErrorRecord.from_metadata(raw_error)
                 if raw_error is not None
@@ -112,6 +123,9 @@ class PendingItemResult:
             "frame_id": self.frame_id,
             "status": self.status,
             "patch_changes": dict(self.patch.changes),
+            "patch_writes": [
+                _state_write_to_metadata(write) for write in self.patch.writes
+            ],
             "error": self.error.to_metadata() if self.error is not None else None,
         }
 
@@ -298,3 +312,53 @@ def _string_tuple(raw: object) -> tuple[str, ...]:
     if isinstance(raw, list) and all(isinstance(item, str) for item in raw):
         return tuple(raw)
     raise WorkflowExecutionError("malformed foreach barrier frame id list")
+
+
+def _state_write_from_metadata(raw: object) -> StateWrite:
+    """Parse one persisted item-lineage write record.
+
+    Barrier metadata must keep reducer-visible values across interrupt/resume;
+    reconstructing from `patch_changes` would downgrade reducer writes to
+    replace-style incoming values.
+    """
+    if not isinstance(raw, dict):
+        raise WorkflowExecutionError("malformed pending foreach write")
+    try:
+        path = raw["path"]
+        incoming_value = raw["incoming_value"]
+        visible_value = raw["visible_value"]
+        reducer = raw["reducer"]
+    except KeyError as exc:
+        raise WorkflowExecutionError(
+            f"malformed pending foreach write missing {exc.args[0]!r}"
+        ) from exc
+    try:
+        return StateWrite(
+            path=_state_path_from_metadata(path),
+            incoming_value=incoming_value,
+            visible_value=visible_value,
+            reducer=ReducerRef.model_validate(reducer),
+        )
+    except Exception as exc:
+        raise WorkflowExecutionError("malformed pending foreach write") from exc
+
+
+def _state_write_to_metadata(write: StateWrite) -> dict[str, Any]:
+    """Serialize one item-lineage write without relying on dotted display paths."""
+    return {
+        "path": {"root": "state", "parts": list(write.path.parts)},
+        "incoming_value": write.incoming_value,
+        "visible_value": write.visible_value,
+        "reducer": write.reducer.model_dump(mode="json"),
+    }
+
+
+def _state_path_from_metadata(raw: object) -> StatePath:
+    if isinstance(raw, str):
+        return StatePath.parse(raw)
+    if not isinstance(raw, dict) or raw.get("root") != "state":
+        raise WorkflowExecutionError("malformed pending foreach write path")
+    parts = raw.get("parts")
+    if not isinstance(parts, list) or not all(isinstance(part, str) for part in parts):
+        raise WorkflowExecutionError("malformed pending foreach write path")
+    return StatePath(tuple(parts))

src/wf_core/runtime/ops/overlays.py

@@ -3,7 +3,6 @@ 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
@@ -33,7 +32,6 @@ def state_view_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]
     # Correctness first: this full copy isolates sibling reads. If state grows
     # large, replace this with a lazy/copy-on-write overlay.
     state_view = deepcopy(run.state)
-    for destination, value in pending.patch.changes.items():
+    for write in pending.patch.writes:
-        path = StatePath.parse(destination)
+        safe_set_nested_value(state_view, list(write.path.parts), write.visible_value)
-        safe_set_nested_value(state_view, list(path.parts), value)
     return state_view

tests/core/test_concurrent_foreach.py

@@ -190,13 +190,6 @@ def test_sync_concurrent_foreach_barrier_replays_add_reducer_inputs() -> None:
     assert foreach_entries[-1].state_changes["state.number"] == 6
 
 
-@pytest.mark.xfail(
-    reason=(
-        "Current foreach overlays use StatePatch.changes, which stores incoming "
-        "reducer values; lineage StateWrite.visible_value should make this pass."
-    ),
-    strict=True,
-)
 def test_sync_concurrent_foreach_same_item_reads_add_reducer_visible_value() -> None:
     workflow = _same_item_reducer_visibility_workflow()
 
@@ -220,9 +213,6 @@ def test_sync_concurrent_foreach_same_item_reads_add_reducer_visible_value() ->
     assert stage_entry.resolved_input["current_number"] == 2
     assert stage_entry.state_changes == {}
     read_entry = next(entry for entry in run.trace if entry.node_id == "read_number")
-    # Current limitation: foreach overlays use StatePatch.changes, which stores
-    # the incoming reducer value. The future lineage StateWrite model should let
-    # this same item read the reducer-visible value 5 instead.
     assert read_entry.resolved_input["number"] == 5
     assert run.state["seen_number"] == [5]
 

tests/core/test_foreach_barrier_state.py

@@ -3,7 +3,9 @@ from __future__ import annotations
 import pytest
 
 from wf_core.errors import WorkflowExecutionError
-from wf_core.run_state import ExecutionFrame
+from wf_core.models.reducers import ReducerRef
+from wf_core.paths import StatePath
+from wf_core.run_state import ExecutionFrame, StateWrite
 from wf_core.runtime.foreach_state import (
     ForeachBarrierState,
     ItemErrorRecord,
@@ -48,6 +50,41 @@ def test_foreach_barrier_state_round_trips_through_frame_metadata() -> None:
     assert loaded.pending_results[1].error.message == "bad item"
 
 
+def test_foreach_barrier_state_round_trips_reducer_write_records() -> None:
+    frame = ExecutionFrame(id="root", kind="root", node_id="each")
+    barrier = ForeachBarrierState(
+        next_index=1,
+        mode="concurrent",
+        pending_results={
+            0: PendingItemResult(
+                index=0,
+                frame_id="child-0",
+                status="succeeded",
+                patch=StatePatch(
+                    writes=[
+                        StateWrite(
+                            path=StatePath(("count",)),
+                            incoming_value=3,
+                            visible_value=5,
+                            reducer=ReducerRef(name="wf.std.add"),
+                        )
+                    ]
+                ),
+            )
+        },
+    )
+
+    barrier.save_to_frame(frame, "each")
+    loaded = ForeachBarrierState.from_frame(frame, "each")
+
+    assert loaded is not None
+    write = loaded.pending_results[0].patch.writes[0]
+    assert write.path == StatePath(("count",))
+    assert write.incoming_value == 3
+    assert write.visible_value == 5
+    assert write.reducer.name == "wf.std.add"
+
+
 def test_foreach_barrier_state_returns_none_when_missing() -> None:
     frame = ExecutionFrame(id="root", kind="root", node_id="each")
 
@@ -109,6 +146,7 @@ def test_foreach_barrier_accumulates_multiple_patches_for_one_item() -> None:
     result = barrier.pending_results[0]
     assert result.patch.changes["state.count"] == 1
     assert result.patch.changes["state.name"] == "a"
+    assert len(result.patch.writes) == 2
 
 
 def test_foreach_barrier_rejects_item_result_frame_mismatch() -> None: