memra-server 0.122.0

//! Predictive-admission SHADOW instrumentation (lane/d2-engine-gaps-20260831).
//!
//! Engine half of darklanes Arc D2 (`research/engines-kv-oversubscription-20260830/
//! SPEC-SESSION-TIERING.md`, shadow analysis `research/d2-shadow-20260831/RESULTS.md`):
//! the D2 lane measured a Mooncake-style predictive admission predictor offline and
//! named the engine state an enforcing (or live-shadow) build needs. This module
//! supplies that state and the shadow verdict; it NEVER rejects anything. Enforcement
//! is a separate, owner-ratified flip that this module only prepares receipts for.
//!
//! What lives here, by D2 gap number:
//!
//! - G2: [`AdmissionBook`]: per-model in-flight count + booked KV bytes, maintained by
//!   the worker at its single admit/retire seam and published to `/metrics` (operator
//!   scope) as `admission_inflight` / `admission_booked_bytes`. The booked figure is the
//!   engine's OWN admission charge (the request-cost estimate the VRAM gate used,
//!   i.e. the P-maxtok book the ladder implicitly reserves today).
//! - G3: [`CompletionHistory`]: rolling per-(tenant-row, model) completion lengths in
//!   fixed-size rings, with a global per-model fallback ring. Bounded memory by
//!   construction (`HISTORY_WINDOW` samples per ring, `HISTORY_ROW_CAP` tenant rows;
//!   overflow tenants simply use the global fallback). DELIBERATELY VOLATILE: the
//!   history survives nothing across restarts; it is admission-shaping state, not a
//!   ledger, and rebuilding it from live traffic after a boot is the design (the D2
//!   predictor's own 20-sample floor covers the cold window by falling back).
//! - G5: [`ShadowConfig`] + [`shadow_verdict_line`]: the `[admit-predict]` receipt,
//!   one line per request at its first decisive admission consideration, behind
//!   `MEMRA_ADMIT_PREDICT_SHADOW` (default 0 = OFF). Computes the D2 contract's
//!   P-tenant-p95 verdict (slot check first, then the KV book against
//!   `MEMRA_ADMIT_PREDICT_BUDGET_MB`) and LOGS it, joined to the request id.
//! - G6: the would-reject lines carry `retry_after_s=`, the value an enforcing 429
//!   would send, computed by [`earliest_completion_retry_s`] from the in-flight
//!   sessions' own predicted remainders and clamped to the shed contract's 1..=60 s
//!   window (`EngineError::rate_limit_after` + `retry_contract_response` are the
//!   byte-compatible response half, wired and tested but not called in shadow mode).
//! - G7: [`ShadowConfig::is_exempt`]: `MEMRA_ADMIT_PREDICT_EXEMPT_TENANTS`, the engine-side
//!   copy of the console's `INTERNAL_ENGINE_TENANTS` list (darklanes
//!   `workers/console/src/activity.ts`); that list is the source of truth and this env
//!   is set FROM it by the deployment. Exempt tenants still enter the book as load;
//!   their would-reject rows are marked `exempt=1` so the confusion matrix can report
//!   them separately as instrument noise (a 429'd probe reads as an outage).
//!
//! Admission-path budget: everything per-request here is arithmetic over fixed-size
//! state. The single percentile computation copies at most `HISTORY_WINDOW` u32 values
//! into a reusable scratch buffer (no steady-state allocation) and runs
//! `select_nth_unstable`, and it runs only while the shadow flag is armed.
//!
//! Scope boundary, stated: this instruments the PRODUCTION admission seam (the batched
//! worker loop). The experimental dsv4 door serves bs=1 on its own dedicated thread and
//! diverts in `handle_cmd` BEFORE that loop, so dsv4 requests appear in neither the
//! book nor the receipts; that door is not a serving-grade path (docs/FLAGS.md §5).

use std::collections::HashMap;

/// Samples kept per completion ring (per tenant-row x model, and per global model row).
/// The D2 report's G3 sizing ("a rolling reservoir (fixed memory, e.g. last 512
/// completions per tenant key)").
pub(crate) const HISTORY_WINDOW: usize = 512;

