feat(placement): place per phase, and let a full fleet queue

Phase 1b: wires the capacity model from 3a2d76a into the launch path, and turns
the existing pending-phase loop into the queue.

Placement moves from mission launch to PHASE launch. A node chosen at launch is
chosen once, minutes before the first VM boots and hours before the last — and
re-placing between phases is free, because mission state lives on the gateway
checkout and every VM is inject -> run -> collect -> destroy. Pinning early
bought nothing and cost the ability to react to a node filling or draining
mid-mission. One call site serves both the solo and composed paths so they cannot
disagree; the composed worker reads `missions.target_node_id`, which placement
writes before dispatch.

QUEUEING, with no new machinery: a phase with no admissible node keeps its
`pending` status and creates no `topology_runs` row. `start_pending_phases`
retries every 10s — that loop already was a queue; nothing downstream ever sees a
run that did not happen.

The risk that creates is the one this codebase keeps paying for: a phase waiting
for capacity looks exactly like a phase nothing is working on. So the wait is
RECORDED, not merely logged — migration 0073 adds `capacity_blocked_since` and
`capacity_note`, stamped once and preserved across retries so the wait is
measured from the first refusal. It is bounded at two full turns: a fleet that
frees will free within one, and a phase that waited two hours must say so rather
than sit pending forever looking like a bug.

Mission launch still fails when NO node could ever run the backend — that is not
transient, waiting will not fix it, and `microvm-negctl` asserts such a mission
stays `draft`. Capacity refusals are transient and queue; capability refusals are
not and fail. The two are separate variants precisely so they cannot be confused.

257 lib tests, 20 binaries.

crates/cm-api/src/mission_orchestrator.rs

@@ -146,33 +146,30 @@ pub async fn on_launch(
             backend.unwrap_or("<name>")
         );
         let how_to_fix = how_to_fix.as_str();
-        let chosen = match mission.target_node_id {
+        // CAPABILITY is checked here; CAPACITY is not, and no node is pinned.
-            // An explicit target is a request, not a guarantee. Honour it only
+        //
-            // if the node actually reports the capability.
+        // Placement moved to phase launch (`phase_runner`). A node chosen now
-            Some(want) => *capable
+        // would be chosen once, minutes before the first VM boots and hours
-                .iter()
+        // before the last — and re-placing between phases is free, because
-                .find(|n| n.as_uuid() == want)
+        // mission state lives on the gateway checkout and every VM is
-                .ok_or_else(|| format!(
+        // inject → run → collect → destroy. Pinning early bought nothing and
-                    "mission targets node {want}, which is not an online node that can \
+        // cost the ability to react to a node filling or draining mid-mission.
-                     run backend {:?} — {how_to_fix}",
+        //
-                    backend.unwrap_or("default")
+        // Launching still FAILS here when no node could ever run this backend:
-                ))?,
+        // that is not transient, waiting will not fix it, and the harness's
-            None => *capable.first().ok_or_else(|| format!(
+        // `microvm-negctl` scenario asserts such a mission stays `draft`.
-                // Names the BACKEND, not just "microvm capability". Both nodes
+        if capable.is_empty() {
-                // report that capability; what one of them lacked was the image.
+            return Err(format!(
-                // The first version of this message would have sent an operator to
-                // reinstall firecracker on a node that already had it.
                 "no online node can run backend {:?} — {how_to_fix}",
                 backend.unwrap_or("default")
-            ))?,
+            ));
-        };
+        }
-        sqlx::query("UPDATE missions SET target_node_id = $1, updated_at = now() WHERE id = $2")
+        eprintln!(
-            .bind(chosen.as_uuid())
+            "mission_orchestrator: mission {mission_id} has {} node(s) able to run \
-            .bind(mission_id)
+             backend {:?}; placement happens per phase",
-            .execute(pool)
+            capable.len(),
-            .await
+            backend.unwrap_or("default")
-            .map_err(|e| format!("pin mission {mission_id} to node {chosen:?}: {e}"))?;
+        );
-        eprintln!("mission_orchestrator: mission {mission_id} placed on microvm node {chosen:?}");
     }
 
     // A microVM mission materialises no team. Its phases run as one `claude -p`

