car-inference 0.60.0

//! Locate + provision the external `vllm-mlx` OpenAI-compatible MLX server that
//! CAR supervises to serve `vllm-mlx/*` models.
//!
//! The in-process MLX backend is text-only, so multimodal and unsupported-arch
//! models (`vllm-mlx/gemma-4-*`, `vllm-mlx/qwen3.6-*`) run through a
//! `vllm-mlx serve <model> --port <P>` process that exposes an OpenAI API. CAR
//! supervises that process (start / health-wait / idle-stop) so those models
//! "just work" — and, when the runtime is absent, provisions it so the user
//! doesn't have to. `vllm-mlx` is a PyPI package; we install it into a
//! dedicated `uv` venv under `~/.car/visual-runtime`, mirroring the managed
//! speech runtime.
//!
//! Resolution prefers an *existing* install (the user may already have it on
//! PATH or via `uv tool install vllm-mlx`); provisioning only happens when none
//! is found, and lands in the CAR-managed venv.

use std::path::{Path, PathBuf};
use std::time::Duration;

use tokio::process::Command;

/// Binary that serves an OpenAI-compatible MLX endpoint (`vllm-mlx serve …`).
const SERVER_BIN: &str = "vllm-mlx";

/// Minimum `vllm-mlx` version required. A binary older than this is skipped
/// during resolution so we provision a working one instead, and because
/// `PIP_SPEC` is the same floor, provisioning over an older install upgrades it.
///
/// - `< 0.3.0` (mlx-vlm `< ~0.6`) hits the upstream `RuntimeError: There is no
///   Stream(gpu, 0)` on every server-side generation.
/// - `< 0.6.0` loads `--mllm-draft-model` only as a Gemma 4 assistant drafter,
///   so a Qwen MTP drafter (`qwen3_5_mtp`) fails at load with "Received 31
///   parameters not in model". 0.6.0 dispatches drafter loading by
///   architecture. On an M5 Pro this took Qwen3.8-27B from 16 to 21.5 tok/s
///   with the venv's mlx-vlm 0.6.15 (26 with mlx-vlm 0.7.10, which would also
///   move mlx and mlx-audio under mflux and ltx, so it is not required here).
///   Upgrading from 0.4.1 changes only vllm-mlx and llguidance in the shared
///   runtime venv.
const MIN_VERSION: (u64, u64, u64) = (0, 6, 0);

/// The PyPI distribution that provides [`SERVER_BIN`]. Pinned to the floor so a
/// provision can never land a known-broken build.
const PIP_SPEC: &str = "vllm-mlx>=0.6.0";

/// Errors from locating or provisioning the vLLM-MLX runtime.
#[derive(Debug, thiserror::Error)]
pub enum RuntimeError {
    /// `remedy` names only installs that [`lookup`] finds (see
    /// [`INSTALL_COMMAND`] and [`upgrade_command`]).
    #[error(
        "`uv` is required to provision the vllm-mlx runtime but was not found on PATH; {remedy}"
    )]
    UvMissing { remedy: String },
    #[error("provisioning step `{step}` failed: {detail}")]
    Provision { step: &'static str, detail: String },
    #[error("vllm-mlx still not found after provisioning into {0}")]
    NotFoundAfterInstall(PathBuf),
    /// A request that may not download reached provisioning: it installs
    /// nothing, and the text says what the person runs instead.
    #[error("{0}")]
    NeverInstall(String),
}

/// The install whose result [`lookup`] finds: `uv tool install` links the
/// binary into uv's executable dir and keeps it under uv's tools dir, and both
/// are searched, as uv resolves them ([`uv_tool_bin_dirs`], [`uv_tool_roots`]).
/// A user `pip install` lands in a per-interpreter `bin` that is not, unless it
/// happens to be on `PATH`.
const INSTALL_COMMAND: &str = "uv tool install vllm-mlx";

/// Upgrades a `uv tool` install in place. Not offered for a binary something
/// else installed: uv will not replace an executable it did not install.
const UPGRADE_COMMAND: &str = "uv tool install --upgrade vllm-mlx";

const UV_DOCS: &str = "https://docs.astral.sh/uv/";

/// What looking for an installed `vllm-mlx` found.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Lookup {
    /// A binary at or above [`MIN_VERSION`] (or the explicit override).
    Ready(PathBuf),
    /// Only binaries older than [`MIN_VERSION`]; the newest of them and where
    /// it was found.
    Outdated((u64, u64, u64), PathBuf),
    /// No binary whose version could be read.
    Missing,
}

impl Lookup {
    /// What is wrong and the command that fixes it, for a request that never
    /// installs anything. `None` when the runtime is ready.
    fn absence(&self) -> Option<String> {
        match self {
            Lookup::Ready(_) => None,
            Lookup::Missing => Some(format!(
                "the vllm-mlx runtime is not installed, and this request never installs one. \
                 Install it with `{INSTALL_COMMAND}`"
            )),
            Lookup::Outdated(version, binary) => Some(format!(
                "{}, and this request never upgrades it. {}",
                outdated_at(*version, binary),
                match upgrade_command(binary) {
                    Some(command) => format!("Upgrade it with `{command}`"),
                    None => format!("Upgrade it with the tool that installed it, {OR_OVERRIDE}"),
                }
            )),
        }
    }
}

/// The way round any install CAR cannot upgrade or find.
const OR_OVERRIDE: &str = "or set `CAR_VLLM_MLX_BIN` to the path of a newer vllm-mlx binary";

fn outdated_at((major, minor, patch): (u64, u64, u64), binary: &Path) -> String {
    let (min_major, min_minor, min_patch) = MIN_VERSION;
    format!(
        "the installed vllm-mlx runtime at `{}` is {major}.{minor}.{patch}, older than the \
         {min_major}.{min_minor}.{min_patch} CAR needs",
        binary.display()
    )
}

