From 7b2f718ad7aed39f4d8f5774efa2dd666c4ce867 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 12:48:01 +0700 Subject: [PATCH] audit: validate-first routing, read-only activation lookup, delta-log barrier patches --- .../2026-09-04-foreach-back-edge-design.md | 7 +- src/wf_core/runtime/foreach_state.py | 40 +++++- src/wf_core/runtime/ops/state.py | 32 ++++- tests/core/test_atomic_state_patches.py | 40 ++++++ tests/core/test_foreach_activations.py | 28 ++++ tests/core/test_foreach_back_edges.py | 124 ++++++++++++++++++ 6 files changed, 256 insertions(+), 15 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 ea1906a8..9a3e40d9 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 @@ -437,8 +437,11 @@ 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 +root, where it commits, or selects the first concurrent boundary as the +buffer target, where it buffers for that barrier to merge (the walk +continues past the selected boundary to validate the full ancestry, so +parent cycles fail closed even when they pass through a concurrent +owner; 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 diff --git a/src/wf_core/runtime/foreach_state.py b/src/wf_core/runtime/foreach_state.py index 89dd5c54..f45c71e8 100644 --- a/src/wf_core/runtime/foreach_state.py +++ b/src/wf_core/runtime/foreach_state.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Literal +from typing import Any, Literal, overload from wf_core.errors import WorkflowExecutionError from wf_core.run_state import ExecutionFrame, RunState @@ -303,9 +303,13 @@ class ForeachBarrierState: def _activation_entry( - frame: ExecutionFrame, table: dict[str, Any], foreach_node_id: str + frame: ExecutionFrame, + table: dict[str, Any] | None, + foreach_node_id: str, ) -> dict[str, Any] | None: """Return the mutable activation entry or fail fast on corrupt state.""" + if table is None: + return None entry = table.get(foreach_node_id) if entry is None: return None @@ -368,7 +372,7 @@ def save_foreach_activation( frame: ExecutionFrame, activation: ForeachActivationState ) -> None: """Persist barrier progress for the named active activation.""" - table = _activation_table(frame) + table = _activation_table(frame, create=False) entry = _activation_entry(frame, table, activation.foreach_node_id) if entry is None: raise WorkflowExecutionError( @@ -391,7 +395,7 @@ def close_foreach_activation( The barrier is removed so a later visit starts fresh; the sequence keeps increasing so child and lineage ids cannot collide across visits. """ - table = _activation_table(frame) + table = _activation_table(frame, create=False) entry = _activation_entry(frame, table, activation.foreach_node_id) if entry is None: raise WorkflowExecutionError( @@ -413,8 +417,11 @@ def load_foreach_activation( A child result naming a closed or different activation must fail closed in the caller rather than buffering into the wrong barrier. + + This is a read-only lookup: a missing table or entry raises without + mutating frame metadata. """ - table = _activation_table(frame) + table = _activation_table(frame, create=False) entry = _activation_entry(frame, table, foreach_node_id) if entry is None: raise WorkflowExecutionError( @@ -469,9 +476,30 @@ def item_frame_owner(frame: ExecutionFrame) -> ForeachItemOwner | None: ) -def _activation_table(frame: ExecutionFrame) -> dict[str, Any]: +@overload +def _activation_table( + frame: ExecutionFrame, *, create: Literal[True] = True +) -> dict[str, Any]: ... + + +@overload +def _activation_table( + frame: ExecutionFrame, *, create: Literal[False] +) -> dict[str, Any] | None: ... + + +def _activation_table( + frame: ExecutionFrame, *, create: bool = True +) -> dict[str, Any] | None: + """Return the activation table, optionally creating it. + + Read-only lookups pass ``create=False`` so a failed lookup leaves + frame metadata untouched. Only ``load_or_begin`` creates the table. + """ raw = frame.metadata.get(_ACTIVATION_METADATA_KEY) if raw is None: + if not create: + return None table: dict[str, Any] = {} frame.metadata[_ACTIVATION_METADATA_KEY] = table return table diff --git a/src/wf_core/runtime/ops/state.py b/src/wf_core/runtime/ops/state.py index 8faff8e6..2a5bc187 100644 --- a/src/wf_core/runtime/ops/state.py +++ b/src/wf_core/runtime/ops/state.py @@ -249,6 +249,15 @@ def build_barrier_patch( committed aggregate values. A barrier trace is the single visible state commit for all buffered item patches, so showing raw per-item incoming values would hide what actually landed in `RunState.state`. + + The emitted `writes` log keeps every constituent item write in order + instead of one merged write per path. A combined patch buffered in a + lineage can itself be re-merged by an outer barrier, and replaying merged + cumulative values would duplicate whatever was already committed when the + constituents were built. Replaying the original per-item deltas stays + correct at any nesting depth. Each kept write still carries the merged + aggregate as its `visible_value`, so overlay reads and `visible_values` + keep showing the final value. """ state_fields = workflow.state_schema.field_index() validate_barrier_writes(item_patches, state_fields, reducers=reducers) @@ -269,22 +278,31 @@ def build_barrier_patch( safe_set_nested_value(staged_state, key_path, merged_value) prepared_patch[destination_path] = (key_path, merged_value) committed_changes[str(destination_path)] = merged_value + merged_visible = { + destination_path: merged_value + for destination_path, (_key_path, merged_value) in prepared_patch.items() + } writes = [ StateWrite( - path=destination_path, - incoming_value=merged_value, - visible_value=merged_value, - reducer=reducer_for_state_path(destination_path, state_fields), + path=write.path, + incoming_value=write.incoming_value, + visible_value=merged_visible[write.path], + reducer=write.reducer, ) - for destination_path, (_key_path, merged_value) in prepared_patch.items() + for item_patch in item_patches + for write in item_patch.writes ] validate_staged_state_patch(staged_state, prepared_patch, state_fields) - return StatePatch( - changes=committed_changes, + combined = StatePatch( writes=writes, _prepared_writes=prepared_patch, _staged_state=staged_state, ) + # The trace-facing view reports the aggregate, while the replay log above + # intentionally carries per-item deltas (see docstring). Assign it after + # construction: passing both to the constructor requires them to agree. + combined.changes = committed_changes + return combined def validate_barrier_writes( diff --git a/tests/core/test_atomic_state_patches.py b/tests/core/test_atomic_state_patches.py index f9fa5f23..d8c10244 100644 --- a/tests/core/test_atomic_state_patches.py +++ b/tests/core/test_atomic_state_patches.py @@ -284,6 +284,46 @@ def test_barrier_replays_incoming_values_not_lineage_visible_values() -> None: assert patch.visible_values["state.number"] == 6 +def test_barrier_combined_patch_remerges_without_duplicating_prefix() -> None: + """A combined patch re-merged by an outer barrier must not duplicate. + + The second barrier is computed after the first aggregate was committed, + so its constituents were built against that prefix. Re-merging must + replay the original per-item deltas, not the cumulative aggregates. + """ + workflow = _workflow( + fields={ + "seen": StateField( + type="array", + reducer=ReducerRef(name="wf.std.append"), + ) + } + ) + + first = build_barrier_patch( + workflow, + [ + StatePatch(changes={"state.seen": "a"}), + StatePatch(changes={"state.seen": "b"}), + ], + {}, + ) + second = build_barrier_patch( + workflow, + [ + StatePatch(changes={"state.seen": "c"}), + StatePatch(changes={"state.seen": "d"}), + ], + {"seen": ["a", "b"]}, + ) + + assert second.changes["state.seen"] == ["a", "b", "c", "d"] + remerged = build_barrier_patch(workflow, [first, second], {}) + + assert remerged.changes["state.seen"] == ["a", "b", "c", "d"] + assert remerged.visible_values["state.seen"] == ["a", "b", "c", "d"] + + def test_build_and_commit_patch_matches_apply_output_bindings() -> None: workflow = _workflow(fields={"person.name": StateField(type="string")}) state_from_apply = {"person": {"name": "old"}} diff --git a/tests/core/test_foreach_activations.py b/tests/core/test_foreach_activations.py index cb94d0f6..0bdea28a 100644 --- a/tests/core/test_foreach_activations.py +++ b/tests/core/test_foreach_activations.py @@ -5,8 +5,11 @@ import pytest from wf_core.errors import WorkflowExecutionError from wf_core.run_state import ExecutionFrame from wf_core.runtime.foreach_state import ( + ForeachActivationState, + ForeachBarrierState, close_foreach_activation, item_frame_owner, + load_foreach_activation, load_or_begin_foreach_activation, save_foreach_activation, ) @@ -100,3 +103,28 @@ def test_item_metadata_requires_activation_identity() -> None: ForeachIterationMetadata.from_frame(frame) with pytest.raises(WorkflowExecutionError, match="activation"): item_frame_owner(frame) + + +def test_failed_activation_lookup_leaves_metadata_untouched() -> None: + """Read-only lookups must not create the activation table on failure.""" + frame = _frame() + stale = ForeachActivationState( + id="root:each#0", + foreach_node_id="each", + barrier=ForeachBarrierState(mode="serial"), + ) + + with pytest.raises(WorkflowExecutionError, match="activation"): + load_foreach_activation(frame, "each", "root:each#0") + with pytest.raises(WorkflowExecutionError, match="activation"): + save_foreach_activation(frame, stale) + with pytest.raises(WorkflowExecutionError, match="activation"): + close_foreach_activation(frame, stale) + + assert frame.metadata == {} + + # The write path still creates the table exactly once. + activation = load_or_begin_foreach_activation(frame, "each", mode="serial") + assert frame.metadata["foreach_activations"]["each"]["active"]["id"] == ( + activation.id + ) diff --git a/tests/core/test_foreach_back_edges.py b/tests/core/test_foreach_back_edges.py index e3f75d5e..bdc368f5 100644 --- a/tests/core/test_foreach_back_edges.py +++ b/tests/core/test_foreach_back_edges.py @@ -539,6 +539,130 @@ def test_nested_foreach_preserves_inner_writes_in_all_modes( assert sorted(seen, key=repr) == sorted([1, 2, 1, 2], key=repr) +def _three_level_workflow( + *, outer_mode: str, middle_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="three_level_nested_foreach", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "items": StateField(type="array"), + "mid_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( + "middle", + over="state.mid_items", + alias="mid_item", + mode=middle_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": "middle"}), + Edge.model_validate({"from": "middle", "outcome": "loop", "to": "inner"}), + Edge.model_validate({"from": "inner", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "inner"}), + Edge.model_validate({"from": "inner", "outcome": "done", "to": "middle"}), + Edge.model_validate({"from": "middle", "outcome": "done", "to": "outer"}), + Edge.model_validate({"from": "outer", "outcome": "done", "to": END}), + ], + ) + + +@pytest.mark.parametrize( + ("outer_mode", "middle_mode", "inner_mode"), + [ + ("serial", "serial", "serial"), + ("serial", "serial", "concurrent"), + ("serial", "concurrent", "serial"), + ("serial", "concurrent", "concurrent"), + ("concurrent", "serial", "serial"), + ("concurrent", "serial", "concurrent"), + ("concurrent", "concurrent", "serial"), + ("concurrent", "concurrent", "concurrent"), + ], +) +def test_three_level_nested_foreach_preserves_writes( + outer_mode: str, middle_mode: str, inner_mode: str +) -> None: + """Write routing holds through three nesting levels in every mode mix. + + Barriers merge items in index order, so each middle visit yields exactly + [1, 2]; only the outer completion order varies. A serial outer admits + in order, making [1, 2, 1, 2] exact, while a concurrent outer leaves + only the multiset contractual. + """ + workflow = _three_level_workflow( + outer_mode=outer_mode, middle_mode=middle_mode, inner_mode=inner_mode + ) + + run = execute_workflow( + workflow, + {"items": ["a", "b"], "mid_items": ["m"], "inner_items": [1, 2]}, + { + "work": lambda payload, _ctx: { + "outcome": "ok", + "output": {"seen": payload["value"]}, + } + }, + ) + + assert run.status == RunStatus.COMPLETED + seen = run.state.get("seen") or [] + if outer_mode == "serial": + 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: """A foreach_iteration frame without a parent is malformed, not ordinary.""" frame = ExecutionFrame(