Merge pull request #747 from AssetOverflow/codex/l10-always-on-heartbeat

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

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

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