/// The command that upgrades the `vllm-mlx` at `binary`, when CAR can tell
/// what installed it: uv's tool manager, or CAR's own managed venv. `None` for
/// anything else (pip, pipx, Homebrew).
fn upgrade_command(binary: &Path) -> Option<String> {
    // As found and with links resolved: `~/.local/bin/vllm-mlx` is a link into
    // uv's tools dir, and a path that no longer exists cannot be resolved.
    let real = |p: &Path| car_winenv::canonicalize(p).unwrap_or_else(|_| p.to_path_buf());
    let resolved = real(binary);
    let under = |root: &Path| binary.starts_with(root) || resolved.starts_with(real(root));
    // uv writes `uv-receipt.toml` into each tool environment it creates; an
    // environment without one (pipx pointed at the same folder) is not uv's,
    // and uv refuses to replace its programs.
    let installed_by_uv = |root: &Path| {
        [
            (binary, root.to_path_buf()),
            (resolved.as_path(), real(root)),
        ]
        .into_iter()
        .any(|(path, root)| {
            path.strip_prefix(&root)
                .ok()
                .and_then(|rest| rest.components().next())
                .is_some_and(|tool| root.join(tool).join("uv-receipt.toml").is_file())
        })
    };
    if uv_tool_roots().iter().any(|root| installed_by_uv(root)) {
        return Some(UPGRADE_COMMAND.into());
    }
    let root = managed_root()?;
    under(&root).then(|| {
        format!(
            "uv pip install --python \"{}\" \"{PIP_SPEC}\"",
            crate::managed_venv::interpreter(&root).display()
        )
    })
}

/// How long routing uses a lookup before looking again, so a runtime the
/// person installs, upgrades or breaks is seen within this long. Server start
/// ([`ensure_runtime`]) never uses it: it always looks again.
pub(crate) const RECHECK_AFTER: Duration = Duration::from_secs(30);

/// Routing's last lookup in this process and when it ran.
static REMEMBERED: std::sync::Mutex<Option<(Lookup, std::time::Instant)>> =
    std::sync::Mutex::new(None);

/// What a remembered lookup still says, or `None` when it is stale: older
/// than [`RECHECK_AFTER`], or a found runtime whose binary or shebang
/// interpreter is gone (a Homebrew upgrade that rotates the Python a venv
/// links to leaves the venv's files in place). Running the interpreter, as
/// `managed_venv::interpreter_healthy` does, would catch more, but this is
/// checked on the async path, so it runs no process; the 30-second recheck
/// covers the rest.
fn still_stands(remembered: Option<&(Lookup, std::time::Instant)>) -> Option<Lookup> {
    let (lookup, at) = remembered?;
    if at.elapsed() >= RECHECK_AFTER {
        return None;
    }
    match lookup {
        Lookup::Ready(server) if !server.is_file() || interpreter_gone(server) => None,
        known => Some(known.clone()),
    }
}

/// Whether `binary` is a script whose `#!` interpreter no longer exists.
fn interpreter_gone(binary: &Path) -> bool {
    use std::io::BufRead;
    let Ok(file) = std::fs::File::open(binary) else {
        return true;
    };
    let mut first = Vec::new();
    if std::io::BufReader::new(file)
        .read_until(b'\n', &mut first)
        .is_err()
    {
        return true;
    }
    let Some(shebang) = first.strip_prefix(b"#!") else {
        return false;
    };
    String::from_utf8_lossy(shebang)
        .split_whitespace()
        .next()
        .is_some_and(|interpreter| !Path::new(interpreter).exists())
}

/// Whether a usable `vllm-mlx` is installed, without installing one, for
/// routing. A remembered answer is used while it stands; otherwise the lookup
/// (which runs each candidate's Python to read its version) runs on a
/// blocking thread, and its answer is remembered.
pub(crate) async fn status() -> Lookup {
    #[cfg(test)]
    if let Some(scope) = test_runtime::current() {
        return scope.status();
    }
    if let Some(known) = still_stands(
        REMEMBERED
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .as_ref(),
    ) {
        return known;
    }
    let found = tokio::task::spawn_blocking(lookup)
        .await
        .unwrap_or(Lookup::Missing);
    *REMEMBERED.lock().unwrap_or_else(|p| p.into_inner()) =
        Some((found.clone(), std::time::Instant::now()));
    found
}

/// The refusal for a request that may not download (or install) whose route
/// is `schema`, a CAR-managed vLLM-MLX model, when the runtime it runs on is
/// missing or too old: it names the model and the command that fixes it.
pub(crate) fn not_ready_error(
    schema: &crate::schema::ModelSchema,
    lookup: &Lookup,
) -> crate::InferenceError {
    let why = lookup
        .absence()
        .unwrap_or_else(|| "the vllm-mlx runtime is not ready".into());
    crate::InferenceError::NoEligibleModel {
        excluded_models: format!(
            "{name} runs on the vllm-mlx runtime: {why}, or connect a provider's API key so a \
             model that is ready can be used",
            name = schema.name
        ),
    }
}

/// Test stand-ins for the runtime lookup, with their own remembered lookup
/// and counters of what ran. A stand-in applies only to the futures a test
/// runs through [`State::run`](test_runtime::State::run): it is task-local,
/// so no other test, and no code outside that future, can see it or move its
/// counters. [`resolve_existing`] never consults it.
#[cfg(test)]
pub(crate) mod test_runtime {
    use super::Lookup;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};

    tokio::task_local! {
        static SCOPE: Arc<State>;
    }

    pub(super) fn current() -> Option<Arc<State>> {
        SCOPE.try_with(Arc::clone).ok()
    }

    pub(crate) struct State {
        found: Mutex<Lookup>,
        remembered: Mutex<Option<(Lookup, std::time::Instant)>>,
        lookups: AtomicUsize,
        provisions: AtomicUsize,
    }

    impl State {
        /// Every lookup finds `found`, and provisioning stops before it runs
        /// anything.
        pub(crate) fn new(found: Lookup) -> Arc<Self> {
            Arc::new(Self {
                found: Mutex::new(found),
                remembered: Mutex::new(None),
                lookups: AtomicUsize::new(0),
                provisions: AtomicUsize::new(0),
            })
        }

        /// Run `future` with this stand-in.
        pub(crate) async fn run<F: std::future::Future>(self: &Arc<Self>, future: F) -> F::Output {
            SCOPE.scope(self.clone(), future).await
        }

        /// From now on every lookup finds `found`; what was remembered stays.
        pub(crate) fn set(&self, found: Lookup) {
            *self.found.lock().unwrap() = found;
        }

        pub(crate) fn lookups(&self) -> usize {
            self.lookups.load(Ordering::SeqCst)
        }

        pub(crate) fn provisions(&self) -> usize {
            self.provisions.load(Ordering::SeqCst)
        }

        /// Make the remembered lookup [`RECHECK_AFTER`](super::RECHECK_AFTER)
        /// older, as if that long had passed.
        pub(crate) fn age_remembered(&self) {
            if let Some((_, at)) = self.remembered.lock().unwrap().as_mut() {
                *at -= super::RECHECK_AFTER;
            }
        }

        pub(super) fn look(&self) -> Lookup {
            self.lookups.fetch_add(1, Ordering::SeqCst);
            self.found.lock().unwrap().clone()
        }

        /// [`super::status`] against this stand-in's own remembered lookup.
        pub(super) fn status(&self) -> Lookup {
            let mut remembered = self.remembered.lock().unwrap();
            if let Some(known) = super::still_stands(remembered.as_ref()) {
                return known;
            }
            let found = self.look();
            *remembered = Some((found.clone(), std::time::Instant::now()));
            found
        }

        pub(super) fn count_provision(&self) {
            self.provisions.fetch_add(1, Ordering::SeqCst);
        }
    }
}

