car-inference 0.55.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.
//!
//! A server must not outlive the CAR process that spawned it. Neither of the
//! above covers that: the pool's `Drop` hands servers to a reaper thread that
//! dies with the process, and only the daemon runs idle eviction, so a `car`
//! CLI run, an embedding host, or a killed daemon all left a model server
//! pinning ~20 GB of GPU memory with nobody left to stop it. Two mechanisms
//! close it. A `Lifeline` watcher kills the group the moment the owner
//! exits, however it exits. A per-group `ServerProcessRecord` lets the next
//! pool on the same state root reclaim a group the lifeline missed, provided a
//! recorded member can still be verified. Neither reaches servers spawned by a
//! CAR build that predates them.

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; dropping an attached `Lifeline` terminates the
    /// whole group.
    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>,
    /// Kills the group if this process exits before teardown confirms it.
    #[cfg(unix)]
    lifeline: Option<Lifeline>,
    /// This group's `ServerProcessRecord`, removed once exit is confirmed.
    #[cfg(unix)]
    record: Option<PathBuf>,
    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);
        };
        // The watcher stays a member until it is retired and says nothing about
        // the server, so counting it would keep a crashed server "alive".
        #[cfg(any(target_os = "macos", target_os = "linux"))]
        if let Some(lifeline) = &self.lifeline {
            let members = process_group_members(process_group_id)?;
            // The watcher is our unreaped child, so it is listed alive or as a
            // zombie until `retire_lifeline` reaps it. A listing without it
            // failed, and libproc reports failure as an empty group: trusting
            // that would declare a live server dead and drop its accounting.
            if !members.contains(&lifeline.watcher_pid) {
                return Err(std::io::Error::other(
                    "process group listing omitted the unreaped lifeline watcher",
                ));
            }
            return Ok(members.into_iter().any(|pid| pid != lifeline.watcher_pid));
        }
        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)
    }

    /// True once the leader is reaped and the group is empty. This is the one
    /// place exit is confirmed, so it also retires the lifeline and the record.
    fn fully_exited(&mut self) -> std::io::Result<bool> {
        if self.child.try_wait()?.is_none() || self.process_group_is_alive()? {
            return Ok(false);
        }
        #[cfg(unix)]
        {
            // Record first: reaping the watcher frees the PGID, and another CAR
            // process could record a new group under that number. If retiring
            // then fails, the live watcher still protects the group.
            if let Some(record) = self.record.take() {
                let _ = std::fs::remove_file(record);
            }
            if !self.retire_lifeline()? {
                return Ok(false);
            }
        }
        Ok(true)
    }

    /// Kill and reap the watcher of a group already confirmed empty. Reaping
    /// matters: a zombie watcher would keep the PGID in use. `Ok(false)` means
    /// it was not reaped within the bound; callers retry like any other
    /// unconfirmed exit.
    ///
    /// The wait blocks, even from `ensure` and `stop` on a tokio worker, and
    /// that is deliberate. `SIGKILL` to a shell parked in `read` is reaped in
    /// about a millisecond, and 100 ms is only the ceiling. Returning
    /// `Ok(false)` straight after the kill would instead turn every
    /// transparent respawn of a crashed server into a "teardown pending" error
    /// for the caller, and making this async would ripple through the
    /// synchronous confirmation predicate the reaper thread also uses.
    #[cfg(unix)]
    fn retire_lifeline(&mut self) -> std::io::Result<bool> {
        let Some(lifeline) = self.lifeline.as_mut() else {
            return Ok(true);
        };
        if lifeline.watcher.try_wait()?.is_none() {
            // Our own unreaped child, so this PID cannot name anyone else.
            lifeline.watcher.start_kill()?;
            let deadline = Instant::now() + Duration::from_millis(100);
            while lifeline.watcher.try_wait()?.is_none() {
                if Instant::now() >= deadline {
                    return Ok(false);
                }
                std::thread::sleep(Duration::from_millis(1));
            }
        }
        self.lifeline = None;
        Ok(true)
    }

    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 for this process's lifetime rather than permit
        // unsafe peer discount. A leaked lifeline still kills the group when
        // this process exits.
        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)
}

/// Every PID currently in `process_group_id`, zombies included.
///
/// libproc turns a failed query into a count of zero, so an empty result is
/// indistinguishable from an empty group. Callers that can know a member must
/// be present check for it (see `ManagedServer::process_group_is_alive`).
#[cfg(target_os = "macos")]
fn process_group_members(process_group_id: i32) -> std::io::Result<Vec<libc::pid_t>> {
    // 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 Err(std::io::Error::last_os_error());
    }
    let capacity = usize::try_from(estimated_count)
        .unwrap_or(0)
        .saturating_add(16);
    let mut pids = vec![0 as libc::pid_t; capacity];
    let buffer_bytes = capacity
        .checked_mul(std::mem::size_of::<libc::pid_t>())
        .and_then(|bytes| i32::try_from(bytes).ok())
        .ok_or_else(|| std::io::Error::other("process group listing buffer overflows"))?;
    // SAFETY: `pids` owns `buffer_bytes` writable bytes for the duration of
    // the call, and the returned count is range-checked before use.
    let count = unsafe {
        libc::proc_listpgrppids(
            process_group_id,
            pids.as_mut_ptr().cast::<libc::c_void>(),
            buffer_bytes,
        )
    };
    if count < 0 {
        return Err(std::io::Error::last_os_error());
    }
    let count = usize::try_from(count).unwrap_or(0);
    if count >= capacity {
        return Err(std::io::Error::other(
            "process group listing may have been truncated",
        ));
    }
    pids.truncate(count);
    pids.retain(|pid| *pid > 0);
    Ok(pids)
}

/// Every PID currently in `process_group_id`, zombies included.
#[cfg(target_os = "linux")]
fn process_group_members(process_group_id: i32) -> std::io::Result<Vec<libc::pid_t>> {
    let mut members = Vec::new();
    for entry in std::fs::read_dir("/proc")?.flatten() {
        let Ok(pid) = entry.file_name().to_string_lossy().parse::<i32>() else {
            continue;
        };
        // A process that exits mid-scan simply is not a member.
        let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else {
            continue;
        };
        if linux_stat_field(&stat, 2).and_then(|value| value.parse::<i32>().ok())
            == Some(process_group_id)
        {
            members.push(pid);
        }
    }
    Ok(members)
}

/// Field `index` of `/proc/<pid>/stat`, counted from the state field (index 0)
/// that follows the parenthesised command name, which may itself hold spaces.
#[cfg(target_os = "linux")]
fn linux_stat_field(stat: &str, index: usize) -> Option<&str> {
    stat.rsplit_once(") ")?.1.split_whitespace().nth(index)
}

#[cfg(target_os = "macos")]
fn process_rss_bytes(pid: libc::pid_t) -> Option<u64> {
    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;
    }
    Some(unsafe { task_info.assume_init() }.pti_resident_size)
}

#[cfg(target_os = "linux")]
fn process_rss_bytes(pid: libc::pid_t) -> Option<u64> {
    let page_size = u64::try_from(unsafe { libc::sysconf(libc::_SC_PAGESIZE) })
        .ok()
        .filter(|size| *size > 0)?;
    let statm = std::fs::read_to_string(format!("/proc/{pid}/statm")).ok()?;
    let resident_pages = statm.split_whitespace().nth(1)?.parse::<u64>().ok()?;
    Some(resident_pages.saturating_mul(page_size))
}

