car-inference 0.49.0

//! On-demand pool of supervised `vllm-mlx` server processes.
//!
//! The engine manages these exactly like its other backends: lazy-started on the
//! first request for a `vllm-mlx/*` model, health-waited before routing, and
//! idle-evicted by the same loop that reaps in-process backends. The point is
//! that a server-backed (multimodal / unsupported-arch) model is
//! indistinguishable from an in-process one to the caller — no manual server to
//! start, no endpoint to configure.
//!
//! `vllm-mlx serve <model> --port <P>` serves **one** model per process, so the
//! pool keys a process per model id, each on a CAR-allocated free loopback port.
//! Teardown relies on `kill_on_drop`: removing a [`ManagedServer`] from the map
//! SIGKILLs its child.

use std::collections::HashMap;
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::{Duration, Instant};

use tokio::process::{Child, Command};
use tokio::sync::Mutex;
use tracing::{info, warn};

use crate::vllm_runtime;

/// Per-`/health` probe timeout during the readiness wait.
const HEALTH_TIMEOUT: Duration = Duration::from_secs(3);
/// How long to wait for a freshly-spawned server to become healthy. A cold model
/// load (weights off disk into the GPU) can take tens of seconds.
const READY_DEADLINE: Duration = Duration::from_secs(120);
/// How long the server may make **no observable progress** before we give up.
///
/// First use of a model downloads its weights, and a 16-40 GB pull cannot
/// finish inside any sane absolute deadline — the flat 120s cap meant CAR could
/// only ever start models that were already cached, which defeated `car models
/// add <repo>` followed by `car infer` for every real-sized model. The server
/// writes download and load progress to its stderr log continuously, so a
/// growing log is proof of life; this bounds the silence, not the total time.
const STALL_TIMEOUT: Duration = Duration::from_secs(180);
/// Poll cadence while waiting for readiness.
const READY_POLL: Duration = Duration::from_millis(500);

struct ManagedServer {
    /// The `vllm-mlx serve` child. `kill_on_drop` is set, so dropping this struct
    /// terminates the server.
    child: Child,
    port: u16,
    last_used: Instant,
}

impl ManagedServer {
    fn endpoint(&self) -> String {
        format!("http://127.0.0.1:{}", self.port)
    }

    /// True while the child is still running (not yet reaped).
    fn is_alive(&mut self) -> bool {
        matches!(self.child.try_wait(), Ok(None))
    }
}

/// On-demand pool of supervised `vllm-mlx` servers, one per model id.
pub struct VllmServerPool {
    servers: Mutex<HashMap<String, ManagedServer>>,
    idle_ttl: Duration,
}

impl VllmServerPool {
    pub fn new(idle_ttl: Duration) -> Self {
        Self {
            servers: Mutex::new(HashMap::new()),
            idle_ttl,
        }
    }

