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::{debug, 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>,
    /// Credential variables whose state the LAST authoritative refresh could
    /// not determine -- the secret store errored rather than answering
    /// "absent". Per-refresh runtime state, deliberately here rather than on
    /// `ModelSchema`, which is catalog content. Read when projecting a row's
    /// `credential_required`: a store failure is not evidence a key is
    /// missing, and rendering "Add key…" for one the person may already hold
    /// is worse than saying nothing (car#1920).
    credential_state_unknown: HashSet<String>,
    /// 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 signed catalog each signed row came from, so a row kept after its
    /// catalog moved on can be re-verified from those exact bytes at the
    /// next startup.
    signed_origin: HashMap<String, std::sync::Arc<crate::catalog::CatalogCacheEnvelope>>,
    /// Signed rows a newer catalog withdrew but something still depends on;
    /// persisted by the engine so a restart keeps them too.
    retained_ids: HashSet<String>,
    /// Ids the current signed catalog revokes. Kept so startup never restores
    /// a retained row the publisher has since revoked.
    revoked_ids: HashSet<String>,
    /// Compiled builtins a revocation marked deprecated; the build's own value
    /// comes back when a later catalog stops revoking them.
    revoked_builtin_ids: HashSet<String>,
    /// Compiled builtin ids whose metadata a signed row currently overlays
    /// (see [`Self::overlay_builtin`]); the next catalog that omits one
    /// restores the compiled values.
    signed_overlay_ids: HashSet<String>,
    /// Exact ids whose current rows came from the signed catalog, so the next
    /// catalog replaces them rather than accreting: a row the publisher
    /// dropped must leave the registry, not linger until restart.
    signed_catalog_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())
}

/// Does a credential exist, in the environment or the OS secret store —
/// **without decrypting it**.
///
/// The distinction is enforced, not stylistic. `car-cli`'s
/// `models_list_explicit_status_refreshes_openrouter_without_value_read` and
/// its `doctor` twin assert that a status surface raises
/// `secret_store_activity().status_attempts` and leaves `get_attempts`
/// unchanged: answering "can I use this model?" is an existence question, and
/// a surface that answers it by reading secret values hands every catalog
/// render a reason to decrypt every provider key on the machine.
///
/// `resolve_env_or_keychain` is the wrong tool here for exactly that reason —
/// it returns the value, and using it made this refresh perform four value
/// reads per render. `status_via_operator_broker_or_keychain` takes the same
/// broker decision and reduces the answer to a boolean before any caller sees
/// it.
///
/// An error — a refused read, a store that will not open — is **not**
/// evidence of absence, but it is also not evidence the model is usable, so it
/// reports `false` the same way a missing key does. That is the conservative
/// direction for a runnability claim: it can under-promise, never over-promise.
pub(crate) fn credential_present_without_reading_it(env_var: &str) -> bool {
    matches!(credential_presence(env_var), CredentialPresence::Present)
}

/// What the store could tell us about a credential, without decrypting it.
///
/// The third answer is the point. Collapsing a store FAILURE into "absent"
/// made a temporarily unreadable key indistinguishable from one that was
/// never set, and the model screen then told the person to add a credential
/// they may already have (car#1920).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CredentialPresence {
    Present,
    Absent,
    /// The store errored. Not evidence either way.
    Unknown,
}

pub(crate) fn credential_presence(env_var: &str) -> CredentialPresence {
    if environment_credential_available(env_var) {
        return CredentialPresence::Present;
    }
    match car_secrets::status_via_operator_broker_or_keychain(&car_secrets::SecretRef::new(
        car_secrets::DEFAULT_SERVICE,
        env_var,
    )) {
        Ok(status) if status.exists => CredentialPresence::Present,
        Ok(_) => CredentialPresence::Absent,
        Err(_) => CredentialPresence::Unknown,
    }
}

/// Process-local hint that an openai-codex sign-in exists. Refreshed by
/// explicit status reads and by each openai-codex request credential lookup,
/// so passive catalog reads never touch the secret store. Relaxed ordering is
/// enough: it is an advisory availability hint, not a synchronisation point.
static CODEX_SIGNED_IN_HINT: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

pub(crate) fn remember_codex_sign_in(present: bool) {
    CODEX_SIGNED_IN_HINT.store(present, std::sync::atomic::Ordering::Relaxed);
}

