//! clawmates-node — the fleet daemon a user installs on each local-hardware node.
//!
//! It dials home to the ClawMates gateway over an OUTBOUND WebSocket
//! (`/api/nodes/agent?token=…`), reports host health on a heartbeat, and runs the
//! commands the gateway sends (host verification today; container placement in a
//! later phase). Outbound-only → no inbound port, NAT-friendly.

use std::collections::HashMap;
use std::io::{Read, Write};
use std::sync::Arc;
use std::time::Duration;

use base64::Engine;
use cm_sandbox::{DockerDriver, SandboxDriver, SandboxHandle, SandboxSpec};
use futures::{SinkExt, StreamExt};
use portable_pty::{native_pty_system, CommandBuilder, MasterPty, PtySize};
use serde_json::{json, Value};
use sysinfo::{Disks, System};
use tokio::sync::{mpsc, Mutex};
use tokio_tungstenite::tungstenite::Message;

mod rtc;

const B64: base64::engine::general_purpose::GeneralPurpose =
    base64::engine::general_purpose::STANDARD;

/// A live host-shell PTY the gateway opened (keyed by session id).
struct Pty {
    master: Box<dyn MasterPty + Send>,
    writer: Box<dyn Write + Send>,
    child: Box<dyn portable_pty::Child + Send + Sync>,
}
type Ptys = Arc<Mutex<HashMap<u64, Pty>>>;

const VERSION: &str = env!("CARGO_PKG_VERSION");

#[tokio::main]
async fn main() {
    // The dep graph enables both rustls crypto providers (tungstenite + bollard),
    // so rustls can't auto-pick — install one explicitly before any TLS.
    let _ = rustls::crypto::ring::default_provider().install_default();
    if std::env::args().any(|a| a == "--selftest") {
        selftest();
        return;
    }
    let (server, token, ts_authkey) = parse_args();
    if server.is_empty() || token.is_empty() {
        eprintln!("usage: clawmates-node --server <https://gateway> --token <token> [--tailscale-authkey <key>]");
        eprintln!("   (or set CLAWMATES_SERVER / CLAWMATES_TOKEN / CLAWMATES_TS_AUTHKEY)");
        std::process::exit(2);
    }
    tailscale_up(&ts_authkey);
    ensure_tmux();
    let ws_url = ws_url(&server, &token);
    println!("clawmates-node {VERSION} connecting to {server}");
    loop {
        if let Err(e) = run(&ws_url).await {
            eprintln!("channel ended: {e}; reconnecting in 5s");
        }
        tokio::time::sleep(Duration::from_secs(5)).await;
    }
}

fn parse_args() -> (String, String, String) {
    let mut server = String::new();
    let mut token = String::new();
    let mut ts_authkey = String::new();
    let mut args = std::env::args().skip(1);
    while let Some(a) = args.next() {
        match a.as_str() {
            "--server" => server = args.next().unwrap_or_default(),
            "--token" => token = args.next().unwrap_or_default(),
            "--tailscale-authkey" => ts_authkey = args.next().unwrap_or_default(),
            _ => {}
        }
    }
    if server.is_empty() {
        server = std::env::var("CLAWMATES_SERVER").unwrap_or_default();
    }
    if token.is_empty() {
        token = std::env::var("CLAWMATES_TOKEN").unwrap_or_default();
    }
    if ts_authkey.is_empty() {
        ts_authkey = std::env::var("CLAWMATES_TS_AUTHKEY").unwrap_or_default();
    }
    (server, token, ts_authkey)
}

fn ws_url(server: &str, token: &str) -> String {
    let base = server.trim_end_matches('/');
    let base = base
        .replacen("https://", "wss://", 1)
        .replacen("http://", "ws://", 1);
    let base = if base.starts_with("ws") {
        base
    } else {
        format!("wss://{base}")
    };
    format!("{base}/api/nodes/agent?token={token}")
}

