car-inference 0.62.0

//! Who pays for a model call, and the gate every dispatch passes
//! (`docs/proposals/spend-caps.md`, "The gate at each entry point").
//!
//! Three things live here:
//!
//! - [`SpendScope`]: the principal a request's model calls are charged to,
//!   the meter keys they charge, the cause, and the run. The daemon sets it
//!   once per request, from the session, and the engine reads it from a
//!   task-local at each dispatch site. Work a handler spawns with
//!   `tokio::spawn` does not inherit a task-local, so it goes through
//!   [`spawn_scoped`] instead, which carries the scope across.
//! - [`SpendGate`]: the trait the daemon implements over its spend meter.
//!   The meter lives in car-registry, which this crate cannot depend on, so
//!   the keys travel as their stable ids (`agent:<id>`, `operator`, …) and
//!   the daemon parses them.
//! - [`SpendTicket`]: an admitted call. It settles on every way the call
//!   ends — a result, an error after dispatch, a timeout, a cancel, a
//!   dropped future — with the usage seen so far, else the pre-dispatch
//!   estimate; and it charges nothing for an error raised before dispatch
//!   ([`InferenceError::raised_before_dispatch`]).
//!
//! The gate is consulted at the engine's own dispatch sites, not in the RPC
//! handlers, because only the engine sees the model actually used: a router
//! fallback, the auto-thinking second dispatch and a substituted model are
//! each admitted and charged as their own call.
//!
//! Nothing here refuses a call on spend yet: every admission answers yes.
//! What is refused is a dispatch with no scope when the engine was told to
//! require one ([`InferenceEngine::set_spend_gate`](crate::InferenceEngine::set_spend_gate)
//! with `require_scope`), so a path that forgot its scope is loud rather than
//! quietly uncounted.

use std::collections::{HashMap, VecDeque};
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde::{Deserialize, Serialize};

use crate::billing::{self, CallCost, CallKind, CallUsage, CostSource, PriceKnown};
use crate::registry::UnifiedRegistry;
use crate::{InferenceError, TokenUsage};

/// Who a model call is charged to. Mirrors the meter's key kinds; `key()` is
/// the meter key id the daemon parses back.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(tag = "kind", content = "id", rename_all = "snake_case")]
pub enum SpendPrincipal {
    /// A supervised or outside agent, bound by `session.auth`.
    Agent(String),
    /// One worker instance, by its handle.
    Worker(String),
    /// An outside credential's own calls.
    Credential(String),
    /// A daemon-token or host session with no agent.
    Operator,
    /// CAR's own background work, by purpose (`scheduled`, `selfheal`, …).
    Car(String),
    /// A declarative agent run inside the daemon.
    Declarative(String),
}

impl SpendPrincipal {
    /// The meter key id: `agent:<id>`, `worker:<handle>`, `credential:<id>`,
    /// `operator`, `car:<purpose>`, `declarative:<id>`.
    pub fn key(&self) -> String {
        match self {
            Self::Agent(id) => format!("agent:{id}"),
            Self::Worker(handle) => format!("worker:{handle}"),
            Self::Credential(id) => format!("credential:{id}"),
            Self::Operator => "operator".to_string(),
            Self::Car(purpose) => format!("car:{purpose}"),
            Self::Declarative(id) => format!("declarative:{id}"),
        }
    }

    /// The kind, as cost views name it.
    pub const fn kind(&self) -> &'static str {
        match self {
            Self::Agent(_) => "agent",
            Self::Worker(_) => "worker",
            Self::Credential(_) => "credential",
            Self::Operator => "operator",
            Self::Car(_) => "car",
            Self::Declarative(_) => "declarative",
        }
    }
}

/// Why CAR made a model call on a principal's behalf. `Direct` is the
/// request's own call; the others are calls CAR makes inside it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum SpendCause {
    #[default]
    Direct,
    Chat,
    MemoryEmbedding,
    Compaction,
    Summary,
    Reflection,
    HelperAgent,
    GoalVerification,
    /// A router retry or failover attempt after the first.
    Fallback,
    Evolution,
}

impl SpendCause {
    /// Stable wire spelling.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Direct => "direct",
            Self::Chat => "chat",
            Self::MemoryEmbedding => "memory_embedding",
            Self::Compaction => "compaction",
            Self::Summary => "summary",
            Self::Reflection => "reflection",
            Self::HelperAgent => "helper_agent",
            Self::GoalVerification => "goal_verification",
            Self::Fallback => "fallback",
            Self::Evolution => "evolution",
        }
    }
}

/// One settled call, as the request that made it can read it back
/// ([`receipts`]): what the record (`InferenceMetered`) carries.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SpendReceipt {
    /// The meter's call id, on the principal's ledger.
    pub call_id: String,
    pub principal: SpendPrincipal,
    pub cause: SpendCause,
    pub model_used: String,
    /// Dollars, when the cost is known.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cost_usd: Option<f64>,
    pub cost_source: CostSource,
    pub input_tokens: u64,
    pub output_tokens: u64,
    /// The call ended with an error raised before any dispatch: it charged
    /// nothing and reached no provider.
    #[serde(default)]
    pub voided: bool,
    /// The engine's outcome trace of the attempt (`InferenceResult::trace_id`,
    /// `TrackedStream::trace_id`), when the site has one: how a record finds
    /// the receipt of its own call rather than the newest in its scope.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub trace_id: Option<String>,
}

/// How many receipts a scope keeps. A request scope sees a handful; a
/// long-lived one (a coder session, a chat) would otherwise grow without
/// bound, and a record only ever looks up its own call's.
pub const RECEIPTS_KEPT: usize = 64;

/// The principal a request's model calls are charged to.
#[derive(Debug, Clone)]
pub struct SpendScope {
    pub principal: SpendPrincipal,
    /// Every meter key the call is charged to, by id, the principal's own
    /// included. The daemon resolves them (a worker's labels, an
    /// orchestrator); the engine passes them through.
    pub keys: Vec<String>,
    pub cause: SpendCause,
    /// The run the request belongs to, when the session has one.
    pub run: Option<String>,
    /// Decision class declared by this call's caller, shared with the ledger.
    pub decision_class: Option<String>,
    /// Refuse every remote dispatch under this scope. Set for work nobody is
    /// watching (the self-healing cadence), so it can only ever run on the
    /// operator's own hardware: a pin, a router fallback and a failover alike
    /// are refused at admission, before a request leaves CAR.
    pub local_only: bool,
    receipts: Arc<Mutex<VecDeque<SpendReceipt>>>,
}

impl SpendScope {
    /// A scope charging `principal`'s own key only.
    pub fn new(principal: SpendPrincipal) -> Self {
        let keys = vec![principal.key()];
        Self {
            principal,
            keys,
            cause: SpendCause::Direct,
            run: None,
            decision_class: None,
            local_only: false,
            receipts: Arc::default(),
        }
    }

    /// This scope, refusing every remote dispatch. See [`Self::local_only`].
    pub fn local_only(mut self) -> Self {
        self.local_only = true;
        self
    }

    /// A scope charging `keys` as well as the principal's own.
    pub fn with_keys(mut self, keys: impl IntoIterator<Item = String>) -> Self {
        for key in keys {
            if !self.keys.contains(&key) {
                self.keys.push(key);
            }
        }
        self
    }

    pub fn with_cause(mut self, cause: SpendCause) -> Self {
        self.cause = cause;
        self
    }

    pub fn with_run(mut self, run: Option<String>) -> Self {
        self.run = run;
        self
    }

    pub fn with_decision_class(mut self, class: Option<String>) -> Self {
        if class.is_some() {
            self.decision_class = class;
        }
        self
    }

    /// The receipts of the calls settled under this scope so far (the newest
    /// [`RECEIPTS_KEPT`]), in order. Shared by every clone of the scope (a
    /// re-caused scope, a spawned task), so a handler reads what its whole
    /// request spent.
    pub fn receipts(&self) -> Vec<SpendReceipt> {
        lock(&self.receipts).iter().cloned().collect()
    }

    /// The receipt of the call whose engine trace is `trace_id`, if it has
    /// settled under this scope: the one a record of that call carries.
    pub fn receipt_for_trace(&self, trace_id: &str) -> Option<SpendReceipt> {
        lock(&self.receipts)
            .iter()
            .rev()
            .find(|receipt| receipt.trace_id.as_deref() == Some(trace_id))
            .cloned()
    }

    fn push_receipt(&self, receipt: SpendReceipt) {
        let mut receipts = lock(&self.receipts);
        if receipts.len() >= RECEIPTS_KEPT {
            receipts.pop_front();
        }
        receipts.push_back(receipt);
    }
}

fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    mutex
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

tokio::task_local! {
    static SCOPE: SpendScope;
    /// While one admitted call's request runs ([`dispatched`]): where the
    /// backend notes that it wrote the request and read no answer.
    static SENT_UNREAD: Arc<std::sync::atomic::AtomicBool>;
}

/// The backend wrote the current call's request and then read no answer: a
/// timeout or a reset after the write, or a 200 whose body never arrived.
/// The provider may be running and billing it, so the call's ticket settles
/// it as cut off ([`SpendTicket::cut_off`]). A failure before the write (a
/// connection that never opened, a request that could not be built) is never
/// noted. Nothing happens outside [`dispatched`].
pub(crate) fn note_sent_unread() {
    let _ = SENT_UNREAD.try_with(|sent| sent.store(true, std::sync::atomic::Ordering::Relaxed));
}

/// Run `future` as one dispatch, and say whether its backend noted the
/// request sent and unread ([`note_sent_unread`]).
#[cfg(test)]
pub(crate) async fn noting_sent_unread<F: Future>(future: F) -> (F::Output, bool) {
    let sent = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let output = SENT_UNREAD.scope(Arc::clone(&sent), future).await;
    (output, sent.load(std::sync::atomic::Ordering::Relaxed))
}

/// Run `dispatch`, the request of the call `ticket` admitted, so that a
/// backend's [`note_sent_unread`] reaches that ticket and no other.
pub(crate) async fn dispatched<F: Future>(ticket: Option<&SpendTicket>, dispatch: F) -> F::Output {
    match ticket {
        Some(ticket) => {
            crate::late_cost::attempt_scope(
                ticket.late_attempt,
                SENT_UNREAD.scope(Arc::clone(&ticket.sent_unread), dispatch),
            )
            .await
        }
        None => dispatch.await,
    }
}

/// Run `future` with `scope` as the principal of every model call made
/// inside it.
pub async fn scope<F: Future>(mut scope: SpendScope, future: F) -> F::Output {
    if scope.decision_class.is_none() {
        scope.decision_class = current().and_then(|parent| parent.decision_class);
    }
    SCOPE.scope(scope, future).await
}

/// Run `future` with the current scope's cause replaced by `cause`: a call
/// CAR makes on the principal's behalf names why. Outside any scope the
/// future runs unchanged.
pub async fn with_cause<F: Future>(cause: SpendCause, future: F) -> F::Output {
    match current() {
        Some(scope) => SCOPE.scope(scope.with_cause(cause), future).await,
        None => future.await,
    }
}

/// Carry the same declared class through model calls made by a decision.
pub async fn with_decision_class<F: Future>(class: Option<String>, future: F) -> F::Output {
    match current() {
        Some(scope) => SCOPE.scope(scope.with_decision_class(class), future).await,
        None => future.await,
    }
}