/// Whether an openai-codex sign-in is known from the process-local hint,
/// without touching the secret store.
pub(crate) fn passive_codex_oauth_available() -> bool {
    CODEX_SIGNED_IN_HINT.load(std::sync::atomic::Ordering::Relaxed)
}

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
        }
        ProprietaryAuth::ChatGptSubscription {} => {
            schema_provider.eq_ignore_ascii_case(OPENAI_CODEX_PROVIDER)
                && source_provider.eq_ignore_ascii_case(OPENAI_CODEX_PROVIDER)
                && passive_codex_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 = crate::catalog::catalog_public_key();
        Self::new_with_catalog_public_key(state_root, models_dir, Some(&catalog_public_key))
    }

    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(),
            credential_state_unknown: HashSet::new(),
            project_model_ids: HashSet::new(),
            signed_catalog_ids: HashSet::new(),
            signed_overlay_ids: HashSet::new(),
            signed_origin: HashMap::new(),
            retained_ids: HashSet::new(),
            revoked_ids: HashSet::new(),
            revoked_builtin_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.
        // Then rows an earlier catalog had that something here still depends
        // on, each re-verified from the catalog it came from.
        let cached = crate::catalog::load_verified(
            &crate::catalog::cache_path(&registry.state_root),
            catalog_public_key,
        );
        if let Some(cached) = cached {
            let origin = std::sync::Arc::new(cached.envelope());
            let revoked: HashSet<String> = cached.revoked().iter().cloned().collect();
            registry.replace_signed_catalog(
                cached.into_models(),
                Some(origin),
                &HashSet::new(),
                &revoked,
            );
        }
        registry.restore_retained(crate::catalog::load_retained(
            &crate::catalog::retained_path(&registry.state_root),
            catalog_public_key,
        ));
        // 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(),
            credential_state_unknown: HashSet::new(),
            project_model_ids: HashSet::new(),
            signed_catalog_ids: HashSet::new(),
            signed_overlay_ids: HashSet::new(),
            signed_origin: HashMap::new(),
            retained_ids: HashSet::new(),
            revoked_ids: HashSet::new(),
            revoked_builtin_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 — and names of
        // models a signed catalog revoked, which must not come back as
        // uncatalogued local ones.
        let known: std::collections::HashSet<String> = self
            .models
            .values()
            .map(|m| m.name.to_ascii_lowercase())
            .chain(
                crate::catalog::load_revoked_names(&crate::catalog::revoked_names_path(
                    &self.state_root,
                ))
                .into_keys()
                .map(|name| 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
    }

    /// Make `rows` the signed catalog, under the one rule both startup and a
    /// live refresh use. Returns how many rows it loaded or overlaid.
    ///
    /// - A row for a compiled builtin id overlays only the builtin's
    ///   [`OVERLAY_FIELDS`](Self::overlay_builtin) — benchmarks, cost,
    ///   performance, deprecation — and only when it names the same weights;
    ///   which weights an id loads stays fixed by the build.
    /// - Any other row is additive: never over another project-owned id, and
    ///   never over a row the user registered, whose definition wins while
    ///   the daemon runs and is what `save_user_config` keeps writing.
    /// - A previous signed row the new catalog drops leaves — unless `keep`
    ///   names it (installed, pinned, a lane default). Then it stays
    ///   resolvable, marked `deprecated` so nothing suggests it, until nothing
    ///   depends on it. Removing it would orphan the weights and blank the
    ///   lane that names it.
    pub(crate) fn replace_signed_catalog(
        &mut self,
        rows: Vec<ModelSchema>,
        origin: Option<std::sync::Arc<crate::catalog::CatalogCacheEnvelope>>,
        keep: &HashSet<String>,
        revoked: &HashSet<String>,
    ) -> usize {
        let incoming: HashSet<&str> = rows.iter().map(|m| m.id.as_str()).collect();
        for id in std::mem::take(&mut self.signed_catalog_ids) {
            // A revoked row goes whatever depends on it: revocation is the
            // publisher saying the weights are bad, not merely superseded.
            if keep.contains(&id) && !incoming.contains(id.as_str()) && !revoked.contains(&id) {
                if let Some(row) = self.models.get_mut(&id) {
                    row.deprecated = true;
                }
                self.retained_ids.insert(id.clone());
                self.signed_catalog_ids.insert(id);
                continue;
            }
            self.models.remove(&id);
            self.project_model_ids.remove(&id);
            self.credential_state_unknown.remove(&id);
            self.signed_origin.remove(&id);
            self.retained_ids.remove(&id);
        }
        let overlaid = std::mem::take(&mut self.signed_overlay_ids);
        let unrevoked = std::mem::take(&mut self.revoked_builtin_ids);
        let restore: HashSet<String> = overlaid.union(&unrevoked).cloned().collect();
        let compiled: HashMap<String, ModelSchema> = if restore.is_empty() {
            HashMap::new()
        } else {
            builtin_catalog()
                .into_iter()
                .filter(|m| restore.contains(&m.id))
                .map(|m| (m.id.clone(), m))
                .collect()
        };
        for id in &restore {
            if let (Some(row), Some(original)) = (self.models.get_mut(id), compiled.get(id)) {
                Self::copy_overlay_fields(original, row);
            }
        }
        self.revoked_ids = revoked.clone();
        let mut loaded = 0;
        for schema in rows {
            let id = schema.id.clone();
            if revoked.contains(&id) {
                warn!(%id, "ignoring a signed catalog row its own catalog revokes");
                continue;
            }
            if self.register_signed_catalog_model(schema) {
                loaded += 1;
                if let Some(origin) = &origin {
                    if self.signed_catalog_ids.contains(&id) {
                        self.signed_origin.insert(id, origin.clone());
                    }
                }
            }
        }
        // A revoked builtin cannot leave the build; it stops being suggested
        // or applied.
        for id in revoked {
            if self.builtin_model_ids.contains(id) {
                if let Some(row) = self.models.get_mut(id) {
                    row.deprecated = true;
                    self.revoked_builtin_ids.insert(id.clone());
                }
            }
        }
        loaded
    }

    /// Whether the current signed catalog revokes `id`.
    pub fn is_revoked(&self, id: &str) -> bool {
        self.revoked_ids.contains(id)
    }

    /// Ids the current signed catalog revokes.
    pub fn revoked_ids(&self) -> Vec<String> {
        self.revoked_ids.iter().cloned().collect()
    }

    /// Whether `id` is a row compiled into this build (signed overlays
    /// included), as opposed to one only a signed catalog or the user added.
    pub fn is_builtin(&self, id: &str) -> bool {
        self.builtin_model_ids.contains(id)
    }

    /// Signed rows kept after their catalog withdrew them, with the catalog
    /// each came from — what the engine persists.
    pub(crate) fn retained_rows(
        &self,
    ) -> Vec<(String, std::sync::Arc<crate::catalog::CatalogCacheEnvelope>)> {
        let mut rows: Vec<_> = self
            .retained_ids
            .iter()
            .filter_map(|id| Some((id.clone(), self.signed_origin.get(id)?.clone())))
            .collect();
        rows.sort_by(|a, b| a.0.cmp(&b.0));
        rows
    }

    /// The signed catalog a row came from, if it is a signed row.
    pub(crate) fn signed_origin_of(
        &self,
        id: &str,
    ) -> Option<std::sync::Arc<crate::catalog::CatalogCacheEnvelope>> {
        self.signed_origin.get(id).cloned()
    }

    /// Bring back verified rows as retained — deprecated, resolvable, never
    /// suggested — unless the id is already present. Startup restores what
    /// the last run kept; a pull that outlived its row's catalog keeps it.
    pub(crate) fn restore_retained(
        &mut self,
        rows: Vec<(
            ModelSchema,
            std::sync::Arc<crate::catalog::CatalogCacheEnvelope>,
        )>,
    ) {
        for (mut schema, origin) in rows {
            if self.models.contains_key(&schema.id) || self.revoked_ids.contains(&schema.id) {
                continue;
            }
            schema.deprecated = true;
            let id = schema.id.clone();
            if self.register_signed_catalog_model(schema) && self.signed_catalog_ids.contains(&id) {
                self.signed_origin.insert(id.clone(), origin);
                self.retained_ids.insert(id);
            }
        }
    }

    /// The fields a signed row may change on a compiled builtin: what CAR
    /// measured or priced about a model, never which weights it loads.
    fn copy_overlay_fields(from: &ModelSchema, to: &mut ModelSchema) {
        to.public_benchmarks = from.public_benchmarks.clone();
        to.cost = from.cost.clone();
        to.performance = from.performance.clone();
        to.deprecated = from.deprecated;
    }

    /// Apply a signed row's measurements to the compiled builtin it names.
    /// Refused when the row names different weights: a publisher may update
    /// what CAR knows about a builtin, not repoint it.
    fn overlay_builtin(&mut self, schema: &ModelSchema) -> bool {
        let Some(row) = self.models.get_mut(&schema.id) else {
            return false;
        };
        let same_weights =
            serde_json::to_value(&row.source).ok() == serde_json::to_value(&schema.source).ok();
        if !same_weights {
            warn!(id = %schema.id, "ignoring signed overlay that names different weights for a compiled builtin");
            return false;
        }
        Self::copy_overlay_fields(schema, row);
        self.signed_overlay_ids.insert(schema.id.clone());
        true
    }

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

    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
                {
                    // A row the in-process loader runs needs the Swift MLX
                    // stack linked into THIS binary; the target cfg above does
                    // not imply it (`CAR_BUILD_SWIFT_LM=0`, or a Swift build
                    // that failed and warned). A row marked available without
                    // it is one the router puts in a fallback chain and then
                    // fails on. Everything else under `ModelSource::Mlx` is a
                    // subprocess and does not care — see
                    // `mlx_row_needs_in_process_loader`.
                    //
                    // Given its executor, available if cached locally OR it
                    // declares an hf_repo — ensure_local() lazy-downloads on
                    // first use, so that is "functionally available" the same
                    // way Ollama/RemoteApi entries are (#164).
                    let loader_ok = !mlx_row_needs_in_process_loader(&schema)
                        || crate::backend::local::in_process_loader_available();
                    let mlx_dir = self.models_dir.join(&schema.name);
                    loader_ok && (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 { .. }
                        | ProprietaryAuth::ChatGptSubscription {} => 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);
        // `debug`, not `info`: this fires once per catalog entry, and the
        // catalog is ~87 rows. At the default filter that put 87 lines of
        // bookkeeping in front of whatever the person actually asked for —
        // `car do "What is 2 plus 2?"`, `car setup`, `car doctor`.
        //
        // car#1788 quieted it per-command, which needs a new entry in
        // `should_quiet_tracing_for_machine_output` for every command that
        // builds a registry; `do` and `setup` were missed and reported as
        // car#1873. Registering one known model is not an event worth an
        // `info` line, so lowering it fixes every command at once and cannot
        // be missed by a new one. `RUST_LOG=debug` still shows the full set.
        debug!(
            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,
        });
        // Record, don't infer: only the files this acquisition actually
        // fetched — never a snapshot it reused, which may be the user's own.
        let (result, fetched) = crate::retire::collecting_fetches(self.ensure_local_inner(
            id,
            sink,
            force,
            managed_dir_override,
        ))
        .await;
        crate::retire::record_download(
            &self.state_root,
            &crate::hf_cache::hub_dir(),
            &model_id,
            &fetched,
        );
        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 is_diffusers = mlx_schema_is_diffusers(schema);

                if mlx_layout_complete(is_diffusers, &model_dir) {
                    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_layout_complete(is_diffusers, 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 = mlx_schema_is_diffusers(schema);

                // 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 a non-empty root config.json, the
                // same bar `download_file` applies per file, so a zero-byte or
                // dangling config is re-fetched below instead of reused;
                // diffusers has none, so recursive weights-present is its
                // completeness signal.
                if !force && mlx_layout_complete(is_diffusers, &model_dir) {
                    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 config or 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_layout_complete(is_diffusers, 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, the non-secret Parslee authority hint,
    /// and the process-local Codex sign-in 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 status request**, resolving
    /// credentials authoritatively (environment *or* OS keychain).
    ///
    /// [`Self::refresh_availability`] is deliberately passive — environment
    /// presence only — because it runs on startup and catalog paths where a
    /// physical secret-store read per provider would be wrong. That is right
    /// for those callers and wrong for a user who typed `car models list`:
    /// with a key held in `car secrets` rather than exported, the passive
    /// verdict reports a perfectly usable model as `RUNNABLE no`. Measured on
    /// a machine with `OPENAI_API_KEY` in the keychain and nothing in the
    /// environment, the embedded list path reported `openai/gpt-5.5` as not
    /// runnable while inference through the same binary worked.
    ///
    /// Use this only for surfaces a person explicitly asked for a status
    /// from, never on a startup or catalog path. It performs one secret-store
    /// read per DISTINCT credential name (deduplicated in
    /// `refresh_availability_with`), not one per model.
    pub fn refresh_availability_for_explicit_status(&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(),
            true,
        );
    }

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

    /// Request-local OAuth availability must not inherit a different organization's
    /// credential or gateway verdict. Other providers retain their usual checks.
    pub(crate) fn refresh_work_context_availability(&mut self, api_base: &str, available: bool) {
        self.refresh_availability();
        for schema in self.models.values_mut() {
            if schema.provider.eq_ignore_ascii_case("parslee") {
                if let ModelSource::Proprietary {
                    provider,
                    endpoint,
                    auth: ProprietaryAuth::OAuth2Pkce { .. },
                    ..
                } = &mut schema.source
                {
                    if provider.eq_ignore_ascii_case("parslee") {
                        *endpoint = api_base.trim_end_matches('/').to_string();
                        schema.available = available;
                    }
                }
            }
        }
    }

    /// Whether the last authoritative refresh failed to determine this
    /// credential variable's state, as opposed to finding it absent.
    pub fn credential_state_is_unknown(&self, env_var: &str) -> bool {
        self.credential_state_unknown.contains(env_var)
    }

    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;
        let mut needs_codex = 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::ChatGptSubscription {} => needs_codex = true,
                    ProprietaryAuth::OAuth2Pkce { .. } => {}
                },
                _ => {}
            }
        }
        let mut credential_unknown: std::collections::HashSet<String> =
            std::collections::HashSet::new();
        // Carried on the registry rather than on each row: `ModelSchema` is
        // catalog content and this is per-refresh runtime state.  See
        // `credential_state_unknown` for what reads it.
        // The Codex record is not an environment credential. Explicit status
        // reads check its existence once through car-auth and refresh the
        // process-local hint; passive catalog reads use only that hint.
        if authoritative_credentials && needs_codex {
            remember_codex_sign_in(car_auth::codex_credential_present_without_reading_it());
        }
        let credential_available: std::collections::HashMap<String, bool> = credential_envs
            .into_iter()
            .map(|env| {
                let available = if authoritative_credentials {
                    // A store failure is not absence. The row still reads
                    // unavailable -- conservative for a runnability claim --
                    // but it must not go on to advise adding a key (car#1920).
                    match credential_presence(&env) {
                        CredentialPresence::Present => true,
                        CredentialPresence::Absent => false,
                        CredentialPresence::Unknown => {
                            credential_unknown.insert(env.clone());
                            false
                        }
                    }
                } else {
                    environment_credential_available(&env)
                };
                (env, available)
            })
            .collect();
        self.credential_state_unknown = credential_unknown;
        // 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");

                        // Only a row the in-process loader runs needs the
                        // Swift MLX stack linked into THIS binary; the target
                        // cfg does not imply it (`CAR_BUILD_SWIFT_LM=0`, or a
                        // Swift build that failed and warned). Both availability
                        // paths ask the same helper, because they are required
                        // to agree — a model whose `available` flips depending
                        // on which ran last is worse than either answer.
                        let in_process_loader_present = !mlx_row_needs_in_process_loader(m)
                            || crate::backend::local::in_process_loader_available();

                        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 —
                            // 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);
                            in_process_loader_present
                                && (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();
        // A floor, not the gate. This answers "does this directory hold a
        // causal-LM checkpoint?" — a question about the *layout*, asked while
        // scanning a local directory on every platform, so it cannot be
        // delegated to a macOS-only loader. The loader's own answer is added
        // below, which is what lets an architecture `mlx-swift-lm` knows about
        // be recognized here without anyone editing this list.
        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",
        ];
        let recognized = KNOWN_LLM_TYPES.iter().any(|t| model_type == *t)
            || crate::backend::local::has_native_backend(&model_type);
        if !recognized {
            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()
}

/// Whether an MLX-sourced speech model can actually run: the managed
/// `mlx-audio` runtime has to be provisioned.
///
/// Platform-independent, and it used to answer an unconditional `true` on Apple
/// Silicon on the grounds that "speech uses native MLX backends — no Python CLI
/// needed". Those backends (`mlx_kokoro`, `mlx_parakeet`) were deleted with the
/// Swift migration, so that branch promised a path that no longer exists —
/// which is the same false `available: true` car#649 was filed about, reached
/// from the other side. The managed runtime is now the only way an
/// `mlx/kokoro-*`, `mlx/parakeet-*` or `mlx/qwen3-tts-*` entry runs anywhere.
///
/// Only `ModelSource::Mlx` / `ManagedVllmMlx` rows reach this; the cross-
/// platform `WhisperCpp` STT entry, Windows OS TTS and the remote ElevenLabs
/// rows are filtered out by the caller's `is_mlx()` guard and keep their own
/// availability.
// Both call sites sit inside `cfg(macos + aarch64 + not(car_skip_mlx))` blocks;
// the else-arm marks every MLX row unavailable without asking, because there is
// no MLX there to ask about. So this is genuinely dead off Apple Silicon.
#[allow(dead_code)]
fn speech_mlx_available() -> bool {
    // Via `venv_program`, so this agrees with `SpeechRuntime` about where a
    // provisioned runtime keeps its console scripts — `Scripts\*.exe` on
    // Windows, `bin/*` elsewhere.
    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()
}

// Dead off Apple Silicon for the same reason as its only caller above.
#[allow(dead_code)]
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,
    /// The credential this row is waiting on, when a missing credential is the
    /// ONLY reason it is unavailable.
    ///
    /// `Some(env_name)` means: this is a remote row, it is not available, and
    /// the single thing standing between the user and using it is that named
    /// credential. `None` means everything else — the row is available, or it
    /// is local, or it is unavailable for a reason a credential will not fix
    /// (deprecated, a gateway with no upstream, the managed
    /// `parslee/openrouter/*` rows whose availability is a session-plus-upstream
    /// question rather than a key).
    ///
    /// **The daemon answers this so the client never has to guess it.** The
    /// obvious client-side shortcut — "a remote row with `available: false`
    /// probably needs a key" — is wrong for every case in the previous
    /// paragraph, and the protocol already tells clients not to reconstruct
    /// state from `available` alone. The alternative, a provider-to-variable
    /// map in the UI, duplicates the catalog that owns `api_key_env`.
    ///
    /// Naming the variable is not a disclosure: `ANTHROPIC_API_KEY` is the
    /// public name of an environment variable, carries no value, and is
    /// already printed by `car models list` and every provider's own docs.
    ///
    /// **Only ever populated on an explicit status read**, never on a passive
    /// one. A passive refresh resolves credentials from the process
    /// environment alone, so a key held in the OS secret store reads as absent
    /// — and a row naming a credential the user already has, telling them to
    /// add it, is the one piece of advice worse than none. The field's
    /// contract is "a missing credential is the SOLE reason this row is
    /// unavailable"; a passive refresh cannot establish that, so it does not
    /// claim it. `available` is left alone and remains passive-accurate on
    /// that path.
    ///
    /// Serialized as `null` rather than omitted, matching every other optional
    /// field on this row (`management_evidence`, `estimated_peak_mb`). The row
    /// has a fixed key set, and `tests/models_surface_gate.rs` asserts it
    /// exactly — a conditionally-present key would make that gate
    /// unmaintainable and hand clients a shape that changes per row.
    #[serde(default)]
    pub credential_required: Option<String>,
    /// Actionable provider setup guidance; absent when no specific remedy is known.
    #[serde(default)]
    pub unavailable_reason: Option<String>,
    /// 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>,
}

