#!/usr/bin/env python3 """ l1-replay-sampler.py — external, read-only progress sampler for a BMF replay. WHY THIS EXISTS --------------- The replay logs progress once per 100 events. In the slow regime that is one data point every ~51 minutes — telemetry coarser than the failure mode it is meant to describe. On 2026-08-04 three separate mechanism hypotheses (machine sleep, entity-pipeline timeout, work-queue drain gating) each survived hours of diagnosis because nothing at that resolution could contradict them. All three were wrong. This samples the substrate directly, every 30s, and writes a curve. It answers, continuously and without being asked: - is the replay advancing, and at what instantaneous rate - edges-per-node and chains-per-node as the graph densifies - coherence_evaluated — the open question of whether that governor ever runs - which idle state the process is in when each sample is taken SAFETY ------ Read-only by discipline: only SELECTs are issued, nothing is restarted, no config is touched. The DB is opened read-write rather than with mode=ro because a mode=ro open of this instance FAILS — SQLite must be able to create the -shm file to read a WAL database, and mode=ro cannot. (Earned 2026-08-03; every read that day ran against a backup for this reason.) WAL readers do not block the writer, so a 30s poll is negligible against a process writing continuously. INSTRUMENT DISCIPLINE --------------------- Every source carries a same-run positive control. An absence is not evidence until the instrument is shown capable of detecting presence — so if any source cannot be read, or reads as structurally implausible, the sampler prints INSTRUMENT NOT VERIFIED and refuses to start rather than emitting a clean- looking stream of zeros. A sampler that silently reports nothing is exactly the failure class it was built to investigate. USAGE ----- l1-replay-sampler.py --selftest run the controls, print, exit l1-replay-sampler.py sample every 30s until stopped l1-replay-sampler.py --interval 60 sample every 60s l1-replay-sampler.py --once emit exactly one sample and exit """ from __future__ import annotations import argparse import json import os import re import sqlite3 import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path INSTANCE = os.environ.get("BM_INSTANCE_NAME", "mindfabric-00") DB_PATH = Path.home() / ".capablemind" / "data" / INSTANCE / "sqlite" / "temporal.sqlite3" LOG_PATH = Path.home() / ".capablemind" / "logs" / "bmf.stderr.log" OUT_PATH = Path.home() / ".capablemind" / "diagnostics" / f"replay-samples-{INSTANCE}.jsonl" TAIL_BYTES = 262_144 # 256 KB of log tail — enough to hold a progress line even when sparse PROGRESS_RE = re.compile(r"Phase (\d) replay: (\d+)/(\d+) events") IDLE_RE = re.compile(r"Idle state transition: (\w+) → (\w+) \(([^)]*)\)") PAUSE_RE = re.compile(r"Phase \d replay paused for resource pressure") def now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") # -------------------------------------------------------------------------- # sources # -------------------------------------------------------------------------- def read_db() -> dict: """Counts from the temporal store. SELECTs only.""" conn = sqlite3.connect(f"file:{DB_PATH}", uri=True, timeout=10.0) try: cur = conn.cursor() out = {} for key, sql in ( ("nodes", "SELECT count(*) FROM temporal_node"), ("edges", "SELECT count(*) FROM caused"), ("chains", "SELECT count(*) FROM causal_chain"), ("coherence_evaluated", "SELECT count(*) FROM causal_chain WHERE coherence_evaluated = 1"), ): out[key] = cur.execute(sql).fetchone()[0] return out finally: conn.close() def read_log_tail() -> dict: """Last replay position, last idle transition, and pause count in the tail window.""" size = LOG_PATH.stat().st_size with LOG_PATH.open("rb") as fh: fh.seek(max(0, size - TAIL_BYTES)) tail = fh.read().decode("utf-8", errors="replace") out: dict = { "phase": None, "events_done": None, "events_total": None, "idle_state": None, "idle_reason": None, "pauses_in_tail": len(PAUSE_RE.findall(tail)), "log_bytes": size, } progress = PROGRESS_RE.findall(tail) if progress: phase, done, total = progress[-1] out["phase"] = int(phase) out["events_done"] = int(done) out["events_total"] = int(total) idle = IDLE_RE.findall(tail) if idle: _frm, to, reason = idle[-1] out["idle_state"] = to out["idle_reason"] = reason return out def read_process() -> dict: """CPU seconds consumed — distinguishes 'working slowly' from 'not working'.""" try: pids = subprocess.run( ["pgrep", "-f", "BetterMemories.io/dist/index.js"], capture_output=True, text=True, timeout=10, ).stdout.split() if not pids: return {"pid": None, "cpu_seconds": None, "alive": False} pid = int(pids[0]) raw = subprocess.run( ["ps", "-o", "time=", "-p", str(pid)], capture_output=True, text=True, timeout=10, ).stdout.strip() # ps time format: [[dd-]hh:]mm:ss days, _, rest = raw.partition("-") if not rest: days, rest = "0", days parts = [float(p) for p in rest.split(":")] while len(parts) < 3: parts.insert(0, 0.0) secs = int(days) * 86400 + parts[0] * 3600 + parts[1] * 60 + parts[2] return {"pid": pid, "cpu_seconds": round(secs, 2), "alive": True} except Exception as exc: # noqa: BLE001 — degradation must be visible, not silent return {"pid": None, "cpu_seconds": None, "alive": False, "error": str(exc)} # -------------------------------------------------------------------------- # positive controls # -------------------------------------------------------------------------- def selftest() -> tuple[bool, list[str]]: """ Each control must demonstrate the instrument can SEE something, not merely that it did not crash. Returns (ok, lines). """ lines: list[str] = [] ok = True # Control 1 — the DB exists, opens, and the four counters return integers. try: db = read_db() if not all(isinstance(v, int) for v in db.values()): ok = False lines.append("FAIL db: a counter returned a non-integer") elif db["nodes"] == 0 and db["edges"] == 0 and db["chains"] == 0: ok = False lines.append("FAIL db: all three counters are zero — cannot distinguish " "'empty store' from 'reading the wrong database'") else: lines.append(f"pass db: nodes={db['nodes']} edges={db['edges']} " f"chains={db['chains']} coherence_evaluated={db['coherence_evaluated']}") except Exception as exc: # noqa: BLE001 ok = False lines.append(f"FAIL db: {exc}") # Control 2 — the log is readable AND the progress regex actually matches. # A regex that matches nothing would report 'no progress' forever. try: log = read_log_tail() if log["events_done"] is None: ok = False lines.append("FAIL log: no 'Phase N replay: X/Y events' line in the last " f"{TAIL_BYTES // 1024} KB — the progress parser is unverified") else: lines.append(f"pass log: phase={log['phase']} " f"events={log['events_done']}/{log['events_total']}") if log["idle_state"] is None: lines.append("warn log: no idle transition in the tail window " "(idle_state will read null; not fatal)") else: lines.append(f"pass log: idle_state={log['idle_state']} ({log['idle_reason']})") except Exception as exc: # noqa: BLE001 ok = False lines.append(f"FAIL log: {exc}") # Control 3 — the process is locatable and reports non-zero CPU. proc = read_process() if not proc["alive"]: ok = False lines.append(f"FAIL proc: BMF process not found ({proc.get('error', 'no match')})") elif not proc["cpu_seconds"]: ok = False lines.append("FAIL proc: cpu_seconds parsed as zero — the ps parser is unverified") else: lines.append(f"pass proc: pid={proc['pid']} cpu_seconds={proc['cpu_seconds']}") # Control 4 — the output path is writable. try: OUT_PATH.parent.mkdir(parents=True, exist_ok=True) with OUT_PATH.open("a"): pass lines.append(f"pass out: {OUT_PATH} writable") except Exception as exc: # noqa: BLE001 ok = False lines.append(f"FAIL out: {exc}") return ok, lines # -------------------------------------------------------------------------- # sampling # -------------------------------------------------------------------------- def sample(prev: dict | None, events_changed_at: float | None = None) -> dict: rec: dict = {"ts": now_iso()} try: rec.update(read_db()) except Exception as exc: # noqa: BLE001 rec["db_error"] = str(exc) try: rec.update(read_log_tail()) except Exception as exc: # noqa: BLE001 rec["log_error"] = str(exc) rec.update(read_process()) # Derived ratios — the densification curve, which is the thing that decides # whether the rebuild bought a fix or only a constant factor. if rec.get("nodes"): rec["edges_per_node"] = round(rec.get("edges", 0) / rec["nodes"], 2) rec["chains_per_node"] = round(rec.get("chains", 0) / rec["nodes"], 2) # Instantaneous rates against the previous sample. if prev and prev.get("ts"): dt = (datetime.fromisoformat(rec["ts"]) - datetime.fromisoformat(prev["ts"])).total_seconds() if dt > 0: if rec.get("nodes") is not None and prev.get("nodes") is not None: rec["nodes_per_min"] = round((rec["nodes"] - prev["nodes"]) * 60 / dt, 2) if rec.get("events_done") is not None and prev.get("events_done") is not None: rec["events_per_min"] = round((rec["events_done"] - prev["events_done"]) * 60 / dt, 2) if rec.get("cpu_seconds") is not None and prev.get("cpu_seconds") is not None: rec["cpu_percent"] = round((rec["cpu_seconds"] - prev["cpu_seconds"]) * 100 / dt, 1) # events_per_min is derived from a counter the replay only writes every 100 # events. At 12 events/min that is one update per ~8 minutes, so the rate # reads 0.0 for most samples and then spikes — the same coarse-telemetry # flaw this sampler exists to escape, reproduced one level in. Publish how # long the counter has been FROZEN so a flat reading is legible as stale # rather than as zero throughput, and treat nodes_per_min as the fine signal. if events_changed_at is not None: rec["events_stale_seconds"] = round(time.time() - events_changed_at, 1) return rec def seed_staleness() -> tuple[int | None, float | None]: """ Recover the events counter and when it last CHANGED from the existing stream, so a restart of the sampler does not reset the staleness clock. Without this, restarting the instrument makes a counter frozen for 33 minutes report stale=0.0s — the instrument's own restart erasing the very signal it exists to publish. Observed 2026-08-04 on the first handoff between two sampler processes. """ if not OUT_PATH.exists(): return None, None try: size = OUT_PATH.stat().st_size with OUT_PATH.open("rb") as fh: fh.seek(max(0, size - TAIL_BYTES)) lines = fh.read().decode("utf-8", errors="replace").splitlines() rows = [] for line in lines: try: rows.append(json.loads(line)) except json.JSONDecodeError: continue # a partial first line from the seek, or a torn write rows = [r for r in rows if r.get("events_done") is not None] if not rows: return None, None current = rows[-1]["events_done"] # Walk back to the first record still holding the current value. changed_ts = rows[-1]["ts"] for row in reversed(rows): if row["events_done"] != current: break changed_ts = row["ts"] return current, datetime.fromisoformat(changed_ts).timestamp() except Exception: # noqa: BLE001 — seeding is best-effort; never block sampling return None, None def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--interval", type=float, default=30.0, help="seconds between samples (default 30)") ap.add_argument("--once", action="store_true", help="emit one sample and exit") ap.add_argument("--selftest", action="store_true", help="run positive controls and exit") args = ap.parse_args() ok, lines = selftest() for line in lines: print(line, file=sys.stderr) if not ok: print("\nINSTRUMENT NOT VERIFIED — refusing to sample. A stream of zeros from a " "blind instrument is worse than no stream at all.", file=sys.stderr) return 2 if args.selftest: print("\ninstrument verified", file=sys.stderr) return 0 OUT_PATH.parent.mkdir(parents=True, exist_ok=True) print(f"sampling every {args.interval:g}s → {OUT_PATH}", file=sys.stderr) prev: dict | None = None last_events, events_changed_at = seed_staleness() if last_events is not None: held = time.time() - (events_changed_at or time.time()) print(f"resumed staleness: events={last_events} unchanged for {held:.0f}s", file=sys.stderr) while True: rec = sample(prev, events_changed_at) if rec.get("events_done") is not None and rec["events_done"] != last_events: last_events = rec["events_done"] events_changed_at = time.time() rec["events_stale_seconds"] = 0.0 with OUT_PATH.open("a") as fh: fh.write(json.dumps(rec) + "\n") prev = rec if args.once: print(json.dumps(rec, indent=2), file=sys.stderr) return 0 time.sleep(args.interval) if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: sys.exit(130)