From 79ce0d3effe32012864b6abc4c238cc9857de4f4 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 10:48:29 +0700 Subject: [PATCH] audit: close review gaps, dedupe control-region and activation helpers --- src/wf_core/analysis/context_scopes.py | 3 +- src/wf_core/analysis/control_regions.py | 53 ++++++++++------------ src/wf_core/runtime/foreach_state.py | 32 ++++++++----- src/wf_core/runtime/ops/foreach.py | 6 +-- tests/core/test_foreach_back_edges.py | 26 +++++++++++ tests/core/test_foreach_control_regions.py | 25 ++++++++++ 6 files changed, 100 insertions(+), 45 deletions(-) diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 87bc8529..1d2b23ce 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -145,7 +145,8 @@ def _available_fields( A single node use has one control region, so foreach fields are either present (inside a body) or absent (outside). Conditional availability is - not used to represent multiple owner stacks. + not used to represent multiple owner stacks: a node reached under two + stacks is a region conflict and receives no foreach fields at all. """ contracts = list(STANDARD_CONTEXT_FIELDS) if active_scope is not None: diff --git a/src/wf_core/analysis/control_regions.py b/src/wf_core/analysis/control_regions.py index 05bf4795..8b8063ba 100644 --- a/src/wf_core/analysis/control_regions.py +++ b/src/wf_core/analysis/control_regions.py @@ -68,6 +68,26 @@ def analyze_control_regions(workflow: Workflow) -> ControlRegionAnalysis: if stack: ambiguous_tops.add(stack[-1]) + def record_region_conflict( + node_id: str, stacks: tuple[ForeachOwnerStack, ...] + ) -> None: + """Drop one node use reached under two regions and report it once.""" + del owner_stack_by_node[node_id] + if node_id not in conflicted: + conflicted.add(node_id) + issues.append( + ControlRegionIssue( + kind=ControlRegionIssueKind.FOREACH_REGION_CONFLICT, + path=f"nodes[{node_id}]", + message=( + f"node {node_id!r} is reachable under two foreach " + "control regions" + ), + ) + ) + for prior_stack in stacks: + mark_ambiguous(prior_stack) + def add_adjacency( source: tuple[str, ForeachOwnerStack], target: tuple[str, ForeachOwnerStack], @@ -99,20 +119,7 @@ def analyze_control_regions(workflow: Workflow) -> ControlRegionAnalysis: # single static owner stack. Drop it so later context analysis # grants no foreach fields, and stop expanding this ambiguous # state so the conflict does not cascade. - del owner_stack_by_node[node_id] - conflicted.add(node_id) - issues.append( - ControlRegionIssue( - kind=ControlRegionIssueKind.FOREACH_REGION_CONFLICT, - path=f"nodes[{node_id}]", - message=( - f"node {node_id!r} is reachable under two foreach " - "control regions" - ), - ) - ) - for prior_stack in (recorded, stack): - mark_ambiguous(prior_stack) + record_region_conflict(node_id, (recorded, stack)) continue for edge_index, edge in edges_by_node.get(node_id, []): # type: ignore[attr-defined] @@ -151,21 +158,9 @@ def analyze_control_regions(workflow: Workflow) -> ControlRegionAnalysis: if recorded_target is None: owner_stack_by_node[target_id] = target_stack elif recorded_target != target_stack: - del owner_stack_by_node[target_id] - if target_id not in conflicted: - conflicted.add(target_id) - issues.append( - ControlRegionIssue( - kind=ControlRegionIssueKind.FOREACH_REGION_CONFLICT, - path=f"nodes[{target_id}]", - message=( - f"node {target_id!r} is reachable under " - "two foreach control regions" - ), - ) - ) - for prior_stack in (recorded_target, target_stack): - mark_ambiguous(prior_stack) + record_region_conflict( + target_id, (recorded_target, target_stack) + ) if target_stack: issues.append( ControlRegionIssue( diff --git a/src/wf_core/runtime/foreach_state.py b/src/wf_core/runtime/foreach_state.py index 6aa8236a..eed386db 100644 --- a/src/wf_core/runtime/foreach_state.py +++ b/src/wf_core/runtime/foreach_state.py @@ -345,6 +345,20 @@ class ForeachBarrierState: ) +def _activation_entry( + frame: ExecutionFrame, table: dict[str, Any], foreach_node_id: str +) -> dict[str, Any] | None: + """Return the mutable activation entry or fail fast on corrupt state.""" + entry = table.get(foreach_node_id) + if entry is None: + return None + if not isinstance(entry, dict): + raise WorkflowExecutionError( + f"malformed foreach activation entry for frame {frame.id!r}" + ) + return entry + + def load_or_begin_foreach_activation( frame: ExecutionFrame, foreach_node_id: str, @@ -359,14 +373,10 @@ def load_or_begin_foreach_activation( with fresh barrier state. Mode mismatches and malformed tables fail fast. """ table = _activation_table(frame) - entry = table.get(foreach_node_id) + entry = _activation_entry(frame, table, foreach_node_id) if entry is None: entry = {"next_sequence": 0, "active": None} table[foreach_node_id] = entry - if not isinstance(entry, dict): - raise WorkflowExecutionError( - f"malformed foreach activation entry for frame {frame.id!r}" - ) next_sequence = entry.get("next_sequence", 0) if not isinstance(next_sequence, int) or next_sequence < 0: raise WorkflowExecutionError( @@ -402,8 +412,8 @@ def save_foreach_activation( ) -> None: """Persist barrier progress for the named active activation.""" table = _activation_table(frame) - entry = table.get(activation.foreach_node_id) - if not isinstance(entry, dict): + entry = _activation_entry(frame, table, activation.foreach_node_id) + if entry is None: raise WorkflowExecutionError( f"malformed foreach activation entry for frame {frame.id!r}" ) @@ -425,8 +435,8 @@ def close_foreach_activation( increasing so child and lineage ids cannot collide across visits. """ table = _activation_table(frame) - entry = table.get(activation.foreach_node_id) - if not isinstance(entry, dict): + entry = _activation_entry(frame, table, activation.foreach_node_id) + if entry is None: raise WorkflowExecutionError( f"malformed foreach activation entry for frame {frame.id!r}" ) @@ -448,8 +458,8 @@ def load_foreach_activation( the caller rather than buffering into the wrong barrier. """ table = _activation_table(frame) - entry = table.get(foreach_node_id) - if not isinstance(entry, dict): + entry = _activation_entry(frame, table, foreach_node_id) + if entry is None: raise WorkflowExecutionError( f"malformed foreach activation entry for frame {frame.id!r}" ) diff --git a/src/wf_core/runtime/ops/foreach.py b/src/wf_core/runtime/ops/foreach.py index dec9178b..99c0c972 100644 --- a/src/wf_core/runtime/ops/foreach.py +++ b/src/wf_core/runtime/ops/foreach.py @@ -170,7 +170,6 @@ def _step_foreach_concurrent( step=step, index=index, activation=activation, - barrier=barrier, iterable=iterable, ) @@ -182,7 +181,6 @@ def _step_foreach_concurrent( step=step, index=index, activation=activation, - barrier=barrier, reducers=reducers, ) @@ -260,12 +258,12 @@ def _admit_concurrent_children( step: ForeachNode, index: WorkflowIndex, activation: ForeachActivationState, - barrier: ForeachBarrierState, iterable: list[object], ) -> None: if step.concurrent is None: raise WorkflowExecutionError("concurrent foreach requires concurrent policy") + barrier = activation.barrier loop_start = index.next_node_id(frame.node_id, "loop") while ( barrier.next_index < len(iterable) @@ -333,9 +331,9 @@ def _finish_concurrent_foreach( step: ForeachNode, index: WorkflowIndex, activation: ForeachActivationState, - barrier: ForeachBarrierState, reducers: Mapping[str, ReducerDefinition] | None = None, ) -> RunState: + barrier = activation.barrier error_records = [ result.error.to_metadata() for result in sorted( diff --git a/tests/core/test_foreach_back_edges.py b/tests/core/test_foreach_back_edges.py index 8a2b969e..5512c60b 100644 --- a/tests/core/test_foreach_back_edges.py +++ b/tests/core/test_foreach_back_edges.py @@ -695,6 +695,32 @@ def test_nonlocal_runtime_return_fails_closed_when_validation_is_bypassed() -> N advance_frame(run, run.frames["inner-item"], outcome="ok", next_node_id="outer") +def test_root_frame_targeting_foreach_enters_normally() -> None: + """A root frame naming a foreach enters it; only an item frame returns.""" + from wf_core.runtime.ops.flow import advance_frame + + run = RunState( + workflow_name="root_entry", + status=RunStatus.RUNNING, + workflow_input={}, + state={}, + frames={}, + ) + add_frame( + run, + ExecutionFrame(id="root", kind="workflow", node_id="start"), + ) + run.current_frame_id = "root" + run.sync_from_current_frame() + + advance_frame(run, run.frames["root"], outcome="ok", next_node_id="each") + + entered = run.frames["root"] + assert entered.node_id == "each" + assert entered.status == FrameStatus.PENDING + assert entered.finished_at_node_id is None + + def test_completed_activation_cannot_consume_later_activation_result_or_wake() -> None: """A closed visit rejects buffered results and wake-ups from other visits.""" from wf_core.runtime.foreach_state import ( diff --git a/tests/core/test_foreach_control_regions.py b/tests/core/test_foreach_control_regions.py index e81110e8..c48469e5 100644 --- a/tests/core/test_foreach_control_regions.py +++ b/tests/core/test_foreach_control_regions.py @@ -265,6 +265,31 @@ def test_skipping_inner_foreach_owner_is_invalid_return() -> None: assert matching[0].path == "edges[2]" +def test_reentering_active_ancestor_foreach_as_nested_controller_is_invalid() -> None: + workflow = _workflow( + start="f1", + nodes=[_foreach("f1"), _foreach("f2")], + edges=[ + {"from": "f1", "outcome": "loop", "to": "f2"}, + {"from": "f2", "outcome": "loop", "to": "f1"}, + {"from": "f2", "outcome": "done", "to": "f1"}, + {"from": "f1", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.INVALID_FOREACH_RETURN, "edges[1]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + matching = [ + issue + for issue in workflow.validate_structure().errors + if issue.code == ValidationIssueCode.INVALID_FOREACH_RETURN + ] + assert matching[0].path == "edges[1]" + + def test_entering_sibling_foreach_body_is_region_conflict() -> None: workflow = _workflow( start="f1",