draupnir 0.1.9

//! **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();
            let outcome =
                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),
                    },
                );
            // Record the readiness verdict: GREEN once the app is up, RED when the
            // probe timed out or the backend errored — the wait_ready surface nornir's
            // matrix reads back (a container that never became ready must be visible).
            crate::functional_status(
                "draupnir/container",
                "wait_ready",
                outcome.is_ok(),
                &match &outcome {
                    Ok(()) => format!("container `{name}` reached {readiness:?}"),
                    Err(e) => format!("container `{name}` never became ready: {e}"),
                },
            );
            outcome
        }
        #[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,
                &spec.security_opt,
            )?;
            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);

    /// **Every container whose name begins with `prefix`, running or not**, as
    /// `(name, state)`.
    ///
    /// The call a *janitor* needs, and the one whose absence made consumers
    /// reach past this seam. gunnar's bench suite carried
    ///
    /// ```text
    /// // draupnir has no list-containers call, so this is `podman` directly
    /// Command::new("podman").args(["ps", "-a", "--filter", "name=…", "--format", "{{.Names}}"])
    /// ```
    ///
    /// to sweep containers a dead run had left behind. That comment was correct
    /// and is the reason this method exists: a boot path with no shell fallback
    /// is worth little if the sweep beside it shells out anyway, because the
    /// shell is where the `podman`-not-on-PATH and the wrong-engine failures come
    /// back in — silently, since a failed `podman ps` and a genuinely empty box
    /// produce the same empty stdout.
    ///
    /// Exited and created containers are included: a janitor that saw only
    /// running ones would leave exactly the corpses it exists to remove.
    ///
    /// The **default** answers `Unsupported`, so a mock backend says so rather
    /// than reporting an empty box — "nothing to sweep" and "I cannot look" must
    /// not be the same answer to a caller about to decide nothing needs removing.
    fn list_containers(&self, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
        let _ = prefix;
        Err(Error::Unsupported(
            "this backend cannot list containers (needs the OCI engine)".into(),
        ))
    }

    /// **The environment the container was created with**, as `KEY=VALUE` lines,
    /// straight out of the daemon's own record of it.
    ///
    /// Not "what we passed" — what the daemon holds. That difference is the
    /// whole value: a variable dropped between a [`BootSpec`] and the running
    /// process is invisible to any check that re-reads the spec, and settings
    /// like `GOMAXPROCS` or `TOKIO_WORKER_THREADS` have no observable effect a
    /// harness can query from outside, so this is the only read-back they have.
    ///
    /// The second call gunnar's bench shelled out for
    /// (`podman inspect --format '{{range .Config.Env}}…'`).
    ///
    /// [`BootSpec`]: crate::BootSpec
    fn container_env(&self, machine: &Machine) -> Result<Vec<String>> {
        let _ = machine;
        Err(Error::Unsupported(
            "this backend cannot inspect a container's environment (needs the OCI engine)".into(),
        ))
    }

    /// **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 list_containers(&self, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
        #[cfg(feature = "backend-oci")]
        {
            self.engine()?.list_containers(prefix)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = prefix;
            Err(Error::Unsupported(
                "this build has no OCI engine, so it cannot list containers".into(),
            ))
        }
    }

    fn container_env(&self, machine: &Machine) -> Result<Vec<String>> {
        #[cfg(feature = "backend-oci")]
        {
            self.engine()?.container_env(&machine.id)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = machine;
            Err(Error::Unsupported(
                "this build has no OCI engine, so it cannot inspect a container".into(),
            ))
        }
    }

    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,
                &spec.security_opt,
            )?;
            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(),
            ))
        }
    }

    /// **Is an OCI image already present locally?** — the `podman image exists <ref>`
    /// / `docker image inspect <ref>` shell twin over the one engine. `Ok(false)` is
    /// reserved for the daemon answering *"no such image"*; any other failure
    /// (socket unreachable, permission denied) is an honest [`Error::Backend`], so a
    /// caller can never mistake a broken engine for a missing image and rebuild the
    /// world. The 404→`false` vs error→`Err` decision lives in [`image_present_on`]
    /// (daemon-free, mock-tested). Requires `backend-oci`.
    pub fn image_present(&self, image: &str) -> Result<bool> {
        #[cfg(feature = "backend-oci")]
        {
            image_present_on(self.engine()?.as_ref(), image)
        }
        #[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(),
            ))
        }
    }

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

    /// **Export a local image to a tar archive on disk** — the `podman save -o` twin
    /// over the one engine, and the missing half of the airgap story: without it a
    /// bundle has no image ARTIFACT to carry, so an offline target is left running
    /// [`build_image`](Self::build_image) at first use, i.e. reaching for a network
    /// on a box that has none.
    ///
    /// **Streams.** Bytes go daemon → `BufWriter` → file a chunk at a time; the
    /// gigabyte is never resident. (`nordisk-spark-iceberg:4.1.2` measures 1.39 GB
    /// on oden, which is also why these images ride the bundle and not a crate.)
    ///
    /// **Three failures stay three failures**, which is the whole point of routing
    /// this through [`ImagePresence`] rather than a shell exit code:
    /// - the image is not on this host → *"no such image locally"*, and **no file is
    ///   created** (a zero-byte tar that later "verifies" as present is the exact
    ///   silent-corruption path this avoids);
    /// - the socket is unreachable/denied → *"container engine unreachable"*;
    /// - the stream died mid-archive → [`ArchiveFault::Truncated`], caught by
    ///   re-reading the written file's tar headers before returning success.
    ///
    /// On success the returned [`ImageArchive`] carries the tags and config digest
    /// **read back out of the archive**, not the ones the caller asked for — so a
    /// manifest records what the tar actually holds. Requires `backend-oci`.
    pub fn export_image(&self, image: &str, dest: &Path) -> Result<ImageArchive> {
        #[cfg(feature = "backend-oci")]
        {
            export_image_on(self.engine()?.as_ref(), image, dest)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = (image, dest);
            Err(Error::Unsupported(
                "container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
                    .into(),
            ))
        }
    }

    /// **Import an image archive back into the local store** — the `podman load -i`
    /// twin over the one engine, and what an airgapped target runs on unfold so the
    /// service that needs the image finds it already there.
    ///
    /// The archive is **verified before a single byte is streamed at the daemon**
    /// ([`inspect_image_archive`]): a truncated or non-image tar fails here, cheaply
    /// and by name, instead of half-loading and leaving the store in a state nobody
    /// can describe. Only then is the file streamed to `/images/load` in chunks —
    /// again never resident in memory.
    ///
    /// Returns the [`ImageArchive`] describing what was loaded, so a caller can log
    /// the tags and digest it actually installed rather than the ones it hoped for.
    /// Requires `backend-oci`.
    pub fn import_image(&self, src: &Path) -> Result<ImageArchive> {
        #[cfg(feature = "backend-oci")]
        {
            import_image_on(self.engine()?.as_ref(), src)
        }
        #[cfg(not(feature = "backend-oci"))]
        {
            let _ = src;
            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)
}

/// The **daemon-touching primitives** of the image *build* and *extract* paths,
/// pulled behind a seam so the surrounding logic — a `BuildInfo.error_detail`
/// failing the build, the returned tag on a clean stream, and the basename-rooted
/// tar unpack — is drivable by a mock with **no socket** (the same treatment
/// [`create_body`](Engine::create_body) / [`context_tar`] already got). The live
/// [`Engine`] implements it over bollard; a unit-test mock scripts the daemon's
/// replies to prove [`build_image_on`] / [`extract_path_on`] with no daemon.
#[cfg(feature = "backend-oci")]
trait ImageDaemon {
    /// Stream a build of the tarred `context` (dockerfile `containerfile`, tagged
    /// `tag`), drained to **one item per `BuildInfo`**: `Ok(None)` = a progress
    /// line, `Ok(Some(msg))` = a daemon-reported build error (an `error_detail`,
    /// its `message` — empty string if the detail carried none), and `Err(..)` = a
    /// transport failure. Draining to a `Vec` keeps the async stream inside the
    /// engine while the pass/fail decision ([`build_image_on`]) stays daemon-free.
    fn build_stream(
        &self,
        context: Vec<u8>,
        containerfile: &str,
        tag: &str,
    ) -> Vec<Result<Option<String>>>;

    /// Create a throwaway (un-started) container from `image`, download a tar of
    /// `container_path` from the daemon, and remove the container (best-effort, even
    /// on a download error). Returns the raw tar bytes — rooted at the basename of
    /// `container_path`, the `podman cp` layout [`extract_path_on`] then unpacks.
    fn download_path_tar(&self, image: &str, container_path: &str) -> Result<Vec<u8>>;

    /// Ask the daemon whether `image` resolves locally, flattened to the three
    /// answers [`image_present_on`] decides on. The engine maps a bollard
    /// `inspect_image` reply here; a mock scripts it with no socket.
    fn inspect_image(&self, image: &str) -> ImagePresence;

    /// Stream `image`'s archive from the daemon into `sink`, returning the byte
    /// count written. Takes a `&mut dyn Write` rather than returning a `Vec<u8>`
    /// **specifically** so the gigabyte never becomes a value: the caller owns a
    /// `BufWriter<File>` and each chunk goes straight through it. The surrounding
    /// decisions (presence probe, dir creation, read-back verification) live in
    /// [`export_image_on`], daemon-free.
    fn export_to(&self, image: &str, sink: &mut dyn std::io::Write) -> Result<u64>;

    /// Stream the archive at `src` into the daemon's image store. Takes a path, not
    /// bytes, for the same reason: the engine reads it in chunks. Verification of
    /// what is at that path already happened in [`import_image_on`] before this is
    /// called, so an implementation may assume a well-formed archive and report only
    /// what the daemon says about loading it.
    fn load_archive(&self, src: &Path) -> Result<()>;
}

/// The daemon's answer to "do you have this image?", split so the **missing** case
/// is distinguishable from a **broken** one. `podman image exists` collapses both
/// into a non-zero exit — that conflation is exactly what this seam exists to kill:
/// a caller that reads an unreachable socket as "absent" self-heals by rebuilding an
/// image it already has, forever.
#[cfg(feature = "backend-oci")]
#[derive(Debug, Clone, PartialEq, Eq)]
enum ImagePresence {
    /// The daemon resolved the reference.
    Present,
    /// The daemon answered, and has no such image (a 404).
    Absent,
    /// The daemon could not be asked (socket down, denied, transport) — carries the
    /// message that becomes the [`Error::Backend`].
    Failed(String),
}

/// **Decide image presence over the [`ImageDaemon`] seam.** `Present` → `Ok(true)`,
/// `Absent` → `Ok(false)`, `Failed` → [`Error::Backend`]. Pure but for the injected
/// daemon, so a mock proves all three arms — in particular that a transport failure
/// never reads as "absent".
#[cfg(feature = "backend-oci")]
fn image_present_on<D: ImageDaemon>(daemon: &D, image: &str) -> Result<bool> {
    match daemon.inspect_image(image) {
        ImagePresence::Present => Ok(true),
        ImagePresence::Absent => Ok(false),
        ImagePresence::Failed(msg) => Err(Error::Backend(format!("inspect image {image}: {msg}"))),
    }
}

/// **Drive an image build over the [`ImageDaemon`] seam.** Tars `context_dir`, streams
/// the build, and fails on the first `error_detail` the daemon reports (an
/// [`Error::Backend`] naming the tag + message) — else returns `tag`. Everything but
/// the injected daemon is pure, so a mock proves the error-detail→`Err` and
/// returned-tag paths with no socket.
#[cfg(feature = "backend-oci")]
fn build_image_on<D: ImageDaemon>(
    daemon: &D,
    context_dir: &Path,
    containerfile: &str,
    tag: &str,
) -> Result<String> {
    let tar = context_tar(context_dir)?;
    for step in daemon.build_stream(tar, containerfile, tag) {
        if let Some(msg) = step? {
            return Err(Error::Backend(format!("build image {tag}: {msg}")));
        }
    }
    Ok(tag.to_string())
}

