connector-usage-only: store-failure tests, README front door

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 04:48:24 +01:00
commit dcb5f109f7
4 changed files with 62 additions and 26 deletions

View file

@ -4,6 +4,11 @@ The connector kit for GameCult minds: one typed, encrypted, replay-safe contract
between a caller that needs a model turn and a daemon that holds the provider between a caller that needs a model turn and a daemon that holds the provider
credential, so no project wires its own provider calls. credential, so no project wires its own provider calls.
The connector meters nothing in dollars. It holds the subscription token, lends it, and
publishes the plan's usage windows (`usage_readout`, and `plan_quota` on every lease grant) for the
minds to read and decide their own burn. The one usage-derived refusal is the provider's own
(`ProviderRejected`), repeated until its reset.
Status: the `model-connector` crate and the `claude-connector` backend crate are live. No Status: the `model-connector` crate and the `claude-connector` backend crate are live. No
connector binary exists yet (the Claude connector binary is first, then an OpenAI-compatible one). connector binary exists yet (the Claude connector binary is first, then an OpenAI-compatible one).
@ -11,7 +16,7 @@ connector binary exists yet (the Claude connector binary is first, then an OpenA
Crate `crates/model-connector`: Crate `crates/model-connector`:
- `src/contract.rs` - the v3 wire contract (`gamecult.model.*.v3`): `ProviderRequest`, `TransportInvocation`, - `src/contract.rs` - the v4 wire contract (`gamecult.model.*.v4`): `ProviderRequest`, `TransportInvocation`,
`TransportResult`, `TransportReceipt`, the typed `StopReason`, `FailureKind` and `Refusal`, `TransportResult`, `TransportReceipt`, the typed `StopReason`, `FailureKind` and `Refusal`,
`InputItem::ReasoningState` (a provider's opaque reasoning block, replayed by callers `InputItem::ReasoningState` (a provider's opaque reasoning block, replayed by callers
verbatim via `TransportResult::transcript_items`; the service refuses another provider's verbatim via `TransportResult::transcript_items`; the service refuses another provider's
@ -23,10 +28,10 @@ Crate `crates/model-connector`:
`cultnet.operation_request.v0` / `operation_response.v0` for service `cultnet.operation_request.v0` / `operation_response.v0` for service
`gamecult.model.connector`, operation `invoke` (or `usage_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. `usage_readout` answers a request id and its payload the base64 of the sealed envelope. `usage_readout` answers a
`UsageReadout` (`gamecult.model.usage_readout.v1`): each caller's spent and reserved `UsageReadout` (`gamecult.model.usage_readout.v1`): the backend's plan quota windows, each with
micro-USD against its ceiling in its current window, and the backend's plan quota; utilization, reset, status and age;
`ConnectorClient::usage_readout` asks. `ConnectorClient::usage_readout` asks.
- `schemas/` - one JSON Schema per v3 document, generated from the types. - `schemas/` - one JSON Schema per v4 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
replay (a retried request is answered byte-identically and the provider runs once), replay (a retried request is answered byte-identically and the provider runs once),
@ -49,7 +54,7 @@ Crate `crates/claude-connector` (library only; the binary that wires it to `serv
group); `SessionStore` and `StderrSink` are the other ports. `ClaudeCodeConfig` is what the group); `SessionStore` and `StderrSink` are the other ports. `ClaudeCodeConfig` is what the
binary hands in; the credential is read from a systemd credential and reaches the child on a binary hands in; the credential is read from a systemd credential and reaches the child on a
pipe (`CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR`), never in an environment. `QuotaCell` keeps the pipe (`CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR`), never in an environment. `QuotaCell` keeps the
plan windows the CLI's `rate_limit_event`s report, for the spend readout. plan windows the CLI's `rate_limit_event`s report, for the usage readout and the lease grant.
Frame arithmetic is public in `src/envelope.rs`, measured on the carriage frame: `frame_bytes_for_payload(n)` is the frame a Frame arithmetic is public in `src/envelope.rs`, measured on the carriage frame: `frame_bytes_for_payload(n)` is the frame a
caller payload bound of `n` needs, `payload_bytes_for_frame(frame)` is the largest payload bound caller payload bound of `n` needs, `payload_bytes_for_frame(frame)` is the largest payload bound
@ -58,7 +63,7 @@ a frame allows, and `MAX_IDENTITY_BYTES` (256) bounds caller ids, request ids an
needed); the smallest legal frame is `frame_bytes_for_payload(MIN_PAYLOAD_BYTES)`. A request id past needed); the smallest legal frame is `frame_bytes_for_payload(MIN_PAYLOAD_BYTES)`. A request id past
the bound is refused on the wire as `Refusal::Malformed`, the reply echoing the id (the connection is dropped only if that reply cannot fit the frame); `ConnectorClient` refuses such an id locally with `RequestIdTooLong`. the bound is refused on the wire as `Refusal::Malformed`, the reply echoing the id (the connection is dropped only if that reply cannot fit the frame); `ConnectorClient` refuses such an id locally with `RequestIdTooLong`.
The default feature set compiles contract, envelope and client only. The kit speaks v3 The default feature set compiles contract, envelope and client only. The kit speaks v4
only: a v2 frame gets no answer. only: a v2 frame gets no answer.
## Verify ## Verify
@ -68,9 +73,9 @@ cargo test --locked --workspace
cargo test --locked --workspace --features model-connector/daemon cargo test --locked --workspace --features model-connector/daemon
``` ```
`crates/model-connector/tests/wire_golden_v3.rs` pins the v3 documents and one sealed carriage `crates/model-connector/tests/wire_golden_v4.rs` pins the v4 documents and one sealed carriage
frame to the fixtures in `crates/model-connector/tests/fixtures/wire_v3` (regenerate with frame to the fixtures in `crates/model-connector/tests/fixtures/wire_v4` (regenerate with
`WIRE_V3_REGENERATE=1`; see the fixtures' README), and `WIRE_V4_REGENERATE=1`; see the fixtures' README), and
`crates/model-connector/tests/published_schemas.rs` fails when `schemas/` differs from the types `crates/model-connector/tests/published_schemas.rs` fails when `schemas/` differs from the types
(`MODEL_CONNECTOR_REGENERATE_SCHEMAS=1` rewrites it). (`MODEL_CONNECTOR_REGENERATE_SCHEMAS=1` rewrites it).

View file

@ -1438,7 +1438,8 @@ fn usage_sums_model_usage_across_models() {
panic!("completed expected"); panic!("completed expected");
}; };
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 the usage alone: the CLI's own total never reaches it.
assert_eq!(parts(&cheap).1, parts(&dear).1);
// 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}}"#;

