hf2q 0.1.1

//! Multi-model in-memory pool for ADR-005 Phase 4 hot-swap (iter-206 W74).
//!
//! This module is the **pure data primitive** that Phase 4's hot-swap
//! orchestrator (iter-208) and AppState integration (iter-209) compose
//! against.  It contains *no* engine load, *no* GPU code, *no* `AppState`
//! wiring — those live downstream.  The pattern mirrors iter-201
//! `serve::quant_select`: a synthetic-fixture-tested data structure that
//! later iters glue into the live serve path.
//!
//! # What this module does
//!
//! - Tracks N currently-loaded model handles (handle metadata only).
//! - Enforces two simultaneous bounds:
//!   1. **Capacity:** at most `capacity_models` distinct repos resident.
//!   2. **Memory budget:** sum of `bytes_resident` across the pool must
//!      stay `<= memory_budget_bytes`.
//! - LRU eviction order: when an `insert` would breach either bound, the
//!   pool drops the least-recently-used handle(s) until both bounds hold.
//! - `touch(repo)` promotes an existing entry to MRU without changing
//!   resident bytes — used by handler entry points before they read
//!   from the engine.
//! - `from_hardware(&HardwareProfile, n)` adapter applies the spec'd
//!   80% of total unified memory ceiling (line 929 of ADR-005).
//!
//! # What this module does NOT do
//!
//! - Hold the actual `Engine` (or any `mlx_native` buffers) — `LoadedHandle`
//!   is a pure descriptor.  iter-208's `HotSwapManager` will own
//!   `Arc<Engine>` and use this pool for the eviction-policy decision
//!   only.
//! - Reach the disk or GGUF reader — entry construction is by-value;
//!   the orchestrator builds handles after the engine load succeeds.
//! - Touch `AppState`, `cmd_serve`, or any handler — wiring is iter-209.
//! - Mutate the on-disk cache manifest — `cache::touch(repo)` is the
//!   parallel on-disk LRU stamp; the in-memory pool's LRU is
//!   independent (different lifetime, different policy).
//!
//! # Eviction algorithm
//!
//! `insert(handle)` runs in two passes:
//!
//! 1. **Capacity pass:** if `len() == capacity_models` and `repo` is not
//!    already in the pool, evict the LRU entry once.
//! 2. **Budget pass:** while `total_resident_bytes + handle.bytes >
//!    memory_budget_bytes`, evict the LRU entry.  Stop when either
//!    the bound is satisfied or the pool is empty.  If the pool is
//!    empty and `handle.bytes > memory_budget_bytes`, the insert is
//!    refused with `PoolError::OversizedHandle`.
//!
//! Re-inserting an already-present `repo_id` is treated as an in-place
//! update: the existing handle's `bytes_resident` and `loaded_at`
//! refresh and the entry is promoted to MRU; no eviction runs unless
//! the new bytes push the total over budget (in which case OTHER
//! entries are evicted, not the re-inserted one — that would be a
//! self-eviction defect; the re-insert intent is "this model just
//! re-loaded with different bytes / weights").
//!
//! # Why a hand-rolled LRU and not a crate
//!
//! `Cargo.toml` is fenced this session (ADR-014 P7 actively bumping
//! deps), so adding `indexmap` or `lru` is not allowed.  The stdlib
//! `HashMap<String, LoadedHandle>` plus `Vec<String>` order list
//! delivers identical semantics for `N <= 8` (Phase 4 default N=3,
//! configurable) at O(N) per op — fine for a pool of ~3 entries
//! where every op is a request-rate event.
//!
//! # Tests
//!
//! Synthetic-fixture unit tests cover:
//!
//! - empty pool + capacity-1 overflow
//! - capacity-3 LRU eviction order
//! - memory-budget eviction without capacity overflow
//! - combined eviction (capacity + budget in the same `insert`)
//! - `touch` promotes MRU; `get` is read-only (does NOT touch)
//! - `remove` returns the dropped handle + updates `total_resident_bytes`
//! - idempotent insert (same `repo_id`) updates bytes + promotes
//! - zero-budget refuses any non-zero handle
//! - oversized single handle refuses (handle.bytes > budget)
//! - `from_hardware` produces the 80% ceiling
//! - `len`, `is_empty`, `iter` semantics
//!
//! # ADR-005 Phase 4 lineage
//!
//! - Lines 928–936: narrative spec.
//! - AC 5357 (line 5357): "Cached pool holds up to 3 loaded models with
//!   LRU eviction bounded by 80% of system unified memory (configurable)".
//!   This module is the foundation primitive that AC closes against.
//! - Plan: line 5354 (Phase 4 audit + iter-by-iter plan, iter 206 row).

use std::collections::HashMap;
use std::path::PathBuf;
use std::time::SystemTime;

use crate::core::hardware::HardwareProfile;

/// Configuration consumed by [`crate::serve::load_engine`] (and by extension
/// the [`ModelLoader`] trait that the [`HotSwapManager`] dispatches against).
///
/// Mirrors the per-load fields previously inlined in `cmd_serve`'s
/// model-load block (`src/serve/mod.rs`, pre-iter-208 lines 1107–1167):
/// the optional sidecar paths the operator supplied via `--tokenizer` /
/// `--config`, the FIFO queue capacity used by `Engine::spawn`, and a
/// `warmup_synchronously` knob that preserves the iter-103 ordering
/// (chat-warmup BEFORE any other Metal device activity).
///
/// Held by-value because every field is small and the loader closure runs
/// once per load (no hot-path concern with cloning).  The `PathBuf` fields
/// are `Option<_>` because `--tokenizer` / `--config` default to a
/// next-to-the-GGUF lookup performed inside `LoadedModel::load`.
#[derive(Clone, Default)]
pub struct EngineConfig {
    /// Optional explicit `tokenizer.json` path.  `None` ⇒ auto-resolve
    /// via `find_tokenizer` in `engine.rs` (sidecar lookup).
    pub tokenizer_path: Option<PathBuf>,
    /// Optional explicit `config.json` path.  `None` ⇒ auto-resolve via
    /// `find_config`.
    pub config_path: Option<PathBuf>,
    /// FIFO queue capacity passed to `Engine::spawn`.  Bounded backpressure
    /// surface (Decision #19): when full, handlers see `queue_full` and
    /// map to 429 + Retry-After.
    pub queue_capacity: usize,
    /// When `true`, run `Engine::warmup()` on a temporary tokio runtime
    /// before returning.  The hot-swap orchestrator and the existing
    /// `cmd_serve` startup both pass `true` so the returned engine is
    /// fully primed; tests using a `MockLoader` can pass `false`.
    pub warmup_synchronously: bool,
    /// ADR-017 Phase E.a iter-2 — optional handle to the AppState-owned
    /// `KvSpillCounters` (upcast to `Arc<dyn KvCacheMetricsSink>`) so
    /// the engine worker thread's per-request LCP probe bumps the same
    /// Arc the `/metrics` handler reads.  Set by `cmd_serve` from
    /// `state.kv_spill_counters`. `None` for tests / standalone engine
    /// constructions; the LCP probe becomes a no-op there.
    pub kv_metrics_sink:
        Option<std::sync::Arc<dyn crate::serve::kv_persist::metrics::KvCacheMetricsSink>>,
    /// ADR-020 AC#5 Iter D — optional path to a DWQ-trained mlx-affine
    /// safetensors overlay.  When `Some`, applied after GGUF load via
    /// `MlxModelWeights::apply_dwq_overlay` (dense families only;
    /// qwen35moe MoE-expert tensors skipped pending Iter C2).  Set by
    /// `cmd_serve` from `args.dwq_overlay`.
    pub dwq_overlay_path: Option<PathBuf>,

    /// ADR-040 Phase C iter-4 (C4) — scheduler-policy selection for the
    /// engine spawned from this config.
    ///
    /// `EngineMode::SerialFifo` (the [`Default`] impl) preserves the
    /// ADR-005 Decision #2 + #19 contract byte-for-byte per ADR-040
    /// §3.6. When `cmd_serve` parses `--scheduler inflight_batched` (or
    /// `HF2Q_SCHEDULER=inflight_batched`), this becomes
    /// `EngineMode::SlotAware { max_slots: N }` and `load_engine` calls
    /// [`Engine::spawn_with_mode`] instead of [`Engine::spawn`]. The
    /// downstream `Engine::spawn_with_mode` returns
    /// [`EngineSpawnError::ModeNotYetWired`] until iter-2b/2c land the
    /// per-family slot-aware worker arms — `load_engine` surfaces that
    /// as an `anyhow::Error` and `cmd_serve` aborts startup with a
    /// non-zero exit code (fail-loud per ADR-040 §7 mantra).
    pub engine_mode: crate::serve::api::engine::EngineMode,
}

impl std::fmt::Debug for EngineConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EngineConfig")
            .field("tokenizer_path", &self.tokenizer_path)
            .field("config_path", &self.config_path)
            .field("queue_capacity", &self.queue_capacity)
            .field("warmup_synchronously", &self.warmup_synchronously)
            .field("kv_metrics_sink_present", &self.kv_metrics_sink.is_some())
            .field("dwq_overlay_path", &self.dwq_overlay_path)
            // ADR-040 Phase C iter-4 (C4) — surface the parsed
            // scheduler-policy selection so operator log greps + test
            // assertions see what mode the engine spawn will request.
            .field("engine_mode", &self.engine_mode)
            .finish()
    }
}

/// Default pool capacity per ADR-005 Phase 4 narrative (line 929).
/// Configurable via [`LoadedPool::with_capacity_and_budget`].
pub const DEFAULT_POOL_CAPACITY: usize = 3;

/// Default memory budget fraction of total unified memory per
/// ADR-005 Phase 4 narrative (line 929: "memory ceiling of **80% of
/// system unified memory**").  Used by [`LoadedPool::from_hardware`].
pub const DEFAULT_MEMORY_BUDGET_FRACTION: f64 = 0.80;

/// Descriptor for one currently-loaded model handle.
///
/// Pure metadata — does NOT hold the actual `Engine` or any GPU
/// buffers.  The hot-swap orchestrator (iter-208) maintains a
/// parallel `HashMap<String, Arc<Engine>>` keyed by `repo_id` and
/// uses this pool's eviction decisions to drop entries from that map.
#[derive(Debug, Clone)]
pub struct LoadedHandle {
    /// HuggingFace repo id (`org/repo`), the canonical pool key.
    pub repo_id: String,
    /// Quantization the loaded weights were materialized at.
    /// Carried for observability + `/v1/models` extension fields.
    pub quant: String,
    /// Wall-clock when the handle was loaded.  Used for diagnostics
    /// (e.g. `Server-Timing: model_load=NNNms` headers); does NOT
    /// drive eviction — eviction reads the pool's internal MRU
    /// order list, which is updated on every `insert` and `touch`.
    pub loaded_at: SystemTime,
    /// On-GPU resident-bytes attributed to this handle.  Sum of all
    /// model weights + KV cache region + any pinned scratch.  The
    /// pool sums these to enforce `memory_budget_bytes`.
    pub bytes_resident: u64,
}

impl LoadedHandle {
    /// Construct a handle with `loaded_at = SystemTime::now()`.
    /// Tests use this; the orchestrator sets `loaded_at` from the
    /// engine-load completion timestamp.
    pub fn new(repo_id: impl Into<String>, quant: impl Into<String>, bytes_resident: u64) -> Self {
        Self {
            repo_id: repo_id.into(),
            quant: quant.into(),
            loaded_at: SystemTime::now(),
            bytes_resident,
        }
    }
}

/// Errors the pool can return from `insert`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PoolError {
    /// `handle.bytes_resident > memory_budget_bytes`: the pool can
    /// never accommodate this handle even when empty.  Caller must
    /// reject the load attempt with a clear operator-facing message.
    OversizedHandle {
        repo_id: String,
        handle_bytes: u64,
        budget_bytes: u64,
    },
    /// `capacity_models == 0`: the pool was constructed disabled.
    /// Caller is expected to refuse the insert with a config-error
    /// message; this exists so a misconfigured deployment fails
    /// loudly rather than silently dropping every load.
    ZeroCapacity,
}

impl std::fmt::Display for PoolError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::OversizedHandle {
                repo_id,
                handle_bytes,
                budget_bytes,
            } => write!(
                f,
                "model {repo_id} resident bytes {handle_bytes} exceed pool memory budget \
                 {budget_bytes}; cannot load (consider raising the budget via the \
                 Phase 4 hot-swap config knob, or pick a smaller quant)"
            ),
            Self::ZeroCapacity => write!(
                f,
                "multi-model pool is configured with capacity_models = 0; refusing \
                 every load.  Either raise capacity (default 3) or remove the pool \
                 entirely from the deployment config."
            ),
        }
    }
}

impl std::error::Error for PoolError {}

/// Bounded in-memory LRU pool of [`LoadedHandle`] descriptors.
///
/// See module docs for the eviction algorithm and what this type does
/// (and explicitly does NOT) do.
#[derive(Debug)]
pub struct LoadedPool {
    /// Maximum number of distinct repos resident.  See module docs.
    capacity_models: usize,
    /// Maximum sum of `bytes_resident` across the pool.  See module docs.
    memory_budget_bytes: u64,
    /// Repo → handle lookup.  O(1) reads.
    entries: HashMap<String, LoadedHandle>,
    /// LRU ordering.  Index 0 is the LRU; the last index is the MRU.
    /// `lru_order.len() == entries.len()` is an invariant.
    lru_order: Vec<String>,
    /// Cumulative `bytes_resident` across `entries`.  Maintained
    /// incrementally so capacity checks are O(1).
    total_resident_bytes: u64,
}

impl LoadedPool {
    /// Construct a pool with explicit capacity and memory budget.
    /// `capacity_models == 0` produces a pool that refuses every
    /// insert with [`PoolError::ZeroCapacity`] — the constructor
    /// allows it so misconfigured deployments fail at insert-time
    /// rather than at construct-time (where the error has less
    /// context).
    pub fn with_capacity_and_budget(capacity_models: usize, memory_budget_bytes: u64) -> Self {
        Self {
            capacity_models,
            memory_budget_bytes,
            entries: HashMap::with_capacity(capacity_models.max(1)),
            lru_order: Vec::with_capacity(capacity_models.max(1)),
            total_resident_bytes: 0,
        }
    }

    /// Construct a pool with [`DEFAULT_POOL_CAPACITY`] and the
    /// 80%-of-total-unified-memory budget per ADR-005 line 929.
    ///
    /// Reads `HardwareProfile::total_memory_bytes` (the pool reserves
    /// against the *physical* unified memory, not the
    /// available-now-bytes — an in-flight model that hasn't released
    /// memory yet should still be allowed to fit, the pool's own
    /// accounting determines that).
    pub fn from_hardware(hw: &HardwareProfile) -> Self {
        // ADR-017 Closure iter-8 (2026-05-05): operator can override the
        // hardware-derived budget via `HF2Q_POOL_BUDGET_BYTES` (raw bytes
        // as a u64). Used by the stress smoke + production deployments
        // that want a tighter cap than the default 80%-of-RAM.
        //
        // Why this exists: on RAM-rich hardware (e.g., 128 GB M5 Max)
        // the default 80% budget = ~102 GB easily fits 4-5 16-GB models
        // simultaneously. The LRU-evict path never fires under the
        // stress-loop's swap-cycle workload, so KV-persist's
        // `pre_evict` + `drop_family` paths are never exercised. With
        // a 20 GB budget, every other model load forces eviction →
        // exercises the spill+restore lifecycle for real leak
        // detection. Iter-7 stress smoke surfaced this gap as a
        // false-positive RSS-leak reading (60 GB of accumulated
        // models, no real leak).
        //
        // Env unset: identical to pre-iter-8 behavior (80% of RAM at
        // DEFAULT_POOL_CAPACITY model slots). Env set but unparseable:
        // logged as a warning and falls through to the default —
        // never panics on a malformed env, so misconfigured operators
        // don't lose their server.
        if let Ok(s) = std::env::var("HF2Q_POOL_BUDGET_BYTES") {
            if let Ok(n) = s.parse::<u64>() {
                return Self::with_capacity_and_budget(DEFAULT_POOL_CAPACITY, n);
            } else {
                eprintln!(
                    "[ADR-017 iter-8] HF2Q_POOL_BUDGET_BYTES={s:?} did not \
                     parse as u64; falling through to hardware-derived budget"
                );
            }
        }
        Self::from_hardware_with(hw, DEFAULT_POOL_CAPACITY, DEFAULT_MEMORY_BUDGET_FRACTION)
    }

