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

Merge pull request #757 from AssetOverflow/codex/l10-heartbeat-soak

Browse source 
test(l10): falsifiable long-horizon soak for the always-on idle heartbeat
This commit is contained in:
Shay 2026-06-14 18:30:35 -07:00 committed by GitHub
commit ee970d8306
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 740 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
evals/l10_always_on/__init__.py Normal file
View file

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

33
evals/l10_always_on/__main__.py Normal file
View file

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

51
evals/l10_always_on/contract.md Normal file
View file

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

192
evals/l10_always_on/predicates.py Normal file
View file

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

110
evals/l10_always_on/report.py Normal file
View file

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

157
evals/l10_always_on/runner.py Normal file
View file

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

185
tests/test_l10_always_on_soak.py Normal file
View file

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