Unrendered is not one thing: split prose from structure and control (#104)

src/orchestrator/__init__.py

@@ -89,6 +89,11 @@
     "create_pipeline_api": ".api",
     "create_advanced_pipeline_compiler": ".api",
     "create_pipeline_executor": ".api",
+    # --- Validation, with its findings ---
+    "validate_pipeline_file": ".validation.pipeline_report",
+    "validate_pipeline_text": ".validation.pipeline_report",
+    "PipelineValidation": ".validation.pipeline_report",
+    "Finding": ".validation.pipeline_report",
     # --- Error hierarchy ---
     "OrchestratorError": ".core.exceptions",
     "PipelineError": ".core.exceptions",

src/orchestrator/cli.py

@@ -367,119 +367,68 @@ def _failed_steps(results) -> list:
     return failed
 
 
-def _reportable_issues(compiler):
-    """Findings a successful validation should still tell the user about.
-
-    The compiler has always collected these; nothing displayed them. A
-    pipeline could report `✓ is valid` while carrying a warning saying a
-    reference could not be checked -- and then fail at run time on exactly
-    that reference (#465). A warning that only reaches the log stream is not
-    a warning: `orchestrator validate` prints to stdout, and a script
-    capturing stdout saw nothing at all.
-
-    Informational findings are excluded: "tool is available" and "execution
-    order computed" are not things anyone needs told.
-    """
-    report = getattr(compiler, "validation_report", None)
-    if report is None:
-        return []
-    return [issue for issue in report.issues if issue.is_error or issue.is_warning]
-
-
-def _issue_payload(issue):
-    """One finding as stable, machine-readable fields.
-
-    Field names are part of the interface: anything reading this should not
-    have to parse the human message to learn which step or reference is at
-    fault.
-    """
-    metadata = issue.metadata or {}
-    return {
-        "code": issue.code,
-        "severity": issue.severity.value,
-        "category": issue.category,
-        "step": metadata.get("step", issue.component),
-        "parameter_path": metadata.get("parameter_path", issue.path),
-        "referenced_step": metadata.get("referenced_step"),
-        "referenced_field": metadata.get("referenced_field"),
-        "message": issue.message,
-        "suggestions": list(issue.suggestions or []),
-    }
-
-
 @cli.command("validate")
 @click.argument("pipeline_file", type=click.Path(exists=True, dir_okay=False))
 @click.option("--json", "as_json", is_flag=True,
               help="Emit the result and every finding as JSON on stdout.")
 @verbose_option
 def validate_pipeline(pipeline_file: str, verbose: bool, as_json: bool):
     """Compile a pipeline without running it, and report its task graph."""
-    import asyncio
     import json as _json
 
     if verbose:
         setup_logging(verbose=True)
 
-    from .compiler.yaml_compiler import YAMLCompiler
-
-    compiler = YAMLCompiler()
-
-    async def _compile():
-        with open(pipeline_file) as fh:
-            return await compiler.compile(fh.read(), {})
+    from .validation.pipeline_report import validate_pipeline_file
 
     try:
-        pipeline = asyncio.run(_compile())
+        result = validate_pipeline_file(pipeline_file)
     except KeyboardInterrupt:
         click.echo("Interrupted.", err=True)
         sys.exit(EXIT_INTERRUPTED)
-    except Exception as exc:
+
+    if not result.valid:
         if as_json:
             click.echo(_json.dumps({
                 "pipeline_file": pipeline_file,
                 "valid": False,
-                "error": f"{type(exc).__name__}: {exc}",
-                "findings": [_issue_payload(i) for i in _reportable_issues(compiler)],
+                "error": result.error,
+                "findings": [f.as_dict() for f in result.findings],
             }, indent=2))
         else:
-            click.echo(f"{type(exc).__name__}: {exc}", err=True)
+            click.echo(result.error, err=True)
         sys.exit(EXIT_VALIDATION_ERROR)
 
-    tasks = getattr(pipeline, "tasks", {}) or {}
-    findings = _reportable_issues(compiler)
+    tasks = result.tasks
+    findings = result.findings
 
     if as_json:
         click.echo(_json.dumps({
             "pipeline_file": pipeline_file,
             "valid": True,
-            "pipeline": getattr(pipeline, "id", None),
-            "tasks": {
-                task_id: list(getattr(task, "dependencies", []) or [])
-                for task_id, task in tasks.items()
-            },
-            "findings": [_issue_payload(issue) for issue in findings],
+            "pipeline": result.pipeline_id,
+            "tasks": tasks,
+            "findings": [f.as_dict() for f in findings],
         }, indent=2))
         sys.exit(EXIT_OK)
 
-    warnings = [issue for issue in findings if issue.is_warning]
+    warnings = result.warnings
     summary = f"✓ {pipeline_file} is valid"
     if warnings:
         summary += f" with {len(warnings)} warning{'s' if len(warnings) > 1 else ''}"
     click.echo(summary)
-    click.echo(f"  pipeline: {getattr(pipeline, 'id', '<unnamed>')}")
+    click.echo(f"  pipeline: {result.pipeline_id or '<unnamed>'}")
     click.echo(f"  tasks: {len(tasks)}")
-    for task_id in tasks:
-        task = tasks[task_id]
-        deps = getattr(task, "dependencies", []) or []
+    for task_id, deps in tasks.items():
         suffix = f"  <- {', '.join(deps)}" if deps else ""
         click.echo(f"    - {task_id}{suffix}")
 
     for issue in findings:
         click.echo("")
-        where = issue.path or issue.component
-        click.echo(f"{issue.severity.value}[{issue.code}] {where}:")
+        where = issue.parameter_path or issue.step
+        click.echo(f"{issue.severity}[{issue.code}] {where}:")
         click.echo(f"  {issue.message}")
-        for suggestion in issue.suggestions or []:
+        for suggestion in issue.suggestions:
             click.echo(f"  suggestion: {suggestion}")
 
     sys.exit(EXIT_OK)

src/orchestrator/compiler/yaml_compiler.py

@@ -455,7 +455,8 @@ async def _validate_templates(
                         component=error.context_path or "unknown",
                         message=error.message,
                         path=error.context_path,
-                        suggestions=error.suggestions
+                        suggestions=error.suggestions,
+                        code=error.error_type,
                     )
                     self.validation_report.add_issue(issue)
 
@@ -466,7 +467,8 @@ async def _validate_templates(
                         component=warning.context_path or "unknown",
                         message=warning.message,
                         path=warning.context_path,
-                        suggestions=warning.suggestions
+                        suggestions=warning.suggestions,
+                        code=warning.error_type,
                     )
                     self.validation_report.add_issue(issue)
 

src/orchestrator/core/step_fields.py

@@ -64,18 +64,63 @@
     "steps",
 )
 
-#: Named so a test can assert they are never scanned, and so the reason is
-#: written down rather than implied by absence.
-INERT_STEP_FIELDS: FrozenSet[str] = frozenset({
-    "id",            # the step's own name
+#: Prose. Nothing renders it and nothing acts on it, so a template here is the
+#: author's mistake to know about but not a defect in the pipeline: the braces
+#: appear in a log line and everything still runs.
+INERT_PROSE_STEP_FIELDS: FrozenSet[str] = frozenset({
     "name",          # human-readable label, copied verbatim
     "description",   # prose
-    "metadata",      # arbitrary author data
-    "tool",          # a registry key, not a template
-    "dependencies",  # already read, with the `declared` origin
+})
+
+#: Also never rendered -- and that makes a template here an error, not a note.
+#: These fields *name* things: a step, a tool in the registry, a dependency to
+#: schedule against. A literal `{{ x }}` names nothing, so the pipeline is
+#: already broken; warning about the wording and calling it valid says the
+#: opposite.
+NON_RENDERED_STRUCTURAL_STEP_FIELDS: FrozenSet[str] = frozenset({
+    "id",            # the step's own name
+    "tool",          # a registry key
+    "dependencies",  # read with the `declared` origin
     "depends_on",
 })
 
+#: `metadata` is arbitrary author data *except* for the keys the runtime
+#: reads. Those are not prose: an unrendered `{{ ... }}` is handed to control
+#: code as a literal string, so `goto: "{{ nosuch }}"` sends execution to a
+#: step named `{{ nosuch }}` and `timeout: "{{ t }}"` is a timeout of that
+#: text. Each key here has a runtime read behind it:
+#:
+#:     goto                  orchestrator.py:1074, control_flow/dynamic_flow.py:53
+#:     priority              orchestrator.py:1345
+#:     requires_model        orchestrator.py:2599
+#:     dynamic_dependencies  orchestrator.py:1166
+#:     on_failure            orchestrator.py:1111
+#:     timeout               orchestrator.py:1888
+#:     max_iterations        orchestrator.py:1001
+#:     condition             runtime/orchestrator_integration.py:123
+#:     output_schema         adapters/enhanced_langgraph_adapter.py:222
+#:     produces              auto_resolution/requirements_analyzer.py:219
+#:     required_capabilities core/control_system.py:130
+#:     validation            compiler/schema_resolver.py:149
+#:
+#: Keys the compiler *writes* to metadata (`step_type`, `retry_count`, the
+#: loop bookkeeping) are absent: an author does not supply them, so a template
+#: in one is not a case that arises.
+OPERATIONAL_METADATA_KEYS: FrozenSet[str] = frozenset({
+    "goto", "priority", "requires_model", "dynamic_dependencies",
+    "on_failure", "timeout", "max_iterations", "condition",
+    "output_schema", "produces", "required_capabilities", "validation",
+})
+
+#: Everything a step carries that the runtime does not render, whatever the
+#: consequence of writing a template in it. Dependency inference only needs
+#: "not scanned"; the distinctions above are for diagnostics.
+INERT_STEP_FIELDS: FrozenSet[str] = (
+    INERT_PROSE_STEP_FIELDS
+    | NON_RENDERED_STRUCTURAL_STEP_FIELDS
+    | frozenset({"metadata"})
+)
+
 #: The same distinction one level up. A pipeline's own `name` and
 #: `description` are prose about the pipeline; nothing renders them either, so
 #: `name: "{{ nosuch }}"` at the top of a document was failing validation for a

src/orchestrator/validation/data_flow_validator.py

@@ -615,6 +615,14 @@ def _suggest_similar_names(self, target: str, available: List[str]) -> List[str]
         """Suggest similar names for typos."""
         suggestions = []
         target_lower = target.lower()
+
+        # Sorted, because callers pass names collected in a set and this
+        # function both emits in iteration order and truncates to three. Two
+        # identical `orchestrator validate` runs offered the same three
+        # suggestions in different orders, and on a longer candidate list
+        # would have offered *different* three. The tiers below still rank the
+        # result; this only makes ties break the same way every time.
+        available = sorted(available)
 
         # Exact match (case insensitive)
         for name in available: