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

audit: fix concurrent subgraph loss, fail-closed ownership, drop barrier compat

Browse Source
This commit is contained in:
lda
2026-09-04 10:58:13 +07:00 Verified
parent 79ce0d3eff
commit f155e6651a
11 changed files with 424 additions and 478 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_core/runtime/foreach_state.py
+34 -125
View File
@@ -4,13 +4,9 @@ from dataclasses import dataclass, field
from typing import Any, Literal
from wf_core.errors import WorkflowExecutionError
from wf_core.models.reducers import ReducerRef
from wf_core.paths import StatePath
from wf_core.run_state import ExecutionFrame, StateWrite
from wf_core.runtime.ops.state import StatePatch
from wf_core.run_state import ExecutionFrame, RunState
from wf_core.runtime.scheduler import ForeachIterationMetadata
_BARRIER_METADATA_KEY = "foreach_barriers"
_ACTIVATION_METADATA_KEY = "foreach_activations"
@@ -93,16 +89,14 @@ class ItemErrorRecord:
class PendingItemResult:
"""Buffered item result waiting for a future foreach barrier commit.
New concurrent foreach execution stores item writes in `RunState.lineages`
and records `lineage_id` here. `patch` remains for old serialized barrier
metadata and direct unit tests that still construct pending patches.
Concurrent item writes live in `RunState.lineages`; the barrier keeps
only the lineage identity per item index.
"""
index: int
frame_id: str
status: Literal["succeeded", "failed"]
lineage_id: str | None = None
patch: StatePatch = field(default_factory=StatePatch)
error: ItemErrorRecord | None = None
@classmethod
@@ -117,34 +111,23 @@ class PendingItemResult:
raise WorkflowExecutionError(
f"malformed pending foreach result missing {exc.args[0]!r}"
) from exc
patch_changes = raw.get("patch_changes", {})
patch_writes = raw.get("patch_writes")
lineage_id = raw.get("lineage_id")
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")
if not isinstance(patch_changes, dict):
raise WorkflowExecutionError("malformed pending foreach result patch")
if patch_writes is not None and not isinstance(patch_writes, list):
raise WorkflowExecutionError("malformed pending foreach result writes")
raw_error = raw.get("error")
return cls(
index=index,
frame_id=frame_id,
status=status,
lineage_id=lineage_id,
patch=(
StatePatch(
writes=[_state_write_from_metadata(item) for item in patch_writes]
)
if patch_writes is not None
else StatePatch(changes=patch_changes)
),
error=(
ItemErrorRecord.from_metadata(raw_error)
if raw_error is not None
@@ -158,10 +141,6 @@ class PendingItemResult:
"frame_id": self.frame_id,
"status": self.status,
"lineage_id": self.lineage_id,
"patch_changes": dict(self.patch.changes),
"patch_writes": [
_state_write_to_metadata(write) for write in self.patch.writes
],
"error": self.error.to_metadata() if self.error is not None else None,
}
@@ -176,33 +155,6 @@ class ForeachBarrierState:
outstanding_frame_ids: tuple[str, ...] = ()
pending_results: dict[int, PendingItemResult] = field(default_factory=dict)
@classmethod
def from_frame(
cls,
frame: ExecutionFrame,
foreach_node_id: str,
) -> ForeachBarrierState | None:
"""Load one foreach barrier state from frame metadata.
Missing metadata means the foreach has not started on this frame yet.
Malformed metadata means runtime state is corrupt and should fail fast.
"""
all_barriers = frame.metadata.get(_BARRIER_METADATA_KEY)
if all_barriers is None:
return None
if not isinstance(all_barriers, dict):
raise WorkflowExecutionError(
f"malformed foreach barrier table for frame {frame.id!r}"
)
raw = all_barriers.get(foreach_node_id)
if raw is None:
return None
if not isinstance(raw, dict):
raise WorkflowExecutionError(
f"malformed foreach barrier state for frame {frame.id!r}"
)
return cls.from_metadata(raw)
@classmethod
def from_metadata(cls, raw: object) -> ForeachBarrierState:
if not isinstance(raw, dict):
@@ -235,20 +187,6 @@ class ForeachBarrierState:
pending_results=parsed_results,
)
def save_to_frame(self, frame: ExecutionFrame, foreach_node_id: str) -> None:
"""Store this barrier state in frame metadata under its foreach node id."""
existing = frame.metadata.get(_BARRIER_METADATA_KEY)
if existing is None:
frame.metadata[_BARRIER_METADATA_KEY] = {
foreach_node_id: self.to_metadata()
}
return
if not isinstance(existing, dict):
raise WorkflowExecutionError(
f"malformed foreach barrier table for frame {frame.id!r}"
)
existing[foreach_node_id] = self.to_metadata()
def to_metadata(self) -> dict[str, Any]:
return {
"next_index": self.next_index,
@@ -291,15 +229,13 @@ class ForeachBarrierState:
*,
index: int,
frame_id: str,
patch: StatePatch,
lineage_id: str | None = None,
lineage_id: str,
) -> None:
"""Buffer or extend successful item patches by item index.
"""Record one completed concurrent item by lineage identity.
New runtime paths pass an empty patch and use `lineage_id`; legacy
callers may still accumulate patches here and replay them at the
barrier. Do not merge `_prepared_writes`: the barrier replays public
write records against one staged parent state.
Registration is idempotent for the same frame and lineage so the
owner back-edge can own it regardless of which operation ran last.
Any conflicting identity fails closed.
"""
existing = self.pending_results.get(index)
if existing is None:
@@ -308,7 +244,6 @@ class ForeachBarrierState:
frame_id=frame_id,
status="succeeded",
lineage_id=lineage_id,
patch=patch,
)
return
if existing.frame_id != frame_id:
@@ -316,14 +251,11 @@ class ForeachBarrierState:
f"foreach item result for index {index!r} belongs to frame "
f"{existing.frame_id!r}, got {frame_id!r}"
)
if lineage_id is not None and existing.lineage_id not in {None, lineage_id}:
if existing.lineage_id != lineage_id:
raise WorkflowExecutionError(
f"foreach item result for index {index!r} belongs to lineage "
f"{existing.lineage_id!r}, got {lineage_id!r}"
)
if existing.lineage_id is None:
existing.lineage_id = lineage_id
existing.patch.extend(patch)
def add_failure(self, *, error: ItemErrorRecord) -> None:
"""Buffer one handled item failure for the foreach barrier.
@@ -547,53 +479,30 @@ def _string_tuple(raw: object) -> tuple[str, ...]:
raise WorkflowExecutionError("malformed foreach barrier frame id list")
def _state_write_from_metadata(raw: object) -> StateWrite:
"""Parse one persisted item-lineage write record.
def register_foreach_item_success(
run: RunState, frame: ExecutionFrame, owner: ForeachItemOwner
) -> None:
"""Record one completed concurrent item at its owner back-edge.
Barrier metadata must keep reducer-visible values across interrupt/resume;
reconstructing from `patch_changes` would downgrade reducer writes to
replace-style incoming values.
Registration keys off the returning frame, so it works regardless of
which operation ran last in the item (node, subgraph, or nested
control). Serial items commit through the parent at operation time and
need no barrier entry. A closed or superseded activation fails closed.
"""
if not isinstance(raw, dict):
raise WorkflowExecutionError("malformed pending foreach write")
try:
path = raw["path"]
incoming_value = raw["incoming_value"]
visible_value = raw["visible_value"]
reducer = raw["reducer"]
except KeyError as exc:
parent_frame = run.frames.get(owner.parent_frame_id)
if parent_frame is None:
raise WorkflowExecutionError(
f"malformed pending foreach write missing {exc.args[0]!r}"
) from exc
try:
return StateWrite(
path=_state_path_from_metadata(path),
incoming_value=incoming_value,
visible_value=visible_value,
reducer=ReducerRef.model_validate(reducer),
"foreach lineage state references missing parent frame "
f"{owner.parent_frame_id!r} for child frame {frame.id!r}"
)
except WorkflowExecutionError:
raise
except (TypeError, ValueError) as exc:
raise WorkflowExecutionError(f"malformed pending foreach write: {exc}") from exc
def _state_write_to_metadata(write: StateWrite) -> dict[str, Any]:
"""Serialize one item-lineage write without relying on dotted display paths."""
return {
"path": {"root": "state", "parts": list(write.path.parts)},
"incoming_value": write.incoming_value,
"visible_value": write.visible_value,
"reducer": write.reducer.model_dump(mode="json"),
}
def _state_path_from_metadata(raw: object) -> StatePath:
if isinstance(raw, str):
return StatePath.parse(raw)
if not isinstance(raw, dict) or raw.get("root") != "state":
raise WorkflowExecutionError("malformed pending foreach write path")
parts = raw.get("parts")
if not isinstance(parts, list) or not all(isinstance(part, str) for part in parts):
raise WorkflowExecutionError("malformed pending foreach write path")
return StatePath(tuple(parts))
activation = require_foreach_activation(
parent_frame, owner.foreach_node_id, owner.activation_id
)
if activation.barrier.mode != "concurrent":
return
activation.barrier.add_success_patch(
index=owner.item_index,
frame_id=frame.id,
lineage_id=frame.lineage_id,
)
save_foreach_activation(parent_frame, activation)
src/wf_core/runtime/lineage.py
+1 -26
View File
@@ -7,7 +7,6 @@ from typing import Any
from wf_core.errors import WorkflowExecutionError
from wf_core.run_state import ExecutionFrame, LineageState, RunState, StateWrite
from wf_core.runtime.foreach_state import item_frame_owner, load_foreach_activation
from wf_core.runtime.ops.state import (
StatePatch,
commit_state_patch,
@@ -57,31 +56,7 @@ def lineage_writes_for_frame(
run, scope_id=frame.scope_id, lineage_id=frame.lineage_id
)
)
# Compatibility fallback: concurrent foreach used barrier-local patches
# before `RunState.lineages` became the primary write store. Keep reading
# those patches so old serialized runs and direct barrier tests still work.
# Barrier lookup includes the activation so a stale visit cannot read a
# later activation's buffered writes.
owner = item_frame_owner(frame)
if owner is None:
return ()
parent_frame = run.frames.get(owner.parent_frame_id)
if parent_frame is None:
raise WorkflowExecutionError(
"foreach lineage compatibility state references missing parent frame "
f"{owner.parent_frame_id!r} for child frame {frame.id!r}"
)
activation = load_foreach_activation(
parent_frame, owner.foreach_node_id, owner.activation_id
)
if activation is None or activation.barrier.mode != "concurrent":
return ()
pending = activation.barrier.pending_results.get(owner.item_index)
if pending is None:
return ()
return pending.patch.writes
return ()
def is_scope_root_lineage_frame(run: RunState, frame: ExecutionFrame) -> bool:
src/wf_core/runtime/ops/flow.py
+8 -1
View File
@@ -80,7 +80,10 @@ def advance_frame(
# Foreach back-edge return is an ownership check, not generic cycle
# detection. Only the frame's immediate recorded owner completes the item;
# a root frame targeting the same foreach enters it normally.
from wf_core.runtime.foreach_state import item_frame_owner
from wf_core.runtime.foreach_state import (
item_frame_owner,
register_foreach_item_success,
)
owner = item_frame_owner(frame)
if owner is not None:
@@ -91,6 +94,10 @@ def advance_frame(
)
if next_node_id == owner.foreach_node_id:
source_node_id = frame.node_id
# Register the completed item with its barrier before completing
# the child, so every final operation (node, subgraph, nested
# control) counts. Closed or superseded activations fail closed.
register_foreach_item_success(run, frame, owner)
frame.prior_outcome = outcome
frame.activated_incoming_edge = source_node_id
frame.node_id = owner.foreach_node_id
src/wf_core/runtime/ops/foreach.py
+81 -72
View File
@@ -96,40 +96,16 @@ def _step_foreach_serial(
loop_start = index.next_node_id(frame.node_id, "loop")
item = iterable[loop_index]
barrier.next_index = loop_index + 1
loop_start, child_id = _admit_item_frame(
run=run,
frame=frame,
step=step,
index=index,
activation=activation,
loop_index=loop_index,
item=item,
)
save_foreach_activation(frame, activation)
child_id = _child_frame_id(activation, loop_index)
child_lineage_id = _child_lineage_id(activation, loop_index)
# Serial items still own a lineage so nested subgraph/boundary commits have
# a parent lineage to buffer into; top-level serial writes commit through
# the parent scope root.
add_lineage(
run,
scope_id=frame.scope_id,
lineage_id=child_lineage_id,
parent_id=frame.lineage_id,
)
add_frame(
run,
ExecutionFrame(
id=child_id,
kind="foreach_iteration",
node_id=loop_start,
status=FrameStatus.PENDING,
parent_frame_id=frame.id,
scope_id=frame.scope_id,
lineage_id=child_lineage_id,
parent_lineage_id=frame.lineage_id,
metadata=ForeachIterationMetadata(
foreach_node_id=step.id,
activation_id=activation.id,
loop_index=loop_index,
loop_item=item,
loop_alias=step.as_,
).to_metadata(),
),
ready=True,
)
block_frame_on_children(run, frame.id, (child_id,))
append_step_result_trace(
run,
@@ -251,6 +227,58 @@ def _item_error_record(child: ExecutionFrame) -> ItemErrorRecord:
)
def _admit_item_frame(
*,
run: RunState,
frame: ExecutionFrame,
step: ForeachNode,
index: WorkflowIndex,
activation: ForeachActivationState,
loop_index: int,
item: object,
) -> tuple[str, str]:
"""Create one activation-qualified child frame and lineage.
Every item owns a lineage so nested subgraph/boundary commits have a
parent lineage to buffer into; top-level serial writes still commit
through the parent scope root. Returns the loop start node and child id;
barrier child bookkeeping stays with the caller. Compare ids by name;
never parse them.
"""
loop_start = index.next_node_id(frame.node_id, "loop")
child_id = _child_frame_id(activation, loop_index)
child_lineage_id = _child_lineage_id(activation, loop_index)
add_lineage(
run,
scope_id=frame.scope_id,
lineage_id=child_lineage_id,
parent_id=frame.lineage_id,
)
activation.barrier.next_index = loop_index + 1
add_frame(
run,
ExecutionFrame(
id=child_id,
kind="foreach_iteration",
node_id=loop_start,
status=FrameStatus.PENDING,
parent_frame_id=frame.id,
scope_id=frame.scope_id,
lineage_id=child_lineage_id,
parent_lineage_id=frame.lineage_id,
metadata=ForeachIterationMetadata(
foreach_node_id=step.id,
activation_id=activation.id,
loop_index=loop_index,
loop_item=item,
loop_alias=step.as_,
).to_metadata(),
),
ready=True,
)
return loop_start, child_id
def _admit_concurrent_children(
*,
run: RunState,
@@ -272,38 +300,17 @@ def _admit_concurrent_children(
):
loop_index = barrier.next_index
item = iterable[loop_index]
child_id = _child_frame_id(activation, loop_index)
child_lineage_id = _child_lineage_id(activation, loop_index)
add_lineage(
run,
scope_id=frame.scope_id,
lineage_id=child_lineage_id,
parent_id=frame.lineage_id,
)
active_count = len(barrier.active_frame_ids)
barrier.next_index = loop_index + 1
barrier.start_child(child_id)
add_frame(
run,
ExecutionFrame(
id=child_id,
kind="foreach_iteration",
node_id=loop_start,
status=FrameStatus.PENDING,
parent_frame_id=frame.id,
scope_id=frame.scope_id,
lineage_id=child_lineage_id,
parent_lineage_id=frame.lineage_id,
metadata=ForeachIterationMetadata(
foreach_node_id=step.id,
activation_id=activation.id,
loop_index=loop_index,
loop_item=item,
loop_alias=step.as_,
).to_metadata(),
),
ready=True,
loop_start, child_id = _admit_item_frame(
run=run,
frame=frame,
step=step,
index=index,
activation=activation,
loop_index=loop_index,
item=item,
)
barrier.start_child(child_id)
append_step_result_trace(
run,
frame_id=frame.id,
@@ -417,14 +424,16 @@ def _patch_for_successful_item(
) -> StatePatch:
"""Return the replayable patch for a completed foreach item.
New concurrent foreach results store writes in `RunState.lineages` and keep
only lineage metadata in the barrier. Old serialized barrier metadata may
still carry `result.patch`, so keep that as the compatibility fallback.
Item writes live in `RunState.lineages`; a success without a known
lineage is corrupt state and fails closed.
"""
if result.lineage_id is not None and result.lineage_id in run.lineages:
return lineage_patch(
run,
scope_id=frame.scope_id,
lineage_id=result.lineage_id,
if result.lineage_id is None or result.lineage_id not in run.lineages:
raise WorkflowExecutionError(
f"foreach item result for index {result.index!r} references "
f"unknown lineage {result.lineage_id!r}"
)
return result.patch
return lineage_patch(
run,
scope_id=frame.scope_id,
lineage_id=result.lineage_id,
)
src/wf_core/runtime/ops/nodes.py
+10 -14
View File
@@ -18,7 +18,6 @@ from wf_core.run_state import (
from wf_core.runtime.foreach_state import (
item_frame_owner,
require_foreach_activation,
save_foreach_activation,
)
from wf_core.runtime.input_bindings import resolve_step_input_bindings
from wf_core.runtime.lineage import (
@@ -30,7 +29,7 @@ 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
from wf_core.runtime.ops.schemas import validate_payload_against_schema
from wf_core.runtime.ops.state import StatePatch, build_output_patch
from wf_core.runtime.ops.state import build_output_patch
NodeHandler = Callable[[dict[str, Any], RuntimeContext], NodeResult | dict[str, Any]]
AsyncNodeHandler = Callable[
@@ -120,29 +119,26 @@ def _finalize_node_execution(
if owner is None:
state_changes = commit_patch_for_frame(run, frame, patch)
else:
parent_frame = run.frames[owner.parent_frame_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 {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
)
barrier = activation.barrier
if barrier.mode == "concurrent":
# New concurrent foreach stores writes in the child lineage; the
# barrier keeps only result metadata plus old patch fallback.
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,
)
barrier.add_success_patch(
index=owner.item_index,
frame_id=frame.id,
patch=StatePatch(),
lineage_id=frame.lineage_id,
)
save_foreach_activation(parent_frame, activation)
state_changes = {}
else:
state_changes = commit_patch_for_frame(run, parent_frame, patch)
src/wf_core/runtime/subgraphs.py
+13 -13
View File
@@ -241,25 +241,25 @@ def _finish_subgraph(
# item writes stay buffered in the item lineage for barrier merge.
from wf_core.runtime.foreach_state import (
item_frame_owner,
load_foreach_activation,
require_foreach_activation,
)
commit_frame = frame
try:
owner = item_frame_owner(frame)
except Exception:
owner = None
owner = item_frame_owner(frame)
if owner is not None:
parent_frame = run.frames.get(owner.parent_frame_id)
if parent_frame is not None:
foreach_activation = load_foreach_activation(
parent_frame, owner.foreach_node_id, owner.activation_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}"
)
if (
foreach_activation is not None
and foreach_activation.barrier.mode == "serial"
):
commit_frame = parent_frame
# 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)
return StepExecutionResult(
outcome=child_outcome,