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