From 89c341e068fa0f9e389dcd9bd91e868582c8be51 Mon Sep 17 00:00:00 2001 From: Jeremy Manning Date: Sun, 2 Aug 2026 10:27:35 -0400 Subject: [PATCH] Make the supported examples prove behaviour, not existence (#104) The acceptance suite asserted that a command exited and that some files existed. A badly broken runtime satisfies that, and one did: 04_conditions ran *both* arms of its conditional, and the test asserted both artifacts, pinning the bug in place. 04_conditions now expresses two genuinely exclusive branches converging on a join. `handle_short` was never a branch -- it was the last link of a dependency chain, so the default path fell through into it. The long arm's `on_success` jumps the short arm; the short arm is reached only by `on_false`. The declared output `taken_branch` used to render `{{ out_dir }}`, echoing an input; it now reads a file the branch itself wrote, so it reports the branch. Each example declares a case per branch, and a case now fixes the exit code, every artifact *and its exact contents*, the complete outputs document, which steps completed / were skipped / failed, and the execution levels -- so "alpha and beta run in parallel" is a checked claim. An unexpected artifact is a failure. `checkpoints/` is named as a run side effect and asserted directly, rather than silently ignored. CLI/API equivalence was nearly vacuous: the comparison dropped every step value and the whole outputs document, on a comment claiming values embed absolute paths. Measured across 5 examples x 3 runs, the only fields that ever differ are the timestamps and the execution id; values are byte-identical and paths in them are relative. `PipelineResult.normalized()` already blanked exactly those, so `normalize_result_payload` now serves both the object and the CLI's JSON and the test cannot drift into comparing less than it claims. Two runtime defects surfaced from the stronger assertions: A timed-out command was abandoned rather than killed. `asyncio.wait_for` cancels the wait, not the process, and the timeout handler never referred to the child again: 06_failure_policy left two `sleep` processes alive after the orchestrator had exited. `_reap()` in a `finally` also covers the cancellation raised when a step rather than the command times out, and closing the pipes on a live loop is what stops the unraisable-exception warnings at teardown. Templates dropped the trailing newline of any parameter referencing another step, because that path alone goes through a Jinja environment left on the default `keep_trailing_newline=False`. Literals and parameter references kept theirs. Of 547 blocking tests the only two affected were the ones pinning the truncated byte. Every new test was run against the un-fixed code: the example cases fail 4x on the old 04, the leak test names the surviving pid, the newline test fails on the lost byte. The first draft of the leak test was vacuous -- its marker sat in a shell comment, and `sh -c "sleep 30 # marker"` execs sleep in place and drops it from argv -- so it now tracks process ancestry instead. Blocking suite 541 -> 548 passed, 13 skipped, 0 failed. Unraisable and resource warnings are errors in both affected suites. Co-Authored-By: Claude Opus 5 (1M context) --- docs/adr/0001-product-contract.md | 8 + examples/supported/04_conditions.yaml | 58 ++- examples/supported/README.md | 32 +- src/orchestrator/core/pipeline_result.py | 36 +- src/orchestrator/core/template_manager.py | 3 +- src/orchestrator/tools/system_tools.py | 30 +- tests/test_failure_policy.py | 81 +++- tests/test_supported_examples.py | 469 +++++++++++++++++----- tests/test_unresolved_templates.py | 70 ++++ 9 files changed, 664 insertions(+), 123 deletions(-) diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md index 13ffde61..c64d6695 100644 --- a/docs/adr/0001-product-contract.md +++ b/docs/adr/0001-product-contract.md @@ -162,6 +162,7 @@ exist, or a step routing to itself, is exit 2 naming the offending target. | | Behaviour | |-|-| | `timeout: N` | The step is cancelled after N seconds and fails with `TimeoutError`. `StepResult.timed_out` distinguishes it from an ordinary failure. | +| A timed-out command | Is **killed and reaped**, not merely abandoned. `asyncio.wait_for` cancels the wait rather than the process, so a command that outran its timeout used to keep running after the orchestrator exited. | | `max_retries: N` | Bounds the **total number of attempts**, not the retries beyond the first. `max_retries: 2` is two attempts, so one retry; `0` and `1` are both a single attempt. | | A timeout | Is retried like any other failure, so worst-case wall time is roughly `timeout × max_retries`. | | `on_failure: fail` (default) | Aborts the run — but only for a step that **raised**. | @@ -280,6 +281,13 @@ looks like a template (#153). Rendering is all-or-nothing per string. One undefined reference aborts the whole render, so every marker in that string is reported, not just the bad one. +**What a template renders is what gets written, to the byte.** A parameter +referring to another step used to lose the trailing newline its `content:` +ended with, while a literal or a pipeline-parameter reference kept theirs — +Jinja's `keep_trailing_newline` defaults to false, and only the step-reference +path went through that environment. Content fidelity does not depend on which +kind of variable a template happens to mention. + ## Test layers | Layer | Marker | Network | Secrets | Runs in default CI | diff --git a/examples/supported/04_conditions.yaml b/examples/supported/04_conditions.yaml index ce2a0cac..054e4e0f 100644 --- a/examples/supported/04_conditions.yaml +++ b/examples/supported/04_conditions.yaml @@ -1,12 +1,28 @@ -# Supported example 04: conditional execution and control-flow routing. +# Supported example 04: exclusive conditional branches and control-flow routing. # -# `check_size` runs only when its `condition:` holds. When the condition is -# false the step is skipped and `on_false` routes execution to the recovery -# branch, marking the steps in between as skipped. Skipping is not failing: -# this pipeline succeeds either way. +# Two branches, and exactly one of them runs. +# +# check_size --(condition false)--> handle_short -> mark_short -\ +# | >-- summarise +# \--(condition true)--> handle_long ---------(on_success)--/ +# +# `check_size` runs only while its `condition:` holds. When the condition is +# false the step is skipped and `on_false` routes to `handle_short`, marking +# `handle_long` skipped on the way past. When it holds, `handle_long` runs and +# its `on_success` routes to `summarise`, marking the whole short branch +# skipped. Routing only ever jumps *forward*, so the branch a run did not take +# is skipped rather than executed. +# +# The two branches therefore write mutually exclusive artifacts -- `long.txt` +# or `short.txt`, never both -- and agree on one file, `branch.txt`, which +# names the branch that ran. `summarise` reads that file, so the declared +# output `taken_branch` reports real branch behaviour rather than echoing an +# input back. +# +# Skipping is not failing: this pipeline succeeds down either branch. id: conditions name: Conditions and Routing -description: Conditional execution with on_false routing and skipped steps +description: Two exclusive branches joined by a converging summary step version: "1.0.0" parameters: @@ -18,6 +34,7 @@ parameters: default: "./output" steps: + # The gate. Skipped when the condition is false, which fires `on_false`. - id: check_size tool: filesystem action: write @@ -27,14 +44,19 @@ steps: path: "{{ out_dir }}/long.txt" content: "long: {{ content }}" + # Long branch tail. `on_success` jumps the short branch entirely, which is + # what makes the two branches exclusive rather than sequential. - id: handle_long tool: filesystem - action: read + action: write + on_success: summarise parameters: - path: "{{ out_dir }}/long.txt" + path: "{{ out_dir }}/branch.txt" + content: "long" dependencies: - check_size + # Short branch. Reached only by the `on_false` jump above. - id: handle_short tool: filesystem action: write @@ -44,5 +66,23 @@ steps: dependencies: - handle_long + - id: mark_short + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/branch.txt" + content: "short" + dependencies: + - handle_short + + # The join. Both branches converge here, so this step runs either way. + - id: summarise + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/branch.txt" + dependencies: + - mark_short + outputs: - taken_branch: "{{ out_dir }}" + taken_branch: "{{ summarise.result.content }}" diff --git a/examples/supported/README.md b/examples/supported/README.md index 86cfa031..9894f883 100644 --- a/examples/supported/README.md +++ b/examples/supported/README.md @@ -5,18 +5,26 @@ Every pipeline in this directory is **tested as product behaviour**. Each one: 1. compiles under `orchestrator validate`; 2. executes through the CLI; 3. executes through the Python API; -4. produces the same normalised result document from both surfaces; -5. produces the artifacts and declared outputs its expectations record. +4. produces the same normalised result document from both surfaces — + *whole documents*, including every step value and the declared outputs, + with only wall-clock times and the execution id blanked; +5. produces exactly the artifacts its case records, **with their exact + contents**, and no others; +6. records exactly which steps completed, which were skipped and which + failed, on every branch it has. -`tests/test_supported_examples.py` enforces all five, and fails if a file is -added here without an entry declaring what it should do — so the set cannot -grow past its coverage. +`tests/test_supported_examples.py` enforces all six, and fails if a file is +added here without a case declaring what it should do — so the set cannot grow +past its coverage. + +A pipeline with a branch declares one case per branch, so the arm a default +run does not take is still executed in CI. | Example | Demonstrates | |-|-| | `01_hello_filesystem.yaml` | typed parameters, templating, dependencies, declared outputs | | `02_parallel_fanout_fanin.yaml` | independent steps sharing an execution level, joined by a dependent step | -| `04_conditions.yaml` | `condition:`, `on_false` routing, skipped steps | +| `04_conditions.yaml` | `condition:`, `on_false`/`on_success` routing, two **exclusive** branches converging on a join | | `06_failure_policy.yaml` | `timeout`, bounded retry, `on_failure: continue`, honest exit code | | `07_templates_and_outputs.yaml` | step-result references and declared outputs | @@ -35,6 +43,18 @@ orchestrator run examples/supported/01_hello_filesystem.yaml `06_failure_policy.yaml` exits **1** on purpose: it contains a step that fails, and the run reports that rather than hiding it. +`04_conditions.yaml` takes its long branch by default. Force the other one +with a shorter input, and note that the two branches produce different files: + +```bash +orchestrator run examples/supported/04_conditions.yaml # long.txt +orchestrator run examples/supported/04_conditions.yaml -i content=hi # short.txt +``` + +Every run also writes a checkpoint under `./checkpoints/` in the working +directory. That is a side effect of running any pipeline rather than of these +examples. + ## Not here yet Numbering follows the planned set, so gaps are deliberate rather than lost: diff --git a/src/orchestrator/core/pipeline_result.py b/src/orchestrator/core/pipeline_result.py index e46f5501..9e44d429 100644 --- a/src/orchestrator/core/pipeline_result.py +++ b/src/orchestrator/core/pipeline_result.py @@ -20,12 +20,39 @@ from __future__ import annotations from collections.abc import Mapping +from copy import deepcopy from dataclasses import dataclass, field from typing import Any, Dict, Iterator, List, Optional, Tuple from .task import Task, TaskStatus +#: Fields whose values are a property of *when* a run happened rather than of +#: what it did. Two runs of the same pipeline differ here and nowhere else. +RUN_SPECIFIC_FIELDS: Tuple[str, ...] = ("started_at", "completed_at", "duration") + + +def normalize_result_payload(data: Dict[str, Any]) -> Dict[str, Any]: + """Blank the run-specific fields of a serialised result, leaving the rest. + + The CLI emits `to_dict()` as JSON and the Python API returns a + `PipelineResult`, so comparing the two surfaces means comparing a plain + dict against an object. This works on the dict form, which both can reach, + and is the *only* definition of what "run-specific" means -- a second + definition living in a test is how a comparison quietly stops comparing + anything. Everything not blanked here, `outputs` and every step `value` + included, is behaviour the two surfaces must agree on exactly. + """ + data = deepcopy(data) + data["execution_id"] = "" + for key in RUN_SPECIFIC_FIELDS: + data[key] = None + for step in data.get("steps", {}).values(): + for key in RUN_SPECIFIC_FIELDS: + step[key] = None + return data + + 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: @@ -231,11 +258,4 @@ def normalized(self) -> Dict[str, Any]: 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 + return normalize_result_payload(self.to_dict()) diff --git a/src/orchestrator/core/template_manager.py b/src/orchestrator/core/template_manager.py index 1530a1bb..affb112c 100644 --- a/src/orchestrator/core/template_manager.py +++ b/src/orchestrator/core/template_manager.py @@ -103,7 +103,8 @@ def __init__(self, debug_mode: bool = False, file_inclusion_processor: Optional[ self.env = Environment( undefined=StrictUndefined, trim_blocks=True, - lstrip_blocks=True + lstrip_blocks=True, + keep_trailing_newline=True, ) # Add custom filters diff --git a/src/orchestrator/tools/system_tools.py b/src/orchestrator/tools/system_tools.py index fca15242..db1d9b23 100644 --- a/src/orchestrator/tools/system_tools.py +++ b/src/orchestrator/tools/system_tools.py @@ -3,11 +3,34 @@ import asyncio import shutil from pathlib import Path -from typing import Any, Dict +from typing import Any, Dict, Optional from .base import Tool +async def _reap(process: Optional[asyncio.subprocess.Process]) -> None: + """Kill a child that is still running, and wait for it. + + `asyncio.wait_for` cancels the *wait*, not the process. A command that + outran its timeout is therefore still running when the tool returns, and + nothing else ever refers to it again: running `06_failure_policy.yaml`, + whose step runs `sleep 5` under a 1 second timeout with two attempts, left + two `sleep` processes alive after the orchestrator itself had exited. + + Killing and then awaiting the child also lets its pipe transports close on + the running loop, instead of leaving them to `__del__` at interpreter + teardown -- which is what surfaces as `PytestUnraisableExceptionWarning` + once the loop is gone. + """ + if process is None or process.returncode is not None: + return + try: + process.kill() + except ProcessLookupError: + return # exited between the check and the kill + await process.wait() + + class TerminalTool(Tool): """Tool for executing terminal/shell commands.""" @@ -42,6 +65,7 @@ async def _execute_impl(self, **kwargs) -> Dict[str, Any]: if command.startswith("!"): command = command[1:] + process: Optional[asyncio.subprocess.Process] = None try: # Create working directory if it doesn't exist Path(working_dir).mkdir(parents=True, exist_ok=True) @@ -98,6 +122,10 @@ async def _execute_impl(self, **kwargs) -> Dict[str, Any]: "success": False, "working_dir": working_dir, } + finally: + # Every exit path, including the cancellation the orchestrator + # raises when a *step* times out rather than the command. + await _reap(process) class FileSystemTool(Tool): diff --git a/tests/test_failure_policy.py b/tests/test_failure_policy.py index d136f13d..740dad06 100644 --- a/tests/test_failure_policy.py +++ b/tests/test_failure_policy.py @@ -25,7 +25,14 @@ from orchestrator.models.model_registry import ModelRegistry from tests.test_infrastructure import create_test_orchestrator -pytestmark = [pytest.mark.contract, pytest.mark.e2e] +pytestmark = [ + pytest.mark.contract, + pytest.mark.e2e, + # Timeouts are where subprocesses get abandoned, and an abandoned child + # reports itself only as an unraisable exception at teardown. Promote it. + pytest.mark.filterwarnings("error::pytest.PytestUnraisableExceptionWarning"), + pytest.mark.filterwarnings("error::ResourceWarning"), +] def _run(yaml_content, cwd): @@ -238,3 +245,75 @@ def test_model_selection_fails_closed_rather_than_substituting(): with pytest.raises(NoEligibleModelsError): asyncio.run(registry.select_model({"tasks": ["generate"]})) + + +def test_a_timed_out_command_leaves_no_surviving_process(tmp_path): + """A command that outran its timeout must be dead, not merely abandoned. + + `asyncio.wait_for` cancels the *wait*, not the process. Before this was + fixed the tool returned its timeout result and never referred to the child + again, so the command kept running: `06_failure_policy.yaml`, whose step + runs `sleep 5` under a one second timeout with two attempts, left two + `sleep` processes alive after the orchestrator had exited. + + The same missing cleanup is what leaves pipe transports for `__del__` to + close after the event loop is gone, which surfaces as + PytestUnraisableExceptionWarning at teardown. + """ + import psutil + + # Identify the leak by process *ancestry* rather than by matching a + # sentinel in the command line. Two simpler approaches do not work: a + # shell comment vanishes because `sh -c "sleep 30 # marker"` execs `sleep` + # in place and the replaced argv keeps neither the comment nor the shell, + # and a renamed copy of `sleep` is SIGKILLed by macOS code signing. What + # the tool owns is its child, so that is what this looks at. + pipeline = """ +id: leak_probe +name: Leak Probe +steps: + - id: slow + tool: terminal + action: execute + timeout: 1 + max_retries: 1 + on_failure: continue + parameters: + command: "sleep 30" +""" + + me = psutil.Process() + + def live_children(): + """Descendants that are actually running, not already-reaped zombies.""" + alive = [] + for child in me.children(recursive=True): + try: + if child.is_running() and child.status() != psutil.STATUS_ZOMBIE: + alive.append(child) + except psutil.NoSuchProcess: + pass # exited while we were looking at it + return alive + + before = {child.pid for child in live_children()} + + try: + result = _run(pipeline, tmp_path) + + assert result.steps["slow"].success is False, ( + "the step was supposed to time out" + ) + assert result.steps["slow"].timed_out is True + + leaked = [child for child in live_children() if child.pid not in before] + assert leaked == [], ( + f"a command that timed out is still running after the pipeline " + f"finished -- the child was abandoned rather than killed: " + f"{[(c.pid, ' '.join(c.cmdline())) for c in leaked]}" + ) + finally: + # Never leave stray processes behind, even when the assertion above is + # the thing that failed. + for child in live_children(): + if child.pid not in before: + child.kill() diff --git a/tests/test_supported_examples.py b/tests/test_supported_examples.py index 74d4f504..b92b3a38 100644 --- a/tests/test_supported_examples.py +++ b/tests/test_supported_examples.py @@ -6,55 +6,229 @@ `examples/supported/` is the answer: a small set that is *tested as product behaviour*. Every file here must validate, must execute through both the CLI -and the Python API, and the two must agree as whole normalised documents. A -pipeline that cannot do all of that does not belong in this directory. - -Adding a file to examples/supported/ without an entry in EXPECTATIONS fails, -so the set cannot grow silently past its coverage. +and the Python API, and the two must agree as whole normalised documents. + +What "tested" means here is deliberately strict, because the first version of +this file was not. It asserted that a command exited and that some files +existed, which a badly broken runtime also satisfies -- and it did: it pinned +04_conditions while that example ran *both* arms of its conditional, and it +compared the CLI against the API after deleting every step value and the whole +outputs document, so the two surfaces could have disagreed about everything +that matters and still passed. + +So a case here declares, exactly: + +* the exit code, +* every artifact the run may create *and its full contents* -- an unexpected + file is a failure, not a shrug, +* the complete declared `outputs` document, +* which steps completed, which were skipped, and which failed, +* the execution levels, so "these steps run in parallel" is a checked claim, +* per-step fields where a step's *manner* of failing is the point + (timeout classification, retry count). + +A pipeline with a branch declares a case per branch, so the arm a default run +does not take is still exercised. + +Adding a file to examples/supported/ without a case fails, so the set cannot +grow past its coverage. """ import json import os import subprocess import sys +from dataclasses import dataclass, field from pathlib import Path +from typing import Any, Dict, Optional, Tuple import pytest +from orchestrator.core.pipeline_result import normalize_result_payload from tests.test_infrastructure import create_test_orchestrator -pytestmark = [pytest.mark.contract, pytest.mark.e2e] +pytestmark = [ + pytest.mark.contract, + pytest.mark.e2e, + # An abandoned subprocess or an unclosed transport does not fail anything; + # it surfaces at teardown as an unraisable exception, which scrolls past. + # This suite runs real pipelines with real side effects, so it is exactly + # where that class of bug appears, and here it is an error. + pytest.mark.filterwarnings("error::pytest.PytestUnraisableExceptionWarning"), + pytest.mark.filterwarnings("error::ResourceWarning"), +] SUPPORTED_DIR = Path(__file__).parent.parent / "examples" / "supported" EXAMPLES = sorted(SUPPORTED_DIR.glob("*.yaml")) -# Exit code 1 is not a failure of the example: 06 deliberately contains a step -# that fails, and the run is supposed to report that honestly. -EXPECTATIONS = { - "01_hello_filesystem.yaml": { - "exit_code": 0, - "artifacts": ["output/greeting.txt"], - "outputs": {"greeting_text": "hello, world"}, - }, - "02_parallel_fanout_fanin.yaml": { - "exit_code": 0, - "artifacts": ["output/alpha.txt", "output/beta.txt", "output/joined.txt"], - "artifact_contents": {"output/joined.txt": "alpha-value+beta-value"}, - }, - "04_conditions.yaml": { - "exit_code": 0, - "artifacts": ["output/long.txt", "output/short.txt"], - }, - "06_failure_policy.yaml": { - "exit_code": 1, - "artifacts": ["output/before.txt", "output/after.txt"], - "failed_steps": ["flaky"], - }, - "07_templates_and_outputs.yaml": { - "exit_code": 0, - "artifacts": ["output/header.md", "output/report.md"], - }, -} +#: Running any pipeline writes a checkpoint under ./checkpoints/. That is a +#: real side effect of a run rather than an artifact of any one example, so it +#: is named here and excluded from the artifact comparison. Naming it is the +#: point: "no unexpected artifacts" then means exactly that, instead of +#: "no artifacts beyond the ones we happened to remember". +SIDE_EFFECT_DIRS = ("checkpoints",) + +#: Files the harness itself drops in the run directory. +HARNESS_FILES = ("stdout.json", "stderr.log") + + +@dataclass(frozen=True) +class Case: + """One run of one example, and everything observable about it.""" + + example: str + #: Distinguishes branches of the same example in test ids. + variant: str + exit_code: int + #: Relative path -> exact file contents. This is the complete set; any + #: other file under the run directory fails the case. + artifacts: Dict[str, str] + #: The complete declared `outputs` document, compared exactly. + outputs: Dict[str, Any] + completed: Tuple[str, ...] + skipped: Tuple[str, ...] = () + failed: Tuple[str, ...] = () + #: Pipeline parameters, passed as `-i k=v` to the CLI and as `context=` to + #: the API -- the two surfaces must reach the same branch the same way. + inputs: Dict[str, Any] = field(default_factory=dict) + #: Steps grouped by dependency level. Steps sharing a level may run + #: concurrently; this is what makes "parallel fan-out" a checked claim. + levels: Optional[Tuple[Tuple[str, ...], ...]] = None + #: step id -> {field: expected value} from the step's serialised form. + step_fields: Dict[str, Dict[str, Any]] = field(default_factory=dict) + + @property + def id(self) -> str: + return f"{self.example}[{self.variant}]" + + +LONG_TEXT = "a reasonably long piece of text" +HEADER_MD = "# Quarterly Report\n\nBy Orchestrator\n" + +CASES = ( + Case( + example="01_hello_filesystem.yaml", + variant="default", + exit_code=0, + artifacts={"output/greeting.txt": "hello, world"}, + outputs={"greeting_text": "hello, world"}, + completed=("write_greeting", "read_back"), + levels=(("write_greeting",), ("read_back",)), + ), + Case( + example="02_parallel_fanout_fanin.yaml", + variant="default", + exit_code=0, + artifacts={ + "output/alpha.txt": "alpha-value", + "output/beta.txt": "beta-value", + "output/joined.txt": "alpha-value+beta-value", + }, + outputs={"joined": "./output/joined.txt"}, + completed=("alpha", "beta", "read_alpha", "read_beta", "join"), + # `alpha` and `beta` share level 0, which is the entire claim the + # example makes. `join` is alone on the last level, strictly after + # both reads. + levels=( + ("alpha", "beta"), + ("read_alpha", "read_beta"), + ("join",), + ), + ), + # 04 declares one case per branch. The default run takes the long arm, so + # without the second case the short arm would never execute in CI. + Case( + example="04_conditions.yaml", + variant="long-branch", + exit_code=0, + artifacts={ + "output/long.txt": f"long: {LONG_TEXT}", + "output/branch.txt": "long", + }, + outputs={"taken_branch": "long"}, + completed=("check_size", "handle_long", "summarise"), + # The short arm must not merely be absent from the artifacts: it must + # be recorded as skipped. Those are different failures. + skipped=("handle_short", "mark_short"), + levels=( + ("check_size",), + ("handle_long",), + ("handle_short",), + ("mark_short",), + ("summarise",), + ), + ), + Case( + example="04_conditions.yaml", + variant="short-branch", + exit_code=0, + inputs={"content": "hi"}, + artifacts={ + "output/short.txt": "short: hi", + "output/branch.txt": "short", + }, + outputs={"taken_branch": "short"}, + completed=("handle_short", "mark_short", "summarise"), + skipped=("check_size", "handle_long"), + levels=( + ("check_size",), + ("handle_long",), + ("handle_short",), + ("mark_short",), + ("summarise",), + ), + ), + Case( + example="06_failure_policy.yaml", + variant="default", + # Not a failure of the example: 06 deliberately contains a step that + # fails, and the run is supposed to report that honestly. + exit_code=1, + artifacts={ + "output/before.txt": "this step succeeds", + "output/after.txt": "the run continued", + }, + outputs={}, + completed=("before", "after"), + failed=("flaky",), + levels=(("before",), ("flaky",), ("after",)), + # *How* `flaky` failed is the example's subject. A step that failed + # for some other reason, or that never retried, would still produce + # the artifacts above. + step_fields={ + "flaky": { + "success": False, + "error_type": "TimeoutError", + "timed_out": True, + # `max_retries: 2` bounds attempts, so two attempts is one + # retry. See the note in the example. + "retries": 1, + } + }, + ), + Case( + example="07_templates_and_outputs.yaml", + variant="default", + exit_code=0, + artifacts={ + "output/header.md": HEADER_MD, + # Both files keep the trailing newline their `content:` ends with. + # They did not always: a parameter containing a *step-result* + # reference was rendered by a Jinja environment left on its + # default `keep_trailing_newline=False`, so this file -- and only + # this file, because it is the only one referring to another step + # -- lost its last byte. Asserting contents exactly is what + # surfaced it; asserting that the file merely existed did not. + "output/report.md": HEADER_MD + "\nBody of the report.\n", + }, + outputs={ + "report_path": "./output/report.md", + "header_text": HEADER_MD, + }, + completed=("write_header", "read_header", "write_report"), + levels=(("write_header",), ("read_header",), ("write_report",)), + ), +) def _run_cli(args, cwd): @@ -74,20 +248,103 @@ def _run_cli(args, cwd): ) -def _normalise(payload): - """Drop the fields that legitimately differ between two runs.""" - payload = json.loads(json.dumps(payload)) - 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 - # Retry timing varies, and a step's value can embed an absolute path - # that differs between the CLI and API working directories. - step.pop("value", None) - payload.pop("outputs", None) - return payload +def _cli_inputs(case: Case): + """`inputs` as repeated `-i key=value` arguments.""" + args = [] + for key, value in case.inputs.items(): + rendered = value if isinstance(value, str) else json.dumps(value) + args += ["-i", f"{key}={rendered}"] + return args + + +def _payload(stdout: str) -> Dict[str, Any]: + """The result document the CLI prints to stdout.""" + assert stdout.lstrip().startswith("{"), ( + f"expected the CLI to print a JSON document, got: {stdout[:200]!r}" + ) + return json.loads(stdout) + + +def _artifacts(run_dir: Path) -> Dict[str, str]: + """Every file a run left behind, minus known side effects.""" + found = {} + for path in sorted(run_dir.rglob("*")): + if not path.is_file(): + continue + relative = path.relative_to(run_dir) + if relative.parts[0] in SIDE_EFFECT_DIRS or relative.name in HARNESS_FILES: + continue + found[relative.as_posix()] = path.read_text() + return found + + +def _assert_case_holds(case: Case, payload: Dict[str, Any], run_dir: Path): + """Everything a case declares, checked against one run of it.""" + steps = payload["steps"] + + by_status = {} + for step_id, step in steps.items(): + by_status.setdefault(step["status"], []).append(step_id) + + assert sorted(by_status.get("completed", [])) == sorted(case.completed), ( + f"{case.id}: completed steps were {sorted(by_status.get('completed', []))}, " + f"expected {sorted(case.completed)}" + ) + assert sorted(by_status.get("skipped", [])) == sorted(case.skipped), ( + f"{case.id}: skipped steps were {sorted(by_status.get('skipped', []))}, " + f"expected {sorted(case.skipped)}" + ) + assert sorted(by_status.get("failed", [])) == sorted(case.failed), ( + f"{case.id}: failed steps were {sorted(by_status.get('failed', []))}, " + f"expected {sorted(case.failed)}" + ) + + # A step can fail *without raising*, leaving it COMPLETED. Checking status + # alone would call that a success, so check the flag too. + for step_id in case.completed: + assert steps[step_id]["success"] is True, ( + f"{case.id}: step {step_id} completed but reported success=False: " + f"{steps[step_id].get('error')!r}" + ) + for step_id in case.failed: + assert steps[step_id]["success"] is False, ( + f"{case.id}: step {step_id} was expected to fail" + ) + + assert payload["success"] is (case.exit_code == 0), ( + f"{case.id}: success={payload['success']} disagrees with the expected " + f"exit code {case.exit_code}" + ) + + assert payload["outputs"] == case.outputs, ( + f"{case.id}: declared outputs were {payload['outputs']!r}, " + f"expected {case.outputs!r}" + ) + + if case.levels is not None: + actual = tuple(tuple(sorted(level)) for level in payload["execution_levels"]) + expected = tuple(tuple(sorted(level)) for level in case.levels) + assert actual == expected, ( + f"{case.id}: execution levels were {actual}, expected {expected}" + ) + + for step_id, fields in case.step_fields.items(): + for name, value in fields.items(): + assert steps[step_id][name] == value, ( + f"{case.id}: step {step_id}.{name} was " + f"{steps[step_id][name]!r}, expected {value!r}" + ) + + found = _artifacts(run_dir) + assert set(found) == set(case.artifacts), ( + f"{case.id}: artifacts were {sorted(found)}, expected " + f"{sorted(case.artifacts)}" + ) + for relative, content in case.artifacts.items(): + assert found[relative] == content, ( + f"{case.id}: {relative} contained {found[relative]!r}, " + f"expected {content!r}" + ) def test_the_supported_directory_is_not_empty(): @@ -95,14 +352,23 @@ def test_the_supported_directory_is_not_empty(): @pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) -def test_every_supported_example_has_declared_expectations(example): +def test_every_supported_example_has_a_case(example): """Coverage cannot lag behind the directory.""" - assert example.name in EXPECTATIONS, ( - f"{example.name} is in examples/supported/ but has no entry in " - f"EXPECTATIONS, so nothing asserts what it should do" + assert any(case.example == example.name for case in CASES), ( + f"{example.name} is in examples/supported/ but no Case declares what " + f"it should do, so nothing asserts its behaviour" ) +def test_every_case_names_a_real_example(): + """And the table cannot outlive the files it describes.""" + names = {example.name for example in EXAMPLES} + for case in CASES: + assert case.example in names, ( + f"case {case.id} names {case.example}, which is not in {SUPPORTED_DIR}" + ) + + @pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) def test_every_supported_example_validates(example, tmp_path): result = _run_cli(["validate", str(example)], cwd=tmp_path) @@ -112,85 +378,94 @@ def test_every_supported_example_validates(example, tmp_path): ) -@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) -def test_every_supported_example_runs_through_the_cli(example, tmp_path): - expected = EXPECTATIONS[example.name] +@pytest.mark.parametrize("case", CASES, ids=lambda c: c.id) +def test_the_cli_run_matches_the_case(case, tmp_path): + example = SUPPORTED_DIR / case.example - result = _run_cli(["run", str(example)], cwd=tmp_path) + result = _run_cli(["run", str(example), *_cli_inputs(case)], cwd=tmp_path) - assert result.returncode == expected["exit_code"], ( - f"{example.name} exited {result.returncode}, expected " - f"{expected['exit_code']}\n{result.stdout[-2000:]}\n{result.stderr[-2000:]}" + assert result.returncode == case.exit_code, ( + f"{case.id} exited {result.returncode}, expected {case.exit_code}\n" + f"{result.stdout[-2000:]}\n{result.stderr[-2000:]}" ) + _assert_case_holds(case, _payload(result.stdout), tmp_path) - for relative in expected["artifacts"]: - assert (tmp_path / relative).is_file(), ( - f"{example.name} did not produce {relative}" - ) - for relative, content in expected.get("artifact_contents", {}).items(): - assert (tmp_path / relative).read_text().strip() == content - - payload = json.loads(result.stdout[result.stdout.index("{"):]) - for step_id in expected.get("failed_steps", []): - assert payload["steps"][step_id]["success"] is False, ( - f"{example.name}: step {step_id} was expected to fail" - ) - - -@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) -def test_every_supported_example_runs_through_the_python_api(example, tmp_path): +@pytest.mark.parametrize("case", CASES, ids=lambda c: c.id) +def test_the_python_api_run_matches_the_case(case, tmp_path): import asyncio - expected = EXPECTATIONS[example.name] + example = SUPPORTED_DIR / case.example + previous = Path.cwd() os.chdir(tmp_path) try: - orchestrator = create_test_orchestrator() result = asyncio.run( - orchestrator.execute_yaml(yaml_content=example.read_text(), context={}) + create_test_orchestrator().execute_yaml( + yaml_content=example.read_text(), context=dict(case.inputs) + ) ) finally: os.chdir(previous) - assert result.success is (expected["exit_code"] == 0), ( - f"{example.name}: success={result.success} disagrees with the " - f"expected exit code {expected['exit_code']}" - ) - for relative in expected["artifacts"]: - assert (tmp_path / relative).is_file() + _assert_case_holds(case, result.to_dict(), tmp_path) - for name, value in expected.get("outputs", {}).items(): - assert result.outputs.get(name) == value, ( - f"{example.name}: declared output {name!r} was " - f"{result.outputs.get(name)!r}, expected {value!r}" - ) +@pytest.mark.parametrize("case", CASES, ids=lambda c: c.id) +def test_the_cli_and_the_api_agree(case, tmp_path): + """Whole normalised documents, including every step value and output. -@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) -def test_the_cli_and_the_api_agree_on_every_supported_example(example, tmp_path): - """Whole normalised documents, not selected nested values.""" + The only thing blanked is what `normalize_result_payload` blanks -- wall + clock times and the execution id -- and it is blanked by the *same* + function the runtime uses, so this test cannot drift into comparing less + than it claims to. + """ import asyncio + example = SUPPORTED_DIR / case.example cli_dir = tmp_path / "cli" api_dir = tmp_path / "api" cli_dir.mkdir() api_dir.mkdir() - cli_result = _run_cli(["run", str(example)], cwd=cli_dir) - from_cli = json.loads(cli_result.stdout[cli_result.stdout.index("{"):]) + cli_result = _run_cli(["run", str(example), *_cli_inputs(case)], cwd=cli_dir) + from_cli = normalize_result_payload(_payload(cli_result.stdout)) previous = Path.cwd() os.chdir(api_dir) try: from_api = asyncio.run( create_test_orchestrator().execute_yaml( - yaml_content=example.read_text(), context={} + yaml_content=example.read_text(), context=dict(case.inputs) ) - ).to_dict() + ).normalized() finally: os.chdir(previous) - assert _normalise(from_cli) == _normalise(from_api), ( - f"{example.name}: the CLI and the Python API disagree" + assert from_cli == from_api, f"{case.id}: the CLI and the Python API disagree" + + # The two surfaces must also leave the same files behind, which the + # document comparison alone would not catch. + assert _artifacts(cli_dir) == _artifacts(api_dir), ( + f"{case.id}: the CLI and the Python API produced different artifacts" + ) + + +def test_running_a_pipeline_writes_a_checkpoint(tmp_path): + """The side effect the artifact comparison excludes, asserted directly. + + `_artifacts` skips ./checkpoints/ so that an unexpected *output* file is a + test failure. That exclusion is only honest if something checks the + excluded thing exists, otherwise the runtime could stop checkpointing and + no test would notice. + """ + example = SUPPORTED_DIR / "01_hello_filesystem.yaml" + + result = _run_cli(["run", str(example)], cwd=tmp_path) + assert result.returncode == 0 + + checkpoints = sorted((tmp_path / "checkpoints").glob("*.json")) + assert checkpoints, "a run left no checkpoint behind" + assert checkpoints[0].name.startswith("hello_filesystem_"), ( + f"checkpoint {checkpoints[0].name} is not named after the pipeline" ) diff --git a/tests/test_unresolved_templates.py b/tests/test_unresolved_templates.py index 8e004b20..209294ae 100644 --- a/tests/test_unresolved_templates.py +++ b/tests/test_unresolved_templates.py @@ -242,3 +242,73 @@ def test_find_template_markers(value, expected): def test_find_template_markers_spans_newlines(): """A multi-line block is still one marker, not two halves of nothing.""" assert _find_template_markers("{{ a\n + b }}") == ["{{ a\n + b }}"] + + +def test_a_template_writes_exactly_what_it_says_including_the_last_newline(tmp_path): + """Rendering must not quietly eat the final byte. + + Three parameters that differ only in what they interpolate must agree on + their trailing newline. They did not: a parameter referring to another + *step* was rendered by a Jinja environment left on its default + `keep_trailing_newline=False`, so it lost the newline its `content:` ended + with, while a literal and a pipeline-parameter reference kept theirs. + + Measured before the fix, from one run of the pipeline below: + + a_literal.txt = 'literal\n' + b_parameter.txt = 'param W\n' + c_step_reference.txt = 'step 8' <- one byte short + + Silent truncation of written content is worth a test of its own: nothing + fails, nothing warns, and the file is simply wrong. + """ + pipeline = tmp_path / "newlines.yaml" + pipeline.write_text( + """ +id: newline_fidelity +name: Newline Fidelity +parameters: + word: + type: string + default: "W" +steps: + - id: literal + tool: filesystem + action: write + parameters: + path: "./a_literal.txt" + content: "literal\\n" + + - id: from_parameter + tool: filesystem + action: write + parameters: + path: "./b_parameter.txt" + content: "param {{ word }}\\n" + + - id: reader + tool: filesystem + action: read + parameters: + path: "./b_parameter.txt" + dependencies: [from_parameter] + + - id: from_step + tool: filesystem + action: write + parameters: + path: "./c_step_reference.txt" + content: "step {{ reader.result.size }}\\n" + dependencies: [reader] +""" + ) + + result = _run_cli(["run", str(pipeline)], cwd=tmp_path) + assert result.returncode == 0, f"{result.stdout}\n{result.stderr}" + + assert (tmp_path / "a_literal.txt").read_text() == "literal\n" + assert (tmp_path / "b_parameter.txt").read_text() == "param W\n" + assert (tmp_path / "c_step_reference.txt").read_text() == "step 8\n", ( + "a parameter that references another step lost the trailing newline " + "its content ended with" + )