car-inference 0.50.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 explicitly kills and waits for the child before clearing residency;
//! `kill_on_drop` remains a crash/cancellation fallback.

use std::collections::HashMap;
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{mpsc, Arc, OnceLock};
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);
/// Finite hard deadline for each freshly spawned attempt. Observable progress
/// cannot extend this bound; it only prevents the shorter stall watchdog from
/// firing while the target model is still loading.
const READY_DEADLINE: Duration = Duration::from_secs(30 * 60);
/// How long the server may make **no observable progress** before we give up.
///
/// The server writes download and load progress to its stderr log, while target
/// repo cache growth covers quiet multi-shard downloads. This bounds silence;
/// [`READY_DEADLINE`] independently bounds the whole attempt.
const STALL_TIMEOUT: Duration = Duration::from_secs(180);
/// Poll cadence while waiting for readiness.
const READY_POLL: Duration = Duration::from_millis(500);
/// Bound transient process-table races after health readiness. A managed
/// server is not published until its captured group has nonzero live RSS.
#[cfg(unix)]
const RSS_MEASURE_DEADLINE: Duration = Duration::from_secs(1);
#[cfg(unix)]
const RSS_MEASURE_POLL: Duration = Duration::from_millis(25);

#[derive(Clone, Copy)]
struct ReadinessLimits {
    health_timeout: Duration,
    ready_deadline: Duration,
    stall_timeout: Duration,
    poll: Duration,
}

impl Default for ReadinessLimits {
    fn default() -> Self {
        Self {
            health_timeout: HEALTH_TIMEOUT,
            ready_deadline: READY_DEADLINE,
            stall_timeout: STALL_TIMEOUT,
            poll: READY_POLL,
        }
    }
}

struct ManagedServer {
    /// The `vllm-mlx serve` child. `kill_on_drop` is set, so dropping this struct
    /// terminates the server.
    child: Child,
    /// Dedicated Unix process group created at spawn. Descendants retain this
    /// identity even if the leader exits and they are reparented.
    #[cfg(unix)]
    process_group_id: Option<i32>,
    port: u16,
    last_used: Instant,
    #[cfg(test)]
    teardown_delay: Duration,
    #[cfg(test)]
    inspection_error: bool,
}

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

    fn leader_has_exited(&mut self) -> std::io::Result<bool> {
        Ok(self.child.try_wait()?.is_some())
    }

    /// True while the captured process group retains any live member. A
    /// launcher may exit after forking the model-bearing server, so leader
    /// state alone is not authoritative on Unix.
    fn is_alive(&mut self) -> std::io::Result<bool> {
        #[cfg(test)]
        if self.inspection_error {
            return Err(std::io::Error::other("injected process inspection failure"));
        }
        if !self.leader_has_exited()? {
            return Ok(true);
        }
        self.process_group_is_alive()
    }

    fn requires_endpoint_health_for_reuse(&mut self) -> std::io::Result<bool> {
        #[cfg(unix)]
        {
            Ok(self.process_group_id.is_some() && self.leader_has_exited()?)
        }
        #[cfg(not(unix))]
        {
            Ok(false)
        }
    }

    #[cfg(unix)]
    fn process_group_is_alive(&self) -> std::io::Result<bool> {
        let Some(process_group_id) = self.process_group_id else {
            return Ok(false);
        };
        let result = unsafe { libc::kill(-process_group_id, 0) };
        if result == 0 {
            return Ok(true);
        }
        let error = std::io::Error::last_os_error();
        match error.raw_os_error() {
            Some(libc::ESRCH) => Ok(false),
            Some(libc::EPERM) => Ok(true),
            _ => Err(error),
        }
    }

    #[cfg(not(unix))]
    fn process_group_is_alive(&self) -> std::io::Result<bool> {
        Ok(false)
    }

    fn fully_exited(&mut self) -> std::io::Result<bool> {
        Ok(self.child.try_wait()?.is_some() && !self.process_group_is_alive()?)
    }

    fn start_kill_all(&mut self) -> std::io::Result<()> {
        #[cfg(unix)]
        if let Some(process_group_id) = self.process_group_id {
            let result = unsafe { libc::killpg(process_group_id, libc::SIGKILL) };
            if result == 0 {
                return Ok(());
            }
            let error = std::io::Error::last_os_error();
            if error.raw_os_error() == Some(libc::ESRCH) {
                return Ok(());
            }
            return Err(error);
        }
        self.child.start_kill()
    }

    /// Stop the supervised process and do not acknowledge release until the OS
    /// has reaped it. `kill_on_drop` remains the fallback if either operation
    /// itself fails or this future is cancelled.
    async fn stop(&mut self) -> Result<(), ()> {
        #[cfg(test)]
        if !self.teardown_delay.is_zero() {
            tokio::time::sleep(self.teardown_delay).await;
        }
        if self.start_kill_all().is_err() {
            return Err(());
        }
        if self.child.wait().await.is_err() {
            return Err(());
        }
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            match self.fully_exited() {
                Ok(true) => return Ok(()),
                Ok(false) if Instant::now() < deadline => {
                    tokio::time::sleep(Duration::from_millis(5)).await;
                }
                Ok(false) | Err(_) => return Err(()),
            }
        }
    }
}

struct StartingServerGuard {
    model_id: String,
    server: Option<ManagedServer>,
    admission: Option<Arc<crate::resource_policy::LocalAdmissionCoordinator>>,
}

impl StartingServerGuard {
    fn server_mut(&mut self) -> &mut ManagedServer {
        self.server.as_mut().expect("starting server owned")
    }

    fn publish(mut self) -> ManagedServer {
        self.server.take().expect("starting server owned")
    }

    /// Fully reap one stalled attempt while retaining the exact pending
    /// allocation charge for the authorized retry. If the process group cannot
    /// be confirmed gone, transfer it to quarantine and fail closed rather
    /// than spawning a replacement beside an unaccounted descendant.
    async fn reap_for_retry(mut self) -> Result<(), String> {
        // `stop` borrows the server from this guard. If this future is
        // cancelled at any await, `Drop` still owns the exact process group
        // and synchronously hands it to the quarantine reaper.
        match self.server_mut().stop().await {
            Ok(()) => {
                self.server.take();
                Ok(())
            }
            Err(()) => {
                let server = self.server.take().expect("starting server owned");
                if let Some(admission) = &self.admission {
                    enqueue_teardown(self.model_id.clone(), server, admission.clone());
                } else {
                    std::mem::forget(server);
                }
                Err(format!(
                    "stalled vllm-mlx process-group teardown is still pending for {}",
                    self.model_id
                ))
            }
        }
    }
}

impl Drop for StartingServerGuard {
    fn drop(&mut self) {
        let Some(server) = self.server.take() else {
            return;
        };
        if let Some(admission) = &self.admission {
            enqueue_teardown(self.model_id.clone(), server, admission.clone());
        }
    }
}

struct PendingServerTeardown {
    model_id: String,
    allocation_id: String,
    servers: Vec<PendingManagedServer>,
    admission: Arc<crate::resource_policy::LocalAdmissionCoordinator>,
}

struct PendingManagedServer {
    server: ManagedServer,
    kill_requested: bool,
    #[cfg(test)]
    enqueued_at: Instant,
}

struct ReaperBackoff {
    initial: Duration,
    current: Duration,
    maximum: Duration,
}

impl ReaperBackoff {
    fn new(initial: Duration, maximum: Duration) -> Self {
        Self {
            initial,
            current: initial,
            maximum,
        }
    }

    fn current(&self) -> Duration {
        self.current
    }

    fn progress(&mut self) {
        self.current = self.initial;
    }

    fn no_progress(&mut self) {
        self.current = self.current.saturating_mul(2).min(self.maximum);
    }
}

fn push_pending_teardown(
    pending: &mut Vec<PendingServerTeardown>,
    mut incoming: PendingServerTeardown,
) {
    if let Some(existing) = pending.iter_mut().find(|item| {
        item.allocation_id == incoming.allocation_id
            && Arc::ptr_eq(&item.admission, &incoming.admission)
    }) {
        existing.servers.append(&mut incoming.servers);
    } else {
        pending.push(incoming);
    }
}

