draupnir 0.1.9

//! # 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.
//! - [`exe`] — a **plain executable** run as a child process on this host: the
//!   non-VM, non-container shape (the `exe` build-thing kind).
//!
//! ## Booting OFF a medium (2026-08-15)
//!
//! A [`BootSpec`] can name a removable [`medium`](BootSpec::medium) and a
//! [`boot_order`](BootOrder), which is what makes an appliance ISO runnable at all:
//!
//! - [`BootSpec::iso_boot`] — live-boot an ISO. **One spec**, routable to the KVM
//!   backend as it stands or to real metal with [`on_metal`](BootSpec::on_metal),
//!   so `iso-kvm` and `iso-metal` are two runs of one claim rather than two claims.
//! - [`BootSpec::kvm_install_from_medium`] — boot off the ISO with a blank target
//!   disk attached (the INSTALL leg).
//! - [`BootSpec::kvm_boot_installed_disk`] — boot that disk with the medium
//!   **absent** (the SECOND VM lifetime). [`validate`](BootSpec::validate) refuses a
//!   spec that claims to boot the installed system while still holding the medium.
//!
//! A spec that names no medium and no boot order behaves exactly as it did before
//! these fields existed — both default to "emit nothing".
//!
//! ## Waiting: a thing that keeps running is the NORMAL case
//!
//! [`exe::ExeBoot::await_marker`] and [`kvm::KvmBoot::await_serial_marker`] wait for
//! a **named marker or a deadline**, never for the instance to terminate, and both
//! return the same [`Seen`] describing what was actually observed. An appliance that
//! boots correctly serves forever; waiting for it to exit is waiting for it to fail.
//!
//! ## 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;
use std::time::Duration;

pub mod container;
pub mod exe;
pub mod kvm;
pub mod redfish;
/// **Draupnir's own BMC** — the Redfish *service* end, fronting KVM
/// ([`redfish_server`](crate::redfish_server)). Behind the `redfish-server` feature,
/// because a client-only build must never grow a listening socket.
#[cfg(feature = "redfish-server")]
pub mod redfish_server;
pub mod seed;
#[cfg(feature = "backend-youki")]
pub mod youki;

/// 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);
    }
}

/// **Drain the buffered functional-status rows** as `(component, check, ok)` triples
/// (feature `testmatrix`) — returns AND clears the process-global buffer nornir's
/// matrix reads back. Exposed so a test can assert that a lifecycle surface emitted
/// the expected GREEN/RED verdict (a failed boot/probe must land as a RED row). It
/// reuses `nornir-testmatrix`'s own buffer verbatim — no twin. When the feature is
/// off there is no buffer, so this is not compiled.
#[cfg(feature = "testmatrix")]
pub fn drain_status_rows() -> Vec<(String, String, bool)> {
    nornir_testmatrix::drain_functional_rows()
        .into_iter()
        .map(|r| (r.suite, r.test_name, r.status == "pass"))
        .collect()
}

/// 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,
    /// A **plain executable** run as a child process on this host — the non-VM,
    /// non-container shape ([`exe::ExeBoot`]). This is the `exe` build-thing kind:
    /// a binary stage 1 produced, started here, its stdout/stderr streamed, and
    /// awaited on a **named marker or a deadline** — never on termination, because
    /// a server that serves forever never terminates.
    Exe,
}

/// 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,
    },
    /// A self-booting disk image with an **in-image bootloader** (the KVM
    /// firmware-boot path → tunnr). No `-kernel`: OVMF boots the disk directly,
    /// exactly as real hardware would. tunnr firmware-boots whenever the kernel
    /// path is empty (`is_direct_kernel_boot()` false), so this variant carries
    /// only the disk.
    BootloaderDisk {
        /// Bootable disk image path (qcow2 or raw), attached as a virtio drive
        /// and booted by firmware (no `-kernel`).
        disk: String,
    },
    /// An OCI image reference, e.g. `docker.io/library/redis:7` (container path).
    OciImage(String),
    /// A **bootable ISO**, booted *off the medium*: served as Redfish **virtual
    /// media** to a BMC (bare metal), or attached to a KVM guest as a `-cdrom`
    /// the firmware boots from ([`BootOrder::Medium`]).
    ///
    /// It suits **both** [`Backend::Redfish`] and [`Backend::Kvm`] — deliberately,
    /// and this is the whole of honesty rule 2: *one* spec, routed to the faked
    /// KVM backend or to real metal, so "it works in the demo" and "it works on the
    /// customer's server" stop being different claims. tunnr already renders an ISO
    /// payload as `-cdrom <iso> -boot d`, so the KVM leg is its existing firmware
    /// path, not a new one. Until 2026-08-15 `suits` rejected `(Iso, Kvm)`, which is
    /// exactly the wall that kept the ISO chain from ever running.
    Iso(String),
    /// A **plain executable** on this host — the `exe` build-thing kind, run as a
    /// child process by [`exe::ExeBoot`] (suits [`Backend::Exe`] only).
    ///
    /// `args` are the process's own argv (argv[1..]); they live here rather than on
    /// [`BootSpec::cmd`] because `cmd` is the container entrypoint override and is
    /// validated container-only — an exe's arguments are part of *what it is*, not a
    /// container knob.
    Executable {
        /// Path to the binary to run (stage 1's artifact).
        path: String,
        /// Arguments passed to it (argv[1..]).
        args: Vec<String>,
    },
}

