rightsize 0.2.0

//! The `Container` builder and the `ContainerGuard` RAII handle it returns.
//!
//! This module owns the hardest moment in the core crate: the two-tier cleanup story.
//! **Happy path:** `ContainerGuard::stop(self)` is an explicit `async fn` that awaits an
//! ordered teardown (backend `stop` then `remove`, then port release). **Fallback:**
//! `Drop` cannot be `async` and must not assume a Tokio runtime is anywhere in the
//! process, so it does the least synchronous work that's still correct — release ports
//! synchronously (`FreePorts` is plain `std::sync::Mutex`-guarded, no runtime needed)
//! and hand the container off to the dedicated cleanup thread in `crate::cleanup`, which
//! tears it down with blocking std I/O only. See `crate::cleanup`'s module docs for why
//! that thread exists and what it promises.
//!
//! `Container::start()`'s ordering (allocate → create → start, with a port-bind-conflict
//! retry loop; then network-link-install → register → wait, as one guarded unit whose
//! failure triggers a full awaited teardown before the error reaches the caller) is
//! deliberate: any partial failure after resources are allocated must still reach a
//! fully-torn-down state before the error propagates to the caller.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use crate::backend::{SandboxBackend, SandboxHandle};
use crate::backends;
use crate::checkpoint::{self, Checkpoint};
use crate::cleanup::{self, CleanupJob};
use crate::error::{Result, RightsizeError};
use crate::free_ports::FreePorts;
use crate::model::{ContainerSpec, ExecResult, FileMount};
use crate::mountable_file::MountableFile;
use crate::network::{Network, NetworkMember};
use crate::run_id::RunId;
use crate::wait::{Wait, WaitStrategy, WaitTarget};

/// How many times `Container::start()` retries the create+start step with fresh host
/// ports before giving up, when every failure is a port-bind conflict.
const PORT_BIND_ATTEMPTS: usize = 5;

/// A process-wide counter for container name suffixes. Retries advance it too (a
/// discarded attempt's name is never reused), so names aren't dense — only unique per
/// process, which is all a name needs to be.
static NAME_COUNTER: AtomicU64 = AtomicU64::new(0);

/// A per-process free-port allocator, shared by every `Container` in this process.
static FREE_PORTS: std::sync::OnceLock<FreePorts> = std::sync::OnceLock::new();

fn free_ports() -> &'static FreePorts {
    FREE_PORTS.get_or_init(FreePorts::new)
}

/// A closure that rewrites a `ContainerSpec` with knowledge of this container's mapped
/// host ports, before `create()` (e.g. Redpanda/Kafka's advertised-listener rewrite).
type SpecCustomizer = dyn Fn(ContainerSpec, &dyn Fn(u16) -> u16) -> ContainerSpec + Send + Sync;

/// A closure run once a guard exists and the wait strategy is satisfied (e.g.
/// Mongo's replica-set init).
type PostStartHook = dyn Fn(&ContainerGuard) -> crate::BoxFuture<'_, Result<()>> + Send + Sync;

/// A single sandboxed container, built with a fluent API and run by whichever
/// [`crate::backend::SandboxBackend`] is active. Configure it with the `with_*`
/// builders, call [`Container::start`] to boot it and get back a [`ContainerGuard`].
pub struct Container {
    image: String,
    env: Vec<(String, String)>,
    exposed_ports: Vec<u16>,
    command: Option<Vec<String>>,
    network: Option<Arc<Network>>,
    aliases: Vec<String>,
    mounts: Vec<FileMount>,
    wait_strategy: Box<dyn WaitStrategy>,
    memory_limit_mb: Option<u64>,
    backend_override: Option<Arc<dyn SandboxBackend>>,
    spec_customizer: Option<Arc<SpecCustomizer>>,
    post_start: Option<Arc<PostStartHook>>,
    reuse: bool,
    cache_dir_override: Option<std::path::PathBuf>,
    reuse_env_override: Option<bool>,
    require_isolation: bool,
    reaper_cache_dir_override: Option<std::path::PathBuf>,
}

impl Container {
    /// Starts building a container from `image`, with no exposed ports, no env, the
    /// default [`Wait::for_listening_port`] readiness check, and every other option at
    /// its default.
    pub fn new(image: &str) -> Container {
        Container {
            image: image.to_string(),
            env: Vec::new(),
            exposed_ports: Vec::new(),
            command: None,
            network: None,
            aliases: Vec::new(),
            mounts: Vec::new(),
            wait_strategy: Wait::for_listening_port(),
            memory_limit_mb: None,
            backend_override: None,
            spec_customizer: None,
            post_start: None,
            reuse: false,
            cache_dir_override: None,
            reuse_env_override: None,
            require_isolation: false,
            reaper_cache_dir_override: None,
        }
    }

    /// Builds a normal `Container` from a [`Checkpoint`]'s image and the source
    /// container's env, command, exposed ports (guest side only — a restored
    /// container gets fresh host ports, chosen by the core allocator exactly like
    /// any other `start()`), and memory limit. Every other field starts at its
    /// ordinary default and can still be overridden with the usual builders before
    /// `start()` (e.g. a different `.waiting_for(...)` wait strategy) — the value
    /// returned here is a plain `Container`, indistinguishable from one built by
    /// [`Container::new`]. A container started from it is ordinary in every other
    /// respect too: a fresh name, fresh host ports, normal reaping, normal `stop()`.
    ///
    /// Deliberately does NOT carry over the checkpoint's `mounts`, `network_id`, or
    /// `aliases` — the checkpoint image already has whatever those mounts wrote,
    /// baked directly into its filesystem (see [`Checkpoint`]'s own doc for the
    /// "filesystem capture, not memory" semantics), and network topology has no
    /// well-defined meaning to carry across a restore.
    pub fn from_checkpoint(cp: &Checkpoint) -> Container {
        let mut c = Container::new(&cp.image_ref);
        c.env = cp.spec.env.clone();
        c.command = cp.spec.command.clone();
        c.exposed_ports = cp.spec.ports.iter().map(|p| p.guest_port).collect();
        c.memory_limit_mb = cp.spec.memory_limit_mb;
        c
    }

    /// Sets a single environment variable for the container process. Call again with
    /// the same key to overwrite it — the *value* from the last call wins, but the
    /// key keeps the position of its *first* `with_env` call in the final spec's
    /// iteration order (see `dedup_env_last_wins`, applied once at `start()` time,
    /// right before the spec reaches a backend — this builder itself still just
    /// appends, so a spec-customizer pushing more env entries is resolved the same way).
    pub fn with_env(mut self, k: &str, v: &str) -> Self {
        self.env.push((k.to_string(), v.to_string()));
        self
    }

    /// Removes every entry for `key` set so far via [`Self::with_env`]. The core-level
    /// primitive a module needs when a later env var must ALSO retract an earlier
    /// one's effect on the running process, not merely be shadowed by last-wins
    /// value resolution (e.g.
    /// `ArangoContainer::with_root_password` clearing `ARANGO_NO_AUTH`, whose
    /// entrypoint checks presence/absence of the variable itself, not just its
    /// value — last-wins on `ARANGO_NO_AUTH`'s own value can't turn it "off").
    /// No-op if `key` was never set.
    pub fn remove_env(mut self, key: &str) -> Self {
        self.env.retain(|(k, _)| k != key);
        self
    }

    /// The env entries set so far, in call order — a plain read-only introspection
    /// point (e.g. for a module's own unit tests asserting what its builder produced,
    /// such as `ArangoContainer::with_root_password`'s NO_AUTH-removed/password-added
    /// contract) rather than a way to mutate the builder. Reflects [`Self::with_env`]
    /// calls made so far exactly, duplicates and all — the last-wins dedup
    /// (`dedup_env_last_wins`) only runs once, at `start()` time.
    pub fn env(&self) -> &[(String, String)] {
        &self.env
    }

    /// Declares guest ports to publish; each gets a host port assigned before boot.
    pub fn with_exposed_ports(mut self, ports: &[u16]) -> Self {
        self.exposed_ports.extend_from_slice(ports);
        self
    }

    /// Overrides the image's default entrypoint/command.
    pub fn with_command(mut self, cmd: &[&str]) -> Self {
        self.command = Some(cmd.iter().map(|s| s.to_string()).collect());
        self
    }

    /// Joins `net`, making this container's exposed ports reachable at its
    /// [`Container::with_network_aliases`].
    pub fn with_network(mut self, net: &Arc<Network>) -> Self {
        self.network = Some(net.clone());
        self
    }

    /// Names this container is reachable as on its network (see [`Network::resolve`]).
    pub fn with_network_aliases(mut self, names: &[&str]) -> Self {
        self.aliases.extend(names.iter().map(|s| s.to_string()));
        self
    }

    /// Mounts `file` read-only into the guest at `guest_path`; takes effect at the next
    /// `start()`.
    pub fn with_copy_file_to_container(mut self, file: MountableFile, guest_path: &str) -> Self {
        self.mounts.push(FileMount::new(file.path(), guest_path));
        self
    }

    /// Overrides the readiness check run after boot; defaults to
    /// [`Wait::for_listening_port`]. Takes the strategy by value — a bare
    /// `Wait::for_http("/health").for_port(80)` or a caller's own `impl WaitStrategy`
    /// works directly, no `Box::new(..)` wart at the call site; this boxes internally
    /// (a `Box<dyn WaitStrategy>` from a built-in factory that already returns one
    /// satisfies the bound too, via the blanket impl in `crate::wait`, so it is never
    /// double-boxed).
    pub fn waiting_for(mut self, strategy: impl WaitStrategy + 'static) -> Self {
        self.wait_strategy = Box::new(strategy);
        self
    }

    /// Caps the container's guest memory at `megabytes`. Leaving this unset lets each
    /// backend apply its own default.
    pub fn with_memory_limit(mut self, megabytes: u64) -> Self {
        self.memory_limit_mb = Some(megabytes);
        self
    }

    /// Marks this container for reuse: a container built from an identical
    /// image/env/command/exposed-ports/memory-limit/mounted-files spec survives
    /// process exit and is ADOPTED — not re-created — by a later `start()`, in this
    /// process or a later one; `stop()` then leaves the sandbox running instead of
    /// tearing it down. Requires a double opt-in: `RIGHTSIZE_REUSE` must ALSO be
    /// set to `"true"` or `"1"` in the real process environment, or `start()`
    /// behaves exactly as an ordinary ephemeral container (with a single stderr
    /// note that reuse was requested but not enabled). Reuse cannot be combined
    /// with [`Self::with_network`] — `start()` returns
    /// [`RightsizeError::ReuseNetworkConflict`] up front. Defaults to `false`.
    pub fn reuse(mut self, enabled: bool) -> Self {
        self.reuse = enabled;
        self
    }

    /// Requires the active backend to provide hardware isolation — its own kernel per
    /// sandbox, see [`crate::backend::Capabilities::hardware_isolated`] — before this
    /// container is allowed to start. Checked in [`Self::start`], before any
    /// create/network work: if the active backend does not provide it (the docker
    /// backend, which shares the host kernel), `start()` returns
    /// [`RightsizeError::IsolationRequired`] and no sandbox is created. Use this for
    /// workloads that genuinely need microVM-strength isolation (untrusted code),
    /// rather than trusting every backend a caller might resolve to. Defaults to
    /// `false`.
    pub fn require_isolation(mut self, enabled: bool) -> Self {
        self.require_isolation = enabled;
        self
    }

    /// Test/module seam: overrides the backend this container starts on, instead of the
    /// process-wide active backend. Every current caller is a unit test in this crate —
    /// `#[cfg_attr(not(test), allow(dead_code))]` reflects that precisely instead of
    /// blanket-silencing dead-code analysis on this method.
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn with_backend(mut self, b: Arc<dyn SandboxBackend>) -> Self {
        self.backend_override = Some(b);
        self
    }

    /// Test/module seam: overrides the reuse registry's cache dir instead of the
    /// real `crate::cache_dir::dir()` (which reads `RIGHTSIZE_CACHE_DIR` from the
    /// real process environment) — every current caller is a unit test exercising
    /// the reuse flow's registry file, which must never touch a real machine's
    /// `~/.cache/rightsize/reuse/` directory.
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn with_cache_dir_override(mut self, dir: std::path::PathBuf) -> Self {
        self.cache_dir_override = Some(dir);
        self
    }

    /// Test/module seam: overrides the `RIGHTSIZE_REUSE` double-opt-in check instead
    /// of reading the real process environment (`crate::reuse::env_enabled`) — lets
    /// unit tests exercise both gating outcomes deterministically, without mutating
    /// real process env (racy across the parallel test threads `cargo test` uses
    /// within one binary) or needing `unsafe` (`std::env::set_var` requires it as of
    /// the 2024 edition, and this crate forbids unsafe code entirely).
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn with_reuse_env_override(mut self, enabled: bool) -> Self {
        self.reuse_env_override = Some(enabled);
        self
    }

    /// Test/module seam: overrides the REAPING ledger's cache dir instead of the
    /// real `crate::cache_dir::dir()` — distinct from [`Self::with_cache_dir_override`],
    /// which only redirects the REUSE registry. Every current caller is a unit
    /// test in this module that asserts against `.sandboxes`/`.networks`
    /// directly (`crate::reaper::Ledger`); without this, those tests read/write
    /// the real, process-wide ledger under this developer machine's actual
    /// `~/.cache/rightsize/runs/` and are inherently coupled to every other
    /// concurrently-running test in the same binary sharing that same file (see
    /// `crate::reaper::ledger`'s module doc — one `WRITE_LOCK`-guarded ledger per
    /// process, keyed only by run id, not by test). Threaded down to every
    /// `crate::reaper::before_create`/`after_stop`/`before_ensure_network`/
    /// `after_remove_network` call this container's own lifecycle makes (see
    /// [`crate::reaper::ledger_for`]'s doc for exactly what does and doesn't
    /// change under the override) — but NOT to the watchdog-spawn-once gate or
    /// this process's `RIGHTSIZE_REAPER` mode, which stay tied to the real,
    /// process-wide reaper state regardless, matching every other test in this
    /// binary. Production never sets this field, so `None` (the real cache dir)
    /// is always what a real caller gets.
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn with_reaper_cache_dir_override(mut self, dir: std::path::PathBuf) -> Self {
        self.reaper_cache_dir_override = Some(dir);
        self
    }

    /// Module hook: rewrites the `ContainerSpec` with knowledge of this container's own
    /// mapped host ports, right before `create()` — e.g. Redpanda/Kafka's
    /// advertised-listener rewrite, which needs to know its own mapped port to advertise
    /// it.
    pub fn with_spec_customizer(
        mut self,
        f: impl Fn(ContainerSpec, &dyn Fn(u16) -> u16) -> ContainerSpec + Send + Sync + 'static,
    ) -> Self {
        self.spec_customizer = Some(Arc::new(f));
        self
    }

