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