car-inference 0.50.0

//! LRU-evicting cache for loaded inference backends.
//!
//! Solves three problems at once:
//!
//! 1. **Cold start on every call.** Before: `FluxBackend::load()` /
//!    `LtxBackend::load()` / `KokoroBackend::load()` ran for every
//!    `generate_image` / `generate_video` / `synth_tts` request, paying
//!    the 1–14 s model-load cost on a hot path. After: first call loads,
//!    subsequent calls get a cheap `Arc<Mutex<T>>` handle.
//!
//! 2. **Concurrent calls racing on the same backend.** `mlx-rs` `Array`
//!    is `!Sync`; two tokio tasks calling the same backend simultaneously
//!    is undefined behavior. The per-entry `Mutex` serializes concurrent
//!    callers onto the same backend. Different backends still run in
//!    parallel (MLX itself queues them on the single Metal driver).
//!
//! 3. **Unbounded RAM growth.** Before: loading Flux + LTX + Kokoro +
//!    every text-gen model held ~30 GB of quantized weights forever. The
//!    cache tracks an approximate per-entry size (sum of the model
//!    directory's `.safetensors` bytes) and evicts LRU entries once the
//!    [`SharedModelBudget`] total exceeds budget. The budget is **shared**
//!    across the engine's caches (one aggregate cap, not N) and defaults to a
//!    **RAM-derived** figure (~60% of usable RAM), overridable by
//!    `CAR_INFERENCE_MODEL_CACHE_MB`. Set to 0 to disable caching. (#427)
//!
//! 4. **Resident memory that never releases at idle.** Capacity eviction
//!    only fires when the total *exceeds* the budget — so a daemon that
//!    loads one 2 GB model and then goes quiet pins that 2 GB resident
//!    forever (it never crosses the configured cap). [`evict_idle`] sweeps out
//!    entries untouched for longer than the idle TTL, so a quiet daemon
//!    drops its model working set back down. TTL via
//!    `CAR_INFERENCE_MODEL_IDLE_SECS` (default 300; 0 disables idle
//!    eviction). The caller drives the sweep on a timer — the cache is
//!    otherwise passive. (car-releases#67)
//!
//! The cache is generic over `T: Send + 'static`. Inference backends
//! don't need to implement any trait — just wrap them on insert.
//!
//! Invariant: an evicted entry is only removed from the cache map; any
//! outstanding `Arc<Mutex<T>>` handle continues to work until the last
//! caller drops it. This makes eviction safe even during a long-running
//! inference call — RAM is reclaimed lazily when the last user finishes.

use std::collections::{HashMap, VecDeque};
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{mpsc, Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

/// Handle to a cached backend. Callers lock the inner mutex for the
/// duration of an inference call to serialize with concurrent requests
/// for the same model.
pub type CachedBackend<T> = Arc<Mutex<T>>;

struct FinalHandleReap {
    released: Box<dyn Fn() -> bool + Send>,
    cleanup: Box<dyn Fn() + Send>,
}

struct ReaperBackoff {
    initial: Duration,
    current: Duration,
    maximum: Duration,
}

impl ReaperBackoff {
    fn new(initial: Duration, maximum: Duration) -> Self {
        Self {
            initial,
            current: initial,
            maximum,
        }
    }

    fn current(&self) -> Duration {
        self.current
    }

    fn progress(&mut self) {
        self.current = self.initial;
    }

    fn no_progress(&mut self) {
        self.current = self.current.saturating_mul(2).min(self.maximum);
    }
}

fn final_handle_reaper() -> &'static mpsc::Sender<FinalHandleReap> {
    static SENDER: OnceLock<mpsc::Sender<FinalHandleReap>> = OnceLock::new();
    SENDER.get_or_init(|| {
        let (sender, receiver) = mpsc::channel::<FinalHandleReap>();
        let _ = std::thread::Builder::new()
            .name("car-backend-handle-reaper".into())
            .spawn(move || {
                let mut pending = Vec::<FinalHandleReap>::new();
                let mut backoff =
                    ReaperBackoff::new(Duration::from_millis(10), Duration::from_millis(250));
                loop {
                    if pending.is_empty() {
                        match receiver.recv() {
                            Ok(item) => pending.push(item),
                            Err(_) => break,
                        }
                    } else {
                        match receiver.recv_timeout(backoff.current()) {
                            Ok(item) => pending.push(item),
                            Err(mpsc::RecvTimeoutError::Timeout) => {}
                            Err(mpsc::RecvTimeoutError::Disconnected) => {}
                        }
                    }
                    while let Ok(item) = receiver.try_recv() {
                        pending.push(item);
                    }
                    let before = pending.len();
                    let mut index = 0;
                    while index < pending.len() {
                        if (pending[index].released)() {
                            let item = pending.swap_remove(index);
                            (item.cleanup)();
                        } else {
                            index += 1;
                        }
                    }
                    if pending.len() < before {
                        backoff.progress();
                    } else {
                        backoff.no_progress();
                    }
                }
            });
        sender
    })
}

/// Whether the loader published weights into a cache/process that survives the
/// current request. Callers must never infer residency merely from successful
/// inference: a zero-budget cache returns a transient backend handle.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BackendRetention {
    Resident,
    Transient,
}

/// A model-cache memory budget, shared across all backend caches so the
/// *aggregate* resident model working set (text + image + video + TTS) is bounded
/// by one figure instead of N independent per-cache caps. Each cache accounts its
/// loads against this shared total and, when the shared total exceeds the budget,
/// evicts its OWN LRU entries.
///
/// Limitation: eviction is per-cache (a cache can only evict its own entries, as
/// the caches are generic over different backend types). So a single cache's
/// growth is bounded strictly, and the cross-modality aggregate is bounded in
/// steady state (every active cache self-trims on insert; idle eviction reclaims
/// inactive ones). A strict per-instant global-LRU coordinator that can evict
/// across caches is a documented follow-up.
pub struct SharedModelBudget {
    budget_bytes: AtomicU64,
    total_bytes: AtomicU64,
}

