From efb9271521f76029bf74ab88147a5e4a8ffef19f Mon Sep 17 00:00:00 2001 From: Jeremy Manning Date: Sun, 2 Aug 2026 17:37:56 -0400 Subject: [PATCH] Stop the validator rejecting pipelines that run correctly (#104) `orchestrator validate` rejected 108 of 117 examples. Only 23 of them actually fail to compile when run. The gate was measuring the validator's strictness rather than whether a pipeline works, and every false rejection told an author their working pipeline was broken. The decisive case -- one pipeline, three ways to name a parameter, all three rendering identically at run time: {{ topic }} validate rc=0 run rc=0 "abc" {{ parameters.topic }} validate rc=2 run rc=0 "abc" {{ inputs.topic }} validate rc=2 run rc=0 "abc" Four causes, each mutation-tested: **Three environments, three filter sets.** The template validator knew 56 filters, the YAML compiler 61, the runtime 70. Each built its own, so `{{ title | slugify }}` was "Unknown filter" -- in a pipeline that renders it fine. Fixing only the validator moved the failure to `truncate_words` in the compiler. They are all built by `create_pipeline_environment` now, from the runtime's registry, so they cannot drift apart again. **`parameters.` / `inputs.` unrecognised.** The data-flow validator went further and read `parameters` as a *task id*, reporting an undefined task by that name. **A declared parameter without a default was reported undefined.** It is declared; only its value is deferred to `-i name=value`. The validator registered only parameters carrying a default, so a pipeline was told `Undefined variable: 'output_path'` about the name it had just declared. **A hardcoded list of twelve output field names.** Any other field was "does not produce output". A model or `` step's output shape is not statically knowable, so this asserted knowledge the validator did not have. It is now an error only when the task declared its outputs, and a warning otherwise -- the warning still records the data-flow edge. Examples validating: 9/117 -> 18/117. Raw validation errors: 1654 -> 836. Blocking suite: 562 -> 573 passed, 0 failed. I predicted this would unblock ~85 files and it unblocked 9. "Compiles at run time" does not mean `validate` has only one objection; most files carry several independent error classes, so removing one halves the error volume without moving the file count. What remains was spot-checked and is real: `output_path` is genuinely undeclared in the files that reference it, and the runtime fails on it identically. Notes in notes/example-catalog-diagnosis.md. Co-Authored-By: Claude Opus 5 (1M context) --- src/orchestrator/compiler/yaml_compiler.py | 4 +- src/orchestrator/core/template_sandbox.py | 24 +++ .../runtime/dependency_resolver.py | 4 +- .../validation/data_flow_validator.py | 36 +++- .../validation/template_validator.py | 98 +++++---- tests/test_validator_agrees_with_runtime.py | 190 ++++++++++++++++++ 6 files changed, 299 insertions(+), 57 deletions(-) create mode 100644 tests/test_validator_agrees_with_runtime.py diff --git a/src/orchestrator/compiler/yaml_compiler.py b/src/orchestrator/compiler/yaml_compiler.py index 01cc905..cd909f0 100644 --- a/src/orchestrator/compiler/yaml_compiler.py +++ b/src/orchestrator/compiler/yaml_compiler.py @@ -10,7 +10,7 @@ import yaml from jinja2 import StrictUndefined -from ..core.template_sandbox import create_sandboxed_environment +from ..core.template_sandbox import create_pipeline_environment from ..core.actions import canonical_action from ..core.pipeline import Pipeline @@ -149,7 +149,7 @@ def __init__( self.ambiguity_resolver = None logger.info("No model registry provided - AUTO tags will be preserved") - self.template_engine = create_sandboxed_environment() + self.template_engine = create_pipeline_environment() # Add custom filters to Jinja2 environment self._register_custom_filters() diff --git a/src/orchestrator/core/template_sandbox.py b/src/orchestrator/core/template_sandbox.py index 8b96efc..af7846f 100644 --- a/src/orchestrator/core/template_sandbox.py +++ b/src/orchestrator/core/template_sandbox.py @@ -36,6 +36,30 @@ def create_sandboxed_environment(**kwargs: Any) -> SandboxedEnvironment: return SandboxedEnvironment(**kwargs) +def create_pipeline_environment(**kwargs: Any) -> SandboxedEnvironment: + """A sandboxed environment that also knows the orchestrator's own filters. + + `create_sandboxed_environment` gives a bare Jinja environment. Anything + that renders *pipeline* templates needs more than that: `slugify`, + `basename`, `from_json` and the rest are part of the pipeline language, and + an environment without them reports a working pipeline as broken. + + Three environments used to build their own filter sets independently -- the + template validator (56 filters), the YAML compiler (61) and the runtime + (70). A pipeline using `{{ title | slugify }}` therefore failed validation + and compiled fine, or compiled and failed on `truncate_words`, depending on + which one it met first. They are all built here now, from the runtime's + registry, so the three cannot drift apart again. + """ + from .template_manager import TemplateManager + + env = create_sandboxed_environment(**kwargs) + # TemplateManager owns the filter set; this copies it rather than + # re-declaring it, which is what let the three sets diverge. + env.filters.update(TemplateManager().env.filters) + return env + + def sandboxed_template(source: str, **kwargs: Any) -> Template: """A single template, compiled under the sandbox. diff --git a/src/orchestrator/runtime/dependency_resolver.py b/src/orchestrator/runtime/dependency_resolver.py index 33a79ad..88f5856 100644 --- a/src/orchestrator/runtime/dependency_resolver.py +++ b/src/orchestrator/runtime/dependency_resolver.py @@ -13,7 +13,7 @@ from dataclasses import dataclass from jinja2 import StrictUndefined, TemplateSyntaxError, UndefinedError, meta -from ..core.template_sandbox import create_sandboxed_environment +from ..core.template_sandbox import create_pipeline_environment import ast from .execution_state import PipelineExecutionState, UnresolvedItem, ItemStatus @@ -61,7 +61,7 @@ def __init__(self, execution_state: PipelineExecutionState, max_iterations: int self.max_iterations = max_iterations # Initialize Jinja2 environment for template parsing - self.jinja_env = create_sandboxed_environment() + self.jinja_env = create_pipeline_environment() # Regex patterns for different types of references self.patterns = { diff --git a/src/orchestrator/validation/data_flow_validator.py b/src/orchestrator/validation/data_flow_validator.py index 6e8fc76..ae81fc3 100644 --- a/src/orchestrator/validation/data_flow_validator.py +++ b/src/orchestrator/validation/data_flow_validator.py @@ -82,6 +82,19 @@ class TaskOutputSchema: produces: Optional[str] = None # what the task produces (file, data, etc.) format: Optional[str] = None # output format + @property + def knows_outputs(self) -> bool: + """Whether anything actually declared what this task produces. + + `outputs` is populated from an explicit `outputs:`/`produces:` on the + step, or from a tool schema. When it is empty -- a model step, an + `` step, a tool the validator has no schema for -- the only + available names are the generic fallbacks below, which say nothing + about this task. Rejecting an unrecognised field in that case asserts + knowledge the validator does not have. + """ + return bool(self.outputs) + def get_available_variables(self) -> Set[str]: """Get all variable names this task makes available.""" variables = set() @@ -385,6 +398,10 @@ def _analyze_template_string(self, if error.severity == "warning": warnings.append(error) + # The reference is still a real edge in the data-flow + # graph; downgrading the message must not lose it. + if validation_result.get("dependency"): + dependencies.add(validation_result["dependency"]) else: errors.append(error) @@ -458,8 +475,11 @@ def _validate_variable_reference(self, base_var = parts[0] - # Check for pipeline inputs - if base_var == "inputs": + # Check for pipeline inputs. `parameters` names the same merged + # namespace -- `pipeline_inputs` is built from both keys -- and without + # it here the reference fell through to the task lookup below and was + # reported as an undefined *task* called "parameters". + if base_var in ("inputs", "parameters"): if len(parts) > 1: input_name = parts[1] if input_name in pipeline_inputs: @@ -508,11 +528,21 @@ def _validate_variable_reference(self, else: # Extract just the output names for suggestions available_outputs = [var.split('.', 1)[1] for var in available_vars if '.' in var] + # Only an error when the task said what it produces. + declared = schema.knows_outputs return { "valid": False, + "severity": "error" if declared else "warning", "error_type": "undefined_output", - "message": f"Task '{source_task}' does not produce output '{output_field}'", + "message": ( + f"Task '{source_task}' does not produce output " + f"'{output_field}'" + if declared else + f"Task '{source_task}' does not declare its outputs, so " + f"'{output_field}' cannot be checked until it runs" + ), "source_task": source_task, + "dependency": source_task, "suggestions": self._suggest_similar_names(output_field, available_outputs) } else: diff --git a/src/orchestrator/validation/template_validator.py b/src/orchestrator/validation/template_validator.py index e9229f0..1e4c4ef 100644 --- a/src/orchestrator/validation/template_validator.py +++ b/src/orchestrator/validation/template_validator.py @@ -19,6 +19,13 @@ logger = logging.getLogger(__name__) +#: The namespaces a pipeline's own parameters can be reached through. +#: `{{ topic }}`, `{{ parameters.topic }}` and `{{ inputs.topic }}` all render +#: the same value, so all three must validate. Deliberately does not include +#: `execution` -- the data-flow validator accepts that one, but the runtime +#: does not populate it, and papering over that here would hide a real bug. +PARAMETER_NAMESPACES = frozenset({"parameters", "inputs"}) + @dataclass class TemplateValidationError: @@ -234,23 +241,24 @@ def validate_pipeline_templates( # Add pipeline inputs and parameters to context full_context = compile_context.copy() - # Add inputs - if "inputs" in pipeline_def: - inputs = pipeline_def["inputs"] - for input_name, input_spec in inputs.items(): - if isinstance(input_spec, dict) and "default" in input_spec: - full_context[input_name] = input_spec["default"] - elif not isinstance(input_spec, dict): - full_context[input_name] = input_spec - - # Add parameters - if "parameters" in pipeline_def: - params = pipeline_def["parameters"] - for param_name, param_spec in params.items(): - if isinstance(param_spec, dict) and "default" in param_spec: - full_context[param_name] = param_spec["default"] - elif not isinstance(param_spec, dict): - full_context[param_name] = param_spec + # Add inputs and parameters. + # + # A declared name counts as available whether or not it has a default. + # These entries are only ever tested for membership -- they answer "is + # this name declared", not "what is its value" -- and a parameter + # without a default is still perfectly well declared; its value simply + # arrives at run time from `-i name=value`. + # + # Registering only the ones with defaults meant a pipeline declaring + # `output_path` with no default was told `Undefined variable: + # 'output_path'`, which is the name it just declared. That single case + # accounted for the largest remaining cluster of false rejections. + for section in ("inputs", "parameters"): + for name, spec in (pipeline_def.get(section) or {}).items(): + if isinstance(spec, dict): + full_context[name] = spec.get("default") + else: + full_context[name] = spec # Validate all templates in pipeline all_errors = [] @@ -356,6 +364,14 @@ def _validate_variables( )) continue + # A pipeline's own parameters can be named three ways -- + # `{{ topic }}`, `{{ parameters.topic }}`, `{{ inputs.topic }}` + # -- and the runtime renders all three identically. Only the + # bare form appears in `available_context`, so the other two + # were reported undefined in pipelines that run correctly. + if var_name in PARAMETER_NAMESPACES: + continue + # Check if variable is available in context if var_name not in available_context: undefined_variables.add(var_name) @@ -632,39 +648,21 @@ def _similar_strings(self, s1: str, s2: str, threshold: float = 0.6) -> bool: return position_similarity >= threshold or char_similarity >= threshold def _register_custom_filters(self): - """Register custom filters that might be used in templates.""" - # Add common filters that might be missing - - def safe_default(value, default_value=""): - """Safe default filter.""" - return value if value is not None else default_value - - def safe_length(value): - """Safe length filter.""" - try: - return len(value) if value is not None else 0 - except TypeError: - return 0 - - def safe_json(value, indent=None): - """Safe JSON serialization.""" - import json - try: - return json.dumps(value, indent=indent, default=str) - except Exception: - return str(value) - - # Register filters - self.env.filters['default'] = safe_default - self.env.filters['length'] = safe_length - self.env.filters['json'] = safe_json - self.env.filters['to_json'] = safe_json - - # Add other commonly used filters - self.env.filters['lower'] = lambda x: str(x).lower() - self.env.filters['upper'] = lambda x: str(x).upper() - self.env.filters['replace'] = lambda x, old, new: str(x).replace(old, new) - + """Take the runtime's filters verbatim. + + This used to define its own small set -- `default`, `length`, `json`, + `lower`, `upper`, `replace` -- in parallel with the ones + `TemplateManager` registers. The two drifted: the runtime grew to 70 + filters and this environment knew 56, so `{{ title | slugify }}` was + reported as an unknown filter in a pipeline that renders it correctly. + + A validator that rejects working pipelines is worse than no validator, + so there is one source of truth and this is not it. + """ + from ..core.template_sandbox import create_pipeline_environment + + self.env = create_pipeline_environment() + if self.debug_mode: logger.debug(f"Registered {len(self.env.filters)} template filters") diff --git a/tests/test_validator_agrees_with_runtime.py b/tests/test_validator_agrees_with_runtime.py new file mode 100644 index 0000000..bf01702 --- /dev/null +++ b/tests/test_validator_agrees_with_runtime.py @@ -0,0 +1,190 @@ +"""`validate` must not reject what `run` executes correctly. + +`orchestrator validate` rejected 108 of 117 example pipelines. Only 23 of them +actually fail to compile when run; the other 85 execute fine. The gate was +measuring the validator's strictness rather than whether a pipeline works, and +every one of those false rejections told an author their working pipeline was +broken. + +Three disagreements produced almost all of it: + +1. The validator's Jinja environment was missing 14 filters the runtime + registers, so `{{ title | slugify }}` was "Unknown filter: 'slugify'". +2. Neither validator recognised the `parameters.` / `inputs.` namespaces. The + data-flow validator went further and read `parameters` as a *task id*. +3. A task's available outputs came from a hardcoded list of twelve field names, + so any domain-specific field was "does not produce output". + +The rule these tests pin is narrow and checkable: **anything the runtime +renders, the validator must accept.** The reverse is not required -- the +validator may still catch things the runtime would fail on later. +""" + +import pytest + +from orchestrator.compiler.yaml_compiler import YAMLCompiler +from orchestrator.core.template_manager import TemplateManager +from orchestrator.validation.template_validator import TemplateValidator + +pytestmark = [pytest.mark.contract] + + +def _validate(yaml_text): + """Compile the way `orchestrator validate` does. Returns None or the error.""" + import asyncio + + try: + asyncio.run(YAMLCompiler().compile(yaml_text, {})) + return None + except Exception as exc: # noqa: BLE001 - the message is the subject + return f"{type(exc).__name__}: {exc}" + + +def _pipeline(content, extra_params=""): + return f""" +id: probe +name: Probe +parameters: + topic: + type: string + default: "Hello World Report" +{extra_params} +steps: + - id: write_it + tool: filesystem + action: write + parameters: + path: "./out.txt" + content: "{content}" +""" + + +# --------------------------------------------------------------------------- +# 1a. Filters +# --------------------------------------------------------------------------- + +def test_every_environment_knows_every_filter_the_runtime_registers(): + """The drift check. + + Three environments built their own filter sets: the validator knew 56, the + compiler 61, the runtime 70. Which error a pipeline got depended on which + environment reached it first -- `{{ title | slugify }}` failed validation, + and fixing only the validator moved the failure to `truncate_words` in the + compiler. + """ + runtime = set(TemplateManager().env.filters) + others = { + "template validator": set(TemplateValidator().env.filters), + "yaml compiler": set(YAMLCompiler().template_engine.filters), + } + + drifted = { + name: sorted(runtime - filters) + for name, filters in others.items() + if runtime - filters + } + assert not drifted, ( + f"environments are missing filters the runtime registers, so pipelines " + f"using them are rejected despite working: {drifted}" + ) + + +@pytest.mark.parametrize( + "expression", + [ + "{{ topic | slugify }}", + "{{ topic | from_json }}", + "{{ '/a/b/c.txt' | basename }}", + "{{ topic | regex_search('World') }}", + "{{ topic | truncate_words(2) }}", + ], +) +def test_a_pipeline_using_a_runtime_filter_validates(expression): + error = _validate(_pipeline(expression)) + + assert error is None or "filter" not in error.lower(), ( + f"{expression} was rejected over its filter: {error}" + ) + + +# --------------------------------------------------------------------------- +# 1b. Parameter namespaces +# --------------------------------------------------------------------------- + +def test_a_parameter_without_a_default_is_still_declared(): + """Having no default does not make a parameter undeclared. + + Its value arrives at run time from `-i name=value`. The validator used to + register only parameters that carried a default, so a pipeline declaring + `output_path` was told `Undefined variable: 'output_path'` -- the name it + had just declared. This was the largest single cluster of false + rejections in the catalogue. + """ + pipeline = """ +id: no_default +name: No Default +parameters: + output_path: + type: string + description: "where to write" +steps: + - id: s + tool: filesystem + action: write + parameters: + path: "{{ output_path }}" + content: "x" +""" + assert _validate(pipeline) is None, ( + "a declared parameter without a default was reported undefined" + ) + + +@pytest.mark.parametrize( + "expression", + ["{{ topic }}", "{{ parameters.topic }}", "{{ inputs.topic }}"], +) +def test_every_way_of_naming_a_parameter_validates(expression): + """All three render identically at runtime; all three must validate. + + Before this, only the bare form did. + """ + assert _validate(_pipeline(expression)) is None, ( + f"{expression} does not validate, but running it renders the parameter " + f"correctly" + ) + + +# --------------------------------------------------------------------------- +# 1c. Step output fields +# --------------------------------------------------------------------------- + +def test_a_step_may_produce_a_field_the_validator_cannot_know(): + """A model or AUTO step's output shape is not statically knowable. + + Rejecting an unrecognised field asserts knowledge the validator does not + have. It may warn; it must not refuse to compile. + """ + pipeline = """ +id: custom_fields +name: Custom Fields +steps: + - id: analyse + action: classify the input and return a strategy + + - id: use_it + tool: filesystem + action: write + parameters: + path: "./out.txt" + content: "{{ analyse.strategy }} / {{ analyse.confidence }}" + dependencies: + - analyse +""" + # Assert it compiles, not that some particular wording is absent. Checking + # for the old message would pass the moment the message was reworded, even + # with the rejection still in force -- which is exactly what happened when + # this was first written. + assert _validate(pipeline) is None, ( + "the validator refused a pipeline over output fields it cannot know" + )