dotfiles/scripts/l1-replay-sampler.py

#!/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)