diff --git a/src/wf_core/validation/core.py b/src/wf_core/validation/core.py index fb5462fa..2dfdce88 100644 --- a/src/wf_core/validation/core.py +++ b/src/wf_core/validation/core.py @@ -25,6 +25,14 @@ from wf_core.validation.steps import ( def validate_workflow(workflow: Workflow) -> ValidationReport: + """Coordinate structural validation including foreach control regions. + + Ordinary node/edge checks run first; the pure control-region analysis runs + once afterwards and its diagnostics are translated verbatim. No second + graph traversal lives inside validation. + """ + from wf_core.analysis.control_regions import analyze_control_regions + report = ValidationReport() node_defs = _collect_node_defs(workflow, report) @@ -33,6 +41,12 @@ def validate_workflow(workflow: Workflow) -> ValidationReport: _validate_start(workflow, nodes_by_id, report) outgoing = _validate_edges(workflow, nodes_by_id, node_defs, report) _validate_reachable_outcomes(workflow, nodes_by_id, node_defs, outgoing, report) + for issue in analyze_control_regions(workflow).issues: + report.add( + ValidationIssueCode(issue.kind.value), + issue.path, + issue.message, + ) return report diff --git a/src/wf_core/validation/issues.py b/src/wf_core/validation/issues.py index ae4fc3a9..c94afffc 100644 --- a/src/wf_core/validation/issues.py +++ b/src/wf_core/validation/issues.py @@ -26,6 +26,12 @@ class ValidationIssueCode(StrEnum): INVALID_FOREACH_COLLECT_DESTINATION = "invalid_foreach_collect_destination" INVALID_INTERRUPT_SOURCE = "invalid_interrupt_source" INVALID_INTERRUPT_DESTINATION = "invalid_interrupt_destination" + 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(slots=True) diff --git a/tests/artifacts/test_draft_adapter.py b/tests/artifacts/test_draft_adapter.py index 3f9d3b63..9422929a 100644 --- a/tests/artifacts/test_draft_adapter.py +++ b/tests/artifacts/test_draft_adapter.py @@ -505,14 +505,16 @@ def test_adapter_lowers_foreach_policy_through_builder() -> None: "collect_to": "state.item_errors", }, } - } + }, + "echo": {"use": "demo.echo"}, }, "routes": { "each_item": { - "loop": "__end__", + "loop": "echo", "done": "__end__", "completed_with_errors": "__end__", - } + }, + "echo": {"ok": "each_item"}, }, } ) diff --git a/tests/artifacts/test_draft_models.py b/tests/artifacts/test_draft_models.py index b839d5dd..22d2b5c5 100644 --- a/tests/artifacts/test_draft_models.py +++ b/tests/artifacts/test_draft_models.py @@ -322,7 +322,7 @@ def test_workflow_draft_foreach_over_dumps_structural_path() -> None: }, "routes": { "each_item": {"loop": "echo", "done": "__end__"}, - "echo": {"ok": "__end__"}, + "echo": {"ok": "each_item"}, }, } ) @@ -338,6 +338,7 @@ def test_workflow_draft_foreach_accepts_canonical_item_error_policy() -> None: **_keyed_echo_draft(), "start": "each_item", "steps": { + **_keyed_echo_draft()["steps"], "each_item": { "foreach": { "over": "state.items", @@ -349,9 +350,12 @@ def test_workflow_draft_foreach_accepts_canonical_item_error_policy() -> None: "collect_to": "state.item_errors", }, } - } + }, + }, + "routes": { + "each_item": {"loop": "echo", "done": "__end__"}, + "echo": {"ok": "each_item"}, }, - "routes": {"each_item": {"loop": "__end__", "done": "__end__"}}, } ) @@ -372,15 +376,19 @@ def test_workflow_draft_foreach_accepts_item_error_action_string() -> None: **_keyed_echo_draft(), "start": "each_item", "steps": { + **_keyed_echo_draft()["steps"], "each_item": { "foreach": { "over": "state.items", "as": "item", "item_error": "skip", } - } + }, + }, + "routes": { + "each_item": {"loop": "echo", "done": "__end__"}, + "echo": {"ok": "each_item"}, }, - "routes": {"each_item": {"loop": "__end__", "done": "__end__"}}, } ) diff --git a/tests/core/test_foreach_control_regions.py b/tests/core/test_foreach_control_regions.py index b3844f2f..e81110e8 100644 --- a/tests/core/test_foreach_control_regions.py +++ b/tests/core/test_foreach_control_regions.py @@ -6,6 +6,7 @@ from wf_core.analysis.control_regions import ( ControlRegionIssueKind, analyze_control_regions, ) +from wf_core.validation.issues import ValidationIssueCode def _workflow( @@ -56,6 +57,21 @@ def _condition(node_id: str) -> dict[str, object]: } +_CONTROL_REGION_CODES = {code.value for code in ControlRegionIssueKind} + + +def _public_control_errors(workflow: Workflow) -> list[tuple[str, str]]: + return [ + (issue.code.value, issue.path) + for issue in workflow.validate_structure().errors + if issue.code.value in _CONTROL_REGION_CODES + ] + + +def _assert_no_public_control_errors(workflow: Workflow) -> None: + assert _public_control_errors(workflow) == [] + + def test_closed_root_cycle_has_one_empty_control_region() -> None: workflow = _workflow( start="a", @@ -70,6 +86,7 @@ def test_closed_root_cycle_has_one_empty_control_region() -> None: assert analysis.issues == () assert analysis.owner_stack_by_node == {"a": (), "b": ()} + _assert_no_public_control_errors(workflow) def test_foreach_cycle_with_possible_return_is_valid() -> None: @@ -89,6 +106,7 @@ def test_foreach_cycle_with_possible_return_is_valid() -> None: assert analysis.issues == () assert analysis.owner_stack_by_node["a"] == ("f",) assert analysis.owner_stack_by_node["f"] == () + _assert_no_public_control_errors(workflow) def test_conditional_foreach_paths_can_both_return() -> None: @@ -109,6 +127,7 @@ def test_conditional_foreach_paths_can_both_return() -> None: assert analysis.issues == () assert analysis.owner_stack_by_node["condition"] == ("f",) assert analysis.owner_stack_by_node["work"] == ("f",) + _assert_no_public_control_errors(workflow) def test_nested_foreach_assigns_static_owner_stacks() -> None: @@ -142,6 +161,7 @@ def test_nested_foreach_assigns_static_owner_stacks() -> None: "after": (), } assert analysis.issues == () + _assert_no_public_control_errors(workflow) def test_reentering_completed_foreach_keeps_one_static_region() -> None: @@ -163,6 +183,7 @@ def test_reentering_completed_foreach_keeps_one_static_region() -> None: assert analysis.owner_stack_by_node["f"] == () assert analysis.owner_stack_by_node["work"] == ("f",) assert analysis.owner_stack_by_node["again"] == () + _assert_no_public_control_errors(workflow) def test_external_entry_into_foreach_body_is_region_conflict() -> None: @@ -184,6 +205,12 @@ def test_external_entry_into_foreach_body_is_region_conflict() -> None: (issue.kind, issue.path) for issue in analysis.issues ] assert "b" not in analysis.owner_stack_by_node + matching = [ + issue + for issue in workflow.validate_structure().errors + if issue.code == ValidationIssueCode.FOREACH_REGION_CONFLICT + ] + assert matching[0].path == "nodes[b]" def test_foreach_body_escape_is_region_conflict() -> None: @@ -204,6 +231,12 @@ def test_foreach_body_escape_is_region_conflict() -> None: (issue.kind, issue.path) for issue in analysis.issues ] assert "after" not in analysis.owner_stack_by_node + matching = [ + issue + for issue in workflow.validate_structure().errors + if issue.code == ValidationIssueCode.FOREACH_REGION_CONFLICT + ] + assert matching[0].path == "nodes[after]" def test_skipping_inner_foreach_owner_is_invalid_return() -> None: @@ -224,6 +257,12 @@ def test_skipping_inner_foreach_owner_is_invalid_return() -> None: assert (ControlRegionIssueKind.INVALID_FOREACH_RETURN, "edges[2]") 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[2]" def test_entering_sibling_foreach_body_is_region_conflict() -> None: @@ -245,6 +284,12 @@ def test_entering_sibling_foreach_body_is_region_conflict() -> None: assert (ControlRegionIssueKind.FOREACH_REGION_CONFLICT, "nodes[b2]") in [ (issue.kind, issue.path) for issue in analysis.issues ] + matching = [ + issue + for issue in workflow.validate_structure().errors + if issue.code == ValidationIssueCode.FOREACH_REGION_CONFLICT + ] + assert matching[0].path == "nodes[b2]" def test_empty_foreach_body_is_rejected() -> None: @@ -262,6 +307,12 @@ def test_empty_foreach_body_is_rejected() -> None: assert (ControlRegionIssueKind.EMPTY_FOREACH_BODY, "edges[0]") in [ (issue.kind, issue.path) for issue in analysis.issues ] + matching = [ + issue + for issue in workflow.validate_structure().errors + if issue.code == ValidationIssueCode.EMPTY_FOREACH_BODY + ] + assert matching[0].path == "edges[0]" def test_closed_foreach_body_cycle_has_no_return() -> None: @@ -281,6 +332,12 @@ def test_closed_foreach_body_cycle_has_no_return() -> None: assert (ControlRegionIssueKind.FOREACH_BODY_NO_RETURN, "nodes[f]") in [ (issue.kind, issue.path) for issue in analysis.issues ] + matching = [ + issue + for issue in workflow.validate_structure().errors + if issue.code == ValidationIssueCode.FOREACH_BODY_NO_RETURN + ] + assert matching[0].path == "nodes[f]" def test_foreach_body_cannot_target_end_token() -> None: @@ -299,6 +356,12 @@ def test_foreach_body_cannot_target_end_token() -> None: assert (ControlRegionIssueKind.INVALID_FOREACH_TERMINAL, "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_TERMINAL + ] + assert matching[0].path == "edges[1]" def test_foreach_body_cannot_target_explicit_end_node() -> None: @@ -321,6 +384,12 @@ def test_foreach_body_cannot_target_explicit_end_node() -> None: assert (ControlRegionIssueKind.INVALID_FOREACH_TERMINAL, "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_TERMINAL + ] + assert matching[0].path == "edges[1]" def test_every_unreachable_node_is_reported() -> None: @@ -345,3 +414,8 @@ def test_every_unreachable_node_is_reported() -> None: ControlRegionIssueKind.UNREACHABLE_NODE, "nodes[detached_b]", ) in by_kind_path + public_by_code_path = [ + (issue.code.value, issue.path) for issue in workflow.validate_structure().errors + ] + assert ("unreachable_node", "nodes[detached_a]") in public_by_code_path + assert ("unreachable_node", "nodes[detached_b]") in public_by_code_path diff --git a/tests/core/test_foreach_policy.py b/tests/core/test_foreach_policy.py index ff752612..ac396605 100644 --- a/tests/core/test_foreach_policy.py +++ b/tests/core/test_foreach_policy.py @@ -137,16 +137,19 @@ def test_collect_policy_requires_completed_with_errors_edge() -> None: workflow = _workflow( item_error={"action": "collect", "collect_to": "state.item_errors"}, edges=[ - {"from": "each", "outcome": "loop", "to": END}, + {"from": "each", "outcome": "loop", "to": "body"}, + {"from": "body", "outcome": "ok", "to": "each"}, {"from": "each", "outcome": "done", "to": END}, ], ) report = validate_workflow(workflow) - assert report.errors - assert report.errors[0].code == "missing_outcome_edge" - assert "completed_with_errors" in report.errors[0].message + matching = [ + issue for issue in report.errors if issue.code == "missing_outcome_edge" + ] + assert matching + assert "completed_with_errors" in matching[0].message def test_collect_policy_destination_must_be_declared_array_field() -> None: @@ -199,11 +202,13 @@ def _workflow( "over": "state.items", "as": "item", "item_error": item_error or {"action": "fail"}, - } + }, + {"id": "body", "type": "node", "node": "noop"}, ], "edges": edges or [ - {"from": "each", "outcome": "loop", "to": END}, + {"from": "each", "outcome": "loop", "to": "body"}, + {"from": "body", "outcome": "ok", "to": "each"}, {"from": "each", "outcome": "done", "to": END}, ], "node_defs": [],