/// A located (and possibly CAR-provisioned) vLLM-MLX runtime.
#[derive(Debug, Clone)]
pub struct VllmRuntime {
    /// Absolute path to the `vllm-mlx` server binary.
    pub server: PathBuf,
}

/// Root of the CAR-managed visual runtime venv (`~/.car/visual-runtime`).
/// Shared with the other Python visual/audio runtimes by convention.
fn managed_root() -> Option<PathBuf> {
    std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".car").join("visual-runtime"))
}

/// The `vllm-mlx` binary inside the managed venv, if HOME is known.
///
/// Resolved through [`crate::managed_venv::venv_program`] rather than a local
/// `bin/` join: a virtualenv puts its executables in `Scripts\` on Windows, and
/// a duplicated copy of that assumption is exactly what made CAR rebuild a
/// healthy speech runtime on every call there (car#956).
fn managed_server_bin() -> Option<PathBuf> {
    managed_root().map(|r| crate::managed_venv::venv_program(&r, SERVER_BIN))
}

/// The absolute directory named by `name`; an empty or relative value is skipped.
fn env_dir(name: &str) -> Option<PathBuf> {
    std::env::var_os(name)
        .map(PathBuf::from)
        .filter(|p| p.is_absolute())
}

/// Where `uv tool install` puts executables, in uv's order: `UV_TOOL_BIN_DIR`,
/// `XDG_BIN_HOME`, `XDG_DATA_HOME/../bin`, `~/.local/bin`.
fn uv_tool_bin_dirs() -> Vec<PathBuf> {
    let mut dirs: Vec<PathBuf> = [
        env_dir("UV_TOOL_BIN_DIR"),
        env_dir("XDG_BIN_HOME"),
        env_dir("XDG_DATA_HOME").map(|d| d.join("..").join("bin")),
    ]
    .into_iter()
    .flatten()
    .collect();
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        dirs.push(home.join(".local").join("bin"));
    }
    dirs
}

/// Where `uv tool install` keeps tool environments, in uv's order:
/// `UV_TOOL_DIR`, `XDG_DATA_HOME/uv/tools`, `~/.local/share/uv/tools`.
fn uv_tool_roots() -> Vec<PathBuf> {
    let mut roots: Vec<PathBuf> = [
        env_dir("UV_TOOL_DIR"),
        env_dir("XDG_DATA_HOME").map(|d| d.join("uv").join("tools")),
    ]
    .into_iter()
    .flatten()
    .collect();
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        roots.push(home.join(".local").join("share").join("uv").join("tools"));
    }
    roots
}

/// Directories searched for an existing `vllm-mlx`, in priority order: PATH,
/// uv's executable dirs, the `vllm-mlx` uv tool environment, and the
/// CAR-managed venv. An explicit override is checked before any of them.
/// Mirrors `backend::mlx_vlm_cli`'s search convention.
fn search_dirs() -> Vec<PathBuf> {
    let mut dirs: Vec<PathBuf> = std::env::var_os("PATH")
        .map(|paths| std::env::split_paths(&paths).collect())
        .unwrap_or_default();
    dirs.extend(uv_tool_bin_dirs());
    dirs.extend(
        uv_tool_roots()
            .into_iter()
            .map(|root| crate::managed_venv::venv_bin_dir(&root.join(SERVER_BIN))),
    );
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        // Same reason `managed_server_bin` goes through `venv_program`: a venv puts
        // its executables in `Scripts\` on Windows, and a hand-rolled `bin` join is
        // the duplicated assumption car#956 was about.
        dirs.push(crate::managed_venv::venv_bin_dir(
            &home.join(".car").join("visual-runtime"),
        ));
    }
    dedupe(dirs)
}

fn dedupe(paths: Vec<PathBuf>) -> Vec<PathBuf> {
    let mut out = Vec::new();
    for p in paths {
        if !out.contains(&p) {
            out.push(p);
        }
    }
    out
}

/// Locate an existing `vllm-mlx` binary without provisioning. Honors the
/// `CAR_VLLM_MLX_BIN` override (used as-is). Auto-discovered binaries must meet
/// [`MIN_VERSION`]; among those that do, the **highest version wins**. Returns
/// an absolute path.
///
/// Best-available rather than first-above-the-floor, because those differ in
/// practice and the difference is silent. A machine that once ran `uv tool
/// install vllm-mlx` has one on `PATH` via `~/.local/bin`, which is searched
/// before the CAR-managed venv — so a stale tool install shadowed the newer
/// runtime CAR provisions for itself, forever, with no diagnostic. It was
/// exactly at `MIN_VERSION`, so the floor check waved it through, and 0.3.0
/// hard-fails on text-only checkpoints of multimodal architectures where 0.4.1
/// disables the absent vision tower and serves the model. Picking the first
/// acceptable binary meant CAR could not load models it was fully equipped to
/// run.
pub fn resolve_existing() -> Option<PathBuf> {
    match lookup() {
        Lookup::Ready(server) => Some(server),
        Lookup::Outdated(..) | Lookup::Missing => None,
    }
}

