box/bin/approvals.py

#!/usr/bin/env python3
"""
approvals.py — Fleet approval detection, classification, and resolution engine.

Supports both:
1. Browser DOM element detection & interaction via CDP (the primary live surface):
   - Confirmed selectors: [data-testid="approval-panel-header"],
     button[data-hatch-approval-primary-action="true"] ("Allow once"),
     and "Deny" / "Always allow this site" actions.
   - Text fallback: "Allow <agent> to share information with <IP>?"
2. Auto-approval against TRUSTED_IPS (our infrastructure).
3. Manual operator resolution (allow once, always allow, deny).
4. Continuous watch & background integration into fleet status and loop health.
"""

import json
import os
import re
import sys
import time
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

try:
    import websocket
except ImportError:
    websocket = None

# Paths
REPO_ROOT = Path("/home/super/Projects/NetVM")
BIN_DIR = REPO_ROOT / "bin"
CTL_LOG = REPO_ROOT / "box-ctl.jsonl"
RESPONDED_WAITS_FILE = REPO_ROOT / ".state" / "approvals-responded.json"
FIRST_SEEN_WAITS_FILE = REPO_ROOT / ".state" / "approvals-first-seen.json"


def load_responded_waits() -> dict:
    """Load the set of (node, task) input waits already responded to.

    Returns {node: {task: iso_timestamp}}. Missing file -> {}.
    """
    try:
        if RESPONDED_WAITS_FILE.exists():
            return json.loads(RESPONDED_WAITS_FILE.read_text())
    except Exception:
        pass
    return {}


def save_responded_waits(data: dict) -> None:
    """Persist the responded-waits map (best effort)."""
    try:
        RESPONDED_WAITS_FILE.parent.mkdir(parents=True, exist_ok=True)
        RESPONDED_WAITS_FILE.write_text(json.dumps(data, indent=1))
    except Exception:
        pass


def load_first_seen_waits() -> dict:
    """Load map of when input waits were first observed: {node: {task: iso_timestamp}}."""
    try:
        if FIRST_SEEN_WAITS_FILE.exists():
            return json.loads(FIRST_SEEN_WAITS_FILE.read_text())
    except Exception:
        pass
    return {}


def save_first_seen_waits(data: dict) -> None:
    """Persist first-seen input waits map."""
    try:
        FIRST_SEEN_WAITS_FILE.parent.mkdir(parents=True, exist_ok=True)
        FIRST_SEEN_WAITS_FILE.write_text(json.dumps(data, indent=1))
    except Exception:
        pass


def is_wait_responded(node: str, task: str) -> bool:
    """True if this (node, task) wait was already answered."""
    return task in load_responded_waits().get(node, {})


def mark_wait_responded(node: str, task: str, caller: str = "approvals") -> None:
    """Record that (node, task) has been responded to, so future
    inspections filter it out of the live input-wait list."""
    data = load_responded_waits()
    node_map = data.setdefault(node, {})
    if task not in node_map:
        node_map[task] = datetime.now(timezone.utc).isoformat()
        save_responded_waits(data)
        log_box_ctl("approval-wait-responded", name=node, caller=caller,
                     extra={"task": task})


def clear_node_waits(node: str = None, caller: str = "box-approvals") -> dict:
    """Clear and dismiss all pending input waits for a specific node or all nodes."""
    target_nodes = [node] if node else VALID_NODES
    total_cleared = 0
    cleared_per_node = {}
    data = load_responded_waits()
    now_iso = datetime.now(timezone.utc).isoformat()

    for n in target_nodes:
        node_map = data.setdefault(n, {})
        n_cleared = 0
        try:
            info = inspect_node_approvals(n)
            for w in info.get("input_waits", []) or []:
                t = w.get("task")
                if t and t not in node_map:
                    node_map[t] = now_iso
                    n_cleared += 1
            if n_cleared:
                log_box_ctl("approval-wait-cleared", name=n, caller=caller,
                            extra={"cleared_count": n_cleared})
        except Exception:
            pass
        cleared_per_node[n] = n_cleared
        total_cleared += n_cleared

    if total_cleared:
        save_responded_waits(data)

    return {"ok": True, "total_cleared": total_cleared, "cleared_per_node": cleared_per_node}

# Add BIN_DIR to sys.path
if str(BIN_DIR) not in sys.path:
    sys.path.insert(0, str(BIN_DIR))

try:
    import netvm_registry
except ImportError:
    netvm_registry = None

VALID_NODES = ["muse", "pip", "646", "opm", "def", "dev"]

# Approval timeout defaults (seconds).
# Key requests are sensitive operations -- give the operator 2h to decide.
# Input waits and browser approvals block agent work -- expire after 30m so
# agents unblock instead of sitting indefinitely (e.g. def waited 4h+).
KEY_REQUEST_TTL_SECONDS = 2 * 3600
INPUT_WAIT_TTL_SECONDS = 30 * 60
BROWSER_APPROVAL_TTL_SECONDS = 30 * 60

# Trusted infrastructure IPs safe for automated approval
TRUSTED_IPS = {
    "34.139.37.135",  # VM (gateway)
    "100.123.153.75", # bl (main compute)
    "100.81.31.9",    # VM tailnet
    "1.1.1.1",        # Cloudflare DNS
    "1.0.0.1",        # Cloudflare DNS
}