fn process_teardown_sender() -> &'static mpsc::Sender<PendingServerTeardown> {
    static SENDER: OnceLock<mpsc::Sender<PendingServerTeardown>> = OnceLock::new();
    SENDER.get_or_init(|| {
        let (sender, receiver) = mpsc::channel::<PendingServerTeardown>();
        let _ = std::thread::Builder::new()
            .name("car-local-process-reaper".into())
            .spawn(move || {
                let mut pending = Vec::<PendingServerTeardown>::new();
                let mut backoff =
                    ReaperBackoff::new(Duration::from_millis(25), Duration::from_millis(250));
                loop {
                    if pending.is_empty() {
                        match receiver.recv() {
                            Ok(item) => push_pending_teardown(&mut pending, item),
                            Err(_) => break,
                        }
                    } else {
                        match receiver.recv_timeout(backoff.current()) {
                            Ok(item) => push_pending_teardown(&mut pending, item),
                            Err(mpsc::RecvTimeoutError::Timeout) => {}
                            Err(mpsc::RecvTimeoutError::Disconnected) => {}
                        }
                    }
                    while let Ok(item) = receiver.try_recv() {
                        push_pending_teardown(&mut pending, item);
                    }
                    let mut made_progress = false;
                    let mut index = 0;
                    while index < pending.len() {
                        let model_id = pending[index].model_id.clone();
                        let mut server_index = 0;
                        while server_index < pending[index].servers.len() {
                            #[cfg(test)]
                            if pending[index].servers[server_index].enqueued_at.elapsed()
                                < pending[index].servers[server_index].server.teardown_delay
                            {
                                server_index += 1;
                                continue;
                            }
                            let owned = &mut pending[index].servers[server_index];
                            // Never clear quarantine until try_wait confirms OS
                            // reaping. Inspection failures remain owned and are
                            // retried with bounded backoff.
                            match owned.server.fully_exited() {
                                Ok(true) => {
                                    pending[index].servers.swap_remove(server_index);
                                    made_progress = true;
                                }
                                Ok(false) => {
                                    if !owned.kill_requested {
                                        match owned.server.start_kill_all() {
                                            Ok(()) => owned.kill_requested = true,
                                            Err(error) => warn!(model = %model_id, %error, "cannot signal local process teardown; retaining quarantine"),
                                        }
                                    }
                                    server_index += 1;
                                }
                                Err(error) => {
                                    warn!(model = %model_id, %error, "cannot confirm local process teardown; retaining quarantine for retry");
                                    server_index += 1;
                                }
                            }
                        }
                        if pending[index].servers.is_empty() {
                            let item = pending.swap_remove(index);
                            item.admission.finish_teardown_allocation(
                                &item.model_id,
                                &item.allocation_id,
                            );
                            made_progress = true;
                        } else {
                            index += 1;
                        }
                    }
                    if made_progress {
                        backoff.progress();
                    } else {
                        backoff.no_progress();
                    }
                }
            });
        sender
    })
}

fn enqueue_teardown(
    model_id: String,
    server: ManagedServer,
    admission: Arc<crate::resource_policy::LocalAdmissionCoordinator>,
) {
    let allocation_id = crate::resource_policy::vllm_process_allocation_id(&model_id);
    admission.mark_teardown_pending_allocation(&model_id, &allocation_id);
    let pending = PendingServerTeardown {
        model_id,
        allocation_id,
        servers: vec![PendingManagedServer {
            server,
            kill_requested: false,
            #[cfg(test)]
            enqueued_at: Instant::now(),
        }],
        admission,
    };
    if let Err(error) = process_teardown_sender().send(pending) {
        // The sole reaper unexpectedly exited. Preserve child ownership and
        // its coordinator forever rather than permit unsafe peer discount.
        std::mem::forget(error.0);
    }
}

#[cfg(test)]
fn sum_process_group_rss_bytes(rows: &[(i32, u64)], process_group_id: i32) -> u64 {
    rows.iter()
        .filter(|(pgid, _)| *pgid == process_group_id)
        .map(|(_, rss_kb)| rss_kb.saturating_mul(1024))
        .fold(0, u64::saturating_add)
}

#[cfg(target_os = "macos")]
fn supervised_process_rss_bytes(server: &ManagedServer) -> Option<u64> {
    let process_group_id = server.process_group_id?;
    // SAFETY: libproc documents a null buffer/zero size as the sizing query.
    let estimated_count =
        unsafe { libc::proc_listpgrppids(process_group_id, std::ptr::null_mut(), 0) };
    if estimated_count <= 0 {
        return None;
    }
    let capacity = usize::try_from(estimated_count).ok()?.saturating_add(16);
    let mut pids = vec![0 as libc::pid_t; capacity];
    let buffer_bytes =
        i32::try_from(pids.len().checked_mul(std::mem::size_of::<libc::pid_t>())?).ok()?;
    // SAFETY: `pids` owns `buffer_bytes` writable bytes for the duration of
    // the call, and the returned count is range-checked before iteration.
    let count = unsafe {
        libc::proc_listpgrppids(
            process_group_id,
            pids.as_mut_ptr().cast::<libc::c_void>(),
            buffer_bytes,
        )
    };
    if count <= 0 || usize::try_from(count).ok()? >= pids.len() {
        return None;
    }

    let mut total = 0u64;
    for pid in pids.into_iter().take(usize::try_from(count).ok()?) {
        if pid <= 0 {
            continue;
        }
        let mut task_info = std::mem::MaybeUninit::<libc::proc_taskinfo>::zeroed();
        let task_info_bytes = i32::try_from(std::mem::size_of::<libc::proc_taskinfo>()).ok()?;
        // SAFETY: `task_info` is a correctly sized writable C buffer and is
        // assumed initialized only after libproc reports the full byte count.
        let read = unsafe {
            libc::proc_pidinfo(
                pid,
                libc::PROC_PIDTASKINFO,
                0,
                task_info.as_mut_ptr().cast::<libc::c_void>(),
                task_info_bytes,
            )
        };
        if read != task_info_bytes {
            return None;
        }
        let task_info = unsafe { task_info.assume_init() };
        total = total.saturating_add(task_info.pti_resident_size);
    }
    Some(total)
}

#[cfg(target_os = "linux")]
fn supervised_process_rss_bytes(server: &ManagedServer) -> Option<u64> {
    let process_group_id = server.process_group_id?;
    let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
    if page_size <= 0 {
        return None;
    }
    let mut total = 0u64;
    let mut found_group_member = false;
    for entry in std::fs::read_dir("/proc").ok()?.flatten() {
        let Ok(pid) = entry.file_name().to_string_lossy().parse::<i32>() else {
            continue;
        };
        let stat = match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
            Ok(stat) => stat,
            Err(_) => continue,
        };
        let Some(after_command) = stat.rsplit_once(") ").map(|(_, rest)| rest) else {
            continue;
        };
        let Some(pgrp) = after_command
            .split_whitespace()
            .nth(2)
            .and_then(|value| value.parse::<i32>().ok())
        else {
            continue;
        };
        if pgrp != process_group_id {
            continue;
        }
        found_group_member = true;
        let statm = std::fs::read_to_string(format!("/proc/{pid}/statm")).ok()?;
        let resident_pages = statm.split_whitespace().nth(1)?.parse::<u64>().ok()?;
        total = total.saturating_add(resident_pages.saturating_mul(page_size as u64));
    }
    found_group_member.then_some(total)
}

#[cfg(all(unix, not(any(target_os = "macos", target_os = "linux"))))]
fn supervised_process_rss_bytes(_server: &ManagedServer) -> Option<u64> {
    None
}

#[cfg(unix)]
async fn wait_for_live_process_group_rss_bytes(server: &mut ManagedServer) -> Result<u64, String> {
    let started = Instant::now();
    loop {
        if let Some(rss_bytes) = supervised_process_rss_bytes(server).filter(|rss| *rss > 0) {
            match server.is_alive() {
                Ok(true) => return Ok(rss_bytes),
                Ok(false) => {
                    return Err(format!(
                        "supervised vllm-mlx process group exited before RSS publication on port {}",
                        server.port
                    ));
                }
                Err(error) => {
                    return Err(format!(
                        "cannot confirm supervised vllm-mlx process group before RSS publication: {error}"
                    ));
                }
            }
        }
        match server.is_alive() {
            Ok(false) => {
                return Err(format!(
                    "supervised vllm-mlx process group exited before RSS could be measured on port {}",
                    server.port
                ));
            }
            Err(error) => {
                return Err(format!(
                    "cannot inspect supervised vllm-mlx process group for RSS publication: {error}"
                ));
            }
            Ok(true) if started.elapsed() < RSS_MEASURE_DEADLINE => {
                tokio::time::sleep(RSS_MEASURE_POLL).await;
            }
            Ok(true) => {
                return Err(format!(
                    "live supervised vllm-mlx process group had no measurable RSS within {:?} on port {}",
                    RSS_MEASURE_DEADLINE, server.port
                ));
            }
        }
    }
}

/// Type-state proof that the exact logical model about to be spawned is owned
/// by the checked reservation. It cannot be constructed outside this module.
struct AuthorizedLocalSpawn<'a> {
    _reservation: &'a crate::resource_policy::LocalLoadReservation,
}

impl<'a> AuthorizedLocalSpawn<'a> {
    fn try_new(
        reservation: &'a crate::resource_policy::LocalLoadReservation,
        model_id: &str,
    ) -> Result<Self, String> {
        if !reservation.authorizes_model(model_id) {
            return Err(format!(
                "local admission reservation does not authorize {model_id}"
            ));
        }
        Ok(Self {
            _reservation: reservation,
        })
    }
}

