//! Which fleet node should run the next microVM phase, and whether any can.
//!
//! # What this replaces
//!
//! Placement was `capable.first()` over a list ordered `last_seen DESC`
//! (`mission_orchestrator`, `nodes::online_for_backend`) — the most recently
//! heartbeated node. Among healthy nodes all heartbeating every 5s that is
//! arbitrary, and it consults nothing about load: two missions launched together
//! land on the same machine. It did not matter while one node held the only
//! rootfs image; all three do now.
//!
//! # Observed memory is not capacity
//!
//! The correctness core, and the reason this is not a one-line sort change. A VM
//! that booted 30 seconds ago holds a fraction of its 8 GiB claim — the guest has
//! not touched the rest — so `mem_pct` reports a sold-out node as nearly idle.
//! Ranking on utilisation alone would happily book five more VMs onto a node with
//! room for one. `capacity_of` therefore takes the WORSE of observed usage and
//! committed usage, and `a_sold_out_node_is_not_mistaken_for_an_idle_one` is the
//! negative control that pins it.
//!
//! Commitments come from two places that must be unioned by IDENTITY, never
//! added: `microvm_client::list` (booted VMs, including orphans nothing has
//! reaped) and `nodes::pinned_microvm_phases` (chosen but not yet booted). The
//! deterministic `vm_id_for` is what lets the same phase be recognised in both.
//!
//! # Fail-closed
//!
//! A node whose health is stale, whose daemon will not answer, or which is
//! draining is INELIGIBLE, not low-scoring. Unknown is not permission — the same
//! rule `nodes::online_for_backend` already applies to capabilities. The one
//! exception is Beszel metrics: they feed `headroom` as a tiebreak only, so stale
//! metrics demote a node instead of excluding it.

use cm_db::repo::node_metrics::EvalRow;
use cm_domain::NodeId;

/// 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
/// overcommits by exactly their difference.
pub(crate) use crate::microvm_executor::MEM_MIB as MEM_PER_VM_MIB;

/// Held back for the host: the daemon, the OS, page cache, and the margin that
/// keeps a node out of swap. A node in swap makes every VM on it slow, so this is
/// cheaper than the alternative.
const HOST_RESERVE_MIB: i64 = 4096;

/// The floor we refuse to believe a host's own footprint is below. Without it, a
/// node reporting less used memory than its VMs have claimed would compute a
/// negative baseline and inflate its free memory.
const HOST_BASELINE_FLOOR_MIB: i64 = 2048;

/// Disk a VM may consume: an 8 GiB sparse rootfs plus room for the collected tar.
const DISK_PER_VM_GIB: i64 = 12;

/// Never let VM disk drive a node below this. `_outputs` and images live on the
/// same filesystem on some nodes.
const DISK_RESERVE_GIB: i64 = 20;

/// Health older than this and the node is ineligible. Deliberately close to the
/// 20s at which `fleet::spawn_node_sweeper` marks a node offline: the window in
/// which a node is "online with unreadable memory" should be narrow.
pub const MAX_HEALTH_AGE_SECS: f64 = 30.0;

/// Beszel metrics older than this rank as zero headroom. Only a tiebreak.
const MAX_METRICS_AGE_SECS: f64 = 60.0;

/// A node that can take at least one more phase VM.
#[derive(Debug, Clone, PartialEq)]
pub struct NodeCapacity {
    pub node_id: NodeId,
    pub name: String,
    /// How many MORE 8 GiB VMs fit.
    pub slots: i64,
    pub headroom: f64,
    pub committed_vms: i64,
    pub mem_total_mib: i64,
    pub used_eff_mib: i64,
    pub disk_free_gib: i64,
}

/// Why a node cannot take this phase. Each renders a distinct, actionable line —
/// "at capacity" and "we could not read it" send an operator to different places.
#[derive(Debug, Clone, PartialEq)]
pub enum Unfit {
    Draining,
    NotConnected,
    NoRecentHealth { age_secs: Option<f64> },
    CapacityUnknown { err: String },
    AtCapacity { committed: i64, used_eff_mib: i64, mem_total_mib: i64 },
    NoDisk { free_gib: i64 },
}

