//! Run mission workloads in Firecracker microVMs on this node.
//!
//! The execution model is the one copy mode already proved for containers —
//! **inject → run → collect → destroy** — with a VM boundary instead of a
//! namespace boundary. Nothing on the host is shared with the guest: files go
//! in as a tar, come back as a tar, and the guest's filesystem dies with it.
//!
//! # Why not `SandboxDriver`
//!
//! The obvious move is to implement `cm_sandbox::SandboxDriver` so the existing
//! `RemoteDriver` marshals these ops over the hub for free. That trait is
//! container-shaped: `attach_pty`, `resize_pty`, argv `exec`. Missions need
//! create/inject/run/collect/destroy, so conforming would mean building
//! PTY-over-vsock and window-resize semantics that no mission path calls,
//! purely to satisfy a signature. These ops ride the node's ordinary frame
//! dispatch instead, which needs no protocol work.
//!
//! # Control plane
//!
//! Length-prefixed JSON over **vsock**, never the serial console. Feeding a
//! guest over stdin races its startup and arrives half-consumed — observed in
//! the spike as `# ho FC-GUEST-ALIVE`, the first two characters eaten. The
//! console stays a log; vsock is the channel.
//!
//! # What runs unprivileged
//!
//! Everything here. `/dev/kvm` is `crw-rw---- root:kvm` and the daemon user is
//! in the `kvm` group, so no `sudo` is needed to start a VM. The one privileged
//! step — baking the guest agent into the shared rootfs, which needs a loop
//! mount — happens once at setup time in `scripts/fc-node-setup.sh`.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixStream;
use tokio::sync::Mutex;

/// Where `fc-node-setup.sh` stages the kernel, the golden rootfs, and per-VM
/// working directories.
fn work_root() -> PathBuf {
    PathBuf::from(std::env::var("CLAWMATES_FC_ROOT").unwrap_or_else(|_| "/opt/clawmates-fc".into()))
}

/// A VM this node started and is responsible for destroying.
pub struct Vm {
    /// Process group, not pid. Firecracker is spawned via `setsid` so the whole
    /// group can be killed at once: a VM that spawned helpers must not be able
    /// to leave one behind holding the workdir open.
    pgid: i32,
    workdir: PathBuf,
    uds: PathBuf,
    /// This VM's egress proxy (see [`crate::egress`]) and its socket. Held so
    /// destroy can abort the task: a listener outliving its VM would accept a
    /// connection from the NEXT VM to reuse the path and proxy it under the dead
    /// one's identity.
    egress: Option<tokio::task::JoinHandle<()>>,
    egress_uds: PathBuf,
    /// The local-model listener, held for the SAME reason as `egress`: a
    /// listener that outlives its VM would accept a connection from the next VM
    /// to reuse the path and serve it under the dead one's identity.
    model: Option<tokio::task::JoinHandle<()>>,
    /// Present only for a backend served by a model on this node. Removed on
    /// destroy alongside the egress socket — a leaked unix socket is the same
    /// class of host litter the TAP approach was rejected for.
    model_uds: Option<PathBuf>,
}

pub type Vms = Arc<Mutex<HashMap<String, Vm>>>;

pub fn new_vms() -> Vms {
    Arc::new(Mutex::new(HashMap::new()))
}

/// A vm id is used to build a filesystem path, so it must not be able to
/// describe one. Rejecting rather than sanitising: a caller that sent
/// `../../etc` wanted something we should not guess at.
fn check_id(vm_id: &str) -> Result<(), String> {
    if vm_id.is_empty() || vm_id.len() > 64 {
        return Err("vm id must be 1..=64 chars".into());
    }
    if !vm_id
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
    {
        return Err(format!("vm id {vm_id:?} may only contain [A-Za-z0-9_-]"));
    }
    Ok(())
}

/// One request, one reply, framed as a 4-byte big-endian length plus JSON.
///
/// The length prefix is the point: a reply larger than a socket buffer arrives
/// in pieces, and reading "whatever was available" would parse a truncated
/// object as a complete one. Claude's `stream-json` output makes that routine
/// rather than theoretical.
/// Largest reply frame accepted from a guest, for both one-shot rpc and tail.
const MAX_REPLY: u32 = 256 * 1024 * 1024;

async fn rpc(uds: &Path, req: &Value) -> Result<Value, String> {
    let mut s = UnixStream::connect(uds)
        .await
        .map_err(|e| format!("connect {}: {e}", uds.display()))?;

    // Firecracker's host-side vsock multiplexes ports over one UDS: send
    // `CONNECT <port>` and it replies `OK <assigned>` before any payload.
    s.write_all(b"CONNECT 9001\n")
        .await
        .map_err(|e| format!("vsock CONNECT: {e}"))?;
    let mut ack = [0u8; 64];
    let n = s
        .read(&mut ack)
        .await
        .map_err(|e| format!("vsock CONNECT ack: {e}"))?;
    let ack = String::from_utf8_lossy(&ack[..n]);
    if !ack.starts_with("OK") {
        return Err(format!("vsock refused the connection: {}", ack.trim()));
    }

    let body = serde_json::to_vec(req).map_err(|e| format!("encode request: {e}"))?;
    s.write_all(&(body.len() as u32).to_be_bytes())
        .await
        .map_err(|e| format!("write length: {e}"))?;
    s.write_all(&body)
        .await
        .map_err(|e| format!("write body: {e}"))?;

    let mut len = [0u8; 4];
    s.read_exact(&mut len)
        .await
        .map_err(|e| format!("read reply length: {e}"))?;
    let len = u32::from_be_bytes(len);
    if len > MAX_REPLY {
        return Err(format!("reply of {len} bytes exceeds the {MAX_REPLY} cap"));
    }
    let mut buf = vec![0u8; len as usize];
    s.read_exact(&mut buf)
        .await
        .map_err(|e| format!("read reply body: {e}"))?;
    serde_json::from_slice(&buf).map_err(|e| format!("decode reply: {e}"))
}

