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

feat(evals): L10 continuity spike — falsifiable long-horizon soak lane

Browse source 
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.
This commit is contained in:
Shay 2026-06-05 11:14:17 -07:00
parent 2d6f9d7ac3
commit 23bc28caf9
8 changed files with 1305 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
evals/l10_continuity/__init__.py Normal file
View file

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

35
evals/l10_continuity/__main__.py Normal file
View file

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

78
evals/l10_continuity/contract.md Normal file
View file

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

52
evals/l10_continuity/corpus.py Normal file
View file

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

389
evals/l10_continuity/predicates.py Normal file
View file

@ -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},
)

152
evals/l10_continuity/report.py Normal file
View file

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

202
evals/l10_continuity/runner.py Normal file
View file

@ -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 ``.<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.
"""
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),
)

376
tests/test_l10_continuity.py Normal file
View file

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