feat(fleet): B2 — vm_* node ops for Firecracker microVMs

create / inject / exec / collect / destroy / list, riding the node's
existing frame dispatch ({t, id, …} -> {t:"result", id, ok, output}), so
no protocol change was needed. Control is length-prefixed JSON over
vsock; the serial console stays a log, because feeding a guest over stdin
races its startup and arrives half-consumed.

DEVIATION FROM THE PLAN, deliberately: this does NOT implement
cm_sandbox::SandboxDriver. That trait is container-shaped —
attach_pty/resize_pty/argv exec — while missions need
create -> inject -> run -> collect -> destroy. Conforming would mean
building PTY-over-vsock and window-resize semantics that no mission path
calls, purely to satisfy a signature. We give up automatic RemoteDriver
marshalling; orphan reaping is a label/id sweep either way.

Three traps from the B0 spike are handled in code rather than remembered:

  - Firecracker does NOT unlink its vsock UDS on exit, so destroy unlinks
    it explicitly, and the selftest ASSERTS it is gone. Assuming the VM
    tidies up after itself is how the mission checkout accumulated four
    uid bugs.
  - firecracker is spawned via setsid and killed as a process GROUP, so a
    background child cannot outlive the VM holding its workdir open.
  - create does not return until the guest agent has answered a ping. A
    VM that booted but serves nothing is worse than one that failed, so a
    half-created VM is destroyed rather than left registered.

A vm id becomes a path component, so ids are restricted to [A-Za-z0-9_-]
and REJECTED rather than sanitised — a caller that sent `../../etc`
wanted something we should not guess at.

Verified on tank through the real Rust path, as the daemon user, with no
sudo: `clawmates-node --vm-selftest` -> 8/8, create in 986ms, and the
host left with zero firecracker processes and zero VM directories. The
selftest asserts every step, including that a destroyed VM can no longer
be exec'd; a test that only reports the steps it completed cannot
distinguish "passed" from "stopped early".

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