From 47e57ce598c3981ea4c18eeccea9c05172194144 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 11:59:35 +0700 Subject: [PATCH] audit: validate ancestry before buffering, tighten nested ordering test --- .../2026-09-04-foreach-back-edge-design.md | 7 +- src/wf_core/runtime/lineage.py | 29 +++-- tests/core/test_foreach_back_edges.py | 116 ++++++++++++++++-- 3 files changed, 129 insertions(+), 23 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 1eb8b0ca..ea1906a8 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 @@ -432,9 +432,10 @@ may independently return and complete the item. ## State and Failure Behavior 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. One shared helper +Concurrent iteration writes remain buffered in the item lineage for the +barrier to merge, while serial owners pass writes outward to the scope +root, which commits them according to the 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 diff --git a/src/wf_core/runtime/lineage.py b/src/wf_core/runtime/lineage.py index fb49ebe3..dcbe077a 100644 --- a/src/wf_core/runtime/lineage.py +++ b/src/wf_core/runtime/lineage.py @@ -98,8 +98,11 @@ def commit_foreach_aware_patch( 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. + merge. The whole ancestry is validated first: the write lands only + after the chain reaches an acyclic non-item ancestor, so a parent + cycle fails closed even when it passes through a concurrent + boundary. Malformed ownership, missing parents, parent cycles, and + closed or superseded activations fail closed. """ from wf_core.runtime.foreach_state import ( item_frame_owner, @@ -108,10 +111,11 @@ def commit_foreach_aware_patch( current = frame seen: set[str] = set() + buffer_in: ExecutionFrame | None = None while True: owner = item_frame_owner(current) if owner is None: - return commit_patch_for_frame(run, current, patch) + break if current.id in seen: raise WorkflowExecutionError( f"cycle detected in foreach parent chain at frame {current.id!r}" @@ -126,15 +130,18 @@ def commit_foreach_aware_patch( activation = require_foreach_activation( parent_frame, owner.foreach_node_id, owner.activation_id ) - if activation.barrier.mode == "concurrent": - append_lineage_writes( - run, - scope_id=current.scope_id, - lineage_id=current.lineage_id, - writes=patch.writes, - ) - return {} + if buffer_in is None and activation.barrier.mode == "concurrent": + buffer_in = current current = parent_frame + if buffer_in is not None: + append_lineage_writes( + run, + scope_id=buffer_in.scope_id, + lineage_id=buffer_in.lineage_id, + writes=patch.writes, + ) + return {} + return commit_patch_for_frame(run, current, patch) def scope_state_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]: diff --git a/tests/core/test_foreach_back_edges.py b/tests/core/test_foreach_back_edges.py index eff749c0..e3f75d5e 100644 --- a/tests/core/test_foreach_back_edges.py +++ b/tests/core/test_foreach_back_edges.py @@ -497,22 +497,26 @@ def _nested_mode_workflow(*, outer_mode: str, inner_mode: str) -> Workflow: @pytest.mark.parametrize( - ("outer_mode", "inner_mode"), + ("outer_mode", "inner_mode", "exact_order"), [ - ("serial", "serial"), - ("serial", "concurrent"), - ("concurrent", "serial"), - ("concurrent", "concurrent"), + ("serial", "serial", True), + ("serial", "concurrent", True), + ("concurrent", "serial", False), + ("concurrent", "concurrent", False), ], ) def test_nested_foreach_preserves_inner_writes_in_all_modes( - outer_mode: str, inner_mode: str + outer_mode: str, inner_mode: str, exact_order: bool ) -> 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. + + Serial outer admission is strictly ordered, so the sequence is exactly + [1, 2, 1, 2]. Concurrent outer completion order depends on scheduling, + so only the multiset is contractual there. """ workflow = _nested_mode_workflow(outer_mode=outer_mode, inner_mode=inner_mode) @@ -528,9 +532,11 @@ def test_nested_foreach_preserves_inner_writes_in_all_modes( ) assert run.status == RunStatus.COMPLETED - assert sorted(run.state.get("seen") or [], key=repr) == sorted( - [1, 2, 1, 2], key=repr - ) + seen = run.state.get("seen") or [] + if exact_order: + assert seen == [1, 2, 1, 2] + else: + assert sorted(seen, key=repr) == sorted([1, 2, 1, 2], key=repr) def test_item_frame_owner_rejects_missing_parent_frame() -> None: @@ -595,6 +601,98 @@ def test_foreach_aware_patch_rejects_parent_cycle() -> None: commit_foreach_aware_patch(run, frame_a, StatePatch(changes={})) +def test_foreach_aware_patch_rejects_concurrent_self_cycle() -> None: + """A self-parented item with a concurrent owner must fail, not buffer.""" + from wf_core.run_state import LineageState + 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 = ExecutionFrame( + id="self", + kind="foreach_iteration", + node_id="work", + scope_id="root", + lineage_id="root", + ) + activation = load_or_begin_foreach_activation(frame, "each", mode="concurrent") + frame.parent_frame_id = "self" + frame.metadata.update( + { + "foreach_node_id": "each", + "activation_id": activation.id, + "loop_index": 0, + "loop_item": "a", + "loop_alias": "item", + } + ) + run = RunState( + workflow_name="concurrent_self_cycle", + status=RunStatus.RUNNING, + workflow_input={}, + state={}, + frames={"self": frame}, + lineages={"root": LineageState(id="root", scope_id="root")}, + ) + + with pytest.raises(WorkflowExecutionError, match="cycle"): + commit_foreach_aware_patch(run, frame, StatePatch(changes={})) + assert run.lineages["root"].writes == [] + + +def test_foreach_aware_patch_rejects_cycle_through_concurrent_boundary() -> None: + """A parent cycle spanning a concurrent boundary must fail, not buffer.""" + from wf_core.run_state import LineageState + 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", + scope_id="root", + lineage_id="root", + ) + frame_b = ExecutionFrame(id="frame-b", kind="foreach_iteration", node_id="work") + activation_on_b = load_or_begin_foreach_activation( + frame_b, "each", mode="concurrent" + ) + 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="mixed_mode_cycle", + status=RunStatus.RUNNING, + workflow_input={}, + state={}, + frames={"frame-a": frame_a, "frame-b": frame_b}, + lineages={"root": LineageState(id="root", scope_id="root")}, + ) + + with pytest.raises(WorkflowExecutionError, match="cycle"): + commit_foreach_aware_patch(run, frame_a, StatePatch(changes={})) + assert run.lineages["root"].writes == [] + + def test_reentering_foreach_uses_fresh_activation_and_item_frames() -> None: workflow = Workflow( name="foreach_reentry",