# Trusted infrastructure domains safe for automated approval
TRUSTED_DOMAINS = {
    "muse-dev.online",
}


def is_trusted_target(target: str, card_text: str = "") -> bool:
    """Check if the extracted approval target is trusted infrastructure.

    Fail-closed: only the parsed target (IP or hostname) is evaluated. Free-form
    card text is deliberately NOT substring-matched, since an untrusted request
    could mention a trusted domain in its purpose string. `card_text` is kept
    for signature compatibility.
    """
    if not target:
        return False
    target = target.strip().lower().rstrip(".")
    if target in TRUSTED_IPS:
        return True
    for dom in TRUSTED_DOMAINS:
        if target == dom or target.endswith("." + dom):
            return True
    return False


REDACT_PATTERNS = [
    (re.compile(r"Bearer\s+[A-Za-z0-9._~+/-]+=*", re.IGNORECASE), "Bearer [REDACTED]"),
    (re.compile(r"eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9._-]{10,}", re.IGNORECASE), "[JWT-REDACTED]"),
    (re.compile(r"(?i)\b(api[_-]?key|token|secret|password|auth|passkey)\s*[:=]\s*(['\"]?)([A-Za-z0-9_\-\.]{4,})\2"), r"\1: \2[REDACTED]\2"),
    (re.compile(r"-----BEGIN [A-Z ]+ PRIVATE KEY-----[\s\S]*?-----END [A-Z ]+ PRIVATE KEY-----"), "[PRIVATE-KEY-REDACTED]"),
]


def redact_sensitive(text: str) -> str:
    """Mask credentials, tokens, and passkeys in text."""
    if not text or not isinstance(text, str):
        return text
    out = text
    for pattern, repl in REDACT_PATTERNS:
        out = pattern.sub(repl, out)
    return out


def _approval_type(action: str) -> str:
    """Classify an approval action into its request type.

    Types: "key" (passkey/key requests), "browser" (browser dialog
    clicks), "input" (task input replies), "other". Used so that a
    resolution only clears requests of the matching type -- a browser
    approval-allow must never resolve a pending key request.
    """
    if action.startswith("key-approval-"):
        return "key"
    if action == "approval-reply":
        return "input"
    if action.startswith("approval-"):
        return "browser"
    return "other"


def log_box_ctl(action: str, name: str = None, caller: str = "box-approvals", extra: dict = None):
    """Log an audit event to box-ctl.jsonl with secret redaction."""
    try:
        rec = {
            "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "action": action,
            "type": _approval_type(action),
            "name": name,
            "caller": caller,
        }
        if extra:
            clean_extra = {}
            for k, v in extra.items():
                if isinstance(v, str):
                    clean_extra[k] = redact_sensitive(v)
                else:
                    clean_extra[k] = v
            rec.update(clean_extra)
        with open(CTL_LOG, "a") as f:
            f.write(json.dumps(rec) + "\n")
    except Exception:
        pass


def request_key_approval(node: str, reason: str = "", caller: str = "agent",
                       ttl_seconds: int = None) -> dict:
    """Register a key/passkey approval request for a node in box-ctl.jsonl.

    ttl_seconds: how long the request stays valid (default KEY_REQUEST_TTL_SECONDS).
    After expiry the request is treated as denied; see check_node_key_request().
    """
    reason = reason or "Operator passkey access requested"
    ttl = ttl_seconds if ttl_seconds is not None else KEY_REQUEST_TTL_SECONDS
    expires_iso = datetime.fromtimestamp(
        datetime.now(timezone.utc).timestamp() + ttl, tz=timezone.utc
    ).strftime("%Y-%m-%dT%H:%M:%SZ")
    log_box_ctl(
        "key-approval-request",
        name=node,
        caller=caller,
        extra={"reason": reason, "ttl_seconds": ttl, "expires_at": expires_iso},
    )
    return {
        "ok": True,
        "node": node,
        "status": "KEY_APPROVAL_REQUESTED",
        "reason": reason,
        "caller": caller,
        "ttl_seconds": ttl,
        "expires_at": expires_iso,
        "note": "Request recorded in audit log. Operator can approve via 'box approvals allow <node>'.",
    }


def _parse_ts(ts_str: str):
    """Parse a box-ctl.jsonl ts ('%Y-%m-%dT%H:%M:%SZ') to epoch seconds. None on failure."""
    if not ts_str:
        return None
    try:
        dt = datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
        return dt.timestamp()
    except Exception:
        return None


def expire_key_request(node: str, caller: str = "approval-sweeper") -> dict:
    """Mark a node's pending key request as expired (auto-deny on TTL)."""
    log_box_ctl("key-approval-expired", name=node, caller=caller,
                extra={"note": "TTL elapsed without operator decision; treated as denied"})
    return {"ok": True, "node": node, "status": "KEY_APPROVAL_EXPIRED"}


def _rec_approval_type(rec: dict) -> str:
    """Return the approval type of a log record.

    Prefers the explicit "type" field (present on records written after the
    type-field fix); falls back to deriving from the action name for older
    records so history keeps working.
    """
    t = rec.get("type")
    if t:
        return t
    return _approval_type(rec.get("action", ""))


