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

execution frame carry a lineage id

Browse Source
This commit is contained in:
lda
2026-05-24 22:51:12 +07:00 Verified
parent 4576127d09
commit e48656c165
7 changed files with 236 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_core/run_state.py
+6
View File
@@ -47,6 +47,9 @@ class ExecutionFrame:
node_id: str
status: FrameStatus = FrameStatus.PENDING
parent_frame_id: str | None = None
scope_id: str = "root"
lineage_id: str = "root"
parent_lineage_id: str | None = None
prior_outcome: str | None = None
activated_incoming_edge: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@@ -57,6 +60,9 @@ class ExecutionFrame:
class RuntimeContext:
current_node_id: str
frame_id: str = "root"
scope_id: str = "root"
lineage_id: str = "root"
parent_lineage_id: str | None = None
retry_count: int = 0
prior_outcome: str | None = None
activated_incoming_edge: str | None = None
src/wf_core/runtime/foreach_state.py
+20 -1
View File
@@ -71,6 +71,7 @@ class PendingItemResult:
index: int
frame_id: str
status: Literal["succeeded", "failed"]
lineage_id: str | None = None
patch: StatePatch = field(default_factory=StatePatch)
error: ItemErrorRecord | None = None
@@ -88,10 +89,13 @@ class PendingItemResult:
) 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 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):
@@ -103,6 +107,7 @@ class PendingItemResult:
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]
@@ -122,6 +127,7 @@ class PendingItemResult:
"index": self.index,
"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
@@ -251,7 +257,12 @@ class ForeachBarrierState:
)
def add_success_patch(
self, *, index: int, frame_id: str, patch: StatePatch
self,
*,
index: int,
frame_id: str,
patch: StatePatch,
lineage_id: str | None = None,
) -> None:
"""Buffer or extend successful item patches by item index.
@@ -266,6 +277,7 @@ class ForeachBarrierState:
index=index,
frame_id=frame_id,
status="succeeded",
lineage_id=lineage_id,
patch=patch,
)
return
@@ -274,6 +286,13 @@ 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}:
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:
src/wf_core/runtime/ops/foreach.py
+17
View File
@@ -81,6 +81,7 @@ def _step_foreach_serial(
barrier.next_index = loop_index + 1
barrier.save_to_frame(frame, step.id)
child_id = f"{frame.id}:{step.id}:{loop_index}"
child_lineage_id = _child_lineage_id(frame, step, loop_index)
add_frame(
run,
ExecutionFrame(
@@ -89,6 +90,9 @@ def _step_foreach_serial(
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,
loop_index=loop_index,
@@ -243,6 +247,7 @@ def _admit_concurrent_children(
loop_index = barrier.next_index
item = iterable[loop_index]
child_id = f"{frame.id}:{step.id}:{loop_index}"
child_lineage_id = _child_lineage_id(frame, step, loop_index)
active_count = len(barrier.active_frame_ids)
barrier.next_index = loop_index + 1
barrier.start_child(child_id)
@@ -254,6 +259,9 @@ def _admit_concurrent_children(
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,
loop_index=loop_index,
@@ -344,3 +352,12 @@ def _finish_concurrent_foreach(
)
advance_frame(run, frame, outcome=outcome, next_node_id=next_node_id)
return run
def _child_lineage_id(frame: ExecutionFrame, step: ForeachNode, loop_index: int) -> str:
"""Return a deterministic opaque lineage id for one foreach child frame.
The readable shape is only for diagnostics. Runtime code should compare the
full id, not parse it; future structured lineage refs can replace this.
"""
return f"{frame.lineage_id}/{step.id}[{loop_index}]"
src/wf_core/runtime/ops/frames.py
+3
View File
@@ -7,6 +7,9 @@ def frame_context_values(frame: ExecutionFrame) -> dict[str, object | None]:
context: dict[str, object | None] = {
"prior_outcome": frame.prior_outcome,
"activated_incoming_edge": frame.activated_incoming_edge,
"scope_id": frame.scope_id,
"lineage_id": frame.lineage_id,
"parent_lineage_id": frame.parent_lineage_id,
}
if frame.kind == "foreach_iteration":
loop_item = frame.metadata.get("loop_item")
src/wf_core/runtime/ops/nodes.py
+4
View File
@@ -79,6 +79,9 @@ def _resolve_node_execution(
context = RuntimeContext(
current_node_id=node.id,
frame_id=frame.id,
scope_id=frame.scope_id,
lineage_id=frame.lineage_id,
parent_lineage_id=frame.parent_lineage_id,
prior_outcome=frame.prior_outcome,
activated_incoming_edge=frame.activated_incoming_edge,
metadata=dict(frame.metadata),
@@ -127,6 +130,7 @@ def _finalize_node_execution(
index=item_index,
frame_id=frame.id,
patch=patch,
lineage_id=frame.lineage_id,
)
barrier.save_to_frame(parent_frame, foreach_node_id)
state_changes = {}
tests/core/test_concurrent_foreach.py
+171
View File
@@ -18,6 +18,8 @@ from wf_core import (
WorkflowExecutionError,
execute_workflow,
)
from wf_core.run_state import ExecutionFrame, RunState, RuntimeContext
from wf_core.runtime.scheduler import ForeachIterationMetadata
def test_sync_concurrent_foreach_interleaves_items_and_commits_at_barrier() -> None:
@@ -96,6 +98,81 @@ def test_sync_concurrent_foreach_respects_max_active_by_refill_trace() -> None:
assert all(entry.resolved_input["active_count"] < 2 for entry in loop_entries)
def test_concurrent_foreach_item_frames_use_distinct_lineages() -> None:
workflow = _workflow(
state_schema=StateSchema.from_field_map(
{
"items": StateField(type="array"),
"seen": StateField(
type="array",
reducer=ReducerRef(name="wf.std.append"),
),
}
),
foreach=ForeachNode.model_validate(
{
"id": "each",
"type": "foreach",
"over": "state.items",
"as": "item",
"mode": "concurrent",
"concurrent": {"max_active": 2, "max_outstanding": 2},
}
),
)
context_lineage_ids: list[str] = []
def record(payload: dict[str, Any], ctx: RuntimeContext) -> dict[str, Any]:
context_lineage_ids.append(ctx.lineage_id)
return {"outcome": "ok", "output": payload}
run = execute_workflow(
workflow,
{"items": ["a", "b"]},
{"record": record},
)
item_frames = [
frame for frame in run.frames.values() if frame.kind == "foreach_iteration"
]
item_lineage_ids = {frame.lineage_id for frame in item_frames}
assert run.frames["root"].scope_id == "root"
assert run.frames["root"].lineage_id == "root"
assert run.frames["root"].parent_lineage_id is None
assert len(item_frames) == 2
assert item_lineage_ids == {"root/each[0]", "root/each[1]"}
assert set(context_lineage_ids) == item_lineage_ids
assert all(frame.scope_id == "root" for frame in item_frames)
assert all(frame.parent_lineage_id == "root" for frame in item_frames)
def test_nested_concurrent_foreach_records_parent_child_lineages() -> None:
workflow = _nested_foreach_lineage_workflow()
run = execute_workflow(
workflow,
{"outer_items": ["a", "b"], "inner_items": [1, 2]},
{"record": lambda payload, _ctx: {"outcome": "ok", "output": payload}},
)
outer_frames = _foreach_frames(run, "outer_each")
inner_frames = _foreach_frames(run, "inner_each")
assert {frame.lineage_id for frame in outer_frames} == {
"root/outer_each[0]",
"root/outer_each[1]",
}
assert all(frame.parent_lineage_id == "root" for frame in outer_frames)
assert {(frame.parent_lineage_id, frame.lineage_id) for frame in inner_frames} == {
("root/outer_each[0]", "root/outer_each[0]/inner_each[0]"),
("root/outer_each[0]", "root/outer_each[0]/inner_each[1]"),
("root/outer_each[1]", "root/outer_each[1]/inner_each[0]"),
("root/outer_each[1]", "root/outer_each[1]/inner_each[1]"),
}
def test_sync_concurrent_foreach_fails_run_on_item_runtime_error() -> None:
workflow = _workflow(
state_schema=StateSchema.from_field_map(
@@ -413,6 +490,91 @@ def _same_item_reducer_visibility_workflow() -> Workflow:
)
def _nested_foreach_lineage_workflow() -> Workflow:
outer = ForeachNode.model_validate(
{
"id": "outer_each",
"type": "foreach",
"over": "state.outer_items",
"as": "outer",
"mode": "concurrent",
"concurrent": {"max_active": 2, "max_outstanding": 2},
}
)
inner = ForeachNode.model_validate(
{
"id": "inner_each",
"type": "foreach",
"over": "state.inner_items",
"as": "inner",
"mode": "concurrent",
"concurrent": {"max_active": 2, "max_outstanding": 2},
}
)
return Workflow(
name="nested_foreach_lineages",
input_schema=SchemaRef(
type="object",
properties={
"outer_items": {"type": "array"},
"inner_items": {"type": "array"},
},
),
state_schema=StateSchema.from_field_map(
{
"outer_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="record",
input_schema=SchemaRef(
type="object",
properties={"seen": {}},
required=["seen"],
),
output_schema=SchemaRef(
type="object",
properties={"seen": {}},
required=["seen"],
),
outcomes=["ok"],
)
],
start="outer_each",
nodes=[
outer,
inner,
NodeUse.model_validate(
{
"id": "record",
"type": "node",
"node": "record",
"input": [{"target": "seen", "path": "context.inner"}],
"output": [{"source": "seen", "target": "state.seen"}],
}
),
],
edges=[
Edge.model_validate(
{"from": "outer_each", "outcome": "loop", "to": "inner_each"}
),
Edge.model_validate(
{"from": "inner_each", "outcome": "loop", "to": "record"}
),
Edge.model_validate({"from": "record", "outcome": "ok", "to": END}),
Edge.model_validate({"from": "inner_each", "outcome": "done", "to": END}),
Edge.model_validate({"from": "outer_each", "outcome": "done", "to": END}),
],
)
def _workflow(
*,
state_schema: StateSchema,
@@ -481,6 +643,15 @@ def _workflow(
)
def _foreach_frames(run: RunState, foreach_node_id: str) -> list[ExecutionFrame]:
frames = []
for frame in run.frames.values():
metadata = ForeachIterationMetadata.from_frame(frame)
if metadata is not None and metadata.foreach_node_id == foreach_node_id:
frames.append(frame)
return frames
def _multi_step_overlay_workflow() -> Workflow:
foreach = ForeachNode.model_validate(
{
tests/core/test_foreach_barrier_state.py
+15 -2
View File
@@ -61,6 +61,7 @@ def test_foreach_barrier_state_round_trips_reducer_write_records() -> None:
index=0,
frame_id="child-0",
status="succeeded",
lineage_id="root/each[0]",
patch=StatePatch(
writes=[
StateWrite(
@@ -80,6 +81,7 @@ def test_foreach_barrier_state_round_trips_reducer_write_records() -> None:
assert loaded is not None
write = loaded.pending_results[0].patch.writes[0]
assert loaded.pending_results[0].lineage_id == "root/each[0]"
assert write.path == StatePath(("count",))
assert write.incoming_value == 3
assert write.visible_value == 5
@@ -169,10 +171,21 @@ def test_foreach_barrier_accumulates_multiple_patches_for_one_item() -> None:
patch = StatePatch(changes={"state.count": 1})
second_patch = StatePatch(changes={"state.name": "a"})
barrier.add_success_patch(index=0, frame_id="child-0", patch=patch)
barrier.add_success_patch(index=0, frame_id="child-0", patch=second_patch)
barrier.add_success_patch(
index=0,
frame_id="child-0",
patch=patch,
lineage_id="root/each[0]",
)
barrier.add_success_patch(
index=0,
frame_id="child-0",
patch=second_patch,
lineage_id="root/each[0]",
)
result = barrier.pending_results[0]
assert result.lineage_id == "root/each[0]"
assert result.patch.changes["state.count"] == 1
assert result.patch.changes["state.name"] == "a"
assert len(result.patch.writes) == 2