draupnir 0.1.8

//! # Draupnir — the nordisk boot / provisioning library
//!
//! Draupnir is the **low-level engine that fires up a runtime instance** from a
//! single [`BootSpec`], across three backends, and drives its power lifecycle:
//!
//! - [`kvm`] — a **KVM/appliance VM**, booted by driving **tunnr**'s
//!   `tunnr_vm::boot_test(BootSpec) -> BootHandle` primitive (feature
//!   `backend-tunnr`). Draupnir does **not** reimplement VM boot — it wires the
//!   [`Boot`] trait against tunnr.
//! - [`container`] — an **OCI container**, brought up over a container runtime.
//! - [`redfish`] — **bare metal**, provisioned out-of-band through a BMC's
//!   **Redfish** REST API (iLO / iDRAC / OpenBMC): insert a virtual-media ISO,
//!   set the one-time boot override to that media, power the node on. This is
//!   the capability Draupnir uniquely owns.
//!
//! ## Where Draupnir sits
//!
//! Draupnir is the **shared low-level boot lib**. Two high-level consumers depend
//! on it **directly** and neither duplicates its boot code:
//!
//! - **jera** (edda's job handler) → depends on Draupnir for job instances
//!   (`process | VM | container`). jera stays thin: just job policy.
//! - **Skidbladnir** (service/systemd, airgap, orchestration) → depends on
//!   Draupnir for service instances.
//!
//! Draupnir itself knows nothing about jobs or services; it just fires up and
//! controls instances. For its KVM backend it calls **down** into tunnr.
//!
//! ```text
//!   jera ─────────┐
//!                 ├──▶ Draupnir ──▶ { tunnr (KVM) | OCI runtime | Redfish BMC }
//!   Skidbladnir ──┘
//! ```
//!
//! ## The mythological nod
//!
//! Draupnir is Odin's gold ring that drips **eight identical copies** of itself
//! every ninth night. Here that is a *natural extension, not the core*: booting a
//! **fleet of identical machines from one ISO** — [`plan_fleet`] fans one
//! [`BootSpec`] out into N identical specs (bare-metal fleet provisioning via
//! Redfish). The core is one library, three boot backends.

use std::collections::BTreeMap;
use std::fmt;

pub mod container;
pub mod kvm;
pub mod redfish;
pub mod seed;

/// Draupnir's result alias.
pub type Result<T> = std::result::Result<T, Error>;

/// **Introspection / emit marker** — record one functional-status row for the
/// nornir test matrix (the constellation-wide introspection-coverage gate).
/// Wraps `nornir_testmatrix::functional_status` behind the optional `testmatrix`
/// feature: ON, it emits a real matrix row nornir reads back; OFF, it is a
/// compiled-out `#[inline]` no-op with NO nornir dependency, so the lean default
/// build never pulls it. `component` is the reporting unit (e.g. `"draupnir/seed"`),
/// `check` what it verified, `ok` the verdict, `detail` a short human note. Mirrors
/// the sibling constellation crates (skidbladnir, ordning-core, korp-collectors).
#[inline]
pub fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
    #[cfg(feature = "testmatrix")]
    nornir_testmatrix::functional_status(component, check, ok, detail);
    #[cfg(not(feature = "testmatrix"))]
    {
        let _ = (component, check, ok, detail);
    }
}

/// Everything that can go wrong firing up or controlling an instance.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Error {
    /// The requested backend is not compiled in (build with its feature) or not
    /// available on this host.
    Unsupported(String),
    /// A live backend (tunnr / OCI runtime / Redfish BMC) reported a failure.
    Backend(String),
    /// The [`BootSpec`] is internally inconsistent (e.g. an ISO image handed to
    /// the KVM backend, or a Redfish spec with no BMC endpoint).
    Spec(String),
}

impl fmt::Display for Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Error::Unsupported(m) => write!(f, "draupnir: unsupported: {m}"),
            Error::Backend(m) => write!(f, "draupnir: backend error: {m}"),
            Error::Spec(m) => write!(f, "draupnir: invalid boot spec: {m}"),
        }
    }
}

impl std::error::Error for Error {}

/// Which runtime a [`BootSpec`] targets.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Backend {
    /// A KVM/appliance VM (driven through tunnr).
    Kvm,
    /// An OCI container.
    Container,
    /// A bare-metal node provisioned out-of-band via Redfish.
    Redfish,
}

/// The bootable payload — the *source* an instance is fired up from.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ImageSource {
    /// A kernel + rootfs/initramfs pair (the KVM appliance path → tunnr).
    KernelRootfs {
        /// Kernel image path (`-kernel`).
        kernel: String,
        /// Rootfs/initramfs path (`-initrd`).
        rootfs: String,
    },
    /// A bootable disk image, qcow2 or raw (the KVM disk path → tunnr). tunnr's
    /// direct-kernel launch still needs a `-kernel`, so the disk carries the
    /// kernel to boot it with explicitly (there is no in-image bootloader path).
    Disk {
        /// Kernel image path (`-kernel`) used to direct-boot the disk.
        kernel: String,
        /// Bootable disk image path (qcow2 or raw), attached as a virtio drive.
        disk: String,
    },
    /// An OCI image reference, e.g. `docker.io/library/redis:7` (container path).
    OciImage(String),
    /// A bootable ISO served as Redfish **virtual media** (bare-metal path).
    Iso(String),
}

impl ImageSource {
    /// Whether this payload is a legal source for `backend` — the KVM backend
    /// boots kernel+rootfs or a disk, the container backend an OCI image, and
    /// Redfish an ISO. Used by [`BootSpec::validate`].
    pub fn suits(&self, backend: Backend) -> bool {
        matches!(
            (self, backend),
            (ImageSource::KernelRootfs { .. }, Backend::Kvm)
                | (ImageSource::Disk { .. }, Backend::Kvm)
                | (ImageSource::OciImage(_), Backend::Container)
                | (ImageSource::Iso(_), Backend::Redfish)
        )
    }
}

/// A BMC (baseboard management controller) endpoint — the out-of-band Redfish
/// service on a bare-metal node (iLO / iDRAC / OpenBMC).
///
/// The secret (password / session token) is supplied out of band at drive time
/// and is deliberately **not** a field here, so a [`BootSpec`] never carries a
/// credential.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BmcEndpoint {
    /// Base URL of the Redfish service, e.g. `https://bmc-42.dc.example`.
    pub host: String,
    /// Redfish account username.
    pub username: String,
    /// The Redfish `ComputerSystem` resource id, e.g. `System.Embedded.1`.
    pub system_id: String,
}

/// The one-time boot device a Redfish node is overridden to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BootTarget {
    /// Boot from virtual media / CD (the ISO we inserted).
    Cd,
    /// Network / PXE.
    Pxe,
    /// The local disk.
    Hdd,
    /// Drop into BIOS/UEFI setup.
    BiosSetup,
}

/// The power state of an instance.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PowerState {
    /// Running.
    On,
    /// Powered off.
    Off,
    /// Not yet observed / indeterminate.
    Unknown,
}

/// The container **network mode** — how the OCI backend attaches the container to
/// a network. Only the [`container`](BootSpec::container) backend acts on it
/// (KVM/Redfish carry no container network). [`Default`](NetMode::Default) is the
/// runtime default (podman/Docker's own choice — **no** `--network` flag, so the
/// created `HostConfig.network_mode` stays unset and the create body is byte-
/// identical to a spec that never named a net mode). The airgap case is
/// [`None`](NetMode::None): it renders `--network none` and cuts the container off
/// from all egress — the load-bearing wire for Skidbladnir's airgap container route.
///
/// Mirrors jera's `ContainerSpec` `NetMode` field-for-field so a jera container-run
/// routes its net choice through this **one** draupnir OCI engine — jera passes its
/// rendered value across with [`NetMode::from_oci_value`]`(jera_spec.net.oci_value())`
/// (no cross-repo variant coupling; the wire is the OCI string `"none"`/`"host"`/`"bridge"`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum NetMode {
    /// The runtime default — **no** `--network` flag; `HostConfig.network_mode`
    /// stays unset (byte-identical to a spec that never set a net mode).
    #[default]
    Default,
    /// Airgap: `--network none` — the container gets no network (loopback only).
    None,
    /// Share the host network namespace (`--network host`).
    Host,
    /// The default bridge network (`--network bridge`).
    Bridge,
}

impl NetMode {
    /// The OCI `HostConfig.network_mode` value this renders to, or [`Option::None`]
    /// for [`Default`](NetMode::Default) (leaves the field unset → the runtime
    /// default). `Some("none")` is the airgap value the container backend threads
    /// onto the create body.
    pub fn oci_value(self) -> Option<&'static str> {
        match self {
            NetMode::Default => Option::None,
            NetMode::None => Some("none"),
            NetMode::Host => Some("host"),
            NetMode::Bridge => Some("bridge"),
        }
    }

    /// Whether this is [`Default`](NetMode::Default) (no net flag) — the additive-
    /// parity guard (a default-net spec must produce the unchanged create body).
    pub fn is_default(self) -> bool {
        matches!(self, NetMode::Default)
    }

    /// Reconstruct a `NetMode` from an OCI network-mode string — jera's
    /// `oci_value()` output: `Some("none"|"host"|"bridge")` maps to the matching
    /// variant; [`Option::None`] (or any unrecognised value) maps to
    /// [`Default`](NetMode::Default). This is the **cross-repo wire** jera passes
    /// without depending on draupnir's variant names.
    pub fn from_oci_value(value: Option<&str>) -> Self {
        match value {
            Some("none") => NetMode::None,
            Some("host") => NetMode::Host,
            Some("bridge") => NetMode::Bridge,
            _ => NetMode::Default,
        }
    }
}

/// **cloud-init NoCloud provisioning** for a KVM appliance boot: the `user-data`
/// (and optional `meta-data`) authored into a small FAT seed image (volume label
/// `cidata`) the guest's cloud-init picks up at first boot.
///
/// Only the [`kvm`] backend consumes it — containers have no init firstboot and
/// the Redfish path provisions the metal itself. It is pure data (zero deps); the
/// seed *image* is authored by the KVM adapter behind `backend-tunnr`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct CloudInit {
    /// The cloud-init `user-data` document (typically begins `#cloud-config`).
    pub user_data: String,
    /// The `meta-data` document; when `None` a minimal default carrying an
    /// `instance-id`/`local-hostname` is supplied by the seed builder.
    pub meta_data: Option<String>,
    /// The optional NoCloud `network-config` document (cloud-init network schema).
    /// `None` → no `network-config` file is written to the seed and the guest keeps
    /// its default (usually DHCP). Present → authored as the third seed file.
    pub network_config: Option<String>,
}

impl CloudInit {
    /// A NoCloud provision from a `user-data` document (default `meta-data`, no
    /// `network-config`).
    pub fn user_data(user_data: impl Into<String>) -> Self {
        Self { user_data: user_data.into(), meta_data: None, network_config: None }
    }