/// **Drive a path extract over the [`ImageDaemon`] seam.** Downloads the path's tar
/// from a throwaway container, then unpacks it into `host_dest` (created if absent).
/// The tar the daemon hands back is rooted at the basename of `container_path`
/// (`podman cp` layout), so the unpacked tree lands at `host_dest/<basename>/…`. A
/// download error surfaces as `Err` **before** any unpack; only the injected daemon
/// touches a socket, so a mock proves the unpack + basename layout with no daemon.
#[cfg(feature = "backend-oci")]
fn extract_path_on<D: ImageDaemon>(
    daemon: &D,
    image: &str,
    container_path: &str,
    host_dest: &Path,
) -> Result<()> {
    let tar_bytes = daemon.download_path_tar(image, container_path)?;
    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(())
}

// ---------------------------------------------------------------------------
// Image ARCHIVES — the export/import half of the one engine (the airgap path).
// ---------------------------------------------------------------------------

/// **What an image archive actually contains**, read out of the archive itself.
///
/// Deliberately NOT "what the caller named the file". An airgap bundle's whole
/// promise is that the target can trust what it received without a registry to ask,
/// so the identity of a packed image has to come from its bytes: the `RepoTags` and
/// the config digest inside the tar's own `manifest.json`.
///
/// The [`config_digest`](Self::config_digest) is the image's content identity (what
/// `podman images` shows as IMAGE ID). It is the ONLY thing tying a shipped image
/// back to the recipe that produced it — an image built on the build host is not
/// byte-reproducible from its `Containerfile` (timestamps, layer ordering, and any
/// network-fetched jar all differ), so "the recipe and the image agree" is a claim
/// no bundle can verify. Recording the digest at pack time and re-reading it on
/// unfold is what CAN be verified, and this type is that record.
///
/// Pure data — defined with no feature so a manifest elsewhere in the constellation
/// can name it without pulling `bollard`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImageArchive {
    /// Where the archive sits on disk.
    pub path: std::path::PathBuf,
    /// Archive size in bytes — the exported tar, uncompressed.
    pub bytes: u64,
    /// Every `RepoTags` entry across the archive's manifest, in archive order. May be
    /// empty for an OCI-layout archive that names its content only by digest.
    pub tags: Vec<String>,
    /// The image config digest as bare lowercase hex, no `sha256:` prefix (the IMAGE
    /// ID). Empty only if the archive named no config, which
    /// [`inspect_image_archive`] treats as malformed rather than passing through.
    pub config_digest: String,
}

/// **Why an image archive could not be believed.** Split into named cases because a
/// bundle that cannot tell "the file is not there" from "the file is half a file"
/// cannot report a useful failure to an operator standing at an offline box — and
/// because the second case is the one that silently produces a target running an
/// image that is missing layers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ArchiveFault {
    /// The file could not be opened or read at all (absent, permissions, IO).
    Unreadable(String),
    /// A tar entry's declared extent runs past the end of the file: the archive was
    /// cut short (a dead socket mid-export, a partial copy, a full disk). `declared`
    /// is where the last entry claims to end; `actual` is the file's real length.
    Truncated {
        /// Byte offset the last entry's header says its data ends at.
        declared: u64,
        /// The file's actual length.
        actual: u64,
    },
    /// A well-formed tar that is not an image archive — no `manifest.json` at the
    /// root. (An empty file lands here too, since an empty tar has no entries.)
    NotAnImageArchive,
    /// `manifest.json` was found but did not parse as the docker-archive shape.
    Malformed(String),
}

impl std::fmt::Display for ArchiveFault {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ArchiveFault::Unreadable(m) => write!(f, "unreadable: {m}"),
            ArchiveFault::Truncated { declared, actual } => write!(
                f,
                "truncated: a tar entry declares bytes out to {declared} but the file is only {actual} bytes"
            ),
            ArchiveFault::NotAnImageArchive => {
                write!(f, "not an image archive: no manifest.json at the tar root")
            }
            ArchiveFault::Malformed(m) => write!(f, "malformed manifest.json: {m}"),
        }
    }
}

/// Chunk size for streaming an archive INTO the daemon (`/images/load`). 4 MiB is
/// large enough that a 1.4 GB image is ~350 round trips rather than ~350 000, and
/// small enough that peak memory is a rounding error. Peak resident is one chunk.
#[cfg(feature = "backend-oci")]
const IMPORT_CHUNK: usize = 4 * 1024 * 1024;

/// Sanity bound on `manifest.json`. A real one is a few KB even for an archive of
/// many images; 8 MiB is far past any plausible value. It exists because the size
/// is read out of a tar header in a file whose trustworthiness is exactly what is
/// being decided, so it must never be used as an unchecked allocation hint.
#[cfg(feature = "backend-oci")]
const MAX_MANIFEST_BYTES: u64 = 8 * 1024 * 1024;

/// **Read an image archive's identity — with NO daemon and NO socket.**
///
/// Opens `path`, walks the tar's *headers* (seek-skipping every entry body, so a
/// 1.4 GB archive costs a handful of seeks rather than a 1.4 GB read), checks the
/// last entry's declared extent against the real file length, and parses the root
/// `manifest.json`.
///
/// This is the function a bundle's verify-on-receipt calls. It is deliberately
/// engine-free: an airgapped target must be able to reject a bad archive **before**
/// it starts a container engine, and a build host must be able to check what it
/// packed without one either. Requires `backend-oci` for the `tar` + `serde_json`
/// it rides.
#[cfg(feature = "backend-oci")]
pub fn inspect_image_archive(path: &Path) -> Result<ImageArchive> {
    let file = std::fs::File::open(path)
        .map_err(|e| archive_error(path, &ArchiveFault::Unreadable(e.to_string())))?;
    let len = file
        .metadata()
        .map_err(|e| archive_error(path, &ArchiveFault::Unreadable(e.to_string())))?
        .len();
    let (tags, config_digest) =
        scan_image_archive(file, len).map_err(|fault| archive_error(path, &fault))?;
    Ok(ImageArchive {
        path: path.to_path_buf(),
        bytes: len,
        tags,
        config_digest,
    })
}

/// Lean build: no `tar`/`serde_json`, so there is nothing to read an archive with.
#[cfg(not(feature = "backend-oci"))]
pub fn inspect_image_archive(path: &Path) -> Result<ImageArchive> {
    let _ = path;
    Err(Error::Unsupported(
        "reading an image archive needs the `backend-oci` feature (tar + serde_json)".into(),
    ))
}

/// One place that turns an [`ArchiveFault`] into the crate's [`Error`], so every
/// caller names the file the same way and the fault's own wording survives.
#[cfg(feature = "backend-oci")]
fn archive_error(path: &Path, fault: &ArchiveFault) -> Error {
    Error::Backend(format!("image archive {}: {fault}", path.display()))
}

/// **The pure scanner** behind [`inspect_image_archive`] — generic over `Read + Seek`
/// so the whole truncation / no-manifest / malformed-JSON decision tree is drivable
/// from an in-memory `Cursor` fixture with no filesystem and no daemon. Returns the
/// archive's `(tags, config_digest)`.
///
/// `len` is passed in rather than measured because a `Cursor` and a `File` learn
/// their length differently, and the truncation check needs the authoritative one.
#[cfg(feature = "backend-oci")]
fn scan_image_archive<R: std::io::Read + std::io::Seek>(
    reader: R,
    len: u64,
) -> std::result::Result<(Vec<String>, String), ArchiveFault> {
    let mut archive = tar::Archive::new(reader);
    let entries = archive
        .entries_with_seek()
        .map_err(|e| ArchiveFault::Unreadable(e.to_string()))?;

    let mut manifest: Option<Vec<u8>> = None;
    // The furthest byte any entry header claims the archive extends to. Compared
    // against the real length once the walk is done — a stream cut mid-layer leaves
    // a perfectly parseable header whose data simply is not all there, and header
    // parsing alone will never notice.
    let mut declared_end: u64 = 0;

    for entry in entries {
        // A header that will not parse is a cut in the HEADER itself; report it as
        // read failure with the real cause rather than inventing byte offsets for a
        // `Truncated` we cannot actually measure.
        let mut entry = entry.map_err(|e| {
            ArchiveFault::Unreadable(format!(
                "tar entry could not be read (archive is {len} bytes): {e}"
            ))
        })?;
        let size = entry.size();
        let data_at = entry.raw_file_position();
        // tar pads every entry body out to a 512-byte block.
        let padded = size.div_ceil(512).saturating_mul(512);
        declared_end = declared_end.max(data_at.saturating_add(padded));

        let is_manifest = entry
            .path()
            .map(|p| p.as_ref() == Path::new("manifest.json"))
            .unwrap_or(false);
        if is_manifest {
            // Bounded: a real manifest.json is a few KB. The size comes from a tar
            // header inside a file we are still deciding whether to trust, so it is
            // never used as an allocation hint unchecked — a corrupt header claiming
            // 4 EiB must not become a 4 EiB reserve.
            if size > MAX_MANIFEST_BYTES {
                return Err(ArchiveFault::Malformed(format!(
                    "manifest.json declares {size} bytes, over the {MAX_MANIFEST_BYTES}-byte \
                     sanity bound — this is not a real image manifest"
                )));
            }
            let mut buf = Vec::with_capacity(size as usize);
            std::io::Read::read_to_end(&mut entry, &mut buf)
                .map_err(|e| ArchiveFault::Unreadable(format!("read manifest.json: {e}")))?;
            manifest = Some(buf);
        }
    }

    if declared_end > len {
        return Err(ArchiveFault::Truncated {
            declared: declared_end,
            actual: len,
        });
    }
    let manifest = manifest.ok_or(ArchiveFault::NotAnImageArchive)?;
    parse_archive_manifest(&manifest)
}

/// **Parse a docker-archive `manifest.json`** into `(tags, config_digest)`. Pure —
/// bytes in, answer out.
///
/// The shape is a JSON array of entries, each with `Config` (a path to the config
/// blob) and `RepoTags`. `Config` appears as either `<hex>.json` (docker) or
/// `blobs/sha256/<hex>` (podman's newer writer), so the digest is taken as the last
/// path segment with any `.json` suffix stripped — both spellings land on the same
/// hex.
#[cfg(feature = "backend-oci")]
fn parse_archive_manifest(
    bytes: &[u8],
) -> std::result::Result<(Vec<String>, String), ArchiveFault> {
    let value: serde_json::Value =
        serde_json::from_slice(bytes).map_err(|e| ArchiveFault::Malformed(e.to_string()))?;
    let entries = value
        .as_array()
        .ok_or_else(|| ArchiveFault::Malformed("top level is not a JSON array".into()))?;
    if entries.is_empty() {
        return Err(ArchiveFault::Malformed(
            "manifest.json is an empty array — the archive names no image".into(),
        ));
    }

    let mut tags: Vec<String> = Vec::new();
    for entry in entries {
        if let Some(list) = entry.get("RepoTags").and_then(|t| t.as_array()) {
            tags.extend(list.iter().filter_map(|t| t.as_str()).map(str::to_string));
        }
    }

    let config = entries[0]
        .get("Config")
        .and_then(|c| c.as_str())
        .ok_or_else(|| ArchiveFault::Malformed("first entry has no Config".into()))?;
    let digest = config
        .rsplit('/')
        .next()
        .unwrap_or(config)
        .trim_end_matches(".json")
        .to_string();
    if digest.is_empty() {
        return Err(ArchiveFault::Malformed(format!(
            "Config {config:?} yields no digest"
        )));
    }
    Ok((tags, digest))
}