impl ImageSource {
    /// Whether this payload is a legal source for `backend` — the KVM backend
    /// boots kernel+rootfs, a disk, **or an ISO off the medium**; the container
    /// backend an OCI image; Redfish an ISO; and [`Backend::Exe`] an
    /// [`Executable`](ImageSource::Executable). Used by [`BootSpec::validate`].
    ///
    /// `(Iso, Kvm)` is legal as of 2026-08-15 — the one deliberate widening in this
    /// pass. It was rejected before, which is precisely why the ISO chain could not
    /// run under KVM at all; tunnr has rendered an ISO payload as `-cdrom <iso>
    /// -boot d` since it was written (`tunnr/src/boot_primitive.rs:758-765`), so
    /// nothing new is being invented here — only un-forbidden.
    pub fn suits(&self, backend: Backend) -> bool {
        matches!(
            (self, backend),
            (ImageSource::KernelRootfs { .. }, Backend::Kvm)
                | (ImageSource::Disk { .. }, Backend::Kvm)
                | (ImageSource::BootloaderDisk { .. }, Backend::Kvm)
                | (ImageSource::OciImage(_), Backend::Container)
                | (ImageSource::Iso(_), Backend::Redfish)
                | (ImageSource::Iso(_), Backend::Kvm)
                | (ImageSource::Executable { .. }, Backend::Exe)
        )
    }
}

/// 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,
}

/// **Which device the firmware is told to boot from** — the boot-order half of
/// the boot-off-a-medium seam ([`BootSpec::boot_order`]).
///
/// This exists because *install* and *boot-the-installed-system* are two different
/// machines, not two phases of one. Honesty rule 1 of the chain driver: install in
/// one VM off the medium, **detach the medium, power-cycle, boot the disk in a
/// SECOND VM**. A caller must be able to say those two things as two distinct
/// specs, and a spec that claims to boot the installed disk must be *unable* to
/// carry the installer medium — otherwise "it booted" can silently be the installer
/// booting a second time.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum BootOrder {
    /// **Unchanged**: draupnir emits no boot-order of its own and the backend keeps
    /// whatever it already did (tunnr's payload-shaped default; a Redfish node's
    /// existing boot list). This is the [`Default`] — a spec that never named a boot
    /// order produces a byte-identical launch to one written before this field
    /// existed.
    #[default]
    Auto,
    /// Boot **off the removable medium** — QEMU `-boot d`, Redfish
    /// [`BootTarget::Cd`]. The install leg.
    Medium,
    /// Boot **off the local disk** — QEMU `-boot c`, Redfish [`BootTarget::Hdd`].
    /// The boot-the-installed-system leg; [`BootSpec::validate`] refuses this
    /// together with a medium, by construction.
    Disk,
}

impl BootOrder {
    /// The QEMU `-boot` value this renders to, or [`None`] for [`Auto`](BootOrder::Auto)
    /// (no `-boot` emitted → the launch is byte-identical to before this field).
    pub fn qemu_value(self) -> Option<&'static str> {
        match self {
            BootOrder::Auto => None,
            BootOrder::Medium => Some("d"),
            BootOrder::Disk => Some("c"),
        }
    }

    /// The Redfish one-time [`BootTarget`] this renders to, or [`None`] for
    /// [`Auto`](BootOrder::Auto) (leave the node's boot list alone).
    ///
    /// This is the *other half* of honesty rule 2: the SAME [`BootSpec`] expresses
    /// "boot the medium" once, and each backend translates it into its own
    /// vocabulary — `-boot d` here, `BootSourceOverrideTarget: "Cd"` there.
    pub fn redfish_target(self) -> Option<BootTarget> {
        match self {
            BootOrder::Auto => None,
            BootOrder::Medium => Some(BootTarget::Cd),
            BootOrder::Disk => Some(BootTarget::Hdd),
        }
    }

    /// **The inverse of [`redfish_target`](BootOrder::redfish_target)** — read a
    /// Redfish `BootSourceOverrideTarget` back as a boot order.
    ///
    /// It lives here, next to its forward direction, because draupnir owns *both*
    /// ends of the Redfish protocol: the client turns a `BootOrder` into a
    /// `BootSourceOverrideTarget`, and [`redfish_server`](crate::redfish_server) —
    /// draupnir's own BMC — has to turn the received target back into the boot order
    /// it will actually apply. Two functions in two modules would be two mappings,
    /// and the one that mattered would be the one nobody tested.
    ///
    /// [`None`] (no override) is [`Auto`](BootOrder::Auto). A target draupnir does
    /// not model as a boot order — `Pxe`, `BiosSetup`, and every other DMTF
    /// `BootSource` token — yields [`Option::None`]: a service that cannot honour a
    /// target must **refuse it by name**, never silently substitute one it can.
    pub fn from_redfish_target(target: Option<BootTarget>) -> Option<BootOrder> {
        match target {
            None => Some(BootOrder::Auto),
            Some(BootTarget::Cd) => Some(BootOrder::Medium),
            Some(BootTarget::Hdd) => Some(BootOrder::Disk),
            Some(BootTarget::Pxe) | Some(BootTarget::BiosSetup) => None,
        }
    }
}

