Merge: B3 microVM client + fix a wire-contract mismatch that would have timed out silently

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

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

@@ -339,9 +339,15 @@ pub async fn list(vms: &Vms) -> Value {
     json!({ "vms": held })
 }
 
-/// Dispatch a `vm_*` frame. Returns `(ok, output)` in the same shape every
+/// Dispatch a `vm_*` frame.
-/// other node op uses, so this needed no protocol change.
+///
-pub async fn handle_op(op: &str, v: &Value, vms: &Vms) -> (bool, Value) {
+/// 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");
@@ -362,8 +368,8 @@ pub async fn handle_op(op: &str, v: &Value, vms: &Vms) -> (bool, Value) {
         // 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),
+        Ok(out) => (true, out.to_string()),
-        Err(e) => (false, json!({ "error": e })),
+        Err(e) => (false, json!({ "error": e }).to_string()),
     }
 }
 

crates/cm-api/src/fleet.rs

@@ -557,7 +557,19 @@ pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socke
                             );
                         }
                     }
-                    Err(_) => {}
+                    // An unparseable frame used to vanish here. That is the
+                    // worst possible handling: a node op whose reply does not
+                    // match `Uplink` never resolves its pending request, so the
+                    // caller times out after 20s with nothing anywhere saying
+                    // why. Caught exactly that way while wiring the vm_* ops —
+                    // `output` was an object where the wire declares a String.
+                    Err(e) => {
+                        let head: String = t.as_str().chars().take(160).collect();
+                        eprintln!(
+                            "fleet: node {node_id} sent a frame we could not parse ({e}); \
+                             any request it was answering will time out. Frame: {head}"
+                        );
+                    }
                 }
             },
         }

crates/cm-api/src/lib.rs

@@ -21,6 +21,7 @@ pub mod corpus;
 pub mod harvest;
 pub mod library;
 pub mod mission_delivery;
+pub mod microvm_client;
 pub mod mission_fs;
 pub mod papers;
 pub mod phase_config;

crates/cm-api/src/microvm_client.rs