    /// Ensure a healthy server for `model_id` (whose backing HF model is
    /// `runtime_model`) and return its loopback endpoint. Starts the process and
    /// health-waits on first use; subsequent calls bump the idle timer and return
    /// the cached endpoint. A crashed server is respawned transparently.
    ///
    /// NB: a single lock serializes starts across models — multimodal traffic is
    /// occasional, so the simplicity is worth more than concurrent cold-starts.
    pub async fn ensure(
        &self,
        model_id: &str,
        runtime_model: &str,
        family: &str,
    ) -> Result<String, String> {
        let mut servers = self.servers.lock().await;

        if let Some(s) = servers.get_mut(model_id) {
            if s.is_alive() {
                s.last_used = Instant::now();
                return Ok(s.endpoint());
            }
            warn!(
                model = model_id,
                "supervised vllm-mlx server died; respawning"
            );
            servers.remove(model_id);
        }

        let runtime = vllm_runtime::ensure_runtime()
            .await
            .map_err(|e| format!("vllm-mlx runtime unavailable: {e}"))?;
        // A wedged Xet transfer is indistinguishable from a hang and never
        // recovers on its own, so one stalled start is retried with Xet off.
        // Bounded at two attempts: the second has no fallback left, and a
        // genuinely broken model should fail fast rather than loop.
        let mut disable_xet = false;
        loop {
            let port = alloc_loopback_port()?;
            let endpoint = format!("http://127.0.0.1:{port}");

            info!(
                model = model_id,
                runtime_model, port, disable_xet, "starting supervised vllm-mlx server"
            );
            let child = spawn_server(&runtime.server, runtime_model, port, disable_xet, family)
                .map_err(|e| format!("failed to spawn vllm-mlx serve: {e}"))?;
            let mut server = ManagedServer {
                child,
                port,
                last_used: Instant::now(),
            };

            match wait_ready(&endpoint, &mut server, runtime_model).await {
                Ok(()) => {
                    info!(model = model_id, endpoint = %endpoint, "vllm-mlx server ready");
                    servers.insert(model_id.to_string(), server);
                    return Ok(endpoint);
                }
                Err(NotReady::Stalled(reason)) if !disable_xet => {
                    // `server` is dropped at the end of this arm; `kill_on_drop`
                    // reaps the wedged child before the retry.
                    warn!(
                        model = model_id,
                        reason, "vllm-mlx startup stalled; retrying with HuggingFace Xet disabled"
                    );
                    disable_xet = true;
                }
                Err(other) => return Err(other.into_message()),
            }
        }
    }

    /// Stop servers idle longer than the TTL (and reap any that have died).
    /// Returns the number stopped. Called from the engine's idle-eviction loop.
    pub async fn evict_idle(&self) -> usize {
        let mut servers = self.servers.lock().await;
        let now = Instant::now();
        let ttl = self.idle_ttl;
        let mut stale: Vec<String> = Vec::new();
        for (k, s) in servers.iter_mut() {
            if now.duration_since(s.last_used) > ttl || !s.is_alive() {
                stale.push(k.clone());
            }
        }
        for k in &stale {
            info!(model = %k, "evicting idle vllm-mlx server");
            servers.remove(k); // Drop → kill_on_drop SIGKILLs the child.
        }
        stale.len()
    }

    /// Number of currently-managed servers (for status/telemetry).
    pub async fn len(&self) -> usize {
        self.servers.lock().await.len()
    }

    /// Whether the pool currently manages no servers.
    pub async fn is_empty(&self) -> bool {
        self.servers.lock().await.is_empty()
    }
}

/// Reserve a free loopback TCP port by binding `:0` and releasing it. There is an
/// unavoidable TOCTOU window before the server binds, but the OS won't hand the
/// same ephemeral port to two binds in quick succession in practice.
fn alloc_loopback_port() -> Result<u16, String> {
    let listener = TcpListener::bind("127.0.0.1:0")
        .map_err(|e| format!("could not allocate a local port: {e}"))?;
    listener
        .local_addr()
        .map(|a| a.port())
        .map_err(|e| format!("could not read allocated port: {e}"))
}

/// Spawn `vllm-mlx serve <runtime_model> --port <port>`. stdout/stderr go to log
/// files under the state root's `logs/` (`~/.car/logs` by default, `CAR_HOME`
/// moves it) when resolvable, else are discarded.
fn spawn_server(
    server_bin: &Path,
    runtime_model: &str,
    port: u16,
    disable_xet: bool,
    family: &str,
) -> std::io::Result<Child> {
    let (out, err) = log_sinks(port);
    let mut cmd = Command::new(server_bin);
    cmd.arg("serve")
        .arg(runtime_model)
        .arg("--port")
        .arg(port.to_string())
        // Tool calling is OFF in vllm-mlx unless asked for, and the server then
        // returns `tool_calls: null` no matter what the model emits — it will
        // happily reason its way to a call and have the result dropped on the
        // floor. Every `vllm-mlx/*` catalog entry advertises `tool_use` and
        // `multi_tool_call`, so without these flags that claim was false and any
        // agent routed to one silently never saw a tool call. `auto` picks the
        // parser from the model's family, which is what keeps this from needing
        // a per-architecture table CAR would have to maintain.
        .arg("--enable-auto-tool-choice")
        .arg("--tool-call-parser")
        .arg("auto")
        .stdin(Stdio::null())
        .stdout(out)
        .stderr(err)
        .kill_on_drop(true);
    if let Some(parser) = reasoning_parser_for(family) {
        cmd.arg("--reasoning-parser").arg(parser);
    }
    if disable_xet {
        let (key, value) = XET_DISABLE_ENV;
        cmd.env(key, value);
    }
    cmd.spawn()
}