/// 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>,
    /// Container **security options** — podman/Docker `--security-opt`, rendered
    /// verbatim onto `HostConfig.security_opt`. Only the [`container`] backend acts
    /// on it (KVM/Redfish ignore it); **empty (the default) keeps the create body
    /// byte-identical** to a spec that never named one, so nothing that exists today
    /// moves.
    ///
    /// It exists because a container's *default* syscall filter is not a property of
    /// the workload and cannot be discovered from inside it. podman's stock
    /// `/usr/share/containers/seccomp.json` allow-lists no `io_uring_*` syscall, so
    /// `io_uring_setup` inside a stock container returns **ENOSYS** — measured on
    /// oden 2026-08-11, kernel 7.0.0-29, with `/proc/sys/kernel/io_uring_disabled=0`
    /// and the identical probe returning a valid fd on the host and under
    /// `--security-opt seccomp=unconfined`. A caller benchmarking an io_uring code
    /// path therefore has no way to run it at all without this field, and "the arm
    /// refused to start" is indistinguishable from "the kernel lacks it".
    ///
    /// Each entry is one `--security-opt` value, e.g. `seccomp=/path/profile.json`,
    /// `seccomp=unconfined`, `label=disable`, `no-new-privileges`. draupnir does not
    /// interpret them: the daemon owns their meaning, and a value this crate parsed
    /// would be a second, staler copy of that grammar.
    pub security_opt: Vec<String>,
    /// **KVM extra QEMU args** — appended verbatim after tunnr's own argv (via
    /// `tunnr_vm::BootSpec::extra_qemu_args`). Only the [`kvm`] backend acts on it
    /// (containers/Redfish ignore it); it is the seam a KVM consumer uses to attach
    /// host-specific QEMU devices tunnr's spec does not model natively — e.g.
    /// Skidbladnir's `org.skidbladnir.report` virtio-serial report port, its
    /// `org.qemu.guest_agent.0` guest-agent port, a read-only data drive, an install
    /// `-cdrom`, `-cpu host`, and a `-snapshot` guard that keeps a shared,
    /// content-addressed base image from being mutated. Empty (the default) leaves
    /// the tunnr argv **byte-identical** to a spec that never named it.
    pub extra_qemu_args: Vec<String>,
    /// **KVM: must the guest's writes to the boot disk SURVIVE the boot?**
    ///
    /// `false` (the default) keeps the long-standing behaviour: an
    /// [`ImageSource::BootloaderDisk`] is opened `snapshot=on`, so guest writes land
    /// in a throwaway overlay and a shared, content-addressed base OS image can never
    /// be mutated by a boot. That is the right default — it is what stops a matrix
    /// run from poisoning the image every later run depends on.
    ///
    /// `true` says the caller owns this disk (typically a per-run copy or an overlay
    /// it made itself) and needs it to persist, because the *point* of the boot is
    /// what the guest wrote. Without this there is no way to express **two boots of
    /// one machine**: boot once to install, boot again to see what systemd does with
    /// the installed system at startup. Under the COW default the second boot silently
    /// gets a pristine image, re-runs first-boot provisioning, and produces
    /// plausible-looking facts about a machine that was never installed. Measured on
    /// oden 2026-08-03: after a full install the per-run disk copy was byte-identical
    /// (md5) to the base, and boot B replayed the whole of boot A.
    ///
    /// Only the [`kvm`] backend acts on it, and only for a boot whose disk would
    /// otherwise be COW-protected; every other [`ImageSource`] already persists.
    /// `false` leaves the tunnr spec **byte-identical** to a spec that never named it.
    pub persist_disk: bool,
    /// **The removable boot MEDIUM** — an installer/appliance ISO attached *in
    /// addition to* the [`image`](BootSpec::image) payload, so a machine can be
    /// booted OFF the medium while a separate, blank target disk is present to be
    /// installed onto. KVM renders it `-cdrom <iso>`; Redfish inserts it as virtual
    /// media. `None` (the default) leaves every launch byte-identical to a spec
    /// written before this field existed.
    ///
    /// The distinction from [`image`](BootSpec::image) is the whole point.
    /// `ImageSource::Iso` is a spec whose *only* payload is the ISO — a live boot,
    /// nothing to install onto. `medium` is the *install* shape: `image` is the
    /// blank disk that will receive the system, `medium` is the ISO doing the
    /// installing. Read them together through [`medium_path`](BootSpec::medium_path),
    /// which is the single accessor both backends use.
    ///
    /// It is meaningless on [`Backend::Container`] and [`Backend::Exe`] (no
    /// firmware, no drive tray), so [`validate`](BootSpec::validate) rejects it
    /// there rather than let it vanish silently.
    pub medium: Option<String>,
    /// **Which device the firmware boots from** — see [`BootOrder`].
    /// [`BootOrder::Auto`] (the default) emits nothing and changes no existing
    /// launch. [`BootOrder::Disk`] together with a [`medium`](BootSpec::medium) is
    /// refused by [`validate`](BootSpec::validate): that combination is the exact
    /// shape of the lie honesty rule 1 exists to stop.
    pub boot_order: BootOrder,
}

impl BootSpec {
    /// **The one writer of a default [`BootSpec`]** — every constructor below is
    /// this plus the two or three fields that make it that kind of boot.
    ///
    /// It exists because the five public constructors were five verbatim copies of
    /// the same 20-line field list (REUSE-law: >10 duplicated lines means share), and
    /// every new field had to be added five times or one constructor silently drifted.
    /// The `medium`/`boot_order` pair added 2026-08-15 is added here **once**, which
    /// is also what makes "a spec that names no medium behaves exactly as before" a
    /// property of the type rather than of five sites agreeing.
    ///
    /// KVM sizing (512 MiB / 2 cores) is applied only to [`Backend::Kvm`]: the
    /// container, Redfish and exe paths carry no VM sizing and stay `0`, exactly the
    /// "lean by design" defaults their constructors always produced.
    fn of(name: impl Into<String>, backend: Backend, image: ImageSource) -> Self {
        let (mem_mb, cores) = if backend == Backend::Kvm {
            (512, 2)
        } else {
            (0, 0)
        };
        Self {
            name: name.into(),
            backend,
            image,
            mem_mb,
            cores,
            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,
            security_opt: Vec::new(),
            extra_qemu_args: Vec::new(),
            persist_disk: false,
            medium: None,
            boot_order: BootOrder::Auto,
        }
    }

