car-server-core 0.47.0

//! Process-wide admission control for inference RPC handlers.
//!
//! ## Why this exists
//!
//! Until this module landed, every WebSocket session shared the
//! `ServerState.inference: OnceLock<Arc<InferenceEngine>>` — which is
//! correct for *model loading* (one set of weights, not N), but the
//! engine itself queues nothing. So when N users (or N
//! parallel-FFI-spawned `car infer` calls that auto-start a daemon)
//! land at once, each call enters the engine concurrently, each
//! triggers its own KV-cache allocation, each holds activations during
//! decode, and the host RAM is overwhelmed long before any single
//! request finishes.
//!
//! The fix is a global semaphore. The number of permits is sized from
//! detected host RAM — roughly "one concurrent generation per 8 GB"
//! with floor 1 and ceiling 8. Embedders and operators can override
//! via `CAR_INFERENCE_MAX_CONCURRENT`.
//!
//! Embedded streaming RPCs (`infer_stream`, `voice.transcribe_stream`,
//! …) hold their permit for the duration of the stream; one-shot RPCs
//! release on response.
//!
//! Memory-pressure-aware *eviction* of loaded weights is a separate
//! concern handled by `car_inference::backend_cache::BackendCache`.
//! This module only gates *new admissions*. Together they form a
//! two-layer defense: admission keeps concurrent activations bounded;
//! the LRU backend cache keeps loaded weights bounded.

use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{OwnedSemaphorePermit, Semaphore};

/// Env var that overrides the auto-sized permit count. Setting it to
/// `1` forces full serialization, which is the right answer on a small
/// laptop running a meeting bot.
pub const ENV_MAX_CONCURRENT: &str = "CAR_INFERENCE_MAX_CONCURRENT";

/// Override for the NON-LOCAL permit count — calls that do no local compute.
pub const ENV_MAX_CONCURRENT_REMOTE: &str = "CAR_INFERENCE_MAX_CONCURRENT_REMOTE";

/// Default ceiling for calls that consume no local RAM. Sized for network
/// concurrency and provider rate limits rather than memory: nothing here
/// allocates a KV cache on this host, so the RAM-derived figure does not apply.
const DEFAULT_REMOTE_PERMITS: usize = 16;

/// Threshold (ms) above which an acquire-wait gets logged. Tuned for
/// "this should normally be instant; surface it when it isn't."
const SLOW_ACQUIRE_LOG_MS: u128 = 100;

/// Process-wide gate on concurrent inference requests. Cheap to clone —
/// internally just an `Arc<Semaphore>` plus the chosen permit count.
#[derive(Clone)]
pub struct InferenceAdmission {
    sem: Arc<Semaphore>,
    permits: usize,
    /// Separate pool for work that does NO local compute.
    ///
    /// The single RAM-sized pool charged a remote call the same permit as a
    /// local generation, even though a remote call allocates no KV cache and
    /// holds no activations on this host. On a 16 GB machine that is ~2 permits
    /// total, so two concurrent remote calls serialized behind a limit that
    /// exists for memory pressure they never create — remote throughput was
    /// capped by local RAM (Parslee-ai/car#800).
    remote_sem: Arc<Semaphore>,
    remote_permits: usize,
}

impl InferenceAdmission {
    /// Build the controller, sizing the permit count from host RAM
    /// unless [`ENV_MAX_CONCURRENT`] is set.
    pub fn new() -> Self {
        let permits = chosen_permit_count();
        tracing::info!(
            permits,
            env = ENV_MAX_CONCURRENT,
            "inference admission controller online"
        );
        Self::with_permits(permits)
    }

    /// Build with an explicit permit count. Skips env probing — useful
    /// for embedders that already know what they want and for tests
    /// that need a deterministic cap without racing on a process-wide
    /// env var.
    pub fn with_permits(permits: usize) -> Self {
        Self::with_permits_split(permits, chosen_remote_permit_count())
    }

    /// Build with explicit local AND non-local permit counts.
    pub fn with_permits_split(permits: usize, remote_permits: usize) -> Self {
        let permits = permits.max(1);
        let remote_permits = remote_permits.max(1);
        Self {
            sem: Arc::new(Semaphore::new(permits)),
            permits,
            remote_sem: Arc::new(Semaphore::new(remote_permits)),
            remote_permits,
        }
    }

    /// Permits available for calls that do no local compute.
    pub fn remote_permits(&self) -> usize {
        self.remote_permits
    }

