car_inference/

registry.rs

//! Unified model registry — local and remote models under one schema.
//!
//! Replaces the hardcoded `ModelRegistry` from `models.rs` with a schema-driven
//! registry that treats all models as first-class typed resources. Users can
//! register custom models (fine-tuned endpoints, private APIs) alongside the
//! built-in catalog.

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::time::SystemTime;

use serde::{Deserialize, Serialize};
use tracing::{info, warn};

use crate::download::{DownloadEvent, ProgressSink};
use crate::schema::*;
use crate::InferenceError;

/// Filter for querying the registry.
#[derive(Debug, Clone, Default)]
pub struct ModelFilter {
    /// Required capabilities (model must have ALL of these).
    pub capabilities: Vec<ModelCapability>,
    /// Maximum on-disk / RAM size in MB.
    pub max_size_mb: Option<u64>,
    /// Maximum expected latency in ms (from declared envelope).
    pub max_latency_ms: Option<u64>,
    /// Maximum cost per 1M output tokens in USD.
    pub max_cost_per_mtok: Option<f64>,
    /// Required tags (model must have ALL of these).
    pub tags: Vec<String>,
    /// Filter by provider.
    pub provider: Option<String>,
    /// Only local models.
    pub local_only: bool,
    /// Only models that are currently available.
    pub available_only: bool,
}

/// A curated replacement for a local model that is installed but no longer
/// the preferred model in its line.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModelUpgrade {
    pub from_id: String,
    pub from_name: String,
    pub to_id: String,
    pub to_name: String,
    pub reason: String,
    pub target_runtime: Option<String>,
    pub target_runtime_requirement: Option<String>,
    pub minimum_runtimes: Vec<ModelRuntimeRequirement>,
    pub target_available: bool,
    pub target_pullable: bool,
    pub remove_old_supported: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModelRuntimeRequirement {
    pub name: String,
    pub minimum_version: String,
}

/// How a registry answers "is a Parslee session signed in?", and whether it may
/// act on a "no" by discarding session-scoped evidence.
///
/// Both call sites used to read `car_auth::access_token_is_available()`
/// directly, which quietly made *constructing* a registry a process-global
/// write: `new_with_catalog_public_key` ends in a `refresh_availability`, and on
/// a machine with no session that refresh clears the learned gateway and
/// credential observations. #989 took the crate's serial lock in the two tests
/// that call `refresh_availability` themselves, but roughly twenty others merely
/// build a registry and reached the same clear through construction — so a test
/// that had just recorded an observation could have it erased under it
/// (Parslee-ai/car#986). This probe is the seam that separates *what the answer
/// is* from *whether acting on it is allowed*.
#[derive(Debug, Clone, Copy)]
pub(crate) enum SessionProbe {
    /// Production. Ask `car_auth` on every refresh and ACT on the answer — a
    /// signed-out answer discards the session-scoped gateway/credential
    /// evidence (Parslee-ai/car#786, Parslee-ai/car#887).
    // Never constructed in a test build — `Inert` is the `cfg(test)` default,
    // which is the whole point of the seam rather than dead code.
    #[cfg_attr(test, allow(dead_code))]
    Live,
    /// A caller-supplied answer, acted on exactly as `Live` acts on it:
    /// `Fixed(false)` discards, `Fixed(true)` does not. A test that means to
    /// exercise the discard asks for it here rather than depending on the
    /// ambient auth state of the machine it runs on.
    #[cfg(test)]
    Fixed(bool),
    /// The real answer, asked exactly as `Live` asks it, and never acted on.
    /// Availability is therefore identical to `Live`, and the only difference
    /// is that nothing outside the registry is written. Used by read-only
    /// diagnostics and as the default under `cfg(test)`.
    Inert,
}

impl SessionProbe {
    /// Whether a Parslee session is signed in, using only the prompt-free
    /// authority hint. Registry construction and catalog refresh must not turn
    /// into a physical Keychain read.
    fn available(&self) -> bool {
        match self {
            Self::Live => passive_parslee_oauth_available(),
            Self::Inert => passive_parslee_oauth_available(),
            #[cfg(test)]
            Self::Fixed(available) => *available,
        }
    }

    /// Whether the authority is explicitly signed out rather than merely
    /// unknown or temporarily unreadable.
    fn signed_out(&self) -> bool {
        match self {
            Self::Live => matches!(
                car_auth::credential_authority_hint().state,
                car_auth::CredentialAuthorityState::SignedOut
            ),
            Self::Inert => matches!(
                car_auth::credential_authority_hint().state,
                car_auth::CredentialAuthorityState::SignedOut
            ),
            #[cfg(test)]
            Self::Fixed(available) => !available,
        }
    }

    /// Whether a "not signed in" answer may discard the session-scoped
    /// gateway/credential observations. False only for [`Self::Inert`], which
    /// exists precisely so that building a registry has no effect outside it.
    fn may_forget_session_evidence(&self) -> bool {
        match self {
            Self::Live => true,
            #[cfg(test)]
            Self::Fixed(_) => true,
            Self::Inert => false,
        }
    }
}

/// Unified registry of all known models.
#[derive(Clone)]
pub struct UnifiedRegistry {
    models_dir: PathBuf,
    /// CAR state root — `~/.car`, or `$CAR_HOME` when the operator relocated
    /// this daemon. Every *state* path the registry touches (`models.json`, the
    /// signed-catalog cache, the discovery cache) hangs off this, never off
    /// `models_dir`, which is the machine-shared weights cache and deliberately
    /// does not move.
    state_root: PathBuf,
    /// All registered models, keyed by id.
    models: HashMap<String, ModelSchema>,
    /// Exact ids owned by the compiled or authenticated project catalog.
    /// Public/user-controlled rows are additive and may never replace these.
    project_model_ids: HashSet<String>,
    /// The strongest reserved namespace: exact ids compiled into this build.
    /// Even an authenticated refresh is additive and cannot change them.
    builtin_model_ids: HashSet<String>,
    /// IDs whose current rows came from the user-controlled `models.json`
    /// boundary or an explicit user registration.
    ///
    /// Tags cannot carry this provenance: a user can add `builtin`, while a
    /// valid signed catalog row does not have to. Keeping the boundary private
    /// prevents generic/builtin/catalog/discovery registration from being
    /// serialized back into user config and losing its trust tier on restart.
    user_config_ids: HashSet<String>,
    /// IDs synthesized solely from directories found under `models_dir`.
    ///
    /// These rows exist only while their directories do. Keep provenance in a
    /// private set rather than trusting the public `auto-discovered` tag: a
    /// user-configured schema may carry any tag and must not disappear merely
    /// because no same-named local directory exists.
    on_disk_discovered_ids: HashSet<String>,
    /// User-added model config file path — `models.json` resolved against
    /// `state_root`, i.e. the same answer the free [`user_config_path`] gives.
    user_config_path: PathBuf,
    /// Where to report downloads that nobody explicitly asked for.
    ///
    /// [`ensure_local`](Self::ensure_local) is the *implicit* acquisition path:
    /// it runs inside `generate`, when the router picks a model whose weights
    /// aren't on disk yet. It used to hard-code `ProgressSink::none()`, so a
    /// multi-gigabyte fetch triggered by a plain `car infer "hi"` produced no
    /// output on any surface — the command simply appeared to hang, and a
    /// user on a metered connection had no signal at all (Parslee-ai/car#620).
    ///
    /// Explicit pulls (`car models pull`) always passed a real sink and were
    /// never affected; this closes the gap for the path users actually hit
    /// first. Defaults to [`ProgressSink::none`], so an embedder that doesn't
    /// opt in behaves exactly as before.
    ambient_progress: ProgressSink,
    /// How this registry learns whether a Parslee session is signed in, and
    /// whether it may act on a "no". See [`SessionProbe`]: normal production
    /// registries carry `Live`; read-only diagnostics and test registries use
    /// `Inert` so construction cannot erase an existing observation.
    session: SessionProbe,
}

#[derive(Debug, Clone, Deserialize)]
struct ModelUpgradeRule {
    from_ids: Vec<String>,
    to_id: String,
    reason: String,
    target_runtime: Option<String>,
    target_runtime_requirement: Option<String>,
    #[serde(default)]
    minimum_runtimes: Vec<ModelRuntimeRequirement>,
    #[serde(default = "default_remove_old_after_available")]
    remove_old_after_available: bool,
}

fn default_remove_old_after_available() -> bool {
    true
}

fn environment_credential_available(env_var: &str) -> bool {
    std::env::var(env_var).is_ok_and(|value| !value.trim().is_empty())
}

fn passive_parslee_oauth_available() -> bool {
    matches!(
        car_auth::credential_authority_hint().state,
        car_auth::CredentialAuthorityState::Configured
    )
}

/// `resolved` is the per-refresh credential memo. Passive catalog refreshes
/// populate it from environment presence only; request-time refreshes may
/// populate it authoritatively from the secret store. Keeping the source of
/// truth outside this helper prevents a single-model registration from
/// silently becoming a physical Keychain read.
fn proprietary_auth_available(
    model_id: &str,
    schema_provider: &str,
    source_provider: &str,
    auth: &ProprietaryAuth,
    parslee_oauth_available: bool,
    resolved: &std::collections::HashMap<String, bool>,
) -> bool {
    // Authentication is necessary but NOT sufficient for a PROXIED namespace.
    // The OAuth2Pkce arm below resolves to exactly "is a Parslee session signed
    // in", which is true regardless of what the gateway can actually serve —
    // so ten `parslee/openrouter/*` aliases advertised `available` while every
    // one of them 503'd with `openrouter_not_configured`, costing a benchmark
    // sweep that picked one on the strength of that claim (Parslee-ai/car#786).
    //
    // There is no discovery endpoint to consult instead (`parslee.capabilities`
    // enumerates product entitlements, not inference upstreams), so the gateway's
    // own answer to a real request is the only truthful signal there is. Once it
    // has told us, stop advertising — the catalog is then optimistic once and
    // self-correcting, rather than permanently wrong.
    //
    // Checked here rather than at the call sites so `register` and
    // `refresh_availability` cannot drift apart.
    if crate::openrouter::is_curated_managed_gateway_alias(model_id)
        && crate::openrouter::gateway_unconfigured()
    {
        return false;
    }
    // The neighbouring half of the same mistake, one layer earlier: a signed-in
    // session is not a WORKING one. `access_token_is_available` is documented
    // existence-only — it reports that a V2 record has an active slot, never
    // that the token is still accepted — so a machine with a stale sign-in
    // offered every managed alias as a top-quality candidate and 401'd on every
    // one, per call, forever (Parslee-ai/car#887).
    //
    // Unlike the gateway check above this is NOT scoped to the curated
    // OpenRouter aliases: a dead credential kills every `parslee/*` row, so the
    // suppression belongs on the OAuth2Pkce arm as a whole.
    //
    // Checked here rather than at the call sites for the same reason as the
    // gateway verdict — so `register` and `refresh_availability` cannot drift.
    if crate::parslee_credential::credential_rejected()
        && matches!(auth, ProprietaryAuth::OAuth2Pkce { .. })
    {
        return false;
    }
    match auth {
        ProprietaryAuth::ApiKeyEnv { env_var } | ProprietaryAuth::BearerTokenEnv { env_var } => {
            resolved.get(env_var).copied().unwrap_or(false)
        }
        ProprietaryAuth::OAuth2Pkce { .. } => {
            schema_provider.eq_ignore_ascii_case("parslee")
                && source_provider.eq_ignore_ascii_case("parslee")
                && parslee_oauth_available
        }
    }
}

fn model_upgrade_rules() -> Vec<ModelUpgradeRule> {
    serde_json::from_str(include_str!("../assets/model-upgrades.json"))
        .expect("built-in model-upgrades.json should parse")
}

/// File name of the user-registered model config under the CAR state root.
pub const USER_MODELS_FILE: &str = "models.json";

/// The one resolver for `models.json`: [`USER_MODELS_FILE`] under the CAR state
/// root (`~/.car/models.json` unless `CAR_HOME` moves the root).
///
/// Every reader and writer in the tree goes through this — the registry's own
/// load/save, and the daemon's `models.register` / `models.unregister`
/// handlers. They have to agree: when the write side moved to `CAR_HOME` and
/// the read side stayed derived from the (deliberately unmoved) weights dir, a
/// relocated daemon persisted a registration to one file and read a different
/// one on the next boot, so the registration silently never took effect.
///
/// `None` only when there is no `CAR_HOME`, `HOME` or `USERPROFILE` — the same
/// condition every other state path treats as "cannot resolve". The registry
/// itself joins [`USER_MODELS_FILE`] onto the state root it was constructed
/// with, which for [`UnifiedRegistry::new`] is the infallible
/// `car_home::root_or_relative()`; same root, same file name, differing only in
/// what they do when nothing resolves at all (error here, relative `.car`
/// there — the pre-existing behaviour of each caller).
pub fn user_config_path() -> Option<PathBuf> {
    car_home::root().map(|root| root.join(USER_MODELS_FILE))
}

impl UnifiedRegistry {
    /// Registry for the install that is actually running: state files resolve
    /// under the CAR state root (`$CAR_HOME`, else `~/.car`), weights under
    /// `models_dir`.
    pub fn new(models_dir: PathBuf) -> Self {
        Self::new_with_state_root(car_home::root_or_relative(), models_dir)
    }

    /// [`new`](Self::new) with the state root supplied rather than resolved
    /// from the environment. The daemon passes its own resolved root; tests and
    /// embedders that want every file inside one directory pass that directory.
    pub fn new_with_state_root(state_root: PathBuf, models_dir: PathBuf) -> Self {
        let catalog_public_key = std::env::var("CAR_CATALOG_PUBKEY").ok();
        Self::new_with_catalog_public_key(state_root, models_dir, catalog_public_key.as_deref())
    }

    fn new_with_catalog_public_key(
        state_root: PathBuf,
        models_dir: PathBuf,
        catalog_public_key: Option<&str>,
    ) -> Self {
        // Production asks `car_auth` live and acts on the answer. Under
        // `cfg(test)` the same question is asked the same way but held INERT, so
        // building a registry — which every registry test does, and most of them
        // without taking the crate's serial lock — reports identical
        // availability while clearing nothing outside itself
        // (Parslee-ai/car#986). A test that means to exercise the clear asks for
        // it explicitly with `new_with_session(.., SessionProbe::Fixed(false))`.
        #[cfg(not(test))]
        let session = SessionProbe::Live;
        #[cfg(test)]
        let session = SessionProbe::Inert;
        Self::new_with_session(state_root, models_dir, catalog_public_key, session)
    }

    /// [`new_with_catalog_public_key`](Self::new_with_catalog_public_key) with
    /// the session answer supplied rather than defaulted. See [`SessionProbe`].
    pub(crate) fn new_with_session(
        state_root: PathBuf,
        models_dir: PathBuf,
        catalog_public_key: Option<&str>,
        session: SessionProbe,
    ) -> Self {
        let user_config_path = state_root.join(USER_MODELS_FILE);

        let mut registry = Self {
            models_dir,
            state_root,
            models: HashMap::new(),
            project_model_ids: HashSet::new(),
            builtin_model_ids: HashSet::new(),
            user_config_ids: HashSet::new(),
            on_disk_discovered_ids: HashSet::new(),
            user_config_path,
            ambient_progress: ProgressSink::none(),
            session,
        };
        registry.load_builtin_catalog();
        // Refreshable signed catalog (E1): a prior `refresh_catalog` stores the
        // exact authenticated body + signature envelope. Startup re-verifies
        // that envelope with the currently configured key before loading any
        // model additively alongside the built-ins.
        for schema in crate::catalog::load_cache(
            &crate::catalog::cache_path(&registry.state_root),
            catalog_public_key,
        ) {
            registry.register_signed_catalog_model(schema);
        }
        // Auto-discovered models (E2): Community-tier entries cached by a prior
        // discovery pass (provider /v1/models). Loaded on top of built-ins +
        // signed catalog, but never *overwriting* a curated/signed entry of the
        // same id — discovery only ever ADDS models the catalog doesn't have.
        for schema in crate::discovery::load_cache(&crate::discovery::cache_path(
            &registry.state_models_dir(),
        )) {
            if !registry.models.contains_key(&schema.id) {
                registry.register(schema);
            }
        }
        registry.refresh_availability();
        // Load user config on top (silently ignore if missing)
        let _ = registry.load_user_config();
        // Surface models the user pulled/placed under ~/.car/models/ that no
        // catalog or user-config entry covers, so a locally-present model is
        // visible and routable instead of invisible (car-releases#62).
        registry.discover_on_disk_models();
        registry
    }

    fn empty_with_state_root(state_root: PathBuf, models_dir: PathBuf) -> Self {
        let user_config_path = state_root.join(USER_MODELS_FILE);
        Self {
            models_dir,
            state_root,
            models: HashMap::new(),
            project_model_ids: HashSet::new(),
            builtin_model_ids: HashSet::new(),
            user_config_ids: HashSet::new(),
            on_disk_discovered_ids: HashSet::new(),
            user_config_path,
            ambient_progress: ProgressSink::none(),
            // No refresh runs during construction, so this probe is never
            // consulted. Keep it inert in case an internal caller refreshes.
            session: SessionProbe::Inert,
        }
    }

    /// Test-only: a registry with NO builtin catalog, signed cache, discovery
    /// cache, or user config — only the models a test explicitly `register`s.
    /// Routing unit tests that assert a synthetic model wins MUST use this:
    /// `new()` loads the real builtin frontier models, which become routable
    /// candidates the moment a test sets their `api_key_env` (e.g.
    /// `OPENAI_API_KEY`) — and once those models carry real benchmark scores
    /// (e.g. car-judged), they legitimately outrank the synthetic stand-ins and
    /// silently break the test's assumption. An empty registry keeps the test
    /// hermetic.
    ///
    /// The state root is taken to be `models_dir`'s parent, so a test that
    /// passes `<tmp>/models` gets `<tmp>` and every state file stays inside the
    /// temp dir. This constructor never reads or writes anything on its own, so
    /// the derivation only matters if the test then calls `save_user_config`.
    #[cfg(test)]
    pub fn new_empty(models_dir: PathBuf) -> Self {
        let state_root = models_dir.parent().unwrap_or(&models_dir).to_path_buf();
        Self::empty_with_state_root(state_root, models_dir)
    }

    /// Registry for a deterministic offline diagnosis fixture.
    ///
    /// It reads only the explicitly named models directory: no builtin,
    /// signed, discovered, or user catalog is loaded, and no credential or
    /// session availability probe runs. Local model directories are inferred
    /// with the same conservative on-disk discovery used by production.
    pub(crate) fn new_isolated_for_diagnosis(state_root: PathBuf, models_dir: PathBuf) -> Self {
        let mut registry = Self::empty_with_state_root(state_root, models_dir);
        registry.discover_on_disk_models();
        registry
    }

    /// Where the registry's own state that has always sat beside the weights
    /// lives: `models/` under the state root. Identical to `models_dir`
    /// whenever `CAR_HOME` is unset, which is why no existing install's files
    /// move; different — and per-daemon — once it is set.
    fn state_models_dir(&self) -> PathBuf {
        self.state_root.join("models")
    }

    /// Register on-disk models under `models_dir` that nothing else already
    /// covers. Curated catalog entries already light up via the
    /// availability check (their `name` maps to `<models_dir>/<name>`), so
    /// this only fills the gap for *uncatalogued* local models — e.g. a
    /// model a user fetched by hand.
    ///
    /// Capability guessing from a bare directory is unreliable, and
    /// mis-tagging a speech/vision/video checkpoint as `Generate` would
    /// poison routing. So this is deliberately conservative: it only
    /// registers text LLMs it can positively identify by a recognized
    /// `model_type` in `config.json` (MLX) or by a GGUF weight file, infers
    /// embed/rerank from the directory name, and skips everything else.
    /// car-releases#62.
    fn discover_on_disk_models(&mut self) {
        let entries = match std::fs::read_dir(&self.models_dir) {
            Ok(e) => e,
            Err(_) => return,
        };
        // Names already registered (case-insensitive) — never shadow a
        // curated/user/discovered entry with an inferred one.
        let known: std::collections::HashSet<String> = self
            .models
            .values()
            .map(|m| m.name.to_ascii_lowercase())
            .collect();

        for entry in entries.flatten() {
            let path = entry.path();
            if !path.is_dir() {
                continue;
            }
            let Some(name) = path
                .file_name()
                .and_then(|n| n.to_str())
                .map(str::to_string)
            else {
                continue;
            };
            if known.contains(&name.to_ascii_lowercase()) {
                continue;
            }

            let Some(schema) = synthesize_local_schema(&name, &path) else {
                continue;
            };
            tracing::info!(
                id = %schema.id,
                name = %name,
                "auto-discovered uncatalogued local model under models_dir (car-releases#62)"
            );
            let id = schema.id.clone();
            self.register(schema);
            if self.models.contains_key(&id) {
                self.on_disk_discovered_ids.insert(id);
            }
        }
    }

    /// Remove inferred rows whose source directories no longer exist.
    ///
    /// The daemon keeps its registry in memory, so an on-disk model removed by
    /// another process otherwise remains visible until restart. Activity and
    /// mutation lock records are durable coordination sentinels, not registry
    /// entries, and deliberately have no bearing on this reconciliation.
    pub(crate) fn prune_missing_on_disk_models(&mut self) {
        let missing = self
            .on_disk_discovered_ids
            .iter()
            .filter(|id| {
                self.models
                    .get(id.as_str())
                    .is_none_or(|schema| !self.models_dir.join(&schema.name).is_dir())
            })
            .cloned()
            .collect::<Vec<_>>();
        for id in missing {
            tracing::debug!(
                model_id = %id,
                "dropping vanished auto-discovered model; coordination records do not register models"
            );
            self.on_disk_discovered_ids.remove(&id);
            self.models.remove(&id);
        }
    }

    /// Register a model at a public runtime boundary.
    ///
    /// Callers cannot confer project curation: even a legacy schema whose
    /// omitted `trust_tier` deserializes as `Curated` is normalized to
    /// `Community` here. Project-owned builtins and signature-verified catalogs
    /// use the crate-private [`register_project_model`](Self::register_project_model)
    /// boundary instead.
    pub fn register(&mut self, mut schema: ModelSchema) {
        if crate::openrouter::is_curated_managed_gateway_alias(&schema.id) {
            warn!(id = %schema.id, "ignoring user registration for reserved Parslee-managed alias");
            return;
        }
        if self.project_model_ids.contains(&schema.id) {
            warn!(id = %schema.id, "ignoring public registration for project-owned exact id");
            return;
        }
        schema.mark_user_registered();
        let id = schema.id.clone();
        if self.register_preserving_trust(schema) {
            self.on_disk_discovered_ids.remove(&id);
        }
    }

    /// Register a model whose provenance was established by CAR itself.
    ///
    /// This trust-preserving path is intentionally crate-private. Production
    /// callers are limited to compiled builtins and signature-verified catalog
    /// rows; tests may use it to construct project-curated fixtures.
    pub(crate) fn register_project_model(&mut self, schema: ModelSchema) -> bool {
        let id = schema.id.clone();
        if !self.register_preserving_trust(schema) {
            return false;
        }
        self.on_disk_discovered_ids.remove(&id);
        self.project_model_ids.insert(id);
        true
    }

    /// Add one signature-verified catalog row without allowing the remote
    /// publisher to redefine an exact id already owned by this build or an
    /// earlier authenticated row in the same document.
    fn register_signed_catalog_model(&mut self, schema: ModelSchema) {
        if self.builtin_model_ids.contains(&schema.id) {
            warn!(id = %schema.id, "ignoring signed catalog row for compiled builtin exact id");
            return;
        }
        if self.project_model_ids.contains(&schema.id) {
            warn!(id = %schema.id, "ignoring duplicate signed catalog row for project-owned exact id");
            return;
        }
        self.register_project_model(schema);
    }

    fn register_preserving_trust(&mut self, mut schema: ModelSchema) -> bool {
        if let Err(error) = crate::catalog_identity::row_digest(&schema) {
            warn!(id = %schema.id, %error, "rejecting model without canonical catalog identity");
            return false;
        }
        // Check availability for local models
        if schema.is_mlx() || schema.is_car_managed_vllm_mlx() {
            // MLX requires Apple Silicon Metal. On any other build target —
            // Intel Mac, Linux, Windows, or `car_skip_mlx` — the backend
            // is not compiled in and any execution attempt would fail at
            // dispatch with a "model not found" or backend-missing error.
            // Mirror the AppleFoundationModels cfg-gating below: the
            // registry must reflect that the model is *not* runnable here
            // so the adaptive router doesn't add it to fallback chains
            // (Parslee-ai/car#231 — §7.1, the "fresh Windows install has
            // no usable inference path" finding). Same shape as the
            // `refresh_availability` MLX branch.
            #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
            {
                schema.available = if schema.tags.contains(&"speech".to_string()) {
                    speech_mlx_available()
                } else if let ModelSource::Mlx { ref hf_repo, .. }
                | ModelSource::ManagedVllmMlx { ref hf_repo, .. } = schema.source
                {
                    // Available if cached locally OR has an hf_repo —
                    // ensure_local() lazy-downloads on first use, so a
                    // declared hf_repo is "functionally available" the
                    // same way Ollama/RemoteApi entries are. Mirrors the
                    // refresh_availability() check below; see #164.
                    let mlx_dir = self.models_dir.join(&schema.name);
                    mlx_dir_has_weights(&mlx_dir) || !hf_repo.is_empty()
                } else {
                    let mlx_dir = self.models_dir.join(&schema.name);
                    mlx_dir_has_weights(&mlx_dir)
                };
            }
            #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
            {
                schema.available = false;
            }
        } else if schema.is_vllm_mlx() {
            // vLLM-MLX: available if endpoint env var set or was manually marked available
            schema.available = std::env::var("VLLM_MLX_ENDPOINT").is_ok() || schema.available;
        } else if matches!(schema.source, ModelSource::WhisperCpp { .. }) {
            // whisper.cpp compiles on every platform and lazy-downloads its ggml
            // model on first use, so it's functionally available everywhere
            // (mirrors refresh_availability). Checked before the is_local() GGUF
            // branch below — whisper is_local() too but has no model.gguf.
            schema.available = true;
        } else if matches!(schema.source, ModelSource::WindowsSpeech {}) {
            // OS-provided WinRT synthesizer — available on Windows only.
            schema.available = cfg!(target_os = "windows");
        } else if schema.is_codex_cli() {
            // The CLI owns auth; availability only answers whether there is a
            // binary to attempt. Login failures surface from the child without
            // CAR reading Codex's credential store.
            schema.available = crate::backend::codex_cli::is_available();
        } else if schema.is_local() {
            let local_path = self.models_dir.join(&schema.name).join("model.gguf");
            // Mirrors the `ModelSource::Local` arm in `refresh_availability`:
            // where the Candle GGUF backend exists, a declared `hf_repo` is
            // functionally available because `ensure_local` fetches from it.
            // Registration and refresh must agree, or a model's availability
            // flips depending on which ran last.
            let lazily_fetchable = matches!(
                schema.source,
                ModelSource::Local { ref hf_repo, .. } if !hf_repo.is_empty()
            ) && !cfg!(all(
                target_os = "macos",
                target_arch = "aarch64",
                not(car_skip_mlx)
            ));
            schema.available = local_path.exists() || lazily_fetchable;
        } else if schema.is_remote() {
            // Registration is catalog work, so it may use environment presence
            // and non-secret authority hints but must never query a secret
            // backend. Request-time snapshots refresh this authoritatively.
            schema.available = match schema.source {
                ModelSource::RemoteApi {
                    protocol: crate::schema::ApiProtocol::OpenRouter,
                    ..
                } => crate::openrouter::credential_source().is_some(),
                ModelSource::RemoteApi {
                    ref api_key_env, ..
                } => environment_credential_available(api_key_env),
                ModelSource::Proprietary {
                    ref provider,
                    ref auth,
                    ..
                } => {
                    let resolved = match auth {
                        ProprietaryAuth::ApiKeyEnv { env_var }
                        | ProprietaryAuth::BearerTokenEnv { env_var } => {
                            std::collections::HashMap::from([(
                                env_var.clone(),
                                environment_credential_available(env_var),
                            )])
                        }
                        ProprietaryAuth::OAuth2Pkce { .. } => Default::default(),
                    };
                    proprietary_auth_available(
                        &schema.id,
                        &schema.provider,
                        provider,
                        auth,
                        self.session.available(),
                        &resolved,
                    )
                }
                _ => schema.available,
            };
        }
        // Ready = usable without a download. For anything local that means
        // weights resolvable on disk *now*; `available` alone can't answer this,
        // since a declared `hf_repo` makes an MLX model "available" before a
        // byte is fetched (#164). Remote models never need a weights download.
        // Their live credential state belongs in `available`; coupling it to
        // `weights_ready` permanently excludes models registered before a key
        // is connected from `require_ready` routes.
        // See `ModelSchema::weights_ready` (Parslee-ai/car#638).
        // NOTE: `is_mlx()` must be tested before `is_local()` — `is_local()`
        // returns true for `ModelSource::Mlx` too, so the GGUF branch would
        // otherwise swallow every MLX model and look for a `model.gguf` that
        // never exists.
        schema.weights_ready = physical_weights_ready(&schema, &self.models_dir);
        info!(
            id = %schema.id,
            name = %schema.name,
            available = schema.available,
            weights_ready = schema.weights_ready,
            "registered model"
        );
        self.models.insert(schema.id.clone(), schema);
        true
    }