    /// Module hook: runs once the guard exists and the wait strategy is satisfied — e.g.
    /// Mongo's replica-set initialization.
    pub fn with_post_start(
        mut self,
        f: impl for<'a> Fn(&'a ContainerGuard) -> crate::BoxFuture<'a, Result<()>>
        + Send
        + Sync
        + 'static,
    ) -> Self {
        self.post_start = Some(Arc::new(f));
        self
    }

    fn active_backend(&self) -> Arc<dyn SandboxBackend> {
        match &self.backend_override {
            Some(b) => b.clone(),
            None => backends::active(),
        }
    }

    fn describe(id: &str, image: &str) -> String {
        format!("container(image={image}, id={id})")
    }

    /// Boots the container and returns an RAII guard. On ANY failure partway (create,
    /// start, install_network_links, register, OR wait), teardown runs and nothing
    /// leaks — `start()` does not return its error to the caller until that teardown
    /// has finished.
    pub async fn start(self) -> Result<ContainerGuard> {
        let backend = self.active_backend();

        if self.require_isolation && !backend.capabilities().hardware_isolated {
            // Checked before any create/network work — reuse's own registry lookup,
            // ensure_network, and create_started_container all come after this, so a
            // non-isolated backend never gets far enough to create anything.
            return Err(RightsizeError::IsolationRequired {
                backend: backend.name().to_string(),
            });
        }

        if self.reuse {
            let reuse_env_enabled = self
                .reuse_env_override
                .unwrap_or_else(crate::reuse::env_enabled);
            if reuse_env_enabled {
                if let Some(net) = &self.network {
                    return Err(RightsizeError::ReuseNetworkConflict {
                        network_id: net.id().to_string(),
                    });
                }
                return start_reuse(self, backend).await;
            }
            // API-marked but env-disabled: the double opt-in requires both, so this
            // container runs as an ordinary ephemeral one — Testcontainers
            // semantics — falling straight through to the unchanged path below,
            // with a single note so a caller who forgot to set RIGHTSIZE_REUSE
            // notices why nothing was adopted.
            eprintln!(
                "rightsize: .reuse(true) was requested but RIGHTSIZE_REUSE is not enabled (set \
                 it to \"true\" or \"1\") — starting an ordinary, non-reused container instead."
            );
        }

        if let Some(net) = &self.network {
            // Append-before-create, same discipline as a sandbox name (see
            // `crate::reaper`'s module doc) — dedupes across repeat joiners of the
            // same network, since `Ledger::append_network` is itself idempotent.
            crate::reaper::before_ensure_network(
                net.id(),
                self.reaper_cache_dir_override.as_deref(),
            );
            if let Err(e) = backend.ensure_network(net.id()).await {
                // This attempt never produced a usable network — undo the ledger
                // append above so a discarded id doesn't sit in `.networks` forever
                // and block the clean-shutdown deletion trigger for the rest of this
                // process. Mirrors `create_started_container`'s `after_stop` cleanup
                // on its own `create`/`start` failure branches.
                crate::reaper::after_remove_network(
                    net.id(),
                    self.reaper_cache_dir_override.as_deref(),
                );
                return Err(e);
            }
        }

        let (handle, mapped_ports) = create_started_container(
            &backend,
            &self.image,
            &self.env,
            &self.command,
            &self.exposed_ports,
            &self.mounts,
            self.network.as_deref(),
            &self.aliases,
            self.memory_limit_mb,
            self.spec_customizer.as_deref(),
            self.reaper_cache_dir_override.as_deref(),
        )
        .await?;

        let name = handle.id().to_string();
        // The reaping ledger tracks `ContainerSpec::name` (`rz-<run_id>-<seq>`), not
        // `SandboxHandle::id()` — the two coincide for msb but NOT for docker, whose
        // `id()` is the daemon-assigned container id. Captured here, before `handle`
        // moves into the guard, for `stop_inner`/`Drop` to hand to
        // `crate::reaper::after_stop`.
        let ledger_name = handle.spec().name.clone();
        let keep_alive = handle.spec().keep_alive;
        // Captured before `handle` moves into the guard below — the diagnostics
        // registry owns its own copy of the handle id/spec, independent of the
        // guard's lifetime (see `crate::diagnostics`'s module doc).
        let diagnostics_handle_id = handle.id().to_string();
        let diagnostics_spec = handle.spec().clone();
        let diagnostics_ports = mapped_ports.clone();
        let guard = ContainerGuard {
            handle: Some(handle),
            backend: backend.clone(),
            mapped_ports: Mutex::new(mapped_ports),
            network: self.network.clone(),
            image: self.image.clone(),
            exposed_ports: self.exposed_ports.clone(),
            name,
            ledger_name,
            keep_alive,
            reaper_cache_dir_override: self.reaper_cache_dir_override.clone(),
        };

        // Guarded block: install_network_links -> register -> wait. On ANY error here,
        // await the guard's own stop to completion, then propagate the original error —
        // a half-started container never leaks regardless of where in this block it
        // failed, and start() does not return until teardown has finished.
        if let Err(e) = link_register_and_wait(
            &guard,
            &backend,
            self.network.as_deref(),
            &self.aliases,
            self.wait_strategy.as_ref(),
        )
        .await
        {
            let _ = guard.stop().await;
            return Err(e);
        }

        // Registered only once the readiness wait has fully succeeded — mirrors the
        // Kotlin port's resolution of this same finding (adopt path registers after
        // its own wait re-run). A container that boots but never becomes ready must
        // never appear in the report, so a mid-wait `diagnostics()` call cannot list
        // it, and the wait-failure branch above has nothing to deregister.
        crate::diagnostics::register(
            &guard.ledger_name,
            &guard.image,
            guard.host(),
            diagnostics_ports,
            backend.clone(),
            &diagnostics_handle_id,
            diagnostics_spec,
        );

        if let Some(post_start) = &self.post_start {
            if let Err(e) = post_start(&guard).await {
                let _ = guard.stop().await;
                return Err(e);
            }
        }

        Ok(guard)
    }
}

#[allow(clippy::too_many_arguments)]
async fn create_started_container(
    backend: &Arc<dyn SandboxBackend>,
    image: &str,
    env: &[(String, String)],
    command: &Option<Vec<String>>,
    exposed_ports: &[u16],
    mounts: &[FileMount],
    network: Option<&Network>,
    aliases: &[String],
    memory_limit_mb: Option<u64>,
    spec_customizer: Option<&SpecCustomizer>,
    reaper_cache_dir_override: Option<&std::path::Path>,
) -> Result<(Box<dyn SandboxHandle>, Vec<(u16, u16)>)> {
    let mut last_conflict: Option<RightsizeError> = None;

    for _ in 0..PORT_BIND_ATTEMPTS {
        let mapped_ports = allocate_ports(exposed_ports)?;
        let seq = NAME_COUNTER.fetch_add(1, Ordering::SeqCst);
        let name = format!("rz-{}-{seq}", RunId::value());

        let mut spec = ContainerSpec {
            name: name.clone(),
            image: image.to_string(),
            env: env.to_vec(),
            command: command.clone(),
            ports: mapped_ports
                .iter()
                .map(|&(guest_port, host_port)| crate::model::PortBinding {
                    host_port,
                    guest_port,
                })
                .collect(),
            mounts: mounts.to_vec(),
            network_id: network.map(|n| n.id().to_string()),
            aliases: aliases.to_vec(),
            run_id: RunId::value().to_string(),
            memory_limit_mb,
            keep_alive: false,
        };

        if let Some(customizer) = spec_customizer {
            let lookup: std::collections::HashMap<u16, u16> =
                mapped_ports.iter().copied().collect();
            let mapped_fn = move |guest: u16| -> u16 {
                *lookup
                    .get(&guest)
                    .expect("customizer looked up an unexposed port")
            };
            spec = customizer(spec, &mapped_fn);
        }

        // Last-wins, insertion-order-of-first-occurrence dedup — see
        // `dedup_env_last_wins`'s doc for why this runs here (after the
        // spec-customizer, which may itself push more env entries) rather than
        // earlier.
        spec.env = dedup_env_last_wins(spec.env);

        // The reaping ledger's own append-before-create discipline: the name must be
        // recorded as a superset BEFORE the backend actually creates it, so a crash
        // between this line and `backend.create` still leaves a (harmlessly
        // not-found-on-remove) name in the ledger rather than a live sandbox with no
        // record at all. See `crate::reaper`'s module doc.
        crate::reaper::before_create(
            backend,
            &spec.name,
            spec.keep_alive,
            reaper_cache_dir_override,
        );
        let attempt_name = spec.name.clone();
        let attempt_keep_alive = spec.keep_alive;

        let handle = match backend.create(spec).await {
            Ok(h) => h,
            Err(e) => {
                // This attempt never produced a live sandbox — undo the ledger append
                // above so a discarded name doesn't sit in `.sandboxes` forever.
                crate::reaper::after_stop(
                    &attempt_name,
                    attempt_keep_alive,
                    reaper_cache_dir_override,
                );
                return Err(e);
            }
        };
        match backend.start(handle.as_ref()).await {
            Ok(()) => return Ok((handle, mapped_ports)),
            Err(e) => {
                let _ = backend.stop(handle.as_ref()).await;
                let _ = backend.remove(handle.as_ref()).await;
                release_ports(&mapped_ports);
                // Same rationale as the `create` failure branch above: this attempt's
                // container was just torn down, so its ledger line must go too.
                crate::reaper::after_stop(
                    &attempt_name,
                    attempt_keep_alive,
                    reaper_cache_dir_override,
                );
                if is_port_bind_conflict(&e) {
                    last_conflict = Some(e);
                    continue;
                }
                return Err(e);
            }
        }
    }

    Err(RightsizeError::Backend(format!(
        "Could not bind free host ports for {} after {PORT_BIND_ATTEMPTS} attempts — another \
         process kept grabbing the allocated ports first; if this persists, check for a port \
         scanner/leaked process racing the allocator on this host{}",
        Container::describe("<unstarted>", image),
        last_conflict
            .map(|c| format!(" (last conflict: {c})"))
            .unwrap_or_default(),
    )))
}

/// Collapses `env` to last-wins-per-key, keeping each key at the position of its
/// **first** occurrence — an overwrite updates the value in place without moving
/// the key to the end of iteration order, the way an insertion-ordered map's `put`
/// behaves.
///
/// `Container::with_env` (and a spec-customizer pushing straight onto
/// `ContainerSpec::env`) is append-only — a `Vec<(String, String)>`, not a map — so
/// calling it twice with the same key previously left *both* entries in the spec,
/// and which one "won" was whatever the backend's own env-merge order happened to
/// do with a duplicate key (Docker and msb are not guaranteed to agree — see the
/// fix commit this function landed in). This function is the single seam that
/// restores map-like last-wins semantics without changing `with_env`'s append-only
/// builder shape or `ContainerSpec::env`'s `Vec` type (both stay exactly as
/// documented elsewhere — order and duplicate-key handling stay under the caller's
/// control up to this point; this is where they get resolved, once, right before
/// the spec reaches a backend).
fn dedup_env_last_wins(env: Vec<(String, String)>) -> Vec<(String, String)> {
    let mut order: Vec<String> = Vec::new();
    let mut values: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for (k, v) in env {
        if !values.contains_key(&k) {
            order.push(k.clone());
        }
        values.insert(k, v);
    }
    order
        .into_iter()
        .map(|k| {
            let v = values.remove(&k).expect("key was just recorded in order");
            (k, v)
        })
        .collect()
}

fn allocate_ports(exposed_ports: &[u16]) -> Result<Vec<(u16, u16)>> {
    let mut mapped = Vec::with_capacity(exposed_ports.len());
    for &guest_port in exposed_ports {
        match free_ports().allocate() {
            Ok(host_port) => mapped.push((guest_port, host_port)),
            Err(e) => {
                // Roll back this attempt's partial allocation before propagating.
                release_ports(&mapped);
                return Err(e);
            }
        }
    }
    Ok(mapped)
}

fn release_ports(mapped_ports: &[(u16, u16)]) {
    for &(_, host_port) in mapped_ports {
        free_ports().release(host_port);
    }
}

/// True if `e` represents a host-port bind conflict worth retrying with fresh ports.
/// Prefers the typed [`RightsizeError::PortBindConflict`], walking the `source` chain;
/// falls back to matching known message phrasings (case-insensitive) for backends that
/// don't throw the typed variant. Negatives — any other error — must NOT retry.
pub(crate) fn is_port_bind_conflict(e: &RightsizeError) -> bool {
    let mut current: Option<&RightsizeError> = Some(e);
    while let Some(err) = current {
        if matches!(err, RightsizeError::PortBindConflict { .. }) {
            return true;
        }
        let msg = err.to_string().to_lowercase();
        if msg.contains("address already in use") || msg.contains("already allocated") {
            return true;
        }
        current = match err {
            RightsizeError::PortBindConflict {
                source: Some(s), ..
            } => Some(s.as_ref()),
            _ => None,
        };
    }
    false
}

async fn link_register_and_wait(
    guard: &ContainerGuard,
    backend: &Arc<dyn SandboxBackend>,
    network: Option<&Network>,
    aliases: &[String],
    wait_strategy: &dyn WaitStrategy,
) -> Result<()> {
    if let Some(net) = network {
        let links = net.links_for_new_member();
        backend
            .install_network_links(guard.handle_ref(), &links)
            .await?;
    }
    if let Some(net) = network {
        net.register(guard.as_network_member(), aliases.to_vec(), backend.clone());
    }
    let target = GuardWaitTarget { guard };
    wait_strategy.wait_until_ready(&target).await
}

/// Adapts a [`ContainerGuard`] to the [`WaitTarget`] a [`WaitStrategy`] needs.
struct GuardWaitTarget<'a> {
    guard: &'a ContainerGuard,
}

#[async_trait::async_trait]
impl WaitTarget for GuardWaitTarget<'_> {
    fn host(&self) -> &str {
        self.guard.host()
    }
    fn mapped_port(&self, guest_port: u16) -> u16 {
        self.guard.get_mapped_port(guest_port).unwrap_or(guest_port)
    }
    fn exposed_guest_ports(&self) -> Vec<u16> {
        self.guard.exposed_ports.clone()
    }
    async fn current_logs(&self) -> String {
        self.guard.logs().await.unwrap_or_default()
    }
    fn describe(&self) -> String {
        self.guard.describe()
    }
}

/// The same [`WaitTarget`] adapter [`GuardWaitTarget`] provides, but usable BEFORE a
/// [`ContainerGuard`] exists — the reuse flow's fresh-create and adopt-verify steps
/// both need to run a wait strategy against a raw handle/mapped-ports pair, and a
/// failed wait there must tear the sandbox down for real (not through a keep_alive
/// guard's own `stop()`, which would leave it running by design). See
/// [`create_fresh_reuse`] and [`try_adopt`].
struct RawWaitTarget<'a> {
    host: &'a str,
    mapped_ports: &'a [(u16, u16)],
    exposed_ports: &'a [u16],
    backend: &'a Arc<dyn SandboxBackend>,
    handle: &'a dyn SandboxHandle,
    name: &'a str,
    image: &'a str,
}

#[async_trait::async_trait]
impl WaitTarget for RawWaitTarget<'_> {
    fn host(&self) -> &str {
        self.host
    }
    fn mapped_port(&self, guest_port: u16) -> u16 {
        self.mapped_ports
            .iter()
            .find(|&&(g, _)| g == guest_port)
            .map(|&(_, h)| h)
            .unwrap_or(guest_port)
    }
    fn exposed_guest_ports(&self) -> Vec<u16> {
        self.exposed_ports.to_vec()
    }
    async fn current_logs(&self) -> String {
        self.backend.logs(self.handle).await.unwrap_or_default()
    }
    fn describe(&self) -> String {
        Container::describe(self.name, self.image)
    }
}

