Merge: B2 vm_* node ops — microVM lifecycle proven on tank (8/8)

Co-Authored-By: Claude Opus 5 <[email protected]>

Cargo.lock

@@ -846,6 +846,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sysinfo",
+ "tar",
  "tokio",
  "tokio-tungstenite 0.26.2",
  "webrtc",

crates/bins/clawmates-node/Cargo.toml

@@ -19,6 +19,7 @@ serde_json = { workspace = true }
 sysinfo = "0.33"
 portable-pty = "0.8"
 base64 = "0.22"
+tar = { workspace = true }
 cm-sandbox = { path = "../../cm-sandbox" }
 # Linking cm-sandbox (bollard) brings a second rustls provider into the graph, so
 # rustls can't auto-pick one — we install `ring` explicitly at startup.

crates/bins/clawmates-node/src/main.rs

@@ -19,6 +19,7 @@ use sysinfo::{Disks, System};
 use tokio::sync::{mpsc, Mutex};
 use tokio_tungstenite::tungstenite::Message;
 
+mod microvm;
 mod rtc;
 
 const B64: base64::engine::general_purpose::GeneralPurpose =
@@ -43,6 +44,15 @@ async fn main() {
         selftest();
         return;
     }
+    // Exercise the microVM lifecycle against a real VM on this node. Separate
+    // from --selftest because it needs KVM, so it can only pass on a node that
+    // actually reports microvm capability.
+    if std::env::args().any(|a| a == "--vm-selftest") {
+        if !microvm::selftest().await {
+            std::process::exit(1);
+        }
+        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>]");
@@ -108,6 +118,11 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
     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()));
+    // microVMs this connection started. Scoped to the connection deliberately:
+    // a reconnect must not inherit VMs it cannot prove are still alive, and
+    // `vm_destroy` cleans a workdir by path even for an unregistered id, so a
+    // VM from a previous incarnation is reapable rather than orphaned.
+    let vms = microvm::new_vms();
     // 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).
@@ -188,8 +203,9 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
                         let out = out_tx.clone();
                         let ptys = ptys.clone();
                         let peers = peers.clone();
+                        let vms = vms.clone();
                         let text = t.to_string();
