Make the supported examples prove behaviour, not existence (#104)

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 |

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 }}"

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:

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"] = "<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"] = "<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())

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

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):

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()