/// Orchestrates a reuse-active `start()` once both opt-ins are confirmed and any
/// custom network has already been rejected by the caller (see `Container::start`).
/// Registry lookup miss/corrupt/stale all fall through to [`create_fresh_reuse`];
/// a name-collision on that create (`crate::reuse::is_name_conflict`) gets exactly
/// one retry back into [`try_adopt`], on the theory that the process that won the
/// race is (or is about to be) the one that wrote the registry entry this retry
/// reads.
async fn start_reuse(
    container: Container,
    backend: Arc<dyn SandboxBackend>,
) -> Result<ContainerGuard> {
    let env = dedup_env_last_wins(container.env.clone());
    let identity = crate::reuse::compute_identity(
        &container.image,
        &env,
        &container.command,
        &container.exposed_ports,
        container.memory_limit_mb,
        &container.mounts,
    )?;

    let cache_dir = container
        .cache_dir_override
        .clone()
        .unwrap_or_else(crate::cache_dir::dir);
    let registry = crate::reuse::Registry::new(&cache_dir, &identity.hash_hex);

    if registry.exists() {
        match registry.read() {
            Some(entry) => {
                if let Some(guard) = try_adopt(&container, &backend, &identity, &entry).await {
                    return Ok(guard);
                }
                // Not adoptable (not running, wait failed, or a port this call's
                // own exposed_ports needs was missing from the entry): best-effort
                // remove whatever's actually there and the stale registry file,
                // then fall through to a fresh create below.
                backend.remove_by_name(&entry.name);
                registry.delete();
            }
            None => {
                // The file exists but didn't parse — we don't know what name it
                // recorded, but the identity-derived name is deterministic
                // regardless of registry content, so best-effort removal still has
                // a target.
                backend.remove_by_name(&identity.name);
                registry.delete();
            }
        }
    }

    // Crash-mid-boot orphan recovery: by this point the adopt path has concluded
    // there is no usable registry entry at all — missing, corrupt, or stale/failed
    // verification (each branch above already best-effort removed what IT knew
    // about). But a sandbox under this identity's FIXED name can still be RUNNING
    // regardless: a process that crashed (or failed its own wait strategy) after
    // `create_and_start_reuse_sandbox` but before `create_fresh_reuse` ever reached
    // its registry write leaves exactly this state, and `keep_alive` hides it from
    // every reaping/sweep path by design (see `crate::reaper`'s module doc and
    // `docs/reuse.md`'s crash-mid-boot orphan section) — this is the only place left
    // that can ever notice and clean it up. Ask the backend directly rather than
    // trusting the registry's absence, and only remove when it actually reports one
    // running: an unconditional remove_by_name here would be a wasted backend call
    // on the overwhelmingly common genuinely-fresh-identity path, and — more
    // importantly — this check must stay a strict subset of "is a sandbox already
    // there right now," never "assume the identity is ours to clear": the
    // name-collision-retry branch below is what handles a LIVE concurrent creator,
    // and it deliberately never calls remove_by_name.
    if find_running_by_name(&backend, &identity.name, &container.image)
        .await
        .is_some()
    {
        backend.remove_by_name(&identity.name);
    }

    match create_fresh_reuse(&container, &backend, &identity, &env, &registry).await {
        Ok(guard) => Ok(guard),
        Err(e) if crate::reuse::is_name_conflict(&e) => {
            // Another process won the create race. Re-enter the adopt path once,
            // reading whatever the winner has (or hasn't yet) written — if that
            // doesn't pan out either, surface the ORIGINAL collision error rather
            // than inventing a new one, and critically do NOT best-effort-remove
            // anything here: unlike the stale-registry branch above, a name
            // collision means a sandbox some OTHER live process just legitimately
            // created is sitting there, and removing it out from under that
            // process would defeat the entire point of reuse.
            match registry.read() {
                Some(entry) => match try_adopt(&container, &backend, &identity, &entry).await {
                    Some(guard) => Ok(guard),
                    None => Err(e),
                },
                None => Err(e),
            }
        }
        Err(e) => Err(e),
    }
}

/// Best-effort query for whether a sandbox is already running under `name` —
/// [`start_reuse`]'s own crash-mid-boot orphan check, built around the minimal
/// [`ContainerSpec`] [`SandboxBackend::find_running`] actually needs (both real
/// backends' implementations key only on `spec.name`; see `rightsize-msb` and
/// `rightsize-docker`'s own `find_running`). `None` on any failure or "not
/// running" — same fold as [`SandboxBackend::find_running`]'s own contract, since
/// this is a pure best-effort probe, never a fatal error in its own right.
async fn find_running_by_name(
    backend: &Arc<dyn SandboxBackend>,
    name: &str,
    image: &str,
) -> Option<Box<dyn SandboxHandle>> {
    let probe = ContainerSpec::new(name, image, RunId::value());
    backend.find_running(&probe).await.ok().flatten()
}

/// Attempts to adopt an already-running reuse sandbox recorded in `entry`: verifies
/// it's actually running (via [`SandboxBackend::find_running`]), then re-runs the
/// container's own wait strategy against the registry's recorded ports. `None` for
/// any failure along the way (not running, a currently-exposed port missing from the
/// registry, or the wait strategy failing) — every failure mode here is the caller's
/// cue to fall back to a fresh create, never a fatal error in its own right.
async fn try_adopt(
    container: &Container,
    backend: &Arc<dyn SandboxBackend>,
    identity: &crate::reuse::Identity,
    entry: &crate::reuse::RegistryEntry,
) -> Option<ContainerGuard> {
    let mut mapped_ports = Vec::with_capacity(container.exposed_ports.len());
    for &guest_port in &container.exposed_ports {
        let host_port = *entry.ports.get(&guest_port.to_string())?;
        mapped_ports.push((guest_port, host_port));
    }

    let adopted_spec = ContainerSpec {
        name: identity.name.clone(),
        image: container.image.clone(),
        env: dedup_env_last_wins(container.env.clone()),
        command: container.command.clone(),
        ports: mapped_ports
            .iter()
            .map(|&(guest_port, host_port)| crate::model::PortBinding {
                host_port,
                guest_port,
            })
            .collect(),
        mounts: container.mounts.clone(),
        network_id: None,
        aliases: container.aliases.clone(),
        run_id: RunId::value().to_string(),
        memory_limit_mb: container.memory_limit_mb,
        keep_alive: true,
    };

    let handle = match backend.find_running(&adopted_spec).await {
        Ok(Some(h)) => h,
        Ok(None) | Err(_) => return None,
    };

    let raw = RawWaitTarget {
        host: "127.0.0.1",
        mapped_ports: &mapped_ports,
        exposed_ports: &container.exposed_ports,
        backend,
        handle: handle.as_ref(),
        name: &identity.name,
        image: &container.image,
    };
    if container
        .wait_strategy
        .wait_until_ready(&raw)
        .await
        .is_err()
    {
        return None;
    }

    let diagnostics_handle_id = handle.id().to_string();
    let diagnostics_spec = handle.spec().clone();
    let diagnostics_ports = mapped_ports.clone();
    let guard = ContainerGuard {
        handle: Some(handle),
        backend: backend.clone(),
        mapped_ports: Mutex::new(mapped_ports),
        network: None,
        image: container.image.clone(),
        exposed_ports: container.exposed_ports.clone(),
        name: identity.name.clone(),
        ledger_name: identity.name.clone(),
        keep_alive: true,
        reaper_cache_dir_override: container.reaper_cache_dir_override.clone(),
    };
    crate::diagnostics::register(
        &guard.ledger_name,
        &guard.image,
        guard.host(),
        diagnostics_ports,
        backend.clone(),
        &diagnostics_handle_id,
        diagnostics_spec,
    );
    Some(guard)
}

/// Creates and starts a reuse sandbox under the identity-derived, FIXED
/// `rz-reuse-<12hex>` name, retrying with freshly allocated ports on a host-port
/// bind conflict — the same retry discipline [`create_started_container`] uses for
/// an ordinary container. Ported here as its own helper (rather than inlined
/// straight-line code) because a reuse sandbox's name is deterministic
/// (identity-derived) instead of a fresh name-per-attempt, so only the ports (and
/// the spec built from them) change between attempts; everything else about the
/// retry — release ports, stop+remove the failed attempt, `is_port_bind_conflict`
/// as the sole retry trigger, the same exhausted-attempts error — mirrors
/// [`create_started_container`] exactly.
async fn create_and_start_reuse_sandbox(
    backend: &Arc<dyn SandboxBackend>,
    container: &Container,
    identity: &crate::reuse::Identity,
    env: &[(String, String)],
) -> Result<(Box<dyn SandboxHandle>, Vec<(u16, u16)>)> {
    let mut last_conflict: Option<RightsizeError> = None;

    for _ in 0..PORT_BIND_ATTEMPTS {
        let mapped_ports = allocate_ports(&container.exposed_ports)?;

        let mut spec = ContainerSpec {
            name: identity.name.clone(),
            image: container.image.clone(),
            env: env.to_vec(),
            command: container.command.clone(),
            ports: mapped_ports
                .iter()
                .map(|&(guest_port, host_port)| crate::model::PortBinding {
                    host_port,
                    guest_port,
                })
                .collect(),
            mounts: container.mounts.clone(),
            network_id: None,
            aliases: container.aliases.clone(),
            run_id: RunId::value().to_string(),
            memory_limit_mb: container.memory_limit_mb,
            keep_alive: true,
        };
        if let Some(customizer) = &container.spec_customizer {
            let lookup: std::collections::HashMap<u16, u16> =
                mapped_ports.iter().copied().collect();
            let mapped_fn = move |guest: u16| -> u16 {
                *lookup
                    .get(&guest)
                    .expect("customizer looked up an unexposed port")
            };
            spec = customizer(spec, &mapped_fn);
        }
        spec.env = dedup_env_last_wins(spec.env);

        let handle = match backend.create(spec).await {
            Ok(h) => h,
            Err(e) => {
                release_ports(&mapped_ports);
                return Err(e);
            }
        };
        match backend.start(handle.as_ref()).await {
            Ok(()) => return Ok((handle, mapped_ports)),
            Err(e) => {
                let _ = backend.stop(handle.as_ref()).await;
                let _ = backend.remove(handle.as_ref()).await;
                release_ports(&mapped_ports);
                if is_port_bind_conflict(&e) {
                    last_conflict = Some(e);
                    continue;
                }
                return Err(e);
            }
        }
    }

    Err(RightsizeError::Backend(format!(
        "Could not bind free host ports for {} after {PORT_BIND_ATTEMPTS} attempts — another \
         process kept grabbing the allocated ports first; if this persists, check for a port \
         scanner/leaked process racing the allocator on this host{}",
        Container::describe(&identity.name, &container.image),
        last_conflict
            .map(|c| format!(" (last conflict: {c})"))
            .unwrap_or_default(),
    )))
}

/// Creates a brand-new reuse sandbox: allocates ports, creates+starts it under the
/// identity-derived `rz-reuse-<12hex>` name with `keep_alive: true` (retrying on a
/// host-port bind conflict — see [`create_and_start_reuse_sandbox`]), runs the wait
/// strategy, and — only on success — writes the registry file. Any failure after
/// resources are allocated tears the sandbox down for real (never through a
/// keep_alive guard's own `stop()`, which would leave a possibly-broken sandbox
/// running with no registry entry pointing at it — an actual leak, not a feature).
async fn create_fresh_reuse(
    container: &Container,
    backend: &Arc<dyn SandboxBackend>,
    identity: &crate::reuse::Identity,
    env: &[(String, String)],
    registry: &crate::reuse::Registry,
) -> Result<ContainerGuard> {
    let (handle, mapped_ports) =
        create_and_start_reuse_sandbox(backend, container, identity, env).await?;

    let raw = RawWaitTarget {
        host: "127.0.0.1",
        mapped_ports: &mapped_ports,
        exposed_ports: &container.exposed_ports,
        backend,
        handle: handle.as_ref(),
        name: &identity.name,
        image: &container.image,
    };
    if let Err(e) = container.wait_strategy.wait_until_ready(&raw).await {
        let _ = backend.stop(handle.as_ref()).await;
        let _ = backend.remove(handle.as_ref()).await;
        release_ports(&mapped_ports);
        return Err(e);
    }

    // Success: write the registry BEFORE handing back the guard — best-effort; a
    // write failure here shouldn't fail a boot that already succeeded (the next
    // start() attempt just won't find a registry and will create fresh again,
    // which is safe, just not the reuse win this call almost delivered).
    let entry = crate::reuse::RegistryEntry {
        name: identity.name.clone(),
        image: container.image.clone(),
        ports: mapped_ports
            .iter()
            .map(|&(guest_port, host_port)| (guest_port.to_string(), host_port))
            .collect(),
        created_iso: crate::reuse::now_iso8601(),
        backend: backend.name().to_string(),
    };
    let _ = registry.write_atomic(&entry);

    let diagnostics_handle_id = handle.id().to_string();
    let diagnostics_spec = handle.spec().clone();
    let diagnostics_ports = mapped_ports.clone();
    let guard = ContainerGuard {
        handle: Some(handle),
        backend: backend.clone(),
        mapped_ports: Mutex::new(mapped_ports),
        network: None,
        image: container.image.clone(),
        exposed_ports: container.exposed_ports.clone(),
        name: identity.name.clone(),
        ledger_name: identity.name.clone(),
        keep_alive: true,
        reaper_cache_dir_override: container.reaper_cache_dir_override.clone(),
    };
    crate::diagnostics::register(
        &guard.ledger_name,
        &guard.image,
        guard.host(),
        diagnostics_ports,
        backend.clone(),
        &diagnostics_handle_id,
        diagnostics_spec,
    );

    if let Some(post_start) = &container.post_start {
        if let Err(e) = post_start(&guard).await {
            let handle_ref = guard.handle_ref();
            let _ = backend.stop(handle_ref).await;
            let _ = backend.remove(handle_ref).await;
            registry.delete();
            for &(_, host_port) in guard
                .mapped_ports
                .lock()
                .expect("mapped_ports mutex poisoned")
                .iter()
            {
                free_ports().release(host_port);
            }
            return Err(e);
        }
    }

    Ok(guard)
}

/// The RAII guard for a running container. Dropping it without calling
/// [`ContainerGuard::stop`] still tears the container down — see the module docs for
/// the two-tier cleanup story.
pub struct ContainerGuard {
    handle: Option<Box<dyn SandboxHandle>>,
    backend: Arc<dyn SandboxBackend>,
    mapped_ports: Mutex<Vec<(u16, u16)>>,
    network: Option<Arc<Network>>,
    image: String,
    exposed_ports: Vec<u16>,
    /// `SandboxHandle::id()` at creation time — the backend-native id (msb: the
    /// same as `ledger_name`; docker: the daemon-assigned container id). Used only
    /// for internal error text ([`Self::describe`]); the public [`Self::name`]
    /// accessor and the diagnostics report both use `ledger_name` instead, since
    /// that's the name a caller can actually act on (e.g. `docker logs <name>`).
    name: String,
    /// The reaping ledger's own name for this sandbox (`ContainerSpec::name`,
    /// e.g. `rz-<run-id>-<seq>`) — see [`Container::start`]'s doc at the capture
    /// site for why this differs from `name` (the raw `SandboxHandle::id()`) on
    /// the docker backend.
    ledger_name: String,
    /// Mirrors `ContainerSpec::keep_alive` — a reuse sandbox is kept out of every
    /// own-process automatic cleanup path (see [`Drop`]'s impl below).
    keep_alive: bool,
    /// Carries [`Container::with_reaper_cache_dir_override`]'s value across into
    /// `stop_inner`/`Drop`'s own `crate::reaper::after_stop` calls — see that
    /// builder's doc. `None` (the real cache dir) for every real caller.
    reaper_cache_dir_override: Option<std::path::PathBuf>,
}

impl ContainerGuard {
    fn handle_ref(&self) -> &dyn SandboxHandle {
        self.handle
            .as_deref()
            .expect("ContainerGuard invariant: handle is only None after being consumed by stop()")
    }