    /// Same as [`Self::from_hardware`] with explicit capacity and
    /// fraction overrides.  `fraction` outside `(0.0, 1.0]` is clamped
    /// to that range — passing 1.5 yields full physical memory; passing
    /// 0.0 yields a zero-budget pool that refuses everything (matches
    /// `from_hardware_with(_, 0, _)` semantically).
    pub fn from_hardware_with(hw: &HardwareProfile, capacity_models: usize, fraction: f64) -> Self {
        let f = fraction.clamp(0.0, 1.0);
        // f64 → u64 with floor; never panics for finite f.
        let budget = ((hw.total_memory_bytes as f64) * f).floor() as u64;
        Self::with_capacity_and_budget(capacity_models, budget)
    }

    /// Capacity (number of distinct loaded models permitted).
    pub fn capacity_models(&self) -> usize {
        self.capacity_models
    }

    /// Memory budget in bytes.
    pub fn memory_budget_bytes(&self) -> u64 {
        self.memory_budget_bytes
    }

    /// Cumulative resident bytes across loaded handles.
    pub fn total_resident_bytes(&self) -> u64 {
        self.total_resident_bytes
    }

    /// Number of currently-resident handles.
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// `true` if the pool has zero resident handles.
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Read-only borrow of a handle by `repo_id`.  Does NOT touch the
    /// LRU order — call [`Self::touch`] explicitly if the caller is
    /// servicing a request and wants to promote.  Separating get
    /// from touch lets diagnostic / metrics endpoints peek without
    /// poisoning the eviction policy.
    pub fn get(&self, repo_id: &str) -> Option<&LoadedHandle> {
        self.entries.get(repo_id)
    }

    /// Promote `repo_id` to MRU.  Returns `true` if the entry existed
    /// and was promoted; `false` if it wasn't in the pool (no-op).
    pub fn touch(&mut self, repo_id: &str) -> bool {
        if !self.entries.contains_key(repo_id) {
            return false;
        }
        // Move the entry to the back of `lru_order` (MRU end).
        if let Some(pos) = self.lru_order.iter().position(|r| r == repo_id) {
            let key = self.lru_order.remove(pos);
            self.lru_order.push(key);
        }
        true
    }

    /// Iterate handles in LRU → MRU order (pool internals' canonical
    /// order).  Useful for `/v1/models` listing and pool diagnostics.
    pub fn iter(&self) -> impl Iterator<Item = &LoadedHandle> {
        self.lru_order
            .iter()
            .filter_map(move |k| self.entries.get(k))
    }

    /// Drop a handle by `repo_id`.  Returns the dropped handle if
    /// present, `None` otherwise.  Updates `total_resident_bytes`.
    /// Does NOT free GPU buffers — that's the orchestrator's
    /// responsibility (drop the parallel `Arc<Engine>`).
    pub fn remove(&mut self, repo_id: &str) -> Option<LoadedHandle> {
        let handle = self.entries.remove(repo_id)?;
        if let Some(pos) = self.lru_order.iter().position(|r| r == repo_id) {
            self.lru_order.remove(pos);
        }
        self.total_resident_bytes = self
            .total_resident_bytes
            .saturating_sub(handle.bytes_resident);
        Some(handle)
    }

    /// Insert a handle into the pool, evicting LRU entries as needed.
    ///
    /// Returns the list of evicted handles (LRU-first order — the
    /// first element is the oldest evictee).  An empty `Vec` means
    /// no eviction was needed.
    ///
    /// Re-inserting an existing `repo_id` updates the handle's
    /// `bytes_resident` + `loaded_at` and promotes the entry to MRU.
    /// In that case the `total_resident_bytes` delta may still drive
    /// budget eviction of *other* entries — but the re-inserted
    /// handle itself is never evicted from the same `insert` call.
    ///
    /// # Errors
    ///
    /// - [`PoolError::ZeroCapacity`] if `capacity_models == 0`.
    /// - [`PoolError::OversizedHandle`] if `handle.bytes_resident >
    ///   memory_budget_bytes` and the pool would have to be empty
    ///   to fit it (i.e. impossible to satisfy under any LRU
    ///   eviction sequence).
    pub fn insert(&mut self, handle: LoadedHandle) -> Result<Vec<LoadedHandle>, PoolError> {
        if self.capacity_models == 0 {
            return Err(PoolError::ZeroCapacity);
        }
        if handle.bytes_resident > self.memory_budget_bytes {
            return Err(PoolError::OversizedHandle {
                repo_id: handle.repo_id,
                handle_bytes: handle.bytes_resident,
                budget_bytes: self.memory_budget_bytes,
            });
        }

        let mut evicted: Vec<LoadedHandle> = Vec::new();

        // Re-insert path: update bytes + promote to MRU.  Eviction of
        // OTHER entries may still happen in the budget pass below.
        if let Some(existing) = self.entries.get_mut(&handle.repo_id) {
            // Drop the old contribution first to avoid an underflow in
            // `total_resident_bytes` if the new bytes are smaller.
            self.total_resident_bytes = self
                .total_resident_bytes
                .saturating_sub(existing.bytes_resident);
            existing.bytes_resident = handle.bytes_resident;
            existing.loaded_at = handle.loaded_at;
            existing.quant = handle.quant.clone();
            self.total_resident_bytes = self
                .total_resident_bytes
                .saturating_add(handle.bytes_resident);
            // Promote: move to MRU end in `lru_order`.
            if let Some(pos) = self.lru_order.iter().position(|r| r == &handle.repo_id) {
                let key = self.lru_order.remove(pos);
                self.lru_order.push(key);
            }
            // Budget pass: evict OTHERS until the bound holds.  The
            // re-inserted entry is now at the MRU end (last index)
            // so the LRU-first eviction loop will not touch it
            // unless it is the only entry (in which case
            // `total_resident_bytes <= memory_budget_bytes` already
            // by the OversizedHandle precheck).
            self.evict_until_within_budget(&mut evicted, &handle.repo_id);
            return Ok(evicted);
        }

        // Capacity pass: if at capacity, evict the LRU entry exactly
        // once.  (We only evict ONE here; the budget pass below may
        // evict more.)
        if self.entries.len() >= self.capacity_models {
            if let Some(victim_key) = self.lru_order.first().cloned() {
                if let Some(victim) = self.entries.remove(&victim_key) {
                    self.total_resident_bytes = self
                        .total_resident_bytes
                        .saturating_sub(victim.bytes_resident);
                    self.lru_order.remove(0);
                    evicted.push(victim);
                }
            }
        }

        // Budget pass (pre-insert).  Evict LRU until adding the new
        // handle would fit.  We hold `handle` outside the pool here
        // so the loop only touches existing entries.
        while !self.lru_order.is_empty()
            && self
                .total_resident_bytes
                .saturating_add(handle.bytes_resident)
                > self.memory_budget_bytes
        {
            let victim_key = self.lru_order.remove(0);
            if let Some(victim) = self.entries.remove(&victim_key) {
                self.total_resident_bytes = self
                    .total_resident_bytes
                    .saturating_sub(victim.bytes_resident);
                evicted.push(victim);
            }
        }

        // The OversizedHandle precheck guarantees the post-eviction
        // budget holds — `handle.bytes_resident <= memory_budget_bytes`
        // and we evicted until `total_resident_bytes + handle.bytes
        // <= memory_budget_bytes` OR the pool is empty.  In the
        // pool-empty case, `0 + handle.bytes <= memory_budget_bytes`
        // holds by the precheck.

        // Insert + push to MRU end.
        self.total_resident_bytes = self
            .total_resident_bytes
            .saturating_add(handle.bytes_resident);
        self.lru_order.push(handle.repo_id.clone());
        self.entries.insert(handle.repo_id.clone(), handle);

        Ok(evicted)
    }

    /// Helper used by the re-insert path: evict LRU entries until
    /// `total_resident_bytes <= memory_budget_bytes`, skipping the
    /// just-promoted entry (`spare_repo_id`).
    fn evict_until_within_budget(&mut self, evicted: &mut Vec<LoadedHandle>, spare_repo_id: &str) {
        while self.total_resident_bytes > self.memory_budget_bytes {
            // Find the LRU entry that is NOT the spare.
            let victim_idx = self.lru_order.iter().position(|k| k != spare_repo_id);
            let Some(idx) = victim_idx else {
                // Only the spare remains; the precheck guaranteed
                // `spare.bytes <= budget`, so the loop terminates here.
                break;
            };
            let victim_key = self.lru_order.remove(idx);
            if let Some(victim) = self.entries.remove(&victim_key) {
                self.total_resident_bytes = self
                    .total_resident_bytes
                    .saturating_sub(victim.bytes_resident);
                evicted.push(victim);
            }
        }
    }
}

// ─────────────────────────────────────────────────────────────────────
// HotSwapManager — pool-backed engine cache with LRU eviction.
//
// ADR-005 Phase 4 spec item 3/5 (W76 iter-208).  Composes the pure
// [`LoadedPool`] data primitive (W74 iter-206) with a pluggable
// [`ModelLoader`] trait so tests can substitute a synthetic engine
// fixture and production wires the [`DefaultModelLoader`] that delegates
// to [`crate::serve::load_engine`].
//
// **Generic over the engine type.**  The production wire-up uses
// `E = crate::serve::api::engine::Engine` ([`HotSwapManager::default`] +
// [`DefaultModelLoader`]).  Unit tests substitute `E = ()`-equivalent
// synthetic fixture types so the manager's eviction + accounting +
// in-flight-Arc-safety logic can be exercised without a real Metal
// device or GGUF on disk.  The shape `Engine` ships with — worker
// thread, GPU buffers, tokenizer, etc. — has no synthetic constructor;
// the trait approach keeps the production type free of test-only
// scaffolding.
// ─────────────────────────────────────────────────────────────────────

use std::path::Path;
use std::sync::Arc;

use crate::serve::api::engine::Engine;
use crate::serve::quant_select::QuantType;

/// A loaded engine + the metadata the pool tracks for eviction.  The
/// `HotSwapManager` hands out `Arc<LoadedEngine<E>>` clones; in-flight
/// requests that hold an Arc keep the engine alive past eviction (the
/// pool slot drops the manager's Arc; the engine itself drops when the
/// last handler releases — refcount semantics).
///
/// Generic over the engine type so tests can substitute a synthetic `E`
/// (the production type [`Engine`] requires a real Metal device + GGUF).
#[derive(Debug)]
pub struct LoadedEngine<E> {
    /// The actual engine handle.  In production this is
    /// [`crate::serve::api::engine::Engine`] (owns the worker thread,
    /// model weights, KV caches).
    pub engine: E,
    /// HuggingFace repo id (or path stem) — same key the pool uses.
    pub repo: String,
    /// Quantization variant resident on this engine.
    pub quant: QuantType,
    /// On-GPU resident-bytes estimate (typically GGUF file size; the
    /// pool sums these to enforce the memory budget).  Set at admission
    /// time and never updated.
    pub bytes_resident: u64,
    /// Wall-clock when the engine finished loading.
    pub loaded_at: SystemTime,
}

/// Trait for loading a GGUF into a live engine of type `E`.
///
/// The production implementation ([`DefaultModelLoader`]) delegates to
/// [`crate::serve::load_engine`], which performs the full mlx-native load
/// (header parse + weights mmap → GPU + tokenizer + chat-template +
/// synchronous warmup).  Tests substitute a `MockLoader` that returns a
/// synthetic engine fixture without touching disk or Metal.
///
/// `Send + Sync` because the manager is held inside an
/// `Arc<RwLock<HotSwapManager>>` (iter-209) and concurrent handlers may
/// call `load_or_get` from multiple tokio tasks.
pub trait ModelLoader<E>: Send + Sync {
    /// Load a GGUF at `path` using `config` and return the live engine.
    /// May take seconds (full GPU weights upload + warmup).
    ///
    /// Errors propagate from header parse, weights load, tokenizer parse,
    /// chat-template resolution, or warmup; the manager treats any error
    /// as a load-failure and does NOT admit a partial entry.
    fn load(&self, path: &Path, config: &EngineConfig) -> anyhow::Result<E>;
}

/// Production [`ModelLoader`] — delegates to
/// [`crate::serve::load_engine`].  Stateless; cheap to clone.
#[derive(Debug, Clone, Default)]
pub struct DefaultModelLoader;

impl ModelLoader<Engine> for DefaultModelLoader {
    fn load(&self, path: &Path, config: &EngineConfig) -> anyhow::Result<Engine> {
        crate::serve::load_engine(path, config)
    }
}

// ─────────────────────────────────────────────────────────────────────
// KvSpiller — eviction-hook trait surface (ADR-005 Phase 4 reopen,
// iter-212 / AC 5471).
//
// Pluggable hook into the manager's eviction + admission lifecycle so
// ADR-017 (Persistent Block Prefix Cache for serve mode) can spill KV
// blocks to disk on swap-out and restore them on swap-back-in WITHOUT
// `HotSwapManager` taking a direct dependency on the persistence
// implementation.  Mirrors the [`ModelLoader`] injection pattern: a
// trait + a no-op default ([`NoopKvSpiller`]) + a production wire-up
// substitution that lives in ADR-017's tree.
//
// Phase 4 reopen ships:
//   - the trait + outcome / error enums + the no-op default
//   - two trigger sites in `load_or_get` (pre-evict per evictee +
//     post-admit on cold load)
//   - one trigger site in `evict()` (symmetric pre-evict)
//   - a `MockSpiller` test fixture (in the tests module) that mirrors
//     the existing `MockLoader` shape
//
// ADR-017 ships:
//   - the real `KvSpiller<Engine>` impl that talks to disk
//   - the `cmd_serve --kv-persist` CLI flag that constructs the real
//     spiller and feeds it through `HotSwapManager::new_with_spiller`
//   - the telemetry counter increments that consume `SpillOutcome` /
//     `RestoreOutcome` (zeroed surface lands iter-213 in this reopen)
//
// **Behavior with the noop spiller wired** is byte-identical to the
// pre-iter-212 manager: the trigger sites call `spiller.pre_evict(...)`
// / `spiller.post_admit(...)` and discard the outcome (`let _ =
// outcome;`).  The 39 existing `serve::multi_model::tests` cases pass
// UNCHANGED — proves the surface is non-disruptive.
// ─────────────────────────────────────────────────────────────────────

/// Failure mode for [`SpillOutcome::Error`].  Carried separately from
/// [`RestoreErrorKind`] because the spill path (snapshotting an in-memory
/// KV block to disk) and the restore path (rehydrating + parity-checking
/// a previously-spilled block) fail in disjoint ways even though both
/// share the same broad categories (codec / IO / parity).
///
/// `Copy` so trigger-site call-sites can store the outcome in a local
/// `let _ = ...` slot without an Arc wrapper; `Debug` so test assertions
/// can match on a specific kind via `matches!(...)`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SpillErrorKind {
    /// Codec-level failure: the in-memory KV block could not be encoded
    /// to the on-disk envelope (e.g. dtype mismatch, layout drift).
    CodecErr,
    /// Filesystem failure: write / fsync / truncate refused (disk full,
    /// permissions, EIO).
    IoErr,
    /// Parity check failure on a round-trip verify.
    ParityFail,
}