View file

@ -94,7 +94,7 @@ impl PeerUnits for FakeUnits {
struct Harness { struct Harness {
dir: PathBuf, dir: PathBuf,
socket: PathBuf, socket: PathBuf,
service: Arc<Mutex<TransportService>>, _service: Arc<Mutex<TransportService>>,
quota: Arc<QuotaCell>, quota: Arc<QuotaCell>,
tokens: Arc<FakeTokens>, tokens: Arc<FakeTokens>,
/// The fake behind the listener, when it has one. /// The fake behind the listener, when it has one.
@ -162,18 +162,6 @@ fn receive(stream: &UnixStream) -> (Vec<u8>, Vec<OwnedFd>) {
} }
} }
fn connect(harness: &Harness) -> UnixStream {
let stream = UnixStream::connect(&harness.socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
stream
}
fn send(stream: &mut UnixStream, message: &LeaseMessage) {
stream.write_all(&encode_frame(message)).unwrap();
}
fn request(ttl_ms: u64) -> LeaseMessage { fn request(ttl_ms: u64) -> LeaseMessage {
LeaseMessage::ClaudeLeaseRequest(LeaseRequest { LeaseMessage::ClaudeLeaseRequest(LeaseRequest {
expires_at_unix_ms: now_ms() + ttl_ms, expires_at_unix_ms: now_ms() + ttl_ms,
@ -182,7 +170,7 @@ fn request(ttl_ms: u64) -> LeaseMessage {
/// What one request to the listener came back with. /// What one request to the listener came back with.
struct Answer { struct Answer {
stream: UnixStream, _stream: UnixStream,
raw: Vec<u8>, raw: Vec<u8>,
message: Option<LeaseMessage>, message: Option<LeaseMessage>,
fds: Vec<OwnedFd>, fds: Vec<OwnedFd>,
@ -202,7 +190,7 @@ fn ask_at(socket: &std::path::Path, ttl_ms: u64) -> Answer {
let (raw, fds) = receive(&stream); let (raw, fds) = receive(&stream);
let message = decode_frame(&raw); let message = decode_frame(&raw);
Answer { Answer {
stream, _stream: stream,
raw, raw,
message, message,
fds, fds,
@ -316,7 +304,7 @@ fn start(setup: Setup) -> Harness {
let harness = Harness { let harness = Harness {
dir, dir,
socket, socket,
service, _service: service,
quota, quota,
tokens, tokens,
units: Some(units), units: Some(units),

View file

@ -2158,4 +2158,46 @@ mod tests {
Some(Operation::UsageReadout) Some(Operation::UsageReadout)
); );
} }
#[test]
fn a_store_that_cannot_be_written_executes_nothing_and_holds_no_slot() {
let store = FlakyStore::new(0);
let allowance = store.allowance();
let mut service =
TransportService::with_store(store, "fixture-sse", [caller(CALLER, KEY, 1)], 5_000)
.unwrap();
let invocation = invocation_for(CALLER, "request-1");
assert!(matches!(
service.begin_unplanned(&raw_envelope(&invocation), 1_000, FRAME),
Err(ServiceError::ReplayStore(_))
));
allowance.store(10, Ordering::SeqCst);
assert!(matches!(
service.begin_unplanned(&raw_envelope(&invocation), 1_000, FRAME),
Ok(TransportAdmission::Execute(_))
));
}
#[test]
fn a_failed_completion_write_answers_the_store_failure_and_frees_the_slot() {
let store = FlakyStore::new(1);
let allowance = store.allowance();
let mut service =
TransportService::with_store(store, "fixture-sse", [caller(CALLER, KEY, 1)], 5_000)
.unwrap();
let (claim, invocation) = claim_for(&mut service, "request-1");
let result = text_result(claim.call(), "done");
assert!(matches!(
service.complete(*claim, result, FRAME),
Err(ServiceError::ReplayStore(_))
));
// The slot is free (one concurrent request is admitted), and the retry
// of the unfinished id is told it is indeterminate, not in flight.
allowance.store(10, Ordering::SeqCst);
let again = service
.begin_unplanned(&raw_envelope(&invocation), 1_000, FRAME)
.unwrap();
assert_eq!(refusal_of(again, &invocation), Refusal::Indeterminate);
let _ = claim_for(&mut service, "request-2");
}
} }