feat(W-022): atomic engine-state checkpoint writes (L10b.1, ADR-0156) (#280)

ADR-0146 specified write-temp+rename for the engine-state
checkpoint to prevent corruption on mid-write process termination.
The W-008 implementation used Path.write_text directly, which
truncates the target before writing — SIGINT/SIGKILL between
truncate and write left a partial / empty file, breaking reboot
recovery (or worse, silently restoring half-state).

- engine_state._atomic_write_text: NamedTemporaryFile in target dir,
  flush + fsync, os.replace (atomic same-FS rename), best-effort
  cleanup of temp on failure
- All three EngineStateStore.save_* methods route through the helper
- Content bytes unchanged → round-trip regression guard passes

Pinned by tests/test_adr_0156_atomic_checkpoint.py (9 tests):
atomic create / overwrite / parent-mkdir, failed-replace preserves
prior target, failed-replace cleans temp, temp lives in target dir
(same-FS atomicity requirement), store-level failure preservation,
round-trip content regression guard.

CLI lanes: smoke (67) + cognition (120+1 skip) green.

Out of scope (next L10b chunks): reboot_event audit entry (W-024),
revision-mismatch warning on load (W-023), parent-dir fsync, cross-
process locking.

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.

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:

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() == []