From f7a9df5cbf666f0a79b099a47eed4a5fe845fb11 Mon Sep 17 00:00:00 2001 From: Jeremy Manning Date: Sat, 1 Aug 2026 19:46:27 -0400 Subject: [PATCH 1/2] Return a typed PipelineResult with an execution trace (#332) execute_pipeline returned a bare {step_id: value} dict -- except when the pipeline declared `outputs:`, when it returned {"steps":..., "outputs":...} instead. Two shapes from one method, distinguishable only by inspecting keys, and a pipeline with a step called `outputs` collided with the second outright. Almost everything else the run knew was computed and then discarded at the return statement. Task has carried status, started_at, completed_at, retry_count, dependencies and the error all along; Pipeline can already report dependency levels; execution_id and start_time were sitting in the executor's context. PipelineResult keeps them. It is a Mapping, so result["step_id"] returns the step's raw value exactly as before and the several thousand existing assertions that index results directly are unaffected. The trace arrives as attributes: status, success, outputs, steps, execution_order, execution_levels, timing, and failed/skipped/retried views. Each StepResult carries its action, status, success, value, structured error and error_type, the tool or model and provider that ran it, start/end/duration, retries and dependencies. status and success answer different questions, and conflating them was a real defect. status records whether the task finished; success whether it worked. A tool returning {"success": False} without raising *does* finish, so selecting failures on status alone reported a failing pipeline as successful -- caught here by the golden exit-code test going green when it should have been red. StepResult.success accounts for both, and the CLI's exit code consults it. The CLI now serialises to_dict() rather than json.dumps(result, default=str). That `default=str` silently stringified anything unserialisable, so the JSON was neither stable nor round-trippable and CLI/API equivalence could not be asserted at all. normalized() drops the execution id and wall-clock times -- the only fields that legitimately differ between runs -- and the golden test now compares whole normalised documents from both surfaces instead of one nested value. Three test expectations changed with the contract, none weakened: - the golden and CLI tests read the typed document rather than the old flat map; - my own assertion that a failed step has status "failed" contradicted the status/success distinction this PR introduces, and was wrong; - a `len(stdout.splitlines()) < 30` ceiling stood in for "no log noise" while the payload was compact. The typed document is legitimately longer, so that bound measured the document rather than the noise. Round-tripping the whole of stdout already proves the point, and an explicit check for log markers replaces the magic number. Blocking suite 476 -> 490 passed, 13 skipped, 0 failed, 0 xfailed. Collection 3271 -> 3285. ruff, uv lock --check, git diff --check and the generated action docs all clean. Co-Authored-By: Claude Opus 5 (1M context) --- docs/adr/0001-product-contract.md | 38 +++ src/orchestrator/cli.py | 14 +- src/orchestrator/core/pipeline_result.py | 229 +++++++++++++++++ src/orchestrator/orchestrator.py | 82 +++++- tests/test_cli_hermetic_and_quiet.py | 31 ++- tests/test_golden_pipelines.py | 34 ++- tests/test_pipeline_result.py | 304 +++++++++++++++++++++++ 7 files changed, 711 insertions(+), 21 deletions(-) create mode 100644 src/orchestrator/core/pipeline_result.py create mode 100644 tests/test_pipeline_result.py diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md index 5081851d..85a03e14 100644 --- a/docs/adr/0001-product-contract.md +++ b/docs/adr/0001-product-contract.md @@ -114,6 +114,44 @@ missing extra must degrade one feature, never break the package import. Exit codes: `0` success, `1` execution failure, `2` validation/compile failure, `130` interrupted. +## The result contract + +`Orchestrator.execute_pipeline` returns a `PipelineResult` +(`core/pipeline_result.py`). It is a `Mapping`, so `result["step_id"]` returns +the step's raw value exactly as before; the trace arrives as attributes +alongside. + +| | | +|-|-| +| `status` / `success` | whether the run as a whole succeeded | +| `outputs` | declared `outputs:`, resolved | +| `steps` | `StepResult` per step | +| `execution_order` / `execution_levels` | dependency order, and what could run together | +| `started_at` / `completed_at` / `duration` | run timing | +| `failed_steps` / `skipped_steps` / `retried_steps` | the trace, by category | + +Each `StepResult` carries its canonical action, status, success, value, +structured error (`error` and `error_type`), the tool or model and provider +that ran it, start/end/duration, retry count and dependencies. + +**`status` and `success` are not the same question.** `status` records whether +the task finished; `success` whether it worked. A tool that returns +`{"success": False}` without raising *finishes* — reading only the status +reported a failing pipeline as successful, and that is why the CLI's exit code +consults `success`. + +Declared outputs do not change the shape of the return value. They used to: +a pipeline with `outputs:` returned `{"steps": …, "outputs": …}` and one +without returned `{step_id: …}`, so a caller could not index a result without +first checking which it had been handed, and a step named `outputs` collided +with the second. + +`to_dict()` is the stable serialisation the CLI emits — no `default=str` +coercion, so it round-trips. `normalized()` drops the execution id and +wall-clock times, which are the only fields that legitimately differ between +two runs. **The CLI and the Python API must produce equal normalised +documents**, compared whole rather than by selected nested values. + ## Actions and template resolution A step names either a tool and an operation on it (`tool: filesystem` with diff --git a/src/orchestrator/cli.py b/src/orchestrator/cli.py index c5eee910..4c82d5b9 100644 --- a/src/orchestrator/cli.py +++ b/src/orchestrator/cli.py @@ -311,7 +311,12 @@ async def _execute(): click.echo(f"{name}: {exc}", err=True) sys.exit(EXIT_VALIDATION_ERROR if is_validation else EXIT_EXECUTION_ERROR) - rendered = json.dumps(results, indent=2, default=str) + # Serialise the typed result, not whatever the executor happened to + # return. `default=str` silently stringified anything unserialisable, so + # the JSON was neither stable nor round-trippable and CLI/API equivalence + # could not be asserted at all. + payload = results.to_dict() if hasattr(results, "to_dict") else results + rendered = json.dumps(payload, indent=2, default=str) if output: Path(output).write_text(rendered) click.echo(f"Results written to {output}") @@ -340,6 +345,13 @@ def _failed_steps(results) -> list: Steps report themselves as ``{"success": bool, "error": ...}``. Anything that does not look like a step result is ignored rather than guessed at. """ + # A PipelineResult knows which steps failed from their recorded status, + # which catches a failure whose value is not a dict at all. The duck-typed + # fallback stays for anything still returning a plain mapping. + typed = getattr(results, "failed_steps", None) + if typed is not None: + return sorted(step.id for step in typed) + if not isinstance(results, dict): return [] failed = [] diff --git a/src/orchestrator/core/pipeline_result.py b/src/orchestrator/core/pipeline_result.py new file mode 100644 index 00000000..d656473e --- /dev/null +++ b/src/orchestrator/core/pipeline_result.py @@ -0,0 +1,229 @@ +"""The typed result of a pipeline run. + +`Orchestrator.execute_pipeline` used to return a bare `{step_id: value}` dict -- +except when the pipeline declared `outputs:`, in which case it returned +`{"steps": ..., "outputs": ...}` instead. Two different shapes from one method, +distinguishable only by inspecting the keys, and a pipeline with a step called +`outputs` collided with the second. Everything else the run knew -- how long +each step took, which model answered, what failed and why, what order things +ran in -- was computed and then dropped on the floor. + +`PipelineResult` is that state, kept. It is a `Mapping`, so `result["step_id"]` +still returns exactly what it always did and existing pipelines and tests are +unaffected; the trace arrives as attributes alongside. + +`to_dict()` is the stable serialisation the CLI emits, so a CLI run and a +Python run of the same pipeline can be compared as data rather than by fishing +selected values out of nested dicts. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any, Dict, Iterator, List, Optional, Tuple + +from .task import Task, TaskStatus + + +def _describe_error(error: Any) -> Tuple[Optional[str], Optional[str]]: + """(message, type name) for whatever a task recorded as its error.""" + if error is None: + return None, None + if isinstance(error, BaseException): + return str(error), type(error).__name__ + return str(error), type(error).__name__ + + +@dataclass(frozen=True) +class StepResult: + """What one step did. + + `value` is the step's raw result -- the same object `result["step_id"]` + has always returned. Everything else is the trace around it. + """ + + id: str + action: str + status: str + success: bool + value: Any = None + + error: Optional[str] = None + error_type: Optional[str] = None + + #: Which tool ran it, when a tool did. + tool: Optional[str] = None + #: Which model answered, and from which provider, when a model did. + model: Optional[str] = None + provider: Optional[str] = None + + started_at: Optional[float] = None + completed_at: Optional[float] = None + #: Seconds, or None when the step never started. + duration: Optional[float] = None + + #: Attempts *after* the first. `Task.retry_count` counts the initial + #: failure too, so it is not the same number. + retries: int = 0 + dependencies: Tuple[str, ...] = () + + @classmethod + def from_task(cls, task: Task, value: Any = None) -> "StepResult": + message, error_type = _describe_error(task.error) + duration = None + if task.started_at is not None and task.completed_at is not None: + duration = task.completed_at - task.started_at + + # A model step records the model it used in its own envelope; there is + # nowhere else the identity survives. + model = provider = None + reported_failure = False + if isinstance(value, Mapping): + model = value.get("model_used") or value.get("model") + provider = value.get("provider") + # A step can fail *without raising*: a tool returns + # {"success": False, "error": ...} and the run continues. The task + # is then COMPLETED -- it did finish -- but the step did not + # succeed, and reading only the task status reports a failing + # pipeline as successful. + reported_failure = value.get("success") is False + if reported_failure and message is None: + message, error_type = _describe_error(value.get("error")) + + succeeded = task.status is TaskStatus.COMPLETED and not reported_failure + + return cls( + id=task.id, + action=str(task.action), + status=task.status.value, + success=succeeded, + value=value, + error=message, + error_type=error_type, + tool=task.metadata.get("tool"), + model=model, + provider=provider, + started_at=task.started_at, + completed_at=task.completed_at, + duration=duration, + retries=max(0, task.retry_count - 1) if task.error else task.retry_count, + dependencies=tuple(task.dependencies), + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "action": self.action, + "status": self.status, + "success": self.success, + "value": self.value, + "error": self.error, + "error_type": self.error_type, + "tool": self.tool, + "model": self.model, + "provider": self.provider, + "started_at": self.started_at, + "completed_at": self.completed_at, + "duration": self.duration, + "retries": self.retries, + "dependencies": list(self.dependencies), + } + + +@dataclass(frozen=True) +class PipelineResult(Mapping): + """The typed result of one pipeline run. + + Behaves as a mapping of step id -> that step's raw value, which is what + `execute_pipeline` has always returned, so existing callers keep working. + """ + + pipeline_id: str + execution_id: str + status: str + success: bool + + steps: Dict[str, StepResult] = field(default_factory=dict) + outputs: Dict[str, Any] = field(default_factory=dict) + + execution_order: Tuple[str, ...] = () + #: Steps grouped by dependency level; members of a level may run together. + execution_levels: Tuple[Tuple[str, ...], ...] = () + + started_at: Optional[float] = None + completed_at: Optional[float] = None + duration: Optional[float] = None + + #: Pipeline-level failure, when the run itself aborted. + error: Optional[str] = None + error_type: Optional[str] = None + + # -- Mapping: backwards compatible access ------------------------------ + + def __getitem__(self, key: str) -> Any: + return self.steps[key].value + + def __iter__(self) -> Iterator[str]: + return iter(self.steps) + + def __len__(self) -> int: + return len(self.steps) + + # -- the trace --------------------------------------------------------- + + @property + def failed_steps(self) -> List[StepResult]: + """Steps that did not succeed, whether they raised or reported it. + + Selecting on `success` rather than on status is deliberate: a tool that + returns `{"success": False}` leaves its task COMPLETED, and a pipeline + containing one is not a successful pipeline. + """ + return [ + s + for s in self.steps.values() + if not s.success and s.status != TaskStatus.SKIPPED.value + ] + + @property + def skipped_steps(self) -> List[StepResult]: + return [s for s in self.steps.values() if s.status == TaskStatus.SKIPPED.value] + + @property + def retried_steps(self) -> List[StepResult]: + return [s for s in self.steps.values() if s.retries > 0] + + def to_dict(self) -> Dict[str, Any]: + """Stable serialisation. The CLI emits exactly this.""" + return { + "pipeline_id": self.pipeline_id, + "execution_id": self.execution_id, + "status": self.status, + "success": self.success, + "steps": {sid: step.to_dict() for sid, step in self.steps.items()}, + "outputs": self.outputs, + "execution_order": list(self.execution_order), + "execution_levels": [list(level) for level in self.execution_levels], + "started_at": self.started_at, + "completed_at": self.completed_at, + "duration": self.duration, + "error": self.error, + "error_type": self.error_type, + } + + def normalized(self) -> Dict[str, Any]: + """`to_dict()` with everything run-specific removed. + + Wall-clock times and the execution id differ between any two runs, so + comparing raw results would always fail. This is the form the CLI and + the Python API must agree on byte for byte. + """ + data = self.to_dict() + data["execution_id"] = "" + for key in ("started_at", "completed_at", "duration"): + data[key] = None + for step in data["steps"].values(): + for key in ("started_at", "completed_at", "duration"): + step[key] = None + return data diff --git a/src/orchestrator/orchestrator.py b/src/orchestrator/orchestrator.py index 6c0f24c2..96816b6a 100644 --- a/src/orchestrator/orchestrator.py +++ b/src/orchestrator/orchestrator.py @@ -11,6 +11,7 @@ from .core.control_system import ControlSystem from .core.error_handler import ErrorHandler from .core.pipeline import Pipeline +from .core.pipeline_result import PipelineResult, StepResult from .core.pipeline_status_tracker import PipelineStatusTracker from .core.pipeline_resume_manager import PipelineResumeManager, ResumeStrategy from .core.resource_allocator import ResourceAllocator @@ -336,11 +337,25 @@ async def execute_pipeline( # Execute pipeline results = await self._execute_pipeline_internal(pipeline, context) - # Extract outputs if defined in pipeline metadata - final_result = results - if pipeline.metadata.get("outputs"): - outputs = self._extract_outputs(pipeline, results) - final_result = {"steps": results, "outputs": outputs} + # Extract outputs if defined in pipeline metadata. + # + # This used to change the *shape* of the return value: a bare + # {step_id: value} normally, but {"steps": ..., "outputs": ...} + # when the pipeline declared outputs. Two shapes from one method, + # told apart only by inspecting keys, and a step called `outputs` + # collided with the second. The outputs now live on a field. + outputs = ( + self._extract_outputs(pipeline, results) + if pipeline.metadata.get("outputs") + else {} + ) + final_result = self._build_pipeline_result( + pipeline=pipeline, + execution_id=execution_id, + results=results, + outputs=outputs, + started_at=context["start_time"], + ) # Record successful execution execution_record = { @@ -2182,6 +2197,63 @@ def _process_enhanced_results(self, results: Dict[str, Any]) -> Dict[str, Any]: return processed_results + def _build_pipeline_result( + self, + pipeline: Pipeline, + execution_id: str, + results: Dict[str, Any], + outputs: Dict[str, Any], + started_at: float, + ) -> PipelineResult: + """Assemble the typed result from state the run already holds. + + Almost nothing here is new information: `Task` has carried status, + timing, retry count and the error all along, and `Pipeline` can already + report dependency order. It was simply discarded at the return + statement. This keeps it. + """ + completed_at = time.time() + + steps: Dict[str, StepResult] = {} + for task_id in pipeline.tasks: + task = pipeline.get_task(task_id) + if task is None: + continue + steps[task_id] = StepResult.from_task(task, results.get(task_id)) + + try: + execution_levels = tuple( + tuple(level) for level in pipeline.get_execution_levels() + ) + except Exception: # pragma: no cover - a cyclic graph cannot reach here + execution_levels = () + execution_order = tuple( + task_id for level in execution_levels for task_id in level + ) + + # Not `status == FAILED`: a tool can report {"success": False} without + # raising, leaving its task COMPLETED. StepResult.success accounts for + # both, and a pipeline containing such a step has not succeeded. + failed = [ + s + for s in steps.values() + if not s.success and s.status != TaskStatus.SKIPPED.value + ] + + return PipelineResult( + pipeline_id=pipeline.id, + execution_id=execution_id, + status="failed" if failed else "completed", + success=not failed, + steps=steps, + outputs=outputs, + execution_order=execution_order, + execution_levels=execution_levels, + started_at=started_at, + completed_at=completed_at, + duration=completed_at - started_at, + ) + def _extract_outputs( self, pipeline: Pipeline, results: Dict[str, Any] ) -> Dict[str, Any]: diff --git a/tests/test_cli_hermetic_and_quiet.py b/tests/test_cli_hermetic_and_quiet.py index af1e0632..0e4c8981 100644 --- a/tests/test_cli_hermetic_and_quiet.py +++ b/tests/test_cli_hermetic_and_quiet.py @@ -221,6 +221,18 @@ def test_registry_populates_when_a_model_is_actually_demanded(tmp_path): # 2. output volume # --------------------------------------------------------------------------- + +def _normalise_run(payload): + """Drop the fields that legitimately differ between two runs.""" + payload["execution_id"] = "" + for key in ("started_at", "completed_at", "duration"): + payload[key] = None + for step in payload.get("steps", {}).values(): + for key in ("started_at", "completed_at", "duration"): + step[key] = None + return payload + + def test_default_output_is_only_the_result_document(tmp_path): """Default stdout is the typed result and nothing else; stderr is empty.""" home = _decoy_home(tmp_path) @@ -232,7 +244,7 @@ def test_default_output_is_only_the_result_document(tmp_path): # Parsable with no leading noise to skip past. payload = json.loads(result.stdout) - assert payload["read_back"]["result"]["content"] == "hi world" + assert payload["steps"]["read_back"]["value"]["result"]["content"] == "hi world" assert result.stderr == "", f"unexpected output on stderr:\n{result.stderr}" @@ -240,7 +252,14 @@ def test_default_output_is_only_the_result_document(tmp_path): # emitted well over a hundred lines around it. rendered = json.dumps(payload, indent=2, default=str) assert result.stdout.strip() == rendered - assert len(result.stdout.splitlines()) < 30 + # Round-tripping the *whole* of stdout is the real check: any log line + # anywhere would make the parse above fail. A line-count ceiling stood in + # for this while the payload was a compact {step_id: value} map, but the + # typed result carries a trace and is legitimately longer, so the bound + # measured the document rather than the noise it was meant to catch. + assert not any( + marker in result.stdout for marker in ("DEBUG", "INFO", "WARNING", "Traceback") + ) def test_verbose_restores_the_detailed_trace(tmp_path): @@ -274,9 +293,11 @@ def test_verbose_restores_the_detailed_trace(tmp_path): assert DECOY_ANTHROPIC not in trace assert DECOY_OPENAI not in trace - # Both surfaces still produce the same result document. - assert json.loads(quiet.stdout) == json.loads( - loud.stdout[loud.stdout.index("{"):] + # Both surfaces still produce the same result document. The typed + # document carries an execution id and wall-clock timings, which differ + # between any two runs, so those are normalised away first. + assert _normalise_run(json.loads(quiet.stdout)) == _normalise_run( + json.loads(loud.stdout[loud.stdout.index("{"):]) ) diff --git a/tests/test_golden_pipelines.py b/tests/test_golden_pipelines.py index 7a405976..a018f8c3 100644 --- a/tests/test_golden_pipelines.py +++ b/tests/test_golden_pipelines.py @@ -142,10 +142,11 @@ def test_basic_pipeline_via_cli(tmp_path): # stdout must be the typed result document. payload = json.loads(result.stdout[result.stdout.index("{"):]) - assert payload["write_greeting"]["success"] is True - assert payload["read_back"]["success"] is True + steps = payload["steps"] + assert steps["write_greeting"]["success"] is True + assert steps["read_back"]["success"] is True # The dependent step read back exactly what the first step wrote. - assert payload["read_back"]["result"]["content"] == "hello world" + assert steps["read_back"]["value"]["result"]["content"] == "hello world" @pytest.mark.asyncio @@ -167,11 +168,22 @@ async def test_cli_and_api_agree(tmp_path): assert cli_result.returncode == 0, cli_result.stderr cli_payload = json.loads(cli_result.stdout[cli_result.stdout.index("{"):]) - api_payload = await _run_api(BASIC, {"greeting": "hello"}, api_dir) + api_result = await _run_api(BASIC, {"greeting": "hello"}, api_dir) + # Whole documents, not selected values: a difference anywhere is caught. + def _normalise(payload): + payload = json.loads(json.dumps(payload)) + payload["execution_id"] = "" + for key in ("started_at", "completed_at", "duration"): + payload[key] = None + for step in payload["steps"].values(): + for key in ("started_at", "completed_at", "duration"): + step[key] = None + return payload + + assert _normalise(cli_payload) == _normalise(api_result.to_dict()) assert ( - cli_payload["read_back"]["result"]["content"] - == api_payload["read_back"]["result"]["content"] + cli_payload["steps"]["read_back"]["value"]["result"]["content"] == "hello world" ) assert (cli_dir / "golden_out" / "greeting.txt").read_text() == ( @@ -216,10 +228,12 @@ def test_failed_step_propagates_to_exit_code(tmp_path): assert "read_missing" in result.stderr payload = json.loads(result.stdout[result.stdout.index("{"):]) - assert payload["read_missing"]["success"] is False - assert payload["read_missing"]["error"] + assert payload["success"] is False + steps = payload["steps"] + assert steps["read_missing"]["success"] is False + assert steps["read_missing"]["error"] # The step before the failure still ran and is reported honestly. - assert payload["write_first"]["success"] is True + assert steps["write_first"]["success"] is True assert (tmp_path / "golden_out" / "before_failure.txt").is_file() @@ -264,4 +278,4 @@ def test_output_file_option(tmp_path): result = _run_cli(["run", str(BASIC), "-o", str(target)], cwd=tmp_path) assert result.returncode == 0, result.stderr payload = json.loads(target.read_text()) - assert payload["read_back"]["success"] is True + assert payload["steps"]["read_back"]["success"] is True diff --git a/tests/test_pipeline_result.py b/tests/test_pipeline_result.py new file mode 100644 index 00000000..806abaf4 --- /dev/null +++ b/tests/test_pipeline_result.py @@ -0,0 +1,304 @@ +"""The PipelineResult contract. + +`execute_pipeline` used to return a bare `{step_id: value}` dict -- except when +the pipeline declared `outputs:`, when it returned +`{"steps": ..., "outputs": ...}` instead. Two shapes from one method, told +apart only by inspecting keys. Everything else the run knew (timing, model +identity, retries, skips, the order things ran in) was computed and dropped. + +These tests pin the typed result, and pin that the CLI and the Python API agree +on it as *data* rather than on a handful of nested values. +""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from orchestrator.core.pipeline_result import PipelineResult, StepResult +from orchestrator.core.task import Task, TaskStatus +from tests.test_infrastructure import create_test_orchestrator + +pytestmark = [pytest.mark.contract, pytest.mark.e2e] + +TOOL_PIPELINE = """ +id: result_contract_pipeline +name: Result Contract Pipeline +parameters: + out_dir: + type: string + default: "./result_out" +steps: + - id: alpha + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/alpha.txt" + content: "alpha-value" + - id: read_alpha + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/alpha.txt" + dependencies: + - alpha +outputs: + saved: "{{ read_alpha.result.content }}" +""" + +FAILING_PIPELINE = """ +id: failing_result_pipeline +name: Failing Result Pipeline +steps: + - id: read_missing + tool: filesystem + action: read + on_failure: continue + parameters: + path: "/nonexistent/definitely/not/here.txt" +""" + + +def _run_api(yaml_content, cwd, context=None): + import asyncio + + previous = Path.cwd() + os.chdir(cwd) + try: + orchestrator = create_test_orchestrator() + return asyncio.run( + orchestrator.execute_yaml(yaml_content=yaml_content, context=context or {}) + ) + finally: + os.chdir(previous) + + +def _run_cli(pipeline_path, cwd): + env = dict(os.environ) + env["PYTHONPATH"] = ( + str(Path(__file__).parent.parent / "src") + os.pathsep + env.get("PYTHONPATH", "") + ) + env.pop("ANTHROPIC_API_KEY", None) + env["ORCHESTRATOR_AUTO_INSTALL"] = "0" + return subprocess.run( + [sys.executable, "-m", "orchestrator.cli", "run", str(pipeline_path)], + cwd=str(cwd), + env=env, + capture_output=True, + text=True, + timeout=300, + ) + + +# --------------------------------------------------------------------------- +# shape +# --------------------------------------------------------------------------- + +def test_the_result_is_typed_and_still_indexes_by_step_id(tmp_path): + """The compatibility promise: `result["step_id"]` is unchanged. + + Thousands of existing assertions index the result directly. A typed result + that broke them would have to be introduced by rewriting them all, so the + type is a Mapping and the trace arrives alongside. + """ + result = _run_api(TOOL_PIPELINE, tmp_path) + + assert isinstance(result, PipelineResult) + assert "alpha" in result and "read_alpha" in result + assert result["read_alpha"] == result.steps["read_alpha"].value + + +def test_declared_outputs_no_longer_change_the_return_shape(tmp_path): + """Outputs are a field, not a different shape. + + Previously a pipeline with `outputs:` returned {"steps":…, "outputs":…} + and one without returned {step_id: …}, so a caller could not index a + result without first checking which it had been handed. A step named + `outputs` collided with the second shape outright. + """ + result = _run_api(TOOL_PIPELINE, tmp_path) + + assert result.outputs, "declared outputs must be resolved" + assert result.outputs["saved"] == "alpha-value" + # The steps are still addressable exactly as in a pipeline with no outputs. + assert sorted(result) == ["alpha", "read_alpha"] + + +def test_overall_status_and_success(tmp_path): + result = _run_api(TOOL_PIPELINE, tmp_path) + + assert result.success is True + assert result.status == "completed" + assert result.pipeline_id == "result_contract_pipeline" + assert result.execution_id + + +# --------------------------------------------------------------------------- +# the trace +# --------------------------------------------------------------------------- + +def test_execution_order_and_levels_reflect_the_dependency_graph(tmp_path): + result = _run_api(TOOL_PIPELINE, tmp_path) + + assert result.execution_order.index("alpha") < result.execution_order.index("read_alpha") + assert result.execution_levels == (("alpha",), ("read_alpha",)), ( + f"a dependent step must occupy a later level: {result.execution_levels}" + ) + + +def test_every_step_carries_timing(tmp_path): + result = _run_api(TOOL_PIPELINE, tmp_path) + + for step in result.steps.values(): + assert step.started_at is not None, f"{step.id} has no start time" + assert step.completed_at is not None, f"{step.id} has no completion time" + assert step.duration is not None and step.duration >= 0 + assert result.duration is not None and result.duration >= 0 + + +def test_the_tool_that_ran_a_step_is_recorded(tmp_path): + result = _run_api(TOOL_PIPELINE, tmp_path) + + assert result.steps["alpha"].tool == "filesystem" + + +def test_the_model_that_answered_a_step_is_recorded(tmp_path): + model_pipeline = """ +id: model_result_pipeline +name: Model Result Pipeline +steps: + - id: think + action: generate + parameters: + prompt: "summarise this" +""" + result = _run_api(model_pipeline, tmp_path) + + assert result.steps["think"].model == "test-model", ( + "the selected model's identity must survive into the trace" + ) + + +def test_a_failed_step_carries_a_structured_error(tmp_path): + """Not a stringified recovery policy -- the reason. + + Every failed step used to record the error handler's decision -- + {"action": "retry", "delay": 5.0, ...} -- in place of its error. + """ + result = _run_api(FAILING_PIPELINE, tmp_path) + + step = result.steps["read_missing"] + assert step.success is False + # `status` records whether the task finished; `success` whether it worked. + # A tool that returns {"success": False} without raising finishes, so the + # two differ here -- and reading only `status` is what let a failing + # pipeline report success. + assert step.status == TaskStatus.COMPLETED.value + assert step.error, "a failed step must say why" + assert step.error_type, "and of what kind" + assert "retry" not in str(step.error_type).lower() + + assert result.success is False + assert result.status == "failed" + assert [s.id for s in result.failed_steps] == ["read_missing"] + + +# --------------------------------------------------------------------------- +# serialisation, and CLI/API equivalence +# --------------------------------------------------------------------------- + +def test_to_dict_is_json_serialisable_without_coercion(tmp_path): + """`json.dumps(..., default=str)` hid unserialisable values by stringifying + them, so the JSON was neither stable nor round-trippable.""" + result = _run_api(TOOL_PIPELINE, tmp_path) + + encoded = json.dumps(result.to_dict()) # no `default=` -- must not need one + assert json.loads(encoded)["pipeline_id"] == "result_contract_pipeline" + + +def test_normalized_drops_everything_run_specific(tmp_path): + """Two runs of the same pipeline must normalise to the same data.""" + first = _run_api(TOOL_PIPELINE, tmp_path) + second = _run_api(TOOL_PIPELINE, tmp_path) + + assert first.normalized() == second.normalized(), ( + "two runs of one pipeline disagree after normalisation" + ) + + +def test_the_cli_and_the_python_api_produce_the_same_result(tmp_path): + """The contract that makes the CLI and the library one product. + + Compared as whole normalised documents, not by fishing out selected nested + values -- that is what lets a difference anywhere be caught. + """ + cli_dir = tmp_path / "cli" + api_dir = tmp_path / "api" + cli_dir.mkdir() + api_dir.mkdir() + + pipeline_path = cli_dir / "pipeline.yaml" + pipeline_path.write_text(TOOL_PIPELINE) + + completed = _run_cli(pipeline_path, cli_dir) + assert completed.returncode == 0, ( + f"CLI run failed:\n{completed.stdout}\n{completed.stderr}" + ) + from_cli = json.loads(completed.stdout) + + from_api = _run_api(TOOL_PIPELINE, api_dir).to_dict() + + # Normalise both the same way the type does. + for payload in (from_cli, from_api): + payload["execution_id"] = "" + for key in ("started_at", "completed_at", "duration"): + payload[key] = None + for step in payload["steps"].values(): + for key in ("started_at", "completed_at", "duration"): + step[key] = None + + assert from_cli == from_api, ( + "the CLI and the Python API disagree about the same pipeline" + ) + + +# --------------------------------------------------------------------------- +# the type in isolation +# --------------------------------------------------------------------------- + +def test_step_result_from_a_skipped_task(): + task = Task(id="s", name="s", action="generate") + task.skip("dependency failed") + + step = StepResult.from_task(task) + + assert step.status == TaskStatus.SKIPPED.value + assert step.success is False + + +def test_step_result_reports_retries_beyond_the_first_attempt(): + task = Task(id="s", name="s", action="generate") + task.fail(RuntimeError("boom")) + task.fail(RuntimeError("boom again")) + + step = StepResult.from_task(task) + + assert step.retries == 1, ( + "Task.retry_count counts the initial failure too, so it is not the " + f"number of retries: got retry_count={task.retry_count}" + ) + assert step.error == "boom again" + assert step.error_type == "RuntimeError" + + +def test_a_result_with_no_steps_is_still_a_mapping(): + result = PipelineResult( + pipeline_id="p", execution_id="e", status="completed", success=True + ) + + assert len(result) == 0 + assert list(result) == [] + assert result.to_dict()["steps"] == {} From 5c4ad845318f70157e2ebf876b125a2bb970a1f2 Mon Sep 17 00:00:00 2001 From: Jeremy Manning Date: Sat, 1 Aug 2026 20:06:34 -0400 Subject: [PATCH 2/2] Implement control-flow routing: on_false, on_success, on_failure (#333) Stacked on #442, which the routing tests use to assert skipped steps through the typed result. `on_false` and `on_success` did not exist at all. `condition:` was already evaluated and already skipped a step, so what was missing was purely the routing: where to go once the answer is known. Routing jumps forward and marks the steps in between as skipped, reusing `_skip_tasks_between` -- the machinery `goto` already had -- rather than adding a second one beside it. Skipping is not failing: a pipeline that routed around a step still reports success, and PipelineResult separates skipped_steps from failed_steps accordingly. The awkward part is `on_failure`, which already meant a failure *policy* (fail / continue / skip / retry) and now also names a step to jump to. Both readings are supported and disambiguated by value: a reserved policy word keeps its policy meaning, anything else is a step id, and when it names a step the run continues there instead of aborting. That is only safe because the genuinely ambiguous case is refused rather than guessed. A pipeline with a step whose id *is* a policy word -- a step called `retry` -- fails to compile, because `on_failure: retry` could not be told apart from selecting the policy. core/routing.py holds that single definition; the compiler and the executor both consume it. Routing targets are validated at compile time: a jump to a step that does not exist, or a step routing to itself, is exit 2 naming the target. That check went into validation/dependency_validator.py, which is the validator actually on the compile path -- my first attempt put it in compiler/schema_validator.py, whose validate_dependencies() nothing calls, so the tests failed while the code looked right. Only the schema widening stayed there: `on_failure` was constrained to an enum of the four policies, which would have rejected every routing target outright. The runtime decision uses StepResult.success, not TaskStatus. A tool returning {"success": False} without raising leaves its task COMPLETED, so routing on status alone sent a failing step down the on_success path -- the same distinction #442 had to draw, reused here rather than restated a third time. The contract is updated: ADR 0001 gains a control-flow section, and docs/actions.md is generated with the routing table and the policy list so the published vocabulary cannot drift from FAILURE_POLICIES. Blocking suite 490 -> 506 passed, 13 skipped, 0 failed, 0 xfailed. Collection 3285 -> 3301. Examples validating unchanged at 3 of 111. Co-Authored-By: Claude Opus 5 (1M context) --- docs/actions.md | 21 ++ docs/adr/0001-product-contract.md | 43 ++++ scripts/generate_action_docs.py | 28 +++ src/orchestrator/compiler/schema_validator.py | 15 +- src/orchestrator/compiler/yaml_compiler.py | 7 + src/orchestrator/core/routing.py | 62 +++++ src/orchestrator/orchestrator.py | 52 +++- .../validation/dependency_validator.py | 50 +++- tests/test_control_flow_routing.py | 233 ++++++++++++++++++ 9 files changed, 506 insertions(+), 5 deletions(-) create mode 100644 src/orchestrator/core/routing.py create mode 100644 tests/test_control_flow_routing.py diff --git a/docs/actions.md b/docs/actions.md index c4f6069e..20326f1f 100644 --- a/docs/actions.md +++ b/docs/actions.md @@ -81,3 +81,24 @@ written as instructions rather than identifiers. handed to the model to interpret. That is an explicit request, which is why it remains supported while the implicit "unrecognised action becomes a prompt" behaviour does not. + +## Control-flow routing + +A step may name where execution goes next. Routing jumps **forward**; +steps between the source and the target are marked `skipped`. Skipping is +not failing -- a pipeline that routed around a step still succeeds. + +| Key | Fires when | +|-|-| +| `on_false` | the step's own `condition:` was false, so it was skipped | +| `on_success` | the step ran and succeeded | +| `on_failure` | the step ran and did not succeed | + +`on_failure` means two things, told apart by value. These are failure +**policies**: `continue`, `fail`, `retry`, `skip`. Any other value names a **step to jump to**, +and the run continues there instead of aborting. + +A step whose id is one of those policy words is refused at compile time, +since routing to it could not be distinguished from selecting the policy. +Routing targets are checked at compile time too: a jump to a step that +does not exist, or a step routing to itself, is exit 2. diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md index 85a03e14..4095e27d 100644 --- a/docs/adr/0001-product-contract.md +++ b/docs/adr/0001-product-contract.md @@ -114,6 +114,49 @@ missing extra must degrade one feature, never break the package import. Exit codes: `0` success, `1` execution failure, `2` validation/compile failure, `130` interrupted. +## Control-flow routing + +A step may say where execution goes next: + +```yaml +- id: check_topic + action: evaluate_condition + condition: "{{ topic | length > 10 }}" + parameters: + condition: "{{ topic | length > 10 }}" + on_false: short_topic_handler + on_success: main_processing + on_failure: error_recovery +``` + +| Key | Fires when | +|-|-| +| `on_false` | the step's own `condition:` was false, so the step was skipped | +| `on_success` | the step ran and succeeded | +| `on_failure` | the step ran and did not succeed | + +Routing jumps **forward**. Steps between the source and the target are marked +`skipped`, which is the same machinery `goto` already used rather than a second +one beside it. Skipping is not failing: a pipeline that routed around a step +still reports `success`. + +"Did not succeed" means `StepResult.success`, not status — a tool returning +`{"success": False}` without raising leaves its task `completed`, and routing +on status alone sent a failing step down the success path. + +**`on_failure` means two things, told apart by value.** It already selected a +failure *policy* — `fail`, `continue`, `skip`, `retry` — and now also names a +step to jump to. A reserved policy word keeps its policy meaning; anything else +is a step id. When `on_failure` names a step, the run continues there instead +of aborting. + +That is only safe because the ambiguous case is refused rather than guessed: +a pipeline containing a step whose id *is* a policy word fails to compile, +because routing to it could not be distinguished from selecting the policy. + +Routing targets are validated at compile time — a jump to a step that does not +exist, or a step routing to itself, is exit 2 naming the offending target. + ## The result contract `Orchestrator.execute_pipeline` returns a `PipelineResult` diff --git a/scripts/generate_action_docs.py b/scripts/generate_action_docs.py index 80275228..4363e6d3 100644 --- a/scripts/generate_action_docs.py +++ b/scripts/generate_action_docs.py @@ -20,6 +20,7 @@ sys.path.insert(0, str(ROOT / "src")) from orchestrator.core.actions import ACTION_FAMILIES, ACTION_SPECS # noqa: E402 +from orchestrator.core.routing import FAILURE_POLICIES # noqa: E402 TARGET = ROOT / "docs" / "actions.md" @@ -106,6 +107,33 @@ def render() -> str: "it remains supported while the implicit \"unrecognised action becomes a\n" "prompt\" behaviour does not.\n" ) + + lines.append("## Control-flow routing\n") + lines.append( + "A step may name where execution goes next. Routing jumps **forward**;\n" + "steps between the source and the target are marked `skipped`. Skipping is\n" + "not failing -- a pipeline that routed around a step still succeeds.\n" + ) + lines.append("| Key | Fires when |") + lines.append("|-|-|") + lines.append( + "| `on_false` | the step's own `condition:` was false, so it was skipped |" + ) + lines.append("| `on_success` | the step ran and succeeded |") + lines.append("| `on_failure` | the step ran and did not succeed |") + lines.append("") + policies = ", ".join(f"`{p}`" for p in sorted(FAILURE_POLICIES)) + lines.append( + f"`on_failure` means two things, told apart by value. These are failure\n" + f"**policies**: {policies}. Any other value names a **step to jump to**,\n" + f"and the run continues there instead of aborting.\n" + ) + lines.append( + "A step whose id is one of those policy words is refused at compile time,\n" + "since routing to it could not be distinguished from selecting the policy.\n" + "Routing targets are checked at compile time too: a jump to a step that\n" + "does not exist, or a step routing to itself, is exit 2.\n" + ) return "\n".join(lines) diff --git a/src/orchestrator/compiler/schema_validator.py b/src/orchestrator/compiler/schema_validator.py index 1e3a8aff..527ee368 100644 --- a/src/orchestrator/compiler/schema_validator.py +++ b/src/orchestrator/compiler/schema_validator.py @@ -133,9 +133,20 @@ def _get_default_schema(self) -> Dict[str, Any]: }, "timeout": {"type": "integer", "minimum": 1}, "max_retries": {"type": "integer", "minimum": 0}, - "on_failure": { + # Either a failure policy or a step to jump + # to (#333). Which one is decided by the + # value; `validate_dependencies` checks + # that a non-policy value names a real + # step, so an enum here would reject valid + # routing outright. + "on_failure": {"type": "string"}, + "on_false": { "type": "string", - "enum": ["continue", "fail", "retry", "skip"], + "pattern": "^[a-zA-Z][a-zA-Z0-9_-]*$", + }, + "on_success": { + "type": "string", + "pattern": "^[a-zA-Z][a-zA-Z0-9_-]*$", }, "requires_model": { "oneOf": [ diff --git a/src/orchestrator/compiler/yaml_compiler.py b/src/orchestrator/compiler/yaml_compiler.py index 345fcd42..203a4866 100644 --- a/src/orchestrator/compiler/yaml_compiler.py +++ b/src/orchestrator/compiler/yaml_compiler.py @@ -1346,6 +1346,13 @@ def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> T # Add additional metadata from task definition if "on_failure" in task_def: metadata["on_failure"] = task_def["on_failure"] + + # Control-flow routing (#333). `on_failure` is deliberately not copied + # again here: it is already above, and whether it names a policy or a + # step is decided by its value, not by a second key. + for routing_key in ("on_false", "on_success"): + if routing_key in task_def: + metadata[routing_key] = task_def[routing_key] if "requires_model" in task_def: metadata["requires_model"] = task_def["requires_model"] if "tool" in task_def: diff --git a/src/orchestrator/core/routing.py b/src/orchestrator/core/routing.py new file mode 100644 index 00000000..1b6dd8f1 --- /dev/null +++ b/src/orchestrator/core/routing.py @@ -0,0 +1,62 @@ +"""Control-flow routing between steps. + +A step may name where execution goes next: + + - id: check_topic + action: evaluate_condition + parameters: + condition: "{{ topic | length > 10 }}" + on_false: short_topic_handler + on_success: main_processing + on_failure: error_recovery + +`on_false` applies when the step's own `condition:` evaluates false, so the +step is skipped. `on_success` and `on_failure` apply to how the step itself +ended. Routing jumps forward: steps between the source and the target are +marked skipped, which is the same machinery `goto` already used. + +## The `on_failure` collision + +`on_failure` already existed, meaning a *failure policy*: one of `fail`, +`continue`, `skip`, `retry`. #333 asks for the same key to name a step to jump +to. Both readings are useful and both are now supported, disambiguated by +value: a reserved policy word keeps its policy meaning, anything else is a +step id. + +That is only safe because the ambiguity is refused rather than guessed at -- +naming a step `retry` and routing to it would silently become a policy, so a +pipeline that does so fails at compile time instead. +""" + +from __future__ import annotations + +from typing import FrozenSet + +#: Values of `on_failure` that select a policy rather than a routing target. +FAILURE_POLICIES: FrozenSet[str] = frozenset({"fail", "continue", "skip", "retry"}) + +#: Step keys that name another step to jump to. +ROUTING_KEYS: FrozenSet[str] = frozenset({"on_false", "on_success", "on_failure"}) + + +def is_failure_policy(value: object) -> bool: + """Whether an `on_failure` value selects a policy instead of a step.""" + return isinstance(value, str) and value.strip().lower() in FAILURE_POLICIES + + +def routing_targets(step: dict) -> dict: + """The routing keys of `step` that name a step id, as {key: target}. + + `on_failure` is included only when its value is not a reserved policy. + """ + targets = {} + for key in ("on_false", "on_success"): + value = step.get(key) + if isinstance(value, str) and value.strip(): + targets[key] = value.strip() + + on_failure = step.get("on_failure") + if isinstance(on_failure, str) and on_failure.strip() and not is_failure_policy(on_failure): + targets["on_failure"] = on_failure.strip() + + return targets diff --git a/src/orchestrator/orchestrator.py b/src/orchestrator/orchestrator.py index 96816b6a..03c69265 100644 --- a/src/orchestrator/orchestrator.py +++ b/src/orchestrator/orchestrator.py @@ -12,6 +12,7 @@ from .core.error_handler import ErrorHandler from .core.pipeline import Pipeline from .core.pipeline_result import PipelineResult, StepResult +from .core.routing import is_failure_policy from .core.pipeline_status_tracker import PipelineStatusTracker from .core.pipeline_resume_manager import PipelineResumeManager, ResumeStrategy from .core.resource_allocator import ResourceAllocator @@ -1029,9 +1030,23 @@ async def _process_goto_jumps( # Check recently completed tasks for goto directives for task_id in completed_tasks: task = pipeline.get_task(task_id) - if not task or task.status != TaskStatus.COMPLETED: + if not task: continue - + + # Control-flow routing (#333), which jumps the same way `goto` + # does. Unlike goto it also applies to steps that were skipped or + # that failed, so the status filter cannot come first. + routed = self._routing_target(task, results.get(task_id)) + if routed and routed in pipeline.tasks: + self.logger.info( + "Task %s routing to %s (%s)", task_id, routed, task.status.value + ) + self._skip_tasks_between(pipeline, task_id, routed) + return routed + + if task.status != TaskStatus.COMPLETED: + continue + # Check for goto in task metadata goto_target = task.metadata.get("goto") if goto_target: @@ -1048,6 +1063,34 @@ async def _process_goto_jumps( return None + def _routing_target(self, task: Task, value: Any = None) -> Optional[str]: + """Where this task's outcome says to go next, if anywhere. + + `on_false` fires when the step's own `condition:` was false, which is + recorded as a skip. `on_success` and `on_failure` fire on how the step + itself ended, and "ended badly" is not the same as status FAILED: a + tool returning {"success": False} without raising leaves its task + COMPLETED. StepResult already owns that distinction, so it is reused + here rather than restated -- routing on status alone sent a failing + step down the success path. + + `on_failure` names a step only when its value is not one of the + reserved failure policies -- see core/routing.py. + """ + if task.status is TaskStatus.SKIPPED: + return task.metadata.get("on_false") + + succeeded = StepResult.from_task(task, value).success + + if succeeded: + return task.metadata.get("on_success") + + on_failure = task.metadata.get("on_failure") + if isinstance(on_failure, str) and not is_failure_policy(on_failure): + return on_failure.strip() or None + + return None + def _skip_tasks_between(self, pipeline: Pipeline, from_task: str, to_task: str) -> None: """Skip tasks between a goto source and target. @@ -2059,6 +2102,11 @@ async def _handle_task_failures( task = pipeline.get_task(task_id) failure_policy = task.metadata.get("on_failure", "fail") + # `on_failure` naming a step is routing, not a policy: the run + # continues at that step rather than aborting here (#333). + if not is_failure_policy(failure_policy): + continue + if failure_policy == "continue": # Continue with other tasks continue diff --git a/src/orchestrator/validation/dependency_validator.py b/src/orchestrator/validation/dependency_validator.py index e6f55813..288858ab 100644 --- a/src/orchestrator/validation/dependency_validator.py +++ b/src/orchestrator/validation/dependency_validator.py @@ -23,6 +23,7 @@ nx = None from ..core.exceptions import ValidationError +from ..core.routing import FAILURE_POLICIES, routing_targets logger = logging.getLogger(__name__) @@ -392,7 +393,54 @@ def _validate_dependency_references(self, tasks: List[Dict[str, Any]]) -> List[D dependency_chain=[task_id, task_id], recommendation=f"Remove self-dependency from task '{task_id}'" )) - + + # Control-flow routing targets (#333). A jump to a step that does not + # exist has to fail at compile time with the bad name, not at runtime + # when the executor tries to route into nothing. + for task in tasks: + task_id = task.get("id") + if not task_id: + continue + for key, target in routing_targets(task).items(): + if target not in task_ids: + issues.append(DependencyIssue( + issue_type="missing_routing_target", + severity="error", + message=( + f"Task '{task_id}' has {key}: '{target}', which is " + f"not a step in this pipeline" + ), + involved_tasks=[task_id, target], + recommendation=( + f"Either add a step '{target}' or correct the " + f"{key} of task '{task_id}'" + ), + )) + elif target == task_id: + issues.append(DependencyIssue( + issue_type="self_routing", + severity="error", + message=f"Task '{task_id}' routes {key} to itself", + involved_tasks=[task_id], + recommendation=f"Route {key} to a different step", + )) + + # A step id that is also a failure policy makes `on_failure` ambiguous: + # routing to it cannot be told apart from selecting the policy. Refuse + # rather than silently pick one reading. + for policy_name in sorted(FAILURE_POLICIES & set(task_ids)): + issues.append(DependencyIssue( + issue_type="ambiguous_task_id", + severity="error", + message=( + f"Task id '{policy_name}' collides with the on_failure " + f"policy of the same name, so routing to it could not be " + f"distinguished from selecting the policy" + ), + involved_tasks=[policy_name], + recommendation=f"Rename the step '{policy_name}'", + )) + return issues def _validate_circular_dependencies( diff --git a/tests/test_control_flow_routing.py b/tests/test_control_flow_routing.py new file mode 100644 index 00000000..6a5e06b9 --- /dev/null +++ b/tests/test_control_flow_routing.py @@ -0,0 +1,233 @@ +"""Control-flow routing: `on_false`, `on_success`, `on_failure` (#333). + +A step may say where execution goes next. Routing jumps forward, marking the +steps in between as skipped -- the same machinery `goto` already used, rather +than a second one beside it. + +The awkward part is `on_failure`, which already meant a *failure policy* +(`fail` / `continue` / `skip` / `retry`) and now also names a step to jump to. +Both are supported, disambiguated by value, and the ambiguous case -- a step +whose id *is* a policy word -- is refused at compile time rather than guessed. +""" + +import asyncio +import os +from pathlib import Path + +import pytest + +from orchestrator.compiler.yaml_compiler import YAMLCompiler +from orchestrator.core.exceptions import YAMLCompilerError +from orchestrator.core.routing import FAILURE_POLICIES, is_failure_policy, routing_targets +from orchestrator.models.model_registry import ModelRegistry +from tests.test_infrastructure import MockTestModel, create_test_orchestrator + +pytestmark = [pytest.mark.contract, pytest.mark.e2e] + + +def _compiler(): + registry = ModelRegistry() + registry.register_model(MockTestModel()) + return YAMLCompiler(model_registry=registry) + + +def _run(yaml_content, cwd): + previous = Path.cwd() + os.chdir(cwd) + try: + return asyncio.run( + create_test_orchestrator().execute_yaml(yaml_content=yaml_content, context={}) + ) + finally: + os.chdir(previous) + + +def _write_step(step_id, filename, *, depends_on=None, **extra): + lines = [ + f" - id: {step_id}", + " tool: filesystem", + " action: write", + " parameters:", + f' path: "out/{filename}"', + f' content: "{step_id}-ran"', + ] + if depends_on: + lines.append(f" dependencies: [{depends_on}]") + for key, value in extra.items(): + lines.append(f" {key}: {value}") + return "\n".join(lines) + + +def _conditional_pipeline(condition): + """check -> (middle) -> recover, where `check` routes on_false to recover.""" + return f""" +id: routing_pipeline +name: Routing Pipeline +steps: + - id: check + action: evaluate_condition + condition: "{condition}" + parameters: + condition: "1 == 1" + on_false: recover +{_write_step("middle", "middle.txt", depends_on="check")} +{_write_step("recover", "recover.txt", depends_on="middle")} +""" + + +# --------------------------------------------------------------------------- +# on_false +# --------------------------------------------------------------------------- + +def test_on_false_jumps_and_skips_the_steps_in_between(tmp_path): + result = _run(_conditional_pipeline("false"), tmp_path) + + assert result.steps["check"].status == "skipped" + assert result.steps["middle"].status == "skipped", ( + "the step routed over must be skipped, not run" + ) + assert result.steps["recover"].success is True + + assert not (tmp_path / "out" / "middle.txt").exists(), ( + "a skipped step must not have had its side effect" + ) + assert (tmp_path / "out" / "recover.txt").read_text() == "recover-ran" + + +def test_a_routed_pipeline_still_reports_success(tmp_path): + """Skipping is not failing. A pipeline that routed around a step worked.""" + result = _run(_conditional_pipeline("false"), tmp_path) + + assert result.success is True + assert result.status == "completed" + assert [s.id for s in result.failed_steps] == [] + assert sorted(s.id for s in result.skipped_steps) == ["check", "middle"] + + +def test_on_false_does_not_fire_when_the_condition_holds(tmp_path): + """The branch not taken must actually not be taken.""" + result = _run(_conditional_pipeline("true"), tmp_path) + + assert result.steps["check"].status == "completed" + assert result.steps["middle"].success is True, "the normal path must run" + assert (tmp_path / "out" / "middle.txt").exists() + + +# --------------------------------------------------------------------------- +# on_success +# --------------------------------------------------------------------------- + +def test_on_success_jumps_past_the_intervening_step(tmp_path): + pipeline = f""" +id: on_success_pipeline +name: On Success Pipeline +steps: +{_write_step("first", "first.txt", on_success="last")} +{_write_step("skipped_middle", "middle.txt", depends_on="first")} +{_write_step("last", "last.txt", depends_on="skipped_middle")} +""" + result = _run(pipeline, tmp_path) + + assert result.steps["first"].success is True + assert result.steps["skipped_middle"].status == "skipped" + assert result.steps["last"].success is True + assert not (tmp_path / "out" / "middle.txt").exists() + + +# --------------------------------------------------------------------------- +# on_failure: policy or step id +# --------------------------------------------------------------------------- + +def test_on_failure_naming_a_step_routes_instead_of_aborting(tmp_path): + pipeline = f""" +id: on_failure_routing +name: On Failure Routing +steps: + - id: boom + tool: filesystem + action: read + on_failure: handler + parameters: + path: "/nonexistent/definitely/not/here.txt" +{_write_step("never", "never.txt", depends_on="boom")} +{_write_step("handler", "handler.txt", depends_on="never")} +""" + result = _run(pipeline, tmp_path) + + assert result.steps["boom"].success is False + assert result.steps["handler"].success is True, ( + "routing on failure must reach the handler" + ) + assert not (tmp_path / "out" / "never.txt").exists() + + +@pytest.mark.parametrize("policy", sorted(FAILURE_POLICIES)) +def test_reserved_policy_words_keep_their_policy_meaning(policy): + """`on_failure: continue` is a policy, not a jump to a step called continue.""" + assert is_failure_policy(policy) + assert routing_targets({"id": "s", "on_failure": policy}) == {} + + +def test_a_non_policy_on_failure_is_a_routing_target(): + assert routing_targets({"id": "s", "on_failure": "handler"}) == { + "on_failure": "handler" + } + + +# --------------------------------------------------------------------------- +# compile-time validation +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("key", ["on_false", "on_success", "on_failure"]) +def test_routing_to_a_step_that_does_not_exist_fails_to_compile(key): + """Exit 2 naming the bad target, not a run that jumps into nothing.""" + pipeline = f""" +id: bad_routing +name: Bad Routing +steps: +{_write_step("only", "only.txt", **{key: "nowhere"})} +""" + with pytest.raises(YAMLCompilerError) as excinfo: + asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False)) + + assert "nowhere" in str(excinfo.value) + + +def test_routing_to_itself_fails_to_compile(): + pipeline = f""" +id: self_routing +name: Self Routing +steps: +{_write_step("only", "only.txt", on_success="only")} +""" + with pytest.raises(YAMLCompilerError) as excinfo: + asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False)) + + assert "itself" in str(excinfo.value) + + +def test_a_step_named_after_a_failure_policy_is_refused(): + """The one genuinely ambiguous case, refused rather than guessed at. + + With a step called `retry`, `on_failure: retry` could mean either the + policy or a jump to that step, and no rule can tell them apart. + """ + pipeline = f""" +id: colliding_ids +name: Colliding Ids +steps: +{_write_step("first", "first.txt")} +{_write_step("retry", "retry.txt", depends_on="first")} +""" + with pytest.raises(YAMLCompilerError) as excinfo: + asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False)) + + message = str(excinfo.value) + assert "retry" in message and "collides" in message + + +def test_valid_routing_compiles_and_is_carried_on_the_task(): + pipeline = _conditional_pipeline("false") + compiled = asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False)) + + assert compiled.tasks["check"].metadata["on_false"] == "recover"