draupnir 0.1.8

//! **Container backend** — fire up an OCI container instance (e.g. a redis
//! service) over a container runtime.
//!
//! Draupnir does not reimplement a runtime. This backend is the thin adapter that
//! maps a Draupnir [`BootSpec`] onto **`bollard`** — the async podman/Docker REST
//! client `jera` already drives for its zero-shell container path — and runs the
//! container lifecycle over it. Reusing bollard keeps one container engine across
//! the constellation rather than a second bespoke one.
//!
//! It sits behind the `backend-oci` feature so the default build stays pure-std;
//! the trait wiring compiles unconditionally. **Zero-shell**: every operation is a
//! Rust API call over the podman/Docker socket — never a `podman`/`docker`
//! subprocess. If no daemon is reachable the backend **degrades with a clear
//! error** (there is deliberately no CLI fallback), it never fakes a boot.

use crate::{Boot, BootSpec, Error, ImageSource, Lifecycle, Machine, PowerState, Result};
// `PortMap` is referenced only inside the bollard-gated engine (create-body /
// create-and-start), so importing it unconditionally warns when `backend-oci` is
// off; the gated sites qualify it as `crate::PortMap` instead.

use std::path::Path;
use std::time::{Duration, Instant};

#[cfg(feature = "backend-oci")]
use std::sync::{Arc, Mutex};

/// **How draupnir decides a container is *app-ready*** — one level above the bare
/// `running` power state. [`ContainerBoot::wait_ready`] blocks on this until it
/// holds or the timeout elapses.
///
/// A freshly `create_and_start`ed container reports [`PowerState::On`] the instant
/// its main process is spawned, which is *not* the same as the app inside having
/// come up (a redis still opening its listen socket, a service still reading its
/// config). [`Readiness::Running`] is the base state poll (parity with the old
/// bare-`running` path); [`Readiness::LogMatch`] waits for the app to *announce*
/// itself on its own logs — the readiness signal the spec can opt into.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum Readiness {
    /// Ready as soon as the container's main process reports `running`
    /// ([`PowerState::On`]) — the base state poll, equivalent to the pre-existing
    /// bare-`running` behaviour.
    #[default]
    Running,
    /// Ready once `needle` appears anywhere in the container's stdout/stderr logs
    /// (e.g. `"Ready to accept connections"` for redis) — a readiness probe the
    /// spec supplies.
    LogMatch(String),
}

/// **Poll `check` until it reports ready, or `timeout` elapses.** The pure,
/// backend-independent core of [`ContainerBoot::wait_ready`]: it owns the deadline
/// arithmetic + sleep cadence and turns a timeout into a clear [`Error::Backend`]
/// naming the instance and the elapsed budget. `check` returns `Ok(true)` when
/// ready, `Ok(false)` to keep polling, and `Err(..)` to fail fast (a backend error
/// is never swallowed as "not ready yet"). Unit-tested with no live daemon.
#[cfg_attr(not(feature = "backend-oci"), allow(dead_code))]
fn poll_until_ready(
    label: &str,
    timeout: Duration,
    interval: Duration,
    mut check: impl FnMut() -> Result<bool>,
) -> Result<()> {
    let deadline = Instant::now() + timeout;
    loop {
        if check()? {
            return Ok(());
        }
        let now = Instant::now();
        if now >= deadline {
            return Err(Error::Backend(format!(
                "container `{label}` not ready within {timeout:?}"
            )));
        }
        // Never sleep past the deadline.
        let remaining = deadline.saturating_duration_since(now);
        std::thread::sleep(interval.min(remaining));
    }
}

/// The OCI container boot backend.
#[derive(Default, Clone)]
pub struct ContainerBoot {
    /// The connected engine (a bollard `Docker` + its own tokio runtime), built
    /// lazily on first use and shared across [`Boot`]/[`Lifecycle`] calls.
    #[cfg(feature = "backend-oci")]
    engine: Arc<Mutex<Option<Arc<Engine>>>>,
}

impl std::fmt::Debug for ContainerBoot {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ContainerBoot").finish_non_exhaustive()
    }
}

impl ContainerBoot {
    /// Construct the container backend.
    pub fn new() -> Self {
        Self::default()
    }

    /// The OCI image reference this spec will pull + run.
    pub fn image_ref<'a>(&self, spec: &'a BootSpec) -> Result<&'a str> {
        match &spec.image {
            ImageSource::OciImage(r) => Ok(r.as_str()),
            other => Err(Error::Spec(format!(
                "container backend needs an OCI image, got {other:?}"
            ))),
        }
    }

    /// The per-instance container name derived from the spec name.
    #[cfg_attr(not(feature = "backend-oci"), allow(dead_code))]
    fn container_name(spec: &BootSpec) -> String {
        format!("draupnir-{}", spec.name)
    }

    /// **Block until the container is app-ready**, or return a clear timeout error.
    ///
    /// Folds in `ContainerController`'s `wait_ready` readiness model: it polls the
    /// container state (and, for [`Readiness::LogMatch`], scans its logs) on a fixed
    /// cadence until `readiness` holds or `timeout` elapses. On timeout it returns an
    /// [`Error::Backend`] naming the instance and the budget — never a fake "ready".
    /// The bare-`running` boot path is unchanged; this is an additive step a caller
    /// runs *after* [`Boot::boot`] when it needs the app up, not just the process.
    pub fn wait_ready(
        &self,
        machine: &Machine,
        readiness: &Readiness,
        timeout: Duration,
    ) -> Result<()> {
        #[cfg(feature = "backend-oci")]
        {
            let engine = self.engine()?;
            let name = machine.id.clone();
            poll_until_ready(&name, timeout, Duration::from_millis(200), || {
                match readiness {
                    Readiness::Running => Ok(matches!(engine.power_state(&name), PowerState::On)),
                    Readiness::LogMatch(needle) => engine.log_contains(&name, needle),
                }
            })
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (machine, readiness, timeout);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature".into(),
            ))
        }
    }

    /// The connected engine, built (and cached) on first use. `Err` when no
    /// podman/Docker socket is reachable — the honest degrade, no shell fallback.
    #[cfg(feature = "backend-oci")]
    fn engine(&self) -> Result<Arc<Engine>> {
        let mut guard = self.engine.lock().unwrap();
        if let Some(e) = guard.as_ref() {
            return Ok(Arc::clone(e));
        }
        let e = Arc::new(Engine::connect()?);
        *guard = Some(Arc::clone(&e));
        Ok(e)
    }
}

impl Boot for ContainerBoot {
    fn boot(&self, spec: &BootSpec) -> Result<Machine> {
        spec.validate()?;
        let image = self.image_ref(spec)?;
        #[cfg(feature = "backend-oci")]
        {
            let name = Self::container_name(spec);
            let env: Vec<String> = spec.env.iter().map(|(k, v)| format!("{k}={v}")).collect();
            // Thread the spec's network mode + resource caps onto the live create
            // body — all default (`None`) leaves the run byte-identical; `NetMode::None`
            // → `--network none`; `cpus`/`mem_limit_mb` set the podman `--cpus`/
            // `--memory` (unset = all host cores, unconstrained memory: the hot-infra
            // default so FalkorDB/Spark are never throttled to one core).
            self.engine()?.create_and_start_with_binds_and_net(
                image, &name, &env, &spec.cmd, &spec.ports, &spec.port_maps, &[], spec.net.oci_value(),
                spec.cpus, spec.mem_limit_mb,
            )?;
            Ok(Machine::started(name, spec))
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = image;
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }
}

impl Lifecycle for ContainerBoot {
    fn power_on(&self, machine: &Machine) -> Result<()> {
        #[cfg(feature = "backend-oci")]
        {
            self.engine()?.start(&machine.id)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            Err(Error::Unsupported("container backend needs the `backend-oci` feature".into()))
        }
    }

    fn power_off(&self, machine: &Machine) -> Result<()> {
        #[cfg(feature = "backend-oci")]
        {
            self.engine()?.stop(&machine.id);
            Ok(())
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            Err(Error::Unsupported("container backend needs the `backend-oci` feature".into()))
        }
    }

    fn status(&self, machine: &Machine) -> Result<PowerState> {
        #[cfg(feature = "backend-oci")]
        {
            Ok(self.engine()?.power_state(&machine.id))
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            Err(Error::Unsupported("container backend needs the `backend-oci` feature".into()))
        }
    }
}

// ---------------------------------------------------------------------------
// ContainerControl — the exit-code-aware + log-streaming container seam.
// ---------------------------------------------------------------------------

/// The lifecycle state of a container as read from the engine — the richer
/// projection a **job runner** ([`jera`](https://codeberg.org/nordisk/edda))
/// needs, one level below the generic power [`Lifecycle`] (which collapses every
/// non-running state to [`PowerState::Off`] and so cannot tell a clean exit from a
/// crash). Mirrors jera's own `EngineState` so a container boot's status/log model
/// is preserved verbatim when it delegates here.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ContainerState {
    /// 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 terminal result of an [`ContainerControl::exec`] into a **running**
/// container: the exec'd process's exit code plus its captured stdout / stderr.
///
/// Shaped like [`RunOutcome`] (a boot-to-completion) so the two read the same, but
/// distinct: a `RunOutcome` is the *container's* lifetime, an `ExecOutcome` is one
/// command run *inside* an already-live container (a `podman exec`). `exit_code` is
/// [`Some`] with the exec'd process status (0 = clean) and [`None`] only if the
/// engine could not report one. A non-zero exit is **not** an error of the call —
/// the command ran, it just failed — so it comes back here, never as an `Err`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ExecOutcome {
    /// The exec'd process's exit code (`Some(0)` = clean), or `None` if the engine
    /// reported no code.
    pub exit_code: Option<i64>,
    /// Captured stdout lines, in order.
    pub stdout: Vec<String>,
    /// Captured stderr lines, in order.
    pub stderr: Vec<String>,
}