def check_node_key_request(node: str) -> dict:
    """Check if node has an active unfulfilled key approval request in box-ctl.jsonl.

    Requests expire after their TTL (default KEY_REQUEST_TTL_SECONDS). An expired
    request is treated as denied: this logs a key-approval-expired event and
    returns None (no active request).
    """
    if not CTL_LOG.exists():
        return None
    latest_req = None
    resolved = False
    now = datetime.now(timezone.utc).timestamp()
    try:
        with open(CTL_LOG, "r") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    rec = json.loads(line)
                except Exception:
                    continue
                if rec.get("name") != node:
                    continue
                act = rec.get("action")
                if act == "key-approval-request":
                    latest_req = rec
                    resolved = False
                elif _rec_approval_type(rec) == "key" and act in (
                    "key-approval-allow", "key-approval-deny", "key-approval-expired",
                ):
                    # Only a KEY-type resolution clears a key request. A browser
                    # approval-allow/deny must never resolve a pending key request
                    # (cross-type resolution bug).
                    resolved = True
    except Exception:
        return None

    if latest_req and not resolved:
        # TTL check: explicit expires_at wins; legacy records fall back to
        # ts + default TTL.
        exp_ts = _parse_ts(latest_req.get("expires_at"))
        if exp_ts is None:
            req_ts = _parse_ts(latest_req.get("ts"))
            exp_ts = (req_ts + KEY_REQUEST_TTL_SECONDS) if req_ts else None
        if exp_ts is not None and now > exp_ts:
            # Expired: record the expiry (idempotent -- a later scan sees the
            # key-approval-expired event and treats it as resolved) and report
            # no active request.
            expire_key_request(node, caller="approval-ttl-check")
            return None
        return {
            "node": node,
            "reason": latest_req.get("reason", "Operator passkey access requested"),
            "requested_at": latest_req.get("ts", ""),
            "caller": latest_req.get("caller", ""),
            "expires_at": latest_req.get("expires_at", ""),
        }
    return None


def sweep_expired_key_requests() -> dict:
    """Proactively expire stale key requests across all nodes.

    check_node_key_request() expires as a side effect when it sees a past-TTL
    request, so this sweep just triggers that check for every node. Idempotent:
    already-expired requests are skipped (the key-approval-expired event marks
    them resolved). Returns {"expired_now": [nodes expired by this sweep]}.
    """
    expired_now = []
    if not CTL_LOG.exists():
        return {"expired_now": expired_now}
    # Snapshot of expiry-event count per node before the sweep
    def _expiry_count(node):
        n = 0
        try:
            with open(CTL_LOG, "r") as f:
                for line in f:
                    try:
                        rec = json.loads(line.strip())
                    except Exception:
                        continue
                    if rec.get("name") == node and rec.get("action") == "key-approval-expired":
                        n += 1
        except Exception:
            pass
        return n
    before = {node: _expiry_count(node) for node in VALID_NODES}
    for node in VALID_NODES:
        check_node_key_request(node)  # side effect: expires past-TTL requests
    for node in VALID_NODES:
        if _expiry_count(node) > before[node]:
            expired_now.append(node)
    return {"expired_now": sorted(expired_now)}


def get_node_connection_info(node: str) -> dict:
    """Return peer_ip, cdp_port, and netns for a given node."""
    if netvm_registry:
        nodes = netvm_registry.load()
        if node in nodes:
            rec = nodes[node]
            return {
                "node": node,
                "peer_ip": rec.get("peer_ip", f"10.201.87.2"),
                "cdp_port": rec.get("cdp_port", 9222),
                "netns": rec.get("netns", f"warp-{node}"),
            }
    # Deterministic fallback
    import hashlib
    tag = hashlib.sha256(node.encode()).hexdigest()[:8]
    idx = int(tag[:3], 16) % 200 + 10
    peer_ip = f"10.201.{idx}.2"
    pinned = {"muse": 9410, "pip": 9420, "646": 9430, "opm": 9440, "def": 9450, "dev": 9455}
    return {
        "node": node,
        "peer_ip": peer_ip,
        "cdp_port": pinned.get(node, 9222),
        "netns": f"warp-{node}",
    }


def get_node_pages(node: str, timeout: float = 3.0) -> list:
    """Return all active page targets for a node."""
    if websocket is None:
        raise RuntimeError("websocket-client library is required")

    info = get_node_connection_info(node)
    peer_ip = info["peer_ip"]
    port = info["cdp_port"]

    urls = [
        f"http://{peer_ip}:{port}/json/list",
        f"http://127.0.0.1:{port}/json/list",
    ]
    tabs = None
    last_err = None
    for u in urls:
        try:
            req = urllib.request.Request(u, headers={"User-Agent": "box-approvals/1.0"})
            with urllib.request.urlopen(req, timeout=timeout) as r:
                tabs = json.load(r)
                break
        except Exception as e:
            last_err = e
            continue

    if not tabs:
        raise ConnectionError(f"Could not reach CDP for {node}: {last_err}")

    pages = [t for t in tabs if t.get("type") == "page"]
    if not pages:
        raise ConnectionError(f"No active page found for node {node}")
    return pages


