feat(missions): Slice 4 — the two engines composed, with the file handoff proven

`team_engine='composed'` (the third name migration 0069 anticipated) runs a
mission as a durable ZeroClaw graph whose every node is a whole
Claude-Code-in-a-microVM session. Engine Z owns checkpoint/resume, cancellation
and per-node heterogeneity; Engine C owns shared context and cheap fan-out;
neither has the other's asset, which is why this is a composition and not a
compromise.

`MicroVmTurnExecutor` implements the existing `TurnExecutor`, so it inherits the
planners, the checkpoint, the stale-run recovery, `close_finished_phases`, the
evaluator, capture and delivery unchanged — the same trick `SubTopologyExecutor`
already plays with a heavy `run_turn`. Producer side emits ONE `queued` row
carrying the real graph and lets the worker claim it: the durability IS being
worker-driven, and the solo path's `tokio::spawn` has none of it. Still exactly
one `topology_runs` row per unit of work and one completion path — `finish()` is
now that one place, shared by every tier.

THE TRAP, solved and proven. A VM is inject → run → collect → destroy, so a
per-node VM with text-only handoff silently loses every file an earlier node
wrote: node 2 boots from the original checkout, sees nothing, and still reports
success. The mission's host checkout is the medium — every node injects from it
and collects back over it — and two properties make that safe rather than lucky:
`execute_resumable` is strictly sequential, so two VMs never write one directory;
and the vm id is deterministic per (phase, iteration, step), so a duplicate is
refused by the node ("vm already exists") instead of becoming a second writer.

NEGATIVE CONTROL, run rather than assumed: with `repo` swapped for a private
per-node workspace, `a_later_node_sees_an_earlier_nodes_files` FAILS with
`saw:[]`; restored, it passes. The `PhaseVm` seam exists for exactly this — it
models inject/collect through the real `mission_fs` tar path in milliseconds.

Two durability traps this tier walks into, both closed:

  - `requeue_stale` fires at 180s on `updated_at`, and one node here can run for
    an hour. `SubTopologyExecutor` keeps its parent alive from each leaf step;
    there is nothing between the start and end of a VM turn, so the turn holds a
    ticker that touches `updated_at` every 30s and aborts on drop. Without it a
    healthy composed run is requeued mid-node and boots a second VM.
  - the 15-minute stuck-run reaper asks "any step records since it was CREATED?",
    which describes a healthy composed run as readily as a wedged one. Hence
    `REAPABLE_TIERS` — worker-driven minus this tier. Reaping it would be #54 in
    a different costume.

`on_launch` mints no team for a microVM mission, deliberately: claws in
containers are what a VM mission does not use. So `mission_orchestrator::
composed_graph` builds the shape from the team template directly — nodes, roles
and pattern, zero claws provisioned. Per-node `attrs["backend"]` and
`attrs["node_id"]` override the mission's, which is what makes a validator node
on another provider's image a first-class graph node; a malformed `node_id`
fails the node rather than quietly running it where the graph did not ask.

Refusals are recorded as a failed run, not returned as an error: `launch_phase`
is swept every ten seconds, so a returned error is a phase that retries forever
while the log repeats itself.

501 tests pass, clippy clean. NOT yet proven end to end: no composed mission has
run on the fleet, so the resume-after-a-killed-worker leg is argued from the DB
test and the step-numbering test, not from a real two-node run.

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

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

