nornir 0.5.3

//! **TEST-BOOT** — spin a *built* target up as a live nornir **job** from the
//! Build tab, in one of three modes, with a bidirectional job↔process lifecycle.
//!
//! Pressing a target's **test-boot** button creates a nornir job
//! ([`crate::jobs`]) and a live in-memory handle. The job tracks the boot's
//! status ([`BootStatus`]) and log, and the two directions of the lifecycle are
//! wired:
//!
//! - **kill the job → kill the boot**: cancelling a boot job SIGKILLs the child
//!   process-*group* ([`BootMode::Exe`]), stops the container
//!   ([`BootMode::Container`]), or calls the VM boot handle's `kill`
//!   ([`BootMode::Vm`]);
//! - **the boot dies → the job terminates**: if the process/container/VM exits on
//!   its own, the next [`TestBootState::poll`] flips the job to
//!   `BootedOk`/`Failed`/`Killed`.
//!
//! The three modes (see [`BootMode`]):
//! 1. [`Exe`](BootMode::Exe) — a plain cargo-variant binary: spawn the built exe
//!    as a child with an interactive console/PROMPT (stdin→child, stdout/stderr→UI
//!    log). Cheapest; no container/VM. **Implemented here** ([`ExeBootRunner`]).
//! 2. [`Container`](BootMode::Container) — a [`ContainerBuild`] image brought up
//!    over the **pure-Rust podman/Docker REST API** THROUGH draupnir's ONE OCI
//!    engine ([`DraupnirContainerBootRunner`], feature `test-boot-container-bollard`)
//!    — the zero-shell default — or, as a fallback when the API socket is absent,
//!    the podman-subprocess [`ContainerBootRunner`].
//! 3. [`Vm`](BootMode::Vm) — a [`VmBuild`] booted as a KVM guest via tunnr's boot
//!    primitive. nornir owns the [`BootSpec`] mapping + the [`VmBootRunner`] seam;
//!    the real boot lands once `tunnr_vm::boot_test` is committed (see
//!    [`StubVmBootRunner`]).
//!
//! The side-work is behind the [`BootRunner`] / [`BootProc`] seams so tests inject
//! a [`FakeBootRunner`] and drive the WHOLE lifecycle (spawn → running → death /
//! kill → terminal job) with no real process, container, or KVM in CI.

use std::io::{BufRead, BufReader, Write};
use std::path::PathBuf;
use std::process::{Child, ChildStdin, Command, Stdio};
use std::sync::{Arc, Mutex};

use nornir_build_thing::{ContainerBuild, VmBuild};

use crate::jobs::{kind, JobHandle, JobSink};

// ---------------------------------------------------------------------------
// Mode + status
// ---------------------------------------------------------------------------

/// Which of the three test-boot modes a built target uses. Selected purely from
/// the target's *type* ([`BootTarget::mode`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BootMode {
    /// A plain cargo-variant binary — spawn it as a child process + console.
    Exe,
    /// A `ContainerBuild` image — bring it up via podman/bollard.
    Container,
    /// A `VmBuild` — boot it as a KVM guest via tunnr.
    Vm,
}

impl BootMode {
    /// Stable label.
    pub fn as_str(&self) -> &'static str {
        match self {
            BootMode::Exe => "exe",
            BootMode::Container => "container",
            BootMode::Vm => "vm",
        }
    }

    /// The [`crate::jobs::kind`] a boot in this mode records.
    pub fn job_kind(&self) -> &'static str {
        match self {
            BootMode::Exe => kind::BOOT_EXE,
            BootMode::Container => kind::BOOT_CONTAINER,
            BootMode::Vm => kind::BOOT_VM,
        }
    }
}

/// Live status of a boot job. The lifecycle is
/// `Booting → Running → (BootedOk | Failed | Killed)`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BootStatus {
    /// Spawn issued; the target is coming up.
    Booting,
    /// The target is up and running (child alive / container up / VM booted).
    Running,
    /// The target booted and then exited cleanly (exit 0).
    BootedOk,
    /// The target failed to boot or exited non-zero. Carries a reason.
    Failed(String),
    /// The boot was killed (via the job) — or died by signal after a kill.
    Killed,
}

impl BootStatus {
    /// Whether the boot has reached a terminal state.
    pub fn is_terminal(&self) -> bool {
        matches!(self, BootStatus::BootedOk | BootStatus::Failed(_) | BootStatus::Killed)
    }

    /// Stable label for the UI / state_json.
    pub fn as_str(&self) -> &'static str {
        match self {
            BootStatus::Booting => "booting",
            BootStatus::Running => "running",
            BootStatus::BootedOk => "booted_ok",
            BootStatus::Failed(_) => "failed",
            BootStatus::Killed => "killed",
        }
    }
}

// ---------------------------------------------------------------------------
// Target + BootSpec (VmBuild → tunnr mapping)
// ---------------------------------------------------------------------------

/// A bootable target chosen from the build surface — the input to a test-boot.
/// Its *variant* fixes the [`BootMode`] (mode selection = target type → mode).
#[derive(Debug, Clone)]
pub enum BootTarget {
    /// A built cargo-variant binary: the exe path + args to run it with.
    Exe {
        /// The variant name (the job target label).
        name: String,
        /// The built executable to spawn.
        exe: PathBuf,
        /// Args passed to the exe.
        args: Vec<String>,
    },
    /// A container image target.
    Container(ContainerBuild),
    /// A VM target.
    Vm(VmBuild),
}

impl BootTarget {
    /// The human name of the target (the job's `target` label).
    pub fn name(&self) -> &str {
        match self {
            BootTarget::Exe { name, .. } => name,
            BootTarget::Container(c) => &c.name,
            BootTarget::Vm(v) => &v.name,
        }
    }

    /// **Mode selection** — the boot mode is fixed by the target's type.
    pub fn mode(&self) -> BootMode {
        match self {
            BootTarget::Exe { .. } => BootMode::Exe,
            BootTarget::Container(_) => BootMode::Container,
            BootTarget::Vm(_) => BootMode::Vm,
        }
    }
}

/// nornir's mapping of a [`VmBuild`] onto the boot primitive's spec — the shape we
/// hand to tunnr's boot primitive. Field-for-field it mirrors the `tunnr_vm`
/// `BootSpec` we will consume once `tunnr_vm::boot_test` lands (see
/// [`VmBootRunner`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BootSpec {
    /// The VM handle / guest name.
    pub name: String,
    /// Kernel image path (`-kernel`).
    pub kernel: String,
    /// Rootfs/initramfs path (`-initrd`), or empty for a disk boot.
    pub rootfs: String,
    /// Disk image path (`-drive`), or empty for a kernel+rootfs boot.
    pub disk: String,
    /// Guest memory (MiB).
    pub mem_mb: u32,
    /// Guest vCPU count.
    pub cores: u32,
    /// Kernel command line (`-append`).
    pub cmdline: String,
    /// Boot flavor (`qemu` / `microvm` / `appliance`).
    pub format: String,
    /// The materialized QEMU argv (the fallback boot recipe when tunnr is absent).
    pub qemu_args: Vec<String>,
}

impl BootSpec {
    /// Map a nornir [`VmBuild`] onto a [`BootSpec`]. This is the "nornir `VmBuild`
    /// → tunnr `BootSpec`" translation the Vm mode hands to the boot primitive.
    pub fn from_vm(vm: &VmBuild) -> Self {
        BootSpec {
            name: vm.name.clone(),
            kernel: vm.kernel.clone(),
            rootfs: vm.rootfs.clone(),
            disk: vm.disk.clone(),
            mem_mb: vm.mem_mb,
            cores: vm.cores,
            cmdline: vm.cmdline(),
            format: vm.format.as_str().to_string(),
            qemu_args: vm.qemu_args(),
        }
    }
}

// ---------------------------------------------------------------------------
// The runner / proc seams
// ---------------------------------------------------------------------------

/// A live boot's process handle — the in-memory side of a boot job. One per boot.
/// The bidirectional lifecycle rides on this trait: [`poll`](Self::poll) reports
/// death, [`kill`](Self::kill) tears the boot down.
pub trait BootProc: Send {
    /// The current status. MUST transition to a terminal state the moment the
    /// underlying process/container/VM dies on its own — that is the "boot dies →
    /// job terminates" direction.
    fn poll(&mut self) -> BootStatus;