#[cfg(any(target_os = "macos", target_os = "linux"))]
fn supervised_process_rss_bytes(server: &ManagedServer) -> Option<u64> {
    let members = process_group_members(server.process_group_id?).ok()?;
    if members.is_empty() {
        return None;
    }
    members.into_iter().try_fold(0u64, |total, pid| {
        Some(total.saturating_add(process_rss_bytes(pid)?))
    })
}

#[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>>,
    /// Where each spawned group's `ServerProcessRecord` is written. `None`
    /// until [`with_process_records`](Self::with_process_records).
    #[cfg(unix)]
    process_record_dir: Option<PathBuf>,
    #[cfg(test)]
    removal_race_hook: std::sync::Mutex<Option<Arc<RemovalRaceHook>>>,
}

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

// The hook is *stored* on every cfg(test) build (`pause_after_removal` reads
// it unconditionally), but it is only ever CONSTRUCTED by the cfg(unix)
// removal-race helper below. Gating the constructor to match its sole caller
// keeps `-D warnings` happy on the Windows leg without an `#[allow(dead_code)]`
// that would also hide a real future orphan.
#[cfg(all(test, unix))]
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(unix)]
            process_record_dir: 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
    }

    /// Record every server this pool spawns under `dir`, after first reclaiming
    /// any group recorded there by a CAR process that has since exited. Pass
    /// the same directory for every pool on one state root.
    #[cfg(unix)]
    pub fn with_process_records(mut self, dir: PathBuf) -> Self {
        reclaim_orphaned_server_groups(&dir);
        self.process_record_dir = Some(dir);
        self
    }

    /// Process groups are Unix-only, so there is nothing to record or reclaim.
    #[cfg(not(unix))]
    pub fn with_process_records(self, _dir: PathBuf) -> Self {
        self
    }

    #[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}"));
                }
            };
            // Attach before the readiness wait: an owner that dies while a
            // model is still loading orphans the group just the same.
            #[cfg(unix)]
            let process_group_id = spawned_process_group_id(&child);
            #[cfg(unix)]
            // Load-bearing order: no `try_wait` on `child` before this. An
            // unreaped leader, even one that already crashed, holds the PGID,
            // so the watcher can only ever join this server's group.
            let lifeline = process_group_id.and_then(spawn_lifeline);
            #[cfg(unix)]
            let record = match (self.process_record_dir.as_deref(), process_group_id) {
                (Some(dir), Some(process_group_id)) => {
                    record_server_process(dir, model_id, port, process_group_id, lifeline.as_ref())
                }
                _ => None,
            };
            let mut starting = StartingServerGuard {
                model_id: model_id.to_string(),
                server: Some(ManagedServer {
                    #[cfg(unix)]
                    process_group_id,
                    #[cfg(unix)]
                    lifeline,
                    #[cfg(unix)]
                    record,
                    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),
                #[cfg(unix)]
                lifeline: None,
                #[cfg(unix)]
                record: None,
                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);
    spawn_tolerating_text_file_busy(&mut command)
}

/// `exec` a command, riding out a transient `ETXTBSY`.
///
/// The kernel refuses to exec a file that any process still holds open for
/// writing. CAR *writes* the binaries it later spawns — `managed_venv`
/// provisions `vllm-mlx` into `~/.car/visual-runtime` — and a fork from any
/// other thread between that write and this exec hands the forked child a copy
/// of the still-open write descriptor until it reaches its own exec. During
/// that window the exec here fails with `ExecutableFileBusy` even though
/// nothing is wrong with the binary. The window is microseconds; a real
/// permission or corruption problem does not clear, so a bounded retry either
/// succeeds almost immediately or returns the same error it would have
/// returned anyway.
///
/// This is not hypothetical: it is what turned the `shared-process-test` job
/// red (`disable_xet_reaches_the_spawned_process`, CI run `33320488099`; the
/// job was added by car#1002), where dozens of sibling tests fork subprocesses
/// inside one process.
///
/// **Which kernels enforce this**, because the retry looks like dead code on
/// the machine most likely to be reading it: Linux does — `do_open_execat`
/// takes `deny_write_access` on the file being executed — and that is where the
/// job runs (`ubuntu-latest`). XNU does not, so on a Mac this branch never
/// fires; a write-open Mach-O execs and is then `SIGKILL`ed for an invalidated
/// signature instead, which is a different fault and not one a retry helps.
/// Do not delete this as macOS-dead: `vllm_pool` compiles and runs on Linux
/// (see the Linux RSS branch above), and the tests pinning it skip themselves
/// where the kernel cannot produce the condition.
///
/// **On the blocking sleep.** `spawn_server` is reached from `ensure`, which is
/// `async`, so a retry parks a tokio worker for up to ~190 ms. That is
/// deliberate, and it is cheaper than it looks:
///
/// - The first attempt is never preceded by a sleep, so the overwhelmingly
///   common path — no `ETXTBSY` — costs exactly nothing.
/// - The worker is already committed to blocking. `Command::spawn` is itself a
///   synchronous fork/exec on this thread, and `ensure` then polls the server
///   for readiness for *seconds*. Adding at most ~190 ms to a call that
///   routinely takes several seconds does not change the shape of the stall.
/// - When it does fire, the window is microseconds, so in practice this is one
///   10 ms sleep, not nineteen.
///
/// The alternatives are all worse here. Making `spawn_server` `async` ripples
/// through the crate for a microsecond-scale race; `spawn_blocking` does not
/// avoid that ripple, it relocates it — its closure must be `'static` and its
/// result awaited, so `spawn_server` ends up `async` anyway, one frame further
/// out and now with the `AuthorizedLocalSpawn<'_>` borrow at the admission
/// boundary to reshape; and busy-spinning burns the same worker harder while
/// shortening nothing.
fn spawn_tolerating_text_file_busy(command: &mut Command) -> std::io::Result<Child> {
    spawn_tolerating_text_file_busy_observed(command, || {})
}

fn spawn_tolerating_text_file_busy_observed(
    command: &mut Command,
    mut on_busy: impl FnMut(),
) -> std::io::Result<Child> {
    const ATTEMPTS: u32 = 20;
    const BACKOFF: std::time::Duration = std::time::Duration::from_millis(10);
    for _ in 1..ATTEMPTS {
        match command.spawn() {
            Err(e) if e.kind() == std::io::ErrorKind::ExecutableFileBusy => {
                on_busy();
                std::thread::sleep(BACKOFF);
            }
            outcome => return outcome,
        }
    }
    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()
}

/// Blocks on a pipe this process alone writes to, and kills its own process
/// group when that pipe reaches EOF. `HUP`/`INT`/`TERM` are ignored so only the
/// owner going away, or an explicit `SIGKILL`, ends it.
#[cfg(unix)]
const LIFELINE_SCRIPT: &str = "trap '' HUP INT TERM; read -r line; kill -s KILL 0";

/// A `/bin/sh` watcher inside a supervised server's process group that
/// `SIGKILL`s the group when the CAR process that spawned it goes away.
///
/// macOS has no parent-death signal, and `vllm-mlx` is third-party code that
/// cannot watch its owner itself. The watcher blocks reading a pipe whose
/// write end this process holds. The kernel closes that end on every kind of
/// exit: `main` returning, a panic, `SIGKILL`, a `car` CLI run finishing. So
/// EOF arrives when the owner goes away, and until then the watcher costs no
/// wakeups.
///
/// The write end can leak, and the lifeline then stays silent. On macOS std
/// creates the pipe with `pipe()` then sets close-on-exec, so a concurrent
/// spawn on another thread can inherit it. A fork without exec (a Python
/// host's `multiprocessing`) inherits it on any OS. Either failure is safe: the
/// group lives while that holder does, and `ServerProcessRecord` reclaims it
/// once the next CAR process on the state root builds its pool.
///
/// The watcher joins the group on purpose. A process group ID cannot be reused
/// while any member lives, so `kill 0` can never reach a stranger's group,
/// which an outside process signalling a remembered PGID could. The price is
/// that liveness checks must not count it (`ManagedServer::process_group_is_alive`).
#[cfg(unix)]
struct Lifeline {
    watcher: Child,
    watcher_pid: i32,
    /// Never written. Closing it is the signal.
    _pipe: tokio::process::ChildStdin,
}