    fn as_network_member(&self) -> Arc<dyn NetworkMember> {
        Arc::new(GuardMemberSnapshot {
            mapped_ports: self
                .mapped_ports
                .lock()
                .expect("mapped_ports mutex poisoned")
                .clone(),
        })
    }

    /// The network this container joined, if any — a module may use this to
    /// resolve a sibling's alias from the guard itself rather than needing to keep a
    /// separate `Network` reference around.
    pub fn network(&self) -> Option<&Arc<Network>> {
        self.network.as_ref()
    }

    /// The host address published ports are reachable on — always loopback.
    pub fn host(&self) -> &str {
        "127.0.0.1"
    }

    /// The backend-facing sandbox name (e.g. `rz-<run-id>-<seq>`, or
    /// `rz-reuse-<12hex>` for a reuse container — see [`Container::reuse`]). This
    /// is always the human-readable ledger/reaping name, on every backend — never
    /// the docker daemon's opaque container id — matching what the diagnostics
    /// report and the reaping ledger both name this sandbox.
    // Deliberately returns `ledger_name`, not the field literally called `name`
    // (which holds the raw `SandboxHandle::id()` — see that field's own doc):
    // the ledger/human name is the one a caller can act on, and is what this
    // accessor has always been documented to return.
    #[allow(clippy::misnamed_getters)]
    pub fn name(&self) -> &str {
        &self.ledger_name
    }

    /// The host port `guest_port` is published on.
    ///
    /// Distinguishes two failure causes: if this guard isn't running (stopped, or never
    /// successfully started), the error says so; if it IS running but `guest_port` was
    /// never declared via `with_exposed_ports`, the error says the port isn't exposed.
    pub fn get_mapped_port(&self, guest_port: u16) -> Result<u16> {
        let mapped = self
            .mapped_ports
            .lock()
            .expect("mapped_ports mutex poisoned");
        if let Some(&(_, host_port)) = mapped.iter().find(|&&(g, _)| g == guest_port) {
            return Ok(host_port);
        }
        if !self.is_running() {
            Err(RightsizeError::Backend(format!(
                "Cannot get mapped port {guest_port} on {}: the container is not running — call \
                 start() first, or check that it did not stop/fail after start()",
                self.describe()
            )))
        } else {
            Err(RightsizeError::Backend(format!(
                "Port {guest_port} is not exposed on {} — call with_exposed_ports({guest_port}) \
                 before start(), or check exposed_ports for the port you actually declared",
                self.describe()
            )))
        }
    }

    /// The full logs captured so far. Requires the container to be running.
    pub async fn logs(&self) -> Result<String> {
        self.backend.logs(self.require_handle()?).await
    }

    /// Runs `cmd` inside the running container and returns its exit code and captured
    /// output.
    pub async fn exec(&self, cmd: &[&str]) -> Result<ExecResult> {
        let cmd: Vec<String> = cmd.iter().map(|s| s.to_string()).collect();
        self.backend.exec(self.require_handle()?, &cmd).await
    }

    /// Checkpoints this RUNNING container: commits its current filesystem state into
    /// a new image via the active backend's `commit_to_image`, and returns a
    /// [`Checkpoint`] carrying that image's ref plus this container's full spec (see
    /// [`Checkpoint`]'s own doc for the "filesystem capture, not memory" semantics
    /// and [`Container::from_checkpoint`] for restoring from the result).
    ///
    /// Gated on [`crate::backend::Capabilities::checkpoint`] BEFORE any backend
    /// call: on a backend that doesn't support it (microsandbox today), this returns
    /// [`RightsizeError::CheckpointUnsupported`] without ever reaching
    /// `commit_to_image`. Requires this guard to currently be running — a state
    /// error otherwise, same shape as [`Self::exec`]/[`Self::logs`].
    pub async fn checkpoint(&self) -> Result<Checkpoint> {
        if !self.backend.capabilities().checkpoint {
            return Err(RightsizeError::CheckpointUnsupported {
                backend: self.backend.name().to_string(),
            });
        }
        let handle = self.require_handle()?;
        let image_ref = checkpoint::generate_image_ref();
        self.backend.commit_to_image(handle, &image_ref).await?;
        Ok(Checkpoint {
            image_ref,
            spec: handle.spec().clone(),
        })
    }

    /// Streams log lines to `consumer` as they're produced. Closing (or dropping) the
    /// returned [`crate::backend::FollowHandle`] halts delivery — no further lines
    /// reach `consumer` afterward, even if the container keeps running.
    pub async fn follow_output(
        &self,
        consumer: impl Fn(String) + Send + Sync + 'static,
    ) -> Result<crate::backend::FollowHandle> {
        self.backend
            .follow_logs(self.require_handle()?, Box::new(consumer))
            .await
    }

    /// True from a successful `start()` until `stop()`; false before the first `start()`
    /// and after.
    pub fn is_running(&self) -> bool {
        self.handle.is_some()
    }

    fn require_handle(&self) -> Result<&dyn SandboxHandle> {
        self.handle.as_deref().ok_or_else(|| {
            RightsizeError::Backend(format!(
                "{} is not running — call start() first",
                self.describe()
            ))
        })
    }

    fn describe(&self) -> String {
        Container::describe(&self.name, &self.image)
    }

    /// Happy-path explicit teardown: stops and removes the container on the backend
    /// (both best-effort — errors are swallowed, since a failed cleanup step
    /// shouldn't block release of the ones after it), then releases its mapped host
    /// ports. Idempotent — a no-op if already stopped
    /// (never re-calls the backend, never double-releases ports), and consumes the
    /// guard so it cannot be used again after this returns.
    pub async fn stop(mut self) -> Result<()> {
        self.stop_inner().await;
        Ok(())
    }

    /// The actual teardown logic, factored out so both `stop(self)` and `Drop` (via the
    /// synchronous fallback) converge on the same "idempotent, ports-then-clear"
    /// contract — `Drop` cannot call this directly (it's async), but it follows the
    /// same shape with blocking primitives instead.
    async fn stop_inner(&mut self) {
        let Some(handle) = self.handle.take() else {
            return; // already stopped (or never started): no-op, no backend call.
        };
        // The diagnostics registry's own "no longer live" moment — mirrors
        // `crate::reaper::after_stop` below, but applies to BOTH branches (keep_alive
        // or not): a reuse sandbox stays alive on the backend, but this guard no
        // longer holds a live handle for it, so it drops out of "what THIS process
        // can currently report on" either way.
        crate::diagnostics::deregister(&self.ledger_name);
        if self.keep_alive {
            // Reuse containers: stop() is the feature's own contract — the sandbox
            // is LEFT RUNNING, and only in-process bookkeeping is cleared. No
            // backend.stop/remove call (that's the whole point), no ledger touch
            // (never listed there in the first place), and no port release: the
            // sandbox is still bound to those host ports for real, and releasing
            // them here would let an unrelated container in this same process grab
            // one out from under it. Mirrors `Drop`'s own keep_alive short-circuit
            // below. `mapped_ports` itself IS in-process bookkeeping, though, so it
            // still gets cleared — `get_mapped_port` must agree with `is_running()`
            // that this guard is no longer live, not keep resolving a port for a
            // handle it no longer holds.
            self.mapped_ports
                .lock()
                .expect("mapped_ports mutex poisoned")
                .clear();
            return;
        }
        let _ = self.backend.stop(handle.as_ref()).await;
        let _ = self.backend.remove(handle.as_ref()).await;
        crate::reaper::after_stop(
            &self.ledger_name,
            self.keep_alive,
            self.reaper_cache_dir_override.as_deref(),
        );
        let mut mapped = self
            .mapped_ports
            .lock()
            .expect("mapped_ports mutex poisoned");
        for &(_, host_port) in mapped.iter() {
            free_ports().release(host_port);
        }
        mapped.clear();
    }
}

struct GuardMemberSnapshot {
    mapped_ports: Vec<(u16, u16)>,
}
impl NetworkMember for GuardMemberSnapshot {
    fn is_running(&self) -> bool {
        true // only constructed for a guard that just successfully started.
    }
    fn mapped_ports(&self) -> Vec<(u16, u16)> {
        self.mapped_ports.clone()
    }
}

impl Drop for ContainerGuard {
    fn drop(&mut self) {
        // Best-effort SYNCHRONOUS fallback (decision 1). MUST NOT panic; MUST work with
        // no Tokio runtime in context.
        let Some(handle) = self.handle.take() else {
            return; // already stopped via stop(self): nothing to do.
        };
        // Synchronous, unlike the ledger update below — the diagnostics registry is
        // an in-memory `Mutex<Vec<_>>`, not a file, so there's no reason to defer
        // this to the cleanup thread the way `crate::reaper::after_stop` is: this
        // guard is done being "live" the moment Drop starts, regardless of whether
        // the async backend teardown below has run yet.
        crate::diagnostics::deregister(&self.ledger_name);
        if self.keep_alive {
            // A reuse sandbox must survive this guard's own automatic teardown
            // entirely — no port release (the container keeps running bound to
            // them; releasing here would let an unrelated container grab the same
            // host port out from under it) and no cleanup-thread enqueue (which
            // would stop+remove a container meant to outlive this process). See
            // `ContainerSpec::keep_alive`'s doc — every own-run cleanup path leaves
            // a keep_alive sandbox alone, and this is core's own piece of that.
            return;
        }
        // Release ports synchronously here — FreePorts is a plain std Mutex, no runtime
        // needed — so a dropped-not-stopped guard doesn't leak its ports even if the
        // cleanup thread is slow or (in a crash) never runs at all.
        if let Ok(mut mapped) = self.mapped_ports.lock() {
            for &(_, host_port) in mapped.iter() {
                free_ports().release(host_port);
            }
            mapped.clear();
        }
        // The Drop-path's own update to the reaping ledger — mirrors `stop_inner`'s
        // `crate::reaper::after_stop` call, but deferred to run on the cleanup thread,
        // AFTER `cleanup_sync` has actually been attempted (see `crate::cleanup`'s
        // `after_teardown` doc), not here in `Drop` itself. Without this, a sandbox
        // torn down only via this fallback path (never an explicit `.stop()`) stays
        // listed in `.sandboxes` for the rest of THIS process's life — reachable only
        // by a future sweep/watchdog, never by this run's own clean-shutdown deletion
        // trigger — even though it's already gone.
        let ledger_name = self.ledger_name.clone();
        let reaper_cache_dir_override = self.reaper_cache_dir_override.clone();
        cleanup::enqueue(CleanupJob {
            backend: self.backend.clone(),
            container_id: handle.id().to_string(),
            after_teardown: Some(Box::new(move || {
                crate::reaper::after_stop(
                    &ledger_name,
                    false,
                    reaper_cache_dir_override.as_deref(),
                );
            })),
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::wait::{WaitStrategy, WaitTarget};
    use std::sync::Mutex as StdMutex;
    use std::time::Duration;

    /// `ContainerGuard` deliberately isn't `Debug` (it holds a `Box<dyn SandboxHandle>`
    /// and friends), so `Result::expect_err`/`unwrap_err` don't work directly on
    /// `Result<ContainerGuard, _>` — this pulls the error out by hand.
    fn expect_start_err(result: Result<ContainerGuard>, msg: &str) -> RightsizeError {
        match result {
            Ok(_) => panic!("{msg}: expected an error, got Ok"),
            Err(e) => e,
        }
    }

    struct FakeHandle {
        id: String,
        spec: ContainerSpec,
    }
    impl SandboxHandle for FakeHandle {
        fn id(&self) -> &str {
            &self.id
        }
        fn spec(&self) -> &ContainerSpec {
            &self.spec
        }
    }

    /// A wait strategy that's immediately ready — the fake backend runs nothing to
    /// actually connect to.
    struct ReadyImmediately;
    #[async_trait::async_trait]
    impl WaitStrategy for ReadyImmediately {
        async fn wait_until_ready(&self, _target: &dyn WaitTarget) -> Result<()> {
            Ok(())
        }
        fn with_startup_timeout(self: Box<Self>, _timeout: Duration) -> Box<dyn WaitStrategy> {
            self
        }
    }

    /// A wait strategy that always fails — forces the start()-then-teardown cleanup
    /// path. Captures the mapped port it saw (via `probe`) before failing, so a test
    /// can assert on the *exact* port that was released, not just "some" port.
    struct NeverReady {
        precaptured_port: Arc<StdMutex<Option<u16>>>,
    }
    #[async_trait::async_trait]
    impl WaitStrategy for NeverReady {
        async fn wait_until_ready(&self, target: &dyn WaitTarget) -> Result<()> {
            *self.precaptured_port.lock().unwrap() = Some(target.mapped_port(6379));
            Err(RightsizeError::ContainerLaunch("never ready".to_string()))
        }
        fn with_startup_timeout(self: Box<Self>, _timeout: Duration) -> Box<dyn WaitStrategy> {
            self
        }
    }

    #[derive(Default)]
    struct FakeBackendState {
        created: Vec<ContainerSpec>,
        started: Vec<String>,
        stopped: Vec<String>,
        installed_links: Vec<(String, Vec<crate::backend::NetworkLink>)>,
        /// `(handle id, image_ref)` for every `commit_to_image` call this backend
        /// actually received — the checkpoint gating tests' proof that a capability
        /// refusal never reaches the backend at all.
        committed: Vec<(String, String)>,
    }

    struct FakeBackend {
        state: StdMutex<FakeBackendState>,
        fail_install_network_links: bool,
        fail_ensure_network: bool,
        hardware_isolated: bool,
        checkpoint_capable: bool,
    }
    impl FakeBackend {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                state: StdMutex::new(FakeBackendState::default()),
                fail_install_network_links: false,
                fail_ensure_network: false,
                hardware_isolated: false,
                checkpoint_capable: false,
            })
        }
        fn failing_install_network_links() -> Arc<Self> {
            Arc::new(Self {
                state: StdMutex::new(FakeBackendState::default()),
                fail_install_network_links: true,
                fail_ensure_network: false,
                hardware_isolated: false,
                checkpoint_capable: false,
            })
        }
        fn failing_ensure_network() -> Arc<Self> {
            Arc::new(Self {
                state: StdMutex::new(FakeBackendState::default()),
                fail_install_network_links: false,
                fail_ensure_network: true,
                hardware_isolated: false,
                checkpoint_capable: false,
            })
        }
        /// A fake backend that reports `capabilities().hardware_isolated == true` —
        /// the require_isolation happy path's fixture.
        fn hardware_isolated() -> Arc<Self> {
            Arc::new(Self {
                state: StdMutex::new(FakeBackendState::default()),
                fail_install_network_links: false,
                fail_ensure_network: false,
                hardware_isolated: true,
                checkpoint_capable: false,
            })
        }
        /// A fake backend that reports `capabilities().checkpoint == true` — the
        /// checkpoint happy-path fixture.
        fn checkpoint_capable() -> Arc<Self> {
            Arc::new(Self {
                state: StdMutex::new(FakeBackendState::default()),
                fail_install_network_links: false,
                fail_ensure_network: false,
                hardware_isolated: false,
                checkpoint_capable: true,
            })
        }
    }
    #[async_trait::async_trait]
    impl SandboxBackend for FakeBackend {
        fn name(&self) -> &str {
            "fake"
        }
        fn supports_native_networks(&self) -> bool {
            false
        }
        fn capabilities(&self) -> crate::backend::Capabilities {
            crate::backend::Capabilities {
                hardware_isolated: self.hardware_isolated,
                checkpoint: self.checkpoint_capable,
            }
        }
        async fn create(&self, spec: ContainerSpec) -> Result<Box<dyn SandboxHandle>> {
            self.state.lock().unwrap().created.push(spec.clone());
            Ok(Box::new(FakeHandle {
                id: spec.name.clone(),
                spec,
            }))
        }
        async fn start(&self, handle: &dyn SandboxHandle) -> Result<()> {
            self.state
                .lock()
                .unwrap()
                .started
                .push(handle.id().to_string());
            Ok(())
        }
        async fn stop(&self, handle: &dyn SandboxHandle) -> Result<()> {
            self.state
                .lock()
                .unwrap()
                .stopped
                .push(handle.id().to_string());
            Ok(())
        }
        async fn remove(&self, _handle: &dyn SandboxHandle) -> Result<()> {
            Ok(())
        }
        async fn exec(&self, _handle: &dyn SandboxHandle, cmd: &[String]) -> Result<ExecResult> {
            Ok(ExecResult {
                exit_code: 0,
                stdout: cmd.join(" "),
                stderr: String::new(),
            })
        }
        async fn logs(&self, _handle: &dyn SandboxHandle) -> Result<String> {
            Ok(String::new())
        }
        async fn follow_logs(
            &self,
            _handle: &dyn SandboxHandle,
            _consumer: Box<dyn Fn(String) + Send + Sync>,
        ) -> Result<crate::backend::FollowHandle> {
            unimplemented!("not exercised by this test suite")
        }
        async fn ensure_network(&self, _network_id: &str) -> Result<()> {
            if self.fail_ensure_network {
                return Err(RightsizeError::Backend("ensure_network failed".to_string()));
            }
            Ok(())
        }
        async fn remove_network(&self, _network_id: &str) -> Result<()> {
            Ok(())
        }
        async fn install_network_links(
            &self,
            handle: &dyn SandboxHandle,
            links: &[crate::backend::NetworkLink],
        ) -> Result<()> {
            if self.fail_install_network_links {
                return Err(RightsizeError::unsupported_with_remedy(
                    "network links (no nc in image; try docker)",
                    "fake",
                    "run with a different backend",
                ));
            }
            if !links.is_empty() {
                self.state
                    .lock()
                    .unwrap()
                    .installed_links
                    .push((handle.id().to_string(), links.to_vec()));
            }
            Ok(())
        }
        fn cleanup_sync(&self, _container_id: &str) {}
        fn remove_by_name(&self, _name: &str) {}
        fn watchdog_kill_command(&self) -> Vec<String> {
            vec!["true".to_string()]
        }
        async fn commit_to_image(&self, handle: &dyn SandboxHandle, image_ref: &str) -> Result<()> {
            self.state
                .lock()
                .unwrap()
                .committed
                .push((handle.id().to_string(), image_ref.to_string()));
            Ok(())
        }
    }