    /// Register a schema that crossed the user-controlled configuration
    /// boundary.
    ///
    /// This is deliberately separate from [`register`](Self::register):
    /// builtin, signed-catalog, and discovery rows must never become
    /// `models.json` rows merely because they share the same in-memory map.
    pub fn register_user_model(&mut self, mut schema: ModelSchema) {
        if crate::openrouter::is_curated_managed_gateway_alias(&schema.id) {
            warn!(id = %schema.id, "ignoring persisted user model for reserved Parslee-managed alias");
            return;
        }
        if self.project_model_ids.contains(&schema.id) {
            warn!(id = %schema.id, "ignoring persisted user model for project-owned exact id");
            return;
        }
        schema.mark_user_registered();
        let id = schema.id.clone();
        if self.register_preserving_trust(schema) {
            self.on_disk_discovered_ids.remove(&id);
            self.user_config_ids.insert(id);
        }
    }

    /// Unregister a model by id. Returns the removed schema if found.
    pub fn unregister(&mut self, id: &str) -> Option<ModelSchema> {
        self.on_disk_discovered_ids.remove(id);
        let removed = self.models.remove(id);
        if let Some(ref m) = removed {
            info!(id = %m.id, "unregistered model");
        }
        removed
    }

    /// Unregister an explicitly user-configured model.
    ///
    /// An untracked builtin, signed-catalog, or discovery row is not removable
    /// through this persistence boundary.
    pub fn unregister_user_model(&mut self, id: &str) -> Option<ModelSchema> {
        if !self.user_config_ids.remove(id) {
            return None;
        }
        self.unregister(id)
    }

    /// List all models.
    pub fn list(&self) -> Vec<&ModelSchema> {
        let mut models: Vec<&ModelSchema> = self.models.values().collect();
        models.sort_by(|a, b| a.id.cmp(&b.id));
        models
    }

    /// Query models matching a filter.
    pub fn query(&self, filter: &ModelFilter) -> Vec<&ModelSchema> {
        self.models
            .values()
            .filter(|m| {
                // Capability check: model must have ALL required capabilities
                if !filter.capabilities.iter().all(|c| m.has_capability(*c)) {
                    return false;
                }
                // Size check
                if let Some(max) = filter.max_size_mb {
                    if m.size_mb() > max && m.is_local() {
                        return false;
                    }
                }
                // Latency check (declared envelope)
                if let Some(max) = filter.max_latency_ms {
                    if let Some(p50) = m.performance.latency_p50_ms {
                        if p50 > max {
                            return false;
                        }
                    }
                }
                // Cost check
                if let Some(max) = filter.max_cost_per_mtok {
                    if let Some(cost) = m.cost.output_per_mtok {
                        if cost > max {
                            return false;
                        }
                    }
                }
                // Tag check
                if !filter.tags.iter().all(|t| m.tags.contains(t)) {
                    return false;
                }
                // Provider check
                if let Some(ref p) = filter.provider {
                    if &m.provider != p {
                        return false;
                    }
                }
                // Local only
                if filter.local_only && !m.is_local() {
                    return false;
                }
                // Available only
                if filter.available_only && !m.available_now() {
                    return false;
                }
                true
            })
            .collect()
    }

    /// Query models by a single capability.
    pub fn query_by_capability(&self, cap: ModelCapability) -> Vec<&ModelSchema> {
        self.query(&ModelFilter {
            capabilities: vec![cap],
            ..Default::default()
        })
    }

    /// Report installed local models with curated newer replacements.
    pub fn available_upgrades(&self) -> Vec<ModelUpgrade> {
        let mut upgrades = Vec::new();
        for rule in model_upgrade_rules() {
            let Some(from) = rule
                .from_ids
                .iter()
                .find_map(|id| self.models.get(id.as_str()))
                .filter(|schema| schema.available)
            else {
                continue;
            };
            let Some(to) = self.models.get(rule.to_id.as_str()) else {
                continue;
            };
            upgrades.push(ModelUpgrade {
                from_id: from.id.clone(),
                from_name: from.name.clone(),
                to_id: to.id.clone(),
                to_name: to.name.clone(),
                reason: rule.reason.clone(),
                target_runtime: rule.target_runtime.clone(),
                target_runtime_requirement: rule.target_runtime_requirement.clone(),
                minimum_runtimes: rule.minimum_runtimes.clone(),
                target_available: to.available,
                target_pullable: matches!(
                    to.source,
                    ModelSource::Local { .. } | ModelSource::Mlx { .. }
                ),
                remove_old_supported: matches!(
                    from.source,
                    ModelSource::Local { .. } | ModelSource::Mlx { .. }
                ) && rule.remove_old_after_available,
            });
        }
        upgrades.sort_by(|a, b| a.from_id.cmp(&b.from_id).then(a.to_id.cmp(&b.to_id)));
        upgrades.dedup_by(|a, b| a.from_id == b.from_id && a.to_id == b.to_id);
        upgrades
    }

    /// Get a specific model by id.
    pub fn get(&self, id: &str) -> Option<&ModelSchema> {
        self.models.get(id)
    }

    /// Return the currently registered schema without refreshing availability.
    ///
    /// Callers that need runtime credential/file state must explicitly use a
    /// refreshing snapshot; identity/provenance checks should use this exact
    /// stored row instead.
    pub fn registered_schema(&self, id: &str) -> Option<&ModelSchema> {
        self.get(id)
    }

    /// Iterate all registered model schemas (built-in + signed + discovered).
    pub fn all(&self) -> impl Iterator<Item = &ModelSchema> {
        self.models.values()
    }

    /// Find a model by name (case-insensitive). For backward compatibility
    /// with the old registry that used short names like "Qwen3-4B".
    pub fn find_by_name(&self, name: &str) -> Option<&ModelSchema> {
        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        if !name.to_ascii_lowercase().ends_with("-mlx") {
            if let Some(mlx_variant) = self
                .models
                .values()
                .find(|m| m.name.eq_ignore_ascii_case(&format!("{name}-MLX")))
            {
                return Some(mlx_variant);
            }
        }

        self.models
            .values()
            .find(|m| m.name.eq_ignore_ascii_case(name))
    }

    /// On Apple Silicon, resolve a GGUF/Candle model to its MLX equivalent.
    /// Returns the MLX model schema if one exists with the same family and
    /// matching capabilities; otherwise returns None.
    #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
    pub fn resolve_mlx_equivalent(&self, schema: &ModelSchema) -> Option<&ModelSchema> {
        // Already MLX — no redirect needed.
        if schema.is_mlx() || schema.is_vllm_mlx() {
            return None;
        }
        // Only redirect local GGUF models.
        if !matches!(schema.source, ModelSource::Local { .. }) {
            return None;
        }
        // Find the MLX twin: same family, SAME PARAMETER COUNT, and at least
        // the same primary capability. Family alone is too coarse — every
        // Qwen3 size shares family "qwen3", so a family-only `.find()` could
        // map an 8B GGUF to a 4B MLX model (whichever the HashMap yielded
        // first), silently swapping model size at execution and routing.
        // Keying on param_count makes the resolution deterministic and 1:1.
        let primary_cap = schema.capabilities.first()?;
        self.models.values().find(|m| {
            m.is_mlx()
                && m.family == schema.family
                && m.param_count == schema.param_count
                && m.capabilities.contains(primary_cap)
        })
    }

    /// Ensure a local model is downloaded, returning its local directory path.
    pub async fn ensure_local(&self, id: &str) -> Result<PathBuf, InferenceError> {
        // Report to the ambient sink rather than dropping the events on the
        // floor — see [`Self::ambient_progress`] (Parslee-ai/car#620). Still a
        // no-op sink unless an embedder opted in, so this is not a behaviour
        // change for anyone who hasn't.
        let sink = self.ambient_progress.clone();
        self.ensure_local_with_progress(id, &sink).await
    }

    /// Route implicit-download progress to `sink`.
    ///
    /// Applies to acquisitions nobody explicitly requested — the ones
    /// [`ensure_local`](Self::ensure_local) performs mid-`generate` when the
    /// router lands on a model that isn't on disk. Explicit pulls take their
    /// sink as an argument and ignore this.
    pub fn set_ambient_progress(&mut self, sink: ProgressSink) {
        self.ambient_progress = sink;
    }

    /// Like [`ensure_local`](Self::ensure_local) but drives a progress sink and
    /// enforces the acquisition lifecycle: a per-model lock (so this can't race
    /// a concurrent pull / remove / upgrade of the same model), a disk-space
    /// preflight, and `Started`/`Completed`/`Failed` events around the work.
    /// When every required local file is already complete, it returns that path
    /// under the lock without a preflight or acquisition events.
    pub async fn ensure_local_with_progress(
        &self,
        id: &str,
        sink: &ProgressSink,
    ) -> Result<PathBuf, InferenceError> {
        self.acquire_and_ensure(id, sink, false, None).await
    }

    /// Download into a caller-created private staging directory. The caller
    /// atomically publishes that directory only after the complete acquisition
    /// succeeds, so a pre-existing same-name directory can never be promoted
    /// into CAR ownership by a pull.
    pub(crate) async fn ensure_local_with_progress_staged(
        &self,
        id: &str,
        sink: &ProgressSink,
        staging_dir: &Path,
    ) -> Result<PathBuf, InferenceError> {
        self.acquire_and_ensure(id, sink, false, Some(staging_dir))
            .await
    }

    /// Force a complete re-acquisition: skip the "reuse what's already on disk"
    /// short-circuits so every expected file is re-checked and any missing one
    /// (e.g. a blob the self-heal path purged as provably corrupt) is
    /// re-downloaded. The per-file checks still skip intact files, so this
    /// re-fetches only what's actually gone — not the whole model.
    pub async fn redownload_local(&self, id: &str) -> Result<PathBuf, InferenceError> {
        self.acquire_and_ensure(id, &ProgressSink::none(), true, None)
            .await
    }

    async fn acquire_and_ensure(
        &self,
        id: &str,
        sink: &ProgressSink,
        force: bool,
        managed_dir_override: Option<&Path>,
    ) -> Result<PathBuf, InferenceError> {
        let schema = self
            .get(id)
            .or_else(|| self.find_by_name(id))
            .ok_or_else(|| InferenceError::ModelNotFound(id.to_string()))?;
        let model_name = schema.name.clone();
        let model_id = schema.id.clone();
        let needed_mb = schema.size_mb();
        let model_dir = self.models_dir.join(&schema.name);

        // Serialize acquisition of this model id against other pull/remove tasks.
        let _guard = crate::download::acquire_model_lock(&model_id).await;

        // Reusing complete bytes touches neither disk nor the network, so it
        // must precede the size-plus-reserve preflight and acquisition events.
        // Forced repairs and staged pulls deliberately keep the full lifecycle.
        if !force && managed_dir_override.is_none() {
            if let Some(path) = self.try_reuse_local(schema).await? {
                return Ok(path);
            }
        }

        // Preflight: fail fast if there isn't room, before touching disk.
        if let Err(e) = crate::download::check_disk_space(&model_dir, needed_mb) {
            sink.emit(DownloadEvent::Failed { error: e.clone() });
            return Err(InferenceError::DownloadFailed(e));
        }

        sink.emit(DownloadEvent::Started {
            model: model_name.clone(),
            total_files: 0,
            total_mb: needed_mb,
        });
        let result = self
            .ensure_local_inner(id, sink, force, managed_dir_override)
            .await;
        match &result {
            Ok(_) => sink.emit(DownloadEvent::Completed { model: model_name }),
            Err(e) => sink.emit(DownloadEvent::Failed {
                error: e.to_string(),
            }),
        }
        result
    }

    async fn try_reuse_local(
        &self,
        schema: &ModelSchema,
    ) -> Result<Option<PathBuf>, InferenceError> {
        match &schema.source {
            ModelSource::Local { .. } => {
                let model_dir = self.models_dir.join(&schema.name);
                let model_path = model_dir.join("model.gguf");
                let tokenizer_path = model_dir.join("tokenizer.json");

                if crate::download::cache_file_usable(&model_path)
                    && crate::download::cache_file_usable(&tokenizer_path)
                {
                    return Ok(Some(model_dir));
                }
            }
            ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
                let model_dir = self.models_dir.join(&schema.name);
                let config_path = model_dir.join("config.json");
                let is_diffusers = schema.capabilities.iter().any(|c| {
                    matches!(
                        c,
                        ModelCapability::ImageGeneration | ModelCapability::VideoGeneration
                    )
                });

                if mlx_dir_has_weights(&model_dir) && (is_diffusers || config_path.exists()) {
                    if auxiliary_mlx_files_missing(&schema.name, hf_repo, &model_dir) {
                        // Preserve managed-directory precedence: the inner path
                        // repairs this directory before considering a shared HF
                        // snapshot, just as it did before reuse moved earlier.
                        return Ok(None);
                    }
                    info!(model = %schema.name, path = %model_dir.display(), "using managed local MLX model");
                    return Ok(Some(model_dir));
                }

                if let Some(snapshot_dir) =
                    latest_huggingface_repo_snapshot(hf_repo).filter(|d| mlx_dir_has_weights(d))
                {
                    if !auxiliary_mlx_files_missing(&schema.name, hf_repo, &snapshot_dir) {
                        info!(model = %schema.name, path = %snapshot_dir.display(), "using cached MLX snapshot");
                        return Ok(Some(snapshot_dir));
                    }
                }
            }
            _ => {}
        }