/// The `--reasoning-parser` vllm-mlx should use for a model family, if any.
///
/// Without one, a reasoning model's chain-of-thought is returned *as the
/// answer*: ask a Qwen3.x model to "say hi" and you get its deliberation about
/// how to greet you. CAR strips `<think>` blocks on its native MLX path, but
/// `strip_thinking` is never called on the remote/OpenAI path that vllm-mlx
/// uses, so the whole external shelf leaked reasoning into `content`. Letting
/// the server split it out is cleaner than teaching CAR's protocol handler a
/// per-family delimiter.
///
/// Matched loosely on purpose: `family` is `qwen3.8` for a catalog entry and
/// `qwen3_5_moe` for one derived from a repo's `config.json`, and both mean the
/// same parser. Unknown families get no flag, which is exactly the old
/// behavior — never a wrong parser.
fn reasoning_parser_for(family: &str) -> Option<&'static str> {
    let f = family.to_ascii_lowercase();
    if f.contains("qwen") {
        Some("qwen3")
    } else if f.contains("gemma") {
        Some("gemma4")
    } else if f.contains("glm") {
        Some("glm4")
    } else if f.contains("deepseek") {
        Some("deepseek_r1")
    } else if f.contains("gpt") && f.contains("oss") {
        Some("gpt_oss")
    } else {
        None
    }
}

/// Environment that forces HuggingFace's classic HTTPS download path instead of
/// Xet chunk transfer.
///
/// Xet is the default and is normally faster, but when its transfer wedges it
/// does so silently and completely — zero bytes, indefinitely, while a plain
/// HTTPS fetch of the very same file proceeds fine. A model that never
/// downloads is a model that never loads, so a stalled startup is retried once
/// with this set rather than surfaced as an unexplained failure.
const XET_DISABLE_ENV: (&str, &str) = ("HF_HUB_DISABLE_XET", "1");

/// stdout/stderr sinks for a server, keyed by port so concurrent servers don't
/// clobber each other's logs. The logs live in `logs/` under the CAR state root
/// — `~/.car/logs` unless `CAR_HOME` moves the root, in which case they move
/// with it, so a relocated daemon's server output lands next to the rest of its
/// state instead of interleaving with the primary's. Falls back to `/dev/null`
/// when the state root can't be resolved or the directory can't be created.
fn log_sinks(port: u16) -> (Stdio, Stdio) {
    let open = |suffix: &str| {
        let path = log_path(port, suffix)?;
        std::fs::create_dir_all(path.parent()?).ok()?;
        std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
            .ok()
            .map(Stdio::from)
    };
    match (open("stdout"), open("stderr")) {
        (Some(o), Some(e)) => (o, e),
        _ => (Stdio::null(), Stdio::null()),
    }
}

/// Where a server's `stdout`/`stderr` log lives, or `None` when the state root
/// cannot be resolved. Single definition so the spawn sink and the readiness
/// watcher cannot disagree about the path.
fn log_path(port: u16, suffix: &str) -> Option<PathBuf> {
    car_home::root().map(|root| {
        root.join("logs")
            .join(format!("vllm-mlx-{port}.{suffix}.log"))
    })
}

