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

test(l10): P5a recall-precision predicate — cross-reboot vault exact-match gate

Browse source 
Adds the P5a recall-precision predicate that was listed as NOT_COVERED in
the L10 continuity lane.  Closes the gap in the schema-as-proof discipline
(CLAUDE.md): every predicate must have both a *_holds test (real soak) and
a *_bites mutation test so a passing lane cannot silently miss the violation
it nominally catches.

Changes:
- runner.py: ProbeRecord dataclass + probe_at/verify_probes_at params on
  run_soak().  Registers a field state (float32 bytes, matching vault's
  _exact_index dtype) at a named turn and recalls it against the vault at a
  later turn — including after a reboot, which is the cross-reboot claim.
- predicates.py: evaluate_p5a_recall_precision — fails if any ProbeRecord
  has rank=None or rank>top_k, or if no cross-reboot probe was recorded.
- report.py: wires P5a into build_report(); probe registered at turn 1,
  verified at reboot_turn+2 (intentionally before the vault's
  null_project auto-reproject cycle at store_count=20, which would destroy
  all CGA inner-product scores — documented as a real finding, deferred to
  a follow-up increment); NOT_COVERED is now empty ().
- contract.md: P5a row in the predicate table + reprojection-boundary
  scope note.
- test_l10_continuity.py: 4 tests — holds (real 6-turn soak across reboot)
  + 3 bites (rank=None, no cross-reboot probe, empty probe_records).

Key finding: vault.null_project() fires every vault_reproject_interval=20
stores and produces CGA-orthogonal versors (inner product → 0.0 with the
original), completely destroying both exact-match and ranked recall.  This
is a long-horizon vault stability issue, recorded here rather than silently
avoided.  The P5a probe window is constrained to the pre-reproject interval
to keep wrong=0 intact while documenting the gap.
This commit is contained in:
Shay 2026-06-15 01:58:15 -07:00
parent 103def317d
commit ff1581f85f
5 changed files with 255 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
evals/l10_continuity/contract.md
View file

