box/bin/dm-log-taxonomy.py

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