/// Bound on distinct (tenant-row, model) history rows. Past the cap a new tenant's
/// completions still feed the global per-model ring, and its predictions fall back to
/// P-global-p95: bounded memory beats per-tenant fidelity for row #257.
pub(crate) const HISTORY_ROW_CAP: usize = 256;

/// Tenant-p95 needs at least this much tenant history before it is trusted
/// (D2 contract §5: "floor 20 samples, else fall back to P-global-p95").
pub(crate) const HISTORY_TENANT_FLOOR: usize = 20;

/// The `+8` context slack the engine's own admission ladder adds
/// (`prompt_tokens + max_tokens + 8`); the D2 contract's `ctx_hat` mirrors it.
pub(crate) const CTX_HAT_SLACK: u64 = 8;

// ---------------------------------------------------------------------------
// G5/G6/G7: shadow configuration
// ---------------------------------------------------------------------------

/// Read-once configuration for the shadow predictor. Built at worker start;
/// the flag cannot arm or disarm mid-process.
pub(crate) struct ShadowConfig {
    /// `MEMRA_ADMIT_PREDICT_SHADOW` != "0"/unset. OFF by design: shadow logging is
    /// a measurement act, and unmeasured behavior does not default ON.
    pub armed: bool,
    /// `MEMRA_ADMIT_PREDICT_BUDGET_MB`: the box KV budget arm the KV check compares
    /// against, in bytes. This is DEPLOYMENT arithmetic (card minus warm floor minus
    /// margin, darklanes LANE.md), not something the engine can honestly derive; when
    /// unset the KV arm never fires (`verdict` can only be admit/reject-slot) and the
    /// line says `budget_bytes=unset` so offline analysis can apply any arm to the
    /// logged books.
    pub budget_bytes: Option<u64>,
    /// `MEMRA_ADMIT_PREDICT_EXEMPT_TENANTS`: comma-separated tenant ids, sourced from the
    /// console's `INTERNAL_ENGINE_TENANTS`. Matching is on the tenant ROW identity
    /// (`auth::meter_key` with any `t:` prefix stripped).
    exempt: Vec<String>,
}

impl ShadowConfig {
    pub(crate) fn from_env() -> Self {
        let armed = std::env::var("MEMRA_ADMIT_PREDICT_SHADOW").is_ok_and(|v| v != "0");
        let budget_bytes = std::env::var("MEMRA_ADMIT_PREDICT_BUDGET_MB")
            .ok()
            .and_then(|v| v.parse::<u64>().ok())
            .map(|mb| mb.saturating_mul(1 << 20));
        let exempt = std::env::var("MEMRA_ADMIT_PREDICT_EXEMPT_TENANTS")
            .ok()
            .map(|v| Self::parse_exempt(&v))
            .unwrap_or_default();
        if armed {
            eprintln!(
                "[admit-predict] shadow armed: budget_bytes={} exempt_tenants={} \
                 (logging only, nothing is rejected; enforcement is a separate flip)",
                budget_bytes.map_or("unset".into(), |b| b.to_string()),
                exempt.len(),
            );
        }
        ShadowConfig {
            armed,
            budget_bytes,
            exempt,
        }
    }

    fn parse_exempt(raw: &str) -> Vec<String> {
        raw.split(',')
            .map(str::trim)
            .filter(|t| !t.is_empty())
            .map(str::to_string)
            .collect()
    }

    /// G7: whether this tenant row is enforcement-exempt. `tenant_row` is
    /// `auth::meter_key(cache_ns)`: `t:<tenant>` under a keyring, the raw salt
    /// otherwise; the console list carries bare tenant ids, so the `t:` prefix is
    /// stripped before comparison. Exempt tenants still occupy the book as load.
    pub(crate) fn is_exempt(&self, tenant_row: &str) -> bool {
        let bare = tenant_row.strip_prefix("t:").unwrap_or(tenant_row);
        self.exempt.iter().any(|t| t == bare)
    }
}

// ---------------------------------------------------------------------------
// G2: the in-flight book
// ---------------------------------------------------------------------------

