car-inference 0.49.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 (`pip install vllm-mlx`); 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. `< 0.3.0` (mlx-vlm `< ~0.6`) hits the
/// upstream `RuntimeError: There is no Stream(gpu, 0)` on every server-side
/// generation — verified fixed at vllm-mlx 0.3.0 / mlx-vlm 0.6.3. A binary older
/// than this is skipped during resolution so we provision a working one instead.
const MIN_VERSION: (u64, u64, u64) = (0, 3, 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.3.0";

/// Errors from locating or provisioning the vLLM-MLX runtime.
#[derive(Debug, thiserror::Error)]
pub enum RuntimeError {
    #[error(
        "`uv` is required to provision the vllm-mlx runtime but was not found on PATH; \
             install uv (https://docs.astral.sh/uv/) or `pip install vllm-mlx` yourself"
    )]
    UvMissing,
    #[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 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))
}

/// Directories searched for an existing `vllm-mlx`, in priority order: an
/// explicit override, then PATH, `~/.local/bin`, the `uv tool` bin dir, and the
/// CAR-managed venv. 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();
    if let Some(home) = std::env::var_os("HOME").map(PathBuf::from) {
        dirs.push(home.join(".local").join("bin"));
        dirs.push(
            home.join(".local")
                .join("share")
                .join("uv")
                .join("tools")
                .join("vllm-mlx")
                .join("bin"),
        );
        dirs.push(home.join(".car").join("visual-runtime").join("bin"));
    }
    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> {
    // 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);
        }
    }
    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 = std::fs::canonicalize(&candidate).unwrap_or_else(|_| candidate.clone());
        if seen.contains(&canon) {
            continue;
        }
        seen.push(canon);
        let version = binary_version(&candidate);
        candidates.push((candidate, version));
    }
    best_candidate(candidates).and_then(|path| absolutize(&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> {
    std::fs::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.
pub async fn ensure_runtime() -> Result<VllmRuntime, RuntimeError> {
    if let Some(server) = resolve_existing() {
        return Ok(VllmRuntime { server });
    }
    provision().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`.
async fn provision() -> Result<(), RuntimeError> {
    if which("uv").is_none() {
        return Err(RuntimeError::UvMissing);
    }
    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(())
}

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

/// 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 mut cmd = Command::new(&hf);
        cmd.arg("download").arg(repo).kill_on_drop(true);
        if disable_xet {
            cmd.env("HF_HUB_DISABLE_XET", "1");
        }
        let output = cmd.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()
}

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

/// 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 {
    let url = format!("{}/health", endpoint.trim_end_matches('/'));
    let client = match reqwest::Client::builder().timeout(timeout).build() {
        Ok(c) => c,
        Err(_) => return false,
    };
    matches!(client.get(&url).send().await, Ok(r) if r.status().is_success())
}

#[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()
        );
        let bin = managed_server_bin().unwrap();
        assert!(bin.ends_with(".car/visual-runtime/bin/vllm-mlx"));
    }

    #[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 std::env::var_os("HOME").is_some() {
            assert!(
                dirs.iter().any(|d| d.ends_with(".car/visual-runtime/bin")),
                "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, 3, 0) >= MIN_VERSION);
        assert!((0, 6, 3) >= MIN_VERSION);
    }

    #[test]
    fn env_override_resolves_when_file_exists() {
        // A non-file override is ignored (falls through to the search dirs).
        std::env::set_var("CAR_VLLM_MLX_BIN", "/nonexistent/vllm-mlx-xyz");
        // Can't assert a positive without a real binary; assert the override path
        // is not blindly returned when it isn't a file.
        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"
        );
        std::env::remove_var("CAR_VLLM_MLX_BIN");
    }
}

#[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, 3, 0))),
            (
                p("/home/u/.car/visual-runtime/bin/vllm-mlx"),
                Some((0, 4, 1)),
            ),
        ]);
        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
        );
    }

    #[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, 4, 1))),
            ]),
            Some(p("/b/vllm-mlx"))
        );
    }

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