connector-spend (B3c): reserve-then-settle spend in the replay record, cost derived on the receipt

The replay store is the connector's one durable ledger. begin reserves the
request's worst case in the Active record it claims the request with, in one
write: a ledger that cannot be written means nothing runs. complete settles
the reservation to the charge derived from the outcome's usage under the
service's price table; a backend reports usage only, never cost, and a receipt
carrying any other cost is answered with the Provider stand-in at the whole
reservation. Amounts are integer micro-USD, prices micro-USD per million
tokens, rounded up once per request.

Refusals: CostBound (the worst case exceeds the invocation's max_cost_micro_usd,
decided before the replay lookup) and SpendCeiling (the caller's window cannot
hold the reservation, after the replay lookup). Each caller's config carries a
spend ceiling and window length, both required nonzero; windows are fixed and
aligned to the unix epoch, and a request belongs to the window it was admitted
in. The per-window spend is re-derived from the records at open.

gamecult.model.replay_record.v1 is replaced by v2 (no reader survives). The
published invocation, receipt and result schemas and their wire goldens are
regenerated; provider_request and input_reasoning_state stay byte-identical.

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,9 @@ 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,
     pub outcome: TransportOutcome,
 }
 
@@ -524,6 +537,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 +552,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,
@@ -751,6 +774,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 +994,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 +1004,7 @@ mod tests {
         blank.outcome = TransportOutcome::Failed {
             kind: FailureKind::Provider,
             message: String::new(),
+            usage: None,
         };
         assert_eq!(
             blank.validate_against(&invocation),
@@ -993,6 +1043,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 +1165,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,70 @@ 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: AtomicUsize,
+}
+
+#[cfg(feature = "daemon")]
+impl FlakyStore {
+    pub(crate) fn new(pushes_allowed: usize) -> Self {
+        Self {
+            held: Arc::default(),
+            pushes_allowed: AtomicUsize::new(pushes_allowed),
+        }
+    }
+
+    /// 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),
+        }
+    }
+}
+
+#[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 +127,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 +149,7 @@ 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,
         outcome: TransportOutcome::Completed {
             provider_response_id: Some("response-1".to_string()),
             stop_reason: StopReason::EndTurn,
@@ -83,6 +163,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 +205,8 @@ pub(crate) fn caller(
         max_concurrent_requests,
         64 * 1024,
         32_768,
+        CEILING,
+        WINDOW_MS,
     )
     .unwrap()
 }
@@ -106,6 +216,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 +242,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,9 @@ 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, 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 +113,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 +139,8 @@ pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>,
                 caller.max_concurrent_requests,
                 caller.max_payload_bytes,
                 caller.max_output_tokens,
+                caller.spend_ceiling_micro_usd,
+                caller.spend_window_ms,
             )
             .map_err(ServeError::Service)?,
         );
@@ -139,12 +149,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 +176,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 +350,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 +429,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 +462,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 +489,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 +556,7 @@ mod tests {
             CALLER,
             invocation.expires_at_unix_ms,
             [7; 32],
+            MAX_COST,
             conflicting_request,
         )
         .unwrap();
@@ -580,6 +641,7 @@ mod tests {
         let mut service = TransportService::open(
             &store.path,
             "fixture-sse",
+            fixture_prices(),
             vec![caller(CALLER, KEY, 4)],
             300_000,
         )
@@ -611,6 +673,7 @@ mod tests {
                 listener,
                 narrowed,
                 vec![caller(CALLER, KEY, 4)],
+                fixture_prices(),
                 &path,
                 served,
             )
@@ -693,7 +756,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 +850,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,
+            CEILING,
+            WINDOW_MS,
+        )
+        .unwrap()
     }
 
     /// `Some(field)` when serve refused the configuration, `None` while it serves.
@@ -887,7 +967,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 +1067,7 @@ mod tests {
                     && receipt.outcome == TransportOutcome::Failed {
                         kind: FailureKind::ResultSize,
                         message: "provider result exceeded its byte bound".to_string(),
+                        usage: None,
                     }
         )
     }
@@ -1001,6 +1082,7 @@ mod tests {
         let mut service = TransportService::open(
             &store.path,
             "fixture-sse",
+            fixture_prices(),
             vec![caller(CALLER, KEY, 4)],
             300_000,
         )
@@ -1103,6 +1185,7 @@ mod tests {
                 TransportService::open(
                     &store.path,
                     "fixture-sse",
+                    fixture_prices(),
                     vec![caller(CALLER, KEY, 4)],
                     300_000,
                 )
@@ -1160,7 +1243,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ﾚ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

crates/model-connector/tests/published_schemas.rs

@@ -110,6 +110,8 @@ 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(), 14);
     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,
+        1 << 40,
+        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,7 @@ 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,
         outcome,
     }
 }
@@ -168,6 +169,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",
@@ -60,6 +66,7 @@
     "model",
     "provider",
     "foreign_reasoning_dropped",
+    "cost_micro_usd",
     "outcome"
   ],
   "$defs": {
@@ -132,6 +139,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 +167,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,6 +46,8 @@
         "ModelNotAllowed",
         "OutputBound",
         "PayloadBound",
+        "CostBound",
+        "SpendCeiling",
         "Capacity",
         "InFlight",
         "Indeterminate",
@@ -250,6 +252,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 +285,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",
@@ -326,11 +345,12 @@
         "model",
         "provider",
         "foreign_reasoning_dropped",
+        "cost_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": {