From 6a5d886962b0a6c4fcefc7fd89bd0b74efa12e36 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 07:15:24 +0700 Subject: [PATCH] feat: analyze foreach control regions --- src/wf_core/analysis/__init__.py | 12 + src/wf_core/analysis/control_regions.py | 255 +++++++++++++++ tests/core/test_foreach_control_regions.py | 347 +++++++++++++++++++++ 3 files changed, 614 insertions(+) create mode 100644 src/wf_core/analysis/control_regions.py create mode 100644 tests/core/test_foreach_control_regions.py diff --git a/src/wf_core/analysis/__init__.py b/src/wf_core/analysis/__init__.py index 44fab703..63815931 100644 --- a/src/wf_core/analysis/__init__.py +++ b/src/wf_core/analysis/__init__.py @@ -5,9 +5,21 @@ from .context_scopes import ( context_analysis_warnings, context_fields_by_node, ) +from .control_regions import ( + ControlRegionAnalysis, + ControlRegionIssue, + ControlRegionIssueKind, + ForeachOwnerStack, + analyze_control_regions, +) __all__ = [ "ContextFieldAvailability", + "ControlRegionAnalysis", + "ControlRegionIssue", + "ControlRegionIssueKind", + "ForeachOwnerStack", + "analyze_control_regions", "context_analysis_warnings", "context_fields_by_node", ] diff --git a/src/wf_core/analysis/control_regions.py b/src/wf_core/analysis/control_regions.py new file mode 100644 index 00000000..81193aad --- /dev/null +++ b/src/wf_core/analysis/control_regions.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +from enum import StrEnum + +from wf_core.models.steps import EndNode, ForeachNode +from wf_core.models.workflow import Workflow +from wf_core.tokens import END + +type ForeachOwnerStack = tuple[str, ...] + + +class ControlRegionIssueKind(StrEnum): + UNREACHABLE_NODE = "unreachable_node" + FOREACH_REGION_CONFLICT = "foreach_region_conflict" + INVALID_FOREACH_RETURN = "invalid_foreach_return" + INVALID_FOREACH_TERMINAL = "invalid_foreach_terminal" + EMPTY_FOREACH_BODY = "empty_foreach_body" + FOREACH_BODY_NO_RETURN = "foreach_body_no_return" + + +@dataclass(frozen=True, slots=True) +class ControlRegionIssue: + kind: ControlRegionIssueKind + path: str + message: str + + +@dataclass(frozen=True, slots=True) +class ControlRegionAnalysis: + owner_stack_by_node: dict[str, ForeachOwnerStack] + issues: tuple[ControlRegionIssue, ...] + + +def analyze_control_regions(workflow: Workflow) -> ControlRegionAnalysis: + """Derive one static foreach-owner stack per reachable node use. + + Traversal is over ``(node_id, owner_stack)`` states. A ``loop`` edge from + a foreach pushes that controller; an edge targeting the immediate owner is + an item return that resumes the owner in the popped stack; targeting an + older ancestor is a non-local return; targeting ``END``/``EndNode`` inside + a body is an invalid terminal. Reaching the same node under two stacks is + a region conflict. After traversal every unreached node is unreachable and + every reached body state must have a structural path back to its top owner. + """ + nodes_by_id = {node.id: node for node in workflow.nodes} + if workflow.start not in nodes_by_id: + return ControlRegionAnalysis(owner_stack_by_node={}, issues=()) + + edges_by_node: dict[str, list[tuple[int, object]]] = {} + for index, edge in enumerate(workflow.edges): + edges_by_node.setdefault(edge.from_, []).append((index, edge)) + + owner_stack_by_node: dict[str, ForeachOwnerStack] = {} + conflicted: set[str] = set() + issues: list[ControlRegionIssue] = [] + # Semantic state adjacency for the structural-return check. Return edges + # also link to the resumed owner state so deeper nested returns are part + # of the path search. + adjacency: dict[ + tuple[str, ForeachOwnerStack], list[tuple[str, ForeachOwnerStack]] + ] = {} + return_owner_by_source: dict[tuple[str, ForeachOwnerStack], str] = {} + ambiguous_tops: set[str] = set() + + def mark_ambiguous(stack: ForeachOwnerStack) -> None: + if stack: + ambiguous_tops.add(stack[-1]) + + def add_adjacency( + source: tuple[str, ForeachOwnerStack], + target: tuple[str, ForeachOwnerStack], + ) -> None: + adjacency.setdefault(source, []).append(target) + + pending: deque[tuple[str, ForeachOwnerStack]] = deque([(workflow.start, ())]) + visited: set[tuple[str, ForeachOwnerStack]] = set() + visited_nodes: set[str] = set() + + while pending: + node_id, stack = pending.popleft() + state = (node_id, stack) + if state in visited: + continue + visited.add(state) + node = nodes_by_id.get(node_id) + if node is None: + continue + visited_nodes.add(node_id) + + if node_id in conflicted: + continue + recorded = owner_stack_by_node.get(node_id) + if recorded is None: + owner_stack_by_node[node_id] = stack + elif recorded != stack: + # Same node use reached under two control regions: it has no + # 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) + continue + + for edge_index, edge in edges_by_node.get(node_id, []): # type: ignore[attr-defined] + target_id: str = edge.to # type: ignore[attr-defined] + source_is_loop = isinstance(node, ForeachNode) and edge.outcome == "loop" # type: ignore[attr-defined] + if source_is_loop: + if target_id == node_id: + issues.append( + ControlRegionIssue( + kind=ControlRegionIssueKind.EMPTY_FOREACH_BODY, + path=f"edges[{edge_index}]", + message=( + f"foreach {node_id!r} loop targets itself; " + "an iteration body needs a distinct node use" + ), + ) + ) + mark_ambiguous(stack) + continue + target_stack: ForeachOwnerStack = (*stack, node_id) + else: + target_stack = stack + + target_node = None if target_id == END else nodes_by_id.get(target_id) + if target_id != END and target_node is None: + # Unknown destinations are owned by ordinary edge validation. + continue + is_terminal = target_id == END or isinstance(target_node, EndNode) + if is_terminal: + if target_stack: + issues.append( + ControlRegionIssue( + kind=ControlRegionIssueKind.INVALID_FOREACH_TERMINAL, + path=f"edges[{edge_index}]", + message=( + f"foreach item path {node_id!r} -> " + f"{target_id!r} targets a workflow terminal " + "from inside a foreach body" + ), + ) + ) + mark_ambiguous(target_stack) + continue + + # At this point target_id is a known non-terminal node id. + if target_stack and target_id == target_stack[-1]: + # Immediate-owner back-edge: the item frame completes at its + # owner without executing the controller again. Resume the + # owner in the popped stack for structural analysis. + resumed: tuple[str, ForeachOwnerStack] = ( + target_id, + target_stack[:-1], + ) + return_owner_by_source[state] = target_id + add_adjacency(state, resumed) + if resumed not in visited: + pending.append(resumed) + continue + if target_id in target_stack: + issues.append( + ControlRegionIssue( + kind=ControlRegionIssueKind.INVALID_FOREACH_RETURN, + path=f"edges[{edge_index}]", + message=( + f"edge {node_id!r} -> {target_id!r} skips the " + "immediate foreach owner" + ), + ) + ) + mark_ambiguous(target_stack) + continue + successor: tuple[str, ForeachOwnerStack] = (target_id, target_stack) + add_adjacency(state, successor) + pending.append(successor) + + for node in workflow.nodes: + if node.id not in visited_nodes: + issues.append( + ControlRegionIssue( + kind=ControlRegionIssueKind.UNREACHABLE_NODE, + path=f"nodes[{node.id}]", + message=f"node {node.id!r} is unreachable from start", + ) + ) + + # Structural returnability: every reached body state needs some graph path + # back to its immediate owner. Data decides whether the exit is taken, so + # one possible path is enough. Skip bodies already made ambiguous by a + # region conflict, invalid return/terminal, or empty body. + tops_with_no_return: set[str] = set() + for node_id, stack in list(visited): + if not stack: + continue + if node_id in conflicted: + continue + top = stack[-1] + if top in ambiguous_tops: + continue + if top in tops_with_no_return: + continue + # Breadth-first search over semantic states for a return to `top`. + seen: set[tuple[str, ForeachOwnerStack]] = set() + queue: deque[tuple[str, ForeachOwnerStack]] = deque([(node_id, stack)]) + found = False + while queue: + current = queue.popleft() + if current in seen: + continue + seen.add(current) + if return_owner_by_source.get(current) == top: + found = True + break + for successor in adjacency.get(current, []): + if successor not in seen: + queue.append(successor) + if not found: + tops_with_no_return.add(top) + + for top in sorted(tops_with_no_return): + # Only report when the owner itself is unambiguous; a conflicted + # owner has no single region to return to. + if top in conflicted: + continue + if top not in owner_stack_by_node and top not in visited_nodes: + continue + issues.append( + ControlRegionIssue( + kind=ControlRegionIssueKind.FOREACH_BODY_NO_RETURN, + path=f"nodes[{top}]", + message=( + f"foreach {top!r} body has no structural path back to " + "its immediate owner" + ), + ) + ) + + return ControlRegionAnalysis( + owner_stack_by_node=dict(owner_stack_by_node), + issues=tuple(issues), + ) diff --git a/tests/core/test_foreach_control_regions.py b/tests/core/test_foreach_control_regions.py new file mode 100644 index 00000000..b3844f2f --- /dev/null +++ b/tests/core/test_foreach_control_regions.py @@ -0,0 +1,347 @@ +from __future__ import annotations + +from wf_core import END, Workflow +from wf_core.analysis.control_regions import ( + ControlRegionAnalysis, + ControlRegionIssueKind, + analyze_control_regions, +) + + +def _workflow( + *, + start: str, + nodes: list[dict[str, object]], + edges: list[dict[str, str]], +) -> Workflow: + return Workflow.model_validate( + { + "name": "control-regions", + "input_schema": {"type": "object", "properties": {}}, + "state_schema": { + "type": "object", + "properties": { + "items": {"type": "array", "items": {"type": "string"}}, + "inner_items": {"type": "array", "items": {"type": "integer"}}, + }, + }, + "output_schema": {"type": "object", "properties": {}}, + "start": start, + "nodes": nodes, + "edges": edges, + "node_defs": [], + } + ) + + +def _node(node_id: str) -> dict[str, object]: + return {"id": node_id, "type": "node", "node": "noop"} + + +def _foreach(node_id: str, *, alias: str = "item") -> dict[str, object]: + return { + "id": node_id, + "type": "foreach", + "over": "state.items", + "as": alias, + "mode": "serial", + } + + +def _condition(node_id: str) -> dict[str, object]: + return { + "id": node_id, + "type": "condition", + "check": {"op": "exists", "path": "state.items"}, + } + + +def test_closed_root_cycle_has_one_empty_control_region() -> None: + workflow = _workflow( + start="a", + nodes=[_node("a"), _node("b")], + edges=[ + {"from": "a", "outcome": "ok", "to": "b"}, + {"from": "b", "outcome": "ok", "to": "a"}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert analysis.issues == () + assert analysis.owner_stack_by_node == {"a": (), "b": ()} + + +def test_foreach_cycle_with_possible_return_is_valid() -> None: + workflow = _workflow( + start="f", + nodes=[_foreach("f"), _node("a")], + edges=[ + {"from": "f", "outcome": "loop", "to": "a"}, + {"from": "a", "outcome": "again", "to": "a"}, + {"from": "a", "outcome": "done", "to": "f"}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert analysis.issues == () + assert analysis.owner_stack_by_node["a"] == ("f",) + assert analysis.owner_stack_by_node["f"] == () + + +def test_conditional_foreach_paths_can_both_return() -> None: + workflow = _workflow( + start="f", + nodes=[_foreach("f"), _condition("condition"), _node("work")], + edges=[ + {"from": "f", "outcome": "loop", "to": "condition"}, + {"from": "condition", "outcome": "true", "to": "work"}, + {"from": "condition", "outcome": "false", "to": "f"}, + {"from": "work", "outcome": "ok", "to": "f"}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert analysis.issues == () + assert analysis.owner_stack_by_node["condition"] == ("f",) + assert analysis.owner_stack_by_node["work"] == ("f",) + + +def test_nested_foreach_assigns_static_owner_stacks() -> None: + workflow = _workflow( + start="f1", + nodes=[ + _foreach("f1"), + _foreach("f2"), + _node("work"), + _node("tail"), + _node("after"), + ], + edges=[ + {"from": "f1", "outcome": "loop", "to": "f2"}, + {"from": "f2", "outcome": "loop", "to": "work"}, + {"from": "work", "outcome": "ok", "to": "f2"}, + {"from": "f2", "outcome": "done", "to": "tail"}, + {"from": "tail", "outcome": "ok", "to": "f1"}, + {"from": "f1", "outcome": "done", "to": "after"}, + {"from": "after", "outcome": "ok", "to": END}, + ], + ) + + analysis: ControlRegionAnalysis = analyze_control_regions(workflow) + + assert analysis.owner_stack_by_node == { + "f1": (), + "f2": ("f1",), + "work": ("f1", "f2"), + "tail": ("f1",), + "after": (), + } + assert analysis.issues == () + + +def test_reentering_completed_foreach_keeps_one_static_region() -> None: + workflow = _workflow( + start="again", + nodes=[_condition("again"), _foreach("f"), _node("work")], + edges=[ + {"from": "again", "outcome": "true", "to": "f"}, + {"from": "f", "outcome": "loop", "to": "work"}, + {"from": "work", "outcome": "ok", "to": "f"}, + {"from": "f", "outcome": "done", "to": "again"}, + {"from": "again", "outcome": "false", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert analysis.issues == () + assert analysis.owner_stack_by_node["f"] == () + assert analysis.owner_stack_by_node["work"] == ("f",) + assert analysis.owner_stack_by_node["again"] == () + + +def test_external_entry_into_foreach_body_is_region_conflict() -> None: + workflow = _workflow( + start="start", + nodes=[_condition("start"), _foreach("f"), _node("b")], + edges=[ + {"from": "start", "outcome": "true", "to": "f"}, + {"from": "start", "outcome": "false", "to": "b"}, + {"from": "f", "outcome": "loop", "to": "b"}, + {"from": "b", "outcome": "ok", "to": "f"}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.FOREACH_REGION_CONFLICT, "nodes[b]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + assert "b" not in analysis.owner_stack_by_node + + +def test_foreach_body_escape_is_region_conflict() -> None: + workflow = _workflow( + start="f", + nodes=[_foreach("f"), _node("b"), _node("after")], + edges=[ + {"from": "f", "outcome": "loop", "to": "b"}, + {"from": "b", "outcome": "ok", "to": "after"}, + {"from": "f", "outcome": "done", "to": "after"}, + {"from": "after", "outcome": "ok", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.FOREACH_REGION_CONFLICT, "nodes[after]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + assert "after" not in analysis.owner_stack_by_node + + +def test_skipping_inner_foreach_owner_is_invalid_return() -> None: + workflow = _workflow( + start="f1", + nodes=[_foreach("f1"), _foreach("f2"), _node("work")], + edges=[ + {"from": "f1", "outcome": "loop", "to": "f2"}, + {"from": "f2", "outcome": "loop", "to": "work"}, + {"from": "work", "outcome": "ok", "to": "f1"}, + {"from": "f1", "outcome": "done", "to": END}, + {"from": "f2", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.INVALID_FOREACH_RETURN, "edges[2]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + + +def test_entering_sibling_foreach_body_is_region_conflict() -> None: + workflow = _workflow( + start="f1", + nodes=[_foreach("f1"), _foreach("f2"), _node("b1"), _node("b2")], + edges=[ + {"from": "f1", "outcome": "loop", "to": "b1"}, + {"from": "b1", "outcome": "ok", "to": "b2"}, + {"from": "f2", "outcome": "loop", "to": "b2"}, + {"from": "b2", "outcome": "ok", "to": "f1"}, + {"from": "f1", "outcome": "done", "to": "f2"}, + {"from": "f2", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.FOREACH_REGION_CONFLICT, "nodes[b2]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + + +def test_empty_foreach_body_is_rejected() -> None: + workflow = _workflow( + start="f", + nodes=[_foreach("f")], + edges=[ + {"from": "f", "outcome": "loop", "to": "f"}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.EMPTY_FOREACH_BODY, "edges[0]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + + +def test_closed_foreach_body_cycle_has_no_return() -> None: + workflow = _workflow( + start="f", + nodes=[_foreach("f"), _node("a"), _node("b")], + edges=[ + {"from": "f", "outcome": "loop", "to": "a"}, + {"from": "a", "outcome": "ok", "to": "b"}, + {"from": "b", "outcome": "ok", "to": "a"}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.FOREACH_BODY_NO_RETURN, "nodes[f]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + + +def test_foreach_body_cannot_target_end_token() -> None: + workflow = _workflow( + start="f", + nodes=[_foreach("f"), _node("body")], + edges=[ + {"from": "f", "outcome": "loop", "to": "body"}, + {"from": "body", "outcome": "ok", "to": END}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.INVALID_FOREACH_TERMINAL, "edges[1]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + + +def test_foreach_body_cannot_target_explicit_end_node() -> None: + workflow = _workflow( + start="f", + nodes=[ + _foreach("f"), + _node("body"), + {"id": "stop", "type": "end", "outcome": "ok"}, + ], + edges=[ + {"from": "f", "outcome": "loop", "to": "body"}, + {"from": "body", "outcome": "ok", "to": "stop"}, + {"from": "f", "outcome": "done", "to": END}, + ], + ) + + analysis = analyze_control_regions(workflow) + + assert (ControlRegionIssueKind.INVALID_FOREACH_TERMINAL, "edges[1]") in [ + (issue.kind, issue.path) for issue in analysis.issues + ] + + +def test_every_unreachable_node_is_reported() -> None: + workflow = _workflow( + start="work", + nodes=[_node("work"), _node("detached_a"), _node("detached_b")], + edges=[ + {"from": "work", "outcome": "ok", "to": END}, + {"from": "detached_a", "outcome": "ok", "to": "detached_b"}, + {"from": "detached_b", "outcome": "ok", "to": "detached_a"}, + ], + ) + + analysis = analyze_control_regions(workflow) + by_kind_path = [(issue.kind, issue.path) for issue in analysis.issues] + + assert ( + ControlRegionIssueKind.UNREACHABLE_NODE, + "nodes[detached_a]", + ) in by_kind_path + assert ( + ControlRegionIssueKind.UNREACHABLE_NODE, + "nodes[detached_b]", + ) in by_kind_path