impl Unfit {
    pub fn reason(&self) -> String {
        match self {
            Unfit::Draining => "draining".into(),
            Unfit::NotConnected => "daemon not connected".into(),
            Unfit::NoRecentHealth { age_secs } => match age_secs {
                Some(a) => format!("health {a:.0}s stale (max {MAX_HEALTH_AGE_SECS:.0}s)"),
                None => "never reported health".into(),
            },
            Unfit::CapacityUnknown { err } => format!("could not read running VMs: {err}"),
            Unfit::AtCapacity { committed, used_eff_mib, mem_total_mib } => format!(
                "at capacity: {committed} VM(s), {used_eff_mib}/{mem_total_mib} MiB committed"
            ),
            Unfit::NoDisk { free_gib } => format!("only {free_gib} GiB free"),
        }
    }
}

/// Why placement produced no node. Distinguished because the operator response
/// differs: wait, fix a daemon, or build an image.
#[derive(Debug, Clone)]
pub enum PlacementError {
    /// No node has the image / KVM at all. Not a capacity problem.
    NoCapableNode { backend: String, how_to_fix: String },
    /// Every capable node is full. Transient — the caller should queue.
    FleetAtCapacity { report: String },
    /// We could not READ capacity. Must never be reported as "full".
    FleetUnreadable { report: String },
}

impl PlacementError {
    /// Whether the caller should wait and retry rather than fail the work.
    pub fn is_transient(&self) -> bool {
        matches!(
            self,
            PlacementError::FleetAtCapacity { .. } | PlacementError::FleetUnreadable { .. }
        )
    }

    pub fn message(&self) -> String {
        match self {
            PlacementError::NoCapableNode { backend, how_to_fix } => {
                format!("no online node can run backend {backend:?} — {how_to_fix}")
            }
            PlacementError::FleetAtCapacity { report } => format!(
                "fleet at capacity — a phase VM runs up to 60 min; this phase waits for a slot.\n{report}"
            ),
            PlacementError::FleetUnreadable { report } => format!(
                "cannot read node capacity — this is NOT a full fleet; check the node daemons.\n{report}"
            ),
        }
    }
}

/// What the host itself costs, excluding its phase VMs.
///
/// With nothing committed the answer is simply what the node reports. With VMs
/// committed it cannot be measured, only remembered or inferred — and inference
/// is where this went wrong: subtracting the VMs' FULL 8 GiB claim from
/// observed usage assumes they have already consumed it. A VM booted seconds
/// ago holds about an eighth of that, so the subtraction goes negative, hits
/// the floor, and hands back memory the host is really using.
///
/// Measured on morpheus (31757 MiB total, 4314 MiB idle, 2 slots) with 2 VMs
/// committed and young: the inferred baseline collapsed to the 2048 floor,
/// freeing 2266 MiB — exactly enough to admit a 3rd VM to a 2-slot node. The
/// `capacity` harness scenario caught it on its first full run.
///
/// So prefer the remembered idle reading, and take the LARGER of it and the
/// inference: a host that has genuinely started doing non-VM work must not be
/// under-charged just because it was once idle at a lower number.
fn host_baseline(mem_used_mib: i64, committed_vms: i64, baseline_mib: Option<i64>) -> i64 {
    if committed_vms <= 0 {
        // Directly observable, and the only moment it is.
        return mem_used_mib.max(HOST_BASELINE_FLOOR_MIB);
    }
    let inferred = mem_used_mib - committed_vms * MEM_PER_VM_MIB as i64;
    inferred
        .max(baseline_mib.unwrap_or(0))
        .max(HOST_BASELINE_FLOOR_MIB)
}

/// The whole capacity decision for one node, as pure arithmetic.
///
/// Separated from every I/O concern so the numbers can be tested against measured
/// fleet values without a database, a hub, or a VM.
pub fn capacity_of(
    node_id: NodeId,
    name: &str,
    mem_total_mib: i64,
    mem_used_mib: i64,
    disk_free_gib: i64,
    committed_vms: i64,
    // What this node used the last time it was seen with nothing committed.
    // `None` before it has ever been observed idle.
    baseline_mib: Option<i64>,
    headroom: f64,
) -> Result<NodeCapacity, Unfit> {
    let host_baseline = host_baseline(mem_used_mib, committed_vms, baseline_mib);
    let committed_use = committed_vms * MEM_PER_VM_MIB as i64 + host_baseline;

    // The worse of the two views. Observed alone under-counts a freshly booted
    // VM; committed alone under-counts a host doing real work outside its VMs.
    let used_eff = mem_used_mib.max(committed_use);
    let free = mem_total_mib - used_eff - HOST_RESERVE_MIB;
    let slots = if free <= 0 { 0 } else { free / MEM_PER_VM_MIB as i64 };

    if disk_free_gib - DISK_PER_VM_GIB < DISK_RESERVE_GIB {
        return Err(Unfit::NoDisk { free_gib: disk_free_gib });
    }
    if slots < 1 {
        return Err(Unfit::AtCapacity {
            committed: committed_vms,
            used_eff_mib: used_eff,
            mem_total_mib,
        });
    }
    Ok(NodeCapacity {
        node_id,
        name: name.to_string(),
        slots,
        headroom,
        committed_vms,
        mem_total_mib,
        used_eff_mib: used_eff,
        disk_free_gib,
    })
}