#[derive(Default, Clone, Copy)]
struct ModelBook {
    inflight: u64,
    /// Sum of in-flight sessions' engine admission charges (the request-cost
    /// estimate the VRAM gate used at each session's admit).
    booked_bytes: u64,
    /// Sum of in-flight sessions' SHADOW kv_hat charges (P-tenant-p95 book).
    /// Zero for every session admitted while the shadow flag is off.
    shadow_booked_bytes: u64,
}

/// Per-model in-flight count + booked KV bytes, owned by the worker loop (single
/// thread, no locks) and snapshotted into `Metrics` on the publish tick. One admit
/// and one retire seam keep it exact: `active.push` books, `active.remove` unbooks
/// (a step-OOM park unbooks too, since its KV is dropped, and its replay re-books at
/// re-admission).
#[derive(Default)]
pub(crate) struct AdmissionBook {
    models: HashMap<String, ModelBook>,
}

impl AdmissionBook {
    pub(crate) fn admit(&mut self, model: &str, booked_bytes: u64, shadow_kv_hat: u64) {
        let row = self.models.entry(model.to_string()).or_default();
        row.inflight += 1;
        row.booked_bytes = row.booked_bytes.saturating_add(booked_bytes);
        row.shadow_booked_bytes = row.shadow_booked_bytes.saturating_add(shadow_kv_hat);
    }

    pub(crate) fn retire(&mut self, model: &str, booked_bytes: u64, shadow_kv_hat: u64) {
        if let Some(row) = self.models.get_mut(model) {
            row.inflight = row.inflight.saturating_sub(1);
            row.booked_bytes = row.booked_bytes.saturating_sub(booked_bytes);
            row.shadow_booked_bytes = row.shadow_booked_bytes.saturating_sub(shadow_kv_hat);
        }
    }

    pub(crate) fn inflight(&self, model: &str) -> u64 {
        self.models.get(model).map_or(0, |row| row.inflight)
    }

    /// The shadow KV book the D2 decision rule sums: EVERY in-flight request's own
    /// admission-time kv_hat, across all models (they share the card's budget arm).
    pub(crate) fn shadow_booked_total(&self) -> u64 {
        self.models
            .values()
            .map(|row| row.shadow_booked_bytes)
            .sum()
    }

    /// G2 metrics snapshots (operator scope): per-model in-flight and booked bytes.
    pub(crate) fn inflight_snapshot(&self) -> HashMap<String, u64> {
        self.models
            .iter()
            .map(|(model, row)| (model.clone(), row.inflight))
            .collect()
    }

    pub(crate) fn booked_snapshot(&self) -> HashMap<String, u64> {
        self.models
            .iter()
            .map(|(model, row)| (model.clone(), row.booked_bytes))
            .collect()
    }
}

// ---------------------------------------------------------------------------
// G3: rolling completion-length history
// ---------------------------------------------------------------------------

struct Ring {
    buf: Vec<u32>,
    next: usize,
    len: usize,
}

impl Ring {
    fn new() -> Self {
        Ring {
            // Allocated at row creation (once per tenant-row x model), never per request.
            buf: vec![0; HISTORY_WINDOW],
            next: 0,
            len: 0,
        }
    }

    fn push(&mut self, sample: u32) {
        self.buf[self.next] = sample;
        self.next = (self.next + 1) % HISTORY_WINDOW;
        self.len = (self.len + 1).min(HISTORY_WINDOW);
    }
}

/// Rolling per-(tenant-row, model) completion-length state (D2 gap G3). Fed at session
/// retirement with the session's generated token count; consulted (only while the
/// shadow flag is armed) for the causal tenant p95. Restart-volatile BY DESIGN; see
/// the module doc.
#[derive(Default)]
pub(crate) struct CompletionHistory {
    rows: HashMap<(String, String), Ring>,
    global: HashMap<String, Ring>,
    /// Reusable percentile scratch: percentile() copies a ring here and selects in
    /// place, so the admission path allocates nothing in steady state.
    scratch: Vec<u32>,
}

/// Which estimator arm produced the predicted completion length: the `reason=` field
/// of the receipt line.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LhatSource {
    /// Per-(tenant, model) causal p95, >= HISTORY_TENANT_FLOOR samples.
    TenantP95,
    /// Global per-model causal p95 (tenant history below the floor).
    GlobalP95,
    /// No usable history at all: fall back to the request's own completion bound
    /// (P-maxtok, today's implicit reservation), the conservative cold-start arm.
    MaxtokFallback,
}

