diff --git a/.claude/memory-handoffs/20260728-030215-issue-567-write-once-sidecars.md b/.claude/memory-handoffs/20260728-030215-issue-567-write-once-sidecars.md new file mode 100644 index 00000000..2b0d4cb6 --- /dev/null +++ b/.claude/memory-handoffs/20260728-030215-issue-567-write-once-sidecars.md @@ -0,0 +1,41 @@ +# Memory Handoff + +## Type + +bugfix + +## Title + +Write-once run sidecar revisions + +## Summary + +Resume and patch-reference updates previously replaced worker and synthesis sidecars in place. They now append private, immutable revisions while retaining the legacy files as current compatibility projections. + +## Durable facts + +- Initial emissions and every update append six-digit JSON revisions below `revisions/worker-results/` and `revisions/synthesis/`. +- A first update of a legacy run imports the compatibility projection as revision 000001 before appending the new revision. +- A failed compatibility projection write must not replace the old projection after immutable evidence has been written. +- Revision files use mode 0600. POSIX writes fsync each new directory entry, the revision file, and the leaf directory before updating the projection. +- Existing readers continue loading `worker-results.json` and `synthesis.json`; no receipt schema or dependency changed. + +## Evidence + +- files changed: `src/brigade/aboyeur.py`, `src/brigade/run_resume.py`, `tests/test_aboyeur.py`, `tests/test_run_resume.py`, `tests/test_runs_cmd.py` +- RED receipts: `20260728-030303-work-verify-9ae890`, `20260728-031100-work-verify-052751`, `20260728-031433-work-verify-ca2331`, `20260728-031552-work-verify-cf2e04` +- GREEN receipts: `20260728-031646-work-verify-d12faa`, `20260728-031656-work-verify-8bfb8d`, `20260728-031822-work-verify-4cb89a`, `20260728-031829-work-verify-d68b11` + +## Recommended memory action + +no-card + +## Target document + +.learnings/LEARNINGS.md + +## Suggested document content + +### Write-once run sidecar revisions + +Worker and synthesis evidence is append-only under `revisions//`, while the original sidecar paths remain current compatibility projections. On the first update of a legacy run, archive the exact old bytes before appending the updated document. Persist revision evidence before replacing the projection, and retain the new revision if projection replacement fails. On POSIX, directory-entry fsync ordering is part of the durability contract. diff --git a/src/brigade/aboyeur.py b/src/brigade/aboyeur.py index 216f477a..1dd231b5 100644 --- a/src/brigade/aboyeur.py +++ b/src/brigade/aboyeur.py @@ -492,6 +492,92 @@ def _write_json(path: Path, payload: object) -> None: localio.write_text_atomic(path, json.dumps(payload, indent=2, sort_keys=True) + "\n") +def _revision_contains(revisions_dir: Path, projection: bytes) -> bool: + """Return whether a preserved sidecar revision matches ``projection``.""" + if not revisions_dir.is_dir(): + return False + for revision_path in revisions_dir.glob("*.json"): + try: + if revision_path.read_bytes() == projection: + return True + except OSError: + continue + return False + + +def _supports_directory_fsync() -> bool: + return os.name == "posix" + + +def _fsync_directory(path: Path) -> None: + """Persist directory entries where the platform supports directory fsync.""" + if not _supports_directory_fsync(): + return + directory_fd = os.open(path, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + + +def _ensure_directory_durable(path: Path) -> None: + """Create missing directory levels and persist each new parent entry.""" + missing: list[Path] = [] + current = path + while not current.exists(): + missing.append(current) + current = current.parent + for directory in reversed(missing): + try: + directory.mkdir() + except FileExistsError: + if not directory.is_dir(): + raise + else: + _fsync_directory(directory.parent) + + +def _write_new_revision(revisions_dir: Path, encoded: bytes) -> None: + """Exclusively create the next immutable sidecar revision.""" + _ensure_directory_durable(revisions_dir) + sequence = max( + (int(path.stem) for path in revisions_dir.glob("*.json") if path.stem.isascii() and path.stem.isdecimal()), + default=0, + ) + while True: + sequence += 1 + revision_path = revisions_dir / f"{sequence:06d}.json" + writer_acquired = False + try: + fd = os.open(revision_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + writer_acquired = True + with os.fdopen(fd, "wb") as handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + _fsync_directory(revisions_dir) + except FileExistsError: + continue + except BaseException: + if writer_acquired: + revision_path.unlink(missing_ok=True) + raise + return + + +def write_sidecar_revision(run_dir: Path, filename: str, payload: object) -> None: + """Append an immutable sidecar revision, then update its compatibility file.""" + projection_path = run_dir / filename + revisions_dir = run_dir / "revisions" / Path(filename).stem + if projection_path.exists(): + legacy_projection = projection_path.read_bytes() + json.loads(legacy_projection) + if not _revision_contains(revisions_dir, legacy_projection): + _write_new_revision(revisions_dir, legacy_projection) + _write_new_revision(revisions_dir, (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8")) + _write_json(projection_path, payload) + + def _utc_iso(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") @@ -1759,7 +1845,7 @@ def set_artifact_patch_ref(output_dir: Path, patch_ref: str = "changes.patch") - else: receipt_schema.stamp_synthesis_document(payload) try: - _write_json(path, payload) + write_sidecar_revision(output_dir, filename, payload) except OSError as exc: raise runguard.RunGuardError(f"failed to record artifact patch reference in {filename}: {exc}") from exc @@ -2983,8 +3069,9 @@ def dispatch_interrupted() -> None: worker_results = _mark_noop_worker_results(worker_results, suspected_noop) if output_dir is not None: worker_results = _write_worker_logs(output_dir, worker_results) - _write_json( - output_dir / "worker-results.json", + write_sidecar_revision( + output_dir, + "worker-results.json", receipt_schema.worker_results_document( _worker_payload(worker_results), ground_truth=ground_truth, @@ -3079,7 +3166,7 @@ def dispatch_interrupted() -> None: ground_truth=ground_truth, ) ) - _write_json(output_dir / "synthesis.json", synthesis_payload) + write_sidecar_revision(output_dir, "synthesis.json", synthesis_payload) if not final.ok: if output_dir is not None: finished_at = datetime.now(timezone.utc) diff --git a/src/brigade/run_resume.py b/src/brigade/run_resume.py index eccbf56c..1736238b 100644 --- a/src/brigade/run_resume.py +++ b/src/brigade/run_resume.py @@ -199,8 +199,9 @@ def _resume_locked(run_dir: Path) -> int: for r in results ] ground_truth = worker_data.get("ground_truth") or {} - aboyeur._write_json( - run_dir / "worker-results.json", + aboyeur.write_sidecar_revision( + run_dir, + "worker-results.json", receipt_schema.worker_results_document( aboyeur._worker_payload(worker_results), ground_truth=ground_truth, @@ -226,8 +227,9 @@ def _resume_locked(run_dir: Path) -> int: reasoning=orchestrator.reasoning, env=dict(orchestrator.env) if orchestrator.env is not None else None, ) - aboyeur._write_json( - run_dir / "synthesis.json", + aboyeur.write_sidecar_revision( + run_dir, + "synthesis.json", receipt_schema.synthesis_document( orchestrator=roster.orchestrator, result={"ok": final.ok, "detail": final.detail, "text": final.text}, diff --git a/tests/test_aboyeur.py b/tests/test_aboyeur.py index 9f0723ee..100fbaa6 100644 --- a/tests/test_aboyeur.py +++ b/tests/test_aboyeur.py @@ -1,6 +1,7 @@ import json import os import signal +import stat import subprocess import sys import time @@ -31,6 +32,10 @@ def _roster(): ) +def _sidecar_revisions(run_dir: Path, sidecar: str) -> list[Path]: + return sorted((run_dir / "revisions" / sidecar).glob("*.json")) + + def _roster_with_incapable_worker(): return Roster( orchestrator="chef", @@ -2613,6 +2618,206 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): assert "Brigade-computed facts:" in calls[-1][1] assert "tracked.txt" in calls[-1][1] assert {call[2] for call in calls} == {run_cwd} + worker_revisions = _sidecar_revisions(output_dir, "worker-results") + synthesis_revisions = _sidecar_revisions(output_dir, "synthesis") + assert [path.name for path in worker_revisions] == ["000001.json"] + assert [path.name for path in synthesis_revisions] == ["000001.json"] + assert json.loads(worker_revisions[0].read_text()) == worker_results + assert json.loads(synthesis_revisions[0].read_text()) == synthesis + + +def test_set_artifact_patch_ref_appends_immutable_sidecar_revisions(tmp_path): + output_dir = tmp_path / "run" + output_dir.mkdir() + initial_worker = { + "results": [{"worker": "coder", "text": "initial"}], + "ground_truth": {"patch_ref": None}, + } + initial_synthesis = { + "orchestrator": "chef", + "result": {"text": "initial"}, + "ground_truth": {"patch_ref": None}, + } + (output_dir / "worker-results.json").write_text(json.dumps(initial_worker) + "\n") + (output_dir / "synthesis.json").write_text(json.dumps(initial_synthesis) + "\n") + + aboyeur.set_artifact_patch_ref(output_dir, "first.patch") + aboyeur.set_artifact_patch_ref(output_dir, "second.patch") + + worker_revisions = _sidecar_revisions(output_dir, "worker-results") + synthesis_revisions = _sidecar_revisions(output_dir, "synthesis") + assert [path.name for path in worker_revisions] == ["000001.json", "000002.json", "000003.json"] + assert [path.name for path in synthesis_revisions] == ["000001.json", "000002.json", "000003.json"] + assert json.loads(worker_revisions[0].read_text()) == initial_worker + assert json.loads(synthesis_revisions[0].read_text()) == initial_synthesis + assert json.loads(worker_revisions[1].read_text())["ground_truth"]["patch_ref"] == "first.patch" + assert json.loads(synthesis_revisions[1].read_text())["ground_truth"]["patch_ref"] == "first.patch" + assert json.loads(worker_revisions[2].read_text())["ground_truth"]["patch_ref"] == "second.patch" + assert json.loads(synthesis_revisions[2].read_text())["ground_truth"]["patch_ref"] == "second.patch" + assert json.loads((output_dir / "worker-results.json").read_text())["ground_truth"]["patch_ref"] == "second.patch" + assert json.loads((output_dir / "synthesis.json").read_text())["ground_truth"]["patch_ref"] == "second.patch" + + +def test_patch_reference_projection_write_failure_keeps_old_file_after_recording_evidence(tmp_path, monkeypatch): + output_dir = tmp_path / "run" + output_dir.mkdir() + worker_path = output_dir / "worker-results.json" + initial_worker = { + "results": [{"worker": "coder", "text": "initial"}], + "ground_truth": {"patch_ref": None}, + } + worker_path.write_text(json.dumps(initial_worker) + "\n") + before = worker_path.read_bytes() + real_write_text_atomic = aboyeur.localio.write_text_atomic + + def fail_worker_projection(path, text): + if path == worker_path: + raise OSError("projection disk full") + real_write_text_atomic(path, text) + + monkeypatch.setattr(aboyeur.localio, "write_text_atomic", fail_worker_projection) + + with pytest.raises( + runguard.RunGuardError, + match="failed to record artifact patch reference in worker-results.json", + ): + aboyeur.set_artifact_patch_ref(output_dir, "changes.patch") + + revisions = _sidecar_revisions(output_dir, "worker-results") + assert worker_path.read_bytes() == before + assert [path.name for path in revisions] == ["000001.json", "000002.json"] + assert json.loads(revisions[0].read_text()) == initial_worker + assert json.loads(revisions[1].read_text())["ground_truth"]["patch_ref"] == "changes.patch" + + +def test_write_sidecar_revision_preserves_legacy_worker_results_bytes(tmp_path): + output_dir = tmp_path / "run" + output_dir.mkdir() + projection_path = output_dir / "worker-results.json" + legacy_bytes = b'{ "z": [3, 2], "a" : { "legacy": true } }\n' + projection_path.write_bytes(legacy_bytes) + payload = {"a": {"legacy": False}, "z": [1]} + + aboyeur.write_sidecar_revision(output_dir, "worker-results.json", payload) + + revisions = _sidecar_revisions(output_dir, "worker-results") + assert [path.name for path in revisions] == ["000001.json", "000002.json"] + assert revisions[0].read_bytes() == legacy_bytes + assert projection_path.read_text() == json.dumps(payload, indent=2, sort_keys=True) + "\n" + + +def test_write_sidecar_revision_creates_private_revision_under_umask_022(tmp_path): + output_dir = tmp_path / "run" + output_dir.mkdir() + previous_umask = os.umask(0o022) + try: + aboyeur.write_sidecar_revision(output_dir, "worker-results.json", {"ok": True}) + finally: + os.umask(previous_umask) + + revision = _sidecar_revisions(output_dir, "worker-results")[0] + assert stat.S_IMODE(revision.stat().st_mode) == 0o600 + + +def test_write_sidecar_revision_fsyncs_revision_directory_before_projection_write(tmp_path, monkeypatch): + output_dir = tmp_path / "run" + output_dir.mkdir() + projection_path = output_dir / "worker-results.json" + revisions_root = output_dir / "revisions" + revisions_dir = output_dir / "revisions" / "worker-results" + events: list[str] = [] + directory_fds: dict[int, str] = {} + real_open = aboyeur.os.open + real_close = aboyeur.os.close + real_fsync = aboyeur.os.fsync + + def recording_open(path, flags, mode=0o777, *, dir_fd=None): + fd = real_open(path, flags, mode, dir_fd=dir_fd) + directory_names = { + output_dir: "run-directory", + revisions_root: "revisions-directory", + revisions_dir: "sidecar-directory", + } + if Path(path) in directory_names: + directory_fds[fd] = directory_names[Path(path)] + return fd + + def recording_fsync(fd: int) -> None: + target = directory_fds.get(fd, "revision-file") + events.append(f"fsync:{target}") + real_fsync(fd) + + def recording_close(fd: int) -> None: + directory_fds.pop(fd, None) + real_close(fd) + + def recording_projection_write(path, text) -> None: + assert path == projection_path + events.append("projection-write") + + monkeypatch.setattr(aboyeur.os, "open", recording_open) + monkeypatch.setattr(aboyeur.os, "close", recording_close) + monkeypatch.setattr(aboyeur.os, "fsync", recording_fsync) + monkeypatch.setattr(aboyeur.localio, "write_text_atomic", recording_projection_write) + + aboyeur.write_sidecar_revision(output_dir, "worker-results.json", {"ok": True}) + + assert events == [ + "fsync:run-directory", + "fsync:revisions-directory", + "fsync:revision-file", + "fsync:sidecar-directory", + "projection-write", + ] + + +def test_write_sidecar_revision_skips_directory_fsync_off_posix(tmp_path, monkeypatch): + output_dir = tmp_path / "run" + output_dir.mkdir() + revisions_dir = output_dir / "revisions" / "worker-results" + real_open = aboyeur.os.open + + def reject_directory_open(path, flags, mode=0o777, *, dir_fd=None): + if Path(path) == revisions_dir: + pytest.fail("non-POSIX platforms must not open directories for fsync") + return real_open(path, flags, mode, dir_fd=dir_fd) + + monkeypatch.setattr(aboyeur, "_supports_directory_fsync", lambda: False) + monkeypatch.setattr(aboyeur.os, "open", reject_directory_open) + monkeypatch.setattr(aboyeur.localio, "write_text_atomic", lambda *_args: None) + + aboyeur.write_sidecar_revision(output_dir, "worker-results.json", {"ok": True}) + + +def test_write_sidecar_revision_removes_partial_revision_when_fsync_fails(tmp_path, monkeypatch): + output_dir = tmp_path / "run" + output_dir.mkdir() + projection_path = output_dir / "worker-results.json" + old_projection = b'{ "status": "old", "items" : [] }\n' + projection_path.write_bytes(old_projection) + revisions_dir = output_dir / "revisions" / "worker-results" + revisions_dir.mkdir(parents=True) + earlier_revision = revisions_dir / "000001.json" + earlier_revision.write_bytes(old_projection) + earlier_bytes = earlier_revision.read_bytes() + partial_revision = revisions_dir / "000002.json" + + def fail_revision_fsync(_fd: int) -> None: + assert partial_revision.read_bytes() + raise OSError("revision fsync failed") + + def fail_projection_write(*_args, **_kwargs) -> None: + pytest.fail("projection write must not follow a failed revision fsync") + + monkeypatch.setattr(aboyeur.os, "fsync", fail_revision_fsync) + monkeypatch.setattr(aboyeur.localio, "write_text_atomic", fail_projection_write) + + with pytest.raises(OSError, match="revision fsync failed"): + aboyeur.write_sidecar_revision(output_dir, "worker-results.json", {"status": "new"}) + + assert projection_path.read_bytes() == old_projection + assert [path.name for path in _sidecar_revisions(output_dir, "worker-results")] == ["000001.json"] + assert earlier_revision.read_bytes() == earlier_bytes def test_run_marks_suspected_noop_for_ok_write_worker_with_no_non_brigade_changes(monkeypatch, tmp_path): diff --git a/tests/test_run_resume.py b/tests/test_run_resume.py index 6e7d9cb2..71f6c749 100644 --- a/tests/test_run_resume.py +++ b/tests/test_run_resume.py @@ -45,6 +45,15 @@ def _write_run_dir(tmp_path: Path, *, results: list[dict]) -> Path: json.dumps({"assignments": [{"stage": 1, "worker": "cook", "task": "write code"}]}) ) (run_dir / "worker-results.json").write_text(json.dumps({"results": results, "ground_truth": {}})) + (run_dir / "synthesis.json").write_text( + json.dumps( + { + "orchestrator": "chef", + "result": {"ok": False, "text": "stale synthesis"}, + "ground_truth": {}, + } + ) + ) return run_dir @@ -163,6 +172,8 @@ def test_resume_reattaches_and_resynthesizes(tmp_path, monkeypatch, capsys): }, ], ) + legacy_worker_results = json.loads((run_dir / "worker-results.json").read_text()) + legacy_synthesis = json.loads((run_dir / "synthesis.json").read_text()) recovered = json.loads((run_dir / "run.json").read_text()) recovered.update( { @@ -194,6 +205,14 @@ def test_resume_reattaches_and_resynthesizes(tmp_path, monkeypatch, capsys): assert run_json["recovery_history"] == [recovered["failure"]] assert "failure_phase" not in run_json assert "failure" not in run_json + worker_revisions = sorted((run_dir / "revisions" / "worker-results").glob("*.json")) + synthesis_revisions = sorted((run_dir / "revisions" / "synthesis").glob("*.json")) + assert [path.name for path in worker_revisions] == ["000001.json", "000002.json"] + assert [path.name for path in synthesis_revisions] == ["000001.json", "000002.json"] + assert json.loads(worker_revisions[0].read_text()) == legacy_worker_results + assert json.loads(synthesis_revisions[0].read_text()) == legacy_synthesis + assert json.loads(worker_revisions[1].read_text()) == json.loads((run_dir / "worker-results.json").read_text()) + assert json.loads(synthesis_revisions[1].read_text()) == json.loads((run_dir / "synthesis.json").read_text()) def test_resume_with_nothing_resumable_reports_and_exits_2(tmp_path, capsys): diff --git a/tests/test_runs_cmd.py b/tests/test_runs_cmd.py index 11376e20..d4c95bb7 100644 --- a/tests/test_runs_cmd.py +++ b/tests/test_runs_cmd.py @@ -184,6 +184,7 @@ def test_runs_show_prints_summary(tmp_path, capsys): assert " [ok] chef" in out assert "final:" in out assert " final answer" in out + assert not (run_dir / "revisions").exists() def test_runs_show_reports_missing_run_json(tmp_path, capsys):