/// [`resolve_existing`], keeping why nothing usable was found. Blocking: it
/// runs each candidate's interpreter.
fn lookup() -> Lookup {
    // An explicit override is the user's responsibility — honored without a
    // version gate.
    if let Some(p) = std::env::var_os("CAR_VLLM_MLX_BIN").map(PathBuf::from) {
        if p.is_file() {
            return absolutize(&p).map_or(Lookup::Missing, Lookup::Ready);
        }
    }
    let mut candidates: Vec<(PathBuf, Option<(u64, u64, u64)>)> = Vec::new();
    let mut seen: Vec<PathBuf> = Vec::new();
    for dir in search_dirs() {
        let candidate = dir.join(SERVER_BIN);
        if !candidate.is_file() {
            continue;
        }
        // `~/.local/bin/vllm-mlx` is typically a symlink into the uv tool dir,
        // which is also searched. Canonicalize so one install is not probed
        // (and version-checked) twice.
        let canon = car_winenv::canonicalize(&candidate).unwrap_or_else(|_| candidate.clone());
        if seen.contains(&canon) {
            continue;
        }
        seen.push(canon);
        let version = binary_version(&candidate);
        candidates.push((candidate, version));
    }
    let newest = candidates
        .iter()
        .filter_map(|(path, version)| version.map(|v| (v, path.clone())))
        .max_by(|(a, _), (b, _)| a.cmp(b));
    match best_candidate(candidates).and_then(|path| absolutize(&path)) {
        Some(server) => Lookup::Ready(server),
        None => newest.map_or(Lookup::Missing, |(version, path)| {
            Lookup::Outdated(version, path)
        }),
    }
}

/// Highest-versioned candidate at or above [`MIN_VERSION`].
///
/// Split out from [`resolve_existing`] so the selection rule is testable
/// without a PATH full of real interpreters. An undeterminable version is
/// treated as unusable rather than as a floor pass — provisioning a known-good
/// build is cheaper than debugging a mystery binary.
fn best_candidate(candidates: Vec<(PathBuf, Option<(u64, u64, u64)>)>) -> Option<PathBuf> {
    candidates
        .into_iter()
        .filter_map(|(path, version)| version.map(|v| (v, path)))
        .filter(|(version, _)| *version >= MIN_VERSION)
        .max_by(|(a, _), (b, _)| a.cmp(b))
        .map(|(_, path)| path)
}

/// Ask `binary`'s shebang interpreter for the installed `vllm-mlx` version.
fn binary_version(binary: &Path) -> Option<(u64, u64, u64)> {
    let head = std::fs::read_to_string(binary).ok()?;
    let interp = head.lines().next()?.strip_prefix("#!")?.trim();
    let out = std::process::Command::new(interp)
        .args([
            "-c",
            "import importlib.metadata as m; print(m.version('vllm-mlx'))",
        ])
        .output()
        .ok()?;
    if !out.status.success() {
        return None;
    }
    parse_semver(String::from_utf8_lossy(&out.stdout).trim())
}

fn parse_semver(s: &str) -> Option<(u64, u64, u64)> {
    let lead = |part: &str| -> u64 {
        part.chars()
            .take_while(|c| c.is_ascii_digit())
            .collect::<String>()
            .parse()
            .unwrap_or(0)
    };
    let mut it = s.split('.');
    let major = it.next()?.trim().parse().ok()?;
    Some((
        major,
        lead(it.next().unwrap_or("0")),
        lead(it.next().unwrap_or("0")),
    ))
}

fn absolutize(p: &Path) -> Option<PathBuf> {
    car_winenv::canonicalize(p)
        .ok()
        .or_else(|| Some(p.to_path_buf()))
}

/// Ensure a usable `vllm-mlx` runtime exists, provisioning into the CAR-managed
/// venv via `uv` when none is found. Idempotent: a second call with the runtime
/// present returns immediately. Network + disk on the install path only.
///
/// Always looks again, never at what routing remembered: a venv whose
/// interpreter has died still has its files, and only this lookup (which runs
/// it) sends that to provisioning, which repairs it.
pub async fn ensure_runtime() -> Result<VllmRuntime, RuntimeError> {
    #[cfg(test)]
    let found = match test_runtime::current() {
        Some(scope) => scope.look(),
        None => lookup(),
    };
    #[cfg(not(test))]
    let found = lookup();
    if let Lookup::Ready(server) = found {
        return Ok(VllmRuntime { server });
    }
    provision(&found).await?;
    // Post-provision, trust the managed venv binary directly (it was pinned to a
    // known-good version), falling back to a fresh resolve.
    let server = managed_server_bin()
        .filter(|p| p.is_file())
        .or_else(resolve_existing)
        .ok_or_else(|| RuntimeError::NotFoundAfterInstall(managed_root().unwrap_or_default()))?;
    Ok(VllmRuntime { server })
}

/// Provision `vllm-mlx` into `~/.car/visual-runtime` via [`crate::managed_venv`]
/// + `uv pip install`. Mirrors `bootstrap_speech_runtime`.
/// `found` is what the lookup before it found.
async fn provision(found: &Lookup) -> Result<(), RuntimeError> {
    #[cfg(test)]
    let scope = test_runtime::current();
    #[cfg(test)]
    if let Some(scope) = &scope {
        scope.count_provision();
    }
    // A request that may not download installs nothing either: the runtime is
    // a Python environment pulled from PyPI (Parslee-ai/car#3718). Routing
    // already skips a model whose runtime is missing; this is the backstop.
    if crate::registry::never_download_in_scope() {
        return Err(RuntimeError::NeverInstall(found.absence().unwrap_or_else(
            || "this request never installs the vllm-mlx runtime".into(),
        )));
    }
    #[cfg(test)]
    if scope.is_some() {
        return Err(RuntimeError::Provision {
            step: "test",
            detail: "this test installs no runtime".into(),
        });
    }
    // Nor does a test outside a stand-in, such as one whose future lost it.
    if cfg!(test) {
        return Err(RuntimeError::Provision {
            step: "test",
            detail: "tests never install the runtime".into(),
        });
    }
    if which("uv").is_none() {
        return Err(uv_missing(found));
    }
    let root = managed_root().ok_or_else(|| RuntimeError::Provision {
        step: "resolve-home",
        detail: "HOME is not set".into(),
    })?;
    // Reuse an existing healthy venv (a prior visual/audio runtime almost always
    // created one) and rebuild it when its interpreter has been rotated away.
    // Delegated to `managed_venv` because a bare `uv venv` hard-fails on an
    // existing directory, which left this whole path unreachable on any machine
    // that had ever provisioned one.
    let outcome = crate::managed_venv::ensure_venv(&root, "python3")
        .await
        .map_err(|e| RuntimeError::Provision {
            step: "venv",
            detail: e.to_string(),
        })?;
    if outcome == crate::managed_venv::VenvOutcome::Recreated {
        tracing::warn!(
            root = %root.display(),
            "managed Python runtime had an unusable interpreter; rebuilt it \
             (installed packages were discarded and are being reinstalled)"
        );
    }

    let venv_python = crate::managed_venv::interpreter(&root);
    run_uv(
        "pip-install",
        &[
            "pip".into(),
            "install".into(),
            "--python".into(),
            venv_python.display().to_string(),
            PIP_SPEC.into(),
        ],
    )
    .await?;

    Ok(())
}

