fix(missions): exclude build output from COLLECT too, not just inject

The other half of the same bug. The previous commit filtered `mission_fs::pack_dir`
(the inject side) and left the guest's `op_get` tarring everything, so the re-run
that proved the #54 fix — it survived 480s where it used to die at 210 — still lost
its work to `vm_collect ... node timed out`. Two modules written, four subagents
used, nothing delivered.

`op_get` now takes an `exclude` list, sent by the host from
`mission_fs::transport_excludes()` — the same list `mission_delivery` uses for the
diff. Policy in one place, applied at both ends of the wire. Matched on directory
NAME at any depth, so a workspace's per-crate `target/` dirs are all covered, with
a test that plants a nested one and asserts it does not come along.

Also proven by that run: the worker no longer kills a live microVM run. It ran 480
seconds straight through the 180s requeue window and the 210s mark where mission
019fd43e died, untouched. And `subagents: 4` — the team addendum did drive real
fan-out this time, which is the first evidence the Slice 3 switch does anything.

483 tests pass, clippy clean. Still to prove: a >3-minute mission that actually
DELIVERS. The collect fix is tested in isolation but has not yet carried a real
mission's work back, and the guest agent needs rebuilding into the rootfs before it
can.

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

@@ -427,9 +427,14 @@ pub async fn exec(
 }
 
 /// Tar a path out of the guest.
-pub async fn collect(vms: &Vms, vm_id: &str, path: &str) -> Result<Value, String> {
+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 })).await
+    rpc(&uds, &json!({ "op": "get", "path": path, "exclude": exclude })).await
 }
 
 /// SIGKILL a whole process group, ignoring "already gone".
@@ -565,7 +570,7 @@ pub async fn handle_op(op: &str, v: &Value, vms: &Vms) -> (bool, String) {
             )
             .await
         }
-        "vm_collect" => collect(vms, &vm_id, &s("path")).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}")),
@@ -801,7 +806,7 @@ pub async fn selftest() -> bool {
 
     // 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").await;
+    let r = collect(&vms, id, "/work", None).await;
     let round_tripped = r
         .as_ref()
         .ok()

crates/bins/fcagent/src/main.rs

@@ -469,6 +469,46 @@ fn op_put(req: &Value) -> Value {
     }
 }
 