    fn container_on(backend: &Arc<FakeBackend>) -> Container {
        Container::new("redis:8.6-alpine")
            .with_backend(backend.clone())
            .waiting_for(ReadyImmediately)
    }

    // U1: port allocate/create/wait/map.
    #[tokio::test]
    async fn u1_start_allocates_ports_creates_spec_waits_and_maps_ports() {
        let backend = FakeBackend::new();
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .with_env("A", "1");
        let guard = c.start().await.expect("start must succeed");

        let spec = {
            let state = backend.state.lock().unwrap();
            assert_eq!(state.created.len(), 1);
            state.created[0].clone()
        };
        assert_eq!(spec.image, "redis:8.6-alpine");
        assert_eq!(spec.env, vec![("A".to_string(), "1".to_string())]);
        assert_eq!(spec.ports.len(), 1);
        assert_eq!(spec.ports[0].guest_port, 6379);
        assert!(spec.ports[0].host_port > 0);
        assert_eq!(
            guard.get_mapped_port(6379).unwrap(),
            spec.ports[0].host_port
        );
        assert!(guard.is_running());

        guard.stop().await.unwrap();
    }

    // require_isolation: a non-isolated backend refuses to start, before any
    // create/network work.
    #[tokio::test]
    async fn require_isolation_on_a_non_isolated_backend_errors_before_any_create() {
        let backend = FakeBackend::new(); // capabilities().hardware_isolated == false
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .require_isolation(true);

        let err = expect_start_err(c.start().await, "require_isolation must refuse to start");
        assert!(
            matches!(err, RightsizeError::IsolationRequired { .. }),
            "{err}"
        );
        let msg = err.to_string();
        assert!(msg.contains("fake"), "{msg}");
        assert!(msg.contains("RIGHTSIZE_BACKEND=microsandbox"), "{msg}");
        assert!(
            backend.state.lock().unwrap().created.is_empty(),
            "no sandbox may be created when isolation is required and unavailable"
        );
    }

    // require_isolation: a hardware-isolated backend starts normally.
    #[tokio::test]
    async fn require_isolation_on_a_hardware_isolated_backend_starts_normally() {
        let backend = FakeBackend::hardware_isolated();
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .require_isolation(true);

        let guard = c.start().await.expect("start must succeed");
        assert!(guard.is_running());
        assert_eq!(backend.state.lock().unwrap().created.len(), 1);

        guard.stop().await.unwrap();
    }