    /// Kill the underlying process-group / container / VM — the "kill the job →
    /// kill the boot" direction. Idempotent.
    fn kill(&mut self);

    /// New log lines since the last drain (child stdout/stderr, serial console).
    fn drain_log(&mut self) -> Vec<String>;

    /// Send a line to the boot's stdin (the [`BootMode::Exe`] console PROMPT).
    /// Default no-op (containers/VMs have no interactive stdin here).
    fn send_stdin(&mut self, _line: &str) {}
}

/// The seam that spawns a boot for a [`BootTarget`]. The production dispatcher is
/// [`DispatchBootRunner`]; tests inject a [`FakeBootRunner`].
pub trait BootRunner: Send + Sync {
    /// Spawn the target; `Ok` carries the live [`BootProc`], `Err` a human reason
    /// (a spawn failure — the boot never started).
    fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String>;
}

// ---------------------------------------------------------------------------
// Exe mode — spawn the built binary as a child + console (REAL).
// ---------------------------------------------------------------------------

/// A shared, append-only log buffer the reader threads push into and the UI drains.
type LogBuf = Arc<Mutex<Vec<String>>>;

/// A live child process (Exe mode): the child, its stdin, a shared log buffer fed
/// by reader threads, and the "killed by us" flag that distinguishes an operator
/// kill from a natural death.
struct ExeProc {
    child: Child,
    stdin: Option<ChildStdin>,
    pid: u32,
    log: LogBuf,
    killed: bool,
    /// An optional engine teardown command run on `kill` BEFORE the process-group
    /// SIGKILL (e.g. `podman kill <name>`) — so a Container boot stops the actual
    /// container, not just the attached client. Best-effort (errors ignored).
    stop_cmd: Option<Vec<String>>,
    /// Cached terminal status once reached (so `poll` is stable + cheap after).
    terminal: Option<BootStatus>,
}

impl ExeProc {
    /// Reap-safe status read. Checks `try_wait`; maps exit → terminal status.
    fn compute(&mut self) -> BootStatus {
        if let Some(t) = &self.terminal {
            return t.clone();
        }
        match self.child.try_wait() {
            Ok(Some(status)) => {
                let t = if self.killed {
                    BootStatus::Killed
                } else if status.success() {
                    BootStatus::BootedOk
                } else {
                    BootStatus::Failed(format!("exited {status}"))
                };
                self.terminal = Some(t.clone());
                t
            }
            Ok(None) => BootStatus::Running,
            Err(e) => {
                let t = BootStatus::Failed(format!("wait failed: {e}"));
                self.terminal = Some(t.clone());
                t
            }
        }
    }
}

impl BootProc for ExeProc {
    fn poll(&mut self) -> BootStatus {
        self.compute()
    }

    fn kill(&mut self) {
        if self.terminal.is_some() {
            return;
        }
        // Engine teardown first (stop the actual container, if any), best-effort.
        if let Some(cmd) = self.stop_cmd.take() {
            if let Some((prog, args)) = cmd.split_first() {
                let _ = Command::new(prog)
                    .args(args)
                    .stdout(Stdio::null())
                    .stderr(Stdio::null())
                    .status();
            }
        }
        // Drop stdin so the child sees EOF, then SIGKILL the whole process GROUP
        // (the child was spawned in its own group = its pid), so any grandchildren
        // die too — not just the direct child.
        self.stdin = None;
        unsafe {
            libc::kill(-(self.pid as i32), libc::SIGKILL);
        }
        // Reap so no zombie lingers; SIGKILL makes this prompt.
        let _ = self.child.wait();
        self.killed = true;
        self.terminal = Some(BootStatus::Killed);
    }

    fn drain_log(&mut self) -> Vec<String> {
        let mut g = self.log.lock().unwrap();
        std::mem::take(&mut *g)
    }

    fn send_stdin(&mut self, line: &str) {
        if let Some(stdin) = self.stdin.as_mut() {
            let _ = stdin.write_all(line.as_bytes());
            let _ = stdin.write_all(b"\n");
            let _ = stdin.flush();
        }
    }
}

/// The Exe boot runner: spawns the built binary as a child in its OWN process
/// group (so a job-kill SIGKILLs the whole group), pipes stdin for the console
/// PROMPT, and streams stdout/stderr into the job log via reader threads.
pub struct ExeBootRunner;

impl ExeBootRunner {
    /// Spawn `program` with `args`, returning the live [`ExeProc`]. `stop_cmd` is an
    /// optional engine teardown run on `kill` (used by the container path). Shared
    /// by the Exe and Container subprocess paths.
    fn spawn_process(
        program: &std::ffi::OsStr,
        args: &[String],
        stop_cmd: Option<Vec<String>>,
    ) -> Result<Box<dyn BootProc>, String> {
        use std::os::unix::process::CommandExt;

        let mut cmd = Command::new(program);
        cmd.args(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        // Own process group (pgid = child pid) so `kill(-pgid, SIGKILL)` reaps the
        // whole tree on a job-kill.
        cmd.process_group(0);

        let mut child = cmd
            .spawn()
            .map_err(|e| format!("spawn `{}`: {e}", program.to_string_lossy()))?;
        let pid = child.id();

        let log: LogBuf = Arc::new(Mutex::new(Vec::new()));
        if let Some(out) = child.stdout.take() {
            spawn_reader(out, "", Arc::clone(&log));
        }
        if let Some(err) = child.stderr.take() {
            spawn_reader(err, "err: ", Arc::clone(&log));
        }
        let stdin = child.stdin.take();

        Ok(Box::new(ExeProc { child, stdin, pid, log, killed: false, stop_cmd, terminal: None }))
    }
}

impl BootRunner for ExeBootRunner {
    fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
        match target {
            BootTarget::Exe { exe, args, .. } => {
                Self::spawn_process(exe.as_os_str(), args, None)
            }
            other => Err(format!(
                "ExeBootRunner cannot boot a {} target",
                other.mode().as_str()
            )),
        }
    }
}

// ---------------------------------------------------------------------------
// Container mode — bring the built image up (podman) as a boot job.
// ---------------------------------------------------------------------------

/// The container engine used for a [`BootMode::Container`] boot (`podman` by
/// default; override with `NORNIR_CONTAINER_ENGINE`).
fn container_engine() -> String {
    std::env::var("NORNIR_CONTAINER_ENGINE").unwrap_or_else(|_| "podman".to_string())
}

/// The **fallback** Container boot runner: brings a built [`ContainerBuild`] image
/// UP as a live container by spawning the engine CLI, used ONLY when the zero-shell
/// [`DraupnirContainerBootRunner`] cannot reach the podman/Docker API socket (or the
/// `test-boot-container-bollard` feature is off).
///
/// Transport: this spawns the engine as an attached `<engine> run` subprocess
/// (`--name` set) so the same process-group + reader-thread machinery as
/// [`ExeBootRunner`] streams logs and reaps the client; `kill` first runs
/// `<engine> kill <name>` to stop the actual container, then SIGKILLs the group.
///
/// ZERO-SHELL: the constellation's charter forbids shelling out to `podman`. The
/// preferred transport is [`DraupnirContainerBootRunner`], which drives the
/// container lifecycle (create/start/logs/kill) through draupnir's ONE OCI engine
/// over the podman/Docker REST API socket — the same engine knut's `knut-spark-server`
/// `server` feature uses. This subprocess runner survives only as the graceful
/// fallback for a host with no API socket; the DEFAULT (feature on + socket present)
/// routes through draupnir.
pub struct ContainerBootRunner;

impl ContainerBootRunner {
    /// The `<engine> run` argv that brings `image` up under `container_name`.
    fn run_argv(image: &str, container_name: &str) -> Vec<String> {
        vec![
            "run".into(),
            "--rm".into(),
            "--name".into(),
            container_name.into(),
            image.into(),
        ]
    }
}

