car_inference/

resource_policy.rs

//! Persisted admission policy for CAR-managed local model memory.
//!
//! This module answers how much host memory CAR may admit for local models. It
//! deliberately does not claim to impose an operating-system RSS limit. Apple
//! Silicon and CPU inference use the host budget; discrete CUDA keeps measured
//! VRAM as a separate fit constraint instead of applying the Apple unified-
//! memory percentage to it.

use std::collections::{HashMap, HashSet};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Weak};
use std::sync::{Mutex, OnceLock};

use serde::{Deserialize, Serialize};
use thiserror::Error;

use crate::hardware::{GpuBackend, HardwareInfo};
use crate::schema::ModelSchema;

/// File name under the CAR state root.
pub const RESOURCE_POLICY_FILE: &str = "model-resource-policy.json";

const EVERYDAY_MODEL_PERCENT: u64 = 40;
const LOCAL_FOCUSED_MODEL_PERCENT: u64 = 80;
const EMERGENCY_RESERVE_PERCENT: u64 = 10;
const MINIMUM_EMERGENCY_RESERVE_MB: u64 = 2 * 1024;
const MAX_POLICY_BYTES: u64 = 64 * 1024;

/// Context window used by recommendation-time memory estimates. This is a
/// typical working window, not a model's advertised maximum.
pub const RECOMMENDATION_CONTEXT_TOKENS: usize = 8_192;

const METAL_RUNTIME_OVERHEAD_MB: u64 = 512;
const CUDA_RUNTIME_OVERHEAD_MB: u64 = 512;
const CPU_RUNTIME_OVERHEAD_MB: u64 = 1_024;
const TRANSIENT_ALLOCATION_MARGIN_MB: u64 = 1_024;

/// The bound an inference worker sets on MLX's buffer cache: the transient
/// allocation margin admission already charges every local model, so the
/// reuse cache stays inside what was admitted rather than MLX's platform
/// default (car#1899).
pub fn allocator_cache_limit_bytes() -> u64 {
    TRANSIENT_ALLOCATION_MARGIN_MB * 1024 * 1024
}
/// Memory accounting uses binary megabytes throughout this module.
const BYTES_PER_MB: u128 = 1024 * 1024;

/// Preset or exact user-selected local-model admission behavior.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ResourceProfile {
    Everyday,
    LocalFocused,
    Custom,
}

/// The persisted local-model memory preference.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct ResourcePolicy {
    pub profile: ResourceProfile,
    pub custom_max_model_mb: Option<u64>,
}

impl Default for ResourcePolicy {
    fn default() -> Self {
        Self::everyday()
    }
}

impl ResourcePolicy {
    pub fn everyday() -> Self {
        Self {
            profile: ResourceProfile::Everyday,
            custom_max_model_mb: None,
        }
    }

    pub fn local_focused() -> Self {
        Self {
            profile: ResourceProfile::LocalFocused,
            custom_max_model_mb: None,
        }
    }

    /// Build an exact Custom policy from user-facing GB input.
    ///
    /// Only finite, nonnegative half-GB increments are accepted. The persisted
    /// value is integer MB, so it round-trips without floating-point drift.
    pub fn custom_gb(gigabytes: f64) -> Result<Self, ResourcePolicyError> {
        if !gigabytes.is_finite() || gigabytes < 0.0 {
            return Err(ResourcePolicyError::InvalidCustomGigabytes(gigabytes));
        }

        let half_gb_steps = gigabytes * 2.0;
        if half_gb_steps.fract() != 0.0 || half_gb_steps > u64::MAX as f64 {
            return Err(ResourcePolicyError::InvalidCustomGigabytes(gigabytes));
        }
        let steps = half_gb_steps as u64;
        let custom_max_model_mb = steps
            .checked_mul(512)
            .ok_or(ResourcePolicyError::InvalidCustomGigabytes(gigabytes))?;

        Ok(Self {
            profile: ResourceProfile::Custom,
            custom_max_model_mb: Some(custom_max_model_mb),
        })
    }

    /// Evaluate the stored preference against the current machine capacity.
    ///
    /// Custom values are clamped for this evaluation only. The persisted value
    /// is left untouched so a preference moved between machines is never
    /// silently rewritten.
    pub fn effective_budget(&self, total_memory_mb: u64) -> EffectiveResourceBudget {
        let emergency_reserve_mb = minimum_emergency_reserve(total_memory_mb);
        let safe_maximum_mb = total_memory_mb.saturating_sub(emergency_reserve_mb);
        let requested_ceiling_mb = match self.profile {
            ResourceProfile::Everyday => percent_of(total_memory_mb, EVERYDAY_MODEL_PERCENT),
            ResourceProfile::LocalFocused => {
                percent_of(total_memory_mb, LOCAL_FOCUSED_MODEL_PERCENT)
            }
            ResourceProfile::Custom => self.custom_max_model_mb.unwrap_or(0),
        };
        let configured_model_ceiling_mb = requested_ceiling_mb.min(safe_maximum_mb);
        let normalization_notice = (matches!(self.profile, ResourceProfile::Custom)
            && requested_ceiling_mb > safe_maximum_mb)
            .then(|| {
                format!(
                    "The saved Custom allocation was adjusted from {requested_ceiling_mb} MB to \
                     {safe_maximum_mb} MB on this machine to preserve the \
                     {emergency_reserve_mb} MB emergency reserve."
                )
            });

        EffectiveResourceBudget {
            total_memory_mb,
            emergency_reserve_mb,
            configured_model_ceiling_mb,
            effective_new_load_ceiling_mb: configured_model_ceiling_mb,
            normalization_notice,
        }
    }

    /// Recommendation target: the profile's own admission ceiling.
    ///
    /// Everyday used to recommend only up to half of its 40% ceiling, which on
    /// a 64 GB Mac (12.8 GB) excluded every local model that scores near the
    /// frontier (Qwen3.8-27B, Qwen3.6-35B-A3B, Gemma-4-12B) from automatic
    /// selection, though each loads within the ceiling. The estimate a row is
    /// compared against already carries its context, runtime and transient
    /// margin, and admission still checks live memory at load.
    pub fn recommendation_target_mb(&self, total_memory_mb: u64) -> u64 {
        self.effective_budget(total_memory_mb)
            .configured_model_ceiling_mb
    }

    /// Validate the exact persisted shape accepted by CAR.
    ///
    /// Keeping this check on the value itself ensures file reads and writes
    /// cannot drift into accepting different policy documents.
    pub fn validate(&self) -> Result<(), ResourcePolicyError> {
        match self.profile {
            ResourceProfile::Custom => match self.custom_max_model_mb {
                Some(value) if value.is_multiple_of(512) => Ok(()),
                Some(value) => Err(ResourcePolicyError::InvalidPolicy {
                    reason: format!(
                        "Custom model RAM must be a 0.5 GB (512 MB) increment; got {value} MB"
                    ),
                }),
                None => Err(ResourcePolicyError::InvalidPolicy {
                    reason: "Custom profile requires custom_max_model_mb".into(),
                }),
            },
            ResourceProfile::Everyday | ResourceProfile::LocalFocused => {
                if self.custom_max_model_mb.is_none() {
                    Ok(())
                } else {
                    Err(ResourcePolicyError::InvalidPolicy {
                        reason: format!(
                            "{:?} profile must not set custom_max_model_mb",
                            self.profile
                        ),
                    })
                }
            }
        }
    }
}

/// Static policy math. Live-memory evidence is added by the admission service.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct EffectiveResourceBudget {
    pub total_memory_mb: u64,
    pub emergency_reserve_mb: u64,
    pub configured_model_ceiling_mb: u64,
    pub effective_new_load_ceiling_mb: u64,
    pub normalization_notice: Option<String>,
}

/// Measured dedicated accelerator memory. This is intentionally separate from
/// host RAM; CAR does not apply Apple unified-memory percentages to CUDA VRAM.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct AcceleratorResourceBudget {
    pub total_mb: u64,
    pub budget_mb: u64,
}

/// Host policy plus an optional distinct accelerator fit pool.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResourceEvaluation {
    pub host_memory: EffectiveResourceBudget,
    pub accelerator_memory: Option<AcceleratorResourceBudget>,
}

/// Why CAR believes the weight portion of a model-memory estimate.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ModelResourceEvidence {
    /// The catalog declared a RAM or artifact-size floor.
    CatalogExact,
    /// A caller supplied a physical on-disk measurement.
    FileSystemMeasured,
    /// CAR estimated from parameter count because no stronger fact existed.
    Heuristic,
}

/// Recommendation-time cold-load estimate. This is static capacity evidence,
/// not a live-memory probe or an operating-system RSS guarantee.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct ModelMemoryEstimate {
    pub weights_mb: u64,
    pub runtime_overhead_mb: u64,
    pub context_overhead_mb: u64,
    pub transient_margin_mb: u64,
    pub estimated_peak_mb: u64,
    pub evidence: ModelResourceEvidence,
}

/// Result of checking one CAR-managed local allocation against both the saved
/// policy and live machine headroom. `LiveMemoryUnknown` is intentionally not
/// collapsed into `Allowed`: callers may choose a conservative static-policy
/// fallback, but must not present the model as safe *right now*.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct LocalLoadPreflight {
    pub model_id: String,
    pub estimate: ModelMemoryEstimate,
    pub configured_ceiling_mb: u64,
    pub resident_model_mb: u64,
    pub active_reservations_mb: u64,
    pub estimated_incremental_mb: u64,
    pub accelerator_total_mb: Option<u64>,
    pub accelerator_resident_mb: Option<u64>,
    pub accelerator_incremental_mb: Option<u64>,
    pub live_available_mb: Option<u64>,
    pub emergency_reserve_mb: u64,
    pub verdict: LocalLoadVerdict,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum LocalLoadVerdict {
    Allowed,
    LiveMemoryUnknown,
    DisabledByPolicy,
    ExceedsConfiguredCeiling,
    InsufficientLiveMemory,
    ModelMaintenance,
    /// A CAR-owned process is still being reaped. Its weights remain charged
    /// and no replacement may peer-discount them until exit is confirmed.
    PendingTeardown,
}

impl LocalLoadVerdict {
    pub fn permits_static_fallback(&self) -> bool {
        matches!(self, Self::Allowed | Self::LiveMemoryUnknown)
    }
}

/// Machine-specific live-memory evidence. Returning `Ok(None)` is distinct
/// from zero available memory and is preserved in the public preflight.
pub trait LiveMemoryProbe: Send + Sync {
    fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError>;
}

#[derive(Default)]
pub struct SystemLiveMemoryProbe;

impl LiveMemoryProbe for SystemLiveMemoryProbe {
    fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError> {
        Ok(crate::hardware::available_ram_mb())
    }
}

#[cfg(test)]
pub(crate) struct FixedLiveMemoryProbe(Option<u64>);

#[cfg(test)]
impl FixedLiveMemoryProbe {
    pub(crate) fn known(available_mb: u64) -> Self {
        Self(Some(available_mb))
    }

    fn unknown() -> Self {
        Self(None)
    }
}