+/// Recursive tar append that skips excluded directory NAMES at any depth.
+///
+/// Hand-rolled because `tar::Builder::append_dir_all` takes no filter. Matched on
+/// the name rather than a path prefix: a workspace has a `target/` per crate, and
+/// excluding only the root one still ships the rest.
+fn append_filtered<W: Write>(
+    b: &mut tar::Builder<W>,
+    dir: &Path,
+    prefix: &Path,
+    exclude: &[String],
+) -> std::io::Result<()> {
+    b.append_dir(prefix, dir)?;
+    let mut entries: Vec<_> = std::fs::read_dir(dir)?.collect::<Result<Vec<_>, _>>()?;
+    entries.sort_by_key(|e| e.file_name());
+    for entry in entries {
+        let name = entry.file_name();
+        let name_str = name.to_string_lossy().to_string();
+        let path = entry.path();
+        let dest = prefix.join(&name);
+        let meta = std::fs::symlink_metadata(&path)?;
+        if meta.is_dir() {
+            if exclude.contains(&name_str) {
+                continue;
+            }
+            append_filtered(b, &path, &dest, exclude)?;
+        } else if meta.is_symlink() {
+            let mut header = tar::Header::new_gnu();
+            header.set_metadata(&meta);
+            header.set_entry_type(tar::EntryType::Symlink);
+            header.set_size(0);
+            let target = std::fs::read_link(&path)?;
+            b.append_link(&mut header, &dest, &target)?;
+        } else {
+            let mut f = std::fs::File::open(&path)?;
+            b.append_file(&dest, &mut f)?;
+        }
+    }
+    Ok(())
+}
+
 fn op_get(req: &Value) -> Value {
     let path = req.get("path").and_then(Value::as_str).unwrap_or_default();
     if path.is_empty() {
@@ -485,12 +525,28 @@ fn op_get(req: &Value) -> Value {
         .map(|s| s.to_string_lossy().to_string())
         .unwrap_or_else(|| "root".to_string());
 
+    // Directory names to leave out, sent by the host so the policy lives in one
+    // place (`mission_fs::transport_excludes`). Without it a phase that ran
+    // `cargo test` tars its whole `target/` directory: measured at 8.9 MB of 9.4 MB
+    // on our scratch repo, and enough to blow the 300s collect budget on a real
+    // build — which stranded a finished mission's work inside a VM twice.
+    let exclude: Vec<String> = req
+        .get("exclude")
+        .and_then(Value::as_array)
+        .map(|a| {
+            a.iter()
+                .filter_map(Value::as_str)
+                .map(str::to_string)
+                .collect()
+        })
+        .unwrap_or_default();
+
     let mut b = tar::Builder::new(Vec::new());
     // Do not follow symlinks: a link pointing outside the collected tree would
     // otherwise be dereferenced and its target smuggled back to the host.
     b.follow_symlinks(false);
     let added = if p.is_dir() {
-        b.append_dir_all(&name, p)
+        append_filtered(&mut b, p, Path::new(&name), &exclude)
     } else {
         b.append_path_with_name(p, &name)
     };
@@ -641,6 +697,62 @@ mod tests {
 
     /// A missing path must be an error, not an empty archive: an empty tar is
     /// indistinguishable from a run that produced nothing.
+    /// Build output is not work. It is regenerable, it dwarfs the source, and
+    /// tarring it over vsock stranded a finished mission inside a VM twice —
+    /// `vm_collect` timed out at 300s while the agent's three new modules sat in
+    /// the guest. Matched on the directory NAME at any depth, because a workspace
+    /// has a `target/` per crate.
+    #[test]
+    fn excluded_directories_stay_out_of_the_archive_at_any_depth() {
+        let dir = std::env::temp_dir().join(format!("fcagent-ex-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(dir.join("src")).unwrap();
+        std::fs::create_dir_all(dir.join("target/debug")).unwrap();
+        std::fs::create_dir_all(dir.join("crates/inner/target")).unwrap();
+        std::fs::write(dir.join("src/lib.rs"), "fn a() {}").unwrap();
+        std::fs::write(dir.join("target/debug/blob"), vec![0u8; 4096]).unwrap();
+        std::fs::write(dir.join("crates/inner/target/blob"), vec![0u8; 4096]).unwrap();
+        std::fs::write(dir.join("crates/inner/keep.rs"), "fn b() {}").unwrap();
+
+        let r = op_get(&json!({
+            "op": "get",
+            "path": dir.to_string_lossy(),
+            "exclude": ["target"],
+        }));
+        assert_eq!(r["ok"], json!(true), "{r}");
+        let bytes = B64.decode(r["tar_b64"].as_str().unwrap()).unwrap();
+        let mut ar = tar::Archive::new(&bytes[..]);
+        let paths: Vec<String> = ar
+            .entries()
+            .unwrap()
+            .filter_map(Result::ok)
+            .map(|e| e.path().unwrap().to_string_lossy().to_string())
+            .collect();
+        let _ = std::fs::remove_dir_all(&dir);
+
+        assert!(paths.iter().any(|p| p.ends_with("src/lib.rs")), "{paths:?}");
+        assert!(paths.iter().any(|p| p.ends_with("inner/keep.rs")), "{paths:?}");
+        assert!(
+            !paths.iter().any(|p| p.contains("target")),
+            "a nested target/ came along: {paths:?}"
+        );
+    }
+
+    /// No exclude list means everything, so an existing caller is unchanged.
+    #[test]
+    fn without_an_exclude_list_nothing_is_dropped() {
+        let dir = std::env::temp_dir().join(format!("fcagent-noex-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(dir.join("target")).unwrap();
+        std::fs::write(dir.join("target/x"), "x").unwrap();
+        let r = op_get(&json!({ "op": "get", "path": dir.to_string_lossy() }));
+        let bytes = B64.decode(r["tar_b64"].as_str().unwrap()).unwrap();
+        let mut ar = tar::Archive::new(&bytes[..]);
+        let n = ar.entries().unwrap().filter_map(Result::ok).count();
+        let _ = std::fs::remove_dir_all(&dir);
+        assert!(n >= 2, "expected the target dir and its file, got {n}");
+    }
+
     #[test]
     fn getting_a_missing_path_is_an_error() {
         let r = op_get(&json!({ "op": "get", "path": "/definitely/not/here" }));

crates/cm-api/src/microvm_client.rs

@@ -184,9 +184,15 @@ impl<'a> MicroVm<'a> {
     }
 
     /// Tar a path out of the guest and return the archive bytes.
-    pub async fn collect(&self, path: &str) -> Result<Vec<u8>, String> {
+    /// `exclude` names directories to leave out — build output, caches. Sent from
+    /// here so the policy lives in one place: `mission_fs::transport_excludes`,
+    /// the same list the delivery diff uses. Shipping `target/` blew this call's
+    /// 300s budget twice, each time with the agent's work finished and stranded.
+    pub async fn collect(&self, path: &str, exclude: &[&str]) -> Result<Vec<u8>, String> {
         use base64::Engine as _;
-        let v = self.call("vm_collect", json!({ "path": path }), 300).await?;
+        let v = self
+            .call("vm_collect", json!({ "path": path, "exclude": exclude }), 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.

crates/cm-api/src/microvm_executor.rs

@@ -479,7 +479,7 @@ async fn run_inside(
     // Collect regardless of the agent's exit code. A turn that failed partway
     // still wrote files, and throwing them away because the CLI exited non-zero
     // would discard exactly the work a retry needs to see.
-    let tar = vm.collect(GUEST_REPO).await;
+    let tar = vm.collect(GUEST_REPO, crate::mission_fs::transport_excludes()).await;
     let collected = match tar {
         Ok(bytes) => {
             let parent = repo