impl LhatSource {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            LhatSource::TenantP95 => "tenant-p95",
            LhatSource::GlobalP95 => "global-p95",
            LhatSource::MaxtokFallback => "maxtok-fallback",
        }
    }
}

impl CompletionHistory {
    /// Record a retired session's completion length. Called once per retirement
    /// (never per token). Only terminal completions feed the history: the caller
    /// excludes step-OOM parks (they re-admit; nothing completed) and client aborts
    /// (a disconnect length is not a completion length; the D2 offline history was
    /// built from completed rows only).
    pub(crate) fn record(&mut self, tenant_row: &str, model: &str, completion_tokens: u32) {
        self.global
            .entry(model.to_string())
            .or_insert_with(Ring::new)
            .push(completion_tokens);
        let key = (tenant_row.to_string(), model.to_string());
        if let Some(ring) = self.rows.get_mut(&key) {
            ring.push(completion_tokens);
            return;
        }
        if self.rows.len() < HISTORY_ROW_CAP {
            self.rows
                .entry(key)
                .or_insert_with(Ring::new)
                .push(completion_tokens);
        }
        // At the row cap: the sample lives in the global ring only, and this tenant
        // predicts through the global fallback. Bounded memory is the contract.
    }

    /// Nearest-rank p95 of a ring (rank = ceil(0.95 * n), 1-based), via
    /// select_nth_unstable on the reusable scratch buffer. O(n), no allocation in
    /// steady state.
    fn p95(scratch: &mut Vec<u32>, ring: &Ring) -> u32 {
        scratch.clear();
        scratch.extend_from_slice(&ring.buf[..ring.len]);
        let n = scratch.len();
        debug_assert!(n > 0, "p95 caller checks emptiness");
        let rank = ((n as f64) * 0.95).ceil() as usize; // 1-based nearest rank
        let idx = rank.clamp(1, n) - 1;
        let (_, value, _) = scratch.select_nth_unstable(idx);
        *value
    }

    /// The D2 P-tenant-p95 estimator: tenant p95 with >= HISTORY_TENANT_FLOOR samples,
    /// else global per-model p95, else the request's own completion bound; always
    /// clipped to that bound when one exists.
    pub(crate) fn lhat(
        &mut self,
        tenant_row: &str,
        model: &str,
        max_tokens: Option<u64>,
    ) -> (u64, LhatSource) {
        let key = (tenant_row.to_string(), model.to_string());
        let predicted = match self.rows.get(&key) {
            Some(ring) if ring.len >= HISTORY_TENANT_FLOOR => Some((
                Self::p95(&mut self.scratch, ring) as u64,
                LhatSource::TenantP95,
            )),
            _ => match self.global.get(model) {
                Some(ring) if ring.len > 0 => Some((
                    Self::p95(&mut self.scratch, ring) as u64,
                    LhatSource::GlobalP95,
                )),
                _ => None,
            },
        };
        match (predicted, max_tokens) {
            (Some((p95, source)), Some(bound)) => (p95.min(bound), source),
            (Some((p95, source)), None) => (p95, source),
            // Cold start with a bound: book the bound (P-maxtok, the engine's own
            // implicit reservation today). Cold start without one: nothing honest to
            // book; 0 keeps the line explicit (reason=maxtok-fallback, predicted 0).
            (None, bound) => (bound.unwrap_or(0), LhatSource::MaxtokFallback),
        }
    }
}

// ---------------------------------------------------------------------------
// G5: the verdict
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Verdict {
    Admit,
    /// In-flight on this model already holds the session cap (D2:
    /// `would-reject(no-decode-slot-at-completion)`).
    RejectSlot,
    /// Predicted KV book (in-flight kv_hats + this request's kv_hat) exceeds the
    /// configured budget arm (D2: `would-reject(kv-budget-exceeded)`).
    RejectKv,
}

impl Verdict {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Verdict::Admit => "admit",
            Verdict::RejectSlot => "reject-slot",
            Verdict::RejectKv => "reject-kv",
        }
    }
}

