feat(placement): capacity model for the fleet — observed memory is not capacity

Phase 1a of the fleet-intelligence plan: the arithmetic and the inputs. Nothing
is wired to it yet; the launch path still picks `capable.first()`.

Placement has been `ORDER BY last_seen DESC` + `.first()` — the most recently
heartbeated node. Among healthy nodes all heartbeating every 5s that is
arbitrary, and it consults nothing about load, so two missions launched together
land on the same machine. It did not matter while tank held the only rootfs
image. All three nodes serve `claude` as of today.

THE correctness point, and the reason this is not a sort change: a VM that booted
30 seconds ago holds a fraction of its 8 GiB claim, so `mem_pct` reports a
sold-out node as nearly idle. `capacity_of` takes the WORSE of observed usage and
committed usage. The negative control pins it with the measured case — tank at
60 GiB total / 12 GiB observed / 5 VMs booted: utilisation alone says 5 more fit,
the node has room for 1. Booking those five is a node in swap, which slows every
VM on it together.

Commitments are unioned BY IDENTITY, never added: `vm_list` reports booted VMs,
`nodes::pinned_microvm_phases` reports phases chosen but not yet booted (a
window of seconds in which a real 8 GiB claim exists that no node can report).
The deterministic `vm_id_for` is what lets the same phase be recognised in both —
counting it twice would shrink the fleet by the number of phases starting.

`EvalRow::headroom()` finally gets a caller. It was written with the doc comment
"for placement ranking" and has had zero callers since. It is a TIEBREAK, not a
gate: ranking is slots first (spread, don't stack), then live headroom, then node
id so the same fleet state yields the same answer twice — which `last_seen DESC`
could never promise.

Fail-closed per house convention: draining, stale health (>30s, tuned just above
the 20s offline sweeper), and an unanswerable `vm_list` are all INELIGIBLE rather
than low-scoring. Stale Beszel metrics are the one exception — they demote a node
to zero headroom instead of excluding it, because they only ever break ties.

`FleetAtCapacity` and `FleetUnreadable` are separate variants with a test
asserting the second never says "at capacity": an operator sent hunting a load
problem that is really a dead daemon wastes the outage.

Also names the two nodes that were both called "New node" (tank, morpheus) — a
capacity report naming two machines identically is one nobody can act on.

257 lib tests.

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

@@ -71,6 +71,18 @@ pub struct EvalRow {
     pub gpu_pct: Option<f64>,
     pub temp_max: Option<f64>,
     pub load1: Option<f64>,
+    /// Absolute memory, for CAPACITY rather than utilisation. `mem_pct` cannot
+    /// answer "does another 8 GiB VM fit" — a node at 20% of 31 GiB and one at
+    /// 20% of 60 GiB report the same percentage and hold a different number of
+    /// VMs. From the 5s heartbeat, which is the only source with absolutes.
+    pub mem_total_bytes: Option<i64>,
+    pub mem_used_bytes: Option<i64>,
+    pub disk_free_bytes: Option<i64>,
+    /// Age of each source. Placement is fail-closed on stale health (a node whose
+    /// RAM we cannot read is one we are guessing at), and demotes rather than
+    /// excludes on stale Beszel metrics, which only ever break ties.
+    pub health_age_secs: Option<f64>,
+    pub metrics_age_secs: Option<f64>,
 }
 
 impl EvalRow {
@@ -91,6 +103,22 @@ impl EvalRow {
         let used = self.cpu_pct.unwrap_or(0.0).max(self.mem_pct.unwrap_or(0.0));
         100.0 - used
     }
+
+    /// [`headroom`] when at least one source is recent, otherwise the WORST
+    /// possible score.
+    ///
+    /// Used only as a placement TIEBREAK, never as an admission gate: stale
+    /// metrics may cost a node a tie, they may never win one. Admission is
+    /// decided by absolute memory from the heartbeat, which has its own
+    /// freshness check.
+    pub fn headroom_fresh(&self, max_age_secs: f64) -> f64 {
+        let fresh = |a: Option<f64>| a.is_some_and(|x| x <= max_age_secs);
+        if fresh(self.health_age_secs) || fresh(self.metrics_age_secs) {
+            self.headroom()
+        } else {
+            0.0
+        }
+    }
 }
 
 /// Every node's current metric scalars (merged Beszel + heartbeat health).
@@ -101,7 +129,12 @@ pub async fn eval_all(pool: &PgPool) -> Result<Vec<EvalRow>, DbError> {
             COALESCE(m.mem_pct, CASE WHEN h.mem_total > 0 THEN h.mem_used::float8 / h.mem_total * 100 END) AS mem_pct,
             COALESCE(m.disk_pct, CASE WHEN h.disk_total > 0 THEN (h.disk_total - h.disk_free)::float8 / h.disk_total * 100 END) AS disk_pct,
             m.gpu_pct, m.temp_max,
-            COALESCE(m.load1, h.load1) AS load1
+            COALESCE(m.load1, h.load1) AS load1,
+            h.mem_total AS mem_total_bytes,
+            h.mem_used  AS mem_used_bytes,
+            h.disk_free AS disk_free_bytes,
+            EXTRACT(EPOCH FROM now() - h.captured_at)::float8 AS health_age_secs,
+            EXTRACT(EPOCH FROM now() - m.updated_at)::float8  AS metrics_age_secs
          FROM nodes n
          LEFT JOIN node_health h ON h.node_id = n.id
          LEFT JOIN node_metrics m ON m.node_id = n.id",
@@ -120,6 +153,11 @@ pub async fn eval_all(pool: &PgPool) -> Result<Vec<EvalRow>, DbError> {
             gpu_pct: r.get("gpu_pct"),
             temp_max: r.get("temp_max"),
             load1: r.get("load1"),
+            mem_total_bytes: r.get("mem_total_bytes"),
+            mem_used_bytes: r.get("mem_used_bytes"),
+            disk_free_bytes: r.get("disk_free_bytes"),
+            health_age_secs: r.get("health_age_secs"),
+            metrics_age_secs: r.get("metrics_age_secs"),
         })
         .collect())
 }

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

