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

feat(l10): always-on heartbeat — the loop that makes the life continuous (T-experience spine)

Browse source 
The 'one continuous life' telos had three built pieces (turn loop, Shape B+ resume,
idle_tick learning) but no process holding the engine alive+learning over uptime. This
adds that heartbeat loop: chat/always_on.run_continuous ticks idle_tick on a cadence so
the engine LIVES and LEARNS even when no one is talking to it, READS (never repairs) the
closure invariant as evidence each beat, and persists so the life survives interruption
and resumes as the SAME life.

Proven by 3 falsifiable soak tests (non-vacuous): lives (closure observed + held, real
field excited, ~7x under the 1e-6 ceiling); learns while idle (Step-D consolidation
DERIVES and STORES member(socrates,mortal) during idle — absent before the heartbeat,
present after); converges (saturated life stops churning); survives interruption as the
SAME life (strict identity guard enforces it on reboot + the consolidated STORED record
survives, not mere re-derivation); never repairs (field byte-identical after the run).

Safety by composition, no new authority: idle_tick stays proposal-only (HITL untouched)
+ sound proof-gated SPECULATIVE session-memory consolidation; closure owned by
algebra/versor.py (heartbeat only reads versor_condition — no hot-path repair).

Adversarial 3-reviewer panel: no-hot-path-repair PASS, wrong=0 PASS; the panel's
proof-quality findings are fixed — removed the tautological identity_continuous claim
(engine identity is config-derived, invariant by construction), reframed the resume test
to prove ACCUMULATED learning survived (stored record, not re-derivation) under the strict
identity guard, surfaced final_checkpoint_ok (no silent swallow), and softened the
docstring (this is the heartbeat LOOP a daemon runs; the daemon shell is a follow-up).

Scope: the reusable loop ships; a production daemon (CLI entry, real cadence, signals)
is a thin follow-up. Purely additive (no existing file touched).
This commit is contained in:
Shay 2026-06-14 16:07:41 -07:00
parent 4831a49720
commit 32466434c2
2 changed files with 281 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

172
chat/always_on.py Normal file
View file

