session 2026-08-04: PENDING-92/93/94 + the replay-never-resumed finding + l1-replay-sampler
Filed PENDING-92 [HARDENING] idle ladder (cool/deep unreachable, spec §9A.1 divergence), PENDING-93 [PROPOSAL] event_seqs normalisation, PENDING-94 [ESCALATE] the resume floor — minCursor pinned at 0 by two non-participating modules, so 13/13 restarts rebuilt from seq 0 and the catch-up branch has never executed. Recall never worked either (retrieval_count = 0 across the whole April-June graph); same fact from the other end. Adds scripts/l1-replay-sampler.py (external read-only sampler, four positive controls, refuses to run blind). Note to Seb pushed separately as CapableMind-AI@ad285df. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WuMjg3ipEVa3n8CoSzoyvc
This commit is contained in:
David F Glidden
co-authored by
Claude Opus 5
Claude Opus 5
parent
ad2fb07698
commit
9c2a7c985c
Executable
+358
@@ -0,0 +1,358 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user