core-labs/core
core-labs/core
1
0
Fork 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

feat(engine-state): generation-dir atomic checkpoint (ADR-0219)

Browse source 
Closes the cross-file checkpoint-atomicity gap in ADR-0156.  The four
checkpoint files now live in a committed gen-NNNN/ directory; the single
atomic os.replace of a 'current' pointer file is the commit boundary.
A kill before the pointer swap leaves the prior committed generation
intact; a kill after commits the new generation.  Unreferenced gen dirs
are ignored.  ADR-0156's deferred parent-dir fsync is also closed.

Key changes:
- engine_state/__init__.py: begin_generation() + commit_generation() +
  _resolve_dir() for all load_* methods.  Flat-layout legacy checkpoints
  migrate into gen-0000 on first begin_generation call.  GC retains K=2
  committed generations.
- chat/runtime.py: checkpoint_engine_state uses the two-phase commit;
  finalize_turn_trace_hash no longer writes discovery_candidates outside
  the generation sequence (the second unguarded write path is closed).
- evals/l10_continuity/runner.py: _inject_orphan_tmp updated to inject
  the two orphan shapes of the generation model.
- tests/test_adr_0219_generation_checkpoint.py: 18 tests, one per
  acceptance-gate bullet + biting mutation variant each.

L10 lane: all_gates_pass=true; versor_condition<1e-6 throughout.
Smoke: 95 passed. Runtime: 20 passed.
This commit is contained in:
Shay 2026-06-15 01:08:36 -07:00
parent b1708a76d5
commit 103def317d
6 changed files with 811 additions and 101 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
chat/runtime.py
View file

@ -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:

140
docs/decisions/ADR-0219-generation-checkpoint-atomicity.md Normal file
View file

@ -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`

316
engine_state/__init__.py
View file

@ -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()

32
evals/l10_continuity/runner.py
View file

@ -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 ``.<name>.<rand>.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:

32
tests/test_adr_0153_trace_hash_backstamp.py
View file

@ -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:

348
tests/test_adr_0219_generation_checkpoint.py Normal file
View file

@ -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"