/// Failure mode for [`RestoreOutcome::Error`].  Symmetric to
/// [`SpillErrorKind`] for the restore path.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestoreErrorKind {
    /// Codec-level failure: the on-disk envelope could not be decoded
    /// back to the in-memory KV layout (envelope version drift, header
    /// mismatch, truncated body).
    CodecErr,
    /// Filesystem failure: read / open refused (file vanished, EIO).
    IoErr,
    /// Parity check failure: the restored KV bytes did not match the
    /// recorded SHA / checksum.  Treated as a hard restore-fail; the
    /// engine continues without the spilled cache (re-prefill on demand).
    ParityFail,
}

/// Result of [`KvSpiller::pre_evict`].  The trigger site discards the
/// outcome with `let _ = ...` in iter-212 (this iter); iter-213 wires
/// each variant to a `/metrics` counter increment with a fixed
/// `outcome` label cardinality (`success` / `codec_err` / `io_err` /
/// `parity_fail` / `skipped`).
///
/// `Skipped` is the no-op default's outcome and is NOT a failure — it
/// signals "no work was attempted" (the noop spiller has nothing to
/// spill; ADR-017 may also short-circuit on heuristics like "block
/// list empty" or "in-flight write quota exhausted").
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SpillOutcome {
    /// No spill work was attempted.  The noop spiller returns this.
    Skipped,
    /// `N` KV blocks were enqueued for spill.  Block count is operator-
    /// observable; the per-block byte count rides on a separate gauge
    /// in ADR-017.
    EnqueuedBlocks(u32),
    /// Spill attempt failed.  Eviction proceeds regardless — the spill
    /// is a best-effort hint, not a precondition.
    Error(SpillErrorKind),
}

/// Result of [`KvSpiller::post_admit`].  Symmetric to [`SpillOutcome`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestoreOutcome {
    /// No restore work was attempted.
    Skipped,
    /// `N` KV blocks were restored from the on-disk envelope.
    RestoredBlocks(u32),
    /// Restore attempt failed.  Admission proceeds regardless — the
    /// restore is a best-effort hint, not a precondition.  The engine
    /// will re-prefill on demand.
    Error(RestoreErrorKind),
}

/// Eviction-hook trait surface for [`HotSwapManager`].
///
/// Two trigger sites:
///
/// - [`Self::pre_evict`] fires from `HotSwapManager::load_or_get` (per
///   evicted handle, before the manager's `Arc<LoadedEngine<E>>` drops
///   from the engines map) AND from `HotSwapManager::evict` (the
///   explicit-evict path).  The Arc is still live at call time so an
///   impl can snapshot live KV state without a race.
/// - [`Self::post_admit`] fires from `HotSwapManager::load_or_get`
///   between `loader.load()` returning an engine and the manager
///   publishing the Arc into the engines map.  The Arc is freshly
///   constructed and held by the manager only at call time, so an
///   impl can rehydrate KV state into the engine before any handler
///   sees it.
///
/// **Best-effort.**  The manager discards the returned outcome at the
/// trigger site in iter-212 — eviction + admission proceed regardless
/// of `Skipped` / `EnqueuedBlocks` / `Error`.  iter-213 wires the
/// outcomes to `/metrics` counter increments.  An impl returning
/// `Error(...)` does NOT block the manager's lifecycle.
///
/// **Send + Sync** because the manager is held inside an
/// `Arc<RwLock<HotSwapManager<E>>>` (see `AppState::pool` at
/// `src/serve/api/state.rs:202`); concurrent handlers may call
/// `load_or_get` from multiple tokio tasks.
pub trait KvSpiller<E>: Send + Sync {
    /// Called BEFORE the manager drops its Arc<LoadedEngine<E>> from the
    /// engines map.  The Arc is live; the impl may inspect engine state
    /// (e.g. via interior mutability or worker-thread message-pass) to
    /// snapshot any KV blocks worth spilling.
    fn pre_evict(&self, handle: &LoadedHandle, engine: &Arc<LoadedEngine<E>>) -> SpillOutcome;

    /// Called AFTER `loader.load()` succeeds and BEFORE the manager
    /// publishes the Arc into its engines map.  The impl may rehydrate
    /// KV blocks into the freshly-loaded engine.
    fn post_admit(
        &self,
        repo: &str,
        quant: QuantType,
        engine: &Arc<LoadedEngine<E>>,
    ) -> RestoreOutcome;

    /// ADR-017 Closure iter-7 (2026-05-04) — called AFTER the manager
    /// has dropped its `Arc<LoadedEngine<E>>` from the engines map
    /// (post-evict). The impl may release any per-family resources
    /// it accumulated since registration (e.g. cloned `Engine` refs
    /// that were keeping the worker thread alive — the iter-3
    /// Phase-E engine_arc Engine clone is exactly this kind of
    /// reference, and without explicit drop it leaks across
    /// model-swap cycles).
    ///
    /// Default = no-op. `BlockPrefixCacheSpiller` overrides to call
    /// its inherent `unregister_family`, which removes the family
    /// hook from the registrations map; the inner spill's
    /// `Arc<EngineInner>` strong-count drops to zero on the
    /// concrete-spill drop, the worker thread exits via channel
    /// close, and the model's RSS frees deterministically.
    fn drop_family(&self, _repo: &str, _quant: QuantType) {}
}

/// No-op default [`KvSpiller`] — every method returns `Skipped` and
/// performs no work.  Wired by [`HotSwapManager::new`] (the back-compat
/// 2-arg constructor) so behavior is byte-identical to pre-iter-212
/// across every existing manager test.
///
/// ADR-017 substitutes this with a real impl via
/// [`HotSwapManager::new_with_spiller`] when the operator passes
/// `cmd_serve --kv-persist`.
///
/// Stateless; cheap to clone.  The manager holds it behind `Arc<dyn
/// KvSpiller<E>>` so impls that need state (file handles, in-flight
/// queues) compose without changing the wiring shape.
#[derive(Debug, Clone, Copy, Default)]
pub struct NoopKvSpiller;

impl<E> KvSpiller<E> for NoopKvSpiller {
    fn pre_evict(&self, _handle: &LoadedHandle, _engine: &Arc<LoadedEngine<E>>) -> SpillOutcome {
        SpillOutcome::Skipped
    }
    fn post_admit(
        &self,
        _repo: &str,
        _quant: QuantType,
        _engine: &Arc<LoadedEngine<E>>,
    ) -> RestoreOutcome {
        RestoreOutcome::Skipped
    }
}

/// Pool diagnostics — bytes used, count, capacity — surfaced for
/// `/v1/models` extension fields and `/metrics` Prometheus output.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PoolStats {
    pub loaded_count: usize,
    pub capacity_models: usize,
    pub total_resident_bytes: u64,
    pub memory_budget_bytes: u64,
}

/// Per-loaded-handle summary for `/v1/models` extension fields.
/// `pool_key` is the manager-internal `format!("{repo}@{quant}")` form;
/// `quant` is the canonical GGML name string (matches
/// [`QuantType::as_str`] output).  Bytes-resident is the on-GPU
/// allocation accounted against the pool's memory budget.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LoadedSummary {
    pub pool_key: String,
    pub quant: String,
    pub bytes_resident: u64,
}

/// Errors returned by [`HotSwapManager`] operations.
#[derive(Debug)]
pub enum HotSwapError {
    /// The pool refused the new entry — wraps a [`PoolError`].  Common
    /// when the requested model exceeds the entire budget even after
    /// evicting every existing entry, or the pool is configured at
    /// zero capacity.
    PoolRefused(PoolError),
    /// The configured loader returned an error (GGUF parse failure,
    /// tokenizer missing, warmup error, etc.).  The manager does not
    /// admit the entry on loader failure.
    LoaderFailed(anyhow::Error),
    /// Filesystem error reading the GGUF file size for budget accounting.
    /// Surfaced as a load-failure rather than swallowed because a
    /// missing or unreadable GGUF is a load-time defect.
    FileSize {
        path: PathBuf,
        source: std::io::Error,
    },
}

impl std::fmt::Display for HotSwapError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::PoolRefused(e) => write!(f, "hot-swap pool refused entry: {e}"),
            Self::LoaderFailed(e) => write!(f, "hot-swap loader failed: {e}"),
            Self::FileSize { path, source } => write!(
                f,
                "hot-swap failed to read GGUF file size for {}: {source}",
                path.display()
            ),
        }
    }
}

impl std::error::Error for HotSwapError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::PoolRefused(e) => Some(e),
            Self::LoaderFailed(e) => Some(e.as_ref()),
            Self::FileSize { source, .. } => Some(source),
        }
    }
}

/// Compose a pool key from `(repo, quant)`.  Two distinct quant variants
/// of the same repo may coexist; the pool keys on `repo_id` so we
/// disambiguate by appending the canonical quant name.
fn pool_key(repo: &str, quant: QuantType) -> String {
    format!("{repo}@{}", quant.as_str())
}

/// Inverse of [`pool_key`] for the iter-213 telemetry path.  Given a
/// victim's composed pool key (`{repo}@{quant_str}`) plus its
/// canonical quant string (already stored on [`LoadedHandle.quant`] —
/// no re-parse required), strip the `@<quant>` suffix and re-parse the
/// quant via [`QuantType::from_canonical_str`].  Returns `None` on
/// either malformed key (no `@`) or unknown quant variant; callers
/// silently skip the counter bump in that case (best-effort
/// telemetry).
///
/// Used at the LRU-eviction trigger site in `load_or_get`: each victim
/// `LoadedHandle` carries a composed `repo_id` rather than the original
/// `(repo, quant)` pair, so the counter-recording call needs to
/// reconstruct the (repo, quant) shape that
/// [`crate::serve::api::state::KvSpillCounters::record_spill`] expects.
fn unpack_pool_key(pool_key_str: &str, quant_str: &str) -> Option<(String, QuantType)> {
    // The canonical key shape is `{repo}@{quant}`; the suffix length
    // is `quant_str.len() + 1` (including the `@`).  We trim that
    // exactly to recover `repo` even if the repo itself contains `@`
    // (defensive — current pool_key callers all pass HF-style
    // `org/name` repos without `@`, but better robust than not).
    let suffix_len = quant_str.len().checked_add(1)?;
    if pool_key_str.len() < suffix_len {
        return None;
    }
    let split = pool_key_str.len() - suffix_len;
    let (repo, sep_and_quant) = pool_key_str.split_at(split);
    if !sep_and_quant.starts_with('@') || &sep_and_quant[1..] != quant_str {
        return None;
    }
    QuantType::from_canonical_str(quant_str)
        .ok()
        .map(|qt| (repo.to_string(), qt))
}

/// Pool-backed engine cache with LRU eviction.
///
/// Composes [`LoadedPool`] (eviction policy, byte accounting) with a
/// [`ModelLoader`] (load path) and a parallel `HashMap<key,
/// Arc<LoadedEngine<E>>>` (the actual engine handles, keyed identically
/// to the pool).
///
/// **Generic over `E`.**  Production wires `E = Engine`; tests use a
/// synthetic stand-in.  The pool eviction + byte accounting logic is
/// engine-agnostic.
///
/// **In-flight request safety.**  Eviction drops the manager's
/// `Arc<LoadedEngine<E>>` from the engines map.  If an axum handler is
/// mid-`generate` and holds its own Arc clone, the engine itself does
/// NOT drop until the handler releases — refcount semantics.  The pool
/// slot becomes available immediately so the new model can admit; the
/// freed bytes don't materialize until the last in-flight request
/// completes.  This is acceptable for a request-rate eviction policy:
/// briefly exceeding the budget while a long generation drains is
/// preferable to either (a) interrupting in-flight work or (b) blocking
/// the new load on an arbitrary in-flight latency.
///
/// **Concurrency.**  The manager itself is not internally synchronized;
/// callers wrap it in `Arc<RwLock<...>>` (iter-209's AppState
/// integration).  `load_or_get` is mutating (LRU touch + insert), so
/// concurrent calls serialize on the write-lock; `try_get` is
/// non-mutating and could be called under the read-lock if the wrapper
/// adds that path.
pub struct HotSwapManager<E> {
    pool: LoadedPool,
    loader: Arc<dyn ModelLoader<E>>,
    /// Eviction-hook surface (ADR-005 Phase 4 reopen iter-212 / AC 5471).
    /// `Arc<dyn KvSpiller<E>>` parallel to the `ModelLoader<E>` field
    /// above so ADR-017 can inject a disk-spilling impl without
    /// `HotSwapManager` taking a direct dependency on the persistence
    /// stack.  The 2-arg back-compat constructor [`Self::new`] defaults
    /// this to `Arc::new(NoopKvSpiller)`; production substitutes via
    /// [`Self::new_with_spiller`].
    spiller: Arc<dyn KvSpiller<E>>,
    /// Optional KV-spill telemetry counters (ADR-005 Phase 4 reopen
    /// iter-213, AC 5472).  When `Some(counters)`, the trigger sites in
    /// `load_or_get` / `evict` bump the per-`(repo, quant, outcome)`
    /// counter row on every `pre_evict` / `post_admit` outcome (Skipped
    /// is excluded — see [`SpillOutcome::Skipped`] semantics).  When
    /// `None`, the trigger sites still call the spiller but no counter
    /// activity is recorded — preserves the back-compat 2-arg
    /// [`HotSwapManager::new`] surface for callers that don't surface
    /// `/metrics` (test fixtures, embedded usages).  Production wiring
    /// via `AppState::new_for_serve` injects the counters with
    /// [`Self::set_kv_counters`].
    kv_counters: Option<Arc<crate::serve::api::state::KvSpillCounters>>,
    /// Parallel map keyed identically to the pool — holds the actual
    /// engine handles.  Invariant: `engines.contains_key(k) ==
    /// pool.get(k).is_some()` for every `k` after every `load_or_get` /
    /// `evict` call returns.
    engines: HashMap<String, Arc<LoadedEngine<E>>>,
}

impl<E> HotSwapManager<E> {
    /// Construct a new manager from a pool + a loader.  Production
    /// passes [`LoadedPool::from_hardware`] (or a fixed-budget pool
    /// for tests) and [`DefaultModelLoader`].  The manager starts
    /// empty; the first `load_or_get` admits the first engine.
    ///
    /// Defaults the eviction-hook surface to [`NoopKvSpiller`].  Use
    /// [`Self::new_with_spiller`] when ADR-017's persistence handler
    /// (or a test fixture) needs to observe eviction + admission.
    pub fn new(pool: LoadedPool, loader: Arc<dyn ModelLoader<E>>) -> Self
    where
        E: 'static,
    {
        Self::new_with_spiller(pool, loader, Arc::new(NoopKvSpiller))
    }

    /// Construct a new manager with an explicit [`KvSpiller`] impl
    /// (ADR-005 Phase 4 reopen iter-212 / AC 5471).  ADR-017's
    /// `cmd_serve --kv-persist` substitutes a disk-backed spiller via
    /// this entry point; tests substitute a `MockSpiller` to assert
    /// trigger-site invariants.
    ///
    /// The manager starts empty; the first `load_or_get` admits the
    /// first engine and fires `spiller.post_admit(...)` between the
    /// loader returning a fresh engine and the engines map publication.
    pub fn new_with_spiller(
        pool: LoadedPool,
        loader: Arc<dyn ModelLoader<E>>,
        spiller: Arc<dyn KvSpiller<E>>,
    ) -> Self {
        Self {
            pool,
            loader,
            spiller,
            kv_counters: None,
            engines: HashMap::new(),
        }
    }

    /// Inject KV-spill telemetry counters (ADR-005 Phase 4 reopen
    /// iter-213, AC 5472).  Production [`crate::serve::api::state::AppState`]
    /// constructors call this immediately after [`Self::new`] so every
    /// trigger-site outcome (`pre_evict` / `post_admit`) bumps the
    /// per-`(repo, quant, outcome)` row that the `/metrics` handler
    /// emits.  Test paths that don't care about counter telemetry
    /// (`mgr_with_spiller` etc.) skip this call — Skipped outcomes
    /// would not increment anyway, and explicit-outcome tests in
    /// router-level tests construct via `AppState` so the counters
    /// thread through automatically.
    pub fn set_kv_counters(&mut self, counters: Arc<crate::serve::api::state::KvSpillCounters>) {
        self.kv_counters = Some(counters);
    }