#[cfg(test)]
impl LiveMemoryProbe for FixedLiveMemoryProbe {
    fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError> {
        Ok(self.0)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum WeightPlacement {
    Host,
    Accelerator,
}

#[derive(Clone, Debug)]
struct ResidentAllocation {
    weights_mb: u64,
    placement: WeightPlacement,
    logical_model_id: String,
}

#[derive(Clone, Default)]
struct AdmissionState {
    resident_models: HashMap<String, ResidentAllocation>,
    active_host_reservations_mb: u64,
    active_accelerator_reservations_mb: u64,
    active_by_model: HashMap<String, usize>,
    maintenance_models: HashSet<String>,
    pending_teardown_models: HashMap<String, HashSet<String>>,
    model_aliases: HashMap<String, HashSet<String>>,
    next_request_id: u64,
}

#[derive(Clone, Default)]
struct MachineAdmissionLedger {
    resident_models: HashMap<(u64, String), ResidentAllocation>,
    pending_allocations: HashMap<(u64, String), ResidentAllocation>,
    active_host_by_owner: HashMap<u64, u64>,
    active_accelerator_by_owner: HashMap<u64, u64>,
}

fn process_machine_admission_ledger() -> Arc<Mutex<MachineAdmissionLedger>> {
    static LEDGER: OnceLock<Arc<Mutex<MachineAdmissionLedger>>> = OnceLock::new();
    LEDGER
        .get_or_init(|| Arc::new(Mutex::new(MachineAdmissionLedger::default())))
        .clone()
}

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

/// Allocation identities are source-qualified so a worker process and a
/// supervised server for the same logical model cannot overwrite or evict one
/// another's physical-memory record.
pub fn worker_process_allocation_id(model_id: &str) -> String {
    format!("worker:{model_id}")
}

pub fn vllm_process_allocation_id(model_id: &str) -> String {
    format!("vllm:{model_id}")
}

/// One process-wide coordinator for local cold loads and per-request runtime
/// allocations. Its mutex makes check-and-reserve one atomic operation, so two
/// individually-safe requests cannot both consume the same headroom.
pub struct LocalAdmissionCoordinator {
    policy: std::sync::RwLock<ResourcePolicy>,
    hardware: HardwareInfo,
    live_probe: Arc<dyn LiveMemoryProbe>,
    state: Mutex<AdmissionState>,
    machine_ledger: Arc<Mutex<MachineAdmissionLedger>>,
    /// Cross-process activity ownership for exact resident allocation ids.
    /// Kept outside cloneable accounting snapshots so an idle cache/process
    /// holds its shared lease until exact eviction or confirmed teardown.
    resident_activity_leases: Mutex<HashMap<String, Arc<crate::model_management::ModelLease>>>,
    models_dir: std::sync::RwLock<Option<PathBuf>>,
    /// The Hugging Face hub a row may be served from instead of `models_dir`
    /// (see [`kv_config_path`]). `None` reads only `models_dir`, which keeps
    /// tests from sizing rows off whatever the developer has cached.
    hf_hub_dir: Option<PathBuf>,
    kv_geometry_cache: Mutex<HashMap<String, (ConfigStamp, KvGeometry)>>,
    owner_id: u64,
}

impl Drop for LocalAdmissionCoordinator {
    fn drop(&mut self) {
        let mut machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        machine
            .resident_models
            .retain(|(owner_id, _), _| *owner_id != self.owner_id);
        machine
            .pending_allocations
            .retain(|(owner_id, _), _| *owner_id != self.owner_id);
        machine.active_host_by_owner.remove(&self.owner_id);
        machine.active_accelerator_by_owner.remove(&self.owner_id);
    }
}

fn scoped_admission_registry() -> &'static Mutex<HashMap<PathBuf, Weak<LocalAdmissionCoordinator>>>
{
    static REGISTRY: OnceLock<Mutex<HashMap<PathBuf, Weak<LocalAdmissionCoordinator>>>> =
        OnceLock::new();
    REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Stable identity for a state root. Existing roots are filesystem-
/// canonicalized so symlink/relative aliases cannot split admission accounting.
/// If canonicalization is unavailable (commonly before first-run directory
/// creation), lexical absolute normalization is deliberately conservative:
/// aliases may be charged independently, but never peer-discount one another.
pub fn normalized_state_root_key(state_root: &Path) -> PathBuf {
    let absolute = if state_root.is_absolute() {
        state_root.to_path_buf()
    } else if let Ok(current) = std::env::current_dir() {
        current.join(state_root)
    } else {
        return state_root.to_path_buf();
    };
    // Do not lexically collapse `..` before asking the filesystem. In
    // `symlink/../state`, `..` applies after resolving the symlink and can name
    // a different physical directory than a lexical pop would produce.
    if let Ok(canonical) = car_winenv::canonicalize(&absolute) {
        return canonical;
    }

    // First-run roots often do not exist yet. Canonicalize the nearest
    // existing ancestor so a missing leaf beneath a symlink still shares the
    // same coordinator/runtime identity as its real path.
    let mut ancestor = absolute.clone();
    let mut missing_suffix = Vec::new();
    while let Some(component) = ancestor.components().next_back() {
        let name = match component {
            std::path::Component::Normal(name) => name.to_owned(),
            std::path::Component::CurDir => std::ffi::OsString::from("."),
            std::path::Component::ParentDir => std::ffi::OsString::from(".."),
            std::path::Component::RootDir | std::path::Component::Prefix(_) => break,
        };
        missing_suffix.push(name);
        ancestor.pop();
        if let Ok(mut canonical) = car_winenv::canonicalize(&ancestor) {
            for component in missing_suffix.iter().rev() {
                if component == std::ffi::OsStr::new(".") {
                    continue;
                }
                if component == std::ffi::OsStr::new("..") {
                    canonical.pop();
                } else {
                    canonical.push(component);
                }
            }
            return canonical;
        }
    }
    // No ancestor could be canonicalized. Keep the absolute spelling rather
    // than inventing peer identity; conservative duplicate charging is safer.
    absolute
}

pub fn install_shared_local_admission(coordinator: Arc<LocalAdmissionCoordinator>) {
    let root = normalized_state_root_key(&car_home::root_or_relative());
    scoped_admission_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .entry(root)
        .or_insert_with(|| Arc::downgrade(&coordinator));
}

pub fn shared_local_admission() -> Arc<LocalAdmissionCoordinator> {
    let root = car_home::root_or_relative();
    let policy = FileResourcePolicyRepository::new(root.clone())
        .load()
        .unwrap_or_else(|_| ResourcePolicy::everyday());
    scoped_local_admission(root, policy, HardwareInfo::detect())
}

pub fn scoped_local_admission(
    state_root: impl AsRef<Path>,
    policy: ResourcePolicy,
    hardware: HardwareInfo,
) -> Arc<LocalAdmissionCoordinator> {
    scoped_local_admission_with_models_dir(state_root, policy, hardware, None)
}

/// Return the admission coordinator for a state root, with optional access to
/// installed model configs for architecture-aware context-memory estimates.
pub fn scoped_local_admission_with_models_dir(
    state_root: impl AsRef<Path>,
    policy: ResourcePolicy,
    hardware: HardwareInfo,
    models_dir: Option<PathBuf>,
) -> Arc<LocalAdmissionCoordinator> {
    let state_root = normalized_state_root_key(state_root.as_ref());
    let mut registry = scoped_admission_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    registry.retain(|_, coordinator| coordinator.strong_count() > 0);
    if let Some(existing) = registry.get(&state_root).and_then(Weak::upgrade) {
        existing.set_policy(policy);
        if let Some(models_dir) = models_dir {
            existing.set_models_dir(models_dir);
        }
        return existing;
    }
    let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe_ledger_and_models_dir(
        policy,
        hardware,
        Arc::new(SystemLiveMemoryProbe),
        process_machine_admission_ledger(),
        models_dir,
        Some(crate::hf_cache::hub_dir()),
    ));
    registry.insert(state_root, Arc::downgrade(&coordinator));
    coordinator
}

pub fn local_admission_for_scope(
    state_root: impl AsRef<Path>,
) -> Option<Arc<LocalAdmissionCoordinator>> {
    let state_root = normalized_state_root_key(state_root.as_ref());
    scoped_admission_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .get(&state_root)
        .and_then(Weak::upgrade)
}

impl LocalAdmissionCoordinator {
    fn built_in_model_identity(model_id: &str) -> Option<String> {
        let lower = model_id.to_ascii_lowercase();
        if lower.contains("kokoro-82m") {
            if lower.contains("6bit") {
                return Some("mlx/kokoro-82m:6bit".into());
            }
            if lower.contains("bf16") {
                return Some("mlx/kokoro-82m:bf16".into());
            }
        }
        None
    }

    fn resolve_model_identity(state: &AdmissionState, model_id: &str) -> String {
        state
            .model_aliases
            .iter()
            .find_map(|(canonical, aliases)| {
                (canonical == model_id || aliases.contains(model_id)).then(|| canonical.clone())
            })
            .or_else(|| Self::built_in_model_identity(model_id))
            .unwrap_or_else(|| model_id.to_string())
    }

    /// Resolve any registered catalog/artifact/provider alias to the single
    /// allocation key used by admission, caches, and maintenance.
    pub fn canonical_model_id(&self, model_id: &str) -> String {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        Self::resolve_model_identity(&state, model_id)
    }

    pub fn register_model_aliases<I, S>(&self, canonical_model_id: &str, aliases: I)
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let mut aliases = aliases
            .into_iter()
            .map(Into::into)
            .collect::<HashSet<String>>();
        aliases.insert(canonical_model_id.to_string());
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut canonical_model_id = Self::built_in_model_identity(canonical_model_id)
            .unwrap_or_else(|| canonical_model_id.to_string());
        let intersecting = state
            .model_aliases
            .iter()
            .filter(|(existing, existing_aliases)| {
                aliases.contains(*existing)
                    || existing_aliases.iter().any(|alias| aliases.contains(alias))
            })
            .map(|(existing, _)| existing.clone())
            .collect::<Vec<_>>();
        if let Some(existing) = intersecting.first() {
            canonical_model_id = existing.clone();
        }
        for existing in intersecting {
            if let Some(existing_aliases) = state.model_aliases.remove(&existing) {
                aliases.extend(existing_aliases);
            }
            aliases.insert(existing);
        }
        aliases.insert(canonical_model_id.clone());

        let mut active = 0usize;
        for alias in &aliases {
            active = active.saturating_add(state.active_by_model.remove(alias).unwrap_or_default());
        }
        if active > 0 {
            *state
                .active_by_model
                .entry(canonical_model_id.clone())
                .or_default() += active;
        }
        let mut maintenance = false;
        for alias in &aliases {
            maintenance |= state.maintenance_models.remove(alias);
        }
        if maintenance {
            state.maintenance_models.insert(canonical_model_id.clone());
        }
        let mut pending = HashSet::new();
        for alias in &aliases {
            pending.extend(
                state
                    .pending_teardown_models
                    .remove(alias)
                    .unwrap_or_default(),
            );
        }
        if !pending.is_empty() {
            state
                .pending_teardown_models
                .entry(canonical_model_id.clone())
                .or_default()
                .extend(pending);
        }
        for resident in state.resident_models.values_mut() {
            if aliases.contains(&resident.logical_model_id) {
                resident.logical_model_id = canonical_model_id.clone();
            }
        }
        state
            .model_aliases
            .insert(canonical_model_id.clone(), aliases.clone());
        // Lock order is always scoped state then machine ledger. Holding both
        // here makes late alias registration one atomic identity migration;
        // reserve cannot observe half-rekeyed maintenance/residency state.
        let mut machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        for ((owner_id, _), resident) in machine.resident_models.iter_mut() {
            if *owner_id == self.owner_id && aliases.contains(&resident.logical_model_id) {
                resident.logical_model_id = canonical_model_id.clone();
            }
        }
        for ((owner_id, _), resident) in machine.pending_allocations.iter_mut() {
            if *owner_id == self.owner_id && aliases.contains(&resident.logical_model_id) {
                resident.logical_model_id = canonical_model_id.clone();
            }
        }
    }
    pub fn new(policy: ResourcePolicy, hardware: HardwareInfo) -> Self {
        Self::with_probe(policy, hardware, Arc::new(SystemLiveMemoryProbe))
    }

    pub fn with_probe(
        policy: ResourcePolicy,
        hardware: HardwareInfo,
        live_probe: Arc<dyn LiveMemoryProbe>,
    ) -> Self {
        Self::with_probe_and_ledger(
            policy,
            hardware,
            live_probe,
            Arc::new(Mutex::new(MachineAdmissionLedger::default())),
        )
    }

    fn with_probe_and_ledger(
        policy: ResourcePolicy,
        hardware: HardwareInfo,
        live_probe: Arc<dyn LiveMemoryProbe>,
        machine_ledger: Arc<Mutex<MachineAdmissionLedger>>,
    ) -> Self {
        Self::with_probe_ledger_and_models_dir(
            policy,
            hardware,
            live_probe,
            machine_ledger,
            None,
            None,
        )
    }

    fn with_probe_ledger_and_models_dir(
        policy: ResourcePolicy,
        hardware: HardwareInfo,
        live_probe: Arc<dyn LiveMemoryProbe>,
        machine_ledger: Arc<Mutex<MachineAdmissionLedger>>,
        models_dir: Option<PathBuf>,
        hf_hub_dir: Option<PathBuf>,
    ) -> Self {
        Self {
            policy: std::sync::RwLock::new(policy),
            hardware,
            live_probe,
            state: Mutex::new(AdmissionState::default()),
            machine_ledger,
            resident_activity_leases: Mutex::new(HashMap::new()),
            models_dir: std::sync::RwLock::new(models_dir),
            hf_hub_dir,
            kv_geometry_cache: Mutex::new(HashMap::new()),
            owner_id: next_admission_owner_id(),
        }
    }

    #[cfg(test)]
    fn with_probe_and_models_dir(
        policy: ResourcePolicy,
        hardware: HardwareInfo,
        live_probe: Arc<dyn LiveMemoryProbe>,
        models_dir: Option<PathBuf>,
    ) -> Self {
        Self::with_probe_ledger_and_models_dir(
            policy,
            hardware,
            live_probe,
            Arc::new(Mutex::new(MachineAdmissionLedger::default())),
            models_dir,
            None,
        )
    }

    fn model_memory_estimate(
        &self,
        model: &ModelSchema,
        context_tokens: usize,
    ) -> ModelMemoryEstimate {
        let geometry = self.kv_geometry(model);
        estimate_model_memory_with_geometry(
            model,
            &self.hardware,
            context_tokens,
            None,
            geometry.as_ref(),
        )
    }

    /// The installed checkpoint's KV geometry under this coordinator's models
    /// directory, cached per model id together with the `config.json` stamp it
    /// was read from. Every lookup stats the file: a removed checkpoint drops
    /// its entry and falls back, and a replaced or re-downloaded one (new
    /// mtime or length) is read again, so neither the catalog nor admission
    /// sizes a row from a config that is no longer there.
    fn kv_geometry(&self, model: &ModelSchema) -> Option<KvGeometry> {
        let models_dir = self
            .models_dir
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        let mut cache = self
            .kv_geometry_cache
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Stat before reading: a write landing between the two leaves an older
        // stamp beside newer contents, which the next lookup reads again.
        let Some(stamp) = models_dir.as_deref().and_then(|dir| {
            ConfigStamp::of(&kv_config_path(model, dir, self.hf_hub_dir.as_deref()))
        }) else {
            cache.remove(&model.id);
            return None;
        };
        if let Some((cached, geometry)) = cache.get(&model.id) {
            if *cached == stamp {
                return Some(geometry.clone());
            }
        }
        let geometry = kv_geometry_for(model, models_dir.as_deref(), self.hf_hub_dir.as_deref());
        match geometry.as_ref() {
            Some(geometry) => {
                cache.insert(model.id.clone(), (stamp, geometry.clone()));
            }
            None => {
                cache.remove(&model.id);
            }
        }
        geometry
    }

    /// The geometry admission sizes these rows' KV cache from, for the
    /// recommendation estimate to use the same input (car#2213).
    pub fn checkpoint_geometry<'a>(
        &self,
        models: impl IntoIterator<Item = &'a ModelSchema>,
    ) -> CheckpointGeometry {
        CheckpointGeometry {
            by_id: models
                .into_iter()
                .filter_map(|model| Some((model.id.clone(), self.kv_geometry(model)?)))
                .collect(),
        }
    }

    fn set_models_dir(&self, models_dir: PathBuf) {
        let mut configured = self
            .models_dir
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if configured.as_ref() == Some(&models_dir) {
            return;
        }
        *configured = Some(models_dir);
        self.kv_geometry_cache
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clear();
    }

    pub fn set_policy(&self, policy: ResourcePolicy) {
        *self
            .policy
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = policy;
    }

    pub fn policy(&self) -> ResourcePolicy {
        self.policy
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    pub fn mark_resident(&self, model_id: &str, weights_mb: u64) {
        let placement = self.default_weight_placement();
        self.mark_resident_with_placement(model_id, model_id, weights_mb, placement);
    }

    pub fn mark_resident_allocation(
        &self,
        logical_model_id: &str,
        allocation_id: &str,
        weights_mb: u64,
    ) {
        let placement = self.default_weight_placement();
        self.mark_resident_with_placement(logical_model_id, allocation_id, weights_mb, placement);
    }

    pub fn mark_evicted(&self, model_id: &str) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Eviction is allocation-owner exact. Logical-model-wide removal can
        // erase a sibling cache/process/voice lease that still owns weights.
        state.resident_models.remove(model_id);
        self.machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .resident_models
            .remove(&(self.owner_id, model_id.to_string()));
        self.resident_activity_leases
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .remove(model_id);
    }

    /// Enter fail-closed process teardown. Residency remains charged while a
    /// background owner retains and reaps the child.
    pub fn mark_teardown_pending(&self, model_id: &str) {
        self.mark_teardown_pending_allocation(model_id, model_id);
    }

    /// Quarantine one concrete process allocation while retaining the logical
    /// model identity used by maintenance and admission. Multiple process
    /// owners for one model are reference-counted independently.
    pub fn mark_teardown_pending_allocation(&self, logical_model_id: &str, allocation_id: &str) {
        self.mark_teardown_pending_allocation_with_charge(logical_model_id, allocation_id, 0);
    }

