connector-spend: delete the public admission_window_start_unix_ms, which no operator surface carries (s4.catch-up-instant-reaches-no-operator)

The operator's view of a clock-locked caller belongs to follow-up connector-spend-readout; ClockBehindAdmissionWindow stays a plain refusal.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Cargo.lock

@@ -490,6 +490,7 @@ name = "model-connector"
 version = "0.1.0"
 dependencies = [
  "aes-gcm",
+ "anyhow",
  "base64",
  "cultcache-rs",
  "cultnet-rs",

crates/model-connector/Cargo.toml

@@ -26,3 +26,6 @@ serde_json = "1"
 sha2 = "0.10"
 thiserror = "2"
 zeroize = "1"
+
+[dev-dependencies]
+anyhow = "1"

crates/model-connector/src/contract.rs

@@ -190,9 +190,10 @@ pub enum FailureKind {
     Provider,
 }
 
-/// Token accounting for one completed request. Cache reads and writes are
-/// counted apart from plain input; `reasoning_tokens` is absent when the
-/// provider does not report it.
+/// Token accounting for one request, as the backend reports it; the service
+/// prices it. Cache reads and writes are counted apart from plain input.
+/// `output_tokens` includes reasoning tokens: `reasoning_tokens` is a detail
+/// (absent when the provider does not report it) and is never priced apart.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(deny_unknown_fields)]
 pub struct Usage {
@@ -251,6 +252,10 @@ pub struct TransportInvocation {
     pub expires_at_unix_ms: u64,
     pub native_request_sha256: [u8; 32],
     pub provider_request_sha256: [u8; 32],
+    /// The most this request may cost, in micro-USD: the caller's own bound
+    /// on it. Required nonzero. It is part of the invocation's identity, so a
+    /// changed budget under a recorded request id is a `ReplayConflict`.
+    pub max_cost_micro_usd: u64,
     pub request: ProviderRequest,
 }
 
@@ -259,6 +264,7 @@ impl TransportInvocation {
         caller_runtime_id: impl Into<String>,
         expires_at_unix_ms: u64,
         native_request_sha256: [u8; 32],
+        max_cost_micro_usd: u64,
         request: ProviderRequest,
     ) -> Result<Self, ContractError> {
         let provider_request_sha256 = provider_request_sha256(&request)?;
@@ -268,6 +274,7 @@ impl TransportInvocation {
             expires_at_unix_ms,
             native_request_sha256,
             provider_request_sha256,
+            max_cost_micro_usd,
             request,
         })
     }
@@ -280,6 +287,9 @@ impl TransportInvocation {
         if self.native_request_sha256 == [0; 32] {
             return Err(ContractError::Invalid("native_request_sha256"));
         }
+        if self.max_cost_micro_usd == 0 {
+            return Err(ContractError::Invalid("max_cost_micro_usd"));
+        }
         self.request.validate()?;
         if self.provider_request_sha256 != provider_request_sha256(&self.request)? {
             return Err(ContractError::ProviderDigest);
@@ -475,6 +485,15 @@ pub struct TransportReceipt {
     /// How many foreign `ReasoningState` items the service removed from the
     /// request before the backend saw it.
     pub foreign_reasoning_dropped: u32,
+    /// 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,
 }
 
@@ -524,6 +543,10 @@ pub enum TransportOutcome {
     Failed {
         kind: FailureKind,
         message: String,
+        /// What the provider billed for the failed request: `Some` when it
+        /// reported usage (`Some` of zeros when it billed nothing), `None`
+        /// when unknown.
+        usage: Option<Usage>,
     },
 }
 
@@ -535,6 +558,12 @@ pub enum Refusal {
     ModelNotAllowed,
     OutputBound,
     PayloadBound,
+    // The worst case of the request exceeds the invocation's own
+    // `max_cost_micro_usd`. Plain comments: a doc comment would split the
+    // published schema into a union.
+    CostBound,
+    // The caller's spend window cannot hold the request's reservation.
+    SpendCeiling,
     Capacity,
     InFlight,
     Indeterminate,
@@ -544,6 +573,12 @@ pub enum Refusal {
     // exchange, where dropping it would break the provider's own rules. A plain
     // comment: a doc comment would split the published schema into a union.
     ForeignReasoningState,
+    // The caller's newest admission window starts after the clock's reading,
+    // and the request does not fit what that window holds: the host's clock
+    // is behind the window and the caller stays refused until it catches up.
+    // A plain comment: a doc comment would split the published schema into a
+    // union.
+    ClockBehindAdmissionWindow,
 }
 
 /// The one wire codec of every v3 document: named MessagePack (struct fields
@@ -751,6 +786,31 @@ mod tests {
         );
     }
 
+    #[test]
+    fn max_cost_zero_is_malformed() {
+        let mut invocation = invocation();
+        invocation.max_cost_micro_usd = 0;
+        assert_eq!(
+            invocation.validate(1_000, 2_000),
+            Err(ContractError::Invalid("max_cost_micro_usd"))
+        );
+        invocation.max_cost_micro_usd = 1;
+        assert_eq!(invocation.validate(1_000, 2_000), Ok(()));
+    }
+
+    #[test]
+    fn max_cost_is_part_of_the_invocations_identity() {
+        let one = invocation();
+        let mut other = one.clone();
+        other.max_cost_micro_usd += 1;
+        assert_ne!(
+            canonical_invocation_bytes(&one).unwrap(),
+            canonical_invocation_bytes(&other).unwrap()
+        );
+        let decoded: TransportInvocation = wire_decode(&wire_encode(&one).unwrap()).unwrap();
+        assert_eq!(decoded.max_cost_micro_usd, one.max_cost_micro_usd);
+    }
+
     #[test]
     fn typed_tool_round_survives_messagepack() {
         let invocation = invocation();
@@ -946,6 +1006,7 @@ mod tests {
             sent.outcome = TransportOutcome::Failed {
                 kind,
                 message: "failed".to_string(),
+                usage: None,
             };
             assert_eq!(sent.validate_against(&invocation), Ok(()));
             let decoded: TransportReceipt = wire_decode(&wire_encode(&sent).unwrap()).unwrap();
@@ -955,6 +1016,7 @@ mod tests {
         blank.outcome = TransportOutcome::Failed {
             kind: FailureKind::Provider,
             message: String::new(),
+            usage: None,
         };
         assert_eq!(
             blank.validate_against(&invocation),
@@ -993,6 +1055,54 @@ mod tests {
         }
     }
 
+    #[test]
+    fn validate_against_refuses_each_binding_field_of_the_receipt() {
+        let invocation = invocation();
+        let good = receipt(&invocation);
+        assert_eq!(good.validate_against(&invocation), Ok(()));
+
+        let mut schema = good.clone();
+        schema.schema_id = "other".to_string();
+        assert_eq!(
+            schema.validate_against(&invocation),
+            Err(ContractError::Schema)
+        );
+        let mut provider_digest = good.clone();
+        provider_digest.provider_request_sha256 = [9; 32];
+        assert_eq!(
+            provider_digest.validate_against(&invocation),
+            Err(ContractError::ProviderDigest)
+        );
+        let mut native_digest = good.clone();
+        native_digest.native_request_sha256 = [9; 32];
+        assert_eq!(
+            native_digest.validate_against(&invocation),
+            Err(ContractError::NativeDigest)
+        );
+        let mut blank_provider = good.clone();
+        blank_provider.provider = " ".to_string();
+        assert_eq!(
+            blank_provider.validate_against(&invocation),
+            Err(ContractError::Invalid("receipt.provider"))
+        );
+        let mut blank_response = good.clone();
+        blank_response.outcome = TransportOutcome::Completed {
+            provider_response_id: Some(String::new()),
+            stop_reason: StopReason::EndTurn,
+            usage: Usage {
+                input_tokens: 0,
+                cache_read_tokens: 0,
+                cache_write_tokens: 0,
+                output_tokens: 0,
+                reasoning_tokens: None,
+            },
+        };
+        assert_eq!(
+            blank_response.validate_against(&invocation),
+            Err(ContractError::Invalid("receipt.provider_response_id"))
+        );
+    }
+
     fn reasoning(state: &[u8]) -> TransportEventPayload {
         TransportEventPayload::ReasoningState {
             state: state.to_vec(),
@@ -1067,6 +1177,7 @@ mod tests {
         failed.outcome = TransportOutcome::Failed {
             kind: FailureKind::RateLimited,
             message: "slow down".to_string(),
+            usage: None,
         };
         let partial = TransportResult::transported(
             vec![event(0, text("half")), event(1, reasoning(&[1, 2, 3]))],

crates/model-connector/src/envelope.rs

@@ -462,6 +462,8 @@ pub enum ServiceError {
     SharedCallerKey,
     #[error("connector replay store failed: {0}")]
     ReplayStore(String),
+    #[error("an allowed model has no price: {0}")]
+    UnpricedModel(String),
     #[error("connector replay record is malformed")]
     InvalidReplayRecord,
     #[error("caller key changed while durable replay records still exist")]

crates/model-connector/src/fixtures.rs

@@ -4,8 +4,15 @@ use crate::contract::*;
 #[cfg(feature = "daemon")]
 use crate::service::*;
 
+#[cfg(feature = "daemon")]
+use std::collections::BTreeMap;
 #[cfg(feature = "daemon")]
 use std::sync::atomic::{AtomicUsize, Ordering};
+#[cfg(feature = "daemon")]
+use std::sync::{Arc, Mutex};
+
+#[cfg(feature = "daemon")]
+use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, PushAllOptions};
 
 #[cfg(feature = "daemon")]
 static TEST_STORE_SEQUENCE: AtomicUsize = AtomicUsize::new(0);
@@ -37,6 +44,76 @@ impl Drop for TestStore {
     }
 }
 
+/// An in-memory ledger that accepts a fixed number of writes and then refuses
+/// every one, so a test can make the replay store fail where it chooses. A
+/// store reopened from it sees what was written, like a restart.
+#[cfg(feature = "daemon")]
+pub(crate) struct FlakyStore {
+    held: Arc<Mutex<BTreeMap<String, CultCacheEnvelope>>>,
+    pushes_allowed: Arc<AtomicUsize>,
+}
+
+#[cfg(feature = "daemon")]
+impl FlakyStore {
+    pub(crate) fn new(pushes_allowed: usize) -> Self {
+        Self {
+            held: Arc::default(),
+            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<AtomicUsize> {
+        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: Arc::new(AtomicUsize::new(pushes_allowed)),
+        }
+    }
+}
+
+#[cfg(feature = "daemon")]
+impl CacheBackingStore for FlakyStore {
+    fn pull_all(&self) -> anyhow::Result<Vec<CultCacheEnvelope>> {
+        Ok(self.held.lock().unwrap().values().cloned().collect())
+    }
+
+    fn push(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result<()> {
+        let allowed = self.pushes_allowed.load(Ordering::SeqCst);
+        if allowed == 0 {
+            return Err(anyhow::anyhow!("ledger write refused"));
+        }
+        self.pushes_allowed.store(allowed - 1, Ordering::SeqCst);
+        self.held
+            .lock()
+            .unwrap()
+            .insert(entry.key.clone(), entry.clone());
+        Ok(())
+    }
+
+    fn delete(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result<()> {
+        self.held.lock().unwrap().remove(&entry.key);
+        Ok(())
+    }
+
+    fn push_all(
+        &mut self,
+        entries: &[CultCacheEnvelope],
+        _options: PushAllOptions,
+    ) -> anyhow::Result<()> {
+        *self.held.lock().unwrap() = entries
+            .iter()
+            .map(|entry| (entry.key.clone(), entry.clone()))
+            .collect();
+        Ok(())
+    }
+}
+
 pub(crate) fn request() -> ProviderRequest {
     let mut request = ProviderRequest::new(
         "request-1",
@@ -56,8 +133,16 @@ pub(crate) fn request() -> ProviderRequest {
     request
 }
 
+/// Generous next to the fixture request's worst case (about 64,000), so it
+/// binds only in the tests that are about spend.
+pub(crate) const MAX_COST: u64 = 10_000_000;
+
+/// What the fixture usage (10 input and 3 output tokens) costs under
+/// [`fixture_price`]: 10 x 3,000,000 + 3 x 15,000,000 = 75,000,000 / 1e6.
+pub(crate) const FIXTURE_USAGE_COST: u64 = 75;
+
 pub(crate) fn invocation() -> TransportInvocation {
-    TransportInvocation::new("epiphany-yggdrasil", 2_000, [7; 32], request()).unwrap()
+    TransportInvocation::new("epiphany-yggdrasil", 2_000, [7; 32], MAX_COST, request()).unwrap()
 }
 
 pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt {
@@ -70,6 +155,8 @@ pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt {
         model: invocation.request.model.clone(),
         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,
@@ -83,6 +170,34 @@ pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt {
         },
     }
 }
+/// A price with a distinct nonzero rate per class, so a swapped rate changes a cost.
+#[cfg(feature = "daemon")]
+pub(crate) fn fixture_price() -> ModelPrice {
+    ModelPrice {
+        rates: Rates {
+            input: 3_000_000,
+            cache_write_short: 3_750_000,
+            cache_write_long: 6_000_000,
+            cache_read: 300_000,
+            output: 15_000_000,
+        },
+        long_context: None,
+        overhead_tokens: 100,
+    }
+}
+
+#[cfg(feature = "daemon")]
+pub(crate) fn fixture_prices() -> PriceTable {
+    PriceTable::from([("gpt-5.4".to_string(), fixture_price())])
+}
+
+/// Far above anything a fixture request reserves: the ceiling binds only in
+/// the tests that are about it.
+#[cfg(feature = "daemon")]
+pub(crate) const CEILING: u64 = 1 << 40;
+#[cfg(feature = "daemon")]
+pub(crate) const WINDOW_MS: u64 = 3_600_000;
+
 #[cfg(feature = "daemon")]
 pub(crate) fn caller(
     caller_runtime_id: &str,
@@ -97,6 +212,8 @@ pub(crate) fn caller(
         max_concurrent_requests,
         64 * 1024,
         32_768,
+        SpendCeiling(CEILING),
+        SpendWindowMs(WINDOW_MS),
     )
     .unwrap()
 }
@@ -106,6 +223,7 @@ pub(crate) fn open_service(store: &TestStore, max_concurrent_requests: usize) ->
     TransportService::open(
         &store.path,
         "fixture-sse",
+        fixture_prices(),
         [
             caller(
                 "epiphany-yggdrasil",
@@ -131,5 +249,5 @@ pub(crate) fn security(key: &str) -> TransportKey {
 pub(crate) fn invocation_for(caller_runtime_id: &str, request_id: &str) -> TransportInvocation {
     let mut request = request();
     request.request_id = request_id.to_string();
-    TransportInvocation::new(caller_runtime_id, 2_000, [7; 32], request).unwrap()
+    TransportInvocation::new(caller_runtime_id, 2_000, [7; 32], MAX_COST, request).unwrap()
 }

crates/model-connector/src/serve.rs

@@ -15,7 +15,10 @@ use crate::envelope::{
     ServiceError, TransportFrameError, decode_invocation_frame, encode_result_frame,
     frame_bytes_for_payload, read_transport_frame, write_transport_frame,
 };
-use crate::service::{CallerAdmission, ProviderCall, TransportAdmission, TransportService};
+use crate::service::{
+    CallerAdmission, PriceTable, ProviderCall, SpendCeiling, SpendWindowMs, TransportAdmission,
+    TransportService,
+};
 
 /// The only provider seam of the connector kit. A connector binary implements
 /// it over one model provider; the kit names no provider.
@@ -111,6 +114,12 @@ pub struct CallerConfig {
     pub max_concurrent_requests: usize,
     pub max_payload_bytes: usize,
     pub max_output_tokens: u32,
+    /// The most the caller may spend in one window, in micro-USD. Required
+    /// nonzero. A fuse set once, generously above what the dial allows.
+    pub spend_ceiling_micro_usd: u64,
+    /// The length of the caller's spend window, in milliseconds. Windows are
+    /// fixed and aligned to the unix epoch. Required nonzero.
+    pub spend_window_ms: u64,
 }
 
 pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>, ServeError> {
@@ -131,6 +140,8 @@ pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>,
                 caller.max_concurrent_requests,
                 caller.max_payload_bytes,
                 caller.max_output_tokens,
+                SpendCeiling(caller.spend_ceiling_micro_usd),
+                SpendWindowMs(caller.spend_window_ms),
             )
             .map_err(ServeError::Service)?,
         );
@@ -139,12 +150,15 @@ pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>,
 }
 
 /// Serves admitted callers on a loopback listener until the listener fails.
+/// `prices` is the connector binary's config document's table of what every
+/// allowed model costs (B4 owns that document).
 /// The caller binds; a listener that is not bound to loopback is refused, and
 /// peers that are not loopback are dropped.
 pub fn serve(
     listener: TcpListener,
     limits: ServeLimits,
     admissions: Vec<CallerAdmission>,
+    prices: PriceTable,
     replay_store: &Path,
     backend: Arc<dyn ProviderBackend>,
 ) -> Result<(), ServeError> {
@@ -163,6 +177,7 @@ pub fn serve(
     let service = Arc::new(Mutex::new(TransportService::open(
         replay_store,
         backend.provider(),
+        prices,
         admissions,
         limits.max_expiry_skew_ms,
     )?));
@@ -336,7 +351,10 @@ mod tests {
         FailureKind, Refusal, TransportDisposition, TransportEvent, TransportEventPayload,
         TransportInvocation, TransportOutcome, TransportResult,
     };
-    use crate::fixtures::{TestStore, caller, invocation_for, receipt, request};
+    use crate::fixtures::{
+        CEILING, MAX_COST, TestStore, WINDOW_MS, caller, fixture_prices, invocation_for, receipt,
+        request,
+    };
 
     const CALLER: &str = "epiphany-yggdrasil";
     const KEY: &str = "epiphany-distinct-test-key";
@@ -412,7 +430,14 @@ mod tests {
         let (sender, receiver) = std::sync::mpsc::channel();
         let path = path.to_path_buf();
         thread::spawn(move || {
-            let _ = sender.send(serve(listener, limits, admissions, &path, backend));
+            let _ = sender.send(serve(
+                listener,
+                limits,
+                admissions,
+                fixture_prices(),
+                &path,
+                backend,
+            ));
         });
         receiver.recv_timeout(Duration::from_secs(2)).ok()
     }
@@ -438,7 +463,16 @@ mod tests {
         let listener = TcpListener::bind("127.0.0.1:0").unwrap();
         let endpoint = listener.local_addr().unwrap();
         let path = store.path.clone();
-        thread::spawn(move || serve(listener, limits(), admissions, &path, backend));
+        thread::spawn(move || {
+            serve(
+                listener,
+                limits(),
+                admissions,
+                fixture_prices(),
+                &path,
+                backend,
+            )
+        });
         let client =
             ConnectorClient::new(endpoint, KEY, FRAME, Some(Duration::from_secs(5))).unwrap();
         (client, endpoint)
@@ -456,10 +490,37 @@ mod tests {
                 max_concurrent_requests: *capacity,
                 max_payload_bytes: 1_048_576,
                 max_output_tokens: 16_384,
+                spend_ceiling_micro_usd: CEILING,
+                spend_window_ms: WINDOW_MS,
             })
             .collect()
     }
 
+    #[test]
+    fn serve_refuses_a_request_over_its_max_cost_without_executing_it() {
+        let store = TestStore::new();
+        let backend = CountingBackend::new(true);
+        let (client, _) = start(backend.clone(), &store);
+        let mut invocation = fresh("request-costly");
+        invocation.max_cost_micro_usd = 1;
+        let refused = client.execute(&invocation).unwrap();
+        assert_eq!(
+            refused.disposition,
+            TransportDisposition::Refused(Refusal::CostBound)
+        );
+        assert_eq!(backend.executions.load(Ordering::SeqCst), 0);
+
+        // The refusal was not recorded: the same id with a budget that holds
+        // the worst case is admitted and executed.
+        invocation.max_cost_micro_usd = MAX_COST;
+        let admitted = client.execute(&invocation).unwrap();
+        assert!(matches!(
+            admitted.disposition,
+            TransportDisposition::Transported { .. }
+        ));
+        assert_eq!(backend.executions.load(Ordering::SeqCst), 1);
+    }
+
     #[test]
     fn serve_end_to_end_replays_without_a_second_execution() {
         let store = TestStore::new();
@@ -496,6 +557,7 @@ mod tests {
             CALLER,
             invocation.expires_at_unix_ms,
             [7; 32],
+            MAX_COST,
             conflicting_request,
         )
         .unwrap();
@@ -580,6 +642,7 @@ mod tests {
         let mut service = TransportService::open(
             &store.path,
             "fixture-sse",
+            fixture_prices(),
             vec![caller(CALLER, KEY, 4)],
             300_000,
         )
@@ -611,6 +674,7 @@ mod tests {
                 listener,
                 narrowed,
                 vec![caller(CALLER, KEY, 4)],
+                fixture_prices(),
                 &path,
                 served,
             )
@@ -693,7 +757,13 @@ mod tests {
     fn the_smallest_frame_also_carries_the_failure_that_stands_in_for_a_result() {
         let long = "x".repeat(crate::envelope::MAX_IDENTITY_BYTES);
         let plaintext = crate::contract::wire_encode(&crate::service::result_size_failure(
-            &long, &long, &long, [0xff; 32], [0xff; 32], 0,
+            &long,
+            &long,
+            &long,
+            [0xff; 32],
+            [0xff; 32],
+            0,
+            u64::MAX,
         ))
         .unwrap()
         .len();
@@ -781,7 +851,18 @@ mod tests {
     }
 
     fn admission(id: &str, key: &str, slots: usize, payload: usize) -> CallerAdmission {
-        CallerAdmission::new(id, key, 1, ["gpt-5.4".to_string()], slots, payload, 32_768).unwrap()
+        CallerAdmission::new(
+            id,
+            key,
+            1,
+            ["gpt-5.4".to_string()],
+            slots,
+            payload,
+            32_768,
+            SpendCeiling(CEILING),
+            SpendWindowMs(WINDOW_MS),
+        )
+        .unwrap()
     }
 
     /// `Some(field)` when serve refused the configuration, `None` while it serves.
@@ -887,7 +968,7 @@ mod tests {
             ..limits()
         };
         let admissions = vec![admission(CALLER, KEY, 4, bound)];
-        thread::spawn(move || serve(listener, tight, admissions, &path, served));
+        thread::spawn(move || serve(listener, tight, admissions, fixture_prices(), &path, served));
         let client =
             ConnectorClient::new(endpoint, KEY, frame, Some(Duration::from_secs(5))).unwrap();
         let reply = client.execute(&invocation).unwrap();
@@ -987,6 +1068,7 @@ mod tests {
                     && receipt.outcome == TransportOutcome::Failed {
                         kind: FailureKind::ResultSize,
                         message: "provider result exceeded its byte bound".to_string(),
+                        usage: None,
                     }
         )
     }
@@ -1001,6 +1083,7 @@ mod tests {
         let mut service = TransportService::open(
             &store.path,
             "fixture-sse",
+            fixture_prices(),
             vec![caller(CALLER, KEY, 4)],
             300_000,
         )
@@ -1103,6 +1186,7 @@ mod tests {
                 TransportService::open(
                     &store.path,
                     "fixture-sse",
+                    fixture_prices(),
                     vec![caller(CALLER, KEY, 4)],
                     300_000,
                 )
@@ -1160,7 +1244,16 @@ mod tests {
         let backend = CountingBackend::new(true);
         let path = store.path.clone();
         let served = backend.clone();
-        thread::spawn(move || serve(listener, limits(), admissions, &path, served));
+        thread::spawn(move || {
+            serve(
+                listener,
+                limits(),
+                admissions,
+                fixture_prices(),
+                &path,
+                served,
+            )
+        });
         let client =
             ConnectorClient::new(endpoint, KEY, FRAME, Some(Duration::from_secs(5))).unwrap();
         client.execute(&fresh("request-key")).unwrap();

crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame

@@ -1 +1 @@
-<EFBFBD>messageIdｮgolden-requestｩoperationｦinvokeｧpayloadﾚ$hqlzY2hlbWFfaWTZJGdhbWVjdWx0Lm1vZGVsLnRyYW5zcG9ydF9lbnZlbG9wZS52M7FjYWxsZXJfcnVudGltZV9pZK1nb2xkZW4tY2FsbGVyqnJlcXVlc3RfaWSuZ29sZGVuLXJlcXVlc3SsbWVzc2FnZV9raW5kqkludm9jYXRpb26lbm9uY2WcS2XMwjgRzMrM9QTM037MsczmqmNpcGhlcnRleHTFAy8eYmQGV5JCOLETwqQE+wIz/6yyPltHTPywbakMD/mgkJUxAvKWLNljXa4RAVVITAb3GxklnGF8zoOX/9qGSM9r0JZ126s+0LqvsRNhllG1S0pfG49fQJy99pevHReOSvX/cNWs2cHppYNdJUbVVAkDuioEXwtEs/vRCsmD/ab0bKUX0XRzrNbmjMU4+dHUCus8yt+iNTK9GdjXbEQAweeKNehirXEgnngLnwylo/MDoI4G237du3PmwhMzyIHztoIbwD0Bv9x7WqY+jj90vqEBsOHl+FmYc+F8+OiP6NKTDNRHypUtdo32G/IgfMfbsP/XHiNpPHtLxDqZVIcYdYqLAvhDfLBReLzQVLuNJnE0a2P8xLVAkfbmQZNsxRqS14RjYVtRoDcacxH1h0afo9OWMSK8Sp8eEBAhAJ7ovhCMklc7xWB3o4ny4cmDSExx5zWRLdI14rC1CZ7nq8k3V8fIIHctSp6am35A5yyiTwDH0m3h7/hvXrzhOqtAwHd0jkfNv+qOXZCEPdtbzOXOz0fMVZwvTAJWhoEYmhfKcGvXu8SmEtZcZe0WjF4xpFIFCFsSxh8HcrDFvxKDa58bnimo6DE2VeEMolpaBvnIGEst6qNZ9VQQnQIpCGbbtFd7jMrncnfcq+36NI10qW30r8sqAqgORp8cAGP5fy2X9SQDmH13SXgSfIdr7uxAHhK0X1r0iopJn4XtVfBiTd+IfoLFN+KO/JTZvLc48e6Jl9QRnxkL/UCgT/5wENd9VRxP+oO1CXG3DthI882rVbZDkhhwxupM9J/EURONajVyVSkeju+r9vzHCgR5odRW8cwqY1RXoq6L6mhj4ph01MBkUugz8MFwBsem+8QpnGatyVbZB99knWjK/01o5J6SZihcXNowRdon6BTydFByJaV5bRvgIK7IGRct3BmC8xidyxN9p7QAZYMPscxhQ0hAM615w76Cy82kx6eAzREBQX3COKApXSlPA/xlOKa/iUUMWHvFKyf/n0KrOlVvHjoDzPflHi3hpxxWcQGt8CxxNdSQMufeqce/5/BCeEwRIJ99+k7CEqIqpFIs7cAsI1GpF+NO5w==ｯpayloadEncodingｲmessagepack-base64ｭpayloadSchemaﾙ$gamecult.model.transport_envelope.v3ｭschemaVersionｼcultnet.operation_request.v0ｩserviceIdｸgamecult.model.connector
+<EFBFBD>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

crates/model-connector/tests/published_schemas.rs

@@ -110,6 +110,9 @@ fn the_refusal_schema_is_one_flat_string_enum() {
         .iter()
         .map(|value| value.as_str().unwrap())
         .collect();
-    assert_eq!(variants.len(), 12);
+    assert_eq!(variants.len(), 15);
+    assert!(variants.contains(&"ClockBehindAdmissionWindow"));
     assert!(variants.contains(&"ForeignReasoningState"));
+    assert!(variants.contains(&"CostBound"));
+    assert!(variants.contains(&"SpendCeiling"));
 }

crates/model-connector/tests/reasoning_state.rs

@@ -106,6 +106,8 @@ fn start_with(backend: Arc<dyn ProviderBackend>, max_concurrent: usize) -> Conne
         max_concurrent,
         64 * 1024,
         32_768,
+        SpendCeiling(1 << 40),
+        SpendWindowMs(3_600_000),
     )
     .unwrap();
     let limits = ServeLimits {
@@ -114,7 +116,21 @@ fn start_with(backend: Arc<dyn ProviderBackend>, max_concurrent: usize) -> Conne
         socket_timeout_ms: 5_000,
         max_expiry_skew_ms: 300_000,
     };
-    thread::spawn(move || serve(listener, limits, vec![admission], &store, backend));
+    let prices = PriceTable::from([(
+        "claude-opus-4-5".to_string(),
+        ModelPrice {
+            rates: Rates {
+                input: 5_000_000,
+                cache_write_short: 6_250_000,
+                cache_write_long: 10_000_000,
+                cache_read: 500_000,
+                output: 25_000_000,
+            },
+            long_context: None,
+            overhead_tokens: 100,
+        },
+    )]);
+    thread::spawn(move || serve(listener, limits, vec![admission], prices, &store, backend));
     ConnectorClient::new(endpoint, KEY, FRAME, Some(Duration::from_secs(5))).unwrap()
 }
 
@@ -127,7 +143,7 @@ fn invocation(request_id: &str, input: Vec<InputItem>) -> TransportInvocation {
         4_096,
     );
     request.input = input;
-    TransportInvocation::new(CALLER, now_ms() + 60_000, [7; 32], request).unwrap()
+    TransportInvocation::new(CALLER, now_ms() + 60_000, [7; 32], 10_000_000, request).unwrap()
 }
 
 fn user(text: &str) -> InputItem {
@@ -313,9 +329,10 @@ impl ProviderBackend for TamperingBackend {
 
 #[test]
 fn a_refused_completion_reaches_the_caller_as_a_typed_failure_and_frees_the_slot() {
-    let edits: [fn(&mut TransportReceipt); 2] = [
+    let edits: [fn(&mut TransportReceipt); 3] = [
         |receipt| receipt.provider = "another-provider".to_string(),
         |receipt| receipt.foreign_reasoning_dropped += 1,
+        |receipt| receipt.cost_micro_usd += 1,
     ];
     for edit in edits {
         // One concurrent request: a wedged slot would refuse the next id.

crates/model-connector/tests/wire_golden_v3.rs

@@ -81,7 +81,7 @@ fn request() -> ProviderRequest {
 }
 
 fn invocation() -> TransportInvocation {
-    TransportInvocation::new(CALLER, 1_800_000_000_000, [7; 32], request()).unwrap()
+    TransportInvocation::new(CALLER, 1_800_000_000_000, [7; 32], 5_000_000, request()).unwrap()
 }
 
 fn receipt(outcome: TransportOutcome) -> TransportReceipt {
@@ -95,6 +95,8 @@ fn receipt(outcome: TransportOutcome) -> TransportReceipt {
         model: invocation.request.model.clone(),
         provider: "anthropic-messages".to_string(),
         foreign_reasoning_dropped: 2,
+        cost_micro_usd: 1_234_567,
+        overrun_micro_usd: 89,
         outcome,
     }
 }
@@ -168,6 +170,13 @@ fn wire_golden_v3() {
     let failed = receipt(TransportOutcome::Failed {
         kind: FailureKind::RateLimited,
         message: "provider asked to slow down".to_string(),
+        usage: Some(Usage {
+            input_tokens: 120,
+            cache_read_tokens: 0,
+            cache_write_tokens: 0,
+            output_tokens: 0,
+            reasoning_tokens: None,
+        }),
     });
     assert_eq!(failed.validate_against(&invocation), Ok(()));
     assert_wire("transport_receipt_failed.mp", &failed);

schemas/transport_invocation.json

@@ -11,6 +11,12 @@
       "format": "uint64",
       "minimum": 0
     },
+    "max_cost_micro_usd": {
+      "description": "The most this request may cost, in micro-USD: the caller's own bound\non it. Required nonzero. It is part of the invocation's identity, so a\nchanged budget under a recorded request id is a `ReplayConflict`.",
+      "type": "integer",
+      "format": "uint64",
+      "minimum": 0
+    },
     "native_request_sha256": {
       "type": "array",
       "items": {
@@ -47,6 +53,7 @@
     "expires_at_unix_ms",
     "native_request_sha256",
     "provider_request_sha256",
+    "max_cost_micro_usd",
     "request"
   ],
   "$defs": {

schemas/transport_receipt.json

@@ -6,6 +6,12 @@
     "caller_runtime_id": {
       "type": "string"
     },
+    "cost_micro_usd": {
+      "description": "What the request cost, in micro-USD. The service derives it from the\noutcome's usage and its price table; a backend never writes it.",
+      "type": "integer",
+      "format": "uint64",
+      "minimum": 0
+    },
     "foreign_reasoning_dropped": {
       "description": "How many foreign `ReasoningState` items the service removed from the\nrequest before the backend saw it.",
       "type": "integer",
@@ -29,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"
     },
@@ -60,6 +72,8 @@
     "model",
     "provider",
     "foreign_reasoning_dropped",
+    "cost_micro_usd",
+    "overrun_micro_usd",
     "outcome"
   ],
   "$defs": {
@@ -132,6 +146,17 @@
                 },
                 "message": {
                   "type": "string"
+                },
+                "usage": {
+                  "description": "What the provider billed for the failed request: `Some` when it\nreported usage (`Some` of zeros when it billed nothing), `None`\nwhen unknown.",
+                  "anyOf": [
+                    {
+                      "$ref": "#/$defs/Usage"
+                    },
+                    {
+                      "type": "null"
+                    }
+                  ]
                 }
               },
               "additionalProperties": false,
@@ -149,7 +174,7 @@
       ]
     },
     "Usage": {
-      "description": "Token accounting for one completed request. Cache reads and writes are\ncounted apart from plain input; `reasoning_tokens` is absent when the\nprovider does not report it.",
+      "description": "Token accounting for one request, as the backend reports it; the service\nprices it. Cache reads and writes are counted apart from plain input.\n`output_tokens` includes reasoning tokens: `reasoning_tokens` is a detail\n(absent when the provider does not report it) and is never priced apart.",
       "type": "object",
       "properties": {
         "cache_read_tokens": {

schemas/transport_result.json

@@ -46,12 +46,15 @@
         "ModelNotAllowed",
         "OutputBound",
         "PayloadBound",
+        "CostBound",
+        "SpendCeiling",
         "Capacity",
         "InFlight",
         "Indeterminate",
         "ReplayConflict",
         "Malformed",
-        "ForeignReasoningState"
+        "ForeignReasoningState",
+        "ClockBehindAdmissionWindow"
       ]
     },
     "StopReason": {
@@ -250,6 +253,17 @@
                 },
                 "message": {
                   "type": "string"
+                },
+                "usage": {
+                  "description": "What the provider billed for the failed request: `Some` when it\nreported usage (`Some` of zeros when it billed nothing), `None`\nwhen unknown.",
+                  "anyOf": [
+                    {
+                      "$ref": "#/$defs/Usage"
+                    },
+                    {
+                      "type": "null"
+                    }
+                  ]
                 }
               },
               "additionalProperties": false,
@@ -272,6 +286,12 @@
         "caller_runtime_id": {
           "type": "string"
         },
+        "cost_micro_usd": {
+          "description": "What the request cost, in micro-USD. The service derives it from the\noutcome's usage and its price table; a backend never writes it.",
+          "type": "integer",
+          "format": "uint64",
+          "minimum": 0
+        },
         "foreign_reasoning_dropped": {
           "description": "How many foreign `ReasoningState` items the service removed from the\nrequest before the backend saw it.",
           "type": "integer",
@@ -295,6 +315,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"
         },
@@ -326,11 +352,13 @@
         "model",
         "provider",
         "foreign_reasoning_dropped",
+        "cost_micro_usd",
+        "overrun_micro_usd",
         "outcome"
       ]
     },
     "Usage": {
-      "description": "Token accounting for one completed request. Cache reads and writes are\ncounted apart from plain input; `reasoning_tokens` is absent when the\nprovider does not report it.",
+      "description": "Token accounting for one request, as the backend reports it; the service\nprices it. Cache reads and writes are counted apart from plain input.\n`output_tokens` includes reasoning tokens: `reasoning_tokens` is a detail\n(absent when the provider does not report it) and is never priced apart.",
       "type": "object",
       "properties": {
         "cache_read_tokens": {