audit: close review gaps, dedupe control-region and activation helpers

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:

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(

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}"
         )

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(