/// Explain why a listed model is unavailable when CAR has one actionable,
/// provider-specific answer.
pub fn model_unavailable_reason(info: &ModelInfo) -> Option<&'static str> {
    (!info.available && info.provider.eq_ignore_ascii_case(OPENAI_CODEX_PROVIDER))
        .then_some(crate::schema::OPENAI_CODEX_SIGN_IN_HINT)
}

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 {
        let mut info = 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,
            credential_required: s.credential_required(),
            unavailable_reason: None,
            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()),
        };
        info.unavailable_reason = model_unavailable_reason(&info).map(str::to_owned);
        info
    }
}

/// 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 = crate::hf_cache::api()
        .build()
        .map_err(|e| InferenceError::DownloadFailed(e.to_string()))?;

    // hf-hub's `get` returns an already-cached file without fetching; only a
    // file the cache did not hold is one CAR downloaded.
    let already_cached = hf_hub::Cache::new(crate::hf_cache::hub_dir())
        .repo(hf_hub::Repo::model(repo.to_string()))
        .get(filename)
        .is_some();
    let repo_id = repo;
    let repo = api.model(repo.to_string());
    let path = repo
        .get(filename)
        .await
        .map_err(|e| InferenceError::DownloadFailed(format!("{filename}: {e}")))?;
    if !already_cached {
        crate::retire::note_fetched(repo_id, filename, crate::retire::snapshot_revision(&path));
    }

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

