From 8a0737fe8b456284b79820b8e200b57ec23d1991 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 11:20:15 +0700 Subject: [PATCH] audit: unify item write routing, fix serial interrupt loss, enforce result coherence --- .../plans/2026-09-04-foreach-back-edges.md | 2 +- src/wf_core/runtime/foreach_state.py | 32 ++++++++-- src/wf_core/runtime/lineage.py | 39 +++++++++++ src/wf_core/runtime/ops/foreach.py | 19 ++++-- src/wf_core/runtime/ops/interrupts.py | 6 +- src/wf_core/runtime/ops/nodes.py | 38 ++--------- src/wf_core/runtime/subgraphs.py | 31 ++------- tests/core/test_foreach_back_edges.py | 64 +++++++++++++++++++ tests/core/test_foreach_barrier_state.py | 50 +++++++++++++++ 9 files changed, 206 insertions(+), 75 deletions(-) diff --git a/docs/historical/superpowers/plans/2026-09-04-foreach-back-edges.md b/docs/historical/superpowers/plans/2026-09-04-foreach-back-edges.md index 44831ced..a019c421 100644 --- a/docs/historical/superpowers/plans/2026-09-04-foreach-back-edges.md +++ b/docs/historical/superpowers/plans/2026-09-04-foreach-back-edges.md @@ -19,7 +19,7 @@ back-edge as item completion. pytest, pytest-asyncio, Ruff, basedpyright, markdownlint-cli2. **Spec:** -[`docs/superpowers/specs/2026-09-04-foreach-back-edge-design.md`](../../superpowers/specs/2026-09-04-foreach-back-edge-design.md) +[`docs/superpowers/specs/2026-09-04-foreach-back-edge-design.md`](../../../superpowers/specs/2026-09-04-foreach-back-edge-design.md) ## Global Constraints diff --git a/src/wf_core/runtime/foreach_state.py b/src/wf_core/runtime/foreach_state.py index 41fc86d6..21fc75d8 100644 --- a/src/wf_core/runtime/foreach_state.py +++ b/src/wf_core/runtime/foreach_state.py @@ -112,17 +112,32 @@ class PendingItemResult: f"malformed pending foreach result missing {exc.args[0]!r}" ) from exc lineage_id = raw.get("lineage_id") + raw_error = raw.get("error") if not isinstance(index, int) or index < 0: raise WorkflowExecutionError("malformed pending foreach result index") if not isinstance(frame_id, str): raise WorkflowExecutionError("malformed pending foreach result frame id") - if status == "succeeded" and not isinstance(lineage_id, str): - raise WorkflowExecutionError("malformed pending foreach result lineage id") - if lineage_id is not None and not isinstance(lineage_id, str): - raise WorkflowExecutionError("malformed pending foreach result lineage id") if status not in {"succeeded", "failed"}: raise WorkflowExecutionError("malformed pending foreach result status") - raw_error = raw.get("error") + if status == "succeeded": + if not isinstance(lineage_id, str): + raise WorkflowExecutionError( + "malformed pending foreach result lineage id" + ) + if raw_error is not None: + raise WorkflowExecutionError( + "malformed pending foreach result: succeeded result must not " + "carry an error" + ) + else: + if raw_error is None: + raise WorkflowExecutionError( + "malformed pending foreach result: failed result requires an error" + ) + if lineage_id is not None and not isinstance(lineage_id, str): + raise WorkflowExecutionError( + "malformed pending foreach result lineage id" + ) return cls( index=index, frame_id=frame_id, @@ -178,7 +193,12 @@ class ForeachBarrierState: raise WorkflowExecutionError( "malformed foreach barrier pending result index" ) from exc - parsed_results[index] = PendingItemResult.from_metadata(raw_result) + parsed = PendingItemResult.from_metadata(raw_result) + if parsed.index != index: + raise WorkflowExecutionError( + "malformed foreach barrier pending result index mismatch" + ) + parsed_results[index] = parsed return cls( next_index=next_index, mode=mode, diff --git a/src/wf_core/runtime/lineage.py b/src/wf_core/runtime/lineage.py index 77c762aa..e9ade5b1 100644 --- a/src/wf_core/runtime/lineage.py +++ b/src/wf_core/runtime/lineage.py @@ -89,6 +89,45 @@ def commit_patch_for_frame( return {} +def commit_foreach_aware_patch( + run: RunState, frame: ExecutionFrame, patch: StatePatch +) -> 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. + """ + 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}" + ) + 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) + + def scope_state_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]: """Return the committed state root for the frame's runtime scope.""" scope = run.scopes.get(frame.scope_id) diff --git a/src/wf_core/runtime/ops/foreach.py b/src/wf_core/runtime/ops/foreach.py index 87739b64..e94961a5 100644 --- a/src/wf_core/runtime/ops/foreach.py +++ b/src/wf_core/runtime/ops/foreach.py @@ -341,13 +341,18 @@ def _finish_concurrent_foreach( reducers: Mapping[str, ReducerDefinition] | None = None, ) -> RunState: barrier = activation.barrier - error_records = [ - result.error.to_metadata() - for result in sorted( - barrier.pending_results.values(), key=lambda item: item.index - ) - if result.status == "failed" and result.error is not None - ] + # Coherence is enforced at load, but re-check here: a failed result + # without an error must never silent-commit as `done`. + error_records = [] + for result in sorted(barrier.pending_results.values(), key=lambda item: item.index): + if result.status != "failed": + continue + if result.error is None: + raise WorkflowExecutionError( + f"foreach item result for index {result.index!r} is failed " + "but carries no error" + ) + error_records.append(result.error.to_metadata()) outcome = "completed_with_errors" if error_records else "done" next_node_id = index.next_node_id(frame.node_id, outcome) success_patches = [ diff --git a/src/wf_core/runtime/ops/interrupts.py b/src/wf_core/runtime/ops/interrupts.py index 3fd11c8b..d56c1e9c 100644 --- a/src/wf_core/runtime/ops/interrupts.py +++ b/src/wf_core/runtime/ops/interrupts.py @@ -14,7 +14,7 @@ from wf_core.run_state import ( StepExecutionResult, ) from wf_core.runtime.input_bindings import resolve_step_input_bindings -from wf_core.runtime.lineage import commit_patch_for_frame +from wf_core.runtime.lineage import commit_foreach_aware_patch from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace from wf_core.runtime.ops.index import WorkflowIndex from wf_core.runtime.ops.merges import ReducerDefinition @@ -115,7 +115,9 @@ def resume_interrupt( reducers=reducers, missing_field_message="interrupt resume payload is missing required field {field}", ) - state_changes = commit_patch_for_frame(run, frame, patch) + # Foreach-aware routing: a serial item resume commits through the parent + # scope, a concurrent one buffers in the item lineage for barrier merge. + state_changes = commit_foreach_aware_patch(run, frame, patch) next_node_id = index.next_node_id(frame.node_id, resume_outcome) append_step_result_trace( run, diff --git a/src/wf_core/runtime/ops/nodes.py b/src/wf_core/runtime/ops/nodes.py index 6c73976b..199b4df8 100644 --- a/src/wf_core/runtime/ops/nodes.py +++ b/src/wf_core/runtime/ops/nodes.py @@ -15,14 +15,9 @@ from wf_core.run_state import ( RuntimeContext, StepExecutionResult, ) -from wf_core.runtime.foreach_state import ( - item_frame_owner, - require_foreach_activation, -) from wf_core.runtime.input_bindings import resolve_step_input_bindings from wf_core.runtime.lineage import ( - append_lineage_writes, - commit_patch_for_frame, + commit_foreach_aware_patch, scope_input_for_frame, ) from wf_core.runtime.ops.frames import frame_context_values @@ -115,33 +110,10 @@ def _finalize_node_execution( state_view, reducers=reducers, ) - owner = item_frame_owner(frame) - if owner is None: - state_changes = commit_patch_for_frame(run, frame, patch) - else: - 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}" - ) - # Fail closed when the child names a closed or superseded activation: - # its writes must not land in a later visit's barrier. - activation = require_foreach_activation( - parent_frame, owner.foreach_node_id, owner.activation_id - ) - if activation.barrier.mode == "concurrent": - # Concurrent writes stay buffered in the child lineage; the owner - # back-edge registers the completed item with the barrier. - append_lineage_writes( - run, - scope_id=frame.scope_id, - lineage_id=frame.lineage_id, - writes=patch.writes, - ) - state_changes = {} - else: - state_changes = commit_patch_for_frame(run, parent_frame, patch) + # Foreach-aware routing (root, serial parent, concurrent lineage) is + # owned by the shared helper so every operation commits the same way. + # Closed or superseded activations fail closed inside. + state_changes = commit_foreach_aware_patch(run, frame, patch) return StepExecutionResult( outcome=result.outcome, resolved_input=resolved_input, diff --git a/src/wf_core/runtime/subgraphs.py b/src/wf_core/runtime/subgraphs.py index 89870993..49daabb1 100644 --- a/src/wf_core/runtime/subgraphs.py +++ b/src/wf_core/runtime/subgraphs.py @@ -17,7 +17,7 @@ from wf_core.run_state import ( StepExecutionResult, ) from wf_core.runtime.input_bindings import resolve_step_input_bindings -from wf_core.runtime.lineage import commit_patch_for_frame +from wf_core.runtime.lineage import commit_foreach_aware_patch from wf_core.runtime.ops.frames import frame_context_values from wf_core.runtime.ops.merges import ReducerDefinition from wf_core.runtime.ops.overlays import state_view_for_frame @@ -236,31 +236,10 @@ def _finish_subgraph( reducers=reducers, missing_field_message="subgraph output did not include required field {field}", ) - # Match node execution: serial item writes commit through the parent - # scope so top-level serial subgraphs land in root state; concurrent - # item writes stay buffered in the item lineage for barrier merge. - from wf_core.runtime.foreach_state import ( - item_frame_owner, - require_foreach_activation, - ) - - commit_frame = frame - owner = item_frame_owner(frame) - if owner is not None: - parent_frame = run.frames.get(owner.parent_frame_id) - if parent_frame is None: - raise WorkflowExecutionError( - "subgraph state references missing parent frame " - f"{owner.parent_frame_id!r} for child frame {frame.id!r}" - ) - # Fail closed when the child names a closed or superseded - # activation: its output must not land in a later visit's state. - foreach_activation = require_foreach_activation( - parent_frame, owner.foreach_node_id, owner.activation_id - ) - if foreach_activation.barrier.mode == "serial": - commit_frame = parent_frame - state_changes = commit_patch_for_frame(run, commit_frame, patch) + # Foreach-aware routing (root, serial parent, concurrent lineage) is + # owned by the shared helper so subgraph output commits exactly like + # node output. Closed or superseded activations fail closed inside. + state_changes = commit_foreach_aware_patch(run, frame, patch) return StepExecutionResult( outcome=child_outcome, resolved_input=activation.child_input, diff --git a/tests/core/test_foreach_back_edges.py b/tests/core/test_foreach_back_edges.py index 800c53d1..fb6ba2ce 100644 --- a/tests/core/test_foreach_back_edges.py +++ b/tests/core/test_foreach_back_edges.py @@ -9,6 +9,7 @@ from wf_core import ( ConditionNode, Edge, ForeachNode, + InterruptNode, NodeDef, NodeUse, ReducerRef, @@ -19,6 +20,8 @@ from wf_core import ( Workflow, WorkflowExecutionError, execute_workflow, + execute_workflow_async, + resume_workflow_async, ) from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, RunStatus from wf_core.runtime.foreach_state import item_frame_owner @@ -742,6 +745,67 @@ def test_concurrent_subgraph_item_returns_through_owner() -> None: assert sorted(run.state["seen"]) == ["a", "b"] +async def test_serial_interrupt_resume_commits_answer_to_parent_state() -> None: + """A serial item resume must land in parent state, not the child lineage.""" + foreach = ForeachNode.model_validate( + { + "id": "each", + "type": "foreach", + "over": "state.items", + "as": "item", + "mode": "serial", + } + ) + workflow = Workflow( + name="foreach_serial_interrupt", + input_schema=SchemaRef(type="object", properties={"items": {"type": "array"}}), + state_schema=StateSchema.from_field_map( + { + "items": StateField(type="array"), + "answers": StateField( + type="array", reducer=ReducerRef(name="wf.std.append") + ), + } + ), + output_schema=SchemaRef( + type="object", properties={"answers": {"type": "array"}} + ), + node_defs=[], + start="each", + nodes=[ + foreach, + InterruptNode.model_validate( + { + "id": "ask", + "type": "interrupt", + "kind": "approval", + "request": [{"target": "item", "path": "context.item"}], + "resume": [{"source": "answer", "target": "state.answers"}], + } + ), + ], + edges=[ + Edge.model_validate({"from": "each", "outcome": "loop", "to": "ask"}), + Edge.model_validate({"from": "ask", "outcome": "submitted", "to": "each"}), + Edge.model_validate({"from": "each", "outcome": "done", "to": END}), + ], + ) + + run = await execute_workflow_async(workflow, {"items": ["a", "b"]}, {}) + assert run.status == RunStatus.INTERRUPTED + + resumed = await resume_workflow_async( + workflow, run, {}, resume_payload={"answer": "a"} + ) + assert resumed.status == RunStatus.INTERRUPTED + + finished = await resume_workflow_async( + workflow, resumed, {}, resume_payload={"answer": "b"} + ) + assert finished.status == RunStatus.COMPLETED + assert finished.state["answers"] == ["a", "b"] + + def test_nonlocal_runtime_return_fails_closed_when_validation_is_bypassed() -> None: run = RunState( workflow_name="nonlocal", diff --git a/tests/core/test_foreach_barrier_state.py b/tests/core/test_foreach_barrier_state.py index 1ac3a439..6df3e5ca 100644 --- a/tests/core/test_foreach_barrier_state.py +++ b/tests/core/test_foreach_barrier_state.py @@ -1,5 +1,7 @@ from __future__ import annotations +from copy import deepcopy + import pytest from wf_core.errors import WorkflowExecutionError @@ -290,3 +292,51 @@ def test_pending_item_result_requires_lineage_for_success() -> None: "lineage_id": None, } ) + + +def test_pending_item_result_rejects_error_on_success() -> None: + with pytest.raises(WorkflowExecutionError, match="must not carry an error"): + PendingItemResult.from_metadata( + { + "index": 0, + "frame_id": "child", + "status": "succeeded", + "lineage_id": "root:each#0[0]", + "error": { + "index": 0, + "frame_id": "child", + "node_id": "work", + "error_type": "ValueError", + "message": "bad", + }, + } + ) + + +def test_pending_item_result_requires_error_for_failure() -> None: + with pytest.raises(WorkflowExecutionError, match="requires an error"): + PendingItemResult.from_metadata( + { + "index": 0, + "frame_id": "child", + "status": "failed", + "lineage_id": None, + "error": None, + } + ) + + +def test_pending_item_result_rejects_index_key_mismatch() -> None: + frame = ExecutionFrame(id="root", kind="root", node_id="each") + activation = load_or_begin_foreach_activation(frame, "each", mode="concurrent") + activation.barrier.add_success_patch( + index=0, frame_id="child-0", lineage_id="root:each#0[0]" + ) + save_foreach_activation(frame, activation) + raw = deepcopy(frame.metadata["foreach_activations"]) + raw["each"]["active"]["barrier"]["pending_results"] = { + "7": raw["each"]["active"]["barrier"]["pending_results"]["0"] + } + + with pytest.raises(WorkflowExecutionError, match="index mismatch"): + ForeachBarrierState.from_metadata(raw["each"]["active"]["barrier"])