/// Admission inputs drawn from an `EvalRow`, or why the node is ineligible.
///
/// Fail-closed on stale or absent health: a node whose memory we cannot read is
/// one whose capacity we would be guessing at.
pub fn from_eval(
    row: &EvalRow,
    name: &str,
    committed_vms: i64,
) -> Result<NodeCapacity, Unfit> {
    if row.status == "draining" {
        return Err(Unfit::Draining);
    }
    let fresh = row
        .health_age_secs
        .is_some_and(|a| a <= MAX_HEALTH_AGE_SECS);
    let (Some(total), Some(used)) = (row.mem_total_bytes, row.mem_used_bytes) else {
        return Err(Unfit::NoRecentHealth { age_secs: row.health_age_secs });
    };
    if !fresh || total <= 0 {
        return Err(Unfit::NoRecentHealth { age_secs: row.health_age_secs });
    }
    const MIB: i64 = 1024 * 1024;
    const GIB: i64 = 1024 * 1024 * 1024;
    capacity_of(
        row.node_id,
        name,
        total / MIB,
        used / MIB,
        row.disk_free_bytes.unwrap_or(0) / GIB,
        committed_vms,
        row.mem_baseline_mib,
        row.headroom_fresh(MAX_METRICS_AGE_SECS),
    )
}

/// Rank admissible nodes: most free slots first, then live headroom, then id.
///
/// Slots before headroom SPREADS load rather than stacking it — two missions
/// launched together go to different machines. Headroom breaks ties with
/// real-time load, which is where a node mid-`cargo build` loses to an idle peer.
/// Node id last so the same fleet state always yields the same answer; the old
/// `last_seen DESC` made placement unreproducible between two identical runs.
pub fn rank(mut fit: Vec<NodeCapacity>) -> Vec<NodeCapacity> {
    fit.sort_by(|a, b| {
        b.slots
            .cmp(&a.slots)
            .then(
                b.headroom
                    .partial_cmp(&a.headroom)
                    .unwrap_or(std::cmp::Ordering::Equal),
            )
            .then(a.node_id.as_uuid().cmp(&b.node_id.as_uuid()))
    });
    fit
}

/// One line per node, for logs and for the message an operator reads.
pub fn report(fit: &[NodeCapacity], unfit: &[(NodeId, String, Unfit)]) -> String {
    let mut out = Vec::new();
    for f in fit {
        out.push(format!(
            "  {}: {} slot(s) free, {} VM(s) committed, {}/{} MiB, headroom {:.0}",
            f.name, f.slots, f.committed_vms, f.used_eff_mib, f.mem_total_mib, f.headroom
        ));
    }
    for (_, name, why) in unfit {
        out.push(format!("  {name}: UNFIT — {}", why.reason()));
    }
    if out.is_empty() {
        out.push("  (no capable nodes)".into());
    }
    out.join("\n")
}

/// Count a node's commitments, unioning booted VMs with pinned-not-yet-booted
/// phases BY IDENTITY.
///
/// A composed graph's step VMs (`...-s0`, `-s1`) each count: each is a real
/// Firecracker process holding 8 GiB. A pinned phase counts only while no live VM
/// carries its id — otherwise the same claim would be counted twice and the fleet
/// would shrink by the number of phases currently starting.
pub fn commitments(live_vm_ids: &[String], pinned_keys: &[String]) -> i64 {
    let live = live_vm_ids.len() as i64;
    let unbooted = pinned_keys
        .iter()
        .filter(|k| !live_vm_ids.iter().any(|v| v.starts_with(k.as_str())))
        .count() as i64;
    live + unbooted
}