def get_cdp_ws(node: str, page_idx: int = 0, timeout: float = 3.0):
    """Connect to a node's browser page over CDP WebSocket."""
    pages = get_node_pages(node, timeout=timeout)
    if page_idx >= len(pages):
        page_idx = 0
    target_page = pages[page_idx]
    ws_url = target_page.get("webSocketDebuggerUrl")
    if not ws_url:
        raise ConnectionError(f"No webSocketDebuggerUrl for node {node}")
    ws = websocket.create_connection(ws_url, timeout=timeout)
    return ws, target_page


def cdp_evaluate(ws, js_expr: str, await_promise: bool = False, timeout: float = 3.0):
    """Evaluate a JavaScript expression via CDP Runtime.evaluate and return the result value."""
    req_id = int(time.time() * 1000) % 100000
    msg = {
        "id": req_id,
        "method": "Runtime.evaluate",
        "params": {
            "expression": js_expr,
            "returnByValue": True,
            "awaitPromise": await_promise,
        },
    }
    ws.send(json.dumps(msg))
    deadline = time.time() + timeout
    while time.time() < deadline:
        raw = ws.recv()
        resp = json.loads(raw)
        if resp.get("id") == req_id:
            res = resp.get("result", {})
            if "exceptionDetails" in res:
                return {"error": res["exceptionDetails"].get("text", "JS exception")}
            return res.get("result", {}).get("value")
    return None


JS_INSPECT_APPROVALS = """(() => {
    // 1. Locate active approval panels
    const headers = Array.from(document.querySelectorAll('[data-testid="approval-panel-header"], [data-testid*="approval"]'));
    let activeCard = null;
    let cardText = '';

    for (const h of headers) {
        let curr = h;
        for (let i = 0; i < 6 && curr && curr.parentElement && curr.parentElement !== document.body; i++) {
            const hasPrimary = !!curr.querySelector('button[data-hatch-approval-primary-action="true"]');
            const btns = Array.from(curr.querySelectorAll('button')).map(b => (b.innerText||'').trim().toLowerCase());
            const hasAllow = btns.some(t => t.includes('allow once') || t === 'allow');
            const hasDeny = btns.some(t => t === 'deny');
            if (hasPrimary || (hasAllow && hasDeny)) {
                activeCard = curr;
                cardText = curr.innerText || '';
                break;
            }
            curr = curr.parentElement;
        }
        if (activeCard) break;
    }

    // Fallback: look for "Allow ... to share" or buttons
    if (!activeCard) {
        const bodyText = document.body ? document.body.innerText : '';
        if (bodyText.includes('Allow') && bodyText.includes('to share')) {
            const btns = Array.from(document.querySelectorAll('button'));
            const allowBtn = btns.find(b => (b.innerText||'').trim().toLowerCase().includes('allow once'));
            if (allowBtn) {
                let curr = allowBtn;
                for (let i = 0; i < 5 && curr && curr.parentElement && curr.parentElement !== document.body; i++) {
                    if (curr.innerText && curr.innerText.includes('Allow') && curr.innerText.includes('to share')) {
                        activeCard = curr;
                        cardText = curr.innerText;
                        break;
                    }
                    curr = curr.parentElement;
                }
            }
        }
    }

    // Inspect buttons in active card
    const buttons = [];
    let hasAllowOnce = false;
    let hasAlwaysAllow = false;
    let hasDeny = false;

    if (activeCard) {
        const btns = Array.from(activeCard.querySelectorAll('button'));
        for (const b of btns) {
            const t = (b.innerText || '').trim();
            const low = t.toLowerCase();
            if (low) buttons.push(t);
            if (low === 'allow once' || b.getAttribute('data-hatch-approval-primary-action') === 'true') hasAllowOnce = true;
            if (low.includes('always allow')) hasAlwaysAllow = true;
            if (low === 'deny') hasDeny = true;
        }
    }

    // Collect historical recent approval badges from chat stream
    const historyBadges = [];
    const allBtns = Array.from(document.querySelectorAll('button'));
    for (const b of allBtns) {
        const txt = (b.innerText || '').trim();
        if (txt.includes('Allowed once ·') || txt.includes('Timed out ·') || txt.includes('Site always allowed ·') || txt.includes('Allowed for this scheduled task ·')) {
            const lines = txt.split('\\n');
            const summary = lines[0] || '';
            const statusLine = lines[lines.length - 1] || '';
            historyBadges.push({ summary, status: statusLine });
        }
    }

    // Task rows in the Activity sidebar waiting on human input
    // (e.g. "Launch 5 OPM Sub-Agents\\nAsked for input to start the launch\\n5:53 pm").
    const inputWaits = [];
    for (const b of document.querySelectorAll('button.rounded-10, a.rounded-10')) {
        const lines = (b.innerText || '').split('\\n').map(s => s.trim()).filter(Boolean);
        if (lines.length >= 2 && /^(asked for (input|details)|waiting for (your )?(input|reply|approval)|needs your input)/i.test(lines[1])) {
            inputWaits.push({ task: lines[0], status: lines[1], when: lines[2] || '' });
        }
    }

    return JSON.stringify({
        has_pending: !!activeCard && (hasAllowOnce || hasDeny),
        card_text: cardText.slice(0, 1000),
        buttons: buttons,
        has_allow_once: hasAllowOnce,
        has_always_allow: hasAlwaysAllow,
        has_deny: hasDeny,
        history: historyBadges.slice(0, 5),
        input_waits: inputWaits.slice(0, 10)
    });
})()"""