/// The scope of the current task, if a request established one.
pub fn current() -> Option<SpendScope> {
    SCOPE.try_with(|scope| scope.clone()).ok()
}

/// The receipts of the current scope so far ([`SpendScope::receipts`]);
/// empty outside any scope.
pub fn receipts() -> Vec<SpendReceipt> {
    current().map(|scope| scope.receipts()).unwrap_or_default()
}

/// The current scope's receipt for the call traced as `trace_id`
/// ([`SpendScope::receipt_for_trace`]); `None` outside any scope, before the
/// call settled, or for a site that records no trace.
pub fn receipt_for_trace(trace_id: &str) -> Option<SpendReceipt> {
    current().and_then(|scope| scope.receipt_for_trace(trace_id))
}

/// `tokio::spawn`, carrying the current spend scope into the task. A task
/// spawned any other way has no scope, and a model call it makes is refused
/// where a scope is required.
pub fn spawn_scoped<F>(future: F) -> tokio::task::JoinHandle<F::Output>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,
{
    match current() {
        Some(scope) => tokio::spawn(SCOPE.scope(scope, future)),
        None => tokio::spawn(future),
    }
}

/// Run `future` under the current scope if there is one, else unchanged.
/// For a future handed to something other than `tokio::spawn` (a `JoinSet`,
/// a `spawn_blocking` runtime handle) that must still carry the scope.
pub fn scoped<F: Future>(future: F) -> impl Future<Output = F::Output> {
    let scope = current();
    async move {
        match scope {
            Some(scope) => SCOPE.scope(scope, future).await,
            None => future.await,
        }
    }
}

/// What the engine knows about a call before it leaves CAR.
#[derive(Debug, Clone)]
pub struct DispatchFacts {
    /// The resolved model, as the catalog names it.
    pub model: String,
    pub provider: String,
    /// `ModelSchema::is_local`: counts $0, never trips a cap.
    pub is_local: bool,
    /// Whether the cost will be known once the call ends.
    pub price: PriceKnown,
    pub kind: CallKind,
    /// Input tokens (or audio seconds, or characters) estimated before
    /// dispatch.
    pub input_estimate: u64,
    /// The most output the call may produce (the request's `max_tokens`,
    /// else the model's maximum); 0 for a call priced by its input alone.
    pub max_output: u64,
    /// The estimated cost at the known price: an upper bound for a
    /// token-priced call. Unknown when the price is.
    pub estimate: CallCost,
}

/// Why a call was not admitted: `budget_reached`, `price_unknown`,
/// `spend_unrecordable` from the meter, or `no_scope` from the engine.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpendRefusal {
    /// A stable reason: `budget_reached`, `price_unknown`,
    /// `spend_unrecordable`, `no_scope`.
    pub reason: String,
    /// The text the caller sees ("budget reached until 3:00 PM EDT").
    pub message: String,
    /// What the wire refusal's `data` carries beside `reason`: the cap, who
    /// holds it, spend so far, the reset time and zone. `None` when the
    /// refusal has nothing to add.
    pub data: Option<serde_json::Value>,
}

/// The stable reason of a refusal by a capped principal's unpriced model.
pub const PRICE_UNKNOWN: &str = "price_unknown";

/// The message of that refusal, for `model`.
pub fn price_unknown_message(model: &str) -> String {
    format!("price unknown for {model}; this agent has a spend cap")
}

/// The message of a `price_unknown` refusal of a model the catalog does
/// price, when its price yields no estimate for this call (a stream whose
/// length is not known when it opens, a row the pricing registry does not
/// hold): the price is known, the call's cost before it is sent is not.
pub fn cost_unestimable_message(model: &str) -> String {
    format!(
        "the cost of this call to {model} cannot be estimated before it is sent; this agent has a spend cap"
    )
}

/// The stable reason of a refusal of a capped principal's second call to a
/// model the provider prices only when the call ends
/// (`PriceKnown::ProviderReports`), while its first is still in flight.
pub const UNPRICED_CALL_IN_FLIGHT: &str = "unpriced_call_in_flight";

/// The message of that refusal, for `model`.
pub fn unpriced_call_in_flight_message(model: &str) -> String {
    format!(
        "{model} is priced only when a call finishes, and this agent already has one such call running; this agent has a spend cap, so it may make one at a time"
    )
}

/// The refusal of a capped principal's remote call whose cost CAR cannot
/// hold in flight at admission (no estimate), worded by why: a model with
/// no price at all, or a priced model whose price yields no estimate for
/// this call. `None` when the call has an estimate, or when its provider
/// reports the cost afterwards (`ProviderReports`), which the engine admits
/// one at a time instead. The engine and the daemon's meter gate both
/// apply it, so every caller of the gate refuses the same calls with the
/// same text.
pub fn unestimable_refusal(facts: &DispatchFacts) -> Option<SpendRefusal> {
    if facts.is_local || facts.estimate.micros.is_some() {
        return None;
    }
    let message = match facts.price {
        PriceKnown::ProviderReports => return None,
        PriceKnown::ListPrice | PriceKnown::Subscription | PriceKnown::Local => {
            cost_unestimable_message(&facts.model)
        }
        PriceKnown::Unknown => price_unknown_message(&facts.model),
    };
    Some(SpendRefusal {
        reason: PRICE_UNKNOWN.to_string(),
        message,
        data: Some(serde_json::json!({
            "reason": PRICE_UNKNOWN,
            "model": facts.model,
        })),
    })
}

/// Whether `error` is one the engine raises for a capped principal's
/// unpriced candidate (`price_unknown`, `unpriced_call_in_flight`): a
/// router-chosen candidate refused this way is passed over for the next.
pub fn is_unpriced_refusal(error: &InferenceError) -> bool {
    matches!(
        error,
        InferenceError::SpendRefused { reason, .. }
            if reason == PRICE_UNKNOWN || reason == UNPRICED_CALL_IN_FLIGHT
    )
}

/// The meter's handle on an admitted call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpendTicketId {
    /// The principal's meter key id.
    pub principal: String,
    pub call_id: String,
}

/// How an admitted call ended, for the meter.
#[derive(Debug, Clone, PartialEq)]
pub struct SettledCall {
    pub model_used: String,
    pub cost: CallCost,
    pub input_tokens: u64,
    pub output_tokens: u64,
    /// The call ended with an error raised before any dispatch: nothing was
    /// sent, so it counts no call and no cost.
    pub voided: bool,
}

/// The daemon's spend meter, as the engine sees it. Every method is
/// synchronous: the meter's writes are small and a call's settle must run
/// where a future is dropped.
pub trait SpendGate: Send + Sync {
    /// Admit a call by `scope`'s principal, or refuse it: at a cap that
    /// applies (`budget_reached`), or because the meter cannot record a
    /// capped principal's spend (`spend_unrecordable`). An uncapped
    /// principal is never refused.
    fn admit(
        &self,
        scope: &SpendScope,
        facts: &DispatchFacts,
    ) -> Result<SpendTicketId, SpendRefusal>;
    /// Whether any cap applies to `scope`'s principal, including an automatic
    /// ceiling. Provider-priced calls without estimates are serialized.
    fn capped(&self, scope: &SpendScope) -> bool {
        let _ = scope;
        false
    }
    /// Whether a cap requires a known price. An automatic ceiling may preserve
    /// unknown-price admission while still checking budgets and metering calls.
    fn requires_price(&self, scope: &SpendScope) -> bool {
        self.capped(scope)
    }
    /// The engine refused a call itself, before consulting `admit` (a capped
    /// principal's pinned or routed model with no known price): the meter
    /// records it like one of its own refusals. Nothing was sent.
    fn refused(&self, scope: &SpendScope, facts: &DispatchFacts, refusal: &SpendRefusal) {
        let _ = (scope, facts, refusal);
    }
    /// The call ended; count `call` into the periods it was admitted in.
    fn settle(&self, ticket: &SpendTicketId, call: &SettledCall);
    /// The provider's final figure for a settled call, reported later.
    fn correct(&self, ticket: &SpendTicketId, cost: CallCost);
}

/// A gate that admits everything and records nothing, for tests of the
/// scope plumbing.
#[derive(Debug, Default)]
pub struct RecordingGate {
    pub admitted: Mutex<Vec<(SpendScope, DispatchFacts)>>,
    pub settled: Mutex<Vec<(SpendTicketId, SettledCall)>>,
    pub corrected: Mutex<Vec<(SpendTicketId, CallCost)>>,
}

impl SpendGate for RecordingGate {
    fn admit(
        &self,
        scope: &SpendScope,
        facts: &DispatchFacts,
    ) -> Result<SpendTicketId, SpendRefusal> {
        let mut admitted = lock(&self.admitted);
        let call_id = format!("call-{}", admitted.len() + 1);
        admitted.push((scope.clone(), facts.clone()));
        Ok(SpendTicketId {
            principal: scope.principal.key(),
            call_id,
        })
    }

    fn settle(&self, ticket: &SpendTicketId, call: &SettledCall) {
        lock(&self.settled).push((ticket.clone(), call.clone()));
    }

    fn correct(&self, ticket: &SpendTicketId, cost: CallCost) {
        lock(&self.corrected).push((ticket.clone(), cost));
    }
}

/// An admitted call. Settles once, on every way it ends: explicitly
/// through [`settle`](Self::settle), or on drop with the usage seen so far,
/// else the pre-dispatch estimate. An error raised before dispatch
/// ([`ended_with`](Self::ended_with)) voids it: nothing was sent, so nothing
/// is charged.
pub struct SpendTicket {
    gate: Arc<dyn SpendGate>,
    id: SpendTicketId,
    scope: SpendScope,
    registry: Arc<UnifiedRegistry>,
    facts: DispatchFacts,
    model_used: String,
    usage: Option<TokenUsage>,
    speech: Option<CallUsage<'static>>,
    trace_id: Option<String>,
    voided: bool,
    settled: bool,
    /// Whether a cap applied to the principal when the call was admitted.
    capped: bool,
    /// The call reached a provider and ended before its final usage
    /// ([`cut_off`](Self::cut_off)).
    cut_off: bool,
    /// The caller said how the call ended ([`ended_with`](Self::ended_with),
    /// [`void`](Self::void), [`cut_off`](Self::cut_off)). A ticket dropped
    /// unsettled without that was abandoned mid-call (the caller's future
    /// was cancelled, or timed out): it settles as cut off.
    ended: bool,
    /// The backend wrote this call's request and read no answer
    /// ([`note_sent_unread`]): an error then ends it cut off.
    sent_unread: Arc<std::sync::atomic::AtomicBool>,
    /// The principal's one in-flight slot for a call priced only when it
    /// ends, held for this call's life. Freed when the ticket settles,
    /// unless the provider's figure is still to come from a late lookup:
    /// then it is held until that figure lands, for at most
    /// [`UNPRICED_LATE_FIGURE_HOLD`].
    unpriced_slot: Option<UnpricedSlot>,
    /// The caller's late-cost probe (`scope_late_cost`), where the engine
    /// leaves the provider's lookup for the call. The ticket runs that
    /// lookup when it settles. `None` when the caller runs no late lookup.
    late_probe: Option<Arc<crate::LateCostProbe>>,
    /// This attempt's dispatch key ([`crate::late_cost::attempt_scope`]),
    /// bound by [`dispatched`], so the ticket takes only the late lookup its
    /// own call left (car#3630).
    late_attempt: u64,
}