/// Every backend a phase needs on ONE node: the mission's, plus each backend
/// named by a node of its composed graph.
///
/// The roster stores them as `config.roster.nodes[].attrs.backend`, and they are
/// the reason this function exists. A 2-member roster with
/// `verifier@canary-claude` was placed on a node holding `claude` and not
/// `canary-claude`; the graph's first node ran, the second died with
/// `no rootfs for backend "canary-claude" on this node`, and the mission
/// delivered half its work and failed. Placement had asked only about the
/// mission's own backend, which was true and insufficient.
pub fn required_backends(mission_backend: Option<&str>, roster: Option<&serde_json::Value>) -> Vec<String> {
    let mut out = vec![cm_db::repo::nodes::backend_key(mission_backend).to_string()];
    if let Some(nodes) = roster.and_then(|r| r.get("nodes")).and_then(|n| n.as_array()) {
        for n in nodes {
            if let Some(b) = n
                .get("attrs")
                .and_then(|a| a.get("backend"))
                .and_then(|b| b.as_str())
                .filter(|b| !b.trim().is_empty())
            {
                out.push(b.to_string());
            }
        }
    }
    out.sort();
    out.dedup();
    out
}

/// Survey every capable node: which can take a phase VM, and why the rest cannot.
///
/// `vm_list` is asked of each candidate in parallel with a short deadline. A node
/// that will not answer is `CapacityUnknown` and therefore ineligible — we cannot
/// count what we cannot see, and guessing zero is how a node gets double-booked.
pub async fn survey(
    pool: &sqlx::PgPool,
    hub: &crate::fleet::NodeHub,
    workspace_id: uuid::Uuid,
    // EVERY backend the work needs, not just the mission's. A composed graph
    // runs on ONE node and its nodes may each name their own — the roster's
    // whole purpose is an independent verifier on another provider — so the
    // node has to hold all of their rootfs images.
    backends: &[String],
) -> Result<(Vec<NodeCapacity>, Vec<(NodeId, String, Unfit)>), String> {
    let candidates = cm_db::repo::nodes::online_for_backends(pool, workspace_id, backends)
        .await
        .map_err(|e| format!("looking up nodes for backends {backends:?}: {e}"))?;
    if candidates.is_empty() {
        return Ok((Vec::new(), Vec::new()));
    }

    let evals = cm_db::repo::node_metrics::eval_all(pool)
        .await
        .map_err(|e| format!("reading node metrics: {e}"))?;
    let pinned = cm_db::repo::nodes::pinned_microvm_phases(pool, workspace_id)
        .await
        .map_err(|e| format!("reading pinned phases: {e}"))?;

    let names = node_names(pool, workspace_id).await;
    let mut fit = Vec::new();
    let mut unfit = Vec::new();
    for node in candidates {
        let row = evals.iter().find(|e| e.node_id == node);
        let name = names
            .get(&node.as_uuid())
            .cloned()
            .unwrap_or_else(|| node.as_uuid().to_string()[..8].to_string());

        // Not connected: nothing can be asked of it, and nothing can run on it.
        if !hub.is_connected(node) {
            unfit.push((node, name, Unfit::NotConnected));
            continue;
        }
        let Some(row) = row else {
            unfit.push((node, name, Unfit::NoRecentHealth { age_secs: None }));
            continue;
        };

        // Commitments: booted VMs unioned with phases pinned here but not yet
        // booted, by the deterministic id both sides agree on.
        let live = match crate::microvm_client::list(hub, node).await {
            Ok(v) => v,
            Err(e) => {
                unfit.push((node, name, Unfit::CapacityUnknown { err: e }));
                continue;
            }
        };
        let keys: Vec<String> = pinned
            .iter()
            .filter(|(n, _, _)| *n == node)
            .map(|(_, phase, iter)| crate::microvm_executor::vm_id_for(*phase, *iter, None))
            .collect();
        let committed = commitments(&live, &keys);

        // An idle node is the ONLY time its own footprint is measurable rather
        // than inferred, so take the reading whenever we get one. Cheap: an
        // UPDATE per idle node per survey, and it is what stops a young VM's
        // unconsumed memory from being handed out a second time.
        if committed == 0 {
            if let Some(used) = row.mem_used_bytes.filter(|_| {
                row.health_age_secs
                    .is_some_and(|a| a <= MAX_HEALTH_AGE_SECS)
            }) {
                let mib = used / (1024 * 1024);
                if row.mem_baseline_mib != Some(mib) {
                    let _ = cm_db::repo::nodes::set_mem_baseline(pool, node, mib).await;
                }
            }
        }

        match from_eval(row, &name, committed) {
            Ok(c) => fit.push(c),
            Err(why) => unfit.push((node, name, why)),
        }
    }
    Ok((rank(fit), unfit))
}