async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
    let (ws, _) = tokio_tungstenite::connect_async(ws_url).await?;
    println!("connected; reporting health every 5s");
    let (mut write, mut read) = ws.split();
    // Everything outbound (heartbeats, command results, PTY output) funnels
    // through one channel so PTY reader threads can push asynchronously.
    let (out_tx, mut out_rx) = mpsc::unbounded_channel::<String>();
    let ptys: Ptys = Arc::new(Mutex::new(HashMap::new()));
    let peers: rtc::RtcPeers = Arc::new(Mutex::new(HashMap::new()));
    // Collect heartbeats on a dedicated thread: the metric helpers shell out to
    // docker/tailscale and stat disks (blocking), which must never stall the
    // async select loop (or heartbeats/pongs would starve during a slow op).
    let hb_tx = out_tx.clone();
    std::thread::spawn(move || {
        let mut sys = System::new_all();
        loop {
            if hb_tx.send(heartbeat(&mut sys)).is_err() {
                break; // connection gone
            }
            std::thread::sleep(Duration::from_secs(5));
        }
    });

    // Probe installed dev-tool versions (docker/claude/kimi-cli/ollama) on a
    // dedicated thread — the version commands shell out (blocking) — on connect,
    // then every 15 min. The server flags updates against upstream.
    let tools_tx = out_tx.clone();
    std::thread::spawn(move || loop {
        let frame = json!({ "t": "node_tools", "tools": probe_tools() }).to_string();
        if tools_tx.send(frame).is_err() {
            break;
        }
        std::thread::sleep(Duration::from_secs(900));
    });

    // Liveness has two independent failure modes to catch, both of which we
    // hit on architect during the Jul 5 2026 outage:
    //   (a) READ-side stall: server never sends anything (or the socket goes
    //       half-open on read). Caught by last_rx / 40s idle window below.
    //   (b) WRITE-side stall: peer's TCP stack has died but our OS buffer is
    //       still soaking heartbeat writes. write.send().await blocks
    //       indefinitely inside the select! branch — tokio::select doesn't
    //       preempt a running future, so the whole loop freezes; idle_tick
    //       never gets to fire. Wrapping the send in a timeout is the fix.
    //
    // WRITE_DEADLINE is short enough (10s) that a stuck send is caught before
    // it can outlast the 40s read-idle threshold and leave the daemon spinning
    // silently for hours (which is what happened pre-patch).
    const WRITE_DEADLINE: Duration = Duration::from_secs(10);
    let mut idle_tick = tokio::time::interval(Duration::from_secs(5));
    let mut last_rx = std::time::Instant::now();
    loop {
        tokio::select! {
            Some(frame) = out_rx.recv() => {
                match tokio::time::timeout(
                    WRITE_DEADLINE,
                    write.send(Message::Text(frame.into())),
                ).await {
                    Ok(Ok(())) => {}
                    Ok(Err(e)) => return Err(e.into()),
                    Err(_) => {
                        eprintln!("write.send timeout > {}s — socket dead, reconnecting",
                            WRITE_DEADLINE.as_secs());
                        return Ok(());
                    }
                }
            }
            _ = idle_tick.tick() => {
                if last_rx.elapsed() > Duration::from_secs(40) {
                    return Ok(());
                }
            }
            msg = read.next() => {
                last_rx = std::time::Instant::now();
                match msg {
                    Some(Ok(Message::Text(t))) => {
                        // Run each op concurrently so long docker/PTY work never
                        // blocks heartbeats, pongs, or other commands.
                        let out = out_tx.clone();
                        let ptys = ptys.clone();
                        let peers = peers.clone();
                        let text = t.to_string();
                        tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers).await; });
                    }
                    Some(Ok(Message::Ping(p))) => {
                        match tokio::time::timeout(WRITE_DEADLINE, write.send(Message::Pong(p))).await {
                            Ok(Ok(())) => {}
                            Ok(Err(e)) => return Err(e.into()),
                            Err(_) => {
                                eprintln!("pong write timeout — socket dead, reconnecting");
                                return Ok(());
                            }
                        }
                    }
                    Some(Ok(Message::Close(_))) | None => return Ok(()),
                    Some(Err(e)) => return Err(e.into()),
                    _ => {}
                }
            }
        }
    }
}

/// Collect a host-health snapshot and serialize the heartbeat frame.
fn heartbeat(sys: &mut System) -> String {
    sys.refresh_cpu_usage();
    sys.refresh_memory();
    let mem_total = sys.total_memory() as i64;
    let mem_used = sys.used_memory() as i64;
    let mem_pressure = if mem_total > 0 {
        mem_used as f64 / mem_total as f64
    } else {
        0.0
    };
    let load = System::load_average();
    let (disk_total, disk_free) = root_disk();
    json!({
        "t": "heartbeat",
        "version": VERSION,
        "tailscale_ip": tailscale_ip(),
        "hostname": System::host_name(),
        "local_ip": local_ip(),
        "health": {
            "cpu_pct": sys.global_cpu_usage() as f64,
            "mem_total": mem_total,
            "mem_used": mem_used,
            "mem_pressure": mem_pressure,
            "swap_used": sys.used_swap() as i64,
            "disk_total": disk_total,
            "disk_free": disk_free,
            "load1": load.one,
            "load5": load.five,
            "load15": load.fifteen,
            "container_count": docker_count(),
        }
    })
    .to_string()
}

/// Probe installed dev-tool versions: for each tool, find its binary across the
/// usual bin dirs and read `--version`. Returns `{ tool: "x.y.z", … }` for the
/// ones found. Probes `kimi-cli` (the real uv tool), not the `kimi` API wrapper.
fn probe_tools() -> Value {
    let home = std::env::var("HOME").unwrap_or_default();
    let dirs = [
        format!("{home}/.local/bin"),
        "/opt/homebrew/bin".to_string(),
        "/usr/local/bin".to_string(),
        "/usr/bin".to_string(),
        "/bin".to_string(),
        format!("{home}/.local/share/uv/tools/kimi-cli/bin"),
        format!("{home}/.cargo/bin"),
    ];
    let mut out = serde_json::Map::new();
    // (report key, binary name) — usually identical; Rust reports "rust" but its
    // binary is `rustc`.
    for (key, bin) in [
        ("docker", "docker"),
        ("claude", "claude"),
        ("kimi-cli", "kimi-cli"),
        ("ollama", "ollama"),
        ("rust", "rustc"),
    ] {
        for d in &dirs {
            let p = std::path::Path::new(d).join(bin);
            if p.exists() {
                if let Some(v) = tool_version(&p) {
                    out.insert(key.to_string(), Value::String(v));
                }
                break;
            }
        }
    }
    Value::Object(out)
}

/// Run `<bin> --version` with a 2s cap (off-thread, so a hung binary can't wedge
/// the prober) and extract the first semver from its output.
fn tool_version(bin: &std::path::Path) -> Option<String> {
    let bin = bin.to_path_buf();
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        let text = std::process::Command::new(&bin)
            .arg("--version")
            .output()
            .ok()
            .map(|o| {
                let mut s = String::from_utf8_lossy(&o.stdout).into_owned();
                s.push_str(&String::from_utf8_lossy(&o.stderr));
                s
            });
        let _ = tx.send(text);
    });
    // 8s, not 2s: right after an update the binary is freshly replaced, and macOS
    // Gatekeeper re-verifies an unsigned binary on first exec (several seconds) — a
    // tight cap would miss the new version on the post-update re-probe.
    let text = rx.recv_timeout(Duration::from_secs(8)).ok().flatten()?;
    extract_semver(&text)
}

