diff --git a/exemptions.json b/exemptions.json index 96d057a7..65a2ed20 100644 --- a/exemptions.json +++ b/exemptions.json @@ -108,14 +108,14 @@ }, { "file": "src/bonfire/engine/pipeline.py", - "symbol_or_line": "154", + "symbol_or_line": "PipelineEngine.run", "rule": "BLE001", "reason": "Top-level pipeline barrier: any stage failure becomes a PipelineFailed event + failed result — the canonical record-all-and-continue boundary.", "approver": "BubbleGum/Elegance-Law architectural blessing (BON-1757); Anta gates via PR merge" }, { "file": "src/bonfire/engine/pipeline.py", - "symbol_or_line": "589", + "symbol_or_line": "PipelineEngine._execute_stage", "rule": "BLE001", "reason": "Per-stage boundary: converts ANY StageHandler failure into a typed ErrorDetail envelope (StageHandler is an open-set extension Protocol).", "approver": "BubbleGum/Elegance-Law architectural blessing (BON-1757); Anta gates via PR merge" diff --git a/file-budget.json b/file-budget.json index 5f3afb57..c84ea2b5 100644 --- a/file-budget.json +++ b/file-budget.json @@ -2,7 +2,7 @@ "files": { "src/bonfire/dispatch/security_hooks.py": 1329, "src/bonfire/dispatch/security_patterns.py": 521, - "src/bonfire/engine/pipeline.py": 989, + "src/bonfire/engine/pipeline.py": 994, "src/bonfire/handlers/merge_preflight.py": 687, "src/bonfire/handlers/sage_correction_bounce.py": 904, "src/bonfire/onboard/config_generator.py": 560, @@ -143,10 +143,10 @@ }, "packages": { "src/bonfire/dispatch": 2869, - "src/bonfire/engine": 1903, + "src/bonfire/engine": 1981, "src/bonfire/handlers": 3250, "src/bonfire/onboard": 4416, - "tests/integration": 1363, + "tests/integration": 1702, "tests/unit": 73410 }, "package_raises": [ @@ -173,6 +173,30 @@ "lines": 211, "reason": "tests/unit measured EXACTLY 73199 on bare origin/main -- the ceiling and the measurement were the same number, so the package had zero headroom and no pull request could add a unit test at all. What the 211 lines buy: tests/unit/test_init_first_run_refusals.py, eight tests over the first command a stranger runs. bonfire init met four hostile-but-ordinary path shapes with a raw Python traceback or, in one case, with a success claim -- a directory named bonfire.toml satisfied Path.exists(), the write was skipped, and the success block printed 'Already present: bonfire.toml (project config)' with exit code 0 over a project no Bonfire command can read. Seven of the eight tests fail on origin/main's behaviour with the fix removed, each for its own stated reason (PermissionError from the first write, PermissionError from the gitignore append, exit 0 with the success banner, IsADirectoryError from inside the safe-read helper, and FileExistsError three times out of mkdir(exist_ok=True) for a regular file, a second regular file, and a dangling symlink). The eighth passes on both sides on purpose: it is the control rod against a guard that learns to refuse everything, and without it the other seven would be satisfied by a command that refuses unconditionally. Four distinct defects, seven distinct failure modes, one negative control -- that is the smallest honest count, not a padded one. Explicitly rejected, in order of how tempting each was: (1) shipping fewer tests to fit -- the control rods ARE the deliverable, and four defects verified by two tests is a weaker claim than the one the pull request makes; (2) parking the file in an unbudgeted tree (tests/smoke, tests/dispatch), which costs zero budget and is dodging the ratchet rather than restructuring, and buries a regression contract where nobody maintaining init would look for it; (3) tests/integration, which is frozen at 1363 with zero headroom of its own and is the wrong home anyway for single-command CLI tests; (4) shrinking another test file to make room, which would have raced two other lanes live in this tree for budget and is worse than asking; (5) a purpose entry for the new file, which is an exemption in everything but name and would drop the file out of the ratchet permanently to accommodate one change -- the same trade the dispatch and handlers raises below both rejected, and exemptions.json is at 52/52 besides. NOTE for the next reader: a ratchet that reaches its own measurement has stopped preventing bloat and started taxing test coverage. Every future unit test in this repo now needs a raise. That is a design question about the tests/unit ceiling, not something a lane can settle.", "approver": "Anta gates via PR merge -- raise surfaced in the PR body and in the lane report, not taken silently" + }, + { + "package": "src/bonfire/engine/pipeline.py", + "from": 989, + "to": 994, + "lines": 5, + "reason": "The engine had no checkpoint write site, so bonfire status / resume / handoff read an artifact bonfire run never produced. Five lines is what the write site costs pipeline.py after the mechanism was moved out of it twice: the PipelineResult construction and the durability reasoning went to SessionStore.save_progress (session/ is unbudgeted and is the layer that owns what a stored record means), and the sink-is-None check plus the error handling went to a write_progress function in engine/checkpoint.py (the module that owns how a checkpoint is written). What is left in pipeline.py is irreducible: one import, one TYPE_CHECKING import, one constructor keyword, one assignment, one call, and a four-line comment stating why the call sits before the budget check rather than after it -- the group is paid for whether or not the next line halts, so a halt that discarded the record would bill it again on resume. Explicitly rejected: deleting that comment to land on exactly 989, which is buying a number with the reasoning for the single most subtle ordering decision in the change. Also rejected: inlining the call site's guard and try/except into _run_inner, which needs no new lines in pipeline.py at all but adds two branches to a function frozen in the complexity snapshot at 23. Also rejected: giving pipeline.py a purpose entry, which would drop 994 lines out of the ratchet to accommodate 5.", + "approver": "Anta gates via PR merge -- raise surfaced in the PR body and in the lane report, not taken silently" + }, + { + "package": "src/bonfire/engine", + "from": 1903, + "to": 1981, + "lines": 78, + "reason": "The same producer, measured at the package. 5 of the 78 are the write site in pipeline.py; the other 73 are in engine/checkpoint.py, which gains the CheckpointSink Protocol the engine writes through and the write_progress function that drives it. The Protocol has to exist because bonfire.session imports bonfire.engine, so naming SessionStore in the engine's signature would point the dependency back into a cycle; import-linter reports 1 contract kept with the Protocol in place. Both belong in checkpoint.py rather than anywhere cheaper: that module already owns the atomic tmp+replace write and the symlink refusals, and the new code is the rule for when that machinery runs. Explicitly rejected: putting the Protocol in bonfire/protocols.py, which is unbudgeted and would have cost zero -- but CLAUDE.md's release gate names 'the four runtime_checkable extension protocols' as a v0.1 trust-triangle item, and quietly making it five edits a documented count from a lane that does not own that doc. Also rejected: writing the checkpoint from an event consumer in the unbudgeted events package -- StageCompleted carries stage_name, agent_name, duration and cost and no Envelope, so a consumer cannot populate CheckpointData.completed at all. That one is not a budget trade; it does not work.", + "approver": "Anta gates via PR merge -- raise surfaced in the PR body and in the lane report, not taken silently" + }, + { + "package": "tests/integration", + "from": 1363, + "to": 1702, + "lines": 339, + "reason": "tests/integration measured EXACTLY its ceiling on origin/main (1363/1363), so the package had zero headroom and no pull request could add an integration test at all. What the 339 lines buy: tests/integration/test_run_checkpoints.py, eight tests that run the engine returned by build_default_engine. They have to live in tests/integration and they have to go through the composition root: this defect family -- Envelope.artifacts, review-verdict.json, costs.jsonl and now the checkpoint -- is invisible to unit tests by construction, because every unit test of the run path injects its own engine factory, so the wiring nothing exercises is the wiring nothing can catch. A test that assembles the object graph by hand re-implements the wiring under test and passes whether or not the product is wired. Both halves are rodded: removing the engine's write site turns 5 of the 8 red, and removing checkpoint_sink= from build_default_engine turns the same 4 red plus the wiring assertion, while the two tests that must not depend on the write stay green. Explicitly rejected: putting these in tests/unit, which would have been the same line count against a ceiling a second live lane needs this round -- and would have meant hand-building the engine, which is the measurement error this file exists to avoid. Also rejected: a purpose entry for the new file, which registers it out of the ratchet entirely rather than paying for it.", + "approver": "Anta gates via PR merge -- raise surfaced in the PR body and in the lane report, not taken silently" } ] } diff --git a/src/bonfire/engine/__init__.py b/src/bonfire/engine/__init__.py index c4a553c2..238b064e 100644 --- a/src/bonfire/engine/__init__.py +++ b/src/bonfire/engine/__init__.py @@ -26,13 +26,15 @@ :class:`CostLimitGate`) plus :class:`GateChain` for sequential evaluation with short-circuit on error severity. - The checkpoint trio (:class:`CheckpointManager`, - :class:`CheckpointData`, :class:`CheckpointSummary`) — an opt-in - persistence surface a caller can drive around - :meth:`PipelineEngine.run`. The engine does not write checkpoints - between stages; callers persist a :class:`PipelineResult` via - :meth:`CheckpointManager.save` and resume by passing the loaded - ``completed`` mapping back into :meth:`PipelineEngine.run` on the - next invocation. + :class:`CheckpointData`, :class:`CheckpointSummary`) — the + persistence surface behind ``bonfire status`` / ``resume`` / + ``handoff``. The engine writes a checkpoint at every stage-group + boundary when its ``checkpoint_sink`` is wired, which is what the + composition root does; a caller constructing an engine by hand and + omitting the sink gets the previous behaviour, no writes. Resume is + unchanged: pass a loaded ``CheckpointData.completed`` mapping back + into :meth:`PipelineEngine.run` and the named stages are skipped, + their cost seeded rather than spent again. """ from bonfire.engine.checkpoint import CheckpointData, CheckpointManager, CheckpointSummary diff --git a/src/bonfire/engine/checkpoint.py b/src/bonfire/engine/checkpoint.py index 7d1f0d0c..d5a452c7 100644 --- a/src/bonfire/engine/checkpoint.py +++ b/src/bonfire/engine/checkpoint.py @@ -20,7 +20,7 @@ import os import time from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol, runtime_checkable from pydantic import BaseModel, ConfigDict, ValidationError @@ -35,6 +35,73 @@ from bonfire.models.plan import WorkflowPlan +# --------------------------------------------------------------------------- +# CheckpointSink -- the seam the engine writes through +# --------------------------------------------------------------------------- + + +@runtime_checkable +class CheckpointSink(Protocol): + """Where :class:`~bonfire.engine.pipeline.PipelineEngine` sends its progress. + + ``bonfire.session.store.SessionStore`` satisfies this structurally and is + what the composition root injects. The engine names a Protocol rather than + that class because ``bonfire.session`` imports ``bonfire.engine``, so + naming the class would point the dependency back into a cycle. + + The engine passes the facts it owns rather than a ``PipelineResult``: a + run still in progress has no result, and asking the engine to build one + would mean stamping a ``success`` value onto a question not yet answered. + Deciding what an in-progress record looks like belongs to the layer that + persists it, not to the layer that is still working. + """ + + def save_progress( + self, + session_id: str, + stages: dict[str, Envelope], + total_cost_usd: float, + plan: WorkflowPlan, + ) -> Path: ... + + +def write_progress( + sink: CheckpointSink | None, + session_id: str, + stages: dict[str, Envelope], + total_cost_usd: float, + plan: WorkflowPlan, +) -> None: + """Send a mid-run snapshot to *sink*, absorbing a failure to write it. + + Lives here rather than on the engine so that "how a checkpoint is written" + stays in the checkpoint module, and so ``pipeline.py`` -- already the + largest file in the package -- gains a call and not a mechanism. + + A checkpoint that cannot be written must not fail an otherwise healthy + run: the work is unaffected by whether a copy of the record reached disk, + and ``PipelineEngine.run`` converts anything raised here into a failed + result, which would report that the run failed when only its record did. + It must not be silent either, or an operator meeting an empty ``bonfire + status`` has no way to learn why. + + Only ``OSError`` and ``ValueError`` are absorbed -- the shapes a disk, + a permission or an oversized payload produces. A sink that does not + implement this Protocol raises ``AttributeError`` and is left to + propagate: that is a wiring mistake, and hiding it would reintroduce + exactly the silence this producer exists to end. + """ + if sink is None: + return + try: + # Copy: ``stages`` is the engine's live accumulator, mutated in place + # as later stages finish. A snapshot must not keep growing after it + # was taken. + sink.save_progress(session_id, dict(stages), total_cost_usd, plan) + except (OSError, ValueError) as exc: + logger.warning("Could not checkpoint session %s: %s", session_id, exc) + + # --------------------------------------------------------------------------- # CheckpointData -- frozen Pydantic model # --------------------------------------------------------------------------- diff --git a/src/bonfire/engine/composition.py b/src/bonfire/engine/composition.py index d7bb7ff4..c6a089db 100644 --- a/src/bonfire/engine/composition.py +++ b/src/bonfire/engine/composition.py @@ -361,6 +361,7 @@ def build_default_engine( from bonfire.events.bus import EventBus from bonfire.events.consumers import CostTracker, SessionLoggerConsumer from bonfire.session.persistence import SessionPersistence + from bonfire.session.store import SessionStore root = resolve_project_root() if project_root is None else project_root.resolve() settings = load_settings_or_default() @@ -400,4 +401,14 @@ def build_default_engine( project_root=root, tool_policy=tool_policy, settings=settings, + # The read side of this store was already wired to three shipped + # verbs: ``bonfire status``, ``bonfire resume`` and ``bonfire + # handoff`` all ask ``SessionStore`` what the last run left behind. + # Nothing answered. ``SessionStore.save`` had no caller anywhere in + # ``src/bonfire`` and the engine had no write site, so every one of + # the three reported an empty store after a run that really happened. + # The same object is passed here so the write and the three reads + # resolve one directory by one rule (``BONFIRE_CHECKPOINT_DIR``, then + # ``~/.bonfire/checkpoints``) rather than agreeing by coincidence. + checkpoint_sink=SessionStore(), ) diff --git a/src/bonfire/engine/pipeline.py b/src/bonfire/engine/pipeline.py index 7518738f..9c37e417 100644 --- a/src/bonfire/engine/pipeline.py +++ b/src/bonfire/engine/pipeline.py @@ -32,6 +32,7 @@ from bonfire.dispatch.runner import execute_with_retry from bonfire.engine import factory +from bonfire.engine.checkpoint import write_progress from bonfire.engine.context import ContextBuilder from bonfire.engine.gates import UnknownGateError from bonfire.engine.model_resolver import resolve_dispatch_model @@ -53,6 +54,7 @@ if TYPE_CHECKING: from bonfire.dispatch.tool_policy import ToolPolicy + from bonfire.engine.checkpoint import CheckpointSink from bonfire.events.bus import EventBus from bonfire.models.config import BonfireSettings, PipelineConfig from bonfire.protocols import AgentBackend, QualityGate, StageHandler @@ -108,6 +110,7 @@ def __init__( project_root: Any | None = None, tool_policy: ToolPolicy | None = None, settings: BonfireSettings | None = None, + checkpoint_sink: CheckpointSink | None = None, ) -> None: self._backend = backend self._bus = bus @@ -118,6 +121,7 @@ def __init__( self._project_root = project_root self._tool_policy = tool_policy self._settings = settings if settings is not None else factory.load_settings_or_default() + self._checkpoint_sink = checkpoint_sink # -- Public API ---------------------------------------------------------- @@ -278,6 +282,12 @@ async def _run_inner( if halt is not None: return halt + # Every stage here passed its gates, so the record can advance. + # Before the budget check below, not after: this group was paid + # for whether or not the next line halts, and a halt discarding + # the record would bill it again on the next attempt. + write_progress(self._checkpoint_sink, session_id, stages_done, total_cost, plan) + # Budget check after each group if total_cost > plan.budget_usd: duration = time.monotonic() - start diff --git a/src/bonfire/session/store.py b/src/bonfire/session/store.py index a8cae010..1cd35a56 100644 --- a/src/bonfire/session/store.py +++ b/src/bonfire/session/store.py @@ -15,12 +15,19 @@ ``~/.bonfire/`` convention the cost ledger and personas already follow), and * exposes the three lookups the verbs need (``latest``, ``load``, - ``summaries``) plus a ``save`` shim so callers and tests persist through the - same resolved location. + ``summaries``) plus the two writes (``save`` for a finished run, + ``save_progress`` for one still going) so every read and every write resolves + the same location. The verbs themselves stay thin: they format what the store returns. Keeping the location logic here means a future change to where checkpoints live is a one-line edit, not a three-command sweep. + +``save_progress`` is the producer side, and it went missing for a release: +``save`` had no caller anywhere in ``src/bonfire/`` and the pipeline had no +write site, so the three verbs above read a store that a real ``bonfire run`` +never wrote to. The composition root now injects this class into the engine as +its checkpoint sink. """ from __future__ import annotations @@ -34,6 +41,7 @@ if TYPE_CHECKING: from bonfire.engine.checkpoint import CheckpointData, CheckpointSummary from bonfire.engine.pipeline import PipelineResult + from bonfire.models.envelope import Envelope from bonfire.models.plan import WorkflowPlan #: Environment variable that overrides the checkpoint directory. Mirrors the @@ -83,3 +91,43 @@ def summaries(self) -> list[CheckpointSummary]: def save(self, result: PipelineResult, plan: WorkflowPlan) -> Path: """Persist a pipeline result so the verbs (and tests) can read it back.""" return self._manager.save(result.session_id, result, plan) + + def save_progress( + self, + session_id: str, + stages: dict[str, Envelope], + total_cost_usd: float, + plan: WorkflowPlan, + ) -> Path: + """Persist a run that is still going, from the facts the engine holds. + + This is the write side of the three lifecycle verbs, and it is what + :class:`~bonfire.engine.pipeline.PipelineEngine` calls at each stage-group + boundary. It exists separately from :meth:`save` because a run in + progress has no :class:`~bonfire.engine.pipeline.PipelineResult` to hand + over: building one at the call site would mean the engine stamping a + ``success`` value onto a question it has not answered yet. Constructing + the snapshot here keeps that decision with the layer that knows what a + stored record means. + + ``success=False`` is therefore chosen here and is deliberate rather + than incidental. It is not persisted -- :class:`CheckpointData` records + the completed stages, the plan and the spend, and nothing else -- but + an in-progress snapshot must not be constructed claiming it finished. + + Durability comes from :meth:`CheckpointManager.save`, which writes a + tmp file and ``os.replace``s it into position. That matters more here + than on the terminal write: this method is called repeatedly while the + run is alive, so an interruption during a write is an ordinary + outcome, and it must leave either the previous checkpoint or the new + one -- never a truncated file naming stages that did not finish. + """ + from bonfire.engine.pipeline import PipelineResult + + snapshot = PipelineResult( + success=False, + session_id=session_id, + stages=stages, + total_cost_usd=total_cost_usd, + ) + return self._manager.save(session_id, snapshot, plan) diff --git a/tests/integration/test_run_checkpoints.py b/tests/integration/test_run_checkpoints.py new file mode 100644 index 00000000..caf0a6ac --- /dev/null +++ b/tests/integration/test_run_checkpoints.py @@ -0,0 +1,339 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 BonfireAI + +"""The checkpoint a run leaves behind — proved through the real composition root. + +``bonfire status``, ``bonfire resume`` and ``bonfire handoff`` are three shipped +verbs that all read one artifact, and nothing wrote it. ``SessionStore.save`` +had no caller anywhere in ``src/bonfire/`` and the pipeline loop had no +checkpoint write site, so a run that really dispatched and really spent money +was followed by three commands reporting an empty store. + +That is the same shape as the ``Envelope.artifacts`` and +``.bonfire/review-verdict.json`` gaps, and it was invisible for the same +reason: every test of the run path injects its own engine factory, so the +wiring nothing exercises is the wiring nothing can catch. This module +therefore calls :func:`bonfire.engine.composition.build_default_engine` and +runs the engine it returns. Assembling the object graph by hand would +re-implement the wiring under test and pass whether or not the product is +wired at all. + +One replacement, stated: ``claude_agent_sdk.query``. Every dispatch below +would otherwise be a billed network call. The fake charges a real +``total_cost_usd`` because the assertions here are partly about money being +recorded, and a zero-cost fake would let an empty figure look correct. +""" + +from __future__ import annotations + +import asyncio +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +import pytest +from claude_agent_sdk import AssistantMessage, ResultMessage, TextBlock + +from bonfire.dispatch import sdk_backend +from bonfire.engine.checkpoint import CheckpointSink +from bonfire.engine.composition import build_default_engine +from bonfire.models.plan import StageSpec, WorkflowPlan, WorkflowType +from bonfire.session.store import CHECKPOINT_DIR_ENV_VAR, SessionStore +from bonfire.workflow.standard import debug + +#: What one faked dispatch charges. Deliberately not round: a total asserted +#: against this figure cannot be satisfied by a default, a zero, or a number +#: someone rounded on the way through. +STAGE_COST_USD = 0.11 + + +class CountingTransport: + """``claude_agent_sdk.query`` stand-in that charges and counts its calls. + + The call count is the load-bearing part for the resume assertions: the + question "was this stage billed again?" is answered by whether the + transport was reached, not by re-reading a total the engine computed. + """ + + def __init__(self, cost_usd: float = STAGE_COST_USD) -> None: + self.cost_usd = cost_usd + self.prompts: list[str] = [] + + def __call__(self, *, prompt: str, options: Any) -> Any: + self.prompts.append(prompt) + cost = self.cost_usd + + async def _stream() -> Any: + yield AssistantMessage(content=[TextBlock(text="done")], model="fake") + yield ResultMessage( + subtype="success", + duration_ms=1, + duration_api_ms=1, + is_error=False, + num_turns=1, + session_id="fake-session", + total_cost_usd=cost, + result="done", + ) + + return _stream() + + +@pytest.fixture +def store_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Point the checkpoint store at a throwaway directory. + + The env override is the same one the three verbs honour, so redirecting it + steers the write and all three reads together -- which is itself part of + what these tests check. + """ + path = tmp_path / "checkpoints" + monkeypatch.setenv(CHECKPOINT_DIR_ENV_VAR, str(path)) + return path + + +@pytest.fixture +def counting(monkeypatch: pytest.MonkeyPatch) -> CountingTransport: + fake = CountingTransport() + monkeypatch.setattr(sdk_backend, "query", fake) + return fake + + +def _plan(*, budget_usd: float = 10.0, gate_second_stage: bool = False) -> WorkflowPlan: + """A three-stage chain. Optionally gate the middle stage on cost. + + ``cost_limit`` is the one built-in gate that fails deterministically with + no filesystem and no subprocess: ``build_default_gates`` registers it + against the plan's own budget, so a budget below the running total fails + the stage that names it. + """ + gates = ["cost_limit"] if gate_second_stage else [] + return WorkflowPlan( + name="debug", + workflow_type=WorkflowType.DEBUG, + description="checkpoint probe", + task_description="ship the checkout refactor", + budget_usd=budget_usd, + stages=[ + StageSpec(name="scout", agent_name="scout", role="scout"), + StageSpec( + name="warrior", + agent_name="warrior", + role="warrior", + depends_on=["scout"], + gates=gates, + ), + StageSpec(name="bard", agent_name="bard", role="bard", depends_on=["warrior"]), + ], + ) + + +def _run(plan: WorkflowPlan, root: Path, **kwargs: Any) -> Any: + return asyncio.run(build_default_engine(plan, project_root=root).run(plan, **kwargs)) + + +# --------------------------------------------------------------------------- +# The producer exists at all +# --------------------------------------------------------------------------- + + +def test_a_run_leaves_a_checkpoint_on_disk( + tmp_path: Path, git_repo: Any, store_dir: Path, counting: CountingTransport +) -> None: + """The regression this module exists for. + + Before the producer landed this directory stayed empty after a run that + dispatched three stages and spent real money. + """ + result = _run(_plan(), git_repo(tmp_path / "r")) + + assert result.success + written = sorted(p.name for p in store_dir.iterdir()) + assert written == [f"{result.session_id}.json"], ( + "a completed run must leave exactly one checkpoint, named for its session" + ) + data = json.loads((store_dir / written[0]).read_text()) + assert sorted(data["completed"]) == ["bard", "scout", "warrior"] + assert data["total_cost_usd"] == pytest.approx(3 * STAGE_COST_USD) + assert data["plan_name"] == "debug" + assert data["task_description"] == "ship the checkout refactor" + + +def test_an_engine_with_no_sink_writes_nothing( + tmp_path: Path, git_repo: Any, store_dir: Path, counting: CountingTransport +) -> None: + """Control rod on the assertion above. + + The engine's sink is optional, and a library caller who omits it keeps the + previous behaviour. If the test above passed with the sink removed it + would be measuring the fixture, not the producer. + """ + plan = _plan() + engine = build_default_engine(plan, project_root=git_repo(tmp_path / "r")) + engine._checkpoint_sink = None + + assert asyncio.run(engine.run(plan)).success + assert not store_dir.exists() or list(store_dir.iterdir()) == [] + + +# --------------------------------------------------------------------------- +# Partway through -- the case resume exists for +# --------------------------------------------------------------------------- + + +def test_a_run_that_halts_partway_records_only_the_stages_that_passed( + tmp_path: Path, git_repo: Any, store_dir: Path, counting: CountingTransport +) -> None: + """A halt at stage two must leave stage one on disk and stage two off it. + + Writing the failed stage into ``completed`` would be worse than writing + nothing: resume skips whatever the checkpoint names, so a failed stage + recorded as done is a stage that silently never runs again. + """ + plan = _plan(budget_usd=STAGE_COST_USD * 1.5, gate_second_stage=True) + result = _run(plan, git_repo(tmp_path / "r")) + + assert not result.success + assert result.failed_stage == "warrior" + data = json.loads((store_dir / f"{result.session_id}.json").read_text()) + assert sorted(data["completed"]) == ["scout"], ( + "the gate-failed stage and the stage after it must not be recorded as done" + ) + assert data["total_cost_usd"] == pytest.approx(STAGE_COST_USD) + + +def test_the_verbs_read_the_checkpoint_from_a_separate_process( + tmp_path: Path, git_repo: Any, store_dir: Path, counting: CountingTransport +) -> None: + """Survival of the process that wrote it, proved by leaving that process. + + ``CliRunner`` would exercise the same interpreter that just ran the + engine, which cannot distinguish a checkpoint on disk from state still + in memory. These three invocations are real ``bonfire`` processes with + no knowledge of the run beyond the file. + + The plan is the registry's own ``debug``, not an ad-hoc one: a checkpoint + stores ``plan_name`` and not the stage list, so ``status`` and ``resume`` + re-derive the total and the remainder from the registry. A test plan + reusing a registered name with a different shape would be graded against + the registered shape and report a denominator nobody ran. + + It halts on the budget check that follows the first stage, which also + pins the ordering the engine promises: work already dispatched and paid + for is on disk before the halt that follows it. + """ + plan = debug().model_copy( + update={"budget_usd": STAGE_COST_USD / 2, "task_description": "ship the refactor"} + ) + result = _run(plan, git_repo(tmp_path / "r")) + assert not result.success and "Budget exceeded" in result.error + bonfire = Path(sys.executable).with_name("bonfire") + + def _verb(name: str) -> str: + done = subprocess.run([str(bonfire), name], capture_output=True, text=True, check=True) + return done.stdout + + status = _verb("status") + assert result.session_id in status + assert "1 / 2 stages" in status, f"status must report progress, not a total: {status!r}" + assert "$0.11" in status + + resume = _verb("resume") + assert "warrior" in resume, f"resume must name what is left: {resume!r}" + + handoff = _verb("handoff") + assert "scout" in handoff and "ship the refactor" in handoff + + +# --------------------------------------------------------------------------- +# Resume does not pay twice +# --------------------------------------------------------------------------- + + +def test_resuming_from_the_checkpoint_does_not_re_dispatch_or_re_bill( + tmp_path: Path, git_repo: Any, store_dir: Path, counting: CountingTransport +) -> None: + """The ruling on re-billing, measured at the transport rather than inferred. + + The first leg halts after ``scout``. Feeding that checkpoint's + ``completed`` map back into a second engine must reach the transport only + for the two stages that remain, and the final total must be the whole + pipeline's spend -- the seeded first stage plus the tail, counted once. + """ + root = git_repo(tmp_path / "r") + first = _run(_plan(budget_usd=STAGE_COST_USD * 1.5, gate_second_stage=True), root) + assert not first.success + assert len(counting.prompts) == 2, "first leg dispatched scout and the failed warrior" + + saved = SessionStore().load(first.session_id) + counting.prompts.clear() + + second = _run(_plan(), root, session_id=first.session_id, completed=dict(saved.completed)) + + assert second.success + assert len(counting.prompts) == 2, ( + f"resume must dispatch only warrior and bard, not scout again: {len(counting.prompts)}" + ) + assert "Output from scout" in counting.prompts[0], ( + "the seeded envelope must be carried forward as context, which is what " + "makes skipping the stage legitimate rather than merely cheaper" + ) + assert second.total_cost_usd == pytest.approx(3 * STAGE_COST_USD), ( + "the seeded stage is counted once, not spent again and not dropped" + ) + + +# --------------------------------------------------------------------------- +# The wiring itself +# --------------------------------------------------------------------------- + + +def test_the_root_wires_a_sink_resolving_the_directory_the_verbs_read( + tmp_path: Path, git_repo: Any, store_dir: Path +) -> None: + """Write side and read side must resolve one directory by one rule. + + Two independent resolutions that happen to agree today are a defect + waiting for the next change to either. + """ + engine = build_default_engine(_plan(), project_root=git_repo(tmp_path / "r")) + sink = engine._checkpoint_sink + + assert isinstance(sink, SessionStore) + assert sink.checkpoint_dir == store_dir == SessionStore().checkpoint_dir + + +def test_session_store_satisfies_the_sink_protocol() -> None: + assert isinstance(SessionStore(), CheckpointSink) + + +def test_a_sink_that_cannot_write_does_not_fail_the_run( + tmp_path: Path, + git_repo: Any, + store_dir: Path, + counting: CountingTransport, + caplog: pytest.LogCaptureFixture, +) -> None: + """A disk problem must not turn a healthy run into a failed one. + + ``run()`` never raises, so an uncaught error here would be converted into + a failed ``PipelineResult`` -- reporting that the work failed when only + the record of it did. It must not be silent either, or an empty + ``bonfire status`` has no explanation. + """ + + class RefusingSink: + def save_progress(self, *args: Any) -> Path: + raise OSError("read-only file system") + + plan = _plan() + engine = build_default_engine(plan, project_root=git_repo(tmp_path / "r")) + engine._checkpoint_sink = RefusingSink() + + with caplog.at_level("WARNING"): + result = asyncio.run(engine.run(plan)) + + assert result.success, "a checkpoint that cannot be written must not fail the run" + assert "read-only file system" in caplog.text