    /// 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::of(
            name,
            Backend::Kvm,
            ImageSource::KernelRootfs {
                kernel: kernel.into(),
                rootfs: rootfs.into(),
            },
        )
    }

    /// 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::of(
            name,
            Backend::Kvm,
            ImageSource::Disk {
                kernel: kernel.into(),
                disk: disk.into(),
            },
        )
    }

    /// A KVM boot from a **self-booting disk image** — a bootloader-in-disk
    /// (firmware) boot with **no `-kernel`** (defaults: 512 MiB, 2 cores). tunnr
    /// attaches the disk as a virtio drive and lets OVMF boot it directly, exactly
    /// as real hardware would; there is no direct-kernel launch.
    pub fn kvm_bootloader_disk(name: impl Into<String>, disk: impl Into<String>) -> Self {
        Self::of(
            name,
            Backend::Kvm,
            ImageSource::BootloaderDisk { disk: disk.into() },
        )
    }

    /// 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::of(
            name,
            Backend::Container,
            ImageSource::OciImage(oci_image.into()),
        )
    }

    /// A bare-metal Redfish boot: an ISO served as virtual media to a BMC node.
    ///
    /// The boot order stays [`BootOrder::Auto`] so this constructor is unchanged
    /// from every call written before boot order existed — the Redfish backend has
    /// always set a one-time [`BootTarget::Cd`] override itself. Say
    /// [`BootOrder::Medium`] explicitly (or use [`iso_boot`](BootSpec::iso_boot)
    /// + [`on_metal`](BootSpec::on_metal)) when the spec must *also* be routable to
    /// KVM.
    pub fn redfish_iso(name: impl Into<String>, iso: impl Into<String>, bmc: BmcEndpoint) -> Self {
        Self {
            bmc: Some(bmc),
            ..Self::of(name, Backend::Redfish, ImageSource::Iso(iso.into()))
        }
    }

    /// **Boot OFF a medium — the backend-agnostic ISO spec.** One spec, routable to
    /// the KVM backend as it stands or to real metal with [`on_metal`](BootSpec::on_metal),
    /// which is honesty rule 2 in a constructor: "it works in the demo" and "it works
    /// on the customer's server" become the same claim about the same value.
    ///
    /// Defaults to [`Backend::Kvm`] with [`BootOrder::Medium`] (`-cdrom <iso> -boot d`),
    /// 512 MiB / 2 cores. There is no target disk: this is a *live* boot of the ISO.
    /// For the install shape (ISO **plus** a blank disk to install onto) use
    /// [`kvm_install_from_medium`](BootSpec::kvm_install_from_medium).
    pub fn iso_boot(name: impl Into<String>, iso: impl Into<String>) -> Self {
        Self {
            boot_order: BootOrder::Medium,
            ..Self::of(name, Backend::Kvm, ImageSource::Iso(iso.into()))
        }
    }

    /// **Route this spec to real metal** (builder style): flip the backend to
    /// [`Backend::Redfish`] and attach the BMC, changing nothing else.
    ///
    /// The point is what it does *not* touch — image, medium and boot order carry
    /// over verbatim, so `iso_boot(..)` and `iso_boot(..).on_metal(bmc)` are the same
    /// boot claim aimed at two different machines. That is the `iso-kvm` / `iso-metal`
    /// pair the chain driver keeps as distinct rows: distinct *runs*, one spec.
    pub fn on_metal(mut self, bmc: BmcEndpoint) -> Self {
        self.backend = Backend::Redfish;
        self.bmc = Some(bmc);
        self
    }

    /// **The INSTALL leg**: boot off `iso` with `target_disk` present and blank, so
    /// the installer has somewhere to install to. KVM renders it as a firmware boot
    /// of a virtio `target_disk` plus `-cdrom <iso> -boot d`.
    ///
    /// [`persist_disk`](BootSpec::persist_disk) is **`true`** here, unlike every other
    /// disk constructor, and that is load-bearing: the default COW guard would send
    /// the installer's writes to a throwaway overlay, so the install would "succeed"
    /// and leave the disk byte-identical to the blank one it started from (measured on
    /// oden 2026-08-03). An install whose writes are discarded is the purest form of
    /// the green we must never accept.
    pub fn kvm_install_from_medium(
        name: impl Into<String>,
        iso: impl Into<String>,
        target_disk: impl Into<String>,
    ) -> Self {
        Self {
            medium: Some(iso.into()),
            boot_order: BootOrder::Medium,
            persist_disk: true,
            ..Self::of(
                name,
                Backend::Kvm,
                ImageSource::BootloaderDisk {
                    disk: target_disk.into(),
                },
            )
        }
    }

    /// **The BOOT-INSTALLED leg**: firmware-boot `disk` with **no medium at all**
    /// (`-boot c`) — the second, separate VM lifetime.
    ///
    /// This is deliberately a different constructor rather than a flag on the install
    /// spec, because the two must be two *values*: the driver's
    /// `an_installer_ready_green_is_refused_inside_one_vm_lifetime`
    /// (`edda/crates/nornir-testmatrix/src/chain.rs:1111`) refuses a READY green that
    /// was observed inside the installing machine. [`validate`](BootSpec::validate)
    /// enforces the same thing from below: a [`BootOrder::Disk`] spec that carries a
    /// medium is refused, so "I detached the ISO" cannot be merely claimed.
    ///
    /// `persist_disk` is `true`: the whole subject of this boot is what the previous
    /// lifetime wrote to that disk.
    pub fn kvm_boot_installed_disk(name: impl Into<String>, disk: impl Into<String>) -> Self {
        Self {
            boot_order: BootOrder::Disk,
            persist_disk: true,
            ..Self::of(
                name,
                Backend::Kvm,
                ImageSource::BootloaderDisk { disk: disk.into() },
            )
        }
    }

    /// **Run a plain executable** — the `exe` build-thing kind ([`exe::ExeBoot`]).
    /// `path` is the binary stage 1 produced; `args` is its argv[1..].
    pub fn exe<I, S>(name: impl Into<String>, path: impl Into<String>, args: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        Self::of(
            name,
            Backend::Exe,
            ImageSource::Executable {
                path: path.into(),
                args: args.into_iter().map(Into::into).collect(),
            },
        )
    }

    /// **The medium this spec boots off, wherever it was named** — the single
    /// accessor both the KVM and the Redfish backend read, so the two can never
    /// disagree about what "the medium" is (REUSE-law: one reader, not a twin per
    /// backend).
    ///
    /// Returns the explicit [`medium`](BootSpec::medium) when set, else the
    /// [`image`](BootSpec::image) when it is an [`ImageSource::Iso`] (a live ISO boot
    /// *is* a boot off the medium), else [`None`].
    pub fn medium_path(&self) -> Option<&str> {
        match (&self.medium, &self.image) {
            (Some(m), _) => Some(m.as_str()),
            (None, ImageSource::Iso(iso)) => Some(iso.as_str()),
            _ => None,
        }
    }

    /// Attach a removable boot [`medium`](BootSpec::medium) (builder style) — an ISO
    /// alongside the payload disk. Does **not** set the boot order; pair it with
    /// [`with_boot_order`](BootSpec::with_boot_order) or use
    /// [`kvm_install_from_medium`](BootSpec::kvm_install_from_medium).
    pub fn with_medium(mut self, medium: impl Into<String>) -> Self {
        self.medium = Some(medium.into());
        self
    }

    /// Set the [`boot_order`](BootSpec::boot_order) (builder style).
    pub fn with_boot_order(mut self, order: BootOrder) -> Self {
        self.boot_order = order;
        self
    }

    /// 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 **KVM extra QEMU args** (builder style) — see
    /// [`extra_qemu_args`](BootSpec::extra_qemu_args). Only the [`kvm`] backend acts
    /// on it (appended verbatim to tunnr's argv); empty leaves the argv byte-identical.
    pub fn with_extra_qemu_args<I, S>(mut self, args: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.extra_qemu_args = args.into_iter().map(Into::into).collect();
        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
    }

    /// Add one container **security option** (podman `--security-opt`, builder
    /// style) — see [`security_opt`](BootSpec::security_opt). Appends rather than
    /// replaces, because `--security-opt` is a repeatable flag and a setter that
    /// silently dropped the previous value would be the quiet kind of wrong.
    pub fn with_security_opt(mut self, opt: impl Into<String>) -> Self {
        self.security_opt.push(opt.into());
        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)?;
            }
            // A bootloader-in-disk boot is kernel-less by design (firmware boots
            // the disk), so require only the disk path — NOT a kernel.
            ImageSource::BootloaderDisk { disk } => require("disk path", disk)?,
            ImageSource::OciImage(image) => require("OCI image reference", image)?,
            ImageSource::Iso(iso) => require("ISO path", iso)?,
            ImageSource::Executable { path, .. } => require("executable path", path)?,
        }
        // ── the boot-off-a-medium seam ──────────────────────────────────────────
        // A medium is a firmware/drive-tray concept: only a machine (KVM guest or
        // Redfish node) has one. On a container or an exe it would vanish without a
        // trace, exactly like the container-only knobs rejected below.
        if self.medium.is_some()
            && !matches!(self.backend, Backend::Kvm | Backend::Redfish)
        {
            return Err(Error::Spec(format!(
                "a {:?} boot has no drive tray (medium is Kvm/Redfish-only)",
                self.backend
            )));
        }
        if let Some(medium) = &self.medium {
            require("medium (ISO) path", medium)?;
        }
        // A boot order is likewise a machine concept.
        if self.boot_order != BootOrder::Auto
            && !matches!(self.backend, Backend::Kvm | Backend::Redfish)
        {
            return Err(Error::Spec(format!(
                "a {:?} boot has no firmware boot order (boot_order is Kvm/Redfish-only)",
                self.backend
            )));
        }
        // "Boot off the medium" with no medium anywhere is a spec that cannot do what
        // it says. Named by field so the failure is actionable, not a generic reject.
        if self.boot_order == BootOrder::Medium && self.medium_path().is_none() {
            return Err(Error::Spec(
                "boot_order = Medium but no medium: set `medium` (an ISO alongside the \
                 payload disk) or make the image an ImageSource::Iso"
                    .into(),
            ));
        }
        // HONESTY RULE 1, enforced by construction. "Boot the installed disk" and
        // "the installer medium is still in the tray" cannot both be true: with the
        // ISO present, a machine that fails to boot its freshly-installed disk falls
        // through to the installer and comes up looking exactly like a success. The
        // driver refuses a READY green observed inside one VM lifetime
        // (edda/crates/nornir-testmatrix/src/chain.rs:1111); this refuses the SPEC
        // that would let it be claimed in the first place — a caller must build a
        // second, medium-less spec, which is a second boot.
        if self.boot_order == BootOrder::Disk && self.medium_path().is_some() {
            return Err(Error::Spec(format!(
                "boot_order = Disk but the medium `{}` is still attached: booting the \
                 INSTALLED system means the medium is ABSENT — build a separate spec \
                 (BootSpec::kvm_boot_installed_disk) for the second VM lifetime",
                self.medium_path().unwrap_or_default()
            )));
        }
        // 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` is container vocabulary EXCEPT the airgap: the kvm backend
            // honours `NetMode::None` as tunnr's `-nic none` (no egress NIC), the
            // exact KVM twin of the container's `--network none`. Every OTHER
            // non-default mode (Host/Bridge) is still container-only — on KVM it
            // would vanish without a trace, so those stay rejected. Redfish
            // provisions metal and honours none of them.
            let net_ok = self.net.is_default()
                || (self.backend == Backend::Kvm && self.net == NetMode::None);
            if !net_ok {
                return Err(Error::Spec(format!(
                    "a {:?} boot takes no container network mode (net is container-only, \
                     save `NetMode::None` on Kvm — the airgap)",
                    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(()),
        }
    }
}

// ---------------------------------------------------------------------------
// What a wait actually SAW — the shared stage-2 observation type.
// ---------------------------------------------------------------------------

/// **What was actually observed** while waiting on a running instance — the return
/// of [`exe::ExeBoot::await_marker`] and of [`kvm::KvmBoot::await_serial_marker`].
///
/// Note what is missing: there is no `Ok`/`Fail`, no "ready". Every one of these
/// outcomes can be the right answer depending on what the caller asked, and a type
/// that guessed would be inventing the verdict the caller exists to make. It carries
/// the evidence instead.
///
/// It lives at the crate root, not inside a backend, because it is the **one**
/// stage-2 observation shape: a process's captured stdout/stderr and a VM's serial
/// console are the same question ("did the named line appear before the deadline?")
/// asked of two different streams, and a per-backend copy would be two definitions
/// of "seen" that drift.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Seen {
    /// The named marker appeared in the captured output. Carries the **whole line**
    /// it appeared on — the applied output, not the fact that a boolean flipped.
    Marker {
        /// The full output line containing the marker.
        line: String,
        /// How long after the wait started it appeared.
        after: Duration,
    },
    /// The deadline elapsed with the process **still running** and the marker not
    /// seen. This is not automatically a failure: for a server, still-running is the
    /// healthy state, and the caller may well have been waiting to confirm exactly
    /// that. Carries the tail of what was captured so the caller can say *why*.
    StillRunning {
        /// The last lines captured, so a report can quote real output.
        tail: String,
        /// The budget that elapsed.
        waited: Duration,
    },
    /// The process **terminated** before the marker appeared. For a server this is
    /// the real failure, and the exit status plus tail is the evidence.
    Exited {
        /// Exit code, or `None` if it was killed by a signal.
        code: Option<i32>,
        /// The last lines captured.
        tail: String,
        /// How long it ran.
        after: Duration,
    },
}

impl Seen {
    /// Whether the named marker was observed — the one question with a single
    /// honest boolean answer.
    pub fn saw_marker(&self) -> bool {
        matches!(self, Seen::Marker { .. })
    }

    /// A one-line, quotable summary carrying the applied output (the matched line,
    /// or the tail that was captured instead) — what a matrix cell's `detail` should
    /// say so a red names what happened rather than that something did not.
    pub fn detail(&self) -> String {
        match self {
            Seen::Marker { line, after } => {
                format!("marker seen after {after:?}: {}", line.trim())
            }
            Seen::StillRunning { tail, waited } => format!(
                "still running after {waited:?}, marker NOT seen; last output: {}",
                one_line(tail)
            ),
            Seen::Exited { code, tail, after } => format!(
                "process EXITED (code {code:?}) after {after:?} without the marker; \
                 last output: {}",
                one_line(tail)
            ),
        }
    }
}

/// Squash a captured tail onto one line so it can live in a matrix cell.
fn one_line(s: &str) -> String {
    let joined = s
        .lines()
        .rev()
        .take(3)
        .collect::<Vec<_>>()
        .into_iter()
        .rev()
        .collect::<Vec<_>>()
        .join(" ⏎ ");
    if joined.trim().is_empty() {
        "<no output>".into()
    } else {
        joined
    }
}