impl std::fmt::Debug for SpendTicket {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SpendTicket")
            .field("id", &self.id)
            .field("principal", &self.scope.principal)
            .field("model", &self.facts.model)
            .field("settled", &self.settled)
            .finish()
    }
}

impl SpendTicket {
    /// The meter's call id.
    pub fn call_id(&self) -> &str {
        &self.id.call_id
    }

    pub fn id(&self) -> &SpendTicketId {
        &self.id
    }

    pub fn scope(&self) -> &SpendScope {
        &self.scope
    }

    /// The engine's outcome trace of this attempt, so the call's record can
    /// find this receipt by `trace_id` rather than taking the scope's newest.
    pub fn set_trace(&mut self, trace_id: &str) {
        self.trace_id = Some(trace_id.to_string());
    }

    /// The model that served the call, and its final usage when known.
    pub fn served(&mut self, model_used: &str, usage: Option<&TokenUsage>) {
        self.ended = true;
        self.model_used = model_used.to_string();
        if let Some(usage) = usage {
            self.usage = Some(usage.clone());
        }
    }

    /// The latest usage reported so far (a stream's running total).
    pub fn usage_so_far(&mut self, usage: TokenUsage) {
        self.usage = Some(usage);
    }

    /// The provider's own dollar figure for the call, seen before the usage
    /// row (a stream's cost event).
    pub fn reported_cost(&mut self, usd: f64) {
        let usage = self.usage.get_or_insert_with(TokenUsage::default);
        if usage.reported_cost_usd.is_none() {
            usage.reported_cost_usd = Some(usd);
        }
    }

    /// A speech call's measured usage (seconds heard, characters spoken).
    pub fn speech_usage(&mut self, usage: CallUsage<'static>) {
        self.speech = Some(usage);
    }

    /// The call ended with `error`. One raised before any dispatch voids
    /// the ticket: nothing was sent, so nothing is charged. One that ended a
    /// dispatched call before its answer was read (the caller's deadline
    /// elapsed, the run was cancelled, or the provider answered 2xx and the
    /// body could not be read) cuts it off ([`cut_off`](Self::cut_off)):
    /// the provider may have billed output that was never seen.
    pub fn ended_with(&mut self, error: &InferenceError) {
        self.ended = true;
        // A refusal is billed; when the provider said what it used, that is
        // the call's usage, not an estimate (car#3704).
        if let InferenceError::ContentRefused {
            usage: Some(usage), ..
        } = error
        {
            self.usage.get_or_insert_with(|| usage.clone());
        }
        if error.raised_before_dispatch() {
            self.voided = true;
        } else if ended_after_dispatch_unread(error)
            || self.sent_unread.load(std::sync::atomic::Ordering::Relaxed)
        {
            self.cut_off = true;
        }
    }

    /// Nothing was sent for this attempt (the host has no runner for a
    /// delegated model, the runner declined the request): it counts no
    /// call and no cost, whatever error the caller reports.
    pub fn void(&mut self) {
        self.ended = true;
        self.voided = true;
    }

    /// The call reached a provider and ended before its final usage: a
    /// stream the caller dropped, or one that broke off. Most providers
    /// report output only in a stream's last chunk, so the output already
    /// generated, and billed, is not in the usage seen. A capped principal's
    /// call is then charged its admission estimate, output ceiling included,
    /// so the cap is never under-counted; an uncapped principal's settles on
    /// what was seen, as before.
    pub fn cut_off(&mut self) {
        self.ended = true;
        self.cut_off = true;
    }

    /// Settle now, with what is known. The same as dropping the ticket;
    /// explicit where the caller wants the receipt in hand.
    pub fn settle(mut self) -> SpendReceipt {
        self.settle_inner()
    }

    fn settle_inner(&mut self) -> SpendReceipt {
        self.settled = true;
        // A voided attempt has no cost to report: `micros: None` with
        // `Unknown` keeps billing's invariant (`micros` is `None` exactly
        // when the source is `Unknown`); `voided` on the settled call and
        // the receipt is what says it counts no call and no money, so the
        // meter never reads it as a priced-unknown call.
        let (cost, input_tokens, output_tokens) = if self.voided {
            (
                CallCost {
                    micros: None,
                    source: CostSource::Unknown,
                },
                0,
                0,
            )
        } else {
            self.price()
        };
        let call = SettledCall {
            model_used: self.model_used.clone(),
            cost,
            input_tokens,
            output_tokens,
            voided: self.voided,
        };
        self.gate.settle(&self.id, &call);
        self.run_late_lookup(cost.source);
        let receipt = SpendReceipt {
            call_id: self.id.call_id.clone(),
            principal: self.scope.principal.clone(),
            cause: self.scope.cause,
            model_used: self.model_used.clone(),
            cost_usd: cost.micros.map(billing::micros_to_usd),
            cost_source: cost.source,
            input_tokens,
            output_tokens,
            voided: self.voided,
            trace_id: self.trace_id.clone(),
        };
        self.scope.push_receipt(receipt.clone());
        receipt
    }

    /// Run the provider's late lookup for this call and free, or hand over,
    /// the principal's unpriced slot. Here rather than in each caller, so it
    /// happens on every way the call ends: a success, an error, a cancelled
    /// or broken stream, a candidate the router fell back from (whose ticket
    /// settles before the next candidate is tried, under its own model).
    ///
    /// A call that ended with no provider figure, whose provider left a
    /// lookup in the caller's probe for the model that served it, has the
    /// lookup run in a spawned task; the figure, when it comes, corrects the
    /// charge. A capped principal's slot waits for that lookup to end (at
    /// most [`UNPRICED_LATE_FIGURE_HOLD`]), so its next unpriced call is
    /// admitted against a spent figure that includes this one. With nothing
    /// to look up, the slot is freed at once.
    fn run_late_lookup(&mut self, source: CostSource) {
        // Taken even when the provider's figure arrived, so a lookup this
        // call left is never run for a later call in the same scope. A voided
        // ticket never dispatched, so a lookup there belongs to another call.
        let late = if self.voided {
            None
        } else {
            self.late_probe
                .as_ref()
                .and_then(|probe| probe.take_for_attempt(self.late_attempt, &self.model_used))
                .filter(|_| !matches!(source, CostSource::Provider | CostSource::Subscription))
        };
        let slot = self.unpriced_slot.take();
        let (Some(late), Ok(runtime)) = (late, tokio::runtime::Handle::try_current()) else {
            // Nothing to wait for: the slot, if held, is freed as it drops.
            drop(slot);
            return;
        };
        let release = slot.and_then(|slot| slot.await_late_figure(&self.id.call_id));
        let ceiling =
            billing::provider_figure_ceiling(&self.registry, &self.model_used, self.usage.as_ref());
        let gate = Arc::clone(&self.gate);
        let id = self.id.clone();
        let model = self.model_used.clone();
        runtime.spawn(async move {
            let figure = late
                .resolve()
                .await
                .and_then(|usd| billing::accept_provider_figure(&model, usd, ceiling));
            if let Some(usd) = figure {
                gate.correct(
                    &id,
                    CallCost {
                        micros: Some(billing::usd_to_micros_up(usd)),
                        source: CostSource::Provider,
                    },
                );
            }
            // After the correction, so the principal's next unpriced call is
            // admitted against a spent figure that includes this one.
            if let Some(slots) = release {
                slots.late_figure_resolved(&id.call_id);
            }
        });
    }

    /// Price the call as it ended: the usage seen, else the pre-dispatch
    /// estimate (an upper bound), at the model that served it.
    fn price(&self) -> (CallCost, u64, u64) {
        if let Some(speech) = self.speech {
            let cost = billing::call_cost(&self.registry, &self.model_used, speech);
            let (input, output) = match speech {
                CallUsage::AudioSeconds(seconds) => (seconds.ceil().max(0.0) as u64, 0),
                CallUsage::Characters(chars) => (chars, 0),
                CallUsage::Tokens(_) => (0, 0),
            };
            return (cost, input, output);
        }
        let provider_figure = self
            .usage
            .as_ref()
            .is_some_and(|usage| usage.reported_cost_usd.is_some());
        if self.cut_off && self.capped && self.facts.kind == CallKind::Tokens && !provider_figure {
            // The usage seen so far (an input count, a running output
            // count), with the output raised to the ceiling the admission
            // held: an upper bound for what the provider billed.
            let mut usage = self.usage.clone().unwrap_or_default();
            if usage.prompt_tokens == 0
                && usage.cache_read_input_tokens == 0
                && usage.cache_creation_input_tokens == 0
            {
                usage.prompt_tokens = self.facts.input_estimate;
            }
            usage.completion_tokens = usage.completion_tokens.max(self.facts.max_output);
            usage.total_tokens = usage.prompt_tokens.saturating_add(usage.completion_tokens);
            let input = usage
                .prompt_tokens
                .saturating_add(usage.cache_read_input_tokens)
                .saturating_add(usage.cache_creation_input_tokens);
            return (
                billing::call_cost(
                    &self.registry,
                    &self.model_used,
                    CallUsage::Tokens(Some(&usage)),
                ),
                input,
                usage.completion_tokens,
            );
        }
        match (&self.usage, self.facts.kind) {
            (Some(usage), CallKind::Tokens) => (
                billing::call_cost(
                    &self.registry,
                    &self.model_used,
                    CallUsage::Tokens(Some(usage)),
                ),
                // The whole prompt, cached parts included: the count the
                // worker meter has always kept.
                usage
                    .prompt_tokens
                    .saturating_add(usage.cache_read_input_tokens)
                    .saturating_add(usage.cache_creation_input_tokens),
                usage.completion_tokens,
            ),
            // The call reached a provider and ended with no usage (a 4xx,
            // a 5xx, a transport failure with no status, or any cut-off
            // call of an uncapped principal; a capped principal's call that
            // was cancelled, timed out or left unread is cut off and charged
            // its estimate above): it is charged at the input it
            // sent, as the worker meter always charged it, not at the
            // output ceiling. The admission estimate, which does include the
            // ceiling, is the upper bound for calls in flight and is
            // released here.
            _ => (
                estimate_cost(
                    &self.registry,
                    &self.model_used,
                    self.facts.kind,
                    self.facts.input_estimate,
                    0,
                ),
                self.facts.input_estimate,
                0,
            ),
        }
    }
}

impl Drop for SpendTicket {
    fn drop(&mut self) {
        if !self.settled {
            if !self.ended {
                // Dropped mid-call: the caller's future was cancelled or
                // timed out, after the call was admitted and sent.
                self.cut_off = true;
            }
            self.settle_inner();
        }
    }
}

/// Whether `error` ended a dispatched call before its answer was read, so
/// the provider may have billed output CAR never saw: the caller's own
/// deadline elapsed, the run was cancelled, or the provider answered 2xx and
/// the body could not be read (`Transient` with that status, from the remote
/// backend). A transport failure with no status is not among them: it covers
/// a connection that never opened, which billed nothing.
fn ended_after_dispatch_unread(error: &InferenceError) -> bool {
    match error {
        InferenceError::DeadlineExceeded { .. } | InferenceError::ControlledTermination => true,
        InferenceError::Transient {
            status: Some(status),
            ..
        } => (200..300).contains(status),
        _ => false,
    }
}