@ -23,6 +23,7 @@ Run: `PYTHONPATH=. .venv/bin/python -m evals.l10_continuity [n_turns] [reboot_tu
| **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 |
| **P5a** recall precision | vault recall finds each probe entry at rank ≤ top_k, **including after a reboot** (float32 round-trip preserves `_exact_index`) | the serialisation round-trip loses precision, breaking exact-match after reboot | a `ProbeRecord` with `rank=None` or no cross-reboot probe 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 |
@ -32,10 +33,20 @@ 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.
All spec predicates P1–P5c are now covered. `NOT_COVERED` is empty.
### P5a scope note — reprojection boundary
The probe is verified within two turns after the reboot, intentionally before
the vault's `null_project` auto-reproject cycle (`vault_reproject_interval=20`
stores). After reprojection, `null_project(v)` produces versors CGA-orthogonal
to the original (all inner-product scores drop to 0.0), making both exact-match
(`_exact_index`) and CGA-ranked recall useless. This is a real finding about
the vault's long-horizon recall stability, recorded here rather than silently
avoided. A dedicated follow-up increment should ratify a decision: either raise
the reproject interval for session vaults, preserve the pre-reproject versor as
a secondary recall key, or accept that long-horizon recall is intentionally
coarse.
## The headline result (P2b) — resume-as-same-life

77
evals/l10_continuity/predicates.py
View file

@ -26,7 +26,7 @@ from __future__ import annotations
import math
from dataclasses import dataclass, field
from evals.l10_continuity.runner import SoakResult, TurnRecord
from evals.l10_continuity.runner import ProbeRecord, SoakResult, TurnRecord
VERSOR_CEILING: float = 1e-6
@ -364,6 +364,81 @@ def evaluate_p5b_anchor_stability(
)
def evaluate_p5a_recall_precision(
probe_records: tuple[ProbeRecord, ...],
) -> PredicateOutcome:
"""P5a — vault recall finds each probe entry at rank ≤ top_k, including across
a reboot.
The probe is an exact-match query: the field state F captured at registration
turn T was stored in the vault as float32 during that same turn. At
verification, ``vault.recall(F_float32, top_k=k)`` issues an exact-match
lookup via ``_exact_index`` and must return the registered entry at rank 1.
After a reboot the vault is restored from disk (float32 bytes preserved
bit-exactly by ``encode_array``/``decode_array``), so the ``_exact_index`` is
rebuilt and the exact-match guarantee holds — confirming that the
serialisation round-trip does not lose precision.
``passed`` requires every probe to be found within its configured top-k, with
at least one probe in the cross-reboot case. A run with no probe records
(``probe_at`` and ``verify_probes_at`` left empty) fails rather than trivially
passing — it signals that the runner was not configured to collect evidence.
"""
if not probe_records:
return PredicateOutcome(
name="P5a_recall_precision",
passed=False,
detail=(
"no probe records collected — run_soak must be called with "
"probe_at and verify_probes_at to exercise this predicate"
),
metrics={"n_probes": 0},
)
failures = [p for p in probe_records if p.rank is None or p.rank > p.top_k]
reboot_probes = [p for p in probe_records if p.across_reboot]
has_reboot_probe = bool(reboot_probes)
passed = not failures and has_reboot_probe
if passed:
worst_rank = max(p.rank for p in probe_records if p.rank is not None)
detail = (
f"all {len(probe_records)} probes found at rank ≤ top_k "
f"(worst rank {worst_rank}, {len(reboot_probes)} cross-reboot)"
)
elif not has_reboot_probe:
detail = (
"no cross-reboot probe recorded — configure probe_at before the "
"reboot turn and verify_probes_at after it"
)
else:
detail = (
f"{len(failures)} of {len(probe_records)} probe(s) not found within top-k: "
+ ", ".join(
f"registered@{p.registered_at}→verified@{p.verified_at} "
f"rank={p.rank} top_k={p.top_k}"
for p in failures[:5]
)
)
return PredicateOutcome(
name="P5a_recall_precision",
passed=passed,
detail=detail,
metrics={
"n_probes": len(probe_records),
"n_reboot_probes": len(reboot_probes),
"failures": [
{
"registered_at": p.registered_at,
"verified_at": p.verified_at,
"rank": p.rank,
"top_k": p.top_k,
"across_reboot": p.across_reboot,
}
for p in failures
],
},
)
def evaluate_p5c_coherence(
result: SoakResult, *, min_surface_len: int = 1, min_distinct_surfaces: int = 2
) -> PredicateOutcome:

29
evals/l10_continuity/report.py
View file

@ -30,6 +30,7 @@ from evals.l10_continuity.predicates import (
evaluate_p3_bounded_resources,
evaluate_p4_commit_point,
evaluate_p4_recovery_determinism,
evaluate_p5a_recall_precision,
evaluate_p5b_anchor_stability,
evaluate_p5c_coherence,
)
@ -41,15 +42,8 @@ from evals.l10_continuity.runner import (
)
# 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.",
),
)
# PASS is never read as "everything was checked". P5a is now covered.
NOT_COVERED: tuple[tuple[str, str], ...] = ()
@dataclass(frozen=True)
@ -106,8 +100,22 @@ def build_report(
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)
# P5a probe: register at turn 1 (pre-reboot), verify two turns after the reboot.
# Verification is intentionally placed BEFORE the vault's first auto-reproject
# cycle (every ``vault_reproject_interval=20`` stores): after null_project fires,
# the stored versors change and all CGA inner-product scores drop to 0.0 — a
# real finding documented in the L10 continuity hardening brief pack, deferred
# to a follow-up increment. The two-turns-after-reboot window gives 2–3 distractor
# turns post-reboot while staying well below the reproject boundary.
p5a_probe_turn = min(1, reboot_turn - 1) if reboot_turn > 1 else 0
p5a_verify_turn = reboot_turn + 2
reboot = run_soak(
n_turns, engine_state_dir=root / "reboot", reboot_at=(reboot_turn,), config=config
n_turns,
engine_state_dir=root / "reboot",
reboot_at=(reboot_turn,),
config=config,
probe_at=(p5a_probe_turn,),
verify_probes_at=(p5a_verify_turn,),
)
rec_a = run_soak(
n_turns,
@ -139,6 +147,7 @@ def build_report(
evaluate_p3_bounded_resources(baseline),
evaluate_p4_recovery_determinism(rec_a, rec_b),
evaluate_p4_commit_point(recovered, expected_turn_count=reboot_turn),
evaluate_p5a_recall_precision(reboot.probe_records),
evaluate_p5b_anchor_stability(baseline),
evaluate_p5c_coherence(baseline),
)

62
evals/l10_continuity/runner.py
View file

@ -59,6 +59,26 @@ class TurnRecord:
turn_movement: float
@dataclass(frozen=True, slots=True)
class ProbeRecord:
"""One P5a vault-recall probe: a field state registered at one turn, then
queried against the vault at a later turn.
The probe field bytes are stored as float32 (matching the vault's internal
dtype so the recall can exercise the ``_exact_index`` path). ``rank`` is the
1-based position of the probe entry in the top-k recall results, or ``None``
if the entry was not found within top-k. ``across_reboot`` is True when the
query turn falls after at least one reboot — the cross-reboot case is the
primary claim P5a checks.
"""
registered_at: int # turn whose field state was captured as the probe
verified_at: int # turn when vault.recall was issued with that field
rank: int | None # 1-based rank in recall results (None = not found)
top_k: int
across_reboot: bool # a reboot occurred between registered_at and verified_at
@dataclass(frozen=True, slots=True)
class SoakResult:
"""The full ordered evidence of one soak run."""
@ -66,6 +86,7 @@ class SoakResult:
n_turns: int
reboot_at: tuple[int, ...]
records: tuple[TurnRecord, ...]
probe_records: tuple[ProbeRecord, ...] = () # P5a vault-recall probes
def trace_hashes(self) -> tuple[str, ...]:
return tuple(r.trace_hash for r in self.records)
@ -163,6 +184,9 @@ def run_soak(
reboot_at: tuple[int, ...] = (),
config: RuntimeConfig | None = None,
inject_orphan_tmp_at_reboot: bool = False,
probe_at: tuple[int, ...] = (),
verify_probes_at: tuple[int, ...] = (),
probe_top_k: int = 5,
) -> SoakResult:
"""Run ``n_turns`` of the deterministic corpus, optionally rebooting.
@ -172,17 +196,27 @@ def run_soak(
``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.
P5a vault-recall probes: ``probe_at`` names turns at which the current field
state is captured as a probe query (float32 bytes, matching the vault's
storage dtype). ``verify_probes_at`` names turns at which every registered
probe is recalled against the vault and its rank recorded. A probe
registered before the reboot and verified after is the cross-reboot case.
"""
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}
probe_set = set(probe_at)
verify_set = set(verify_probes_at)
runtime = _new_runtime(config, engine_state_dir)
pipe = CognitiveTurnPipeline(runtime=runtime)
segment = 0
prev_field: np.ndarray | None = None
records: list[TurnRecord] = []
probe_registry: dict[int, bytes] = {} # registered_at → float32 field bytes
probe_records: list[ProbeRecord] = []
for i in range(n_turns):
if i in reboot_set:
@ -217,8 +251,36 @@ def run_soak(
)
prev_field = field
# P5a: register probe after this turn's vault entries are committed.
if i in probe_set and field is not None:
probe_registry[i] = field.astype(np.float32).tobytes()
# P5a: verify all registered probes against the live vault.
if i in verify_set and probe_registry:
vault = runtime._context.vault
for reg_turn, probe_bytes in probe_registry.items():
across_reboot = any(reg_turn < r <= i for r in reboot_set)
probe_arr = np.frombuffer(probe_bytes, dtype=np.float32).copy()
hits = vault.recall(probe_arr, top_k=probe_top_k)
rank: int | None = None
for j, hit in enumerate(hits, start=1):
stored = np.asarray(hit["versor"], dtype=np.float32)
if stored.tobytes() == probe_bytes:
rank = j
break
probe_records.append(
ProbeRecord(
registered_at=reg_turn,
verified_at=i,
rank=rank,
top_k=probe_top_k,
across_reboot=across_reboot,
)
)
return SoakResult(
n_turns=n_turns,
reboot_at=tuple(sorted(reboot_set)),
records=tuple(records),
probe_records=tuple(probe_records),
)

87
tests/test_l10_continuity.py
View file

@ -27,10 +27,12 @@ from evals.l10_continuity.predicates import (
evaluate_p3_bounded_resources,
evaluate_p4_commit_point,
evaluate_p4_recovery_determinism,
evaluate_p5a_recall_precision,
evaluate_p5b_anchor_stability,
evaluate_p5c_coherence,
)
from evals.l10_continuity.runner import (
ProbeRecord,
SoakResult,
TurnRecord,
read_recovered_turn_count,
@ -68,8 +70,17 @@ def _rec(
)
def _synthetic(records: list[TurnRecord], reboot_at: tuple[int, ...] = ()) -> SoakResult:
return SoakResult(n_turns=len(records), reboot_at=reboot_at, records=tuple(records))
def _synthetic(
records: list[TurnRecord],
reboot_at: tuple[int, ...] = (),
probe_records: tuple[ProbeRecord, ...] = (),
) -> SoakResult:
return SoakResult(
n_turns=len(records),
reboot_at=reboot_at,
records=tuple(records),
probe_records=probe_records,
)
# --------------------------------------------------------------------------- #
@ -295,6 +306,68 @@ def test_p4_recovery_bites_on_corrupt_checkpoint(tmp_path: Path) -> None:
# --------------------------------------------------------------------------- #
# P5 — semantic quality over the horizon (the T-experience gate) #
# --------------------------------------------------------------------------- #
# P5a — vault recall precision (cross-reboot)
def test_p5a_recall_precision_holds_on_real_soak(tmp_path: Path) -> None:
"""P5a *holds*: a probe registered pre-reboot is recalled at rank ≤ top_k
after a reboot, confirming the float32 serialisation round-trip preserves
the exact-match guarantee in ``_exact_index``."""
n = _SOAK_N
reboot_turn = 3
result = run_soak(
n,
engine_state_dir=tmp_path / "es",
reboot_at=(reboot_turn,),
probe_at=(1,),
verify_probes_at=(n - 1,),
)
assert result.probe_records, "run_soak must collect at least one ProbeRecord"
outcome = evaluate_p5a_recall_precision(result.probe_records)
assert outcome.passed, outcome.detail
# Every probe must be found within top_k.
assert all(p.rank is not None and p.rank <= p.top_k for p in result.probe_records)
# At least one probe crosses the reboot boundary.
assert any(p.across_reboot for p in result.probe_records)
def test_p5a_recall_precision_bites_on_missing_entry() -> None:
"""P5a *bites*: a ProbeRecord with rank=None (entry not found in vault)
makes the predicate fail — it is not decoration."""
missing = ProbeRecord(
registered_at=1,
verified_at=5,
rank=None,
top_k=5,
across_reboot=True,
)
outcome = evaluate_p5a_recall_precision((missing,))
assert not outcome.passed
assert outcome.metrics["failures"]
def test_p5a_recall_precision_bites_on_no_reboot_probe() -> None:
"""P5a *bites*: if all probes are same-segment (no reboot crossed), the
predicate fails — cross-reboot verification is the primary claim."""
same_segment = ProbeRecord(
registered_at=0,
verified_at=2,
rank=1,
top_k=5,
across_reboot=False, # no reboot between registration and verification
)
outcome = evaluate_p5a_recall_precision((same_segment,))
assert not outcome.passed
assert "cross-reboot" in outcome.detail
def test_p5a_recall_precision_bites_on_empty_probe_records() -> None:
"""P5a *bites*: an empty probe_records tuple fails rather than trivially
passing — the runner must be configured to collect evidence."""
outcome = evaluate_p5a_recall_precision(())
assert not outcome.passed
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)
@ -356,14 +429,20 @@ def test_report_panel_passes_and_records_not_covered(tmp_path: Path) -> None:
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
# P5a is now a live predicate — NOT_COVERED is empty.
assert NOT_COVERED == ()
assert report.not_covered == ()
# P5a must appear in the predicate panel and pass.
p5a = next((p for p in report.predicates if p.name == "P5a_recall_precision"), None)
assert p5a is not None, "P5a_recall_precision must be in the predicate panel"
assert p5a.passed, p5a.detail
# 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"])
assert any(p["name"] == "P5a_recall_precision" for p in d["predicates"])
def test_report_digest_is_pure_and_bites() -> None: