feat(l10): always-on daemon/CLI — the process that runs the continuous-life heartbeat

The L10 heartbeat loop (run_continuous) had no process to drive it; `core always-on`
is that process — the T-experience spine made runnable. It ticks idle_tick on a wall-clock
cadence so the engine LIVES and LEARNS with no user turn, persists lived_life.json (the
workbench Lived Life surface) + the checkpoint, and resumes the SAME life on restart.

- chat/always_on.py: run_continuous gains unbounded operation (heartbeats=None, runs until
  stop) + an interruptible inter-beat wait (_sleep_until_stop) so shutdown latency is one
  slice, not the cadence interval. Persists the run's identity pack ids (resume-verdict
  faithfulness). on_heartbeat is now best-effort (a broken log pipe can't kill the life).
- chat/always_on_daemon.py (new): the daemon shell — a single-instance OS lock (fcntl.flock:
  kernel-held, atomic, auto-released on death — no stale window, no PID-reuse, no half-written
  race), SIGINT/SIGTERM -> graceful stop (handlers saved/restored), the continuous-life config
  FORCED on, ephemeral (--no-load-state) writes no durable artifact. Foreground + explicit:
  no hidden background execution (CLAUDE.md); only writes the engine-state dir it was given.
- core/cli.py: `core always-on [--interval --max-beats --no-load-state --quiet]` with per-beat
  + summary logging; validates --interval; reports IdentityContinuityError / IncompatibleEngine
  StateError (the "different life / newer build" cases) as clean refusals, not tracebacks.
- workbench/readers.py: ENGINE_STATE_ROOT now honors CORE_ENGINE_STATE_DIR (= the daemon's
  resolved dir), so the workbench can't be split-brained (reading REPO_ROOT/engine_state while
  the daemon writes elsewhere); the Lived Life resume verdict recomputes from the persisted
  pack config, not a default config (no false substrate_changed for a non-default-pack life).
- Lived Life absence state now points at the real `core always-on` command (loop closed).

Adversarial 4-lens review (lock/concurrency, signals/shutdown, invariants/trust-boundary,
test-vacuity/CLI) caught 16 findings; this fixes all real ones — the HIGH lock races (two
daemons over one life), the env split-brain, the IO-kill, the uncaught identity/schema errors,
the unvalidated interval, the ephemeral-artifact shadow, and the resume-verdict pack-id bug —
and closes the two test-coverage gaps it flagged (real SIGTERM path + config-forcing-at-the-
runtime boundary).

Tests (non-vacuous): 11 daemon (flock live-holder refusal, leftover-lock reclaim, unbounded+
stop, interruptible sleep, forced-config-at-boundary, no-load-state guard, REAL SIGTERM
subprocess) + the reader pack-id discrimination test (fails under the old default-config bug).
245 workbench+invariants+always-on Python green; frontend tsc + vitest green; `core always-on`
verified end-to-end (bounded, real SIGTERM graceful stop, interval rejection).

engine_state/always_on.lock is runtime state (gitignored, ADR-0146 pattern).

.gitignore

@@ -60,6 +60,7 @@ engine_state/manifest.json
 engine_state/recognizers.jsonl
 engine_state/discovery_candidates.jsonl
 engine_state/lived_life.json
+engine_state/always_on.lock
 
 # Private outreach / personal packet — never tracked
 01_Anthropic/

chat/always_on.py

@@ -36,9 +36,10 @@ the idle/heartbeat half.
 
 from __future__ import annotations
 
+import itertools
 import json
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Callable
 
@@ -55,6 +56,27 @@ LIVED_LIFE_SCHEMA_VERSION = "lived_life_v1"
 # it never repairs to keep it true — closure is owned by ``algebra/versor.py``.
 CLOSURE_CEILING = 1e-6
 
+# Granularity of the interruptible inter-beat wait. A daemon's clean-shutdown latency is
+# bounded by THIS, not by the (possibly long) cadence interval — a SIGTERM mid-wait is
+# honored within a slice instead of waiting out the whole interval.
+_SLEEP_SLICE_SECONDS = 0.25
+
+
+def _sleep_until_stop(seconds: float, stop: Callable[[], bool] | None) -> None:
+    """Sleep up to ``seconds``, returning early the moment ``stop`` fires.
+
+    Without a ``stop`` predicate this is a plain ``time.sleep``. With one, the wait is
+    sliced so a long cadence interval never delays a clean shutdown beyond one slice."""
+    if stop is None:
+        time.sleep(seconds)
+        return
+    deadline = time.monotonic() + seconds
+    while not stop():
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return
+        time.sleep(min(remaining, _SLEEP_SLICE_SECONDS))
+
 
 @dataclass(frozen=True, slots=True)
 class HeartbeatRecord:
@@ -89,6 +111,9 @@ class AlwaysOnReport:
     final_checkpoint_ok: bool
     total_facts_consolidated: int
     total_proposals_created: int