-                        tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers).await; });
+                        tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers, &vms).await; });
                     }
                     Some(Ok(Message::Ping(p))) => {
                         match tokio::time::timeout(WRITE_DEADLINE, write.send(Message::Pong(p))).await {
@@ -520,6 +536,7 @@ async fn handle_frame(
     out: &mpsc::UnboundedSender<String>,
     ptys: &Ptys,
     peers: &rtc::RtcPeers,
+    vms: &microvm::Vms,
 ) {
     let Ok(v) = serde_json::from_str::<Value>(text) else {
         return;
@@ -582,6 +599,23 @@ async fn handle_frame(
         // 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.
+        // microVM ops. Same envelope as every other op, so adding them needed
+        // no protocol change. `vm_create` blocks until the guest agent answers:
+        // a VM that booted but serves nothing is worse than one that failed.
+        op @ ("vm_create" | "vm_inject" | "vm_exec" | "vm_collect" | "vm_destroy" | "vm_list") => {
+            if let Some(id) = v.get("id").and_then(Value::as_u64) {
+                let (op, v, out, vms) = (op.to_string(), v.clone(), out.clone(), vms.clone());
+                // Spawned: a VM boot takes ~1s and an exec can take an hour.
+                // Running it inline would stall heartbeats and the daemon would
+                // be declared offline mid-mission.
+                tokio::spawn(async move {
+                    let (ok, output) = microvm::handle_op(&op, &v, &vms).await;
+                    let _ = out.send(
+                        json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
+                    );
+                });
+            }
+        }
         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;

crates/bins/clawmates-node/src/microvm.rs

@@ -0,0 +1,537 @@
+//! 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,
+}
+
+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.
+async fn rpc(uds: &Path, req: &Value) -> Result<Value, String> {
+    const MAX_REPLY: u32 = 256 * 1024 * 1024;
+
+    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}"))
+}
+
+/// 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) -> Result<Value, String> {
+    check_id(vm_id)?;
+    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(root.join("rootfs.ext4"))
+        .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()))?;
+
+    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 vm = Vm {
+        pgid,
+        workdir: workdir.clone(),
+        uds: 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();
+    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) => 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() }))
+}
+
+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.
+pub async fn exec(
+    vms: &Vms,
+    vm_id: &str,
+    cmd: &str,
+    cwd: Option<&str>,
+    timeout_secs: u64,
+) -> Result<Value, String> {
+    let uds = uds_of(vms, vm_id).await?;
+    rpc(
+        &uds,
+        &json!({ "op": "exec", "cmd": cmd, "cwd": cwd, "timeout": timeout_secs }),
+    )
+    .await
+}
+
+/// Tar a path out of the guest.
+pub async fn collect(vms: &Vms, vm_id: &str, path: &str) -> Result<Value, String> {
+    let uds = uds_of(vms, vm_id).await?;
+    rpc(&uds, &json!({ "op": "get", "path": path })).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) = match vm {
+        Some(v) => (Some(v.pgid), v.workdir, v.uds),
+        // Not registered: still clean the path, 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);
+            (None, wd.clone(), wd.join("v.sock"))
+        }
+    };
+    if let Some(pgid) = pgid {
+        kill_group(pgid).await;
+    }
+    // Give the group a moment to die before removing the files it has open.
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    let _ = tokio::fs::remove_file(&uds).await;
+    let removed = tokio::fs::remove_dir_all(&workdir).await.is_ok();
+    Ok(json!({ "vm_id": vm_id, "killed": 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 `(ok, output)` in the same shape every
+/// other node op uses, so this needed no protocol change.
+pub async fn handle_op(op: &str, v: &Value, vms: &Vms) -> (bool, Value) {
+    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).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)).await
+        }
+        "vm_collect" => collect(vms, &vm_id, &s("path")).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),
+        Err(e) => (false, json!({ "error": e })),
+    }
+}
+
+/// 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;
+
+    let started = std::time::Instant::now();
+    match create(&vms, id, 2, 1024).await {
+        Ok(v) => check(
+            true,
+            &format!("create ({} ms) {}", started.elapsed().as_millis(), v["vm_id"]),
+            String::new(),
+        ),
+        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).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).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:?}"),
+    );
+
+    // Work produced in the guest must come back out.
+    let _ = exec(&vms, id, "echo PRODUCED-OK > /work/out.txt", None, 30).await;
+    let r = collect(&vms, id, "/work").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:?}"),
+    );
+
+    let r = destroy(&vms, id).await;
+    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).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::*;
+
+    /// 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"
+        );
+    }
+}

scripts/fc-node-setup.sh

@@ -96,42 +96,161 @@ for host in "$@"; do
   " >/dev/null 2>&1 || { fail "$host" "could not stage kernel/rootfs"; continue; }
   pass "$host" "kernel + rootfs staged in $WORK"
 