/// First `\d+\.\d+(\.\d+)?` run in `s` (e.g. "29.1.3" from "Docker version
/// 29.1.3, build …"; stops before a trailing `-0ubuntu…`).
/// Run the FIXED update command for a tool (no arbitrary shell — the key maps to a
/// hardcoded command). Returns (success, combined output). Async so it never blocks
/// the read loop; ~170s cap.
async fn update_tool(tool: &str) -> (bool, String) {
    let home = std::env::var("HOME").unwrap_or_default();
    let dirs = [
        format!("{home}/.local/bin"),
        "/opt/homebrew/bin".to_string(),
        "/usr/local/bin".to_string(),
        "/usr/bin".to_string(),
        "/bin".to_string(),
        format!("{home}/.cargo/bin"),
    ];
    let find = |name: &str| {
        dirs.iter()
            .map(|d| std::path::Path::new(d).join(name))
            .find(|p| p.exists())
    };
    let mut cmd = match tool {
        "claude" | "glm" => match find("claude") {
            Some(p) => {
                let mut c = tokio::process::Command::new(p);
                c.arg("update");
                c
            }
            None => return (false, "claude not found".into()),
        },
        "kimi" => match find("uv") {
            Some(p) => {
                let mut c = tokio::process::Command::new(p);
                c.args(["tool", "upgrade", "kimi-cli"]);
                c
            }
            None => return (false, "uv not found".into()),
        },
        "ollama" => {
            if cfg!(target_os = "macos") {
                match find("brew") {
                    Some(p) => {
                        let mut c = tokio::process::Command::new(p);
                        c.args(["upgrade", "ollama"]);
                        c
                    }
                    None => return (false, "brew not found".into()),
                }
            } else {
                let mut c = tokio::process::Command::new("sh");
                c.args(["-c", "curl -fsSL https://ollama.com/install.sh | sh"]);
                c
            }
        }
        "rust" => match find("rustup") {
            Some(p) => {
                let mut c = tokio::process::Command::new(p);
                c.args(["update", "stable"]);
                c
            }
            None => return (false, "rustup not found".into()),
        },
        _ => return (false, "unsupported tool".into()),
    };
    match tokio::time::timeout(Duration::from_secs(170), cmd.output()).await {
        Ok(Ok(o)) => {
            let mut s = String::from_utf8_lossy(&o.stdout).into_owned();
            s.push_str(&String::from_utf8_lossy(&o.stderr));
            if s.len() > 4096 {
                s.truncate(4096);
            }
            (o.status.success(), s)
        }
        Ok(Err(e)) => (false, format!("spawn error: {e}")),
        Err(_) => (false, "update timed out (170s)".into()),
    }
}

fn extract_semver(s: &str) -> Option<String> {
    let c: Vec<char> = s.chars().collect();
    let mut i = 0;
    while i < c.len() {
        if c[i].is_ascii_digit() {
            let start = i;
            while i < c.len() && (c[i].is_ascii_digit() || c[i] == '.') {
                i += 1;
            }
            let mut end = i;
            while end > start && c[end - 1] == '.' {
                end -= 1;
            }
            let cand: String = c[start..end].iter().collect();
            let parts: Vec<&str> = cand.split('.').collect();
            if parts.len() >= 2 && parts.iter().all(|p| !p.is_empty()) {
                return Some(cand);
            }
        } else {
            i += 1;
        }
    }
    None
}

/// Total + available bytes of the filesystem backing `/` (largest disk as a
/// fallback).
fn root_disk() -> (i64, i64) {
    let disks = Disks::new_with_refreshed_list();
    let mut best: Option<(i64, i64)> = None;
    for d in &disks {
        let total = d.total_space() as i64;
        let free = d.available_space() as i64;
        if d.mount_point().as_os_str() == "/" {
            return (total, free);
        }
        if best.map(|(t, _)| total > t).unwrap_or(true) {
            best = Some((total, free));
        }
    }
    best.unwrap_or((0, 0))
}

/// Number of running Docker containers (0 if Docker is absent).
fn docker_count() -> i32 {
    std::process::Command::new("docker")
        .args(["ps", "-q"])
        .output()
        .ok()
        .filter(|o| o.status.success())
        .map(|o| String::from_utf8_lossy(&o.stdout).lines().count() as i32)
        .unwrap_or(0)
}

/// The node's primary outbound IPv4 (the interface a default route would use).
/// Uses a connectionless UDP socket — no packet is actually sent.
fn local_ip() -> Option<String> {
    let sock = std::net::UdpSocket::bind("0.0.0.0:0").ok()?;
    sock.connect("8.8.8.8:80").ok()?;
    Some(sock.local_addr().ok()?.ip().to_string())
}

/// This node's Tailscale IP, if Tailscale is up (BYO tailnet).
fn tailscale_ip() -> Option<String> {
    let out = std::process::Command::new("tailscale")
        .args(["ip", "-4"])
        .output()
        .ok()?;
    if !out.status.success() {
        return None;
    }
    let ip = String::from_utf8_lossy(&out.stdout).trim().to_owned();
    (!ip.is_empty()).then_some(ip)
}