/// Follow a file inside a guest, handing each chunk to `on_chunk` as it arrives.
///
/// Unlike [`rpc`], which is one request and one reply, this keeps the connection
/// open and reads MANY framed replies — the guest's `tail` op emits one per
/// chunk and a final `eof`. That is what makes a turn's output visible while the
/// turn is still running, and it works only because the guest agent now accepts
/// concurrent connections.
///
/// Returns the byte offset reached, so a caller that reconnects resumes instead
/// of replaying.
pub async fn tail_into<F>(
    vms: &Vms,
    vm_id: &str,
    path: &str,
    from: u64,
    mut on_chunk: F,
) -> Result<u64, String>
where
    F: FnMut(u64, String),
{
    let uds = uds_of(vms, vm_id).await?;
    let mut s = UnixStream::connect(&uds)
        .await
        .map_err(|e| format!("connect {}: {e}", uds.display()))?;
    s.write_all(b"CONNECT 9001\n")
        .await
        .map_err(|e| format!("vsock CONNECT: {e}"))?;
    let mut ack = [0u8; 64];
    let n = s
        .read(&mut ack)
        .await
        .map_err(|e| format!("vsock CONNECT ack: {e}"))?;
    if !String::from_utf8_lossy(&ack[..n]).starts_with("OK") {
        return Err("vsock refused the tail connection".into());
    }

    let req = json!({
        "op": "tail", "path": path, "from": from,
        // Long enough that a quiet agent is not mistaken for a finished one,
        // short enough that the thread is released soon after the turn ends.
        "idle_ms": 15_000, "max_secs": 3_600,
    });
    let body = serde_json::to_vec(&req).map_err(|e| format!("encode tail: {e}"))?;
    s.write_all(&(body.len() as u32).to_be_bytes())
        .await
        .map_err(|e| format!("write tail length: {e}"))?;
    s.write_all(&body)
        .await
        .map_err(|e| format!("write tail body: {e}"))?;

    let mut at = from;
    loop {
        let mut len = [0u8; 4];
        if s.read_exact(&mut len).await.is_err() {
            // The guest hung up: the turn ended or the VM went away. Not an
            // error — the caller already has the turn's own result.
            return Ok(at);
        }
        let len = u32::from_be_bytes(len);
        if len > MAX_REPLY {
            return Err(format!("tail frame of {len} bytes exceeds the cap"));
        }
        let mut buf = vec![0u8; len as usize];
        if s.read_exact(&mut buf).await.is_err() {
            return Ok(at);
        }
        let v: Value = match serde_json::from_slice(&buf) {
            Ok(v) => v,
            Err(e) => return Err(format!("decode tail frame: {e}")),
        };
        if let Some(d) = v.get("data").and_then(Value::as_str) {
            at = v.get("at").and_then(Value::as_u64).unwrap_or(at);
            on_chunk(at, d.to_string());
        }
        if v.get("eof").and_then(Value::as_bool) == Some(true) {
            return Ok(v.get("at").and_then(Value::as_u64).unwrap_or(at));
        }
    }
}

/// Resolve a backend name to the rootfs image on this node.
///
/// `None` (or `"default"`) means the golden `rootfs.ext4`; anything else selects
/// `rootfs-<backend>.ext4`, built by `scripts/fc-build-rootfs.sh`.
///
/// A missing image is an explicit error naming the file and how to build it. The
/// tempting fallback — quietly boot the default when the requested image is
/// absent — would run a claude mission in a kimi VM, or in a rootfs with no CLI
/// at all, and report success for whatever came out.
fn rootfs_for(backend: Option<&str>) -> Result<PathBuf, String> {
    let root = work_root();
    let name = match backend {
        None | Some("") | Some("default") => return Ok(root.join("rootfs.ext4")),
        Some(b) => b,
    };
    // Becomes a filename, so the same rule as vm ids applies.
    check_id(name).map_err(|e| format!("backend {name:?}: {e}"))?;
    let path = root.join(format!("rootfs-{name}.ext4"));
    if !path.is_file() {
        return Err(format!(
            "no rootfs for backend {name:?} on this node ({} is missing) — build it: \
             scripts/fc-build-rootfs.sh <host> <docker-image> {name}",
            path.display()
        ));
    }
    Ok(path)
}

/// The backend names this node can actually boot, derived from the images on
/// disk.
///
/// Reported as a capability so **placement can require it**. Without this,
/// `missions.backend` is invisible to the scheduler: the first real microVM
/// mission was placed on morpheus because it reports `microvm: true`, while only
/// tank had `rootfs-claude.ext4`. It failed by name rather than booting the wrong
/// image — but whether a mission ran came down to which capable node was picked
/// first, which is a coin flip dressed as scheduling.
///
/// Deliberately in this module: it is the inverse of [`rootfs_for`], and the two
/// must agree about what a backend name means. Split apart, one of them drifts
/// and the scheduler starts promising images the booter cannot find.
pub fn available_backends() -> Vec<String> {
    let root = work_root();
    let mut out = Vec::new();
    // `rootfs.ext4` is what `None`/`""`/`"default"` resolve to.
    if root.join("rootfs.ext4").is_file() {
        out.push("default".to_string());
    }
    if let Ok(entries) = std::fs::read_dir(&root) {
        for e in entries.flatten() {
            let name = e.file_name().to_string_lossy().to_string();
            if let Some(rest) = name.strip_prefix("rootfs-") {
                if let Some(backend) = rest.strip_suffix(".ext4") {
                    // Only what `rootfs_for` would accept, so the list cannot
                    // advertise a name the booter would reject.
                    if check_id(backend).is_ok() && e.path().is_file() {
                        out.push(backend.to_string());
                    }
                }
            }
        }
    }
    out.sort();
    out
}