/// On-demand pool of supervised `vllm-mlx` servers, one per model id.
pub struct VllmServerPool {
    servers: Mutex<HashMap<String, ManagedServer>>,
    /// Per-model gate spanning engine-side reap, admission, artifact resolution,
    /// spawn, and resident publication. Pool-only singleflight is too late:
    /// callers otherwise observe the startup quarantine before reaching
    /// [`ensure`](Self::ensure) and fail instead of waiting for publication.
    dispatch_gates: std::sync::Mutex<HashMap<String, std::sync::Weak<tokio::sync::Mutex<()>>>>,
    #[cfg(test)]
    runtime_override: Option<std::path::PathBuf>,
    #[cfg(test)]
    startup_teardown_delay: Duration,
    readiness_limits: ReadinessLimits,
    idle_ttl: Duration,
    admission: Option<Arc<crate::resource_policy::LocalAdmissionCoordinator>>,
    #[cfg(test)]
    removal_race_hook: std::sync::Mutex<Option<Arc<RemovalRaceHook>>>,
}

#[cfg(test)]
struct RemovalRaceHook {
    removed: tokio::sync::Barrier,
    resume: tokio::sync::Barrier,
}

#[cfg(test)]
impl RemovalRaceHook {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            removed: tokio::sync::Barrier::new(2),
            resume: tokio::sync::Barrier::new(2),
        })
    }
}

impl Drop for VllmServerPool {
    fn drop(&mut self) {
        let servers = std::mem::take(self.servers.get_mut());
        for (model_id, server) in servers {
            if let Some(admission) = &self.admission {
                enqueue_teardown(model_id, server, admission.clone());
            }
        }
    }
}

impl VllmServerPool {
    pub fn new(idle_ttl: Duration) -> Self {
        Self {
            servers: Mutex::new(HashMap::new()),
            dispatch_gates: std::sync::Mutex::new(HashMap::new()),
            #[cfg(test)]
            runtime_override: None,
            #[cfg(test)]
            startup_teardown_delay: Duration::ZERO,
            readiness_limits: ReadinessLimits::default(),
            idle_ttl,
            admission: None,
            #[cfg(test)]
            removal_race_hook: std::sync::Mutex::new(None),
        }
    }

    pub fn with_admission(
        idle_ttl: Duration,
        admission: Arc<crate::resource_policy::LocalAdmissionCoordinator>,
    ) -> Self {
        let mut pool = Self::new(idle_ttl);
        pool.admission = Some(admission);
        pool
    }

    #[cfg(test)]
    pub(crate) fn with_test_runtime(
        idle_ttl: Duration,
        admission: Arc<crate::resource_policy::LocalAdmissionCoordinator>,
        runtime: std::path::PathBuf,
    ) -> Self {
        let mut pool = Self::with_admission(idle_ttl, admission);
        pool.runtime_override = Some(runtime);
        pool.startup_teardown_delay = Duration::from_millis(250);
        pool
    }

    /// Serialize the complete managed dispatch for one logical model. Weak
    /// entries keep this map bounded after the last waiter/owner releases.
    pub(crate) async fn acquire_dispatch(
        &self,
        model_id: &str,
    ) -> tokio::sync::OwnedMutexGuard<()> {
        let gate = {
            let mut gates = self
                .dispatch_gates
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            gates.retain(|_, gate| gate.strong_count() > 0);
            match gates.get(model_id).and_then(std::sync::Weak::upgrade) {
                Some(gate) => gate,
                None => {
                    let gate = Arc::new(tokio::sync::Mutex::new(()));
                    gates.insert(model_id.to_string(), Arc::downgrade(&gate));
                    gate
                }
            }
        };
        gate.lock_owned().await
    }

    /// A cancelled startup transfers its process and exact charge to the
    /// background reaper before releasing the dispatch gate. The next waiter
    /// may safely retry only after that teardown is acknowledged; bound the
    /// wait and fail closed if the OS never confirms exit.
    pub(crate) async fn wait_for_teardown(
        &self,
        model_id: &str,
        timeout: Duration,
    ) -> Result<(), String> {
        let Some(admission) = &self.admission else {
            return Ok(());
        };
        let deadline = Instant::now() + timeout;
        while admission.teardown_pending(model_id) {
            if Instant::now() >= deadline {
                return Err(format!(
                    "timed out waiting for supervised vllm-mlx process-group teardown for {model_id}"
                ));
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        Ok(())
    }

    #[cfg(test)]
    async fn pause_after_removal(&self) {
        let hook = self
            .removal_race_hook
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        if let Some(hook) = hook {
            hook.removed.wait().await;
            hook.resume.wait().await;
        }
    }

    /// 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,
        reservation: &crate::resource_policy::LocalLoadReservation,
        family: &str,
    ) -> Result<String, String> {
        let mut servers = self.servers.lock().await;

        if let Some(s) = servers.get_mut(model_id) {
            let alive = s.is_alive().map_err(|error| {
                format!("cannot confirm supervised vllm-mlx process state for {model_id}: {error}")
            })?;
            if alive {
                let endpoint = s.endpoint();
                let needs_health = s.requires_endpoint_health_for_reuse().map_err(|error| {
                    format!(
                        "cannot confirm supervised vllm-mlx launcher state for {model_id}: {error}"
                    )
                })?;
                if !needs_health
                    || vllm_runtime::health_ok(&endpoint, self.readiness_limits.health_timeout)
                        .await
                {
                    s.last_used = Instant::now();
                    return Ok(endpoint);
                }
                warn!(
                    model = model_id,
                    "supervised vllm-mlx launcher exited and its surviving process group is unhealthy"
                );
            }
            warn!(
                model = model_id,
                "supervised vllm-mlx server is not reusable; reconciling its process group"
            );
            let mut dead = servers.remove(model_id).expect("checked server present");
            match dead.fully_exited() {
                Ok(true) => {
                    if let Some(admission) = &self.admission {
                        admission.mark_evicted(
                            &crate::resource_policy::vllm_process_allocation_id(model_id),
                        );
                    }
                }
                Ok(false) | Err(_) => {
                    if let Some(admission) = &self.admission {
                        enqueue_teardown(model_id.to_string(), dead, admission.clone());
                    } else {
                        std::mem::forget(dead);
                    }
                    return Err(format!(
                        "supervised vllm-mlx leader exited but its process group teardown is still pending for {model_id}"
                    ));
                }
            }
        }

        // The server map and teardown quarantine form one ownership state
        // machine. Check pending only after taking the map lock and after
        // giving a live/starting singleflight owner the chance to be reused.
        if self
            .admission
            .as_ref()
            .is_some_and(|admission| admission.teardown_pending(model_id))
        {
            return Err(format!(
                "supervised vllm-mlx process-group teardown is still pending for {model_id}"
            ));
        }

        #[cfg(test)]
        let runtime_override = self.runtime_override.clone();
        #[cfg(not(test))]
        let runtime_override: Option<std::path::PathBuf> = None;
        let server_bin = match runtime_override {
            Some(runtime) => runtime,
            None => {
                vllm_runtime::ensure_runtime()
                    .await
                    .map_err(|e| format!("vllm-mlx runtime unavailable: {e}"))?
                    .server
            }
        };
        let allocation_id = crate::resource_policy::vllm_process_allocation_id(model_id);
        let mut disable_xet = false;
        loop {
            let authorization = AuthorizedLocalSpawn::try_new(reservation, model_id)?;
            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"
            );
            // LOCAL_ADMISSION_BOUNDARY:supervised-vllm-mlx
            // The engine reconciles installed weights and holds the reservation
            // before entering `ensure`; publication follows the readiness ACK.
            // Re-marking the same allocation id on the one retry is a max
            // reconciliation, not a second charge.
            if self.admission.is_some() {
                reservation.transfer_cold_weights_to_pending_allocation(
                    &allocation_id,
                    reservation.reconciled_weights_bytes(),
                );
            }
            let child = match spawn_server_authorized(
                authorization,
                &server_bin,
                runtime_model,
                port,
                disable_xet,
                family,
            ) {
                Ok(child) => child,
                Err(error) => {
                    if let Some(admission) = &self.admission {
                        admission.finish_teardown_allocation(model_id, &allocation_id);
                    }
                    return Err(format!("failed to spawn vllm-mlx serve: {error}"));
                }
            };
            let mut starting = StartingServerGuard {
                model_id: model_id.to_string(),
                server: Some(ManagedServer {
                    #[cfg(unix)]
                    process_group_id: spawned_process_group_id(&child),
                    child,
                    port,
                    last_used: Instant::now(),
                    #[cfg(test)]
                    teardown_delay: self.startup_teardown_delay,
                    #[cfg(test)]
                    inspection_error: false,
                }),
                admission: self.admission.clone(),
            };

            match wait_ready(
                &endpoint,
                starting.server_mut(),
                runtime_model,
                self.readiness_limits,
            )
            .await
            {
                Ok(()) => {
                    // Disk size can understate a persistent server's real
                    // footprint. Fold the supervised process group's live RSS
                    // into the same pending allocation before publication.
                    #[cfg(unix)]
                    if self.admission.is_some() {
                        let rss_bytes =
                            wait_for_live_process_group_rss_bytes(starting.server_mut()).await?;
                        reservation.transfer_cold_weights_to_pending_allocation(
                            &allocation_id,
                            reservation.reconciled_weights_bytes().max(rss_bytes),
                        );
                    }

                    info!(model = model_id, endpoint = %endpoint, "vllm-mlx server ready");
                    servers.insert(model_id.to_string(), starting.publish());
                    return Ok(endpoint);
                }
                Err(NotReady::Stalled(reason)) if !disable_xet => {
                    warn!(
                        model = model_id,
                        reason, "vllm-mlx startup stalled; retrying with HuggingFace Xet disabled"
                    );
                    starting.reap_for_retry().await?;
                    disable_xet = true;
                }
                Err(error) => return Err(error.into_message()),
            }
        }
    }