    /// Attach a NoCloud `network-config` document (builder style).
    pub fn with_network_config(mut self, network_config: impl Into<String>) -> Self {
        self.network_config = Some(network_config.into());
        self
    }
}

/// A container **published-port mapping** — a distinct `host:container` pair
/// (podman `-p HOST:CONTAINER`). This is the general publish form: a service
/// listening on a **fixed port inside** the container (FalkorDB always binds
/// `6379`, Spark-Connect `15002`) can be published on a **different host port**
/// so several isolated copies coexist on one host — e.g. per-zone offsets
/// (`Demo` on `6379`, `Test` on `6380`, `Prod` on `6381`) that all reach the
/// same in-container `6379`. Carried on [`BootSpec::port_maps`].
///
/// The single-port field ([`BootSpec::ports`], a bare `u16`) is exactly the
/// `host == container` special case and stays the byte-identical default — a
/// spec that uses only `ports` renders the same create body it always did. Both
/// forms are published (`ports` as `host==container`, `port_maps` as the pair),
/// so a spec may carry either or both.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct PortMap {
    /// The **host** port the publish binds on the host (the `HOST` in `-p
    /// HOST:CONTAINER`); this is what a client on the host connects to.
    pub host: u16,
    /// The **container** port the service listens on *inside* the container
    /// (the `CONTAINER` in `-p HOST:CONTAINER`); fixed by the image/service.
    pub container: u16,
}

impl PortMap {
    /// A distinct `host:container` publish — the general form (the host port
    /// may differ from the in-container port, e.g. a per-zone offset).
    pub fn new(host: u16, container: u16) -> Self {
        Self { host, container }
    }

    /// The `host == container` publish — the single-port special case (the same
    /// mapping the bare-`u16` [`BootSpec::ports`] form produces).
    pub fn same(port: u16) -> Self {
        Self { host: port, container: port }
    }
}

impl From<u16> for PortMap {
    /// A bare port maps `host == container` (the single-port form).
    fn from(port: u16) -> Self {
        Self::same(port)
    }
}

impl From<(u16, u16)> for PortMap {
    /// A `(host, container)` tuple is the distinct-mapping form.
    fn from((host, container): (u16, u16)) -> Self {
        Self::new(host, container)
    }
}

/// A **self-contained boot request**. One shape fires up any backend; the
/// [`backend`](BootSpec::backend) selects the driver and [`validate`] enforces
/// that the [`image`](BootSpec::image) (and, for Redfish, the [`bmc`]) match.
///
/// [`bmc`]: BootSpec::bmc
// NB: `PartialEq` only (not `Eq`): the container `cpus` cap is an `Option<f64>`,
// and `f64` is not `Eq`. Every `==`/`assert_eq!` on a `BootSpec` needs only
// `PartialEq`; `BootSpec` is never used as a hash/btree key, so dropping `Eq` is
// additive (no consumer required it).
#[derive(Debug, Clone, PartialEq)]
pub struct BootSpec {
    /// Human/instance name (also the fleet-member prefix).
    pub name: String,
    /// Which backend fires this up.
    pub backend: Backend,
    /// The bootable payload source.
    pub image: ImageSource,
    /// Guest/appliance RAM in MiB (ignored by the bare-metal Redfish path).
    pub mem_mb: u32,
    /// vCPU count (ignored by the bare-metal Redfish path).
    pub cores: u32,
    /// Kernel/boot command line, if the backend takes one.
    pub cmdline: String,
    /// Container command / entrypoint override (empty = the image's own default).
    /// Only the [`container`] backend acts on it; ignored by KVM/Redfish.
    pub cmd: Vec<String>,
    /// Container ports to publish, each bound to the **same host port**
    /// (`host == container`) — the single-port form. Only the [`container`]
    /// backend acts on them; ignored by KVM/Redfish. For a **distinct**
    /// `host:container` publish (a per-zone host offset onto a fixed in-container
    /// port) use [`port_maps`](BootSpec::port_maps); both are published.
    pub ports: Vec<u16>,
    /// Container **distinct-mapping** published ports — each a [`PortMap`]
    /// `{ host, container }` rendered as podman `-p host:container` /
    /// `HostConfig.port_bindings[container/tcp] = host`. This is the form that
    /// publishes a **different host port than the in-container port**: FalkorDB
    /// listens on `6379` inside every zone's container, but `Test` publishes it
    /// on host `6380` and `Prod` on `6381` (`PortMap::new(6380, 6379)`), so the
    /// zones don't collide on the host yet each reaches the fixed in-container
    /// port. Empty (the default) leaves the create body **byte-identical** to a
    /// spec that never named it; it composes with [`ports`](BootSpec::ports)
    /// (the `host == container` form) — both sets are published. Only the
    /// [`container`] backend acts on it; ignored by KVM/Redfish.
    pub port_maps: Vec<PortMap>,
    /// Environment for the instance (container env; appliance kernel env).
    pub env: BTreeMap<String, String>,
    /// The BMC endpoint — **required** for [`Backend::Redfish`], `None` otherwise.
    pub bmc: Option<BmcEndpoint>,
    /// Optional cloud-init NoCloud provisioning. Consumed by the [`kvm`] backend,
    /// which authors it into a seed image the guest reads at first boot; ignored by
    /// the container/Redfish backends. `None` → no seed is attached.
    pub cloud_init: Option<CloudInit>,
    /// Container **network mode** — how the OCI backend attaches the container to a
    /// network. Only the [`container`] backend acts on it (KVM/Redfish ignore it).
    /// [`NetMode::Default`] (the field's [`Default`]) leaves the create body
    /// byte-identical to a spec that never named a net mode; [`NetMode::None`]
    /// renders `--network none` (the airgap wire). Added additively — every existing
    /// constructor defaults it to [`NetMode::Default`].
    pub net: NetMode,
    /// Container **CPU quota** (podman `--cpus`) — how many host cores the container
    /// may use. `None` (the default) leaves it **unconstrained**, so the container
    /// sees **all** host cores: the deliberate default for hot infra (FalkorDB's
    /// OpenMP pool, a Spark executor) that must never be throttled to one core.
    /// `Some(n)` caps it at `n` cores (rendered as `HostConfig.nano_cpus = n * 1e9`).
    /// Only the [`container`] backend acts on it (KVM sizing is [`mem_mb`]/[`cores`]);
    /// `None` keeps the create body byte-identical to a spec that never named it.
    ///
    /// [`cores`]: BootSpec::cores
    pub cpus: Option<f64>,
    /// Container **memory limit** in MiB (podman `--memory`). `None` (the default)
    /// leaves it **unconstrained** (the container may use the box's memory) — the hot-
    /// infra default. `Some(m)` caps it at `m` MiB (`HostConfig.memory = m * 1MiB`).
    /// Only the [`container`] backend acts on it; the KVM guest RAM is the separate
    /// [`mem_mb`](BootSpec::mem_mb). `None` keeps the create body byte-identical.
    pub mem_limit_mb: Option<u32>,
}

