box/bin/dm.py

#!/usr/bin/env python3
"""
DM: Headless Direct Message API (muse.ai) — tagged, logged, verifiable.

Wire format (every send, signed or not):
    [from:<sender>] [id:<8-hex>] <body>

Signed DMs (produced by bin/dm-sign.sh, namespace "dm") append the SSH
signature block after a blank line; `verify-sig` checks it against the
sender's key in dm-signers/<sender>.pub. An unsigned message carrying a
[from:X] header is just a claim — only a GOOD verify-sig result is proof.

Every send is appended to dm-log.jsonl. Delivery is confirmed by reading
the RECIPIENT's chat for the message id (up to 3 attempts): SENT+VERIFIED
means the id was seen in the recipient's chat; FAILED means it wasn't
after 3 attempts. `send --raw` transmits verbatim (for pre-signed
messages): no tagging, no truncation — the signed payload must survive
byte-identical.

Usage:
  dm.py send --agent opm --to 646 --target main "message"
  dm.py send --agent opm --target main --raw "$(dm-sign.sh --from operator-main 'hi')"
  dm.py verify-sig --agent opm --target main        # scan recent reads for signed DMs
  dm.py verify-sig "$(dm-sign.sh --from operator-main 'hi')"  # verify text directly
  dm.py read --agent 646 --target main [n]
  dm.py log [--n 20]
  dm.py thread --from opm --to 646 --target main "message"
"""
import argparse
import os
import re
import subprocess
import time
import sys
import uuid
import json
import os
import urllib.request
import urllib.parse
from datetime import datetime, timezone, timedelta

API = "/home/super/Projects/NetVM/bin/muse-chat-api.py"
NETVM_EXEC = "/home/super/Projects/NetVM/bin/netvm-exec.sh"
VALID_AGENTS = ["muse", "pip", "646", "opm"]
VALID_SENDERS = ["muse", "pip", "646", "opm", "super"]
VALID_RECIPIENTS = ["muse", "pip", "646", "opm"]

# ---- Canonical follow-up tags (2026-10-04, DEPLOY-DECISIONS.md) ----
# Tags declare follow-up policy at send time. They are metadata only:
# stripped from the delivered text, recorded on send_start/sent log
# events. The request-store sweeper (VM side) guarantees nudges,
# retries, closure, and escalation. Untagged DMs behave exactly as before.
TAG_TIMEOUT_MIN = 60
TAG_TIMEOUT_MAX = 604800
TAG_TIMEOUT_DEFAULT = 3600  # 1h, per deploy decisions
TAG_NUDGES_MIN = 0
TAG_NUDGES_MAX = 10
TAG_NUDGES_DEFAULT = 2
TAG_ROUTE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
TAG_THREAD_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
# Trailing-token form: [reply:expected] [reply:timeout=7200] ...
TAG_TOKEN_RE = re.compile(r"\[([A-Za-z][A-Za-z0-9:]*)"
                           r"(?:=([^\]]*))?\]\s*$")