/// Void `ticket` (when there is one) for `error`, an error raised before
/// anything was sent, and hand the error back: for the `?` exits of a
/// candidate loop, so the attempt counts no call.
pub fn void_pre_dispatch(
    ticket: &mut Option<SpendTicket>,
    error: InferenceError,
) -> InferenceError {
    if let Some(ticket) = ticket.as_mut() {
        ticket.void();
    }
    error
}

/// The cost of a `kind` call to `model` that uses `input_estimate` and
/// produces `max_output`, at the catalog price: an upper bound for a
/// token-priced call. Unknown when the price is.
pub fn estimate_cost(
    registry: &UnifiedRegistry,
    model: &str,
    kind: CallKind,
    input_estimate: u64,
    max_output: u64,
) -> CallCost {
    let usage = match kind {
        CallKind::Tokens => {
            let usage = TokenUsage {
                prompt_tokens: input_estimate,
                completion_tokens: max_output,
                total_tokens: input_estimate.saturating_add(max_output),
                ..Default::default()
            };
            return billing::call_cost(registry, model, CallUsage::Tokens(Some(&usage)));
        }
        CallKind::AudioSeconds => CallUsage::AudioSeconds(input_estimate as f64),
        CallKind::Characters => CallUsage::Characters(input_estimate),
    };
    billing::call_cost(registry, model, usage)
}

/// The engine's gate, with whether a scope is required and whether the
/// engine may dispatch to anything but an on-device model.
#[derive(Clone)]
pub(crate) struct InstalledGate {
    pub gate: Arc<dyn SpendGate>,
    pub require_scope: bool,
    /// Refuse every dispatch whose resolved model is not `is_local()`,
    /// scope or no scope, before the gate is consulted. Set by
    /// [`InferenceEngine::restrict_to_local_dispatch`](crate::InferenceEngine::restrict_to_local_dispatch)
    /// in the on-device inference worker child, whose parent already
    /// admitted the call at $0 as local: a remote call from there would be
    /// uncounted, so it is refused as a CAR bug rather than let through.
    pub local_only: bool,
    /// The principals with a call in flight to a model priced only when the
    /// call ends ([`UnpricedSlots`]). Shared by every clone of this gate and
    /// kept across a later
    /// [`set_spend_gate`](crate::InferenceEngine::set_spend_gate).
    pub unpriced: Arc<UnpricedSlots>,
}

/// How long a provider-priced call that ended with no figure keeps its
/// principal's slot while the provider's late lookup for it runs: the
/// lookup's own window ([`LATE_COST_WINDOW`](crate::late_cost::LATE_COST_WINDOW),
/// 10 minutes from the end of the call, after which it gives up) plus a
/// minute for the gap between the settle and the lookup starting. A
/// provider that never answers holds the slot no longer than this; the
/// figure landing, or the lookup giving up, frees it at once.
pub const UNPRICED_LATE_FIGURE_HOLD: Duration =
    Duration::from_secs(crate::late_cost::LATE_COST_WINDOW.as_secs() + 60);

/// The capped principals that have a call in flight to a remote model with
/// no list price whose provider reports the cost when the call ends
/// (`PriceKnown::ProviderReports`, for example OpenRouter), each with the
/// slot id of that call. Such a call has no estimate to hold in flight at
/// admission, so the cap could not bound a burst of them; one at a time it
/// can: a period's total is then at most the cap, plus the priced overshoot
/// the settled rule allows, plus the one unpriced call in flight (owner
/// decision, Keenan Keeling, 2026-10-26). A call that settles with its
/// figure still to come from the provider's late lookup keeps the slot until
/// that figure lands, for at most [`UNPRICED_LATE_FIGURE_HOLD`]. In memory,
/// by design: a restart ends every call the old process had in flight.
#[derive(Debug, Default)]
pub struct UnpricedSlots {
    held: Mutex<HashMap<String, HeldSlot>>,
    next: std::sync::atomic::AtomicU64,
}

/// One principal's held slot: the call holding it, and once that call has
/// settled with its figure still to come, which call that is and until
/// when the slot waits for it.
#[derive(Debug)]
struct HeldSlot {
    id: u64,
    awaiting: Option<(String, tokio::time::Instant)>,
}

impl HeldSlot {
    /// A slot waiting for a late figure past its deadline is free.
    fn lapsed(&self) -> bool {
        self.awaiting
            .as_ref()
            .is_some_and(|(_, until)| tokio::time::Instant::now() >= *until)
    }
}

impl UnpricedSlots {
    /// Take `principal`'s slot, or `None` while another call holds it.
    fn take(self: &Arc<Self>, principal: &str) -> Option<UnpricedSlot> {
        let id = self.next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let mut held = lock(&self.held);
        if held.get(principal).is_some_and(|slot| !slot.lapsed()) {
            return None;
        }
        held.insert(principal.to_string(), HeldSlot { id, awaiting: None });
        Some(UnpricedSlot {
            slots: Arc::clone(self),
            principal: principal.to_string(),
            id,
            armed: true,
        })
    }

    /// Whether `principal` has such a call in flight now, or one whose
    /// late figure is still to land.
    pub fn held(&self, principal: &str) -> bool {
        lock(&self.held)
            .get(principal)
            .is_some_and(|slot| !slot.lapsed())
    }

    /// The late lookup for call `call_id` finished: its figure was applied,
    /// or none came. The slot that waited for it is free.
    pub fn late_figure_resolved(&self, call_id: &str) {
        lock(&self.held).retain(|_, slot| {
            slot.awaiting
                .as_ref()
                .is_none_or(|(waiting_for, _)| waiting_for != call_id)
        });
    }
}

/// One principal's slot in [`UnpricedSlots`], freed when dropped: when the
/// call's ticket settles, or when its admission is refused after the slot
/// was taken. A call whose figure is still to come hands the slot over to
/// wait for it instead ([`await_late_figure`](Self::await_late_figure)).
struct UnpricedSlot {
    slots: Arc<UnpricedSlots>,
    principal: String,
    id: u64,
    armed: bool,
}

impl UnpricedSlot {
    /// Keep the slot held for call `call_id` until its late figure lands
    /// ([`UnpricedSlots::late_figure_resolved`]) or
    /// [`UNPRICED_LATE_FIGURE_HOLD`] passes. The slots to release it from,
    /// when this call still held it.
    fn await_late_figure(mut self, call_id: &str) -> Option<Arc<UnpricedSlots>> {
        let mut held = lock(&self.slots.held);
        let slot = held.get_mut(&self.principal)?;
        if slot.id != self.id {
            return None;
        }
        slot.awaiting = Some((
            call_id.to_string(),
            tokio::time::Instant::now() + UNPRICED_LATE_FIGURE_HOLD,
        ));
        self.armed = false;
        drop(held);
        Some(Arc::clone(&self.slots))
    }
}

impl Drop for UnpricedSlot {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }
        let mut held = lock(&self.slots.held);
        if held
            .get(&self.principal)
            .is_some_and(|slot| slot.id == self.id)
        {
            held.remove(&self.principal);
        }
    }
}

/// A refusal the engine raised for a candidate before consulting the gate
/// (`price_unknown`, `unpriced_call_in_flight`), not yet recorded. The
/// router passes over such a candidate when it chose it, and a candidate it
/// passed over leaves no refusal record: only the refusal the caller
/// actually receives is recorded ([`record`](Self::record)).
pub(crate) struct UnrecordedRefusal {
    gate: Arc<dyn SpendGate>,
    scope: SpendScope,
    facts: DispatchFacts,
    refusal: SpendRefusal,
}

impl UnrecordedRefusal {
    /// The refusal as the caller would receive it.
    pub(crate) fn error(&self) -> InferenceError {
        InferenceError::SpendRefused {
            reason: self.refusal.reason.clone(),
            message: self.refusal.message.clone(),
            data: self.refusal.data.clone(),
        }
    }

    /// The refusal's stable reason (`price_unknown`,
    /// `unpriced_call_in_flight`).
    pub(crate) fn reason(&self) -> &str {
        &self.refusal.reason
    }

    /// Whether `error` is this refusal: what the caller receives when the
    /// candidate loop ends on it.
    pub(crate) fn is(&self, error: &InferenceError) -> bool {
        matches!(
            error,
            InferenceError::SpendRefused { reason, message, .. }
                if *reason == self.refusal.reason && *message == self.refusal.message
        )
    }

    /// Record the refusal (the caller receives it) and hand it back.
    pub(crate) fn record(self) -> InferenceError {
        self.gate.refused(&self.scope, &self.facts, &self.refusal);
        self.error()
    }
}

/// Why [`InstalledGate::admit_candidate`] did not admit a call.
pub(crate) enum CandidateRefusal {
    /// The gate's own refusal (recorded by the gate), or the engine's for a
    /// reason no router skips (no scope, a remote call in the local-only
    /// worker).
    Final(InferenceError),
    /// An unpriced candidate for a capped principal: not recorded yet.
    Unrecorded(UnrecordedRefusal),
}

impl CandidateRefusal {
    /// The refusal the caller receives, recorded.
    pub(crate) fn recorded(self) -> InferenceError {
        match self {
            Self::Final(error) => error,
            Self::Unrecorded(refusal) => refusal.record(),
        }
    }
}

/// The refusal for a dispatch with no scope where one is required
/// (`docs/proposals/spend-caps.md`: "never let through uncounted").
pub const NO_SCOPE_MESSAGE: &str = "model call has no spend scope (CAR bug)";

/// The stable reason of the refusal a local-only engine (the on-device
/// inference worker child) answers a remote dispatch with.
pub const REMOTE_IN_LOCAL_WORKER: &str = "remote_in_local_worker";

/// The refusal for a remote dispatch under a [`SpendScope::local_only`] scope.
pub const REMOTE_IN_LOCAL_ONLY_SCOPE: &str = "remote_in_local_only_scope";

pub const REMOTE_IN_LOCAL_ONLY_SCOPE_MESSAGE: &str = "this work runs with no one watching, so it may use only local models; a remote model would spend money unattended";

/// Refuse a dispatch to anything but a local model under a
/// [`SpendScope::local_only`] scope. Checked by the engine's admission entries
/// BEFORE they look for an installed gate, so the refusal holds where no gate
/// is installed; [`InstalledGate`] checks again for its own callers. An
/// unknown model (`None`) is not local.
pub fn refuse_remote_in_local_only_scope(
    schema: Option<&crate::schema::ModelSchema>,
) -> Result<(), InferenceError> {
    if current().is_some_and(|scope| scope.local_only)
        && !schema.is_some_and(crate::schema::ModelSchema::is_local)
    {
        return Err(InferenceError::SpendRefused {
            reason: REMOTE_IN_LOCAL_ONLY_SCOPE.to_string(),
            message: REMOTE_IN_LOCAL_ONLY_SCOPE_MESSAGE.to_string(),
            data: None,
        });
    }
    Ok(())
}

/// The message of that refusal.
pub const REMOTE_IN_LOCAL_WORKER_MESSAGE: &str =
    "remote model dispatched inside the on-device inference worker (CAR bug)";

/// The gate a local-only engine installs. It is consulted only for a
/// dispatch that is already local (the local-only check runs first), so it
/// admits and records nothing: a local call costs $0 and the parent, not
/// the worker child, holds the meter.
#[derive(Debug, Default)]
pub struct LocalOnlyGate;