/// Handle a typed frame from the gateway. The gateway never sends arbitrary
/// shell — only vetted ops (verify, and an interactive host terminal the user
/// explicitly opened), so the host attack surface stays minimal.
async fn handle_frame(
    text: &str,
    out: &mpsc::UnboundedSender<String>,
    ptys: &Ptys,
    peers: &rtc::RtcPeers,
) {
    let Ok(v) = serde_json::from_str::<Value>(text) else {
        return;
    };
    match v.get("t").and_then(Value::as_str).unwrap_or_default() {
        "verify" => {
            if let Some(id) = v.get("id").and_then(Value::as_u64) {
                let (ok, output) = verify().await;
                let _ = out.send(
                    json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
                );
            }
        }
        "sb_check" => {
            if let Some(id) = v.get("id").and_then(Value::as_u64) {
                let (ok, output) = sandbox_check().await;
                let _ = out.send(
                    json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
                );
            }
        }
        // One-click dev-tool update. Runs a FIXED per-tool command (no arbitrary
        // shell), then re-probes so the new version reports. Spawned so the ~170s
        // command never stalls the read loop (heartbeats keep flowing).
        "tool_update" => {
            if let Some(id) = v.get("id").and_then(Value::as_u64) {
                let tool = v
                    .get("tool")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_owned();
                let out = out.clone();
                tokio::spawn(async move {
                    let (ok, output) = update_tool(&tool).await;
                    let _ = out.send(
                        json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
                    );
                    let tools = tokio::task::spawn_blocking(probe_tools)
                        .await
                        .unwrap_or_else(|_| json!({}));
                    let _ = out.send(json!({ "t": "node_tools", "tools": tools }).to_string());
                });
            }
        }
        // Herdr dispatch ops. Server sends `herdr_dispatch` to open a
        // sibling pane on the node's Herdr session and start the requested
        // CLI (claude / codex / kimi / etc.) with a prompt. `herdr_status`
        // polls that pane's agent_status; `herdr_read` scrapes its recent
        // transcript. Node just shells out to the `herdr` binary — the
        // Herdr background daemon is expected to already be running.
        op @ ("herdr_dispatch" | "herdr_status" | "herdr_read") => {
            if let Some(id) = v.get("id").and_then(Value::as_u64) {
                let (ok, output) = herdr_op(op, &v).await;
                let _ = out.send(
                    json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
                );
            }
        }
        // Agent-sandbox container ops: drive the REAL DockerDriver so the
        // hardening (cap-drop ALL, seccomp, no-net, read-only, non-root) is
        // byte-identical to the gateway's local sandboxes.
        op @ ("sb_provision" | "sb_exec" | "sb_destroy" | "sb_health" | "sb_list") => {
            if let Some(id) = v.get("id").and_then(Value::as_u64) {
                let (ok, output) = sb_op(op, &v).await;
                let _ = out.send(
                    json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
                );
            }
        }
        "pty_open" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            let cols = v.get("cols").and_then(Value::as_u64).unwrap_or(80) as u16;
            let rows = v.get("rows").and_then(Value::as_u64).unwrap_or(24) as u16;
            eprintln!("[pty] received pty_open sid={sid}");
            if let Err(e) = open_pty(
                sid,
                cols,
                rows,
                PtyTarget::from_frame(&v),
                out.clone(),
                ptys.clone(),
            )
            .await
            {
                eprintln!("[pty] open_pty FAILED sid={sid}: {e}");
                let _ = out.send(json!({ "t": "pty_out", "sid": sid, "data": B64.encode(format!("\r\n\x1b[31m[clawmates] could not start shell: {e}\x1b[0m\r\n").as_bytes()) }).to_string());
                let _ = out.send(json!({ "t": "pty_exit", "sid": sid, "error": e }).to_string());
            }
        }
        "pty_in" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            if let Some(bytes) = v
                .get("data")
                .and_then(Value::as_str)
                .and_then(|d| B64.decode(d).ok())
            {
                if let Some(p) = ptys.lock().await.get_mut(&sid) {
                    let _ = p.writer.write_all(&bytes);
                    let _ = p.writer.flush();
                }
            }
        }
        "pty_resize" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            let cols = v.get("cols").and_then(Value::as_u64).unwrap_or(80) as u16;
            let rows = v.get("rows").and_then(Value::as_u64).unwrap_or(24) as u16;
            if let Some(p) = ptys.lock().await.get(&sid) {
                let _ = p.master.resize(PtySize {
                    rows,
                    cols,
                    pixel_width: 0,
                    pixel_height: 0,
                });
            }
        }
        "pty_close" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            if let Some(mut p) = ptys.lock().await.remove(&sid) {
                let _ = p.child.kill();
            }
        }
        // WebRTC signaling (browser is offerer). The DataChannel, once open,
        // carries terminal I/O directly peer-to-peer; resize/close still arrive
        // as pty_resize/pty_close keyed by the same sid.
        "webrtc_offer" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            let sdp = v
                .get("sdp")
                .and_then(Value::as_str)
                .unwrap_or("")
                .to_owned();
            rtc::handle_offer(
                sid,
                sdp,
                PtyTarget::from_frame(&v),
                out.clone(),
                ptys.clone(),
                peers.clone(),
            )
            .await;
        }
        "webrtc_ice" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            rtc::handle_ice(sid, &v, peers).await;
        }
        "webrtc_close" => {
            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
            rtc::handle_close(sid, peers, ptys).await;
        }
        _ => {}
    }
}