/// The D2 contract's per-request KV charge:
/// `kv_hat = B/token x (prompt + L_hat + slack) + fixed`.
pub(crate) fn kv_hat(
    prompt_tokens: u64,
    predicted_completion: u64,
    bytes_per_token: u64,
    fixed_bytes: u64,
) -> u64 {
    bytes_per_token
        .saturating_mul(
            prompt_tokens
                .saturating_add(predicted_completion)
                .saturating_add(CTX_HAT_SLACK),
        )
        .saturating_add(fixed_bytes)
}

/// G6 (shadow half): the Retry-After an enforcing reject would send: the earliest
/// predicted in-flight completion, from each session's own shadow book (predicted
/// total minus generated so far) at the current p50 step latency. Clamped to the shed
/// contract's 1..=60 s window (`retry_contract_response` applies the same clamp on the
/// wire, so the logged value IS the header value). None when nothing is in flight or
/// no latency estimate exists yet (an enforcing path would fall back to the class
/// default (RETRY_AFTER_S_RATE_LIMIT) there, and the line logs `retry_after_s=-`.
pub(crate) fn earliest_completion_retry_s(
    inflight: impl Iterator<Item = (u64, u64)>, // (shadow predicted total, generated so far)
    p50_step_ms: f32,
) -> Option<u64> {
    if p50_step_ms <= 0.0 {
        return None;
    }
    inflight
        .map(|(predicted_total, generated)| predicted_total.saturating_sub(generated))
        .min()
        .map(|remaining_tokens| {
            let secs = (remaining_tokens as f64 * p50_step_ms as f64 / 1000.0).ceil() as u64;
            secs.clamp(1, 60)
        })
}

/// Everything one receipt line says. Assembled by the worker at the request's first
/// decisive admission consideration; formatted by [`shadow_verdict_line`].
pub(crate) struct VerdictLine<'a> {
    pub request_id: &'a str,
    pub tenant_row: &'a str,
    pub model: &'a str,
    pub verdict: Verdict,
    /// Which estimator arm produced `predicted_completion` (the decomposed "why":
    /// the verdict already carries the reject reason per the D2 contract).
    pub reason: LhatSource,
    pub prompt_tokens: Option<u64>,
    pub predicted_completion: u64,
    /// This request's own kv_hat charge (None on slot rejects logged before tokenize).
    pub kv_hat_bytes: Option<u64>,
    /// The in-flight shadow book at decision time, EXCLUDING this request.
    pub booked_bytes: u64,
    pub inflight: u64,
    pub cap: u64,
    pub budget_bytes: Option<u64>,
    /// What an enforcing 429 would have sent (would-reject rows only).
    pub retry_after_s: Option<u64>,
    pub exempt: bool,
}

