box/bin/box-chat.py

#!/usr/bin/env python3
"""box-chat.py — allowlisted bl helper for Box thread oversight (READ-ONLY).

The board server (VM) never scrapes browsers directly. All thread reads go
through this helper, invoked as:

    /home/super/Projects/NetVM/bin/box-chat.py thread-list <agent> [--kind K] [--limit N]
    /home/super/Projects/NetVM/bin/box-chat.py thread-messages <agent> <thread-id> [--limit N] [--before MSGID]

Security properties (mirrors box-ctl.py):
- Fixed verb set; every argument validated before acting.
- <agent> must be a known node (muse, pip, 646, opm); anything else exits
  before any netns/SSH/CDP work.
- <thread-id> must match ^[a-zA-Z0-9-]{1,64}$ or be the literal "main".
- READ-ONLY by construction: the CDP driver (box-chat-cdp.py) only runs
  Runtime.evaluate reads plus benign navigation clicks. No sends, no
  composer interaction, no shell=True anywhere. All subprocess calls use
  argv lists.
- Every invocation audit-logged to box-chat.jsonl with caller identity.

Output: JSON to stdout ({"ok": true, ...} or {"ok": false, ...}),
exit 0 on success, nonzero on failure.
"""

import json
import os
import re
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

NETVM_ROOT = Path("/home/super/Projects/NetVM")
BIN = NETVM_ROOT / "bin"
CDP_DRIVER = BIN / "box-chat-cdp.py"
NETVM_EXEC = BIN / "netvm-exec.sh"
CHAT_LOG = NETVM_ROOT / "box-chat.jsonl"
DM_LOG = NETVM_ROOT / "dm-log.jsonl"

VALID_AGENTS = {"muse", "pip", "646", "opm"}
VALID_KINDS = {"main", "sidechat", "dm", "all"}
AGENT_RE = re.compile(r"^[a-z0-9-]{1,64}$")
THREAD_RE = re.compile(r"^[a-zA-Z0-9-]{1,64}$")
MSGID_RE = re.compile(r"^[a-zA-Z0-9-]{1,128}$")

REL_MONTHS = {m: i + 1 for i, m in enumerate(
    ["jan", "feb", "mar", "apr", "may", "jun",
     "jul", "aug", "sep", "oct", "nov", "dec"])}


def utcnow():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def out(ok, **kw):
    payload = {"ok": ok}
    payload.update(kw)
    print(json.dumps(payload))


def fail(code, error, detail=None, exit_code=1):
    payload = {"ok": False, "code": code, "error": error}
    if detail is not None:
        payload["detail"] = detail
    print(json.dumps(payload))
    sys.exit(exit_code)


def audit(action, agent=None, name=None, kind=None):
    """Append invocation record to the bl-side log."""
    try:
        entry = {
            "ts": utcnow(),
            "action": action,
            "agent": agent,
            "name": name,
            "kind": kind,
            "caller": os.environ.get("BOX_CALLER", "unknown"),
        }
        with open(CHAT_LOG, "a") as f:
            f.write(json.dumps(entry) + "\n")
    except Exception:
        pass  # audit failure must not break the action


def check_agent(agent):
    if not agent or agent not in VALID_AGENTS:
        fail("BAD_AGENT", "agent must be one of %s" % sorted(VALID_AGENTS),
             {"field": "agent", "value": agent})
    return agent


def check_thread_id(tid):
    if tid != "main" and not (tid and THREAD_RE.match(tid)):
        fail("BAD_THREAD", "thread-id must be 'main' or match ^[a-zA-Z0-9-]{1,64}$",
             {"field": "thread-id", "value": tid})
    return tid


def check_limit(val, default):
    if val is None:
        return default
    try:
        n = int(val)
    except (TypeError, ValueError):
        fail("BAD_LIMIT", "limit must be an integer 1-200", {"value": val})
    if not 1 <= n <= 200:
        fail("BAD_LIMIT", "limit must be an integer 1-200", {"value": val})
    return n


def run_cdp(agent, op, args, timeout):
    """Run the CDP driver inside the agent's netns. argv only, no shell."""
    cmd = [str(NETVM_EXEC), agent, "--", sys.executable,
           str(CDP_DRIVER), agent, op] + args
    try:
        r = subprocess.run(cmd, capture_output=True, text=True,
                           timeout=timeout)
    except subprocess.TimeoutExpired:
        fail("CDP_TIMEOUT", "CDP driver timed out in netns",
             {"agent": agent, "op": op})
    if r.returncode != 0 and not r.stdout.strip():
        fail("CDP_ERROR", "CDP driver failed",
             {"stderr": (r.stderr or "")[-300:]})
    try:
        data = json.loads(r.stdout.strip())
    except Exception:
        fail("CDP_ERROR", "CDP driver returned non-JSON",
             {"stdout": r.stdout[-300:], "stderr": (r.stderr or "")[-300:]})
    if not data.get("ok"):
        code = data.get("code", "CDP_ERROR")
        if "THREAD_NOT_FOUND" in str(data.get("error", "")):
            code = "THREAD_NOT_FOUND"
        fail(code, data.get("error", "cdp failed"))
    return data