    /// Acquire from the pool appropriate to the work.
    ///
    /// `does_local_compute` must be TRUE whenever this host will run the model
    /// itself, and when we cannot tell. Defaulting the unknown case to the local
    /// pool is deliberate: a delegated model may be a cloud API (no local
    /// pressure) or a host driving a local llama.cpp (plenty), and they are
    /// indistinguishable to CAR. Guessing "remote" there would remove the guard
    /// from exactly the call that needs it.
    pub async fn acquire_for(&self, does_local_compute: bool) -> OwnedSemaphorePermit {
        if does_local_compute {
            return self.acquire().await;
        }
        let started = std::time::Instant::now();
        let permit = self
            .remote_sem
            .clone()
            .acquire_owned()
            .await
            .expect("remote admission semaphore is never closed");
        let waited_ms = started.elapsed().as_millis();
        if waited_ms >= SLOW_ACQUIRE_LOG_MS {
            tracing::info!(
                waited_ms,
                permits_total = self.remote_permits,
                permits_available = self.remote_sem.available_permits(),
                "remote inference request queued behind concurrency limit"
            );
        }
        permit
    }

    /// Acquire a permit. Returns an owned guard whose `Drop` releases
    /// the slot — keep it alive for the full duration of the inference
    /// call (including any token streaming).
    pub async fn acquire(&self) -> OwnedSemaphorePermit {
        let started = std::time::Instant::now();
        let permit = self
            .sem
            .clone()
            .acquire_owned()
            .await
            .expect("inference admission semaphore is never closed");
        let waited_ms = started.elapsed().as_millis();
        if waited_ms >= SLOW_ACQUIRE_LOG_MS {
            tracing::info!(
                waited_ms,
                permits_total = self.permits,
                permits_available = self.sem.available_permits(),
                "inference request queued behind concurrency limit"
            );
        }
        permit
    }

    /// Try to acquire a permit without blocking. Returns `None` when
    /// every slot is busy. Handy for non-essential paths (e.g. health
    /// probes) that prefer to fail fast over queueing.
    pub fn try_acquire(&self) -> Option<OwnedSemaphorePermit> {
        self.sem.clone().try_acquire_owned().ok()
    }

    /// Acquire with an upper bound on wait time. Returns `None` on
    /// timeout. The wait time is observed at acquisition; the caller
    /// keeps the permit for as long as it likes once granted.
    pub async fn acquire_with_timeout(&self, max_wait: Duration) -> Option<OwnedSemaphorePermit> {
        match tokio::time::timeout(max_wait, self.acquire()).await {
            Ok(permit) => Some(permit),
            Err(_) => {
                tracing::warn!(
                    max_wait_ms = max_wait.as_millis() as u64,
                    permits_total = self.permits,
                    "inference admission acquire timed out"
                );
                None
            }
        }
    }

    /// Total permits configured. Useful for status surfaces (`car-host`
    /// tray, `car daemon status`) so operators can see the cap.
    pub fn permits(&self) -> usize {
        self.permits
    }

    /// Permits currently free. Snapshot — racy by definition but
    /// sufficient for status panels.
    pub fn permits_available(&self) -> usize {
        self.sem.available_permits()
    }

    /// Inference requests currently in flight (held permits). Snapshot — racy,
    /// but the right coarse "is the machine busy" signal for deferring the
    /// proactive upgrade nudge so it never fires mid-inference.
    pub fn in_flight(&self) -> usize {
        self.permits.saturating_sub(self.sem.available_permits())
    }
}

impl Default for InferenceAdmission {
    fn default() -> Self {
        Self::new()
    }
}

/// Permit count for work that does no local compute. Not RAM-derived: nothing
/// on this path allocates a KV cache here, so the constraint is network
/// concurrency and provider rate limits, not memory.
fn chosen_remote_permit_count() -> usize {
    if let Ok(raw) = std::env::var(ENV_MAX_CONCURRENT_REMOTE) {
        if let Ok(n) = raw.trim().parse::<usize>() {
            if n >= 1 {
                return n;
            }
        }
        tracing::warn!(
            value = %raw,
            "{} must be a positive integer; ignoring",
            ENV_MAX_CONCURRENT_REMOTE
        );
    }
    DEFAULT_REMOTE_PERMITS
}

fn chosen_permit_count() -> usize {
    // Operator override wins.
    if let Ok(raw) = std::env::var(ENV_MAX_CONCURRENT) {
        if let Ok(n) = raw.trim().parse::<usize>() {
            if n >= 1 {
                return n;
            }
        }
        tracing::warn!(
            value = %raw,
            "{} must be a positive integer; ignoring and falling back to auto-sizing",
            ENV_MAX_CONCURRENT
        );
    }

    // Auto-size from host RAM. ~8 GB per concurrent generation is a
    // conservative floor for the model-size mix CAR users hit in
    // practice (Qwen3-4B + a remote model + vLLM-MLX VL); raise the
    // env var when running with smaller models or more headroom.
    let total_ram_mb = host_ram_mb();

    (total_ram_mb / 8192).clamp(1, 8) as usize
}

