nornir 0.5.2

//! **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** ([`BollardBootRunner`], 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::collections::BTreeMap;
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
/// [`BollardBootRunner`] 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 [`BollardBootRunner`], which drives the container
/// lifecycle (create/start/logs/kill) over the podman/Docker REST API socket via
/// [`bollard`](https://docs.rs/bollard) — exactly like knut's `knut-spark-server`
/// `server` feature. This subprocess runner survives only as the graceful fallback
/// for a host with no API socket; the DEFAULT (feature on + socket present) is
/// bollard.
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 bollard transport (podman/Docker REST API).
// ---------------------------------------------------------------------------

/// The lifecycle state of a container as read from the engine API — the minimal
/// projection the boot lifecycle needs (running / exited-with-code / gone).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EngineState {
    /// The container is created and/or running (up).
    Running,
    /// The container exited on its own with this code (0 = clean).
    Exited(i64),
    /// No such container — removed, or never created.
    Gone,
}

/// The zero-shell container-engine seam: the create / start / logs / state / stop
/// operations a [`BollardBootRunner`] boot needs, over the podman/Docker REST API.
///
/// The production implementation is [`BollardEngine`] (feature
/// `test-boot-container-bollard`), which drives bollard against the podman socket.
/// Tests inject a [`FakeContainerEngine`] to exercise the whole create→logs→state→
/// kill wiring with NO real daemon. Crucially there is **no `std::process::Command`
/// anywhere on this path** — that is the difference from [`ContainerBootRunner`].
/// The boot-time **run overrides** applied to a container image when it is brought
/// up: an entrypoint/`cmd` override, extra `env`, and published `ports`. Derived
/// from the [`ContainerBuild`] descriptor (`cmd`/`env`/`expose`) in
/// [`BollardBootRunner::spawn`] so a boot can override the entrypoint/env and
/// publish ports instead of accepting only the image defaults. Plain data — the
/// [`FakeContainerEngine`] records it and the pure [`BollardEngine::create_body`]
/// folds it into a `ContainerCreateBody`, both without a daemon.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ContainerRunSpec {
    /// Command / entrypoint override (empty = the image's own default).
    pub cmd: Vec<String>,
    /// Extra environment, `KEY=VALUE` at the engine boundary (sorted, stable).
    pub env: BTreeMap<String, String>,
    /// Container ports to publish (each bound to the same host port).
    pub ports: Vec<u16>,
}

impl ContainerRunSpec {
    /// Derive the boot run overrides from a [`ContainerBuild`] descriptor: its
    /// `cmd`, `env`, and exposed ports (`expose`).
    pub fn from_container(c: &ContainerBuild) -> Self {
        ContainerRunSpec { cmd: c.cmd.clone(), env: c.env.clone(), ports: c.expose.clone() }
    }

    /// The `env` as engine `KEY=VALUE` strings (sorted for a stable body).
    pub fn env_pairs(&self) -> Vec<String> {
        self.env.iter().map(|(k, v)| format!("{k}={v}")).collect()
    }
}

pub trait ContainerEngine: Send + Sync {
    /// Create + start `image` as a detached container named `name` — applying the
    /// [`ContainerRunSpec`] boot overrides (cmd/env/published ports) — and begin
    /// streaming its logs. `Err` carries a human reason (the boot never started).
    fn create_and_start(&self, image: &str, name: &str, spec: &ContainerRunSpec) -> Result<(), String>;
    /// The container's current lifecycle state (a transient API hiccup is reported
    /// as [`EngineState::Running`] so it never fakes a death).
    fn state(&self, name: &str) -> EngineState;
    /// New log lines streamed since the last drain.
    fn drain_logs(&self, name: &str) -> Vec<String>;
    /// Stop + remove the container (idempotent, best-effort).
    fn stop(&self, name: &str);
}