/// (master for resize, cloned reader, writer, child) — one PTY spawn reused by
/// both the WS-relay path and the WebRTC DataChannel path.
type PtyParts = (
    Box<dyn MasterPty + Send>,
    Box<dyn Read + Send>,
    Box<dyn Write + Send>,
    Box<dyn portable_pty::Child + Send + Sync>,
);

/// Spawn `cmd` in a fresh PTY, returning handles for resize/read/write/child.
fn spawn_pty(cmd: CommandBuilder, cols: u16, rows: u16) -> Result<PtyParts, String> {
    let pair = native_pty_system()
        .openpty(PtySize {
            rows,
            cols,
            pixel_width: 0,
            pixel_height: 0,
        })
        .map_err(|e| e.to_string())?;
    let child = pair.slave.spawn_command(cmd).map_err(|e| e.to_string())?;
    drop(pair.slave);
    let reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
    let writer = pair.master.take_writer().map_err(|e| e.to_string())?;
    Ok((pair.master, reader, writer, child))
}

/// Spawn a host login shell (tmux `-L clawmates`) in a fresh PTY.
fn spawn_terminal_pty(cols: u16, rows: u16) -> Result<PtyParts, String> {
    spawn_pty(terminal_command(), cols, rows)
}

/// `docker exec -it` a tmux session inside an agent's container — the node-placed
/// agent terminal, which shares the container's node-local `~/drives`.
fn spawn_container_pty(
    container: &str,
    session: &str,
    cols: u16,
    rows: u16,
) -> Result<PtyParts, String> {
    let mut c = CommandBuilder::new("docker");
    for a in [
        "exec",
        "-it",
        container,
        "tmux",
        "new-session",
        "-A",
        "-s",
        session,
    ] {
        c.arg(a);
    }
    spawn_pty(c, cols, rows)
}

/// Where a session's PTY runs: the node's host shell, or `docker exec` into a
/// specific agent container on this node.
pub(crate) enum PtyTarget {
    Host,
    Container { container: String, session: String },
    /// Custom argv (Herdr Live Pane uses this to spawn `herdr` directly
    /// so the browser xterm attaches straight into the node's Herdr TUI
    /// instead of a login shell).
    Command { argv: Vec<String> },
}

impl PtyTarget {
    /// Parse from a control frame. Precedence: explicit `command` (non-
    /// empty array) → Command; else `container` → Container; else Host.
    pub(crate) fn from_frame(v: &Value) -> Self {
        if let Some(argv) = v.get("command").and_then(Value::as_array) {
            let parts: Vec<String> = argv
                .iter()
                .filter_map(|x| x.as_str().map(str::to_owned))
                .collect();
            if !parts.is_empty() {
                return PtyTarget::Command { argv: parts };
            }
        }
        match v.get("container").and_then(Value::as_str) {
            Some(c) if !c.is_empty() => PtyTarget::Container {
                container: c.to_owned(),
                session: v
                    .get("session")
                    .and_then(Value::as_str)
                    .unwrap_or("main")
                    .to_owned(),
            },
            _ => PtyTarget::Host,
        }
    }

    pub(crate) fn spawn(&self, cols: u16, rows: u16) -> Result<PtyParts, String> {
        match self {
            PtyTarget::Host => spawn_terminal_pty(cols, rows),
            PtyTarget::Container { container, session } => {
                spawn_container_pty(container, session, cols, rows)
            }
            PtyTarget::Command { argv } => spawn_command_pty(argv, cols, rows),
        }
    }
}

/// Spawn an arbitrary command in a PTY. Argv[0] must be the program;
/// if it's a bare name (no slash), it's resolved via the process PATH.
/// Missing binary returns a clean error the client sees as a banner.
fn spawn_command_pty(argv: &[String], cols: u16, rows: u16) -> Result<PtyParts, String> {
    let program = argv
        .first()
        .ok_or_else(|| "command argv is empty".to_string())?;
    // Resolve bare names against common bin dirs so a headless daemon
    // (no login shell / no PATH set for herdr install dir) still finds it.
    let resolved = if program.contains('/') {
        program.clone()
    } else {
        let home = std::env::var("HOME").unwrap_or_default();
        let candidates = [
            format!("{home}/.local/bin/{program}"),
            format!("/opt/homebrew/bin/{program}"),
            format!("/usr/local/bin/{program}"),
            format!("/usr/bin/{program}"),
        ];
        candidates
            .into_iter()
            .find(|p| std::path::Path::new(p).exists())
            .unwrap_or_else(|| program.clone())
    };
    let mut c = CommandBuilder::new(&resolved);
    for a in argv.iter().skip(1) {
        c.arg(a);
    }
    spawn_pty(c, cols, rows)
}

