diff --git a/crates/claude-connector/src/backend.rs b/crates/claude-connector/src/backend.rs index e10d6be..bc72890 100644 --- a/crates/claude-connector/src/backend.rs +++ b/crates/claude-connector/src/backend.rs @@ -187,7 +187,7 @@ impl ClaudeCodeBackend { child.as_mut(), deadline, &request.model, - request.output_schema_json.is_some(), + request.output_schema_json.as_deref(), &self.quota, ); child.kill(); @@ -220,14 +220,20 @@ impl ClaudeCodeBackend { text: finished.texts.concat(), }); } - let handle = ClaudeCodeHandle::Ended { - session_id: finished.session_id, - instance: self.config.instance.clone(), - prefix_sha256: prefix_digest(&request.instructions, &request.model, &prefix), - }; - payloads.push(TransportEventPayload::ReasoningState { - state: handle.encode(), - }); + // A schema turn leaves the CLI's StructuredOutput tool exchange in its + // session, and a later turn without the schema runs with no tools: it + // must not resume that session. So a schema turn hands out no handle, + // and whatever follows it is seeded from the caller's transcript. + if request.output_schema_json.is_none() { + let handle = ClaudeCodeHandle::Ended { + session_id: finished.session_id, + instance: self.config.instance.clone(), + prefix_sha256: prefix_digest(&request.instructions, &request.model, &prefix), + }; + payloads.push(TransportEventPayload::ReasoningState { + state: handle.encode(), + }); + } let events = payloads .into_iter() diff --git a/crates/claude-connector/src/backend/tests.rs b/crates/claude-connector/src/backend/tests.rs index 009775a..a48aa9c 100644 --- a/crates/claude-connector/src/backend/tests.rs +++ b/crates/claude-connector/src/backend/tests.rs @@ -95,7 +95,7 @@ fn structured_tool_result_line() -> String { } fn plan_rate_limit_line() -> String { - r#"{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","resetsAt":1793307000,"rateLimitType":"five_hour","overageStatus":"rejected","isUsingOverage":false,"unifiedWindows":{"five_hour":{"utilization":0.08,"resetsAt":1793307000,"status":"allowed"},"seven_day":{"utilization":0.3,"resetsAt":1793656800,"status":"allowed"}}}}"#.to_string() + r#"{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","resetsAt":4102444800,"rateLimitType":"five_hour","overageStatus":"rejected","isUsingOverage":false,"unifiedWindows":{"five_hour":{"utilization":0.08,"resetsAt":4102444800,"status":"allowed"},"seven_day":{"utilization":0.3,"resetsAt":4102531200,"status":"allowed"}}}}"#.to_string() } fn structured_result_line() -> String { @@ -1923,12 +1923,22 @@ fn quota_of(rig: &Rig) -> model_connector::PlanQuotaSnapshot { rig.backend.plan_quota() } +/// The windows with their observation stamps zeroed, for comparing readings. +fn readings(rig: &Rig) -> Vec { + let mut windows = quota_of(rig).windows; + for window in &mut windows { + window.observed_at_unix_ms = 0; + } + windows +} + fn window(name: &str, permille: u32, resets_at_unix_ms: u64) -> model_connector::PlanQuota { model_connector::PlanQuota { window: name.to_string(), - utilization_permille: permille, + utilization_permille: Some(permille), resets_at_unix_ms, status: "allowed".to_string(), + observed_at_unix_ms: 0, } } @@ -1943,20 +1953,26 @@ fn rate_limit_event_updates_quota_cell() { rig.run(schema_request()); let snapshot = quota_of(&rig); assert_eq!( - snapshot.windows, + readings(&rig), vec![ - window("five_hour", 80, 1_793_307_000_000), - window("seven_day", 300, 1_793_656_800_000), + window("five_hour", 80, 4_102_444_800_000), + window("seven_day", 300, 4_102_531_200_000), ] ); - assert!(snapshot.observed_at_unix_ms.unwrap() >= before); + let after = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + for each in &snapshot.windows { + assert!((before..=after).contains(&each.observed_at_unix_ms)); + } // The cell is the one the backend hands out. assert_eq!(rig.backend.quota().snapshot(), snapshot); } #[test] fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself() { - let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":1793307000,"status":"rejected"},"seven_day":{"resetsAt":1793656800,"status":"allowed"},"idle":{"utilization":0,"resetsAt":1793307000,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#; + let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":4102444800,"status":"rejected"},"seven_day":{"resetsAt":4102531200,"status":"allowed"},"idle":{"utilization":0,"resetsAt":4102444800,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#; let rig = rig(vec![Script::lines(vec![ init_line(MODEL, SESSION), event.to_string(), @@ -1964,20 +1980,20 @@ fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself( ])]); let (kind, _, _) = failure_of(&rig.run(request(vec![user("hello")]))); assert_eq!(kind, FailureKind::RateLimited); - let mut rejected = window("five_hour", 1000, 1_793_307_000_000); + let mut rejected = window("five_hour", 1000, 4_102_444_800_000); rejected.status = "rejected".to_string(); // Unknown is never zero: a window missing a field or of another type is // skipped, not read as empty. // A window at exactly zero use is a reading, and is kept. assert_eq!( - quota_of(&rig).windows, - vec![rejected, window("idle", 0, 1_793_307_000_000)] + readings(&rig), + vec![rejected, window("idle", 0, 4_102_444_800_000)] ); } #[test] fn a_later_event_replaces_only_the_windows_it_names() { - let only_five = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","unifiedWindows":{"five_hour":{"utilization":0.5,"resetsAt":1793310000,"status":"allowed"}}}}"#; + let only_five = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","unifiedWindows":{"five_hour":{"utilization":0.5,"resetsAt":4102448400,"status":"allowed"}}}}"#; let rig = rig(vec![ structured_script(), Script::lines(vec![ @@ -1990,10 +2006,10 @@ fn a_later_event_replaces_only_the_windows_it_names() { rig.run(schema_request()); rig.run(request(vec![user("again")])); assert_eq!( - quota_of(&rig).windows, + readings(&rig), vec![ - window("five_hour", 500, 1_793_310_000_000), - window("seven_day", 300, 1_793_656_800_000), + window("five_hour", 500, 4_102_448_400_000), + window("seven_day", 300, 4_102_531_200_000), ] ); } @@ -2415,3 +2431,127 @@ mod real { )); } } + +// ---- a call that never reached the provider ---- + +/// The shape claude 2.1.312 gave with no usable credential: an error result +/// with no status, no API time and no usage rows. `extra` replaces the +/// per-turn usage field. +fn not_logged_in_line(status: &str, api_ms: u64, extra: &str) -> String { + format!( + r#"{{"type":"result","subtype":"success","is_error":true,{status}"duration_ms":118,"duration_api_ms":{api_ms},"stop_reason":"stop_sequence","result":"Not logged in","total_cost_usd":0,"modelUsage":{{}}{extra}}}"# + ) +} + +#[test] +fn a_result_that_made_no_api_call_bills_nothing() { + let reservation = reservation(); + let empty = r#","model_usage_since_last_result":{}"#; + let cases = [ + // (status field, api ms, usage field, billed nothing) + (r#""api_error_status":null,"#, 0, empty, true), + ("", 0, empty, true), + ("", 0, "", true), + // An API call was made, or the status names a server fault, or the + // CLI reported rows: the reservation stands. + ("", 2300, empty, false), + (r#""api_error_status":500,"#, 0, empty, false), + (r#""api_error_status":null,"#, 0, r#","model_usage_since_last_result":"?""#, false), + ]; + for (status, api_ms, extra, free) in cases { + let line = not_logged_in_line(status, api_ms, extra); + let result = rig(vec![Script::lines(vec![ + init_line(MODEL, SESSION), + line.clone(), + ])]) + .run(request(vec![user("hi")])); + let (_, _, usage) = failure_of(&result); + if free { + assert_eq!(usage, Some(usage_zero()), "{line}"); + assert_eq!(parts(&result).1.cost_micro_usd, 0, "{line}"); + } else { + assert_eq!(usage, None, "{line}"); + assert_eq!(parts(&result).1.cost_micro_usd, reservation, "{line}"); + } + } +} + +#[test] +fn a_no_call_result_that_reports_rows_is_billed_those_rows() { + let line = not_logged_in_line( + "", + 0, + &format!(r#","model_usage_since_last_result":{USAGE}"#), + ); + let result = rig(vec![Script::lines(vec![init_line(MODEL, SESSION), line])]) + .run(request(vec![user("hi")])); + let (_, _, billed) = failure_of(&result); + assert_eq!(billed, Some(usage(2, 7, 1043, 4, Some(3)))); +} + +// ---- the structured output is an object ---- + +#[test] +fn a_structured_output_that_is_not_an_object_fails_as_provider_with_usage() { + for output in ["\"text\"", "[1]", "42", "true", "null"] { + let result = rig(vec![Script::lines(vec![ + init_line(MODEL, SESSION), + structured_call_line("StructuredOutput"), + result_line_with( + "tool_use", + USAGE, + &format!(r#","structured_output":{output}"#), + ), + ])]) + .run(schema_request()); + let (kind, message, usage) = failure_of(&result); + assert_eq!(kind, FailureKind::Provider, "{output}"); + assert!(message.contains("structured output"), "{message}"); + assert!(usage.is_some(), "{output}: the turn ran and is billed"); + } +} + +#[test] +fn an_object_answer_to_a_schema_whose_top_type_is_not_object_fails() { + for (schema, accepted) in [ + (r#"{"type":"array"}"#, false), + (r#"{"type":"string"}"#, false), + (r#"{"type":["object","null"]}"#, false), + (r#"{"type":"object"}"#, true), + (r#"{"properties":{}}"#, true), + ] { + let mut asked = request(vec![user("hello")]); + asked.output_schema_json = Some(schema.to_string()); + let result = rig(vec![structured_script()]).run(asked); + assert_eq!( + matches!(outcome(&result), TransportOutcome::Completed { .. }), + accepted, + "{schema}" + ); + } +} + +// ---- a schema turn does not hand out a session to resume ---- + +#[test] +fn a_turn_after_a_schema_turn_is_seeded_not_resumed() { + let rig = rig(vec![structured_script(), ok_script(SESSION, "next")]); + let first = rig.run(schema_request()); + assert!( + !parts(&first) + .0 + .iter() + .any(|e| matches!(e.payload, TransportEventPayload::ReasoningState { .. })), + "a schema turn leaves no handle" + ); + rig.run(request(vec![ + user("hello"), + InputItem::AssistantText { + text: STRUCTURED.to_string(), + }, + user("and now plainly"), + ])); + let argv = rig.seen().argv; + assert!(!argv.iter().any(|a| a == "--resume"), "{argv:?}"); + assert!(!argv.iter().any(|a| a == "--json-schema")); +} diff --git a/crates/claude-connector/src/quota.rs b/crates/claude-connector/src/quota.rs index 3c02d2c..a8477fd 100644 --- a/crates/claude-connector/src/quota.rs +++ b/crates/claude-connector/src/quota.rs @@ -1,7 +1,7 @@ //! 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 -//! place that number appears, so the latest one seen is kept here and read out -//! beside the connector's own spend ledger. +//! place that number appears, so the latest observation of each window is kept +//! here and read out beside the connector's own spend ledger. use std::collections::BTreeMap; use std::sync::Mutex; @@ -10,45 +10,60 @@ use std::time::{SystemTime, UNIX_EPOCH}; use model_connector::{PlanQuota, PlanQuotaSnapshot}; /// The latest observation of each plan window, shared by the backend that -/// feeds it and whatever reads it out. +/// feeds it and whatever reads it out. Each window carries the time of its +/// own observation: an event that names one window says nothing about the +/// others. #[derive(Default)] -pub struct QuotaCell(Mutex); - -#[derive(Default)] -struct Observed { - windows: BTreeMap, - observed_at_unix_ms: Option, -} +pub struct QuotaCell(Mutex>); impl QuotaCell { pub fn new() -> Self { Self::default() } - /// Records the windows one event reported: each replaces that window's - /// earlier observation, and windows the event did not mention stay. An - /// event that reported none changes nothing, not even the time. + /// Records the windows one event reported, stamping each with now. Each + /// replaces that window's earlier observation, and windows the event did + /// not mention stay as they were, stamp included. pub fn observe(&self, windows: impl IntoIterator) { + self.observe_at(unix_ms(), windows); + } + + fn observe_at(&self, now: u64, windows: impl IntoIterator) { let mut held = self.0.lock().unwrap_or_else(|e| e.into_inner()); - let mut any = false; - for quota in windows { - held.windows.insert(quota.window.clone(), quota); - any = true; - } - if any { - held.observed_at_unix_ms = Some(unix_ms()); + for mut quota in windows { + quota.observed_at_unix_ms = now; + held.insert(quota.window.clone(), quota); } } pub fn snapshot(&self) -> PlanQuotaSnapshot { + self.snapshot_at(unix_ms()) + } + + /// The windows as of `now`. One whose reset time has passed is reported + /// `expired` with no utilization: the reading described a window that no + /// longer exists. + fn snapshot_at(&self, now: u64) -> PlanQuotaSnapshot { let held = self.0.lock().unwrap_or_else(|e| e.into_inner()); PlanQuotaSnapshot { - windows: held.windows.values().cloned().collect(), - observed_at_unix_ms: held.observed_at_unix_ms, + windows: held + .values() + .cloned() + .map(|mut quota| { + if now >= quota.resets_at_unix_ms { + quota.utilization_permille = None; + quota.status = EXPIRED.to_string(); + } + quota + }) + .collect(), } } } +/// The status of a window whose reset has passed since it was observed. +const EXPIRED: &str = "expired"; + fn unix_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -61,47 +76,70 @@ fn unix_ms() -> u64 { mod tests { use super::*; - fn quota(window: &str, permille: u32) -> PlanQuota { + const RESET: u64 = 1_793_307_000_000; + + fn quota(window: &str, permille: u32, observed_at: u64) -> PlanQuota { PlanQuota { window: window.to_string(), - utilization_permille: permille, - resets_at_unix_ms: 1_793_307_000_000, + utilization_permille: Some(permille), + resets_at_unix_ms: RESET, status: "allowed".to_string(), + observed_at_unix_ms: observed_at, } } #[test] fn an_empty_cell_has_observed_nothing() { - let snapshot = QuotaCell::new().snapshot(); - assert!(snapshot.windows.is_empty()); - assert_eq!(snapshot.observed_at_unix_ms, None); + assert!(QuotaCell::new().snapshot().windows.is_empty()); } #[test] - fn a_window_keeps_only_its_latest_observation_and_others_stay() { + fn each_window_keeps_its_own_latest_observation_and_its_own_time() { + let cell = QuotaCell::new(); + cell.observe_at(100, [quota("five_hour", 80, 0), quota("seven_day", 300, 0)]); + cell.observe_at(200, [quota("five_hour", 120, 0)]); + assert_eq!( + cell.snapshot_at(250).windows, + vec![quota("five_hour", 120, 200), quota("seven_day", 300, 100)], + "the window the later event did not name keeps its older stamp" + ); + } + + #[test] + fn an_event_with_no_windows_changes_nothing() { + let cell = QuotaCell::new(); + cell.observe_at(100, [quota("five_hour", 80, 0)]); + cell.observe_at(900, Vec::new()); + assert_eq!(cell.snapshot_at(950).windows, vec![quota("five_hour", 80, 100)]); + } + + #[test] + fn a_window_past_its_reset_is_expired_not_old_utilization() { + let cell = QuotaCell::new(); + cell.observe_at(100, [quota("five_hour", 80, 0)]); + let live = cell.snapshot_at(RESET - 1).windows; + assert_eq!(live, vec![quota("five_hour", 80, 100)]); + let expired = cell.snapshot_at(RESET).windows; + assert_eq!(expired.len(), 1); + assert_eq!(expired[0].status, "expired"); + assert_eq!(expired[0].utilization_permille, None); + assert_eq!(expired[0].observed_at_unix_ms, 100, "its age stays readable"); + assert_eq!(expired[0].resets_at_unix_ms, RESET); + } + + #[test] + fn the_live_clock_stamps_and_expires() { let cell = QuotaCell::new(); let before = unix_ms(); - cell.observe([quota("five_hour", 80), quota("seven_day", 300)]); - cell.observe([quota("five_hour", 120)]); + let mut past = quota("old", 10, 0); + past.resets_at_unix_ms = 1; + cell.observe([past, quota("five_hour", 80, 0)]); let after = unix_ms(); - let snapshot = cell.snapshot(); - assert_eq!( - snapshot.windows, - vec![quota("five_hour", 120), quota("seven_day", 300)] - ); - let at = snapshot.observed_at_unix_ms.expect("stamped"); - assert!((before..=after).contains(&at)); - } - - #[test] - fn an_event_with_no_windows_does_not_move_the_stamp() { - let cell = QuotaCell::new(); - cell.observe(Vec::new()); - assert_eq!(cell.snapshot().observed_at_unix_ms, None); - cell.observe([quota("five_hour", 80)]); - let first = cell.snapshot().observed_at_unix_ms; - std::thread::sleep(std::time::Duration::from_millis(3)); - cell.observe(Vec::new()); - assert_eq!(cell.snapshot().observed_at_unix_ms, first); + let windows = cell.snapshot().windows; + let stamp = windows[0].observed_at_unix_ms; + assert!((before..=after).contains(&stamp)); + assert_eq!(windows[0].window, "five_hour"); + assert_eq!(windows[0].utilization_permille, Some(80)); + assert_eq!(windows[1].status, "expired"); } } diff --git a/crates/claude-connector/src/stream.rs b/crates/claude-connector/src/stream.rs index a75a37f..1c56821 100644 --- a/crates/claude-connector/src/stream.rs +++ b/crates/claude-connector/src/stream.rs @@ -75,6 +75,8 @@ struct Line { is_error: Option, stop_reason: Option, api_error_status: Option, + /// Milliseconds spent in API calls; zero means no call was made. + duration_api_ms: Option, result: Option, errors: Option, /// The answer to a `--json-schema` request, an object. @@ -116,6 +118,9 @@ pub(crate) struct StreamState<'a> { /// Whether the request carried an output schema, so the answer is the /// result's structured output. structured: bool, + /// Whether that schema's top type admits an object (absent counts): the + /// structured output must be an object, never a bare scalar or array. + schema_top_is_object: bool, quota: &'a QuotaCell, session_id: Option, texts: Vec, @@ -124,10 +129,15 @@ pub(crate) struct StreamState<'a> { } impl<'a> StreamState<'a> { - pub(crate) fn new(model: &'a str, structured: bool, quota: &'a QuotaCell) -> Self { + /// `schema` is the request's output schema text, when it carried one. + pub(crate) fn new(model: &'a str, schema: Option<&str>, quota: &'a QuotaCell) -> Self { + let top = schema + .and_then(|text| serde_json::from_str::(text).ok()) + .and_then(|schema| schema.get("type").cloned()); Self { model, - structured, + structured: schema.is_some(), + schema_top_is_object: top.is_none_or(|top| top.as_str() == Some("object")), quota, session_id: None, texts: Vec::new(), @@ -211,7 +221,16 @@ impl<'a> StreamState<'a> { line.api_error_status.as_ref().and_then(Value::as_u64), Some(401 | 403 | 429) ); - let usage = read_usage(line.model_usage_since_last_result, door_refusal); + // The CLI found no credential and never called the API: an error with + // no status and no API time at all. + let no_call = !success + && line.duration_api_ms == Some(0) + && line.api_error_status.as_ref().is_none_or(Value::is_null); + let usage = read_usage( + line.model_usage_since_last_result, + door_refusal || no_call, + no_call, + ); if !success { return Err(Failure { kind: self.error_kind(&line.api_error_status, &line.result, &line.errors), @@ -230,10 +249,13 @@ impl<'a> StreamState<'a> { if self.structured { // The answer is the structured output alone; any prose around the // tool call is not part of it. - let Some(output) = line.structured_output.filter(|value| !value.is_null()) else { + let Some(output) = line + .structured_output + .filter(|value| value.is_object() && self.schema_top_is_object) + else { return Err(Failure { kind: FailureKind::Provider, - message: "claude-code: a schema was sent and the result carried no structured output", + message: "claude-code: a schema was sent and the result carried no structured output object", usage: Some(usage), }); }; @@ -314,9 +336,11 @@ fn plan_windows(info: &Value) -> Vec { Some(PlanQuota { window: window.clone(), // `as` saturates a float at the integer's bounds. - utilization_permille: (utilization * 1000.0).round() as u32, + utilization_permille: Some((utilization * 1000.0).round() as u32), resets_at_unix_ms: resets_at.saturating_mul(1000), status: status.to_string(), + // Stamped by the cell that keeps it. + observed_at_unix_ms: 0, }) }) .collect() @@ -328,14 +352,22 @@ fn fail(kind: FailureKind, message: &'static str) -> Step { /// What this turn consumed, summed over every model it used (harness side /// calls included), or `None` when the result does not say: the field absent, -/// of another shape, or empty. An empty report is believed only for a refusal -/// at the door (401, 403 or 429), which the provider cannot have billed; elsewhere -/// the CLI also zeroes usage on crashes after real work. `None` is billed the -/// reservation by the kit. -fn read_usage(reported: Option, door_refusal: bool) -> Option { - let models: BTreeMap = serde_json::from_value(reported?).ok()?; +/// of another shape, or empty. An empty report is believed only where the +/// 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 +/// crashes after real work. A call never made (`absent_is_zero`) may also +/// omit the field. `None` is billed the reservation by the kit. +fn read_usage( + reported: Option, + empty_is_zero: bool, + absent_is_zero: bool, +) -> Option { + let Some(reported) = reported else { + return absent_is_zero.then(|| sum_usage(std::iter::empty())); + }; + let models: BTreeMap = serde_json::from_value(reported).ok()?; if models.is_empty() { - return door_refusal.then(|| sum_usage(std::iter::empty())); + return empty_is_zero.then(|| sum_usage(std::iter::empty())); } Some(sum_usage(models.into_values())) } diff --git a/crates/model-connector/src/readout.rs b/crates/model-connector/src/readout.rs index a870ce3..e4e34df 100644 --- a/crates/model-connector/src/readout.rs +++ b/crates/model-connector/src/readout.rs @@ -14,18 +14,22 @@ pub const SPEND_READOUT_SCHEMA_ID: &str = "gamecult.model.spend_readout.v1"; #[serde(deny_unknown_fields)] pub struct PlanQuota { pub window: String, - /// How much of the window is used, in thousandths. - pub utilization_permille: u32, + /// How much of the window was used when it was observed, in thousandths; + /// `None` once the window has reset (`status` is then `expired`), because + /// the old reading says nothing about the new window. + pub utilization_permille: Option, pub resets_at_unix_ms: u64, pub status: String, + /// When this window was last reported. Its age in a readout is the + /// readout's `at_unix_ms` minus this. + pub observed_at_unix_ms: u64, } -/// What a backend last saw of the plan's windows and when it saw them. A -/// backend with no plan to report returns the default. +/// What a backend last saw of the plan's windows, each stamped with its own +/// observation time. A backend with no plan to report returns the default. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PlanQuotaSnapshot { pub windows: Vec, - pub observed_at_unix_ms: Option, } /// One caller's current spend window in micro-USD. `spent` is settled charges @@ -48,7 +52,6 @@ pub struct SpendReadout { pub at_unix_ms: u64, pub callers: Vec, pub plan_quota: Vec, - pub plan_quota_observed_at_unix_ms: Option, } impl SpendReadout { @@ -58,7 +61,6 @@ impl SpendReadout { at_unix_ms, callers, plan_quota: plan.windows, - plan_quota_observed_at_unix_ms: plan.observed_at_unix_ms, } } @@ -88,11 +90,11 @@ mod tests { PlanQuotaSnapshot { windows: vec![PlanQuota { window: "five_hour".into(), - utilization_permille: 80, + utilization_permille: Some(80), resets_at_unix_ms: 1_793_307_000_000, status: "allowed".into(), + observed_at_unix_ms: 999, }], - observed_at_unix_ms: Some(999), }, ) } @@ -111,8 +113,7 @@ mod tests { "schema_id", "at_unix_ms", "callers", - "plan_quota", - "plan_quota_observed_at_unix_ms" + "plan_quota" ] ); assert_eq!(wire_decode::(&bytes).unwrap(), readout()); @@ -123,7 +124,6 @@ mod tests { let bare = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default()); let decoded: SpendReadout = wire_decode(&wire_encode(&bare).unwrap()).unwrap(); assert!(decoded.plan_quota.is_empty()); - assert_eq!(decoded.plan_quota_observed_at_unix_ms, None); } #[test] diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs index ef54663..26f5e2d 100644 --- a/crates/model-connector/src/serve.rs +++ b/crates/model-connector/src/serve.rs @@ -544,11 +544,11 @@ mod tests { let plan = PlanQuotaSnapshot { windows: vec![crate::readout::PlanQuota { window: "five_hour".into(), - utilization_permille: 80, + utilization_permille: Some(80), resets_at_unix_ms: 1_793_307_000_000, status: "allowed".into(), + observed_at_unix_ms: 1_793_300_000_000, }], - observed_at_unix_ms: Some(1_793_300_000_000), }; let backend = Arc::new(PlanBackend { inner: inner.clone(), @@ -562,10 +562,6 @@ mod tests { let readout = client.spend_readout(CALLER).unwrap(); assert!(readout.at_unix_ms >= before && readout.at_unix_ms <= now_ms()); assert_eq!(readout.plan_quota, plan.windows); - assert_eq!( - readout.plan_quota_observed_at_unix_ms, - plan.observed_at_unix_ms - ); let [row] = readout.callers.as_slice() else { panic!("one admitted caller"); }; @@ -596,7 +592,6 @@ mod tests { let (client, _) = start(CountingBackend::new(true), &store); let readout = client.spend_readout(CALLER).unwrap(); assert!(readout.plan_quota.is_empty()); - assert_eq!(readout.plan_quota_observed_at_unix_ms, None); } #[test] diff --git a/schemas/spend_readout.json b/schemas/spend_readout.json index cfdd957..86aa926 100644 --- a/schemas/spend_readout.json +++ b/schemas/spend_readout.json @@ -20,14 +20,6 @@ "$ref": "#/$defs/PlanQuota" } }, - "plan_quota_observed_at_unix_ms": { - "type": [ - "integer", - "null" - ], - "format": "uint64", - "minimum": 0 - }, "schema_id": { "type": "string" } @@ -81,6 +73,13 @@ "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 +readout's `at_unix_ms` minus this.", + "type": "integer", + "format": "uint64", + "minimum": 0 + }, "resets_at_unix_ms": { "type": "integer", "format": "uint64", @@ -90,8 +89,13 @@ "type": "string" }, "utilization_permille": { - "description": "How much of the window is used, in thousandths.", - "type": "integer", + "description": "How much of the window was used when it was observed, in thousandths; +`None` once the window has reset (`status` is then `expired`), because +the old reading says nothing about the new window.", + "type": [ + "integer", + "null" + ], "format": "uint32", "minimum": 0 }, @@ -104,7 +108,8 @@ "window", "utilization_permille", "resets_at_unix_ms", - "status" + "status", + "observed_at_unix_ms" ] } }