impl SharedModelBudget {
    /// A budget of `budget_bytes` (0 disables caching).
    pub fn new(budget_bytes: u64) -> Arc<Self> {
        Arc::new(Self {
            budget_bytes: AtomicU64::new(budget_bytes),
            total_bytes: AtomicU64::new(0),
        })
    }

    /// `CAR_INFERENCE_MODEL_CACHE_MB` if set, else `default_mb` (the caller's
    /// RAM-derived figure). The env override always wins.
    pub fn from_env_or(default_mb: u64) -> Arc<Self> {
        let mb = std::env::var("CAR_INFERENCE_MODEL_CACHE_MB")
            .ok()
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(default_mb);
        Self::new(mb.saturating_mul(1024 * 1024))
    }

    fn add(&self, n: u64) {
        self.total_bytes.fetch_add(n, Ordering::Relaxed);
    }
    fn sub(&self, n: u64) {
        // Saturating: every add is matched by a sub, but guard against a stray
        // double-sub wrapping the counter.
        let _ = self
            .total_bytes
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
                Some(cur.saturating_sub(n))
            });
    }
    fn total(&self) -> u64 {
        self.total_bytes.load(Ordering::Relaxed)
    }
    /// Over budget? Always false for a disabled (0) budget.
    fn over_budget(&self) -> bool {
        let budget = self.budget_bytes();
        budget != 0 && self.total() > budget
    }
    /// A disabled (zero-budget) budget — caching off, every call loads fresh.
    pub fn is_disabled(&self) -> bool {
        self.budget_bytes() == 0
    }

    /// Update the aggregate cache ceiling without replacing the shared budget
    /// object held by each modality cache.
    pub fn set_budget_bytes(&self, budget_bytes: u64) {
        self.budget_bytes.store(budget_bytes, Ordering::Release);
    }

    pub fn budget_bytes(&self) -> u64 {
        self.budget_bytes.load(Ordering::Acquire)
    }
}

/// RAM-derived default cache budget (MB): ~60% of usable RAM, from `HardwareInfo`
/// (the same figure the model recommender uses). Replaces the old flat 24 GB so
/// the cache scales with the machine instead of pinning a fixed ceiling.
pub fn default_model_cache_mb() -> u64 {
    crate::hardware::HardwareInfo::detect().max_model_mb
}

/// Idle-eviction TTL from `CAR_INFERENCE_MODEL_IDLE_SECS` (default 300; 0 = off).
pub fn idle_ttl_from_env() -> Option<Duration> {
    let idle_secs = std::env::var("CAR_INFERENCE_MODEL_IDLE_SECS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(300);
    (idle_secs > 0).then(|| Duration::from_secs(idle_secs))
}

struct Entry<T> {
    backend: CachedBackend<T>,
    allocation_id: String,
    size_bytes: u64,
    /// Last time this entry was looked up or inserted. Drives idle
    /// (time-based) eviction, distinct from the LRU capacity eviction.
    last_used: Instant,
    /// A failed backend may be invalidated while its current request still
    /// owns a handle. It remains accounted until that request releases it.
    invalidated: bool,
}

struct Inner<T> {
    /// Loaded backends keyed by stable ID (typically `ModelSchema.id`).
    map: HashMap<String, Entry<T>>,
    /// LRU recency list — back is most recent, front is oldest.
    lru: VecDeque<String>,
}

/// LRU-bounded cache of loaded inference backends.
///
/// Not itself `Sync` over `T` — the cache stores handles only, and the
/// lock guarding the map is coarse (held only for look-up / insert /
/// evict). Per-entry mutation is serialized by each entry's own `Mutex`.
pub struct BackendCache<T: Send + 'static> {
    inner: Mutex<Inner<T>>,
    /// Shared budget governing this cache's evictions (may be shared with other
    /// backend caches so the aggregate working set is bounded by one figure).
    budget: Arc<SharedModelBudget>,
    /// Idle eviction window. `None` disables time-based eviction (the
    /// cache then only evicts on capacity pressure). See [`evict_idle`].
    idle_ttl: Option<Duration>,
    /// Per-key singleflight gates. A cold load for one model does not block a
    /// different model, while duplicate requests share the first published
    /// backend instead of allocating the same weights twice.
    load_locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
    resident_accounting: Option<Arc<crate::resource_policy::LocalAdmissionCoordinator>>,
    allocation_scope: u64,
}

fn next_cache_allocation_scope() -> u64 {
    static NEXT: AtomicU64 = AtomicU64::new(1);
    NEXT.fetch_add(1, Ordering::Relaxed)
}

impl<T: Send + 'static> Drop for BackendCache<T> {
    fn drop(&mut self) {
        let inner = self
            .inner
            .get_mut()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Only acknowledge allocations whose cache-owned Arc is the final
        // backend handle. An inference call can outlive its engine/cache; in
        // that case the weights remain resident and must stay conservatively
        // accounted rather than being advertised as freed too early.
        let idle = inner
            .map
            .iter()
            .filter(|(_, entry)| Arc::strong_count(&entry.backend) == 1)
            .map(|(_, entry)| (entry.allocation_id.clone(), entry.size_bytes))
            .collect::<Vec<_>>();
        let active = inner
            .map
            .iter()
            .filter(|(_, entry)| Arc::strong_count(&entry.backend) > 1)
            .map(|(_, entry)| {
                (
                    entry.allocation_id.clone(),
                    entry.size_bytes,
                    Arc::downgrade(&entry.backend),
                )
            })
            .collect::<Vec<_>>();
        let bytes = idle
            .iter()
            .map(|(_, size_bytes)| *size_bytes)
            .fold(0_u64, u64::saturating_add);
        inner.lru.clear();
        self.budget.sub(bytes);
        if let Some(accounting) = &self.resident_accounting {
            for (key, _) in idle {
                accounting.mark_evicted(&key);
            }
        }

        // An inference handle may outlive the cache/engine. Transfer all such
        // leases to one shared reaper instead of creating an unbounded OS
        // thread per active entry.
        for (key, size_bytes, backend) in active {
            let accounting = self.resident_accounting.clone();
            let budget = self.budget.clone();
            let item = FinalHandleReap {
                released: Box::new(move || backend.strong_count() == 0),
                cleanup: Box::new(move || {
                    budget.sub(size_bytes);
                    if let Some(accounting) = &accounting {
                        accounting.mark_evicted(&key);
                    }
                }),
            };
            if let Err(error) = final_handle_reaper().send(item) {
                // Fail closed if the singleton reaper ever exits.
                std::mem::forget(error.0);
            }
        }
        inner.map.clear();
    }
}