-  # 4. Boot one. Installing proves nothing; this is the check that counts.
+  # 4. Bake the guest agent into the shared rootfs.
+  #
+  # It has to be baked rather than injected per VM: mounting an ext4 image to
+  # write into it needs root, and the node daemon deliberately runs as an
+  # ordinary user (in the kvm group). Baking once at setup time is the only
+  # place root is available, so the per-VM path stays unprivileged.
+  #
+  # Control is length-prefixed JSON over vsock, NOT the serial console — see
+  # the spike: stdin races the guest's startup and arrives half-consumed.
+  ssh "$host" "
+    set -e
+    cd '$WORK'
+    sudo mkdir -p /mnt/fcroot && sudo mount -o loop rootfs.ext4 /mnt/fcroot
+    sudo tee /mnt/fcroot/usr/local/bin/fcagent >/dev/null <<'PY'
+#!/usr/bin/env python3
+# ClawMates microVM guest agent. One request per frame, framed as a 4-byte
+# big-endian length followed by JSON, so a reply larger than a socket buffer
+# cannot be mistaken for a complete one.
+import socket, struct, subprocess, json, os, base64, tarfile, io, sys
+
+def recv_exact(c, n):
+    buf = b''
+    while len(buf) < n:
+        chunk = c.recv(n - len(buf))
+        if not chunk:
+            return None
+        buf += chunk
+    return buf
+
+def handle(req):
+    op = req.get('op')
+    if op == 'ping':
+        return {'ok': True, 'pid': os.getpid()}
+    if op == 'exec':
+        # Setsid so background children join a new process group the host can
+        # kill wholesale; without it a stray daemon keeps the run alive forever.
+        p = subprocess.run(req['cmd'], shell=True, capture_output=True, text=True,
+                           cwd=req.get('cwd') or '/', timeout=req.get('timeout', 3600),
+                           preexec_fn=os.setsid)
+        return {'ok': True, 'rc': p.returncode, 'stdout': p.stdout, 'stderr': p.stderr}
+    if op == 'put':
+        # A tar, not a raw file: it carries directories, modes and multiple
+        # entries, and is the same shape the mission checkout already travels in.
+        raw = base64.b64decode(req['tar_b64'])
+        dest = req['dest']
+        os.makedirs(dest, exist_ok=True)
+        with tarfile.open(fileobj=io.BytesIO(raw)) as t:
+            t.extractall(dest)
+        return {'ok': True, 'dest': dest}
+    if op == 'get':
+        src = req['path']
+        if not os.path.exists(src):
+            return {'ok': False, 'error': 'no such path: ' + src}
+        buf = io.BytesIO()
+        with tarfile.open(fileobj=buf, mode='w') as t:
+            t.add(src, arcname=os.path.basename(src.rstrip('/')))
+        return {'ok': True, 'tar_b64': base64.b64encode(buf.getvalue()).decode()}
+    return {'ok': False, 'error': 'unknown op: ' + str(op)}
+
+s = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
+s.bind((socket.VMADDR_CID_ANY, 9001))
+s.listen(8)
+print('FC-AGENT-LISTENING', flush=True)
+while True:
+    try:
+        conn, _ = s.accept()
+        hdr = recv_exact(conn, 4)
+        if hdr:
+            body = recv_exact(conn, struct.unpack('>I', hdr)[0])
+            try:
+                resp = handle(json.loads(body))
+            except Exception as e:
+                resp = {'ok': False, 'error': '%s: %s' % (type(e).__name__, e)}
+            out = json.dumps(resp).encode()
+            conn.sendall(struct.pack('>I', len(out)) + out)
+        conn.close()
+    except Exception as e:
+        # A bad request must never kill the agent — the VM would look booted
+        # and answer nothing, the worst of both outcomes.
+        print('FC-AGENT-ERROR', e, file=sys.stderr, flush=True)
+PY
+    sudo chmod 0755 /mnt/fcroot/usr/local/bin/fcagent
+    printf '%s\n' '#!/bin/sh' \\
+      'mount -t proc proc /proc 2>/dev/null' \\
+      'mount -t sysfs sys /sys 2>/dev/null' \\
+      'mount -t devtmpfs dev /dev 2>/dev/null' \\
+      'echo FC-GUEST-ALIVE kernel=\$(uname -r) cpus=\$(nproc)' \\
+      'exec /usr/local/bin/fcagent' \\
+      | sudo tee /mnt/fcroot/usr/local/bin/fcinit >/dev/null
+    sudo chmod 0755 /mnt/fcroot/usr/local/bin/fcinit
+    sudo umount /mnt/fcroot
+  " >/dev/null 2>&1 || { fail "$host" "could not bake the guest agent into rootfs"; continue; }
+  pass "$host" "guest agent baked into rootfs"
+
+  # The daemon copies the shared rootfs per VM, so it must be readable by the
+  # daemon user without sudo.
+  ssh "$host" "sudo chmod 0644 '$WORK/rootfs.ext4' && sudo chmod 0644 '$WORK/vmlinux' && sudo mkdir -p '$WORK/vms' && sudo chown \$(id -u):\$(id -g) '$WORK/vms'" >/dev/null 2>&1 \
+    && pass "$host" "rootfs/vmlinux readable, $WORK/vms writable by the daemon user" \
+    || fail "$host" "could not make $WORK usable by the daemon user"
+
+  # 5. Boot one. Installing proves nothing; this is the check that counts.
   #
   # The guest marker is printed by an init script rather than typed at a shell
   # over the serial console — feeding stdin races the shell's startup and
   # arrives half-consumed (observed: `# ho FC-GUEST-ALIVE`, the first two
   # characters eaten).