/// A live container boot over a [`ContainerEngine`] — the API-driven analogue of
/// [`ExeProc`]. `poll` reads the engine state (so a container that dies on its own
/// flips the job), `kill` stops+removes it, `drain_log` pulls the streamed logs.
pub struct BollardProc {
    engine: Arc<dyn ContainerEngine>,
    /// The per-boot container name (`nornir-boot-<image>-<uuid>`).
    name: String,
    /// 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 BootProc for BollardProc {
    fn poll(&mut self) -> BootStatus {
        if let Some(t) = &self.terminal {
            return t.clone();
        }
        match self.engine.state(&self.name) {
            EngineState::Running => BootStatus::Running,
            EngineState::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
            }
            EngineState::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.engine.stop(&self.name);
        self.killed = true;
        self.terminal = Some(BootStatus::Killed);
    }

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

/// The **zero-shell** Container boot runner: brings a built [`ContainerBuild`] image
/// UP as a live container over the podman/Docker REST API (no `podman` subprocess),
/// streams its logs, and tears it down with an API stop/kill. This is the DEFAULT
/// container transport when `test-boot-container-bollard` is on and the API socket
/// is reachable; [`ContainerBootRunner`] is the fallback otherwise.
pub struct BollardBootRunner {
    engine: Arc<dyn ContainerEngine>,
}

impl BollardBootRunner {
    /// A runner over an injected [`ContainerEngine`] — the test seam (a
    /// [`FakeContainerEngine`] drives the whole lifecycle with no daemon).
    pub fn with_engine(engine: Arc<dyn ContainerEngine>) -> Self {
        BollardBootRunner { engine }
    }

    /// The per-boot container name for image `c`: stable prefix + a fresh uuid so a
    /// kill can address the exact container. Shared shape with [`ContainerBootRunner`].
    fn container_name(c: &ContainerBuild) -> String {
        format!("nornir-boot-{}-{}", c.name, uuid::Uuid::new_v4())
    }

    /// Connect to the real podman/Docker API socket (feature
    /// `test-boot-container-bollard`). `Err` when the socket is unavailable — the
    /// caller then falls back to the subprocess [`ContainerBootRunner`].
    #[cfg(feature = "test-boot-container-bollard")]
    pub fn connect() -> Result<Self, String> {
        Ok(Self::with_engine(Arc::new(BollardEngine::connect()?)))
    }
}

impl BootRunner for BollardBootRunner {
    fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
        let BootTarget::Container(c) = target else {
            return Err(format!("BollardBootRunner cannot boot a {} target", target.mode().as_str()));
        };
        let image = c.primary_tag().unwrap_or(c.name.as_str()).to_string();
        let name = Self::container_name(c);
        let spec = ContainerRunSpec::from_container(c);
        self.engine.create_and_start(&image, &name, &spec)?;
        Ok(Box::new(BollardProc { engine: Arc::clone(&self.engine), name, killed: false, terminal: None }))
    }
}

/// The production [`ContainerEngine`]: bollard against the podman/Docker API socket
/// (feature `test-boot-container-bollard`). Mirrors knut's `knut-spark-server`
/// zero-shell path — `Docker::connect_with_unix` → `create_container` /
/// `start_container` / `logs(follow)` / `stop_container`, tonic-free HTTP. A
/// dedicated tokio runtime drives the async API from the synchronous [`BootProc`]
/// seam; a background task streams each container's logs into a shared buffer.
#[cfg(feature = "test-boot-container-bollard")]
pub struct BollardEngine {
    docker: bollard::Docker,
    rt: tokio::runtime::Runtime,
    /// Per-container streamed-log buffers, filled by the follow tasks, drained by `drain_logs`.
    logs: Mutex<std::collections::HashMap<String, LogBuf>>,
}

#[cfg(feature = "test-boot-container-bollard")]
impl BollardEngine {
    /// Resolve the podman API socket URL: honour `DOCKER_HOST`, else the rootless
    /// user socket under `XDG_RUNTIME_DIR` (mirrors knut's resolution).
    fn socket_url() -> String {
        if let Ok(h) = std::env::var("DOCKER_HOST") {
            return h;
        }
        let xdg = std::env::var("XDG_RUNTIME_DIR").unwrap_or_else(|_| "/run/user/1000".into());
        format!("unix://{xdg}/podman/podman.sock")
    }