    /// Transfer an exact process allocation into quarantine. If it has not yet
    /// published residency (the pre-ACK cancellation case), `measured_bytes`
    /// supplies the physical charge that must outlive the caller reservation.
    /// Repeating this for the same allocation is idempotent.
    pub fn mark_teardown_pending_allocation_with_charge(
        &self,
        logical_model_id: &str,
        allocation_id: &str,
        measured_bytes: u64,
    ) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, logical_model_id);
        state
            .pending_teardown_models
            .entry(model_id.clone())
            .or_default()
            .insert(allocation_id.to_string());
        let mut machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let key = (self.owner_id, allocation_id.to_string());
        let machine_resident = machine.resident_models.remove(&key);
        let scoped_resident = state.resident_models.get(allocation_id).cloned();
        let resident = machine_resident.or(scoped_resident);
        let measured_mb = measured_bytes.div_ceil(1024 * 1024);
        let placement = self.default_weight_placement();
        machine
            .pending_allocations
            .entry(key)
            .and_modify(|pending| {
                pending.weights_mb = pending.weights_mb.max(measured_mb);
                pending.logical_model_id = model_id.clone();
            })
            .or_insert_with(|| {
                resident.unwrap_or(ResidentAllocation {
                    weights_mb: measured_mb,
                    placement,
                    logical_model_id: model_id,
                })
            });
    }

    /// Clear a quarantined allocation only after its OS process was reaped.
    pub fn finish_teardown(&self, model_id: &str) {
        self.finish_teardown_allocation(model_id, model_id);
    }

    /// Finish one concrete process allocation after confirmed OS exit. This
    /// removes only that owner's resident record; sibling allocations for the
    /// same logical model remain charged and pending until their own ACK.
    pub fn finish_teardown_allocation(&self, logical_model_id: &str, allocation_id: &str) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let logical_model_id = Self::resolve_model_identity(&state, logical_model_id);
        let exact_pending = state
            .pending_teardown_models
            .get_mut(&logical_model_id)
            .is_some_and(|pending| pending.remove(allocation_id));
        if exact_pending
            && state
                .pending_teardown_models
                .get(&logical_model_id)
                .is_some_and(HashSet::is_empty)
        {
            state.pending_teardown_models.remove(&logical_model_id);
        }
        if !exact_pending {
            return;
        }
        state.resident_models.remove(allocation_id);
        let mut machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        machine
            .resident_models
            .remove(&(self.owner_id, allocation_id.to_string()));
        machine
            .pending_allocations
            .remove(&(self.owner_id, allocation_id.to_string()));
        self.resident_activity_leases
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .remove(allocation_id);
    }

    pub fn teardown_pending(&self, model_id: &str) -> bool {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, model_id);
        state.pending_teardown_models.contains_key(&model_id)
    }

    pub fn is_resident(&self, model_id: &str) -> bool {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, model_id);
        state
            .resident_models
            .iter()
            .any(|(allocation_id, resident)| {
                allocation_id == &model_id || resident.logical_model_id == model_id
            })
    }

    pub fn resident_model_mb(&self) -> u64 {
        let machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        machine
            .resident_models
            .values()
            .chain(machine.pending_allocations.values())
            .map(|resident| resident.weights_mb)
            .fold(0, u64::saturating_add)
    }

    /// Test support for asserting one coordinator's accounting without
    /// observing unrelated coordinators running in parallel.
    #[doc(hidden)]
    pub fn owned_resident_model_mb_for_testing(&self) -> u64 {
        let machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        machine
            .resident_models
            .iter()
            .chain(machine.pending_allocations.iter())
            .filter(|((owner_id, _), _)| *owner_id == self.owner_id)
            .map(|(_, resident)| resident.weights_mb)
            .fold(0, u64::saturating_add)
    }

    pub fn active_request_count(&self, model_id: &str) -> usize {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, model_id);
        state
            .active_by_model
            .get(&model_id)
            .copied()
            .unwrap_or_default()
    }

    /// The logical models this controller holds resident, with the megabytes
    /// each holds, largest first (ties by id, so the order is stable).
    pub fn resident_logical_models(&self) -> Vec<(String, u64)> {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut by_model: HashMap<String, u64> = HashMap::new();
        for resident in state.resident_models.values() {
            let held = by_model
                .entry(resident.logical_model_id.clone())
                .or_default();
            *held = held.saturating_add(resident.weights_mb);
        }
        let mut models = by_model.into_iter().collect::<Vec<_>>();
        models.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
        models
    }

    /// Exact resident allocation keys for a logical model. Engine caches use
    /// the logical ID directly; independent voice/provider instances use
    /// allocation-qualified keys while retaining the same logical alias.
    pub fn resident_allocation_ids(&self, model_id: &str) -> Vec<String> {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, model_id);
        let mut allocations = state
            .resident_models
            .iter()
            .filter(|(allocation_id, resident)| {
                allocation_id.as_str() == model_id || resident.logical_model_id == model_id
            })
            .map(|(allocation_id, _)| allocation_id.clone())
            .collect::<Vec<_>>();
        if let Some(pending) = state.pending_teardown_models.get(&model_id) {
            allocations.extend(pending.iter().cloned());
        }
        allocations.sort();
        allocations.dedup();
        allocations
    }

    /// Whether model weights are charged to host memory (every Mac, and any
    /// machine without a CUDA GPU of known size), where the configured
    /// allocation is the ceiling that refuses a load.
    pub fn places_weights_on_host(&self) -> bool {
        self.default_weight_placement() == WeightPlacement::Host
    }

    fn default_weight_placement(&self) -> WeightPlacement {
        if matches!(self.hardware.gpu_backend, GpuBackend::Cuda)
            && self.hardware.gpu_memory_mb.is_some()
        {
            WeightPlacement::Accelerator
        } else {
            WeightPlacement::Host
        }
    }

    fn mark_resident_with_placement(
        &self,
        logical_model_id: &str,
        allocation_id: &str,
        weights_mb: u64,
        placement: WeightPlacement,
    ) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let logical_model_id = Self::resolve_model_identity(&state, logical_model_id);
        let allocation = ResidentAllocation {
            weights_mb,
            placement,
            logical_model_id: logical_model_id.clone(),
        };
        state
            .resident_models
            .entry(allocation_id.to_string())
            .and_modify(|resident| {
                resident.weights_mb = resident.weights_mb.max(weights_mb);
                resident.placement = placement;
                resident.logical_model_id = logical_model_id.clone();
            })
            .or_insert(ResidentAllocation {
                weights_mb,
                placement,
                logical_model_id,
            });
        // Preserve the global lock order: scoped state, then machine ledger.
        self.machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .resident_models
            .insert((self.owner_id, allocation_id.to_string()), allocation);
    }

    pub fn preflight(&self, model: &ModelSchema, context_tokens: usize) -> LocalLoadPreflight {
        let live_available_mb = self.live_probe.available_memory_mb().ok().flatten();
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        self.preflight_locked(model, context_tokens, &state, &machine, live_available_mb)
    }

    pub fn reserve(
        self: &Arc<Self>,
        model: &ModelSchema,
        context_tokens: usize,
    ) -> Result<LocalLoadReservation, LocalAdmissionError> {
        let live_available_mb = self.live_probe.available_memory_mb().ok().flatten();
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, &model.id);
        let estimate = self.model_memory_estimate(model, context_tokens);
        let mut preflight = self.preflight_estimate_locked(
            &model_id,
            estimate,
            self.default_weight_placement(),
            &state,
            &machine,
            live_available_mb,
        );
        if state.pending_teardown_models.contains_key(&model_id) {
            preflight.verdict = LocalLoadVerdict::PendingTeardown;
            return Err(LocalAdmissionError { preflight });
        }
        if state.maintenance_models.contains(&model_id) {
            preflight.verdict = LocalLoadVerdict::ModelMaintenance;
            return Err(LocalAdmissionError { preflight });
        }
        if !preflight.verdict.permits_static_fallback() {
            return Err(LocalAdmissionError { preflight });
        }
        if preflight.verdict == LocalLoadVerdict::LiveMemoryUnknown {
            tracing::warn!(
                model = %model.id,
                configured_ceiling_mb = preflight.configured_ceiling_mb,
                estimated_incremental_mb = preflight.estimated_incremental_mb,
                "live memory is unknown; proceeding under the static configured ceiling only"
            );
        }
        state.active_host_reservations_mb = state
            .active_host_reservations_mb
            .saturating_add(preflight.estimated_incremental_mb);
        state.active_accelerator_reservations_mb = state
            .active_accelerator_reservations_mb
            .saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
        let host = machine
            .active_host_by_owner
            .entry(self.owner_id)
            .or_default();
        *host = host.saturating_add(preflight.estimated_incremental_mb);
        let accelerator = machine
            .active_accelerator_by_owner
            .entry(self.owner_id)
            .or_default();
        *accelerator =
            accelerator.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
        *state.active_by_model.entry(model_id.clone()).or_default() += 1;
        state.next_request_id = state.next_request_id.wrapping_add(1);
        let request_id = state.next_request_id;
        Ok(LocalLoadReservation {
            request_id,
            model_id: model_id.clone(),
            maintenance_model_id: model_id.clone(),
            weights_mb: preflight.estimate.weights_mb,
            reserved_incremental_mb: preflight.estimated_incremental_mb,
            reserved_accelerator_mb: preflight.accelerator_incremental_mb.unwrap_or_default(),
            cold_weights_reserved: !Self::has_resident_model(&state, &model_id),
            placement: self.default_weight_placement(),
            coordinator: Arc::clone(self),
            charge: Arc::new(ReservationCharge::new(
                model_id,
                preflight.estimated_incremental_mb,
                preflight.accelerator_incremental_mb.unwrap_or_default(),
                Arc::clone(self),
            )),
        })
    }

    /// Reserve an installed host-memory allocation (Whisper/ONNX voice paths).
    /// The caller downloads first, measures the resulting artifact, then calls
    /// this immediately before constructing the native runtime.
    pub fn reserve_measured_host(
        self: &Arc<Self>,
        model_id: &str,
        measured_weights_bytes: u64,
        request_overhead_mb: u64,
    ) -> Result<LocalLoadReservation, LocalAdmissionError> {
        self.reserve_measured(
            model_id,
            model_id,
            measured_weights_bytes,
            request_overhead_mb,
            WeightPlacement::Host,
        )
    }

    pub fn reserve_measured_host_allocation(
        self: &Arc<Self>,
        logical_model_id: &str,
        allocation_id: &str,
        measured_weights_bytes: u64,
        request_overhead_mb: u64,
    ) -> Result<LocalLoadReservation, LocalAdmissionError> {
        self.reserve_measured(
            logical_model_id,
            allocation_id,
            measured_weights_bytes,
            request_overhead_mb,
            WeightPlacement::Host,
        )
    }

    fn reserve_measured(
        self: &Arc<Self>,
        logical_model_id: &str,
        allocation_id: &str,
        measured_weights_bytes: u64,
        request_overhead_mb: u64,
        placement: WeightPlacement,
    ) -> Result<LocalLoadReservation, LocalAdmissionError> {
        let weights_mb = measured_weights_bytes.div_ceil(1024 * 1024);
        let estimate = ModelMemoryEstimate {
            weights_mb,
            runtime_overhead_mb: request_overhead_mb,
            context_overhead_mb: 0,
            transient_margin_mb: 0,
            estimated_peak_mb: weights_mb.saturating_add(request_overhead_mb),
            evidence: ModelResourceEvidence::FileSystemMeasured,
        };
        let live_available_mb = self.live_probe.available_memory_mb().ok().flatten();
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut machine = self
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let logical_model_id = Self::resolve_model_identity(&state, logical_model_id);
        let mut preflight = self.preflight_estimate_locked(
            allocation_id,
            estimate,
            placement,
            &state,
            &machine,
            live_available_mb,
        );
        if state
            .pending_teardown_models
            .contains_key(&logical_model_id)
        {
            preflight.verdict = LocalLoadVerdict::PendingTeardown;
        }
        if state.maintenance_models.contains(&logical_model_id) {
            preflight.verdict = LocalLoadVerdict::ModelMaintenance;
        }
        if !preflight.verdict.permits_static_fallback() {
            return Err(LocalAdmissionError { preflight });
        }
        state.active_host_reservations_mb = state
            .active_host_reservations_mb
            .saturating_add(preflight.estimated_incremental_mb);
        state.active_accelerator_reservations_mb = state
            .active_accelerator_reservations_mb
            .saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
        let host = machine
            .active_host_by_owner
            .entry(self.owner_id)
            .or_default();
        *host = host.saturating_add(preflight.estimated_incremental_mb);
        let accelerator = machine
            .active_accelerator_by_owner
            .entry(self.owner_id)
            .or_default();
        *accelerator =
            accelerator.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
        *state
            .active_by_model
            .entry(logical_model_id.clone())
            .or_default() += 1;
        state.next_request_id = state.next_request_id.wrapping_add(1);
        Ok(LocalLoadReservation {
            request_id: state.next_request_id,
            model_id: allocation_id.to_string(),
            maintenance_model_id: logical_model_id.clone(),
            weights_mb,
            reserved_incremental_mb: preflight.estimated_incremental_mb,
            reserved_accelerator_mb: preflight.accelerator_incremental_mb.unwrap_or_default(),
            cold_weights_reserved: !state.resident_models.contains_key(allocation_id),
            placement,
            coordinator: Arc::clone(self),
            charge: Arc::new(ReservationCharge::new(
                logical_model_id,
                preflight.estimated_incremental_mb,
                preflight.accelerator_incremental_mb.unwrap_or_default(),
                Arc::clone(self),
            )),
        })
    }

    /// Exclude new reservations for one model while removal/maintenance checks
    /// cache and cross-process lease state. The check and exclusion happen
    /// under the same mutex as `reserve`, closing the removal-vs-load race.
    pub fn begin_model_maintenance(
        self: &Arc<Self>,
        model_id: &str,
    ) -> Result<LocalModelMaintenanceGuard, ModelMaintenanceError> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let model_id = Self::resolve_model_identity(&state, model_id);
        if state
            .active_by_model
            .get(&model_id)
            .copied()
            .unwrap_or_default()
            > 0
        {
            return Err(ModelMaintenanceError::ModelInUse(model_id));
        }
        if !state.maintenance_models.insert(model_id.clone()) {
            return Err(ModelMaintenanceError::AlreadyInMaintenance(model_id));
        }
        Ok(LocalModelMaintenanceGuard {
            model_id,
            coordinator: Arc::clone(self),
        })
    }

    fn preflight_locked(
        &self,
        model: &ModelSchema,
        context_tokens: usize,
        state: &AdmissionState,
        machine: &MachineAdmissionLedger,
        live_available_mb: Option<u64>,
    ) -> LocalLoadPreflight {
        let estimate = self.model_memory_estimate(model, context_tokens);
        let model_id = Self::resolve_model_identity(state, &model.id);
        self.preflight_estimate_locked(
            &model_id,
            estimate,
            self.default_weight_placement(),
            state,
            machine,
            live_available_mb,
        )
    }

    fn preflight_estimate_locked(
        &self,
        model_id: &str,
        estimate: ModelMemoryEstimate,
        placement: WeightPlacement,
        state: &AdmissionState,
        machine: &MachineAdmissionLedger,
        live_available_mb: Option<u64>,
    ) -> LocalLoadPreflight {
        let policy = self.policy();
        let budget = policy.effective_budget(self.hardware.total_ram_mb);
        let host_resident_mb = machine
            .resident_models
            .values()
            .chain(machine.pending_allocations.values())
            .filter(|resident| resident.placement == WeightPlacement::Host)
            .map(|resident| resident.weights_mb)
            .fold(0, u64::saturating_add);
        let accelerator_resident_weights_mb = machine
            .resident_models
            .values()
            .chain(machine.pending_allocations.values())
            .filter(|resident| resident.placement == WeightPlacement::Accelerator)
            .map(|resident| resident.weights_mb)
            .fold(0, u64::saturating_add);
        let already_resident = Self::has_resident_model(state, model_id);
        let request_overhead_mb = estimate
            .context_overhead_mb
            .saturating_add(estimate.runtime_overhead_mb)
            .saturating_add(estimate.transient_margin_mb);
        let cold_weights_mb = if already_resident {
            0
        } else {
            estimate.weights_mb
        };
        let resident_model_mb = host_resident_mb;
        let host_cold_weights_mb = if placement == WeightPlacement::Host {
            cold_weights_mb
        } else {
            0
        };
        let estimated_incremental_mb = request_overhead_mb.saturating_add(host_cold_weights_mb);
        let accelerator_total_mb = (placement == WeightPlacement::Accelerator)
            .then_some(self.hardware.gpu_memory_mb)
            .flatten();
        let accelerator_resident_mb =
            (placement == WeightPlacement::Accelerator).then_some(accelerator_resident_weights_mb);
        let accelerator_incremental_mb =
            (placement == WeightPlacement::Accelerator).then_some(cold_weights_mb);
        let projected_static_mb = resident_model_mb
            .saturating_add(machine.active_host_by_owner.values().copied().sum::<u64>())
            .saturating_add(estimated_incremental_mb);
        let projected_accelerator_mb = accelerator_resident_weights_mb
            .saturating_add(
                machine
                    .active_accelerator_by_owner
                    .values()
                    .copied()
                    .sum::<u64>(),
            )
            .saturating_add(cold_weights_mb);

        let verdict = if budget.effective_new_load_ceiling_mb == 0 && !already_resident {
            LocalLoadVerdict::DisabledByPolicy
        } else if (budget.configured_model_ceiling_mb != 0
            && projected_static_mb > budget.configured_model_ceiling_mb)
            || accelerator_total_mb.is_some_and(|vram_mb| projected_accelerator_mb > vram_mb)
        {
            LocalLoadVerdict::ExceedsConfiguredCeiling
        } else if let Some(available_mb) = live_available_mb {
            let unreserved_available_mb = available_mb
                .saturating_sub(machine.active_host_by_owner.values().copied().sum::<u64>());
            if unreserved_available_mb
                < estimated_incremental_mb.saturating_add(budget.emergency_reserve_mb)
            {
                LocalLoadVerdict::InsufficientLiveMemory
            } else {
                LocalLoadVerdict::Allowed
            }
        } else {
            LocalLoadVerdict::LiveMemoryUnknown
        };

        LocalLoadPreflight {
            model_id: model_id.to_string(),
            estimate,
            configured_ceiling_mb: budget.configured_model_ceiling_mb,
            resident_model_mb,
            active_reservations_mb: machine.active_host_by_owner.values().copied().sum(),
            estimated_incremental_mb,
            accelerator_total_mb,
            accelerator_resident_mb,
            accelerator_incremental_mb,
            live_available_mb,
            emergency_reserve_mb: budget.emergency_reserve_mb,
            verdict,
        }
    }

    fn has_resident_model(state: &AdmissionState, model_id: &str) -> bool {
        state
            .resident_models
            .iter()
            .any(|(allocation_id, resident)| {
                allocation_id == model_id || resident.logical_model_id == model_id
            })
    }
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct LocalAdmissionError {
    pub preflight: LocalLoadPreflight,
}

#[derive(Clone, Debug, Error, PartialEq, Eq)]
pub enum ModelMaintenanceError {
    #[error("local model '{0}' is in use")]
    ModelInUse(String),
    #[error("local model '{0}' already has maintenance in progress")]
    AlreadyInMaintenance(String),
    #[error("failed to release local model residency: {0}")]
    ReleaseFailed(String),
    #[error("local worker did not acknowledge release of model '{0}'")]
    WorkerReleaseUnacknowledged(String),
    #[error("supervised local process did not acknowledge release of model '{0}'")]
    ProcessReleaseUnacknowledged(String),
    #[error("an in-process cache still has active work for local model '{0}'")]
    CacheReleaseBlocked(String),
    #[error("local model '{model_id}' still has resident allocations: {allocation_ids:?}")]
    ResidualResidency {
        model_id: String,
        allocation_ids: Vec<String>,
    },
}

pub struct LocalModelMaintenanceGuard {
    model_id: String,
    coordinator: Arc<LocalAdmissionCoordinator>,
}

impl Drop for LocalModelMaintenanceGuard {
    fn drop(&mut self) {
        self.coordinator
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .maintenance_models
            .remove(&self.model_id);
    }
}

impl std::fmt::Display for LocalAdmissionError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "local model '{}' blocked by {:?}: needs {} MB incremental, {} MB live available",
            self.preflight.model_id,
            self.preflight.verdict,
            self.preflight.estimated_incremental_mb,
            self.preflight
                .live_available_mb
                .map(|value| value.to_string())
                .unwrap_or_else(|| "unknown".into())
        )
    }
}

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

/// RAII guard for one request's estimated incremental allocation. Dropping on
/// success, error, panic unwind, or cancellation releases the reservation.
pub struct LocalLoadReservation {
    request_id: u64,
    model_id: String,
    maintenance_model_id: String,
    weights_mb: u64,
    reserved_incremental_mb: u64,
    reserved_accelerator_mb: u64,
    cold_weights_reserved: bool,
    placement: WeightPlacement,
    coordinator: Arc<LocalAdmissionCoordinator>,
    charge: Arc<ReservationCharge>,
}

/// Cloneable ownership for native work that outlives its awaiting request.
/// `spawn_blocking` tasks and supervised subprocess cleanup retain this lease
/// after cancellation so machine headroom is not advertised while allocation
/// work is still running.
#[derive(Clone)]
pub struct DetachedLocalLease {
    _charge: Arc<ReservationCharge>,
}

struct ReservationCharge {
    maintenance_model_id: String,
    host_mb: AtomicU64,
    accelerator_mb: AtomicU64,
    cold_weights_transferred: AtomicBool,
    coordinator: Arc<LocalAdmissionCoordinator>,
    activity_lease: Mutex<Option<Arc<crate::model_management::ModelLease>>>,
}

impl ReservationCharge {
    fn new(
        maintenance_model_id: String,
        host_mb: u64,
        accelerator_mb: u64,
        coordinator: Arc<LocalAdmissionCoordinator>,
    ) -> Self {
        Self {
            maintenance_model_id,
            host_mb: AtomicU64::new(host_mb),
            accelerator_mb: AtomicU64::new(accelerator_mb),
            cold_weights_transferred: AtomicBool::new(false),
            coordinator,
            activity_lease: Mutex::new(None),
        }
    }

    fn update(&self, host_mb: u64, accelerator_mb: u64) {
        self.host_mb.store(host_mb, Ordering::Release);
        self.accelerator_mb.store(accelerator_mb, Ordering::Release);
    }
}

impl Drop for ReservationCharge {
    fn drop(&mut self) {
        let host_mb = self.host_mb.load(Ordering::Acquire);
        let accelerator_mb = self.accelerator_mb.load(Ordering::Acquire);
        let mut state = self
            .coordinator
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut machine = self
            .coordinator
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        state.active_host_reservations_mb =
            state.active_host_reservations_mb.saturating_sub(host_mb);
        state.active_accelerator_reservations_mb = state
            .active_accelerator_reservations_mb
            .saturating_sub(accelerator_mb);
        if let Some(host) = machine
            .active_host_by_owner
            .get_mut(&self.coordinator.owner_id)
        {
            *host = host.saturating_sub(host_mb);
        }
        if let Some(accelerator) = machine
            .active_accelerator_by_owner
            .get_mut(&self.coordinator.owner_id)
        {
            *accelerator = accelerator.saturating_sub(accelerator_mb);
        }
        let maintenance_model_id =
            LocalAdmissionCoordinator::resolve_model_identity(&state, &self.maintenance_model_id);
        if let Some(active) = state.active_by_model.get_mut(&maintenance_model_id) {
            *active = active.saturating_sub(1);
            if *active == 0 {
                state.active_by_model.remove(&maintenance_model_id);
            }
        }
    }
}

impl std::fmt::Debug for LocalLoadReservation {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("LocalLoadReservation")
            .field("request_id", &self.request_id)
            .field("model_id", &self.model_id)
            .field("reserved_incremental_mb", &self.reserved_incremental_mb)
            .field("reserved_accelerator_mb", &self.reserved_accelerator_mb)
            .finish_non_exhaustive()
    }
}

impl LocalLoadReservation {
    pub fn request_id(&self) -> u64 {
        self.request_id
    }

    /// Model whose allocation this reservation protects. Callers that walk an
    /// adaptive route can use this to avoid reusing one candidate's reservation
    /// for a different local fallback.
    pub fn model_id(&self) -> &str {
        &self.model_id
    }

    pub fn authorizes_model(&self, model_id: &str) -> bool {
        self.model_id == model_id || self.maintenance_model_id == model_id
    }

    /// Bind an admitted cache load to its concrete allocation owner. Distinct
    /// cache/runtime generations for one logical model must not peer-discount
    /// each other's physical weights.
    pub(crate) fn bind_allocation_id(&mut self, allocation_id: &str) {
        self.model_id = allocation_id.to_string();
    }

    pub fn reserved_incremental_mb(&self) -> u64 {
        self.reserved_incremental_mb
    }

    pub fn reconciled_weights_bytes(&self) -> u64 {
        self.weights_mb.saturating_mul(1024 * 1024)
    }

    pub fn detached_lease(&self) -> DetachedLocalLease {
        DetachedLocalLease {
            _charge: self.charge.clone(),
        }
    }