@ -0,0 +1,172 @@
"""The L10 always-on heartbeat — the loop that makes the life CONTINUOUS.
CORE is meant to be ONE continuous life (listen -> comprehend -> recall -> think ->
articulate -> learn -> replay), not "many lives sharing a checkpoint." Three pieces of
that spine are already built:
* the turn loop (``chat/runtime.py``) handles user turns;
* Shape B+ persistence makes a reboot resume the SAME life (field/vault/anchor/graph
restored bit-exactly, ``config.persist_session_state``);
* ``ChatRuntime.idle_tick`` advances continuous learning *between* turns
(proposal-only + sound session-memory consolidation).
What was missing is a runtime that holds the engine alive and learning over uptime with
no user turn — the T-experience direction. This module is the reusable **heartbeat
loop**: ``run_continuous`` ticks ``idle_tick`` on a cadence so the engine lives and
learns even when no one is talking to it, READS (never repairs) the closure invariant as
evidence each beat, and persists so the life survives interruption and resumes as the
SAME life. It is the core a daemon would call — the production daemon shell (a real
wall-clock cadence, signal handling, a ``core always-on`` CLI entry) is a thin follow-up
on top of this loop, not built here.
Safety by composition, no new authority:
* ``idle_tick`` is proposal-only (HITL ratification untouched) + sound, proof-gated
session-memory consolidation — the heartbeat introduces no unreviewed mutation.
* closure (``versor_condition < 1e-6``) holds BY CONSTRUCTION (the sanctioned session
anchoring; L10 Decision-0 ruling), so the heartbeat only *reads* ``versor_condition``
as telemetry — never a hot-path repair or watchdog (CLAUDE.md no-hot-path-repair).
* the engine's content identity is invariant within a life (config-derived, not
lived-state); cross-reboot "same life" enforcement is owned by the ``ChatRuntime``
load guard (``IdentityContinuityError`` when the stamped checkpoint identity differs
from the recomputed one) — this loop does not re-implement it.
See the L10 continuity soak (``evals/l10_continuity``) for the turn-loop half; this is
the idle/heartbeat half.
"""
from __future__ import annotations
import time
from dataclasses import dataclass
from typing import Callable
from algebra.versor import versor_condition
from core.engine_identity import engine_identity_for_config
from engine_state import get_git_revision
# The non-negotiable field invariant (CLAUDE.md). The heartbeat READS this as evidence;
# it never repairs to keep it true — closure is owned by ``algebra/versor.py``.
CLOSURE_CEILING = 1e-6
@dataclass(frozen=True, slots=True)
class HeartbeatRecord:
"""Per-beat evidence of one continuous-life heartbeat (read-only telemetry)."""
tick: int
versor_condition: float | None # closure of the live field; None before any turn built one
field_valid: bool # versor_condition < CLOSURE_CEILING (vacuously True when no field yet)
facts_consolidated: int # Step-D facts learned this beat (continuous learning)
proposals_created: int # reviewable proposals emitted this beat (proposal-only)
pending_proposals: int
did_work: bool
@dataclass(frozen=True, slots=True)
class AlwaysOnReport:
"""The ordered evidence of one always-on run — the soak subject.
``identity`` is the engine's content identity for the run (honest telemetry — it is
invariant within a life BY CONSTRUCTION, config-derived, so it is recorded once and
is NOT a continuity proof; cross-reboot "same life" is enforced by the ``ChatRuntime``
load guard). ``closure_observed`` says whether any beat actually observed a field, so
a consumer can distinguish "closure held over N observations" from "no field ever
existed" (``closure_held`` is vacuously True with zero observations).
``final_checkpoint_ok`` surfaces whether the exit checkpoint persisted (never silently
swallowed)."""
records: tuple[HeartbeatRecord, ...]
identity: str
closure_observed: bool
closure_held: bool # every OBSERVED versor_condition < CLOSURE_CEILING
final_checkpoint_ok: bool
total_facts_consolidated: int
total_proposals_created: int
@property
def heartbeats(self) -> int:
return len(self.records)
def _live_versor_condition(runtime) -> float | None:
"""Read the closure of the runtime's live field, or None if no turn built one yet.
Pure telemetry — never mutates or repairs the field."""
context = getattr(runtime, "_context", None)
state = getattr(context, "state", None) if context is not None else None
if state is None:
return None
return float(versor_condition(state.F))
def run_continuous(
runtime,
*,
heartbeats: int,
sleep_seconds: float = 0.0,
on_heartbeat: Callable[[HeartbeatRecord], None] | None = None,
stop: Callable[[], bool] | None = None,
) -> AlwaysOnReport:
"""Run the always-on heartbeat for up to ``heartbeats`` beats.
Each beat: advance continuous learning (``idle_tick``), then record the closure +
learning evidence. ``idle_tick`` self-checkpoints on real work; this loop also
checkpoints once at exit, so the life survives interruption — the next ``ChatRuntime``
over the same engine-state dir resumes the SAME life (with Shape B+ persistence on,
and the load-time identity guard enforcing it).
Bounded for falsifiable soaks; a daemon passes a large ``heartbeats`` + a ``stop``
predicate (and a real ``sleep_seconds`` cadence). ``stop`` is checked BEFORE each
beat so a clean shutdown still persists the final state.
"""
if heartbeats < 0:
raise ValueError("heartbeats must be >= 0")
git_revision = get_git_revision()
identity = engine_identity_for_config(runtime.config, git_revision)
records: list[HeartbeatRecord] = []
final_checkpoint_ok = True
try:
for tick in range(heartbeats):
if stop is not None and stop():
break
result = runtime.idle_tick()
vc = _live_versor_condition(runtime)
did_work = (
result.facts_consolidated > 0
or result.proposals_created > 0
or result.candidates_contemplated > 0
)
record = HeartbeatRecord(
tick=tick,
versor_condition=vc,
field_valid=(vc is None or vc < CLOSURE_CEILING),
facts_consolidated=result.facts_consolidated,
proposals_created=result.proposals_created,
pending_proposals=result.pending_proposals,
did_work=did_work,
)
records.append(record)
if on_heartbeat is not None:
on_heartbeat(record)
if sleep_seconds:
time.sleep(sleep_seconds)
finally:
# Final checkpoint even on a mid-beat interruption — the life persists and resumes
# as the SAME life. Best-effort so a checkpoint failure cannot mask the original
# error, but the outcome is SURFACED (final_checkpoint_ok), never silently swallowed.
try:
runtime.checkpoint_engine_state()
except Exception: # noqa: BLE001 — persistence is best-effort at the boundary
final_checkpoint_ok = False
observed = [r.versor_condition for r in records if r.versor_condition is not None]
return AlwaysOnReport(
records=tuple(records),
identity=identity,
closure_observed=bool(observed),
closure_held=all(vc < CLOSURE_CEILING for vc in observed),
final_checkpoint_ok=final_checkpoint_ok,
total_facts_consolidated=sum(r.facts_consolidated for r in records),
total_proposals_created=sum(r.proposals_created for r in records),
)

109
tests/test_l10_always_on.py Normal file
View file