/// The Flux checkpoint whose T5 tokenizer the MLX Flux row fetches from its base
/// model: `(mlx repo, mlx model name, base repo, file)`. One definition for the
/// downloader here and for [`crate::retire`]'s ownership, which must know that
/// CAR takes one file from the base repo and owns nothing else in it.
pub(crate) const FLUX_AUXILIARY: (&str, &str, &str, &str) = (
    "mlx-community/Flux-1.lite-8B-MLX-Q4",
    "Flux-1.lite-8B-MLX-Q4",
    "Freepik/flux.1-lite-8B",
    "tokenizer_2/tokenizer.json",
);

/// The text encoder every LTX video generation loads. CAR passes it to the
/// `ltx-2-mlx` runtime explicitly rather than inheriting the runtime's default,
/// so what CAR's video rows fetch is stated here — for the runtime, and for
/// [`crate::retire`], which would otherwise see an 8 GB repo nothing references
/// and report it as the user's.
pub(crate) const LTX_TEXT_ENCODER: &str = "mlx-community/gemma-3-12b-it-4bit";

/// Whether `schema` generates video through the `ltx-2-mlx` runtime — every
/// MLX video row does.
pub(crate) fn uses_ltx_text_encoder(schema: &ModelSchema) -> bool {
    matches!(schema.source, ModelSource::Mlx { .. })
        && schema.has_capability(ModelCapability::VideoGeneration)
}

pub(crate) fn uses_flux_auxiliary(model_name: &str, hf_repo: &str) -> bool {
    hf_repo == FLUX_AUXILIARY.0 || model_name == FLUX_AUXILIARY.1
}

