From 71dc5ba8e8763cc71a319e338a8d599bf52e64f2 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 21:42:38 +0700 Subject: [PATCH] feat: validate structured context paths --- src/wf_core/validation/context_paths.py | 306 ++++++++++++++ src/wf_core/validation/core.py | 17 +- src/wf_core/validation/issues.py | 2 + .../test_structured_context_validation.py | 395 ++++++++++++++++++ 4 files changed, 717 insertions(+), 3 deletions(-) create mode 100644 src/wf_core/validation/context_paths.py create mode 100644 tests/core/test_structured_context_validation.py diff --git a/src/wf_core/validation/context_paths.py b/src/wf_core/validation/context_paths.py new file mode 100644 index 00000000..2d7a8892 --- /dev/null +++ b/src/wf_core/validation/context_paths.py @@ -0,0 +1,306 @@ +from __future__ import annotations + +from collections.abc import Iterator, Mapping +from typing import Any + +from wf_core.analysis.context_scopes import ContextSchema, root_context_schema +from wf_core.analysis.control_regions import ControlRegionAnalysis +from wf_core.context_contracts import RESERVED_CONTEXT_KEYS +from wf_core.models.conditions import ( + BinaryCondition, + Condition, + ExistsCondition, + LiteralOperand, + NotCondition, + PathOperand, + VariadicCondition, +) +from wf_core.models.input_bindings import ( + ArrayExpression, + InputExpression, + InputExpressionBinding, + InputPathBinding, + LiteralExpression, + ObjectExpression, + PathExpression, +) +from wf_core.models.steps import ( + ConditionNode, + ForeachNode, + InterruptNode, + NodeUse, + SubgraphNode, +) +from wf_core.models.workflow import Workflow +from wf_core.paths import GraphSourcePath +from wf_core.validation.issues import ValidationIssueCode, ValidationReport + + +def validate_context_paths( + workflow: Workflow, + *, + context_schemas: Mapping[str, ContextSchema], + report: ValidationReport, +) -> None: + """Validate every ``context.*`` path against its consuming location schema. + + Ordinary input/state validation stays where it is; this pass owns the + stronger program-location-aware meaning of ``context.*``. A path is valid + only if every literal segment is a declared object property in the + consuming node's generated schema. The whole ``context`` object and the + ``context.foreach`` map remain readable; unknown dynamic keys do not. + """ + nodes_by_index = list(workflow.nodes) + node_index_by_id = {node.id: idx for idx, node in enumerate(nodes_by_index)} + _validate_alias_ownership(workflow, node_index_by_id, report) + for idx, node in enumerate(nodes_by_index): + schema = context_schemas.get(node.id) + if isinstance(node, NodeUse): + _validate_step_input_bindings( + node.input, f"nodes[{idx}].input", node.id, schema, report + ) + elif isinstance(node, SubgraphNode): + _validate_step_input_bindings( + node.input, f"nodes[{idx}].input", node.id, schema, report + ) + elif isinstance(node, ConditionNode): + for location, path in _condition_paths( + node.check, f"nodes[{idx}].check" + ): + _validate_one_context_path( + path, location, node.id, schema, report + ) + elif isinstance(node, ForeachNode): + # Context-rooted `over` paths reach this pass; the old + # input/state-only check stays permissive for them. + if node.over.root == "context": + _validate_one_context_path( + node.over, f"nodes[{idx}].over", node.id, schema, report + ) + elif isinstance(node, InterruptNode): + _validate_step_input_bindings( + node.request, f"nodes[{idx}].request", node.id, schema, report + ) + _validate_workflow_output(workflow, report) + + +def _validate_step_input_bindings( + bindings: list[Any], + base: str, + node_id: str, + schema: ContextSchema | None, + report: ValidationReport, +) -> None: + """Validate context paths in one input/request binding list. + + `base` is the list location such as `nodes[3].input` or + `nodes[1].request`; each binding contributes `base[i]` and each path + field contributes a further suffix like `.path` or + `.expression.items[0].path`. + """ + for binding_index, binding in enumerate(bindings): + binding_location = f"{base}[{binding_index}]" + if isinstance(binding, InputPathBinding): + if binding.path.root == "context": + _validate_one_context_path( + binding.path, f"{binding_location}.path", node_id, schema, report + ) + elif isinstance(binding, InputExpressionBinding): + for location, path in _expression_paths( + binding.expression, f"{binding_location}.expression" + ): + if path.root == "context": + _validate_one_context_path( + path, location, node_id, schema, report + ) + + +def _expression_paths( + expression: InputExpression, + location: str, +) -> Iterator[tuple[str, GraphSourcePath]]: + """Yield ``(model path, graph path)`` for every path leaf in an expression. + + Finite recursion mirrors the input-expression model: paths, arrays, and + objects. Literals contribute no paths. + """ + match expression: + case PathExpression(path=path): + yield location + ".path", path + case ArrayExpression(items=items): + for index, item in enumerate(items): + yield from _expression_paths(item, f"{location}.items[{index}]") + case ObjectExpression(fields=fields): + for name, item in fields.items(): + yield from _expression_paths(item, f"{location}.fields.{name}") + case LiteralExpression(): + return + + +def _condition_paths( + condition: Condition, + location: str, +) -> Iterator[tuple[str, GraphSourcePath]]: + """Yield context-candidate paths from a condition tree with model locations.""" + if isinstance(condition, ExistsCondition): + yield location + ".path", condition.path + return + if isinstance(condition, NotCondition): + yield from _condition_paths(condition.arg, f"{location}.arg") + return + if isinstance(condition, VariadicCondition): + for index, arg in enumerate(condition.args): + yield from _condition_paths(arg, f"{location}.args[{index}]") + return + if isinstance(condition, BinaryCondition): + yield from _operand_paths(condition.left, f"{location}.left") + yield from _operand_paths(condition.right, f"{location}.right") + return + + +def _operand_paths( + operand: PathOperand | LiteralOperand, location: str +) -> Iterator[tuple[str, GraphSourcePath]]: + if isinstance(operand, LiteralOperand): + return + yield location + ".path", operand.path + + +def _validate_one_context_path( + path: GraphSourcePath, + location: str, + node_id: str | None, + schema: ContextSchema | None, + report: ValidationReport, +) -> None: + if path.root != "context": + return + if schema is None: + report.add( + ValidationIssueCode.INVALID_CONTEXT_PATH, + location, + f"invalid context path {str(path)!r} at {node_id or location!r}: " + "no context schema for this program location", + ) + return + if not _path_in_schema(schema, path.parts): + report.add( + ValidationIssueCode.INVALID_CONTEXT_PATH, + location, + f"invalid context path {str(path)!r} at {node_id or location!r}", + ) + + +def _path_in_schema(schema: Mapping[str, Any], parts: tuple[str, ...]) -> bool: + """Return whether literal parts walk declared object properties. + + The whole ``context`` object (no parts) and the ``foreach`` map itself + are readable. Unknown segments fail closed, except beneath an + unconstrained item schema (``{}`` or an open object with no declared + properties): a generic ``array`` collection infers ``{}``, and runtime + values may carry fields the static schema cannot see, so such subpaths + stay permissive. Closed maps with ``additionalProperties: False`` (like + the ``foreach`` map itself) still reject unknown keys. + """ + if not parts: + return True + current: Any = schema + for part in parts: + if not isinstance(current, Mapping): + return False + # Resolve local $ref if present (defensive; generated schemas are inline). + while isinstance(current.get("$ref"), str): + # Generated context schemas keep entry schemas inline, so an + # unresolvable ref here means the path cannot be proven valid. + return False + properties = current.get("properties") + if not isinstance(properties, Mapping): + # No declared properties: unconstrained `{}` or an open object + # allows subpaths; scalar or closed schemas do not. + if current == {}: + return True + if current.get("type") == "object" and current.get( + "additionalProperties", True + ) is not False: + return True + return False + if part not in properties: + return False + current = properties[part] + return True + + +def _validate_workflow_output(workflow: Workflow, report: ValidationReport) -> None: + schema = root_context_schema() + for output_index, binding in enumerate(workflow.output): + if isinstance(binding, InputPathBinding): + if binding.path.root == "context": + _validate_one_context_path( + binding.path, + f"output[{output_index}].path", + "workflow_output", + schema, + report, + ) + elif isinstance(binding, InputExpressionBinding): + for location, path in _expression_paths( + binding.expression, f"output[{output_index}].expression" + ): + if path.root == "context": + _validate_one_context_path( + path, location, "workflow_output", schema, report + ) + + +def _validate_alias_ownership( + workflow: Workflow, + node_index_by_id: dict[str, int], + report: ValidationReport, +) -> None: + """Reject reserved or colliding active foreach aliases. + + Reserved names are every standard context field plus ``foreach``, + ``loop_item``, and ``loop_index`` (that is, ``RESERVED_CONTEXT_KEYS``). + Siblings in separate control regions may reuse an alias because they are + never active together; only aliases active in the same owner stack + collide. Failures point at the inner foreach's ``as`` field. + """ + from wf_core.analysis.control_regions import analyze_control_regions + + analysis: ControlRegionAnalysis = analyze_control_regions(workflow) + foreach_by_id = { + node.id: node for node in workflow.nodes if isinstance(node, ForeachNode) + } + reported: set[str] = set() + for stack in analysis.owner_stack_by_node.values(): + seen_aliases: dict[str, str] = {} + for owner_id in stack: + foreach = foreach_by_id.get(owner_id) + if foreach is None: + continue + alias = foreach.as_ + idx = node_index_by_id.get(owner_id) + location = ( + f"nodes[{idx}].as" if idx is not None else f"nodes[{owner_id}]" + ) + if not alias or alias in RESERVED_CONTEXT_KEYS: + if owner_id not in reported: + reported.add(owner_id) + report.add( + ValidationIssueCode.FOREACH_CONTEXT_ALIAS_CONFLICT, + location, + f"foreach alias {alias!r} for node {owner_id!r} " + "collides with reserved context keys", + ) + continue + if alias in seen_aliases: + if owner_id not in reported: + reported.add(owner_id) + report.add( + ValidationIssueCode.FOREACH_CONTEXT_ALIAS_CONFLICT, + location, + f"foreach alias {alias!r} for node {owner_id!r} " + f"collides with active alias from {seen_aliases[alias]!r}", + ) + continue + seen_aliases[alias] = owner_id diff --git a/src/wf_core/validation/core.py b/src/wf_core/validation/core.py index 2dfdce88..9685bad7 100644 --- a/src/wf_core/validation/core.py +++ b/src/wf_core/validation/core.py @@ -28,10 +28,14 @@ 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. + once afterwards and its diagnostics are translated verbatim. The same + analysis feeds structured context-schema construction and the + location-aware context-path pass, so no second traversal hides inside + context helpers. No second graph traversal lives inside validation. """ + from wf_core.analysis.context_scopes import context_schemas_by_node from wf_core.analysis.control_regions import analyze_control_regions + from wf_core.validation.context_paths import validate_context_paths report = ValidationReport() @@ -41,12 +45,19 @@ 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: + analysis = analyze_control_regions(workflow) + for issue in analysis.issues: report.add( ValidationIssueCode(issue.kind.value), issue.path, issue.message, ) + context_schemas = context_schemas_by_node( + workflow, control_regions=analysis + ) + validate_context_paths( + workflow, context_schemas=context_schemas, report=report + ) return report diff --git a/src/wf_core/validation/issues.py b/src/wf_core/validation/issues.py index c94afffc..4e5ccc1e 100644 --- a/src/wf_core/validation/issues.py +++ b/src/wf_core/validation/issues.py @@ -32,6 +32,8 @@ class ValidationIssueCode(StrEnum): INVALID_FOREACH_TERMINAL = "invalid_foreach_terminal" EMPTY_FOREACH_BODY = "empty_foreach_body" FOREACH_BODY_NO_RETURN = "foreach_body_no_return" + INVALID_CONTEXT_PATH = "invalid_context_path" + FOREACH_CONTEXT_ALIAS_CONFLICT = "foreach_context_alias_conflict" @dataclass(slots=True) diff --git a/tests/core/test_structured_context_validation.py b/tests/core/test_structured_context_validation.py new file mode 100644 index 00000000..5769f3af --- /dev/null +++ b/tests/core/test_structured_context_validation.py @@ -0,0 +1,395 @@ +from __future__ import annotations + +import pytest + +from wf_core import END, Edge, ForeachNode, NodeDef, NodeUse, SchemaRef, Workflow +from wf_core.models.schemas import StateField, StateSchema +from wf_core.validation.issues import ValidationIssueCode + + +def _foreach(node_id: str, *, over: str = "state.items", alias: str) -> ForeachNode: + return ForeachNode.model_validate( + {"id": node_id, "type": "foreach", "over": over, "as": alias} + ) + + +def _node_use( + node_id: str, *, path: str | None = None, expression: dict | None = None +) -> NodeUse: + if expression is not None: + binding = {"target": "value", "expression": expression} + elif path is not None: + binding = {"target": "value", "path": path} + else: + binding = {"target": "value", "path": "state.items"} + return NodeUse.model_validate( + {"id": node_id, "type": "node", "node": "record", "input": [binding]} + ) + + +def _record_def() -> NodeDef: + return NodeDef( + name="record", + input_schema=SchemaRef(type="object", properties={"value": {}}), + output_schema=SchemaRef(type="object", properties={}), + outcomes=["ok"], + ) + + +def _base_workflow(*, work_path: str = "context.foreach.orders.item") -> Workflow: + return Workflow( + name="validation_structured", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "items": StateField(type="array"), + "orders_list": StateField(type="array"), + } + ), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[_record_def()], + start="customers", + nodes=[ + _foreach("customers", over="state.items", alias="customer"), + _foreach("orders", over="state.orders_list", alias="order"), + _node_use("work", path=work_path), + NodeUse.model_validate( + {"id": "after_inner", "type": "node", "node": "record"} + ), + ], + edges=[ + Edge.model_validate( + {"from": "customers", "outcome": "loop", "to": "orders"} + ), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "orders"}), + Edge.model_validate( + {"from": "orders", "outcome": "done", "to": "after_inner"} + ), + Edge.model_validate( + {"from": "after_inner", "outcome": "ok", "to": "customers"} + ), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ], + ) + + +def _issue(report, code: ValidationIssueCode, path: str): + return next( + (issue for issue in report.errors if issue.code == code and issue.path == path), + None, + ) + + +def test_active_structured_foreach_item_path_is_valid() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow(work_path="context.foreach.orders.item") + report = validate_workflow(workflow) + assert _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].input[0].path", + ) is None + + +def test_nested_body_can_read_outer_and_inner_entries() -> None: + from wf_core.validation import validate_workflow + + for path in ( + "context.foreach.customers.item", + "context.foreach.orders.item", + "context.foreach.customers.index", + "context.foreach.orders.index", + ): + workflow = _base_workflow(work_path=path) + report = validate_workflow(workflow) + assert _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path" + ) is None, path + + +def test_inactive_foreach_entry_is_rejected() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow() + workflow.nodes[3] = _node_use( + "after_inner", path="context.foreach.orders.item" + ) + report = validate_workflow(workflow) + issue = _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[3].input[0].path" + ) + assert issue is not None + assert "context.foreach.orders.item" in issue.message + assert "after_inner" in issue.message + + +def test_missing_foreach_id_is_rejected() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow(work_path="context.foreach.missing.item") + report = validate_workflow(workflow) + issue = _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path" + ) + assert issue is not None + assert "context.foreach.missing.item" in issue.message + assert "work" in issue.message + + +def test_unknown_foreach_entry_field_is_rejected() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow(work_path="context.foreach.orders.bogus") + report = validate_workflow(workflow) + issue = _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path" + ) + assert issue is not None + + +def test_unreachable_node_does_not_receive_root_context_fallback() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow() + workflow.nodes.append(_node_use("ghost", path="context.foreach.orders.item")) + report = validate_workflow(workflow) + issue = _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[4].input[0].path" + ) + assert issue is not None + + +def test_workflow_output_cannot_read_completed_foreach_entry() -> None: + from wf_core.models.steps import InputPathBinding + from wf_core.validation import validate_workflow + + workflow = _base_workflow() + workflow.output = [ + InputPathBinding.model_validate( + {"target": "result", "path": "context.foreach.orders.item"} + ) + ] + report = validate_workflow(workflow) + issue = _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "output[0].path" + ) + assert issue is not None + + +@pytest.mark.parametrize( + ("make_node", "expected_path"), + [ + ( + lambda: _node_use("work", path="context.foreach.missing.item"), + "nodes[2].input[0].path", + ), + ( + lambda: _node_use( + "work", + expression={"kind": "path", "path": "context.foreach.missing.item"}, + ), + "nodes[2].input[0].expression.path", + ), + ( + lambda: _node_use( + "work", + expression={ + "kind": "array", + "items": [ + {"kind": "path", "path": "context.foreach.missing.item"} + ], + }, + ), + "nodes[2].input[0].expression.items[0].path", + ), + ], +) +def test_node_input_surfaces_report_exact_model_paths(make_node, expected_path) -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow() + workflow.nodes[2] = make_node() + report = validate_workflow(workflow) + assert ( + _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, expected_path) + is not None + ) + + +def test_all_model_surfaces_reject_missing_foreach_id() -> None: + from wf_core.models.steps import ConditionNode, InterruptNode, SubgraphNode + from wf_core.validation import validate_workflow + + bad = "context.foreach.missing.item" + # Subgraph input + workflow = _base_workflow() + workflow.nodes[2] = SubgraphNode.model_validate( + { + "id": "work", + "type": "subgraph", + "workflow": {"name": "child"}, + "input": [{"target": "order", "path": bad}], + } + ) + report = validate_workflow(workflow) + assert ( + _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path") + is not None + ) + + # Condition check + workflow = _base_workflow() + workflow.nodes[2] = ConditionNode.model_validate( + {"id": "work", "type": "condition", "check": {"op": "exists", "path": bad}} + ) + workflow.edges = [ + Edge.model_validate({"from": "customers", "outcome": "loop", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "true", "to": "orders"}), + Edge.model_validate({"from": "work", "outcome": "false", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "done", "to": "after_inner"}), + Edge.model_validate({"from": "after_inner", "outcome": "ok", "to": "customers"}), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ] + report = validate_workflow(workflow) + assert ( + _issue( + report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].check.path" + ) + is not None + ) + + # Foreach over + workflow = _base_workflow() + foreach = workflow.nodes[1] + assert isinstance(foreach, ForeachNode) + workflow.nodes[1] = ForeachNode.model_validate( + {"id": "orders", "type": "foreach", "over": bad, "as": "order"} + ) + report = validate_workflow(workflow) + assert ( + _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[1].over") + is not None + ) + + # Interrupt request + workflow = _base_workflow() + workflow.nodes[2] = InterruptNode.model_validate( + { + "id": "work", + "type": "interrupt", + "kind": "approval", + "request": [{"target": "order", "path": bad}], + } + ) + workflow.edges = [ + Edge.model_validate({"from": "customers", "outcome": "loop", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "submitted", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "done", "to": "after_inner"}), + Edge.model_validate({"from": "after_inner", "outcome": "ok", "to": "customers"}), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ] + report = validate_workflow(workflow) + assert ( + _issue( + report, + ValidationIssueCode.INVALID_CONTEXT_PATH, + "nodes[2].request[0].path", + ) + is not None + ) + + # Workflow output + from wf_core.models.steps import InputPathBinding as _IPB + + workflow = _base_workflow() + workflow.output = [ + _IPB.model_validate({"target": "result", "path": bad}) + ] + report = validate_workflow(workflow) + assert ( + _issue(report, ValidationIssueCode.INVALID_CONTEXT_PATH, "output[0].path") + is not None + ) + + +def test_foreach_alias_cannot_use_reserved_context_name() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow() + workflow.nodes[1] = _foreach("orders", over="state.orders_list", alias="loop_item") + report = validate_workflow(workflow) + issue = next( + ( + issue + for issue in report.errors + if issue.code == ValidationIssueCode.FOREACH_CONTEXT_ALIAS_CONFLICT + ), + None, + ) + assert issue is not None + assert issue.path == "nodes[1].as" + + +def test_nested_active_foreach_aliases_must_be_unique() -> None: + from wf_core.validation import validate_workflow + + workflow = _base_workflow() + workflow.nodes[0] = _foreach("customers", over="state.items", alias="same") + workflow.nodes[1] = _foreach("orders", over="state.orders_list", alias="same") + report = validate_workflow(workflow) + issue = next( + ( + issue + for issue in report.errors + if issue.code == ValidationIssueCode.FOREACH_CONTEXT_ALIAS_CONFLICT + ), + None, + ) + assert issue is not None + assert issue.path == "nodes[1].as" + + +def test_sibling_foreach_aliases_may_match_when_never_active_together() -> None: + from wf_core.validation import validate_workflow + + workflow = Workflow( + name="siblings", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "items": StateField(type="array"), + "other": StateField(type="array"), + } + ), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[_record_def()], + start="start", + nodes=[ + NodeUse(id="start", type="node", node="record"), + _foreach("left", over="state.items", alias="same"), + _foreach("right", over="state.other", alias="same"), + NodeUse(id="left_body", type="node", node="record"), + NodeUse(id="right_body", type="node", node="record"), + NodeUse(id="join", type="node", node="record"), + ], + edges=[ + Edge.model_validate({"from": "start", "outcome": "ok", "to": "left"}), + Edge.model_validate({"from": "left", "outcome": "loop", "to": "left_body"}), + Edge.model_validate({"from": "left_body", "outcome": "ok", "to": "left"}), + Edge.model_validate({"from": "left", "outcome": "done", "to": "right"}), + Edge.model_validate({"from": "right", "outcome": "loop", "to": "right_body"}), + Edge.model_validate({"from": "right_body", "outcome": "ok", "to": "right"}), + Edge.model_validate({"from": "right", "outcome": "done", "to": "join"}), + Edge.model_validate({"from": "join", "outcome": "ok", "to": END}), + ], + ) + report = validate_workflow(workflow) + assert not [ + issue + for issue in report.errors + if issue.code == ValidationIssueCode.FOREACH_CONTEXT_ALIAS_CONFLICT + ]