/// The refusal when `uv` is absent, naming what fixes what the lookup found:
/// an install when there is none, the upgrade that fits an outdated one.
fn uv_missing(found: &Lookup) -> RuntimeError {
    let remedy = match found {
        Lookup::Outdated(version, binary) => {
            let at = outdated_at(*version, binary);
            match upgrade_command(binary) {
                Some(command) => {
                    format!("{at}; install uv ({UV_DOCS}) and run `{command}`, {OR_OVERRIDE}")
                }
                None => format!("{at}; upgrade it with the tool that installed it, {OR_OVERRIDE}"),
            }
        }
        Lookup::Ready(_) | Lookup::Missing => format!(
            "install uv ({UV_DOCS}) and run `{INSTALL_COMMAND}`, or set `CAR_VLLM_MLX_BIN` to \
             the path of a vllm-mlx binary"
        ),
    };
    RuntimeError::UvMissing { remedy }
}

async fn run_uv(step: &'static str, args: &[String]) -> Result<(), RuntimeError> {
    let output = Command::new("uv")
        .args(args)
        // Installing a heavy MLX wheel set can take a while on a cold cache.
        .kill_on_drop(true)
        .output()
        .await
        .map_err(|e| RuntimeError::Provision {
            step,
            detail: e.to_string(),
        })?;
    if output.status.success() {
        Ok(())
    } else {
        Err(RuntimeError::Provision {
            step,
            detail: format!(
                "uv exited with {}: {}",
                output.status,
                String::from_utf8_lossy(&output.stderr).trim()
            ),
        })
    }
}

/// The `hf download` command for one attempt. `HF_TOKEN` is handed over
/// explicitly when it lives only in the secret store, because the daemon does
/// not carry stored keys in its own environment for the child to inherit.
fn hf_download_command(hf: &Path, repo: &str, disable_xet: bool) -> Command {
    let mut cmd = Command::new(hf);
    crate::hf_cache::apply_with_token(&mut cmd);
    cmd.arg("download").arg(repo).kill_on_drop(true);
    if disable_xet {
        cmd.env("HF_HUB_DISABLE_XET", "1");
    }
    cmd
}

/// Fetch a HuggingFace repo into the shared HF cache, in the layout the
/// external runtimes expect.
///
/// Delegated to the managed venv's `hf` CLI rather than reimplemented, because
/// the cache layout (`blobs/` + `snapshots/` + `refs/`, content-addressed and
/// Xet-aware) is `huggingface_hub`'s contract, not a format worth cloning. CAR's
/// own `pull` writes its native `~/.car/models/<name>` layout, which is the
/// wrong shape for a model that a Python runtime will open by repo id.
///
/// A stalled transfer is retried once with Xet disabled, for the reason
/// documented on [`crate::vllm_pool`]: a wedged Xet download produces zero bytes
/// indefinitely while plain HTTPS for the same file works.
pub async fn download_repo(repo: &str) -> Result<(), RuntimeError> {
    // Provisioning also guarantees the venv (and therefore `hf`) exists.
    ensure_runtime().await?;
    let hf = managed_root()
        .map(|r| crate::managed_venv::venv_program(&r, "hf"))
        .filter(|p| p.is_file())
        .ok_or_else(|| RuntimeError::Provision {
            step: "resolve-hf-cli",
            detail: "the managed runtime has no `hf` CLI".into(),
        })?;

    for disable_xet in [false, true] {
        let output = hf_download_command(&hf, repo, disable_xet)
            .output()
            .await
            .map_err(|e| RuntimeError::Provision {
                step: "hf-download",
                detail: e.to_string(),
            })?;
        if output.status.success() {
            return Ok(());
        }
        if disable_xet {
            return Err(RuntimeError::Provision {
                step: "hf-download",
                detail: format!(
                    "hf download {repo} failed: {}",
                    String::from_utf8_lossy(&output.stderr).trim()
                ),
            });
        }
        tracing::warn!(repo, "hf download failed; retrying with Xet disabled");
    }
    unreachable!("the loop returns on both iterations")
}

/// Whether CAR can serve a `vllm-mlx/*` model on this machine — either the
/// runtime is already resolvable, or `uv` is present so [`ensure_runtime`] can
/// provision it on first use.
///
/// This is the loose "functionally available" question, matching the
/// convention local MLX models use (a declared `hf_repo` counts as available
/// even though first use downloads weights). The strict "usable this instant"
/// question is `weights_ready`.
pub fn serviceable() -> bool {
    resolve_existing().is_some() || which("uv").is_some()
}

/// Whether an installed `vllm-mlx` at or above the floor CAR needs is present
/// now, without installing or upgrading anything. The strict question a caller
/// that never provisions (an unattended sweep) must ask: `serviceable` counts
/// a runtime `uv` could still install, and a model whose weights are on disk
/// is not servable while its runtime is outdated.
pub fn runtime_ready() -> bool {
    matches!(lookup(), Lookup::Ready(_))
}

/// Why the installed `vllm-mlx` cannot serve now, with the command that fixes
/// it, or `None` when it can.
pub fn runtime_unready_reason() -> Option<String> {
    lookup().absence()
}

