diff --git a/bin/dm-log-taxonomy.py b/bin/dm-log-taxonomy.py new file mode 100644 index 0000000..4b8bef8 --- /dev/null +++ b/bin/dm-log-taxonomy.py @@ -0,0 +1,267 @@ +#!/usr/bin/env python3 +"""dm-log-taxonomy.py — READ-ONLY failure taxonomy for fleet DM sidechat reliability. + +Reads /home/super/Projects/NetVM/dm-log.jsonl, prints: + 1. Event-type counts and send outcome rates + 2. Sidechat nav failure taxonomy (per-target, per-agent-pair) + 3. Failure timeline (hourly buckets, worst 10-min windows, by node) + 4. "Ghost" rate: verified:true sidechat sends with no UUID anywhere + 5. Hypothesis evidence tables (nav_failed reasons, placement pairs, + alias sources, autoprovision success, retry distribution) + +No writes to any state files. Runs in <1s on the current log size. +""" +import json +import re +import sys +from collections import Counter, defaultdict +from datetime import datetime, timezone, timedelta + +LOG = "/home/super/Projects/NetVM/dm-log.jsonl" +WINDOW_H = 48 + + +def parse_ts(s): + if not s: + return None + try: + dt = datetime.fromisoformat(s.replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + except Exception: + return None + + +def main(): + now = datetime.now(timezone.utc) + cutoff = now - timedelta(hours=WINDOW_H) + events = [] + parse_err = 0 + with open(LOG, encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + e = json.loads(line) + except Exception: + parse_err += 1 + continue + e["_dt"] = parse_ts(e.get("ts")) + events.append(e) + in_win = [e for e in events if e["_dt"] and e["_dt"] >= cutoff] + print(f"dm-log.jsonl: {len(events)} total lines ({parse_err} parse errors)") + print(f"window: last {WINDOW_H}h -> {len(in_win)} events") + if in_win: + print(f" range: {in_win[0]['ts']} .. {in_win[-1]['ts']}") + print("=" * 78) + + # ---- 1. event-type counts ------------------------------------------------ + types = Counter(e.get("type", "?") for e in in_win) + print("\n[1] EVENT-TYPE COUNTS") + for t, n in types.most_common(): + print(f" {n:6d} {t}") + + # ---- per-send assembly --------------------------------------------------- + sends = {} + order = [] + + def S(e): + sid = e.get("id") + if not sid: + return None + if sid not in sends: + sends[sid] = {"id": sid, "events": [], "first_ts": e["_dt"]} + order.append(sid) + s = sends[sid] + s["events"].append(e) + for k in ("agent", "to", "target"): + if k not in s and e.get(k) is not None: + s[k] = e.get(k) + t = e.get("type") + if t == "verified": + s["verified_seen"] = True + if e.get("thread_uuid"): + s["uuids"] = s.get("uuids", set()) | {e["thread_uuid"]} + if t == "sent": + s["sent_seen"] = True + s["sent_verified"] = bool(e.get("verified")) + if e.get("thread_uuid"): + s["uuids"] = s.get("uuids", set()) | {e["thread_uuid"]} + tags = e.get("tags") or {} + if isinstance(tags, dict) and tags.get("thread"): + s["uuids"] = s.get("uuids", set()) | {tags["thread"]} + if t == "send_done": + s["done"] = True + if t == "sidechat_uuid_capture_failed": + s["capture_failed"] = True + s["capture_out"] = (e.get("out") or "")[:80] + if t == "sidechat_autoprovisioned" and e.get("thread_uuid"): + s["uuids"] = s.get("uuids", set()) | {e["thread_uuid"]} + if t == "alias_resolved" and e.get("thread_uuid"): + s["uuids"] = s.get("uuids", set()) | {e["thread_uuid"]} + if t == "placement_mismatch": + s["placement_mismatch"] = True + if t == "pre_send_assert_failed": + s["gate_failed"] = True + s["gate_reason"] = e.get("reason") + if t == "nav_failed": + s["nav_failed"] = True + s["nav_reason"] = e.get("reason") or "transport" + return s + + for e in in_win: + S(e) + + started = [sends[i] for i in order + if any(e.get("type") == "send_start" for e in sends[i]["events"])] + print(f"\n[1b] SEND OUTCOMES: {len(started)} sends attempted") + oc = Counter() + for s in started: + if s.get("gate_failed"): + oc["loud-fail: pre_send_assert_failed"] += 1 + elif s.get("nav_failed"): + oc["loud-fail: nav_failed"] += 1 + elif s.get("sent_seen") and s.get("sent_verified"): + oc["sent verified:true"] += 1 + elif s.get("sent_seen"): + oc["sent verified:false"] += 1 + elif s.get("done"): + oc["send_done, no sent event"] += 1 + else: + oc["abandoned (no terminal event)"] += 1 + for k, n in oc.most_common(): + print(f" {n:5d} {k}") + print(" (note: main_chat_blocked events are policy blocks, not failures;") + print(" followup_register_failed=signing issues, tangential)") + + # ---- 2. sidechat taxonomy ------------------------------------------------ + def is_sc(s): + return (s.get("target") or "") not in ("main", "", None) + + sc = [s for s in started if is_sc(s)] + print(f"\n[2] SIDECHAT TAXONOMY ({len(sc)} sidechat-targeted sends)") + tax = Counter() + per_target = defaultdict(Counter) + per_pair = defaultdict(Counter) + for s in sc: + if s.get("gate_failed"): + cls = "gate_failed:" + str(s.get("gate_reason")) + elif s.get("nav_failed"): + cls = "nav_failed:" + str(s.get("nav_reason")) + elif s.get("capture_failed"): + cls = "uuid_capture_failed" + elif s.get("placement_mismatch"): + cls = "placement_mismatch" + elif s.get("sent_seen") and s.get("sent_verified"): + cls = ("verified:true, NO uuid anywhere (ghost)" + if not s.get("uuids") else "clean verified") + elif s.get("sent_seen"): + cls = "sent verified:false" + elif s.get("done"): + cls = "done, no sent event" + else: + cls = "abandoned" + tax[cls] += 1 + per_target[s.get("target") or "?"][cls] += 1 + per_pair[f"{s.get('agent') or '?'}->{s.get('to') or '?'}"][cls] += 1 + for k, n in tax.most_common(): + print(f" {n:5d} {k}") + print("\n per-target (sends, non-clean, top classes):") + for tgt, c in sorted(per_target.items(), key=lambda x: -sum(x[1].values())): + tot = sum(c.values()) + bad = tot - c.get("clean verified", 0) + print(f" {tgt}: {tot} sends, {bad} non-clean {dict(c.most_common(4))}") + print("\n per agent-pair:") + for pair, c in sorted(per_pair.items(), key=lambda x: -sum(x[1].values())): + tot = sum(c.values()) + bad = tot - c.get("clean verified", 0) + print(f" {pair}: {tot} sends, {bad} non-clean {dict(c.most_common(4))}") + + # ---- 4. ghosts ------------------------------------------------------------ + ghosts = [s for s in sc if s.get("sent_seen") and s.get("sent_verified") + and not s.get("uuids")] + print(f"\n[4] GHOST RATE: {len(ghosts)}/{len(sc)} " + f"({100.0 * len(ghosts) / len(sc) if sc else 0:.0f}%) verified:true " + f"sidechat sends with no UUID in any event") + gh = Counter(s["first_ts"].strftime("%m-%d %H") for s in ghosts if s["first_ts"]) + print(" ghost hours:", dict(sorted(gh.items()))) + print(" proxies: uuid_capture_failed=" + f"{sum(1 for s in sc if s.get('capture_failed'))}, " + f"placement_mismatch={sum(1 for s in sc if s.get('placement_mismatch'))}, " + f"pre_send_assert_failed={sum(1 for s in sc if s.get('gate_failed'))}") + + # ---- 3. timeline ------------------------------------------------------------ + print("\n[3] TIMELINE (hourly, sidechat sends; #=bad, .=ok)") + buckets = defaultdict(Counter) + for s in sc: + if not s.get("first_ts"): + continue + hr = s["first_ts"].strftime("%m-%d %H:00") + buckets[hr]["total"] += 1 + bad = not (s.get("sent_seen") and s.get("sent_verified") + and not s.get("capture_failed") + and not s.get("placement_mismatch") + and not s.get("gate_failed") and not s.get("nav_failed")) + if bad: + buckets[hr]["bad"] += 1 + for hr in sorted(buckets): + t, b = buckets[hr]["total"], buckets[hr]["bad"] + print(f" {hr} total={t:3d} bad={b:3d} {'#' * b}{'.' * (t - b)}") + wins = defaultdict(Counter) + for s in sc: + if not s.get("first_ts"): + continue + w = s["first_ts"].strftime("%m-%d %H:%M")[:-1] + "0" + wins[w]["total"] += 1 + if not (s.get("sent_seen") and s.get("sent_verified")): + wins[w]["bad"] += 1 + print(" worst 10-min windows (>=3 sends):") + shown = 0 + for w, c in sorted(wins.items(), key=lambda x: -x[1]["bad"]): + if c["total"] >= 3 and shown < 8: + print(f" {w} total={c['total']} bad={c['bad']}") + shown += 1 + print(" bad rate by sending node:") + by_node = defaultdict(Counter) + for s in sc: + by_node[s.get("agent") or "?"]["total"] += 1 + if not (s.get("sent_seen") and s.get("sent_verified")): + by_node[s.get("agent") or "?"]["bad"] += 1 + for node, c in sorted(by_node.items(), key=lambda x: -x[1]["total"]): + r = 100.0 * c["bad"] / c["total"] if c["total"] else 0 + print(f" {node}: {c['bad']}/{c['total']} bad ({r:.0f}%)") + + # ---- 5. hypothesis evidence --------------------------------------------------- + print("\n[5] HYPOTHESIS EVIDENCE") + nfr = Counter(s.get("nav_reason") for s in sc if s.get("nav_failed")) + print(f" nav_failed reasons: {dict(nfr)}") + land = [s for s in sc if s.get("capture_failed")] + print(f" uuid_capture_failed: {len(land)}, " + f"landing-page outs: {sum(1 for s in land if 'muse.ai/' in (s.get('capture_out') or ''))}") + pairs = Counter() + for e in in_win: + if e.get("type") == "placement_mismatch": + pairs[((e.get("expected_uuid") or "?")[:8], + (e.get("actual_uuid") or "?")[:8], e.get("target"))] += 1 + print(" placement_mismatch expected->actual:") + for (a, b, t), n in pairs.most_common(6): + print(f" {n:3d} exp={a}.. act={b}.. target={t}") + ar = Counter(e.get("source") for e in in_win if e.get("type") == "alias_resolved") + print(f" alias_resolved sources: {dict(ar)} (None = field absent, older events)") + ap_try = sum(1 for e in in_win if e.get("type") == "sidechat_autoprovision_start") + ap_ok = sum(1 for e in in_win if e.get("type") == "sidechat_autoprovisioned" + and e.get("thread_uuid")) + print(f" autoprovision: {ap_try} attempts -> {ap_ok} with uuid ({100.0 * ap_ok / ap_try if ap_try else 0:.0f}%)") + rt = Counter() + for s in started: + n = sum(1 for e in s["events"] if e.get("type") == "retry") + if n: + rt[n] += 1 + print(f" retry distribution (sends with >=1 retry): {dict(sorted(rt.items()))}") + print("\n DONE.") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bin/dm.py b/bin/dm.py old mode 100755 new mode 100644 index ac47d0e..26c7e48 --- a/bin/dm.py +++ b/bin/dm.py @@ -62,6 +62,9 @@ TAG_NUDGES_MAX = 10 TAG_NUDGES_DEFAULT = 2 TAG_ROUTE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") TAG_THREAD_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +# Thread-UUID pattern, shared by nav parsing and the pre-send gate. +UUID_RE = (r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-" + r"[0-9a-f]{4}-[0-9a-f]{12}") # Trailing-token form: [reply:expected] [reply:timeout=7200] ... TAG_TOKEN_RE = re.compile(r"\[([A-Za-z][A-Za-z0-9:]*)" r"(?:=([^\]]*))?\]\s*$") @@ -365,6 +368,52 @@ def verify_placement(recipient, msg_id, target, thread_uuid): detail["error"] = str(e)[:200] return False, detail +def assert_pre_send_placement(recipient, target, thread_uuid, direct_nav_done=False): + """Pre-send placement gate (2026-10-04; restored after 19:43Z clobber). + + Independently samples the recipient browser's URL and requires the + expected thread UUID in it BEFORE any send. Fails closed: returns + (False, detail) and the caller must abort the send. 'main' skips. + """ + detail = {"target": target} + if target == "main": + return True, detail + if not thread_uuid: + # Fresh autoprovision: SPA assigns UUID only on first send. + # Strongest pre-send signal: browser must be on a /thread/ page + # (the incident was a send parked on the muse.ai landing page). + _u_rc, _u_out, _u_err = run_full( + f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} url") + actual = (_u_out or "").strip() + detail["actual_url"] = actual[:200] + if "/thread/" not in actual: + detail["reason"] = "not_on_thread_page" + detail["expected"] = "a /thread/ URL (fresh sidechat, UUID assigned on first send)" + return False, detail + return True, detail + expected = thread_uuid.lower() + detail["expected_uuid"] = expected + if not direct_nav_done: + # Title-search nav only: re-navigate by direct /thread/ URL. + _rc, _out, _err = run_full( + f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat use {expected}") + if _rc != 0 or "NOTFOUND" in _out or expected not in _out.lower(): + detail.update({"reason": "direct_nav_failed", "rc": _rc, + "out": _out[:200], "err": _err[:200]}) + return False, detail + time.sleep(2) + else: + time.sleep(1) + _u_rc, _u_out, _u_err = run_full( + f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} url") + actual = (_u_out or "").strip() + detail["actual_url"] = actual[:200] + if expected not in actual.lower(): + detail["reason"] = "url_mismatch" + return False, detail + return True, detail + + def dm_send(agent, target, message, verify=True, raw=False, to_agent=None, tags=None, nudge_meta=None, allow_main_chat=False): @@ -444,6 +493,7 @@ def dm_send(agent, target, message, verify=True, raw=False, thread_uuid = None thread_url = None is_new_sidechat = False + nav_is_uuid = False # Resolve well-known aliases or dynamic thread mappings to UUIDs. nav_target = resolve_sidechat_target(target) if nav_target != target: @@ -454,6 +504,7 @@ def dm_send(agent, target, message, verify=True, raw=False, else: # Reset-at-Begin: If searching sidebar by title (not a direct UUID), reset to main first to expose the sidebar is_uuid = bool(re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", nav_target.strip().lower())) + nav_is_uuid = is_uuid if not is_uuid: run_full(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat main") time.sleep(1) @@ -479,6 +530,12 @@ def dm_send(agent, target, message, verify=True, raw=False, file=sys.stderr) sys.exit(1) is_new_sidechat = True + # If create returned a real thread URL (not /thread/new), + # pin it now so the pre-send gate can assert it. + _cm = re.search(r"/thread/(" + UUID_RE + r")", _c_out, re.I) + if _cm: + thread_uuid = _cm.group(1).lower() + thread_url = "https://muse.ai/thread/" + thread_uuid log_event({"type": "nav_ok", "id": msg_id, "agent": agent, "to": recipient, "target": target, "status": "sidechat_created_pending_uuid", "browser_url": thread_url}) @@ -509,6 +566,20 @@ def dm_send(agent, target, message, verify=True, raw=False, "target": target, "thread_uuid": thread_uuid, "browser_url": thread_url}) + # Pre-send placement assertion: independently confirm the browser is on + # the target thread BEFORE any send. Failure fails loudly (no ghost sends). + _gate_uuid = thread_uuid or (nav_target.lower() if nav_is_uuid else None) + _ps_ok, _ps_detail = assert_pre_send_placement( + recipient, target, _gate_uuid, direct_nav_done=nav_is_uuid) + if not _ps_ok: + log_event({"type": "pre_send_assert_failed", "id": msg_id, "agent": agent, + "to": recipient, **_ps_detail}) + print(f"DM {msg_id} from {agent} to {recipient}/{target}: FAILED " + f"(pre-send placement assertion failed: {_ps_detail.get('reason')}; " + f"expected={_ps_detail.get('expected_uuid') or _ps_detail.get('expected')}; " + f"actual_url={_ps_detail.get('actual_url')})", file=sys.stderr) + sys.exit(1) + time.sleep(2) # Send (raw mode: no truncation — signatures must survive intact) diff --git a/bin/followup-sweeper.py b/bin/followup-sweeper.py old mode 100755 new mode 100644 index b2922e0..e0c8bc0 --- a/bin/followup-sweeper.py +++ b/bin/followup-sweeper.py @@ -14,6 +14,7 @@ Usage: import argparse import json import os +import re import subprocess import sys import time @@ -26,6 +27,7 @@ BIN_DIR = NETVM_ROOT / "bin" JOBS_DIR = NETVM_ROOT / "jobs" FOLLOWUPS_FILE = NETVM_ROOT / "followups.json" JOB_LOG = NETVM_ROOT / "job-log.jsonl" +DM_LOG = NETVM_ROOT / "dm-log.jsonl" DM_PY = BIN_DIR / "dm.py" DISPATCH_PY = BIN_DIR / "job-dispatch.py" @@ -101,6 +103,46 @@ def send_dm(sender, recipient, target, text): except Exception as e: return False, str(e) +_DM_ID_RE = re.compile(r"\bDM ([0-9a-f]{8})\b") +_THREAD_UUID_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", + re.IGNORECASE, +) + +def _resolve_nudge_thread_uuid(nudge_output): + """Parse the DM id from dm.py's stdout, scan the tail (last 5000 lines) + of dm-log.jsonl for that id's sidechat_autoprovisioned event, falling + back to the sent event's tags.thread. Returns the UUID or None.""" + m = _DM_ID_RE.search(nudge_output or "") + if not m: + return None + nudge_id = m.group(1) + fallback = None + try: + with open(DM_LOG, "r", encoding="utf-8") as f: + lines = f.readlines() + except FileNotFoundError: + return None + for line in lines[-5000:]: + line = line.strip() + if not line: + continue + try: + e = json.loads(line) + except Exception: + continue + if e.get("id") != nudge_id: + continue + if e.get("type") == "sidechat_autoprovisioned": + uuid = e.get("thread_uuid") or "" + if _THREAD_UUID_RE.fullmatch(uuid): + return uuid + elif e.get("type") == "sent": + cand = ((e.get("tags") or {}).get("thread")) or "" + if _THREAD_UUID_RE.fullmatch(cand): + fallback = cand + return fallback + def sweep_cycle(dry_run=False): followups = load_followups() @@ -181,6 +223,15 @@ def sweep_cycle(dry_run=False): timeout_s = rec.get("timeout_s", 1800) step_s = max(300, timeout_s // (nudges_allowed + 1)) rec["deadline"] = (now + timedelta(seconds=step_s)).isoformat() + # C1: backfill the thread the nudge actually landed in. + landed_uuid = _resolve_nudge_thread_uuid(out) + if landed_uuid and rec.get("thread_uuid") != landed_uuid: + rec["thread_uuid"] = landed_uuid + print(f"Sweeper: backfilled thread_uuid={landed_uuid} " + f"for followup {dm_id}", file=sys.stderr) + # C2: record final-nudge routing for the harvester. + if is_final: + rec["final_nudge_target"] = "main" modified = True append_job_log({ "ts": utcnow_str(), @@ -189,6 +240,7 @@ def sweep_cycle(dry_run=False): "nudge_num": nudge_num, "recipient": recipient, "target": delivery_target, + "thread_uuid": rec.get("thread_uuid"), }) else: print(f"Sweeper WARNING: nudge send failed: {out}", file=sys.stderr) diff --git a/bin/placement-audit.py b/bin/placement-audit.py new file mode 100755 index 0000000..c47fbb0 --- /dev/null +++ b/bin/placement-audit.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +"""placement-audit.py -- readable audit of sidechat placement / pre-send gate +activity in dm-log.jsonl. + +The gate and navigation emit machine-readable events (pre_send_assert_failed, +placement_failed, placement_mismatch, sidechat_uuid_capture_failed, nav_failed, +...). This CLI turns them into human-readable summaries so the logs stay +useful for troubleshooting sidechat reliability. + +Usage: + placement-audit.py --since 24h # summary over the last 24 hours + placement-audit.py --since 7d # ... 7 days + placement-audit.py --since 60m # ... 60 minutes + placement-audit.py --watch # live tail of placement-relevant events + +Read-only: never writes, no network, no browser. +""" + +import argparse +import json +import re +import sys +import time +from collections import Counter, defaultdict +from datetime import datetime, timedelta, timezone + +DM_LOG = "/home/super/Projects/NetVM/dm-log.jsonl" +JOB_LOG = "/home/super/Projects/NetVM/job-log.jsonl" + +# Event types that count as placement/navigation failures. +FAILURE_TYPES = { + "sidechat_uuid_capture_failed", + "placement_mismatch", + "placement_failed", + "pre_send_assert_failed", + "nav_failed", + "send_failed", + "failed", +} + +# Event types worth streaming in --watch (dm-log). +WATCH_TYPES_DM = FAILURE_TYPES | { + "verified", + "sent", + "nav_ok", + "sidechat_autoprovisioned", +} + + +def parse_since(s): + m = re.fullmatch(r"(\d+)(m|h|d)", (s or "").strip().lower()) + if not m: + raise ValueError(f"bad --since value {s!r}; use like 60m, 24h, 7d") + n, unit = int(m.group(1)), m.group(2) + return timedelta(minutes=n) if unit == "m" else timedelta(hours=n) if unit == "h" else timedelta(days=n) + + +def parse_ts(ts): + if not ts: + return None + try: + dt = datetime.fromisoformat(str(ts).replace("Z", "+00:00")) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def short_uuid(u): + u = str(u or "") + return u[:8] + "..." if len(u) > 12 else u + + +def fmt_ts(e): + dt = parse_ts(e.get("ts")) + return dt.strftime("%m-%d %H:%M:%S") if dt else "?" + + +def expected_actual(e): + """Return (expected, actual) strings for a failure event.""" + t = e.get("type") + if t in ("pre_send_assert_failed", "placement_failed"): + return e.get("expected_uuid"), e.get("actual_url") + if t == "placement_mismatch": + return e.get("expected_uuid"), e.get("actual_uuid") + if t == "sidechat_uuid_capture_failed": + return "(uuid capture)", e.get("out") + if t == "nav_failed": + return e.get("expected") or e.get("reason"), e.get("got") or e.get("out") + if t in ("send_failed", "failed"): + return e.get("reason"), (e.get("send_out") or e.get("err") or "")[:120] + return None, None + + +def one_line(e): + """Compact one-line summary of an event.""" + t = e.get("type", "?") + who = f"{e.get('agent', '?')}->{e.get('to', '?')}/{e.get('target', '?')}" + base = f"{fmt_ts(e)} {t:28s} {str(e.get('id', ''))[:8]:8s} {who}" + if t in FAILURE_TYPES: + reason = e.get("reason") or "" + exp, act = expected_actual(e) + extra = f" reason={reason}" if reason else "" + if exp or act: + extra += f" expected={short_uuid(exp) if exp and len(str(exp)) > 20 else exp} actual={act}" + return base + extra + if t == "sent": + return base + f" verified={e.get('verified')}" + if t == "verified": + return base + f" placement={e.get('placement')} thread={short_uuid(e.get('thread_uuid'))}" + if t == "nav_ok": + return base + f" status={e.get('status')} url={e.get('browser_url')}" + if t == "sidechat_autoprovisioned": + return base + f" thread={short_uuid(e.get('thread_uuid'))}" + return base + + +def iter_log(path, cutoff=None): + """Yield parsed events from a jsonl file, optionally filtered by ts.""" + try: + f = open(path, encoding="utf-8") + except OSError as ex: + print(f"warning: cannot open {path}: {ex}", file=sys.stderr) + return + with f: + for line in f: + line = line.strip() + if not line: + continue + try: + e = json.loads(line) + except json.JSONDecodeError: + continue + if cutoff is not None: + dt = parse_ts(e.get("ts")) + if dt is None or dt < cutoff: + continue + yield e + + +def cmd_since(args): + try: + delta = parse_since(args.since) + except ValueError as ex: + print(str(ex), file=sys.stderr) + return 1 + cutoff = datetime.now(timezone.utc) - delta + + sends = Counter() # (agent, target) -> sent events + sent_ok = Counter() # (agent, target) -> sent with verified=True + taxonomy = Counter() # failure label -> count + failures = [] # failure events, for the recent list + total = 0 + + for e in iter_log(DM_LOG, cutoff): + total += 1 + t = e.get("type") + key = (e.get("agent") or "?", e.get("target") or "?") + if t == "sent": + sends[key] += 1 + if e.get("verified") is True: + sent_ok[key] += 1 + if t in FAILURE_TYPES: + reason = e.get("reason") or "" + label = f"{t}" + (f":{reason}" if reason else "") + taxonomy[label] += 1 + failures.append(e) + + # followup nudge failures live in job-log.jsonl + for e in iter_log(JOB_LOG, cutoff): + if e.get("type") == "followup_nudge_failed": + taxonomy["followup_nudge_failed"] += 1 + failures.append({"type": "followup_nudge_failed", + "ts": e.get("ts"), "id": e.get("dm_id"), + "agent": "sweeper", "to": e.get("recipient"), + "target": e.get("target"), + "reason": (e.get("error") or "")[:100]}) + + print(f"== placement audit: last {args.since} (since {cutoff.strftime('%Y-%m-%d %H:%M UTC')}) ==") + print(f"dm-log events scanned: {total}") + print() + print("-- per-target sends --") + print(f"{'agent':10s} {'target':28s} {'sent':>5s} {'verified':>8s} {'unver':>6s}") + for (agent, target), n in sorted(sends.items(), key=lambda kv: -kv[1]): + ok = sent_ok.get((agent, target), 0) + print(f"{agent:10s} {target:28s} {n:5d} {ok:8d} {n - ok:6d}") + if not sends: + print("(no sends in window)") + print() + print("-- failure taxonomy --") + if taxonomy: + for label, n in taxonomy.most_common(): + print(f"{n:5d} {label}") + else: + print("(no placement failures in window)") + print() + print("-- 10 most recent failures --") + failures.sort(key=lambda e: parse_ts(e.get("ts")) or datetime.min.replace(tzinfo=timezone.utc), + reverse=True) + for e in failures[:10]: + print(one_line(e)) + if not failures: + print("(none)") + return 0 + + +def cmd_watch(_args): + print("watching dm-log.jsonl for placement events (Ctrl-C to stop)...", file=sys.stderr) + try: + f = open(DM_LOG, encoding="utf-8") + except OSError as ex: + print(f"cannot open {DM_LOG}: {ex}", file=sys.stderr) + return 1 + with f: + f.seek(0, 2) # start at end: live view only + try: + while True: + line = f.readline() + if not line: + time.sleep(2) + continue + line = line.strip() + if not line: + continue + try: + e = json.loads(line) + except json.JSONDecodeError: + continue + if e.get("type") in WATCH_TYPES_DM: + print(one_line(e), flush=True) + except KeyboardInterrupt: + print("\nstopped.", file=sys.stderr) + return 0 + + +def main(argv=None): + ap = argparse.ArgumentParser(description="Audit sidechat placement / gate events in dm-log.jsonl") + ap.add_argument("--since", metavar="60m|24h|7d", + help="summarize events newer than this (e.g. 60m, 24h, 7d)") + ap.add_argument("--watch", action="store_true", + help="live tail of placement-relevant events") + args = ap.parse_args(argv) + if args.watch: + return cmd_watch(args) + if args.since: + return cmd_since(args) + ap.print_help() + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bin/response-harvester.py b/bin/response-harvester.py old mode 100755 new mode 100644 index 04b7705..b4288e3 --- a/bin/response-harvester.py +++ b/bin/response-harvester.py @@ -71,6 +71,18 @@ except ImportError: VALID_AGENTS = ["muse", "pip", "646", "opm"] DEFAULT_PORTS = {"muse": 9410, "pip": 9420, "646": 9430, "opm": 9440} +# Matches EVERY [RESULT ] marker in a message (use with finditer, not +# search). The result text is lazy and stops before the next marker (or end of +# text), so a message closing two jobs records each with its own text instead +# of the first marker greedily swallowing the second. +RESULT_RE = re.compile(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*?)(?=\[RESULT\s|\Z)", re.S) + +def iter_result_markers(text): + """Yield (job_id, result_text) for every [RESULT ] marker in text.""" + for m in RESULT_RE.finditer(text or ""): + yield m.group(1).strip(), m.group(2).strip() + + def utcnow(): return datetime.now(timezone.utc).isoformat() @@ -353,30 +365,29 @@ def harvest_main_feed(cdp, agent, watermarks, followups, dry_run=False): append_jsonl(CHAT_HISTORY_LOG, record) if author == "assistant": - m_res = re.search(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*)", text, re.S) - result_job_id = None - if m_res: - job_id = m_res.group(1).strip() - result_job_id = job_id - result_text = m_res.group(2).strip() - is_fail = is_fail_result(result_text) - job_results += 1 - job_record = { - "ts": utcnow(), - "type": "job_result", - "job_id": job_id, - "agent": agent, - "success": not is_fail, - "result_snippet": result_text[:300], - "thread_id": "main", - "msg_id": mid, - } - if not dry_run: - append_jsonl(JOB_LOG, job_record) - trigger_chain_next(job_id, result_text, success=not is_fail) + markers = list(iter_result_markers(text)) + if markers: + for job_id, result_text in markers: + is_fail = is_fail_result(result_text) + job_results += 1 - clear_matching_followups(followups, agent, "main", mid, text, dry_run, - job_id=result_job_id) + job_record = { + "ts": utcnow(), + "type": "job_result", + "job_id": job_id, + "agent": agent, + "success": not is_fail, + "result_snippet": result_text[:300], + "thread_id": "main", + "msg_id": mid, + } + if not dry_run: + append_jsonl(JOB_LOG, job_record) + trigger_chain_next(job_id, result_text, success=not is_fail) + clear_matching_followups(followups, agent, "main", mid, text, + dry_run, job_id=job_id) + else: + clear_matching_followups(followups, agent, "main", mid, text, dry_run) return new_messages, new_wm, job_results @@ -474,35 +485,31 @@ def harvest_agent_thread(cdp, agent, thread_info, watermarks, followups, dry_run if not dry_run: append_jsonl(CHAT_HISTORY_LOG, record) - # Check for [RESULT ] in assistant messages if author == "assistant": - m_res = re.search(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*)", text, re.S) - result_job_id = None - if m_res: - job_id = m_res.group(1).strip() - result_job_id = job_id - result_text = m_res.group(2).strip() - is_fail = is_fail_result(result_text) - job_results += 1 + markers = list(iter_result_markers(text)) + if markers: + for job_id, result_text in markers: + is_fail = is_fail_result(result_text) + job_results += 1 - job_record = { - "ts": utcnow(), - "type": "job_result", - "job_id": job_id, - "agent": agent, - "success": not is_fail, - "result_snippet": result_text[:300], - "thread_id": thread_id, - "msg_id": mid, - } - if not dry_run: - append_jsonl(JOB_LOG, job_record) - # Trigger pipeline chaining or next job if configured - trigger_chain_next(job_id, result_text, success=not is_fail) - - # Check and clear pending follow-ups - clear_matching_followups(followups, agent, thread_id, mid, text, dry_run, - job_id=result_job_id) + job_record = { + "ts": utcnow(), + "type": "job_result", + "job_id": job_id, + "agent": agent, + "success": not is_fail, + "result_snippet": result_text[:300], + "thread_id": thread_id, + "msg_id": mid, + } + if not dry_run: + append_jsonl(JOB_LOG, job_record) + # Trigger pipeline chaining or next job if configured + trigger_chain_next(job_id, result_text, success=not is_fail) + clear_matching_followups(followups, agent, thread_id, mid, text, + dry_run, job_id=job_id) + else: + clear_matching_followups(followups, agent, thread_id, mid, text, dry_run) return new_messages, new_wm, job_results @@ -610,6 +617,8 @@ def clear_matching_followups(followups, agent, thread_id, mid, text, dry_run=Fal Matches on thread identity (thread_uuid or target='main') OR on job_id (from a [RESULT ] reply). The job_id path works regardless of thread_uuid or target, fixing ghost followups with null thread_uuid. + Also matches a main-chat reply when the sweeper recorded + final_nudge_target='main' (final nudge routed to main chat). """ if not followups: return @@ -627,6 +636,11 @@ def clear_matching_followups(followups, agent, thread_id, mid, text, dry_run=Fal match_thread = True elif f_rec.get("thread_uuid") and f_rec.get("thread_uuid") == thread_id: match_thread = True + elif f_rec.get("final_nudge_target") == "main" and thread_id == "main": + # C3: the sweeper routed the final nudge to main chat, so a + # main-chat reply resolves even when the followup target is a + # sidechat. + match_thread = True # Match by job_id (from [RESULT ]) -- works regardless of # thread_uuid or target. This is an ADDITIONAL path, not a replacement. diff --git a/bin/session-probe.py b/bin/session-probe.py new file mode 100755 index 0000000..e3f7620 --- /dev/null +++ b/bin/session-probe.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +""" +session-probe.py — classify a fleet agent's muse.ai login state via CDP. + +READ-ONLY: performs a single Runtime.evaluate reading document.title, +location.href, and body markers. Never clicks, navigates, types, or mutates +session state in any way. Uses the shared CDP queue at PRIORITY_LOW so it +never blocks operator sends or the harvester. + +Login states (per operator AGENTS.md, 2026-10-03): + LOGGED_IN title contains "Chat —"/"Muse —" (SPA booted authenticated) + LANDING title == "muse.ai" (landing page, not logged in) + LOGGED_OUT body shows a "Log in" affordance + OTP_PROMPT body contains "To log in, enter the code" + ACCOUNT_SELECTION body contains "Your email matches multiple accounts" + UNKNOWN none of the above matched + CDP_UNREACHABLE browser/CDP could not be reached at all + + parked_on_landing (bool, separate field): LOGGED_IN but the current URL is + https://muse.ai/ — the watchdog relaunches browsers with the landing page + as start URL, so a fresh browser is parked there until first nav. Session + is fine; nav may proceed (but allow for SPA boot race). + +Exit code: 0 when LOGGED_IN, 1 otherwise (suitable for pre-nav gating). +Stdout: one JSON line: {agent, state, title, url, has_input, checked_at}. + +INVOCATION (important): CDP is only reachable from inside the node's netns +(Chromium binds DevTools to loopback). Always run via: + netvm-exec.sh -- python3 /home/super/Projects/NetVM/bin/session-probe.py --agent +Running it on the bl host directly will report CDP_UNREACHABLE even when the +browser is healthy. +""" +import argparse +import contextlib +import importlib.util +import json +import os +import sys +import urllib.request +from datetime import datetime, timezone + +BIN_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, BIN_DIR) + +try: + from cdp_queue import cdp_slot, PRIORITY_LOW + HAS_CDP_QUEUE = True +except ImportError: + HAS_CDP_QUEUE = False + +import websocket # noqa: E402 (after sys.path tweak, mirrors muse-chat-api.py) + + +def _load_accounts(): + path = os.path.join(BIN_DIR, "netvm-registry.py") + spec = importlib.util.spec_from_file_location("netvm_registry", path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + accounts = {} + for node, rec in mod.load().items(): + accounts[node] = (node, "http://127.0.0.1:%d/json/list" % rec["cdp_port"]) + return accounts + + +def _ev(ws, expr): + """Runtime.evaluate with event draining (copied pattern from muse-chat-api.py).""" + ws.send(json.dumps({ + "id": 1, "method": "Runtime.evaluate", + "params": {"expression": expr, "returnByValue": True}, + })) + for _ in range(50): + resp = json.loads(ws.recv()) + if resp.get("id") == 1: + break + else: + return None + return resp.get("result", {}).get("result", {}).get("value") + + +_STATE_JS = """(() => { + const body = document.body ? document.body.innerText.slice(0, 4000) : ""; + const has_input = !!document.querySelector( + '[contenteditable="true"], textarea[placeholder*="Message"], div[role="textbox"]'); + return JSON.stringify({ + title: document.title || "", + url: location.href || "", + body: body, + has_input: has_input, + }); +})()""" + + +def classify(title, url, body, has_input): + t = (title or "").strip() + b = (body or "") + if "Your email matches multiple accounts" in b: + return "ACCOUNT_SELECTION" + if "To log in, enter the code" in b: + return "OTP_PROMPT" + if t == "muse.ai": + return "LANDING" + # A "Chat —"/"Muse —" title means the SPA booted with an authenticated + # session (the landing page title is exactly "muse.ai"). The browser may + # still be parked on "/" (fresh relaunch start URL) — reported separately + # via parked_on_landing, not as a session failure. + if ("Chat \u2014" in t) or ("Muse \u2014" in t) or ("Chat -" in t) or ("Muse -" in t): + return "LOGGED_IN" + if "Log in" in b: + return "LOGGED_OUT" + return "UNKNOWN" + + +def probe(agent, timeout=15): + accounts = _load_accounts() + if agent not in accounts: + return {"agent": agent, "state": "UNKNOWN", + "error": "no such agent in registry", + "checked_at": _now()} + node, cdp_url = accounts[agent] + try: + with urllib.request.urlopen(cdp_url, timeout=5) as r: + targets = json.load(r) + except Exception as e: + return {"agent": agent, "state": "CDP_UNREACHABLE", + "error": "cdp list failed: %s" % str(e)[:120], + "checked_at": _now()} + pages = [t for t in targets if t.get("type") == "page"] + if not pages: + return {"agent": agent, "state": "CDP_UNREACHABLE", + "error": "no page target", "checked_at": _now()} + + slot = cdp_slot(node, priority=PRIORITY_LOW) if HAS_CDP_QUEUE \ + else contextlib.nullcontext() + try: + with slot: + ws = websocket.create_connection( + pages[0]["webSocketDebuggerUrl"], timeout=timeout) + try: + raw = _ev(ws, _STATE_JS) + finally: + ws.close() + except Exception as e: + return {"agent": agent, "state": "CDP_UNREACHABLE", + "error": "cdp session failed: %s" % str(e)[:120], + "checked_at": _now()} + if not raw: + return {"agent": agent, "state": "UNKNOWN", + "error": "empty evaluate result", "checked_at": _now()} + try: + snap = json.loads(raw) + except Exception: + return {"agent": agent, "state": "UNKNOWN", + "error": "unparseable evaluate result", "checked_at": _now()} + + state = classify(snap.get("title"), snap.get("url"), + snap.get("body"), snap.get("has_input")) + url = snap.get("url") or "" + parked = url.rstrip("/") in ("https://muse.ai", "https://muse.ai/") + return {"agent": agent, "state": state, "title": snap.get("title"), + "url": url[:120], "parked_on_landing": parked, + "has_input": snap.get("has_input"), + "checked_at": _now()} + + +def _now(): + return datetime.now(timezone.utc).isoformat() + + +def main(): + accounts = _load_accounts() + p = argparse.ArgumentParser( + description="Classify a fleet agent's muse.ai login state (read-only).") + p.add_argument("--agent", required=True, choices=sorted(accounts.keys())) + p.add_argument("--timeout", type=int, default=15) + args = p.parse_args() + result = probe(args.agent, timeout=args.timeout) + print(json.dumps(result)) + sys.stdout.flush() + sys.exit(0 if result.get("state") == "LOGGED_IN" else 1) + + +if __name__ == "__main__": + main() diff --git a/bin/sidechat_manager.py b/bin/sidechat_manager.py index fc45c5f..8516075 100644 --- a/bin/sidechat_manager.py +++ b/bin/sidechat_manager.py @@ -4,6 +4,7 @@ Robust sidechat management for muse-chat-api.py. Provides: - ensure_sidebar(ws): Opens sidebar if closed, with retry +- wait_for_chat_list(ws, name=None): Polls until sidebar list content loads - list_sidechats(ws): Returns list of side chat names, with retry - All operations logged for audit @@ -56,13 +57,51 @@ def ensure_sidebar(ws, max_retries=3): return !!document.querySelector('[data-testid="hatch-chat-compose"]'); })()""") +def wait_for_chat_list(ws, name=None, timeout_s=15): + """Poll until the sidebar chat list content has loaded (and optionally + contains `name`, case-insensitive substring match). + + Readiness requires at least one plausible chat-title line under the + "Side chats" header -- the header itself renders before items populate, + and the compose button that ensure_sidebar() keys on renders earlier + still. Clicking a chat title before the list is ready is a known + nav-miss contributor (the click lands nowhere and the SPA stays on /). + + Per-poll CDP errors are treated as not-ready (browser may be mid-render + or mid-flap). Returns True when ready, False on timeout. Note: an + account with genuinely zero sidechats will always time out here. + """ + import re as _re + want = (name or "").strip().lower() + deadline = time.time() + timeout_s + while time.time() < deadline: + try: + text = _ev(ws, "document.body.innerText") or "" + except Exception: + text = "" + idx = text.find("Side chats") + if idx != -1: + section = text[idx:idx + 2000].lower() + lines = [l.strip() for l in section.split("\n")] + titles = [l for l in lines[1:21] + if 5 < len(l) < 80 + and not _re.fullmatch(r"\d+[mh]", l) + and l != "unread updates"] + if titles and (not want or want in section): + return True + time.sleep(1) + return False + def list_sidechats(ws): """ List side chat names. Returns list of strings. - Ensures sidebar is open first. + Ensures sidebar is open first, then waits for the list content to + populate before scraping (the open signal fires before content loads). """ if not ensure_sidebar(ws): return [] + if not wait_for_chat_list(ws): + return [] result = _ev(ws, """(() => { const text = document.body.innerText; diff --git a/bin/tests/test_followup_fixes.py b/bin/tests/test_followup_fixes.py new file mode 100644 index 0000000..ea70b91 --- /dev/null +++ b/bin/tests/test_followup_fixes.py @@ -0,0 +1,694 @@ +#!/usr/bin/env python3 +""" +test_followup_fixes.py -- verification harness for the fleet DM machinery fixes. + +Workstream 4 of 5 (verification). Covers the contracts of three sibling +workstreams editing code in /home/super/Projects/NetVM/bin/ : + + ws-1 followup-sweeper.py + - backfill rec["thread_uuid"] from dm-log sidechat_autoprovisioned + events (so a followup whose thread provisioned late becomes + resolvable instead of a ghost); + - record rec["final_nudge_target"] = "main" when the final nudge is + routed to main chat. + ws-2 response-harvester.py + - resolve followups on main-chat assistant replies when the followup + carries final_nudge_target == "main"; + - extract ALL [RESULT ] markers per message (finditer), + not just the first (re.search). + ws-3 dm.py + - assert the post-nav browser URL contains the target thread UUID + BEFORE sending; fail loudly otherwise. target == "main" skips the + assertion. + +Test layout +----------- + test_contract_* Executable specs of the intended behavior, written against + small local reference predicates. These run GREEN now and + pin the exact semantics the siblings must satisfy. + test_wired_* The same behaviors exercised against the REAL modules + (importlib-loaded from bin/; stdlib-only, side-effect-free + imports; tmp fixtures). Where a sibling has not landed the + change yet, these FAIL with an exact deviation report -- + that is the intended signal, not a bug in the harness. + +Pure unit tests: NO browser, NO network, NO live DMs, NO writes to live state +files (followups.json, job-log.jsonl, dm-log.jsonl). Tmp copies/fixtures only. + +Run: python3 test_followup_fixes.py (built-in runner below) + pytest test_followup_fixes.py (also compatible) + +Snapshot note (2026-10-04 ~19:45Z): ws-2 has fully landed -- +clear_matching_followups() grew final_nudge_target AND job_id matching paths, +and iter_result_markers() extracts all markers via finditer. ws-1 has +partially landed -- DM_LOG constant, _resolve_nudge_thread_uuid() helper, and +final_nudge_target marking are in; the ghost fail-fast for null-thread_uuid +recs is unchanged (no pre-check backfill). ws-3 (dm.py pre-send URL +assertion) had not landed at the time of writing. +""" + +import importlib.util +import inspect +import json +import os +import re +import sys +import tempfile +from pathlib import Path + +BIN_DIR = Path(__file__).resolve().parent.parent # .../bin/tests -> .../bin + + +# -------------------------------------------------------------------------- +# module loading (side-effect-free: all three modules are stdlib-only at +# import time and do real work only inside functions / __main__) +# -------------------------------------------------------------------------- + +def _load(mod_name, filename): + path = BIN_DIR / filename + spec = importlib.util.spec_from_file_location(mod_name, str(path)) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class _Skip(Exception): + pass + + +def skip(reason): + raise _Skip(reason) + + +# -------------------------------------------------------------------------- +# CONTRACT REFERENCE IMPLEMENTATIONS (executable specs -- green now) +# -------------------------------------------------------------------------- + +def contract_backfill_thread_uuid(rec, dm_log_events): + """ws-1 spec: adopt thread_uuid from a matching sidechat_autoprovisioned + dm-log event. Never overwrite an existing UUID; never write null.""" + if rec.get("thread_uuid"): + return False + for ev in dm_log_events or []: + if ev.get("type") != "sidechat_autoprovisioned": + continue + uuid = ev.get("thread_uuid") + if not uuid: + continue + if ev.get("id") == rec.get("dm_id") or ev.get("target") == rec.get("target"): + rec["thread_uuid"] = uuid + return True + return False + + +def contract_route_nudge(rec): + """ws-1 spec: final-nudge routing + marking. Returns (delivery_target, + is_final); marks rec['final_nudge_target']='main' on the final nudge.""" + nudge_num = rec.get("nudges_sent", 0) + 1 + nudges_allowed = rec.get("nudges_allowed", 2) + is_final = (nudge_num == nudges_allowed) + delivery_target = "main" if is_final else rec.get("target", "main") + if is_final: + rec["final_nudge_target"] = "main" + return delivery_target, is_final + + +def contract_followup_matches(rec, agent, thread_id): + """ws-2 spec: extended matching predicate for clear_matching_followups.""" + if rec.get("status") != "pending": + return False + if rec.get("recipient") != agent: + return False + if rec.get("target") == "main" and thread_id == "main": + return True + if rec.get("thread_uuid") and rec.get("thread_uuid") == thread_id: + return True + # NEW (ws-2 contract): the final nudge went to main, so a main-chat + # assistant reply resolves the followup even though target != "main". + if rec.get("final_nudge_target") == "main" and thread_id == "main": + return True + return False + + +_RESULT_MARKER_RE = re.compile(r"\[RESULT\s+([A-Za-z0-9_-]+)\]") + + +def contract_extract_results(text): + """ws-2 spec: extract EVERY [RESULT ] marker; each result text + runs from its marker to the next marker (or end of text).""" + ms = list(_RESULT_MARKER_RE.finditer(text or "")) + out = [] + for i, m in enumerate(ms): + seg_end = ms[i + 1].start() if i + 1 < len(ms) else len(text) + out.append((m.group(1), text[m.end():seg_end].strip())) + return out + + +_UUID_RE = re.compile( + r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I) + + +def contract_assert_post_nav_url(actual_url, expected_uuid, target): + """ws-3 spec: (ok, reason). target == 'main' skips the assertion.""" + if target == "main": + return True, "main_skipped" + if not expected_uuid: + return False, "no_expected_uuid" + if expected_uuid.lower() in (actual_url or "").lower(): + return True, "uuid_present" + return False, "url_mismatch" + + +# -------------------------------------------------------------------------- +# fixtures +# -------------------------------------------------------------------------- + +# Exact text of opm's 2026-10-04 16:54:17Z message (chat-history.jsonl, +# msg assistant-msg-a5afa9c6-...): carries TWO [RESULT] markers; the old +# re.search path recorded only the first, swallowing the 163439 marker. +TWO_MARKER_TEXT = 'All 10 public services are up, bl is up, tunnels are holding except muse\'s long-standing 7682 terminal gap, and the DM queue is drained — fleet operational.\n\n[RESULT pipe-demo-step1-20261004-165058-b08f5fb5] OK: Fleet healthy and operational\n\nOn the nudge: I never received the 16:35:04 request. I traced it in bl\'s dm-log — job `3ef91457` (pipe-demo-step1 from 16:34:39) targeted side chat "pipe-1b4579", but `sidechat_uuid_capture_failed` fired at 16:34:59 (nav landed on the muse.ai landing page, thread_uuid null), and the "verified: true" at 16:35:04 is the known placement-blind verification. The message went somewhere I can\'t see, so the followup tracker correctly saw no reply and nudged. Two things to fix on the pipeline side: assert the post-nav URL holds the target UUID before sending (never trust the verified flag alone), and dedupe nudges — a second identical nudge (`5e078808`) fired at 16:52:01.\n\nClosing the original loop anyway:\n\n[RESULT pipe-demo-step1-20261004-163439-8a408441] OK: Fleet healthy and operational' + +UUID_A = "bf7bf3e1-3ee8-4816-9b66-b77b34387986" +UUID_B = "1e75a740-d08f-443d-a0f9-793db196e24f" + + +def _mk_rec(**kw): + rec = { + "dm_id": "testdm01", + "sender": "opm", + "recipient": "646", + "target": "646 tasks", + "thread_uuid": UUID_B, + "sent_at": "2026-10-04T17:00:00+00:00", + "deadline": "2020-01-01T00:00:00+00:00", # long past -> sweep acts now + "timeout_s": 900, + "nudges_allowed": 2, + "nudges_sent": 0, + "escalate_to": "opm", + "status": "pending", + } + rec.update(kw) + return rec + + +# -------------------------------------------------------------------------- +# 1. sweeper backfill -- contract +# -------------------------------------------------------------------------- + +def test_contract_backfill_sets_uuid_on_matching_event(): + rec = _mk_rec(thread_uuid=None) + evs = [{"type": "sidechat_autoprovisioned", "id": "testdm01", + "target": "646 tasks", "thread_uuid": UUID_A}] + assert contract_backfill_thread_uuid(rec, evs) is True + assert rec["thread_uuid"] == UUID_A, rec + + +def test_contract_backfill_no_event_stays_null(): + rec = _mk_rec(thread_uuid=None) + assert contract_backfill_thread_uuid(rec, []) is False + assert rec["thread_uuid"] is None, rec + evs = [{"type": "nav_ok", "id": "testdm01", "target": "646 tasks"}] + assert contract_backfill_thread_uuid(rec, evs) is False + assert rec["thread_uuid"] is None, rec + + +def test_contract_backfill_never_overwrites_existing_uuid(): + rec = _mk_rec(thread_uuid=UUID_B) + evs = [{"type": "sidechat_autoprovisioned", "id": "testdm01", + "target": "646 tasks", "thread_uuid": UUID_A}] + assert contract_backfill_thread_uuid(rec, evs) is False + assert rec["thread_uuid"] == UUID_B, "existing UUID must never be overwritten" + # ... and a null-UUID event must never blank it either + evs2 = [{"type": "sidechat_autoprovisioned", "id": "testdm01", + "target": "646 tasks", "thread_uuid": None}] + assert contract_backfill_thread_uuid(rec, evs2) is False + assert rec["thread_uuid"] == UUID_B, rec + + +def test_contract_backfill_ignores_unrelated_events(): + rec = _mk_rec(thread_uuid=None) + evs = [{"type": "sidechat_autoprovisioned", "id": "otherdm99", + "target": "other-target", "thread_uuid": UUID_A}] + assert contract_backfill_thread_uuid(rec, evs) is False + assert rec["thread_uuid"] is None, rec + + +# -------------------------------------------------------------------------- +# 2. sweeper final-nudge marking -- contract +# -------------------------------------------------------------------------- + +def test_contract_final_nudge_marks_and_routes_main(): + rec = _mk_rec(nudges_sent=1, nudges_allowed=2) # about to send 2/2 + target, is_final = contract_route_nudge(rec) + assert is_final is True + assert target == "main", target + assert rec.get("final_nudge_target") == "main", rec + + +def test_contract_nonfinal_nudge_keeps_target_unmarked(): + rec = _mk_rec(nudges_sent=0, nudges_allowed=2) # about to send 1/2 + target, is_final = contract_route_nudge(rec) + assert is_final is False + assert target == "646 tasks", target + assert "final_nudge_target" not in rec, rec + + +# -------------------------------------------------------------------------- +# 3. harvester matching -- contract +# -------------------------------------------------------------------------- + +def test_contract_harvester_main_reply_resolves_via_final_nudge_target(): + rec = _mk_rec(target="646 tasks", thread_uuid=UUID_B, + final_nudge_target="main") + assert contract_followup_matches(rec, "646", "main") is True + + +def test_contract_harvester_unrelated_thread_does_not_resolve(): + rec = _mk_rec(target="646 tasks", thread_uuid=UUID_B, + final_nudge_target="main") + assert contract_followup_matches(rec, "646", "deadbeef-thread") is False + assert contract_followup_matches(rec, "pip", "main") is False + + +def test_contract_harvester_existing_rules_still_work(): + # exact thread_uuid match + assert contract_followup_matches(_mk_rec(), "646", UUID_B) is True + # target == "main" + main thread + assert contract_followup_matches(_mk_rec(target="main"), "646", "main") is True + # non-pending never resolves + r = _mk_rec(status="resolved", final_nudge_target="main") + assert contract_followup_matches(r, "646", "main") is False + + +# -------------------------------------------------------------------------- +# 4. multi-RESULT extraction -- contract +# -------------------------------------------------------------------------- + +def test_contract_multi_result_extracts_both(): + got = contract_extract_results(TWO_MARKER_TEXT) + ids = [j for j, _ in got] + assert ids == ["pipe-demo-step1-20261004-165058-b08f5fb5", + "pipe-demo-step1-20261004-163439-8a408441"], ids + texts = dict(got) + assert texts["pipe-demo-step1-20261004-163439-8a408441"] == \ + "OK: Fleet healthy and operational", texts + first = texts["pipe-demo-step1-20261004-165058-b08f5fb5"] + assert first.startswith("OK: Fleet healthy and operational"), first + assert "[RESULT" not in first, "first result text must stop at the 2nd marker" + + +def test_contract_single_result(): + got = contract_extract_results("[RESULT abc-123] OK: done") + assert got == [("abc-123", "OK: done")], got + + +def test_contract_no_result(): + assert contract_extract_results("just a normal message") == [] + assert contract_extract_results("") == [] + + +# -------------------------------------------------------------------------- +# 5. dm.py URL assertion -- contract +# -------------------------------------------------------------------------- + +def test_contract_dm_url_contains_uuid_passes(): + ok, reason = contract_assert_post_nav_url( + "https://muse.ai/thread/" + UUID_A, UUID_A, "pipe-1b4579") + assert ok is True and reason == "uuid_present", (ok, reason) + + +def test_contract_dm_url_landing_page_fails(): + ok, reason = contract_assert_post_nav_url( + "https://muse.ai/", UUID_A, "pipe-1b4579") + assert ok is False and reason == "url_mismatch", (ok, reason) + + +def test_contract_dm_url_wrong_thread_fails(): + ok, reason = contract_assert_post_nav_url( + "https://muse.ai/thread/" + UUID_B, UUID_A, "pipe-1b4579") + assert ok is False and reason == "url_mismatch", (ok, reason) + + +def test_contract_dm_url_main_skips_assertion(): + ok, reason = contract_assert_post_nav_url("https://muse.ai/", None, "main") + assert ok is True and reason == "main_skipped", (ok, reason) + + +def test_contract_dm_url_empty_actual_fails(): + ok, reason = contract_assert_post_nav_url("", UUID_A, "pipe-1b4579") + assert ok is False, (ok, reason) + + +# -------------------------------------------------------------------------- +# wiring: real modules +# -------------------------------------------------------------------------- + +def _tmp_sweeper_env(sweeper): + """Point the sweeper's file IO at a tmp dir; return (tmpdir, saved). + + NOTE: the module uses pathlib.Path objects (FOLLOWUPS_FILE.exists()), + so the monkeypatched values must be Paths, not strs.""" + tmp = tempfile.mkdtemp(prefix="w4sweep") + saved = {} + for attr, fname in (("FOLLOWUPS_FILE", "followups.json"), + ("JOB_LOG", "job-log.jsonl")): + saved[attr] = getattr(sweeper, attr) + setattr(sweeper, attr, Path(tmp) / fname) + return tmp, saved + + +def _restore(mod, saved): + for attr, val in saved.items(): + setattr(mod, attr, val) + + +def test_wired_sweeper_final_nudge_routes_main_and_marks(): + """Real sweep_cycle on a followup due its final nudge: delivery must go + to main (existing behavior) AND rec must gain final_nudge_target='main' + (ws-1 contract).""" + sweeper = _load("sweeper_under_test", "followup-sweeper.py") + tmp, saved = _tmp_sweeper_env(sweeper) + saved["send_dm"] = sweeper.send_dm + seen = {} + + def fake_send_dm(sender, recipient, target, text): + seen.update(sender=sender, recipient=recipient, target=target) + return True, "SENT id=fake01" + + sweeper.send_dm = fake_send_dm + try: + rec = _mk_rec(nudges_sent=1, nudges_allowed=2) # final nudge due + with open(os.path.join(tmp, "followups.json"), "w") as f: + json.dump({"w2nudge": rec}, f) + stats = sweeper.sweep_cycle(dry_run=False) + assert stats["nudges_sent"] == 1, stats + assert seen.get("target") == "main", \ + f"final nudge must route to main, went to {seen.get('target')!r}" + with open(os.path.join(tmp, "followups.json")) as f: + rec2 = json.load(f)["w2nudge"] + assert rec2.get("final_nudge_target") == "main", ( + "DEVIATION: ws-1 has not landed final_nudge_target marking -- " + f"sweep_cycle routed the final nudge to main but did not record " + f"final_nudge_target on the followup rec (rec keys: " + f"{sorted(rec2.keys())})") + finally: + _restore(sweeper, saved) + + +def test_wired_sweeper_backfills_thread_uuid(): + """Real _resolve_nudge_thread_uuid (ws-1): parses the nudge DM id from + send output, adopts the thread UUID from that id's + sidechat_autoprovisioned dm-log event (falling back to the sent event's + tags.thread); returns None when nothing usable exists (never fabricates + a UUID). Fixture dm-log via the module's DM_LOG hook -- no live state.""" + sweeper = _load("sweeper_under_test", "followup-sweeper.py") + resolver = getattr(sweeper, "_resolve_nudge_thread_uuid", None) + assert resolver is not None, ( + "DEVIATION: ws-1 backfill not landed -- followup-sweeper.py has no " + "_resolve_nudge_thread_uuid helper") + tmp = tempfile.mkdtemp(prefix="w4blog") + dmpath = Path(tmp) / "dm-log.jsonl" + saved = sweeper.DM_LOG + sweeper.DM_LOG = dmpath + nid = "f00dbabe" + out = (f"DM {nid} from opm to 646/pipe-x: SENT and VERIFIED " + f"thread={UUID_A}") + try: + # 1. autoprovisioned event -> UUID adopted + with open(dmpath, "w") as f: + f.write(json.dumps({"type": "sidechat_autoprovisioned", "id": nid, + "target": "pipe-x", "thread_uuid": UUID_A}) + "\n") + f.write(json.dumps({"type": "sent", "id": nid, + "tags": {"thread": UUID_A}}) + "\n") + assert resolver(out) == UUID_A, "autoprovisioned UUID not adopted" + # 2. no autoprovisioned event -> falls back to sent.tags.thread + with open(dmpath, "w") as f: + f.write(json.dumps({"type": "sent", "id": nid, + "tags": {"thread": UUID_A}}) + "\n") + assert resolver(out) == UUID_A, "sent.tags.thread fallback broken" + # 3. no usable events -> None (never fabricates / never null-writes) + with open(dmpath, "w") as f: + f.write(json.dumps({"type": "sent", "id": "other12", + "tags": {}}) + "\n") + assert resolver(out) is None, "resolver fabricated a UUID" + # 4. malformed UUID in event -> skipped + with open(dmpath, "w") as f: + f.write(json.dumps({"type": "sidechat_autoprovisioned", "id": nid, + "thread_uuid": "not-a-uuid"}) + "\n") + assert resolver(out) is None, "malformed UUID accepted" + # 5. unparseable nudge output -> None + assert resolver("some garbage without an id") is None + finally: + sweeper.DM_LOG = saved + + +def test_wired_sweeper_never_overwrites_uuid_with_null(): + """Real sweep_cycle: after a successful nudge whose thread cannot be + determined, an existing rec thread_uuid is left untouched (ws-1 + contract: only a real UUID is ever written).""" + sweeper = _load("sweeper_under_test", "followup-sweeper.py") + tmp, saved = _tmp_sweeper_env(sweeper) + saved["send_dm"] = sweeper.send_dm + saved_dm_log = sweeper.DM_LOG + sweeper.DM_LOG = Path(tmp) / "dm-log.jsonl" # empty: no events + Path(tmp, "dm-log.jsonl").write_text("") + nid = "b00bf00d" + sweeper.send_dm = lambda *a: ( + True, f"DM {nid} from opm to 646/pipe-x: SENT and VERIFIED") + try: + rec = _mk_rec(thread_uuid=UUID_B, nudges_sent=0, nudges_allowed=2) + with open(os.path.join(tmp, "followups.json"), "w") as f: + json.dump({"w2null": rec}, f) + sweeper.sweep_cycle(dry_run=False) + with open(os.path.join(tmp, "followups.json")) as f: + rec2 = json.load(f)["w2null"] + assert rec2["thread_uuid"] == UUID_B, ( + f"DEVIATION: existing thread_uuid was overwritten " + f"(now {rec2['thread_uuid']!r}) despite no resolvable nudge thread") + assert rec2["nudges_sent"] == 1 + finally: + sweeper.DM_LOG = saved_dm_log + _restore(sweeper, saved) + + +def test_wired_sweeper_backfill_reprovision_updates_uuid(): + """Real sweep_cycle: when a nudge lands in a newly provisioned thread, + the rec's stale UUID is replaced by the real new one (ws-1 reprovision + case).""" + sweeper = _load("sweeper_under_test", "followup-sweeper.py") + tmp, saved = _tmp_sweeper_env(sweeper) + saved["send_dm"] = sweeper.send_dm + saved_dm_log = sweeper.DM_LOG + sweeper.DM_LOG = Path(tmp) / "dm-log.jsonl" + nid = "c0ffee42" + with open(os.path.join(tmp, "dm-log.jsonl"), "w") as f: + f.write(json.dumps({"type": "sidechat_autoprovisioned", "id": nid, + "target": "pipe-x", "thread_uuid": UUID_A}) + "\n") + sweeper.send_dm = lambda *a: ( + True, f"DM {nid} from opm to 646/pipe-x: SENT and VERIFIED " + f"thread={UUID_A}") + try: + rec = _mk_rec(thread_uuid=UUID_B, nudges_sent=0, nudges_allowed=2) + with open(os.path.join(tmp, "followups.json"), "w") as f: + json.dump({"w2re": rec}, f) + sweeper.sweep_cycle(dry_run=False) + with open(os.path.join(tmp, "followups.json")) as f: + rec2 = json.load(f)["w2re"] + assert rec2["thread_uuid"] == UUID_A, ( + f"DEVIATION: reprovisioned thread UUID not adopted " + f"(still {rec2['thread_uuid']!r})") + finally: + sweeper.DM_LOG = saved_dm_log + _restore(sweeper, saved) + + +def test_wired_harvester_final_nudge_target_main_resolves(): + """Real clear_matching_followups: followup target='646 tasks', + final_nudge_target='main' must resolve on an assistant message in + thread 'main' (ws-2 contract).""" + harv = _load("harvester_under_test", "response-harvester.py") + rec = _mk_rec(target="646 tasks", thread_uuid=UUID_B, + final_nudge_target="main") + fups = {"hx1": rec} + harv.clear_matching_followups(fups, "646", "main", "mid-1", + "some assistant reply", dry_run=True) + assert rec.get("status") == "resolved", ( + "DEVIATION: ws-2 final_nudge_target path not landed -- " + "clear_matching_followups() does not resolve a followup with " + "final_nudge_target='main' on a main-thread assistant reply " + f"(status={rec.get('status')!r}; fn signature: " + f"{inspect.signature(harv.clear_matching_followups)})") + # ... and must NOT resolve on an unrelated thread + rec2 = _mk_rec(target="646 tasks", thread_uuid=UUID_B, + final_nudge_target="main") + fups2 = {"hx2": rec2} + harv.clear_matching_followups(fups2, "646", "unrelated-thread", "mid-2", + "some assistant reply", dry_run=True) + assert rec2.get("status") == "pending", \ + f"unrelated thread must not resolve (status={rec2.get('status')!r})" + + +def test_wired_harvester_existing_rules_still_hold(): + """Real clear_matching_followups: pre-existing rules keep working.""" + harv = _load("harvester_under_test", "response-harvester.py") + r1 = _mk_rec() # thread_uuid == UUID_B + fups = {"e1": r1} + harv.clear_matching_followups(fups, "646", UUID_B, "m", "t", dry_run=True) + assert r1["status"] == "resolved", "exact thread_uuid match broke" + r2 = _mk_rec(target="main", thread_uuid=None) + fups = {"e2": r2} + harv.clear_matching_followups(fups, "646", "main", "m", "t", dry_run=True) + assert r2["status"] == "resolved", "target=='main' rule broke" + r3 = _mk_rec() + fups = {"e3": r3} + harv.clear_matching_followups(fups, "646", "nope", "m", "t", dry_run=True) + assert r3["status"] == "pending", "unrelated thread wrongly resolved" + + +def test_wired_harvester_extracts_all_result_markers(): + """The real extraction path must yield BOTH [RESULT] markers (ws-2: + finditer instead of first-only re.search).""" + harv = _load("harvester_under_test", "response-harvester.py") + extractor = next( + (getattr(harv, n) for n in dir(harv) + if "result" in n.lower() + and any(k in n.lower() for k in ("extract", "iter", "marker")) + and callable(getattr(harv, n))), + None) + src = inspect.getsource(harv) + if extractor is not None: + got = extractor(TWO_MARKER_TEXT) + ids = [j for j, _ in got] + assert ids == ["pipe-demo-step1-20261004-165058-b08f5fb5", + "pipe-demo-step1-20261004-163439-8a408441"], \ + f"extractor {extractor.__name__} missed markers: {ids}" + return + if "finditer" in src and "RESULT" in src: + return # inline finditer implementation detected; contract assumed met + # Deviation evidence: the current inline path uses re.search (first only). + cur = re.search(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*)", + TWO_MARKER_TEXT, re.S) + raise AssertionError( + "DEVIATION: ws-2 finditer change not landed -- response-harvester.py " + "exposes no result extractor and its inline path is re.search " + f"(first marker only): it captures {cur.group(1)!r} and swallows the " + "second marker [RESULT pipe-demo-step1-20261004-163439-8a408441] " + "inside group(2)") + + +def test_wired_dm_send_asserts_post_nav_url_before_send(): + """dm_send must independently assert the post-nav browser URL contains + the target thread UUID BEFORE sending (ws-3 contract); 'main' skips.""" + path = BIN_DIR / "dm.py" + dm = None + via = "file-text fallback" + try: + dm = _load("dm_under_test", "dm.py") + src = inspect.getsource(dm.dm_send) + via = "inspect(dm.dm_send)" + except Exception: + src = path.read_text() + m = re.search(r"def dm_send\(.*?(?=\ndef |\Z)", src, re.S) + src = m.group(0) if m else src + # Exposed predicate? test it directly. + pred = None + if dm is not None: + pred = next( + (getattr(dm, n) for n in dir(dm) + if callable(getattr(dm, n)) + and "nav" in n.lower() and "url" in n.lower()), + None) + if pred is not None: + assert pred("https://muse.ai/thread/" + UUID_A, UUID_A, "pipe-x")[0] is True + assert pred("https://muse.ai/", UUID_A, "pipe-x")[0] is False + return + # ws-3 landed shape: module-level assert_pre_send_placement() called from + # dm_send before the send. Verify the call site and exercise the helper. + gate = getattr(dm, "assert_pre_send_placement", None) if dm is not None else None + if callable(gate): + assert "assert_pre_send_placement" in src, \ + "gate exists but dm_send never calls it" + _orig_run_full = dm.run_full + _calls = [] + + def _stub(cmd, timeout=60): + _calls.append(cmd) + if "sidechat use" in cmd: + return 0, "navigated https://muse.ai/thread/" + UUID_A, "" + return 0, _stub.url, "" + + try: + dm.run_full = _stub + # 1. UUID-known thread, correct placement -> pass + _stub.url = "https://muse.ai/thread/" + UUID_A + ok, detail = gate("opm", "pipe-x", UUID_A, direct_nav_done=True) + assert ok is True, f"expected pass on matching URL: {detail}" + # 2. landing page -> loud fail (the 2026-10-04 incident mode) + _stub.url = "https://muse.ai/" + ok, detail = gate("opm", "pipe-x", UUID_A, direct_nav_done=True) + assert ok is False and detail.get("reason") == "url_mismatch", detail + # 3. wrong thread -> loud fail + _stub.url = "https://muse.ai/thread/" + UUID_B + ok, detail = gate("opm", "pipe-x", UUID_A, direct_nav_done=True) + assert ok is False and detail.get("reason") == "url_mismatch", detail + # 4. main target skips assertion with zero subprocess calls + _calls.clear() + ok, _ = gate("opm", "main", None, direct_nav_done=False) + assert ok is True and not _calls, "main must skip without subprocess" + # 5. re-nav path (direct_nav_done=False) -> pass + _stub.url = "https://muse.ai/thread/" + UUID_A + ok, detail = gate("opm", "pipe-x", UUID_A, direct_nav_done=False) + assert ok is True, f"re-nav path should pass: {detail}" + finally: + dm.run_full = _orig_run_full + return + nav_i = src.find("sidechat use") + send_i = src.find("Send with verification retries") + segment = src[nav_i:send_i] if 0 <= nav_i < send_i else "" + has_url_fetch = re.search(r"""['"]\s*url['"]|account\s+\S+\s+url\b""", segment) + has_abort = ("sys.exit" in segment) or ("raise " in segment) + assert has_url_fetch and has_abort, ( + f"DEVIATION: ws-3 pre-send URL assertion not landed ({via}) -- " + "dm_send's nav->send path performs no independent post-nav URL " + "fetch+containment check that aborts the send; it trusts the " + "`sidechat use` command output (no_thread_url/uuid_mismatch on the " + "nav output only). Note: verify_placement() asserts the URL " + "post-send, which is a different (later) check.") + + +# -------------------------------------------------------------------------- +# runner (also pytest-compatible: plain test_* functions, no args) +# -------------------------------------------------------------------------- + +def main(): + fns = [(n, f) for n, f in sorted(globals().items()) + if n.startswith("test_") and callable(f)] + results = [] + for name, fn in fns: + try: + fn() + results.append((name, "PASS", "")) + except _Skip as e: + results.append((name, "SKIP", str(e))) + except AssertionError as e: + results.append((name, "FAIL", str(e) or "assertion failed")) + except Exception as e: # noqa: BLE001 - harness must not crash + results.append((name, "ERROR", + f"{type(e).__name__}: {e}")) + npass = sum(1 for _, s, _ in results if s == "PASS") + nfail = sum(1 for _, s, _ in results if s in ("FAIL", "ERROR")) + nskip = sum(1 for _, s, _ in results if s == "SKIP") + print(f"\n{'test':58} result") + print("-" * 80) + for name, status, detail in results: + print(f"{name:58} {status}") + if detail and status in ("FAIL", "ERROR"): + for line in detail.splitlines(): + print(f" {line}") + print("-" * 80) + print(f"{len(results)} tests: {npass} pass, {nfail} fail/error, {nskip} skip") + return 1 if nfail else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/SIDECHAT-RELIABILITY.md b/docs/SIDECHAT-RELIABILITY.md new file mode 100644 index 0000000..0a802d6 --- /dev/null +++ b/docs/SIDECHAT-RELIABILITY.md @@ -0,0 +1,228 @@ +# Sidechat Reliability Runbook + +Fleet sidechat DMs are the fleet's nervous system. On 2026-10-04 they were +measured at ~1/3 navigation success. This runbook is the single place that +explains how the machinery works, how each known failure looks in the logs, +and what to do when a DM doesn't arrive. + +All paths below are on **bl** (`super@100.123.153.75`, via the VM). +Repo: `/home/super/Projects/NetVM` (git, local-only, no remotes). + +Companion tools (all read-only unless noted): +- `bin/dm-log-taxonomy.py` — failure taxonomy over dm-log.jsonl (counts, nav-failure + breakdown per target/agent-pair, hourly bursts, ghost rate). Run: `python3 bin/dm-log-taxonomy.py` +- `bin/placement-audit.py` — human-readable placement/gate activity. + `placement-audit.py --since 24h`, `--since 60m`, `--watch` +- `bin/tests/test_followup_fixes.py` — 25 unit tests pinning the sweeper/harvester/gate contracts + +--- + +## 1. Architecture (10 lines) + +1. Operator (or timer) runs `bin/dm.py send --agent --to --target ""`. +2. `dm.py` resolves the target name → thread UUID: `SIDCHAT_ALIASES` (intentionally + empty — hardcoded UUIDs rot), then `job-sidechats.json` dynamic mapping, then + passthrough name for sidebar search. +3. Unknown names autoprovision: `sidechat create` in the recipient's browser, and the + fresh UUID is registered in `job-sidechats.json` (self-healing). +4. Browser control goes `netvm-exec.sh -- python3 muse-chat-api.py --account `, + which drives headless Chromium over CDP inside the node's Warp netns. +5. `dm.py` navigates to `/thread/` (or sidebar title search), then the + **pre-send gate** (2026-10-04) independently samples the browser URL and requires + the expected UUID in it — fail-closed, never sends blind. +6. Send happens via CDP; `verify_placement()` re-navigates and reads the thread back + (post-send check, complementary to the pre-send gate). +7. Every step appends machine-readable events to `dm-log.jsonl` (`send_start`, + `nav_ok`, `sent`, `verified`, `placement_failed`, `pre_send_assert_failed`, …). +8. If the message carries `reply:expected` tags, `_register_followup()` arms a record in + `followups.json` (deadline, nudges, escalate target). +9. `followup-sweeper.py` (driven by gravity loop) sends nudges on expiry — in-thread + first, main chat on the final nudge — then escalates. It backfills `thread_uuid` + when a nudge provisions the thread. +10. `response-harvester.py` (every ~60s via `response-harvester.service`) reads agent + threads, records `[RESULT job_id]` markers to `job-log.jsonl`, and resolves + followups when an assistant reply lands in the matching thread. + +--- + +## 2. Failure-mode catalog + +How to read the signatures: `grep '"type": ""' dm-log.jsonl`, or use +`placement-audit.py --since 24h` / `dm-log-taxonomy.py`. + +### 2a. Landing-page nav (caught by the pre-send gate) + +- **Signature:** `{"type":"pre_send_assert_failed","id":…,"agent":…,"to":…, + "target":…,"expected_uuid":…,"actual_url":"https://muse.ai/", + "reason":"url_mismatch"|"not_on_thread_page"|"direct_nav_failed"}` +- **Cause:** the browser never left the muse.ai landing page (or parked on the wrong + thread) after `sidechat use`; the old code would have sent anyway. +- **Effect:** send aborted, exit 1, nothing delivered, nothing marked verified. +- **Fix:** re-run the send; if it repeats, the nav layer is flapping — see §2g and the + nav workstream. The gate doing its job is *good news*, not a bug. +- **Owner:** dm.py placement-gate workstream (landed 2026-10-04). + +### 2b. `sidechat_uuid_capture_failed` (autoprovision went nowhere) + +- **Signature:** `{"type":"sidechat_uuid_capture_failed","id":…,"out":"https://muse.ai/"}` +- **Cause:** `sidechat create` ran but UUID capture saw the landing page — the thread + may or may not exist; dm.py cannot know. +- **Effect (pre-gate era):** the send proceeded with `thread_uuid: null` and reported + `verified:true` — a ghost send (see §2d). With the gate live, the send now fails + closed instead. +- **Owner:** dm.py autoprovision path; gate workstream. + +### 2c. Stale UUID rotation (the UUID you have is dead) + +- **Signature:** `{"type":"placement_failed",…,"expected_uuid":"", + "actual_url":"https://muse.ai/thread/","reason":"url_mismatch"}` + or `{"type":"placement_mismatch",…,"expected_uuid":…,"actual_uuid":…}`. +- **Cause:** thread UUIDs rotate — observed same-day (`heartbeat-opm`: + `5bd5b350` → `0077e918` → dead). Any hardcoded UUID is a time bomb; the SPA may + also redirect a dead-UUID nav to whatever thread was last open (misdelivery, not + just failure). +- **Fix:** never hardcode UUIDs — `SIDCHAT_ALIASES` in dm.py is intentionally empty + (sibling workstream removed the last hardcoded entries 2026-10-04, incl. the + twice-dead heartbeat UUID). Use name-based autoprovision + `job-sidechats.json` + dynamic mappings, which self-heal on re-provision. +- **Owner:** UUID-rotation workstream. TODO: `docs/UUID-ROTATION.md` not yet written; + this section is the interim reference. + +### 2d. Placement-blind `verified:true` (the ghost send) + +- **Signature:** `{"type":"verified","id":…,"agent":…,"target":…}` with **no** + `thread_uuid` field (or `"thread_uuid": null`), paired with a `sent` event — + and no `placement_failed`/`pre_send_assert_failed` anywhere near it. +- **Cause:** the old verify step checked the *current* browser chat, not the target + (see operator AGENTS.md). A send parked on the landing page still "verified". +- **Effect:** the message is lost but every downstream system believes it landed — + followups nudge a ghost, pipelines stall as `dispatched`. +- **Fix:** the pre-send gate (§2a) makes new ghost sends impossible (fail-closed). + For historical ghosts: backfill `thread_uuid` in `followups.json` once the thread + provisions, register the alias in `job-sidechats.json`, resolve manually with a note. +- **Owner:** gate workstream (prevention); operators (historical repair). + +### 2e. Logged-out / dead session + +- **Signature:** no single event type. Look for `nav_failed` with `"rc": 1` and CDP + tracebacks, or muse-chat-api.py login-state detection: page `Title == "muse.ai"` + (LANDING), body containing `"To log in, enter the code"` (OTP_PROMPT), or + `"Your email matches multiple accounts"` (ACCOUNT_SELECTION). +- **Cause:** Meta session expired, OTP pending, or (2026-10-04 ~19:43 UTC) all four + chromebox browsers dying silently every 10–15 min with no crash log. +- **Fix:** `agent-health.sh` watchdog relaunches dead Chromium; OTP/account-selection + states need the human (approval flow). Track: board task `9e8b15d69ca9` for the + 2026-10-04 browser-flapping root cause. +- **Owner:** session-state workstream. TODO: dedicated session-probe tooling pending. + +### 2f. Ghost followup (`thread_uuid: null` — can never auto-resolve) + +- **Signature:** in `followups.json`: `"thread_uuid": null`, `"status": "pending"`; + in dm-log: the original send's `sidechat_uuid_capture_failed`, then sweeper + `followup_nudged` events that never resolve. +- **Cause:** the harvester's `clear_matching_followups` only matches exact + `thread_uuid` (or `target=="main"` in main). A null-UUID followup matches nothing, + forever — the sweeper nudges then escalates on a ghost. +- **Fix:** sweeper now backfills `thread_uuid` when a nudge successfully provisions + the thread (workstream landed 2026-10-04). For existing ghosts: backfill manually, + register the alias, resolve with a note. Longer-term TODO: sweeper should also + backfill on sweep ticks by scanning dm-log for the *original* dm_id's + `sidechat_autoprovisioned` event (currently only backfills on nudge success). +- **Owner:** sweeper-policy workstream. + +### 2g. Flaky nav (~1/3 success) and timing races + +- **Signature:** bursts of `nav_failed` / `pre_send_assert_failed` / + `placement_mismatch` for the same target in `placement-audit.py --since 60m`; + `dm-log-taxonomy.py` shows the per-target failure rate. +- **Cause:** under investigation 2026-10-04 — sidebar DOM races, CDP timing, + possibly the browser-flapping issue (§2e). The gate converts these from silent + misdeliveries into loud, retried failures (strictly better). +- **Owner:** nav-shootout, timing-races, and retry-module workstreams. + TODO: their findings land here when complete. + +### 2h. `main_chat_blocked` + +- **Signature:** `{"type":"main_chat_blocked","id":…,"agent":…,"to":…,"target":"main"}` +- **Cause:** a send targeted `main` without `--allow-main-chat`. Deliberate guardrail: + automation stays out of main chat unless explicitly allowed. +- **Fix:** pass `--allow-main-chat` if main delivery is really intended. + +--- + +## 3. Diagnosis playbook: "a DM didn't arrive" + +Run in order. All commands on bl unless noted. + +1. **Find the send.** `grep '"id": "<8-hex-id>"' /home/super/Projects/NetVM/dm-log.jsonl` + (or grep by target). Walk its event sequence: `send_start` → `nav_ok`? → + `sent`? → `verified`? Any `*_failed` event is your answer. +2. **Placement first.** `python3 bin/placement-audit.py --since 24h` — if you see + `pre_send_assert_failed` / `placement_failed` / `placement_mismatch` for the id, + the message never left bl correctly. Check `actual_url`: landing page (§2a), + wrong thread (§2c). +3. **Check the taxonomy.** `python3 bin/dm-log-taxonomy.py` — is this an isolated + incident or is the target/agent-pair failing at a high rate right now (§2g)? +4. **Check the session.** If `nav_failed` with rc=1: is the browser alive? + `ss -tln` for the CDP port in the netns; muse-chat-api.py login-state detection + (LANDING vs LOGGED_IN per operator AGENTS.md). If all four browsers are flapping, + see board task `9e8b15d69ca9` (§2e). +5. **Check UUID freshness.** `python3 -c "import json;print(json.load(open( + '/home/super/Projects/NetVM/job-sidechats.json')).get(''))"` — does the + mapped UUID still open the right thread? If the send used a UUID from anywhere + else, suspect rotation (§2c). +6. **Check the followup.** `grep -A12 '' /home/super/Projects/NetVM/followups.json` + — `thread_uuid: null` + `pending` = ghost (§2f). If nudges are firing, the reply + (if any) landed in a thread the harvester doesn't match: backfill + resolve manually. +7. **Check the pipeline.** If it was a pipeline step: `grep '' + /home/super/Projects/NetVM/pipelines.json` — `dispatched` with no result after a + ghost send needs `pipeline_engine.record_step_result()` + a decision on chaining. +8. **Repair, don't re-fire blindly.** Fix the cause (nav, session, UUID), then resend + once. Repeated blind retries of a ghost just burn the nudge budget. + +--- + +## 4. Hard lessons (non-negotiable) + +1. **Never trust `verified:true` alone.** It historically meant "found the msg_id in + whatever chat the browser happened to be parked in." Always corroborate with a + placement assertion (pre-send gate) or a thread read-back. +2. **Assert the post-nav URL before sending.** The gate is now the enforcement; any + new send path must go through it. Fail closed, fail loudly (non-zero exit + + `pre_send_assert_failed` event), never degrade to blind send. +3. **Thread UUIDs rotate — never hardcode them.** Observed same-day rotation. + `SIDCHAT_ALIASES` stays empty; name-based autoprovision + `job-sidechats.json` + self-heals. If you find a hardcoded UUID anywhere, delete it. +4. **Backfill `thread_uuid`.** A followup with null UUID can never auto-resolve. + Backfill on provision (sweeper does this now), and repair historical ghosts by + hand with a note. +5. **Atomic writes on live files.** siphon/response-harvester fire every ~60s; the + sweeper loop is tight. Edit to `/tmp`, `py_compile`, atomic `mv`. Never half-write + a live file. Back up (`*.bak--`) before touching shared code. +6. **One owner per file per change.** Parallel fixes on shared surfaces caused the + 2026-10-03 502s: reconcile first, disjoint ownership, review the combined diff + before it goes live. dm.py is the single chokepoint — treat it accordingly. +7. **The dm-log is the source of truth for delivery; the harvester is the source of + truth for replies.** `verified` ≠ delivered; `sent` ≠ read. When they disagree, + believe the thread read-back. + +--- + +## 5. Sibling workstream map (2026-10-04 troubleshooting sprint) + +| # | Workstream | Output | Status | +|---|-----------|--------|--------| +| 1 | Gate restore | pre-send gate in `dm.py` (`assert_pre_send_placement`) | **Landed** | +| 2 | Log taxonomy | `bin/dm-log-taxonomy.py` | **Landed** | +| 3 | UUID rotation | hardcoded UUID removal (`SIDCHAT_ALIASES` emptied) | **Landed** | +| 4 | Nav shootout | — | TODO: findings pending | +| 5 | Session-state | — | TODO: probe tooling pending | +| 6 | Timing races | — | TODO: findings pending | +| 7 | Retry module | shared retry helper | TODO: pending | +| 8 | Sweeper policy | thread_uuid backfill + `final_nudge_target` | **Landed** | +| 9 | Audit CLI | `bin/placement-audit.py` | **Landed** | +| 10 | This runbook | `docs/SIDECHAT-RELIABILITY.md` | **This file** | + +Related docs: `docs/DM-SPEC.md` (DM system spec), `docs/SIDECHAT_SPEC.md`, +`docs/SIDECHAT-POLICY-TRIAGE.md`, `docs/CHROMEBOX-RUNBOOK.md`. +TODO: `docs/UUID-ROTATION.md` (rotation mechanics deep-dive) — §2c is the interim reference. diff --git a/docs/UUID-ROTATION.md b/docs/UUID-ROTATION.md new file mode 100644 index 0000000..181f488 --- /dev/null +++ b/docs/UUID-ROTATION.md @@ -0,0 +1,132 @@ +# Thread-UUID rotation: measurement & mapping hygiene proposal + +Date: 2026-10-04. Source: `dm-log.jsonl` (window 2026-10-03 21:36Z → 2026-10-04 19:46Z, +~22h, 3318 lines — the log does not cover a full 7 days; rates below are +extrapolated from this window and should be re-measured on a longer one). +Read-only analysis; no code changed. + +## 1. Measured rotation per alias + +Two distinct phenomena share the symptom "alias → different UUID": + +- **(A) Real thread death/rotation.** The SPA retires threads. Observed earlier + for heartbeat-opm (5bd5b350 → 0077e918 → dead, twice in one day, per operator + notes; predates this log window). +- **(B) Fuzzy-search nondeterminism.** `sidechat use ""` scans every DOM + element for `textContent.includes(name)`, picks the shortest match, and clicks + it (`muse-chat-api.py: cmd_sidechat_use`). The candidate set changes between + React renders, so the same alias resolves to different live threads run to run + — with no thread dying at all. + +| alias | distinct UUIDs (window) | autoprovisioned (real rotations) | send-target flaps (fuzzy) | changes/day* | +|---|---|---|---|---| +| heartbeat | 4 (5f18476d, 1e75a740, 0077e918, …) | 0 | 3 | ~3.3 flap | +| 646 tasks | 4 (1e75a740, a7d0b08a, 1dfb3199) | 1 (18:30 → 1dfb3199) | 2 | ~1.1 real, ~2.2 flap | +| 646-pip | 2 (82779e55 → 75feb3a2) | 1 (18:01) | 0 | ~1.1 real | +| ops-audit | 2 (75feb3a2 → 34a8a87c) | 1 (19:28) | 0 | ~1.1 real | +| main-loop brain | 1 (5f18476d) | 0 | — | collision, see §3 | +| pipe-* (6 aliases) | 1 each | 1 each (creation) | 0 | 0 post-creation | +| 646-pip-coord | 1 (4466d0c1) | 0 | 0 | 0 | +| 646-opm-coord | 1 (4139dd4e) | 0 | 0 | 0 | + +\* changes/day extrapolated from ~22h window; demo-day traffic inflates flaps. + +Smoking gun for (B): at 16:25:53 a `super`→646 send for "646 tasks" landed on +a7d0b08a, and 52 seconds later an `opm`→646 send for the same alias in the same +recipient browser landed on 1e75a740. No thread died in between — the fuzzy +search is nondeterministic across runs. + +## 2. Who writes / reads job-sidechats.json today + +Writers: +- `dm.py` autoprovision path only (atomic tmp+mv write of + `{thread_uuid, agent, created_at}` on `sidechat_autoprovisioned`). +- Manual edits (e.g. today's removal of stale heartbeat-opm / pipe-9735f2). + +NOT writers: +- The new sweeper thread_uuid backfill (workstream 1) writes **followups.json + only** — it does not refresh job-sidechats.json. +- `thread_lifecycle.py` exists but is marked "prototype, not deployed". + +Readers: `dm.py resolve_sidechat_target` (exact key match → UUID, else +passthrough to fuzzy search), harvester/siphon (monitor list), box-ctl, +job-dispatch, super-cli. + +Critical gap in the refresh story: a **dead mapped UUID does not self-heal**. +`sidechat use ` → SPA redirects to `/` → output is +"Navigated to: https://muse.ai/" (not NOTFOUND) → `nav_failed no_thread_url` → +hard fail. Autoprovision only fires on NOTFOUND from *name* search. So a dead +mapping = permanent send failures for that alias until a human removes the +entry. (This is fail-closed, not misdelivery — but it is not self-healing.) + +Second gap: an **unmapped alias whose fuzzy search keeps "succeeding" never +autoprovision** and therefore never earns a canonical mapping — it flaps +forever. This is heartbeat's exact situation right now (see §3). + +## 3. Flags + +- **MISSING mapping, highest churn: `heartbeat`.** 4 UUIDs in 3h, zero + autoprovision events, no entry in job-sidechats.json (the old `heartbeat-opm` + entry was removed during today's hardcoded-alias cleanup). Every send does a + fresh fuzzy search. At 16:45 a heartbeat send landed on 1e75a740 — 646 tasks' + thread. **Misdelivery already happened here.** +- **MISSING mapping: `ops-audit`.** Re-provisioned 19:28 (34a8a87c) but the + alias was never registered; next fuzzy search may flap. +- **COLLISION, unverified: `main-loop brain` → 5f18476d.** The brain's init + send (19:40) fuzzy-matched into 5f18476d — the same thread heartbeat used at + 16:35 and 19:41. No `sidechat_autoprovisioned` event for the brain alias, so + no new thread was created. Either the fuzzy search false-positived into + heartbeat's thread, or the brain was deliberately placed there. **Confirm + intent — if unintended, digests and heartbeat traffic are cross-talking.** +- **No current mapping points to a UUID superseded by a newer autoprovision + for the same alias** — all 11 entries match their alias's latest + autoprovision event. Nothing stale in that narrow sense. +- **Aging, low-risk:** `test-auto-prov` (last confirmed 16:20), `646-opm-coord` + (last confirmed 12:52), and six single-use `pipe-*` demo entries accumulate + with no TTL. Harmless today, cruft tomorrow. +- **Confounder:** as of ~19:43Z all four chromebox browsers are flapping + (dying every 10–15 min, no crash signature). Nav flakiness measured today is + polluted by this; re-measure rotation after the browsers stabilize. + +## 4. Proposal: mapping hygiene + +1. **Canonical entry for every recurring alias.** `heartbeat`, `ops-audit`, + `main-loop brain` (once its thread is confirmed), `646 tasks`, `646-pip`, + `646-pip-coord`, `646-opm-coord` must each have exactly one entry, created + by explicit provisioning — never by fuzzy luck. Unmapped + fuzzy-matched is + the misdelivery configuration; it should be treated as a bug, not a + fallback. +2. **Dead-UUID reprovisioning.** On `no_thread_url` / `uuid_mismatch` where + the nav target came from a *mapping* (source == `dynamic_mapping`), fall + through to autoprovision and overwrite the mapping — the same path name + search uses today. This closes the "dead mapping = permanent failure" gap + and makes rotation self-healing. (Keep failing closed when even + autoprovision can't produce a thread page.) +3. **Exact-match-first fuzzy search.** Before the DOM substring scan, + `sidechat use` should try an exact title match (normalized case/space). + Only fall back to substring on no exact hit. This kills most cross-alias + false positives (the 16:45 heartbeat→646-tasks landing) without changing + the autoprovision contract. +4. **TTL + audit job.** Add `last_confirmed_at` to each mapping entry, refreshed + on every successful send through it. A periodic audit (daily is plenty; + weekly at minimum) flags: entries unconfirmed > TTL (7d job, 24h heartbeat), + aliases seen in dm-log with no entry, and UUIDs serving >1 alias + (collision detector — would have caught 1e75a740 and 5f18476d). Report-only + at first; auto-prune single-use `pipe-*`/test entries after TTL. +5. **Sweeper backfill → also refresh the mapping.** The workstream-1 backfill + already learns the live UUID per followup; writing it through to + job-sidechats.json when the followup's target is a mapped alias is nearly + free and covers the "nudge provisioned a thread dm.py didn't" case. +6. **Do not reintroduce hardcoded UUIDs** (the empty `SIDCHAT_ALIASES` was the + right call). The mapping file is the single source of truth; the audit job + in (4) is what keeps it honest. + +## 5. Suggested next steps (no code in this workstream) + +- Confirm `main-loop brain` thread intent (collision with heartbeat's 5f18476d). +- Provision canonical threads for `heartbeat` and `ops-audit`; register them. +- Implement proposal items 2 and 3 (dm.py + muse-chat-api.py — small, surgical). +- Stand up the audit job (4) as report-only; review one week of flags before + enabling any auto-prune. +- Re-run this measurement on a 7-day dm-log window once rotation stabilizes + post-browser-flap.