/// Poll `/health` until the server is ready. Aborts early (with a precise
/// error) if the child exits during startup, or if it goes quiet.
///
/// Bounds *silence* rather than total time whenever the server's log is
/// observable, because first use of a model includes downloading its weights
/// and that is legitimately unbounded — see [`STALL_TIMEOUT`]. When the log
/// cannot be observed (state root unresolvable, so output went to `/dev/null`)
/// there is no progress signal to use and this falls back to the absolute
/// [`READY_DEADLINE`].
async fn wait_ready(
    endpoint: &str,
    server: &mut ManagedServer,
    runtime_model: &str,
) -> Result<(), NotReady> {
    let start = Instant::now();
    let log = log_path(server.port, "stderr");
    let cache = crate::registry::huggingface_repo_dir(runtime_model);
    let xet = xet_cache_root();
    let mut last_mark = progress_mark(log.as_deref(), &cache, &xet);
    let mut last_progress = Instant::now();

    loop {
        if vllm_runtime::health_ok(endpoint, HEALTH_TIMEOUT).await {
            return Ok(());
        }
        if !server.is_alive() {
            return Err(NotReady::Other(format!(
                "vllm-mlx server exited during startup (see ~/.car/logs/vllm-mlx-{}.stderr.log)",
                server.port
            )));
        }

        match progress_mark(log.as_deref(), &cache, &xet) {
            // Either the log or the weight cache growing means real work.
            Some(mark) => {
                if Some(mark) != last_mark {
                    last_mark = Some(mark);
                    last_progress = Instant::now();
                }
                if last_progress.elapsed() > STALL_TIMEOUT {
                    return Err(NotReady::Stalled(format!(
                        "vllm-mlx server made no progress for {}s and never became healthy at \
                         {endpoint} (see ~/.car/logs/vllm-mlx-{}.stderr.log)",
                        STALL_TIMEOUT.as_secs(),
                        server.port
                    )));
                }
            }
            // Nothing observable — cannot distinguish "working" from "wedged".
            None => {
                if start.elapsed() > READY_DEADLINE {
                    return Err(NotReady::Stalled(format!(
                        "vllm-mlx server did not become healthy at {endpoint} within {}s",
                        READY_DEADLINE.as_secs()
                    )));
                }
            }
        }
        tokio::time::sleep(READY_POLL).await;
    }
}

/// Why a server never reached readiness. `Stalled` is separated because it is
/// the one case worth retrying differently — it means the process is alive but
/// achieving nothing, which is what a wedged download looks like.
#[derive(Debug)]
enum NotReady {
    Stalled(String),
    Other(String),
}

impl NotReady {
    fn into_message(self) -> String {
        match self {
            NotReady::Stalled(m) | NotReady::Other(m) => m,
        }
    }
}

/// Byte length of `path`, or `None` if it cannot be read.
fn file_len(path: &Path) -> Option<u64> {
    std::fs::metadata(path).ok().map(|m| m.len())
}

/// A number that strictly increases while the server is doing real work:
/// stderr bytes written, plus bytes of model weights on disk.
///
/// The log alone is not enough, which is the trap this exists to avoid. The
/// downloader's progress bar ticks once per *file completed* (`Fetching 13
/// files: 77%|...| 10/13`), so while it pulls a multi-gigabyte weight shard the
/// log is completely silent for minutes — precisely the window a startup
/// watchdog must not treat as a hang. The partially written blobs in the
/// HuggingFace cache are the honest signal there.
///
/// `None` only when neither is observable, which is the caller's cue to fall
/// back to an absolute deadline.
fn progress_mark(log: Option<&Path>, cache_dir: &Path, xet_root: &Path) -> Option<u64> {
    let log_bytes = log.and_then(file_len);
    let cache_bytes = dir_bytes(cache_dir);
    // HuggingFace serves large files through Xet, whose chunks land in a
    // separate cache (`~/.cache/huggingface/xet`) and are only materialized
    // into the repo's `blobs/` at the end. During the bulk of a multi-gigabyte
    // pull neither the log nor `cache_dir` moves at all, so without this a
    // Xet-backed download looks identical to a hang. Directory mtimes change as
    // chunk files are created, and a handful of `stat`s is cheap enough to do
    // every poll — far cheaper than walking a cache that is routinely tens of
    // gigabytes.
    let xet_activity = newest_dir_mtime(xet_root);
    match (log_bytes, cache_bytes, xet_activity) {
        (None, None, None) => None,
        (a, b, c) => Some(
            a.unwrap_or(0)
                .saturating_add(b.unwrap_or(0))
                .saturating_add(c.unwrap_or(0)),
        ),
    }
}