impl<T: Send + 'static> BackendCache<T> {
    /// Create a cache with its OWN budget of `budget_bytes` and no idle eviction.
    /// For a budget shared across caches use [`from_shared`](Self::from_shared).
    pub fn new(budget_bytes: u64) -> Self {
        Self::from_shared(SharedModelBudget::new(budget_bytes), None)
    }

    /// Same as [`new`](Self::new) but also arms idle (time-based) eviction.
    pub fn with_idle_ttl(budget_bytes: u64, idle_ttl: Option<Duration>) -> Self {
        Self::from_shared(SharedModelBudget::new(budget_bytes), idle_ttl)
    }

    /// Create a cache bound to a (possibly shared) [`SharedModelBudget`], so
    /// several caches can enforce one aggregate budget.
    pub fn from_shared(budget: Arc<SharedModelBudget>, idle_ttl: Option<Duration>) -> Self {
        Self::from_shared_with_admission(budget, idle_ttl, None)
    }

    pub fn from_shared_with_admission(
        budget: Arc<SharedModelBudget>,
        idle_ttl: Option<Duration>,
        resident_accounting: Option<Arc<crate::resource_policy::LocalAdmissionCoordinator>>,
    ) -> Self {
        Self {
            inner: Mutex::new(Inner {
                map: HashMap::new(),
                lru: VecDeque::new(),
            }),
            budget,
            idle_ttl,
            load_locks: Mutex::new(HashMap::new()),
            resident_accounting,
            allocation_scope: next_cache_allocation_scope(),
        }
    }

    /// A standalone cache configured from the environment:
    /// - `CAR_INFERENCE_MODEL_CACHE_MB` — capacity budget; default is now
    ///   **RAM-derived** (~60% of usable RAM via `HardwareInfo`), not a flat
    ///   24 GB. 0 disables caching.
    /// - `CAR_INFERENCE_MODEL_IDLE_SECS` — idle eviction TTL, default 300; 0 off.
    ///
    /// NOTE: this builds its OWN budget. To share one budget across the engine's
    /// caches, construct a [`SharedModelBudget`] once and pass it to
    /// [`from_shared`](Self::from_shared).
    pub fn from_env() -> Self {
        Self::from_shared(
            SharedModelBudget::from_env_or(default_model_cache_mb()),
            idle_ttl_from_env(),
        )
    }

    /// Is this a disabled (zero-budget) cache?
    pub fn is_disabled(&self) -> bool {
        self.budget.is_disabled()
    }

    /// Get a handle to the cached backend, or load + insert it.
    ///
    /// `loader` is invoked only on a cache miss. If the new entry plus
    /// existing entries exceed the budget, the oldest entries are
    /// evicted until we're within budget (or only the new entry remains).
    ///
    /// `size_bytes` should be an approximate on-disk or in-memory size
    /// for the loaded backend; [`estimate_model_size`] is a reasonable
    /// default that sums the `.safetensors` file sizes in a model dir.
    pub fn get_or_load<E>(
        &self,
        key: &str,
        size_bytes: u64,
        loader: impl FnOnce() -> Result<T, E>,
    ) -> Result<CachedBackend<T>, E> {
        self.get_or_load_with_publish(key, size_bytes, loader, |allocation_id| {
            if let Some(accounting) = &self.resident_accounting {
                accounting.mark_resident_allocation(
                    key,
                    allocation_id,
                    size_bytes.div_ceil(1024 * 1024),
                );
            }
            Ok(())
        })
        .map(|(handle, _retained)| handle)
    }

    /// Load through the cache while tying measured-size admission to the exact
    /// publication point. Cache hits transfer the caller's redundant cold
    /// reservation back to the already-resident allocation.
    pub fn get_or_load_admitted(
        &self,
        key: &str,
        size_bytes: u64,
        reservation: &mut crate::resource_policy::LocalLoadReservation,
        loader: impl FnOnce() -> Result<T, crate::InferenceError>,
    ) -> Result<(CachedBackend<T>, BackendRetention), crate::InferenceError> {
        let allocation_id = format!("cache:{}:{key}", self.allocation_scope);
        reservation.bind_allocation_id(&allocation_id);
        reservation
            .reconcile_measured_weights(size_bytes)
            .map_err(crate::InferenceError::from)?;
        let (handle, retained) =
            self.get_or_load_with_publish(key, size_bytes, loader, |allocation_id| {
                reservation.publish_resident_weights_as(allocation_id, size_bytes);
                Ok(())
            })?;
        if retained {
            reservation.publish_resident_weights(size_bytes);
        }
        Ok((
            handle,
            if retained {
                BackendRetention::Resident
            } else {
                BackendRetention::Transient
            },
        ))
    }

    fn get_or_load_with_publish<E>(
        &self,
        key: &str,
        size_bytes: u64,
        loader: impl FnOnce() -> Result<T, E>,
        on_publish: impl FnOnce(&str) -> Result<(), E>,
    ) -> Result<(CachedBackend<T>, bool), E> {
        // Fast path: already cached.
        {
            let mut guard = self.inner.lock().expect("backend cache poisoned");
            let stale_is_idle = guard
                .map
                .get(key)
                .is_some_and(|entry| entry.invalidated && Arc::strong_count(&entry.backend) == 1);
            if stale_is_idle {
                let stale = guard.map.remove(key).expect("entry checked above");
                guard.lru.retain(|candidate| candidate != key);
                self.budget.sub(stale.size_bytes);
                if let Some(accounting) = &self.resident_accounting {
                    accounting.mark_evicted(&stale.allocation_id);
                }
            }
            if let Some(entry) = guard.map.get_mut(key) {
                let handle = Arc::clone(&entry.backend);
                // Refresh recency for both LRU (capacity) and idle (time)
                // eviction so a model in active use is never swept.
                entry.last_used = Instant::now();
                guard.lru.retain(|k| k != key);
                guard.lru.push_back(key.to_string());
                return Ok((handle, true));
            }
        }

        let load_gate = {
            let mut gates = self.load_locks.lock().expect("backend load gates poisoned");
            Arc::clone(
                gates
                    .entry(key.to_string())
                    .or_insert_with(|| Arc::new(Mutex::new(()))),
            )
        };
        let _singleflight = load_gate.lock().expect("backend load gate poisoned");

        // A same-key caller may have published while this caller waited.
        {
            let mut guard = self.inner.lock().expect("backend cache poisoned");
            if let Some(entry) = guard.map.get_mut(key) {
                let handle = Arc::clone(&entry.backend);
                entry.last_used = Instant::now();
                guard.lru.retain(|candidate| candidate != key);
                guard.lru.push_back(key.to_string());
                drop(_singleflight);
                self.release_load_gate(key, &load_gate);
                return Ok((handle, true));
            }
        }

        // Different keys still load concurrently.
        let backend = match loader() {
            Ok(backend) => backend,
            Err(error) => {
                drop(_singleflight);
                self.release_load_gate(key, &load_gate);
                return Err(error);
            }
        };
        let handle = Arc::new(Mutex::new(backend));

        if self.budget.is_disabled() {
            // Caching disabled: return the handle without retaining it.
            drop(_singleflight);
            self.release_load_gate(key, &load_gate);
            return Ok((handle, false));
        }

        let mut guard = self.inner.lock().expect("backend cache poisoned");

        // Re-check in case someone else inserted while we were loading.
        if let Some(existing) = guard.map.get(key) {
            let existing = Arc::clone(&existing.backend);
            drop(guard);
            drop(_singleflight);
            self.release_load_gate(key, &load_gate);
            return Ok((existing, true));
        }

        let allocation_id = format!("cache:{}:{key}", self.allocation_scope);
        if let Err(error) = on_publish(&allocation_id) {
            drop(guard);
            drop(_singleflight);
            self.release_load_gate(key, &load_gate);
            return Err(error);
        }
        self.budget.add(size_bytes);
        guard.map.insert(
            key.to_string(),
            Entry {
                backend: Arc::clone(&handle),
                allocation_id,
                size_bytes,
                last_used: Instant::now(),
                invalidated: false,
            },
        );
        guard.lru.push_back(key.to_string());

        // Evict this cache's LRU entries until the SHARED total is within budget
        // (but never evict the just-inserted key — it's the newest, so it would
        // only be reached once everything else is gone, which defeats the load).
        let mut examined = guard.lru.len();
        while self.budget.over_budget() && examined > 0 {
            examined -= 1;
            let Some(victim_key) = guard.lru.pop_front() else {
                break;
            };
            if victim_key == key {
                // Only our own (just-inserted) entry is left to evict; stop. The
                // shared total may still exceed budget due to OTHER caches'
                // entries — they self-trim on their next insert, and idle
                // eviction reclaims inactive ones.
                guard.lru.push_front(victim_key);
                break;
            }
            let victim_is_idle = guard
                .map
                .get(&victim_key)
                .is_some_and(|entry| Arc::strong_count(&entry.backend) == 1);
            if !victim_is_idle {
                guard.lru.push_back(victim_key);
                continue;
            }
            if let Some(victim) = guard.map.remove(&victim_key) {
                self.budget.sub(victim.size_bytes);
                if let Some(accounting) = &self.resident_accounting {
                    accounting.mark_evicted(&victim.allocation_id);
                }
                // The `Arc` may still have outstanding handles; they
                // continue to work. Memory is reclaimed when the last
                // one drops.
                drop(victim);
            }
        }

        drop(guard);
        drop(_singleflight);
        self.release_load_gate(key, &load_gate);
        Ok((handle, true))
    }

    fn release_load_gate(&self, key: &str, gate: &Arc<Mutex<()>>) {
        let mut gates = self.load_locks.lock().expect("backend load gates poisoned");
        if gates
            .get(key)
            .is_some_and(|current| Arc::ptr_eq(current, gate) && Arc::strong_count(current) == 2)
        {
            gates.remove(key);
        }
    }

    /// Manually remove a key, e.g. when model weights have been updated on
    /// disk. Active handles remain fully accounted and are marked stale; the
    /// entry is removed before its first lookup after becoming idle.
    pub fn invalidate(&self, key: &str) {
        let mut guard = self.inner.lock().expect("backend cache poisoned");
        let idle = guard
            .map
            .get(key)
            .is_some_and(|entry| Arc::strong_count(&entry.backend) == 1);
        if idle {
            let Some(entry) = guard.map.remove(key) else {
                return;
            };
            self.budget.sub(entry.size_bytes);
            if let Some(accounting) = &self.resident_accounting {
                accounting.mark_evicted(&entry.allocation_id);
            }
            guard.lru.retain(|k| k != key);
        } else if let Some(entry) = guard.map.get_mut(key) {
            entry.invalidated = true;
        }
    }

    pub fn contains(&self, key: &str) -> bool {
        self.inner
            .lock()
            .expect("backend cache poisoned")
            .map
            .contains_key(key)
    }

    /// Remove one model only when the cache owns its last handle. Used by the
    /// model-removal seam after the admission coordinator has excluded new
    /// reservations for this ID.
    pub fn evict_if_idle(&self, key: &str) -> bool {
        let mut guard = self.inner.lock().expect("backend cache poisoned");
        let Some(entry) = guard.map.get(key) else {
            return true;
        };
        if Arc::strong_count(&entry.backend) != 1 {
            return false;
        }
        let entry = guard.map.remove(key).expect("entry checked above");
        guard.lru.retain(|candidate| candidate != key);
        self.budget.sub(entry.size_bytes);
        if let Some(accounting) = &self.resident_accounting {
            accounting.mark_evicted(&entry.allocation_id);
        }
        true
    }

    /// Reclaim idle LRU entries after a live policy decrease. Entries with an
    /// outstanding caller handle are skipped; a later access/sweep retries
    /// after active work completes.
    pub fn enforce_budget(&self) -> (usize, u64) {
        let mut guard = self.inner.lock().expect("backend cache poisoned");
        let mut evicted = 0usize;
        let mut bytes = 0u64;
        let mut examined = guard.lru.len();
        while (self.budget.over_budget() || self.budget.is_disabled()) && examined > 0 {
            examined -= 1;
            let Some(key) = guard.lru.pop_front() else {
                break;
            };
            let is_idle = guard
                .map
                .get(&key)
                .is_some_and(|entry| Arc::strong_count(&entry.backend) == 1);
            if !is_idle {
                guard.lru.push_back(key);
                continue;
            }
            if let Some(entry) = guard.map.remove(&key) {
                self.budget.sub(entry.size_bytes);
                if let Some(accounting) = &self.resident_accounting {
                    accounting.mark_evicted(&entry.allocation_id);
                }
                evicted += 1;
                bytes = bytes.saturating_add(entry.size_bytes);
            }
        }
        (evicted, bytes)
    }

    /// Evict every entry untouched for longer than the configured idle
    /// TTL. Returns `(entries_evicted, bytes_evicted)`. A no-op (returns
    /// `(0, 0)`) when idle eviction is disabled (`idle_ttl == None`).
    ///
    /// Like capacity eviction, this only drops the cache's own handle;
    /// outstanding `Arc<Mutex<T>>` handles keep working and the RAM is
    /// reclaimed when the last one drops. At idle there are none, so the
    /// model's resident memory is released promptly. Intended to be
    /// driven on a timer by the daemon (car-releases#67).
    pub fn evict_idle(&self) -> (usize, u64) {
        let Some(ttl) = self.idle_ttl else {
            return (0, 0);
        };
        let now = Instant::now();
        let mut guard = self.inner.lock().expect("backend cache poisoned");
        let stale: Vec<String> = guard
            .map
            .iter()
            .filter(|(_, e)| now.duration_since(e.last_used) >= ttl)
            .map(|(k, _)| k.clone())
            .collect();
        let mut entries = 0usize;
        let mut bytes = 0u64;
        for key in stale {
            let is_idle = guard
                .map
                .get(&key)
                .is_some_and(|entry| Arc::strong_count(&entry.backend) == 1);
            if !is_idle {
                continue;
            }
            if let Some(victim) = guard.map.remove(&key) {
                self.budget.sub(victim.size_bytes);
                if let Some(accounting) = &self.resident_accounting {
                    accounting.mark_evicted(&victim.allocation_id);
                }
                guard.lru.retain(|k| k != &key);
                entries += 1;
                bytes = bytes.saturating_add(victim.size_bytes);
                drop(victim);
            }
        }
        (entries, bytes)
    }

    /// Evict all entries. Outstanding handles continue to work.
    pub fn clear(&self) {
        let mut guard = self.inner.lock().expect("backend cache poisoned");
        // Subtract only THIS cache's bytes from the shared total (it may be
        // shared with other caches whose entries must remain accounted).
        let idle_keys = guard
            .map
            .iter()
            .filter(|(_, entry)| Arc::strong_count(&entry.backend) == 1)
            .map(|(key, _)| key.clone())
            .collect::<Vec<_>>();
        let mut freed = 0u64;
        for key in idle_keys {
            if let Some(entry) = guard.map.remove(&key) {
                freed = freed.saturating_add(entry.size_bytes);
                if let Some(accounting) = &self.resident_accounting {
                    accounting.mark_evicted(&entry.allocation_id);
                }
            }
            guard.lru.retain(|candidate| candidate != &key);
        }
        self.budget.sub(freed);
    }

    /// Returns `(entries, total_bytes, budget_bytes)` for diagnostics. Note:
    /// `total_bytes` is the SHARED total across all caches on this budget, not
    /// just this cache's; `entries` is this cache's count.
    pub fn stats(&self) -> (usize, u64, u64) {
        let guard = self.inner.lock().expect("backend cache poisoned");
        (
            guard.map.len(),
            self.budget.total(),
            self.budget.budget_bytes(),
        )
    }
}