/// Spawn a host login shell in a PTY; stream its output back as pty_out frames.
async fn open_pty(
    sid: u64,
    cols: u16,
    rows: u16,
    target: PtyTarget,
    out: mpsc::UnboundedSender<String>,
    ptys: Ptys,
) -> Result<(), String> {
    let (master, mut reader, writer, child) = target.spawn(cols, rows)?;
    ptys.lock().await.insert(
        sid,
        Pty {
            master,
            writer,
            child,
        },
    );
    let label = match &target {
        PtyTarget::Host => format!(
            "host shell ({})",
            if has_tmux() { "tmux" } else { "login shell" }
        ),
        PtyTarget::Container { container, .. } => format!("container {container}"),
        PtyTarget::Command { argv } => format!("command {}", argv.join(" ")),
    };
    eprintln!("[pty] open sid={sid} cols={cols} rows={rows} target={label}");
    // Immediate banner over the channel: if the browser shows this but no shell,
    // the relay works and the shell is the problem (vs. a dead relay → nothing).
    let host = System::host_name().unwrap_or_else(|| "node".into());
    let banner = format!("\r\n\x1b[2m[clawmates] {label} on {host} — starting…\x1b[0m\r\n");
    let _ = out.send(
        json!({ "t": "pty_out", "sid": sid, "data": B64.encode(banner.as_bytes()) }).to_string(),
    );
    // Blocking PTY reads on a thread → base64 pty_out frames into the out channel.
    std::thread::spawn(move || {
        let mut buf = [0u8; 8192];
        let mut total = 0usize;
        loop {
            match reader.read(&mut buf) {
                Ok(0) => {
                    eprintln!("[pty] sid={sid} EOF after {total} bytes (shell exited)");
                    break;
                }
                Err(e) => {
                    eprintln!("[pty] sid={sid} read error after {total} bytes: {e}");
                    break;
                }
                Ok(n) => {
                    if total == 0 {
                        eprintln!("[pty] sid={sid} first read: {n} bytes");
                    }
                    total += n;
                    let data = B64.encode(&buf[..n]);
                    if out
                        .send(json!({ "t": "pty_out", "sid": sid, "data": data }).to_string())
                        .is_err()
                    {
                        eprintln!("[pty] sid={sid} out channel closed");
                        break;
                    }
                }
            }
        }
        eprintln!("[pty] sid={sid} reader done, {total} bytes total");
        let _ = out.send(json!({ "t": "pty_exit", "sid": sid }).to_string());
    });
    Ok(())
}

/// The daemon's built-in host check (fixed command — nothing caller-supplied
/// executes): kernel info + Docker presence.
async fn verify() -> (bool, String) {
    let out = tokio::process::Command::new("sh")
        .arg("-lc")
        .arg("uname -a; echo '---'; docker version --format 'docker {{.Server.Version}}' 2>/dev/null || echo 'docker: not found'")
        .output()
        .await;
    match out {
        Ok(o) => {
            let mut s = String::from_utf8_lossy(&o.stdout).into_owned();
            if !o.stderr.is_empty() {
                s.push_str(&String::from_utf8_lossy(&o.stderr));
            }
            (true, s.trim().to_owned())
        }
        Err(e) => (false, format!("verify error: {e}")),
    }
}

/// Readiness check: provision a fully locked-down throwaway container (the same
/// hardening agent sandboxes use — cap-drop ALL, no-new-privileges, no network,
/// read-only rootfs, non-root, resource caps), run it, and tear it down. Proves
/// the node can host hardened agent workloads. Fixed command — nothing
/// caller-supplied runs.
async fn sandbox_check() -> (bool, String) {
    let _ = tokio::process::Command::new("docker")
        .args(["pull", "-q", "alpine:latest"])
        .output()
        .await;
    let out = tokio::process::Command::new("docker")
        .args([
            "run",
            "--rm",
            "--cap-drop=ALL",
            "--security-opt",
            "no-new-privileges",
            "--network",
            "none",
            "--read-only",
            "--tmpfs",
            "/tmp",
            "--user",
            "10001:10001",
            "--memory",
            "256m",
            "--pids-limit",
            "128",
            "alpine:latest",
            "sh",
            "-c",
            "echo sandbox-ok; id; uname -sm",
        ])
        .output()
        .await;
    match out {
        Ok(o) if o.status.success() => (true, String::from_utf8_lossy(&o.stdout).trim().to_owned()),
        Ok(o) => {
            let mut s = String::from_utf8_lossy(&o.stdout).into_owned();
            s.push_str(&String::from_utf8_lossy(&o.stderr));
            (false, s.trim().to_owned())
        }
        Err(e) => (false, format!("docker error: {e} — is Docker installed?")),
    }
}

fn handle_of(v: &Value) -> SandboxHandle {
    SandboxHandle {
        id: v
            .get("cid")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
        name: v
            .get("cname")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
    }
}

