From 08b770c7f508fa0f2305017e2aca6ad80e26f97a Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 23:39:24 -0400 Subject: [PATCH] feat(run-journal): measure lifecycle journal ceiling for #635 Extend the measurement harness with volume mode (real journal scan, representative and worst-case synthetic runs) and record the raise-bounds design decision before any production ceiling change. Co-authored-by: Cursor --- docs/phase-635-journal-ceiling-decision.md | 73 +++++ scripts/measure_run_journal.py | 356 ++++++++++++++++++++- tests/test_run_journal_measurement.py | 75 +++++ 3 files changed, 494 insertions(+), 10 deletions(-) create mode 100644 docs/phase-635-journal-ceiling-decision.md diff --git a/docs/phase-635-journal-ceiling-decision.md b/docs/phase-635-journal-ceiling-decision.md new file mode 100644 index 00000000..be5618fe --- /dev/null +++ b/docs/phase-635-journal-ceiling-decision.md @@ -0,0 +1,73 @@ +# Issue #635: journal ceiling decision (measure, then design) + +Status: design decision awaiting grading. No production implementation in this change. + +## Measured numbers + +Artifact: `.brigade/measurements/issue-635-journal-ceiling.json` (local, same shape family as `issue-568-slice6.json`: environment metadata plus structured aggregates; volume mode adds bounds and scenario blocks). + +Current hard bounds: `MAX_JOURNAL_EVENTS = 512`, `MAX_JOURNAL_BYTES = 8 MiB`. + +| Scenario | Events | Bytes | % event ceiling | % byte ceiling | Hit ceiling? | +| --- | ---: | ---: | ---: | ---: | --- | +| Scanned real runs (n=13, single-seat journal-hardening runs) | p50/max 17 | p50 14,279 / max 14,286 | 3.3% | 0.17% | no | +| Synthetic representative (1 seat, 1 attempt, 0 pauses) | 11 | 7,780 | 2.1% | 0.09% | no | +| Synthetic roster-sized (11 seats, 2 attempts, 3 pause/resume cycles) | 155 | 124,777 | 30.3% | 1.5% | no | +| Synthetic worst case (32 seats, 3 attempts, 8 pause cycles) | 512 | 420,683 | 100% | 5.0% | yes, mid-dispatch | + +Worst-case stop reason: `LifecycleJournalError: bound exceeded: journal event sequence above MAX_JOURNAL_EVENTS` at `seat-28` attempt 1. Bytes were still ~5% of the 8 MiB cap. + +## Decision + +Raise the event bound. Do not segment the journal yet. + +Reasoning: + +1. Event count is the binding constraint. At a full 512-event journal the file is only ~0.4 MiB. +2. Realistic and roster-sized load stays far from 512 (3% real, 30% for 11 seats with retries and three approval cycles). +3. The ceiling is reachable, but only under a synthetic fleet larger than today's 11-seat roster with three attempts per seat. +4. Segmentation (cross-linked `lifecycle.NNNNN.jsonl`, reader/projector/doctor/recovery awareness) is a large surface change. The measured gap does not justify that cost before a cheaper bound raise plus a soft warning. +5. Keep segmentation as the explicit escape hatch if #593 budget events or larger fleets push roster-sized runs past ~50% of the raised ceiling. + +Recommended production change (separate implementation issue, after this design is graded): + +- Raise `MAX_JOURNAL_EVENTS` from 512 to 2048 (4x). +- Leave `MAX_JOURNAL_BYTES` at 8 MiB for now (still ~20x headroom at 2048 median-sized events). +- Add a doctor soft threshold at 75% of the event bound (1536 of 2048): `WARN`, not `FAIL`. +- Keep the hard refuse-to-append behavior only at the new ceiling, and make that failure visible (already true today via `LifecycleJournalError`). + +## Soft-threshold doctor warning + +When `brigade doctor` inspects a run journal and `event_count >= ceil(0.75 * MAX_JOURNAL_EVENTS)` while still under the hard ceiling: + +- status: `WARN` +- check name: `runs: journal event headroom` +- detail shape: `lifecycle journal at {event_count}/{MAX_JOURNAL_EVENTS} events ({pct}%); raise or segment before the hard ceiling halts appends` + +Below 75%: silent on this check. At or above the hard ceiling: keep today's `FAIL` / `bound exceeded` path for recovery verdicts that already treat oversize journals as failed. + +## Recovery after upgrade for a run already at the old ceiling + +A run that stopped at sequence 512 under the old bound has a complete, chain-valid journal and paired checkpoints. It does not need rewrite or segmentation to become readable again. + +After the bound raise ships: + +1. Readers (`read_journal` / `read_journal_bounded`) accept the existing 512-event file because it is under the new ceiling. +2. Doctor recovery checks that previously returned `FAIL` / `bound exceeded` solely for the old event ceiling clear once the constant moves. +3. If the run is still live or being resumed, the active lock owner may append sequence 513+ again; checkpoint pairing and projection continue from the existing chain tip. +4. No migration tool is required for journals that stopped exactly at 512. Journals that somehow exceed the new byte bound remain fail-closed (unchanged). + +## Explicit non-goals for the follow-up implementation issue + +- No journal segmentation. +- No change to worker/provider event streams (#592 stays separate). +- No silent truncation or dropping of lifecycle facts when the hard ceiling is hit. + +## Reproducing the measurement + +```bash +python scripts/measure_run_journal.py \ + --mode volume \ + --scan /path/to/.brigade/runs \ + --output .brigade/measurements/issue-635-journal-ceiling.json +``` diff --git a/scripts/measure_run_journal.py b/scripts/measure_run_journal.py index 99c8b449..2c2a009f 100644 --- a/scripts/measure_run_journal.py +++ b/scripts/measure_run_journal.py @@ -1,10 +1,17 @@ #!/usr/bin/env python3 -"""Measure journal-authority append latency across N real run directories. +"""Measure journal-authority append latency and lifecycle journal volume. -Drives each run directory through the real Brigade journal-authority path -(``run_lifecycle`` + ``run_checkpoint`` under a real ``runguard.run_lock``), -records per-append wall-clock latency with ``time.perf_counter_ns``, and -reports deterministic integer percentiles plus environment metadata. +Latency mode drives each run directory through the real Brigade +journal-authority path (``run_lifecycle`` + ``run_checkpoint`` under a real +``runguard.run_lock``), records per-append wall-clock latency with +``time.perf_counter_ns``, and reports deterministic integer percentiles plus +environment metadata. + +Volume mode (issue #635) reports event counts and journal bytes for: + +- scanned real ``lifecycle.jsonl`` files (optional ``--scan`` roots) +- a synthetic representative run (one seat, no retries, no pause cycles) +- a synthetic worst case (configurable seats, attempts-per-seat, pause cycles) This is a measurement harness only: it carries no acceptance threshold and no pass/fail verdict. Standard library plus the in-repo ``brigade`` package; @@ -23,15 +30,20 @@ import sys import tempfile import time +from collections import Counter from pathlib import Path from typing import Any -from brigade import run_checkpoint, run_lifecycle, runguard +from brigade import run_checkpoint, run_journal, run_lifecycle, runguard _MEASURED_STATUS = "planning" _MOUNTPOINT_ESCAPE_RE = re.compile(r"\\([0-7]{3})") +_DEFAULT_WORST_SEATS = 32 +_DEFAULT_WORST_ATTEMPTS = 3 +_DEFAULT_WORST_PAUSE_CYCLES = 8 + def _bootstrap_run(run_dir: Path) -> bytes: """Write the minimal canonical run.json carrying both durable request fields. @@ -80,8 +92,14 @@ def _measure_run(run_dir: Path, workspace: Path) -> dict[str, Any]: workspace=workspace, ) latencies.append(time.perf_counter_ns() - start) - journal_bytes = (run_dir / "events" / "lifecycle.jsonl").stat().st_size - return {"journal_bytes": journal_bytes, "append_latencies_ns": latencies} + journal_path = run_dir / "events" / "lifecycle.jsonl" + journal_bytes = journal_path.stat().st_size + event_count = _count_journal_events(journal_path) + return { + "journal_bytes": journal_bytes, + "event_count": event_count, + "append_latencies_ns": latencies, + } def _percentile_nearest_rank(sorted_values: list[int], percentile: float) -> int: @@ -160,6 +178,13 @@ def _environment(root: Path) -> dict[str, Any]: } +def _bounds() -> dict[str, int]: + return { + "max_journal_events": int(run_checkpoint.MAX_JOURNAL_EVENTS), + "max_journal_bytes": int(run_checkpoint.MAX_JOURNAL_BYTES), + } + + def _lock_workspace_parent(root: Path, staging_root: Path) -> Path | None: """Pick a parent directory for the runguard lock workspace. @@ -186,6 +211,221 @@ def _lock_workspace_parent(root: Path, staging_root: Path) -> Path | None: return staging_root +def _count_journal_events(journal_path: Path) -> int: + """Count complete journal events, preferring the journal reader. + + Falls back to nonempty line count when the reader rejects the file (bound + exceeded, corrupt chain) or returns no parseable events while the file + still has content. Volume surveys must not abort on a single bad journal. + """ + line_count = sum(1 for line in journal_path.read_text(encoding="utf-8").splitlines() if line.strip()) + try: + report = run_journal.read_journal(journal_path) + except (run_journal.RunJournalError, OSError, UnicodeError): + return line_count + if report.events: + return len(report.events) + return line_count + + +def _journal_volume_entry(journal_path: Path, *, label: str) -> dict[str, Any]: + journal_bytes = journal_path.stat().st_size + event_count = _count_journal_events(journal_path) + event_types: dict[str, int] = {} + try: + report = run_journal.read_journal(journal_path) + event_types = dict(sorted(Counter(event.event_type for event in report.events).items())) + except run_journal.RunJournalError: + event_types = {} + bounds = _bounds() + return { + "label": label, + "path": str(journal_path), + "event_count": event_count, + "journal_bytes": journal_bytes, + "event_type_counts": event_types, + "headroom_events": bounds["max_journal_events"] - event_count, + "headroom_bytes": bounds["max_journal_bytes"] - journal_bytes, + "pct_events": round(100.0 * event_count / bounds["max_journal_events"], 4), + "pct_bytes": round(100.0 * journal_bytes / bounds["max_journal_bytes"], 6), + } + + +def scan_lifecycle_journals(roots: list[Path]) -> dict[str, Any]: + """Collect event-count/byte stats for ``lifecycle.jsonl`` under ``roots``.""" + journals: list[Path] = [] + for root in roots: + root = root.expanduser().resolve() + if not root.exists(): + continue + journals.extend(sorted(root.rglob("lifecycle.jsonl"))) + per_run = [_journal_volume_entry(path, label=path.parent.parent.name) for path in journals] + event_counts = sorted(entry["event_count"] for entry in per_run) + byte_counts = sorted(entry["journal_bytes"] for entry in per_run) + aggregate: dict[str, Any] + if event_counts: + aggregate = { + "runs": len(per_run), + "event_count_p50": _percentile_nearest_rank(event_counts, 50), + "event_count_p95": _percentile_nearest_rank(event_counts, 95), + "event_count_max": int(event_counts[-1]), + "journal_bytes_p50": _percentile_nearest_rank(byte_counts, 50), + "journal_bytes_p95": _percentile_nearest_rank(byte_counts, 95), + "journal_bytes_max": int(byte_counts[-1]), + } + else: + aggregate = { + "runs": 0, + "event_count_p50": 0, + "event_count_p95": 0, + "event_count_max": 0, + "journal_bytes_p50": 0, + "journal_bytes_p95": 0, + "journal_bytes_max": 0, + } + return {"roots": [str(path.expanduser().resolve()) for path in roots], "per_run": per_run, "aggregate": aggregate} + + +def _set_approval_reference(run_dir: Path, approval_id: str, cycle: int, *, decision_state: str) -> dict[str, Any]: + snap = json.loads((run_dir / "run.json").read_text(encoding="utf-8")) + if not isinstance(snap, dict): + raise TypeError("run.json must contain an object") + snap["approval_reference"] = { + "approval_id": approval_id, + "source": "daily", + "fingerprint": f"fp-{cycle}", + "source_fingerprint": f"sfp-{cycle}", + "contract_fingerprint": f"cfp-{cycle}", + "evidence_fingerprint": f"efp-{cycle}", + "decision_state": decision_state, + } + (run_dir / "run.json").write_text(json.dumps(snap, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return snap + + +def _drive_volume_scenario( + run_dir: Path, + workspace: Path, + *, + seats: int, + attempts_per_seat: int, + pause_cycles: int, +) -> dict[str, Any]: + """Drive a synthetic authoritative run and return volume stats.""" + if seats < 0 or attempts_per_seat < 1 or pause_cycles < 0: + raise ValueError("seats>=0, attempts_per_seat>=1, pause_cycles>=0 required") + _bootstrap_run(run_dir) + stopped_reason: str | None = None + with runguard.run_lock(workspace, run_dir=run_dir): + run_lifecycle.prepare_lifecycle_journal(run_dir, workspace=workspace) + for status in ("planning", "dispatching"): + try: + run_lifecycle.record_lifecycle_transition(run_dir, status=status, workspace=workspace) + except Exception as exc: # noqa: BLE001 - harness records and continues + stopped_reason = f"status {status}: {type(exc).__name__}: {exc}" + break + if stopped_reason is None: + for seat_index in range(seats): + seat = f"seat-{seat_index:02d}" + for attempt_index in range(attempts_per_seat): + try: + requested = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=workspace, + event_type="run.dispatch.requested", + seat=seat, + ) + if requested is None: + raise RuntimeError("dispatch requested returned None") + attempt = requested.payload["attempt"] + if not isinstance(attempt, int): + raise TypeError("dispatch attempt must be int") + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=workspace, + event_type="run.dispatch.observed", + seat=seat, + attempt=attempt, + ) + terminal = ( + "run.dispatch.completed" + if attempt_index == attempts_per_seat - 1 + else "run.dispatch.failed" + ) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=workspace, + event_type=terminal, + seat=seat, + attempt=attempt, + ) + except Exception as exc: # noqa: BLE001 + stopped_reason = f"dispatch seat={seat} attempt={attempt_index}: {type(exc).__name__}: {exc}" + break + if stopped_reason is not None: + break + if stopped_reason is None: + for cycle in range(pause_cycles): + approval_id = f"approval-{cycle}" + try: + run_lifecycle.record_lifecycle_event( + run_dir, + event_type="approval.requested", + payload={ + "approval_id": approval_id, + "source": "daily", + "contract_fingerprint": f"cfp-{cycle}", + }, + idempotency_key=run_lifecycle.approval_idempotency_key(approval_id, "requested"), + workspace=workspace, + ) + snap = _set_approval_reference(run_dir, approval_id, cycle, decision_state="pending") + run_lifecycle.record_lifecycle_transition( + run_dir, + status="paused", + workspace=workspace, + incoming_snapshot=snap, + ) + snap = _set_approval_reference(run_dir, approval_id, cycle, decision_state="approved") + run_lifecycle.record_lifecycle_event( + run_dir, + event_type="approval.consumed", + payload={"approval_id": approval_id, "consuming_run_id": run_dir.name}, + idempotency_key=run_lifecycle.approval_idempotency_key( + approval_id, "consumed", scope=run_dir.name + ), + workspace=workspace, + ) + run_lifecycle.record_lifecycle_transition( + run_dir, + status="running", + workspace=workspace, + incoming_snapshot=snap, + ) + except Exception as exc: # noqa: BLE001 + stopped_reason = f"pause cycle={cycle}: {type(exc).__name__}: {exc}" + break + if stopped_reason is None and seats > 0: + for status in ("result-processing", "synthesizing", "ok"): + try: + run_lifecycle.record_lifecycle_transition(run_dir, status=status, workspace=workspace) + except Exception as exc: # noqa: BLE001 + stopped_reason = f"terminal {status}: {type(exc).__name__}: {exc}" + break + + journal_path = run_dir / "events" / "lifecycle.jsonl" + entry = _journal_volume_entry(journal_path, label=run_dir.name) + entry["config"] = { + "seats": seats, + "attempts_per_seat": attempts_per_seat, + "pause_cycles": pause_cycles, + } + entry["stopped_reason"] = stopped_reason + entry["hit_event_ceiling"] = entry["event_count"] >= run_checkpoint.MAX_JOURNAL_EVENTS + entry["hit_byte_ceiling"] = entry["journal_bytes"] >= run_checkpoint.MAX_JOURNAL_BYTES + return entry + + def measure_runs(runs: int, root: Path) -> dict[str, Any]: """Drive ``runs`` run directories under ``root`` through journal authority. @@ -237,6 +477,63 @@ def measure_runs(runs: int, root: Path) -> dict[str, Any]: } +def measure_volume( + root: Path, + *, + scan_roots: list[Path] | None = None, + roster_seats: int = 11, + roster_attempts_per_seat: int = 2, + roster_pause_cycles: int = 3, + worst_seats: int = _DEFAULT_WORST_SEATS, + worst_attempts_per_seat: int = _DEFAULT_WORST_ATTEMPTS, + worst_pause_cycles: int = _DEFAULT_WORST_PAUSE_CYCLES, +) -> dict[str, Any]: + """Measure lifecycle journal event counts and bytes for issue #635.""" + if not isinstance(root, Path): + raise TypeError("root must be a pathlib.Path") + root = root.expanduser().resolve() + root.mkdir(parents=True, exist_ok=True) + + scanned = scan_lifecycle_journals(scan_roots or []) + with tempfile.TemporaryDirectory(prefix="brigade-run-journal-volume-", dir=root) as staging: + staging_root = Path(staging) + lock_parent = _lock_workspace_parent(root, staging_root) + with tempfile.TemporaryDirectory(prefix="brigade-run-journal-lock-", dir=lock_parent) as lock_dir: + workspace = Path(lock_dir) + representative = _drive_volume_scenario( + staging_root / "runs" / "representative", + workspace, + seats=1, + attempts_per_seat=1, + pause_cycles=0, + ) + roster_sized = _drive_volume_scenario( + staging_root / "runs" / "roster-sized", + workspace, + seats=roster_seats, + attempts_per_seat=roster_attempts_per_seat, + pause_cycles=roster_pause_cycles, + ) + worst_case = _drive_volume_scenario( + staging_root / "runs" / "worst-case", + workspace, + seats=worst_seats, + attempts_per_seat=worst_attempts_per_seat, + pause_cycles=worst_pause_cycles, + ) + + return { + "issue": 635, + "kind": "journal-volume", + "bounds": _bounds(), + "environment": _environment(root), + "scanned_real_runs": scanned, + "representative_synthetic": representative, + "roster_sized_synthetic": roster_sized, + "worst_case_synthetic": worst_case, + } + + def write_report(report: dict[str, Any], output: Path) -> None: """Write ``report`` as canonical JSON (indent=2, sort_keys) plus newline.""" output = Path(output) @@ -246,12 +543,51 @@ def write_report(report: dict[str, Any], output: Path) -> None: def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--runs", type=int, default=1000, help="number of run directories to measure") + parser.add_argument( + "--mode", + choices=("latency", "volume", "all"), + default="latency", + help="latency (default, issue #568), volume (issue #635), or both", + ) + parser.add_argument("--runs", type=int, default=1000, help="number of run directories for latency mode") parser.add_argument("--output", required=True, help="exact file path for the JSON report") + parser.add_argument( + "--scan", + action="append", + default=[], + help="root directory to scan for lifecycle.jsonl (repeatable; volume mode)", + ) + parser.add_argument("--worst-seats", type=int, default=_DEFAULT_WORST_SEATS) + parser.add_argument("--worst-attempts-per-seat", type=int, default=_DEFAULT_WORST_ATTEMPTS) + parser.add_argument("--worst-pause-cycles", type=int, default=_DEFAULT_WORST_PAUSE_CYCLES) args = parser.parse_args(argv) if args.runs <= 0: parser.error("--runs must be a positive integer") - report = measure_runs(args.runs, root=Path.cwd()) + if args.worst_seats < 0 or args.worst_attempts_per_seat < 1 or args.worst_pause_cycles < 0: + parser.error("worst-case knobs must be seats>=0, attempts>=1, pause_cycles>=0") + + root = Path.cwd() + if args.mode == "latency": + report: dict[str, Any] = measure_runs(args.runs, root=root) + elif args.mode == "volume": + report = measure_volume( + root, + scan_roots=[Path(path) for path in args.scan], + worst_seats=args.worst_seats, + worst_attempts_per_seat=args.worst_attempts_per_seat, + worst_pause_cycles=args.worst_pause_cycles, + ) + else: + report = { + "latency": measure_runs(args.runs, root=root), + "volume": measure_volume( + root, + scan_roots=[Path(path) for path in args.scan], + worst_seats=args.worst_seats, + worst_attempts_per_seat=args.worst_attempts_per_seat, + worst_pause_cycles=args.worst_pause_cycles, + ), + } write_report(report, Path(args.output)) return 0 diff --git a/tests/test_run_journal_measurement.py b/tests/test_run_journal_measurement.py index b3a43352..ab50f1e9 100644 --- a/tests/test_run_journal_measurement.py +++ b/tests/test_run_journal_measurement.py @@ -39,6 +39,8 @@ def test_measure_runs_returns_report_shape(tmp_path): for entry in per_run: assert isinstance(entry["journal_bytes"], int) assert entry["journal_bytes"] > 0 + assert isinstance(entry["event_count"], int) + assert entry["event_count"] > 0 latencies = entry["append_latencies_ns"] assert isinstance(latencies, list) assert latencies @@ -198,3 +200,76 @@ def test_measure_runs_nested_under_outer_run_lock(tmp_path): assert repo_leftovers == [], f"measurement temp dirs left under repo: {repo_leftovers}" parent_leftovers = [path.name for path in tmp_path.iterdir() if path.name.startswith("brigade-run-journal-")] assert parent_leftovers == [], f"lock workspace temp dirs left under repo parent: {parent_leftovers}" + + +def test_measure_volume_report_shape(tmp_path): + module = _load_measure_run_journal_module() + + report = module.measure_volume( + tmp_path, + roster_seats=2, + roster_attempts_per_seat=1, + roster_pause_cycles=0, + worst_seats=2, + worst_attempts_per_seat=2, + worst_pause_cycles=1, + ) + + assert report["issue"] == 635 + assert report["kind"] == "journal-volume" + assert report["bounds"]["max_journal_events"] > 0 + assert report["bounds"]["max_journal_bytes"] > 0 + assert {"python_version", "platform", "filesystem_type", "directory_fsync_supported"} <= set(report["environment"]) + for key in ("representative_synthetic", "roster_sized_synthetic", "worst_case_synthetic"): + entry = report[key] + assert isinstance(entry["event_count"], int) and entry["event_count"] > 0 + assert isinstance(entry["journal_bytes"], int) and entry["journal_bytes"] > 0 + assert "headroom_events" in entry + assert "pct_events" in entry + assert "event_type_counts" in entry + assert report["representative_synthetic"]["event_count"] < report["worst_case_synthetic"]["event_count"] + assert report["scanned_real_runs"]["aggregate"]["runs"] == 0 + + +def test_scan_lifecycle_journals_reads_existing_files(tmp_path): + module = _load_measure_run_journal_module() + journal = tmp_path / "runs" / "run-a" / "events" / "lifecycle.jsonl" + journal.parent.mkdir(parents=True) + # Non-canonical lines still contribute to the volume survey via line count. + journal.write_text('{"event_type":"run.created"}\n{"event_type":"run.completed"}\n') + + scanned = module.scan_lifecycle_journals([tmp_path]) + + assert scanned["aggregate"]["runs"] == 1 + assert scanned["aggregate"]["event_count_max"] == 2 + assert scanned["per_run"][0]["journal_bytes"] == journal.stat().st_size + + +def test_cli_volume_mode_writes_report(tmp_path): + output = tmp_path / "out" / "volume.json" + + result = subprocess.run( + [ + sys.executable, + str(SCRIPT), + "--mode", + "volume", + "--worst-seats", + "1", + "--worst-attempts-per-seat", + "1", + "--worst-pause-cycles", + "0", + "--output", + str(output), + ], + cwd=tmp_path, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + report = json.loads(output.read_text()) + assert report["kind"] == "journal-volume" + assert "threshold" not in output.read_text() + assert "pass" not in output.read_text()