From 267e2c33dc9c9094df5c104952be24eb5c93eb78 Mon Sep 17 00:00:00 2001 From: Jeremy Manning Date: Sat, 1 Aug 2026 20:44:15 -0400 Subject: [PATCH] Pin timeout and retry semantics; establish that fallback fails closed The last unpinned part of the execution contract. Everything here was already implemented and none of it was asserted anywhere, so none of it was safe to write down. Measuring it found two defects and one thing I had assumed was missing but is deliberately absent. Retry was double-counted. `Task.fail()` increments retry_count, and it was called twice for one failure: once by the retry handler in _execute_task_with_resources, again when the gathered result was processed. A step with `max_retries: 0` therefore reported one retry. The outer call now only records a failure the inner path has not already recorded. With that fixed, the measured semantics are: `max_retries` bounds the total number of ATTEMPTS, not retries beyond the first. `max_retries: 2` is two attempts and one retry; 0 and 1 are both a single attempt. The name and the behaviour disagree. Changing it would shift every existing pipeline's retry budget, so the behaviour is pinned by tests and the mismatch recorded rather than silently corrected. Timeouts are enforced with asyncio.wait_for and are retried like any other failure, so worst-case wall time is roughly timeout x max_retries -- measured, not assumed: 1s timeout with max_retries 2 takes ~2.7s. StepResult.timed_out distinguishes a timeout from an ordinary failure, because "raise the timeout" and "fix the step" are different responses. There is no model fallback on the canonical path, and that is correct rather than an omission. ModelRegistry.select_model raises NoEligibleModelsError instead of substituting a model the pipeline did not ask for; models/model_selector.py has fallback strategies but nothing on the canonical path uses it. Quietly selecting an unrequested model is exactly how a cost-control policy gets bypassed, so this is pinned as a test rather than "fixed". Fallback events had no emission point because fallback does not happen. A fail-fast gap is recorded rather than closed. `fail` is the default policy but fires only for a step that RAISED; a tool returning {"success": False} without raising leaves its task COMPLETED, so the run continues past a step the author asked to abort on. Making the policy consult StepResult.success closes it in one line and was tried -- and reverted, because the policy aborts by raising, so the run then produced no result document at all, discarding the trace exactly when it is most wanted. Two golden tests caught that immediately. Closing it properly means the execution loop stops scheduling rather than throwing, and still returns a PipelineResult. ADR 0001 records it, and a test pins the current behaviour so closing it is a visible change. Blocking suite 506 -> 515 passed, 13 skipped, 0 failed, 0 xfailed. Collection 3301 -> 3310. Co-Authored-By: Claude Opus 5 (1M context) --- docs/adr/0001-product-contract.md | 35 ++++ src/orchestrator/core/pipeline_result.py | 12 ++ src/orchestrator/orchestrator.py | 24 ++- tests/test_failure_policy.py | 240 +++++++++++++++++++++++ 4 files changed, 309 insertions(+), 2 deletions(-) create mode 100644 tests/test_failure_policy.py diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md index 4095e27..13ffde6 100644 --- a/docs/adr/0001-product-contract.md +++ b/docs/adr/0001-product-contract.md @@ -157,6 +157,41 @@ because routing to it could not be distinguished from selecting the policy. Routing targets are validated at compile time — a jump to a step that does not exist, or a step routing to itself, is exit 2 naming the offending target. +## Failure, timeout and retry + +| | Behaviour | +|-|-| +| `timeout: N` | The step is cancelled after N seconds and fails with `TimeoutError`. `StepResult.timed_out` distinguishes it from an ordinary failure. | +| `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**. | +| `on_failure: continue` | The run carries on; the failure still surfaces in the result and the exit code. | +| Model fallback | **Does not exist.** `ModelRegistry.select_model` raises `NoEligibleModelsError` rather than substituting a model the pipeline did not ask for. | + +The `max_retries` name and its behaviour disagree — it reads as "retries beyond +the first" and acts as "total attempts". The behaviour is pinned by tests so +that changing it has to be deliberate, since every pipeline's retry budget +would shift. + +Fallback failing closed is deliberate, not an omission. Quietly selecting an +unrequested model is how a cost-control policy gets bypassed: the pipeline +believes it ran on the free model it asked for. Any future fallback has to +record what it did, and cannot cross the free/paid boundary silently. + +### Known gap: fail-fast does not fire for non-raising failures + +A tool returning `{"success": False}` without raising leaves its task +`completed`, so the failure policy never sees it and the run continues past a +step the author asked to abort on. The failure is not lost — it surfaces in the +result and the exit code — but "fail fast" does not currently fail fast for +this class. + +Making the policy consult `StepResult.success` closes it in one line and was +tried; it is reverted, because the policy aborts by *raising*, so the run then +produces no result document at all — discarding the trace exactly when it is +most wanted. Closing this properly means the execution loop stops scheduling +rather than throwing, and still returns a `PipelineResult`. + ## The result contract `Orchestrator.execute_pipeline` returns a `PipelineResult` diff --git a/src/orchestrator/core/pipeline_result.py b/src/orchestrator/core/pipeline_result.py index d656473..e46f550 100644 --- a/src/orchestrator/core/pipeline_result.py +++ b/src/orchestrator/core/pipeline_result.py @@ -111,6 +111,17 @@ def from_task(cls, task: Task, value: Any = None) -> "StepResult": dependencies=tuple(task.dependencies), ) + @property + def timed_out(self) -> bool: + """Whether this step ran out of time rather than failing on its merits. + + A timeout is retried like any other failure, so a step with + `timeout: 2` and `max_retries: 3` can occupy roughly eight seconds + before giving up. Distinguishing it from an ordinary failure is the + difference between "make the timeout bigger" and "fix the step". + """ + return self.error_type == "TimeoutError" + def to_dict(self) -> Dict[str, Any]: return { "id": self.id, @@ -127,6 +138,7 @@ def to_dict(self) -> Dict[str, Any]: "completed_at": self.completed_at, "duration": self.duration, "retries": self.retries, + "timed_out": self.timed_out, "dependencies": list(self.dependencies), } diff --git a/src/orchestrator/orchestrator.py b/src/orchestrator/orchestrator.py index 03c6926..4aa3ee5 100644 --- a/src/orchestrator/orchestrator.py +++ b/src/orchestrator/orchestrator.py @@ -880,7 +880,21 @@ async def _execute_pipeline_internal( self.logger.debug("Accumulated results so far: %s", list(results.keys())) level_results = await self._execute_level(pipeline, executable_tasks, context, results) - # Check for failures + # Check for failures. + # + # This selects on status, so it fires only for a step that + # *raised*. A tool returning {"success": False} without raising + # leaves its task COMPLETED and does not trigger the failure + # policy at all -- the run continues, and the failure surfaces + # in the result and the exit code instead. + # + # Making this consult StepResult.success so the default `fail` + # policy fired for those too was tried and reverted: the policy + # aborts by raising, so the run produced no result document at + # all -- discarding the trace precisely when it is most useful. + # Fail-fast that still returns a trace is a real gap, recorded + # in ADR 0001; it needs the execution loop to stop scheduling + # rather than to throw, which is more than a predicate change. failed_tasks = [ task_id for task_id in executable_tasks @@ -1675,7 +1689,13 @@ async def _execute_level( # reason it failed, and `task.error` was not even an # exception. The reason is the original exception; the # decision is kept beside it rather than on top of it. - task.fail(result) + # `Task.fail()` also increments `retry_count`, and the + # retry handler in _execute_task_with_resources has + # already called it for this failure. Calling it again + # here counted one attempt twice, which is why a step + # with `max_retries: 0` reported one retry. + if task.status is not TaskStatus.FAILED: + task.fail(result) results[task_id] = { "success": False, "error": str(result), diff --git a/tests/test_failure_policy.py b/tests/test_failure_policy.py new file mode 100644 index 0000000..d136f13 --- /dev/null +++ b/tests/test_failure_policy.py @@ -0,0 +1,240 @@ +"""Failure-policy semantics: timeout, retry, and what happens after. + +These were the last unpinned part of the execution contract. Everything here +was already implemented; none of it was asserted anywhere, so none of it was +safe to describe in the documentation. + +Measured, not assumed: a timeout is retried like any other failure, so a step's +worst-case wall time is roughly `timeout x (max_retries + 1)`. + +There is deliberately no test for automatic model fallback. It does not exist +on the canonical path -- `ModelRegistry.select_model` raises +`NoEligibleModelsError` rather than substituting a different model -- and the +last test in this file pins that fail-closed behaviour, because silently +selecting an unrequested model is how a cost-control policy gets bypassed. +""" + +import asyncio +import os +import time +from pathlib import Path + +import pytest + +from orchestrator.core.exceptions import NoEligibleModelsError +from orchestrator.models.model_registry import ModelRegistry +from tests.test_infrastructure import create_test_orchestrator + +pytestmark = [pytest.mark.contract, pytest.mark.e2e] + + +def _run(yaml_content, cwd): + previous = Path.cwd() + os.chdir(cwd) + try: + return asyncio.run( + create_test_orchestrator().execute_yaml(yaml_content=yaml_content, context={}) + ) + finally: + os.chdir(previous) + + +def _slow_step_pipeline(timeout, max_retries, sleep_for=3): + """A step that cannot finish inside its timeout, using a real subprocess.""" + return f""" +id: timeout_pipeline +name: Timeout Pipeline +steps: + - id: slow + tool: terminal + action: execute + timeout: {timeout} + max_retries: {max_retries} + on_failure: continue + parameters: + command: "sleep {sleep_for}" +""" + + +# --------------------------------------------------------------------------- +# timeout +# --------------------------------------------------------------------------- + +def test_a_step_that_exceeds_its_timeout_fails(tmp_path): + result = _run(_slow_step_pipeline(timeout=1, max_retries=0), tmp_path) + + step = result.steps["slow"] + assert step.success is False + assert step.error, "a timed-out step must say why" + assert "timeout" in step.error.lower() + + +def test_a_timeout_is_distinguishable_from_an_ordinary_failure(tmp_path): + """"Raise the timeout" and "fix the step" are different responses.""" + result = _run(_slow_step_pipeline(timeout=1, max_retries=0), tmp_path) + + step = result.steps["slow"] + assert step.error_type == "TimeoutError" + assert step.timed_out is True + assert step.to_dict()["timed_out"] is True + + +def test_a_step_that_finishes_in_time_is_not_marked_timed_out(tmp_path): + quick = """ +id: quick_pipeline +name: Quick Pipeline +steps: + - id: quick + tool: terminal + action: execute + timeout: 30 + parameters: + command: "echo done" +""" + result = _run(quick, tmp_path) + + assert result.steps["quick"].timed_out is False + + +# --------------------------------------------------------------------------- +# retry +# --------------------------------------------------------------------------- + +def test_retries_are_bounded_by_max_retries(tmp_path): + """`Task.reset()` deliberately does not clear `retry_count`. + + If it did, every retry would restore the budget and a permanently failing + step would retry for ever. This pins that it does not. + """ + result = _run(_slow_step_pipeline(timeout=1, max_retries=2), tmp_path) + + # Measured: `max_retries` bounds total ATTEMPTS, not retries beyond the + # first, so 2 means two attempts and therefore one retry. The name and the + # behaviour disagree; the behaviour is pinned here so changing it has to be + # a deliberate decision rather than a silent one. + assert result.steps["slow"].retries == 1 + + +def test_max_retries_zero_means_a_single_attempt(tmp_path): + result = _run(_slow_step_pipeline(timeout=1, max_retries=0), tmp_path) + + assert result.steps["slow"].retries == 0 + + +def test_a_timeout_is_retried_so_wall_time_multiplies(tmp_path): + """The consequence worth knowing before setting these numbers. + + A timeout is an exception like any other, so it goes through the same + retry path. `timeout: 1` with `max_retries: 2` is three attempts of about + a second each, not one second total. + """ + started = time.monotonic() + result = _run(_slow_step_pipeline(timeout=1, max_retries=2), tmp_path) + elapsed = time.monotonic() - started + + assert result.steps["slow"].retries == 1 + assert elapsed >= 2.0, ( + f"three one-second attempts cannot take {elapsed:.1f}s -- if this " + f"fails, the timeout is no longer being retried" + ) + # Generous ceiling: this asserts the bound exists, not the exact timing. + assert elapsed < 15.0, f"retrying took far longer than the bound: {elapsed:.1f}s" + + +# --------------------------------------------------------------------------- +# what happens after a failure +# --------------------------------------------------------------------------- + +def test_on_failure_continue_lets_the_run_finish(tmp_path): + pipeline = """ +id: continue_pipeline +name: Continue Pipeline +steps: + - id: boom + tool: filesystem + action: read + on_failure: continue + parameters: + path: "/nonexistent/definitely/not/here.txt" + - id: after + tool: filesystem + action: write + dependencies: [boom] + parameters: + path: "out/after.txt" + content: "ran anyway" +""" + result = _run(pipeline, tmp_path) + + assert result.steps["boom"].success is False + assert result.steps["after"].success is True, ( + "`continue` must let the following step run" + ) + # The pipeline still reports failure: a step in it did not succeed. + assert result.success is False + assert [s.id for s in result.failed_steps] == ["boom"] + + +def test_a_non_raising_failure_does_not_trigger_the_failure_policy(tmp_path): + """The measured behaviour, and a real gap in it. + + `fail` is the default policy, but it only fires for a step that *raised*. + A tool returning {"success": False} without raising leaves its task + COMPLETED, so the policy never sees it and the run continues -- the + downstream step below runs on a failure the pipeline author asked to abort + on. + + The failure is not lost: it surfaces in the result and in the exit code. + But "fail fast" does not currently fail fast for this class of failure. + + Making the policy consult StepResult.success was tried and reverted, + because the policy aborts by raising and the run then produced no result + document at all -- discarding the trace exactly when it is most wanted. + Closing this properly means the execution loop stops scheduling rather + than throwing, and still returns a PipelineResult. Recorded in ADR 0001. + """ + pipeline = """ +id: default_policy_pipeline +name: Default Policy Pipeline +steps: + - id: boom + tool: filesystem + action: read + parameters: + path: "/nonexistent/definitely/not/here.txt" + - id: after + tool: filesystem + action: write + dependencies: [boom] + parameters: + path: "out/after.txt" + content: "ran despite the default policy" +""" + result = _run(pipeline, tmp_path) + + assert result.steps["boom"].success is False + assert result.success is False, "the run must still report failure" + assert [s.id for s in result.failed_steps] == ["boom"] + + # The gap, pinned so that closing it is a visible change rather than a + # silent one: the downstream step ran anyway. + assert result.steps["after"].success is True + assert (tmp_path / "out" / "after.txt").exists() + + +# --------------------------------------------------------------------------- +# no automatic model fallback +# --------------------------------------------------------------------------- + +def test_model_selection_fails_closed_rather_than_substituting(): + """There is no automatic fallback, and that is the safe behaviour. + + Quietly selecting a model the pipeline did not ask for is how a + cost-control policy gets bypassed -- the pipeline believes it ran on the + free model it requested. Selection raises instead, so an unsatisfiable + requirement is an error the caller sees. + """ + registry = ModelRegistry() # deliberately empty + + with pytest.raises(NoEligibleModelsError): + asyncio.run(registry.select_model({"tasks": ["generate"]}))