diff --git a/docs/phase-568-slice-3-run-projector-plan.md b/docs/phase-568-slice-3-run-projector-plan.md new file mode 100644 index 00000000..8ffb1eca --- /dev/null +++ b/docs/phase-568-slice-3-run-projector-plan.md @@ -0,0 +1,970 @@ +# Issue #568 Slice 3: Run Snapshot Projector - Execution Plan + +Goal: implement the approved spec at `docs/phase-568-slice-3-run-projector.md` by adding one pure projector module (`src/brigade/run_projector.py`), its 20 tests, and two golden fixtures, with zero changes to any existing writer, reader, or CLI path. + +Architecture: `project_run_snapshot(base_snapshot, events, *, journal_present)` re-validates every envelope through `run_events.validate_event`, re-verifies the chain (contiguous sequence from 1, digest linkage, single `run_id`), then derives exactly five fields (`status`, `projector_version`, `journal_present`, `journal_last_sequence`, `journal_last_event_digest`) while deep-copying the 44 preserved fields verbatim from the base. Output bytes come from the single encoding path `(json.dumps(snapshot, indent=2, sort_keys=True) + "\n").encode("utf-8")`, matching `aboyeur._write_json` byte for byte. The module performs no I/O, reads no clock or environment, and nothing in the runtime calls it in this slice. + +The spec is the authority. This plan quotes the spec's API and the real source contracts in `src/brigade/run_events.py`, `src/brigade/run_journal.py`, `src/brigade/run_lifecycle.py`, `src/brigade/aboyeur.py`, `src/brigade/run_resume.py`, and `src/brigade/runguard.py` as of the base commit. If any quoted contract disagrees with the checked-out code, stop and re-read the code. Do not improvise. + +## File map + +| Path | Action | +| --- | --- | +| `tests/test_run_projector.py` | Create first (RED). 20 test functions, listed below | +| `src/brigade/run_projector.py` | Create. The entire production surface of this slice | +| `tests/fixtures/run-lifecycle/golden-projection.base.json` | Create. Hand-authored base snapshot carrying all 44 preserved fields | +| `tests/fixtures/run-lifecycle/golden-projection.expected.json` | Create. Generated once by the implementation, reviewed by hand before commit | + +Do not edit `src/brigade/aboyeur.py`, `src/brigade/run_lifecycle.py`, `src/brigade/run_journal.py`, `src/brigade/run_events.py`, any CLI module, `pyproject.toml`, or any other existing file. The existing fixture `tests/fixtures/run-lifecycle/golden-lifecycle.jsonl` is read, never modified. + +## Repository conventions used below + +- Tests run with `python3 -m pytest` from the repo root. `tests/conftest.py` inserts `src/` on `sys.path`, so no install step is needed. Ad-hoc scripts outside pytest need `PYTHONPATH=src`. +- Focused runs name the test file: `python3 -m pytest tests/test_run_projector.py -q`. +- Lint and type gates: `python3 -m ruff check ` (line length 120, rules B, E4, E7, E9, F) and `python3 -m mypy src/brigade/run_projector.py` (new modules are checked from day one. Only legacy modules carry overrides in `pyproject.toml`). +- Fixtures for this work live under `tests/fixtures/run-lifecycle/`, beside `golden-lifecycle.jsonl` (6 events, run_id `20260727-153045-a1b2c3d4`, final event `run.completed` with `payload.status == "ok"`, final `event_digest` `176259cd00408b637f1c13496fcdfafea7dd96753fb8a9e76dbea58817442fc7`). + +## Task 1: RED tests + +- [x] Create `tests/test_run_projector.py` with the approved RED coverage: + +```python +"""Tests for brigade.run_projector pure run snapshot projector (issue #568 slice 3). + +Covers the slice-3 contract: golden replay against the committed lifecycle +journal, determinism, empty-sequence journal presence semantics, the +event-to-status mapping with its payload status rules, closed field ownership +over run.json, fail-closed chain verification, preservation deep-equality, +re-projection idempotence, encoding parity with aboyeur._write_json, and +bounded typed errors for every failure mode. +""" + +from __future__ import annotations + +import dataclasses +import json +from pathlib import Path + +import pytest + +from brigade import run_events, run_journal, run_projector + +FIXTURES = Path(__file__).resolve().parent / "fixtures" / "run-lifecycle" +GOLDEN_LIFECYCLE_PATH = FIXTURES / "golden-lifecycle.jsonl" +GOLDEN_BASE_PATH = FIXTURES / "golden-projection.base.json" +GOLDEN_EXPECTED_PATH = FIXTURES / "golden-projection.expected.json" + +RUN_ID = "20260727-153045-a1b2c3d4" +RECORDED_AT = "2026-07-27T15:30:45.123456Z" +GOLDEN_FINAL_DIGEST = "176259cd00408b637f1c13496fcdfafea7dd96753fb8a9e76dbea58817442fc7" + + +def _load_golden_base() -> dict: + if not GOLDEN_BASE_PATH.is_file(): + pytest.fail(f"missing golden fixture: {GOLDEN_BASE_PATH}") + return json.loads(GOLDEN_BASE_PATH.read_text(encoding="utf-8")) + + +def _read_golden_events() -> list[run_journal.RunEvent]: + if not GOLDEN_LIFECYCLE_PATH.is_file(): + pytest.fail(f"missing golden fixture: {GOLDEN_LIFECYCLE_PATH}") + report = run_journal.read_journal(GOLDEN_LIFECYCLE_PATH) + assert report.partial_tail is None + assert report.chain_errors == [] + return report.events + + +def _minimal_base() -> dict: + return { + "schema": "brigade.run.v1", + "schema_version": 1, + "task": "project the lifecycle journal", + "orchestrator": "chef", + "dry_run": False, + "read_only": False, + "status": "started", + "started_at": "2026-07-27T15:30:45.123456+00:00", + "status_started_at": "2026-07-27T15:30:45.123456+00:00", + "suspected_noop": False, + "code_graph_brief": {"attached": False, "bytes": 0}, + "drift_impact_brief": {"attached": False, "bytes": 0, "pending_count": 0}, + "evidence_brief": {"attached": False, "bytes": 0}, + "brief_budget": {"bytes": 8192, "attached": []}, + } + + +def _build_chain( + steps: list[tuple[str, dict]], + *, + run_id: str = RUN_ID, + start_sequence: int = 1, + previous_digest: str | None = None, +) -> list[dict]: + envelopes = [] + sequence = start_sequence + for event_type, payload in steps: + env = run_events.build_event( + run_id=run_id, + sequence=sequence, + event_type=event_type, + payload=payload, + idempotency_key=f"test-{run_id}-{sequence}", + recorded_at=RECORDED_AT, + previous_digest=previous_digest, + ) + envelopes.append(env) + previous_digest = env["event_digest"] + sequence += 1 + return envelopes + + +def _failed_chain(status_payload: dict) -> list[dict]: + return _build_chain( + [ + ("run.created", {"status": "started"}), + ("run.failed", status_payload), + ] + ) + + +def test_golden_replay_matches_expected_bytes(): + if not GOLDEN_EXPECTED_PATH.is_file(): + pytest.fail(f"missing golden fixture: {GOLDEN_EXPECTED_PATH}") + projection = run_projector.project_run_snapshot( + _load_golden_base(), _read_golden_events(), journal_present=True + ) + assert projection.to_bytes() == GOLDEN_EXPECTED_PATH.read_bytes() + + +def test_golden_projection_is_byte_deterministic(): + base = _load_golden_base() + events = _read_golden_events() + first = run_projector.project_run_snapshot(base, events, journal_present=True) + second = run_projector.project_run_snapshot(base, events, journal_present=True) + assert first.to_bytes() == second.to_bytes() + + +def test_empty_sequence_without_journal_preserves_base(): + base = _minimal_base() + projection = run_projector.project_run_snapshot(base, [], journal_present=False) + assert projection.status == "started" + assert projection.journal_present is False + assert projection.last_sequence == 0 + assert projection.last_event_digest is None + snapshot = projection.snapshot + assert snapshot["status"] == "started" + assert snapshot["projector_version"] == 1 + assert snapshot["journal_present"] is False + assert snapshot["journal_last_sequence"] == 0 + assert snapshot["journal_last_event_digest"] is None + for key, value in base.items(): + assert snapshot[key] == value + assert snapshot["code_graph_brief"] is not base["code_graph_brief"] + + +def test_empty_created_journal_projects_presence_without_facts(): + projection = run_projector.project_run_snapshot(_minimal_base(), [], journal_present=True) + assert projection.journal_present is True + assert projection.last_sequence == 0 + assert projection.last_event_digest is None + assert projection.status == "started" + assert projection.snapshot["journal_present"] is True + assert projection.snapshot["journal_last_sequence"] == 0 + assert projection.snapshot["journal_last_event_digest"] is None + + +def test_full_golden_sequence_derives_terminal_facts(): + projection = run_projector.project_run_snapshot( + _load_golden_base(), _read_golden_events(), journal_present=True + ) + assert projection.status == "ok" + assert projection.last_sequence == 6 + assert projection.last_event_digest == GOLDEN_FINAL_DIGEST + + +def test_run_failed_timeout_payload_derives_timeout(): + chain = _failed_chain({"status": "timeout", "detail": "bounded"}) + projection = run_projector.project_run_snapshot(_minimal_base(), chain, journal_present=True) + assert projection.status == "timeout" + assert projection.last_sequence == 2 + + +def test_run_failed_failed_payload_derives_failed(): + chain = _failed_chain({"status": "failed", "detail": "bounded"}) + projection = run_projector.project_run_snapshot(_minimal_base(), chain, journal_present=True) + assert projection.status == "failed" + assert projection.last_sequence == 2 + + +@pytest.mark.parametrize("payload", [{"status": "ok"}, {"detail": "bounded"}]) +def test_run_failed_bad_or_missing_payload_status_raises(payload): + chain = _failed_chain(payload) + with pytest.raises(run_projector.EventPayloadError) as excinfo: + run_projector.project_run_snapshot(_minimal_base(), chain, journal_present=True) + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + assert "run.failed" in excinfo.value.diagnostic + + +def test_run_interrupted_derives_canceled_and_wrong_status_raises(): + chain = _build_chain( + [ + ("run.created", {"status": "started"}), + ("run.interrupted", {"status": "canceled"}), + ] + ) + projection = run_projector.project_run_snapshot(_minimal_base(), chain, journal_present=True) + assert projection.status == "canceled" + + bad = _build_chain( + [ + ("run.created", {"status": "started"}), + ("run.interrupted", {"status": "failed"}), + ] + ) + with pytest.raises(run_projector.EventPayloadError) as excinfo: + run_projector.project_run_snapshot(_minimal_base(), bad, journal_present=True) + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + + +def test_unknown_base_key_raises_unknown_snapshot_field(): + base = _minimal_base() + base["mystery_key"] = 1 + with pytest.raises(run_projector.UnknownSnapshotFieldError) as excinfo: + run_projector.project_run_snapshot(base, [], journal_present=False) + assert "mystery_key" in excinfo.value.diagnostic + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + + +def test_registered_unmapped_event_type_raises(): + chain = _build_chain( + [ + ("run.created", {"status": "started"}), + ("run.paused", {"approval_id": "ap-1", "reason": "review"}), + ] + ) + with pytest.raises(run_projector.UnmappedEventTypeError) as excinfo: + run_projector.project_run_snapshot(_minimal_base(), chain, journal_present=True) + assert "run.paused" in excinfo.value.diagnostic + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + + +def test_chain_breaks_raise_event_chain_error(): + base = _minimal_base() + first = _build_chain([("run.created", {"status": "started"})]) + + gap = first + _build_chain( + [("run.planning.started", {"detail": "planning"})], + start_sequence=3, + previous_digest=first[0]["event_digest"], + ) + with pytest.raises(run_projector.EventChainError): + run_projector.project_run_snapshot(base, gap, journal_present=True) + + duplicate = first + first + with pytest.raises(run_projector.EventChainError): + run_projector.project_run_snapshot(base, duplicate, journal_present=True) + + broken = first + _build_chain( + [("run.planning.started", {"detail": "planning"})], + start_sequence=2, + previous_digest="0" * 64, + ) + with pytest.raises(run_projector.EventChainError): + run_projector.project_run_snapshot(base, broken, journal_present=True) + + mixed = first + _build_chain( + [("run.planning.started", {"detail": "planning"})], + run_id="20260727-153045-b2c3d4e5", + start_sequence=2, + previous_digest=first[0]["event_digest"], + ) + with pytest.raises(run_projector.EventChainError): + run_projector.project_run_snapshot(base, mixed, journal_present=True) + + +def test_invalid_envelope_mapping_raises_event_chain_error(): + env = dict(_build_chain([("run.created", {"status": "started"})])[0]) + env["event_type"] = "run.unknown" + with pytest.raises(run_projector.EventChainError): + run_projector.project_run_snapshot(_minimal_base(), [env], journal_present=True) + + +def test_invalid_envelope_diagnostic_excludes_rejected_value(): + private_marker = "private-marker-568" + env = dict(_build_chain([("run.created", {"status": "started"})])[0]) + env["recorded_at"] = private_marker + with pytest.raises(run_projector.EventChainError) as excinfo: + run_projector.project_run_snapshot(_minimal_base(), [env], journal_present=True) + assert private_marker not in excinfo.value.diagnostic + + +def test_mutated_typed_run_event_raises_event_chain_error(): + env = _build_chain([("run.created", {"status": "started"})])[0] + event = run_journal.RunEvent(**env) + mutated = dataclasses.replace(event, payload={"status": "ok"}) + with pytest.raises(run_projector.EventChainError): + run_projector.project_run_snapshot(_minimal_base(), [mutated], journal_present=True) + + +def test_full_coverage_base_preservation_deep_equals(): + base = _load_golden_base() + assert run_projector.PRESERVED_FIELDS <= set(base.keys()) + projection = run_projector.project_run_snapshot(base, [], journal_present=False) + snapshot = projection.snapshot + for key in run_projector.PRESERVED_FIELDS: + assert snapshot[key] == base[key] + preserved_in_output = set(snapshot.keys()) & run_projector.PRESERVED_FIELDS + assert preserved_in_output == set(base.keys()) & run_projector.PRESERVED_FIELDS + assert snapshot["failure"] is not base["failure"] + assert snapshot["recovery_history"] is not base["recovery_history"] + assert snapshot["active_seats"] is not base["active_seats"] + + +def test_reprojection_is_idempotent(): + events = _read_golden_events() + first = run_projector.project_run_snapshot(_load_golden_base(), events, journal_present=True) + second = run_projector.project_run_snapshot(first.snapshot, events, journal_present=True) + assert first.to_bytes() == second.to_bytes() + + +def test_to_bytes_matches_run_json_encoding(): + projection = run_projector.project_run_snapshot(_minimal_base(), [], journal_present=False) + expected = (json.dumps(projection.snapshot, indent=2, sort_keys=True) + "\n").encode("utf-8") + assert projection.to_bytes() == expected + assert run_projector.encode_snapshot_bytes(projection.snapshot) == expected + + +def test_unencodable_preserved_value_raises_snapshot_encoding_error(): + base = _minimal_base() + base["failure"] = {"phase": "dispatch", "detail": object()} + with pytest.raises(run_projector.SnapshotEncodingError) as excinfo: + run_projector.project_run_snapshot(base, [], journal_present=False) + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + + circular = _minimal_base() + loop: dict = {} + loop["self"] = loop + circular["failure"] = loop + with pytest.raises(run_projector.SnapshotEncodingError) as excinfo: + run_projector.project_run_snapshot(circular, [], journal_present=False) + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + assert "Circular" not in excinfo.value.diagnostic + + +@pytest.mark.parametrize("value", ["yes", 1, None]) +def test_non_boolean_journal_present_raises_projection_input_error(value): + with pytest.raises(run_projector.ProjectionInputError): + run_projector.project_run_snapshot(_minimal_base(), [], journal_present=value) +``` + +- [x] Run the RED command and confirm the expected failure: + +```bash +python3 -m pytest tests/test_run_projector.py -q +``` + +Expected: collection fails with `ModuleNotFoundError: No module named 'brigade.run_projector'`. Any other failure means the test file has a syntax or import error. Fix the test file before continuing. + +## Task 2: production module + +- [x] Create `src/brigade/run_projector.py` with exactly this content: + +```python +"""Pure run snapshot projector (issue #568, slice 3). + +Derives a complete ``run.json`` snapshot from a base snapshot plus a verified +lifecycle event sequence. The projector owns exactly five derived fields +(``status``, ``projector_version``, ``journal_present``, +``journal_last_sequence``, ``journal_last_event_digest``) and deep-copies the +44 preserved fields of the current ``run.json`` contract verbatim from the +base. Every envelope is re-validated through ``run_events.validate_event`` +and the chain is re-verified (contiguous sequence from 1, previous-digest +linkage, single run_id) even when the caller already read the events through +``run_journal.read_journal``. Any deviation fails closed with a bounded typed +error; there is no partial projection. + +The projector performs no I/O, reads no clock, reads no environment, and holds +no mutable module state. Nothing in the runtime calls it in this slice. +Standard library only. Brigade is zero-runtime-dependency. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from copy import deepcopy +from dataclasses import dataclass +from typing import Any + +from brigade import run_events, run_journal + +PROJECTOR_VERSION: int = 1 + +# Field ownership over the run.json contract. Every current run.json key is +# in exactly one of these two sets; any other base key stops projection. +DERIVED_FIELDS: frozenset[str] = frozenset( + { + "status", + "projector_version", + "journal_present", + "journal_last_sequence", + "journal_last_event_digest", + } +) + +PRESERVED_FIELDS: frozenset[str] = frozenset( + { + # Identity and schema + "schema", + "schema_version", + "task", + "orchestrator", + "roster", + "worker", + "scheduler", + # Run configuration + "dry_run", + "read_only", + "cwd", + "lock_workspace", + "codex_transport", + "lifecycle_journal_requested", + # Timing + "started_at", + "status_started_at", + "finished_at", + "duration_seconds", + "resumed_at", + # Recovery + "recovery_history", + "recovery_preserved_artifact", + # Briefs and routing telemetry + "code_graph_brief", + "drift_impact_brief", + "evidence_brief", + "brief_budget", + "route", + "skill_route_policy", + "pre_run_snapshot", + "git", + "code_graph_delta", + "context_eval", + "suspected_noop", + # Control transport + "control_transport", + "control_socket", + # Live progress (record_dispatch_stage, record_result_processing) + "active_stage", + "active_seats", + "phase_owner", + # Outcome and failure + "error", + "failure_phase", + "failure_kind", + "failure", + "transport_warning", + "artifact_collection", + # Artifact references + "artifacts", + "handoff", + } +) + +OWNED_FIELDS: frozenset[str] = DERIVED_FIELDS | PRESERVED_FIELDS + +# Payload-independent rows of the event-to-status mapping. The remaining four +# mapped event types carry their run.json status in the payload and are +# handled by _EVENT_PAYLOAD_STATUS. +EVENT_STATUS: dict[str, str] = { + "run.planning.started": "planning", + "run.dispatch.requested": "dispatching", + "run.dispatch.completed": "result-processing", + "run.synthesis.started": "synthesizing", + "run.synthesis.completed": "handoff", +} + +# Payload-dependent rows: event_type -> allowed payload status values. The +# derived status is the payload value itself after the membership check. +_EVENT_PAYLOAD_STATUS: dict[str, frozenset[str]] = { + "run.created": frozenset({"started"}), + "run.completed": frozenset({"ok"}), + "run.failed": frozenset({"failed", "timeout"}), + "run.interrupted": frozenset({"canceled"}), +} + +_ENCODING_CATEGORY = "snapshot cannot be encoded under the run.json JSON contract" + + +class ProjectionError(RuntimeError): + """Base class for projection failures. Carries a bounded ``diagnostic``.""" + + def __init__(self, diagnostic: str) -> None: + super().__init__(diagnostic) + self.diagnostic = diagnostic + + +class UnknownSnapshotFieldError(ProjectionError): + """The base snapshot carried a key outside ``OWNED_FIELDS``.""" + + +class UnmappedEventTypeError(ProjectionError): + """A validated event's type has no status mapping.""" + + +class EventChainError(ProjectionError): + """Envelope validation or chain verification failed.""" + + +class EventPayloadError(ProjectionError): + """A payload status rule failed for a mapped event type.""" + + +class ProjectionInputError(ProjectionError): + """An input argument violated the projector contract.""" + + +class SnapshotEncodingError(ProjectionError): + """Projected values cannot be encoded by the run.json JSON contract.""" + + +@dataclass(frozen=True) +class RunProjection: + """Result of a successful projection.""" + + snapshot: dict[str, Any] + status: str + journal_present: bool + last_sequence: int + last_event_digest: str | None + + def to_bytes(self) -> bytes: + return encode_snapshot_bytes(self.snapshot) + + +def encode_snapshot_bytes(snapshot: Mapping[str, Any]) -> bytes: + """Encode a snapshot with the existing run.json byte contract. + + Sorted keys, two-space indent, trailing newline, UTF-8, matching + ``aboyeur._write_json`` byte for byte. Encoding failures (non-JSON values, + circular references, recursion or overflow) surface as a bounded + category-only SnapshotEncodingError, never as a raw serializer exception. + """ + try: + return (json.dumps(snapshot, indent=2, sort_keys=True) + "\n").encode("utf-8") + except (TypeError, ValueError, OverflowError, RecursionError) as exc: + raise SnapshotEncodingError(_ENCODING_CATEGORY) from exc + + +def _validated_envelopes( + events: Sequence[run_journal.RunEvent | Mapping[str, Any]], +) -> list[Mapping[str, Any]]: + """Normalize every item to an envelope mapping and re-validate it.""" + if isinstance(events, (str, bytes)) or not isinstance(events, Sequence): + raise ProjectionInputError("events must be a sequence of envelopes") + envelopes: list[Mapping[str, Any]] = [] + for index, item in enumerate(events, start=1): + env = item.to_dict() if isinstance(item, run_journal.RunEvent) else item + errors = run_events.validate_event(env) + if errors: + raise EventChainError(run_events._bound(f"event envelope invalid at sequence {index}")) + envelopes.append(env) + return envelopes + + +def _event_status(env: Mapping[str, Any]) -> str: + """Map one validated envelope to its run.json status, enforcing payload rules.""" + event_type = env["event_type"] + static = EVENT_STATUS.get(event_type) + if static is not None: + return static + expected = _EVENT_PAYLOAD_STATUS.get(event_type) + if expected is None: + raise UnmappedEventTypeError( + run_events._bound(f"event type {event_type} has no status mapping at sequence {env['sequence']}") + ) + payload = env["payload"] + value = payload.get("status") + if not isinstance(value, str) or value not in expected: + wanted = ", ".join(sorted(expected)) + raise EventPayloadError( + run_events._bound(f"event type {event_type} requires payload status in {{{wanted}}}") + ) + return value + + +def project_run_snapshot( + base_snapshot: Mapping[str, Any], + events: Sequence[run_journal.RunEvent | Mapping[str, Any]], + *, + journal_present: bool, +) -> RunProjection: + """Project a complete run.json snapshot from a base plus a verified event sequence. + + Fail closed: unknown base keys, invalid envelopes, chain breaks, mixed + run_id values, unmapped event types, payload rule violations, and + unencodable preserved values each raise a typed ProjectionError subclass + with a bounded diagnostic. An empty event sequence is valid and preserves + status from the base. + """ + if not isinstance(journal_present, bool): + raise ProjectionInputError("journal_present must be a real boolean") + if not isinstance(base_snapshot, Mapping): + raise ProjectionInputError("base snapshot must be a mapping") + + unknown = sorted(set(base_snapshot.keys()) - OWNED_FIELDS) + if unknown: + raise UnknownSnapshotFieldError( + run_events._bound("base snapshot carries unmapped fields: " + ", ".join(unknown[:8])) + ) + + envelopes = _validated_envelopes(events) + + last_sequence = 0 + last_event_digest: str | None = None + run_id: str | None = None + status: str | None = None + for env in envelopes: + sequence = env["sequence"] + if sequence != last_sequence + 1: + raise EventChainError(run_events._bound(f"event chain sequence break at sequence {sequence}")) + previous_digest = env["previous_digest"] + if sequence == 1: + if previous_digest is not None: + raise EventChainError("event chain sequence 1 previous_digest must be null") + elif previous_digest != last_event_digest: + raise EventChainError(run_events._bound(f"event chain digest link break at sequence {sequence}")) + if run_id is None: + run_id = env["run_id"] + elif env["run_id"] != run_id: + raise EventChainError(run_events._bound(f"event chain mixes run_id at sequence {sequence}")) + status = _event_status(env) + last_sequence = sequence + last_event_digest = env["event_digest"] + + if status is None: + base_status = base_snapshot.get("status") + if not isinstance(base_status, str): + raise ProjectionInputError("base snapshot must carry a string status when events is empty") + status = base_status + + snapshot: dict[str, Any] = {} + for key in PRESERVED_FIELDS: + if key in base_snapshot: + snapshot[key] = deepcopy(base_snapshot[key]) + snapshot["status"] = status + snapshot["projector_version"] = PROJECTOR_VERSION + snapshot["journal_present"] = journal_present + snapshot["journal_last_sequence"] = last_sequence + snapshot["journal_last_event_digest"] = last_event_digest + + # Fail closed at projection time on preserved values the run.json JSON + # contract cannot encode; to_bytes() re-encodes deterministically. + encode_snapshot_bytes(snapshot) + + return RunProjection( + snapshot=snapshot, + status=status, + journal_present=journal_present, + last_sequence=last_sequence, + last_event_digest=last_event_digest, + ) + + +__all__ = [ + "DERIVED_FIELDS", + "EVENT_STATUS", + "OWNED_FIELDS", + "PRESERVED_FIELDS", + "PROJECTOR_VERSION", + "EventChainError", + "EventPayloadError", + "ProjectionError", + "ProjectionInputError", + "RunProjection", + "SnapshotEncodingError", + "UnknownSnapshotFieldError", + "UnmappedEventTypeError", + "encode_snapshot_bytes", + "project_run_snapshot", +] +``` + +Implementation notes that resolve the spec's implicit corners (these are decisions, not open questions): + +- The empty-sequence rule preserves `status` from the base. Because the invariants require all five derived fields in every output and every failure to be a `ProjectionError` subclass, a base without a string `status` on an empty sequence raises `ProjectionInputError` (category-only diagnostic). Every real base from `aboyeur._run_payload` carries a string status, so this only fires on synthetic input. +- Every event in the sequence must map to a status. The mapping walk computes each event's status (raising `UnmappedEventTypeError` or `EventPayloadError` on the offending event) and keeps the final one, matching the spec's "nothing is skipped silently" rule. +- Diagnostics follow the spec's bounded-error table: categories plus field names, event types, sequences, and expected value sets only. Raw payload values, envelope bytes, paths, and snapshot contents never appear. Bounding reuses `run_events._bound` (240 chars, ellipsis), the same cross-module reuse `run_lifecycle.py` already makes. + +- [x] Run the focused suite and confirm the expected intermediate state: + +```bash +python3 -m pytest tests/test_run_projector.py -q +``` + +Observed: 22 passed, 1 failed. The only failure was `Failed: missing golden base fixture: .../golden-projection.base.json` from the golden replay test. Any other failure is a real bug in the module. + +## Task 3: golden base fixture + +- [x] Create `tests/fixtures/run-lifecycle/golden-projection.base.json` with exactly this content (all 44 preserved fields, nested objects, lists, and optional fields, plus the derived `status` key the live writer always carries): + +```json +{ + "schema": "brigade.run.v1", + "schema_version": 1, + "task": "Project the lifecycle journal", + "orchestrator": "chef", + "roster": { + "path": ".brigade/roster.toml", + "source": "project", + "shadowed": [] + }, + "worker": "coder", + "scheduler": { + "requested": "auto", + "used": "serial" + }, + "dry_run": false, + "read_only": false, + "cwd": "/work/repo", + "lock_workspace": "/work/repo", + "codex_transport": "app-server", + "lifecycle_journal_requested": true, + "started_at": "2026-07-27T15:30:45.123456+00:00", + "status_started_at": "2026-07-27T15:30:49.000000+00:00", + "finished_at": "2026-07-27T15:30:50.000000+00:00", + "duration_seconds": 4.877, + "resumed_at": [ + "2026-07-27T15:31:00.000000+00:00", + "2026-07-27T15:32:00.000000+00:00" + ], + "recovery_history": [ + { + "kind": "worker-error", + "at": "2026-07-27T15:31:30.000000+00:00", + "seat": "coder" + } + ], + "recovery_preserved_artifact": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4/run.json.corrupt", + "code_graph_brief": { + "attached": true, + "bytes": 2048 + }, + "drift_impact_brief": { + "attached": false, + "bytes": 0, + "pending_count": 0 + }, + "evidence_brief": { + "attached": true, + "bytes": 512 + }, + "brief_budget": { + "bytes": 8192, + "attached": [ + "code-graph" + ] + }, + "route": { + "skill": "brigade-work", + "matched": [ + "run" + ], + "score": 3 + }, + "skill_route_policy": { + "policy_applied": true, + "extensions": [ + "force-verify" + ] + }, + "pre_run_snapshot": { + "head": "abc123", + "dirty": false + }, + "git": { + "head": "abc123", + "branch": "main", + "dirty": false + }, + "code_graph_delta": { + "summary": "no drift", + "changed": 0 + }, + "context_eval": { + "verdict": "pass", + "score": 0.9 + }, + "suspected_noop": false, + "control_transport": { + "kind": "unix", + "path": "/run/brigade/ctl.sock" + }, + "control_socket": "/run/brigade/ctl.sock", + "active_stage": 2, + "active_seats": [ + "coder", + "reviewer" + ], + "phase_owner": "chef", + "error": null, + "failure_phase": null, + "failure_kind": null, + "failure": { + "phase": "dispatch", + "kind": "worker-error", + "detail": "bounded detail", + "seat": "coder" + }, + "transport_warning": { + "kind": "retry", + "attempts": 1 + }, + "artifact_collection": { + "status": "ok" + }, + "artifacts": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4", + "handoff": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4/handoff.md", + "status": "handoff" +} +``` + +- [x] Confirm the fixture carries every preserved field and nothing outside the ownership map: + +```bash +PYTHONPATH=src python3 - <<'PY' +import json +from pathlib import Path +from brigade import run_projector + +base = json.loads(Path("tests/fixtures/run-lifecycle/golden-projection.base.json").read_text(encoding="utf-8")) +missing = run_projector.PRESERVED_FIELDS - set(base.keys()) +extra = set(base.keys()) - run_projector.OWNED_FIELDS +assert not missing, f"fixture missing preserved fields: {sorted(missing)}" +assert not extra, f"fixture carries unmapped fields: {sorted(extra)}" +print(f"base fixture covers all {len(run_projector.PRESERVED_FIELDS)} preserved fields") +PY +``` + +Expected output: `base fixture covers all 44 preserved fields`. + +- [x] Run the focused suite and confirm the expected intermediate state: + +```bash +python3 -m pytest tests/test_run_projector.py -q +``` + +Observed: 22 passed, 1 failed. The one failure was `Failed: missing golden expected fixture: .../golden-projection.expected.json` from `test_golden_replay_matches_expected_bytes`. + +## Task 4: generate and hand-review the golden expected fixture + +The convention from `golden-lifecycle.jsonl` applies: the implementation generates the expected bytes once, a human reviews them, and only then are they committed. + +- [x] Generate `tests/fixtures/run-lifecycle/golden-projection.expected.json`: + +```bash +PYTHONPATH=src python3 - <<'PY' +import json +from pathlib import Path +from brigade import run_journal, run_projector + +fixtures = Path("tests/fixtures/run-lifecycle") +base = json.loads((fixtures / "golden-projection.base.json").read_text(encoding="utf-8")) +report = run_journal.read_journal(fixtures / "golden-lifecycle.jsonl") +assert report.partial_tail is None and report.chain_errors == [] +projection = run_projector.project_run_snapshot(base, report.events, journal_present=True) +(fixtures / "golden-projection.expected.json").write_bytes(projection.to_bytes()) +print("wrote", fixtures / "golden-projection.expected.json") +PY +``` + +- [x] Hand-review the generated bytes against this checklist. The review is the point of the golden convention, so do not skip it: + +```bash +PYTHONPATH=src python3 - <<'PY' +import json +from pathlib import Path + +raw = Path("tests/fixtures/run-lifecycle/golden-projection.expected.json").read_bytes() +assert raw.endswith(b"\n") and not raw.endswith(b"\n\n"), "exactly one trailing newline" +data = json.loads(raw) +assert len(data) == 49, f"expected 49 owned keys, got {len(data)}" +assert list(data) == sorted(data), "keys must be sorted" +assert data["status"] == "ok" +assert data["projector_version"] == 1 +assert data["journal_present"] is True +assert data["journal_last_sequence"] == 6 +assert data["journal_last_event_digest"] == "176259cd00408b637f1c13496fcdfafea7dd96753fb8a9e76dbea58817442fc7" +base = json.loads(Path("tests/fixtures/run-lifecycle/golden-projection.base.json").read_text(encoding="utf-8")) +for key, value in base.items(): + if key != "status": + assert data[key] == value, f"preserved field drifted: {key}" +print("golden expected fixture reviewed ok") +PY +``` + +Also read the file top to bottom and confirm by eye: two-space indent, preserved nested values (`failure`, `code_graph_brief`, `control_transport`, `active_seats`) match the base verbatim, and no key outside the 49 owned fields appears. Expected output: `golden expected fixture reviewed ok`. + +- [x] Run the focused suite to green: + +```bash +python3 -m pytest tests/test_run_projector.py -q +``` + +Observed: `23 passed` (20 test functions, one parametrized over four cases). + +## Task 5: full verification + +- [x] Focused suite plus the neighboring journal and lifecycle suites: + +```bash +python3 -m pytest tests/test_run_projector.py tests/test_run_events.py tests/test_run_journal.py tests/test_run_lifecycle.py -q +``` + +Observed: `132 passed`. + +- [x] Full repository suite: + +```bash +python3 -m pytest -q +``` + +Observed through `./scripts/verify`: `4864 passed, 3 skipped`. + +- [x] Lint and type gates: + +```bash +python3 -m ruff check src/brigade/run_projector.py tests/test_run_projector.py +python3 -m mypy src/brigade/run_projector.py +``` + +Observed through `./scripts/verify`: ruff lint and format checks passed, and mypy reported `Success: no issues found in 331 source files`. + +- [x] Diff review: confirm the slice adds only its module, tests, fixtures, specification, and plan: + +```bash +git status --porcelain +``` + +Observed: exactly six new paths, the four implementation paths above plus this plan and `docs/phase-568-slice-3-run-projector.md`. No pre-existing path is modified relative to the slice branch point. + +## Task 6: commit + +- [x] Commit each completed task checkpoint with a conventional message. + +The slice is split into specification, contract-test, projector, and fixture commits. This is an in-house repo, so each implementation commit carries a `Co-authored-by` trailer only for a tool that did substantial work on that commit. + +## Task 7: Brigade verification + +Per the mandatory Brigade work loop and Definition of Done in `AGENTS.md`, +the final result must run the complete repository gate through Brigade: + +```bash +brigade work verify run --target . --command "./scripts/verify" --capture brigade-work +``` + +Expected: the receipt records ruff lint, ruff format check, version sync, +mypy, and the full pytest suite with its coverage floor. Atomic +`--capture brigade-work` records the outcome once. Do not run a second +`brigade outcome capture` for the same receipt. + +- [x] Receipt `20260729-024813-work-verify-0a6ae6` completed after the review fix: ruff lint and format passed, version `0.25.1` was synchronized across 13 locations, mypy passed 331 source files, and pytest finished with 4864 passed, 3 skipped, and 82.95% coverage. diff --git a/docs/phase-568-slice-3-run-projector.md b/docs/phase-568-slice-3-run-projector.md new file mode 100644 index 00000000..7d287ca9 --- /dev/null +++ b/docs/phase-568-slice-3-run-projector.md @@ -0,0 +1,406 @@ +# Issue #568 Slice 3: Run Snapshot Projector + +This is the proposed implementation spec for the approved slice 3 boundary +of GitHub issue #568. +Slice 1 landed the append-only journal kernel (`run_journal`, `run_events`). +Slice 2 landed opt-in lifecycle journaling of run status transitions +(`run_lifecycle`, wired into `aboyeur._write_json`). Slice 3 adds a pure +projector that derives a `run.json` snapshot from a base snapshot plus a +verified lifecycle event sequence. + +**This slice does not write `run.json` and does not change runtime behavior.** +It adds one pure module and its tests. No existing writer, reader, CLI +command, or recovery path calls the projector in this slice. Shadow +comparison, recovery integration, and live writer integration are later +slices, listed under Deferrals. + +Source contracts this spec builds on: + +- `src/brigade/run_events.py`: `brigade.run_event.v1` envelope, `EVENT_TYPES` + registry, `validate_event`, canonical byte rules, `MAX_DIAGNOSTIC_LEN`. +- `src/brigade/run_journal.py`: `RunEvent`, `JournalReport`, `read_journal`, + chain verification, typed bounded errors. +- `src/brigade/run_lifecycle.py`: `STATUS_EVENT_TYPE`, the slice-2 status to + event mapping, and the allowlisted payload construction in + `_allowlisted_payload`. +- `src/brigade/aboyeur.py`: the current `run.json` field contract + (`_run_payload` plus the merge-write recorders) and the snapshot encoding + in `_write_json` (`json.dumps(payload, indent=2, sort_keys=True) + "\n"`, + UTF-8, atomic replace). +- `src/brigade/run_resume.py`: writes the `resumed_at` and `recovery_history` + run.json fields during resume. +- `src/brigade/runguard.py`: writes the `recovery_preserved_artifact` + run.json field during guarded recovery. + +Standard library only. Brigade is zero-runtime-dependency. + +## Boundary + +`project_run_snapshot(base_snapshot, events, *, +journal_present=...)` is a pure function with an explicit ownership map over +every current `run.json` field: + +- It validates that `events` is a verified lifecycle sequence: every envelope + passes `run_events.validate_event`, sequences are contiguous from 1, + `previous_digest` links to the prior `event_digest`, and every event + carries the same `run_id`. Any deviation fails closed. The expected caller + obtains events from `run_journal.read_journal` after checking + `partial_tail is None` and `chain_errors == []`. The projector re-verifies + anyway as a second check. +- It derives exactly five fields: `status`, `projector_version`, + `journal_present`, `journal_last_sequence`, `journal_last_event_digest`. + Journal presence is an explicit input because an empty journal and a + missing journal both produce an empty event sequence. +- It preserves every other current `run.json` field verbatim from + `base_snapshot` through an explicit ownership map. +- It emits deterministic bytes using the existing sorted, indented + `run.json` encoding, not the compact journal-line canonical encoding. +- It rejects unknown or unmapped fields in `base_snapshot` and unmapped + event types in `events`. Nothing is skipped silently. + +The projector performs no I/O, reads no clock, reads no environment, and +holds no mutable module state. The same inputs always produce byte-identical +output. + +## API and types + +New module: `src/brigade/run_projector.py`. + +```python +PROJECTOR_VERSION: int = 1 + +# Field ownership over the run.json contract. Every current run.json key is +# in exactly one of these two sets. See the ownership inventory below. +DERIVED_FIELDS: frozenset[str] # {"status", "projector_version", "journal_present", + # "journal_last_sequence", "journal_last_event_digest"} +PRESERVED_FIELDS: frozenset[str] # the 44 remaining current run.json keys +OWNED_FIELDS: frozenset[str] # DERIVED_FIELDS | PRESERVED_FIELDS + +# Static event_type -> run.json status mappings (payload-independent rows of +# the event-to-status table below). +EVENT_STATUS: dict[str, str] + +class ProjectionError(RuntimeError): + """Base class for projection failures. Carries a bounded ``diagnostic``.""" + diagnostic: str + +class UnknownSnapshotFieldError(ProjectionError): ... +class UnmappedEventTypeError(ProjectionError): ... +class EventChainError(ProjectionError): ... +class EventPayloadError(ProjectionError): ... +class ProjectionInputError(ProjectionError): ... +class SnapshotEncodingError(ProjectionError): ... + +@dataclass(frozen=True) +class RunProjection: + """Result of a successful projection.""" + snapshot: dict[str, Any] # complete projected run.json object + status: str # derived status (equals snapshot["status"]) + journal_present: bool + last_sequence: int # 0 when events is empty + last_event_digest: str | None # None when events is empty + def to_bytes(self) -> bytes: ... + +def project_run_snapshot( + base_snapshot: Mapping[str, Any], + events: Sequence[run_journal.RunEvent | Mapping[str, Any]], + *, + journal_present: bool, +) -> RunProjection: ... + +def encode_snapshot_bytes(snapshot: Mapping[str, Any]) -> bytes: ... +``` + +Semantics: + +- `events` items may be typed `run_journal.RunEvent` instances or raw + envelope mappings. Every item is normalized to an envelope mapping and + passed through `run_events.validate_event`. Typed events use + `RunEvent.to_dict()`. Validation must produce an empty error list before + projection continues. This keeps direct typed inputs under the same + structural and digest checks as raw mappings. +- Chain verification inside the projector: the first event has `sequence == + 1` and `previous_digest is None`. Each subsequent event has `sequence == + prior + 1` and `previous_digest == prior.event_digest`. All events share + one `run_id`. Any break raises `EventChainError`. An empty sequence is + valid and means "no committed journal facts". +- `base_snapshot` keys must be a subset of `OWNED_FIELDS`. Any other key + raises `UnknownSnapshotFieldError`. Derived-field keys present in + `base_snapshot` are ignored and recomputed, so re-projecting a projected + snapshot is idempotent. +- `projector_version` is the constant `PROJECTOR_VERSION`, starting at 1. It + bumps whenever the ownership map, the event-to-status mapping, or the + encoding contract changes. +- `journal_present` must be a real boolean supplied by the caller from the + journal path check. It is not inferred from `events`: a created empty + journal projects `journal_present: True`, while a missing journal projects + `journal_present: False`. Either case has `journal_last_sequence: 0`, + `journal_last_event_digest: None`, and preserves `status` from + `base_snapshot`. A non-boolean value raises `ProjectionInputError`. +- `encode_snapshot_bytes` is the single encoding path: + `(json.dumps(snapshot, indent=2, sort_keys=True) + "\n").encode("utf-8")`, + matching `aboyeur._write_json` byte for byte. `RunProjection.to_bytes()` + delegates to it. JSON encoding failures are wrapped as + `SnapshotEncodingError` with a bounded diagnostic that does not include + snapshot values. + +## run.json field ownership inventory + +Derived (projector-owned, recomputed on every call): + +| Field | Derivation | +| --- | --- | +| `status` | Event-to-status mapping of the final event, or preserved from base when the sequence is empty | +| `projector_version` | Constant `PROJECTOR_VERSION` (1) | +| `journal_present` | Explicit boolean input describing journal path presence | +| `journal_last_sequence` | Final event `sequence`, or 0 | +| `journal_last_event_digest` | Final event `event_digest`, or null | + +Preserved (snapshot-owned and copied verbatim from `base_snapshot`, never +added, removed, retyped, or reformatted by the projector): + +Identity and schema: `schema` (`"brigade.run.v1"`), `schema_version` (1), +`task`, `orchestrator`, `roster`, `worker`, `scheduler`. + +Run configuration: `dry_run`, `read_only`, `cwd`, `lock_workspace`, +`codex_transport`, `lifecycle_journal_requested`. + +Timing: `started_at`, `status_started_at`, `finished_at`, +`duration_seconds`, `resumed_at`. + +Recovery (written by `run_resume.py` and `runguard.py`): +`recovery_history`, `recovery_preserved_artifact`. + +Briefs and routing telemetry: `code_graph_brief`, `drift_impact_brief`, +`evidence_brief`, `brief_budget`, `route`, `skill_route_policy`, +`pre_run_snapshot`, `git`, `code_graph_delta`, `context_eval`, +`suspected_noop`. + +Control transport: `control_transport`, `control_socket`. + +Live progress (written by `record_dispatch_stage` and +`record_result_processing`): `active_stage`, `active_seats`, `phase_owner`. + +Outcome and failure: `error`, `failure_phase`, `failure_kind`, `failure`, +`transport_warning`, `artifact_collection`. + +Artifact references: `artifacts`, `handoff`. + +That is 44 preserved fields plus 5 derived fields, 49 owned keys total. +This inventory mirrors `_run_payload` and the merge-write recorders in +`aboyeur.py` as of this spec, plus the recovery writers `run_resume.py` +(`resumed_at`, `recovery_history`) and `runguard.py` +(`recovery_preserved_artifact`). Any future writer that adds a `run.json` +field must extend the ownership map in the same change. The projector's +rejection of unmapped fields is the enforcement mechanism. + +## Event-to-status mapping + +Slice 2 (`run_lifecycle.STATUS_EVENT_TYPE`) journals these run.json status +transitions. The projector inverts that mapping. Where two run.json +statuses share one event type, the event payload `status` disambiguates, +because `_allowlisted_payload` always records the run.json status string in +the payload for these types. + +| Event type | Derived status | Payload rule | +| --- | --- | --- | +| `run.created` | `started` | `payload.status` must be `"started"` | +| `run.planning.started` | `planning` | none | +| `run.dispatch.requested` | `dispatching` | none | +| `run.dispatch.completed` | `result-processing` | none | +| `run.synthesis.started` | `synthesizing` | none | +| `run.synthesis.completed` | `handoff` | none | +| `run.completed` | `ok` | `payload.status` must be `"ok"` | +| `run.failed` | value of `payload.status` | `payload.status` must be `"failed"` or `"timeout"` | +| `run.interrupted` | `canceled` | `payload.status` must be `"canceled"` | + +Timeout vs failed: both run.json statuses journal as `run.failed`. The +projector reads `payload.status` and uses it directly as the derived status, +after checking it is `"failed"` or `"timeout"`. Any other value, or a +missing `status` key where the rule requires one, raises +`EventPayloadError`. The payload `detail` field is never read by the +projector. + +Registered event types with no current status mapping +(`run.planning.completed`, `run.planning.failed`, `run.dispatch.observed`, +`run.dispatch.failed`, `run.synthesis.failed`, `run.paused`, `run.resumed`, +`approval.requested`, `approval.granted`, `approval.rejected`, +`approval.held`, `approval.consumed`, `run.recovery.started`, +`run.recovery.completed`) are never emitted by the slice-2 writer. If one +appears in the sequence, projection stops with `UnmappedEventTypeError`. +They are never skipped silently. Mapping them is deferred to the work listed +under Deferrals. + +Status lag note: the run.json statuses `dry-run`, `incomplete`, and +`artifact-collection` have no event mapping in slice 2, so the journal +never records those transitions. When a run ends in one of those statuses, +the projected status (from the final journaled transition) legitimately +lags the live writer's `run.json` status. This is by design in slice 3. +Detecting and reconciling the gap belongs to shadow comparison (slice 4), +and journaling those statuses belongs to event enrichment. Both are +deferred. + +## Invariants + +1. Purity: no I/O, no clock, no environment, no randomness. Output depends + only on `(base_snapshot, events, journal_present)`. +2. Determinism: identical inputs produce byte-identical `to_bytes()` output + on every call and every host. +3. Closed ownership: output keys are a subset of `OWNED_FIELDS`. A base + snapshot key outside `OWNED_FIELDS` stops projection. It is never + dropped or passed through. +4. Preservation: for every preserved key present in `base_snapshot`, the + output value deep-equals the base value. The projector adds no preserved + key that base lacks and removes none that base has. +5. Derivation completeness: all five derived fields are always present in + the output and always recomputed, even when present in the base. +6. Fail-closed chain: any sequence gap, duplicate, digest-link break, + run_id mix, invalid envelope, unmapped event type, or payload rule + violation raises a typed error. There is no partial projection. +7. Bounded errors: every failure is a `ProjectionError` subclass whose + `diagnostic` is at most `run_events.MAX_DIAGNOSTIC_LEN` (240) + characters, truncated with the same ellipsis convention as the journal + layer. +8. Encoding parity: `to_bytes()` equals the `aboyeur._write_json` encoding + of the same object: sorted keys, two-space indent, trailing newline, + UTF-8. + +## Bounded errors + +| Error | Raised when | Diagnostic content | +| --- | --- | --- | +| `UnknownSnapshotFieldError` | base snapshot carries a key outside `OWNED_FIELDS` | category plus up to 8 sorted offending key names | +| `UnmappedEventTypeError` | a validated event's type has no status mapping | category plus the event type and its sequence | +| `EventChainError` | envelope invalid, sequence gap or duplicate, digest-link break, mixed run_id | category plus the failing sequence, never raw envelope bytes | +| `EventPayloadError` | a payload status rule fails | category plus event type and the expected value set, never the raw payload value | +| `ProjectionInputError` | `journal_present` is not a real boolean | category only | +| `SnapshotEncodingError` | projected values cannot be encoded by the existing `run.json` JSON contract | category only, never the raw value or serializer message | + +Diagnostics carry categories and field names only. Raw payload values, +paths, and snapshot contents never appear in a diagnostic, matching the +slice-2 bounded-failure rule. + +## Golden fixture and unit tests + +Fixtures (under `tests/fixtures/run-lifecycle/`, alongside the existing +`golden-lifecycle.jsonl`): + +- `golden-projection.base.json`: a base snapshot exercising the ownership + map, including nested objects (`failure`, `code_graph_brief`, + `control_transport`), lists (`active_seats`, `resumed_at`, + `recovery_history`), and optional fields (`scheduler`, `handoff`, + `artifact_collection`, `recovery_preserved_artifact`). +- `golden-projection.expected.json`: the exact expected projected bytes for + that base plus the full `golden-lifecycle.jsonl` sequence. + +Tests in `tests/test_run_projector.py`: + +1. Golden replay: read `golden-lifecycle.jsonl` via + `run_journal.read_journal`, assert no chain errors and no partial tail, + project against the golden base, assert `to_bytes()` equals the golden + expected bytes exactly. +2. Golden determinism: a second projection of the same inputs is + byte-identical to the first. +3. Empty sequence with no journal: `journal_present` is False, last sequence + 0, digest None, status and all preserved fields match base. +4. Empty created journal: the same empty sequence with + `journal_present=True` projects presence as True without inventing a last + sequence or digest. +5. Full golden sequence derives `status == "ok"`, + `journal_last_sequence == 6`, and the digest of the final golden event. +6. `run.failed` with `payload.status == "timeout"` derives `timeout`. +7. `run.failed` with `payload.status == "failed"` derives `failed`. +8. `run.failed` with any other `payload.status`, or a missing `status`, + raises `EventPayloadError` with a diagnostic of at most 240 chars. +9. `run.interrupted` derives `canceled`. A wrong payload status raises. +10. Unknown base snapshot key raises `UnknownSnapshotFieldError`. The key + name appears in the bounded diagnostic. +11. Registered-but-unmapped event type (for example `approval.requested`) + in the sequence raises `UnmappedEventTypeError`. +12. Sequence gap, duplicate sequence, broken `previous_digest` link, and + mixed run_id each raise `EventChainError` and produce no output. +13. Invalid envelope mapping (for example an unknown event type, which + fails `run_events.validate_event`) raises `EventChainError`. +14. Invalid-envelope diagnostics do not include rejected envelope values. +15. A mutated typed `RunEvent` replacement whose envelope digest or ID does + not validate raises `EventChainError`. +16. Preservation: every preserved field in a full-coverage base deep-equals + the output, including nested objects, and no preserved key appears or + disappears. +17. Re-projection idempotence: feeding a projected snapshot back as the + base with the same events yields byte-identical output. +18. Encoding parity: `to_bytes()` equals + `json.dumps(snapshot, indent=2, sort_keys=True) + "\n"` encoded UTF-8. +19. A non-JSON or circular preserved value raises `SnapshotEncodingError` + with a bounded category-only diagnostic. +20. A non-boolean `journal_present` value raises `ProjectionInputError`. + +The golden expected file is generated once by the implementation and +reviewed by hand before commit, the same convention as +`golden-lifecycle.jsonl`. + +## Implementation steps + +1. Add `src/brigade/run_projector.py` with the constants, ownership sets, + error classes, `RunProjection`, `project_run_snapshot`, and + `encode_snapshot_bytes` as specified above. Standard library only. +2. Implement event normalization: convert typed `RunEvent` instances with + `to_dict()`, validate every normalized mapping through + `run_events.validate_event`, then verify the chain (contiguous sequence + from 1, digest linkage, single run_id). +3. Implement the ownership check and the preservation copy (deep copy of + preserved values so the result shares no mutable state with the caller). +4. Implement status derivation per the event-to-status table, including the + payload status rules and the empty-sequence preservation rule. +5. Implement `encode_snapshot_bytes` and `RunProjection.to_bytes()`. +6. Add the two golden fixtures under `tests/fixtures/run-lifecycle/`. +7. Add `tests/test_run_projector.py` with the 20 tests above. +8. No changes to `aboyeur.py`, `run_lifecycle.py`, `run_journal.py`, + `run_events.py`, any CLI module, or any writer path. + +## Compatibility + +- Runtime behavior is unchanged. No code path invokes the projector in this + slice, so `run.json` bytes, journal bytes, CLI output, and reader behavior + (`runs watch`, `show`, `steer`, `interrupt`, `recover`, `resume`) are all + exactly as before. +- The field names `projector_version`, `journal_present`, + `journal_last_sequence`, and `journal_last_event_digest` are reserved by + this spec for slice 4 and later. Nothing writes them yet. When a later + slice starts writing them, existing readers see additive keys in a + `brigade.run.v1` object, which the current reader set tolerates. +- `PROJECTOR_VERSION` starts at 1 and bumps on any change to the ownership + map, the event-to-status mapping, or the encoding contract. +- Legacy run directories without `events/lifecycle.jsonl` are unaffected. + The projector is never called for them at runtime. + +## Non-goals + +- Writing `run.json` or any other file from the projector. +- Comparing projected output against the live writer's output. +- Rebuilding `run.json` from the journal during recovery. +- Journaling or projecting the unmapped statuses (`dry-run`, `incomplete`, + `artifact-collection`). +- Projecting approval, pause, resume, or recovery events. +- Journal compaction, redaction, retention, or quarantine flows. +- Any CLI surface. + +## Deferrals + +- Shadow comparison: running the projector alongside the live writer after + each transition, recording mismatches, and gating automatic projection + replacement is slice 4 (issue #568 step 4). +- Recovery: `brigade runs recover` journal verification and projection + repair is slice 5 (issue #568 step 5). This slice's fail-closed validation + is the contract recovery will reuse. +- Live writer integration: routing `aboyeur._write_json` through the + projector, or making the journal authoritative for new runs, waits for + shadow parity (issue #568 step 6). +- Journal-only reconstruction: projecting without a base snapshot (a + deleted or corrupt `run.json`) requires deriving preserved fields from + events, which the current event payloads do not carry. It needs event + enrichment first and is out of scope here. +- Event enrichment: new payload keys or event types that would let the + journal represent unmapped statuses, dispatch observation detail, or + approval pause and resume projection are future slices under the issue's + event contract, not this one. diff --git a/src/brigade/run_projector.py b/src/brigade/run_projector.py new file mode 100644 index 00000000..fc340982 --- /dev/null +++ b/src/brigade/run_projector.py @@ -0,0 +1,334 @@ +"""Pure run.json snapshot projector (issue #568, slice 3). + +Derives a deterministic ``run.json`` snapshot from a base snapshot plus a +verified lifecycle event sequence. The projector owns exactly five derived +fields (``status``, ``projector_version``, ``journal_present``, +``journal_last_sequence``, ``journal_last_event_digest``) and preserves every +other current ``run.json`` field verbatim from the base snapshot through an +explicit ownership map. Unknown base fields and registered event types with +no status mapping fail closed with bounded typed errors. + +Purity: no I/O, no clock, no environment, no mutable module state. Identical +inputs produce byte-identical output on every call and every host. Encoding +matches ``aboyeur._write_json`` byte for byte: sorted keys, two-space indent, +one trailing newline, UTF-8. + +This slice does not write ``run.json`` and is not wired into any runtime +path. Standard library only. Brigade is zero-runtime-dependency. +""" + +from __future__ import annotations + +import copy +import json +from dataclasses import dataclass +from typing import Any, Mapping, Sequence + +from brigade import run_events, run_journal + +PROJECTOR_VERSION: int = 1 + +# Field ownership over the run.json contract. Every current run.json key is +# in exactly one of these two sets; see the ownership inventory in +# docs/phase-568-slice-3-run-projector.md. +DERIVED_FIELDS: frozenset[str] = frozenset( + { + "status", + "projector_version", + "journal_present", + "journal_last_sequence", + "journal_last_event_digest", + } +) + +PRESERVED_FIELDS: frozenset[str] = frozenset( + { + # Identity and schema + "schema", + "schema_version", + "task", + "orchestrator", + "roster", + "worker", + "scheduler", + # Run configuration + "dry_run", + "read_only", + "cwd", + "lock_workspace", + "codex_transport", + "lifecycle_journal_requested", + # Timing + "started_at", + "status_started_at", + "finished_at", + "duration_seconds", + "resumed_at", + # Recovery (written by run_resume.py and runguard.py) + "recovery_history", + "recovery_preserved_artifact", + # Briefs and routing telemetry + "code_graph_brief", + "drift_impact_brief", + "evidence_brief", + "brief_budget", + "route", + "skill_route_policy", + "pre_run_snapshot", + "git", + "code_graph_delta", + "context_eval", + "suspected_noop", + # Control transport + "control_transport", + "control_socket", + # Live progress + "active_stage", + "active_seats", + "phase_owner", + # Outcome and failure + "error", + "failure_phase", + "failure_kind", + "failure", + "transport_warning", + "artifact_collection", + # Artifact references + "artifacts", + "handoff", + } +) + +OWNED_FIELDS: frozenset[str] = DERIVED_FIELDS | PRESERVED_FIELDS + +# Static event_type -> run.json status mappings (payload-independent rows of +# the event-to-status table). +EVENT_STATUS: dict[str, str] = { + "run.planning.started": "planning", + "run.dispatch.requested": "dispatching", + "run.dispatch.completed": "result-processing", + "run.synthesis.started": "synthesizing", + "run.synthesis.completed": "handoff", +} + +# Payload-driven rows: event_type -> (allowed payload status values, derived +# status). A None derived status means the payload status value is used +# directly as the derived status (run.failed: "failed" vs "timeout"). +_PAYLOAD_STATUS_RULES: dict[str, tuple[frozenset[str], str | None]] = { + "run.created": (frozenset({"started"}), "started"), + "run.completed": (frozenset({"ok"}), "ok"), + "run.failed": (frozenset({"failed", "timeout"}), None), + "run.interrupted": (frozenset({"canceled"}), "canceled"), +} + + +class ProjectionError(RuntimeError): + """Base class for projection failures. Carries a bounded ``diagnostic``.""" + + def __init__(self, diagnostic: str) -> None: + super().__init__(diagnostic) + self.diagnostic = diagnostic + + +class UnknownSnapshotFieldError(ProjectionError): + """Base snapshot carries a key outside ``OWNED_FIELDS``.""" + + +class UnmappedEventTypeError(ProjectionError): + """A validated event's type has no status mapping.""" + + +class EventChainError(ProjectionError): + """Envelope invalid, sequence gap/duplicate, digest-link break, mixed run_id.""" + + +class EventPayloadError(ProjectionError): + """A payload status rule failed for a payload-driven event type.""" + + +class ProjectionInputError(ProjectionError): + """An input argument violates the projector contract.""" + + +class SnapshotEncodingError(ProjectionError): + """Projected values cannot be encoded by the run.json JSON contract.""" + + +def _bound(msg: str) -> str: + limit = run_events.MAX_DIAGNOSTIC_LEN + if len(msg) <= limit: + return msg + return msg[: limit - 1] + "…" + + +@dataclass(frozen=True) +class RunProjection: + """Result of a successful projection.""" + + snapshot: dict[str, Any] # complete projected run.json object + status: str # derived status (equals snapshot["status"]) + journal_present: bool + last_sequence: int # 0 when events is empty + last_event_digest: str | None # None when events is empty + + def to_bytes(self) -> bytes: + """Encode the snapshot with the existing run.json byte contract.""" + return encode_snapshot_bytes(self.snapshot) + + +def encode_snapshot_bytes(snapshot: Mapping[str, Any]) -> bytes: + """Encode a snapshot as ``json.dumps(indent=2, sort_keys=True) + newline``, UTF-8. + + This matches ``aboyeur._write_json`` byte for byte. Encoding failures are + wrapped as a bounded ``SnapshotEncodingError`` whose diagnostic carries a + category only -- never the raw value or the serializer message. + """ + try: + return (json.dumps(snapshot, indent=2, sort_keys=True) + "\n").encode("utf-8") + except (TypeError, ValueError, OverflowError, RecursionError) as exc: + raise SnapshotEncodingError(_bound("snapshot cannot be encoded as run.json JSON bytes")) from exc + + +def _normalize_events(events: Sequence[run_journal.RunEvent | Mapping[str, Any]]) -> list[Mapping[str, Any]]: + """Normalize typed RunEvent instances and raw mappings to envelope mappings.""" + normalized: list[Mapping[str, Any]] = [] + for item in events: + if isinstance(item, run_journal.RunEvent): + normalized.append(item.to_dict()) + elif isinstance(item, Mapping): + normalized.append(item) + else: + raise EventChainError(_bound("event chain item is neither a RunEvent nor a mapping")) + return normalized + + +def _verify_chain(envelopes: list[Mapping[str, Any]]) -> None: + """Fail closed on any invalid envelope, sequence break, or run_id mix. + + Every envelope must pass ``run_events.validate_event``; the first event + must have ``sequence == 1`` and a null ``previous_digest``; each later + event must continue the sequence and link its ``previous_digest`` to the + prior ``event_digest``; all events share one ``run_id``. + """ + run_id: Any = None + expected_sequence = 1 + previous_digest: Any = None + for position, env in enumerate(envelopes, start=1): + errors = run_events.validate_event(env) + if errors: + raise EventChainError(_bound(f"event chain envelope invalid at input position {position}")) + sequence = env["sequence"] + if run_id is None: + run_id = env["run_id"] + elif env["run_id"] != run_id: + raise EventChainError(_bound(f"event chain mixes run_id values at sequence {sequence}")) + if sequence != expected_sequence: + raise EventChainError(_bound(f"event chain sequence break: expected {expected_sequence}, got {sequence}")) + if sequence == 1: + if env["previous_digest"] is not None: + raise EventChainError(_bound("event chain sequence 1 previous_digest must be null")) + elif env["previous_digest"] != previous_digest: + raise EventChainError( + _bound(f"event chain previous_digest does not link to prior event_digest at sequence {sequence}") + ) + previous_digest = env["event_digest"] + expected_sequence = sequence + 1 + + +def _derive_status(envelopes: list[Mapping[str, Any]], base_status: Any) -> str: + """Derive the run.json status from the event sequence. + + Every event must have a status mapping: static rows come from + ``EVENT_STATUS``; payload-driven rows validate the payload ``status`` + against the allowed set. Registered types with no mapping raise + ``UnmappedEventTypeError``; a missing or disallowed payload status raises + ``EventPayloadError``. An empty sequence preserves the base status. + """ + if not envelopes: + if not isinstance(base_status, str): + raise ProjectionInputError(_bound("base snapshot status must be a string when the event sequence is empty")) + return base_status + status = "" + for env in envelopes: + event_type = env["event_type"] + sequence = env["sequence"] + if event_type in EVENT_STATUS: + status = EVENT_STATUS[event_type] + continue + rule = _PAYLOAD_STATUS_RULES.get(event_type) + if rule is None: + raise UnmappedEventTypeError( + _bound(f"event type {event_type!r} at sequence {sequence} has no status mapping") + ) + allowed, derived = rule + payload = env["payload"] + payload_status = payload.get("status") + if payload_status not in allowed: + raise EventPayloadError( + _bound(f"event type {event_type!r} at sequence {sequence} requires payload status in {sorted(allowed)}") + ) + status = derived if derived is not None else payload_status + return status + + +def project_run_snapshot( + base_snapshot: Mapping[str, Any], + events: Sequence[run_journal.RunEvent | Mapping[str, Any]], + *, + journal_present: bool, +) -> RunProjection: + """Project a run.json snapshot from a base snapshot plus a verified event sequence. + + Pure: no I/O, no clock, no environment. ``journal_present`` must be a real + boolean supplied by the caller from the journal path check; it is never + inferred from ``events``. Base snapshot keys must be a subset of + ``OWNED_FIELDS``; derived-field keys present in the base are ignored and + recomputed, so re-projecting a projected snapshot is idempotent. Every + preserved field present in the base is deep-copied into the result. The + result is verified to encode under the run.json byte contract before it + is returned. + """ + if not isinstance(journal_present, bool): + raise ProjectionInputError(_bound("journal_present must be a boolean")) + if not isinstance(base_snapshot, Mapping): + raise ProjectionInputError(_bound("base_snapshot must be a mapping")) + + unknown = sorted(set(base_snapshot.keys()) - OWNED_FIELDS) + if unknown: + shown = ", ".join(unknown[:8]) + raise UnknownSnapshotFieldError(_bound(f"base snapshot carries unknown fields: {shown}")) + + envelopes = _normalize_events(events) + _verify_chain(envelopes) + + status = _derive_status(envelopes, base_snapshot.get("status")) + + snapshot: dict[str, Any] = {} + for field_name in PRESERVED_FIELDS: + if field_name in base_snapshot: + snapshot[field_name] = copy.deepcopy(base_snapshot[field_name]) + + last_sequence = 0 + last_event_digest: str | None = None + if envelopes: + final = envelopes[-1] + last_sequence = final["sequence"] + last_event_digest = final["event_digest"] + + snapshot["status"] = status + snapshot["projector_version"] = PROJECTOR_VERSION + snapshot["journal_present"] = journal_present + snapshot["journal_last_sequence"] = last_sequence + snapshot["journal_last_event_digest"] = last_event_digest + + # Verify the result encodes under the run.json byte contract before + # returning; failures surface as a bounded SnapshotEncodingError. + encode_snapshot_bytes(snapshot) + + return RunProjection( + snapshot=snapshot, + status=status, + journal_present=journal_present, + last_sequence=last_sequence, + last_event_digest=last_event_digest, + ) diff --git a/tests/fixtures/run-lifecycle/golden-projection.base.json b/tests/fixtures/run-lifecycle/golden-projection.base.json new file mode 100644 index 00000000..d9846d87 --- /dev/null +++ b/tests/fixtures/run-lifecycle/golden-projection.base.json @@ -0,0 +1,118 @@ +{ + "schema": "brigade.run.v1", + "schema_version": 1, + "task": "Project the lifecycle journal", + "orchestrator": "chef", + "roster": { + "path": ".brigade/roster.toml", + "source": "project", + "shadowed": [] + }, + "worker": "coder", + "scheduler": { + "requested": "auto", + "used": "serial" + }, + "dry_run": false, + "read_only": false, + "cwd": "/work/repo", + "lock_workspace": "/work/repo", + "codex_transport": "app-server", + "lifecycle_journal_requested": true, + "started_at": "2026-07-27T15:30:45.123456+00:00", + "status_started_at": "2026-07-27T15:30:49.000000+00:00", + "finished_at": "2026-07-27T15:30:50.000000+00:00", + "duration_seconds": 4.877, + "resumed_at": [ + "2026-07-27T15:31:00.000000+00:00", + "2026-07-27T15:32:00.000000+00:00" + ], + "recovery_history": [ + { + "kind": "worker-error", + "at": "2026-07-27T15:31:30.000000+00:00", + "seat": "coder" + } + ], + "recovery_preserved_artifact": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4/run.json.corrupt", + "code_graph_brief": { + "attached": true, + "bytes": 2048 + }, + "drift_impact_brief": { + "attached": false, + "bytes": 0, + "pending_count": 0 + }, + "evidence_brief": { + "attached": true, + "bytes": 512 + }, + "brief_budget": { + "bytes": 8192, + "attached": [ + "code-graph" + ] + }, + "route": { + "skill": "brigade-work", + "matched": [ + "run" + ], + "score": 3 + }, + "skill_route_policy": { + "policy_applied": true, + "extensions": [ + "force-verify" + ] + }, + "pre_run_snapshot": { + "head": "abc123", + "dirty": false + }, + "git": { + "head": "abc123", + "branch": "main", + "dirty": false + }, + "code_graph_delta": { + "summary": "no drift", + "changed": 0 + }, + "context_eval": { + "verdict": "pass", + "score": 0.9 + }, + "suspected_noop": false, + "control_transport": { + "kind": "unix", + "path": "/run/brigade/ctl.sock" + }, + "control_socket": "/run/brigade/ctl.sock", + "active_stage": 2, + "active_seats": [ + "coder", + "reviewer" + ], + "phase_owner": "chef", + "error": null, + "failure_phase": null, + "failure_kind": null, + "failure": { + "phase": "dispatch", + "kind": "worker-error", + "detail": "bounded detail", + "seat": "coder" + }, + "transport_warning": { + "kind": "retry", + "attempts": 1 + }, + "artifact_collection": { + "status": "ok" + }, + "artifacts": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4", + "handoff": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4/handoff.md", + "status": "handoff" +} diff --git a/tests/fixtures/run-lifecycle/golden-projection.expected.json b/tests/fixtures/run-lifecycle/golden-projection.expected.json new file mode 100644 index 00000000..06630faf --- /dev/null +++ b/tests/fixtures/run-lifecycle/golden-projection.expected.json @@ -0,0 +1,122 @@ +{ + "active_seats": [ + "coder", + "reviewer" + ], + "active_stage": 2, + "artifact_collection": { + "status": "ok" + }, + "artifacts": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4", + "brief_budget": { + "attached": [ + "code-graph" + ], + "bytes": 8192 + }, + "code_graph_brief": { + "attached": true, + "bytes": 2048 + }, + "code_graph_delta": { + "changed": 0, + "summary": "no drift" + }, + "codex_transport": "app-server", + "context_eval": { + "score": 0.9, + "verdict": "pass" + }, + "control_socket": "/run/brigade/ctl.sock", + "control_transport": { + "kind": "unix", + "path": "/run/brigade/ctl.sock" + }, + "cwd": "/work/repo", + "drift_impact_brief": { + "attached": false, + "bytes": 0, + "pending_count": 0 + }, + "dry_run": false, + "duration_seconds": 4.877, + "error": null, + "evidence_brief": { + "attached": true, + "bytes": 512 + }, + "failure": { + "detail": "bounded detail", + "kind": "worker-error", + "phase": "dispatch", + "seat": "coder" + }, + "failure_kind": null, + "failure_phase": null, + "finished_at": "2026-07-27T15:30:50.000000+00:00", + "git": { + "branch": "main", + "dirty": false, + "head": "abc123" + }, + "handoff": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4/handoff.md", + "journal_last_event_digest": "176259cd00408b637f1c13496fcdfafea7dd96753fb8a9e76dbea58817442fc7", + "journal_last_sequence": 6, + "journal_present": true, + "lifecycle_journal_requested": true, + "lock_workspace": "/work/repo", + "orchestrator": "chef", + "phase_owner": "chef", + "pre_run_snapshot": { + "dirty": false, + "head": "abc123" + }, + "projector_version": 1, + "read_only": false, + "recovery_history": [ + { + "at": "2026-07-27T15:31:30.000000+00:00", + "kind": "worker-error", + "seat": "coder" + } + ], + "recovery_preserved_artifact": "/work/repo/.brigade/runs/20260727-153045-a1b2c3d4/run.json.corrupt", + "resumed_at": [ + "2026-07-27T15:31:00.000000+00:00", + "2026-07-27T15:32:00.000000+00:00" + ], + "roster": { + "path": ".brigade/roster.toml", + "shadowed": [], + "source": "project" + }, + "route": { + "matched": [ + "run" + ], + "score": 3, + "skill": "brigade-work" + }, + "scheduler": { + "requested": "auto", + "used": "serial" + }, + "schema": "brigade.run.v1", + "schema_version": 1, + "skill_route_policy": { + "extensions": [ + "force-verify" + ], + "policy_applied": true + }, + "started_at": "2026-07-27T15:30:45.123456+00:00", + "status": "ok", + "status_started_at": "2026-07-27T15:30:49.000000+00:00", + "suspected_noop": false, + "task": "Project the lifecycle journal", + "transport_warning": { + "attempts": 1, + "kind": "retry" + }, + "worker": "coder" +} diff --git a/tests/test_run_projector.py b/tests/test_run_projector.py new file mode 100644 index 00000000..fe1bfdf3 --- /dev/null +++ b/tests/test_run_projector.py @@ -0,0 +1,454 @@ +"""Tests for brigade.run_projector pure snapshot projection.""" + +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from brigade import run_events, run_journal +from brigade.run_projector import ( + DERIVED_FIELDS, + EVENT_STATUS, + OWNED_FIELDS, + PRESERVED_FIELDS, + PROJECTOR_VERSION, + EventChainError, + EventPayloadError, + ProjectionError, + ProjectionInputError, + RunProjection, + SnapshotEncodingError, + UnknownSnapshotFieldError, + UnmappedEventTypeError, + encode_snapshot_bytes, + project_run_snapshot, +) + +FIXTURES = Path(__file__).resolve().parent / "fixtures" / "run-lifecycle" +GOLDEN_LIFECYCLE_PATH = FIXTURES / "golden-lifecycle.jsonl" +GOLDEN_BASE_PATH = FIXTURES / "golden-projection.base.json" +GOLDEN_EXPECTED_PATH = FIXTURES / "golden-projection.expected.json" + +RUN_ID = "20260727-153045-a1b2c3d4" +OTHER_RUN_ID = "20260727-153045-z9y8x7w6" +RECORDED_AT = "2026-07-27T15:30:45.123456Z" +GOLDEN_FINAL_DIGEST = "176259cd00408b637f1c13496fcdfafea7dd96753fb8a9e76dbea58817442fc7" + + +def _golden_events() -> list[run_journal.RunEvent]: + """Load the verified golden lifecycle event sequence from the fixture.""" + report = run_journal.read_journal(GOLDEN_LIFECYCLE_PATH) + assert report.partial_tail is None + assert report.chain_errors == [] + return report.events + + +def _build_event( + sequence: int, + event_type: str, + payload: dict, + idempotency_key: str, + recorded_at: str, + previous_digest: str | None, + *, + run_id: str = RUN_ID, +) -> dict: + """Build a validated run_event.v1 envelope using the real run_events contract.""" + return run_events.build_event( + run_id=run_id, + sequence=sequence, + event_type=event_type, + payload=payload, + idempotency_key=idempotency_key, + recorded_at=recorded_at, + previous_digest=previous_digest, + ) + + +def _minimal_base_snapshot(status: str = "started") -> dict: + """Return a minimal base snapshot that satisfies the ownership map.""" + return { + "schema": "brigade.run.v1", + "schema_version": 1, + "status": status, + "task": "test-task", + "orchestrator": "test-orchestrator", + "roster": "test-roster", + "worker": "test-worker", + "scheduler": "immediate", + } + + +def _full_base_snapshot() -> dict: + """Return a base snapshot covering every preserved run.json field.""" + base = { + "schema": "brigade.run.v1", + "schema_version": 1, + "status": "started", + "task": "task-id", + "orchestrator": "orchestrator-id", + "roster": "roster-id", + "worker": "worker-id", + "scheduler": "immediate", + "dry_run": False, + "read_only": False, + "cwd": "/tmp", + "lock_workspace": True, + "codex_transport": "app-server", + "lifecycle_journal_requested": True, + "started_at": "2026-07-27T15:30:00.000000Z", + "status_started_at": "2026-07-27T15:30:00.000000Z", + "finished_at": None, + "duration_seconds": 0, + "resumed_at": ["2026-07-27T15:31:00.000000Z"], + "recovery_history": [{"kind": "worker-error"}], + "recovery_preserved_artifact": "/tmp/run.json.corrupt", + "code_graph_brief": {"entries": []}, + "drift_impact_brief": {"entries": []}, + "evidence_brief": {"entries": []}, + "brief_budget": 100, + "route": "route-id", + "skill_route_policy": "auto", + "pre_run_snapshot": {"files": []}, + "git": {"branch": "main"}, + "code_graph_delta": {"changed": []}, + "context_eval": {"score": 0}, + "suspected_noop": False, + "control_transport": "unix", + "control_socket": "/tmp/control.sock", + "active_stage": "planning", + "active_seats": ["coder"], + "phase_owner": "coder", + "error": None, + "failure_phase": None, + "failure_kind": None, + "failure": {"detail": None}, + "transport_warning": None, + "artifact_collection": None, + "artifacts": [], + "handoff": None, + } + for field in PRESERVED_FIELDS: + if field not in base: + base[field] = None + return base + + +def _chain_events(scenario: str) -> list[dict]: + """Build event sequences that fail chain verification for named reasons.""" + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + if scenario == "gap": + planning = _build_event( + 3, + "run.planning.started", + {"detail": "planning"}, + "plan-3", + "2026-07-27T15:30:46.000000Z", + created["event_digest"], + ) + return [created, planning] + if scenario == "duplicate": + duplicate = _build_event( + 1, + "run.created", + {"status": "started"}, + "create-dup", + "2026-07-27T15:30:46.000000Z", + None, + ) + return [created, duplicate] + if scenario == "broken_digest": + planning = _build_event( + 2, + "run.planning.started", + {"detail": "planning"}, + "plan-2", + "2026-07-27T15:30:46.000000Z", + "0" * 64, + ) + return [created, planning] + if scenario == "mixed_run_id": + other_created = _build_event( + 1, + "run.created", + {"status": "started"}, + "create-other", + RECORDED_AT, + None, + run_id=OTHER_RUN_ID, + ) + planning = _build_event( + 2, + "run.planning.started", + {"detail": "planning"}, + "plan-2", + "2026-07-27T15:30:46.000000Z", + other_created["event_digest"], + ) + return [other_created, planning] + raise ValueError(f"unknown scenario: {scenario}") + + +def _events_ending_with_failed(*, status: str | None) -> list[dict]: + """Build a two-event sequence ending with run.failed with the given status.""" + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + payload = {"detail": "failed"} + if status is not None: + payload["status"] = status + failed = _build_event( + 2, + "run.failed", + payload, + "failed-1", + "2026-07-27T15:30:46.000000Z", + created["event_digest"], + ) + return [created, failed] + + +def _events_ending_with_interrupted(*, status: str) -> list[dict]: + """Build a two-event sequence ending with run.interrupted with the given status.""" + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + interrupted = _build_event( + 2, + "run.interrupted", + {"status": status}, + "interrupt-1", + "2026-07-27T15:30:46.000000Z", + created["event_digest"], + ) + return [created, interrupted] + + +def _assert_bounded_projection_error(excinfo: pytest.ExceptionInfo[ProjectionError]) -> None: + assert len(excinfo.value.diagnostic) <= run_events.MAX_DIAGNOSTIC_LEN + + +def test_golden_replay_matches_expected_bytes(): + if not GOLDEN_BASE_PATH.is_file(): + pytest.fail(f"missing golden base fixture: {GOLDEN_BASE_PATH}") + if not GOLDEN_EXPECTED_PATH.is_file(): + pytest.fail(f"missing golden expected fixture: {GOLDEN_EXPECTED_PATH}") + base = json.loads(GOLDEN_BASE_PATH.read_text()) + expected = GOLDEN_EXPECTED_PATH.read_bytes() + projection = project_run_snapshot(base, _golden_events(), journal_present=True) + assert projection.to_bytes() == expected + + +def test_repeated_projection_is_byte_deterministic(): + base = _minimal_base_snapshot() + events = _golden_events() + first = project_run_snapshot(base, events, journal_present=True) + second = project_run_snapshot(base, events, journal_present=True) + assert first.to_bytes() == second.to_bytes() + + +def test_empty_events_no_journal_preserves_base_status_and_deep_copies(): + base = _minimal_base_snapshot(status="planning") + base["failure"] = {"detail": "pending"} + projection = project_run_snapshot(base, [], journal_present=False) + assert isinstance(projection, RunProjection) + assert projection.status == "planning" + assert projection.snapshot["status"] == "planning" + assert projection.snapshot["failure"] == base["failure"] + assert projection.snapshot["failure"] is not base["failure"] + assert projection.last_sequence == 0 + assert projection.last_event_digest is None + assert projection.journal_present is False + + +def test_empty_created_journal_reports_presence_with_zero_sequence_and_null_digest(): + base = _minimal_base_snapshot(status="dispatching") + projection = project_run_snapshot(base, [], journal_present=True) + assert projection.status == "dispatching" + assert projection.journal_present is True + assert projection.last_sequence == 0 + assert projection.last_event_digest is None + assert projection.snapshot["journal_last_sequence"] == 0 + assert projection.snapshot["journal_last_event_digest"] is None + + +def test_golden_sequence_derives_ok_sequence_and_digest(): + projection = project_run_snapshot(_minimal_base_snapshot(), _golden_events(), journal_present=True) + assert projection.status == "ok" + assert projection.last_sequence == 6 + assert projection.last_event_digest == GOLDEN_FINAL_DIGEST + assert projection.snapshot["status"] == "ok" + assert projection.snapshot["projector_version"] == PROJECTOR_VERSION + assert projection.snapshot["journal_last_sequence"] == 6 + assert projection.snapshot["journal_last_event_digest"] == GOLDEN_FINAL_DIGEST + + +def test_run_failed_derives_timeout(): + projection = project_run_snapshot( + _minimal_base_snapshot(), + _events_ending_with_failed(status="timeout"), + journal_present=True, + ) + assert projection.status == "timeout" + assert projection.snapshot["status"] == "timeout" + + +def test_run_failed_derives_failed(): + projection = project_run_snapshot( + _minimal_base_snapshot(), + _events_ending_with_failed(status="failed"), + journal_present=True, + ) + assert projection.status == "failed" + assert projection.snapshot["status"] == "failed" + + +def test_run_failed_bad_or_missing_status_raises_bounded_event_payload_error(): + for status in ("bad-status", None): + with pytest.raises(EventPayloadError) as excinfo: + project_run_snapshot( + _minimal_base_snapshot(), + _events_ending_with_failed(status=status), + journal_present=True, + ) + _assert_bounded_projection_error(excinfo) + + +def test_run_interrupted_derives_canceled_and_rejects_wrong_status(): + projection = project_run_snapshot( + _minimal_base_snapshot(), + _events_ending_with_interrupted(status="canceled"), + journal_present=True, + ) + assert projection.status == "canceled" + with pytest.raises(EventPayloadError) as excinfo: + project_run_snapshot( + _minimal_base_snapshot(), + _events_ending_with_interrupted(status="interrupted"), + journal_present=True, + ) + _assert_bounded_projection_error(excinfo) + + +def test_unknown_base_key_raises_bounded_unknown_snapshot_field_error(): + base = _minimal_base_snapshot() + base["unknown_field"] = "nope" + with pytest.raises(UnknownSnapshotFieldError) as excinfo: + project_run_snapshot(base, [], journal_present=False) + _assert_bounded_projection_error(excinfo) + assert "unknown_field" in excinfo.value.diagnostic + + +def test_registered_unmapped_run_paused_raises_bounded_unmapped_event_type_error(): + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + paused = _build_event( + 2, + "run.paused", + {"approval_id": "a-1", "reason": "wait"}, + "pause-1", + "2026-07-27T15:30:46.000000Z", + created["event_digest"], + ) + with pytest.raises(UnmappedEventTypeError) as excinfo: + project_run_snapshot(_minimal_base_snapshot(), [created, paused], journal_present=True) + _assert_bounded_projection_error(excinfo) + + +@pytest.mark.parametrize( + "scenario", + ["gap", "duplicate", "broken_digest", "mixed_run_id"], +) +def test_chain_errors_raise_event_chain_error(scenario): + with pytest.raises(EventChainError) as excinfo: + project_run_snapshot(_minimal_base_snapshot(), _chain_events(scenario), journal_present=True) + _assert_bounded_projection_error(excinfo) + + +def test_invalid_raw_envelope_raises_event_chain_error(): + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + invalid = {**created, "event_type": "run.not-real"} + with pytest.raises(EventChainError) as excinfo: + project_run_snapshot(_minimal_base_snapshot(), [created, invalid], journal_present=True) + _assert_bounded_projection_error(excinfo) + + +def test_invalid_envelope_diagnostic_excludes_rejected_value(): + private_marker = "private-marker-568" + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + invalid = {**created, "recorded_at": private_marker} + with pytest.raises(EventChainError) as excinfo: + project_run_snapshot(_minimal_base_snapshot(), [invalid], journal_present=True) + _assert_bounded_projection_error(excinfo) + assert private_marker not in excinfo.value.diagnostic + + +def test_dataclasses_replace_mutation_of_typed_run_event_raises_event_chain_error(): + event = _golden_events()[0] + mutated = replace(event, event_type="run.planning.started") + with pytest.raises(EventChainError) as excinfo: + project_run_snapshot(_minimal_base_snapshot(), [mutated], journal_present=True) + _assert_bounded_projection_error(excinfo) + + +def test_full_field_fixture_preserves_deep_equality_and_copies_nested_values(): + base = _full_base_snapshot() + assert len(PRESERVED_FIELDS) == 44 + assert DERIVED_FIELDS == { + "status", + "projector_version", + "journal_present", + "journal_last_sequence", + "journal_last_event_digest", + } + assert EVENT_STATUS == { + "run.planning.started": "planning", + "run.dispatch.requested": "dispatching", + "run.dispatch.completed": "result-processing", + "run.synthesis.started": "synthesizing", + "run.synthesis.completed": "handoff", + } + projection = project_run_snapshot(base, [], journal_present=False) + for field in PRESERVED_FIELDS: + assert field in projection.snapshot + assert projection.snapshot[field] == base[field] + assert projection.snapshot["failure"] is not base["failure"] + assert projection.snapshot["recovery_history"] is not base["recovery_history"] + assert projection.snapshot["active_seats"] is not base["active_seats"] + assert projection.snapshot["code_graph_brief"] is not base["code_graph_brief"] + assert set(projection.snapshot.keys()) <= OWNED_FIELDS + + +def test_reprojection_is_byte_idempotent(): + base = _minimal_base_snapshot() + events = _golden_events() + first = project_run_snapshot(base, events, journal_present=True) + second = project_run_snapshot(first.snapshot, events, journal_present=True) + assert first.to_bytes() == second.to_bytes() + + +def test_to_bytes_and_encode_snapshot_bytes_match_sorted_two_space_json(): + projection = project_run_snapshot(_minimal_base_snapshot(), _golden_events(), journal_present=True) + expected = (json.dumps(projection.snapshot, indent=2, sort_keys=True) + "\n").encode("utf-8") + assert projection.to_bytes() == expected + assert encode_snapshot_bytes(projection.snapshot) == expected + + +def test_object_and_circular_preserved_values_raise_bounded_snapshot_encoding_error(): + for value in (object(), {"self": None}): + base = _minimal_base_snapshot() + base["failure"] = value + if isinstance(value, dict): + value["self"] = value + with pytest.raises(SnapshotEncodingError) as excinfo: + project_run_snapshot(base, [], journal_present=False) + _assert_bounded_projection_error(excinfo) + diagnostic = excinfo.value.diagnostic + assert "object" not in diagnostic + assert "circular" not in diagnostic + assert repr(value) not in diagnostic + + +def test_non_boolean_journal_present_raises_projection_input_error(): + for value in (None, 1): + with pytest.raises(ProjectionInputError) as excinfo: + project_run_snapshot(_minimal_base_snapshot(), [], journal_present=value) + _assert_bounded_projection_error(excinfo)