        Ok(None)
    }

    async fn ensure_local_inner(
        &self,
        id: &str,
        sink: &ProgressSink,
        force: bool,
        managed_dir_override: Option<&Path>,
    ) -> Result<PathBuf, InferenceError> {
        let schema = self
            .get(id)
            .or_else(|| self.find_by_name(id))
            .ok_or_else(|| InferenceError::ModelNotFound(id.to_string()))?;

        match &schema.source {
            ModelSource::Local {
                hf_repo,
                hf_filename,
                tokenizer_repo,
            } => {
                let model_dir = managed_dir_override
                    .map(Path::to_path_buf)
                    .unwrap_or_else(|| self.models_dir.join(&schema.name));
                let model_path = model_dir.join("model.gguf");
                let tokenizer_path = model_dir.join("tokenizer.json");

                if !force
                    && crate::download::cache_file_usable(&model_path)
                    && crate::download::cache_file_usable(&tokenizer_path)
                {
                    return Ok(model_dir);
                }

                // A row with no `hf_repo` came from the local-directory scan:
                // its weights are whatever the user put on disk, and there is
                // nowhere to fetch the missing pieces from. Downloading with an
                // empty repo asks HuggingFace for `https://huggingface.co//…`
                // and reports whatever that returns, which explains nothing.
                // Say what CAR actually needs instead.
                if hf_repo.is_empty() {
                    let missing = [
                        ("model.gguf", &model_path),
                        ("tokenizer.json", &tokenizer_path),
                    ]
                    .into_iter()
                    .filter(|(_, path)| !crate::download::cache_file_usable(path))
                    .map(|(name, _)| name)
                    .collect::<Vec<_>>()
                    .join(" and ");
                    return Err(InferenceError::InferenceFailed(format!(
                        "{}: discovered on disk at {} but not loadable — the GGUF \
                         backend reads `model.gguf` and `tokenizer.json` from the \
                         model directory, and this one is missing {missing}. Rename \
                         the weight file to `model.gguf` and add the tokenizer, or \
                         register the model against its HuggingFace repo so CAR can \
                         fetch both.",
                        schema.name,
                        model_dir.display()
                    )));
                }

                std::fs::create_dir_all(&model_dir)?;

                if !crate::download::cache_file_usable(&model_path) {
                    info!(model = %schema.name, repo = %hf_repo, "downloading model weights");
                    sink.emit(DownloadEvent::FileStarted {
                        filename: "model weights".into(),
                        index: 1,
                        total_files: 2,
                        size_mb: schema.size_mb(),
                    });
                    download_file(hf_repo, hf_filename, &model_path).await?;
                    sink.emit(DownloadEvent::FileCompleted {
                        filename: "model weights".into(),
                    });
                }
                if !crate::download::cache_file_usable(&tokenizer_path) {
                    info!(model = %schema.name, repo = %tokenizer_repo, "downloading tokenizer");
                    sink.emit(DownloadEvent::FileStarted {
                        filename: "tokenizer".into(),
                        index: 2,
                        total_files: 2,
                        size_mb: 0,
                    });
                    download_file(tokenizer_repo, "tokenizer.json", &tokenizer_path).await?;
                    sink.emit(DownloadEvent::FileCompleted {
                        filename: "tokenizer".into(),
                    });
                }

                Ok(model_dir)
            }
            ModelSource::Mlx {
                hf_repo,
                hf_weight_file,
            }
            | ModelSource::ManagedVllmMlx {
                hf_repo,
                hf_weight_file,
            } => {
                let model_dir = managed_dir_override
                    .map(Path::to_path_buf)
                    .unwrap_or_else(|| self.models_dir.join(&schema.name));
                let config_path = model_dir.join("config.json");

                // Diffusers-layout models (Flux image-gen, LTX video) keep their
                // weights in component subdirs (transformer/, vae/, text_encoder/,
                // tokenizer/) and ship NO root config.json / top-level weight
                // index. The standard MLX flow below hard-fetches config.json
                // (404 for these repos) and a weight index (absent), so they need
                // the whole-repo snapshot path. Detect by capability so DOWNLOAD,
                // REUSE, and (via mlx_dir_has_weights) LOAD all agree on the
                // layout — the recurring image-gen 404 was fixing only the load
                // path (recurse subdirs) while download + reuse still demanded a
                // config.json these repos don't have.
                let is_diffusers = schema.capabilities.iter().any(|c| {
                    matches!(
                        c,
                        ModelCapability::ImageGeneration | ModelCapability::VideoGeneration
                    )
                });

                // Reuse the managed dir ONLY if it actually holds weights — a
                // config-only stub (or dangling-symlink install) must fall
                // through to the download below, not no-op (car-releases#391).
                // `force` (self-heal re-pull) skips reuse so a purged shard is
                // actually re-fetched rather than masked by the surviving shards.
                // Standard MLX also gates on the root config.json; diffusers has
                // none, so recursive weights-present is its completeness signal.
                if !force
                    && mlx_dir_has_weights(&model_dir)
                    && (is_diffusers || config_path.exists())
                {
                    ensure_auxiliary_mlx_files(&schema.name, hf_repo, &model_dir).await?;
                    info!(model = %schema.name, path = %model_dir.display(), "using managed local MLX model");
                    return Ok(model_dir);
                }

                // Same guard for a cached HF snapshot: a partial/pruned snapshot
                // whose weights don't resolve isn't usable — re-download instead.
                if !force {
                    if let Some(snapshot_dir) =
                        latest_huggingface_repo_snapshot(hf_repo).filter(|d| mlx_dir_has_weights(d))
                    {
                        ensure_auxiliary_mlx_files(&schema.name, hf_repo, &snapshot_dir).await?;
                        info!(model = %schema.name, path = %snapshot_dir.display(), "using cached MLX snapshot");
                        return Ok(snapshot_dir);
                    }
                }

                std::fs::create_dir_all(&model_dir)?;

                info!(model = %schema.name, repo = %hf_repo, "downloading MLX model");

                // Diffusers-layout: mirror the whole repo (component subdirs
                // preserved). There is no root config.json or weight index to
                // drive the file-by-file flow below, and demanding one is the
                // 404 that broke image/video generation. Verify component weights
                // actually landed so a half-finished pull can't masquerade as
                // installed (the false-availability the caller relies on).
                if is_diffusers {
                    download_repo_snapshot(hf_repo, &model_dir, sink).await?;
                    ensure_auxiliary_mlx_files(&schema.name, hf_repo, &model_dir).await?;
                    if !mlx_dir_has_weights(&model_dir) {
                        return Err(InferenceError::DownloadFailed(format!(
                            "{hf_repo}: snapshot fetched but no component weights found"
                        )));
                    }
                    info!(model = %schema.name, path = %model_dir.display(), "downloaded diffusers model");
                    return Ok(model_dir);
                }

                // Download config, tokenizer, and weight files. total_files is
                // not known up front for MLX (single vs sharded), so events
                // carry total_files = 0 ("unknown") and the UI shows names.
                emit_file(sink, "config", 0, schema.size_mb());
                download_file(hf_repo, "config.json", &config_path).await?;
                download_tokenizer_assets(hf_repo, &model_dir, sink).await;
                let tok_config_path = model_dir.join("tokenizer_config.json");
                if !crate::download::cache_file_usable(&tok_config_path) {
                    let _ = download_file(hf_repo, "tokenizer_config.json", &tok_config_path).await;
                }

                // Download weight files
                if let Some(ref wf) = hf_weight_file {
                    let wf_path = model_dir.join(wf);
                    if !crate::download::cache_file_usable(&wf_path) {
                        emit_file(sink, "model weights", 0, schema.size_mb());
                        download_file(hf_repo, wf, &wf_path).await?;
                    }
                } else {
                    // Try single file first, then sharded
                    let single = model_dir.join("model.safetensors");
                    if !crate::download::cache_file_usable(&single) {
                        emit_file(sink, "model weights", 0, schema.size_mb());
                        match download_file(hf_repo, "model.safetensors", &single).await {
                            Ok(()) => {}
                            Err(_) => {
                                // Sharded: download index and then each shard
                                let index_path = model_dir.join("model.safetensors.index.json");
                                download_file(hf_repo, "model.safetensors.index.json", &index_path)
                                    .await?;

                                let index_json: serde_json::Value =
                                    serde_json::from_str(&std::fs::read_to_string(&index_path)?)
                                        .map_err(|e| {
                                            InferenceError::InferenceFailed(format!(
                                                "parse index: {e}"
                                            ))
                                        })?;

                                if let Some(weight_map) =
                                    index_json.get("weight_map").and_then(|m| m.as_object())
                                {
                                    let mut files: std::collections::HashSet<String> =
                                        std::collections::HashSet::new();
                                    for filename in weight_map.values() {
                                        if let Some(f) = filename.as_str() {
                                            files.insert(f.to_string());
                                        }
                                    }
                                    let shard_total = files.len() as u32;
                                    for (i, file) in files.iter().enumerate() {
                                        let dest = model_dir.join(file);
                                        if !crate::download::cache_file_usable(&dest) {
                                            info!(file = %file, "downloading weight shard");
                                            sink.emit(DownloadEvent::FileStarted {
                                                filename: format!("weights part {}", i + 1),
                                                index: (i + 1) as u32,
                                                total_files: shard_total,
                                                size_mb: 0,
                                            });
                                            download_file(hf_repo, file, &dest).await?;
                                            sink.emit(DownloadEvent::FileCompleted {
                                                filename: format!("weights part {}", i + 1),
                                            });
                                        }
                                    }
                                }
                            }
                        }
                    }
                }

                ensure_auxiliary_mlx_files(&schema.name, hf_repo, &model_dir).await?;

                // Verify what we actually landed, exactly as the diffusers branch
                // above does (#808). Without this, a pull whose shard fetch was
                // interrupted returned `Ok(model_dir)` — indistinguishable from
                // success — and the model stayed broken until inference failed
                // with `load model-00001-of-00002.safetensors: Path must point to
                // a local file`, inside whatever job was running.
                //
                // A caller reaching for the two obvious mechanisms — check
                // availability, then pull — got "yes" and "ok" from both while
                // the model was unusable. Report the missing shards by name so
                // the failure is actionable at pull time.
                let missing = missing_weight_shards(&model_dir);
                if !missing.is_empty() {
                    return Err(InferenceError::DownloadFailed(format!(
                        "{}: pull finished but {} weight shard(s) are still missing: {}. \
                         The download was interrupted; re-run the pull to resume it.",
                        schema.name,
                        missing.len(),
                        missing.join(", ")
                    )));
                }
                if !mlx_dir_has_weights(&model_dir) {
                    return Err(InferenceError::DownloadFailed(format!(
                        "{}: pull finished but no usable weights are present under {}",
                        schema.name,
                        model_dir.display()
                    )));
                }
                Ok(model_dir)
            }
            _ => Err(InferenceError::InferenceFailed(format!(
                "model {} is not local",
                id
            ))),
        }
    }

    /// Legacy registry-local deletion is intentionally disabled because it
    /// recursively removed shared Hugging Face caches without receipts or
    /// runtime/lease coordination. Use `InferenceEngine::remove_model_from_car`.
    #[deprecated(note = "use InferenceEngine::remove_model_from_car")]
    pub fn remove_local(&mut self, id: &str) -> Result<(), InferenceError> {
        Err(InferenceError::InferenceFailed(format!(
            "legacy registry removal for {id} is disabled; use receipt-backed model management"
        )))
    }

    /// Refresh passive catalog availability flags for all models.
    ///
    /// Physical local/runtime probes remain live, but credential-backed rows
    /// use only environment presence and the non-secret Parslee authority hint.
    /// This method is safe for startup, catalog, setup, and health surfaces.
    pub fn refresh_availability(&mut self) {
        let parslee_oauth_available = self.session.available();
        self.refresh_availability_with(
            parslee_oauth_available,
            self.session.signed_out() && self.session.may_forget_session_evidence(),
            false,
        );
    }

    /// Refresh availability for an explicit request-time routing/use path.
    /// `parslee_api_base` comes from Task 2's coordinator-owned credential
    /// snapshot; `parslee_signed_out` is true only for an authoritative absence,
    /// never for denial/cooldown/unreadable failures.
    pub(crate) fn refresh_routing_availability(
        &mut self,
        parslee_api_base: Option<&str>,
        parslee_signed_out: bool,
    ) {
        if let Some(api_base) = parslee_api_base {
            let api_base = api_base.trim_end_matches('/');
            for schema in self.models.values_mut() {
                if schema.provider.eq_ignore_ascii_case("parslee") {
                    if let ModelSource::Proprietary {
                        provider, endpoint, ..
                    } = &mut schema.source
                    {
                        if provider.eq_ignore_ascii_case("parslee") {
                            *endpoint = api_base.to_string();
                        }
                    }
                }
            }
        }
        self.refresh_availability_with(parslee_api_base.is_some(), parslee_signed_out, true);
    }

    fn refresh_availability_with(
        &mut self,
        parslee_oauth_available: bool,
        clear_parslee_observations: bool,
        authoritative_credentials: bool,
    ) {
        // `models_dir` is consumed only inside the MLX and Local arms.
        // On non-MLX targets the MLX arm is a cfg-gated `available = false`
        // and never touches `models_dir`; the Local arm still needs it.
        let models_dir = self.models_dir.clone();
        // mlx-vlm CLI is the same probe call no matter which model
        // requires it; do it once per refresh, not per-model. On
        // non-MLX targets the variable is unused (the consuming arm
        // is cfg-gated out), so suppress the unused-variable warning.
        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        let mlx_vlm_cli_present = crate::backend::mlx_vlm_cli::is_available();
        #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
        #[allow(unused_variables)]
        let mlx_vlm_cli_present = false;
        // A learned "gateway has no OpenRouter upstream" is about the
        // environment behind THIS session. When the session is gone, so is the
        // evidence — otherwise a sign-in to a properly-configured org would
        // inherit the previous one's suppression (Parslee-ai/car#786).
        if clear_parslee_observations {
            crate::openrouter::clear_gateway_unconfigured();
            // Same reasoning for the credential verdict: it was learned about
            // the credential that is now gone, and inheriting it would suppress
            // the managed lanes through the next sign-in (Parslee-ai/car#887).
            crate::parslee_credential::clear_credential_rejected();
        }

        // Resolve each DISTINCT credential once, not once per model
        // (car-releases#75).
        //
        // Every `RemoteApi` and `Proprietary` row used to call
        // `resolve_env_or_keychain` inside the loop below, and on macOS that is
        // a keychain query — ~14 ms each. A stock catalog is ~72 models over
        // roughly a dozen providers, so a refresh spent ~1 s asking the keychain
        // the same handful of questions dozens of times. `routing_registry_
        // snapshot` runs a refresh per request AND `estimated_tokens` runs
        // another, so a single delegated call paid it twice: ~2 s before the
        // request reached the runner, which is the reported latency.
        //
        // Deduplicating changes no semantics. Within ONE refresh the same env
        // var must yield the same answer — reading it 30 times cannot be more
        // correct than reading it once. Credentials are still re-read on every
        // refresh, so the "pasted/OAuth key changes take effect without a
        // restart" property above is untouched.
        let mut credential_envs: std::collections::BTreeSet<String> = Default::default();
        let mut needs_openrouter = false;
        for m in self.models.values() {
            match &m.source {
                ModelSource::RemoteApi {
                    protocol: crate::schema::ApiProtocol::OpenRouter,
                    ..
                } => needs_openrouter = true,
                ModelSource::RemoteApi { api_key_env, .. } => {
                    credential_envs.insert(api_key_env.clone());
                }
                ModelSource::Proprietary { auth, .. } => match auth {
                    ProprietaryAuth::ApiKeyEnv { env_var }
                    | ProprietaryAuth::BearerTokenEnv { env_var } => {
                        credential_envs.insert(env_var.clone());
                    }
                    ProprietaryAuth::OAuth2Pkce { .. } => {}
                },
                _ => {}
            }
        }
        let credential_available: std::collections::HashMap<String, bool> = credential_envs
            .into_iter()
            .map(|env| {
                let available = if authoritative_credentials {
                    car_secrets::resolve_env_or_keychain(&env).is_some()
                } else {
                    environment_credential_available(&env)
                };
                (env, available)
            })
            .collect();
        // Only probe OpenRouter when a row actually needs it — an unnecessary
        // probe is the same class of waste this block exists to remove.
        let openrouter_available = needs_openrouter
            && if authoritative_credentials {
                crate::openrouter::refresh_credential_source().is_some()
            } else {
                crate::openrouter::credential_source().is_some()
            };

        for m in self.models.values_mut() {
            match &m.source {
                ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
                    // MLX requires Apple Silicon Metal. On any other
                    // build target — Intel Mac, Linux, Windows, or
                    // `car_skip_mlx` — the backend is not compiled in
                    // and the adaptive router must not select these
                    // models. Cfg-gate identical to the
                    // AppleFoundationModels branch below and the
                    // `register()` MLX branch above. Closes the §7.1
                    // arm of Parslee-ai/car#231.
                    #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
                    {
                        // Models tagged `requires-mlx-vlm` shell out to the
                        // mlx_vlm Python CLI for image inference (#115).
                        // If the CLI isn't on PATH, the runtime reaches it
                        // anyway and bails — the registry MUST reflect that
                        // by marking such entries unavailable until the
                        // user installs `uv tool install mlx-vlm`. #137.
                        let needs_mlx_vlm = m.tags.iter().any(|t| t == "requires-mlx-vlm");

                        m.available = if needs_mlx_vlm {
                            mlx_vlm_cli_present
                        } else if m.tags.contains(&"speech".to_string()) {
                            speech_mlx_available()
                        } else {
                            // Available if cached locally OR has an hf_repo —
                            // the native MLX path's ensure_local() lazy-
                            // downloads on first use, so a declared hf_repo
                            // is "functionally available" the same way
                            // Ollama and RemoteApi entries are ("should
                            // work in principle" not "physically cached").
                            // Closes #164: mlx/ltx-2.3:q4 reported
                            // unavailable even though `car video` would
                            // just download-and-run successfully.
                            let mlx_dir = models_dir.join(&m.name);
                            mlx_dir_has_weights(&mlx_dir) || !hf_repo.is_empty()
                        };
                    }
                    #[cfg(not(all(
                        target_os = "macos",
                        target_arch = "aarch64",
                        not(car_skip_mlx)
                    )))]
                    {
                        let _ = hf_repo; // unused on non-MLX targets
                        m.available = false;
                    }
                }
                ModelSource::Local {
                    hf_repo: local_repo,
                    ..
                } => {
                    let local_path = models_dir.join(&m.name).join("model.gguf");
                    // Where the Candle GGUF backend exists — every target that
                    // is not Apple-Silicon-with-MLX, i.e. the CUDA machines
                    // where GGUF *is* the local path — a declared `hf_repo` is
                    // functionally available, because `ensure_local` fetches
                    // `model.gguf` and `tokenizer.json` from it on first use.
                    // Same rule #164 gave the MLX path and whisper.cpp already
                    // has; this branch was left behind, so a fresh CUDA install
                    // reported every local model unavailable, the router
                    // dropped them from fallback chains, and the user was sent
                    // to cloud with a GPU sitting idle one download away.
                    //
                    // An empty `hf_repo` means a row from the local-directory
                    // scan, which has nowhere to fetch from and stays gated on
                    // the file actually being there.
                    #[cfg(not(all(
                        target_os = "macos",
                        target_arch = "aarch64",
                        not(car_skip_mlx)
                    )))]
                    {
                        m.available = local_path.exists() || !local_repo.is_empty();
                    }
                    // On Apple Silicon `backend::candle` is not compiled at
                    // all: a GGUF row runs only by `resolve_mlx_equivalent`
                    // substitution, so it stays gated on physical presence
                    // rather than claiming a lazy download that has no loader.
                    #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
                    {
                        let _ = local_repo;
                        m.available = local_path.exists();
                    }
                }
                ModelSource::WhisperCpp { .. } => {
                    // whisper.cpp compiles on every platform and lazy-downloads
                    // its ggml model on first use (car-whisper), so it is
                    // functionally available everywhere — the cross-platform
                    // on-device STT the MLX speech models can't be off Apple.
                    m.available = true;
                }
                ModelSource::WindowsSpeech {} => {
                    // OS-provided WinRT synthesizer — available on Windows only
                    // (like AppleFoundationModels is Apple-only).
                    #[cfg(target_os = "windows")]
                    {
                        m.available = true;
                    }
                    #[cfg(not(target_os = "windows"))]
                    {
                        m.available = false;
                    }
                }
                ModelSource::RemoteApi {
                    protocol: crate::schema::ApiProtocol::OpenRouter,
                    ..
                } => {
                    m.available = openrouter_available;
                }
                ModelSource::RemoteApi { api_key_env, .. } => {
                    // env OR keychain — see `register`. Resolved once per
                    // refresh above, not once per model (car-releases#75).
                    m.available = credential_available
                        .get(api_key_env)
                        .copied()
                        .unwrap_or(false);
                }
                ModelSource::CodexCli { .. } => {
                    m.available = crate::backend::codex_cli::is_available();
                }
                ModelSource::Ollama { .. } => {
                    // Assume available; health check is async and done lazily
                    m.available = true;
                }
                ModelSource::VllmMlx { .. } => {
                    // External endpoints remain server-provided even on
                    // loopback. CAR's managed runtime says nothing about the
                    // health or ownership of this separately registered server.
                    m.available = std::env::var("VLLM_MLX_ENDPOINT").is_ok() || m.available;
                }
                ModelSource::Proprietary { provider, auth, .. } => {
                    m.available = proprietary_auth_available(
                        &m.id,
                        &m.provider,
                        provider,
                        auth,
                        parslee_oauth_available,
                        &credential_available,
                    );
                }
                ModelSource::AppleFoundationModels { .. } => {
                    // Apple Silicon macOS 26+ AND iOS 26+ both expose
                    // the FoundationModels framework. The shim's
                    // runtime probe handles per-device availability
                    // (Apple Intelligence may be off, the device may
                    // be pre-A17, etc.); cfg-gating here just hides
                    // the call on targets where the shim isn't built.
                    #[cfg(any(
                        all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
                        all(target_os = "ios", target_arch = "aarch64")
                    ))]
                    {
                        m.available = crate::backend::foundation_models::is_available();
                        // Parallel tool calls are a HOST property here, not a
                        // catalog one: the same shim yields one captured call
                        // per turn on macOS 26 and every proposed call on 27,
                        // because the framework's post-throw batch behaviour
                        // differs between them. The catalog literal stays
                        // conservative and the row is upgraded only where the
                        // behaviour was measured, so a host that cannot do it
                        // never advertises it.
                        //
                        // Precedent: `available` on the line above is already
                        // host-dependent at this same hook, so a capability
                        // that varies by host is not a new shape for this row.
                        if crate::backend::foundation_models::supports_parallel_tool_calls()
                            && !m.capabilities.contains(&ModelCapability::MultiToolCall)
                        {
                            m.capabilities.push(ModelCapability::MultiToolCall);
                        }
                        // Vision, same shape, one extra constraint: the claim is
                        // only made where the request can actually be SERVED.
                        // `supports_vision()` reads the model's own
                        // `LanguageModelCapabilities`, and the only path that
                        // consumes images — `generate_with_images` — is gated on
                        // the same probe. Claiming vision without that path is
                        // the 2b bug again: a route the router picks and the
                        // backend then fails at generation time.
                        if crate::backend::foundation_models::supports_vision()
                            && !m.capabilities.contains(&ModelCapability::Vision)
                        {
                            m.capabilities.push(ModelCapability::Vision);
                        }
                        // Prefer the framework's own answer over the catalog
                        // literal. `contextSize` exists from 26.4 on, so this
                        // is None on 26.0-26.3 and the declared value stands.
                        // The literal over-claiming the window is not cosmetic:
                        // `build_context_for_model` sizes the assembly budget
                        // from it, so a wrong number fails at generation time
                        // with the assembly already paid for.
                        if let Some(window) = crate::backend::foundation_models::context_size() {
                            m.context_length = window as usize;
                        }
                    }
                    #[cfg(not(any(
                        all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
                        all(target_os = "ios", target_arch = "aarch64")
                    )))]
                    {
                        m.available = false;
                    }
                }
                ModelSource::Delegated { .. } => {
                    // Availability tracks whether a runner is registered.
                    // Hosts call `registerInferenceRunner` (or its
                    // language equivalent) at startup; until then the
                    // model is unavailable.
                    m.available = crate::runner::current_inference_runner().is_some();
                }
            }
            m.weights_ready = physical_weights_ready(m, &models_dir);
        }
    }

    /// Persist only explicitly user-registered models to disk.
    pub fn save_user_config(&self) -> Result<(), InferenceError> {
        let mut user_models: Vec<ModelSchema> = self
            .user_config_ids
            .iter()
            .filter_map(|id| self.models.get(id))
            .cloned()
            .map(|mut model| {
                // `models.json` is a user-controlled trust boundary regardless
                // of how the in-memory row was originally registered.
                model.mark_user_registered();
                model
            })
            .collect();
        user_models.sort_by(|a, b| a.id.cmp(&b.id));

        for model in &user_models {
            crate::catalog_identity::row_digest(model).map_err(|error| {
                InferenceError::InferenceFailed(format!(
                    "refuse to persist model without canonical catalog identity: {error}"
                ))
            })?;
        }

        let json = serde_json::to_string_pretty(&user_models)
            .map_err(|e| InferenceError::InferenceFailed(format!("serialize: {e}")))?;
        std::fs::write(&self.user_config_path, json)?;
        Ok(())
    }

    /// Load user-registered models from disk.
    pub fn load_user_config(&mut self) -> Result<(), InferenceError> {
        if !self.user_config_path.exists() {
            return Ok(());
        }

        let json = std::fs::read_to_string(&self.user_config_path)?;
        let models: Vec<ModelSchema> = serde_json::from_str(&json)
            .map_err(|e| InferenceError::InferenceFailed(format!("parse models.json: {e}")))?;

        for m in models {
            // Legacy rows omit `trust_tier` and serde must keep accepting that
            // shape, but a manual config can never confer project curation.
            self.register_user_model(m);
        }
        Ok(())
    }

    /// Get the models directory path.
    pub fn models_dir(&self) -> &Path {
        &self.models_dir
    }

    /// Whether invoking this model can start without CAR-managed model setup.
    ///
    /// `available` deliberately includes lazy-downloadable local models so
    /// normal CLI inference can pull on first use. Interactive host actions
    /// need the stricter answer: avoid starting a large download from a
    /// ready-to-use assistant button and point the user at Models setup
    /// instead.
    pub fn ready_without_download(&self, id: &str) -> Option<bool> {
        let schema = self.get(id).or_else(|| self.find_by_name(id))?;
        Some(match &schema.source {
            ModelSource::Local { .. } => {
                let model_dir = self.models_dir.join(&schema.name);
                crate::download::cache_file_usable(&model_dir.join("model.gguf"))
                    && crate::download::cache_file_usable(&model_dir.join("tokenizer.json"))
            }
            ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
                let managed_dir = self.models_dir.join(&schema.name);
                let managed_ready = mlx_snapshot_complete(schema, &managed_dir)
                    && crate::download::cache_file_usable(&managed_dir.join("tokenizer.json"));
                let snapshot_ready =
                    latest_huggingface_repo_snapshot(hf_repo).is_some_and(|snapshot| {
                        mlx_snapshot_complete(schema, &snapshot)
                            && crate::download::cache_file_usable(&snapshot.join("tokenizer.json"))
                    });
                managed_ready || snapshot_ready
            }
            ModelSource::WindowsSpeech {} => true, // OS-provided; nothing to download
            ModelSource::WhisperCpp { model } => {
                // "ready without download" = the ggml file is physically cached
                // at ~/.tokhn/whisper/. (Availability is looser — whisper always
                // lazy-downloads — but this is the strict host-action answer.)
                car_whisper::model_cached(model)
            }
            ModelSource::RemoteApi { .. }
            | ModelSource::CodexCli { .. }
            | ModelSource::Ollama { .. }
            | ModelSource::VllmMlx { .. }
            | ModelSource::AppleFoundationModels { .. }
            | ModelSource::Proprietary { .. }
            | ModelSource::Delegated { .. } => true,
        })
    }

    /// Resolve a usable local artifact without downloading anything. This is
    /// the only source path accepted by explicit model adoption; callers do
    /// not get to supply a filesystem path.
    pub fn existing_local_artifact(&self, id: &str) -> Option<PathBuf> {
        let schema = self.get(id).or_else(|| self.find_by_name(id))?;
        let managed = self.models_dir.join(&schema.name);
        if std::fs::symlink_metadata(&managed).is_ok()
            && self.ready_without_download(&schema.id) == Some(true)
        {
            return Some(managed);
        }
        match &schema.source {
            ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
                latest_huggingface_repo_snapshot(hf_repo).filter(|snapshot| {
                    mlx_snapshot_complete(schema, snapshot)
                        && crate::download::cache_file_usable(&snapshot.join("tokenizer.json"))
                })
            }
            _ => None,
        }
    }

    /// Load the built-in Qwen3 catalog as ModelSchema objects.
    fn load_builtin_catalog(&mut self) {
        for schema in builtin_catalog() {
            let id = schema.id.clone();
            if self.register_project_model(schema) {
                self.builtin_model_ids.insert(id);
            }
        }
    }
}

/// Build a `ModelSchema` for an uncatalogued local model directory, or
/// `None` if the directory isn't a recognized text-LLM checkpoint.
///
/// Recognizes two layouts: MLX (a `config.json` with a known causal-LM
/// `model_type` plus safetensors weights) and GGUF (a `*.gguf` weight
/// file). Capabilities are inferred conservatively from the directory
/// name — anything not positively identified as a text LLM is skipped so
/// speech/vision/image/video checkpoints are never mistaken for
/// generators. car-releases#62.
fn synthesize_local_schema(name: &str, dir: &Path) -> Option<ModelSchema> {
    let lower = name.to_ascii_lowercase();

    // Hard skip directory names that are unambiguously non-text models —
    // these would be catastrophic to route as `Generate`.
    const NON_TEXT_HINTS: &[&str] = &[
        "vad",
        "whisper",
        "parakeet",
        "kokoro",
        "tts",
        "stt",
        "flux",
        "ltx",
        "yume",
        "sd-",
        "stable-diffusion",
        "wan",
        "mochi",
        "sana",
        "diffusion",
    ];
    if NON_TEXT_HINTS.iter().any(|h| lower.contains(h)) {
        return None;
    }

    // Capabilities by name. Embedding / reranker checkpoints share the same
    // causal-LM architecture as generators, so the name is the only signal.
    let capabilities: Vec<ModelCapability> =
        if lower.contains("embedding") || lower.contains("embed") {
            vec![ModelCapability::Embed]
        } else if lower.contains("reranker") || lower.contains("rerank") {
            vec![ModelCapability::Rerank]
        } else {
            vec![
                ModelCapability::Generate,
                ModelCapability::Code,
                ModelCapability::Reasoning,
            ]
        };

    // Locate weights and (for MLX) validate the architecture.
    let config_path = dir.join("config.json");
    let has_safetensors =
        dir.join("model.safetensors").exists() || dir.join("model.safetensors.index.json").exists();

    let (source, context_length, quantization) = if config_path.exists() && has_safetensors {
        // MLX layout. Only accept a recognized causal-LM model_type.
        let cfg: serde_json::Value = std::fs::read_to_string(&config_path)
            .ok()
            .and_then(|s| serde_json::from_str(&s).ok())?;
        let model_type = cfg
            .get("model_type")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_ascii_lowercase();
        const KNOWN_LLM_TYPES: &[&str] = &[
            "qwen",
            "qwen2",
            "qwen3",
            "qwen3_moe",
            "llama",
            "mistral",
            "mixtral",
            "gemma",
            "gemma2",
            "gemma3",
            "gemma4_unified",
            "gemma4_unified_text",
            "phi",
            "phi3",
            "phimoe",
            "starcoder2",
            "deepseek",
            "deepseek_v2",
            "internlm2",
            "cohere",
            "olmo",
        ];
        if !KNOWN_LLM_TYPES.iter().any(|t| model_type == *t) {
            return None;
        }
        let ctx = cfg
            .get("max_position_embeddings")
            .and_then(|v| v.as_u64())
            .unwrap_or(32_768) as usize;
        // Same block `hf_schema::quantization_of` reads, and for the same
        // reason: `bits` alone cannot tell an affine group quant from an MX
        // float, and the group size is what a kernel has to match. This used
        // to emit `"4-bit"` while the other ingest path emitted `"4bit"` —
        // two spellings of one thing, in a field nothing could compare.
        let quant = cfg
            .get("quantization")
            .filter(|q| q.is_object())
            .and_then(|q| {
                let bits = q
                    .get("bits")
                    .and_then(|b| b.as_u64())
                    .and_then(|b| u8::try_from(b).ok());
                let group_size = q
                    .get("group_size")
                    .and_then(|g| g.as_u64())
                    .and_then(|g| u32::try_from(g).ok());
                let mode = q.get("mode").and_then(|m| m.as_str());
                crate::schema::Quantization::from_mlx_config(bits, group_size, mode)
            });
        (
            serde_json::json!({ "type": "mlx", "hf_repo": "" }),
            ctx,
            quant,
        )
    } else {
        let gguf = std::fs::read_dir(dir).ok().and_then(|rd| {
            rd.flatten().map(|e| e.path()).find(|p| {
                p.extension()
                    .and_then(|x| x.to_str())
                    .is_some_and(|x| x.eq_ignore_ascii_case("gguf"))
            })
        })?;
        // GGUF layout (Candle). file name carried for the Local source.
        let filename = gguf
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("model.gguf")
            .to_string();
        // A GGUF file names its own quantization and this path used to throw
        // that away, so a user's own local checkpoints were the one ingest
        // route that recorded nothing at all.
        let quant = crate::schema::Quantization::from_gguf_filename(&filename);
        (
            serde_json::json!({
                "type": "local",
                "hf_repo": "",
                "hf_filename": filename,
                "tokenizer_repo": "",
            }),
            4_096,
            quant,
        )
    };

    let id = format!("local/{}", lower.replace(['/', ' '], "-"));
    serde_json::from_value(serde_json::json!({
        "id": id,
        "name": name,
        "provider": "local",
        "family": "local",
        "capabilities": capabilities,
        "context_length": context_length,
        "quantization": quantization,
        "source": source,
        "tags": ["auto-discovered"],
        "trust_tier": "community",
    }))
    .ok()
}

#[allow(dead_code)] // callers are Apple-Silicon/MLX-gated; dead in car_skip_mlx builds
fn speech_mlx_available() -> bool {
    // On Apple Silicon, speech uses native MLX backends — no Python CLI needed.
    // Models are available if we're on the right platform (weights are downloaded on demand).
    #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
    {
        true
    }

    // On other platforms, check for the Python mlx-audio CLI.
    #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
    {
        // Via `venv_program`, so this agrees with `SpeechRuntime` about where a
        // provisioned runtime keeps its console scripts — `Scripts\*.exe` on
        // Windows, `bin/*` elsewhere. This branch is the non-Apple-Silicon one,
        // so Windows is exactly who reaches it.
        let runtime_root = speech_runtime_root();
        crate::managed_venv::venv_program(&runtime_root, "mlx_audio.stt.generate").exists()
            || crate::managed_venv::venv_program(&runtime_root, "mlx_audio.tts.generate").exists()
    }
}

#[allow(dead_code)] // general (no MLX deps); called from speech_mlx_available's non-MLX branch
fn speech_runtime_root() -> PathBuf {
    if let Ok(path) = std::env::var("CAR_SPEECH_RUNTIME_DIR") {
        if !path.trim().is_empty() {
            return PathBuf::from(path);
        }
    }
    std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("."))
        .join(".car")
        .join("speech-runtime")
}

/// Backward-compatible ModelInfo for listing (used by CLI and old callers).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModelInfo {
    pub id: String,
    pub name: String,
    pub provider: String,
    pub capabilities: Vec<ModelCapability>,
    pub param_count: String,
    pub size_mb: u64,
    pub context_length: usize,
    pub available: bool,
    pub is_local: bool,
    /// True only for a raw `VllmMlx` source whose runtime is operated outside
    /// CAR. The endpoint may be loopback or remote; its address is not
    /// ownership evidence. CAR-owned `ManagedVllmMlx` remains a local row.
    #[serde(default)]
    pub operator_managed_external_runtime: bool,
    /// Weights resolvable on disk *now* — i.e. usable without a download.
    /// Distinct from `available`, which for a lazy-downloading MLX entry is
    /// true before a byte is fetched (#164). Remote models need no weights
    /// and report `true`. See `ModelSchema::weights_ready` (#638).
    ///
    /// `#[serde(default)]` so a client built against this version can still
    /// parse a catalog from a daemon that predates the field.
    #[serde(default)]
    pub weights_ready: bool,
    /// Whether CAR fetches weights to disk for this entry at all — i.e.
    /// whether `weights_ready` is a question with a meaningful answer.
    ///
    /// `false` for OS-provided models (`windows/speech-synthesis:os`,
    /// `apple/foundation:default`), for server-backed local models
    /// (`vllm-mlx/*`, Ollama) and for every remote entry. `is_local` is NOT a
    /// substitute: the first four of those are local and still download
    /// nothing. Consumers should render no install status when this is
    /// `false`. See `ModelSchema::downloads_weights` (#894).
    ///
    /// `#[serde(default)]` so a client built against this version can still
    /// parse a catalog from a daemon that predates the field.
    #[serde(default)]
    pub downloads_weights: bool,
    /// Per-model maximum OUTPUT tokens (registry-declared). `None` when the
    /// catalog entry omits it; callers fall back to a fraction of
    /// `context_length` via `ModelSchema::effective_max_output()`.
    #[serde(default)]
    pub max_output_tokens: Option<usize>,
    /// Public benchmark scores carried straight through from `ModelSchema`.
    /// The built-in catalog ships this empty; populating it is a curation
    /// step (see `BenchmarkScore` in the schema for shape and conventions).
    #[serde(default)]
    pub public_benchmarks: Vec<crate::schema::BenchmarkScore>,
    /// Declared per-token prices carried straight through from
    /// `ModelSchema::cost` — USD per 1M tokens for uncached input, output,
    /// cache reads and cache writes, plus any prompt-size pricing tiers the
    /// catalog declares.
    ///
    /// Every price component is `Option`: a local model that declares no
    /// prices reports them as `null`, which is deliberately **not** the same
    /// as a declared `0.0`. Consumers that price a request must distinguish
    /// "free" from "unpriced" — collapsing the two is how a caller ends up
    /// publishing a fabricated $0.00.
    ///
    /// `#[serde(default)]` so a newly built client can parse a catalog from
    /// an older daemon that predates this field: the row deserializes with an
    /// all-`None` cost rather than failing the whole response.
    #[serde(default)]
    pub cost: crate::schema::CostModel,
    #[serde(default = "default_true")]
    pub car_enabled: bool,
    #[serde(default)]
    pub can_remove: bool,
    #[serde(default)]
    pub in_use: bool,
    #[serde(default)]
    pub management_evidence: Option<String>,
    /// Whether this row fits the machine that answered, judged by the same
    /// rule `models.recommend` partitions on: the cold-load estimate at the
    /// recommendation context against the active resource policy's ceiling.
    /// `fits` for remote, delegated and operator-managed rows, whose memory is
    /// not this machine's. Computed per request from `HardwareInfo::detect()`
    /// and the engine's active policy; never persisted.
    ///
    /// `#[serde(default)]` (→ `unknown`) so a client built against this
    /// version can parse a catalog from a daemon that predates the field, and
    /// so an absent verdict never reads as too-big.
    #[serde(default)]
    pub fit: crate::recommend::ModelFitStatus,
    /// The cold-load peak `fit` was judged against, in MB. `null` when the
    /// row's memory is not this machine's or it declares no size/RAM figure.
    #[serde(default)]
    pub estimated_peak_mb: Option<u64>,
    /// Whether this machine's accelerator can run the row at all — `false`
    /// for a Metal-only row (MLX, managed vLLM-MLX, Apple FoundationModels)
    /// on a non-Apple machine, the rows `models.recommend` never offers.
    /// Defaults to `true` for the same older-daemon reason as `fit`.
    #[serde(default = "default_true")]
    pub platform_compatible: bool,
    /// Carried through from `ModelSchema::deprecated`: superseded in the
    /// catalog. The row stays listed (an installed deprecated model is still
    /// usable); clients lead with current models by sorting or hiding on it.
    #[serde(default)]
    pub deprecated: bool,
    /// Model line for grouping (`qwen3`, `gemma-4`), published for local rows
    /// only. Remote rows report `null` because this response deliberately
    /// carries no upstream identifier for the managed `parslee/openrouter/*`
    /// aliases, whose `family` names the upstream model; `models.search`
    /// publishes it for every row.
    #[serde(default)]
    pub family: Option<String>,
    /// Version or checkpoint label within the family, local rows only for
    /// the same reason as `family` (an alias's `version` is the upstream
    /// snapshot).
    #[serde(default)]
    pub version: Option<String>,
}

