box/bin/placement-audit.py

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