    pub(crate) fn attach_activity_lease(&mut self, lease: crate::model_management::ModelLease) {
        *self
            .charge
            .activity_lease
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Arc::new(lease));
    }

    /// Atomically replace this request's cold-weight charge with a pending
    /// process allocation. Request overhead remains active until the request
    /// reservation and any detached cleanup lease are released.
    pub(crate) fn transfer_cold_weights_to_pending_allocation(
        &self,
        allocation_id: &str,
        measured_weights_bytes: u64,
    ) {
        let mut state = self
            .coordinator
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let logical_model_id =
            LocalAdmissionCoordinator::resolve_model_identity(&state, &self.maintenance_model_id);
        state
            .pending_teardown_models
            .entry(logical_model_id.clone())
            .or_default()
            .insert(allocation_id.to_string());
        let mut machine = self
            .coordinator
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let key = (self.coordinator.owner_id, allocation_id.to_string());
        let machine_resident = machine.resident_models.remove(&key);
        let scoped_resident = state.resident_models.get(allocation_id).cloned();
        let resident = machine_resident.or(scoped_resident);
        let measured_mb = measured_weights_bytes.div_ceil(1024 * 1024);
        machine
            .pending_allocations
            .entry(key)
            .and_modify(|pending| {
                pending.weights_mb = pending.weights_mb.max(measured_mb);
                pending.logical_model_id = logical_model_id.clone();
            })
            .or_insert_with(|| {
                resident.unwrap_or(ResidentAllocation {
                    weights_mb: measured_mb,
                    placement: self.placement,
                    logical_model_id,
                })
            });

        if self.cold_weights_reserved
            && self
                .charge
                .cold_weights_transferred
                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
        {
            match self.placement {
                WeightPlacement::Host => {
                    state.active_host_reservations_mb = state
                        .active_host_reservations_mb
                        .saturating_sub(self.weights_mb);
                    let active = machine
                        .active_host_by_owner
                        .entry(self.coordinator.owner_id)
                        .or_default();
                    *active = active.saturating_sub(self.weights_mb);
                    let _ = self.charge.host_mb.fetch_update(
                        Ordering::AcqRel,
                        Ordering::Acquire,
                        |host_mb| Some(host_mb.saturating_sub(self.weights_mb)),
                    );
                }
                WeightPlacement::Accelerator => {
                    state.active_accelerator_reservations_mb = state
                        .active_accelerator_reservations_mb
                        .saturating_sub(self.weights_mb);
                    let active = machine
                        .active_accelerator_by_owner
                        .entry(self.coordinator.owner_id)
                        .or_default();
                    *active = active.saturating_sub(self.weights_mb);
                    let _ = self.charge.accelerator_mb.fetch_update(
                        Ordering::AcqRel,
                        Ordering::Acquire,
                        |accelerator_mb| Some(accelerator_mb.saturating_sub(self.weights_mb)),
                    );
                }
            }
        }
    }

    fn sync_shared_charge(&self) {
        self.charge
            .update(self.reserved_incremental_mb, self.reserved_accelerator_mb);
    }

    /// Replace the catalog estimate with an installed on-disk measurement.
    /// Check and reservation replacement happen under the same coordinator
    /// mutex; rejection leaves the original reservation untouched.
    pub fn reconcile_measured_weights(
        &mut self,
        measured_weights_bytes: u64,
    ) -> Result<LocalLoadPreflight, LocalAdmissionError> {
        if !self.cold_weights_reserved {
            let live_available_mb = self
                .coordinator
                .live_probe
                .available_memory_mb()
                .ok()
                .flatten();
            let state = self
                .coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let machine = self
                .coordinator
                .machine_ledger
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if LocalAdmissionCoordinator::has_resident_model(&state, &self.model_id) {
                let estimate = self.measured_estimate(self.weights_mb);
                return Ok(self.coordinator.preflight_estimate_locked(
                    &self.model_id,
                    estimate,
                    self.placement,
                    &state,
                    &machine,
                    live_available_mb,
                ));
            }
            drop(state);
            self.cold_weights_reserved = true;
        }
        let measured_mb = measured_weights_bytes.div_ceil(1024 * 1024);
        let live_available_mb = self
            .coordinator
            .live_probe
            .available_memory_mb()
            .ok()
            .flatten();
        let mut state = self
            .coordinator
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut machine = self
            .coordinator
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let published_by_peer =
            LocalAdmissionCoordinator::has_resident_model(&state, &self.model_id);
        let mut without_current = state.clone();
        without_current.active_host_reservations_mb = without_current
            .active_host_reservations_mb
            .saturating_sub(self.reserved_incremental_mb);
        without_current.active_accelerator_reservations_mb = without_current
            .active_accelerator_reservations_mb
            .saturating_sub(self.reserved_accelerator_mb);
        let mut without_machine = machine.clone();
        let host = without_machine
            .active_host_by_owner
            .entry(self.coordinator.owner_id)
            .or_default();
        *host = host.saturating_sub(self.reserved_incremental_mb);
        let accelerator = without_machine
            .active_accelerator_by_owner
            .entry(self.coordinator.owner_id)
            .or_default();
        *accelerator = accelerator.saturating_sub(self.reserved_accelerator_mb);
        let estimate = self.measured_estimate(measured_mb);
        let preflight = self.coordinator.preflight_estimate_locked(
            &self.model_id,
            estimate,
            self.placement,
            &without_current,
            &without_machine,
            live_available_mb,
        );
        if !preflight.verdict.permits_static_fallback() {
            return Err(LocalAdmissionError { preflight });
        }
        state.active_host_reservations_mb = without_current
            .active_host_reservations_mb
            .saturating_add(preflight.estimated_incremental_mb);
        state.active_accelerator_reservations_mb = without_current
            .active_accelerator_reservations_mb
            .saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
        machine.active_host_by_owner.insert(
            self.coordinator.owner_id,
            without_machine
                .active_host_by_owner
                .get(&self.coordinator.owner_id)
                .copied()
                .unwrap_or_default()
                .saturating_add(preflight.estimated_incremental_mb),
        );
        machine.active_accelerator_by_owner.insert(
            self.coordinator.owner_id,
            without_machine
                .active_accelerator_by_owner
                .get(&self.coordinator.owner_id)
                .copied()
                .unwrap_or_default()
                .saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default()),
        );
        self.weights_mb = measured_mb;
        self.reserved_incremental_mb = preflight.estimated_incremental_mb;
        self.reserved_accelerator_mb = preflight.accelerator_incremental_mb.unwrap_or_default();
        if published_by_peer {
            self.cold_weights_reserved = false;
        }
        self.sync_shared_charge();
        Ok(preflight)
    }

    fn measured_estimate(&self, weights_mb: u64) -> ModelMemoryEstimate {
        let host_cold_mb = if self.cold_weights_reserved && self.placement == WeightPlacement::Host
        {
            self.weights_mb
        } else {
            0
        };
        let request_overhead_mb = self.reserved_incremental_mb.saturating_sub(host_cold_mb);
        ModelMemoryEstimate {
            weights_mb,
            runtime_overhead_mb: request_overhead_mb,
            context_overhead_mb: 0,
            transient_margin_mb: 0,
            estimated_peak_mb: weights_mb.saturating_add(request_overhead_mb),
            evidence: ModelResourceEvidence::FileSystemMeasured,
        }
    }

    /// Atomically transfer the cold portion of this request reservation into
    /// resident accounting at the exact point a cache publishes the backend.
    pub fn publish_resident_weights(&mut self, measured_weights_bytes: u64) {
        let allocation_id = self.model_id.clone();
        self.publish_resident_weights_as(&allocation_id, measured_weights_bytes);
    }

    /// Conservatively account an external allocation whose acknowledgement
    /// reported a different concrete allocation id than the requested model.
    /// This is only used when the external owner cannot confirm teardown.
    pub fn publish_resident_weights_as(
        &mut self,
        allocation_id: &str,
        measured_weights_bytes: u64,
    ) {
        // Publication happens after the allocation already exists. It must not
        // run another live-memory probe or reject at that point: doing so can
        // return an error while a cache/process remains resident but
        // unaccounted. The caller reconciles before loading; if an external
        // allocator reports a different final size, conservatively account the
        // actual size atomically even when it exceeds the earlier reservation.
        let measured_mb = measured_weights_bytes.div_ceil(1024 * 1024);
        let mut state = self
            .coordinator
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut machine = self
            .coordinator
            .machine_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let key = (self.coordinator.owner_id, allocation_id.to_string());
        let pending = machine.pending_allocations.remove(&key);
        let logical_model_id =
            LocalAdmissionCoordinator::resolve_model_identity(&state, &self.maintenance_model_id);
        if let Some(allocations) = state.pending_teardown_models.get_mut(&logical_model_id) {
            allocations.remove(allocation_id);
            if allocations.is_empty() {
                state.pending_teardown_models.remove(&logical_model_id);
            }
        }
        if !self.cold_weights_reserved && pending.is_none() {
            return;
        }
        if self.cold_weights_reserved {
            let cold_weights_transferred =
                self.charge.cold_weights_transferred.load(Ordering::Acquire);
            match self.placement {
                WeightPlacement::Host => {
                    if !cold_weights_transferred {
                        state.active_host_reservations_mb = state
                            .active_host_reservations_mb
                            .saturating_sub(self.weights_mb);
                        let active = machine
                            .active_host_by_owner
                            .entry(self.coordinator.owner_id)
                            .or_default();
                        *active = active.saturating_sub(self.weights_mb);
                    }
                    self.reserved_incremental_mb =
                        self.reserved_incremental_mb.saturating_sub(self.weights_mb);
                }
                WeightPlacement::Accelerator => {
                    if !cold_weights_transferred {
                        state.active_accelerator_reservations_mb = state
                            .active_accelerator_reservations_mb
                            .saturating_sub(self.weights_mb);
                        let active = machine
                            .active_accelerator_by_owner
                            .entry(self.coordinator.owner_id)
                            .or_default();
                        *active = active.saturating_sub(self.weights_mb);
                    }
                    self.reserved_accelerator_mb =
                        self.reserved_accelerator_mb.saturating_sub(self.weights_mb);
                }
            }
        }
        let resident = ResidentAllocation {
            weights_mb: pending
                .as_ref()
                .map(|allocation| allocation.weights_mb)
                .unwrap_or_default()
                .max(measured_mb),
            placement: pending
                .as_ref()
                .map(|allocation| allocation.placement)
                .unwrap_or(self.placement),
            logical_model_id,
        };
        state
            .resident_models
            .insert(allocation_id.to_string(), resident.clone());
        machine.resident_models.insert(key, resident.clone());
        if let Some(lease) = self
            .charge
            .activity_lease
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
        {
            self.coordinator
                .resident_activity_leases
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .insert(allocation_id.to_string(), lease);
        }
        self.weights_mb = resident.weights_mb;
        self.cold_weights_reserved = false;
        self.charge
            .cold_weights_transferred
            .store(false, Ordering::Release);
        self.sync_shared_charge();
    }

    /// Transfer cold weight accounting into the resident set after a loader
    /// successfully publishes the backend into its cache.
    pub fn commit_resident_weights(&mut self) {
        self.publish_resident_weights(self.weights_mb * 1024 * 1024);
    }
}

/// What a checkpoint's own `config.json` says about the memory a request
/// costs beyond its weights, reduced to bytes.
///
/// Measured against vllm-mlx 0.6.0 on an M5 Pro (2026-10-26, car#3563
/// follow-up): Qwen3.8-27B 4-bit, whose cache grows 64 KiB per token, and
/// GLM-4.7-Flash 8-bit, whose MLA cache grows 53 KiB per token while its
/// prompt pass peaks near 340 KiB per token. Counting every layer as a
/// full-attention KV layer had overstated the first 4x and understated the
/// second by 8 GiB at 64k tokens.
#[derive(Clone, Debug, PartialEq, Eq)]
struct KvGeometry {
    /// Cache bytes one context token adds, summed over the layers that keep a
    /// per-token cache. A linear-attention layer keeps none.
    kv_bytes_per_token: u64,
    /// Bytes a sequence holds whatever its length: the recurrent and
    /// convolution state of linear-attention layers.
    fixed_state_bytes: u64,
    /// Bytes per prompt token alive at once while the prompt is processed,
    /// beyond the cache itself.
    prefill_bytes_per_token: u64,
    /// The language model's vocabulary, which sizes the logits of a prompt
    /// chunk on runtimes that compute them for every position.
    vocab_size: Option<u64>,
}

/// mlx-lm's and mlx-vlm's `prefill_step_size` default: the prompt is
/// processed in chunks of this many tokens.
const PREFILL_CHUNK_TOKENS: u64 = 2_048;

/// Installed checkpoints' own KV geometry, keyed by model id: the input that
/// lets the recommendation estimate size an installed row's KV cache from its
/// `config.json`, as load-time admission does, instead of from the parameter
/// count. A row absent here — not installed, or a config without readable
/// geometry — keeps the parameter-count fallback, as it does at admission.
/// [`CheckpointGeometry::default`] holds nothing, so every row falls back.
#[derive(Clone, Debug, Default)]
pub struct CheckpointGeometry {
    by_id: HashMap<String, KvGeometry>,
}

impl CheckpointGeometry {
    /// Read each row's `<models_dir>/<name>/config.json`, the file admission
    /// reads. Callers that hold an engine use
    /// [`LocalAdmissionCoordinator::checkpoint_geometry`] instead, which shares
    /// admission's models directory and cache.
    pub fn read<'a>(models: impl IntoIterator<Item = &'a ModelSchema>, models_dir: &Path) -> Self {
        Self {
            by_id: models
                .into_iter()
                .filter_map(|model| {
                    Some((
                        model.id.clone(),
                        kv_geometry_for(model, Some(models_dir), None)?,
                    ))
                })
                .collect(),
        }
    }

    fn get(&self, model: &ModelSchema) -> Option<&KvGeometry> {
        self.by_id.get(&model.id)
    }
}

/// Where multimodal checkpoints nest their language model's settings:
/// `text_config` (gemma-4, Qwen-VL), `llm_config` (InternVL),
/// `language_config` (DeepSeek-VL). Defined here, not beside the MLX loader,
/// because the loader compiles only on Apple Silicon and this file everywhere.
pub(crate) const NESTED_TEXT_CONFIGS: [&str; 3] = ["text_config", "llm_config", "language_config"];

#[derive(Deserialize)]
struct ModelGeometryConfig {
    /// The architecture (`text_config.model_type` for a nested language
    /// model): it decides how an MLA cache is laid out.
    model_type: Option<String>,
    num_hidden_layers: Option<usize>,
    num_key_value_heads: Option<usize>,
    head_dim: Option<usize>,
    hidden_size: Option<usize>,
    num_attention_heads: Option<usize>,
    kv_cache_dtype: Option<String>,
    torch_dtype: Option<String>,
    vocab_size: Option<usize>,
    /// Per-layer attention kind (Qwen3.5+, gemma). A `linear_attention` layer
    /// keeps a fixed-size state, not a per-token cache.
    layer_types: Option<Vec<String>>,
    /// Qwen3.5+'s hybrid period, when `layer_types` is absent: one
    /// full-attention layer in every `full_attention_interval`.
    full_attention_interval: Option<usize>,
    linear_num_key_heads: Option<usize>,
    linear_num_value_heads: Option<usize>,
    linear_key_head_dim: Option<usize>,
    linear_value_head_dim: Option<usize>,
    linear_conv_kernel_dim: Option<usize>,
    /// Multi-head latent attention (DeepSeek-V3, GLM-4.7-Flash): the cache
    /// holds this compressed latent and the rope key, not per-head K and V.
    kv_lora_rank: Option<usize>,
    qk_rope_head_dim: Option<usize>,
    qk_nope_head_dim: Option<usize>,
    v_head_dim: Option<usize>,
}

fn dtype_bytes(dtype: Option<&str>) -> usize {
    match dtype.map(str::to_ascii_lowercase).as_deref() {
        Some("float32" | "fp32" | "f32") => 4,
        Some("float64" | "fp64" | "f64") => 8,
        Some("float8" | "fp8" | "int8" | "uint8" | "i8" | "u8") => 1,
        // MLX-community quantizes weights, while its KV tensors remain fp16 or
        // bf16. Missing and unrecognized dtype declarations therefore retain
        // the safe two-byte default rather than inheriting weight bit width.
        _ => 2,
    }
}

/// What identifies one version of a checkpoint's `config.json`: its path, and
/// a remove, a replace or a re-download changes the rest.
#[derive(Clone, Debug, PartialEq, Eq)]
struct ConfigStamp {
    /// Which file was read: the managed copy or a Hugging Face snapshot, and
    /// a newer snapshot is a different file even at the same size and mtime.
    path: PathBuf,
    modified: Option<std::time::SystemTime>,
    len: u64,
}

impl ConfigStamp {
    fn of(path: &Path) -> Option<Self> {
        let metadata = std::fs::metadata(path).ok()?;
        Some(Self {
            path: path.to_path_buf(),
            modified: metadata.modified().ok(),
            len: metadata.len(),
        })
    }
}

/// The `config.json` of the checkpoint a load would serve: the managed
/// directory when it holds one, else the Hugging Face snapshot that
/// `ensure_local` reuses for an MLX or managed vllm-mlx row instead of
/// copying it (`registry::try_reuse_local`). Without the second, a row served
/// from the shared cache was sized from its parameter count: Qwen3.8-27B at
/// 51 GiB for a 64k context it serves in 24.
fn kv_config_path(model: &ModelSchema, models_dir: &Path, hf_hub: Option<&Path>) -> PathBuf {
    let managed = models_dir.join(&model.name).join("config.json");
    let (Some(hub), false) = (hf_hub, managed.exists()) else {
        return managed;
    };
    let repo = match &model.source {
        crate::schema::ModelSource::Mlx { hf_repo, .. }
        | crate::schema::ModelSource::ManagedVllmMlx { hf_repo, .. } => hf_repo,
        _ => return managed,
    };
    crate::registry::latest_huggingface_repo_snapshot_in(&crate::hf_cache::repo_dir_in(hub, repo))
        .map(|snapshot| snapshot.join("config.json"))
        .filter(|config| config.exists())
        .unwrap_or(managed)
}

/// Resolve KV geometry from `<models_dir>/<model.name>/config.json`.
///
/// The config may explicitly state `kv_cache_dtype` (or the model-wide
/// `torch_dtype`); otherwise KV elements default to two bytes for fp16/bf16.
/// Missing, unreadable, malformed, or incomplete configs have no geometry and
/// use the conservative parameter-count fallback below.
///
/// A multimodal checkpoint (gemma-4's `gemma4_unified`) keeps its language
/// model's geometry under a nested config, so when the top level has no layer
/// count, the nested one is read — the same keys the loader reads.
fn kv_geometry_for(
    model: &ModelSchema,
    models_dir: Option<&Path>,
    hf_hub: Option<&Path>,
) -> Option<KvGeometry> {
    let bytes = std::fs::read(kv_config_path(model, models_dir?, hf_hub)).ok()?;
    let raw: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
    let source = if raw.get("num_hidden_layers").is_some() {
        &raw
    } else {
        NESTED_TEXT_CONFIGS
            .iter()
            .find_map(|key| {
                raw.get(*key)
                    .filter(|v| v.get("num_hidden_layers").is_some())
            })
            .unwrap_or(&raw)
    };
    let mut config: ModelGeometryConfig = serde_json::from_value(source.clone()).ok()?;
    // The KV dtype may be declared once, at the top.
    if config.kv_cache_dtype.is_none() && config.torch_dtype.is_none() {
        config.torch_dtype = raw
            .get("torch_dtype")
            .and_then(serde_json::Value::as_str)
            .map(str::to_string);
    }
    geometry_from_config(&config)
}

