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;
/// 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 within the operating ceiling.
    ///
    /// Everyday intentionally targets half of its 40% admission ceiling so
    /// the primary leaves room for context growth, CAR, and changing app
    /// load. Other profiles may use their full configured ceiling.
    pub fn recommendation_target_mb(&self, total_memory_mb: u64) -> u64 {
        let ceiling = self
            .effective_budget(total_memory_mb)
            .configured_model_ceiling_mb;
        if self.profile == ResourceProfile::Everyday {
            ceiling / 2
        } else {
            ceiling
        }
    }

    /// 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>>,
    kv_geometry_cache: Mutex<HashMap<String, 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) = std::fs::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) = std::fs::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,
    ));
    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)
    }

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

    fn model_memory_estimate(
        &self,
        model: &ModelSchema,
        context_tokens: usize,
    ) -> ModelMemoryEstimate {
        let models_dir = self
            .models_dir
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        let geometry = {
            let mut cache = self
                .kv_geometry_cache
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if let Some(geometry) = cache.get(&model.id).cloned() {
                Some(geometry)
            } else {
                let geometry = kv_geometry_for(model, models_dir.as_deref());
                if let Some(geometry) = geometry.as_ref() {
                    cache.insert(model.id.clone(), geometry.clone());
                }
                geometry
            }
        };
        estimate_model_memory_with_geometry(
            model,
            &self.hardware,
            context_tokens,
            None,
            geometry.as_ref(),
        )
    }

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

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

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

#[derive(Clone, Debug, PartialEq, Eq)]
struct KvGeometry {
    num_hidden_layers: usize,
    num_key_value_heads: usize,
    head_dim: usize,
    kv_bytes_per_element: usize,
    hidden_size: Option<usize>,
}

/// 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 {
    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>,
}

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

/// 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>) -> Option<KvGeometry> {
    let path = models_dir?.join(&model.name).join("config.json");
    let bytes = std::fs::read(path).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);
    }
    let num_hidden_layers = config.num_hidden_layers.filter(|value| *value > 0)?;
    let num_key_value_heads = config.num_key_value_heads.filter(|value| *value > 0)?;
    let head_dim = config.head_dim.filter(|value| *value > 0).or_else(|| {
        let hidden_size = config.hidden_size?;
        let attention_heads = config.num_attention_heads.filter(|value| *value > 0)?;
        hidden_size
            .is_multiple_of(attention_heads)
            .then_some(hidden_size / attention_heads)
    })?;
    let dtype = config
        .kv_cache_dtype
        .as_deref()
        .or(config.torch_dtype.as_deref());
    Some(KvGeometry {
        num_hidden_layers,
        num_key_value_heads,
        head_dim,
        kv_bytes_per_element: dtype_bytes(dtype),
        hidden_size: config.hidden_size.filter(|value| *value > 0),
    })
}

/// Binary-megabyte KV allocation for decoder geometry and context length:
/// `2 × layers × KV heads × head dimension × element bytes × tokens`.
fn kv_cache_mb_for(geometry: &KvGeometry, context_tokens: usize) -> u64 {
    let bytes = 2_u128
        .saturating_mul(geometry.num_hidden_layers as u128)
        .saturating_mul(geometry.num_key_value_heads as u128)
        .saturating_mul(geometry.head_dim as u128)
        .saturating_mul(geometry.kv_bytes_per_element as u128)
        .saturating_mul(context_tokens 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
}

/// Four fp16 `[tokens × hidden]` tensors conservatively alive together during
/// prompt prefill. This is additive to persistent KV allocation.
fn prefill_activation_mb(geometry: Option<&KvGeometry>, context_tokens: usize) -> u64 {
    let Some(hidden_size) = geometry.and_then(|geometry| geometry.hidden_size) else {
        return 0;
    };
    let bytes = (hidden_size as u128)
        .saturating_mul(context_tokens as u128)
        .saturating_mul(2)
        .saturating_mul(4);
    bytes.div_ceil(BYTES_PER_MB).min(u64::MAX as u128) as u64
}

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

    #[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 {
            num_hidden_layers: 36,
            num_key_value_heads: 8,
            head_dim: 128,
            kv_bytes_per_element: 2,
            hidden_size: Some(4_096),
        }
    }

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

    #[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;
        std::fs::write(&config_path, b"not json anymore").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) {}
}