impl BootRunner for ContainerBootRunner {
    fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
        let BootTarget::Container(c) = target else {
            return Err(format!("ContainerBootRunner cannot boot a {} target", target.mode().as_str()));
        };
        let image = c.primary_tag().unwrap_or(c.name.as_str()).to_string();
        // A stable per-boot container name so `kill` can `<engine> kill <name>`.
        let container_name = format!("nornir-boot-{}-{}", c.name, uuid::Uuid::new_v4());
        let engine = container_engine();
        let args = ContainerBootRunner::run_argv(&image, &container_name);
        let stop_cmd = vec![engine.clone(), "kill".into(), container_name.clone()];
        ExeBootRunner::spawn_process(std::ffi::OsStr::new(&engine), &args, Some(stop_cmd))
    }
}

/// Stream one child pipe into the shared log, line by line, with a `prefix`.
fn spawn_reader<R: std::io::Read + Send + 'static>(reader: R, prefix: &'static str, log: LogBuf) {
    std::thread::spawn(move || {
        let buf = BufReader::new(reader);
        for line in buf.lines() {
            match line {
                Ok(l) => log.lock().unwrap().push(format!("{prefix}{l}")),
                Err(_) => break,
            }
        }
    });
}

// ---------------------------------------------------------------------------
// Container mode — ZERO-SHELL, routed THROUGH draupnir's ONE OCI engine.
// ---------------------------------------------------------------------------

/// Map a nornir [`ContainerBuild`] onto a `draupnir::BootSpec` (the OCI container
/// path). The image (`primary_tag`, else the name), a fresh per-boot name
/// (`nornir-boot-<name>-<uuid>` so a kill addresses the exact container), and the
/// run overrides (cmd / env / published ports) all carry across, so a boot through
/// draupnir brings up the same container nornir's old in-tree engine did.
fn container_build_to_draupnir(c: &ContainerBuild) -> draupnir::BootSpec {
    let image = c.primary_tag().unwrap_or(c.name.as_str()).to_string();
    let name = format!("nornir-boot-{}-{}", c.name, uuid::Uuid::new_v4());
    let mut d = draupnir::BootSpec::container(name, image).with_cmd(c.cmd.clone());
    d.ports = c.expose.clone();
    d.env = c.env.clone();
    d
}

/// The **zero-shell** Container boot runner that provisions THROUGH **draupnir** —
/// the shared boot library that owns the single bollard/podman REST engine for the
/// whole constellation. It maps a built [`ContainerBuild`] onto a `draupnir::BootSpec`
/// ([`container_build_to_draupnir`]), boots it via `draupnir::Boot`, and adapts the
/// returned `draupnir::Machine` into a [`BootProc`] over draupnir's
/// [`ContainerControl`](draupnir::container::ContainerControl) seam. There is **no
/// second engine and no `std::process::Command`** on this path — that is the
/// difference from the [`ContainerBootRunner`] subprocess fallback.
///
/// Generic over the draupnir backend `B` so a test injects a mock and proves the
/// nornir→draupnir routing with no daemon (draupnir's `Boot`/`ContainerControl` are
/// always compiled); the production `B` is [`draupnir::container::ContainerBoot`]
/// (the live engine, feature `test-boot-container-bollard = ["draupnir/backend-oci"]`).
#[derive(Default)]
pub struct DraupnirContainerBootRunner<B = draupnir::container::ContainerBoot> {
    backend: B,
}

impl<B> DraupnirContainerBootRunner<B>
where
    B: draupnir::Boot + draupnir::container::ContainerControl + Clone + Send + Sync + 'static,
{
    /// A runner over an explicit draupnir backend — the test seam (a mock backend
    /// drives the whole lifecycle with no daemon).
    pub fn with_backend(backend: B) -> Self {
        DraupnirContainerBootRunner { backend }
    }
}

impl<B> BootRunner for DraupnirContainerBootRunner<B>
where
    B: draupnir::Boot + draupnir::container::ContainerControl + Clone + Send + Sync + 'static,
{
    fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
        let BootTarget::Container(c) = target else {
            return Err(format!("DraupnirContainerBootRunner cannot boot a {} target", target.mode().as_str()));
        };
        let machine = self
            .backend
            .boot(&container_build_to_draupnir(c))
            .map_err(|e| format!("draupnir container boot: {e}"))?;
        Ok(Box::new(DraupnirContainerProc {
            backend: self.backend.clone(),
            machine,
            killed: false,
            terminal: None,
        }))
    }
}

/// Adapts a draupnir container `draupnir::Machine` to nornir's [`BootProc`], driving
/// its lifecycle over draupnir's [`ContainerControl`](draupnir::container::ContainerControl)
/// seam. `poll` maps draupnir's exit-code-aware
/// [`ContainerState`](draupnir::container::ContainerState) onto [`BootStatus`]
/// exactly as the old in-tree engine did (clean exit → `BootedOk`, crash →
/// `Failed(code)`, gone → `Failed`/`Killed`).
struct DraupnirContainerProc<B: draupnir::container::ContainerControl> {
    backend: B,
    machine: draupnir::Machine,
    /// Set once we asked the engine to stop it (distinguishes kill from natural death).
    killed: bool,
    /// Cached terminal status once reached (so `poll` is stable + cheap after).
    terminal: Option<BootStatus>,
}

impl<B: draupnir::container::ContainerControl + Send> BootProc for DraupnirContainerProc<B> {
    fn poll(&mut self) -> BootStatus {
        use draupnir::container::ContainerState;
        if let Some(t) = &self.terminal {
            return t.clone();
        }
        match self.backend.container_state(&self.machine) {
            ContainerState::Running => BootStatus::Running,
            ContainerState::Exited(code) => {
                let t = if self.killed {
                    BootStatus::Killed
                } else if code == 0 {
                    BootStatus::BootedOk
                } else {
                    BootStatus::Failed(format!("container exited {code}"))
                };
                self.terminal = Some(t.clone());
                t
            }
            ContainerState::Gone => {
                let t = if self.killed { BootStatus::Killed } else { BootStatus::Failed("container gone".into()) };
                self.terminal = Some(t.clone());
                t
            }
        }
    }

    fn kill(&mut self) {
        if self.terminal.is_some() {
            return;
        }
        self.backend.stop(&self.machine);
        self.killed = true;
        self.terminal = Some(BootStatus::Killed);
    }

    fn drain_log(&mut self) -> Vec<String> {
        self.backend.drain_logs(&self.machine)
    }
}

/// True iff the podman/Docker API socket path exists (honours `DOCKER_HOST`, else
/// the rootless user socket) — a cheap pre-check so the container runner can fall
/// back to the subprocess path when there is no socket, without linking bollard.
#[cfg(feature = "test-boot-container-bollard")]
fn podman_socket_present() -> bool {
    let url = std::env::var("DOCKER_HOST").unwrap_or_else(|_| {
        let xdg = std::env::var("XDG_RUNTIME_DIR").unwrap_or_else(|_| "/run/user/1000".into());
        format!("unix://{xdg}/podman/podman.sock")
    });
    let path = url.strip_prefix("unix://").unwrap_or(&url);
    std::path::Path::new(path).exists()
}

/// Pick the container [`BootRunner`] for the running build: the zero-shell
/// [`DraupnirContainerBootRunner`] (routed through draupnir's ONE OCI engine) when
/// `test-boot-container-bollard` is on AND the podman API socket is present, else the
/// subprocess [`ContainerBootRunner`] fallback.
fn default_container_runner() -> Box<dyn BootRunner> {
    #[cfg(feature = "test-boot-container-bollard")]
    {
        if podman_socket_present() {
            return Box::new(DraupnirContainerBootRunner::<draupnir::container::ContainerBoot>::default());
        }
        eprintln!("test-boot: podman API socket unavailable; falling back to the podman subprocess runner");
    }
    Box::new(ContainerBootRunner)
}

// ---------------------------------------------------------------------------
// Vm mode — nornir owns the seam; the real boot lands via tunnr.
// ---------------------------------------------------------------------------