impl SpendGate for LocalOnlyGate {
    fn admit(
        &self,
        scope: &SpendScope,
        facts: &DispatchFacts,
    ) -> Result<SpendTicketId, SpendRefusal> {
        debug_assert!(facts.is_local, "the local-only check runs before the gate");
        Ok(SpendTicketId {
            principal: scope.principal.key(),
            call_id: "local".to_string(),
        })
    }

    fn settle(&self, _ticket: &SpendTicketId, _call: &SettledCall) {}

    fn correct(&self, _ticket: &SpendTicketId, _cost: CallCost) {}
}

impl InstalledGate {
    /// Admit one dispatch attempt. `attempt` is 0 for the first candidate
    /// of a request and counts up through router fallbacks, each of which
    /// is its own call with cause `fallback`. A refusal is recorded before it
    /// is returned: for the dispatch sites whose caller receives it.
    ///
    /// `Ok(None)` means the call runs unmetered: no scope, and none is
    /// required or the model is local. A local call with no scope is not
    /// refused because it costs nothing and the daemon's own local work
    /// (an on-device worker child) has no request to be scoped by.
    ///
    /// Under `local_only`, a dispatch whose model is not local is refused
    /// first, whatever the scope and however the router arrived at it
    /// (a pin, a fallback, a failover): the check sits at the one entry
    /// every dispatch site passes through, so no routing change can walk
    /// around it.
    pub(crate) fn admit(
        &self,
        registry: &Arc<UnifiedRegistry>,
        facts: DispatchFacts,
        attempt: usize,
    ) -> Result<Option<SpendTicket>, InferenceError> {
        self.admit_candidate(registry, facts, attempt)
            .map_err(CandidateRefusal::recorded)
    }

    /// [`admit`](Self::admit) for a router candidate: a capped principal's
    /// unpriced candidate comes back [`CandidateRefusal::Unrecorded`], so the
    /// loop can pass over it with no refusal on the record, and records it
    /// only when the request ends on it.
    pub(crate) fn admit_candidate(
        &self,
        registry: &Arc<UnifiedRegistry>,
        facts: DispatchFacts,
        attempt: usize,
    ) -> Result<Option<SpendTicket>, CandidateRefusal> {
        if self.local_only && !facts.is_local {
            return Err(CandidateRefusal::Final(InferenceError::SpendRefused {
                reason: REMOTE_IN_LOCAL_WORKER.to_string(),
                message: REMOTE_IN_LOCAL_WORKER_MESSAGE.to_string(),
                data: None,
            }));
        }
        if !facts.is_local && current().is_some_and(|scope| scope.local_only) {
            return Err(CandidateRefusal::Final(InferenceError::SpendRefused {
                reason: REMOTE_IN_LOCAL_ONLY_SCOPE.to_string(),
                message: REMOTE_IN_LOCAL_ONLY_SCOPE_MESSAGE.to_string(),
                data: None,
            }));
        }
        let Some(scope) = current() else {
            if self.require_scope && !facts.is_local {
                return Err(CandidateRefusal::Final(InferenceError::SpendRefused {
                    reason: "no_scope".to_string(),
                    message: NO_SCOPE_MESSAGE.to_string(),
                    data: None,
                }));
            }
            return Ok(None);
        };
        // A capped principal's remote call CAR cannot price before it is sent
        // (A5, CAR-246). With no list price and no provider figure to come
        // (`Unknown`), or a price that yields no estimate for this call, it
        // is refused: the admission would have nothing to hold in flight, so
        // a burst of such calls could not be bounded. A model whose provider
        // reports the cost when the call ends (`ProviderReports`) is admitted
        // one call at a time instead, and charged the provider's figure at
        // settle (owner decision, Keenan Keeling, 2026-10-26). The candidate
        // loop skips such a model when the router chose it and refuses when
        // the caller pinned it; an uncapped principal's call goes through
        // with its cost unknown until the provider reports it. An automatic
        // owner ceiling can preserve Unknown admission through `requires_price`
        // while retaining serialization and cutoff accounting for paid calls.
        let capped = !facts.is_local && self.gate.capped(&scope);
        let unestimable = capped && facts.estimate.micros.is_none();
        let mut unpriced_slot = None;
        if unestimable {
            if let Some(refusal) =
                unestimable_refusal(&facts).filter(|_| self.gate.requires_price(&scope))
            {
                return Err(CandidateRefusal::Unrecorded(UnrecordedRefusal {
                    gate: Arc::clone(&self.gate),
                    scope,
                    facts,
                    refusal,
                }));
            }
        }
        if unestimable && matches!(facts.price, PriceKnown::ProviderReports) {
            match self.unpriced.take(&scope.principal.key()) {
                Some(slot) => unpriced_slot = Some(slot),
                None => {
                    let refusal = SpendRefusal {
                        reason: UNPRICED_CALL_IN_FLIGHT.to_string(),
                        message: unpriced_call_in_flight_message(&facts.model),
                        data: Some(serde_json::json!({
                            "reason": UNPRICED_CALL_IN_FLIGHT,
                            "model": facts.model,
                        })),
                    };
                    return Err(CandidateRefusal::Unrecorded(UnrecordedRefusal {
                        gate: Arc::clone(&self.gate),
                        scope,
                        facts,
                        refusal,
                    }));
                }
            }
        }
        // The spec names `fallback` as the cause of "each router retry or
        // failover attempt": every attempt after the first is one, whatever
        // the request's own cause (a chat turn's second candidate is a
        // fallback, not a chat). The record of the request still carries the
        // served attempt's receipt.
        let scope = if attempt > 0 {
            scope.with_cause(SpendCause::Fallback)
        } else {
            scope
        };
        // A refusal by the gate drops the slot taken above with it.
        let id = self.gate.admit(&scope, &facts).map_err(|refusal| {
            CandidateRefusal::Final(InferenceError::SpendRefused {
                reason: refusal.reason,
                message: refusal.message,
                data: refusal.data,
            })
        })?;
        Ok(Some(SpendTicket {
            gate: Arc::clone(&self.gate),
            id,
            scope,
            registry: Arc::clone(registry),
            model_used: facts.model.clone(),
            facts,
            usage: None,
            speech: None,
            trace_id: None,
            voided: false,
            settled: false,
            capped,
            cut_off: false,
            ended: false,
            sent_unread: Default::default(),
            late_probe: crate::late_cost::current_probe(),
            late_attempt: crate::late_cost::next_attempt(),
            unpriced_slot,
        }))
    }
}

