box/bin/muse-chat-api.py

#!/usr/bin/env python3
# GOLDEN PATH: container -> VM (34.139.37.135) -> bl (100.123.153.75) -> netns -> browser -> agent
# This API is the bridge. Every call traverses 4 hops. Respect the path.

"""
Multi-account muse.ai chat API with approval handling.

Approvals: The browser may show permission dialogs (e.g., "Allow pip to share
information with 34.139.37.135?"). The API detects these and handles them:
- Known-safe (our infrastructure IPs): auto-approve
- Unknown: raise APPROVAL_NEEDED, operator decides via chat

Usage:
  muse-chat-api.py --account <agent> send "message"
  muse-chat-api.py --account <agent> messages [n]
  muse-chat-api.py --account <agent> wait [timeout]
  muse-chat-api.py --account <agent> approvals  # check pending approvals
  muse-chat-api.py --account <agent> upload <file> [--message txt] [--dry-run]

Exit codes: 0 ok, 1 error, 2 APPROVAL_NEEDED (human decision),
3 DOM_NOT_READY (browser not in expected chat state - safe to retry).
"""
import json, urllib.request, websocket, time, sys, argparse, importlib.util, re
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
try:
    from sidechat_manager import ensure_sidebar, log_sidechat_op
    HAS_SIDECHAT_MANAGER = True
except ImportError:
    HAS_SIDECHAT_MANAGER = False
try:
    from cdp_queue import cdp_slot, PRIORITY_HIGH, PRIORITY_NORMAL, PRIORITY_LOW
    HAS_CDP_QUEUE = True
except ImportError:
    HAS_CDP_QUEUE = False