/// Root of HuggingFace's Xet chunk cache, honoring `HF_HOME`.
fn xet_cache_root() -> PathBuf {
    crate::registry::huggingface_cache_root()
        .parent()
        .map(|hf| hf.join("xet"))
        .unwrap_or_else(|| PathBuf::from("xet"))
}

/// Newest mtime (epoch seconds) among `dir` and its subdirectories, two levels
/// deep. A directory's mtime advances when entries are added or removed, which
/// is exactly what an in-flight chunked download does continuously.
fn newest_dir_mtime(dir: &Path) -> Option<u64> {
    if !dir.is_dir() {
        return None;
    }
    fn mtime_secs(path: &Path) -> u64 {
        std::fs::metadata(path)
            .ok()
            .and_then(|m| m.modified().ok())
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_secs())
            .unwrap_or(0)
    }
    let mut newest = mtime_secs(dir);
    let mut level = vec![dir.to_path_buf()];
    for _ in 0..2 {
        let mut next = Vec::new();
        for d in &level {
            let Ok(entries) = std::fs::read_dir(d) else {
                continue;
            };
            for entry in entries.flatten() {
                if entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
                    let path = entry.path();
                    newest = newest.max(mtime_secs(&path));
                    next.push(path);
                }
            }
        }
        level = next;
    }
    Some(newest)
}