    // require_isolation(false) (the default) never consults capabilities — a
    // non-isolated backend is fine.
    #[tokio::test]
    async fn require_isolation_defaults_to_false_and_does_not_gate_a_normal_start() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.expect("start must succeed");
        guard.stop().await.unwrap();
    }

    // =========================== checkpoint / restore ==============================

    // Capability gating: a backend with checkpoint == false refuses before any
    // backend call — the typed error, and `commit_to_image` is never invoked.
    #[tokio::test]
    async fn checkpoint_refuses_before_any_backend_call_when_capability_is_false() {
        let backend = FakeBackend::new(); // capabilities().checkpoint == false
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.expect("start must succeed");

        let err = guard
            .checkpoint()
            .await
            .expect_err("checkpoint must refuse on a non-checkpoint-capable backend");
        assert!(
            matches!(err, RightsizeError::CheckpointUnsupported { .. }),
            "{err}"
        );
        let msg = err.to_string();
        assert!(msg.contains("fake"), "{msg}");
        assert!(msg.contains("RIGHTSIZE_BACKEND=docker"), "{msg}");
        assert!(
            backend.state.lock().unwrap().committed.is_empty(),
            "commit_to_image must never be called once capability gating refuses"
        );

        guard.stop().await.unwrap();
    }

    // A non-running container: state error, same shape as exec/logs.
    #[tokio::test]
    async fn checkpoint_on_a_non_running_container_is_a_state_error() {
        let backend = FakeBackend::checkpoint_capable();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let mut guard = c.start().await.unwrap();
        guard.stop_inner().await; // stops it in place, without consuming the guard.

        let err = guard
            .checkpoint()
            .await
            .expect_err("checkpoint must refuse on a stopped guard");
        assert!(err.to_string().contains("not running"), "{err}");
        assert!(
            backend.state.lock().unwrap().committed.is_empty(),
            "commit_to_image must never be called on a non-running guard"
        );
    }

    // The returned Checkpoint carries a well-formed imageRef and the full source
    // spec — env, command, exposed ports, memory limit and all.
    #[tokio::test]
    async fn checkpoint_returns_the_image_ref_and_the_full_source_spec() {
        let backend = FakeBackend::checkpoint_capable();
        let c = container_on(&backend)
            .with_env("A", "1")
            .with_exposed_ports(&[6379])
            .with_command(&["redis-server"])
            .with_memory_limit(256);
        let guard = c.start().await.unwrap();

        let cp = guard.checkpoint().await.expect("checkpoint must succeed");

        let tag = cp
            .image_ref
            .strip_prefix("rightsize/checkpoint:")
            .unwrap_or_else(|| {
                panic!(
                    "expected the rightsize/checkpoint: prefix, got {}",
                    cp.image_ref
                )
            });
        assert_eq!(tag.len(), 12, "{}", cp.image_ref);
        assert!(
            tag.chars()
                .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
            "{}",
            cp.image_ref
        );

        assert_eq!(cp.spec.env, vec![("A".to_string(), "1".to_string())]);
        assert_eq!(cp.spec.command, Some(vec!["redis-server".to_string()]));
        assert_eq!(cp.spec.ports.len(), 1);
        assert_eq!(cp.spec.ports[0].guest_port, 6379);
        assert_eq!(cp.spec.memory_limit_mb, Some(256));

        let committed = backend.state.lock().unwrap().committed.clone();
        assert_eq!(committed.len(), 1);
        assert_eq!(committed[0].0, guard.name());
        assert_eq!(committed[0].1, cp.image_ref);

        guard.stop().await.unwrap();
    }

    // `Container::from_checkpoint` applies the checkpoint's image/env/command/
    // exposed-ports/memory-limit as defaults, and an ordinary builder call after it
    // still overrides (command, here — with_command *replaces* rather than appends).
    #[tokio::test]
    async fn from_checkpoint_applies_the_spec_defaults_and_allows_overrides() {
        let source_backend = FakeBackend::checkpoint_capable();
        let source = container_on(&source_backend)
            .with_env("A", "1")
            .with_exposed_ports(&[6379])
            .with_command(&["redis-server"])
            .with_memory_limit(256);
        let source_guard = source.start().await.unwrap();
        let cp = source_guard.checkpoint().await.unwrap();
        source_guard.stop().await.unwrap();

        // Defaults applied, no override.
        let restore_backend = FakeBackend::new();
        let restored = Container::from_checkpoint(&cp)
            .with_backend(restore_backend.clone())
            .waiting_for(ReadyImmediately);
        let restored_guard = restored.start().await.expect("restore must start");
        let created = restore_backend.state.lock().unwrap().created[0].clone();
        assert_eq!(created.image, cp.image_ref);
        assert_eq!(created.env, vec![("A".to_string(), "1".to_string())]);
        assert_eq!(created.command, Some(vec!["redis-server".to_string()]));
        assert_eq!(created.ports.len(), 1);
        assert_eq!(created.ports[0].guest_port, 6379);
        assert_eq!(created.memory_limit_mb, Some(256));
        restored_guard.stop().await.unwrap();

        // Override: a caller's own `.with_command(...)` after `from_checkpoint`
        // replaces the checkpoint spec's command rather than being ignored.
        let override_backend = FakeBackend::new();
        let overridden = Container::from_checkpoint(&cp)
            .with_backend(override_backend.clone())
            .waiting_for(ReadyImmediately)
            .with_command(&["redis-server", "--appendonly", "yes"]);
        let overridden_guard = overridden.start().await.expect("restore must start");
        let created = override_backend.state.lock().unwrap().created[0].clone();
        assert_eq!(
            created.command,
            Some(vec![
                "redis-server".to_string(),
                "--appendonly".to_string(),
                "yes".to_string()
            ])
        );
        overridden_guard.stop().await.unwrap();
    }

    // Interlock with the reaping ledger: a restored container is registered and
    // reaped exactly like any other — its name appears in `.sandboxes` while
    // running, same as `dropping_a_guard_without_stop_removes_it_from_the_reaping_ledger`
    // proves for an ordinary container.
    #[tokio::test]
    async fn restored_container_is_registered_in_the_reaping_ledger_like_any_other() {
        // Test isolation seam: a per-test scratch cache dir for the reaping
        // ledger (see `Container::with_reaper_cache_dir_override`'s doc) — without
        // it, this test's assertion reads the real, process-wide ledger every
        // OTHER concurrently-running test in this binary also writes to.
        let cache_dir = temp_cache_dir("restored-ledger");

        let source_backend = FakeBackend::checkpoint_capable();
        let source = container_on(&source_backend)
            .with_exposed_ports(&[6379])
            .with_reaper_cache_dir_override(cache_dir.clone());
        let source_guard = source.start().await.unwrap();
        let cp = source_guard.checkpoint().await.unwrap();
        source_guard.stop().await.unwrap();

        let restore_backend = FakeBackend::new();
        let restored = Container::from_checkpoint(&cp)
            .with_backend(restore_backend.clone())
            .with_reaper_cache_dir_override(cache_dir.clone())
            .waiting_for(ReadyImmediately);
        let restored_guard = restored.start().await.expect("restore must start");
        let ledger_name = restore_backend.state.lock().unwrap().created[0]
            .name
            .clone();

        let ledger = crate::reaper::Ledger::new(&cache_dir, crate::RunId::value());
        assert!(
            ledger.sandbox_names().contains(&ledger_name),
            "a restored container must be listed in the reaping ledger like any other"
        );

        restored_guard.stop().await.unwrap();
    }

    // Diagnostics registration: a running container is reachable through the report
    // by its unique name (`rz-<run-id>-<seq>`, unique regardless of concurrently
    // running tests sharing the process-wide registry); stop() removes it again.
    #[tokio::test]
    async fn a_started_container_is_registered_for_diagnostics_and_deregistered_on_stop() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.unwrap();
        let name = guard.name().to_string();
        // The report's own header delimiter — plain `report.contains(&name)` would be
        // a substring false-positive against another concurrently-running test's
        // longer name sharing this one as a numeric prefix (e.g. name `..-42` is a
        // substring of a sibling test's `..-420`); the header's trailing `" ("` rules
        // that out, since a longer name's next character there is never `(`.
        let header = format!("-- {name} (");

        let report = crate::diagnostics().await;
        assert!(report.contains(&header), "{report}");

        guard.stop().await.unwrap();
        let report_after_stop = crate::diagnostics().await;
        assert!(
            !report_after_stop.contains(&header),
            "stop() must deregister this container from the diagnostics report"
        );
    }

    /// A wait strategy that captures a `diagnostics()` snapshot from mid-wait (before
    /// deciding readiness), then always succeeds — proves registration happens only
    /// AFTER the wait passes, not as soon as the guard exists.
    struct CaptureDiagnosticsThenReady {
        mid_wait_report: Arc<StdMutex<Option<String>>>,
    }
    #[async_trait::async_trait]
    impl WaitStrategy for CaptureDiagnosticsThenReady {
        async fn wait_until_ready(&self, _target: &dyn WaitTarget) -> Result<()> {
            *self.mid_wait_report.lock().unwrap() = Some(crate::diagnostics().await);
            Ok(())
        }
        fn with_startup_timeout(self: Box<Self>, _timeout: Duration) -> Box<dyn WaitStrategy> {
            self
        }
    }

    // Registration timing: a container must never be diagnosable while its readiness
    // wait is still running — only a FULLY-successful start (wait passed) makes it
    // reachable through the report. Mirrors the Kotlin port's resolution of the same
    // finding (register only after the wait succeeds).
    #[tokio::test]
    async fn a_container_is_not_diagnosable_until_its_readiness_wait_succeeds() {
        let backend = FakeBackend::new();
        let mid_wait_report = Arc::new(StdMutex::new(None));
        let c = Container::new("redis:8.6-alpine")
            .with_backend(backend.clone())
            .waiting_for(CaptureDiagnosticsThenReady {
                mid_wait_report: mid_wait_report.clone(),
            })
            .with_exposed_ports(&[6379]);

        let guard = c.start().await.expect("this wait strategy always succeeds");
        let name = guard.name().to_string();
        let header = format!("-- {name} (");

        let captured = mid_wait_report
            .lock()
            .unwrap()
            .clone()
            .expect("wait strategy must have run");
        assert!(
            !captured.contains(&header),
            "a container must not be diagnosable while its readiness wait is still \
             running: {captured}"
        );

        let report_after_start = crate::diagnostics().await;
        assert!(report_after_start.contains(&header), "{report_after_start}");

        guard.stop().await.unwrap();
    }

    // Registration timing, failure branch: when the wait never succeeds, the
    // container must never become diagnosable at all — there is nothing for the
    // failure path's teardown to deregister.
    #[tokio::test]
    async fn a_container_that_never_becomes_ready_is_never_diagnosable() {
        let backend = FakeBackend::new();
        let precaptured_port = Arc::new(StdMutex::new(None));
        let c = Container::new("redis:8.6-alpine")
            .with_backend(backend.clone())
            .waiting_for(NeverReady {
                precaptured_port: precaptured_port.clone(),
            })
            .with_exposed_ports(&[6379]);

        let err = expect_start_err(c.start().await, "wait strategy must fail start()");
        assert!(err.to_string().contains("never ready"), "{err}");

        let name = backend.state.lock().unwrap().created[0].name.clone();
        let header = format!("-- {name} (");
        let report = crate::diagnostics().await;
        assert!(
            !report.contains(&header),
            "a container that never became ready must never appear in the \
             diagnostics report: {report}"
        );
    }

    // Not-running vs not-exposed disambiguation.
    #[tokio::test]
    async fn get_mapped_port_reports_not_running_after_stop_clears_the_mappings() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.unwrap();
        assert!(guard.get_mapped_port(6379).unwrap() > 0);
        guard.stop().await.unwrap();

        // The guard was consumed by stop(); re-derive a fresh one to exercise the
        // stopped-state error message via a boot that we then let the wait strategy
        // fail after capturing the port — simpler: build a second guard, call stop via
        // an explicit helper path. Since `stop` consumes `self`, we instead assert the
        // not-running message shape directly against a guard we stop through the
        // internal helper without consuming it, mirroring what Drop/stop share.
        let backend2 = FakeBackend::new();
        let c2 = container_on(&backend2).with_exposed_ports(&[6379]);
        let mut guard2 = c2.start().await.unwrap();
        guard2.stop_inner().await;
        let err = guard2.get_mapped_port(6379).unwrap_err().to_string();
        assert!(err.contains("not running"), "{err}");
        assert!(!err.contains("not exposed"), "{err}");
    }

    // U8 (part 2): not-exposed for a port never declared.
    #[tokio::test]
    async fn get_mapped_port_reports_not_exposed_for_an_undeclared_port() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.unwrap();
        let err = guard.get_mapped_port(9999).unwrap_err().to_string();
        assert!(err.contains("not exposed"), "{err}");
        guard.stop().await.unwrap();
    }

    // U2: network links installed on the new member for the running sibling, before
    // wait, after registering siblings only (never self).
    #[tokio::test]
    async fn u2_starting_on_a_network_installs_links_to_running_siblings() {
        let backend = FakeBackend::new();
        let net = Arc::new(Network::new_network());
        let stub = container_on(&backend)
            .with_exposed_ports(&[8888])
            .with_network(&net)
            .with_network_aliases(&["configuration-stub"]);
        let stub_guard = stub.start().await.unwrap();

        let app = container_on(&backend)
            .with_exposed_ports(&[8080])
            .with_network(&net);
        let app_guard = app.start().await.unwrap();

        let (consumer_id, links) = {
            let state = backend.state.lock().unwrap();
            assert_eq!(state.installed_links.len(), 1);
            state.installed_links[0].clone()
        };
        assert_eq!(
            consumer_id,
            backend.state.lock().unwrap().created.last().unwrap().name
        );
        assert_eq!(
            links,
            vec![crate::backend::NetworkLink {
                alias: "configuration-stub".to_string(),
                guest_port: 8888,
                target_host_port: stub_guard.get_mapped_port(8888).unwrap(),
            }]
        );
        assert_eq!(
            net.resolve("configuration-stub", 8888).unwrap(),
            "configuration-stub:8888"
        );
        assert!(net.resolve("nope", 1).is_err());

        app_guard.stop().await.unwrap();
        stub_guard.stop().await.unwrap();
    }

    // U8 (part 3): a single container on a network installs no links but is still
    // registered, so a later joiner links back to it.
    #[tokio::test]
    async fn single_container_on_network_installs_no_links_but_is_registered() {
        let backend = FakeBackend::new();
        let net = Arc::new(Network::new_network());
        let solo = container_on(&backend)
            .with_exposed_ports(&[9999])
            .with_network(&net)
            .with_network_aliases(&["solo"]);
        let solo_guard = solo.start().await.unwrap();
        assert!(
            backend.state.lock().unwrap().installed_links.is_empty(),
            "a lone container must not link to itself"
        );

        let joiner = container_on(&backend)
            .with_exposed_ports(&[8080])
            .with_network(&net);
        let joiner_guard = joiner.start().await.unwrap();
        let (_, links) = {
            let state = backend.state.lock().unwrap();
            assert_eq!(state.installed_links.len(), 1);
            state.installed_links[0].clone()
        };
        assert_eq!(
            links,
            vec![crate::backend::NetworkLink {
                alias: "solo".to_string(),
                guest_port: 9999,
                target_host_port: solo_guard.get_mapped_port(9999).unwrap(),
            }]
        );

        joiner_guard.stop().await.unwrap();
        solo_guard.stop().await.unwrap();
    }

    // U5 (injection point 1/2): wait-strategy failure stops the container and releases
    // its host ports — proven against FreePorts' own issued_view(), not just "the port
    // is bindable" (which would pass even if release were never called, since the fake
    // backend never binds OS ports either).
    #[tokio::test]
    async fn u5_wait_strategy_failure_stops_the_container_and_releases_ports() {
        let backend = FakeBackend::new();
        let precaptured_port = Arc::new(StdMutex::new(None));
        let c = Container::new("redis:8.6-alpine")
            .with_backend(backend.clone())
            .waiting_for(NeverReady {
                precaptured_port: precaptured_port.clone(),
            })
            .with_exposed_ports(&[6379]);

        let err = expect_start_err(c.start().await, "wait strategy must fail start()");
        assert!(err.to_string().contains("never ready"), "{err}");

        let name = backend.state.lock().unwrap().created[0].name.clone();
        assert!(
            backend.state.lock().unwrap().stopped.contains(&name),
            "started container must be stopped when the wait strategy fails"
        );

        let port = precaptured_port
            .lock()
            .unwrap()
            .expect("wait strategy must have observed a real mapped port");
        assert!(port > 0);
        assert!(
            !free_ports().issued_view().contains(&port),
            "port {port} must be released by the wait-strategy-failure cleanup path"
        );
    }

    // U5 (injection point 2/2): install_network_links failure stops the container too,
    // and — like the wait-strategy injection point above — releases its host ports back
    // to FreePorts. Proven against issued_view() directly, not just "stop was called",
    // so this fails the same way U5's first injection point would if `start()` ever
    // stopped releasing ports on this seam specifically.
    #[tokio::test]
    async fn u5_install_network_links_failure_stops_the_container() {
        let backend = FakeBackend::failing_install_network_links();
        let net = Arc::new(Network::new_network());
        let c = container_on(&backend)
            .with_exposed_ports(&[8080])
            .with_network(&net);

        let err = expect_start_err(
            c.start().await,
            "install_network_links failure must propagate",
        );
        assert!(err.to_string().contains("nc"), "{err}");

        let created = backend.state.lock().unwrap().created[0].clone();
        assert!(
            backend
                .state
                .lock()
                .unwrap()
                .stopped
                .contains(&created.name),
            "started container must be stopped on link-install failure"
        );

        let port = created
            .ports
            .first()
            .expect("spec must carry the allocated host port")
            .host_port;
        assert!(
            !free_ports().issued_view().contains(&port),
            "port {port} must be released when install_network_links fails, \
             not just when the wait strategy fails"
        );
    }

    // `before_ensure_network` appends to the reaping ledger's `.networks` file BEFORE
    // `backend.ensure_network` is even called, mirroring the sandbox-name discipline in
    // `create_started_container`. Unlike that path, a failed `ensure_network` used to
    // have no matching cleanup — this proves the fix: a failed ensure_network must not
    // leave a phantom `.networks` entry behind (which would otherwise block this run's
    // own clean-shutdown deletion trigger for the rest of the process).
    #[tokio::test]
    async fn ensure_network_failure_does_not_leave_a_phantom_networks_ledger_entry() {
        // Test isolation seam: see `restored_container_is_registered_in_the_
        // reaping_ledger_like_any_other`'s own comment.
        let cache_dir = temp_cache_dir("ensure-network-failure-ledger");

        let backend = FakeBackend::failing_ensure_network();
        let net = Arc::new(Network::new_network());
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .with_network(&net)
            .with_reaper_cache_dir_override(cache_dir.clone());

        let err = expect_start_err(c.start().await, "ensure_network failure must propagate");
        assert!(err.to_string().contains("ensure_network failed"), "{err}");

        let ledger = crate::reaper::Ledger::new(&cache_dir, crate::RunId::value());
        assert!(
            !ledger.network_ids().contains(&net.id().to_string()),
            "a failed ensure_network must not leave a phantom .networks entry behind"
        );
    }

    // U7: stop is a no-op before start, and Drop after an explicit stop() doesn't
    // double-release ports or double-call the backend (stop() itself can't be called
    // twice in Rust: it consumes the guard by value, so "calling stop() twice" is a
    // compile-time guarantee here instead of a runtime assertion — what remains to
    // prove is that Drop, which still runs after stop() returns, doesn't redo any of
    // stop()'s work).
    #[tokio::test]
    async fn u7_stop_is_idempotent_across_the_stop_then_drop_sequence() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.unwrap();
        let name = backend.state.lock().unwrap().created[0].name.clone();
        let mapped_port = guard.get_mapped_port(6379).unwrap();

        guard.stop().await.unwrap(); // guard is consumed here; Drop still runs after.

        assert_eq!(
            backend
                .state
                .lock()
                .unwrap()
                .stopped
                .iter()
                .filter(|n| **n == name)
                .count(),
            1,
            "backend.stop must be called exactly once"
        );
        assert!(
            !free_ports().issued_view().contains(&mapped_port),
            "port must be released by stop()"
        );
    }

    #[tokio::test]
    async fn stop_before_start_is_a_no_op() {
        // There is no "unstarted guard" in this API shape (a guard only exists after a
        // successful start()) — the closest analogue is starting then immediately
        // stopping via the internal helper twice, proving the second call is a no-op.
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let mut guard = c.start().await.unwrap();
        let name = backend.state.lock().unwrap().created[0].name.clone();

        guard.stop_inner().await;
        assert_eq!(
            backend
                .state
                .lock()
                .unwrap()
                .stopped
                .iter()
                .filter(|n| **n == name)
                .count(),
            1
        );
        guard.stop_inner().await; // second call: must not re-call backend.stop or double-release.
        assert_eq!(
            backend
                .state
                .lock()
                .unwrap()
                .stopped
                .iter()
                .filter(|n| **n == name)
                .count(),
            1,
            "a second stop must not re-call backend.stop"
        );
        assert!(!guard.is_running());
    }

    // U6: port-bind-conflict retry — a fake backend that fails start with a typed
    // PortBindConflict on attempts 1-2 then succeeds; ports are reallocated per attempt.
    struct PortConflictBackend {
        fail_first: usize,
        conflict: Box<dyn Fn(u16) -> RightsizeError + Send + Sync>,
        created: StdMutex<Vec<ContainerSpec>>,
        started_ports: StdMutex<Vec<u16>>,
        start_attempts: std::sync::atomic::AtomicUsize,
    }
    impl PortConflictBackend {
        fn new(fail_first: usize) -> Arc<Self> {
            Self::with_conflict(fail_first, |port| {
                RightsizeError::Backend(format!(
                    "driver failed programming external connectivity: failed to bind host port 127.0.0.1:{port}/tcp: address already in use"
                ))
            })
        }
        fn with_conflict(
            fail_first: usize,
            conflict: impl Fn(u16) -> RightsizeError + Send + Sync + 'static,
        ) -> Arc<Self> {
            Arc::new(Self {
                fail_first,
                conflict: Box::new(conflict),
                created: StdMutex::new(Vec::new()),
                started_ports: StdMutex::new(Vec::new()),
                start_attempts: std::sync::atomic::AtomicUsize::new(0),
            })
        }
        fn create_count(&self) -> usize {
            self.created.lock().unwrap().len()
        }
    }
    #[async_trait::async_trait]
    impl SandboxBackend for PortConflictBackend {
        fn name(&self) -> &str {
            "port-conflict"
        }
        fn supports_native_networks(&self) -> bool {
            false
        }
        async fn create(&self, spec: ContainerSpec) -> Result<Box<dyn SandboxHandle>> {
            self.created.lock().unwrap().push(spec.clone());
            Ok(Box::new(FakeHandle {
                id: spec.name.clone(),
                spec,
            }))
        }
        async fn start(&self, handle: &dyn SandboxHandle) -> Result<()> {
            let attempt = self.start_attempts.fetch_add(1, Ordering::SeqCst) + 1;
            let port = handle.spec().ports[0].host_port;
            self.started_ports.lock().unwrap().push(port);
            if attempt <= self.fail_first {
                return Err((self.conflict)(port));
            }
            Ok(())
        }
        async fn stop(&self, _handle: &dyn SandboxHandle) -> Result<()> {
            Ok(())
        }
        async fn remove(&self, _handle: &dyn SandboxHandle) -> Result<()> {
            Ok(())
        }
        async fn exec(&self, _handle: &dyn SandboxHandle, _cmd: &[String]) -> Result<ExecResult> {
            Ok(ExecResult {
                exit_code: 0,
                stdout: String::new(),
                stderr: String::new(),
            })
        }
        async fn logs(&self, _handle: &dyn SandboxHandle) -> Result<String> {
            Ok(String::new())
        }
        async fn follow_logs(
            &self,
            _handle: &dyn SandboxHandle,
            _consumer: Box<dyn Fn(String) + Send + Sync>,
        ) -> Result<crate::backend::FollowHandle> {
            unimplemented!()
        }
        async fn ensure_network(&self, _network_id: &str) -> Result<()> {
            Ok(())
        }
        async fn remove_network(&self, _network_id: &str) -> Result<()> {
            Ok(())
        }
        fn cleanup_sync(&self, _container_id: &str) {}
        fn remove_by_name(&self, _name: &str) {}
        fn watchdog_kill_command(&self) -> Vec<String> {
            vec!["true".to_string()]
        }
    }

    #[tokio::test]
    async fn u6_start_retries_with_fresh_host_ports_on_a_bind_conflict() {
        let backend = PortConflictBackend::new(2);
        let c = Container::new("redis:8.6-alpine")
            .with_backend(backend.clone())
            .waiting_for(ReadyImmediately)
            .with_exposed_ports(&[6379]);
        let guard = c.start().await.expect("must eventually succeed");
        assert!(guard.is_running());
        assert_eq!(
            backend.create_count(),
            3,
            "each attempt recreates the container"
        );
        let started_ports = backend.started_ports.lock().unwrap().clone();
        assert_eq!(started_ports.len(), 3, "start attempted three times");
        let distinct: std::collections::HashSet<u16> = started_ports.iter().copied().collect();
        assert_eq!(
            distinct.len(),
            started_ports.len(),
            "ports are reallocated per attempt, not reused after a conflict"
        );
        guard.stop().await.unwrap();
    }

    /// A reuse fresh-create (no registry entry yet) must retry with fresh host
    /// ports on a bind conflict exactly like an ordinary container's `start()`
    /// does (see [`u6_start_retries_with_fresh_host_ports_on_a_bind_conflict`]) —
    /// the sibling ports use the same retry discipline for their own reuse
    /// fresh-create path, and this crate must not silently fail-fast instead on
    /// the very first transient port collision a reuse boot happens to hit.
    #[tokio::test]
    async fn u6_reuse_fresh_create_retries_with_fresh_host_ports_on_a_bind_conflict() {
        let backend = PortConflictBackend::new(2);
        let cache_dir = temp_cache_dir("fresh-create-port-conflict");
        let c = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir)
            .with_reuse_env_override(true)
            .waiting_for(ReadyImmediately)
            .with_exposed_ports(&[6379])
            .reuse(true);
        let guard = c
            .start()
            .await
            .expect("reuse fresh-create must retry and eventually succeed");
        assert!(guard.is_running());
        assert_eq!(
            backend.create_count(),
            3,
            "each attempt recreates the container"
        );
        let started_ports = backend.started_ports.lock().unwrap().clone();
        assert_eq!(started_ports.len(), 3, "start attempted three times");
        let distinct: std::collections::HashSet<u16> = started_ports.iter().copied().collect();
        assert_eq!(
            distinct.len(),
            started_ports.len(),
            "ports are reallocated per attempt, not reused after a conflict"
        );
        guard.stop().await.unwrap();
    }

    #[tokio::test]
    async fn u6_start_retries_on_the_typed_port_bind_conflict_bare_or_nested() {
        let bare = PortConflictBackend::with_conflict(1, |port| RightsizeError::PortBindConflict {
            message: format!("could not bind host port {port}"),
            source: None,
        });
        let c = Container::new("redis:8.6-alpine")
            .with_backend(bare.clone())
            .waiting_for(ReadyImmediately)
            .with_exposed_ports(&[6379]);
        let guard = c
            .start()
            .await
            .expect("must retry exactly once for the typed exception");
        assert!(guard.is_running());
        assert_eq!(bare.create_count(), 2);
        guard.stop().await.unwrap();

        // Wrapped: a typed PortBindConflict nested two levels deep under other
        // PortBindConflicts whose own messages say nothing about ports — only walking
        // the `source` chain (not the string fallback) finds it.
        let nested =
            PortConflictBackend::with_conflict(1, |port| RightsizeError::PortBindConflict {
                message: "start failed".to_string(),
                source: Some(Box::new(RightsizeError::PortBindConflict {
                    message: "io error".to_string(),
                    source: Some(Box::new(RightsizeError::PortBindConflict {
                        message: format!("could not bind host port {port}"),
                        source: None,
                    })),
                })),
            });
        let c = Container::new("redis:8.6-alpine")
            .with_backend(nested.clone())
            .waiting_for(ReadyImmediately)
            .with_exposed_ports(&[6379]);
        let guard = c
            .start()
            .await
            .expect("must unwrap to find a nested typed exception");
        assert!(guard.is_running());
        assert_eq!(nested.create_count(), 2);
        guard.stop().await.unwrap();
    }

    #[tokio::test]
    async fn u6_truth_table_known_phrasings_retry_negative_does_not() {
        let phrasings = [
            "address already in use",
            "port is already allocated",
            "bind: address already in use",
            "Bind for 0.0.0.0:32770 failed: PORT IS ALREADY ALLOCATED",
        ];
        for phrasing in phrasings {
            let backend = PortConflictBackend::with_conflict(1, move |_port| {
                RightsizeError::Backend(phrasing.to_string())
            });
            let c = Container::new("redis:8.6-alpine")
                .with_backend(backend.clone())
                .waiting_for(ReadyImmediately)
                .with_exposed_ports(&[6379]);
            let guard = c.start().await.unwrap_or_else(|e| {
                panic!("must retry and succeed for phrasing '{phrasing}': {e}")
            });
            assert!(guard.is_running());
            assert_eq!(
                backend.create_count(),
                2,
                "must retry exactly once for phrasing: {phrasing}"
            );
            guard.stop().await.unwrap();
        }

        // Negative: an unrelated failure must NOT be treated as a conflict.
        let boom = PortConflictBackend::with_conflict(99, |_port| {
            RightsizeError::Backend("boom".to_string())
        });
        let c = Container::new("redis:8.6-alpine")
            .with_backend(boom.clone())
            .waiting_for(ReadyImmediately)
            .with_exposed_ports(&[6379]);
        let err = expect_start_err(
            c.start().await,
            "a non-conflict exception must fail fast, no retry",
        );
        assert_eq!(err.to_string(), "boom");
        assert_eq!(
            boom.create_count(),
            1,
            "a non-conflict exception must fail fast, no retry"
        );
    }

    #[tokio::test]
    async fn u4_and_u6_start_gives_up_after_the_retry_budget_is_exhausted_and_releases_every_attempts_ports()
     {
        let backend = PortConflictBackend::new(99);
        let c = Container::new("redis:8.6-alpine")
            .with_backend(backend.clone())
            .waiting_for(ReadyImmediately)
            .with_exposed_ports(&[6379]);
        let err = expect_start_err(c.start().await, "must give up after the retry budget");
        assert!(err.to_string().contains("free host ports"), "{err}");

        let started_ports = backend.started_ports.lock().unwrap().clone();
        assert_eq!(
            started_ports.len(),
            PORT_BIND_ATTEMPTS,
            "all attempts must have been tried"
        );
        let distinct: std::collections::HashSet<u16> = started_ports.iter().copied().collect();
        assert_eq!(
            distinct.len(),
            PORT_BIND_ATTEMPTS,
            "each retry must allocate a fresh port"
        );

        let issued = free_ports().issued_view();
        for port in started_ports {
            assert!(
                !issued.contains(&port),
                "port {port} from a discarded attempt must be released — mutation-verified: this is the U4 port-release gate"
            );
        }
    }

    // Memory-limit knob: with_memory_limit reaches the ContainerSpec; None when unset.
    #[tokio::test]
    async fn with_memory_limit_carries_through_to_the_container_spec() {
        let backend = FakeBackend::new();
        let limited = container_on(&backend)
            .with_exposed_ports(&[6379])
            .with_memory_limit(1024);
        let guard = limited.start().await.unwrap();
        assert_eq!(
            backend.state.lock().unwrap().created[0].memory_limit_mb,
            Some(1024)
        );
        guard.stop().await.unwrap();

        let unset = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = unset.start().await.unwrap();
        assert_eq!(
            backend
                .state
                .lock()
                .unwrap()
                .created
                .last()
                .unwrap()
                .memory_limit_mb,
            None
        );
        guard.stop().await.unwrap();
    }

    // dedup_env_last_wins in isolation — last value wins, key keeps the position
    // of its FIRST occurrence: an insertion-ordered map's put never moves an
    // existing key.
    #[test]
    fn dedup_env_last_wins_keeps_first_position_but_last_value() {
        let env = vec![
            ("A".to_string(), "1".to_string()),
            ("B".to_string(), "2".to_string()),
            ("A".to_string(), "override".to_string()),
            ("C".to_string(), "3".to_string()),
            ("B".to_string(), "final".to_string()),
        ];
        let deduped = dedup_env_last_wins(env);
        assert_eq!(
            deduped,
            vec![
                ("A".to_string(), "override".to_string()),
                ("B".to_string(), "final".to_string()),
                ("C".to_string(), "3".to_string()),
            ],
            "A and B must keep their FIRST-occurrence position; each must carry its LAST value"
        );
    }

    #[test]
    fn dedup_env_last_wins_is_a_no_op_on_already_unique_keys() {
        let env = vec![
            ("X".to_string(), "1".to_string()),
            ("Y".to_string(), "2".to_string()),
        ];
        assert_eq!(dedup_env_last_wins(env.clone()), env);
    }

    // Fix 3 (end-to-end through start()): a duplicate key set via with_env twice
    // reaches the backend's ContainerSpec exactly once, with the LAST value, in the
    // position of the FIRST with_env call — proving the dedup runs on the real
    // start() path, not just as a unit in isolation.
    #[tokio::test]
    async fn duplicate_with_env_calls_resolve_last_wins_in_the_spec_reaching_the_backend() {
        let backend = FakeBackend::new();
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .with_env("MODE", "first")
            .with_env("OTHER", "x")
            .with_env("MODE", "second");
        let guard = c.start().await.unwrap();

        let spec = backend.state.lock().unwrap().created[0].clone();
        assert_eq!(
            spec.env,
            vec![
                ("MODE".to_string(), "second".to_string()),
                ("OTHER".to_string(), "x".to_string()),
            ],
            "MODE must appear exactly once, in its first-occurrence position, with its last value"
        );
        guard.stop().await.unwrap();
    }

    // Fix 3 (spec-customizer interaction): a customizer pushing a key that
    // duplicates one already set via with_env must ALSO resolve last-wins — the
    // dedup runs after the customizer, not just on the pre-customizer env.
    #[tokio::test]
    async fn a_spec_customizer_overriding_an_existing_key_also_resolves_last_wins() {
        let backend = FakeBackend::new();
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .with_env("KAFKA_ADVERTISED_LISTENERS", "PLACEHOLDER")
            .with_spec_customizer(|mut spec, _mapped| {
                spec.env.push((
                    "KAFKA_ADVERTISED_LISTENERS".to_string(),
                    "PLAINTEXT://127.0.0.1:9999".to_string(),
                ));
                spec
            });
        let guard = c.start().await.unwrap();

        let spec = backend.state.lock().unwrap().created[0].clone();
        assert_eq!(
            spec.env,
            vec![(
                "KAFKA_ADVERTISED_LISTENERS".to_string(),
                "PLAINTEXT://127.0.0.1:9999".to_string()
            )],
            "the customizer's later push must win, deduped to a single entry"
        );
        guard.stop().await.unwrap();
    }

    // U3: exec/get_mapped_port on a not-running guard errors.
    #[tokio::test]
    async fn u3_exec_and_mapped_port_require_a_running_container() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let mut guard = c.start().await.unwrap();
        guard.stop_inner().await;

        assert!(guard.exec(&["ls"]).await.is_err());
        assert!(guard.get_mapped_port(6379).is_err());
    }

    #[tokio::test]
    async fn exec_returns_the_backends_result() {
        let backend = FakeBackend::new();
        let c = container_on(&backend);
        let guard = c.start().await.unwrap();
        let result = guard.exec(&["ls", "-la"]).await.unwrap();
        assert_eq!(result.stdout, "ls -la");
        guard.stop().await.unwrap();
    }

    #[tokio::test]
    async fn dropping_a_guard_without_stop_releases_its_ports_synchronously() {
        let backend = FakeBackend::new();
        let c = container_on(&backend).with_exposed_ports(&[6379]);
        let guard = c.start().await.unwrap();
        let port = guard.get_mapped_port(6379).unwrap();
        assert!(free_ports().issued_view().contains(&port));

        drop(guard); // no explicit stop(): Drop's synchronous fallback must still release the port.

        assert!(
            !free_ports().issued_view().contains(&port),
            "Drop must release mapped ports synchronously even without an explicit stop()"
        );
    }

    // The Drop-path's own opportunity to update the reaping ledger (see
    // `crate::cleanup`'s `after_teardown` doc) — without this, a sandbox torn down
    // only via `Drop` (never an explicit `.stop()`) stays listed in `.sandboxes` for
    // the rest of this process's life even though it's already gone, and this run's
    // own clean-shutdown deletion trigger never fires for it.
    #[tokio::test]
    async fn dropping_a_guard_without_stop_removes_it_from_the_reaping_ledger() {
        // Test isolation seam: see `restored_container_is_registered_in_the_
        // reaping_ledger_like_any_other`'s own comment — this test's ledger file
        // is now exclusive to it, not the real, process-wide one every other test
        // in this binary also writes to.
        let cache_dir = temp_cache_dir("dropping-guard-ledger");

        let backend = FakeBackend::new();
        let c = container_on(&backend)
            .with_exposed_ports(&[6379])
            .with_reaper_cache_dir_override(cache_dir.clone());
        let guard = c.start().await.unwrap();
        let ledger_name = backend.state.lock().unwrap().created[0].name.clone();

        let ledger = crate::reaper::Ledger::new(&cache_dir, crate::RunId::value());
        assert!(
            ledger.sandbox_names().contains(&ledger_name),
            "before_create must have listed the sandbox in the reaping ledger"
        );

        drop(guard); // no explicit stop(): the cleanup thread's fallback path runs instead.

        // The cleanup thread updates the ledger on a background thread, genuinely
        // asynchronously relative to this test (not a cross-test race — this
        // ledger file is exclusive to this test now) — poll until this entry is
        // gone.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while ledger.sandbox_names().contains(&ledger_name) && std::time::Instant::now() < deadline
        {
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(
            !ledger.sandbox_names().contains(&ledger_name),
            "Drop's cleanup-thread fallback must remove the sandbox from the reaping ledger \
             too, not just tear it down on the backend"
        );
    }

    // ======================================================================
    // Container reuse
    // ======================================================================

    fn temp_cache_dir(label: &str) -> std::path::PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "rz-reuse-container-test-{label}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[derive(Default)]
    struct ReuseFakeState {
        created: Vec<ContainerSpec>,
        started: Vec<String>,
        stopped: Vec<String>,
        removed: Vec<String>,
        removed_by_name: Vec<String>,
        running: std::collections::HashSet<String>,
        find_running_calls: usize,
    }

    /// A fake backend for the reuse flow's own tests: unlike [`FakeBackend`] (which
    /// has no notion of "currently running by name" at all), this tracks a
    /// `running` name set directly, so [`SandboxBackend::find_running`] and
    /// [`SandboxBackend::remove_by_name`] behave like a real backend's would —
    /// exactly what the adopt/stale/collision scenarios below need to drive.
    struct ReuseFakeBackend {
        state: StdMutex<ReuseFakeState>,
        conflict_once_for_name: StdMutex<Option<String>>,
        on_conflict: StdMutex<Option<Box<dyn FnMut() + Send>>>,
    }

    impl ReuseFakeBackend {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                state: StdMutex::new(ReuseFakeState::default()),
                conflict_once_for_name: StdMutex::new(None),
                on_conflict: StdMutex::new(None),
            })
        }

        /// Marks `name` as already running, as if some other (or earlier, same-
        /// process) call had already created and started it.
        fn mark_running(&self, name: &str) {
            self.state.lock().unwrap().running.insert(name.to_string());
        }

        /// The NEXT `create()` call for a spec named `name` fails with a typed
        /// [`RightsizeError::NameConflict`] instead of succeeding (armed once,
        /// consumed on that call), running `on_conflict` right before returning the
        /// error — the test's chance to simulate the concurrent winner's own
        /// side effects (marking itself running, writing the registry) landing
        /// right as this call loses the race.
        fn fail_next_create_with_conflict(
            &self,
            name: &str,
            on_conflict: impl FnMut() + Send + 'static,
        ) {
            *self.conflict_once_for_name.lock().unwrap() = Some(name.to_string());
            *self.on_conflict.lock().unwrap() = Some(Box::new(on_conflict));
        }
    }

    #[async_trait::async_trait]
    impl SandboxBackend for ReuseFakeBackend {
        fn name(&self) -> &str {
            "reuse-fake"
        }
        fn supports_native_networks(&self) -> bool {
            false
        }
        async fn create(&self, spec: ContainerSpec) -> Result<Box<dyn SandboxHandle>> {
            {
                let mut conflict = self.conflict_once_for_name.lock().unwrap();
                if conflict.as_deref() == Some(spec.name.as_str()) {
                    *conflict = None;
                    drop(conflict);
                    if let Some(cb) = self.on_conflict.lock().unwrap().as_mut() {
                        cb();
                    }
                    return Err(RightsizeError::NameConflict {
                        message: format!("sandbox '{}' already exists", spec.name),
                        source: None,
                    });
                }
            }
            self.state.lock().unwrap().created.push(spec.clone());
            Ok(Box::new(FakeHandle {
                id: spec.name.clone(),
                spec,
            }))
        }
        async fn start(&self, handle: &dyn SandboxHandle) -> Result<()> {
            let id = handle.id().to_string();
            let mut state = self.state.lock().unwrap();
            state.started.push(id.clone());
            state.running.insert(id);
            Ok(())
        }
        async fn stop(&self, handle: &dyn SandboxHandle) -> Result<()> {
            let id = handle.id().to_string();
            let mut state = self.state.lock().unwrap();
            state.stopped.push(id.clone());
            state.running.remove(&id);
            Ok(())
        }
        async fn remove(&self, handle: &dyn SandboxHandle) -> Result<()> {
            self.state
                .lock()
                .unwrap()
                .removed
                .push(handle.id().to_string());
            Ok(())
        }
        async fn exec(&self, _handle: &dyn SandboxHandle, cmd: &[String]) -> Result<ExecResult> {
            Ok(ExecResult {
                exit_code: 0,
                stdout: cmd.join(" "),
                stderr: String::new(),
            })
        }
        async fn logs(&self, _handle: &dyn SandboxHandle) -> Result<String> {
            Ok(String::new())
        }
        async fn follow_logs(
            &self,
            _handle: &dyn SandboxHandle,
            _consumer: Box<dyn Fn(String) + Send + Sync>,
        ) -> Result<crate::backend::FollowHandle> {
            unimplemented!("not exercised by the reuse test suite")
        }
        async fn ensure_network(&self, _network_id: &str) -> Result<()> {
            Ok(())
        }
        async fn remove_network(&self, _network_id: &str) -> Result<()> {
            Ok(())
        }
        fn cleanup_sync(&self, _container_id: &str) {}
        fn remove_by_name(&self, name: &str) {
            let mut state = self.state.lock().unwrap();
            state.removed_by_name.push(name.to_string());
            state.running.remove(name);
        }
        fn watchdog_kill_command(&self) -> Vec<String> {
            vec!["true".to_string()]
        }
        async fn find_running(
            &self,
            spec: &ContainerSpec,
        ) -> Result<Option<Box<dyn SandboxHandle>>> {
            let mut state = self.state.lock().unwrap();
            state.find_running_calls += 1;
            if state.running.contains(&spec.name) {
                Ok(Some(Box::new(FakeHandle {
                    id: spec.name.clone(),
                    spec: spec.clone(),
                })))
            } else {
                Ok(None)
            }
        }
    }

    fn sample_registry_entry(
        identity: &crate::reuse::Identity,
        host_port: u16,
    ) -> crate::reuse::RegistryEntry {
        crate::reuse::RegistryEntry {
            name: identity.name.clone(),
            image: "redis:7-alpine".to_string(),
            ports: std::collections::BTreeMap::from([("6379".to_string(), host_port)]),
            created_iso: "2025-01-01T00:00:00Z".to_string(),
            backend: "reuse-fake".to_string(),
        }
    }

    // Double opt-in: only the marker-AND-env-both-on combination produces a reuse
    // (`rz-reuse-<hash>`-named, `keep_alive`) sandbox; every other combination
    // behaves exactly like an ordinary ephemeral container.
    #[tokio::test]
    async fn reuse_double_opt_in_gating_all_four_combinations() {
        async fn is_reuse_name(marker: bool, env_enabled: bool) -> bool {
            let backend = ReuseFakeBackend::new();
            let cache_dir = temp_cache_dir("gating");
            let guard = Container::new("redis:7-alpine")
                .with_backend(backend)
                .with_cache_dir_override(cache_dir)
                .with_reuse_env_override(env_enabled)
                .with_exposed_ports(&[6379])
                .waiting_for(ReadyImmediately)
                .reuse(marker)
                .start()
                .await
                .expect("start must succeed regardless of the gating outcome");
            let reused = guard.name().starts_with("rz-reuse-");
            guard.stop().await.unwrap();
            reused
        }

        assert!(
            is_reuse_name(true, true).await,
            "marker on + env on must produce a reuse name"
        );
        assert!(
            !is_reuse_name(true, false).await,
            "marker on, env off: ordinary container"
        );
        assert!(
            !is_reuse_name(false, true).await,
            "env on, marker off: reuse never considered"
        );
        assert!(
            !is_reuse_name(false, false).await,
            "both off: ordinary container"
        );
    }

    // Adopt path: a registry hit whose sandbox the backend reports running, and
    // whose re-run wait strategy succeeds, adopts — no create() call, and the
    // guard's mapped port comes straight from the registry, not a fresh allocation.
    #[tokio::test]
    async fn adopt_path_registry_hit_running_and_wait_ok_skips_create_and_uses_registry_ports() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("adopt-hit");
        let identity =
            crate::reuse::compute_identity("redis:7-alpine", &[], &None, &[6379], None, &[])
                .unwrap();

        // A previous process already created, started, and registered this
        // sandbox, then exited cleanly (reuse containers are never torn down by
        // clean exit) — this process's first start() should adopt it.
        backend.mark_running(&identity.name);
        crate::reuse::Registry::new(&cache_dir, &identity.hash_hex)
            .write_atomic(&sample_registry_entry(&identity, 40321))
            .unwrap();

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir)
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("adopt must succeed");

        assert_eq!(guard.name(), identity.name);
        assert_eq!(
            guard.get_mapped_port(6379).unwrap(),
            40321,
            "must use the REGISTRY's port, not a freshly allocated one"
        );
        {
            let state = backend.state.lock().unwrap();
            assert!(
                state.created.is_empty(),
                "adopt must not call backend.create"
            );
            assert!(state.find_running_calls >= 1);
        }
        guard.stop().await.unwrap();
    }

    // Stale registry: the backend reports the recorded sandbox is NOT running ->
    // best-effort remove-by-name + delete the registry file, then fall through to a
    // fresh create that rewrites the registry.
    #[tokio::test]
    async fn stale_registry_not_running_removes_and_creates_fresh_and_rewrites_registry() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("adopt-stale");
        let identity =
            crate::reuse::compute_identity("redis:7-alpine", &[], &None, &[6379], None, &[])
                .unwrap();

        crate::reuse::Registry::new(&cache_dir, &identity.hash_hex)
            .write_atomic(&sample_registry_entry(&identity, 40321))
            .unwrap();
        // Deliberately NOT marked running: find_running must report `None`.

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir.clone())
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("a stale registry must fall back to a fresh create");

        assert_eq!(guard.name(), identity.name);
        {
            let state = backend.state.lock().unwrap();
            assert_eq!(state.created.len(), 1, "must create exactly once");
            assert_eq!(
                state.removed_by_name,
                vec![identity.name.clone()],
                "the stale sandbox must be best-effort removed by name"
            );
        }

        let rewritten = crate::reuse::Registry::new(&cache_dir, &identity.hash_hex)
            .read()
            .expect("the registry must be rewritten after the fresh create");
        let new_port = *rewritten.ports.get("6379").unwrap();
        assert_eq!(guard.get_mapped_port(6379).unwrap(), new_port);

        guard.stop().await.unwrap();
    }

    // Corrupted registry JSON: unparseable, but still on disk — best-effort remove
    // the identity-derived name (the only name we can know without parsing the
    // file) and the file itself, then fall through to a fresh create.
    #[tokio::test]
    async fn corrupted_registry_json_falls_back_to_fresh_create() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("adopt-corrupt");
        let identity =
            crate::reuse::compute_identity("redis:7-alpine", &[], &None, &[6379], None, &[])
                .unwrap();

        let reuse_dir = cache_dir.join("reuse");
        std::fs::create_dir_all(&reuse_dir).unwrap();
        std::fs::write(
            reuse_dir.join(format!("{}.json", identity.hash_hex)),
            b"not json",
        )
        .unwrap();

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir.clone())
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("corrupt registry JSON must fall back to a fresh create, not fail start()");

        {
            let state = backend.state.lock().unwrap();
            assert_eq!(state.created.len(), 1);
            assert_eq!(state.removed_by_name, vec![identity.name.clone()]);
        }
        assert!(
            crate::reuse::Registry::new(&cache_dir, &identity.hash_hex)
                .read()
                .is_some(),
            "a valid registry must exist after the fresh create"
        );

        guard.stop().await.unwrap();
    }

    // Stop semantics: stop() on a reuse-active guard clears only in-process
    // bookkeeping — no backend.stop/remove call, and the sandbox never appears in
    // the reaping ledger (never listed there in the first place).
    #[tokio::test]
    async fn stop_on_a_reuse_container_leaves_the_sandbox_running_and_never_touches_the_ledger() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("stop-semantics");

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir.clone())
            .with_reuse_env_override(true)
            // Test isolation seam: see `restored_container_is_registered_in_the_
            // reaping_ledger_like_any_other`'s own comment. Shares the same scratch
            // dir as the reuse registry override above — the reuse registry lives
            // under `<dir>/reuse/`, the reaper ledger under `<dir>/runs/`, so the
            // two don't collide.
            .with_reaper_cache_dir_override(cache_dir.clone())
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .unwrap();
        let name = guard.name().to_string();
        assert!(name.starts_with("rz-reuse-"));

        guard.stop().await.unwrap();

        {
            let state = backend.state.lock().unwrap();
            assert!(
                state.stopped.is_empty(),
                "stop() must not call backend.stop for a reuse container"
            );
            assert!(
                state.removed.is_empty(),
                "stop() must not call backend.remove for a reuse container"
            );
        }

        let ledger = crate::reaper::Ledger::new(&cache_dir, crate::RunId::value());
        assert!(
            !ledger.sandbox_names().contains(&name),
            "a reuse container must never appear in the reaping ledger, before or after stop()"
        );
    }

    // Reuse + a custom network is a typed, fail-fast error — never reaches create().
    #[tokio::test]
    async fn reuse_plus_custom_network_is_a_typed_error() {
        let backend = ReuseFakeBackend::new();
        let net = Arc::new(Network::new_network());
        let start_result = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_reuse_env_override(true)
            .with_network(&net)
            .reuse(true)
            .start()
            .await;
        let err = expect_start_err(start_result, "reuse + a custom network must fail fast");
        assert!(
            matches!(err, RightsizeError::ReuseNetworkConflict { .. }),
            "{err}"
        );
        assert!(
            backend.state.lock().unwrap().created.is_empty(),
            "must fail before any create() call"
        );
    }

    // Name collision on create (another process won the race): exactly one retry
    // back into the adopt path, using whatever the winner has by then written.
    #[tokio::test]
    async fn name_collision_on_create_retries_the_adopt_path_once() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("collision");
        let identity =
            crate::reuse::compute_identity("redis:7-alpine", &[], &None, &[6379], None, &[])
                .unwrap();

        // Deliberately NOT marked running yet: this call's own crash-mid-boot
        // orphan check (find_running, right before create) must see nothing and
        // must not remove anything — the concurrent winner only actually starts
        // (and registers) its sandbox at the exact moment THIS call's create()
        // loses the race, simulated below inside `on_conflict`, not before it.
        let entry = sample_registry_entry(&identity, 41111);
        let winner_cache_dir = cache_dir.clone();
        let winner_hash = identity.hash_hex.clone();
        let winner_backend = backend.clone();
        let winner_name = identity.name.clone();
        backend.fail_next_create_with_conflict(&identity.name, move || {
            // Simulate the concurrent winner's own create+start landing (marking
            // itself running) and its registry write, right as this call loses
            // the create race.
            winner_backend.mark_running(&winner_name);
            let _ =
                crate::reuse::Registry::new(&winner_cache_dir, &winner_hash).write_atomic(&entry);
        });

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir)
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("a name collision must retry the adopt path and succeed");

        assert_eq!(guard.name(), identity.name);
        assert_eq!(guard.get_mapped_port(6379).unwrap(), 41111);
        assert_eq!(
            backend.state.lock().unwrap().created.len(),
            0,
            "the losing create() attempt must not count as a successful create"
        );

        guard.stop().await.unwrap();
    }

    // ======================================================================
    // Crash-mid-boot orphan recovery (fresh-create's own find_running/remove
    // check, run once the adopt path has already concluded there is no usable
    // registry entry) — see docs/reuse.md's own section on this.
    // ======================================================================

    // (a) A sandbox is running under the identity's fixed name, but NO registry
    // entry points at it at all — exactly what a process that crashed (or failed
    // its own wait strategy) between `create` and the registry write leaves
    // behind. The next start() for the same identity must find it via
    // find_running and best-effort remove it BEFORE attempting a fresh create.
    #[tokio::test]
    async fn fresh_create_removes_a_running_but_unregistered_orphan_before_creating() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("orphan-recovery");
        let identity =
            crate::reuse::compute_identity("redis:7-alpine", &[], &None, &[6379], None, &[])
                .unwrap();

        // No registry file at all — but a sandbox under the identity's name is
        // already running, exactly as a crash-mid-boot orphan would leave it.
        backend.mark_running(&identity.name);

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir)
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("an orphaned running sandbox must not fail start(), just be replaced");

        assert_eq!(guard.name(), identity.name);
        {
            let state = backend.state.lock().unwrap();
            assert_eq!(
                state.removed_by_name,
                vec![identity.name.clone()],
                "the orphaned sandbox must be best-effort removed before the fresh create"
            );
            assert_eq!(state.created.len(), 1, "must still create exactly once");
            assert!(state.find_running_calls >= 1);
        }

        guard.stop().await.unwrap();
    }

    // (b) A registry entry IS present and verifies (adopt succeeds) — the
    // orphan-recovery find_running/remove step must never even run: adopt
    // short-circuits before `start_reuse` ever reaches it, so no remove_by_name
    // call happens.
    #[tokio::test]
    async fn adopt_with_a_verified_registry_never_calls_remove_by_name() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("orphan-recovery-adopt");
        let identity =
            crate::reuse::compute_identity("redis:7-alpine", &[], &None, &[6379], None, &[])
                .unwrap();

        backend.mark_running(&identity.name);
        crate::reuse::Registry::new(&cache_dir, &identity.hash_hex)
            .write_atomic(&sample_registry_entry(&identity, 40321))
            .unwrap();

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir)
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("a verified registry entry must adopt");

        assert_eq!(guard.name(), identity.name);
        assert!(
            backend.state.lock().unwrap().removed_by_name.is_empty(),
            "adopting a verified registry entry must never call remove_by_name"
        );

        guard.stop().await.unwrap();
    }

    // (c) No registry AND nothing running under the identity's name —
    // find_running reports None, so remove_by_name must never be called; the
    // fresh create proceeds exactly as it always has for a genuinely first-time
    // identity.
    #[tokio::test]
    async fn fresh_create_with_nothing_running_never_calls_remove_by_name() {
        let backend = ReuseFakeBackend::new();
        let cache_dir = temp_cache_dir("orphan-recovery-clean");

        let guard = Container::new("redis:7-alpine")
            .with_backend(backend.clone())
            .with_cache_dir_override(cache_dir)
            .with_reuse_env_override(true)
            .with_exposed_ports(&[6379])
            .waiting_for(ReadyImmediately)
            .reuse(true)
            .start()
            .await
            .expect("a genuinely fresh identity must create normally");

        {
            let state = backend.state.lock().unwrap();
            assert!(
                state.removed_by_name.is_empty(),
                "nothing was running, so remove_by_name must never be called"
            );
            assert!(
                state.find_running_calls >= 1,
                "the orphan-recovery check must still run"
            );
            assert_eq!(state.created.len(), 1);
        }

        guard.stop().await.unwrap();
    }
}