diff --git a/chat/always_on.py b/chat/always_on.py new file mode 100644 index 00000000..2301f8ec --- /dev/null +++ b/chat/always_on.py @@ -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), + ) diff --git a/tests/test_l10_always_on.py b/tests/test_l10_always_on.py new file mode 100644 index 00000000..ff8b114e --- /dev/null +++ b/tests/test_l10_always_on.py @@ -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