def parse_canonical_tag(item):
    """Parse one canonical tag token. Returns (key, value).
    Bare 'reply:expected' -> ('reply:expected', True).
    'route:X' / 'thread:X' use colon form; 'reply:timeout=N' etc use '='.
    Raises ValueError with a human message on any problem."""
    item = item.strip()
    if "=" in item:
        k, v = item.split("=", 1)
        k, v = k.strip(), v.strip()
    elif item.startswith("route:"):
        k, v = "route", item[len("route:"):].strip()
    elif item.startswith("thread:"):
        k, v = "thread", item[len("thread:"):].strip()
    else:
        k, v = item, True
    if k == "reply:expected":
        # Bare 'reply:expected' or 'reply:expected=' (empty value from
        # --tag flag generation); anything else is an error.
        if v is not True and v != "":
            raise ValueError("reply:expected takes no value")
        return k, True
    if k == "reply:timeout":
        try:
            n = int(v)
        except (TypeError, ValueError):
            raise ValueError("reply:timeout must be integer seconds")
        if not TAG_TIMEOUT_MIN <= n <= TAG_TIMEOUT_MAX:
            raise ValueError("reply:timeout must be %d..%d"
                             % (TAG_TIMEOUT_MIN, TAG_TIMEOUT_MAX))
        return k, n
    if k == "reply:nudges":
        try:
            n = int(v)
        except (TypeError, ValueError):
            raise ValueError("reply:nudges must be an integer")
        if not TAG_NUDGES_MIN <= n <= TAG_NUDGES_MAX:
            raise ValueError("reply:nudges must be %d..%d"
                             % (TAG_NUDGES_MIN, TAG_NUDGES_MAX))
        return k, n
    if k == "reply:escalate":
        if not v or not TAG_ROUTE_RE.fullmatch(str(v)):
            raise ValueError("reply:escalate must be an identity")
        return k, str(v)
    if k == "route":
        if not TAG_ROUTE_RE.fullmatch(str(v)):
            raise ValueError("route must match ^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
        return k, str(v)
    if k == "thread":
        if not TAG_THREAD_RE.fullmatch(str(v)):
            raise ValueError("thread must match ^[A-Za-z0-9_-]{1,64}$")
        return k, str(v)
    raise ValueError("unknown tag %r; known: reply:expected, reply:timeout=, "
                     "reply:nudges=, reply:escalate=, route:, thread:" % (k,))


def parse_tags(tag_list):
    """Validate --tag items. Returns canonical dict. Raises ValueError."""
    tags = {}
    for item in tag_list or []:
        k, v = parse_canonical_tag(item)
        if k in tags:
            raise ValueError("duplicate tag %r" % (k,))
        tags[k] = v
    # Cross-field consistency: policy tags require reply:expected.
    if "reply:expected" not in tags:
        for k in ("reply:timeout", "reply:nudges", "reply:escalate"):
            if k in tags:
                raise ValueError("tag %s requires reply:expected" % (k,))
    # Defaults when reply:expected without explicit policy.
    if "reply:expected" in tags:
        tags.setdefault("reply:timeout", TAG_TIMEOUT_DEFAULT)
        tags.setdefault("reply:nudges", TAG_NUDGES_DEFAULT)
    return tags


def extract_trailing_tags(message):
    """Pull trailing [tag] tokens off message text. Returns
    (stripped_message, tags_dict). Unknown brackets are left alone."""
    tags = {}
    text = message.rstrip()
    while True:
        m = TAG_TOKEN_RE.search(text)
        if not m:
            break
        key = m.group(1)
        # Only consume known canonical keys; anything else stays.
        probe = key + ("=" + m.group(2) if m.group(2) is not None else "")
        try:
            k, v = parse_canonical_tag(probe)
        except ValueError:
            break
        if k in tags:
            break  # duplicate; leave it in the text
        tags[k] = v
        text = text[:m.start()].rstrip()
    return text, tags


def merge_tags(flag_tags, cli_tags, text_tags):
    """Precedence: --tag > dedicated flags > trailing text tokens."""
    merged = dict(text_tags)
    merged.update(flag_tags)
    merged.update(cli_tags)
    # Defaults when reply:expected without explicit policy.
    if "reply:expected" in merged:
        merged.setdefault("reply:timeout", TAG_TIMEOUT_DEFAULT)
        merged.setdefault("reply:nudges", TAG_NUDGES_DEFAULT)
    return merged


def tags_from_flags(a):
    """Build canonical tags from argparse --expect-reply et al."""
    tags = {}
    if getattr(a, "expect_reply", False):
        tags["reply:expected"] = True
    if getattr(a, "reply_timeout", None) is not None:
        k, v = parse_canonical_tag("reply:timeout=%s" % a.reply_timeout)
        tags[k] = v
    if getattr(a, "reply_nudges", None) is not None:
        k, v = parse_canonical_tag("reply:nudges=%s" % a.reply_nudges)
        tags[k] = v
    if getattr(a, "reply_escalate", None):
        k, v = parse_canonical_tag("reply:escalate=%s" % a.reply_escalate)
        tags[k] = v
    if getattr(a, "route", None):
        k, v = parse_canonical_tag("route:%s" % a.route)
        tags[k] = v
    if getattr(a, "thread", None):
        k, v = parse_canonical_tag("thread:%s" % a.thread)
        tags[k] = v
    return tags

LOG_FILE = "/home/super/Projects/NetVM/dm-log.jsonl"

# Well-known sidechat name -> thread UUID aliases.
# These bypass fuzzy name matching for reliable placement.
# Resolved 2026-10-04. Add new entries as sidechats are created.
SIDCHAT_ALIASES = {
    "646-opm-work": "d410b9ad-f667-465f-a103-43fabc0f69fe",
}

def resolve_sidechat_target(target):
    """Resolve target alias or name to UUID dynamically from job-sidechats.json."""
    if not target or target == "main":
        return target
    if re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", target.lower()):
        return target.lower()
    if target in SIDCHAT_ALIASES:
        return SIDCHAT_ALIASES[target]
    sc_file = "/home/super/Projects/NetVM/job-sidechats.json"
    if os.path.exists(sc_file):
        try:
            with open(sc_file, "r", encoding="utf-8") as f:
                sc_data = json.load(f)
            val = sc_data.get(target)
            if isinstance(val, dict):
                return val.get("thread_uuid") or val.get("uuid") or target
            elif isinstance(val, str):
                return val
        except Exception:
            pass
    return target


def log_event(event):
    """Append to JSONL log."""
    event["ts"] = datetime.now(timezone.utc).isoformat()
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(event) + "\n")

