From 23bc28caf90a6a23a628f400e967301a7da764e5 Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 5 Jun 2026 11:14:17 -0700 Subject: [PATCH] =?UTF-8?q?feat(evals):=20L10=20continuity=20spike=20?= =?UTF-8?q?=E2=80=94=20falsifiable=20long-horizon=20soak=20lane?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Build evals/l10_continuity/, the empirical gate between the two L10 targets (T-resume: provable same-life resume; T-experience: continuous experiencing field-life). Drives the REAL turn loop (ChatRuntime + CognitiveTurnPipeline) over a deterministic in-vocab corpus, with reboot and orphan-crash legs, and evaluates falsifiable predicates over recorded evidence. Additive only; no existing file touched; read-only over the runtime; no serving-path import. Predicates (each with a *_holds real-soak test AND a *_bites mutation test, per the CLAUDE.md schema-as-proof discipline): - P1 closure: versor_condition < 1e-6 every turn (green guard). - P2a determinism: two independent runtimes -> byte-identical trace_hash. - P2b reboot transparency (the diagnostic): a reboot never alters pre-reboot turns (hard guard); post-reboot transparency is MEASURED and today FALSE -- the mechanical proof that Shape B (ADR-0146) discards the lived field/vault, i.e. "many lives sharing a checkpoint". A pinned test flips if persistence is ever added, forcing a doc update so the gap can't close silently. - P3 bounded resources: vault grows linear-bounded/monotonic (RSS recorded). - P4 crash recovery: two recoveries from one checkpoint converge (determinism) + commit-point/ARIES force boundary (recovered turn_count == committed) + atomic-write survives mid-os.replace kill (ADR-0156). - P5b anchor stability (T-experience crux): field anchors without COLLAPSE (dist_to_anchor not -> 0) or FREEZE (turn_movement not -> 0); the long-horizon test of the sanctioned _session_anchor_pull (alpha=0.05). Thresholds measured. - P5c coherence: surfaces stay non-empty and not collapsed to one output, over more than one corpus cycle. - P5a recall precision@k: recorded as not_covered (needs a held-out probe set). report.py assembles the panel into a structured report with a hardware-stable deterministic_digest (trace_hash sequence + verdicts; excludes RSS/wall-clock) as the freeze handle. Run: python -m evals.l10_continuity [n_turns] [reboot_turn]. 24 tests pass; adversarially reviewed across 4 lenses (bite-discipline, invariant/trust-boundary, honesty/determinism, correctness) before landing. --- evals/l10_continuity/__init__.py | 21 ++ evals/l10_continuity/__main__.py | 35 +++ evals/l10_continuity/contract.md | 78 ++++++ evals/l10_continuity/corpus.py | 52 ++++ evals/l10_continuity/predicates.py | 389 +++++++++++++++++++++++++++++ evals/l10_continuity/report.py | 152 +++++++++++ evals/l10_continuity/runner.py | 202 +++++++++++++++ tests/test_l10_continuity.py | 376 ++++++++++++++++++++++++++++ 8 files changed, 1305 insertions(+) create mode 100644 evals/l10_continuity/__init__.py create mode 100644 evals/l10_continuity/__main__.py create mode 100644 evals/l10_continuity/contract.md create mode 100644 evals/l10_continuity/corpus.py create mode 100644 evals/l10_continuity/predicates.py create mode 100644 evals/l10_continuity/report.py create mode 100644 evals/l10_continuity/runner.py create mode 100644 tests/test_l10_continuity.py diff --git a/evals/l10_continuity/__init__.py b/evals/l10_continuity/__init__.py new file mode 100644 index 00000000..f0548faa --- /dev/null +++ b/evals/l10_continuity/__init__.py @@ -0,0 +1,21 @@ +"""L10 continuity spike — a falsifiable long-horizon soak of the real turn loop. + +This lane is the empirical gate between the two L10 targets (see +``docs/analysis/L10-continuity-spike-design-2026-06-05.md``): + +- **T-resume** (provable same-life *resume*): determinism + recovery (P1, P2, P3, P4). +- **T-experience** (a continuous *experiencing* field-life): the field's *content* + stays meaningful over indefinite uptime (P5). + +It is NOT a proof of correctness of any single turn (that is the cognition lane), +nor a wall-clock endurance certificate. It is a falsifiable soak: every predicate +must be able to fail loudly, and each predicate is mutation-verified to *bite* +before any PASS is trusted (CLAUDE.md schema-as-proof discipline). + +The lane drives the full runtime, but it never *directly* imports the GSM8K +serving path (``generate.derivation`` / ``core.reliability_gate``) and it is +read-only over the runtime — it records evidence and never mutates serving code +or any gold lane — so it cannot regress the serving metric. +""" + +from __future__ import annotations diff --git a/evals/l10_continuity/__main__.py b/evals/l10_continuity/__main__.py new file mode 100644 index 00000000..af3329a9 --- /dev/null +++ b/evals/l10_continuity/__main__.py @@ -0,0 +1,35 @@ +"""On-demand entrypoint for the L10 continuity soak panel. + +Run: PYTHONPATH=. .venv/bin/python -m evals.l10_continuity [n_turns] [reboot_turn] + +Prints the structured report as JSON and exits non-zero if any gate fails. This +lane is a soak — it is NOT in the default smoke suite; run it on demand or +nightly. The ``deterministic_digest`` in the output is the freeze handle: pin it +once the lane is trusted so a regression flips it. +""" + +from __future__ import annotations + +import json +import sys +import tempfile +from pathlib import Path + +from evals.l10_continuity.report import build_report + + +def main(argv: list[str]) -> int: + n_turns = int(argv[0]) if len(argv) > 0 else 12 + reboot_turn = int(argv[1]) if len(argv) > 1 else 3 + with tempfile.TemporaryDirectory(prefix="l10_continuity_") as tmp: + report = build_report( + n_turns=n_turns, + reboot_turn=reboot_turn, + engine_state_root=Path(tmp), + ) + print(json.dumps(report.to_dict(), indent=2, ensure_ascii=False)) + return 0 if report.all_gates_pass() else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/evals/l10_continuity/contract.md b/evals/l10_continuity/contract.md new file mode 100644 index 00000000..1722afb4 --- /dev/null +++ b/evals/l10_continuity/contract.md @@ -0,0 +1,78 @@ +# L10 Continuity Lane — Contract + +**Status:** spike (falsifiable experiment) · **Parent:** `docs/analysis/L10-continuity-spike-design-2026-06-05.md` · **Not in default smoke** (a soak; run on demand / nightly). + +This lane drives the **real** turn loop (`ChatRuntime` + `CognitiveTurnPipeline`) +over a deterministic, cyclic, in-vocabulary corpus for N turns, with optional +reboot and orphan-crash legs, and evaluates falsifiable predicates over the +recorded evidence. It is the empirical gate between the two L10 targets: + +- **T-resume** — provable same-life *resume* (determinism + recovery): P1–P4. +- **T-experience** — a continuous *experiencing* field-life (content stays + meaningful over a long horizon): P5. + +Run: `PYTHONPATH=. .venv/bin/python -m evals.l10_continuity [n_turns] [reboot_turn]` + +## Predicates + +| ID | Proves | Fails loudly when | Mutation-verified bite | +|----|--------|-------------------|------------------------| +| **P1** closure | `versor_condition < 1e-6` every turn | a construction breaks closure (no repair allowed) | a record with `versor_condition ≥ 1e-6` trips it | +| **P2a** determinism | two independent runtimes → byte-identical `trace_hash` sequence | the pipeline is nondeterministic | a perturbed hash trips it | +| **P2b** reboot transparency | a reboot never changes turns *before* the reboot point | determinism/state leaks backward across reboot | a pre-reboot hash divergence trips it | +| **P3** bounded resources | vault grows linear-bounded/monotonic per turn | an unbounded cache/store leaks | a 10k-entry vault record trips it | +| **P4** recovery determinism | two crash-recoveries from one checkpoint converge | torn read / nondeterministic boot | divergent recovery tails trip it | +| **P4** commit point | recovered `turn_count` == committed turns (ARIES force boundary) | the checkpoint isn't the atomic commit boundary | `None`/mismatched count trips it | +| **P5b** anchor stability | field anchors without **collapse** (`dist_to_anchor`↛0) or **freeze** (`turn_movement`↛0) | the field is swallowed by the attractor, or frozen | collapsing distance / zero movement trips it | +| **P5c** coherence | surfaces stay non-empty and not collapsed to one repeated output | the field wanders into noise or freezes onto one output | empty / single-surface records trip it | + +Each predicate has a `*_holds` test (real soak) **and** a `*_bites` test +(mutation), per the CLAUDE.md schema-as-proof discipline: a predicate that cannot +fail under the violation it nominally catches is decoration, not proof. + +## Not covered (no silent skips) + +- **P5a — recall precision@k stability.** Requires a held-out probe set with + known-relevant entries and a metric grounded in the vault's actual scoring + semantics (the raw recall score is not a clean similarity). Recorded as + `not_covered` in the report; a follow-up increment. + +## The headline diagnostic (P2b) + +Today a reboot restores only recognizers / discovery candidates / `turn_count` +(Shape B, ADR-0146) and discards the lived field / vault / anchor / graph / +referents. So `post_reboot_transparent == False`: a reboot diverges from the +uninterrupted run **at the first post-reboot turn**. This is the mechanical +proof of "many lives sharing a checkpoint" and the precise definition of the +Shape-B+ persistence work that closes resume-as-same-life. The +`test_p2b_documents_current_resume_gap` test pins this reality and will **flip** +if persistence is later added — forcing a doc update so the gap cannot close +silently. + +## Thresholds (empirical basis, not arbitrary) + +All gating thresholds are set from measured real-soak data and deliberately +conservative — these are **catastrophe gates** (collapse / freeze / unbounded +leak are yes/no failures of the T-experience claim), not early-warning trend +detectors. Gradual-drift detection is a deliberate long-horizon follow-up; +tightening toward the healthy band would risk false positives on a different +corpus or longer horizon. + +| Threshold | Measured healthy | Default floor/ceiling | Rationale | +|-----------|------------------|-----------------------|-----------| +| P5b `dist_to_anchor` | ~4.0–6.2 (steady) | `collapse_floor=1.0` | ~75%+ drop toward anchor = pathological collapse | +| P5b `turn_movement` | median ~1.5 | `freeze_floor=0.05` | ~1/30th of healthy = frozen field | +| P3 vault growth | ~2–3 entries/turn | `vault_per_turn_ceiling=4` | ~130–200% of as-designed writes | + +**P5b vs P5c division of labour:** P5b catches the field *freezing* (movement→0) +or *collapsing onto the anchor* (distance→0); P5c catches the *output* collapsing +to a single repeated surface. The P5c real-soak test runs **over more than one +corpus cycle** so the horizon exercises repetition (a 6-turn run == the cycle +length would trivially yield 6 distinct surfaces and prove nothing). + +## Freeze handle + +`report.deterministic_digest` is a SHA-256 over only hardware-stable evidence +(the `trace_hash` sequence + each predicate's `(name, passed)` verdict), +excluding RSS / wall-clock / raw floats. Pin it once the lane is trusted; a +regression flips it. diff --git a/evals/l10_continuity/corpus.py b/evals/l10_continuity/corpus.py new file mode 100644 index 00000000..a28b659f --- /dev/null +++ b/evals/l10_continuity/corpus.py @@ -0,0 +1,52 @@ +"""The deterministic scripted corpus that drives the L10 continuity soak. + +The corpus is a fixed, committed sequence of in-vocabulary natural-language +prompts. Determinism is the point: turn N of a soak is always ``prompt_at(N)``, +so two independent runs over the same N are byte-identical in their inputs, and +a reboot leg replays the exact same tail it would have seen uninterrupted. + +There is NO randomness here — "seeded/replayable" means a fixed cycle, not an +RNG. The prompts are hand-picked to be resident in the default cognition packs +so ``ChatRuntime.chat`` never raises ``no in-vocabulary tokens``; the runner +verifies this at turn 0 and fails loudly if a prompt drifts out of vocabulary. +""" + +from __future__ import annotations + +# A small, fixed ring of in-vocabulary prompts. Each is a complete cognition +# turn the default packs can tokenize and ground (mirrors the inputs used by +# the cognition lane and the ADR-0153 trace-hash tests). The ring is cycled to +# reach an arbitrary soak length; cycling (not appending novelty) is deliberate +# — the soak measures whether *repetition over a long horizon* stays closed, +# deterministic, bounded, and meaningful, not whether novel input is handled. +_BASE_PROMPTS: tuple[str, ...] = ( + "What causes light?", + "What is a concept?", + "Hello.", + "What causes rain?", + "What is a principle?", + "What is memory?", +) + + +def base_prompts() -> tuple[str, ...]: + """Return the immutable base ring of prompts (a safe copy of the tuple).""" + return _BASE_PROMPTS + + +def prompt_at(turn_index: int) -> str: + """The prompt for a given 0-based turn index, by cycling the base ring. + + Deterministic and total: any non-negative ``turn_index`` maps to exactly + one base prompt, so the corpus is replayable across runs and reboots. + """ + if turn_index < 0: + raise ValueError(f"turn_index must be non-negative, got {turn_index}") + return _BASE_PROMPTS[turn_index % len(_BASE_PROMPTS)] + + +def scripted_corpus(n_turns: int) -> tuple[str, ...]: + """The first ``n_turns`` prompts of the deterministic soak corpus.""" + if n_turns < 0: + raise ValueError(f"n_turns must be non-negative, got {n_turns}") + return tuple(prompt_at(i) for i in range(n_turns)) diff --git a/evals/l10_continuity/predicates.py b/evals/l10_continuity/predicates.py new file mode 100644 index 00000000..5774d004 --- /dev/null +++ b/evals/l10_continuity/predicates.py @@ -0,0 +1,389 @@ +"""Pure pass/fail predicates over soak evidence — the falsifiable gates. + +Each predicate is a pure function of ``SoakResult`` evidence (it runs no turns +and mutates nothing), so it is trivially replayable and each can be +mutation-verified to *bite*. The predicates: + +- **P1 closure** — every turn satisfies ``versor_condition < 1e-6``. A hard + green guard backed by algebra-owned construction (Decision 0). +- **P2a determinism** — two independent, no-reboot runs of equal length produce + byte-identical ``trace_hash`` sequences. A hard green guard; a failure is a + real nondeterminism bug. +- **P2b reboot transparency** — a rebooted run vs an uninterrupted baseline. The + *diagnostic*: today a reboot restores only recognizers / candidates / + turn_count (Shape B, ADR-0146) and discards the lived field / vault / anchor, + so the first post-reboot turn is expected to diverge. P2b LOCATES that + divergence; it does not pretend it is absent. The structural invariant it + enforces is weaker and always-true: a reboot must never change turns *before* + the reboot point. +- **P3 bounded resources** — vault growth stays linear-bounded per turn (no + unbounded cache/store leak). RSS is recorded for the long lane; on a short + soak it is dominated by startup and only loosely bounded here. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field + +from evals.l10_continuity.runner import SoakResult, TurnRecord + +VERSOR_CEILING: float = 1e-6 + + +@dataclass(frozen=True, slots=True) +class PredicateOutcome: + name: str + passed: bool + detail: str + metrics: dict = field(default_factory=dict) + + +def evaluate_p1_closure( + result: SoakResult, *, ceiling: float = VERSOR_CEILING +) -> PredicateOutcome: + """P1 — every turn's field is a valid versor (``versor_condition < ceiling``).""" + violations = [ + (r.turn_index, r.versor_condition) + for r in result.records + if not (r.versor_condition < ceiling) + ] + worst = max((r.versor_condition for r in result.records), default=0.0) + passed = not violations + detail = ( + f"all {len(result.records)} turns closed (worst={worst:.3e} < {ceiling:.0e})" + if passed + else f"{len(violations)} turn(s) breached the versor ceiling: {violations[:5]}" + ) + return PredicateOutcome( + name="P1_closure", + passed=passed, + detail=detail, + metrics={"worst_versor_condition": worst, "violations": violations}, + ) + + +def _first_divergence(a: tuple[str, ...], b: tuple[str, ...]) -> int | None: + """Index of the first position where two trace-hash sequences differ. + + A length mismatch counts as a divergence at the first extra/missing index. + Returns ``None`` when the sequences are byte-identical. + """ + for i in range(min(len(a), len(b))): + if a[i] != b[i]: + return i + if len(a) != len(b): + return min(len(a), len(b)) + return None + + +def evaluate_p2a_determinism( + run_a: SoakResult, run_b: SoakResult +) -> PredicateOutcome: + """P2a — two independent no-reboot runs are byte-identical in trace_hash.""" + if run_a.reboot_at or run_b.reboot_at: + raise ValueError("P2a compares two NO-reboot runs; pass reboot_at=().") + ha, hb = run_a.trace_hashes(), run_b.trace_hashes() + div = _first_divergence(ha, hb) + passed = div is None and len(ha) == len(hb) + detail = ( + f"{len(ha)} turns byte-identical across two independent runtimes" + if passed + else f"trace_hash diverged at turn {div} " + f"({ha[div] if div is not None and div < len(ha) else '∅'} != " + f"{hb[div] if div is not None and div < len(hb) else '∅'})" + ) + return PredicateOutcome( + name="P2a_determinism", + passed=passed, + detail=detail, + metrics={"n_turns": len(ha), "first_divergence": div}, + ) + + +@dataclass(frozen=True, slots=True) +class RebootTransparency: + """The measured outcome of a reboot leg vs an uninterrupted baseline.""" + + pre_reboot_identical: bool + post_reboot_transparent: bool + first_divergence: int | None + reboot_turn: int + + +def evaluate_p2b_reboot_transparency( + rebooted: SoakResult, baseline: SoakResult +) -> tuple[PredicateOutcome, RebootTransparency]: + """P2b — locate where a rebooted run diverges from an uninterrupted one. + + The predicate PASSES on the structural invariant only: a reboot must not + change any turn *before* the reboot point (those are the same first + segment, so they must be identical — a failure here is a real determinism + or state-leak bug). Full post-reboot transparency is the *measured* + diagnostic, returned alongside; it is expected to be ``False`` until the + lived field/vault are persisted across reboot (the Shape-B+ work). + """ + if not rebooted.reboot_at: + raise ValueError("P2b expects a rebooted run (reboot_at non-empty).") + if baseline.reboot_at: + raise ValueError("P2b baseline must be an uninterrupted run (reboot_at=()).") + + reboot_turn = rebooted.reboot_at[0] + hr, hb = rebooted.trace_hashes(), baseline.trace_hashes() + div = _first_divergence(hr, hb) + + pre_reboot_identical = div is None or div >= reboot_turn + post_reboot_transparent = div is None + + transparency = RebootTransparency( + pre_reboot_identical=pre_reboot_identical, + post_reboot_transparent=post_reboot_transparent, + first_divergence=div, + reboot_turn=reboot_turn, + ) + if not pre_reboot_identical: + detail = ( + f"determinism violated BEFORE reboot: diverged at turn {div} " + f"(reboot was at {reboot_turn}) — a reboot must not change earlier turns" + ) + elif post_reboot_transparent: + detail = ( + f"reboot at turn {reboot_turn} is FULLY transparent " + f"({len(hr)} turns byte-identical to the uninterrupted run)" + ) + else: + detail = ( + f"reboot at turn {reboot_turn} is NOT transparent: first divergence " + f"at turn {div} (lived field/vault not persisted — Shape B). " + "Pre-reboot turns are identical; the resume gap is post-reboot." + ) + return ( + PredicateOutcome( + name="P2b_reboot_transparency", + passed=pre_reboot_identical, + detail=detail, + metrics={ + "reboot_turn": reboot_turn, + "first_divergence": div, + "post_reboot_transparent": post_reboot_transparent, + }, + ), + transparency, + ) + + +def evaluate_p3_bounded_resources( + result: SoakResult, *, vault_per_turn_ceiling: int = 4 +) -> PredicateOutcome: + """P3 — vault growth is linear-bounded per turn (no unbounded store leak). + + The real turn loop stores a small fixed number of vault entries per turn + (user + assistant + occasional promotion); an unbounded cache or a per-turn + accumulator that grows super-linearly would breach the ceiling. RSS is + recorded for the long lane but is dominated by startup on a short soak, so + it is reported, not gated, here. + + Ceiling basis (measured): the real soak grows ~2–3 vault entries/turn; the + default ``vault_per_turn_ceiling=4`` is ~130–200% of that, so it tolerates + the as-designed user+assistant(+promotion) writes while a genuinely + unbounded store (a per-turn cache) breaches it. A leak slower than the + ceiling is by design out of scope for this linear-bound check; it is the + long-horizon RSS lane's job. + """ + if result.reboot_at: + raise ValueError("P3 expects a no-reboot run (vault resets on reboot).") + records: tuple[TurnRecord, ...] = result.records + sizes = [r.vault_size for r in records] + monotonic = all(b >= a for a, b in zip(sizes, sizes[1:])) + breaches = [ + (r.turn_index, r.vault_size) + for r in records + if r.vault_size > vault_per_turn_ceiling * (r.turn_index + 1) + ] + passed = monotonic and not breaches + peak_first = records[0].peak_rss_raw if records else 0 + peak_last = records[-1].peak_rss_raw if records else 0 + detail = ( + f"vault grew monotonically within {vault_per_turn_ceiling}/turn " + f"(final size {sizes[-1] if sizes else 0} over {len(records)} turns)" + if passed + else f"resource bound breached: monotonic={monotonic}, breaches={breaches[:5]}" + ) + return PredicateOutcome( + name="P3_bounded_resources", + passed=passed, + detail=detail, + metrics={ + "final_vault_size": sizes[-1] if sizes else 0, + "vault_monotonic": monotonic, + "vault_breaches": breaches, + "peak_rss_raw_first": peak_first, + "peak_rss_raw_last": peak_last, + }, + ) + + +def evaluate_p4_recovery_determinism( + recovery_a: SoakResult, recovery_b: SoakResult +) -> PredicateOutcome: + """P4 — two independent crash-recoveries from the same checkpoint converge. + + The L10 kill-9 claim: a hard kill (incl. mid-checkpoint-write) always + next-boots onto a valid prior checkpoint (ADR-0156 atomicity) and resumes + *deterministically*. Because Shape B discards the lived field/vault, a + recovered run does NOT match the uninterrupted baseline (that is the P2b + gap) — so determinism here means: two independent recoveries from the same + durable checkpoint produce byte-identical continuations. A non-deterministic + recovery (torn read, partial state, nondeterministic boot) breaks this. + """ + if not recovery_a.reboot_at or not recovery_b.reboot_at: + raise ValueError("P4 expects two crash-recovery runs (reboot_at non-empty).") + tail_a = tuple(r.trace_hash for r in recovery_a.post_reboot_records()) + tail_b = tuple(r.trace_hash for r in recovery_b.post_reboot_records()) + div = _first_divergence(tail_a, tail_b) + passed = div is None and len(tail_a) == len(tail_b) and len(tail_a) > 0 + detail = ( + f"two crash-recoveries produced byte-identical {len(tail_a)}-turn tails" + if passed + else f"recovery diverged at post-reboot index {div} " + f"(|a|={len(tail_a)}, |b|={len(tail_b)})" + ) + return PredicateOutcome( + name="P4_recovery_determinism", + passed=passed, + detail=detail, + metrics={"recovered_tail_len": len(tail_a), "first_divergence": div}, + ) + + +def evaluate_p4_commit_point( + recovered_turn_count: int | None, expected_turn_count: int +) -> PredicateOutcome: + """P4 (WAL/ARIES force boundary) — the checkpoint IS the commit boundary. + + The engine-state checkpoint is the last durable act of a turn, so a kill + next-boots onto a checkpoint whose ``turn_count`` equals the number of + fully-committed turns — never a partially-applied turn. A recovered count + that is ``None`` (no checkpoint) or != the committed count means the durable + record did not gate the turn as a unit. + """ + passed = recovered_turn_count == expected_turn_count + detail = ( + f"recovered checkpoint turn_count={recovered_turn_count} " + f"== {expected_turn_count} committed turns" + if passed + else f"recovered turn_count={recovered_turn_count} != " + f"expected {expected_turn_count} (commit boundary not atomic)" + ) + return PredicateOutcome( + name="P4_commit_point", + passed=passed, + detail=detail, + metrics={ + "recovered_turn_count": recovered_turn_count, + "expected_turn_count": expected_turn_count, + }, + ) + + +def evaluate_p5b_anchor_stability( + result: SoakResult, + *, + warmup: int = 2, + collapse_floor: float = 1.0, + freeze_floor: float = 0.05, +) -> PredicateOutcome: + """P5b — the field anchors without collapsing onto the attractor or freezing. + + The crux of the T-experience gate and the direct long-horizon test of the + sanctioned ``_session_anchor_pull`` (α=0.05). Two failure modes, both fatal + to "continuous experiencing life": + + - **collapse** — ``dist_to_anchor`` trends to 0 (the field is swallowed by + the anchor; every turn becomes the same concept). Guard: the minimum + steady-state distance stays above ``collapse_floor``. + - **freeze** — ``turn_movement`` trends to 0 (the field stops moving with + content). Guard: the median steady-state movement stays above + ``freeze_floor``. + + Evaluated over the steady state (after ``warmup`` turns) because turn 0 is + the anchor itself (distance 0) and turn 1 is a large transient. + + Threshold basis (measured, not arbitrary): on the real soak the steady-state + ``dist_to_anchor`` sits in a ~4.0–6.2 band and the median ``turn_movement`` + is ~1.5. The defaults are set deliberately BELOW that band — + ``collapse_floor=1.0`` (a ~75%+ drop toward the anchor) and + ``freeze_floor=0.05`` (movement ~1/30th of healthy) — so P5b is a *binary + catastrophe* gate (the T-experience question is "does the field collapse or + freeze?", a yes/no), NOT an early-warning trend detector. A gradual-drift + detector would need a long-horizon trend test and is a deliberate follow-up; + tightening these floors toward the healthy band risks false positives on a + different corpus or a longer horizon. + """ + if result.reboot_at: + raise ValueError("P5b expects a no-reboot run (anchor resets on reboot).") + tail = result.records[warmup:] + dists = [r.dist_to_anchor for r in tail if not math.isnan(r.dist_to_anchor)] + moves = [r.turn_movement for r in tail if not math.isnan(r.turn_movement)] + if len(dists) < 2 or len(moves) < 2: + return PredicateOutcome( + name="P5b_anchor_stability", + passed=False, + detail=f"insufficient steady-state turns to evaluate (warmup={warmup})", + metrics={"n_steady": len(dists)}, + ) + min_dist = min(dists) + sorted_moves = sorted(moves) + median_move = sorted_moves[len(sorted_moves) // 2] + no_collapse = min_dist > collapse_floor + no_freeze = median_move > freeze_floor + passed = no_collapse and no_freeze + if passed: + detail = ( + f"anchored without collapse (min dist {min_dist:.3f} > {collapse_floor}) " + f"or freeze (median move {median_move:.3f} > {freeze_floor})" + ) + else: + cause = [] + if not no_collapse: + cause.append(f"COLLAPSE (min dist {min_dist:.3f} ≤ {collapse_floor})") + if not no_freeze: + cause.append(f"FREEZE (median move {median_move:.3f} ≤ {freeze_floor})") + detail = "; ".join(cause) + return PredicateOutcome( + name="P5b_anchor_stability", + passed=passed, + detail=detail, + metrics={ + "min_steady_dist_to_anchor": min_dist, + "median_steady_movement": median_move, + "n_steady": len(dists), + }, + ) + + +def evaluate_p5c_coherence( + result: SoakResult, *, min_surface_len: int = 1, min_distinct_surfaces: int = 2 +) -> PredicateOutcome: + """P5c — the field does not wander into noise or collapse to one output. + + Two degeneracies: empty/trivial surfaces (the field drifted into noise) and + a single repeated surface across the whole horizon (the field froze onto one + output). Both are caught by surface non-emptiness + a distinct-surface floor. + """ + surfaces = [r.surface for r in result.records] + empties = [r.turn_index for r in result.records if len(r.surface) < min_surface_len] + distinct = len(set(surfaces)) + passed = not empties and distinct >= min_distinct_surfaces + detail = ( + f"surfaces stayed coherent ({distinct} distinct, none empty) " + f"over {len(surfaces)} turns" + if passed + else f"incoherent: empties={empties[:5]}, distinct_surfaces={distinct}" + ) + return PredicateOutcome( + name="P5c_coherence", + passed=passed, + detail=detail, + metrics={"distinct_surfaces": distinct, "empty_turns": empties}, + ) diff --git a/evals/l10_continuity/report.py b/evals/l10_continuity/report.py new file mode 100644 index 00000000..130b3a82 --- /dev/null +++ b/evals/l10_continuity/report.py @@ -0,0 +1,152 @@ +"""Assemble the L10 continuity panel into a structured, freeze-gateable report. + +The panel runs the soaks the predicates need (an uninterrupted baseline, a +reboot leg, and two crash-recoveries), evaluates every predicate, and emits a +structured report with per-predicate PASS/FAIL, metrics, the explicitly +*not-covered* legs (no silent skips — CLAUDE.md), and a **deterministic digest**. + +The digest is a SHA-256 over only the hardware-stable evidence: the canonical +``trace_hash`` sequence (``core.cognition.trace`` already rounds floats so the +hash is stable across hardware) and each predicate's ``(name, passed)`` verdict. +It deliberately EXCLUDES RSS, wall-clock, and raw float metrics, which are not +reproducible across machines. The digest is the freeze handle: pin it once the +lane is trusted and a regression flips it. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict, dataclass +from pathlib import Path + +from core.config import RuntimeConfig + +from evals.l10_continuity.predicates import ( + PredicateOutcome, + evaluate_p1_closure, + evaluate_p2a_determinism, + evaluate_p2b_reboot_transparency, + evaluate_p3_bounded_resources, + evaluate_p4_commit_point, + evaluate_p4_recovery_determinism, + evaluate_p5b_anchor_stability, + evaluate_p5c_coherence, +) +from evals.l10_continuity.runner import ( + SoakResult, + _inject_orphan_tmp, + read_recovered_turn_count, + run_soak, +) + +# Legs the spec names but this lane does not yet cover, recorded explicitly so a +# PASS is never read as "everything was checked". +NOT_COVERED: tuple[tuple[str, str], ...] = ( + ( + "P5a_recall_stability", + "recall precision@k over a held-out probe set requires a probe set with " + "known-relevant entries and a metric grounded in the vault's scoring " + "semantics (the raw recall score is not a clean similarity); deferred.", + ), +) + + +@dataclass(frozen=True) +class L10ContinuityReport: + n_turns: int + reboot_turn: int + predicates: tuple[PredicateOutcome, ...] + not_covered: tuple[tuple[str, str], ...] + deterministic_digest: str + + def all_gates_pass(self) -> bool: + return all(p.passed for p in self.predicates) + + def to_dict(self) -> dict: + return { + "n_turns": self.n_turns, + "reboot_turn": self.reboot_turn, + "all_gates_pass": self.all_gates_pass(), + "deterministic_digest": self.deterministic_digest, + "predicates": [asdict(p) for p in self.predicates], + "not_covered": [ + {"leg": leg, "reason": reason} for leg, reason in self.not_covered + ], + } + + +def deterministic_digest( + baseline: SoakResult, predicates: tuple[PredicateOutcome, ...] +) -> str: + """SHA-256 over hardware-stable evidence: trace_hash sequence + verdicts.""" + payload = { + "trace_hashes": list(baseline.trace_hashes()), + "verdicts": [[p.name, p.passed] for p in predicates], + "not_covered": [leg for leg, _ in NOT_COVERED], + } + serialized = json.dumps(payload, sort_keys=True, ensure_ascii=False) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +def build_report( + *, + n_turns: int = 12, + reboot_turn: int = 3, + engine_state_root: Path, + config: RuntimeConfig | None = None, +) -> L10ContinuityReport: + """Run the full panel and assemble the report. + + Soaks: an uninterrupted ``baseline``; a second independent ``run_b`` (P2a); + a ``reboot`` leg (P2b); and two orphan-crash recoveries (P4). + """ + config = config or RuntimeConfig() + root = engine_state_root + + baseline = run_soak(n_turns, engine_state_dir=root / "baseline", config=config) + run_b = run_soak(n_turns, engine_state_dir=root / "run_b", config=config) + reboot = run_soak( + n_turns, engine_state_dir=root / "reboot", reboot_at=(reboot_turn,), config=config + ) + rec_a = run_soak( + n_turns, + engine_state_dir=root / "rec_a", + reboot_at=(reboot_turn,), + inject_orphan_tmp_at_reboot=True, + config=config, + ) + rec_b = run_soak( + n_turns, + engine_state_dir=root / "rec_b", + reboot_at=(reboot_turn,), + inject_orphan_tmp_at_reboot=True, + config=config, + ) + # Commit-point probe: run exactly ``reboot_turn`` turns, simulate the torn + # write, and read the recovered turn_count AT the crash boundary (not after + # the recovery continues and re-checkpoints). + probe_dir = root / "commit_probe" + run_soak(reboot_turn, engine_state_dir=probe_dir, config=config) + _inject_orphan_tmp(probe_dir) + recovered = read_recovered_turn_count(probe_dir) + + p2b_outcome, _ = evaluate_p2b_reboot_transparency(reboot, baseline) + predicates: tuple[PredicateOutcome, ...] = ( + evaluate_p1_closure(baseline), + evaluate_p2a_determinism(baseline, run_b), + p2b_outcome, + evaluate_p3_bounded_resources(baseline), + evaluate_p4_recovery_determinism(rec_a, rec_b), + evaluate_p4_commit_point(recovered, expected_turn_count=reboot_turn), + evaluate_p5b_anchor_stability(baseline), + evaluate_p5c_coherence(baseline), + ) + digest = deterministic_digest(baseline, predicates) + return L10ContinuityReport( + n_turns=n_turns, + reboot_turn=reboot_turn, + predicates=predicates, + not_covered=NOT_COVERED, + deterministic_digest=digest, + ) diff --git a/evals/l10_continuity/runner.py b/evals/l10_continuity/runner.py new file mode 100644 index 00000000..9cb90515 --- /dev/null +++ b/evals/l10_continuity/runner.py @@ -0,0 +1,202 @@ +"""The L10 continuity soak runner — drives the REAL turn loop over N turns. + +It runs the deterministic corpus through ``CognitiveTurnPipeline`` over a fresh +``ChatRuntime`` whose engine-state checkpoint lives in a caller-supplied +directory. Optionally it injects *reboot legs*: at a chosen turn boundary it +drops the live runtime and reconstructs a new one from the on-disk checkpoint — +exactly the lifecycle the L10 telos asks about ("resume as the same life") — and +optionally simulates a kill mid-checkpoint-write by leaving an orphan temp file +the reconstruct must ignore (ADR-0156 atomicity). + +The runner is pure instrumentation: it records per-turn evidence +(``versor_condition``, canonical ``trace_hash``, vault size, peak RSS, anchor +distance, turn-to-turn field movement, and which boot segment produced the turn) +and returns it. It makes NO pass/fail judgement — that is ``predicates.py`` — and +it never repairs, normalizes, or mutates field state (it only reads what the real +pipeline produced). + +What a reboot restores (today, Shape B / ADR-0146): recognizers, discovery +candidates, and ``turn_count`` — and NOTHING else. The lived field, vault, +session graph, referents, and session anchor are process-local and discarded on +exit. The ``booted_segment`` tag on each record exists precisely so the +reboot-transparency predicate (P2b) can locate where a rebooted run diverges +from an uninterrupted one. +""" + +from __future__ import annotations + +import resource +from dataclasses import dataclass +from pathlib import Path + +import numpy as np + +from chat.runtime import ChatRuntime +from core.cognition.pipeline import CognitiveTurnPipeline +from core.config import RuntimeConfig + +from evals.l10_continuity.corpus import prompt_at + + +@dataclass(frozen=True, slots=True) +class TurnRecord: + """Per-turn evidence captured from the real pipeline (no judgement).""" + + turn_index: int + input_text: str + trace_hash: str + versor_condition: float + surface: str + vault_size: int + peak_rss_raw: int + booted_segment: int + # P5 signals (NaN when undefined — e.g. movement on a segment's first turn, + # or distance before an anchor exists). + dist_to_anchor: float + turn_movement: float + + +@dataclass(frozen=True, slots=True) +class SoakResult: + """The full ordered evidence of one soak run.""" + + n_turns: int + reboot_at: tuple[int, ...] + records: tuple[TurnRecord, ...] + + def trace_hashes(self) -> tuple[str, ...]: + return tuple(r.trace_hash for r in self.records) + + def versor_conditions(self) -> tuple[float, ...]: + return tuple(r.versor_condition for r in self.records) + + def post_reboot_records(self) -> tuple[TurnRecord, ...]: + """Records produced at/after the first reboot (the recovered tail).""" + if not self.reboot_at: + return () + first = self.reboot_at[0] + return tuple(r for r in self.records if r.turn_index >= first) + + +def _peak_rss_raw() -> int: + """Process peak RSS as the OS reports it (bytes on macOS, KiB on Linux). + + The unit differs by platform, so callers must use this only for + *ratio*/monotonic checks (P3), never as an absolute byte ceiling. + """ + return int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) + + +def _new_runtime(config: RuntimeConfig, engine_state_dir: Path) -> ChatRuntime: + """Construct a ChatRuntime bound to the checkpoint dir. + + Reconstruction is the reboot: ``ChatRuntime.__init__`` loads the on-disk + engine-state checkpoint when one exists (recognizers / candidates / + turn_count), so a second instance over the same directory resumes from the + last durable checkpoint. + """ + return ChatRuntime(config=config, engine_state_path=engine_state_dir) + + +def _inject_orphan_tmp(engine_state_dir: Path) -> None: + """Simulate a kill mid-checkpoint-write: leave an orphan temp file. + + 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. + """ + 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") + + +def read_recovered_turn_count(engine_state_dir: Path) -> int | None: + """Read ``turn_count`` from the on-disk manifest, or None if absent.""" + from engine_state import EngineStateStore + + manifest = EngineStateStore(engine_state_dir).load_manifest() + return None if manifest is None else int(manifest.get("turn_count", 0)) + + +def _anchor_distance(runtime: ChatRuntime) -> float: + ctx = runtime._context + if ctx.state is None or ctx._anchor_field is None: + return float("nan") + f = np.asarray(ctx.state.F, dtype=np.float64) + anchor = np.asarray(ctx._anchor_field, dtype=np.float64) + return float(np.linalg.norm(f - anchor)) + + +def _current_field(runtime: ChatRuntime) -> np.ndarray | None: + ctx = runtime._context + return None if ctx.state is None else np.asarray(ctx.state.F, dtype=np.float64) + + +def run_soak( + n_turns: int, + *, + engine_state_dir: Path, + reboot_at: tuple[int, ...] = (), + config: RuntimeConfig | None = None, + inject_orphan_tmp_at_reboot: bool = False, +) -> SoakResult: + """Run ``n_turns`` of the deterministic corpus, optionally rebooting. + + ``reboot_at`` is a set of turn indices at which, *before* running that turn, + the live runtime is dropped and reconstructed from the checkpoint. A reboot + at turn 0 is meaningless (nothing checkpointed yet) and is ignored. When + ``inject_orphan_tmp_at_reboot`` is set, a torn-write orphan temp file is left + in the checkpoint dir immediately before each reconstruct, so the reboot + exercises ADR-0156 crash recovery rather than a clean restart. + """ + if n_turns < 0: + raise ValueError(f"n_turns must be non-negative, got {n_turns}") + config = config or RuntimeConfig() + reboot_set = {i for i in reboot_at if i > 0} + + runtime = _new_runtime(config, engine_state_dir) + pipe = CognitiveTurnPipeline(runtime=runtime) + segment = 0 + prev_field: np.ndarray | None = None + records: list[TurnRecord] = [] + + for i in range(n_turns): + if i in reboot_set: + if inject_orphan_tmp_at_reboot: + _inject_orphan_tmp(engine_state_dir) + runtime = _new_runtime(config, engine_state_dir) + pipe = CognitiveTurnPipeline(runtime=runtime) + segment += 1 + prev_field = None # movement is undefined across a reboot boundary + + text = prompt_at(i) + result = pipe.run(text) + field = _current_field(runtime) + movement = ( + float(np.linalg.norm(field - prev_field)) + if field is not None and prev_field is not None + else float("nan") + ) + records.append( + TurnRecord( + turn_index=i, + input_text=text, + trace_hash=result.trace_hash, + versor_condition=float(result.versor_condition), + surface=result.surface, + vault_size=len(runtime._context.vault), + peak_rss_raw=_peak_rss_raw(), + booted_segment=segment, + dist_to_anchor=_anchor_distance(runtime), + turn_movement=movement, + ) + ) + prev_field = field + + return SoakResult( + n_turns=n_turns, + reboot_at=tuple(sorted(reboot_set)), + records=tuple(records), + ) diff --git a/tests/test_l10_continuity.py b/tests/test_l10_continuity.py new file mode 100644 index 00000000..62f9dce3 --- /dev/null +++ b/tests/test_l10_continuity.py @@ -0,0 +1,376 @@ +"""L10 continuity spike — foundation lane (P1, P2a, P2b, P3). + +Two kinds of test per predicate: + +- a ``*_holds`` test that drives the REAL turn loop over a short soak and asserts + the predicate passes on genuine evidence, and +- a ``*_bites`` test that feeds the predicate a single mutated record and asserts + it FAILS — the schema-as-proof obligation (CLAUDE.md): a predicate that cannot + fail under the violation it nominally catches is decoration, not proof. + +The soak-running tests use a small N and a tmp engine-state dir; they are NOT in +the default smoke suite (this is a soak lane, run on demand / nightly). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from evals.l10_continuity.corpus import prompt_at, scripted_corpus +from evals.l10_continuity.predicates import ( + VERSOR_CEILING, + evaluate_p1_closure, + evaluate_p2a_determinism, + evaluate_p2b_reboot_transparency, + evaluate_p3_bounded_resources, + evaluate_p4_commit_point, + evaluate_p4_recovery_determinism, + evaluate_p5b_anchor_stability, + evaluate_p5c_coherence, +) +from evals.l10_continuity.runner import ( + SoakResult, + TurnRecord, + read_recovered_turn_count, + run_soak, +) + +_SOAK_N = 6 # short horizon: enough to cycle the corpus and cross a reboot + + +# --------------------------------------------------------------------------- # +# Synthetic-evidence helpers (fast; no pipeline) — used by the *_bites tests. # +# --------------------------------------------------------------------------- # +def _rec( + i: int, + *, + trace_hash: str | None = None, + versor_condition: float = 1e-13, + vault_size: int | None = None, + booted_segment: int = 0, + surface: str | None = None, + dist_to_anchor: float = 5.0, + turn_movement: float = 1.0, +) -> TurnRecord: + return TurnRecord( + turn_index=i, + input_text=prompt_at(i), + trace_hash=trace_hash if trace_hash is not None else f"hash-{i}", + versor_condition=versor_condition, + surface=surface if surface is not None else f"surface-{i}", + vault_size=vault_size if vault_size is not None else 2 * (i + 1), + peak_rss_raw=1_000_000, + booted_segment=booted_segment, + dist_to_anchor=dist_to_anchor, + turn_movement=turn_movement, + ) + + +def _synthetic(records: list[TurnRecord], reboot_at: tuple[int, ...] = ()) -> SoakResult: + return SoakResult(n_turns=len(records), reboot_at=reboot_at, records=tuple(records)) + + +# --------------------------------------------------------------------------- # +# Corpus determinism # +# --------------------------------------------------------------------------- # +def test_corpus_is_deterministic_and_total() -> None: + assert scripted_corpus(10) == scripted_corpus(10) + assert prompt_at(0) == prompt_at(6) # ring of length 6 cycles + assert scripted_corpus(3) == tuple(prompt_at(i) for i in range(3)) + with pytest.raises(ValueError): + prompt_at(-1) + + +# --------------------------------------------------------------------------- # +# P1 — closure # +# --------------------------------------------------------------------------- # +def test_p1_closure_holds_on_real_soak(tmp_path: Path) -> None: + result = run_soak(_SOAK_N, engine_state_dir=tmp_path / "es") + outcome = evaluate_p1_closure(result) + assert outcome.passed, outcome.detail + assert outcome.metrics["worst_versor_condition"] < VERSOR_CEILING + + +def test_p1_closure_bites_on_breached_versor() -> None: + bad = _synthetic([_rec(0), _rec(1, versor_condition=1e-3), _rec(2)]) + outcome = evaluate_p1_closure(bad) + assert not outcome.passed + assert (1, 1e-3) in outcome.metrics["violations"] + + +# --------------------------------------------------------------------------- # +# P2a — pipeline determinism (two independent no-reboot runs) # +# --------------------------------------------------------------------------- # +def test_p2a_determinism_holds_across_independent_runtimes(tmp_path: Path) -> None: + run_a = run_soak(_SOAK_N, engine_state_dir=tmp_path / "a") + run_b = run_soak(_SOAK_N, engine_state_dir=tmp_path / "b") + outcome = evaluate_p2a_determinism(run_a, run_b) + assert outcome.passed, outcome.detail + # And the trace_hashes are genuinely populated (not all empty strings). + assert all(h for h in run_a.trace_hashes()), "pipeline must produce trace_hashes" + + +def test_p2a_determinism_bites_on_perturbed_hash() -> None: + base = [_rec(i) for i in range(4)] + perturbed = [_rec(i) for i in range(4)] + perturbed[2] = _rec(2, trace_hash="DIVERGED") + outcome = evaluate_p2a_determinism(_synthetic(base), _synthetic(perturbed)) + assert not outcome.passed + assert outcome.metrics["first_divergence"] == 2 + + +# --------------------------------------------------------------------------- # +# P2b — reboot transparency (the diagnostic) # +# --------------------------------------------------------------------------- # +def test_p2b_pre_reboot_invariant_holds_on_real_soak(tmp_path: Path) -> None: + reboot_turn = 3 + rebooted = run_soak( + _SOAK_N, engine_state_dir=tmp_path / "r", reboot_at=(reboot_turn,) + ) + baseline = run_soak(_SOAK_N, engine_state_dir=tmp_path / "base") + outcome, transparency = evaluate_p2b_reboot_transparency(rebooted, baseline) + # The structural invariant ALWAYS holds: a reboot cannot change earlier turns. + assert outcome.passed, outcome.detail + assert transparency.pre_reboot_identical + # Diagnostic record: whatever the persistence story, a divergence (if any) + # must not appear before the reboot turn. + if transparency.first_divergence is not None: + assert transparency.first_divergence >= reboot_turn + + +def test_p2b_documents_current_resume_gap(tmp_path: Path) -> None: + """Diagnostic: TODAY (Shape B, field/vault not persisted) a reboot is NOT + transparent — the first post-reboot turn diverges because the lived field + and vault were discarded. This test pins that empirical reality so that if + persistence is later built, it flips and we are forced to update the claim. + """ + reboot_turn = 3 + rebooted = run_soak( + _SOAK_N, engine_state_dir=tmp_path / "r", reboot_at=(reboot_turn,) + ) + baseline = run_soak(_SOAK_N, engine_state_dir=tmp_path / "base") + _, transparency = evaluate_p2b_reboot_transparency(rebooted, baseline) + assert not transparency.post_reboot_transparent, ( + "Reboot transparency is unexpected under Shape B: if this fails, the " + "lived field/vault are now surviving reboot — update the L10 spike doc " + "and this assertion to assert transparency." + ) + assert transparency.first_divergence == reboot_turn + + +def test_p2b_bites_on_pre_reboot_divergence() -> None: + reboot_turn = 3 + baseline = [_rec(i) for i in range(6)] + # A rebooted run that (wrongly) differs BEFORE the reboot point. + rebooted_records = [_rec(i) for i in range(6)] + rebooted_records[1] = _rec(1, trace_hash="LEAKED-BACKWARD") + outcome, transparency = evaluate_p2b_reboot_transparency( + _synthetic(rebooted_records, reboot_at=(reboot_turn,)), + _synthetic(baseline), + ) + assert not outcome.passed + assert not transparency.pre_reboot_identical + assert transparency.first_divergence == 1 + + +# --------------------------------------------------------------------------- # +# P3 — bounded resources # +# --------------------------------------------------------------------------- # +def test_p3_bounded_resources_holds_on_real_soak(tmp_path: Path) -> None: + result = run_soak(_SOAK_N, engine_state_dir=tmp_path / "es") + outcome = evaluate_p3_bounded_resources(result) + assert outcome.passed, outcome.detail + assert outcome.metrics["vault_monotonic"] + + +def test_p3_bounded_resources_bites_on_unbounded_vault() -> None: + records = [_rec(i) for i in range(5)] + # Simulate an unbounded store: turn 4 holds far more than ceiling*turns. + records[4] = _rec(4, vault_size=10_000) + outcome = evaluate_p3_bounded_resources(_synthetic(records)) + assert not outcome.passed + assert (4, 10_000) in outcome.metrics["vault_breaches"] + + +# --------------------------------------------------------------------------- # +# P4 — kill-9 crash recovery (ADR-0156 atomicity + WAL commit boundary) # +# --------------------------------------------------------------------------- # +def test_p4_atomic_write_survives_mid_replace_kill(tmp_path: Path, monkeypatch) -> None: + """ADR-0156: a kill at the os.replace instant leaves the PRIOR checkpoint + fully intact and no partial target behind.""" + import engine_state + from engine_state import EngineStateStore + + store = EngineStateStore(tmp_path) + store.save_manifest(turn_count=7) # a good prior checkpoint + good = (tmp_path / "manifest.json").read_bytes() + + def _boom(*_args, **_kwargs): # simulate SIGKILL between fsync and rename + raise OSError("killed mid-replace") + + monkeypatch.setattr(engine_state.os, "replace", _boom) + with pytest.raises(OSError): + store.save_manifest(turn_count=8) + + # Prior target intact; no orphan .tmp left in place by the except cleanup. + assert (tmp_path / "manifest.json").read_bytes() == good + assert not list(tmp_path.glob(".manifest.json.*.tmp")) + + +def test_p4_recovered_checkpoint_is_valid_prior(tmp_path: Path) -> None: + """A reboot after an orphan-leaving crash loads the valid prior turn_count.""" + state_dir = tmp_path / "es" + k = 4 + run_soak(k, engine_state_dir=state_dir) # K committed turns + # Simulate the torn write, then verify the loader recovers turn_count == K. + from evals.l10_continuity.runner import _inject_orphan_tmp + + _inject_orphan_tmp(state_dir) + recovered = read_recovered_turn_count(state_dir) + outcome = evaluate_p4_commit_point(recovered, expected_turn_count=k) + assert outcome.passed, outcome.detail + + +def test_p4_recovery_is_deterministic_across_orphan_crash(tmp_path: Path) -> None: + reboot_turn = 3 + rec_a = run_soak( + _SOAK_N, + engine_state_dir=tmp_path / "a", + reboot_at=(reboot_turn,), + inject_orphan_tmp_at_reboot=True, + ) + rec_b = run_soak( + _SOAK_N, + engine_state_dir=tmp_path / "b", + reboot_at=(reboot_turn,), + inject_orphan_tmp_at_reboot=True, + ) + outcome = evaluate_p4_recovery_determinism(rec_a, rec_b) + assert outcome.passed, outcome.detail + assert outcome.metrics["recovered_tail_len"] == _SOAK_N - reboot_turn + + +def test_p4_recovery_determinism_bites_on_divergent_tail() -> None: + """Synthetic bite for the predicate itself: two recoveries whose post-reboot + tails differ MUST fail (the corrupt-checkpoint test below exercises system + atomicity, not this predicate's logic).""" + reboot_turn = 2 + base = [_rec(i, booted_segment=0 if i < reboot_turn else 1) for i in range(5)] + div = [_rec(i, booted_segment=0 if i < reboot_turn else 1) for i in range(5)] + div[3] = _rec(3, trace_hash="DIVERGED", booted_segment=1) # post-reboot index 1 + outcome = evaluate_p4_recovery_determinism( + _synthetic(base, reboot_at=(reboot_turn,)), + _synthetic(div, reboot_at=(reboot_turn,)), + ) + assert not outcome.passed + assert outcome.metrics["first_divergence"] == 1 + + +def test_p4_commit_point_bites_on_missing_or_partial_checkpoint() -> None: + # No checkpoint at all → recovered count is None → not the committed count. + assert not evaluate_p4_commit_point(None, expected_turn_count=5).passed + # A checkpoint that recorded fewer turns than were committed (non-atomic). + assert not evaluate_p4_commit_point(3, expected_turn_count=5).passed + + +def test_p4_recovery_bites_on_corrupt_checkpoint(tmp_path: Path) -> None: + """In-place corruption (the thing atomicity prevents) MUST fail recovery + loudly — proving the atomic write is load-bearing, not decoration.""" + from chat.runtime import ChatRuntime + from core.config import RuntimeConfig + + state_dir = tmp_path / "es" + run_soak(3, engine_state_dir=state_dir) + (state_dir / "manifest.json").write_text("{ torn-write garbage <<<", encoding="utf-8") + with pytest.raises(Exception): + ChatRuntime(config=RuntimeConfig(), engine_state_path=state_dir) + + +# --------------------------------------------------------------------------- # +# P5 — semantic quality over the horizon (the T-experience gate) # +# --------------------------------------------------------------------------- # +def test_p5b_anchor_stability_holds_on_real_soak(tmp_path: Path) -> None: + result = run_soak(12, engine_state_dir=tmp_path / "es") + outcome = evaluate_p5b_anchor_stability(result) + assert outcome.passed, outcome.detail + assert outcome.metrics["min_steady_dist_to_anchor"] > 1.0 + + +def test_p5b_bites_on_anchor_collapse() -> None: + # dist_to_anchor monotonically collapsing to ~0 → field swallowed by anchor. + records = [ + _rec(i, dist_to_anchor=max(0.0, 5.0 - i), turn_movement=1.0) for i in range(8) + ] + outcome = evaluate_p5b_anchor_stability(_synthetic(records)) + assert not outcome.passed + assert "COLLAPSE" in outcome.detail + + +def test_p5b_bites_on_field_freeze() -> None: + # turn_movement ~0 → field stopped moving with content (frozen attractor). + records = [_rec(i, dist_to_anchor=5.0, turn_movement=0.0) for i in range(8)] + outcome = evaluate_p5b_anchor_stability(_synthetic(records)) + assert not outcome.passed + assert "FREEZE" in outcome.detail + + +def test_p5c_coherence_holds_over_multiple_corpus_cycles(tmp_path: Path) -> None: + # Span >2 corpus cycles (ring length 6) so the horizon exercises REPETITION, + # not just 6 unique prompts — a total output collapse across cycles would + # drop distinct_surfaces toward 1 and trip the predicate. + from evals.l10_continuity.corpus import base_prompts + + n = len(base_prompts()) * 2 + 2 # 14 turns over a 6-prompt ring + result = run_soak(n, engine_state_dir=tmp_path / "es") + outcome = evaluate_p5c_coherence(result) + assert outcome.passed, outcome.detail + assert n > len(base_prompts()), "horizon must exceed one cycle to be meaningful" + + +def test_p5c_bites_on_empty_surfaces() -> None: + records = [_rec(i, surface="") for i in range(4)] + outcome = evaluate_p5c_coherence(_synthetic(records)) + assert not outcome.passed + + +def test_p5c_bites_on_frozen_single_surface() -> None: + records = [_rec(i, surface="the same thing") for i in range(6)] + outcome = evaluate_p5c_coherence(_synthetic(records)) + assert not outcome.passed + assert outcome.metrics["distinct_surfaces"] == 1 + + +# --------------------------------------------------------------------------- # +# Report panel + freeze-gate digest # +# --------------------------------------------------------------------------- # +def test_report_panel_passes_and_records_not_covered(tmp_path: Path) -> None: + from evals.l10_continuity.report import NOT_COVERED, build_report + + report = build_report(n_turns=8, reboot_turn=3, engine_state_root=tmp_path) + assert report.all_gates_pass(), [ + (p.name, p.detail) for p in report.predicates if not p.passed + ] + # Every spec leg the lane does NOT cover is recorded explicitly (no silent skip). + assert ("P5a_recall_stability", NOT_COVERED[0][1]) in report.not_covered + # The deterministic digest is a 64-hex SHA-256. + assert len(report.deterministic_digest) == 64 + # The report serializes cleanly (for the on-disk artifact). + d = report.to_dict() + assert d["all_gates_pass"] is True + assert any(p["name"] == "P2b_reboot_transparency" for p in d["predicates"]) + + +def test_report_digest_is_pure_and_bites() -> None: + from evals.l10_continuity.predicates import evaluate_p1_closure + from evals.l10_continuity.report import deterministic_digest + + baseline = _synthetic([_rec(i) for i in range(5)]) + outcomes = (evaluate_p1_closure(baseline),) + a = deterministic_digest(baseline, outcomes) + b = deterministic_digest(baseline, outcomes) + assert a == b # pure + # A flipped verdict changes the digest (the freeze handle bites). + flipped = (evaluate_p1_closure(_synthetic([_rec(0, versor_condition=1e-3)])),) + assert deterministic_digest(baseline, flipped) != a