/// The agent CLI a backend image is named for, and the command that proves it
/// is present.
///
/// A rootfs built from the wrong Dockerfile boots perfectly well and then has
/// no agent inside it. The first rootfs built on this track came from
/// `clawmates/agent-terminal:dev`, which turned out to contain git and nothing
/// else — no `claude`, no `node`. Nothing about the boot said so; it would have
/// surfaced as a mission that ran, produced no files, and reported completed.
/// So the selftest asks the image directly rather than trusting its name.
///
/// An unrecognised backend has no required CLI — that is reported as unchecked,
/// never as a pass.
fn required_cli(backend: Option<&str>) -> Option<(&'static str, &'static str)> {
    match backend {
        Some("claude") => Some(("claude", "claude --version")),
        // GLM and Kimi are both Claude Code pointed at another provider's
        // Anthropic-compatible endpoint (z.ai, and api.kimi.com/coding), so the
        // binary in all three images is `claude`.
        //
        // `kimi` used to expect Moonshot's own `kimi` CLI here. That was written
        // before the endpoint was measured, and it failed the selftest of a
        // rootfs that was in fact correct — the image runs Claude Code because
        // the whole mission harness is Claude-Code-shaped.
        Some("glm") | Some("kimi") => Some(("claude", "claude --version")),
        _ => None,
    }
}