def _load_accounts():
    """Build the accounts table from the NODES.md registry — new nodes
    propagate automatically instead of being hardcoded here."""
    path = "/home/super/Projects/NetVM/bin/netvm-registry.py"
    spec = importlib.util.spec_from_file_location("netvm_registry", path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    accounts = {}
    for node, rec in mod.load().items():
        peer_ip = rec.get("peer_ip", "127.0.0.1")
        accounts[node] = (node, "http://%s:%d/json/list" % (peer_ip, rec["cdp_port"]))
    return accounts


ACCOUNTS = _load_accounts()

# IPs we trust for auto-approval (our infrastructure)
TRUSTED_IPS = {
    "34.139.37.135",  # VM (gateway)
    "100.123.153.75", # bl (main compute)
    "100.81.31.9",    # VM tailnet
}

def get_page(node, cdp_url):
    urls = [cdp_url]
    m = re.search(r":(\d+)/", cdp_url)
    if m:
        port = m.group(1)
        if "127.0.0.1" in cdp_url:
            path = "/home/super/Projects/NetVM/bin/netvm-registry.py"
            spec = importlib.util.spec_from_file_location("netvm_registry", path)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            pip = mod.peer_ip_for(node)
            if pip:
                urls.append(f"http://{pip}:{port}/json/list")
        else:
            urls.append(f"http://127.0.0.1:{port}/json/list")
    ts = None
    last_err = None
    for u in urls:
        try:
            with urllib.request.urlopen(u, timeout=5) as r:
                ts = json.load(r)
                break
        except Exception as e:
            last_err = e
            continue
    if not ts:
        print(f"ERROR: No page found ({last_err})", file=sys.stderr)
        sys.exit(1)
    pages = [t for t in ts if t.get('type') == 'page']
    if not pages:
        print("ERROR: No page found", file=sys.stderr)
        sys.exit(1)
    return pages[0]

def ev(ws, expr, await_p=False):
    """Returns None (no traceback) if the CDP WebSocket drops.
    (Fix 2026-10-06: uncaught WebSocketConnectionClosedException.)"""
    try:
        ws.send(json.dumps({
            "id": 1, "method": "Runtime.evaluate",
            "params": {"expression": expr, "returnByValue": True, "awaitPromise": await_p}
        }))
        # Drain CDP events until we get our command response (id 1).
        # The browser can emit events (Runtime.executionContextCreated, etc.)
        # at any time; taking the first recv() blindly returns None on a
        # busy page (observed as transient navigation failures in dm.py
        # sidechat sends, 2026-10-04 — same class as the NO_SWITCHER fix
        # in box-chat-cdp.py commit 8d4bfa7).
        for _ in range(50):
            resp = json.loads(ws.recv())
            if resp.get("id") == 1:
                break
        else:
            return None
        return resp.get('result', {}).get('result', {}).get('value')
    except Exception as e:
        print(f"CDP evaluate failed: {type(e).__name__}: {e}", file=sys.stderr)
        return None

def check_approvals(ws):
    """
    Check for browser permission dialogs.
    Returns list of (dialog_text, is_trusted, action_taken).
    """
    result = ev(ws, """(() => {
      const dialogs = [];
      // 1. Check confirmed structural selectors
      const headers = [...document.querySelectorAll('[data-testid="approval-panel-header"], [data-testid*="approval"]')];
      for (const h of headers) {
        let card = h;
        for (let i = 0; i < 6 && card && card.parentElement && card.parentElement !== document.body; i++) {
          if (card.querySelector('button[data-hatch-approval-primary-action="true"]') ||
              [...card.querySelectorAll('button')].some(b => {
                const t = (b.innerText||'').toLowerCase();
                return t.includes('allow once') || t.includes('deny');
              })) {
            dialogs.push(card.innerText.slice(0, 500));
            break;
          }
          card = card.parentElement;
        }
      }
      // 2. Check for "Allow ... to share" pattern if not found
      if (dialogs.length === 0) {
        const body = document.body ? document.body.innerText : '';
        if (body.includes('Allow') && body.includes('to share')) {
          const els = [...document.querySelectorAll('*')].filter(el => {
            const t = el.innerText || '';
            return t.includes('Allow') && t.includes('to share') && t.length < 500;
          });
          for (const el of els.slice(0,3)) {
            dialogs.push(el.innerText.slice(0,200));
          }
        }
      }
      // 3. Check for other permission patterns
      if (dialogs.length === 0) {
        const perm_btns = [...document.querySelectorAll('button')].filter(b => {
          const t = (b.innerText||'').toLowerCase();
          return t.includes('allow') || t.includes('deny') || t.includes('block');
        });
        if (perm_btns.length >= 2) {
          const parent = perm_btns[0].closest('div');
          if (parent) dialogs.push(parent.innerText.slice(0,200));
        }
      }
      return JSON.stringify(dialogs);
    })()""")
    try:
        dialogs = json.loads(result) if result else []
    except:
        dialogs = []
    
    actions = []
    for d in dialogs:
        # Extract IP if present
        import re
        ips = re.findall(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', d)
        # Check trust: if IP present, must be in TRUSTED_IPS; if no IP, untrusted approval dialog
        if ips:
            is_trusted = any(ip in TRUSTED_IPS for ip in ips)
        else:
            # Check if this is a known false positive or real dialog
            if "allow once" in d.lower() or "approval-panel" in d.lower() or "wants to" in d.lower():
                is_trusted = False
            else:
                continue

        if is_trusted:
            # Auto-approve: click primary action or "Allow once" / "Allow"
            clicked = ev(ws, """(async()=>{
              const b = document.querySelector('button[data-hatch-approval-primary-action="true"]') ||
                        [...document.querySelectorAll('button')].find(x=>{
                          const t = (x.innerText||'').toLowerCase().trim();
                          return t === 'allow once' || t.includes('allow once') || t === 'allow';
                        });
              if (b) { b.click(); return 'clicked:'+b.innerText.slice(0,20); }
              return 'NOTFOUND';
            })()""", True)
            actions.append((d[:80], True, clicked))
        else:
            actions.append((d[:80], False, "APPROVAL_NEEDED"))

    return actions

# Exit code for "browser DOM not ready" — distinct from 1 (generic error)
# and 2 (APPROVAL_NEEDED, needs a human). DOM_NOT_READY is safe to retry
# after a few seconds: the React app may still be hydrating (S3 partial
# load) or the Warp tunnel may have just hiccupped. dm.py treats any
# non-"sent" send output as a failed attempt and retries, so this is a
# fail-closed retry signal, not a crash.
DOM_NOT_READY_EXIT = 3

# Shimmer spans (span.animate-pulse-light) exist in the settled state too
# (2 observed — avatar/media skeleton slots, not load progress). Only a
# mass-skeleton count indicates a stuck/broken render. Heuristic threshold
# from docs/DOM-EDGE-STATES.md 1.1; the settled-state signature below is
# the primary readiness signal.
SHIMMER_STORM_THRESHOLD = 30


def chat_ready_probe(ws):
    """One-shot DOM readiness probe. Returns a dict of observed state, or
    {} if the page could not be evaluated at all."""
    result = ev1(ws, """(() => {
      const q = s => { try { return document.querySelector(s); } catch(e) { return null; } };
      const qa = s => { try { return document.querySelectorAll(s); } catch(e) { return []; } };
      const title = document.title || '';
      const url = window.location.href || '';
      const switcher = !!q('[data-testid="hatch-chat-switcher-trigger"]');
      const buttons = qa('button').length;
      const msglog = !!q('div[role="log"]');
      const composer = !!(q('[contenteditable="true"]') ||
                          q('textarea[placeholder*="Message"]') ||
                          q('div[role="textbox"]'));
      const shimmer = qa('span.animate-pulse-light').length;
      let alertText = '';
      try {
        alertText = [...qa('[role="alert"]')]
          .filter(e => e.tagName !== 'SCRIPT' && (e.innerText||'').trim().length > 0)
          .slice(0, 2).map(e => e.innerText.slice(0, 160)).join(' | ');
      } catch(e) {}
      let onLine = true;
      try { onLine = navigator.onLine; } catch(e) {}
      return JSON.stringify({title, url, switcher, buttons, msglog,
                             composer, shimmer, alertText, onLine});
    })()""")
    try:
        return json.loads(result) if result else {}
    except Exception:
        return {}


def classify_chat_state(p):
    """Classify a readiness probe. Returns (state, diagnosis). READY is the
    only state a send may proceed from. Reference: docs/DOM-EDGE-STATES.md.
    Note: 'Chat \u2014' is the 'Chat \u2014 <identity>' title prefix."""
    if not p:
        return ("NO_PROBE",
                "CDP evaluate returned nothing - page blank, navigating, or CDP wedged")
    title = p.get("title", "")
    url = p.get("url", "")
    if title == "muse.ai":
        return ("LANDING",
                "browser is on the muse.ai landing page, not in chat (state: LANDING)")
    if not p.get("onLine", True):
        return ("OFFLINE",
                "navigator.onLine is false - browser reports no network")
    if p.get("alertText"):
        return ("REAL_ERROR",
                "page shows an error banner: %s" % p["alertText"][:160])
    if "/thread/new" in url:
        # Legitimate only right after `sidechat create`: the new-thread
        # composer is up and the title has settled. A stripped /thread/new
        # (no title, no composer) is the S8 broken state.
        if p.get("composer") and "Chat \u2014" in title:
            return ("READY", "new-thread composer state (/thread/new)")
        return ("S8_STRIPPED",
                "stripped /thread/new page - no composer/title (state: S8)")
    switcher = p.get("switcher", False)
    buttons = p.get("buttons", 0)
    if title == "Muse" and (not switcher or buttons < 30):
        return ("S3_PARTIAL",
                "React still hydrating: title 'Muse', %d buttons, switcher=%s (state: S3)"
                % (buttons, switcher))
    if not ("Chat \u2014" in title and switcher and buttons > 30 and p.get("msglog")):
        return ("NOT_SETTLED",
                "settled-state signature not met (title=%r switcher=%s buttons=%d msglog=%s)" %
                (title[:40], switcher, buttons, p.get("msglog", False)))
    if p.get("shimmer", 0) > SHIMMER_STORM_THRESHOLD:
        return ("SHIMMER_STORM",
                "mass skeleton render: %d shimmer spans (settled has ~2)" % p["shimmer"])
    if not p.get("composer"):
        return ("COMPOSER_MISSING",
                "chat settled but no composer input found - send has nowhere to type")
    return ("READY", "settled chat state (title=%r)" % title[:40])


def assert_chat_ready(ws, context="send"):
    """Fail-closed React-readiness gate ("matching chromebox").

    Call before any DOM mutation (send, upload). On mismatch prints a
    structured diagnosis to stderr and exits DOM_NOT_READY_EXIT (3) - a
    retryable signal, distinct from APPROVAL_NEEDED (2, needs a human).
    S3 partial load gets one 5s grace re-probe (transient render phase);
    everything else fails immediately.
    """
    probe = chat_ready_probe(ws)
    state, detail = classify_chat_state(probe)
    if state == "S3_PARTIAL":
        time.sleep(5)
        probe = chat_ready_probe(ws)
        state, detail = classify_chat_state(probe)
    if state != "READY":
        print("DOM_NOT_READY [%s] %s" % (state, detail), file=sys.stderr)
        print("DOM_NOT_READY probe: %s" % json.dumps(probe), file=sys.stderr)
        sys.exit(DOM_NOT_READY_EXIT)
    return probe

def cmd_approvals(ws):
    """Check and handle pending approvals."""
    actions = check_approvals(ws)
    if not actions:
        print("No pending approvals")
        return
    for dialog, trusted, action in actions:
        print(f"Dialog: {dialog}")
        print(f"  Trusted: {trusted}, Action: {action}")
        if not trusted:
            print("  APPROVAL_NEEDED: Manual review required")
            sys.exit(2)

def cmd_send(ws, message):
    # React-readiness gate ("matching chromebox"): fail closed if the
    # browser is not in the expected chat state (landing page,
    # partial load, partition). Exit 3 = safe to retry, unlike
    # APPROVAL_NEEDED (2, needs a human).
    assert_chat_ready(ws, context="send")
    # Check approvals
    actions = check_approvals(ws)
    for dialog, trusted, action in actions:
        if not trusted:
            print(f"APPROVAL_NEEDED: {dialog[:80]}", file=sys.stderr)
            sys.exit(2)
    
    msg_esc = message.replace('\\', '\\\\').replace('`', '\\`').replace('$', '\\$')
    result = ev1(ws, f"""(async()=>{{
      const input = document.querySelector('[contenteditable="true"]') ||
                    document.querySelector('textarea[placeholder*="Message"]') ||
                    [...document.querySelectorAll('div[role="textbox"]')][0];
      if (!input) return 'NOINPUT';
      input.focus();
      document.execCommand('insertText', false, `{msg_esc}`);
      await new Promise(r=>setTimeout(r,500));
      const send = [...document.querySelectorAll('button')].find(b=>
        b.getAttribute('aria-label')&&b.getAttribute('aria-label').toLowerCase().includes('send')
      );
      if (send) {{ send.click(); return 'sent'; }}
      const ke = new KeyboardEvent('keydown', {{key:'Enter', code:'Enter', bubbles:true}});
      input.dispatchEvent(ke);
      return 'enter-sent';
    }})()""", True)
    print(result)

def cmd_messages(ws, n=5, width=200):
    # Check approvals first (non-blocking)
    check_approvals(ws)
    # Exclude the compose box subtree: a failed send leaves the draft text
    # (including the [id:...] tag) in the composer, and scraping it would
    # produce a false "verified" (2026-10-04 dm.py false-confirmation bug).
    # 2026-10-05: row-aware scrape. The message feed alternates sender-header
    # rows (div.group/stacked-row, per-message timestamp in
    # div.text-caption-1) and message units. Each unit is prefixed with its
    # header's timestamp ([8:57 pm]) so sweeps can compute message age. The
    # feed is the row-parent whose non-row children hold <p> elements (the
    # sidebar shares the row classes). The feed hydrates async after
    # navigation, so poll up to ~8s before falling back to the legacy
    # paragraph scrape.
    result = ev1(ws, f"""(async () => {{
      const composer = document.querySelector('[contenteditable="true"]') ||
                       document.querySelector('textarea[placeholder*="Message"]');
      const noComposer = p => !(composer && composer.contains(p));
      const legacy = () => {{
        const ps = [...document.querySelectorAll('p')]
          .filter(noComposer)
          .slice(-{n*2}).map(p=>p.innerText.slice(0,{width}));
        return ps.join('\\n---\\n');
      }};
      const ROWSEL = 'div[class*="group/stacked-row"]';
      const findFeed = () => {{
        const byParent = new Map();
        for (const r of document.querySelectorAll(ROWSEL)) {{
          const p = r.parentElement;
          if (p) {{
            if (!byParent.has(p)) byParent.set(p, []);
            byParent.get(p).push(r);
          }}
        }}
        for (const [p, rs] of byParent) {{
          const hasMsg = [...p.children].some(c => rs.indexOf(c) === -1 &&
            c.querySelectorAll('p').length > 0);
          if (hasMsg) return [p, rs];
        }}
        return [null, null];
      }};
      let list = null, rows = null;
      for (let i = 0; i < 16 && !list; i++) {{
        [list, rows] = findFeed();
        if (!list) await new Promise(r => setTimeout(r, 500));
      }}
      if (!list) return legacy();
      let curTs = '';
      const out = [];
      for (const child of [...list.children]) {{
        if (composer && child.contains(composer)) continue;
        if (rows.indexOf(child) !== -1) {{
          const t = child.querySelector('div.text-caption-1');
          const txt = t ? t.innerText.trim() : '';
          if (txt) curTs = txt;
        }} else {{
          const ps = [...child.querySelectorAll('p')].filter(noComposer)
            .map(p=>p.innerText.slice(0,{width}));
          if (ps.length) out.push((curTs ? '[' + curTs + '] ' : '') + ps.join('\\n'));
        }}
      }}
      const res = out.slice(-{n}).join('\\n---\\n');
      return res || legacy();
    }})()""", True)
    print(result)

def cmd_compose_check(ws):
    """Print the current compose-box text (empty string if clear).
    Used by dm.py to confirm a send actually left the composer."""
    result = ev1(ws, """(() => {
      const input = document.querySelector('[contenteditable="true"]') ||
                    document.querySelector('textarea[placeholder*="Message"]') ||
                    [...document.querySelectorAll('div[role="textbox"]')][0];
      if (!input) return 'NOCOMPOSE';
      return input.innerText || '';
    })()""")
    print(result if result is not None else '')

def cmd_wait(ws, timeout=30):
    print(f"Waiting {timeout}s for response...")
    # Check approvals periodically during wait
    for i in range(timeout // 5):
        actions = check_approvals(ws)
        for dialog, trusted, action in actions:
            if not trusted:
                print(f"APPROVAL_NEEDED: {dialog[:80]}", file=sys.stderr)
                sys.exit(2)
        time.sleep(5)
    cmd_messages(ws, 2)

def cdp_navigate(ws, url, timeout_s=30):
    """Navigate via CDP Page.navigate (proper navigation, waits for commit).
    Returns True if the page URL matches the target after navigation.
    Returns False (no traceback) if the CDP WebSocket drops mid-call --
    the caller retries on False. (Fix 2026-10-06: uncaught
    WebSocketConnectionClosedException crashed dm.py sends as nav_failed.)"""
    import time as _time
    try:
        ws.send(json.dumps({"id": 2, "method": "Page.navigate",
                            "params": {"url": url}}))
        # Drain until we get the Page.navigate response (id 2).
        for _ in range(50):
            resp = json.loads(ws.recv())
            if resp.get("id") == 2:
                break
        else:
            return False
    except Exception as e:
        # Browser CDP connection dropped (crash/restart/relay flake).
        # Fail cleanly so dm.py logs nav_failed without a traceback.
        print(f"CDP navigate failed: {type(e).__name__}: {e}", file=sys.stderr)
        return False
    # Wait for the URL to settle (SPA client-side routing).
    for _ in range(timeout_s):
        try:
            cur = ev1(ws, "window.location.href", True)
        except Exception as e:
            print(f"CDP read failed: {type(e).__name__}: {e}", file=sys.stderr)
            return False
        if cur and url.rstrip("/").lower() in cur.lower():
            return True
        _time.sleep(1)
    return False

def cmd_sidechat_use(ws, chat_id):
    """Open a sidechat by name (sidebar text search) or by thread UUID
    (direct navigation). The sidebar shows titles, not UUIDs, so the
    text search below can never match a UUID."""
    import base64, re
    cid = chat_id.strip()
    if re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", cid.lower()):
        url = "https://muse.ai/thread/" + cid.lower()
        want_uuid = cid.lower()
        # Use CDP Page.navigate (proper navigation lifecycle) and confirm
        # the URL actually changed. Hardened 2026-10-04: the React SPA
        # sometimes redirects back to / when the thread page hasn't finished
        # loading; retry instead of failing immediately.
        # (dm.py sidechat 1/3 flake; was window.location.href via evaluate.)
        cur = None
        for _try in range(5):
            ok = cdp_navigate(ws, url, timeout_s=15)
            cur = ev1(ws, "window.location.href", True)
            if ok and cur and want_uuid in cur:
                break
            # SPA dropped the nav or hasn't routed yet; wait and retry.
            time.sleep(3)
        print(f"Navigated to: {cur}")
        return
    b64 = base64.b64encode(chat_id.encode()).decode()
    js = (
        "(async()=>{"
        # Open sidebar first
        "const ob=[...document.querySelectorAll('button')].find("
        "b=>b.textContent.includes('Open chat and side chats'));"
        "if(ob)ob.click();"
        "await new Promise(r=>setTimeout(r,2000));"
        # Find and click the chat
        "const name=atob('" + b64 + "');"
        "const els=[...document.querySelectorAll('*')];"
        "const cands=els.filter(e=>e.textContent&&e.textContent.includes(name));"
        "cands.sort((a,b)=>a.textContent.length-b.textContent.length);"
        "const el=cands[0];"
        "if(!el)return 'NOTFOUND';"
        "el.click();"
        "await new Promise(r=>setTimeout(r,4000));"
        "return window.location.href;"
        "})()"
    )
    # Hardened 2026-10-04: the click sometimes doesn't navigate (React
    # mid-render, or the SPA drops it). Retry the whole nav if we didn't
    # land on a /thread/ URL.
    result = None
    for _try in range(3):
        result = ev(ws, js, True)
        if result and "/thread/" in result:
            break
        time.sleep(3)
    print(f"Navigated to: {result}")

def cmd_sidechat_main(ws):
    """Navigate back to main chat via Ctrl+J keyboard shortcut."""
    # Ctrl+J (the Chat button shortcut) properly refocuses Main Chat.
    # More reliable than DOM scraping for buttons/text.
    # Do NOT navigate to https://muse.ai/ (landing page) - it restores the
    # last-viewed chat which may be a side chat.
    import json as _json
    # Send Ctrl+J via Input.dispatchKeyEvent
    ws.send(_json.dumps({
        "id": 10, "method": "Input.dispatchKeyEvent",
        "params": {"type": "keyDown", "key": "j", "code": "KeyJ",
                   "ctrlKey": True, "modifiers": 2}
    }))
    _json.loads(ws.recv())
    ws.send(_json.dumps({
        "id": 11, "method": "Input.dispatchKeyEvent",
        "params": {"type": "keyUp", "key": "j", "code": "KeyJ",
                   "ctrlKey": True, "modifiers": 2}
    }))
    _json.loads(ws.recv())
    time.sleep(4)
    # Get current URL to confirm navigation
    ws.send(_json.dumps({"id": 12, "method": "Runtime.evaluate",
        "params": {"expression": "window.location.href"}}))
    resp = _json.loads(ws.recv())
    url = resp.get("result", {}).get("result", {}).get("value", "unknown")
    print(f"Back to main chat: {url}")

def cmd_sidechat_list(ws):
    """List side chats via DOM."""
    ev(ws, """(() => {
      const btn = [...document.querySelectorAll("button")].find(b => 
        b.textContent.includes("Open chat and side chats")
      );
      if (btn) btn.click();
    })()""")
    time.sleep(2)
    chats = ev(ws, """(() => {
      const text = document.body.innerText;
      const idx = text.indexOf("Side chats");
      if (idx === -1) return "NOSIDEBAR";
      return text.slice(idx, idx + 1000);
    })()""")
    print(chats[:500])

def cmd_sidechat_create(ws):
    """Create a new side chat via the '+' button.
    Returns the new thread URL.
    Selectors (verified 2026-10-04 via DOM investigation):
    - Panel opener: [data-testid="hatch-chat-switcher-trigger"] (idempotent)
    - + button: [data-testid="hatch-chat-compose"] (SVG icon, no text)
    """
    import time as _time
    import json as _json
    import sys as _sys2
    # Ensure chat is active via DOM clicks (replaces Ctrl+J).
    # Ctrl+J needs keyboard focus which stripped states (/thread/new) lack.
    # Priority: compose (fast path) -> switcher -> nav-chat (recovery).
    # Verified 2026-10-04: hatch-nav-chat click recovers from /thread/new.
    def _ensure_chat_active(timeout=20):
        t0 = _time.time()
        while _time.time() - t0 < timeout:
            st = ev(ws, """(() => ({
                compose: !!document.querySelector('[data-testid="hatch-chat-compose"]'),
                switcher: !!document.querySelector('[data-testid="hatch-chat-switcher-trigger"]'),
                navchat: !!document.querySelector('[data-testid="hatch-nav-chat"]')
            }))()""")
            if not isinstance(st, dict):
                _time.sleep(1)
                continue
            if st.get("compose"):
                return True
            if st.get("switcher"):
                ev(ws, """document.querySelector('[data-testid="hatch-chat-switcher-trigger"]').click()""")
                _time.sleep(1.5)
                continue
            if st.get("navchat"):
                ev(ws, """document.querySelector('[data-testid="hatch-nav-chat"]').click()""")
                _time.sleep(2)
                continue
            _time.sleep(1)
        return False
    if not _ensure_chat_active():
        print("ERROR: Could not activate chat (no compose/switcher/nav-chat)", file=_sys2.stderr)
        sys.exit(1)
    print("Chat active", file=_sys2.stderr)
    # Find + button via data-testid (verified selector, SVG icon no text)
    result = ev(ws, """(() => {
      const btn = document.querySelector('[data-testid="hatch-chat-compose"]');
      if (!btn) return 'NOT_FOUND';
      btn.click();
      return 'CLICKED';
    })()""")
    if result == 'NOT_FOUND':
        print("ERROR: + button [data-testid=hatch-chat-compose] not found", file=sys.stderr)
        sys.exit(1)
    # Log which strategy worked for debugging
    import sys as _sys
    print(f"Sidechat create: {result}", file=_sys.stderr)
    import time as _time
    # Poll for URL to change to a thread URL (up to 15s)
    # Fixed 2026-10-04: was sleeping 5s and reading once, often captured
    # base URL before navigation settled.
    url = None
    for i in range(15):
        _time.sleep(1)
        url = ev1(ws, "window.location.href")
        if url and "/thread/" in url:
            break
    if not url or "/thread/" not in url:
        print(f"ERROR: Sidechat creation did not navigate to thread URL (got: {url})", file=sys.stderr)
        sys.exit(1)
    # Return the URL (may be /thread/new placeholder).
    # The caller sends directly to current chat via cmd_send (no ID needed).
    # Thread ID can be looked up later via URL polling if needed.
    # Fixed 2026-10-04: don't chase real ID at create time, just create.
    print(f"Created: {url}")
    return url




def cdp_call(ws, method, params=None):
    """Raw CDP method call for non-Runtime domains (DOM, Page, ...).
    The ev() helper only speaks Runtime.evaluate; uploads need DOM.
    Skips CDP event chatter (messages without our id) while waiting."""
    cdp_call.counter += 1
    cid = cdp_call.counter
    ws.send(json.dumps({"id": cid, "method": method,
                        "params": params or {}}))
    for _ in range(20):
        resp = json.loads(ws.recv())
        if resp.get("id") != cid:
            continue  # event, not our response
        if "error" in resp:
            raise RuntimeError("CDP %s failed: %s" % (method, resp["error"]))
        return resp.get("result", {})
    raise RuntimeError("CDP %s: no response after 20 reads" % method)


cdp_call.counter = 100


def ev1(ws, expr, await_p=False):
    """Runtime.evaluate that skips CDP event chatter while awaiting its
    response. ev() reads a single message and can catch an event
    instead (the known None-result quirk); uploads do several DOM
    calls first, so chatter is likely.
    Returns None (no traceback) if the CDP WebSocket drops.
    (Fix 2026-10-06: uncaught WebSocketConnectionClosedException.)"""
    try:
        ws.send(json.dumps({
            "id": 1, "method": "Runtime.evaluate",
            "params": {"expression": expr, "returnByValue": True,
                       "awaitPromise": await_p}
        }))
        for _ in range(30):
            resp = json.loads(ws.recv())
            if resp.get("id") != 1:
                continue
            return resp.get("result", {}).get("result", {}).get("value")
        return None
    except Exception as e:
        print(f"CDP evaluate failed: {type(e).__name__}: {e}", file=sys.stderr)
        return None


def cmd_url(ws):
    """Print current browser URL."""
    url = ev1(ws, "window.location.href")
    print(url)

def cmd_upload(ws, filepath, message=None, dry_run=False):
    """Attach a file to the chat composer via CDP DOM.setFileInputFiles.

    Dry-run stages the attachment and verifies the preview chip without
    sending. Otherwise sends (with optional caption) and verifies via
    read-back — never trust the CDP result alone.
    """
    import os
    if not os.path.isfile(filepath):
        print("ERROR: file not found: %s" % filepath, file=sys.stderr)
        sys.exit(1)
    # React-readiness gate ("matching chromebox") - see cmd_send.
    assert_chat_ready(ws, context="upload")
    actions = check_approvals(ws)
    for dialog, trusted, action in actions:
        if not trusted:
            print("APPROVAL_NEEDED: %s" % dialog[:80], file=sys.stderr)
            sys.exit(2)
    # The file input may only render after the attach button is clicked.
    node_id = 0
    for _ in range(2):
        doc = cdp_call(ws, "DOM.getDocument")
        root = doc["root"]["nodeId"]
        q = cdp_call(ws, "DOM.querySelector",
                     {"nodeId": root, "selector": "input[type=file]"})
        node_id = q.get("nodeId", 0)
        if node_id:
            break
        ev(ws, """(() => {
          const b = [...document.querySelectorAll('button')].find(x =>
            (x.getAttribute('aria-label')||'').toLowerCase().includes('attach'));
          if (b) { b.click(); return 'clicked'; }
          return 'notfound';
        })()""")
        time.sleep(2)
    if not node_id:
        print("ERROR: no file input in composer", file=sys.stderr)
        sys.exit(1)
    cdp_call(ws, "DOM.setFileInputFiles",
             {"nodeId": node_id, "files": [os.path.abspath(filepath)]})
    time.sleep(2)
    # Read-back: the attachment must be staged. React removes the
    # file input after change, so input.files is unreliable — the
    # composer's own signal is the "Remove attachment" button plus the
    # chip text beside it.
    base = os.path.basename(filepath)
    preview = ev1(ws, """(() => {
      const btns = [...document.querySelectorAll('button')].filter(b =>
        (b.getAttribute('aria-label') || '') === 'Remove attachment');
      if (!btns.length) return JSON.stringify([]);
      const chip = btns[0].closest('div');
      const text = chip ? chip.innerText.slice(0, 120) : '';
      return JSON.stringify([{button: true, chip: text}]);
    })()""")
    print("preview: %s" % preview)
    try:
        hits = json.loads(preview) if preview else []
    except Exception:
        hits = []
    if not hits:
        print("ERROR: attachment not staged (no Remove-attachment button)",
              file=sys.stderr)
        sys.exit(1)
    if dry_run:
        print("DRY-RUN OK: '%s' staged in composer, not sent." % base)
        return
    if message:
        msg_esc = message.replace('\\', '\\\\').replace('`', '\\`').replace('$', '\\$')
        ev(ws, """(async()=>{
          const input = document.querySelector('[contenteditable="true"]') ||
                        document.querySelector('textarea[placeholder*="Message"]') ||
                        [...document.querySelectorAll('div[role="textbox"]')][0];
          if (!input) return 'NOINPUT';
          input.focus();
          document.execCommand('insertText', false, `%s`);
          return 'caption-set';
        })()""" % msg_esc, True)
        time.sleep(1)
    sent = ev(ws, """(async()=>{
      const send = [...document.querySelectorAll('button')].find(b =>
        b.getAttribute('aria-label') &&
        b.getAttribute('aria-label').toLowerCase().includes('send'));
      if (send) { send.click(); return 'sent'; }
      return 'NOSEND';
    })()""", True)
    print("send: %s" % sent)
    time.sleep(5)
    recent = ev1(ws, """(() => {
      const ps = [...document.querySelectorAll('p')].slice(-10)
        .map(p => p.innerText.slice(0,200));
      return ps.join('\\n---\\n');
    })()""")
    if base in (recent or ''):
        print("VERIFIED: attachment '%s' present in chat." % base)
    else:
        print("WARNING: attachment not found in read-back; "
              "check the composer manually.")

def main():
    p = argparse.ArgumentParser()
    p.add_argument('--account', required=True, choices=list(ACCOUNTS.keys()),
                   help='Agent name (matches node, profile, ACCOUNTS.md)')
    p.add_argument('command', choices=['send', 'messages', 'wait', 'approvals', 'sidechat', 'upload', 'url', 'compose_check'])
    p.add_argument('arg', nargs='*', default=[])
    p.add_argument('--dry-run', action='store_true',
                 help='upload: stage attachment without sending')
    p.add_argument('--message', default=None,
                 help='upload: caption text sent with the file')
    args = p.parse_args()

    node, cdp_url = ACCOUNTS[args.account]
    page = get_page(node, cdp_url)

    # CDP operation queue: serialize browser access across processes.
    # Priority from CDP_PRIORITY env (high/normal/low); default normal.
    # Falls back to unqueued if cdp_queue is unavailable.
    import contextlib
    if HAS_CDP_QUEUE:
        _prio_name = os.environ.get("CDP_PRIORITY", "normal").lower()
        _prio = {"high": PRIORITY_HIGH, "low": PRIORITY_LOW}.get(
            _prio_name, PRIORITY_NORMAL)
        _slot = cdp_slot(node, priority=_prio)
    else:
        _slot = contextlib.nullcontext()
    with _slot:
        ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=15)

        try:
            if args.command == 'send':
                if not args.arg:
                    print("ERROR: send requires a message", file=sys.stderr)
                    sys.exit(1)
                cmd_send(ws, " ".join(args.arg))
            elif args.command == 'messages':
                n = int(args.arg[0]) if args.arg else 5
                w = int(args.arg[1]) if len(args.arg) > 1 else 200
                cmd_messages(ws, n, w)
            elif args.command == 'compose_check':
                cmd_compose_check(ws)
            elif args.command == 'wait':
                t = int(args.arg[0]) if args.arg else 30
                cmd_wait(ws, t)
            elif args.command == 'approvals':
                cmd_approvals(ws)
            elif args.command == 'sidechat':
                if not args.arg:
                    print("ERROR: sidechat requires subcommand (use|main|list|create)", file=sys.stderr)
                    sys.exit(1)
                sub = args.arg[0]
                if sub == "create":
                    cmd_sidechat_create(ws)
                elif sub == "use":
                    if len(args.arg) < 2:
                        print("ERROR: sidechat use requires chat_id", file=sys.stderr)
                        sys.exit(1)
                    cmd_sidechat_use(ws, args.arg[1])
                elif sub == "main":
                    cmd_sidechat_main(ws)
                elif sub == "list":
                    cmd_sidechat_list(ws)
                else:
                    print(f"ERROR: unknown sidechat subcommand: {sub}", file=sys.stderr)
                    sys.exit(1)
            elif args.command == 'url':
                cmd_url(ws)
            elif args.command == 'upload':
                if not args.arg:
                    print("ERROR: upload requires a filepath", file=sys.stderr)
                    sys.exit(1)
                cmd_upload(ws, args.arg[0], message=args.message,
                           dry_run=args.dry_run)
        finally:
            ws.close()

if __name__ == '__main__':
    main()