/// The VM boot seam nornir owns until tunnr's boot primitive is committed.
///
/// TODO(tunnr): replace [`StubVmBootRunner`] with a thin adapter over the tunnr
/// API once the tunnr agent lands it. The exact API nornir needs:
///
/// ```ignore
/// // in crate `tunnr_vm`:
/// pub struct BootSpec { name, kernel, rootfs, disk, mem_mb, cores, cmdline, format }
/// pub fn boot_test(spec: BootSpec) -> Result<BootHandle, String>;
/// impl BootHandle {
///     fn status(&mut self) -> BootStatus;   // Booting/Running/BootedOk/Failed/Killed
///     fn kill(&mut self);                    // tear the guest down
///     fn drain_serial(&mut self) -> Vec<String>;
/// }
/// ```
///
/// nornir maps [`VmBuild`] → [`BootSpec`] ([`BootSpec::from_vm`]) and then
/// [`BootSpec`] → `tunnr_vm::BootSpec` field-for-field.
pub trait VmBootRunner: Send + Sync {
    /// Boot the VM described by `spec`, returning the live [`BootProc`].
    fn boot(&self, spec: &BootSpec) -> Result<Box<dyn BootProc>, String>;
}

/// The default VM runner when the real tunnr wire is not compiled in
/// (`--features test-boot-vm` off): it maps the spec but reports a clear, honest
/// failure rather than pretending to boot.
pub struct StubVmBootRunner;

impl VmBootRunner for StubVmBootRunner {
    fn boot(&self, spec: &BootSpec) -> Result<Box<dyn BootProc>, String> {
        Err(format!(
            "vm boot '{}' not wired in this build: rebuild with `--features test-boot-vm` to \
             drive tunnr_vm::boot_test(BootSpec) -> BootHandle (spec has {} qemu args)",
            spec.name,
            spec.qemu_args.len()
        ))
    }
}

/// The REAL VM runner (feature `test-boot-vm`): maps nornir's [`BootSpec`] onto
/// tunnr's `tunnr_vm::BootSpec`, calls `tunnr_vm::boot_test`, and adapts the
/// returned `BootHandle` into a [`BootProc`] — so the job lifecycle drives the
/// KVM: job-kill → `BootHandle::kill()`; `poll` → `BootHandle::poll_status()`
/// (which probes the QEMU pid, so an externally-dead VM flips the job too).
#[cfg(feature = "test-boot-vm")]
pub struct TunnrVmBootRunner;

#[cfg(feature = "test-boot-vm")]
impl VmBootRunner for TunnrVmBootRunner {
    fn boot(&self, spec: &BootSpec) -> Result<Box<dyn BootProc>, String> {
        use std::path::PathBuf;
        use std::time::Duration;

        // nornir BootSpec → tunnr BootSpec. tunnr takes a single boot image
        // (`rootfs_or_disk`): prefer the rootfs (initramfs), else the disk.
        let image = if !spec.rootfs.is_empty() { &spec.rootfs } else { &spec.disk };
        let tunnr_spec = tunnr_vm::BootSpec {
            kernel: PathBuf::from(&spec.kernel),
            rootfs_or_disk: PathBuf::from(image),
            mem_mb: spec.mem_mb,
            cores: spec.cores,
            timeout: Duration::from_secs(
                std::env::var("NORNIR_TEST_BOOT_VM_TIMEOUT")
                    .ok()
                    .and_then(|s| s.parse().ok())
                    .unwrap_or(120),
            ),
            headless: true,
            extra_qemu_args: Vec::new(),
            success_marker: tunnr_vm::DEFAULT_SUCCESS_MARKER.to_string(),
            kernel_cmdline: if spec.cmdline.is_empty() { "quiet".into() } else { spec.cmdline.clone() },
        };
        let handle = tunnr_vm::boot_test(tunnr_spec).map_err(|e| format!("tunnr boot_test: {e}"))?;
        Ok(Box::new(TunnrProc { handle, last_log_len: 0 }))
    }
}

/// Adapts a tunnr `BootHandle` to nornir's [`BootProc`] (feature `test-boot-vm`).
#[cfg(feature = "test-boot-vm")]
struct TunnrProc {
    handle: tunnr_vm::BootHandle,
    /// Byte length of the serial log already drained (tunnr returns the full log).
    last_log_len: usize,
}

#[cfg(feature = "test-boot-vm")]
impl BootProc for TunnrProc {
    fn poll(&mut self) -> BootStatus {
        match self.handle.poll_status() {
            tunnr_vm::BootStatus::Booting => BootStatus::Booting,
            tunnr_vm::BootStatus::BootedOk => BootStatus::BootedOk,
            tunnr_vm::BootStatus::Failed(r) => BootStatus::Failed(r),
            tunnr_vm::BootStatus::Killed => BootStatus::Killed,
            tunnr_vm::BootStatus::TimedOut => BootStatus::Failed("timed out".to_string()),
        }
    }
    fn kill(&mut self) {
        self.handle.kill();
    }
    fn drain_log(&mut self) -> Vec<String> {
        let full = self.handle.boot_log();
        if full.len() <= self.last_log_len {
            return Vec::new();
        }
        let new = full[self.last_log_len..].to_string();
        self.last_log_len = full.len();
        new.lines().map(|l| l.to_string()).collect()
    }
}

// ---------------------------------------------------------------------------
// Dispatch runner — routes a BootTarget to its mode's runner.
// ---------------------------------------------------------------------------

/// The production [`BootRunner`]: routes each [`BootTarget`] to the runner for its
/// mode — Exe → [`ExeBootRunner`], Container → [`ContainerBootRunner`], Vm → the
/// [`VmBootRunner`] seam.
pub struct DispatchBootRunner {
    exe: ExeBootRunner,
    container: Box<dyn BootRunner>,
    vm: Box<dyn VmBootRunner>,
}

impl Default for DispatchBootRunner {
    fn default() -> Self {
        // Feature `test-boot-vm` swaps the honest stub for the real tunnr KVM wire.
        #[cfg(feature = "test-boot-vm")]
        let vm: Box<dyn VmBootRunner> = Box::new(TunnrVmBootRunner);
        #[cfg(not(feature = "test-boot-vm"))]
        let vm: Box<dyn VmBootRunner> = Box::new(StubVmBootRunner);
        // Container path: zero-shell draupnir OCI engine (feature on + socket present) else
        // the podman-subprocess fallback.
        DispatchBootRunner { exe: ExeBootRunner, container: default_container_runner(), vm }
    }
}

impl BootRunner for DispatchBootRunner {
    fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
        match target {
            BootTarget::Exe { .. } => self.exe.spawn(target),
            BootTarget::Container(_) => self.container.spawn(target),
            BootTarget::Vm(vm) => self.vm.boot(&BootSpec::from_vm(vm)),
        }
    }
}

// ---------------------------------------------------------------------------
// The boot-job manager — job list integration + bidirectional lifecycle.
// ---------------------------------------------------------------------------

/// One live boot job: the durable job record's id + the observable status/log, and
/// the live [`BootProc`] handle + the [`JobHandle`] that writes the ledger row.
pub struct BootJob {
    /// The nornir job id (the ledger key).
    pub job_id: String,
    /// The target name (variant / image / vm).
    pub target: String,
    /// The boot mode.
    pub mode: BootMode,
    /// Live status (mirrors the ledger row + the last `poll`).
    pub status: BootStatus,
    /// The boot log (child stdout/stderr / serial), newest last.
    pub log: Vec<String>,
    /// The live process handle (None once terminal + reaped).
    proc: Option<Box<dyn BootProc>>,
    /// The ledger handle; `take`n and finished/failed on the terminal transition.
    handle: Option<JobHandle>,
}

impl BootJob {
    /// The observable JSON for state_json / a robot assert.
    pub fn state_json(&self) -> serde_json::Value {
        serde_json::json!({
            "job_id": self.job_id,
            "target": self.target,
            "mode": self.mode.as_str(),
            "status": self.status.as_str(),
            "terminal": self.status.is_terminal(),
            "log_lines": self.log.len(),
        })
    }
}

/// Manages the Build tab's live test-boot jobs: presses a boot (creating a nornir
/// job), polls each boot's lifecycle both directions, and kills on demand.
pub struct TestBootState {
    runner: Arc<dyn BootRunner>,
    sink: JobSink,
    workspace: String,
    jobs: Vec<BootJob>,
}