fn auxiliary_mlx_files_missing(model_name: &str, hf_repo: &str, model_dir: &Path) -> bool {
    uses_flux_auxiliary(model_name, hf_repo)
        && !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(FLUX_AUXILIARY.2, FLUX_AUXILIARY.3, &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(&crate::hf_cache::repo_dir_in(
                    root, hf_repo,
                )),
                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 {
    mlx_layout_complete(mlx_schema_is_diffusers(schema), dir)
        && mlx_auxiliary_ready_without_download(&schema.name, dir)
}

/// Whether a row stores its weights in the diffusers component layout (Flux
/// image generation, LTX video): component subdirectories with no root
/// `config.json` and no top-level weight index.
fn mlx_schema_is_diffusers(schema: &ModelSchema) -> bool {
    schema.capabilities.iter().any(|capability| {
        matches!(
            capability,
            ModelCapability::ImageGeneration | ModelCapability::VideoGeneration
        )
    })
}

/// Root metadata plus weights: the one layout test shared by every reuse
/// gate and by the installed-state projection, so the two can never disagree
/// about the same directory. A standard MLX directory needs a `config.json`
/// that resolves and is non-empty (`cache_file_usable`, the bar each
/// per-file download applies) and every weight file present; a zero-byte or
/// dangling config is therefore re-fetched rather than reused. Diffusers
/// layouts have no root config, so their recursively validated component
/// weights are the whole signal.
fn mlx_layout_complete(is_diffusers: bool, dir: &Path) -> bool {
    (is_diffusers || crate::download::cache_file_usable(&dir.join("config.json")))
        && mlx_dir_has_weights(dir)
}

/// Whether a `ModelSource::Mlx` row is executed by the **in-process** Swift
/// loader, as opposed to one of the subprocesses that also read MLX weights.
///
/// `ModelSource::Mlx` is a statement about the weight format, not about who
/// runs it. Five executors share it today: the in-process text loader, the
/// managed `mlx-audio` venv (speech), `mflux` (image), `ltx-2-mlx` (video) and
/// the `mlx_vlm` Python CLI. Only the first needs the Swift stack linked into
/// this binary, and gating the other four on it marked working subprocess
/// models unavailable — `car image` and `car video` lose their only local
/// backend on a machine whose Swift build merely failed.
///
/// Both availability paths call this, because they are required to agree: a
/// model whose `available` flips depending on whether registration or refresh
/// ran last is worse than either answer.
// Both call sites sit inside `cfg(macos + aarch64 + not(car_skip_mlx))`; the
// else-arm marks every MLX row unavailable without asking, because there is no
// MLX there to ask about. So this is genuinely dead off Apple Silicon — as the
// `cargo check --cfg=car_skip_mlx` gate in ci-local.sh reported the moment it
// was added.
#[allow(dead_code)]
fn mlx_row_needs_in_process_loader(schema: &ModelSchema) -> bool {
    use crate::schema::ModelCapability as C;
    let tagged = |t: &str| schema.tags.iter().any(|x| x == t);
    if tagged("speech") || tagged("requires-mlx-vlm") {
        return false;
    }
    if schema
        .capabilities
        .iter()
        .any(|c| matches!(c, C::ImageGeneration | C::VideoGeneration))
    {
        return false;
    }
    matches!(schema.source, ModelSource::Mlx { .. })
}

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

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

pub(crate) fn latest_huggingface_repo_snapshot(repo_id: &str) -> Option<PathBuf> {
    let repo_dir = crate::hf_cache::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
}

/// Credential environment names declared by CAR's built-in model catalog.
///
/// This is a value-free catalog projection for child-process scrubbing. It
/// does not construct a live registry, probe credentials, read caches, or
/// perform discovery I/O.
pub fn builtin_credential_env_names() -> std::collections::BTreeSet<String> {
    let mut names = std::collections::BTreeSet::new();
    for model in builtin_catalog() {
        match model.source {
            ModelSource::RemoteApi {
                api_key_env,
                api_key_envs,
                ..
            } => {
                names.insert(api_key_env);
                names.extend(api_key_envs);
            }
            ModelSource::Proprietary { auth, .. } => match auth {
                ProprietaryAuth::ApiKeyEnv { env_var }
                | ProprietaryAuth::BearerTokenEnv { env_var } => {
                    names.insert(env_var);
                }
                ProprietaryAuth::OAuth2Pkce { .. } | ProprietaryAuth::ChatGptSubscription {} => {}
            },
            ModelSource::Local { .. }
            | ModelSource::CodexCli { .. }
            | ModelSource::Ollama { .. }
            | ModelSource::Mlx { .. }
            | ModelSource::WhisperCpp { .. }
            | ModelSource::WindowsSpeech {}
            | ModelSource::VllmMlx { .. }
            | ModelSource::ManagedVllmMlx { .. }
            | ModelSource::AppleFoundationModels { .. }
            | ModelSource::Delegated { .. } => {}
        }
    }
    names
}

/// Report whether a built-in model is paid for by a ChatGPT subscription.
///
/// Some callers hold only the durable model id, not a [`UnifiedRegistry`]. In
/// particular, the coder A/B harness reads a completed session after the
/// registry that ran it may be gone. Giving those callers a narrow catalog
/// lookup lets them render `subscription` instead of inventing a dollar cost,
/// without starting a registry merely to recover billing provenance.
pub fn is_builtin_subscription_billed(model_id: &str) -> bool {
    let latest_id = (!model_id.contains(':')).then(|| format!("{model_id}:latest"));
    builtin_catalog().into_iter().any(|model| {
        (model.id == model_id || latest_id.as_deref() == Some(model.id.as_str()))
            && model.is_subscription_billed()
    })
}

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

    /// **`credential_required` must only ever name a credential that would
    /// actually fix the row.**
    ///
    /// The field exists so a client never has to infer "this needs a key"
    /// from `available: false`. That inference is wrong for several real
    /// shapes, and saying "add ANTHROPIC_API_KEY" to someone whose row is
    /// deprecated, OAuth-backed, or gated on a gateway upstream is worse than
    /// saying nothing — it sends them to do something that cannot help.
    /// car#1920. A secret store that ERRORS is not a store that said "absent",
    /// and the difference decides whether the model screen tells someone to add
    /// a credential they may already hold. The registry records the
    /// indeterminate case so the projection can stay silent; the row itself
    /// remains unavailable, which is the conservative direction.
    #[test]
    fn an_unreadable_credential_is_recorded_as_unknown_not_absent() {
        // The default registry has run no authoritative refresh, so it claims
        // nothing is unknown -- the value that lets `credential_required`
        // behave exactly as before wherever this state is not established.
        let registry = UnifiedRegistry::new_empty(std::path::PathBuf::from("/nonexistent"));
        assert!(
            !registry.credential_state_is_unknown("VENDOR_API_KEY"),
            "an unprobed registry must not claim a credential state is unknown"
        );

        // And the three-way probe keeps the distinction the bool could not:
        // a name that cannot exist in any store is ABSENT, not unknown, so a
        // genuinely missing key still gets its "add this key" advice.
        assert_eq!(
            credential_presence("CAR_TEST_CREDENTIAL_THAT_DOES_NOT_EXIST"),
            CredentialPresence::Absent
        );
    }

    #[test]
    fn credential_required_names_a_key_only_when_a_key_is_the_answer() {
        let remote = |available: bool, deprecated: bool| {
            let mut m = test_generate_schema(
                "vendor/m",
                "m",
                ModelSource::RemoteApi {
                    protocol: crate::schema::ApiProtocol::OpenAiCompat,
                    endpoint: "https://example.invalid/v1".into(),
                    api_key_env: "VENDOR_API_KEY".into(),
                    api_key_envs: vec![],
                    api_version: None,
                },
            );
            m.provider = "vendor".into();
            m.available = available;
            m.deprecated = deprecated;
            m
        };

        assert_eq!(
            remote(false, false).credential_required().as_deref(),
            Some("VENDOR_API_KEY"),
            "an unavailable plain remote row is exactly the case this field is for"
        );

        assert_eq!(
            remote(true, false).credential_required(),
            None,
            "an available row needs nothing"
        );
        assert_eq!(
            remote(false, true).credential_required(),
            None,
            "a deprecated row does not become usable by adding a key"
        );

        // Several alternative variables: telling someone to set one of four
        // names is not actionable, and choosing for them guesses which
        // account they hold.
        let mut multi = remote(false, false);
        if let ModelSource::RemoteApi {
            ref mut api_key_envs,
            ..
        } = multi.source
        {
            *api_key_envs = vec!["VENDOR_A".into(), "VENDOR_B".into()];
        }
        assert_eq!(multi.credential_required(), None);

        // OpenRouter resolves through a credential SOURCE (pasted key or
        // OAuth), not a variable a person types.
        let mut router = remote(false, false);
        router.source = 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,
        };
        assert_eq!(router.credential_required(), None);

        // A local row has no credential at all.
        let mut local = test_generate_schema(
            "mlx/local",
            "local",
            ModelSource::Mlx {
                hf_repo: "org/repo".into(),
                hf_weight_file: None,
            },
        );
        local.available = false;
        assert_eq!(local.credential_required(), None);
    }

    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(car_mlxlm_swift_built)]
        {
            // With the Swift MLX backend built, 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(car_mlxlm_swift_built))]
        {
            // Without the Swift MLX backend it is neither runnable nor
            // installed — still not a contradiction.
            assert!(
                !registered.available,
                "the Swift MLX backend is absent, 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"
        );
    }

    /// **An explicit status request resolves env-or-keychain; the passive
    /// refresh resolves the environment alone.**
    ///
    /// The distinction is deliberate — the passive refresh runs on startup and
    /// catalog paths, where a secret-store read per provider would be wrong —
    /// but it reached a user-facing column. With a key kept in `car secrets`
    /// rather than exported, `car models list` reported a usable model as
    /// `RUNNABLE no`; measured with `OPENAI_API_KEY` in the keychain and
    /// nothing in the environment, `openai/gpt-5.5` read as not runnable while
    /// inference through the same binary worked.
    ///
    /// What this pins is the ENVIRONMENT half, which is all a test can reach:
    /// both modes must agree when the credential is exported, so the shared
    /// machinery is wired and the explicit mode has not silently become a
    /// no-op. The keychain half needs a populated OS store and cannot run in
    /// CI; it was verified by hand against the state above.
    #[test]
    fn an_explicit_status_refresh_sees_the_same_environment_credential() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let env_var = "CAR_TEST_EXPLICIT_STATUS_KEY";
        let mut registry = UnifiedRegistry::new_empty(tmp.path().join("models"));
        let mut schema = test_generate_schema(
            "vendor/explicit-status",
            "explicit-status",
            ModelSource::RemoteApi {
                protocol: crate::schema::ApiProtocol::OpenAiCompat,
                endpoint: "https://example.invalid/v1".into(),
                api_key_env: env_var.into(),
                api_key_envs: vec![],
                api_version: None,
            },
        );
        schema.provider = "vendor".into();
        registry.register(schema);

        std::env::remove_var(env_var);
        registry.refresh_availability();
        assert!(
            !registry.get("vendor/explicit-status").unwrap().available,
            "no credential anywhere: the passive refresh must report unavailable"
        );
        registry.refresh_availability_for_explicit_status();
        assert!(
            !registry.get("vendor/explicit-status").unwrap().available,
            "no credential anywhere: the explicit refresh must agree"
        );

        std::env::set_var(env_var, "sk-present");
        registry.refresh_availability();
        assert!(
            registry.get("vendor/explicit-status").unwrap().available,
            "exported credential: the passive refresh must see it"
        );
        registry.refresh_availability_for_explicit_status();
        assert!(
            registry.get("vendor/explicit-status").unwrap().available,
            "exported credential: the explicit refresh must see it too — if this fails \
             the authoritative mode is resolving through a path that cannot see the \
             environment, and `car models list` is answering from something else again"
        );
        std::env::remove_var(env_var);
    }

    #[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 = crate::hf_cache::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")
        );
    }

    fn signed_remote(id: &str, name: &str) -> ModelSchema {
        test_generate_schema(
            id,
            name,
            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,
            },
        )
    }

    #[test]
    fn a_user_registered_row_wins_a_live_signed_swap_and_stays_in_models_json() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
            None,
        );
        let mut mine = signed_remote("shared/id", "mine");
        mine.context_length = 1234;
        registry.register_user_model(mine);
        let mut publishers = signed_remote("shared/id", "publishers");
        publishers.context_length = 9999;
        assert_eq!(
            registry.replace_signed_catalog(
                vec![publishers],
                None,
                &HashSet::new(),
                &HashSet::new()
            ),
            0
        );
        assert_eq!(registry.get("shared/id").unwrap().context_length, 1234);
        registry.save_user_config().unwrap();
        let saved = std::fs::read_to_string(tmp.path().join(USER_MODELS_FILE)).unwrap();
        assert!(saved.contains("1234") && !saved.contains("9999"), "{saved}");
        assert!(registry.unregister_user_model("shared/id").is_some());
    }

    #[test]
    fn a_withdrawn_row_something_depends_on_stays_deprecated_until_nothing_does() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
            None,
        );
        let used = signed_remote("signed/used", "used");
        let idle = signed_remote("signed/idle", "idle");
        assert_eq!(
            registry.replace_signed_catalog(
                vec![used, idle],
                None,
                &HashSet::new(),
                &HashSet::new()
            ),
            2
        );
        let keep: HashSet<String> = ["signed/used".to_string()].into();
        assert_eq!(
            registry.replace_signed_catalog(vec![], None, &keep, &HashSet::new()),
            0
        );
        assert!(
            registry.get("signed/idle").is_none(),
            "nothing depended on it"
        );
        let kept = registry.get("signed/used").expect("a dependent keeps it");
        assert!(kept.deprecated, "kept, but never suggested");
        // Republished: it comes back current.
        let used = signed_remote("signed/used", "used");
        assert_eq!(
            registry.replace_signed_catalog(vec![used], None, &keep, &HashSet::new()),
            1
        );
        assert!(!registry.get("signed/used").unwrap().deprecated);
        // Nothing depends on it any more: the next catalog that omits it drops it.
        registry.replace_signed_catalog(vec![], None, &HashSet::new(), &HashSet::new());
        assert!(registry.get("signed/used").is_none());
    }

    #[test]
    fn a_kept_row_survives_a_restart_only_while_its_signed_catalog_verifies() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let open = |key: Option<&str>| {
            UnifiedRegistry::new_with_catalog_public_key(
                tmp.path().to_path_buf(),
                tmp.path().join("models"),
                key,
            )
        };
        let cache = crate::catalog::cache_path(tmp.path());
        let retained = crate::catalog::retained_path(tmp.path());
        let catalog = |models, version| {
            crate::catalog::signed_test_catalog(
                crate::catalog::CatalogDoc {
                    revoked: Vec::new(),
                    version,
                    models,
                },
                21,
            )
        };
        let (v1, key) = catalog(
            vec![
                signed_remote("signed/used", "used"),
                signed_remote("signed/idle", "idle"),
            ],
            1,
        );
        crate::catalog::save_verified(&cache, &v1).unwrap();
        let mut running = open(Some(&key));
        assert!(running.get("signed/used").is_some());

        // v2 withdraws both; something depends on `used`.
        let (v2, _) = catalog(vec![], 2);
        let keep: HashSet<String> = ["signed/used".to_string()].into();
        running.replace_signed_catalog(
            v2.clone().into_models(),
            Some(std::sync::Arc::new(v2.envelope())),
            &keep,
            &HashSet::new(),
        );
        crate::catalog::save_retained(&retained, &running.retained_rows()).unwrap();
        crate::catalog::save_verified(&cache, &v2).unwrap();

        let restarted = open(Some(&key));
        let kept = restarted.get("signed/used").expect("kept across a restart");
        assert!(kept.deprecated);
        assert!(restarted.get("signed/idle").is_none());
        // Another key: nothing verifies, nothing comes back.
        let (_, other_key) = crate::catalog::signed_test_catalog(
            crate::catalog::CatalogDoc {
                revoked: Vec::new(),
                version: 9,
                models: vec![],
            },
            22,
        );
        assert!(open(Some(&other_key)).get("signed/used").is_none());
        // A tampered file: the envelope no longer verifies.
        let json = std::fs::read_to_string(&retained).unwrap();
        std::fs::write(&retained, json.replace("signed/idle", "signed/idlx")).unwrap();
        assert!(open(Some(&key)).get("signed/used").is_none());
    }

    #[test]
    fn a_revoked_row_leaves_even_when_something_depends_on_it() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
            None,
        );
        let none = HashSet::new();
        let bad: HashSet<String> = ["signed/bad".to_string()].into();
        registry.replace_signed_catalog(
            vec![signed_remote("signed/bad", "bad")],
            None,
            &none,
            &none,
        );
        // Withdrawn, but a lane depends on it: an ordinary withdrawal keeps it…
        registry.replace_signed_catalog(vec![], None, &bad, &none);
        assert!(registry.get("signed/bad").is_some());
        // …a revocation does not, even with nothing else changing.
        registry.replace_signed_catalog(vec![], None, &bad, &bad);
        assert!(registry.get("signed/bad").is_none());
        // And a catalog never registers a row it revokes itself.
        registry.replace_signed_catalog(
            vec![signed_remote("signed/bad", "bad")],
            None,
            &none,
            &bad,
        );
        assert!(registry.get("signed/bad").is_none());
        assert!(registry.retained_rows().is_empty());

        // A revoked builtin is deprecated, and restored once it is not.
        let builtin = builtin_catalog()
            .into_iter()
            .find(|m| !m.deprecated)
            .expect("a current builtin");
        let revoked: HashSet<String> = [builtin.id.clone()].into();
        registry.replace_signed_catalog(vec![], None, &none, &revoked);
        assert!(registry.get(&builtin.id).unwrap().deprecated);
        registry.replace_signed_catalog(vec![], None, &none, &none);
        assert!(!registry.get(&builtin.id).unwrap().deprecated);
    }

    #[test]
    fn a_restart_never_restores_a_row_the_current_catalog_revokes() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let catalog = |models, revoked, version| {
            crate::catalog::signed_test_catalog(
                crate::catalog::CatalogDoc {
                    version,
                    models,
                    revoked,
                },
                23,
            )
        };
        let (v1, key) = catalog(vec![signed_remote("signed/kept", "kept")], vec![], 1);
        let mut running = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
            Some(&key),
        );
        running.replace_signed_catalog(
            v1.clone().into_models(),
            Some(std::sync::Arc::new(v1.envelope())),
            &HashSet::new(),
            &HashSet::new(),
        );
        // v2 withdraws it while a lane depends on it: kept and recorded.
        let (v2, _) = catalog(vec![], vec![], 2);
        let keep: HashSet<String> = ["signed/kept".to_string()].into();
        running.replace_signed_catalog(
            vec![],
            Some(std::sync::Arc::new(v2.envelope())),
            &keep,
            &HashSet::new(),
        );
        let retained = crate::catalog::retained_path(tmp.path());
        crate::catalog::save_retained(&retained, &running.retained_rows()).unwrap();
        // v3 revokes it; the daemon restarts before anything rewrites the file.
        let (v3, _) = catalog(vec![], vec!["signed/kept".to_string()], 3);
        crate::catalog::save_verified(&crate::catalog::cache_path(tmp.path()), &v3).unwrap();
        let restarted = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
            Some(&key),
        );
        assert!(restarted.get("signed/kept").is_none());
    }

    #[test]
    fn a_signed_overlay_updates_a_builtins_measurements_never_its_weights() {
        let _environment = crate::openrouter::test_environment_scope();
        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_with_catalog_public_key(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
            None,
        );
        let compiled = builtin_catalog()
            .into_iter()
            .find(|m| m.public_benchmarks.iter().any(|b| b.name == "car-judged"))
            .expect("a scored builtin");
        let mut remeasured = compiled.clone();
        remeasured.public_benchmarks[0].score = 0.42;
        remeasured.public_benchmarks[0].runs = Some(3);
        remeasured.context_length = compiled.context_length + 1;
        remeasured.name = "renamed by the publisher".into();
        assert_eq!(
            registry.replace_signed_catalog(
                vec![remeasured.clone()],
                None,
                &HashSet::new(),
                &HashSet::new()
            ),
            1
        );
        let row = registry.get(&compiled.id).unwrap();
        assert_eq!(row.public_benchmarks[0].score, 0.42);
        assert_eq!(row.public_benchmarks[0].runs, Some(3));
        assert_eq!(
            row.context_length, compiled.context_length,
            "not an overlay field"
        );
        assert_eq!(row.name, compiled.name, "not an overlay field");
        // A catalog that omits it restores what the build shipped.
        registry.replace_signed_catalog(vec![], None, &HashSet::new(), &HashSet::new());
        assert_eq!(
            registry.get(&compiled.id).unwrap().public_benchmarks[0].score,
            compiled.public_benchmarks[0].score
        );
        // Different weights under a builtin id: refused outright.
        let mut repointed = remeasured;
        repointed.source = signed_remote("x/y", "y").source;
        assert_eq!(
            registry.replace_signed_catalog(
                vec![repointed],
                None,
                &HashSet::new(),
                &HashSet::new()
            ),
            0
        );
        assert_eq!(
            registry.get(&compiled.id).unwrap().public_benchmarks[0].score,
            compiled.public_benchmarks[0].score
        );
    }

    #[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 {
                revoked: Vec::new(),
                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 }
        }

        fn unset(name: &'static str) -> Self {
            let previous = std::env::var_os(name);
            // SAFETY: as for `set`.
            unsafe { std::env::remove_var(name) };
            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");
        }
    }

    // The fixture redirects the cache through env vars and unix paths; keep it
    // 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"));
        let _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");

        // `download_file` writes the resolved hub — `HF_HOME/hub` here, written
        // out literally so the fixture does not borrow the code under test.
        // Populate the exact base-model pointer it asks for, fully offline.
        let cached_auxiliary = tmp
            .path()
            .join("hf-home/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 = tmp
            .path()
            .join("hf-home/hub/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");
        }
        // The auxiliary came from a file the hub already held: nothing was
        // fetched, so nothing is recorded as CAR's.
        assert!(
            !tmp.path()
                .join("model-management")
                .join(crate::retire::PROVENANCE_FILE)
                .exists(),
            "reusing a cached file must not record it as a CAR download"
        );
    }

    // The fixture redirects the cache through env vars and unix paths; keep it
    // 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);
        let _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");

        // The managed repair fetches Flux's auxiliary from the base model.
        // Populate the resolved hub (`HF_HOME/hub`, written out literally) so
        // that repair succeeds offline.
        let cached_auxiliary = hf_home
            .join("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 = hf_home.join("hub/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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        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());
    }

    #[tokio::test]
    async fn a_zero_byte_managed_config_json_starts_acquisition() {
        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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        let models = tmp.path().join("models");
        let name = "Zero-Byte-Config-MLX";
        write_mlx_dir(&models, name, "qwen3");
        let model_dir = models.join(name);
        std::fs::write(model_dir.join("tokenizer.json"), b"{}").unwrap();
        std::fs::write(model_dir.join("config.json"), b"").unwrap();
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(mlx_schema(
            "mlx/zero-byte-config",
            name,
            "example/zero-byte-config",
        ));
        let recorder =
            std::sync::Arc::new(AcquisitionRecorder::replacing_dir_on_started(model_dir));
        let sink = ProgressSink::new(recorder.clone());

        let result = reg
            .ensure_local_with_progress("mlx/zero-byte-config", &sink)
            .await;

        let events = recorder.events();
        assert!(
            matches!(events.first(), Some(DownloadEvent::Started { .. })),
            "weights beside a zero-byte config.json must start an acquisition, not be reused; events: {events:?}"
        );
        assert_eq!(started_count(&events), 1);
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn a_zero_byte_hf_snapshot_config_json_starts_acquisition() {
        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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        let models = tmp.path().join("models");
        let name = "Zero-Byte-Snapshot-Config-MLX";
        let repo = "example/zero-byte-snapshot-config";
        let snapshot = hf_home
            .join("hub")
            .join("models--example--zero-byte-snapshot-config")
            .join("snapshots")
            .join("fixture");
        write_complete_mlx_snapshot(&snapshot);
        std::fs::write(snapshot.join("config.json"), b"").unwrap();
        // An empty managed directory leaves the snapshot as the only reuse
        // candidate.
        let model_dir = models.join(name);
        std::fs::create_dir_all(&model_dir).unwrap();
        let mut reg = UnifiedRegistry::new_empty(models);
        reg.register(mlx_schema("mlx/zero-byte-snapshot-config", name, repo));
        let recorder =
            std::sync::Arc::new(AcquisitionRecorder::replacing_dir_on_started(model_dir));
        let sink = ProgressSink::new(recorder.clone());

        let result = reg
            .ensure_local_with_progress("mlx/zero-byte-snapshot-config", &sink)
            .await;

        let events = recorder.events();
        assert!(
            matches!(events.first(), Some(DownloadEvent::Started { .. })),
            "a snapshot with a zero-byte config.json must start an acquisition, not be reused; events: {events:?}"
        );
        assert_eq!(started_count(&events), 1);
        assert!(result.is_err());
    }

    #[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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        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 _hub = ScopedEnvVar::unset("HF_HUB_CACHE");
        let _legacy_hub = ScopedEnvVar::unset("HUGGINGFACE_HUB_CACHE");
        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 {
                revoked: Vec::new(),
                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 {
                revoked: Vec::new(),
                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 {
                revoked: Vec::new(),
                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 subscription_billing_lookup_uses_exact_or_implicit_latest_ids() {
        let codex_ids: Vec<_> = builtin_catalog()
            .into_iter()
            .filter(|model| model.id.starts_with("openai-codex/"))
            .map(|model| model.id)
            .collect();
        assert!(!codex_ids.is_empty());
        for id in codex_ids {
            assert!(is_builtin_subscription_billed(&id), "{id}");
            if let Some(untagged) = id.strip_suffix(":latest") {
                assert!(is_builtin_subscription_billed(untagged), "{untagged}");
            }
        }
        assert!(!is_builtin_subscription_billed("parslee/reasoning"));
        assert!(!is_builtin_subscription_billed("openai/gpt-5.5-2026-04-23"));
        assert!(!is_builtin_subscription_billed("unknown/not-a-model"));
    }

    #[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.
    /// The `provider/name[:tag]` form `InferenceModelIdentity::resolved_model_id`
    /// documents for built-in ids: each part is `[A-Za-z0-9][A-Za-z0-9._-]*`.
    fn is_builtin_model_id_form(id: &str) -> bool {
        fn part(s: &str) -> bool {
            let mut chars = s.chars();
            chars.next().is_some_and(|c| c.is_ascii_alphanumeric())
                && chars.all(|c| c.is_ascii_alphanumeric() || "._-".contains(c))
        }
        let (body, tag) = match id.split_once(':') {
            Some((body, tag)) => (body, Some(tag)),
            None => (id, None),
        };
        let Some((provider, name)) = body.split_once('/') else {
            return false;
        };
        part(provider) && part(name) && tag.is_none_or(part)
    }

    #[test]
    fn builtin_model_ids_follow_the_resolved_model_id_contract() {
        for good in [
            "anthropic/claude-opus-4-8",
            "openai/gpt-5.5-2026-04-23",
            "qwen/qwen3-4b:q4_k_m",
        ] {
            assert!(is_builtin_model_id_form(good), "{good} should pass");
        }
        for bad in [
            "",
            "gpt-5.5",
            "/name",
            "openai/",
            "-openai/x",
            "openai/-x",
            "open ai/x",
            "openai/x y",
            "openai/é",
            "parslee/openrouter/open-fast",
            "openai/x:",
            "openai/x:-a",
            "openai/x:a b",
            "openai/x:a:b",
            "openai/x:a/b",
        ] {
            assert!(!is_builtin_model_id_form(bad), "{bad:?} should fail");
        }
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
        let off: Vec<&str> = catalog
            .iter()
            .map(|m| m.id.as_str())
            .filter(|id| !is_builtin_model_id_form(id))
            .collect();
        assert!(off.is_empty(), "built-in ids outside the contract: {off:?}");
    }

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

    #[test]
    fn codex_subscription_rows_publish_the_gpt_5_6_family() {
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
        assert_eq!(
            catalog
                .iter()
                .filter(|model| model.provider == "openai-codex")
                .count(),
            6,
            "the explicit checks below must cover every openai-codex row"
        );
        for id in [
            "openai-codex/gpt-5.6-sol:latest",
            "openai-codex/gpt-5.6-sol:high",
            "openai-codex/gpt-5.6-luna:latest",
            "openai-codex/gpt-5.6-luna:high",
            "openai-codex/gpt-5.6-terra:latest",
            "openai-codex/gpt-5.6-terra:high",
        ] {
            let row = catalog
                .iter()
                .find(|model| model.id == id)
                .unwrap_or_else(|| panic!("missing {id}"));
            assert_eq!(row.provider, "openai-codex", "{id}");
            assert!(
                matches!(&row.source, ModelSource::Proprietary { provider, auth: ProprietaryAuth::ChatGptSubscription {}, .. } if provider == "openai-codex"),
                "{id}"
            );
            assert!(row.is_subscription_billed(), "{id}");
            assert!(
                row.cost.input_per_mtok.is_none() && row.cost.output_per_mtok.is_none(),
                "{id} must carry no dollar price"
            );
            assert!(row.tags.iter().any(|tag| tag == "subscription"), "{id}");
            assert!(
                row.cost
                    .estimated_usd_bounded(Some(1_000), 1_000, 1_000, 0, 0)
                    .is_none(),
                "{id} must not produce an approximate dollar cost"
            );
            let mut accidentally_priced = row.clone();
            accidentally_priced.cost.input_per_mtok = Some(1.0);
            accidentally_priced.cost.output_per_mtok = Some(1.0);
            assert!(
                crate::scoreboard_price_model(&accidentally_priced).is_none(),
                "{id} must stay out of the dollar scoreboard"
            );
            assert!(row.has_capability(ModelCapability::ToolUse), "{id}");
            assert_eq!(row.name.ends_with(":high"), id.ends_with(":high"), "{id}");
        }
        let cli = catalog
            .iter()
            .find(|model| model.id == "openai/gpt-5.6-sol:high")
            .unwrap();
        assert!(
            matches!(&cli.source, ModelSource::CodexCli { .. }),
            "the Codex CLI row stays as it was"
        );
        assert!(!cli.is_subscription_billed());
        let api = catalog
            .iter()
            .find(|model| model.id == "openai/gpt-5.6-sol:latest")
            .unwrap();
        assert!(
            matches!(&api.source, ModelSource::RemoteApi { .. }),
            "the API-key row stays as it was"
        );
        assert!(!api.is_subscription_billed());
        let parslee = catalog
            .iter()
            .find(|model| model.id == "parslee/fast")
            .expect("catalog publishes a Parslee row");
        assert!(!parslee.is_subscription_billed());
    }

    #[test]
    fn codex_rows_unavailable_without_sign_in_carry_the_sign_in_reason() {
        let _environment = crate::openrouter::test_environment_scope();
        remember_codex_sign_in(false);

        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_with_state_root(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
        );
        let expected_ids = [
            "openai-codex/gpt-5.6-sol:latest",
            "openai-codex/gpt-5.6-sol:high",
            "openai-codex/gpt-5.6-luna:latest",
            "openai-codex/gpt-5.6-luna:high",
            "openai-codex/gpt-5.6-terra:latest",
            "openai-codex/gpt-5.6-terra:high",
        ];

        for id in expected_ids {
            let info = ModelInfo::from(
                registry
                    .get(id)
                    .unwrap_or_else(|| panic!("missing builtin Codex row {id}")),
            );
            assert!(!info.available, "{id} must be unavailable while signed out");
            let reason = model_unavailable_reason(&info).expect("signed-out Codex reason");
            assert_eq!(reason, crate::schema::OPENAI_CODEX_SIGN_IN_HINT);
            assert_eq!(
                serde_json::to_value(&info).unwrap()["unavailable_reason"],
                reason
            );
            assert!(reason.contains("car auth login --provider openai-codex"));
        }

        remember_codex_sign_in(true);
        registry.refresh_availability();

        for id in expected_ids {
            let info = ModelInfo::from(registry.get(id).unwrap());
            assert!(info.available, "{id} must be available while signed in");
            assert_eq!(model_unavailable_reason(&info), None, "{id}");
            assert!(serde_json::to_value(&info).unwrap()["unavailable_reason"].is_null());
        }

        remember_codex_sign_in(false);
    }

    #[test]
    fn work_context_availability_is_independent_of_other_org_rejection() {
        let _environment = crate::openrouter::test_environment_scope();
        crate::parslee_credential::clear_credential_rejected();
        let tmp = TempDir::new().unwrap();
        let mut parslee = UnifiedRegistry::new_with_state_root(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
        );
        let mut flyexclusive = parslee.clone();
        crate::parslee_credential::note_credential_rejected();
        parslee.refresh_work_context_availability("https://authority.example", true);
        flyexclusive.refresh_work_context_availability("https://authority.example", false);
        let scoped_rows: Vec<_> = parslee.all().filter(|row| matches!(
            &row.source, ModelSource::Proprietary { provider, auth: ProprietaryAuth::OAuth2Pkce { .. }, .. }
                if provider == "parslee"
        )).collect();
        assert!(!scoped_rows.is_empty());
        for row in scoped_rows {
            assert!(row.available);
            assert!(!flyexclusive.get(&row.id).unwrap().available);
        }
        assert!(
            crate::parslee_credential::credential_rejected(),
            "scoped routing must not clear another context's observation"
        );
        crate::parslee_credential::clear_credential_rejected();
    }

    #[test]
    fn codex_availability_leaves_openai_and_parslee_rows_alone() {
        let _environment = crate::openrouter::test_environment_scope();
        remember_codex_sign_in(false);

        let tmp = TempDir::new().unwrap();
        let mut registry = UnifiedRegistry::new_with_state_root(
            tmp.path().to_path_buf(),
            tmp.path().join("models"),
        );
        registry.refresh_availability();

        let unaffected_ids: Vec<String> = registry
            .list()
            .into_iter()
            .filter(|row| {
                matches!(
                    row.id.as_str(),
                    "openai/gpt-5.6-sol:latest" | "openai/gpt-5.6-sol:high"
                ) || row.id.starts_with("parslee/")
            })
            .map(|row| row.id.clone())
            .collect();
        assert!(unaffected_ids
            .iter()
            .any(|id| id == "openai/gpt-5.6-sol:latest"));
        assert!(unaffected_ids
            .iter()
            .any(|id| id == "openai/gpt-5.6-sol:high"));
        assert!(
            unaffected_ids.iter().any(|id| id.starts_with("parslee/")),
            "builtin catalog must contain Parslee rows"
        );
        let signed_out: std::collections::HashMap<String, bool> = unaffected_ids
            .iter()
            .map(|id| (id.clone(), registry.get(id).unwrap().available))
            .collect();

        remember_codex_sign_in(true);
        registry.refresh_availability();

        for id in unaffected_ids {
            assert_eq!(
                registry.get(&id).unwrap().available,
                signed_out[&id],
                "Codex sign-in must not change {id} availability"
            );
        }

        remember_codex_sign_in(false);
    }

    #[test]
    fn builtin_catalog_names_are_unique_for_codex_rows() {
        let catalog: Vec<ModelSchema> = serde_json::from_str(BUILTIN_CATALOG_JSON).unwrap();
        for row in catalog.iter().filter(|row| row.provider == "openai-codex") {
            let occurrences = catalog
                .iter()
                .filter(|candidate| candidate.name == row.name)
                .count();
            assert_eq!(
                occurrences, 1,
                "openai-codex name must be unique: {}",
                row.name
            );
        }
    }

    /// 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()),
                runs: None,
                spread: None,
            },
            BenchmarkScore {
                name: "HumanEval".into(),
                score: 0.713,
                harness: Some("pass@1".into()),
                source_url: None,
                measured_at: None,
                runs: None,
                spread: 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",
                "7eeaecc6e83ce409b37cc68e9c44ba84d64c5f9690179a494bdcb25a93f0aac3",
            ),
            (
                "anthropic/claude-opus-4-8",
                "claude-opus-4-8",
                "4.8",
                "10504959e51dc76c3563df91ae2eaba57cf814834ecd657232c50f264f9e735e",
            ),
            (
                "openai/gpt-5.6-sol:high",
                "gpt-5.6-sol:high",
                "latest",
                "0f580e5c2cebd169be21d511ee4c8c7b2e97e83dc62d8462366a3520ae74de5b",
            ),
        ] {
            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 mlx_executor_discrimination_tests {
    use super::*;

    /// `ModelSource::Mlx` is a weight-format tag shared by five executors, and
    /// only one of them is the in-process Swift loader. Gating the whole source
    /// on that loader marked `mflux`, `ltx-2-mlx` and the `mlx_vlm` CLI rows
    /// unavailable on any Mac whose Swift build merely failed — subprocesses
    /// that never touch the linked library, and the only local image/video
    /// backends there are.
    ///
    /// Asserted against the real builtin catalog rather than a hand-built
    /// fixture: a fixture written beside this helper would agree with it by
    /// construction, and the thing under test is precisely whether the
    /// predicate matches the catalog's actual shape.
    #[test]
    fn only_in_process_text_rows_need_the_swift_loader() {
        let catalog = builtin_catalog();
        let row = |id: &str| {
            catalog
                .iter()
                .find(|m| m.id == id)
                .unwrap_or_else(|| panic!("{id} missing from the builtin catalog"))
        };

        for id in [
            "mlx/flux-1-lite-8b:q4",    // mflux subprocess
            "mlx/ltx-2.3:q4",           // ltx-2-mlx subprocess
            "mlx-vlm/qwen3-vl-2b:bf16", // mlx_vlm Python CLI
            "mlx/kokoro-82m:bf16",      // managed mlx-audio venv
            "mlx/parakeet-tdt-0.6b-v3:default",
        ] {
            let m = row(id);
            assert!(
                matches!(m.source, ModelSource::Mlx { .. }),
                "{id}: precondition — this test is only meaningful for Mlx-sourced rows"
            );
            assert!(
                !mlx_row_needs_in_process_loader(m),
                "{id} runs in a subprocess; its availability must not depend on the \
                 in-process Swift loader"
            );
        }

        // Positive control: the predicate must still say yes to something, or
        // the assertions above would pass against a function returning false.
        for id in ["mlx/qwen3-4b:4bit", "mlx/gemma-4-12b-it:4bit"] {
            assert!(
                mlx_row_needs_in_process_loader(row(id)),
                "{id} is served in-process and does need the loader"
            );
        }
    }
}

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

    #[test]
    fn builtin_credential_names_cover_remote_and_proprietary_auth() {
        let names = builtin_credential_env_names();
        for name in [
            "OPENAI_API_KEY",
            "ANTHROPIC_API_KEY",
            crate::openrouter::API_KEY_ENV,
        ] {
            assert!(names.contains(name), "built-in catalog omitted {name}");
        }
    }
}