    /// Connect to podman/Docker over its API socket. Does NOT try to start it (no
    /// shelling out) — returns `Err` if the socket is absent so the caller can fall
    /// back to the subprocess runner.
    pub fn connect() -> Result<Self, String> {
        let url = Self::socket_url();
        let path = url.strip_prefix("unix://").unwrap_or(&url);
        if !std::path::Path::new(path).exists() {
            return Err(format!(
                "podman API socket not found at {path} (enable with `systemctl --user enable --now podman.socket`, \
                 or point DOCKER_HOST at a running Docker/podman socket)"
            ));
        }
        let docker = bollard::Docker::connect_with_unix(&url, 120, bollard::API_DEFAULT_VERSION)
            .map_err(|e| format!("connect podman socket {url}: {e}"))?;
        let rt = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .map_err(|e| format!("build tokio runtime for bollard boot: {e}"))?;
        Ok(BollardEngine { docker, rt, logs: Mutex::new(std::collections::HashMap::new()) })
    }
}

#[cfg(feature = "test-boot-container-bollard")]
impl BollardEngine {
    /// **Pure** builder of the `ContainerCreateBody` for `image`/`name` under the
    /// boot `spec` — factored out so the cmd/env/exposed-port wiring is testable
    /// without a daemon. A non-empty `spec.cmd` overrides the image entrypoint;
    /// `spec.env` is carried as `KEY=VALUE`; each `spec.ports[p]` is both exposed
    /// and published to the same host port via a `HostConfig` port binding.
    fn create_body(image: &str, spec: &ContainerRunSpec) -> bollard::models::ContainerCreateBody {
        use bollard::models::{ContainerCreateBody, HostConfig, PortBinding};
        use std::collections::HashMap;

        // `exposed_ports` is a list of `"<port>/<proto>"` keys; the actual host
        // publish lives in the `HostConfig` port bindings.
        let mut exposed: Vec<String> = Vec::new();
        let mut bindings: HashMap<String, Option<Vec<PortBinding>>> = HashMap::new();
        for p in &spec.ports {
            let key = format!("{p}/tcp");
            exposed.push(key.clone());
            bindings.insert(
                key,
                Some(vec![PortBinding {
                    host_ip: Some("0.0.0.0".to_string()),
                    host_port: Some(p.to_string()),
                }]),
            );
        }
        let host_config = if bindings.is_empty() {
            None
        } else {
            Some(HostConfig { port_bindings: Some(bindings), ..Default::default() })
        };
        ContainerCreateBody {
            image: Some(image.to_string()),
            cmd: if spec.cmd.is_empty() { None } else { Some(spec.cmd.clone()) },
            env: if spec.env.is_empty() { None } else { Some(spec.env_pairs()) },
            exposed_ports: if exposed.is_empty() { None } else { Some(exposed) },
            host_config,
            ..Default::default()
        }
    }
}

#[cfg(feature = "test-boot-container-bollard")]
impl ContainerEngine for BollardEngine {
    fn create_and_start(&self, image: &str, name: &str, spec: &ContainerRunSpec) -> Result<(), String> {
        use bollard::query_parameters::{
            CreateContainerOptionsBuilder, CreateImageOptionsBuilder, RemoveContainerOptionsBuilder,
        };
        use futures::StreamExt;

        let docker = &self.docker;
        self.rt.block_on(async {
            // Remove any stale container of the same name (ignore "not found").
            let _ = docker
                .remove_container(name, Some(RemoveContainerOptionsBuilder::new().force(true).build()))
                .await;
            // Pull the image only if it isn't present locally.
            if docker.inspect_image(image).await.is_err() {
                let (repo, tag) = image.rsplit_once(':').unwrap_or((image, "latest"));
                let opts = CreateImageOptionsBuilder::new().from_image(repo).tag(tag).build();
                let mut pull = docker.create_image(Some(opts), None, None);
                while let Some(item) = pull.next().await {
                    item.map_err(|e| format!("pull image {image}: {e}"))?;
                }
            }
            let config = Self::create_body(image, spec);
            docker
                .create_container(Some(CreateContainerOptionsBuilder::new().name(name).build()), config)
                .await
                .map_err(|e| format!("create container {name}: {e}"))?;
            docker
                .start_container(name, None::<bollard::query_parameters::StartContainerOptions>)
                .await
                .map_err(|e| format!("start container {name}: {e}"))?;
            Ok::<(), String>(())
        })?;

        // Wire a background log-follow task into a shared buffer this container's
        // `drain_logs` drains — the API-streamed equivalent of the reader threads.
        let buf: LogBuf = Arc::new(Mutex::new(Vec::new()));
        self.logs.lock().unwrap().insert(name.to_string(), Arc::clone(&buf));
        let docker = self.docker.clone();
        let name_owned = name.to_string();
        self.rt.spawn(async move {
            use bollard::container::LogOutput;
            use bollard::query_parameters::LogsOptionsBuilder;
            let mut stream = docker.logs(
                &name_owned,
                Some(LogsOptionsBuilder::new().follow(true).stdout(true).stderr(true).build()),
            );
            while let Some(item) = stream.next().await {
                match item {
                    Ok(out) => {
                        let line = LogOutput::to_string(&out);
                        let line = line.trim_end_matches(['\n', '\r']).to_string();
                        if !line.is_empty() {
                            buf.lock().unwrap().push(line);
                        }
                    }
                    Err(_) => break,
                }
            }
        });
        Ok(())
    }

