From 0beb1f6d02f1c69978a908d9a85f304477dbf939 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 15:56:38 +0100 Subject: [PATCH] connector-spend: settle records the true cost with an overrun marker; prune window spend; checked reservation; replay before cost bound; settle failure frees the slot; typed ceiling and window Co-Authored-By: Claude Opus 5.5 --- crates/model-connector/src/contract.rs | 6 + crates/model-connector/src/fixtures.rs | 17 +- crates/model-connector/src/serve.rs | 11 +- crates/model-connector/src/service.rs | 523 +++++++++++++++--- .../wire_v3/carriage_invocation.frame | 2 +- .../wire_v3/transport_receipt_failed.mp | Bin 498 -> 517 bytes .../wire_v3/transport_result_transported.mp | Bin 934 -> 953 bytes .../model-connector/tests/reasoning_state.rs | 4 +- .../model-connector/tests/wire_golden_v3.rs | 1 + schemas/transport_receipt.json | 7 + schemas/transport_result.json | 7 + 11 files changed, 501 insertions(+), 77 deletions(-) diff --git a/crates/model-connector/src/contract.rs b/crates/model-connector/src/contract.rs index 760324d..5953b8e 100644 --- a/crates/model-connector/src/contract.rs +++ b/crates/model-connector/src/contract.rs @@ -488,6 +488,12 @@ pub struct TransportReceipt { /// What the request cost, in micro-USD. The service derives it from the /// outcome's usage and its price table; a backend never writes it. pub cost_micro_usd: u64, + /// How far `cost_micro_usd` exceeds what the service reserved for the + /// request: zero unless the backend reported usage above the worst case + /// the request allowed, which is a breach of the backend's contract. The + /// cost is still the true one and the caller's window is charged all of + /// it. The service derives this; a backend never writes it. + pub overrun_micro_usd: u64, pub outcome: TransportOutcome, } diff --git a/crates/model-connector/src/fixtures.rs b/crates/model-connector/src/fixtures.rs index 44d8ede..079a99c 100644 --- a/crates/model-connector/src/fixtures.rs +++ b/crates/model-connector/src/fixtures.rs @@ -50,7 +50,7 @@ impl Drop for TestStore { #[cfg(feature = "daemon")] pub(crate) struct FlakyStore { held: Arc>>, - pushes_allowed: AtomicUsize, + pushes_allowed: Arc, } #[cfg(feature = "daemon")] @@ -58,15 +58,21 @@ impl FlakyStore { pub(crate) fn new(pushes_allowed: usize) -> Self { Self { held: Arc::default(), - pushes_allowed: AtomicUsize::new(pushes_allowed), + pushes_allowed: Arc::new(AtomicUsize::new(pushes_allowed)), } } + /// The remaining writes this store accepts, shared with it after it has + /// been handed to a service. + pub(crate) fn allowance(&self) -> Arc { + self.pushes_allowed.clone() + } + /// A store over the same records that accepts `pushes_allowed` more writes. pub(crate) fn reopened(&self, pushes_allowed: usize) -> Self { Self { held: self.held.clone(), - pushes_allowed: AtomicUsize::new(pushes_allowed), + pushes_allowed: Arc::new(AtomicUsize::new(pushes_allowed)), } } } @@ -150,6 +156,7 @@ pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt { provider: "fixture-sse".to_string(), foreign_reasoning_dropped: 0, cost_micro_usd: FIXTURE_USAGE_COST, + overrun_micro_usd: 0, outcome: TransportOutcome::Completed { provider_response_id: Some("response-1".to_string()), stop_reason: StopReason::EndTurn, @@ -205,8 +212,8 @@ pub(crate) fn caller( max_concurrent_requests, 64 * 1024, 32_768, - CEILING, - WINDOW_MS, + SpendCeiling(CEILING), + SpendWindowMs(WINDOW_MS), ) .unwrap() } diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs index 6853f79..6d87f6e 100644 --- a/crates/model-connector/src/serve.rs +++ b/crates/model-connector/src/serve.rs @@ -16,7 +16,8 @@ use crate::envelope::{ frame_bytes_for_payload, read_transport_frame, write_transport_frame, }; use crate::service::{ - CallerAdmission, PriceTable, ProviderCall, TransportAdmission, TransportService, + CallerAdmission, PriceTable, ProviderCall, SpendCeiling, SpendWindowMs, TransportAdmission, + TransportService, }; /// The only provider seam of the connector kit. A connector binary implements @@ -139,8 +140,8 @@ pub fn load_admissions(callers: &[CallerConfig]) -> Result, caller.max_concurrent_requests, caller.max_payload_bytes, caller.max_output_tokens, - caller.spend_ceiling_micro_usd, - caller.spend_window_ms, + SpendCeiling(caller.spend_ceiling_micro_usd), + SpendWindowMs(caller.spend_window_ms), ) .map_err(ServeError::Service)?, ); @@ -858,8 +859,8 @@ mod tests { slots, payload, 32_768, - CEILING, - WINDOW_MS, + SpendCeiling(CEILING), + SpendWindowMs(WINDOW_MS), ) .unwrap() } diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index eea2fab..13ae039 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -17,6 +17,34 @@ use crate::envelope::{ encode_result_frame, encrypt_result, validate_envelope, }; +/// What a caller may spend in one window, in micro-USD. A distinct type from +/// [`SpendWindowMs`] so the pair cannot be passed swapped: +/// +/// ``` +/// use model_connector::{CallerAdmission, SpendCeiling, SpendWindowMs}; +/// CallerAdmission::new( +/// "caller-1", "a-connection-key", 1, ["model-1".to_string()], 1, 1, 1, +/// SpendCeiling(50_000_000), SpendWindowMs(3_600_000), +/// ) +/// .unwrap(); +/// ``` +/// +/// ```compile_fail +/// use model_connector::{CallerAdmission, SpendCeiling, SpendWindowMs}; +/// CallerAdmission::new( +/// "caller-1", "a-connection-key", 1, ["model-1".to_string()], 1, 1, 1, +/// SpendWindowMs(3_600_000), SpendCeiling(50_000_000), +/// ) +/// .unwrap(); +/// ``` +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SpendCeiling(pub u64); + +/// How long a spend window lasts, in milliseconds. Windows are fixed and +/// aligned to the unix epoch. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SpendWindowMs(pub u64); + pub struct CallerAdmission { caller_runtime_id: String, connection_key_epoch: u32, @@ -41,9 +69,10 @@ impl CallerAdmission { max_concurrent_requests: usize, max_payload_bytes: usize, max_output_tokens: u32, - spend_ceiling_micro_usd: u64, - spend_window_ms: u64, + spend_ceiling: SpendCeiling, + spend_window: SpendWindowMs, ) -> Result { + let (spend_ceiling_micro_usd, spend_window_ms) = (spend_ceiling.0, spend_window.0); let caller_runtime_id = caller_runtime_id.into(); require_id(&caller_runtime_id, "caller_runtime_id")?; let allowed_models = allowed_models.into_iter().collect::>(); @@ -122,10 +151,28 @@ pub type PriceTable = BTreeMap; const TOKENS_PER_RATE: u128 = 1_000_000; -fn ceil_micro_usd(rate_weighted_tokens: u128) -> u64 { +/// Rounds up to whole micro-USD; `None` when the amount does not fit a u64. +fn checked_ceil_micro_usd(rate_weighted_tokens: u128) -> Option { let whole = rate_weighted_tokens / TOKENS_PER_RATE; let rounded = whole + u128::from(!rate_weighted_tokens.is_multiple_of(TOKENS_PER_RATE)); - u64::try_from(rounded).unwrap_or(u64::MAX) + u64::try_from(rounded).ok() +} + +fn ceil_micro_usd(rate_weighted_tokens: u128) -> u64 { + checked_ceil_micro_usd(rate_weighted_tokens).unwrap_or(u64::MAX) +} + +impl Rates { + /// The dearer of each class. + fn dearest(self, other: Rates) -> Rates { + Rates { + input: self.input.max(other.input), + cache_write_short: self.cache_write_short.max(other.cache_write_short), + cache_write_long: self.cache_write_long.max(other.cache_write_long), + cache_read: self.cache_read.max(other.cache_read), + output: self.output.max(other.output), + } + } } fn rates_for(price: &ModelPrice, prompt_tokens: u128) -> &Rates { @@ -168,20 +215,29 @@ pub(crate) fn cost_micro_usd( /// The worst case `request` can cost, in micro-USD: its encoded length plus /// the model's overhead as input tokens (a token is at least one byte) at the /// dearest input rate the cache retention allows, plus `max_output_tokens` at -/// the output rate, rounded up once. +/// the output rate, rounded up once. A prompt that may cross the long-context +/// threshold is bounded at the dearer of the two rate sets, so a backend whose +/// usage prices above this broke its contract. `None` when the worst case does +/// not fit a u64: such a request can never be afforded. pub(crate) fn reservation_micro_usd( price: &ModelPrice, request: &ProviderRequest, -) -> Result { +) -> Result, ContractError> { let bound = (wire_encode(request)?.len() as u128) + u128::from(price.overhead_tokens); - let rates = rates_for(price, bound); + let rates = match &price.long_context { + Some(long) if bound > u128::from(long.above_prompt_tokens) => { + price.rates.dearest(long.rates) + } + _ => price.rates, + }; let input_rate = rates .input .max(rates.cache_read) - .max(write_rate(rates, request.cache_retention)); - let total = (bound * u128::from(input_rate)) - .saturating_add(u128::from(request.max_output_tokens) * u128::from(rates.output)); - Ok(ceil_micro_usd(total)) + .max(write_rate(&rates, request.cache_retention)); + let total = bound.checked_mul(u128::from(input_rate)).and_then(|input| { + input.checked_add(u128::from(request.max_output_tokens) * u128::from(rates.output)) + }); + Ok(total.and_then(checked_ceil_micro_usd)) } /// Moves `held` of a caller's window: releases `release`, then adds `charge`. @@ -200,6 +256,41 @@ fn add_spend( *held = held.saturating_sub(release).saturating_add(charge); } +/// Drops the windows of `caller_runtime_id` that nothing can still read: +/// those before `floor` that hold no live request's reservation. Admission +/// reads the current window alone, and a live request settles into the window +/// it was admitted in, so nothing older is kept. +fn prune_windows( + window_spend: &mut HashMap>, + live: &HashMap, + caller_runtime_id: &str, + floor: u64, +) { + if let Some(windows) = window_spend.get_mut(caller_runtime_id) { + windows.retain(|window, _| { + *window >= floor + || live.iter().any(|(identity, live_window)| { + live_window == window && identity.caller_runtime_id == caller_runtime_id + }) + }); + } +} + +/// `prune_windows` with the caller's newest window as the floor, for the +/// places that do not know the clock (open, settle). +fn prune_to_latest( + window_spend: &mut HashMap>, + live: &HashMap, + caller_runtime_id: &str, +) { + let latest = window_spend + .get(caller_runtime_id) + .and_then(|windows| windows.keys().next_back().copied()); + if let Some(latest) = latest { + prune_windows(window_spend, live, caller_runtime_id, latest); + } +} + #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct RequestIdentity { caller_runtime_id: String, @@ -212,9 +303,12 @@ enum ReplayState { /// caller's window until the request settles (an indeterminate one is /// billed at it). Active { reserved_micro_usd: u64 }, + /// `overrun_micro_usd` is what the charge exceeded the reservation by: a + /// backend contract breach, kept so a replayed stand-in still says so. Completed { response: TransportEnvelope, charged_micro_usd: u64, + overrun_micro_usd: u64, }, } @@ -270,14 +364,22 @@ impl ProviderCall { events: Vec, outcome: TransportOutcome, ) -> TransportResult { - TransportResult::transported( - events, + TransportResult::transported(events, { + let cost_micro_usd = self.cost_of(&outcome); TransportReceipt { - cost_micro_usd: self.cost_of(&outcome), + cost_micro_usd, + overrun_micro_usd: self.overrun_of(cost_micro_usd), outcome, ..self.receipt.clone() - }, - ) + } + }) + } + + /// How far `cost_micro_usd` exceeds what was reserved. The reservation is + /// the worst case the kit can compute, so a nonzero overrun means the + /// backend broke its contract (usage above its own request's bounds). + fn overrun_of(&self, cost_micro_usd: u64) -> u64 { + cost_micro_usd.saturating_sub(self.reserved_micro_usd) } /// What `outcome` costs: its usage priced, or the whole reservation when a @@ -314,7 +416,9 @@ pub enum TransportAdmission { pub struct TransportService { callers: HashMap, - live: HashSet, + /// The requests between claim and settlement, each with the window it was + /// admitted in. + live: HashMap, replay_store: CultCache, max_expiry_skew_ms: u64, provider: String, @@ -323,7 +427,9 @@ pub struct TransportService { /// Active records and the charges of its Completed ones. A cache of the /// replay records, re-derived at open and moved only after a record's put /// returns Ok; the records are the ledger. A request belongs to the window - /// it was admitted in, so settlement never moves it to another. + /// it was admitted in, so settlement never moves it to another. Only what + /// admission and settlement can still read is kept: the caller's newest + /// window and any older one a live request is reserved in. window_spend: HashMap>, } @@ -434,9 +540,14 @@ impl TransportService { held, ); } + let live = HashMap::new(); + let spenders: Vec = window_spend.keys().cloned().collect(); + for caller_runtime_id in &spenders { + prune_to_latest(&mut window_spend, &live, caller_runtime_id); + } Ok(Self { callers, - live: HashSet::new(), + live, replay_store, max_expiry_skew_ms, provider: provider.to_string(), @@ -506,17 +617,6 @@ impl TransportService { Ok(partition) => partition, Err(refusal) => return self.reply_refusal(&invocation, refusal, &security), }; - // The worst case is decided by the invocation alone, so it is refused - // before the replay lookup, like every other refusal of this kind. - let price = *self - .prices - .get(&invocation.request.model) - .ok_or_else(|| ServiceError::UnpricedModel(invocation.request.model.clone()))?; - let reserved_micro_usd = reservation_micro_usd(&price, &request)?; - if reserved_micro_usd > invocation.max_cost_micro_usd { - return self.reply_refusal(&invocation, Refusal::CostBound, &security); - } - let identity = RequestIdentity { caller_runtime_id: invocation.caller_runtime_id.clone(), request_id: invocation.request_id().to_string(), @@ -535,6 +635,7 @@ impl TransportService { ReplayState::Completed { response, charged_micro_usd, + overrun_micro_usd, } => { if fits_frame(response, max_frame_bytes)? { return Ok(TransportAdmission::Reply(response.clone())); @@ -546,7 +647,10 @@ impl TransportService { &security, max_frame_bytes, FailureKind::ResultSize, - *charged_micro_usd, + Charge { + cost_micro_usd: *charged_micro_usd, + overrun_micro_usd: *overrun_micro_usd, + }, )?; self.replay_store .put( @@ -555,6 +659,7 @@ impl TransportService { state: ReplayState::Completed { response: stand_in.clone(), charged_micro_usd: *charged_micro_usd, + overrun_micro_usd: *overrun_micro_usd, }, ..record.clone() }, @@ -563,7 +668,7 @@ impl TransportService { Ok(TransportAdmission::Reply(stand_in)) } ReplayState::Active { .. } => { - let refusal = if self.live.contains(&identity) { + let refusal = if self.live.contains_key(&identity) { Refusal::InFlight } else { Refusal::Indeterminate @@ -572,9 +677,20 @@ impl TransportService { } }; } + // The worst case is priced at today's table, so it is checked only for + // new work: a request already recorded is answered whatever it would + // cost now. + let price = *self + .prices + .get(&invocation.request.model) + .ok_or_else(|| ServiceError::UnpricedModel(invocation.request.model.clone()))?; + let reserved_micro_usd = match reservation_micro_usd(&price, &request)? { + Some(reserved) if reserved <= invocation.max_cost_micro_usd => reserved, + _ => return self.reply_refusal(&invocation, Refusal::CostBound, &security), + }; let active_for_caller = self .live - .iter() + .keys() .filter(|key| key.caller_runtime_id == invocation.caller_runtime_id) .count(); if active_for_caller >= max_concurrent_requests { @@ -584,6 +700,12 @@ impl TransportService { // checked after the replay lookup, so a recorded answer is replayed // whatever the caller has left. let window = now_unix_ms / spend_window_ms; + prune_windows( + &mut self.window_spend, + &self.live, + &identity.caller_runtime_id, + window, + ); if self .window_spend(&identity.caller_runtime_id, window) .saturating_add(reserved_micro_usd) @@ -613,7 +735,7 @@ impl TransportService { 0, reserved_micro_usd, ); - self.live.insert(identity); + self.live.insert(identity, window); Ok(TransportAdmission::Execute(Box::new(ExecutionClaim { call: ProviderCall { receipt: TransportReceipt { @@ -627,6 +749,7 @@ impl TransportService { foreign_reasoning_dropped, // Replaced by the backend's outcome in `transported`. cost_micro_usd: 0, + overrun_micro_usd: 0, outcome: TransportOutcome::Failed { kind: FailureKind::Provider, message: String::new(), @@ -658,7 +781,7 @@ impl TransportService { caller_runtime_id: invocation.caller_runtime_id.clone(), request_id: invocation.request_id().to_string(), }; - if !self.live.contains(&identity) { + if !self.live.contains_key(&identity) { return Err(ServiceError::NoActiveRequest); } let active = self @@ -688,21 +811,28 @@ impl TransportService { // An admissible result charges the cost its receipt carries (the real // cost, even when the answer is replaced by a stand-in that fits the // frame); a result the service refused charges the whole reservation. - let charged_micro_usd = match &result.disposition { - TransportDisposition::Transported { receipt, .. } if admissible => { - receipt.cost_micro_usd - } - _ => reserved_micro_usd, + let charge = match &result.disposition { + TransportDisposition::Transported { receipt, .. } if admissible => Charge { + cost_micro_usd: receipt.cost_micro_usd, + overrun_micro_usd: receipt.overrun_micro_usd, + }, + _ => Charge { + cost_micro_usd: reserved_micro_usd, + overrun_micro_usd: 0, + }, }; + let charged_micro_usd = charge.cost_micro_usd; let mut response = if admissible { - encrypt_result(&result, &security)? + encrypt_result(&result, &security).inspect_err(|_| { + self.live.remove(&identity); + })? } else { self.seal_stand_in( invocation, &security, max_frame_bytes, FailureKind::Provider, - charged_micro_usd, + charge, ) .inspect_err(|_| { self.live.remove(&identity); @@ -718,7 +848,7 @@ impl TransportService { &security, max_frame_bytes, FailureKind::ResultSize, - charged_micro_usd, + charge, ) .inspect_err(|_| { self.live.remove(&identity); @@ -733,12 +863,19 @@ impl TransportService { state: ReplayState::Completed { response: response.clone(), charged_micro_usd, + overrun_micro_usd: charge.overrun_micro_usd, }, admitted_at_unix_ms: active.admitted_at_unix_ms, }; + // A failed settle write frees the caller's slot and answers the typed + // store failure; the Active record stays, so the reservation stays + // counted and a retry is told the request is indeterminate. self.replay_store .put(replay_key(&identity), &replay_record) - .map_err(replay_store_error)?; + .map_err(replay_store_error) + .inspect_err(|_| { + self.live.remove(&identity); + })?; // Settled: the window the request was admitted in holds the charge in // place of the reservation. A failed put above leaves the reservation // counted, as the store does. @@ -750,6 +887,11 @@ impl TransportService { charged_micro_usd, ); self.live.remove(&identity); + prune_to_latest( + &mut self.window_spend, + &self.live, + &identity.caller_runtime_id, + ); Ok(response) } @@ -760,7 +902,7 @@ impl TransportService { security: &TransportKey, max_frame_bytes: usize, kind: FailureKind, - cost_micro_usd: u64, + charge: Charge, ) -> Result { let (_, foreign_reasoning_dropped) = partition_reasoning(&invocation.request, &self.provider) @@ -774,7 +916,7 @@ impl TransportService { invocation.provider_request_sha256, foreign_reasoning_dropped, kind, - cost_micro_usd, + charge, ), security, )?; @@ -796,6 +938,7 @@ impl TransportService { && receipt.provider == self.provider && receipt.foreign_reasoning_dropped == claim.call.receipt.foreign_reasoning_dropped && receipt.cost_micro_usd == claim.call.cost_of(&receipt.outcome) + && receipt.overrun_micro_usd == claim.call.overrun_of(receipt.cost_micro_usd) } fn reply_refusal( @@ -831,6 +974,10 @@ pub(crate) fn result_size_failure( foreign_reasoning_dropped: u32, cost_micro_usd: u64, ) -> TransportResult { + let charge = Charge { + cost_micro_usd, + overrun_micro_usd: 0, + }; stand_in_failure( caller_runtime_id, request_id, @@ -839,10 +986,17 @@ pub(crate) fn result_size_failure( provider_request_sha256, foreign_reasoning_dropped, FailureKind::ResultSize, - cost_micro_usd, + charge, ) } +/// What a request cost, and by how much that exceeded its reservation. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Charge { + cost_micro_usd: u64, + overrun_micro_usd: u64, +} + /// The stand-in failure for `kind`: `ResultSize` for a result that cannot be /// carried, `Provider` for one the service refused. // One argument per fact the stand-in repeats from the invocation it answers. @@ -855,7 +1009,7 @@ fn stand_in_failure( provider_request_sha256: [u8; 32], foreign_reasoning_dropped: u32, kind: FailureKind, - cost_micro_usd: u64, + charge: Charge, ) -> TransportResult { let message = match kind { FailureKind::ResultSize => "provider result exceeded its byte bound", @@ -876,7 +1030,8 @@ fn stand_in_failure( model: model.to_string(), provider: STAND_IN_PROVIDER.to_string(), foreign_reasoning_dropped, - cost_micro_usd, + cost_micro_usd: charge.cost_micro_usd, + overrun_micro_usd: charge.overrun_micro_usd, outcome: TransportOutcome::Failed { kind, message: message.to_string(), @@ -1184,8 +1339,8 @@ mod tests { 1, 64 * 1024, 32_768, - CEILING, - WINDOW_MS, + SpendCeiling(CEILING), + SpendWindowMs(WINDOW_MS) ) .unwrap()], 5_000, @@ -1285,8 +1440,8 @@ mod tests { 4, max_payload_bytes, 32_768, - CEILING, - WINDOW_MS, + SpendCeiling(CEILING), + SpendWindowMs(WINDOW_MS), ) .unwrap(); TransportService::open( @@ -1440,6 +1595,7 @@ mod tests { record.state = ReplayState::Completed { response: encrypt_result(&result, &security(KEY)).unwrap(), charged_micro_usd: 0, + overrun_micro_usd: 0, }; service.replay_store.put(key, &record).unwrap(); assert_eq!( @@ -1459,8 +1615,8 @@ mod tests { 4, 64 * 1024, 32_768, - CEILING, - WINDOW_MS, + SpendCeiling(CEILING), + SpendWindowMs(WINDOW_MS), ) }; let at = "i".repeat(MAX_IDENTITY_BYTES); @@ -2199,8 +2355,8 @@ mod tests { 8, 65_536, 32_768, - ceiling, - window_ms, + SpendCeiling(ceiling), + SpendWindowMs(window_ms), ) .unwrap()], 5_000, @@ -2326,11 +2482,15 @@ mod tests { let length = wire_encode(&request).unwrap().len() as u64; let bound = length + 7; assert_eq!( - reservation_micro_usd(&price(bound, 7), &request).unwrap(), + reservation_micro_usd(&price(bound, 7), &request) + .unwrap() + .unwrap(), expected_reservation(&request, &fixture_price().rates, 7) ); assert_eq!( - reservation_micro_usd(&price(bound - 1, 7), &request).unwrap(), + reservation_micro_usd(&price(bound - 1, 7), &request) + .unwrap() + .unwrap(), expected_reservation(&request, &long, 7) ); } @@ -2394,7 +2554,7 @@ mod tests { let mut request = request(); request.cache_retention = retention; assert_eq!( - reservation_micro_usd(&price, &request).unwrap(), + reservation_micro_usd(&price, &request).unwrap().unwrap(), expected_reservation(&request, &rates, overhead_tokens), "{rates:?} {retention:?}" ); @@ -2407,7 +2567,7 @@ mod tests { for max_output_tokens in [1, 1_000, 1_001_000] { let mut request = request(); request.max_output_tokens = max_output_tokens; - let amount = reservation_micro_usd(&price, &request).unwrap(); + let amount = reservation_micro_usd(&price, &request).unwrap().unwrap(); assert_eq!( amount, expected_reservation(&request, &price.rates, 100), @@ -2684,7 +2844,8 @@ mod tests { drop(service); let mut service = open_limited(&store, FIXTURE_USAGE_COST + 2 * reserved, window); - assert_eq!(service.window_spend(CALLER, 0), FIXTURE_USAGE_COST); + // Only the current window is kept: the earlier one is pruned. + assert_eq!(service.window_spend(CALLER, 0), 0); assert_eq!( service.window_spend(CALLER, 1), FIXTURE_USAGE_COST + reserved @@ -2741,8 +2902,8 @@ mod tests { 4, 64 * 1024, 32_768, - ceiling, - window_ms, + SpendCeiling(ceiling), + SpendWindowMs(window_ms), ) }; assert!(admit(1, 1).is_ok()); @@ -2750,6 +2911,240 @@ mod tests { assert!(matches!(admit(1, 0), Err(ServiceError::InvalidAdmission))); } + // ---- settle past the reservation, pruning, extremes ---- + + fn overrun_result(call: &ProviderCall, text: &str) -> TransportResult { + let mut outcome = receipt(&invocation_for(CALLER, "outcome")).outcome; + let TransportOutcome::Completed { usage, .. } = &mut outcome else { + unreachable!(); + }; + // 2 million output tokens at 15,000,000 per million: 30,000,000. + *usage = self::usage(0, 0, 0, 2 * MTOK); + call.transported( + vec![TransportEvent { + sequence: 0, + payload: TransportEventPayload::TextDelta { + text: text.to_string(), + }, + }], + outcome, + ) + } + + const OVERRUN_COST: u64 = 30_000_000; + + #[test] + fn an_overrun_settles_at_the_true_cost_and_the_receipt_says_so() { + let reserved = reservation(); + let store = TestStore::new(); + let mut service = open_limited(&store, 2 * reserved, 3_600_000); + let claim = claim_at(&mut service, "request-1", 1_000); + let result = overrun_result(claim.call(), "done"); + let invocation = invocation_at("request-1", 1_000, MAX_COST); + let response = service.complete(*claim, result, FRAME).unwrap(); + let receipt = receipt_of(&response, &invocation); + // The provider's own answer, not a stand-in, carrying the true cost + // and the marker. + assert_eq!(receipt.provider, "fixture-sse"); + assert_eq!(receipt.cost_micro_usd, OVERRUN_COST); + assert_eq!(receipt.overrun_micro_usd, OVERRUN_COST - reserved); + assert_eq!(charged_of(&service, "request-1"), OVERRUN_COST); + assert_eq!(service.window_spend(CALLER, 0), OVERRUN_COST); + // The ceiling holds for every later request in the window. + assert_eq!( + refusal_at(&mut service, "request-2", 1_000, MAX_COST), + Refusal::SpendCeiling + ); + // A result inside its reservation carries no marker. + drop(service); + let mut service = open_service(&store, 4); + let claim = claim_at(&mut service, "request-3", 1_000); + let result = text_result(claim.call(), "done"); + let invocation = invocation_at("request-3", 1_000, MAX_COST); + let response = service.complete(*claim, result, FRAME).unwrap(); + assert_eq!(receipt_of(&response, &invocation).overrun_micro_usd, 0); + } + + #[test] + fn an_overrun_survives_the_result_size_stand_in_and_its_replay() { + let reserved = reservation(); + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let claim = claim_at(&mut service, "request-1", 1_000); + let big = overrun_result(claim.call(), &"y".repeat(4000)); + let invocation = invocation_at("request-1", 1_000, MAX_COST); + let response = service.complete(*claim, big, 2000).unwrap(); + let receipt = receipt_of(&response, &invocation); + assert_eq!(receipt.provider, STAND_IN_PROVIDER); + assert_eq!(receipt.cost_micro_usd, OVERRUN_COST); + assert_eq!(receipt.overrun_micro_usd, OVERRUN_COST - reserved); + + let claim = claim_at(&mut service, "request-2", 1_000); + let big = overrun_result(claim.call(), &"y".repeat(4000)); + service.complete(*claim, big, FRAME).unwrap(); + drop(service); + let mut service = open_service(&store, 4); + let invocation = invocation_at("request-2", 1_000, MAX_COST); + let TransportAdmission::Reply(smaller) = service + .begin(&raw_envelope(&invocation), 1_000, 2000) + .unwrap() + else { + panic!("expected a replay"); + }; + let receipt = receipt_of(&smaller, &invocation); + assert_eq!(receipt.provider, STAND_IN_PROVIDER); + assert_eq!(receipt.cost_micro_usd, OVERRUN_COST); + assert_eq!(receipt.overrun_micro_usd, OVERRUN_COST - reserved); + } + + #[test] + fn a_backend_overrun_misreport_is_answered_with_the_stand_in() { + let reserved = reservation(); + for (index, overrun) in [1, 0].into_iter().enumerate() { + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let id = format!("request-{index}"); + let claim = claim_at(&mut service, &id, 1_000); + // The first result is inside its reservation and claims an overrun; + // the second overran and claims none. + let mut result = if index == 0 { + text_result(claim.call(), "done") + } else { + overrun_result(claim.call(), "done") + }; + let TransportDisposition::Transported { receipt, .. } = &mut result.disposition else { + unreachable!(); + }; + receipt.overrun_micro_usd = overrun; + let invocation = invocation_at(&id, 1_000, MAX_COST); + let response = service.complete(*claim, result, FRAME).unwrap(); + let receipt = receipt_of(&response, &invocation); + assert_eq!(receipt.provider, STAND_IN_PROVIDER, "{index}"); + assert_eq!(receipt.cost_micro_usd, reserved, "{index}"); + assert_eq!(receipt.overrun_micro_usd, 0, "{index}"); + } + } + + #[test] + fn the_window_map_stays_bounded_live_and_after_reopen() { + let window = 10; + let store = TestStore::new(); + let mut service = open_limited(&store, 1 << 40, window); + let held = + |service: &TransportService| service.window_spend.get(CALLER).map_or(0, |w| w.len()); + for index in 0..300u64 { + let id = format!("request-{index}"); + let claim = claim_at(&mut service, &id, 1_000 + index * window); + let result = text_result(claim.call(), "done"); + service.complete(*claim, result, FRAME).unwrap(); + assert_eq!(held(&service), 1, "{index}"); + } + // A request still in flight keeps its own window until it settles. + let early = claim_at(&mut service, "request-e", 1_000 + 300 * window); + claim_at(&mut service, "request-l", 1_000 + 305 * window); + assert_eq!(held(&service), 2); + let result = text_result(early.call(), "done"); + service.complete(*early, result, FRAME).unwrap(); + assert_eq!(held(&service), 1); + drop(service); + let service = open_limited(&store, 1 << 40, window); + assert_eq!(held(&service), 1); + assert_eq!( + service.window_spend(CALLER, 1_000 / window + 305), + reservation() + ); + } + + #[test] + fn a_reservation_that_overflows_is_refused_cost_bound_not_wrapped() { + for price in [ + ModelPrice { + overhead_tokens: u64::MAX, + ..flat_price(u64::MAX) + }, + ModelPrice { + overhead_tokens: u64::MAX, + ..flat_price(1 << 40) + }, + ] { + assert_eq!(reservation_micro_usd(&price, &request()).unwrap(), None); + let store = TestStore::new(); + let mut service = TransportService::open( + &store.path, + "fixture-sse", + PriceTable::from([("gpt-5.4".to_string(), price)]), + [caller(CALLER, KEY, 4)], + 5_000, + ) + .unwrap(); + // Not even a caller that bounds its request at u64::MAX can afford it. + assert_eq!( + refusal_at(&mut service, "request-1", 1_000, u64::MAX), + Refusal::CostBound + ); + assert!(record_of(&service, "request-1").is_none()); + } + } + + #[test] + fn a_recorded_answer_replays_after_a_price_raise() { + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let (admission, _) = begin_at(&mut service, "request-1", 1_000, reservation()); + let TransportAdmission::Execute(claim) = admission else { + panic!("expected an execution claim"); + }; + let result = text_result(claim.call(), "done"); + service.complete(*claim, result, FRAME).unwrap(); + drop(service); + + let mut dear = fixture_price(); + dear.rates.output *= 2; + let mut service = TransportService::open( + &store.path, + "fixture-sse", + PriceTable::from([("gpt-5.4".to_string(), dear)]), + [caller(CALLER, KEY, 4)], + 5_000, + ) + .unwrap(); + let (admission, invocation) = begin_at(&mut service, "request-1", 1_000, reservation()); + let TransportAdmission::Reply(response) = admission else { + panic!("expected the recorded answer"); + }; + assert_eq!( + receipt_of(&response, &invocation).cost_micro_usd, + FIXTURE_USAGE_COST + ); + // New work is still checked against today's prices. + assert_eq!( + refusal_at(&mut service, "request-2", 1_000, reservation()), + Refusal::CostBound + ); + } + + #[test] + fn a_settle_write_failure_releases_the_slot_and_keeps_the_reservation() { + let store = FlakyStore::new(1); + let allowance = store.allowance(); + let mut service = open_flaky(store); + let claim = claim_at(&mut service, "request-1", 1_000); + let result = text_result(claim.call(), "done"); + assert!(matches!( + service.complete(*claim, result, FRAME), + Err(ServiceError::ReplayStore(_)) + )); + assert!(service.live.is_empty()); + assert_eq!(service.window_spend(CALLER, 0), reservation()); + // The store recovers: the caller's only slot is free again, and the + // unsettled request is told it is indeterminate, not in flight. + allowance.store(5, std::sync::atomic::Ordering::SeqCst); + claim_at(&mut service, "request-2", 1_000); + assert_eq!(service.window_spend(CALLER, 0), 2 * reservation()); + let (admission, invocation) = begin_at(&mut service, "request-1", 1_000, MAX_COST); + assert_eq!(refusal_of(admission, &invocation), Refusal::Indeterminate); + } + fn open_flaky(store: FlakyStore) -> TransportService { TransportService::with_store( store, diff --git a/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame b/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame index 5950d4b..b531c71 100644 --- a/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame +++ b/crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame @@ -1 +1 @@ -‡©messageId®golden-request©operation¦invoke§payloadÚDhqlzY2hlbWFfaWTZJGdhbWVjdWx0Lm1vZGVsLnRyYW5zcG9ydF9lbnZlbG9wZS52M7FjYWxsZXJfcnVudGltZV9pZK1nb2xkZW4tY2FsbGVyqnJlcXVlc3RfaWSuZ29sZGVuLXJlcXVlc3SsbWVzc2FnZV9raW5kqkludm9jYXRpb26lbm9uY2WcR8ylzIDMx8ysBMyLER9vzL8cqmNpcGhlcnRleHTFA0eCklvJd0ylcRitrB12Lz0CutMrwr519cC0vRtdkl3zrWX37ymWvqDsLVdUeDjc0FoLaMAH1CRD6BzwQHbRw//no2hBiTe8hkS5Mdej3DWJ7cYvzPNwgJeCFIp7qv5eKfsgiVJMXnWSQN9TYZwBQ6BpstwWiC+/iwbSruycp7ERWoR9PYC8CTzDKgSodkAJ08z9k1ItLhH25mhBy70Mg4p/HuTOC5CIwTu0waGP35+gfkG/RstpO1NDzMsRkdLnMaAl4N6icBYN1dllTwbeJM7EaVeUpZgJ4u7EVrRDgfCnZj8LdXJcIHn8KkXD0P3ZIF7BIoQeF9tbVzZttNYDv0XBtprg5OwztQYoRD2o0AxGKSyVKeOFIC/viF9CaVVQLQo2pqnuL1zqvNbqPmOQgeP4mbOEQbVem0lXh0cQocrpvKiDlFit2+OYHMSnVldHGcdhEHw+chy8HgtmJUlgnsKoLQlBiI5Y5ahJqUSbJdwrJg2Lfj5TZx0NgVCY4onWBfvD9jdayUtm6Fq457cAQ18wfrvZn9kUe4y5LheDPTtNQZoROUKqRTuF/8xP7fpLR4lJPi1OZBfWbiydbq+nueM1YaZgNH9VaEkeArWqTxy532eeMTDVw2ENVqXgrUEa/m7pLj1YBwdoLNpxXJVhviXO8oqwPQM/Fh0RwFlcjiJ+vBSDMZbw4icuQNZfJ4zcDRm95Xu9Js5H2TtbA1rb6p1wFjwKSFTXe5gXeT+P1YcDlBONOtGIHvYOuVhGrjWL9xrXqtml2kes8Zp9QFOOw/bd9YoQtpmbQQl4ZLM41nyZC6br7HTz9hacRslLuUsA3Cf18raaSk8JuomQIoET5Dg/DZtR9Nb0JNVnkLoKZrTCXuTZ9JuxvrOk9NmH7/tdj3tt2de3VW5JnUm/X2qBMmaguEkILci5KDS90Ohc1/NbGbhQeBGGlHOR4j7NeOqsZwA1rCkjB98X4sukSdWexfeiZee88pVMevdH5OJVCYZUHdRqG5pL8hb7U8JsySuUdy9v+S1OolaALaVoQ89MTHT2lxo7dJ1gGxeoEhPsPNDhdk7jjKmsQeM4SZB1pX4vYyUK3tUlPFwAmuz58rhL2ZicqwD97b2IDw==¯payloadEncoding²messagepack-base64­payloadSchemaÙ$gamecult.model.transport_envelope.v3­schemaVersion¼cultnet.operation_request.v0©serviceId¸gamecult.model.connector \ No newline at end of file +‡©messageId®golden-request©operation¦invoke§payloadÚDhqlzY2hlbWFfaWTZJGdhbWVjdWx0Lm1vZGVsLnRyYW5zcG9ydF9lbnZlbG9wZS52M7FjYWxsZXJfcnVudGltZV9pZK1nb2xkZW4tY2FsbGVyqnJlcXVlc3RfaWSuZ29sZGVuLXJlcXVlc3SsbWVzc2FnZV9raW5kqkludm9jYXRpb26lbm9uY2WczO7MxRrMxw7MhTh7SMzwzNB2qmNpcGhlcnRleHTFA0d65EskLG6kIwJ/G+XDK03gZOGtLFi6vZHJbI+/KpRLwnLT0TPF7RQa0KD98CrW+VwkKY3sRa4+Z9+/5OdPq1YmDFwbxvMVjZsFjXXw84tzvuLKHsb/3kQSEvAlFE4K6FXHvznLcMo4CnGUzoXwNGALf+xDTqFG0aFbdG4xGynV/kttsL6f34F4+l6xiXMUy87Q8/1UyOK/YaVxMMmlT3tdTllhq8YU8Dvfq0KuMaj/f9Anya8A5bJwlxafZPtc53RoH1/oNjLdaDqvwrTEnTB9TvKlnEOxKyyKhmmwGaC03qqPJXRaj8cDQO6EHDbML4J0zMN8TvssQISd5UCeVMkvDGjmvEOmRUoPk1/SnSrIHdha4+6rU4lUSumloLmlx5vv++Mq4AqzOdd/vPB6Ov/AoF32X27o4RI6H3xCKKHjKC4m3+cVjXCrOcGojuRf9yeDFkgbiPaC31hmSc8rAm6HO/NcaIHvx+G+6AaurAEhDLdF6DdrXb4xwjuv6yWQKihNvLXpnwmhRFkSeD5lS6FHW+CoOIdq/RBnFFyDrboTPWFoxQYIrBuqFoHDOajIz7dXYif0sf9sB9JBrDhlcbTpiGcyURfzDhW55NOPqKFGi6ropaJ1/SyLP3f7E57kA22aQ2Op7o36Tu06Pnzo2wI7UsL/Qz27F+AGAukQCvTJyjxARiGr3TRpChjVXx6GXhLrG1T9P+Mrhyrt9vwwhboN8qz1uvDZwKhGWBKXPXkxluq/NY1DwB3JTCCZfEwukWjMZbqiINLEKvW9BStcBR1kfXcK6sQvn/F0g9r81e9brbe+w3U24GERnHwgruOqytTeax/xPbQ7uISV48g5L8DAYf788KhUXwI43QDHNBdBUyo0rCsgoWJKeO05fcAgKgYYGsyLtMW10Vstw3q4POyn3o7rrBz4S2cnl2iZgr0ixg/KgF0V+loEVuj/6rYrcW0SLCHwV1+w1ClFMn0ZPsrz28EoZMA617G0Rrch66cS5uI9/Zf/wLX3z6n7xoPLNXk51ZsLSZ2bqEQkcBJbW/dT5MdH6anWVogRDvIW1coUbkIq9l+pkQSuelWsfakX4UQwYuUD0UwNbqxHiGwlEFj0f1bdhEj26Q==¯payloadEncoding²messagepack-base64­payloadSchemaÙ$gamecult.model.transport_envelope.v3­schemaVersion¼cultnet.operation_request.v0©serviceId¸gamecult.model.connector \ No newline at end of file diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp index c71c5bb23ac6ce99cbfca921bf98c3acab0ace2f..385a3188e4656e1de2360443cb81bdd64346d8da 100644 GIT binary patch delta 34 pcmeyw+{(h(y^+zEQFvp1S!z*HX>sK4C4R* delta 14 VcmZo=`NYiFwUN=6adIN#5&$Dy1fu`| diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_result_transported.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_result_transported.mp index 22e85aa995af5b08f2ef0de535731ef9cbfbc55b..79e6ebd3cd6a3a79292b265afe20da18941a2f1a 100644 GIT binary patch delta 38 ucmZ3+zLR~!c1FhT%{v%<8HG3Im!%dJmFC6gW+oTq$CnnTL{4U9UI+jnqYkP7 delta 18 acmdnVzKngtc1Fgo%{v%<87E6HF9ZNV+6HO> diff --git a/crates/model-connector/tests/reasoning_state.rs b/crates/model-connector/tests/reasoning_state.rs index 546dd7b..f42e0f1 100644 --- a/crates/model-connector/tests/reasoning_state.rs +++ b/crates/model-connector/tests/reasoning_state.rs @@ -106,8 +106,8 @@ fn start_with(backend: Arc, max_concurrent: usize) -> Conne max_concurrent, 64 * 1024, 32_768, - 1 << 40, - 3_600_000, + SpendCeiling(1 << 40), + SpendWindowMs(3_600_000), ) .unwrap(); let limits = ServeLimits { diff --git a/crates/model-connector/tests/wire_golden_v3.rs b/crates/model-connector/tests/wire_golden_v3.rs index 7eeef18..9364035 100644 --- a/crates/model-connector/tests/wire_golden_v3.rs +++ b/crates/model-connector/tests/wire_golden_v3.rs @@ -96,6 +96,7 @@ fn receipt(outcome: TransportOutcome) -> TransportReceipt { provider: "anthropic-messages".to_string(), foreign_reasoning_dropped: 2, cost_micro_usd: 1_234_567, + overrun_micro_usd: 89, outcome, } } diff --git a/schemas/transport_receipt.json b/schemas/transport_receipt.json index e0afcfb..01ff920 100644 --- a/schemas/transport_receipt.json +++ b/schemas/transport_receipt.json @@ -35,6 +35,12 @@ "outcome": { "$ref": "#/$defs/TransportOutcome" }, + "overrun_micro_usd": { + "description": "How far `cost_micro_usd` exceeds what the service reserved for the\nrequest: zero unless the backend reported usage above the worst case\nthe request allowed, which is a breach of the backend's contract. The\ncost is still the true one and the caller's window is charged all of\nit. The service derives this; a backend never writes it.", + "type": "integer", + "format": "uint64", + "minimum": 0 + }, "provider": { "type": "string" }, @@ -67,6 +73,7 @@ "provider", "foreign_reasoning_dropped", "cost_micro_usd", + "overrun_micro_usd", "outcome" ], "$defs": { diff --git a/schemas/transport_result.json b/schemas/transport_result.json index 183b913..e85fd51 100644 --- a/schemas/transport_result.json +++ b/schemas/transport_result.json @@ -314,6 +314,12 @@ "outcome": { "$ref": "#/$defs/TransportOutcome" }, + "overrun_micro_usd": { + "description": "How far `cost_micro_usd` exceeds what the service reserved for the\nrequest: zero unless the backend reported usage above the worst case\nthe request allowed, which is a breach of the backend's contract. The\ncost is still the true one and the caller's window is charged all of\nit. The service derives this; a backend never writes it.", + "type": "integer", + "format": "uint64", + "minimum": 0 + }, "provider": { "type": "string" }, @@ -346,6 +352,7 @@ "provider", "foreign_reasoning_dropped", "cost_micro_usd", + "overrun_micro_usd", "outcome" ] },