/// **Pure** builder of the `podman exec <id> <argv…>` command vector for a running
/// container — factored out so the exec wiring is testable with **no daemon** (the
/// same treatment as [`Engine::create_body`] for the create path). The returned
/// vector is the canonical command form (`["exec", id, argv0, argv1, …]`) the live
/// [`Engine::exec`] drives over bollard's `/exec` REST op (it takes `id` explicitly
/// and `argv = command[2..]`), and it is exactly what a `podman exec` subprocess
/// would run — so a consumer/test can assert the constructed command without a
/// container. `argv` is the command to run inside the container; the caller
/// ([`ContainerControl::exec`]) rejects an empty one.
pub fn exec_argv(id: &str, argv: &[&str]) -> Vec<String> {
    let mut command = Vec::with_capacity(argv.len() + 2);
    command.push("exec".to_string());
    command.push(id.to_string());
    command.extend(argv.iter().map(|a| a.to_string()));
    command
}

/// **Container-specific control** beyond the generic power [`Lifecycle`]: an
/// exit-code-aware [`ContainerState`] and a streamed-log drain, an [`exec`] into a
/// running container, plus a stop that removes the container. This is the seam a job
/// handler (jera) maps onto its own boot status + log model, so the ONE bollard
/// engine lives here in draupnir and jera keeps only job policy — no second engine.
///
/// [`exec`]: ContainerControl::exec
///
/// It is a trait (not inherent methods) so a consumer can inject a mock and prove
/// its delegation wiring with no daemon; [`ContainerBoot`] is the production impl.
pub trait ContainerControl {
    /// The container's exit-code-aware lifecycle state (running / exited-with-code
    /// / gone). A transient inspect hiccup reads [`ContainerState::Gone`].
    fn container_state(&self, machine: &Machine) -> ContainerState;
    /// New streamed log lines since the last drain (the follow-task buffer), as one
    /// **combined** ordered stream. Empty when the backend is not compiled in.
    fn drain_logs(&self, machine: &Machine) -> Vec<String>;
    /// New streamed log lines since the last drain, **split** into `(stdout,
    /// stderr)`. This is the shape a run-to-completion job ([`run_to_completion`])
    /// records so stdout and stderr stay apart (jera's `ContainerOutcome` keeps
    /// them separate). The **default** routes every combined line to `stdout` (a
    /// backend that does not distinguish the streams loses nothing observable);
    /// [`ContainerBoot`] overrides it to preserve the real stdout/stderr tag. It
    /// drains the same buffer as [`drain_logs`](Self::drain_logs) — call one or the
    /// other per tick, not both.
    fn drain_logs_split(&self, machine: &Machine) -> (Vec<String>, Vec<String>) {
        (self.drain_logs(machine), Vec::new())
    }
    /// Stop + remove the container (idempotent, best-effort).
    fn stop(&self, machine: &Machine);

    /// **Exec `argv` inside the already-running container `machine`** — the
    /// `podman exec <id> <argv…>` analogue over the ONE OCI engine, returning the
    /// exec'd process's [`ExecOutcome`] (exit code + captured stdout/stderr). This
    /// is the control primitive a **live start/stop/exec lifecycle** needs: after a
    /// detached [`Boot::boot`] hands back a long-lived [`Machine`], a surviving
    /// handle can run commands *into* it (health probe, live reconfigure, drain)
    /// without tearing it down — the piece [`run_to_completion`] (which owns the
    /// whole container lifetime) cannot express.
    ///
    /// **Container-only** (guarded parity with the container-only `net`/`cmd`/`ports`
    /// `validate()` checks): a non-container [`Machine`] (a KVM guest / Redfish node)
    /// has nothing to exec into and is rejected with an [`Error::Spec`] *before* the
    /// engine is touched, and an **empty `argv`** (nothing to run) is likewise
    /// rejected. This provided method does the guards + assembles the command via
    /// [`exec_argv`] and then delegates the live run to
    /// [`exec_command`](Self::exec_command); a backend without an OCI engine (a mock)
    /// keeps the default `exec_command` and so this whole path is exercised with no
    /// daemon. A non-zero exit of the exec'd process is **not** an `Err` — it is the
    /// [`ExecOutcome::exit_code`]; only a guard failure or an engine/connect error
    /// returns `Err`.
    fn exec(&self, machine: &Machine, argv: &[&str]) -> Result<ExecOutcome> {
        if machine.backend != crate::Backend::Container {
            return Err(Error::Spec(format!(
                "exec is container-only; a {:?} machine has no container to exec into",
                machine.backend
            )));
        }
        if argv.is_empty() {
            return Err(Error::Spec(
                "a container exec needs a non-empty argv (nothing to run)".into(),
            ));
        }
        let command = exec_argv(&machine.id, argv);
        self.exec_command(&command)
    }

    /// The backend hook [`exec`](Self::exec) delegates the *live run* to, once the
    /// container-only + non-empty-argv guards have passed and the canonical command
    /// (`["exec", id, argv…]`, from [`exec_argv`]) is assembled. **Default**:
    /// [`Error::Unsupported`] — an engine-less backend (a mock in a unit test)
    /// cannot exec, but the guard + command-assembly path in [`exec`](Self::exec)
    /// is still fully exercised against it. [`ContainerBoot`] overrides it to drive
    /// bollard's `/exec` REST op (`id = command[1]`, `argv = command[2..]`). A
    /// consumer normally calls [`exec`](Self::exec), not this.
    fn exec_command(&self, command: &[String]) -> Result<ExecOutcome> {
        let _ = command;
        Err(Error::Unsupported(
            "this backend cannot exec into a container (needs the OCI engine)".into(),
        ))
    }
}

impl ContainerControl for ContainerBoot {
    fn container_state(&self, machine: &Machine) -> ContainerState {
        #[cfg(feature = "backend-oci")]
        {
            match self.engine() {
                Ok(e) => e.container_state(&machine.id),
                Err(_) => ContainerState::Gone,
            }
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            ContainerState::Gone
        }
    }

    fn drain_logs(&self, machine: &Machine) -> Vec<String> {
        #[cfg(feature = "backend-oci")]
        {
            match self.engine() {
                Ok(e) => e.drain_logs(&machine.id),
                Err(_) => Vec::new(),
            }
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            Vec::new()
        }
    }

    fn drain_logs_split(&self, machine: &Machine) -> (Vec<String>, Vec<String>) {
        #[cfg(feature = "backend-oci")]
        {
            match self.engine() {
                Ok(e) => e.drain_logs_split(&machine.id),
                Err(_) => (Vec::new(), Vec::new()),
            }
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            (Vec::new(), Vec::new())
        }
    }

    fn stop(&self, machine: &Machine) {
        #[cfg(feature = "backend-oci")]
        {
            if let Ok(e) = self.engine() {
                e.stop(&machine.id);
            }
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
        }
    }

    fn exec_command(&self, command: &[String]) -> Result<ExecOutcome> {
        // The guards + assembly ran in the provided `exec`; `command` is the
        // canonical `["exec", <id>, <argv…>]` (non-empty argv => len >= 3). Drive
        // bollard's `/exec` REST op against `id` with `argv = command[2..]`.
        #[cfg(feature = "backend-oci")]
        {
            let id = &command[1];
            let argv: Vec<String> = command[2..].to_vec();
            self.engine()?.exec(id, &argv)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = command;
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }
}

// ---------------------------------------------------------------------------
// run_to_completion — the single-call run-to-completion container seam.
// ---------------------------------------------------------------------------

/// The terminal result of a [`run_to_completion`] job: the container's exit code
/// plus its captured logs, split into stdout / stderr.
///
/// `exit_code` is [`Some`] with the process exit status (0 = clean) when the
/// container exited on its own, and [`None`] when it vanished before a code could be
/// read (removed out from under us, killed by signal with no reported status). This
/// mirrors jera's `ContainerOutcome` field-for-field, so `nornir::jobs::run_container`
/// can repoint onto this seam and delete its duplicate `BollardEngine`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RunOutcome {
    /// The container's exit code (`Some(0)` = clean), or `None` if it went away
    /// before a code was observed.
    pub exit_code: Option<i64>,
    /// Captured stdout lines, in order.
    pub stdout: Vec<String>,
    /// Captured stderr lines, in order.
    pub stderr: Vec<String>,
}

/// Knobs for [`run_to_completion`]: how long to wait for the container to exit and
/// how often to poll its state. [`Default`] waits **indefinitely** (parity with
/// jera's blocking `wait_container`) and polls every 200 ms.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunOptions {
    /// Overall budget before giving up. `None` = wait forever for the container to
    /// exit (jera parity). `Some(d)` stops + removes the container and returns an
    /// [`Error::Backend`] if it has not exited within `d`.
    pub timeout: Option<Duration>,
    /// How often the container state is polled (and logs drained) while it runs.
    pub poll_interval: Duration,
}

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

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

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

/// **Run a container to completion in one call** — start it, wait for it to exit,
/// collect its logs, and return an exit-code-aware [`RunOutcome`]. This is the
/// missing seam that lets jera's run-to-completion `run_container` (and, above it,
/// `nornir::jobs::run_container`) route through draupnir's **one** OCI engine
/// instead of jera's duplicate `BollardEngine` — the run-to-completion analogue of
/// how a VM/container *boot* already delegates through [`Boot`].
///
/// It is written against the always-compiled [`Boot`] + [`ContainerControl`] seam,
/// generic over the backend, so it drives a live [`ContainerBoot`] in production
/// **and** a mock in a unit test with no daemon. The flow is exactly jera's
/// run_container: [`boot`](Boot::boot) (create + start) → poll
/// [`container_state`](ContainerControl::container_state) draining
/// [`drain_logs_split`](ContainerControl::drain_logs_split) each tick until the
/// container reports [`ContainerState::Exited`] (or [`ContainerState::Gone`]) → a
/// final drain → [`stop`](ContainerControl::stop) (remove). A non-zero exit is NOT
/// an `Err` — it comes back in [`RunOutcome::exit_code`] (the *container* failed,
/// the *call* succeeded); only a boot/connect failure or a `timeout` returns `Err`.
///
/// Note: on a live engine the log follow-task flushes asynchronously, so the final
/// drain after exit is what captures the tail — a chatty container's last lines
/// arrive on the buffer as the stream closes.
pub fn run_to_completion<B>(backend: &B, spec: &BootSpec, opts: &RunOptions) -> Result<RunOutcome>
where
    B: Boot + ContainerControl,
{
    let machine = backend.boot(spec)?;
    drive_to_completion(backend, &machine, opts)
}