def inspect_node_approvals(node: str) -> dict:
    """Inspect a node across all open page targets for active approval prompts and input waits."""
    try:
        pages = get_node_pages(node, timeout=2.5)
    except Exception as e:
        key_req = check_node_key_request(node)
        if key_req:
            return {
                "node": node,
                "status": "KEY_APPROVAL",
                "has_pending": True,
                "title": f"Passkey requested: {key_req.get('reason')}",
                "purpose": key_req.get("reason"),
                "ip": "34.139.37.135",
                "target": "34.139.37.135 (VM passkey)",
                "is_trusted": True,
                "buttons": ["Allow", "Deny"],
                "has_allow_once": True,
                "has_always_allow": False,
                "has_deny": True,
                "raw_text": f"Agent on node {node} requested passkey: {key_req.get('reason')}",
                "history": [],
                "input_waits": [],
                "page_title": "",
                "page_url": "",
                "ws_url": "",
                "key_request": key_req,
            }
        return {
            "node": node,
            "status": "UNREACHABLE",
            "error": str(e),
            "has_pending": False,
        }

    all_input_waits = []
    first_page = pages[0]
    last_err = None

    for page in pages:
        ws_url = page.get("webSocketDebuggerUrl")
        if not ws_url:
            continue
        ws = None
        try:
            ws = websocket.create_connection(ws_url, timeout=2.0)
            val_str = cdp_evaluate(ws, JS_INSPECT_APPROVALS, timeout=2.5)
            ws.close()
            ws = None
            if not val_str or not isinstance(val_str, str):
                continue
            data = json.loads(val_str)
            if data.get("input_waits"):
                all_input_waits.extend(data["input_waits"])

            if data.get("has_pending"):
                card_text = data.get("card_text", "")
                ip = None
                target = None
                m_t = re.search(
                    r"(?:connection to|share information with|contact|connect to)\s+([A-Za-z0-9][A-Za-z0-9.-]*[A-Za-z0-9])",
                    card_text,
                )
                if m_t:
                    target = m_t.group(1)
                m_ip = re.search(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b", target or card_text)
                if m_ip:
                    ip = m_ip.group(0)
                if not target:
                    target = ip
                if not target:
                    m_domain = re.search(r"\b([a-zA-Z0-9-]+\.)+(?:online|com|net|org|io|dev)\b", card_text)
                    if m_domain:
                        target = m_domain.group(0)

                lines = [line.strip() for line in card_text.split("\n") if line.strip()]
                title = redact_sensitive(lines[0] if lines else "Permission request")
                purpose = redact_sensitive(lines[1] if len(lines) > 1 else "")

                is_trusted = is_trusted_target(target or ip, card_text)

                return {
                    "node": node,
                    "status": "PENDING",
                    "has_pending": True,
                    "title": title,
                    "purpose": purpose,
                    "ip": ip,
                    "target": target or ip or "-",
                    "is_trusted": is_trusted,
                    "buttons": data.get("buttons", []),
                    "has_allow_once": data.get("has_allow_once", False),
                    "has_always_allow": data.get("has_always_allow", False),
                    "has_deny": data.get("has_deny", False),
                    "raw_text": redact_sensitive(card_text),
                    "history": data.get("history", []),
                    "input_waits": all_input_waits,
                    "page_title": page.get("title", ""),
                    "page_url": page.get("url", ""),
                    "ws_url": ws_url,
                }
        except Exception as e:
            last_err = e
            if ws:
                try:
                    ws.close()
                except Exception:
                    pass

    # Deduplicate input waits by task name
    unique_waits = []
    seen_tasks = set()
    for w in all_input_waits:
        t_name = redact_sensitive(w.get("task", ""))
        status_text = redact_sensitive(w.get("status", ""))
        if t_name not in seen_tasks:
            seen_tasks.add(t_name)
            unique_waits.append({
                "task": t_name,
                "status": status_text,
                "when": w.get("when", ""),
            })

    # Filter out waits already responded to (stale sidebar entries that
    # muse.ai never cleared). Responded set is maintained by
    # reply_node_task() and mark_wait_responded().
    responded = load_responded_waits().get(node, {})
    if responded:
        unique_waits = [w for w in unique_waits if w.get("task") not in responded]

    # TTL check for input waits: auto-expire stale waits older than INPUT_WAIT_TTL_SECONDS
    if unique_waits:
        first_seen = load_first_seen_waits()
        node_seen = first_seen.setdefault(node, {})
        now_dt = datetime.now(timezone.utc)
        now_iso = now_dt.isoformat()
        first_seen_changed = False
        surviving_waits = []

        for w in unique_waits:
            t = w.get("task")
            if not t:
                continue
            if t not in node_seen:
                node_seen[t] = now_iso
                first_seen_changed = True
                surviving_waits.append(w)
            else:
                try:
                    seen_dt = datetime.fromisoformat(node_seen[t])
                    age_seconds = (now_dt - seen_dt).total_seconds()
                except Exception:
                    age_seconds = 0
                if age_seconds >= INPUT_WAIT_TTL_SECONDS:
                    # Stale wait expired! Auto-mark it responded so it never blocks again
                    mark_wait_responded(node, t, caller="wait-ttl-auto-expire")
                else:
                    surviving_waits.append(w)

        if first_seen_changed:
            save_first_seen_waits(first_seen)
        unique_waits = surviving_waits

    key_req = check_node_key_request(node)
    if key_req:
        return {
            "node": node,
            "status": "KEY_APPROVAL",
            "has_pending": True,
            "title": f"Passkey requested: {key_req.get('reason')}",
            "purpose": key_req.get("reason"),
            "ip": "34.139.37.135",
            "target": "34.139.37.135 (VM passkey)",
            "is_trusted": True,
            "buttons": ["Allow", "Deny"],
            "has_allow_once": True,
            "has_always_allow": False,
            "has_deny": True,
            "raw_text": f"Agent on node {node} requested passkey: {key_req.get('reason')}",
            "history": [],
            "input_waits": unique_waits,
            "page_title": first_page.get("title", ""),
            "page_url": first_page.get("url", ""),
            "ws_url": first_page.get("webSocketDebuggerUrl", ""),
            "key_request": key_req,
        }

    status = "INPUT_WAIT" if unique_waits else ("ERROR" if last_err and not first_page else "CLEAR")
    return {
        "node": node,
        "status": status,
        "has_pending": False,
        "title": "No pending approvals",
        "purpose": "",
        "ip": None,
        "target": "-",
        "is_trusted": False,
        "buttons": [],
        "has_allow_once": False,
        "has_always_allow": False,
        "has_deny": False,
        "raw_text": "",
        "history": [],
        "input_waits": unique_waits,
        "page_title": first_page.get("title", ""),
        "page_url": first_page.get("url", ""),
        "ws_url": first_page.get("webSocketDebuggerUrl", ""),
    }


def check_fleet_approvals(nodes: list = None) -> list:
    """Check approval status across the fleet."""
    target_nodes = nodes or VALID_NODES
    results = []
    for node in target_nodes:
        results.append(inspect_node_approvals(node))
    return results



def _notify_key_decision(node: str, decision: str, reason: str, message: str,
                         allow_main_chat: bool = False, caller: str = "box-approvals") -> dict:
    """Notify the waiting agent of a key-approval decision via their thread.

    Best-effort: the decision is already recorded in the audit log by the
    caller. If notification fails, the operator must follow up manually.
    Returns the reply_node_task result dict.
    """
    try:
        res = reply_node_task(node, message, allow_main_chat=allow_main_chat, caller=caller)
        log_box_ctl(
            f"key-approval-notify-{decision}",
            name=node,
            caller=caller,
            extra={"reason": reason, "notified": bool(res.get("ok")), "notify_error": res.get("error", "")},
        )
        return res
    except Exception as e:
        return {"ok": False, "node": node, "error": f"notify exception: {e}"}


def allow_node_approval(node: str, always: bool = False, force: bool = False, caller: str = "box-approvals", message: str = None, allow_main_chat: bool = False) -> dict:
    """Approve a pending approval on a node (click 'Allow once' or 'Always allow this site')."""
    info = inspect_node_approvals(node)
    if not info.get("has_pending"):
        return {"ok": False, "node": node, "error": "No pending approval dialog found on node"}

    if info.get("status") == "KEY_APPROVAL":
        key_req = info.get("key_request") or check_node_key_request(node) or {}
        reason = key_req.get("reason", info.get("purpose", ""))
        log_box_ctl(
            "key-approval-allow",
            name=node,
            caller=caller,
            extra={
                "reason": reason,
                "forced": force,
            },
        )
        # Execute-gap fix: notify the waiting agent. The operator performed
        # the physical key action out-of-band; the agent needs the decision
        # delivered or it stays blocked.
        notify_msg = message or (
            f"[operator] Key/passkey request APPROVED ({reason}). "
            "Operator action complete - you may proceed."
        )
        notify_res = _notify_key_decision(
            node, "allow", reason, notify_msg, allow_main_chat, caller
        )
        return {
            "ok": True,
            "node": node,
            "type": "key_approval",
            "decision": "allow",
            "dismissed": True,
            "reason": reason,
            "title": info.get("title"),
            "notified": bool(notify_res.get("ok")),
            "notify_result": notify_res,
        }

    if not info.get("is_trusted") and not force:
        target = info.get("ip") or "unrecognized target"
        return {
            "ok": False,
            "node": node,
            "error": f"Untrusted origin ({target}). Human review required. Use --force to override.",
            "approval": info,
        }

    try:
        ws_url = info.get("ws_url")
        if ws_url:
            ws = websocket.create_connection(ws_url, timeout=3.0)
        else:
            ws, _ = get_cdp_ws(node, timeout=3.0)
    except Exception as e:
        return {"ok": False, "node": node, "error": f"Failed to connect to CDP: {e}"}

    try:
        if always:
            js_click = """(() => {
                const btns = Array.from(document.querySelectorAll('button'));
                let btn = btns.find(b => (b.innerText||'').toLowerCase().includes('always allow'));
                if (btn) {
                    btn.click();
                    return 'CLICKED_ALWAYS';
                }
                btn = document.querySelector('button[data-hatch-approval-primary-action="true"]');
                if (!btn) {
                    btn = btns.find(b => (b.innerText||'').toLowerCase().includes('allow once') || (b.innerText||'').trim().toLowerCase() === 'allow');
                }
                if (btn) {
                    btn.click();
                    return 'CLICKED_PRIMARY_FALLBACK';
                }
                return 'NOT_FOUND';
            })()"""
        else:
            js_click = """(() => {
                const primary = document.querySelector('button[data-hatch-approval-primary-action="true"]');
                if (primary) {
                    primary.click();
                    return 'CLICKED_PRIMARY';
                }
                const btns = Array.from(document.querySelectorAll('button'));
                const btn = btns.find(b => {
                    const t = (b.innerText||'').trim().toLowerCase();
                    return t === 'allow once' || t === 'allow';
                });
                if (btn) {
                    btn.click();
                    return 'CLICKED_ALLOW';
                }
                return 'NOT_FOUND';
            })()"""

        click_res = cdp_evaluate(ws, js_click, timeout=3.0)

        # Verify dismissal
        time.sleep(0.8)
        js_verify = """(() => {
            const primary = document.querySelector('button[data-hatch-approval-primary-action="true"]');
            if (primary) return 'STILL_PRESENT';
            const headers = document.querySelectorAll('[data-testid="approval-panel-header"]');
            return headers.length === 0 ? 'DISMISSED' : 'STILL_PRESENT';
        })()"""
        verify_res = cdp_evaluate(ws, js_verify, timeout=2.0)
        ws.close()

        dismissed = verify_res == "DISMISSED"
        mode = "always" if always else "allow_once"
        log_box_ctl(
            "approval-allow",
            name=node,
            caller=caller,
            extra={
                "decision": mode,
                "target_ip": info.get("ip"),
                "dismissed": dismissed,
                "forced": force,
            },
        )

        return {
            "ok": True,
            "node": node,
            "decision": mode,
            "click_result": click_res,
            "dismissed": dismissed,
            "target_ip": info.get("ip"),
            "title": info.get("title"),
        }
    except Exception as e:
        try:
            ws.close()
        except Exception:
            pass
        return {"ok": False, "node": node, "error": str(e)}


def deny_node_approval(node: str, caller: str = "box-approvals", message: str = None, allow_main_chat: bool = False) -> dict:
    """Deny a pending approval on a node (click 'Deny' or deny key request)."""
    info = inspect_node_approvals(node)
    if not info.get("has_pending"):
        return {"ok": False, "node": node, "error": "No pending approval dialog found on node"}

    if info.get("status") == "KEY_APPROVAL":
        key_req = info.get("key_request") or check_node_key_request(node) or {}
        reason = key_req.get("reason", info.get("purpose", ""))
        log_box_ctl(
            "key-approval-deny",
            name=node,
            caller=caller,
            extra={
                "reason": reason,
            },
        )
        # Execute-gap fix: notify the waiting agent of the denial.
        notify_msg = message or (
            f"[operator] Key/passkey request DENIED ({reason}). "
            "Do not proceed with the protected action."
        )
        notify_res = _notify_key_decision(
            node, "deny", reason, notify_msg, allow_main_chat, caller
        )
        return {
            "ok": True,
            "node": node,
            "type": "key_approval",
            "decision": "deny",
            "dismissed": True,
            "reason": reason,
            "title": info.get("title"),
            "notified": bool(notify_res.get("ok")),
            "notify_result": notify_res,
        }

    try:
        ws_url = info.get("ws_url")
        if ws_url:
            ws = websocket.create_connection(ws_url, timeout=3.0)
        else:
            ws, _ = get_cdp_ws(node, timeout=3.0)
    except Exception as e:
        return {"ok": False, "node": node, "error": f"Failed to connect to CDP: {e}"}

    try:
        js_deny = """(() => {
            const btns = Array.from(document.querySelectorAll('button'));
            const btn = btns.find(b => (b.innerText||'').trim().toLowerCase() === 'deny');
            if (btn) {
                btn.click();
                return 'CLICKED_DENY';
            }
            return 'NOT_FOUND';
        })()"""
        click_res = cdp_evaluate(ws, js_deny, timeout=3.0)

        # Verify dismissal
        time.sleep(0.8)
        js_verify = """(() => {
            const headers = document.querySelectorAll('[data-testid="approval-panel-header"]');
            return headers.length === 0 ? 'DISMISSED' : 'STILL_PRESENT';
        })()"""
        verify_res = cdp_evaluate(ws, js_verify, timeout=2.0)
        ws.close()

        dismissed = verify_res == "DISMISSED"
        log_box_ctl(
            "approval-deny",
            name=node,
            caller=caller,
            extra={
                "decision": "deny",
                "target_ip": info.get("ip"),
                "dismissed": dismissed,
            },
        )

        return {
            "ok": True,
            "node": node,
            "decision": "deny",
            "click_result": click_res,
            "dismissed": dismissed,
            "target_ip": info.get("ip"),
        }
    except Exception as e:
        try:
            ws.close()
        except Exception:
            pass
        return {"ok": False, "node": node, "error": str(e)}


def auto_approve_fleet(nodes: list = None, always: bool = True, caller: str = "box-approvals") -> dict:
    """Scan fleet nodes and automatically approve any requests to TRUSTED_IPS with Always Allow."""
    fleet = check_fleet_approvals(nodes)
    approved = []
    untrusted = []
    clear = []

    for item in fleet:
        node = item["node"]
        if item.get("has_pending"):
            if item.get("status") == "KEY_APPROVAL":
                # Key approvals require explicit operator decision, never auto-approve
                untrusted.append(item)
            elif item.get("is_trusted"):
                res = allow_node_approval(node, always=always, caller=caller)
                approved.append({
                    "node": node,
                    "target_ip": item.get("ip"),
                    "title": item.get("title"),
                    "decision": "always" if always else "allow_once",
                    "res": res,
                })
            else:
                untrusted.append(item)
        else:
            clear.append(node)

    return {
        "ok": True,
        "auto_approved": approved,
        "untrusted_pending": untrusted,
        "clear_nodes": clear,
    }


def reply_node_task(node: str, message: str, allow_main_chat: bool = False, caller: str = "box-approvals") -> dict:
    """Send an operator reply into the waiting agent's thread to answer an input prompt."""
    info = inspect_node_approvals(node)
    page_url = info.get("page_url", "")
    page_title = info.get("page_title", "")
    ws_url = info.get("ws_url")

    # Check if target is Main Chat
    # Main chat signatures: not sidechat and (no /thread/ or title contains 'Chat —')
    is_main = "sidechat" not in page_url.lower() and ("/thread/" not in page_url or "chat —" in page_title.lower())
    if is_main and not allow_main_chat:
        return {
            "ok": False,
            "node": node,
            "error": "Active thread appears to be Main Chat. Refusing reply by sidechat-first policy. Pass --allow-main-chat to confirm intentional operator override.",
            "page_title": page_title,
            "page_url": page_url,
        }

    try:
        if ws_url:
            ws = websocket.create_connection(ws_url, timeout=3.0)
        else:
            ws, _ = get_cdp_ws(node, timeout=3.0)
    except Exception as e:
        return {"ok": False, "node": node, "error": f"Failed to connect to CDP: {e}"}

    try:
        msg_esc = message.replace('\\', '\\\\').replace('`', '\\`').replace('$', '\\$').replace('"', '\\"')
        js_type_and_send = f"""(async() => {{
            const input = document.querySelector('[contenteditable="true"]') ||
                          document.querySelector('textarea[placeholder*="Message"]') ||
                          document.querySelector('textarea');
            if (!input) return 'NO_INPUT';
            input.focus();
            document.execCommand('insertText', false, "{msg_esc}");
            await new Promise(r => setTimeout(r, 400));
            const sendBtn = Array.from(document.querySelectorAll('button')).find(b => {{
                const a = (b.getAttribute('aria-label') || '').toLowerCase();
                const t = (b.innerText || '').toLowerCase();
                return a.includes('send') || t === 'send';
            }});
            if (sendBtn && !sendBtn.disabled) {{
                sendBtn.click();
                return 'CLICKED_SEND';
            }}
            const ke = new KeyboardEvent('keydown', {{key: 'Enter', code: 'Enter', keyCode: 13, bubbles: true}});
            input.dispatchEvent(ke);
            return 'ENTER_SENT';
        }})()"""
        send_res = cdp_evaluate(ws, js_type_and_send, await_promise=True, timeout=5.0)
        ws.close()

        log_box_ctl(
            "approval-reply",
            name=node,
            caller=caller,
            extra={
                "message_len": len(message),
                "page_url": page_url,
                "allow_main_chat": allow_main_chat,
                "send_res": send_res,
            },
        )
        # The wait(s) visible at reply time are now answered — record them
        # so the sidebar's stale entries stop re-alerting.
        for w in info.get("input_waits", []) or []:
            t = w.get("task")
            if t:
                mark_wait_responded(node, t, caller=caller)
        return {
            "ok": True,
            "node": node,
            "send_result": send_res,
            "page_title": page_title,
            "page_url": page_url,
        }
    except Exception as e:
        try:
            ws.close()
        except Exception:
            pass
        return {"ok": False, "node": node, "error": str(e)}


def dismiss_node_task(node: str, caller: str = "box-approvals") -> dict:
    """Close any open task modal dialog or popup on a node and clear active input waits."""
    clear_res = clear_node_waits(node, caller=caller)
    try:
        ws, _ = get_cdp_ws(node, timeout=3.0)
    except Exception as e:
        return {
            "ok": True,
            "node": node,
            "result": "WAITS_CLEARED",
            "cleared_waits": clear_res.get("cleared_per_node", {}).get(node, 0),
            "warning": f"CDP unreachable ({e}), but input waits cleared",
        }

    try:
        js_dismiss = """(() => {
            const close = document.querySelector('[aria-label="Close"], button[data-slot="dialog-close"]');
            if (close) { close.click(); return 'CLICKED_CLOSE'; }
            document.dispatchEvent(new KeyboardEvent('keydown', {key: 'Escape', code: 'Escape', keyCode: 27, bubbles: true}));
            return 'ESCAPE_SENT';
        })()"""
        res = cdp_evaluate(ws, js_dismiss, timeout=2.0)
        ws.close()
        return {
            "ok": True,
            "node": node,
            "result": res,
            "cleared_waits": clear_res.get("cleared_per_node", {}).get(node, 0),
        }
    except Exception as e:
        try:
            ws.close()
        except Exception:
            pass
        return {
            "ok": True,
            "node": node,
            "result": "WAITS_CLEARED",
            "cleared_waits": clear_res.get("cleared_per_node", {}).get(node, 0),
        }