impl TestBootState {
    /// A manager wired to the real [`DispatchBootRunner`] and the given ledger
    /// `sink` (use [`JobSink::noop`] when no warehouse is configured).
    pub fn new(sink: JobSink, workspace: impl Into<String>) -> Self {
        Self::with_runner(Arc::new(DispatchBootRunner::default()), sink, workspace)
    }

    /// A manager with an injected [`BootRunner`] — the test seam (a
    /// [`FakeBootRunner`] drives the whole lifecycle with no real process).
    pub fn with_runner(
        runner: Arc<dyn BootRunner>,
        sink: JobSink,
        workspace: impl Into<String>,
    ) -> Self {
        TestBootState { runner, sink, workspace: workspace.into(), jobs: Vec::new() }
    }

    /// The live boot jobs (newest last).
    pub fn jobs(&self) -> &[BootJob] {
        &self.jobs
    }

    /// A clone of the injected [`BootRunner`] `Arc` — lets a caller re-wrap the
    /// manager around a new sink while preserving the (possibly faked) runner.
    pub fn runner_arc(&self) -> Arc<dyn BootRunner> {
        Arc::clone(&self.runner)
    }

    /// Whether a non-terminal boot job exists (drives the repaint request).
    pub fn any_running(&self) -> bool {
        self.jobs.iter().any(|j| !j.status.is_terminal())
    }

    /// The observable JSON — the boot jobs, for the Build tab's state_json.
    pub fn state_json(&self) -> serde_json::Value {
        serde_json::json!({
            "boot_jobs": self.jobs.iter().map(|j| j.state_json()).collect::<Vec<_>>(),
            "running": self.jobs.iter().filter(|j| !j.status.is_terminal()).count(),
        })
    }

    /// **Press test-boot.** Spawn `target` through the [`BootRunner`], open a
    /// `running` nornir job (kind = the mode's boot kind), and register the live
    /// boot job. On a spawn failure the job is opened and immediately failed (an
    /// honest terminal row), so a failed boot is still visible in the ledger.
    /// Returns the new job id.
    pub fn boot(&mut self, target: BootTarget) -> String {
        let mode = target.mode();
        let name = target.name().to_string();
        let handle = JobHandle::start(
            self.sink.clone(),
            mode.job_kind(),
            &name,
            &self.workspace,
            serde_json::json!({ "mode": mode.as_str() }),
        );
        let job_id = handle.job_id().to_string();

        match self.runner.spawn(&target) {
            Ok(proc) => {
                // EMIT-DOCTRINE: a launched boot is a REAL functional signal (green).
                nornir_testmatrix::functional_status(
                    "viz-test-boot",
                    &format!("boot_{}_{name}", mode.as_str()),
                    true,
                    &format!("test-boot '{name}' ({}) started", mode.as_str()),
                );
                self.jobs.push(BootJob {
                    job_id: job_id.clone(),
                    target: name,
                    mode,
                    status: BootStatus::Booting,
                    log: Vec::new(),
                    proc: Some(proc),
                    handle: Some(handle),
                });
            }
            Err(reason) => {
                nornir_testmatrix::functional_status(
                    "viz-test-boot",
                    &format!("boot_{}_{name}", mode.as_str()),
                    false,
                    &format!("test-boot '{name}' ({}) FAILED to start: {reason}", mode.as_str()),
                );
                handle.fail(&anyhow::anyhow!("{reason}"));
                self.jobs.push(BootJob {
                    job_id: job_id.clone(),
                    target: name,
                    mode,
                    status: BootStatus::Failed(reason),
                    log: Vec::new(),
                    proc: None,
                    handle: None,
                });
            }
        }
        job_id
    }

    /// **Poll every boot both directions.** Drain each boot's log, read its status;
    /// when a boot has DIED on its own (status went terminal), finish/fail the
    /// nornir job — the "boot dies → job terminates" direction.
    pub fn poll(&mut self) {
        for job in &mut self.jobs {
            if job.status.is_terminal() {
                continue;
            }
            let Some(proc) = job.proc.as_mut() else { continue };
            let new_lines = proc.drain_log();
            if !new_lines.is_empty() {
                job.log.extend(new_lines);
            }
            let status = proc.poll();
            if status != job.status {
                job.status = status.clone();
            }
            if status.is_terminal() {
                Self::terminate_job(job, &status);
            }
        }
    }

    /// **Kill a boot job by id** — the "kill the job → kill the boot" direction.
    /// SIGKILLs the process-group / stops the container / kills the VM, flips the
    /// status to `Killed`, and writes the terminal ledger row. Returns whether a
    /// live job matched.
    pub fn kill(&mut self, job_id: &str) -> bool {
        let Some(job) = self.jobs.iter_mut().find(|j| j.job_id == job_id) else {
            return false;
        };
        if job.status.is_terminal() {
            return false;
        }
        if let Some(proc) = job.proc.as_mut() {
            proc.kill();
            let drained = proc.drain_log();
            job.log.extend(drained);
        }
        job.status = BootStatus::Killed;
        Self::terminate_job(job, &BootStatus::Killed);
        true
    }

    /// Send a line to an Exe boot's console (the PROMPT).
    pub fn send_stdin(&mut self, job_id: &str, line: &str) {
        if let Some(job) = self.jobs.iter_mut().find(|j| j.job_id == job_id) {
            if let Some(proc) = job.proc.as_mut() {
                proc.send_stdin(line);
                job.log.push(format!("> {line}"));
            }
        }
    }

    /// Write the terminal ledger row for a boot that has reached `status`, and drop
    /// the live handle. Idempotent-safe: only runs when a `JobHandle` is still held.
    fn terminate_job(job: &mut BootJob, status: &BootStatus) {
        let Some(handle) = job.handle.take() else { return };
        let detail = serde_json::json!({
            "mode": job.mode.as_str(),
            "status": status.as_str(),
            "log_tail": job.log.iter().rev().take(20).rev().cloned().collect::<Vec<_>>(),
        });
        match status {
            BootStatus::BootedOk => handle.finish(detail, ""),
            BootStatus::Killed => handle.fail_with_detail(detail),
            BootStatus::Failed(reason) => {
                let mut d = detail;
                d["reason"] = serde_json::json!(reason);
                handle.fail_with_detail(d);
            }
            // Non-terminal — never called here, but keep the handle if so.
            _ => job.handle = Some(handle),
        }
        job.proc = None;
        // EMIT-DOCTRINE: the boot reached a terminal state — a functional row.
        let ok = matches!(status, BootStatus::BootedOk);
        nornir_testmatrix::functional_status(
            "viz-test-boot",
            &format!("boot_{}_{}_terminal", job.mode.as_str(), job.target),
            ok,
            &format!("test-boot '{}' → {}", job.target, status.as_str()),
        );
    }
}

// ---------------------------------------------------------------------------
// Test seam — a fake runner that drives the whole lifecycle with no real boot.
// ---------------------------------------------------------------------------

/// A scriptable fake [`BootProc`] for tests: its status is driven by a shared cell
/// (a test flips it to simulate the boot dying), and [`kill`](BootProc::kill)
/// records that it was asked to die.
#[derive(Clone)]
pub struct FakeProc {
    /// The status the fake reports on `poll` (a test mutates it to simulate death).
    pub status: Arc<Mutex<BootStatus>>,
    /// Set true when `kill` was called (asserts the kill reached the boot).
    pub killed: Arc<Mutex<bool>>,
    /// Pending log lines the fake emits on the next `drain_log`.
    pub pending_log: Arc<Mutex<Vec<String>>>,
}

impl FakeProc {
    /// A fake that starts `Running`.
    pub fn running() -> Self {
        FakeProc {
            status: Arc::new(Mutex::new(BootStatus::Running)),
            killed: Arc::new(Mutex::new(false)),
            pending_log: Arc::new(Mutex::new(Vec::new())),
        }
    }
}