#[cfg(any(target_os = "macos", target_os = "linux"))]
fn spawn_lifeline(process_group_id: i32) -> Option<Lifeline> {
    let mut command = Command::new("/bin/sh");
    command
        .arg("-c")
        .arg(LIFELINE_SCRIPT)
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .process_group(process_group_id);
    let mut watcher = match command.spawn() {
        Ok(watcher) => watcher,
        // The group has no members left to join, so nothing needs protecting.
        Err(error) if error.raw_os_error() == Some(libc::EPERM) => return None,
        Err(error) => {
            warn!(
                process_group_id,
                %error,
                "cannot attach a lifeline to the vllm-mlx process group; it will outlive this process unless teardown runs"
            );
            return None;
        }
    };
    let pipe = watcher.stdin.take();
    let watcher_pid = watcher.id().and_then(|id| i32::try_from(id).ok());
    match (pipe, watcher_pid) {
        (Some(pipe), Some(watcher_pid)) => Some(Lifeline {
            watcher,
            watcher_pid,
            _pipe: pipe,
        }),
        (pipe, _) => {
            // Kill before the pipe closes, or EOF would take the server with it.
            let _ = watcher.start_kill();
            drop(pipe);
            None
        }
    }
}

#[cfg(all(unix, not(any(target_os = "macos", target_os = "linux"))))]
fn spawn_lifeline(_process_group_id: i32) -> Option<Lifeline> {
    None
}

/// Enough to reclaim a server group whose owning CAR process exited without
/// tearing it down and whose `Lifeline` never fired (it was killed, or a
/// leaked descriptor kept its pipe open).
///
/// Limit: reclaim needs a recorded member it can verify. If the lifeline never
/// attached and the launcher has since exited, only unrecorded descendants
/// remain, and the record is dropped with the group still running.
///
/// Every PID is paired with its start time, because a PID alone names whatever
/// process holds that number today. The group is signalled only when one of its
/// recorded members is still the same process and still in the group, which
/// also proves the PGID has not been reused.
#[cfg(unix)]
#[derive(serde::Serialize, serde::Deserialize)]
struct ServerProcessRecord {
    owner_pid: i32,
    owner_started: String,
    process_group_id: i32,
    leader_started: Option<String>,
    watcher_pid: Option<i32>,
    watcher_started: Option<String>,
    model_id: String,
    port: u16,
}

/// When `pid` started, as a string that differs between any two processes that
/// ever held that PID. `None` when the process is gone or unreadable.
#[cfg(target_os = "macos")]
fn process_started(pid: i32) -> Option<String> {
    let mut info = std::mem::MaybeUninit::<libc::proc_bsdinfo>::zeroed();
    let info_bytes = i32::try_from(std::mem::size_of::<libc::proc_bsdinfo>()).ok()?;
    // SAFETY: `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_PIDTBSDINFO,
            0,
            info.as_mut_ptr().cast::<libc::c_void>(),
            info_bytes,
        )
    };
    if read != info_bytes {
        return None;
    }
    let info = unsafe { info.assume_init() };
    Some(format!(
        "{}.{:06}",
        info.pbi_start_tvsec, info.pbi_start_tvusec
    ))
}

/// Linux start times count ticks since boot, so the boot ID keeps a PID and
/// tick count from a previous boot from matching.
#[cfg(target_os = "linux")]
fn process_started(pid: i32) -> Option<String> {
    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
    let started = linux_stat_field(&stat, 19)?;
    let boot_id = std::fs::read_to_string("/proc/sys/kernel/random/boot_id").ok()?;
    Some(format!("{}:{started}", boot_id.trim()))
}

#[cfg(all(unix, not(any(target_os = "macos", target_os = "linux"))))]
fn process_started(_pid: i32) -> Option<String> {
    None
}

#[cfg(unix)]
fn is_same_process(pid: i32, started: &str) -> bool {
    process_started(pid).as_deref() == Some(started)
}

/// Write `record` under `dir` as `<pgid>.json`, atomically so a concurrent
/// reclaim never reads half a record. `None` if it could not be written; the
/// lifeline still protects the group.
#[cfg(unix)]
fn write_server_process_record(dir: &Path, record: &ServerProcessRecord) -> Option<PathBuf> {
    let path = dir.join(format!("{}.json", record.process_group_id));
    let staged = dir.join(format!("{}.json.tmp", record.process_group_id));
    let written = (|| -> std::io::Result<()> {
        std::fs::create_dir_all(dir)?;
        std::fs::write(
            &staged,
            serde_json::to_vec(record).map_err(std::io::Error::other)?,
        )?;
        std::fs::rename(&staged, &path)
    })();
    match written {
        Ok(()) => Some(path),
        Err(error) => {
            warn!(path = %path.display(), %error, "cannot record the vllm-mlx process group for later reclamation");
            let _ = std::fs::remove_file(&staged);
            None
        }
    }
}

#[cfg(unix)]
fn record_server_process(
    dir: &Path,
    model_id: &str,
    port: u16,
    process_group_id: i32,
    lifeline: Option<&Lifeline>,
) -> Option<PathBuf> {
    let owner_pid = i32::try_from(std::process::id()).ok()?;
    let record = ServerProcessRecord {
        owner_pid,
        owner_started: process_started(owner_pid)?,
        process_group_id,
        leader_started: process_started(process_group_id),
        watcher_pid: lifeline.map(|lifeline| lifeline.watcher_pid),
        watcher_started: lifeline.and_then(|lifeline| process_started(lifeline.watcher_pid)),
        model_id: model_id.to_string(),
        port,
    };
    // With no member identity there is nothing a reclaim could safely signal.
    if record.leader_started.is_none() && record.watcher_started.is_none() {
        return None;
    }
    write_server_process_record(dir, &record)
}

/// How long a staged `.json.tmp` must sit untouched before reclaim treats it as
/// left by a writer that died between write and rename. A live writer renames
/// within microseconds.
#[cfg(unix)]
const STALE_STAGED_RECORD: Duration = Duration::from_secs(60);

/// How long reclaim waits for a killed group to leave the process table before
/// keeping its record for the next pool to retry.
#[cfg(unix)]
const RECLAIM_EXIT_DEADLINE: Duration = Duration::from_secs(1);

/// Whether any process, zombies included, still holds `process_group_id`. XNU's
/// `kill(-pgid, 0)` skips zombies while Linux counts them, so the member
/// listing is the one answer both agree on. A failed listing reads as present
/// on Linux, which only delays forgetting a record.
#[cfg(any(target_os = "macos", target_os = "linux"))]
fn process_group_is_listed(process_group_id: i32) -> bool {
    process_group_members(process_group_id).map_or(true, |members| !members.is_empty())
}

#[cfg(all(unix, not(any(target_os = "macos", target_os = "linux"))))]
fn process_group_is_listed(process_group_id: i32) -> bool {
    (unsafe { libc::kill(-process_group_id, 0) }) == 0
}