impl BootSpec {
    /// A KVM appliance boot from a kernel + rootfs (defaults: 512 MiB, 2 cores).
    pub fn kvm_kernel_rootfs(
        name: impl Into<String>,
        kernel: impl Into<String>,
        rootfs: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            backend: Backend::Kvm,
            image: ImageSource::KernelRootfs { kernel: kernel.into(), rootfs: rootfs.into() },
            mem_mb: 512,
            cores: 2,
            cmdline: String::new(),
            cmd: Vec::new(),
            ports: Vec::new(),
            port_maps: Vec::new(),
            env: BTreeMap::new(),
            bmc: None,
            cloud_init: None,
            net: NetMode::Default,
            cpus: None,
            mem_limit_mb: None,
        }
    }

    /// A KVM boot from a bootable **disk image** direct-launched with `kernel`
    /// (defaults: 512 MiB, 2 cores). tunnr attaches the disk as a virtio drive;
    /// the kernel is required because there is no in-image bootloader path.
    pub fn kvm_disk(
        name: impl Into<String>,
        kernel: impl Into<String>,
        disk: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            backend: Backend::Kvm,
            image: ImageSource::Disk { kernel: kernel.into(), disk: disk.into() },
            mem_mb: 512,
            cores: 2,
            cmdline: String::new(),
            cmd: Vec::new(),
            ports: Vec::new(),
            port_maps: Vec::new(),
            env: BTreeMap::new(),
            bmc: None,
            cloud_init: None,
            net: NetMode::Default,
            cpus: None,
            mem_limit_mb: None,
        }
    }

    /// A container boot from an OCI image reference (e.g. a redis service).
    pub fn container(name: impl Into<String>, oci_image: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            backend: Backend::Container,
            image: ImageSource::OciImage(oci_image.into()),
            mem_mb: 0,
            cores: 0,
            cmdline: String::new(),
            cmd: Vec::new(),
            ports: Vec::new(),
            port_maps: Vec::new(),
            env: BTreeMap::new(),
            bmc: None,
            cloud_init: None,
            net: NetMode::Default,
            cpus: None,
            mem_limit_mb: None,
        }
    }

    /// A bare-metal Redfish boot: an ISO served as virtual media to a BMC node.
    pub fn redfish_iso(name: impl Into<String>, iso: impl Into<String>, bmc: BmcEndpoint) -> Self {
        Self {
            name: name.into(),
            backend: Backend::Redfish,
            image: ImageSource::Iso(iso.into()),
            mem_mb: 0,
            cores: 0,
            cmdline: String::new(),
            cmd: Vec::new(),
            ports: Vec::new(),
            port_maps: Vec::new(),
            env: BTreeMap::new(),
            bmc: Some(bmc),
            cloud_init: None,
            net: NetMode::Default,
            cpus: None,
            mem_limit_mb: None,
        }
    }

    /// Set an environment variable (builder style).
    pub fn with_env(mut self, key: impl Into<String>, val: impl Into<String>) -> Self {
        self.env.insert(key.into(), val.into());
        self
    }

    /// Set the container command / entrypoint override (builder style). Only the
    /// [`container`] backend acts on it.
    pub fn with_cmd<I, S>(mut self, cmd: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.cmd = cmd.into_iter().map(Into::into).collect();
        self
    }

    /// Publish a container port (builder style), bound to the same host port. Only
    /// the [`container`] backend acts on it.
    pub fn with_port(mut self, port: u16) -> Self {
        self.ports.push(port);
        self
    }

    /// Publish a **distinct** `host:container` mapping (builder style) — the host
    /// port may differ from the in-container port (a per-zone offset onto a fixed
    /// service port; see [`port_maps`](BootSpec::port_maps) and [`PortMap`]). Only
    /// the [`container`] backend acts on it.
    pub fn with_port_map(mut self, host: u16, container: u16) -> Self {
        self.port_maps.push(PortMap::new(host, container));
        self
    }

    /// Attach cloud-init NoCloud provisioning (builder style). Only the KVM backend
    /// acts on it — it authors a seed image the guest reads at first boot.
    pub fn with_cloud_init(mut self, ci: CloudInit) -> Self {
        self.cloud_init = Some(ci);
        self
    }

    /// Set the container **network mode** (builder style). Only the [`container`]
    /// backend acts on it; [`NetMode::None`] is the airgap `--network none` case.
    /// [`NetMode::Default`] (unchanged) leaves the create body byte-identical.
    pub fn with_net(mut self, net: NetMode) -> Self {
        self.net = net;
        self
    }

    /// Set the container **CPU quota** (podman `--cpus`, builder style) — see
    /// [`cpus`](BootSpec::cpus). `None` (unset) leaves the container **unconstrained**
    /// (all host cores); `Some(n)` caps it at `n` cores. Only the [`container`]
    /// backend acts on it.
    pub fn with_cpus(mut self, cpus: f64) -> Self {
        self.cpus = Some(cpus);
        self
    }

    /// Set the container **memory limit** in MiB (podman `--memory`, builder style)
    /// — see [`mem_limit_mb`](BootSpec::mem_limit_mb). `Some(m)` caps it at `m` MiB;
    /// unset leaves it unconstrained. Only the [`container`] backend acts on it.
    pub fn with_mem_limit_mb(mut self, mem_mb: u32) -> Self {
        self.mem_limit_mb = Some(mem_mb);
        self
    }

    /// Reject an internally inconsistent spec **before** touching a backend:
    /// the image must suit the backend, a Redfish spec must carry a BMC (and no
    /// other backend may), every required payload path/ref is non-empty, the
    /// instance `name` is non-empty, a Redfish `bmc` carries non-empty
    /// host/username/system-id, a KVM spec is sized (`mem_mb`/`cores` > 0), the
    /// container-only `cmd`/`ports` are not set on a non-container backend, and
    /// every published container `port` is non-zero and listed at most once.
    /// This is pure and unit-tested.
    pub fn validate(&self) -> Result<()> {
        // Reject an empty/all-whitespace required string up front so a mistyped
        // spec fails here with a clear message, not verbatim-passed to a backend
        // that only fails deep in a runtime call (tunnr's direct-kernel launch
        // needs a real `-kernel`; the OCI daemon a real image ref; Redfish a real
        // ISO and a reachable BMC). `require` trims, so whitespace is caught too.
        let require = |what: &str, val: &str| -> Result<()> {
            if val.trim().is_empty() {
                Err(Error::Spec(format!("a {:?} boot needs a non-empty {what}", self.backend)))
            } else {
                Ok(())
            }
        };
        // The instance name is the `Machine` id prefix and the fleet-member prefix
        // (`plan_fleet` mints `"{name}-{i}"`); a blank one yields `"-1"`/`"-2"`
        // members and a leading-dash id, so reject it on every backend.
        require("instance name", &self.name)?;
        if !self.image.suits(self.backend) {
            return Err(Error::Spec(format!(
                "{:?} image is not bootable by the {:?} backend",
                self.image, self.backend
            )));
        }
        match &self.image {
            ImageSource::KernelRootfs { kernel, rootfs } => {
                require("kernel path", kernel)?;
                require("rootfs path", rootfs)?;
            }
            ImageSource::Disk { kernel, disk } => {
                require("kernel path", kernel)?;
                require("disk path", disk)?;
            }
            ImageSource::OciImage(image) => require("OCI image reference", image)?,
            ImageSource::Iso(iso) => require("ISO path", iso)?,
        }
        // A KVM VM is booted with the spec's RAM/vCPU verbatim (the tunnr adapter
        // sets `mem_mb`/`cores` from these); 0 would launch a 0-RAM/0-CPU guest
        // that dies at boot, so a KVM spec must be sized. Container/Redfish carry
        // no VM sizing (both default to 0 by design) and are exempt.
        if self.backend == Backend::Kvm {
            if self.mem_mb == 0 {
                return Err(Error::Spec("a Kvm boot needs mem_mb > 0 (VM RAM)".into()));
            }
            if self.cores == 0 {
                return Err(Error::Spec("a Kvm boot needs cores > 0 (vCPUs)".into()));
            }
        }
        // `cmd` (entrypoint override) and `ports` (published ports) are
        // container-only knobs: only the container backend runs the cmd and
        // publishes the ports — KVM/Redfish silently ignore both. Setting either
        // on a non-container backend is a misconfiguration that would vanish
        // without a trace (the caller asked to run a command / expose a port that
        // never happens), so reject it up front — parity with a `bmc` on a
        // non-Redfish backend (rejected below) and the same shape as the empty /
        // zero / duplicate required-field checks.
        if self.backend != Backend::Container {
            if !self.cmd.is_empty() {
                return Err(Error::Spec(format!(
                    "a {:?} boot takes no container cmd (cmd is container-only)",
                    self.backend
                )));
            }
            if !self.ports.is_empty() || !self.port_maps.is_empty() {
                return Err(Error::Spec(format!(
                    "a {:?} boot publishes no ports (ports are container-only)",
                    self.backend
                )));
            }
            // `net` (the container network mode) is likewise container-only: only
            // the container backend threads it onto the OCI create body — KVM/Redfish
            // have no container network, so a non-default net there is a
            // misconfiguration that would vanish without a trace (an airgap the caller
            // asked for that never happens). Reject it up front (parity with cmd/ports
            // above). `NetMode::Default` is the no-op and always allowed.
            if !self.net.is_default() {
                return Err(Error::Spec(format!(
                    "a {:?} boot takes no container network mode (net is container-only)",
                    self.backend
                )));
            }
            // `cpus` (podman `--cpus`) and `mem_limit_mb` (podman `--memory`) are
            // container resource knobs the OCI backend threads onto `HostConfig`;
            // KVM sizing is the separate `mem_mb`/`cores`, and Redfish provisions the
            // metal itself — so a container CPU/memory cap on a non-container backend
            // is a misconfiguration that would vanish without a trace. Reject it up
            // front (parity with cmd/ports/net above).
            if self.cpus.is_some() {
                return Err(Error::Spec(format!(
                    "a {:?} boot takes no container cpus quota (cpus is container-only)",
                    self.backend
                )));
            }
            if self.mem_limit_mb.is_some() {
                return Err(Error::Spec(format!(
                    "a {:?} boot takes no container memory limit (mem_limit_mb is container-only)",
                    self.backend
                )));
            }
        }
        // A container CPU/memory cap, when set, must be positive: `--cpus 0` /
        // `--memory 0` are not a real cap (they would either error at the daemon or
        // mean "unlimited", which is what `None` already expresses). Reject a
        // non-positive value here rather than pass a nonsense limit to the engine.
        if let Some(c) = self.cpus {
            if !c.is_finite() || c <= 0.0 {
                return Err(Error::Spec(
                    "a container cpus quota must be a finite value > 0 (use None for all host cores)".into(),
                ));
            }
        }
        if self.mem_limit_mb == Some(0) {
            return Err(Error::Spec(
                "a container memory limit must be > 0 MiB (use None for unconstrained)".into(),
            ));
        }
        // Published container ports are exposed and host-bound verbatim by the
        // container backend (`ports`: `{p}/tcp` → host_port `{p}`; `port_maps`:
        // `{container}/tcp` → host_port `{host}`). Port `0` is never a real
        // published port (it would expose `0/tcp` / bind host port 0, which is
        // not what any caller means), and the same **host** port bound twice —
        // whether from `ports`, `port_maps`, or a mix — is a self-colliding host
        // binding, so reject both up front (parity with the other required-field
        // checks). By the guard above, a non-empty ports set here is necessarily
        // a container spec. Two different host ports mapping to the same
        // *container* port is fine (that is exactly the multi-zone case).
        if !self.ports.is_empty() || !self.port_maps.is_empty() {
            let mut seen_host = std::collections::BTreeSet::new();
            // `ports` (the host==container single-port form).
            for &p in &self.ports {
                if p == 0 {
                    return Err(Error::Spec(
                        "a published container port must be > 0".into(),
                    ));
                }
                if !seen_host.insert(p) {
                    return Err(Error::Spec(format!(
                        "published container port {p} is listed twice"
                    )));
                }
            }
            // `port_maps` (the distinct host:container form): both ends must be a
            // real port, and the host port must not collide with any already
            // bound (from `ports` or an earlier map).
            for pm in &self.port_maps {
                if pm.host == 0 || pm.container == 0 {
                    return Err(Error::Spec(
                        "a published container port map needs host > 0 and container > 0".into(),
                    ));
                }
                if !seen_host.insert(pm.host) {
                    return Err(Error::Spec(format!(
                        "published container host port {} is bound twice",
                        pm.host
                    )));
                }
            }
        }
        match (self.backend, &self.bmc) {
            (Backend::Redfish, None) => {
                Err(Error::Spec("Redfish boot needs a BMC endpoint".into()))
            }
            (Backend::Redfish, Some(bmc)) => {
                // The BMC fields are woven into the Redfish REST URLs the backend
                // drives; a blank host/username/system-id yields a malformed
                // request that only fails on the wire, so require them here too
                // (parity with the non-empty ISO path check above).
                require("BMC host", &bmc.host)?;
                require("BMC username", &bmc.username)?;
                require("BMC system id", &bmc.system_id)?;
                Ok(())
            }
            (_, Some(_)) => Err(Error::Spec(
                "only the Redfish backend takes a BMC endpoint".into(),
            )),
            (_, None) => Ok(()),
        }
    }
}

/// A booted (or booting) instance handle — what a [`Boot::boot`] returns and
/// what [`Lifecycle`] acts on.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Machine {
    /// Backend-scoped instance id (VM handle / container id / Redfish system id).
    pub id: String,
    /// The [`BootSpec::name`] this was fired up from.
    pub spec_name: String,
    /// Which backend owns it.
    pub backend: Backend,
    /// Last-observed power state.
    pub power: PowerState,
}

impl Machine {
    /// Record a freshly fired-up instance (power state assumed `On`).
    pub fn started(id: impl Into<String>, spec: &BootSpec) -> Self {
        Self {
            id: id.into(),
            spec_name: spec.name.clone(),
            backend: spec.backend,
            power: PowerState::On,
        }
    }
}

/// **Fire up** an instance from a [`BootSpec`]. One trait, three implementations
/// ([`kvm::KvmBoot`], [`container::ContainerBoot`], [`redfish::RedfishBoot`]).
pub trait Boot {
    /// Boot the instance described by `spec`, returning its live [`Machine`].
    fn boot(&self, spec: &BootSpec) -> Result<Machine>;
}

/// Drive an instance's **power lifecycle** after it is fired up.
pub trait Lifecycle {
    /// Power the instance on.
    fn power_on(&self, machine: &Machine) -> Result<()>;
    /// Power the instance off.
    fn power_off(&self, machine: &Machine) -> Result<()>;
    /// Observe the instance's current power state.
    fn status(&self, machine: &Machine) -> Result<PowerState>;
}