@ -0,0 +1,109 @@
"""L10 — the always-on heartbeat: one continuous life that lives + learns when idle
and survives interruption as the SAME life.
These exercise the T-experience heartbeat (``chat/always_on.run_continuous``) directly:
the engine ticks ``idle_tick`` over uptime with no user turn, the closure invariant holds
by construction (read as evidence, never repaired), and a reboot resumes the SAME life
with its accumulated heartbeat learning intact (enforced by the strict identity guard).
Wrong=0 / no-hot-path-repair are preserved by composition — the heartbeat only proposes
(HITL untouched) + consolidates sound session memory.
"""
from __future__ import annotations
from chat.always_on import CLOSURE_CEILING, run_continuous
from chat.runtime import ChatRuntime, RuntimeConfig
from core.cognition.pipeline import CognitiveTurnPipeline
from generate.meaning_graph.reader import comprehend
from generate.realize import realize_comprehension, recall_realized
from session.context import SessionContext
_RESUME_CONFIG = RuntimeConfig(
persist_session_state=True, # Shape B+ — the lived field/vault survive reboot
consolidate_determinations=True, # Step D — the loop learns from determined facts
strict_identity_continuity=True, # a reboot must be the SAME life or refuse (load guard)
)
def _stored_members(ctx: SessionContext, subject: str) -> set[str]:
"""The set of ``b`` for which ``member(subject, b)`` is a STORED realized record
(recall, not on-the-fly re-derivation)."""
return {
f.relation_arguments[1]
for f in recall_realized(ctx, subject=subject, predicate="member")
}
def _seed_continuous_life(runtime: ChatRuntime) -> None:
"""Seed a consolidatable held self — member(socrates, man) + subset(man, mortal),
from which the idle heartbeat DERIVES member(socrates, mortal) — and excite the field
so closure is observed, not vacuous."""
ctx = runtime._context
realize_comprehension(comprehend("Socrates is a man."), ctx)
realize_comprehension(comprehend("All men are mortals."), ctx)
# A real cognitive turn excites the field so ``versor_condition`` is observable.
CognitiveTurnPipeline(runtime=runtime).run("Socrates is a man.")
def test_heartbeat_lives_learns_and_converges(tmp_path) -> None:
runtime = ChatRuntime(config=_RESUME_CONFIG, engine_state_path=tmp_path / "engine_state")
_seed_continuous_life(runtime)
# The derived membership is NOT stored before the heartbeat runs.
assert "mortal" not in _stored_members(runtime._context, "socrates")
report = run_continuous(runtime, heartbeats=5)
assert report.heartbeats == 5
assert report.final_checkpoint_ok
assert report.identity # the life carries a content identity (telemetry)
# LIVES: closure holds by construction across the uptime, and it is OBSERVED (a real
# field exists), not vacuously true.
assert report.closure_observed
assert report.closure_held
observed = [r.versor_condition for r in report.records if r.versor_condition is not None]
assert observed and all(vc < CLOSURE_CEILING for vc in observed)
# LEARNS while idle: Step-D consolidation derived AND STORED member(socrates, mortal)
# during the heartbeat (non-vacuous — it was absent before run_continuous).
assert report.total_facts_consolidated >= 1
assert "mortal" in _stored_members(runtime._context, "socrates")
# CONVERGES: a saturated life stops churning — the final beat does no work.
assert report.records[-1].did_work is False
def test_life_survives_interruption_as_the_same_life(tmp_path) -> None:
state_dir = tmp_path / "engine_state"
runtime = ChatRuntime(config=_RESUME_CONFIG, engine_state_path=state_dir)
_seed_continuous_life(runtime)
report = run_continuous(runtime, heartbeats=3)
# The heartbeat's ACCUMULATED learning before the interruption.
assert report.total_facts_consolidated >= 1
assert "mortal" in _stored_members(runtime._context, "socrates")
# Interruption: the live runtime is dropped. The next runtime over the SAME engine-state
# dir is the reboot. Under strict_identity_continuity, constructing it ENFORCES the
# same-life identity guard — a clean construction means the stamped checkpoint identity
# matched the recomputed one (the SAME life, not a new one).
del runtime
resumed = ChatRuntime(config=_RESUME_CONFIG, engine_state_path=state_dir)
# The heartbeat's accumulated learning survived the reboot as a STORED record — not
# merely re-derivable from the seeds: the consolidated member(socrates, mortal) is
# recalled directly. (With consolidation off, this would NOT be stored — non-vacuous.)
assert "mortal" in _stored_members(resumed._context, "socrates")
def test_heartbeat_never_repairs_closure(tmp_path) -> None:
"""The heartbeat READS versor_condition as evidence; it must never repair the field.
A no-op idle life (nothing to learn) keeps closure stable without any mutation."""
runtime = ChatRuntime(config=_RESUME_CONFIG, engine_state_path=tmp_path / "engine_state")
_seed_continuous_life(runtime)
assert runtime._context.state is not None
field_before = runtime._context.state.F.copy()
report = run_continuous(runtime, heartbeats=4)
# The field is byte-unchanged by the idle heartbeat (no propagation, no repair).
assert runtime._context.state is not None
assert (runtime._context.state.F == field_before).all()
assert report.closure_observed and report.closure_held