def run(cmd, timeout=60, priority=None):
    # priority: CDP queue priority for the subprocess (high/normal/low).
    # DM sends are user-facing -> high. Reads stay at default (normal).
    env = dict(os.environ, CDP_PRIORITY=priority) if priority else None
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True,
                            timeout=timeout, env=env)
    return result.stdout.strip()

def run_full(cmd, timeout=60):
    """Variant returning (rc, stdout, stderr) for calls where a silent
    failure is worse than noise (observed 2026-10-03: opm's browser was dead
    and `dm.py read` printed nothing with exit 0, hiding the outage)."""
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
    return result.returncode, result.stdout.strip(), result.stderr.strip()

def dm_send(agent, target, message, verify=True, raw=False,
            to_agent=None, tags=None, nudge_meta=None):
    """Send a DM. raw=True sends verbatim (for pre-signed messages from
    dm-sign.sh, which already carry [from:X] [id:Y]): no tagging, no
    truncation. Non-raw messages are tagged [from:<agent>] [id:<uuid8>]."""
    # Cross-operator: to_agent is the recipient (whose browser/chat to use).
    # agent is the sender (for attribution). If to_agent is None, send to own chat.
    recipient = to_agent if to_agent else agent

    if agent not in VALID_SENDERS:
        print(f"ERROR: Unknown agent {agent}", file=sys.stderr)
        sys.exit(1)
    if recipient not in VALID_RECIPIENTS:
        print(f"ERROR: Unknown recipient {recipient}", file=sys.stderr)
        sys.exit(1)

    if raw:
        tagged = message
        m = re.search(r'\[id:([^\]]+)\]', message)
        msg_id = m.group(1) if m else "raw"
    else:
        msg_id = str(uuid.uuid4())[:8]
        # Single unified attribution format (matches verify-sig's regex).
        tagged = f"[from:{agent}] [id:{msg_id}] {message}"

    # Canonical follow-up tags: trailing [bracket] tokens are metadata,
    # stripped from the delivered text, recorded on the log events.
    tags = tags or {}
    if not raw:
        message, text_tags = extract_trailing_tags(message)
        if text_tags:
            # Re-tag with the stripped body so the wire text is clean.
            tagged = f"[from:{agent}] [id:{msg_id}] {message}"
        for k, v in text_tags.items():
            tags.setdefault(k, v)
    _log = {"type": "send_start", "id": msg_id, "agent": agent, "to": recipient, "target": target, "msg": message[:100], "tags": tags}
    if nudge_meta:
        _log["nudge_meta"] = nudge_meta
    log_event(_log)

    # Navigate the RECIPIENT's browser to the target chat (the send and the
    # read-back both happen there; navigating the sender's browser was a bug
    # for cross-operator side-chat targets).
    # Use run_full: a silent navigation failure used to send the message to
    # whatever chat the browser happened to be parked on (2026-10-04).
    thread_uuid = None
    thread_url = None
    is_new_sidechat = False
    # Resolve well-known aliases or dynamic thread mappings to UUIDs.
    nav_target = resolve_sidechat_target(target)
    if nav_target != target:
        log_event({"type": "alias_resolved", "id": msg_id, "target": target, "thread_uuid": nav_target})
    if target == "main":
        _rc, _out, _err = run_full(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat main")
    else:
        # Reset-at-Begin: If searching sidebar by title (not a direct UUID), reset to main first to expose the sidebar
        is_uuid = bool(re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", nav_target.strip().lower()))
        if not is_uuid:
            run_full(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat main")
            time.sleep(1)
        _rc, _out, _err = run_full(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat use {nav_target}")
    if _rc != 0:
        log_event({"type": "nav_failed", "id": msg_id, "agent": agent, "to": recipient,
                   "target": target, "rc": _rc, "err": _err[:200]})
        print(f"DM {msg_id} from {agent} to {recipient}/{target}: FAILED (navigation rc={_rc}: {_err[:120]})",
              file=sys.stderr)
        sys.exit(1)
    # cmd_sidechat_use exits 0 even on NOTFOUND (it prints "Navigated to: NOTFOUND").
    # If sidechat is not found, auto-provision a new thread via `sidechat create`!
    if target != "main":
        if "NOTFOUND" in _out:
            log_event({"type": "sidechat_autoprovision_start", "id": msg_id, "agent": agent,
                       "to": recipient, "target": target})
            _c_rc, _c_out, _c_err = run_full(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat create")
            if _c_rc != 0 or "Created:" not in _c_out:
                log_event({"type": "nav_failed", "id": msg_id, "agent": agent, "to": recipient,
                           "target": target, "reason": "autoprovision_create_failed",
                           "rc": _c_rc, "err": _c_err[:200], "out": _c_out[:200]})
                print(f"DM {msg_id} from {agent} to {recipient}/{target}: FAILED (sidechat auto-creation failed: {_c_err[:120]})",
                      file=sys.stderr)
                sys.exit(1)
            is_new_sidechat = True
            log_event({"type": "nav_ok", "id": msg_id, "agent": agent, "to": recipient,
                       "target": target, "status": "sidechat_created_pending_uuid"})
        else:
            m = re.search(r"/thread/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})", _out, re.I)
            if m:
                thread_uuid = m.group(1).lower()
                thread_url = "https://muse.ai/thread/" + thread_uuid
            else:
                # Navigation "succeeded" but we can't confirm where we landed.
                # Don't trust it: fail instead of sending to an unknown chat.
                log_event({"type": "nav_failed", "id": msg_id, "agent": agent, "to": recipient,
                           "target": target, "reason": "no_thread_url", "out": _out[:200]})
                print(f"DM {msg_id} from {agent} to {recipient}/{target}: FAILED (could not confirm thread URL)",
                      file=sys.stderr)
                sys.exit(1)
            # If we resolved via alias, assert the browser actually landed on
            # the expected thread. A mismatch means the navigation didn't take
            # (stale URL, blocked nav) -- fail instead of misdelivering.
            if nav_target != target and thread_uuid != nav_target.lower():
                log_event({"type": "nav_failed", "id": msg_id, "agent": agent, "to": recipient,
                           "target": target, "reason": "uuid_mismatch",
                           "expected": nav_target.lower(), "got": thread_uuid})
                print(f"DM {msg_id} from {agent} to {recipient}/{target}: FAILED (navigation landed on wrong thread)",
                      file=sys.stderr)
                sys.exit(1)
            log_event({"type": "nav_ok", "id": msg_id, "agent": agent, "to": recipient,
                       "target": target, "thread_uuid": thread_uuid})

    time.sleep(2)

    # Send (raw mode: no truncation — signatures must survive intact)
    safe = tagged.replace('"', '\\"').replace('$', '\\$').replace('`', '\\`')
    if not raw:
        safe = safe[:1000]

    # Send with verification retries
    # The underlying muse-chat-api.py send returns None/unreliable status,
    # so we verify by reading the recipient's chat for our message ID.
    max_retries = 3
    delivered = False
    for attempt in range(max_retries):
        send_out = run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} send "{safe}"',
                       priority="high")
        time.sleep(3)  # Wait for message to propagate

        # The send command reports 'sent' / 'enter-sent' / 'NOINPUT'.
        # 'NOINPUT' (or empty output) means the compose box was never found:
        # retrying verification would be meaningless, so fail fast.
        if send_out.strip() not in ("sent", "enter-sent"):
            log_event({"type": "send_failed", "id": msg_id, "agent": agent, "to": recipient,
                       "target": target, "attempt": attempt + 1, "send_out": send_out[:80]})
            if attempt < max_retries - 1:
                time.sleep(2)
            continue

        # Assert the composer cleared: if our text (with the [id:...] tag) is
        # still sitting in the compose box, the send click never fired and any
        # later "verification" would be a false positive (2026-10-04 bug).
        compose = run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} compose_check')
        if compose and compose != "NOCOMPOSE" and msg_id in compose:
            log_event({"type": "compose_stuck", "id": msg_id, "agent": agent, "to": recipient,
                       "target": target, "attempt": attempt + 1})
            # Clear the stuck draft so a retry starts clean.
            run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} send ""',
                priority="high")
            if attempt < max_retries - 1:
                time.sleep(2)
            continue

        # Verify by reading recipient's chat (independent check, not local echo).
        # cmd_messages now excludes the composer subtree, so a hit here means
        # the message is actually in the chat history.
        try:
            check_msgs = run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} messages 5 200')
            if msg_id in check_msgs:
                delivered = True
                log_event({"type": "verified", "id": msg_id, "agent": agent, "to": recipient, "target": target,
                           "thread_uuid": thread_uuid, "attempt": attempt + 1})
                break
            else:
                log_event({"type": "retry", "id": msg_id, "agent": agent, "to": recipient, "attempt": attempt + 1})
        except Exception as e:
            log_event({"type": "verify_error", "id": msg_id, "error": str(e)[:100]})

        if attempt < max_retries - 1:
            time.sleep(2)  # Brief pause before retry

    if delivered and is_new_sidechat:
        # We sent to /thread/new; muse.ai now assigns a permanent UUID.
        # Capture the current URL BEFORE navigating away to main.
        for _ in range(10):
            _u_rc, _u_out, _u_err = run_full(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} url")
            m_uuid = re.search(r"/thread/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})", _u_out, re.I)
            if m_uuid:
                thread_uuid = m_uuid.group(1).lower()
                thread_url = "https://muse.ai/thread/" + thread_uuid
                break
            time.sleep(1)

        if thread_uuid:
            sc_file = "/home/super/Projects/NetVM/job-sidechats.json"
            try:
                sc_data = {}
                if os.path.exists(sc_file):
                    with open(sc_file, "r", encoding="utf-8") as f:
                        sc_data = json.load(f)
                sc_data[target] = {
                    "thread_uuid": thread_uuid,
                    "agent": recipient,
                    "created_at": datetime.now(timezone.utc).isoformat()
                }
                tmp_sc = f"{sc_file}.tmp.{os.getpid()}"
                with open(tmp_sc, "w", encoding="utf-8") as f:
                    json.dump(sc_data, f, indent=2)
                os.replace(tmp_sc, sc_file)
                log_event({"type": "sidechat_autoprovisioned", "id": msg_id, "target": target,
                           "thread_uuid": thread_uuid, "agent": recipient})
            except Exception as e:
                log_event({"type": "sidechat_persist_error", "id": msg_id, "error": str(e)[:100]})
            tags["thread"] = thread_uuid
        else:
            log_event({"type": "sidechat_uuid_capture_failed", "id": msg_id, "out": _u_out[:200]})

    if delivered and thread_uuid and "thread" not in tags:
        tags["thread"] = thread_uuid

    # Note: Leave recipient browser parked in the sidechat thread to preserve Main Chat DOM
    log_event({"type": "send_done", "id": msg_id, "agent": agent, "to": recipient, "target": target})

    if delivered:
        _sent = {"type": "sent", "id": msg_id, "agent": agent, "to": recipient, "target": target, "verified": True, "tags": tags}
        if nudge_meta:
            _sent["nudge_meta"] = nudge_meta
        log_event(_sent)
        print(f"DM {msg_id} from {agent} to {recipient}/{target}: SENT and VERIFIED")
        # Follow-up record creation: if tags request tracking, POST to the
        # VM's /api/box/followups endpoint. Fail gracefully -- the DM already
        # sent, so a record-creation failure is logged, not fatal.
        if tags.get("reply:expected"):
            _register_followup(msg_id, agent, recipient, target, tags)
    else:
        log_event({"type": "failed", "id": msg_id, "agent": agent, "to": recipient, "target": target, "verified": False})
        print(f"DM {msg_id} from {agent} to {recipient}/{target}: FAILED (not found in recipient chat after {max_retries} attempts)", file=sys.stderr)
        sys.exit(1)
    return msg_id