/// **Redfish virtual-media + boot-override** control — the out-of-band steps that
/// make a bare-metal node boot our ISO. Only the [`redfish::RedfishBoot`] backend
/// implements it; the KVM/container backends have no BMC.
pub trait VirtualMedia {
    /// Attach `iso` to the node as Redfish virtual media (CD/DVD).
    fn insert_media(&self, node: &BmcEndpoint, iso: &str) -> Result<()>;
    /// Detach any virtual media from the node.
    fn eject_media(&self, node: &BmcEndpoint) -> Result<()>;
    /// Set the node's **one-time** boot override to `target`.
    fn set_boot_override(&self, node: &BmcEndpoint, target: BootTarget) -> Result<()>;
}

/// **The unifying entry point** — fire up one instance from a [`BootSpec`] across
/// *whichever* backend is handed in. It [`validate`](BootSpec::validate)s the spec
/// first (so a mismatched image/BMC is rejected before any backend is touched),
/// then delegates to the backend's [`Boot::boot`]. The same `spec` boots the same
/// image on a [`kvm::KvmBoot`], a [`container::ContainerBoot`], or a
/// [`redfish::RedfishBoot`] — one call, three backends.
pub fn boot(spec: &BootSpec, backend: &dyn Boot) -> Result<Machine> {
    spec.validate()?;
    backend.boot(spec)
}

/// **The ring drips eight copies** — fan one [`BootSpec`] out into `n` identical
/// specs, each with a distinct `"{name}-{i}"` name (1-based), for booting a
/// fleet of identical machines from one image/ISO.
///
/// Pure bookkeeping: it plans the fleet; the caller boots each member through the
/// backend. Unit-tested.
pub fn plan_fleet(spec: &BootSpec, n: usize) -> Vec<BootSpec> {
    (1..=n)
        .map(|i| {
            let mut member = spec.clone();
            member.name = format!("{}-{i}", spec.name);
            member
        })
        .collect()
}

/// **Drip a fleet from one image** — [`plan_fleet`] the spec into `n` members and
/// [`boot`] each through `backend`, returning a per-member result (a partial fleet
/// is observable: some members may boot while a later one errors).
pub fn boot_fleet(spec: &BootSpec, n: usize, backend: &dyn Boot) -> Vec<Result<Machine>> {
    plan_fleet(spec, n)
        .iter()
        .map(|member| boot(member, backend))
        .collect()
}

// ---------------------------------------------------------------------------
// Power-state readback — confirm a booted instance actually reached a state.
// ---------------------------------------------------------------------------

/// Knobs for [`await_power_state`] / [`boot_and_await`]: how long to wait for the
/// instance to reach the target power state and how often to poll it. [`Default`]
/// waits **indefinitely** (parity with [`container::RunOptions`]) and polls every
/// 200 ms — a bounded budget ([`WaitOptions::bounded`]) is recommended for a boot
/// readback so a node that never comes up is a timeout, not a hang.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WaitOptions {
    /// Overall budget before giving up. `None` = wait forever for the state.
    /// `Some(d)` returns an [`Error::Backend`] if the state is not reached within `d`.
    pub timeout: Option<std::time::Duration>,
    /// How often [`Lifecycle::status`] is polled while waiting.
    pub poll_interval: std::time::Duration,
}

impl Default for WaitOptions {
    fn default() -> Self {
        Self { timeout: None, poll_interval: std::time::Duration::from_millis(200) }
    }
}

impl WaitOptions {
    /// Wait indefinitely, polling on `poll_interval`.
    pub fn poll_every(poll_interval: std::time::Duration) -> Self {
        Self { timeout: None, poll_interval }
    }

    /// Cap the wait at `timeout`, polling on `poll_interval`.
    pub fn bounded(timeout: std::time::Duration, poll_interval: std::time::Duration) -> Self {
        Self { timeout: Some(timeout), poll_interval }
    }
}

/// **Confirm an instance reached a power state** — poll a [`Lifecycle`]'s
/// [`status`](Lifecycle::status) until it reports `want`, returning `Ok(())` the
/// moment it does. This is the cross-backend **boot-status readback** seam: a
/// [`Boot::boot`] fires an instance up but returns before it has actually powered
/// on (a KVM guest is still booting, a Redfish node is still POSTing, a container is
/// still being scheduled), so a consumer (jera / Skidbladnir) that needs to *know*
/// the instance is up polls this — the power-lifecycle analogue of the container
/// [`run_to_completion`](container::run_to_completion) drive loop, but written
/// against the plain [`Lifecycle`] trait so it drives **any** backend (KVM,
/// container, Redfish) and a mock in a unit test with no live instance.
///
/// A live backend's own `status` error propagates as `Err` (the readback failed).
/// A `want` of [`PowerState::Unknown`] is a nonsensical target (`Unknown` means "not
/// observed") and is rejected as an [`Error::Spec`] before polling. With a bounded
/// [`WaitOptions`] a state never reached is an [`Error::Backend`] timeout carrying
/// the last-observed state; with the default (unbounded) options it waits forever.
pub fn await_power_state<L: Lifecycle>(
    lifecycle: &L,
    machine: &Machine,
    want: PowerState,
    opts: &WaitOptions,
) -> Result<()> {
    if want == PowerState::Unknown {
        return Err(Error::Spec(
            "cannot await PowerState::Unknown (it means \"not observed\")".into(),
        ));
    }
    let deadline = opts.timeout.map(|t| std::time::Instant::now() + t);
    loop {
        let observed = lifecycle.status(machine)?;
        if observed == want {
            functional_status(
                "draupnir/lifecycle",
                "await_power_state",
                true,
                &format!("instance {} reached {want:?}", machine.id),
            );
            return Ok(());
        }
        if let Some(dl) = deadline {
            if std::time::Instant::now() >= dl {
                functional_status(
                    "draupnir/lifecycle",
                    "await_power_state",
                    false,
                    &format!("instance {} never reached {want:?}", machine.id),
                );
                return Err(Error::Backend(format!(
                    "instance `{}` did not reach {want:?} within {:?} (last observed {observed:?})",
                    machine.id,
                    opts.timeout.unwrap()
                )));
            }
        }
        std::thread::sleep(opts.poll_interval);
    }
}

/// **Boot an instance and confirm it is up** — the one-call provision seam jera /
/// Skidbladnir want: [`validate`](BootSpec::validate) + [`boot`] the `spec` on
/// `backend`, then [`await_power_state`] it to [`PowerState::On`], returning the
/// live [`Machine`] only once it has actually powered on. A validation or boot
/// failure short-circuits before any wait (the backend is never touched on an
/// invalid spec — [`boot`] enforces that); a boot that never comes up within a
/// bounded [`WaitOptions`] is a timeout [`Error::Backend`]. Written against
/// `Boot + Lifecycle` so it drives any backend and a mock alike.
pub fn boot_and_await<B>(backend: &B, spec: &BootSpec, opts: &WaitOptions) -> Result<Machine>
where
    B: Boot + Lifecycle,
{
    let machine = boot(spec, backend)?;
    await_power_state(backend, &machine, PowerState::On, opts)?;
    Ok(machine)
}

// ---------------------------------------------------------------------------
// Fleet-level boot-status readback — boot N members and roll up who is ready.
// ---------------------------------------------------------------------------

/// The boot-readback verdict for **one** fleet member in a
/// [`boot_fleet_and_await`] rollup — the three ways a member can land.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MemberOutcome {
    /// Booted **and confirmed** at [`PowerState::On`] within the wait budget — the
    /// member is ready. Carries the live [`Machine`] handle.
    Up(Machine),
    /// The backend **did not get the member to [`PowerState::On`]** within the
    /// bounded budget: the readback timed out (booted but never powered on) or the
    /// backend failed / went unreachable while bringing it up (an [`Error::Backend`]
    /// — the fleet's "this node did not come up in time" bucket). Carries the
    /// failure detail. Counts as **failed**, never blocks the rest of the rollup.
    Timeout(String),
    /// The member could **not even be launched**: its spec was invalid or the
    /// backend is not available ([`Error::Spec`] / [`Error::Unsupported`]) — a hard
    /// misconfiguration that waiting can never fix. Carries the failure detail.
    /// Counts as **failed**.
    Error(String),
}

/// One member's line in a [`FleetReadback`]: its name paired with its verdict.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberReadback {
    /// The fleet member's name (`"{spec.name}-{i}"`, as minted by [`plan_fleet`]).
    pub name: String,
    /// Its boot-readback [`outcome`](MemberOutcome).
    pub outcome: MemberOutcome,
}

/// **The fleet-level boot-readback rollup** returned by [`boot_fleet_and_await`]:
/// one verdict per member (in fleet order `node-1`, `node-2`, …) plus the aggregate
/// ready/failed tallies a dispatcher (jera) reads to decide whether the fleet is up.
///
/// It is *infallible by construction* — a dead or misconfigured member is a
/// per-member [`MemberOutcome::Timeout`]/[`Error`](MemberOutcome::Error) line, never
/// an early return, so a partial fleet is always observable (like [`boot_fleet`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FleetReadback {
    /// Per-member verdicts, in fleet order.
    pub members: Vec<MemberReadback>,
}

impl FleetReadback {
    /// How many members booted **and** confirmed [`PowerState::On`].
    pub fn ready(&self) -> usize {
        self.members.iter().filter(|m| matches!(m.outcome, MemberOutcome::Up(_))).count()
    }

    /// How many members did **not** confirm up (timeout **or** error) — the
    /// complement of [`ready`](Self::ready).
    pub fn failed(&self) -> usize {
        self.members.len() - self.ready()
    }

    /// `true` only when the fleet is non-empty and **every** member is up.
    pub fn all_ready(&self) -> bool {
        !self.members.is_empty() && self.failed() == 0
    }
}

/// **Boot a fleet and roll up who is actually ready** — the fleet-level analogue of
/// [`boot_and_await`], and the multi-machine provision-readback seam jera's
/// dispatcher wants (jera Roster/WorkPayload → draupnir boots the fleet → this rolls
/// up who came up). It [`plan_fleet`]s `spec` into `n` members and drives each
/// through [`boot_and_await`] (validate → [`boot`] → confirm [`PowerState::On`]),
/// classifying every member's result into a [`MemberOutcome`] and returning the
/// [`FleetReadback`] rollup (per-member verdict + aggregate ready/failed counts).
///
/// **Pass a bounded [`WaitOptions`]** ([`WaitOptions::bounded`]): the per-member
/// timeout is what keeps one dead node from hanging the whole rollup — each member is
/// awaited independently within `opts`, so a member that never powers on rolls up as
/// a [`MemberOutcome::Timeout`] while the healthy members are [`Up`](MemberOutcome::Up)
/// (the default unbounded `WaitOptions` would block forever on the first dead node).
/// Written against `Boot + Lifecycle` so it drives any backend (KVM / container /
/// Redfish) and a mock alike, exactly like [`boot_and_await`] — no per-backend twin.
pub fn boot_fleet_and_await<B>(
    spec: &BootSpec,
    n: usize,
    backend: &B,
    opts: &WaitOptions,
) -> FleetReadback
where
    B: Boot + Lifecycle,
{
    let members =
        plan_fleet(spec, n).iter().map(|member| classify_member(backend, member, opts)).collect();
    let rollup = FleetReadback { members };
    functional_status(
        "draupnir/lifecycle",
        "boot_fleet_and_await",
        rollup.all_ready(),
        &format!("fleet of {n}: {} ready, {} failed", rollup.ready(), rollup.failed()),
    );
    rollup
}

