diff --git a/src/orchestrator/compiler/control_flow_compiler.py b/src/orchestrator/compiler/control_flow_compiler.py index 99ab67c..af138aa 100644 --- a/src/orchestrator/compiler/control_flow_compiler.py +++ b/src/orchestrator/compiler/control_flow_compiler.py @@ -418,16 +418,24 @@ def _task_to_step_def(self, task: Task) -> Dict[str, Any]: return step_def - def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> Task: - """Build Task object with control flow support. - - Args: - task_def: Task definition - available_steps: List of all step IDs in the pipeline - - Returns: - Task object + def _build_task(self, task_def: Dict[str, Any], available_steps: List[str], + graph=None) -> Task: + """Build a control-flow task, ordered by the canonical graph. + + This method has many exits -- conditional, goto, for-each, action-loop + -- and each builds its task through a different handler that reads + `dependencies` straight off the definition. Applying the graph once, + here, is what stops a control-flow task from being the one kind of + step that silently keeps the old behaviour (#465). """ + task = self._build_control_flow_task(task_def, available_steps, graph) + if graph is not None: + task.dependencies = graph.dependencies_for(task.id) + return task + + def _build_control_flow_task(self, task_def: Dict[str, Any], + available_steps: List[str], graph=None) -> Task: + """Build the task itself; `_build_task` applies the ordering.""" # First, ensure metadata is properly built # This is important for conditional tasks that also have tools if "metadata" not in task_def: @@ -444,7 +452,7 @@ def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> T condition = task_def.pop("condition", None) or task_def.pop("if", None) # First, let the parent build the base task with template analysis - base_task = super()._build_task(task_def, available_steps) + base_task = super()._build_task(task_def, available_steps, graph) # Restore condition to task_def if condition: diff --git a/src/orchestrator/compiler/yaml_compiler.py b/src/orchestrator/compiler/yaml_compiler.py index de8efc3..c65cced 100644 --- a/src/orchestrator/compiler/yaml_compiler.py +++ b/src/orchestrator/compiler/yaml_compiler.py @@ -8,14 +8,15 @@ from typing import Any, Dict, List, Optional import yaml -from jinja2 import StrictUndefined from ..core.template_sandbox import create_pipeline_environment from ..core.actions import canonical_action from ..core.pipeline import Pipeline from ..core.task import Task +from ..core.dependency_graph import DependencyGraph, build_dependency_graph from ..core.template_metadata import TemplateMetadata +from ..core.template_scope import template_references from ..core.exceptions import YAMLCompilerError from ..core.error_handling import ErrorHandler from ..core.file_inclusion import FileInclusionProcessor, FileInclusionError @@ -1201,9 +1202,53 @@ def _build_pipeline(self, pipeline_def: Dict[str, Any], compile_context: Dict[st if "id" in step_def: available_steps.append(step_def["id"]) + # The one graph: explicit `dependencies:`, template references and + # control-flow references together. It is what cycle validation + # inspects and what every Task.dependencies is built from, so the + # schedule cannot disagree with what was validated (#465). + graph = build_dependency_graph(pipeline_def) + + cycles = graph.cycles() + if cycles: + described = "; ".join(" -> ".join(cycle) for cycle in cycles) + raise YAMLCompilerError( + f"Dependency cycle detected: {described}. No step in a cycle can " + f"ever become ready, so the pipeline would deadlock rather than " + f"fail. A cycle can come from a template reference as well as " + f"from an explicit `dependencies:` entry." + ) + + # Ordering that is implied but not written down. Reported, never + # enforced: an author who prefers explicit graphs gets told exactly + # which lines to add, and one who is happy with inference is not + # nagged into failure. + if self.validation_report is not None: + for edge in graph.inferred_only(): + self.validation_report.add_issue(ValidationIssue( + severity=ValidationSeverity.INFO, + category="dependency", + component=edge.task, + message=( + f"Step '{edge.task}' depends on '{edge.depends_on}' because " + f"{edge.location} refers to it. The ordering is applied; add " + f"'{edge.depends_on}' to its dependencies to state it." + ), + code="implicit_dependency", + path=f"{edge.task}.{edge.location}", + suggestions=[ + f"dependencies:\n - {edge.depends_on}" + ], + metadata={ + "step": edge.task, + "referenced_step": edge.depends_on, + "parameter_path": edge.location, + "origin": edge.origin, + }, + )) + # Second pass: build tasks with template analysis for step_def in steps: - task = self._build_task(step_def, available_steps) + task = self._build_task(step_def, available_steps, graph) pipeline.add_task(task) return pipeline @@ -1224,13 +1269,16 @@ def _analyze_template(self, template_str: str, available_steps: List[str], dependencies = set() context_requirements = set() - # Extract step references (e.g., step_id.result, step_id.outputs.data) + # Step references, from the same extractor the dependency graph and + # the data-flow validator use. This was a third regex, and it matched + # `step_id.` anywhere in the raw text -- including inside a string + # literal, a comment, or a loop that rebinds the name. if available_steps: - # Build pattern to match step references - # Match: step_id.property, step_id.nested.property, etc. - step_pattern = r'\b(' + '|'.join(re.escape(step) for step in available_steps) + r')\.' - for match in re.finditer(step_pattern, template_str): - dependencies.add(match.group(1)) + known = set(available_steps) + for reference in template_references(template_str, self.template_engine): + base = reference.split(".", 1)[0] + if base in known: + dependencies.add(base) # Extract loop variables loop_vars = ['$item', '$index', '$is_first', '$is_last', '$iteration', '$loop'] @@ -1292,7 +1340,8 @@ def analyze_value(value: Any, path: str) -> None: analyze_value(params, path_prefix) return template_metadata - def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> Task: + def _build_task(self, task_def: Dict[str, Any], available_steps: List[str], + graph: Optional[DependencyGraph] = None) -> Task: """ Build Task object from definition with template analysis. @@ -1328,15 +1377,21 @@ def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> T parameters = task_def.get("parameters", {}) - # Handle dependencies which may be string or array - # Support both 'dependencies' and 'depends_on' for backward compatibility - dependencies = task_def.get("dependencies", task_def.get("depends_on", [])) - if isinstance(dependencies, str): - # Handle single dependency as string or comma-separated list - if "," in dependencies: - dependencies = [dep.strip() for dep in dependencies.split(",")] - else: - dependencies = [dependencies.strip()] if dependencies.strip() else [] + # Dependencies come from the canonical graph, which already merged the + # explicit `dependencies:` key with the references found in templates + # and control-flow expressions. Appending inferred edges here instead + # would build a schedule the cycle check never inspected -- which is + # the shape of the bug this replaced (#465). + if graph is not None: + dependencies = graph.dependencies_for(task_id) + else: + # Building one task in isolation: fall back to the declared key. + dependencies = task_def.get("dependencies", task_def.get("depends_on", [])) + if isinstance(dependencies, str): + if "," in dependencies: + dependencies = [dep.strip() for dep in dependencies.split(",")] + else: + dependencies = [dependencies.strip()] if dependencies.strip() else [] timeout = task_def.get("timeout") max_retries = task_def.get("max_retries", 3) @@ -1477,7 +1532,6 @@ def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> T # Set output metadata if provided if produces or location or format_type: - from ..core.output_metadata import create_output_metadata task.set_output_metadata( produces=produces, location=location, diff --git a/src/orchestrator/core/dependency_graph.py b/src/orchestrator/core/dependency_graph.py new file mode 100644 index 0000000..52438c8 --- /dev/null +++ b/src/orchestrator/core/dependency_graph.py @@ -0,0 +1,273 @@ +"""One graph of what must run before what. + +`{{ make.path }}` in a step's parameters is a data dependency. It says so +plainly: the value cannot exist until `make` has run. Three separate parts of +the compiler agreed about that and none of them told the scheduler: + +* `DependencyValidator` inferred edges from `for_each`, `condition` and + `while` -- but never looked inside `parameters`, so the commonest reference + of all was invisible to the cycle check. +* `DataFlowValidator` built a `data_flow_graph` that *did* include parameter + references, then logged it and threw it away. +* `YAMLCompiler._analyze_template` found the same references with a third + regex and stored them on `Task.template_metadata`, where nothing scheduled + anything from them. + +Meanwhile `Task.dependencies` was read from the explicit `dependencies:` key +alone. So this validated and then failed at run time (#465):: + + - id: make + parameters: {path: "./a.txt", content: "A"} + - id: use + parameters: {content: "{{ make.path }}"} # no `dependencies:` + +`use` was scheduled beside `make`, `make`'s result was not in context, and the +render failed. Validation had promised the pipeline was fine. + +This module is the single answer. Explicit dependencies, template-derived +dependencies and control-flow dependencies go into one graph; that graph is +what gets validated for cycles, and the same graph is what `Task.dependencies` +is built from. A fourth mechanism would have recreated the original problem. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, Iterator, List, Set, Tuple + +from .template_scope import template_references + +#: Where an edge came from. Kept on the edge so a diagnostic can say *why* +#: two steps are ordered -- "you wrote it" reads very differently from "your +#: template implies it". +DECLARED = "declared" +TEMPLATE = "template" +CONTROL_FLOW = "control_flow" + +#: Step keys whose value is a condition or iterable rather than a parameter. +#: A reference in any of them orders the step just as a parameter does. +CONTROL_FLOW_KEYS: Tuple[str, ...] = ( + "for_each", "foreach", "condition", "if", "while", "until", +) + + +@dataclass(frozen=True) +class DependencyEdge: + """`task` cannot start until `depends_on` has finished.""" + + task: str + depends_on: str + origin: str + location: str + + @property + def is_inferred(self) -> bool: + return self.origin != DECLARED + + +@dataclass(frozen=True) +class DependencyGraph: + """Every ordering constraint in a pipeline, and where each came from.""" + + steps: Tuple[str, ...] + edges: Tuple[DependencyEdge, ...] + #: References a step makes to itself. Never edges -- a self-edge cannot be + #: scheduled -- but recorded so validation can report them. + self_references: Tuple[DependencyEdge, ...] = () + + def dependencies_for(self, task_id: str) -> List[str]: + """What must run before `task_id`, deduplicated, in declaration order. + + Declaration order rather than discovery order so that a pipeline + compiles to the same task graph however its templates are written. + """ + needed = {edge.depends_on for edge in self.edges if edge.task == task_id} + return [step for step in self.steps if step in needed] + + def adjacency(self) -> Dict[str, List[str]]: + return {step: self.dependencies_for(step) for step in self.steps} + + def origins_for(self, task_id: str, depends_on: str) -> Set[str]: + return { + edge.origin for edge in self.edges + if edge.task == task_id and edge.depends_on == depends_on + } + + def inferred_only(self) -> List[DependencyEdge]: + """Edges no `dependencies:` entry also states. + + The basis for the `implicit_dependency` lint: an author who wants the + ordering written down can be told exactly which lines to add, without + the omission being an error. + """ + declared = { + (edge.task, edge.depends_on) for edge in self.edges + if edge.origin == DECLARED + } + seen: Set[Tuple[str, str]] = set() + result = [] + for edge in self.edges: + key = (edge.task, edge.depends_on) + if edge.origin == DECLARED or key in declared or key in seen: + continue + seen.add(key) + result.append(edge) + return result + + def unknown_references(self) -> List[DependencyEdge]: + """Declared dependencies naming a step that does not exist. + + Only `declared` edges can be unknown: a template reference to a name + that is not a step is not a dependency at all -- it may be a pipeline + input, a loop variable or a typo, and deciding which is the data-flow + validator's job. + """ + known = set(self.steps) + return [edge for edge in self.edges if edge.depends_on not in known] + + def cycles(self) -> List[List[str]]: + """Every dependency cycle, each as a list of step ids. + + Reported before execution because a cycle is a deadlock: no step in it + can ever become ready. + """ + adjacency = self.adjacency() + found: List[List[str]] = [] + seen_signatures: Set[Tuple[str, ...]] = set() + visiting: List[str] = [] + state: Dict[str, int] = {} + + def visit(node: str) -> None: + state[node] = 1 + visiting.append(node) + for dependency in adjacency.get(node, []): + if dependency not in state: + visit(dependency) + elif state[dependency] == 1: + cycle = visiting[visiting.index(dependency):] + [dependency] + signature = tuple(sorted(set(cycle))) + if signature not in seen_signatures: + seen_signatures.add(signature) + found.append(cycle) + visiting.pop() + state[node] = 2 + + for step in self.steps: + if step not in state: + visit(step) + return found + + def levels(self) -> List[List[str]]: + """Steps grouped so that everything in a level may run concurrently. + + Raises `ValueError` on a cycle: producing a partial schedule for an + unschedulable pipeline is how a deadlock becomes a silent hang. + """ + adjacency = self.adjacency() + remaining = {step: set(deps) & set(self.steps) for step, deps in adjacency.items()} + done: Set[str] = set() + result: List[List[str]] = [] + + while remaining: + ready = [step for step, deps in remaining.items() if deps <= done] + if not ready: + raise ValueError( + f"dependency cycle among {sorted(remaining)}: no step is ready" + ) + ready = [step for step in self.steps if step in ready] + result.append(ready) + done.update(ready) + for step in ready: + del remaining[step] + return result + + +def _strings_in(value: Any, path: str = "") -> Iterator[Tuple[str, str]]: + """Every string inside a nested structure, with the path that reached it. + + Templates hide in nested dictionaries and lists as readily as in a plain + parameter, and a reference is a dependency wherever it appears. + """ + if isinstance(value, str): + yield value, path + elif isinstance(value, dict): + for key, item in value.items(): + yield from _strings_in(item, f"{path}.{key}" if path else str(key)) + elif isinstance(value, (list, tuple)): + for index, item in enumerate(value): + yield from _strings_in(item, f"{path}[{index}]") + + +def _declared_dependencies(step: Dict[str, Any]) -> List[str]: + """The `dependencies:` / `depends_on:` value, however it is written.""" + raw = step.get("dependencies", step.get("depends_on", [])) + if isinstance(raw, str): + return [part.strip() for part in raw.split(",") if part.strip()] + if isinstance(raw, (list, tuple)): + return [str(item).strip() for item in raw if str(item).strip()] + return [] + + +def build_dependency_graph( + pipeline_def: Dict[str, Any], env: Any = None +) -> DependencyGraph: + """The one graph, from a raw pipeline definition. + + Only names that are actually step ids become edges. A template reference + to anything else -- a pipeline input, a loop variable, a typo -- is not an + ordering constraint, and judging which of those it is belongs to the + data-flow validator. + """ + steps_def = pipeline_def.get("steps", pipeline_def.get("tasks", [])) or [] + step_ids: List[str] = [ + step["id"] for step in steps_def + if isinstance(step, dict) and step.get("id") + ] + known = set(step_ids) + + edges: List[DependencyEdge] = [] + self_references: List[DependencyEdge] = [] + + def add(task: str, depends_on: str, origin: str, location: str) -> None: + edge = DependencyEdge(task, depends_on, origin, location) + if depends_on == task: + # A self-edge can never be satisfied. Recorded, never scheduled. + self_references.append(edge) + return + edges.append(edge) + + for step in steps_def: + if not isinstance(step, dict) or not step.get("id"): + continue + task_id = step["id"] + + for dependency in _declared_dependencies(step): + add(task_id, dependency, DECLARED, "dependencies") + + for key in CONTROL_FLOW_KEYS: + if key not in step: + continue + for text, path in _strings_in(step[key], key): + for reference in template_references(text, env): + base = reference.split(".", 1)[0] + if base in known: + add(task_id, base, CONTROL_FLOW, path) + + # Everything else the step carries that may hold a template. `id`, + # `dependencies` and the control-flow keys above are excluded: the + # first two are not templates, and the third is already covered with + # a more precise origin. + for key, value in step.items(): + if key in ("id", "dependencies", "depends_on") or key in CONTROL_FLOW_KEYS: + continue + for text, path in _strings_in(value, key): + for reference in template_references(text, env): + base = reference.split(".", 1)[0] + if base in known: + add(task_id, base, TEMPLATE, path) + + return DependencyGraph( + steps=tuple(step_ids), + edges=tuple(edges), + self_references=tuple(self_references), + ) diff --git a/src/orchestrator/core/template_scope.py b/src/orchestrator/core/template_scope.py index a9f728e..085d7cd 100644 --- a/src/orchestrator/core/template_scope.py +++ b/src/orchestrator/core/template_scope.py @@ -39,7 +39,7 @@ from __future__ import annotations -from typing import Any, FrozenSet, MutableSet, Set +from typing import Any, FrozenSet, List, MutableSet, Optional, Set from jinja2 import nodes @@ -51,6 +51,91 @@ MACRO_IMPLICIT: FrozenSet[str] = frozenset({"varargs", "kwargs", "caller"}) +def template_references(source: str, env: Any = None) -> List[str]: + """Every name a template reads from its context, as dotted paths. + + `{{ a.b['c'] }}` is one reference to `a.b.c`, not three names. Loop + targets, `{% set %}` bindings and macro arguments are the template's own + and are excluded -- but only where their binding reaches, which is what + `shadowed_name_nodes` decides. + + This is the one implementation. It began as a private method on the + data-flow validator, while the dependency validator had a second + regex-based one whose task-reference pattern only recognised six + hard-coded suffixes (`result|output|data|content|status|metadata`), so + `{{ make.path }}` was a dependency to one of them and invisible to the + other. Two answers to "what does this template refer to" is one too many. + """ + from .template_sandbox import create_pipeline_environment, pipeline_global_names + + if env is None: + env = create_pipeline_environment() + + try: + ast = env.parse(source) + except Exception: + # Syntax is the template validator's business; a template that does + # not parse has no references anyone can be sure of. + return [] + + shadowed = shadowed_name_nodes(ast) + + # `a.b.c` is one reference. Only the outermost node of a chain is + # reported; the links inside it are skipped. + inner = {id(node.node) for node in ast.find_all((nodes.Getattr, nodes.Getitem))} + + # `{{ range(3) }}` names a function the environment provides, not a step. + provided = set(env.globals) | pipeline_global_names() + + references: List[str] = [] + for node in ast.find_all((nodes.Name, nodes.Getattr, nodes.Getitem)): + if id(node) in inner: + continue + base = base_name_node(node) + if base is None or id(base) in shadowed or base.name in provided: + continue + path = dotted_path(node) + if path: + references.append(path) + return references + + +def base_name_node(node: Any) -> Optional[nodes.Name]: + """The `Name` node a reference chain starts from. + + `a.b['c']` refers to `a`. Identity of that particular node decides + whether the use is shadowed, since one template can hold both a shadowed + and an unshadowed use of the same word. + """ + while isinstance(node, (nodes.Getattr, nodes.Getitem)): + node = node.node + if isinstance(node, nodes.Name) and getattr(node, "ctx", "load") == "load": + return node + return None + + +def dotted_path(node: Any) -> Optional[str]: + """`a.b`, `a['b']` and `a[0]` as the one name they refer to.""" + if isinstance(node, nodes.Name): + return node.name if getattr(node, "ctx", "load") == "load" else None + + if isinstance(node, nodes.Getattr): + base = dotted_path(node.node) + return f"{base}.{node.attr}" if base else None + + if isinstance(node, nodes.Getitem): + base = dotted_path(node.node) + if not base: + return None + key = node.arg + if isinstance(key, nodes.Const) and isinstance(key.value, str): + return f"{base}.{key.value}" + # An index selects an element of `base`; it is not its own name. + return base + + return None + + def shadowed_name_nodes(ast: nodes.Node) -> FrozenSet[int]: """The `id()` of every `Name` node that refers to a template-local binding. diff --git a/src/orchestrator/validation/data_flow_validator.py b/src/orchestrator/validation/data_flow_validator.py index 4c66a4f..507e4a7 100644 --- a/src/orchestrator/validation/data_flow_validator.py +++ b/src/orchestrator/validation/data_flow_validator.py @@ -25,7 +25,7 @@ RUNTIME_NAMESPACE, ) from ..core.template_sandbox import create_sandboxed_environment, pipeline_global_names -from ..core.template_scope import shadowed_name_nodes +from ..core.template_scope import template_references logger = logging.getLogger(__name__) @@ -429,105 +429,14 @@ def _analyze_template_string(self, def _extract_template_variables(self, template_str: str) -> List[str]: """Every variable reference in a template, as dotted paths. - This used to chop the raw text -- `split('|')[0]`, then `split(' ')[0]` - -- which does not survive an expression. Anything with a space or a - bracket in it came out mangled, and the mangled fragment was then - looked up as a task id: - - Undefined task reference: '(row' - Undefined task reference: 'from_json)' - Undefined task reference: 'analysis_topics[loop' - - None of those is a name anyone wrote. The parser knows what a - reference is, so it is asked instead -- the same move #451 made for - global calls. + The implementation lives in `core.template_scope` because the + dependency graph needs the same answer. It used to live here, while + the dependency validator had its own regex version recognising only + six hard-coded suffixes -- so the two disagreed about what a template + refers to, which is how a pipeline came to validate and then fail to + run (#465). """ - from jinja2 import nodes - - try: - ast = self.jinja_env.parse(template_str) - except Exception: - # Syntax is the template validator's business; a broken template - # simply has no references this validator can be sure of. - return [] - - # `{% for row in rows %}` binds `row` *inside its body*; it is not a - # reference to anything this validator tracks. Which uses the binding - # reaches is a question of scope, and getting that wrong in the - # permissive direction is worse than the false positive it replaced: - # - # {{ ghost.done }} <- undefined, and reported nothing - # {% for ghost in rows %} - # {{ ghost.name }} <- the binding that silenced it - # {% endfor %} - # - # A template-wide set of bound names made adding a loop anywhere in a - # file suppress that typo everywhere in it. - shadowed = shadowed_name_nodes(ast) - - # `a.b.c` is one reference, not three. Only the outermost node of a - # chain is reported; the links inside it are skipped. - inner = { - id(node.node) - for node in ast.find_all((nodes.Getattr, nodes.Getitem)) - } - - # `{{ range(3) }}` names a function the environment provides, not a - # step. Jinja's own globals and the pipeline's both count. - provided = set(self.jinja_env.globals) | pipeline_global_names() - - variables: List[str] = [] - for node in ast.find_all((nodes.Name, nodes.Getattr, nodes.Getitem)): - if id(node) in inner: - continue - base_node = self._base_name(node) - if base_node is None or id(base_node) in shadowed: - continue - if base_node.name in provided: - continue - path = self._dotted_path(node) - if path: - variables.append(path) - - return variables - - def _base_name(self, node): - """The `Name` node a reference chain starts from. - - `a.b['c']` is a reference to `a`; identity of that particular node is - what decides whether this use is shadowed, since one template can hold - both a shadowed and an unshadowed use of the same word. - """ - from jinja2 import nodes - - while isinstance(node, (nodes.Getattr, nodes.Getitem)): - node = node.node - if isinstance(node, nodes.Name) and getattr(node, "ctx", "load") == "load": - return node - return None - - def _dotted_path(self, node) -> Optional[str]: - """`a.b`, `a['b']` and `a[0]` as the one name they refer to.""" - from jinja2 import nodes - - if isinstance(node, nodes.Name): - return node.name if getattr(node, "ctx", "load") == "load" else None - - if isinstance(node, nodes.Getattr): - base = self._dotted_path(node.node) - return f"{base}.{node.attr}" if base else None - - if isinstance(node, nodes.Getitem): - base = self._dotted_path(node.node) - if not base: - return None - key = node.arg - if isinstance(key, nodes.Const) and isinstance(key.value, str): - return f"{base}.{key.value}" - # An index selects an element of `base`; it is not its own name. - return base - - return None + return template_references(template_str, self.jinja_env) def _validate_variable_reference(self, var_ref: str, diff --git a/src/orchestrator/validation/dependency_validator.py b/src/orchestrator/validation/dependency_validator.py index 3e2ebd3..ba42b63 100644 --- a/src/orchestrator/validation/dependency_validator.py +++ b/src/orchestrator/validation/dependency_validator.py @@ -12,8 +12,9 @@ import logging from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Set, Union, Tuple -import re +from typing import Any, Dict, List, Optional, Set + +from ..core.template_scope import template_references try: import networkx as nx @@ -22,7 +23,6 @@ NETWORKX_AVAILABLE = False nx = None -from ..core.exceptions import ValidationError from ..core.routing import FAILURE_POLICIES, routing_targets logger = logging.getLogger(__name__) @@ -313,35 +313,25 @@ def _get_task_dependencies(self, task: Dict[str, Any]) -> List[str]: return unique_deps def _extract_template_dependencies(self, template_str: str) -> List[str]: - """Extract task dependencies from template expressions.""" + """Task references inside a template expression. + + This was a regex whose task-reference pattern recognised only six + hard-coded suffixes -- `result|output|data|content|status|metadata` -- + so `{{ make.path }}` was not a reference at all here, while the + data-flow validator counted it. Two answers to the same question is + how a pipeline came to validate and then fail to run (#465). + + The AST extractor is now shared with the dependency graph, so the + edges this validator checks for cycles are the edges the scheduler + will use. + """ if not isinstance(template_str, str): return [] - - dependencies = set() # Use set to avoid duplicates - - # Pattern to match task references like "task_id.result", "task_id.output", etc. - # This is a simplified pattern - in practice, you'd want more sophisticated parsing - task_ref_pattern = r'\b([a-zA-Z][a-zA-Z0-9_-]*)\.(result|output|data|content|status|metadata)' - - matches = re.finditer(task_ref_pattern, template_str) - for match in matches: - task_id = match.group(1) - dependencies.add(task_id) - - # Also look for simple variable references that might be task IDs - # Pattern for variables in templates like {{ task_id }} - var_pattern = r'\{\{\s*([a-zA-Z][a-zA-Z0-9_-]*)' - var_matches = re.finditer(var_pattern, template_str) - for match in var_matches: - var_name = match.group(1) - # Only include if it looks like a task ID (not built-in variables) - if not var_name.startswith('$') and var_name not in ['item', 'index', 'is_first', 'is_last', 'iteration', 'loop']: - # Don't add if already captured by task reference pattern - if not any(var_name == dep for dep in dependencies): - dependencies.add(var_name) - - return list(dependencies) - + return [ + reference.split(".", 1)[0] + for reference in template_references(template_str) + ] + def _validate_unique_task_ids(self, tasks: List[Dict[str, Any]]) -> List[DependencyIssue]: """Validate that all task IDs are unique.""" issues = [] diff --git a/tests/test_dependency_graph.py b/tests/test_dependency_graph.py new file mode 100644 index 0000000..c8b1d2c --- /dev/null +++ b/tests/test_dependency_graph.py @@ -0,0 +1,396 @@ +"""A template reference is a dependency, and the schedule agrees. + +`{{ make.path }}` says plainly that the value cannot exist until `make` has +run. Three parts of the compiler agreed and none told the scheduler: +`DependencyValidator` inferred edges from `for_each`/`condition`/`while` but +never looked inside `parameters`; `DataFlowValidator` built a graph including +parameters and logged it; `YAMLCompiler._analyze_template` found the same +references with a third regex and stored them where nothing scheduled from +them. `Task.dependencies` came from the explicit key alone, so this validated +and then failed at run time (#465):: + + - id: make + parameters: {path: "./a.txt", content: "A"} + - id: use + parameters: {content: "{{ make.path }}"} # no `dependencies:` + +These tests pin the replacement: one graph, and the same graph is what cycle +validation inspects and what every `Task.dependencies` is built from. The +danger of fixing this by appending edges inside `_build_task` is a schedule +nobody validated, so several of these ask the compiled pipeline rather than +the graph object. +""" + +import asyncio +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from orchestrator.compiler.yaml_compiler import YAMLCompiler +from orchestrator.core.dependency_graph import ( + CONTROL_FLOW, + DECLARED, + TEMPLATE, + build_dependency_graph, +) + +pytestmark = [pytest.mark.contract] + +REPO = Path(__file__).resolve().parent.parent + + +def _graph(*steps): + return build_dependency_graph({"id": "p", "steps": list(steps)}) + + +def _step(step_id, content=None, **extra): + step = {"id": step_id, "tool": "filesystem", "action": "write", + "parameters": {"path": f"./{step_id}.txt", "content": content or step_id}} + step.update(extra) + return step + + +def _compile(*steps): + """Compile the way `orchestrator validate` does.""" + import yaml as _yaml + + source = _yaml.safe_dump({"id": "p", "name": "P", "steps": list(steps)}) + return asyncio.run(YAMLCompiler().compile(source, {})) + + +def _compile_error(*steps): + try: + _compile(*steps) + return None + except Exception as exc: # noqa: BLE001 - the message is the subject + return str(exc) + + +def _cli(command, pipeline, cwd): + env = dict(os.environ) + env["PYTHONPATH"] = str(REPO / "src") + os.pathsep + env.get("PYTHONPATH", "") + env["ORCHESTRATOR_AUTO_INSTALL"] = "0" + env.pop("ANTHROPIC_API_KEY", None) + return subprocess.run( + [sys.executable, "-m", "orchestrator.cli", command, str(pipeline)], + cwd=str(cwd), env=env, capture_output=True, text=True, timeout=300, + ) + + +# --------------------------------------------------------------------------- +# The reference orders the steps +# --------------------------------------------------------------------------- + +def test_a_referenced_step_precedes_its_consumer(): + graph = _graph(_step("make"), _step("use", "{{ make.path }}")) + assert graph.dependencies_for("use") == ["make"] + assert graph.dependencies_for("make") == [] + + +def test_the_compiled_task_carries_the_inferred_edge(): + """The graph being right is not enough; `Task.dependencies` is what runs.""" + pipeline = _compile(_step("make"), _step("use", "{{ make.path }}")) + assert pipeline.tasks["use"].dependencies == ["make"] + + +def test_execution_levels_put_the_producer_first(): + pipeline = _compile(_step("make"), _step("use", "{{ make.path }}")) + levels = pipeline.get_execution_levels() + assert levels.index(["make"]) < levels.index(["use"]), levels + + +def test_the_edge_records_where_it_came_from(): + """A diagnostic that says *why* two steps are ordered reads very + differently from one that only says they are.""" + graph = _graph(_step("make"), _step("use", "{{ make.path }}")) + edge = next(e for e in graph.edges if e.task == "use") + assert edge.origin == TEMPLATE + assert edge.location == "parameters.content" + + +# --------------------------------------------------------------------------- +# The issue #465 reproduction +# --------------------------------------------------------------------------- + +def _repro_pipeline(reference): + return ( + "id: uo\nname: UO\nsteps:\n" + " - id: make\n tool: filesystem\n action: write\n" + " parameters:\n path: \"./out_a.txt\"\n content: \"A ran\"\n" + " - id: use\n tool: filesystem\n action: write\n" + " parameters:\n path: \"./out_b.txt\"\n" + f" content: \"made={reference}\"\n" + ) + + +@pytest.mark.e2e +def test_the_issue_465_reproduction_runs(tmp_path): + """The pipeline that validated and then could not run. + + `use` references `make` and declares no `dependencies:`. Before the graph + was shared, the edge was computed and discarded, so both steps went into + the same level, `make`'s result was not in context, and the render failed. + + The reference is `make.result.path`. The issue was filed with + `make.path`, which is not a field the filesystem tool returns -- so that + reproduction conflated the missing edge with a genuinely wrong field name, + and would fail here for the second reason even with the first fixed. The + companion test below covers that case separately. + """ + pipeline = tmp_path / "p.yaml" + pipeline.write_text(_repro_pipeline("{{ make.result.path }}")) + + assert _cli("validate", pipeline, tmp_path).returncode == 0 + result = _cli("run", pipeline, tmp_path) + assert result.returncode == 0, f"{result.stdout[-700:]}{result.stderr[-700:]}" + + written = (tmp_path / "out_b.txt").read_text() + assert "{{" not in written, f"the reference never resolved: {written!r}" + assert "out_a" in written, f"the producer's output did not reach it: {written!r}" + + +@pytest.mark.e2e +def test_a_field_the_producer_does_not_return_still_fails_loudly(tmp_path): + """Ordering a step correctly does not invent a field it never produced. + + `make.path` does not exist -- the filesystem tool returns + `{'result': {'path': ...}, 'success': ...}`. The step is stopped rather + than writing an empty value, which is the right outcome. What is wrong is + that `validate` says nothing: the data-flow validator warns, because + `make` declares no outputs, and that warning never reaches stdout. + """ + pipeline = tmp_path / "p.yaml" + pipeline.write_text(_repro_pipeline("{{ make.path }}")) + + assert _cli("validate", pipeline, tmp_path).returncode == 0 + result = _cli("run", pipeline, tmp_path) + assert result.returncode != 0, "a reference to a field that does not exist ran" + assert not (tmp_path / "out_b.txt").exists(), ( + "the step wrote a file despite an unresolved reference" + ) + + +# --------------------------------------------------------------------------- +# Explicit and inferred together +# --------------------------------------------------------------------------- + +def test_an_explicit_and_inferred_edge_is_one_dependency(): + graph = _graph(_step("make"), + _step("use", "{{ make.path }}", dependencies=["make"])) + assert graph.dependencies_for("use") == ["make"] + assert graph.origins_for("use", "make") == {DECLARED, TEMPLATE} + + +def test_an_explicit_dependency_with_no_reference_survives(): + """Ordering that is only stated, never implied, must not be dropped.""" + graph = _graph(_step("first"), _step("second", dependencies=["first"])) + assert graph.dependencies_for("second") == ["first"] + + +def test_dependencies_are_reported_in_declaration_order(): + """So a pipeline compiles to the same graph however its templates are + written -- discovery order would make the schedule depend on text layout.""" + graph = _graph(_step("a"), _step("b"), + _step("c", "{{ b.path }} {{ a.path }}")) + assert graph.dependencies_for("c") == ["a", "b"] + + +def test_a_forward_reference_orders_correctly(): + """A step may reference one written later in the file.""" + graph = _graph(_step("first", "{{ later.path }}"), _step("later")) + assert graph.dependencies_for("first") == ["later"] + assert graph.levels() == [["later"], ["first"]] + + +# --------------------------------------------------------------------------- +# What must still be refused +# --------------------------------------------------------------------------- + +def test_an_inferred_cycle_is_rejected_before_execution(): + """A cycle is a deadlock: no step in it can ever become ready. + + It must be refused at compile time whether it was written as an explicit + `dependencies:` entry or implied by a template. + """ + error = _compile_error(_step("a", "{{ b.path }}"), _step("b", "{{ a.path }}")) + assert error is not None, "a cyclic pipeline compiled" + assert "cycle" in error.lower(), error + + +def test_a_longer_inferred_cycle_is_rejected(): + error = _compile_error(_step("a", "{{ c.path }}"), + _step("b", "{{ a.path }}"), + _step("c", "{{ b.path }}")) + assert error is not None and "cycle" in error.lower(), error + + +def test_a_self_reference_is_never_an_edge(): + """A self-edge cannot be satisfied, so it is recorded and not scheduled.""" + graph = _graph(_step("s", "{{ s.path }}")) + assert graph.edges == () + assert [(e.task, e.depends_on) for e in graph.self_references] == [("s", "s")] + + +def test_a_self_reference_is_rejected_by_validation(): + error = _compile_error(_step("s", "{{ s.path }}")) + assert error is not None, "a step referencing itself compiled" + + +def test_a_reference_to_a_nonexistent_step_is_still_an_error(): + """Inference must not turn a typo into silence: `{{ ghost.path }}` names + nothing, so it is not an edge, and the data-flow validator still reports + it.""" + graph = _graph(_step("real"), _step("use", "{{ ghost.path }}")) + assert graph.dependencies_for("use") == [] + + error = _compile_error(_step("real"), _step("use", "{{ ghost.path }}")) + assert error is not None and "ghost" in error, error + + +# --------------------------------------------------------------------------- +# Where references hide +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize( + "content", + [ + "{{ src.path }}", # plain + "{{ src.path | upper }}", # through a filter + "{{ src['path'] }}", # subscript + "{{ src.items[0] }}", # index + "{{ 'yes' if src.ok else 'no' }}", # conditional + "{% if src.ok %}x{% endif %}", # statement + "{{ [src.a, src.b] | join(',') }}", # inside a list literal + ], +) +def test_a_reference_is_found_however_it_is_written(content): + graph = _graph(_step("src"), _step("use", content)) + assert graph.dependencies_for("use") == ["src"], content + + +def test_a_reference_nested_in_dictionaries_and_lists_is_found(): + """Templates hide in nested structures as readily as in a plain parameter.""" + step = {"id": "deep", "action": "generate", + "parameters": {"cfg": {"items": ["{{ src.a }}", {"k": "{{ src.b }}"}]}}} + graph = _graph(_step("src"), step) + assert graph.dependencies_for("deep") == ["src"] + + +@pytest.mark.parametrize("key", ["condition", "for_each", "while", "until"]) +def test_a_control_flow_reference_orders_the_step(key): + graph = _graph(_step("src"), _step("use", **{key: "{{ src.items }}"})) + assert graph.dependencies_for("use") == ["src"] + assert graph.origins_for("use", "src") == {CONTROL_FLOW} + + +def test_a_loop_local_name_is_not_a_step_dependency(): + """`{% for src in ... %}` rebinds the name; the body is talking about the + loop variable, not the step that happens to share its id.""" + graph = _graph( + _step("src"), + _step("use", "{% for src in [1, 2] %}{{ src }}{% endfor %}"), + ) + assert graph.dependencies_for("use") == [] + + +def test_a_reference_outside_the_loop_that_shadows_it_still_counts(): + graph = _graph( + _step("src"), + _step("use", "{{ src.path }}{% for src in [1] %}{{ src }}{% endfor %}"), + ) + assert graph.dependencies_for("use") == ["src"] + + +def test_a_pipeline_input_is_not_a_step_dependency(): + graph = _graph(_step("src"), _step("use", "{{ inputs.topic }}")) + assert graph.dependencies_for("use") == [] + + +# --------------------------------------------------------------------------- +# The unknown-output contract is unchanged +# --------------------------------------------------------------------------- + +def test_an_unknown_field_of_a_declared_output_is_still_an_error(): + """Inferring the edge must not soften the check on the field name.""" + producer = {"id": "make", "action": "generate", + "outputs": {"content": {"type": "string"}}, "parameters": {}} + error = _compile_error(producer, _step("use", "{{ make.contnet }}")) + assert error is not None, "a misspelled declared output compiled" + + +def test_an_unknown_field_of_an_undeclared_output_is_still_only_a_warning(): + """When a task does not declare its outputs the validator has no basis to + reject a field name -- the false-positive class removed in #448/#450/#461. + """ + assert _compile_error(_step("make"), _step("use", "{{ make.anything }}")) is None + + +def test_the_undeclared_case_still_orders_the_steps(): + """The warning is about the *field*; the *step* reference is certain.""" + graph = _graph(_step("make"), _step("use", "{{ make.anything }}")) + assert graph.dependencies_for("use") == ["make"] + + +# --------------------------------------------------------------------------- +# The lint for authors who want it written down +# --------------------------------------------------------------------------- + +def test_an_inferred_edge_is_reported_as_implicit(): + graph = _graph(_step("make"), _step("use", "{{ make.path }}")) + implicit = graph.inferred_only() + assert [(e.task, e.depends_on) for e in implicit] == [("use", "make")] + + +def test_an_edge_that_is_also_declared_is_not_implicit(): + graph = _graph(_step("make"), + _step("use", "{{ make.path }}", dependencies=["make"])) + assert graph.inferred_only() == [] + + +def _implicit_issues(*steps): + """Compile and return the `implicit_dependency` findings.""" + import yaml as _yaml + + compiler = YAMLCompiler() + source = _yaml.safe_dump({"id": "p", "name": "P", "steps": list(steps)}) + asyncio.run(compiler.compile(source, {})) + return [ + issue for issue in compiler.validation_report.issues + if issue.code == "implicit_dependency" + ] + + +def test_the_lint_reaches_the_validation_report(): + """The graph knowing an edge is implicit is no use unless something says so.""" + issues = _implicit_issues(_step("make"), _step("use", "{{ make.result.path }}")) + assert len(issues) == 1, [i.message for i in issues] + assert issues[0].metadata["referenced_step"] == "make" + assert issues[0].metadata["parameter_path"] == "parameters.content" + + +def test_the_lint_is_informational_and_does_not_fail_validation(): + """Inference is a supported way to write a pipeline, not a mistake. + + Making this an error would break every pipeline that relies on it, which + is the option this issue deliberately did not take. + """ + issues = _implicit_issues(_step("make"), _step("use", "{{ make.result.path }}")) + assert issues[0].severity.value == "info" + assert not issues[0].is_error and not issues[0].is_warning + + +def test_the_lint_names_the_line_to_add(): + """A lint that reports a problem without the fix is a nag.""" + issues = _implicit_issues(_step("make"), _step("use", "{{ make.result.path }}")) + assert any("make" in suggestion for suggestion in issues[0].suggestions) + + +def test_a_declared_dependency_produces_no_lint(): + assert _implicit_issues( + _step("make"), + _step("use", "{{ make.result.path }}", dependencies=["make"]), + ) == []