+  # It does NOT rewrite fcinit — the baked one prints the marker and then execs
+  # the agent, and an earlier version of this script clobbered the agent here,
+  # which would have left every VM booting into a dead end.
+  #
+  # Booted is not the same as reachable, so the check goes all the way to an
+  # agent round trip. It runs as the DAEMON USER, without sudo, because that is
+  # who will actually be starting VMs; the B0 spike passed under sudo and hid a
+  # /dev/kvm permission problem for exactly this reason.
   boot=$(ssh "$host" "
-    set -e
     cd '$WORK'
-    sudo mkdir -p /mnt/fcroot
+    cat > selftest-vm.json <<JSON
-    sudo mount -o loop rootfs.ext4 /mnt/fcroot
-    printf '%s\n' '#!/bin/sh' 'mount -t proc proc /proc 2>/dev/null' \
-      'echo FC-GUEST-ALIVE kernel=\$(uname -r) cpus=\$(nproc)' 'sync' 'reboot -f' \
-      | sudo tee /mnt/fcroot/usr/local/bin/fcinit >/dev/null
-    sudo chmod 0755 /mnt/fcroot/usr/local/bin/fcinit
-    sudo umount /mnt/fcroot
-    cat > vm.json <<JSON
 {
   \"boot-source\": {
     \"kernel_image_path\": \"$WORK/vmlinux\",
     \"boot_args\": \"console=ttyS0 reboot=k panic=1 pci=off init=/usr/local/bin/fcinit\"
   },
-  \"drives\": [{\"drive_id\":\"rootfs\",\"path_on_host\":\"$WORK/rootfs.ext4\",\"is_root_device\":true,\"is_read_only\":false}],
+  \"drives\": [{\"drive_id\":\"rootfs\",\"path_on_host\":\"$WORK/selftest-rootfs.ext4\",\"is_root_device\":true,\"is_read_only\":false}],
-  \"machine-config\": {\"vcpu_count\":2,\"mem_size_mib\":1024,\"smt\":false}
+  \"machine-config\": {\"vcpu_count\":2,\"mem_size_mib\":1024,\"smt\":false},
+  \"vsock\": {\"guest_cid\": 3, \"uds_path\": \"$WORK/selftest.sock\"}
 }
 JSON
-    rm -f boot.log
+    cp --sparse=always rootfs.ext4 selftest-rootfs.ext4
+    rm -f selftest.sock boot.log
     S=\$(date +%s%N)
-    sudo timeout 30 firecracker --no-api --config-file vm.json > boot.log 2>&1 || true
+    setsid timeout 30 firecracker --no-api --config-file selftest-vm.json > boot.log 2>&1 &
+    FCPGID=\$!
+    for i in \$(seq 1 200); do grep -qa FC-AGENT-LISTENING boot.log && break; sleep 0.05; done
     E=\$(date +%s%N)
     echo \"ms=\$(( (E - S) / 1000000 ))\"
-    grep -a FC-GUEST-ALIVE boot.log || echo MARKER-ABSENT
+    python3 - <<'PY' 2>&1 | tail -2
+import socket, struct, json
+s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+s.connect('$WORK/selftest.sock')
+s.sendall(b'CONNECT 9001\n'); s.recv(64)
+req = json.dumps({'op': 'exec', 'cmd': 'echo AGENT-RPC-OK'}).encode()
+s.sendall(struct.pack('>I', len(req)) + req)
+n = struct.unpack('>I', s.recv(4))[0]
+buf = b''
+while len(buf) < n: buf += s.recv(n - len(buf))
+print(json.loads(buf).get('stdout', '').strip())
+PY
+    kill -- -\$FCPGID 2>/dev/null
+    rm -f selftest.sock selftest-rootfs.ext4 selftest-vm.json
+    grep -a FC-GUEST-ALIVE boot.log | head -1 || echo MARKER-ABSENT
   " 2>&1)
 
   case "$boot" in
-    *FC-GUEST-ALIVE*) pass "$host" "microVM booted and ran our code ($(printf '%s' "$boot" | grep -o 'ms=[0-9]*'))" ;;
+    *AGENT-RPC-OK*) pass "$host" "microVM booted and its agent answered over vsock ($(printf '%s' "$boot" | grep -o 'ms=[0-9]*' | head -1))" ;;
+    *FC-GUEST-ALIVE*) fail "$host" "microVM booted but its agent did not answer: $(printf '%s' "$boot" | tail -2 | tr '\n' ' ')" ;;
     *)              fail "$host" "microVM did not boot: $(printf '%s' "$boot" | tail -3 | tr '\n' ' ')" ;;
   esac