@@ -238,6 +238,48 @@ pub async fn online_with_capabilities(
 /// daemon has no `rootfs` key at all and matches nothing, which is the same
 /// treatment an unqueried node gets for every other capability: unknown is not
 /// permission.
+///
+/// Capability only — this says a node COULD run the backend, not that it has room.
+/// Capacity is `cm_api::vm_placement`'s job.
+/// microVM phases already pinned to a node but whose VM may not exist yet.
+///
+/// The other half of "how much is this node committed to". `vm_list` reports
+/// BOOTED VMs; between `phase_runner` choosing a node and the guest answering,
+/// there is a window of seconds in which a phase is a real 8 GiB claim that no
+/// node can report. Two missions launched together both survey a node as empty
+/// and both land on it.
+///
+/// Returns (node, phase_id, iteration) so the caller can build the same
+/// deterministic vm id the executor uses and union the two sets by identity
+/// rather than adding them — a phase whose VM HAS booted must count once, not
+/// twice.
+pub async fn pinned_microvm_phases(
+    pool: &PgPool,
+    workspace_id: uuid::Uuid,
+) -> Result<Vec<(NodeId, uuid::Uuid, i32)>, DbError> {
+    let rows: Vec<(uuid::Uuid, uuid::Uuid, i32)> = sqlx::query_as(
+        "SELECT m.target_node_id, p.id, p.iteration
+           FROM mission_phases p
+           JOIN missions m ON m.id = p.mission_id
+          WHERE m.workspace_id = $1
+            AND m.runtime_kind = 'microvm'
+            AND m.status = 'running'
+            AND m.target_node_id IS NOT NULL
+            -- `pending` counts: it is about to become a VM. `completed`/`failed`
+            -- do not: their VM is destroyed on every exit path of
+            -- `run_phase_in_vm`, so counting them would shrink the fleet by the
+            -- number of missions it has ever run.
+            AND p.status IN ('pending', 'running')",
+    )
+    .bind(workspace_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows
+        .into_iter()
+        .map(|(n, p, i)| (NodeId::from(n), p, i))
+        .collect())
+}
+
 pub async fn online_for_backend(
     pool: &PgPool,
     workspace_id: uuid::Uuid,
@@ -251,7 +293,13 @@ pub async fn online_for_backend(
           WHERE workspace_id = $1 AND status = 'online'
             AND capabilities @> '{\"microvm\": true}'::jsonb
             AND capabilities -> 'rootfs' @> $2::jsonb
-          ORDER BY last_seen DESC NULLS LAST",
+          -- Deterministic, NOT `last_seen DESC`. Ranking now happens in
+          -- `cm_api::vm_placement`, against real capacity. Ordering by last_seen
+          -- and taking `.first()` was the placement algorithm until now: among
+          -- healthy nodes all heartbeating every 5s, that is arbitrary — it sent
+          -- concurrent missions to whichever node's packet landed most recently,
+          -- with no regard for what was already running there.
+          ORDER BY id",
     )
     .bind(workspace_id)
     .bind(serde_json::Value::Array(vec![serde_json::Value::String(