diff --git a/.gitignore b/.gitignore index c5f48a98..4c7868be 100644 --- a/.gitignore +++ b/.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/ diff --git a/chat/always_on.py b/chat/always_on.py index c272713c..3f1bca00 100644 --- a/chat/always_on.py +++ b/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`` - predicate (and a real ``sleep_seconds`` cadence). ``stop`` is checked BEFORE each - beat so a clean shutdown still persists the final state. + A finite ``heartbeats`` bounds a falsifiable soak; ``heartbeats=None`` runs unbounded + until ``stop`` fires — the daemon contract (``chat.always_on_daemon``), where ``stop`` + 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 ``/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: - raise ValueError("heartbeats must be >= 0") + if heartbeats is not None and heartbeats < 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) diff --git a/chat/always_on_daemon.py b/chat/always_on_daemon.py new file mode 100644 index 00000000..d2d80135 --- /dev/null +++ b/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() + ) diff --git a/core/cli.py b/core/cli.py index 44876e7e..492341bd 100644 --- a/core/cli.py +++ b/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") diff --git a/docs/workbench/UI-UX-GUIDE.md b/docs/workbench/UI-UX-GUIDE.md index 8f252955..b4896782 100644 --- a/docs/workbench/UI-UX-GUIDE.md +++ b/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. | diff --git a/tests/test_l10_always_on_daemon.py b/tests/test_l10_always_on_daemon.py new file mode 100644 index 00000000..2670a007 --- /dev/null +++ b/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) diff --git a/tests/test_workbench_lived_life.py b/tests/test_workbench_lived_life.py index a0617e50..fdc26287 100644 --- a/tests/test_workbench_lived_life.py +++ b/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() diff --git a/workbench-ui/src/app/lived-life/LivedLifeRoute.tsx b/workbench-ui/src/app/lived-life/LivedLifeRoute.tsx index 909edf77..42165f74 100644 --- a/workbench-ui/src/app/lived-life/LivedLifeRoute.tsx +++ b/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."; -export const LIVED_LIFE_ABSENCE_ACTION = - "Run the always-on heartbeat to persist engine_state/lived_life.json"; + "No always-on run recorded yet. Run the continuous-life heartbeat and it persists its evidence here."; +export const LIVED_LIFE_ABSENCE_ACTION = "core always-on"; function errorMessage(error: unknown): string { return error instanceof WorkbenchApiError @@ -338,7 +337,7 @@ export function LivedLifeRoute() { return ( ); } diff --git a/workbench/readers.py b/workbench/readers.py index 69459da1..1461241c 100644 --- a/workbench/readers.py +++ b/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