@@ -280,6 +280,9 @@ async fn run() -> Result<(), String> {
     cm_api::topology_worker::spawn(
         pool.clone(),
         runtime.clone(),
+        // The composed tier (`microvm_graph`) runs each graph node as a VM on a
+        // fleet node, so the worker needs the same hub the phase runner uses.
+        node_hub.clone(),
         std::time::Duration::from_secs(3),
     );
     // Boot-time content loaders — skills first, then team templates

crates/cm-api/src/lib.rs

@@ -23,6 +23,7 @@ pub mod library;
 pub mod mission_delivery;
 pub mod microvm_client;
 pub mod microvm_executor;
+pub mod microvm_turn_executor;
 pub mod mission_fs;
 pub mod papers;
 pub mod phase_config;

crates/cm-api/src/microvm_executor.rs

@@ -57,8 +57,20 @@ const GUEST_REPO: &str = "/mission/repo";
 /// A vm id must be `[A-Za-z0-9_-]` — the node rejects anything else, since it
 /// becomes a path component. Hyphenated uuids qualify; this also keeps the id
 /// readable in `vm_list` output and in the node's egress log.
-fn vm_id_for(phase_id: Uuid, iteration: i32) -> String {
+///
-    format!("m-{}-{}", &phase_id.simple().to_string()[..12], iteration)
+/// `step` distinguishes the nodes of a composed run, whose graph runs several VMs
+/// for one phase and one iteration. Deliberately DETERMINISTIC rather than
+/// random: the node refuses to create a vm id that already exists, so if a
+/// restarted worker resumes a step whose VM is somehow still alive, the second
+/// attempt fails loudly instead of running a duplicate agent against the same
+/// checkout. A random id would make that collision invisible and let two VMs
+/// collect over each other's work.
+fn vm_id_for(phase_id: Uuid, iteration: i32, step: Option<u32>) -> String {
+    let base = format!("m-{}-{}", &phase_id.simple().to_string()[..12], iteration);
+    match step {
+        Some(s) => format!("{base}-s{s}"),
+        None => base,
+    }
 }
 
 /// Ceiling on teammates, stated in the prompt.
@@ -337,7 +349,7 @@ pub async fn run_phase_in_vm(hub: &NodeHub, p: VmPhase<'_>) -> Result<VmOutcome,
     // phase, not boot a VM whose agent will sit there unauthenticated.
     let env = crate::mission_runtime::microvm_provider_env(p.backend)?;
 
-    let vm = MicroVm::new(hub, p.node_id, vm_id_for(p.phase_id, p.iteration));
+    let vm = MicroVm::new(hub, p.node_id, vm_id_for(p.phase_id, p.iteration, p.step));
     let created = vm.create(VCPUS, MEM_MIB, p.backend).await?;
 
     // From here on every early return must still destroy the VM, so the work is
@@ -374,6 +386,41 @@ pub struct VmPhase<'a> {
     /// `missions.team_engine` — `Some("claude_code")` asks the lead to form a
     /// team. `None` is solo, which is the default.
     pub team_engine: Option<&'a str>,
+    /// Which node of a composed graph this VM is running, if any. `None` is the
+    /// solo path, where the phase is one VM and the id needs no further
+    /// qualification. Part of the vm id, so the nodes of one phase-iteration
+    /// cannot collide on a fleet node.
+    pub step: Option<u32>,
+}
+
+/// Runs one phase (or one graph node of one) in a VM.
+///
+/// A seam with exactly one production implementation, and it exists for one
+/// reason: the composed executor's defining property is that node 2 sees node 1's
+/// files, and that property is untestable against real Firecracker in a unit
+/// test. A fake that models inject → run → collect faithfully can prove it in
+/// milliseconds, including the negative control where the handoff is broken.
+#[allow(async_fn_in_trait)]
+pub trait PhaseVm {
+    /// Boot a VM, run this phase in it, collect the result, destroy it.
+    async fn run(&self, p: VmPhase<'_>) -> Result<VmOutcome, String>;
+}
+
+/// The real thing: VMs on the fleet, over the node hub.
+pub struct HubVms {
+    hub: std::sync::Arc<NodeHub>,
+}
+
+impl HubVms {
+    pub fn new(hub: std::sync::Arc<NodeHub>) -> Self {
+        Self { hub }
+    }
+}
+
+impl PhaseVm for HubVms {
+    async fn run(&self, p: VmPhase<'_>) -> Result<VmOutcome, String> {
+        run_phase_in_vm(&self.hub, p).await
+    }
 }
 
 async fn run_inside(
@@ -516,13 +563,15 @@ mod tests {
     /// outside `[A-Za-z0-9_-]` rather than sanitising it.
     #[test]
     fn a_vm_id_is_acceptable_to_the_node() {
-        let id = vm_id_for(Uuid::now_v7(), 3);
+        for step in [None, Some(0), Some(11)] {
-        assert!(
+            let id = vm_id_for(Uuid::now_v7(), 3, step);
-            id.chars()
+            assert!(
-                .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'),
+                id.chars()
-            "{id}"
+                    .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'),
-        );
+                "{id}"
-        assert!(id.len() <= 64, "{id}");
+            );
+            assert!(id.len() <= 64, "{id}");
+        }
     }
 
     /// Two passes of the same phase must not collide: the second would fail with
@@ -530,7 +579,22 @@ mod tests {
     #[test]
     fn each_iteration_gets_its_own_vm() {
         let p = Uuid::now_v7();
-        assert_ne!(vm_id_for(p, 1), vm_id_for(p, 2));
+        assert_ne!(vm_id_for(p, 1, None), vm_id_for(p, 2, None));
+    }
+
+    /// A composed run boots one VM per graph node within a single phase and
+    /// iteration. Without the step in the id they would all be the same vm, and
+    /// the second node would fail with "vm already exists" — or worse, if the
+    /// first were already destroyed, succeed while looking like a re-run.
+    #[test]
+    fn each_graph_node_gets_its_own_vm() {
+        let p = Uuid::now_v7();
+        assert_ne!(vm_id_for(p, 1, Some(0)), vm_id_for(p, 1, Some(1)));
+        // And a composed node never collides with the solo id for the same pass.
+        assert_ne!(vm_id_for(p, 1, Some(0)), vm_id_for(p, 1, None));
+        // Deterministic: resuming the same step asks for the same vm, which is
+        // what makes a still-live duplicate fail loudly on the node.
+        assert_eq!(vm_id_for(p, 1, Some(2)), vm_id_for(p, 1, Some(2)));
     }
 
     /// A task description contains quotes and newlines as a matter of course,
@@ -683,6 +747,22 @@ mod tests {
         for t in ["team", "swarm"] {
             assert!(cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&t), "{t}");
         }
+        // The composed tier is the opposite case and must not be confused with
+        // the solo one: its durability comes FROM being worker-driven.
+        assert!(cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&"microvm_graph"));
+        // But the 15-minute stuck-run reaper must not touch it: one of its nodes
+        // is a whole agent session, so journaling nothing for 15 minutes is what
+        // a healthy composed run looks like.
+        assert!(
+            !cm_db::repo::topology_runs::REAPABLE_TIERS.contains(&"microvm_graph"),
+            "the reaper would kill a healthy composed run and orphan its VM"
+        );
+        for t in cm_db::repo::topology_runs::REAPABLE_TIERS {
+            assert!(
+                cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(t),
+                "{t} is reapable but not worker-driven"
+            );
+        }
     }
 
     /// Solo is the default, and asking for a team must be explicit. Multi-agent

crates/cm-api/src/microvm_turn_executor.rs

@@ -0,0 +1,579 @@
+//! The two engines composed — Slice 4.
+//!
+//! Engine Z (the ZeroClaw graph in `cm_orchestrator`) owns durability and
+//! heterogeneity: deterministic planners, per-step checkpoint/resume, a stale-run
+//! sweep, cancellation, and a different model per node. Engine C (Claude Code in
+//! a microVM) owns shared context, self-sizing and cheap fan-out. Neither has the
+//! other's asset, which is why keeping both is a composition rather than a
+//! compromise.
+//!
+//! This module is the join: a [`TurnExecutor`] whose "turn" is a whole
+//! Claude-Code-in-a-VM session. Because `topology_worker` already dispatches by
+//! tier, implementing the existing trait inherits the planners, checkpointing,
+//! reaper, cancellation, `close_finished_phases`, evaluation, capture and
+//! delivery unchanged. `recursive_exec::SubTopologyExecutor` is the precedent: a
+//! `run_turn` may be arbitrarily heavy.
+//!
+//! # The file-handoff trap
+//!
+//! A VM is inject-tar → run → collect-tar → destroy. A graph of per-node VMs with
+//! **text-only** handoff would silently lose every file an earlier node wrote:
+//! node 2 would boot from the original checkout, see none of node 1's work, and
+//! still report success — the exact silent-success shape this project keeps
+//! paying for.
+//!
+//! The answer here is that the mission's **host checkout is the medium**. Every
+//! node injects from `repo` and collects back over `repo`, so the tree carries
+//! forward node to node and the last node's tree is what delivery diffs. Two
+//! properties make that safe rather than lucky:
+//!
+//! - `execute_resumable` runs steps strictly **sequentially**, so two VMs are
+//!   never writing the same host directory at once;
+//! - the vm id is deterministic per (phase, iteration, step), so a resumed step
+//!   whose VM is somehow still alive is refused by the node ("vm already exists")
+//!   instead of quietly producing a second writer.
+//!
+//! `a_later_node_sees_an_earlier_nodes_files` proves the handoff, and
+//! `text_only_handoff_loses_the_earlier_nodes_work` is its negative control.
+//!
+//! # Keeping a long turn alive
+//!
+//! `requeue_stale` requeues a `running` job that has not touched `updated_at` in
+//! 180 seconds, and one node here can run for an hour. `SubTopologyExecutor`
+//! keeps its parent alive from each *leaf step*, which it has and this does not:
+//! there is nothing between the start and end of a VM turn. So the turn holds a
+//! ticker that touches `updated_at` every [`KEEPALIVE_SECS`] and is aborted on
+//! drop. Without it a healthy composed run is requeued mid-node, claimed again,
+//! and boots a second VM against the same checkout.
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+
+use cm_domain::NodeId;
+use cm_orchestrator::{OrchestratorError, TurnExecutor, TurnOutcome, TurnRequest};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::microvm_executor::{PhaseVm, VmPhase};
+
+/// How often a running VM turn touches its run's `updated_at`.
+///
+/// Comfortably inside the 180s stale window, and cheap: one UPDATE per node per
+/// half minute against a row nothing else is writing.
+const KEEPALIVE_SECS: u64 = 30;
+
+/// A [`TurnExecutor`] that runs each graph node as a full Claude-Code session
+/// inside its own microVM, against the mission's shared host checkout.
+pub struct MicroVmTurnExecutor<V: PhaseVm> {
+    vms: V,
+    pool: PgPool,
+    /// The durable outer run. Touched for keepalive; its status gates the turn.
+    run_id: Uuid,
+    mission_id: Uuid,
+    phase_id: Uuid,
+    iteration: i32,
+    /// The mission's host checkout — injected into every node's VM and collected
+    /// back over, which is how file work survives a node boundary.
+    repo: PathBuf,
+    /// `missions.target_node_id`: the fleet node a mission was placed on. A node
+    /// may override it with `attrs["node_id"]`.
+    default_fleet_node: Option<Uuid>,
+    /// `missions.backend`: which rootfs image. A node may override it with
+    /// `attrs["backend"]`, which is what makes a graph heterogeneous — a
+    /// `validator` node on a different provider's image is then a first-class
+    /// graph node rather than a bolt-on.
+    default_backend: Option<String>,
+    /// `missions.team_engine`, passed through so a composed node can itself ask
+    /// for Claude Code fan-out inside its VM.
+    team_engine: Option<String>,
+    /// Which step is next. `execute_resumable` is sequential and gives the
+    /// executor no index, so the executor counts — and the count starts from the
+    /// checkpoint on resume, or two VMs would share an id across a restart.
+    step: std::sync::atomic::AtomicU32,
+}
+
+/// Everything a composed run needs that is not the graph itself.
+pub struct ComposedRun {
+    pub run_id: Uuid,
+    pub mission_id: Uuid,
+    pub phase_id: Uuid,
+    pub iteration: i32,
+    pub repo: PathBuf,
+    pub target_node_id: Option<Uuid>,
+    pub backend: Option<String>,
+    pub team_engine: Option<String>,
+    /// Steps already completed, from the durable checkpoint. Nonzero on resume.
+    pub completed_steps: u32,
+}
+
+impl<V: PhaseVm> MicroVmTurnExecutor<V> {
+    pub fn new(vms: V, pool: PgPool, r: ComposedRun) -> Self {
+        Self {
+            vms,
+            pool,
+            run_id: r.run_id,
+            mission_id: r.mission_id,
+            phase_id: r.phase_id,
+            iteration: r.iteration,
+            repo: r.repo,
+            default_fleet_node: r.target_node_id,
+            default_backend: r.backend,
+            team_engine: r.team_engine,
+            step: std::sync::atomic::AtomicU32::new(r.completed_steps),
+        }
+    }
+
+    /// Which fleet node this graph node runs on.
+    ///
+    /// Fail-closed on a malformed override: placing a node on the mission's node
+    /// because its own `node_id` did not parse would run the work somewhere the
+    /// graph did not ask for and say nothing.
+    fn fleet_node(&self, req: &TurnRequest) -> Result<NodeId, OrchestratorError> {
+        let id = match req.attrs.get("node_id") {
+            Some(raw) => Uuid::parse_str(raw.trim()).map_err(|_| {
+                OrchestratorError::Executor(format!(
+                    "node {} has an invalid node_id attr: {raw}",
+                    req.node_id
+                ))
+            })?,
+            None => self.default_fleet_node.ok_or_else(|| {
+                OrchestratorError::Executor(format!(
+                    "node {} has no node_id attr and the mission has no \
+                     target_node_id — a microVM node cannot run on the gateway, \
+                     which has no /dev/kvm",
+                    req.node_id
+                ))
+            })?,
+        };
+        Ok(NodeId::from(id))
+    }
+}
+
+impl<V: PhaseVm> TurnExecutor for MicroVmTurnExecutor<V> {
+    async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
+        let fleet_node = self.fleet_node(&req)?;
+        let backend = req
+            .attrs
+            .get("backend")
+            .map(|s| s.trim().to_string())
+            .filter(|s| !s.is_empty())
+            .or_else(|| self.default_backend.clone());
+
+        if !self.repo.is_dir() {
+            return Err(OrchestratorError::Executor(format!(
+                "mission has no checkout at {} — a composed node needs the \
+                 repository, and it is also how the previous node's work reaches \
+                 this one",
+                self.repo.display()
+            )));
+        }
+
+        let step = self
+            .step
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+
+        // Held for the length of the VM turn: an hour of silence would otherwise
+        // look exactly like a dead worker to `requeue_stale`.
+        let _alive = Keepalive::spawn(self.pool.clone(), self.run_id);
+
+        let task = node_task_text(&req);
+        let outcome = self
+            .vms
+            .run(VmPhase {
+                node_id: fleet_node,
+                mission_id: self.mission_id,
+                phase_id: self.phase_id,
+                iteration: self.iteration,
+                task: &task,
+                backend: backend.as_deref(),
+                repo: &self.repo,
+                team_engine: self.team_engine.as_deref(),
+                step: Some(step),
+            })
+            .await
+            .map_err(|e| {
+                OrchestratorError::Executor(format!("node {} in a microVM: {e}", req.node_id))
+            })?;
+
+        // A node whose work never came back must fail the run rather than hand
+        // the next node a tree missing the previous one's edits. On this path an
+        // uncollected turn is worse than on the solo one: the loss is silent,
+        // because the next node still boots from a checkout that looks fine.
+        if !outcome.collected {
+            return Err(OrchestratorError::Executor(format!(
+                "node {}'s work could not be collected from its VM, so the next \
+                 node would not see it: {}",
+                req.node_id,
+                outcome.summary.chars().take(400).collect::<String>()
+            )));
+        }
+        if outcome.rc != 0 {
+            return Err(OrchestratorError::Executor(format!(
+                "node {} exited {}: {}",
+                req.node_id,
+                outcome.rc,
+                outcome.summary.chars().take(400).collect::<String>()
+            )));
+        }
+
+        eprintln!(
+            "microvm_turn_executor: run {} node {} (role {}, step {}) ok — subagents: {}",
+            self.run_id,
+            req.node_id,
+            req.role,
+            step,
+            outcome
+                .subagents
+                .map(|n| n.to_string())
+                .unwrap_or_else(|| "?".into()),
+        );
+
+        Ok(TurnOutcome {
+            output: outcome.summary,
+            // `claude -p` does not report token usage on stdout, and inventing a
+            // number here would corrupt the run totals the harness reads. Zero is
+            // the honest value for "not measured on this path".
+            tokens: 0,
+            gated: Vec::new(),
+        })
+    }
+}
+
+/// What one graph node is told.
+///
+/// The upstream outputs are included as context, but the load-bearing sentence is
+/// that the previous node's *files* are already in the tree: a node told only
+/// about the text would re-do work it is standing on.
+fn node_task_text(req: &TurnRequest) -> String {
+    let mut s = format!(
+        "You are the `{}` stage of a multi-stage mission.\n\nMISSION TASK\n{}\n",
+        req.role, req.task
+    );
+    if !req.context.is_empty() {
+        s.push_str(
+            "\nWHAT CAME BEFORE\nThe earlier stages' work is ALREADY IN THIS \
+             WORKING TREE — the repository you have been given is their output, \
+             not a fresh checkout. Read the files before changing them, and do \
+             not redo what is already done. Their closing reports:\n",
+        );
+        for (i, c) in req.context.iter().enumerate() {
+            s.push_str(&format!("\n--- stage {} ---\n{}\n", i + 1, c));
+        }
+    }
+    s
+}
+
+/// Touches a run's `updated_at` until dropped.
+struct Keepalive(tokio::task::JoinHandle<()>);
+
+impl Keepalive {
+    fn spawn(pool: PgPool, run_id: Uuid) -> Self {
+        Keepalive(tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(Duration::from_secs(KEEPALIVE_SECS));
+            loop {
+                ticker.tick().await;
+                let _ = cm_db::repo::topology_runs::touch(&pool, run_id).await;
+            }
+        }))
+    }
+}
+
+impl Drop for Keepalive {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
+/// Build the executor the worker uses, over real VMs on the fleet.
+pub fn for_fleet(
+    hub: Arc<crate::fleet::NodeHub>,
+    pool: PgPool,
+    r: ComposedRun,
+) -> MicroVmTurnExecutor<crate::microvm_executor::HubVms> {
+    MicroVmTurnExecutor::new(crate::microvm_executor::HubVms::new(hub), pool, r)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::microvm_executor::VmOutcome;
+    use std::collections::BTreeMap;
+    use std::sync::Mutex;
+
+    /// A VM modelled honestly: the host tree is packed in, the "agent" works on a
+    /// COPY that no host path points at, and the result is unpacked back over the
+    /// host tree. That is the real inject → run → collect shape, which is what
+    /// makes the negative control below meaningful — remove the collect and the
+    /// handoff breaks exactly as it would in production.
+    struct FakeVms {
+        /// Whether the guest's tree is collected back to the host.
+        collect: bool,
+        /// vm ids used, in order — the id is what stops two nodes colliding.
+        ids: Mutex<Vec<String>>,
+        /// (backend, fleet node) per call, for the heterogeneity assertions.
+        placements: Mutex<Vec<(Option<String>, NodeId)>>,
+    }
+
+    impl FakeVms {
+        fn new(collect: bool) -> Self {
+            Self {
+                collect,
+                ids: Mutex::new(Vec::new()),
+                placements: Mutex::new(Vec::new()),
+            }
+        }
+    }
+
+    impl PhaseVm for FakeVms {
+        async fn run(&self, p: VmPhase<'_>) -> Result<VmOutcome, String> {
+            self.ids
+                .lock()
+                .unwrap()
+                .push(format!("{}-{:?}", p.phase_id.simple(), p.step));
+            self.placements
+                .lock()
+                .unwrap()
+                .push((p.backend.map(str::to_string), p.node_id));
+
+            // inject: the host checkout goes in as a tar.
+            let tar = crate::mission_fs::pack_dir(p.repo, "repo")?;
+            let guest = tempfile::tempdir().map_err(|e| e.to_string())?;
+            crate::mission_fs::unpack_into(&tar, guest.path())?;
+            let guest_repo = guest.path().join("repo");
+
+            // run: the agent records that it was here, and reports what it found
+            // of the previous stages — the observation the handoff test reads.
+            let seen: Vec<String> = std::fs::read_dir(&guest_repo)
+                .map_err(|e| e.to_string())?
+                .filter_map(|e| e.ok())
+                .map(|e| e.file_name().to_string_lossy().to_string())
+                .filter(|n| n.starts_with("stage-"))
+                .collect();
+            let mine = guest_repo.join(format!("stage-{}.txt", p.step.unwrap_or(0)));
+            std::fs::write(&mine, "work").map_err(|e| e.to_string())?;
+
+            // collect: the guest tree comes back over the same host path.
+            if self.collect {
+                let back = crate::mission_fs::pack_dir(&guest_repo, "repo")?;
+                let parent = p.repo.parent().ok_or("no parent")?;
+                crate::mission_fs::unpack_into(&back, parent)?;
+            }
+
+            Ok(VmOutcome {
+                summary: format!("saw:[{}]", seen.join(",")),
+                rc: 0,
+                collected: true,
+                subagents: Some(0),
+                teammates: None,
+            })
+        }
+    }
+
+    fn req(node: &str, role: &str, context: Vec<String>) -> TurnRequest {
+        TurnRequest {
+            node_id: node.into(),
+            role: role.into(),
+            agent: None,
+            attrs: BTreeMap::new(),
+            task: "build the thing".into(),
+            context,
+        }
+    }
+
+    fn exec<V: PhaseVm>(vms: V, repo: PathBuf) -> MicroVmTurnExecutor<V> {
+        // A pool that is never connected: every test here fails the turn before
+        // any query, or drives one whose only DB touch is the best-effort
+        // keepalive (which swallows its own errors by design).
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(1)
+            .connect_lazy("postgres://invalid/invalid")
+            .expect("a lazy pool never dials");
+        MicroVmTurnExecutor::new(
+            vms,
+            pool,
+            ComposedRun {
+                run_id: Uuid::now_v7(),
+                mission_id: Uuid::now_v7(),
+                phase_id: Uuid::now_v7(),
+                iteration: 1,
+                repo,
+                target_node_id: Some(Uuid::now_v7()),
+                backend: Some("claude".into()),
+                team_engine: None,
+                completed_steps: 0,
+            },
+        )
+    }
+
+    fn a_checkout() -> tempfile::TempDir {
+        let d = tempfile::tempdir().unwrap();
+        std::fs::create_dir_all(d.path().join("repo")).unwrap();
+        std::fs::write(d.path().join("repo").join("README.md"), "hello").unwrap();
+        d
+    }
+
+    /// THE trap this slice exists to solve. A per-node VM is destroyed with its
+    /// filesystem, so unless the tree is carried forward, node 2 works from the
+    /// original checkout and silently loses node 1's edits — while still
+    /// reporting success.
+    #[tokio::test]
+    async fn a_later_node_sees_an_earlier_nodes_files() {
+        let d = a_checkout();
+        let e = exec(FakeVms::new(true), d.path().join("repo"));
+
+        let first = e.run_turn(req("n1", "implementer", vec![])).await.unwrap();
+        assert_eq!(first.output, "saw:[]", "the first node starts clean");
+
+        let second = e
+            .run_turn(req("n2", "verifier", vec![first.output.clone()]))
+            .await
+            .unwrap();
+        assert!(
+            second.output.contains("stage-0.txt"),
+            "node 2 could not see node 1's file: {}",
+            second.output
+        );
+        // And the host tree — what delivery diffs — holds both nodes' work.
+        for f in ["stage-0.txt", "stage-1.txt"] {
+            assert!(d.path().join("repo").join(f).exists(), "{f} missing on the host");
+        }
+    }
+
+    /// The negative control, run rather than assumed: with the collect removed —
+    /// i.e. a text-only handoff between nodes — the test above fails. A guard
+    /// that cannot detect the bug it was written for is decoration.
+    #[tokio::test]
+    async fn text_only_handoff_loses_the_earlier_nodes_work() {
+        let d = a_checkout();
+        let e = exec(FakeVms::new(false), d.path().join("repo"));
+
+        e.run_turn(req("n1", "implementer", vec![])).await.unwrap();
+        let second = e.run_turn(req("n2", "verifier", vec![])).await.unwrap();
+        assert_eq!(
+            second.output, "saw:[]",
+            "without a collect, node 2 must NOT see node 1's work — if it does, \
+             this test is no longer controlling anything"
+        );
+        assert!(!d.path().join("repo").join("stage-0.txt").exists());
+    }
+
+    /// Each node gets its own vm id within one phase and iteration. Two nodes
+    /// sharing an id means the second is refused by the fleet node while the
+    /// first is alive, and indistinguishable from a re-run once it is not.
+    #[tokio::test]
+    async fn every_node_runs_in_its_own_vm() {
+        let d = a_checkout();
+        let vms = FakeVms::new(true);
+        let e = exec(vms, d.path().join("repo"));
+        for n in ["n1", "n2", "n3"] {
+            e.run_turn(req(n, "worker", vec![])).await.unwrap();
+        }
+        let ids = e.vms.ids.lock().unwrap().clone();
+        let unique: std::collections::HashSet<_> = ids.iter().collect();
+        assert_eq!(unique.len(), ids.len(), "{ids:?}");
+    }
+
+    /// Resume must not re-use a completed step's vm id. The executor counts steps
+    /// itself, so the count has to start where the checkpoint left off.
+    #[tokio::test]
+    async fn a_resumed_run_continues_the_step_numbering() {
+        let d = a_checkout();
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(1)
+            .connect_lazy("postgres://invalid/invalid")
+            .unwrap();
+        let e = MicroVmTurnExecutor::new(
+            FakeVms::new(true),
+            pool,
+            ComposedRun {
+                run_id: Uuid::now_v7(),
+                mission_id: Uuid::now_v7(),
+                phase_id: Uuid::now_v7(),
+                iteration: 1,
+                repo: d.path().join("repo"),
+                target_node_id: Some(Uuid::now_v7()),
+                backend: None,
+                team_engine: None,
+                completed_steps: 2,
+            },
+        );
+        e.run_turn(req("n3", "worker", vec![])).await.unwrap();
+        let ids = e.vms.ids.lock().unwrap().clone();
+        assert!(
+            ids[0].ends_with("Some(2)"),
+            "the first step after a resume must be step 2, not 0: {ids:?}"
+        );
+    }
+
+    /// Per-node `backend` is what makes the outer graph heterogeneous — a
+    /// validator node on another provider's image. It must override the
+    /// mission's, and the mission's must still apply to nodes that say nothing.
+    #[tokio::test]
+    async fn a_node_may_pick_its_own_backend_and_fleet_node() {
+        let d = a_checkout();
+        let e = exec(FakeVms::new(true), d.path().join("repo"));
+        let elsewhere = Uuid::now_v7();
+
+        let mut r = req("n1", "worker", vec![]);
+        r.attrs.insert("backend".into(), "glm".into());
+        r.attrs.insert("node_id".into(), elsewhere.to_string());
+        e.run_turn(r).await.unwrap();
+        e.run_turn(req("n2", "worker", vec![])).await.unwrap();
+
+        let p = e.vms.placements.lock().unwrap().clone();
+        assert_eq!(p[0].0.as_deref(), Some("glm"));
+        assert_eq!(p[0].1, NodeId::from(elsewhere));
+        assert_eq!(p[1].0.as_deref(), Some("claude"), "the mission default");
+        assert_ne!(p[1].1, NodeId::from(elsewhere));
+    }
+
+    /// A malformed `node_id` must fail the node, not fall back to the mission's.
+    /// Silently running work somewhere the graph did not ask for is the same
+    /// class of bug as an alias that serde dropped.
+    #[tokio::test]
+    async fn a_malformed_node_placement_fails_closed() {
+        let d = a_checkout();
+        let e = exec(FakeVms::new(true), d.path().join("repo"));
+        let mut r = req("n1", "worker", vec![]);
+        r.attrs.insert("node_id".into(), "not-a-uuid".into());
+        let err = e.run_turn(r).await.unwrap_err().to_string();
+        assert!(err.contains("invalid node_id"), "{err}");
+    }
+
+    /// An uncollected node is a failed run here, not a warning: the next node
+    /// would boot from a tree that looks fine and is missing this node's work.
+    #[tokio::test]
+    async fn an_uncollected_node_fails_the_run() {
+        struct Lost;
+        impl PhaseVm for Lost {
+            async fn run(&self, _p: VmPhase<'_>) -> Result<VmOutcome, String> {
+                Ok(VmOutcome {
+                    summary: "did plenty".into(),
+                    rc: 0,
+                    collected: false,
+                    subagents: None,
+                    teammates: None,
+                })
+            }
+        }
+        let d = a_checkout();
+        let e = exec(Lost, d.path().join("repo"));
+        let err = e.run_turn(req("n1", "worker", vec![])).await.unwrap_err().to_string();
+        assert!(err.contains("could not be collected"), "{err}");
+    }
+
+    /// A node must be told its predecessors' files are already in the tree.
+    /// Given only the text, an agent re-does work it is standing on.
+    #[test]
+    fn a_downstream_node_is_told_the_work_is_already_in_the_tree() {
+        let solo = node_task_text(&req("n1", "implementer", vec![]));
+        assert!(solo.contains("build the thing"));
+        assert!(!solo.contains("WHAT CAME BEFORE"), "{solo}");
+
+        let later = node_task_text(&req("n2", "verifier", vec!["I wrote foo.rs".into()]));
+        assert!(later.contains("ALREADY IN THIS WORKING TREE"), "{later}");
+        assert!(later.contains("I wrote foo.rs"), "{later}");
+        assert!(later.contains("verifier"), "the node's role: {later}");
+    }
+}

crates/cm-api/src/mission_orchestrator.rs

@@ -574,6 +574,73 @@ async fn mint_team_from_template(
     Ok(team_id)
 }
 
+/// The graph a COMPOSED microVM mission runs, built from its team template
+/// without minting a single claw.
+///
+/// A composed mission needs the template's *shape* — how many nodes, in what
+/// pattern, playing what roles — and nothing else it carries. Its nodes are VMs,
+/// so provisioning claws for them would create agents, containers and `.brain`
+/// files that nothing ever dials; that is exactly why `on_launch` returns early
+/// for a microVM mission, and this is how the composed path gets its graph
+/// anyway rather than by undoing that.
+///
+/// `purposes` is the phase's purpose list, matched against `config.phase_teams`;
+/// missions using the legacy single `team_template_id` fall back to it.
+/// Returns `None` when the mission picked no template at all.
+pub async fn composed_graph(
+    pool: &PgPool,
+    mission_id: Uuid,
+    purposes: &[&str],
+) -> Result<Option<serde_json::Value>, String> {
+    let row: Option<(serde_json::Value, Option<Uuid>)> =
+        sqlx::query_as("SELECT config, team_template_id FROM missions WHERE id = $1")
+            .bind(mission_id)
+            .fetch_optional(pool)
+            .await
+            .map_err(|e| format!("load mission {mission_id}: {e}"))?;
+    let Some((config, legacy_template)) = row else {
+        return Err(format!("mission {mission_id} not found"));
+    };
+
+    let template_id = config
+        .get("phase_teams")
+        .and_then(|v| v.as_object())
+        .and_then(|pt| {
+            // First template named by any purpose this phase answers to, in the
+            // phase's own preference order — the same order `launch_phase` uses
+            // to pick teams, so a composed mission and a ZeroClaw one resolve the
+            // same template for the same phase.
+            purposes.iter().find_map(|p| {
+                pt.get(*p)
+                    .and_then(|v| v.as_array())
+                    .and_then(|a| a.first())
+                    .and_then(|v| v.as_str())
+                    .and_then(|s| Uuid::parse_str(s).ok())
+            })
+        })
+        .or(legacy_template);
+    let Some(template_id) = template_id else {
+        return Ok(None);
+    };
+
+    let template = cm_db::repo::team_templates::get(pool, template_id)
+        .await
+        .map_err(|e| format!("load template {template_id}: {e}"))?
+        .ok_or_else(|| format!("template {template_id} not found"))?;
+    let roles: Vec<&str> = template.roles.iter().map(|r| r.slot.as_str()).collect();
+    if roles.is_empty() {
+        return Err(format!("template {template_id} defines no roles"));
+    }
+    let graph = cm_topology::build(
+        parse_topology_kind(&template.template.default_topology),
+        &roles,
+    )
+    .map_err(|e| format!("build topology graph for template {template_id}: {e}"))?;
+    serde_json::to_value(&graph)
+        .map(Some)
+        .map_err(|e| format!("serialize topology graph: {e}"))
+}
+
 fn parse_topology_kind(s: &str) -> cm_topology::TopologyKind {
     use cm_topology::TopologyKind;
     match s {

crates/cm-api/src/phase_runner.rs

@@ -524,6 +524,32 @@ async fn launch_phase(
     // this mission specifically, and an env var that happens to be set must not
     // silently run it somewhere else.
     if p.runtime_kind == "microvm" {
+        // A composed mission's graph comes from the same `mission_teams` row a
+        // ZeroClaw mission would use — only its nodes run as VMs instead of
+        // claws. That is the whole point of composing the engines: the graph,
+        // the planners and the durability are Engine Z's, unchanged.
+        if wants_composed(p.team_engine) {
+            let team = team_rows
+                .iter()
+                .map(|r| {
+                    (
+                        r.get::<Uuid, _>("team_id"),
+                        r.get::<serde_json::Value, _>("graph"),
+                    )
+                })
+                .collect::<Vec<_>>();
+            return launch_composed_microvm_phase(
+                pool,
+                mission_id,
+                phase_id,
+                workspace_id,
+                iteration,
+                &task,
+                team,
+                purposes,
+            )
+            .await;
+        }
         return launch_microvm_phase(
             pool,
             hub,
@@ -611,6 +637,162 @@ async fn launch_phase(
 /// Returns as soon as the session is spawned: `launch_phase` runs inside the
 /// sweep loop, and blocking it for the length of a coding session would stall
 /// every other mission.
+/// Does this mission want the composed engines — a durable ZeroClaw graph whose
+/// every node is a Claude-Code-in-a-microVM session?
+///
+/// The third `team_engine` name migration 0069 anticipated. Exact match, like
+/// `wants_claude_code_team`: a typo must run solo rather than half-compose.
+fn wants_composed(team_engine: Option<&str>) -> bool {
+    matches!(team_engine, Some("composed"))
+}
+
+/// Enqueue a composed run — Slice 4's producer side.
+///
+/// Unlike every other executor here this one does **not** run the work: it emits
+/// ONE `queued` row carrying the real graph and lets `topology_worker` claim it.
+/// That is where the durability comes from — the checkpoint, the resume after a
+/// crash, and the cancellation all belong to the worker, and a run that drove
+/// itself from a `tokio::spawn` (as the solo microVM path does) would have none
+/// of them.
+///
+/// One row, not one per team, and the constraint is physical: every node injects
+/// from and collects back over the SAME host checkout, so two concurrent runs of
+/// one phase would be two VMs writing one directory. A mission with two matching
+/// teams is refused, visibly, rather than silently running only the first.
+#[allow(clippy::too_many_arguments)]
+async fn launch_composed_microvm_phase(
+    pool: &PgPool,
+    mission_id: Uuid,
+    phase_id: Uuid,
+    workspace_id: Uuid,
+    iteration: i32,
+    task: &str,
+    teams: Vec<(Uuid, serde_json::Value)>,
+    purposes: &[&str],
+) -> Result<(), String> {
+    sqlx::query(
+        "DELETE FROM topology_runs
+          WHERE mission_phase_id = $1 AND status IN ('failed', 'cancelled')",
+    )
+    .bind(phase_id)
+    .execute(pool)
+    .await
+    .map_err(|e| format!("purge prior runs for phase {phase_id}: {e}"))?;
+
+    // Refusals are recorded as a failed run rather than returned as an error:
+    // `launch_phase` is called from the sweep every ten seconds, so a returned
+    // error is a phase that retries forever while the log repeats itself. A
+    // failed row closes the phase and says why on the card.
+    let refuse = |why: String| async move {
+        eprintln!("phase_runner: composed phase {phase_id} of mission {mission_id} refused: {why}");
+        let _ = sqlx::query(
+            "INSERT INTO topology_runs
+                (id, workspace_id, task, kind, status, graph, tier,
+                 mission_id, mission_phase_id, iteration, error)
+             VALUES ($1, $2, $3, 'run', 'failed', $4, 'microvm_graph', $5, $6, $7, $8)",
+        )
+        .bind(Uuid::now_v7())
+        .bind(workspace_id)
+        .bind(task)
+        .bind(serde_json::json!({ "nodes": [], "edges": [] }))
+        .bind(mission_id)
+        .bind(phase_id)
+        .bind(iteration)
+        .bind(&why)
+        .execute(pool)
+        .await;
+        // The phase must still leave `pending`, or the sweep re-launches it.
+        let _ = sqlx::query(
+            "UPDATE mission_phases SET status = 'running', started_at = now()
+              WHERE id = $1 AND status = 'pending'",
+        )
+        .bind(phase_id)
+        .execute(pool)
+        .await;
+        Ok(())
+    };
+
+    let (team_id, graph) = match teams.len() {
+        1 => {
+            let (id, g) = teams.into_iter().next().expect("len == 1");
+            (Some(id), g)
+        }
+        // The usual case, and not an error: `mission_orchestrator::on_launch`
+        // deliberately mints NO team for a microVM mission, because claws in
+        // containers are exactly what a VM mission does not use. A composed run
+        // needs the template's shape, not its claws, so it builds the graph from
+        // the template directly.
+        0 => match crate::mission_orchestrator::composed_graph(pool, mission_id, purposes).await {
+            Ok(Some(g)) => (None, g),
+            Ok(None) => {
+                return refuse(
+                    "team_engine='composed' needs a team template to give the run its \
+                     shape, and this mission picked none — a composed mission is a \
+                     ZeroClaw graph whose nodes happen to be VMs, so without the graph \
+                     there is nothing to compose"
+                        .to_string(),
+                )
+                .await
+            }
+            Err(e) => return refuse(format!("could not build the composed graph: {e}")).await,
+        },
+        n => {
+            return refuse(format!(
+                "{n} teams match this phase, and a composed run must be exactly one: \
+                 every node injects from and collects back over the same host \
+                 checkout, so two runs would be two VMs writing one directory"
+            ))
+            .await
+        }
+    };
+
+    // Parse here rather than letting the worker discover it: a graph the
+    // orchestrator cannot plan would otherwise be claimed, fail with "missing or
+    // invalid graph", and look like a runtime fault instead of a bad team.
+    if let Err(e) = serde_json::from_value::<cm_topology::TopologyGraph>(graph.clone()) {
+        return refuse(format!(
+            "the graph for this composed phase (team {team_id:?}) is not a runnable \
+             topology: {e}"
+        ))
+        .await;
+    }
+
+    let run_id = Uuid::now_v7();
+    sqlx::query(
+        "INSERT INTO topology_runs
+            (id, workspace_id, task, kind, status, graph, tier,
+             team_id, mission_id, mission_phase_id, iteration)
+         VALUES ($1, $2, $3, 'run', 'queued', $4, 'microvm_graph', $5, $6, $7, $8)",
+    )
+    .bind(run_id)
+    .bind(workspace_id)
+    .bind(task)
+    .bind(&graph)
+    .bind(team_id)
+    .bind(mission_id)
+    .bind(phase_id)
+    .bind(iteration)
+    .execute(pool)
+    .await
+    .map_err(|e| format!("enqueue composed run for phase {phase_id}: {e}"))?;
+
+    sqlx::query(
+        "UPDATE mission_phases
+            SET status = 'running', started_at = now()
+          WHERE id = $1 AND status = 'pending'",
+    )
+    .bind(phase_id)
+    .execute(pool)
+    .await
+    .map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
+
+    eprintln!(
+        "phase_runner: mission {mission_id} phase {phase_id} queued as a COMPOSED run \
+         {run_id} (team {team_id:?}) — a ZeroClaw graph with microVM nodes"
+    );
+    Ok(())
+}
+
 /// Launch a phase inside a Firecracker microVM on the mission's placed node.
 ///
 /// Mirrors [`launch_direct_session`] on purpose, down to creating exactly ONE
@@ -707,6 +889,9 @@ async fn launch_microvm_phase(
                     backend: backend.as_deref(),
                     repo: &repo,
                     team_engine: team_engine.as_deref(),
+                    // The solo path is one VM for the whole phase; only a
+                    // composed run needs the id qualified per graph node.
+                    step: None,
                 },
             )
             .await
@@ -1247,6 +1432,36 @@ mod tests {
         pairs.iter().map(|(k, v)| (k.to_string(), *v)).collect()
     }
 
+    /// Composing the engines must be asked for by name. A mission that says
+    /// nothing — or says something adjacent — runs the solo VM path it ran
+    /// before, one VM for the whole phase.
+    #[test]
+    fn only_an_explicit_request_composes_the_engines() {
+        assert!(wants_composed(Some("composed")));
+        for engine in [
+            None,
+            Some(""),
+            Some("claude_code"),
+            Some("zeroclaw"),
+            Some("Composed"),
+            Some("compose"),
+        ] {
+            assert!(!wants_composed(engine), "{engine:?} must not compose");
+        }
+    }
+
+    /// The composed tier the producer inserts and the tier the worker dispatches
+    /// on are the same string, and nothing but this test connects them. A typo
+    /// would enqueue a row no worker ever claims: the phase would sit `running`
+    /// with a `queued` run behind it and no error anywhere.
+    #[test]
+    fn the_composed_tier_is_one_the_worker_actually_drives() {
+        assert!(
+            cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&"microvm_graph"),
+            "the producer would enqueue a run nobody claims"
+        );
+    }
+
     /// The regression from mission `019fcf62`: a coding phase whose agents were
     /// silently unpinned from the repo wrote nothing and reported `completed`,
     /// the same status a fully delivered phase gets.

crates/cm-api/src/topology_worker.rs

@@ -33,7 +33,12 @@ const REAP_STUCK_AFTER_SECS: i64 = 15 * 60;
 
 /// Spawn the durable topology job worker. Polls for queued jobs every `poll`
 /// interval; runs each to completion (or failure), checkpointing per step.
-pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime, poll: Duration) {
+pub fn spawn(
+    pool: PgPool,
+    runtime: cm_runtime::Runtime,
+    hub: Arc<crate::fleet::NodeHub>,
+    poll: Duration,
+) {
     // Fire the stuck-container reaper on its own cadence — checking
     // once a minute is plenty and keeps this off the hot claim loop.
     let reaper_pool = pool.clone();
@@ -55,7 +60,7 @@ pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime, poll: Duration) {
                 eprintln!("topology_worker: requeue_stale failed: {e}");
             }
             match cm_db::repo::topology_runs::claim_next_queued(&pool).await {
-                Ok(Some(job)) => run_job(&pool, &runtime, job).await,
+                Ok(Some(job)) => run_job(&pool, &runtime, &hub, job).await,
                 Ok(None) => tokio::time::sleep(poll).await,
                 Err(e) => {
                     eprintln!("topology_worker: claim failed: {e}");
@@ -90,7 +95,11 @@ async fn reap_stuck_runs(pool: &PgPool) -> Result<(), sqlx::Error> {
     )
     .bind(REAP_STUCK_AFTER_SECS as f64)
     .bind(
-        cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS
+        // REAPABLE, not worker-driven: `microvm_graph` is driven by this worker
+        // and must NOT be reaped — one of its steps is a whole agent session in a
+        // VM, so "no step records in 15 minutes" describes a healthy composed run
+        // as readily as a wedged one.
+        cm_db::repo::topology_runs::REAPABLE_TIERS
             .iter()
             .map(|s| (*s).to_string())
             .collect::<Vec<_>>(),
@@ -119,6 +128,7 @@ async fn reap_stuck_runs(pool: &PgPool) -> Result<(), sqlx::Error> {
 async fn run_job(
     pool: &PgPool,
     runtime: &cm_runtime::Runtime,
+    hub: &Arc<crate::fleet::NodeHub>,
     job: cm_db::repo::topology_runs::ClaimedTopologyRun,
 ) {
     let id = job.id;
@@ -156,9 +166,21 @@ async fn run_job(
     // Resume from the last checkpoint, or start fresh.
     let progress: RunProgress = job
         .checkpoint
+        .clone()
         .and_then(|c| serde_json::from_value(c).ok())
         .unwrap_or_default();
 
+    // The composed engines (Slice 4): this graph's nodes are not claws, they are
+    // Claude-Code-in-a-microVM sessions. Branched BEFORE the leaf executor is
+    // built, because that build reads the ZeroClaw gateway config — a composed
+    // run must not fail for want of a runtime it never dials.
+    if job.tier == "microvm_graph" {
+        let result = run_composed(pool, hub, &job, &graph, progress).await;
+        finish(pool, id, result).await;
+        maybe_teardown_ephemeral_team(pool, runtime, id).await;
+        return;
+    }
+
     // C3: prefer the mission's per-run runtime endpoint when set on
     // the missions row; else fall back to the shared env-derived
     // gateway (pre-C3 missions + non-mission runs). This is what
@@ -212,6 +234,14 @@ async fn run_job(
         _ => drive(pool, id, &graph, &job.task, progress, &leaf).await,
     };
 
+    finish(pool, id, result).await;
+    maybe_teardown_ephemeral_team(pool, runtime, id).await;
+}
+
+/// Write a driven run's terminal state. The single place a run finishes, shared
+/// by every tier — a second one would be a second completion path, which is where
+/// every microVM bug this project has hit came from.
+async fn finish(pool: &PgPool, id: Uuid, result: Result<RunRecord, OrchestratorError>) {
     match result {
         Ok(record) => {
             let value = serde_json::to_value(&record).unwrap_or(serde_json::Value::Null);
@@ -230,7 +260,57 @@ async fn run_job(
             }
         }
     }
-    maybe_teardown_ephemeral_team(pool, runtime, id).await;
+}
+
+/// Drive a composed run: the outer graph is Engine Z, every node is a
+/// Claude-Code-in-a-microVM session (Engine C).
+///
+/// The mission columns are read here rather than carried on the run row so a
+/// re-placed or re-backed mission takes effect on resume, and so the composed
+/// path has exactly one source of truth for where a VM boots.
+async fn run_composed(
+    pool: &PgPool,
+    hub: &Arc<crate::fleet::NodeHub>,
+    job: &cm_db::repo::topology_runs::ClaimedTopologyRun,
+    graph: &TopologyGraph,
+    progress: RunProgress,
+) -> Result<RunRecord, OrchestratorError> {
+    let mission_id = job.mission_id.ok_or_else(|| {
+        OrchestratorError::Executor(
+            "a composed run has no mission, so there is no checkout for its nodes \
+             to share"
+                .into(),
+        )
+    })?;
+    let phase_id = job.mission_phase_id.ok_or_else(|| {
+        OrchestratorError::Executor("a composed run must belong to a mission phase".into())
+    })?;
+
+    let mission: (Option<Uuid>, Option<String>, Option<String>) =
+        sqlx::query_as("SELECT target_node_id, backend, team_engine FROM missions WHERE id = $1")
+            .bind(mission_id)
+            .fetch_one(pool)
+            .await
+            .map_err(|e| OrchestratorError::Executor(format!("load mission {mission_id}: {e}")))?;
+
+    let exec = crate::microvm_turn_executor::for_fleet(
+        hub.clone(),
+        pool.clone(),
+        crate::microvm_turn_executor::ComposedRun {
+            run_id: job.id,
+            mission_id,
+            phase_id,
+            iteration: job.iteration.unwrap_or(1),
+            repo: crate::mission_workspace::checkout_path(mission_id),
+            target_node_id: mission.0,
+            backend: mission.1,
+            team_engine: mission.2,
+            // Resume continues the step numbering; restarting it would re-use a
+            // finished node's vm id.
+            completed_steps: progress.completed as u32,
+        },
+    );
+    drive(pool, job.id, graph, &job.task, progress, &exec).await
 }
 
 /// Post-terminal hook: if this run's team is `ephemeral` and no siblings are

crates/cm-db/src/repo/topology_runs.rs

@@ -54,6 +54,15 @@ pub struct ClaimedTopologyRun {
     /// Deploy tier: `team` drives claws directly; `company`/`org` drive the
     /// recursive sub-topology executor.
     pub tier: String,
+    /// The mission this run belongs to, when it belongs to one. The composed
+    /// (`microvm_graph`) tier needs it: its nodes share the mission's checkout,
+    /// and that shared tree is how file work survives a node boundary.
+    pub mission_id: Option<Uuid>,
+    /// The mission phase, for the same reason — the phase and pass identify the
+    /// VMs a composed run may boot.
+    pub mission_phase_id: Option<Uuid>,
+    /// Which pass of the phase produced this run.
+    pub iteration: Option<i32>,
 }
 
 /// Lifecycle status + progress for a durable run (status endpoint).
@@ -222,7 +231,34 @@ pub async fn check_ephemeral_teardown(
 ///
 /// An allowlist rather than a denylist on purpose: the next self-driven tier is
 /// then safe by default, instead of exposed until someone remembers this file.
-pub const WORKER_DRIVEN_TIERS: &[&str] = &["team", "company", "org", "swarm", "compare"];
+pub const WORKER_DRIVEN_TIERS: &[&str] = &[
+    "team",
+    "company",
+    "org",
+    "swarm",
+    "compare",
+    // The composed engines: a ZeroClaw graph whose every node is a
+    // Claude-Code-in-a-microVM session. Worker-driven BY DESIGN — the outer
+    // graph's durability (checkpoint, resume, cancellation) is the entire reason
+    // the tier exists, and it comes from being claimed like any other job. It
+    // survives `requeue_stale` because the executor touches `updated_at` from a
+    // ticker for the whole length of a VM turn, not only between steps.
+    "microvm_graph",
+];
+
+/// Tiers the stuck-run reaper may fail.
+///
+/// A subset of [`WORKER_DRIVEN_TIERS`], and the difference matters. The reaper
+/// asks "has this run journaled a step within 15 minutes of being CREATED?",
+/// which assumes a step is short. A `microvm_graph` node is a whole agent session
+/// in a VM with an hour's budget, so a healthy composed run can legitimately
+/// journal nothing for far longer than the reaper's patience — it would kill the
+/// run and orphan a live VM, which is #54 wearing a different tier.
+///
+/// Losing the reaper for that tier costs little: a composed run that genuinely
+/// wedges stops touching `updated_at` and `requeue_stale` recovers it at 180s,
+/// which is the mechanism the reaper was a backstop for in the first place.
+pub const REAPABLE_TIERS: &[&str] = &["team", "company", "org", "swarm", "compare"];
 
 /// The allowlist as owned strings, for binding as `text[]`.
 fn worker_driven() -> Vec<String> {
@@ -251,7 +287,8 @@ pub async fn claim_next_queued(pool: &PgPool) -> Result<Option<ClaimedTopologyRu
              FOR UPDATE SKIP LOCKED
              LIMIT 1
          )
-         RETURNING id, workspace_id, task, graph, checkpoint, last_event_id, tier",
+         RETURNING id, workspace_id, task, graph, checkpoint, last_event_id, tier,
+                   mission_id, mission_phase_id, iteration",
     )
     .bind(worker_driven())
     .fetch_optional(pool)
@@ -264,6 +301,9 @@ pub async fn claim_next_queued(pool: &PgPool) -> Result<Option<ClaimedTopologyRu
         checkpoint: r.get("checkpoint"),
         last_event_id: r.get("last_event_id"),
         tier: r.get("tier"),
+        mission_id: r.get("mission_id"),
+        mission_phase_id: r.get("mission_phase_id"),
+        iteration: r.get("iteration"),
     }))
 }
 

crates/cm-db/tests/self_driven_runs.rs

@@ -139,6 +139,53 @@ async fn the_worker_will_not_claim_a_self_driven_run() {
     assert_eq!(status_of(&pool, id).await, "queued", "and it must be left as it was");
 }
 
+/// The composed tier is the mirror image of the two above and must not be
+/// mistaken for them: it runs VMs, but the WORKER drives its graph, so being
+/// claimed and requeued is exactly what gives it checkpointing and resume.
+#[tokio::test]
+async fn the_worker_claims_and_rescues_a_composed_run() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = workspace(&pool).await;
+
+    let id = Uuid::now_v7();
+    topology_runs::enqueue_run_tier(
+        &pool,
+        id,
+        ws,
+        "compose the engines",
+        // A real graph, unlike the self-driven placeholder: the worker plans it.
+        &serde_json::json!({
+            "kind": "pipeline",
+            "nodes": [{ "id": "a", "role": "worker", "attrs": {} }],
+            "edges": []
+        }),
+        "microvm_graph",
+    )
+    .await
+    .expect("enqueue");
+
+    let claimed = topology_runs::claim_next_queued(&pool)
+        .await
+        .expect("claim")
+        .expect("a composed run must be claimable, or it never runs at all");
+    assert_eq!(claimed.tier, "microvm_graph");
+    assert_eq!(claimed.id, id);
+
+    // And a composed run whose worker died must come back: its checkpoint is
+    // what makes resume possible, and requeue is what triggers it.
+    sqlx::query("UPDATE topology_runs SET updated_at = now() - interval '30 minutes' WHERE id = $1")
+        .bind(id)
+        .execute(&pool)
+        .await
+        .expect("age the run");
+    topology_runs::requeue_stale(&pool, 180.0).await.expect("requeue");
+    assert_eq!(
+        status_of(&pool, id).await,
+        "queued",
+        "a composed run orphaned by a dead worker must be recovered"
+    );
+}
+
 /// The allowlist is the single place this policy lives, so assert its membership
 /// directly — a new self-driven tier added without touching it would otherwise be
 /// exposed exactly as microvm was.
@@ -156,4 +203,16 @@ fn the_allowlist_names_only_worker_driven_tiers() {
             "{self_driven} owns its own lifecycle; sweeping it kills live work"
         );
     }
+    // `microvm_graph` is worker-driven but NOT reapable: one of its steps is a
+    // whole agent session in a VM, so "no step records in 15 minutes" is what a
+    // healthy composed run looks like, and reaping it would orphan a live VM —
+    // #54 in a different tier.
+    assert!(topology_runs::WORKER_DRIVEN_TIERS.contains(&"microvm_graph"));
+    assert!(!topology_runs::REAPABLE_TIERS.contains(&"microvm_graph"));
+    for reapable in topology_runs::REAPABLE_TIERS {
+        assert!(
+            topology_runs::WORKER_DRIVEN_TIERS.contains(reapable),
+            "{reapable} is reaped but never driven"
+        );
+    }
 }