/// Boot a VM and wait until its agent answers.
///
/// "Started" is not "usable": a VM whose agent never comes up is a process that
/// looks healthy and serves nothing, so create does not return until a `ping`
/// has round-tripped. If it never does, the VM is destroyed rather than left
/// registered — a half-created VM in the map is a leak with a plausible alibi.
pub async fn create(
    vms: &Vms,
    vm_id: &str,
    vcpus: u32,
    mem_mib: u32,
    backend: Option<&str>,
) -> Result<Value, String> {
    check_id(vm_id)?;
    let golden = rootfs_for(backend)?;
    if vms.lock().await.contains_key(vm_id) {
        return Err(format!("vm {vm_id} already exists"));
    }
    let root = work_root();
    let workdir = root.join("vms").join(vm_id);
    if workdir.exists() {
        // Left over from a crash. Reusing it would inherit a dirty rootfs.
        let _ = tokio::fs::remove_dir_all(&workdir).await;
    }
    tokio::fs::create_dir_all(&workdir)
        .await
        .map_err(|e| format!("mkdir {}: {e}", workdir.display()))?;

    // Sparse copy of the golden image: the guest gets its own writable disk
    // without paying for a full 1 GB copy per VM.
    let rootfs = workdir.join("rootfs.ext4");
    let cp = tokio::process::Command::new("cp")
        .arg("--sparse=always")
        .arg(&golden)
        .arg(&rootfs)
        .output()
        .await
        .map_err(|e| format!("spawn cp: {e}"))?;
    if !cp.status.success() {
        return Err(format!(
            "copy rootfs: {}",
            String::from_utf8_lossy(&cp.stderr).trim()
        ));
    }

    let uds = workdir.join("v.sock");
    let cfg = json!({
        "boot-source": {
            "kernel_image_path": root.join("vmlinux").display().to_string(),
            "boot_args": "console=ttyS0 reboot=k panic=1 pci=off init=/usr/local/bin/fcinit",
        },
        "drives": [{
            "drive_id": "rootfs",
            "path_on_host": rootfs.display().to_string(),
            "is_root_device": true,
            "is_read_only": false,
        }],
        "machine-config": { "vcpu_count": vcpus, "mem_size_mib": mem_mib, "smt": false },
        // guest_cid 3 is the lowest usable; the guest is addressed through this
        // VM's own UDS, so it need not be unique across VMs on the host.
        "vsock": { "guest_cid": 3, "uds_path": uds.display().to_string() },
    });
    let cfg_path = workdir.join("vm.json");
    tokio::fs::write(&cfg_path, serde_json::to_vec_pretty(&cfg).unwrap_or_default())
        .await
        .map_err(|e| format!("write {}: {e}", cfg_path.display()))?;

    // Bound BEFORE firecracker starts: a guest that dials the host before the
    // host is listening gets a connection refused it will not retry.
    let (egress_uds, egress_task) = match crate::egress::start(&uds, vm_id, backend) {
        Ok((p, t)) => (p, Some(t)),
        // Not fatal — a VM is still useful for work that needs no network — but
        // it must be visible. `create`'s reply says whether egress exists, and
        // the guest's own `proxy` flag says whether the guest end came up.
        Err(e) => {
            eprintln!("microvm {vm_id}: NO EGRESS ({e}) — the guest cannot reach any network");
            (PathBuf::from(format!("{}_{}", uds.display(), crate::egress::EGRESS_PORT)), None)
        }
    };

    // The node's own model, for a backend that has one. Bound before firecracker
    // starts for the same reason egress is: a guest that dials before the host
    // listens gets a refusal it will not retry.
    let (model_uds, model_task) = match crate::local_model::start(&uds, vm_id, backend) {
        Ok(Some((p, t))) => (Some(p), Some(t)),
        Ok(None) => (None, None),
        // This backend was supposed to have a local model and does not. Not
        // fatal, but the mission WILL fail on its first turn, so say why here
        // rather than leaving it to look like a hung agent.
        Err(e) => {
            eprintln!(
                "microvm {vm_id}: NO LOCAL MODEL ({e}) — a {backend:?} turn cannot reach one"
            );
            (None, None)
        }
    };

    let log = std::fs::File::create(workdir.join("console.log"))
        .map_err(|e| format!("create console.log: {e}"))?;
    let errlog = log
        .try_clone()
        .map_err(|e| format!("clone console.log handle: {e}"))?;

    // `setsid` puts firecracker in its own process group so destroy can kill the
    // group. Without it a background child outlives the VM and keeps the
    // workdir busy — the "background children hang the run" failure the
    // Firecracker write-ups warn about.
    let child = tokio::process::Command::new("setsid")
        .arg("firecracker")
        .arg("--no-api")
        .arg("--config-file")
        .arg(&cfg_path)
        .stdout(log)
        .stderr(errlog)
        .stdin(std::process::Stdio::null())
        .spawn()
        .map_err(|e| format!("spawn firecracker: {e}"))?;
    // setsid's child IS the new group leader, and its pid is the pgid.
    let pgid = child.id().ok_or("firecracker exited immediately")? as i32;

    let host_egress = egress_task.is_some();
    let vm = Vm {
        pgid,
        workdir: workdir.clone(),
        uds: uds.clone(),
        egress: egress_task,
        egress_uds: egress_uds.clone(),
        model: model_task,
        model_uds: model_uds.clone(),
    };

    // Poll for the agent. 10s is generous: the measured boot-to-agent is under
    // a second, so anything near the ceiling means something is wrong.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let mut last = String::new();
    // Assigned on the only path that reaches the use below; the timeout path
    // returns. Declared without a default so a future edit cannot make "we never
    // asked" read as "the guest said no".
    let guest_egress;
    loop {
        if std::time::Instant::now() > deadline {
            let console = tokio::fs::read_to_string(workdir.join("console.log"))
                .await
                .unwrap_or_default();
            kill_group(pgid).await;
            let _ = tokio::fs::remove_dir_all(&workdir).await;
            return Err(format!(
                "vm {vm_id} booted but its agent never answered ({last}); last console output: {}",
                console.lines().rev().take(3).collect::<Vec<_>>().join(" | ")
            ));
        }
        match rpc(&uds, &json!({"op": "ping"})).await {
            Ok(v) if v.get("ok").and_then(Value::as_bool) == Some(true) => {
                // The guest reports whether ITS end of the tunnel is listening.
                // Both ends must be up for the VM to have egress, and only the
                // guest knows whether its image could bring loopback up.
                guest_egress = v.get("proxy").and_then(Value::as_bool) == Some(true);
                break;
            }
            Ok(v) => last = v.to_string(),
            Err(e) => last = e,
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    vms.lock().await.insert(vm_id.to_string(), vm);
    Ok(json!({
        "vm_id": vm_id,
        "pgid": pgid,
        "workdir": workdir.display().to_string(),
        // Which image actually booted, not which was asked for. A mission
        // artifact that records the request cannot show that the wrong VM ran.
        "rootfs": golden.display().to_string(),
        // Egress needs BOTH ends. Reported rather than assumed so a caller that
        // requires the network can refuse the VM up front, instead of a mission
        // discovering it as an agent that cannot reach its API.
        "egress": host_egress && guest_egress,
        "egress_host": host_egress,
        "egress_guest": guest_egress,
    }))
}

async fn uds_of(vms: &Vms, vm_id: &str) -> Result<PathBuf, String> {
    vms.lock()
        .await
        .get(vm_id)
        .map(|v| v.uds.clone())
        .ok_or_else(|| format!("no such vm: {vm_id}"))
}

/// Unpack a tar into the guest at `dest`.
pub async fn inject(vms: &Vms, vm_id: &str, dest: &str, tar_b64: &str) -> Result<Value, String> {
    let uds = uds_of(vms, vm_id).await?;
    rpc(
        &uds,
        &json!({ "op": "put", "dest": dest, "tar_b64": tar_b64 }),
    )
    .await
}

/// Run a command in the guest and return its exit code and output.
/// `env` is passed straight to the guest, which validates it and refuses the
/// exec if any entry is unusable. It is NOT logged here or anywhere on the way:
/// this is the channel credentials travel on.
pub async fn exec(
    vms: &Vms,
    vm_id: &str,
    cmd: &str,
    cwd: Option<&str>,
    timeout_secs: u64,
    env: Option<&Value>,
) -> Result<Value, String> {
    let uds = uds_of(vms, vm_id).await?;
    rpc(
        &uds,
        &json!({
            "op": "exec", "cmd": cmd, "cwd": cwd,
            "timeout": timeout_secs, "env": env,
        }),
    )
    .await
}

/// Tar a path out of the guest.
pub async fn collect(
    vms: &Vms,
    vm_id: &str,
    path: &str,
    exclude: Option<&Value>,
) -> Result<Value, String> {
    let uds = uds_of(vms, vm_id).await?;
    rpc(&uds, &json!({ "op": "get", "path": path, "exclude": exclude })).await
}

/// SIGKILL a whole process group, ignoring "already gone".
///
/// Shells out rather than calling `killpg`: the workspace denies `unsafe`, and
/// one `kill(1)` per teardown is not worth either an FFI exception or a `libc`
/// dependency. `-- -PGID` is the POSIX spelling for "the group", and the `--`
/// matters — without it the negative number parses as a flag.
async fn kill_group(pgid: i32) {
    let _ = tokio::process::Command::new("kill")
        .arg("-9")
        .arg("--")
        .arg(format!("-{pgid}"))
        .output()
        .await;
}

/// Stop a VM and remove everything it owned.
///
/// Idempotent, and deliberately thorough about the socket: **Firecracker does
/// not unlink its vsock UDS on exit**, and leaves it owned by whoever ran the
/// VM. A driver that assumed the VM tidied up after itself would accumulate
/// root-owned sockets it could not remove — the same uid trap that cost this
/// codebase four bugs on the mission checkout.
pub async fn destroy(vms: &Vms, vm_id: &str) -> Result<Value, String> {
    check_id(vm_id)?;
    let vm = vms.lock().await.remove(vm_id);
    let (pgid, workdir, uds, egress_uds, model_uds) = match vm {
        Some(v) => {
            // Abort first: a live listener would keep accepting on a path the
            // next VM is about to reuse.
            if let Some(t) = v.egress {
                t.abort();
            }
            if let Some(t) = v.model {
                t.abort();
            }
            (Some(v.pgid), v.workdir, v.uds, v.egress_uds, v.model_uds)
        }
        // Not registered: still clean the paths, so a VM created by a previous
        // incarnation of the daemon can be reaped rather than orphaned forever.
        None => {
            let wd = work_root().join("vms").join(vm_id);
            let uds = wd.join("v.sock");
            let eg = PathBuf::from(format!("{}_{}", uds.display(), crate::egress::EGRESS_PORT));
            let md = PathBuf::from(format!(
                "{}_{}",
                uds.display(),
                crate::local_model::MODEL_PORT
            ));
            (None, wd.clone(), uds, eg, Some(md))
        }
    };
    // Killed means OBSERVED GONE, not asked-to-die.
    //
    // This used to report `killed: pgid.is_some()` — true whenever there was a
    // pgid to signal, whether or not anything died. A firecracker process was
    // found alive 1h37m after its VM was destroyed, holding a config file in a
    // workdir that no longer existed, while destroy had reported success. So the
    // signal is sent, the process is polled, and the answer is what was seen.
    let mut killed = false;
    if let Some(pgid) = pgid {
        for attempt in 0..20 {
            kill_group(pgid).await;
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            // The pgid is the group leader's pid, so /proc/<pgid> is the check.
            if !std::path::Path::new(&format!("/proc/{pgid}")).exists() {
                killed = true;
                break;
            }
            if attempt == 19 {
                eprintln!(
                    "microvm {vm_id}: process group {pgid} SURVIVED {} kill attempts — \
                     a firecracker process is still holding this VM's resources",
                    attempt + 1
                );
            }
        }
    }
    let _ = tokio::fs::remove_file(&uds).await;
    // Same trap as firecracker's own socket: nothing unlinks these for us, and a
    // stale file makes the next bind fail with EADDRINUSE.
    let _ = tokio::fs::remove_file(&egress_uds).await;
    if let Some(m) = &model_uds {
        let _ = tokio::fs::remove_file(m).await;
    }
    let removed = tokio::fs::remove_dir_all(&workdir).await.is_ok();
    // `killed` is now observed rather than assumed; `signalled` keeps the old
    // meaning so a caller can tell "there was nothing to kill" from
    // "we tried and it would not die".
    Ok(json!({
        "vm_id": vm_id,
        "killed": killed,
        "signalled": pgid.is_some(),
        "workdir_removed": removed,
    }))
}

/// VMs this node currently holds, so the server can reap orphans.
pub async fn list(vms: &Vms) -> Value {
    let held: Vec<Value> = vms
        .lock()
        .await
        .iter()
        .map(|(id, v)| json!({ "vm_id": id, "pgid": v.pgid }))
        .collect();
    json!({ "vms": held })
}

/// Dispatch a `vm_*` frame.
///
/// Returns `output` as a **String**, not a `Value`, because the server's
/// `Uplink::Result` declares `output: String`. Sending an object made the whole
/// result frame fail to deserialize, and the server's uplink match ends in
/// `Err(_) => {}` — so the reply vanished and the caller timed out after 20s
/// with nothing to explain why. The type had to match the wire contract, not
/// merely look tidier.
pub async fn handle_op(op: &str, v: &Value, vms: &Vms) -> (bool, String) {
    let s = |k: &str| v.get(k).and_then(Value::as_str).unwrap_or_default().to_string();
    let u = |k: &str, d: u64| v.get(k).and_then(Value::as_u64).unwrap_or(d);
    let vm_id = s("vm_id");

    let r: Result<Value, String> = match op {
        "vm_create" => {
            create(
                vms,
                &vm_id,
                u("vcpus", 2) as u32,
                u("mem_mib", 2048) as u32,
                v.get("backend").and_then(Value::as_str),
            )
            .await
        }
        "vm_inject" => inject(vms, &vm_id, &s("dest"), &s("tar_b64")).await,
        "vm_exec" => {
            let cwd = v.get("cwd").and_then(Value::as_str);
            exec(
                vms,
                &vm_id,
                &s("cmd"),
                cwd,
                u("timeout", 3600),
                v.get("env"),
            )
            .await
        }
        "vm_collect" => collect(vms, &vm_id, &s("path"), v.get("exclude")).await,
        "vm_destroy" => destroy(vms, &vm_id).await,
        "vm_list" => Ok(list(vms).await),
        other => Err(format!("unknown vm op: {other}")),
    };
    match r {
        // The guest reports its own `ok`, and a command that ran and failed is
        // not a transport failure — the caller needs `rc` either way, so the
        // frame is ok:true and the verdict lives in the payload.
        Ok(out) => (true, out.to_string()),
        Err(e) => (false, json!({ "error": e }).to_string()),
    }
}

/// Exercise the whole lifecycle against a real VM: `clawmates-node --vm-selftest`.
///
/// The bash setup script proves firecracker and the guest agent work; it proves
/// nothing about *this* code. This runs create → inject → exec → collect →
/// destroy through the same functions the server will call, on the node, as the
/// daemon user, and checks the host is clean afterwards.
///
/// Every step is asserted. A selftest that only reports the steps it completed
/// cannot distinguish "passed" from "stopped early".
pub async fn selftest() -> bool {
    use base64::Engine as _;
    let b64 = base64::engine::general_purpose::STANDARD;
    let vms = new_vms();
    let id = "selftest";
    let mut failures = 0;
    let mut check = |ok: bool, what: &str, detail: String| {
        if ok {
            println!("PASS  {what}");
        } else {
            println!("FAIL  {what}: {detail}");
            failures += 1;
        }
    };

    // Start from a clean slate even if a previous run died mid-way.
    let _ = destroy(&vms, id).await;

    // Which image to exercise. Defaults to the golden rootfs; set
    // CLAWMATES_FC_BACKEND=agent-terminal to prove a built per-CLI image boots.
    let backend = std::env::var("CLAWMATES_FC_BACKEND").ok();
    println!(
        "      backend: {}",
        backend.as_deref().unwrap_or("(default rootfs.ext4)")
    );

    // A backend whose image is absent must fail by name, not fall back to the
    // default — booting the wrong rootfs would report success for whatever came
    // out of it. Checked here so the guarantee is exercised on real hardware and
    // not only in a unit test with a temp dir.
    match create(&vms, "selftest-absent", 2, 512, Some("definitely-not-built")).await {
        Err(e) if e.contains("rootfs-definitely-not-built.ext4") => {
            check(true, "an absent backend image fails by name", String::new())
        }
        Err(e) => check(false, "an absent backend image fails by name", e),
        Ok(_) => {
            let _ = destroy(&vms, "selftest-absent").await;
            check(
                false,
                "an absent backend image fails by name",
                "it BOOTED — a missing image fell back to the default".into(),
            )
        }
    }

    let started = std::time::Instant::now();
    let created = match create(&vms, id, 2, 1024, backend.as_deref()).await {
        Ok(v) => {
            check(
                true,
                &format!("create ({} ms) {}", started.elapsed().as_millis(), v["vm_id"]),
                String::new(),
            );
            v
        }
        Err(e) => {
            check(false, "create", e);
            println!("\n1 or more checks failed");
            return false;
        }
    };

    // Inject a tar the way the mission checkout will travel.
    let mut tar = tar::Builder::new(Vec::new());
    let body = b"INJECTED-OK\n";
    let mut hdr = tar::Header::new_gnu();
    hdr.set_path("marker.txt").unwrap();
    hdr.set_size(body.len() as u64);
    hdr.set_mode(0o644);
    hdr.set_entry_type(tar::EntryType::Regular);
    hdr.set_cksum();
    tar.append(&hdr, &body[..]).unwrap();
    let archive = tar.into_inner().unwrap();
    let r = inject(&vms, id, "/work", &b64.encode(&archive)).await;
    check(
        r.as_ref().map(|v| v["ok"] == json!(true)).unwrap_or(false),
        "inject a tar into /work",
        format!("{r:?}"),
    );

    // The guest must SEE what we injected — an inject that reports ok while
    // landing nothing is the failure shape this codebase keeps paying for.
    let r = exec(&vms, id, "cat /work/marker.txt", None, 30, None).await;
    let saw = r
        .as_ref()
        .map(|v| v["stdout"].as_str().unwrap_or_default().contains("INJECTED-OK"))
        .unwrap_or(false);
    check(saw, "the guest reads the injected file", format!("{r:?}"));

    // A failing command must come back as rc != 0, not as a transport error:
    // the caller needs to tell "the command failed" from "we could not run it".
    let r = exec(&vms, id, "exit 3", None, 30, None).await;
    check(
        r.as_ref().map(|v| v["rc"] == json!(3)).unwrap_or(false),
        "a failing command reports rc=3 rather than an error",
        format!("{r:?}"),
    );

    // Credentials reach the agent CLI as exec env, and this is the only place
    // that is proven over the real vsock wire rather than in a unit test: the
    // failure it guards against is a `claude -p` with no token, which does not
    // error — it hangs.
    let r = exec(
        &vms,
        id,
        "printf %s \"$CLAWMATES_ENV_PROBE\"",
        None,
        30,
        Some(&json!({ "CLAWMATES_ENV_PROBE": "env-injection-ok" })),
    )
    .await;
    check(
        r.as_ref()
            .map(|v| v["stdout"] == json!("env-injection-ok"))
            .unwrap_or(false),
        "injected env reaches the guest command",
        format!("{r:?}"),
    );

    // And an entry the guest cannot honour must fail the exec rather than run
    // the command without it.
    let r = exec(
        &vms,
        id,
        "true",
        None,
        30,
        Some(&json!({ "BAD=NAME": "x" })),
    )
    .await;
    let refused = r
        .as_ref()
        .map(|v| v["ok"] == json!(false) && v["rc"].is_null())
        .unwrap_or(false);
    check(
        refused,
        "an unusable env entry refuses the exec instead of dropping it",
        format!("{r:?}"),
    );

    // Egress. Both ends of the tunnel must be up, and `create` says so rather
    // than leaving the caller to find out from an agent that cannot reach its
    // API. Only asserted for a backend whose image is expected to have iproute2.
    if backend.is_some() {
        check(
            created["egress"] == json!(true),
            "the VM reports working egress (both ends of the tunnel)",
            format!(
                "host={} guest={}",
                created["egress_host"], created["egress_guest"]
            ),
        );

        // The allow-list has to actually let the model API through — this is the
        // check that says a mission could run — and it goes over the real tunnel:
        // guest loopback → vsock → host proxy → TLS to Anthropic. `-sS -o
        // /dev/null -w %{http_code}` because any 2xx/4xx from the API proves the
        // connection completed; only a transport failure gives no code at all.
        let r = exec(
            &vms,
            id,
            "curl -sS -o /dev/null -w '%{http_code}' --max-time 25 https://api.anthropic.com/v1/messages",
            None,
            40,
            None,
        )
        .await;
        let code = r
            .as_ref()
            .map(|v| v["stdout"].as_str().unwrap_or_default().trim().to_string())
            .unwrap_or_default();
        check(
            code.len() == 3 && code != "000",
            "an allow-listed host is reachable from inside the VM",
            format!("http_code={code:?} {r:?}"),
        );

        // And the denial must fire. Without this the allow-list is decoration:
        // a proxy that allows everything passes the check above just as well.
        let r = exec(
            &vms,
            id,
            "curl -sS -o /dev/null -w '%{http_code}' --max-time 25 https://example.com",
            None,
            40,
            None,
        )
        .await;
        let out = r
            .as_ref()
            .map(|v| {
                format!(
                    "{}{}",
                    v["stdout"].as_str().unwrap_or_default(),
                    v["stderr"].as_str().unwrap_or_default()
                )
            })
            .unwrap_or_default();
        // The refusal must come from THE PROXY, not from a dead network. When
        // this check merely asserted "did not reach it", it passed while the
        // guest had no proxy env at all and curl was failing with "Could not
        // resolve host" — a green result for a broken tunnel, which is the exact
        // failure shape this project keeps paying for. `403` is the status the
        // proxy returns after CONNECT for a host off the list.
        check(
            out.contains("403") || out.contains("allow-list"),
            "a host that is NOT allow-listed is refused BY THE PROXY",
            format!("expected a 403 from the proxy, got: {out:?}"),
        );
    }

    // Work produced in the guest must come back out.
    let _ = exec(&vms, id, "echo PRODUCED-OK > /work/out.txt", None, 30, None).await;
    let r = collect(&vms, id, "/work", None).await;
    let round_tripped = r
        .as_ref()
        .ok()
        .and_then(|v| v["tar_b64"].as_str().map(|s| s.to_string()))
        .and_then(|s| b64.decode(s).ok())
        .map(|bytes| {
            let mut ar = tar::Archive::new(&bytes[..]);
            ar.entries()
                .map(|es| {
                    es.filter_map(Result::ok)
                        .any(|e| e.path().map(|p| p.ends_with("out.txt")).unwrap_or(false))
                })
                .unwrap_or(false)
        })
        .unwrap_or(false);
    check(
        round_tripped,
        "collect brings the guest's work back as a tar",
        format!("{r:?}"),
    );

    // The image must actually contain the agent the mission will run. This is
    // the check that separates "a VM booted" from "a mission could run in it",
    // and it is exec'd inside the guest rather than inferred from the image name.
    match required_cli(backend.as_deref()) {
        Some((cli, probe)) => {
            let r = exec(&vms, id, probe, None, 60, None).await;
            let (rc, out) = match r.as_ref() {
                Ok(v) => (
                    v["rc"].as_i64(),
                    v["stdout"].as_str().unwrap_or_default().trim().to_string(),
                ),
                Err(e) => (None, e.clone()),
            };
            check(
                rc == Some(0),
                &format!("the guest provides the {cli} CLI"),
                format!("`{probe}` gave rc={rc:?} {out}"),
            );
            if rc == Some(0) {
                println!("      {cli}: {out}");
            }
            // git is what delivery is built on: the host captures a phase by
            // diffing the collected tree, so an image without git delivers
            // nothing no matter which CLI it has.
            let r = exec(&vms, id, "git --version", None, 30, None).await;
            check(
                r.as_ref().map(|v| v["rc"] == json!(0)).unwrap_or(false),
                "the guest provides git",
                format!("{r:?}"),
            );

            // The whole track in one check: a real agent turn, in a VM with no
            // network card, reaching the API through the vsock tunnel on the
            // subscription credential. Everything above can pass while this
            // fails, which is why it is asked separately.
            //
            // Gated on a token being present, and SKIPPED loudly when it is not —
            // it spends a small amount of subscription budget, so it must be a
            // deliberate act rather than something every infra check does.
            match std::env::var("CLAUDE_CODE_OAUTH_TOKEN") {
                Ok(t) if !t.trim().is_empty() => {
                    let r = exec(
                        &vms,
                        id,
                        "cd /work && claude -p 'Reply with exactly: VM-OK'",
                        None,
                        180,
                        // Subscription only. An ANTHROPIC_API_KEY would outrank
                        // this token and bill the API instead of the plan.
                        Some(&json!({ "CLAUDE_CODE_OAUTH_TOKEN": t })),
                    )
                    .await;
                    let said = r
                        .as_ref()
                        .map(|v| v["stdout"].as_str().unwrap_or_default().to_string())
                        .unwrap_or_default();
                    check(
                        said.contains("VM-OK"),
                        "the agent completes a real turn inside the VM (subscription auth)",
                        format!("said {said:?} — {r:?}"),
                    );
                }
                _ => println!(
                    "SKIP  no CLAUDE_CODE_OAUTH_TOKEN in the environment — the agent's \
                     own turn is UNPROVEN by this run"
                ),
            }
        }
        None => println!(
            "SKIP  no agent CLI is required of backend {} — its contents are unchecked",
            backend.as_deref().unwrap_or("(default rootfs.ext4)")
        ),
    }

    let r = destroy(&vms, id).await;
    // `killed` is now an OBSERVATION — the process was polled and found gone —
    // rather than "we sent a signal". A firecracker was found alive 1h37m after a
    // destroy that had reported success.
    check(
        r.as_ref()
            .map(|v| v["killed"] == json!(true) && v["workdir_removed"] == json!(true))
            .unwrap_or(false),
        "destroy kills the VM and removes its workdir",
        format!("{r:?}"),
    );

    // Firecracker does not unlink its vsock UDS; if destroy did not, it is still
    // there. This is the check for the trap the spike found.
    let uds = work_root().join("vms").join(id).join("v.sock");
    check(
        !uds.exists(),
        "the vsock socket is gone after destroy",
        format!("{} still exists", uds.display()),
    );

    // And the VM must not be usable afterwards — a destroy that leaves a live
    // guest answering is worse than one that errors.
    let r = exec(&vms, id, "echo still-here", None, 5, None).await;
    check(
        r.is_err(),
        "a destroyed VM can no longer be exec'd",
        format!("{r:?}"),
    );

    if failures == 0 {
        println!("\nall microvm checks passed");
        true
    } else {
        println!("\n{failures} microvm check(s) failed");
        false
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Each per-CLI backend must name a CLI to probe for, or a rootfs built
    /// from the wrong image passes the selftest by saying nothing.
    #[test]
    fn every_per_cli_backend_declares_the_cli_it_must_contain() {
        for (backend, want) in [("claude", "claude"), ("kimi", "claude"), ("glm", "claude")] {
            let (cli, probe) = required_cli(Some(backend))
                .unwrap_or_else(|| panic!("backend {backend} requires no CLI"));
            assert_eq!(cli, want, "backend {backend}");
            assert!(probe.starts_with(cli), "probe {probe:?} must run {cli}");
        }
    }

    /// And an unknown backend reports "unchecked", which the selftest prints as
    /// SKIP. Returning a plausible default here would claim a guarantee about
    /// an image nobody has looked inside.
    #[test]
    fn an_unknown_backend_requires_no_cli_rather_than_a_guessed_one() {
        for b in [None, Some(""), Some("default"), Some("agent-terminal")] {
            assert!(required_cli(b).is_none(), "backend {b:?}");
        }
    }

    /// A vm id becomes a path component, so it must not be able to describe a
    /// path. These are rejected, not sanitised — a caller that sent `../../etc`
    /// wanted something we should not silently reinterpret.
    #[test]
    fn a_vm_id_cannot_escape_the_work_directory() {
        for bad in [
            "../etc",
            "a/b",
            "..",
            ".",
            "/abs",
            "with space",
            "semi;colon",
            "",
            "nul\0byte",
        ] {
            assert!(check_id(bad).is_err(), "{bad:?} must be rejected");
        }
        for good in ["ok", "mission-019fcf62", "a_b-C9", &"x".repeat(64)] {
            assert!(check_id(good).is_ok(), "{good:?} must be accepted");
        }
        assert!(
            check_id(&"x".repeat(65)).is_err(),
            "an over-long id must be rejected"
        );
    }

    /// An absent image must be an ERROR, never a silent fall back to the golden
    /// rootfs: booting the default would run a claude mission in a kimi VM, or
    /// in a rootfs with no CLI at all, and report success for whatever came out.
    #[test]
    fn a_missing_backend_image_is_an_error_not_a_fallback() {
        let tmp = tempfile::tempdir().unwrap();
        std::env::set_var("CLAWMATES_FC_ROOT", tmp.path());

        let e = rootfs_for(Some("kimi")).expect_err("a missing image must fail");
        assert!(e.contains("rootfs-kimi.ext4"), "must name the file: {e}");
        assert!(e.contains("fc-build-rootfs.sh"), "must say how to fix: {e}");

        // Present → selected.
        std::fs::write(tmp.path().join("rootfs-kimi.ext4"), b"x").unwrap();
        assert!(rootfs_for(Some("kimi"))
            .unwrap()
            .ends_with("rootfs-kimi.ext4"));

        // The default is the golden image, and is not required to exist for the
        // name to resolve — vm_create's copy reports that.
        for none_ish in [None, Some(""), Some("default")] {
            assert!(rootfs_for(none_ish).unwrap().ends_with("rootfs.ext4"));
        }
        std::env::remove_var("CLAWMATES_FC_ROOT");
    }

    /// A backend name becomes a filename, so it must not be able to describe a
    /// path any more than a vm id can.
    #[test]
    fn a_backend_name_cannot_escape_the_work_directory() {
        for bad in ["../etc/passwd", "a/b", ".."] {
            assert!(rootfs_for(Some(bad)).is_err(), "{bad:?} must be rejected");
        }
    }
}