/// mlx-lm architectures whose MLA cache holds the compressed latent and the
/// rope key (`cache.update_and_fetch(kv_latent, k_pe)`), checked in mlx-lm
/// 0.31.3. Another MLA architecture may cache expanded per-head keys and
/// values (minicpm3 does), so it is sized that way: the larger figure.
const LATENT_CACHE_MLA: [&str; 5] = [
    "deepseek_v3",
    "deepseek_v32",
    "glm4_moe_lite",
    "kimi_linear",
    "longcat_flash",
];

/// Product of byte factors that saturates instead of wrapping: a malformed
/// `config.json` must read as enormous, never as a tiny estimate that admits
/// the load.
fn bytes(factors: &[u64]) -> u64 {
    factors
        .iter()
        .fold(1_u64, |product, factor| product.saturating_mul(*factor))
}

fn geometry_from_config(config: &ModelGeometryConfig) -> Option<KvGeometry> {
    let positive = |value: Option<usize>| value.filter(|value| *value > 0).map(|v| v as u64);
    let num_hidden_layers = positive(config.num_hidden_layers)?;
    let element = dtype_bytes(
        config
            .kv_cache_dtype
            .as_deref()
            .or(config.torch_dtype.as_deref()),
    ) as u64;
    // Four fp16 `[tokens × hidden]` tensors conservatively alive together
    // during prompt prefill, additive to the cache.
    let activations = positive(config.hidden_size).map_or(0, |hidden| bytes(&[hidden, 2, 4]));

    // A linear-attention layer keeps a fixed state, not a per-token cache,
    // and is left out of the cache only when that state can be sized:
    // without its dimensions it stays counted as a cache layer, which errs
    // high. Sliding layers stay counted in full too: capping them at their
    // window is not measured yet.
    let linear_dims = (
        positive(config.linear_num_key_heads),
        positive(config.linear_num_value_heads),
        positive(config.linear_key_head_dim),
        positive(config.linear_value_head_dim),
    );
    let linear_layers = match (
        &config.layer_types,
        positive(config.full_attention_interval),
        linear_dims,
    ) {
        (_, _, (None, _, _, _) | (_, None, _, _) | (_, _, None, _) | (_, _, _, None)) => 0,
        (Some(kinds), _, _) => kinds
            .iter()
            .filter(|kind| *kind == "linear_attention")
            .count() as u64,
        (None, Some(interval), _) => num_hidden_layers - num_hidden_layers / interval,
        (None, None, _) => 0,
    }
    .min(num_hidden_layers);
    let cache_layers = num_hidden_layers - linear_layers;
    let fixed_state_bytes = match linear_dims {
        (Some(key_heads), Some(value_heads), Some(key_dim), Some(value_dim)) => {
            // Gated-delta state in fp32, plus the causal-conv tail over the
            // q/k/v projections in the element dtype.
            let recurrent = bytes(&[value_heads, key_dim, value_dim, 4]);
            let conv_channels =
                bytes(&[key_heads, key_dim, 2]).saturating_add(bytes(&[value_heads, value_dim]));
            let kernel_tail = positive(config.linear_conv_kernel_dim)
                .unwrap_or(4)
                .saturating_sub(1);
            let conv = bytes(&[conv_channels, kernel_tail, element]);
            bytes(&[linear_layers, recurrent.saturating_add(conv)])
        }
        _ => 0,
    };

    let mla = (
        positive(config.kv_lora_rank),
        positive(config.qk_rope_head_dim),
        positive(config.num_attention_heads),
    );
    let (kv_bytes_per_token, prefill_bytes_per_token) = match mla {
        (Some(rank), Some(rope), Some(heads))
            if config
                .model_type
                .as_deref()
                .is_some_and(|kind| LATENT_CACHE_MLA.contains(&kind)) =>
        {
            // The cache holds the latent and the rope key once per layer.
            // The prompt pass expands it per head (mlx-lm's glm4_moe_lite):
            // rope scores, their masked copy, and the attention scores and
            // probabilities of an unfused SDPA, each `heads × chunk` per
            // context token, plus the expanded K and V.
            let expanded = positive(config.qk_nope_head_dim)
                .unwrap_or(0)
                .saturating_add(positive(config.v_head_dim).unwrap_or(0));
            let scratch = bytes(&[4, heads, PREFILL_CHUNK_TOKENS, element])
                .saturating_add(bytes(&[heads, expanded, element]));
            (
                bytes(&[cache_layers, rank.saturating_add(rope), element]),
                activations.saturating_add(scratch),
            )
        }
        (Some(_), Some(rope), Some(heads)) => {
            // MLA on an architecture not known to cache the latent: per-head
            // keys (nope + rope) and values for every head.
            let key = positive(config.qk_nope_head_dim)
                .unwrap_or(0)
                .saturating_add(rope);
            let value = positive(config.v_head_dim).unwrap_or(key);
            (
                bytes(&[cache_layers, heads, key.saturating_add(value), element]),
                activations,
            )
        }
        _ => {
            let kv_heads = positive(config.num_key_value_heads)?;
            let head_dim = positive(config.head_dim).or_else(|| {
                let hidden = positive(config.hidden_size)?;
                let heads = positive(config.num_attention_heads)?;
                hidden.is_multiple_of(heads).then_some(hidden / heads)
            })?;
            (
                bytes(&[cache_layers, 2, kv_heads, head_dim, element]),
                activations,
            )
        }
    };
    Some(KvGeometry {
        kv_bytes_per_token,
        fixed_state_bytes,
        prefill_bytes_per_token,
        vocab_size: positive(config.vocab_size),
    })
}

/// Binary-megabyte cache for a context: per-token cache bytes times tokens,
/// plus the sequence's fixed state.
fn kv_cache_mb_for(geometry: &KvGeometry, context_tokens: usize) -> u64 {
    let bytes = (geometry.kv_bytes_per_token as u128)
        .saturating_mul(context_tokens as u128)
        .saturating_add(geometry.fixed_state_bytes as u128);
    bytes.div_ceil(BYTES_PER_MB).min(u64::MAX as u128) as u64
}

/// Conservative unknown-geometry fallback. Qwen3-8B fp16 KV uses 147,456
/// bytes/token (`2 × 36 layers × 8 KV heads × 128 head_dim × 2 bytes`). Dense
/// 0.6B–14B models sit within roughly 0.8–1.2× of this figure per token; larger
/// active parameter counts scale it upward. Installed, known geometry wins.
fn fallback_kv_cache_mb(model: &ModelSchema, context_tokens: usize) -> u64 {
    let active_billions = model_parameter_billions_active(model) as f64;
    let bytes_per_token = 147_456.0 * (active_billions / 8.0).max(1.0);
    ((bytes_per_token * context_tokens as f64) / BYTES_PER_MB as f64).ceil() as u64
}

/// Memory alive during the prompt pass beyond the cache, additive to it.
fn prefill_activation_mb(geometry: Option<&KvGeometry>, context_tokens: usize) -> u64 {
    let Some(geometry) = geometry else {
        return 0;
    };
    let bytes = (geometry.prefill_bytes_per_token as u128).saturating_mul(context_tokens as u128);
    bytes.div_ceil(BYTES_PER_MB).min(u64::MAX as u128) as u64
}

/// The logits of one prompt chunk, for a model served through mlx-vlm (a
/// managed vllm-mlx row with a drafter runs `--mllm`): it computes them for
/// every chunk position, in fp32 and then bf16. Measured on Qwen3.8-27B as a
/// 2.85 GiB step at any context length; `248,320 × 2048 × 6` is 2.84 GiB.
/// mlx-lm, which serves every other managed row, computes them lazily and
/// never materializes them during prefill.
fn chunk_logits_mb(model: &ModelSchema, geometry: Option<&KvGeometry>) -> u64 {
    let served_by_mlx_vlm = matches!(
        &model.source,
        crate::schema::ModelSource::ManagedVllmMlx {
            draft_hf_repo: Some(_),
            ..
        }
    );
    match geometry.and_then(|geometry| geometry.vocab_size) {
        Some(vocab) if served_by_mlx_vlm => {
            ((vocab as u128) * (PREFILL_CHUNK_TOKENS as u128) * 6).div_ceil(BYTES_PER_MB) as u64
        }
        _ => 0,
    }
}

/// Central recommendation-time memory estimator.
pub fn estimate_model_memory(
    model: &ModelSchema,
    hardware: &HardwareInfo,
    context_tokens: usize,
) -> ModelMemoryEstimate {
    estimate_model_memory_with_measured_weights(model, hardware, context_tokens, None)
}

/// The recommendation estimate for a row whose installed checkpoint may
/// declare its own KV geometry: the KV cache is sized from that geometry when
/// `geometry` holds the row, and from the parameter count otherwise.
pub fn estimate_model_memory_for_checkpoint(
    model: &ModelSchema,
    hardware: &HardwareInfo,
    context_tokens: usize,
    geometry: &CheckpointGeometry,
) -> ModelMemoryEstimate {
    estimate_model_memory_with_geometry(model, hardware, context_tokens, None, geometry.get(model))
}

/// The same estimate with optional physical weight evidence. Registry/load
/// paths can provide a measured value without duplicating the remaining math.
pub fn estimate_model_memory_with_measured_weights(
    model: &ModelSchema,
    hardware: &HardwareInfo,
    context_tokens: usize,
    measured_weights_mb: Option<u64>,
) -> ModelMemoryEstimate {
    estimate_model_memory_with_geometry(model, hardware, context_tokens, measured_weights_mb, None)
}

fn estimate_model_memory_with_geometry(
    model: &ModelSchema,
    hardware: &HardwareInfo,
    context_tokens: usize,
    measured_weights_mb: Option<u64>,
    geometry: Option<&KvGeometry>,
) -> ModelMemoryEstimate {
    let declared = model.cost.ram_mb.or(model.cost.size_mb);
    let (weights_mb, evidence) = if let Some(measured) = measured_weights_mb {
        (measured, ModelResourceEvidence::FileSystemMeasured)
    } else if let Some(declared) = declared {
        (
            declared.max(model.cost.size_mb.unwrap_or(0)),
            ModelResourceEvidence::CatalogExact,
        )
    } else {
        (
            heuristic_weights_mb(model),
            ModelResourceEvidence::Heuristic,
        )
    };
    let kv_cache_mb = geometry
        .map(|geometry| kv_cache_mb_for(geometry, context_tokens))
        .unwrap_or_else(|| fallback_kv_cache_mb(model, context_tokens));
    let context_overhead_mb = kv_cache_mb
        .saturating_add(prefill_activation_mb(geometry, context_tokens))
        .saturating_add(chunk_logits_mb(model, geometry));
    let runtime_overhead_mb = backend_runtime_overhead_mb(hardware);
    let transient_margin_mb = TRANSIENT_ALLOCATION_MARGIN_MB;
    let estimated_peak_mb = weights_mb
        .saturating_add(context_overhead_mb)
        .saturating_add(runtime_overhead_mb)
        .saturating_add(transient_margin_mb);

    ModelMemoryEstimate {
        weights_mb,
        runtime_overhead_mb,
        context_overhead_mb,
        transient_margin_mb,
        estimated_peak_mb,
        evidence,
    }
}

fn backend_runtime_overhead_mb(hardware: &HardwareInfo) -> u64 {
    match hardware.gpu_backend {
        GpuBackend::Metal => METAL_RUNTIME_OVERHEAD_MB,
        GpuBackend::Cuda => CUDA_RUNTIME_OVERHEAD_MB,
        _ => CPU_RUNTIME_OVERHEAD_MB,
    }
}

fn heuristic_weights_mb(model: &ModelSchema) -> u64 {
    let billions = model_parameter_billions_total(model);
    (billions as f64 * 600.0).ceil() as u64
}

pub(crate) fn model_parameter_billions_active(model: &ModelSchema) -> f32 {
    model
        .param_count
        .split_once('(')
        .and_then(|(_, rest)| rest.split_once("active"))
        .and_then(|(number, _)| parse_parameter_billions(number))
        .unwrap_or_else(|| model_parameter_billions_total(model))
}

pub(crate) fn model_parameter_billions_total(model: &ModelSchema) -> f32 {
    parse_parameter_billions(&model.param_count).unwrap_or_else(|| {
        let size_mb = model.size_mb();
        if size_mb > 0 {
            (size_mb as f32 / 600.0).max(0.1)
        } else {
            0.0
        }
    })
}

pub(crate) fn parse_parameter_billions(value: &str) -> Option<f32> {
    let value = value.trim();
    let number: String = value
        .chars()
        .take_while(|character| character.is_ascii_digit() || *character == '.')
        .collect();
    let parsed: f32 = number.parse().ok()?;
    if value[number.len()..]
        .trim_start()
        .to_ascii_lowercase()
        .starts_with('m')
    {
        Some(parsed / 1_000.0)
    } else {
        Some(parsed)
    }
}

/// Evaluate policy math for the hardware CAR actually detected.
pub fn evaluate_resources(hardware: &HardwareInfo, policy: &ResourcePolicy) -> ResourceEvaluation {
    let accelerator_memory = match (hardware.gpu_backend.clone(), hardware.gpu_memory_mb) {
        (GpuBackend::Cuda, Some(total_mb)) => Some(AcceleratorResourceBudget {
            total_mb,
            budget_mb: total_mb,
        }),
        _ => None,
    };

    ResourceEvaluation {
        host_memory: policy.effective_budget(hardware.total_ram_mb),
        accelerator_memory,
    }
}

fn percent_of(total_mb: u64, percent: u64) -> u64 {
    let value = (total_mb as u128).saturating_mul(percent as u128) / 100;
    value.min(u64::MAX as u128) as u64
}

fn minimum_emergency_reserve(total_memory_mb: u64) -> u64 {
    MINIMUM_EMERGENCY_RESERVE_MB.max(percent_of(total_memory_mb, EMERGENCY_RESERVE_PERCENT))
}

/// Persistence errors that callers can distinguish from safe defaulting on a
/// missing or malformed policy document.
#[derive(Debug, Error)]
pub enum ResourcePolicyError {
    #[error("resource policy I/O failed: {0}")]
    Io(#[from] io::Error),
    #[error("resource policy serialization failed: {0}")]
    Serialization(#[from] serde_json::Error),
    #[error("Custom model RAM must be finite, nonnegative, and a 0.5 GB increment; got {0}")]
    InvalidCustomGigabytes(f64),
    #[error("invalid resource policy: {reason}")]
    InvalidPolicy { reason: String },
}

/// Where the effective policy returned by a repository load came from.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ResourcePolicyLoadSource {
    Loaded,
    MissingDefault,
    CorruptDefault,
}

/// A policy plus recoverable evidence about safe defaulting.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct ResourcePolicyLoadEvidence {
    pub policy: ResourcePolicy,
    pub source: ResourcePolicyLoadSource,
    pub warning: Option<String>,
}

pub trait ResourcePolicyRepository: Send + Sync {
    fn load(&self) -> Result<ResourcePolicy, ResourcePolicyError>;
    fn save(&self, policy: &ResourcePolicy) -> Result<(), ResourcePolicyError>;
}

/// Private, atomic resource-policy storage rooted at `CAR_HOME`.
#[derive(Clone, Debug)]
pub struct FileResourcePolicyRepository {
    root: PathBuf,
}

impl Default for FileResourcePolicyRepository {
    fn default() -> Self {
        Self::new(car_home::root_or_relative())
    }
}

/// File under the CAR state root recording that the saved resource policy
/// is the concierge's choice, not the user's.
pub const CONCIERGE_POLICY_MARKER: &str = "model-resource-policy.concierge.json";

#[derive(Debug, Serialize, Deserialize)]
struct ConciergePolicyMarker {
    policy: ResourcePolicy,
    reason: String,
    at: u64,
    /// The concierge undid its own raise: it does not raise again.
    #[serde(default)]
    reverted: bool,
}

/// What the concierge's marker records about the saved policy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConciergeChoice {
    pub policy: ResourcePolicy,
    pub at: u64,
    pub reverted: bool,
}

impl FileResourcePolicyRepository {
    pub fn new(root: PathBuf) -> Self {
        Self { root }
    }

    /// The policy the concierge last saved, if the saved policy was its
    /// choice. A user's save clears this ([`Self::clear_concierge_marker`]),
    /// so the saved policy differing from it, or it being absent while a
    /// policy is saved, means the user chose.
    pub fn concierge_marker(&self) -> Option<ResourcePolicy> {
        self.concierge_choice().map(|choice| choice.policy)
    }

    /// The full marker: the policy, when it was set, and whether it was a
    /// reversal of the concierge's own raise.
    pub fn concierge_choice(&self) -> Option<ConciergeChoice> {
        let bytes = std::fs::read(self.root.join(CONCIERGE_POLICY_MARKER)).ok()?;
        serde_json::from_slice::<ConciergePolicyMarker>(&bytes)
            .ok()
            .map(|marker| ConciergeChoice {
                policy: marker.policy,
                at: marker.at,
                reverted: marker.reverted,
            })
    }

    /// Save `policy` as the concierge's choice: the policy, then the marker.
    pub fn save_concierge_choice(
        &self,
        policy: &ResourcePolicy,
        reason: &str,
        at: u64,
        reverted: bool,
    ) -> Result<(), ResourcePolicyError> {
        ResourcePolicyRepository::save(self, policy)?;
        let marker = ConciergePolicyMarker {
            policy: policy.clone(),
            reason: reason.to_string(),
            at,
            reverted,
        };
        let path = self.root.join(CONCIERGE_POLICY_MARKER);
        let tmp = self.root.join(format!(".{CONCIERGE_POLICY_MARKER}.tmp"));
        let bytes = serde_json::to_vec_pretty(&marker).map_err(io::Error::other)?;
        std::fs::write(&tmp, bytes)?;
        std::fs::rename(&tmp, &path)?;
        Ok(())
    }

    /// The user chose: the saved policy is theirs from now on.
    pub fn clear_concierge_marker(&self) -> Result<(), ResourcePolicyError> {
        match std::fs::remove_file(self.root.join(CONCIERGE_POLICY_MARKER)) {
            Err(error) if error.kind() != io::ErrorKind::NotFound => Err(error.into()),
            _ => Ok(()),
        }
    }

    pub fn path(&self) -> PathBuf {
        self.root.join(RESOURCE_POLICY_FILE)
    }

