test(l10): falsifiable long-horizon soak for the always-on IDLE heartbeat

The always-on PROCESS shipped (run_continuous + daemon + CLI), but the claim "it holds
over uptime" was UNFALSIFIED: the existing evals/l10_continuity soak drives the TURN loop
and is disjoint from run_continuous; the daemon path was covered only by ≤5-beat unit
tests. This lane converts "built" → "proven over horizon" for the idle path.

evals/l10_always_on/ mirrors the l10_continuity predicate/mutation harness but drives the
IDLE heartbeat: it seeds a real continuous life (held self + a cognitive turn to excite the
field), runs N beats with NO user turn, and gates four falsifiable predicates, each with a
*_holds (real soak) AND a *_bites (mutation) test per CLAUDE.md schema-as-proof:

- H1 closure — every observed idle beat versor_condition < 1e-6 (held over uptime, READ
  never repaired); bites on a breached beat AND on a vacuous no-field soak.
- H2 bounded idle — a no-work beat adds NOTHING to the vault (no idle leak — invisible at
  5 beats, fatal at 100k); bites on idle vault growth.
- H3 convergence — a saturated idle life SETTLES and stays settled (no re-awakening), at
  rest at the end, closure intact on the tail; bites on never-settling + on re-awakening.
- H4 reboot resume — a reboot mid-soak resumes the SAME life (strict identity guard passes,
  pre-reboot DERIVED learning survives, post-reboot closure holds); bites on failed
  resume + on lost learning.

Measured (5000 idle beats, reboot@2500, ~20s): ALL gates pass — versor_condition flat at
1.389e-07 (no drift, no repair), vault bounded at 6 (no idle leak), converged at beat 1
with a 4999-beat at-rest tail, reboot resumed the same life with learning intact. This is
the empirical resolution of the L10 riskiest-unknown FOR THE IDLE PATH (the closure-by-
construction ruling covered the field-transition walk; this covers indefinite idle uptime).

Honestly NOT covered (recorded in the report, no silent skip): the continuously-LEARNING
life's resource cost under a sustained new-fact stream (O(n²) snapshot; per-run lived_life)
— out of scope until an afferent/intake feed + incremental persistence exist (the next
frontier).

Run on demand: PYTHONPATH=. python -m evals.l10_always_on [n_beats] [reboot_beat]. Path-
scoped out of CI smoke (like l10_continuity); runs in post-merge full-pytest. Additive — a
new eval lane, no existing files touched. 12/12 lane tests + architectural invariants green.

evals/l10_always_on/__init__.py

@@ -0,0 +1,12 @@
+"""L10 always-on heartbeat soak — the falsifiable long-horizon gate for the IDLE path.
+
+The continuity lane (``evals/l10_continuity``) soaks the TURN loop. This lane soaks the
+IDLE heartbeat (``chat/always_on.run_continuous``): it seeds a real continuous life, then
+drives the engine over many beats with NO user turn and evaluates falsifiable predicates
+over the per-beat evidence — closure holds over indefinite idle uptime (read, never
+repaired), idle resources stay bounded, the saturated life converges to rest, and a reboot
+mid-soak resumes the SAME life.
+
+Not in default smoke (it is a soak; run on demand / nightly):
+``PYTHONPATH=. .venv/bin/python -m evals.l10_always_on [n_beats] [reboot_beat]``
+"""

evals/l10_always_on/__main__.py