/// The post-boot run-to-completion loop — drain logs → poll state → final drain →
/// stop — factored out of [`run_to_completion`] so it AND the bind-mount variant
/// ([`ContainerBoot::run_to_completion_with_binds`]) share one implementation
/// (single-source: the drive loop lives once). Takes an already-booted `machine`.
pub fn drive_to_completion<B>(backend: &B, machine: &Machine, opts: &RunOptions) -> Result<RunOutcome>
where
    B: ContainerControl,
{
    let mut stdout: Vec<String> = Vec::new();
    let mut stderr: Vec<String> = Vec::new();
    let deadline = opts.timeout.map(|t| Instant::now() + t);

    let exit_code = loop {
        // Drain incrementally so a long-running, chatty container doesn't buffer
        // unboundedly before we ever read it.
        let (mut out, mut err) = backend.drain_logs_split(machine);
        stdout.append(&mut out);
        stderr.append(&mut err);

        match backend.container_state(machine) {
            ContainerState::Exited(code) => break Some(code),
            ContainerState::Gone => break None,
            ContainerState::Running => {}
        }

        if let Some(dl) = deadline {
            if Instant::now() >= dl {
                backend.stop(machine);
                return Err(Error::Backend(format!(
                    "container `{}` did not run to completion within {:?}",
                    machine.id,
                    opts.timeout.unwrap()
                )));
            }
        }
        std::thread::sleep(opts.poll_interval);
    };

    // Final drain: catch the lines emitted between the last poll and exit.
    let (mut out, mut err) = backend.drain_logs_split(machine);
    stdout.append(&mut out);
    stderr.append(&mut err);

    // Remove the container (idempotent) now that we have its code + logs.
    backend.stop(machine);

    Ok(RunOutcome { exit_code, stdout, stderr })
}