    /// Reap a dead server before admission is reconciled for a replacement.
    pub async fn reap_dead(&self, model_id: &str) -> Result<(), String> {
        if self
            .admission
            .as_ref()
            .is_some_and(|admission| admission.teardown_pending(model_id))
        {
            return Err(format!(
                "supervised vllm-mlx process-group teardown is still pending for {model_id}"
            ));
        }
        let mut servers = self.servers.lock().await;
        let dead = match servers.get_mut(model_id) {
            Some(server) => match server.is_alive() {
                Ok(false) => true,
                Ok(true) => match server.requires_endpoint_health_for_reuse() {
                    Ok(true) => {
                        !vllm_runtime::health_ok(
                            &server.endpoint(),
                            self.readiness_limits.health_timeout,
                        )
                        .await
                    }
                    Ok(false) => false,
                    Err(error) => {
                        warn!(model = model_id, %error, "cannot inspect supervised vllm-mlx launcher; retaining ownership and residency");
                        false
                    }
                },
                Err(error) => {
                    warn!(model = model_id, %error, "cannot inspect supervised vllm-mlx process; retaining ownership and residency");
                    false
                }
            },
            None => false,
        };
        if dead {
            let mut server = servers.remove(model_id).expect("checked server present");
            match server.fully_exited() {
                Ok(true) => {
                    if let Some(admission) = &self.admission {
                        admission.mark_evicted(
                            &crate::resource_policy::vllm_process_allocation_id(model_id),
                        );
                    }
                }
                Ok(false) | Err(_) => {
                    if let Some(admission) = &self.admission {
                        enqueue_teardown(model_id.to_string(), server, admission.clone());
                    } else {
                        std::mem::forget(server);
                    }
                    return Err(format!(
                        "supervised vllm-mlx leader exited; process-group teardown is pending for {model_id}"
                    ));
                }
            }
        }
        Ok(())
    }

    /// 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() {
            let confirmed_dead = match s.is_alive() {
                Ok(alive) => !alive,
                Err(error) => {
                    warn!(model = %k, %error, "cannot inspect supervised vllm-mlx process; retaining ownership and residency");
                    false
                }
            };
            if now.duration_since(s.last_used) > ttl || confirmed_dead {
                stale.push(k.clone());
            }
        }
        let removed = stale
            .iter()
            .filter_map(|model_id| {
                info!(model = %model_id, "evicting idle vllm-mlx server");
                servers
                    .remove(model_id)
                    .map(|server| (model_id.clone(), server))
            })
            .collect::<Vec<_>>();
        let scheduled = removed.len();
        if let Some(admission) = &self.admission {
            for (model_id, _) in &removed {
                let allocation_id = crate::resource_policy::vllm_process_allocation_id(model_id);
                admission.mark_teardown_pending_allocation(model_id, &allocation_id);
            }
        }
        #[cfg(test)]
        self.pause_after_removal().await;
        drop(servers);
        for (model_id, server) in removed {
            if let Some(admission) = &self.admission {
                enqueue_teardown(model_id, server, admission.clone());
            }
        }
        scheduled
    }

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

    pub async fn contains(&self, model_id: &str) -> bool {
        self.servers.lock().await.contains_key(model_id)
    }

    #[cfg(test)]
    pub(crate) async fn insert_test_process(&self, model_id: &str, child: Child) {
        self.insert_test_process_with_teardown_delay(model_id, child, Duration::ZERO)
            .await;
    }

    #[cfg(test)]
    async fn insert_test_process_with_teardown_delay(
        &self,
        model_id: &str,
        child: Child,
        teardown_delay: Duration,
    ) {
        self.servers.lock().await.insert(
            model_id.to_string(),
            ManagedServer {
                #[cfg(unix)]
                process_group_id: child_process_group_id(&child),
                child,
                port: 1,
                last_used: Instant::now(),
                teardown_delay,
                inspection_error: false,
            },
        );
    }

    /// Stop one supervised model process and acknowledge only after it exits.
    /// `Ok(false)` means no process was present; an error means teardown could
    /// not be confirmed and resident accounting intentionally remains held.
    pub async fn release_model_if_present(&self, model_id: &str) -> Result<bool, String> {
        let mut servers = self.servers.lock().await;
        let Some(mut server) = servers.remove(model_id) else {
            return Ok(false);
        };
        if let Some(admission) = &self.admission {
            let allocation_id = crate::resource_policy::vllm_process_allocation_id(model_id);
            admission.mark_teardown_pending_allocation(model_id, &allocation_id);
        }
        #[cfg(test)]
        self.pause_after_removal().await;
        drop(servers);
        if let Some(admission) = &self.admission {
            enqueue_teardown(model_id.to_string(), server, admission.clone());
            let wait = async {
                while admission.teardown_pending(model_id) {
                    tokio::time::sleep(Duration::from_millis(5)).await;
                }
            };
            tokio::time::timeout(Duration::from_secs(5), wait)
                .await
                .map_err(|_| {
                    format!("timed out confirming supervised vllm-mlx exit for {model_id}")
                })?;
        } else if server.stop().await.is_err() {
            return Err(format!(
                "could not confirm supervised vllm-mlx process exit for {model_id}"
            ));
        }
        Ok(true)
    }

    /// Compatibility wrapper for callers that only need present-and-released.
    pub async fn evict_model(&self, model_id: &str) -> bool {
        matches!(self.release_model_if_present(model_id).await, Ok(true))
    }
}

/// 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 command = Command::new(server_bin);
    command
        .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) {
        command.arg("--reasoning-parser").arg(parser);
    }
    if disable_xet {
        let (key, value) = XET_DISABLE_ENV;
        command.env(key, value);
    }
    #[cfg(unix)]
    command.process_group(0);
    command.spawn()
}

#[cfg(all(unix, test))]
fn child_process_group_id(child: &Child) -> Option<i32> {
    let pid = i32::try_from(child.id()?).ok()?;
    let process_group_id = unsafe { libc::getpgid(pid) };
    (process_group_id == pid).then_some(process_group_id)
}

/// `process_group(0)` makes the spawned child's PID its PGID by contract. Use
/// the captured PID directly rather than querying `getpgid`: a launcher may
/// fork the model worker and exit before CAR inspects it, at which point the
/// leader no longer exists but its descendants retain this PGID.
#[cfg(unix)]
fn spawned_process_group_id(child: &Child) -> Option<i32> {
    i32::try_from(child.id()?).ok()
}

fn spawn_server_authorized(
    _authorization: AuthorizedLocalSpawn<'_>,
    server_bin: &Path,
    runtime_model: &str,
    port: u16,
    disable_xet: bool,
    family: &str,
) -> std::io::Result<Child> {
    // LOCAL_ADMISSION_BOUNDARY:supervised-vllm-mlx-raw-spawn
    spawn_server(server_bin, runtime_model, port, disable_xet, family)
}

/// 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.
///
/// Every attempt has a hard total deadline. The target model's stderr and repo
/// cache additionally provide a no-progress deadline; unrelated global cache
/// activity is deliberately excluded.
async fn wait_ready(
    endpoint: &str,
    server: &mut ManagedServer,
    runtime_model: &str,
    limits: ReadinessLimits,
) -> Result<(), NotReady> {
    let start = Instant::now();
    let log = log_path(server.port, "stderr");
    let cache = crate::registry::huggingface_repo_dir(runtime_model);
    let mut last_mark = progress_mark(log.as_deref(), &cache);
    let mut last_progress = Instant::now();

    loop {
        if vllm_runtime::health_ok(endpoint, limits.health_timeout).await {
            return Ok(());
        }
        match server.is_alive() {
            Ok(false) => {
                return Err(NotReady::Other(format!(
                    "vllm-mlx server exited during startup (see ~/.car/logs/vllm-mlx-{}.stderr.log)",
                    server.port
                )));
            }
            Err(error) => {
                return Err(NotReady::Other(format!(
                    "cannot inspect vllm-mlx server during startup: {error}"
                )));
            }
            Ok(true) => {}
        }

        if let Some(mark) = progress_mark(log.as_deref(), &cache) {
            // Either the log or the weight cache growing means real work.
            if Some(mark) != last_mark {
                last_mark = Some(mark);
                last_progress = Instant::now();
            }
        }
        if let Some(error) = readiness_timeout(
            start.elapsed(),
            last_progress.elapsed(),
            endpoint,
            server.port,
            limits,
        ) {
            return Err(error);
        }
        tokio::time::sleep(limits.poll).await;
    }
}