@@ -0,0 +1,33 @@
+"""On-demand entrypoint for the L10 always-on heartbeat soak panel.
+
+Run:  PYTHONPATH=. .venv/bin/python -m evals.l10_always_on [n_beats] [reboot_beat]
+
+Drives the IDLE heartbeat over a seeded continuous life for ``n_beats`` beats (default 24;
+pass a large N — e.g. 100000 — for a true long-horizon soak), prints the structured report
+as JSON, and exits non-zero if any gate fails. NOT in the default smoke suite — a soak, run
+on demand / nightly. ``deterministic_digest`` is the freeze handle.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+from evals.l10_always_on.report import build_report
+
+
+def main(argv: list[str]) -> int:
+    n_beats = int(argv[0]) if len(argv) > 0 else 24
+    reboot_beat = int(argv[1]) if len(argv) > 1 else max(1, n_beats // 2)
+    with tempfile.TemporaryDirectory(prefix="l10_always_on_") as tmp:
+        report = build_report(
+            n_beats=n_beats, reboot_beat=reboot_beat, 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:]))

evals/l10_always_on/contract.md

@@ -0,0 +1,51 @@
+# L10 Always-On Heartbeat Soak — Contract
+
+**Status:** soak (falsifiable long-horizon gate) · **Not in default smoke** (run on demand / nightly).
+
+The continuity lane (`evals/l10_continuity`) soaks the **turn loop**. This lane soaks the
+**idle heartbeat** (`chat/always_on.run_continuous` — the loop the `core always-on` daemon
+drives). It seeds a real continuous life (a held self + a cognitive turn to excite the
+field), then runs the engine over N beats with **no user turn**, and evaluates falsifiable
+predicates over the per-beat evidence. It converts the claim *"the always-on process is
+built"* into *"the always-on process holds over uptime"* — the idle-path claims that the
+short daemon unit tests (≤5 beats) and the turn-loop soak (disjoint from `run_continuous`)
+do **not** prove at horizon.
+
+Run: `PYTHONPATH=. .venv/bin/python -m evals.l10_always_on [n_beats] [reboot_beat]`
+(pass a large `n_beats` — e.g. `100000` — for a true long-horizon soak; default `24`).
+
+## Predicates
+
+| ID | Proves | Fails loudly when | Mutation-verified bite |
+|----|--------|-------------------|------------------------|
+| **H1** closure | every OBSERVED idle beat has `versor_condition < 1e-6` (closure holds over idle uptime, READ never repaired) | the idle heartbeat drifts/corrupts the field, or the field never existed (vacuous) | a beat with `versor_condition ≥ 1e-6`; an all-`None` (no-field) soak |
+| **H2** bounded idle | a `did_work=False` beat adds NOTHING to the vault (no idle resource leak) | a converged idle life keeps growing a store — invisible at 5 beats, fatal at 100k | an idle beat whose `vault_size` grew over the prior beat |
+| **H3** convergence | a saturated idle life SETTLES and stays settled (no re-awakening), at rest at the end, closure intact on the tail | the life churns forever, thrashes (work-after-rest), or breaks closure once settled | a never-settling run; a rest→work re-awakening |
+| **H4** reboot resume | a reboot mid-soak resumes the SAME life (strict identity guard passes, pre-reboot DERIVED learning survives, post-reboot closure holds) | a reboot forks a new life, loses learning, or breaks closure | `resumed_cleanly=False`; `learned_fact_survived=False` |
+
+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.
+
+## The measured result
+
+On a 5000-beat soak (reboot at 2500): **all gates pass.** `versor_condition` is flat at
+`1.389e-07` across all 5000 beats (no drift — the idle heartbeat never perturbs the field,
+no repair), the vault stays bounded at 6 entries (no idle leak), the life converges at
+beat 1 and the 4999-beat tail stays at rest with closure intact, and the reboot at 2500
+resumes the same life with its derived learning intact. This is the empirical resolution of
+the L10 riskiest-unknown **for the idle path** (the closure-by-construction ruling covered
+the field-transition walk; this covers indefinite idle uptime).
+
+## Not covered (no silent skips)
+
+- **H5 — learning-life resource cost.** This lane proves the **idle (converged)** life is
+  resource-bounded. The cost of a continuously-**learning** life under a sustained
+  new-fact stream — the full-snapshot checkpoint is O(n²) in facts, `lived_life.json` is
+  per-run — is out of scope until an afferent/intake feed and incremental persistence
+  exist. Recorded as `not_covered` in the report; a follow-up.
+
+The `deterministic_digest` in the report freezes the per-beat shape (`did_work` /
+`field_valid` / learning counts / vault size — all deterministic) + the verdicts, excluding
+the machine-variant raw `versor_condition` float. Pin it once the lane is trusted so a
+regression flips it.

evals/l10_always_on/predicates.py

@@ -0,0 +1,192 @@
+"""Pure pass/fail predicates over heartbeat-soak evidence — the falsifiable gates.
+
+Each predicate is a pure function of ``HeartbeatSoakResult`` evidence (it runs no beats and
+mutates nothing), so each can be mutation-verified to *bite*. The idle-path claims today's
+always-on work makes, none of which the short daemon unit tests or the turn-loop soak prove
+at horizon:
+
+- **H1 closure** — every OBSERVED idle beat satisfies ``versor_condition < 1e-6``. The L10
+  riskiest-unknown for the IDLE path: the heartbeat holds closure over long uptime WITHOUT
+  any repair (it only reads ``versor_condition``).
+- **H2 bounded idle** — a beat that does NO work adds NOTHING to the vault (no idle leak); a
+  growing store on an idle beat is a resource leak that only manifests at horizon.
+- **H3 convergence** — a saturated idle life SETTLES: once it stops working it stays at rest
+  (no re-awakening) and the final beat is at rest — a continuously-IDLING steady state, not
+  a churning or thrashing one — and closure still holds across the converged tail.
+- **H4 reboot resume** — a reboot mid-soak resumes the SAME life: the reconstruct under the
+  strict identity guard succeeds, the pre-reboot DERIVED learning survives, and post-reboot
+  closure holds.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from evals.l10_always_on.runner import HeartbeatSoakResult
+
+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_h1_closure(
+    result: HeartbeatSoakResult, *, ceiling: float = VERSOR_CEILING
+) -> PredicateOutcome:
+    """H1 — every observed idle beat is a valid versor (``versor_condition < ceiling``).
+
+    Requires at least one OBSERVED beat (a real field) so the gate is non-vacuous — a soak
+    where the field never existed cannot pass H1 by saying nothing."""
+    observed = result.observed()
+    violations = [
+        (r.beat_index, r.versor_condition)
+        for r in observed
+        if not (r.versor_condition is not None and r.versor_condition < ceiling)
+    ]
+    worst = max((r.versor_condition for r in observed if r.versor_condition is not None), default=0.0)
+    passed = bool(observed) and not violations
+    if not observed:
+        detail = "no beat observed a field — closure is vacuous, not held"
+    elif passed:
+        detail = f"all {len(observed)} observed beats closed (worst={worst:.3e} < {ceiling:.0e})"
+    else:
+        detail = f"{len(violations)} idle beat(s) breached the versor ceiling: {violations[:5]}"
+    return PredicateOutcome(
+        name="H1_closure",
+        passed=passed,
+        detail=detail,
+        metrics={"observed_beats": len(observed), "worst_versor_condition": worst, "violations": violations},
+    )
+
+
+def evaluate_h2_bounded_idle(result: HeartbeatSoakResult) -> PredicateOutcome:
+    """H2 — a no-work idle beat adds nothing to the vault (no idle resource leak).
+
+    A ``did_work=False`` beat that GROWS the vault over the previous beat is an idle leak —
+    the kind that is invisible at 5 beats and fatal at 100k. Work beats (consolidation)
+    legitimately grow it and are exempt."""
+    records = result.records
+    leaks = [
+        (r.beat_index, prev.vault_size, r.vault_size)
+        for prev, r in zip(records, records[1:])
+        if not r.did_work and r.vault_size > prev.vault_size
+    ]
+    monotonic = all(b.vault_size >= a.vault_size for a, b in zip(records, records[1:]))
+    passed = not leaks
+    final = records[-1].vault_size if records else 0
+    detail = (
+        f"no idle beat grew the vault (final size {final} over {len(records)} beats, monotonic={monotonic})"
+        if passed
+        else f"idle resource leak: {len(leaks)} no-work beat(s) grew the vault: {leaks[:5]}"
+    )
+    return PredicateOutcome(
+        name="H2_bounded_idle",
+        passed=passed,
+        detail=detail,
+        metrics={"idle_leaks": leaks, "final_vault_size": final, "vault_monotonic": monotonic},
+    )
+
+
+def evaluate_h3_convergence(
+    result: HeartbeatSoakResult, *, min_converged_tail: int = 2
+) -> PredicateOutcome:
+    """H3 — a saturated idle life SETTLES and stays settled (no re-awakening), at rest at
+    the end, with closure intact across the converged tail.
+
+    Three failure modes: it never settles (the final beat still works — churns forever); it
+    re-awakens (a ``did_work=True`` beat after it had gone to rest — a nondeterministic idle
+    leak); or closure breaks on the converged tail. The converged tail must be at least
+    ``min_converged_tail`` beats so 'settled' is observed, not assumed."""
+    records = result.records
+    if not records:
+        return PredicateOutcome("H3_convergence", False, "no beats to evaluate", {})
+
+    rest_indices = [i for i, r in enumerate(records) if not r.did_work]
+    if not rest_indices:
+        return PredicateOutcome(
+            "H3_convergence", False, "the life never went to rest (every beat did work)", {}
+        )
+    convergence_at = rest_indices[0]
+    tail = records[convergence_at:]
+    reawakenings = [r.beat_index for r in tail if r.did_work]
+    final_at_rest = not records[-1].did_work
+    closure_breaks = [r.beat_index for r in tail if not r.field_valid]
+    long_enough = len(tail) >= min_converged_tail
+    passed = not reawakenings and final_at_rest and not closure_breaks and long_enough
+
+    if passed:
+        detail = (
+            f"converged at beat {convergence_at}; the {len(tail)}-beat tail stayed at rest "
+            f"with closure intact"
+        )
+    else:
+        cause = []
+        if reawakenings:
+            cause.append(f"RE-AWAKENED at {reawakenings[:5]}")
+        if not final_at_rest:
+            cause.append("never settled (final beat still working)")
+        if closure_breaks:
+            cause.append(f"closure broke on the tail at {closure_breaks[:5]}")
+        if not long_enough:
+            cause.append(f"converged tail too short ({len(tail)} < {min_converged_tail})")
+        detail = "; ".join(cause)
+    return PredicateOutcome(
+        name="H3_convergence",
+        passed=passed,
+        detail=detail,
+        metrics={
+            "convergence_beat": convergence_at,
+            "converged_tail_len": len(tail),
+            "reawakenings": reawakenings,
+        },
+    )
+
+
+def evaluate_h4_reboot_resume(result: HeartbeatSoakResult) -> PredicateOutcome:
+    """H4 — a reboot mid-soak resumes the SAME life.
+
+    Three obligations: the reconstruct under the strict identity guard succeeded
+    (``resumed_cleanly`` — a different-life checkpoint would have raised
+    ``IdentityContinuityError``); the pre-reboot DERIVED learning survived
+    (``learned_fact_survived`` — recalled post-reboot, not merely re-derivable); and
+    post-reboot closure holds (every segment>0 beat is a valid versor)."""
+    if not result.reboot_at:
+        raise ValueError("H4 expects a soak with a reboot leg (reboot_at non-empty).")
+    post = result.post_reboot_records()
+    post_closure_breaks = [
+        r.beat_index for r in post if r.versor_condition is not None and not r.field_valid
+    ]
+    passed = (
+        result.resumed_cleanly
+        and result.learned_fact_survived is True
+        and not post_closure_breaks
+    )
+    if passed:
+        detail = (
+            f"reboot at {result.reboot_at[0]} resumed the SAME life: identity guard passed, "
+            f"learning survived, {len(post)} post-reboot beats closed"
+        )
+    else:
+        cause = []
+        if not result.resumed_cleanly:
+            cause.append("reconstruct RAISED IdentityContinuityError (not the same life)")
+        if result.learned_fact_survived is not True:
+            cause.append(f"pre-reboot learning did NOT survive (={result.learned_fact_survived})")
+        if post_closure_breaks:
+            cause.append(f"post-reboot closure broke at {post_closure_breaks[:5]}")
+        detail = "; ".join(cause)
+    return PredicateOutcome(
+        name="H4_reboot_resume",
+        passed=passed,
+        detail=detail,
+        metrics={
+            "resumed_cleanly": result.resumed_cleanly,
+            "learned_fact_survived": result.learned_fact_survived,
+            "post_reboot_beats": len(post),
+        },
+    )

