Give Python callers the findings the CLI already prints (#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/core/step_fields.py

@@ -75,3 +75,20 @@
     "dependencies",  # already read, with the `declared` origin
     "depends_on",
 })
+
+#: 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
+#: pipeline that runs.
+#:
+#: Deliberately short. `outputs` *is* rendered; `parameters` declares names
+#: rather than using them; and `id` and `version` are schema-constrained --
+#: `version` must match `\d+\.\d+\.\d+`, so a template there is a real error
+#: and calling the field inert would describe it wrongly. Only fields a real
+#: run tolerates an unresolvable reference in are listed, which is what
+#: `test_a_pipeline_with_templates_in_prose_still_runs` checks.
+INERT_PIPELINE_FIELDS: FrozenSet[str] = frozenset({
+    "name",
+    "description",
+    "metadata",
+})

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:

src/orchestrator/validation/pipeline_report.py

@@ -0,0 +1,169 @@
+"""Validating a pipeline, with its findings, from Python.
+
+`orchestrator validate` has reported structured findings since #467. The
+Python API had no equivalent: `PipelineAPI.validate_yaml` returns a bare
+`bool`, so a caller embedding the orchestrator could learn *that* a pipeline
+was rejected and nothing about why -- and could not see warnings at all, which
+is where "this reference could not be checked" lives.
+
+The findings were not missing, only private: `cli._reportable_issues` and
+`cli._issue_payload` built them for the JSON output and nothing else could
+reach them. A second implementation for the API would be a second thing to
+drift; this module is the one implementation, and the CLI formats what it
+returns.
+
+The payload field names are interface. Anything consuming a finding should not
+have to parse the human message to learn which step or reference is at fault.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple, Union
+
+
+@dataclass(frozen=True)
+class Finding:
+    """One validation finding, in stable machine-readable fields."""
+
+    code: str
+    severity: str
+    category: Optional[str] = None
+    step: Optional[str] = None
+    parameter_path: Optional[str] = None
+    referenced_step: Optional[str] = None
+    referenced_field: Optional[str] = None
+    message: str = ""
+    suggestions: Tuple[str, ...] = ()
+
+    @property
+    def is_error(self) -> bool:
+        return self.severity == "error"
+
+    @property
+    def is_warning(self) -> bool:
+        return self.severity == "warning"
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "code": self.code,
+            "severity": self.severity,
+            "category": self.category,
+            "step": self.step,
+            "parameter_path": self.parameter_path,
+            "referenced_step": self.referenced_step,
+            "referenced_field": self.referenced_field,
+            "message": self.message,
+            "suggestions": list(self.suggestions),
+        }
+
+
+@dataclass(frozen=True)
+class PipelineValidation:
+    """The result of validating one pipeline document."""
+
+    valid: bool
+    findings: Tuple[Finding, ...] = ()
+    pipeline_id: Optional[str] = None
+    #: Step id -> its dependencies, so a caller can check the graph the
+    #: compiler actually built rather than re-deriving it.
+    tasks: Dict[str, List[str]] = field(default_factory=dict)
+    #: The exception that stopped compilation, if one did.
+    error: Optional[str] = None
+
+    @property
+    def errors(self) -> Tuple[Finding, ...]:
+        return tuple(f for f in self.findings if f.is_error)
+
+    @property
+    def warnings(self) -> Tuple[Finding, ...]:
+        return tuple(f for f in self.findings if f.is_warning)
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "valid": self.valid,
+            "pipeline": self.pipeline_id,
+            "tasks": self.tasks,
+            "findings": [f.as_dict() for f in self.findings],
+            **({"error": self.error} if self.error is not None else {}),
+        }
+
+
+def _findings_from(compiler) -> Tuple[Finding, ...]:
+    """Findings a successful validation should still tell the caller about.
+
+    The compiler has always collected these; nothing displayed them. A
+    pipeline could report valid while carrying a warning saying a reference
+    could not be checked -- and then fail at run time on exactly that
+    reference (#465).
+
+    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 tuple(
+        _finding(issue)
+        for issue in report.issues
+        if issue.is_error or issue.is_warning
+    )
+
+
+def _finding(issue) -> Finding:
+    metadata = issue.metadata or {}
+    severity = issue.severity
+    return Finding(
+        code=issue.code,
+        severity=getattr(severity, "value", severity),
+        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=tuple(issue.suggestions or ()),
+    )
+
+
+async def validate_pipeline_text(yaml_text: str) -> PipelineValidation:
+    """Compile `yaml_text` without running it and report what was found.
+
+    Compilation failure is a result, not an exception: a caller validating
+    user input wants the findings alongside the failure, and raising would
+    discard them.
+    """
+    from ..compiler.yaml_compiler import YAMLCompiler
+
+    compiler = YAMLCompiler()
+    try:
+        pipeline = await compiler.compile(yaml_text, {})
+    except Exception as exc:
+        return PipelineValidation(
+            valid=False,
+            findings=_findings_from(compiler),
+            error=f"{type(exc).__name__}: {exc}",
+        )
+
+    tasks = getattr(pipeline, "tasks", {}) or {}
+    return PipelineValidation(
+        valid=True,
+        findings=_findings_from(compiler),
+        pipeline_id=getattr(pipeline, "id", None),
+        tasks={
+            task_id: list(getattr(task, "dependencies", []) or [])
+            for task_id, task in tasks.items()
+        },
+    )
+
+
+def validate_pipeline_file(path: Union[str, Path]) -> PipelineValidation:
+    """`validate_pipeline_text` for a file on disk, without an event loop.
+
+    Provided because the common case -- checking a pipeline someone wrote --
+    is synchronous, and requiring callers to build an event loop to learn
+    whether a file is valid is why the boolean survived so long.
+    """
+    return asyncio.run(validate_pipeline_text(Path(path).read_text()))