/// 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()?;
    let outcome = backend.boot(spec);
    // Record the boot verdict as a functional-status row: GREEN when the instance
    // came up, RED when the backend failed to fire it — the boot surface nornir's
    // matrix reads back (a failed boot must be visible, never swallowed).
    functional_status(
        "draupnir/boot",
        "boot",
        outcome.is_ok(),
        &match &outcome {
            Ok(m) => format!("instance `{}` booted", m.id),
            Err(e) => format!("instance `{}` failed to boot: {e}", spec.name),
        },
    );
    outcome
}

/// **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)
}

/// **Roll a booted instance back** — tear down an instance a boot brought up (or a
/// half-finished provision) by powering it off through the backend's [`Lifecycle`],
/// recording the teardown verdict as a functional-status row. This is the *undo*
/// seam jera / Skidbladnir call when a provision must be reverted: a boot that never
/// reached [`PowerState::On`], a [`wait_ready`](container::ContainerBoot::wait_ready)
/// that timed out, or a fleet member being reclaimed. GREEN when the backend confirms
/// the power-off; a live backend failure to power it down is a **RED** row AND a
/// propagated [`Err`] — a rollback that could not complete must be visible, never
/// swallowed. Written against the plain [`Lifecycle`] trait so it rolls back **any**
/// backend (KVM / container / Redfish) and a mock in a unit test alike — one shared
/// teardown, never a per-backend twin.
pub fn rollback<L: Lifecycle>(lifecycle: &L, machine: &Machine) -> Result<()> {
    let outcome = lifecycle.power_off(machine);
    functional_status(
        "draupnir/lifecycle",
        "rollback",
        outcome.is_ok(),
        &match &outcome {
            Ok(()) => format!("instance `{}` rolled back (powered off)", machine.id),
            Err(e) => format!("instance `{}` rollback failed: {e}", machine.id),
        },
    );
    outcome
}