fn host_ram_mb() -> u64 {
    #[cfg(target_os = "macos")]
    {
        if let Ok(output) = std::process::Command::new("sysctl")
            .args(["-n", "hw.memsize"])
            .output()
        {
            if output.status.success() {
                if let Ok(s) = String::from_utf8(output.stdout) {
                    if let Ok(bytes) = s.trim().parse::<u64>() {
                        return bytes / (1024 * 1024);
                    }
                }
            }
        }
    }
    #[cfg(target_os = "linux")]
    {
        if let Ok(content) = std::fs::read_to_string("/proc/meminfo") {
            for line in content.lines() {
                if let Some(rest) = line.strip_prefix("MemTotal:") {
                    let parts: Vec<&str> = rest.split_whitespace().collect();
                    if let Some(kb_str) = parts.first() {
                        if let Ok(kb) = kb_str.parse::<u64>() {
                            return kb / 1024;
                        }
                    }
                }
            }
        }
    }
    // Windows: reuse car-inference's CRLF-aware Win32_ComputerSystem detector
    // (wmic → PowerShell) rather than hand-rolling a subprocess parse. Without
    // this arm the function fell straight through to the 16 GB default, so
    // inference concurrency was auto-sized off a fixed 16 GB regardless of the
    // box's real RAM.
    #[cfg(target_os = "windows")]
    {
        if let Some(mb) = car_inference::hardware::detect_ram_mb_windows() {
            return mb;
        }
    }
    // Final fallback: assume 16 GB so auto-sizing yields 2 permits —
    // a sane default for the laptop demographic.
    16 * 1024
}

#[cfg(test)]
mod tests {
    // Tests use `with_permits` exclusively to stay deterministic —
    // `new()` reads `CAR_INFERENCE_MAX_CONCURRENT` from a
    // process-global env, which would race when cargo test runs
    // sibling tests in parallel.
    use super::*;

    #[tokio::test]
    async fn permits_clamps_to_at_least_one() {
        let admission = InferenceAdmission::with_permits(0);
        assert_eq!(admission.permits(), 1);
    }

    #[tokio::test]
    async fn try_acquire_returns_none_when_full() {
        let admission = InferenceAdmission::with_permits(1);
        let _held = admission.acquire().await;
        assert!(admission.try_acquire().is_none());
    }

    #[tokio::test]
    async fn acquire_with_timeout_returns_none_on_full_queue() {
        let admission = InferenceAdmission::with_permits(1);
        let _held = admission.acquire().await;
        let started = std::time::Instant::now();
        let result = admission
            .acquire_with_timeout(Duration::from_millis(50))
            .await;
        assert!(result.is_none());
        assert!(started.elapsed() >= Duration::from_millis(45));
    }

    #[tokio::test]
    async fn permits_available_reflects_outstanding_holds() {
        let admission = InferenceAdmission::with_permits(2);
        assert_eq!(admission.permits_available(), 2);
        let _a = admission.acquire().await;
        assert_eq!(admission.permits_available(), 1);
        let _b = admission.acquire().await;
        assert_eq!(admission.permits_available(), 0);
    }

    #[test]
    fn host_ram_mb_returns_a_positive_value() {
        // Even on the fallback path the function must report >0 so
        // chosen_permit_count produces a usable cap.
        assert!(host_ram_mb() > 0);
    }

    /// The two pools are independent: exhausting the local one must not block a
    /// call that does no local compute. That is the whole point — a RAM-sized
    /// limit was capping remote throughput that consumes no RAM (car#800).
    #[tokio::test]
    async fn remote_work_does_not_queue_behind_the_local_pool() {
        let admission = InferenceAdmission::with_permits_split(1, 4);
        // Take the only local permit and hold it.
        let _local = admission.acquire_for(true).await;
        // A second LOCAL acquire must block; a remote one must not.
        assert!(
            admission.try_acquire().is_none(),
            "the local pool should be exhausted"
        );
        let remote = tokio::time::timeout(
            std::time::Duration::from_millis(200),
            admission.acquire_for(false),
        )
        .await;
        assert!(
            remote.is_ok(),
            "remote work queued behind a full local pool — the pools are not independent"
        );
    }

    /// The unknown case must fall to the LOCAL pool. A delegated model may be a
    /// cloud API or a host driving a local llama.cpp; they look identical to
    /// CAR, and guessing "remote" would strip the guard from the one that
    /// actually allocates memory here.
    #[tokio::test]
    async fn unknown_work_is_treated_as_local() {
        let admission = InferenceAdmission::with_permits_split(1, 4);
        let _held = admission.acquire_for(true).await;
        // `does_local_compute = true` is what callers pass when unsure, so it
        // must contend for the scarce pool rather than the roomy one.
        let blocked = tokio::time::timeout(
            std::time::Duration::from_millis(150),
            admission.acquire_for(true),
        )
        .await;
        assert!(
            blocked.is_err(),
            "an unknown-provenance call must contend for the local pool"
        );
    }

    #[test]
    fn remote_pool_is_sized_independently_of_ram() {
        let admission = InferenceAdmission::with_permits_split(1, 16);
        assert_eq!(admission.permits(), 1);
        assert_eq!(admission.remote_permits(), 16);
    }
}