/// Total bytes of files directly under `dir` and its immediate subdirectories
/// — enough to see HuggingFace's `blobs/` (including `.incomplete` partials)
/// grow, without paying for a deep walk on every poll. `None` when `dir` does
/// not exist yet.
fn dir_bytes(dir: &Path) -> Option<u64> {
    if !dir.is_dir() {
        return None;
    }
    let mut total = 0u64;
    let mut stack = vec![dir.to_path_buf()];
    let mut depth = 0;
    while let Some(current) = stack.pop() {
        let Ok(entries) = std::fs::read_dir(&current) else {
            continue;
        };
        for entry in entries.flatten() {
            match entry.metadata() {
                Ok(m) if m.is_file() => total = total.saturating_add(m.len()),
                Ok(m) if m.is_dir() && depth < 2 => stack.push(entry.path()),
                _ => {}
            }
        }
        depth += 1;
    }
    Some(total)
}

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

    #[test]
    fn alloc_port_returns_distinct_usable_ports() {
        let a = alloc_loopback_port().unwrap();
        let b = alloc_loopback_port().unwrap();
        assert_ne!(a, 0);
        assert_ne!(b, 0);
        // Re-binding the freed port must succeed (it was released).
        assert!(TcpListener::bind(("127.0.0.1", a)).is_ok());
    }

    #[tokio::test]
    async fn evict_idle_on_empty_pool_is_zero() {
        let pool = VllmServerPool::new(Duration::from_secs(300));
        assert_eq!(pool.evict_idle().await, 0);
        assert_eq!(pool.len().await, 0);
    }

    /// End-to-end mechanics — spawn a stand-in `serve <model> --port <P>` process
    /// that answers `/health`, then prove the pool's spawn + readiness-wait reach
    /// "healthy" and that `kill_on_drop` tears it down. Uses a tiny Python HTTP
    /// server so it doesn't depend on the real vllm-mlx runtime or any weights.
    #[tokio::test]
    async fn spawns_and_health_waits_a_stand_in_server() {
        let Some(python) = vllm_runtime::which("python3") else {
            eprintln!("SKIP: python3 not available");
            return;
        };
        // A fake `vllm-mlx`: `serve <model> --port <P>` → HTTP 200 on every GET.
        let dir = std::env::temp_dir().join(format!("car-vllm-pool-test-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let script = dir.join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\n\
                 import sys, http.server\n\
                 port = int(sys.argv[sys.argv.index('--port') + 1])\n\
                 class H(http.server.BaseHTTPRequestHandler):\n\
                 \x20   def do_GET(self):\n\
                 \x20       self.send_response(200); self.end_headers(); self.wfile.write(b'ok')\n\
                 \x20   def log_message(self, *a): pass\n\
                 http.server.HTTPServer(('127.0.0.1', port), H).serve_forever()\n",
                python.display()
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let port = alloc_loopback_port().unwrap();
        let child = spawn_server(&script, "dummy/model", port, false, "qwen3").expect("spawn");
        let mut server = ManagedServer {
            child,
            port,
            last_used: Instant::now(),
        };
        let endpoint = server.endpoint();

        wait_ready(&endpoint, &mut server, "test-org/stand-in-model")
            .await
            .expect("stand-in server should become healthy");
        assert!(vllm_runtime::health_ok(&endpoint, Duration::from_secs(2)).await);
        assert!(server.is_alive());

        // kill_on_drop tears the child down; the port frees up afterwards.
        drop(server);
        let _ = std::fs::remove_dir_all(&dir);
    }
}

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

    /// The bug this guards: a first-run model download cannot finish inside any
    /// absolute deadline, so readiness must bound *silence*, not total time. A
    /// 16 GB pull under the old flat 120s cap always failed with "did not
    /// become healthy", which made `car models add` + `car infer` unusable for
    /// every real-sized model.
    #[test]
    fn stall_timeout_is_the_bound_not_total_elapsed_time() {
        assert!(
            STALL_TIMEOUT >= Duration::from_secs(120),
            "a stall bound shorter than a slow model-load step would reintroduce \
             spurious startup failures"
        );
    }

    #[test]
    fn log_path_is_stable_and_suffix_keyed() {
        let (out, err) = (log_path(4242, "stdout"), log_path(4242, "stderr"));
        // When there is no resolvable state root both are `None`: readiness then
        // falls back to the absolute deadline, which is documented behavior, so
        // there is nothing to assert in that environment.
        if let (Some(o), Some(e)) = (out, err) {
            assert_ne!(o, e, "stdout and stderr must not share a file");
            assert!(o.ends_with("vllm-mlx-4242.stdout.log"), "{}", o.display());
            assert!(e.ends_with("vllm-mlx-4242.stderr.log"), "{}", e.display());
            // The watcher reads exactly what the spawn sink writes.
            assert_eq!(e.parent(), o.parent());
        }
    }

    /// The trap: the downloader's progress bar ticks once per *file completed*,
    /// so during a multi-gigabyte weight shard the log is silent for minutes.
    /// A watchdog keyed only on log growth calls that a hang and kills a
    /// perfectly healthy download — which is how a 16 GB model died at 26 MB.
    /// Bytes landing in the weight cache are the signal that keeps moving.
    #[test]
    fn a_silent_log_still_counts_as_progress_while_weights_land() {
        let dir = tempfile::tempdir().unwrap();
        let log = dir.path().join("server.stderr.log");
        std::fs::write(&log, b"Fetching 13 files: 77%").unwrap();
        let cache = dir.path().join("models--org--big");
        std::fs::create_dir_all(cache.join("blobs")).unwrap();

        let no_xet = dir.path().join("no-xet");
        let before = progress_mark(Some(&log), &cache, &no_xet).expect("observable");

        // The log does not move; a partial blob grows. This is the whole case.
        std::fs::write(
            cache.join("blobs").join("shard.incomplete"),
            vec![0u8; 4096],
        )
        .unwrap();
        let after = progress_mark(Some(&log), &cache, &no_xet).expect("observable");

        assert!(
            after > before,
            "weights landing must register as progress even with a silent log: {before} -> {after}"
        );
    }

    /// HuggingFace serves large files through Xet: chunks land in a separate
    /// cache and are materialized into `blobs/` only at the end. So for the
    /// biggest models — the ones most likely to outlast any timeout — neither
    /// the log nor the repo dir moves for most of the download.
    #[test]
    fn directory_activity_is_visible_without_any_byte_growth() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("xet");
        std::fs::create_dir_all(root.join("cas-server").join("chunk-cache")).unwrap();

        let before = newest_dir_mtime(&root).expect("existing dir is observable");
        // A new chunk directory appearing advances its parent's mtime, which is
        // what an in-flight chunked download does continuously.
        std::fs::create_dir_all(root.join("cas-server").join("staging")).unwrap();
        let after = newest_dir_mtime(&root).expect("still observable");

        assert!(
            after >= before,
            "mtime must not go backwards: {before} -> {after}"
        );
        assert!(newest_dir_mtime(&dir.path().join("absent")).is_none());
    }

    #[test]
    fn progress_is_unobservable_only_when_neither_source_exists() {
        let dir = tempfile::tempdir().unwrap();
        let missing_log = dir.path().join("nope.log");
        let missing_cache = dir.path().join("nope-cache");
        let missing_xet = dir.path().join("nope-xet");
        assert_eq!(
            progress_mark(Some(&missing_log), &missing_cache, &missing_xet),
            None,
            "with nothing to watch, the caller must fall back to an absolute deadline"
        );

        std::fs::create_dir_all(&missing_cache).unwrap();
        assert!(
            progress_mark(Some(&missing_log), &missing_cache, &missing_xet).is_some(),
            "an existing cache dir is observable even before any bytes arrive"
        );

        // Any one observable source is enough — the Xet cache alone counts,
        // which is the only signal a large Xet-backed download provides.
        std::fs::create_dir_all(dir.path().join("real-xet")).unwrap();
        assert!(
            progress_mark(None, &missing_cache, &dir.path().join("real-xet")).is_some(),
            "an observable Xet cache must make progress measurable on its own"
        );
    }

    #[test]
    fn file_len_reports_growth_and_tolerates_a_missing_file() {
        let dir = tempfile::tempdir().unwrap();
        let f = dir.path().join("x.log");
        assert_eq!(
            file_len(&f),
            None,
            "missing file must not look like progress"
        );

        std::fs::write(&f, b"loading").unwrap();
        let first = file_len(&f).expect("written file has a length");
        std::fs::write(&f, b"loading... fetching shard 2 of 7").unwrap();
        let second = file_len(&f).expect("still readable");
        assert!(
            second > first,
            "growth must be observable: {first} -> {second}"
        );
    }
}

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

    /// vllm-mlx returns `tool_calls: null` for every request unless tool
    /// calling is explicitly enabled — the model reasons its way to a call and
    /// the result is dropped. Since every `vllm-mlx/*` catalog entry advertises
    /// `tool_use`, omitting these flags made that claim false.
    #[tokio::test]
    async fn tool_calling_flags_reach_the_spawned_process() {
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("argv.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintf '%s\\n' \"$@\" > {}\n",
                sh.display(),
                out.display()
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let mut child = spawn_server(&script, "org/model", 1234, false, "qwen3.8").expect("spawn");
        let _ = child.wait().await;
        let argv: Vec<String> = std::fs::read_to_string(&out)
            .unwrap()
            .lines()
            .map(str::to_string)
            .collect();

        assert!(
            argv.contains(&"--enable-auto-tool-choice".to_string()),
            "tool calling must be enabled or the advertised tool_use capability is a lie: {argv:?}"
        );
        let parser = argv
            .iter()
            .position(|a| a == "--tool-call-parser")
            .and_then(|i| argv.get(i + 1));
        assert_eq!(
            parser.map(String::as_str),
            Some("auto"),
            "--enable-auto-tool-choice requires a parser; `auto` avoids a \
             per-architecture table CAR would have to maintain: {argv:?}"
        );
        // The basics must survive the addition.
        assert_eq!(argv.first().map(String::as_str), Some("serve"));
        assert!(argv.contains(&"org/model".to_string()));
        assert!(argv.contains(&"1234".to_string()));
    }

    /// The retry is only worth anything if the variable actually reaches the
    /// child, so assert it end-to-end through a real spawn rather than trusting
    /// the builder.
    #[tokio::test]
    async fn disable_xet_reaches_the_spawned_process() {
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("env.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintenv {} > {} 2>&1 || echo UNSET > {}\n",
                sh.display(),
                XET_DISABLE_ENV.0,
                out.display(),
                out.display()
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        // First attempt: Xet untouched, so the child must not see the variable.
        let mut child = spawn_server(&script, "dummy/model", 1, false, "qwen3").expect("spawn");
        let _ = child.wait().await;
        assert_eq!(
            std::fs::read_to_string(&out).unwrap().trim(),
            "UNSET",
            "the default path must leave Xet enabled — it is normally the faster one"
        );

        // Retry attempt: the fallback must be visible to the downloader.
        let mut child = spawn_server(&script, "dummy/model", 1, true, "qwen3").expect("spawn");
        let _ = child.wait().await;
        assert_eq!(
            std::fs::read_to_string(&out).unwrap().trim(),
            XET_DISABLE_ENV.1,
            "the stalled-start retry must actually disable Xet in the child"
        );
    }
}

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

    /// `family` arrives in two spellings — `qwen3.8` from a catalog entry,
    /// `qwen3_5_moe` from a schema derived off a repo's `config.json` — and both
    /// mean the same parser.
    #[test]
    fn both_family_spellings_map_to_the_same_parser() {
        for family in ["qwen3.8", "qwen3.5", "qwen3_5_moe", "qwen3", "Qwen3.6"] {
            assert_eq!(
                reasoning_parser_for(family),
                Some("qwen3"),
                "family `{family}` should select the qwen3 reasoning parser"
            );
        }
        assert_eq!(reasoning_parser_for("gemma4_unified"), Some("gemma4"));
        assert_eq!(reasoning_parser_for("glm4_moe_lite"), Some("glm4"));
        assert_eq!(reasoning_parser_for("glm4.7"), Some("glm4"));
        assert_eq!(reasoning_parser_for("deepseek_v3"), Some("deepseek_r1"));
    }

    /// An unknown family must get no flag at all. Guessing a parser is worse
    /// than leaving reasoning inline: a wrong one mis-parses real output.
    #[test]
    fn an_unknown_family_selects_no_parser() {
        for family in ["llama", "mistral-nemo", "phi3", "", "something-new"] {
            assert_eq!(
                reasoning_parser_for(family),
                None,
                "unknown family `{family}` must not be given a guessed parser"
            );
        }
    }

    #[tokio::test]
    async fn the_reasoning_parser_reaches_the_spawned_process() {
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("argv.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintf '%s\\n' \"$@\" > {}\n",
                sh.display(),
                out.display()
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let read_argv = |path: &std::path::Path| -> Vec<String> {
            std::fs::read_to_string(path)
                .unwrap()
                .lines()
                .map(str::to_string)
                .collect()
        };

        let mut child = spawn_server(&script, "org/m", 1, false, "qwen3.8").expect("spawn");
        let _ = child.wait().await;
        let argv = read_argv(&out);
        let parser = argv
            .iter()
            .position(|a| a == "--reasoning-parser")
            .and_then(|i| argv.get(i + 1));
        assert_eq!(parser.map(String::as_str), Some("qwen3"), "{argv:?}");

        let mut child = spawn_server(&script, "org/m", 1, false, "llama").expect("spawn");
        let _ = child.wait().await;
        let argv = read_argv(&out);
        assert!(
            !argv.contains(&"--reasoning-parser".to_string()),
            "an unknown family must spawn without the flag: {argv:?}"
        );
    }
}