lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

audit: unify item write routing, fix serial interrupt loss, enforce result coherence

Browse Source
This commit is contained in:
lda
2026-09-04 11:20:15 +07:00 Verified
parent f155e6651a
commit 8a0737fe8b
9 changed files with 206 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/historical/superpowers/plans/2026-09-04-foreach-back-edges.md
+1 -1
View File
@@ -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
src/wf_core/runtime/foreach_state.py
+26 -6
View File
@@ -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,
src/wf_core/runtime/lineage.py
+39
View File
@@ -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)
src/wf_core/runtime/ops/foreach.py
+12 -7
View File
@@ -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 = [
src/wf_core/runtime/ops/interrupts.py
+4 -2
View File
@@ -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,
src/wf_core/runtime/ops/nodes.py
+5 -33
View File
@@ -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,
src/wf_core/runtime/subgraphs.py
+5 -26
View File
@@ -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,
tests/core/test_foreach_back_edges.py
+64
View File
@@ -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",
tests/core/test_foreach_barrier_state.py
+50
View File
@@ -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"])