Stop the validator rejecting pipelines that run correctly (#104)

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()

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.
 

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 = {

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
+        `<AUTO>` 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:

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")