diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md index 13ffde6..c64d669 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 ce2a0ca..054e4e0 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 86cfa03..9894f88 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 e46f550..9e44d42 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 1530a1b..affb112 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 fca1524..db1d9b2 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 d136f13..740dad0 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 74d4f50..b92b3a3 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 8e004b2..209294a 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" + )