diff --git a/exemptions.json b/exemptions.json index 65a2ed20..cdb46e74 100644 --- a/exemptions.json +++ b/exemptions.json @@ -269,28 +269,28 @@ }, { "file": "src/bonfire/onboard/flow.py", - "symbol_or_line": "147", + "symbol_or_line": "run_front_door", "rule": "C901", "reason": "grandfathered at v1.0.1 cf-gate adoption 2026-06-22 (ratchet: BON-1757) — pre-existing under the kit ruff config; tracked for fix-forward", "approver": "BubbleGum mount (v1.0.1 gate-adoption), 2026-06-22 — pending Anta's blessing via PR #227" }, { "file": "src/bonfire/onboard/flow.py", - "symbol_or_line": "147", + "symbol_or_line": "run_front_door", "rule": "PLR0915", "reason": "grandfathered at v1.0.1 cf-gate adoption 2026-06-22 (ratchet: BON-1757) — pre-existing under the kit ruff config; tracked for fix-forward", "approver": "BubbleGum mount (v1.0.1 gate-adoption), 2026-06-22 — pending Anta's blessing via PR #227" }, { "file": "src/bonfire/onboard/flow.py", - "symbol_or_line": "239", + "symbol_or_line": "run_front_door", "rule": "BLE001", "reason": "Shutdown drain of cancelled asyncio tasks: the asyncio.CancelledError arm is load-bearing (deliberately swallows the cancel during drain); the broad arm catches a drained task's own arbitrary error.", "approver": "BubbleGum/Elegance-Law architectural blessing (BON-1757); Anta gates via PR merge" }, { "file": "src/bonfire/onboard/flow.py", - "symbol_or_line": "239", + "symbol_or_line": "run_front_door", "rule": "S110", "reason": "grandfathered at v1.0.1 cf-gate adoption 2026-06-22 (ratchet: BON-1757) — pre-existing under the kit ruff config; tracked for fix-forward", "approver": "BubbleGum mount (v1.0.1 gate-adoption), 2026-06-22 — pending Anta's blessing via PR #227" @@ -311,7 +311,7 @@ }, { "file": "src/bonfire/onboard/orchestrator.py", - "symbol_or_line": "106", + "symbol_or_line": "_run_one", "rule": "BLE001", "reason": "Per-scanner boundary: scanners are a pluggable open set; one scanner failing must not abort onboarding (log + count=0 + continue).", "approver": "BubbleGum/Elegance-Law architectural blessing (BON-1757); Anta gates via PR merge" diff --git a/file-budget.json b/file-budget.json index c84ea2b5..22744875 100644 --- a/file-budget.json +++ b/file-budget.json @@ -145,9 +145,9 @@ "src/bonfire/dispatch": 2869, "src/bonfire/engine": 1981, "src/bonfire/handlers": 3250, - "src/bonfire/onboard": 4416, + "src/bonfire/onboard": 4460, "tests/integration": 1702, - "tests/unit": 73410 + "tests/unit": 73777 }, "package_raises": [ { @@ -197,6 +197,22 @@ "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" + }, + { + "package": "tests/unit", + "from": 73410, + "to": 73777, + "lines": 367, + "reason": "tests/unit measured 73409 against a ceiling of 73410 -- one line of headroom, the second consecutive lane to arrive at a ratchet sitting on its own measurement (the raise immediately below moved it 73199 -> 73410 and landed it right back on the number). What the 367 lines buy: tests/unit/test_failure_is_not_a_completion.py, twenty tests over four places where a failure recorded itself as a completion or gave a reason that was not the reason that occurred. The money-facing one is the cost ledger: a halted run and a run that finished one stage wrote rows differing ONLY in timestamp, so 'bonfire cost' -- the operator's record of what they were charged and why -- could report the spend but not whether it bought a finished run or a crash. The others are the XP penalty's reason, a crashed scanner reported to the browser as a clean scan of nothing, and every client-frame validation error reported as message_too_long. Each defect gets a regression test that fails on origin/main's behaviour AND a negative control asserting a genuine success still records as a success, so none of it can be satisfied by code that reports everything as broken; two further tests pin the on-disk migration (pre-field rows still load, and still aggregate through CostAnalyzer, without being fabricated into successes). Explicitly rejected: (1) shipping fewer tests -- the control rods ARE the deliverable and the negative controls are what stop the fix from being 'call everything a failure'; (2) parking the file in an unbudgeted tree, which costs zero budget and is dodging the ratchet rather than restructuring, and was rejected on the same grounds by the raise below; (3) a purpose entry, an exemption in all but name that would drop the file out of the ratchet permanently; (4) shrinking another test file to make room, which races two lanes live in this tree right now; (5) padding the ask to leave headroom for the next lane -- the number is the measurement, and a lane granting itself slack is how a ratchet dies. The file was tightened before the ask was sized (a shared sink helper and a shared legacy-row constant replaced duplicated setup, and two pairs of tests became parametrized), which took the measurement from 389 to 368.", + "approver": "Anta gates via PR merge -- raise surfaced in the PR body and in the lane report, not taken silently" + }, + { + "package": "src/bonfire/onboard", + "from": 4416, + "to": 4460, + "lines": 44, + "reason": "src/bonfire/onboard measured EXACTLY 4416 against a ceiling of 4416 -- zero headroom, verified by a one-line probe file producing 'package src/bonfire/onboard is 4417 lines'. The 44 lines fix two records in this package that reported a failure as a success. In orchestrator.py a scanner that CRASHED emitted ScanComplete(item_count=0), identical to a scanner that ran clean and found nothing, so the browser said 'we scanned and found nothing' over a scan that had died; _run_one now returns (count, failed) and reports the real exception, and run_scan totals failed panels so an all-crashed run does not summarise as total_items=0. In flow.py every ValidationError on a client frame was reported as message_too_long -- including a frame with no text field at all and a frame whose text was a number -- sending the user to shorten a message that was never long; _frame_rejection now keeps message_too_long only for a genuine pydantic string_too_long and names the actual cause otherwise. protocol.py carries the three new fields, all defaulted to the success shape so an older producer's frame still parses. Explicitly rejected: (1) reporting the crash only to the log, which is where it already went and is not visible to the browser that is the actual consumer; (2) compressing docstrings elsewhere in the package to buy the lines back, which trades documentation for budget and measures as nothing anyway since the unit is max(physical, statements); (3) moving the fix out of onboard into an unbudgeted package, which would take the producer off the path that produces. The protocol.py docstrings WERE trimmed once for size before this number was taken.", + "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/cost/consumer.py b/src/bonfire/cost/consumer.py index b8964266..a386dacd 100644 --- a/src/bonfire/cost/consumer.py +++ b/src/bonfire/cost/consumer.py @@ -68,6 +68,7 @@ async def _on_pipeline_completed(self, event: PipelineCompleted) -> None: total_cost_usd=event.total_cost_usd, duration_seconds=event.duration_seconds, stages_completed=event.stages_completed, + outcome="completed", ) self._append(record) @@ -81,6 +82,15 @@ async def _on_pipeline_failed(self, event: PipelineFailed) -> None: — every failed session no longer looks instant with zero stages done, and downstream analyzers can compute meaningful success-rate / mean-time-to-halt over the ledger. + + Carrying those two fields alone made the halt row a perfect + forgery of a success row: a run that died in the builder and a + run that finished one stage both wrote ``stages_completed=1`` + and nothing else differed but ``timestamp``. ``outcome`` is the + field that separates them, and ``failed_stage`` / + ``error_message`` say WHICH failure it was — copied from the + event rather than inferred, so the ledger's reason is the + reason that actually occurred. """ record = PipelineRecord( timestamp=event.timestamp, @@ -88,6 +98,9 @@ async def _on_pipeline_failed(self, event: PipelineFailed) -> None: total_cost_usd=event.total_cost_usd, duration_seconds=event.duration_seconds, stages_completed=event.stages_completed, + outcome="failed", + failed_stage=event.failed_stage, + error_message=event.error_message, ) self._append(record) diff --git a/src/bonfire/cost/models.py b/src/bonfire/cost/models.py index 79c26c8a..e6969f41 100644 --- a/src/bonfire/cost/models.py +++ b/src/bonfire/cost/models.py @@ -56,7 +56,27 @@ class DispatchRecord(BaseModel): class PipelineRecord(BaseModel): - """One pipeline completion with total cost.""" + """One pipeline run with total cost, and how that run ENDED. + + ``outcome`` exists because a halt and a completion used to write + byte-identical rows apart from ``timestamp``. A run that died in + the builder and a run that finished one stage both landed as + ``stages_completed=1``, so ``bonfire cost`` — the operator's record + of what they were charged and why — could report spend but could + not say whether the money bought a finished run or a crash. + + ``failed_stage`` and ``error_message`` carry the reason across from + ``PipelineFailed`` rather than being re-derived, so the ledger + states the cause that actually occurred instead of an inferred one. + + Migration: the default is ``"unknown"``, NOT ``"completed"``. Rows + written before this field existed genuinely do not record how the + run ended, and defaulting them to success would fabricate exactly + the history this defect corrupted. ``CostAnalyzer`` does not list + ``outcome`` in ``_PIPELINE_REQUIRED_FIELDS``, so those rows keep + validating and keep aggregating unchanged; they simply decline to + claim an outcome nobody recorded. + """ type: Literal["pipeline"] = "pipeline" timestamp: float @@ -64,6 +84,9 @@ class PipelineRecord(BaseModel): total_cost_usd: float duration_seconds: float stages_completed: int + outcome: Literal["completed", "failed", "unknown"] = "unknown" + failed_stage: str | None = None + error_message: str | None = None class SessionCost(BaseModel): diff --git a/src/bonfire/onboard/flow.py b/src/bonfire/onboard/flow.py index 3db57c10..61ba5c89 100644 --- a/src/bonfire/onboard/flow.py +++ b/src/bonfire/onboard/flow.py @@ -90,6 +90,28 @@ class MessageTooLargeError(BrowserDisconnectedError): """ +def _frame_rejection(exc: ValidationError) -> ServerError: + """Name the ACTUAL reason a client frame was refused. + + Every ``ValidationError`` on a ``user_message`` frame used to be + reported as ``message_too_long``. A frame with no ``text`` field at + all, and a frame whose ``text`` was a number, both told the user to + shorten a message that was never long — sending them to fix + something that is not broken. Only a genuine length violation + (pydantic ``string_too_long``) keeps that code; anything else is + reported as what it is. + """ + if any(err["type"] == "string_too_long" for err in exc.errors()): + return ServerError( + code="message_too_long", + message="Message too long; please keep under 8 KiB.", + ) + detail = "; ".join( + f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}" for err in exc.errors() + ) + return ServerError(code="invalid_message", message=f"Message rejected — {detail}") + + async def dispatch_user_message( data: dict[str, Any], *, @@ -105,7 +127,9 @@ async def dispatch_user_message( - Non-``user_message`` frames are ignored. - Overlong payloads (>``MAX_USER_MESSAGE_LEN``) trigger a ``server_error`` frame with code ``message_too_long`` and never reach - the conversation analyzer. + the conversation analyzer. Frames rejected for any OTHER reason get + code ``invalid_message`` naming the real cause — see + :func:`_frame_rejection`. - ``ConversationCompleteError`` after the third answer is broadcast as a polite ``server_error`` and is NOT propagated. - Once the conversation is complete, ``conversation_done`` is signalled @@ -120,13 +144,8 @@ async def dispatch_user_message( try: msg = UserMessage.model_validate(data) - except ValidationError: - await broadcast( - ServerError( - code="message_too_long", - message="Message too long; please keep under 8 KiB.", - ) - ) + except ValidationError as exc: + await broadcast(_frame_rejection(exc)) return try: diff --git a/src/bonfire/onboard/orchestrator.py b/src/bonfire/onboard/orchestrator.py index 2a15e0cf..95a3edab 100644 --- a/src/bonfire/onboard/orchestrator.py +++ b/src/bonfire/onboard/orchestrator.py @@ -77,10 +77,11 @@ async def run_scan( await emit(ScanStart(panels=panel_names)) tasks = [_run_one(panel, module, project_path, emit) for panel, module in scanners] - results: list[int] = await asyncio.gather(*tasks) + results: list[tuple[int, bool]] = await asyncio.gather(*tasks) - total = sum(results) - await emit(AllScansComplete(total_items=total)) + total = sum(count for count, _ in results) + failed_panels = sum(1 for _, failed in results if failed) + await emit(AllScansComplete(total_items=total, failed_panels=failed_panels)) return total @@ -94,18 +95,29 @@ async def _run_one( module: ModuleType, project_path: Path, emit: Callable[[FrontDoorMessage], Awaitable[None]], -) -> int: - """Execute a single scanner, catch failures, emit ScanComplete.""" +) -> tuple[int, bool]: + """Execute a single scanner, emit ScanComplete, report ``(count, failed)``. + + A crashed scanner used to be indistinguishable from a clean scan + that found nothing: both emitted ``item_count=0`` and the browser + said "we scanned and found nothing" over a scanner that had died. + The exception is still swallowed — one broken scanner must not + abort the other five — but it is now REPORTED rather than erased. + """ async def _narrow_emit(event: ScanUpdate) -> None: """Forward ScanUpdate from the scanner to the orchestrator emit.""" await emit(event) + failed = False + error: str | None = None try: count = await module.scan(project_path, _narrow_emit) - except Exception: # noqa: BLE001 + except Exception as exc: # noqa: BLE001 _log.exception("Scanner %s failed", panel) count = 0 + failed = True + error = f"{type(exc).__name__}: {exc}" - await emit(ScanComplete(panel=panel, item_count=count)) - return count + await emit(ScanComplete(panel=panel, item_count=count, failed=failed, error=error)) + return count, failed diff --git a/src/bonfire/onboard/protocol.py b/src/bonfire/onboard/protocol.py index 8c4474a7..4fcf735c 100644 --- a/src/bonfire/onboard/protocol.py +++ b/src/bonfire/onboard/protocol.py @@ -90,18 +90,31 @@ class ScanUpdate(FrontDoorMessage): class ScanComplete(FrontDoorMessage): - """One scan panel has finished.""" + """One scan panel has stopped — successfully or otherwise. + + ``failed``/``error`` separate a scanner that CRASHED from one that + ran clean and found nothing; both used to report only + ``item_count=0``. Both default to the success shape, so a frame + from an older producer still parses as a clean scan. + """ type: Literal["scan_complete"] = "scan_complete" panel: str item_count: int + failed: bool = False + error: str | None = None class AllScansComplete(FrontDoorMessage): - """All scan panels have finished.""" + """All scan panels have stopped. + + ``failed_panels``: an all-crashed run used to summarise as + ``total_items=0``, identical to a clean scan of an empty project. + """ type: Literal["all_scans_complete"] = "all_scans_complete" total_items: int + failed_panels: int = 0 class ConversationStart(FrontDoorMessage): diff --git a/src/bonfire/onboard/ui.html b/src/bonfire/onboard/ui.html index 8b82a770..04220035 100644 --- a/src/bonfire/onboard/ui.html +++ b/src/bonfire/onboard/ui.html @@ -139,6 +139,19 @@ content: ' \2713'; color: var(--green); } + /* A scanner that crashed. Deliberately NOT the completion tick: a + dead scan and an empty one both arrived as item_count=0 before. */ + .scan-panel.failed { + border-color: var(--amber, #ffb000); + transition: border-color 0.5s ease; + } + .scan-panel.failed h2 { + color: var(--amber, #ffb000); + } + .scan-panel.failed h2::after { + content: ' \26A0'; + color: var(--amber, #ffb000); + } @keyframes panel-pulse { 0%, 100% { box-shadow: none; } 50% { box-shadow: 0 0 8px rgba(0, 255, 65, 0.15); } @@ -375,8 +388,8 @@ switch (msg.type) { case 'scan_start': statusEl.textContent = 'scanning ' + msg.panels.length + ' systems...'; break; case 'scan_update': addScanItem(msg.panel, msg.label, msg.value, msg.detail || ''); break; - case 'scan_complete': markPanelComplete(msg.panel, msg.item_count); break; - case 'all_scans_complete': statusEl.textContent = msg.total_items + ' findings catalogued'; break; + case 'scan_complete': markPanelComplete(msg.panel, msg.item_count, msg.failed, msg.error); break; + case 'all_scans_complete': statusEl.textContent = summariseScans(msg.total_items, msg.failed_panels); break; case 'conversation_start': activateChat(); break; case 'falcor_message': if (msg.subtype === 'narration') { setNarration(msg.text); } @@ -400,13 +413,27 @@ el.appendChild(item); } - function markPanelComplete(panel, count) { + // A scanner that DIED must not render as one that found nothing. + // Both used to arrive as item_count=0; 'failed' is what separates them. + function markPanelComplete(panel, count, failed, error) { if (KNOWN_PANELS.indexOf(panel) === -1) return; var panelEl = document.querySelector('[data-panel="' + panel + '"]'); if (!panelEl) return; - panelEl.classList.add('complete'); + panelEl.classList.add(failed ? 'failed' : 'complete'); var h2 = panelEl.querySelector('h2'); - if (h2) h2.textContent += ' [' + count + ']'; + if (!h2) return; + if (failed) { + h2.textContent += ' [scan failed]'; + if (error) panelEl.setAttribute('title', error); + } else { + h2.textContent += ' [' + count + ']'; + } + } + + function summariseScans(total, failedPanels) { + var summary = total + ' findings catalogued'; + if (failedPanels) summary += ' — ' + failedPanels + ' scan(s) failed'; + return summary; } function setNarration(text) { enqueueType(narrationEl, text, 35); } diff --git a/src/bonfire/xp/consumer.py b/src/bonfire/xp/consumer.py index 76a90a78..ae7ea2bf 100644 --- a/src/bonfire/xp/consumer.py +++ b/src/bonfire/xp/consumer.py @@ -20,6 +20,11 @@ from bonfire.xp.calculator import XPCalculator from bonfire.xp.tracker import XPTracker +#: ``PipelineFailed.failed_handler`` sentinel for outer-exception halts, +#: which cannot name a bounce target. Distinct from ``None``, which the +#: schema uses for "no bounce happened". +_OUTER_HALT_SENTINEL = "__outer__" + class XPConsumer: """Subscribes to PipelineCompleted and drives the XP system. @@ -52,6 +57,31 @@ async def _handle_pipeline_completed(self, event: PipelineCompleted) -> None: """ await self.on_pipeline_completed(event, success=True, stages_failed=0) + @classmethod + def _halt_reason(cls, event: PipelineFailed) -> str: + """Render the halt's REAL cause for the emitted penalty event. + + ``PipelineFailed`` already names what broke (``failed_stage``, + ``error_message``) and, on bounce-target halts, which handler + actually died (``failed_handler``). The penalty event used to + discard all three and announce a counted-up substitute — + "Pipeline failed with 1 stage failures" — where the 1 was a + constant, not an observation. A reader was told a number that + was never measured instead of the cause that was. + + Two halt paths (budget-exceeded, outer-exception) legitimately + carry no stage name, so this states no stage rather than + inventing one; ``__outer__`` is a schema sentinel, not a real + bounce target, and is never rendered as one. + """ + detail = event.error_message.strip() if event.error_message else "" + stage = event.failed_stage + handler = event.failed_handler + if handler and handler not in (_OUTER_HALT_SENTINEL, stage): + stage = f"{stage or 'unknown stage'} (bounce target {handler})" + where = f" at {stage}" if stage else "" + return f"Pipeline halted{where}: {detail}" if detail else f"Pipeline halted{where}" + async def _handle_pipeline_failed(self, event: PipelineFailed) -> None: """Bus handler for pipeline failures — applies XP penalty or respawn. @@ -64,6 +94,13 @@ async def _handle_pipeline_failed(self, event: PipelineFailed) -> None: We still build a ``PipelineCompleted``-shaped wrapper so the existing ``on_pipeline_completed`` logic stays the single path. + + ``stages_failed=1`` is deliberately NOT derived from the event: + every ``PipelineFailed`` emit site in ``engine/pipeline.py`` + halts on the first failing stage, so exactly one stage failed + and 1 is the measured truth rather than a placeholder. What WAS + a placeholder is the reason, which is why ``halt_reason`` + carries the event's own account of the failure through. """ compat = PipelineCompleted( session_id=event.session_id, @@ -72,7 +109,12 @@ async def _handle_pipeline_failed(self, event: PipelineFailed) -> None: duration_seconds=event.duration_seconds, stages_completed=event.stages_completed, ) - await self.on_pipeline_completed(compat, success=False, stages_failed=1) + await self.on_pipeline_completed( + compat, + success=False, + stages_failed=1, + halt_reason=self._halt_reason(event), + ) async def on_pipeline_completed( self, @@ -80,6 +122,7 @@ async def on_pipeline_completed( *, success: bool, stages_failed: int, + halt_reason: str | None = None, ) -> None: """Process a pipeline completion event. @@ -87,6 +130,11 @@ async def on_pipeline_completed( event: The PipelineCompleted event from the bus. success: Whether the pipeline succeeded. stages_failed: Number of stages that failed. + halt_reason: The failure's own account of what went wrong, + used verbatim as the emitted penalty's reason. Callers + that have a real cause pass it; when it is absent the + penalty falls back to the stage-failure count, which + describes how many stages failed but not why. """ # Snapshot XP before recording old_xp = self._tracker.total_xp() @@ -118,7 +166,7 @@ async def on_pipeline_completed( ), ) elif not success: - reason = f"Pipeline failed with {stages_failed} stage failures" + reason = halt_reason or f"Pipeline failed with {stages_failed} stage failures" await self._bus.emit( XPPenalty( session_id=event.session_id, diff --git a/tests/unit/test_failure_is_not_a_completion.py b/tests/unit/test_failure_is_not_a_completion.py new file mode 100644 index 00000000..1b8cec50 --- /dev/null +++ b/tests/unit/test_failure_is_not_a_completion.py @@ -0,0 +1,368 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 BonfireAI + +"""A halt must never record itself as a completion. + +Four records had the same shape of defect: the failure path produced an +artifact indistinguishable from the success path, or gave a reason that +was not the reason that occurred. Each test below fails on the +pre-fix behaviour; each is paired with a negative control asserting a +genuine success still records as a success, so the suite cannot be +satisfied by code that simply reports everything as broken. +""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from bonfire.cost.analyzer import CostAnalyzer +from bonfire.cost.consumer import CostLedgerConsumer +from bonfire.cost.models import PipelineRecord +from bonfire.events.bus import EventBus +from bonfire.models.events import ( + PipelineCompleted, + PipelineFailed, + XPAwarded, + XPPenalty, +) +from bonfire.onboard.flow import dispatch_user_message +from bonfire.onboard.orchestrator import _run_one, run_scan +from bonfire.onboard.protocol import UserMessage +from bonfire.xp.calculator import XPCalculator +from bonfire.xp.consumer import XPConsumer +from bonfire.xp.tracker import XPTracker + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_SESSION = "sess-halt-vs-done" + + +def _completed(**over: object) -> PipelineCompleted: + kwargs: dict = { + "session_id": _SESSION, + "sequence": 1, + "total_cost_usd": 1.5, + "duration_seconds": 10.0, + "stages_completed": 1, + } + kwargs.update(over) + return PipelineCompleted(**kwargs) + + +def _failed(**over: object) -> PipelineFailed: + kwargs: dict = { + "session_id": _SESSION, + "sequence": 1, + "failed_stage": "builder", + "error_message": "builder raised RuntimeError: disk full", + "total_cost_usd": 1.5, + "duration_seconds": 10.0, + "stages_completed": 1, + } + kwargs.update(over) + return PipelineFailed(**kwargs) + + +def _rows(path: Path) -> list[dict]: + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + + +def _sink() -> tuple[list, object]: + """A capture list and the async emit/broadcast callable that fills it.""" + captured: list = [] + + async def emit(message: object) -> None: + captured.append(message) + + return captured, emit + + +#: A ledger row as written BEFORE ``outcome`` existed — the shape the +#: operator already has on disk. +_LEGACY_ROW: dict = { + "type": "pipeline", + "timestamp": 1785232588.0, + "session_id": "legacy", + "total_cost_usd": 0.37, + "duration_seconds": 0.0004, + "stages_completed": 2, +} + + +class _Crashy: + """A scanner that dies.""" + + @staticmethod + async def scan(path: Path, emit: object) -> int: + raise RuntimeError("no git binary on PATH") + + +class _Empty: + """A scanner that runs clean and legitimately finds nothing.""" + + @staticmethod + async def scan(path: Path, emit: object) -> int: + return 0 + + +# --------------------------------------------------------------------------- +# Defect 1 — the cost ledger, the operator's record of what they paid for +# --------------------------------------------------------------------------- + + +class TestLedgerDistinguishesHaltFromCompletion: + async def test_halt_and_one_stage_success_are_distinguishable(self, tmp_path: Path) -> None: + """The observed defect: a failed run showed up as '1 stages', exactly + like a run that finished one stage. Rows differing only in timestamp + are not distinguishable to any reader that aggregates by session.""" + ledger = tmp_path / "cost_ledger.jsonl" + consumer = CostLedgerConsumer(ledger_path=ledger) + + await consumer._on_pipeline_completed(_completed()) + await consumer._on_pipeline_failed(_failed()) + + success, failure = _rows(ledger) + distinguishing = {k for k in success if success[k] != failure.get(k)} - {"timestamp"} + assert distinguishing, ( + "halt row and success row differ only by timestamp — the ledger " + f"cannot tell a crash from a completion: {success!r} vs {failure!r}" + ) + assert success["outcome"] == "completed" + assert failure["outcome"] == "failed" + + async def test_halt_row_carries_the_reason_that_actually_occurred(self, tmp_path: Path) -> None: + ledger = tmp_path / "cost_ledger.jsonl" + consumer = CostLedgerConsumer(ledger_path=ledger) + + await consumer._on_pipeline_failed(_failed()) + + (row,) = _rows(ledger) + assert row["failed_stage"] == "builder" + assert "disk full" in row["error_message"] + + async def test_success_still_records_as_success(self, tmp_path: Path) -> None: + """Negative control: the fix must not make completions look like halts.""" + ledger = tmp_path / "cost_ledger.jsonl" + consumer = CostLedgerConsumer(ledger_path=ledger) + + await consumer._on_pipeline_completed(_completed()) + + (row,) = _rows(ledger) + assert row["outcome"] == "completed" + assert row["failed_stage"] is None + assert row["error_message"] is None + assert row["stages_completed"] == 1 + assert row["total_cost_usd"] == 1.5 + + def test_rows_written_before_the_field_existed_still_load(self) -> None: + """Migration: history on disk predates ``outcome``. Those rows must keep + validating, and must NOT be fabricated into successes.""" + record = PipelineRecord.model_validate(_LEGACY_ROW) + assert record.outcome == "unknown", ( + "a row that never recorded an outcome must not claim one; " + "defaulting to 'completed' would fabricate the very history " + "this defect corrupted" + ) + + def test_legacy_rows_still_aggregate_through_the_analyzer(self, tmp_path: Path) -> None: + """The sole reader must not start skipping pre-migration rows.""" + ledger = tmp_path / "cost_ledger.jsonl" + ledger.write_text(json.dumps(_LEGACY_ROW) + "\n") + sessions = CostAnalyzer(ledger_path=ledger).all_sessions() + assert [s.session_id for s in sessions] == ["legacy"] + assert sessions[0].stages_completed == 2 + + +# --------------------------------------------------------------------------- +# Defect 2 — the XP substrate. The reported defect was NOT what is broken. +# --------------------------------------------------------------------------- + + +class TestXPPenaltyStatesTheRealReason: + @staticmethod + def _consumer(tmp_path: Path) -> tuple[EventBus, list[object], XPTracker]: + bus = EventBus() + seen: list[object] = [] + + async def capture(event: object) -> None: + seen.append(event) + + bus.subscribe(XPAwarded, capture) + bus.subscribe(XPPenalty, capture) + tracker = XPTracker(tmp_path) + XPConsumer(tracker=tracker, calculator=XPCalculator(), bus=bus) + return bus, seen, tracker + + async def test_penalty_reason_names_the_stage_and_error(self, tmp_path: Path) -> None: + """The reason was invented from a hardcoded count while the event's own + ``failed_stage``/``error_message`` were discarded.""" + bus, seen, _ = self._consumer(tmp_path) + + await bus.emit(_failed()) + await asyncio.sleep(0) + + (penalty,) = seen + assert isinstance(penalty, XPPenalty) + assert "builder" in penalty.reason + assert "disk full" in penalty.reason + assert "1 stage failures" not in penalty.reason + + async def test_stageless_halt_does_not_invent_a_stage(self, tmp_path: Path) -> None: + """Budget-exceeded and outer-exception halts carry no stage name.""" + bus, seen, _ = self._consumer(tmp_path) + + await bus.emit(_failed(failed_stage="", error_message="Budget exceeded: $9.00 > $5.00")) + await asyncio.sleep(0) + + (penalty,) = seen + assert "Budget exceeded" in penalty.reason + assert "unknown stage" not in penalty.reason + + async def test_outer_sentinel_is_not_rendered_as_a_bounce_target(self, tmp_path: Path) -> None: + bus, seen, _ = self._consumer(tmp_path) + + await bus.emit(_failed(failed_handler="__outer__")) + await asyncio.sleep(0) + + (penalty,) = seen + assert "__outer__" not in penalty.reason + + async def test_bounce_target_is_named_when_it_is_a_real_handler(self, tmp_path: Path) -> None: + bus, seen, _ = self._consumer(tmp_path) + + await bus.emit(_failed(failed_handler="sage")) + await asyncio.sleep(0) + + (penalty,) = seen + assert "sage" in penalty.reason + + async def test_xp_store_already_recorded_failure_honestly(self, tmp_path: Path) -> None: + """Control on the REPORTED defect, which did not reproduce: the store + already carried ``success=False``. This pins that it stays true.""" + bus, _, tracker = self._consumer(tmp_path) + + await bus.emit(_failed()) + await asyncio.sleep(0) + + (event,) = tracker.events() + assert event["success"] is False + + async def test_success_still_awards_and_reads_as_success(self, tmp_path: Path) -> None: + """Negative control for the whole XP path.""" + bus, seen, tracker = self._consumer(tmp_path) + + await bus.emit(_completed()) + await asyncio.sleep(0) + + (award,) = seen + assert isinstance(award, XPAwarded) + assert "halted" not in award.reason + assert tracker.events()[0]["success"] is True + + +# --------------------------------------------------------------------------- +# Defect 3 — a scanner that crashed is not a scanner that found nothing +# --------------------------------------------------------------------------- + + +class TestCrashedScannerIsNotAnEmptyScan: + async def test_crash_and_empty_scan_are_distinguishable(self) -> None: + crashed, crash_emit = _sink() + empty, empty_emit = _sink() + await _run_one("git_state", _Crashy, Path("."), crash_emit) + await _run_one("git_state", _Empty, Path("."), empty_emit) + + (crash_frame,) = crashed + (empty_frame,) = empty + assert crash_frame.model_dump() != empty_frame.model_dump(), ( + "a dead scanner and an empty one emit identical frames — the " + "browser cannot tell 'the scan died' from 'we found nothing'" + ) + assert (crash_frame.failed, empty_frame.failed) == (True, False) + assert "RuntimeError" in crash_frame.error + assert "no git binary" in crash_frame.error + + async def test_clean_scan_is_untouched(self) -> None: + """Negative control: a working scanner still reports its count.""" + sink, emit = _sink() + result = await _run_one("git_state", _Empty, Path("."), emit) + + assert result == (0, False) + assert sink[0].error is None + + @pytest.mark.parametrize( + ("scanners", "expected_failed"), + [ + ([("git_state", _Crashy), ("mcp_servers", _Empty)], 1), + ([("git_state", _Empty), ("mcp_servers", _Empty)], 0), + ], + ) + async def test_summary_frame_counts_failed_panels( + self, monkeypatch, scanners: list, expected_failed: int + ) -> None: + """The same defect one level up: an all-crashed run summarised as + ``total_items=0``, identical to a clean scan of an empty project. + The zero-failure row is the negative control.""" + monkeypatch.setattr("bonfire.onboard.orchestrator._get_scanners", lambda: scanners) + sink, emit = _sink() + await run_scan(Path("."), emit) + + assert sink[-1].total_items == 0 + assert sink[-1].failed_panels == expected_failed + + +# --------------------------------------------------------------------------- +# Defect 4 — a wrong reason sends the user to fix something that is not broken +# --------------------------------------------------------------------------- + + +class TestFrameRejectionNamesTheRealCause: + @staticmethod + async def _reject(frame: dict) -> object: + sink, broadcast = _sink() + await dispatch_user_message( + frame, + conversation=None, + broadcast=broadcast, + conversation_done=asyncio.Event(), + ) + return sink[0] + + @pytest.mark.parametrize( + ("label", "frame"), + [ + ("missing text", {"type": "user_message"}), + ("text is a number", {"type": "user_message", "text": 12345}), + ], + ) + async def test_non_length_failures_are_not_called_too_long( + self, label: str, frame: dict + ) -> None: + error = await self._reject(frame) + assert error.code != "message_too_long", ( + f"{label}: reported as a length problem over a message that was " + "never long — the user is sent to fix something that is not broken" + ) + assert error.code == "invalid_message" + + async def test_missing_field_says_which_field(self) -> None: + error = await self._reject({"type": "user_message"}) + assert "text" in error.message + + async def test_genuinely_too_long_still_says_too_long(self) -> None: + """Negative control: the real length violation keeps its own code, and + the existing cap contract is unchanged.""" + error = await self._reject({"type": "user_message", "text": "x" * 9000}) + assert error.code == "message_too_long" + assert "8 KiB" in error.message + + def test_the_length_cap_itself_is_still_enforced(self) -> None: + with pytest.raises(ValidationError): + UserMessage.model_validate({"type": "user_message", "text": "x" * 9000})