#!/usr/bin/env python3 """ DM: Headless Direct Message API (muse.ai) — tagged, logged, verifiable. Wire format (every send, signed or not): [from:] [id:<8-hex>] 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/.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 from datetime import datetime, timezone 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"] LOG_FILE = "/home/super/Projects/NetVM/dm-log.jsonl" 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): """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:] [id:].""" # 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_AGENTS: print(f"ERROR: Unknown agent {agent}", file=sys.stderr) sys.exit(1) if recipient not in VALID_AGENTS: 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}" log_event({"type": "send_start", "id": msg_id, "agent": agent, "to": recipient, "target": target, "msg": message[:100]}) # 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). if target == "main": run(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat main") else: run(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat use {target}") 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): run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} send "{safe}"', priority="high") time.sleep(3) # Wait for message to propagate # Verify by reading recipient's chat (independent check, not local echo) 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, "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 # Park the recipient's browser back on main run(f"{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} sidechat main") log_event({"type": "send_done", "id": msg_id, "agent": agent, "to": recipient, "target": target}) if delivered: log_event({"type": "sent", "id": msg_id, "agent": agent, "to": recipient, "target": target, "verified": True}) print(f"DM {msg_id} from {agent} to {recipient}/{target}: SENT and VERIFIED") 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) 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 {target}") time.sleep(2) rc, msgs, err = run_full(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} messages {n} {width}") run(f"{NETVM_EXEC} {agent} -- python3 {API} --account {agent} sidechat main") 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: # 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 {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 ") # 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_AGENTS) ps.add_argument('--to', required=False, choices=VALID_AGENTS, 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('message') ps.set_defaults(func=lambda a: dm_send(a.agent, a.target, a.message, verify=not a.no_verify, raw=a.raw, to_agent=a.to)) 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_AGENTS) pt.add_argument('--to', dest='to_agent', required=True, choices=VALID_AGENTS) 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) if __name__ == '__main__': main()