#!/usr/bin/env python3 """main-chat-watchdog.py — enforce the sidechat-first DM policy. Tails dm-log.jsonl (read-only) and flags any DM actually SENT to main chat without an explicit allow_main_chat marker, plus any placement_mismatch / placement_failed events (message landed in the wrong chat). Distinguishes: VIOLATION : type=sent, target=main, no tags.allow_main_chat -> policy broken (P0) AUTHORIZED: type=sent, target=main, tags.allow_main_chat=true -> explicit opt-in BLOCKED : type=main_chat_blocked -> policy working MISMATCH : type=placement_mismatch -> placed in wrong chat (P0) PFAILED : type=placement_failed -> placement hard-failed (P0) State: byte-offset watermark in main-chat-watchdog.state (handles log rotation via inode check). First run starts at EOF (no historical backfill — old entries predate the allow_main_chat audit marker). Violations are appended to logs/main-chat-violations.jsonl and printed to stdout (journal). Exit 0 = clean, 1 = violations/P0s found, 2 = error. Read-only: never modifies dm-log.jsonl. """ import json import os import sys from datetime import datetime, timezone BASE = "/home/super/Projects/NetVM" DM_LOG = os.path.join(BASE, "dm-log.jsonl") STATE_FILE = os.path.join(BASE, "main-chat-watchdog.state") VIOLATIONS_LOG = os.path.join(BASE, "logs", "main-chat-violations.jsonl") def utcnow(): return datetime.now(timezone.utc).isoformat() def load_state(): try: with open(STATE_FILE) as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError, ValueError): return {} def save_state(state): tmp = STATE_FILE + ".tmp" with open(tmp, "w") as f: json.dump(state, f) os.replace(tmp, STATE_FILE) def main(): try: st = os.stat(DM_LOG) except FileNotFoundError: print(f"watchdog ERROR: {DM_LOG} not found", file=sys.stderr) return 2 state = load_state() # First run (or rotation): start at EOF, don't backfill history that # predates the allow_main_chat audit marker. if state.get("inode") != st.st_ino: offset = st.st_size if state: print(f"watchdog: log rotated or first run (inode {st.st_ino}), " f"starting at EOF offset {offset}") else: offset = min(state.get("offset", 0), st.st_size) violations = [] # P0: gate bypass (sent to main, no opt-in) mismatches = [] # P0: placement_mismatch / placement_failed blocked = 0 authorized = 0 scanned = 0 malformed = 0 try: with open(DM_LOG, "r", encoding="utf-8", errors="replace") as f: f.seek(offset) for line in f: line = line.strip() if not line: continue scanned += 1 try: ev = json.loads(line) except json.JSONDecodeError: malformed += 1 continue etype = ev.get("type") if etype == "main_chat_blocked": # Policy working: the dm.py gate refused a main send. blocked += 1 elif etype == "placement_mismatch": # P0: verify-time URL check found the browser parked in a # different chat than the intended target. The send # completed but landed in the wrong place. Schema has two # variants: sidechat ("expected_uuid") and main-drift # ("expected": "main"). mismatches.append({ "ts": utcnow(), "kind": "placement_mismatch", "severity": "P0", "dm_id": ev.get("id"), "agent": ev.get("agent"), "to": ev.get("to"), "target": ev.get("target"), "expected_uuid": ev.get("expected_uuid") or ev.get("expected"), "actual_uuid": ev.get("actual_uuid"), "attempt": ev.get("attempt"), "event_ts": ev.get("ts"), }) elif etype == "placement_failed": # P0: the authoritative post-send placement check failed # hard (sender exited 1, send NOT marked verified). d = ev.get("detail") or {} mismatches.append({ "ts": utcnow(), "kind": "placement_failed", "severity": "P0", "dm_id": ev.get("id"), "agent": ev.get("agent"), "to": ev.get("to"), "target": ev.get("target"), "reason": ev.get("reason") or d.get("reason"), "expected_uuid": ev.get("expected_uuid") or d.get("expected_uuid"), "actual_url": ev.get("actual_url") or d.get("actual_url"), "loop_attempt": ev.get("loop_attempt"), "event_ts": ev.get("ts"), }) elif etype == "sent" and ev.get("target") == "main": tags = ev.get("tags") or {} if tags.get("allow_main_chat"): authorized += 1 else: violations.append({ "ts": utcnow(), "kind": "main_chat_send", "severity": "P0", "dm_id": ev.get("id"), "agent": ev.get("agent"), "to": ev.get("to"), "target": "main", "msg_preview": (ev.get("msg") or "")[:120], "event_ts": ev.get("ts"), }) new_offset = f.tell() except OSError as e: print(f"watchdog ERROR reading log: {e}", file=sys.stderr) return 2 save_state({"offset": new_offset, "inode": st.st_ino}) findings = violations + mismatches summary = (f"watchdog: scanned={scanned} blocked={blocked} " f"authorized_main={authorized} " f"violations={len(violations)} " f"placement_mismatches={len(mismatches)} " f"malformed={malformed}") print(summary) if findings: try: os.makedirs(os.path.dirname(VIOLATIONS_LOG), exist_ok=True) with open(VIOLATIONS_LOG, "a", encoding="utf-8") as vf: for v in findings: vf.write(json.dumps(v) + "\n") except OSError as e: print(f"watchdog ERROR writing violations log: {e}", file=sys.stderr) return 2 for v in violations: print(f"VIOLATION main-chat send without opt-in: " f"id={v['dm_id']} agent={v['agent']} to={v['to']} " f"at={v['event_ts']} preview={v['msg_preview']!r}") for m in mismatches: if m["kind"] == "placement_mismatch": print(f"P0 PLACEMENT_MISMATCH message landed in wrong chat: " f"id={m['dm_id']} agent={m['agent']} to={m['to']} " f"target={m['target']} expected={m['expected_uuid']} " f"actual={m['actual_uuid']} at={m['event_ts']}") else: print(f"P0 PLACEMENT_FAILED placement check failed: " f"id={m['dm_id']} agent={m['agent']} to={m['to']} " f"target={m['target']} reason={m['reason']} " f"expected={m['expected_uuid']} " f"actual_url={m['actual_url']} at={m['event_ts']}") return 1 return 0 if __name__ == "__main__": sys.exit(main())