evals/l10_always_on/report.py

@@ -0,0 +1,110 @@
+"""Assemble the L10 always-on heartbeat panel into a freeze-gateable report.
+
+The panel runs the soaks the predicates need (an uninterrupted idle baseline for H1/H2/H3
+and a reboot leg for H4), 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 per-beat SHAPE
+(``did_work`` / ``field_valid`` / learning counts / vault size — all deterministic ints &
+bools) and each predicate's ``(name, passed)`` verdict. It deliberately EXCLUDES the raw
+``versor_condition`` float (machine-variant; only the ``field_valid`` comparison against the
+ceiling is stable). Pin the digest 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_always_on.predicates import (
+    PredicateOutcome,
+    evaluate_h1_closure,
+    evaluate_h2_bounded_idle,
+    evaluate_h3_convergence,
+    evaluate_h4_reboot_resume,
+)
+from evals.l10_always_on.runner import HeartbeatSoakResult, run_heartbeat_soak
+
+# Legs this lane does NOT cover, recorded so a PASS is never read as "everything checked".
+NOT_COVERED: tuple[tuple[str, str], ...] = (
+    (
+        "H5_learning_life_resource_cost",
+        "This lane proves the IDLE (converged) life is resource-bounded. The cost of a "
+        "continuously-LEARNING life under a sustained NEW-fact stream (the full-snapshot "
+        "checkpoint is O(n^2) in facts; lived_life.json is per-run) is out of scope until "
+        "an afferent/intake feed and incremental persistence exist; deferred.",
+    ),
+)
+
+
+@dataclass(frozen=True)
+class L10AlwaysOnReport:
+    n_beats: int
+    reboot_beat: 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_beats": self.n_beats,
+            "reboot_beat": self.reboot_beat,
+            "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: HeartbeatSoakResult, predicates: tuple[PredicateOutcome, ...]
+) -> str:
+    """SHA-256 over hardware-stable evidence: the per-beat shape + the verdicts."""
+    payload = {
+        "beat_shape": [
+            [r.did_work, r.field_valid, r.facts_consolidated, r.proposals_created, r.vault_size]
+            for r in baseline.records
+        ],
+        "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_beats: int = 24,
+    reboot_beat: int = 12,
+    engine_state_root: Path,
+    config: RuntimeConfig | None = None,
+) -> L10AlwaysOnReport:
+    """Run the full idle panel and assemble the report.
+
+    Soaks: an uninterrupted idle ``baseline`` (H1/H2/H3) and a ``reboot`` leg (H4)."""
+    root = engine_state_root
+    baseline = run_heartbeat_soak(n_beats, engine_state_dir=root / "baseline", config=config)
+    reboot = run_heartbeat_soak(
+        n_beats, engine_state_dir=root / "reboot", reboot_at=(reboot_beat,), config=config
+    )
+    predicates: tuple[PredicateOutcome, ...] = (
+        evaluate_h1_closure(baseline),
+        evaluate_h2_bounded_idle(baseline),
+        evaluate_h3_convergence(baseline),
+        evaluate_h4_reboot_resume(reboot),
+    )
+    return L10AlwaysOnReport(
+        n_beats=n_beats,
+        reboot_beat=reboot_beat,
+        predicates=predicates,
+        not_covered=NOT_COVERED,
+        deterministic_digest=deterministic_digest(baseline, predicates),
+    )

evals/l10_always_on/runner.py

@@ -0,0 +1,157 @@
+"""The L10 always-on heartbeat soak runner — drives the REAL idle loop over N beats.
+
+It seeds a continuous life (a held self + a cognitive turn to excite the field), then runs
+``chat/always_on.run_continuous`` over a fresh ``ChatRuntime`` whose checkpoint lives in a
+caller-supplied dir. Optionally it injects a *reboot leg*: at a chosen beat it drops the
+live runtime and reconstructs one from the on-disk checkpoint — the always-on lifecycle
+("resume as the same life") — under the strict identity guard.
+
+Pure instrumentation: it records per-beat evidence (``versor_condition``, ``field_valid``,
+the learning counts, ``did_work``, vault size, boot segment) and returns it. It makes NO
+pass/fail judgement (that is ``predicates.py``) and NEVER repairs or normalizes the field
+(it reads only what the heartbeat produced — CLAUDE.md no-hot-path-repair).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from chat.always_on import HeartbeatRecord, run_continuous
+from chat.always_on_daemon import continuous_life_config
+from chat.runtime import ChatRuntime
+from core.config import RuntimeConfig
+from core.engine_identity import IdentityContinuityError, engine_identity_for_config
+from engine_state import get_git_revision
+
+
+@dataclass(frozen=True, slots=True)
+class BeatRecord:
+    """Per-beat evidence captured from the real idle heartbeat (no judgement)."""
+
+    beat_index: int  # global, across reboot segments
+    segment_tick: int  # the run_continuous tick within this boot segment
+    versor_condition: float | None  # None before any field exists
+    field_valid: bool
+    facts_consolidated: int
+    proposals_created: int
+    pending_proposals: int
+    did_work: bool
+    vault_size: int
+    booted_segment: int
+
+
+@dataclass(frozen=True, slots=True)
+class HeartbeatSoakResult:
+    """The full ordered evidence of one idle soak run."""
+
+    n_beats: int
+    reboot_at: tuple[int, ...]
+    records: tuple[BeatRecord, ...]
+    identity: str
+    # Reboot-leg evidence (only meaningful when reboot_at is non-empty):
+    resumed_cleanly: bool  # the reconstruct under the strict identity guard did not raise
+    learned_fact_survived: bool | None  # a pre-reboot DERIVED fact is recalled post-reboot
+
+    def observed(self) -> tuple[BeatRecord, ...]:
+        return tuple(r for r in self.records if r.versor_condition is not None)
+
+    def post_reboot_records(self) -> tuple[BeatRecord, ...]:
+        return tuple(r for r in self.records if r.booted_segment > 0)
+
+
+def _seed_life(runtime: ChatRuntime) -> None:
+    """Seed a consolidatable held self + excite the field so closure is OBSERVED.
+
+    member(socrates, man) + subset(man, mortal) — from which the idle heartbeat DERIVES
+    member(socrates, mortal) — plus a real cognitive turn so ``versor_condition`` is
+    observable (an idle life never excites its own field)."""
+    from core.cognition.pipeline import CognitiveTurnPipeline
+    from generate.meaning_graph.reader import comprehend
+    from generate.realize import realize_comprehension
+
+    ctx = runtime._context
+    realize_comprehension(comprehend("Socrates is a man."), ctx)
+    realize_comprehension(comprehend("All men are mortals."), ctx)
+    CognitiveTurnPipeline(runtime=runtime).run("Socrates is a man.")
+
+
+def _mortal_is_stored(runtime: ChatRuntime) -> bool:
+    """True iff member(socrates, mortal) is a STORED realized record (recall, not redo)."""
+    from generate.realize import recall_realized
+
+    return any(
+        f.relation_arguments[1] == "mortal"
+        for f in recall_realized(runtime._context, subject="socrates", predicate="member")
+    )
+
+
+def run_heartbeat_soak(
+    n_beats: int,
+    *,
+    engine_state_dir: Path,
+    reboot_at: tuple[int, ...] = (),
+    config: RuntimeConfig | None = None,
+    seed: bool = True,
+) -> HeartbeatSoakResult:
+    """Run ``n_beats`` idle heartbeats, optionally rebooting at given beat boundaries.
+
+    The config is forced to the continuous-life config (persist + consolidate + strict
+    identity) — the daemon's contract. ``reboot_at`` beats split the soak into boot
+    segments: before each, the live runtime is dropped and reconstructed from the
+    checkpoint (the reboot). A reboot at beat 0 is meaningless and ignored.
+    """
+    if n_beats < 0:
+        raise ValueError(f"n_beats must be non-negative, got {n_beats}")
+    config = continuous_life_config(config)
+    reboot_set = sorted({i for i in reboot_at if 0 < i < n_beats})
+    boundaries = [0, *reboot_set, n_beats]
+
+    runtime = ChatRuntime(config=config, engine_state_path=engine_state_dir)
+    if seed:
+        _seed_life(runtime)
+    identity = engine_identity_for_config(config, get_git_revision())
+    records: list[BeatRecord] = []
+    resumed_cleanly = True
+    learned_fact_survived: bool | None = None
+
+    for seg, (start, end) in enumerate(zip(boundaries, boundaries[1:])):
+        if seg > 0:
+            # The reboot: reconstruct from the prior segment's checkpoint. Under the strict
+            # identity guard this RAISES if the checkpoint is a different life.
+            try:
+                runtime = ChatRuntime(config=config, engine_state_path=engine_state_dir)
+            except IdentityContinuityError:
+                resumed_cleanly = False
+                break
+            if learned_fact_survived is None:
+                learned_fact_survived = _mortal_is_stored(runtime)
+
+        rt = runtime
+
+        def _capture(record: HeartbeatRecord, *, _seg: int = seg, _rt: ChatRuntime = rt) -> None:
+            records.append(
+                BeatRecord(
+                    beat_index=len(records),
+                    segment_tick=record.tick,
+                    versor_condition=record.versor_condition,
+                    field_valid=record.field_valid,
+                    facts_consolidated=record.facts_consolidated,
+                    proposals_created=record.proposals_created,
+                    pending_proposals=record.pending_proposals,
+                    did_work=record.did_work,
+                    vault_size=len(_rt._context.vault),
+                    booted_segment=_seg,
+                )
+            )
+
+        run_continuous(rt, heartbeats=end - start, on_heartbeat=_capture)
+
+    return HeartbeatSoakResult(
+        n_beats=n_beats,
+        reboot_at=tuple(reboot_set),
+        records=tuple(records),
+        identity=identity,
+        resumed_cleanly=resumed_cleanly,
+        learned_fact_survived=learned_fact_survived,
+    )