fn default_true() -> bool {
    true
}

impl ModelInfo {
    /// Stamp a computed fit annotation onto the row. Kept as one call so no
    /// caller can publish `fit` without the estimate and platform flag it
    /// was judged with.
    pub fn with_fit(mut self, fit: crate::recommend::ModelFit) -> Self {
        self.fit = fit.fit;
        self.estimated_peak_mb = fit.estimated_peak_mb;
        self.platform_compatible = fit.platform_compatible;
        self
    }
}

impl From<&ModelSchema> for ModelInfo {
    fn from(s: &ModelSchema) -> Self {
        ModelInfo {
            id: s.id.clone(),
            name: s.name.clone(),
            provider: s.provider.clone(),
            capabilities: s.capabilities.clone(),
            param_count: s.param_count.clone(),
            size_mb: s.size_mb(),
            context_length: s.context_length,
            available: s.available_now(),
            is_local: s.is_local(),
            operator_managed_external_runtime: matches!(s.source, ModelSource::VllmMlx { .. }),
            weights_ready: s.weights_ready,
            downloads_weights: s.downloads_weights(),
            max_output_tokens: s.max_output_tokens,
            public_benchmarks: s.public_benchmarks.clone(),
            // Prices only — the managed `parslee/…` aliases carry the same
            // `CostModel` as the personal row they front, and a `CostModel`
            // holds no identifiers, so publishing it discloses nothing about
            // the upstream model id.
            cost: s.cost.clone(),
            car_enabled: true,
            can_remove: false,
            in_use: false,
            management_evidence: None,
            // Not a verdict: the machine and policy are the engine's to
            // supply. `list_models_unified` stamps the real annotation with
            // `with_fit`; an unannotated projection reads as "unknown", the
            // value that hides nothing.
            fit: crate::recommend::ModelFitStatus::Unknown,
            estimated_peak_mb: None,
            platform_compatible: true,
            deprecated: s.deprecated,
            // Local rows only — see the field docs for the non-disclosure
            // rule on managed aliases.
            family: s.is_local().then(|| s.family.clone()),
            version: s.is_local().then(|| s.version.clone()),
        }
    }
}

/// Emit a lightweight `FileStarted` marker for an MLX auxiliary/weight file.
/// MLX pulls don't know `total_files` up front (single vs sharded), so these
/// carry `total_files = 0` ("unknown") and the UI shows the file name. The
/// sharded weight loop, which knows its count, pairs Started/Completed itself.
fn emit_file(sink: &ProgressSink, name: &str, index: u32, size_mb: u64) {
    sink.emit(DownloadEvent::FileStarted {
        filename: name.to_string(),
        index,
        total_files: 0,
        size_mb,
    });
}

/// Download a single file from a HuggingFace repo.
/// Mirror every file in a HuggingFace repo into `model_dir`, preserving its
/// subdirectory structure. For diffusers-layout models (Flux/LTX) whose weights
/// live in component subdirs (`transformer/`, `vae/`, `text_encoder/`, …) with
/// no root `config.json` or top-level weight index — the file-by-file MLX flow
/// can't enumerate them, so the whole snapshot is fetched. The repo file list
/// comes from the HF models API (version-independent, unlike hf-hub's `info()`);
/// metadata (`.gitattributes`, dotfiles, markdown) is skipped.
async fn download_repo_snapshot(
    repo: &str,
    model_dir: &Path,
    sink: &ProgressSink,
) -> Result<(), InferenceError> {
    #[derive(serde::Deserialize)]
    struct RepoInfo {
        siblings: Vec<Sibling>,
    }
    #[derive(serde::Deserialize)]
    struct Sibling {
        rfilename: String,
    }
    let url = format!("https://huggingface.co/api/models/{repo}");
    // Not `Client::new()`: that is `build().expect(..)`, which panics when the
    // OS trust store loads zero valid certificates — the failure the ladder in
    // `tls_client` exists to survive. huggingface.co is a public endpoint, so
    // the public-CA rung serves this call fully.
    let info: RepoInfo = crate::tls_client::model_download_client()
        .get(&url)
        .send()
        .await
        .map_err(|e| InferenceError::DownloadFailed(format!("list {repo}: {e}")))?
        .error_for_status()
        .map_err(|e| InferenceError::DownloadFailed(format!("list {repo}: {e}")))?
        .json()
        .await
        .map_err(|e| InferenceError::DownloadFailed(format!("parse {repo} file list: {e}")))?;

    let files: Vec<String> = info
        .siblings
        .into_iter()
        .map(|s| s.rfilename)
        .filter(|f| !f.starts_with('.') && !f.to_ascii_lowercase().ends_with(".md"))
        .collect();
    if files.is_empty() {
        return Err(InferenceError::DownloadFailed(format!(
            "{repo}: repo lists no downloadable files"
        )));
    }

    let total = files.len() as u32;
    for (i, fname) in files.iter().enumerate() {
        let dest = model_dir.join(fname);
        if crate::download::cache_file_usable(&dest) {
            continue;
        }
        if let Some(parent) = dest.parent() {
            std::fs::create_dir_all(parent)?;
        }
        sink.emit(DownloadEvent::FileStarted {
            filename: fname.clone(),
            index: (i + 1) as u32,
            total_files: total,
            size_mb: 0,
        });
        download_file(repo, fname, &dest).await?;
        sink.emit(DownloadEvent::FileCompleted {
            filename: fname.clone(),
        });
    }
    Ok(())
}

/// Tokenizer filenames seen across the repos CAR pulls, in preference order.
///
/// There is no single convention, which is the whole point of this list:
///
/// | Layout | Files | Example |
/// |---|---|---|
/// | `tokenizers` library | `tokenizer.json` | most modern text MLX repos |
/// | GPT-2 BPE pair | `vocab.json`, `merges.txt` | `Qwen3-TTS-12Hz-1.7B-Base-5bit` |
/// | SentencePiece | `tokenizer.model`, `tokenizer.vocab`, `vocab.txt` | `parakeet-tdt-0.6b-v3` |
/// | none at all | — | `Kokoro-82M-*` (phoneme-based, needs no tokenizer) |
const TOKENIZER_FILENAMES: &[&str] = &[
    "tokenizer.json",
    "vocab.json",
    "merges.txt",
    "tokenizer.model",
    "tokenizer.vocab",
    "vocab.txt",
];

/// Fetch whatever tokenizer assets a repo actually has. **Never fails the pull
/// over their absence.**
///
/// This used to be a single `download_file(repo, "tokenizer.json", …).await?` —
/// an unconditional hard requirement that aborted the entire multi-gigabyte
/// acquisition when the file 404'd. That made every curated speech model
/// impossible to install, in either direction: both Kokoro TTS builds, the
/// Qwen3-TTS default, and the Parakeet STT default, so local speech did not work
/// at all on a clean machine (Parslee-ai/car#639).
///
/// Best-effort is the correct posture here, not a workaround. The downloader
/// cannot know what a given backend needs — and Kokoro settles it: it ships
/// **no** tokenizer files whatsoever, because a phoneme-based TTS has no use for
/// one. Any rule of the form "a tokenizer must be present" is therefore wrong
/// for some model CAR legitimately supports. A loader that genuinely needs a
/// tokenizer will fail at load with a precise, model-specific error, which beats
/// a blanket download-time abort that names a file the model was never going to
/// have. This also matches the treatment `tokenizer_config.json` already gets
/// on the very next line.
async fn download_tokenizer_assets(hf_repo: &str, model_dir: &Path, sink: &ProgressSink) {
    // Already satisfied by any recognized layout? Then don't re-probe the API.
    if TOKENIZER_FILENAMES
        .iter()
        .any(|f| crate::download::cache_file_usable(&model_dir.join(f)))
    {
        return;
    }
    emit_file(sink, "tokenizer", 0, 0);
    let mut fetched: Vec<&str> = Vec::new();
    for name in TOKENIZER_FILENAMES {
        let dest = model_dir.join(name);
        if crate::download::cache_file_usable(&dest) {
            continue;
        }
        if download_file(hf_repo, name, &dest).await.is_ok() {
            fetched.push(name);
        }
    }
    if fetched.is_empty() {
        // Not an error: see the doc comment — Kokoro has none by design.
        tracing::debug!(
            repo = %hf_repo,
            "no tokenizer assets in this repo; continuing (the backend may not need one)"
        );
    } else {
        tracing::debug!(repo = %hf_repo, files = ?fetched, "fetched tokenizer assets");
    }
}

async fn download_file(repo: &str, filename: &str, dest: &Path) -> Result<(), InferenceError> {
    // Do not even consult the remote/cache when a prior install is usable.
    // Conversely, bare existence is not enough: zero-byte files and dangling
    // symlinks must be repaired rather than accepted forever.
    if crate::download::cache_file_usable(dest) {
        return Ok(());
    }

    let api = hf_hub::api::tokio::Api::new()
        .map_err(|e| InferenceError::DownloadFailed(e.to_string()))?;

    let repo = api.model(repo.to_string());
    let path = repo
        .get(filename)
        .await
        .map_err(|e| InferenceError::DownloadFailed(format!("{filename}: {e}")))?;

    install_fetched_file(&path, dest)
}

/// Install an already-fetched cache file without consulting the network.
fn install_fetched_file(src: &Path, dest: &Path) -> Result<(), InferenceError> {
    if crate::download::cache_file_usable(dest) {
        return Ok(());
    }

    match std::fs::symlink_metadata(dest) {
        Ok(_) => std::fs::remove_file(dest).map_err(|e| {
            InferenceError::DownloadFailed(format!(
                "remove unusable destination {}: {e}",
                dest.display()
            ))
        })?,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => {
            return Err(InferenceError::DownloadFailed(format!(
                "inspect destination {}: {error}",
                dest.display()
            )));
        }
    }

    // Symlink creation publishes one complete directory entry. If the
    // platform/filesystem does not support it, copy beside the destination and
    // rename only after the complete source bytes are present.
    #[cfg(unix)]
    {
        if std::os::unix::fs::symlink(src, dest).is_ok() {
            return Ok(());
        }
    }

    static INSTALL_SEQUENCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let sequence = INSTALL_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let file_name = dest
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("download");
    let temp = dest.with_file_name(format!(
        ".{file_name}.car-install-{}-{sequence}.tmp",
        std::process::id()
    ));
    let _ = std::fs::remove_file(&temp);
    std::fs::copy(src, &temp).map_err(|error| {
        InferenceError::DownloadFailed(format!(
            "copy to temporary destination {}: {error}",
            temp.display()
        ))
    })?;
    if let Err(error) = std::fs::rename(&temp, dest) {
        let _ = std::fs::remove_file(&temp);
        return Err(InferenceError::DownloadFailed(format!(
            "publish downloaded file at {}: {error}",
            dest.display()
        )));
    }
    Ok(())
}

fn auxiliary_mlx_files_missing(model_name: &str, hf_repo: &str, model_dir: &Path) -> bool {
    (hf_repo == "mlx-community/Flux-1.lite-8B-MLX-Q4" || model_name == "Flux-1.lite-8B-MLX-Q4")
        && !crate::download::cache_file_usable(
            &model_dir.join("tokenizer_2").join("tokenizer.json"),
        )
}

async fn ensure_auxiliary_mlx_files(
    model_name: &str,
    hf_repo: &str,
    model_dir: &Path,
) -> Result<(), InferenceError> {
    if auxiliary_mlx_files_missing(model_name, hf_repo, model_dir) {
        let t5_tokenizer_path = model_dir.join("tokenizer_2").join("tokenizer.json");
        std::fs::create_dir_all(
            t5_tokenizer_path
                .parent()
                .ok_or_else(|| InferenceError::InferenceFailed("invalid tokenizer path".into()))?,
        )?;
        info!(
            path = %t5_tokenizer_path.display(),
            "downloading missing Flux tokenizer_2/tokenizer.json from base model"
        );
        download_file(
            "Freepik/flux.1-lite-8B",
            "tokenizer_2/tokenizer.json",
            &t5_tokenizer_path,
        )
        .await?;
    }
    Ok(())
}

fn mlx_auxiliary_ready_without_download(model_name: &str, model_dir: &Path) -> bool {
    if model_name == "Flux-1.lite-8B-MLX-Q4" {
        return crate::download::cache_file_usable(
            &model_dir.join("tokenizer_2").join("tokenizer.json"),
        );
    }
    true
}

/// The single physical-readiness projection used by registration, refresh,
/// recommendation/setup state, and ready-without-download checks.
fn physical_weights_ready(schema: &ModelSchema, models_dir: &Path) -> bool {
    physical_weights_ready_with_huggingface_hub(schema, models_dir, None)
}

pub(crate) fn physical_weights_ready_with_huggingface_hub(
    schema: &ModelSchema,
    models_dir: &Path,
    huggingface_hub_root: Option<&Path>,
) -> bool {
    match &schema.source {
        ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
            let managed_dir = models_dir.join(&schema.name);
            if mlx_snapshot_complete(schema, &managed_dir) {
                return true;
            }
            let shared_snapshot = match huggingface_hub_root {
                Some(root) => latest_huggingface_repo_snapshot_in(
                    &root.join(format!("models--{}", hf_repo.replace('/', "--"))),
                ),
                None => latest_huggingface_repo_snapshot(hf_repo),
            };
            shared_snapshot
                .as_deref()
                .is_some_and(|snapshot| mlx_snapshot_complete(schema, snapshot))
        }
        ModelSource::WhisperCpp { model } => car_whisper::model_cached(model),
        ModelSource::Local { .. } => {
            crate::download::cache_file_usable(&models_dir.join(&schema.name).join("model.gguf"))
        }
        // These sources have no CAR-downloaded artifact. `weights_ready` is a
        // readiness sentinel for legacy routing; callers must gate install
        // language with `downloads_weights()`.
        ModelSource::WindowsSpeech {}
        | ModelSource::AppleFoundationModels { .. }
        | ModelSource::VllmMlx { .. }
        | ModelSource::Ollama { .. }
        | ModelSource::RemoteApi { .. }
        | ModelSource::CodexCli { .. }
        | ModelSource::Proprietary { .. }
        | ModelSource::Delegated { .. } => true,
    }
}

#[cfg(test)]
fn mlx_weights_ready_at(
    schema: &ModelSchema,
    managed_dir: &Path,
    shared_snapshot: Option<&Path>,
) -> bool {
    mlx_snapshot_complete(schema, managed_dir)
        || shared_snapshot.is_some_and(|snapshot| mlx_snapshot_complete(schema, snapshot))
}

/// A complete MLX location, whether CAR's managed directory or a shared
/// Hugging Face snapshot. Standard text models need config + all indexed
/// shards. Diffusers layouts have no root config, so their recursively
/// validated component weights are the completeness authority. Tokenizer
/// readiness is a stricter invocation concern handled by
/// `ready_without_download`; `weights_ready` remains literal physical weights.
fn mlx_snapshot_complete(schema: &ModelSchema, dir: &Path) -> bool {
    let is_diffusers = schema.capabilities.iter().any(|capability| {
        matches!(
            capability,
            ModelCapability::ImageGeneration | ModelCapability::VideoGeneration
        )
    });
    let metadata_ready =
        is_diffusers || crate::download::cache_file_usable(&dir.join("config.json"));

    metadata_ready
        && mlx_dir_has_weights(dir)
        && mlx_auxiliary_ready_without_download(&schema.name, dir)
}

/// Does a managed MLX model dir actually contain resolvable weights?
///
/// A dir holding only `config.json`/`tokenizer.json` stubs — or a dangling
/// `*.safetensors` symlink into a pruned HuggingFace cache — is NOT a complete
/// install. Treating it as installed makes `ensure_local`/`car models pull`
/// no-op and inference then fails at load with "no safetensors weights found"
/// (car-releases#391). We check for at least one `*.safetensors` whose target
/// resolves: `Path::exists()` follows symlinks, so a dangling link counts as
/// absent. This covers single-file (`model.safetensors`) and sharded
/// (`model-00001-of-0000N.safetensors`) layouts; the bare
/// `model.safetensors.index.json` (extension `.json`) correctly doesn't count.
///
/// **Sharded models are checked against their index, not by existence of any
/// one shard.** "At least one `*.safetensors` resolves" is the right question
/// for a single-file model and the wrong one for a sharded install: an
/// interrupted download that landed shard 2 of 2 satisfies it, so
/// `ensure_local` returns early without repairing, `car models pull` reports
/// success having done nothing, and the failure surfaces only at load —
/// `load model-00001-of-00002.safetensors: Path must point to a local file`
/// (Parslee-ai/car#808). The index enumerates every required shard, so when it
/// is present it is authoritative and cheap to consult.
pub(crate) fn mlx_dir_has_weights(dir: &Path) -> bool {
    // An index present at the top level means a sharded layout: require EVERY
    // shard it names. Absent (single-file, or a diffusers-layout model whose
    // components live in subdirs), fall back to the recursive any-weights walk.
    let index = dir.join("model.safetensors.index.json");
    if index.is_file() {
        return sharded_weight_files(&index).is_some_and(|required| {
            !required.is_empty()
                && required
                    .iter()
                    .all(|shard| crate::download::cache_file_usable(&dir.join(shard)))
        });
    }
    mlx_dir_has_weights_depth(dir, 0)
}

/// Weight shards the index requires but that are not present on disk (#808).
///
/// Empty means "no shard filename could be reported" — including for
/// single-file/diffusers layouts and malformed indexes. Callers must pair this
/// with [`mlx_dir_has_weights`]; an empty diagnostic is never proof that the
/// model is complete.
///
/// Exists so a failed pull can say *which* shard is absent. The reported case
/// (#808) left a 6.7 GB model with one of two shards on disk for weeks, and the
/// only symptom was an inference-time `Path must point to a local file` naming
/// a shard the caller had no reason to know about.
pub(crate) fn missing_weight_shards(dir: &Path) -> Vec<String> {
    let index = dir.join("model.safetensors.index.json");
    if !index.is_file() {
        return Vec::new();
    }
    let Some(required) = sharded_weight_files(&index) else {
        // Unreadable index: there is no trustworthy shard filename to report.
        // Readiness still fails closed in `mlx_dir_has_weights`.
        return Vec::new();
    };
    required
        .into_iter()
        .filter(|shard| !dir.join(shard).exists())
        .collect()
}

/// Shard filenames an MLX/HF `model.safetensors.index.json` says are required.
///
/// `weight_map` maps every tensor name to the file holding it, so the distinct
/// values are exactly the shard set. Returns `None` when the index cannot be
/// read or has no `weight_map`; readiness callers must fail closed when an
/// index exists but this metadata cannot be validated.
fn sharded_weight_files(index: &Path) -> Option<Vec<String>> {
    let raw = std::fs::read_to_string(index).ok()?;
    let parsed: serde_json::Value = serde_json::from_str(&raw).ok()?;
    let map = parsed.get("weight_map")?.as_object()?;
    let mut files: Vec<String> = map
        .values()
        .filter_map(|v| v.as_str().map(str::to_string))
        .collect();
    files.sort();
    files.dedup();
    Some(files)
}

/// Recurse into component subdirs to find weights, but bounded. Diffusers-layout
/// models (Flux image, LTX video, …) keep their component weights in subdirs
/// (`transformer/`, `vae/`, `text_encoder/`, …), not at the top level, so a flat
/// scan misjudges a fully-cached snapshot as weightless (the miss that sent the
/// image/video path into a re-download that 404s on the `config.json` these
/// repos don't ship). `load_all_tensors` reads those component safetensors
/// recursively. The walk is depth-capped and skips symlinked directories so a
/// symlink cycle in a cache dir can't stack-overflow us — the HF cache nests
/// real component dirs at most ~2 deep, with only leaf blob files symlinked
/// (those are files, resolved by `cache_file_usable`, not followed here).
fn mlx_dir_has_weights_depth(dir: &Path, depth: usize) -> bool {
    if depth > 4 {
        return false;
    }
    let Ok(rd) = std::fs::read_dir(dir) else {
        return false;
    };
    rd.flatten().any(|e| {
        let p = e.path();
        let is_symlink = std::fs::symlink_metadata(&p)
            .map(|m| m.file_type().is_symlink())
            .unwrap_or(true);
        if p.is_dir() {
            !is_symlink && mlx_dir_has_weights_depth(&p, depth + 1)
        } else {
            // `cache_file_usable` follows the symlink and requires non-empty, so
            // a dangling weight symlink (pruned blob) or a zero-length partial
            // write does not count as installed.
            p.extension().and_then(|x| x.to_str()) == Some("safetensors")
                && crate::download::cache_file_usable(&p)
        }
    })
}

#[allow(dead_code)] // conditionally compiled — used only on MLX-backend (macOS) snapshot-resolution paths
fn huggingface_repo_has_snapshot(repo_id: &str) -> bool {
    latest_huggingface_repo_snapshot(repo_id).is_some()
}

pub(crate) fn huggingface_cache_root() -> PathBuf {
    std::env::var("HF_HOME")
        .map(PathBuf::from)
        .unwrap_or_else(|_| {
            std::env::var_os("HOME")
                .or_else(|| std::env::var_os("USERPROFILE"))
                .map(PathBuf::from)
                .unwrap_or_else(|| PathBuf::from("."))
                .join(".cache")
                .join("huggingface")
        })
        .join("hub")
}

pub(crate) fn huggingface_repo_dir(repo_id: &str) -> PathBuf {
    huggingface_cache_root().join(format!("models--{}", repo_id.replace('/', "--")))
}

fn resolve_huggingface_ref_snapshot(repo_dir: &Path, name: &str) -> Option<PathBuf> {
    let sha = std::fs::read_to_string(repo_dir.join("refs").join(name))
        .ok()?
        .trim()
        .to_string();
    if sha.is_empty() {
        return None;
    }

    let snapshot = repo_dir.join("snapshots").join(sha);
    if snapshot_looks_ready(&snapshot) {
        Some(snapshot)
    } else {
        None
    }
}

fn latest_huggingface_repo_snapshot(repo_id: &str) -> Option<PathBuf> {
    let repo_dir = huggingface_repo_dir(repo_id);
    latest_huggingface_repo_snapshot_in(&repo_dir)
}

fn latest_huggingface_repo_snapshot_in(repo_dir: &Path) -> Option<PathBuf> {
    if let Some(snapshot) = resolve_huggingface_ref_snapshot(repo_dir, "main") {
        return Some(snapshot);
    }

    let snapshots = repo_dir.join("snapshots");
    let mut candidates: Vec<(SystemTime, PathBuf)> = std::fs::read_dir(snapshots)
        .ok()?
        .filter_map(Result::ok)
        .map(|e| e.path())
        .filter(|p| p.is_dir() && snapshot_looks_ready(p))
        .map(|path| {
            let modified = path
                .metadata()
                .and_then(|metadata| metadata.modified())
                .unwrap_or(SystemTime::UNIX_EPOCH);
            (modified, path)
        })
        .collect();
    candidates.sort();
    candidates.pop().map(|(_, path)| path)
}

fn snapshot_looks_ready(path: &Path) -> bool {
    if path.join("config.json").exists() || path.join("model_index.json").exists() {
        return true;
    }
    snapshot_contains_ext(path, "safetensors")
}

fn snapshot_contains_ext(root: &Path, ext: &str) -> bool {
    let Ok(entries) = std::fs::read_dir(root) else {
        return false;
    };
    entries.filter_map(Result::ok).any(|entry| {
        let path = entry.path();
        if path.is_dir() {
            snapshot_contains_ext(&path, ext)
        } else {
            let ext_matches = path
                .extension()
                .and_then(|value| value.to_str())
                .map(|value| value.eq_ignore_ascii_case(ext))
                .unwrap_or(false);
            // A matching extension only counts when the file is actually usable
            // — a dangling symlink into a pruned blob or a zero-length partial
            // must not make a snapshot look ready.
            ext_matches && crate::download::cache_file_usable(&path)
        }
    })
}

/// Built-in catalog parsed from `builtin_catalog.json`.
///
/// Adding, removing, or editing a model is a JSON-only change — Rust
/// source stays put. The JSON is embedded at compile time via
/// `include_str!`, parsed once into a `LazyLock`, and cloned on each
/// call. A malformed JSON file fails the integration test
/// `builtin_catalog_json_parses` so the binary never ships unable
/// to load its own catalog.
const BUILTIN_CATALOG_JSON: &str = include_str!("builtin_catalog.json");

static BUILTIN_CATALOG: std::sync::LazyLock<Vec<ModelSchema>> = std::sync::LazyLock::new(|| {
    serde_json::from_str(BUILTIN_CATALOG_JSON)
        .expect("builtin_catalog.json failed to parse — fix the JSON, not this code")
});

pub(crate) fn builtin_catalog() -> Vec<ModelSchema> {
    let mut catalog = BUILTIN_CATALOG.clone();
    catalog.extend(crate::openrouter::builtin_schemas());
    catalog
}

/// Test support for consumers that need real built-in rows without consulting
/// the developer machine's shared Hugging Face cache.
#[doc(hidden)]
pub fn builtin_catalog_with_huggingface_hub_for_testing(
    models_dir: &Path,
    huggingface_hub_root: &Path,
) -> Vec<ModelSchema> {
    let mut catalog = builtin_catalog();
    for schema in &mut catalog {
        schema.weights_ready = physical_weights_ready_with_huggingface_hub(
            schema,
            models_dir,
            Some(huggingface_hub_root),
        );
    }
    catalog
}

#[cfg(test)]
mod tests {
    use crate::openrouter::StateRootScope;