// ---------------------------------------------------------------------------
// 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 the **gatling fork-join pool** ([`gatling::gatling_forkjoin::gatling_for_each`]
/// — the constellation's one sanctioned threading home, ROOT-LAW #0), one worker per
/// member, 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 gatling workers); 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 the gatling fork-join pool (the constellation's ONE
    // sanctioned home for threading — ROOT-LAW #0: no bare `std::thread::spawn`).
    // `gatling_for_each` self-dispatches units and returns results in **index
    // order**, so the rollup stays deterministic and order-identical to the serial
    // path regardless of which member finishes first — identical effect to the
    // former `std::thread::scope` fan-out.
    //
    // Worker count is pinned to `plan.len()` (ONE worker per member), NOT the
    // default core count: `classify_member` is **blocking-IO** (boot a backend +
    // poll it up), not CPU work, so every member must have its own thread parked
    // on its own boot at the same time for the fleet wall-clock to collapse to the
    // *slowest* member (the whole point of this call). A core-bounded pool would
    // serialise `n > cores` members and break that guarantee. This is deliberately
    // exempt from the core-saturation law, which governs CPU hot paths, not
    // IO-bound waiters.
    let members: Vec<MemberReadback> =
        gatling::gatling_forkjoin::gatling_for_each(plan.len(), plan.len(), |i| {
            classify_member(backend, &plan[i], opts)
        });
    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));
        assert!(ImageSource::Executable {
            path: "/bin/true".into(),
            args: Vec::new()
        }
        .suits(Backend::Exe));
        // An ISO suits BOTH machine backends — changed 2026-08-15, and the point of
        // the change. `(Iso, Kvm)` used to be asserted here as REJECTED, which is
        // literally the wall that made the ISO chain unrunnable: there was no way to
        // hand the KVM backend an ISO at all. One spec must route to the faked KVM
        // machine and to real metal (honesty rule 2), so the pairing is legal on both.
        assert!(
            ImageSource::Iso("/boot.iso".into()).suits(Backend::Kvm),
            "an ISO is bootable under KVM (-cdrom): this is the iso-kvm leg"
        );
        // Cross pairings that remain nonsense are still rejected.
        assert!(!ImageSource::OciImage("redis:7".into()).suits(Backend::Redfish));
        assert!(!ImageSource::Iso("/boot.iso".into()).suits(Backend::Container));
        assert!(!ImageSource::OciImage("redis:7".into()).suits(Backend::Exe));
        assert!(!ImageSource::Executable {
            path: "/bin/true".into(),
            args: Vec::new()
        }
        .suits(Backend::Kvm));
    }

    #[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"
        );
    }

    // -- Boot off a medium: the two legs, and the lie between them ---------------

    #[test]
    fn the_install_leg_and_the_boot_installed_leg_are_two_distinct_specs() {
        // Honesty rule 1 in the type system. The install spec carries the medium and
        // boots off it; the boot-installed spec carries NO medium and boots the disk.
        // They are two values, so they can only be two boots.
        let install = BootSpec::kvm_install_from_medium("app", "/gunnar.iso", "/target.qcow2");
        install.validate().unwrap();
        assert_eq!(install.medium_path(), Some("/gunnar.iso"));
        assert_eq!(install.boot_order, BootOrder::Medium);
        assert_eq!(
            install.image,
            ImageSource::BootloaderDisk {
                disk: "/target.qcow2".into()
            },
            "the blank target disk is present to be installed ONTO"
        );
        assert!(
            install.persist_disk,
            "an install whose writes are thrown away installed nothing"
        );

        let installed = BootSpec::kvm_boot_installed_disk("app", "/target.qcow2");
        installed.validate().unwrap();
        assert_eq!(installed.medium_path(), None, "the medium is ABSENT");
        assert_eq!(installed.boot_order, BootOrder::Disk);
        assert!(installed.persist_disk);

        // Same disk, different machine: that is the whole claim.
        assert_ne!(install, installed);
    }

    #[test]
    fn a_boot_installed_spec_that_still_carries_the_medium_is_refused() {
        // THE guard. With the ISO still in the tray, a machine that fails to boot its
        // freshly-installed disk falls through to the installer and looks like a pass.
        // RED-when-broken: delete the `BootOrder::Disk && medium_path().is_some()`
        // arm in validate() and this spec sails through.
        let lie = BootSpec::kvm_boot_installed_disk("app", "/target.qcow2")
            .with_medium("/gunnar.iso");
        let err = lie.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(matches!(err, Error::Spec(_)));
        assert!(msg.contains("/gunnar.iso"), "names the medium: {msg}");
        assert!(msg.contains("ABSENT"), "says what boot-installed MEANS: {msg}");
        assert!(
            msg.contains("second VM lifetime"),
            "points at the fix: {msg}"
        );

        // ...and the same lie spelled the other way (an ISO payload + boot from disk).
        let lie2 = BootSpec::iso_boot("app", "/gunnar.iso").with_boot_order(BootOrder::Disk);
        assert!(
            matches!(lie2.validate(), Err(Error::Spec(_))),
            "an Iso payload counts as an attached medium too"
        );
    }

    #[test]
    fn boot_order_medium_with_no_medium_anywhere_is_refused_by_name() {
        // "Boot off the medium" with nothing in the tray cannot do what it says.
        // RED-when-broken: drop the `Medium && medium_path().is_none()` arm.
        let empty = BootSpec::kvm_bootloader_disk("app", "/d.qcow2")
            .with_boot_order(BootOrder::Medium);
        let err = empty.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("no medium"), "names the problem: {msg}");
        assert!(msg.contains("medium"), "names the field to set: {msg}");
    }

    #[test]
    fn one_iso_spec_routes_to_kvm_or_to_metal_without_changing_the_boot_claim() {
        // Honesty rule 2: `iso-kvm` and `iso-metal` are distinct RUNS of ONE spec.
        // Everything that describes the boot must survive the routing; only the
        // machine it is aimed at changes.
        let kvm = BootSpec::iso_boot("gunnar-server", "/images/gunnar-server.iso");
        kvm.validate().unwrap();
        assert_eq!(kvm.backend, Backend::Kvm);
        assert_eq!(kvm.boot_order, BootOrder::Medium);

        let metal = kvm.clone().on_metal(bmc());
        metal.validate().unwrap();
        assert_eq!(metal.backend, Backend::Redfish);
        // The boot claim is untouched by the routing.
        assert_eq!(metal.image, kvm.image);
        assert_eq!(metal.medium_path(), kvm.medium_path());
        assert_eq!(metal.boot_order, kvm.boot_order);
        // Each backend reads the SAME field in its own vocabulary — one field, two
        // renderings, so a KVM green and a metal green are the same claim.
        assert_eq!(kvm.boot_order.qemu_value(), Some("d"));
        assert_eq!(metal.boot_order.redfish_target(), Some(BootTarget::Cd));
    }

    #[test]
    fn a_medium_is_rejected_on_the_backends_that_have_no_drive_tray() {
        // RED-when-broken: drop the Kvm|Redfish guard and a container silently
        // accepts an ISO it will never read.
        let ctr = BootSpec::container("cache", "redis:7").with_medium("/boot.iso");
        assert!(matches!(ctr.validate(), Err(Error::Spec(_))), "medium on container rejected");
        let ctr_order = BootSpec::container("cache", "redis:7").with_boot_order(BootOrder::Disk);
        assert!(
            matches!(ctr_order.validate(), Err(Error::Spec(_))),
            "boot order on container rejected"
        );
        // A blank medium path is rejected like every other required path.
        let blank = BootSpec::kvm_bootloader_disk("app", "/d.qcow2").with_medium("  ");
        assert!(matches!(blank.validate(), Err(Error::Spec(_))), "blank medium rejected");
    }

    #[test]
    fn every_pre_existing_constructor_still_names_no_medium_and_no_boot_order() {
        // The additive-parity guard: the two new fields must leave every spec that
        // existed before them exactly as it was, so no caller's launch moves.
        for spec in [
            BootSpec::kvm_kernel_rootfs("a", "/bzImage", "/rootfs.cpio.gz"),
            BootSpec::kvm_disk("b", "/bzImage", "/d.qcow2"),
            BootSpec::kvm_bootloader_disk("c", "/d.qcow2"),
            BootSpec::container("d", "redis:7"),
            BootSpec::redfish_iso("e", "/boot.iso", bmc()),
        ] {
            assert_eq!(spec.medium, None, "{} names no explicit medium", spec.name);
            assert_eq!(
                spec.boot_order,
                BootOrder::Auto,
                "{} names no boot order",
                spec.name
            );
            assert_eq!(
                spec.boot_order.qemu_value(),
                None,
                "{} emits no -boot",
                spec.name
            );
            spec.validate().unwrap();
        }
        // The one pre-existing spec whose image IS an ISO still resolves a medium
        // through the shared accessor — that is how the Redfish backend keeps working.
        assert_eq!(
            BootSpec::redfish_iso("e", "/boot.iso", bmc()).medium_path(),
            Some("/boot.iso")
        );
    }

    #[test]
    fn the_one_constructor_writer_keeps_the_lean_and_kvm_defaults_apart() {
        // `BootSpec::of` is the single writer; prove it did not flatten the two
        // default shapes into one (KVM is sized, everything else is lean).
        assert_eq!(
            (
                BootSpec::kvm_bootloader_disk("k", "/d.qcow2").mem_mb,
                BootSpec::kvm_bootloader_disk("k", "/d.qcow2").cores
            ),
            (512, 2)
        );
        for lean in [
            BootSpec::container("c", "redis:7"),
            BootSpec::redfish_iso("r", "/b.iso", bmc()),
            BootSpec::exe("x", "/bin/true", Vec::<String>::new()),
        ] {
            assert_eq!((lean.mem_mb, lean.cores), (0, 0), "{} stays lean", lean.name);
        }
    }

    #[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());
    }
}