    /// Iter-215 Wedge-2 test-only admission helper: inject a
    /// pre-built `LoadedEngine` into the pool without invoking the
    /// loader.  Used by router / handlers tests that need a
    /// populated pool to scrape `/metrics`, `/v1/models`, or exercise
    /// the chat 501 short-circuit, without standing up a real GGUF +
    /// GPU + tokenizer.
    ///
    /// Mirrors the production `load_or_get` admission shape (insert
    /// into pool + engines map) but skips loader, spiller, and
    /// counter triggers — those are exercised by other tests.
    /// Returns `Err(PoolError)` if the pool refuses admission.
    #[cfg(test)]
    pub fn admit_for_test(
        &mut self,
        repo: &str,
        quant: QuantType,
        bytes_resident: u64,
        engine: E,
    ) -> Result<Arc<LoadedEngine<E>>, PoolError>
    where
        E: 'static,
    {
        let k = pool_key(repo, quant);
        let loaded_engine = Arc::new(LoadedEngine {
            engine,
            repo: repo.to_string(),
            quant,
            bytes_resident,
            loaded_at: SystemTime::now(),
        });
        let handle = LoadedHandle {
            repo_id: k.clone(),
            quant: quant.as_str().to_string(),
            loaded_at: loaded_engine.loaded_at,
            bytes_resident,
        };
        let evicted = self.pool.insert(handle)?;
        for victim in evicted {
            self.engines.remove(&victim.repo_id);
        }
        self.engines.insert(k, Arc::clone(&loaded_engine));
        Ok(loaded_engine)
    }

    /// Loaded count + capacity + memory budget — surfaced for diagnostics.
    pub fn pool_stats(&self) -> PoolStats {
        PoolStats {
            loaded_count: self.pool.len(),
            capacity_models: self.pool.capacity_models(),
            total_resident_bytes: self.pool.total_resident_bytes(),
            memory_budget_bytes: self.pool.memory_budget_bytes(),
        }
    }

    /// Read-only borrow — does NOT touch the LRU order, does NOT trigger
    /// a load.  Returns `None` when the requested `(repo, quant)` is not
    /// resident in the pool.  Used by routing-only / metrics paths that
    /// need to observe the cache state without mutating it.
    pub fn try_get(&self, repo: &str, quant: QuantType) -> Option<Arc<LoadedEngine<E>>> {
        let k = pool_key(repo, quant);
        self.engines.get(&k).cloned()
    }

    /// Snapshot every currently-pooled `Arc<LoadedEngine<E>>`.  Used by
    /// `cmd_serve`'s graceful-shutdown path (iter-209) to enumerate the
    /// worker handles and join them in parallel.  Cheap-clones every
    /// Arc; does NOT touch the LRU order, does NOT trigger a load.
    /// LRU → MRU iteration order (matches [`LoadedPool::iter`]).
    pub fn snapshot_engines(&self) -> Vec<Arc<LoadedEngine<E>>> {
        self.pool
            .iter()
            .filter_map(|h| self.engines.get(&h.repo_id).cloned())
            .collect()
    }

    /// Iterate the pool's `(pool_key, repo, quant_str, bytes_resident, loaded)` tuples
    /// for `/v1/models` extension fields.  LRU → MRU order.  Read-only.
    pub fn iter_loaded(&self) -> impl Iterator<Item = LoadedSummary> + '_ {
        self.pool.iter().map(|h| LoadedSummary {
            pool_key: h.repo_id.clone(),
            quant: h.quant.clone(),
            bytes_resident: h.bytes_resident,
        })
    }

    /// Force-drop the entry for `(repo, quant)` — symmetric to
    /// [`LoadedPool::remove`].  Returns the bytes freed (0 if the entry
    /// was not in the pool — idempotent).  In-flight requests holding
    /// their own Arc clones keep the engine alive until they release;
    /// the pool slot becomes available immediately.
    ///
    /// **Eviction-hook trigger (iter-212 / AC 5471).**  Fires
    /// `spiller.pre_evict(handle, &Arc<LoadedEngine<E>>)` BEFORE
    /// `engines.remove(&k)` drops the manager's Arc.  The Arc is still
    /// live at call time so the spiller may snapshot KV state without a
    /// race.  The returned [`SpillOutcome`] is intentionally discarded
    /// here in iter-212; iter-213 wires it into the
    /// `hf2q_pool_kv_spills_total{outcome=...}` counter.
    pub fn evict(&mut self, repo: &str, quant: QuantType) -> u64 {
        let k = pool_key(repo, quant);
        let removed = self.pool.remove(&k);
        // Eviction-hook trigger.  Fire BEFORE the manager's Arc drops
        // from `engines` so the spiller observes a live Arc.  Lookup
        // happens before remove() to capture the handle for the call;
        // we only fire when both the pool AND the engines map agree
        // the entry exists (the back-compat defensive `engines.remove`
        // below covers the desync case without firing the hook).
        if let (Some(handle), Some(arc)) = (removed.as_ref(), self.engines.get(&k).cloned()) {
            let outcome = self.spiller.pre_evict(handle, &arc);
            // ADR-005 Phase 4 reopen iter-213 (AC 5472): record outcome
            // to `hf2q_pool_kv_spills_total{repo,quant,outcome=...}`.
            // Skipped does NOT increment (closed-enum guard).
            if let Some(counters) = self.kv_counters.as_ref() {
                counters.record_spill(repo, quant, outcome);
            }
        }
        // Drop the engines map entry symmetrically.  If the pool didn't
        // know about the key (already-evicted from a prior call), the
        // engines map shouldn't have it either by the invariant — but
        // we remove unconditionally to be defensive.
        self.engines.remove(&k);
        // ADR-017 Closure iter-7 (2026-05-04): drop the spiller's
        // per-family hook AFTER engines.remove. The hook holds a
        // cloned `Engine` (iter-3 Phase-E fix) whose inner
        // `Arc<EngineInner>` keeps the worker thread alive — without
        // this drop_family call, every model-swap cycle leaks ~16
        // GB of RSS (verified iter-7 stress smoke: 19.5 GB → 80 GB
        // after 3 iters). drop_family is a no-op for `NoopKvSpiller`
        // (no state to release); `BlockPrefixCacheSpiller` overrides
        // to call `unregister_family` on its registrations map.
        self.spiller.drop_family(repo, quant);
        removed.map(|h| h.bytes_resident).unwrap_or(0)
    }

    /// Return the engine for `(repo, quant)`, loading + admitting if
    /// not already pooled.  On a hit: promotes to MRU and returns the
    /// existing `Arc<LoadedEngine<E>>` clone.  On a miss: invokes the
    /// loader (may take seconds), reads the GGUF file size for byte
    /// accounting, admits the new entry to the pool (which may evict
    /// LRU entries), inserts into the engines map, and returns the
    /// fresh Arc.
    ///
    /// **Errors.**
    ///
    /// - [`HotSwapError::FileSize`] when the GGUF cannot be `metadata`'d.
    /// - [`HotSwapError::LoaderFailed`] when the loader returns an
    ///   error.  Pool + engines map are untouched on loader failure.
    /// - [`HotSwapError::PoolRefused`] when the pool refuses admission
    ///   (oversized handle, zero capacity).  Loader still ran; the
    ///   engine produced is dropped immediately to free GPU buffers.
    pub fn load_or_get(
        &mut self,
        repo: &str,
        quant: QuantType,
        gguf_path: &Path,
        config: &EngineConfig,
    ) -> Result<Arc<LoadedEngine<E>>, HotSwapError> {
        let k = pool_key(repo, quant);

        // Fast path: already loaded.  Touch the LRU and clone the Arc.
        if let Some(existing) = self.engines.get(&k).cloned() {
            self.pool.touch(&k);
            return Ok(existing);
        }

        // Slow path: load.  Read the file size FIRST so a missing GGUF
        // fails fast without driving the loader (which would itself
        // bail at header-open time, but a clean upfront error is
        // better tracing).
        let bytes_resident = std::fs::metadata(gguf_path)
            .map_err(|source| HotSwapError::FileSize {
                path: gguf_path.to_path_buf(),
                source,
            })?
            .len();

        let engine = self
            .loader
            .load(gguf_path, config)
            .map_err(HotSwapError::LoaderFailed)?;

        let loaded_engine = Arc::new(LoadedEngine {
            engine,
            repo: repo.to_string(),
            quant,
            bytes_resident,
            loaded_at: SystemTime::now(),
        });

        // Admit to the pool.  May evict LRU entries; we drop the
        // corresponding engine map entries here so the freed Arcs
        // release.
        let handle = LoadedHandle {
            repo_id: k.clone(),
            quant: quant.as_str().to_string(),
            loaded_at: loaded_engine.loaded_at,
            bytes_resident,
        };

        let evicted = match self.pool.insert(handle) {
            Ok(evicted) => evicted,
            Err(e) => {
                // Pool refused — drop the freshly-loaded engine
                // immediately to free its GPU buffers.  No state
                // mutation visible to the caller.
                drop(loaded_engine);
                return Err(HotSwapError::PoolRefused(e));
            }
        };

        for victim in evicted {
            // Eviction-hook trigger (iter-212 / AC 5471).  Fire
            // `pre_evict` per evictee BEFORE dropping the manager's Arc
            // from `engines`, so the spiller observes a live Arc.  An
            // `Error(...)` outcome MUST NOT block eviction — the spill
            // is best-effort.
            if let Some(arc) = self.engines.get(&victim.repo_id).cloned() {
                let outcome = self.spiller.pre_evict(&victim, &arc);
                // ADR-005 Phase 4 reopen iter-213 (AC 5472): record
                // outcome to
                // `hf2q_pool_kv_spills_total{repo,quant,outcome=...}`.
                // Per-call, NOT per-block: EnqueuedBlocks(N) increments
                // the success row by 1 regardless of N.  The victim's
                // `repo_id` is the pool key (`format!("{repo}@{quant}")`);
                // we re-parse it back into (repo, quant) below.  The
                // pool key shape is enforced by `pool_key()`; both halves
                // are non-empty by construction.
                if let Some(counters) = self.kv_counters.as_ref() {
                    if let Some((victim_repo, victim_quant)) =
                        unpack_pool_key(&victim.repo_id, &victim.quant)
                    {
                        counters.record_spill(&victim_repo, victim_quant, outcome);
                    }
                }
            }
            // Drop the manager's Arc — in-flight requests holding their
            // own clones keep the engine alive until they release; the
            // pool slot is already free.
            self.engines.remove(&victim.repo_id);
            // ADR-017 Closure iter-7 (2026-05-04): drop the spiller's
            // per-family hook AFTER engines.remove. Symmetric with the
            // explicit `evict()` path. victim.repo_id is the pool-key
            // form `{repo}@{quant}`; unpack via the same helper that
            // the metrics path uses so the spiller's bare-repo
            // registration key is matched.
            if let Some((victim_repo, victim_quant)) =
                unpack_pool_key(&victim.repo_id, &victim.quant)
            {
                self.spiller.drop_family(&victim_repo, victim_quant);
            }
        }

        // Admission-hook trigger (iter-212 / AC 5471).  Fire
        // `post_admit` AFTER `loader.load()` returns and BEFORE the
        // manager publishes the Arc into `engines`.  At call time the
        // Arc is held only by the manager's local `loaded_engine`
        // binding; the spiller may rehydrate KV state into the engine
        // before any handler can observe it.  An `Error(...)` outcome
        // MUST NOT block admission — the restore is best-effort.
        let outcome = self.spiller.post_admit(repo, quant, &loaded_engine);
        // ADR-005 Phase 4 reopen iter-213 (AC 5472): record outcome to
        // `hf2q_pool_kv_restores_total{repo,quant,outcome=...}`.
        if let Some(counters) = self.kv_counters.as_ref() {
            counters.record_restore(repo, quant, outcome);
        }

        self.engines.insert(k, Arc::clone(&loaded_engine));
        Ok(loaded_engine)
    }
}