/// Run an agent-sandbox container op via the local DockerDriver (full hardening),
/// returning the result payload as JSON or an error message.
/// Herdr control ops. Requires `herdr` in PATH and a running background
/// session (Phase 0 install). Returns raw JSON strings from the herdr
/// CLI so the server can parse pane_id / agent_status without a
/// second RPC hop.
///
/// Ops:
///   herdr_dispatch { mission_id, cli, prompt, direction? } →
///       runs `herdr pane split ... && herdr pane run ... "prompt"`
///       output = the split's JSON response so the server can extract
///       result.pane.pane_id
///   herdr_status { pane_id } → `herdr pane get <pane_id>` JSON
///   herdr_read   { pane_id, lines? } → recent-unwrapped scrollback
async fn herdr_op(op: &str, v: &Value) -> (bool, String) {
    let herdr = match std::env::var("HOME").ok().and_then(|h| {
        [
            format!("{h}/.local/bin/herdr"),
            "/opt/homebrew/bin/herdr".to_string(),
            "/usr/local/bin/herdr".to_string(),
        ]
        .into_iter()
        .find(|p| std::path::Path::new(p).exists())
    }) {
        Some(p) => p,
        None => return (false, "herdr binary not found on PATH".into()),
    };
    let herdr = std::sync::Arc::new(herdr);
    let run = move |args: Vec<String>| {
        let herdr = herdr.clone();
        async move {
            let fut = tokio::process::Command::new(herdr.as_str())
                .args(&args)
                .output();
            match tokio::time::timeout(Duration::from_secs(30), fut).await {
                Ok(Ok(o)) => {
                    let mut s = String::from_utf8_lossy(&o.stdout).into_owned();
                    if !o.status.success() {
                        s.push_str(&String::from_utf8_lossy(&o.stderr));
                    }
                    (o.status.success(), s)
                }
                Ok(Err(e)) => (false, format!("spawn error: {e}")),
                Err(_) => (false, "herdr op timed out".into()),
            }
        }
    };
    match op {
        "herdr_dispatch" => {
            let mission_id = v
                .get("mission_id")
                .and_then(Value::as_str)
                .unwrap_or("unknown");
            let cli = v.get("cli").and_then(Value::as_str).unwrap_or("claude");
            let prompt = v.get("prompt").and_then(Value::as_str).unwrap_or("");
            let direction = v
                .get("direction")
                .and_then(Value::as_str)
                .unwrap_or("right");
            // 1. Ensure a mission workspace exists (idempotent — label collision falls through).
            let _ = run(vec![
                "workspace".into(),
                "create".into(),
                "--label".into(),
                format!("mission-{mission_id}"),
            ])
            .await;
            // 2. Split off a fresh pane in that workspace and read its pane_id.
            let (ok, split_out) = run(vec![
                "pane".into(),
                "split".into(),
                "--direction".into(),
                direction.into(),
                "--no-focus".into(),
            ])
            .await;
            if !ok {
                return (false, format!("split failed: {split_out}"));
            }
            let pane_id = match serde_json::from_str::<Value>(&split_out) {
                Ok(j) => j
                    .pointer("/result/pane/pane_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                Err(_) => String::new(),
            };
            if pane_id.is_empty() {
                return (false, format!("no pane_id in split response: {split_out}"));
            }
            // 3. Rename for operator readability.
            let _ = run(vec![
                "pane".into(),
                "rename".into(),
                pane_id.clone(),
                format!("mission-{mission_id}"),
            ])
            .await;
            // 4. Launch the CLI with the prompt inline.
            let launch = if prompt.is_empty() {
                cli.to_string()
            } else {
                // Single-quoted so shell metacharacters in the prompt don't
                // reinterpret. Herdr's pane.run sends this verbatim to the shell.
                let escaped = prompt.replace('\'', "'\\''");
                format!("{cli} '{escaped}'")
            };
            let (rok, rout) = run(vec![
                "pane".into(),
                "run".into(),
                pane_id.clone(),
                launch,
            ])
            .await;
            let payload = serde_json::json!({
                "pane_id": pane_id,
                "split": split_out,
                "run_output": rout,
                "run_ok": rok,
            });
            (rok, payload.to_string())
        }
        "herdr_status" => {
            let pane = v.get("pane_id").and_then(Value::as_str).unwrap_or_default();
            if pane.is_empty() {
                return (false, "pane_id required".into());
            }
            run(vec!["pane".into(), "get".into(), pane.to_string()]).await
        }
        "herdr_read" => {
            let pane = v.get("pane_id").and_then(Value::as_str).unwrap_or_default();
            let lines = v.get("lines").and_then(Value::as_u64).unwrap_or(200);
            if pane.is_empty() {
                return (false, "pane_id required".into());
            }
            run(vec![
                "pane".into(),
                "read".into(),
                pane.to_string(),
                "--source".into(),
                "recent-unwrapped".into(),
                "--lines".into(),
                lines.to_string(),
            ])
            .await
        }
        _ => (false, format!("unknown herdr op {op}")),
    }
}

async fn sb_op(op: &str, v: &Value) -> (bool, String) {
    let driver = match DockerDriver::connect() {
        Ok(d) => d,
        Err(e) => return (false, format!("docker unavailable: {e}")),
    };
    match op {
        "sb_provision" => {
            let spec: SandboxSpec = match v
                .get("spec")
                .and_then(|s| serde_json::from_value(s.clone()).ok())
            {
                Some(s) => s,
                None => return (false, "invalid spec".to_owned()),
            };
            // Pull the agent image if the node doesn't have it yet (best effort).
            let _ = tokio::process::Command::new("docker")
                .args(["pull", "-q", &spec.image])
                .output()
                .await;
            match driver.provision(&spec).await {
                Ok(h) => (true, json!({ "id": h.id, "name": h.name }).to_string()),
                Err(e) => (false, format!("provision: {e}")),
            }
        }
        "sb_exec" => {
            let handle = handle_of(v);
            let cmd: Vec<String> = v
                .get("cmd")
                .and_then(Value::as_array)
                .map(|a| {
                    a.iter()
                        .filter_map(|x| x.as_str().map(str::to_owned))
                        .collect()
                })
                .unwrap_or_default();
            let refs: Vec<&str> = cmd.iter().map(String::as_str).collect();
            match driver.exec(&handle, &refs).await {
                Ok(r) => (
                    r.exit_code == 0,
                    json!({ "exit": r.exit_code, "stdout": r.stdout, "stderr": r.stderr })
                        .to_string(),
                ),
                Err(e) => (false, format!("exec: {e}")),
            }
        }
        "sb_destroy" => match driver.destroy(&handle_of(v)).await {
            Ok(()) => (true, "ok".to_owned()),
            Err(e) => (false, format!("destroy: {e}")),
        },
        "sb_health" => match driver.health(&handle_of(v)).await {
            Ok(alive) => (true, json!({ "alive": alive }).to_string()),
            Err(e) => (false, format!("health: {e}")),
        },
        "sb_list" => {
            let kind = v.get("kind").and_then(Value::as_str).unwrap_or("agent");
            match driver.list_managed(kind).await {
                Ok(list) => (
                    true,
                    json!(list
                        .iter()
                        .map(|m| json!({ "id": m.id, "created_unix": m.created_unix }))
                        .collect::<Vec<_>>())
                    .to_string(),
                ),
                Err(e) => (false, format!("list: {e}")),
            }
        }
        _ => (false, "unknown op".to_owned()),
    }
}

/// Best-effort: join the user's tailnet (BYO Tailscale) and enable Tailscale SSH
/// so they can reach this node keylessly. ONLY when an auth key is explicitly
/// passed — enabling SSH unprompted can drop the user's current SSH session.
/// `--accept-risk` avoids the interactive abort when they're on Tailscale.
/// Non-fatal: the WSS control channel works regardless of Tailscale.
fn tailscale_up(authkey: &str) {
    if authkey.is_empty() {
        return;
    }
    let _ = std::process::Command::new("tailscale")
        .args([
            "up",
            "--authkey",
            authkey,
            "--ssh",
            "--accept-routes",
            "--accept-risk=lose-ssh",
        ])
        .status();
}

/// The command a host terminal runs: a resumable host tmux session (like the
/// agent terminal — `-A` attaches-or-creates "clawmates" and tmux redraws on
/// attach so the view is never blank), falling back to a login shell.
fn terminal_command() -> CommandBuilder {
    let mut cmd = if has_tmux() {
        let mut c = CommandBuilder::new("tmux");
        // A DEDICATED server socket (-L) so we never collide with — or get refused
        // by — the operator's own tmux. Without this, running the daemon inside a
        // tmux makes the spawned tmux refuse to nest and exit instantly (writing
        // its warning to stderr, not the PTY → the browser sees nothing).
        c.arg("-L");
        c.arg("clawmates");
        c.arg("new-session");
        c.arg("-A");
        c.arg("-s");
        c.arg("main");
        c
    } else {
        let shell = std::env::var("SHELL").unwrap_or_else(|_| "/bin/bash".to_owned());
        let mut c = CommandBuilder::new(shell);
        c.arg("-l");
        c
    };
    cmd.env("TERM", "xterm-256color");
    // Clear any inherited $TMUX so the new tmux doesn't think it's nested.
    cmd.env_remove("TMUX");
    if let Ok(home) = std::env::var("HOME") {
        cmd.cwd(home);
    }
    cmd
}

/// `--selftest`: open the host terminal PTY locally and print ~2.5s of its raw
/// output (diagnostic — confirms tmux/shell actually draws).
fn selftest() {
    let pair = match native_pty_system().openpty(PtySize {
        rows: 40,
        cols: 120,
        pixel_width: 0,
        pixel_height: 0,
    }) {
        Ok(p) => p,
        Err(e) => {
            eprintln!("openpty failed: {e}");
            return;
        }
    };
    let mut child = match pair.slave.spawn_command(terminal_command()) {
        Ok(c) => c,
        Err(e) => {
            eprintln!("spawn failed: {e}");
            return;
        }
    };
    drop(pair.slave);
    let mut reader = pair.master.try_clone_reader().expect("reader");
    let (tx, rx) = std::sync::mpsc::channel::<Vec<u8>>();
    std::thread::spawn(move || {
        let mut buf = [0u8; 8192];
        while let Ok(n) = reader.read(&mut buf) {
            if n == 0 || tx.send(buf[..n].to_vec()).is_err() {
                break;
            }
        }
    });
    let mut total = 0usize;
    let start = std::time::Instant::now();
    while start.elapsed() < std::time::Duration::from_millis(2500) {
        if let Ok(bytes) = rx.recv_timeout(std::time::Duration::from_millis(300)) {
            total += bytes.len();
            print!("{}", String::from_utf8_lossy(&bytes));
        }
    }
    let _ = child.kill();
    eprintln!(
        "\n--- selftest: tmux={}, {} bytes of output ---",
        has_tmux(),
        total
    );
}

/// Is tmux on PATH?
fn has_tmux() -> bool {
    std::process::Command::new("tmux")
        .arg("-V")
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false)
}