def parse_rel_time(rel):
    """Panel relative times ('just now', '4m', '2h', '3d', 'Oct 3') ->
    (approx ISO8601, True). Returns (None, False) when unparseable."""
    now = datetime.now(timezone.utc)
    rel = (rel or "").strip().lower()
    if rel in ("just now", "now"):
        return now.strftime("%Y-%m-%dT%H:%M:%SZ"), True
    m = re.match(r"^(\d+)\s*m(in(ute)?s?)?$", rel)
    if m:
        return (now - timedelta(minutes=int(m.group(1)))).strftime("%Y-%m-%dT%H:%M:%SZ"), True
    m = re.match(r"^(\d+)\s*h((ou)?rs?)?$", rel)
    if m:
        return (now - timedelta(hours=int(m.group(1)))).strftime("%Y-%m-%dT%H:%M:%SZ"), True
    m = re.match(r"^(\d+)\s*d(ays?)?$", rel)
    if m:
        return (now - timedelta(days=int(m.group(1)))).strftime("%Y-%m-%dT%H:%M:%SZ"), True
    m = re.match(r"^([a-z]{3})\s+(\d{1,2})$", rel)
    if m and m.group(1) in REL_MONTHS:
        dt = now.replace(month=REL_MONTHS[m.group(1)], day=int(m.group(2)),
                         hour=12, minute=0, second=0, microsecond=0)
        if dt > now:
            dt = dt.replace(year=dt.year - 1)
        return dt.strftime("%Y-%m-%dT%H:%M:%SZ"), True
    return None, False


def dm_conversations(agent):
    """DM conversations involving <agent>, synthesized from dm-log.jsonl.

    Real timestamps and counts; bodies are not stored in the log (by design)
    — message text for these threads is the wire tag, and full bodies live
    in the target chat's messages.
    """
    convos = {}
    try:
        with open(DM_LOG) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    e = json.loads(line)
                except Exception:
                    continue
                if e.get("type") not in ("sent", "send_done"):
                    continue
                frm, to = e.get("agent"), e.get("to")
                if not frm or not to:
                    continue
                if agent not in (frm, to):
                    continue
                key = tuple(sorted([frm, to]))
                c = convos.setdefault(key, {"count": 0, "last_ts": "",
                                            "entries": []})
                c["count"] += 1
                if e.get("ts", "") > c["last_ts"]:
                    c["last_ts"] = e["ts"]
                c["entries"].append(e)
    except FileNotFoundError:
        return []
    out = []
    for (a, b), c in sorted(convos.items()):
        out.append({
            "id": "dm-%s-%s" % (a, b),
            "kind": "dm",
            "title": "dm:%s:%s" % (a, b),
            "participants": [a, b],
            "last_message_at": c["last_ts"] or None,
            "last_message_approx": False,
            "message_count": c["count"],
        })
    return out


def dm_thread_messages(agent, thread_id):
    """Messages for a dm-<a>-<b> thread, from dm-log.jsonl (complete log)."""
    parts = thread_id[3:].split("-")
    if len(parts) != 2:
        fail("THREAD_NOT_FOUND", "no such DM thread: %s" % thread_id)
    a, b = parts
    if agent not in (a, b):
        fail("THREAD_NOT_FOUND", "no such DM thread: %s" % thread_id)
    msgs = []
    try:
        with open(DM_LOG) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    e = json.loads(line)
                except Exception:
                    continue
                if e.get("type") not in ("sent", "send_done"):
                    continue
                frm, to = e.get("agent"), e.get("to")
                if not frm or not to:
                    continue
                if tuple(sorted([frm, to])) != (a, b):
                    continue
                sender = e.get("agent")
                msgs.append({
                    "id": str(e.get("id", "")),
                    "from": {"role": "agent", "name": sender},
                    "text": "[from:%s] [id:%s] -> %s" % (
                        sender, e.get("id"), e.get("target", "?")),
                    "ts": e.get("ts"),
                    "target": e.get("target"),
                    "verified": e.get("verified"),
                })
    except FileNotFoundError:
        pass
    msgs.sort(key=lambda m: m.get("ts") or "")
    # de-dupe send_done/sent pairs on DM id, keep the richer record
    seen = {}
    for m in msgs:
        prev = seen.get(m["id"])
        if prev is None or (m.get("verified") and not prev.get("verified")):
            seen[m["id"]] = m
    msgs = sorted(seen.values(), key=lambda m: m.get("ts") or "")
    return msgs