/// Node names for readable reports. A capacity report naming two machines
/// "New node" is a report nobody can act on.
async fn node_names(
    pool: &sqlx::PgPool,
    workspace_id: uuid::Uuid,
) -> std::collections::HashMap<uuid::Uuid, String> {
    sqlx::query_as::<_, (uuid::Uuid, String)>(
        "SELECT id, name FROM nodes WHERE workspace_id = $1",
    )
    .bind(workspace_id)
    .fetch_all(pool)
    .await
    .unwrap_or_default()
    .into_iter()
    .collect()
}

/// Choose a node for a phase, honouring an explicit target as a REQUEST.
///
/// `want` is honoured only if that node is genuinely admissible — the same
/// "a request, not a guarantee" rule the orchestrator already applied to
/// capability, now extended to capacity and draining.
pub async fn choose(
    pool: &sqlx::PgPool,
    hub: &crate::fleet::NodeHub,
    workspace_id: uuid::Uuid,
    backends: &[String],
    want: Option<uuid::Uuid>,
) -> Result<NodeId, PlacementError> {
    let named = backends.join(", ");
    let how_to_fix = format!(
        "needs /dev/kvm + firecracker (scripts/fc-node-setup.sh) AND the {named} rootfs \
         built on ONE node (scripts/fc-build-rootfs.sh <host> <image> <name>) — a \
         composed graph runs on a single node, so that node needs every image its \
         nodes ask for"
    );
    let (fit, unfit) = survey(pool, hub, workspace_id, backends).await.map_err(|e| {
        PlacementError::FleetUnreadable { report: format!("  survey failed: {e}") }
    })?;

    if fit.is_empty() && unfit.is_empty() {
        return Err(PlacementError::NoCapableNode {
            backend: named,
            how_to_fix,
        });
    }
    let report = report(&fit, &unfit);

    // `want` is ADVISORY, always. The only caller passes `missions.target_node_id`,
    // which is simply where the PREVIOUS phase ran — not an operator's choice.
    // Treating it as a requirement had two consequences, both wrong:
    //
    //   - a previous node that had since filled up (or gone unreadable) failed
    //     the phase outright: `TargetUnfit` is not transient, so it never
    //     reached the queue. Note this was NOT the drain case — a draining node
    //     is already excluded by `online_for_backend`'s `status = 'online'`, so
    //     it never reaches `unfit` at all and the pin simply falls through.
    //     `drain-midmission` passes either way; the path it does not cover is
    //     "phase 1's node is now full", which is the one that used to fail.
    //   - and while the node stayed fit, every later phase went back to it
    //     regardless of ranking — accidental mission-to-node affinity, which
    //     this module's own header says must not exist.
    //
    // Mission state lives on the gateway (inject -> run -> collect -> destroy),
    // so re-placing costs nothing. Prefer the pin when it still fits; say out
    // loud why it did not when it does not, and rank as usual.
    if let Some(want) = want {
        if let Some(c) = fit.iter().find(|c| c.node_id.as_uuid() == want) {
            return Ok(c.node_id);
        }
        if let Some((_, name, why)) = unfit.iter().find(|(n, _, _)| n.as_uuid() == want) {
            eprintln!(
                "vm_placement: the previous phase's node {name} is {} — re-placing this phase",
                why.reason()
            );
        }
    }

    if let Some(best) = fit.into_iter().next() {
        return Ok(best.node_id);
    }

    // Nothing fit. Distinguish "full" from "blind": an operator sent to look for
    // a load problem that is really a dead daemon wastes the outage.
    let blind = unfit.iter().all(|(_, _, w)| {
        matches!(w, Unfit::CapacityUnknown { .. } | Unfit::NotConnected | Unfit::NoRecentHealth { .. })
    });
    Err(if blind {
        PlacementError::FleetUnreadable { report }
    } else {
        PlacementError::FleetAtCapacity { report }
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn nid(n: u128) -> NodeId {
        NodeId::from(uuid::Uuid::from_u128(n))
    }

    /// THE test. Measured on tank: 60 GiB total, and five VMs booted moments ago
    /// showing only ~12 GiB used because the guests have not touched their claim.
    ///
    /// Observed-usage-only arithmetic says (61440-12000-4096)/8192 = 5 more VMs.
    /// The node has room for ONE. Booking those five is a node in swap, and every
    /// VM on it slows down together.
    /// A node whose VMs have not yet consumed their claim must not hand the
    /// difference out again.
    ///
    /// This is the bug the `capacity` harness scenario found on its first full
    /// run — "morpheus peaked at 3 concurrent VM(s) with only 2 slot(s)" — and
    /// the numbers here are that node's real ones. Idle it reports 4314 MiB of
    /// 31757 and the survey correctly gives it 2 slots. Two VMs later, each
    /// holding roughly 1 GiB of its 8 GiB, observed usage is ~6314 MiB;
    /// inferring the baseline as 6314 - 16384 goes negative, clamps to the
    /// 2048 floor, and invents 2266 MiB — exactly one more VM than exists.
    /// A previous node that is no longer usable re-places the next phase; it
    /// does not fail it.
    ///
    /// `choose` treated `missions.target_node_id` — which is only ever "where
    /// the last phase ran" — as a hard requirement, so a pinned node that had
    /// since FILLED UP produced `TargetUnfit`, which is not transient, and the
    /// phase failed instead of queueing or moving. It also gave every later
    /// phase silent affinity back to the first node.
    ///
    /// The drain case is not this one and never was: `online_for_backend`
    /// filters on `status = 'online'`, so a draining node is not a candidate
    /// and the pin falls through to ranking. `drain-midmission` passes on both
    /// the old and new code, which is why the capacity half needs this test.
    /// A composed graph's per-node backends are part of what placement needs.
    ///
    /// The full harness found this: a 2-member roster with
    /// `verifier@canary-claude` was placed on a node holding `claude` and not
    /// `canary-claude`. The first graph node ran, the second died with
    /// `no rootfs for backend "canary-claude" on this node`, and the mission
    /// delivered half its work and failed. Placement had asked only about the
    /// mission's own backend — true, and insufficient.
    #[test]
    fn a_composed_graph_needs_every_backend_its_nodes_name() {
        let roster = serde_json::json!({
            "kind": "pipeline",
            "nodes": [
                {"id": "n0", "role": "implementer"},
                {"id": "n1", "role": "verifier", "attrs": {"backend": "canary-claude"}},
            ],
        });
        assert_eq!(
            required_backends(Some("claude"), Some(&roster)),
            vec!["canary-claude".to_string(), "claude".to_string()],
            "both images have to be on the ONE node the graph runs on"
        );

        // A solo mission is unchanged — this must not make ordinary placement
        // stricter than it was.
        assert_eq!(required_backends(Some("claude"), None), vec!["claude"]);
        assert_eq!(required_backends(None, None), vec!["default"]);

        // A node with no explicit backend inherits the mission's, so it adds
        // nothing. Deduped, or a 5-node graph would ask for `claude` five times
        // and the containment query would still be right but the error message
        // would be nonsense.
        let inherit = serde_json::json!({"nodes": [
            {"id": "n0", "role": "a"},
            {"id": "n1", "role": "b", "attrs": {}},
            {"id": "n2", "role": "c", "attrs": {"backend": ""}},
        ]});
        assert_eq!(required_backends(Some("claude"), Some(&inherit)), vec!["claude"]);
    }

    #[test]
    fn an_unfit_previous_node_is_re_placed_not_refused() {
        let drained = uuid::Uuid::from_u128(1);
        let healthy = capacity_of(nid(2), "tank", 61440, 6144, 800, 0, None, 90.0).unwrap();

        // Stand in for `choose`'s decision: the pin is consulted, then dropped.
        let fit = vec![healthy.clone()];
        let picked = fit
            .iter()
            .find(|c| c.node_id.as_uuid() == drained)
            .or_else(|| fit.first())
            .expect("a fit node exists");
        assert_eq!(
            picked.node_id,
            nid(2),
            "with the pinned node absent from `fit`, ranking must still yield a node"
        );

        // And the error that used to be produced here no longer exists, so it
        // cannot be reintroduced as a non-transient failure by accident.
        for e in [
            PlacementError::FleetAtCapacity { report: String::new() },
            PlacementError::FleetUnreadable { report: String::new() },
        ] {
            assert!(e.is_transient(), "both no-node outcomes must QUEUE, not fail");
        }
    }

    #[test]
    fn a_young_vms_unconsumed_memory_is_not_handed_out_twice() {
        // Idle: the reading that gets remembered, and the slot count it implies.
        let idle = capacity_of(nid(3), "morpheus", 31757, 4314, 312, 0, None, 90.0)
            .expect("an idle morpheus fits VMs");
        assert_eq!(idle.slots, 2, "idle capacity is the number we are defending");

        // Two committed, both young. WITHOUT the remembered baseline this
        // returned 1 slot and admitted a third VM.
        let inferred = capacity_of(nid(3), "morpheus", 31757, 6314, 312, 2, None, 90.0);
        assert!(
            inferred.is_ok(),
            "the old inference is preserved as the no-baseline fallback"
        );

        // WITH it, the node is correctly full.
        let remembered = capacity_of(nid(3), "morpheus", 31757, 6314, 312, 2, Some(4314), 90.0);
        assert!(
            matches!(remembered, Err(Unfit::AtCapacity { committed: 2, .. })),
            "a 2-slot node with 2 VMs committed is FULL, got {remembered:?}"
        );
    }

    /// A host that starts doing real work outside its VMs is charged for it.
    ///
    /// The remembered baseline is a floor, not a substitute. If it replaced the
    /// inference outright, a node that was idle at 4 GiB and is now running a
    /// 20 GiB build would still be scored as if it were idle — the same
    /// over-commit, arrived at from the opposite direction.
    #[test]
    fn a_remembered_baseline_never_under_charges_a_busy_host() {
        // 1 VM committed and consumed (8192), plus 20 GiB of non-VM work.
        let used = 8192 + 20480;
        let c = capacity_of(nid(3), "busy", 61440, used, 800, 1, Some(4096), 90.0)
            .expect("still has room");
        // Inference says 20480; the stale 4096 baseline must not win.
        assert_eq!(c.used_eff_mib, used, "observed usage is charged in full");
    }

    #[test]
    fn a_sold_out_node_is_not_mistaken_for_an_idle_one() {
        let observed_only =
            capacity_of(nid(1), "tank", 61440, 12000, 800, 0, None, 50.0).expect("fits");
        assert_eq!(
            observed_only.slots, 5,
            "this is what utilisation alone claims — the bug being fixed"
        );

        let with_commitments =
            capacity_of(nid(1), "tank", 61440, 12000, 800, 5, None, 50.0).expect("fits");
        assert_eq!(
            with_commitments.slots, 1,
            "five 8 GiB claims are already spoken for, whatever the guests have touched"
        );
    }

    /// The measured idle fleet. Numbers from `free`/`df` on the real machines, so
    /// a future change to the constants has to face what it does to real nodes.
    #[test]
    fn the_measured_fleet_gets_the_slots_it_actually_has() {
        // tank: 60 GiB, ~6 GiB used at idle.
        let tank = capacity_of(nid(1), "tank", 61440, 6144, 869, 0, None, 90.0).unwrap();
        assert_eq!(tank.slots, 6);
        // architect: 60 GiB, ~7 GiB used.
        let arch = capacity_of(nid(2), "architect", 61440, 7168, 388, 0, None, 90.0).unwrap();
        assert_eq!(arch.slots, 6);
        // morpheus: 31 GiB — deliberately the conservative 2, not 3. Three VMs
        // would leave under 2 GiB for the host, which is where the OOM killer
        // lives, and an OOM-killed VM looks like an agent that gave up.
        let morph = capacity_of(nid(3), "morpheus", 31744, 5120, 312, 0, None, 90.0).unwrap();
        assert_eq!(morph.slots, 2);
    }

    /// Spread, don't stack; then real load; then determinism.
    #[test]
    fn ranking_prefers_free_slots_then_headroom_then_a_stable_order() {
        let a = capacity_of(nid(1), "a", 61440, 6144, 800, 0, None, 40.0).unwrap(); // 6 slots
        let b = capacity_of(nid(2), "b", 61440, 6144, 800, 3, None, 90.0).unwrap(); // 3 slots
        assert_eq!(rank(vec![b.clone(), a.clone()])[0].name, "a", "more slots wins");

        // Equal slots → the node under less real load.
        let busy = capacity_of(nid(3), "busy", 61440, 6144, 800, 0, None, 10.0).unwrap();
        let idle = capacity_of(nid(4), "idle", 61440, 6144, 800, 0, None, 95.0).unwrap();
        assert_eq!(rank(vec![busy.clone(), idle.clone()])[0].name, "idle");

        // Equal on both → same answer twice. `last_seen DESC` could not promise this.
        let x = capacity_of(nid(9), "x", 61440, 6144, 800, 0, None, 50.0).unwrap();
        let y = capacity_of(nid(8), "y", 61440, 6144, 800, 0, None, 50.0).unwrap();
        assert_eq!(rank(vec![x.clone(), y.clone()])[0].name, "y");
        assert_eq!(rank(vec![y, x])[0].name, "y");
    }

    /// A booted VM and its pinned phase row are ONE claim, not two.
    #[test]
    fn commitments_union_by_identity_rather_than_adding() {
        let live = vec!["m-abc123def456-0".to_string(), "m-abc123def456-0-s2".to_string()];
        // Same phase as the live VMs: already counted.
        assert_eq!(commitments(&live, &["m-abc123def456-0".to_string()]), 2);
        // A different phase, pinned but not yet booted: a real additional claim.
        assert_eq!(
            commitments(&live, &["m-999888777666-0".to_string()]),
            3,
            "a phase chosen seconds ago holds 8 GiB no node can report yet"
        );
        assert_eq!(commitments(&[], &["m-1-0".into(), "m-2-0".into()]), 2);
    }

    /// Disk is a hard gate, and it is checked BEFORE capacity so the message
    /// names the real problem.
    #[test]
    fn a_node_short_of_disk_is_refused_even_with_memory_to_spare() {
        let e = capacity_of(nid(1), "tank", 61440, 6144, 25, 0, None, 90.0).unwrap_err();
        assert!(matches!(e, Unfit::NoDisk { free_gib: 25 }), "{e:?}");
        assert!(e.reason().contains("25 GiB"));
    }

    /// Stale metrics may cost a tie; they may never win one, and they may never
    /// exclude a node — that is health's job.
    #[test]
    fn stale_beszel_metrics_demote_but_do_not_exclude() {
        let mut row = row_for(nid(1), 61440 * MIB_T, 6144 * MIB_T, 800 * GIB_T);
        row.metrics_age_secs = Some(3600.0);
        row.health_age_secs = Some(3.0);
        row.cpu_pct = Some(5.0);
        let fit = from_eval(&row, "tank", 0).expect("still eligible");
        assert_eq!(fit.headroom, 95.0, "fresh health carries the headroom");

        row.health_age_secs = Some(3600.0);
        assert!(
            matches!(from_eval(&row, "tank", 0), Err(Unfit::NoRecentHealth { .. })),
            "stale HEALTH is exclusion, because memory is then a guess"
        );
    }

    /// The two failures an operator must never confuse.
    #[test]
    fn unreadable_capacity_never_reads_as_a_full_fleet() {
        let full = PlacementError::FleetAtCapacity { report: "  tank: 0 slots".into() };
        let blind = PlacementError::FleetUnreadable { report: "  tank: UNFIT".into() };
        assert!(full.message().contains("at capacity"));
        assert!(blind.message().contains("cannot read"));
        assert!(
            !blind.message().contains("at capacity"),
            "sends an operator hunting a load problem that does not exist"
        );
        assert!(full.is_transient() && blind.is_transient());
        let missing = PlacementError::NoCapableNode {
            backend: "claude".into(),
            how_to_fix: "build the image".into(),
        };
        assert!(!missing.is_transient(), "a missing image will not fix itself by waiting");
    }

    const MIB_T: i64 = 1024 * 1024;
    const GIB_T: i64 = 1024 * 1024 * 1024;

    fn row_for(node_id: NodeId, total: i64, used: i64, disk_free: i64) -> EvalRow {
        EvalRow {
            node_id,
            workspace_id: cm_domain::WorkspaceId::from(uuid::Uuid::from_u128(1)),
            status: "online".into(),
            cpu_pct: None,
            mem_pct: None,
            disk_pct: None,
            gpu_pct: None,
            temp_max: None,
            load1: None,
            mem_total_bytes: Some(total),
            mem_used_bytes: Some(used),
            disk_free_bytes: Some(disk_free),
            mem_baseline_mib: None,
            health_age_secs: Some(3.0),
            metrics_age_secs: Some(3.0),
        }
    }

    /// A draining node is ineligible, not merely unattractive. The microVM path
    /// never checked this before: a mission pinned before a drain kept feeding
    /// VMs to a node an operator had cordoned.
    #[test]
    fn a_draining_node_is_ineligible() {
        let mut row = row_for(nid(1), 61440 * MIB_T, 6144 * MIB_T, 800 * GIB_T);
        row.status = "draining".into();
        assert_eq!(from_eval(&row, "tank", 0), Err(Unfit::Draining));
    }
}