crates/cm-api/src/phase_runner.rs

@@ -363,6 +363,34 @@ fn empty_delivery_is_a_failure(
 /// so it is not held to producing them.
 const PRODUCING_KINDS: &[&str] = &["coding", "research", "benchmark", "security_scan"];
 
+/// How long a phase may wait for a VM slot before it is failed.
+///
+/// Two full turns. A fleet that genuinely frees will free within one, so this
+/// only fires when nothing is coming — and a phase that waited two hours must say
+/// so rather than sit `pending` forever looking like a bug.
+const CAPACITY_WAIT_MAX_SECS: f64 = 2.0 * 3600.0;
+
+/// Stamp why a phase is waiting, returning how long it has waited so far.
+///
+/// The timestamp is set once and preserved across retries, so the wait is
+/// measured from the first refusal rather than reset every 10s sweep.
+async fn record_capacity_block(pool: &PgPool, phase_id: Uuid, note: &str) -> f64 {
+    let waited: Option<f64> = sqlx::query_scalar(
+        "UPDATE mission_phases
+            SET capacity_blocked_since = COALESCE(capacity_blocked_since, now()),
+                capacity_note = $2
+          WHERE id = $1
+      RETURNING EXTRACT(EPOCH FROM now() - capacity_blocked_since)::float8",
+    )
+    .bind(phase_id)
+    .bind(note)
+    .fetch_optional(pool)
+    .await
+    .ok()
+    .flatten();
+    waited.unwrap_or(0.0)
+}
+
 /// Enqueue topology_runs for every phase whose predecessors are done.
 async fn start_pending_phases(
     pool: &PgPool,
@@ -668,6 +696,81 @@ 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" {
+        // PLACEMENT, for both the solo and composed paths, in one place so they
+        // cannot disagree about which node this phase runs on.
+        //
+        // Per phase rather than per mission: re-placing is free because mission
+        // state lives on the gateway checkout (inject -> run -> collect ->
+        // destroy), so a node that filled or drained since the last phase simply
+        // is not chosen for the next one.
+        // The node this phase will use. Starts as the mission's current pin (a
+        // request), becomes whatever placement actually chose.
+        let mut chosen_node = p.target_node_id;
+        match crate::vm_placement::choose(pool, hub, workspace_id, p.backend, chosen_node).await {
+            Ok(node) => {
+                if chosen_node != Some(node.as_uuid()) {
+                    eprintln!(
+                        "phase_runner: mission {mission_id} phase {phase_id} placed on node {} \
+                         (was {:?})",
+                        node.as_uuid(),
+                        chosen_node
+                    );
+                }
+                // Pin for the executors, which read `missions.target_node_id`,
+                // and clear any capacity wait now that one is satisfied.
+                let _ = sqlx::query(
+                    "UPDATE missions SET target_node_id = $1, updated_at = now() WHERE id = $2",
+                )
+                .bind(node.as_uuid())
+                .bind(mission_id)
+                .execute(pool)
+                .await;
+                let _ = sqlx::query(
+                    "UPDATE mission_phases
+                        SET capacity_blocked_since = NULL, capacity_note = NULL
+                      WHERE id = $1 AND capacity_blocked_since IS NOT NULL",
+                )
+                .bind(phase_id)
+                .execute(pool)
+                .await;
+                chosen_node = Some(node.as_uuid());
+            }
+            // TRANSIENT: the fleet is full, or we cannot read it. Leave the phase
+            // `pending` — `start_pending_phases` retries every 10s, and that loop
+            // IS the queue. No topology_runs row is created, so nothing downstream
+            // sees a run that never happened.
+            Err(e) if e.is_transient() => {
+                let msg = e.message();
+                // Recorded, not just logged: a phase waiting for capacity and a
+                // phase nothing is working on look identical from the outside,
+                // and this codebase has paid for that confusion repeatedly.
+                let blocked_for = record_capacity_block(pool, phase_id, &msg).await;
+                if blocked_for >= CAPACITY_WAIT_MAX_SECS {
+                    eprintln!(
+                        "phase_runner: mission {mission_id} phase {phase_id} waited \
+                         {blocked_for:.0}s for a VM slot — failing it.\n{msg}"
+                    );
+                    let _ = sqlx::query(
+                        "UPDATE mission_phases SET status = 'failed', completed_at = now(),
+                            capacity_note = $2 WHERE id = $1",
+                    )
+                    .bind(phase_id)
+                    .bind(format!("waited {blocked_for:.0}s for fleet capacity:\n{msg}"))
+                    .execute(pool)
+                    .await;
+                    return Err(format!("phase {phase_id} exhausted its capacity wait"));
+                }
+                eprintln!(
+                    "phase_runner: mission {mission_id} phase {phase_id} WAITING for a VM \
+                     slot ({blocked_for:.0}s so far)\n{msg}"
+                );
+                return Ok(());
+            }
+            // Not transient — no node could ever run this backend. Waiting will
+            // not fix it, so fail the phase rather than queue it forever.
+            Err(e) => return Err(e.message()),
+        }
+
         // 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,
@@ -704,7 +807,7 @@ async fn launch_phase(
             iteration,
             &task,
             p.backend,
-            p.target_node_id,
+            chosen_node,
             p.team_engine,
             crate::vm_stop_gate::StopGate::for_phase(kind, p.config),
         )

crates/cm-api/src/vm_placement.rs

@@ -34,7 +34,6 @@
 
 use cm_db::repo::node_metrics::EvalRow;
 use cm_domain::NodeId;
-use uuid::Uuid;
 
 /// Memory a phase VM claims. Re-exported from the executor so there is ONE number
 /// — a scheduler and a launcher that disagree about VM size is a fleet that
@@ -438,7 +437,7 @@ mod tests {
     use super::*;
 
     fn nid(n: u128) -> NodeId {
-        NodeId::from(Uuid::from_u128(n))
+        NodeId::from(uuid::Uuid::from_u128(n))
     }
 
     /// THE test. Measured on tank: 60 GiB total, and five VMs booted moments ago
@@ -567,7 +566,7 @@ mod tests {
     fn row_for(node_id: NodeId, total: i64, used: i64, disk_free: i64) -> EvalRow {
         EvalRow {
             node_id,
-            workspace_id: cm_domain::WorkspaceId::from(Uuid::from_u128(1)),
+            workspace_id: cm_domain::WorkspaceId::from(uuid::Uuid::from_u128(1)),
             status: "online".into(),
             cpu_pct: None,
             mem_pct: None,

migrations/0073_phase_capacity_wait.sql

@@ -0,0 +1,15 @@
+-- Why a phase is waiting, so a queue never looks like a stall.
+--
+-- Placement moves from mission launch to PHASE launch, which makes the existing
+-- `start_pending_phases` loop the queue: a phase with no admissible node simply
+-- stays `pending` and is retried every 10s. That is the whole queue mechanism —
+-- no new status, no sweeper, no migration for a `queued` state.
+--
+-- The risk it creates is the one this codebase keeps paying for: a phase waiting
+-- for capacity is indistinguishable from a phase nothing is working on. These two
+-- columns are what make the difference visible. `capacity_blocked_since` also
+-- bounds the wait, so a fleet that never frees fails the phase with a reason
+-- instead of holding it forever.
+ALTER TABLE mission_phases
+    ADD COLUMN IF NOT EXISTS capacity_blocked_since timestamptz,
+    ADD COLUMN IF NOT EXISTS capacity_note          text;