    /// Load the effective policy without losing evidence that CAR recovered
    /// from a missing or invalid source document.
    pub fn load_with_evidence(&self) -> Result<ResourcePolicyLoadEvidence, ResourcePolicyError> {
        let path = self.path();
        let file = match open_resource_policy(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == io::ErrorKind::NotFound => {
                return Ok(ResourcePolicyLoadEvidence {
                    policy: ResourcePolicy::everyday(),
                    source: ResourcePolicyLoadSource::MissingDefault,
                    warning: None,
                });
            }
            Err(error) => {
                if std::fs::symlink_metadata(&path)
                    .is_ok_and(|metadata| !metadata.is_file() || metadata.file_type().is_symlink())
                {
                    return Ok(corrupt_default(
                        "The saved resource policy could not be loaded because it is not a regular file.",
                    ));
                }
                return Err(error.into());
            }
        };
        let metadata = file.metadata()?;
        if !metadata.is_file() {
            return Ok(corrupt_default(
                "The saved resource policy could not be loaded because it is not a regular file.",
            ));
        }
        if metadata.len() > MAX_POLICY_BYTES {
            return Ok(corrupt_default(
                "The saved resource policy could not be loaded because it exceeds the size limit.",
            ));
        }

        let mut raw = Vec::new();
        file.take(MAX_POLICY_BYTES + 1).read_to_end(&mut raw)?;
        if raw.len() as u64 > MAX_POLICY_BYTES {
            return Ok(corrupt_default(
                "The saved resource policy could not be loaded because it exceeds the size limit.",
            ));
        }
        let policy = match serde_json::from_slice::<ResourcePolicy>(&raw) {
            Ok(policy) => policy,
            Err(error) => {
                return Ok(corrupt_default(format!(
                    "The saved resource policy could not be loaded: {error}"
                )));
            }
        };
        if let Err(error) = policy.validate() {
            return Ok(corrupt_default(format!(
                "The saved resource policy could not be loaded: {error}"
            )));
        }

        Ok(ResourcePolicyLoadEvidence {
            policy,
            source: ResourcePolicyLoadSource::Loaded,
            warning: None,
        })
    }
}

fn open_resource_policy(path: &Path) -> io::Result<std::fs::File> {
    let mut options = std::fs::OpenOptions::new();
    options.read(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.custom_flags(libc::O_NOFOLLOW);
    }
    options.open(path)
}

fn corrupt_default(warning: impl Into<String>) -> ResourcePolicyLoadEvidence {
    ResourcePolicyLoadEvidence {
        policy: ResourcePolicy::everyday(),
        source: ResourcePolicyLoadSource::CorruptDefault,
        warning: Some(warning.into()),
    }
}

impl ResourcePolicyRepository for FileResourcePolicyRepository {
    fn load(&self) -> Result<ResourcePolicy, ResourcePolicyError> {
        Ok(self.load_with_evidence()?.policy)
    }

    fn save(&self, policy: &ResourcePolicy) -> Result<(), ResourcePolicyError> {
        policy.validate()?;
        let _guard = mutation_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        ensure_private_directory(&self.root)?;

        let temp_path = unique_temp_path(&self.root);
        let result = (|| {
            let body = serde_json::to_vec_pretty(policy)?;
            let mut file = open_private_temp(&temp_path)?;
            file.write_all(&body)?;
            file.sync_all()?;
            atomic_replace(&temp_path, &self.path())?;
            car_secrets::harden_owner_only(&self.path());
            sync_directory(&self.root)?;
            Ok(())
        })();
        if result.is_err() {
            let _ = std::fs::remove_file(&temp_path);
        }
        result
    }
}

fn mutation_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

fn unique_temp_path(root: &Path) -> PathBuf {
    static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(0);
    let sequence = NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
    let epoch_nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| duration.as_nanos())
        .unwrap_or(0);
    root.join(format!(
        ".{RESOURCE_POLICY_FILE}.{}.{}.{}.tmp",
        std::process::id(),
        epoch_nanos,
        sequence
    ))
}

fn ensure_private_directory(path: &Path) -> io::Result<()> {
    std::fs::create_dir_all(path)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
    }
    car_secrets::harden_owner_only(path);
    Ok(())
}

#[cfg(unix)]
fn open_private_temp(path: &Path) -> io::Result<std::fs::File> {
    use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};

    let file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .mode(0o600)
        .open(path)?;
    file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
    Ok(file)
}

#[cfg(not(unix))]
fn open_private_temp(path: &Path) -> io::Result<std::fs::File> {
    let file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(path)?;
    car_secrets::harden_owner_only(path);
    Ok(file)
}

#[cfg(not(windows))]
fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> {
    std::fs::rename(source, destination)
}