/// Kill every group recorded under `dir` whose owner has exited, and drop the
/// record once the group is confirmed gone. A record whose owner is still
/// running is left alone: a daemon and a `car` CLI share one state root, and
/// each must not reclaim the other's live servers. A record none of whose
/// members can be verified is dropped without signalling anything, because its
/// PGID may belong to someone else by now.
#[cfg(unix)]
fn reclaim_orphaned_server_groups(dir: &Path) {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        let name = entry.file_name();
        let name = name.to_string_lossy();
        // A staged record is complete once written, so one whose writer died
        // before the rename still describes a group worth reclaiming.
        let is_record = name.ends_with(".json")
            || (name.ends_with(".json.tmp")
                && entry
                    .metadata()
                    .and_then(|metadata| metadata.modified())
                    .ok()
                    .and_then(|modified| modified.elapsed().ok())
                    .is_some_and(|age| age >= STALE_STAGED_RECORD));
        if !is_record {
            continue;
        }
        // Records are renamed into place, so an unparsable one is corrupt,
        // not half-written.
        let Some(record) = std::fs::read(&path)
            .ok()
            .and_then(|bytes| serde_json::from_slice::<ServerProcessRecord>(&bytes).ok())
        else {
            let _ = std::fs::remove_file(&path);
            continue;
        };
        // An owner whose start time cannot be read counts as exited. That read
        // fails only for another user's process (EPERM) or no process at all
        // (ESRCH, zombies included), and the CAR process that wrote a record
        // in this user's state root is always readable while it lives. Killing
        // still requires a verified member below.
        if is_same_process(record.owner_pid, &record.owner_started) {
            continue;
        }
        let still_in_group = |pid: Option<i32>, started: Option<&str>| match (pid, started) {
            (Some(pid), Some(started)) => {
                is_same_process(pid, started)
                    && unsafe { libc::getpgid(pid) } == record.process_group_id
            }
            _ => false,
        };
        // Verify-then-signal is not atomic, but reusing a PGID in that window
        // needs every member gone and the PID counter wrapped in between.
        if still_in_group(
            Some(record.process_group_id),
            record.leader_started.as_deref(),
        ) || still_in_group(record.watcher_pid, record.watcher_started.as_deref())
        {
            if unsafe { libc::killpg(record.process_group_id, libc::SIGKILL) } != 0 {
                let error = std::io::Error::last_os_error();
                if error.raw_os_error() != Some(libc::ESRCH) {
                    warn!(
                        model = %record.model_id,
                        process_group_id = record.process_group_id,
                        %error,
                        "cannot reclaim an orphaned vllm-mlx process group; keeping its record"
                    );
                    continue;
                }
            }
            // Members of an exited owner are reparented to init and reaped
            // promptly. One still listed afterwards keeps the record, so the
            // next pool verifies and retries rather than forgetting the group.
            let deadline = Instant::now() + RECLAIM_EXIT_DEADLINE;
            while process_group_is_listed(record.process_group_id) {
                if Instant::now() >= deadline {
                    warn!(
                        model = %record.model_id,
                        process_group_id = record.process_group_id,
                        "killed an orphaned vllm-mlx process group that has not exited yet; keeping its record"
                    );
                    break;
                }
                std::thread::sleep(Duration::from_millis(10));
            }
            if process_group_is_listed(record.process_group_id) {
                continue;
            }
            info!(
                model = %record.model_id,
                process_group_id = record.process_group_id,
                port = record.port,
                owner_pid = record.owner_pid,
                "reclaimed a vllm-mlx process group left running by an exited CAR process"
            );
        }
        let _ = std::fs::remove_file(&path);
    }
}

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