/// Best-effort: install tmux via the host package manager (works when the daemon
/// runs as root, e.g. systemd). Non-interactive; if it can't, host terminals fall
/// back to a plain login shell.
fn ensure_tmux() {
    if has_tmux() {
        return;
    }
    let mgrs: &[(&str, &[&str])] = &[
        ("apt-get", &["install", "-y", "tmux"]),
        ("dnf", &["install", "-y", "tmux"]),
        ("yum", &["install", "-y", "tmux"]),
        ("apk", &["add", "--no-cache", "tmux"]),
        ("pacman", &["-S", "--noconfirm", "tmux"]),
        ("brew", &["install", "tmux"]),
    ];
    for (mgr, args) in mgrs {
        let present = std::process::Command::new("sh")
            .arg("-c")
            .arg(format!("command -v {mgr}"))
            .output()
            .map(|o| o.status.success())
            .unwrap_or(false);
        if !present {
            continue;
        }
        if *mgr == "apt-get" {
            let _ = std::process::Command::new("apt-get")
                .arg("update")
                .env("DEBIAN_FRONTEND", "noninteractive")
                .output();
        }
        let _ = std::process::Command::new(mgr)
            .args(*args)
            .env("DEBIAN_FRONTEND", "noninteractive")
            .output();
        break;
    }
    if has_tmux() {
        println!("tmux ready for host terminals");
    } else {
        eprintln!("tmux not found (auto-install unavailable) — host terminal will use a plain shell; `apt install tmux` for resumable sessions");
    }
}