/// **Drive an image export over the [`ImageDaemon`] seam.** Presence is probed
/// FIRST, so "you never built this image" and "your container engine is down" stay
/// two different sentences and neither one leaves a stub file behind. Then the
/// archive is streamed to `dest` and immediately read back
/// ([`inspect_image_archive`]) — a stream that died mid-layer becomes an
/// [`ArchiveFault::Truncated`] here, at pack time, rather than a mystery on the
/// airgapped box a week later.
///
/// The archive's own tags are checked against `image`: an archive that does not
/// contain the tag it was asked for is a hard error, because the tag is where an
/// image's *pairing* lives (korp's `nordisk-falkordb-valkey:9-<date>` encodes both
/// the Valkey major and the FalkorDB module version, and a bundle silently carrying
/// the wrong one is worse than a bundle carrying nothing). An archive with NO tags
/// at all is accepted — an OCI-layout export identifies by digest alone, and the
/// digest is recorded.
///
/// Everything but the injected daemon is pure, so a mock proves all of it with no
/// socket.
#[cfg(feature = "backend-oci")]
fn export_image_on<D: ImageDaemon>(daemon: &D, image: &str, dest: &Path) -> Result<ImageArchive> {
    match daemon.inspect_image(image) {
        ImagePresence::Present => {}
        ImagePresence::Absent => {
            return Err(Error::Backend(format!(
                "export image {image}: no such image locally — nothing was written to {}",
                dest.display()
            )));
        }
        ImagePresence::Failed(msg) => {
            return Err(Error::Backend(format!(
                "export image {image}: container engine unreachable: {msg}"
            )));
        }
    }

    if let Some(parent) = dest.parent().filter(|p| !p.as_os_str().is_empty()) {
        std::fs::create_dir_all(parent).map_err(|e| {
            Error::Backend(format!("create export dir {}: {e}", parent.display()))
        })?;
    }
    {
        let file = std::fs::File::create(dest).map_err(|e| {
            Error::Backend(format!("create export file {}: {e}", dest.display()))
        })?;
        let mut sink = std::io::BufWriter::new(file);
        daemon.export_to(image, &mut sink)?;
        // Flush explicitly: a BufWriter dropped with a failing flush swallows the
        // error, which would hand back a short archive that looks written.
        std::io::Write::flush(&mut sink).map_err(|e| {
            Error::Backend(format!("flush export file {}: {e}", dest.display()))
        })?;
    }

    let archive = inspect_image_archive(dest)?;
    if !archive.tags.is_empty() && !archive.tags.iter().any(|t| t == image) {
        return Err(Error::Backend(format!(
            "export image {image}: the archive at {} carries {:?}, not {image} — refusing to \
             record an image under a tag it does not have",
            dest.display(),
            archive.tags
        )));
    }
    Ok(archive)
}

/// **Drive an image import over the [`ImageDaemon`] seam.** Verifies the archive
/// BEFORE handing it to the daemon, so a truncated or non-image file costs a few
/// seeks instead of a gigabyte pushed at a socket followed by a store in a state
/// nobody can describe. Returns what was loaded, read from the archive.
#[cfg(feature = "backend-oci")]
fn import_image_on<D: ImageDaemon>(daemon: &D, src: &Path) -> Result<ImageArchive> {
    let archive = inspect_image_archive(src)?;
    daemon.load_archive(src)?;
    Ok(archive)
}

// ---------------------------------------------------------------------------
// Socket reachability — WHY the API socket cannot be used, said honestly.
// ---------------------------------------------------------------------------
//
// This whole section exists because of one line that used to read:
//
//     if !std::path::Path::new(path).exists() { "…socket not found at {path}…" }
//
// `Path::exists()` is `fs::metadata(path).is_ok()`, so it collapses EVERY stat
// failure into "does not exist". **EACCES and ENOENT are not the same fact and
// they do not have the same fix**, and reporting the first as the second sends
// the reader to enable a unit that is already running.
//
// MEASURED on oden 2026-08-11 — the misreport that cost the debugging time:
//
//     draupnir: backend error: podman/Docker API socket not found at
//     /run/podman/podman.sock (enable with `systemctl --user enable --now
//     podman.socket`, …)
//
// while at the same moment `systemctl is-active podman.socket` said **active**,
// `sudo podman ps` listed a running container, and the socket was right there:
// `/run/podman/podman.sock` `root root 0660` inside `/run/podman` `root root
// 0700`. korp-server runs `User=korp` (uid 111), so the stat was refused at the
// DIRECTORY, and `exists()` turned "you may not look" into "it is not there".
//
// Two further defects in that one message, both fixed here:
//
//   * **Scope.** It always said `systemctl --user`, even when the path came from
//     a `DOCKER_HOST` naming a SYSTEM socket (`/run/podman/podman.sock`) — i.e.
//     wrong advice for precisely the configuration that produced it. A path under
//     `/run/user/<uid>/` is the rootless USER socket; anything else is the system
//     one, and the two have different unit scopes and different fixes.
//
//   * **Silence about ownership.** The one thing the reader needs for an EACCES
//     — who owns it, what group, what mode — was never printed, though the
//     process could stat the blocking component perfectly well.
//
// Everything below is pure std and **not** behind `backend-oci`, deliberately:
// the diagnosis is the part worth testing, and gating it would mean the lean
// default build never exercises it.

/// Owner, group and permission bits of one path component, with uid/gid resolved
/// to NAMES where `/etc/passwd` and `/etc/group` can resolve them.
///
/// Names are read out of those files with plain `std::fs` — no `getpwuid` FFI and
/// no `id`/`stat` subprocess (LAW: pure Rust, zero shell). An unresolvable id is
/// rendered as its decimal number, which is honest rather than blank.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Ownership {
    /// Owning user — name if resolvable, else the decimal uid.
    pub user: String,
    /// Owning group — name if resolvable, else the decimal gid.
    pub group: String,
    /// Permission bits only (`mode & 0o7777`).
    pub mode: u32,
}

impl std::fmt::Display for Ownership {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}:{} {:04o}", self.user, self.group, self.mode)
    }
}

/// Resolve a numeric id to a name out of a colon-separated `/etc/passwd`-style
/// database (`name:x:id:…`), falling back to the decimal id.
///
/// One parser for both files because `/etc/passwd` and `/etc/group` agree on the
/// first three fields. Reading the file is the pure-Rust route; the FFI
/// (`getpwuid_r`) is the thing the LAW keeps for last resort, and it is not
/// needed for a diagnostic string.
fn name_for_id(db: &str, id: u32) -> Option<String> {
    let text = std::fs::read_to_string(db).ok()?;
    for line in text.lines() {
        let mut f = line.split(':');
        let name = f.next()?;
        let _passwd = f.next();
        if f.next().and_then(|n| n.trim().parse::<u32>().ok()) == Some(id) {
            return Some(name.to_string());
        }
    }
    None
}

/// Ownership of an already-stat'd path, names resolved where possible.
fn ownership_of(md: &std::fs::Metadata) -> Ownership {
    use std::os::unix::fs::MetadataExt;
    let (uid, gid) = (md.uid(), md.gid());
    Ownership {
        user: name_for_id("/etc/passwd", uid).unwrap_or_else(|| uid.to_string()),
        group: name_for_id("/etc/group", gid).unwrap_or_else(|| gid.to_string()),
        mode: md.mode() & 0o7777,
    }
}

/// **What a stat of the container API socket path actually found** — the fact the
/// old `Path::exists()` threw away.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SocketProbe {
    /// The path stats. The socket is there and this process can at least reach it.
    Reachable,
    /// **ENOENT** — genuinely absent. Nothing is listening and nothing created it.
    Absent,
    /// **EACCES/EPERM** — something on the path is there and this process may not
    /// get past it. `blocked_at` is the deepest component that DID stat (i.e. the
    /// one whose permissions are doing the blocking, or the socket itself), with
    /// its ownership when it could be read.
    Denied {
        /// The path component the probe was stopped at.
        blocked_at: String,
        /// Its owner/group/mode, when the component could be stat'd.
        owner: Option<Ownership>,
    },
    /// Any other stat failure (ELOOP, ENOTDIR, EIO …) — carried verbatim rather
    /// than folded into one of the above, because guessing is what this whole
    /// section is a fix for.
    Other(String),
}

/// Stat `path` and classify the outcome — the honest replacement for
/// `Path::exists()`.
///
/// On `PermissionDenied` it walks UP the ancestors to find the deepest component
/// it *can* stat, because that component is the one whose mode is refusing the
/// traversal. On oden that turns a useless "not found at
/// /run/podman/podman.sock" into "refused at /run/podman (root:root 0700)",
/// which names the actual obstacle.
pub fn probe_socket_path(path: &Path) -> SocketProbe {
    match std::fs::metadata(path) {
        Ok(_) => SocketProbe::Reachable,
        Err(e) => match e.kind() {
            std::io::ErrorKind::NotFound => SocketProbe::Absent,
            std::io::ErrorKind::PermissionDenied => {
                // The first ancestor that stats is the gatekeeper: everything
                // below it was invisible precisely because of its mode.
                for anc in path.ancestors().skip(1) {
                    if let Ok(md) = std::fs::metadata(anc) {
                        return SocketProbe::Denied {
                            blocked_at: anc.display().to_string(),
                            owner: Some(ownership_of(&md)),
                        };
                    }
                }
                SocketProbe::Denied {
                    blocked_at: path.display().to_string(),
                    owner: None,
                }
            }
            _ => SocketProbe::Other(e.to_string()),
        },
    }
}

/// **Is this the rootless USER socket or the SYSTEM one?** — the distinction the
/// old hint ignored, and got backwards for the exact path that produced it.
///
/// A rootless podman socket lives under `/run/user/<uid>/`, is owned by that user
/// and is managed by that user's `systemd --user`. Anything else (`/run/podman/…`,
/// `/var/run/docker.sock`) belongs to the SYSTEM manager, where `systemctl --user`
/// is not merely unhelpful but addresses a different init instance entirely.
pub fn is_user_scope_socket(path: &Path) -> bool {
    path.starts_with("/run/user/") || path.starts_with("/var/run/user/")
}

/// The `systemctl` invocation that would actually apply to THIS socket's scope.
fn enable_hint(path: &Path) -> &'static str {
    if is_user_scope_socket(path) {
        "systemctl --user enable --now podman.socket"
    } else {
        "sudo systemctl enable --now podman.socket"
    }
}

/// **Turn a probe into the message a reader can act on** — `None` when the socket
/// is reachable, `Some(reason)` when it is not.
///
/// Split out from the stat so the wording is testable with no filesystem at all:
/// every branch here is a pure function of the probe.
pub fn socket_unreachable_reason(path: &Path, probe: &SocketProbe) -> Option<String> {
    let p = path.display();
    match probe {
        SocketProbe::Reachable => None,
        SocketProbe::Absent => Some(format!(
            "podman/Docker API socket not found at {p} — nothing exists at that path \
             (enable with `{}`, or point DOCKER_HOST at a running socket)",
            enable_hint(path)
        )),
        SocketProbe::Denied { blocked_at, owner } => {
            let mut m = format!(
                "podman/Docker API socket at {p} EXISTS but this process may not reach it \
                 (permission denied) — do NOT enable the unit, it is already there. \
                 Refused at {blocked_at}"
            );
            match owner {
                Some(o) => {
                    m.push_str(&format!(" ({o})."));
                    // The honest split: a group other than root is something a
                    // service declaration can actually join. `root` is not — saying
                    // "add yourself to root" would be advice nobody should take.
                    if o.group == "root" || o.group == "0" {
                        m.push_str(
                            " Its group is root, so joining a group CANNOT fix this: \
                             either the host gives the socket a non-root group \
                             (a `SocketGroup=` drop-in on podman.socket, a HOST decision) \
                             or the service runs as root. Adding a service account to the \
                             root group is not a fix.",
                        );
                    } else {
                        m.push_str(&format!(
                            " Grant access by making the service account a member of `{}` — \
                             for a systemd unit that is `SupplementaryGroups={}`, \
                             which skidbladnir renders from the service's own declaration.",
                            o.group, o.group
                        ));
                    }
                }
                None => m.push('.'),
            }
            Some(m)
        }
        SocketProbe::Other(e) => Some(format!(
            "podman/Docker API socket at {p} could not be examined: {e}"
        )),
    }
}

/// The one call the engine makes: stat, classify, and word it. `None` ⇒ go ahead.
pub fn socket_unreachable(path: &Path) -> Option<String> {
    socket_unreachable_reason(path, &probe_socket_path(path))
}

// ---------------------------------------------------------------------------
// ROOTLESS ENGINE READINESS — what the engine REQUIRES of the host, and which
// half of it draupnir owns.
// ---------------------------------------------------------------------------
//
// The 2026-08-11 oden setup took five hand-typed commands to get korp-server
// (`User=korp`, uid 111, `nologin`) a podman socket. They split cleanly in two,
// and the split is the design:
//
//   HOST / ACCOUNT — `loginctl enable-linger korp`, `/etc/subuid`, `/etc/subgid`.
//     Nothing to do with containers. They are logind and shadow-utils facts about
//     an ACCOUNT, they need root, and skidbladnir already owns account and service
//     lifecycle and already drives systemd over D-Bus. It performs them
//     (`skidbladnir::rootless`); draupnir does not, and must not grow a second
//     `/etc/subuid` writer.
//
//   ENGINE — where the socket is, whether it answers, whether `podman system
//     migrate` is owed. These are facts about podman, which is this module's
//     subject. draupnir STATES them and VERIFIES them; it mutates nothing.
//
// So: skidbladnir sets the host up, draupnir says what "set up" has to mean and
// checks afterwards whether it took. Neither repeats the other, and the two facts
// they share (the uid, and the socket path derived from it) are derived by the
// same arithmetic on both sides rather than passed between them.