/// **Boot one fleet member and classify how it landed** — the shared per-member step
/// behind both the serial [`boot_fleet_and_await`] and the parallel
/// [`boot_fleet_and_await_parallel`], so the two paths produce a **byte-identical**
/// [`MemberReadback`] for the same member by construction (there is exactly ONE
/// classification, never a twin — L5). Drives the member through [`boot_and_await`]
/// (validate → [`boot`] → confirm [`PowerState::On`]) and buckets the result:
/// `Ok` → [`Up`](MemberOutcome::Up); `Err(Backend)` → [`Timeout`](MemberOutcome::Timeout)
/// (the backend did not get it up in budget — a readback timeout or a transient boot
/// failure); `Err(Spec | Unsupported)` → [`Error`](MemberOutcome::Error) (a hard
/// misconfiguration / no backend — waiting cannot help).
fn classify_member<B>(backend: &B, member: &BootSpec, opts: &WaitOptions) -> MemberReadback
where
    B: Boot + Lifecycle,
{
    let outcome = match boot_and_await(backend, member, opts) {
        Ok(machine) => MemberOutcome::Up(machine),
        // A backend failure to bring the member up in time (readback timeout or a
        // live backend error while booting/polling) — the node did not come up
        // within the budget.
        Err(Error::Backend(msg)) => MemberOutcome::Timeout(msg),
        // A hard misconfiguration (invalid spec) or an unavailable backend — waiting
        // cannot help, so it is an error, not a timeout.
        Err(Error::Spec(msg)) | Err(Error::Unsupported(msg)) => MemberOutcome::Error(msg),
    };
    MemberReadback { name: member.name.clone(), outcome }
}