/// Minimal `which`: first hit for `name` across PATH. Used to gate the
/// provisioning preflight (`uv` present?) and by tests.
pub(crate) fn which(name: &str) -> Option<PathBuf> {
    std::env::var_os("PATH")?
        .to_str()?
        .split(':')
        .map(|d| Path::new(d).join(name))
        .find(|p| p.is_file())
}

/// Resolve the executable behind a Python launcher for generated test-script
/// shebangs. pyenv's `python3` is itself a shebang script, and kernels do not
/// recursively interpret shebangs; asking Python for `sys.executable` yields
/// the real binary that can safely be placed after `#!`.
#[cfg(all(test, unix))]
pub(crate) fn test_python_interpreter() -> Option<PathBuf> {
    let launcher = which("python3")?;
    let output = std::process::Command::new(launcher)
        .args(["-c", "import sys; print(sys.executable)"])
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    let resolved = PathBuf::from(String::from_utf8(output.stdout).ok()?.trim());
    let healthy = std::process::Command::new(&resolved)
        .args(["-c", ""])
        .status()
        .is_ok_and(|status| status.success());
    healthy.then_some(resolved)
}

/// Probe a `vllm-mlx` server's `/health` once. `true` on any 2xx. Short timeout —
/// callers poll this in a readiness loop after spawning the process.
pub async fn health_ok(endpoint: &str, timeout: Duration) -> bool {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    // Managed endpoints are allocated as plain HTTP on loopback. A direct
    // socket probe avoids proxy/TLS client initialization, which can block a
    // current-thread executor long enough to hide a just-spawned process from
    // concurrent admission waiters.
    let Ok(url) = reqwest::Url::parse(endpoint) else {
        return false;
    };
    if url.scheme() != "http" {
        return false;
    }
    let Some(host) = url.host_str() else {
        return false;
    };
    let Some(port) = url.port_or_known_default() else {
        return false;
    };
    let probe = async {
        let mut stream = tokio::net::TcpStream::connect((host, port)).await.ok()?;
        let request =
            format!("GET /health HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
        stream.write_all(request.as_bytes()).await.ok()?;
        let mut response = Vec::with_capacity(128);
        while !response.windows(2).any(|window| window == b"\r\n") {
            if response.len() >= 1024 {
                return None;
            }
            let mut chunk = [0u8; 128];
            let count = stream.read(&mut chunk).await.ok()?;
            if count == 0 {
                return None;
            }
            response.extend_from_slice(&chunk[..count]);
        }
        let status = std::str::from_utf8(&response)
            .ok()?
            .lines()
            .next()?
            .split_whitespace()
            .nth(1)?
            .parse::<u16>()
            .ok()?;
        Some((200..300).contains(&status))
    };
    matches!(tokio::time::timeout(timeout, probe).await, Ok(Some(true)))
}

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

    #[test]
    fn managed_paths_are_under_dot_car() {
        if std::env::var_os("HOME").is_none() {
            return; // CI sandbox without HOME: nothing to assert
        }
        let root = managed_root().unwrap();
        assert!(
            root.ends_with(".car/visual-runtime"),
            "root: {}",
            root.display()
        );
        // NOT a hardcoded `bin/vllm-mlx`: the production path deliberately goes
        // through `venv_program`, which is `Scriptsllm-mlx.exe` on Windows. A
        // literal POSIX suffix here asserted the one layout the code avoids, so this
        // failed on any Windows box whose shell sets HOME (Git Bash does).
        let bin = managed_server_bin().unwrap();
        assert_eq!(bin, crate::managed_venv::venv_program(&root, SERVER_BIN));
        assert!(bin.starts_with(&root), "bin: {}", bin.display());
    }

    #[test]
    fn search_dirs_include_path_and_managed() {
        // PATH entries always lead; the managed venv bin is always a candidate.
        let dirs = search_dirs();
        if let Some(root) = managed_root() {
            assert!(
                dirs.iter()
                    .any(|d| d == &crate::managed_venv::venv_bin_dir(&root)),
                "managed bin dir missing from search set"
            );
        }
    }

    #[test]
    fn parse_semver_handles_plain_and_prerelease() {
        assert_eq!(parse_semver("0.3.0"), Some((0, 3, 0)));
        assert_eq!(parse_semver("1.2.10"), Some((1, 2, 10)));
        assert_eq!(parse_semver("0.6.3rc1"), Some((0, 6, 3)));
        assert_eq!(parse_semver("0.2"), Some((0, 2, 0)));
        assert!(parse_semver("not-a-version").is_none());
        // The version floor must order correctly.
        assert!((0, 2, 9) < MIN_VERSION);
        assert!((0, 4, 1) < MIN_VERSION);
        assert!((0, 6, 0) >= MIN_VERSION);
        assert!((0, 6, 3) >= MIN_VERSION);
    }

    /// Puts the named variables back as they were, even when an assertion
    /// fails. Drop it before the environment lock.
    pub(super) struct RestoredEnvironment(Vec<(&'static str, Option<std::ffi::OsString>)>);

    impl RestoredEnvironment {
        pub(super) fn capture(names: &[&'static str]) -> Self {
            Self(
                names
                    .iter()
                    .map(|name| (*name, std::env::var_os(name)))
                    .collect(),
            )
        }
    }

    impl Drop for RestoredEnvironment {
        fn drop(&mut self) {
            for (name, value) in &self.0 {
                match value {
                    Some(value) => std::env::set_var(name, value),
                    None => std::env::remove_var(name),
                }
            }
        }
    }

    #[test]
    fn env_override_resolves_when_file_exists() {
        // Declared before the restore guard so the variable is put back
        // before the lock is released.
        let _environment = crate::openrouter::test_environment_scope();
        let _restore = RestoredEnvironment::capture(&["CAR_VLLM_MLX_BIN"]);

        // An override naming a file is used as-is, with no version check.
        let tmp = tempfile::TempDir::new().unwrap();
        let binary = tmp.path().join(SERVER_BIN);
        std::fs::write(&binary, b"fixture").unwrap();
        std::env::set_var("CAR_VLLM_MLX_BIN", &binary);
        assert_eq!(resolve_existing(), absolutize(&binary));

        // A non-file override is ignored (falls through to the search dirs).
        std::env::set_var("CAR_VLLM_MLX_BIN", "/nonexistent/vllm-mlx-xyz");
        let resolved = resolve_existing();
        assert!(
            resolved
                .as_ref()
                .map(|p| !p.ends_with("vllm-mlx-xyz"))
                .unwrap_or(true),
            "non-file override must not be returned"
        );
    }

    /// A uv configured with its own tool and executable dirs puts
    /// `uv tool install vllm-mlx` where the lookup still looks.
    #[test]
    fn uv_tool_dirs_named_by_the_environment_are_searched() {
        let _environment = crate::openrouter::test_environment_scope();
        let _restore = RestoredEnvironment::capture(&["UV_TOOL_BIN_DIR", "UV_TOOL_DIR"]);
        let tmp = tempfile::TempDir::new().unwrap();
        let bin = tmp.path().join("uv-bin");
        let tools = tmp.path().join("uv-tools");
        std::env::set_var("UV_TOOL_BIN_DIR", &bin);
        std::env::set_var("UV_TOOL_DIR", &tools);

        let dirs = search_dirs();
        assert!(dirs.contains(&bin), "{dirs:?}");
        let tool_env = crate::managed_venv::venv_bin_dir(&tools.join(SERVER_BIN));
        assert!(dirs.contains(&tool_env), "{dirs:?}");
    }

    /// Without `uv`, the error names an install the lookup finds, and for an
    /// outdated runtime an upgrade that fits what installed it: uv will not
    /// replace an executable it did not install, so a pip or pipx one is not
    /// sent to `uv tool install --upgrade`.
    #[test]
    fn missing_uv_names_a_fix_that_works_for_what_was_found() {
        let _environment = crate::openrouter::test_environment_scope();
        let _restore = RestoredEnvironment::capture(&["UV_TOOL_DIR"]);
        let tmp = tempfile::TempDir::new().unwrap();
        let tools = tmp.path().join("uv-tools");
        std::env::set_var("UV_TOOL_DIR", &tools);
        let from_uv = tools.join(SERVER_BIN).join("bin").join(SERVER_BIN);
        std::fs::create_dir_all(tools.join(SERVER_BIN)).unwrap();
        std::fs::write(tools.join(SERVER_BIN).join("uv-receipt.toml"), "[tool]\n").unwrap();
        let from_pip = tmp.path().join("Library/Python/3.12/bin").join(SERVER_BIN);
        // A pipx environment in the same folder has no uv receipt.
        let from_pipx = tools.join("pipx-venv").join("bin").join(SERVER_BIN);

        let missing = uv_missing(&Lookup::Missing).to_string();
        assert!(missing.contains("`uv tool install vllm-mlx`"), "{missing}");
        assert!(missing.contains("`CAR_VLLM_MLX_BIN`"), "{missing}");

        let uv_texts = [
            uv_missing(&Lookup::Outdated((0, 4, 1), from_uv.clone())).to_string(),
            Lookup::Outdated((0, 4, 1), from_uv.clone())
                .absence()
                .unwrap(),
        ];
        for text in &uv_texts {
            assert!(
                text.contains("`uv tool install --upgrade vllm-mlx`")
                    && text.contains(&from_uv.display().to_string()),
                "{text}"
            );
        }

        let mut pip_texts = Vec::new();
        for other in [&from_pip, &from_pipx] {
            for text in [
                uv_missing(&Lookup::Outdated((0, 4, 1), other.clone())).to_string(),
                Lookup::Outdated((0, 4, 1), other.clone())
                    .absence()
                    .unwrap(),
            ] {
                assert!(
                    !text.contains("uv tool install")
                        && text.contains("the tool that installed it")
                        && text.contains("`CAR_VLLM_MLX_BIN`")
                        && text.contains(&other.display().to_string()),
                    "{text}"
                );
                pip_texts.push(text);
            }
        }

        if let Some(root) = managed_root() {
            let managed = managed_server_bin().unwrap();
            let text = Lookup::Outdated((0, 4, 1), managed).absence().unwrap();
            let interpreter = crate::managed_venv::interpreter(&root);
            assert!(
                text.contains(&format!(
                    "`uv pip install --python \"{}\" \"{PIP_SPEC}\"`",
                    interpreter.display()
                )),
                "{text}"
            );
        }

        for text in uv_texts.iter().chain(&pip_texts).chain([&missing]) {
            assert!(!text.contains("`pip install"), "{text}");
        }
    }

    #[tokio::test]
    async fn health_probe_accepts_a_fragmented_http_status_line() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut request = [0u8; 256];
            let _ = stream.read(&mut request).await.unwrap();
            stream.write_all(b"HTTP/1.1 2").await.unwrap();
            tokio::time::sleep(Duration::from_millis(25)).await;
            stream
                .write_all(b"00 OK\r\nContent-Length: 0\r\n\r\n")
                .await
                .unwrap();
        });

        assert!(health_ok(&format!("http://{address}"), crate::LOADED_WAIT).await);
        server.await.unwrap();
    }
}

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

    fn p(s: &str) -> PathBuf {
        PathBuf::from(s)
    }

    /// The regression: a stale `uv tool` install on PATH sat at exactly
    /// MIN_VERSION and shadowed the newer CAR-managed runtime, because
    /// selection stopped at the first binary above the floor. 0.3.0 hard-fails
    /// on text-only checkpoints of multimodal architectures; 0.4.1 serves them.
    #[test]
    fn highest_version_wins_not_first_above_the_floor() {
        let picked = best_candidate(vec![
            // PATH order: the uv tool install is found first.
            (p("/home/u/.local/bin/vllm-mlx"), Some((0, 6, 0))),
            (
                p("/home/u/.car/visual-runtime/bin/vllm-mlx"),
                Some((0, 6, 2)),
            ),
        ]);
        assert_eq!(picked, Some(p("/home/u/.car/visual-runtime/bin/vllm-mlx")));
    }

    #[test]
    fn below_the_floor_is_rejected_even_when_it_is_the_only_one() {
        assert_eq!(
            best_candidate(vec![(p("/a/vllm-mlx"), Some((0, 2, 8)))]),
            None
        );
        // The install every machine had before the drafter floor: rejecting
        // it is what makes provisioning upgrade the managed runtime.
        assert_eq!(
            best_candidate(vec![(p("/a/vllm-mlx"), Some((0, 4, 1)))]),
            None
        );
    }

    #[test]
    fn an_undeterminable_version_is_not_selected() {
        assert_eq!(best_candidate(vec![(p("/a/vllm-mlx"), None)]), None);
        // ...but it must not mask a usable sibling.
        assert_eq!(
            best_candidate(vec![
                (p("/a/vllm-mlx"), None),
                (p("/b/vllm-mlx"), Some((0, 6, 0))),
            ]),
            Some(p("/b/vllm-mlx"))
        );
    }

    #[test]
    fn no_candidates_means_provision() {
        assert_eq!(best_candidate(vec![]), None);
    }

    /// Routing uses a found runtime with no further lookup until it is 30
    /// seconds old, or until the interpreter its binary names is gone, which
    /// is what a Homebrew Python upgrade does to a venv while leaving its
    /// files.
    #[tokio::test(flavor = "multi_thread")]
    async fn routing_reuses_a_found_runtime_until_it_is_stale() {
        let tmp = tempfile::TempDir::new().unwrap();
        let python = tmp.path().join("python");
        std::fs::write(&python, b"").unwrap();
        let server = tmp.path().join(SERVER_BIN);
        std::fs::write(&server, format!("#!{}\n", python.display())).unwrap();
        let runtime = test_runtime::State::new(Lookup::Ready(server.clone()));
        let ready = Lookup::Ready(server.clone());

        assert_eq!(runtime.run(status()).await, ready);
        assert_eq!(runtime.run(status()).await, ready);
        assert_eq!(
            runtime.lookups(),
            1,
            "the second call used what the first found"
        );

        std::fs::remove_file(&python).unwrap();
        runtime.run(status()).await;
        assert_eq!(
            runtime.lookups(),
            2,
            "a dead interpreter is looked up again"
        );

        runtime.age_remembered();
        runtime.run(status()).await;
        assert_eq!(runtime.lookups(), 3, "so is one 30 seconds old");
    }

    /// Server start always looks again, whatever routing remembered: here
    /// routing found a runtime, then its interpreter died (files in place),
    /// and a never-download request still routed to it. Server start finds it
    /// missing and the provisioning backstop refuses with the named error.
    ///
    /// Watched failing: with server start using the remembered runtime, it
    /// returned that runtime and never reached the backstop.
    #[tokio::test(flavor = "multi_thread")]
    async fn server_start_looks_again_whatever_routing_remembered() {
        let tmp = tempfile::TempDir::new().unwrap();
        let server = tmp.path().join(SERVER_BIN);
        std::fs::write(&server, b"fixture").unwrap();
        let runtime = test_runtime::State::new(Lookup::Ready(server.clone()));
        assert_eq!(runtime.run(status()).await, Lookup::Ready(server));

        runtime.set(Lookup::Missing);
        let refused = runtime
            .run(crate::registry::never_download_scope(
                true,
                ensure_runtime(),
            ))
            .await
            .expect_err("server start looked again and found no runtime");
        assert!(
            refused.to_string().contains("not installed")
                && refused.to_string().contains("`uv tool install vllm-mlx`"),
            "{refused}"
        );
        assert_eq!(runtime.lookups(), 2);
        assert_eq!(
            runtime.provisions(),
            1,
            "the backstop refused at provisioning"
        );
    }

    /// A runtime older than the floor is named with its version and the
    /// command that upgrades it, by the routing refusal and by the backstop.
    #[tokio::test(flavor = "multi_thread")]
    async fn an_outdated_runtime_is_named_with_the_upgrade_command() {
        let _environment = crate::openrouter::test_environment_scope_async().await;
        let _restore = super::tests::RestoredEnvironment::capture(&["UV_TOOL_DIR"]);
        let tmp = tempfile::TempDir::new().unwrap();
        std::env::set_var("UV_TOOL_DIR", tmp.path());
        let from_uv = tmp.path().join(SERVER_BIN).join("bin").join(SERVER_BIN);
        std::fs::create_dir_all(tmp.path().join(SERVER_BIN)).unwrap();
        std::fs::write(
            tmp.path().join(SERVER_BIN).join("uv-receipt.toml"),
            "[tool]\n",
        )
        .unwrap();
        let runtime = test_runtime::State::new(Lookup::Outdated((0, 4, 1), from_uv));
        let schema = crate::registry::builtin_catalog()
            .into_iter()
            .find(crate::schema::ModelSchema::is_car_managed_vllm_mlx)
            .expect("managed vllm fixture schema");

        let found = runtime.run(status()).await;
        let refusal = not_ready_error(&schema, &found).to_string();
        let backstop = runtime
            .run(crate::registry::never_download_scope(
                true,
                ensure_runtime(),
            ))
            .await
            .unwrap_err()
            .to_string();
        for text in [&refusal, &backstop] {
            assert!(
                text.contains("is 0.4.1, older than the 0.6.0 CAR needs")
                    && text.contains("`uv tool install --upgrade vllm-mlx`"),
                "{text}"
            );
        }
        assert!(refusal.contains(&schema.name), "{refusal}");
        assert_eq!(
            runtime.provisions(),
            1,
            "the backstop refused at provisioning"
        );
    }
}

#[cfg(all(test, unix))]
mod hf_token_tests {
    use super::*;
    use crate::hf_cache::token_probe::{Probe, WITH_TOKEN};

    async fn recorded_for(credentials: &[(&str, &str)]) -> String {
        let probe = Probe::new();
        let hf = probe.dir().join("hf");
        probe.fake_program(&hf);
        let mut cmd = car_secrets::with_child_credentials(credentials, || {
            hf_download_command(&hf, "org/repo", false)
        });
        cmd.output().await.unwrap();
        probe.recorded()
    }

    #[tokio::test]
    async fn hf_download_is_handed_the_token() {
        assert_eq!(recorded_for(WITH_TOKEN).await, "yes");
    }

    #[tokio::test]
    async fn hf_download_gets_no_token_when_there_is_none() {
        assert_eq!(recorded_for(&[]).await, "no");
    }
}