/// **Where a rootless podman puts its API socket for `uid`.**
///
/// One place, so that a service unit's `DOCKER_HOST`, this engine's connect path
/// and skidbladnir's post-setup verification cannot disagree by a directory. On
/// oden this resolved to `/run/user/111/podman/podman.sock` — VERIFIED as
/// `srw-rw---- korp korp` once lingering was on.
pub fn rootless_socket_path(uid: u32) -> std::path::PathBuf {
    std::path::PathBuf::from(format!("/run/user/{uid}/podman/podman.sock"))
}

/// **The uid this process is actually running as**, read from `/proc/self/status`.
///
/// Zero shell (no `id -u`) and no FFI (no `getuid()`), per the LAW — the kernel
/// publishes it as text and `std::fs` can read text. The `Uid:` line is
/// `real effective saved filesystem`; the REAL uid is taken, because that is the
/// uid logind created `/run/user/<uid>` for.
///
/// `None` off Linux or where `/proc` is not mounted — in which case a caller must
/// be told the uid rather than guess one, which is precisely the defect this
/// replaced: [`Engine::socket_url`] fell back to a hardcoded `/run/user/1000`, so
/// a service running as uid 111 with no `DOCKER_HOST` looked for its socket in
/// *another account's* runtime directory and got EACCES or ENOENT depending on
/// that account's luck.
pub fn current_uid() -> Option<u32> {
    let status = std::fs::read_to_string("/proc/self/status").ok()?;
    parse_uid_line(&status)
}

/// Pull the real uid out of `/proc/<pid>/status` text (pure → testable).
fn parse_uid_line(status: &str) -> Option<u32> {
    for line in status.lines() {
        if let Some(rest) = line.strip_prefix("Uid:") {
            return rest.split_whitespace().next()?.parse().ok();
        }
    }
    None
}

/// **Is `podman system migrate` owed for this account?**
///
/// It is the one command from the oden session draupnir will NOT run, and saying
/// so plainly is better than a wrapper that pretends. Two independent reasons:
///
/// 1. **There is no API for it.** The libpod REST surface exposes
///    `/libpod/system/{info,df,prune,events,version}` and no `migrate`. It exists
///    only in the CLI, so "running" it means `Command::new("podman")` — a
///    shell-out under the LAW, and this crate deleted its last two of those (the
///    `podman build`/`podman run -v` subprocesses) to get here.
/// 2. **It is usually not owed at all.** What it does is re-create the user
///    namespace mapping recorded for EXISTING containers and pods. An account that
///    has never run podman has no stored mappings, so a fresh account picks the new
///    `/etc/subuid` range up on its first run and needs nothing. It is owed only
///    when podman ran BEFORE the subordinate-id grant — which is exactly what
///    happened on oden, and is exactly why it was needed there.
///
/// The detection is therefore "has podman ever run as this account", i.e. does its
/// libpod storage exist. That is a fact on disk, not a guess.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MigrateVerdict {
    /// podman has no stored state for this account: the next run adopts the new
    /// range by itself.
    NotOwed(String),
    /// podman has run here before. If that was before the subordinate-id grant, the
    /// stored mappings are stale and only `podman system migrate` refreshes them.
    Owed(String),
}

impl MigrateVerdict {
    pub fn detail(&self) -> &str {
        match self {
            MigrateVerdict::NotOwed(s) | MigrateVerdict::Owed(s) => s,
        }
    }
}

/// Decide from the ONE fact that decides it: does this account already have libpod
/// storage. Pure — `home` and the existence flag are passed in — so both arms are
/// testable without a podman install.
pub fn migrate_verdict(account: &str, storage_exists: bool) -> MigrateVerdict {
    if !storage_exists {
        MigrateVerdict::NotOwed(format!(
            "podman has never run as `{account}` (no libpod storage), so there are no \
             stored user-namespace mappings to refresh — the first run adopts the \
             /etc/subuid grant by itself. `podman system migrate` is not needed."
        ))
    } else {
        MigrateVerdict::Owed(format!(
            "podman has already run as `{account}` and holds stored user-namespace \
             mappings. If the /etc/subuid grant was made AFTER that, they are stale and \
             every container keeps the old single-id mapping. Only `podman system \
             migrate`, run AS `{account}`, refreshes them — and draupnir will not run \
             it: there is no libpod REST endpoint for it, so running it means a podman \
             subprocess, which this crate does not do. Run it once, as the service \
             account, or delete the account's containers and let them be re-created."
        ))
    }
}

/// The libpod storage directory for an account — the path [`migrate_verdict`] asks
/// about. `XDG_DATA_HOME` is not consulted: a `nologin` service account has no
/// session to set it, which is the whole shape of this problem.
pub fn libpod_storage_dir(home: &str) -> std::path::PathBuf {
    std::path::Path::new(home).join(".local/share/containers/storage/libpod")
}

/// **Everything the engine needs the host to have already done**, checked.
///
/// This is the "did it take?" side of the seam: skidbladnir applies the host setup,
/// then this says whether a rootless engine can now actually be reached for that
/// uid, and what the remaining obstacle is if not.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RootlessReadiness {
    pub uid: u32,
    pub socket: std::path::PathBuf,
    pub probe: SocketProbe,
    /// `None` when the engine is reachable; otherwise the obstacle, worded for a
    /// reader — including the ENOENT/EACCES distinction the rest of this section
    /// exists to preserve.
    pub obstacle: Option<String>,
}

impl RootlessReadiness {
    pub fn is_ready(&self) -> bool {
        self.obstacle.is_none()
    }
}

/// Probe the rootless socket for `uid` and word the result.
///
/// Note what this does NOT say. An `Absent` rootless socket has a *different* first
/// question from an absent system socket: not "is the unit enabled" but "does
/// `/run/user/<uid>` exist at all", because without lingering it does not, and
/// `systemctl --user` inside it has nothing to talk to. So the wording branches on
/// the runtime directory before it branches on the unit.
pub fn rootless_readiness(uid: u32) -> RootlessReadiness {
    let socket = rootless_socket_path(uid);
    let probe = probe_socket_path(&socket);
    let runtime_dir = std::path::PathBuf::from(format!("/run/user/{uid}"));
    let obstacle = match &probe {
        SocketProbe::Reachable => None,
        SocketProbe::Absent if !runtime_dir.exists() => Some(format!(
            "no rootless engine for uid {uid}: {} does not exist, so there is no user \
             manager to enable `podman.socket` in and nowhere for the socket to live. \
             The account must LINGER first — `skidbladnir::rootless` performs that over \
             org.freedesktop.login1 — and the runtime directory then appears by itself.",
            runtime_dir.display()
        )),
        _ => socket_unreachable_reason(&socket, &probe),
    };
    RootlessReadiness {
        uid,
        socket,
        probe,
        obstacle,
    }
}