/// One grep-stable receipt line, `[admit-predict]`-prefixed, all fields `key=value`.
/// The confusion matrix is a grep, not a reconstruction (D2 G5).
pub(crate) fn shadow_verdict_line(line: &VerdictLine<'_>) -> String {
    format!(
        "[admit-predict] id={} tenant={:?} model={:?} verdict={} reason={} prompt={} \
         predicted_completion={} kv_hat={} booked_bytes={} inflight={} cap={} \
         budget_bytes={} retry_after_s={} exempt={}",
        line.request_id,
        line.tenant_row,
        line.model,
        line.verdict.as_str(),
        line.reason.as_str(),
        line.prompt_tokens.map_or("-".into(), |v| v.to_string()),
        line.predicted_completion,
        line.kv_hat_bytes.map_or("-".into(), |v| v.to_string()),
        line.booked_bytes,
        line.inflight,
        line.cap,
        line.budget_bytes.map_or("unset".into(), |v| v.to_string()),
        line.retry_after_s.map_or("-".into(), |v| v.to_string()),
        u8::from(line.exempt),
    )
}

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

    fn line(verdict: Verdict) -> VerdictLine<'static> {
        VerdictLine {
            request_id: "chatcmpl-abc123",
            tenant_row: "t:acme",
            model: "qwen/qwen3.8-27b",
            verdict,
            reason: LhatSource::TenantP95,
            prompt_tokens: Some(1200),
            predicted_completion: 2906,
            kv_hat_bytes: Some(130_000_000),
            booked_bytes: 9_000_000_000,
            inflight: 7,
            cap: 32,
            budget_bytes: Some(35_423 << 20),
            retry_after_s: Some(4),
            exempt: false,
        }
    }

    /// Locks the receipt's field NAMES and joinability: request id, verdict, reason,
    /// predicted_completion, booked_bytes (the fields the D2 confusion-matrix grep
    /// keys on), plus the shadow-Retry-After and exemption markers.
    #[test]
    fn verdict_line_locks_fields() {
        let s = shadow_verdict_line(&line(Verdict::RejectKv));
        assert!(s.starts_with("[admit-predict] "), "grep-stable prefix: {s}");
        for field in [
            "id=chatcmpl-abc123",
            "tenant=\"t:acme\"",
            "model=\"qwen/qwen3.8-27b\"",
            "verdict=reject-kv",
            "reason=tenant-p95",
            "prompt=1200",
            "predicted_completion=2906",
            "kv_hat=130000000",
            "booked_bytes=9000000000",
            "inflight=7",
            "cap=32",
            "budget_bytes=37143707648",
            "retry_after_s=4",
            "exempt=0",
        ] {
            assert!(s.contains(field), "line must carry `{field}`: {s}");
        }
        // Exactly one line: a multi-line receipt breaks the grep contract.
        assert_eq!(s.lines().count(), 1);
    }

    #[test]
    fn verdict_line_optional_fields_render_placeholders() {
        let mut l = line(Verdict::RejectSlot);
        l.prompt_tokens = None;
        l.kv_hat_bytes = None;
        l.budget_bytes = None;
        l.retry_after_s = None;
        l.exempt = true;
        let s = shadow_verdict_line(&l);
        for field in [
            "verdict=reject-slot",
            "prompt=-",
            "kv_hat=-",
            "budget_bytes=unset",
            "retry_after_s=-",
            "exempt=1",
        ] {
            assert!(s.contains(field), "line must carry `{field}`: {s}");
        }
    }

    #[test]
    fn book_admit_retire_round_trip() {
        let mut book = AdmissionBook::default();
        book.admit("m", 100, 40);
        book.admit("m", 50, 10);
        book.admit("other", 7, 3);
        assert_eq!(book.inflight("m"), 2);
        assert_eq!(book.booked_snapshot()["m"], 150);
        assert_eq!(book.inflight_snapshot()["m"], 2);
        // The shadow decision sums across models (one card, one budget arm).
        assert_eq!(book.shadow_booked_total(), 53);
        book.retire("m", 100, 40);
        assert_eq!(book.inflight("m"), 1);
        assert_eq!(book.booked_snapshot()["m"], 50);
        assert_eq!(book.shadow_booked_total(), 13);
        book.retire("m", 50, 10);
        book.retire("other", 7, 3);
        assert_eq!(book.inflight("m"), 0);
        assert_eq!(book.shadow_booked_total(), 0);
        // Underflow is saturating, never a panic in the serving loop.
        book.retire("m", 1, 1);
        assert_eq!(book.inflight("m"), 0);
    }

    #[test]
    fn history_tenant_floor_then_tenant_p95() {
        let mut h = CompletionHistory::default();
        // Below the 20-sample floor: global fallback.
        for i in 0..HISTORY_TENANT_FLOOR - 1 {
            h.record("t:a", "m", 100 + i as u32);
        }
        h.record("t:b", "m", 9_000); // global ring gains a whale
        let (lhat, source) = h.lhat("t:a", "m", None);
        assert_eq!(source, LhatSource::GlobalP95);
        assert!(lhat >= 118, "global p95 sees the whale: {lhat}");
        // At the floor: tenant p95 (nearest rank ceil(0.95 x 20) = 19 -> the 19th of
        // 20 sorted samples), so the tenant's own single outlier sits above it.
        h.record("t:a", "m", 200);
        let (lhat, source) = h.lhat("t:a", "m", None);
        assert_eq!(source, LhatSource::TenantP95);
        assert_eq!(lhat, 118, "nearest-rank p95 of {{100..=118, 200}}");
    }

    #[test]
    fn history_clips_to_max_tokens_and_cold_start_falls_back() {
        let mut h = CompletionHistory::default();
        for _ in 0..HISTORY_TENANT_FLOOR {
            h.record("t:a", "m", 5_000);
        }
        let (lhat, source) = h.lhat("t:a", "m", Some(256));
        assert_eq!(
            (lhat, source),
            (256, LhatSource::TenantP95),
            "clip to bound"
        );
        // No history at all: the request's own bound (P-maxtok arm), else 0.
        let (lhat, source) = h.lhat("t:zzz", "unknown-model", Some(1024));
        assert_eq!((lhat, source), (1024, LhatSource::MaxtokFallback));
        let (lhat, source) = h.lhat("t:zzz", "unknown-model", None);
        assert_eq!((lhat, source), (0, LhatSource::MaxtokFallback));
    }

    #[test]
    fn history_ring_is_rolling_and_rows_are_bounded() {
        let mut h = CompletionHistory::default();
        // Overfill one ring: old samples must age out.
        for _ in 0..HISTORY_WINDOW {
            h.record("t:a", "m", 10);
        }
        for _ in 0..HISTORY_WINDOW {
            h.record("t:a", "m", 20);
        }
        let (lhat, _) = h.lhat("t:a", "m", None);
        assert_eq!(lhat, 20, "the whole window rolled over");
        // Row cap: row #HISTORY_ROW_CAP+1 gets no per-tenant ring but still feeds global.
        let mut h = CompletionHistory::default();
        for i in 0..HISTORY_ROW_CAP {
            h.record(&format!("t:{i}"), "m", 100);
        }
        assert_eq!(h.rows.len(), HISTORY_ROW_CAP);
        h.record("t:overflow", "m", 100);
        assert_eq!(h.rows.len(), HISTORY_ROW_CAP, "row map is bounded");
        for _ in 0..HISTORY_TENANT_FLOOR + 5 {
            h.record("t:overflow", "m", 300);
        }
        let (_, source) = h.lhat("t:overflow", "m", None);
        assert_eq!(
            source,
            LhatSource::GlobalP95,
            "an over-cap tenant predicts through the global arm"
        );
    }

    #[test]
    fn kv_hat_matches_contract_arithmetic() {
        // kv_hat = B/token x (p + L + 8) + fixed, the D2 §4 decision rule.
        assert_eq!(kv_hat(100, 50, 1_000, 7), 1_000 * (100 + 50 + 8) + 7);
        // Saturates rather than overflowing on absurd inputs.
        assert_eq!(kv_hat(u64::MAX, u64::MAX, 2, 1), u64::MAX);
    }

    #[test]
    fn retry_hint_is_earliest_completion_clamped_to_shed_window() {
        // Two in-flight: 100 remaining and 5 remaining; p50 200 ms/step.
        let hint = earliest_completion_retry_s([(150, 50), (30, 25)].into_iter(), 200.0);
        assert_eq!(hint, Some(1), "5 tokens x 0.2 s = 1 s ceil");
        let hint = earliest_completion_retry_s([(10_000, 0)].into_iter(), 200.0);
        assert_eq!(hint, Some(60), "clamped to the shed contract's 60 s max");
        assert_eq!(earliest_completion_retry_s(std::iter::empty(), 200.0), None);
        assert_eq!(
            earliest_completion_retry_s([(10, 0)].into_iter(), 0.0),
            None,
            "no latency estimate yet"
        );
    }

    #[test]
    fn exempt_matching_strips_tenant_row_prefix() {
        let cfg = ShadowConfig {
            armed: true,
            budget_bytes: None,
            exempt: ShadowConfig::parse_exempt("watchdog-orn, ten_UBB1F6Lf ,,orn-probe-dry"),
        };
        assert!(cfg.is_exempt("t:watchdog-orn"), "keyring row form");
        assert!(cfg.is_exempt("ten_UBB1F6Lf"), "bare row form");
        assert!(cfg.is_exempt("orn-probe-dry"));
        assert!(!cfg.is_exempt("t:acme"));
        assert!(!cfg.is_exempt(""));
    }
}