def dm_read(agent, target, n=5, quiet=False, width=200):
    """Read DMs via headless. width widens the per-paragraph slice
    (needed for multi-line signature blocks)."""
    if agent not in VALID_AGENTS:
        print(f"ERROR: Unknown agent {agent}", file=sys.stderr)
        sys.exit(1)

    nav_target = resolve_sidechat_target(target)
    if target == "main":
        run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat main")
    else:
        is_uuid = bool(re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", nav_target.strip().lower()))
        if not is_uuid:
            run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat main")
            time.sleep(1)
        run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat use {nav_target}")

    time.sleep(2)
    rc, msgs, err = run_full(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} messages {n} {width}")

    if rc != 0 or not msgs:
        # Never fail silently: an empty read with exit 0 hid a dead browser
        # for hours (opm, 2026-10-03). Surface the last error line.
        detail = err.splitlines()[-1] if err.strip() else "no output"
        print(f"WARNING: read of {agent}/{target} failed (rc={rc}): {detail}", file=sys.stderr)
    if not quiet:
        print(msgs)
    return msgs

SIGNERS_DIR = "/home/super/Projects/NetVM/dm-signers"

def dm_select(agent, target=None):
    """Select active conversation for an agent.
    - With --target: switch directly to that conversation.
    - Without --target: list available conversations.
    Uses sidechat use/main under the hood; stores selection for future commands.
    """
    import os, json, subprocess

    NETVM_EXEC = os.path.expanduser("~/Projects/NetVM/bin/netvm-exec.sh")
    API = os.path.expanduser("~/Projects/NetVM/bin/muse-chat-api.py")
    state_file = os.path.expanduser(f"~/.dm-select-{agent}.json")

    def run(cmd):
        r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
        return r.stdout.strip()

    if target:
        nav_target = resolve_sidechat_target(target)
        # Switch to target conversation
        if target == "main":
            run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat main")
        else:
            run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat use {nav_target}")
        # Store selection
        with open(state_file, "w") as f:
            json.dump({"agent": agent, "target": target}, f)
        print(f"Selected: {agent}/{target}")
        return target

    # List available conversations
    out = run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat list")
    print(f"Conversations for {agent}:")
    print("  main")
    # Parse: filter out headers, timestamps, and status lines
    import re
    lines = [l.strip() for l in out.split("\n") if l.strip()]
    ts = re.compile(r"^\d+[mhd]$")
    skip = {"NOSIDEBAR", "Side chats", "Unread updates"}
    for line in lines:
        if line in skip:
            continue
        if ts.match(line):
            continue
        # Remaining lines are chat names
        print(f"  {line}")
    print()
    print(f"Use: dm.py select --agent {agent} --target <name>")
    # Show current selection
    if os.path.exists(state_file):
        with open(state_file) as f:
            sel = json.load(f)
        print(f"Current: {sel.get('target', 'main')}")
    return None


def dm_verify_sig(message=None, agent=None, target=None):
    """Verify an SSH-signed DM. Pass message text directly, or give
    --agent/--target to scan recent reads for signed messages."""
    import tempfile, re
    texts = []
    if message:
        texts = [message]
    elif agent and target:
        msgs = dm_read(agent, target, n=10, quiet=True, width=2000)
        # split read output into candidate blocks containing a signature
        chunks = re.split(r'\n---\n', msgs)
        texts = [c for c in chunks if '-----BEGIN SSH SIGNATURE-----' in c]
        if not texts:
            print("no signed messages found in recent reads", file=sys.stderr)
            sys.exit(1)
    else:
        print("ERROR: provide a message or --agent/--target", file=sys.stderr)
        sys.exit(1)

    ok_any = False
    for text in texts:
        text = text.strip()
        m = re.match(r'\[from:([^\]]+)\]\s*\[id:([^\]]+)\]', text)
        if not m:
            print("BAD: no [from:]/[id:] header", file=sys.stderr)
            continue
        sender, mid = m.group(1), m.group(2)
        sm = re.search(r'\n-----BEGIN SSH SIGNATURE-----\n(.*?)\n-----END SSH SIGNATURE-----',
                       text, re.S)
        if not sm:
            print(f"BAD: [{mid}] no signature block", file=sys.stderr)
            continue
        payload = text[:sm.start()].strip()
        sigblock = "-----BEGIN SSH SIGNATURE-----\n" + sm.group(1).strip() + "\n-----END SSH SIGNATURE-----\n"
        pubpath = os.path.join(SIGNERS_DIR, sender + ".pub")
        if not os.path.exists(pubpath):
            print(f"BAD: [{mid}] no public key registered for sender '{sender}'", file=sys.stderr)
            continue
        with open(pubpath) as f:
            pubkey = f.read().strip()
        with tempfile.TemporaryDirectory() as td:
            allowed = os.path.join(td, "allowed")
            with open(allowed, "w") as f:
                f.write(f"{sender} {pubkey}\n")
            sigf = os.path.join(td, "sig")
            with open(sigf, "w") as f:
                f.write(sigblock)
            payf = os.path.join(td, "payload")
            with open(payf, "w") as f:
                f.write(payload)
            # verify reads the payload from stdin; feed it from the temp file
            # (redirect, not a pipe) per the file-based lesson from chat-400
            with open(payf, "rb") as fin:
                r = subprocess.run(
                    ["ssh-keygen", "-Y", "verify", "-f", allowed, "-I", sender,
                     "-n", "dm", "-s", sigf],
                    stdin=fin, capture_output=True, text=True, timeout=15)
        if r.returncode == 0:
            print(f"GOOD: [{mid}] signature valid — really from '{sender}'")
            ok_any = True
        else:
            print(f"BAD: [{mid}] signature FAILED for claimed sender '{sender}': "
                  f"{r.stderr.strip()[:120]}", file=sys.stderr)
    sys.exit(0 if ok_any else 1)

def dm_log(n=20):
    """Show recent log entries."""
    if not os.path.exists(LOG_FILE):
        print("No log file yet")
        return
    with open(LOG_FILE) as f:
        lines = f.readlines()
    for line in lines[-n:]:
        e = json.loads(line)
        print(f"{e['ts'][:19]} {e.get('type','?'):12} {e.get('id','-'):8} {e.get('agent','-')}/{e.get('target','-')}")

def dm_thread(from_agent, to_agent, target, message):
    """Thread from one agent to another. Attribution is applied exactly once,
    in the unified [from:X] [id:Y] format, by dm_send."""
    return dm_send(from_agent, target, message, to_agent=to_agent)

def main():
    p = argparse.ArgumentParser(description="DM: Headless Direct Messages (tagged [from:X] [id:Y], logged; delivery confirmed by recipient read-back)")
    sub = p.add_subparsers(dest='cmd', required=True)

    ps = sub.add_parser('send', help='Send a DM (tagged; delivery confirmed by recipient read-back, up to 3 attempts)')
    ps.add_argument('--agent', required=True, choices=VALID_SENDERS)
    ps.add_argument('--to', required=False, choices=VALID_RECIPIENTS, default=None,
                    help='Recipient operator (for cross-operator DMs). Uses recipient\'s browser/chat.')
    ps.add_argument('--target', required=True)
    ps.add_argument('--no-verify', action='store_true',
                    help='Accepted for compatibility but ignored: recipient-side read-back verification always runs.')
    ps.add_argument('--raw', action='store_true', help='Send verbatim: no tagging, no truncation (for pre-signed messages from dm-sign.sh)')
    ps.add_argument('--expect-reply', action='store_true',
                    help='Tag reply:expected: create a follow-up record (nudge/escalate per policy)')
    ps.add_argument('--reply-timeout', default=None,
                    help='reply:timeout= seconds until nudge/escalation (60..604800, default 3600)')
    ps.add_argument('--reply-nudges', default=None,
                    help='reply:nudges= max nudges before escalation (0..10, default 2)')
    ps.add_argument('--reply-escalate', default=None,
                    help='reply:escalate= identity to alert on timeout')
    ps.add_argument('--route', default=None, help='route: attach follow-up to a conversation route')
    ps.add_argument('--thread', default=None, help='thread: associate DM with a thread UUID')
    ps.add_argument('--tag', action='append', default=[],
                    help='Raw canonical tag (repeatable): reply:expected, reply:timeout=N, reply:nudges=N, reply:escalate=X, route:R, thread:U')
    ps.add_argument('--nudge-meta', default=None,
                    help='JSON string of nudge metadata (followup_id, nudge_n, dm_id) for audit logging')
    ps.add_argument('message')
    def _send_with_tags(a):
        try:
            cli_tags = parse_tags(a.tag)
            flag_tags = tags_from_flags(a)
            tags = merge_tags(flag_tags, cli_tags, {})
        except ValueError as e:
            print(f"TAG_ERROR: {e}", file=sys.stderr)
            sys.exit(2)
        nudge_meta = None
        if a.nudge_meta:
            try:
                import json as _json
                nudge_meta = _json.loads(a.nudge_meta)
                if not isinstance(nudge_meta, dict):
                    raise ValueError("nudge-meta must be a JSON object")
            except Exception as e:
                print(f"NUDGE_META_ERROR: {e}", file=sys.stderr)
                sys.exit(2)
        return dm_send(a.agent, a.target, a.message, verify=not a.no_verify,
                       raw=a.raw, to_agent=a.to, tags=tags,
                       nudge_meta=nudge_meta)
    ps.set_defaults(func=_send_with_tags)

    psel = sub.add_parser('select', help='Select active conversation')
    psel.add_argument('--agent', required=True, choices=VALID_AGENTS)
    psel.add_argument('--target', required=False, default=None,
                      help='Conversation: main or side chat name')
    psel.set_defaults(func=lambda a: dm_select(a.agent, a.target))

    pr = sub.add_parser('read', help='Read DMs')
    pr.add_argument('--agent', required=True, choices=VALID_AGENTS)
    pr.add_argument('--target', required=True)
    pr.add_argument('--n', type=int, default=5)
    pr.set_defaults(func=lambda a: dm_read(a.agent, a.target, a.n))

    pvs = sub.add_parser('verify-sig', help='Verify SSH signature on a signed DM (pass message text, or --agent/--target to scan recent reads)')
    pvs.add_argument('--agent', required=False, choices=VALID_AGENTS)
    pvs.add_argument('--target', required=False)
    pvs.add_argument('message', nargs='?')
    pvs.set_defaults(func=lambda a: dm_verify_sig(message=a.message, agent=a.agent, target=a.target))

    pl = sub.add_parser('log', help='Show DM log')
    pl.add_argument('--n', type=int, default=20)
    pl.set_defaults(func=lambda a: dm_log(a.n))

    pt = sub.add_parser('thread', help='Thread from one agent to another')
    pt.add_argument('--from', dest='from_agent', required=True, choices=VALID_SENDERS)
    pt.add_argument('--to', dest='to_agent', required=True, choices=VALID_RECIPIENTS)
    pt.add_argument('--target', required=True)
    pt.add_argument('message')
    pt.set_defaults(func=lambda a: dm_thread(a.from_agent, a.to_agent, a.target, a.message))

    args = p.parse_args()
    args.func(args)

BOX_API_BASE = os.environ.get("BOX_API_BASE", "https://box.muse-dev.online")
BOX_SIGN_KEY = os.environ.get("BOX_SIGN_KEY", os.path.expanduser("~/.ssh/id_ed25519"))


def _box_sign(identity, endpoint):
    """Sign a box API request. Returns (ts, sig_armored) or (None, None)."""
    ts = str(int(time.time()))
    payload = f"{ts}\n{endpoint}".encode()
    try:
        p = subprocess.run(
            ["ssh-keygen", "-Y", "sign", "-f", BOX_SIGN_KEY, "-n", "box"],
            input=payload, capture_output=True, timeout=15)
        if p.returncode != 0:
            return None, None
        return ts, p.stdout.decode()
    except Exception:
        return None, None


def _register_followup(msg_id, agent, recipient, target, tags):
    """POST a follow-up record to the VM. Never raises -- logs and returns."""
    identity_map = {
        "bl": "bl",
        "opm": "operator-main",
        "646": "operator-646",
    }
    identity = "bl"

    ts, sig = _box_sign(identity, "followups")
    if not ts or not sig:
        log_event({"type": "followup_register_failed", "id": msg_id,
                   "error": "signing failed"})
        return

    tags_obj = {}
    if tags.get("reply:expected"):
        tags_obj["reply:expected"] = True
    for key in ("reply:timeout", "reply:nudges", "reply:escalate",
                "route", "thread"):
        if key in tags:
            tags_obj[key] = tags[key]

    body = {
        "dm_id": msg_id,
        "to": recipient,
        "from": agent,
        "target": target,
        "tags": tags_obj,
        "dm_sent_ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }

    query = urllib.parse.urlencode({
        "identity": identity,
        "ts": ts,
        "sig": sig,
    })
    url = f"{BOX_API_BASE}/api/box/followups?{query}"
    try:
        req = urllib.request.Request(
            url, data=json.dumps(body).encode(),
            headers={"Content-Type": "application/json",
                     # Cloudflare 1010-blocks python-urllib's default UA;
                     # use an identifiable custom UA instead.
                     "User-Agent": "NetVM-dm.py/1.0 (bl)"},
            method="POST")
        with urllib.request.urlopen(req, timeout=30) as r:
            resp = json.load(r)
            log_event({"type": "followup_registered", "id": msg_id,
                       "request_id": resp.get("request_id"),
                       "tracked": resp.get("tracked")})
    except Exception as e:
        log_event({"type": "followup_register_failed", "id": msg_id,
                   "error": str(e)[:200]})

    # Also persist to local followups.json for bl autonomy
    try:
        f_path = "/home/super/Projects/NetVM/followups.json"
        followups = {}
        if os.path.exists(f_path):
            with open(f_path, "r", encoding="utf-8") as f:
                followups = json.load(f)
        now_dt = datetime.now(timezone.utc)
        timeout_s = int(tags.get("reply:timeout", 3600))
        nudges_n = int(tags.get("reply:nudges", 2))
        deadline_dt = now_dt + timedelta(seconds=timeout_s)
        followups[msg_id] = {
            "dm_id": msg_id,
            "sender": agent,
            "recipient": recipient,
            "target": target,
            "thread_uuid": tags.get("thread"),
            "route": tags.get("route"),
            "sent_at": now_dt.isoformat(),
            "deadline": deadline_dt.isoformat(),
            "timeout_s": timeout_s,
            "nudges_allowed": nudges_n,
            "nudges_sent": 0,
            "escalate_to": tags.get("reply:escalate", "opm"),
            "status": "pending"
        }
        tmp = f"{f_path}.tmp.{os.getpid()}"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(followups, f, indent=2)
        os.replace(tmp, f_path)
    except Exception as e:
        log_event({"type": "local_followup_save_failed", "id": msg_id, "error": str(e)[:200]})

if __name__ == '__main__':
    main()