//! Persisted admission policy for CAR-managed local model memory.
//!
//! This module answers how much host memory CAR may admit for local models. It
//! deliberately does not claim to impose an operating-system RSS limit. Apple
//! Silicon and CPU inference use the host budget; discrete CUDA keeps measured
//! VRAM as a separate fit constraint instead of applying the Apple unified-
//! memory percentage to it.
use std::collections::{HashMap, HashSet};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Weak};
use std::sync::{Mutex, OnceLock};
use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::hardware::{GpuBackend, HardwareInfo};
use crate::schema::ModelSchema;
/// File name under the CAR state root.
pub const RESOURCE_POLICY_FILE: &str = "model-resource-policy.json";
const EVERYDAY_MODEL_PERCENT: u64 = 40;
const LOCAL_FOCUSED_MODEL_PERCENT: u64 = 80;
const EMERGENCY_RESERVE_PERCENT: u64 = 10;
const MINIMUM_EMERGENCY_RESERVE_MB: u64 = 2 * 1024;
const MAX_POLICY_BYTES: u64 = 64 * 1024;
/// Context window used by recommendation-time memory estimates. This is a
/// typical working window, not a model's advertised maximum.
pub const RECOMMENDATION_CONTEXT_TOKENS: usize = 8_192;
const METAL_RUNTIME_OVERHEAD_MB: u64 = 512;
const CUDA_RUNTIME_OVERHEAD_MB: u64 = 512;
const CPU_RUNTIME_OVERHEAD_MB: u64 = 1_024;
const TRANSIENT_ALLOCATION_MARGIN_MB: u64 = 1_024;
/// The bound an inference worker sets on MLX's buffer cache: the transient
/// allocation margin admission already charges every local model, so the
/// reuse cache stays inside what was admitted rather than MLX's platform
/// default (car#1899).
pub fn allocator_cache_limit_bytes() -> u64 {
TRANSIENT_ALLOCATION_MARGIN_MB * 1024 * 1024
}
/// Memory accounting uses binary megabytes throughout this module.
const BYTES_PER_MB: u128 = 1024 * 1024;
/// Preset or exact user-selected local-model admission behavior.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ResourceProfile {
Everyday,
LocalFocused,
Custom,
}
/// The persisted local-model memory preference.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct ResourcePolicy {
pub profile: ResourceProfile,
pub custom_max_model_mb: Option<u64>,
}
impl Default for ResourcePolicy {
fn default() -> Self {
Self::everyday()
}
}
impl ResourcePolicy {
pub fn everyday() -> Self {
Self {
profile: ResourceProfile::Everyday,
custom_max_model_mb: None,
}
}
pub fn local_focused() -> Self {
Self {
profile: ResourceProfile::LocalFocused,
custom_max_model_mb: None,
}
}
/// Build an exact Custom policy from user-facing GB input.
///
/// Only finite, nonnegative half-GB increments are accepted. The persisted
/// value is integer MB, so it round-trips without floating-point drift.
pub fn custom_gb(gigabytes: f64) -> Result<Self, ResourcePolicyError> {
if !gigabytes.is_finite() || gigabytes < 0.0 {
return Err(ResourcePolicyError::InvalidCustomGigabytes(gigabytes));
}
let half_gb_steps = gigabytes * 2.0;
if half_gb_steps.fract() != 0.0 || half_gb_steps > u64::MAX as f64 {
return Err(ResourcePolicyError::InvalidCustomGigabytes(gigabytes));
}
let steps = half_gb_steps as u64;
let custom_max_model_mb = steps
.checked_mul(512)
.ok_or(ResourcePolicyError::InvalidCustomGigabytes(gigabytes))?;
Ok(Self {
profile: ResourceProfile::Custom,
custom_max_model_mb: Some(custom_max_model_mb),
})
}
/// Evaluate the stored preference against the current machine capacity.
///
/// Custom values are clamped for this evaluation only. The persisted value
/// is left untouched so a preference moved between machines is never
/// silently rewritten.
pub fn effective_budget(&self, total_memory_mb: u64) -> EffectiveResourceBudget {
let emergency_reserve_mb = minimum_emergency_reserve(total_memory_mb);
let safe_maximum_mb = total_memory_mb.saturating_sub(emergency_reserve_mb);
let requested_ceiling_mb = match self.profile {
ResourceProfile::Everyday => percent_of(total_memory_mb, EVERYDAY_MODEL_PERCENT),
ResourceProfile::LocalFocused => {
percent_of(total_memory_mb, LOCAL_FOCUSED_MODEL_PERCENT)
}
ResourceProfile::Custom => self.custom_max_model_mb.unwrap_or(0),
};
let configured_model_ceiling_mb = requested_ceiling_mb.min(safe_maximum_mb);
let normalization_notice = (matches!(self.profile, ResourceProfile::Custom)
&& requested_ceiling_mb > safe_maximum_mb)
.then(|| {
format!(
"The saved Custom allocation was adjusted from {requested_ceiling_mb} MB to \
{safe_maximum_mb} MB on this machine to preserve the \
{emergency_reserve_mb} MB emergency reserve."
)
});
EffectiveResourceBudget {
total_memory_mb,
emergency_reserve_mb,
configured_model_ceiling_mb,
effective_new_load_ceiling_mb: configured_model_ceiling_mb,
normalization_notice,
}
}
/// Recommendation target: the profile's own admission ceiling.
///
/// Everyday used to recommend only up to half of its 40% ceiling, which on
/// a 64 GB Mac (12.8 GB) excluded every local model that scores near the
/// frontier (Qwen3.8-27B, Qwen3.6-35B-A3B, Gemma-4-12B) from automatic
/// selection, though each loads within the ceiling. The estimate a row is
/// compared against already carries its context, runtime and transient
/// margin, and admission still checks live memory at load.
pub fn recommendation_target_mb(&self, total_memory_mb: u64) -> u64 {
self.effective_budget(total_memory_mb)
.configured_model_ceiling_mb
}
/// Validate the exact persisted shape accepted by CAR.
///
/// Keeping this check on the value itself ensures file reads and writes
/// cannot drift into accepting different policy documents.
pub fn validate(&self) -> Result<(), ResourcePolicyError> {
match self.profile {
ResourceProfile::Custom => match self.custom_max_model_mb {
Some(value) if value.is_multiple_of(512) => Ok(()),
Some(value) => Err(ResourcePolicyError::InvalidPolicy {
reason: format!(
"Custom model RAM must be a 0.5 GB (512 MB) increment; got {value} MB"
),
}),
None => Err(ResourcePolicyError::InvalidPolicy {
reason: "Custom profile requires custom_max_model_mb".into(),
}),
},
ResourceProfile::Everyday | ResourceProfile::LocalFocused => {
if self.custom_max_model_mb.is_none() {
Ok(())
} else {
Err(ResourcePolicyError::InvalidPolicy {
reason: format!(
"{:?} profile must not set custom_max_model_mb",
self.profile
),
})
}
}
}
}
}
/// Static policy math. Live-memory evidence is added by the admission service.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct EffectiveResourceBudget {
pub total_memory_mb: u64,
pub emergency_reserve_mb: u64,
pub configured_model_ceiling_mb: u64,
pub effective_new_load_ceiling_mb: u64,
pub normalization_notice: Option<String>,
}
/// Measured dedicated accelerator memory. This is intentionally separate from
/// host RAM; CAR does not apply Apple unified-memory percentages to CUDA VRAM.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct AcceleratorResourceBudget {
pub total_mb: u64,
pub budget_mb: u64,
}
/// Host policy plus an optional distinct accelerator fit pool.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResourceEvaluation {
pub host_memory: EffectiveResourceBudget,
pub accelerator_memory: Option<AcceleratorResourceBudget>,
}
/// Why CAR believes the weight portion of a model-memory estimate.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ModelResourceEvidence {
/// The catalog declared a RAM or artifact-size floor.
CatalogExact,
/// A caller supplied a physical on-disk measurement.
FileSystemMeasured,
/// CAR estimated from parameter count because no stronger fact existed.
Heuristic,
}
/// Recommendation-time cold-load estimate. This is static capacity evidence,
/// not a live-memory probe or an operating-system RSS guarantee.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct ModelMemoryEstimate {
pub weights_mb: u64,
pub runtime_overhead_mb: u64,
pub context_overhead_mb: u64,
pub transient_margin_mb: u64,
pub estimated_peak_mb: u64,
pub evidence: ModelResourceEvidence,
}
/// Result of checking one CAR-managed local allocation against both the saved
/// policy and live machine headroom. `LiveMemoryUnknown` is intentionally not
/// collapsed into `Allowed`: callers may choose a conservative static-policy
/// fallback, but must not present the model as safe *right now*.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct LocalLoadPreflight {
pub model_id: String,
pub estimate: ModelMemoryEstimate,
pub configured_ceiling_mb: u64,
pub resident_model_mb: u64,
pub active_reservations_mb: u64,
pub estimated_incremental_mb: u64,
pub accelerator_total_mb: Option<u64>,
pub accelerator_resident_mb: Option<u64>,
pub accelerator_incremental_mb: Option<u64>,
pub live_available_mb: Option<u64>,
pub emergency_reserve_mb: u64,
pub verdict: LocalLoadVerdict,
}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum LocalLoadVerdict {
Allowed,
LiveMemoryUnknown,
DisabledByPolicy,
ExceedsConfiguredCeiling,
InsufficientLiveMemory,
ModelMaintenance,
/// A CAR-owned process is still being reaped. Its weights remain charged
/// and no replacement may peer-discount them until exit is confirmed.
PendingTeardown,
}
impl LocalLoadVerdict {
pub fn permits_static_fallback(&self) -> bool {
matches!(self, Self::Allowed | Self::LiveMemoryUnknown)
}
}
/// Machine-specific live-memory evidence. Returning `Ok(None)` is distinct
/// from zero available memory and is preserved in the public preflight.
pub trait LiveMemoryProbe: Send + Sync {
fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError>;
}
#[derive(Default)]
pub struct SystemLiveMemoryProbe;
impl LiveMemoryProbe for SystemLiveMemoryProbe {
fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError> {
Ok(crate::hardware::available_ram_mb())
}
}
#[cfg(test)]
pub(crate) struct FixedLiveMemoryProbe(Option<u64>);
#[cfg(test)]
impl FixedLiveMemoryProbe {
pub(crate) fn known(available_mb: u64) -> Self {
Self(Some(available_mb))
}
fn unknown() -> Self {
Self(None)
}
}
#[cfg(test)]
impl LiveMemoryProbe for FixedLiveMemoryProbe {
fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError> {
Ok(self.0)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum WeightPlacement {
Host,
Accelerator,
}
#[derive(Clone, Debug)]
struct ResidentAllocation {
weights_mb: u64,
placement: WeightPlacement,
logical_model_id: String,
}
#[derive(Clone, Default)]
struct AdmissionState {
resident_models: HashMap<String, ResidentAllocation>,
active_host_reservations_mb: u64,
active_accelerator_reservations_mb: u64,
active_by_model: HashMap<String, usize>,
maintenance_models: HashSet<String>,
pending_teardown_models: HashMap<String, HashSet<String>>,
model_aliases: HashMap<String, HashSet<String>>,
next_request_id: u64,
}
#[derive(Clone, Default)]
struct MachineAdmissionLedger {
resident_models: HashMap<(u64, String), ResidentAllocation>,
pending_allocations: HashMap<(u64, String), ResidentAllocation>,
active_host_by_owner: HashMap<u64, u64>,
active_accelerator_by_owner: HashMap<u64, u64>,
}
fn process_machine_admission_ledger() -> Arc<Mutex<MachineAdmissionLedger>> {
static LEDGER: OnceLock<Arc<Mutex<MachineAdmissionLedger>>> = OnceLock::new();
LEDGER
.get_or_init(|| Arc::new(Mutex::new(MachineAdmissionLedger::default())))
.clone()
}
fn next_admission_owner_id() -> u64 {
static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
}
/// Allocation identities are source-qualified so a worker process and a
/// supervised server for the same logical model cannot overwrite or evict one
/// another's physical-memory record.
pub fn worker_process_allocation_id(model_id: &str) -> String {
format!("worker:{model_id}")
}
pub fn vllm_process_allocation_id(model_id: &str) -> String {
format!("vllm:{model_id}")
}
/// One process-wide coordinator for local cold loads and per-request runtime
/// allocations. Its mutex makes check-and-reserve one atomic operation, so two
/// individually-safe requests cannot both consume the same headroom.
pub struct LocalAdmissionCoordinator {
policy: std::sync::RwLock<ResourcePolicy>,
hardware: HardwareInfo,
live_probe: Arc<dyn LiveMemoryProbe>,
state: Mutex<AdmissionState>,
machine_ledger: Arc<Mutex<MachineAdmissionLedger>>,
/// Cross-process activity ownership for exact resident allocation ids.
/// Kept outside cloneable accounting snapshots so an idle cache/process
/// holds its shared lease until exact eviction or confirmed teardown.
resident_activity_leases: Mutex<HashMap<String, Arc<crate::model_management::ModelLease>>>,
models_dir: std::sync::RwLock<Option<PathBuf>>,
/// The Hugging Face hub a row may be served from instead of `models_dir`
/// (see [`kv_config_path`]). `None` reads only `models_dir`, which keeps
/// tests from sizing rows off whatever the developer has cached.
hf_hub_dir: Option<PathBuf>,
kv_geometry_cache: Mutex<HashMap<String, (ConfigStamp, KvGeometry)>>,
owner_id: u64,
}
impl Drop for LocalAdmissionCoordinator {
fn drop(&mut self) {
let mut machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
machine
.resident_models
.retain(|(owner_id, _), _| *owner_id != self.owner_id);
machine
.pending_allocations
.retain(|(owner_id, _), _| *owner_id != self.owner_id);
machine.active_host_by_owner.remove(&self.owner_id);
machine.active_accelerator_by_owner.remove(&self.owner_id);
}
}
fn scoped_admission_registry() -> &'static Mutex<HashMap<PathBuf, Weak<LocalAdmissionCoordinator>>>
{
static REGISTRY: OnceLock<Mutex<HashMap<PathBuf, Weak<LocalAdmissionCoordinator>>>> =
OnceLock::new();
REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}
/// Stable identity for a state root. Existing roots are filesystem-
/// canonicalized so symlink/relative aliases cannot split admission accounting.
/// If canonicalization is unavailable (commonly before first-run directory
/// creation), lexical absolute normalization is deliberately conservative:
/// aliases may be charged independently, but never peer-discount one another.
pub fn normalized_state_root_key(state_root: &Path) -> PathBuf {
let absolute = if state_root.is_absolute() {
state_root.to_path_buf()
} else if let Ok(current) = std::env::current_dir() {
current.join(state_root)
} else {
return state_root.to_path_buf();
};
// Do not lexically collapse `..` before asking the filesystem. In
// `symlink/../state`, `..` applies after resolving the symlink and can name
// a different physical directory than a lexical pop would produce.
if let Ok(canonical) = car_winenv::canonicalize(&absolute) {
return canonical;
}
// First-run roots often do not exist yet. Canonicalize the nearest
// existing ancestor so a missing leaf beneath a symlink still shares the
// same coordinator/runtime identity as its real path.
let mut ancestor = absolute.clone();
let mut missing_suffix = Vec::new();
while let Some(component) = ancestor.components().next_back() {
let name = match component {
std::path::Component::Normal(name) => name.to_owned(),
std::path::Component::CurDir => std::ffi::OsString::from("."),
std::path::Component::ParentDir => std::ffi::OsString::from(".."),
std::path::Component::RootDir | std::path::Component::Prefix(_) => break,
};
missing_suffix.push(name);
ancestor.pop();
if let Ok(mut canonical) = car_winenv::canonicalize(&ancestor) {
for component in missing_suffix.iter().rev() {
if component == std::ffi::OsStr::new(".") {
continue;
}
if component == std::ffi::OsStr::new("..") {
canonical.pop();
} else {
canonical.push(component);
}
}
return canonical;
}
}
// No ancestor could be canonicalized. Keep the absolute spelling rather
// than inventing peer identity; conservative duplicate charging is safer.
absolute
}
pub fn install_shared_local_admission(coordinator: Arc<LocalAdmissionCoordinator>) {
let root = normalized_state_root_key(&car_home::root_or_relative());
scoped_admission_registry()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.entry(root)
.or_insert_with(|| Arc::downgrade(&coordinator));
}
pub fn shared_local_admission() -> Arc<LocalAdmissionCoordinator> {
let root = car_home::root_or_relative();
let policy = FileResourcePolicyRepository::new(root.clone())
.load()
.unwrap_or_else(|_| ResourcePolicy::everyday());
scoped_local_admission(root, policy, HardwareInfo::detect())
}
pub fn scoped_local_admission(
state_root: impl AsRef<Path>,
policy: ResourcePolicy,
hardware: HardwareInfo,
) -> Arc<LocalAdmissionCoordinator> {
scoped_local_admission_with_models_dir(state_root, policy, hardware, None)
}
/// Return the admission coordinator for a state root, with optional access to
/// installed model configs for architecture-aware context-memory estimates.
pub fn scoped_local_admission_with_models_dir(
state_root: impl AsRef<Path>,
policy: ResourcePolicy,
hardware: HardwareInfo,
models_dir: Option<PathBuf>,
) -> Arc<LocalAdmissionCoordinator> {
let state_root = normalized_state_root_key(state_root.as_ref());
let mut registry = scoped_admission_registry()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
registry.retain(|_, coordinator| coordinator.strong_count() > 0);
if let Some(existing) = registry.get(&state_root).and_then(Weak::upgrade) {
existing.set_policy(policy);
if let Some(models_dir) = models_dir {
existing.set_models_dir(models_dir);
}
return existing;
}
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe_ledger_and_models_dir(
policy,
hardware,
Arc::new(SystemLiveMemoryProbe),
process_machine_admission_ledger(),
models_dir,
Some(crate::hf_cache::hub_dir()),
));
registry.insert(state_root, Arc::downgrade(&coordinator));
coordinator
}
pub fn local_admission_for_scope(
state_root: impl AsRef<Path>,
) -> Option<Arc<LocalAdmissionCoordinator>> {
let state_root = normalized_state_root_key(state_root.as_ref());
scoped_admission_registry()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(&state_root)
.and_then(Weak::upgrade)
}
impl LocalAdmissionCoordinator {
fn built_in_model_identity(model_id: &str) -> Option<String> {
let lower = model_id.to_ascii_lowercase();
if lower.contains("kokoro-82m") {
if lower.contains("6bit") {
return Some("mlx/kokoro-82m:6bit".into());
}
if lower.contains("bf16") {
return Some("mlx/kokoro-82m:bf16".into());
}
}
None
}
fn resolve_model_identity(state: &AdmissionState, model_id: &str) -> String {
state
.model_aliases
.iter()
.find_map(|(canonical, aliases)| {
(canonical == model_id || aliases.contains(model_id)).then(|| canonical.clone())
})
.or_else(|| Self::built_in_model_identity(model_id))
.unwrap_or_else(|| model_id.to_string())
}
/// Resolve any registered catalog/artifact/provider alias to the single
/// allocation key used by admission, caches, and maintenance.
pub fn canonical_model_id(&self, model_id: &str) -> String {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
Self::resolve_model_identity(&state, model_id)
}
pub fn register_model_aliases<I, S>(&self, canonical_model_id: &str, aliases: I)
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
let mut aliases = aliases
.into_iter()
.map(Into::into)
.collect::<HashSet<String>>();
aliases.insert(canonical_model_id.to_string());
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut canonical_model_id = Self::built_in_model_identity(canonical_model_id)
.unwrap_or_else(|| canonical_model_id.to_string());
let intersecting = state
.model_aliases
.iter()
.filter(|(existing, existing_aliases)| {
aliases.contains(*existing)
|| existing_aliases.iter().any(|alias| aliases.contains(alias))
})
.map(|(existing, _)| existing.clone())
.collect::<Vec<_>>();
if let Some(existing) = intersecting.first() {
canonical_model_id = existing.clone();
}
for existing in intersecting {
if let Some(existing_aliases) = state.model_aliases.remove(&existing) {
aliases.extend(existing_aliases);
}
aliases.insert(existing);
}
aliases.insert(canonical_model_id.clone());
let mut active = 0usize;
for alias in &aliases {
active = active.saturating_add(state.active_by_model.remove(alias).unwrap_or_default());
}
if active > 0 {
*state
.active_by_model
.entry(canonical_model_id.clone())
.or_default() += active;
}
let mut maintenance = false;
for alias in &aliases {
maintenance |= state.maintenance_models.remove(alias);
}
if maintenance {
state.maintenance_models.insert(canonical_model_id.clone());
}
let mut pending = HashSet::new();
for alias in &aliases {
pending.extend(
state
.pending_teardown_models
.remove(alias)
.unwrap_or_default(),
);
}
if !pending.is_empty() {
state
.pending_teardown_models
.entry(canonical_model_id.clone())
.or_default()
.extend(pending);
}
for resident in state.resident_models.values_mut() {
if aliases.contains(&resident.logical_model_id) {
resident.logical_model_id = canonical_model_id.clone();
}
}
state
.model_aliases
.insert(canonical_model_id.clone(), aliases.clone());
// Lock order is always scoped state then machine ledger. Holding both
// here makes late alias registration one atomic identity migration;
// reserve cannot observe half-rekeyed maintenance/residency state.
let mut machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
for ((owner_id, _), resident) in machine.resident_models.iter_mut() {
if *owner_id == self.owner_id && aliases.contains(&resident.logical_model_id) {
resident.logical_model_id = canonical_model_id.clone();
}
}
for ((owner_id, _), resident) in machine.pending_allocations.iter_mut() {
if *owner_id == self.owner_id && aliases.contains(&resident.logical_model_id) {
resident.logical_model_id = canonical_model_id.clone();
}
}
}
pub fn new(policy: ResourcePolicy, hardware: HardwareInfo) -> Self {
Self::with_probe(policy, hardware, Arc::new(SystemLiveMemoryProbe))
}
pub fn with_probe(
policy: ResourcePolicy,
hardware: HardwareInfo,
live_probe: Arc<dyn LiveMemoryProbe>,
) -> Self {
Self::with_probe_and_ledger(
policy,
hardware,
live_probe,
Arc::new(Mutex::new(MachineAdmissionLedger::default())),
)
}
fn with_probe_and_ledger(
policy: ResourcePolicy,
hardware: HardwareInfo,
live_probe: Arc<dyn LiveMemoryProbe>,
machine_ledger: Arc<Mutex<MachineAdmissionLedger>>,
) -> Self {
Self::with_probe_ledger_and_models_dir(
policy,
hardware,
live_probe,
machine_ledger,
None,
None,
)
}
fn with_probe_ledger_and_models_dir(
policy: ResourcePolicy,
hardware: HardwareInfo,
live_probe: Arc<dyn LiveMemoryProbe>,
machine_ledger: Arc<Mutex<MachineAdmissionLedger>>,
models_dir: Option<PathBuf>,
hf_hub_dir: Option<PathBuf>,
) -> Self {
Self {
policy: std::sync::RwLock::new(policy),
hardware,
live_probe,
state: Mutex::new(AdmissionState::default()),
machine_ledger,
resident_activity_leases: Mutex::new(HashMap::new()),
models_dir: std::sync::RwLock::new(models_dir),
hf_hub_dir,
kv_geometry_cache: Mutex::new(HashMap::new()),
owner_id: next_admission_owner_id(),
}
}
#[cfg(test)]
fn with_probe_and_models_dir(
policy: ResourcePolicy,
hardware: HardwareInfo,
live_probe: Arc<dyn LiveMemoryProbe>,
models_dir: Option<PathBuf>,
) -> Self {
Self::with_probe_ledger_and_models_dir(
policy,
hardware,
live_probe,
Arc::new(Mutex::new(MachineAdmissionLedger::default())),
models_dir,
None,
)
}
fn model_memory_estimate(
&self,
model: &ModelSchema,
context_tokens: usize,
) -> ModelMemoryEstimate {
let geometry = self.kv_geometry(model);
estimate_model_memory_with_geometry(
model,
&self.hardware,
context_tokens,
None,
geometry.as_ref(),
)
}
/// The installed checkpoint's KV geometry under this coordinator's models
/// directory, cached per model id together with the `config.json` stamp it
/// was read from. Every lookup stats the file: a removed checkpoint drops
/// its entry and falls back, and a replaced or re-downloaded one (new
/// mtime or length) is read again, so neither the catalog nor admission
/// sizes a row from a config that is no longer there.
fn kv_geometry(&self, model: &ModelSchema) -> Option<KvGeometry> {
let models_dir = self
.models_dir
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
let mut cache = self
.kv_geometry_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
// Stat before reading: a write landing between the two leaves an older
// stamp beside newer contents, which the next lookup reads again.
let Some(stamp) = models_dir.as_deref().and_then(|dir| {
ConfigStamp::of(&kv_config_path(model, dir, self.hf_hub_dir.as_deref()))
}) else {
cache.remove(&model.id);
return None;
};
if let Some((cached, geometry)) = cache.get(&model.id) {
if *cached == stamp {
return Some(geometry.clone());
}
}
let geometry = kv_geometry_for(model, models_dir.as_deref(), self.hf_hub_dir.as_deref());
match geometry.as_ref() {
Some(geometry) => {
cache.insert(model.id.clone(), (stamp, geometry.clone()));
}
None => {
cache.remove(&model.id);
}
}
geometry
}
/// The geometry admission sizes these rows' KV cache from, for the
/// recommendation estimate to use the same input (car#2213).
pub fn checkpoint_geometry<'a>(
&self,
models: impl IntoIterator<Item = &'a ModelSchema>,
) -> CheckpointGeometry {
CheckpointGeometry {
by_id: models
.into_iter()
.filter_map(|model| Some((model.id.clone(), self.kv_geometry(model)?)))
.collect(),
}
}
fn set_models_dir(&self, models_dir: PathBuf) {
let mut configured = self
.models_dir
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if configured.as_ref() == Some(&models_dir) {
return;
}
*configured = Some(models_dir);
self.kv_geometry_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clear();
}
pub fn set_policy(&self, policy: ResourcePolicy) {
*self
.policy
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner) = policy;
}
pub fn policy(&self) -> ResourcePolicy {
self.policy
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
pub fn mark_resident(&self, model_id: &str, weights_mb: u64) {
let placement = self.default_weight_placement();
self.mark_resident_with_placement(model_id, model_id, weights_mb, placement);
}
pub fn mark_resident_allocation(
&self,
logical_model_id: &str,
allocation_id: &str,
weights_mb: u64,
) {
let placement = self.default_weight_placement();
self.mark_resident_with_placement(logical_model_id, allocation_id, weights_mb, placement);
}
pub fn mark_evicted(&self, model_id: &str) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
// Eviction is allocation-owner exact. Logical-model-wide removal can
// erase a sibling cache/process/voice lease that still owns weights.
state.resident_models.remove(model_id);
self.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.resident_models
.remove(&(self.owner_id, model_id.to_string()));
self.resident_activity_leases
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(model_id);
}
/// Enter fail-closed process teardown. Residency remains charged while a
/// background owner retains and reaps the child.
pub fn mark_teardown_pending(&self, model_id: &str) {
self.mark_teardown_pending_allocation(model_id, model_id);
}
/// Quarantine one concrete process allocation while retaining the logical
/// model identity used by maintenance and admission. Multiple process
/// owners for one model are reference-counted independently.
pub fn mark_teardown_pending_allocation(&self, logical_model_id: &str, allocation_id: &str) {
self.mark_teardown_pending_allocation_with_charge(logical_model_id, allocation_id, 0);
}
/// Transfer an exact process allocation into quarantine. If it has not yet
/// published residency (the pre-ACK cancellation case), `measured_bytes`
/// supplies the physical charge that must outlive the caller reservation.
/// Repeating this for the same allocation is idempotent.
pub fn mark_teardown_pending_allocation_with_charge(
&self,
logical_model_id: &str,
allocation_id: &str,
measured_bytes: u64,
) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, logical_model_id);
state
.pending_teardown_models
.entry(model_id.clone())
.or_default()
.insert(allocation_id.to_string());
let mut machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let key = (self.owner_id, allocation_id.to_string());
let machine_resident = machine.resident_models.remove(&key);
let scoped_resident = state.resident_models.get(allocation_id).cloned();
let resident = machine_resident.or(scoped_resident);
let measured_mb = measured_bytes.div_ceil(1024 * 1024);
let placement = self.default_weight_placement();
machine
.pending_allocations
.entry(key)
.and_modify(|pending| {
pending.weights_mb = pending.weights_mb.max(measured_mb);
pending.logical_model_id = model_id.clone();
})
.or_insert_with(|| {
resident.unwrap_or(ResidentAllocation {
weights_mb: measured_mb,
placement,
logical_model_id: model_id,
})
});
}
/// Clear a quarantined allocation only after its OS process was reaped.
pub fn finish_teardown(&self, model_id: &str) {
self.finish_teardown_allocation(model_id, model_id);
}
/// Finish one concrete process allocation after confirmed OS exit. This
/// removes only that owner's resident record; sibling allocations for the
/// same logical model remain charged and pending until their own ACK.
pub fn finish_teardown_allocation(&self, logical_model_id: &str, allocation_id: &str) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let logical_model_id = Self::resolve_model_identity(&state, logical_model_id);
let exact_pending = state
.pending_teardown_models
.get_mut(&logical_model_id)
.is_some_and(|pending| pending.remove(allocation_id));
if exact_pending
&& state
.pending_teardown_models
.get(&logical_model_id)
.is_some_and(HashSet::is_empty)
{
state.pending_teardown_models.remove(&logical_model_id);
}
if !exact_pending {
return;
}
state.resident_models.remove(allocation_id);
let mut machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
machine
.resident_models
.remove(&(self.owner_id, allocation_id.to_string()));
machine
.pending_allocations
.remove(&(self.owner_id, allocation_id.to_string()));
self.resident_activity_leases
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(allocation_id);
}
pub fn teardown_pending(&self, model_id: &str) -> bool {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, model_id);
state.pending_teardown_models.contains_key(&model_id)
}
pub fn is_resident(&self, model_id: &str) -> bool {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, model_id);
state
.resident_models
.iter()
.any(|(allocation_id, resident)| {
allocation_id == &model_id || resident.logical_model_id == model_id
})
}
pub fn resident_model_mb(&self) -> u64 {
let machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
machine
.resident_models
.values()
.chain(machine.pending_allocations.values())
.map(|resident| resident.weights_mb)
.fold(0, u64::saturating_add)
}
/// Test support for asserting one coordinator's accounting without
/// observing unrelated coordinators running in parallel.
#[doc(hidden)]
pub fn owned_resident_model_mb_for_testing(&self) -> u64 {
let machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
machine
.resident_models
.iter()
.chain(machine.pending_allocations.iter())
.filter(|((owner_id, _), _)| *owner_id == self.owner_id)
.map(|(_, resident)| resident.weights_mb)
.fold(0, u64::saturating_add)
}
pub fn active_request_count(&self, model_id: &str) -> usize {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, model_id);
state
.active_by_model
.get(&model_id)
.copied()
.unwrap_or_default()
}
/// The logical models this controller holds resident, with the megabytes
/// each holds, largest first (ties by id, so the order is stable).
pub fn resident_logical_models(&self) -> Vec<(String, u64)> {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut by_model: HashMap<String, u64> = HashMap::new();
for resident in state.resident_models.values() {
let held = by_model
.entry(resident.logical_model_id.clone())
.or_default();
*held = held.saturating_add(resident.weights_mb);
}
let mut models = by_model.into_iter().collect::<Vec<_>>();
models.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
models
}
/// Exact resident allocation keys for a logical model. Engine caches use
/// the logical ID directly; independent voice/provider instances use
/// allocation-qualified keys while retaining the same logical alias.
pub fn resident_allocation_ids(&self, model_id: &str) -> Vec<String> {
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, model_id);
let mut allocations = state
.resident_models
.iter()
.filter(|(allocation_id, resident)| {
allocation_id.as_str() == model_id || resident.logical_model_id == model_id
})
.map(|(allocation_id, _)| allocation_id.clone())
.collect::<Vec<_>>();
if let Some(pending) = state.pending_teardown_models.get(&model_id) {
allocations.extend(pending.iter().cloned());
}
allocations.sort();
allocations.dedup();
allocations
}
/// Whether model weights are charged to host memory (every Mac, and any
/// machine without a CUDA GPU of known size), where the configured
/// allocation is the ceiling that refuses a load.
pub fn places_weights_on_host(&self) -> bool {
self.default_weight_placement() == WeightPlacement::Host
}
fn default_weight_placement(&self) -> WeightPlacement {
if matches!(self.hardware.gpu_backend, GpuBackend::Cuda)
&& self.hardware.gpu_memory_mb.is_some()
{
WeightPlacement::Accelerator
} else {
WeightPlacement::Host
}
}
fn mark_resident_with_placement(
&self,
logical_model_id: &str,
allocation_id: &str,
weights_mb: u64,
placement: WeightPlacement,
) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let logical_model_id = Self::resolve_model_identity(&state, logical_model_id);
let allocation = ResidentAllocation {
weights_mb,
placement,
logical_model_id: logical_model_id.clone(),
};
state
.resident_models
.entry(allocation_id.to_string())
.and_modify(|resident| {
resident.weights_mb = resident.weights_mb.max(weights_mb);
resident.placement = placement;
resident.logical_model_id = logical_model_id.clone();
})
.or_insert(ResidentAllocation {
weights_mb,
placement,
logical_model_id,
});
// Preserve the global lock order: scoped state, then machine ledger.
self.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.resident_models
.insert((self.owner_id, allocation_id.to_string()), allocation);
}
pub fn preflight(&self, model: &ModelSchema, context_tokens: usize) -> LocalLoadPreflight {
let live_available_mb = self.live_probe.available_memory_mb().ok().flatten();
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
self.preflight_locked(model, context_tokens, &state, &machine, live_available_mb)
}
pub fn reserve(
self: &Arc<Self>,
model: &ModelSchema,
context_tokens: usize,
) -> Result<LocalLoadReservation, LocalAdmissionError> {
let live_available_mb = self.live_probe.available_memory_mb().ok().flatten();
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, &model.id);
let estimate = self.model_memory_estimate(model, context_tokens);
let mut preflight = self.preflight_estimate_locked(
&model_id,
estimate,
self.default_weight_placement(),
&state,
&machine,
live_available_mb,
);
if state.pending_teardown_models.contains_key(&model_id) {
preflight.verdict = LocalLoadVerdict::PendingTeardown;
return Err(LocalAdmissionError { preflight });
}
if state.maintenance_models.contains(&model_id) {
preflight.verdict = LocalLoadVerdict::ModelMaintenance;
return Err(LocalAdmissionError { preflight });
}
if !preflight.verdict.permits_static_fallback() {
return Err(LocalAdmissionError { preflight });
}
if preflight.verdict == LocalLoadVerdict::LiveMemoryUnknown {
tracing::warn!(
model = %model.id,
configured_ceiling_mb = preflight.configured_ceiling_mb,
estimated_incremental_mb = preflight.estimated_incremental_mb,
"live memory is unknown; proceeding under the static configured ceiling only"
);
}
state.active_host_reservations_mb = state
.active_host_reservations_mb
.saturating_add(preflight.estimated_incremental_mb);
state.active_accelerator_reservations_mb = state
.active_accelerator_reservations_mb
.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
let host = machine
.active_host_by_owner
.entry(self.owner_id)
.or_default();
*host = host.saturating_add(preflight.estimated_incremental_mb);
let accelerator = machine
.active_accelerator_by_owner
.entry(self.owner_id)
.or_default();
*accelerator =
accelerator.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
*state.active_by_model.entry(model_id.clone()).or_default() += 1;
state.next_request_id = state.next_request_id.wrapping_add(1);
let request_id = state.next_request_id;
Ok(LocalLoadReservation {
request_id,
model_id: model_id.clone(),
maintenance_model_id: model_id.clone(),
weights_mb: preflight.estimate.weights_mb,
reserved_incremental_mb: preflight.estimated_incremental_mb,
reserved_accelerator_mb: preflight.accelerator_incremental_mb.unwrap_or_default(),
cold_weights_reserved: !Self::has_resident_model(&state, &model_id),
placement: self.default_weight_placement(),
coordinator: Arc::clone(self),
charge: Arc::new(ReservationCharge::new(
model_id,
preflight.estimated_incremental_mb,
preflight.accelerator_incremental_mb.unwrap_or_default(),
Arc::clone(self),
)),
})
}
/// Reserve an installed host-memory allocation (Whisper/ONNX voice paths).
/// The caller downloads first, measures the resulting artifact, then calls
/// this immediately before constructing the native runtime.
pub fn reserve_measured_host(
self: &Arc<Self>,
model_id: &str,
measured_weights_bytes: u64,
request_overhead_mb: u64,
) -> Result<LocalLoadReservation, LocalAdmissionError> {
self.reserve_measured(
model_id,
model_id,
measured_weights_bytes,
request_overhead_mb,
WeightPlacement::Host,
)
}
pub fn reserve_measured_host_allocation(
self: &Arc<Self>,
logical_model_id: &str,
allocation_id: &str,
measured_weights_bytes: u64,
request_overhead_mb: u64,
) -> Result<LocalLoadReservation, LocalAdmissionError> {
self.reserve_measured(
logical_model_id,
allocation_id,
measured_weights_bytes,
request_overhead_mb,
WeightPlacement::Host,
)
}
fn reserve_measured(
self: &Arc<Self>,
logical_model_id: &str,
allocation_id: &str,
measured_weights_bytes: u64,
request_overhead_mb: u64,
placement: WeightPlacement,
) -> Result<LocalLoadReservation, LocalAdmissionError> {
let weights_mb = measured_weights_bytes.div_ceil(1024 * 1024);
let estimate = ModelMemoryEstimate {
weights_mb,
runtime_overhead_mb: request_overhead_mb,
context_overhead_mb: 0,
transient_margin_mb: 0,
estimated_peak_mb: weights_mb.saturating_add(request_overhead_mb),
evidence: ModelResourceEvidence::FileSystemMeasured,
};
let live_available_mb = self.live_probe.available_memory_mb().ok().flatten();
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut machine = self
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let logical_model_id = Self::resolve_model_identity(&state, logical_model_id);
let mut preflight = self.preflight_estimate_locked(
allocation_id,
estimate,
placement,
&state,
&machine,
live_available_mb,
);
if state
.pending_teardown_models
.contains_key(&logical_model_id)
{
preflight.verdict = LocalLoadVerdict::PendingTeardown;
}
if state.maintenance_models.contains(&logical_model_id) {
preflight.verdict = LocalLoadVerdict::ModelMaintenance;
}
if !preflight.verdict.permits_static_fallback() {
return Err(LocalAdmissionError { preflight });
}
state.active_host_reservations_mb = state
.active_host_reservations_mb
.saturating_add(preflight.estimated_incremental_mb);
state.active_accelerator_reservations_mb = state
.active_accelerator_reservations_mb
.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
let host = machine
.active_host_by_owner
.entry(self.owner_id)
.or_default();
*host = host.saturating_add(preflight.estimated_incremental_mb);
let accelerator = machine
.active_accelerator_by_owner
.entry(self.owner_id)
.or_default();
*accelerator =
accelerator.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
*state
.active_by_model
.entry(logical_model_id.clone())
.or_default() += 1;
state.next_request_id = state.next_request_id.wrapping_add(1);
Ok(LocalLoadReservation {
request_id: state.next_request_id,
model_id: allocation_id.to_string(),
maintenance_model_id: logical_model_id.clone(),
weights_mb,
reserved_incremental_mb: preflight.estimated_incremental_mb,
reserved_accelerator_mb: preflight.accelerator_incremental_mb.unwrap_or_default(),
cold_weights_reserved: !state.resident_models.contains_key(allocation_id),
placement,
coordinator: Arc::clone(self),
charge: Arc::new(ReservationCharge::new(
logical_model_id,
preflight.estimated_incremental_mb,
preflight.accelerator_incremental_mb.unwrap_or_default(),
Arc::clone(self),
)),
})
}
/// Exclude new reservations for one model while removal/maintenance checks
/// cache and cross-process lease state. The check and exclusion happen
/// under the same mutex as `reserve`, closing the removal-vs-load race.
pub fn begin_model_maintenance(
self: &Arc<Self>,
model_id: &str,
) -> Result<LocalModelMaintenanceGuard, ModelMaintenanceError> {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let model_id = Self::resolve_model_identity(&state, model_id);
if state
.active_by_model
.get(&model_id)
.copied()
.unwrap_or_default()
> 0
{
return Err(ModelMaintenanceError::ModelInUse(model_id));
}
if !state.maintenance_models.insert(model_id.clone()) {
return Err(ModelMaintenanceError::AlreadyInMaintenance(model_id));
}
Ok(LocalModelMaintenanceGuard {
model_id,
coordinator: Arc::clone(self),
})
}
fn preflight_locked(
&self,
model: &ModelSchema,
context_tokens: usize,
state: &AdmissionState,
machine: &MachineAdmissionLedger,
live_available_mb: Option<u64>,
) -> LocalLoadPreflight {
let estimate = self.model_memory_estimate(model, context_tokens);
let model_id = Self::resolve_model_identity(state, &model.id);
self.preflight_estimate_locked(
&model_id,
estimate,
self.default_weight_placement(),
state,
machine,
live_available_mb,
)
}
fn preflight_estimate_locked(
&self,
model_id: &str,
estimate: ModelMemoryEstimate,
placement: WeightPlacement,
state: &AdmissionState,
machine: &MachineAdmissionLedger,
live_available_mb: Option<u64>,
) -> LocalLoadPreflight {
let policy = self.policy();
let budget = policy.effective_budget(self.hardware.total_ram_mb);
let host_resident_mb = machine
.resident_models
.values()
.chain(machine.pending_allocations.values())
.filter(|resident| resident.placement == WeightPlacement::Host)
.map(|resident| resident.weights_mb)
.fold(0, u64::saturating_add);
let accelerator_resident_weights_mb = machine
.resident_models
.values()
.chain(machine.pending_allocations.values())
.filter(|resident| resident.placement == WeightPlacement::Accelerator)
.map(|resident| resident.weights_mb)
.fold(0, u64::saturating_add);
let already_resident = Self::has_resident_model(state, model_id);
let request_overhead_mb = estimate
.context_overhead_mb
.saturating_add(estimate.runtime_overhead_mb)
.saturating_add(estimate.transient_margin_mb);
let cold_weights_mb = if already_resident {
0
} else {
estimate.weights_mb
};
let resident_model_mb = host_resident_mb;
let host_cold_weights_mb = if placement == WeightPlacement::Host {
cold_weights_mb
} else {
0
};
let estimated_incremental_mb = request_overhead_mb.saturating_add(host_cold_weights_mb);
let accelerator_total_mb = (placement == WeightPlacement::Accelerator)
.then_some(self.hardware.gpu_memory_mb)
.flatten();
let accelerator_resident_mb =
(placement == WeightPlacement::Accelerator).then_some(accelerator_resident_weights_mb);
let accelerator_incremental_mb =
(placement == WeightPlacement::Accelerator).then_some(cold_weights_mb);
let projected_static_mb = resident_model_mb
.saturating_add(machine.active_host_by_owner.values().copied().sum::<u64>())
.saturating_add(estimated_incremental_mb);
let projected_accelerator_mb = accelerator_resident_weights_mb
.saturating_add(
machine
.active_accelerator_by_owner
.values()
.copied()
.sum::<u64>(),
)
.saturating_add(cold_weights_mb);
let verdict = if budget.effective_new_load_ceiling_mb == 0 && !already_resident {
LocalLoadVerdict::DisabledByPolicy
} else if (budget.configured_model_ceiling_mb != 0
&& projected_static_mb > budget.configured_model_ceiling_mb)
|| accelerator_total_mb.is_some_and(|vram_mb| projected_accelerator_mb > vram_mb)
{
LocalLoadVerdict::ExceedsConfiguredCeiling
} else if let Some(available_mb) = live_available_mb {
let unreserved_available_mb = available_mb
.saturating_sub(machine.active_host_by_owner.values().copied().sum::<u64>());
if unreserved_available_mb
< estimated_incremental_mb.saturating_add(budget.emergency_reserve_mb)
{
LocalLoadVerdict::InsufficientLiveMemory
} else {
LocalLoadVerdict::Allowed
}
} else {
LocalLoadVerdict::LiveMemoryUnknown
};
LocalLoadPreflight {
model_id: model_id.to_string(),
estimate,
configured_ceiling_mb: budget.configured_model_ceiling_mb,
resident_model_mb,
active_reservations_mb: machine.active_host_by_owner.values().copied().sum(),
estimated_incremental_mb,
accelerator_total_mb,
accelerator_resident_mb,
accelerator_incremental_mb,
live_available_mb,
emergency_reserve_mb: budget.emergency_reserve_mb,
verdict,
}
}
fn has_resident_model(state: &AdmissionState, model_id: &str) -> bool {
state
.resident_models
.iter()
.any(|(allocation_id, resident)| {
allocation_id == model_id || resident.logical_model_id == model_id
})
}
}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct LocalAdmissionError {
pub preflight: LocalLoadPreflight,
}
#[derive(Clone, Debug, Error, PartialEq, Eq)]
pub enum ModelMaintenanceError {
#[error("local model '{0}' is in use")]
ModelInUse(String),
#[error("local model '{0}' already has maintenance in progress")]
AlreadyInMaintenance(String),
#[error("failed to release local model residency: {0}")]
ReleaseFailed(String),
#[error("local worker did not acknowledge release of model '{0}'")]
WorkerReleaseUnacknowledged(String),
#[error("supervised local process did not acknowledge release of model '{0}'")]
ProcessReleaseUnacknowledged(String),
#[error("an in-process cache still has active work for local model '{0}'")]
CacheReleaseBlocked(String),
#[error("local model '{model_id}' still has resident allocations: {allocation_ids:?}")]
ResidualResidency {
model_id: String,
allocation_ids: Vec<String>,
},
}
pub struct LocalModelMaintenanceGuard {
model_id: String,
coordinator: Arc<LocalAdmissionCoordinator>,
}
impl Drop for LocalModelMaintenanceGuard {
fn drop(&mut self) {
self.coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.maintenance_models
.remove(&self.model_id);
}
}
impl std::fmt::Display for LocalAdmissionError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
formatter,
"local model '{}' blocked by {:?}: needs {} MB incremental, {} MB live available",
self.preflight.model_id,
self.preflight.verdict,
self.preflight.estimated_incremental_mb,
self.preflight
.live_available_mb
.map(|value| value.to_string())
.unwrap_or_else(|| "unknown".into())
)
}
}
impl std::error::Error for LocalAdmissionError {}
/// RAII guard for one request's estimated incremental allocation. Dropping on
/// success, error, panic unwind, or cancellation releases the reservation.
pub struct LocalLoadReservation {
request_id: u64,
model_id: String,
maintenance_model_id: String,
weights_mb: u64,
reserved_incremental_mb: u64,
reserved_accelerator_mb: u64,
cold_weights_reserved: bool,
placement: WeightPlacement,
coordinator: Arc<LocalAdmissionCoordinator>,
charge: Arc<ReservationCharge>,
}
/// Cloneable ownership for native work that outlives its awaiting request.
/// `spawn_blocking` tasks and supervised subprocess cleanup retain this lease
/// after cancellation so machine headroom is not advertised while allocation
/// work is still running.
#[derive(Clone)]
pub struct DetachedLocalLease {
_charge: Arc<ReservationCharge>,
}
struct ReservationCharge {
maintenance_model_id: String,
host_mb: AtomicU64,
accelerator_mb: AtomicU64,
cold_weights_transferred: AtomicBool,
coordinator: Arc<LocalAdmissionCoordinator>,
activity_lease: Mutex<Option<Arc<crate::model_management::ModelLease>>>,
}
impl ReservationCharge {
fn new(
maintenance_model_id: String,
host_mb: u64,
accelerator_mb: u64,
coordinator: Arc<LocalAdmissionCoordinator>,
) -> Self {
Self {
maintenance_model_id,
host_mb: AtomicU64::new(host_mb),
accelerator_mb: AtomicU64::new(accelerator_mb),
cold_weights_transferred: AtomicBool::new(false),
coordinator,
activity_lease: Mutex::new(None),
}
}
fn update(&self, host_mb: u64, accelerator_mb: u64) {
self.host_mb.store(host_mb, Ordering::Release);
self.accelerator_mb.store(accelerator_mb, Ordering::Release);
}
}
impl Drop for ReservationCharge {
fn drop(&mut self) {
let host_mb = self.host_mb.load(Ordering::Acquire);
let accelerator_mb = self.accelerator_mb.load(Ordering::Acquire);
let mut state = self
.coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut machine = self
.coordinator
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.active_host_reservations_mb =
state.active_host_reservations_mb.saturating_sub(host_mb);
state.active_accelerator_reservations_mb = state
.active_accelerator_reservations_mb
.saturating_sub(accelerator_mb);
if let Some(host) = machine
.active_host_by_owner
.get_mut(&self.coordinator.owner_id)
{
*host = host.saturating_sub(host_mb);
}
if let Some(accelerator) = machine
.active_accelerator_by_owner
.get_mut(&self.coordinator.owner_id)
{
*accelerator = accelerator.saturating_sub(accelerator_mb);
}
let maintenance_model_id =
LocalAdmissionCoordinator::resolve_model_identity(&state, &self.maintenance_model_id);
if let Some(active) = state.active_by_model.get_mut(&maintenance_model_id) {
*active = active.saturating_sub(1);
if *active == 0 {
state.active_by_model.remove(&maintenance_model_id);
}
}
}
}
impl std::fmt::Debug for LocalLoadReservation {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("LocalLoadReservation")
.field("request_id", &self.request_id)
.field("model_id", &self.model_id)
.field("reserved_incremental_mb", &self.reserved_incremental_mb)
.field("reserved_accelerator_mb", &self.reserved_accelerator_mb)
.finish_non_exhaustive()
}
}
impl LocalLoadReservation {
pub fn request_id(&self) -> u64 {
self.request_id
}
/// Model whose allocation this reservation protects. Callers that walk an
/// adaptive route can use this to avoid reusing one candidate's reservation
/// for a different local fallback.
pub fn model_id(&self) -> &str {
&self.model_id
}
pub fn authorizes_model(&self, model_id: &str) -> bool {
self.model_id == model_id || self.maintenance_model_id == model_id
}
/// Bind an admitted cache load to its concrete allocation owner. Distinct
/// cache/runtime generations for one logical model must not peer-discount
/// each other's physical weights.
pub(crate) fn bind_allocation_id(&mut self, allocation_id: &str) {
self.model_id = allocation_id.to_string();
}
pub fn reserved_incremental_mb(&self) -> u64 {
self.reserved_incremental_mb
}
pub fn reconciled_weights_bytes(&self) -> u64 {
self.weights_mb.saturating_mul(1024 * 1024)
}
pub fn detached_lease(&self) -> DetachedLocalLease {
DetachedLocalLease {
_charge: self.charge.clone(),
}
}
pub(crate) fn attach_activity_lease(&mut self, lease: crate::model_management::ModelLease) {
*self
.charge
.activity_lease
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Arc::new(lease));
}
/// Atomically replace this request's cold-weight charge with a pending
/// process allocation. Request overhead remains active until the request
/// reservation and any detached cleanup lease are released.
pub(crate) fn transfer_cold_weights_to_pending_allocation(
&self,
allocation_id: &str,
measured_weights_bytes: u64,
) {
let mut state = self
.coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let logical_model_id =
LocalAdmissionCoordinator::resolve_model_identity(&state, &self.maintenance_model_id);
state
.pending_teardown_models
.entry(logical_model_id.clone())
.or_default()
.insert(allocation_id.to_string());
let mut machine = self
.coordinator
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let key = (self.coordinator.owner_id, allocation_id.to_string());
let machine_resident = machine.resident_models.remove(&key);
let scoped_resident = state.resident_models.get(allocation_id).cloned();
let resident = machine_resident.or(scoped_resident);
let measured_mb = measured_weights_bytes.div_ceil(1024 * 1024);
machine
.pending_allocations
.entry(key)
.and_modify(|pending| {
pending.weights_mb = pending.weights_mb.max(measured_mb);
pending.logical_model_id = logical_model_id.clone();
})
.or_insert_with(|| {
resident.unwrap_or(ResidentAllocation {
weights_mb: measured_mb,
placement: self.placement,
logical_model_id,
})
});
if self.cold_weights_reserved
&& self
.charge
.cold_weights_transferred
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
match self.placement {
WeightPlacement::Host => {
state.active_host_reservations_mb = state
.active_host_reservations_mb
.saturating_sub(self.weights_mb);
let active = machine
.active_host_by_owner
.entry(self.coordinator.owner_id)
.or_default();
*active = active.saturating_sub(self.weights_mb);
let _ = self.charge.host_mb.fetch_update(
Ordering::AcqRel,
Ordering::Acquire,
|host_mb| Some(host_mb.saturating_sub(self.weights_mb)),
);
}
WeightPlacement::Accelerator => {
state.active_accelerator_reservations_mb = state
.active_accelerator_reservations_mb
.saturating_sub(self.weights_mb);
let active = machine
.active_accelerator_by_owner
.entry(self.coordinator.owner_id)
.or_default();
*active = active.saturating_sub(self.weights_mb);
let _ = self.charge.accelerator_mb.fetch_update(
Ordering::AcqRel,
Ordering::Acquire,
|accelerator_mb| Some(accelerator_mb.saturating_sub(self.weights_mb)),
);
}
}
}
}
fn sync_shared_charge(&self) {
self.charge
.update(self.reserved_incremental_mb, self.reserved_accelerator_mb);
}
/// Replace the catalog estimate with an installed on-disk measurement.
/// Check and reservation replacement happen under the same coordinator
/// mutex; rejection leaves the original reservation untouched.
pub fn reconcile_measured_weights(
&mut self,
measured_weights_bytes: u64,
) -> Result<LocalLoadPreflight, LocalAdmissionError> {
if !self.cold_weights_reserved {
let live_available_mb = self
.coordinator
.live_probe
.available_memory_mb()
.ok()
.flatten();
let state = self
.coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let machine = self
.coordinator
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if LocalAdmissionCoordinator::has_resident_model(&state, &self.model_id) {
let estimate = self.measured_estimate(self.weights_mb);
return Ok(self.coordinator.preflight_estimate_locked(
&self.model_id,
estimate,
self.placement,
&state,
&machine,
live_available_mb,
));
}
drop(state);
self.cold_weights_reserved = true;
}
let measured_mb = measured_weights_bytes.div_ceil(1024 * 1024);
let live_available_mb = self
.coordinator
.live_probe
.available_memory_mb()
.ok()
.flatten();
let mut state = self
.coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut machine = self
.coordinator
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let published_by_peer =
LocalAdmissionCoordinator::has_resident_model(&state, &self.model_id);
let mut without_current = state.clone();
without_current.active_host_reservations_mb = without_current
.active_host_reservations_mb
.saturating_sub(self.reserved_incremental_mb);
without_current.active_accelerator_reservations_mb = without_current
.active_accelerator_reservations_mb
.saturating_sub(self.reserved_accelerator_mb);
let mut without_machine = machine.clone();
let host = without_machine
.active_host_by_owner
.entry(self.coordinator.owner_id)
.or_default();
*host = host.saturating_sub(self.reserved_incremental_mb);
let accelerator = without_machine
.active_accelerator_by_owner
.entry(self.coordinator.owner_id)
.or_default();
*accelerator = accelerator.saturating_sub(self.reserved_accelerator_mb);
let estimate = self.measured_estimate(measured_mb);
let preflight = self.coordinator.preflight_estimate_locked(
&self.model_id,
estimate,
self.placement,
&without_current,
&without_machine,
live_available_mb,
);
if !preflight.verdict.permits_static_fallback() {
return Err(LocalAdmissionError { preflight });
}
state.active_host_reservations_mb = without_current
.active_host_reservations_mb
.saturating_add(preflight.estimated_incremental_mb);
state.active_accelerator_reservations_mb = without_current
.active_accelerator_reservations_mb
.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default());
machine.active_host_by_owner.insert(
self.coordinator.owner_id,
without_machine
.active_host_by_owner
.get(&self.coordinator.owner_id)
.copied()
.unwrap_or_default()
.saturating_add(preflight.estimated_incremental_mb),
);
machine.active_accelerator_by_owner.insert(
self.coordinator.owner_id,
without_machine
.active_accelerator_by_owner
.get(&self.coordinator.owner_id)
.copied()
.unwrap_or_default()
.saturating_add(preflight.accelerator_incremental_mb.unwrap_or_default()),
);
self.weights_mb = measured_mb;
self.reserved_incremental_mb = preflight.estimated_incremental_mb;
self.reserved_accelerator_mb = preflight.accelerator_incremental_mb.unwrap_or_default();
if published_by_peer {
self.cold_weights_reserved = false;
}
self.sync_shared_charge();
Ok(preflight)
}
fn measured_estimate(&self, weights_mb: u64) -> ModelMemoryEstimate {
let host_cold_mb = if self.cold_weights_reserved && self.placement == WeightPlacement::Host
{
self.weights_mb
} else {
0
};
let request_overhead_mb = self.reserved_incremental_mb.saturating_sub(host_cold_mb);
ModelMemoryEstimate {
weights_mb,
runtime_overhead_mb: request_overhead_mb,
context_overhead_mb: 0,
transient_margin_mb: 0,
estimated_peak_mb: weights_mb.saturating_add(request_overhead_mb),
evidence: ModelResourceEvidence::FileSystemMeasured,
}
}
/// Atomically transfer the cold portion of this request reservation into
/// resident accounting at the exact point a cache publishes the backend.
pub fn publish_resident_weights(&mut self, measured_weights_bytes: u64) {
let allocation_id = self.model_id.clone();
self.publish_resident_weights_as(&allocation_id, measured_weights_bytes);
}
/// Conservatively account an external allocation whose acknowledgement
/// reported a different concrete allocation id than the requested model.
/// This is only used when the external owner cannot confirm teardown.
pub fn publish_resident_weights_as(
&mut self,
allocation_id: &str,
measured_weights_bytes: u64,
) {
// Publication happens after the allocation already exists. It must not
// run another live-memory probe or reject at that point: doing so can
// return an error while a cache/process remains resident but
// unaccounted. The caller reconciles before loading; if an external
// allocator reports a different final size, conservatively account the
// actual size atomically even when it exceeds the earlier reservation.
let measured_mb = measured_weights_bytes.div_ceil(1024 * 1024);
let mut state = self
.coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut machine = self
.coordinator
.machine_ledger
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let key = (self.coordinator.owner_id, allocation_id.to_string());
let pending = machine.pending_allocations.remove(&key);
let logical_model_id =
LocalAdmissionCoordinator::resolve_model_identity(&state, &self.maintenance_model_id);
if let Some(allocations) = state.pending_teardown_models.get_mut(&logical_model_id) {
allocations.remove(allocation_id);
if allocations.is_empty() {
state.pending_teardown_models.remove(&logical_model_id);
}
}
if !self.cold_weights_reserved && pending.is_none() {
return;
}
if self.cold_weights_reserved {
let cold_weights_transferred =
self.charge.cold_weights_transferred.load(Ordering::Acquire);
match self.placement {
WeightPlacement::Host => {
if !cold_weights_transferred {
state.active_host_reservations_mb = state
.active_host_reservations_mb
.saturating_sub(self.weights_mb);
let active = machine
.active_host_by_owner
.entry(self.coordinator.owner_id)
.or_default();
*active = active.saturating_sub(self.weights_mb);
}
self.reserved_incremental_mb =
self.reserved_incremental_mb.saturating_sub(self.weights_mb);
}
WeightPlacement::Accelerator => {
if !cold_weights_transferred {
state.active_accelerator_reservations_mb = state
.active_accelerator_reservations_mb
.saturating_sub(self.weights_mb);
let active = machine
.active_accelerator_by_owner
.entry(self.coordinator.owner_id)
.or_default();
*active = active.saturating_sub(self.weights_mb);
}
self.reserved_accelerator_mb =
self.reserved_accelerator_mb.saturating_sub(self.weights_mb);
}
}
}
let resident = ResidentAllocation {
weights_mb: pending
.as_ref()
.map(|allocation| allocation.weights_mb)
.unwrap_or_default()
.max(measured_mb),
placement: pending
.as_ref()
.map(|allocation| allocation.placement)
.unwrap_or(self.placement),
logical_model_id,
};
state
.resident_models
.insert(allocation_id.to_string(), resident.clone());
machine.resident_models.insert(key, resident.clone());
if let Some(lease) = self
.charge
.activity_lease
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
{
self.coordinator
.resident_activity_leases
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(allocation_id.to_string(), lease);
}
self.weights_mb = resident.weights_mb;
self.cold_weights_reserved = false;
self.charge
.cold_weights_transferred
.store(false, Ordering::Release);
self.sync_shared_charge();
}
/// Transfer cold weight accounting into the resident set after a loader
/// successfully publishes the backend into its cache.
pub fn commit_resident_weights(&mut self) {
self.publish_resident_weights(self.weights_mb * 1024 * 1024);
}
}
/// What a checkpoint's own `config.json` says about the memory a request
/// costs beyond its weights, reduced to bytes.
///
/// Measured against vllm-mlx 0.6.0 on an M5 Pro (2026-10-26, car#3563
/// follow-up): Qwen3.8-27B 4-bit, whose cache grows 64 KiB per token, and
/// GLM-4.7-Flash 8-bit, whose MLA cache grows 53 KiB per token while its
/// prompt pass peaks near 340 KiB per token. Counting every layer as a
/// full-attention KV layer had overstated the first 4x and understated the
/// second by 8 GiB at 64k tokens.
#[derive(Clone, Debug, PartialEq, Eq)]
struct KvGeometry {
/// Cache bytes one context token adds, summed over the layers that keep a
/// per-token cache. A linear-attention layer keeps none.
kv_bytes_per_token: u64,
/// Bytes a sequence holds whatever its length: the recurrent and
/// convolution state of linear-attention layers.
fixed_state_bytes: u64,
/// Bytes per prompt token alive at once while the prompt is processed,
/// beyond the cache itself.
prefill_bytes_per_token: u64,
/// The language model's vocabulary, which sizes the logits of a prompt
/// chunk on runtimes that compute them for every position.
vocab_size: Option<u64>,
}
/// mlx-lm's and mlx-vlm's `prefill_step_size` default: the prompt is
/// processed in chunks of this many tokens.
const PREFILL_CHUNK_TOKENS: u64 = 2_048;
/// Installed checkpoints' own KV geometry, keyed by model id: the input that
/// lets the recommendation estimate size an installed row's KV cache from its
/// `config.json`, as load-time admission does, instead of from the parameter
/// count. A row absent here — not installed, or a config without readable
/// geometry — keeps the parameter-count fallback, as it does at admission.
/// [`CheckpointGeometry::default`] holds nothing, so every row falls back.
#[derive(Clone, Debug, Default)]
pub struct CheckpointGeometry {
by_id: HashMap<String, KvGeometry>,
}
impl CheckpointGeometry {
/// Read each row's `<models_dir>/<name>/config.json`, the file admission
/// reads. Callers that hold an engine use
/// [`LocalAdmissionCoordinator::checkpoint_geometry`] instead, which shares
/// admission's models directory and cache.
pub fn read<'a>(models: impl IntoIterator<Item = &'a ModelSchema>, models_dir: &Path) -> Self {
Self {
by_id: models
.into_iter()
.filter_map(|model| {
Some((
model.id.clone(),
kv_geometry_for(model, Some(models_dir), None)?,
))
})
.collect(),
}
}
fn get(&self, model: &ModelSchema) -> Option<&KvGeometry> {
self.by_id.get(&model.id)
}
}
/// Where multimodal checkpoints nest their language model's settings:
/// `text_config` (gemma-4, Qwen-VL), `llm_config` (InternVL),
/// `language_config` (DeepSeek-VL). Defined here, not beside the MLX loader,
/// because the loader compiles only on Apple Silicon and this file everywhere.
pub(crate) const NESTED_TEXT_CONFIGS: [&str; 3] = ["text_config", "llm_config", "language_config"];
#[derive(Deserialize)]
struct ModelGeometryConfig {
/// The architecture (`text_config.model_type` for a nested language
/// model): it decides how an MLA cache is laid out.
model_type: Option<String>,
num_hidden_layers: Option<usize>,
num_key_value_heads: Option<usize>,
head_dim: Option<usize>,
hidden_size: Option<usize>,
num_attention_heads: Option<usize>,
kv_cache_dtype: Option<String>,
torch_dtype: Option<String>,
vocab_size: Option<usize>,
/// Per-layer attention kind (Qwen3.5+, gemma). A `linear_attention` layer
/// keeps a fixed-size state, not a per-token cache.
layer_types: Option<Vec<String>>,
/// Qwen3.5+'s hybrid period, when `layer_types` is absent: one
/// full-attention layer in every `full_attention_interval`.
full_attention_interval: Option<usize>,
linear_num_key_heads: Option<usize>,
linear_num_value_heads: Option<usize>,
linear_key_head_dim: Option<usize>,
linear_value_head_dim: Option<usize>,
linear_conv_kernel_dim: Option<usize>,
/// Multi-head latent attention (DeepSeek-V3, GLM-4.7-Flash): the cache
/// holds this compressed latent and the rope key, not per-head K and V.
kv_lora_rank: Option<usize>,
qk_rope_head_dim: Option<usize>,
qk_nope_head_dim: Option<usize>,
v_head_dim: Option<usize>,
}
fn dtype_bytes(dtype: Option<&str>) -> usize {
match dtype.map(str::to_ascii_lowercase).as_deref() {
Some("float32" | "fp32" | "f32") => 4,
Some("float64" | "fp64" | "f64") => 8,
Some("float8" | "fp8" | "int8" | "uint8" | "i8" | "u8") => 1,
// MLX-community quantizes weights, while its KV tensors remain fp16 or
// bf16. Missing and unrecognized dtype declarations therefore retain
// the safe two-byte default rather than inheriting weight bit width.
_ => 2,
}
}
/// What identifies one version of a checkpoint's `config.json`: its path, and
/// a remove, a replace or a re-download changes the rest.
#[derive(Clone, Debug, PartialEq, Eq)]
struct ConfigStamp {
/// Which file was read: the managed copy or a Hugging Face snapshot, and
/// a newer snapshot is a different file even at the same size and mtime.
path: PathBuf,
modified: Option<std::time::SystemTime>,
len: u64,
}
impl ConfigStamp {
fn of(path: &Path) -> Option<Self> {
let metadata = std::fs::metadata(path).ok()?;
Some(Self {
path: path.to_path_buf(),
modified: metadata.modified().ok(),
len: metadata.len(),
})
}
}
/// The `config.json` of the checkpoint a load would serve: the managed
/// directory when it holds one, else the Hugging Face snapshot that
/// `ensure_local` reuses for an MLX or managed vllm-mlx row instead of
/// copying it (`registry::try_reuse_local`). Without the second, a row served
/// from the shared cache was sized from its parameter count: Qwen3.8-27B at
/// 51 GiB for a 64k context it serves in 24.
fn kv_config_path(model: &ModelSchema, models_dir: &Path, hf_hub: Option<&Path>) -> PathBuf {
let managed = models_dir.join(&model.name).join("config.json");
let (Some(hub), false) = (hf_hub, managed.exists()) else {
return managed;
};
let repo = match &model.source {
crate::schema::ModelSource::Mlx { hf_repo, .. }
| crate::schema::ModelSource::ManagedVllmMlx { hf_repo, .. } => hf_repo,
_ => return managed,
};
crate::registry::latest_huggingface_repo_snapshot_in(&crate::hf_cache::repo_dir_in(hub, repo))
.map(|snapshot| snapshot.join("config.json"))
.filter(|config| config.exists())
.unwrap_or(managed)
}
/// Resolve KV geometry from `<models_dir>/<model.name>/config.json`.
///
/// The config may explicitly state `kv_cache_dtype` (or the model-wide
/// `torch_dtype`); otherwise KV elements default to two bytes for fp16/bf16.
/// Missing, unreadable, malformed, or incomplete configs have no geometry and
/// use the conservative parameter-count fallback below.
///
/// A multimodal checkpoint (gemma-4's `gemma4_unified`) keeps its language
/// model's geometry under a nested config, so when the top level has no layer
/// count, the nested one is read — the same keys the loader reads.
fn kv_geometry_for(
model: &ModelSchema,
models_dir: Option<&Path>,
hf_hub: Option<&Path>,
) -> Option<KvGeometry> {
let bytes = std::fs::read(kv_config_path(model, models_dir?, hf_hub)).ok()?;
let raw: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
let source = if raw.get("num_hidden_layers").is_some() {
&raw
} else {
NESTED_TEXT_CONFIGS
.iter()
.find_map(|key| {
raw.get(*key)
.filter(|v| v.get("num_hidden_layers").is_some())
})
.unwrap_or(&raw)
};
let mut config: ModelGeometryConfig = serde_json::from_value(source.clone()).ok()?;
// The KV dtype may be declared once, at the top.
if config.kv_cache_dtype.is_none() && config.torch_dtype.is_none() {
config.torch_dtype = raw
.get("torch_dtype")
.and_then(serde_json::Value::as_str)
.map(str::to_string);
}
geometry_from_config(&config)
}
/// mlx-lm architectures whose MLA cache holds the compressed latent and the
/// rope key (`cache.update_and_fetch(kv_latent, k_pe)`), checked in mlx-lm
/// 0.31.3. Another MLA architecture may cache expanded per-head keys and
/// values (minicpm3 does), so it is sized that way: the larger figure.
const LATENT_CACHE_MLA: [&str; 5] = [
"deepseek_v3",
"deepseek_v32",
"glm4_moe_lite",
"kimi_linear",
"longcat_flash",
];
/// Product of byte factors that saturates instead of wrapping: a malformed
/// `config.json` must read as enormous, never as a tiny estimate that admits
/// the load.
fn bytes(factors: &[u64]) -> u64 {
factors
.iter()
.fold(1_u64, |product, factor| product.saturating_mul(*factor))
}
fn geometry_from_config(config: &ModelGeometryConfig) -> Option<KvGeometry> {
let positive = |value: Option<usize>| value.filter(|value| *value > 0).map(|v| v as u64);
let num_hidden_layers = positive(config.num_hidden_layers)?;
let element = dtype_bytes(
config
.kv_cache_dtype
.as_deref()
.or(config.torch_dtype.as_deref()),
) as u64;
// Four fp16 `[tokens × hidden]` tensors conservatively alive together
// during prompt prefill, additive to the cache.
let activations = positive(config.hidden_size).map_or(0, |hidden| bytes(&[hidden, 2, 4]));
// A linear-attention layer keeps a fixed state, not a per-token cache,
// and is left out of the cache only when that state can be sized:
// without its dimensions it stays counted as a cache layer, which errs
// high. Sliding layers stay counted in full too: capping them at their
// window is not measured yet.
let linear_dims = (
positive(config.linear_num_key_heads),
positive(config.linear_num_value_heads),
positive(config.linear_key_head_dim),
positive(config.linear_value_head_dim),
);
let linear_layers = match (
&config.layer_types,
positive(config.full_attention_interval),
linear_dims,
) {
(_, _, (None, _, _, _) | (_, None, _, _) | (_, _, None, _) | (_, _, _, None)) => 0,
(Some(kinds), _, _) => kinds
.iter()
.filter(|kind| *kind == "linear_attention")
.count() as u64,
(None, Some(interval), _) => num_hidden_layers - num_hidden_layers / interval,
(None, None, _) => 0,
}
.min(num_hidden_layers);
let cache_layers = num_hidden_layers - linear_layers;
let fixed_state_bytes = match linear_dims {
(Some(key_heads), Some(value_heads), Some(key_dim), Some(value_dim)) => {
// Gated-delta state in fp32, plus the causal-conv tail over the
// q/k/v projections in the element dtype.
let recurrent = bytes(&[value_heads, key_dim, value_dim, 4]);
let conv_channels =
bytes(&[key_heads, key_dim, 2]).saturating_add(bytes(&[value_heads, value_dim]));
let kernel_tail = positive(config.linear_conv_kernel_dim)
.unwrap_or(4)
.saturating_sub(1);
let conv = bytes(&[conv_channels, kernel_tail, element]);
bytes(&[linear_layers, recurrent.saturating_add(conv)])
}
_ => 0,
};
let mla = (
positive(config.kv_lora_rank),
positive(config.qk_rope_head_dim),
positive(config.num_attention_heads),
);
let (kv_bytes_per_token, prefill_bytes_per_token) = match mla {
(Some(rank), Some(rope), Some(heads))
if config
.model_type
.as_deref()
.is_some_and(|kind| LATENT_CACHE_MLA.contains(&kind)) =>
{
// The cache holds the latent and the rope key once per layer.
// The prompt pass expands it per head (mlx-lm's glm4_moe_lite):
// rope scores, their masked copy, and the attention scores and
// probabilities of an unfused SDPA, each `heads × chunk` per
// context token, plus the expanded K and V.
let expanded = positive(config.qk_nope_head_dim)
.unwrap_or(0)
.saturating_add(positive(config.v_head_dim).unwrap_or(0));
let scratch = bytes(&[4, heads, PREFILL_CHUNK_TOKENS, element])
.saturating_add(bytes(&[heads, expanded, element]));
(
bytes(&[cache_layers, rank.saturating_add(rope), element]),
activations.saturating_add(scratch),
)
}
(Some(_), Some(rope), Some(heads)) => {
// MLA on an architecture not known to cache the latent: per-head
// keys (nope + rope) and values for every head.
let key = positive(config.qk_nope_head_dim)
.unwrap_or(0)
.saturating_add(rope);
let value = positive(config.v_head_dim).unwrap_or(key);
(
bytes(&[cache_layers, heads, key.saturating_add(value), element]),
activations,
)
}
_ => {
let kv_heads = positive(config.num_key_value_heads)?;
let head_dim = positive(config.head_dim).or_else(|| {
let hidden = positive(config.hidden_size)?;
let heads = positive(config.num_attention_heads)?;
hidden.is_multiple_of(heads).then_some(hidden / heads)
})?;
(
bytes(&[cache_layers, 2, kv_heads, head_dim, element]),
activations,
)
}
};
Some(KvGeometry {
kv_bytes_per_token,
fixed_state_bytes,
prefill_bytes_per_token,
vocab_size: positive(config.vocab_size),
})
}
/// Binary-megabyte cache for a context: per-token cache bytes times tokens,
/// plus the sequence's fixed state.
fn kv_cache_mb_for(geometry: &KvGeometry, context_tokens: usize) -> u64 {
let bytes = (geometry.kv_bytes_per_token as u128)
.saturating_mul(context_tokens as u128)
.saturating_add(geometry.fixed_state_bytes as u128);
bytes.div_ceil(BYTES_PER_MB).min(u64::MAX as u128) as u64
}
/// Conservative unknown-geometry fallback. Qwen3-8B fp16 KV uses 147,456
/// bytes/token (`2 × 36 layers × 8 KV heads × 128 head_dim × 2 bytes`). Dense
/// 0.6B–14B models sit within roughly 0.8–1.2× of this figure per token; larger
/// active parameter counts scale it upward. Installed, known geometry wins.
fn fallback_kv_cache_mb(model: &ModelSchema, context_tokens: usize) -> u64 {
let active_billions = model_parameter_billions_active(model) as f64;
let bytes_per_token = 147_456.0 * (active_billions / 8.0).max(1.0);
((bytes_per_token * context_tokens as f64) / BYTES_PER_MB as f64).ceil() as u64
}
/// Memory alive during the prompt pass beyond the cache, additive to it.
fn prefill_activation_mb(geometry: Option<&KvGeometry>, context_tokens: usize) -> u64 {
let Some(geometry) = geometry else {
return 0;
};
let bytes = (geometry.prefill_bytes_per_token as u128).saturating_mul(context_tokens as u128);
bytes.div_ceil(BYTES_PER_MB).min(u64::MAX as u128) as u64
}
/// The logits of one prompt chunk, for a model served through mlx-vlm (a
/// managed vllm-mlx row with a drafter runs `--mllm`): it computes them for
/// every chunk position, in fp32 and then bf16. Measured on Qwen3.8-27B as a
/// 2.85 GiB step at any context length; `248,320 × 2048 × 6` is 2.84 GiB.
/// mlx-lm, which serves every other managed row, computes them lazily and
/// never materializes them during prefill.
fn chunk_logits_mb(model: &ModelSchema, geometry: Option<&KvGeometry>) -> u64 {
let served_by_mlx_vlm = matches!(
&model.source,
crate::schema::ModelSource::ManagedVllmMlx {
draft_hf_repo: Some(_),
..
}
);
match geometry.and_then(|geometry| geometry.vocab_size) {
Some(vocab) if served_by_mlx_vlm => {
((vocab as u128) * (PREFILL_CHUNK_TOKENS as u128) * 6).div_ceil(BYTES_PER_MB) as u64
}
_ => 0,
}
}
/// Central recommendation-time memory estimator.
pub fn estimate_model_memory(
model: &ModelSchema,
hardware: &HardwareInfo,
context_tokens: usize,
) -> ModelMemoryEstimate {
estimate_model_memory_with_measured_weights(model, hardware, context_tokens, None)
}
/// The recommendation estimate for a row whose installed checkpoint may
/// declare its own KV geometry: the KV cache is sized from that geometry when
/// `geometry` holds the row, and from the parameter count otherwise.
pub fn estimate_model_memory_for_checkpoint(
model: &ModelSchema,
hardware: &HardwareInfo,
context_tokens: usize,
geometry: &CheckpointGeometry,
) -> ModelMemoryEstimate {
estimate_model_memory_with_geometry(model, hardware, context_tokens, None, geometry.get(model))
}
/// The same estimate with optional physical weight evidence. Registry/load
/// paths can provide a measured value without duplicating the remaining math.
pub fn estimate_model_memory_with_measured_weights(
model: &ModelSchema,
hardware: &HardwareInfo,
context_tokens: usize,
measured_weights_mb: Option<u64>,
) -> ModelMemoryEstimate {
estimate_model_memory_with_geometry(model, hardware, context_tokens, measured_weights_mb, None)
}
fn estimate_model_memory_with_geometry(
model: &ModelSchema,
hardware: &HardwareInfo,
context_tokens: usize,
measured_weights_mb: Option<u64>,
geometry: Option<&KvGeometry>,
) -> ModelMemoryEstimate {
let declared = model.cost.ram_mb.or(model.cost.size_mb);
let (weights_mb, evidence) = if let Some(measured) = measured_weights_mb {
(measured, ModelResourceEvidence::FileSystemMeasured)
} else if let Some(declared) = declared {
(
declared.max(model.cost.size_mb.unwrap_or(0)),
ModelResourceEvidence::CatalogExact,
)
} else {
(
heuristic_weights_mb(model),
ModelResourceEvidence::Heuristic,
)
};
let kv_cache_mb = geometry
.map(|geometry| kv_cache_mb_for(geometry, context_tokens))
.unwrap_or_else(|| fallback_kv_cache_mb(model, context_tokens));
let context_overhead_mb = kv_cache_mb
.saturating_add(prefill_activation_mb(geometry, context_tokens))
.saturating_add(chunk_logits_mb(model, geometry));
let runtime_overhead_mb = backend_runtime_overhead_mb(hardware);
let transient_margin_mb = TRANSIENT_ALLOCATION_MARGIN_MB;
let estimated_peak_mb = weights_mb
.saturating_add(context_overhead_mb)
.saturating_add(runtime_overhead_mb)
.saturating_add(transient_margin_mb);
ModelMemoryEstimate {
weights_mb,
runtime_overhead_mb,
context_overhead_mb,
transient_margin_mb,
estimated_peak_mb,
evidence,
}
}
fn backend_runtime_overhead_mb(hardware: &HardwareInfo) -> u64 {
match hardware.gpu_backend {
GpuBackend::Metal => METAL_RUNTIME_OVERHEAD_MB,
GpuBackend::Cuda => CUDA_RUNTIME_OVERHEAD_MB,
_ => CPU_RUNTIME_OVERHEAD_MB,
}
}
fn heuristic_weights_mb(model: &ModelSchema) -> u64 {
let billions = model_parameter_billions_total(model);
(billions as f64 * 600.0).ceil() as u64
}
pub(crate) fn model_parameter_billions_active(model: &ModelSchema) -> f32 {
model
.param_count
.split_once('(')
.and_then(|(_, rest)| rest.split_once("active"))
.and_then(|(number, _)| parse_parameter_billions(number))
.unwrap_or_else(|| model_parameter_billions_total(model))
}
pub(crate) fn model_parameter_billions_total(model: &ModelSchema) -> f32 {
parse_parameter_billions(&model.param_count).unwrap_or_else(|| {
let size_mb = model.size_mb();
if size_mb > 0 {
(size_mb as f32 / 600.0).max(0.1)
} else {
0.0
}
})
}
pub(crate) fn parse_parameter_billions(value: &str) -> Option<f32> {
let value = value.trim();
let number: String = value
.chars()
.take_while(|character| character.is_ascii_digit() || *character == '.')
.collect();
let parsed: f32 = number.parse().ok()?;
if value[number.len()..]
.trim_start()
.to_ascii_lowercase()
.starts_with('m')
{
Some(parsed / 1_000.0)
} else {
Some(parsed)
}
}
/// Evaluate policy math for the hardware CAR actually detected.
pub fn evaluate_resources(hardware: &HardwareInfo, policy: &ResourcePolicy) -> ResourceEvaluation {
let accelerator_memory = match (hardware.gpu_backend.clone(), hardware.gpu_memory_mb) {
(GpuBackend::Cuda, Some(total_mb)) => Some(AcceleratorResourceBudget {
total_mb,
budget_mb: total_mb,
}),
_ => None,
};
ResourceEvaluation {
host_memory: policy.effective_budget(hardware.total_ram_mb),
accelerator_memory,
}
}
fn percent_of(total_mb: u64, percent: u64) -> u64 {
let value = (total_mb as u128).saturating_mul(percent as u128) / 100;
value.min(u64::MAX as u128) as u64
}
fn minimum_emergency_reserve(total_memory_mb: u64) -> u64 {
MINIMUM_EMERGENCY_RESERVE_MB.max(percent_of(total_memory_mb, EMERGENCY_RESERVE_PERCENT))
}
/// Persistence errors that callers can distinguish from safe defaulting on a
/// missing or malformed policy document.
#[derive(Debug, Error)]
pub enum ResourcePolicyError {
#[error("resource policy I/O failed: {0}")]
Io(#[from] io::Error),
#[error("resource policy serialization failed: {0}")]
Serialization(#[from] serde_json::Error),
#[error("Custom model RAM must be finite, nonnegative, and a 0.5 GB increment; got {0}")]
InvalidCustomGigabytes(f64),
#[error("invalid resource policy: {reason}")]
InvalidPolicy { reason: String },
}
/// Where the effective policy returned by a repository load came from.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ResourcePolicyLoadSource {
Loaded,
MissingDefault,
CorruptDefault,
}
/// A policy plus recoverable evidence about safe defaulting.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct ResourcePolicyLoadEvidence {
pub policy: ResourcePolicy,
pub source: ResourcePolicyLoadSource,
pub warning: Option<String>,
}
pub trait ResourcePolicyRepository: Send + Sync {
fn load(&self) -> Result<ResourcePolicy, ResourcePolicyError>;
fn save(&self, policy: &ResourcePolicy) -> Result<(), ResourcePolicyError>;
}
/// Private, atomic resource-policy storage rooted at `CAR_HOME`.
#[derive(Clone, Debug)]
pub struct FileResourcePolicyRepository {
root: PathBuf,
}
impl Default for FileResourcePolicyRepository {
fn default() -> Self {
Self::new(car_home::root_or_relative())
}
}
/// File under the CAR state root recording that the saved resource policy
/// is the concierge's choice, not the user's.
pub const CONCIERGE_POLICY_MARKER: &str = "model-resource-policy.concierge.json";
#[derive(Debug, Serialize, Deserialize)]
struct ConciergePolicyMarker {
policy: ResourcePolicy,
reason: String,
at: u64,
/// The concierge undid its own raise: it does not raise again.
#[serde(default)]
reverted: bool,
}
/// What the concierge's marker records about the saved policy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConciergeChoice {
pub policy: ResourcePolicy,
pub at: u64,
pub reverted: bool,
}
impl FileResourcePolicyRepository {
pub fn new(root: PathBuf) -> Self {
Self { root }
}
/// The policy the concierge last saved, if the saved policy was its
/// choice. A user's save clears this ([`Self::clear_concierge_marker`]),
/// so the saved policy differing from it, or it being absent while a
/// policy is saved, means the user chose.
pub fn concierge_marker(&self) -> Option<ResourcePolicy> {
self.concierge_choice().map(|choice| choice.policy)
}
/// The full marker: the policy, when it was set, and whether it was a
/// reversal of the concierge's own raise.
pub fn concierge_choice(&self) -> Option<ConciergeChoice> {
let bytes = std::fs::read(self.root.join(CONCIERGE_POLICY_MARKER)).ok()?;
serde_json::from_slice::<ConciergePolicyMarker>(&bytes)
.ok()
.map(|marker| ConciergeChoice {
policy: marker.policy,
at: marker.at,
reverted: marker.reverted,
})
}
/// Save `policy` as the concierge's choice: the policy, then the marker.
pub fn save_concierge_choice(
&self,
policy: &ResourcePolicy,
reason: &str,
at: u64,
reverted: bool,
) -> Result<(), ResourcePolicyError> {
ResourcePolicyRepository::save(self, policy)?;
let marker = ConciergePolicyMarker {
policy: policy.clone(),
reason: reason.to_string(),
at,
reverted,
};
let path = self.root.join(CONCIERGE_POLICY_MARKER);
let tmp = self.root.join(format!(".{CONCIERGE_POLICY_MARKER}.tmp"));
let bytes = serde_json::to_vec_pretty(&marker).map_err(io::Error::other)?;
std::fs::write(&tmp, bytes)?;
std::fs::rename(&tmp, &path)?;
Ok(())
}
/// The user chose: the saved policy is theirs from now on.
pub fn clear_concierge_marker(&self) -> Result<(), ResourcePolicyError> {
match std::fs::remove_file(self.root.join(CONCIERGE_POLICY_MARKER)) {
Err(error) if error.kind() != io::ErrorKind::NotFound => Err(error.into()),
_ => Ok(()),
}
}
pub fn path(&self) -> PathBuf {
self.root.join(RESOURCE_POLICY_FILE)
}
/// Load the effective policy without losing evidence that CAR recovered
/// from a missing or invalid source document.
pub fn load_with_evidence(&self) -> Result<ResourcePolicyLoadEvidence, ResourcePolicyError> {
let path = self.path();
let file = match open_resource_policy(&path) {
Ok(file) => file,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
return Ok(ResourcePolicyLoadEvidence {
policy: ResourcePolicy::everyday(),
source: ResourcePolicyLoadSource::MissingDefault,
warning: None,
});
}
Err(error) => {
if std::fs::symlink_metadata(&path)
.is_ok_and(|metadata| !metadata.is_file() || metadata.file_type().is_symlink())
{
return Ok(corrupt_default(
"The saved resource policy could not be loaded because it is not a regular file.",
));
}
return Err(error.into());
}
};
let metadata = file.metadata()?;
if !metadata.is_file() {
return Ok(corrupt_default(
"The saved resource policy could not be loaded because it is not a regular file.",
));
}
if metadata.len() > MAX_POLICY_BYTES {
return Ok(corrupt_default(
"The saved resource policy could not be loaded because it exceeds the size limit.",
));
}
let mut raw = Vec::new();
file.take(MAX_POLICY_BYTES + 1).read_to_end(&mut raw)?;
if raw.len() as u64 > MAX_POLICY_BYTES {
return Ok(corrupt_default(
"The saved resource policy could not be loaded because it exceeds the size limit.",
));
}
let policy = match serde_json::from_slice::<ResourcePolicy>(&raw) {
Ok(policy) => policy,
Err(error) => {
return Ok(corrupt_default(format!(
"The saved resource policy could not be loaded: {error}"
)));
}
};
if let Err(error) = policy.validate() {
return Ok(corrupt_default(format!(
"The saved resource policy could not be loaded: {error}"
)));
}
Ok(ResourcePolicyLoadEvidence {
policy,
source: ResourcePolicyLoadSource::Loaded,
warning: None,
})
}
}
fn open_resource_policy(path: &Path) -> io::Result<std::fs::File> {
let mut options = std::fs::OpenOptions::new();
options.read(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.custom_flags(libc::O_NOFOLLOW);
}
options.open(path)
}
fn corrupt_default(warning: impl Into<String>) -> ResourcePolicyLoadEvidence {
ResourcePolicyLoadEvidence {
policy: ResourcePolicy::everyday(),
source: ResourcePolicyLoadSource::CorruptDefault,
warning: Some(warning.into()),
}
}
impl ResourcePolicyRepository for FileResourcePolicyRepository {
fn load(&self) -> Result<ResourcePolicy, ResourcePolicyError> {
Ok(self.load_with_evidence()?.policy)
}
fn save(&self, policy: &ResourcePolicy) -> Result<(), ResourcePolicyError> {
policy.validate()?;
let _guard = mutation_lock()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
ensure_private_directory(&self.root)?;
let temp_path = unique_temp_path(&self.root);
let result = (|| {
let body = serde_json::to_vec_pretty(policy)?;
let mut file = open_private_temp(&temp_path)?;
file.write_all(&body)?;
file.sync_all()?;
atomic_replace(&temp_path, &self.path())?;
car_secrets::harden_owner_only(&self.path());
sync_directory(&self.root)?;
Ok(())
})();
if result.is_err() {
let _ = std::fs::remove_file(&temp_path);
}
result
}
}
fn mutation_lock() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}
fn unique_temp_path(root: &Path) -> PathBuf {
static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(0);
let sequence = NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
let epoch_nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_nanos())
.unwrap_or(0);
root.join(format!(
".{RESOURCE_POLICY_FILE}.{}.{}.{}.tmp",
std::process::id(),
epoch_nanos,
sequence
))
}
fn ensure_private_directory(path: &Path) -> io::Result<()> {
std::fs::create_dir_all(path)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
}
car_secrets::harden_owner_only(path);
Ok(())
}
#[cfg(unix)]
fn open_private_temp(path: &Path) -> io::Result<std::fs::File> {
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
let file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(path)?;
file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
Ok(file)
}
#[cfg(not(unix))]
fn open_private_temp(path: &Path) -> io::Result<std::fs::File> {
let file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(path)?;
car_secrets::harden_owner_only(path);
Ok(file)
}
#[cfg(not(windows))]
fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> {
std::fs::rename(source, destination)
}
#[cfg(windows)]
fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> {
use std::os::windows::ffi::OsStrExt;
const MOVEFILE_REPLACE_EXISTING: u32 = 0x1;
const MOVEFILE_WRITE_THROUGH: u32 = 0x8;
#[link(name = "kernel32")]
unsafe extern "system" {
fn MoveFileExW(existing: *const u16, new: *const u16, flags: u32) -> i32;
}
let source = source
.as_os_str()
.encode_wide()
.chain(Some(0))
.collect::<Vec<_>>();
let destination = destination
.as_os_str()
.encode_wide()
.chain(Some(0))
.collect::<Vec<_>>();
// SAFETY: both pointers address NUL-terminated UTF-16 buffers for the
// duration of the call, and the flags request atomic replacement.
let replaced = unsafe {
MoveFileExW(
source.as_ptr(),
destination.as_ptr(),
MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
)
};
if replaced == 0 {
Err(io::Error::last_os_error())
} else {
Ok(())
}
}
#[cfg(unix)]
fn sync_directory(path: &Path) -> io::Result<()> {
std::fs::File::open(path)?.sync_all()
}
#[cfg(not(unix))]
fn sync_directory(_path: &Path) -> io::Result<()> {
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn hardware(
total_ram_mb: u64,
gpu_backend: crate::hardware::GpuBackend,
vram_mb: Option<u64>,
) -> crate::hardware::HardwareInfo {
crate::hardware::HardwareInfo {
os: "test".into(),
arch: "test".into(),
cpu_cores: 8,
total_ram_mb,
gpu_backend,
gpu_memory_mb: vram_mb,
gpu_devices: Vec::new(),
recommended_model: "fixture".into(),
recommended_context: 4_096,
max_model_mb: total_ram_mb,
}
}
#[test]
fn profiles_compute_exact_32_gb_budgets() {
let total = 32 * 1024;
assert_eq!(
ResourcePolicy::everyday()
.effective_budget(total)
.configured_model_ceiling_mb,
13_107
);
assert_eq!(
ResourcePolicy::local_focused()
.effective_budget(total)
.configured_model_ceiling_mb,
26_214
);
assert_eq!(
ResourcePolicy::custom_gb(12.5)
.unwrap()
.effective_budget(total)
.configured_model_ceiling_mb,
12_800
);
assert_eq!(
ResourcePolicy::everyday().recommendation_target_mb(total),
13_107
);
}
#[test]
fn model_memory_estimate_keeps_transient_margin_distinct_and_totals_exactly() {
let catalog = crate::registry::builtin_catalog();
let model = catalog
.iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let estimate = estimate_model_memory(
model,
&hardware(32 * 1024, GpuBackend::Metal, None),
RECOMMENDATION_CONTEXT_TOKENS,
);
assert_eq!(estimate.evidence, ModelResourceEvidence::CatalogExact);
assert_eq!(estimate.weights_mb, 2_400);
assert_eq!(estimate.runtime_overhead_mb, 512);
assert_eq!(estimate.transient_margin_mb, 1_024);
assert_eq!(
estimate.estimated_peak_mb,
estimate.weights_mb
+ estimate.context_overhead_mb
+ estimate.runtime_overhead_mb
+ estimate.transient_margin_mb
);
let measured = estimate_model_memory_with_measured_weights(
model,
&hardware(32 * 1024, GpuBackend::Metal, None),
RECOMMENDATION_CONTEXT_TOKENS,
Some(2_321),
);
assert_eq!(measured.weights_mb, 2_321);
assert_eq!(measured.evidence, ModelResourceEvidence::FileSystemMeasured);
}
fn qwen3_8b_model() -> ModelSchema {
crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-8b:4bit")
.unwrap()
}
fn qwen3_8b_geometry() -> KvGeometry {
KvGeometry {
// 2 × 36 layers × 8 KV heads × 128 head_dim × 2 bytes.
kv_bytes_per_token: 147_456,
fixed_state_bytes: 0,
prefill_bytes_per_token: 4_096 * 2 * 4,
vocab_size: None,
}
}
#[test]
fn kv_cache_qwen3_8b_geometry_matches_binary_mb_table() {
let geometry = qwen3_8b_geometry();
assert_eq!(kv_cache_mb_for(&geometry, 9_000), 1_266);
assert_eq!(kv_cache_mb_for(&geometry, 13_096), 1_842);
assert_eq!(prefill_activation_mb(Some(&geometry), 9_000), 282);
}
#[test]
fn kv_cache_unknown_geometry_uses_conservative_qwen3_8b_floor() {
let model = qwen3_8b_model();
let estimate =
estimate_model_memory(&model, &hardware(32 * 1024, GpuBackend::Metal, None), 9_000);
assert_eq!(estimate.context_overhead_mb, 1_266);
}
#[test]
fn kv_cache_unknown_geometry_scales_models_larger_than_8b() {
let mut model = qwen3_8b_model();
model.param_count = "32B".into();
let estimate =
estimate_model_memory(&model, &hardware(64 * 1024, GpuBackend::Metal, None), 9_000);
assert_eq!(estimate.context_overhead_mb, 5_063);
}
#[test]
fn kv_geometry_installed_config_drives_admission_context_charge() {
let root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
let model_dir = root.path().join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
std::fs::write(
model_dir.join("config.json"),
r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32,"torch_dtype":"bfloat16"}"#,
)
.unwrap();
let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(root.path().to_path_buf()),
);
let estimate = coordinator.preflight(&model, 9_000).estimate;
assert_eq!(estimate.context_overhead_mb, 1_548);
}
/// The recommendation estimate takes the geometry admission reads, from
/// the same file and through the same cache, so for an installed row the
/// two publish one peak and only live memory separates their verdicts
/// (car#2213).
#[test]
fn recommendation_estimate_uses_the_installed_geometry_admission_uses() {
let root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
let model_dir = root.path().join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
std::fs::write(
model_dir.join("config.json"),
r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32,"torch_dtype":"bfloat16"}"#,
)
.unwrap();
let hardware = hardware(32 * 1024, GpuBackend::Metal, None);
let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware.clone(),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(root.path().to_path_buf()),
);
let admitted = coordinator
.preflight(&model, RECOMMENDATION_CONTEXT_TOKENS)
.estimate
.estimated_peak_mb;
let peak = |geometry: &CheckpointGeometry| {
estimate_model_memory_for_checkpoint(
&model,
&hardware,
RECOMMENDATION_CONTEXT_TOKENS,
geometry,
)
.estimated_peak_mb
};
assert_eq!(
peak(&CheckpointGeometry::read([&model], root.path())),
admitted
);
assert_eq!(peak(&coordinator.checkpoint_geometry([&model])), admitted);
// Not installed: the parameter-count fallback, which differs here.
assert_ne!(peak(&CheckpointGeometry::default()), admitted);
assert_eq!(
peak(&CheckpointGeometry::default()),
estimate_model_memory(&model, &hardware, RECOMMENDATION_CONTEXT_TOKENS)
.estimated_peak_mb
);
}
#[test]
fn kv_geometry_reads_a_multimodal_checkpoints_nested_text_config() {
let root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
let model_dir = root.path().join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
// Same geometry as the flat test above, nested the way gemma-4 is.
std::fs::write(
model_dir.join("config.json"),
r#"{"model_type":"gemma4_unified","torch_dtype":"bfloat16","vision_config":{},"text_config":{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32}}"#,
)
.unwrap();
let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(root.path().to_path_buf()),
);
assert_eq!(
coordinator
.preflight(&model, 9_000)
.estimate
.context_overhead_mb,
1_548
);
}
#[test]
fn kv_geometry_cache_forgets_missing_config_after_install() {
let root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(root.path().to_path_buf()),
);
let before_install = coordinator.preflight(&model, 9_000).estimate;
let model_dir = root.path().join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
std::fs::write(
model_dir.join("config.json"),
r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096,"num_attention_heads":32,"torch_dtype":"bfloat16"}"#,
)
.unwrap();
let after_install = coordinator.preflight(&model, 9_000).estimate;
assert_eq!(before_install.context_overhead_mb, 1_266);
assert_eq!(after_install.context_overhead_mb, 1_548);
}
#[test]
fn kv_geometry_derives_head_dim_and_memoizes_installed_config() {
let root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
let model_dir = root.path().join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
let config_path = model_dir.join("config.json");
std::fs::write(
&config_path,
r#"{"num_hidden_layers":36,"num_key_value_heads":8,"hidden_size":4096,"num_attention_heads":32}"#,
)
.unwrap();
let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(root.path().to_path_buf()),
);
let first = coordinator.preflight(&model, 9_000).estimate;
// Same length and mtime: the stamp is unchanged, so the cached
// geometry stands and the file is not parsed again.
let modified = std::fs::metadata(&config_path).unwrap().modified().unwrap();
let len = std::fs::metadata(&config_path).unwrap().len() as usize;
std::fs::write(&config_path, "x".repeat(len)).unwrap();
std::fs::File::options()
.write(true)
.open(&config_path)
.unwrap()
.set_modified(modified)
.unwrap();
let second = coordinator.preflight(&model, 9_000).estimate;
assert_eq!(first.context_overhead_mb, 1_548);
assert_eq!(second.context_overhead_mb, first.context_overhead_mb);
}
#[test]
fn kv_geometry_cache_is_scoped_to_coordinator_models_dir() {
let first_root = tempfile::tempdir().unwrap();
let second_root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
for (root, config) in [
(
first_root.path(),
r#"{"num_hidden_layers":36,"num_key_value_heads":8,"head_dim":128,"hidden_size":4096}"#,
),
(
second_root.path(),
r#"{"num_hidden_layers":2,"num_key_value_heads":2,"head_dim":64,"hidden_size":256}"#,
),
] {
let model_dir = root.join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
std::fs::write(model_dir.join("config.json"), config).unwrap();
}
let coordinator = |models_dir: &Path| {
LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(models_dir.to_path_buf()),
)
};
// Both coordinators look up the same model id. Their distinct results
// prove geometry memoization belongs to the coordinator/models root,
// rather than leaking one test or engine's config across the process.
assert_eq!(
coordinator(first_root.path())
.preflight(&model, 9_000)
.estimate
.context_overhead_mb,
1_548
);
assert_eq!(
coordinator(second_root.path())
.preflight(&model, 9_000)
.estimate
.context_overhead_mb,
27
);
}
#[test]
fn kv_geometry_malformed_config_falls_back_without_panicking() {
let root = tempfile::tempdir().unwrap();
let model = qwen3_8b_model();
let model_dir = root.path().join(&model.name);
std::fs::create_dir_all(&model_dir).unwrap();
std::fs::write(model_dir.join("config.json"), b"not json").unwrap();
let coordinator = LocalAdmissionCoordinator::with_probe_and_models_dir(
ResourcePolicy::local_focused(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
Some(root.path().to_path_buf()),
);
let estimate = coordinator.preflight(&model, 9_000).estimate;
assert_eq!(estimate.context_overhead_mb, 1_266);
}
#[test]
fn kv_cache_estimate_is_monotone_and_keeps_non_context_terms_fixed() {
let model = qwen3_8b_model();
let hardware = hardware(32 * 1024, GpuBackend::Metal, None);
let short = estimate_model_memory(&model, &hardware, 9_000);
let long = estimate_model_memory(&model, &hardware, 13_096);
assert!(long.context_overhead_mb >= short.context_overhead_mb);
assert_eq!(short.weights_mb, long.weights_mb);
assert_eq!(short.runtime_overhead_mb, long.runtime_overhead_mb);
assert_eq!(short.transient_margin_mb, long.transient_margin_mb);
}
#[test]
fn custom_zero_disables_loads_and_overlarge_values_clamp_below_emergency_reserve() {
let total = 32 * 1024;
assert_eq!(
ResourcePolicy::custom_gb(0.0)
.unwrap()
.effective_budget(total)
.effective_new_load_ceiling_mb,
0
);
let result = ResourcePolicy::custom_gb(99.0)
.unwrap()
.effective_budget(total);
assert_eq!(result.emergency_reserve_mb, 3_276);
assert_eq!(result.configured_model_ceiling_mb, total - 3_276);
assert!(result.normalization_notice.is_some());
}
#[test]
fn custom_gigabytes_reject_non_half_steps_negative_and_non_finite_values() {
for invalid in [10.3, -0.5, f64::INFINITY, f64::NEG_INFINITY, f64::NAN] {
assert!(
ResourcePolicy::custom_gb(invalid).is_err(),
"accepted {invalid:?}"
);
}
assert_eq!(
ResourcePolicy::custom_gb(10.5).unwrap().custom_max_model_mb,
Some(10_752)
);
}
#[test]
fn repository_round_trips_exact_half_gb_and_uses_private_atomic_files() {
let dir = tempfile::tempdir().unwrap();
let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
repository
.save(&ResourcePolicy::custom_gb(10.5).unwrap())
.unwrap();
assert_eq!(
repository.path(),
dir.path().join("model-resource-policy.json")
);
assert_eq!(repository.load().unwrap().custom_max_model_mb, Some(10_752));
assert_private_mode(&repository.path(), 0o600);
assert_private_mode(dir.path(), 0o700);
assert_no_temp_files(dir.path());
}
#[test]
fn repository_missing_or_corrupt_file_falls_back_without_deleting_source() {
let dir = tempfile::tempdir().unwrap();
let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
assert_eq!(repository.load().unwrap(), ResourcePolicy::everyday());
let corrupt = br#"{"profile":"custom","custom_max_model_mb":"broken"}"#;
std::fs::write(repository.path(), corrupt).unwrap();
assert_eq!(repository.load().unwrap(), ResourcePolicy::everyday());
assert_eq!(std::fs::read(repository.path()).unwrap(), corrupt);
}
#[test]
fn repository_rejects_invalid_policy_shapes_before_writing() {
let dir = tempfile::tempdir().unwrap();
let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
let invalid = [
ResourcePolicy {
profile: ResourceProfile::Custom,
custom_max_model_mb: None,
},
ResourcePolicy {
profile: ResourceProfile::Custom,
custom_max_model_mb: Some(1),
},
ResourcePolicy {
profile: ResourceProfile::Custom,
custom_max_model_mb: Some(513),
},
ResourcePolicy {
profile: ResourceProfile::Everyday,
custom_max_model_mb: Some(512),
},
ResourcePolicy {
profile: ResourceProfile::LocalFocused,
custom_max_model_mb: Some(512),
},
];
for policy in invalid {
let error = repository.save(&policy).unwrap_err();
assert!(matches!(error, ResourcePolicyError::InvalidPolicy { .. }));
assert!(!repository.path().exists());
}
}
#[test]
fn load_evidence_distinguishes_loaded_missing_and_corrupt_defaults() {
let dir = tempfile::tempdir().unwrap();
let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
let missing = repository.load_with_evidence().unwrap();
assert_eq!(missing.policy, ResourcePolicy::everyday());
assert_eq!(missing.source, ResourcePolicyLoadSource::MissingDefault);
assert!(missing.warning.is_none());
repository.save(&ResourcePolicy::local_focused()).unwrap();
let loaded = repository.load_with_evidence().unwrap();
assert_eq!(loaded.policy, ResourcePolicy::local_focused());
assert_eq!(loaded.source, ResourcePolicyLoadSource::Loaded);
assert!(loaded.warning.is_none());
let corrupt = br#"{"profile":"custom","custom_max_model_mb":"broken"}"#;
std::fs::write(repository.path(), corrupt).unwrap();
let recovered = repository.load_with_evidence().unwrap();
assert_eq!(recovered.policy, ResourcePolicy::everyday());
assert_eq!(recovered.source, ResourcePolicyLoadSource::CorruptDefault);
assert!(recovered
.warning
.as_deref()
.is_some_and(|warning| { warning.contains("could not be loaded") }));
assert_eq!(std::fs::read(repository.path()).unwrap(), corrupt);
}
#[test]
fn corrupt_evidence_covers_unknown_fields_invalid_shapes_and_oversized_files() {
let dir = tempfile::tempdir().unwrap();
let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
let corrupt_documents = [
br#"{"#.to_vec(),
br#"{"profile":"everyday","custom_max_model_mb":null,"extra":true}"#.to_vec(),
br#"{"profile":"custom","custom_max_model_mb":null}"#.to_vec(),
br#"{"profile":"custom","custom_max_model_mb":1}"#.to_vec(),
br#"{"profile":"custom","custom_max_model_mb":513}"#.to_vec(),
br#"{"profile":"everyday","custom_max_model_mb":512}"#.to_vec(),
br#"{"profile":"local_focused","custom_max_model_mb":512}"#.to_vec(),
vec![b' '; MAX_POLICY_BYTES as usize + 1],
];
for document in corrupt_documents {
std::fs::write(repository.path(), &document).unwrap();
let recovered = repository.load_with_evidence().unwrap();
assert_eq!(recovered.source, ResourcePolicyLoadSource::CorruptDefault);
assert!(recovered.warning.is_some());
assert_eq!(std::fs::read(repository.path()).unwrap(), document);
}
}
#[cfg(unix)]
#[test]
fn non_regular_policy_source_is_reported_as_corrupt_without_following_it() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let repository = FileResourcePolicyRepository::new(dir.path().to_path_buf());
let target = dir.path().join("target.json");
std::fs::write(
&target,
br#"{"profile":"local_focused","custom_max_model_mb":null}"#,
)
.unwrap();
symlink(&target, repository.path()).unwrap();
let recovered = repository.load_with_evidence().unwrap();
assert_eq!(recovered.source, ResourcePolicyLoadSource::CorruptDefault);
assert!(recovered.warning.is_some());
assert!(repository.path().is_symlink());
}
#[test]
fn concurrent_saves_leave_one_complete_document_and_no_staging_files() {
let dir = tempfile::tempdir().unwrap();
let repository =
std::sync::Arc::new(FileResourcePolicyRepository::new(dir.path().to_path_buf()));
let mut writers = Vec::new();
for index in 0..16_u64 {
let repository = repository.clone();
writers.push(std::thread::spawn(move || {
repository
.save(&ResourcePolicy {
profile: ResourceProfile::Custom,
custom_max_model_mb: Some(index * 512),
})
.unwrap();
}));
}
for writer in writers {
writer.join().unwrap();
}
let loaded = repository.load().unwrap();
assert_eq!(loaded.profile, ResourceProfile::Custom);
assert!(loaded.custom_max_model_mb.unwrap().is_multiple_of(512));
assert_no_temp_files(dir.path());
}
#[test]
fn cuda_uses_separate_vram_fit_and_host_ram_policy() {
let hardware = hardware(
64 * 1024,
crate::hardware::GpuBackend::Cuda,
Some(12 * 1024),
);
let evidence = evaluate_resources(&hardware, &ResourcePolicy::everyday());
assert_eq!(evidence.host_memory.configured_model_ceiling_mb, 26_214);
let accelerator = evidence.accelerator_memory.unwrap();
assert_eq!(accelerator.total_mb, 12 * 1024);
assert_eq!(accelerator.budget_mb, 12 * 1024);
assert_ne!(
accelerator.budget_mb,
evidence.host_memory.configured_model_ceiling_mb
);
}
#[test]
fn resource_preflight_zero_custom_budget_blocks_new_load() {
let coordinator = LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(0.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
);
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let preflight = coordinator.preflight(&model, 2_048);
assert_eq!(preflight.verdict, LocalLoadVerdict::DisabledByPolicy);
}
#[test]
fn resource_preflight_zero_budget_does_not_kill_resident_inference() {
let coordinator = LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(0.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
);
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
coordinator.mark_resident(&model.id, 2_400);
let preflight = coordinator.preflight(&model, 2_048);
assert_eq!(preflight.verdict, LocalLoadVerdict::Allowed);
assert_eq!(
preflight.estimated_incremental_mb,
preflight.estimate.context_overhead_mb
+ preflight.estimate.runtime_overhead_mb
+ preflight.estimate.transient_margin_mb
);
}
#[test]
fn resource_preflight_resident_weights_are_incremental_only() {
let coordinator = LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(6_000)),
);
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
coordinator.mark_resident(&model.id, 2_400);
let preflight = coordinator.preflight(&model, 2_048);
assert_eq!(
preflight.estimated_incremental_mb,
preflight.estimate.context_overhead_mb
+ preflight.estimate.runtime_overhead_mb
+ preflight.estimate.transient_margin_mb
);
assert_eq!(preflight.verdict, LocalLoadVerdict::Allowed);
}
#[test]
fn resource_preflight_unavailable_live_probe_is_explicit() {
let coordinator = LocalAdmissionCoordinator::with_probe(
ResourcePolicy::everyday(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::unknown()),
);
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let preflight = coordinator.preflight(&model, 2_048);
assert_eq!(preflight.live_available_mb, None);
assert_eq!(preflight.verdict, LocalLoadVerdict::LiveMemoryUnknown);
}
#[test]
fn resource_preflight_simultaneous_reservations_are_atomic() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(5.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(6_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
coordinator.mark_resident(&model.id, 2_400);
let first = coordinator.reserve(&model, 2_048).unwrap();
let second = coordinator.reserve(&model, 2_048).unwrap_err();
assert_eq!(
second.preflight.verdict,
LocalLoadVerdict::ExceedsConfiguredCeiling
);
drop(first);
assert!(coordinator.reserve(&model, 2_048).is_ok());
}
#[test]
fn distinct_state_roots_reserve_atomically_against_one_machine_ledger() {
let machine_ledger = Arc::new(Mutex::new(MachineAdmissionLedger::default()));
let policy = ResourcePolicy::custom_gb(6.0).unwrap();
let hardware = hardware(32 * 1024, GpuBackend::Metal, None);
let first = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
policy.clone(),
hardware.clone(),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
machine_ledger.clone(),
));
let second = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
policy,
hardware,
Arc::new(FixedLiveMemoryProbe::known(24_000)),
machine_ledger,
));
let mut model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
model.cost.ram_mb = Some(2 * 1024);
model.cost.size_mb = Some(2 * 1024);
let barrier = Arc::new(std::sync::Barrier::new(2));
let attempts = [first, second].map(|coordinator| {
let model = model.clone();
let barrier = barrier.clone();
std::thread::spawn(move || {
barrier.wait();
coordinator.reserve(&model, 0)
})
});
let outcomes = attempts.map(|attempt| attempt.join().unwrap());
assert_eq!(outcomes.iter().filter(|result| result.is_ok()).count(), 1);
assert_eq!(
outcomes
.iter()
.find_map(|result| result.as_ref().err())
.expect("one cross-root request must be blocked")
.preflight
.verdict,
LocalLoadVerdict::ExceedsConfiguredCeiling
);
}
#[test]
fn resource_preflight_cuda_charges_weights_to_vram_and_overhead_to_host() {
let coordinator = LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(2.0).unwrap(),
hardware(64 * 1024, GpuBackend::Cuda, Some(12 * 1024)),
Arc::new(FixedLiveMemoryProbe::known(20_000)),
);
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let preflight = coordinator.preflight(&model, 2_048);
assert_eq!(preflight.resident_model_mb, 0);
assert_eq!(preflight.accelerator_total_mb, Some(12 * 1024));
assert_eq!(
preflight.accelerator_incremental_mb,
Some(preflight.estimate.weights_mb)
);
assert_eq!(
preflight.estimated_incremental_mb,
preflight.estimate.context_overhead_mb
+ preflight.estimate.runtime_overhead_mb
+ preflight.estimate.transient_margin_mb
);
assert_eq!(preflight.verdict, LocalLoadVerdict::Allowed);
}
#[test]
fn resource_preflight_cuda_vram_reservations_are_atomic() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(64 * 1024, GpuBackend::Cuda, Some(3_000)),
Arc::new(FixedLiveMemoryProbe::known(20_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let first = coordinator.reserve(&model, 2_048).unwrap();
assert_eq!(
coordinator
.reserve(&model, 2_048)
.unwrap_err()
.preflight
.verdict,
LocalLoadVerdict::ExceedsConfiguredCeiling
);
drop(first);
assert!(coordinator.reserve(&model, 2_048).is_ok());
}
#[test]
fn resource_preflight_model_maintenance_races_atomically_with_reserve() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::everyday(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let active = coordinator.reserve(&model, 2_048).unwrap();
assert!(matches!(
coordinator.begin_model_maintenance(&model.id),
Err(ModelMaintenanceError::ModelInUse(_))
));
drop(active);
let maintenance = coordinator.begin_model_maintenance(&model.id).unwrap();
assert_eq!(
coordinator
.reserve(&model, 2_048)
.unwrap_err()
.preflight
.verdict,
LocalLoadVerdict::ModelMaintenance
);
drop(maintenance);
assert!(coordinator.reserve(&model, 2_048).is_ok());
}
#[test]
fn measured_weights_are_rechecked_atomically_before_allocation() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(5.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(16_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let mut reservation = coordinator.reserve(&model, 2_048).unwrap();
let blocked = reservation
.reconcile_measured_weights(6 * 1024 * 1024 * 1024)
.unwrap_err();
assert_eq!(
blocked.preflight.verdict,
LocalLoadVerdict::ExceedsConfiguredCeiling
);
assert_eq!(
coordinator.preflight(&model, 2_048).active_reservations_mb,
reservation.reserved_incremental_mb(),
"a rejected resize must not mutate the live reservation"
);
}
#[test]
fn cache_publication_transfers_cold_reservation_without_double_counting() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(16_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let mut reservation = coordinator.reserve(&model, 2_048).unwrap();
let measured_bytes = 3 * 1024 * 1024 * 1024_u64;
reservation
.reconcile_measured_weights(measured_bytes)
.unwrap();
reservation.publish_resident_weights(measured_bytes);
let after = coordinator.preflight(&model, 2_048);
let request_overhead = after.estimate.context_overhead_mb
+ after.estimate.runtime_overhead_mb
+ after.estimate.transient_margin_mb;
assert_eq!(after.resident_model_mb, 3 * 1024);
assert_eq!(after.active_reservations_mb, request_overhead);
assert_eq!(reservation.reserved_incremental_mb(), request_overhead);
}
#[test]
fn resident_publication_never_reprobes_or_rejects_after_allocation() {
struct CountingProbe {
calls: AtomicU64,
available_mb: u64,
}
impl LiveMemoryProbe for CountingProbe {
fn available_memory_mb(&self) -> Result<Option<u64>, ResourcePolicyError> {
self.calls.fetch_add(1, Ordering::Relaxed);
Ok(Some(self.available_mb))
}
}
let probe = Arc::new(CountingProbe {
calls: AtomicU64::new(0),
available_mb: 24_000,
});
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
probe.clone(),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let mut reservation = coordinator.reserve(&model, 2_048).unwrap();
let measured = 3 * 1024 * 1024 * 1024_u64;
reservation.reconcile_measured_weights(measured).unwrap();
let calls_before_publish = probe.calls.load(Ordering::Relaxed);
reservation.publish_resident_weights(measured + 512 * 1024 * 1024);
assert_eq!(probe.calls.load(Ordering::Relaxed), calls_before_publish);
assert!(coordinator.is_resident(&model.id));
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.resident_models
.get(&model.id)
.map(|allocation| allocation.weights_mb),
Some(3 * 1024 + 512)
);
}
#[test]
fn simultaneous_cold_reservations_converge_on_one_resident_allocation() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(16.0).unwrap(),
hardware(32 * 1024, GpuBackend::Metal, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let mut first = coordinator.reserve(&model, 2_048).unwrap();
let mut second = coordinator.reserve(&model, 2_048).unwrap();
let measured = 3 * 1024 * 1024 * 1024_u64;
first.reconcile_measured_weights(measured).unwrap();
second.reconcile_measured_weights(measured).unwrap();
first.publish_resident_weights(measured);
second.publish_resident_weights(measured);
let after = coordinator.preflight(&model, 2_048);
let per_request_overhead = after.estimate.context_overhead_mb
+ after.estimate.runtime_overhead_mb
+ after.estimate.transient_margin_mb;
assert_eq!(after.resident_model_mb, 3 * 1024);
assert_eq!(
after.active_reservations_mb,
per_request_overhead * 2,
"peer publication must release only this reservation's redundant cold weights"
);
}
#[test]
fn resident_eviction_before_allocation_repromotes_reservation_to_cold() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let model = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.id == "mlx/qwen3-4b:4bit")
.unwrap();
let measured = 3 * 1024 * 1024 * 1024_u64;
let mut first = coordinator.reserve(&model, 2_048).unwrap();
first.reconcile_measured_weights(measured).unwrap();
first.publish_resident_weights(measured);
drop(first);
let mut replacement = coordinator.reserve(&model, 2_048).unwrap();
coordinator.mark_evicted(&model.id);
let preflight = replacement.reconcile_measured_weights(measured).unwrap();
assert!(preflight.estimated_incremental_mb >= 3 * 1024);
}
#[test]
fn maintenance_catalog_alias_blocks_provider_alias_reservation() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
coordinator
.register_model_aliases("mlx/kokoro-82m:6bit", ["mlx-community/Kokoro-82M-6bit"]);
let _maintenance = coordinator
.begin_model_maintenance("mlx/kokoro-82m:6bit")
.unwrap();
let mut provider_schema = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.is_local())
.expect("one local schema");
provider_schema.id = "mlx-community/Kokoro-82M-6bit".into();
let blocked = coordinator.reserve(&provider_schema, 512).unwrap_err();
assert_eq!(
blocked.preflight.verdict,
LocalLoadVerdict::ModelMaintenance
);
assert_eq!(blocked.preflight.model_id, "mlx/kokoro-82m:6bit");
}
#[test]
fn late_alias_registration_rekeys_active_and_resident_state_atomically() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let mut schema = crate::registry::builtin_catalog()
.into_iter()
.find(|model| model.is_local())
.unwrap();
schema.id = "provider/artifact".into();
let active = coordinator.reserve(&schema, 0).unwrap();
coordinator.register_model_aliases("catalog/model:default", ["provider/artifact"]);
assert!(matches!(
coordinator.begin_model_maintenance("catalog/model:default"),
Err(ModelMaintenanceError::ModelInUse(_))
));
drop(active);
let maintenance = coordinator
.begin_model_maintenance("catalog/model:default")
.unwrap();
let error = coordinator.reserve(&schema, 0).unwrap_err();
assert_eq!(error.preflight.verdict, LocalLoadVerdict::ModelMaintenance);
drop(maintenance);
}
#[test]
fn process_allocations_teardown_by_exact_owner_without_sibling_erasure() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(16.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let logical = "local/shared-model";
let worker = worker_process_allocation_id(logical);
let vllm = vllm_process_allocation_id(logical);
let bytes = 512 * 1024 * 1024_u64;
let mut worker_load = coordinator
.reserve_measured_host_allocation(logical, &worker, bytes, 0)
.unwrap();
worker_load.publish_resident_weights_as(&worker, bytes);
drop(worker_load);
let mut vllm_load = coordinator
.reserve_measured_host_allocation(logical, &vllm, bytes, 0)
.unwrap();
vllm_load.publish_resident_weights_as(&vllm, bytes);
drop(vllm_load);
assert_eq!(coordinator.resident_model_mb(), 1024);
coordinator.mark_teardown_pending_allocation(logical, &worker);
coordinator.mark_teardown_pending_allocation(logical, &vllm);
coordinator.finish_teardown_allocation(logical, &worker);
assert!(coordinator.teardown_pending(logical));
assert_eq!(
coordinator.resident_allocation_ids(logical),
vec![vllm.clone()]
);
assert_eq!(coordinator.resident_model_mb(), 512);
coordinator.finish_teardown_allocation(logical, &vllm);
assert!(!coordinator.teardown_pending(logical));
assert!(!coordinator.is_resident(logical));
}
#[test]
fn replacement_generation_reconcile_removes_peer_resident_discount_before_load() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(1.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let logical = "local/replaced-worker";
let bytes = 512 * 1024 * 1024_u64;
let mut old = coordinator
.reserve_measured_host_allocation(logical, "worker:old", bytes, 0)
.unwrap();
old.publish_resident_weights_as("worker:old", bytes);
drop(old);
let mut replacement = coordinator
.reserve_measured_host(logical, bytes, 0)
.expect("logical preflight initially sees the old resident");
assert_eq!(replacement.reserved_incremental_mb(), 0);
replacement.bind_allocation_id("worker:new");
replacement
.reconcile_measured_weights(bytes)
.expect("the exact replacement generation fits by itself");
assert_eq!(replacement.reserved_incremental_mb(), 512);
assert!(
coordinator
.reserve_measured_host("different/model", bytes, 0)
.is_err(),
"a second cold model must see both old residency and the replacement generation"
);
drop(replacement);
assert!(coordinator
.reserve_measured_host("different/model", bytes, 0)
.is_ok());
}
#[test]
fn pre_ack_pending_charge_blocks_other_scope_until_exact_exit_ack() {
let machine = Arc::new(Mutex::new(MachineAdmissionLedger::default()));
let first = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
ResourcePolicy::custom_gb(2.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
machine.clone(),
));
let second = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
ResourcePolicy::custom_gb(2.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
machine,
));
let logical = "managed/starting";
let allocation = worker_process_allocation_id(logical);
let cold = first
.reserve_measured_host_allocation(logical, &allocation, 1024 * 1024 * 1024, 0)
.unwrap();
first.mark_teardown_pending_allocation_with_charge(
logical,
&allocation,
cold.reconciled_weights_bytes(),
);
drop(cold);
let blocked = second
.reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
.unwrap_err();
assert_eq!(
blocked.preflight.verdict,
LocalLoadVerdict::ExceedsConfiguredCeiling
);
first.finish_teardown_allocation(logical, "worker:unrelated-sibling");
assert!(first.teardown_pending(logical));
assert_eq!(first.resident_model_mb(), 1024);
first.finish_teardown_allocation(logical, &allocation);
assert!(!first.teardown_pending(logical));
assert!(second
.reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
.is_ok());
}
#[test]
fn detached_native_lease_keeps_machine_charge_after_request_cancellation() {
let machine = Arc::new(Mutex::new(MachineAdmissionLedger::default()));
let first = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
ResourcePolicy::custom_gb(2.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
machine.clone(),
));
let second = Arc::new(LocalAdmissionCoordinator::with_probe_and_ledger(
ResourcePolicy::custom_gb(2.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
machine,
));
let reservation = first
.reserve_measured_host("detached/model", 1024 * 1024 * 1024, 0)
.unwrap();
let detached = reservation.detached_lease();
drop(reservation);
assert!(second
.reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
.is_err());
drop(detached);
assert!(second
.reserve_measured_host("other/model", 1536 * 1024 * 1024, 0)
.is_ok());
}
#[test]
fn normal_awaited_detached_work_shares_one_charge_with_its_request() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(1.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let request = coordinator
.reserve_measured_host("native/model-a", 512 * 1024 * 1024, 0)
.unwrap();
let detached = request.detached_lease();
let peer = coordinator
.reserve_measured_host("native/model-b", 512 * 1024 * 1024, 0)
.expect("request + its detached job are one 512 MB allocation, not two");
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_host_reservations_mb,
1024
);
drop(peer);
drop(detached);
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_host_reservations_mb,
512
);
drop(request);
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_host_reservations_mb,
0
);
}
#[test]
fn concurrent_non_aligned_starts_transfer_cold_weights_to_pending_without_double_charge() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
let first_allocation = vllm_process_allocation_id("managed/model-a");
let second_allocation = vllm_process_allocation_id("managed/model-b");
let mib = 1024 * 1024_u64;
let first_measured_bytes = 1024 * mib + 1;
let second_measured_bytes = 512 * mib + 1;
let mut first = coordinator
.reserve_measured_host_allocation(
"managed/model-a",
&first_allocation,
first_measured_bytes,
128,
)
.unwrap();
let mut second = coordinator
.reserve_measured_host_allocation(
"managed/model-b",
&second_allocation,
second_measured_bytes,
256,
)
.unwrap();
first.transfer_cold_weights_to_pending_allocation(&first_allocation, first_measured_bytes);
second
.transfer_cold_weights_to_pending_allocation(&second_allocation, second_measured_bytes);
assert_eq!(
coordinator.resident_model_mb(),
1538,
"each non-MiB-aligned allocation must be rounded up independently"
);
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_host_reservations_mb,
384,
"pending weights replace cold request weights while request overhead remains active"
);
first.publish_resident_weights_as(&first_allocation, first_measured_bytes);
second.publish_resident_weights_as(&second_allocation, second_measured_bytes);
assert_eq!(coordinator.resident_model_mb(), 1538);
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_host_reservations_mb,
384,
"publication must not subtract transferred weights twice"
);
drop(first);
drop(second);
assert_eq!(
coordinator
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_host_reservations_mb,
0
);
assert_eq!(coordinator.resident_model_mb(), 1538);
}
#[test]
fn pending_allocation_identity_is_idempotent_and_sibling_exact() {
let coordinator = Arc::new(LocalAdmissionCoordinator::with_probe(
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
Arc::new(FixedLiveMemoryProbe::known(24_000)),
));
coordinator.mark_teardown_pending_allocation_with_charge(
"same/model",
"worker:same/model",
512 * 1024 * 1024,
);
coordinator.mark_teardown_pending_allocation_with_charge(
"same/model",
"worker:same/model",
512 * 1024 * 1024,
);
coordinator.mark_teardown_pending_allocation_with_charge(
"same/model",
"vllm:same/model",
256 * 1024 * 1024,
);
assert_eq!(coordinator.resident_model_mb(), 768);
coordinator.finish_teardown_allocation("same/model", "worker:same/model");
assert!(coordinator.teardown_pending("same/model"));
assert_eq!(coordinator.resident_model_mb(), 256);
coordinator.finish_teardown_allocation("same/model", "worker:same/model");
assert_eq!(coordinator.resident_model_mb(), 256);
coordinator.finish_teardown_allocation("same/model", "vllm:same/model");
assert!(!coordinator.teardown_pending("same/model"));
}
#[test]
fn scoped_coordinator_identity_is_stable_for_one_state_root() {
let root = tempfile::tempdir().unwrap();
let first = scoped_local_admission(
root.path(),
ResourcePolicy::custom_gb(4.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
);
let second = scoped_local_admission(
root.path(),
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
);
assert!(Arc::ptr_eq(&first, &second));
assert_eq!(second.policy(), ResourcePolicy::custom_gb(8.0).unwrap());
}
#[cfg(unix)]
#[test]
fn scoped_coordinator_identity_unifies_symlinked_state_roots() {
use std::os::unix::fs::symlink;
let fixture = tempfile::tempdir().unwrap();
let real = fixture.path().join("real-state");
std::fs::create_dir(&real).unwrap();
let alias = fixture.path().join("state-alias");
symlink(&real, &alias).unwrap();
let first = scoped_local_admission(
&real,
ResourcePolicy::custom_gb(4.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
);
let second = scoped_local_admission(
&alias,
ResourcePolicy::custom_gb(8.0).unwrap(),
hardware(32 * 1024, GpuBackend::Cpu, None),
);
assert!(Arc::ptr_eq(&first, &second));
assert_eq!(second.policy(), ResourcePolicy::custom_gb(8.0).unwrap());
}
#[test]
fn unavailable_state_root_normalization_is_absolute_and_lexically_stable() {
let fixture = tempfile::tempdir().unwrap();
let missing = fixture.path().join("not-created").join("..").join("state");
assert_eq!(
normalized_state_root_key(&missing),
normalized_state_root_key(&fixture.path().join("state"))
);
}
#[cfg(unix)]
#[test]
fn missing_leaf_under_symlinked_parent_keeps_one_scope_identity() {
use std::os::unix::fs::symlink;
let fixture = tempfile::tempdir().unwrap();
let real = fixture.path().join("real");
std::fs::create_dir(&real).unwrap();
let alias = fixture.path().join("alias");
symlink(&real, &alias).unwrap();
assert_eq!(
normalized_state_root_key(&alias.join("missing").join("state")),
normalized_state_root_key(&real.join("missing").join("state"))
);
}
#[cfg(unix)]
#[test]
fn parent_components_are_resolved_after_symlinks_not_lexically_before_them() {
use std::os::unix::fs::symlink;
let fixture = tempfile::tempdir().unwrap();
let physical_parent = fixture.path().join("physical");
let physical_child = physical_parent.join("child");
std::fs::create_dir_all(&physical_child).unwrap();
let aliases = fixture.path().join("aliases");
std::fs::create_dir(&aliases).unwrap();
let alias = aliases.join("runtime");
symlink(&physical_child, &alias).unwrap();
let through_alias = alias.join("..").join("missing-state");
assert_eq!(
normalized_state_root_key(&through_alias),
normalized_state_root_key(&physical_parent.join("missing-state"))
);
assert_ne!(
normalized_state_root_key(&through_alias),
aliases.join("missing-state")
);
}
fn assert_no_temp_files(directory: &std::path::Path) {
let entries = std::fs::read_dir(directory)
.unwrap()
.map(|entry| entry.unwrap().path())
.collect::<Vec<_>>();
assert_eq!(entries, vec![directory.join("model-resource-policy.json")]);
}
#[cfg(unix)]
fn assert_private_mode(path: &std::path::Path, expected: u32) {
use std::os::unix::fs::PermissionsExt;
assert_eq!(
std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
expected
);
}
#[cfg(not(unix))]
fn assert_private_mode(_path: &std::path::Path, _expected: u32) {}
}
/// Excerpts of the real `config.json` files (the geometry keys, verbatim) of
/// the two checkpoints whose memory was measured through vllm-mlx 0.6.0 on an
/// M5 Pro, 2026-10-26.
#[cfg(test)]
mod measured_vllm_mlx_memory {
use super::*;
const QWEN3_8_27B_4BIT: &str = r#"{"model_type":"qwen3_5","text_config":{"model_type":"qwen3_5_text","num_hidden_layers":64,"num_attention_heads":24,"num_key_value_heads":4,"head_dim":256,"hidden_size":5120,"vocab_size":248320,"full_attention_interval":4,"layer_types":["linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention","linear_attention","linear_attention","linear_attention","full_attention"],"linear_conv_kernel_dim":4,"linear_key_head_dim":128,"linear_num_key_heads":16,"linear_num_value_heads":48,"linear_value_head_dim":128}}"#;
const GLM_4_7_FLASH_8BIT: &str = r#"{"model_type":"glm4_moe_lite","num_hidden_layers":47,"num_attention_heads":20,"num_key_value_heads":20,"hidden_size":2048,"vocab_size":154880,"kv_lora_rank":512,"qk_rope_head_dim":64,"qk_nope_head_dim":192,"v_head_dim":256}"#;
fn geometry(raw: &str) -> KvGeometry {
let raw: serde_json::Value = serde_json::from_str(raw).unwrap();
let source = raw.get("text_config").cloned().unwrap_or(raw);
geometry_from_config(&serde_json::from_value(source).unwrap()).unwrap()
}
fn m5_pro_64gb() -> crate::hardware::HardwareInfo {
crate::hardware::HardwareInfo {
os: "macos".into(),
arch: "aarch64".into(),
cpu_cores: 12,
total_ram_mb: 64 * 1024,
gpu_backend: GpuBackend::Metal,
gpu_memory_mb: None,
gpu_devices: Vec::new(),
recommended_model: "fixture".into(),
recommended_context: 4_096,
max_model_mb: 64 * 1024,
}
}
#[test]
fn a_hybrid_model_caches_only_its_full_attention_layers() {
let g = geometry(QWEN3_8_27B_4BIT);
// 16 of 64 layers are full attention: 2 × 16 × 4 KV heads × 256 × 2.
assert_eq!(g.kv_bytes_per_token, 65_536);
// 48 gated-delta layers' fixed state; measured as the cache's
// length-independent 0.15 GiB.
assert_eq!(g.fixed_state_bytes, 153_944_064);
assert_eq!(g.vocab_size, Some(248_320));
}
#[test]
fn an_mla_model_caches_its_latent_and_pays_for_the_expansion_in_prefill() {
let g = geometry(GLM_4_7_FLASH_8BIT);
// (512 latent + 64 rope) × 2 bytes × 47 layers, as measured.
assert_eq!(g.kv_bytes_per_token, 54_144);
// Measured near 290 KiB/token beyond the cache; the derivation is
// within 25% above it.
let measured_scratch = 289 * 1024;
assert!(g.prefill_bytes_per_token >= measured_scratch);
assert!(g.prefill_bytes_per_token <= measured_scratch * 5 / 4);
}
/// MLX peak (`/v1/status` metal `peak_memory_gb`) through vllm-mlx 0.6.0,
/// served with CAR's own flags (Qwen3.8 with its MTP drafter, so through
/// mlx-vlm), at about 8k, 32k and 64k prompt tokens.
const MEASURED_PEAK_GIB: [(&str, &str, [(usize, f64); 3]); 2] = [
(
"vllm-mlx/qwen3.8-27b:4bit",
QWEN3_8_27B_4BIT,
[(8_192, 19.78), (32_768, 21.64), (65_536, 24.24)],
),
(
"vllm-mlx/glm-4.7-flash:8bit",
GLM_4_7_FLASH_8BIT,
[(8_192, 34.84), (32_768, 42.45), (65_536, 53.47)],
),
];
#[test]
fn the_estimate_brackets_what_vllm_mlx_was_measured_to_use() {
let catalog = crate::registry::builtin_catalog();
for (id, config, points) in MEASURED_PEAK_GIB {
let model = catalog.iter().find(|m| m.id == id).unwrap();
let g = geometry(config);
for (tokens, measured) in points {
let estimate = estimate_model_memory_with_geometry(
model,
&m5_pro_64gb(),
tokens,
None,
Some(&g),
);
let gib = estimate.estimated_peak_mb as f64 / 1024.0;
// Never below what the server used (an under-estimate admits a
// load the machine cannot hold), and not so far above that a
// load that fits is refused.
assert!(
gib >= measured,
"{id} at {tokens}: estimate {gib:.1} GiB < measured {measured} GiB"
);
assert!(
gib <= measured + 6.0,
"{id} at {tokens}: estimate {gib:.1} GiB > measured {measured} + 6 GiB"
);
}
}
}
fn parse(raw: &str) -> Option<KvGeometry> {
geometry_from_config(&serde_json::from_str(raw).unwrap())
}
#[test]
fn an_absurd_config_saturates_instead_of_wrapping_to_a_small_estimate() {
let huge = parse(r#"{"num_hidden_layers":4294967296,"num_key_value_heads":4294967296,"head_dim":4294967296,"hidden_size":4611686018427387904}"#).unwrap();
assert_eq!(huge.kv_bytes_per_token, u64::MAX);
assert_eq!(huge.prefill_bytes_per_token, u64::MAX);
}
#[test]
fn mla_on_an_architecture_not_known_to_cache_the_latent_is_sized_per_head() {
// GLM-4.7-Flash's geometry under minicpm3's name, which mlx-lm caches
// as expanded keys and values.
let renamed = GLM_4_7_FLASH_8BIT.replace("glm4_moe_lite", "minicpm3");
let expanded = parse(&renamed).unwrap();
// 47 layers × 20 heads × ((192 nope + 64 rope) keys + 256 values) × 2.
assert_eq!(expanded.kv_bytes_per_token, 47 * 20 * (256 + 256) * 2);
assert!(expanded.kv_bytes_per_token > geometry(GLM_4_7_FLASH_8BIT).kv_bytes_per_token);
}
#[test]
fn linear_layers_without_state_dimensions_stay_counted_as_cache_layers() {
let undimensioned = parse(
r#"{"num_hidden_layers":4,"num_key_value_heads":4,"head_dim":256,"layer_types":["linear_attention","linear_attention","linear_attention","full_attention"]}"#,
)
.unwrap();
assert_eq!(undimensioned.kv_bytes_per_token, 4 * 2 * 4 * 256 * 2);
assert_eq!(undimensioned.fixed_state_bytes, 0);
}
#[test]
fn a_row_served_from_the_hugging_face_cache_is_sized_from_that_snapshot() {
let tmp = tempfile::tempdir().unwrap();
let hub = tmp.path().join("hub");
let repo = hub.join("models--mlx-community--Qwen3.8-27B-4bit");
let snapshot = repo.join("snapshots/fixture");
std::fs::create_dir_all(&snapshot).unwrap();
std::fs::write(snapshot.join("config.json"), QWEN3_8_27B_4BIT).unwrap();
std::fs::create_dir_all(repo.join("refs")).unwrap();
std::fs::write(repo.join("refs/main"), "fixture").unwrap();
// The managed directory holds only the drafter, as on a real install.
let models_dir = tmp.path().join("models");
std::fs::create_dir_all(models_dir.join("Qwen3.8-27B-MLX-vLLM/mtp-drafter")).unwrap();
let catalog = crate::registry::builtin_catalog();
let model = catalog
.iter()
.find(|m| m.id == "vllm-mlx/qwen3.8-27b:4bit")
.unwrap();
let served = kv_geometry_for(model, Some(&models_dir), Some(&hub));
assert_eq!(served.map(|g| g.kv_bytes_per_token), Some(65_536));
// Without a hub only the managed directory counts, so a test (or a
// caller that did not ask) never sizes a row from someone's cache.
assert_eq!(kv_geometry_for(model, Some(&models_dir), None), None);
}
}