/// **Boot a fleet CONCURRENTLY and roll up who is ready** — the parallel sibling of
/// [`boot_fleet_and_await`]. Where the serial call awaits members one-at-a-time (so a
/// fleet's wall-clock is the *sum* of the per-member waits), this fans every member
/// onto its own scoped thread ([`std::thread::scope`] — **pure std, no new
/// dependency**) so `n` nodes are booted and awaited **at the same time** and the
/// wall-clock collapses to roughly the *slowest* member's wait, not the sum. This is
/// the real win for jera's multi-machine dispatcher: booting a fleet of `n` nodes with
/// a 5-minute per-member budget takes ~5 minutes, not ~`5n` minutes.
///
/// It produces an **identical [`FleetReadback`]** to the serial call for the same
/// inputs — same per-member [`MemberOutcome`], same **fleet order** (`node-1`…`node-n`,
/// not completion order: the threads are joined back in fleet order), same aggregate
/// tallies — because both paths classify each member through the one shared
/// [`classify_member`] step. It is likewise *infallible by construction*: a dead or
/// misconfigured member is its own per-member line, and — because each member is
/// awaited on its **own** thread within the bounded `opts` — one dead node cannot hang
/// the others (they finish independently and are joined). **Pass a bounded
/// [`WaitOptions`]** for the same reason as the serial call.
///
/// Requires `B: Sync` (the backend is shared `&B` across the scoped threads); all three
/// real backends ([`kvm::KvmBoot`], [`container::ContainerBoot`], [`redfish::RedfishBoot`])
/// satisfy it. A backend that is not `Sync` simply uses the serial
/// [`boot_fleet_and_await`] instead — the serial path stays the fully-general fallback
/// (L2: the working path is kept, never replaced).
pub fn boot_fleet_and_await_parallel<B>(
    spec: &BootSpec,
    n: usize,
    backend: &B,
    opts: &WaitOptions,
) -> FleetReadback
where
    B: Boot + Lifecycle + Sync,
{
    let plan = plan_fleet(spec, n);
    // Fan each member onto its own scoped thread; join back in fleet order so the
    // rollup is deterministic and order-identical to the serial path regardless of
    // which member finishes first.
    let members: Vec<MemberReadback> = std::thread::scope(|scope| {
        let handles: Vec<_> = plan
            .iter()
            .map(|member| scope.spawn(move || classify_member(backend, member, opts)))
            .collect();
        handles.into_iter().map(|h| h.join().expect("fleet member thread panicked")).collect()
    });
    let rollup = FleetReadback { members };
    functional_status(
        "draupnir/lifecycle",
        "boot_fleet_and_await_parallel",
        rollup.all_ready(),
        &format!("fleet of {n} (parallel): {} ready, {} failed", rollup.ready(), rollup.failed()),
    );
    rollup
}

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

    fn bmc() -> BmcEndpoint {
        BmcEndpoint {
            host: "https://bmc-42.dc.example".into(),
            username: "admin".into(),
            system_id: "System.Embedded.1".into(),
        }
    }

    #[test]
    fn image_source_suits_the_right_backend() {
        assert!(ImageSource::Disk { kernel: "/bzImage".into(), disk: "/d.qcow2".into() }
            .suits(Backend::Kvm));
        assert!(ImageSource::OciImage("redis:7".into()).suits(Backend::Container));
        assert!(ImageSource::Iso("/boot.iso".into()).suits(Backend::Redfish));
        // Cross pairings are rejected.
        assert!(!ImageSource::Iso("/boot.iso".into()).suits(Backend::Kvm));
        assert!(!ImageSource::OciImage("redis:7".into()).suits(Backend::Redfish));
    }

    #[test]
    fn valid_specs_pass_validation() {
        BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
            .validate()
            .unwrap();
        BootSpec::container("cache", "docker.io/library/redis:7")
            .validate()
            .unwrap();
        BootSpec::redfish_iso("node-42", "/images/installer.iso", bmc())
            .validate()
            .unwrap();
        BootSpec::kvm_disk("disky", "/bzImage", "/disk.qcow2")
            .validate()
            .unwrap();
    }

    #[test]
    fn disk_boot_without_a_kernel_is_rejected() {
        // The whole point of the fix: a Disk spec that carries no kernel would
        // pass an empty `-kernel` to tunnr and fail at runtime, so reject it here.
        let mut spec = BootSpec::kvm_disk("disky", "", "/disk.qcow2");
        assert!(matches!(spec.validate(), Err(Error::Spec(_))));
        // ...and an empty disk path is likewise rejected.
        spec = BootSpec::kvm_disk("disky", "/bzImage", "");
        assert!(matches!(spec.validate(), Err(Error::Spec(_))));
    }

    #[test]
    fn empty_required_payload_paths_are_rejected_on_every_image_source() {
        // Parity with the Disk checks: an empty required path/ref on ANY image
        // source is rejected at validate() rather than handed empty to a backend.
        // RED-when-broken — drop any arm's `require` and one of these passes.

        // KernelRootfs: empty kernel, then empty rootfs.
        let mut kr = BootSpec::kvm_kernel_rootfs("kr", "", "/rootfs.cpio.gz");
        assert!(matches!(kr.validate(), Err(Error::Spec(_))), "empty kernel rejected");
        kr = BootSpec::kvm_kernel_rootfs("kr", "/bzImage", "   ");
        assert!(matches!(kr.validate(), Err(Error::Spec(_))), "whitespace rootfs rejected");

        // OciImage: an empty image reference is not bootable.
        let oci = BootSpec::container("cache", "");
        assert!(matches!(oci.validate(), Err(Error::Spec(_))), "empty OCI ref rejected");

        // Iso: an empty ISO path is not bootable (BMC present so only the ISO fails).
        let iso = BootSpec::redfish_iso("node", "  ", bmc());
        assert!(matches!(iso.validate(), Err(Error::Spec(_))), "empty ISO path rejected");

        // The well-formed constructors still pass (no regression).
        BootSpec::kvm_kernel_rootfs("kr", "/bzImage", "/rootfs.cpio.gz").validate().unwrap();
        BootSpec::container("cache", "redis:7").validate().unwrap();
        BootSpec::redfish_iso("node", "/boot.iso", bmc()).validate().unwrap();
    }

    #[test]
    fn blank_instance_name_is_rejected_on_every_backend() {
        // The name is the Machine-id / fleet-member prefix; a blank one yields
        // "-1"/"-2" members and a leading-dash id, so reject it up front.
        // RED-when-broken — drop the `require("instance name", …)` and these pass.
        let mut vm = BootSpec::kvm_kernel_rootfs("", "/bzImage", "/rootfs.cpio.gz");
        assert!(matches!(vm.validate(), Err(Error::Spec(_))), "blank KVM name rejected");
        vm = BootSpec::kvm_kernel_rootfs("   ", "/bzImage", "/rootfs.cpio.gz");
        assert!(matches!(vm.validate(), Err(Error::Spec(_))), "whitespace KVM name rejected");

        let ctr = BootSpec::container("", "redis:7");
        assert!(matches!(ctr.validate(), Err(Error::Spec(_))), "blank container name rejected");

        let node = BootSpec::redfish_iso("", "/boot.iso", bmc());
        assert!(matches!(node.validate(), Err(Error::Spec(_))), "blank Redfish name rejected");
    }

    #[test]
    fn kvm_spec_without_ram_or_cpus_is_rejected() {
        // A KVM VM is booted with the spec's RAM/vCPU verbatim; 0 launches a dead
        // guest. RED-when-broken — drop either sizing guard and one of these passes.
        let mut vm = BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz");
        vm.mem_mb = 0;
        assert!(matches!(vm.validate(), Err(Error::Spec(_))), "0 MiB RAM rejected");
        vm = BootSpec::kvm_disk("disky", "/bzImage", "/disk.qcow2");
        vm.cores = 0;
        assert!(matches!(vm.validate(), Err(Error::Spec(_))), "0 vCPUs rejected");
        // Container/Redfish carry no VM sizing (both default to 0) and stay valid.
        BootSpec::container("cache", "redis:7").validate().unwrap();
        BootSpec::redfish_iso("node", "/boot.iso", bmc()).validate().unwrap();
    }

    #[test]
    fn redfish_spec_with_a_blank_bmc_field_is_rejected() {
        // Parity with the ISO-path check: each BMC field is woven into a Redfish
        // URL, so a blank one is rejected at validate(), not on the wire.
        // RED-when-broken — drop any `require("BMC …", …)` and its arm passes.
        let mut spec = BootSpec::redfish_iso("node", "/boot.iso", bmc());
        spec.bmc = Some(BmcEndpoint { host: "  ".into(), ..bmc() });
        assert!(matches!(spec.validate(), Err(Error::Spec(_))), "blank BMC host rejected");

        spec = BootSpec::redfish_iso("node", "/boot.iso", bmc());
        spec.bmc = Some(BmcEndpoint { username: String::new(), ..bmc() });
        assert!(matches!(spec.validate(), Err(Error::Spec(_))), "blank BMC username rejected");

        spec = BootSpec::redfish_iso("node", "/boot.iso", bmc());
        spec.bmc = Some(BmcEndpoint { system_id: String::new(), ..bmc() });
        assert!(matches!(spec.validate(), Err(Error::Spec(_))), "blank BMC system id rejected");

        // A fully-populated BMC still validates (no regression).
        BootSpec::redfish_iso("node", "/boot.iso", bmc()).validate().unwrap();
    }

    #[test]
    fn container_spec_with_a_zero_or_duplicate_port_is_rejected() {
        // Published ports are exposed and host-bound verbatim; `0` is never a real
        // published port and a doubled port self-collides. RED-when-broken — drop
        // either guard and its arm passes.
        let zero = BootSpec::container("cache", "redis:7").with_port(0);
        assert!(matches!(zero.validate(), Err(Error::Spec(_))), "0 published port rejected");

        let dup = BootSpec::container("cache", "redis:7").with_port(8080).with_port(8080);
        assert!(matches!(dup.validate(), Err(Error::Spec(_))), "duplicate published port rejected");

        // A mix with a single 0 among valid ports is still rejected.
        let mixed = BootSpec::container("cache", "redis:7").with_port(8080).with_port(0);
        assert!(matches!(mixed.validate(), Err(Error::Spec(_))), "0 among valid ports rejected");

        // A valid, distinct port set passes; no ports at all passes (no regression).
        BootSpec::container("cache", "redis:7").with_port(8080).with_port(8443).validate().unwrap();
        BootSpec::container("cache", "redis:7").validate().unwrap();
    }

    #[test]
    fn container_port_map_validates_distinct_host_container_and_host_collisions() {
        // The distinct `host:container` form (the per-zone offset fix): a `Test`
        // zone publishes FalkorDB on host 6380 → container 6379. RED-when-broken —
        // drop a guard and its arm passes.

        // A valid distinct map passes — this is korp's Test/Prod zone shape.
        BootSpec::container("falkor", "docker.io/falkordb/falkordb:v4.20.0")
            .with_port_map(6380, 6379)
            .validate()
            .unwrap();

        // Host 0 or container 0 is not a real published port.
        let host0 = BootSpec::container("f", "img:1").with_port_map(0, 6379);
        assert!(matches!(host0.validate(), Err(Error::Spec(_))), "host 0 rejected");
        let cont0 = BootSpec::container("f", "img:1").with_port_map(6380, 0);
        assert!(matches!(cont0.validate(), Err(Error::Spec(_))), "container 0 rejected");

        // The same HOST port bound twice self-collides — across two maps...
        let dup_map = BootSpec::container("f", "img:1").with_port_map(6380, 6379).with_port_map(6380, 15002);
        assert!(matches!(dup_map.validate(), Err(Error::Spec(_))), "duplicate host port across maps rejected");
        // ...and across the `ports` + `port_maps` fields (a host 8080 in both).
        let dup_mix = BootSpec::container("f", "img:1").with_port(8080).with_port_map(8080, 80);
        assert!(matches!(dup_mix.validate(), Err(Error::Spec(_))), "host port shared by ports+port_maps rejected");

        // Two DIFFERENT host ports mapping to the SAME container port is fine —
        // that is exactly the multi-zone case (never both live on one host at
        // once, but a spec listing both is legitimate) — no false collision.
        BootSpec::container("f", "img:1").with_port_map(6380, 6379).with_port_map(6381, 6379).validate().unwrap();
        // `ports` (host==container) composes with a distinct map on another port.
        BootSpec::container("f", "img:1").with_port(6379).with_port_map(15003, 15002).validate().unwrap();
    }

    #[test]
    fn container_only_cmd_or_ports_on_a_non_container_backend_is_rejected() {
        // `cmd` and `ports` are container-only knobs; KVM/Redfish silently ignore
        // them, so setting either there is a misconfiguration that would vanish
        // without a trace — reject it (parity with a bmc on a non-Redfish backend).
        // RED-when-broken — drop either guard and its arm passes.

        // A cmd on a KVM spec is rejected.
        let kvm_cmd = BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
            .with_cmd(["/bin/init"]);
        assert!(matches!(kvm_cmd.validate(), Err(Error::Spec(_))), "cmd on KVM rejected");

        // A published port on a KVM spec is rejected.
        let kvm_port =
            BootSpec::kvm_disk("disky", "/bzImage", "/disk.qcow2").with_port(8080);
        assert!(matches!(kvm_port.validate(), Err(Error::Spec(_))), "port on KVM rejected");

        // A cmd on a Redfish spec is rejected...
        let redfish_cmd =
            BootSpec::redfish_iso("node", "/boot.iso", bmc()).with_cmd(["/bin/init"]);
        assert!(matches!(redfish_cmd.validate(), Err(Error::Spec(_))), "cmd on Redfish rejected");

        // ...and a published port on a Redfish spec is rejected.
        let redfish_port = BootSpec::redfish_iso("node", "/boot.iso", bmc()).with_port(443);
        assert!(matches!(redfish_port.validate(), Err(Error::Spec(_))), "port on Redfish rejected");

        // The container backend still carries both (no regression).
        BootSpec::container("web", "nginx:latest")
            .with_cmd(["nginx", "-g", "daemon off;"])
            .with_port(8080)
            .with_port(8443)
            .validate()
            .unwrap();
    }

    #[test]
    fn container_resource_knobs_are_container_only_and_must_be_positive() {
        // `cpus` / `mem_limit_mb` are container-only (parity with cmd/ports/net):
        // a CPU/memory cap on a KVM or Redfish spec is rejected up front.
        // RED-when-broken — drop either container-only guard and its arm passes.
        let kvm_cpus = BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
            .with_cpus(8.0);
        assert!(matches!(kvm_cpus.validate(), Err(Error::Spec(_))), "cpus on KVM rejected");
        let kvm_mem = BootSpec::kvm_disk("disky", "/bzImage", "/disk.qcow2").with_mem_limit_mb(4096);
        assert!(matches!(kvm_mem.validate(), Err(Error::Spec(_))), "mem limit on KVM rejected");
        let redfish_cpus = BootSpec::redfish_iso("node", "/boot.iso", bmc()).with_cpus(4.0);
        assert!(matches!(redfish_cpus.validate(), Err(Error::Spec(_))), "cpus on Redfish rejected");

        // A non-positive cap is rejected on the container backend (None = all cores /
        // unconstrained is the way to express "no limit", never 0).
        let zero_cpus = BootSpec::container("cache", "redis:7").with_cpus(0.0);
        assert!(matches!(zero_cpus.validate(), Err(Error::Spec(_))), "0 cpus rejected");
        let mut neg = BootSpec::container("cache", "redis:7");
        neg.cpus = Some(-1.0);
        assert!(matches!(neg.validate(), Err(Error::Spec(_))), "negative cpus rejected");
        let zero_mem = BootSpec::container("cache", "redis:7").with_mem_limit_mb(0);
        assert!(matches!(zero_mem.validate(), Err(Error::Spec(_))), "0 MiB memory rejected");

        // A hot-infra container (all cores, sized memory) validates; and the default
        // (no caps = all host cores, unconstrained memory) validates too (no regression).
        BootSpec::container("falkordb", "docker.io/falkordb/falkordb:v4.20.0")
            .with_port(6379)
            .with_cpus(12.0)
            .with_mem_limit_mb(16384)
            .validate()
            .unwrap();
        BootSpec::container("cache", "redis:7").validate().unwrap();
        assert_eq!(BootSpec::container("cache", "redis:7").cpus, None, "default = all host cores");
    }

    #[test]
    fn image_backend_mismatch_is_rejected() {
        let mut spec = BootSpec::container("bad", "redis:7");
        spec.image = ImageSource::Iso("/boot.iso".into());
        assert!(matches!(spec.validate(), Err(Error::Spec(_))));
    }

    #[test]
    fn redfish_without_bmc_is_rejected() {
        let mut spec = BootSpec::redfish_iso("node", "/boot.iso", bmc());
        spec.bmc = None;
        assert!(matches!(spec.validate(), Err(Error::Spec(_))));
    }

    #[test]
    fn non_redfish_with_bmc_is_rejected() {
        let mut spec = BootSpec::container("cache", "redis:7");
        spec.bmc = Some(bmc());
        assert!(matches!(spec.validate(), Err(Error::Spec(_))));
    }

    #[test]
    fn started_machine_records_the_spec() {
        let spec = BootSpec::container("cache", "redis:7").with_env("PORT", "6379");
        let m = Machine::started("ctr-abc123", &spec);
        assert_eq!(m.spec_name, "cache");
        assert_eq!(m.backend, Backend::Container);
        assert_eq!(m.power, PowerState::On);
        assert_eq!(spec.env.get("PORT").map(String::as_str), Some("6379"));
    }

    /// A fake [`Boot`] that records the specs it was handed and mints a stable id,
    /// so the unifying [`boot`]/[`boot_fleet`] entry points are testable with no
    /// live backend.
    #[derive(Default)]
    struct RecordingBoot {
        seen: std::cell::RefCell<Vec<String>>,
    }

    impl Boot for RecordingBoot {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            self.seen.borrow_mut().push(spec.name.clone());
            Ok(Machine::started(format!("id-{}", spec.name), spec))
        }
    }

    #[test]
    fn boot_validates_then_delegates_to_the_backend() {
        let backend = RecordingBoot::default();
        let spec = BootSpec::container("cache", "redis:7");
        let m = boot(&spec, &backend).unwrap();
        assert_eq!(m.id, "id-cache");
        assert_eq!(m.backend, Backend::Container);
        assert_eq!(backend.seen.borrow().as_slice(), &["cache".to_string()]);
    }

    #[test]
    fn boot_rejects_an_invalid_spec_before_touching_the_backend() {
        let backend = RecordingBoot::default();
        let mut spec = BootSpec::container("bad", "redis:7");
        spec.image = ImageSource::Iso("/boot.iso".into()); // ISO can't boot on Container
        assert!(matches!(boot(&spec, &backend), Err(Error::Spec(_))));
        assert!(backend.seen.borrow().is_empty(), "backend never touched on an invalid spec");
    }

    #[test]
    fn boot_fleet_drips_and_boots_every_member_through_one_backend() {
        let backend = RecordingBoot::default();
        let one = BootSpec::redfish_iso("node", "/images/installer.iso", bmc());
        let results = boot_fleet(&one, 3, &backend);
        assert_eq!(results.len(), 3);
        let ids: Vec<_> = results.into_iter().map(|r| r.unwrap().id).collect();
        assert_eq!(ids, vec!["id-node-1", "id-node-2", "id-node-3"]);
        assert_eq!(backend.seen.borrow().as_slice(), &["node-1", "node-2", "node-3"]);
    }

    /// A [`Boot`] that mints ids until it has booted `ok_before` members, then
    /// errors — so the documented "a partial fleet is observable" contract of
    /// [`boot_fleet`] can be exercised.
    struct FlakyBoot {
        ok_before: usize,
        booted: std::cell::RefCell<usize>,
    }

    impl Boot for FlakyBoot {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            let mut n = self.booted.borrow_mut();
            if *n >= self.ok_before {
                return Err(Error::Backend(format!("backend went away booting {}", spec.name)));
            }
            *n += 1;
            Ok(Machine::started(format!("id-{}", spec.name), spec))
        }
    }

    #[test]
    fn boot_fleet_reports_a_partial_fleet_when_a_later_member_fails() {
        // The doc promises a partial fleet is observable: the first members boot,
        // a later one errors, and every per-member result is returned in order.
        let backend = FlakyBoot { ok_before: 2, booted: std::cell::RefCell::new(0) };
        let one = BootSpec::redfish_iso("node", "/images/installer.iso", bmc());
        let results = boot_fleet(&one, 4, &backend);
        assert_eq!(results.len(), 4);
        assert_eq!(results[0].as_ref().unwrap().id, "id-node-1");
        assert_eq!(results[1].as_ref().unwrap().id, "id-node-2");
        assert!(matches!(results[2], Err(Error::Backend(_))), "3rd member fails");
        assert!(matches!(results[3], Err(Error::Backend(_))), "4th member fails too");
        let ok = results.iter().filter(|r| r.is_ok()).count();
        assert_eq!(ok, 2, "exactly the first two members booted");
    }

    #[test]
    fn plan_fleet_of_zero_is_empty_and_of_one_keeps_a_suffix() {
        assert!(plan_fleet(&BootSpec::container("c", "redis:7"), 0).is_empty());
        // Even n=1 gets the 1-based suffix (a fleet member is always "{name}-{i}").
        let one = plan_fleet(&BootSpec::container("c", "redis:7"), 1);
        assert_eq!(one.len(), 1);
        assert_eq!(one[0].name, "c-1");
    }

    #[test]
    fn cloud_init_user_data_constructor_defaults_meta_data_to_none() {
        let ci = CloudInit::user_data("#cloud-config\n");
        assert_eq!(ci.user_data, "#cloud-config\n");
        assert_eq!(ci.meta_data, None);
        assert_eq!(ci.network_config, None);
    }

    #[test]
    fn container_spec_defaults_are_lean() {
        // A container boot carries no VM sizing and no overrides until asked — the
        // "lean by design" contract the KVM path does not share.
        let spec = BootSpec::container("cache", "docker.io/library/redis:7");
        assert_eq!(spec.mem_mb, 0);
        assert_eq!(spec.cores, 0);
        assert!(spec.cmd.is_empty());
        assert!(spec.ports.is_empty());
        assert!(spec.env.is_empty());
        assert!(spec.bmc.is_none());
        assert!(spec.cloud_init.is_none());
    }

    #[test]
    fn cloud_init_on_a_container_spec_still_validates_and_is_backend_ignored() {
        // cloud_init is a KVM-only provisioning payload; attaching it to a container
        // spec is not an error (the container backend simply ignores it).
        let spec = BootSpec::container("cache", "redis:7")
            .with_cloud_init(CloudInit::user_data("#cloud-config\n"));
        spec.validate().unwrap();
        assert!(spec.cloud_init.is_some());
    }

    #[test]
    fn plan_fleet_drips_n_identical_but_distinctly_named_members() {
        let one = BootSpec::redfish_iso("node", "/images/installer.iso", bmc());
        let fleet = plan_fleet(&one, 8);
        assert_eq!(fleet.len(), 8);
        assert_eq!(fleet[0].name, "node-1");
        assert_eq!(fleet[7].name, "node-8");
        // Identical payload + backend across the whole ring.
        assert!(fleet.iter().all(|m| m.image == one.image && m.backend == one.backend));
        // Names are unique.
        let mut names: Vec<_> = fleet.iter().map(|m| m.name.clone()).collect();
        names.sort();
        names.dedup();
        assert_eq!(names.len(), 8);
    }

    /// A backend that is both [`Boot`] and [`Lifecycle`], reporting a scripted
    /// sequence of power states from `status` (then a `fallback` once the sequence
    /// is exhausted) — so the [`await_power_state`] / [`boot_and_await`] readback
    /// seam is testable with no live instance.
    struct ScriptedNode {
        states: std::cell::RefCell<std::collections::VecDeque<PowerState>>,
        fallback: PowerState,
        status_calls: std::cell::Cell<usize>,
    }

    impl ScriptedNode {
        fn new(seq: impl IntoIterator<Item = PowerState>, fallback: PowerState) -> Self {
            Self {
                states: std::cell::RefCell::new(seq.into_iter().collect()),
                fallback,
                status_calls: std::cell::Cell::new(0),
            }
        }
    }

    impl Boot for ScriptedNode {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            Ok(Machine::started(format!("id-{}", spec.name), spec))
        }
    }

    impl Lifecycle for ScriptedNode {
        fn power_on(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn power_off(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn status(&self, _m: &Machine) -> Result<PowerState> {
            self.status_calls.set(self.status_calls.get() + 1);
            Ok(self.states.borrow_mut().pop_front().unwrap_or(self.fallback))
        }
    }

    fn tiny_bounded() -> WaitOptions {
        WaitOptions::bounded(
            std::time::Duration::from_millis(200),
            std::time::Duration::from_millis(1),
        )
    }

    #[test]
    fn await_power_state_returns_ok_once_the_state_is_reached() {
        // status reports Unknown twice, then On — await must poll past the
        // Unknowns and return Ok the moment it observes On.
        let node =
            ScriptedNode::new([PowerState::Unknown, PowerState::Unknown, PowerState::On], PowerState::On);
        let m = Machine::started("node-1", &BootSpec::container("c", "redis:7"));
        await_power_state(&node, &m, PowerState::On, &tiny_bounded()).unwrap();
        assert!(node.status_calls.get() >= 3, "polled past the two Unknowns to On");
    }

    #[test]
    fn await_power_state_times_out_when_the_state_is_never_reached() {
        // The node is stuck Off forever; a bounded wait must time out with a
        // Backend error rather than block. RED-when-broken — drop the `== want`
        // gate (return Ok on the first poll) and this Err expectation fails.
        let node = ScriptedNode::new(std::iter::empty(), PowerState::Off);
        let m = Machine::started("node-2", &BootSpec::container("c", "redis:7"));
        let r = await_power_state(
            &node,
            &m,
            PowerState::On,
            &WaitOptions::bounded(
                std::time::Duration::from_millis(20),
                std::time::Duration::from_millis(1),
            ),
        );
        assert!(matches!(r, Err(Error::Backend(_))), "never-up node times out");
    }

    #[test]
    fn await_power_state_rejects_awaiting_unknown() {
        // Awaiting Unknown is nonsensical (Unknown = "not observed") and is
        // rejected up front as a Spec error, before any poll.
        let node = ScriptedNode::new([PowerState::On], PowerState::On);
        let m = Machine::started("node-3", &BootSpec::container("c", "redis:7"));
        let r = await_power_state(&node, &m, PowerState::Unknown, &tiny_bounded());
        assert!(matches!(r, Err(Error::Spec(_))), "await Unknown rejected");
        assert_eq!(node.status_calls.get(), 0, "rejected before polling");
    }

    #[test]
    fn await_power_state_propagates_a_backend_status_error() {
        // A live backend's status() error is the readback failing — it propagates.
        struct ErrLifecycle;
        impl Lifecycle for ErrLifecycle {
            fn power_on(&self, _m: &Machine) -> Result<()> {
                Ok(())
            }
            fn power_off(&self, _m: &Machine) -> Result<()> {
                Ok(())
            }
            fn status(&self, _m: &Machine) -> Result<PowerState> {
                Err(Error::Backend("BMC unreachable".into()))
            }
        }
        let m = Machine::started("node-4", &BootSpec::container("c", "redis:7"));
        let r = await_power_state(&ErrLifecycle, &m, PowerState::On, &tiny_bounded());
        assert!(matches!(r, Err(Error::Backend(_))), "status error propagates");
    }

    #[test]
    fn boot_and_await_boots_then_confirms_power_on() {
        // The one-call provision seam: boot, then confirm On. The node reports
        // Unknown once, then On — boot_and_await returns the live Machine only
        // after the On readback.
        let node = ScriptedNode::new([PowerState::Unknown, PowerState::On], PowerState::On);
        let spec = BootSpec::container("cache", "redis:7");
        let m = boot_and_await(&node, &spec, &tiny_bounded()).unwrap();
        assert_eq!(m.id, "id-cache");
        assert_eq!(m.power, PowerState::On);
        assert!(node.status_calls.get() >= 2, "awaited past Unknown to On");
    }

    #[test]
    fn boot_and_await_times_out_when_the_instance_never_comes_up() {
        // A node that never powers on makes boot_and_await a bounded timeout, not a
        // hang — the readback is what distinguishes a booted-but-dead instance.
        let node = ScriptedNode::new(std::iter::empty(), PowerState::Off);
        let spec = BootSpec::container("cache", "redis:7");
        let r = boot_and_await(
            &node,
            &spec,
            &WaitOptions::bounded(
                std::time::Duration::from_millis(20),
                std::time::Duration::from_millis(1),
            ),
        );
        assert!(matches!(r, Err(Error::Backend(_))), "never-up boot times out");
    }

    #[test]
    fn boot_and_await_rejects_an_invalid_spec_before_booting() {
        // boot_and_await goes through boot(), so an invalid spec is rejected at
        // validate() before the backend (or any wait) is ever touched.
        let node = ScriptedNode::new([PowerState::On], PowerState::On);
        let mut spec = BootSpec::container("bad", "redis:7");
        spec.image = ImageSource::Iso("/boot.iso".into()); // ISO can't boot a container
        let r = boot_and_await(&node, &spec, &tiny_bounded());
        assert!(matches!(r, Err(Error::Spec(_))), "invalid spec rejected pre-boot");
        assert_eq!(node.status_calls.get(), 0, "never awaited an unbooted instance");
    }

    /// A fleet backend where members whose name is in `dead` never power on (their
    /// `status` stays [`PowerState::Off`]) while every other member boots and reports
    /// [`PowerState::On`] — so a *partial-fleet* boot-readback rollup (some up, one
    /// timed out) is testable with no live instances. Keys on `Machine::spec_name`,
    /// which carries the fleet member name.
    struct FleetNode {
        dead: std::collections::HashSet<String>,
    }

    impl FleetNode {
        fn with_dead<'a>(dead: impl IntoIterator<Item = &'a str>) -> Self {
            Self { dead: dead.into_iter().map(String::from).collect() }
        }
    }

    impl Boot for FleetNode {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            Ok(Machine::started(format!("id-{}", spec.name), spec))
        }
    }

    impl Lifecycle for FleetNode {
        fn power_on(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn power_off(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn status(&self, m: &Machine) -> Result<PowerState> {
            // A dead member is stuck Off forever (never confirms up); everyone else
            // is On the moment they are polled.
            if self.dead.contains(&m.spec_name) {
                Ok(PowerState::Off)
            } else {
                Ok(PowerState::On)
            }
        }
    }

    #[test]
    fn boot_fleet_and_await_rolls_up_one_dead_member_as_timeout_others_up() {
        // The headline fleet-readback contract: boot a fleet of 3 where the middle
        // member never powers on. The rollup must show node-2 = Timeout while node-1
        // and node-3 = Up, with the aggregate ready/failed tallies correct — a dead
        // node is a per-member verdict, never a hang or an early return.
        //
        // RED-when-broken: this leans on `boot_and_await` confirming On per member
        // AND the Err(Backend)→Timeout classification. Neuter await_power_state's
        // `observed == want` gate (return Ok on the first poll) and node-2 rolls up
        // Up → ready becomes 3, this fails; flip the Timeout classification arm to
        // Error and the `matches!(.., Timeout)` assertion fails.
        let backend = FleetNode::with_dead(["node-2"]);
        let one = BootSpec::redfish_iso("node", "/images/installer.iso", bmc());
        let rollup = boot_fleet_and_await(&one, 3, &backend, &tiny_bounded());

        assert_eq!(rollup.members.len(), 3, "one verdict per member, in fleet order");
        assert_eq!(rollup.members[0].name, "node-1");
        assert_eq!(rollup.members[1].name, "node-2");
        assert_eq!(rollup.members[2].name, "node-3");

        // The two healthy members are Up with their live handle...
        match &rollup.members[0].outcome {
            MemberOutcome::Up(m) => assert_eq!(m.id, "id-node-1"),
            other => panic!("node-1 should be Up, was {other:?}"),
        }
        assert!(matches!(rollup.members[2].outcome, MemberOutcome::Up(_)), "node-3 up");
        // ...and the dead member is a Timeout, not an Up and not a hang.
        assert!(
            matches!(rollup.members[1].outcome, MemberOutcome::Timeout(_)),
            "node-2 never powered on → Timeout, was {:?}",
            rollup.members[1].outcome
        );

        // Aggregate tallies (mutation-verified).
        assert_eq!(rollup.ready(), 2, "exactly the two healthy members are ready");
        assert_eq!(rollup.failed(), 1, "exactly the one dead member failed");
        assert!(!rollup.all_ready(), "a partial fleet is not all-ready");
    }

    #[test]
    fn boot_fleet_and_await_reports_a_fully_ready_fleet() {
        // The all-healthy path: every member boots and confirms On, so the rollup is
        // all_ready with ready == n and no failures.
        let backend = FleetNode::with_dead(std::iter::empty());
        let one = BootSpec::container("cache", "redis:7");
        let rollup = boot_fleet_and_await(&one, 4, &backend, &tiny_bounded());
        assert_eq!(rollup.members.len(), 4);
        assert!(rollup.members.iter().all(|m| matches!(m.outcome, MemberOutcome::Up(_))));
        assert_eq!(rollup.ready(), 4);
        assert_eq!(rollup.failed(), 0);
        assert!(rollup.all_ready(), "a fully-up fleet is all-ready");
    }

    #[test]
    fn boot_fleet_and_await_rolls_up_an_invalid_spec_as_error_not_timeout() {
        // An invalid spec is rejected at validate() inside boot_and_await, before any
        // backend/await — so every member is an Error (hard misconfig), never a
        // Timeout, and the backend is never touched. Distinguishes the two failure
        // buckets: Spec → Error, Backend → Timeout.
        let backend = FleetNode::with_dead(std::iter::empty());
        let mut bad = BootSpec::container("bad", "redis:7");
        bad.image = ImageSource::Iso("/boot.iso".into()); // ISO can't boot a container
        let rollup = boot_fleet_and_await(&bad, 2, &backend, &tiny_bounded());
        assert_eq!(rollup.members.len(), 2);
        assert!(
            rollup.members.iter().all(|m| matches!(m.outcome, MemberOutcome::Error(_))),
            "an invalid spec rolls up as Error on every member, not Timeout"
        );
        assert_eq!(rollup.ready(), 0);
        assert_eq!(rollup.failed(), 2);
    }

    #[test]
    fn boot_fleet_and_await_of_zero_is_an_empty_not_ready_rollup() {
        // An empty fleet: no members, and all_ready is false (vacuously "nothing up"
        // is not a ready fleet — parity with plan_fleet(0) being empty).
        let backend = FleetNode::with_dead(std::iter::empty());
        let rollup =
            boot_fleet_and_await(&BootSpec::container("c", "redis:7"), 0, &backend, &tiny_bounded());
        assert!(rollup.members.is_empty());
        assert_eq!(rollup.ready(), 0);
        assert_eq!(rollup.failed(), 0);
        assert!(!rollup.all_ready(), "an empty fleet is not all-ready");
    }

    // -- Parallel fleet-boot: identical rollup + proven overlap ------------------

    /// A `Sync` fleet backend that produces a *mixed* rollup with no live instances:
    /// members named in `unbootable` fail at [`Boot::boot`] (`Err(Unsupported)` →
    /// `Error`), members named in `dead` boot but never power on (`status` stays `Off`
    /// → `Timeout`), everyone else comes `Up`. `Sync` (only owned `HashSet`s), so it
    /// drives BOTH the serial and the scoped-thread parallel path — letting the two be
    /// asserted byte-for-byte equal.
    struct MixedNode {
        dead: std::collections::HashSet<String>,
        unbootable: std::collections::HashSet<String>,
    }

    impl Boot for MixedNode {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            if self.unbootable.contains(&spec.name) {
                return Err(Error::Unsupported(format!("no backend slot for {}", spec.name)));
            }
            Ok(Machine::started(format!("id-{}", spec.name), spec))
        }
    }

    impl Lifecycle for MixedNode {
        fn power_on(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn power_off(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn status(&self, m: &Machine) -> Result<PowerState> {
            if self.dead.contains(&m.spec_name) {
                Ok(PowerState::Off)
            } else {
                Ok(PowerState::On)
            }
        }
    }

    #[test]
    fn parallel_fleet_boot_is_identical_to_serial_for_a_mixed_fleet() {
        // The equivalence guard: a mixed fleet (some Up, one Timeout, one Error) MUST
        // roll up to the SAME FleetReadback — same per-member outcomes, same fleet
        // order, same aggregate tallies — whether booted serially or in parallel. The
        // parallel path only changes WHEN members are awaited, never WHAT the rollup
        // says.
        //
        // RED-when-broken: the two share the one `classify_member` step, so any
        // divergence (a reordered join, a different bucket, a dropped member) breaks
        // this `assert_eq!`. Fleet of 5: node-3 dead (→ Timeout), node-5 unbootable
        // (→ Error), node-1/2/4 Up.
        let backend = MixedNode {
            dead: std::collections::HashSet::from(["node-3".to_string()]),
            unbootable: std::collections::HashSet::from(["node-5".to_string()]),
        };
        let spec = BootSpec::container("node", "redis:7");

        let serial = boot_fleet_and_await(&spec, 5, &backend, &tiny_bounded());
        let parallel = boot_fleet_and_await_parallel(&spec, 5, &backend, &tiny_bounded());

        // Byte-for-byte identical rollups (names, order, outcomes, and the carried
        // Machine handles / failure strings all compared by derived PartialEq).
        assert_eq!(parallel, serial, "parallel rollup must equal the serial rollup");

        // And it is genuinely the mixed shape we intended (not two identical *empty*
        // rollups trivially matching).
        assert_eq!(parallel.members.len(), 5);
        assert!(matches!(parallel.members[0].outcome, MemberOutcome::Up(_)), "node-1 up");
        assert!(matches!(parallel.members[2].outcome, MemberOutcome::Timeout(_)), "node-3 timeout");
        assert!(matches!(parallel.members[4].outcome, MemberOutcome::Error(_)), "node-5 error");
        assert_eq!(parallel.ready(), 3);
        assert_eq!(parallel.failed(), 2);
    }

    /// A `Sync` backend that proves the members are awaited **concurrently**: every
    /// member, on its first `status` poll, bumps a shared "in flight" counter, records
    /// the running peak, then waits (bounded) until all `n` members have entered before
    /// returning `On`. In the parallel path all `n` threads enter together so the peak
    /// reaches `n`; a serial path would only ever have one member in flight (peak 1),
    /// timing out the entry wait instead of hanging.
    struct BarrierNode {
        n: usize,
        in_flight: std::sync::atomic::AtomicUsize,
        peak: std::sync::atomic::AtomicUsize,
    }

    impl BarrierNode {
        fn new(n: usize) -> Self {
            Self {
                n,
                in_flight: std::sync::atomic::AtomicUsize::new(0),
                peak: std::sync::atomic::AtomicUsize::new(0),
            }
        }
    }

    impl Boot for BarrierNode {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            Ok(Machine::started(format!("id-{}", spec.name), spec))
        }
    }

    impl Lifecycle for BarrierNode {
        fn power_on(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn power_off(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn status(&self, _m: &Machine) -> Result<PowerState> {
            use std::sync::atomic::Ordering::SeqCst;
            let now = self.in_flight.fetch_add(1, SeqCst) + 1;
            self.peak.fetch_max(now, SeqCst);
            // Bounded wait for all members to have entered — proves they overlap
            // without ever deadlocking a (hypothetical) serial caller.
            let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
            while self.in_flight.load(SeqCst) < self.n && std::time::Instant::now() < deadline {
                std::thread::yield_now();
            }
            self.in_flight.fetch_sub(1, SeqCst);
            Ok(PowerState::On)
        }
    }

    #[test]
    fn parallel_fleet_boot_actually_overlaps_the_members() {
        // The concurrency guard: prove the parallel path really runs members at the
        // same time (not a serial loop wearing a parallel name). All `n` members must
        // be in their `status` poll simultaneously → observed peak concurrency == n.
        //
        // RED-when-broken: replace `boot_fleet_and_await_parallel` with the serial
        // `boot_fleet_and_await` here and the peak collapses to 1 (each member enters,
        // waits out the 500ms barrier alone, and leaves before the next starts), so
        // `peak == n` fails.
        use std::sync::atomic::Ordering::SeqCst;
        let n = 6;
        let backend = BarrierNode::new(n);
        let spec = BootSpec::container("node", "redis:7");
        let rollup = boot_fleet_and_await_parallel(&spec, n, &backend, &tiny_bounded());

        assert!(rollup.all_ready(), "every member comes up");
        assert_eq!(
            backend.peak.load(SeqCst),
            n,
            "all {n} members were awaited concurrently (peak in-flight == n)"
        );
    }

    #[test]
    fn parallel_fleet_boot_of_zero_is_an_empty_rollup() {
        // Parity with the serial n=0 case: no members, not all-ready, no threads.
        let backend = MixedNode {
            dead: std::collections::HashSet::new(),
            unbootable: std::collections::HashSet::new(),
        };
        let rollup = boot_fleet_and_await_parallel(
            &BootSpec::container("c", "redis:7"),
            0,
            &backend,
            &tiny_bounded(),
        );
        assert!(rollup.members.is_empty());
        assert!(!rollup.all_ready());
    }
}