    fn state(&self, name: &str) -> EngineState {
        let docker = &self.docker;
        self.rt.block_on(async {
            match docker
                .inspect_container(name, None::<bollard::query_parameters::InspectContainerOptions>)
                .await
            {
                Ok(info) => {
                    let state = info.state;
                    let running = state.as_ref().and_then(|s| s.running).unwrap_or(false);
                    if running {
                        EngineState::Running
                    } else {
                        // Not running: exited (use the code) unless it never ran.
                        let code = state.and_then(|s| s.exit_code).unwrap_or(0);
                        EngineState::Exited(code)
                    }
                }
                // 404 / not found → the container is gone. Other transient API
                // errors also land here; a killed boot is already terminal, and a
                // live boot's next poll retries — we never fabricate a death mid-run.
                Err(_) => EngineState::Gone,
            }
        })
    }

    fn drain_logs(&self, name: &str) -> Vec<String> {
        match self.logs.lock().unwrap().get(name) {
            Some(buf) => std::mem::take(&mut *buf.lock().unwrap()),
            None => Vec::new(),
        }
    }

    fn stop(&self, name: &str) {
        use bollard::query_parameters::RemoveContainerOptionsBuilder;
        let docker = &self.docker;
        self.rt.block_on(async {
            let _ = docker
                .stop_container(name, None::<bollard::query_parameters::StopContainerOptions>)
                .await;
            let _ = docker
                .remove_container(name, Some(RemoveContainerOptionsBuilder::new().force(true).build()))
                .await;
        });
        self.logs.lock().unwrap().remove(name);
    }
}