@@ -0,0 +1,295 @@
+//! Drive a fleet node's microVMs from the server.
+//!
+//! Thin by design: the node owns the VM lifecycle (see
+//! `clawmates-node::microvm`), and this is the typed way to ask it. Every call
+//! is one `vm_*` op over the existing `NodeHub` request/response channel, so
+//! there is no new transport, correlation or timeout machinery.
+//!
+//! # Not a `SandboxDriver`
+//!
+//! `RemoteDriver` exists to marshal `SandboxDriver` over the hub, and reusing it
+//! was the plan. That trait is container-shaped — `attach_pty`, `resize_pty`,
+//! argv `exec` — while a mission needs inject → run → collect. Conforming would
+//! mean implementing PTY-over-vsock semantics that nothing calls, so this speaks
+//! the smaller interface the mission path actually uses.
+//!
+//! # Timeouts
+//!
+//! The hub defaults to 20s, which is right for a create (measured: ~1s) and
+//! badly wrong for an agent turn. `exec` therefore takes its own budget and
+//! passes it to BOTH the hub and the guest, with the hub's slightly longer: if
+//! the guest's own timeout fires first the reply says so, whereas a hub timeout
+//! leaves us guessing whether the command is still running.
+
+use cm_domain::NodeId;
+use serde_json::{json, Value};
+
+use crate::fleet::NodeHub;
+
+/// Slack between the guest's deadline and the hub's, so the guest's own timeout
+/// wins the race and we get a real answer rather than a transport error.
+const HUB_GRACE_SECS: u64 = 30;
+
+/// How long the hub waits for a command whose own budget is `guest_secs`.
+///
+/// Saturating, not `+`: a caller passing a very large budget would otherwise
+/// overflow and panic in debug or wrap to a tiny timeout in release — the second
+/// being far worse, since it turns a long-running agent turn into a spurious
+/// transport failure.
+fn hub_deadline(guest_secs: u64) -> u64 {
+    guest_secs.saturating_add(HUB_GRACE_SECS)
+}
+
+pub struct MicroVm<'a> {
+    hub: &'a NodeHub,
+    node_id: NodeId,
+    vm_id: String,
+}
+
+impl<'a> MicroVm<'a> {
+    pub fn new(hub: &'a NodeHub, node_id: NodeId, vm_id: impl Into<String>) -> Self {
+        Self {
+            hub,
+            node_id,
+            vm_id: vm_id.into(),
+        }
+    }
+
+    pub fn vm_id(&self) -> &str {
+        &self.vm_id
+    }
+
+    /// One op, with the node's `output` string parsed back into JSON.
+    ///
+    /// `output` is a String on the wire (`Uplink::Result`), and a node that
+    /// answered with a JSON object instead made the whole frame unparseable —
+    /// the reply then vanished into the uplink's error arm and the call timed
+    /// out with nothing explaining why. Parsing here, loudly, keeps that
+    /// mismatch a visible error rather than a mystery timeout.
+    async fn call(&self, op: &str, mut args: Value, secs: u64) -> Result<Value, String> {
+        if let Some(o) = args.as_object_mut() {
+            o.insert("vm_id".into(), Value::String(self.vm_id.clone()));
+        }
+        let out = self
+            .hub
+            .call_timeout(self.node_id, op, args, secs)
+            .await
+            .map_err(|e| format!("{op} on node {:?}: {e}", self.node_id))?;
+        let body: Value = serde_json::from_str(&out.output)
+            .map_err(|e| format!("{op} returned unparseable output ({e}): {}", out.output))?;
+        if !out.ok {
+            let why = body
+                .get("error")
+                .and_then(Value::as_str)
+                .unwrap_or(&out.output);
+            return Err(format!("{op} failed: {why}"));
+        }
+        Ok(body)
+    }
+
+    /// Boot the VM. Returns only once its guest agent has answered.
+    pub async fn create(&self, vcpus: u32, mem_mib: u32) -> Result<Value, String> {
+        // 60s, not the hub default: a create that has to copy a rootfs and boot
+        // is measured near 1s, but a node under load has no reason to be fast.
+        self.call(
+            "vm_create",
+            json!({ "vcpus": vcpus, "mem_mib": mem_mib }),
+            60,
+        )
+        .await
+    }
+
+    /// Unpack a tar inside the guest at `dest`.
+    ///
+    /// Takes the archive bytes rather than a path: the server holds the mission
+    /// checkout, the node does not, and shipping the tar is the whole point of
+    /// the inject → run → collect model.
+    pub async fn inject(&self, dest: &str, tar: &[u8]) -> Result<Value, String> {
+        use base64::Engine as _;
+        let b64 = base64::engine::general_purpose::STANDARD.encode(tar);
+        self.call("vm_inject", json!({ "dest": dest, "tar_b64": b64 }), 120)
+            .await
+    }
+
+    /// Run a shell command in the guest.
+    ///
+    /// `Ok` means the command RAN; the exit code is in the payload. A non-zero
+    /// exit is not an error here — the caller has to be able to tell "the build
+    /// failed" from "we could not reach the VM", and collapsing them is the
+    /// defect this codebase keeps paying for.
+    pub async fn exec(
+        &self,
+        cmd: &str,
+        cwd: Option<&str>,
+        timeout_secs: u64,
+    ) -> Result<ExecOut, String> {
+        let v = self
+            .call(
+                "vm_exec",
+                json!({ "cmd": cmd, "cwd": cwd, "timeout": timeout_secs }),
+                hub_deadline(timeout_secs),
+            )
+            .await?;
+        // A missing rc is not "success" — it means the guest did not report one,
+        // which we must not read as zero.
+        let rc = v
+            .get("rc")
+            .and_then(Value::as_i64)
+            .ok_or_else(|| format!("vm_exec gave no exit code: {v}"))?;
+        Ok(ExecOut {
+            rc,
+            stdout: v
+                .get("stdout")
+                .and_then(Value::as_str)
+                .unwrap_or_default()
+                .to_string(),
+            stderr: v
+                .get("stderr")
+                .and_then(Value::as_str)
+                .unwrap_or_default()
+                .to_string(),
+        })
+    }
+
+    /// Tar a path out of the guest and return the archive bytes.
+    pub async fn collect(&self, path: &str) -> Result<Vec<u8>, String> {
+        use base64::Engine as _;
+        let v = self.call("vm_collect", json!({ "path": path }), 300).await?;
+        // The guest reports its own `ok`: a missing path is a real failure that
+        // must not come back as an empty archive, which would look exactly like
+        // a run that produced nothing.
+        if v.get("ok").and_then(Value::as_bool) != Some(true) {
+            return Err(format!(
+                "vm_collect {path}: {}",
+                v.get("error").and_then(Value::as_str).unwrap_or("unknown")
+            ));
+        }
+        let b64 = v
+            .get("tar_b64")
+            .and_then(Value::as_str)
+            .ok_or_else(|| format!("vm_collect {path} returned no archive: {v}"))?;
+        base64::engine::general_purpose::STANDARD
+            .decode(b64)
+            .map_err(|e| format!("vm_collect {path}: undecodable archive: {e}"))
+    }
+
+    /// Stop the VM and remove everything it owned. Idempotent.
+    pub async fn destroy(&self) -> Result<Value, String> {
+        self.call("vm_destroy", json!({}), 60).await
+    }
+}
+
+/// The result of a command that RAN. `rc != 0` is a normal outcome.
+#[derive(Debug, Clone)]
+pub struct ExecOut {
+    pub rc: i64,
+    pub stdout: String,
+    pub stderr: String,
+}
+
+impl ExecOut {
+    pub fn ok(&self) -> bool {
+        self.rc == 0
+    }
+    /// One line for a log or an artifact, without dumping a whole build.
+    pub fn summary(&self) -> String {
+        let tail = |s: &str| {
+            s.lines()
+                .rev()
+                .take(3)
+                .collect::<Vec<_>>()
+                .into_iter()
+                .rev()
+                .collect::<Vec<_>>()
+                .join(" | ")
+        };
+        if self.ok() {
+            format!("rc=0 {}", tail(&self.stdout))
+        } else {
+            format!("rc={} {}", self.rc, tail(&self.stderr))
+        }
+    }
+}
+
+/// VMs a node currently holds, so orphans can be reaped.
+pub async fn list(hub: &NodeHub, node_id: NodeId) -> Result<Vec<String>, String> {
+    let out = hub
+        .call(node_id, "vm_list", json!({}))
+        .await
+        .map_err(|e| format!("vm_list on node {node_id:?}: {e}"))?;
+    let body: Value = serde_json::from_str(&out.output)
+        .map_err(|e| format!("vm_list returned unparseable output ({e}): {}", out.output))?;
+    Ok(body
+        .get("vms")
+        .and_then(Value::as_array)
+        .map(|a| {
+            a.iter()
+                .filter_map(|v| v.get("vm_id").and_then(Value::as_str))
+                .map(str::to_string)
+                .collect()
+        })
+        .unwrap_or_default())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A command that ran and failed must be distinguishable from one that
+    /// could not be reached. `rc` carries the verdict; `Err` is for transport.
+    #[test]
+    fn a_nonzero_exit_is_an_outcome_not_an_error() {
+        let failed = ExecOut {
+            rc: 3,
+            stdout: String::new(),
+            stderr: "boom\n".into(),
+        };
+        assert!(!failed.ok());
+        assert!(failed.summary().starts_with("rc=3"));
+        assert!(failed.summary().contains("boom"));
+
+        let passed = ExecOut {
+            rc: 0,
+            stdout: "fine\n".into(),
+            stderr: String::new(),
+        };
+        assert!(passed.ok());
+        assert_eq!(passed.summary(), "rc=0 fine");
+    }
+
+    /// The summary is for logs, so it must stay short even when a build prints
+    /// thousands of lines — and it must keep the LAST lines, where the error is.
+    #[test]
+    fn the_summary_keeps_the_tail_and_stays_short() {
+        let noisy = ExecOut {
+            rc: 1,
+            stdout: String::new(),
+            stderr: (1..=500)
+                .map(|i| format!("line {i}"))
+                .collect::<Vec<_>>()
+                .join("\n"),
+        };
+        let s = noisy.summary();
+        assert!(s.contains("line 500"), "the last line must survive: {s}");
+        assert!(!s.contains("line 400"), "older lines must be dropped: {s}");
+        assert!(s.len() < 200, "summary must stay log-sized, got {}", s.len());
+    }
+
+    /// The guest's deadline must fire before the hub's, so a slow command comes
+    /// back as a reported timeout rather than an unexplained transport failure.
+    #[test]
+    fn the_hub_always_outlives_the_guests_own_timeout() {
+        for guest in [0u64, 1, 30, 3600, 86_400] {
+            assert!(
+                hub_deadline(guest) > guest,
+                "hub deadline for {guest}s must exceed it"
+            );
+        }
+        // A caller passing a huge budget must not wrap to a tiny timeout, which
+        // would turn a long agent turn into a spurious transport failure.
+        assert!(
+            hub_deadline(u64::MAX) >= u64::MAX - 1,
+            "an extreme budget must saturate, not wrap"
+        );
+    }
+}