From 6e87cf2e6c0109863899028dacea7fa3d2cd961f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 6 Nov 2026 22:55:06 +0100 Subject: [PATCH 1/5] lease socket: a usage readout from the one QuotaCell, no lease, no caller check, no token, no meter wake Co-Authored-By: Claude Sonnet 5.5 --- README.md | 2 +- crates/claude-connector/src/lease.rs | 79 +++++++++++++--- crates/claude-connector/src/lib.rs | 4 +- .../tests/fixtures/lease_v2/README.md | 4 + crates/claude-connector/tests/lease.rs | 93 ++++++++++++++++++- .../tests/lease_wire_golden.rs | 41 +++++++- 6 files changed, 203 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 74bd5b5..7ffd262 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ between a caller that needs a model turn and a daemon that holds the provider 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 +publishes the plan's usage windows (`usage_readout`, `claude_usage_readout` on the lease socket, 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. diff --git a/crates/claude-connector/src/lease.rs b/crates/claude-connector/src/lease.rs index 7e503aa..1749f6f 100644 --- a/crates/claude-connector/src/lease.rs +++ b/crates/claude-connector/src/lease.rs @@ -10,6 +10,10 @@ //! provider's own, repeated until its own reset. Nothing the run says about //! itself decides who it is or what the plan allows. //! +//! The same socket also answers a usage readout to any peer that can reach it +//! (the claude-lease group): no lease, no caller check, no token read and no +//! wake of the meter. Usage is state every Mind reads. +//! //! 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 //! a reply frame. The connector spawns nothing here: the Self run's `claude` @@ -27,7 +31,7 @@ use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use model_connector::{ - LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseTurn, PlanQuota, TransportService, + LeaseAdmission, LeaseGrant, LeaseRefusal, LeaseTurn, PlanQuota, TransportService, UsageReadout, }; use serde::{Deserialize, Serialize}; @@ -48,6 +52,10 @@ pub enum LeaseMessage { /// holds the token, its write end closed. ClaudeLeaseGrant(LeaseGranted), ClaudeLeaseRefusal(LeaseRefused), + /// Asks for the plan's usage windows. Opens no lease. + ClaudeUsageReadoutRequest(UsageReadoutRequest), + /// The answer to a readout request: the cell's windows as of now, no fd. + ClaudeUsageReadout(UsageReadout), } /// A request names only when the lease ends. Who asks is the connector's to know. @@ -56,6 +64,10 @@ pub struct LeaseRequest { pub expires_at_unix_ms: u64, } +/// A readout request names nothing: the answer is the same for every asker. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct UsageReadoutRequest {} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct LeaseGranted { pub lease_id: String, @@ -385,8 +397,11 @@ struct Lender { units: Arc, } -/// Serves leases on `listener` until it fails. Each connection is one lease: -/// request, then grant or refusal, then (after a grant) the hold. A caller is +/// Serves leases on `listener` until it fails. Each connection is one lease or +/// one usage readout. A lease: request, then grant or refusal, then (after a +/// grant) the hold. A readout: request, then one frame, then close; its reach +/// is the socket's group, and it reads the plan's windows without waking the +/// meter. A caller is /// the account whose uid the kernel reports for the connection, and only a /// process in that caller's unit may lease, once per unit invocation for a Self /// run and once per turn for a Persona; no @@ -421,21 +436,33 @@ pub fn serve_leases( let _permit = permit; // A connection that fails is dropped; the lease it may have // held is ended inside `lend`. - let _ = lender.lend(stream); + let _ = lender.serve_one(stream); })?; } } impl Lender { - fn lend(&self, mut stream: UnixStream) -> io::Result<()> { + fn serve_one(&self, mut stream: UnixStream) -> io::Result<()> { stream.set_write_timeout(Some(self.limits.socket_timeout))?; - let peer = peer_credentials(&stream)?; let request_deadline = Instant::now() + self.limits.socket_timeout; - let LeaseMessage::ClaudeLeaseRequest(request) = - read_message(&mut stream, request_deadline)? - else { - return Err(io::Error::from(io::ErrorKind::InvalidData)); - }; + match read_message(&mut stream, request_deadline)? { + LeaseMessage::ClaudeLeaseRequest(request) => { + let peer = peer_credentials(&stream)?; + self.lend(stream, peer, request) + } + LeaseMessage::ClaudeUsageReadoutRequest(_) => { + stream.write_all(&encode_frame(&readout(&self.quota, unix_ms()))) + } + _ => Err(io::Error::from(io::ErrorKind::InvalidData)), + } + } + + fn lend( + &self, + mut stream: UnixStream, + peer: PeerCredentials, + request: LeaseRequest, + ) -> io::Result<()> { let Some(caller) = self.callers.iter().find(|caller| caller.uid == peer.uid) else { return refuse( &mut stream, @@ -609,6 +636,12 @@ impl Lender { } } +/// The plan's windows as the cell holds them at `now_ms`. It takes the cell by +/// reference so it cannot note a request, and never waits for a first read. +fn readout(quota: &QuotaCell, now_ms: u64) -> LeaseMessage { + LeaseMessage::ClaudeUsageReadout(UsageReadout::new(now_ms, quota.snapshot())) +} + /// The longest a lease waits for the meter's first read of the week. A call /// takes a few seconds; the holder's own wait for the grant is far longer. const FIRST_READ_WAIT: Duration = Duration::from_secs(10); @@ -629,6 +662,30 @@ fn refuse(stream: &mut UnixStream, kind: LeaseRefusalKind, detail: &str) -> io:: mod tests { use super::*; + #[test] + fn a_readout_of_an_empty_metered_cell_does_not_wait() { + let quota = QuotaCell::new(); + quota.attach_meter(); + let started = Instant::now(); + let LeaseMessage::ClaudeUsageReadout(answer) = readout("a, 77) else { + panic!("a readout answers with a readout"); + }; + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(answer.plan_quota.is_empty()); + assert_eq!(answer.at_unix_ms, 77); + } + + #[test] + fn a_readout_is_no_sign_of_life() { + let quota = Arc::new(QuotaCell::new()); + let within = Duration::from_secs(15 * 60); + let now = unix_ms(); + let _ = readout("a, now); + assert!(!quota.wanted_at(now, within)); + let _asking = quota.asking(); + assert!(quota.wanted_at(now, within)); + } + /// A refusal frame whose body is exactly `body_bytes` long. fn frame_of(body_bytes: usize) -> Vec { let frame_with = |pad: usize| { diff --git a/crates/claude-connector/src/lib.rs b/crates/claude-connector/src/lib.rs index 671ee36..52f957f 100644 --- a/crates/claude-connector/src/lib.rs +++ b/crates/claude-connector/src/lib.rs @@ -29,8 +29,8 @@ pub use launch::ChildEnv; #[cfg(target_os = "linux")] pub use lease::{ LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused, - LeaseRequest, MAX_LEASE_FRAME_BYTES, PeerCredentials, PeerUnit, PeerUnits, decode_frame, - encode_frame, peer_credentials, serve_leases, user_uid, + LeaseRequest, MAX_LEASE_FRAME_BYTES, PeerCredentials, PeerUnit, PeerUnits, UsageReadoutRequest, + decode_frame, encode_frame, peer_credentials, serve_leases, user_uid, }; #[cfg(unix)] pub use process::StdProcessPort; diff --git a/crates/claude-connector/tests/fixtures/lease_v2/README.md b/crates/claude-connector/tests/fixtures/lease_v2/README.md index 3cc9766..4204337 100644 --- a/crates/claude-connector/tests/fixtures/lease_v2/README.md +++ b/crates/claude-connector/tests/fixtures/lease_v2/README.md @@ -11,6 +11,10 @@ big-endian length, then the named-MessagePack map `{kind, value}`). `seven_day` windows (utilization_permille, resets_at_unix_ms, status, observed_at_unix_ms). On the wire the grant travels with one SCM_RIGHTS fd; the file holds the frame only. +- `usage_readout_request.bin`: `claude_usage_readout_request`, an empty map; opens no lease. +- `usage_readout.bin`: `claude_usage_readout`, the kit's `UsageReadout` (`schema_id`, `at_unix_ms`, + `plan_quota` with `five_hour` and `seven_day` windows); no fd travels with it. + Huginn's lease client vendors these files from this repository and tests against the same bytes. Generator: `tests/lease_wire_golden.rs`; run it with `LEASE_V2_REGENERATE=1`, then review the diff: a changed byte is a wire change. diff --git a/crates/claude-connector/tests/lease.rs b/crates/claude-connector/tests/lease.rs index 77c8295..5115e8a 100644 --- a/crates/claude-connector/tests/lease.rs +++ b/crates/claude-connector/tests/lease.rs @@ -15,10 +15,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use claude_connector::{ LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused, - LeaseRequest, PeerUnit, PeerUnits, QuotaCell, TokenSource, decode_frame, encode_frame, - serve_leases, + LeaseRequest, PeerUnit, PeerUnits, QuotaCell, TokenSource, UsageReadoutRequest, decode_frame, + encode_frame, serve_leases, }; -use model_connector::{CallerAdmission, LeaseTurn, PlanQuota, TransportService}; +use model_connector::{CallerAdmission, LeaseTurn, PlanQuota, TransportService, UsageReadout}; use zeroize::Zeroizing; const CALLER: &str = "eureka"; @@ -649,3 +649,90 @@ fn a_request_trickled_past_the_socket_timeout_is_dropped_without_a_lease() { assert!(started.elapsed() < Duration::from_secs(6)); assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 0); } + +/// What a usage readout request came back with. +fn read_usage(harness: &Harness) -> Answer { + let mut stream = UnixStream::connect(&harness.socket).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + stream + .write_all(&encode_frame(&LeaseMessage::ClaudeUsageReadoutRequest( + UsageReadoutRequest {}, + ))) + .unwrap(); + let (raw, fds) = receive(&stream); + let message = decode_frame(&raw); + Answer { + _stream: stream, + raw, + message, + fds, + } +} + +fn usage_of(answer: &Answer) -> UsageReadout { + match &answer.message { + Some(LeaseMessage::ClaudeUsageReadout(readout)) => readout.clone(), + other => panic!("expected a usage readout, got {other:?}"), + } +} + +#[test] +fn a_readout_answers_the_cells_windows_and_lends_nothing() { + let harness = start(setup()); + let reset = now_ms() + 3_600_000; + harness.quota.observe([ + window("five_hour", 420, "allowed", reset), + window("seven_day", 999, "allowed_warning", reset + 86_400_000), + ]); + let before = now_ms(); + let answer = read_usage(&harness); + let readout = usage_of(&answer); + assert!(answer.fds.is_empty(), "a readout lends no descriptor"); + assert_eq!(readout.plan_quota, harness.quota.snapshot().windows); + assert_eq!(readout.plan_quota.len(), 2); + assert!(readout.at_unix_ms >= before && readout.at_unix_ms <= now_ms()); + assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 0); + // No lease was opened: the next request is granted. + assert!(granted_ok(&ask(&harness, 60_000))); +} + +#[test] +fn a_peer_outside_the_lease_callers_still_reads_usage() { + let harness = start(Setup { + uid: own_uid().wrapping_add(1), + ..setup() + }); + harness + .quota + .observe([window("five_hour", 420, "allowed", now_ms() + 3_600_000)]); + let refused = ask(&harness, 60_000); + assert_eq!(refusal_of(&refused), Some(LeaseRefusalKind::UnknownCaller)); + let readout = usage_of(&read_usage(&harness)); + assert_eq!(readout.plan_quota.len(), 1); + assert_eq!(readout.plan_quota[0].utilization_permille, Some(420)); + assert_eq!(harness.tokens.reads.load(Ordering::SeqCst), 0); +} + +#[test] +fn a_readout_beside_a_held_lease_leaves_it_held() { + let harness = start(setup()); + let held = ask(&harness, 60_000); + assert!(granted_ok(&held)); + assert_eq!(usage_of(&read_usage(&harness)).plan_quota.len(), 0); + harness.enter(2); + let second = ask(&harness, 60_000); + assert_eq!(refusal_of(&second), Some(LeaseRefusalKind::Busy)); + assert!(granted_ok(&held)); +} + +#[test] +fn a_readout_does_not_wait_for_the_meter() { + let harness = start(setup()); + harness.quota.attach_meter(); + let started = Instant::now(); + let readout = usage_of(&read_usage(&harness)); + assert!(started.elapsed() < Duration::from_secs(5)); + assert!(readout.plan_quota.is_empty()); +} diff --git a/crates/claude-connector/tests/lease_wire_golden.rs b/crates/claude-connector/tests/lease_wire_golden.rs index 7b9cea2..c713889 100644 --- a/crates/claude-connector/tests/lease_wire_golden.rs +++ b/crates/claude-connector/tests/lease_wire_golden.rs @@ -8,10 +8,10 @@ use std::path::{Path, PathBuf}; use claude_connector::{ - LeaseGranted, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseRequest, decode_frame, - encode_frame, + LeaseGranted, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseRequest, UsageReadoutRequest, + decode_frame, encode_frame, }; -use model_connector::PlanQuota; +use model_connector::{PlanQuota, PlanQuotaSnapshot, UsageReadout}; const REGENERATE: &str = "LEASE_V2_REGENERATE"; @@ -97,3 +97,38 @@ fn a_grant_with_both_usage_windows_is_pinned() { }), ); } + +#[test] +fn the_usage_readout_frames_are_pinned() { + let window = |name: &str, permille, status: &str, resets| PlanQuota { + window: name.to_string(), + utilization_permille: Some(permille), + resets_at_unix_ms: resets, + status: status.to_string(), + observed_at_unix_ms: 1_793_300_000_000, + }; + let request = LeaseMessage::ClaudeUsageReadoutRequest(UsageReadoutRequest {}); + let answer = LeaseMessage::ClaudeUsageReadout(UsageReadout::new( + 1_793_300_100_000, + PlanQuotaSnapshot { + windows: vec![ + window("five_hour", 420, "allowed", 1_793_307_000_000), + window("seven_day", 999, "allowed_warning", 1_793_800_000_000), + ], + }, + )); + assert_frame("usage_readout_request.bin", &request); + assert_frame("usage_readout.bin", &answer); + for (message, spelled) in [ + (&request, "claude_usage_readout_request"), + (&answer, "claude_usage_readout"), + ] { + let bytes = encode_frame(message); + let value: rmpv::Value = rmp_serde::from_slice(&bytes[4..]).unwrap(); + let rmpv::Value::Map(outer) = value else { + panic!("a frame is a map") + }; + assert_eq!(outer[0].0.as_str(), Some("kind")); + assert_eq!(outer[0].1.as_str(), Some(spelled)); + } +} From 8331859537341e8664691caad02a99fc94387a3f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 6 Nov 2026 22:58:15 +0100 Subject: [PATCH 2/5] lease_v2: pin the usage readout request and answer frames Co-Authored-By: Claude Sonnet 5.5 --- .../tests/fixtures/lease_v2/usage_readout.bin | Bin 0 -> 346 bytes .../fixtures/lease_v2/usage_readout_request.bin | Bin 0 -> 46 bytes 2 files changed, 0 insertions(+), 0 deletions(-) create mode 100644 crates/claude-connector/tests/fixtures/lease_v2/usage_readout.bin create mode 100644 crates/claude-connector/tests/fixtures/lease_v2/usage_readout_request.bin diff --git a/crates/claude-connector/tests/fixtures/lease_v2/usage_readout.bin b/crates/claude-connector/tests/fixtures/lease_v2/usage_readout.bin new file mode 100644 index 0000000000000000000000000000000000000000..3fe12c81fddd65bf7e8c8502634c971d35c68414 GIT binary patch literal 346 zcmaLQJx&8L5I|usxz9@UoP%~WtcT5ru^l`fFHnLgXgL8$VNuc`M0&UYio_*202hEF zLgKf7dY^O6?&w2#1ymUY?J`hn0rKK39Tivb?gw!Y5cbMVL_zEJDxC{63RX zQTRR#J5Qq)ET*?8k_Pe*X-_Fv@;}lAW2CvL`twu{0$$zO*v;;^OmZlb$EG Date: Fri, 6 Nov 2026 22:59:12 +0100 Subject: [PATCH 3/5] lease readout tests: drop the integration test that needs a crate-private meter hook (the unit test owns it) Co-Authored-By: Claude Sonnet 5.5 --- crates/claude-connector/tests/lease.rs | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/crates/claude-connector/tests/lease.rs b/crates/claude-connector/tests/lease.rs index 5115e8a..0316b9f 100644 --- a/crates/claude-connector/tests/lease.rs +++ b/crates/claude-connector/tests/lease.rs @@ -726,13 +726,3 @@ fn a_readout_beside_a_held_lease_leaves_it_held() { assert_eq!(refusal_of(&second), Some(LeaseRefusalKind::Busy)); assert!(granted_ok(&held)); } - -#[test] -fn a_readout_does_not_wait_for_the_meter() { - let harness = start(setup()); - harness.quota.attach_meter(); - let started = Instant::now(); - let readout = usage_of(&read_usage(&harness)); - assert!(started.elapsed() < Duration::from_secs(5)); - assert!(readout.plan_quota.is_empty()); -} From 51982b7f63c08dc6c6d60aa57028575104eac8ae Mon Sep 17 00:00:00 2001 From: Hands Date: Fri, 6 Nov 2026 23:31:24 +0100 Subject: [PATCH 4/5] lease readouts: a pool of their own, a refusal over it, the arm's promises pinned, the request type named for the lease Readouts no longer take a lease's connection slot: the accept loop admits max_connections + READOUT_CONNECTIONS, a lease holds one of max_connections after it is read, a readout one of READOUT_CONNECTIONS, and a readout over its limit is refused Busy rather than dropped. Socket tests hold the service lock and a metered cell across readouts (no asking, no week wait, no lock), flood the readout pool beside leases, and pin the refusal. The new public request type is LeaseUsageReadoutRequest so it no longer shares a name with the kit's EnvelopeKind::UsageReadoutRequest; wire bytes and fixtures are unchanged. Co-Authored-By: Claude Sonnet 5.5 --- crates/claude-connector/src/lease.rs | 210 +++++++++++++++++- crates/claude-connector/src/lib.rs | 4 +- crates/claude-connector/tests/lease.rs | 4 +- .../tests/lease_wire_golden.rs | 4 +- 4 files changed, 213 insertions(+), 9 deletions(-) diff --git a/crates/claude-connector/src/lease.rs b/crates/claude-connector/src/lease.rs index 1749f6f..a049bba 100644 --- a/crates/claude-connector/src/lease.rs +++ b/crates/claude-connector/src/lease.rs @@ -53,7 +53,7 @@ pub enum LeaseMessage { ClaudeLeaseGrant(LeaseGranted), ClaudeLeaseRefusal(LeaseRefused), /// Asks for the plan's usage windows. Opens no lease. - ClaudeUsageReadoutRequest(UsageReadoutRequest), + ClaudeUsageReadoutRequest(LeaseUsageReadoutRequest), /// The answer to a readout request: the cell's windows as of now, no fd. ClaudeUsageReadout(UsageReadout), } @@ -66,7 +66,7 @@ pub struct LeaseRequest { /// A readout request names nothing: the answer is the same for every asker. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct UsageReadoutRequest {} +pub struct LeaseUsageReadoutRequest {} #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct LeaseGranted { @@ -395,8 +395,17 @@ struct Lender { quota: Arc, tokens: Arc, units: Arc, + /// Connections holding or waiting on a lease, at most `max_connections`. + leases: Arc, + /// Connections being answered a readout, at most `READOUT_CONNECTIONS`. + readouts: Arc, } +/// How many usage readouts are served at once, beside the leases. A readout +/// holds nothing, so a few suffice, and a flood of them cannot take a lease's +/// slot: the accept loop admits `max_connections` plus this many. +const READOUT_CONNECTIONS: usize = 4; + /// Serves leases on `listener` until it fails. Each connection is one lease or /// one usage readout. A lease: request, then grant or refusal, then (after a /// grant) the hold. A readout: request, then one frame, then close; its reach @@ -422,11 +431,14 @@ pub fn serve_leases( quota, tokens, units, + leases: Arc::new(AtomicUsize::new(0)), + readouts: Arc::new(AtomicUsize::new(0)), }); let active = Arc::new(AtomicUsize::new(0)); loop { let (stream, _) = listener.accept()?; - let Some(permit) = Permit::acquire(&active, limits.max_connections) else { + let admitted = limits.max_connections.saturating_add(READOUT_CONNECTIONS); + let Some(permit) = Permit::acquire(&active, admitted) else { continue; }; let lender = lender.clone(); @@ -447,10 +459,22 @@ impl Lender { let request_deadline = Instant::now() + self.limits.socket_timeout; match read_message(&mut stream, request_deadline)? { LeaseMessage::ClaudeLeaseRequest(request) => { + // A lease over the limit is dropped unanswered; readouts + // never count against it. + let Some(_slot) = Permit::acquire(&self.leases, self.limits.max_connections) else { + return Ok(()); + }; let peer = peer_credentials(&stream)?; self.lend(stream, peer, request) } LeaseMessage::ClaudeUsageReadoutRequest(_) => { + let Some(_slot) = Permit::acquire(&self.readouts, READOUT_CONNECTIONS) else { + return refuse( + &mut stream, + LeaseRefusalKind::Busy, + "too many usage readouts at once", + ); + }; stream.write_all(&encode_frame(&readout(&self.quota, unix_ms()))) } _ => Err(io::Error::from(io::ErrorKind::InvalidData)), @@ -662,6 +686,186 @@ fn refuse(stream: &mut UnixStream, kind: LeaseRefusalKind, detail: &str) -> io:: mod tests { use super::*; + struct NoTokens; + impl TokenSource for NoTokens { + fn read(&self) -> Option> { + panic!("a readout read the token") + } + } + struct NoUnits; + impl PeerUnits for NoUnits { + fn invocation_in(&self, _: BorrowedFd<'_>, _: &str) -> Option { + panic!("a readout looked up the peer's unit") + } + } + + fn lender( + quota: &Arc, + dir: &std::path::Path, + max_connections: usize, + ) -> (Arc>, Lender) { + let service = Arc::new(Mutex::new( + TransportService::open( + &dir.join("replay.cc"), + "claude-code", + [model_connector::CallerAdmission::new( + "lease-test-caller", + "lease-test-connection-key", + 1, + ["model".to_string()], + 1, + 65_536, + 32_768, + ) + .unwrap()], + 300_000, + ) + .unwrap(), + )); + let lender = Lender { + callers: Vec::new(), + limits: LeaseLimits { + socket_timeout: Duration::from_secs(2), + max_lease_ms: 3_600_000, + max_connections, + }, + service: service.clone(), + quota: quota.clone(), + tokens: Arc::new(NoTokens), + units: Arc::new(NoUnits), + leases: Arc::new(AtomicUsize::new(0)), + readouts: Arc::new(AtomicUsize::new(0)), + }; + (service, lender) + } + + /// Serves `lender`'s parts on a socket in `dir`; the path comes back. + fn serve(lender: Lender, dir: &std::path::Path) -> std::path::PathBuf { + let socket = dir.join("lease.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let (service, quota) = (lender.service.clone(), lender.quota.clone()); + thread::spawn(move || { + let _ = serve_leases( + &listener, + lender.callers, + lender.limits, + service, + quota, + lender.tokens, + lender.units, + ); + }); + socket + } + + fn ask(socket: &std::path::Path, message: &LeaseMessage) -> Option { + let mut stream = UnixStream::connect(socket).ok()?; + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + stream.write_all(&encode_frame(message)).ok()?; + let mut raw = Vec::new(); + let _ = stream.read_to_end(&mut raw); + decode_frame(&raw) + } + + fn readout_request() -> LeaseMessage { + LeaseMessage::ClaudeUsageReadoutRequest(LeaseUsageReadoutRequest {}) + } + + fn scratch(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("{name}-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn a_socket_readout_never_asks_the_meter_waits_for_a_week_or_takes_the_service_lock() { + let dir = scratch("readout-arm"); + let quota = Arc::new(QuotaCell::new()); + quota.attach_meter(); + let (service, lender) = lender("a, &dir, 8); + let socket = serve(lender, &dir); + // Held throughout: a readout that takes it hangs past the read timeout. + let guard = service.lock().unwrap(); + for _ in 0..20 { + let started = Instant::now(); + let Some(LeaseMessage::ClaudeUsageReadout(answer)) = ask(&socket, &readout_request()) + else { + panic!("no readout"); + }; + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(answer.plan_quota.is_empty()); + } + drop(guard); + assert_eq!(quota.asks_so_far(), 0); + assert!(!quota.wanted_at(unix_ms(), Duration::from_secs(15 * 60))); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn a_flood_of_readouts_at_their_limit_leaves_every_lease_request_answered() { + let dir = scratch("readout-flood"); + let quota = Arc::new(QuotaCell::new()); + // Fewer lease slots than flooding readers: the readers must not need them. + let (_service, lender) = lender("a, &dir, 2); + let socket = serve(lender, &dir); + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let flood: Vec<_> = (0..READOUT_CONNECTIONS) + .map(|_| { + let (socket, stop) = (socket.clone(), stop.clone()); + thread::spawn(move || { + while !stop.load(Ordering::Acquire) { + let _ = ask(&socket, &readout_request()); + } + }) + }) + .collect(); + // No caller is configured, so each lease is answered UnknownCaller: + // an answer is all that is asked. + let lease = LeaseMessage::ClaudeLeaseRequest(LeaseRequest { + expires_at_unix_ms: unix_ms() + 60_000, + }); + let unanswered = (0..40) + .filter(|_| { + !matches!( + ask(&socket, &lease), + Some(LeaseMessage::ClaudeLeaseRefusal(LeaseRefused { + kind: LeaseRefusalKind::UnknownCaller, + .. + })) + ) + }) + .count(); + stop.store(true, Ordering::Release); + for client in flood { + client.join().unwrap(); + } + assert_eq!(unanswered, 0); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn a_readout_over_the_readout_limit_is_refused_busy_not_dropped() { + let dir = scratch("readout-limit"); + let quota = Arc::new(QuotaCell::new()); + let (_service, lender) = lender("a, &dir, 8); + let held: Vec<_> = (0..READOUT_CONNECTIONS) + .map(|_| Permit::acquire(&lender.readouts, READOUT_CONNECTIONS).unwrap()) + .collect(); + let (mut client, server) = UnixStream::pair().unwrap(); + client.write_all(&encode_frame(&readout_request())).unwrap(); + lender.serve_one(server).unwrap(); + let mut raw = Vec::new(); + client.read_to_end(&mut raw).unwrap(); + let Some(LeaseMessage::ClaudeLeaseRefusal(refusal)) = decode_frame(&raw) else { + panic!("no refusal"); + }; + assert_eq!(refusal.kind, LeaseRefusalKind::Busy); + drop(held); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn a_readout_of_an_empty_metered_cell_does_not_wait() { let quota = QuotaCell::new(); diff --git a/crates/claude-connector/src/lib.rs b/crates/claude-connector/src/lib.rs index 52f957f..88de37b 100644 --- a/crates/claude-connector/src/lib.rs +++ b/crates/claude-connector/src/lib.rs @@ -29,8 +29,8 @@ pub use launch::ChildEnv; #[cfg(target_os = "linux")] pub use lease::{ LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused, - LeaseRequest, MAX_LEASE_FRAME_BYTES, PeerCredentials, PeerUnit, PeerUnits, UsageReadoutRequest, - decode_frame, encode_frame, peer_credentials, serve_leases, user_uid, + LeaseRequest, LeaseUsageReadoutRequest, MAX_LEASE_FRAME_BYTES, PeerCredentials, PeerUnit, + PeerUnits, decode_frame, encode_frame, peer_credentials, serve_leases, user_uid, }; #[cfg(unix)] pub use process::StdProcessPort; diff --git a/crates/claude-connector/tests/lease.rs b/crates/claude-connector/tests/lease.rs index 0316b9f..61e5a5e 100644 --- a/crates/claude-connector/tests/lease.rs +++ b/crates/claude-connector/tests/lease.rs @@ -15,7 +15,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use claude_connector::{ LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused, - LeaseRequest, PeerUnit, PeerUnits, QuotaCell, TokenSource, UsageReadoutRequest, decode_frame, + LeaseRequest, PeerUnit, PeerUnits, QuotaCell, TokenSource, LeaseUsageReadoutRequest, decode_frame, encode_frame, serve_leases, }; use model_connector::{CallerAdmission, LeaseTurn, PlanQuota, TransportService, UsageReadout}; @@ -658,7 +658,7 @@ fn read_usage(harness: &Harness) -> Answer { .unwrap(); stream .write_all(&encode_frame(&LeaseMessage::ClaudeUsageReadoutRequest( - UsageReadoutRequest {}, + LeaseUsageReadoutRequest {}, ))) .unwrap(); let (raw, fds) = receive(&stream); diff --git a/crates/claude-connector/tests/lease_wire_golden.rs b/crates/claude-connector/tests/lease_wire_golden.rs index c713889..83820e3 100644 --- a/crates/claude-connector/tests/lease_wire_golden.rs +++ b/crates/claude-connector/tests/lease_wire_golden.rs @@ -8,7 +8,7 @@ use std::path::{Path, PathBuf}; use claude_connector::{ - LeaseGranted, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseRequest, UsageReadoutRequest, + LeaseGranted, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseRequest, LeaseUsageReadoutRequest, decode_frame, encode_frame, }; use model_connector::{PlanQuota, PlanQuotaSnapshot, UsageReadout}; @@ -107,7 +107,7 @@ fn the_usage_readout_frames_are_pinned() { status: status.to_string(), observed_at_unix_ms: 1_793_300_000_000, }; - let request = LeaseMessage::ClaudeUsageReadoutRequest(UsageReadoutRequest {}); + let request = LeaseMessage::ClaudeUsageReadoutRequest(LeaseUsageReadoutRequest {}); let answer = LeaseMessage::ClaudeUsageReadout(UsageReadout::new( 1_793_300_100_000, PlanQuotaSnapshot { From 028224411477b1f99fec746775f0d1cda1eab660 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 6 Nov 2026 23:55:01 +0100 Subject: [PATCH 5/5] lease tests: rustfmt the renamed imports (Soul s2 test-imports-unformatted) Co-Authored-By: Claude Opus 5.5 --- crates/claude-connector/tests/lease.rs | 4 ++-- crates/claude-connector/tests/lease_wire_golden.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/claude-connector/tests/lease.rs b/crates/claude-connector/tests/lease.rs index 61e5a5e..9c6f27d 100644 --- a/crates/claude-connector/tests/lease.rs +++ b/crates/claude-connector/tests/lease.rs @@ -15,8 +15,8 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use claude_connector::{ LeaseCaller, LeaseGranted, LeaseLimits, LeaseMessage, LeaseRefusalKind, LeaseRefused, - LeaseRequest, PeerUnit, PeerUnits, QuotaCell, TokenSource, LeaseUsageReadoutRequest, decode_frame, - encode_frame, serve_leases, + LeaseRequest, LeaseUsageReadoutRequest, PeerUnit, PeerUnits, QuotaCell, TokenSource, + decode_frame, encode_frame, serve_leases, }; use model_connector::{CallerAdmission, LeaseTurn, PlanQuota, TransportService, UsageReadout}; use zeroize::Zeroizing; diff --git a/crates/claude-connector/tests/lease_wire_golden.rs b/crates/claude-connector/tests/lease_wire_golden.rs index 83820e3..5b51f8a 100644 --- a/crates/claude-connector/tests/lease_wire_golden.rs +++ b/crates/claude-connector/tests/lease_wire_golden.rs @@ -8,8 +8,8 @@ use std::path::{Path, PathBuf}; use claude_connector::{ - LeaseGranted, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseRequest, LeaseUsageReadoutRequest, - decode_frame, encode_frame, + LeaseGranted, LeaseMessage, LeaseRefusalKind, LeaseRefused, LeaseRequest, + LeaseUsageReadoutRequest, decode_frame, encode_frame, }; use model_connector::{PlanQuota, PlanQuotaSnapshot, UsageReadout};