+    # The identity-determining pack ids of the run's config, so a reader can recompute the
+    # SAME identity (the resume verdict) instead of assuming default packs. Empty = default.
+    identity_pack_ids: dict[str, str] = field(default_factory=dict)
 
     @property
     def heartbeats(self) -> int:
@@ -108,6 +133,7 @@ def serialize_report(report: AlwaysOnReport) -> dict[str, Any]:
         "final_checkpoint_ok": report.final_checkpoint_ok,
         "total_facts_consolidated": report.total_facts_consolidated,
         "total_proposals_created": report.total_proposals_created,
+        "identity_pack_ids": dict(report.identity_pack_ids),
         "records": [
             {
                 "tick": r.tick,
@@ -148,13 +174,13 @@ def _live_versor_condition(runtime) -> float | None:
 def run_continuous(
     runtime,
     *,
-    heartbeats: int,
+    heartbeats: int | None,
     sleep_seconds: float = 0.0,
     on_heartbeat: Callable[[HeartbeatRecord], None] | None = None,
     stop: Callable[[], bool] | None = None,
     report_path: Path | None = None,
 ) -> AlwaysOnReport:
-    """Run the always-on heartbeat for up to ``heartbeats`` beats.
+    """Run the always-on heartbeat for up to ``heartbeats`` beats (``None`` = until ``stop``).
 
     Each beat: advance continuous learning (``idle_tick``), then record the closure +
     learning evidence. ``idle_tick`` self-checkpoints on real work; this loop also
@@ -162,9 +188,10 @@ def run_continuous(
     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``
+    A finite ``heartbeats`` bounds a falsifiable soak; ``heartbeats=None`` runs unbounded
-    predicate (and a real ``sleep_seconds`` cadence). ``stop`` is checked BEFORE each
+    until ``stop`` fires — the daemon contract (``chat.always_on_daemon``), where ``stop``
-    beat so a clean shutdown still persists the final state.
+    is wired to SIGINT/SIGTERM. ``stop`` is checked BEFORE each beat AND interrupts the
+    inter-beat wait, so a clean shutdown is prompt and still persists the final state.
 
     When ``report_path`` is given, the run's lived-life evidence is persisted there after
     the loop exits — point it at ``<engine_state>/lived_life.json`` so the workbench Lived
@@ -173,15 +200,25 @@ def run_continuous(
     state is still checkpointed in ``finally`` for recovery, but the workbench report is
     best-effort and simply not refreshed — the surface keeps the last good run.
     """
-    if heartbeats < 0:
+    if heartbeats is not None and heartbeats < 0:
-        raise ValueError("heartbeats must be >= 0")
+        raise ValueError("heartbeats must be >= 0 or None")
 
     git_revision = get_git_revision()
-    identity = engine_identity_for_config(runtime.config, git_revision)
+    config = runtime.config
+    identity = engine_identity_for_config(config, git_revision)
+    # The identity-determining pack ids (empty == default) — persisted so a reader recomputes
+    # the SAME identity for the resume verdict instead of assuming default packs.
+    identity_pack_ids = {
+        "identity_pack": getattr(config, "identity_pack", "") or "",
+        "ethics_pack": getattr(config, "ethics_pack", "") or "",
+        "register_pack_id": getattr(config, "register_pack_id", "") or "",
+        "anchor_lens_id": getattr(config, "anchor_lens_id", "") or "",
+    }
     records: list[HeartbeatRecord] = []
     final_checkpoint_ok = True
     try:
-        for tick in range(heartbeats):
+        ticks = itertools.count() if heartbeats is None else range(heartbeats)
+        for tick in ticks:
             if stop is not None and stop():
                 break
             result = runtime.idle_tick()
@@ -202,9 +239,14 @@ def run_continuous(
             )
             records.append(record)
             if on_heartbeat is not None:
-                on_heartbeat(record)
+                try:
+                    on_heartbeat(record)
+                except OSError:
+                    # Telemetry is best-effort like the checkpoint: a broken stderr/log
+                    # pipe (BrokenPipeError) must NOT kill an indefinite-uptime life.
+                    pass
             if sleep_seconds:
-                time.sleep(sleep_seconds)
+                _sleep_until_stop(sleep_seconds, stop)
     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
@@ -223,6 +265,7 @@ def run_continuous(
         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),
+        identity_pack_ids=identity_pack_ids,
     )
     if report_path is not None:
         write_lived_life(report, report_path)

chat/always_on_daemon.py

@@ -0,0 +1,195 @@
+"""The always-on daemon shell — the thin process that RUNS the continuous-life heartbeat.
+
+``chat/always_on.run_continuous`` is the reusable loop; this module is the daemon a
+``core always-on`` invocation drives. It resolves the engine-state dir, takes a
+single-instance lock (ONE life per engine-state dir — two daemons would corrupt the one
+continuous life), installs SIGINT/SIGTERM handlers for a graceful stop, builds the
+continuous-life ``ChatRuntime``, and runs the heartbeat unbounded until a signal — writing
+the ``lived_life.json`` evidence the workbench reads and checkpointing so the next start
+resumes the SAME life.
+
+No new authority (CLAUDE.md): the heartbeat only ticks the proposal-only ``idle_tick`` and
+READS ``versor_condition`` (no hot-path repair). It is FOREGROUND and explicit — there is
+no hidden background execution; an operator backgrounds it with their shell / service
+manager. The only writes are to the engine-state dir it was pointed at (the lock, the
+manifest checkpoint, ``lived_life.json``).
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import fcntl
+import os
+import signal
+import threading
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable
+
+from chat.always_on import (
+    LIVED_LIFE_FILENAME,
+    AlwaysOnReport,
+    HeartbeatRecord,
+    run_continuous,
+)
+from chat.runtime import ChatRuntime
+from core.config import RuntimeConfig
+from engine_state import EngineStateStore
+
+LOCK_FILENAME = "always_on.lock"
+
+# The continuous-life config: persist the lived field/vault across reboot (Shape B+),
+# learn from determined facts each beat (Step D), and REFUSE to resume a different-identity
+# checkpoint (the load-time identity guard). So a daemon restart is the SAME life or it
+# stops — never a silent fork.
+CONTINUOUS_LIFE_CONFIG_FLAGS: dict[str, Any] = {
+    "persist_session_state": True,
+    "consolidate_determinations": True,
+    "strict_identity_continuity": True,
+}
+
+
+class AlwaysOnLockedError(RuntimeError):
+    """Another live always-on daemon already holds the lock for this engine-state dir."""
+
+
+class _SingleInstanceLock:
+    """Single-instance lock backed by an advisory OS lock (``fcntl.flock``).
+
+    The kernel holds the lock for the lifetime of the open fd and releases it AUTOMATICALLY
+    when the process dies — so there is no stale-lock window, no PID-reuse ambiguity, and no
+    empty / half-written-file race (a pid-file scheme has all three: a loser of the create
+    race can read a not-yet-written file and mistake a live holder for stale, then reclaim,
+    and two daemons proceed over one life). The lock FILE is intentionally never unlinked —
+    unlinking it would let a waiting peer flock a different inode — so the persistent marker
+    also carries the holder's pid for a human-readable refusal message.
+    """
+
+    def __init__(self, path: Path) -> None:
+        self._path = path
+        self._fd: int | None = None
+
+    def _holder_pid(self) -> int | None:
+        try:
+            return int(self._path.read_text(encoding="utf-8").strip())
+        except (OSError, ValueError):
+            return None
+
+    def acquire(self) -> None:
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        fd = os.open(self._path, os.O_CREAT | os.O_RDWR, 0o644)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError as exc:
+            # The lock is held by another LIVE process (the kernel released it if the holder
+            # died). Read the marker pid for the message; it is informational only.
+            holder = self._holder_pid()
+            os.close(fd)
+            suffix = f" (pid {holder})" if holder is not None else ""
+            raise AlwaysOnLockedError(
+                f"another always-on daemon{suffix} holds {self._path}"
+            ) from exc
+        # We hold the kernel lock; stamp our pid as a human-readable marker (best-effort —
+        # exclusion is the flock, not this content).
+        try:
+            os.ftruncate(fd, 0)
+            os.write(fd, f"{os.getpid()}\n".encode("utf-8"))
+        except OSError:
+            pass
+        self._fd = fd
+
+    def release(self) -> None:
+        if self._fd is None:
+            return
+        try:
+            fcntl.flock(self._fd, fcntl.LOCK_UN)
+        except OSError:
+            pass
+        os.close(self._fd)
+        self._fd = None
+
+
+@dataclass(frozen=True, slots=True)
+class DaemonResult:
+    """The outcome of one daemon run — the heartbeat report + how it ended."""
+
+    report: AlwaysOnReport
+    engine_state_path: Path
+    stopped_by_signal: bool
+
+
+def continuous_life_config(base: RuntimeConfig | None = None) -> RuntimeConfig:
+    """``base`` (or a default config) with the continuous-life flags forced on.
+
+    Forced, not defaulted: the daemon's whole purpose is ONE continuous life, so persistence
+    + consolidation + the strict identity guard are not optional knobs here."""
+    cfg = base if base is not None else RuntimeConfig()
+    return dataclasses.replace(cfg, **CONTINUOUS_LIFE_CONFIG_FLAGS)
+
+
+def run_daemon(
+    *,
+    config: RuntimeConfig | None = None,
+    engine_state_path: Path | None = None,
+    interval: float = 1.0,
+    max_beats: int | None = None,
+    no_load_state: bool = False,
+    on_record: Callable[[HeartbeatRecord], None] | None = None,
+    install_signals: bool = True,
+    stop_event: threading.Event | None = None,
+) -> DaemonResult:
+    """Run the always-on heartbeat as a daemon until a signal (or ``max_beats``).
+
+    Resolves the engine-state dir, takes the single-instance lock (raising
+    ``AlwaysOnLockedError`` if a live daemon already owns this life), installs
+    SIGINT/SIGTERM handlers (unless ``install_signals=False`` — tests drive ``stop_event``
+    directly), then runs ``run_continuous`` with the continuous-life config forced on and
+    persists ``lived_life.json`` beside the checkpoint.
+
+    ``install_signals=True`` must be called from the main thread (a Python signal-handler
+    constraint); tests pass ``install_signals=False`` with their own ``stop_event``.
+    """
+    resolved = EngineStateStore(engine_state_path).path
+    cfg = continuous_life_config(config)
+    stop_event = stop_event if stop_event is not None else threading.Event()
+    signalled = threading.Event()  # distinguishes a signal stop from a max_beats / test stop
+
+    lock = _SingleInstanceLock(resolved / LOCK_FILENAME)
+    lock.acquire()  # fail fast BEFORE any signal/runtime setup if another life is live
+
+    previous_handlers: list[tuple[int, Any]] = []
+    try:
+        if install_signals:
+
+            def _handle(_signum: int, _frame: Any) -> None:
+                signalled.set()
+                stop_event.set()
+
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                old = signal.getsignal(sig)
+                signal.signal(sig, _handle)  # record only AFTER it actually changed
+                previous_handlers.append((sig, old))
+
+        runtime = ChatRuntime(
+            config=cfg, engine_state_path=resolved, no_load_state=no_load_state
+        )
+        # An ephemeral (no-load-state) life persists nothing — including no durable
+        # lived_life.json, which would otherwise overwrite the persistent life's artifact.
+        report_path = None if no_load_state else resolved / LIVED_LIFE_FILENAME
+        report = run_continuous(
+            runtime,
+            heartbeats=max_beats,
+            sleep_seconds=interval,
+            stop=stop_event.is_set,
+            on_heartbeat=on_record,
+            report_path=report_path,
+        )
+    finally:
+        for sig, handler in previous_handlers:
+            # getsignal can return None for a non-Python handler; restore SIG_DFL then.
+            signal.signal(sig, handler if handler is not None else signal.SIG_DFL)
+        lock.release()
+
+    return DaemonResult(
+        report=report, engine_state_path=resolved, stopped_by_signal=signalled.is_set()
+    )

core/cli.py

@@ -324,6 +324,89 @@ def cmd_chat(args: argparse.Namespace) -> int:
     return 0
 
 
+# How often (in beats) the always-on daemon logs an "alive" line when idle, so the
+# operator sees a living process without per-beat spam on a saturated (converged) life.
+_ALWAYS_ON_ALIVE_EVERY = 60
+
+
+def cmd_always_on(args: argparse.Namespace) -> int:
+    """Run the continuous-life heartbeat daemon (T-experience) until SIGINT/SIGTERM.
+
+    The thin shell over ``chat.always_on_daemon.run_daemon``: it forces the continuous-life
+    config on, takes the single-instance lock, and ticks ``idle_tick`` on a cadence so the
+    engine LIVES and LEARNS with no user turn — persisting ``lived_life.json`` (the
+    workbench Lived Life surface) and resuming the SAME life on the next start."""
+    from chat.always_on_daemon import AlwaysOnLockedError, run_daemon
+    from core.engine_identity import IdentityContinuityError
+    from engine_state import IncompatibleEngineStateError
+
+    max_beats = getattr(args, "max_beats", None)
+    if max_beats is not None and max_beats < 0:
+        _die("--max-beats must be >= 0", code=2)
+    interval = float(getattr(args, "interval", 1.0))
+    if interval < 0:
+        _die("--interval must be >= 0", code=2)
+    quiet = bool(getattr(args, "quiet", False))
+
+    def _log(record) -> None:
+        if quiet:
+            return
+        vc = "—" if record.versor_condition is None else f"{record.versor_condition:.2e}"
+        if record.did_work:
+            print(
+                f"[beat {record.tick}] learned +{record.facts_consolidated} facts "
+                f"+{record.proposals_created} proposals · closure {vc}",
+                file=sys.stderr,
+            )
+        elif record.tick % _ALWAYS_ON_ALIVE_EVERY == 0:
+            print(
+                f"[beat {record.tick}] alive · closure {vc} · valid={record.field_valid}",
+                file=sys.stderr,
+            )
+
+    bound = "unbounded (until SIGINT/SIGTERM)" if max_beats is None else f"{max_beats} beats"
+    print(
+        f"core always-on: continuous-life heartbeat · interval {interval}s · {bound}",
+        file=sys.stderr,
+    )
+    try:
+        result = run_daemon(
+            config=_runtime_config_from_args(args),
+            interval=interval,
+            max_beats=max_beats,
+            no_load_state=bool(getattr(args, "no_load_state", False)),
+            on_record=_log,
+        )
+    except AlwaysOnLockedError as exc:
+        _die(f"always-on already running for this engine state: {exc}", code=2)
+    except IdentityContinuityError as exc:
+        _die(
+            "this engine state belongs to a different life (substrate identity "
+            f"changed): {exc}",
+            code=2,
+        )
+    except IncompatibleEngineStateError as exc:
+        _die(f"this engine state was written by a newer build: {exc}", code=2)
+
+    report = result.report
+    how = "signal" if result.stopped_by_signal else "completed"
+    if not report.closure_observed:
+        closure = "no field observed"
+    elif report.closure_held:
+        closure = "held"
+    else:
+        closure = "BREACHED"
+    print(
+        f"core always-on: stopped ({how}) after {report.heartbeats} heartbeats · "
+        f"learned {report.total_facts_consolidated} facts · "
+        f"{report.total_proposals_created} proposals · closure {closure} · "
+        f"checkpoint {'ok' if report.final_checkpoint_ok else 'FAILED'} · "
+        f"life {report.identity}",
+        file=sys.stderr,
+    )
+    return 0
+
+
 def _pytest_args_for_suite(suite: str, extra_args: Sequence[str]) -> list[str]:
     paths = _TEST_SUITES[suite]
     forwarded = list(extra_args)
@@ -4143,6 +4226,42 @@ def build_parser() -> argparse.ArgumentParser:
     )
     chat.set_defaults(func=cmd_chat)
 
+    always_on = subparsers.add_parser(
+        "always-on",
+        help="run the continuous-life heartbeat daemon (T-experience) until SIGINT/SIGTERM",
+    )
+    _add_runtime_policy_args(always_on)
+    always_on.add_argument(
+        "--interval",
+        type=float,
+        default=1.0,
+        metavar="SECONDS",
+        help="seconds between heartbeats (the idle-learning cadence); default: 1.0",
+    )
+    always_on.add_argument(
+        "--max-beats",
+        type=int,
+        default=None,
+        metavar="N",
+        help="stop after N heartbeats; default: unbounded (run until SIGINT/SIGTERM)",
+    )
+    always_on.add_argument(
+        "--no-load-state",
+        action="store_true",
+        default=False,
+        help=(
+            "run a fresh EPHEMERAL life (do not load or persist engine_state); "
+            "default: load + persist + resume the SAME life across restarts"
+        ),
+    )
+    always_on.add_argument(
+        "--quiet",
+        action="store_true",
+        default=False,
+        help="suppress per-beat logging; print only the final summary",
+    )
+    always_on.set_defaults(func=cmd_always_on)
+
     test = subparsers.add_parser("test", help="run pytest with curated suite aliases or direct passthrough")
     test.add_argument("--suite", choices=sorted(_TEST_SUITES), help="curated suite alias to run")
     test.add_argument("--list-suites", action="store_true", help="list curated test suite aliases and exit")

docs/workbench/UI-UX-GUIDE.md

@@ -94,7 +94,7 @@ the command palette.
 | Demos | Registered demo scenarios pass or fail with recorded scenario evidence and proof/entailment DAGs where the demo emits them. |
 | Proposals | Proposal evidence, replay facts, and ratification commands are inspectable without applying mutation. |
 | Runs | Recorded run/session references, checkpoint gaps, and identity-continuity verdicts are discoverable. |
-| Lived Life | A persisted always-on run shows the engine living + learning over uptime, with closure (`versor_condition < 1e-6`) read as evidence each beat (never repaired); the surface's `closure_held` is consistency-checked against the per-beat measurements, so it can never paint a breached field as valid. |
+| Lived Life | A persisted always-on run (produced by the `core always-on` daemon) shows the engine living + learning over uptime, with closure (`versor_condition < 1e-6`) read as evidence each beat (never repaired); the surface's `closure_held` is consistency-checked against the per-beat measurements, so it can never paint a breached field as valid, and its resume verdict shows whether a reboot resumes this life. |
 | Vault | Persisted vault metadata is inspectable when persistence is configured. |
 | Audit | Audit events are readable with payload digests and mutation-boundary flags. |
 | Evals | Allowlisted eval lanes and wrong/correct/refused metrics are visible. |

tests/test_l10_always_on_daemon.py

@@ -0,0 +1,212 @@
+"""L10 — the always-on DAEMON shell: the process that runs the continuous-life heartbeat.
+
+``run_continuous`` (the loop) is proven in ``test_l10_always_on``; this covers what the
+daemon adds: the single-instance OS lock (one life per engine-state dir), the unbounded
+run + prompt interruptible stop, the forced continuous-life config applied AT the runtime,
+real SIGTERM shutdown, and that the daemon leaves a readable ``lived_life.json``. Most
+tests drive ``stop_event`` directly (``install_signals=False``) so they never touch the
+test process's signal handlers; one subprocess test exercises the real signal path.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import threading
+import time
+
+import pytest
+
+from chat.always_on import LIVED_LIFE_FILENAME, _sleep_until_stop
+from chat.always_on_daemon import (
+    LOCK_FILENAME,
+    AlwaysOnLockedError,
+    _SingleInstanceLock,
+    continuous_life_config,
+    run_daemon,
+)
+from core.config import RuntimeConfig
+from workbench.lived_life import lived_life_from_payload
+
+
+def _read_lived_life(state_dir):
+    payload = json.loads((state_dir / LIVED_LIFE_FILENAME).read_text(encoding="utf-8"))
+    return lived_life_from_payload(payload)
+
+
+def _bounded(state_dir, **kw):
+    return run_daemon(
+        engine_state_path=state_dir,
+        interval=kw.pop("interval", 0.0),
+        max_beats=kw.pop("max_beats", 1),
+        install_signals=False,
+        stop_event=threading.Event(),
+        **kw,
+    )
+
+
+def test_daemon_runs_bounded_and_leaves_a_valid_lived_life(tmp_path) -> None:
+    state_dir = tmp_path / "engine_state"
+    result = _bounded(state_dir, max_beats=3)
+
+    assert result.report.heartbeats == 3
+    assert result.stopped_by_signal is False  # ended on max_beats, not a signal
+    assert result.report.final_checkpoint_ok
+    assert result.engine_state_path == state_dir
+    # The daemon left the workbench a readable, validated surface where it reads it.
+    surface = _read_lived_life(state_dir)
+    assert surface.status == "recorded"
+    assert surface.heartbeats == 3
+    # The lock is RELEASED on exit — a second run over the same dir acquires it cleanly.
+    assert _bounded(state_dir, max_beats=1).report.heartbeats == 1
+
+
+def test_daemon_forces_the_continuous_life_config_helper() -> None:
+    # Even handed a config with persistence/continuity OFF, the helper forces them ON.
+    base = RuntimeConfig(
+        persist_session_state=False,
+        consolidate_determinations=False,
+        strict_identity_continuity=False,
+    )
+    cfg = continuous_life_config(base)
+    assert cfg.persist_session_state is True
+    assert cfg.consolidate_determinations is True
+    assert cfg.strict_identity_continuity is True
+
+
+def test_run_daemon_applies_the_forced_config_to_the_runtime(tmp_path, monkeypatch) -> None:
+    # The wiring obligation: run_daemon must apply continuous_life_config to the ChatRuntime
+    # it builds, not just expose the helper. Capture the config the runtime is constructed
+    # with and assert the flags are forced on even though the base config had them off.
+    import chat.always_on_daemon as daemon_mod
+
+    captured: dict = {}
+    real_ctor = daemon_mod.ChatRuntime
+
+    def _spy(*, config, engine_state_path, no_load_state):
+        captured["config"] = config
+        return real_ctor(
+            config=config, engine_state_path=engine_state_path, no_load_state=no_load_state
+        )
+
+    monkeypatch.setattr(daemon_mod, "ChatRuntime", _spy)
+    run_daemon(
+        config=RuntimeConfig(
+            persist_session_state=False,
+            consolidate_determinations=False,
+            strict_identity_continuity=False,
+        ),
+        engine_state_path=tmp_path / "engine_state",
+        interval=0.0,
+        max_beats=1,
+        install_signals=False,
+        stop_event=threading.Event(),
+    )
+    cfg = captured["config"]
+    assert cfg.persist_session_state is True
+    assert cfg.consolidate_determinations is True
+    assert cfg.strict_identity_continuity is True
+
+
+def test_unbounded_run_stops_promptly_on_stop_signal(tmp_path) -> None:
+    state_dir = tmp_path / "engine_state"
+    stop_event = threading.Event()
+
+    def _stop_after_two(record) -> None:
+        if record.tick >= 2:
+            stop_event.set()
+
+    result = run_daemon(
+        engine_state_path=state_dir,
+        interval=0.01,
+        max_beats=None,  # would run forever without the stop
+        install_signals=False,
+        stop_event=stop_event,
+        on_record=_stop_after_two,
+    )
+    assert result.report.heartbeats == 3  # beats 0,1,2 then the top-of-loop stop breaks
+    assert result.stopped_by_signal is False  # a stop_event, not a SIGINT/SIGTERM
+
+
+def test_live_flock_holder_refuses_a_second_daemon(tmp_path) -> None:
+    state_dir = tmp_path / "engine_state"
+    state_dir.mkdir(parents=True)
+    # A process actually HOLDING the flock (not merely a pid written in the file) -> refuse.
+    holder = _SingleInstanceLock(state_dir / LOCK_FILENAME)
+    holder.acquire()
+    try:
+        with pytest.raises(AlwaysOnLockedError):
+            _bounded(state_dir)
+    finally:
+        holder.release()
+    # After release, a daemon acquires cleanly — the lock was contended, not permanent.
+    assert _bounded(state_dir).report.heartbeats == 1
+
+
+def test_leftover_lock_file_from_a_dead_holder_does_not_block(tmp_path) -> None:
+    state_dir = tmp_path / "engine_state"
+    state_dir.mkdir(parents=True)
+    # A crash leaves a lock FILE with a dead pid, but the kernel released the flock on death
+    # -> a new daemon acquires (no manual stale reclaim, no deadlock, no PID-reuse ambiguity).
+    (state_dir / LOCK_FILENAME).write_text("999999\n", encoding="utf-8")
+    assert _bounded(state_dir).report.heartbeats == 1
+
+
+def test_no_load_state_writes_no_durable_lived_life(tmp_path) -> None:
+    state_dir = tmp_path / "engine_state"
+    state_dir.mkdir(parents=True)
+    # An ephemeral life must not drop a lived_life.json that would shadow a persistent life.
+    run_daemon(
+        engine_state_path=state_dir,
+        interval=0.0,
+        max_beats=2,
+        no_load_state=True,
+        install_signals=False,
+        stop_event=threading.Event(),
+    )
+    assert not (state_dir / LIVED_LIFE_FILENAME).exists()
+
+
+def test_real_sigterm_stops_the_daemon_cleanly(tmp_path) -> None:
+    # The real signal path the other tests skip (install_signals=True on the main thread):
+    # launch the actual CLI, send SIGTERM, assert a clean signal stop + a persisted artifact.
+    import os
+
+    state_dir = tmp_path / "engine_state"
+    env = {
+        **os.environ,
+        "CORE_ENGINE_STATE_DIR": str(state_dir),
+        "CORE_BACKEND": "numpy",
+        "CORE_STRICT_MLX_ON_APPLE": "0",
+    }
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "core.cli", "always-on", "--interval", "0.05"],
+        env=env,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    time.sleep(3)
+    proc.terminate()  # SIGTERM on POSIX
+    try:
+        _, err = proc.communicate(timeout=30)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.communicate()
+        pytest.fail("daemon did not stop on SIGTERM (interruptible shutdown broken)")
+
+    assert proc.returncode == 0
+    assert "stopped (signal)" in err
+    assert (state_dir / LIVED_LIFE_FILENAME).exists()
+
+
+def test_interruptible_sleep_returns_immediately_when_stopped() -> None:
+    start = time.monotonic()
+    _sleep_until_stop(30.0, lambda: True)
+    assert time.monotonic() - start < 1.0  # 30s vs <1s: a generous, non-flaky margin
+
+
+def test_interruptible_sleep_without_stop_is_a_plain_sleep() -> None:
+    start = time.monotonic()
+    _sleep_until_stop(0.05, None)
+    assert time.monotonic() - start >= 0.04  # actually waited (no stop predicate)

tests/test_workbench_lived_life.py

@@ -99,6 +99,40 @@ def test_resume_verdict_tracks_identity_vs_substrate() -> None:
     assert unknown.resume_status == "unknown"
 
 
+def test_reader_resume_verdict_uses_persisted_pack_ids_not_default(
+    tmp_path, monkeypatch
+) -> None:
+    # A life that ran with a NON-default identity pack must read as would_resume, because the
+    # reader recomputes the current identity from the PERSISTED pack config — not a default
+    # config (which would echo a different identity and falsely read substrate_changed).
+    import json
+
+    from workbench import readers
+
+    # Echo the config's identity_pack so the test can see WHICH config the reader recomputed with.
+    monkeypatch.setattr(
+        readers, "engine_identity_for_config", lambda cfg, rev: f"id:{cfg.identity_pack}"
+    )
+    state_dir = tmp_path / "engine_state"
+    state_dir.mkdir(parents=True)
+    monkeypatch.setattr(readers, "ENGINE_STATE_ROOT", state_dir)
+
+    payload = serialize_report(_report())
+    payload["identity"] = "id:custom_pack_v1"  # the run's persisted identity (non-default pack)
+    payload["identity_pack_ids"] = {
+        "identity_pack": "custom_pack_v1",
+        "ethics_pack": "",
+        "register_pack_id": "",
+        "anchor_lens_id": "",
+    }
+    (state_dir / "lived_life.json").write_text(json.dumps(payload), encoding="utf-8")
+
+    surface = readers.lived_life()
+    # Rebuilt RuntimeConfig(identity_pack="custom_pack_v1") -> "id:custom_pack_v1" == persisted.
+    # A default-config recompute would echo "id:" -> substrate_changed: the bug this guards.
+    assert surface.resume_status == "would_resume"
+
+
 def test_reader_reads_persisted_artifact(tmp_path, monkeypatch) -> None:
     state_dir = tmp_path / "engine_state"
     state_dir.mkdir()

workbench-ui/src/app/lived-life/LivedLifeRoute.tsx

@@ -18,9 +18,8 @@ import type { LivedLife, LivedLifeHeartbeat } from "../../types/api";
 // always-on run lands an artifact (fail-closed, like Vault/Calibration).
 export const LIVED_LIFE_LOADING = "Loading lived life...";
 export const LIVED_LIFE_ABSENCE_STATEMENT =
-  "No always-on run recorded yet. The continuous-life heartbeat persists its evidence here when it runs.";
+  "No always-on run recorded yet. Run the continuous-life heartbeat and it persists its evidence here.";
-export const LIVED_LIFE_ABSENCE_ACTION =
+export const LIVED_LIFE_ABSENCE_ACTION = "core always-on";
-  "Run the always-on heartbeat to persist engine_state/lived_life.json";
 
 function errorMessage(error: unknown): string {
   return error instanceof WorkbenchApiError
@@ -338,7 +337,7 @@ export function LivedLifeRoute() {
     return (
       <EmptyState
         statement={LIVED_LIFE_ABSENCE_STATEMENT}
-        nextAction={LIVED_LIFE_ABSENCE_ACTION}
+        nextAction={{ kind: "cli", command: LIVED_LIFE_ABSENCE_ACTION }}
       />
     );
   }

workbench/readers.py

@@ -55,6 +55,10 @@ from workbench.schemas import (
 from workbench.lived_life import lived_life_from_payload, missing_lived_life
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
+# The engine-state dir the RUNTIME actually uses — honors CORE_ENGINE_STATE_DIR exactly as
+# EngineStateStore / the always-on daemon do, so the workbench can never be split-brained
+# (reading REPO_ROOT/engine_state while the daemon writes to the env dir).
+ENGINE_STATE_ROOT = EngineStateStore().path
 SAFE_EVAL_LANES = frozenset({"contemplation_quality"})
 MAX_ARTIFACT_BYTES = 16 * 1024 * 1024
 READ_CHUNK_BYTES = 64 * 1024
@@ -64,7 +68,7 @@ ENGINE_STATE_RUN_ID = "engine_state_checkpoint"
 _EVAL_RUN_LOCK = threading.Lock()
 _REVIEW_STATES = frozenset(get_args(ReviewState))
 ALLOWED_ARTIFACT_ROOTS = (
-    REPO_ROOT / "engine_state",
+    ENGINE_STATE_ROOT,
     REPO_ROOT / "teaching" / "proposals",
     REPO_ROOT / "teaching" / "math_proposals",
     REPO_ROOT / "evals",
@@ -75,7 +79,6 @@ MATH_PROPOSALS_JSONL = REPO_ROOT / "teaching" / "math_proposals" / "proposals.js
 LANGUAGE_PACK_ROOT = REPO_ROOT / "language_packs" / "data"
 RUNTIME_PACK_ROOT = REPO_ROOT / "packs"
 WORKBENCH_TELEMETRY_ROOT = REPO_ROOT / "workbench_data"
-ENGINE_STATE_ROOT = REPO_ROOT / "engine_state"
 DEMOS_ROOT = REPO_ROOT / "demos"
 CONTEMPLATION_RUNS_ROOT = REPO_ROOT / "contemplation" / "runs"
 _DEFAULT_MATH_AUDIT_PATH = (
@@ -1754,10 +1757,19 @@ def lived_life() -> LivedLife:
     payload = _read_json_object(path)
     artifact = _artifact_ref_for_path(path, "lived_life")
     # The resume verdict: would a reboot resume THIS life? Recompute the current substrate
-    # identity with the canonical function (fail-soft -> "unknown", like IdentityContinuity).
+    # identity with the SAME pack config the run used (persisted in the artifact) — not a
+    # default config, which would falsely read as substrate_changed for a non-default-pack
+    # life. Fail-soft -> "unknown", like IdentityContinuity.
+    pack_ids = payload.get("identity_pack_ids") or {}
     try:
+        recompute_config = RuntimeConfig(
+            identity_pack=pack_ids.get("identity_pack", "") or "",
+            ethics_pack=pack_ids.get("ethics_pack", "") or "",
+            register_pack_id=pack_ids.get("register_pack_id") or None,
+            anchor_lens_id=pack_ids.get("anchor_lens_id") or None,
+        )
         current_identity: str | None = engine_identity_for_config(
-            RuntimeConfig(), get_git_revision()
+            recompute_config, get_git_revision()
         )
     except EngineIdentityError:
         current_identity = None