/// The facts of a `kind` dispatch to `schema` (or to an unresolved
/// `model`), with its estimate priced from `registry`.
pub fn facts_for(
    registry: &UnifiedRegistry,
    schema: Option<&crate::schema::ModelSchema>,
    model: &str,
    kind: CallKind,
    input_estimate: u64,
    max_output: u64,
) -> DispatchFacts {
    let (provider, is_local, price) = match schema {
        Some(schema) => (
            schema.provider.clone(),
            schema.is_local(),
            billing::price_known(schema, kind),
        ),
        None => (String::new(), false, PriceKnown::Unknown),
    };
    let model = schema.map_or_else(|| model.to_string(), |schema| schema.id.clone());
    let estimate = estimate_cost(registry, &model, kind, input_estimate, max_output);
    DispatchFacts {
        model,
        provider,
        is_local,
        price,
        kind,
        input_estimate,
        max_output,
        estimate,
    }
}

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

    fn registry_with(model: &str, per_token: f64) -> Arc<UnifiedRegistry> {
        let mut registry = UnifiedRegistry::new_empty(std::env::temp_dir());
        registry.register(crate::ModelSchema {
            id: model.into(),
            name: model.into(),
            provider: "test".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![crate::ModelCapability::Generate],
            context_length: 8_192,
            max_output_tokens: Some(1_024),
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: crate::schema::CostModel {
                input_per_mtok: Some(per_token * 1_000_000.0),
                output_per_mtok: Some(per_token * 1_000_000.0),
                ..Default::default()
            },
            source: crate::ModelSource::Delegated { hint: None },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::TrustTier::Community,
            deprecated: false,
            available: true,
            weights_ready: true,
        });
        Arc::new(registry)
    }

    fn installed(gate: &Arc<RecordingGate>, require_scope: bool) -> InstalledGate {
        InstalledGate {
            gate: gate.clone() as Arc<dyn SpendGate>,
            require_scope,
            local_only: false,
            unpriced: Default::default(),
        }
    }

    /// The gate the on-device worker child installs
    /// (`InferenceEngine::restrict_to_local_dispatch`).
    fn local_only() -> InstalledGate {
        InstalledGate {
            gate: Arc::new(LocalOnlyGate),
            require_scope: false,
            local_only: true,
            unpriced: Default::default(),
        }
    }

    /// A registry holding one on-device row, `local/m`.
    fn local_registry() -> Arc<UnifiedRegistry> {
        let mut registry = UnifiedRegistry::new_empty(std::env::temp_dir());
        registry.register(crate::ModelSchema {
            id: "local/m".into(),
            name: "local/m".into(),
            provider: "local".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![crate::ModelCapability::Generate],
            context_length: 8_192,
            max_output_tokens: Some(1_024),
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: crate::ModelSource::Mlx {
                hf_repo: "x".into(),
                hf_weight_file: None,
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::TrustTier::Community,
            deprecated: false,
            available: true,
            weights_ready: true,
        });
        Arc::new(registry)
    }

    fn facts(registry: &UnifiedRegistry, model: &str) -> DispatchFacts {
        let schema = registry.get(model).cloned();
        facts_for(registry, schema.as_ref(), model, CallKind::Tokens, 10, 20)
    }

    /// A registry holding one `RemoteApi` row on a protocol whose handler
    /// reports a dollar figure (OpenRouter) and no list price: the catalog
    /// cannot price it before dispatch (`PriceKnown::ProviderReports`, an
    /// estimate of `None`), but the record will be priced afterwards.
    fn provider_reports_registry() -> Arc<UnifiedRegistry> {
        let mut registry = UnifiedRegistry::new_empty(std::env::temp_dir());
        registry.register(crate::ModelSchema {
            id: "openrouter/unlisted".into(),
            name: "openrouter/unlisted".into(),
            provider: "openrouter".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![crate::ModelCapability::Generate],
            context_length: 8_192,
            max_output_tokens: Some(1_024),
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: crate::ModelSource::RemoteApi {
                endpoint: "https://openrouter.ai/api/v1".into(),
                api_key_env: "OPENROUTER_API_KEY".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: crate::schema::ApiProtocol::OpenRouter,
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::TrustTier::Community,
            deprecated: false,
            available: true,
            weights_ready: true,
        });
        Arc::new(registry)
    }

    /// A recording gate whose principal is capped: what the engine asks
    /// before consulting `admit`.
    #[derive(Default)]
    pub(crate) struct CappedGate {
        pub(crate) inner: RecordingGate,
        pub(crate) refused: Mutex<Vec<(DispatchFacts, SpendRefusal)>>,
    }

    impl SpendGate for CappedGate {
        fn admit(
            &self,
            scope: &SpendScope,
            facts: &DispatchFacts,
        ) -> Result<SpendTicketId, SpendRefusal> {
            self.inner.admit(scope, facts)
        }
        fn capped(&self, _scope: &SpendScope) -> bool {
            true
        }
        fn refused(&self, _scope: &SpendScope, facts: &DispatchFacts, refusal: &SpendRefusal) {
            lock(&self.refused).push((facts.clone(), refusal.clone()));
        }
        fn settle(&self, ticket: &SpendTicketId, call: &SettledCall) {
            self.inner.settle(ticket, call);
        }
        fn correct(&self, ticket: &SpendTicketId, cost: CallCost) {
            self.inner.correct(ticket, cost);
        }
    }

    fn capped_installed(capped: &Arc<CappedGate>) -> InstalledGate {
        InstalledGate {
            gate: capped.clone() as Arc<dyn SpendGate>,
            require_scope: true,
            local_only: false,
            unpriced: Default::default(),
        }
    }

    fn bot() -> SpendScope {
        SpendScope::new(SpendPrincipal::Agent("bot".into()))
    }

    /// A capped principal's call needs a bounded estimate (A5, CAR-246): a
    /// remote model with no list price and no provider figure to come
    /// (`PriceKnown::Unknown`) has none, so a burst of such calls could not
    /// be bounded. It is refused `price_unknown` before the gate's `admit` is
    /// consulted and recorded as a refusal; an uncapped principal's call goes
    /// through with its estimate unknown. Watched failing: with the
    /// condition back to `price == PriceKnown::Unknown` alone, the capped
    /// call was admitted with `estimate.micros == None` (round 1).
    #[tokio::test]
    async fn a_capped_principal_is_refused_a_model_with_no_price_at_all() {
        let registry = provider_reports_registry();
        let mut facts = facts(&registry, "openrouter/unlisted");
        facts.price = PriceKnown::Unknown;
        assert_eq!(facts.estimate.micros, None);
        assert!(!facts.is_local);

        let capped = Arc::new(CappedGate::default());
        let error = scope(bot(), async {
            capped_installed(&capped)
                .admit(&registry, facts.clone(), 0)
                .expect_err("refused before anything is sent")
        })
        .await;
        assert!(matches!(
            error,
            InferenceError::SpendRefused { ref reason, ref message, .. }
                if reason == PRICE_UNKNOWN
                    && message == "price unknown for openrouter/unlisted; this agent has a spend cap"
        ));
        assert!(error.raised_before_dispatch());
        assert!(
            capped.inner.admitted.lock().unwrap().is_empty(),
            "admit was never consulted"
        );
        {
            let refused = capped.refused.lock().unwrap();
            assert_eq!(refused.len(), 1);
            assert_eq!(refused[0].1.reason, PRICE_UNKNOWN);
        }

        // An uncapped principal: admitted, its estimate unknown.
        let uncapped = Arc::new(RecordingGate::default());
        let ticket = scope(
            SpendScope::new(SpendPrincipal::Agent("free".into())),
            async {
                installed(&uncapped, true)
                    .admit(&registry, facts, 0)
                    .expect("an uncapped principal is never refused")
            },
        )
        .await;
        assert!(ticket.is_some());
        let admitted = uncapped.admitted.lock().unwrap();
        assert_eq!(admitted.len(), 1);
        assert_eq!(admitted[0].1.estimate.micros, None);
    }

    /// The owner decision of 2026-10-26: a capped principal may call a
    /// remote model with no list price whose provider reports the cost
    /// (`ProviderReports`, OpenRouter), one call at a time. While one such
    /// call is in flight a second for the same principal is refused
    /// `unpriced_call_in_flight` with nothing sent, and recorded; another
    /// principal, and a priced call by the same one, are unaffected; the
    /// slot is free again once the first call settles. The provider's figure
    /// is what the call is charged; with no figure the call settles unpriced
    /// (`Unknown`, counted $0 and as an unpriced call), the existing
    /// fallback. Watched failing: with the slot check removed
    /// (`self.unpriced.take` answering a fresh slot every time), the second
    /// call was admitted.
    #[tokio::test]
    async fn a_capped_principal_makes_provider_priced_calls_one_at_a_time() {
        let registry = provider_reports_registry();
        let facts = facts(&registry, "openrouter/unlisted");
        assert_eq!(facts.price, PriceKnown::ProviderReports);
        assert_eq!(facts.estimate.micros, None);

        let capped = Arc::new(CappedGate::default());
        let gate = capped_installed(&capped);
        let (receipt_reported, receipt_silent) = scope(bot(), async {
            let mut first = gate
                .admit(&registry, facts.clone(), 0)
                .expect("the first unpriced call is admitted")
                .expect("metered");
            assert!(gate.unpriced.held("agent:bot"));

            let second = gate
                .admit(&registry, facts.clone(), 0)
                .expect_err("a second while the first runs is refused");
            assert!(matches!(
                second,
                InferenceError::SpendRefused { ref reason, ref message, .. }
                    if reason == UNPRICED_CALL_IN_FLIGHT
                        && message == &unpriced_call_in_flight_message("openrouter/unlisted")
            ));
            assert!(second.raised_before_dispatch());

            // A priced call by the same principal is not held up.
            let priced_registry = registry_with("test/m", 0.000_001);
            gate.admit(
                &priced_registry,
                super::tests::facts(&priced_registry, "test/m"),
                0,
            )
            .expect("a priced call is admitted")
            .expect("metered");

            // The provider reports $0.42: that is the charge.
            first.served(
                "openrouter/unlisted",
                Some(&TokenUsage {
                    prompt_tokens: 10,
                    completion_tokens: 5,
                    total_tokens: 15,
                    reported_cost_usd: Some(0.42),
                    ..Default::default()
                }),
            );
            let reported = first.settle();
            assert!(!gate.unpriced.held("agent:bot"), "freed at settle");

            // The slot is free again; this call ends with no figure.
            let mut silent = gate
                .admit(&registry, facts.clone(), 0)
                .expect("admitted once the first settled")
                .expect("metered");
            silent.served(
                "openrouter/unlisted",
                Some(&TokenUsage {
                    prompt_tokens: 10,
                    completion_tokens: 5,
                    total_tokens: 15,
                    ..Default::default()
                }),
            );
            (reported, silent.settle())
        })
        .await;
        assert_eq!(receipt_reported.cost_usd, Some(0.42));
        assert_eq!(receipt_reported.cost_source, CostSource::Provider);
        assert_eq!(receipt_silent.cost_usd, None);
        assert_eq!(receipt_silent.cost_source, CostSource::Unknown);

        // Another principal holds its own slot.
        scope(
            SpendScope::new(SpendPrincipal::Agent("other".into())),
            async {
                gate.admit(&registry, facts.clone(), 0)
                    .expect("another principal is not held up")
            },
        )
        .await;

        let refused = capped.refused.lock().unwrap();
        assert_eq!(refused.len(), 1, "the refusal the caller received");
        assert_eq!(refused[0].1.reason, UNPRICED_CALL_IN_FLIGHT);
        let settled = capped.inner.settled.lock().unwrap();
        assert_eq!(settled[1].1.cost.micros, Some(420_000));
        assert_eq!(settled[1].1.cost.source, CostSource::Provider);
    }

    /// The OpenRouter lookup of `generation` for `openrouter/unlisted`, at
    /// the local mock provider `base`. Never a real provider.
    fn lookup_at(base: &str, generation: &str) -> crate::LateCost {
        crate::LateCost::openrouter(
            crate::tls_client::test_seam::trust_store_independent_client(),
            base,
            generation,
            Vec::new(),
            "openrouter/unlisted",
            "openrouter/unlisted",
        )
        .expect("a generation id")
    }

    /// Poll (real time, bounded) until `done`.
    async fn eventually(what: &str, mut done: impl FnMut() -> bool) {
        let deadline = std::time::Instant::now() + crate::LOADED_WAIT;
        while !done() {
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for {what}"
            );
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    }

    /// A provider-priced call that ends with no figure, whose provider left
    /// a late lookup in the caller's probe, has that lookup run by its own
    /// ticket when it settles, and keeps the principal's slot until the
    /// figure lands: the next such call would otherwise be admitted against
    /// a spent figure that does not yet include this one. The figure lands
    /// as a correction, and frees the slot. Watched failing: with the slot
    /// freed at settle (round 2), the second call was admitted while the
    /// lookup was pending; with the ticket running no lookup (the handler ran
    /// it, on success only), no correction came and the slot stayed held.
    #[tokio::test]
    async fn a_provider_priced_call_holds_its_slot_until_its_late_figure_lands() {
        use wiremock::matchers::{method, path, query_param};
        use wiremock::{Mock, MockServer, ResponseTemplate};
        let provider = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/v1/generation"))
            .and(query_param("id", "gen-1"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"data": {"total_cost": 0.42}}))
                    .set_delay(Duration::from_millis(500)),
            )
            .mount(&provider)
            .await;
        let registry = provider_reports_registry();
        let facts = facts(&registry, "openrouter/unlisted");
        let capped = Arc::new(CappedGate::default());
        let gate = capped_installed(&capped);
        let probe = Arc::new(crate::LateCostProbe::default());
        crate::scope_late_cost(
            probe.clone(),
            scope(bot(), async {
                let mut first = gate
                    .admit(&registry, facts.clone(), 0)
                    .expect("admitted")
                    .expect("metered");
                // The provider named the generation; its cost comes later.
                probe.set(lookup_at(&provider.uri(), "gen-1"));
                first.served("openrouter/unlisted", None);
                let receipt = first.settle();
                assert_eq!(receipt.cost_source, CostSource::Unknown);

                let held = gate
                    .admit(&registry, facts.clone(), 0)
                    .expect_err("the slot is held until the late figure lands");
                assert!(matches!(
                    held,
                    InferenceError::SpendRefused { ref reason, .. }
                        if reason == UNPRICED_CALL_IN_FLIGHT
                ));

                eventually("the late figure", || {
                    !capped.inner.corrected.lock().unwrap().is_empty()
                })
                .await;
                eventually("the slot freed", || !gate.unpriced.held("agent:bot")).await;
                gate.admit(&registry, facts.clone(), 0)
                    .expect("admitted once the late figure landed")
                    .expect("metered");
            }),
        )
        .await;
        let corrected = capped.inner.corrected.lock().unwrap();
        assert_eq!(corrected.len(), 1);
        assert_eq!(corrected[0].0.call_id, "call-1", "the call it was for");
        assert_eq!(corrected[0].1.micros, Some(420_000));
        assert_eq!(corrected[0].1.source, CostSource::Provider);
    }

    /// A late figure that never comes (the provider never answers) holds the
    /// slot no longer than [`UNPRICED_LATE_FIGURE_HOLD`]; and a call that
    /// ends with no figure and no lookup pending frees the slot at once.
    /// Watched failing: with the hold's deadline ignored, the slot was still
    /// held after the hold.
    #[tokio::test(start_paused = true)]
    async fn a_late_figure_that_never_comes_holds_the_slot_for_a_bounded_time() {
        let slots = Arc::new(UnpricedSlots::default());
        let slot = slots.take("agent:bot").expect("free");
        let _release = slot
            .await_late_figure("call-1")
            .expect("the call still held the slot");
        tokio::time::advance(UNPRICED_LATE_FIGURE_HOLD - Duration::from_secs(1)).await;
        assert!(slots.held("agent:bot"), "still within the hold");
        tokio::time::advance(Duration::from_secs(2)).await;
        assert!(!slots.held("agent:bot"), "the hold ended");

        // No lookup pending: freed at settle.
        let registry = provider_reports_registry();
        let capped = Arc::new(CappedGate::default());
        let gate = capped_installed(&capped);
        let probe = Arc::new(crate::LateCostProbe::default());
        crate::scope_late_cost(
            probe,
            scope(bot(), async {
                let mut call = gate
                    .admit(&registry, facts(&registry, "openrouter/unlisted"), 0)
                    .expect("admitted")
                    .expect("metered");
                call.served("openrouter/unlisted", None);
                call.settle();
                assert!(!gate.unpriced.held("agent:bot"), "nothing to wait for");
            }),
        )
        .await;
    }

    /// A capped principal's call that ended after dispatch before its answer
    /// was read is charged its admission estimate (10 in + the 20-token
    /// ceiling, at $1): the provider may have billed output CAR never saw.
    /// That is a call dropped mid-flight (the caller's future cancelled or
    /// timed out), the caller's deadline elapsing, a cancelled run, and a 2xx
    /// whose body could not be read. An error that billed no output (a
    /// failure with no status, a provider's error) stays on its input.
    /// Watched failing: with an unsettled drop and these errors not marked
    /// cut off, every case settled at the input alone (#3605).
    #[tokio::test]
    async fn a_capped_call_that_ended_unread_is_charged_its_estimate() {
        let registry = registry_with("test/m", 1.0);
        let cases = [
            (None, 30_000_000, "dropped mid-call"),
            (
                Some(InferenceError::DeadlineExceeded {
                    applied_ms: 1_000,
                    elapsed_ms: 1_000,
                    last_error: String::new(),
                }),
                30_000_000,
                "the caller's deadline elapsed",
            ),
            (
                Some(InferenceError::ControlledTermination),
                30_000_000,
                "the run was cancelled",
            ),
            (
                Some(InferenceError::Transient {
                    status: Some(200),
                    message: String::new(),
                }),
                30_000_000,
                "a 200 whose body could not be read",
            ),
            (
                Some(InferenceError::Transient {
                    status: None,
                    message: String::new(),
                }),
                10_000_000,
                "a failure with no status",
            ),
            (
                Some(InferenceError::InferenceFailed(String::new())),
                10_000_000,
                "a provider error",
            ),
        ];
        for (error, expected, case) in cases {
            let capped = Arc::new(CappedGate::default());
            let gate = capped_installed(&capped);
            scope(bot(), async {
                let mut ticket = gate
                    .admit(&registry, facts(&registry, "test/m"), 0)
                    .unwrap()
                    .expect("a ticket");
                if let Some(error) = &error {
                    ticket.ended_with(error);
                }
                drop(ticket);
            })
            .await;
            let settled = capped.inner.settled.lock().unwrap();
            assert_eq!(settled[0].1.cost.micros, Some(expected), "{case}");
        }
    }

    /// A slot taken for a call the gate then refuses (a cap reached) is
    /// freed with the refusal, so the principal's next call is not blocked
    /// by a call that never ran.
    #[tokio::test]
    async fn a_refused_admission_frees_the_slot_it_took() {
        struct AtCap;
        impl SpendGate for AtCap {
            fn admit(
                &self,
                _scope: &SpendScope,
                _facts: &DispatchFacts,
            ) -> Result<SpendTicketId, SpendRefusal> {
                Err(SpendRefusal {
                    reason: "budget_reached".into(),
                    message: "budget reached until 3:00 PM EDT".into(),
                    data: None,
                })
            }
            fn capped(&self, _scope: &SpendScope) -> bool {
                true
            }
            fn settle(&self, _ticket: &SpendTicketId, _call: &SettledCall) {}
            fn correct(&self, _ticket: &SpendTicketId, _cost: CallCost) {}
        }
        let registry = provider_reports_registry();
        let gate = InstalledGate {
            gate: Arc::new(AtCap),
            require_scope: true,
            local_only: false,
            unpriced: Default::default(),
        };
        scope(bot(), async {
            let error = gate
                .admit(&registry, facts(&registry, "openrouter/unlisted"), 0)
                .expect_err("at the cap");
            assert!(matches!(
                error,
                InferenceError::SpendRefused { ref reason, .. } if reason == "budget_reached"
            ));
            assert!(!gate.unpriced.held("agent:bot"));
        })
        .await;
    }

    /// A model the catalog prices whose price yields no estimate for this
    /// call (here, a pricing registry that does not hold the row) is refused
    /// for a capped principal, with text that says so rather than "price
    /// unknown". Watched failing: with `unestimable_refusal` worded by
    /// `price_unknown_message` for every price, the message read "price
    /// unknown for test/m".
    #[tokio::test]
    async fn a_listed_model_with_no_estimate_is_refused_with_accurate_text() {
        let registry = registry_with("test/m", 0.000_001);
        let schema = registry.get("test/m").cloned();
        // The estimate is priced from a registry without the row.
        let empty = Arc::new(UnifiedRegistry::new_empty(std::env::temp_dir()));
        let facts = facts_for(&empty, schema.as_ref(), "test/m", CallKind::Tokens, 10, 20);
        assert_eq!(facts.price, PriceKnown::ListPrice);
        assert_eq!(facts.estimate.micros, None);

        let capped = Arc::new(CappedGate::default());
        let error = scope(bot(), async {
            capped_installed(&capped)
                .admit(&registry, facts, 0)
                .expect_err("no estimate to hold in flight")
        })
        .await;
        let InferenceError::SpendRefused {
            reason, message, ..
        } = error
        else {
            panic!("a spend refusal");
        };
        assert_eq!(reason, PRICE_UNKNOWN);
        assert_eq!(message, cost_unestimable_message("test/m"));
        assert!(!message.contains("price unknown"), "{message}");
    }

    /// A router candidate refused for being unpriced comes back unrecorded:
    /// nothing reaches the gate's refusal record until the caller records
    /// the refusal it actually receives. Watched failing: with the refusal
    /// recorded in `admit_candidate` (the round-1 `gate.refused` call before
    /// returning), `refused` held one row for the skipped candidate.
    #[tokio::test]
    async fn an_unpriced_candidate_is_refused_unrecorded_until_the_caller_receives_it() {
        let registry = provider_reports_registry();
        let mut facts = facts(&registry, "openrouter/unlisted");
        facts.price = PriceKnown::Unknown;
        let capped = Arc::new(CappedGate::default());
        let gate = capped_installed(&capped);
        let refusal = scope(bot(), async {
            match gate.admit_candidate(&registry, facts, 0) {
                Err(CandidateRefusal::Unrecorded(refusal)) => refusal,
                Err(CandidateRefusal::Final(error)) => panic!("final: {error}"),
                Ok(_) => panic!("admitted"),
            }
        })
        .await;
        assert!(
            capped.refused.lock().unwrap().is_empty(),
            "not recorded yet"
        );
        assert!(refusal.is(&refusal.error()));
        let error = refusal.record();
        assert!(matches!(
            error,
            InferenceError::SpendRefused { ref reason, .. } if reason == PRICE_UNKNOWN
        ));
        assert_eq!(capped.refused.lock().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn unscoped_remote_dispatch_is_refused_when_a_scope_is_required() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let error = installed(&gate, true)
            .admit(&registry, facts(&registry, "test/m"), 0)
            .expect_err("refused");
        assert!(matches!(
            error,
            InferenceError::SpendRefused { ref reason, ref message, .. }
                if reason == "no_scope" && message == NO_SCOPE_MESSAGE
        ));
        assert!(error.raised_before_dispatch());
        assert!(gate.admitted.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn unscoped_dispatch_runs_unmetered_when_no_scope_is_required() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let ticket = installed(&gate, false)
            .admit(&registry, facts(&registry, "test/m"), 0)
            .unwrap();
        assert!(ticket.is_none());
    }

    #[tokio::test]
    async fn unscoped_local_dispatch_is_not_refused() {
        let registry = local_registry();
        let gate = Arc::new(RecordingGate::default());
        let facts = facts(&registry, "local/m");
        assert!(facts.is_local);
        assert!(installed(&gate, true)
            .admit(&registry, facts, 0)
            .unwrap()
            .is_none());
    }

    /// The worker child's gate refuses a remote dispatch before the scope is
    /// looked at: with no scope (the child's normal state, where a plain gate
    /// is never consulted) and under one alike. Watched failing: with the
    /// `local_only` check removed from `InstalledGate::admit`, the unscoped
    /// call ran unmetered (`Ok(None)`) and the scoped one was admitted.
    #[tokio::test]
    async fn a_local_only_gate_refuses_a_remote_dispatch_with_or_without_a_scope() {
        let registry = registry_with("test/m", 1.0);
        let remote = facts(&registry, "test/m");
        assert!(!remote.is_local);
        let unscoped = local_only()
            .admit(&registry, remote.clone(), 0)
            .expect_err("refused");
        assert!(matches!(
            unscoped,
            InferenceError::SpendRefused { ref reason, ref message, .. }
                if reason == REMOTE_IN_LOCAL_WORKER && message == REMOTE_IN_LOCAL_WORKER_MESSAGE
        ));
        assert!(unscoped.raised_before_dispatch());
        let (scoped, receipts) = scope(
            SpendScope::new(SpendPrincipal::Agent("bot".into())),
            async {
                let error = local_only()
                    .admit(&registry, remote, 1)
                    .expect_err("refused under a scope too");
                (error, receipts())
            },
        )
        .await;
        assert!(matches!(
            scoped,
            InferenceError::SpendRefused { ref reason, .. } if reason == REMOTE_IN_LOCAL_WORKER
        ));
        assert!(
            receipts.is_empty(),
            "nothing was admitted, so nothing settled"
        );
    }

    /// A local-only SCOPE refuses a remote dispatch through an ordinary,
    /// uncapping gate, and still admits a local one: the unattended heal sweep
    /// runs under such a scope so it can never spend money with no one
    /// watching, whatever model a pin, the router or a failover reaches.
    #[tokio::test]
    async fn a_local_only_scope_refuses_remote_and_admits_local_dispatch() {
        let remote_registry = registry_with("test/m", 1.0);
        let local_registry = local_registry();
        let gate = Arc::new(RecordingGate::default());
        let unattended = SpendScope::new(SpendPrincipal::Car("heal".into())).local_only();

        let refused = scope(unattended.clone(), async {
            installed(&gate, true)
                .admit(&remote_registry, facts(&remote_registry, "test/m"), 0)
                .expect_err("remote refused under a local-only scope")
        })
        .await;
        assert!(matches!(
            refused,
            InferenceError::SpendRefused { ref reason, ref message, .. }
                if reason == REMOTE_IN_LOCAL_ONLY_SCOPE && message == REMOTE_IN_LOCAL_ONLY_SCOPE_MESSAGE
        ));
        assert!(refused.raised_before_dispatch());
        assert!(gate.admitted.lock().unwrap().is_empty());

        let local = scope(unattended, async {
            installed(&gate, true).admit(&local_registry, facts(&local_registry, "local/m"), 0)
        })
        .await;
        assert!(local.is_ok(), "a local model runs unattended");

        // Positive control: the same remote call under an ordinary scope.
        let ordinary = scope(SpendScope::new(SpendPrincipal::Car("heal".into())), async {
            installed(&gate, true).admit(&remote_registry, facts(&remote_registry, "test/m"), 0)
        })
        .await;
        assert!(ordinary.is_ok(), "only the local-only scope refuses");
    }

    /// The same gate lets an on-device dispatch through: unmetered with no
    /// scope, and under a scope a $0 ticket that settles to a receipt.
    #[tokio::test]
    async fn a_local_only_gate_admits_a_local_dispatch() {
        let registry = local_registry();
        let local = facts(&registry, "local/m");
        assert!(local.is_local);
        assert!(local_only()
            .admit(&registry, local.clone(), 0)
            .unwrap()
            .is_none());
        let receipts = scope(
            SpendScope::new(SpendPrincipal::Agent("bot".into())),
            async {
                let ticket = local_only()
                    .admit(&registry, local, 0)
                    .unwrap()
                    .expect("a ticket");
                drop(ticket);
                receipts()
            },
        )
        .await;
        assert_eq!(receipts.len(), 1);
        assert_eq!(receipts[0].model_used, "local/m");
    }

    #[tokio::test]
    async fn a_scoped_call_is_admitted_charged_and_receipted() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        let scope_in = SpendScope::new(SpendPrincipal::Agent("bot".into()))
            .with_keys(["label:orch/label-abc".to_string()]);
        let receipts = scope(scope_in.clone(), async {
            let mut ticket = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap()
                .expect("a ticket");
            let usage = TokenUsage {
                prompt_tokens: 3,
                completion_tokens: 4,
                total_tokens: 7,
                cache_read_input_tokens: 2,
                ..Default::default()
            };
            ticket.served("test/m", Some(&usage));
            drop(ticket);
            receipts()
        })
        .await;
        let (scope_seen, facts_seen) = gate.admitted.lock().unwrap()[0].clone();
        assert_eq!(scope_seen.principal, SpendPrincipal::Agent("bot".into()));
        assert_eq!(
            scope_seen.keys,
            vec!["agent:bot".to_string(), "label:orch/label-abc".to_string()]
        );
        // 10 in + 20 out at $1 a token: an upper-bound estimate.
        assert_eq!(facts_seen.estimate.micros, Some(30_000_000));
        let (ticket, settled) = gate.settled.lock().unwrap()[0].clone();
        assert_eq!(ticket.principal, "agent:bot");
        assert_eq!(settled.input_tokens, 5, "the cached prompt counts as input");
        assert_eq!(settled.output_tokens, 4);
        // Cached prompt tokens count as input but are priced at the model's
        // cache rate, which this fixture does not set: 3 in + 4 out at $1.
        assert_eq!(settled.cost.micros, Some(7_000_000));
        assert_eq!(settled.cost.source, CostSource::ListPrice);
        assert!(!settled.voided);
        assert_eq!(receipts.len(), 1);
        assert_eq!(receipts[0].call_id, "call-1");
        assert_eq!(receipts[0].cost_usd, Some(7.0));
        assert_eq!(receipts[0].cause, SpendCause::Direct);
    }

    /// A call that reached a provider and reported no usage settles at the
    /// input it sent, not at the output ceiling its admission estimate held.
    #[tokio::test]
    async fn a_dropped_ticket_settles_at_the_input_estimate() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        scope(SpendScope::new(SpendPrincipal::Operator), async {
            let ticket = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap();
            drop(ticket);
        })
        .await;
        let (admitted_scope, facts) = gate.admitted.lock().unwrap()[0].clone();
        assert_eq!(admitted_scope.principal, SpendPrincipal::Operator);
        assert_eq!(
            facts.estimate.micros,
            Some(30_000_000),
            "admitted at the ceiling"
        );
        let (_, settled) = gate.settled.lock().unwrap()[0].clone();
        assert_eq!(
            settled.cost.micros,
            Some(10_000_000),
            "settled at the input alone"
        );
        assert_eq!(settled.input_tokens, 10);
        assert_eq!(settled.output_tokens, 0);
    }

    /// A refusal whose response said what it used is settled on those
    /// figures, partial output included, not on the input estimate
    /// (car#3704).
    #[tokio::test]
    async fn a_refusal_is_charged_what_the_provider_reported() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        scope(SpendScope::new(SpendPrincipal::Operator), async {
            let mut ticket = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap()
                .unwrap();
            ticket.ended_with(&InferenceError::ContentRefused {
                provider: "anthropic".into(),
                kind: Some("refusal".into()),
                code: None,
                message: "declined".into(),
                usage: Some(TokenUsage {
                    prompt_tokens: 12,
                    completion_tokens: 7,
                    total_tokens: 19,
                    ..Default::default()
                }),
            });
            drop(ticket);
        })
        .await;
        let (_, settled) = gate.settled.lock().unwrap()[0].clone();
        assert!(!settled.voided);
        assert_eq!(settled.input_tokens, 12);
        assert_eq!(settled.output_tokens, 7);
    }

    #[tokio::test]
    async fn an_error_raised_before_dispatch_charges_nothing() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        let receipts = scope(SpendScope::new(SpendPrincipal::Operator), async {
            let mut ticket = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap()
                .unwrap();
            ticket.ended_with(&InferenceError::ProviderKeyMissing {
                provider: "test".into(),
                model: "test/m".into(),
                env_vars: vec!["TEST_KEY".into()],
                message: String::new(),
            });
            drop(ticket);
            receipts()
        })
        .await;
        let (_, settled) = gate.settled.lock().unwrap()[0].clone();
        assert!(settled.voided);
        assert_eq!(settled.cost.micros, None, "no cost to report");
        assert_eq!(settled.cost.source, CostSource::Unknown);
        assert!(receipts[0].voided);
        assert_eq!(receipts[0].cost_usd, None);
    }

    /// Watched failing: with the fallback cause applied only to a `direct`
    /// scope, a chat turn's second attempt stayed `chat`.
    #[tokio::test]
    async fn a_fallback_attempt_names_its_cause_whatever_the_requests_cause() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        scope(
            SpendScope::new(SpendPrincipal::Operator).with_cause(SpendCause::Chat),
            async {
                let _first = installed
                    .admit(&registry, facts(&registry, "test/m"), 0)
                    .unwrap();
                let _second = installed
                    .admit(&registry, facts(&registry, "test/m"), 1)
                    .unwrap();
            },
        )
        .await;
        {
            let admitted = gate.admitted.lock().unwrap();
            assert_eq!(admitted[0].0.cause, SpendCause::Chat);
            assert_eq!(admitted[1].0.cause, SpendCause::Fallback);
        }
        gate.admitted.lock().unwrap().clear();
        scope(SpendScope::new(SpendPrincipal::Operator), async {
            let _first = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap();
            let _second = installed
                .admit(&registry, facts(&registry, "test/m"), 1)
                .unwrap();
        })
        .await;
        let admitted = gate.admitted.lock().unwrap();
        assert_eq!(admitted[0].0.cause, SpendCause::Direct);
        assert_eq!(admitted[1].0.cause, SpendCause::Fallback);
    }

    #[tokio::test]
    async fn spawn_scoped_carries_the_scope_and_a_plain_spawn_does_not() {
        let carried = scope(SpendScope::new(SpendPrincipal::Car("test".into())), async {
            let scoped = spawn_scoped(async { current().map(|s| s.principal.key()) })
                .await
                .unwrap();
            let plain = tokio::spawn(async { current().map(|s| s.principal.key()) })
                .await
                .unwrap();
            (scoped, plain)
        })
        .await;
        assert_eq!(carried, (Some("car:test".to_string()), None));
    }

    #[tokio::test]
    async fn decision_class_survives_nested_work_and_scoped_spawns() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        scope(SpendScope::new(SpendPrincipal::Operator), async {
            with_decision_class(Some("scheduling replies".into()), async {
                spawn_scoped(async move {
                    scope(
                        SpendScope::new(SpendPrincipal::Car("helper".into())),
                        with_decision_class(None, async {
                            let ticket = installed
                                .admit(&registry, facts(&registry, "test/m"), 0)
                                .unwrap();
                            drop(ticket);
                        }),
                    )
                    .await;
                })
                .await
                .unwrap();
            })
            .await;
            assert!(receipts().is_empty());
            assert!(current().unwrap().decision_class.is_none());
        })
        .await;
        assert_eq!(
            gate.admitted.lock().unwrap()[0].0.decision_class.as_deref(),
            Some("scheduling replies")
        );
    }

    #[tokio::test]
    async fn with_cause_rescopes_and_shares_receipts() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        let outer = scope(SpendScope::new(SpendPrincipal::Operator), async {
            with_cause(SpendCause::Summary, async {
                let ticket = installed
                    .admit(&registry, facts(&registry, "test/m"), 0)
                    .unwrap();
                drop(ticket);
            })
            .await;
            receipts()
        })
        .await;
        assert_eq!(outer.len(), 1);
        assert_eq!(outer[0].cause, SpendCause::Summary);
        assert_eq!(
            gate.admitted.lock().unwrap()[0].0.cause,
            SpendCause::Summary
        );
    }

    /// A record finds the receipt of its own call by the engine trace, not
    /// the scope's newest; the scope keeps a bounded number. Watched
    /// failing: with `receipt_for_trace` returning the newest receipt, the
    /// first call's record carried the second call's id; with no bound, a
    /// scope of 100 calls held 100 receipts.
    #[tokio::test]
    async fn a_record_finds_its_own_calls_receipt_and_the_scope_is_bounded() {
        let registry = registry_with("test/m", 1.0);
        let gate = Arc::new(RecordingGate::default());
        let installed = installed(&gate, true);
        scope(SpendScope::new(SpendPrincipal::Operator), async {
            let mut first = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap()
                .unwrap();
            first.set_trace("t-1");
            let mut second = installed
                .admit(&registry, facts(&registry, "test/m"), 0)
                .unwrap()
                .unwrap();
            second.set_trace("t-2");
            // The later call settles first, as concurrent calls may.
            let second_receipt = second.settle();
            let first_receipt = first.settle();
            assert_eq!(
                receipt_for_trace("t-1").unwrap().call_id,
                first_receipt.call_id
            );
            assert_eq!(
                receipt_for_trace("t-2").unwrap().call_id,
                second_receipt.call_id
            );
            assert!(receipt_for_trace("t-9").is_none());
            for n in 0..RECEIPTS_KEPT + 10 {
                let mut ticket = installed
                    .admit(&registry, facts(&registry, "test/m"), 0)
                    .unwrap()
                    .unwrap();
                ticket.set_trace(&format!("t-bulk-{n}"));
                drop(ticket);
            }
            assert_eq!(
                receipts().len(),
                RECEIPTS_KEPT,
                "the scope keeps the newest"
            );
            assert!(
                receipt_for_trace("t-1").is_none(),
                "the oldest were dropped"
            );
            assert!(receipt_for_trace(&format!("t-bulk-{}", RECEIPTS_KEPT + 9)).is_some());
        })
        .await;
    }

    #[test]
    fn principal_keys_round_trip_the_meter_spelling() {
        assert_eq!(SpendPrincipal::Agent("a".into()).key(), "agent:a");
        assert_eq!(SpendPrincipal::Worker("wk_1".into()).key(), "worker:wk_1");
        assert_eq!(SpendPrincipal::Credential("c".into()).key(), "credential:c");
        assert_eq!(SpendPrincipal::Operator.key(), "operator");
        assert_eq!(SpendPrincipal::Car("selfheal".into()).key(), "car:selfheal");
        assert_eq!(
            SpendPrincipal::Declarative("d".into()).key(),
            "declarative:d"
        );
    }
}