tests/test_l10_always_on_soak.py

@@ -0,0 +1,185 @@
+"""L10 always-on heartbeat soak — the falsifiable long-horizon gate for the IDLE path.
+
+Per predicate, two tests (CLAUDE.md schema-as-proof):
+- a ``*_holds`` test that drives the REAL idle heartbeat over a short soak and asserts the
+  predicate passes on genuine evidence, and
+- a ``*_bites`` test that feeds the predicate mutated evidence and asserts it FAILS.
+
+These soak a seeded continuous life over the idle heartbeat (no user turn). Short N + a tmp
+checkpoint dir; NOT in the default smoke suite (a soak — run on demand / nightly). The long
+horizon is the ``python -m evals.l10_always_on`` CLI's job.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from evals.l10_always_on.predicates import (
+    VERSOR_CEILING,
+    evaluate_h1_closure,
+    evaluate_h2_bounded_idle,
+    evaluate_h3_convergence,
+    evaluate_h4_reboot_resume,
+)
+from evals.l10_always_on.runner import (
+    BeatRecord,
+    HeartbeatSoakResult,
+    run_heartbeat_soak,
+)
+
+_SOAK_N = 12  # beat 0 learns the derived fact; beats 1+ converge to rest
+
+
+# --------------------------------------------------------------------------- #
+# Synthetic-evidence helpers (fast; no heartbeat) — used by the *_bites tests.  #
+# --------------------------------------------------------------------------- #
+def _beat(
+    i: int,
+    *,
+    versor_condition: float | None = 8.2e-13,
+    field_valid: bool = True,
+    did_work: bool = False,
+    vault_size: int = 2,
+    segment: int = 0,
+) -> BeatRecord:
+    return BeatRecord(
+        beat_index=i,
+        segment_tick=i,
+        versor_condition=versor_condition,
+        field_valid=field_valid,
+        facts_consolidated=1 if did_work else 0,
+        proposals_created=0,
+        pending_proposals=0,
+        did_work=did_work,
+        vault_size=vault_size,
+        booted_segment=segment,
+    )
+
+
+def _soak(
+    records: list[BeatRecord],
+    *,
+    reboot_at: tuple[int, ...] = (),
+    resumed_cleanly: bool = True,
+    learned_fact_survived: bool | None = True,
+) -> HeartbeatSoakResult:
+    return HeartbeatSoakResult(
+        n_beats=len(records),
+        reboot_at=reboot_at,
+        records=tuple(records),
+        identity="sha256:soak",
+        resumed_cleanly=resumed_cleanly,
+        learned_fact_survived=learned_fact_survived,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# H1 — closure over idle uptime                                                 #
+# --------------------------------------------------------------------------- #
+def test_h1_closure_holds_on_real_soak(tmp_path: Path) -> None:
+    result = run_heartbeat_soak(_SOAK_N, engine_state_dir=tmp_path / "es")
+    outcome = evaluate_h1_closure(result)
+    assert outcome.passed, outcome.detail
+    assert outcome.metrics["observed_beats"] >= 1  # the field really existed (non-vacuous)
+    assert outcome.metrics["worst_versor_condition"] < VERSOR_CEILING
+
+
+def test_h1_bites_on_breached_versor() -> None:
+    bad = _soak([_beat(0), _beat(1, versor_condition=1e-3, field_valid=False), _beat(2)])
+    outcome = evaluate_h1_closure(bad)
+    assert not outcome.passed
+    assert (1, 1e-3) in outcome.metrics["violations"]
+
+
+def test_h1_bites_on_no_field_observed() -> None:
+    # A life whose field never existed must NOT pass H1 vacuously.
+    vacuous = _soak([_beat(0, versor_condition=None), _beat(1, versor_condition=None)])
+    outcome = evaluate_h1_closure(vacuous)
+    assert not outcome.passed
+    assert "vacuous" in outcome.detail
+
+
+# --------------------------------------------------------------------------- #
+# H2 — bounded idle (no leak on no-work beats)                                   #
+# --------------------------------------------------------------------------- #
+def test_h2_bounded_idle_holds_on_real_soak(tmp_path: Path) -> None:
+    result = run_heartbeat_soak(_SOAK_N, engine_state_dir=tmp_path / "es")
+    outcome = evaluate_h2_bounded_idle(result)
+    assert outcome.passed, outcome.detail
+
+
+def test_h2_bites_on_idle_vault_growth() -> None:
+    # A no-work beat that grows the vault is an idle resource leak.
+    leaky = _soak(
+        [
+            _beat(0, did_work=True, vault_size=3),
+            _beat(1, did_work=False, vault_size=7),  # idle but grew 3 -> 7
+            _beat(2, did_work=False, vault_size=7),
+        ]
+    )
+    outcome = evaluate_h2_bounded_idle(leaky)
+    assert not outcome.passed
+    assert outcome.metrics["idle_leaks"] and outcome.metrics["idle_leaks"][0][0] == 1
+
+
+# --------------------------------------------------------------------------- #
+# H3 — convergence (settles and stays settled)                                  #
+# --------------------------------------------------------------------------- #
+def test_h3_convergence_holds_on_real_soak(tmp_path: Path) -> None:
+    result = run_heartbeat_soak(_SOAK_N, engine_state_dir=tmp_path / "es")
+    outcome = evaluate_h3_convergence(result)
+    assert outcome.passed, outcome.detail
+    assert outcome.metrics["converged_tail_len"] >= 2
+    assert outcome.metrics["reawakenings"] == []
+
+
+def test_h3_bites_on_never_settling() -> None:
+    churning = _soak([_beat(i, did_work=True, vault_size=2 + i) for i in range(4)])
+    outcome = evaluate_h3_convergence(churning)
+    assert not outcome.passed
+
+
+def test_h3_bites_on_reawakening() -> None:
+    # Went to rest, then a beat did work again — a nondeterministic idle re-awakening.
+    reawaken = _soak(
+        [_beat(0, did_work=False), _beat(1, did_work=True, vault_size=3), _beat(2, did_work=False)]
+    )
+    outcome = evaluate_h3_convergence(reawaken)
+    assert not outcome.passed
+    assert 1 in outcome.metrics["reawakenings"]
+
+
+# --------------------------------------------------------------------------- #
+# H4 — reboot mid-soak resumes the SAME life                                    #
+# --------------------------------------------------------------------------- #
+def test_h4_reboot_resume_holds_on_real_soak(tmp_path: Path) -> None:
+    result = run_heartbeat_soak(_SOAK_N, engine_state_dir=tmp_path / "es", reboot_at=(6,))
+    outcome = evaluate_h4_reboot_resume(result)
+    assert outcome.passed, outcome.detail
+    assert result.resumed_cleanly is True
+    assert result.learned_fact_survived is True  # the pre-reboot DERIVED fact was recalled
+
+
+def test_h4_bites_on_failed_resume() -> None:
+    broke = _soak([_beat(0, segment=0), _beat(1, segment=1)], reboot_at=(1,), resumed_cleanly=False)
+    outcome = evaluate_h4_reboot_resume(broke)
+    assert not outcome.passed
+    assert "IdentityContinuityError" in outcome.detail
+
+
+def test_h4_bites_on_lost_learning() -> None:
+    lost = _soak(
+        [_beat(0, segment=0), _beat(1, segment=1)],
+        reboot_at=(1,),
+        learned_fact_survived=False,
+    )
+    outcome = evaluate_h4_reboot_resume(lost)
+    assert not outcome.passed
+    assert "did NOT survive" in outcome.detail
+
+
+def test_h4_requires_a_reboot_leg() -> None:
+    with pytest.raises(ValueError):
+        evaluate_h4_reboot_resume(_soak([_beat(0)]))