    /// A sharded install missing one shard is NOT installed.
    ///
    /// The interrupted-download shape from car#808: shard 2 of 2 landed, shard
    /// 1 did not. The old any-weights check said "installed", so `ensure_local`
    /// returned early without repairing, `pull_model` reported success having
    /// done nothing, and the failure surfaced only at load. Synthetic dir — no
    /// real weights needed, which is what makes it runnable in CI.
    #[test]
    fn a_sharded_model_missing_one_shard_is_not_installed() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path();
        std::fs::write(
            dir.join("model.safetensors.index.json"),
            r#"{"weight_map":{"a":"model-00001-of-00002.safetensors",
                              "b":"model-00002-of-00002.safetensors"}}"#,
        )
        .unwrap();
        // Only shard 2 present — exactly the reported state.
        std::fs::write(dir.join("model-00002-of-00002.safetensors"), b"x").unwrap();
        assert!(
            !mlx_dir_has_weights(dir),
            "a missing shard must read as not-installed, or pull silently no-ops"
        );

        // Completing the set flips it.
        std::fs::write(dir.join("model-00001-of-00002.safetensors"), b"x").unwrap();
        assert!(
            mlx_dir_has_weights(dir),
            "a complete shard set must read as installed"
        );
    }

    /// A failed pull must be able to NAME the shard it is missing (car#808).
    ///
    /// The reported symptom was an inference-time
    /// `load model-00001-of-00002.safetensors: Path must point to a local file`
    /// — a filename the caller had no way to anticipate, surfacing inside a
    /// benchmarking run rather than at pull time. `missing_weight_shards` is
    /// what lets the pull itself say which shard is absent.
    #[test]
    fn missing_shards_are_reported_by_name() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path();
        std::fs::write(
            dir.join("model.safetensors.index.json"),
            r#"{"weight_map":{"a":"model-00001-of-00002.safetensors",
                              "b":"model-00002-of-00002.safetensors"}}"#,
        )
        .unwrap();
        std::fs::write(dir.join("model-00002-of-00002.safetensors"), b"x").unwrap();

        assert_eq!(
            missing_weight_shards(dir),
            vec!["model-00001-of-00002.safetensors".to_string()],
            "the absent shard must be named, not just counted"
        );

        std::fs::write(dir.join("model-00001-of-00002.safetensors"), b"x").unwrap();
        assert!(
            missing_weight_shards(dir).is_empty(),
            "a complete shard set must report nothing missing"
        );
    }

    /// No index (single-file or diffusers layout) means there is no shard list
    /// to check against — report nothing missing rather than inventing a
    /// failure. `mlx_dir_has_weights` remains the completeness signal there.
    #[test]
    fn missing_shards_is_empty_without_an_index() {
        let tmp = tempfile::tempdir().unwrap();
        std::fs::write(tmp.path().join("model.safetensors"), b"x").unwrap();
        assert!(missing_weight_shards(tmp.path()).is_empty());

        // An unreadable index is a metadata problem, not proof of a missing
        // shard — same call `mlx_dir_has_weights` makes, so they cannot disagree.
        let bad = tempfile::tempdir().unwrap();
        std::fs::write(bad.path().join("model.safetensors.index.json"), b"not json").unwrap();
        assert!(missing_weight_shards(bad.path()).is_empty());
    }

    /// A single-file model has no index; the any-weights walk still governs.
    #[test]
    fn a_single_file_model_still_counts_without_an_index() {
        let tmp = tempfile::tempdir().unwrap();
        std::fs::write(tmp.path().join("model.safetensors"), b"x").unwrap();
        assert!(mlx_dir_has_weights(tmp.path()));
    }

    /// A present but unreadable index is authoritative evidence we cannot
    /// validate, so readiness fails closed even if a stray weight file exists.
    #[test]
    fn an_unparseable_index_fails_closed() {
        let tmp = tempfile::tempdir().unwrap();
        std::fs::write(
            tmp.path().join("model.safetensors.index.json"),
            b"{not-json",
        )
        .unwrap();
        std::fs::write(tmp.path().join("model.safetensors"), b"x").unwrap();
        assert!(
            !mlx_dir_has_weights(tmp.path()),
            "an unreadable index must not fall back to a stray weight"
        );
    }

    use super::*;
    use tempfile::TempDir;

    #[test]
    fn mlx_dir_has_weights_detects_completeness() {
        let tmp = TempDir::new().unwrap();
        let dir = tmp.path();

        // A config-only stub is NOT complete (car-releases#391).
        std::fs::write(dir.join("config.json"), "{}").unwrap();
        std::fs::write(dir.join("tokenizer.json"), "{}").unwrap();
        assert!(
            !mlx_dir_has_weights(dir),
            "config-only stub must not count as installed"
        );

        // The bare sharded index without shards is still incomplete.
        std::fs::write(dir.join("model.safetensors.index.json"), "{}").unwrap();
        assert!(!mlx_dir_has_weights(dir), "index.json alone is not weights");

        // A real single-file layout has no shard index.
        std::fs::remove_file(dir.join("model.safetensors.index.json")).unwrap();
        std::fs::write(dir.join("model.safetensors"), b"\x00\x01\x02").unwrap();
        assert!(mlx_dir_has_weights(dir));
    }

    #[test]
    fn mlx_dir_has_weights_handles_sharded_and_dangling_symlinks() {
        let sharded = TempDir::new().unwrap();
        std::fs::write(sharded.path().join("config.json"), "{}").unwrap();
        std::fs::write(
            sharded.path().join("model-00001-of-00002.safetensors"),
            b"\x00",
        )
        .unwrap();
        assert!(mlx_dir_has_weights(sharded.path()), "sharded shard counts");

        // A DANGLING *.safetensors symlink (into a pruned HF cache) must NOT
        // count — Path::exists() follows the link and returns false.
        #[cfg(unix)]
        {
            let dangling = TempDir::new().unwrap();
            std::fs::write(dangling.path().join("config.json"), "{}").unwrap();
            std::os::unix::fs::symlink(
                dangling.path().join("does-not-exist"),
                dangling.path().join("model.safetensors"),
            )
            .unwrap();
            assert!(
                !mlx_dir_has_weights(dangling.path()),
                "dangling weight symlink must count as absent"
            );
        }
    }

    /// A registry over a temp state root, bundled with the two things that
    /// must outlive it: the temp dir, and this crate's environment lock.
    ///
    /// Holding the lock is not optional, and RETURNING it rather than binding
    /// it inside the helper is the whole point — a guard dropped at the end of
    /// `test_registry()` protects nothing. Every `UnifiedRegistry::new*` ends
    /// in `refresh_availability()`, which clears the process-global gateway
    /// observation AND `$CAR_HOME/gateway-state.json` whenever no Parslee
    /// token is present — exactly a CI runner's state. Two tests in this
    /// module (`a_gateway_that_reports_no_upstream_stops_being_advertised`,
    /// `a_rejected_credential_stops_the_managed_lane_being_advertised`) assert
    /// on that same observation, so ANY unserialized construction in this
    /// module can fail them. #989 locked the two direct
    /// `refresh_availability()` call sites but not the constructor path; this
    /// closes it. `cargo nextest` hides the family by giving every test its
    /// own process, which is why it stayed hidden until `shared-process-test`
    /// put a threaded `cargo test` on the per-PR path — see
    /// docs/solutions/process-global-state-in-tests.md.
    ///
    /// Modelled on `adaptive_router::tests::TestRegistry`, which already has
    /// this shape for the same reason.
    struct TestRegistry {
        registry: UnifiedRegistry,
        // Field drop order is intentional: tear the temp state root down
        // before releasing the lock, so the next test never sees a half-gone
        // `CAR_HOME`-shaped directory.
        _tmp: TempDir,
        _environment: tokio::sync::MutexGuard<'static, ()>,
    }

    impl std::ops::Deref for TestRegistry {
        type Target = UnifiedRegistry;

        fn deref(&self) -> &Self::Target {
            &self.registry
        }
    }

    impl std::ops::DerefMut for TestRegistry {
        fn deref_mut(&mut self) -> &mut Self::Target {
            &mut self.registry
        }
    }

    fn test_registry() -> TestRegistry {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let registry = UnifiedRegistry::new_with_state_root(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
        );
        TestRegistry {
            registry,
            _tmp: tmp,
            _environment,
        }
    }

    fn test_generate_schema(id: &str, name: &str, source: ModelSource) -> ModelSchema {
        ModelSchema {
            id: id.into(),
            name: name.into(),
            provider: "local".into(),
            family: "qwen3".into(),
            version: "test".into(),
            capabilities: vec![ModelCapability::Generate],
            context_length: 4096,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: PerformanceEnvelope::default(),
            cost: CostModel::default(),
            source,
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: false,
            weights_ready: false,
        }
    }

    /// Parslee-ai/car#894: the schema knows whether weights are on disk, but
    /// the schema → `ModelInfo` projection dropped that fact, so the CLI could
    /// only render `available` and contradicted `car doctor`. The projection
    /// must carry `weights_ready` through in both directions.
    #[test]
    fn model_info_carries_weights_ready_through_the_projection() {
        let mut schema = test_generate_schema(
            "mlx-community/car894-test-4bit",
            "car894-test-4bit",
            ModelSource::Mlx {
                hf_repo: "mlx-community/car894-test-4bit".into(),
                hf_weight_file: None,
            },
        );

        schema.weights_ready = false;
        assert!(
            !ModelInfo::from(&schema).weights_ready,
            "a schema with no weights on disk must project weights_ready = false"
        );

        schema.weights_ready = true;
        assert!(
            ModelInfo::from(&schema).weights_ready,
            "a schema with weights on disk must project weights_ready = true"
        );
    }

    /// The `weights_ready` flag is only meaningful for entries CAR actually
    /// downloads. `downloads_weights` is that gate, and it has to survive the
    /// schema → `ModelInfo` projection or the CLI is back to guessing from
    /// `is_local` — which is what made `windows/speech-synthesis:os` claim an
    /// install it does not have (#894).
    #[test]
    fn model_info_carries_downloads_weights_through_the_projection() {
        let mlx = test_generate_schema(
            "mlx-community/car894-test-4bit",
            "car894-test-4bit",
            ModelSource::Mlx {
                hf_repo: "mlx-community/car894-test-4bit".into(),
                hf_weight_file: None,
            },
        );
        assert!(
            ModelInfo::from(&mlx).downloads_weights,
            "an MLX entry downloads weights"
        );

        // OS-owned sources are local even though CAR never downloads for
        // them, so a projection that reused `is_local` would get them wrong.
        for (label, source) in [
            ("windows speech", ModelSource::WindowsSpeech {}),
            (
                "apple foundation",
                ModelSource::AppleFoundationModels { use_case: None },
            ),
        ] {
            let schema = test_generate_schema("car894/os-model", "os-model", source);
            let info = ModelInfo::from(&schema);
            assert!(
                !info.downloads_weights,
                "{label} installs nothing, so the projection must say so"
            );
            assert!(
                info.is_local,
                "{label} is still local — which is exactly why is_local cannot stand in"
            );
        }

        // Raw vLLM-MLX is independently owned and therefore remote even when
        // its endpoint happens to use a loopback spelling. CAR neither
        // downloads its weights nor imposes the client machine's hardware
        // requirements on it.
        let external = test_generate_schema(
            "car894/external-model",
            "external-model",
            ModelSource::VllmMlx {
                endpoint: "http://localhost:8000".into(),
                model_name: "mlx-community/car894-test-4bit".into(),
            },
        );
        assert!(!external.is_local());
        assert!(external.is_remote());
        assert!(!external.requires_apple_silicon());
        let info = ModelInfo::from(&external);
        assert!(!info.is_local);
        assert!(
            !info.downloads_weights,
            "external vllm-mlx owns its weights, so CAR installs nothing"
        );
    }

    #[test]
    fn model_info_classifies_only_raw_vllm_mlx_as_operator_managed_external() {
        for endpoint in ["http://localhost:8000", "https://models.example.invalid/v1"] {
            let schema = test_generate_schema(
                "external/model",
                "external-model",
                ModelSource::VllmMlx {
                    endpoint: endpoint.into(),
                    model_name: "mlx-community/external-model".into(),
                },
            );
            let info = ModelInfo::from(&schema);
            assert!(info.operator_managed_external_runtime);
            assert_eq!(
                serde_json::to_value(info).unwrap()["operator_managed_external_runtime"],
                true
            );
        }

        for source in [
            ModelSource::RemoteApi {
                endpoint: "https://cloud.example.invalid/v1".into(),
                api_key_env: "CAR_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
            ModelSource::ManagedVllmMlx {
                hf_repo: "mlx-community/car-owned-model".into(),
                hf_weight_file: None,
            },
        ] {
            assert!(
                !ModelInfo::from(&test_generate_schema(
                    "not-external/model",
                    "not-external-model",
                    source,
                ))
                .operator_managed_external_runtime
            );
        }
    }

    /// The fresh-machine state from Parslee-ai/car#894, end to end: an MLX
    /// entry with a declared `hf_repo` and an EMPTY models dir. It is
    /// `available` (ensure_local lazy-downloads on first use, #164) and
    /// simultaneously NOT `weights_ready` (nothing has been fetched). Those
    /// two facts are what `car models list` and `car doctor` were each
    /// reporting in isolation, which is why they disagreed.
    #[test]
    fn fresh_machine_mlx_entry_is_available_but_not_weights_ready() {
        let mut reg = test_registry();
        let id = "mlx-community/car894-fresh-4bit";
        reg.register(test_generate_schema(
            id,
            "car894-fresh-4bit",
            ModelSource::Mlx {
                hf_repo: "mlx-community/car894-fresh-4bit".into(),
                hf_weight_file: None,
            },
        ));

        let registered = reg
            .get(id)
            .expect("the model just registered must be in the registry");
        let info = ModelInfo::from(registered);

        // The models dir is a fresh TempDir, so nothing is downloaded.
        assert!(
            !registered.weights_ready,
            "an empty models dir means no weights on disk"
        );
        assert!(
            !info.weights_ready,
            "the CLI-facing projection must report the same: nothing installed"
        );

        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        {
            // On a real MLX target this is the exact contradiction #894
            // reported: runnable here, but not installed.
            assert!(
                registered.available,
                "a declared hf_repo makes an MLX entry runnable before download (#164)"
            );
            assert!(
                info.available,
                "the projection must keep reporting it as runnable"
            );
        }
        #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
        {
            // Off an MLX target the backend isn't compiled in, so it is
            // neither runnable nor installed — still not a contradiction.
            assert!(
                !registered.available,
                "MLX cannot execute on this target, so it must not be runnable"
            );
            assert!(!info.available);
        }
    }

    fn write_complete_mlx_snapshot(dir: &Path) {
        std::fs::create_dir_all(dir).unwrap();
        std::fs::write(dir.join("config.json"), b"{}").unwrap();
        std::fs::write(dir.join("tokenizer.json"), b"{}").unwrap();
        std::fs::write(
            dir.join("model.safetensors.index.json"),
            r#"{"weight_map":{"a":"model-00001-of-00002.safetensors","b":"model-00002-of-00002.safetensors"}}"#,
        )
        .unwrap();
        std::fs::write(dir.join("model-00001-of-00002.safetensors"), b"one").unwrap();
        std::fs::write(dir.join("model-00002-of-00002.safetensors"), b"two").unwrap();
    }

    #[test]
    fn complete_managed_and_shared_mlx_snapshots_are_physically_ready() {
        let schema = test_generate_schema(
            "mlx/qwen3-4b:4bit",
            "Qwen3-4B-MLX",
            ModelSource::Mlx {
                hf_repo: "mlx-community/Qwen3-4B-4bit".into(),
                hf_weight_file: None,
            },
        );
        let root = tempfile::tempdir().unwrap();
        let managed = root.path().join("managed");
        let shared = root.path().join("shared");

        write_complete_mlx_snapshot(&managed);
        assert!(mlx_weights_ready_at(&schema, &managed, None));

        std::fs::remove_dir_all(&managed).unwrap();
        write_complete_mlx_snapshot(&shared);
        assert!(mlx_weights_ready_at(&schema, &managed, Some(&shared)));
    }

    #[test]
    fn zero_byte_gguf_is_not_physically_ready() {
        let schema = test_generate_schema(
            "qwen/qwen3-4b:q4_k_m",
            "Qwen3-4B",
            ModelSource::Local {
                hf_repo: "Qwen/Qwen3-4B-GGUF".into(),
                hf_filename: "model.gguf".into(),
                tokenizer_repo: "Qwen/Qwen3-4B".into(),
            },
        );
        let root = tempfile::tempdir().unwrap();
        let model_dir = root.path().join(&schema.name);
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(model_dir.join("model.gguf"), b"").unwrap();

        assert!(!physical_weights_ready(&schema, root.path()));
        std::fs::write(model_dir.join("model.gguf"), b"gguf").unwrap();
        assert!(physical_weights_ready(&schema, root.path()));
    }

    #[test]
    fn shared_mlx_snapshot_missing_an_indexed_shard_is_not_physically_ready() {
        let schema = test_generate_schema(
            "mlx/qwen3-8b:4bit",
            "Qwen3-8B-MLX",
            ModelSource::Mlx {
                hf_repo: "mlx-community/Qwen3-8B-4bit".into(),
                hf_weight_file: None,
            },
        );
        let root = tempfile::tempdir().unwrap();
        let managed = root.path().join("managed");
        let shared = root.path().join("shared");
        write_complete_mlx_snapshot(&shared);
        std::fs::remove_file(shared.join("model-00001-of-00002.safetensors")).unwrap();

        assert!(!mlx_weights_ready_at(&schema, &managed, Some(&shared)));
    }

    /// `refresh_availability` must probe each DISTINCT credential once, not
    /// once per model (car-releases#75).
    ///
    /// A credential probe is a keychain query on macOS — ~14 ms. The stock
    /// catalog is ~72 models over roughly a dozen providers, so probing
    /// per-model cost ~1 s per refresh; a snapshot runs per request, and
    /// `estimated_tokens` ran a second one, so a delegated call waited ~2 s
    /// before it ever reached the host's runner.
    ///
    /// Asserted by COUNTING probes rather than by timing: a latency assertion
    /// would be flaky on a loaded CI runner, and the invariant that actually
    /// matters is "O(providers), not O(models)".
    /// Parslee-ai/car#786 — the catalog must stop advertising a managed
    /// namespace once the gateway has said it has no upstream for it.
    ///
    /// This is the reported symptom directly: `car models list` showed all ten
    /// `parslee/openrouter/*` aliases as `avail=yes` while every one of them
    /// 503'd, because availability for those rows resolves to "is a Parslee
    /// session signed in" and nothing more. A harness picked one on the
    /// strength of that claim and lost a full benchmark sweep to it.
    ///
    /// Asserted through the real `curated_schemas()` rows rather than a
    /// fixture, because the bug lives in how THOSE rows derive availability.
    ///
    /// Isolates `CAR_HOME` for the same reason its credential sibling does.
    /// `note_gateway_unconfigured` writes `gateway-state.json` under the state
    /// root, and cleaning up afterwards is not enough: nextest runs every test
    /// in its own process, so while this one holds the flag a DIFFERENT process
    /// can load it and fail on an availability assertion it never made
    /// (Parslee-ai/car#986). The crate's serial lock cannot serialize across
    /// processes; a private state root can.
    #[test]
    fn a_gateway_that_reports_no_upstream_stops_being_advertised() {
        let _guard = crate::openrouter::test_environment_scope();
        // See the note above: `StateRootScope` is the RAII form, so a panic in
        // an assertion below cannot leave `CAR_HOME` pointing at a deleted dir.
        let _home = StateRootScope::new();
        crate::openrouter::clear_gateway_unconfigured();

        let managed: Vec<ModelSchema> = crate::openrouter::curated_schemas()
            .into_iter()
            .filter(|s| crate::openrouter::is_curated_managed_gateway_alias(&s.id))
            .collect();
        assert!(
            !managed.is_empty(),
            "precondition: the curated catalog must still carry managed aliases"
        );

        let availability_of = |schema: &ModelSchema| match &schema.source {
            ModelSource::Proprietary { provider, auth, .. } => proprietary_auth_available(
                &schema.id,
                &schema.provider,
                provider,
                auth,
                // Signed in — the state in which the bug reported `yes`.
                true,
                &std::collections::HashMap::new(),
            ),
            other => panic!("managed aliases must be Proprietary, got {other:?}"),
        };

        assert!(
            managed.iter().all(availability_of),
            "precondition: an authenticated session advertises these today"
        );

        crate::openrouter::note_gateway_unconfigured();
        assert!(
            managed.iter().all(|s| !availability_of(s)),
            "after the gateway says it has no OpenRouter upstream, every alias in \
             the namespace must report unavailable — that claim is what cost the \
             benchmark sweep in #786"
        );

        // Self-correcting: forgetting the observation restores the optimistic
        // claim, so an environment that later gains OpenRouter is not
        // permanently written off.
        crate::openrouter::clear_gateway_unconfigured();
        assert!(
            managed.iter().all(availability_of),
            "the suppression must be recoverable, not a one-way latch"
        );
    }

    /// Parslee-ai/car#986 — building a registry must not clear the gateway
    /// observation as a side effect of construction.
    ///
    /// `new_with_catalog_public_key` ends in a `refresh_availability`, and that
    /// refresh forgets the session-scoped evidence whenever no Parslee session
    /// is present. #989 took the crate's serial lock in the two tests that call
    /// `refresh_availability` directly, but roughly twenty others reach the same
    /// clear just by constructing a registry, and none of them took the lock —
    /// so the class of bug survived the fix that was supposed to close it.
    ///
    /// This first case pins the harmless direction: a live session, so nothing
    /// is forgotten. `CAR_HOME` is isolated so the durable half of the
    /// observation lands in a temp dir rather than the developer's `~/.car`.
    #[test]
    fn constructing_a_registry_with_a_live_session_leaves_the_gateway_observation_alone() {
        let _guard = crate::openrouter::test_environment_scope();
        let home = StateRootScope::new();

        crate::openrouter::note_gateway_unconfigured();
        let _registry = UnifiedRegistry::new_with_session(
            home.path().to_path_buf(),
            home.path().join("models"),
            None,
            SessionProbe::Fixed(true),
        );

        let observed = crate::openrouter::gateway_unconfigured();
        let persisted = crate::openrouter::gateway_state_path().exists();

        crate::openrouter::clear_gateway_unconfigured();

        assert!(
            observed,
            "a signed-in session has no reason to forget what the gateway said"
        );
        assert!(
            persisted,
            "the durable half of the observation must survive construction too"
        );
    }

    /// The other half of Parslee-ai/car#986: the production guarantee from
    /// Parslee-ai/car#786 is unchanged. When the session really is gone, the
    /// evidence learned about it is discarded — in memory and on disk — because
    /// the next sign-in may be to an org that has OpenRouter configured.
    ///
    /// Pinned rather than bypassed: the fix narrows *who* may forget, not
    /// whether forgetting still happens.
    #[test]
    fn constructing_a_registry_with_no_session_still_forgets_the_gateway_observation() {
        let _guard = crate::openrouter::test_environment_scope();
        let home = StateRootScope::new();

        crate::openrouter::note_gateway_unconfigured();
        let recorded = crate::openrouter::gateway_unconfigured();
        let _registry = UnifiedRegistry::new_with_session(
            home.path().to_path_buf(),
            home.path().join("models"),
            None,
            SessionProbe::Fixed(false),
        );

        let observed = crate::openrouter::gateway_unconfigured();
        let persisted = crate::openrouter::gateway_state_path().exists();

        crate::openrouter::clear_gateway_unconfigured();

        assert!(
            recorded,
            "precondition: the observation is on record before construction"
        );
        assert!(
            !observed,
            "sign-out must still discard the session-scoped verdict (#786)"
        );
        assert!(
            !persisted,
            "and the durable copy with it — otherwise the next sign-in inherits it from disk"
        );
    }

    /// The case that actually failed before Parslee-ai/car#986 was fixed: an
    /// ordinary test registry, built the way every other test in this crate
    /// builds one, with no serial lock and no opinion about auth.
    ///
    /// On a machine with no Parslee session — a CI runner — construction alone
    /// used to wipe the observation, which is how
    /// `a_gateway_that_reports_no_upstream_stops_being_advertised` lost its flag
    /// between the `note_` call and the assertion under `cargo test`.
    ///
    /// Two assertions, deliberately. The observation one reproduces the CI
    /// condition and is the regression proper — but it only *fails* on main
    /// when the runner is signed out, so on a developer Mac with a live
    /// keychain session it would pass either way. The probe one holds
    /// regardless of the machine: the default a test registry is built with
    /// must be one that cannot forget session evidence, which is the whole
    /// content of the fix.
    #[test]
    fn an_ordinary_test_registry_does_not_disturb_a_separately_set_observation() {
        let _guard = crate::openrouter::test_environment_scope();
        let home = StateRootScope::new();

        crate::openrouter::note_gateway_unconfigured();
        let _registry = UnifiedRegistry::new_with_state_root(
            home.path().to_path_buf(),
            home.path().join("models"),
        );

        let observed = crate::openrouter::gateway_unconfigured();

        crate::openrouter::clear_gateway_unconfigured();

        assert!(
            observed,
            "constructing a registry is not a statement about the session, so it \
             must not erase an observation another test just recorded (#986)"
        );
        assert!(
            !SessionProbe::Inert.may_forget_session_evidence(),
            "the `cfg(test)` construction default must be a probe that answers \
             the session question without acting on it — this is the half of \
             the guarantee that does not depend on whether the runner happens \
             to be signed in"
        );
    }

    /// Parslee-ai/car#887 — a signed-in session is not a WORKING one, and the
    /// catalog must stop advertising `parslee/*` once the server has rejected
    /// the credential.
    ///
    /// The reported symptom: on a machine with a stale sign-in, every routing
    /// pass offered the managed aliases as top-quality candidates and every
    /// request 401'd, so the router walked the fallback chain silently on every
    /// call. `access_token_is_available()` is existence-only, so `true` is
    /// exactly the state the bug reported as available.
    ///
    /// Isolates `CAR_HOME` rather than writing the observation into the
    /// developer's real `~/.car`; the sibling gateway test now does the same.
    #[test]
    fn a_rejected_credential_stops_the_managed_lane_being_advertised() {
        let _guard = crate::openrouter::test_environment_scope();
        let _home = StateRootScope::new();
        crate::parslee_credential::clear_credential_rejected();

        let managed: Vec<ModelSchema> = crate::openrouter::curated_schemas()
            .into_iter()
            .filter(|s| s.provider == "parslee")
            .collect();
        assert!(
            !managed.is_empty(),
            "precondition: the curated catalog must still carry parslee rows"
        );

        let availability_of = |schema: &ModelSchema| match &schema.source {
            ModelSource::Proprietary { provider, auth, .. } => proprietary_auth_available(
                &schema.id,
                &schema.provider,
                provider,
                auth,
                // Signed in — the state in which the bug reported `yes`.
                true,
                &std::collections::HashMap::new(),
            ),
            other => panic!("parslee rows must be Proprietary, got {other:?}"),
        };

        assert!(
            managed.iter().all(availability_of),
            "precondition: an authenticated session advertises these today"
        );

        crate::parslee_credential::note_credential_rejected();
        assert!(
            managed.iter().all(|s| !availability_of(s)),
            "after the server rejects the credential, EVERY parslee row must \
             report unavailable — unlike the gateway verdict this is not scoped \
             to the curated OpenRouter aliases, because a dead credential kills \
             the whole namespace"
        );

        // Recoverable, not a one-way latch: the operator signs in again, the
        // next org lookup succeeds and clears the verdict.
        crate::parslee_credential::clear_credential_rejected();
        assert!(
            managed.iter().all(availability_of),
            "the suppression must lift once the credential works again"
        );
    }

    #[test]
    fn refresh_availability_probes_each_credential_once_not_per_model() {
        // Serialized with the other tests that touch the gateway observation.
        // `refresh_availability` CLEARS it when no Parslee token is present
        // (see the `!parslee_oauth_available` branch above), which is exactly
        // the state a CI runner is in — so without this lock, running under
        // `cargo test` (one process per crate target, tests threaded in
        // parallel inside it) this test can wipe the flag that
        // `a_gateway_that_reports_no_upstream_stops_being_advertised` set,
        // between its `note_` call and its assertion. `cargo nextest` hides it
        // by giving every test its own process, and every per-PR job runs
        // nextest — which is why it surfaced in `check-windows` instead
        // (Parslee-ai/car#986). See docs/solutions/process-global-state-in-tests.md
        // for the family and the full list of shared-process CI runs.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_empty(tmp.path().join("models"));
        for i in 0..25 {
            let mut schema = test_generate_schema(
                &format!("openrouter/model-{i}"),
                &format!("model-{i}"),
                ModelSource::RemoteApi {
                    protocol: crate::schema::ApiProtocol::OpenRouter,
                    endpoint: "https://openrouter.ai/api/v1".into(),
                    api_key_env: "OPENROUTER_API_KEY".into(),
                    api_key_envs: vec![],
                    api_version: None,
                },
            );
            schema.provider = "openrouter".into();
            registry.register(schema);
        }

        crate::openrouter::reset_credential_source_call_count();
        registry.refresh_availability();
        let calls = crate::openrouter::credential_source_call_count();

        assert_eq!(
            calls, 1,
            "refresh_availability probed the OpenRouter credential {calls} times for 25 models; \
             it must resolve each distinct credential once per refresh, not once per model"
        );
    }

    /// `pull` on a `vllm-mlx` model must report the HuggingFace cache directory
    /// the external runtime will actually open — CAR's own
    /// `~/.car/models/<name>` layout is the wrong shape for a model a Python
    /// runtime resolves by repo id, and reporting the wrong path would make
    /// "where did my 16 GB go" unanswerable.
    #[test]
    fn a_vllm_mlx_pull_targets_the_shared_huggingface_cache() {
        let repo = "mlx-community/Qwen3.8-27B-4bit";
        let dir = huggingface_repo_dir(repo);
        assert!(
            dir.ends_with("models--mlx-community--Qwen3.8-27B-4bit"),
            "got {}",
            dir.display()
        );
        assert!(
            dir.parent().is_some_and(|p| p.ends_with("hub")),
            "must live under the HF cache's hub/ root, got {}",
            dir.display()
        );
    }

    /// A raw `VllmMlx` row is operator-managed, even when its endpoint is a
    /// loopback URL. Only `ManagedVllmMlx` opts into CAR's provisioned runtime.
    #[test]
    fn external_vllm_mlx_does_not_become_managed_from_a_loopback_endpoint() {
        // Serialized with the other tests that touch the gateway observation.
        // `refresh_availability` CLEARS it when no Parslee token is present
        // (see the `!parslee_oauth_available` branch above), which is exactly
        // the state a CI runner is in — so without this lock, running under
        // `cargo test` (one process per crate target, tests threaded in
        // parallel inside it) this test can wipe the flag that
        // `a_gateway_that_reports_no_upstream_stops_being_advertised` set,
        // between its `note_` call and its assertion. `cargo nextest` hides it
        // by giving every test its own process, and every per-PR job runs
        // nextest — which is why it surfaced in `check-windows` instead
        // (Parslee-ai/car#986). See docs/solutions/process-global-state-in-tests.md
        // for the family and the full list of shared-process CI runs.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_empty(tmp.path().join("models"));
        let schema = test_generate_schema(
            "vllm-mlx/arch-the-rust-backend-cannot-load",
            "external-only-model",
            ModelSource::VllmMlx {
                endpoint: "http://localhost:8000".into(),
                model_name: "mlx-community/Qwen3.8-27B-4bit".into(),
            },
        );
        registry.register(schema);

        // Deliberately NOT set: no external endpoint has been declared healthy.
        assert!(
            std::env::var("VLLM_MLX_ENDPOINT").is_err(),
            "test precondition: VLLM_MLX_ENDPOINT must be unset"
        );
        registry.refresh_availability();

        let model = registry
            .get("vllm-mlx/arch-the-rust-backend-cannot-load")
            .expect("registered model should be present");
        assert!(
            !model.available,
            "an external vllm-mlx row remains external even on loopback; only an \
             explicit ManagedVllmMlx source may use CAR's runtime"
        );
    }

    #[test]
    fn user_config_load_and_save_force_community_trust() {
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let config_path = tmp.path().join("models.json");
        let schema = test_generate_schema(
            "user/test-model",
            "user-test-model",
            ModelSource::RemoteApi {
                endpoint: "https://attacker.invalid/v1/chat/completions".into(),
                api_key_env: "CAR_USER_MODEL_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        );
        let mut omitted_tier = serde_json::to_value(schema.clone()).unwrap();
        omitted_tier.as_object_mut().unwrap().remove("trust_tier");
        std::fs::write(
            &config_path,
            serde_json::to_vec_pretty(&vec![omitted_tier]).unwrap(),
        )
        .unwrap();

        let mut loaded = UnifiedRegistry::new_empty(models_dir.clone());
        loaded.load_user_config().unwrap();
        assert_eq!(
            loaded.get("user/test-model").unwrap().trust_tier,
            crate::schema::TrustTier::Community
        );

        let mut persisted = UnifiedRegistry::new_empty(models_dir);
        persisted.register_user_model(schema);
        persisted.save_user_config().unwrap();
        let saved: Vec<ModelSchema> =
            serde_json::from_slice(&std::fs::read(config_path).unwrap()).unwrap();
        assert_eq!(saved.len(), 1);
        assert_eq!(saved[0].trust_tier, crate::schema::TrustTier::Community);
    }

    #[test]
    fn persisted_user_model_cannot_shadow_managed_openrouter_alias() {
        // Constructing a `UnifiedRegistry` ends in `refresh_availability()`,
        // which WRITES the process-global gateway observation. Full reasoning
        // on `TestRegistry` in this crate; family in
        // docs/solutions/process-global-state-in-tests.md.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let config_path = tmp.path().join("models.json");
        let mut shadow = crate::openrouter::curated_schemas()
            .into_iter()
            .find(|schema| schema.id == "parslee/openrouter/frontier-general")
            .unwrap();
        shadow.provider = "attacker".into();
        std::fs::write(
            &config_path,
            serde_json::to_vec_pretty(&vec![shadow]).unwrap(),
        )
        .unwrap();

        let registry = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models_dir);
        let actual = registry
            .get("parslee/openrouter/frontier-general")
            .expect("compiled managed alias must remain present");
        assert_eq!(actual.provider, "parslee");
        assert_eq!(
            crate::openrouter::canonical_managed_gateway_selector(actual),
            Some("parslee/openrouter/frontier-general")
        );
    }

    #[test]
    fn user_config_persistence_excludes_signed_rows_and_keeps_builtin_tagged_user_rows() {
        // Constructing a `UnifiedRegistry` ends in `refresh_availability()`,
        // which WRITES the process-global gateway observation. Full reasoning
        // on `TestRegistry` in this crate; family in
        // docs/solutions/process-global-state-in-tests.md.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let config_path = tmp.path().join("models.json");

        let signed = test_generate_schema(
            "signed/catalog-only",
            "signed-catalog-only",
            ModelSource::RemoteApi {
                endpoint: "https://catalog.example/v1".into(),
                api_key_env: "SIGNED_CATALOG_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        );
        assert!(!signed.tags.iter().any(|tag| tag == "builtin"));
        let (verified, public_key) = crate::catalog::signed_test_catalog(
            crate::catalog::CatalogDoc {
                version: 81,
                models: vec![signed],
            },
            81,
        );
        crate::catalog::save_verified(&crate::catalog::cache_path(tmp.path()), &verified).unwrap();

        let mut registry = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            models_dir.clone(),
            Some(public_key.as_str()),
        );
        let mut user = test_generate_schema(
            "user/builtin-tagged",
            "user-builtin-tagged",
            ModelSource::RemoteApi {
                endpoint: "https://user.example/v1".into(),
                api_key_env: "USER_MODEL_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        );
        user.tags.push("builtin".into());
        registry.register_user_model(user);
        registry.save_user_config().unwrap();

        let saved: Vec<ModelSchema> =
            serde_json::from_slice(&std::fs::read(&config_path).unwrap()).unwrap();
        assert_eq!(
            saved
                .iter()
                .map(|model| model.id.as_str())
                .collect::<Vec<_>>(),
            vec!["user/builtin-tagged"],
            "models.json must contain only explicitly user-registered rows"
        );
        assert_eq!(saved[0].trust_tier, crate::schema::TrustTier::Community);

        let mut restarted = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            models_dir,
            Some(public_key.as_str()),
        );
        assert_eq!(
            restarted.get("signed/catalog-only").unwrap().trust_tier,
            crate::schema::TrustTier::Curated,
            "user persistence must not demote an unrelated signed catalog row"
        );
        assert_eq!(
            restarted.get("user/builtin-tagged").unwrap().trust_tier,
            crate::schema::TrustTier::Community
        );
        let mut signed_shadow = restarted.get("signed/catalog-only").unwrap().clone();
        signed_shadow.name = "user-shadow-of-signed-row".into();
        restarted.register_user_model(signed_shadow);
        assert_eq!(
            restarted.get("signed/catalog-only").unwrap().name,
            "signed-catalog-only",
            "a user row must not shadow a signature-verified project exact id"
        );
    }

    #[test]
    fn empty_user_config_save_clears_stale_rows() {
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let config_path = tmp.path().join("models.json");
        let stale = test_generate_schema(
            "user/stale",
            "stale",
            ModelSource::RemoteApi {
                endpoint: "https://stale.example/v1".into(),
                api_key_env: "STALE_USER_MODEL_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        );
        std::fs::write(
            &config_path,
            serde_json::to_vec_pretty(&vec![stale]).unwrap(),
        )
        .unwrap();

        UnifiedRegistry::new_empty(models_dir)
            .save_user_config()
            .unwrap();

        let saved: Vec<ModelSchema> =
            serde_json::from_slice(&std::fs::read(config_path).unwrap()).unwrap();
        assert!(
            saved.is_empty(),
            "saving an empty user set must overwrite stale models.json rows"
        );
    }

    #[test]
    fn unregister_then_save_removes_the_user_row_from_disk() {
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let config_path = tmp.path().join("models.json");
        let mut registry = UnifiedRegistry::new_empty(models_dir);
        registry.register_project_model(test_generate_schema(
            "signed/not-user-removable",
            "not-user-removable",
            ModelSource::RemoteApi {
                endpoint: "https://catalog.example/v1".into(),
                api_key_env: "SIGNED_CATALOG_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        ));
        assert!(
            registry
                .unregister_user_model("signed/not-user-removable")
                .is_none(),
            "the user boundary cannot unregister an untracked catalog row"
        );
        assert!(registry.get("signed/not-user-removable").is_some());
        registry.register_user_model(test_generate_schema(
            "user/removable",
            "removable",
            ModelSource::RemoteApi {
                endpoint: "https://user.example/v1".into(),
                api_key_env: "REMOVABLE_USER_MODEL_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        ));
        registry.save_user_config().unwrap();
        assert!(registry.unregister_user_model("user/removable").is_some());
        registry.save_user_config().unwrap();

        let saved: Vec<ModelSchema> =
            serde_json::from_slice(&std::fs::read(config_path).unwrap()).unwrap();
        assert!(saved.is_empty());
    }

    /// The daemon's `models.register` handler writes through
    /// [`user_config_path`]; the registry reads through the state root it was
    /// built with. Those two resolvers must land on one file.
    ///
    /// They did not. The write side moved to `CAR_HOME` while the read side
    /// stayed `<models_dir>/../models.json` — and `models_dir` is deliberately
    /// the machine-shared weights cache, which does NOT move. So against a
    /// relocated daemon the registration was persisted to
    /// `$CAR_HOME/models.json`, the next boot read `~/.car/models.json`, and
    /// the model silently never appeared. No error anywhere; the handler even
    /// reported the path it had written.
    #[test]
    fn a_registration_written_under_car_home_is_the_file_the_registry_reads() {
        let _environment = crate::openrouter::test_environment_scope();
        let prior = std::env::var_os(car_home::ENV_VAR);

        let state_root = TempDir::new().unwrap();
        // Deliberately NOT inside the state root — this stands in for the
        // machine-shared `~/.car/models`, which stays put under `CAR_HOME`.
        let weights = TempDir::new().unwrap();
        let models_dir = weights.path().join("models");
        std::fs::create_dir_all(&models_dir).unwrap();

        unsafe { std::env::set_var(car_home::ENV_VAR, state_root.path()) };

        // Exactly what `handle_models_register` resolves and writes.
        let write_path = user_config_path().expect("CAR_HOME must resolve a models.json path");
        assert_eq!(write_path, state_root.path().join(USER_MODELS_FILE));
        let registered = test_generate_schema(
            "user/relocated-daemon-model",
            "relocated-daemon-model",
            ModelSource::RemoteApi {
                endpoint: "https://relocated.example/v1".into(),
                api_key_env: "RELOCATED_DAEMON_MODEL_TEST_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
            },
        );
        std::fs::write(
            &write_path,
            serde_json::to_vec_pretty(&vec![registered]).unwrap(),
        )
        .unwrap();

        // …and exactly what the next daemon boot constructs.
        let registry = UnifiedRegistry::new(models_dir.clone());

        match prior {
            Some(value) => unsafe { std::env::set_var(car_home::ENV_VAR, value) },
            None => unsafe { std::env::remove_var(car_home::ENV_VAR) },
        }

        assert!(
            registry.get("user/relocated-daemon-model").is_some(),
            "the registry must load the models.json that `models.register` wrote; \
             it looked at {} instead",
            registry.user_config_path.display(),
        );
        assert_eq!(registry.user_config_path, write_path);
        assert!(
            !weights.path().join(USER_MODELS_FILE).exists(),
            "nothing may be written beside the shared weights cache",
        );
    }

    #[test]
    fn ready_without_download_is_strict_for_local_model_files() {
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        let mut reg = UnifiedRegistry::new_empty(models.clone());
        reg.register(test_generate_schema(
            "local/test",
            "TestLocal",
            ModelSource::Local {
                hf_repo: "example/repo".into(),
                hf_filename: "model.gguf".into(),
                tokenizer_repo: "example/repo".into(),
            },
        ));

        assert_eq!(reg.ready_without_download("local/test"), Some(false));

        let dir = models.join("TestLocal");
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("model.gguf"), b"weights").unwrap();
        assert_eq!(
            reg.ready_without_download("local/test"),
            Some(false),
            "tokenizer is required too"
        );
        std::fs::write(dir.join("tokenizer.json"), "{}").unwrap();
        assert_eq!(reg.ready_without_download("local/test"), Some(true));
    }

    #[test]
    fn ready_without_download_rejects_mlx_config_only_stub() {
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        let mut reg = UnifiedRegistry::new_empty(models.clone());
        reg.register(test_generate_schema(
            "mlx/test",
            "TestMlx",
            ModelSource::Mlx {
                hf_repo: "example/repo".into(),
                hf_weight_file: None,
            },
        ));

        let dir = models.join("TestMlx");
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("config.json"), "{}").unwrap();
        std::fs::write(dir.join("tokenizer.json"), "{}").unwrap();
        assert_eq!(
            reg.ready_without_download("mlx/test"),
            Some(false),
            "config/tokenizer stubs must not start assistant inference"
        );

        std::fs::write(dir.join("model.safetensors"), b"weights").unwrap();
        assert_eq!(reg.ready_without_download("mlx/test"), Some(true));
    }

    fn write_mlx_dir(root: &Path, name: &str, model_type: &str) {
        let dir = root.join(name);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(
            dir.join("config.json"),
            serde_json::json!({
                "model_type": model_type,
                "max_position_embeddings": 40_960,
                "quantization": { "bits": 8, "group_size": 32, "mode": "mxfp8" },
            })
            .to_string(),
        )
        .unwrap();
        std::fs::write(dir.join("model.safetensors"), b"weights").unwrap();
    }

    struct ScopedEnvVar {
        name: &'static str,
        previous: Option<std::ffi::OsString>,
    }

    impl ScopedEnvVar {
        fn set(name: &'static str, value: &Path) -> Self {
            let previous = std::env::var_os(name);
            // SAFETY: acquisition tests hold `test_environment_scope_async` for
            // the lifetime of this process-global environment mutation.
            unsafe { std::env::set_var(name, value) };
            Self { name, previous }
        }
    }

    impl Drop for ScopedEnvVar {
        fn drop(&mut self) {
            // SAFETY: the environment mutex is still held because this guard is
            // declared after it and therefore drops first.
            unsafe {
                match self.previous.take() {
                    Some(value) => std::env::set_var(self.name, value),
                    None => std::env::remove_var(self.name),
                }
            }
        }
    }

    #[derive(Default)]
    struct AcquisitionRecorder {
        events: std::sync::Mutex<Vec<DownloadEvent>>,
        replace_dir_on_started: std::sync::Mutex<Option<PathBuf>>,
    }

    impl AcquisitionRecorder {
        fn replacing_dir_on_started(path: PathBuf) -> Self {
            Self {
                events: std::sync::Mutex::new(Vec::new()),
                replace_dir_on_started: std::sync::Mutex::new(Some(path)),
            }
        }

        fn events(&self) -> Vec<DownloadEvent> {
            self.events.lock().unwrap().clone()
        }
    }

    impl crate::download::DownloadProgress for AcquisitionRecorder {
        fn on_event(&self, event: &DownloadEvent) {
            self.events.lock().unwrap().push(event.clone());
            if matches!(event, DownloadEvent::Started { .. }) {
                if let Some(path) = self.replace_dir_on_started.lock().unwrap().take() {
                    std::fs::remove_dir_all(&path).unwrap();
                    std::fs::write(path, b"make create_dir_all fail before any network access")
                        .unwrap();
                }
            }
        }
    }

    fn started_count(events: &[DownloadEvent]) -> usize {
        events
            .iter()
            .filter(|event| matches!(event, DownloadEvent::Started { .. }))
            .count()
    }

    fn mlx_schema(id: &str, name: &str, hf_repo: &str) -> ModelSchema {
        test_generate_schema(
            id,
            name,
            ModelSource::Mlx {
                hf_repo: hf_repo.into(),
                hf_weight_file: None,
            },
        )
    }

    #[test]
    fn installing_a_fetched_file_replaces_only_unusable_destinations() {
        let tmp = TempDir::new().unwrap();
        let src = tmp.path().join("fetched");
        std::fs::write(&src, b"fetched bytes").unwrap();

        let missing = tmp.path().join("missing");
        install_fetched_file(&src, &missing).unwrap();
        assert_eq!(std::fs::read(&missing).unwrap(), b"fetched bytes");

        let usable = tmp.path().join("usable");
        std::fs::write(&usable, b"keep these bytes").unwrap();
        let usable_before = std::fs::symlink_metadata(&usable).unwrap();
        let modified_before = usable_before.modified().unwrap();
        #[cfg(unix)]
        let inode_before = {
            use std::os::unix::fs::MetadataExt;
            usable_before.ino()
        };
        install_fetched_file(&src, &usable).unwrap();
        let usable_after = std::fs::symlink_metadata(&usable).unwrap();
        assert_eq!(std::fs::read(&usable).unwrap(), b"keep these bytes");
        assert_eq!(usable_after.modified().unwrap(), modified_before);
        #[cfg(unix)]
        {
            use std::os::unix::fs::MetadataExt;
            assert_eq!(usable_after.ino(), inode_before);
        }

        let zero_byte = tmp.path().join("zero-byte");
        std::fs::write(&zero_byte, b"").unwrap();
        install_fetched_file(&src, &zero_byte).unwrap();
        assert_eq!(std::fs::read(&zero_byte).unwrap(), b"fetched bytes");

        #[cfg(unix)]
        {
            let dangling = tmp.path().join("dangling");
            std::os::unix::fs::symlink(tmp.path().join("absent"), &dangling).unwrap();
            assert!(std::fs::symlink_metadata(&dangling).unwrap().is_symlink());
            install_fetched_file(&src, &dangling).unwrap();
            assert_eq!(std::fs::read(&dangling).unwrap(), b"fetched bytes");
        }
    }

    // hf-hub's default cache resolves the OS home, not `$HOME`, on Windows.
    // Keep this redirected offline fixture out of nightly Windows lib tests.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_zero_byte_flux_auxiliary_file_is_not_accepted_as_present() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let home = tmp.path().join("home");
        let _home = ScopedEnvVar::set("HOME", &home);
        let _hf_home = ScopedEnvVar::set("HF_HOME", &tmp.path().join("hf-home"));

        // `download_file` uses hf-hub's default HOME cache. Populate the exact
        // base-model pointer it asks for so this regression is fully offline.
        let cached_auxiliary = home
            .join(".cache/huggingface/hub")
            .join("models--Freepik--flux.1-lite-8B")
            .join("snapshots/fixture/tokenizer_2/tokenizer.json");
        std::fs::create_dir_all(cached_auxiliary.parent().unwrap()).unwrap();
        std::fs::write(&cached_auxiliary, b"repaired tokenizer").unwrap();
        let refs = home
            .join(".cache/huggingface/hub")
            .join("models--Freepik--flux.1-lite-8B/refs");
        std::fs::create_dir_all(&refs).unwrap();
        std::fs::write(refs.join("main"), b"fixture").unwrap();

        let models = tmp.path().join("models");
        let name = "Flux-1.lite-8B-MLX-Q4";
        write_mlx_dir(&models, name, "flux");
        let auxiliary = models.join(name).join("tokenizer_2/tokenizer.json");
        std::fs::create_dir_all(auxiliary.parent().unwrap()).unwrap();
        std::fs::write(&auxiliary, b"").unwrap();
        let mut reg = UnifiedRegistry::new_empty(models.clone());
        reg.register(mlx_schema(
            "mlx/flux-zero-byte-aux",
            name,
            "mlx-community/Flux-1.lite-8B-MLX-Q4",
        ));
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let result = reg
            .acquire_and_ensure("mlx/flux-zero-byte-aux", &sink, false, None)
            .await;
        let events = recorder.events();
        assert!(matches!(
            events.first(),
            Some(DownloadEvent::Started { .. })
        ));
        assert_eq!(started_count(&events), 1);
        assert!(
            result.is_err() || crate::download::cache_file_usable(&auxiliary),
            "acquisition must replace the zero-byte auxiliary file or report failure"
        );
        if result.is_ok() {
            assert_eq!(std::fs::read(auxiliary).unwrap(), b"repaired tokenizer");
        }
    }

    // hf-hub's default cache resolves the OS home, not `$HOME`, on Windows.
    // Keep this redirected offline fixture out of nightly Windows lib tests.
    #[cfg(unix)]
    #[tokio::test]
    async fn managed_flux_missing_auxiliary_does_not_fall_through_to_hf_snapshot() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let home = tmp.path().join("home");
        let hf_home = tmp.path().join("hf-home");
        let _home = ScopedEnvVar::set("HOME", &home);
        let _hf_home = ScopedEnvVar::set("HF_HOME", &hf_home);

        // The managed repair fetches Flux's auxiliary from the base model.
        // Populate hf-hub's HOME cache so that repair succeeds offline.
        let cached_auxiliary = home
            .join(".cache/huggingface/hub")
            .join("models--Freepik--flux.1-lite-8B")
            .join("snapshots/fixture/tokenizer_2/tokenizer.json");
        std::fs::create_dir_all(cached_auxiliary.parent().unwrap()).unwrap();
        std::fs::write(&cached_auxiliary, b"managed repair tokenizer").unwrap();
        let refs = home
            .join(".cache/huggingface/hub")
            .join("models--Freepik--flux.1-lite-8B/refs");
        std::fs::create_dir_all(&refs).unwrap();
        std::fs::write(refs.join("main"), b"fixture").unwrap();

        let models = tmp.path().join("models");
        let name = "Flux-1.lite-8B-MLX-Q4";
        let managed = models.join(name);
        write_mlx_dir(&models, name, "flux");
        let managed_auxiliary = managed.join("tokenizer_2/tokenizer.json");

        // Also provide a fully reusable shared snapshot. The managed install
        // must still be repaired and returned, preserving base precedence.
        let snapshot = hf_home
            .join("hub/models--mlx-community--Flux-1.lite-8B-MLX-Q4")
            .join("snapshots/fixture");
        write_complete_mlx_snapshot(&snapshot);
        let snapshot_auxiliary = snapshot.join("tokenizer_2/tokenizer.json");
        std::fs::create_dir_all(snapshot_auxiliary.parent().unwrap()).unwrap();
        std::fs::write(&snapshot_auxiliary, b"snapshot tokenizer").unwrap();

        let id = "mlx/flux-managed-precedence";
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(mlx_schema(id, name, "mlx-community/Flux-1.lite-8B-MLX-Q4"));
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let path = reg
            .acquire_and_ensure(id, &sink, false, None)
            .await
            .unwrap();

        assert_eq!(path, managed);
        assert_eq!(
            std::fs::read(managed_auxiliary).unwrap(),
            b"managed repair tokenizer"
        );
        let events = recorder.events();
        assert!(matches!(
            events.first(),
            Some(DownloadEvent::Started { .. })
        ));
        assert_eq!(started_count(&events), 1);
    }

    #[tokio::test]
    async fn reusing_a_complete_managed_mlx_dir_emits_no_acquisition_lifecycle() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let _hf_home = ScopedEnvVar::set("HF_HOME", &tmp.path().join("hf-home"));
        let models = tmp.path().join("models");
        let mut reg = UnifiedRegistry::new_empty(models.clone());
        reg.register(mlx_schema(
            "mlx/reuse-complete",
            "Reuse-Complete-MLX",
            "example/reuse-complete",
        ));
        write_mlx_dir(&models, "Reuse-Complete-MLX", "qwen3");
        std::fs::write(
            models.join("Reuse-Complete-MLX").join("tokenizer.json"),
            b"{}",
        )
        .unwrap();
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let path = reg
            .ensure_local_with_progress("mlx/reuse-complete", &sink)
            .await
            .unwrap();

        assert_eq!(path, models.join("Reuse-Complete-MLX"));
        assert!(recorder.events().is_empty());
    }

    #[tokio::test]
    async fn reusing_a_managed_mlx_dir_without_tokenizer_json_emits_no_lifecycle() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let _hf_home = ScopedEnvVar::set("HF_HOME", &tmp.path().join("hf-home"));
        let models = tmp.path().join("models");
        let mut reg = UnifiedRegistry::new_empty(models.clone());
        reg.register(mlx_schema(
            "mlx/reuse-no-tokenizer",
            "Reuse-No-Tokenizer-MLX",
            "example/reuse-no-tokenizer",
        ));
        write_mlx_dir(&models, "Reuse-No-Tokenizer-MLX", "qwen3");
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let path = reg
            .ensure_local_with_progress("mlx/reuse-no-tokenizer", &sink)
            .await
            .unwrap();

        assert_eq!(path, models.join("Reuse-No-Tokenizer-MLX"));
        assert!(recorder.events().is_empty());
    }

    #[tokio::test]
    async fn reusing_a_complete_hf_snapshot_emits_no_acquisition_lifecycle() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let hf_home = tmp.path().join("hf-home");
        let _hf_home = ScopedEnvVar::set("HF_HOME", &hf_home);
        let models = tmp.path().join("models");
        let repo = "example/reuse-hf-snapshot";
        let snapshot = hf_home
            .join("hub")
            .join("models--example--reuse-hf-snapshot")
            .join("snapshots")
            .join("fixture");
        write_complete_mlx_snapshot(&snapshot);
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(mlx_schema("mlx/reuse-hf", "Reuse-HF-MLX", repo));
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let path = reg
            .ensure_local_with_progress("mlx/reuse-hf", &sink)
            .await
            .unwrap();

        assert_eq!(path, snapshot);
        assert!(recorder.events().is_empty());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn an_oversized_installed_model_is_reused_below_the_disk_threshold() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let _hf_home = ScopedEnvVar::set("HF_HOME", &tmp.path().join("hf-home"));
        let models = tmp.path().join("models");
        let mut schema = mlx_schema(
            "mlx/reuse-oversized",
            "Reuse-Oversized-MLX",
            "example/reuse-oversized",
        );
        // One petabyte in MB: safely representable by the canonical catalog
        // identity while still exceeding any machine this test can run on.
        schema.cost.size_mb = Some(1_000_000_000);
        let mut reg = UnifiedRegistry::new_empty(models.clone());
        reg.register(schema);
        write_mlx_dir(&models, "Reuse-Oversized-MLX", "qwen3");
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let path = reg
            .ensure_local_with_progress("mlx/reuse-oversized", &sink)
            .await
            .unwrap();

        assert_eq!(path, models.join("Reuse-Oversized-MLX"));
        assert!(recorder.events().is_empty());
    }

    #[tokio::test]
    async fn a_missing_shard_starts_acquisition_exactly_once_after_the_preflight() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let _hf_home = ScopedEnvVar::set("HF_HOME", &tmp.path().join("hf-home"));
        let models = tmp.path().join("models");
        let model_dir = models.join("Missing-Shard-MLX");
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(model_dir.join("config.json"), b"{}").unwrap();
        std::fs::write(
            model_dir.join("model.safetensors.index.json"),
            r#"{"weight_map":{"a":"model-00001-of-00002.safetensors","b":"model-00002-of-00002.safetensors"}}"#,
        )
        .unwrap();
        std::fs::write(model_dir.join("model-00002-of-00002.safetensors"), b"two").unwrap();
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(mlx_schema(
            "mlx/missing-shard-lifecycle",
            "Missing-Shard-MLX",
            "example/missing-shard",
        ));
        let recorder =
            std::sync::Arc::new(AcquisitionRecorder::replacing_dir_on_started(model_dir));
        let sink = ProgressSink::new(recorder.clone());

        assert!(reg
            .ensure_local_with_progress("mlx/missing-shard-lifecycle", &sink)
            .await
            .is_err());
        let events = recorder.events();
        assert!(matches!(
            events.first(),
            Some(DownloadEvent::Started { .. })
        ));
        assert_eq!(started_count(&events), 1);
    }

    #[tokio::test]
    async fn a_flux_dir_missing_its_auxiliary_tokenizer_keeps_the_lifecycle() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let _hf_home = ScopedEnvVar::set("HF_HOME", &tmp.path().join("hf-home"));
        let models = tmp.path().join("models");
        let name = "Flux-1.lite-8B-MLX-Q4";
        write_mlx_dir(&models, name, "flux");
        std::fs::write(models.join(name).join("tokenizer_2"), b"not a directory").unwrap();
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(mlx_schema(
            "mlx/flux-missing-aux",
            name,
            "mlx-community/Flux-1.lite-8B-MLX-Q4",
        ));
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        assert!(reg
            .ensure_local_with_progress("mlx/flux-missing-aux", &sink)
            .await
            .is_err());
        let events = recorder.events();
        assert!(matches!(
            events.first(),
            Some(DownloadEvent::Started { .. })
        ));
        assert_eq!(started_count(&events), 1);
    }

    #[tokio::test]
    async fn force_bypasses_reuse() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        let name = "Force-Local";
        let model_dir = models.join(name);
        std::fs::create_dir_all(&model_dir).unwrap();
        std::fs::write(model_dir.join("model.gguf"), b"weights").unwrap();
        std::fs::write(model_dir.join("tokenizer.json"), b"{}").unwrap();
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(test_generate_schema(
            "local/force-lifecycle",
            name,
            ModelSource::Local {
                hf_repo: String::new(),
                hf_filename: "model.gguf".into(),
                tokenizer_repo: String::new(),
            },
        ));
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        assert!(reg
            .acquire_and_ensure("local/force-lifecycle", &sink, true, None)
            .await
            .is_err());
        let events = recorder.events();
        assert!(matches!(
            events.first(),
            Some(DownloadEvent::Started { .. })
        ));
        assert_eq!(started_count(&events), 1);
    }

    #[tokio::test]
    async fn staged_pulls_keep_the_lifecycle() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        let canonical = models.join("Staged-Local");
        std::fs::create_dir_all(&canonical).unwrap();
        std::fs::write(canonical.join("model.gguf"), b"canonical weights").unwrap();
        std::fs::write(canonical.join("tokenizer.json"), b"{}").unwrap();
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(test_generate_schema(
            "local/staged-lifecycle",
            "Staged-Local",
            ModelSource::Local {
                hf_repo: String::new(),
                hf_filename: "model.gguf".into(),
                tokenizer_repo: String::new(),
            },
        ));
        let staging = tmp.path().join("staging");
        std::fs::create_dir_all(&staging).unwrap();
        std::fs::write(staging.join("model.gguf"), b"weights").unwrap();
        std::fs::write(staging.join("tokenizer.json"), b"{}").unwrap();
        let recorder = std::sync::Arc::new(AcquisitionRecorder::default());
        let sink = ProgressSink::new(recorder.clone());

        let path = reg
            .acquire_and_ensure("local/staged-lifecycle", &sink, false, Some(&staging))
            .await
            .unwrap();
        assert_eq!(path, staging);
        let events = recorder.events();
        assert!(matches!(
            events.first(),
            Some(DownloadEvent::Started { .. })
        ));
        assert_eq!(started_count(&events), 1);
    }

    #[test]
    fn synthesize_local_schema_classifies_by_name_and_arch() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path();

        write_mlx_dir(root, "MyCustom-Qwen3-7B", "qwen3");
        let gen = synthesize_local_schema("MyCustom-Qwen3-7B", &root.join("MyCustom-Qwen3-7B"))
            .expect("text LLM should be recognized");
        assert_eq!(
            gen.capabilities,
            vec![
                ModelCapability::Generate,
                ModelCapability::Code,
                ModelCapability::Reasoning
            ]
        );
        assert_eq!(gen.context_length, 40_960);
        assert_eq!(gen.provider, "local");
        assert!(matches!(gen.source, ModelSource::Mlx { .. }));

        write_mlx_dir(root, "Some-Embedding-0.6B", "qwen3");
        let emb = synthesize_local_schema("Some-Embedding-0.6B", &root.join("Some-Embedding-0.6B"))
            .expect("embedding model recognized");
        assert_eq!(emb.capabilities, vec![ModelCapability::Embed]);

        // Unknown architecture: refuse rather than guess.
        write_mlx_dir(root, "Mystery-Net", "some_unknown_arch");
        assert!(synthesize_local_schema("Mystery-Net", &root.join("Mystery-Net")).is_none());

        // Speech/vision/etc. names are skipped even with a valid LLM config.
        write_mlx_dir(root, "silero-vad-v6-mlx", "qwen3");
        assert!(
            synthesize_local_schema("silero-vad-v6-mlx", &root.join("silero-vad-v6-mlx")).is_none()
        );

        // A bare directory with no weights is not a model.
        std::fs::create_dir_all(root.join("empty")).unwrap();
        assert!(synthesize_local_schema("empty", &root.join("empty")).is_none());
    }

    /// A GGUF directory found by the scan carries no `hf_repo` — its weights
    /// are whatever the user put there. When the layout is not what the Candle
    /// backend reads, CAR used to attempt a download against the empty repo and
    /// report whatever `https://huggingface.co//…` returned, which explains
    /// nothing about the actual problem.
    #[test]
    fn a_scanned_gguf_directory_diagnoses_instead_of_downloading_from_nowhere() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        let dir = models.join("Dropped-In-Llama");
        std::fs::create_dir_all(&dir).unwrap();
        // A real user drop-in: the file keeps the name it was published under.
        std::fs::write(dir.join("Llama-3-8B-Q4_K_M.gguf"), b"weights").unwrap();

        // Constructed outside the runtime: `new_with_state_root` ends in
        // `refresh_availability`, which blocks.
        let reg = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models);
        let err = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(reg.ensure_local("Dropped-In-Llama"))
            .expect_err("an unloadable layout must not report success");
        let err = err.to_string();

        assert!(err.contains("model.gguf"), "must name what it reads: {err}");
        assert!(
            err.contains("tokenizer.json"),
            "must name the missing tokenizer too: {err}"
        );
        assert!(
            !err.contains("huggingface.co//"),
            "must not have tried to fetch from an empty repo: {err}"
        );
    }

    #[test]
    fn discovery_registers_uncatalogued_local_model() {
        // Constructing a `UnifiedRegistry` ends in `refresh_availability()`,
        // which WRITES the process-global gateway observation. Full reasoning
        // on `TestRegistry` in this crate; family in
        // docs/solutions/process-global-state-in-tests.md.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        write_mlx_dir(&models, "Totally-Custom-Llama-3B", "llama");

        let reg = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models);
        let found = reg
            .list()
            .into_iter()
            .find(|m| m.name == "Totally-Custom-Llama-3B");
        assert!(
            found.is_some(),
            "uncatalogued on-disk model should be registered"
        );
        assert!(found.unwrap().tags.iter().any(|t| t == "auto-discovered"));
    }

    #[test]
    fn explicit_user_row_is_not_pruned_by_auto_discovery_tag() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        let model_dir = models.join("Explicit-Custom-Llama");
        write_mlx_dir(&models, "Explicit-Custom-Llama", "llama");

        let mut registry = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models);
        let model_id = "local/explicit-custom-llama";
        let schema = registry.get(model_id).cloned().unwrap();
        assert!(schema.tags.iter().any(|tag| tag == "auto-discovered"));

        registry.register_user_model(schema);
        std::fs::remove_dir_all(model_dir).unwrap();
        registry.prune_missing_on_disk_models();

        assert!(
            registry.get(model_id).is_some(),
            "explicit registration provenance must outrank a user-controlled tag"
        );
    }

    #[test]
    fn signed_catalog_cannot_shadow_builtin_exact_id() {
        // Constructing a `UnifiedRegistry` ends in `refresh_availability()`,
        // which WRITES the process-global gateway observation. Full reasoning
        // on `TestRegistry` in this crate; family in
        // docs/solutions/process-global-state-in-tests.md.
        let _environment = crate::openrouter::test_environment_scope();
        // A signature authenticates the catalog publisher, but it does not
        // authorize changing a project-owned exact id compiled into this CAR
        // build. An additive signed row may not shadow a builtin.
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");

        let builtin = builtin_catalog();
        let mut overriding = builtin.first().expect("a built-in model").clone();
        let target_id = overriding.id.clone();
        overriding.name = "REPLACED-BY-CATALOG".into();

        let (verified, public_key) = crate::catalog::signed_test_catalog(
            crate::catalog::CatalogDoc {
                version: 1,
                models: vec![overriding],
            },
            51,
        );
        crate::catalog::save_verified(&crate::catalog::cache_path(tmp.path()), &verified).unwrap();

        // new() loads the verified cache additively after the built-ins.
        let reg = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            models_dir,
            Some(public_key.as_str()),
        );
        assert_eq!(
            reg.get(&target_id).map(|m| m.name.as_str()),
            Some(builtin.first().unwrap().name.as_str()),
            "a signed cache row must not replace a builtin exact id"
        );
    }

    #[test]
    fn user_model_cannot_shadow_project_owned_exact_id() {
        let mut reg = test_registry();
        let original = builtin_catalog().first().expect("a builtin").clone();
        let mut forged = original.clone();
        forged.name = "USER-SHADOW".into();

        reg.register_user_model(forged);

        assert_eq!(
            reg.get(&original.id).map(|model| model.name.as_str()),
            Some(original.name.as_str()),
            "a user row must not replace a project-owned exact id"
        );
    }

    #[test]
    fn legacy_unsigned_catalog_cache_cannot_replace_builtin() {
        // Constructing a `UnifiedRegistry` ends in `refresh_availability()`,
        // which WRITES the process-global gateway observation. Full reasoning
        // on `TestRegistry` in this crate; family in
        // docs/solutions/process-global-state-in-tests.md.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let builtin = builtin_catalog();
        let original = builtin.first().expect("a built-in model");
        let mut forged = original.clone();
        forged.name = "FORGED-UNSIGNED-CATALOG".into();
        let path = crate::catalog::cache_path(tmp.path());
        std::fs::write(
            &path,
            serde_json::to_vec_pretty(&crate::catalog::CatalogDoc {
                version: u64::MAX,
                models: vec![forged],
            })
            .unwrap(),
        )
        .unwrap();

        let reg = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models_dir);
        assert_eq!(
            reg.get(&original.id).map(|model| model.name.as_str()),
            Some(original.name.as_str()),
            "legacy unsigned cache JSON must fail closed and preserve the built-in"
        );
    }

    #[test]
    fn tampered_signed_managed_row_preserves_builtin() {
        // Constructing a `UnifiedRegistry` ends in `refresh_availability()`,
        // which WRITES the process-global gateway observation. Full reasoning
        // on `TestRegistry` in this crate; family in
        // docs/solutions/process-global-state-in-tests.md.
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models_dir = tmp.path().join("models");
        let original = builtin_catalog()
            .into_iter()
            .find(|model| model.id == "parslee/openrouter/frontier-general")
            .expect("managed frontier alias");
        let mut forged = original.clone();
        forged.name = "SIGNED-THEN-TAMPERED-MANAGED".into();
        let (verified, public_key) = crate::catalog::signed_test_catalog(
            crate::catalog::CatalogDoc {
                version: 9,
                models: vec![forged],
            },
            52,
        );
        let path = crate::catalog::cache_path(tmp.path());
        crate::catalog::save_verified(&path, &verified).unwrap();
        let cache = std::fs::read_to_string(&path)
            .unwrap()
            .replace("SIGNED-THEN-TAMPERED-MANAGED", "ATTACKER-MUTATION");
        std::fs::write(&path, cache).unwrap();

        let reg = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            models_dir,
            Some(public_key.as_str()),
        );
        assert_eq!(
            reg.get(&original.id).map(|model| model.name.as_str()),
            Some(original.name.as_str()),
            "a tampered same-id managed row must fail verification and preserve the builtin"
        );
    }

    #[test]
    fn builtin_catalog_loads() {
        let reg = test_registry();
        let all = reg.list();
        assert_eq!(all.len(), builtin_catalog().len());
    }

    #[test]
    fn shipped_supervised_vllm_models_use_the_managed_source_contract() {
        let managed = builtin_catalog()
            .into_iter()
            .filter(|model| model.id.starts_with("vllm-mlx/"))
            .collect::<Vec<_>>();
        assert_eq!(managed.len(), 8);
        assert!(managed.iter().all(ModelSchema::is_car_managed_vllm_mlx));
        assert!(managed.iter().all(ModelSchema::downloads_weights));
    }

    /// #137: a model tagged `requires-mlx-vlm` must report
    /// `available: true` if and only if `mlx_vlm_cli::is_available()`
    /// returns true. Without this, registry consumers (FFI
    /// `listModelsUnified`, the tray Models submenu, agent routing)
    /// see a model as available, the user picks it, and inference
    /// bails with `mlx-vlm CLI not found on PATH`.
    ///
    /// The probe is environmental — runs the same check on the host
    /// the test executes on. CI usually doesn't have `mlx_vlm`
    /// installed → expected unavailable; a dev box with it installed
    /// → expected available. Either way, the registry tracks the
    /// runtime probe.
    #[test]
    fn mlx_vlm_models_reflect_runtime_availability() {
        let reg = test_registry();
        let mlx_vlm_models: Vec<&ModelSchema> = reg
            .list()
            .into_iter()
            .filter(|m| m.tags.iter().any(|t| t == "requires-mlx-vlm"))
            .collect();
        assert!(
            !mlx_vlm_models.is_empty(),
            "catalog should contain at least one model tagged \
             `requires-mlx-vlm` — otherwise this regression has \
             nothing to guard"
        );

        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        let expected = crate::backend::mlx_vlm_cli::is_available();
        #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
        let expected = false;

        for m in mlx_vlm_models {
            assert_eq!(
                m.available, expected,
                "model {} `available` field should reflect \
                 mlx_vlm CLI presence (expected {expected}, got {})",
                m.id, m.available
            );
        }
    }

    /// F1 (Parslee-ai/car#231 §7.1): MLX models must report
    /// `available: false` on any non-Apple-Silicon-with-MLX build target.
    /// Without this, the Windows / Linux / Intel-Mac router happily
    /// adds MLX entries to fallback chains and inference fails at
    /// dispatch with "model not found", producing a 7-deep cascade
    /// of useless errors on a fresh install.
    ///
    /// The test runs on every platform. On macOS arm64 (default-features),
    /// at least one MLX model with an `hf_repo` should report available;
    /// on Linux / Windows / Intel-Mac / `car_skip_mlx`, every plain MLX
    /// model must report unavailable.
    #[test]
    fn mlx_models_unavailable_on_non_mlx_targets() {
        let reg = test_registry();
        let mlx_models: Vec<&ModelSchema> = reg
            .list()
            .into_iter()
            .filter(|m| {
                m.is_mlx()
                    // Exclude `requires-mlx-vlm` and `speech` — those have
                    // their own availability logic (mlx_vlm CLI probe,
                    // speech_mlx_available). The plain MLX models are
                    // what §7.1 surfaced as broken on Windows.
                    && !m.tags.iter().any(|t| t == "requires-mlx-vlm")
                    && !m.tags.contains(&"speech".to_string())
            })
            .collect();
        assert!(
            !mlx_models.is_empty(),
            "catalog should contain at least one plain MLX model — \
             otherwise this F1 regression guard has nothing to guard"
        );

        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        {
            // On a real MLX target, models with an `hf_repo` should be
            // marked available (the `ensure_local()` lazy-download path).
            // At least one must qualify.
            let any_available = mlx_models.iter().any(|m| m.available);
            assert!(
                any_available,
                "on macOS arm64 with MLX enabled, at least one plain MLX \
                 model with hf_repo should be available — none were"
            );
        }
        #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
        {
            // On every other target, MLX models cannot execute, so the
            // registry must mark them all unavailable.
            for m in &mlx_models {
                assert!(
                    !m.available,
                    "MLX model {} is marked available on a non-MLX target — \
                     the adaptive router will add it to fallback chains \
                     and dispatch will fail (Parslee-ai/car#231 §7.1)",
                    m.id
                );
            }
        }
    }

    /// Embedded JSON must parse cleanly — if it doesn't, the runtime
    /// would panic on first registry load. Catch it in CI instead.
    #[test]
    fn builtin_catalog_json_parses() {
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON)
            .expect("builtin_catalog.json must be valid ModelSchema array");
        assert!(
            !catalog.is_empty(),
            "embedded catalog has no entries — that's almost certainly wrong"
        );

        let mut seen = std::collections::HashSet::new();
        for entry in &catalog {
            assert!(
                seen.insert(entry.id.clone()),
                "duplicate id in builtin_catalog.json: {}",
                entry.id
            );
        }
    }

    #[test]
    fn codex_subscription_row_has_pinnable_text_only_identity() {
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
        let row = catalog
            .iter()
            .find(|model| model.id == "openai/gpt-5.6-sol:high")
            .expect("catalog publishes the Codex subscription identity");
        assert_eq!(row.name, "gpt-5.6-sol:high");
        assert!(matches!(
            &row.source,
            ModelSource::CodexCli { model } if model == "gpt-5.6-sol:high"
        ));
        assert!(row.has_capability(ModelCapability::Generate));
        assert!(row.has_capability(ModelCapability::Reasoning));
        assert!(!row.has_capability(ModelCapability::ToolUse));
        assert!(!row.has_capability(ModelCapability::MultiToolCall));
        assert!(!row.has_capability(ModelCapability::Vision));
        assert_eq!(row.supported_params, vec![GenerateParam::MaxTokens]);
        assert!(!row.downloads_weights());
    }

    /// The Apple FoundationModels row must advertise `ToolUse` and must NOT
    /// advertise `MultiToolCall`, and its context window must match the
    /// framework's real limit. All three were wrong or missing, and each
    /// failure was silent:
    ///
    /// - `AdaptiveRouter` pushes `ModelCapability::ToolUse` as a HARD required
    ///   capability whenever `has_tools` is set, so a row without it can never
    ///   be selected for a tool-calling route. Omitting it made the entire
    ///   `backend::foundation_models::generate_with_tools` path — and the Swift
    ///   shim's capture-and-return bridge behind it — unreachable through the
    ///   router, with no error anywhere: the model was simply never a candidate.
    /// - The same filter pushes `MultiToolCall` for multi-step prompts. The
    ///   bridge ends the turn on the first captured call, so at most one tool
    ///   call per turn is possible and claiming it would be a lie the router
    ///   would act on.
    /// - `SystemLanguageModel`'s context window is 4096 tokens, not 8192.
    ///   `build_context_for_model` sizes the assembly budget from this number,
    ///   so an inflated value overcommits the prompt and the framework rejects
    ///   the request at generation time.
    ///
    /// Pins all three so the catalog claim and the implementation stay in
    /// lockstep, in the same spirit as the in-process Qwen3 test below.
    #[test]
    fn apple_foundation_row_claims_single_tool_call_and_the_real_context_window() {
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
        let row = catalog
            .iter()
            .find(|model| model.id == "apple/foundation:default")
            .expect("catalog publishes the Apple FoundationModels row");
        assert!(matches!(
            &row.source,
            ModelSource::AppleFoundationModels { .. }
        ));
        assert!(row.has_capability(ModelCapability::Generate));
        assert!(
            row.has_capability(ModelCapability::ToolUse),
            "generate_with_tools exists, so the router must be able to pick this row for tool routes"
        );
        assert!(
            !row.has_capability(ModelCapability::MultiToolCall),
            "the capture sentinel ends the turn on the first tool call"
        );
        assert!(!row.has_capability(ModelCapability::Vision));
        assert_eq!(
            row.context_length, 4096,
            "SystemLanguageModel's context window is 4096 tokens"
        );
        assert!(!row.downloads_weights());
    }

    /// The row gains `multi_tool_call` exactly where the host can do it.
    ///
    /// The catalog literal deliberately omits it, because macOS 26 yields one
    /// captured call per turn; macOS 27 yields every proposed call through the
    /// same unchanged shim. So the claim is host-dependent, and the registry
    /// is where that gets resolved — the same hook that already makes
    /// `available` host-dependent for this source.
    ///
    /// Pins the agreement rather than a platform: whatever
    /// `supports_parallel_tool_calls()` reports is what the live row must
    /// claim. A regression that upgrades the row on a host that cannot do it
    /// (or fails to upgrade one that can) fails here.
    #[test]
    fn apple_foundation_multi_tool_call_tracks_what_the_host_can_actually_do() {
        #[cfg(any(
            all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
            all(target_os = "ios", target_arch = "aarch64")
        ))]
        {
            let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
            let declared = catalog
                .iter()
                .find(|m| m.id == "apple/foundation:default")
                .expect("catalog publishes the Apple FoundationModels row");
            assert!(
                !declared.has_capability(ModelCapability::MultiToolCall),
                "the catalog literal must stay conservative; the host decides"
            );

            // A temp models dir keeps this off the developer's real ~/.car.
            let models_dir = tempfile::tempdir().unwrap();
            let registry = UnifiedRegistry::new(models_dir.path().to_path_buf());
            // expect(), not a silent early return: the row comes from the
            // builtin catalog and is therefore always present here, so a
            // missing one is a real regression rather than a reason to skip.
            // A test that can quietly pass by doing nothing is the failure
            // mode this whole row keeps hitting.
            let live = registry
                .models
                .get("apple/foundation:default")
                .expect("the builtin catalog row must survive into the live registry");
            let host_can = crate::backend::foundation_models::supports_parallel_tool_calls();
            assert_eq!(
                live.capabilities.contains(&ModelCapability::MultiToolCall),
                host_can,
                "live row must claim multi_tool_call iff the host supports it"
            );
        }
    }

    /// The row gains `vision` exactly where images can actually be served.
    ///
    /// Stricter than the `multi_tool_call` rule, and deliberately so. That one
    /// only has to be truthful; this one also has to be *serviceable*, because
    /// `AdaptiveRouter` treats vision as a hard required capability and will
    /// hand image requests to whatever claims it. A row that advertised vision
    /// without `generate_with_images` behind it would be picked and then fail
    /// at generation — the same shape as the 8192-vs-4096 context bug, where
    /// the claim outran what the backend could do.
    ///
    /// Both sides read the same probe, so they cannot drift apart.
    #[test]
    fn apple_foundation_claims_vision_only_where_images_can_be_served() {
        #[cfg(any(
            all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
            all(target_os = "ios", target_arch = "aarch64")
        ))]
        {
            let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
            let declared = catalog
                .iter()
                .find(|m| m.id == "apple/foundation:default")
                .expect("catalog publishes the Apple FoundationModels row");
            assert!(
                !declared.has_capability(ModelCapability::Vision),
                "the catalog literal must stay conservative; the device decides"
            );

            let models_dir = tempfile::tempdir().unwrap();
            let registry = UnifiedRegistry::new(models_dir.path().to_path_buf());
            let live = registry
                .models
                .get("apple/foundation:default")
                .expect("the builtin catalog row must survive into the live registry");

            assert_eq!(
                live.capabilities.contains(&ModelCapability::Vision),
                crate::backend::foundation_models::supports_vision(),
                "the live row must claim vision iff this device's model accepts images"
            );
        }
    }

    /// On a host WITHOUT the macOS 27 behaviour, the live row must be
    /// byte-identical to the catalog row.
    ///
    /// This is the "does it hurt previous versions?" guard, and it is
    /// deliberately written as a property rather than a platform check,
    /// because the machine that can run macOS 26 and the machine that runs
    /// this suite are not the same machine. Whatever the host reports, the
    /// contract is: no parallel support means nothing was added.
    ///
    /// It therefore proves the macOS 26 no-regression claim by construction
    /// instead of by testimony — on such a host
    /// `supports_parallel_tool_calls()` is false, this assertion runs, and a
    /// capability that leaked in anyway fails the build. On macOS 27 the
    /// branch is inert and
    /// `apple_foundation_multi_tool_call_tracks_what_the_host_can_actually_do`
    /// covers the positive case.
    #[test]
    fn a_host_without_parallel_support_gets_exactly_the_catalog_capabilities() {
        #[cfg(any(
            all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
            all(target_os = "ios", target_arch = "aarch64")
        ))]
        {
            if crate::backend::foundation_models::supports_parallel_tool_calls() {
                return;
            }
            let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
            let declared = catalog
                .iter()
                .find(|m| m.id == "apple/foundation:default")
                .expect("catalog publishes the Apple FoundationModels row");

            let models_dir = tempfile::tempdir().unwrap();
            let registry = UnifiedRegistry::new(models_dir.path().to_path_buf());
            let live = registry
                .models
                .get("apple/foundation:default")
                .expect("the builtin catalog row must survive into the live registry");

            assert_eq!(
                live.capabilities, declared.capabilities,
                "a host without the macOS 27 behaviour must advertise exactly what \
                 the catalog declares — nothing added, nothing removed"
            );
        }
    }

    /// `context_size()` reports the OS's own answer, and the catalog literal
    /// is the fallback when it can't. Both halves matter: the literal above
    /// is what macOS 26.0-26.3 and every non-Apple host keep using, because
    /// `SystemLanguageModel.contextSize` only exists from 26.4 on.
    ///
    /// Deliberately not asserting a specific number on a supported host --
    /// the whole point of reading it is that Apple owns the value. Measured
    /// 4096 on both macOS 26 and macOS 27; a future OS moving it should
    /// change behaviour here without failing this test.
    #[test]
    fn apple_foundation_context_size_is_either_the_os_answer_or_none() {
        #[cfg(any(
            all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
            all(target_os = "ios", target_arch = "aarch64")
        ))]
        {
            if let Some(window) = crate::backend::foundation_models::context_size() {
                assert!(
                    window >= 512,
                    "a context window the framework reports must be usable, got {window}"
                );
            }
        }

        // Everywhere else the shim isn't built, so the sentinel must be None
        // and the declared `context_length` stands untouched.
        #[cfg(not(any(
            all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)),
            all(target_os = "ios", target_arch = "aarch64")
        )))]
        {
            let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
            let row = catalog
                .iter()
                .find(|model| model.id == "apple/foundation:default")
                .expect("catalog publishes the Apple FoundationModels row");
            assert_eq!(row.context_length, 4096);
        }
    }

    /// In-process Qwen3 models advertise tool capability because the local
    /// generate path now renders Qwen3's `<tools>` chat format and parses
    /// `<tool_call>` blocks back out (`tasks::generate::render_chat_prompt` /
    /// `parse_tool_calls`). This pins the catalog so the capability claim and
    /// the implementation stay in lockstep — if the in-process tool path is
    /// ever removed, this test should fail until the claim is dropped too.
    #[test]
    fn in_process_qwen3_models_declare_tool_use() {
        use crate::schema::ModelSource;
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
        // The mid/large Qwen3 sizes are reliable tool-callers; the 0.6b/1.7b
        // tiers legitimately don't advertise tool_use.
        let tool_sizes = ["qwen3-4b", "qwen3-8b", "qwen3-30b-a3b"];
        let mut checked = 0;
        for entry in &catalog {
            let in_process = matches!(
                entry.source,
                ModelSource::Mlx { .. } | ModelSource::Local { .. }
            );
            if !in_process || !tool_sizes.iter().any(|s| entry.id.contains(s)) {
                continue;
            }
            assert!(
                entry.capabilities.contains(&ModelCapability::ToolUse),
                "in-process Qwen3 model {} should advertise ToolUse — the local \
                 generate path renders/parses tool calls",
                entry.id
            );
            checked += 1;
        }
        assert_eq!(
            checked, 6,
            "expected 6 in-process tool-capable Qwen3 entries (3 mlx + 3 gguf)"
        );
    }

    #[test]
    fn public_benchmarks_round_trip_through_model_info() {
        use crate::schema::BenchmarkScore;
        let mut reg = test_registry();
        let mut schema = reg
            .find_by_name("Qwen3-4B")
            .expect("catalog has Qwen3-4B")
            .clone();
        schema.id = "test/qwen3-4b-with-bench".into();
        schema.public_benchmarks = vec![
            BenchmarkScore {
                name: "MMLU-Pro".into(),
                score: 0.482,
                harness: Some("5-shot CoT".into()),
                source_url: Some("https://example.invalid/qwen3-4b-card".into()),
                measured_at: Some("2025-08-12".into()),
            },
            BenchmarkScore {
                name: "HumanEval".into(),
                score: 0.713,
                harness: Some("pass@1".into()),
                source_url: None,
                measured_at: None,
            },
        ];
        reg.register(schema);

        let stored = reg
            .get("test/qwen3-4b-with-bench")
            .expect("registered model is retrievable");
        let info = ModelInfo::from(stored);
        assert_eq!(info.public_benchmarks.len(), 2);

        // The serialized JSON shape is what the WS / FFI clients consume.
        let json = serde_json::to_string(&info).unwrap();
        assert!(json.contains("\"public_benchmarks\""));
        assert!(json.contains("\"MMLU-Pro\""));
        assert!(json.contains("\"5-shot CoT\""));

        // Round-trip back through serde to confirm deserialization works.
        let decoded: ModelInfo = serde_json::from_str(&json).unwrap();
        assert_eq!(decoded.public_benchmarks.len(), 2);
        assert_eq!(decoded.public_benchmarks[0].name, "MMLU-Pro");
        assert_eq!(decoded.public_benchmarks[1].name, "HumanEval");
    }

    #[test]
    fn public_benchmarks_default_to_empty_when_absent_in_json() {
        // Older user-config JSON written before this field existed must
        // still deserialize cleanly into the new ModelSchema shape.
        let legacy_json = r#"{
            "id": "legacy/test:1",
            "name": "Legacy Test",
            "provider": "test",
            "family": "test",
            "version": "",
            "capabilities": ["generate"],
            "context_length": 4096,
            "param_count": "1B",
            "quantization": null,
            "performance": {},
            "cost": {},
            "source": { "type": "ollama", "model_tag": "legacy:1" },
            "tags": [],
            "supported_params": []
        }"#;
        let schema: ModelSchema = serde_json::from_str(legacy_json).unwrap();
        assert!(schema.public_benchmarks.is_empty());
    }

    #[test]
    fn find_by_name() {
        let reg = test_registry();
        let m = reg.find_by_name("Qwen3-4B").unwrap();
        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        assert_eq!(m.id, "mlx/qwen3-4b:4bit");
        #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
        assert_eq!(m.id, "qwen/qwen3-4b:q4_k_m");
        assert!(m.has_capability(ModelCapability::Code));
    }

    #[test]
    fn query_by_capability() {
        let reg = test_registry();
        let embed_models = reg.query_by_capability(ModelCapability::Embed);
        assert_eq!(embed_models.len(), 2);
        assert!(embed_models
            .iter()
            .any(|model| model.name == "Qwen3-Embedding-0.6B"));
        assert!(embed_models
            .iter()
            .any(|model| model.name == "Qwen3-Embedding-0.6B-MLX"));
    }

    #[test]
    fn query_with_filter() {
        let reg = test_registry();
        let code_small = reg.query(&ModelFilter {
            capabilities: vec![ModelCapability::Code],
            max_size_mb: Some(3000),
            local_only: true,
            ..Default::default()
        });
        // Qwen3-1.7B, Qwen3-1.7B-MLX, Qwen3-4B, and Qwen3-4B-MLX fit and have Code capability.
        assert_eq!(code_small.len(), 4);
    }

    #[test]
    fn register_remote() {
        let mut reg = test_registry();
        let initial_len = reg.list().len();
        let initial_reasoning_len = reg
            .query(&ModelFilter {
                capabilities: vec![ModelCapability::Reasoning, ModelCapability::ToolUse],
                ..Default::default()
            })
            .len();
        let remote = ModelSchema {
            id: "anthropic/claude-sonnet-4-6:latest".into(),
            name: "Claude Sonnet 4.6".into(),
            provider: "anthropic".into(),
            family: "claude-4".into(),
            version: "latest".into(),
            capabilities: vec![
                ModelCapability::Generate,
                ModelCapability::Code,
                ModelCapability::Reasoning,
                ModelCapability::ToolUse,
            ],
            context_length: 200000,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: PerformanceEnvelope {
                latency_p50_ms: Some(2000),
                ..Default::default()
            },
            cost: CostModel {
                input_per_mtok: Some(3.0),
                output_per_mtok: Some(15.0),
                ..Default::default()
            },
            source: ModelSource::RemoteApi {
                endpoint: "https://api.anthropic.com/v1/messages".into(),
                api_key_env: "ANTHROPIC_API_KEY".into(),
                api_key_envs: vec![],
                api_version: Some("2023-06-01".into()),
                protocol: ApiProtocol::Anthropic,
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: false,
            weights_ready: false,
        };

        reg.register(remote);
        // Same ID as builtin claude-sonnet-4-6 — replaces, count stays same
        assert_eq!(reg.list().len(), initial_len);

        let reasoning = reg.query(&ModelFilter {
            capabilities: vec![ModelCapability::Reasoning, ModelCapability::ToolUse],
            ..Default::default()
        });
        // Replacing an existing remote slot should not change the reasoning/tool-use lineup size.
        assert_eq!(reasoning.len(), initial_reasoning_len);
    }

    #[test]
    fn unregister() {
        let mut reg = test_registry();
        let initial_len = reg.list().len();
        let removed = reg.unregister("qwen/qwen3-0.6b:q8_0");
        assert!(removed.is_some());
        assert_eq!(reg.list().len(), initial_len - 1);
    }

    #[test]
    fn speech_models_are_curated() {
        let reg = test_registry();
        let stt = reg.query_by_capability(ModelCapability::SpeechToText);
        let tts = reg.query_by_capability(ModelCapability::TextToSpeech);
        // Parakeet-TDT-MLX + Whisper-large-v3-turbo (cross-platform) + scribe_v1.
        assert_eq!(stt.len(), 3);
        // Kokoro-6bit/bf16 + Windows-Speech (OS, Windows-only) + Qwen3-TTS + eleven.
        assert_eq!(tts.len(), 5);
        // whisper.cpp STT is a first-class catalog citizen — the cross-platform
        // local STT that runs where MLX (Apple-only) can't.
        let whisper = stt
            .iter()
            .find(|m| m.name == "Whisper-large-v3-turbo-q5_0")
            .expect("whisper STT model should be curated");
        assert!(whisper.is_local());
        assert!(matches!(
            whisper.source,
            crate::schema::ModelSource::WhisperCpp { .. }
        ));
    }

    #[test]
    fn qwen_8b_variants_keep_tool_use_consistent() {
        // The GGUF and MLX twins of Qwen3-8B must agree on tool capability, and
        // both advertise it: the in-process generate path renders Qwen3's
        // <tools> chat format and parses <tool_call> blocks back out (see
        // tasks::generate::render_chat_prompt / parse_tool_calls).
        let reg = test_registry();
        for name in ["Qwen3-8B", "Qwen3-8B-MLX"] {
            let model = reg.find_by_name(name).expect("model should exist");
            assert!(model.has_capability(ModelCapability::ToolUse));
            assert!(model.has_capability(ModelCapability::MultiToolCall));
        }
    }

    #[test]
    fn mac_name_resolution_prefers_mlx_siblings() {
        // Only used inside the aarch64-macos cfg below; non-mac targets
        // keep the test as a smoke compile.
        #[allow(unused_variables)]
        let reg = test_registry();
        #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
        {
            assert_eq!(
                reg.find_by_name("Qwen3-0.6B").unwrap().id,
                "mlx/qwen3-0.6b:6bit"
            );
            assert_eq!(
                reg.find_by_name("Qwen3-1.7B").unwrap().id,
                "mlx/qwen3-1.7b:3bit"
            );
            assert_eq!(
                reg.find_by_name("Qwen3-Embedding-0.6B").unwrap().id,
                "mlx/qwen3-embedding-0.6b:mxfp8"
            );
        }
    }

    #[test]
    fn remote_multimodal_models_are_curated_as_vision_capable() {
        let reg = test_registry();
        for name in [
            "claude-opus-4-7",
            "claude-opus-4-6",
            "claude-sonnet-4-6",
            "claude-haiku-4-5",
            "gpt-5.4",
            "gpt-5.4-mini",
            "o3",
            "o4-mini",
            "gpt-4.1-mini",
            "gemini-2.5-pro",
            "gemini-2.5-flash",
        ] {
            let model = reg.find_by_name(name).expect("model should exist");
            assert!(
                model.has_capability(ModelCapability::Vision),
                "{name} should be curated as vision-capable"
            );
        }
    }

    #[test]
    fn qwen25vl_entries_are_replaced_by_qwen3vl_in_builtin_catalog() {
        let reg = test_registry();

        let stale_ids = [
            // Native MLX text tower can't tokenize images — never advertise.
            "mlx/qwen2.5-vl-3b:4bit",
            "mlx/qwen2.5-vl-7b:4bit",
            // Qwen2.5-VL is superseded by Qwen3-VL; drop the mlx-vlm CLI
            // catalog entries so callers route to the upgraded family.
            "mlx-vlm/qwen2.5-vl-3b:4bit",
            "mlx-vlm/qwen2.5-vl-7b:4bit",
            // Same supersession applies to the vLLM-MLX route.
            "vllm-mlx/qwen2.5-vl-3b:4bit",
        ];
        for id in stale_ids {
            assert!(
                reg.get(id).is_none(),
                "{id} is superseded by Qwen3-VL; the catalog must not advertise it"
            );
        }

        let vision_ids: Vec<&str> = reg
            .query_by_capability(ModelCapability::Vision)
            .into_iter()
            .map(|model| model.id.as_str())
            .collect();
        for stale in stale_ids {
            assert!(
                !vision_ids.contains(&stale),
                "{stale} must not be reachable through the Vision capability index"
            );
        }
        assert!(
            vision_ids.contains(&"mlx-vlm/qwen3-vl-2b:bf16"),
            "Qwen3-VL is the supported local VL family and must route as Vision"
        );
    }

    #[test]
    fn gemini_models_are_curated_for_multimodal_tool_use() {
        let reg = test_registry();
        for name in ["gemini-2.5-pro", "gemini-2.5-flash"] {
            let model = reg.find_by_name(name).expect("model should exist");
            assert!(model.has_capability(ModelCapability::Vision));
            assert!(model.has_capability(ModelCapability::ToolUse));
            assert!(model.has_capability(ModelCapability::MultiToolCall));
        }
    }

    #[test]
    fn model_info_publishes_declared_prices_and_keeps_unpriced_distinct_from_free() {
        let reg = test_registry();

        // A priced remote row publishes every declared rate plus its tiers.
        let opus = reg
            .list()
            .into_iter()
            .find(|m| m.id == "openrouter/anthropic/claude-opus-4.8")
            .map(ModelInfo::from)
            .expect("curated opus-4.8 row is present on first boot");
        assert_eq!(opus.cost.input_per_mtok, Some(5.0));
        assert_eq!(opus.cost.output_per_mtok, Some(25.0));
        assert_eq!(opus.cost.cache_read_input_per_mtok, Some(0.5));
        assert_eq!(opus.cost.cache_write_input_per_mtok, Some(6.25));

        // Tiered pricing survives the projection where a model declares it.
        let gpt = reg
            .list()
            .into_iter()
            .find(|m| m.id == "openrouter/openai/gpt-5.4")
            .map(ModelInfo::from)
            .expect("curated gpt-5.4 row");
        assert_eq!(gpt.cost.pricing_tiers.len(), 1);
        assert_eq!(gpt.cost.prices_for(272_000).input_per_mtok, Some(5.0));

        // A local model that declares no prices reports them ABSENT. `null`
        // and `0.0` are different facts and must stay different on the wire.
        let local = reg
            .list()
            .into_iter()
            .find(|m| m.is_local() && m.cost.input_per_mtok.is_none())
            .map(ModelInfo::from)
            .expect("the built-in catalog ships unpriced local models");
        let json = serde_json::to_value(&local).unwrap();
        assert!(json["cost"]["input_per_mtok"].is_null());
        assert!(json["cost"]["output_per_mtok"].is_null());
        assert_ne!(json["cost"]["input_per_mtok"], serde_json::json!(0.0));
    }

    #[test]
    fn a_hand_registered_copy_of_a_curated_id_does_not_double_the_row() {
        let mut reg = test_registry();
        let id = "openrouter/anthropic/claude-opus-4.8";
        assert_eq!(reg.list().iter().filter(|m| m.id == id).count(), 1);

        // The runtime registry is keyed by id, so a user registration of the
        // same id replaces the curated row rather than appending a duplicate.
        let mut copy = reg
            .list()
            .into_iter()
            .find(|m| m.id == id)
            .cloned()
            .expect("curated row");
        copy.name = "hand-registered".into();
        reg.register_user_model(copy);
        assert_eq!(reg.list().iter().filter(|m| m.id == id).count(), 1);
    }

    /// Scoped to `ModelInfo` — the `models.list_unified` projection — on
    /// purpose. `models.search` wraps this same struct but adds `family`,
    /// which for a managed alias names the upstream model family and is
    /// matched against by queries. That is pre-existing (`frontier-deep`
    /// already carries `claude-4.6`) and out of scope here; this test locks
    /// the surface the guarantee actually holds on rather than implying a
    /// registry-wide one it does not.
    #[test]
    fn managed_alias_publishes_prices_without_disclosing_the_upstream_id_in_the_catalog_view() {
        let reg = test_registry();
        let alias = reg
            .list()
            .into_iter()
            .find(|m| m.id == "parslee/openrouter/frontier-deep-next")
            .map(ModelInfo::from)
            .expect("managed alias for the new curated row");

        assert_eq!(alias.cost.input_per_mtok, Some(5.0));
        assert_eq!(alias.cost.output_per_mtok, Some(25.0));
        assert_eq!(alias.cost.cache_read_input_per_mtok, Some(0.5));
        assert_eq!(alias.cost.cache_write_input_per_mtok, Some(6.25));

        let wire = serde_json::to_string(&alias).unwrap();
        assert!(!wire.contains("claude-opus-4.8"));
        assert!(!wire.contains("anthropic/"));
    }

    #[test]
    fn model_info_from_an_older_daemon_without_cost_still_parses() {
        // A newly built client must not fail to read a catalog produced before
        // `cost` existed; the row deserializes with an all-absent cost.
        let legacy = serde_json::json!({
            "id": "legacy/model",
            "name": "legacy",
            "provider": "legacy",
            "capabilities": ["generate"],
            "param_count": "",
            "size_mb": 0,
            "context_length": 8192,
            "available": true,
            "is_local": false
        });
        let info: ModelInfo = serde_json::from_value(legacy).expect("older catalog row parses");
        assert!(info.cost.input_per_mtok.is_none());
        assert!(info.cost.output_per_mtok.is_none());
        assert!(info.cost.pricing_tiers.is_empty());
        assert!(info.max_output_tokens.is_none());
        assert!(info.car_enabled, "legacy rows default to enabled");
        assert!(!info.can_remove);
        assert!(!info.in_use);
        assert!(info.management_evidence.is_none());
    }

    #[test]
    fn visual_generation_models_are_curated() {
        let reg = test_registry();
        assert_eq!(
            reg.query_by_capability(ModelCapability::ImageGeneration)
                .len(),
            1
        );
        assert_eq!(
            reg.query_by_capability(ModelCapability::VideoGeneration)
                .len(),
            1
        );
    }
}