// ─────────────────────────────────────────────────────────────────────
// Tests — synthetic fixtures, no engine load, no GPU.
// ─────────────────────────────────────────────────────────────────────

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

    fn h(repo: &str, bytes: u64) -> LoadedHandle {
        LoadedHandle::new(repo, "Q4_K_M", bytes)
    }

    // --- Construction --------------------------------------------------

    #[test]
    fn empty_pool_is_empty() {
        let p = LoadedPool::with_capacity_and_budget(3, 1_000);
        assert!(p.is_empty());
        assert_eq!(p.len(), 0);
        assert_eq!(p.total_resident_bytes(), 0);
        assert_eq!(p.capacity_models(), 3);
        assert_eq!(p.memory_budget_bytes(), 1_000);
        assert!(p.iter().next().is_none());
    }

    /// Synthetic [`HardwareProfile`] for the from-hardware tests.
    /// `HardwareProfile` does NOT derive `Default` (the bandwidth
    /// field is computed from the chip model elsewhere), so tests
    /// construct one fixture-style here.
    fn synthetic_hw(total_memory_bytes: u64) -> HardwareProfile {
        HardwareProfile {
            chip_model: "Synthetic Test Chip".into(),
            total_memory_bytes,
            available_memory_bytes: total_memory_bytes,
            performance_cores: 8,
            efficiency_cores: 4,
            total_cores: 12,
            memory_bandwidth_gbs: 400.0,
        }
    }

    #[test]
    fn from_hardware_applies_eighty_percent_default() {
        let hw = synthetic_hw(128 * 1024 * 1024 * 1024); // 128 GiB
        let p = LoadedPool::from_hardware(&hw);
        assert_eq!(p.capacity_models(), DEFAULT_POOL_CAPACITY);
        // 0.80 × 128 GiB = 102.4 GiB exact (fp64 has the precision)
        let expected = ((128.0_f64 * 1024.0 * 1024.0 * 1024.0) * 0.80).floor() as u64;
        assert_eq!(p.memory_budget_bytes(), expected);
    }

    #[test]
    fn from_hardware_with_clamps_fraction() {
        let hw = synthetic_hw(1_000_000);
        // 1.5 → clamps to 1.0 → full physical
        let p_high = LoadedPool::from_hardware_with(&hw, 3, 1.5);
        assert_eq!(p_high.memory_budget_bytes(), 1_000_000);
        // -0.1 → clamps to 0.0 → zero budget
        let p_low = LoadedPool::from_hardware_with(&hw, 3, -0.1);
        assert_eq!(p_low.memory_budget_bytes(), 0);
    }

    // --- Insert: capacity-driven eviction ------------------------------

    #[test]
    fn capacity_one_evicts_on_second_insert() {
        let mut p = LoadedPool::with_capacity_and_budget(1, 1_000_000);
        let evicted_a = p.insert(h("a/1", 100)).unwrap();
        assert!(evicted_a.is_empty());
        let evicted_b = p.insert(h("b/2", 200)).unwrap();
        assert_eq!(evicted_b.len(), 1);
        assert_eq!(evicted_b[0].repo_id, "a/1");
        assert_eq!(p.len(), 1);
        assert!(p.get("a/1").is_none());
        assert!(p.get("b/2").is_some());
        assert_eq!(p.total_resident_bytes(), 200);
    }

    #[test]
    fn capacity_three_evicts_lru_first() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 1_000_000_000);
        let _ = p.insert(h("a/1", 100)).unwrap(); // LRU
        let _ = p.insert(h("b/2", 200)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap(); // MRU
        assert_eq!(p.len(), 3);
        // 4th insert evicts a/1 (LRU)
        let evicted = p.insert(h("d/4", 400)).unwrap();
        assert_eq!(evicted.len(), 1);
        assert_eq!(evicted[0].repo_id, "a/1");
        // pool order is now b, c, d (LRU → MRU)
        let order: Vec<&str> = p.iter().map(|h| h.repo_id.as_str()).collect();
        assert_eq!(order, vec!["b/2", "c/3", "d/4"]);
        assert_eq!(p.total_resident_bytes(), 200 + 300 + 400);
    }

    // --- Insert: budget-driven eviction --------------------------------

    #[test]
    fn budget_evicts_without_capacity_overflow() {
        // Capacity 5, budget 1_000.  Three handles totalling 900 fit;
        // 4th handle of 200 forces budget eviction even though capacity
        // is not breached.
        let mut p = LoadedPool::with_capacity_and_budget(5, 1_000);
        let _ = p.insert(h("a/1", 300)).unwrap(); // LRU, 300
        let _ = p.insert(h("b/2", 300)).unwrap(); // total 600
        let _ = p.insert(h("c/3", 300)).unwrap(); // total 900, MRU
        assert_eq!(p.total_resident_bytes(), 900);
        // 4th handle 200: 900+200=1100 > 1000 → evict LRU until fit.
        // Evict a/1 (300) → total 800, +200 = 1000 ≤ 1000. Stop.
        let evicted = p.insert(h("d/4", 200)).unwrap();
        assert_eq!(evicted.len(), 1);
        assert_eq!(evicted[0].repo_id, "a/1");
        assert_eq!(p.len(), 3);
        assert_eq!(p.total_resident_bytes(), 800);
    }

    #[test]
    fn budget_eviction_chains_until_fit() {
        // Three small handles + one big new handle that requires
        // evicting all three.
        let mut p = LoadedPool::with_capacity_and_budget(5, 1_000);
        let _ = p.insert(h("a/1", 300)).unwrap();
        let _ = p.insert(h("b/2", 300)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap();
        let evicted = p.insert(h("big/1", 950)).unwrap();
        assert_eq!(evicted.len(), 3);
        assert_eq!(
            evicted
                .iter()
                .map(|h| h.repo_id.as_str())
                .collect::<Vec<_>>(),
            vec!["a/1", "b/2", "c/3"]
        );
        assert_eq!(p.len(), 1);
        assert_eq!(p.total_resident_bytes(), 950);
    }

    #[test]
    fn capacity_and_budget_evict_in_one_insert() {
        // Capacity 2, budget 800.  Insert two handles at 400 each;
        // third at 500 trips capacity AND budget — both passes run
        // and the eviction list reflects both.
        let mut p = LoadedPool::with_capacity_and_budget(2, 800);
        let _ = p.insert(h("a/1", 400)).unwrap();
        let _ = p.insert(h("b/2", 400)).unwrap(); // total 800, capacity 2/2
                                                  // Third: capacity pass evicts a/1 → total 400, len 1.  Budget
                                                  // pass: 400+500=900 > 800 → evict b/2 → total 0.  +500 = 500.
        let evicted = p.insert(h("c/3", 500)).unwrap();
        assert_eq!(evicted.len(), 2);
        let names: Vec<&str> = evicted.iter().map(|h| h.repo_id.as_str()).collect();
        assert_eq!(names, vec!["a/1", "b/2"]);
        assert_eq!(p.len(), 1);
        assert_eq!(p.total_resident_bytes(), 500);
    }

    // --- touch / get separation ----------------------------------------

    #[test]
    fn touch_promotes_to_mru() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 1_000_000);
        let _ = p.insert(h("a/1", 100)).unwrap(); // LRU
        let _ = p.insert(h("b/2", 200)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap(); // MRU
        assert!(p.touch("a/1"));
        // Order should now be b, c, a (LRU → MRU).
        let order: Vec<&str> = p.iter().map(|h| h.repo_id.as_str()).collect();
        assert_eq!(order, vec!["b/2", "c/3", "a/1"]);
        // 4th insert evicts b/2 (the new LRU), not a/1.
        let evicted = p.insert(h("d/4", 400)).unwrap();
        assert_eq!(evicted.len(), 1);
        assert_eq!(evicted[0].repo_id, "b/2");
    }

    #[test]
    fn get_does_not_touch() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 1_000_000);
        let _ = p.insert(h("a/1", 100)).unwrap(); // LRU
        let _ = p.insert(h("b/2", 200)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap();
        let _ = p.get("a/1").unwrap();
        // a/1 must still be LRU — get is read-only.
        let evicted = p.insert(h("d/4", 400)).unwrap();
        assert_eq!(evicted.len(), 1);
        assert_eq!(evicted[0].repo_id, "a/1");
    }

    #[test]
    fn touch_unknown_is_noop() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 1_000);
        assert!(!p.touch("nope/0"));
    }

    // --- remove --------------------------------------------------------

    #[test]
    fn remove_returns_handle_and_updates_bytes() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 1_000);
        let _ = p.insert(h("a/1", 100)).unwrap();
        let _ = p.insert(h("b/2", 200)).unwrap();
        assert_eq!(p.total_resident_bytes(), 300);
        let removed = p.remove("a/1").unwrap();
        assert_eq!(removed.repo_id, "a/1");
        assert_eq!(removed.bytes_resident, 100);
        assert_eq!(p.total_resident_bytes(), 200);
        assert_eq!(p.len(), 1);
        // Removing again is a no-op.
        assert!(p.remove("a/1").is_none());
    }

    // --- Idempotent insert (re-load) -----------------------------------

    #[test]
    fn reinsert_same_repo_updates_bytes_and_promotes() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 10_000);
        let _ = p.insert(h("a/1", 100)).unwrap(); // LRU
        let _ = p.insert(h("b/2", 200)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap(); // MRU
                                                  // Re-insert a/1 with new bytes 1500: it should NOT be evicted,
                                                  // total goes 100→1500 = +1400, no other eviction (well under
                                                  // budget).  a/1 promotes to MRU.
        let evicted = p.insert(h("a/1", 1500)).unwrap();
        assert!(evicted.is_empty(), "re-insert must not self-evict");
        let order: Vec<&str> = p.iter().map(|h| h.repo_id.as_str()).collect();
        assert_eq!(order, vec!["b/2", "c/3", "a/1"]);
        assert_eq!(p.total_resident_bytes(), 1500 + 200 + 300);
        assert_eq!(p.get("a/1").unwrap().bytes_resident, 1500);
    }

    #[test]
    fn reinsert_with_budget_overflow_evicts_others_not_self() {
        // Budget 1000.  a:300, b:300, c:300 = 900.  Re-insert b:800
        // would push total to 1400 → evict OTHERS (a is LRU because b
        // promoted; then c) but never b.
        let mut p = LoadedPool::with_capacity_and_budget(5, 1_000);
        let _ = p.insert(h("a/1", 300)).unwrap();
        let _ = p.insert(h("b/2", 300)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap();
        // Re-insert b with bigger bytes.  After the bytes update the
        // total is 300+800+300 = 1400; we evict LRU (a) → 1100, still
        // > 1000, evict next non-spare LRU (c) → 800.  Stop.
        let evicted = p.insert(h("b/2", 800)).unwrap();
        let evicted_names: Vec<&str> = evicted.iter().map(|h| h.repo_id.as_str()).collect();
        assert_eq!(evicted_names, vec!["a/1", "c/3"]);
        assert!(p.get("b/2").is_some(), "self never evicted on re-insert");
        assert_eq!(p.len(), 1);
        assert_eq!(p.total_resident_bytes(), 800);
    }

    // --- Refusal cases -------------------------------------------------

    #[test]
    fn zero_capacity_refuses_every_insert() {
        let mut p = LoadedPool::with_capacity_and_budget(0, 1_000);
        let err = p.insert(h("a/1", 100)).unwrap_err();
        assert_eq!(err, PoolError::ZeroCapacity);
        let msg = format!("{err}");
        assert!(msg.contains("capacity_models = 0"), "msg = {msg}");
    }

    #[test]
    fn oversized_handle_refused() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 1_000);
        let err = p.insert(h("big/1", 1_001)).unwrap_err();
        match err {
            PoolError::OversizedHandle {
                repo_id,
                handle_bytes,
                budget_bytes,
            } => {
                assert_eq!(repo_id, "big/1");
                assert_eq!(handle_bytes, 1_001);
                assert_eq!(budget_bytes, 1_000);
            }
            other => panic!("unexpected error: {other:?}"),
        }
        // Pool is unchanged.
        assert!(p.is_empty());
    }

    #[test]
    fn zero_budget_refuses_any_nonzero_handle() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 0);
        let err = p.insert(h("a/1", 1)).unwrap_err();
        assert!(matches!(err, PoolError::OversizedHandle { .. }));
        // A zero-byte handle would technically fit a zero budget, but
        // a real handle is never zero bytes — the test below documents
        // that we do not gratuitously refuse it.
        let evicted = p.insert(h("zerobyte/0", 0)).unwrap();
        assert!(evicted.is_empty());
        assert_eq!(p.len(), 1);
    }

    // --- Iter order ----------------------------------------------------

    #[test]
    fn iter_yields_lru_to_mru() {
        let mut p = LoadedPool::with_capacity_and_budget(3, 10_000);
        let _ = p.insert(h("a/1", 100)).unwrap();
        let _ = p.insert(h("b/2", 200)).unwrap();
        let _ = p.insert(h("c/3", 300)).unwrap();
        let order: Vec<&str> = p.iter().map(|h| h.repo_id.as_str()).collect();
        assert_eq!(order, vec!["a/1", "b/2", "c/3"]);
    }

    // --- Display + error formatting ------------------------------------

    #[test]
    fn oversized_error_message_names_budget_and_repo() {
        let err = PoolError::OversizedHandle {
            repo_id: "huge/model".into(),
            handle_bytes: 9_000,
            budget_bytes: 1_000,
        };
        let msg = format!("{err}");
        assert!(msg.contains("huge/model"));
        assert!(msg.contains("9000"));
        assert!(msg.contains("1000"));
        assert!(msg.contains("Phase 4"));
    }

    // ─────────────────────────────────────────────────────────────────
    // HotSwapManager tests — synthetic engine fixture (`MockEngine` is
    // a unit struct), MockLoader writes a temp GGUF (just bytes) so
    // `std::fs::metadata().len()` returns a deterministic byte count
    // for the pool's budget accounting.  Production E = Engine path
    // is exercised by tests/multi_model_hotswap.rs (env-gated E2E).
    // ─────────────────────────────────────────────────────────────────

    /// Synthetic engine type used by the manager unit tests.  The pool's
    /// eviction logic is engine-agnostic — it only sees `bytes_resident`
    /// and the LRU order — so the test fixture just needs a `Send +
    /// Sync` placeholder that survives an `Arc::clone`.  Carries an id
    /// so tests can assert "the same Arc came back" via field equality.
    #[derive(Debug, Clone, PartialEq, Eq)]
    struct MockEngine {
        load_serial: u64,
    }

    /// Test loader.  Tracks call count + an optional error injection
    /// so tests can exercise both happy + failure paths.  Holds an
    /// AtomicU64 for the load-serial counter so the manager can
    /// observe distinct engine instances across calls.
    struct MockLoader {
        calls: std::sync::atomic::AtomicU64,
        fail_on_call: Option<u64>, // 1-indexed; None = never fail
    }

    impl MockLoader {
        fn new() -> Self {
            Self {
                calls: std::sync::atomic::AtomicU64::new(0),
                fail_on_call: None,
            }
        }
        fn fail_on(call_num: u64) -> Self {
            Self {
                calls: std::sync::atomic::AtomicU64::new(0),
                fail_on_call: Some(call_num),
            }
        }
        fn call_count(&self) -> u64 {
            self.calls.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    impl ModelLoader<MockEngine> for MockLoader {
        fn load(&self, _path: &Path, _config: &EngineConfig) -> anyhow::Result<MockEngine> {
            let n = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
            if self.fail_on_call == Some(n) {
                anyhow::bail!("MockLoader synthetic failure on call {n}");
            }
            Ok(MockEngine { load_serial: n })
        }
    }

    /// Make a temp GGUF-shaped fixture file of `size` bytes.  The
    /// manager reads the file size for budget accounting; the loader
    /// is a mock so the bytes' content doesn't matter.  Returns a
    /// `tempfile::NamedTempFile` so the file lives until the test
    /// function returns.
    fn synthetic_gguf(size: usize) -> tempfile::NamedTempFile {
        use std::io::Write;
        let mut f = tempfile::NamedTempFile::new().expect("temp file");
        // Write `size` zero bytes — the manager only cares about
        // `metadata().len()`, not the content.
        let chunk = vec![0u8; 4096.min(size)];
        let mut remaining = size;
        while remaining > 0 {
            let n = remaining.min(chunk.len());
            f.write_all(&chunk[..n]).expect("write");
            remaining -= n;
        }
        f.flush().expect("flush");
        f
    }

    fn empty_config() -> EngineConfig {
        EngineConfig::default()
    }

    // --- Load + reuse path ---------------------------------------------

    #[test]
    fn hotswap_loads_on_first_request() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());

        let f = synthetic_gguf(1_000);
        let cfg = empty_config();
        let engine = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("first load");

        assert_eq!(engine.repo, "acme/m1");
        assert_eq!(engine.quant, QuantType::Q4_K_M);
        assert_eq!(engine.bytes_resident, 1_000);
        assert_eq!(engine.engine.load_serial, 1);
        assert_eq!(loader.call_count(), 1);
        // Pool reflects the load.
        let stats = mgr.pool_stats();
        assert_eq!(stats.loaded_count, 1);
        assert_eq!(stats.total_resident_bytes, 1_000);
    }

    #[test]
    fn hotswap_reuses_pooled_engine() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let e1 = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("first");
        let e2 = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("second");

        // Same Arc — refcount-equality.
        assert!(Arc::ptr_eq(&e1, &e2), "second call must return same Arc");
        // Loader called exactly once.
        assert_eq!(loader.call_count(), 1);
    }

    #[test]
    fn hotswap_evicts_lru_on_pressure() {
        // Capacity 2, budget large enough to bypass byte-budget
        // eviction so we exercise the capacity path cleanly.
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(2, 1_000_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f = synthetic_gguf(1_000);
        let cfg = empty_config();

        let _e1 = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a/1");
        let _e2 = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b/2");
        let _e3 = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("c/3");

        // a/1 was LRU at the third load → evicted.
        assert!(mgr.try_get("a/1", QuantType::Q4_K_M).is_none());
        assert!(mgr.try_get("b/2", QuantType::Q4_K_M).is_some());
        assert!(mgr.try_get("c/3", QuantType::Q4_K_M).is_some());
        let stats = mgr.pool_stats();
        assert_eq!(stats.loaded_count, 2);
    }

    #[test]
    fn hotswap_evicts_lru_on_byte_pressure() {
        // Capacity comfortable, budget tight — third load forces a
        // byte-budget eviction even though capacity is 5.
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(5, 2_500);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f1 = synthetic_gguf(1_000);
        let f2 = synthetic_gguf(1_000);
        let f3 = synthetic_gguf(1_000);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f1.path(), &cfg)
            .expect("a");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f2.path(), &cfg)
            .expect("b");
        // 1000 + 1000 + 1000 = 3000 > 2500 → evict a/1 (LRU) → 2000.
        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f3.path(), &cfg)
            .expect("c");

        assert!(mgr.try_get("a/1", QuantType::Q4_K_M).is_none());
        let stats = mgr.pool_stats();
        assert_eq!(stats.loaded_count, 2);
        assert_eq!(stats.total_resident_bytes, 2_000);
    }

    #[test]
    fn hotswap_errors_when_no_evictable_fits() {
        // Budget 500, GGUF file 1500 → oversized handle even with
        // empty pool.  Loader still runs (the manager invokes it
        // before the pool admission attempt by design — the alternative
        // would be to file-stat first, but file size IS the byte
        // estimate so we already do that).  Verify the engine is
        // dropped (`Arc::strong_count == 0` after the error).
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 500);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f = synthetic_gguf(1_500);
        let cfg = empty_config();

        let err = mgr
            .load_or_get("big/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect_err("should refuse oversized");
        match err {
            HotSwapError::PoolRefused(PoolError::OversizedHandle {
                repo_id,
                handle_bytes,
                budget_bytes,
            }) => {
                assert_eq!(repo_id, "big/1@Q4_K_M");
                assert_eq!(handle_bytes, 1_500);
                assert_eq!(budget_bytes, 500);
            }
            other => panic!("unexpected error: {other:?}"),
        }
        // Loader was invoked (engine produced + dropped).
        assert_eq!(loader.call_count(), 1);
        // Manager state is unchanged.
        assert_eq!(mgr.pool_stats().loaded_count, 0);
        assert!(mgr.try_get("big/1", QuantType::Q4_K_M).is_none());
    }

    #[test]
    fn hotswap_evict_explicit_removes() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f = synthetic_gguf(700);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("load");
        assert_eq!(mgr.pool_stats().loaded_count, 1);

        let bytes_freed = mgr.evict("acme/m1", QuantType::Q4_K_M);
        assert_eq!(bytes_freed, 700);
        assert!(mgr.try_get("acme/m1", QuantType::Q4_K_M).is_none());
        assert_eq!(mgr.pool_stats().loaded_count, 0);
        assert_eq!(mgr.pool_stats().total_resident_bytes, 0);

        // Idempotent: second evict returns 0.
        assert_eq!(mgr.evict("acme/m1", QuantType::Q4_K_M), 0);
    }

    #[test]
    fn hotswap_try_get_returns_none_when_absent() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 1_000);
        let mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        assert!(mgr.try_get("nope/0", QuantType::Q4_K_M).is_none());
    }

    #[test]
    fn hotswap_try_get_returns_arc_when_present() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let loaded = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("load");
        let viewed = mgr.try_get("acme/m1", QuantType::Q4_K_M).expect("present");
        // Same Arc.
        assert!(Arc::ptr_eq(&loaded, &viewed));
    }

    #[test]
    fn hotswap_try_get_does_not_touch_lru() {
        // Mirrors W74's `get_does_not_touch` test for the manager
        // surface: try_get must NOT promote the entry, otherwise a
        // diagnostic / metrics path peeking at the cache would poison
        // the LRU policy.
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(2, 1_000_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f = synthetic_gguf(1_000);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b");
        // Peek at a/1 — must not promote.
        let _peek = mgr.try_get("a/1", QuantType::Q4_K_M).unwrap();
        // Third load: a/1 should still be LRU and evicted.
        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("c");
        assert!(
            mgr.try_get("a/1", QuantType::Q4_K_M).is_none(),
            "try_get must NOT promote — a/1 should evict as LRU"
        );
    }

    #[test]
    fn hotswap_pool_stats_reflects_loads_and_evictions() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(2, 5_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f1 = synthetic_gguf(800);
        let f2 = synthetic_gguf(800);
        let cfg = empty_config();

        // Empty.
        let s0 = mgr.pool_stats();
        assert_eq!(s0.loaded_count, 0);
        assert_eq!(s0.total_resident_bytes, 0);
        assert_eq!(s0.capacity_models, 2);
        assert_eq!(s0.memory_budget_bytes, 5_000);

        // One load.
        let _ = mgr.load_or_get("a/1", QuantType::Q4_K_M, f1.path(), &cfg);
        let s1 = mgr.pool_stats();
        assert_eq!(s1.loaded_count, 1);
        assert_eq!(s1.total_resident_bytes, 800);

        // Two loads.
        let _ = mgr.load_or_get("b/2", QuantType::Q4_K_M, f2.path(), &cfg);
        let s2 = mgr.pool_stats();
        assert_eq!(s2.loaded_count, 2);
        assert_eq!(s2.total_resident_bytes, 1_600);

        // Explicit evict drops to 1.
        mgr.evict("a/1", QuantType::Q4_K_M);
        let s3 = mgr.pool_stats();
        assert_eq!(s3.loaded_count, 1);
        assert_eq!(s3.total_resident_bytes, 800);
    }

    #[test]
    fn hotswap_loader_error_propagates() {
        // Loader fails on first call → manager returns LoaderFailed
        // and does NOT admit the entry.
        let loader = Arc::new(MockLoader::fail_on(1));
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let err = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect_err("loader failure must propagate");
        match err {
            HotSwapError::LoaderFailed(e) => {
                let msg = format!("{e}");
                assert!(
                    msg.contains("synthetic failure"),
                    "expected synthetic failure msg, got: {msg}"
                );
            }
            other => panic!("unexpected error: {other:?}"),
        }
        assert_eq!(loader.call_count(), 1);
        // No state mutation — the entry was never admitted.
        assert_eq!(mgr.pool_stats().loaded_count, 0);
        assert!(mgr.try_get("acme/m1", QuantType::Q4_K_M).is_none());
    }

    #[test]
    fn hotswap_file_size_error_when_gguf_missing() {
        // GGUF path doesn't exist → FileSize error before the loader
        // is invoked.  Verifies the manager's pre-load file-stat
        // catches missing files with a clean named error.
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let cfg = empty_config();

        let err = mgr
            .load_or_get(
                "acme/m1",
                QuantType::Q4_K_M,
                Path::new("/nonexistent/path/to/no.gguf"),
                &cfg,
            )
            .expect_err("missing GGUF must error");
        match err {
            HotSwapError::FileSize { path, .. } => {
                assert!(path.to_string_lossy().contains("/nonexistent/"));
            }
            other => panic!("unexpected error: {other:?}"),
        }
        // Loader was NOT invoked.
        assert_eq!(loader.call_count(), 0);
        assert_eq!(mgr.pool_stats().loaded_count, 0);
    }

    #[test]
    fn hotswap_two_quants_of_same_repo_coexist() {
        // The pool key is `format!("{repo}@{quant}")` so two distinct
        // quant variants of the same repo can both be resident.
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("Q4_K_M");
        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q8_0, f.path(), &cfg)
            .expect("Q8_0");

        assert!(mgr.try_get("acme/m1", QuantType::Q4_K_M).is_some());
        assert!(mgr.try_get("acme/m1", QuantType::Q8_0).is_some());
        assert_eq!(mgr.pool_stats().loaded_count, 2);
        assert_eq!(loader.call_count(), 2);
    }

    #[test]
    fn hotswap_in_flight_arc_survives_eviction() {
        // Capacity 1.  Hold the Arc from the first load through the
        // second load; the manager evicts the first, but the Arc
        // refcount keeps the engine alive.  Drop the held Arc and
        // confirm refcount drops to 1 (just the held local, since the
        // manager already released its reference).
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(1, 100_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let inflight = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a/1");
        // strong_count: 1 (manager) + 1 (inflight) = 2
        assert_eq!(Arc::strong_count(&inflight), 2);

        // Second load evicts a/1 from the manager.
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b/2");

        // Manager dropped its Arc → strong_count = 1 (just inflight).
        assert_eq!(
            Arc::strong_count(&inflight),
            1,
            "manager must have released its Arc on eviction"
        );
        // Inflight Arc still valid — the engine wasn't dropped.
        assert_eq!(inflight.repo, "a/1");
        assert_eq!(inflight.engine.load_serial, 1);

        // Drop inflight → engine drops now.
        drop(inflight);
        // (We can't assert "engine dropped" directly without a Drop
        // impl on MockEngine; the strong_count == 1 invariant above
        // is the load-bearing assertion for this test.)
    }

    // ────────────────────────────────────────────────────────────────────
    // ADR-005 Phase 4 iter-210 (W78) — AC 5467 closure
    // ────────────────────────────────────────────────────────────────────
    //
    // AC 5467 ("Cached pool holds up to 3 loaded models with LRU eviction
    // bounded by 80% of system unified memory (configurable)") composes
    // three already-tested invariants:
    //
    //   1. `LoadedPool::from_hardware` reads `total_memory_bytes` and
    //      applies the 80% default fraction (covered by
    //      `from_hardware_applies_eighty_percent_default`).
    //   2. `HotSwapManager` admits up to capacity_models entries in
    //      LRU order (covered by `hotswap_evicts_lru_on_pressure`).
    //   3. `pool_stats()` surfaces loaded_count + total_resident_bytes +
    //      memory_budget_bytes for /metrics consumption (covered by
    //      `hotswap_pool_stats_reflects_loads_and_evictions`).
    //
    // The new test below asserts all three invariants TOGETHER through
    // the manager surface — three distinct repos, all admitted, all
    // resident, total bytes under the budget, count == 3.  Combined
    // with the metrics gauge tests in `serve::api::router::tests::
    // iter210_metrics_emits_pool_gauges`, this closes the AC at the
    // unit-test level without requiring a 100 GiB-budget E2E run.

    #[test]
    fn pool_holds_three_models_within_budget() {
        // ADR-005 line 929 spec: capacity_models = 3 (DEFAULT_POOL_CAPACITY).
        // Synthetic budget large enough to fit three 1 KiB handles with
        // significant headroom — proves no eviction fires under nominal
        // multi-model load.
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(DEFAULT_POOL_CAPACITY, 1_000_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let cfg = empty_config();

        let f1 = synthetic_gguf(1_024);
        let f2 = synthetic_gguf(1_024);
        let f3 = synthetic_gguf(1_024);

        let _e1 = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f1.path(), &cfg)
            .expect("admit m1");
        let _e2 = mgr
            .load_or_get("acme/m2", QuantType::Q4_K_M, f2.path(), &cfg)
            .expect("admit m2");
        let _e3 = mgr
            .load_or_get("acme/m3", QuantType::Q4_K_M, f3.path(), &cfg)
            .expect("admit m3");

        // All three present — no eviction at capacity boundary +
        // no byte-budget eviction (3 KiB << 1 MiB budget).
        let stats = mgr.pool_stats();
        assert_eq!(
            stats.loaded_count, 3,
            "expected 3 distinct repos pooled; stats={stats:?}"
        );
        assert_eq!(stats.capacity_models, DEFAULT_POOL_CAPACITY);
        assert_eq!(stats.total_resident_bytes, 3 * 1_024);
        assert!(
            stats.total_resident_bytes < stats.memory_budget_bytes,
            "AC 5467 invariant violated: total_resident_bytes={} >= memory_budget_bytes={}",
            stats.total_resident_bytes,
            stats.memory_budget_bytes,
        );
        // Loader called once per distinct repo.
        assert_eq!(loader.call_count(), 3);

        // All three reachable via try_get (no LRU promotion side-effect).
        assert!(mgr.try_get("acme/m1", QuantType::Q4_K_M).is_some());
        assert!(mgr.try_get("acme/m2", QuantType::Q4_K_M).is_some());
        assert!(mgr.try_get("acme/m3", QuantType::Q4_K_M).is_some());
    }

    #[test]
    fn pool_from_hardware_yields_eighty_percent_budget_for_three_models() {
        // AC 5467 second leg: the 80% memory ceiling is sourced from
        // `HardwareProfile::total_memory_bytes` via
        // `LoadedPool::from_hardware`.  This test wires the production
        // factory + a synthetic 128 GiB hardware profile (the M5 Max
        // unified-memory tier the AC references) and asserts the
        // budget yields a pool that admits three Gemma-sized (16 GiB)
        // handles with headroom.
        const M5_MAX_UNIFIED_BYTES: u64 = 128 * 1024 * 1024 * 1024;
        const GEMMA_SIZED_BYTES: u64 = 16 * 1024 * 1024 * 1024;

        let hw = synthetic_hw(M5_MAX_UNIFIED_BYTES);
        let pool = LoadedPool::from_hardware(&hw);

        // 80% of 128 GiB = 102.4 GiB.  Floor to u64; parity with
        // the existing `from_hardware_applies_eighty_percent_default`
        // test's arithmetic.
        let expected_budget = ((M5_MAX_UNIFIED_BYTES as f64) * 0.80).floor() as u64;
        assert_eq!(pool.memory_budget_bytes(), expected_budget);
        assert_eq!(pool.capacity_models(), DEFAULT_POOL_CAPACITY);

        // Three 16 GiB handles must fit (3 × 16 = 48 GiB < 102.4 GiB).
        assert!(
            3 * GEMMA_SIZED_BYTES < pool.memory_budget_bytes(),
            "M5 Max budget {} cannot fit 3× 16 GiB Gemma; AC 5467 sizing assumption broken",
            pool.memory_budget_bytes()
        );
    }

    // ─────────────────────────────────────────────────────────────────────
    // ADR-005 Wave 3 W-B1 (T1.11) — hot-swap LRU evict gap-test enforcement.
    //
    // Context:  the eviction *loop* (`LoadedPool::insert` capacity + budget
    // two-pass) and its `HotSwapManager::load_or_get` driver shipped at
    // iter-208 / iter-210 (W76 + W78) and closed AC 5467 ("Cached pool
    // holds up to 3 loaded models with LRU eviction bounded by 80% of
    // system unified memory").  The four tests below close the testing
    // *gap* the W-B1 audit surfaced — production behavior under memory
    // pressure WAS exercised at the pool-primitive level, but four
    // manager-surface invariants the Wave-2 W-2 llama.cpp reference
    // (server-models.cpp:498-549) calls out were not asserted through
    // `HotSwapManager`'s public API:
    //
    //   1. **Multi-evict-in-one-load** (`hotswap_chains_multiple_evictions_in_one_load`):
    //      a single `load_or_get` may need to evict TWO OR MORE LRU
    //      entries to fit the new model — the chained-budget pass at
    //      `LoadedPool::insert` lines 434-447 already does this, but no
    //      manager-level test asserts the chain.
    //
    //   2. **Oversized-with-prior-pool** (`hotswap_oversized_handle_preserves_existing_entries`):
    //      `LoadedPool::insert` pre-checks `handle.bytes > budget` at
    //      line 377 *before* any eviction runs.  The existing
    //      `hotswap_errors_when_no_evictable_fits` test only exercises
    //      the empty-pool case; this new test asserts that an oversized
    //      load against a *populated* pool refuses cleanly without
    //      touching the existing residents.
    //
    //   3. **Cache-hit-promotes-LRU** (`hotswap_load_or_get_promotes_on_cache_hit`):
    //      the fast path at `HotSwapManager::load_or_get` line 789 calls
    //      `self.pool.touch(&k)` on cache hit.  The existing
    //      `hotswap_try_get_does_not_touch_lru` test asserts the
    //      *inverse* (the diagnostic surface `try_get` MUST NOT touch);
    //      this new test asserts the request-path symmetric: a second
    //      `load_or_get` for an already-pooled repo SHOULD touch and
    //      thereby protect the entry from being evicted as LRU.
    //
    //   4. **TOCTOU-safe under shared lock** (`hotswap_concurrent_load_or_get_serializes_under_mutex`):
    //      `HotSwapManager` itself is not internally synchronized
    //      (struct doc lines 676-681 + AppState line 202 wrap it in
    //      `Arc<RwLock<...>>`).  llama.cpp's
    //      `server_models.cpp::load` (lines 545-558) re-checks capacity
    //      under the load-lock to defeat the
    //      check-budget → release-lock → load → re-acquire window.
    //      Our wrapper holds a single std::sync RwLock write-guard for
    //      the entire `load_or_get` call (file-stat + loader invoke +
    //      pool insert), so the budget-check → admit sequence is
    //      atomic by construction.  This regression test wraps the
    //      manager in the same `Arc<Mutex<...>>` shape and spawns N
    //      concurrent threads; it asserts (a) the loader is called
    //      exactly N times (no double-load on the same key + no missed
    //      load), (b) the final pool state respects capacity, and (c)
    //      `total_resident_bytes <= memory_budget_bytes` holds at every
    //      observable point post-join (the budget invariant never tears).
    //
    // Reference:  `/opt/llama.cpp/tools/server/server-models.cpp:498-549`
    // (the canonical LRU+TOCTOU pattern; our equivalent is the
    // single-lock-guard wrapping the entire `load_or_get` call).
    // ─────────────────────────────────────────────────────────────────────

    /// W-B1 test 1/4: a single `load_or_get` can evict multiple LRU
    /// entries when one eviction is insufficient to fit the new model.
    /// Capacity 5 (no capacity pressure); budget 1500.  Load A=500,
    /// B=500, C=500 sequentially → total 1500 (= budget, no eviction).
    /// Load D=1000: 1500+1000=2500>1500 → evict A → 1000+1000=2000>1500
    /// → evict B → 500+1000=1500 ≤ 1500.  Stop.  D evicts A AND B in a
    /// single `load_or_get` call.
    #[test]
    fn hotswap_chains_multiple_evictions_in_one_load() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(5, 1_500);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f_small = synthetic_gguf(500);
        let f_big = synthetic_gguf(1_000);
        let cfg = empty_config();

        // Three 500-byte loads — pool fills exactly at budget.
        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("a");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("b");
        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("c");
        assert_eq!(mgr.pool_stats().loaded_count, 3);
        assert_eq!(mgr.pool_stats().total_resident_bytes, 1_500);

        // 1000-byte load forces TWO evictions in one call.
        let _ = mgr
            .load_or_get("d/4", QuantType::Q4_K_M, f_big.path(), &cfg)
            .expect("d");

        // A and B (the two LRU entries) both evicted; C survives;
        // D resident.  total = 500 (C) + 1000 (D) = 1500.
        assert!(
            mgr.try_get("a/1", QuantType::Q4_K_M).is_none(),
            "A (LRU) must be evicted"
        );
        assert!(
            mgr.try_get("b/2", QuantType::Q4_K_M).is_none(),
            "B (second-LRU) must also be evicted — single eviction insufficient"
        );
        assert!(
            mgr.try_get("c/3", QuantType::Q4_K_M).is_some(),
            "C (MRU before D) must survive — eviction stops as soon as budget fits"
        );
        assert!(
            mgr.try_get("d/4", QuantType::Q4_K_M).is_some(),
            "D (newly admitted) must be resident"
        );
        let stats = mgr.pool_stats();
        assert_eq!(stats.loaded_count, 2);
        assert_eq!(stats.total_resident_bytes, 1_500);
        assert!(
            stats.total_resident_bytes <= stats.memory_budget_bytes,
            "post-eviction budget invariant must hold: total={} budget={}",
            stats.total_resident_bytes,
            stats.memory_budget_bytes
        );
    }

    /// W-B1 test 2/4: an oversized handle (`bytes > budget`) is refused
    /// cleanly without touching ANY existing pooled entries.  Mirrors
    /// `LoadedPool::insert`'s pre-check semantics at the manager surface
    /// (the existing `hotswap_errors_when_no_evictable_fits` only
    /// exercises the empty-pool branch).
    #[test]
    fn hotswap_oversized_handle_preserves_existing_entries() {
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(3, 1_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f_small = synthetic_gguf(300);
        let f_oversized = synthetic_gguf(1_500); // > budget 1000
        let cfg = empty_config();

        // Pre-populate with two well-formed entries.
        let arc_a = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("a admit");
        let arc_b = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("b admit");
        assert_eq!(mgr.pool_stats().loaded_count, 2);
        let bytes_before = mgr.pool_stats().total_resident_bytes;

        // Attempt to load an oversized model.  Must refuse with
        // OversizedHandle WITHOUT evicting A or B (the pre-check at
        // `LoadedPool::insert` line 377 fires before the eviction
        // passes — no partial mutation).
        let err = mgr
            .load_or_get("huge/1", QuantType::Q4_K_M, f_oversized.path(), &cfg)
            .expect_err("oversized must refuse");
        match err {
            HotSwapError::PoolRefused(PoolError::OversizedHandle {
                repo_id,
                handle_bytes,
                budget_bytes,
            }) => {
                assert_eq!(repo_id, "huge/1@Q4_K_M");
                assert_eq!(handle_bytes, 1_500);
                assert_eq!(budget_bytes, 1_000);
            }
            other => panic!("expected OversizedHandle, got: {other:?}"),
        }

        // Pool state UNCHANGED — A and B both still resident with
        // identical Arc identities (cache hit on follow-up read).
        assert_eq!(
            mgr.pool_stats().loaded_count,
            2,
            "oversized refusal must not change loaded_count"
        );
        assert_eq!(
            mgr.pool_stats().total_resident_bytes,
            bytes_before,
            "oversized refusal must not change total_resident_bytes"
        );
        let arc_a_after = mgr
            .try_get("a/1", QuantType::Q4_K_M)
            .expect("A must remain resident");
        let arc_b_after = mgr
            .try_get("b/2", QuantType::Q4_K_M)
            .expect("B must remain resident");
        assert!(
            Arc::ptr_eq(&arc_a, &arc_a_after),
            "A's Arc identity must be preserved across refused oversized load"
        );
        assert!(
            Arc::ptr_eq(&arc_b, &arc_b_after),
            "B's Arc identity must be preserved across refused oversized load"
        );
        // Loader was invoked once for the oversized attempt (the
        // manager admits the file-stat + load before pool insert);
        // the freshly-produced engine drops on PoolRefused.
        // Total calls: 2 (a, b admits) + 1 (huge attempt) = 3.
        assert_eq!(loader.call_count(), 3);
    }

    /// W-B1 test 3/4: a `load_or_get` cache hit promotes the entry to
    /// MRU (the symmetric path to `try_get`'s "must NOT touch" — the
    /// request path SHOULD touch).  Without this promotion, a steady
    /// stream of requests against entry A would leave A as LRU and let
    /// it be evicted under cap pressure even though it is the actively-
    /// served model.
    #[test]
    fn hotswap_load_or_get_promotes_on_cache_hit() {
        let loader = Arc::new(MockLoader::new());
        // Capacity 2 so the third load forces exactly one eviction.
        let pool = LoadedPool::with_capacity_and_budget(2, 1_000_000);
        let mut mgr = HotSwapManager::<MockEngine>::new(pool, loader.clone());
        let f = synthetic_gguf(1_000);
        let cfg = empty_config();

        // Initial state: A is LRU, B is MRU.
        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a admit");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b admit");

        // Cache-hit `load_or_get` against A → A promotes to MRU, B
        // becomes LRU.  Loader call count must NOT advance (cache hit).
        let calls_before = loader.call_count();
        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a cache-hit");
        assert_eq!(
            loader.call_count(),
            calls_before,
            "cache-hit must not invoke the loader"
        );

        // Third distinct load: B (now LRU after A's promotion) evicts,
        // A survives.  This is the load-bearing assertion: without the
        // `pool.touch(&k)` call at HotSwapManager::load_or_get line 789,
        // A would be the LRU and would be the evictee here.
        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("c admit");

        assert!(
            mgr.try_get("a/1", QuantType::Q4_K_M).is_some(),
            "A must SURVIVE — cache-hit promoted it to MRU"
        );
        assert!(
            mgr.try_get("b/2", QuantType::Q4_K_M).is_none(),
            "B must EVICT — it became LRU after A's promotion"
        );
        assert!(
            mgr.try_get("c/3", QuantType::Q4_K_M).is_some(),
            "C must be resident (newly admitted)"
        );
    }

    /// W-B1 test 4/4: TOCTOU regression.  Concurrent `load_or_get`
    /// calls under the production-shape `Arc<Mutex<HotSwapManager<E>>>`
    /// wrapper (mirrors `AppState::pool` at `src/serve/api/state.rs:202`,
    /// which uses `Arc<RwLock<...>>` — Mutex is conceptually the same
    /// for the write-only path `load_or_get` exercises) must NOT race
    /// past the budget gate.  llama.cpp's reference at
    /// `server-models.cpp:545-558` re-checks capacity under the load-lock
    /// for the same reason; our defense is structural — a single lock
    /// guard wraps file-stat + loader invocation + pool insert.
    ///
    /// Invariants asserted post-join:
    ///   - exactly N loader invocations (no double-load + no missed load)
    ///   - `pool_stats().loaded_count == capacity_models`
    ///     (capacity strictly enforced under contention)
    ///   - `total_resident_bytes <= memory_budget_bytes` (budget never
    ///     exceeds — the post-condition the entire eviction-loop exists
    ///     to maintain)
    #[test]
    fn hotswap_concurrent_load_or_get_serializes_under_mutex() {
        use std::sync::Mutex;

        // Capacity = 3, budget large enough that the budget pass never
        // fires — we want the capacity LRU eviction to be the load-
        // bearing critical section under contention.
        const CAPACITY: usize = 3;
        const N_THREADS: usize = 8;
        let loader = Arc::new(MockLoader::new());
        let pool = LoadedPool::with_capacity_and_budget(CAPACITY, 1_000_000);
        let mgr = Arc::new(Mutex::new(HotSwapManager::<MockEngine>::new(
            pool,
            loader.clone(),
        )));

        // Pre-create per-thread synthetic GGUFs so each thread has a
        // distinct (repo, file) pair; tempfile lifetimes outlive the
        // join via the outer-scope Vec.
        let fixtures: Vec<tempfile::NamedTempFile> =
            (0..N_THREADS).map(|_| synthetic_gguf(500)).collect();
        let paths: Vec<PathBuf> = fixtures.iter().map(|f| f.path().to_path_buf()).collect();

        let cfg = empty_config();
        let cfg = Arc::new(cfg);

        // Spawn N threads, each loading a distinct repo.  They contend
        // on the Mutex-wrapped manager; the lock guarantees the
        // file-stat → loader invoke → pool insert sequence is atomic
        // per thread.
        let mut handles = Vec::with_capacity(N_THREADS);
        for i in 0..N_THREADS {
            let mgr = Arc::clone(&mgr);
            let path = paths[i].clone();
            let cfg = Arc::clone(&cfg);
            let repo = format!("repo/{i}");
            handles.push(std::thread::spawn(move || {
                let mut guard = mgr.lock().expect("lock manager");
                let _ = guard
                    .load_or_get(&repo, QuantType::Q4_K_M, &path, &cfg)
                    .expect("load_or_get");
                // Snapshot under the same lock the load happened under
                // — no observer can ever see the budget torn open.
                let stats = guard.pool_stats();
                assert!(
                    stats.total_resident_bytes <= stats.memory_budget_bytes,
                    "budget invariant violated under contention: \
                     total={} budget={} thread={}",
                    stats.total_resident_bytes,
                    stats.memory_budget_bytes,
                    i,
                );
                assert!(
                    stats.loaded_count <= stats.capacity_models,
                    "capacity invariant violated under contention: \
                     loaded={} capacity={} thread={}",
                    stats.loaded_count,
                    stats.capacity_models,
                    i,
                );
            }));
        }

        for h in handles {
            h.join().expect("thread join");
        }

        // Post-join invariants:
        // (a) exactly N loader invocations — every thread admitted its
        //     distinct repo cleanly, no duplicates, none lost to a race.
        assert_eq!(
            loader.call_count(),
            N_THREADS as u64,
            "expected exactly {N_THREADS} loader calls under Mutex serialization; \
             got {} (TOCTOU: duplicate load OR missed load)",
            loader.call_count()
        );

        let final_mgr = mgr.lock().expect("lock manager final");
        let stats = final_mgr.pool_stats();
        // (b) capacity strictly enforced — exactly capacity_models
        //     entries resident even though N > capacity threads ran.
        assert_eq!(
            stats.loaded_count, CAPACITY,
            "capacity must hold post-join: loaded={} capacity={}",
            stats.loaded_count, CAPACITY
        );
        // (c) budget invariant holds.
        assert!(
            stats.total_resident_bytes <= stats.memory_budget_bytes,
            "budget invariant violated post-join: total={} budget={}",
            stats.total_resident_bytes,
            stats.memory_budget_bytes
        );
        // The final 3 surviving entries are deterministically the
        // last-3 to acquire the lock, but threads have nondeterministic
        // scheduling — assert structurally that *some* 3-of-N survived.
        let survivors: usize = (0..N_THREADS)
            .filter(|i| {
                final_mgr
                    .try_get(&format!("repo/{i}"), QuantType::Q4_K_M)
                    .is_some()
            })
            .count();
        assert_eq!(
            survivors, CAPACITY,
            "exactly CAPACITY={CAPACITY} survivors expected post-join; got {survivors}"
        );
    }

    // ─────────────────────────────────────────────────────────────────────
    // ADR-005 Phase 4 reopen iter-212 (AC 5471) — KvSpiller trigger-site
    // unit tests.  Mirrors the iter-208 W76 test density for HotSwapManager
    // (14 in-binary unit tests under MockEngine + MockLoader) — these 9
    // additional tests assert the trigger-site invariants for the new
    // KvSpiller<E> hook surface.
    //
    // MockSpiller mirrors the MockEngine / MockLoader pattern at
    // multi_model.rs:1175 (now post-iter-212 ~line 1295): atomic call
    // counters + optional outcome injection per trigger so tests can
    // exercise both the happy path AND the Skipped / Error paths without
    // a real disk-backed spiller.  Captures Arc::strong_count at call
    // time so the in-flight-Arc invariant from
    // hotswap_pre_evict_fires_before_engine_drop has a load-bearing
    // assertion.
    // ─────────────────────────────────────────────────────────────────────

    /// Test fixture KvSpiller — mirrors MockEngine / MockLoader pattern.
    /// Tracks per-trigger call counts and the Arc::strong_count snapshot
    /// each call observed; tests can also inject a one-shot outcome
    /// override via `set_pre_evict_outcome` / `set_post_admit_outcome` to
    /// exercise the Skipped / Error paths.
    struct MockSpiller {
        pre_evict_calls: std::sync::atomic::AtomicU64,
        post_admit_calls: std::sync::atomic::AtomicU64,
        /// Each entry: (call_index, observed Arc::strong_count of
        /// engine at trigger-site time).  Captured under a Mutex so
        /// concurrent test paths (the multi-thread regression test)
        /// can observe consistent snapshots.
        pre_evict_strong_counts: std::sync::Mutex<Vec<(u64, usize)>>,
        post_admit_strong_counts: std::sync::Mutex<Vec<(u64, usize)>>,
        /// Optional outcome-injection slots.  When `Some(...)`, the
        /// trigger returns the injected outcome instead of the default
        /// `Skipped`.  Mirrors MockLoader::fail_on but per-trigger.
        pre_evict_outcome: std::sync::Mutex<Option<SpillOutcome>>,
        post_admit_outcome: std::sync::Mutex<Option<RestoreOutcome>>,
    }

    impl MockSpiller {
        fn new() -> Self {
            Self {
                pre_evict_calls: std::sync::atomic::AtomicU64::new(0),
                post_admit_calls: std::sync::atomic::AtomicU64::new(0),
                pre_evict_strong_counts: std::sync::Mutex::new(Vec::new()),
                post_admit_strong_counts: std::sync::Mutex::new(Vec::new()),
                pre_evict_outcome: std::sync::Mutex::new(None),
                post_admit_outcome: std::sync::Mutex::new(None),
            }
        }
        fn pre_evict_count(&self) -> u64 {
            self.pre_evict_calls
                .load(std::sync::atomic::Ordering::SeqCst)
        }
        fn post_admit_count(&self) -> u64 {
            self.post_admit_calls
                .load(std::sync::atomic::Ordering::SeqCst)
        }
        fn set_pre_evict_outcome(&self, outcome: SpillOutcome) {
            *self.pre_evict_outcome.lock().expect("lock") = Some(outcome);
        }
        fn set_post_admit_outcome(&self, outcome: RestoreOutcome) {
            *self.post_admit_outcome.lock().expect("lock") = Some(outcome);
        }
        fn pre_evict_strong_counts_snapshot(&self) -> Vec<(u64, usize)> {
            self.pre_evict_strong_counts.lock().expect("lock").clone()
        }
    }

    impl ModelLoader<MockEngine> for MockSpiller {
        // never used — only here to keep the test fixture compact.  The
        // real loader is MockLoader; MockSpiller only impls KvSpiller.
        fn load(&self, _path: &Path, _config: &EngineConfig) -> anyhow::Result<MockEngine> {
            unreachable!("MockSpiller is not a loader")
        }
    }

    impl KvSpiller<MockEngine> for MockSpiller {
        fn pre_evict(
            &self,
            _handle: &LoadedHandle,
            engine: &Arc<LoadedEngine<MockEngine>>,
        ) -> SpillOutcome {
            let n = self
                .pre_evict_calls
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
                + 1;
            // Capture Arc::strong_count BEFORE returning so the test
            // sees the live count at trigger-site time.
            let sc = Arc::strong_count(engine);
            self.pre_evict_strong_counts
                .lock()
                .expect("lock")
                .push((n, sc));
            self.pre_evict_outcome
                .lock()
                .expect("lock")
                .unwrap_or(SpillOutcome::Skipped)
        }
        fn post_admit(
            &self,
            _repo: &str,
            _quant: QuantType,
            engine: &Arc<LoadedEngine<MockEngine>>,
        ) -> RestoreOutcome {
            let n = self
                .post_admit_calls
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
                + 1;
            let sc = Arc::strong_count(engine);
            self.post_admit_strong_counts
                .lock()
                .expect("lock")
                .push((n, sc));
            self.post_admit_outcome
                .lock()
                .expect("lock")
                .unwrap_or(RestoreOutcome::Skipped)
        }
    }

    /// Helper: build a manager wired to a MockSpiller alongside the
    /// existing MockLoader scaffolding.  Returns (manager, loader,
    /// spiller) so tests can assert against any of them.
    fn mgr_with_spiller(
        capacity_models: usize,
        memory_budget_bytes: u64,
    ) -> (
        HotSwapManager<MockEngine>,
        Arc<MockLoader>,
        Arc<MockSpiller>,
    ) {
        let loader = Arc::new(MockLoader::new());
        let spiller = Arc::new(MockSpiller::new());
        let pool = LoadedPool::with_capacity_and_budget(capacity_models, memory_budget_bytes);
        let mgr =
            HotSwapManager::<MockEngine>::new_with_spiller(pool, loader.clone(), spiller.clone());
        (mgr, loader, spiller)
    }

    /// 1/9: capacity-2 pool; admit 3 distinct repos; the third load
    /// evicts the LRU.  Asserts MockSpiller::pre_evict_calls == 1
    /// (single LRU evict; no double-fire).
    #[test]
    fn hotswap_pre_evict_fires_on_lru_eviction() {
        let (mut mgr, _loader, spiller) = mgr_with_spiller(2, 1_000_000);
        let f = synthetic_gguf(1_000);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b");
        // a/1 and b/2 admit clean — no eviction yet.
        assert_eq!(
            spiller.pre_evict_count(),
            0,
            "no eviction at capacity boundary; pre_evict must NOT fire"
        );

        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("c");
        // c/3 evicts a/1 (LRU) → exactly one pre_evict call.
        assert_eq!(
            spiller.pre_evict_count(),
            1,
            "exactly one pre_evict call expected on single LRU eviction"
        );
        assert!(mgr.try_get("a/1", QuantType::Q4_K_M).is_none());
    }

    /// 2/9: explicit evict() — `mgr.evict(repo, quant)` should also
    /// fire pre_evict.  Symmetric to the LRU-eviction trigger.
    #[test]
    fn hotswap_pre_evict_fires_on_explicit_evict() {
        let (mut mgr, _loader, spiller) = mgr_with_spiller(3, 100_000);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("admit");
        // pre_evict has not fired yet — only post_admit on cold load.
        assert_eq!(spiller.pre_evict_count(), 0);
        assert_eq!(spiller.post_admit_count(), 1);

        let bytes_freed = mgr.evict("acme/m1", QuantType::Q4_K_M);
        assert_eq!(bytes_freed, 500);
        assert_eq!(
            spiller.pre_evict_count(),
            1,
            "explicit evict() must fire pre_evict exactly once"
        );

        // Idempotent evict — second call is a no-op AND must NOT fire
        // pre_evict (no live entry to spill).
        let bytes_freed_2 = mgr.evict("acme/m1", QuantType::Q4_K_M);
        assert_eq!(bytes_freed_2, 0);
        assert_eq!(
            spiller.pre_evict_count(),
            1,
            "idempotent evict() must NOT double-fire pre_evict"
        );
    }

    /// 3/9: post_admit fires on cold load AFTER the loader returns.
    /// Asserts call sequencing: post_admit_calls == 1 AFTER
    /// loader.call_count == 1.
    #[test]
    fn hotswap_post_admit_fires_on_cold_load() {
        let (mut mgr, loader, spiller) = mgr_with_spiller(3, 100_000);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        // Pre-load: loader has not been called; spiller untouched.
        assert_eq!(loader.call_count(), 0);
        assert_eq!(spiller.post_admit_count(), 0);

        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("admit");

        // Sequencing invariant: loader fires first; post_admit second.
        // Both should have fired exactly once.
        assert_eq!(
            loader.call_count(),
            1,
            "loader must have run before post_admit — sequencing"
        );
        assert_eq!(
            spiller.post_admit_count(),
            1,
            "post_admit must fire exactly once on cold load"
        );
    }

    /// 4/9: post_admit must NOT fire on a cache hit.  The fast path in
    /// load_or_get short-circuits before any admission work; the
    /// spiller MUST NOT observe a phantom admission.
    #[test]
    fn hotswap_post_admit_does_not_fire_on_cache_hit() {
        let (mut mgr, loader, spiller) = mgr_with_spiller(3, 100_000);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("first");
        assert_eq!(spiller.post_admit_count(), 1);
        assert_eq!(loader.call_count(), 1);

        // Cache-hit: same (repo, quant) — fast path, no loader.load.
        let _ = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("cache-hit");
        assert_eq!(
            loader.call_count(),
            1,
            "cache-hit must not invoke the loader"
        );
        assert_eq!(
            spiller.post_admit_count(),
            1,
            "cache-hit must NOT fire post_admit a second time"
        );
    }

    /// 5/9: Arc::strong_count invariant.  The pre_evict trigger must
    /// observe a live Arc (strong_count >= 2: manager's clone +
    /// pre_evict's borrow).  After pre_evict returns, the manager's
    /// Arc drops from `engines` and (assuming no other holders) the
    /// engine drops normally.  This test re-verifies iter-208's
    /// `hotswap_in_flight_arc_survives_eviction` invariant holds
    /// post-trigger-insertion.
    #[test]
    fn hotswap_pre_evict_fires_before_engine_drop() {
        // Capacity 1 → second load forces eviction.
        let (mut mgr, _loader, spiller) = mgr_with_spiller(1, 100_000);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        // Hold an in-flight Arc clone of the first load — mirrors the
        // iter-208 invariant test.
        let inflight = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a");
        // strong_count: 1 (manager) + 1 (inflight) = 2.
        assert_eq!(Arc::strong_count(&inflight), 2);

        // Second load evicts a/1 — pre_evict fires before the manager
        // drops its Arc from `engines`.
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b");

        // pre_evict observed the live Arc.  At trigger-site time the
        // manager's Arc was still in the `engines` map AND the
        // pre_evict signature borrows the Arc (`&Arc<LoadedEngine<E>>`)
        // — so strong_count at call-time is at least:
        //   1 (manager's `engines` map slot)
        //   + 1 (inflight clone held by this test)
        //   = 2.
        // The `engines.get(...).cloned()` call site additionally clones
        // briefly into a local before passing `&arc` — so the snapshot
        // observed by the spiller may be 3.  Either is correct per
        // ADR-005 line 666-674: the load-bearing assertion is `>= 2`.
        let snapshots = spiller.pre_evict_strong_counts_snapshot();
        assert_eq!(snapshots.len(), 1, "exactly one pre_evict call");
        let (call_n, observed_count) = snapshots[0];
        assert_eq!(call_n, 1);
        assert!(
            observed_count >= 2,
            "pre_evict must observe a live Arc (strong_count >= 2); observed = {observed_count}"
        );

        // Post-eviction invariant: manager's Arc dropped → strong_count
        // drops to 1 (just inflight).  This mirrors iter-208's
        // `hotswap_in_flight_arc_survives_eviction` post-condition.
        assert_eq!(
            Arc::strong_count(&inflight),
            1,
            "manager must have released its Arc on eviction; \
             pre_evict trigger must NOT extend the Arc lifetime"
        );
        // Inflight engine still valid — pre_evict did not interfere.
        assert_eq!(inflight.repo, "a/1");
        assert_eq!(inflight.engine.load_serial, 1);
        // Engines map post-eviction: a/1 absent (the post-pre_evict
        // remove() ran), b/2 present.
        assert!(mgr.try_get("a/1", QuantType::Q4_K_M).is_none());
        assert!(mgr.try_get("b/2", QuantType::Q4_K_M).is_some());
    }

    /// 6/9: a single load that chains MULTIPLE evictions must fire
    /// pre_evict per evictee.  Mirrors the existing
    /// `hotswap_chains_multiple_evictions_in_one_load` shape but
    /// asserts the spiller call count.
    #[test]
    fn hotswap_chained_evictions_fire_pre_evict_per_evictee() {
        let (mut mgr, _loader, spiller) = mgr_with_spiller(3, 1_500);
        let f_small = synthetic_gguf(500);
        let f_big = synthetic_gguf(1_000);
        let cfg = empty_config();

        // Three 500-byte loads — pool fills exactly at budget.
        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("a");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("b");
        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f_small.path(), &cfg)
            .expect("c");
        assert_eq!(spiller.pre_evict_count(), 0);
        assert_eq!(spiller.post_admit_count(), 3);

        // 1000-byte load forces TWO evictions in one call (a/1 then b/2).
        let _ = mgr
            .load_or_get("d/4", QuantType::Q4_K_M, f_big.path(), &cfg)
            .expect("d");

        assert_eq!(
            spiller.pre_evict_count(),
            2,
            "chained eviction must fire pre_evict once PER evictee"
        );
        assert_eq!(
            spiller.post_admit_count(),
            4,
            "post_admit fires once for the new admission (d/4)"
        );
    }

    /// 7/9: Skipped outcome — pre_evict returning `Skipped` must NOT
    /// block the eviction.  The spill is best-effort; eviction
    /// completes regardless.
    #[test]
    fn hotswap_pre_evict_skipped_outcome_does_not_block_eviction() {
        let (mut mgr, _loader, spiller) = mgr_with_spiller(2, 1_000_000);
        let f = synthetic_gguf(1_000);
        let cfg = empty_config();

        // Inject Skipped explicitly (default is also Skipped, but this
        // exercises the override path explicitly).
        spiller.set_pre_evict_outcome(SpillOutcome::Skipped);

        let _ = mgr
            .load_or_get("a/1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("a");
        let _ = mgr
            .load_or_get("b/2", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("b");
        let _ = mgr
            .load_or_get("c/3", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("c");

        // Eviction proceeded normally despite Skipped outcome.
        assert!(mgr.try_get("a/1", QuantType::Q4_K_M).is_none());
        assert!(mgr.try_get("b/2", QuantType::Q4_K_M).is_some());
        assert!(mgr.try_get("c/3", QuantType::Q4_K_M).is_some());
        let stats = mgr.pool_stats();
        assert_eq!(
            stats.loaded_count, 2,
            "eviction must complete normally regardless of pre_evict outcome"
        );
        assert_eq!(spiller.pre_evict_count(), 1);
    }

    /// 8/9: Error outcome on post_admit must NOT block admission.  The
    /// restore is best-effort; admission proceeds regardless and the
    /// engine becomes resident in the pool.  iter-213 surfaces the
    /// error to a /metrics counter.
    #[test]
    fn hotswap_post_admit_error_does_not_block_admission() {
        let (mut mgr, loader, spiller) = mgr_with_spiller(3, 100_000);
        let f = synthetic_gguf(500);
        let cfg = empty_config();

        // Inject a parity-fail error outcome.
        spiller.set_post_admit_outcome(RestoreOutcome::Error(RestoreErrorKind::ParityFail));

        let arc = mgr
            .load_or_get("acme/m1", QuantType::Q4_K_M, f.path(), &cfg)
            .expect("admission must succeed even when post_admit errors");
        assert_eq!(arc.repo, "acme/m1");
        assert_eq!(arc.engine.load_serial, 1);

        // Loader was invoked, post_admit fired, engine is in the pool.
        assert_eq!(loader.call_count(), 1);
        assert_eq!(spiller.post_admit_count(), 1);
        let stats = mgr.pool_stats();
        assert_eq!(
            stats.loaded_count, 1,
            "engine must be admitted to the pool despite post_admit error"
        );
        assert!(
            mgr.try_get("acme/m1", QuantType::Q4_K_M).is_some(),
            "post-error admission must publish the engine to the engines map"
        );
    }

    /// 9/9: TOCTOU regression.  Extends iter-210 W-B1
    /// `hotswap_concurrent_load_or_get_serializes_under_mutex` with the
    /// spiller wired in; asserts per-thread pre_evict + post_admit
    /// counts are consistent with the eviction sequence (no double-fire
    /// under contention).
    #[test]
    fn hotswap_concurrent_load_with_spiller_serializes_under_mutex() {
        use std::sync::Mutex;

        const CAPACITY: usize = 3;
        const N_THREADS: usize = 8;
        let loader = Arc::new(MockLoader::new());
        let spiller = Arc::new(MockSpiller::new());
        let pool = LoadedPool::with_capacity_and_budget(CAPACITY, 1_000_000);
        let mgr = Arc::new(Mutex::new(HotSwapManager::<MockEngine>::new_with_spiller(
            pool,
            loader.clone(),
            spiller.clone(),
        )));

        let fixtures: Vec<tempfile::NamedTempFile> =
            (0..N_THREADS).map(|_| synthetic_gguf(500)).collect();
        let paths: Vec<PathBuf> = fixtures.iter().map(|f| f.path().to_path_buf()).collect();
        let cfg = Arc::new(empty_config());

        let mut handles = Vec::with_capacity(N_THREADS);
        for i in 0..N_THREADS {
            let mgr = Arc::clone(&mgr);
            let path = paths[i].clone();
            let cfg = Arc::clone(&cfg);
            let repo = format!("repo/{i}");
            handles.push(std::thread::spawn(move || {
                let mut guard = mgr.lock().expect("lock manager");
                let _ = guard
                    .load_or_get(&repo, QuantType::Q4_K_M, &path, &cfg)
                    .expect("load_or_get");
            }));
        }
        for h in handles {
            h.join().expect("thread join");
        }

        // (a) Loader called exactly N times — no double-load + no
        // missed load.  This is the pre-existing TOCTOU invariant.
        assert_eq!(
            loader.call_count(),
            N_THREADS as u64,
            "expected exactly {N_THREADS} loader calls under Mutex serialization"
        );

        // (b) post_admit fires exactly N times — once per cold load.
        assert_eq!(
            spiller.post_admit_count(),
            N_THREADS as u64,
            "post_admit must fire exactly once per cold load under contention"
        );

        // (c) pre_evict count + accounting consistency.  N_THREADS = 8
        // distinct repos against capacity = 3 → exactly N - CAPACITY
        // total evictions over the run (each load past capacity evicts
        // one).  pre_evict fires once per evictee.
        let expected_evictions = (N_THREADS - CAPACITY) as u64;
        assert_eq!(
            spiller.pre_evict_count(),
            expected_evictions,
            "pre_evict must fire exactly {expected_evictions} times \
             (N - CAPACITY = {N_THREADS} - {CAPACITY}) under contention; \
             observed = {}",
            spiller.pre_evict_count()
        );

        // (d) per-call no-double-fire: every captured call_index in the
        // strong-count snapshot is unique (the AtomicU64 fetch_add gave
        // each call a distinct sequence number; if any pair shared an
        // index, fetch_add was bypassed somewhere).
        let pre_evict_snapshots = spiller.pre_evict_strong_counts_snapshot();
        let mut seen = std::collections::HashSet::new();
        for (idx, _) in &pre_evict_snapshots {
            assert!(
                seen.insert(*idx),
                "pre_evict call index {idx} repeated — double-fire under contention"
            );
        }

        // (e) final state — capacity strictly enforced.
        let final_mgr = mgr.lock().expect("lock manager final");
        let stats = final_mgr.pool_stats();
        assert_eq!(stats.loaded_count, CAPACITY);
        assert!(stats.total_resident_bytes <= stats.memory_budget_bytes);
    }
}