/// Measure every installed regular file under a model path. Native loaders span
/// safetensors, GGUF/GGML, ONNX, vocab tables, and single-file checkpoints; a
/// safetensors-only walk silently reported zero for several real allocations.
/// Counting small configs/tokenizers is conservative and keeps the measurement
/// tied to the exact installed artifact rather than catalog metadata.
pub fn estimate_model_size(model_dir: &Path) -> u64 {
    fn visit(dir: &Path, total: &mut u64) {
        // `metadata` follows a leaf symlink, which is how Hugging Face
        // snapshots expose blob files. A directory symlink is deliberately not
        // traversed, avoiding cycles and out-of-snapshot recursion.
        if let Ok(meta) = dir.metadata() {
            if meta.is_file() {
                *total = total.saturating_add(meta.len());
                return;
            }
        }
        let Ok(entries) = std::fs::read_dir(dir) else {
            return;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            let Ok(link_meta) = path.symlink_metadata() else {
                continue;
            };
            if link_meta.file_type().is_symlink() {
                if let Ok(target_meta) = path.metadata() {
                    if target_meta.is_file() {
                        *total = total.saturating_add(target_meta.len());
                    }
                }
                continue;
            }
            if link_meta.is_dir() {
                visit(&path, total);
                continue;
            }
            if link_meta.is_file() {
                *total = total.saturating_add(link_meta.len());
            }
        }
    }
    let mut total = 0u64;
    visit(model_dir, &mut total);
    total
}

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

    struct FixedProbe;

    impl crate::resource_policy::LiveMemoryProbe for FixedProbe {
        fn available_memory_mb(
            &self,
        ) -> Result<Option<u64>, crate::resource_policy::ResourcePolicyError> {
            Ok(Some(24_000))
        }
    }

    #[test]
    fn final_handle_reaper_backoff_is_bounded_and_resets_after_progress() {
        let mut backoff = ReaperBackoff::new(Duration::from_millis(10), Duration::from_millis(250));
        assert_eq!(backoff.current(), Duration::from_millis(10));
        for _ in 0..10 {
            backoff.no_progress();
        }
        assert_eq!(backoff.current(), Duration::from_millis(250));
        backoff.progress();
        assert_eq!(backoff.current(), Duration::from_millis(10));
    }

    #[test]
    fn cache_hit_returns_same_handle() {
        let cache: BackendCache<u32> = BackendCache::new(1024);
        let a = cache.get_or_load::<()>("a", 100, || Ok(42)).unwrap();
        let b = cache
            .get_or_load::<()>("a", 100, || panic!("should not reload"))
            .unwrap();
        assert!(Arc::ptr_eq(&a, &b));
    }

    #[test]
    fn evicts_lru_when_over_budget() {
        let cache: BackendCache<u32> = BackendCache::new(250);
        let _a = cache.get_or_load::<()>("a", 100, || Ok(1)).unwrap();
        let b = cache.get_or_load::<()>("b", 100, || Ok(2)).unwrap();
        // Total 200, still under 250. Touch `a` so `b` is LRU.
        let _a_again = cache
            .get_or_load::<()>("a", 100, || panic!("cached"))
            .unwrap();
        drop(b);
        // Insert `c`: pushes us to 300, evicts `b` (the LRU one).
        let _c = cache.get_or_load::<()>("c", 100, || Ok(3)).unwrap();
        let (n, bytes, budget) = cache.stats();
        assert_eq!(n, 2, "a + c should remain, b evicted");
        assert_eq!(bytes, 200);
        assert_eq!(budget, 250);
    }

    #[test]
    fn zero_budget_disables_cache_but_returns_handle() {
        let cache: BackendCache<u32> = BackendCache::new(0);
        let mut load_count = 0u32;
        let a = cache
            .get_or_load::<()>("a", 100, || {
                load_count += 1;
                Ok(1)
            })
            .unwrap();
        assert_eq!(*a.lock().unwrap(), 1);
        let b = cache
            .get_or_load::<()>("a", 100, || {
                load_count += 1;
                Ok(1)
            })
            .unwrap();
        assert_eq!(*b.lock().unwrap(), 1);
        assert_eq!(load_count, 2, "disabled cache reloads every call");
        assert!(!Arc::ptr_eq(&a, &b));
    }

    #[test]
    fn evict_idle_drops_stale_entries_below_capacity() {
        // Tiny TTL so the test doesn't sleep long; budget far above usage
        // so capacity eviction never fires — this exercises *idle* eviction.
        let cache: BackendCache<u32> =
            BackendCache::with_idle_ttl(1_000_000, Some(Duration::from_millis(20)));
        let _a = cache.get_or_load::<()>("a", 100, || Ok(1)).unwrap();
        let b = cache.get_or_load::<()>("b", 100, || Ok(2)).unwrap();
        assert_eq!(cache.stats().0, 2);
        // Nothing is idle yet.
        assert_eq!(cache.evict_idle(), (0, 0));
        std::thread::sleep(Duration::from_millis(40));
        drop(b);
        // Touch `a` so only `b` is stale.
        let _a_again = cache
            .get_or_load::<()>("a", 100, || panic!("cached"))
            .unwrap();
        let (entries, bytes) = cache.evict_idle();
        assert_eq!((entries, bytes), (1, 100), "only b should be swept");
        let (n, total, _) = cache.stats();
        assert_eq!(n, 1, "a remains");
        assert_eq!(total, 100);
    }

    #[test]
    fn evict_idle_noop_when_disabled() {
        let cache: BackendCache<u32> = BackendCache::new(1024); // idle_ttl: None
        let _a = cache.get_or_load::<()>("a", 100, || Ok(1)).unwrap();
        assert_eq!(cache.evict_idle(), (0, 0));
        assert_eq!(cache.stats().0, 1, "disabled idle eviction keeps the entry");
    }

    #[test]
    fn invalidate_removes_key() {
        let cache: BackendCache<u32> = BackendCache::new(1024);
        let a = cache.get_or_load::<()>("a", 100, || Ok(1)).unwrap();
        assert_eq!(cache.stats().0, 1);
        drop(a);
        cache.invalidate("a");
        assert_eq!(cache.stats().0, 0);
    }

    #[test]
    fn active_invalidation_is_deferred_without_losing_resident_accounting() {
        let cache: BackendCache<u32> = BackendCache::new(1024);
        let active = cache.get_or_load::<()>("a", 100, || Ok(1)).unwrap();

        cache.invalidate("a");
        assert_eq!(cache.stats().1, 100, "active weights remain accounted");
        drop(active);

        let replacement = cache.get_or_load::<()>("a", 100, || Ok(2)).unwrap();
        assert_eq!(*replacement.lock().unwrap(), 2);
        assert_eq!(cache.stats().1, 100, "replacement is not double-counted");
    }

    #[test]
    fn installed_size_measurement_covers_single_files_and_non_safetensor_weights() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("model.gguf"), vec![0_u8; 17]).unwrap();
        std::fs::write(dir.path().join("encoder.onnx"), vec![0_u8; 23]).unwrap();
        let single = dir.path().join("whisper.bin");
        std::fs::write(&single, vec![0_u8; 31]).unwrap();

        assert_eq!(estimate_model_size(dir.path()), 71);
        assert_eq!(estimate_model_size(&single), 31);
    }

    #[cfg(unix)]
    #[test]
    fn installed_size_measurement_follows_hf_leaf_symlinks_but_not_directory_symlinks() {
        use std::os::unix::fs::symlink;

        let dir = tempfile::tempdir().unwrap();
        let blobs = dir.path().join("blobs");
        let snapshot = dir.path().join("snapshots/revision");
        std::fs::create_dir_all(&blobs).unwrap();
        std::fs::create_dir_all(&snapshot).unwrap();
        std::fs::write(blobs.join("weights"), vec![0_u8; 47]).unwrap();
        symlink("../../blobs/weights", snapshot.join("model.safetensors")).unwrap();
        symlink("../../blobs", snapshot.join("directory-link")).unwrap();
        symlink("../../blobs/missing", snapshot.join("dangling")).unwrap();

        assert_eq!(estimate_model_size(&snapshot), 47);
    }

    #[test]
    fn separate_cache_generations_keep_exact_resident_owners_until_final_handle_drop() {
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let first: BackendCache<u32> = BackendCache::from_shared_with_admission(
            SharedModelBudget::new(1024),
            None,
            Some(coordinator.clone()),
        );
        let second: BackendCache<u32> = BackendCache::from_shared_with_admission(
            SharedModelBudget::new(1024),
            None,
            Some(coordinator.clone()),
        );
        let mut first_reservation = coordinator
            .reserve_measured_host("same-model", 100, 0)
            .unwrap();
        let (first_handle, _) = first
            .get_or_load_admitted("same-model", 100, &mut first_reservation, || {
                Ok::<_, crate::InferenceError>(1)
            })
            .unwrap();
        drop(first_reservation);
        let mut second_reservation = coordinator
            .reserve_measured_host("same-model", 100, 0)
            .unwrap();
        let (_second_handle, _) = second
            .get_or_load_admitted("same-model", 100, &mut second_reservation, || {
                Ok::<_, crate::InferenceError>(2)
            })
            .unwrap();
        drop(second_reservation);

        assert_eq!(coordinator.resident_allocation_ids("same-model").len(), 2);
        drop(first);
        assert_eq!(coordinator.resident_allocation_ids("same-model").len(), 2);
        drop(first_handle);
        for _ in 0..100 {
            if coordinator.resident_allocation_ids("same-model").len() == 1 {
                break;
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(coordinator.resident_allocation_ids("same-model").len(), 1);
    }

    #[test]
    fn shared_budget_spans_caches_and_each_self_trims() {
        // One budget shared by two caches (the #427 aggregate-bound fix).
        let budget = SharedModelBudget::new(250);
        let a: BackendCache<u32> = BackendCache::from_shared(budget.clone(), None);
        let b: BackendCache<u32> = BackendCache::from_shared(budget.clone(), None);

        let a1 = a.get_or_load::<()>("a1", 100, || Ok(1)).unwrap();
        let _ = b.get_or_load::<()>("b1", 100, || Ok(2)).unwrap();
        // The total is SHARED across both caches.
        assert_eq!(budget.total(), 200);

        // Growing A past the shared budget makes A evict its OWN LRU (a1) back
        // under the shared cap — B's entry in the other cache is untouched.
        drop(a1);
        let _ = a.get_or_load::<()>("a2", 100, || Ok(3)).unwrap(); // shared→300>250
        assert_eq!(budget.total(), 200, "A self-trimmed to the shared budget");
        assert_eq!(a.stats().0, 1, "A kept a2, evicted a1");
        assert_eq!(b.stats().0, 1, "B's b1 not evicted by A");
    }

    #[test]
    fn default_budget_is_ram_derived_not_flat() {
        // The default scales with detected RAM (~60%); never the old flat 24 GB
        // sentinel on a normal machine, and always positive.
        let mb = default_model_cache_mb();
        assert!(mb > 0, "RAM-derived default should be positive");
    }

    #[test]
    fn backend_cache_policy_decrease_evicts_idle_entries() {
        let budget = SharedModelBudget::new(1_000);
        let cache: BackendCache<u32> = BackendCache::from_shared(budget.clone(), None);
        let handle = cache.get_or_load::<()>("model", 600, || Ok(1)).unwrap();
        drop(handle);

        budget.set_budget_bytes(500);
        assert_eq!(cache.enforce_budget(), (1, 600));
        assert_eq!(cache.stats(), (0, 0, 500));
    }

    #[test]
    fn backend_cache_policy_decrease_does_not_kill_active_handle() {
        let budget = SharedModelBudget::new(1_000);
        let cache: BackendCache<u32> = BackendCache::from_shared(budget.clone(), None);
        let handle = cache.get_or_load::<()>("model", 600, || Ok(1)).unwrap();

        budget.set_budget_bytes(0);
        assert_eq!(cache.enforce_budget(), (0, 0));
        assert!(cache.contains("model"));
        drop(handle);
        assert_eq!(cache.enforce_budget(), (1, 600));
    }

    #[test]
    fn backend_cache_same_key_load_is_singleflight() {
        let cache = Arc::new(BackendCache::new(1_000));
        let loads = Arc::new(AtomicU64::new(0));
        let mut threads = Vec::new();
        for _ in 0..8 {
            let cache = Arc::clone(&cache);
            let loads = Arc::clone(&loads);
            threads.push(std::thread::spawn(move || {
                cache
                    .get_or_load::<()>("model", 100, || {
                        loads.fetch_add(1, Ordering::SeqCst);
                        std::thread::sleep(Duration::from_millis(5));
                        Ok(42_u32)
                    })
                    .unwrap()
            }));
        }
        let handles = threads
            .into_iter()
            .map(|thread| thread.join().unwrap())
            .collect::<Vec<_>>();
        assert_eq!(loads.load(Ordering::SeqCst), 1);
        assert!(handles
            .windows(2)
            .all(|pair| Arc::ptr_eq(&pair[0], &pair[1])));
    }

    #[test]
    fn backend_cache_eviction_updates_admission_residency() {
        let hw = crate::hardware::HardwareInfo {
            os: "test".into(),
            arch: "test".into(),
            cpu_cores: 8,
            total_ram_mb: 32 * 1024,
            gpu_backend: crate::hardware::GpuBackend::Metal,
            gpu_memory_mb: None,
            gpu_devices: Vec::new(),
            recommended_model: "fixture".into(),
            recommended_context: 4_096,
            max_model_mb: 32 * 1024,
        };
        let admission = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::everyday(),
            hw,
        ));
        let budget = SharedModelBudget::new(2 * 1024 * 1024);
        let cache: BackendCache<u32> =
            BackendCache::from_shared_with_admission(budget.clone(), None, Some(admission.clone()));
        let handle = cache
            .get_or_load::<()>("model", 1024 * 1024, || Ok(1))
            .unwrap();
        assert_eq!(admission.resident_model_mb(), 1);
        drop(handle);

        budget.set_budget_bytes(0);
        assert_eq!(cache.enforce_budget().0, 1);
        assert_eq!(admission.resident_model_mb(), 0);
    }

    #[test]
    fn backend_cache_drop_never_peer_discounts_an_outstanding_handle() {
        let admission = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo {
                total_ram_mb: 32 * 1024,
                ..crate::hardware::HardwareInfo::detect()
            },
        ));
        let budget = SharedModelBudget::new(8 * 1024 * 1024);
        let cache: BackendCache<u32> =
            BackendCache::from_shared_with_admission(budget.clone(), None, Some(admission.clone()));
        let active = cache
            .get_or_load::<()>("model", 1024 * 1024, || Ok(1))
            .unwrap();
        assert_eq!(admission.resident_model_mb(), 1);

        drop(cache);
        assert_eq!(
            admission.resident_model_mb(),
            1,
            "dropping a cache must not call an outstanding allocation safe"
        );
        assert_eq!(budget.total(), 1024 * 1024);
        drop(active);
        for _ in 0..100 {
            if admission.resident_model_mb() == 0 && budget.total() == 0 {
                break;
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(
            admission.resident_model_mb(),
            0,
            "the final external handle drop must release resident accounting"
        );
        assert_eq!(budget.total(), 0);
    }

    #[test]
    fn backend_cache_targeted_eviction_refuses_active_then_releases_idle() {
        let cache: BackendCache<u32> = BackendCache::new(1_000);
        let handle = cache.get_or_load::<()>("model", 100, || Ok(1)).unwrap();
        assert!(!cache.evict_if_idle("model"));
        assert!(cache.contains("model"));
        drop(handle);
        assert!(cache.evict_if_idle("model"));
        assert!(!cache.contains("model"));
    }

    #[test]
    fn admitted_cache_publication_atomically_transfers_residency() {
        let hw = crate::hardware::HardwareInfo {
            os: "test".into(),
            arch: "test".into(),
            cpu_cores: 8,
            total_ram_mb: 32 * 1024,
            gpu_backend: crate::hardware::GpuBackend::Metal,
            gpu_memory_mb: None,
            gpu_devices: Vec::new(),
            recommended_model: "fixture".into(),
            recommended_context: 4_096,
            max_model_mb: 32 * 1024,
        };
        let admission = Arc::new(
            crate::resource_policy::LocalAdmissionCoordinator::with_probe(
                crate::resource_policy::ResourcePolicy::local_focused(),
                hw,
                Arc::new(FixedProbe),
            ),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let budget = SharedModelBudget::new(8 * 1024 * 1024 * 1024);
        let cache: BackendCache<u32> =
            BackendCache::from_shared_with_admission(budget, None, Some(admission.clone()));
        let mut reservation = admission.reserve(&model, 2_048).unwrap();

        let (handle, retention) = cache
            .get_or_load_admitted(&model.id, 3 * 1024 * 1024 * 1024, &mut reservation, || {
                Ok::<_, crate::InferenceError>(42)
            })
            .unwrap();

        assert_eq!(*handle.lock().unwrap(), 42);
        assert_eq!(retention, BackendRetention::Resident);
        assert_eq!(admission.resident_model_mb(), 3 * 1024);
        assert_eq!(
            admission.preflight(&model, 2_048).active_reservations_mb,
            reservation.reserved_incremental_mb()
        );
    }

    #[test]
    fn admitted_zero_cache_success_is_explicitly_transient_and_not_resident() {
        let admission = Arc::new(
            crate::resource_policy::LocalAdmissionCoordinator::with_probe(
                crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
                crate::hardware::HardwareInfo {
                    total_ram_mb: 32 * 1024,
                    ..crate::hardware::HardwareInfo::detect()
                },
                Arc::new(FixedProbe),
            ),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let cache: BackendCache<u32> = BackendCache::from_shared_with_admission(
            SharedModelBudget::new(0),
            None,
            Some(admission.clone()),
        );
        let mut reservation = admission.reserve(&model, 64).unwrap();
        let (handle, retention) = cache
            .get_or_load_admitted(&model.id, 1024 * 1024, &mut reservation, || {
                Ok::<_, crate::InferenceError>(42)
            })
            .unwrap();

        assert_eq!(*handle.lock().unwrap(), 42);
        assert_eq!(retention, BackendRetention::Transient);
        drop(handle);
        drop(reservation);
        assert!(!admission.is_resident(&model.id));
    }
}