GameCult/ModelConnector
Watch
9
0
Fork
You've already forked ModelConnector
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

connector-usage-only: delete the dollar authorities (WIP, does not yet build)

Browse source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 04:43:18 +01:00
commit 06a47bb13f
39 changed files with 1243 additions and 6010 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.gitattributes vendored
View file

@ -1,2 +1,2 @@
schemas/*.json text eol=lf schemas/*.json text eol=lf
crates/model-connector/tests/fixtures/wire_v3/* -text crates/model-connector/tests/fixtures/wire_v4/* -text

8
README.md
View file

@ -21,11 +21,11 @@ Crate `crates/model-connector`:
- `src/envelope.rs` - the AES-256-GCM `TransportEnvelope` (binary ciphertext), its - `src/envelope.rs` - the AES-256-GCM `TransportEnvelope` (binary ciphertext), its
identity-binding AAD, the length-prefixed framing, and the CultNet carriage: each frame is a identity-binding AAD, the length-prefixed framing, and the CultNet carriage: each frame is a
`cultnet.operation_request.v0` / `operation_response.v0` for service `cultnet.operation_request.v0` / `operation_response.v0` for service
`gamecult.model.connector`, operation `invoke` (or `spend_readout`), its `message_id` the `gamecult.model.connector`, operation `invoke` (or `usage_readout`), its `message_id` the
request id and its payload the base64 of the sealed envelope. `spend_readout` answers a request id and its payload the base64 of the sealed envelope. `usage_readout` answers a
`SpendReadout` (`gamecult.model.spend_readout.v1`): each caller's spent and reserved `UsageReadout` (`gamecult.model.usage_readout.v1`): each caller's spent and reserved
micro-USD against its ceiling in its current window, and the backend's plan quota; micro-USD against its ceiling in its current window, and the backend's plan quota;
`ConnectorClient::spend_readout` asks. `ConnectorClient::usage_readout` asks.
- `schemas/` - one JSON Schema per v3 document, generated from the types. - `schemas/` - one JSON Schema per v3 document, generated from the types.
- `src/client.rs` - `ConnectorClient`, the one-request caller side. - `src/client.rs` - `ConnectorClient`, the one-request caller side.
- `src/service.rs` (feature `daemon`) - `TransportService`: caller admission, durable - `src/service.rs` (feature `daemon`) - `TransportService`: caller admission, durable

3
crates/claude-connector/Cargo.toml
View file

@ -25,3 +25,6 @@ zeroize = "1"
[target.'cfg(unix)'.dependencies] [target.'cfg(unix)'.dependencies]
libc = "0.2" libc = "0.2"
[dev-dependencies]
rmpv = "1"

81
crates/claude-connector/src/backend/tests.rs
View file

@ -5,8 +5,7 @@ use std::sync::{Arc, Mutex};
use std::time::Duration; use std::time::Duration;
use model_connector::{ use model_connector::{
CallerAdmission, ModelPrice, PriceTable, Rates, ReasoningEffort, SpendCeiling, SpendWindowMs, CallerAdmission, PlanQuotaSnapshot, ReasoningEffort, StopReason, ToolDefinition, TransportAdmission, TransportDisposition, TransportInvocation,
StopReason, ToolDefinition, TransportAdmission, TransportDisposition, TransportInvocation,
TransportKey, TransportReceipt, TransportService, decrypt_result, encrypt_invocation, TransportKey, TransportReceipt, TransportService, decrypt_result, encrypt_invocation,
}; };
@ -460,25 +459,6 @@ struct Door {
service: TransportService, service: TransportService,
} }
fn prices() -> PriceTable {
PriceTable::from([MODEL, OTHER_MODEL].map(|model| {
(
model.to_string(),
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,
},
)
}))
}
impl Door { impl Door {
fn new() -> Self { fn new() -> Self {
let dir = std::env::temp_dir().join(format!( let dir = std::env::temp_dir().join(format!(
@ -496,13 +476,11 @@ impl Door {
4, 4,
FRAME / 2, FRAME / 2,
32_768, 32_768,
SpendCeiling(1 << 40),
SpendWindowMs(3_600_000),
) )
.unwrap() .unwrap()
}); });
let service = let service =
TransportService::open(&dir.join("replay.cc"), PROVIDER, prices(), callers, 5_000) TransportService::open(&dir.join("replay.cc"), PROVIDER, callers, 5_000)
.unwrap(); .unwrap();
Self { dir, service } Self { dir, service }
} }
@ -521,9 +499,12 @@ impl Door {
) -> (model_connector::ExecutionClaim, TransportInvocation) { ) -> (model_connector::ExecutionClaim, TransportInvocation) {
let key = TransportKey::from_connection_secret(&format!("{caller}-key")).unwrap(); let key = TransportKey::from_connection_secret(&format!("{caller}-key")).unwrap();
let invocation = let invocation =
TransportInvocation::new(caller, 2_000, [7; 32], 10_000_000, request).unwrap(); TransportInvocation::new(caller, 2_000, [7; 32], request).unwrap();
let envelope = encrypt_invocation(&invocation, &key).unwrap(); let envelope = encrypt_invocation(&invocation, &key).unwrap();
match self.service.begin(&envelope, 1_000, FRAME).unwrap() { match self
.service
.begin(&envelope, 1_000, &PlanQuotaSnapshot::default(), FRAME)
.unwrap() {
TransportAdmission::Execute(claim) => (*claim, invocation), TransportAdmission::Execute(claim) => (*claim, invocation),
TransportAdmission::Reply(_) => panic!("the door answered instead of admitting"), TransportAdmission::Reply(_) => panic!("the door answered instead of admitting"),
} }
@ -1458,11 +1439,6 @@ fn usage_sums_model_usage_across_models() {
}; };
assert_eq!(*summed, usage(12, 37, 1083, 24, Some(8))); assert_eq!(*summed, usage(12, 37, 1083, 24, Some(8)));
// The receipt is priced from usage alone: 12*3 + 37*0.3 + 1083*6 + 24*15 micro-USD, rounded up. // The receipt is priced from usage alone: 12*3 + 37*0.3 + 1083*6 + 24*15 micro-USD, rounded up.
assert_eq!(parts(&cheap).1.cost_micro_usd, 36 + 12 + 6498 + 360);
assert_eq!(
parts(&cheap).1.cost_micro_usd,
parts(&dear).1.cost_micro_usd
);
// No model reported thinking tokens: none is claimed. // No model reported thinking tokens: none is claimed.
let plain = r#"{"claude-haiku-5-5":{"inputTokens":1,"outputTokens":1,"cacheReadInputTokens":0,"cacheCreationInputTokens":0}}"#; let plain = r#"{"claude-haiku-5-5":{"inputTokens":1,"outputTokens":1,"cacheReadInputTokens":0,"cacheCreationInputTokens":0}}"#;
@ -1599,7 +1575,6 @@ fn no_result_event_is_a_failure_with_unknown_usage() {
}; };
assert_eq!(zero_billed, Some(usage_zero())); assert_eq!(zero_billed, Some(usage_zero()));
let mut costs = Vec::new();
for (script, expected) in ends { for (script, expected) in ends {
let rig = rig(vec![script]); let rig = rig(vec![script]);
let result = rig.run(request(vec![user("hi")])); let result = rig.run(request(vec![user("hi")]));
@ -1607,20 +1582,7 @@ fn no_result_event_is_a_failure_with_unknown_usage() {
assert_eq!(kind, expected); assert_eq!(kind, expected);
assert_eq!(usage, None, "{message}"); assert_eq!(usage, None, "{message}");
assert!(result.transcript_items().is_empty()); assert!(result.transcript_items().is_empty());
costs.push(parts(&result).1.cost_micro_usd);
} }
// Unknown usage is charged the reservation, which is far above a typical turn.
assert!(costs.iter().all(|cost| *cost >= 15_000), "{costs:?}");
assert_eq!(
parts(&rig(vec![]).run({
let mut bad = request(vec![user("a")]);
bad.tools.push(a_tool());
bad
}))
.1
.cost_micro_usd,
0
);
} }
#[test] #[test]
@ -1719,7 +1681,6 @@ fn a_port_that_cannot_spawn_is_a_provider_failure() {
Some(usage_zero()), Some(usage_zero()),
"a child that never ran bills nothing" "a child that never ran bills nothing"
); );
assert_eq!(parts(&result).1.cost_micro_usd, 0);
assert!( assert!(
!message.contains(dir.path().to_str().unwrap()), !message.contains(dir.path().to_str().unwrap()),
"no path in a message" "no path in a message"
@ -1743,8 +1704,6 @@ fn backend_end_to_end_through_the_service() {
assert_eq!(receipt.provider, "claude-code"); assert_eq!(receipt.provider, "claude-code");
assert_eq!(receipt.model, MODEL); assert_eq!(receipt.model, MODEL);
// USAGE priced at the table: 2*3 + 7*0.3 + 1043*6 + 4*15 micro-USD, rounded up. // USAGE priced at the table: 2*3 + 7*0.3 + 1043*6 + 4*15 micro-USD, rounded up.
assert_eq!(receipt.cost_micro_usd, 6 + 3 + 6258 + 60);
assert_eq!(receipt.overrun_micro_usd, 0);
assert_eq!(events.len(), 2); assert_eq!(events.len(), 2);
let items = opened.transcript_items(); let items = opened.transcript_items();
assert_eq!(items[0], assistant("hello back")); assert_eq!(items[0], assistant("hello back"));
@ -1757,13 +1716,6 @@ fn backend_end_to_end_through_the_service() {
const USAGE_TURN_2: &str = r#"{"claude-haiku-5-5":{"inputTokens":3,"outputTokens":5,"cacheReadInputTokens":1100,"cacheCreationInputTokens":9}}"#; const USAGE_TURN_2: &str = r#"{"claude-haiku-5-5":{"inputTokens":3,"outputTokens":5,"cacheReadInputTokens":1100,"cacheCreationInputTokens":9}}"#;
/// What a call whose usage is unknown is billed: the kit's reservation.
fn reservation() -> u64 {
let result = rig(vec![Script::lines(vec![])]).run(request(vec![user("hi")]));
assert_eq!(failure_of(&result).2, None);
parts(&result).1.cost_micro_usd
}
#[test] #[test]
fn a_resumed_turn_is_charged_only_its_own_usage() { fn a_resumed_turn_is_charged_only_its_own_usage() {
let rig = rig(vec![ let rig = rig(vec![
@ -1788,13 +1740,10 @@ fn a_resumed_turn_is_charged_only_its_own_usage() {
}; };
assert_eq!(*billed, usage(3, 1100, 9, 5, None)); assert_eq!(*billed, usage(3, 1100, 9, 5, None));
// 3*3 + 1100*0.3 + 9*6 + 5*15 micro-USD. // 3*3 + 1100*0.3 + 9*6 + 5*15 micro-USD.
assert_eq!(parts(&second).1.cost_micro_usd, 9 + 330 + 54 + 75);
} }
#[test] #[test]
fn usage_the_result_does_not_report_is_billed_the_reservation() { fn usage_the_result_does_not_report_stays_unknown() {
let reservation = reservation();
assert!(reservation >= 15_000, "{reservation}");
let success = |fields: &str| { let success = |fields: &str| {
format!( format!(
r#"{{"type":"result","subtype":"success","is_error":false,"stop_reason":"end_turn","modelUsage":{CUMULATIVE}{fields}}}"# r#"{{"type":"result","subtype":"success","is_error":false,"stop_reason":"end_turn","modelUsage":{CUMULATIVE}{fields}}}"#
@ -1838,7 +1787,6 @@ fn usage_the_result_does_not_report_is_billed_the_reservation() {
let (kind, _, usage) = failure_of(&result); let (kind, _, usage) = failure_of(&result);
assert_eq!(kind, expected, "{line}"); assert_eq!(kind, expected, "{line}");
assert_eq!(usage, None, "{line}"); assert_eq!(usage, None, "{line}");
assert_eq!(parts(&result).1.cost_micro_usd, reservation, "{line}");
} }
for line in cases { for line in cases {
let result = rig(vec![Script::lines(vec![ let result = rig(vec![Script::lines(vec![
@ -1850,7 +1798,6 @@ fn usage_the_result_does_not_report_is_billed_the_reservation() {
let (kind, _, usage) = failure_of(&result); let (kind, _, usage) = failure_of(&result);
assert_eq!(kind, FailureKind::Provider, "{line}"); assert_eq!(kind, FailureKind::Provider, "{line}");
assert_eq!(usage, None, "{line}"); assert_eq!(usage, None, "{line}");
assert_eq!(parts(&result).1.cost_micro_usd, reservation, "{line}");
assert!(result.transcript_items().is_empty(), "{line}"); assert!(result.transcript_items().is_empty(), "{line}");
} }
} }
@ -1865,7 +1812,6 @@ fn door_set_is_exactly_401_403_429() {
// with a bogus token. // with a bogus token.
let retry = r#"{"type":"system","subtype":"api_retry","error":"authentication_failed"}"#; let retry = r#"{"type":"system","subtype":"api_retry","error":"authentication_failed"}"#;
let synthetic = r#"{"type":"assistant","api_error_status":401,"error":"authentication_failed","message":{"id":"e48001b7-cf4f-49b1-9810-db8fd7f73c73","model":"<synthetic>","stop_reason":"stop_sequence","content":[{"type":"text","text":"Failed to authenticate. API Error: 401 OAuth access token is invalid."}]}}"#; let synthetic = r#"{"type":"assistant","api_error_status":401,"error":"authentication_failed","message":{"id":"e48001b7-cf4f-49b1-9810-db8fd7f73c73","model":"<synthetic>","stop_reason":"stop_sequence","content":[{"type":"text","text":"Failed to authenticate. API Error: 401 OAuth access token is invalid."}]}}"#;
let reservation = reservation();
let statuses: [Option<u16>; 9] = [ let statuses: [Option<u16>; 9] = [
Some(400), Some(400),
Some(401), Some(401),
@ -1904,10 +1850,8 @@ fn door_set_is_exactly_401_403_429() {
let case = format!("{status:?} {shape}"); let case = format!("{status:?} {shape}");
if matches!(status, Some(401 | 403 | 429)) && shape == "empty" { if matches!(status, Some(401 | 403 | 429)) && shape == "empty" {
assert_eq!(usage, Some(usage_zero()), "{case}"); assert_eq!(usage, Some(usage_zero()), "{case}");
assert_eq!(parts(&result).1.cost_micro_usd, 0, "{case}");
} else { } else {
assert_eq!(usage, None, "{case}"); assert_eq!(usage, None, "{case}");
assert_eq!(parts(&result).1.cost_micro_usd, reservation, "{case}");
} }
} }
} }
@ -1979,7 +1923,6 @@ fn prompt_file_lives_in_prompt_dir_exclusively() {
let result = rig.run(request(vec![user("hi")])); let result = rig.run(request(vec![user("hi")]));
let (kind, _, usage) = failure_of(&result); let (kind, _, usage) = failure_of(&result);
assert_eq!((kind, usage), (FailureKind::Provider, Some(usage_zero()))); assert_eq!((kind, usage), (FailureKind::Provider, Some(usage_zero())));
assert_eq!(parts(&result).1.cost_micro_usd, 0);
assert_eq!(rig.spawns(), 0); assert_eq!(rig.spawns(), 0);
} }
@ -1996,11 +1939,6 @@ fn pin_is_checked_before_every_spawn() {
let result = rig.run(request(vec![user("hi")])); let result = rig.run(request(vec![user("hi")]));
let (kind, message, usage) = failure_of(&result); let (kind, message, usage) = failure_of(&result);
assert_eq!((kind, usage), (FailureKind::Provider, Some(usage_zero()))); assert_eq!((kind, usage), (FailureKind::Provider, Some(usage_zero())));
assert_eq!(
parts(&result).1.cost_micro_usd,
0,
"billed nothing, not the reservation"
);
assert!(!message.contains(rig.dir.path().to_str().unwrap())); assert!(!message.contains(rig.dir.path().to_str().unwrap()));
assert_eq!(rig.spawns(), 1, "no child for a binary off its pin"); assert_eq!(rig.spawns(), 1, "no child for a binary off its pin");
assert!(!rig.backend.binary_matches_pin()); assert!(!rig.backend.binary_matches_pin());
@ -2695,7 +2633,6 @@ fn not_logged_in_line(status: &str, api_ms: u64, extra: &str) -> String {
#[test] #[test]
fn a_result_that_made_no_api_call_bills_nothing() { fn a_result_that_made_no_api_call_bills_nothing() {
let reservation = reservation();
let empty = r#","model_usage_since_last_result":{}"#; let empty = r#","model_usage_since_last_result":{}"#;
let cases = [ let cases = [
// (status field, api ms, usage field, billed nothing) // (status field, api ms, usage field, billed nothing)
@ -2723,10 +2660,8 @@ fn a_result_that_made_no_api_call_bills_nothing() {
let (_, _, usage) = failure_of(&result); let (_, _, usage) = failure_of(&result);
if free { if free {
assert_eq!(usage, Some(usage_zero()), "{line}"); assert_eq!(usage, Some(usage_zero()), "{line}");
assert_eq!(parts(&result).1.cost_micro_usd, 0, "{line}");
} else { } else {
assert_eq!(usage, None, "{line}"); assert_eq!(usage, None, "{line}");
assert_eq!(parts(&result).1.cost_micro_usd, reservation, "{line}");
} }
} }
} }

66
crates/claude-connector/src/daemon.rs
View file

@ -3,7 +3,7 @@
//! run with; nothing in the connector's environment, flags or JSON can change //! run with; nothing in the connector's environment, flags or JSON can change
//! them. Validation errors name a field and never echo a value. //! them. Validation errors name a field and never echo a value.
use std::collections::{BTreeMap, BTreeSet}; use std::collections::BTreeSet;
use std::net::{SocketAddr, TcpListener}; use std::net::{SocketAddr, TcpListener};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
@ -11,7 +11,7 @@ use std::time::Duration;
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use model_connector::{ use model_connector::{
CallerAdmission, CallerConfig, LeaseTurn, MIN_PAYLOAD_BYTES, ModelPrice, ProviderBackend, CallerAdmission, CallerConfig, LeaseTurn, MIN_PAYLOAD_BYTES, ProviderBackend,
ServeError, ServeLimits, ServiceError, TransportService, load_admissions, serve_service, ServeError, ServeLimits, ServiceError, TransportService, load_admissions, serve_service,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -31,8 +31,8 @@ const MODEL_PREFIX: &str = "claude-";
/// A unix account that may lease the subscription token for a Self run or a /// A unix account that may lease the subscription token for a Self run or a
/// Persona turn, the unit its process must run in (`mind-self@<unit_instance>` /// Persona turn, the unit its process must run in (`mind-self@<unit_instance>`
/// or `mind-persona@<unit_instance>`, by `turn`), the configured caller whose /// or `mind-persona@<unit_instance>`, by `turn`), and the configured caller
/// spend window its leases reserve in, and what each lease reserves there. /// whose name its leases carry.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeaseCallerConfig { pub struct LeaseCallerConfig {
pub unix_user: String, pub unix_user: String,
@ -41,11 +41,10 @@ pub struct LeaseCallerConfig {
/// or `mind-persona@`, its process must run in. /// or `mind-persona@`, its process must run in.
pub turn: LeaseTurn, pub turn: LeaseTurn,
pub unit_instance: String, pub unit_instance: String,
pub reservation_micro_usd: u64,
} }
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "gamecult.model_connector.claude_connector_config.v1")] #[cultcache(type = "gamecult.model_connector.claude_connector_config.v2")]
pub struct ClaudeConnectorConfig { pub struct ClaudeConnectorConfig {
/// A loopback socket address. /// A loopback socket address.
#[cultcache(key = 0)] #[cultcache(key = 0)]
@ -86,14 +85,10 @@ pub struct ClaudeConnectorConfig {
#[cultcache(key = 16)] #[cultcache(key = 16)]
pub max_expiry_skew_ms: u64, pub max_expiry_skew_ms: u64,
/// Every caller of this connector: its own connection key and epoch, models, /// Every caller of this connector: its own connection key and epoch, models,
/// concurrency, payload and output bounds, and spend ceiling and window. /// concurrency, payload and output bounds.
/// Ids are unique. Handles are bound to the caller that was given them. /// Ids are unique. Handles are bound to the caller that was given them.
#[cultcache(key = 17)] #[cultcache(key = 17)]
pub callers: Vec<CallerConfig>, pub callers: Vec<CallerConfig>,
/// What each allowed model costs. The keys are the known list of models:
/// ids exactly as the CLI reports them in its init event.
#[cultcache(key = 18)]
pub prices: BTreeMap<String, ModelPrice>,
/// Where the Self-run lease socket is bound (a systemd RuntimeDirectory), /// Where the Self-run lease socket is bound (a systemd RuntimeDirectory),
/// or none when this connector lends no token. /// or none when this connector lends no token.
#[cultcache(key = 19)] #[cultcache(key = 19)]
@ -143,7 +138,6 @@ impl From<ServeError> for ConfigError {
ServeError::ConnectionKey(_) => Self::Unusable("connection_key_file"), ServeError::ConnectionKey(_) => Self::Unusable("connection_key_file"),
ServeError::Service(error) => match error { ServeError::Service(error) => match error {
ServiceError::ReplayStore(_) => Self::Unusable("replay_store"), ServiceError::ReplayStore(_) => Self::Unusable("replay_store"),
ServiceError::UnpricedModel(_) => Self::Invalid("prices"),
ServiceError::InvalidAdmission ServiceError::InvalidAdmission
| ServiceError::DuplicateCaller | ServiceError::DuplicateCaller
| ServiceError::SharedCallerKey => Self::Invalid("callers"), | ServiceError::SharedCallerKey => Self::Invalid("callers"),
@ -228,8 +222,6 @@ impl ClaudeConnectorConfig {
if caller.max_concurrent_requests == 0 if caller.max_concurrent_requests == 0
|| caller.max_payload_bytes < MIN_PAYLOAD_BYTES || caller.max_payload_bytes < MIN_PAYLOAD_BYTES
|| caller.max_output_tokens == 0 || caller.max_output_tokens == 0
|| caller.spend_ceiling_micro_usd == 0
|| caller.spend_window_ms == 0
{ {
return Err(ConfigError::Invalid("caller bounds")); return Err(ConfigError::Invalid("caller bounds"));
} }
@ -240,19 +232,11 @@ impl ClaudeConnectorConfig {
for model in &caller.allowed_models { for model in &caller.allowed_models {
if !has_model_id_form(model) if !has_model_id_form(model)
|| !seen.insert(model.as_str()) || !seen.insert(model.as_str())
|| !self.prices.contains_key(model)
{ {
return Err(ConfigError::Invalid("allowed_models")); return Err(ConfigError::Invalid("allowed_models"));
} }
} }
} }
if !self
.prices
.iter()
.all(|(model, price)| has_model_id_form(model) && is_fully_priced(price))
{
return Err(ConfigError::Invalid("prices"));
}
self.validate_leases()?; self.validate_leases()?;
Ok(listen) Ok(listen)
} }
@ -278,11 +262,7 @@ impl ClaudeConnectorConfig {
&& self && self
.callers .callers
.iter() .iter()
.find(|caller| caller.caller_runtime_id == lease.caller_id) .any(|caller| caller.caller_runtime_id == lease.caller_id)
.is_some_and(|caller| {
(1..=caller.spend_ceiling_micro_usd)
.contains(&lease.reservation_micro_usd)
})
&& is_unit_instance(&lease.unit_instance) && is_unit_instance(&lease.unit_instance)
&& users.insert(lease.unix_user.as_str()) && users.insert(lease.unix_user.as_str())
}) })
@ -336,8 +316,7 @@ fn is_trimmed_nonempty(text: &str) -> bool {
/// The loose shape of a model id, only to catch typos: /// The loose shape of a model id, only to catch typos:
/// `claude-<family>-<major>`, then an optional `-<minor>` and an optional /// `claude-<family>-<major>`, then an optional `-<minor>` and an optional
/// `-YYYYMMDD`, the family lowercase letters. No shape can tell an alias from /// `-YYYYMMDD`, the family lowercase letters. No shape can tell an alias from
/// a canonical id, so the known list is the price table's keys (ruling /// a canonical id, so the known list is `allowed_models` itself; an alias that slips in fails its first
/// claude-model-ids-from-price-table); an alias that slips in fails its first
/// call, when the stream's `init.model` differs from the request. /// call, when the stream's `init.model` differs from the request.
fn has_model_id_form(model: &str) -> bool { fn has_model_id_form(model: &str) -> bool {
let Some(rest) = model.strip_prefix(MODEL_PREFIX) else { let Some(rest) = model.strip_prefix(MODEL_PREFIX) else {
@ -365,25 +344,6 @@ fn has_model_id_form(model: &str) -> bool {
&& after_ok && after_ok
} }
/// A price is a fuse only if every component it bills is nonzero: a zero rate
/// lets a call cost nothing against the caller's ceiling.
fn is_fully_priced(price: &ModelPrice) -> bool {
let long = price.long_context.map(|long| long.rates);
[Some(price.rates), long]
.into_iter()
.flatten()
.all(|rates| {
[
rates.input,
rates.cache_write_short,
rates.cache_write_long,
rates.cache_read,
rates.output,
]
.iter()
.all(|rate| *rate > 0)
})
}
/// One file name under `$CREDENTIALS_DIRECTORY`, never a path. /// One file name under `$CREDENTIALS_DIRECTORY`, never a path.
fn is_credential_name(name: &str) -> bool { fn is_credential_name(name: &str) -> bool {
@ -509,7 +469,6 @@ fn lease_callers_of(config: &ClaudeConnectorConfig) -> Result<Vec<LeaseCaller>,
caller_id: lease.caller_id.clone(), caller_id: lease.caller_id.clone(),
turn: lease.turn, turn: lease.turn,
unit_instance: lease.unit_instance.clone(), unit_instance: lease.unit_instance.clone(),
reservation_micro_usd: lease.reservation_micro_usd,
}) })
.ok_or(ConfigError::Unusable("lease_callers")) .ok_or(ConfigError::Unusable("lease_callers"))
}) })
@ -531,6 +490,7 @@ fn start_leases(
config: &ClaudeConnectorConfig, config: &ClaudeConnectorConfig,
socket: &Path, socket: &Path,
service: Arc<Mutex<TransportService>>, service: Arc<Mutex<TransportService>>,
quota: Arc<crate::QuotaCell>,
) -> Result<(), ConfigError> { ) -> Result<(), ConfigError> {
use std::os::unix::fs::{FileTypeExt, PermissionsExt}; use std::os::unix::fs::{FileTypeExt, PermissionsExt};
use std::os::unix::net::UnixListener; use std::os::unix::net::UnixListener;
@ -559,6 +519,7 @@ fn start_leases(
callers, callers,
limits, limits,
service, service,
quota,
tokens, tokens,
Arc::new(SystemdUnits::system()), Arc::new(SystemdUnits::system()),
); );
@ -589,7 +550,6 @@ pub fn serve_config(path: &Path) -> Result<(), ConfigError> {
TransportService::open( TransportService::open(
&config.replay_store, &config.replay_store,
backend.provider(), backend.provider(),
config.prices.clone(),
admissions, admissions,
config.max_expiry_skew_ms, config.max_expiry_skew_ms,
) )
@ -597,7 +557,7 @@ pub fn serve_config(path: &Path) -> Result<(), ConfigError> {
)); ));
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
if let Some(socket) = &config.lease_socket { if let Some(socket) = &config.lease_socket {
start_leases(&config, socket, service.clone())?; start_leases(&config, socket, service.clone(), backend.quota())?;
} }
serve_service(listener, config.serve_limits(), service, backend)?; serve_service(listener, config.serve_limits(), service, backend)?;
Ok(()) Ok(())
@ -643,10 +603,6 @@ mod tests {
ServeError::Service(ServiceError::ReplayStore(format!("{CANARY}/replay.cc"))), ServeError::Service(ServiceError::ReplayStore(format!("{CANARY}/replay.cc"))),
"replay_store", "replay_store",
), ),
(
ServeError::Service(ServiceError::UnpricedModel(CANARY.to_string())),
"prices",
),
( (
ServeError::Service(ServiceError::InvalidAdmission), ServeError::Service(ServiceError::InvalidAdmission),
"callers", "callers",

241
crates/claude-connector/src/lease.rs
View file

@ -1,15 +1,14 @@
//! The Self-run credential lease. The connector is the only holder of the //! The Self-run credential lease. The connector is the only holder of the
//! subscription token and the one owner of Self-run admission, reservation and //! subscription token and the one owner of Self-run admission (rulings
//! the spend ledger (rulings self-runs-through-connector, //! self-runs-through-connector and self-run-credential-lease). A Self run asks
//! self-run-credential-lease and self-run-lease-billing-and-binding). A Self //! over a unix socket for a lease. The connector decides who may ask (the
//! run asks over a unix socket for a lease. The connector decides who may ask //! peer's uid, and the systemd unit invocation its process runs in), and hands
//! (the peer's uid, and the systemd unit invocation its process runs in), //! back one pipe fd that holds the token, with the plan's usage windows. The
//! reserves its own configured amount in the caller's spend window, and only //! lease lasts while the holder keeps the socket open: it ends when the holder
//! then hands back one pipe fd that holds the token. The run settles with the //! closes it, when its unit invocation ends, or when its expiry passes. Nothing
//! per-model usage it reports, and the connector bills that at true cost, //! is reserved and nothing is billed; the only usage-derived refusal is the
//! uncapped; a run that vanishes or lets its lease expire is billed the whole //! provider's own, repeated until its own reset. Nothing the run says about
//! reservation. Nothing the run says about itself decides who it is, what it //! itself decides who it is or what the plan allows.
//! reserves, or what the plan allows.
//! //!
//! The token exists in this process only as the value the token source returns //! The token exists in this process only as the value the token source returns
//! and in the pipe. It is not in an environment, an argument, a file, a log or //! and in the pipe. It is not in an environment, an argument, a file, a log or
@ -28,12 +27,11 @@ use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use model_connector::{ use model_connector::{
LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseSettlement, LeaseTurn, LeaseUse, ModelUsage, LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseTurn, PlanQuota, TransportService,
TransportService,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use crate::TokenSource; use crate::{QuotaCell, TokenSource};
/// The largest frame either side sends or accepts, in bytes. It is wire /// The largest frame either side sends or accepts, in bytes. It is wire
/// contract: a Huginn client sizes against it, so it is pinned as a literal. /// contract: a Huginn client sizes against it, so it is pinned as a literal.
@ -50,12 +48,9 @@ pub enum LeaseMessage {
/// holds the token, its write end closed. /// holds the token, its write end closed.
ClaudeLeaseGrant(LeaseGranted), ClaudeLeaseGrant(LeaseGranted),
ClaudeLeaseRefusal(LeaseRefused), ClaudeLeaseRefusal(LeaseRefused),
ClaudeLeaseSettle(LeaseSettle),
ClaudeLeaseSettled(LeaseSettled),
} }
/// A request names only when the lease ends. Who asks, and what is reserved, /// A request names only when the lease ends. Who asks is the connector's to know.
/// are the connector's to know.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeaseRequest { pub struct LeaseRequest {
pub expires_at_unix_ms: u64, pub expires_at_unix_ms: u64,
@ -64,16 +59,17 @@ pub struct LeaseRequest {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeaseGranted { pub struct LeaseGranted {
pub lease_id: String, pub lease_id: String,
pub reserved_micro_usd: u64, /// The plan's usage windows as the connector last saw them, so the run
/// can read its own burn.
pub plan_quota: Vec<PlanQuota>,
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum LeaseRefusalKind { pub enum LeaseRefusalKind {
SpendCeiling,
CostBound,
UnknownCaller, UnknownCaller,
Expired, Expired,
Busy, Busy,
ProviderRejected,
} }
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@ -82,20 +78,6 @@ pub struct LeaseRefused {
pub detail: String, pub detail: String,
} }
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeaseSettle {
pub lease_id: String,
/// What the run used of each model, as the run reports it.
pub usage: Vec<ModelUsage>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeaseSettled {
pub billed_micro_usd: u64,
/// How far the bill passed the reservation; zero when it did not.
pub overrun_micro_usd: u64,
}
/// A frame: the length, then the named-map MessagePack of `message`. /// A frame: the length, then the named-map MessagePack of `message`.
pub fn encode_frame(message: &LeaseMessage) -> Vec<u8> { pub fn encode_frame(message: &LeaseMessage) -> Vec<u8> {
// Every field of the message is plain data, so encoding cannot fail. // Every field of the message is plain data, so encoding cannot fail.
@ -180,8 +162,8 @@ fn read_message(stream: &mut impl Bounded, deadline: Instant) -> io::Result<Leas
const PIPE_BUF: usize = 4096; const PIPE_BUF: usize = 4096;
/// A unix account allowed to lease, the unit its process must run in (the /// A unix account allowed to lease, the unit its process must run in (the
/// Self or the Persona unit of its instance, by `turn`), and the caller whose /// Self or the Persona unit of its instance, by `turn`), and the
/// spend window its leases reserve in. /// caller whose name its leases carry.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct LeaseCaller { pub struct LeaseCaller {
pub uid: u32, pub uid: u32,
@ -191,8 +173,6 @@ pub struct LeaseCaller {
/// The `<instance>` of `mind-self@<instance>.service` or /// The `<instance>` of `mind-self@<instance>.service` or
/// `mind-persona@<instance>.service`. /// `mind-persona@<instance>.service`.
pub unit_instance: String, pub unit_instance: String,
/// What each of this caller's leases reserves, whatever the run asks.
pub reservation_micro_usd: u64,
} }
impl LeaseCaller { impl LeaseCaller {
@ -400,6 +380,7 @@ struct Lender {
callers: Vec<LeaseCaller>, callers: Vec<LeaseCaller>,
limits: LeaseLimits, limits: LeaseLimits,
service: Arc<Mutex<TransportService>>, service: Arc<Mutex<TransportService>>,
quota: Arc<QuotaCell>,
tokens: Arc<dyn TokenSource>, tokens: Arc<dyn TokenSource>,
units: Arc<dyn PeerUnits>, units: Arc<dyn PeerUnits>,
} }
@ -415,6 +396,7 @@ pub fn serve_leases(
callers: Vec<LeaseCaller>, callers: Vec<LeaseCaller>,
limits: LeaseLimits, limits: LeaseLimits,
service: Arc<Mutex<TransportService>>, service: Arc<Mutex<TransportService>>,
quota: Arc<QuotaCell>,
tokens: Arc<dyn TokenSource>, tokens: Arc<dyn TokenSource>,
units: Arc<dyn PeerUnits>, units: Arc<dyn PeerUnits>,
) -> io::Result<()> { ) -> io::Result<()> {
@ -422,6 +404,7 @@ pub fn serve_leases(
callers, callers,
limits, limits,
service, service,
quota,
tokens, tokens,
units, units,
}); });
@ -484,9 +467,9 @@ impl Lender {
"the expiry is not in the future or is past the lease bound", "the expiry is not in the future or is past the lease bound",
); );
} }
// The reservation is claimed in the ledger before the token is read, // The lease is claimed before the token is read, let alone lent. The
// let alone lent. A ledger that fails closes the connection with no // plan the refusal reads is the one this process last observed.
// answer and no lease. let plan = self.quota.snapshot();
let admission = self let admission = self
.service .service
.lock() .lock()
@ -495,120 +478,134 @@ impl Lender {
&caller.caller_id, &caller.caller_id,
caller.turn, caller.turn,
&invocation.invocation_id, &invocation.invocation_id,
caller.reservation_micro_usd,
request.expires_at_unix_ms, request.expires_at_unix_ms,
now, now,
&plan,
) )
.map_err(|_| io::Error::other("ledger"))?; .map_err(|_| io::Error::other("lease"))?;
let grant = match admission { let grant = match admission {
LeaseAdmission::Granted(grant) => grant, LeaseAdmission::Granted(grant) => grant,
LeaseAdmission::Refused(why) => { LeaseAdmission::Refused(why) => {
let (kind, detail) = match why { let (kind, detail) = match why {
LeaseRefusal::SpendCeiling => (
LeaseRefusalKind::SpendCeiling,
"the window cannot hold this reservation",
),
LeaseRefusal::CostBound => (
LeaseRefusalKind::CostBound,
"the reservation must be above zero",
),
LeaseRefusal::UnknownCaller => ( LeaseRefusal::UnknownCaller => (
LeaseRefusalKind::UnknownCaller, LeaseRefusalKind::UnknownCaller,
"this account is bound to no admitted caller", "this account is bound to no admitted caller",
), ),
LeaseRefusal::Busy => ( LeaseRefusal::Busy => (
LeaseRefusalKind::Busy, LeaseRefusalKind::Busy,
"a lease is open for this caller or this unit invocation leased", "a lease is open for this caller or this unit invocation",
),
LeaseRefusal::ProviderRejected => (
LeaseRefusalKind::ProviderRejected,
"the provider is refusing until its own reset",
), ),
}; };
return refuse(&mut stream, kind, detail); return refuse(&mut stream, kind, detail);
} }
}; };
self.hold(stream, caller, &grant, request.expires_at_unix_ms) self.hold(
stream,
caller,
&grant,
&invocation,
peer.process,
request.expires_at_unix_ms,
plan.windows,
)
} }
/// Everything after a reservation exists: lend the token, then settle the /// Everything after a lease exists: lend the token, then keep the lease
/// lease exactly once whatever happens. /// while its holder keeps the socket and its unit invocation, and end it
/// exactly once whatever happens.
// One argument per fact the held lease binds.
#[allow(clippy::too_many_arguments)]
fn hold( fn hold(
&self, &self,
mut stream: UnixStream, mut stream: UnixStream,
caller: &LeaseCaller, caller: &LeaseCaller,
grant: &LeaseGrant, grant: &LeaseGrant,
invocation: &PeerUnit,
process: Option<OwnedFd>,
expires_at_unix_ms: u64, expires_at_unix_ms: u64,
plan_quota: Vec<PlanQuota>,
) -> io::Result<()> { ) -> io::Result<()> {
// A token that cannot be read or piped was never lent: nothing was let held = self.keep(
// used, so nothing is billed. &mut stream,
caller,
grant,
invocation,
process.as_ref(),
expires_at_unix_ms,
plan_quota,
);
self.service
.lock()
.map_err(|_| io::Error::other("service lock"))?
.end_lease(&caller.caller_id, &grant.lease_id);
held
}
// One argument per fact the held lease binds.
#[allow(clippy::too_many_arguments)]
fn keep(
&self,
stream: &mut UnixStream,
caller: &LeaseCaller,
grant: &LeaseGrant,
invocation: &PeerUnit,
process: Option<&OwnedFd>,
expires_at_unix_ms: u64,
plan_quota: Vec<PlanQuota>,
) -> io::Result<()> {
// A token that cannot be read or piped was never lent: the lease ends.
let Some(read_end) = self let Some(read_end) = self
.tokens .tokens
.read() .read()
.and_then(|token| token_pipe(token.as_bytes()).ok()) .and_then(|token| token_pipe(token.as_bytes()).ok())
else { else {
self.settle(caller, grant, LeaseUse::NeverLent)?;
return Ok(()); return Ok(());
}; };
let frame = encode_frame(&LeaseMessage::ClaudeLeaseGrant(LeaseGranted { let frame = encode_frame(&LeaseMessage::ClaudeLeaseGrant(LeaseGranted {
lease_id: grant.lease_id.clone(), lease_id: grant.lease_id.clone(),
reserved_micro_usd: grant.reserved_micro_usd, plan_quota,
})); }));
let sent = send_with_fd(&stream, &frame, read_end.as_raw_fd()); let sent = send_with_fd(stream, &frame, read_end.as_raw_fd());
drop(read_end); drop(read_end);
if sent.is_err() { sent?;
// The descriptor may have been delivered: the reservation is billed. // The holder says nothing more. The lease ends when it closes the
self.settle(caller, grant, LeaseUse::Reservation)?; // socket, when its unit invocation is no longer the one that leased,
// or at the expiry.
let mut byte = [0_u8; 1];
loop {
let left = expires_at_unix_ms.saturating_sub(unix_ms());
if left == 0 {
return Ok(()); return Ok(());
} }
let outcome = self.await_settle(&mut stream, grant, expires_at_unix_ms); stream.set_read_timeout(Some(Duration::from_millis(left.min(HOLD_POLL_MS))))?;
match outcome { match stream.read(&mut byte) {
Some(settle) => { Ok(0) => return Ok(()),
let settlement = self.settle(caller, grant, LeaseUse::Reported(&settle.usage))?; Ok(_) => {}
let _ = stream.write_all(&encode_frame(&LeaseMessage::ClaudeLeaseSettled( Err(error)
LeaseSettled { if matches!(
billed_micro_usd: settlement.billed_micro_usd, error.kind(),
overrun_micro_usd: settlement.overrun_micro_usd, io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
}, ) => {}
))); Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
Err(_) => return Ok(()),
} }
None => { let still = process
self.settle(caller, grant, LeaseUse::Reservation)?; .and_then(|process| self.units.invocation_in(process.as_fd(), &caller.unit()))
.is_some_and(|found| found == *invocation);
if !still {
return Ok(());
} }
} }
Ok(())
}
/// The run's settle, if its whole frame arrives for this lease before the
/// lease expires. Anything else (end of file, silence, a frame still
/// trickling at the expiry, a malformed frame, the wrong lease) is no
/// report.
fn await_settle(
&self,
stream: &mut UnixStream,
grant: &LeaseGrant,
expires_at_unix_ms: u64,
) -> Option<LeaseSettle> {
let remaining = expires_at_unix_ms.saturating_sub(unix_ms());
let deadline = Instant::now() + Duration::from_millis(remaining);
match read_message(stream, deadline).ok()? {
LeaseMessage::ClaudeLeaseSettle(settle) if settle.lease_id == grant.lease_id => {
Some(settle)
}
_ => None,
}
}
fn settle(
&self,
caller: &LeaseCaller,
grant: &LeaseGrant,
usage: LeaseUse<'_>,
) -> io::Result<LeaseSettlement> {
self.service
.lock()
.map_err(|_| io::Error::other("service lock"))?
.settle_lease(&caller.caller_id, &grant.lease_id, usage)
.map_err(|_| io::Error::other("ledger"))
} }
} }
/// How often a held lease looks again at its holder's unit invocation.
const HOLD_POLL_MS: u64 = 500;
fn refuse(stream: &mut UnixStream, kind: LeaseRefusalKind, detail: &str) -> io::Result<()> { fn refuse(stream: &mut UnixStream, kind: LeaseRefusalKind, detail: &str) -> io::Result<()> {
stream.write_all(&encode_frame(&LeaseMessage::ClaudeLeaseRefusal( stream.write_all(&encode_frame(&LeaseMessage::ClaudeLeaseRefusal(
LeaseRefused { LeaseRefused {
@ -622,12 +619,12 @@ fn refuse(stream: &mut UnixStream, kind: LeaseRefusalKind, detail: &str) -> io::
mod tests { mod tests {
use super::*; use super::*;
/// A settle frame whose body is exactly `body_bytes` long. /// A refusal frame whose body is exactly `body_bytes` long.
fn settle_frame_of(body_bytes: usize) -> Vec<u8> { fn frame_of(body_bytes: usize) -> Vec<u8> {
let frame_with = |pad: usize| { let frame_with = |pad: usize| {
encode_frame(&LeaseMessage::ClaudeLeaseSettle(LeaseSettle { encode_frame(&LeaseMessage::ClaudeLeaseRefusal(LeaseRefused {
lease_id: "x".repeat(pad), kind: LeaseRefusalKind::Busy,
usage: Vec::new(), detail: "x".repeat(pad),
})) }))
}; };
// The string header is the same width between 256 and 65535 bytes. // The string header is the same width between 256 and 65535 bytes.
@ -646,10 +643,10 @@ mod tests {
#[test] #[test]
fn a_frame_is_one_whole_message_within_the_bound() { fn a_frame_is_one_whole_message_within_the_bound() {
let at_bound = settle_frame_of(MAX_LEASE_FRAME_BYTES); let at_bound = frame_of(MAX_LEASE_FRAME_BYTES);
assert_eq!(at_bound.len(), 4 + MAX_LEASE_FRAME_BYTES); assert_eq!(at_bound.len(), 4 + MAX_LEASE_FRAME_BYTES);
assert!(decode_frame(&at_bound).is_some()); assert!(decode_frame(&at_bound).is_some());
let past_bound = settle_frame_of(MAX_LEASE_FRAME_BYTES + 1); let past_bound = frame_of(MAX_LEASE_FRAME_BYTES + 1);
assert!(decode_frame(&past_bound).is_none()); assert!(decode_frame(&past_bound).is_none());
let mut trailing = at_bound.clone(); let mut trailing = at_bound.clone();
trailing.push(0); trailing.push(0);
@ -661,9 +658,9 @@ mod tests {
#[test] #[test]
fn a_stream_frame_over_the_bound_is_refused_before_it_is_read() { fn a_stream_frame_over_the_bound_is_refused_before_it_is_read() {
let past_bound = settle_frame_of(MAX_LEASE_FRAME_BYTES + 1); let past_bound = frame_of(MAX_LEASE_FRAME_BYTES + 1);
assert!(within(&past_bound).is_err()); assert!(within(&past_bound).is_err());
let at_bound = settle_frame_of(MAX_LEASE_FRAME_BYTES); let at_bound = frame_of(MAX_LEASE_FRAME_BYTES);
assert!(within(&at_bound).is_ok()); assert!(within(&at_bound).is_ok());
assert!(within(&[0_u8, 0, 0, 0]).is_err()); assert!(within(&[0_u8, 0, 0, 0]).is_err());
// Short of the length it announced is end of file, not a message. // Short of the length it announced is end of file, not a message.
@ -673,9 +670,9 @@ mod tests {
#[test] #[test]
fn a_frame_that_trickles_past_its_deadline_is_refused_however_small_each_read() { fn a_frame_that_trickles_past_its_deadline_is_refused_however_small_each_read() {
let (mut reader, mut writer) = UnixStream::pair().unwrap(); let (mut reader, mut writer) = UnixStream::pair().unwrap();
let frame = encode_frame(&LeaseMessage::ClaudeLeaseSettle(LeaseSettle { let frame = encode_frame(&LeaseMessage::ClaudeLeaseRefusal(LeaseRefused {
lease_id: "lease:x".to_string(), kind: LeaseRefusalKind::Busy,
usage: Vec::new(), detail: "lease:x".to_string(),
})); }));
let trickle = thread::spawn(move || { let trickle = thread::spawn(move || {
for byte in frame { for byte in frame {

2
crates/claude-connector/src/lib.rs
View file

@ -28,7 +28,7 @@ pub use launch::ChildEnv;
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
pub use lease::{ pub use lease::{
LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused,
LeaseRequest, LeaseSettle, LeaseSettled, MAX_LEASE_FRAME_BYTES, PeerCredentials, PeerUnit, LeaseRequest, MAX_LEASE_FRAME_BYTES, PeerCredentials, PeerUnit,
PeerUnits, decode_frame, encode_frame, peer_credentials, serve_leases, user_uid, PeerUnits, decode_frame, encode_frame, peer_credentials, serve_leases, user_uid,
}; };
#[cfg(unix)] #[cfg(unix)]

58
crates/claude-connector/src/main.rs
View file

@ -18,7 +18,7 @@ use claude_connector::{
ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, MIN_SECRET_BYTES, check_ready, ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, MIN_SECRET_BYTES, check_ready,
load_config, write_config, load_config, write_config,
}; };
use model_connector::{CallerConfig, LeaseTurn, ModelPrice, Rates, frame_bytes_for_payload}; use model_connector::{CallerConfig, LeaseTurn, frame_bytes_for_payload};
fn main() -> ExitCode { fn main() -> ExitCode {
let args: Vec<String> = std::env::args().skip(1).collect(); let args: Vec<String> = std::env::args().skip(1).collect();
@ -72,7 +72,6 @@ fn check(path: &str) -> Result<(), Failure> {
println!("claude connector configuration is valid and ready to serve"); println!("claude connector configuration is valid and ready to serve");
println!("callers: {}", config.callers.len()); println!("callers: {}", config.callers.len());
println!("lease_callers: {}", config.lease_callers.len()); println!("lease_callers: {}", config.lease_callers.len());
println!("prices: {}", config.prices.len());
Ok(()) Ok(())
} }
@ -85,17 +84,15 @@ struct Flags {
/// The flags that describe one caller, shared by `config init` (which writes /// The flags that describe one caller, shared by `config init` (which writes
/// the first caller) and `config add-caller`. /// the first caller) and `config add-caller`.
const CALLER: [&str; 7] = [ const CALLER: [&str; 5] = [
"--caller-id", "--caller-id",
"--connection-key-file", "--connection-key-file",
"--connection-key-epoch", "--connection-key-epoch",
"--spend-ceiling-micro-usd",
"--spend-window-ms",
"--max-concurrent-requests", "--max-concurrent-requests",
"--max-output-tokens", "--max-output-tokens",
]; ];
const INIT: [&str; 22] = [ const INIT: [&str; 21] = [
"--listen", "--listen",
"--replay-store", "--replay-store",
"--claude-binary", "--claude-binary",
@ -117,15 +114,9 @@ const INIT: [&str; 22] = [
"--lease-max-ms", "--lease-max-ms",
"--turn", "--turn",
"--lease-unit-instance", "--lease-unit-instance",
"--lease-reservation-micro-usd",
]; ];
const LEASE_CALLER: [&str; 4] = [ const LEASE_CALLER: [&str; 3] = ["--caller-id", "--turn", "--lease-unit-instance"];
"--caller-id",
"--turn",
"--lease-unit-instance",
"--lease-reservation-micro-usd",
];
impl Flags { impl Flags {
fn parse( fn parse(
@ -213,8 +204,6 @@ impl Flags {
max_concurrent_requests: self.number_or("--max-concurrent-requests", 4)?, max_concurrent_requests: self.number_or("--max-concurrent-requests", 4)?,
max_payload_bytes: self.number_or("--max-payload-bytes", 1 << 20)?, max_payload_bytes: self.number_or("--max-payload-bytes", 1 << 20)?,
max_output_tokens: self.number_or("--max-output-tokens", 32_768)?, max_output_tokens: self.number_or("--max-output-tokens", 32_768)?,
spend_ceiling_micro_usd: self.number("--spend-ceiling-micro-usd")?,
spend_window_ms: self.number("--spend-window-ms")?,
}) })
} }
} }
@ -230,33 +219,6 @@ fn sha256(hex: &str) -> Option<[u8; 32]> {
Some(digest) Some(digest)
} }
/// MODEL:INPUT:OUTPUT:CACHE_WRITE:CACHE_READ:OVERHEAD, rates in micro-USD per
/// million tokens. The one cache-write rate prices both retentions.
fn price(text: &str) -> Option<(String, ModelPrice)> {
let mut parts = text.split(':');
let model = parts.next()?.to_string();
let mut next = || parts.next()?.parse::<u64>().ok();
let (input, output, cache_write, cache_read, overhead_tokens) =
(next()?, next()?, next()?, next()?, next()?);
if parts.next().is_some() {
return None;
}
Some((
model,
ModelPrice {
rates: Rates {
input,
cache_write_short: cache_write,
cache_write_long: cache_write,
cache_read,
output,
},
long_context: None,
overhead_tokens,
},
))
}
/// Writes 32 random bytes as 64 hex characters, mode 0600, unless the file is /// Writes 32 random bytes as 64 hex characters, mode 0600, unless the file is
/// already there: the secret is never read back or shown, and a file that /// already there: the secret is never read back or shown, and a file that
/// exists is the secret handles already depend on, so it is kept as it is when /// exists is the secret handles already depend on, so it is kept as it is when
@ -322,16 +284,9 @@ fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
let flags = Flags::parse( let flags = Flags::parse(
args, args,
&[&INIT, &CALLER], &[&INIT, &CALLER],
&["--allowed-model", "--price", "--lease-user"], &["--allowed-model", "--lease-user"],
)?; )?;
let first = flags.caller()?; let first = flags.caller()?;
let mut prices = BTreeMap::new();
for text in flags.list("--price") {
let (model, price) = price(text).ok_or(Failure::Malformed("--price"))?;
if prices.insert(model, price).is_some() {
return Err(Failure::Malformed("--price"));
}
}
let lease_callers = flags let lease_callers = flags
.list("--lease-user") .list("--lease-user")
.iter() .iter()
@ -341,7 +296,6 @@ fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
caller_id: first.caller_runtime_id.clone(), caller_id: first.caller_runtime_id.clone(),
turn: flags.turn()?, turn: flags.turn()?,
unit_instance: flags.text("--lease-unit-instance")?, unit_instance: flags.text("--lease-unit-instance")?,
reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?,
}) })
}) })
.collect::<Result<Vec<_>, Failure>>()?; .collect::<Result<Vec<_>, Failure>>()?;
@ -366,7 +320,6 @@ fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
socket_timeout_ms: flags.number_or("--socket-timeout-ms", 10_000)?, socket_timeout_ms: flags.number_or("--socket-timeout-ms", 10_000)?,
max_expiry_skew_ms: 300_000, max_expiry_skew_ms: 300_000,
callers: vec![first], callers: vec![first],
prices,
lease_socket: flags.single.get("--lease-socket").map(PathBuf::from), lease_socket: flags.single.get("--lease-socket").map(PathBuf::from),
lease_callers, lease_callers,
lease_max_ms: flags.number_or("--lease-max-ms", 3_600_000)?, lease_max_ms: flags.number_or("--lease-max-ms", 3_600_000)?,
@ -412,7 +365,6 @@ fn add_lease_caller(path: &str, args: &[&str]) -> Result<(), Failure> {
caller_id: flags.text("--caller-id")?, caller_id: flags.text("--caller-id")?,
turn: flags.turn()?, turn: flags.turn()?,
unit_instance: flags.text("--lease-unit-instance")?, unit_instance: flags.text("--lease-unit-instance")?,
reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?,
}); });
write_config(path.as_ref(), &config)?; write_config(path.as_ref(), &config)?;
Ok(()) Ok(())

2
crates/claude-connector/src/quota.rs
View file

@ -1,7 +1,7 @@
//! The plan's own usage windows, as the CLI reports them. The subscription //! The plan's own usage windows, as the CLI reports them. The subscription
//! decides how much a plan may burn; the CLI's `rate_limit_event` is the only //! decides how much a plan may burn; the CLI's `rate_limit_event` is the only
//! place that number appears, so the latest observation of each window is kept //! place that number appears, so the latest observation of each window is kept
//! here and read out beside the connector's own spend ledger. //! here and published as the usage readout and lent with each lease.
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::sync::Mutex; use std::sync::Mutex;

6
crates/claude-connector/src/stream.rs
View file

@ -14,7 +14,7 @@ use crate::quota::QuotaCell;
const STRUCTURED_OUTPUT_TOOL: &str = "StructuredOutput"; const STRUCTURED_OUTPUT_TOOL: &str = "StructuredOutput";
/// A call that cannot complete, with the stage that failed. `usage` is `None` /// A call that cannot complete, with the stage that failed. `usage` is `None`
/// (the kit bills its reservation) unless a result event reported it or the /// (unknown) unless a result event reported it or the
/// call never reached the provider (zero). /// call never reached the provider (zero).
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
pub(crate) struct Failure { pub(crate) struct Failure {
@ -25,7 +25,7 @@ pub(crate) struct Failure {
impl Failure { impl Failure {
/// A failure after the child ran: what the provider billed is unknown, so /// A failure after the child ran: what the provider billed is unknown, so
/// the kit bills its reservation. /// the usage is unknown.
pub(crate) fn cost_unknown(kind: FailureKind, message: &'static str) -> Self { pub(crate) fn cost_unknown(kind: FailureKind, message: &'static str) -> Self {
Self { Self {
kind, kind,
@ -356,7 +356,7 @@ fn fail(kind: FailureKind, message: &'static str) -> Step {
/// provider cannot have billed (`empty_is_zero`: a refusal at the door, 401, /// provider cannot have billed (`empty_is_zero`: a refusal at the door, 401,
/// 403 or 429, or a call never made); elsewhere the CLI also zeroes usage on /// 403 or 429, or a call never made); elsewhere the CLI also zeroes usage on
/// crashes after real work. A call never made (`absent_is_zero`) may also /// crashes after real work. A call never made (`absent_is_zero`) may also
/// omit the field. `None` is billed the reservation by the kit. /// omit the field. `None` is unknown usage.
fn read_usage(reported: Option<Value>, empty_is_zero: bool, absent_is_zero: bool) -> Option<Usage> { fn read_usage(reported: Option<Value>, empty_is_zero: bool, absent_is_zero: bool) -> Option<Usage> {
let Some(reported) = reported else { let Some(reported) = reported else {
return absent_is_zero.then(|| sum_usage(std::iter::empty())); return absent_is_zero.then(|| sum_usage(std::iter::empty()));

1670
crates/claude-connector/tests/config_document.rs
View file

File diff suppressed because it is too large Load diff

BIN
crates/claude-connector/tests/fixtures/lease_v1/grant.bin vendored
View file

Binary file not shown.

BIN
crates/claude-connector/tests/fixtures/lease_v1/refusal.bin vendored
View file

Binary file not shown.

BIN
crates/claude-connector/tests/fixtures/lease_v1/request.bin vendored
View file

Binary file not shown.

BIN
crates/claude-connector/tests/fixtures/lease_v1/settle.bin vendored
View file

Binary file not shown.

BIN
crates/claude-connector/tests/fixtures/lease_v1/settled.bin vendored
View file

Binary file not shown.

1302
crates/claude-connector/tests/lease.rs
View file

File diff suppressed because it is too large Load diff

34
crates/model-connector/src/client.rs
View file

@ -9,11 +9,11 @@ use rand_core::{OsRng, RngCore};
use crate::contract::{TransportInvocation, TransportKey, TransportResult}; use crate::contract::{TransportInvocation, TransportKey, TransportResult};
use crate::envelope::{ use crate::envelope::{
MAX_IDENTITY_BYTES, ServiceError, TransportFrameError, decode_result_frame, MAX_IDENTITY_BYTES, ServiceError, TransportFrameError, decode_result_frame,
decode_spend_readout_response_frame, decrypt_result, decrypt_spend_readout, decode_usage_readout_response_frame, decrypt_result, decrypt_usage_readout,
encode_invocation_frame, encode_spend_readout_request_frame, encrypt_invocation, encode_invocation_frame, encode_usage_readout_request_frame, encrypt_invocation,
encrypt_spend_readout_request, read_transport_frame, write_transport_frame, encrypt_usage_readout_request, read_transport_frame, write_transport_frame,
}; };
use crate::readout::SpendReadout; use crate::readout::UsageReadout;
#[derive(Clone)] #[derive(Clone)]
pub struct ConnectorClient { pub struct ConnectorClient {
@ -62,25 +62,25 @@ impl ConnectorClient {
decrypt_result(&envelope, &self.security, invocation).map_err(Into::into) decrypt_result(&envelope, &self.security, invocation).map_err(Into::into)
} }
/// Reads the connector's spend ledger and plan quota as `caller_runtime_id`, /// Reads the plan's usage windows as `caller_runtime_id`,
/// which must be a caller whose connection key this client holds. /// which must be a caller whose connection key this client holds.
pub fn spend_readout( pub fn usage_readout(
&self, &self,
caller_runtime_id: &str, caller_runtime_id: &str,
) -> Result<SpendReadout, ConnectorClientError> { ) -> Result<UsageReadout, ConnectorClientError> {
let mut nonce = [0_u8; 16]; let mut nonce = [0_u8; 16];
OsRng.fill_bytes(&mut nonce); OsRng.fill_bytes(&mut nonce);
let request_id = format!( let request_id = format!(
"spend-readout-{}", "usage-readout-{}",
nonce.iter().map(|b| format!("{b:02x}")).collect::<String>() nonce.iter().map(|b| format!("{b:02x}")).collect::<String>()
); );
let sealed = encrypt_spend_readout_request(caller_runtime_id, &request_id, &self.security)?; let sealed = encrypt_usage_readout_request(caller_runtime_id, &request_id, &self.security)?;
let response = self.exchange(&encode_spend_readout_request_frame(&sealed)?)?; let response = self.exchange(&encode_usage_readout_request_frame(&sealed)?)?;
let envelope = decode_spend_readout_response_frame(&response)?; let envelope = decode_usage_readout_response_frame(&response)?;
if envelope.caller_runtime_id != caller_runtime_id || envelope.request_id != request_id { if envelope.caller_runtime_id != caller_runtime_id || envelope.request_id != request_id {
return Err(ServiceError::OuterIdentity.into()); return Err(ServiceError::OuterIdentity.into());
} }
decrypt_spend_readout(&envelope, &self.security).map_err(Into::into) decrypt_usage_readout(&envelope, &self.security).map_err(Into::into)
} }
/// Sends one frame on a fresh connection and reads the one answer. /// Sends one frame on a fresh connection and reads the one answer.
@ -274,17 +274,17 @@ mod fixture_tests {
let server = std::thread::spawn(move || { let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap(); let (mut stream, _) = listener.accept().unwrap();
let request = read_transport_frame(&mut stream, 64 * 1024).unwrap(); let request = read_transport_frame(&mut stream, 64 * 1024).unwrap();
let RequestFrame::SpendReadout(asked) = decode_request_frame(&request).unwrap() let RequestFrame::UsageReadout(asked) = decode_request_frame(&request).unwrap()
else { else {
panic!("the client sent something other than a readout request"); panic!("the client sent something other than a readout request");
}; };
let request_id = answer_request.unwrap_or(&asked.request_id); let request_id = answer_request.unwrap_or(&asked.request_id);
let readout = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default()); let readout = UsageReadout::new(1, PlanQuotaSnapshot::default());
let sealed = let sealed =
encrypt_spend_readout(answer_caller, request_id, &readout, &key).unwrap(); encrypt_usage_readout(answer_caller, request_id, &readout, &key).unwrap();
write_transport_frame( write_transport_frame(
&mut stream, &mut stream,
&encode_spend_readout_response_frame(&sealed).unwrap(), &encode_usage_readout_response_frame(&sealed).unwrap(),
64 * 1024, 64 * 1024,
) )
.unwrap(); .unwrap();
@ -296,7 +296,7 @@ mod fixture_tests {
Some(Duration::from_secs(1)), Some(Duration::from_secs(1)),
) )
.unwrap(); .unwrap();
let answer = client.spend_readout("caller-1"); let answer = client.usage_readout("caller-1");
if accepted { if accepted {
assert_eq!(answer.unwrap().at_unix_ms, 1); assert_eq!(answer.unwrap().at_unix_ms, 1);
} else { } else {

83
crates/model-connector/src/contract.rs
View file

@ -9,11 +9,11 @@ use zeroize::Zeroize;
use crate::envelope::{ServiceError, binary_schema}; use crate::envelope::{ServiceError, binary_schema};
pub const PROVIDER_REQUEST_SCHEMA_ID: &str = "gamecult.model.provider_request.v3"; pub const PROVIDER_REQUEST_SCHEMA_ID: &str = "gamecult.model.provider_request.v4";
pub const INVOCATION_SCHEMA_ID: &str = "gamecult.model.transport_invocation.v3"; pub const INVOCATION_SCHEMA_ID: &str = "gamecult.model.transport_invocation.v4";
pub const RESULT_SCHEMA_ID: &str = "gamecult.model.transport_result.v3"; pub const RESULT_SCHEMA_ID: &str = "gamecult.model.transport_result.v4";
pub const RECEIPT_SCHEMA_ID: &str = "gamecult.model.transport_receipt.v3"; pub const RECEIPT_SCHEMA_ID: &str = "gamecult.model.transport_receipt.v4";
pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.model.transport_envelope.v3"; pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.model.transport_envelope.v4";
#[derive(Clone, PartialEq, Eq)] #[derive(Clone, PartialEq, Eq)]
pub struct TransportKey(pub(crate) [u8; 32]); pub struct TransportKey(pub(crate) [u8; 32]);
@ -190,10 +190,10 @@ pub enum FailureKind {
Provider, Provider,
} }
/// Token accounting for one request, as the backend reports it; the service /// Token accounting for one request, as the backend reports it. Cache reads
/// prices it. Cache reads and writes are counted apart from plain input. /// and writes are counted apart from plain input.
/// `output_tokens` includes reasoning tokens: `reasoning_tokens` is a detail /// `output_tokens` includes reasoning tokens: `reasoning_tokens` is a detail
/// (absent when the provider does not report it) and is never priced apart. /// (absent when the provider does not report it) and is never counted apart.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct Usage { pub struct Usage {
@ -252,10 +252,6 @@ pub struct TransportInvocation {
pub expires_at_unix_ms: u64, pub expires_at_unix_ms: u64,
pub native_request_sha256: [u8; 32], pub native_request_sha256: [u8; 32],
pub provider_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, pub request: ProviderRequest,
} }
@ -264,7 +260,6 @@ impl TransportInvocation {
caller_runtime_id: impl Into<String>, caller_runtime_id: impl Into<String>,
expires_at_unix_ms: u64, expires_at_unix_ms: u64,
native_request_sha256: [u8; 32], native_request_sha256: [u8; 32],
max_cost_micro_usd: u64,
request: ProviderRequest, request: ProviderRequest,
) -> Result<Self, ContractError> { ) -> Result<Self, ContractError> {
let provider_request_sha256 = provider_request_sha256(&request)?; let provider_request_sha256 = provider_request_sha256(&request)?;
@ -274,7 +269,6 @@ impl TransportInvocation {
expires_at_unix_ms, expires_at_unix_ms,
native_request_sha256, native_request_sha256,
provider_request_sha256, provider_request_sha256,
max_cost_micro_usd,
request, request,
}) })
} }
@ -287,9 +281,6 @@ impl TransportInvocation {
if self.native_request_sha256 == [0; 32] { if self.native_request_sha256 == [0; 32] {
return Err(ContractError::Invalid("native_request_sha256")); 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()?; self.request.validate()?;
if self.provider_request_sha256 != provider_request_sha256(&self.request)? { if self.provider_request_sha256 != provider_request_sha256(&self.request)? {
return Err(ContractError::ProviderDigest); return Err(ContractError::ProviderDigest);
@ -485,15 +476,6 @@ pub struct TransportReceipt {
/// How many foreign `ReasoningState` items the service removed from the /// How many foreign `ReasoningState` items the service removed from the
/// request before the backend saw it. /// request before the backend saw it.
pub foreign_reasoning_dropped: u32, 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, pub outcome: TransportOutcome,
} }
@ -558,12 +540,6 @@ pub enum Refusal {
ModelNotAllowed, ModelNotAllowed,
OutputBound, OutputBound,
PayloadBound, 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, Capacity,
InFlight, InFlight,
Indeterminate, Indeterminate,
@ -573,15 +549,12 @@ pub enum Refusal {
// exchange, where dropping it would break the provider's own rules. A plain // 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. // comment: a doc comment would split the published schema into a union.
ForeignReasoningState, ForeignReasoningState,
// The caller's newest admission window starts after the clock's reading, // The provider itself is refusing right now: a plan window it reported
// and the request does not fit what that window holds: the host's clock // `rejected` has not reached its own reset. Plain comments, as above.
// is behind the window and the caller stays refused until it catches up. ProviderRejected,
// 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 /// The one wire codec of every v4 document: named MessagePack (struct fields
/// are keyed by name), with unknown fields refused by the types themselves. /// are keyed by name), with unknown fields refused by the types themselves.
pub(crate) fn wire_encode<T: Serialize + ?Sized>(document: &T) -> Result<Vec<u8>, ContractError> { pub(crate) fn wire_encode<T: Serialize + ?Sized>(document: &T) -> Result<Vec<u8>, ContractError> {
rmp_serde::to_vec_named(document).map_err(|_| ContractError::Encoding) rmp_serde::to_vec_named(document).map_err(|_| ContractError::Encoding)
@ -787,28 +760,16 @@ mod tests {
} }
#[test] #[test]
fn max_cost_zero_is_malformed() { fn the_invocations_identity_covers_its_request() {
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 one = invocation();
let mut other = one.clone(); let mut other = one.clone();
other.max_cost_micro_usd += 1; other.request.max_output_tokens += 1;
assert_ne!( assert_ne!(
canonical_invocation_bytes(&one).unwrap(), canonical_invocation_bytes(&one).unwrap(),
canonical_invocation_bytes(&other).unwrap() canonical_invocation_bytes(&other).unwrap()
); );
let decoded: TransportInvocation = wire_decode(&wire_encode(&one).unwrap()).unwrap(); let decoded: TransportInvocation = wire_decode(&wire_encode(&one).unwrap()).unwrap();
assert_eq!(decoded.max_cost_micro_usd, one.max_cost_micro_usd); assert_eq!(decoded, one);
} }
#[test] #[test]
@ -869,7 +830,7 @@ mod tests {
} }
#[test] #[test]
fn v3_documents_carry_their_v3_ids() { fn v4_documents_carry_their_v4_ids() {
assert_eq!( assert_eq!(
[ [
PROVIDER_REQUEST_SCHEMA_ID, PROVIDER_REQUEST_SCHEMA_ID,
@ -879,11 +840,11 @@ mod tests {
ENVELOPE_SCHEMA_ID ENVELOPE_SCHEMA_ID
], ],
[ [
"gamecult.model.provider_request.v3", "gamecult.model.provider_request.v4",
"gamecult.model.transport_invocation.v3", "gamecult.model.transport_invocation.v4",
"gamecult.model.transport_result.v3", "gamecult.model.transport_result.v4",
"gamecult.model.transport_receipt.v3", "gamecult.model.transport_receipt.v4",
"gamecult.model.transport_envelope.v3" "gamecult.model.transport_envelope.v4"
] ]
); );
let invocation = invocation(); let invocation = invocation();
@ -922,7 +883,7 @@ mod tests {
let bytes = wire_encode(document).unwrap(); let bytes = wire_encode(document).unwrap();
let mut value = rmpv::decode::read_value(&mut bytes.as_slice()).unwrap(); let mut value = rmpv::decode::read_value(&mut bytes.as_slice()).unwrap();
let rmpv::Value::Map(entries) = &mut value else { let rmpv::Value::Map(entries) = &mut value else {
panic!("a v3 document is a named map"); panic!("a v4 document is a named map");
}; };
entries.push((rmpv::Value::from("unexpected_field"), rmpv::Value::from(1))); entries.push((rmpv::Value::from("unexpected_field"), rmpv::Value::from(1)));
let mut out = Vec::new(); let mut out = Vec::new();

94
crates/model-connector/src/envelope.rs
View file

@ -17,14 +17,14 @@ use crate::contract::{
ContractError, ENVELOPE_SCHEMA_ID, TransportInvocation, TransportKey, TransportResult, ContractError, ENVELOPE_SCHEMA_ID, TransportInvocation, TransportKey, TransportResult,
canonical_invocation_bytes, require_id, wire_decode, wire_encode, canonical_invocation_bytes, require_id, wire_decode, wire_encode,
}; };
use crate::readout::SpendReadout; use crate::readout::UsageReadout;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub enum EnvelopeKind { pub enum EnvelopeKind {
Invocation, Invocation,
Result, Result,
SpendReadoutRequest, UsageReadoutRequest,
SpendReadoutResult, UsageReadoutResult,
} }
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
@ -54,8 +54,8 @@ pub(crate) fn binary_schema(_: &mut schemars::SchemaGenerator) -> schemars::Sche
pub const CONNECTOR_SERVICE_ID: &str = "gamecult.model.connector"; pub const CONNECTOR_SERVICE_ID: &str = "gamecult.model.connector";
/// The operation that runs one sealed invocation. /// The operation that runs one sealed invocation.
pub const INVOKE_OPERATION: &str = "invoke"; pub const INVOKE_OPERATION: &str = "invoke";
/// The operation that reads the spend ledger and the plan's quota. /// The operation that reads the plan's usage.
pub const SPEND_READOUT_OPERATION: &str = "spend_readout"; pub const USAGE_READOUT_OPERATION: &str = "usage_readout";
const PAYLOAD_ENCODING: &str = "messagepack-base64"; const PAYLOAD_ENCODING: &str = "messagepack-base64";
const OK_STATUS: &str = "ok"; const OK_STATUS: &str = "ok";
@ -78,19 +78,19 @@ fn seal_payload(envelope: &TransportEnvelope) -> Result<String, ServiceError> {
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Operation { pub(crate) enum Operation {
Invoke, Invoke,
SpendReadout, UsageReadout,
} }
impl Operation { impl Operation {
fn name(self) -> &'static str { fn name(self) -> &'static str {
match self { match self {
Operation::Invoke => INVOKE_OPERATION, Operation::Invoke => INVOKE_OPERATION,
Operation::SpendReadout => SPEND_READOUT_OPERATION, Operation::UsageReadout => USAGE_READOUT_OPERATION,
} }
} }
fn from_name(name: &str) -> Option<Self> { pub(crate) fn from_name(name: &str) -> Option<Self> {
[Operation::Invoke, Operation::SpendReadout] [Operation::Invoke, Operation::UsageReadout]
.into_iter() .into_iter()
.find(|operation| operation.name() == name) .find(|operation| operation.name() == name)
} }
@ -141,10 +141,10 @@ pub(crate) fn encode_invocation_frame(
request_frame(Operation::Invoke, envelope) request_frame(Operation::Invoke, envelope)
} }
pub(crate) fn encode_spend_readout_request_frame( pub(crate) fn encode_usage_readout_request_frame(
envelope: &TransportEnvelope, envelope: &TransportEnvelope,
) -> Result<Vec<u8>, ServiceError> { ) -> Result<Vec<u8>, ServiceError> {
request_frame(Operation::SpendReadout, envelope) request_frame(Operation::UsageReadout, envelope)
} }
#[cfg(any(feature = "daemon", test))] #[cfg(any(feature = "daemon", test))]
@ -153,10 +153,10 @@ pub(crate) fn encode_result_frame(envelope: &TransportEnvelope) -> Result<Vec<u8
} }
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
pub(crate) fn encode_spend_readout_response_frame( pub(crate) fn encode_usage_readout_response_frame(
envelope: &TransportEnvelope, envelope: &TransportEnvelope,
) -> Result<Vec<u8>, ServiceError> { ) -> Result<Vec<u8>, ServiceError> {
response_frame(Operation::SpendReadout, envelope) response_frame(Operation::UsageReadout, envelope)
} }
/// Opens the carriage of one frame. Everything the carriage says is checked /// Opens the carriage of one frame. Everything the carriage says is checked
@ -193,7 +193,7 @@ fn read_carriage(bytes: &[u8]) -> Result<CultNetMessage, ServiceError> {
#[cfg(any(feature = "daemon", test))] #[cfg(any(feature = "daemon", test))]
pub(crate) enum RequestFrame { pub(crate) enum RequestFrame {
Invoke(TransportEnvelope), Invoke(TransportEnvelope),
SpendReadout(TransportEnvelope), UsageReadout(TransportEnvelope),
} }
#[cfg(any(feature = "daemon", test))] #[cfg(any(feature = "daemon", test))]
@ -215,7 +215,7 @@ pub(crate) fn decode_request_frame(bytes: &[u8]) -> Result<RequestFrame, Service
&payload, &payload,
)? { )? {
(Operation::Invoke, envelope) => Ok(RequestFrame::Invoke(envelope)), (Operation::Invoke, envelope) => Ok(RequestFrame::Invoke(envelope)),
(Operation::SpendReadout, envelope) => Ok(RequestFrame::SpendReadout(envelope)), (Operation::UsageReadout, envelope) => Ok(RequestFrame::UsageReadout(envelope)),
} }
} }
_ => Err(ServiceError::Carriage), _ => Err(ServiceError::Carriage),
@ -226,7 +226,7 @@ pub(crate) fn decode_request_frame(bytes: &[u8]) -> Result<RequestFrame, Service
pub(crate) fn decode_invocation_frame(bytes: &[u8]) -> Result<TransportEnvelope, ServiceError> { pub(crate) fn decode_invocation_frame(bytes: &[u8]) -> Result<TransportEnvelope, ServiceError> {
match decode_request_frame(bytes)? { match decode_request_frame(bytes)? {
RequestFrame::Invoke(envelope) => Ok(envelope), RequestFrame::Invoke(envelope) => Ok(envelope),
RequestFrame::SpendReadout(_) => Err(ServiceError::Carriage), RequestFrame::UsageReadout(_) => Err(ServiceError::Carriage),
} }
} }
@ -263,10 +263,10 @@ pub(crate) fn decode_result_frame(bytes: &[u8]) -> Result<TransportEnvelope, Ser
decode_response_frame(bytes, Operation::Invoke) decode_response_frame(bytes, Operation::Invoke)
} }
pub(crate) fn decode_spend_readout_response_frame( pub(crate) fn decode_usage_readout_response_frame(
bytes: &[u8], bytes: &[u8],
) -> Result<TransportEnvelope, ServiceError> { ) -> Result<TransportEnvelope, ServiceError> {
decode_response_frame(bytes, Operation::SpendReadout) decode_response_frame(bytes, Operation::UsageReadout)
} }
/// The most bytes one frame holding a sealed envelope of `ciphertext_len` /// The most bytes one frame holding a sealed envelope of `ciphertext_len`
@ -416,9 +416,9 @@ pub fn decrypt_result(
Ok(result) Ok(result)
} }
/// A spend readout request carries no content: the sealed envelope's /// A usage readout request carries no content: the sealed envelope's
/// authenticated identity and kind are the whole request. /// authenticated identity and kind are the whole request.
pub(crate) fn encrypt_spend_readout_request( pub(crate) fn encrypt_usage_readout_request(
caller_runtime_id: &str, caller_runtime_id: &str,
request_id: &str, request_id: &str,
security: &TransportKey, security: &TransportKey,
@ -426,45 +426,45 @@ pub(crate) fn encrypt_spend_readout_request(
encrypt_message( encrypt_message(
caller_runtime_id, caller_runtime_id,
request_id, request_id,
EnvelopeKind::SpendReadoutRequest, EnvelopeKind::UsageReadoutRequest,
&[], &[],
security, security,
) )
} }
#[cfg(any(feature = "daemon", test))] #[cfg(any(feature = "daemon", test))]
pub(crate) fn decrypt_spend_readout_request( pub(crate) fn decrypt_usage_readout_request(
envelope: &TransportEnvelope, envelope: &TransportEnvelope,
security: &TransportKey, security: &TransportKey,
) -> Result<(), ServiceError> { ) -> Result<(), ServiceError> {
validate_envelope(envelope, EnvelopeKind::SpendReadoutRequest)?; validate_envelope(envelope, EnvelopeKind::UsageReadoutRequest)?;
decrypt_message(envelope, security).map(drop) decrypt_message(envelope, security).map(drop)
} }
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
pub(crate) fn encrypt_spend_readout( pub(crate) fn encrypt_usage_readout(
caller_runtime_id: &str, caller_runtime_id: &str,
request_id: &str, request_id: &str,
readout: &SpendReadout, readout: &UsageReadout,
security: &TransportKey, security: &TransportKey,
) -> Result<TransportEnvelope, ServiceError> { ) -> Result<TransportEnvelope, ServiceError> {
let plaintext = wire_encode(readout).map_err(|_| ServiceError::Encoding)?; let plaintext = wire_encode(readout).map_err(|_| ServiceError::Encoding)?;
encrypt_message( encrypt_message(
caller_runtime_id, caller_runtime_id,
request_id, request_id,
EnvelopeKind::SpendReadoutResult, EnvelopeKind::UsageReadoutResult,
&plaintext, &plaintext,
security, security,
) )
} }
pub(crate) fn decrypt_spend_readout( pub(crate) fn decrypt_usage_readout(
envelope: &TransportEnvelope, envelope: &TransportEnvelope,
security: &TransportKey, security: &TransportKey,
) -> Result<SpendReadout, ServiceError> { ) -> Result<UsageReadout, ServiceError> {
validate_envelope(envelope, EnvelopeKind::SpendReadoutResult)?; validate_envelope(envelope, EnvelopeKind::UsageReadoutResult)?;
let plaintext = decrypt_message(envelope, security)?; let plaintext = decrypt_message(envelope, security)?;
let readout: SpendReadout = wire_decode(&plaintext).map_err(|_| ServiceError::Encoding)?; let readout: UsageReadout = wire_decode(&plaintext).map_err(|_| ServiceError::Encoding)?;
readout.validate()?; readout.validate()?;
Ok(readout) Ok(readout)
} }
@ -607,8 +607,6 @@ pub enum ServiceError {
SharedCallerKey, SharedCallerKey,
#[error("connector replay store failed: {0}")] #[error("connector replay store failed: {0}")]
ReplayStore(String), ReplayStore(String),
#[error("an allowed model has no price: {0}")]
UnpricedModel(String),
#[error("connector replay record is malformed")] #[error("connector replay record is malformed")]
InvalidReplayRecord, InvalidReplayRecord,
#[error("caller key changed while durable replay records still exist")] #[error("caller key changed while durable replay records still exist")]
@ -1072,7 +1070,7 @@ mod tests {
fn golden_frame() -> Vec<u8> { fn golden_frame() -> Vec<u8> {
std::fs::read( std::fs::read(
std::path::Path::new(env!("CARGO_MANIFEST_DIR")) std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures/wire_v3/carriage_invocation.frame"), .join("tests/fixtures/wire_v4/carriage_invocation.frame"),
) )
.expect("carriage_invocation.frame is committed") .expect("carriage_invocation.frame is committed")
} }
@ -1109,7 +1107,7 @@ mod tests {
assert_eq!(text(&request, "operation"), "invoke"); assert_eq!(text(&request, "operation"), "invoke");
assert_eq!( assert_eq!(
text(&request, "payloadSchema"), text(&request, "payloadSchema"),
"gamecult.model.transport_envelope.v3" "gamecult.model.transport_envelope.v4"
); );
assert_eq!(text(&request, "payloadEncoding"), "messagepack-base64"); assert_eq!(text(&request, "payloadEncoding"), "messagepack-base64");
assert_eq!(text(&request, "messageId"), envelope.request_id); assert_eq!(text(&request, "messageId"), envelope.request_id);
@ -1123,7 +1121,7 @@ mod tests {
assert_eq!(text(&response, "status"), "ok"); assert_eq!(text(&response, "status"), "ok");
assert_eq!( assert_eq!(
text(&response, "payloadSchema"), text(&response, "payloadSchema"),
"gamecult.model.transport_envelope.v3" "gamecult.model.transport_envelope.v4"
); );
assert_eq!(text(&response, "payloadEncoding"), "messagepack-base64"); assert_eq!(text(&response, "payloadEncoding"), "messagepack-base64");
assert_eq!(text(&response, "messageId"), envelope.request_id); assert_eq!(text(&response, "messageId"), envelope.request_id);
@ -1287,23 +1285,23 @@ mod tests {
#[test] #[test]
fn a_readout_request_is_not_an_invocation_and_names_its_own_operation() { fn a_readout_request_is_not_an_invocation_and_names_its_own_operation() {
let key = TransportKey::from_connection_secret("readout-test-key").unwrap(); let key = TransportKey::from_connection_secret("readout-test-key").unwrap();
let sealed = encrypt_spend_readout_request("caller-1", "readout-1", &key).unwrap(); let sealed = encrypt_usage_readout_request("caller-1", "readout-1", &key).unwrap();
let frame = encode_spend_readout_request_frame(&sealed).unwrap(); let frame = encode_usage_readout_request_frame(&sealed).unwrap();
assert!(matches!( assert!(matches!(
decode_request_frame(&frame), decode_request_frame(&frame),
Ok(RequestFrame::SpendReadout(found)) if found == sealed Ok(RequestFrame::UsageReadout(found)) if found == sealed
)); ));
assert_eq!(decode_invocation_frame(&frame), Err(ServiceError::Carriage)); assert_eq!(decode_invocation_frame(&frame), Err(ServiceError::Carriage));
// Sealed as one kind, the authenticated data refuses it as another. // Sealed as one kind, the authenticated data refuses it as another.
assert!(decrypt_spend_readout_request(&sealed, &key).is_ok()); assert!(decrypt_usage_readout_request(&sealed, &key).is_ok());
assert_eq!( assert_eq!(
decrypt_spend_readout(&sealed, &key).unwrap_err(), decrypt_usage_readout(&sealed, &key).unwrap_err(),
ServiceError::Envelope ServiceError::Envelope
); );
let mut relabelled = sealed.clone(); let mut relabelled = sealed.clone();
relabelled.message_kind = EnvelopeKind::SpendReadoutResult; relabelled.message_kind = EnvelopeKind::UsageReadoutResult;
assert_eq!( assert_eq!(
decrypt_spend_readout(&relabelled, &key).unwrap_err(), decrypt_usage_readout(&relabelled, &key).unwrap_err(),
ServiceError::Encryption ServiceError::Encryption
); );
} }
@ -1313,12 +1311,12 @@ mod tests {
fn a_response_is_read_as_its_own_operation_only() { fn a_response_is_read_as_its_own_operation_only() {
use crate::readout::PlanQuotaSnapshot; use crate::readout::PlanQuotaSnapshot;
let key = TransportKey::from_connection_secret("readout-test-key").unwrap(); let key = TransportKey::from_connection_secret("readout-test-key").unwrap();
let readout = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default()); let readout = UsageReadout::new(1, PlanQuotaSnapshot::default());
let sealed = encrypt_spend_readout("caller-1", "readout-1", &readout, &key).unwrap(); let sealed = encrypt_usage_readout("caller-1", "readout-1", &readout, &key).unwrap();
let readout_frame = encode_spend_readout_response_frame(&sealed).unwrap(); let readout_frame = encode_usage_readout_response_frame(&sealed).unwrap();
let invoke_frame = encode_result_frame(&sealed).unwrap(); let invoke_frame = encode_result_frame(&sealed).unwrap();
assert_eq!( assert_eq!(
decode_spend_readout_response_frame(&readout_frame).unwrap(), decode_usage_readout_response_frame(&readout_frame).unwrap(),
sealed sealed
); );
assert_eq!(decode_result_frame(&invoke_frame).unwrap(), sealed); assert_eq!(decode_result_frame(&invoke_frame).unwrap(), sealed);
@ -1327,9 +1325,9 @@ mod tests {
Err(ServiceError::Carriage) Err(ServiceError::Carriage)
); );
assert_eq!( assert_eq!(
decode_spend_readout_response_frame(&invoke_frame), decode_usage_readout_response_frame(&invoke_frame),
Err(ServiceError::Carriage) Err(ServiceError::Carriage)
); );
assert_eq!(decrypt_spend_readout(&sealed, &key).unwrap(), readout); assert_eq!(decrypt_usage_readout(&sealed, &key).unwrap(), readout);
} }
} }

43
crates/model-connector/src/fixtures.rs
View file

@ -133,16 +133,9 @@ pub(crate) fn request() -> ProviderRequest {
request 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 { pub(crate) fn invocation() -> TransportInvocation {
TransportInvocation::new("epiphany-yggdrasil", 2_000, [7; 32], MAX_COST, request()).unwrap() TransportInvocation::new("epiphany-yggdrasil", 2_000, [7; 32], request()).unwrap()
} }
pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt { pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt {
@ -155,8 +148,6 @@ pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt {
model: invocation.request.model.clone(), model: invocation.request.model.clone(),
provider: "fixture-sse".to_string(), provider: "fixture-sse".to_string(),
foreign_reasoning_dropped: 0, foreign_reasoning_dropped: 0,
cost_micro_usd: FIXTURE_USAGE_COST,
overrun_micro_usd: 0,
outcome: TransportOutcome::Completed { outcome: TransportOutcome::Completed {
provider_response_id: Some("response-1".to_string()), provider_response_id: Some("response-1".to_string()),
stop_reason: StopReason::EndTurn, stop_reason: StopReason::EndTurn,
@ -170,33 +161,6 @@ 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")] #[cfg(feature = "daemon")]
pub(crate) fn caller( pub(crate) fn caller(
@ -212,8 +176,6 @@ pub(crate) fn caller(
max_concurrent_requests, max_concurrent_requests,
64 * 1024, 64 * 1024,
32_768, 32_768,
SpendCeiling(CEILING),
SpendWindowMs(WINDOW_MS),
) )
.unwrap() .unwrap()
} }
@ -223,7 +185,6 @@ pub(crate) fn open_service(store: &TestStore, max_concurrent_requests: usize) ->
TransportService::open( TransportService::open(
&store.path, &store.path,
"fixture-sse", "fixture-sse",
fixture_prices(),
[ [
caller( caller(
"epiphany-yggdrasil", "epiphany-yggdrasil",
@ -249,5 +210,5 @@ pub(crate) fn security(key: &str) -> TransportKey {
pub(crate) fn invocation_for(caller_runtime_id: &str, request_id: &str) -> TransportInvocation { pub(crate) fn invocation_for(caller_runtime_id: &str, request_id: &str) -> TransportInvocation {
let mut request = request(); let mut request = request();
request.request_id = request_id.to_string(); request.request_id = request_id.to_string();
TransportInvocation::new(caller_runtime_id, 2_000, [7; 32], MAX_COST, request).unwrap() TransportInvocation::new(caller_runtime_id, 2_000, [7; 32], request).unwrap()
} }

55
crates/model-connector/src/readout.rs
View file

@ -1,13 +1,14 @@
//! The spend readout: what each caller has spent and holds in its current //! The usage readout: the subscription plan's own usage windows as the backend
//! window, and the plan's own usage as the backend last saw it. The ledger is //! last saw them. The connector meters nothing in dollars; it publishes what
//! the replay store's; a readout only reads it. //! the provider reports, and the minds read it and decide their own burn.
use schemars::JsonSchema; use schemars::JsonSchema;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use crate::contract::ContractError; use crate::contract::ContractError;
pub const SPEND_READOUT_SCHEMA_ID: &str = "gamecult.model.spend_readout.v1"; pub const USAGE_READOUT_SCHEMA_ID: &str = "gamecult.model.usage_readout.v1";
/// One window of the subscription plan, as the provider reported it. /// One window of the subscription plan, as the provider reported it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
@ -32,40 +33,25 @@ pub struct PlanQuotaSnapshot {
pub windows: Vec<PlanQuota>, pub windows: Vec<PlanQuota>,
} }
/// One caller's current spend window in micro-USD. `spent` is settled charges
/// plus what a crash left counted at its full reservation; `reserved` is what
/// calls in flight hold, and both count against the ceiling.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct CallerSpend { pub struct UsageReadout {
pub caller_id: String,
pub window_ms: u64,
pub ceiling_micro_usd: u64,
pub spent_micro_usd: u64,
pub reserved_micro_usd: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct SpendReadout {
pub schema_id: String, pub schema_id: String,
pub at_unix_ms: u64, pub at_unix_ms: u64,
pub callers: Vec<CallerSpend>,
pub plan_quota: Vec<PlanQuota>, pub plan_quota: Vec<PlanQuota>,
} }
impl SpendReadout { impl UsageReadout {
pub fn new(at_unix_ms: u64, callers: Vec<CallerSpend>, plan: PlanQuotaSnapshot) -> Self { pub fn new(at_unix_ms: u64, plan: PlanQuotaSnapshot) -> Self {
Self { Self {
schema_id: SPEND_READOUT_SCHEMA_ID.to_string(), schema_id: USAGE_READOUT_SCHEMA_ID.to_string(),
at_unix_ms, at_unix_ms,
callers,
plan_quota: plan.windows, plan_quota: plan.windows,
} }
} }
pub(crate) fn validate(&self) -> Result<(), ContractError> { pub(crate) fn validate(&self) -> Result<(), ContractError> {
if self.schema_id != SPEND_READOUT_SCHEMA_ID { if self.schema_id != USAGE_READOUT_SCHEMA_ID {
return Err(ContractError::Schema); return Err(ContractError::Schema);
} }
Ok(()) Ok(())
@ -77,16 +63,9 @@ mod tests {
use super::*; use super::*;
use crate::contract::{wire_decode, wire_encode}; use crate::contract::{wire_decode, wire_encode};
fn readout() -> SpendReadout { fn readout() -> UsageReadout {
SpendReadout::new( UsageReadout::new(
1_000, 1_000,
vec![CallerSpend {
caller_id: "eureka".into(),
window_ms: 3_600_000,
ceiling_micro_usd: 50_000_000,
spent_micro_usd: 7,
reserved_micro_usd: 3,
}],
PlanQuotaSnapshot { PlanQuotaSnapshot {
windows: vec![PlanQuota { windows: vec![PlanQuota {
window: "five_hour".into(), window: "five_hour".into(),
@ -107,21 +86,21 @@ mod tests {
panic!("a readout is a map, not a positional array"); panic!("a readout is a map, not a positional array");
}; };
let names: Vec<&str> = fields.iter().filter_map(|(k, _)| k.as_str()).collect(); let names: Vec<&str> = fields.iter().filter_map(|(k, _)| k.as_str()).collect();
assert_eq!(names, ["schema_id", "at_unix_ms", "callers", "plan_quota"]); assert_eq!(names, ["schema_id", "at_unix_ms", "plan_quota"]);
assert_eq!(wire_decode::<SpendReadout>(&bytes).unwrap(), readout()); assert_eq!(wire_decode::<UsageReadout>(&bytes).unwrap(), readout());
} }
#[test] #[test]
fn an_unobserved_plan_is_none_not_zero() { fn an_unobserved_plan_is_none_not_zero() {
let bare = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default()); let bare = UsageReadout::new(1, PlanQuotaSnapshot::default());
let decoded: SpendReadout = wire_decode(&wire_encode(&bare).unwrap()).unwrap(); let decoded: UsageReadout = wire_decode(&wire_encode(&bare).unwrap()).unwrap();
assert!(decoded.plan_quota.is_empty()); assert!(decoded.plan_quota.is_empty());
} }
#[test] #[test]
fn a_readout_of_another_schema_is_refused() { fn a_readout_of_another_schema_is_refused() {
let mut other = readout(); let mut other = readout();
other.schema_id = "gamecult.model.spend_readout.v2".into(); other.schema_id = "gamecult.model.usage_readout.v2".into();
assert!(other.validate().is_err()); assert!(other.validate().is_err());
assert!(readout().validate().is_ok()); assert!(readout().validate().is_ok());
} }

127
crates/model-connector/src/serve.rs
View file

@ -13,13 +13,12 @@ use zeroize::Zeroizing;
use crate::contract::TransportResult; use crate::contract::TransportResult;
use crate::envelope::{ use crate::envelope::{
RequestFrame, ServiceError, TransportFrameError, decode_request_frame, encode_result_frame, RequestFrame, ServiceError, TransportFrameError, decode_request_frame, encode_result_frame,
encode_spend_readout_response_frame, frame_bytes_for_payload, read_transport_frame, encode_usage_readout_response_frame, frame_bytes_for_payload, read_transport_frame,
write_transport_frame, write_transport_frame,
}; };
use crate::readout::PlanQuotaSnapshot; use crate::readout::PlanQuotaSnapshot;
use crate::service::{ use crate::service::{
CallerAdmission, PriceTable, ProviderCall, SpendCeiling, SpendWindowMs, TransportAdmission, CallerAdmission, ProviderCall, TransportAdmission, TransportService, distinct_admissions,
TransportService, distinct_admissions,
}; };
/// The only provider seam of the connector kit. A connector binary implements /// The only provider seam of the connector kit. A connector binary implements
@ -34,8 +33,9 @@ pub trait ProviderBackend: Send + Sync {
/// `serve` on: it names the provider on every receipt and decides which /// `serve` on: it names the provider on every receipt and decides which
/// reasoning state is foreign. /// reasoning state is foreign.
fn provider(&self) -> &str; fn provider(&self) -> &str;
/// What the provider's own plan allows and has used, for the spend /// What the provider's own plan allows and has used, for the usage
/// readout. A provider with no plan to report keeps the default. /// readout and the provider's own refusals. A provider with no plan to report
/// keeps the default.
fn plan_quota(&self) -> PlanQuotaSnapshot { fn plan_quota(&self) -> PlanQuotaSnapshot {
PlanQuotaSnapshot::default() PlanQuotaSnapshot::default()
} }
@ -123,12 +123,6 @@ pub struct CallerConfig {
pub max_concurrent_requests: usize, pub max_concurrent_requests: usize,
pub max_payload_bytes: usize, pub max_payload_bytes: usize,
pub max_output_tokens: u32, 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> { pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>, ServeError> {
@ -149,8 +143,6 @@ pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>,
caller.max_concurrent_requests, caller.max_concurrent_requests,
caller.max_payload_bytes, caller.max_payload_bytes,
caller.max_output_tokens, caller.max_output_tokens,
SpendCeiling(caller.spend_ceiling_micro_usd),
SpendWindowMs(caller.spend_window_ms),
) )
.map_err(ServeError::Service)?, .map_err(ServeError::Service)?,
); );
@ -159,15 +151,12 @@ pub fn load_admissions(callers: &[CallerConfig]) -> Result<Vec<CallerAdmission>,
} }
/// Serves admitted callers on a loopback listener until the listener fails. /// 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 /// The caller binds; a listener that is not bound to loopback is refused, and
/// peers that are not loopback are dropped. /// peers that are not loopback are dropped.
pub fn serve( pub fn serve(
listener: TcpListener, listener: TcpListener,
limits: ServeLimits, limits: ServeLimits,
admissions: Vec<CallerAdmission>, admissions: Vec<CallerAdmission>,
prices: PriceTable,
replay_store: &Path, replay_store: &Path,
backend: Arc<dyn ProviderBackend>, backend: Arc<dyn ProviderBackend>,
) -> Result<(), ServeError> { ) -> Result<(), ServeError> {
@ -176,7 +165,6 @@ pub fn serve(
let service = Arc::new(Mutex::new(TransportService::open( let service = Arc::new(Mutex::new(TransportService::open(
replay_store, replay_store,
backend.provider(), backend.provider(),
prices,
admissions, admissions,
limits.max_expiry_skew_ms, limits.max_expiry_skew_ms,
)?)); )?));
@ -184,8 +172,8 @@ pub fn serve(
} }
/// `serve` over a service the caller opened and keeps a handle to, so another /// `serve` over a service the caller opened and keeps a handle to, so another
/// front of the same ledger (the Self-run credential lease) reserves in the /// front of the same service (the Self-run credential lease) shares its
/// same windows. The limits must already have been validated against the /// replay store and plan. The limits must already have been validated against the
/// service's admissions. /// service's admissions.
pub fn serve_service( pub fn serve_service(
listener: TcpListener, listener: TcpListener,
@ -197,7 +185,7 @@ pub fn serve_service(
accept_loop(|| listener.accept(), limits, service, backend) accept_loop(|| listener.accept(), limits, service, backend)
} }
/// What must be true before a serve loop touches the ledger: the listener is /// What must be true before a serve loop touches the store: the listener is
/// on loopback and the backend can take work. /// on loopback and the backend can take work.
fn admit_listener(listener: &TcpListener, backend: &dyn ProviderBackend) -> Result<(), ServeError> { fn admit_listener(listener: &TcpListener, backend: &dyn ProviderBackend) -> Result<(), ServeError> {
if !listener if !listener
@ -272,20 +260,25 @@ fn answer_frame(
) -> Result<Vec<u8>, ServeError> { ) -> Result<Vec<u8>, ServeError> {
let envelope = match decode_request_frame(request).map_err(|_| ServeError::FrameEncoding)? { let envelope = match decode_request_frame(request).map_err(|_| ServeError::FrameEncoding)? {
RequestFrame::Invoke(envelope) => envelope, RequestFrame::Invoke(envelope) => envelope,
RequestFrame::SpendReadout(envelope) => { RequestFrame::UsageReadout(envelope) => {
let plan = backend.plan_quota(); let plan = backend.plan_quota();
let response = service let response = service
.lock() .lock()
.map_err(|_| ServeError::ServicePoisoned)? .map_err(|_| ServeError::ServicePoisoned)?
.spend_readout(&envelope, unix_ms()?, plan, limits.max_frame_bytes)?; .usage_readout(&envelope, unix_ms()?, plan, limits.max_frame_bytes)?;
return encode_spend_readout_response_frame(&response) return encode_usage_readout_response_frame(&response)
.map_err(|_| ServeError::FrameEncoding); .map_err(|_| ServeError::FrameEncoding);
} }
}; };
let admission = service let admission = service
.lock() .lock()
.map_err(|_| ServeError::ServicePoisoned)? .map_err(|_| ServeError::ServicePoisoned)?
.begin(&envelope, unix_ms()?, limits.max_frame_bytes)?; .begin(
&envelope,
unix_ms()?,
&backend.plan_quota(),
limits.max_frame_bytes,
)?;
let response = match admission { let response = match admission {
TransportAdmission::Reply(response) => response, TransportAdmission::Reply(response) => response,
TransportAdmission::Execute(claim) => { TransportAdmission::Execute(claim) => {
@ -390,7 +383,7 @@ mod tests {
TransportInvocation, TransportOutcome, TransportResult, TransportInvocation, TransportOutcome, TransportResult,
}; };
use crate::fixtures::{ use crate::fixtures::{
CEILING, MAX_COST, TestStore, WINDOW_MS, caller, fixture_prices, invocation_for, receipt, TestStore, caller, invocation_for, receipt,
request, request,
}; };
@ -472,7 +465,6 @@ mod tests {
listener, listener,
limits, limits,
admissions, admissions,
fixture_prices(),
&path, &path,
backend, backend,
)); ));
@ -506,7 +498,6 @@ mod tests {
listener, listener,
limits(), limits(),
admissions, admissions,
fixture_prices(),
&path, &path,
backend, backend,
) )
@ -528,8 +519,6 @@ mod tests {
max_concurrent_requests: *capacity, max_concurrent_requests: *capacity,
max_payload_bytes: 1_048_576, max_payload_bytes: 1_048_576,
max_output_tokens: 16_384, max_output_tokens: 16_384,
spend_ceiling_micro_usd: CEILING,
spend_window_ms: WINDOW_MS,
}) })
.collect() .collect()
} }
@ -559,7 +548,7 @@ mod tests {
} }
#[test] #[test]
fn serve_reads_out_spend_and_plan_quota_without_executing_anything() { fn serve_reads_out_plan_usage_without_executing_anything() {
let store = TestStore::new(); let store = TestStore::new();
let inner = CountingBackend::new(true); let inner = CountingBackend::new(true);
let plan = PlanQuotaSnapshot { let plan = PlanQuotaSnapshot {
@ -580,16 +569,9 @@ mod tests {
assert_eq!(inner.executions.load(Ordering::SeqCst), 1); assert_eq!(inner.executions.load(Ordering::SeqCst), 1);
let before = now_ms(); let before = now_ms();
let readout = client.spend_readout(CALLER).unwrap(); let readout = client.usage_readout(CALLER).unwrap();
assert!(readout.at_unix_ms >= before && readout.at_unix_ms <= now_ms()); assert!(readout.at_unix_ms >= before && readout.at_unix_ms <= now_ms());
assert_eq!(readout.plan_quota, plan.windows); assert_eq!(readout.plan_quota, plan.windows);
let [row] = readout.callers.as_slice() else {
panic!("one admitted caller");
};
assert_eq!(row.caller_id, CALLER);
assert_eq!((row.ceiling_micro_usd, row.window_ms), (CEILING, WINDOW_MS));
assert!(row.spent_micro_usd > 0, "the settled call was charged");
assert_eq!(row.reserved_micro_usd, 0);
assert_eq!( assert_eq!(
inner.executions.load(Ordering::SeqCst), inner.executions.load(Ordering::SeqCst),
1, 1,
@ -604,39 +586,62 @@ mod tests {
Some(Duration::from_secs(5)), Some(Duration::from_secs(5)),
) )
.unwrap(); .unwrap();
assert!(stranger.spend_readout(CALLER).is_err()); assert!(stranger.usage_readout(CALLER).is_err());
} }
#[test] #[test]
fn a_backend_with_no_plan_reads_out_none() { fn a_backend_with_no_plan_reads_out_none() {
let store = TestStore::new(); let store = TestStore::new();
let (client, _) = start(CountingBackend::new(true), &store); let (client, _) = start(CountingBackend::new(true), &store);
let readout = client.spend_readout(CALLER).unwrap(); let readout = client.usage_readout(CALLER).unwrap();
assert!(readout.plan_quota.is_empty()); assert!(readout.plan_quota.is_empty());
} }
fn window_plan(permille: u32, status: &str, resets_at_unix_ms: u64) -> PlanQuotaSnapshot {
PlanQuotaSnapshot {
windows: vec![crate::readout::PlanQuota {
window: "five_hour".into(),
utilization_permille: Some(permille),
resets_at_unix_ms,
status: status.into(),
observed_at_unix_ms: 1,
}],
}
}
#[test] #[test]
fn serve_refuses_a_request_over_its_max_cost_without_executing_it() { fn serve_repeats_the_providers_rejection_and_executes_nothing_while_it_stands() {
let store = TestStore::new(); let store = TestStore::new();
let backend = CountingBackend::new(true); let inner = CountingBackend::new(true);
let (client, _) = start(backend.clone(), &store); let backend = Arc::new(PlanBackend {
let mut invocation = fresh("request-costly"); inner: inner.clone(),
invocation.max_cost_micro_usd = 1; plan: window_plan(1_000, "rejected", now_ms() + 3_600_000),
let refused = client.execute(&invocation).unwrap(); });
let (client, _) = start_with(backend, &store, vec![caller(CALLER, KEY, 4)]);
let refused = client.execute(&fresh("request-rejected")).unwrap();
assert_eq!( assert_eq!(
refused.disposition, refused.disposition,
TransportDisposition::Refused(Refusal::CostBound) TransportDisposition::Refused(Refusal::ProviderRejected)
); );
assert_eq!(backend.executions.load(Ordering::SeqCst), 0); assert_eq!(inner.executions.load(Ordering::SeqCst), 0);
}
// The refusal was not recorded: the same id with a budget that holds #[test]
// the worst case is admitted and executed. fn serve_admits_a_full_window_the_provider_still_allows() {
invocation.max_cost_micro_usd = MAX_COST; let store = TestStore::new();
let admitted = client.execute(&invocation).unwrap(); let inner = CountingBackend::new(true);
let backend = Arc::new(PlanBackend {
inner: inner.clone(),
plan: window_plan(999, "allowed", now_ms() + 3_600_000),
});
let (client, _) = start_with(backend, &store, vec![caller(CALLER, KEY, 4)]);
let admitted = client.execute(&fresh("request-full")).unwrap();
assert!(matches!( assert!(matches!(
admitted.disposition, admitted.disposition,
TransportDisposition::Transported { .. } TransportDisposition::Transported { .. }
)); ));
assert_eq!(inner.executions.load(Ordering::SeqCst), 1);
}
assert_eq!(backend.executions.load(Ordering::SeqCst), 1); assert_eq!(backend.executions.load(Ordering::SeqCst), 1);
} }
@ -676,7 +681,6 @@ mod tests {
CALLER, CALLER,
invocation.expires_at_unix_ms, invocation.expires_at_unix_ms,
[7; 32], [7; 32],
MAX_COST,
conflicting_request, conflicting_request,
) )
.unwrap(); .unwrap();
@ -761,7 +765,6 @@ mod tests {
let mut service = TransportService::open( let mut service = TransportService::open(
&store.path, &store.path,
"fixture-sse", "fixture-sse",
fixture_prices(),
vec![caller(CALLER, KEY, 4)], vec![caller(CALLER, KEY, 4)],
300_000, 300_000,
) )
@ -770,7 +773,9 @@ mod tests {
crate::envelope::encrypt_invocation(&invocation, &crate::fixtures::security(KEY)) crate::envelope::encrypt_invocation(&invocation, &crate::fixtures::security(KEY))
.unwrap(); .unwrap();
let TransportAdmission::Execute(claim) = let TransportAdmission::Execute(claim) =
service.begin(&envelope, now_ms(), 1 << 20).unwrap() service
.begin(&envelope, now_ms(), &PlanQuotaSnapshot::default(), 1 << 20)
.unwrap()
else { else {
panic!("a fresh request must execute"); panic!("a fresh request must execute");
}; };
@ -793,7 +798,6 @@ mod tests {
listener, listener,
narrowed, narrowed,
vec![caller(CALLER, KEY, 4)], vec![caller(CALLER, KEY, 4)],
fixture_prices(),
&path, &path,
served, served,
) )
@ -1001,8 +1005,6 @@ mod tests {
slots, slots,
payload, payload,
32_768, 32_768,
SpendCeiling(CEILING),
SpendWindowMs(WINDOW_MS),
) )
.unwrap() .unwrap()
} }
@ -1110,7 +1112,7 @@ mod tests {
..limits() ..limits()
}; };
let admissions = vec![admission(CALLER, KEY, 4, bound)]; let admissions = vec![admission(CALLER, KEY, 4, bound)];
thread::spawn(move || serve(listener, tight, admissions, fixture_prices(), &path, served)); thread::spawn(move || serve(listener, tight, admissions, &path, served));
let client = let client =
ConnectorClient::new(endpoint, KEY, frame, Some(Duration::from_secs(5))).unwrap(); ConnectorClient::new(endpoint, KEY, frame, Some(Duration::from_secs(5))).unwrap();
let reply = client.execute(&invocation).unwrap(); let reply = client.execute(&invocation).unwrap();
@ -1225,7 +1227,6 @@ mod tests {
let mut service = TransportService::open( let mut service = TransportService::open(
&store.path, &store.path,
"fixture-sse", "fixture-sse",
fixture_prices(),
vec![caller(CALLER, KEY, 4)], vec![caller(CALLER, KEY, 4)],
300_000, 300_000,
) )
@ -1236,7 +1237,9 @@ mod tests {
) )
.unwrap(); .unwrap();
let TransportAdmission::Execute(claim) = let TransportAdmission::Execute(claim) =
service.begin(&envelope, now_ms(), 1 << 20).unwrap() service
.begin(&envelope, now_ms(), &PlanQuotaSnapshot::default(), 1 << 20)
.unwrap()
else { else {
panic!("a fresh request must execute"); panic!("a fresh request must execute");
}; };
@ -1328,7 +1331,6 @@ mod tests {
TransportService::open( TransportService::open(
&store.path, &store.path,
"fixture-sse", "fixture-sse",
fixture_prices(),
vec![caller(CALLER, KEY, 4)], vec![caller(CALLER, KEY, 4)],
300_000, 300_000,
) )
@ -1391,7 +1393,6 @@ mod tests {
listener, listener,
limits(), limits(),
admissions, admissions,
fixture_prices(),
&path, &path,
served, served,
) )

2340
crates/model-connector/src/service.rs
View file

File diff suppressed because it is too large Load diff

880
crates/model-connector/src/service/lease.rs
View file

File diff suppressed because it is too large Load diff

8
crates/model-connector/tests/fixtures/wire_v3/README.md → crates/model-connector/tests/fixtures/wire_v4/README.md vendored
View file

@ -1,6 +1,6 @@
# wire_v3 fixtures # wire_v4 fixtures
The named-MessagePack bytes of the v3 documents and one sealed carriage frame, The named-MessagePack bytes of the v4 documents and one sealed carriage frame,
produced by the kit at the revision that added them. They pin drift; the types in produced by the kit at the revision that added them. They pin drift; the types in
`src/contract.rs` own the shape, and `schemas/` publishes it. `src/contract.rs` own the shape, and `schemas/` publishes it.
@ -10,7 +10,7 @@ produced by the kit at the revision that added them. They pin drift; the types i
the plaintext documents. the plaintext documents.
- `carriage_invocation.frame`: a `cultnet.operation_request.v0` for service - `carriage_invocation.frame`: a `cultnet.operation_request.v0` for service
`gamecult.model.connector`, operation `invoke`, whose payload is the base64 of an envelope `gamecult.model.connector`, operation `invoke`, whose payload is the base64 of an envelope
sealed under the fixed test key in `tests/wire_golden_v3.rs`. sealed under the fixed test key in `tests/wire_golden_v4.rs`.
Generator: `tests/wire_golden_v3.rs` itself. Run it with `WIRE_V3_REGENERATE=1` to rewrite Generator: `tests/wire_golden_v4.rs` itself. Run it with `WIRE_V4_REGENERATE=1` to rewrite
the files from the kit, then review the diff: a changed byte is a wire change. the files from the kit, then review the diff: a changed byte is a wire change.

0
crates/model-connector/tests/fixtures/wire_v3/carriage_invocation.frame → crates/model-connector/tests/fixtures/wire_v4/carriage_invocation.frame vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/input_reasoning_state.mp → crates/model-connector/tests/fixtures/wire_v4/input_reasoning_state.mp vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/provider_request.mp → crates/model-connector/tests/fixtures/wire_v4/provider_request.mp vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/transport_envelope.mp → crates/model-connector/tests/fixtures/wire_v4/transport_envelope.mp vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/transport_invocation.mp → crates/model-connector/tests/fixtures/wire_v4/transport_invocation.mp vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp → crates/model-connector/tests/fixtures/wire_v4/transport_receipt_failed.mp vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/transport_result_refused.mp → crates/model-connector/tests/fixtures/wire_v4/transport_result_refused.mp vendored
View file

0
crates/model-connector/tests/fixtures/wire_v3/transport_result_transported.mp → crates/model-connector/tests/fixtures/wire_v4/transport_result_transported.mp vendored
View file

12
crates/model-connector/tests/published_schemas.rs
View file

@ -1,4 +1,4 @@
//! `schemas/` at the repository root publishes one JSON Schema per v3 document, //! `schemas/` at the repository root publishes one JSON Schema per v4 document,
//! generated from the types. This test regenerates them and fails on any //! generated from the types. This test regenerates them and fails on any
//! difference; run it with `MODEL_CONNECTOR_REGENERATE_SCHEMAS=1` to rewrite the //! difference; run it with `MODEL_CONNECTOR_REGENERATE_SCHEMAS=1` to rewrite the
//! files from the types. //! files from the types.
@ -28,7 +28,7 @@ fn published_schemas_are_the_types() {
("transport_result", render::<TransportResult>()), ("transport_result", render::<TransportResult>()),
("transport_receipt", render::<TransportReceipt>()), ("transport_receipt", render::<TransportReceipt>()),
("transport_envelope", render::<TransportEnvelope>()), ("transport_envelope", render::<TransportEnvelope>()),
("spend_readout", render::<SpendReadout>()), ("usage_readout", render::<UsageReadout>()),
]; ];
let dir = schemas_dir(); let dir = schemas_dir();
if std::env::var_os(REGENERATE).is_some() { if std::env::var_os(REGENERATE).is_some() {
@ -80,7 +80,7 @@ fn the_published_ciphertext_is_what_the_wire_sends() {
assert_eq!(&published["properties"]["ciphertext"], ciphertext); assert_eq!(&published["properties"]["ciphertext"], ciphertext);
let envelope = TransportEnvelope { let envelope = TransportEnvelope {
schema_id: "gamecult.model.transport_envelope.v3".to_string(), schema_id: "gamecult.model.transport_envelope.v4".to_string(),
caller_runtime_id: "caller".to_string(), caller_runtime_id: "caller".to_string(),
request_id: "request".to_string(), request_id: "request".to_string(),
message_kind: EnvelopeKind::Invocation, message_kind: EnvelopeKind::Invocation,
@ -111,9 +111,7 @@ fn the_refusal_schema_is_one_flat_string_enum() {
.iter() .iter()
.map(|value| value.as_str().unwrap()) .map(|value| value.as_str().unwrap())
.collect(); .collect();
assert_eq!(variants.len(), 15); assert_eq!(variants.len(), 13);
assert!(variants.contains(&"ClockBehindAdmissionWindow"));
assert!(variants.contains(&"ForeignReasoningState")); assert!(variants.contains(&"ForeignReasoningState"));
assert!(variants.contains(&"CostBound")); assert!(variants.contains(&"ProviderRejected"));
assert!(variants.contains(&"SpendCeiling"));
} }

22
crates/model-connector/tests/reasoning_state.rs
View file

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

18
crates/model-connector/tests/wire_golden_v3.rs → crates/model-connector/tests/wire_golden_v4.rs
View file

@ -1,5 +1,5 @@
//! Pins the bytes of the v3 wire so a drift is a failing test. The fixtures in //! Pins the bytes of the v4 wire so a drift is a failing test. The fixtures in
//! `fixtures/wire_v3` are produced by this file (see its README); they pin the //! `fixtures/wire_v4` are produced by this file (see its README); they pin the
//! kit against itself and are not a second contract: the types in //! kit against itself and are not a second contract: the types in
//! `model_connector::contract` own the shape. //! `model_connector::contract` own the shape.
@ -17,10 +17,10 @@ use serde::de::DeserializeOwned;
const KEY: &str = "golden-distinct-test-key"; const KEY: &str = "golden-distinct-test-key";
const CALLER: &str = "golden-caller"; const CALLER: &str = "golden-caller";
const REGENERATE: &str = "WIRE_V3_REGENERATE"; const REGENERATE: &str = "WIRE_V4_REGENERATE";
fn fixtures() -> PathBuf { fn fixtures() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/wire_v3") Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/wire_v4")
} }
fn named<T: Serialize>(value: &T) -> Vec<u8> { fn named<T: Serialize>(value: &T) -> Vec<u8> {
@ -28,7 +28,7 @@ fn named<T: Serialize>(value: &T) -> Vec<u8> {
} }
/// The value serializes to the committed bytes and the committed bytes decode /// The value serializes to the committed bytes and the committed bytes decode
/// back to it. With `WIRE_V3_REGENERATE` set, the committed bytes are rewritten /// back to it. With `WIRE_V4_REGENERATE` set, the committed bytes are rewritten
/// from the value first. /// from the value first.
fn assert_wire<T: Serialize + DeserializeOwned + PartialEq + std::fmt::Debug>( fn assert_wire<T: Serialize + DeserializeOwned + PartialEq + std::fmt::Debug>(
name: &str, name: &str,
@ -81,7 +81,7 @@ fn request() -> ProviderRequest {
} }
fn invocation() -> TransportInvocation { fn invocation() -> TransportInvocation {
TransportInvocation::new(CALLER, 1_800_000_000_000, [7; 32], 5_000_000, request()).unwrap() TransportInvocation::new(CALLER, 1_800_000_000_000, [7; 32], request()).unwrap()
} }
fn receipt(outcome: TransportOutcome) -> TransportReceipt { fn receipt(outcome: TransportOutcome) -> TransportReceipt {
@ -95,8 +95,6 @@ fn receipt(outcome: TransportOutcome) -> TransportReceipt {
model: invocation.request.model.clone(), model: invocation.request.model.clone(),
provider: "anthropic-messages".to_string(), provider: "anthropic-messages".to_string(),
foreign_reasoning_dropped: 2, foreign_reasoning_dropped: 2,
cost_micro_usd: 1_234_567,
overrun_micro_usd: 89,
outcome, outcome,
} }
} }
@ -116,7 +114,7 @@ fn completed() -> TransportOutcome {
} }
#[test] #[test]
fn wire_golden_v3() { fn wire_golden_v4() {
let invocation = invocation(); let invocation = invocation();
assert_wire("provider_request.mp", &invocation.request); assert_wire("provider_request.mp", &invocation.request);
assert_wire("transport_invocation.mp", &invocation); assert_wire("transport_invocation.mp", &invocation);
@ -233,7 +231,7 @@ fn the_golden_carriage_frame_decrypts_and_validates_under_the_fixed_key() {
}; };
assert_eq!(service_id, "gamecult.model.connector"); assert_eq!(service_id, "gamecult.model.connector");
assert_eq!(operation, "invoke"); assert_eq!(operation, "invoke");
assert_eq!(payload_schema, "gamecult.model.transport_envelope.v3"); assert_eq!(payload_schema, "gamecult.model.transport_envelope.v4");
assert_eq!(message_id, invocation.request_id()); assert_eq!(message_id, invocation.request_id());
let envelope: TransportEnvelope = let envelope: TransportEnvelope =
rmp_serde::from_slice(&STANDARD.decode(payload).unwrap()).unwrap(); rmp_serde::from_slice(&STANDARD.decode(payload).unwrap()).unwrap();

112
schemas/spend_readout.json
View file

@ -1,112 +0,0 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "SpendReadout",
"type": "object",
"properties": {
"at_unix_ms": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"callers": {
"type": "array",
"items": {
"$ref": "#/$defs/CallerSpend"
}
},
"plan_quota": {
"type": "array",
"items": {
"$ref": "#/$defs/PlanQuota"
}
},
"schema_id": {
"type": "string"
}
},
"additionalProperties": false,
"required": [
"schema_id",
"at_unix_ms",
"callers",
"plan_quota"
],
"$defs": {
"CallerSpend": {
"description": "One caller's current spend window in micro-USD. `spent` is settled charges\nplus what a crash left counted at its full reservation; `reserved` is what\ncalls in flight hold, and both count against the ceiling.",
"type": "object",
"properties": {
"caller_id": {
"type": "string"
},
"ceiling_micro_usd": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"reserved_micro_usd": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"spent_micro_usd": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"window_ms": {
"type": "integer",
"format": "uint64",
"minimum": 0
}
},
"additionalProperties": false,
"required": [
"caller_id",
"window_ms",
"ceiling_micro_usd",
"spent_micro_usd",
"reserved_micro_usd"
]
},
"PlanQuota": {
"description": "One window of the subscription plan, as the provider reported it.",
"type": "object",
"properties": {
"observed_at_unix_ms": {
"description": "When this window was last reported. Its age in a readout is the\nreadout's `at_unix_ms` minus this.",
"type": "integer",
"format": "uint64",
"minimum": 0
},
"resets_at_unix_ms": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"status": {
"type": "string"
},
"utilization_permille": {
"description": "How much of the window was used when it was observed, in thousandths;\n`None` once the window has reset (`status` is then `expired`), because\nthe old reading says nothing about the new window.",
"type": [
"integer",
"null"
],
"format": "uint32",
"minimum": 0
},
"window": {
"type": "string"
}
},
"additionalProperties": false,
"required": [
"window",
"resets_at_unix_ms",
"status",
"observed_at_unix_ms"
]
}
}
}

4
schemas/transport_envelope.json
View file

@ -47,8 +47,8 @@
"enum": [ "enum": [
"Invocation", "Invocation",
"Result", "Result",
"SpendReadoutRequest", "UsageReadoutRequest",
"SpendReadoutResult" "UsageReadoutResult"
] ]
} }
} }