/// Structural validation of the built-in catalog.
///
/// The catalog is hand-edited JSON compiled into the binary, so a typo in it is
/// a shipped bug that no compiler catches. It went seven weeks without a new
/// text model while the open-weight world moved through three generations, and
/// the entry added to end that drought carried `param_count: "MoE"` — a string
/// the scorer cannot parse — which nothing would have caught. These are the
/// invariants the rest of the code already assumes.
#[cfg(test)]
mod builtin_catalog_validation {
    use super::*;
    use crate::schema::ModelSource;

    /// Sources whose weights CAR (or a runtime CAR supervises) actually loads.
    fn weight_repo(source: &ModelSource) -> Option<&str> {
        match source {
            ModelSource::Mlx { hf_repo, .. } => Some(hf_repo),
            ModelSource::Local { hf_repo, .. } => Some(hf_repo),
            ModelSource::ManagedVllmMlx { hf_repo, .. } => Some(hf_repo),
            _ => None,
        }
    }

    #[test]
    fn ids_are_unique() {
        let catalog = builtin_catalog();
        let mut seen: Vec<&str> = Vec::new();
        for model in &catalog {
            assert!(
                !seen.contains(&model.id.as_str()),
                "duplicate catalog id `{}` — the later entry silently shadows the earlier",
                model.id
            );
            seen.push(&model.id);
        }
    }

