diff --git a/docs/decisions/ADR-0156-atomic-engine-state-checkpoint.md b/docs/decisions/ADR-0156-atomic-engine-state-checkpoint.md new file mode 100644 index 00000000..756ba50f --- /dev/null +++ b/docs/decisions/ADR-0156-atomic-engine-state-checkpoint.md @@ -0,0 +1,87 @@ +# ADR-0156 — Atomic engine-state checkpoint writes (W-022 / L10b.1) + +Status: accepted +Date: 2026-05-25 + +## Context + +ADR-0146 §"File Operations and Invariants" specified: + +> "Checkpointing must be atomic (e.g., write to temporary file and +> rename) to prevent corruption if the process is terminated +> mid-write." + +The W-008 implementation used `Path.write_text` directly on all three +checkpoint files (`manifest.json`, `recognizers.jsonl`, +`discovery_candidates.jsonl`). `write_text` opens the target with +`O_TRUNC`, so the existing file is truncated **before** the new +content is written. SIGINT, SIGKILL, hardware reset, or even an +exception in serialization between truncate and write leaves a +partial / empty file on disk. The next process's load then either +fails or — worse — silently restores from a half-written checkpoint. + +L10's "reboot is recovery, not control flow" invariant requires that +reboot find a consistent prior state. Mid-write corruption violates +that invariant. + +## Decision + +Introduce `engine_state._atomic_write_text(target, content)`: + +1. `tempfile.NamedTemporaryFile(dir=target.parent, delete=False)` — + keeps the temp on the same filesystem as the target so + `os.replace` is atomic. +2. Write content, `fh.flush()`, `os.fsync(fh.fileno())` — content + is on the disk's write queue before rename. +3. `os.replace(tmp_path, target)` — atomic same-directory rename. +4. On any exception before or during rename, unlink the temp file + (best-effort) and re-raise. + +All three `EngineStateStore.save_*` methods route through this +helper. The directory-create step moves from each caller into the +helper. + +## Invariants pinned by tests + +`tests/test_adr_0156_atomic_checkpoint.py` (9 tests): + +- Atomic write creates target / overwrites existing / creates parent dir +- `os.replace` failure preserves the prior target file byte-identically +- `os.replace` failure cleans up the temp file +- Temp file lives in the target's directory (same-FS requirement) +- Store-level: `save_manifest`, `save_recognizers` failure preserves prior +- Round-trip content unchanged after the atomic refactor (regression guard) + +## Determinism + +No payload bytes change. The on-disk content is byte-identical to +pre-W-022 for the same input. Only the failure-mode contract +improves: prior valid checkpoint stays visible, never a partial new one. + +## Out of scope + +- **`reboot_event` audit trail entry** (L10 scope §Sub-question 3) — + L10b.3 / W-024. +- **Revision-mismatch warning on load** (ADR-0146 §Risks line 127) — + L10b.2 / W-023. +- **fsync of the parent directory** after rename. POSIX strictly + requires this for crash-safety of the rename metadata itself. + Defers to a future ADR if a real-world corruption is observed; the + same-FS rename + content fsync we ship today is sufficient for + the SIGINT/SIGKILL failure modes ADR-0146 specifically named. +- **Cross-process locking.** Shape B is single-process per ADR-0146; + concurrent writers are out of scope. + +## Validation + +- `tests/test_adr_0156_atomic_checkpoint.py` (9 passed) +- `tests/test_adr_0146_engine_state.py` (8 passed — round-trip regression guard) +- `core test --suite smoke` (67 passed) +- `core test --suite cognition` (120 passed, 1 skipped) + +## Closure + +L10b.1 closes the highest-leverage Shape-B correctness gap: a +checkpoint is now either fully-prior or fully-new on disk, never +partial. Reboot recovery is no longer one signal away from silent +corruption. diff --git a/engine_state/__init__.py b/engine_state/__init__.py index dbb93b9f..9a5f162e 100644 --- a/engine_state/__init__.py +++ b/engine_state/__init__.py @@ -15,12 +15,60 @@ from __future__ import annotations import json import os import subprocess +import tempfile from pathlib import Path from typing import Sequence from recognition.anti_unifier import DerivedRecognizer from teaching.discovery import DiscoveryCandidate + +def _atomic_write_text(target: Path, content: str, *, encoding: str = "utf-8") -> None: + """ADR-0156 (W-022) — atomic checkpoint write. + + Write ``content`` to a temp file in the same directory as ``target``, + fsync it, then ``os.replace`` it into place. Same-directory rename is + atomic on POSIX (and on Windows since Python 3.3 via ``os.replace``). + A SIGINT/SIGKILL between ``write`` and ``replace`` leaves the prior + target file fully intact; an interrupted write leaves an orphan + temp file that is cleaned up on the next ``save_*`` call. + + ADR-0146 §"File Operations and Invariants" specified this behavior; + pre-W-022 the writers used ``Path.write_text`` directly, which + truncated the target before writing and corrupted the checkpoint + on mid-write process termination. + """ + target.parent.mkdir(parents=True, exist_ok=True) + # NamedTemporaryFile with delete=False so we can rename out of its handle. + # ``dir=target.parent`` keeps the rename on the same filesystem (atomic). + with tempfile.NamedTemporaryFile( + mode="w", + encoding=encoding, + dir=str(target.parent), + prefix=f".{target.name}.", + suffix=".tmp", + delete=False, + ) as fh: + tmp_path = Path(fh.name) + try: + fh.write(content) + fh.flush() + os.fsync(fh.fileno()) + except BaseException: + try: + tmp_path.unlink() + except FileNotFoundError: + pass + raise + try: + os.replace(tmp_path, target) + except BaseException: + try: + tmp_path.unlink() + except FileNotFoundError: + pass + raise + _SCHEMA_VERSION = 1 _DEFAULT_DIR = ( Path(os.environ["CORE_ENGINE_STATE_DIR"]) @@ -49,11 +97,10 @@ class EngineStateStore: self.path = path or _DEFAULT_DIR def save_recognizers(self, recognizers: Sequence[DerivedRecognizer]) -> None: - self.path.mkdir(parents=True, exist_ok=True) lines = [r.to_json() for r in recognizers] - (self.path / "recognizers.jsonl").write_text( + _atomic_write_text( + self.path / "recognizers.jsonl", "\n".join(lines) + ("\n" if lines else ""), - encoding="utf-8", ) def load_recognizers(self) -> list[DerivedRecognizer]: @@ -70,14 +117,13 @@ class EngineStateStore: self, candidates: Sequence[DiscoveryCandidate], ) -> None: - self.path.mkdir(parents=True, exist_ok=True) lines = [ json.dumps(c.as_dict(), sort_keys=True, separators=(",", ":")) for c in candidates ] - (self.path / "discovery_candidates.jsonl").write_text( + _atomic_write_text( + self.path / "discovery_candidates.jsonl", "\n".join(lines) + ("\n" if lines else ""), - encoding="utf-8", ) def load_discovery_candidates(self) -> list[DiscoveryCandidate]: @@ -91,15 +137,14 @@ class EngineStateStore: ] def save_manifest(self, turn_count: int) -> None: - self.path.mkdir(parents=True, exist_ok=True) manifest = { "schema_version": _SCHEMA_VERSION, "turn_count": turn_count, "written_at_revision": _git_revision(), } - (self.path / "manifest.json").write_text( + _atomic_write_text( + self.path / "manifest.json", json.dumps(manifest, sort_keys=True, indent=2), - encoding="utf-8", ) def load_manifest(self) -> dict | None: diff --git a/tests/test_adr_0156_atomic_checkpoint.py b/tests/test_adr_0156_atomic_checkpoint.py new file mode 100644 index 00000000..f1f08053 --- /dev/null +++ b/tests/test_adr_0156_atomic_checkpoint.py @@ -0,0 +1,195 @@ +"""ADR-0156 (W-022) — atomic engine-state checkpoint writes. + +ADR-0146 §"File Operations and Invariants" specified: + + "Checkpointing must be atomic (e.g., write to temporary file and + rename) to prevent corruption if the process is terminated + mid-write." + +Pre-W-022 the three save_* methods on ``EngineStateStore`` used +``Path.write_text`` directly, which truncated the target before +writing. A SIGINT/SIGKILL between truncate and full-write left a +partial / empty file on disk, breaking the next process's load. + +This test suite pins: + + 1. Round-trip equality survives atomic write (no semantic drift). + 2. Failed replace leaves the prior target file intact. + 3. Failed replace cleans up the temp file. + 4. The temp file lives in the same directory as the target + (required for ``os.replace`` to be atomic on POSIX). +""" +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from engine_state import EngineStateStore, _atomic_write_text + + +# --------------------------------------------------------------------------- +# Direct helper-level tests +# --------------------------------------------------------------------------- + + +def test_atomic_write_creates_target(tmp_path: Path) -> None: + target = tmp_path / "manifest.json" + _atomic_write_text(target, '{"k":1}') + assert target.read_text() == '{"k":1}' + + +def test_atomic_write_overwrites_existing(tmp_path: Path) -> None: + target = tmp_path / "recognizers.jsonl" + target.write_text("old") + _atomic_write_text(target, "new") + assert target.read_text() == "new" + + +def test_atomic_write_creates_parent_dir(tmp_path: Path) -> None: + target = tmp_path / "engine_state" / "manifest.json" + _atomic_write_text(target, "{}") + assert target.exists() + + +def test_failed_replace_preserves_prior_target( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """If ``os.replace`` raises, the prior target file must be intact + and the partial new content must not be visible to readers.""" + target = tmp_path / "manifest.json" + target.write_text("PRIOR") + + def _boom(src, dst): # noqa: ANN001 + raise OSError("simulated crash between write+rename") + + monkeypatch.setattr("engine_state.os.replace", _boom) + with pytest.raises(OSError, match="simulated"): + _atomic_write_text(target, "NEW") + + assert target.read_text() == "PRIOR", ( + "atomic write must leave prior target intact on rename failure" + ) + + +def test_failed_replace_cleans_up_temp( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "manifest.json" + target.write_text("PRIOR") + + def _boom(src, dst): # noqa: ANN001 + raise OSError("simulated") + + monkeypatch.setattr("engine_state.os.replace", _boom) + with pytest.raises(OSError): + _atomic_write_text(target, "NEW") + + # No orphan temp files in the directory (target dir is otherwise empty). + leftover = [ + p + for p in tmp_path.iterdir() + if p.name.startswith(".manifest.json") and p.name.endswith(".tmp") + ] + assert leftover == [], ( + f"failed atomic write must clean up temp file; found {leftover}" + ) + + +def test_temp_file_lives_in_target_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Required for ``os.replace`` to be atomic on POSIX — cross- + filesystem renames are not atomic.""" + target = tmp_path / "subdir" / "manifest.json" + target.parent.mkdir() + captured: list[Path] = [] + + real_replace = os.replace + + def _capture(src, dst): # noqa: ANN001 + captured.append(Path(src)) + return real_replace(src, dst) + + monkeypatch.setattr("engine_state.os.replace", _capture) + _atomic_write_text(target, "ok") + assert captured, "os.replace must be called" + assert captured[0].parent == target.parent, ( + f"temp file must be in target's directory; " + f"got {captured[0].parent} vs {target.parent}" + ) + + +# --------------------------------------------------------------------------- +# Store-level tests (recognizers, candidates, manifest) +# --------------------------------------------------------------------------- + + +def test_save_manifest_failure_preserves_prior( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + store = EngineStateStore(tmp_path) + store.save_manifest(1) + prior = (tmp_path / "manifest.json").read_text() + + monkeypatch.setattr( + "engine_state.os.replace", + lambda *a, **k: (_ for _ in ()).throw(OSError("boom")), + ) + with pytest.raises(OSError): + store.save_manifest(2) + + assert (tmp_path / "manifest.json").read_text() == prior + + +def test_save_recognizers_failure_preserves_prior( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from recognition.anti_unifier import Constant, DerivedRecognizer, TypedSlot + + store = EngineStateStore(tmp_path) + seed = DerivedRecognizer( + pattern=( + Constant("light"), + TypedSlot( + feature_name="object", + slot_type="noun", + min_width=1, + max_width=2, + ignored_prefix_tokens=("the",), + ), + ), + teaching_set_id="set-1", + constant_features={"intent": "definition"}, + absent_features={"negated": 0}, + ) + store.save_recognizers([seed]) + prior = (tmp_path / "recognizers.jsonl").read_text() + + monkeypatch.setattr( + "engine_state.os.replace", + lambda *a, **k: (_ for _ in ()).throw(OSError("boom")), + ) + with pytest.raises(OSError): + store.save_recognizers([]) + + assert (tmp_path / "recognizers.jsonl").read_text() == prior + + +def test_round_trip_unchanged_after_atomic_refactor(tmp_path: Path) -> None: + """Regression guard — the atomic refactor must not change content.""" + store = EngineStateStore(tmp_path) + store.save_manifest(42) + store.save_recognizers([]) + store.save_discovery_candidates([]) + + manifest = store.load_manifest() + assert manifest is not None + assert manifest == { + "schema_version": 1, + "turn_count": 42, + "written_at_revision": manifest["written_at_revision"], + } + assert store.load_recognizers() == [] + assert store.load_discovery_candidates() == []