def act_thread_list(agent, kind, limit):
    threads = []
    if kind in ("main", "sidechat", "all"):
        data = run_cdp(agent, "threads", [], timeout=240)
        for t in data.get("threads", [])[:limit]:
            if kind != "all" and t.get("kind") != kind:
                continue
            ts, approx = parse_rel_time(t.get("rel", ""))
            threads.append({
                "id": t.get("id"),
                "kind": t.get("kind"),
                "title": t.get("title"),
                "participants": [agent, "human"],
                "last_message_at": ts,
                "last_message_approx": approx,
                "message_count": None,  # list is a panel scrape; counts need a thread open
            })
    if kind in ("dm", "all"):
        threads.extend(dm_conversations(agent))
    audit("thread-list", agent=agent, kind=kind)
    out(True, agent=agent, kind=kind, threads=threads,
        fetched_at=utcnow())


def act_thread_messages(agent, thread_id, limit, before):
    if thread_id.startswith("dm-"):
        msgs = dm_thread_messages(agent, thread_id)
        thread = {"id": thread_id, "kind": "dm",
                  "title": "dm:%s" % thread_id[3:].replace("-", ":"),
                  "participants": thread_id[3:].split("-")}
    else:
        data = run_cdp(agent, "messages", [thread_id], timeout=150)
        raw = data.get("messages", [])
        thread = {"id": thread_id,
                  "kind": "main" if thread_id == "main" else "sidechat",
                  "participants": [agent, "human"]}
        msgs = [{
            "id": m.get("id"),
            "from": ({"role": "agent", "name": agent}
                     if m.get("author") == "assistant"
                     else {"role": "human", "name": "human"}),
            "text": m.get("text", ""),
            "ts": m.get("ts"),
        } for m in raw]
    if before:
        if not MSGID_RE.match(before):
            fail("BAD_CURSOR", "before must match ^[a-zA-Z0-9-]{1,128}$",
                 {"value": before})
        idx = next((i for i, m in enumerate(msgs) if m["id"] == before), None)
        if idx is None:
            fail("BAD_CURSOR", "no message with that id in loaded window",
                 {"value": before})
        msgs = msgs[:idx]
    older = len(msgs)
    if len(msgs) > limit:
        msgs = msgs[-limit:]
    next_before = msgs[0]["id"] if older > len(msgs) and msgs else None
    audit("thread-messages", agent=agent, name=thread_id)
    out(True, agent=agent, thread=thread, messages=msgs,
        next_before=next_before, loaded_count=older, fetched_at=utcnow())


USAGE = """usage: box-chat.py <action> [args]

  thread-list <agent> [--kind main|sidechat|dm|all] [--limit N]
  thread-messages <agent> <thread-id> [--limit N] [--before MSGID]

read-only. <agent> is one of: muse, pip, 646, opm."""


def parse_flags(rest, names):
    """Parse [--flag value] pairs; returns (positionals, {flag: value})."""
    pos, flags = [], {}
    i = 0
    while i < len(rest):
        tok = rest[i]
        if tok.startswith("--") and tok[2:] in names:
            if i + 1 >= len(rest):
                fail("BAD_ARGS", "flag %s needs a value" % tok)
            flags[tok[2:]] = rest[i + 1]
            i += 2
        elif tok.startswith("--"):
            fail("BAD_ARGS", "unknown flag: %s" % tok)
        else:
            pos.append(tok)
            i += 1
    return pos, flags


def main(argv):
    if len(argv) < 2:
        print(USAGE, file=sys.stderr)
        sys.exit(2)
    action = argv[1]

    if action == "thread-list":
        pos, flags = parse_flags(argv[2:], {"kind", "limit"})
        if len(pos) != 1:
            fail("BAD_ARGS", "usage: thread-list <agent> [--kind K] [--limit N]")
        agent = check_agent(pos[0])
        kind = flags.get("kind", "all")
        if kind not in VALID_KINDS:
            fail("BAD_ARGS", "kind must be one of %s" % sorted(VALID_KINDS))
        act_thread_list(agent, kind, check_limit(flags.get("limit"), 50))
    elif action == "thread-messages":
        pos, flags = parse_flags(argv[2:], {"limit", "before"})
        if len(pos) != 2:
            fail("BAD_ARGS",
                 "usage: thread-messages <agent> <thread-id> [--limit N] [--before MSGID]")
        agent = check_agent(pos[0])
        tid = check_thread_id(pos[1])
        act_thread_messages(agent, tid, check_limit(flags.get("limit"), 50),
                            flags.get("before"))
    else:
        print(USAGE, file=sys.stderr)
        fail("BAD_ARGS", "unknown action: %s" % action)


if __name__ == "__main__":
    main(sys.argv)