#[cfg(windows)]
fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> {
    use std::os::windows::ffi::OsStrExt;

    const MOVEFILE_REPLACE_EXISTING: u32 = 0x1;
    const MOVEFILE_WRITE_THROUGH: u32 = 0x8;
    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn MoveFileExW(existing: *const u16, new: *const u16, flags: u32) -> i32;
    }
    let source = source
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect::<Vec<_>>();
    let destination = destination
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect::<Vec<_>>();
    // SAFETY: both pointers address NUL-terminated UTF-16 buffers for the
    // duration of the call, and the flags request atomic replacement.
    let replaced = unsafe {
        MoveFileExW(
            source.as_ptr(),
            destination.as_ptr(),
            MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
        )
    };
    if replaced == 0 {
        Err(io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(unix)]
fn sync_directory(path: &Path) -> io::Result<()> {
    std::fs::File::open(path)?.sync_all()
}

#[cfg(not(unix))]
fn sync_directory(_path: &Path) -> io::Result<()> {
    Ok(())
}

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

    fn hardware(
        total_ram_mb: u64,
        gpu_backend: crate::hardware::GpuBackend,
        vram_mb: Option<u64>,
    ) -> crate::hardware::HardwareInfo {
        crate::hardware::HardwareInfo {
            os: "test".into(),
            arch: "test".into(),
            cpu_cores: 8,
            total_ram_mb,
            gpu_backend,
            gpu_memory_mb: vram_mb,
            gpu_devices: Vec::new(),
            recommended_model: "fixture".into(),
            recommended_context: 4_096,
            max_model_mb: total_ram_mb,
        }
    }

    #[test]
    fn profiles_compute_exact_32_gb_budgets() {
        let total = 32 * 1024;

        assert_eq!(
            ResourcePolicy::everyday()
                .effective_budget(total)
                .configured_model_ceiling_mb,
            13_107
        );
        assert_eq!(
            ResourcePolicy::local_focused()
                .effective_budget(total)
                .configured_model_ceiling_mb,
            26_214
        );
        assert_eq!(
            ResourcePolicy::custom_gb(12.5)
                .unwrap()
                .effective_budget(total)
                .configured_model_ceiling_mb,
            12_800
        );
        assert_eq!(
            ResourcePolicy::everyday().recommendation_target_mb(total),
            13_107
        );
    }

    #[test]
    fn model_memory_estimate_keeps_transient_margin_distinct_and_totals_exactly() {
        let catalog = crate::registry::builtin_catalog();
        let model = catalog
            .iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let estimate = estimate_model_memory(
            model,
            &hardware(32 * 1024, GpuBackend::Metal, None),
            RECOMMENDATION_CONTEXT_TOKENS,
        );

        assert_eq!(estimate.evidence, ModelResourceEvidence::CatalogExact);
        assert_eq!(estimate.weights_mb, 2_400);
        assert_eq!(estimate.runtime_overhead_mb, 512);
        assert_eq!(estimate.transient_margin_mb, 1_024);
        assert_eq!(
            estimate.estimated_peak_mb,
            estimate.weights_mb
                + estimate.context_overhead_mb
                + estimate.runtime_overhead_mb
                + estimate.transient_margin_mb
        );

        let measured = estimate_model_memory_with_measured_weights(
            model,
            &hardware(32 * 1024, GpuBackend::Metal, None),
            RECOMMENDATION_CONTEXT_TOKENS,
            Some(2_321),
        );
        assert_eq!(measured.weights_mb, 2_321);
        assert_eq!(measured.evidence, ModelResourceEvidence::FileSystemMeasured);
    }

    fn qwen3_8b_model() -> ModelSchema {
        crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-8b:4bit")
            .unwrap()
    }

    fn qwen3_8b_geometry() -> KvGeometry {
        KvGeometry {
            // 2 × 36 layers × 8 KV heads × 128 head_dim × 2 bytes.
            kv_bytes_per_token: 147_456,
            fixed_state_bytes: 0,
            prefill_bytes_per_token: 4_096 * 2 * 4,
            vocab_size: None,
        }
    }

    #[test]
    fn kv_cache_qwen3_8b_geometry_matches_binary_mb_table() {
        let geometry = qwen3_8b_geometry();
        assert_eq!(kv_cache_mb_for(&geometry, 9_000), 1_266);
        assert_eq!(kv_cache_mb_for(&geometry, 13_096), 1_842);
        assert_eq!(prefill_activation_mb(Some(&geometry), 9_000), 282);
    }

    #[test]
    fn kv_cache_unknown_geometry_uses_conservative_qwen3_8b_floor() {
        let model = qwen3_8b_model();
        let estimate =
            estimate_model_memory(&model, &hardware(32 * 1024, GpuBackend::Metal, None), 9_000);

        assert_eq!(estimate.context_overhead_mb, 1_266);
    }

    #[test]
    fn kv_cache_unknown_geometry_scales_models_larger_than_8b() {
        let mut model = qwen3_8b_model();
        model.param_count = "32B".into();

        let estimate =
            estimate_model_memory(&model, &hardware(64 * 1024, GpuBackend::Metal, None), 9_000);

        assert_eq!(estimate.context_overhead_mb, 5_063);
    }

    #[test]
    fn kv_geometry_installed_config_drives_admission_context_charge() {
        let root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        let model_dir = root.path().join(&model.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(
            model_dir.join("config.json"),
            r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32,"torch_dtype":"bfloat16"}"#,
        )
        .unwrap();
        let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
            ResourcePolicy::local_focused(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            Some(root.path().to_path_buf()),
        );

        let estimate = coordinator.preflight(&model, 9_000).estimate;

        assert_eq!(estimate.context_overhead_mb, 1_548);
    }

    /// The recommendation estimate takes the geometry admission reads, from
    /// the same file and through the same cache, so for an installed row the
    /// two publish one peak and only live memory separates their verdicts
    /// (car#2213).
    #[test]
    fn recommendation_estimate_uses_the_installed_geometry_admission_uses() {
        let root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        let model_dir = root.path().join(&model.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(
            model_dir.join("config.json"),
            r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32,"torch_dtype":"bfloat16"}"#,
        )
        .unwrap();
        let hardware = hardware(32 * 1024, GpuBackend::Metal, None);
        let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
            ResourcePolicy::local_focused(),
            hardware.clone(),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            Some(root.path().to_path_buf()),
        );
        let admitted = coordinator
            .preflight(&model, RECOMMENDATION_CONTEXT_TOKENS)
            .estimate
            .estimated_peak_mb;
        let peak = |geometry: &CheckpointGeometry| {
            estimate_model_memory_for_checkpoint(
                &model,
                &hardware,
                RECOMMENDATION_CONTEXT_TOKENS,
                geometry,
            )
            .estimated_peak_mb
        };

        assert_eq!(
            peak(&CheckpointGeometry::read([&model], root.path())),
            admitted
        );
        assert_eq!(peak(&coordinator.checkpoint_geometry([&model])), admitted);
        // Not installed: the parameter-count fallback, which differs here.
        assert_ne!(peak(&CheckpointGeometry::default()), admitted);
        assert_eq!(
            peak(&CheckpointGeometry::default()),
            estimate_model_memory(&model, &hardware, RECOMMENDATION_CONTEXT_TOKENS)
                .estimated_peak_mb
        );
    }

    #[test]
    fn kv_geometry_reads_a_multimodal_checkpoints_nested_text_config() {
        let root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        let model_dir = root.path().join(&model.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        // Same geometry as the flat test above, nested the way gemma-4 is.
        std::fs::write(
            model_dir.join("config.json"),
            r#"{"model_type":"gemma4_unified","torch_dtype":"bfloat16","vision_config":{},"text_config":{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32}}"#,
        )
        .unwrap();
        let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
            ResourcePolicy::local_focused(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            Some(root.path().to_path_buf()),
        );
        assert_eq!(
            coordinator
                .preflight(&model, 9_000)
                .estimate
                .context_overhead_mb,
            1_548
        );
    }

    #[test]
    fn kv_geometry_cache_forgets_missing_config_after_install() {
        let root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
            ResourcePolicy::local_focused(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            Some(root.path().to_path_buf()),
        );

        let before_install = coordinator.preflight(&model, 9_000).estimate;
        let model_dir = root.path().join(&model.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(
            model_dir.join("config.json"),
            r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32,"torch_dtype":"bfloat16"}"#,
        )
        .unwrap();
        let after_install = coordinator.preflight(&model, 9_000).estimate;

        assert_eq!(before_install.context_overhead_mb, 1_266);
        assert_eq!(after_install.context_overhead_mb, 1_548);
    }

    #[test]
    fn kv_geometry_derives_head_dim_and_memoizes_installed_config() {
        let root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        let model_dir = root.path().join(&model.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        let config_path = model_dir.join("config.json");
        std::fs::write(
            &config_path,
            r#"{"num_hidden_layers":36,"num_key_value_heads":8,"hidden_size":4096,"num_attention_heads":32}"#,
        )
        .unwrap();
        let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
            ResourcePolicy::local_focused(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            Some(root.path().to_path_buf()),
        );

        let first = coordinator.preflight(&model, 9_000).estimate;
        // Same length and mtime: the stamp is unchanged, so the cached
        // geometry stands and the file is not parsed again.
        let modified = std::fs::metadata(&config_path).unwrap().modified().unwrap();
        let len = std::fs::metadata(&config_path).unwrap().len() as usize;
        std::fs::write(&config_path, "x".repeat(len)).unwrap();
        std::fs::File::options()
            .write(true)
            .open(&config_path)
            .unwrap()
            .set_modified(modified)
            .unwrap();
        let second = coordinator.preflight(&model, 9_000).estimate;

        assert_eq!(first.context_overhead_mb, 1_548);
        assert_eq!(second.context_overhead_mb, first.context_overhead_mb);
    }

    #[test]
    fn kv_geometry_cache_is_scoped_to_coordinator_models_dir() {
        let first_root = tempfile::tempdir().unwrap();
        let second_root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        for (root, config) in [
            (
                first_root.path(),
                r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096}"#,
            ),
            (
                second_root.path(),
                r#"{"num_hidden_layers":2,"num_key_value_heads":2,"head_dim":64,"hidden_size":256}"#,
            ),
        ] {
            let model_dir = root.join(&model.name);
            std::fs::create_dir_all(&model_dir).unwrap();
            std::fs::write(model_dir.join("config.json"), config).unwrap();
        }
        let coordinator = |models_dir: &Path| {
            LocalAdmissionCoordinator::with_probe_and_models_dir(
                ResourcePolicy::local_focused(),
                hardware(32 * 1024, GpuBackend::Metal, None),
                Arc::new(FixedLiveMemoryProbe::known(24_000)),
                Some(models_dir.to_path_buf()),
            )
        };

        // Both coordinators look up the same model id. Their distinct results
        // prove geometry memoization belongs to the coordinator/models root,
        // rather than leaking one test or engine's config across the process.
        assert_eq!(
            coordinator(first_root.path())
                .preflight(&model, 9_000)
                .estimate
                .context_overhead_mb,
            1_548
        );
        assert_eq!(
            coordinator(second_root.path())
                .preflight(&model, 9_000)
                .estimate
                .context_overhead_mb,
            27
        );
    }

    #[test]
    fn kv_geometry_malformed_config_falls_back_without_panicking() {
        let root = tempfile::tempdir().unwrap();
        let model = qwen3_8b_model();
        let model_dir = root.path().join(&model.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(model_dir.join("config.json"), b"not json").unwrap();
        let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
            ResourcePolicy::local_focused(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            Some(root.path().to_path_buf()),
        );

        let estimate = coordinator.preflight(&model, 9_000).estimate;

        assert_eq!(estimate.context_overhead_mb, 1_266);
    }

    #[test]
    fn kv_cache_estimate_is_monotone_and_keeps_non_context_terms_fixed() {
        let model = qwen3_8b_model();
        let hardware = hardware(32 * 1024, GpuBackend::Metal, None);
        let short = estimate_model_memory(&model, &hardware, 9_000);
        let long = estimate_model_memory(&model, &hardware, 13_096);

        assert!(long.context_overhead_mb >= short.context_overhead_mb);
        assert_eq!(short.weights_mb, long.weights_mb);
        assert_eq!(short.runtime_overhead_mb, long.runtime_overhead_mb);
        assert_eq!(short.transient_margin_mb, long.transient_margin_mb);
    }

    #[test]
    fn custom_zero_disables_loads_and_overlarge_values_clamp_below_emergency_reserve() {
        let total = 32 * 1024;

        assert_eq!(
            ResourcePolicy::custom_gb(0.0)
                .unwrap()
                .effective_budget(total)
                .effective_new_load_ceiling_mb,
            0
        );
        let result = ResourcePolicy::custom_gb(99.0)
            .unwrap()
            .effective_budget(total);
        assert_eq!(result.emergency_reserve_mb, 3_276);
        assert_eq!(result.configured_model_ceiling_mb, total - 3_276);
        assert!(result.normalization_notice.is_some());
    }

    #[test]
    fn custom_gigabytes_reject_non_half_steps_negative_and_non_finite_values() {
        for invalid in [10.3, -0.5, f64::INFINITY, f64::NEG_INFINITY, f64::NAN] {
            assert!(
                ResourcePolicy::custom_gb(invalid).is_err(),
                "accepted {invalid:?}"
            );
        }

        assert_eq!(
            ResourcePolicy::custom_gb(10.5).unwrap().custom_max_model_mb,
            Some(10_752)
        );
    }

    #[test]
    fn repository_round_trips_exact_half_gb_and_uses_private_atomic_files() {
        let dir = tempfile::tempdir().unwrap();
        let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());

        repository
            .save(&ResourcePolicy::custom_gb(10.5).unwrap())
            .unwrap();

        assert_eq!(
            repository.path(),
            dir.path().join("model-resource-policy.json")
        );
        assert_eq!(repository.load().unwrap().custom_max_model_mb, Some(10_752));
        assert_private_mode(&repository.path(), 0o600);
        assert_private_mode(dir.path(), 0o700);
        assert_no_temp_files(dir.path());
    }

    #[test]
    fn repository_missing_or_corrupt_file_falls_back_without_deleting_source() {
        let dir = tempfile::tempdir().unwrap();
        let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
        assert_eq!(repository.load().unwrap(), ResourcePolicy::everyday());

        let corrupt = br#"{"profile":"custom","custom_max_model_mb":"broken"}"#;
        std::fs::write(repository.path(), corrupt).unwrap();

        assert_eq!(repository.load().unwrap(), ResourcePolicy::everyday());
        assert_eq!(std::fs::read(repository.path()).unwrap(), corrupt);
    }

    #[test]
    fn repository_rejects_invalid_policy_shapes_before_writing() {
        let dir = tempfile::tempdir().unwrap();
        let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
        let invalid = [
            ResourcePolicy {
                profile: ResourceProfile::Custom,
                custom_max_model_mb: None,
            },
            ResourcePolicy {
                profile: ResourceProfile::Custom,
                custom_max_model_mb: Some(1),
            },
            ResourcePolicy {
                profile: ResourceProfile::Custom,
                custom_max_model_mb: Some(513),
            },
            ResourcePolicy {
                profile: ResourceProfile::Everyday,
                custom_max_model_mb: Some(512),
            },
            ResourcePolicy {
                profile: ResourceProfile::LocalFocused,
                custom_max_model_mb: Some(512),
            },
        ];

        for policy in invalid {
            let error = repository.save(&policy).unwrap_err();
            assert!(matches!(error, ResourcePolicyError::InvalidPolicy { .. }));
            assert!(!repository.path().exists());
        }
    }

    #[test]
    fn load_evidence_distinguishes_loaded_missing_and_corrupt_defaults() {
        let dir = tempfile::tempdir().unwrap();
        let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());

        let missing = repository.load_with_evidence().unwrap();
        assert_eq!(missing.policy, ResourcePolicy::everyday());
        assert_eq!(missing.source, ResourcePolicyLoadSource::MissingDefault);
        assert!(missing.warning.is_none());

        repository.save(&ResourcePolicy::local_focused()).unwrap();
        let loaded = repository.load_with_evidence().unwrap();
        assert_eq!(loaded.policy, ResourcePolicy::local_focused());
        assert_eq!(loaded.source, ResourcePolicyLoadSource::Loaded);
        assert!(loaded.warning.is_none());

        let corrupt = br#"{"profile":"custom","custom_max_model_mb":"broken"}"#;
        std::fs::write(repository.path(), corrupt).unwrap();
        let recovered = repository.load_with_evidence().unwrap();
        assert_eq!(recovered.policy, ResourcePolicy::everyday());
        assert_eq!(recovered.source, ResourcePolicyLoadSource::CorruptDefault);
        assert!(recovered
            .warning
            .as_deref()
            .is_some_and(|warning| { warning.contains("could not be loaded") }));
        assert_eq!(std::fs::read(repository.path()).unwrap(), corrupt);
    }

    #[test]
    fn corrupt_evidence_covers_unknown_fields_invalid_shapes_and_oversized_files() {
        let dir = tempfile::tempdir().unwrap();
        let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
        let corrupt_documents = [
            br#"{"#.to_vec(),
            br#"{"profile":"everyday","custom_max_model_mb":null,"extra":true}"#.to_vec(),
            br#"{"profile":"custom","custom_max_model_mb":null}"#.to_vec(),
            br#"{"profile":"custom","custom_max_model_mb":1}"#.to_vec(),
            br#"{"profile":"custom","custom_max_model_mb":513}"#.to_vec(),
            br#"{"profile":"everyday","custom_max_model_mb":512}"#.to_vec(),
            br#"{"profile":"local_focused","custom_max_model_mb":512}"#.to_vec(),
            vec![b' '; MAX_POLICY_BYTES as usize + 1],
        ];

        for document in corrupt_documents {
            std::fs::write(repository.path(), &document).unwrap();
            let recovered = repository.load_with_evidence().unwrap();
            assert_eq!(recovered.source, ResourcePolicyLoadSource::CorruptDefault);
            assert!(recovered.warning.is_some());
            assert_eq!(std::fs::read(repository.path()).unwrap(), document);
        }
    }

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

        let dir = tempfile::tempdir().unwrap();
        let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
        let target = dir.path().join("target.json");
        std::fs::write(
            &target,
            br#"{"profile":"local_focused","custom_max_model_mb":null}"#,
        )
        .unwrap();
        symlink(&target, repository.path()).unwrap();

        let recovered = repository.load_with_evidence().unwrap();
        assert_eq!(recovered.source, ResourcePolicyLoadSource::CorruptDefault);
        assert!(recovered.warning.is_some());
        assert!(repository.path().is_symlink());
    }

    #[test]
    fn concurrent_saves_leave_one_complete_document_and_no_staging_files() {
        let dir = tempfile::tempdir().unwrap();
        let repository =
            std::sync::Arc::new(FileResourcePolicyRepository::new(dir.path().to_path_buf()));
        let mut writers = Vec::new();
        for index in 0..16_u64 {
            let repository = repository.clone();
            writers.push(std::thread::spawn(move || {
                repository
                    .save(&ResourcePolicy {
                        profile: ResourceProfile::Custom,
                        custom_max_model_mb: Some(index * 512),
                    })
                    .unwrap();
            }));
        }
        for writer in writers {
            writer.join().unwrap();
        }

        let loaded = repository.load().unwrap();
        assert_eq!(loaded.profile, ResourceProfile::Custom);
        assert!(loaded.custom_max_model_mb.unwrap().is_multiple_of(512));
        assert_no_temp_files(dir.path());
    }

    #[test]
    fn cuda_uses_separate_vram_fit_and_host_ram_policy() {
        let hardware = hardware(
            64 * 1024,
            crate::hardware::GpuBackend::Cuda,
            Some(12 * 1024),
        );

        let evidence = evaluate_resources(&hardware, &ResourcePolicy::everyday());

        assert_eq!(evidence.host_memory.configured_model_ceiling_mb, 26_214);
        let accelerator = evidence.accelerator_memory.unwrap();
        assert_eq!(accelerator.total_mb, 12 * 1024);
        assert_eq!(accelerator.budget_mb, 12 * 1024);
        assert_ne!(
            accelerator.budget_mb,
            evidence.host_memory.configured_model_ceiling_mb
        );
    }

    #[test]
    fn resource_preflight_zero_custom_budget_blocks_new_load() {
        let coordinator = LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(0.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();

        let preflight = coordinator.preflight(&model, 2_048);
        assert_eq!(preflight.verdict, LocalLoadVerdict::DisabledByPolicy);
    }

    #[test]
    fn resource_preflight_zero_budget_does_not_kill_resident_inference() {
        let coordinator = LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(0.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        coordinator.mark_resident(&model.id, 2_400);

        let preflight = coordinator.preflight(&model, 2_048);
        assert_eq!(preflight.verdict, LocalLoadVerdict::Allowed);
        assert_eq!(
            preflight.estimated_incremental_mb,
            preflight.estimate.context_overhead_mb
                + preflight.estimate.runtime_overhead_mb
                + preflight.estimate.transient_margin_mb
        );
    }

    #[test]
    fn resource_preflight_resident_weights_are_incremental_only() {
        let coordinator = LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(6_000)),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        coordinator.mark_resident(&model.id, 2_400);

        let preflight = coordinator.preflight(&model, 2_048);
        assert_eq!(
            preflight.estimated_incremental_mb,
            preflight.estimate.context_overhead_mb
                + preflight.estimate.runtime_overhead_mb
                + preflight.estimate.transient_margin_mb
        );
        assert_eq!(preflight.verdict, LocalLoadVerdict::Allowed);
    }

    #[test]
    fn resource_preflight_unavailable_live_probe_is_explicit() {
        let coordinator = LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::everyday(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::unknown()),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();

        let preflight = coordinator.preflight(&model, 2_048);
        assert_eq!(preflight.live_available_mb, None);
        assert_eq!(preflight.verdict, LocalLoadVerdict::LiveMemoryUnknown);
    }

    #[test]
    fn resource_preflight_simultaneous_reservations_are_atomic() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(5.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(6_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        coordinator.mark_resident(&model.id, 2_400);

        let first = coordinator.reserve(&model, 2_048).unwrap();
        let second = coordinator.reserve(&model, 2_048).unwrap_err();
        assert_eq!(
            second.preflight.verdict,
            LocalLoadVerdict::ExceedsConfiguredCeiling
        );
        drop(first);
        assert!(coordinator.reserve(&model, 2_048).is_ok());
    }

    #[test]
    fn distinct_state_roots_reserve_atomically_against_one_machine_ledger() {
        let machine_ledger = Arc::new(Mutex::new(MachineAdmissionLedger::default()));
        let policy = ResourcePolicy::custom_gb(6.0).unwrap();
        let hardware = hardware(32 * 1024, GpuBackend::Metal, None);
        let first = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
            policy.clone(),
            hardware.clone(),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            machine_ledger.clone(),
        ));
        let second = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
            policy,
            hardware,
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            machine_ledger,
        ));
        let mut model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        model.cost.ram_mb = Some(2 * 1024);
        model.cost.size_mb = Some(2 * 1024);
        let barrier = Arc::new(std::sync::Barrier::new(2));

        let attempts = [first, second].map(|coordinator| {
            let model = model.clone();
            let barrier = barrier.clone();
            std::thread::spawn(move || {
                barrier.wait();
                coordinator.reserve(&model, 0)
            })
        });
        let outcomes = attempts.map(|attempt| attempt.join().unwrap());

        assert_eq!(outcomes.iter().filter(|result| result.is_ok()).count(), 1);
        assert_eq!(
            outcomes
                .iter()
                .find_map(|result| result.as_ref().err())
                .expect("one cross-root request must be blocked")
                .preflight
                .verdict,
            LocalLoadVerdict::ExceedsConfiguredCeiling
        );
    }

    #[test]
    fn resource_preflight_cuda_charges_weights_to_vram_and_overhead_to_host() {
        let coordinator = LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(2.0).unwrap(),
            hardware(64 * 1024, GpuBackend::Cuda, Some(12 * 1024)),
            Arc::new(FixedLiveMemoryProbe::known(20_000)),
        );
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();

        let preflight = coordinator.preflight(&model, 2_048);
        assert_eq!(preflight.resident_model_mb, 0);
        assert_eq!(preflight.accelerator_total_mb, Some(12 * 1024));
        assert_eq!(
            preflight.accelerator_incremental_mb,
            Some(preflight.estimate.weights_mb)
        );
        assert_eq!(
            preflight.estimated_incremental_mb,
            preflight.estimate.context_overhead_mb
                + preflight.estimate.runtime_overhead_mb
                + preflight.estimate.transient_margin_mb
        );
        assert_eq!(preflight.verdict, LocalLoadVerdict::Allowed);
    }

    #[test]
    fn resource_preflight_cuda_vram_reservations_are_atomic() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(64 * 1024, GpuBackend::Cuda, Some(3_000)),
            Arc::new(FixedLiveMemoryProbe::known(20_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();

        let first = coordinator.reserve(&model, 2_048).unwrap();
        assert_eq!(
            coordinator
                .reserve(&model, 2_048)
                .unwrap_err()
                .preflight
                .verdict,
            LocalLoadVerdict::ExceedsConfiguredCeiling
        );
        drop(first);
        assert!(coordinator.reserve(&model, 2_048).is_ok());
    }

    #[test]
    fn resource_preflight_model_maintenance_races_atomically_with_reserve() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::everyday(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();

        let active = coordinator.reserve(&model, 2_048).unwrap();
        assert!(matches!(
            coordinator.begin_model_maintenance(&model.id),
            Err(ModelMaintenanceError::ModelInUse(_))
        ));
        drop(active);

        let maintenance = coordinator.begin_model_maintenance(&model.id).unwrap();
        assert_eq!(
            coordinator
                .reserve(&model, 2_048)
                .unwrap_err()
                .preflight
                .verdict,
            LocalLoadVerdict::ModelMaintenance
        );
        drop(maintenance);
        assert!(coordinator.reserve(&model, 2_048).is_ok());
    }

    #[test]
    fn measured_weights_are_rechecked_atomically_before_allocation() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(5.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(16_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let mut reservation = coordinator.reserve(&model, 2_048).unwrap();

        let blocked = reservation
            .reconcile_measured_weights(6 * 1024 * 1024 * 1024)
            .unwrap_err();

        assert_eq!(
            blocked.preflight.verdict,
            LocalLoadVerdict::ExceedsConfiguredCeiling
        );
        assert_eq!(
            coordinator.preflight(&model, 2_048).active_reservations_mb,
            reservation.reserved_incremental_mb(),
            "a rejected resize must not mutate the live reservation"
        );
    }

    #[test]
    fn cache_publication_transfers_cold_reservation_without_double_counting() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(16_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let mut reservation = coordinator.reserve(&model, 2_048).unwrap();
        let measured_bytes = 3 * 1024 * 1024 * 1024_u64;

        reservation
            .reconcile_measured_weights(measured_bytes)
            .unwrap();
        reservation.publish_resident_weights(measured_bytes);
        let after = coordinator.preflight(&model, 2_048);
        let request_overhead = after.estimate.context_overhead_mb
            + after.estimate.runtime_overhead_mb
            + after.estimate.transient_margin_mb;

        assert_eq!(after.resident_model_mb, 3 * 1024);
        assert_eq!(after.active_reservations_mb, request_overhead);
        assert_eq!(reservation.reserved_incremental_mb(), request_overhead);
    }

    #[test]
    fn resident_publication_never_reprobes_or_rejects_after_allocation() {
        struct CountingProbe {
            calls: AtomicU64,
            available_mb: u64,
        }

        impl LiveMemoryProbe for CountingProbe {
            fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError> {
                self.calls.fetch_add(1, Ordering::Relaxed);
                Ok(Some(self.available_mb))
            }
        }

        let probe = Arc::new(CountingProbe {
            calls: AtomicU64::new(0),
            available_mb: 24_000,
        });
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            probe.clone(),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let mut reservation = coordinator.reserve(&model, 2_048).unwrap();
        let measured = 3 * 1024 * 1024 * 1024_u64;
        reservation.reconcile_measured_weights(measured).unwrap();
        let calls_before_publish = probe.calls.load(Ordering::Relaxed);

        reservation.publish_resident_weights(measured + 512 * 1024 * 1024);

        assert_eq!(probe.calls.load(Ordering::Relaxed), calls_before_publish);
        assert!(coordinator.is_resident(&model.id));
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .resident_models
                .get(&model.id)
                .map(|allocation| allocation.weights_mb),
            Some(3 * 1024 + 512)
        );
    }

    #[test]
    fn simultaneous_cold_reservations_converge_on_one_resident_allocation() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(16.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Metal, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let mut first = coordinator.reserve(&model, 2_048).unwrap();
        let mut second = coordinator.reserve(&model, 2_048).unwrap();
        let measured = 3 * 1024 * 1024 * 1024_u64;

        first.reconcile_measured_weights(measured).unwrap();
        second.reconcile_measured_weights(measured).unwrap();
        first.publish_resident_weights(measured);
        second.publish_resident_weights(measured);

        let after = coordinator.preflight(&model, 2_048);
        let per_request_overhead = after.estimate.context_overhead_mb
            + after.estimate.runtime_overhead_mb
            + after.estimate.transient_margin_mb;
        assert_eq!(after.resident_model_mb, 3 * 1024);
        assert_eq!(
            after.active_reservations_mb,
            per_request_overhead * 2,
            "peer publication must release only this reservation's redundant cold weights"
        );
    }

    #[test]
    fn resident_eviction_before_allocation_repromotes_reservation_to_cold() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let model = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.id == "mlx/qwen3-4b:4bit")
            .unwrap();
        let measured = 3 * 1024 * 1024 * 1024_u64;
        let mut first = coordinator.reserve(&model, 2_048).unwrap();
        first.reconcile_measured_weights(measured).unwrap();
        first.publish_resident_weights(measured);
        drop(first);
        let mut replacement = coordinator.reserve(&model, 2_048).unwrap();
        coordinator.mark_evicted(&model.id);

        let preflight = replacement.reconcile_measured_weights(measured).unwrap();
        assert!(preflight.estimated_incremental_mb >= 3 * 1024);
    }

    #[test]
    fn maintenance_catalog_alias_blocks_provider_alias_reservation() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        coordinator
            .register_model_aliases("mlx/kokoro-82m:6bit", ["mlx-community/Kokoro-82M-6bit"]);
        let _maintenance = coordinator
            .begin_model_maintenance("mlx/kokoro-82m:6bit")
            .unwrap();
        let mut provider_schema = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.is_local())
            .expect("one local schema");
        provider_schema.id = "mlx-community/Kokoro-82M-6bit".into();

        let blocked = coordinator.reserve(&provider_schema, 512).unwrap_err();
        assert_eq!(
            blocked.preflight.verdict,
            LocalLoadVerdict::ModelMaintenance
        );
        assert_eq!(blocked.preflight.model_id, "mlx/kokoro-82m:6bit");
    }

    #[test]
    fn late_alias_registration_rekeys_active_and_resident_state_atomically() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let mut schema = crate::registry::builtin_catalog()
            .into_iter()
            .find(|model| model.is_local())
            .unwrap();
        schema.id = "provider/artifact".into();
        let active = coordinator.reserve(&schema, 0).unwrap();

        coordinator.register_model_aliases("catalog/model:default", ["provider/artifact"]);
        assert!(matches!(
            coordinator.begin_model_maintenance("catalog/model:default"),
            Err(ModelMaintenanceError::ModelInUse(_))
        ));
        drop(active);

        let maintenance = coordinator
            .begin_model_maintenance("catalog/model:default")
            .unwrap();
        let error = coordinator.reserve(&schema, 0).unwrap_err();
        assert_eq!(error.preflight.verdict, LocalLoadVerdict::ModelMaintenance);
        drop(maintenance);
    }

    #[test]
    fn process_allocations_teardown_by_exact_owner_without_sibling_erasure() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(16.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let logical = "local/shared-model";
        let worker = worker_process_allocation_id(logical);
        let vllm = vllm_process_allocation_id(logical);
        let bytes = 512 * 1024 * 1024_u64;
        let mut worker_load = coordinator
            .reserve_measured_host_allocation(logical, &worker, bytes, 0)
            .unwrap();
        worker_load.publish_resident_weights_as(&worker, bytes);
        drop(worker_load);
        let mut vllm_load = coordinator
            .reserve_measured_host_allocation(logical, &vllm, bytes, 0)
            .unwrap();
        vllm_load.publish_resident_weights_as(&vllm, bytes);
        drop(vllm_load);
        assert_eq!(coordinator.resident_model_mb(), 1024);

        coordinator.mark_teardown_pending_allocation(logical, &worker);
        coordinator.mark_teardown_pending_allocation(logical, &vllm);
        coordinator.finish_teardown_allocation(logical, &worker);
        assert!(coordinator.teardown_pending(logical));
        assert_eq!(
            coordinator.resident_allocation_ids(logical),
            vec![vllm.clone()]
        );
        assert_eq!(coordinator.resident_model_mb(), 512);

        coordinator.finish_teardown_allocation(logical, &vllm);
        assert!(!coordinator.teardown_pending(logical));
        assert!(!coordinator.is_resident(logical));
    }

    #[test]
    fn replacement_generation_reconcile_removes_peer_resident_discount_before_load() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(1.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let logical = "local/replaced-worker";
        let bytes = 512 * 1024 * 1024_u64;
        let mut old = coordinator
            .reserve_measured_host_allocation(logical, "worker:old", bytes, 0)
            .unwrap();
        old.publish_resident_weights_as("worker:old", bytes);
        drop(old);

        let mut replacement = coordinator
            .reserve_measured_host(logical, bytes, 0)
            .expect("logical preflight initially sees the old resident");
        assert_eq!(replacement.reserved_incremental_mb(), 0);
        replacement.bind_allocation_id("worker:new");
        replacement
            .reconcile_measured_weights(bytes)
            .expect("the exact replacement generation fits by itself");
        assert_eq!(replacement.reserved_incremental_mb(), 512);
        assert!(
            coordinator
                .reserve_measured_host("different/model", bytes, 0)
                .is_err(),
            "a second cold model must see both old residency and the replacement generation"
        );

        drop(replacement);
        assert!(coordinator
            .reserve_measured_host("different/model", bytes, 0)
            .is_ok());
    }

    #[test]
    fn pre_ack_pending_charge_blocks_other_scope_until_exact_exit_ack() {
        let machine = Arc::new(Mutex::new(MachineAdmissionLedger::default()));
        let first = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
            ResourcePolicy::custom_gb(2.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            machine.clone(),
        ));
        let second = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
            ResourcePolicy::custom_gb(2.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            machine,
        ));
        let logical = "managed/starting";
        let allocation = worker_process_allocation_id(logical);
        let cold = first
            .reserve_measured_host_allocation(logical, &allocation, 1024 * 1024 * 1024, 0)
            .unwrap();
        first.mark_teardown_pending_allocation_with_charge(
            logical,
            &allocation,
            cold.reconciled_weights_bytes(),
        );
        drop(cold);

        let blocked = second
            .reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
            .unwrap_err();
        assert_eq!(
            blocked.preflight.verdict,
            LocalLoadVerdict::ExceedsConfiguredCeiling
        );
        first.finish_teardown_allocation(logical, "worker:unrelated-sibling");
        assert!(first.teardown_pending(logical));
        assert_eq!(first.resident_model_mb(), 1024);

        first.finish_teardown_allocation(logical, &allocation);
        assert!(!first.teardown_pending(logical));
        assert!(second
            .reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
            .is_ok());
    }

    #[test]
    fn detached_native_lease_keeps_machine_charge_after_request_cancellation() {
        let machine = Arc::new(Mutex::new(MachineAdmissionLedger::default()));
        let first = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
            ResourcePolicy::custom_gb(2.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            machine.clone(),
        ));
        let second = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
            ResourcePolicy::custom_gb(2.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
            machine,
        ));
        let reservation = first
            .reserve_measured_host("detached/model", 1024 * 1024 * 1024, 0)
            .unwrap();
        let detached = reservation.detached_lease();
        drop(reservation);
        assert!(second
            .reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
            .is_err());
        drop(detached);
        assert!(second
            .reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
            .is_ok());
    }

    #[test]
    fn normal_awaited_detached_work_shares_one_charge_with_its_request() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(1.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let request = coordinator
            .reserve_measured_host("native/model-a", 512 * 1024 * 1024, 0)
            .unwrap();
        let detached = request.detached_lease();

        let peer = coordinator
            .reserve_measured_host("native/model-b", 512 * 1024 * 1024, 0)
            .expect("request + its detached job are one 512 MB allocation, not two");
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .active_host_reservations_mb,
            1024
        );

        drop(peer);
        drop(detached);
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .active_host_reservations_mb,
            512
        );
        drop(request);
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .active_host_reservations_mb,
            0
        );
    }

    #[test]
    fn concurrent_non_aligned_starts_transfer_cold_weights_to_pending_without_double_charge() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        let first_allocation = vllm_process_allocation_id("managed/model-a");
        let second_allocation = vllm_process_allocation_id("managed/model-b");
        let mib = 1024 * 1024_u64;
        let first_measured_bytes = 1024 * mib + 1;
        let second_measured_bytes = 512 * mib + 1;
        let mut first = coordinator
            .reserve_measured_host_allocation(
                "managed/model-a",
                &first_allocation,
                first_measured_bytes,
                128,
            )
            .unwrap();
        let mut second = coordinator
            .reserve_measured_host_allocation(
                "managed/model-b",
                &second_allocation,
                second_measured_bytes,
                256,
            )
            .unwrap();

        first.transfer_cold_weights_to_pending_allocation(&first_allocation, first_measured_bytes);
        second
            .transfer_cold_weights_to_pending_allocation(&second_allocation, second_measured_bytes);

        assert_eq!(
            coordinator.resident_model_mb(),
            1538,
            "each non-MiB-aligned allocation must be rounded up independently"
        );
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .active_host_reservations_mb,
            384,
            "pending weights replace cold request weights while request overhead remains active"
        );

        first.publish_resident_weights_as(&first_allocation, first_measured_bytes);
        second.publish_resident_weights_as(&second_allocation, second_measured_bytes);
        assert_eq!(coordinator.resident_model_mb(), 1538);
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .active_host_reservations_mb,
            384,
            "publication must not subtract transferred weights twice"
        );

        drop(first);
        drop(second);
        assert_eq!(
            coordinator
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .active_host_reservations_mb,
            0
        );
        assert_eq!(coordinator.resident_model_mb(), 1538);
    }

    #[test]
    fn pending_allocation_identity_is_idempotent_and_sibling_exact() {
        let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
            Arc::new(FixedLiveMemoryProbe::known(24_000)),
        ));
        coordinator.mark_teardown_pending_allocation_with_charge(
            "same/model",
            "worker:same/model",
            512 * 1024 * 1024,
        );
        coordinator.mark_teardown_pending_allocation_with_charge(
            "same/model",
            "worker:same/model",
            512 * 1024 * 1024,
        );
        coordinator.mark_teardown_pending_allocation_with_charge(
            "same/model",
            "vllm:same/model",
            256 * 1024 * 1024,
        );
        assert_eq!(coordinator.resident_model_mb(), 768);
        coordinator.finish_teardown_allocation("same/model", "worker:same/model");
        assert!(coordinator.teardown_pending("same/model"));
        assert_eq!(coordinator.resident_model_mb(), 256);
        coordinator.finish_teardown_allocation("same/model", "worker:same/model");
        assert_eq!(coordinator.resident_model_mb(), 256);
        coordinator.finish_teardown_allocation("same/model", "vllm:same/model");
        assert!(!coordinator.teardown_pending("same/model"));
    }

    #[test]
    fn scoped_coordinator_identity_is_stable_for_one_state_root() {
        let root = tempfile::tempdir().unwrap();
        let first = scoped_local_admission(
            root.path(),
            ResourcePolicy::custom_gb(4.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
        );
        let second = scoped_local_admission(
            root.path(),
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
        );
        assert!(Arc::ptr_eq(&first, &second));
        assert_eq!(second.policy(), ResourcePolicy::custom_gb(8.0).unwrap());
    }

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

        let fixture = tempfile::tempdir().unwrap();
        let real = fixture.path().join("real-state");
        std::fs::create_dir(&real).unwrap();
        let alias = fixture.path().join("state-alias");
        symlink(&real, &alias).unwrap();

        let first = scoped_local_admission(
            &real,
            ResourcePolicy::custom_gb(4.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
        );
        let second = scoped_local_admission(
            &alias,
            ResourcePolicy::custom_gb(8.0).unwrap(),
            hardware(32 * 1024, GpuBackend::Cpu, None),
        );
        assert!(Arc::ptr_eq(&first, &second));
        assert_eq!(second.policy(), ResourcePolicy::custom_gb(8.0).unwrap());
    }

    #[test]
    fn unavailable_state_root_normalization_is_absolute_and_lexically_stable() {
        let fixture = tempfile::tempdir().unwrap();
        let missing = fixture.path().join("not-created").join("..").join("state");
        assert_eq!(
            normalized_state_root_key(&missing),
            normalized_state_root_key(&fixture.path().join("state"))
        );
    }

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

        let fixture = tempfile::tempdir().unwrap();
        let real = fixture.path().join("real");
        std::fs::create_dir(&real).unwrap();
        let alias = fixture.path().join("alias");
        symlink(&real, &alias).unwrap();

        assert_eq!(
            normalized_state_root_key(&alias.join("missing").join("state")),
            normalized_state_root_key(&real.join("missing").join("state"))
        );
    }

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

        let fixture = tempfile::tempdir().unwrap();
        let physical_parent = fixture.path().join("physical");
        let physical_child = physical_parent.join("child");
        std::fs::create_dir_all(&physical_child).unwrap();
        let aliases = fixture.path().join("aliases");
        std::fs::create_dir(&aliases).unwrap();
        let alias = aliases.join("runtime");
        symlink(&physical_child, &alias).unwrap();

        let through_alias = alias.join("..").join("missing-state");
        assert_eq!(
            normalized_state_root_key(&through_alias),
            normalized_state_root_key(&physical_parent.join("missing-state"))
        );
        assert_ne!(
            normalized_state_root_key(&through_alias),
            aliases.join("missing-state")
        );
    }

    fn assert_no_temp_files(directory: &std::path::Path) {
        let entries = std::fs::read_dir(directory)
            .unwrap()
            .map(|entry| entry.unwrap().path())
            .collect::<Vec<_>>();
        assert_eq!(entries, vec![directory.join("model-resource-policy.json")]);
    }

    #[cfg(unix)]
    fn assert_private_mode(path: &std::path::Path, expected: u32) {
        use std::os::unix::fs::PermissionsExt;
        assert_eq!(
            std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
            expected
        );
    }

    #[cfg(not(unix))]
    fn assert_private_mode(_path: &std::path::Path, _expected: u32) {}
}

