From 08cd4e1e5ef9db9b51177bc5730772ab415a2219 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 11:43:41 +0700 Subject: [PATCH] audit: walk serial owners in write routing, harden result identity --- .../2026-09-04-foreach-back-edge-design.md | 9 +- src/wf_core/runtime/foreach_state.py | 25 ++- src/wf_core/runtime/lineage.py | 59 +++--- src/wf_core/runtime/ops/foreach.py | 4 +- tests/core/test_foreach_back_edges.py | 168 ++++++++++++++++++ tests/core/test_foreach_barrier_state.py | 48 +++++ 6 files changed, 278 insertions(+), 35 deletions(-) diff --git a/docs/superpowers/specs/2026-09-04-foreach-back-edge-design.md b/docs/superpowers/specs/2026-09-04-foreach-back-edge-design.md index 10180cca..1eb8b0ca 100644 --- a/docs/superpowers/specs/2026-09-04-foreach-back-edge-design.md +++ b/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 diff --git a/src/wf_core/runtime/foreach_state.py b/src/wf_core/runtime/foreach_state.py index 21fc75d8..89dd5c54 100644 --- a/src/wf_core/runtime/foreach_state.py +++ b/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 diff --git a/src/wf_core/runtime/lineage.py b/src/wf_core/runtime/lineage.py index e9ade5b1..fb49ebe3 100644 --- a/src/wf_core/runtime/lineage.py +++ b/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]: diff --git a/src/wf_core/runtime/ops/foreach.py b/src/wf_core/runtime/ops/foreach.py index e94961a5..52989dcf 100644 --- a/src/wf_core/runtime/ops/foreach.py +++ b/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, diff --git a/tests/core/test_foreach_back_edges.py b/tests/core/test_foreach_back_edges.py index fb6ba2ce..eff749c0 100644 --- a/tests/core/test_foreach_back_edges.py +++ b/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", diff --git a/tests/core/test_foreach_barrier_state.py b/tests/core/test_foreach_barrier_state.py index 6df3e5ca..23dd9a8c 100644 --- a/tests/core/test_foreach_barrier_state.py +++ b/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"