    #[test]
    fn exact_frontier_rows_lock_native_selectors_and_digests() {
        let catalog = builtin_catalog();
        for (id, name, version, expected_digest) in [
            (
                "openai/gpt-5.5-2026-04-23",
                "gpt-5.5-2026-04-23",
                "2026-04-23",
                "aa1b0741114e6d9d0e1a758a3ab76005fe55dbcfe050d6a10a5dc37675b07b8f",
            ),
            (
                "anthropic/claude-opus-4-8",
                "claude-opus-4-8",
                "4.8",
                "10504959e51dc76c3563df91ae2eaba57cf814834ecd657232c50f264f9e735e",
            ),
            (
                "openai/gpt-5.6-sol:high",
                "gpt-5.6-sol:high",
                "latest",
                "6cc4a9a80708dbb99c8e4edb6c13be46b2b727a798cb354dcc5598d9004c4acd",
            ),
        ] {
            let row = catalog
                .iter()
                .find(|model| model.id == id)
                .unwrap_or_else(|| panic!("missing exact production row {id}"));
            assert_eq!(row.name, name);
            assert_eq!(row.version, version);
            assert_eq!(
                crate::catalog_identity::row_digest(row).unwrap(),
                expected_digest,
                "CAR row digest drifted for {id}"
            );
        }
    }