/// Excerpts of the real `config.json` files (the geometry keys, verbatim) of
/// the two checkpoints whose memory was measured through vllm-mlx 0.6.0 on an
/// M5 Pro, 2026-10-26.
#[cfg(test)]
mod measured_vllm_mlx_memory {
    use super::*;

    const QWEN3_8_27B_4BIT: &str = r#"{"model_type":"qwen3_5","text_config":{"model_type":"qwen3_5_text","num_hidden_layers":64,"num_attention_heads":24,"num_key_value_heads":4,"head_dim":256,"hidden_size":5120,"vocab_size":248320,"full_attention_interval":4,"layer_types":["linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention"],"linear_conv_kernel_dim":4,"linear_key_head_dim":128,"linear_num_key_heads":16,"linear_num_value_heads":48,"linear_value_head_dim":128}}"#;
    const GLM_4_7_FLASH_8BIT: &str = r#"{"model_type":"glm4_moe_lite","num_hidden_layers":47,"num_attention_heads":20,"num_key_value_heads":20,"hidden_size":2048,"vocab_size":154880,"kv_lora_rank":512,"qk_rope_head_dim":64,"qk_nope_head_dim":192,"v_head_dim":256}"#;

    fn geometry(raw: &str) -> KvGeometry {
        let raw: serde_json::Value = serde_json::from_str(raw).unwrap();
        let source = raw.get("text_config").cloned().unwrap_or(raw);
        geometry_from_config(&serde_json::from_value(source).unwrap()).unwrap()
    }

    fn m5_pro_64gb() -> crate::hardware::HardwareInfo {
        crate::hardware::HardwareInfo {
            os: "macos".into(),
            arch: "aarch64".into(),
            cpu_cores: 12,
            total_ram_mb: 64 * 1024,
            gpu_backend: GpuBackend::Metal,
            gpu_memory_mb: None,
            gpu_devices: Vec::new(),
            recommended_model: "fixture".into(),
            recommended_context: 4_096,
            max_model_mb: 64 * 1024,
        }
    }

    #[test]
    fn a_hybrid_model_caches_only_its_full_attention_layers() {
        let g = geometry(QWEN3_8_27B_4BIT);
        // 16 of 64 layers are full attention: 2 × 16 × 4 KV heads × 256 × 2.
        assert_eq!(g.kv_bytes_per_token, 65_536);
        // 48 gated-delta layers' fixed state; measured as the cache's
        // length-independent 0.15 GiB.
        assert_eq!(g.fixed_state_bytes, 153_944_064);
        assert_eq!(g.vocab_size, Some(248_320));
    }

    #[test]
    fn an_mla_model_caches_its_latent_and_pays_for_the_expansion_in_prefill() {
        let g = geometry(GLM_4_7_FLASH_8BIT);
        // (512 latent + 64 rope) × 2 bytes × 47 layers, as measured.
        assert_eq!(g.kv_bytes_per_token, 54_144);
        // Measured near 290 KiB/token beyond the cache; the derivation is
        // within 25% above it.
        let measured_scratch = 289 * 1024;
        assert!(g.prefill_bytes_per_token >= measured_scratch);
        assert!(g.prefill_bytes_per_token <= measured_scratch * 5 / 4);
    }

    /// MLX peak (`/v1/status` metal `peak_memory_gb`) through vllm-mlx 0.6.0,
    /// served with CAR's own flags (Qwen3.8 with its MTP drafter, so through
    /// mlx-vlm), at about 8k, 32k and 64k prompt tokens.
    const MEASURED_PEAK_GIB: [(&str, &str, [(usize, f64); 3]); 2] = [
        (
            "vllm-mlx/qwen3.8-27b:4bit",
            QWEN3_8_27B_4BIT,
            [(8_192, 19.78), (32_768, 21.64), (65_536, 24.24)],
        ),
        (
            "vllm-mlx/glm-4.7-flash:8bit",
            GLM_4_7_FLASH_8BIT,
            [(8_192, 34.84), (32_768, 42.45), (65_536, 53.47)],
        ),
    ];

    #[test]
    fn the_estimate_brackets_what_vllm_mlx_was_measured_to_use() {
        let catalog = crate::registry::builtin_catalog();
        for (id, config, points) in MEASURED_PEAK_GIB {
            let model = catalog.iter().find(|m| m.id == id).unwrap();
            let g = geometry(config);
            for (tokens, measured) in points {
                let estimate = estimate_model_memory_with_geometry(
                    model,
                    &m5_pro_64gb(),
                    tokens,
                    None,
                    Some(&g),
                );
                let gib = estimate.estimated_peak_mb as f64 / 1024.0;
                // Never below what the server used (an under-estimate admits a
                // load the machine cannot hold), and not so far above that a
                // load that fits is refused.
                assert!(
                    gib >= measured,
                    "{id} at {tokens}: estimate {gib:.1} GiB < measured {measured} GiB"
                );
                assert!(
                    gib <= measured + 6.0,
                    "{id} at {tokens}: estimate {gib:.1} GiB > measured {measured} + 6 GiB"
                );
            }
        }
    }

    fn parse(raw: &str) -> Option<KvGeometry> {
        geometry_from_config(&serde_json::from_str(raw).unwrap())
    }

    #[test]
    fn an_absurd_config_saturates_instead_of_wrapping_to_a_small_estimate() {
        let huge = parse(r#"{"num_hidden_layers":4294967296,"num_key_value_heads":4294967296,"head_dim":4294967296,"hidden_size":4611686018427387904}"#).unwrap();
        assert_eq!(huge.kv_bytes_per_token, u64::MAX);
        assert_eq!(huge.prefill_bytes_per_token, u64::MAX);
    }

    #[test]
    fn mla_on_an_architecture_not_known_to_cache_the_latent_is_sized_per_head() {
        // GLM-4.7-Flash's geometry under minicpm3's name, which mlx-lm caches
        // as expanded keys and values.
        let renamed = GLM_4_7_FLASH_8BIT.replace("glm4_moe_lite", "minicpm3");
        let expanded = parse(&renamed).unwrap();
        // 47 layers × 20 heads × ((192 nope + 64 rope) keys + 256 values) × 2.
        assert_eq!(expanded.kv_bytes_per_token, 47 * 20 * (256 + 256) * 2);
        assert!(expanded.kv_bytes_per_token > geometry(GLM_4_7_FLASH_8BIT).kv_bytes_per_token);
    }

    #[test]
    fn linear_layers_without_state_dimensions_stay_counted_as_cache_layers() {
        let undimensioned = parse(
            r#"{"num_hidden_layers":4,"num_key_value_heads":4,"head_dim":256,"layer_types":["linear_attention","linear_attention","linear_attention","full_attention"]}"#,
        )
        .unwrap();
        assert_eq!(undimensioned.kv_bytes_per_token, 4 * 2 * 4 * 256 * 2);
        assert_eq!(undimensioned.fixed_state_bytes, 0);
    }

    #[test]
    fn a_row_served_from_the_hugging_face_cache_is_sized_from_that_snapshot() {
        let tmp = tempfile::tempdir().unwrap();
        let hub = tmp.path().join("hub");
        let repo = hub.join("models--mlx-community--Qwen3.8-27B-4bit");
        let snapshot = repo.join("snapshots/fixture");
        std::fs::create_dir_all(&snapshot).unwrap();
        std::fs::write(snapshot.join("config.json"), QWEN3_8_27B_4BIT).unwrap();
        std::fs::create_dir_all(repo.join("refs")).unwrap();
        std::fs::write(repo.join("refs/main"), "fixture").unwrap();
        // The managed directory holds only the drafter, as on a real install.
        let models_dir = tmp.path().join("models");
        std::fs::create_dir_all(models_dir.join("Qwen3.8-27B-MLX-vLLM/mtp-drafter")).unwrap();
        let catalog = crate::registry::builtin_catalog();
        let model = catalog
            .iter()
            .find(|m| m.id == "vllm-mlx/qwen3.8-27b:4bit")
            .unwrap();

        let served = kv_geometry_for(model, Some(&models_dir), Some(&hub));
        assert_eq!(served.map(|g| g.kv_bytes_per_token), Some(65_536));
        // Without a hub only the managed directory counts, so a test (or a
        // caller that did not ask) never sizes a row from someone's cache.
        assert_eq!(kv_geometry_for(model, Some(&models_dir), None), None);
    }
}