impl ContainerBoot {
    /// Run `spec` to completion on the live OCI engine — the ergonomic production
    /// entry point that forwards to the generic [`run_to_completion`] with `self`.
    /// Requires the `backend-oci` feature (else an honest [`Error::Unsupported`]).
    pub fn run_to_completion(&self, spec: &BootSpec, opts: &RunOptions) -> Result<RunOutcome> {
        #[cfg(feature = "backend-oci")]
        {
            run_to_completion(self, spec, opts)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (spec, opts);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }

    /// Boot `spec` with additional host **bind mounts** (`host:container[:opts]`,
    /// `-v` semantics) — the mount-aware twin of [`Boot::boot`]. `binds` empty ⇒
    /// identical to `boot`. Requires `backend-oci`.
    pub fn boot_with_binds(&self, spec: &BootSpec, binds: &[String]) -> Result<Machine> {
        spec.validate()?;
        let image = self.image_ref(spec)?;
        #[cfg(feature = "backend-oci")]
        {
            let name = Self::container_name(spec);
            let env: Vec<String> = spec.env.iter().map(|(k, v)| format!("{k}={v}")).collect();
            self.engine()?.create_and_start_with_binds_and_net(
                image, &name, &env, &spec.cmd, &spec.ports, &spec.port_maps, binds, spec.net.oci_value(),
                spec.cpus, spec.mem_limit_mb,
            )?;
            Ok(Machine::started(name, spec))
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (image, binds);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }

    /// [`run_to_completion`](Self::run_to_completion) with host **bind mounts** —
    /// lets a build-in-a-container job (WiX/MSI under Wine, `pack`) mount its input/
    /// output dirs and run through this ONE OCI engine, replacing a
    /// `Command::new("podman") -v …` shell twin. `binds` empty ⇒ identical to
    /// `run_to_completion`. Requires `backend-oci`.
    pub fn run_to_completion_with_binds(
        &self,
        spec: &BootSpec,
        opts: &RunOptions,
        binds: &[String],
    ) -> Result<RunOutcome> {
        #[cfg(feature = "backend-oci")]
        {
            let machine = self.boot_with_binds(spec, binds)?;
            drive_to_completion(self, &machine, opts)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (spec, opts, binds);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }

    /// **Build an OCI image** from a build context directory + a Containerfile
    /// (`podman build` over the one engine). `context_dir` is tarred in-memory and
    /// sent to the daemon's `/build`; `containerfile` is its path relative to the
    /// context; `tag` names the result. Returns `tag` on success. Kills the
    /// `Command::new("podman") build …` shell twin (Skidbladnir `pack.rs`). Requires
    /// `backend-oci`; honest [`Error::Backend`] when the socket is unreachable, never
    /// a fake image.
    pub fn build_image(&self, context_dir: &Path, containerfile: &str, tag: &str) -> Result<String> {
        #[cfg(feature = "backend-oci")]
        {
            self.engine()?.build_image(context_dir, containerfile, tag)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (context_dir, containerfile, tag);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }

    /// **Extract a path from an image** to the host — the `podman create` + `podman
    /// cp <container>:<path> <host>` twin over the one engine. Creates a throwaway
    /// (un-started) container from `image`, downloads a tar of `container_path` via
    /// the daemon, and unpacks it into `host_dest`, then removes the container. The
    /// unpacked tree is rooted at the basename of `container_path` (same layout
    /// `podman cp` yields). Requires `backend-oci`; honest [`Error::Backend`] when the
    /// socket is unreachable, never a partial fake.
    pub fn extract_path(&self, image: &str, container_path: &str, host_dest: &Path) -> Result<()> {
        #[cfg(feature = "backend-oci")]
        {
            self.engine()?.extract_path(image, container_path, host_dest)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (image, container_path, host_dest);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }
}

/// **Pure** in-memory tar of an OCI build-context directory — factored out of
/// [`Engine::build_image`] so the context packing is testable with no daemon
/// (build a temp dir → tar → read back the entries). Every file under `context_dir`
/// is added at its relative path (archive root `.`), the shape the daemon's `/build`
/// endpoint expects. Feature-gated because it rides the `tar` crate that comes with
/// `backend-oci`.
#[cfg(feature = "backend-oci")]
fn context_tar(context_dir: &Path) -> Result<Vec<u8>> {
    let mut buf: Vec<u8> = Vec::new();
    {
        let mut builder = tar::Builder::new(&mut buf);
        builder
            .append_dir_all(".", context_dir)
            .map_err(|e| Error::Backend(format!("tar build context {}: {e}", context_dir.display())))?;
        builder
            .finish()
            .map_err(|e| Error::Backend(format!("finish build-context tar: {e}")))?;
    }
    Ok(buf)
}

// ---------------------------------------------------------------------------
// Engine — the real podman/Docker REST engine (feature `backend-oci`).
// ---------------------------------------------------------------------------

/// A shared, append-only log buffer the follow tasks push into and the caller
/// drains. Each entry is `(is_stderr, line)` so a drain can either flatten to the
/// combined ordered stream ([`Engine::drain_logs`]) or split it back into
/// stdout/stderr ([`Engine::drain_logs_split`]) — jera's `run_container` keeps the
/// two apart, so preserving the tag makes that repoint lossless.
#[cfg(feature = "backend-oci")]
type LogBuf = Arc<Mutex<Vec<(bool, String)>>>;

/// The live bollard engine: a `Docker` handle + a dedicated tokio runtime that
/// drives its async API from draupnir's synchronous [`Boot`]/[`Lifecycle`] seam.
#[cfg(feature = "backend-oci")]
struct Engine {
    docker: bollard::Docker,
    rt: tokio::runtime::Runtime,
    /// Per-container streamed-log buffers, filled by the follow tasks, drained by
    /// [`Engine::drain_logs`] (the API-streamed equivalent of reader threads).
    logs: Mutex<std::collections::HashMap<String, LogBuf>>,
}

#[cfg(feature = "backend-oci")]
impl Engine {
    /// Resolve the podman/Docker API socket URL: honour `DOCKER_HOST`, else the
    /// rootless user socket under `XDG_RUNTIME_DIR` (parity with jera).
    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 over the API socket. Does NOT try to start a daemon (zero-shell) —
    /// returns a clear [`Error::Backend`] if the socket is absent/unreachable.
    fn connect() -> Result<Self> {
        let url = Self::socket_url();
        let path = url.strip_prefix("unix://").unwrap_or(&url);
        if url.starts_with("unix://") && !std::path::Path::new(path).exists() {
            return Err(Error::Backend(format!(
                "podman/Docker API socket not found at {path} (enable with \
                 `systemctl --user enable --now podman.socket`, or point DOCKER_HOST at a running socket)"
            )));
        }
        let docker = bollard::Docker::connect_with_unix(&url, 120, bollard::API_DEFAULT_VERSION)
            .map_err(|e| Error::Backend(format!("connect container socket {url}: {e}")))?;
        let rt = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .map_err(|e| Error::Backend(format!("build tokio runtime for bollard: {e}")))?;
        Ok(Engine { docker, rt, logs: Mutex::new(std::collections::HashMap::new()) })
    }

    /// **Pure** builder of the `ContainerCreateBody` for `image` — factored out so
    /// the env/cmd/exposed-port wiring is testable with no daemon. A non-empty `cmd`
    /// overrides the image entrypoint; `env` is carried as `KEY=VALUE`; each port is
    /// both exposed AND published to the same host port via a `HostConfig` binding.
    /// Byte-for-byte parity with jera's former `BollardEngine::create_body`, so the
    /// one engine here produces the same container jera did.
    // Kept as the documented jera-parity entry (bind-free); the engine now routes
    // through `create_body_with_binds`, so this is a thin delegate + public API.
    #[allow(dead_code)]
    pub fn create_body(image: &str, env: &[String], cmd: &[String], ports: &[u16]) -> bollard::models::ContainerCreateBody {
        Self::create_body_with_binds(image, env, cmd, ports, &[])
    }

    /// Like [`create_body`](Self::create_body) but also attaches host **bind
    /// mounts** — `host:container[:opts]` strings (`-v` semantics) — to the
    /// `HostConfig.binds`. With `binds` empty this is **byte-for-byte identical**
    /// to [`create_body`](Self::create_body) (the mount field stays `None`), so
    /// every existing caller is unchanged; a non-empty `binds` lets a build-in-a-
    /// container job (WiX/MSI, pack) route through this one OCI engine instead of a
    /// `Command::new("podman") -v …` shell twin.
    pub fn create_body_with_binds(
        image: &str,
        env: &[String],
        cmd: &[String],
        ports: &[u16],
        binds: &[String],
    ) -> bollard::models::ContainerCreateBody {
        Self::create_body_with_binds_and_net(image, env, cmd, ports, binds, None)
    }

    /// Like [`create_body_with_binds`](Self::create_body_with_binds) but also sets
    /// the container **network mode** on `HostConfig.network_mode` — `Some("none")`
    /// is the airgap `--network none` (loopback-only, no egress). With `network_mode`
    /// `None` this is **byte-for-byte identical** to `create_body_with_binds` (the
    /// field stays unset, and no `HostConfig` is minted when ports+binds are empty
    /// too), so every existing caller is unchanged. This is the **load-bearing wire**
    /// for Skidbladnir's airgap container route: jera renders `--network none` and
    /// threads its `oci_value()` (`"none"`) here so it reaches the live podman run.
    pub fn create_body_with_binds_and_net(
        image: &str,
        env: &[String],
        cmd: &[String],
        ports: &[u16],
        binds: &[String],
        network_mode: Option<&str>,
    ) -> bollard::models::ContainerCreateBody {
        // The `ports`-only (host==container) form: no distinct maps, no res caps.
        // Byte-identical to before — it now flows through the one pair-aware core.
        Self::create_body_full(image, env, cmd, ports, &[], binds, network_mode, None, None)
    }

    /// The **one** create-body builder every other `create_body*` entry routes
    /// through: it publishes both the `host == container` single-port form
    /// ([`ports`]) **and** the distinct [`PortMap`] `host:container` form
    /// ([`port_maps`]), attaches host `binds`, sets the `network_mode`, and applies
    /// the `cpus`/`mem_mb` resource caps — every knob optional and each defaulting
    /// to the byte-identical no-op (empty ports/maps/binds + `None` net/caps ⇒ no
    /// `HostConfig` is minted, exactly as a bare image spec always rendered).
    ///
    /// A `PortMap { host, container }` renders `exposed_ports[container/tcp]` +
    /// `port_bindings[container/tcp] = host` — i.e. podman `-p host:container`, so
    /// a service on a fixed in-container port (FalkorDB's `6379`) is reachable on a
    /// different host port (a per-zone offset `6380`/`6381`). This is the load-
    /// bearing wire for korp's per-zone infra: the demo zone (`6379:6379`) already
    /// worked via `ports`; test/prod (`6380:6379`, `6381:6379`) need this map.
    ///
    /// [`ports`]: BootSpec::ports
    /// [`port_maps`]: BootSpec::port_maps
    #[allow(clippy::too_many_arguments)]
    pub fn create_body_full(
        image: &str,
        env: &[String],
        cmd: &[String],
        ports: &[u16],
        port_maps: &[crate::PortMap],
        binds: &[String],
        network_mode: Option<&str>,
        cpus: Option<f64>,
        mem_mb: Option<u32>,
    ) -> bollard::models::ContainerCreateBody {
        use bollard::models::{ContainerCreateBody, HostConfig, PortBinding};
        use std::collections::HashMap;

        let mut exposed: Vec<String> = Vec::new();
        let mut bindings: HashMap<String, Option<Vec<PortBinding>>> = HashMap::new();
        // Helper: expose `container/tcp` and bind it to `host` on 0.0.0.0. Kept as
        // a closure so the two forms (host==container `ports` and distinct
        // `port_maps`) render identically — a `ports` entry `p` is just the
        // `host == container == p` case, so `[6379]` and `PortMap::same(6379)`
        // produce the same wire.
        let mut publish = |host: u16, container: u16| {
            let key = format!("{container}/tcp");
            if !exposed.contains(&key) {
                exposed.push(key.clone());
            }
            bindings.insert(
                key,
                Some(vec![PortBinding {
                    host_ip: Some("0.0.0.0".to_string()),
                    host_port: Some(host.to_string()),
                }]),
            );
        };
        for &p in ports {
            publish(p, p);
        }
        for pm in port_maps {
            publish(pm.host, pm.container);
        }

        // Resource caps: MiB → bytes, cores → nano-cpus. Filter a non-positive
        // value so an unconstrained (`None`) or mistakenly-zero cap never mints a
        // `HostConfig` and stays byte-identical to the cap-less create body.
        let nano_cpus = cpus.filter(|c| *c > 0.0).map(|c| (c * 1e9) as i64);
        let memory = mem_mb.filter(|m| *m > 0).map(|m| i64::from(m) * 1024 * 1024);

        // Mint a `HostConfig` when there is anything to carry — ports, binds, a
        // network mode, OR a resource cap. When all are absent, `host_config`
        // stays `None`, exactly as a bare image spec always rendered (byte-parity).
        let host_config = if bindings.is_empty()
            && binds.is_empty()
            && network_mode.is_none()
            && nano_cpus.is_none()
            && memory.is_none()
        {
            None
        } else {
            Some(HostConfig {
                port_bindings: if bindings.is_empty() { None } else { Some(bindings) },
                binds: if binds.is_empty() { None } else { Some(binds.to_vec()) },
                network_mode: network_mode.map(str::to_string),
                nano_cpus,
                memory,
                ..Default::default()
            })
        };
        ContainerCreateBody {
            image: Some(image.to_string()),
            cmd: if cmd.is_empty() { None } else { Some(cmd.to_vec()) },
            env: if env.is_empty() { None } else { Some(env.to_vec()) },
            exposed_ports: if exposed.is_empty() { None } else { Some(exposed) },
            host_config,
            ..Default::default()
        }
    }

    /// Like [`create_body_with_binds_and_net`](Self::create_body_with_binds_and_net)
    /// but also sets the container **resource caps** on `HostConfig`: `cpus` →
    /// `nano_cpus` (podman `--cpus`, `n * 1e9`) and `mem_mb` → `memory` (podman
    /// `--memory`, MiB → bytes). Both **default to `None` = unconstrained**, which is
    /// the deliberate hot-infra default — a container with no `--cpus` sees **all**
    /// host cores (FalkorDB's OpenMP pool, a Spark executor must never be throttled to
    /// one core). With `cpus`+`mem_mb` both `None` this is **byte-for-byte identical**
    /// to `create_body_with_binds_and_net` (no `HostConfig` is minted purely for an
    /// unset cap), so every existing caller is unchanged. A non-positive cap is dropped
    /// here (guarded at [`BootSpec::validate`]) rather than passed to the daemon.
    // Now a thin `port_maps`-free delegate to `create_body_full`; kept as a public
    // parity entry (and used by the resource-cap unit test), so it is dead in a
    // non-test lib build — same treatment as the `create_body` parity delegate.
    #[allow(dead_code)]
    #[allow(clippy::too_many_arguments)]
    pub fn create_body_with_res(
        image: &str,
        env: &[String],
        cmd: &[String],
        ports: &[u16],
        binds: &[String],
        network_mode: Option<&str>,
        cpus: Option<f64>,
        mem_mb: Option<u32>,
    ) -> bollard::models::ContainerCreateBody {
        // The `ports`-only (host==container) form with resource caps: no distinct
        // maps. Delegates to the one pair-aware core with empty `port_maps`, so it
        // is byte-identical to the former standalone implementation.
        Self::create_body_full(image, env, cmd, ports, &[], binds, network_mode, cpus, mem_mb)
    }

    /// Pull `image` if not present, then create + start it as a detached container
    /// named `name` with `env` (`KEY=VALUE`), an optional `cmd` entrypoint override,
    /// and published `ports`. A background follow task streams the container's logs
    /// into a shared buffer this container's [`drain_logs`](Self::drain_logs) drains.
    // The boot paths now route through the net-aware
    // `create_and_start_with_binds_and_net` (so `spec.net` reaches the live run), so
    // these two thin delegates are the kept net-less/bind-less parity entries — same
    // treatment as the public `create_body`/`create_body_with_binds` delegate pair.
    #[allow(dead_code)]
    fn create_and_start(&self, image: &str, name: &str, env: &[String], cmd: &[String], ports: &[u16]) -> Result<()> {
        self.create_and_start_with_binds(image, name, env, cmd, ports, &[])
    }

    /// [`create_and_start`](Self::create_and_start) plus host **bind mounts**
    /// (`host:container[:opts]`, `-v` semantics). `binds` empty ⇒ identical to
    /// `create_and_start`.
    #[allow(dead_code)]
    fn create_and_start_with_binds(&self, image: &str, name: &str, env: &[String], cmd: &[String], ports: &[u16], binds: &[String]) -> Result<()> {
        self.create_and_start_with_binds_and_net(image, name, env, cmd, ports, &[], binds, None, None, None)
    }

    /// [`create_and_start_with_binds`](Self::create_and_start_with_binds) plus the
    /// container **network mode** (`Some("none")` ⇒ `--network none`, the airgap
    /// case). `network_mode` `None` ⇒ byte-identical to `create_and_start_with_binds`
    /// (the live create body is unchanged). This is where the net choice actually
    /// reaches the live podman `create_container` call.
    // The engine's create-body carries this many independent knobs (image/name/env/
    // cmd/ports/binds/net); grouping them into a struct would just shadow the
    // `ContainerCreateBody` fields, so the flat arg list is the honest shape here.
    #[allow(clippy::too_many_arguments)]
    fn create_and_start_with_binds_and_net(&self, image: &str, name: &str, env: &[String], cmd: &[String], ports: &[u16], port_maps: &[crate::PortMap], binds: &[String], network_mode: Option<&str>, cpus: Option<f64>, mem_mb: Option<u32>) -> Result<()> {
        use bollard::query_parameters::{
            CreateContainerOptionsBuilder, CreateImageOptionsBuilder, RemoveContainerOptionsBuilder,
            StartContainerOptions,
        };
        use futures::StreamExt;

        let docker = &self.docker;
        self.rt.block_on(async {
            // Drop 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 if it is not already local.
            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| Error::Backend(format!("pull image {image}: {e}")))?;
                }
            }
            let body = Self::create_body_full(image, env, cmd, ports, port_maps, binds, network_mode, cpus, mem_mb);
            docker
                .create_container(Some(CreateContainerOptionsBuilder::new().name(name).build()), body)
                .await
                .map_err(|e| Error::Backend(format!("create container {name}: {e}")))?;
            docker
                .start_container(name, None::<StartContainerOptions>)
                .await
                .map_err(|e| Error::Backend(format!("start container {name}: {e}")))?;
            Ok::<(), Error>(())
        })?;

        // Wire a background log-follow task into a shared buffer this container's
        // `drain_logs` drains — the API-streamed equivalent of 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 is_err = matches!(out, LogOutput::StdErr { .. });
                        let line = LogOutput::to_string(&out);
                        let line = line.trim_end_matches(['\n', '\r']).to_string();
                        if !line.is_empty() {
                            buf.lock().unwrap().push((is_err, line));
                        }
                    }
                    Err(_) => break,
                }
            }
        });
        Ok(())
    }

    /// Build an OCI image from `context_dir` + `containerfile`, tagging it `tag`.
    /// Tars the context in-memory ([`context_tar`]) and streams it to the daemon's
    /// `/build`; a `BuildInfo` carrying an `error` fails the call. Returns `tag`.
    fn build_image(&self, context_dir: &Path, containerfile: &str, tag: &str) -> Result<String> {
        use bollard::body_full;
        use bollard::query_parameters::BuildImageOptionsBuilder;
        use futures::StreamExt;

        let tar = context_tar(context_dir)?;
        let docker = &self.docker;
        self.rt.block_on(async {
            let opts = BuildImageOptionsBuilder::default()
                .dockerfile(containerfile)
                .t(tag)
                .rm(true)
                .build();
            let mut stream = docker.build_image(opts, None, Some(body_full(tar.into())));
            while let Some(item) = stream.next().await {
                let info = item.map_err(|e| Error::Backend(format!("build image {tag}: {e}")))?;
                if let Some(detail) = info.error_detail {
                    return Err(Error::Backend(format!(
                        "build image {tag}: {}",
                        detail.message.unwrap_or_default()
                    )));
                }
            }
            Ok::<(), Error>(())
        })?;
        Ok(tag.to_string())
    }

    /// Extract `container_path` from `image` to `host_dest`: create a throwaway
    /// (un-started) container, download a tar of the path from the daemon, remove the
    /// container, and unpack the tar into `host_dest`. The `podman create`+`cp` twin.
    fn extract_path(&self, image: &str, container_path: &str, host_dest: &Path) -> Result<()> {
        use bollard::models::ContainerCreateBody;
        use bollard::query_parameters::{
            CreateContainerOptionsBuilder, DownloadFromContainerOptionsBuilder,
            RemoveContainerOptionsBuilder,
        };
        use futures::StreamExt;

        let docker = &self.docker;
        // A throwaway container name from the (sanitised) image + pid; force-removed
        // first so a stale one never blocks the extract.
        let safe: String = image
            .chars()
            .map(|c| if c.is_ascii_alphanumeric() { c } else { '-' })
            .collect();
        let name = format!("draupnir-extract-{safe}-{}", std::process::id());

        let tar_bytes: Vec<u8> = self.rt.block_on(async {
            let _ = docker
                .remove_container(&name, Some(RemoveContainerOptionsBuilder::new().force(true).build()))
                .await;
            let body = ContainerCreateBody { image: Some(image.to_string()), ..Default::default() };
            docker
                .create_container(Some(CreateContainerOptionsBuilder::new().name(name.as_str()).build()), body)
                .await
                .map_err(|e| Error::Backend(format!("create extract container from {image}: {e}")))?;
            let opts = DownloadFromContainerOptionsBuilder::default().path(container_path).build();
            let mut stream = docker.download_from_container(&name, Some(opts));
            let mut buf: Vec<u8> = Vec::new();
            let mut dl_err: Option<Error> = None;
            while let Some(item) = stream.next().await {
                match item {
                    Ok(chunk) => buf.extend_from_slice(&chunk),
                    Err(e) => {
                        dl_err = Some(Error::Backend(format!(
                            "download {container_path} from {image}: {e}"
                        )));
                        break;
                    }
                }
            }
            // Always remove the throwaway container (best-effort), even on error.
            let _ = docker
                .remove_container(&name, Some(RemoveContainerOptionsBuilder::new().force(true).build()))
                .await;
            match dl_err {
                Some(e) => Err(e),
                None => Ok(buf),
            }
        })?;

        // Unpack the downloaded tar into the host destination (sync fs).
        std::fs::create_dir_all(host_dest)
            .map_err(|e| Error::Backend(format!("create extract dest {}: {e}", host_dest.display())))?;
        tar::Archive::new(std::io::Cursor::new(tar_bytes))
            .unpack(host_dest)
            .map_err(|e| Error::Backend(format!("unpack extracted tar into {}: {e}", host_dest.display())))?;
        Ok(())
    }

    /// The container's exit-code-aware state (running / exited-with-code / gone) —
    /// what [`ContainerControl::container_state`] surfaces. A 404 / transient inspect
    /// error reads [`ContainerState::Gone`]; a live boot's next poll retries.
    fn container_state(&self, name: &str) -> ContainerState {
        use bollard::query_parameters::InspectContainerOptions;
        self.rt.block_on(async {
            match self.docker.inspect_container(name, None::<InspectContainerOptions>).await {
                Ok(info) => {
                    let state = info.state;
                    let running = state.as_ref().and_then(|s| s.running).unwrap_or(false);
                    if running {
                        ContainerState::Running
                    } else {
                        ContainerState::Exited(state.and_then(|s| s.exit_code).unwrap_or(0))
                    }
                }
                Err(_) => ContainerState::Gone,
            }
        })
    }

    /// Drain the streamed log lines accumulated for `name` since the last drain,
    /// as one **combined** ordered stream (stdout + stderr interleaved as emitted).
    fn drain_logs(&self, name: &str) -> Vec<String> {
        match self.logs.lock().unwrap().get(name) {
            Some(buf) => std::mem::take(&mut *buf.lock().unwrap())
                .into_iter()
                .map(|(_, line)| line)
                .collect(),
            None => Vec::new(),
        }
    }

    /// Drain the streamed log lines for `name`, **split** into `(stdout, stderr)` —
    /// the shape jera's run-to-completion `ContainerOutcome` keeps apart. Order
    /// within each stream is preserved. Empties the same buffer `drain_logs` reads.
    fn drain_logs_split(&self, name: &str) -> (Vec<String>, Vec<String>) {
        match self.logs.lock().unwrap().get(name) {
            Some(buf) => {
                let mut out = Vec::new();
                let mut err = Vec::new();
                for (is_err, line) in std::mem::take(&mut *buf.lock().unwrap()) {
                    if is_err {
                        err.push(line);
                    } else {
                        out.push(line);
                    }
                }
                (out, err)
            }
            None => (Vec::new(), Vec::new()),
        }
    }

    /// Start a previously-created (stopped) container.
    fn start(&self, name: &str) -> Result<()> {
        use bollard::query_parameters::StartContainerOptions;
        self.rt.block_on(async {
            self.docker
                .start_container(name, None::<StartContainerOptions>)
                .await
                .map_err(|e| Error::Backend(format!("start container {name}: {e}")))
        })
    }

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

    /// **Exec `argv` inside the running container `name`** — the live drive behind
    /// [`ContainerControl::exec`], over podman's `/exec` REST op (`create_exec` →
    /// `start_exec` → `inspect_exec`), never a subprocess (zero-shell). Attaches
    /// stdout+stderr, drains them split (same tagging as the log-follow task), and
    /// reads the process's exit code back from `inspect_exec`. A non-zero exit is
    /// surfaced in [`ExecOutcome::exit_code`], not as an `Err`; only a create/start/
    /// inspect/transport failure returns `Err`.
    fn exec(&self, name: &str, argv: &[String]) -> Result<ExecOutcome> {
        use bollard::container::LogOutput;
        use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults};
        use futures::StreamExt;

        let docker = &self.docker;
        self.rt.block_on(async {
            let config = CreateExecOptions::<String> {
                cmd: Some(argv.to_vec()),
                attach_stdout: Some(true),
                attach_stderr: Some(true),
                ..Default::default()
            };
            let created = docker
                .create_exec(name, config)
                .await
                .map_err(|e| Error::Backend(format!("create exec in {name}: {e}")))?;

            let mut stdout: Vec<String> = Vec::new();
            let mut stderr: Vec<String> = Vec::new();
            match docker
                .start_exec(&created.id, None::<StartExecOptions>)
                .await
                .map_err(|e| Error::Backend(format!("start exec in {name}: {e}")))?
            {
                StartExecResults::Attached { mut output, .. } => {
                    while let Some(item) = output.next().await {
                        match item {
                            Ok(out) => {
                                let is_err = matches!(out, LogOutput::StdErr { .. });
                                let line = LogOutput::to_string(&out);
                                let line = line.trim_end_matches(['\n', '\r']).to_string();
                                if !line.is_empty() {
                                    if is_err {
                                        stderr.push(line);
                                    } else {
                                        stdout.push(line);
                                    }
                                }
                            }
                            Err(e) => {
                                return Err(Error::Backend(format!(
                                    "read exec output in {name}: {e}"
                                )))
                            }
                        }
                    }
                }
                StartExecResults::Detached => {}
            }

            let inspect = docker
                .inspect_exec(&created.id)
                .await
                .map_err(|e| Error::Backend(format!("inspect exec in {name}: {e}")))?;
            Ok(ExecOutcome { exit_code: inspect.exit_code, stdout, stderr })
        })
    }

    /// The container's power state: `On` while running, else `Off` (a gone/unknown
    /// container reads `Off`).
    fn power_state(&self, name: &str) -> PowerState {
        use bollard::query_parameters::InspectContainerOptions;
        self.rt.block_on(async {
            match self
                .docker
                .inspect_container(name, None::<InspectContainerOptions>)
                .await
            {
                Ok(info) => {
                    let running = info.state.as_ref().and_then(|s| s.running).unwrap_or(false);
                    if running {
                        PowerState::On
                    } else {
                        PowerState::Off
                    }
                }
                Err(_) => PowerState::Off,
            }
        })
    }

    /// Whether `needle` has appeared in the container's stdout/stderr logs so far —
    /// the log-match readiness probe. Drains the (non-follow) log stream once and
    /// scans it; a container that has produced no output yet simply reads `false`.
    fn log_contains(&self, name: &str, needle: &str) -> Result<bool> {
        use bollard::query_parameters::LogsOptionsBuilder;
        use futures::StreamExt;
        self.rt.block_on(async {
            let opts = LogsOptionsBuilder::new().stdout(true).stderr(true).build();
            let mut stream = self.docker.logs(name, Some(opts));
            let mut buf = String::new();
            while let Some(item) = stream.next().await {
                match item {
                    Ok(chunk) => buf.push_str(&String::from_utf8_lossy(&chunk.into_bytes())),
                    Err(e) => {
                        return Err(Error::Backend(format!("read logs for {name}: {e}")))
                    }
                }
            }
            Ok(buf.contains(needle))
        })
    }
}

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

    #[test]
    fn image_ref_extracts_the_oci_reference() {
        let spec = BootSpec::container("cache", "docker.io/library/redis:7");
        assert_eq!(ContainerBoot::new().image_ref(&spec).unwrap(), "docker.io/library/redis:7");
    }

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

    #[test]
    fn container_name_is_derived_from_the_spec_name() {
        let spec = BootSpec::container("cache", "redis:7");
        assert_eq!(ContainerBoot::container_name(&spec), "draupnir-cache");
    }

    #[test]
    fn wait_ready_returns_a_clear_timeout_error_when_never_ready() {
        // A probe that never reports ready must time out with an Error::Backend
        // naming the instance and the budget — not hang, not fake-succeed.
        let err = poll_until_ready(
            "cache",
            Duration::from_millis(40),
            Duration::from_millis(5),
            || Ok(false),
        )
        .unwrap_err();
        match err {
            Error::Backend(m) => {
                assert!(m.contains("cache"), "names the instance: {m}");
                assert!(m.contains("not ready"), "says it wasn't ready: {m}");
            }
            other => panic!("expected Error::Backend, got {other:?}"),
        }
    }

    #[test]
    fn wait_ready_returns_ok_as_soon_as_the_probe_reports_ready() {
        // Ready on the 3rd poll — proves it polls rather than checking once.
        let mut n = 0;
        let r = poll_until_ready("cache", Duration::from_secs(5), Duration::from_millis(1), || {
            n += 1;
            Ok(n >= 3)
        });
        assert!(r.is_ok());
        assert_eq!(n, 3);
    }

    #[test]
    fn wait_ready_fails_fast_on_a_backend_error() {
        // A backend error from the probe is surfaced, never swallowed as "not ready".
        let r = poll_until_ready("cache", Duration::from_secs(5), Duration::from_millis(1), || {
            Err(Error::Backend("socket vanished".into()))
        });
        assert!(matches!(r, Err(Error::Backend(m)) if m.contains("socket vanished")));
    }

    #[test]
    fn readiness_defaults_to_running() {
        assert_eq!(Readiness::default(), Readiness::Running);
    }

    #[test]
    fn container_boot_carries_cmd_and_ports_through_the_spec() {
        // The consolidated engine must accept cmd/entrypoint override + published
        // ports (jera parity) — proven on the pure BootSpec, no daemon.
        let spec = BootSpec::container("web", "docker.io/library/nginx:alpine")
            .with_cmd(["nginx", "-g", "daemon off;"])
            .with_port(8080)
            .with_port(8443)
            .with_env("TZ", "UTC");
        assert_eq!(spec.cmd, vec!["nginx", "-g", "daemon off;"]);
        assert_eq!(spec.ports, vec![8080, 8443]);
        assert_eq!(spec.env.get("TZ").map(String::as_str), Some("UTC"));
        spec.validate().unwrap();
    }

    /// Without the `backend-oci` engine, the [`ContainerControl`] seam is an honest
    /// no-op: a not-connected container reads `Gone`, drains no logs. (Under the
    /// feature these route to the live engine; that needs a daemon.)
    #[test]
    fn container_control_is_an_honest_noop_without_the_engine() {
        let boot = ContainerBoot::new();
        let m = Machine::started("draupnir-x", &BootSpec::container("x", "redis:7"));
        // With backend-oci the engine can't connect in CI (no socket) → Gone; without
        // it, the compiled-out path is Gone too. Either way: never a fake "Running".
        assert_eq!(boot.container_state(&m), ContainerState::Gone);
        assert!(boot.drain_logs(&m).is_empty());
        boot.stop(&m); // must not panic
    }

    /// The pure `create_body` builder folds env/cmd/ports into a `ContainerCreateBody`
    /// — cmd carried, each port both exposed AND bound. Byte-for-byte the shape jera
    /// produced, so moving the engine here changes no observable container. No daemon.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn create_body_carries_env_cmd_and_port_bindings() {
        let body = Engine::create_body(
            "app:1",
            &["A=1".to_string(), "B=2".to_string()],
            &["/bin/app".to_string(), "--serve".to_string()],
            &[8080],
        );
        assert_eq!(body.image.as_deref(), Some("app:1"));
        assert_eq!(body.cmd, Some(vec!["/bin/app".to_string(), "--serve".to_string()]));
        assert_eq!(body.env, Some(vec!["A=1".to_string(), "B=2".to_string()]));
        let exposed = body.exposed_ports.expect("exposed ports set");
        assert!(exposed.iter().any(|s| s == "8080/tcp"));
        let hc = body.host_config.expect("host config");
        let b = hc.port_bindings.expect("bindings").get("8080/tcp").and_then(|v| v.clone()).expect("8080");
        assert_eq!(b[0].host_port.as_deref(), Some("8080"));
    }

    /// Bind-mount wiring (kills the `podman -v …` shell twin): a non-empty `binds`
    /// attaches `HostConfig.binds`, and empty `binds` is byte-for-byte identical to
    /// `create_body` (no `host_config` at all when there are no ports either) — so
    /// every existing caller is unchanged. No daemon.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn create_body_with_binds_attaches_host_mounts_and_stays_parity_when_empty() {
        // With binds, no ports: host_config present, binds set, port_bindings None.
        let with = Engine::create_body_with_binds(
            "wix:4",
            &[],
            &["build".to_string()],
            &[],
            &["/host/in:/work/in:ro".to_string(), "/host/out:/work/out".to_string()],
        );
        let hc = with.host_config.expect("host config for binds");
        assert_eq!(
            hc.binds,
            Some(vec!["/host/in:/work/in:ro".to_string(), "/host/out:/work/out".to_string()])
        );
        assert!(hc.port_bindings.is_none(), "no ports => no port bindings");

        // Empty binds + no ports => identical to create_body: no host_config.
        let plain = Engine::create_body("wix:4", &[], &["build".to_string()], &[]);
        let via_empty = Engine::create_body_with_binds("wix:4", &[], &["build".to_string()], &[], &[]);
        assert!(plain.host_config.is_none());
        assert_eq!(plain, via_empty, "empty binds is byte-parity with create_body");
    }

    /// **Airgap network-mode wiring** (the load-bearing wire for Skidbladnir's
    /// airgap container route): `Some("none")` sets `HostConfig.network_mode =
    /// "none"` on the create body — even with no ports/binds a `HostConfig` is
    /// minted to carry it — while `None` is **byte-for-byte identical** to the
    /// net-less builder (no `network_mode`, no `HostConfig` when ports+binds are
    /// empty too). RED-when-broken: drop the `network_mode` thread-through and the
    /// `"none"` assert fails; break the byte-parity and the last assert fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn create_body_sets_network_mode_none_and_default_is_byte_identical() {
        // net=none, no ports/binds => host_config minted purely to carry network_mode.
        let airgap = Engine::create_body_with_binds_and_net(
            "job:1", &[], &["run".to_string()], &[], &[], Some("none"),
        );
        let hc = airgap.host_config.expect("host config minted for network mode");
        assert_eq!(hc.network_mode.as_deref(), Some("none"), "airgap => --network none");
        assert!(hc.port_bindings.is_none(), "no ports => no port bindings");
        assert!(hc.binds.is_none(), "no binds => no binds");

        // net=default (None) with no ports/binds => byte-identical to the net-less
        // builder: no host_config at all (unchanged create body).
        let plain = Engine::create_body_with_binds("job:1", &[], &["run".to_string()], &[], &[]);
        let via_none = Engine::create_body_with_binds_and_net(
            "job:1", &[], &["run".to_string()], &[], &[], None,
        );
        assert!(plain.host_config.is_none(), "default net + no ports/binds => no host_config");
        assert_eq!(plain, via_none, "None network_mode is byte-parity with the net-less builder");

        // net=default WITH ports => host_config present but network_mode unset
        // (still byte-identical to the pre-net builder for an existing port spec).
        let ported = Engine::create_body_with_binds("web:1", &[], &[], &[8080], &[]);
        let ported_none = Engine::create_body_with_binds_and_net("web:1", &[], &[], &[8080], &[], None);
        assert_eq!(ported, ported_none, "None net leaves a ported spec byte-identical");
        assert!(ported.host_config.unwrap().network_mode.is_none(), "default net sets no network_mode");
    }

    /// **Published ports on a NON-isolated network = a host-reachable local-zone
    /// container** (the korp default-infra fix). The create body must publish each
    /// port as a host binding (`{p}/tcp` → `0.0.0.0:{p}`) AND leave `network_mode`
    /// UNSET (the default NAT/bridge, so the published port is actually reachable) —
    /// this is the non-airgap counterpart to `--network none`. RED-when-broken: if a
    /// published port stopped host-binding, or the default net started emitting
    /// `"none"` (which would make the published port unreachable), this fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn published_ports_bind_to_the_host_on_a_non_isolated_network() {
        // FalkorDB :6379 on the default (non-isolated) network — the korp local-zone
        // shape. `network_mode` = None (default NAT) so the binding is reachable.
        let body = Engine::create_body_with_binds_and_net(
            "docker.io/falkordb/falkordb:v4.20.0", &[], &[], &[6379], &[], None,
        );
        let hc = body.host_config.expect("host config for published port");
        assert!(hc.network_mode.is_none(), "non-isolated: no --network none (reachable)");
        let bindings = hc.port_bindings.expect("port bindings");
        let b = bindings.get("6379/tcp").and_then(|v| v.clone()).expect("6379 published");
        assert_eq!(b[0].host_ip.as_deref(), Some("0.0.0.0"), "bound on the host");
        assert_eq!(b[0].host_port.as_deref(), Some("6379"), "host port == container port");
        assert!(
            body.exposed_ports.unwrap().iter().any(|s| s == "6379/tcp"),
            "6379 exposed",
        );
        crate::functional_status(
            "draupnir/container",
            "published_port_non_isolated",
            hc.network_mode.is_none(),
            "published ports bind 0.0.0.0:host on the default (non-isolated) net → host-reachable local-zone container",
        );
    }

    /// **Full-machine resource knobs reach the podman invocation** (the hot-infra
    /// law: no spawned infra throttled to one core). `cpus` → `HostConfig.nano_cpus`
    /// (`n * 1e9`), `mem_mb` → `HostConfig.memory` (MiB → bytes); both `None` is
    /// **byte-for-byte identical** to the resource-less create body (no `HostConfig`
    /// minted purely for an unset cap → the container sees ALL host cores). RED-when-
    /// broken: drop the `nano_cpus`/`memory` thread-through and the cap asserts fail;
    /// break the None byte-parity and the last assert fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn resource_caps_reach_the_create_body_and_none_is_byte_identical() {
        // FalkorDB hot: all 12 Loki cores + 16 GiB — the caps land on HostConfig.
        let hot = Engine::create_body_with_res(
            "docker.io/falkordb/falkordb:v4.20.0", &[], &[], &[6379], &[], None,
            Some(12.0), Some(16384),
        );
        let hc = hot.host_config.expect("host config for resource caps");
        assert_eq!(hc.nano_cpus, Some(12_000_000_000), "12 cores → nano_cpus");
        assert_eq!(hc.memory, Some(16384_i64 * 1024 * 1024), "16384 MiB → bytes");
        // The published port survives alongside the caps.
        assert!(hc.port_bindings.unwrap().contains_key("6379/tcp"), "port still published");

        // Unconstrained (None, None) = byte-identical to the resource-less builder:
        // no cap is minted, so the container sees ALL host cores (hot-infra default).
        let plain = Engine::create_body_with_binds_and_net("redis:7", &[], &[], &[], &[], None);
        let via_none =
            Engine::create_body_with_res("redis:7", &[], &[], &[], &[], None, None, None);
        assert!(plain.host_config.is_none(), "no caps + no ports/binds → no host_config");
        assert_eq!(plain, via_none, "None caps is byte-parity with the resource-less builder");

        crate::functional_status(
            "draupnir/container",
            "resource_caps_reach_create_body",
            hc.nano_cpus == Some(12_000_000_000) && hc.memory == Some(16384_i64 * 1024 * 1024),
            "cpus/mem reach HostConfig.nano_cpus/memory; None = all host cores (hot-infra never throttled)",
        );
    }

    /// **Distinct `host:container` publish reaches the create body** (the per-zone
    /// port fix). A `PortMap { host: 6380, container: 6379 }` must expose the
    /// **container** port (`6379/tcp`) and bind it to the **host** port (`6380`) —
    /// podman `-p 6380:6379` — so korp's `Test` zone reaches FalkorDB (fixed on
    /// `6379` INSIDE) on host `6380` while `Demo` keeps `6379:6379`. The old wire
    /// (host==container only) published `6380:6380`, which never reached the
    /// container. RED-when-broken: if the map keyed the exposed/binding on the HOST
    /// port, or bound the CONTAINER port on the host, the two `assert_eq!`s below
    /// flip; the single-port arm guards the `6379:6379` back-compat path.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn distinct_host_container_port_map_publishes_host_to_container() {
        // Test zone: FalkorDB on host 6380 → container 6379 (the deferred-bug fix).
        let body = Engine::create_body_full(
            "docker.io/falkordb/falkordb:v4.20.0",
            &[], &[], &[], &[crate::PortMap::new(6380, 6379)], &[], None, None, None,
        );
        let hc = body.host_config.expect("host config for published map");
        assert!(hc.network_mode.is_none(), "non-isolated: reachable (no --network none)");
        let bindings = hc.port_bindings.expect("port bindings");
        // The binding is keyed on the CONTAINER port and bound to the HOST port.
        let b = bindings.get("6379/tcp").and_then(|v| v.clone()).expect("container 6379 published");
        assert_eq!(b[0].host_ip.as_deref(), Some("0.0.0.0"), "bound on the host");
        assert_eq!(b[0].host_port.as_deref(), Some("6380"), "host 6380 -> container 6379");
        assert!(!bindings.contains_key("6380/tcp"), "the HOST port is NOT the container key");
        assert!(
            body.exposed_ports.as_ref().unwrap().iter().any(|s| s == "6379/tcp"),
            "the CONTAINER port is exposed, not the host port",
        );

        // The single-port (host==container) form still yields 6379:6379 (demo/back-
        // compat) — proven equivalent to the PortMap::same form for the same port.
        let demo = Engine::create_body_full(
            "docker.io/falkordb/falkordb:v4.20.0",
            &[], &[], &[6379], &[], &[], None, None, None,
        );
        let demo_via_map = Engine::create_body_full(
            "docker.io/falkordb/falkordb:v4.20.0",
            &[], &[], &[], &[crate::PortMap::same(6379)], &[], None, None, None,
        );
        let db = demo.host_config.clone().unwrap().port_bindings.unwrap();
        let sb = db.get("6379/tcp").and_then(|v| v.clone()).expect("demo 6379 published");
        assert_eq!(sb[0].host_port.as_deref(), Some("6379"), "single-port form: host==container==6379");
        assert_eq!(demo, demo_via_map, "[6379] and PortMap::same(6379) render the same wire");

        // Empty ports + empty maps + no net/caps/binds = byte-identical no-op
        // (no HostConfig minted), exactly as a bare image spec always rendered.
        let bare = Engine::create_body_full("redis:7", &[], &[], &[], &[], &[], None, None, None);
        assert!(bare.host_config.is_none(), "no ports/maps/net/caps => no host_config (byte-parity)");

        crate::functional_status(
            "draupnir/container",
            "distinct_host_container_port_map",
            sb[0].host_port.as_deref() == Some("6379")
                && b[0].host_port.as_deref() == Some("6380")
                && bindings.contains_key("6379/tcp"),
            "PortMap{host,container} publishes host:container (-p 6380:6379) so a per-zone host port reaches a fixed in-container port",
        );
    }

    // -----------------------------------------------------------------------
    // run_to_completion — driven entirely over the always-compiled Boot +
    // ContainerControl seam by a mock, so it proves start→wait→exit-code→logs
    // with no daemon.
    // -----------------------------------------------------------------------

    use std::cell::RefCell;

    /// A scripted container backend: `boot` records the spec and mints a Machine;
    /// each `container_state` call pops the next scripted state (staying on the last
    /// once the script is exhausted); `drain_logs_split` hands out the next batch of
    /// (stdout, stderr) lines then empties. Proves the run-to-completion driver with
    /// no daemon.
    #[derive(Default)]
    struct ScriptedBackend {
        booted: RefCell<Vec<String>>,
        stopped: RefCell<Vec<String>>,
        states: RefCell<std::collections::VecDeque<ContainerState>>,
        /// Each entry is the (stdout, stderr) lines yielded by one drain.
        log_batches: RefCell<std::collections::VecDeque<(Vec<String>, Vec<String>)>>,
    }

    impl ScriptedBackend {
        fn with_states(states: Vec<ContainerState>) -> Self {
            Self { states: RefCell::new(states.into()), ..Default::default() }
        }
        fn push_logs(&self, out: &[&str], err: &[&str]) {
            self.log_batches.borrow_mut().push_back((
                out.iter().map(|s| s.to_string()).collect(),
                err.iter().map(|s| s.to_string()).collect(),
            ));
        }
    }

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

    impl ContainerControl for ScriptedBackend {
        fn container_state(&self, _m: &Machine) -> ContainerState {
            let mut q = self.states.borrow_mut();
            if q.len() > 1 {
                q.pop_front().unwrap()
            } else {
                q.front().cloned().unwrap_or(ContainerState::Gone)
            }
        }
        fn drain_logs(&self, _m: &Machine) -> Vec<String> {
            let (mut out, mut err) = self.log_batches.borrow_mut().pop_front().unwrap_or_default();
            out.append(&mut err);
            out
        }
        fn drain_logs_split(&self, _m: &Machine) -> (Vec<String>, Vec<String>) {
            self.log_batches.borrow_mut().pop_front().unwrap_or_default()
        }
        fn stop(&self, m: &Machine) {
            self.stopped.borrow_mut().push(m.id.clone());
        }
    }

    fn fast_opts() -> RunOptions {
        RunOptions::poll_every(Duration::from_millis(1))
    }

    #[test]
    fn run_to_completion_starts_waits_collects_split_logs_and_exit_code() {
        // Running for two polls, then a clean exit(0). Logs arrive across ticks and
        // in a final flush after exit — split into stdout/stderr.
        let backend = ScriptedBackend::with_states(vec![
            ContainerState::Running,
            ContainerState::Running,
            ContainerState::Exited(0),
        ]);
        backend.push_logs(&["booting"], &[]); // tick 1 drain
        backend.push_logs(&["serving"], &["a warning"]); // tick 2 drain
        backend.push_logs(&["bye"], &[]); // final drain after exit

        let spec = BootSpec::container("job", "docker.io/library/busybox:latest");
        let out = run_to_completion(&backend, &spec, &fast_opts()).unwrap();

        assert_eq!(out.exit_code, Some(0));
        assert_eq!(out.stdout, vec!["booting", "serving", "bye"]);
        assert_eq!(out.stderr, vec!["a warning"]);
        // It booted exactly the spec and removed the container it started.
        assert_eq!(backend.booted.borrow().as_slice(), &["job".to_string()]);
        assert_eq!(backend.stopped.borrow().as_slice(), &["draupnir-job".to_string()]);
    }

    #[test]
    fn run_to_completion_surfaces_a_nonzero_exit_as_ok_not_err() {
        // A crashing container is a successful CALL carrying a non-zero code — the
        // job failed, not the API (parity with jera's run_container).
        let backend = ScriptedBackend::with_states(vec![ContainerState::Exited(137)]);
        backend.push_logs(&[], &["oom-killed"]);
        let spec = BootSpec::container("crash", "img:1");
        let out = run_to_completion(&backend, &spec, &fast_opts()).unwrap();
        assert_eq!(out.exit_code, Some(137));
        assert_eq!(out.stderr, vec!["oom-killed"]);
    }

    #[test]
    fn run_to_completion_reports_none_when_the_container_is_gone() {
        // Vanished before a code could be read → exit_code None (not a fake 0).
        let backend = ScriptedBackend::with_states(vec![ContainerState::Gone]);
        let spec = BootSpec::container("vanished", "img:1");
        let out = run_to_completion(&backend, &spec, &fast_opts()).unwrap();
        assert_eq!(out.exit_code, None);
        assert_eq!(backend.stopped.borrow().len(), 1, "still removed on the way out");
    }

    #[test]
    fn run_to_completion_times_out_with_a_clear_error_and_stops_the_container() {
        // Never exits → the bounded budget elapses → Error::Backend naming the
        // instance, and the container is stopped (no leak).
        let backend = ScriptedBackend::with_states(vec![ContainerState::Running]);
        let spec = BootSpec::container("hang", "img:1");
        let opts = RunOptions::bounded(Duration::from_millis(20), Duration::from_millis(2));
        let err = run_to_completion(&backend, &spec, &opts).unwrap_err();
        match err {
            Error::Backend(m) => {
                assert!(m.contains("draupnir-hang"), "names the instance: {m}");
                assert!(m.contains("run to completion"), "says what timed out: {m}");
            }
            other => panic!("expected Error::Backend, got {other:?}"),
        }
        assert_eq!(backend.stopped.borrow().len(), 1, "container stopped on timeout");
    }

    #[test]
    fn run_to_completion_propagates_a_boot_failure_without_polling() {
        // A backend whose boot fails must surface Err and never poll/stop.
        struct FailBoot;
        impl Boot for FailBoot {
            fn boot(&self, _spec: &BootSpec) -> Result<Machine> {
                Err(Error::Backend("no socket".into()))
            }
        }
        impl ContainerControl for FailBoot {
            fn container_state(&self, _m: &Machine) -> ContainerState {
                panic!("must not poll after a boot failure")
            }
            fn drain_logs(&self, _m: &Machine) -> Vec<String> {
                Vec::new()
            }
            fn stop(&self, _m: &Machine) {
                panic!("must not stop after a boot failure")
            }
        }
        let spec = BootSpec::container("x", "img:1");
        let err = run_to_completion(&FailBoot, &spec, &fast_opts()).unwrap_err();
        assert!(matches!(err, Error::Backend(m) if m.contains("no socket")));
    }

    #[test]
    fn default_drain_logs_split_routes_combined_logs_to_stdout() {
        // The trait default (a backend that doesn't distinguish streams) puts every
        // combined line on stdout, stderr empty — nothing observable is lost.
        struct CombinedOnly;
        impl ContainerControl for CombinedOnly {
            fn container_state(&self, _m: &Machine) -> ContainerState {
                ContainerState::Gone
            }
            fn drain_logs(&self, _m: &Machine) -> Vec<String> {
                vec!["one".into(), "two".into()]
            }
            fn stop(&self, _m: &Machine) {}
        }
        let m = Machine::started("draupnir-x", &BootSpec::container("x", "img:1"));
        let (out, err) = CombinedOnly.drain_logs_split(&m);
        assert_eq!(out, vec!["one", "two"]);
        assert!(err.is_empty());
    }

    // -----------------------------------------------------------------------
    // ContainerControl::exec — the guards + the assembled `podman exec <id>
    // <argv…>` command are proven over the always-compiled seam by a mock, with
    // no daemon (the live bollard `/exec` drive is a Loki integration concern).
    // -----------------------------------------------------------------------

    /// Records the exact command [`ContainerControl::exec`] assembled and hands back
    /// a scripted outcome — proves `exec` builds `["exec", id, argv…]` and returns
    /// the engine's result, with no daemon.
    #[derive(Default)]
    struct ExecRecorder {
        seen: RefCell<Vec<Vec<String>>>,
        outcome: ExecOutcome,
    }
    impl ContainerControl for ExecRecorder {
        fn container_state(&self, _m: &Machine) -> ContainerState {
            ContainerState::Running
        }
        fn drain_logs(&self, _m: &Machine) -> Vec<String> {
            Vec::new()
        }
        fn stop(&self, _m: &Machine) {}
        fn exec_command(&self, command: &[String]) -> Result<ExecOutcome> {
            self.seen.borrow_mut().push(command.to_vec());
            Ok(self.outcome.clone())
        }
    }

    #[test]
    fn exec_argv_builds_the_podman_exec_command() {
        // The pure command builder: ["exec", id, argv…] — the canonical podman-exec
        // form the live engine drives (`id` + `argv = command[2..]`).
        assert_eq!(
            exec_argv("draupnir-cache", &["redis-cli", "ping"]),
            vec!["exec", "draupnir-cache", "redis-cli", "ping"]
        );
        assert_eq!(exec_argv("draupnir-x", &["true"]), vec!["exec", "draupnir-x", "true"]);
    }

    #[test]
    fn exec_assembles_the_command_and_returns_the_outcome() {
        // RED-when-broken: `exec` must build ["exec", <machine.id>, <argv…>] and
        // hand back the engine's ExecOutcome. A recorder mock captures the command
        // with no daemon; neutralize exec_argv (e.g. stop pushing the id) and this
        // recorded-command assert fails.
        let recorder = ExecRecorder {
            outcome: ExecOutcome {
                exit_code: Some(0),
                stdout: vec!["PONG".into()],
                stderr: vec![],
            },
            ..Default::default()
        };
        let m = Machine::started("draupnir-cache", &BootSpec::container("cache", "redis:7"));
        let out = recorder.exec(&m, &["redis-cli", "ping"]).unwrap();
        assert_eq!(
            recorder.seen.borrow().as_slice(),
            &[vec![
                "exec".to_string(),
                "draupnir-cache".to_string(),
                "redis-cli".to_string(),
                "ping".to_string(),
            ]]
        );
        assert_eq!(out.exit_code, Some(0));
        assert_eq!(out.stdout, vec!["PONG"]);
    }

    #[test]
    fn exec_is_rejected_on_a_non_container_machine() {
        // Container-only guard (parity with the container-only net/cmd/ports checks):
        // a KVM Machine has nothing to exec into — reject BEFORE the engine, and
        // never record a command.
        let recorder = ExecRecorder::default();
        let kvm = Machine::started(
            "vm-1",
            &BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz"),
        );
        assert!(matches!(recorder.exec(&kvm, &["ls"]), Err(Error::Spec(_))));
        assert!(recorder.seen.borrow().is_empty(), "guard runs before the engine");
    }

    #[test]
    fn exec_rejects_an_empty_argv() {
        // Nothing to run => Error::Spec, no command assembled.
        let recorder = ExecRecorder::default();
        let m = Machine::started("draupnir-cache", &BootSpec::container("cache", "redis:7"));
        assert!(matches!(recorder.exec(&m, &[]), Err(Error::Spec(_))));
        assert!(recorder.seen.borrow().is_empty());
    }

    #[test]
    fn exec_without_an_engine_is_unsupported_not_faked() {
        // A backend that keeps the default `exec_command` (no OCI engine) passes the
        // guards then honestly reports Unsupported — never a fake exec.
        struct NoEngine;
        impl ContainerControl for NoEngine {
            fn container_state(&self, _m: &Machine) -> ContainerState {
                ContainerState::Gone
            }
            fn drain_logs(&self, _m: &Machine) -> Vec<String> {
                Vec::new()
            }
            fn stop(&self, _m: &Machine) {}
        }
        let m = Machine::started("draupnir-x", &BootSpec::container("x", "img:1"));
        assert!(matches!(NoEngine.exec(&m, &["true"]), Err(Error::Unsupported(_))));
    }

    /// An empty spec yields a bare create body: no cmd override, no env, no ports.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn create_body_empty_spec_is_bare() {
        let body = Engine::create_body("scratch", &[], &[], &[]);
        assert_eq!(body.image.as_deref(), Some("scratch"));
        assert!(body.cmd.is_none());
        assert!(body.env.is_none());
        assert!(body.exposed_ports.is_none());
        assert!(body.host_config.is_none());
    }

    /// The OCI **image-build context** is tarred correctly (kills the `podman build`
    /// shell twin): every file under the context dir — nested included — lands in the
    /// tar at its relative path, the shape the daemon's `/build` endpoint expects.
    /// Pure, no daemon: build a temp context → `context_tar` → read the tar back.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn context_tar_packs_the_build_context() {
        let dir = std::env::temp_dir().join(format!("draupnir-ctx-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(dir.join("sub")).unwrap();
        std::fs::write(dir.join("Containerfile"), b"FROM scratch\n").unwrap();
        std::fs::write(dir.join("sub").join("app.txt"), b"hi").unwrap();

        let bytes = context_tar(&dir).unwrap();
        assert!(!bytes.is_empty(), "the tar carries the context");

        let mut ar = tar::Archive::new(std::io::Cursor::new(bytes));
        let names: Vec<String> = ar
            .entries()
            .unwrap()
            .map(|e| e.unwrap().path().unwrap().to_string_lossy().replace('\\', "/"))
            .collect();
        assert!(
            names.iter().any(|n| n.ends_with("Containerfile")),
            "Containerfile packed: {names:?}"
        );
        assert!(
            names.iter().any(|n| n.ends_with("sub/app.txt")),
            "nested file packed: {names:?}"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Without the `backend-oci` engine, `build_image`/`extract_path` are honest
    /// [`Error::Unsupported`] — never a fake image or a partial extract.
    #[cfg(not(feature = "backend-oci"))]
    #[test]
    fn build_and_extract_are_unsupported_without_the_engine() {
        let boot = ContainerBoot::new();
        assert!(matches!(
            boot.build_image(Path::new("."), "Containerfile", "x:test"),
            Err(Error::Unsupported(_))
        ));
        assert!(matches!(
            boot.extract_path("x:test", "/out", Path::new("/tmp/draupnir-x")),
            Err(Error::Unsupported(_))
        ));
    }
}