diff --git a/chat/runtime.py b/chat/runtime.py index fa6a1444..c3e5117e 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -829,7 +829,6 @@ class ChatRuntime: recognizer = derive_recognizer(tuple(self._pending_recognizer_examples)) self._recognizer_registry.register(recognizer) self._pending_recognizer_examples.clear() - store.save_recognizers(self._recognizer_registry.all()) candidates_to_save = self._pending_candidates if self.config.auto_contemplate and candidates_to_save: from teaching.contemplation import contemplate @@ -838,22 +837,28 @@ class ChatRuntime: contemplate(c, vault_probe=vault_probe) for c in candidates_to_save ] - store.save_discovery_candidates(candidates_to_save) + # ADR-0219 — generation-dir atomic checkpoint. All files are written + # into a fresh gen-NNNN/ directory; the commit is the atomic pointer + # swap in commit_generation. A kill before the swap leaves the prior + # committed generation intact. The manifest is the last file written + # into the gen dir (audit convention); the pointer swap is the true + # commit boundary. + gen_num, gen_dir = store.begin_generation() + gen_store = EngineStateStore(gen_dir) + gen_store.save_recognizers(self._recognizer_registry.all()) + gen_store.save_discovery_candidates(candidates_to_save) # Shape B+ (schema v2): persist the lived session state (field, vault, - # anchor, graph, referents, dialogue) BEFORE the manifest, so the - # manifest stays the last durable act — the commit marker for the turn. - # Opt-in (config.persist_session_state): a deliberate resume mode, off by - # default so one-shot runtimes don't pay the per-turn snapshot cost. + # anchor, graph, referents, dialogue) BEFORE the manifest. + # Opt-in (config.persist_session_state): off by default. if self._context is not None and self.config.persist_session_state: - store.save_session_state(self._context.snapshot()) - # L11 — stamp the engine's identity and its lineage parent (the identity - # of the prior checkpoint). Same substrate -> identity == parent (a stable - # life); a ratified substrate change -> identity != parent (the bump). - store.save_manifest( + gen_store.save_session_state(self._context.snapshot()) + # L11 — stamp the engine's identity and its lineage parent. + gen_store.save_manifest( self._turn_count, engine_identity=self._engine_identity, parent_engine_identity=self._loaded_engine_identity, ) + store.commit_generation(gen_num) self._loaded_engine_identity = self._engine_identity def _count_pending_proposals(self) -> int: @@ -1027,10 +1032,13 @@ class ChatRuntime: ``compute_trace_hash`` produces the turn's canonical SHA-256. Stamps the trace_hash onto the most recent TurnEvent and any DiscoveryCandidate emitted during this - turn (i.e., the unstamped tail of ``_pending_candidates``), - then re-persists the candidates checkpoint so the on-disk - audit trail names the originating turn instead of the - prior empty-string default. + turn (i.e., the unstamped tail of ``_pending_candidates``). + + The stamped candidates are held in memory and persisted at the + next ``checkpoint_engine_state`` call (ADR-0219: only + ``checkpoint_engine_state`` writes to the checkpoint dir, so the + trace-hash back-stamp is always part of a coherent committed + generation rather than a mid-turn partial write). No-op when ``trace_hash`` is empty (pre-pipeline call sites, refusal stub path). Idempotent: stamping a tail whose @@ -1043,7 +1051,6 @@ class ChatRuntime: last_event = self.turn_log[-1] if not last_event.trace_hash: self.turn_log[-1] = replace(last_event, trace_hash=trace_hash) - stamped = False for idx in range(len(self._pending_candidates) - 1, -1, -1): cand = self._pending_candidates[idx] if cand.source_turn_trace: @@ -1051,11 +1058,6 @@ class ChatRuntime: self._pending_candidates[idx] = replace( cand, source_turn_trace=trace_hash ) - stamped = True - if stamped and self._engine_state_store is not None: - self._engine_state_store.save_discovery_candidates( - self._pending_candidates - ) def first_admitted_recognizer(self): if not self.config.recognition_grounded_graph: diff --git a/docs/decisions/ADR-0219-generation-checkpoint-atomicity.md b/docs/decisions/ADR-0219-generation-checkpoint-atomicity.md new file mode 100644 index 00000000..1b82eaa1 --- /dev/null +++ b/docs/decisions/ADR-0219-generation-checkpoint-atomicity.md @@ -0,0 +1,140 @@ +# ADR-0219 — Generation-dir atomic checkpoint (L10 continuity hardening) + +Status: accepted +Date: 2026-06-15 +Extends: ADR-0146 (engine-state persistence), ADR-0156 (per-file atomic writes) + +## Context + +ADR-0156 (W-022) introduced `_atomic_write_text` to guarantee **per-file** +atomicity: each of the four checkpoint files (`recognizers.jsonl`, +`discovery_candidates.jsonl`, `session_state.json`, `manifest.json`) is +written via `temp + os.replace`, so a kill mid-write leaves the prior file +intact. + +ADR-0156 §"Out of scope" explicitly excluded **cross-file atomicity**: + +> "There is no cross-file consistency guarantee: recognizers and candidates +> are written in separate atomic ops, so a kill between them leaves the +> store in a mixed state (e.g., recognizers@N over manifest@N-1)." + +The flat layout is therefore not a committed state — it is four +independently-atomic files with no shared commit boundary. L10's telos +("one continuous life") requires that a reboot always resumes from a +**coherent, complete** checkpoint, not a mixed-generation one. + +A second deficiency: `finalize_turn_trace_hash` (ADR-0153) called +`save_discovery_candidates` **outside** the main checkpoint sequence, +creating a second write path into the checkpoint directory without going +through the generation commit. + +ADR-0156 also deferred the **parent-directory fsync** after `os.replace`. +POSIX strictly requires fsyncing the directory to make the rename metadata +durable. This ADR closes that deferral. + +## Decision + +### Generation-dir model + +The checkpoint becomes a **committed generation**: a complete, fsync-ed +`gen-NNNN/` directory pointed to by a single `current` file. The atomic +replacement of `current` is the only commit boundary. + +``` +engine_state/ + gen-0041/ + recognizers.jsonl + discovery_candidates.jsonl + session_state.json + manifest.json + current ← one line: "gen-0041" +``` + +Two-phase commit protocol: +1. **`begin_generation()`** — allocate `gen-NNNN/`, return `(gen_num, gen_dir)`. +2. Write all checkpoint files into `gen_dir` via `EngineStateStore(gen_dir).save_*`. +3. **`commit_generation(gen_num)`** — fsync gen dir → `os.replace(tmp_current, current)` → fsync parent → GC old gens. + +A SIGKILL before the `os.replace` in step 3 leaves the prior `current` +intact. A SIGKILL after commits the new generation. Incomplete `gen-NNNN/` +directories with no `current` entry are garbage, ignored by all load paths. + +### `finalize_turn_trace_hash` write path closed + +`finalize_turn_trace_hash` no longer calls `save_discovery_candidates`. +The in-memory candidates carry the updated `source_turn_trace`; they are +persisted atomically as part of the next `checkpoint_engine_state` call. +(If the process dies between back-stamp and next checkpoint, the trace hash +is lost from the candidate but the candidate itself survives in the prior +committed generation — acceptable: the trace hash is audit metadata, not +state-determining.) + +### Legacy flat layout — explicit migration + +On the first `begin_generation()` call against a flat-layout store (pre-0219 +`manifest.json` at root, no `current`): +1. Flat files are copied into `gen-0000/` (fsync-ed). +2. `current` is written pointing to `gen-0000`. +3. The next generation (`gen-0001`) is allocated and returned. + +Load methods fall back to the flat root when no `current` exists, so a v1 +or v2 checkpoint is readable without a write (the migration only happens at +write time). + +### Parent-dir fsync + +`commit_generation` fsyncs both the generation directory (content durability) +and the parent directory after the `os.replace` (pointer rename metadata +durability), closing the ADR-0156 deferral. + +### GC + +`commit_generation` retains the last K=2 committed generations (the current +and its predecessor) and prunes older ones. K≥2 ensures the predecessor is +available if the committed gen dir is somehow corrupted after commit. +Pruned generation names are logged at DEBUG level. + +## Implementation + +- `engine_state/__init__.py`: `begin_generation`, `commit_generation`, + `_current_gen_dir`, `_resolve_dir`, `_gc_old_generations`, `_fsync_dir`. + All `load_*` methods updated to use `_resolve_dir()`. `exists()` checks + both layouts. `save_*` methods unchanged (write relative to `self.path`, + so they naturally write into a gen dir when instantiated as + `EngineStateStore(gen_dir)`). +- `chat/runtime.py`: `checkpoint_engine_state` uses the two-phase commit. + `finalize_turn_trace_hash` no longer calls `save_discovery_candidates`. +- `evals/l10_continuity/runner.py`: `_inject_orphan_tmp` updated to inject + both orphan shapes (unreferenced gen dir + torn `current` temp file). + +## Acceptance gate + +All of the following must hold (proven by `tests/test_adr_0219_generation_checkpoint.py`): +- Crash before pointer swap restores the prior committed generation. +- Crash after pointer swap restores the new generation. +- Incomplete gen dirs (not pointed to by `current`) are ignored on load. +- No load path mixes files across generations. +- Legacy flat layout migrates explicitly on first write; reads cleanly without migration. +- No normalization or repair on restore. +- `versor_condition < 1e-6` throughout (proven by L10 lane). + +## Invariants pinned by tests + +`tests/test_adr_0219_generation_checkpoint.py`: +- `test_fresh_store_writes_gen0000_and_current` — first checkpoint creates gen-0000 + current. +- `test_second_checkpoint_advances_generation` — gen-0001 replaces gen-0000. +- `test_orphan_gen_dir_ignored_before_pointer_swap` — unreferenced gen-9999 is invisible. +- `test_torn_current_tmp_ignored` — `.current.*.tmp` orphan does not affect load. +- `test_no_cross_generation_mixing` — load always reads from a single gen dir. +- `test_legacy_flat_layout_migrates_on_first_write` — flat state → gen-0000 on write. +- `test_legacy_flat_layout_readable_without_write` — load falls back to flat root. +- `test_commit_point_matches_turn_count` — recovered turn_count == committed turns. +- `test_gc_retains_last_two_generations` — GC prunes only generations outside the window. + +## Related + +- ADR-0146 — engine-state persistence foundation +- ADR-0156 — per-file atomic writes (extended here to cross-file) +- ADR-0157 — revision-mismatch warning on load (unaffected) +- ADR-0158 — reboot event audit (unaffected) +- L10 continuity hardening brief pack: `docs/handoff/l10-continuity-hardening-briefs-2026-06-15.md` diff --git a/engine_state/__init__.py b/engine_state/__init__.py index 86ed33ff..b7e41d4b 100644 --- a/engine_state/__init__.py +++ b/engine_state/__init__.py @@ -1,19 +1,37 @@ -"""Shape B engine-state persistence (ADR-0146). +"""Generation-dir checkpoint persistence (ADR-0146 / ADR-0156 / ADR-0219). engine_state/ is the mutable checkpoint directory for per-session engine -state that must survive reboot. It is NOT append-only (unlike substrate- -state); each checkpoint overwrites the previous. +state that must survive reboot. Each checkpoint is now an atomic committed +**generation**: a complete, fsync-ed ``gen-NNNN/`` directory pointed to by a +single ``current`` file whose atomic replacement is the commit boundary. -Layout: - engine_state/recognizers.jsonl -- one DerivedRecognizer per line - engine_state/discovery_candidates.jsonl -- one DiscoveryCandidate per line - engine_state/manifest.json -- schema_version, git revision, turn_count +Layout (ADR-0219 generation-dir model): + engine_state/ + gen-0041/ + recognizers.jsonl -- one DerivedRecognizer per line + discovery_candidates.jsonl -- one DiscoveryCandidate per line + session_state.json -- bit-exact field/vault/anchor/graph snapshot + manifest.json -- schema_version, git revision, turn_count + current -- one line: "gen-0041"; pointer swap = commit + +Commit is the single atomic ``os.replace`` of ``current``. A kill before the +swap leaves the prior ``current`` intact (the prior generation is the committed +state). A kill after the swap commits the new generation. Incomplete +``gen-NNNN/`` directories without a ``current`` entry are garbage, ignored. + +**Legacy (flat) layout — read-only fallback and migration:** +A pre-0219 flat ``engine_state/`` (manifest.json at root, no ``current``) +is read transparently by all ``load_*`` methods. On the first +``begin_generation()`` call, flat files are copied into ``gen-0000/`` and +``current`` is written; subsequent checkpoints use the generation model. """ from __future__ import annotations import json +import logging import os +import shutil import stat import subprocess import tempfile @@ -25,6 +43,18 @@ from typing import Sequence from recognition.anti_unifier import DerivedRecognizer from teaching.discovery import DiscoveryCandidate +_logger = logging.getLogger(__name__) + +_SCHEMA_VERSION = 2 # v2 adds session_state.json (Shape B+ lived-state persistence) +_GEN_PREFIX = "gen-" +_CURRENT_FILE = "current" + +_DEFAULT_DIR = ( + Path(os.environ["CORE_ENGINE_STATE_DIR"]) + if os.environ.get("CORE_ENGINE_STATE_DIR") + else Path(__file__).parents[1] / "engine_state" +) + def _atomic_write_text(target: Path, content: str, *, encoding: str = "utf-8") -> None: """ADR-0156 (W-022) — atomic checkpoint write. @@ -71,12 +101,20 @@ def _atomic_write_text(target: Path, content: str, *, encoding: str = "utf-8") - pass raise -_SCHEMA_VERSION = 2 # v2 adds session_state.json (Shape B+ lived-state persistence) -_DEFAULT_DIR = ( - Path(os.environ["CORE_ENGINE_STATE_DIR"]) - if os.environ.get("CORE_ENGINE_STATE_DIR") - else Path(__file__).parents[1] / "engine_state" -) + +def _fsync_dir(dir_path: Path) -> None: + """Fsync a directory file descriptor. + + Required for POSIX crash-safety: fsyncing directory metadata makes + rename/create operations inside it durable. Called after all files in a + generation directory are written (before pointer swap) and after the + pointer swap itself (to make the new ``current`` durable). + """ + fd = os.open(str(dir_path), os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) @lru_cache(maxsize=1) @@ -115,9 +153,158 @@ class IncompatibleEngineStateError(RuntimeError): class EngineStateStore: + """Manages the engine-state checkpoint directory. + + Two layouts are supported: + + - **Generation-dir** (ADR-0219): ``gen-NNNN/`` dirs pointed to by + ``current``. ``begin_generation`` + ``commit_generation`` form the + two-phase commit. This is the active layout for all new writes. + + - **Flat** (legacy pre-0219): files at the store root. ``load_*`` + methods fall back to this when no ``current`` file exists, so old + checkpoints are readable without migration. The first + ``begin_generation`` call migrates flat state into ``gen-0000/`` and + writes ``current``. + + The ``save_*`` / ``load_*`` methods write/read relative to ``self.path`` + directly. In the generation-dir model, callers pass a temporary + ``EngineStateStore(gen_dir)`` instance so ``save_*`` naturally writes + into the pending generation directory. Load methods resolve the active + generation via ``_resolve_dir()`` (or ``self.path`` for flat layout). + """ + def __init__(self, path: Path | None = None) -> None: self.path = path or _DEFAULT_DIR + # ------------------------------------------------------------------ + # Generation-dir protocol (ADR-0219) + # ------------------------------------------------------------------ + + def _current_gen_dir(self) -> Path | None: + """Return the currently committed generation directory, or None. + + Returns None when no ``current`` file exists (flat layout or fresh + store) or when the pointed-to directory does not exist. + """ + current_file = self.path / _CURRENT_FILE + if not current_file.exists(): + return None + gen_name = current_file.read_text(encoding="utf-8").strip() + gen_dir = self.path / gen_name + return gen_dir if gen_dir.is_dir() else None + + def _resolve_dir(self) -> Path: + """Return the directory to read checkpoint files from. + + Generation-dir layout: the committed gen dir (resolved via ``current``). + Flat / fresh layout: ``self.path`` directly. + """ + gen_dir = self._current_gen_dir() + return gen_dir if gen_dir is not None else self.path + + def begin_generation(self) -> tuple[int, Path]: + """Allocate the next pending generation directory and return (gen_num, gen_dir). + + If a flat-layout checkpoint exists at the store root (pre-0219 migration), + it is atomically wrapped into ``gen-0000/`` and ``current`` is written + before the new generation is allocated. This migration happens at most + once per store. + + The returned ``gen_dir`` is an empty (or freshly created) directory. + Write all checkpoint files into it, then call ``commit_generation``. + An exception before ``commit_generation`` leaves the prior committed + generation intact — the incomplete ``gen-NNNN/`` is treated as garbage + on the next load (no ``current`` update occurred). + """ + current_file = self.path / _CURRENT_FILE + if not current_file.exists(): + flat_manifest = self.path / "manifest.json" + if flat_manifest.exists(): + # Migrate flat layout into gen-0000 before proceeding. + gen0 = self.path / f"{_GEN_PREFIX}0000" + gen0.mkdir(parents=True, exist_ok=True) + for fname in ( + "manifest.json", + "recognizers.jsonl", + "discovery_candidates.jsonl", + "session_state.json", + ): + src = self.path / fname + if src.exists(): + shutil.copy2(src, gen0 / fname) + _fsync_dir(gen0) + _atomic_write_text(current_file, f"{_GEN_PREFIX}0000") + _fsync_dir(self.path) + _logger.info("engine_state: migrated flat layout to gen-0000") + next_num = 1 + else: + # Fresh store — first generation. + next_num = 0 + else: + current_name = current_file.read_text(encoding="utf-8").strip() + current_num = int(current_name[len(_GEN_PREFIX):]) + next_num = current_num + 1 + + gen_name = f"{_GEN_PREFIX}{next_num:04d}" + gen_dir = self.path / gen_name + gen_dir.mkdir(parents=True, exist_ok=True) + return next_num, gen_dir + + def commit_generation(self, gen_num: int, *, keep: int = 2) -> None: + """Atomically commit a completed generation as the new checkpoint. + + Sequence (all steps needed for POSIX crash-safety): + 1. Fsync the generation directory (content durability). + 2. Atomic ``os.replace`` of ``current`` pointer (the commit). + 3. Fsync the parent directory (pointer-rename metadata durability). + 4. GC old generations, retaining the last ``keep`` committed ones. + + A SIGKILL before step 2 leaves the prior ``current`` intact. + A SIGKILL between steps 2 and 3 may lose the pointer on a hard + crash, but ``os.replace`` is atomic at the kernel level so the + pointer is either the old or the new value — never torn. + """ + gen_name = f"{_GEN_PREFIX}{gen_num:04d}" + gen_dir = self.path / gen_name + _fsync_dir(gen_dir) + _atomic_write_text(self.path / _CURRENT_FILE, gen_name) + _fsync_dir(self.path) + pruned = self._gc_old_generations(committed_num=gen_num, keep=keep) + if pruned: + _logger.debug("engine_state GC: pruned old generations %s", pruned) + + def _gc_old_generations(self, *, committed_num: int, keep: int = 2) -> list[str]: + """Prune generation directories older than the retention window. + + Retains the ``keep`` most-recent committed generations (including + the just-committed one). Unreferenced gen dirs from an in-progress + write (gen_num > committed_num) are left alone — they are garbage + by definition but harmless, and pruning them here would require + knowing which one is in-progress, which we do not. + """ + cutoff = committed_num - keep + 1 # prune gen_num < cutoff + pruned: list[str] = [] + try: + for item in self.path.iterdir(): + if not item.is_dir() or not item.name.startswith(_GEN_PREFIX): + continue + try: + num = int(item.name[len(_GEN_PREFIX):]) + except ValueError: + continue + if num < cutoff: + shutil.rmtree(item, ignore_errors=True) + pruned.append(item.name) + except OSError: + pass # best-effort; GC failure never blocks a checkpoint + return sorted(pruned) + + # ------------------------------------------------------------------ + # Save methods — write to self.path (used both for flat layout and, + # via EngineStateStore(gen_dir), to write into a pending generation). + # ------------------------------------------------------------------ + def save_recognizers(self, recognizers: Sequence[DerivedRecognizer]) -> None: lines = [r.to_json() for r in recognizers] _atomic_write_text( @@ -125,16 +312,6 @@ class EngineStateStore: "\n".join(lines) + ("\n" if lines else ""), ) - def load_recognizers(self) -> list[DerivedRecognizer]: - p = self.path / "recognizers.jsonl" - if not p.exists(): - return [] - return [ - DerivedRecognizer.from_json(line) - for line in p.read_text(encoding="utf-8").splitlines() - if line.strip() - ] - def save_discovery_candidates( self, candidates: Sequence[DiscoveryCandidate], @@ -148,16 +325,6 @@ class EngineStateStore: "\n".join(lines) + ("\n" if lines else ""), ) - def load_discovery_candidates(self) -> list[DiscoveryCandidate]: - p = self.path / "discovery_candidates.jsonl" - if not p.exists(): - return [] - return [ - DiscoveryCandidate.from_dict(json.loads(line)) - for line in p.read_text(encoding="utf-8").splitlines() - if line.strip() - ] - def save_manifest( self, turn_count: int, @@ -167,12 +334,15 @@ class EngineStateStore: ) -> None: """Write the checkpoint manifest. - ``engine_identity`` (L11) stamps the content-derived identity the engine - was running under when this checkpoint was written; ``parent_engine_ - identity`` links it to the identity of the prior checkpoint, forming a - git-like lineage chain (the parent differs only across a ratified - substrate change). Both are additive-optional — an empty string omits the - key, preserving the manifest bytes for pre-L11 callers. + In the generation-dir model, the manifest is the last file written + into the pending gen dir; ``commit_generation`` then swaps the + ``current`` pointer. The manifest is no longer the commit boundary — + the pointer swap is. + + ``engine_identity`` (L11) stamps the content-derived identity the + engine was running under when this checkpoint was written. + ``parent_engine_identity`` links it to the identity of the prior + checkpoint, forming a lineage chain. Both are additive-optional. """ manifest: dict = { "schema_version": _SCHEMA_VERSION, @@ -188,8 +358,45 @@ class EngineStateStore: json.dumps(manifest, sort_keys=True, indent=2), ) + def save_session_state(self, snapshot: dict) -> None: + """Persist the lived session state (Shape B+ / schema v2). + + ``snapshot`` is ``SessionContext.snapshot()`` — a bit-exact, JSON-safe + dict of the field, vault, anchor, graph, referents, and dialogue. + In the generation-dir model, written before ``save_manifest`` inside + the pending gen dir. + """ + _atomic_write_text( + self.path / "session_state.json", + json.dumps(snapshot, sort_keys=True, separators=(",", ":")), + ) + + # ------------------------------------------------------------------ + # Load methods — resolve via _resolve_dir() (gen dir or flat root). + # ------------------------------------------------------------------ + + def load_recognizers(self) -> list[DerivedRecognizer]: + p = self._resolve_dir() / "recognizers.jsonl" + if not p.exists(): + return [] + return [ + DerivedRecognizer.from_json(line) + for line in p.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + def load_discovery_candidates(self) -> list[DiscoveryCandidate]: + p = self._resolve_dir() / "discovery_candidates.jsonl" + if not p.exists(): + return [] + return [ + DiscoveryCandidate.from_dict(json.loads(line)) + for line in p.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + def load_manifest(self) -> dict | None: - p = self.path / "manifest.json" + p = self._resolve_dir() / "manifest.json" if not p.exists(): return None content = p.read_text(encoding="utf-8").strip() @@ -211,7 +418,11 @@ class EngineStateStore: # Never refuse to load; reboot is recovery, not control flow. stored_rev = manifest.get("written_at_revision", "unknown") current_rev = get_git_revision() - if stored_rev not in ("unknown", "") and current_rev not in ("unknown", "") and stored_rev != current_rev: + if ( + stored_rev not in ("unknown", "") + and current_rev not in ("unknown", "") + and stored_rev != current_rev + ): warnings.warn( f"engine_state checkpoint was written at revision {stored_rev!r} " f"but the current revision is {current_rev!r}. " @@ -222,27 +433,13 @@ class EngineStateStore: ) return manifest - def save_session_state(self, snapshot: dict) -> None: - """Persist the lived session state (Shape B+ / schema v2). - - ``snapshot`` is ``SessionContext.snapshot()`` — a bit-exact, JSON-safe - dict of the field, vault, anchor, graph, referents, and dialogue. Written - atomically (ADR-0156). Save this BEFORE ``save_manifest`` so the manifest - stays the last durable act of a checkpoint (the commit marker). - """ - _atomic_write_text( - self.path / "session_state.json", - json.dumps(snapshot, sort_keys=True, separators=(",", ":")), - ) - def load_session_state(self) -> dict | None: """Load the lived session state, or None when absent (v1 checkpoint). A v1 checkpoint has no ``session_state.json`` — returning None lets the - caller fall back to a fresh session (today's Shape B behavior), so old - checkpoints stay loadable. + caller fall back to a fresh session (the historical Shape B behavior). """ - p = self.path / "session_state.json" + p = self._resolve_dir() / "session_state.json" if not p.exists(): return None content = p.read_text(encoding="utf-8").strip() @@ -251,6 +448,11 @@ class EngineStateStore: return json.loads(content) def exists(self) -> bool: + """True if a committed checkpoint exists (generation-dir or flat layout).""" + # Generation-dir layout: current pointer exists and points to a gen dir. + if (self.path / _CURRENT_FILE).exists(): + return self._current_gen_dir() is not None + # Flat legacy layout: manifest at root. return (self.path / "manifest.json").exists() diff --git a/evals/l10_continuity/runner.py b/evals/l10_continuity/runner.py index 69e2923d..7f920de6 100644 --- a/evals/l10_continuity/runner.py +++ b/evals/l10_continuity/runner.py @@ -105,17 +105,33 @@ def _new_runtime(config: RuntimeConfig, engine_state_dir: Path) -> ChatRuntime: def _inject_orphan_tmp(engine_state_dir: Path) -> None: - """Simulate a kill mid-checkpoint-write: leave an orphan temp file. + """Simulate a kill mid-checkpoint-write under the generation-dir model (ADR-0219). - ADR-0156 writes ``content`` to a ``...tmp`` file, fsyncs, then - ``os.replace``s it into place. A SIGKILL between fsync and replace leaves - exactly such an orphan with the *real* target fully intact. The loader reads - only the canonical filenames, so the orphan must be ignored. We write a - deliberately-corrupt orphan to prove the loader never reads it. + Two orphan shapes, both harmless to a correct loader: + + 1. An unreferenced generation directory (kill before the ``current`` pointer + swap): ``gen-9999/`` exists with content, but ``current`` still names the + prior committed generation. The loader follows ``current``; the + unreferenced gen dir is invisible to it. + + 2. A torn ``current`` temp file (kill during the ``os.replace`` of + ``current``): ``.current.deadbeef.tmp`` exists, but ``os.replace`` is + atomic so ``current`` is either the old or the new value — never the temp. + The loader reads only the canonical ``current`` filename. + + Neither orphan is reachable from a consistent load path. """ engine_state_dir.mkdir(parents=True, exist_ok=True) - orphan = engine_state_dir / ".manifest.json.deadbeef.tmp" - orphan.write_text("{ this is a torn, half-written checkpoint <<<", encoding="utf-8") + # Orphan 1: unreferenced gen dir (kill before pointer swap) + orphan_gen = engine_state_dir / "gen-9999" + orphan_gen.mkdir(exist_ok=True) + (orphan_gen / "manifest.json").write_text( + '{ "TORN": true, "note": "this gen was never committed via current" }', + encoding="utf-8", + ) + # Orphan 2: torn current temp file (kill during pointer swap) + torn_current = engine_state_dir / ".current.deadbeef.tmp" + torn_current.write_text("gen-9999", encoding="utf-8") def read_recovered_turn_count(engine_state_dir: Path) -> int | None: diff --git a/tests/test_adr_0153_trace_hash_backstamp.py b/tests/test_adr_0153_trace_hash_backstamp.py index 4eb88707..f156d5ef 100644 --- a/tests/test_adr_0153_trace_hash_backstamp.py +++ b/tests/test_adr_0153_trace_hash_backstamp.py @@ -109,8 +109,11 @@ def test_pipeline_back_stamps_pending_discovery_candidates( def test_persisted_candidates_jsonl_carries_trace_hash( tmp_path: Path, ) -> None: - """The on-disk ``discovery_candidates.jsonl`` checkpoint reflects the - back-stamped trace_hash (not the empty default).""" + """The persisted discovery candidates carry the back-stamped trace_hash. + + Uses the store's public load interface (load_discovery_candidates) rather + than reading file paths directly, so the test is layout-agnostic. + """ subject = _pick_cold_cause_subject() state_path = tmp_path / "engine_state" runtime = ChatRuntime( @@ -121,25 +124,24 @@ def test_persisted_candidates_jsonl_carries_trace_hash( result = pipe.run(f"What causes {subject}?") runtime.checkpoint_engine_state() - candidates_path = state_path / "discovery_candidates.jsonl" - if not candidates_path.exists(): + from engine_state import EngineStateStore + loaded = EngineStateStore(state_path).load_discovery_candidates() + if not loaded: pytest.skip("checkpoint did not write candidates this turn") - lines = [ - json.loads(line) - for line in candidates_path.read_text("utf-8").splitlines() - if line.strip() - ] cold = [ - ln - for ln in lines - if ln["proposed_chain"]["subject"] == subject - and ln["proposed_chain"]["intent"] == "cause" + c + for c in loaded + if c.proposed_chain.get("subject") == subject + and c.proposed_chain.get("intent") == "cause" ] if not cold: pytest.skip("discovery did not fire under default config; fixture stale") - for ln in cold: - assert ln["source_turn_trace"] == result.trace_hash + for c in cold: + assert c.source_turn_trace == result.trace_hash, ( + f"persisted candidate source_turn_trace must equal the trace_hash; " + f"got {c.source_turn_trace!r} vs {result.trace_hash!r}" + ) def test_finalize_is_noop_for_empty_trace_hash(tmp_path: Path) -> None: diff --git a/tests/test_adr_0219_generation_checkpoint.py b/tests/test_adr_0219_generation_checkpoint.py new file mode 100644 index 00000000..b01b5621 --- /dev/null +++ b/tests/test_adr_0219_generation_checkpoint.py @@ -0,0 +1,348 @@ +"""ADR-0219 — Generation-dir atomic checkpoint. + +Each test covers one bullet of the acceptance gate and includes a ``*_bites`` +mutation variant (CLAUDE.md schema-as-proof discipline: a predicate that cannot +fail under the violation it nominally catches is decoration, not proof). +""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from engine_state import EngineStateStore + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _write_gen(store: EngineStateStore, turn_count: int) -> int: + """Write one generation checkpoint and return the gen_num.""" + gen_num, gen_dir = store.begin_generation() + gs = EngineStateStore(gen_dir) + gs.save_recognizers([]) + gs.save_discovery_candidates([]) + gs.save_manifest(turn_count) + store.commit_generation(gen_num) + return gen_num + + +# --------------------------------------------------------------------------- +# Gate: first checkpoint creates gen-0000 + current pointer +# --------------------------------------------------------------------------- + +def test_fresh_store_writes_gen0000_and_current(tmp_path: Path) -> None: + store = EngineStateStore(tmp_path) + gen_num = _write_gen(store, turn_count=1) + + assert gen_num == 0 + assert (tmp_path / "gen-0000").is_dir() + assert (tmp_path / "current").exists() + assert (tmp_path / "current").read_text("utf-8").strip() == "gen-0000" + assert (tmp_path / "gen-0000" / "manifest.json").exists() + + +def test_fresh_store_writes_gen0000_and_current_bites(tmp_path: Path) -> None: + """Without commit_generation the current pointer is absent.""" + store = EngineStateStore(tmp_path) + _gen_num, gen_dir = store.begin_generation() + EngineStateStore(gen_dir).save_manifest(1) + # No commit_generation — current pointer must not exist yet. + assert not (tmp_path / "current").exists(), ( + "current pointer must only be written by commit_generation, not begin_generation" + ) + + +# --------------------------------------------------------------------------- +# Gate: second checkpoint advances the generation +# --------------------------------------------------------------------------- + +def test_second_checkpoint_advances_generation(tmp_path: Path) -> None: + store = EngineStateStore(tmp_path) + _write_gen(store, turn_count=1) + gen_num2 = _write_gen(store, turn_count=2) + + assert gen_num2 == 1 + assert (tmp_path / "current").read_text("utf-8").strip() == "gen-0001" + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 2 + + +def test_second_checkpoint_advances_generation_bites(tmp_path: Path) -> None: + """Stale current (not advanced) → stale turn_count.""" + store = EngineStateStore(tmp_path) + _write_gen(store, turn_count=1) + # Deliberately skip the second commit and write gen-0001 without committing. + _gen_num2, gen_dir2 = store.begin_generation() + EngineStateStore(gen_dir2).save_manifest(2) + # current still points to gen-0000 (turn_count=1). + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 1, "uncommitted gen must not be visible via current" + + +# --------------------------------------------------------------------------- +# Gate: orphan gen dir ignored before pointer swap +# --------------------------------------------------------------------------- + +def test_orphan_gen_dir_ignored_before_pointer_swap(tmp_path: Path) -> None: + store = EngineStateStore(tmp_path) + _write_gen(store, turn_count=5) + + # Simulate a kill between writing gen-9999 and swapping current. + orphan = tmp_path / "gen-9999" + orphan.mkdir() + (orphan / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 9999, "written_at_revision": "x"}), + encoding="utf-8", + ) + + # Load must still read from the committed generation (gen-0000, turn_count=5). + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 5, ( + "unreferenced gen-9999 must be invisible; load must follow current" + ) + + +def test_orphan_gen_dir_ignored_before_pointer_swap_bites(tmp_path: Path) -> None: + """Without a current pointer the orphan gen dir IS what gets read (flat fallback).""" + # Build the store with ONLY the flat manifest (no current pointer). + (tmp_path / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 42, "written_at_revision": "x"}), + encoding="utf-8", + ) + manifest = EngineStateStore(tmp_path).load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 42, "flat layout fallback must work when no current exists" + + +# --------------------------------------------------------------------------- +# Gate: torn current temp file is ignored +# --------------------------------------------------------------------------- + +def test_torn_current_tmp_ignored(tmp_path: Path) -> None: + store = EngineStateStore(tmp_path) + _write_gen(store, turn_count=3) + + # Inject a torn .current temp file (simulate kill during os.replace of current). + torn = tmp_path / ".current.deadbeef.tmp" + torn.write_text("gen-9999", encoding="utf-8") + + # The loader reads "current" (canonical), not ".current.*.tmp". + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 3, ( + ".current.*.tmp orphan must be invisible to the loader" + ) + + +def test_torn_current_tmp_ignored_bites(tmp_path: Path) -> None: + """If current itself is overwritten with the orphan name, load diverges.""" + store = EngineStateStore(tmp_path) + _write_gen(store, turn_count=3) + # Corrupt current directly (simulates what we're proving the loader avoids). + (tmp_path / "current").write_text("gen-9999", encoding="utf-8") + # gen-9999 doesn't exist → _current_gen_dir() returns None → flat fallback. + # The flat manifest doesn't exist either → load_manifest() returns None. + assert store.load_manifest() is None, ( + "a current pointer to a non-existent gen dir must fall back to flat/None" + ) + + +# --------------------------------------------------------------------------- +# Gate: no cross-generation file mixing +# --------------------------------------------------------------------------- + +def test_no_cross_generation_mixing(tmp_path: Path) -> None: + """load_* always reads from a single generation directory.""" + store = EngineStateStore(tmp_path) + # Gen 0: turn_count=10 + _write_gen(store, turn_count=10) + # Gen 1: turn_count=20 + _write_gen(store, turn_count=20) + + # current → gen-0001. manifest must say turn_count=20. + # recognizers and candidates must also come from gen-0001 (not gen-0000). + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 20 + + recs = store.load_recognizers() + cands = store.load_discovery_candidates() + # Both loads resolve to gen-0001; not a mix of gen-0000 and gen-0001. + assert recs == [] + assert cands == [] + + +def test_no_cross_generation_mixing_bites(tmp_path: Path) -> None: + """Forcing current to gen-0000 after two writes returns gen-0000 data.""" + store = EngineStateStore(tmp_path) + # Gen 0 with turn_count=10 + gen0_num, gen0_dir = store.begin_generation() + EngineStateStore(gen0_dir).save_manifest(10) + store.commit_generation(gen0_num) + # Gen 1 with turn_count=20 + gen1_num, gen1_dir = store.begin_generation() + EngineStateStore(gen1_dir).save_manifest(20) + store.commit_generation(gen1_num) + + # Manually rewind current to gen-0000. + (tmp_path / "current").write_text("gen-0000", encoding="utf-8") + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 10, ( + "rewinding current to gen-0000 must expose gen-0000 data, not gen-0001" + ) + + +# --------------------------------------------------------------------------- +# Gate: legacy flat layout migrates on first write +# --------------------------------------------------------------------------- + +def test_legacy_flat_layout_migrates_on_first_write(tmp_path: Path) -> None: + """A flat-layout checkpoint is wrapped into gen-0000 on the first begin_generation.""" + # Seed a flat layout (pre-0219). + (tmp_path / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 7, "written_at_revision": "old"}), + encoding="utf-8", + ) + (tmp_path / "recognizers.jsonl").write_text("", encoding="utf-8") + + store = EngineStateStore(tmp_path) + gen_num, _gen_dir = store.begin_generation() + + assert gen_num == 1, "first gen after migration must be gen-0001 (gen-0000 = migrated flat)" + assert (tmp_path / "gen-0000" / "manifest.json").exists() + assert (tmp_path / "current").read_text("utf-8").strip() == "gen-0000" + + # Commit gen-0001 with new turn_count. + EngineStateStore(_gen_dir).save_manifest(8) + store.commit_generation(gen_num) + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 8, "committed gen-0001 must be visible after migration" + + +def test_legacy_flat_layout_migrates_on_first_write_bites(tmp_path: Path) -> None: + """Without migration, a second gen would incorrectly become gen-0000.""" + # Seed a flat layout but do NOT call begin_generation (no migration). + (tmp_path / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 7, "written_at_revision": "old"}), + encoding="utf-8", + ) + # Directly create gen-0000 manually to simulate what migration does. + gen0 = tmp_path / "gen-0000" + gen0.mkdir() + (gen0 / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 7, "written_at_revision": "old"}), + encoding="utf-8", + ) + # No current pointer yet. + assert not (tmp_path / "current").exists() + # begin_generation sees no current AND no flat manifest (we deleted it conceptually by + # just not having one) → would allocate gen-0000 again. Since we created gen-0000 + # manually, begin_generation will return (0, gen-0000 dir) and overwrite it. + # This is the hazard that the migration path avoids. + # The test just confirms our understanding: without migration the gen num is 0. + (tmp_path / "manifest.json").unlink() # remove flat file so no-migration path taken + store = EngineStateStore(tmp_path) + gen_num, _ = store.begin_generation() + assert gen_num == 0, "without migration and without flat manifest, gen_num starts at 0" + + +# --------------------------------------------------------------------------- +# Gate: legacy flat layout readable without write +# --------------------------------------------------------------------------- + +def test_legacy_flat_layout_readable_without_write(tmp_path: Path) -> None: + """load_manifest falls back to the flat root when no current pointer exists.""" + (tmp_path / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 99, "written_at_revision": "x"}), + encoding="utf-8", + ) + store = EngineStateStore(tmp_path) + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 99 + + +def test_legacy_flat_layout_readable_without_write_bites(tmp_path: Path) -> None: + """Moving the flat manifest to a gen dir without writing current → None.""" + gen0 = tmp_path / "gen-0000" + gen0.mkdir() + (gen0 / "manifest.json").write_text( + json.dumps({"schema_version": 2, "turn_count": 99, "written_at_revision": "x"}), + encoding="utf-8", + ) + # No flat manifest at root, no current → load_manifest returns None. + assert EngineStateStore(tmp_path).load_manifest() is None, ( + "a gen dir with no current pointer must not be read by load_manifest" + ) + + +# --------------------------------------------------------------------------- +# Gate: commit-point / turn_count matches committed turns +# --------------------------------------------------------------------------- + +def test_commit_point_matches_turn_count(tmp_path: Path) -> None: + """Recovered turn_count always equals the number of fully committed turns.""" + store = EngineStateStore(tmp_path) + for t in range(1, 4): + _write_gen(store, turn_count=t) + + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 3 + + +def test_commit_point_matches_turn_count_bites(tmp_path: Path) -> None: + """A mid-write kill (no commit) does not advance the recovered turn_count.""" + store = EngineStateStore(tmp_path) + _write_gen(store, turn_count=1) + + # Simulate a kill mid-write of gen-0001 (no commit_generation). + _gen_num2, gen_dir2 = store.begin_generation() + EngineStateStore(gen_dir2).save_manifest(2) + # No commit — current still names gen-0000. + + manifest = store.load_manifest() + assert manifest is not None + assert manifest["turn_count"] == 1, ( + "an uncommitted generation must not advance the recovered turn_count" + ) + + +# --------------------------------------------------------------------------- +# Gate: GC retains the last two generations +# --------------------------------------------------------------------------- + +def test_gc_retains_last_two_generations(tmp_path: Path) -> None: + store = EngineStateStore(tmp_path) + for t in range(1, 5): + _write_gen(store, turn_count=t) + + # After 4 commits (gen-0000..gen-0003), GC should have pruned gen-0000 and gen-0001. + gen_dirs = sorted(d.name for d in tmp_path.iterdir() if d.is_dir() and d.name.startswith("gen-")) + assert "gen-0000" not in gen_dirs, "gen-0000 should be pruned after 4 commits (keep=2)" + assert "gen-0001" not in gen_dirs, "gen-0001 should be pruned after 4 commits (keep=2)" + assert "gen-0002" in gen_dirs, "gen-0002 (N-1) must be retained" + assert "gen-0003" in gen_dirs, "gen-0003 (N, current) must be retained" + + +def test_gc_retains_last_two_generations_bites(tmp_path: Path) -> None: + """With keep=3 the three most-recent generations are all retained.""" + store = EngineStateStore(tmp_path) + for t in range(1, 5): + gen_num, gen_dir = store.begin_generation() + EngineStateStore(gen_dir).save_manifest(t) + store.commit_generation(gen_num, keep=3) + + gen_dirs = sorted(d.name for d in tmp_path.iterdir() if d.is_dir() and d.name.startswith("gen-")) + assert "gen-0000" not in gen_dirs, "gen-0000 pruned even with keep=3 (4 gens total)" + assert "gen-0001" in gen_dirs, "gen-0001 retained with keep=3" + assert "gen-0002" in gen_dirs, "gen-0002 retained with keep=3" + assert "gen-0003" in gen_dirs, "gen-0003 retained with keep=3"