/// Pick the container [`BootRunner`] for the running build: the zero-shell
/// [`BollardBootRunner`] when `test-boot-container-bollard` is on AND the podman
/// API socket is reachable, else the subprocess [`ContainerBootRunner`] fallback.
fn default_container_runner() -> Box<dyn BootRunner> {
    #[cfg(feature = "test-boot-container-bollard")]
    {
        match BollardBootRunner::connect() {
            Ok(r) => return Box::new(r),
            Err(reason) => {
                eprintln!("test-boot: bollard API unavailable ({reason}); 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 bollard (feature on + socket reachable) 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 bollard runner: fake engine (no daemon) exercises the whole
    //    create → logs → state → kill wiring the real BollardEngine drives. ──────

    /// A scriptable fake [`ContainerEngine`] — a "fake bollard transport" that
    /// records create/start + stop calls and reports a preset state/logs, so the
    /// [`BollardBootRunner`]/[`BollardProc`] lifecycle is driven with NO real podman.
    #[derive(Default)]
    struct FakeEngineInner {
        /// (image, name) pairs `create_and_start` was called with.
        started: Vec<(String, String)>,
        /// The [`ContainerRunSpec`] forwarded on each `create_and_start` call.
        specs: Vec<ContainerRunSpec>,
        /// The state reported on `state` (defaults to Running once started).
        state: EngineStateOpt,
        /// Log lines emitted on the next `drain_logs`.
        logs: Vec<String>,
        /// Names `stop` was called with.
        stopped: Vec<String>,
        /// If set, `create_and_start` fails with this reason.
        fail_start: Option<String>,
    }
    /// `None` = "Running" (the fake's default before a test overrides it).
    type EngineStateOpt = Option<EngineState>;

    #[derive(Clone, Default)]
    struct FakeContainerEngine {
        inner: Arc<Mutex<FakeEngineInner>>,
    }
    impl FakeContainerEngine {
        fn set_state(&self, s: EngineState) {
            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 started(&self) -> Vec<(String, String)> {
            self.inner.lock().unwrap().started.clone()
        }
        fn specs(&self) -> Vec<ContainerRunSpec> {
            self.inner.lock().unwrap().specs.clone()
        }
        fn stopped(&self) -> Vec<String> {
            self.inner.lock().unwrap().stopped.clone()
        }
    }
    impl ContainerEngine for FakeContainerEngine {
        fn create_and_start(&self, image: &str, name: &str, spec: &ContainerRunSpec) -> Result<(), String> {
            let mut g = self.inner.lock().unwrap();
            if let Some(reason) = g.fail_start.clone() {
                return Err(reason);
            }
            g.started.push((image.to_string(), name.to_string()));
            g.specs.push(spec.clone());
            Ok(())
        }
        fn state(&self, _name: &str) -> EngineState {
            self.inner.lock().unwrap().state.clone().unwrap_or(EngineState::Running)
        }
        fn drain_logs(&self, _name: &str) -> Vec<String> {
            std::mem::take(&mut self.inner.lock().unwrap().logs)
        }
        fn stop(&self, name: &str) {
            self.inner.lock().unwrap().stopped.push(name.to_string());
        }
    }

    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 bollard_runner_creates_starts_streams_logs_and_kills() {
        // The fake transport drives the whole lifecycle with no daemon: spawn →
        // create_and_start (image + a nornir-boot-* name) → Running → streamed logs
        // → kill → engine stop. This is the zero-shell path's own wiring proof.
        let engine = FakeContainerEngine::default();
        engine.set_logs(&["holger up", "listening on :8080"]);
        let runner = BollardBootRunner::with_engine(Arc::new(engine.clone()));

        let mut proc = runner.spawn(&container_target()).expect("bollard spawn");
        // create_and_start got the primary tag as the image + a per-boot name.
        let started = engine.started();
        assert_eq!(started.len(), 1, "one create+start");
        assert_eq!(started[0].0, "holger-appliance:0.1", "image = primary tag");
        assert!(started[0].1.starts_with("nornir-boot-holger-appliance-"), "per-boot name: {}", started[0].1);

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

        proc.kill();
        assert_eq!(proc.poll(), BootStatus::Killed, "kill is terminal");
        assert_eq!(engine.stopped(), started.iter().map(|(_, n)| n.clone()).collect::<Vec<_>>(), "stop hit the exact container");
        // Idempotent: a second kill is a no-op (no extra stop).
        proc.kill();
        assert_eq!(engine.stopped().len(), 1, "kill is idempotent");
    }

    #[test]
    fn bollard_runner_maps_engine_state_to_terminal_status() {
        let engine = FakeContainerEngine::default();
        let runner = BollardBootRunner::with_engine(Arc::new(engine.clone()));

        // Exit 0 → BootedOk.
        engine.set_state(EngineState::Exited(0));
        let mut p = runner.spawn(&container_target()).unwrap();
        assert_eq!(p.poll(), BootStatus::BootedOk);

        // Non-zero exit → Failed with the code.
        engine.set_state(EngineState::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.
        engine.set_state(EngineState::Gone);
        let mut p = runner.spawn(&container_target()).unwrap();
        assert!(matches!(p.poll(), BootStatus::Failed(_)));
    }

    #[test]
    fn bollard_runner_drives_the_boot_job_lifecycle_via_the_state_manager() {
        // The BollardBootRunner 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 engine = FakeContainerEngine::default();
        let runner = Arc::new(BollardBootRunner::with_engine(Arc::new(engine.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).
        engine.set_state(EngineState::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 bollard_runner_start_failure_is_reported() {
        let engine = FakeContainerEngine::default();
        engine.inner.lock().unwrap().fail_start = Some("no such image".into());
        let runner = BollardBootRunner::with_engine(Arc::new(engine));
        let err = runner.spawn(&container_target()).err().expect("start failure surfaces");
        assert!(err.contains("no such image"), "reason carried: {err}");
    }

    #[test]
    fn bollard_runner_rejects_non_container_target() {
        let runner = BollardBootRunner::with_engine(Arc::new(FakeContainerEngine::default()));
        let err = runner.spawn(&exe_target()).err().expect("exe is not a container");
        assert!(err.contains("cannot boot a exe"), "{err}");
    }

    #[test]
    fn bollard_runner_forwards_cmd_env_and_ports_from_the_descriptor() {
        // A ContainerBuild that overrides cmd/env and publishes a port: the runner
        // must derive a ContainerRunSpec and hand it to the engine's create_and_start.
        let engine = FakeContainerEngine::default();
        let runner = BollardBootRunner::with_engine(Arc::new(engine.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("bollard spawn");

        let specs = engine.specs();
        assert_eq!(specs.len(), 1, "one create+start carried a spec");
        let spec = &specs[0];
        assert_eq!(spec.cmd, vec!["/usr/local/bin/holger", "--served"], "cmd override forwarded");
        assert_eq!(spec.env_pairs(), vec!["RUST_LOG=info"], "env forwarded as KEY=VALUE");
        assert_eq!(spec.ports, vec![8080], "published port forwarded");
    }

    #[test]
    fn empty_descriptor_yields_an_empty_run_spec() {
        // No overrides → an empty spec (the image's own defaults are used).
        let spec = ContainerRunSpec::from_container(&ContainerBuild {
            name: "x".into(),
            base: "scratch".into(),
            ..Default::default()
        });
        assert!(spec.cmd.is_empty() && spec.env.is_empty() && spec.ports.is_empty());
        assert_eq!(spec, ContainerRunSpec::default());
    }

    /// The pure `create_body` builder folds the run spec into a
    /// `ContainerCreateBody` — cmd/env carried, each port both exposed AND bound in
    /// the `HostConfig`. Verified WITHOUT a daemon (feature gates the bollard type).
    #[cfg(feature = "test-boot-container-bollard")]
    #[test]
    fn create_body_carries_cmd_env_and_port_bindings() {
        let spec = ContainerRunSpec {
            cmd: vec!["/bin/app".into(), "--serve".into()],
            env: BTreeMap::from([
                ("A".to_string(), "1".to_string()),
                ("B".to_string(), "2".to_string()),
            ]),
            ports: vec![8080],
        };
        let body = BollardEngine::create_body("app:1", &spec);
        assert_eq!(body.image.as_deref(), Some("app:1"));
        assert_eq!(body.cmd, Some(vec!["/bin/app".into(), "--serve".into()]));
        // env sorted KEY=VALUE.
        assert_eq!(body.env, Some(vec!["A=1".to_string(), "B=2".to_string()]));
        // The port is exposed AND bound to the same host port.
        let exposed = body.exposed_ports.expect("exposed ports set");
        assert!(exposed.iter().any(|s| s == "8080/tcp"), "8080/tcp exposed: {exposed:?}");
        let hc = body.host_config.expect("host config for port bindings");
        let bindings = hc.port_bindings.expect("port bindings");
        let b = bindings.get("8080/tcp").and_then(|v| v.clone()).expect("binding for 8080/tcp");
        assert_eq!(b[0].host_port.as_deref(), Some("8080"));
    }

    /// A spec with no ports yields no `HostConfig` and no exposed ports (nothing to
    /// publish) — and an empty cmd/env leaves those `None` (image defaults win).
    #[cfg(feature = "test-boot-container-bollard")]
    #[test]
    fn create_body_empty_spec_is_all_defaults() {
        let body = BollardEngine::create_body("app:1", &ContainerRunSpec::default());
        assert_eq!(body.image.as_deref(), Some("app:1"));
        assert!(body.cmd.is_none(), "no cmd override → image default entrypoint");
        assert!(body.env.is_none());
        assert!(body.exposed_ports.is_none());
        assert!(body.host_config.is_none());
    }

    /// REAL daemon path (feature `test-boot-container-bollard`): connect to the
    /// podman API socket, boot a tiny image, and kill it — proving the bollard
    /// 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_bollard -- --ignored --nocapture`.
    #[cfg(feature = "test-boot-container-bollard")]
    #[test]
    #[ignore = "real podman API: needs a running podman.socket + network pull"]
    fn real_bollard_container_boot_reaches_terminal() {
        let runner = BollardBootRunner::connect().expect("connect podman API socket");
        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. The bollard
        // path 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);
    }
}