/// Wait until a subprocess readiness file contains a complete PID line.
///
/// Checking only `Path::exists` races shell redirection: the kernel creates and
/// truncates the file before `printf`/`echo` writes its bytes. Under host load
/// that gap is large enough for the reader to observe an empty file. Keep each
/// caller's existing wall bound, but wait for the actual parseable signal.
#[cfg(all(test, unix))]
async fn wait_for_pid_file(
    path: &Path,
    bound: Duration,
) -> Result<i32, tokio::time::error::Elapsed> {
    tokio::time::timeout(bound, async {
        loop {
            if let Ok(pid) = std::fs::read_to_string(path)
                .and_then(|line| line.trim().parse::<i32>().map_err(std::io::Error::other))
            {
                break pid;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await
}

#[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),
                #[cfg(unix)]
                lifeline: None,
                #[cfg(unix)]
                record: None,
                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());
        // The teardown delay is the discriminator, so it must dwarf host-load
        // jitter: a BLOCKING drop waits at least this long before it can even
        // begin killing the 60 s-sleeping child, while a non-blocking drop
        // returns in microseconds. 2 s delay vs a 1 s bound keeps the meaning
        // of the old 250 ms/100 ms pair with 10x the scheduling headroom
        // (car-sap7: the tight pair flaked under parallel builds).
        pool.insert_test_process_with_teardown_delay(
            "vllm-mlx/drop-test",
            child,
            Duration::from_secs(2),
        )
        .await;

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

        assert!(
            started.elapsed() < Duration::from_secs(1),
            "drop must hand off to the background reaper, not wait out the teardown delay"
        );
        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(30), 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(),
        ));
        // Wide enough that the abort below reliably lands while the release
        // is still parked in the delay — with a narrow window, host load could
        // let the release complete before the abort, which tests nothing.
        pool.insert_test_process_with_teardown_delay(
            "vllm-mlx/cancel",
            child,
            Duration::from_secs(2),
        )
        .await;

        let owner = pool.clone();
        let release =
            tokio::spawn(async move { owner.release_model_if_present("vllm-mlx/cancel").await });
        // Wait for the release to publish quarantine rather than sampling a
        // fixed 20 ms later — under host load the spawned task may not have
        // been scheduled at all by then (car-sap7). The 2 s teardown delay
        // keeps the release parked well past this observation.
        tokio::time::timeout(Duration::from_secs(10), async {
            while !coordinator.teardown_pending("vllm-mlx/cancel") {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("release must publish teardown quarantine");
        release.abort();
        let _ = release.await;
        assert!(coordinator.is_resident("vllm-mlx/cancel"));
        tokio::time::timeout(Duration::from_secs(30), 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();
        let descendant_pid = wait_for_pid_file(&descendant_pid_file, Duration::from_secs(30))
            .await
            .expect("stand-in server must publish its model-bearing descendant PID");

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

        // Poll rather than sampling once. `kill(pid, 0)` succeeds for a
        // ZOMBIE — a process that has already exited but whose parent has not
        // reaped it — so an instantaneous check cannot tell "still running"
        // from "dying". This test's `sleep` is a grandchild: `release` waits
        // for the direct child (`sh`) to exit, and the group signal reaches
        // the grandchild at the same moment, but its exit, reparenting to
        // init, and reaping are all asynchronous after that. The old
        // single-sample assertion therefore raced the reaper and went red on
        // an unrelated PR (car#1126) with nothing wrong in the runtime.
        //
        // The bound does NOT weaken the invariant: a `release` that failed to
        // signal the group at all leaves `sleep 60` running for the full
        // window and still fails. It tolerates reap latency, not a missing
        // kill. Mirrors the 2 s poll this same test already uses to wait for
        // the pid file to appear.
        let deadline = Duration::from_secs(30);
        let started = std::time::Instant::now();
        let mut descendant_survived = true;
        while started.elapsed() < deadline {
            if unsafe { libc::kill(descendant_pid, 0) } != 0 {
                descendant_survived = false;
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        if descendant_survived {
            unsafe {
                libc::kill(descendant_pid, libc::SIGKILL);
            }
        }
        assert!(
            !descendant_survived,
            "release must not acknowledge while a model-bearing descendant survives \
             (still alive {deadline:?} after release returned)"
        );
    }

    #[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));
        let descendant_pid = wait_for_pid_file(&descendant_pid_file, Duration::from_secs(30))
            .await
            .expect("launcher must publish its descendant PID");
        // The launcher exits on its own schedule; poll for it rather than
        // sampling once 50 ms later (car-sap7: single samples flake under load).
        let launcher_gone = Instant::now() + Duration::from_secs(30);
        while child.try_wait().unwrap().is_none() {
            assert!(Instant::now() < launcher_gone, "launcher must be gone");
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(
            unsafe { libc::getpgid(leader_pid) },
            -1,
            "OS lookup reproduces the old ESRCH race"
        );

        let mut server = ManagedServer {
            child,
            process_group_id: recorded_process_group,
            lifeline: None,
            record: None,
            port: 1,
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        server.start_kill_all().unwrap();
        tokio::time::timeout(Duration::from_secs(30), 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::test_python_interpreter() else {
            panic!("a real Python interpreter is required for the leaderless fixture");
        };
        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,
        );
        // Anti-hang bounds only: the fixture's expected path is healthy, and
        // a loaded host can hold a forking Python stand-in below its bind for
        // several seconds — tight values here were exactly the car-sap7 flake.
        pool.readiness_limits = ReadinessLimits {
            health_timeout: Duration::from_millis(25),
            ready_deadline: Duration::from_secs(60),
            stall_timeout: Duration::from_secs(30),
            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();
        let descendant_pid = wait_for_pid_file(&descendant_pid_file, Duration::from_secs(30))
            .await
            .expect("launcher must publish its descendant PID");
        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),
                lifeline: None,
                record: None,
                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(30), 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.
    #[cfg(unix)]
    #[tokio::test]
    async fn spawns_and_health_waits_a_stand_in_server() {
        let Some(python) = vllm_runtime::test_python_interpreter() else {
            panic!("a real Python interpreter is required for the stand-in server fixture");
        };
        // 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),
            #[cfg(unix)]
            lifeline: None,
            #[cfg(unix)]
            record: None,
            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),
            lifeline: None,
            record: None,
            child,
            port: alloc_loopback_port().unwrap(),
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        let endpoint = server.endpoint();
        // The fixture produces zero observable progress by construction, so
        // the stall watchdog CANNOT fire falsely — a short stall bound is
        // load-safe here. The hard deadline exists only to lose the race: it
        // must be far enough out that a scheduler stall can never let it win
        // over the stall verdict (car-sap7: at 750 ms it sometimes did).
        let limits = ReadinessLimits {
            health_timeout: Duration::from_millis(5),
            ready_deadline: Duration::from_secs(30),
            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),
            // The hard deadline is an anti-hang bound, not part of any
            // fixture's expected path — healthy attempts finish in well under
            // a second warm, and stalled attempts are decided by the stall
            // watchdog. Generous on purpose: under parallel builds a Python
            // stand-in server can take several seconds to bind (car-sap7,
            // car-9xug: 3 s lost 2 of 3 serialized runs under load).
            ready_deadline: Duration::from_secs(60),
            // The stall bound decides the STALLING fixtures — but every
            // assertion afterwards reads what the fixture SHELL recorded, and
            // under host starvation a freshly forked shell can fail to run
            // even its first line inside a small window (measured on this
            // bead: at 1 s, a loaded run stalled both attempts before either
            // script wrote its attempt line, and the healthy retry stalled
            // before its heartbeat subshell was scheduled). Ten seconds
            // dwarfs fork/exec starvation while the isolated-child mutex
            // keeps the serial cost bounded to these few tests.
            stall_timeout: Duration::from_secs(10),
            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 ready = dir.path().join("ready.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}}); server=http.server.HTTPServer(('127.0.0.1',p),H); open('{}','w').write('ready\\n'); server.serve_forever()\" \"$@\"\n",
                sh.display(),
                attempts.display(),
                python.display(),
                ready.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(120),
            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_eq!(
            std::fs::read_to_string(&ready).unwrap(),
            "ready\n",
            "the healthy retry must publish readiness only after its port binds"
        );
        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(120),
            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(30))
            .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),
            // This test proves the END-TO-END cleanup when the finite hard
            // deadline fires; the progress-extends-stall-but-not-deadline
            // semantics are pinned deterministically by the injected-duration
            // unit test recent_target_progress_can_outlive_stall_but_not_the_
            // hard_deadline. Setting stall ABOVE the deadline makes the
            // deadline verdict load-independent: however late the fixture's
            // first stderr write lands, the stall watchdog can never win the
            // race and turn the expected "startup deadline" error into
            // "no progress" (car-sap7). The deadline itself is wide enough
            // that a starved shell still records its attempt line first.
            ready_deadline: Duration::from_secs(10),
            stall_timeout: Duration::from_secs(30),
            poll: Duration::from_millis(5),
        };

        let error = tokio::time::timeout(
            Duration::from_secs(60),
            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(30))
            .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,
        );
        // Wide reap window so the abort below reliably lands DURING the reap
        // of the first stalled group, which is the scenario under test.
        pool.startup_teardown_delay = Duration::from_secs(5);
        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
        });
        let process_group_id = wait_for_pid_file(&pid_file, Duration::from_secs(30))
            .await
            .expect("the first process must record its complete group leader PID");
        // Wait for the stall verdict to actually publish quarantine instead
        // of sleeping stall_timeout+100ms and hoping the startup task was
        // scheduled — under host load it often was not (car-sap7). The 5 s
        // teardown delay above keeps the startup parked in the reap while the
        // abort lands.
        tokio::time::timeout(Duration::from_secs(60), async {
            while !coordinator.teardown_pending(model_id) {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("the stalled first attempt must enter teardown quarantine");

        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(30))
            .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);
    }

    /// Write an executable `sh` script into `dir` that records the fact that it
    /// ran, so a caller can tell a real `exec` from a swallowed error.
    #[cfg(unix)]
    fn write_marker_script(dir: &Path, sh: &Path, name: &str, marker: &Path) -> PathBuf {
        use std::os::unix::fs::PermissionsExt;
        let script = dir.join(name);
        std::fs::write(
            &script,
            format!("#!{}\n: > '{}'\n", sh.display(), marker.display()),
        )
        .unwrap();
        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
        script
    }

    /// Whether this kernel refuses to `exec` a file that another descriptor
    /// holds open for writing — i.e. whether the fixture the two tests below
    /// stand on is real here.
    ///
    /// Linux does: `do_open_execat` takes `deny_write_access` on the file being
    /// executed, and for a `#!` script that file is the script itself, so
    /// `ETXTBSY` is available on demand with no race to lose. XNU does **not**
    /// — measured on darwin 25.6.0/arm64, a write-open shebang script execs and
    /// runs normally (a write-open Mach-O execs and is then `SIGKILL`ed for an
    /// invalidated signature, which is a different fault with a different
    /// remedy). `shared-process-test`, the job whose red these tests pin, runs
    /// on `ubuntu-latest`, so asserting where the kernel provides the condition
    /// covers the failure; asserting on a Mac would only pin XNU's silence.
    #[cfg(unix)]
    fn kernel_refuses_exec_of_a_write_open_file(dir: &Path, sh: &Path) -> bool {
        let probe = write_marker_script(dir, sh, "etxtbsy-probe", &dir.join("probe-ran"));
        let held = std::fs::OpenOptions::new()
            .write(true)
            .open(&probe)
            .expect("the probe script must be openable for writing");
        let outcome = std::process::Command::new(&probe)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn();
        drop(held);
        match outcome {
            Err(e) => e.kind() == std::io::ErrorKind::ExecutableFileBusy,
            Ok(mut child) => {
                let _ = child.wait();
                false
            }
        }
    }

    /// CAR writes the binaries it then spawns, and a `fork` on any other thread
    /// carries a copy of the still-open write descriptor until its own `exec`.
    /// During that window the kernel refuses this `exec` with `ETXTBSY` even
    /// though nothing is wrong with the file — which is what turned
    /// `disable_xet_reaches_the_spawned_process` red in CI run `33320488099`,
    /// where dozens of sibling tests fork subprocesses inside one process.
    ///
    /// Holding the write handle here makes that condition available on demand
    /// rather than by winning a race, and releasing it well inside the retry's
    /// ~190 ms ceiling is exactly the transient the retry exists for.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_transient_text_file_busy_is_ridden_out_not_surfaced() {
        let Some(sh) = vllm_runtime::which("sh") else {
            eprintln!(
                "SKIP a_transient_text_file_busy_is_ridden_out_not_surfaced: no `sh` on PATH, so the fixture cannot build an executable script"
            );
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        if !kernel_refuses_exec_of_a_write_open_file(dir.path(), &sh) {
            eprintln!(
                "SKIP a_transient_text_file_busy_is_ridden_out_not_surfaced: this kernel execs a write-open file instead of refusing it \
                 (expected on XNU), so ETXTBSY cannot be produced here"
            );
            return;
        }
        let ran = dir.path().join("ran.txt");
        let script = write_marker_script(dir.path(), &sh, "fake-vllm-mlx", &ran);

        let held = std::fs::OpenOptions::new()
            .write(true)
            .open(&script)
            .expect("the fixture must hold a write descriptor on the target");
        let mut held = Some(held);
        let mut command = Command::new(&script);
        // Release the write descriptor only after the real first ETXTBSY. The
        // old 60 ms sleeper measured thread scheduling under host load and
        // could outlive the retry ceiling without testing a different runtime
        // behavior. This callback is the fixture's actual readiness signal.
        let spawned = spawn_tolerating_text_file_busy_observed(&mut command, || {
            drop(held.take());
        });
        let mut child = spawned.expect(
            "a write descriptor released inside the retry ceiling is transient and must be \
             ridden out, not surfaced to the caller",
        );
        let status = child.wait().await.unwrap();
        assert!(status.success(), "the retried attempt must run: {status:?}");
        assert!(
            ran.exists(),
            "the retry must end in a real exec, not in a swallowed error"
        );
    }

    /// The other half of the contract: a bounded retry must not turn a
    /// permanent fault into a delayed lie. A descriptor still open past the
    /// ceiling is a real failure and has to reach the caller as one, with its
    /// kind intact so the caller can tell it apart from a missing binary.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_text_file_busy_that_outlasts_the_ceiling_is_still_surfaced() {
        let Some(sh) = vllm_runtime::which("sh") else {
            eprintln!(
                "SKIP a_text_file_busy_that_outlasts_the_ceiling_is_still_surfaced: no `sh` on PATH, so the fixture cannot build an executable script"
            );
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        if !kernel_refuses_exec_of_a_write_open_file(dir.path(), &sh) {
            eprintln!(
                "SKIP a_text_file_busy_that_outlasts_the_ceiling_is_still_surfaced: this kernel execs a write-open file instead of refusing it \
                 (expected on XNU), so ETXTBSY cannot be produced here"
            );
            return;
        }
        let ran = dir.path().join("ran.txt");
        let script = write_marker_script(dir.path(), &sh, "fake-vllm-mlx", &ran);

        let _held = std::fs::OpenOptions::new()
            .write(true)
            .open(&script)
            .expect("the fixture must hold a write descriptor on the target");
        let started = Instant::now();
        let err = spawn_server(&script, "dummy/model", 1, false, "qwen3")
            .expect_err("a fault that outlasts the retry ceiling must not be swallowed");

        assert_eq!(
            err.kind(),
            std::io::ErrorKind::ExecutableFileBusy,
            "the original error kind must survive the retry: {err:?}"
        );
        assert!(!ran.exists(), "nothing may have executed");
        let elapsed = started.elapsed();
        assert!(
            elapsed >= Duration::from_millis(100),
            "the retry must actually have been attempted, not skipped: {elapsed:?}"
        );
        assert!(
            elapsed < Duration::from_secs(60),
            "the retry must stay bounded rather than waiting the fault out: {elapsed:?}"
        );
    }
}

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

/// A server must not outlive the CAR process that spawned it, and reclaiming
/// what an exited process left behind must never signal a stranger's group.
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
mod orphan_tests {
    use super::*;

    /// A stand-in server: its own process group, doing nothing for a minute.
    fn spawn_stand_in_group(script: &str) -> (Child, i32) {
        let mut command = Command::new("/bin/sh");
        command
            .arg("-c")
            .arg(script)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .process_group(0);
        let child = command.spawn().expect("spawn stand-in server group");
        let process_group_id = spawned_process_group_id(&child).unwrap();
        (child, process_group_id)
    }

    async fn wait_for_group_exit(process_group_id: i32, bound: Duration) -> bool {
        let deadline = Instant::now() + bound;
        while Instant::now() < deadline {
            if unsafe { libc::kill(-process_group_id, 0) } != 0 {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        false
    }

    /// The case that cost ~20 GB of GPU memory for ten days: the owner dies
    /// without running any teardown at all. A real SIGKILL of a real owner
    /// process rather than a dropped handle, so the kernel's own descriptor
    /// cleanup is what fires the lifeline.
    #[tokio::test]
    async fn lifeline_kills_the_group_when_its_owner_is_sigkilled() {
        const OWNER_ENV: &str = "CAR_VLLM_LIFELINE_OWNER_PGID_FILE";
        if let Some(pgid_file) = std::env::var_os(OWNER_ENV) {
            let (child, process_group_id) = spawn_stand_in_group("sleep 60 & wait");
            let lifeline = spawn_lifeline(process_group_id).expect("lifeline attaches");
            let staged = PathBuf::from(&pgid_file).with_extension("tmp");
            std::fs::write(&staged, process_group_id.to_string()).unwrap();
            std::fs::rename(&staged, &pgid_file).unwrap();
            tokio::time::sleep(Duration::from_secs(60)).await;
            drop((lifeline, child));
            return;
        }

        let dir = tempfile::tempdir().unwrap();
        let pgid_file = dir.path().join("server.pgid");
        let mut owner = Command::new(std::env::current_exe().unwrap())
            .arg("--exact")
            .arg("vllm_pool::orphan_tests::lifeline_kills_the_group_when_its_owner_is_sigkilled")
            .env(OWNER_ENV, &pgid_file)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
            .unwrap();
        let process_group_id = wait_for_pid_file(&pgid_file, Duration::from_secs(60))
            .await
            .expect("owner must publish its server group PID");

        // Positive control: the watcher must not fire while its owner lives.
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(
            unsafe { libc::kill(-process_group_id, 0) },
            0,
            "the server group must stay up while its owner is alive"
        );

        owner.start_kill().unwrap();
        owner.wait().await.unwrap();
        let exited = wait_for_group_exit(process_group_id, Duration::from_secs(30)).await;
        if !exited {
            unsafe { libc::killpg(process_group_id, libc::SIGKILL) };
        }
        assert!(exited, "the server group must die with its SIGKILLed owner");
    }

    /// The watcher is a group member, so a naive liveness check would report
    /// a crashed server as alive for as long as its owner runs.
    #[tokio::test]
    async fn a_crashed_server_is_not_kept_alive_by_its_lifeline() {
        let (child, process_group_id) = spawn_stand_in_group("sleep 1");
        let lifeline = spawn_lifeline(process_group_id).expect("lifeline attaches");
        let mut server = ManagedServer {
            child,
            process_group_id: Some(process_group_id),
            lifeline: Some(lifeline),
            record: None,
            port: 1,
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        assert!(server.is_alive().unwrap(), "a running server is alive");

        tokio::time::timeout(Duration::from_secs(30), async {
            while server.is_alive().unwrap() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("a server whose only remaining member is its watcher has exited");
        assert!(
            server
                .lifeline
                .as_mut()
                .unwrap()
                .watcher
                .try_wait()
                .unwrap()
                .is_none(),
            "the watcher was still running, so only the exclusion made the server read as dead"
        );

        tokio::time::timeout(Duration::from_secs(30), async {
            while !server.fully_exited().unwrap() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("confirmed exit must retire the watcher");
        assert!(server.lifeline.is_none());
        assert_ne!(
            unsafe { libc::kill(-process_group_id, 0) },
            0,
            "a retired watcher must be reaped, not left holding the PGID as a zombie"
        );
    }

    /// libproc reports a failed listing as an empty group. If that were taken at
    /// face value a live server would read as dead, lose its lifeline and record,
    /// and have its admission charge released while it keeps running.
    #[tokio::test]
    async fn a_listing_that_omits_the_watcher_is_an_error_not_an_empty_group() {
        let (child, process_group_id) = spawn_stand_in_group("sleep 60");
        let mut lifeline = spawn_lifeline(process_group_id).expect("lifeline attaches");
        let real_watcher_pid = lifeline.watcher_pid;
        let mut server = ManagedServer {
            child,
            process_group_id: Some(process_group_id),
            lifeline: None,
            record: None,
            port: 1,
            last_used: Instant::now(),
            teardown_delay: Duration::ZERO,
            inspection_error: false,
        };
        // A PID that cannot be in the group stands in for a listing that lost it.
        lifeline.watcher_pid = i32::try_from(std::process::id()).unwrap();
        server.lifeline = Some(lifeline);
        assert!(server.leader_has_exited().is_ok());
        let error = server
            .process_group_is_alive()
            .expect_err("a listing without the unreaped watcher must not be trusted");
        assert!(error.to_string().contains("lifeline watcher"), "{error}");

        // Positive control: the real watcher makes the same group read as alive.
        server.lifeline.as_mut().unwrap().watcher_pid = real_watcher_pid;
        assert!(server.process_group_is_alive().unwrap());
    }

    #[tokio::test]
    async fn ensure_records_its_group_and_confirmed_teardown_removes_the_record() {
        const CHILD_ENV: &str = "CAR_VLLM_PROCESS_RECORD_CHILD";
        if !crate::run_in_isolated_test_process(
            "vllm_pool::orphan_tests::ensure_records_its_group_and_confirmed_teardown_removes_the_record",
            CHILD_ENV,
        ) {
            return;
        }
        let Some(python) = vllm_runtime::test_python_interpreter() else {
            panic!("a real Python interpreter is required for the stand-in server fixture");
        };
        let dir = tempfile::tempdir().unwrap();
        unsafe { std::env::set_var(car_home::ENV_VAR, dir.path().join("car-home")) };
        let script = dir.path().join("fake-vllm-mlx");
        std::fs::write(
            &script,
            format!(
                "#!{}\n\
                 import sys,http.server\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(),
            ),
        )
        .unwrap();
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();

        let model_id = "vllm-mlx/recorded";
        let records = dir.path().join("run").join("vllm-mlx");
        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, 64)
            .unwrap();
        let mut pool =
            VllmServerPool::with_test_runtime(Duration::from_secs(300), coordinator, script)
                .with_process_records(records.clone());
        pool.readiness_limits = ReadinessLimits {
            health_timeout: Duration::from_millis(25),
            ready_deadline: Duration::from_secs(60),
            stall_timeout: Duration::from_secs(30),
            poll: Duration::from_millis(10),
        };

        pool.ensure(model_id, "fixture/model", &reservation, "qwen3")
            .await
            .expect("stand-in server becomes healthy");
        let (process_group_id, watcher_pid) = {
            let servers = pool.servers.lock().await;
            let server = &servers[model_id];
            let lifeline = server
                .lifeline
                .as_ref()
                .expect("ensure attaches a lifeline");
            (server.process_group_id.unwrap(), lifeline.watcher_pid)
        };
        let record_path = records.join(format!("{process_group_id}.json"));
        let record: ServerProcessRecord =
            serde_json::from_slice(&std::fs::read(&record_path).expect("ensure writes a record"))
                .unwrap();
        assert_eq!(record.process_group_id, process_group_id);
        assert_eq!(record.owner_pid, i32::try_from(std::process::id()).unwrap());
        assert_eq!(record.watcher_pid, Some(watcher_pid));
        assert!(record.leader_started.is_some() && record.watcher_started.is_some());
        assert_eq!(record.model_id, model_id);

        assert!(pool.release_model_if_present(model_id).await.unwrap());
        assert!(
            !record_path.exists(),
            "confirmed teardown must remove the record"
        );
        assert_ne!(unsafe { libc::kill(-process_group_id, 0) }, 0);
    }

    fn stand_in_record(
        process_group_id: i32,
        owner_started: String,
        leader_started: Option<String>,
    ) -> ServerProcessRecord {
        ServerProcessRecord {
            owner_pid: i32::try_from(std::process::id()).unwrap(),
            owner_started,
            process_group_id,
            leader_started,
            watcher_pid: None,
            watcher_started: None,
            model_id: "vllm-mlx/stand-in".into(),
            port: 1,
        }
    }

    /// Run reclaim off the runtime so this test can reap its own stand-ins
    /// meanwhile. In production an exited owner's members are reaped by init;
    /// here the test is their parent, and unreaped they still hold the PGID.
    async fn reclaim_while_reaping(dir: &Path) -> tokio::task::JoinHandle<()> {
        let dir = dir.to_path_buf();
        tokio::task::spawn_blocking(move || reclaim_orphaned_server_groups(&dir))
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn reclaim_kills_a_verified_group_whose_owner_exited() {
        let dir = tempfile::tempdir().unwrap();
        let (mut child, process_group_id) = spawn_stand_in_group("sleep 60");
        // Same PID, different start time: exactly what a dead owner whose PID
        // was reused looks like.
        let record = stand_in_record(
            process_group_id,
            "an exited owner".into(),
            process_started(process_group_id),
        );
        let path = write_server_process_record(dir.path(), &record).unwrap();

        let reclaim = reclaim_while_reaping(dir.path()).await;
        let status = tokio::time::timeout(Duration::from_secs(10), child.wait())
            .await
            .expect("reclaim must kill the orphaned group")
            .unwrap();
        reclaim.await.unwrap();
        assert!(!status.success());
        assert!(!path.exists(), "a reclaimed group's record is removed");
    }

    /// A signalled group that has not left the process table is not reclaimed
    /// yet. Forgetting its record would leave nothing to retry with.
    #[tokio::test]
    async fn reclaim_keeps_the_record_until_the_killed_group_is_gone() {
        let dir = tempfile::tempdir().unwrap();
        let (mut child, process_group_id) = spawn_stand_in_group("sleep 60");
        let record = stand_in_record(
            process_group_id,
            "an exited owner".into(),
            process_started(process_group_id),
        );
        let path = write_server_process_record(dir.path(), &record).unwrap();

        // Nothing reaps the killed leader while reclaim runs, so it lingers as
        // a zombie member for the whole exit deadline.
        reclaim_orphaned_server_groups(dir.path());

        assert!(path.exists(), "an unconfirmed reclaim must keep the record");
        let status = tokio::time::timeout(Duration::from_secs(10), child.wait())
            .await
            .expect("the group was still signalled")
            .unwrap();
        assert!(!status.success());
    }

    /// A writer that died between writing its staged record and renaming it
    /// still left a group behind; a fresh staged record is someone mid-write.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn reclaim_adopts_only_stale_staged_records() {
        let dir = tempfile::tempdir().unwrap();
        let (mut fresh_child, fresh_group) = spawn_stand_in_group("sleep 60");
        let (mut stale_child, stale_group) = spawn_stand_in_group("sleep 60");
        for (group, age) in [
            (fresh_group, Duration::ZERO),
            (stale_group, STALE_STAGED_RECORD + Duration::from_secs(5)),
        ] {
            let record = stand_in_record(group, "an exited owner".into(), process_started(group));
            let staged = dir.path().join(format!("{group}.json.tmp"));
            std::fs::write(&staged, serde_json::to_vec(&record).unwrap()).unwrap();
            std::fs::File::options()
                .write(true)
                .open(&staged)
                .unwrap()
                .set_modified(std::time::SystemTime::now() - age)
                .unwrap();
        }

        let reclaim = reclaim_while_reaping(dir.path()).await;
        let status = tokio::time::timeout(Duration::from_secs(10), stale_child.wait())
            .await
            .expect("a stale staged record's group is reclaimed")
            .unwrap();
        reclaim.await.unwrap();
        assert!(!status.success());
        assert!(!dir.path().join(format!("{stale_group}.json.tmp")).exists());

        assert!(
            fresh_child.try_wait().unwrap().is_none(),
            "a fresh staged record may belong to a live writer"
        );
        assert!(dir.path().join(format!("{fresh_group}.json.tmp")).exists());
    }

    #[tokio::test]
    async fn reclaim_leaves_a_live_owners_group_and_record_alone() {
        let dir = tempfile::tempdir().unwrap();
        let (mut child, process_group_id) = spawn_stand_in_group("sleep 60");
        let own_pid = i32::try_from(std::process::id()).unwrap();
        let record = stand_in_record(
            process_group_id,
            process_started(own_pid).expect("this process has a start time"),
            process_started(process_group_id),
        );
        let path = write_server_process_record(dir.path(), &record).unwrap();

        reclaim_orphaned_server_groups(dir.path());

        tokio::time::sleep(Duration::from_millis(200)).await;
        assert!(
            child.try_wait().unwrap().is_none(),
            "a live owner's server must not be killed"
        );
        assert!(path.exists(), "a live owner's record must be kept");
    }

    #[tokio::test]
    async fn reclaim_never_signals_a_group_it_cannot_verify() {
        let dir = tempfile::tempdir().unwrap();
        let (mut child, process_group_id) = spawn_stand_in_group("sleep 60");
        // Owner gone, and the PGID now names a different process: the reuse
        // case, where signalling would kill a stranger.
        let record = stand_in_record(
            process_group_id,
            "an exited owner".into(),
            Some("a different process".into()),
        );
        let path = write_server_process_record(dir.path(), &record).unwrap();

        reclaim_orphaned_server_groups(dir.path());

        tokio::time::sleep(Duration::from_millis(200)).await;
        assert!(
            child.try_wait().unwrap().is_none(),
            "an unverifiable group must never be signalled"
        );
        assert!(!path.exists(), "an unverifiable record is dropped");
    }

    /// The real `vllm-mlx` shape: the launcher is gone and reaped, so only the
    /// watcher can prove the PGID still names this server's group.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn reclaim_verifies_a_group_by_its_watcher_after_the_launcher_exited() {
        let dir = tempfile::tempdir().unwrap();
        let descendant_pid_file = dir.path().join("descendant.pid");
        let (mut launcher, process_group_id) = spawn_stand_in_group(&format!(
            "sleep 60 & echo $! > '{}'; exit 0",
            descendant_pid_file.display()
        ));
        // Attaches even if the launcher already exited: it is still unreaped.
        let mut lifeline = spawn_lifeline(process_group_id).expect("lifeline attaches");
        let descendant_pid = wait_for_pid_file(&descendant_pid_file, Duration::from_secs(30))
            .await
            .expect("launcher must publish its model-bearing descendant PID");
        launcher.wait().await.unwrap();
        assert_eq!(unsafe { libc::kill(descendant_pid, 0) }, 0);

        let mut record = stand_in_record(process_group_id, "an exited owner".into(), None);
        record.watcher_pid = Some(lifeline.watcher_pid);
        record.watcher_started = process_started(lifeline.watcher_pid);
        let path = write_server_process_record(dir.path(), &record).unwrap();

        // The pipe stays open throughout, so only reclaim can kill the group.
        let reclaim = reclaim_while_reaping(dir.path()).await;
        let watcher_status = tokio::time::timeout(Duration::from_secs(10), lifeline.watcher.wait())
            .await
            .expect("reclaim must kill the watcher-verified group")
            .unwrap();
        reclaim.await.unwrap();
        assert!(!watcher_status.success());
        let deadline = Instant::now() + Duration::from_secs(30);
        while unsafe { libc::kill(descendant_pid, 0) } == 0 && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let descendant_survived = unsafe { libc::kill(descendant_pid, 0) } == 0;
        if descendant_survived {
            unsafe { libc::kill(descendant_pid, libc::SIGKILL) };
        }
        assert!(
            !descendant_survived,
            "the model-bearing descendant must die"
        );
        assert!(!path.exists());
    }

    /// A recorded member that is still the same process but sits in another
    /// group proves nothing about the recorded PGID.
    #[tokio::test]
    async fn reclaim_never_signals_a_group_its_verified_member_has_left() {
        let dir = tempfile::tempdir().unwrap();
        let (mut member, member_group) = spawn_stand_in_group("sleep 60");
        let (mut recorded, recorded_group) = spawn_stand_in_group("sleep 60");
        let mut record = stand_in_record(recorded_group, "an exited owner".into(), None);
        record.watcher_pid = Some(member_group);
        record.watcher_started = process_started(member_group);
        assert!(record.watcher_started.is_some());
        let path = write_server_process_record(dir.path(), &record).unwrap();

        reclaim_orphaned_server_groups(dir.path());

        tokio::time::sleep(Duration::from_millis(200)).await;
        assert!(
            recorded.try_wait().unwrap().is_none(),
            "the recorded group must not be signalled on another group's member"
        );
        assert!(member.try_wait().unwrap().is_none());
        assert!(!path.exists(), "an unverifiable record is dropped");
    }

    #[test]
    fn reclaim_discards_a_corrupt_record() {
        let dir = tempfile::tempdir().unwrap();
        let corrupt = dir.path().join("12345.json");
        std::fs::write(&corrupt, b"not a record").unwrap();
        reclaim_orphaned_server_groups(dir.path());
        assert!(!corrupt.exists());
    }
}