impl BootProc for FakeProc {
    fn poll(&mut self) -> BootStatus {
        self.status.lock().unwrap().clone()
    }
    fn kill(&mut self) {
        *self.killed.lock().unwrap() = true;
        *self.status.lock().unwrap() = BootStatus::Killed;
    }
    fn drain_log(&mut self) -> Vec<String> {
        std::mem::take(&mut *self.pending_log.lock().unwrap())
    }
    fn send_stdin(&mut self, line: &str) {
        self.pending_log.lock().unwrap().push(format!("echo: {line}"));
    }
}

/// A fake [`BootRunner`] that hands back a preset [`FakeProc`] per spawn — the test
/// keeps a clone so it can flip the status (simulate death) or read `killed`.
pub struct FakeBootRunner {
    /// The proc handed to the NEXT spawn (a test presets it, then clones to drive).
    pub next: Mutex<Option<FakeProc>>,
    /// If set, spawn fails with this reason (test the failed-boot job path).
    pub fail: Mutex<Option<String>>,
}

impl FakeBootRunner {
    /// A runner whose next spawn yields `proc`.
    pub fn yielding(proc: FakeProc) -> Self {
        FakeBootRunner { next: Mutex::new(Some(proc)), fail: Mutex::new(None) }
    }
}

impl BootRunner for FakeBootRunner {
    fn spawn(&self, _target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
        if let Some(reason) = self.fail.lock().unwrap().clone() {
            return Err(reason);
        }
        let proc = self.next.lock().unwrap().clone().unwrap_or_else(FakeProc::running);
        Ok(Box::new(proc))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::jobs::{JobRecord, JobSink};

    /// A capturing sink: records every ledger write so a test can assert the job's
    /// running→terminal transitions landed.
    fn capturing_sink() -> (JobSink, Arc<Mutex<Vec<JobRecord>>>) {
        let cap: Arc<Mutex<Vec<JobRecord>>> = Arc::new(Mutex::new(Vec::new()));
        let c2 = Arc::clone(&cap);
        let sink = JobSink::remote(move |rec: &JobRecord| c2.lock().unwrap().push(rec.clone()));
        (sink, cap)
    }

    fn exe_target() -> BootTarget {
        BootTarget::Exe { name: "full".into(), exe: PathBuf::from("/bin/true"), args: vec![] }
    }

    #[test]
    fn mode_selection_maps_target_type_to_mode() {
        assert_eq!(exe_target().mode(), BootMode::Exe);
        assert_eq!(BootMode::Exe.job_kind(), kind::BOOT_EXE);

        let c = BootTarget::Container(ContainerBuild { name: "img".into(), ..Default::default() });
        assert_eq!(c.mode(), BootMode::Container);
        assert_eq!(BootMode::Container.job_kind(), kind::BOOT_CONTAINER);

        let v = BootTarget::Vm(VmBuild { name: "vm".into(), ..Default::default() });
        assert_eq!(v.mode(), BootMode::Vm);
        assert_eq!(BootMode::Vm.job_kind(), kind::BOOT_VM);
    }

    #[test]
    fn vm_boot_creates_a_boot_vm_job_via_fake_runner() {
        // Fake runner: the whole Vm lifecycle without QEMU/KVM. Proves the mode
        // wires a boot_vm job + bidirectional kill, independent of tunnr.
        let (sink, cap) = capturing_sink();
        let proc = FakeProc::running();
        let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
        let mut st = TestBootState::with_runner(runner, sink, "ws");

        let target = BootTarget::Vm(VmBuild {
            name: "holger-appliance".into(),
            kernel: "/boot/vmlinuz".into(),
            rootfs: "rootfs.cpio.gz".into(),
            ..Default::default()
        });
        assert_eq!(target.mode(), BootMode::Vm);
        let id = st.boot(target);
        st.poll();
        assert_eq!(st.jobs()[0].mode, BootMode::Vm);
        assert_eq!(st.jobs()[0].status, BootStatus::Running);

        // The VM dies on its own → job terminates (the "KVM dies → job" edge).
        *proc.status.lock().unwrap() = BootStatus::Failed("qemu vanished".into());
        st.poll();
        assert!(matches!(st.jobs()[0].status, BootStatus::Failed(_)));
        let recs = cap.lock().unwrap();
        assert_eq!(recs[0].kind, kind::BOOT_VM);
        let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
        assert_eq!(last.status, crate::jobs::status::FAILED);
    }

    /// REAL KVM boot via tunnr — heavy, needs QEMU+OVMF+KVM+a kernel/rootfs, so it
    /// is `#[ignore]`d and only meaningful under `--features test-boot-vm`. Mirrors
    /// tunnr's own ignored `boot_matrix` heavy path. Point it at real images with
    /// `TUNNR_KERNEL` / `TUNNR_INITRD`-style paths in a `VmBuild` and run:
    /// `cargo test --features test-boot-vm real_vm_boot -- --ignored --nocapture`.
    #[cfg(feature = "test-boot-vm")]
    #[test]
    #[ignore = "real KVM boot: needs QEMU+OVMF+KVM and a kernel/rootfs"]
    fn real_vm_boot_via_tunnr_reaches_terminal() {
        let kernel = std::env::var("NORNIR_VM_KERNEL").expect("set NORNIR_VM_KERNEL");
        let rootfs = std::env::var("NORNIR_VM_ROOTFS").expect("set NORNIR_VM_ROOTFS");
        let mut st = TestBootState::new(JobSink::noop(), "ws");
        let id = st.boot(BootTarget::Vm(VmBuild {
            name: "real".into(),
            kernel,
            rootfs,
            mem_mb: 512,
            cores: 2,
            ..Default::default()
        }));
        // Give it a moment, then kill — proves the bidirectional KVM teardown.
        std::thread::sleep(std::time::Duration::from_secs(3));
        st.poll();
        assert!(st.kill(&id), "kill the real VM");
        assert_eq!(st.jobs()[0].status, BootStatus::Killed);
    }

    #[test]
    fn vmbuild_maps_to_bootspec_field_for_field() {
        let vm = VmBuild {
            name: "holger-appliance".into(),
            kernel: "/boot/vmlinuz".into(),
            rootfs: "rootfs.cpio.gz".into(),
            mem_mb: 1024,
            cores: 4,
            boot_params: vec!["console=ttyS0".into()],
            ..Default::default()
        };
        let spec = BootSpec::from_vm(&vm);
        assert_eq!(spec.name, "holger-appliance");
        assert_eq!(spec.kernel, "/boot/vmlinuz");
        assert_eq!(spec.rootfs, "rootfs.cpio.gz");
        assert_eq!(spec.mem_mb, 1024);
        assert_eq!(spec.cores, 4);
        assert_eq!(spec.cmdline, "console=ttyS0");
        assert_eq!(spec.format, "qemu");
        assert_eq!(spec.qemu_args, vm.qemu_args());
    }

    #[test]
    fn container_run_argv_brings_image_up_named() {
        let argv = ContainerBootRunner::run_argv("holger:0.1", "nornir-boot-x");
        assert_eq!(argv, vec!["run", "--rm", "--name", "nornir-boot-x", "holger:0.1"]);
    }

    #[test]
    fn container_boot_creates_a_boot_container_job() {
        // Fake runner: no real podman. Proves the Container mode wires a job with the
        // boot_container kind + bidirectional kill, exactly like Exe.
        let (sink, cap) = capturing_sink();
        let proc = FakeProc::running();
        let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
        let mut st = TestBootState::with_runner(runner, sink, "ws");

        let target = BootTarget::Container(ContainerBuild {
            name: "holger-appliance".into(),
            base: "debian".into(),
            tags: vec!["holger-appliance:0.1".into()],
            ..Default::default()
        });
        assert_eq!(target.mode(), BootMode::Container);
        let id = st.boot(target);
        st.poll();
        assert_eq!(st.jobs()[0].status, BootStatus::Running);
        assert_eq!(st.jobs()[0].mode, BootMode::Container);

        assert!(st.kill(&id));
        assert!(*proc.killed.lock().unwrap());
        let recs = cap.lock().unwrap();
        assert_eq!(recs[0].kind, kind::BOOT_CONTAINER);
    }

    // ── zero-shell container runner: a MOCK draupnir backend (no daemon, no
    //    feature) exercises the whole boot → logs → state → kill wiring that the
    //    real draupnir::container::ContainerBoot drives — the ONE engine. ─────────

    use std::collections::BTreeMap;

    #[derive(Default)]
    struct MockDraupnirInner {
        booted: Vec<draupnir::BootSpec>,
        stopped: Vec<String>,
        state: Option<draupnir::container::ContainerState>,
        logs: Vec<String>,
        fail_boot: Option<String>,
    }

    /// A fake draupnir OCI backend: records the specs it was handed to boot + the
    /// machines it was asked to stop, and reports a scriptable container state /
    /// logs — so the nornir→draupnir wiring is driven with no real podman. It
    /// exercises only draupnir's always-compiled `Boot` + `ContainerControl` traits.
    #[derive(Clone, Default)]
    struct MockDraupnirBackend {
        inner: Arc<Mutex<MockDraupnirInner>>,
    }
    impl MockDraupnirBackend {
        fn set_state(&self, s: draupnir::container::ContainerState) {
            self.inner.lock().unwrap().state = Some(s);
        }
        fn set_logs(&self, lines: &[&str]) {
            self.inner.lock().unwrap().logs = lines.iter().map(|s| s.to_string()).collect();
        }
        fn booted(&self) -> Vec<draupnir::BootSpec> {
            self.inner.lock().unwrap().booted.clone()
        }
        fn stopped(&self) -> Vec<String> {
            self.inner.lock().unwrap().stopped.clone()
        }
    }
    impl draupnir::Boot for MockDraupnirBackend {
        fn boot(&self, spec: &draupnir::BootSpec) -> draupnir::Result<draupnir::Machine> {
            let mut g = self.inner.lock().unwrap();
            if let Some(reason) = g.fail_boot.clone() {
                return Err(draupnir::Error::Backend(reason));
            }
            g.booted.push(spec.clone());
            Ok(draupnir::Machine::started(format!("draupnir-{}", spec.name), spec))
        }
    }
    impl draupnir::container::ContainerControl for MockDraupnirBackend {
        fn container_state(&self, _m: &draupnir::Machine) -> draupnir::container::ContainerState {
            self.inner
                .lock()
                .unwrap()
                .state
                .clone()
                .unwrap_or(draupnir::container::ContainerState::Running)
        }
        fn drain_logs(&self, _m: &draupnir::Machine) -> Vec<String> {
            std::mem::take(&mut self.inner.lock().unwrap().logs)
        }
        fn stop(&self, m: &draupnir::Machine) {
            self.inner.lock().unwrap().stopped.push(m.id.clone());
        }
    }

    fn container_target() -> BootTarget {
        BootTarget::Container(ContainerBuild {
            name: "holger-appliance".into(),
            base: "debian".into(),
            tags: vec!["holger-appliance:0.1".into()],
            ..Default::default()
        })
    }

    #[test]
    fn draupnir_runner_boots_streams_logs_and_kills() {
        // The mock backend drives the whole lifecycle with no daemon: spawn → boot
        // through draupnir (image + a nornir-boot-* name) → Running → streamed logs
        // → kill → draupnir stop. This is the zero-shell path's own wiring proof.
        let backend = MockDraupnirBackend::default();
        backend.set_logs(&["holger up", "listening on :8080"]);
        let runner = DraupnirContainerBootRunner::with_backend(backend.clone());

        let mut proc = runner.spawn(&container_target()).expect("draupnir spawn");
        // The boot was routed to draupnir with the primary tag as image + a per-boot name.
        let booted = backend.booted();
        assert_eq!(booted.len(), 1, "one boot routed through draupnir");
        assert_eq!(booted[0].image, draupnir::ImageSource::OciImage("holger-appliance:0.1".into()));
        assert!(booted[0].name.starts_with("nornir-boot-holger-appliance-"), "per-boot name: {}", booted[0].name);

        assert_eq!(proc.poll(), BootStatus::Running, "container is up (draupnir state)");
        assert_eq!(proc.drain_log(), vec!["holger up", "listening on :8080"], "draupnir-streamed logs drained");

        proc.kill();
        assert_eq!(proc.poll(), BootStatus::Killed, "kill is terminal");
        assert_eq!(backend.stopped().len(), 1, "kill routed to draupnir stop");
        // Idempotent: a second kill is a no-op (no extra stop).
        proc.kill();
        assert_eq!(backend.stopped().len(), 1, "kill is idempotent");
    }

    #[test]
    fn draupnir_runner_maps_container_state_to_terminal_status() {
        let make = || {
            let backend = MockDraupnirBackend::default();
            let runner = DraupnirContainerBootRunner::with_backend(backend.clone());
            (backend, runner)
        };

        // Exit 0 → BootedOk.
        let (backend, runner) = make();
        backend.set_state(draupnir::container::ContainerState::Exited(0));
        let mut p = runner.spawn(&container_target()).unwrap();
        assert_eq!(p.poll(), BootStatus::BootedOk);

        // Non-zero exit → Failed with the code.
        let (backend, runner) = make();
        backend.set_state(draupnir::container::ContainerState::Exited(137));
        let mut p = runner.spawn(&container_target()).unwrap();
        assert!(matches!(p.poll(), BootStatus::Failed(r) if r.contains("137")));

        // Gone (removed out from under us) → Failed.
        let (backend, runner) = make();
        backend.set_state(draupnir::container::ContainerState::Gone);
        let mut p = runner.spawn(&container_target()).unwrap();
        assert!(matches!(p.poll(), BootStatus::Failed(_)));
    }

    #[test]
    fn draupnir_runner_drives_the_boot_job_lifecycle_via_the_state_manager() {
        // The DraupnirContainerBootRunner is a drop-in BootRunner: press-boot → job,
        // the container dies on its own → job terminates, all over the zero-shell seam.
        let (sink, cap) = capturing_sink();
        let backend = MockDraupnirBackend::default();
        let runner = Arc::new(DraupnirContainerBootRunner::with_backend(backend.clone()));
        let mut st = TestBootState::with_runner(runner, sink, "ws");

        let id = st.boot(container_target());
        st.poll();
        assert_eq!(st.jobs()[0].mode, BootMode::Container);
        assert_eq!(st.jobs()[0].status, BootStatus::Running);

        // Container exits cleanly on its own → the boot job terminates (done).
        backend.set_state(draupnir::container::ContainerState::Exited(0));
        st.poll();
        assert_eq!(st.jobs()[0].status, BootStatus::BootedOk);
        let recs = cap.lock().unwrap();
        assert_eq!(recs[0].kind, kind::BOOT_CONTAINER);
        let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
        assert_eq!(last.status, crate::jobs::status::DONE);
    }

    #[test]
    fn draupnir_runner_boot_failure_is_reported() {
        let backend = MockDraupnirBackend::default();
        backend.inner.lock().unwrap().fail_boot = Some("no such image".into());
        let runner = DraupnirContainerBootRunner::with_backend(backend);
        let err = runner.spawn(&container_target()).err().expect("boot failure surfaces");
        assert!(err.contains("no such image"), "reason carried: {err}");
    }

    #[test]
    fn draupnir_runner_rejects_non_container_target() {
        let runner = DraupnirContainerBootRunner::with_backend(MockDraupnirBackend::default());
        let err = runner.spawn(&exe_target()).err().expect("exe is not a container");
        assert!(err.contains("cannot boot a exe"), "{err}");
    }

    #[test]
    fn draupnir_runner_forwards_cmd_env_and_ports_from_the_descriptor() {
        // A ContainerBuild that overrides cmd/env and publishes a port: the runner
        // must map it onto a draupnir::BootSpec and hand that to draupnir's boot.
        let backend = MockDraupnirBackend::default();
        let runner = DraupnirContainerBootRunner::with_backend(backend.clone());
        let target = BootTarget::Container(ContainerBuild {
            name: "holger-appliance".into(),
            base: "debian".into(),
            tags: vec!["holger-appliance:0.1".into()],
            cmd: vec!["/usr/local/bin/holger".into(), "--served".into()],
            env: BTreeMap::from([("RUST_LOG".to_string(), "info".to_string())]),
            expose: vec![8080],
            ..Default::default()
        });
        let _ = runner.spawn(&target).expect("draupnir spawn");

        let booted = backend.booted();
        assert_eq!(booted.len(), 1, "one boot carried the mapped spec");
        let spec = &booted[0];
        assert_eq!(spec.image, draupnir::ImageSource::OciImage("holger-appliance:0.1".into()));
        assert_eq!(spec.cmd, vec!["/usr/local/bin/holger", "--served"], "cmd override forwarded");
        assert_eq!(spec.env.get("RUST_LOG").map(String::as_str), Some("info"), "env forwarded");
        assert_eq!(spec.ports, vec![8080], "published port forwarded");
    }

    #[test]
    fn container_build_with_no_overrides_maps_to_a_bare_draupnir_spec() {
        // No overrides → image + name only (the image's own defaults are used).
        let d = container_build_to_draupnir(&ContainerBuild {
            name: "x".into(),
            base: "scratch".into(),
            tags: vec!["x:1".into()],
            ..Default::default()
        });
        assert_eq!(d.backend, draupnir::Backend::Container);
        assert_eq!(d.image, draupnir::ImageSource::OciImage("x:1".into()));
        assert!(d.name.starts_with("nornir-boot-x-"), "per-boot name: {}", d.name);
        assert!(d.cmd.is_empty() && d.env.is_empty() && d.ports.is_empty());
        d.validate().expect("mapped container spec is valid");
    }

    /// REAL daemon path (feature `test-boot-container-bollard = ["draupnir/backend-oci"]`):
    /// boot a tiny image through draupnir's ONE OCI engine and kill it — proving the
    /// create/start/logs/kill wiring against a live engine. `#[ignore]`d (needs a
    /// running `podman.socket`): run with
    /// `cargo test --features test-boot-container-bollard real_draupnir -- --ignored --nocapture`.
    #[cfg(feature = "test-boot-container-bollard")]
    #[test]
    #[ignore = "real podman API: needs a running podman.socket + network pull"]
    fn real_draupnir_container_boot_reaches_terminal() {
        let runner = DraupnirContainerBootRunner::<draupnir::container::ContainerBoot>::default();
        let mut st = TestBootState::with_runner(Arc::new(runner), JobSink::noop(), "ws");
        // A long-lived image so the container is up when we kill it. draupnir boots
        // the image's own entrypoint/cmd (parity with the subprocess runner), so
        // point NORNIR_TEST_BOOT_IMAGE at an image that stays running.
        let image = std::env::var("NORNIR_TEST_BOOT_IMAGE")
            .unwrap_or_else(|_| "docker.io/library/nginx:alpine".to_string());
        let id = st.boot(BootTarget::Container(ContainerBuild {
            name: "nornir-real-boot".into(),
            base: image.clone(),
            tags: vec![image],
            ..Default::default()
        }));
        std::thread::sleep(std::time::Duration::from_secs(3));
        st.poll();
        // Kill if still up (long-lived image), then confirm a terminal state either
        // way (a short-lived image would already have exited on its own).
        let _ = st.kill(&id);
        st.poll();
        assert!(st.jobs()[0].status.is_terminal(), "real container reaches a terminal state");
    }

    #[test]
    fn stub_vm_runner_reports_honest_not_wired() {
        let vm = VmBuild { name: "vm".into(), kernel: "/k".into(), rootfs: "r".into(), ..Default::default() };
        let err = match StubVmBootRunner.boot(&BootSpec::from_vm(&vm)) {
            Ok(_) => panic!("stub must not pretend to boot"),
            Err(e) => e,
        };
        assert!(err.contains("tunnr_vm::boot_test"), "names the exact API: {err}");
    }

    #[test]
    fn press_boot_creates_a_running_job() {
        let (sink, cap) = capturing_sink();
        let runner = Arc::new(FakeBootRunner::yielding(FakeProc::running()));
        let mut st = TestBootState::with_runner(runner, sink, "ws");

        let id = st.boot(exe_target());
        assert_eq!(st.jobs().len(), 1);
        assert_eq!(st.jobs()[0].job_id, id);
        assert_eq!(st.jobs()[0].mode, BootMode::Exe);
        // Booting until first poll.
        assert_eq!(st.jobs()[0].status, BootStatus::Booting);
        assert!(st.any_running());

        // The ledger got a `running` boot_exe row.
        let recs = cap.lock().unwrap();
        assert_eq!(recs.len(), 1);
        assert_eq!(recs[0].kind, kind::BOOT_EXE);
        assert_eq!(recs[0].target, "full");
        assert_eq!(recs[0].status, crate::jobs::status::RUNNING);
    }

    #[test]
    fn poll_tracks_boot_status_running_then_death_terminates_job() {
        let (sink, cap) = capturing_sink();
        let proc = FakeProc::running();
        let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
        let mut st = TestBootState::with_runner(runner, sink, "ws");

        let id = st.boot(exe_target());
        st.poll();
        assert_eq!(st.jobs()[0].status, BootStatus::Running, "job tracks Running");

        // The boot DIES on its own (fake flips to BootedOk) → job terminates.
        *proc.status.lock().unwrap() = BootStatus::BootedOk;
        st.poll();
        assert_eq!(st.jobs()[0].status, BootStatus::BootedOk);
        assert!(!st.any_running());

        // The ledger saw running → done for this job.
        let recs = cap.lock().unwrap();
        let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
        assert_eq!(last.status, crate::jobs::status::DONE, "natural exit 0 → done");
    }

    #[test]
    fn killing_the_job_kills_the_boot_and_marks_killed() {
        let (sink, cap) = capturing_sink();
        let proc = FakeProc::running();
        let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
        let mut st = TestBootState::with_runner(runner, sink, "ws");

        let id = st.boot(exe_target());
        st.poll();
        assert_eq!(st.jobs()[0].status, BootStatus::Running);

        // Kill the JOB → the boot is killed (bidirectional).
        assert!(st.kill(&id));
        assert!(*proc.killed.lock().unwrap(), "kill reached the boot proc");
        assert_eq!(st.jobs()[0].status, BootStatus::Killed);
        assert!(!st.any_running());

        // The ledger row is terminal (failed carries the killed detail).
        let recs = cap.lock().unwrap();
        let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
        assert!(crate::jobs::status::is_terminal(&last.status));
        assert!(last.detail_json.contains("killed"), "killed detail carried: {}", last.detail_json);

        // Killing an already-terminal job is a no-op.
        assert!(!st.kill(&id));
    }

    #[test]
    fn spawn_failure_opens_and_immediately_fails_the_job() {
        let (sink, cap) = capturing_sink();
        let runner = FakeBootRunner::yielding(FakeProc::running());
        *runner.fail.lock().unwrap() = Some("no such exe".into());
        let mut st = TestBootState::with_runner(Arc::new(runner), sink, "ws");

        let id = st.boot(exe_target());
        assert!(matches!(st.jobs()[0].status, BootStatus::Failed(_)));
        let recs = cap.lock().unwrap();
        let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
        assert_eq!(last.status, crate::jobs::status::FAILED);
    }

    /// REAL Exe boot (no fake): spawn a short-lived binary, watch it reach a
    /// terminal state, and prove the process-group kill path compiles + runs. Uses
    /// `/bin/sh -c` so we control lifetime; gated to unix.
    #[test]
    fn real_exe_boot_spawns_and_reaches_terminal() {
        let runner = Arc::new(DispatchBootRunner::default());
        let mut st = TestBootState::with_runner(runner, JobSink::noop(), "ws");
        // A sleeper we will KILL — proves the bidirectional process-group kill.
        let target = BootTarget::Exe {
            name: "sleeper".into(),
            exe: PathBuf::from("/bin/sh"),
            args: vec!["-c".into(), "sleep 30".into()],
        };
        let id = st.boot(target);
        st.poll();
        assert_eq!(st.jobs()[0].status, BootStatus::Running, "sleeper is running");
        assert!(st.kill(&id), "kill the running sleeper");
        assert_eq!(st.jobs()[0].status, BootStatus::Killed);
    }
}