fn readiness_timeout(
    elapsed: Duration,
    since_progress: Duration,
    endpoint: &str,
    port: u16,
    limits: ReadinessLimits,
) -> Option<NotReady> {
    if elapsed > limits.ready_deadline {
        return Some(NotReady::Stalled(format!(
            "vllm-mlx server exceeded the hard startup deadline of {:?} and never became healthy at {endpoint}",
            limits.ready_deadline
        )));
    }
    (since_progress > limits.stall_timeout).then(|| {
        NotReady::Stalled(format!(
            "vllm-mlx server made no progress for {}s and never became healthy at \
             {endpoint} (see ~/.car/logs/vllm-mlx-{port}.stderr.log)",
            limits.stall_timeout.as_secs(),
        ))
    })
}

/// 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:
/// Target-process stderr bytes written, plus bytes of target-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.
///
/// The global Xet cache is intentionally excluded because it cannot attribute
/// activity to this model; a different download must not extend this attempt.
fn progress_mark(log: Option<&Path>, cache_dir: &Path) -> Option<u64> {
    let log_bytes = log.and_then(file_len);
    let cache_bytes = dir_bytes(cache_dir);
    match (log_bytes, cache_bytes) {
        (None, None) => None,
        (a, b) => Some(a.unwrap_or(0).saturating_add(b.unwrap_or(0))),
    }
}

