more code review and plans
This commit is contained in:
@@ -222,6 +222,13 @@ class ForeachBarrierState:
|
||||
|
||||
def finish_child(self, frame_id: str) -> None:
|
||||
"""Record one child frame as no longer active or outstanding."""
|
||||
if (
|
||||
frame_id not in self.active_frame_ids
|
||||
or frame_id not in self.outstanding_frame_ids
|
||||
):
|
||||
raise WorkflowExecutionError(
|
||||
f"foreach child frame {frame_id!r} is not active"
|
||||
)
|
||||
self.active_frame_ids = tuple(
|
||||
item for item in self.active_frame_ids if item != frame_id
|
||||
)
|
||||
@@ -233,6 +240,10 @@ class ForeachBarrierState:
|
||||
self, *, index: int, frame_id: str, patch: StatePatch
|
||||
) -> None:
|
||||
"""Buffer one successful item patch by item index."""
|
||||
if index in self.pending_results:
|
||||
raise WorkflowExecutionError(
|
||||
f"foreach item result for index {index!r} already recorded"
|
||||
)
|
||||
self.pending_results[index] = PendingItemResult(
|
||||
index=index,
|
||||
frame_id=frame_id,
|
||||
|
||||
@@ -223,11 +223,10 @@ def _admit_concurrent_children(
|
||||
index: WorkflowIndex,
|
||||
barrier: ForeachBarrierState,
|
||||
iterable: list[object],
|
||||
) -> int:
|
||||
) -> None:
|
||||
if step.concurrent is None:
|
||||
raise WorkflowExecutionError("concurrent foreach requires concurrent policy")
|
||||
|
||||
admitted = 0
|
||||
loop_start = index.next_node_id(frame.node_id, "loop")
|
||||
while (
|
||||
barrier.next_index < len(iterable)
|
||||
@@ -274,8 +273,6 @@ def _admit_concurrent_children(
|
||||
state_changes={},
|
||||
),
|
||||
)
|
||||
admitted += 1
|
||||
return admitted
|
||||
|
||||
|
||||
def _finish_concurrent_foreach(
|
||||
|
||||
@@ -158,6 +158,11 @@ def build_barrier_patch(
|
||||
writes cannot be blindly merged because reducers must see the value produced
|
||||
by earlier item patches. The barrier therefore replays trace-facing incoming
|
||||
changes against a single staged state in deterministic item order.
|
||||
|
||||
Unlike ordinary node patches, barrier patch `changes` report the final
|
||||
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`.
|
||||
"""
|
||||
state_fields = workflow.state_schema.field_index()
|
||||
staged_state = deepcopy(state)
|
||||
|
||||
Reference in New Issue
Block a user