    #[test]
    fn weight_repos_are_well_formed_huggingface_ids() {
        for model in builtin_catalog() {
            let Some(repo) = weight_repo(&model.source) else {
                continue;
            };
            assert_eq!(
                repo.split('/').count(),
                2,
                "{}: `{repo}` is not an `org/name` HuggingFace id",
                model.id
            );
            assert!(
                !repo.split('/').any(str::is_empty),
                "{}: `{repo}` has an empty path segment",
                model.id
            );
            assert!(
                !repo.contains(char::is_whitespace),
                "{}: `{repo}` contains whitespace",
                model.id
            );
        }
    }

    /// `param_billions_total` parses a leading number; anything else silently
    /// falls back to a size estimate, so an unparseable value is a scoring bug
    /// that presents as a mysteriously bad recommendation.
    #[test]
    fn param_counts_are_parseable_or_deliberately_empty() {
        for model in builtin_catalog() {
            if weight_repo(&model.source).is_none() || model.param_count.is_empty() {
                continue;
            }
            assert!(
                model.param_count.starts_with(|c: char| c.is_ascii_digit()),
                "{}: param_count `{}` does not start with a number, so the quality \
                 prior cannot read it — leave it empty rather than descriptive",
                model.id,
                model.param_count
            );
        }
    }

    /// CAR supervision is an explicit source contract. Endpoint topology must
    /// never be used to infer ownership.
    #[test]
    fn catalog_vllm_mlx_entries_use_explicit_managed_ownership() {
        for model in builtin_catalog() {
            if !model.is_vllm_mlx() {
                continue;
            }
            assert!(
                model.is_car_managed_vllm_mlx(),
                "{}: a CAR-supervised catalog row must opt into ManagedVllmMlx; \
                 loopback alone cannot confer ownership",
                model.id
            );
        }
    }

    #[test]
    fn generate_capable_models_declare_a_context_window() {
        for model in builtin_catalog() {
            if !model.has_capability(crate::schema::ModelCapability::Generate) {
                continue;
            }
            assert!(
                model.context_length > 0,
                "{}: a generate-capable model with no context_length breaks budget sizing",
                model.id
            );
        }
    }

    #[test]
    fn every_entry_declares_at_least_one_capability() {
        for model in builtin_catalog() {
            assert!(
                !model.capabilities.is_empty(),
                "{}: an entry with no capabilities can never be routed to",
                model.id
            );
        }
    }
}

#[cfg(test)]
mod gguf_quantization_tests {
    use crate::schema::{QuantScheme, Quantization};

    fn quantization_from_gguf_filename(name: &str) -> Option<Quantization> {
        Quantization::from_gguf_filename(name)
    }

    #[test]
    fn reads_the_quantization_a_gguf_file_names() {
        let cases = [
            ("Qwen3-8B-Q4_K_M.gguf", "Q4_K_M", QuantScheme::KQuantMixed),
            (
                "Qwen3-Embedding-0.6B-Q8_0.gguf",
                "Q8_0",
                QuantScheme::RtnBlock,
            ),
            (
                "ggml-large-v3-turbo-q5_0.gguf",
                "q5_0",
                QuantScheme::RtnBlock,
            ),
            ("model-IQ4_XS.gguf", "IQ4_XS", QuantScheme::KQuantMixed),
        ];
        for (filename, label, scheme) in cases {
            let q = quantization_from_gguf_filename(filename)
                .unwrap_or_else(|| panic!("no quantization found in {filename}"));
            assert_eq!(q.label, label, "label for {filename}");
            assert_eq!(q.scheme, scheme, "scheme for {filename}");
        }
    }

    /// A name with nothing quant-shaped in it must yield nothing, not a guess.
    #[test]
    fn returns_none_when_the_name_says_nothing() {
        for filename in ["model.gguf", "llama-2-7b-chat.gguf", "ggml-base.gguf"] {
            assert!(
                quantization_from_gguf_filename(filename).is_none(),
                "should not have guessed from {filename}"
            );
        }
    }

    /// Scanning from the right means a quant-shaped word earlier in the model
    /// name cannot outrank the real suffix.
    #[test]
    fn the_rightmost_match_wins() {
        let q = quantization_from_gguf_filename("q8-experiment-Q4_K_M.gguf").unwrap();
        assert_eq!(q.label, "Q4_K_M");
    }
}

#[cfg(test)]
mod local_availability_tests {
    use super::*;
    use crate::schema::ModelSchema;
    use tempfile::TempDir;

    fn gguf_row(id: &str, hf_repo: &str) -> ModelSchema {
        let mut schema: ModelSchema = serde_json::from_value(serde_json::json!({
            "id": id,
            "name": id.replace('/', "-"),
            "provider": "qwen",
            "family": "qwen3",
            "capabilities": ["generate"],
            "context_length": 32768,
            "param_count": "8B",
            "source": {
                "type": "local",
                "hf_repo": hf_repo,
                "hf_filename": "model.gguf",
                "tokenizer_repo": hf_repo,
            },
            "cost": { "size_mb": 4900 },
        }))
        .unwrap();
        schema.available = false;
        schema
    }

    /// GGUF is the local path on every machine that is not Apple-Silicon-MLX —
    /// i.e. the CUDA machines. `ensure_local` downloads `model.gguf` and
    /// `tokenizer.json` from the declared repo on first use, exactly as the MLX
    /// path does, so a declared repo is functionally available.
    ///
    /// This branch was left out of #164, which gave the MLX path that rule, and
    /// out of whisper.cpp's unconditional `true`. The result was a fresh CUDA
    /// install reporting every local model unavailable, the router dropping
    /// them from fallback chains, and the user routed to cloud with an idle GPU
    /// one download away.
    #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
    #[test]
    fn a_declared_repo_is_available_before_it_is_downloaded() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        std::fs::create_dir_all(&models).unwrap();

        let mut reg = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models);
        reg.register(gguf_row("qwen/test-8b:q4_k_m", "Qwen/Qwen3-8B-GGUF"));

        let model = reg.get("qwen/test-8b:q4_k_m").expect("registered");
        assert!(
            model.available,
            "a GGUF model with a repo to fetch from must not report unavailable \
             just because nothing has downloaded it yet"
        );
    }

    /// The exception, and the reason the rule is `!hf_repo.is_empty()` rather
    /// than blanket `true`: rows from the local-directory scan have nowhere to
    /// fetch from, so they stay gated on the file being physically present.
    #[test]
    fn a_row_with_nowhere_to_fetch_from_stays_unavailable() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let models = tmp.path().join("models");
        std::fs::create_dir_all(&models).unwrap();

        let mut reg = UnifiedRegistry::new_with_state_root(tmp.path().to_path_buf(), models);
        reg.register(gguf_row("local/scanned", ""));

        let model = reg.get("local/scanned").expect("registered");
        assert!(
            !model.available,
            "an empty hf_repo has no download to promise"
        );
    }
}