// ---------------------------------------------------------------------------
// 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), else the
    /// runtime directory of the uid this process is ACTUALLY running as.
    ///
    /// That last fallback used to be the literal string `/run/user/1000`, and it is
    /// wrong for precisely the deployment this engine exists in. A systemd system
    /// unit with `User=korp` runs as uid 111 and is given NO `XDG_RUNTIME_DIR` (only
    /// a session or a `--user` manager sets one), so with no `DOCKER_HOST` the
    /// engine went looking in **uid 1000's** runtime directory — another account's
    /// `/run/user/1000`, mode 0700. On a box where that account is logged in the
    /// answer is EACCES, on a box where it is not the answer is ENOENT, and neither
    /// mentions the actual mistake. `1000` is a desktop's first user, not a fact
    /// about the process.
    fn socket_url() -> String {
        if let Ok(h) = std::env::var("DOCKER_HOST") {
            return h;
        }
        if let Ok(xdg) = std::env::var("XDG_RUNTIME_DIR") {
            return format!("unix://{xdg}/podman/podman.sock");
        }
        match current_uid() {
            Some(uid) => format!("unix://{}", rootless_socket_path(uid).display()),
            // No /proc: say which socket is meant rather than guess a uid. The
            // system socket is the only one whose path does not depend on an
            // identity we could not read.
            None => "unix:///run/podman/podman.sock".to_string(),
        }
    }

    /// 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);
        // ENOENT and EACCES are different facts with different fixes — see the
        // `SocketProbe` section above for the oden measurement that forced this
        // apart. `Path::exists()` reported the second as the first.
        if url.starts_with("unix://") {
            if let Some(reason) = socket_unreachable(Path::new(path)) {
                return Err(Error::Backend(reason));
            }
        }
        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>,
        security_opt: &[String],
    ) -> 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()
            && security_opt.is_empty()
        {
            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,
                // Verbatim, and `None` when empty: an empty `Vec` here would
                // render `"SecurityOpt": []` and stop being byte-identical to a
                // create body that never named one.
                security_opt: if security_opt.is_empty() {
                    None
                } else {
                    Some(security_opt.to_vec())
                },
                ..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>,
        security_opt: &[String],
    ) -> 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,
                security_opt,
            );
            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.
        //
        // ROOT-LAW #0 SANCTIONED EXCEPTION: this is `tokio::spawn`, but the task is
        // pure blocking-IO — it parks on `stream.next().await` (a docker `/logs`
        // follow socket) and only trims/pushes each line into a `Mutex<Vec<..>>`.
        // No CPU fan-out, no compute. ROOT-LAW #0 forbids `tokio::spawn`-for-CPU
        // and bare `std::thread::spawn`, NOT an IO task on the backend's own async
        // runtime; a `gatling::background::Job` would be wrong here — it wants an
        // OS thread blocking on the async stream, re-entering this very runtime.
        // So it correctly stays an async task on `self.rt`.
        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`. The
    /// error-detail→fail + returned-tag decision lives in [`build_image_on`] (daemon-
    /// free, mock-tested); this engine only supplies the live `/build` stream.
    fn build_image(&self, context_dir: &Path, containerfile: &str, tag: &str) -> Result<String> {
        build_image_on(self, context_dir, containerfile, tag)
    }

    /// 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.
    /// The download+create+force-remove is [`download_path_tar`](Self::download_path_tar);
    /// the unpack + basename layout is [`extract_path_on`] (daemon-free, mock-tested).
    fn extract_path(&self, image: &str, container_path: &str, host_dest: &Path) -> Result<()> {
        extract_path_on(self, image, container_path, host_dest)
    }

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

    /// Every container whose name begins with `prefix`, over bollard's
    /// `list_containers` — the live drive behind
    /// [`ContainerControl::list_containers`].
    ///
    /// `all(true)`, so exited and created containers come back too: a sweep that
    /// saw only running containers would leave precisely the corpses it exists
    /// to remove.
    ///
    /// The name filter is applied **here** rather than handed to the daemon.
    /// Docker's and podman's `name=` filter is a *substring* match, not a
    /// prefix, so `name=foo-` also returns `bar-foo-1`; a janitor that removed
    /// on that basis would delete another owner's containers. The daemon's
    /// filter is a bandwidth optimisation with the wrong semantics, so this asks
    /// for the list and matches the prefix itself.
    ///
    /// Names come back from the daemon with a leading `/`, which is stripped.
    ///
    /// ⚠ `prefix` matches the **daemon-side** name, and that is not the name in
    /// the [`BootSpec`]: [`Machine::container_name`] prepends `draupnir-`, and
    /// `machine.id` is the result. `list_containers(&spec.name)` matches nothing.
    /// Pass `&machine.id`, or a `draupnir-…` prefix. Caught by this crate's own
    /// live test on the first run.
    ///
    /// [`BootSpec`]: crate::BootSpec
    fn list_containers(&self, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
        use bollard::query_parameters::ListContainersOptionsBuilder;
        self.rt.block_on(async {
            let summaries = self
                .docker
                .list_containers(Some(ListContainersOptionsBuilder::new().all(true).build()))
                .await
                .map_err(|e| Error::Backend(format!("listing containers: {e}")))?;
            let mut out = Vec::new();
            for c in summaries {
                for name in c.names.iter().flatten() {
                    let name = name.strip_prefix('/').unwrap_or(name);
                    if !name.starts_with(prefix) {
                        continue;
                    }
                    // `state` is the daemon's own word for it. Anything that is
                    // not plainly running is reported through the exit code the
                    // summary carries, so a caller can tell "still working" from
                    // "finished, badly" without a second round trip.
                    let state = match c.state {
                        Some(bollard::models::ContainerSummaryStateEnum::RUNNING) => {
                            ContainerState::Running
                        }
                        Some(bollard::models::ContainerSummaryStateEnum::EXITED) => {
                            ContainerState::Exited(0)
                        }
                        _ => ContainerState::Gone,
                    };
                    out.push((name.to_owned(), state));
                }
            }
            // By name; `ContainerState` has no ordering and does not need one
            // for a janitor's stable output.
            out.sort_by(|a, b| a.0.cmp(&b.0));
            Ok(out)
        })
    }

    /// The container's configured environment, over bollard's
    /// `inspect_container` — the live drive behind
    /// [`ContainerControl::container_env`].
    ///
    /// The daemon's record of what the process was started with, not the spec it
    /// was asked for. A container with no environment at all answers an empty
    /// vector; a container that is not there is an [`Error::Backend`], because
    /// "no variables" and "no container" are different answers and a read-back
    /// that conflated them would pass on a container that had vanished.
    fn container_env(&self, name: &str) -> Result<Vec<String>> {
        use bollard::query_parameters::InspectContainerOptions;
        self.rt.block_on(async {
            let info = self
                .docker
                .inspect_container(name, None::<InspectContainerOptions>)
                .await
                .map_err(|e| Error::Backend(format!("inspecting container `{name}`: {e}")))?;
            Ok(info
                .config
                .and_then(|c| c.env)
                .unwrap_or_default()
                .into_iter()
                .collect())
        })
    }

    /// **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(feature = "backend-oci")]
impl ImageDaemon for Engine {
    /// Drive the live `/build` stream, draining each `BuildInfo` into the
    /// daemon-free shape [`build_image_on`] decides on: `Ok(None)` for a progress
    /// line, `Ok(Some(msg))` when the `BuildInfo` carried an `error_detail` (its
    /// `message`, empty string if the detail had none — parity with the previous
    /// `unwrap_or_default()`), and `Err` for a transport failure.
    fn build_stream(
        &self,
        context: Vec<u8>,
        containerfile: &str,
        tag: &str,
    ) -> Vec<Result<Option<String>>> {
        use bollard::body_full;
        use bollard::query_parameters::BuildImageOptionsBuilder;
        use futures::StreamExt;

        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(context.into())));
            let mut steps: Vec<Result<Option<String>>> = Vec::new();
            while let Some(item) = stream.next().await {
                steps.push(match item {
                    Ok(info) => Ok(info.error_detail.map(|d| d.message.unwrap_or_default())),
                    Err(e) => Err(Error::Backend(format!("build image {tag}: {e}"))),
                });
            }
            steps
        })
    }

    /// Create a throwaway container from `image`, stream a tar of `container_path`
    /// from the daemon, and force-remove the container afterward **even on a
    /// download error** (best-effort, no leak). Returns the raw (basename-rooted)
    /// tar bytes [`extract_path_on`] unpacks.
    /// Resolve `image` against the daemon's local store via `inspect_image`. A 404
    /// (and podman's "no such image" text, which it can return under other codes)
    /// is [`ImagePresence::Absent`]; every other error is
    /// [`ImagePresence::Failed`] so a dead socket never reads as a missing image.
    fn inspect_image(&self, image: &str) -> ImagePresence {
        let docker = &self.docker;
        self.rt.block_on(async {
            match docker.inspect_image(image).await {
                Ok(_) => ImagePresence::Present,
                Err(bollard::errors::Error::DockerResponseServerError {
                    status_code: 404, ..
                }) => ImagePresence::Absent,
                Err(bollard::errors::Error::DockerResponseServerError { message, .. })
                    if message.to_ascii_lowercase().contains("no such image") =>
                {
                    ImagePresence::Absent
                }
                Err(e) => ImagePresence::Failed(e.to_string()),
            }
        })
    }

    /// Live `/images/{name}/get`, drained chunk-by-chunk into `sink`. bollard hands
    /// back a `Stream<Item = Result<Bytes, _>>`; each chunk is written and dropped,
    /// so peak memory is one chunk regardless of image size. `block_on` keeps this
    /// inside the engine's own runtime, matching every other call here.
    fn export_to(&self, image: &str, sink: &mut dyn std::io::Write) -> Result<u64> {
        use futures::StreamExt;

        let docker = &self.docker;
        self.rt.block_on(async move {
            let mut stream = docker.export_image(image);
            let mut written: u64 = 0;
            while let Some(chunk) = stream.next().await {
                let chunk =
                    chunk.map_err(|e| Error::Backend(format!("export image {image}: {e}")))?;
                sink.write_all(chunk.as_ref()).map_err(|e| {
                    Error::Backend(format!("write exported image {image} to sink: {e}"))
                })?;
                written = written.saturating_add(chunk.len() as u64);
            }
            Ok(written)
        })
    }

    /// Live `/images/load` over bollard's STREAMING import. The body is a
    /// `try_unfold` over an owned `File` handed out in [`IMPORT_CHUNK`]-sized pieces
    /// — the streaming variant is required here, because the non-streaming
    /// `import_image` wants the entire archive as one `Bytes` (1.4 GB resident for
    /// the Spark image).
    ///
    /// The `File::read` inside the async body is a blocking read on a runtime
    /// worker. That is deliberate and matches the rest of this engine: draupnir
    /// pulls in `tokio` with `rt-multi-thread` only — no `fs` feature, no
    /// `tokio-util` — and adding an async-fs stack to move a local file into a local
    /// socket would buy nothing. One chunk is in flight at a time either way.
    fn load_archive(&self, src: &Path) -> Result<()> {
        use bollard::query_parameters::ImportImageOptionsBuilder;
        use futures::StreamExt;
        use std::io::Read;

        let file = std::fs::File::open(src)
            .map_err(|e| Error::Backend(format!("open image archive {}: {e}", src.display())))?;
        let label = src.display().to_string();
        let docker = &self.docker;
        self.rt.block_on(async move {
            let body = futures::stream::try_unfold(file, |mut file| async move {
                let mut buf = vec![0u8; IMPORT_CHUNK];
                let n = file.read(&mut buf)?;
                if n == 0 {
                    Ok::<_, std::io::Error>(None)
                } else {
                    buf.truncate(n);
                    Ok(Some((bytes::Bytes::from(buf), file)))
                }
            });
            let mut stream =
                docker.import_image_stream(ImportImageOptionsBuilder::new().build(), body, None);
            while let Some(item) = stream.next().await {
                item.map_err(|e| Error::Backend(format!("load image archive {label}: {e}")))?;
            }
            Ok(())
        })
    }

    fn download_path_tar(&self, image: &str, container_path: &str) -> Result<Vec<u8>> {
        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());

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

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

    // ── The socket diagnosis: EACCES is not ENOENT ───────────────────────────
    //
    // These are pure-std and run in the LEAN default build, deliberately: the
    // wording is the part that misled a reader for an afternoon, and gating it
    // behind `backend-oci` would mean the default `cargo test` never sees it.

    /// **A socket that is PRESENT but refused must not be reported as missing.**
    ///
    /// This is the exact oden shape: `/run/podman` is `root:root 0700`, the socket
    /// inside it is `root:root 0660`, and `korp` (uid 111) stats the socket path
    /// and gets EACCES at the directory. `Path::exists()` returned `false` and the
    /// engine said "socket not found … enable with `systemctl --user enable --now
    /// podman.socket`" — while `systemctl is-active podman.socket` said **active**.
    ///
    /// RED direction: fold `Denied` back into the `Absent` arm of
    /// [`socket_unreachable_reason`] (i.e. what `Path::exists()` did) and both the
    /// "not `not found`" and the "does not tell you to enable it" assertions fail.
    #[test]
    fn a_permission_denied_socket_is_not_reported_as_missing() {
        let path = Path::new("/run/podman/podman.sock");
        let probe = SocketProbe::Denied {
            blocked_at: "/run/podman".into(),
            owner: Some(Ownership {
                user: "root".into(),
                group: "root".into(),
                mode: 0o700,
            }),
        };
        let m = socket_unreachable_reason(path, &probe).expect("denied is not reachable");

        assert!(
            !m.contains("not found"),
            "EACCES must not be worded as ENOENT — that is the whole defect: {m}"
        );
        assert!(
            m.contains("EXISTS") && m.contains("permission denied"),
            "it must say the socket is there and the process is refused: {m}"
        );
        assert!(
            !m.contains("enable --now"),
            "never advise enabling a unit that is already running: {m}"
        );
        // The obstacle, named — owner, group and mode of the component that refused.
        assert!(
            m.contains("/run/podman") && m.contains("root:root") && m.contains("0700"),
            "it must name the blocking component and its ownership: {m}"
        );
        // …and it must NOT propose joining the root group, which is not a fix.
        assert!(
            m.contains("SocketGroup=") && m.contains("HOST decision"),
            "a root-group socket needs a host decision, and the message must say so: {m}"
        );
        assert!(
            !m.contains("SupplementaryGroups=root"),
            "'add the service to the root group' is advice nobody should take: {m}"
        );
    }

    /// A **genuinely absent** socket keeps the old, correct advice — the fix must
    /// not blur the two directions.
    ///
    /// RED direction: report `Absent` with the `Denied` wording and the "not found"
    /// assertion fails.
    #[test]
    fn an_absent_socket_still_says_it_is_absent_and_how_to_enable_it() {
        let path = Path::new("/run/podman/podman.sock");
        let m = socket_unreachable_reason(path, &SocketProbe::Absent).expect("absent");
        assert!(m.contains("not found"), "ENOENT is genuinely 'not found': {m}");
        assert!(
            m.contains("enable --now podman.socket"),
            "and enabling the unit is the right advice here: {m}"
        );
    }

    /// **The SCOPE bug.** The old hint always said `systemctl --user`, including
    /// for a `DOCKER_HOST` naming the SYSTEM socket — wrong advice for exactly the
    /// configuration that produced it. Scope follows the PATH: `/run/user/<uid>/…`
    /// is the rootless user socket, anything else is the system manager's.
    ///
    /// RED direction: hardcode `systemctl --user …` in `enable_hint` and the system
    /// assertion fails while the user one still passes — the silent half.
    #[test]
    fn the_enable_hint_matches_the_sockets_scope() {
        let system = Path::new("/run/podman/podman.sock");
        let user = Path::new("/run/user/1000/podman/podman.sock");

        assert!(!is_user_scope_socket(system), "/run/podman is system scope");
        assert!(is_user_scope_socket(user), "/run/user/1000 is user scope");

        let sys_msg = socket_unreachable_reason(system, &SocketProbe::Absent).unwrap();
        assert!(
            sys_msg.contains("sudo systemctl enable") && !sys_msg.contains("--user"),
            "a SYSTEM socket must not be pointed at `systemctl --user`: {sys_msg}"
        );

        let usr_msg = socket_unreachable_reason(user, &SocketProbe::Absent).unwrap();
        assert!(
            usr_msg.contains("systemctl --user enable"),
            "a rootless socket keeps the --user scope: {usr_msg}"
        );
    }

    /// A socket group that is NOT root is a real, declarable fix, and the message
    /// must name it and the directive that grants it.
    #[test]
    fn a_non_root_socket_group_is_advised_as_a_supplementary_group() {
        let m = socket_unreachable_reason(
            Path::new("/run/podman/podman.sock"),
            &SocketProbe::Denied {
                blocked_at: "/run/podman/podman.sock".into(),
                owner: Some(Ownership {
                    user: "root".into(),
                    group: "podman".into(),
                    mode: 0o660,
                }),
            },
        )
        .unwrap();
        assert!(
            m.contains("SupplementaryGroups=podman"),
            "a joinable group IS the fix, and the directive must be named: {m}"
        );
        assert!(
            !m.contains("HOST decision"),
            "this one does not need a host decision — the group already exists: {m}"
        );
    }

    /// The probe itself, against the real filesystem: a path under a directory that
    /// certainly does not exist is `Absent`, and a path that does exist is
    /// `Reachable`. (The `Denied` branch needs a 0700 directory owned by *another*
    /// uid, which a test running as one unprivileged user cannot manufacture — its
    /// wording is covered purely above, and its stat behaviour was VERIFIED on oden.)
    #[test]
    fn probe_separates_absent_from_reachable_on_the_real_filesystem() {
        assert_eq!(
            probe_socket_path(Path::new(
                "/nonexistent-draupnir-probe/podman/podman.sock"
            )),
            SocketProbe::Absent
        );
        assert_eq!(
            probe_socket_path(Path::new("/etc/passwd")),
            SocketProbe::Reachable
        );
        assert!(
            socket_unreachable(Path::new("/etc/passwd")).is_none(),
            "a reachable path yields no complaint"
        );
    }

    /// uid/gid → name resolution reads the passwd/group databases with `std::fs` —
    /// no `getpwuid` FFI, no `id`/`stat` subprocess. uid 0 is `root` on every Linux
    /// host; an id nobody owns falls back to its decimal form rather than blank.
    #[test]
    fn ids_resolve_to_names_without_ffi_or_a_subprocess() {
        assert_eq!(name_for_id("/etc/passwd", 0).as_deref(), Some("root"));
        assert_eq!(name_for_id("/etc/group", 0).as_deref(), Some("root"));
        assert_eq!(name_for_id("/etc/passwd", 4_294_967_294), None);
    }

    // ── Rootless engine readiness: the ENGINE half of the oden setup ─────────

    /// **The socket path is arithmetic on the uid, in ONE place.**
    ///
    /// oden's, VERIFIED: `/run/user/111/podman/podman.sock`, `srw-rw---- korp korp`
    /// once lingering was on. Three parties need this string — the unit's
    /// `DOCKER_HOST`, this engine's connect, skidbladnir's post-setup check — and
    /// three spellings of it is how they end up disagreeing by a directory.
    #[test]
    fn the_rootless_socket_path_is_derived_from_the_uid() {
        assert_eq!(
            rootless_socket_path(111),
            Path::new("/run/user/111/podman/podman.sock")
        );
        assert!(is_user_scope_socket(&rootless_socket_path(111)));
    }

    /// **The uid comes from `/proc/self/status`, not from `id -u` and not from FFI.**
    ///
    /// The `Uid:` line is `real effective saved filesystem`; the real uid is the one
    /// logind created `/run/user/<uid>` for.
    #[test]
    fn the_current_uid_is_read_from_proc_without_shell_or_ffi() {
        assert_eq!(parse_uid_line("Name:\tsh\nUid:\t111\t111\t111\t111\n"), Some(111));
        assert_eq!(parse_uid_line("Uid:\t0\t0\t0\t0"), Some(0));
        assert_eq!(parse_uid_line("Gid:\t111\t111\t111\t111\n"), None);
        // …and against the real /proc on this box.
        assert!(current_uid().is_some(), "Linux always publishes /proc/self/status");
    }

    /// **The `/run/user/1000` hardcode was a real defect, and this is its shape.**
    ///
    /// A systemd system unit with `User=korp` (uid 111) gets no `XDG_RUNTIME_DIR` —
    /// only a session or a `--user` manager sets one. With no `DOCKER_HOST` the old
    /// fallback therefore sent uid 111 into **uid 1000's** runtime directory, mode
    /// 0700: EACCES if that account is logged in, ENOENT if it is not, and neither
    /// message mentions the actual mistake. `1000` is a desktop's first user, never
    /// a fact about the running process.
    ///
    /// Asserted through the pure derivation rather than by setting process env,
    /// which a parallel test run cannot do safely.
    ///
    /// RED direction: put the literal back and `rootless_socket_path(current_uid())`
    /// stops matching what the engine would connect to on any non-1000 account.
    #[test]
    fn the_socket_fallback_follows_the_processs_own_uid_not_a_hardcoded_1000() {
        let uid = current_uid().expect("Linux");
        let derived = format!("unix://{}", rootless_socket_path(uid).display());
        assert!(
            derived.contains(&format!("/run/user/{uid}/")),
            "the fallback must name THIS process's runtime dir: {derived}"
        );
        // The specific regression: a service account is not uid 1000.
        assert_eq!(
            rootless_socket_path(111),
            Path::new("/run/user/111/podman/podman.sock"),
            "uid 111 must never be sent to /run/user/1000"
        );
    }

    /// **`podman system migrate` is REFUSED, and the refusal explains itself.**
    ///
    /// It is the one command from the oden session that has no libpod REST endpoint
    /// — it exists only in the CLI — so performing it would mean a podman
    /// subprocess, which this crate removed its last two of to get here. It is also
    /// usually not owed: it refreshes stored user-namespace mappings for containers
    /// that ALREADY EXIST, so a fresh account picks up a new /etc/subuid range by
    /// itself. Both facts have to be in the message or the reader will either run it
    /// needlessly or skip it when it matters.
    ///
    /// RED direction: return one verdict for both cases and the "not needed" arm
    /// starts telling every fresh account to run a command it does not need.
    #[test]
    fn system_migrate_is_owed_only_where_podman_already_ran_and_is_never_performed() {
        let fresh = migrate_verdict("korp", false);
        assert!(matches!(fresh, MigrateVerdict::NotOwed(_)), "{fresh:?}");
        assert!(
            fresh.detail().contains("not needed"),
            "a fresh account must be told to skip it: {}",
            fresh.detail()
        );

        // oden's case: podman had already run as korp before the subuid grant.
        let stale = migrate_verdict("korp", true);
        assert!(matches!(stale, MigrateVerdict::Owed(_)), "{stale:?}");
        let m = stale.detail();
        assert!(m.contains("podman system migrate"), "{m}");
        assert!(
            m.contains("as `korp`"),
            "it must say WHICH account has to run it — as root it does nothing: {m}"
        );
        assert!(
            m.contains("no libpod REST endpoint") && m.contains("will not run it"),
            "and it must say plainly that draupnir refuses, and why: {m}"
        );

        assert_eq!(
            libpod_storage_dir("/home/korp"),
            Path::new("/home/korp/.local/share/containers/storage/libpod")
        );
    }

    /// **A rootless socket that is absent because there is no runtime directory gets
    /// a DIFFERENT first question** from an absent system socket.
    ///
    /// "enable podman.socket" is useless advice when `/run/user/<uid>` does not
    /// exist: there is no user manager to enable it in. The account has to linger
    /// first, and then the directory appears by itself. This is the rootless twin of
    /// the EACCES/ENOENT fix — right fact, wrong next step.
    #[test]
    fn an_absent_rootless_socket_asks_about_lingering_before_the_unit() {
        // A uid that certainly has no runtime directory on any box.
        let r = rootless_readiness(4_294_967_200);
        assert!(!r.is_ready());
        let m = r.obstacle.expect("not ready ⇒ an obstacle");
        assert!(m.contains("does not exist"), "{m}");
        assert!(
            m.contains("LINGER") && m.contains("org.freedesktop.login1"),
            "the first obstacle is lingering, and it must name the mechanism: {m}"
        );
        assert!(
            !m.contains("systemctl --user enable"),
            "…and must NOT advise enabling a unit in a user manager that does not \
             exist yet: {m}"
        );
    }

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

    /// **A security option reaches `HostConfig.security_opt`, and an empty one
    /// mints nothing at all.**
    ///
    /// Both halves matter and they fail in opposite directions. Without the
    /// first, `BootSpec::with_security_opt` is a setter nobody reads and a
    /// caller that needs `io_uring` inside a container gets ENOSYS with the
    /// field set — the exact "the request is not the setting" shape this crate
    /// checks everywhere else. Without the second, every existing create body
    /// grows a `"SecurityOpt": []` and stops being byte-identical to what
    /// shipped, which is the change nobody asked for.
    ///
    /// Seen red by dropping the field from the `HostConfig` literal
    /// (`security_opt: None` unconditionally): "left: None, right:
    /// Some([\"seccomp=unconfined\"])". Seen red the other way by rendering
    /// `Some(security_opt.to_vec())` unguarded: the `bare` body below mints a
    /// `HostConfig` and `host_config.is_none()` fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn a_security_option_reaches_the_create_body_and_an_empty_one_changes_nothing() {
        let opts = vec!["seccomp=unconfined".to_string()];
        let with = Engine::create_body_full(
            "redis:7",
            &[],
            &[],
            &[],
            &[],
            &[],
            None,
            None,
            None,
            &opts,
        );
        let hc = with
            .host_config
            .clone()
            .expect("a security option alone mints a host config");
        assert_eq!(
            hc.security_opt.as_deref(),
            Some(opts.as_slice()),
            "the security option did not reach HostConfig.security_opt, so the daemon would \
             apply its default profile while the caller believes otherwise"
        );

        // …and nothing else moved: with the option stripped, the body is the
        // same bare, host-config-less body it has always been.
        let bare =
            Engine::create_body_full("redis:7", &[], &[], &[], &[], &[], None, None, None, &[]);
        assert!(
            bare.host_config.is_none(),
            "an empty security_opt minted a host config — every existing create body would \
             stop being byte-identical to what shipped"
        );

        // The builder is the door a caller actually uses, and it APPENDS:
        // `--security-opt` is a repeatable flag.
        let spec = crate::BootSpec::container("probe", "redis:7")
            .with_security_opt("seccomp=unconfined")
            .with_security_opt("no-new-privileges");
        assert_eq!(
            spec.security_opt,
            vec![
                "seccomp=unconfined".to_string(),
                "no-new-privileges".to_string()
            ],
            "with_security_opt replaced instead of appending"
        );

        crate::functional_status(
            "draupnir/container",
            "security_opt",
            hc.security_opt.is_some() && bare.host_config.is_none(),
            "BootSpec::with_security_opt renders --security-opt onto HostConfig.security_opt; empty stays byte-identical",
        );
    }

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

    // -----------------------------------------------------------------------
    // build_image / extract_path — driven over the ImageDaemon seam by a mock,
    // so the returned-tag happy path, the BuildInfo error_detail→Err path, and
    // the basename-flattened unpack are proven with NO daemon (the live bollard
    // /build + download_from_container wire is a Loki integration concern).
    // -----------------------------------------------------------------------

    /// A scripted [`ImageDaemon`]: hands out canned build steps and a canned
    /// download tar (or error), recording the calls it saw — so `build_image_on`
    /// / `extract_path_on` run their full decision + unpack logic with no socket.
    #[cfg(feature = "backend-oci")]
    #[derive(Default)]
    struct ScriptedDaemon {
        /// Drained on the single `build_stream` call — one entry per `BuildInfo`.
        build_steps: RefCell<Vec<Result<Option<String>>>>,
        /// Taken on the single `download_path_tar` call (default: an empty tar).
        download: RefCell<Option<Result<Vec<u8>>>>,
        /// Recorded `(context tar length, tag)` per build.
        saw_build: RefCell<Vec<(usize, String)>>,
        /// Recorded `(image, container_path)` per download.
        saw_download: RefCell<Vec<(String, String)>>,
        /// Canned reply to `inspect_image` (default: `Present`).
        presence: RefCell<Option<ImagePresence>>,
        /// Recorded image ref per presence probe.
        saw_inspect: RefCell<Vec<String>>,
        /// Archive bytes the scripted daemon "exports" (default: nothing at all).
        export_bytes: RefCell<Option<Result<Vec<u8>>>>,
        /// Recorded image ref per export.
        saw_export: RefCell<Vec<String>>,
        /// Canned reply to `load_archive` (default: `Ok`).
        load_result: RefCell<Option<Result<()>>>,
        /// Recorded archive path per load.
        saw_load: RefCell<Vec<String>>,
    }

    #[cfg(feature = "backend-oci")]
    impl ImageDaemon for ScriptedDaemon {
        fn build_stream(
            &self,
            context: Vec<u8>,
            _containerfile: &str,
            tag: &str,
        ) -> Vec<Result<Option<String>>> {
            self.saw_build
                .borrow_mut()
                .push((context.len(), tag.to_string()));
            std::mem::take(&mut *self.build_steps.borrow_mut())
        }
        fn download_path_tar(&self, image: &str, container_path: &str) -> Result<Vec<u8>> {
            self.saw_download
                .borrow_mut()
                .push((image.to_string(), container_path.to_string()));
            self.download
                .borrow_mut()
                .take()
                .unwrap_or_else(|| Ok(Vec::new()))
        }
        fn inspect_image(&self, image: &str) -> ImagePresence {
            self.saw_inspect.borrow_mut().push(image.to_string());
            self.presence
                .borrow_mut()
                .take()
                .unwrap_or(ImagePresence::Present)
        }
        fn export_to(&self, image: &str, sink: &mut dyn std::io::Write) -> Result<u64> {
            self.saw_export.borrow_mut().push(image.to_string());
            let payload = self.export_bytes.borrow_mut().take();
            match payload {
                Some(Err(e)) => Err(e),
                Some(Ok(bytes)) => {
                    sink.write_all(&bytes).unwrap();
                    Ok(bytes.len() as u64)
                }
                // Default: an empty write, which the read-back then rejects as
                // "not an image archive" — the honest outcome for a daemon that
                // streamed nothing.
                None => Ok(0),
            }
        }
        fn load_archive(&self, src: &Path) -> Result<()> {
            self.saw_load
                .borrow_mut()
                .push(src.display().to_string());
            self.load_result.borrow_mut().take().unwrap_or(Ok(()))
        }
    }

    /// A daemon that resolves the reference reads `true`, and the probe asked about
    /// the image it was given. RED-when-broken: invert the `Present` arm and this
    /// flips.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_present_reads_true_when_the_daemon_resolves_it() {
        let daemon = ScriptedDaemon::default();
        *daemon.presence.borrow_mut() = Some(ImagePresence::Present);
        assert!(image_present_on(&daemon, "localhost/nordisk-falkordb-valkey:v4.20.1").unwrap());
        assert_eq!(
            daemon.saw_inspect.borrow().as_slice(),
            ["localhost/nordisk-falkordb-valkey:v4.20.1"],
            "the probe asked the daemon about the image it was handed"
        );
    }

    /// A daemon that answers "no such image" reads `false` — NOT an error. This is
    /// the self-heal path: absent means build it.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_present_reads_false_when_the_daemon_has_no_such_image() {
        let daemon = ScriptedDaemon::default();
        *daemon.presence.borrow_mut() = Some(ImagePresence::Absent);
        assert!(!image_present_on(&daemon, "localhost/nope:1").unwrap());
    }

    /// **The whole reason this seam exists**: an unreachable daemon is an `Err`, never
    /// a silent `false`. `podman image exists` collapses both into a non-zero exit, so
    /// a shell-out caller rebuilds an image it already has every time the socket is
    /// down. RED-when-broken: map `Failed` to `Ok(false)` and this test fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_present_errors_when_the_daemon_cannot_be_asked() {
        let daemon = ScriptedDaemon::default();
        *daemon.presence.borrow_mut() = Some(ImagePresence::Failed("connection refused".into()));
        let err = image_present_on(&daemon, "localhost/spark:1").unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("localhost/spark:1") && m.contains("connection refused")),
            "a transport failure surfaces as a Backend error naming the image + cause, not as absence: {err:?}"
        );
    }

    /// Build a `podman cp`-style tar in memory: entries rooted at `root` (the
    /// basename of the extracted path), each `(relpath, bytes)` a file under it.
    #[cfg(feature = "backend-oci")]
    fn make_cp_tar(root: &str, files: &[(&str, &[u8])]) -> Vec<u8> {
        let mut buf: Vec<u8> = Vec::new();
        {
            let mut b = tar::Builder::new(&mut buf);
            for (rel, data) in files {
                let mut header = tar::Header::new_gnu();
                header.set_size(data.len() as u64);
                header.set_mode(0o644);
                header.set_cksum();
                b.append_data(&mut header, format!("{root}/{rel}"), *data)
                    .unwrap();
            }
            b.finish().unwrap();
        }
        buf
    }

    /// **Happy path**: a build stream carrying only progress lines (no
    /// `error_detail`) succeeds, returning the tag — and the context was tarred and
    /// handed to the daemon under that tag. RED-when-broken: if `build_image_on`
    /// stopped returning `tag`, or skipped tarring the context, an assert flips.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn build_image_returns_the_tag_on_a_clean_stream() {
        let dir = std::env::temp_dir().join(format!("draupnir-build-ok-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("Containerfile"), b"FROM scratch\n").unwrap();

        let daemon = ScriptedDaemon::default();
        *daemon.build_steps.borrow_mut() = vec![Ok(None), Ok(None)]; // two progress lines
        let tag = build_image_on(&daemon, &dir, "Containerfile", "app:test").unwrap();
        assert_eq!(tag, "app:test", "a clean build returns its tag");

        let saw = daemon.saw_build.borrow();
        assert_eq!(saw.len(), 1, "the daemon was streamed exactly one build");
        assert!(
            saw[0].0 > 0,
            "a non-empty context tar was sent: {}",
            saw[0].0
        );
        assert_eq!(saw[0].1, "app:test", "the tag reached the daemon");

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

    /// **Error-detail path**: a `BuildInfo` carrying an `error_detail` message fails
    /// the build with an [`Error::Backend`] naming the tag + the daemon message —
    /// never a fake image. RED-when-broken: drop the `error_detail`→`Err` check and
    /// this returns `Ok("app:bad")`, failing `unwrap_err`.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn build_image_surfaces_an_error_detail_as_err() {
        let dir = std::env::temp_dir().join(format!("draupnir-build-fail-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("Containerfile"), b"FROM scratch\n").unwrap();

        let daemon = ScriptedDaemon::default();
        *daemon.build_steps.borrow_mut() = vec![
            Ok(None),                                            // a progress line first
            Ok(Some("no such file or directory: /nope".into())), // then the daemon errors
        ];
        let err = build_image_on(&daemon, &dir, "Containerfile", "app:bad").unwrap_err();
        match err {
            Error::Backend(m) => {
                assert!(m.contains("app:bad"), "names the tag: {m}");
                assert!(
                    m.contains("no such file"),
                    "carries the daemon message: {m}"
                );
            }
            other => panic!("expected Error::Backend, got {other:?}"),
        }
        crate::functional_status(
            "draupnir/container",
            "build_image_error_detail_fails",
            true,
            "a BuildInfo error_detail surfaces as Err(Error::Backend) naming the tag + message — never a fake image",
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A **transport error** mid-stream (not an `error_detail`, a broken `/build`
    /// socket) is surfaced by the `?` in `build_image_on`, not swallowed as success.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn build_image_propagates_a_transport_error() {
        let dir = std::env::temp_dir().join(format!("draupnir-build-tx-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("Containerfile"), b"FROM scratch\n").unwrap();

        let daemon = ScriptedDaemon::default();
        *daemon.build_steps.borrow_mut() = vec![Err(Error::Backend(
            "build image app:tx: connection reset".into(),
        ))];
        let err = build_image_on(&daemon, &dir, "Containerfile", "app:tx").unwrap_err();
        assert!(matches!(err, Error::Backend(m) if m.contains("connection reset")));

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

    /// **Extract happy path**: the daemon hands back a basename-rooted tar (`podman
    /// cp` layout — downloading `/opt/out` yields entries under `out/…`) and
    /// `extract_path_on` unpacks it so the files land at `host_dest/out/…` with
    /// content intact (nested dirs preserved). RED-when-broken: skip the unpack, or
    /// unpack somewhere other than `host_dest`, and the file reads fail.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn extract_path_unpacks_the_downloaded_tar_with_basename_layout() {
        let tar = make_cp_tar(
            "out",
            &[
                ("app.msi", b"windows-msi-bytes"),
                ("logs/build.log", b"built ok"),
            ],
        );

        let dest = std::env::temp_dir().join(format!("draupnir-extract-ok-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dest);

        let daemon = ScriptedDaemon::default();
        *daemon.download.borrow_mut() = Some(Ok(tar));
        extract_path_on(&daemon, "app:test", "/opt/out", &dest).unwrap();

        // Files land under host_dest/<basename>/… with bytes intact.
        assert_eq!(
            std::fs::read(dest.join("out").join("app.msi")).unwrap(),
            b"windows-msi-bytes",
            "the top-level extracted file lands at host_dest/out/app.msi"
        );
        assert_eq!(
            std::fs::read(dest.join("out").join("logs").join("build.log")).unwrap(),
            b"built ok",
            "the nested file keeps its subtree under host_dest/out/logs/"
        );
        // The image + container_path reached the daemon verbatim.
        assert_eq!(
            daemon.saw_download.borrow().as_slice(),
            &[("app:test".to_string(), "/opt/out".to_string())]
        );

        crate::functional_status(
            "draupnir/container",
            "extract_path_basename_unpack",
            dest.join("out").join("app.msi").exists(),
            "extract_path unpacks the downloaded tar into host_dest with the podman-cp basename layout (host_dest/<basename>/…)",
        );
        let _ = std::fs::remove_dir_all(&dest);
    }

    /// **Extract error path**: a download error surfaces as `Err` *before* any
    /// unpack — no partial fake tree at `host_dest`. RED-when-broken: swallow the
    /// download error and this both returns Ok and leaves an unpacked dir.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn extract_path_surfaces_a_download_error_and_does_not_unpack() {
        let dest =
            std::env::temp_dir().join(format!("draupnir-extract-err-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dest);

        let daemon = ScriptedDaemon::default();
        *daemon.download.borrow_mut() = Some(Err(Error::Backend(
            "download /out from app:x: broken pipe".into(),
        )));
        let err = extract_path_on(&daemon, "app:x", "/out", &dest).unwrap_err();
        assert!(matches!(err, Error::Backend(m) if m.contains("broken pipe")));
        assert!(
            !dest.exists(),
            "nothing is unpacked (dest never created) on a download error"
        );
    }

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

    // -----------------------------------------------------------------------
    // Image ARCHIVES — export/import, the airgap path.
    //
    // NOT EXERCISED HERE (stated plainly rather than implied): no live podman
    // socket is touched, so bollard's `export_image` / `import_image_stream`
    // wire calls themselves are unproven by this suite. What IS proven is
    // everything that decides whether a bundle can be trusted — the archive
    // scanner, the truncation and not-an-image rejections, the missing-image vs
    // dead-socket split, the tag guard, and that a digest survives an
    // export→import round trip — all against fixtures, with no daemon.
    // -----------------------------------------------------------------------

    /// Build a docker-archive-shaped tar in memory: a `manifest.json` naming
    /// `tags` + `config`, plus a fake layer blob of `layer_len` bytes so the
    /// truncation tests have something with real extent to cut.
    #[cfg(feature = "backend-oci")]
    fn make_image_archive(config: &str, tags: &[&str], layer_len: usize) -> Vec<u8> {
        let repo_tags = tags
            .iter()
            .map(|t| format!("{t:?}"))
            .collect::<Vec<_>>()
            .join(",");
        let manifest = format!(
            r#"[{{"Config":"{config}","RepoTags":[{repo_tags}],"Layers":["blobs/sha256/layer0"]}}]"#
        );
        let mut buf: Vec<u8> = Vec::new();
        {
            let mut b = tar::Builder::new(&mut buf);
            let layer = vec![0xABu8; layer_len];
            let mut h = tar::Header::new_gnu();
            h.set_size(layer.len() as u64);
            h.set_mode(0o644);
            h.set_cksum();
            b.append_data(&mut h, "blobs/sha256/layer0", layer.as_slice())
                .unwrap();
            let mut h = tar::Header::new_gnu();
            h.set_size(manifest.len() as u64);
            h.set_mode(0o644);
            h.set_cksum();
            b.append_data(&mut h, "manifest.json", manifest.as_bytes())
                .unwrap();
            b.finish().unwrap();
        }
        buf
    }

    /// A well-formed archive yields the tags and the config digest that are
    /// **inside it**. RED-when-broken: return the caller's requested ref instead of
    /// the parsed one and the digest assertion fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_archive_reads_its_tags_and_config_digest() {
        let tar = make_image_archive(
            "blobs/sha256/deadbeefcafe",
            &["localhost/nordisk-falkordb-valkey:9-20260724"],
            4096,
        );
        let len = tar.len() as u64;
        let (tags, digest) = scan_image_archive(std::io::Cursor::new(tar), len).unwrap();
        assert_eq!(tags, ["localhost/nordisk-falkordb-valkey:9-20260724"]);
        assert_eq!(digest, "deadbeefcafe");
    }

    /// **The failure that matters most**: an archive cut short. A tar header parses
    /// perfectly happily while the data it describes is simply absent, so header
    /// walking alone will never notice — the declared-extent vs file-length check
    /// is what catches it. RED-when-broken: drop the `declared_end > len` check and
    /// a half-copied 1.4 GB image verifies clean.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_archive_rejects_a_truncated_archive() {
        let mut tar = make_image_archive("blobs/sha256/abc123", &["x:1"], 64 * 1024);
        let full = tar.len() as u64;
        // Cut inside the layer body: the layer's header survives, its data does not.
        tar.truncate(2048);
        let len = tar.len() as u64;
        let fault = scan_image_archive(std::io::Cursor::new(tar), len).unwrap_err();
        match fault {
            ArchiveFault::Truncated { declared, actual } => {
                assert!(
                    declared > actual,
                    "the cut is reported as declared({declared}) past actual({actual})"
                );
                assert_eq!(actual, len);
                assert!(declared <= full);
            }
            other => panic!("a cut archive must read as Truncated, got {other:?}"),
        }
    }

    /// A tar that is a perfectly good tar but carries no `manifest.json` is not an
    /// image archive — a distinct fault from "truncated", because the operator
    /// action differs (wrong file vs re-copy the file).
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_archive_rejects_a_tar_that_is_not_an_image() {
        let mut buf: Vec<u8> = Vec::new();
        {
            let mut b = tar::Builder::new(&mut buf);
            let mut h = tar::Header::new_gnu();
            h.set_size(3);
            h.set_mode(0o644);
            h.set_cksum();
            b.append_data(&mut h, "hello.txt", &b"hi\n"[..]).unwrap();
            b.finish().unwrap();
        }
        let len = buf.len() as u64;
        assert_eq!(
            scan_image_archive(std::io::Cursor::new(buf), len).unwrap_err(),
            ArchiveFault::NotAnImageArchive
        );
    }

    /// An empty file is not "an empty image" — it has no manifest, so it is not an
    /// image archive. This is the shape a failed export leaves behind if nothing
    /// verifies the result.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn image_archive_rejects_an_empty_file() {
        assert_eq!(
            scan_image_archive(std::io::Cursor::new(Vec::new()), 0).unwrap_err(),
            ArchiveFault::NotAnImageArchive
        );
    }

    /// Both `Config` spellings in the wild — docker's `<hex>.json` and podman's
    /// `blobs/sha256/<hex>` — must land on the SAME digest, or the same image
    /// exported by two engines would appear to be two images.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn manifest_config_digest_is_the_same_for_both_spellings() {
        let podman =
            br#"[{"Config":"blobs/sha256/05455e6bc39e","RepoTags":["a:1"],"Layers":[]}]"#.to_vec();
        let docker = br#"[{"Config":"05455e6bc39e.json","RepoTags":["a:1"],"Layers":[]}]"#.to_vec();
        assert_eq!(
            parse_archive_manifest(&podman).unwrap(),
            parse_archive_manifest(&docker).unwrap()
        );
        assert_eq!(parse_archive_manifest(&podman).unwrap().1, "05455e6bc39e");
    }

    /// A manifest that parses as JSON but is not the docker-archive shape is
    /// `Malformed`, carrying why — not silently "no tags".
    #[cfg(feature = "backend-oci")]
    #[test]
    fn manifest_that_is_not_the_archive_shape_is_malformed() {
        assert!(matches!(
            parse_archive_manifest(br#"{"Config":"x"}"#).unwrap_err(),
            ArchiveFault::Malformed(m) if m.contains("not a JSON array")
        ));
        assert!(matches!(
            parse_archive_manifest(b"[]").unwrap_err(),
            ArchiveFault::Malformed(m) if m.contains("empty array")
        ));
        assert!(matches!(
            parse_archive_manifest(br#"[{"RepoTags":["a:1"]}]"#).unwrap_err(),
            ArchiveFault::Malformed(m) if m.contains("no Config")
        ));
    }

    /// **A missing image and a dead socket are two different failures.** An image
    /// that was never built reads as "no such image locally" and — critically —
    /// leaves NO file behind, so a later verify cannot mistake a stub for an
    /// export. RED-when-broken: create the file before probing presence and the
    /// `!dest.exists()` assertion fails.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn export_refuses_a_missing_image_and_writes_no_file() {
        let dest =
            std::env::temp_dir().join(format!("draupnir-export-absent-{}.tar", std::process::id()));
        let _ = std::fs::remove_file(&dest);

        let daemon = ScriptedDaemon::default();
        *daemon.presence.borrow_mut() = Some(ImagePresence::Absent);
        let err = export_image_on(&daemon, "localhost/nordisk-spark-iceberg:4.1.2", &dest)
            .unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("no such image locally")),
            "got {err:?}"
        );
        assert!(!dest.exists(), "a missing image leaves no stub archive");
        assert!(
            daemon.saw_export.borrow().is_empty(),
            "the export stream is never opened for an image that is not there"
        );
    }

    /// An unreachable engine must NOT read as "image missing" — otherwise a caller
    /// self-heals by rebuilding an image it already has, forever (the same defect
    /// `image_present_on` exists to prevent, now on the export path).
    #[cfg(feature = "backend-oci")]
    #[test]
    fn export_reports_an_unreachable_engine_as_such() {
        let dest = std::env::temp_dir()
            .join(format!("draupnir-export-nosock-{}.tar", std::process::id()));
        let _ = std::fs::remove_file(&dest);

        let daemon = ScriptedDaemon::default();
        *daemon.presence.borrow_mut() =
            Some(ImagePresence::Failed("connection refused".into()));
        let err = export_image_on(&daemon, "localhost/x:1", &dest).unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m)
                if m.contains("container engine unreachable") && m.contains("connection refused")),
            "got {err:?}"
        );
        assert!(!dest.exists());
    }

    /// A daemon that streams nothing (or dies instantly) must not produce a
    /// "successful" export: the read-back rejects the zero-byte file.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn export_rejects_an_empty_stream_instead_of_claiming_success() {
        let dest =
            std::env::temp_dir().join(format!("draupnir-export-empty-{}.tar", std::process::id()));
        let _ = std::fs::remove_file(&dest);

        let daemon = ScriptedDaemon::default();
        let err = export_image_on(&daemon, "localhost/x:1", &dest).unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("not an image archive")),
            "got {err:?}"
        );
        let _ = std::fs::remove_file(&dest);
    }

    /// The tag guard: an archive whose `RepoTags` do not include the image that was
    /// asked for is refused. This is what stops a bundle silently carrying a
    /// version-mismatched image — korp's `nordisk-falkordb-valkey:9-<date>` encodes
    /// BOTH the Valkey major and the FalkorDB module version in its tag, so a
    /// mislabelled archive is a wrong product, not a cosmetic slip.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn export_refuses_an_archive_tagged_as_a_different_image() {
        let dest =
            std::env::temp_dir().join(format!("draupnir-export-wrong-{}.tar", std::process::id()));
        let _ = std::fs::remove_file(&dest);

        let daemon = ScriptedDaemon::default();
        *daemon.export_bytes.borrow_mut() = Some(Ok(make_image_archive(
            "blobs/sha256/f00d",
            &["localhost/nordisk-falkordb-redis:9-20260724"],
            512,
        )));
        let err = export_image_on(
            &daemon,
            "localhost/nordisk-falkordb-valkey:9-20260724",
            &dest,
        )
        .unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("refusing to record an image under a tag it does not have")),
            "got {err:?}"
        );
        let _ = std::fs::remove_file(&dest);
    }

    /// **Import verifies BEFORE it streams.** A truncated archive must be rejected
    /// without the daemon ever being handed a byte — pushing a gigabyte at a socket
    /// and then discovering the file was short leaves the store in a state nobody
    /// can describe. RED-when-broken: call `load_archive` first and `saw_load` is
    /// non-empty.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn import_verifies_the_archive_before_touching_the_daemon() {
        let src =
            std::env::temp_dir().join(format!("draupnir-import-cut-{}.tar", std::process::id()));
        let mut tar = make_image_archive("blobs/sha256/abc", &["x:1"], 64 * 1024);
        tar.truncate(2048);
        std::fs::write(&src, &tar).unwrap();

        let daemon = ScriptedDaemon::default();
        let err = import_image_on(&daemon, &src).unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("truncated")),
            "got {err:?}"
        );
        assert!(
            daemon.saw_load.borrow().is_empty(),
            "the daemon is never asked to load an archive that failed verification"
        );
        let _ = std::fs::remove_file(&src);
    }

    /// An import of a file that is not there is `Unreadable`, named as such —
    /// distinct from truncated and from not-an-image (three different operator
    /// actions: find the file / re-copy it / use the right file).
    #[cfg(feature = "backend-oci")]
    #[test]
    fn import_reports_a_missing_archive_as_unreadable() {
        let daemon = ScriptedDaemon::default();
        let err =
            import_image_on(&daemon, Path::new("/nonexistent/draupnir-no-such.tar")).unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("unreadable")),
            "got {err:?}"
        );
        assert!(daemon.saw_load.borrow().is_empty());
    }

    /// **THE round trip.** Export an image to disk, import it back, and prove the
    /// config digest and tags survive unchanged — the assertion that a bundle
    /// carries the same image it packed. The daemon is scripted (no live podman),
    /// so what this proves is the archive-identity path end to end: written bytes →
    /// file → re-read → loaded, with the digest read from the archive at both ends
    /// rather than remembered from the request.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn export_then_import_round_trip_preserves_the_digest() {
        let dest =
            std::env::temp_dir().join(format!("draupnir-roundtrip-{}.tar", std::process::id()));
        let _ = std::fs::remove_file(&dest);
        let image = "localhost/nordisk-spark-iceberg:4.1.2";

        let daemon = ScriptedDaemon::default();
        *daemon.export_bytes.borrow_mut() = Some(Ok(make_image_archive(
            "blobs/sha256/05455e6bc39e",
            &[image],
            32 * 1024,
        )));
        let exported = export_image_on(&daemon, image, &dest).unwrap();
        assert_eq!(exported.config_digest, "05455e6bc39e");
        assert_eq!(exported.tags, [image]);
        assert_eq!(exported.path, dest);
        assert_eq!(
            exported.bytes,
            std::fs::metadata(&dest).unwrap().len(),
            "the recorded size is the size on disk"
        );

        // A fresh daemon: nothing carries over but the file itself.
        let target = ScriptedDaemon::default();
        let imported = import_image_on(&target, &dest).unwrap();
        assert_eq!(
            imported.config_digest, exported.config_digest,
            "the digest survives the round trip — same image in, same image out"
        );
        assert_eq!(imported.tags, exported.tags);
        assert_eq!(
            target.saw_load.borrow().as_slice(),
            [dest.display().to_string()],
            "the verified archive is the exact file handed to the daemon"
        );

        let _ = std::fs::remove_file(&dest);
    }

    /// A daemon error while loading is surfaced, not swallowed — the target must
    /// never believe it has an image the engine refused.
    #[cfg(feature = "backend-oci")]
    #[test]
    fn import_surfaces_a_daemon_load_failure() {
        let src =
            std::env::temp_dir().join(format!("draupnir-import-fail-{}.tar", std::process::id()));
        std::fs::write(
            &src,
            make_image_archive("blobs/sha256/abc", &["x:1"], 1024),
        )
        .unwrap();

        let daemon = ScriptedDaemon::default();
        *daemon.load_result.borrow_mut() =
            Some(Err(Error::Backend("no space left on device".into())));
        let err = import_image_on(&daemon, &src).unwrap_err();
        assert!(
            matches!(&err, Error::Backend(m) if m.contains("no space left")),
            "got {err:?}"
        );
        let _ = std::fs::remove_file(&src);
    }

    /// Without the engine feature, the archive verbs are honest
    /// [`Error::Unsupported`] — never a silent success that a bundle would record.
    #[cfg(not(feature = "backend-oci"))]
    #[test]
    fn image_archive_verbs_are_unsupported_without_the_engine() {
        let boot = ContainerBoot::new();
        assert!(matches!(
            boot.export_image("x:test", Path::new("/tmp/draupnir-x.tar")),
            Err(Error::Unsupported(_))
        ));
        assert!(matches!(
            boot.import_image(Path::new("/tmp/draupnir-x.tar")),
            Err(Error::Unsupported(_))
        ));
        assert!(matches!(
            inspect_image_archive(Path::new("/tmp/draupnir-x.tar")),
            Err(Error::Unsupported(_))
        ));
    }
}