/// 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 process_reaper_backoff_is_bounded_and_resets_after_progress() {
        let mut backoff = ReaperBackoff::new(Duration::from_millis(25), Duration::from_millis(250));
        assert_eq!(backoff.current(), Duration::from_millis(25));
        for _ in 0..10 {
            backoff.no_progress();
        }
        assert_eq!(backoff.current(), Duration::from_millis(250));
        backoff.progress();
        assert_eq!(backoff.current(), Duration::from_millis(25));
    }

    #[test]
    fn supervised_process_rss_includes_the_captured_group_only() {
        let rows = vec![(10, 100), (10, 200), (10, 300), (20, 900)];
        assert_eq!(sum_process_group_rss_bytes(&rows, 10), 600 * 1024);
    }

    #[test]
    fn alloc_port_returns_distinct_usable_ports() {
        if !crate::run_in_isolated_test_process(
            "vllm_pool::tests::alloc_port_returns_distinct_usable_ports",
            "CAR_VLLM_PORT_ALLOCATION_CHILD",
        ) {
            return;
        }
        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);
    }

    #[tokio::test]
    async fn targeted_evict_waits_for_process_exit_before_clearing_residency() {
        const CHILD_ENV: &str = "CAR_VLLM_EVICTION_TEST_CHILD";
        if std::env::var_os(CHILD_ENV).is_some() {
            tokio::time::sleep(Duration::from_secs(60)).await;
            return;
        }

        let child = Command::new(std::env::current_exe().expect("current test executable"))
            .arg("--exact")
            .arg(
                "vllm_pool::tests::targeted_evict_waits_for_process_exit_before_clearing_residency",
            )
            .env(CHILD_ENV, "1")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
            .expect("spawn stand-in supervised process");
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        coordinator.mark_resident_allocation(
            "vllm-mlx/test",
            &crate::resource_policy::vllm_process_allocation_id("vllm-mlx/test"),
            1,
        );
        let pool = VllmServerPool::with_admission(Duration::from_secs(300), coordinator.clone());
        pool.servers.lock().await.insert(
            "vllm-mlx/test".into(),
            ManagedServer {
                #[cfg(unix)]
                process_group_id: child_process_group_id(&child),
                child,
                port: 1,
                last_used: Instant::now(),
                teardown_delay: Duration::ZERO,
                inspection_error: false,
            },
        );

        assert!(pool.evict_model("vllm-mlx/test").await);
        assert!(pool.is_empty().await);
        assert!(!coordinator.is_resident("vllm-mlx/test"));
    }

    #[tokio::test]
    async fn process_inspection_error_retains_vllm_owner_and_residency() {
        const CHILD_ENV: &str = "CAR_VLLM_INSPECTION_ERROR_CHILD";
        if std::env::var_os(CHILD_ENV).is_some() {
            tokio::time::sleep(Duration::from_secs(60)).await;
            return;
        }
        let child = Command::new(std::env::current_exe().unwrap())
            .arg("--exact")
            .arg("vllm_pool::tests::process_inspection_error_retains_vllm_owner_and_residency")
            .env(CHILD_ENV, "1")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
            .unwrap();
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        coordinator.mark_resident_allocation(
            "vllm-mlx/inspect",
            &crate::resource_policy::vllm_process_allocation_id("vllm-mlx/inspect"),
            1,
        );
        let pool = VllmServerPool::with_admission(Duration::from_secs(300), coordinator.clone());
        pool.insert_test_process("vllm-mlx/inspect", child).await;
        pool.servers
            .lock()
            .await
            .get_mut("vllm-mlx/inspect")
            .unwrap()
            .inspection_error = true;

        pool.reap_dead("vllm-mlx/inspect").await.unwrap();
        assert!(pool.contains("vllm-mlx/inspect").await);
        assert!(coordinator.is_resident("vllm-mlx/inspect"));
    }

    #[tokio::test]
    async fn dropping_last_pool_reaps_children_before_clearing_residency() {
        const CHILD_ENV: &str = "CAR_VLLM_DROP_TEST_CHILD";
        if std::env::var_os(CHILD_ENV).is_some() {
            tokio::time::sleep(Duration::from_secs(60)).await;
            return;
        }

        let child = Command::new(std::env::current_exe().expect("current test executable"))
            .arg("--exact")
            .arg("vllm_pool::tests::dropping_last_pool_reaps_children_before_clearing_residency")
            .env(CHILD_ENV, "1")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
            .expect("spawn stand-in supervised process");
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        coordinator.mark_resident_allocation(
            "vllm-mlx/drop-test",
            &crate::resource_policy::vllm_process_allocation_id("vllm-mlx/drop-test"),
            1,
        );
        let pool = VllmServerPool::with_admission(Duration::from_secs(300), coordinator.clone());
        pool.insert_test_process_with_teardown_delay(
            "vllm-mlx/drop-test",
            child,
            Duration::from_millis(250),
        )
        .await;

        let started = Instant::now();
        drop(pool);

        assert!(started.elapsed() < Duration::from_millis(100));
        assert!(coordinator.teardown_pending("vllm-mlx/drop-test"));
        assert!(
            coordinator.is_resident("vllm-mlx/drop-test"),
            "residency stays charged until the background reaper confirms exit"
        );
        let blocked = coordinator
            .reserve_measured_host("vllm-mlx/drop-test", 1024 * 1024, 0)
            .unwrap_err();
        assert_eq!(
            blocked.preflight.verdict,
            crate::resource_policy::LocalLoadVerdict::PendingTeardown
        );
        tokio::time::timeout(Duration::from_secs(2), async {
            while coordinator.teardown_pending("vllm-mlx/drop-test") {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("background reaper should eventually confirm exit");
        assert!(!coordinator.is_resident("vllm-mlx/drop-test"));
    }

    #[tokio::test]
    async fn cancelled_targeted_release_transfers_to_reaper_until_confirmed_exit() {
        const CHILD_ENV: &str = "CAR_VLLM_CANCELLED_RELEASE_CHILD";
        if std::env::var_os(CHILD_ENV).is_some() {
            tokio::time::sleep(Duration::from_secs(60)).await;
            return;
        }
        let child = Command::new(std::env::current_exe().unwrap())
            .arg("--exact")
            .arg("vllm_pool::tests::cancelled_targeted_release_transfers_to_reaper_until_confirmed_exit")
            .env(CHILD_ENV, "1")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
            .unwrap();
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        coordinator.mark_resident_allocation(
            "vllm-mlx/cancel",
            &crate::resource_policy::vllm_process_allocation_id("vllm-mlx/cancel"),
            1,
        );
        let pool = Arc::new(VllmServerPool::with_admission(
            Duration::from_secs(300),
            coordinator.clone(),
        ));
        pool.insert_test_process_with_teardown_delay(
            "vllm-mlx/cancel",
            child,
            Duration::from_millis(250),
        )
        .await;

        let owner = pool.clone();
        let release =
            tokio::spawn(async move { owner.release_model_if_present("vllm-mlx/cancel").await });
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(coordinator.teardown_pending("vllm-mlx/cancel"));
        release.abort();
        let _ = release.await;
        assert!(coordinator.is_resident("vllm-mlx/cancel"));
        tokio::time::timeout(Duration::from_secs(2), async {
            while coordinator.teardown_pending("vllm-mlx/cancel") {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        assert!(!coordinator.is_resident("vllm-mlx/cancel"));
    }

    #[cfg(unix)]
    async fn assert_removal_never_exposes_unquarantined_empty_slot(targeted: bool) {
        let model_id = if targeted {
            "vllm-mlx/release-race"
        } else {
            "vllm-mlx/evict-race"
        };
        let mut command = Command::new("sh");
        command
            .arg("-c")
            .arg("sleep 60")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().unwrap();
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        coordinator.mark_resident_allocation(
            model_id,
            &crate::resource_policy::vllm_process_allocation_id(model_id),
            1,
        );
        let pool = Arc::new(VllmServerPool::with_admission(
            if targeted {
                Duration::from_secs(300)
            } else {
                Duration::ZERO
            },
            coordinator.clone(),
        ));
        pool.insert_test_process(model_id, child).await;
        let hook = RemovalRaceHook::new();
        *pool
            .removal_race_hook
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(hook.clone());

        let owner = pool.clone();
        let removal = tokio::spawn(async move {
            if targeted {
                owner.release_model_if_present(model_id).await.map(|_| ())
            } else {
                owner.evict_idle().await;
                Ok(())
            }
        });
        hook.removed.wait().await;
        let pending_was_visible = coordinator.teardown_pending(model_id);
        let empty_slot_was_visible = pool.servers.try_lock().is_ok();
        hook.resume.wait().await;
        removal.await.unwrap().unwrap();

        assert!(
            pending_was_visible,
            "removal must publish exact teardown quarantine before an empty slot is visible"
        );
        assert!(
            !empty_slot_was_visible,
            "ensure must not acquire an empty server map between removal and quarantine"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn targeted_release_serializes_empty_slot_and_pending_transition_against_ensure() {
        assert_removal_never_exposes_unquarantined_empty_slot(true).await;
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn idle_eviction_serializes_empty_slot_and_pending_transition_against_ensure() {
        assert_removal_never_exposes_unquarantined_empty_slot(false).await;
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn concurrent_same_model_ensure_reuses_live_singleflight_owner_while_publication_pending()
    {
        let model_id = "vllm-mlx/concurrent-start";
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .unwrap();
        coordinator.mark_teardown_pending_allocation_with_charge(
            model_id,
            &crate::resource_policy::vllm_process_allocation_id(model_id),
            1024 * 1024,
        );
        let mut command = Command::new("sh");
        command
            .arg("-c")
            .arg("sleep 60")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().unwrap();
        let pool = VllmServerPool::with_admission(Duration::from_secs(300), coordinator.clone());
        pool.insert_test_process(model_id, child).await;

        let endpoint = pool
            .ensure(model_id, "already-started", &reservation, "qwen3")
            .await
            .expect("the second waiter must reuse the live singleflight owner");
        assert_eq!(endpoint, "http://127.0.0.1:1");

        assert!(pool.release_model_if_present(model_id).await.unwrap());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn targeted_evict_terminates_model_bearing_process_group_descendants() {
        let fixture = tempfile::tempdir().unwrap();
        let descendant_pid_file = fixture.path().join("descendant.pid");
        let mut command = Command::new("sh");
        command
            .arg("-c")
            .arg("sleep 60 & echo $! > \"$1\"; wait")
            .arg("car-vllm-process-group-test")
            .arg(&descendant_pid_file)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().unwrap();
        tokio::time::timeout(Duration::from_secs(2), async {
            while !descendant_pid_file.exists() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("stand-in server must spawn its model-bearing descendant");
        let descendant_pid: i32 = std::fs::read_to_string(&descendant_pid_file)
            .unwrap()
            .trim()
            .parse()
            .unwrap();

        let pool = VllmServerPool::new(Duration::from_secs(300));
        pool.insert_test_process("vllm-mlx/process-tree", child)
            .await;
        assert!(pool
            .release_model_if_present("vllm-mlx/process-tree")
            .await
            .unwrap());

        let descendant_survived = unsafe { libc::kill(descendant_pid, 0) } == 0;
        if descendant_survived {
            unsafe {
                libc::kill(descendant_pid, libc::SIGKILL);
            }
        }
        assert!(
            !descendant_survived,
            "release must not acknowledge while a model-bearing descendant survives"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn launcher_exit_cannot_erase_spawn_time_process_group_identity() {
        let fixture = tempfile::tempdir().unwrap();
        let descendant_pid_file = fixture.path().join("descendant.pid");
        let mut command = Command::new("sh");
        command
            .arg("-c")
            .arg("sleep 60 & echo $! > \"$1\"; exit 0")
            .arg("car-vllm-launcher-exit-test")
            .arg(&descendant_pid_file)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let mut child = command.spawn().unwrap();
        let leader_pid = i32::try_from(child.id().unwrap()).unwrap();
        let recorded_process_group = spawned_process_group_id(&child);
        assert_eq!(recorded_process_group, Some(leader_pid));
        tokio::time::timeout(Duration::from_secs(2), async {
            while !descendant_pid_file.exists() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(child.try_wait().unwrap().is_some(), "launcher must be gone");
        assert_eq!(
            unsafe { libc::getpgid(leader_pid) },
            -1,
            "OS lookup reproduces the old ESRCH race"
        );

        let descendant_pid: i32 = std::fs::read_to_string(&descendant_pid_file)
            .unwrap()
            .trim()
            .parse()
            .unwrap();
        let mut server = ManagedServer {
            child,
            process_group_id: recorded_process_group,
            port: 1,
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        server.start_kill_all().unwrap();
        tokio::time::timeout(Duration::from_secs(2), async {
            while server.process_group_is_alive().unwrap() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("captured spawn-time PGID must still terminate descendants");
        assert_ne!(unsafe { libc::kill(descendant_pid, 0) }, 0);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn launcher_exit_preserves_readiness_rss_publication_and_healthy_reuse() {
        const CHILD_ENV: &str = "CAR_VLLM_LEADERLESS_HEALTHY_CHILD";
        if !crate::run_in_isolated_test_process(
            "vllm_pool::tests::launcher_exit_preserves_readiness_rss_publication_and_healthy_reuse",
            CHILD_ENV,
        ) {
            return;
        }
        let Some(python) = vllm_runtime::which("python3") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        unsafe { std::env::set_var(car_home::ENV_VAR, dir.path().join("car-home")) };
        let attempts = dir.path().join("attempts.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\n\
                 import os,sys,time,http.server\n\
                 with open({:?}, 'a') as f: f.write('launch\\n')\n\
                 if os.fork() != 0: sys.exit(0)\n\
                 time.sleep(0.2)\n\
                 p=int(sys.argv[sys.argv.index('--port')+1])\n\
                 H=type('H',(http.server.BaseHTTPRequestHandler,),{{'do_GET':lambda s:(s.send_response(200),s.end_headers()),'log_message':lambda *a:None}})\n\
                 http.server.HTTPServer(('127.0.0.1',p),H).serve_forever()\n",
                python.display(),
                attempts,
            ),
        )
        .unwrap();
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();

        let model_id = "vllm-mlx/leaderless-healthy";
        let allocation_id = crate::resource_policy::vllm_process_allocation_id(model_id);
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let mut reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 64)
            .unwrap();
        let mut pool = VllmServerPool::with_test_runtime(
            Duration::from_secs(300),
            coordinator.clone(),
            script,
        );
        pool.readiness_limits = ReadinessLimits {
            health_timeout: Duration::from_millis(25),
            ready_deadline: Duration::from_secs(3),
            stall_timeout: Duration::from_secs(3),
            poll: Duration::from_millis(10),
        };

        // Process ownership/accounting cannot depend on a shell utility being
        // available or permitted. This also isolates the fixture from sibling
        // tests that deliberately replace PATH while probing external tools.
        let empty_path = dir.path().join("empty-path");
        std::fs::create_dir_all(&empty_path).unwrap();
        unsafe { std::env::set_var("PATH", empty_path) };

        let endpoint = pool
            .ensure(model_id, "fixture/model", &reservation, "qwen3")
            .await
            .expect("a healthy descendant must survive launcher exit through readiness");
        let pending_rss_mb = coordinator.resident_model_mb();
        assert!(
            pending_rss_mb > 1,
            "RSS publication must measure the live process group, not the exited 1 MB launcher"
        );
        reservation.publish_resident_weights_as(&allocation_id, 1024 * 1024);
        assert_eq!(coordinator.resident_model_mb(), pending_rss_mb);

        let reused = pool
            .ensure(model_id, "fixture/model", &reservation, "qwen3")
            .await
            .expect("a healthy leaderless process group must be reused");
        assert_eq!(reused, endpoint);
        assert_eq!(
            std::fs::read_to_string(&attempts).unwrap().lines().count(),
            1,
            "reuse must not spawn a second process group"
        );

        assert!(pool.release_model_if_present(model_id).await.unwrap());
        assert!(!coordinator.is_resident(model_id));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn dead_leader_with_live_descendant_stays_quarantined_until_group_exit() {
        let fixture = tempfile::tempdir().unwrap();
        let descendant_pid_file = fixture.path().join("descendant.pid");
        let mut command = Command::new("sh");
        command
            .arg("-c")
            .arg("sleep 60 & echo $! > \"$1\"; exit 0")
            .arg("car-vllm-dead-leader-test")
            .arg(&descendant_pid_file)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let mut child = command.spawn().unwrap();
        let process_group_id = i32::try_from(child.id().unwrap()).unwrap();
        tokio::time::timeout(Duration::from_secs(2), async {
            while !descendant_pid_file.exists() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        let descendant_pid: i32 = std::fs::read_to_string(&descendant_pid_file)
            .unwrap()
            .trim()
            .parse()
            .unwrap();
        child.wait().await.unwrap();

        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let model_id = "vllm-mlx/dead-leader";
        coordinator.mark_resident_allocation(
            model_id,
            &crate::resource_policy::vllm_process_allocation_id(model_id),
            1,
        );
        let warm_reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .expect("pre-reap reservation sees the old resident generation");
        let pool = VllmServerPool::with_admission(Duration::from_secs(300), coordinator.clone());
        pool.servers.lock().await.insert(
            model_id.into(),
            ManagedServer {
                child,
                process_group_id: Some(process_group_id),
                port: 1,
                last_used: Instant::now(),
                teardown_delay: Duration::from_millis(250),
                inspection_error: false,
            },
        );

        let reap_error = pool.reap_dead(model_id).await.unwrap_err();
        assert!(reap_error.contains("teardown is pending"));
        assert!(!pool.contains(model_id).await);
        assert!(coordinator.teardown_pending(model_id));
        assert!(
            coordinator.is_resident(model_id),
            "leader death must not erase accounting while a descendant retains the group"
        );
        assert_eq!(unsafe { libc::kill(descendant_pid, 0) }, 0);
        let replacement_error = pool
            .ensure(model_id, "must-not-spawn", &warm_reservation, "qwen3")
            .await
            .unwrap_err();
        assert!(replacement_error.contains("teardown is still pending"));

        tokio::time::timeout(Duration::from_secs(2), async {
            while coordinator.teardown_pending(model_id) {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("reaper must terminate and confirm the complete process group");
        assert!(!coordinator.is_resident(model_id));
        assert_ne!(unsafe { libc::kill(descendant_pid, 0) }, 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 {
            #[cfg(unix)]
            process_group_id: spawned_process_group_id(&child),
            child,
            port,
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        let endpoint = server.endpoint();

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

        // 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::*;

    /// A finite hard deadline is the last-resort bound, not a replacement for
    /// the no-progress watchdog. It must leave materially more time for a
    /// genuinely progressing first-run download than a silent process gets.
    #[test]
    fn default_hard_deadline_is_materially_longer_than_the_stall_timeout() {
        assert!(
            STALL_TIMEOUT >= Duration::from_secs(120),
            "a stall bound shorter than a slow model-load step would reintroduce \
             spurious startup failures"
        );
        assert!(
            READY_DEADLINE.as_secs() >= STALL_TIMEOUT.as_secs().saturating_mul(4),
            "the hard deadline must be materially longer than the no-progress watchdog: \
             ready={READY_DEADLINE:?} stall={STALL_TIMEOUT:?}"
        );
    }

    #[test]
    fn recent_target_progress_can_outlive_stall_but_not_the_hard_deadline() {
        let limits = ReadinessLimits {
            health_timeout: Duration::from_millis(1),
            ready_deadline: Duration::from_secs(10),
            stall_timeout: Duration::from_secs(2),
            poll: Duration::from_millis(1),
        };
        assert!(
            readiness_timeout(
                Duration::from_secs(6),
                Duration::from_millis(10),
                "http://127.0.0.1:1",
                1,
                limits,
            )
            .is_none(),
            "recent target progress must permit startup beyond one stall interval"
        );
        assert!(readiness_timeout(
            Duration::from_secs(6),
            Duration::from_secs(3),
            "http://127.0.0.1:1",
            1,
            limits,
        )
        .unwrap()
        .into_message()
        .contains("no progress"));
        assert!(
            readiness_timeout(
                Duration::from_secs(11),
                Duration::ZERO,
                "http://127.0.0.1:1",
                1,
                limits,
            )
            .unwrap()
            .into_message()
            .contains("hard startup deadline"),
            "continuous progress must never extend the finite hard bound"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn absent_log_and_cache_evidence_still_hits_the_no_progress_timeout() {
        const TEST_ENV: &str = "CAR_VLLM_NO_OBSERVABLE_PROGRESS_TEST";
        const SERVER_ENV: &str = "CAR_VLLM_NO_OBSERVABLE_PROGRESS_SERVER";
        if std::env::var_os(SERVER_ENV).is_some() {
            tokio::time::sleep(Duration::from_secs(60)).await;
            return;
        }
        if !crate::run_in_isolated_test_process(
            "vllm_pool::readiness_tests::absent_log_and_cache_evidence_still_hits_the_no_progress_timeout",
            TEST_ENV,
        ) {
            return;
        }

        let child = Command::new(std::env::current_exe().unwrap())
            .arg("--exact")
            .arg("vllm_pool::readiness_tests::absent_log_and_cache_evidence_still_hits_the_no_progress_timeout")
            .env(SERVER_ENV, "1")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0)
            .spawn()
            .unwrap();
        let mut server = ManagedServer {
            process_group_id: spawned_process_group_id(&child),
            child,
            port: alloc_loopback_port().unwrap(),
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        let endpoint = server.endpoint();
        let limits = ReadinessLimits {
            health_timeout: Duration::from_millis(5),
            ready_deadline: Duration::from_millis(750),
            stall_timeout: Duration::from_millis(100),
            poll: Duration::from_millis(5),
        };
        let started = Instant::now();
        let error = wait_ready(
            &endpoint,
            &mut server,
            "fixture/no-observable-progress-evidence",
            limits,
        )
        .await
        .unwrap_err()
        .into_message();

        assert!(error.contains("no progress"), "{error}");
        assert!(
            started.elapsed() < limits.ready_deadline,
            "absence of evidence must hit the stall bound before the hard deadline"
        );
        let _ = server.start_kill_all();
        let _ = server.child.wait().await;
    }

    #[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 before = progress_mark(Some(&log), &cache).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).expect("observable");

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

    #[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");
        assert_eq!(
            progress_mark(Some(&missing_log), &missing_cache),
            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).is_some(),
            "an existing cache dir is observable even before any bytes arrive"
        );
    }

    #[test]
    fn unrelated_global_xet_activity_is_not_target_model_progress() {
        let dir = tempfile::tempdir().unwrap();
        let target_cache = dir.path().join("target-model-cache");
        let global_xet = dir.path().join("xet");
        std::fs::create_dir_all(global_xet.join("unrelated-repo").join("staging")).unwrap();

        assert_eq!(
            progress_mark(None, &target_cache),
            None,
            "global Xet traffic for another model must not keep this startup alive"
        );
    }

    #[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::*;

    fn fast_readiness_limits() -> ReadinessLimits {
        ReadinessLimits {
            health_timeout: Duration::from_millis(10),
            ready_deadline: Duration::from_secs(3),
            // The isolated fixture still competes with the parent suite for
            // CPU. A one-second startup window keeps the retry proof bounded
            // without killing the shell before it can record the attempt.
            stall_timeout: Duration::from_secs(1),
            poll: Duration::from_millis(5),
        }
    }

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

    /// The production break: a wedged Xet download returned the first stall
    /// without trying the known-good HTTPS path. This drives the real process,
    /// readiness watcher, process-group reap, and second spawn; the attempt log
    /// proves there is exactly one fallback and that the family parser survives
    /// the admission-authorized path.
    #[tokio::test]
    async fn xet_stall_retries_exactly_once_then_publishes_the_healthy_owner() {
        const CHILD_ENV: &str = "CAR_VLLM_XET_RETRY_SUCCESS_CHILD";
        if !crate::run_in_isolated_test_process(
            "vllm_pool::xet_fallback_tests::xet_stall_retries_exactly_once_then_publishes_the_healthy_owner",
            CHILD_ENV,
        ) {
            return;
        }
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let Some(python) = vllm_runtime::which("python3") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        unsafe { std::env::set_var(car_home::ENV_VAR, dir.path().join("car-home")) };
        let attempts = dir.path().join("attempts.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintf '%s|%s\\n' \"${{HF_HUB_DISABLE_XET:-UNSET}}\" \"$*\" >> '{}'\n\
                 if [ \"${{HF_HUB_DISABLE_XET:-}}\" != \"1\" ]; then exec sleep 60; fi\n\
                 exec '{}' -c \"import sys,http.server; p=int(sys.argv[sys.argv.index('--port')+1]); H=type('H',(http.server.BaseHTTPRequestHandler,),{{'do_GET':lambda s:(s.send_response(200),s.end_headers()),'log_message':lambda *a:None}}); http.server.HTTPServer(('127.0.0.1',p),H).serve_forever()\" \"$@\"\n",
                sh.display(),
                attempts.display(),
                python.display(),
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let model_id = "vllm-mlx/xet-retry-success";
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let mut reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .unwrap();
        let mut pool = VllmServerPool::with_test_runtime(
            Duration::from_secs(300),
            coordinator.clone(),
            script,
        );
        pool.startup_teardown_delay = Duration::ZERO;
        pool.readiness_limits = fast_readiness_limits();

        let endpoint = tokio::time::timeout(
            Duration::from_secs(5),
            pool.ensure(model_id, "fixture/model", &reservation, "qwen3.8"),
        )
        .await
        .expect("startup retry must remain bounded")
        .expect("the Xet-disabled retry should become healthy");
        reservation.publish_resident_weights_as(
            &crate::resource_policy::vllm_process_allocation_id(model_id),
            1024 * 1024,
        );

        let lines: Vec<String> = std::fs::read_to_string(&attempts)
            .unwrap()
            .lines()
            .map(str::to_string)
            .collect();
        assert_eq!(
            lines.len(),
            2,
            "one normal attempt plus one fallback: {lines:?}"
        );
        assert!(lines[0].starts_with("UNSET|"), "{lines:?}");
        assert!(lines[1].starts_with("1|"), "{lines:?}");
        assert!(
            lines[1].contains("--reasoning-parser qwen3"),
            "family parser was lost between admission and retry: {lines:?}"
        );
        assert!(vllm_runtime::health_ok(&endpoint, Duration::from_secs(1)).await);
        assert!(pool.release_model_if_present(model_id).await.unwrap());
    }

    /// The production break: an unbounded fallback loop could keep spawning
    /// model-bearing children forever, or return while the last failed owner
    /// was still charged but not quarantined. Two real stalls must return an
    /// error after exactly two spawns and leave no process/residency behind.
    #[tokio::test]
    async fn continued_stall_after_retry_is_bounded_and_cleans_up_accounting() {
        const CHILD_ENV: &str = "CAR_VLLM_XET_RETRY_FAILURE_CHILD";
        if !crate::run_in_isolated_test_process(
            "vllm_pool::xet_fallback_tests::continued_stall_after_retry_is_bounded_and_cleans_up_accounting",
            CHILD_ENV,
        ) {
            return;
        }
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        unsafe { std::env::set_var(car_home::ENV_VAR, dir.path().join("car-home")) };
        let attempts = dir.path().join("attempts.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintf '%s\\n' \"${{HF_HUB_DISABLE_XET:-UNSET}}\" >> '{}'\nexec sleep 60\n",
                sh.display(),
                attempts.display(),
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let model_id = "vllm-mlx/xet-retry-failure";
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .unwrap();
        let mut pool = VllmServerPool::with_test_runtime(
            Duration::from_secs(300),
            coordinator.clone(),
            script,
        );
        pool.startup_teardown_delay = Duration::ZERO;
        pool.readiness_limits = fast_readiness_limits();

        let error = tokio::time::timeout(
            Duration::from_secs(5),
            pool.ensure(model_id, "fixture/model", &reservation, "qwen3"),
        )
        .await
        .expect("two failed attempts must remain bounded")
        .unwrap_err();
        assert!(error.contains("no progress") || error.contains("never became healthy"));
        let lines: Vec<String> = std::fs::read_to_string(&attempts)
            .unwrap()
            .lines()
            .map(str::to_string)
            .collect();
        assert_eq!(lines, ["UNSET", "1"], "the fallback must not loop");

        pool.wait_for_teardown(model_id, Duration::from_secs(2))
            .await
            .expect("failed startup owner must be fully reaped");
        assert_eq!(pool.len().await, 0);
        assert!(!coordinator.is_resident(model_id));
        drop(reservation);
        assert_eq!(coordinator.active_request_count(model_id), 0);
    }

    /// A process can emit stderr forever without ever binding its health
    /// endpoint. Progress extends the no-progress watchdog, but must not extend
    /// the finite per-attempt startup deadline.
    #[tokio::test]
    async fn continuously_growing_stderr_hits_both_hard_deadlines_and_cleans_up() {
        const CHILD_ENV: &str = "CAR_VLLM_GROWING_STDERR_FAILURE_CHILD";
        if !crate::run_in_isolated_test_process(
            "vllm_pool::xet_fallback_tests::continuously_growing_stderr_hits_both_hard_deadlines_and_cleans_up",
            CHILD_ENV,
        ) {
            return;
        }
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        unsafe { std::env::set_var(car_home::ENV_VAR, dir.path().join("car-home")) };
        let attempts = dir.path().join("attempts.txt");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintf '%s\\n' \"${{HF_HUB_DISABLE_XET:-UNSET}}\" >> '{}'\nwhile true; do printf progress >&2; sleep 0.01; done\n",
                sh.display(),
                attempts.display(),
            ),
        )
        .unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let model_id = "vllm-mlx/growing-stderr-failure";
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .unwrap();
        let mut pool = VllmServerPool::with_test_runtime(
            Duration::from_secs(300),
            coordinator.clone(),
            script,
        );
        pool.startup_teardown_delay = Duration::ZERO;
        pool.readiness_limits = ReadinessLimits {
            health_timeout: Duration::from_millis(10),
            // Leave enough scheduling headroom for both fixture processes to
            // record their attempt even on a loaded CI worker. Their growing
            // logs keep the shorter stall watchdog refreshed until this hard
            // bound fires.
            ready_deadline: Duration::from_secs(2),
            stall_timeout: Duration::from_secs(1),
            poll: Duration::from_millis(5),
        };

        let error = tokio::time::timeout(
            Duration::from_secs(6),
            pool.ensure(model_id, "fixture/model", &reservation, "qwen3"),
        )
        .await
        .expect("both finite startup deadlines must beat the outer bound")
        .unwrap_err();
        assert!(error.contains("startup deadline"), "{error}");
        assert_eq!(
            std::fs::read_to_string(&attempts)
                .unwrap()
                .lines()
                .collect::<Vec<_>>(),
            ["UNSET", "1"]
        );

        pool.wait_for_teardown(model_id, Duration::from_secs(2))
            .await
            .expect("failed startup owner must be fully reaped");
        assert_eq!(pool.len().await, 0);
        assert!(!coordinator.is_resident(model_id));
        drop(reservation);
        assert_eq!(coordinator.active_request_count(model_id), 0);
    }

    /// Cancelling while the first stalled process group is being reaped must
    /// transfer that exact owner to quarantine. The pending allocation may
    /// clear only after the background reaper confirms the whole group exited.
    #[cfg(unix)]
    #[tokio::test]
    async fn cancelling_during_retry_reap_retains_group_and_charge_until_exit_ack() {
        const CHILD_ENV: &str = "CAR_VLLM_CANCEL_RETRY_REAP_CHILD";
        if !crate::run_in_isolated_test_process(
            "vllm_pool::xet_fallback_tests::cancelling_during_retry_reap_retains_group_and_charge_until_exit_ack",
            CHILD_ENV,
        ) {
            return;
        }
        let Some(sh) = vllm_runtime::which("sh") else {
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        unsafe { std::env::set_var(car_home::ENV_VAR, dir.path().join("car-home")) };
        let pid_file = dir.path().join("attempt.pid");
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\nprintf '%s\\n' \"$$\" > '{}'\nexec sleep 60\n",
                sh.display(),
                pid_file.display(),
            ),
        )
        .unwrap();
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();

        let model_id = "vllm-mlx/cancel-retry-reap";
        let coordinator = Arc::new(crate::resource_policy::LocalAdmissionCoordinator::new(
            crate::resource_policy::ResourcePolicy::custom_gb(8.0).unwrap(),
            crate::hardware::HardwareInfo::detect(),
        ));
        let reservation = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .unwrap();
        let mut pool = VllmServerPool::with_test_runtime(
            Duration::from_secs(300),
            coordinator.clone(),
            script,
        );
        pool.startup_teardown_delay = Duration::from_secs(1);
        pool.readiness_limits = fast_readiness_limits();
        let pool = Arc::new(pool);

        let startup_pool = pool.clone();
        let startup = tokio::spawn(async move {
            startup_pool
                .ensure(model_id, "fixture/model", &reservation, "qwen3")
                .await
        });
        tokio::time::timeout(Duration::from_secs(2), async {
            while !pid_file.exists() {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("the first process must record its group leader");
        let process_group_id: i32 = std::fs::read_to_string(&pid_file)
            .unwrap()
            .trim()
            .parse()
            .unwrap();
        tokio::time::sleep(pool.readiness_limits.stall_timeout + Duration::from_millis(100)).await;

        startup.abort();
        let _ = startup.await;
        assert!(coordinator.teardown_pending(model_id));
        assert_eq!(
            coordinator.resident_model_mb(),
            1,
            "the pending process allocation must stay charged after cancellation"
        );
        assert_eq!(
            unsafe { libc::kill(-process_group_id, 0) },
            0,
            "the reaper must retain the exact live process group during quarantine"
        );
        let blocked = coordinator
            .reserve_measured_host(model_id, 1024 * 1024, 0)
            .unwrap_err();
        assert_eq!(
            blocked.preflight.verdict,
            crate::resource_policy::LocalLoadVerdict::PendingTeardown
        );

        pool.wait_for_teardown(model_id, Duration::from_secs(3))
            .await
            .expect("quarantine must clear after the reaper confirms group exit");
        assert_eq!(coordinator.resident_model_mb(), 0);
        assert_eq!(coordinator.active_request_count(model_id), 0);
        assert_ne!(unsafe { libc::kill(-process_group_id, 0) }, 0);
    }
}

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