1use std::collections::{BTreeMap, BTreeSet, HashMap};
2
3use serde::{Deserialize, Serialize};
4
5use crate::pricing::{CostConfidence, UsdAmount};
6use crate::routing_explain::{RoutingExplainCandidate, RoutingExplainResponse};
7use crate::runtime_identity::ProviderEndpointKey;
8use crate::state::{
9 BalanceSnapshotStatus, FinishedRequest, ProviderBalanceSnapshot, UsageBucket, UsageRollupView,
10};
11use crate::usage_providers::UsageProviderRefreshSummary;
12
13#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default, Hash)]
14#[serde(rename_all = "snake_case")]
15pub enum UsageBalanceStatus {
16 #[default]
17 Unknown,
18 Ok,
19 Unlimited,
20 Exhausted,
21 Stale,
22 Error,
23}
24
25impl UsageBalanceStatus {
26 pub fn as_str(self) -> &'static str {
27 match self {
28 UsageBalanceStatus::Unknown => "unknown",
29 UsageBalanceStatus::Ok => "ok",
30 UsageBalanceStatus::Unlimited => "unlimited",
31 UsageBalanceStatus::Exhausted => "exhausted",
32 UsageBalanceStatus::Stale => "stale",
33 UsageBalanceStatus::Error => "error",
34 }
35 }
36
37 pub fn is_attention(self) -> bool {
38 matches!(
39 self,
40 UsageBalanceStatus::Unknown
41 | UsageBalanceStatus::Exhausted
42 | UsageBalanceStatus::Stale
43 | UsageBalanceStatus::Error
44 )
45 }
46}
47
48#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
49pub struct UsageBalanceStatusCounts {
50 pub ok: usize,
51 pub unlimited: usize,
52 pub exhausted: usize,
53 pub stale: usize,
54 pub error: usize,
55 pub unknown: usize,
56}
57
58impl UsageBalanceStatusCounts {
59 pub fn total(&self) -> usize {
60 self.ok
61 .saturating_add(self.unlimited)
62 .saturating_add(self.exhausted)
63 .saturating_add(self.stale)
64 .saturating_add(self.error)
65 .saturating_add(self.unknown)
66 }
67
68 pub fn record(&mut self, status: UsageBalanceStatus) {
69 match status {
70 UsageBalanceStatus::Ok => self.ok += 1,
71 UsageBalanceStatus::Unlimited => self.unlimited += 1,
72 UsageBalanceStatus::Exhausted => self.exhausted += 1,
73 UsageBalanceStatus::Stale => self.stale += 1,
74 UsageBalanceStatus::Error => self.error += 1,
75 UsageBalanceStatus::Unknown => self.unknown += 1,
76 }
77 }
78
79 pub fn aggregate_status(&self) -> UsageBalanceStatus {
80 if self.total() == 0 {
81 UsageBalanceStatus::Unknown
82 } else if self.error > 0 {
83 UsageBalanceStatus::Error
84 } else if self.exhausted > 0 {
85 UsageBalanceStatus::Exhausted
86 } else if self.stale > 0 {
87 UsageBalanceStatus::Stale
88 } else if self.unknown > 0 {
89 UsageBalanceStatus::Unknown
90 } else if self.unlimited > 0 {
91 UsageBalanceStatus::Unlimited
92 } else {
93 UsageBalanceStatus::Ok
94 }
95 }
96}
97
98#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
99pub struct UsageBalanceView {
100 pub service_name: String,
101 pub window_days: usize,
102 pub generated_at_ms: u64,
103 pub totals: UsageBalanceTotals,
104 pub provider_rows: Vec<UsageBalanceProviderRow>,
105 pub endpoint_rows: Vec<UsageBalanceEndpointRow>,
106 pub routing_impacts: Vec<UsageBalanceRouteImpact>,
107 pub refresh_status: UsageBalanceRefreshStatus,
108 #[serde(default, skip_serializing_if = "Vec::is_empty")]
109 pub warnings: Vec<String>,
110}
111
112#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
113pub struct UsageBalanceTotals {
114 pub requests_total: u64,
115 pub requests_error: u64,
116 pub success_per_mille: Option<u16>,
117 pub input_tokens: i64,
118 pub cached_input_tokens: i64,
119 pub output_tokens: i64,
120 pub reasoning_output_tokens: i64,
121 pub total_tokens: i64,
122 pub cost_total_usd: Option<String>,
123 pub cost_display: String,
124 pub cost_confidence: CostConfidence,
125 pub balance_status_counts: UsageBalanceStatusCounts,
126}
127
128#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
129pub struct UsageBalanceProviderRow {
130 pub provider_id: String,
131 pub display_name: String,
132 pub usage: UsageBucket,
133 pub success_per_mille: Option<u16>,
134 pub output_tokens_per_second: Option<u64>,
135 pub avg_ttfb_ms: Option<u64>,
136 pub cost_total_usd: Option<String>,
137 pub cost_display: String,
138 pub cost_confidence: CostConfidence,
139 pub balance_status: UsageBalanceStatus,
140 pub balance_counts: UsageBalanceStatusCounts,
141 pub primary_balance: Option<UsageBalanceSnapshotSummary>,
142 pub latest_balance_error: Option<String>,
143 pub balance_age_ms: Option<u64>,
144 pub routing: UsageBalanceRouteImpact,
145 pub endpoint_count: usize,
146 pub endpoints_with_balance: usize,
147 pub recent_endpoint_requests: u64,
148}
149
150impl UsageBalanceProviderRow {
151 pub fn needs_attention(&self) -> bool {
152 self.balance_status.is_attention()
153 || self
154 .latest_balance_error
155 .as_deref()
156 .is_some_and(|value| !value.trim().is_empty())
157 || self.usage.requests_error > 0
158 }
159}
160
161#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
162pub struct UsageBalanceEndpointRow {
163 pub provider_id: String,
164 pub endpoint_id: String,
165 pub provider_endpoint_key: String,
166 pub base_url: Option<String>,
167 pub usage: UsageBucket,
168 pub balance_status: UsageBalanceStatus,
169 pub balance: Option<UsageBalanceSnapshotSummary>,
170 pub route_selected: bool,
171 pub route_skip_reasons: Vec<String>,
172}
173
174#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
175pub struct UsageBalanceSnapshotSummary {
176 pub provider_id: String,
177 pub provider_endpoint_key: Option<String>,
178 pub source: String,
179 pub status: UsageBalanceStatus,
180 pub amount_summary: String,
181 pub fetched_at_ms: u64,
182 pub stale_after_ms: Option<u64>,
183 pub age_ms: Option<u64>,
184 pub error: Option<String>,
185 pub exhaustion_affects_routing: bool,
186 pub routing_exhausted: bool,
187 pub routing_ignored_exhaustion: bool,
188}
189
190#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
191pub struct UsageBalanceRouteImpact {
192 pub provider_id: String,
193 pub selected: bool,
194 pub selected_endpoint_id: Option<String>,
195 pub selected_provider_endpoint_key: Option<String>,
196 pub candidate_count: usize,
197 pub skip_reasons: Vec<String>,
198 pub route_paths: Vec<Vec<String>>,
199}
200
201#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
202pub struct UsageBalanceRefreshStatus {
203 pub refreshing: bool,
204 pub total_snapshots: usize,
205 pub latest_fetched_at_ms: Option<u64>,
206 pub latest_error: Option<String>,
207 #[serde(default, skip_serializing_if = "Option::is_none")]
208 pub latest_error_provider_id: Option<String>,
209 pub last_message: Option<String>,
210 pub last_error: Option<String>,
211 #[serde(default, skip_serializing_if = "Option::is_none")]
212 pub last_provider_refresh: Option<UsageProviderRefreshSummary>,
213 pub status_counts: UsageBalanceStatusCounts,
214}
215
216#[derive(Debug, Clone, Default)]
217pub struct UsageBalanceRefreshInput {
218 pub refreshing: bool,
219 pub last_message: Option<String>,
220 pub last_error: Option<String>,
221 pub last_provider_refresh: Option<UsageProviderRefreshSummary>,
222}
223
224pub struct UsageBalanceBuildInput<'a> {
225 pub service_name: &'a str,
226 pub window_days: usize,
227 pub generated_at_ms: u64,
228 pub usage_rollup: &'a UsageRollupView,
229 pub provider_balances: &'a HashMap<String, Vec<ProviderBalanceSnapshot>>,
230 pub recent: &'a [FinishedRequest],
231 pub routing_explain: Option<&'a RoutingExplainResponse>,
232 pub refresh: UsageBalanceRefreshInput,
233}
234
235impl UsageBalanceView {
236 pub fn build(input: UsageBalanceBuildInput<'_>) -> Self {
237 let route_index = RouteIndex::from_explain(input.routing_explain);
238 let mut endpoint_rows = build_endpoint_rows(
239 input.service_name,
240 input.recent,
241 input.provider_balances,
242 &route_index,
243 input.generated_at_ms,
244 );
245 sort_endpoint_rows(&mut endpoint_rows);
246
247 let provider_rows = build_provider_rows(&input, &route_index, &endpoint_rows);
248 let totals = build_totals(
249 input.usage_rollup,
250 input.provider_balances,
251 input.generated_at_ms,
252 );
253 let refresh_status = build_refresh_status(
254 input.provider_balances,
255 input.generated_at_ms,
256 input.refresh,
257 );
258 let routing_impacts = route_index
259 .provider_impacts
260 .values()
261 .cloned()
262 .collect::<Vec<_>>();
263
264 Self {
265 service_name: input.service_name.to_string(),
266 window_days: input.window_days,
267 generated_at_ms: input.generated_at_ms,
268 totals,
269 provider_rows,
270 endpoint_rows,
271 routing_impacts,
272 refresh_status,
273 warnings: Vec::new(),
274 }
275 }
276}
277
278fn build_totals(
279 rollup: &UsageRollupView,
280 provider_balances: &HashMap<String, Vec<ProviderBalanceSnapshot>>,
281 now_ms: u64,
282) -> UsageBalanceTotals {
283 let window = &rollup.window;
284 let mut balance_status_counts = UsageBalanceStatusCounts::default();
285 for snapshot in provider_balances.values().flatten() {
286 balance_status_counts.record(classify_snapshot(snapshot, now_ms));
287 }
288
289 UsageBalanceTotals {
290 requests_total: window.requests_total,
291 requests_error: window.requests_error,
292 success_per_mille: per_mille(
293 window.requests_total.saturating_sub(window.requests_error),
294 window.requests_total,
295 ),
296 input_tokens: window.usage.input_tokens,
297 cached_input_tokens: window.usage.cache_read_tokens_total(),
298 output_tokens: window.usage.output_tokens,
299 reasoning_output_tokens: window.usage.reasoning_output_tokens_total(),
300 total_tokens: window.usage.total_tokens,
301 cost_total_usd: window.cost.total_cost_usd.clone(),
302 cost_display: window.cost.display_total(),
303 cost_confidence: window.cost.confidence,
304 balance_status_counts,
305 }
306}
307
308fn build_provider_rows(
309 input: &UsageBalanceBuildInput<'_>,
310 route_index: &RouteIndex,
311 endpoint_rows: &[UsageBalanceEndpointRow],
312) -> Vec<UsageBalanceProviderRow> {
313 let mut provider_ids = BTreeSet::new();
314 for (provider_id, _) in &input.usage_rollup.by_provider {
315 provider_ids.insert(provider_id.clone());
316 }
317 for (provider_id, _) in group_balances_by_provider(input.provider_balances) {
318 provider_ids.insert(provider_id);
319 }
320 for request in input.recent {
321 if let Some(provider_id) = request
322 .provider_id
323 .as_deref()
324 .map(str::trim)
325 .filter(|value| !value.is_empty())
326 {
327 provider_ids.insert(provider_id.to_string());
328 }
329 }
330 for provider_id in route_index.provider_impacts.keys() {
331 provider_ids.insert(provider_id.clone());
332 }
333
334 let usage_by_provider = input
335 .usage_rollup
336 .by_provider
337 .iter()
338 .cloned()
339 .collect::<BTreeMap<_, _>>();
340 let balances_by_provider = group_balances_by_provider(input.provider_balances);
341 let mut endpoint_counts = BTreeMap::<String, (usize, usize, u64)>::new();
342 for endpoint in endpoint_rows {
343 let entry = endpoint_counts
344 .entry(endpoint.provider_id.clone())
345 .or_insert((0, 0, 0));
346 entry.0 += 1;
347 if endpoint.balance.is_some() {
348 entry.1 += 1;
349 }
350 entry.2 = entry.2.saturating_add(endpoint.usage.requests_total);
351 }
352
353 let mut rows = provider_ids
354 .into_iter()
355 .map(|provider_id| {
356 let usage = usage_by_provider
357 .get(provider_id.as_str())
358 .cloned()
359 .unwrap_or_default();
360 let balances = balances_by_provider
361 .get(provider_id.as_str())
362 .cloned()
363 .unwrap_or_default();
364 let balance_counts = balance_counts_for_snapshots(&balances, input.generated_at_ms);
365 let primary_balance = primary_balance_snapshot(&balances, input.generated_at_ms)
366 .map(|snapshot| summarize_snapshot(snapshot, input.generated_at_ms));
367 let latest_balance_error = latest_balance_error(&balances);
368 let balance_age_ms = primary_balance.as_ref().and_then(|summary| summary.age_ms);
369 let balance_status = balance_counts.aggregate_status();
370 let routing = route_index
371 .provider_impacts
372 .get(provider_id.as_str())
373 .cloned()
374 .unwrap_or_else(|| UsageBalanceRouteImpact {
375 provider_id: provider_id.clone(),
376 ..UsageBalanceRouteImpact::default()
377 });
378 let (endpoint_count, endpoints_with_balance, recent_endpoint_requests) =
379 endpoint_counts
380 .get(provider_id.as_str())
381 .copied()
382 .unwrap_or_default();
383
384 UsageBalanceProviderRow {
385 display_name: provider_id.clone(),
386 provider_id,
387 success_per_mille: per_mille(
388 usage.requests_total.saturating_sub(usage.requests_error),
389 usage.requests_total,
390 ),
391 output_tokens_per_second: output_tokens_per_second(&usage),
392 avg_ttfb_ms: (usage.ttfb_samples > 0)
393 .then(|| usage.ttfb_ms_total / usage.ttfb_samples),
394 cost_total_usd: usage.cost.total_cost_usd.clone(),
395 cost_display: usage.cost.display_total(),
396 cost_confidence: usage.cost.confidence,
397 usage,
398 balance_status,
399 balance_counts,
400 primary_balance,
401 latest_balance_error,
402 balance_age_ms,
403 routing,
404 endpoint_count,
405 endpoints_with_balance,
406 recent_endpoint_requests,
407 }
408 })
409 .collect::<Vec<_>>();
410
411 sort_provider_rows(&mut rows);
412 rows
413}
414
415fn build_refresh_status(
416 provider_balances: &HashMap<String, Vec<ProviderBalanceSnapshot>>,
417 now_ms: u64,
418 input: UsageBalanceRefreshInput,
419) -> UsageBalanceRefreshStatus {
420 let mut out = UsageBalanceRefreshStatus {
421 refreshing: input.refreshing,
422 last_message: input.last_message,
423 last_error: input.last_error,
424 last_provider_refresh: input.last_provider_refresh,
425 ..UsageBalanceRefreshStatus::default()
426 };
427 let mut latest_error_fetched_at_ms: Option<u64> = None;
428
429 for snapshot in provider_balances.values().flatten() {
430 out.total_snapshots += 1;
431 out.latest_fetched_at_ms = Some(
432 out.latest_fetched_at_ms
433 .unwrap_or(0)
434 .max(snapshot.fetched_at_ms),
435 );
436 let status = classify_snapshot(snapshot, now_ms);
437 out.status_counts.record(status);
438 if let Some(error) = snapshot
439 .error
440 .as_deref()
441 .map(str::trim)
442 .filter(|v| !v.is_empty())
443 && latest_error_fetched_at_ms.is_none_or(|latest| snapshot.fetched_at_ms >= latest)
444 {
445 out.latest_error = Some(error.to_string());
446 out.latest_error_provider_id = non_empty(snapshot.provider_id.clone())
447 .or_else(|| non_empty(snapshot.station_name.clone().unwrap_or_default()));
448 latest_error_fetched_at_ms = Some(snapshot.fetched_at_ms);
449 }
450 }
451
452 out
453}
454
455fn build_endpoint_rows(
456 service_name: &str,
457 recent: &[FinishedRequest],
458 provider_balances: &HashMap<String, Vec<ProviderBalanceSnapshot>>,
459 route_index: &RouteIndex,
460 now_ms: u64,
461) -> Vec<UsageBalanceEndpointRow> {
462 let mut acc = BTreeMap::<EndpointKey, EndpointAccum>::new();
463
464 for candidate in route_index.candidates.values() {
465 let key = EndpointKey {
466 provider_id: candidate.provider_id.clone(),
467 endpoint_id: candidate.endpoint_id.clone(),
468 };
469 let row = acc
470 .entry(key.clone())
471 .or_insert_with(|| EndpointAccum::new(service_name, key));
472 row.provider_endpoint_key = candidate.provider_endpoint_key.clone();
473 row.base_url = non_empty(candidate.base_url.clone()).or(row.base_url.take());
474 row.route_selected |= candidate.selected;
475 row.route_skip_reasons
476 .extend(candidate.skip_reasons.iter().cloned());
477 }
478
479 for request in recent {
480 let Some(provider_id) = request
481 .provider_id
482 .as_deref()
483 .map(str::trim)
484 .filter(|value| !value.is_empty())
485 else {
486 continue;
487 };
488 let base_url = request
489 .upstream_base_url
490 .as_deref()
491 .map(str::trim)
492 .filter(|value| !value.is_empty());
493 let endpoint_id = base_url
494 .and_then(|base_url| {
495 route_index
496 .endpoint_by_provider_base_url
497 .get(&(provider_id.to_string(), base_url.to_string()))
498 .cloned()
499 })
500 .unwrap_or_else(|| {
501 base_url
502 .map(endpoint_id_from_base_url)
503 .unwrap_or_else(|| "unknown_endpoint".to_string())
504 });
505 let key = EndpointKey {
506 provider_id: provider_id.to_string(),
507 endpoint_id,
508 };
509 let row = acc
510 .entry(key.clone())
511 .or_insert_with(|| EndpointAccum::new(service_name, key));
512 if row.base_url.is_none() {
513 row.base_url = base_url.map(ToOwned::to_owned);
514 }
515 record_finished_into_bucket(&mut row.usage, request);
516 }
517
518 for (station_name, snapshots) in provider_balances {
519 for snapshot in snapshots {
520 let provider_id = balance_provider_id(station_name, snapshot);
521 let endpoint_id = snapshot_endpoint_id(snapshot, &provider_id, route_index)
522 .or_else(|| snapshot.upstream_index.map(|idx| format!("upstream#{idx}")))
523 .or_else(|| non_empty(snapshot.source.clone()))
524 .unwrap_or_else(|| "balance".to_string());
525 let key = EndpointKey {
526 provider_id: provider_id.clone(),
527 endpoint_id,
528 };
529 let row = acc
530 .entry(key.clone())
531 .or_insert_with(|| EndpointAccum::new(service_name, key));
532 if let Some(provider_endpoint_key) =
533 non_empty(snapshot.provider_endpoint_key.clone().unwrap_or_default())
534 {
535 row.provider_endpoint_key = provider_endpoint_key;
536 }
537 let summary = summarize_snapshot(snapshot, now_ms);
538 let replace = row.balance.as_ref().is_none_or(|existing| {
539 snapshot_display_rank(summary.status) < snapshot_display_rank(existing.status)
540 });
541 if replace {
542 row.balance = Some(summary);
543 }
544 }
545 }
546
547 acc.into_values().map(EndpointAccum::finish).collect()
548}
549
550fn snapshot_endpoint_id(
551 snapshot: &ProviderBalanceSnapshot,
552 provider_id: &str,
553 route_index: &RouteIndex,
554) -> Option<String> {
555 let provider_endpoint_key = snapshot
556 .provider_endpoint_key
557 .as_deref()
558 .map(str::trim)
559 .filter(|value| !value.is_empty())?;
560 route_index
561 .endpoint_by_provider_endpoint_key
562 .get(provider_endpoint_key)
563 .cloned()
564 .or_else(|| endpoint_id_from_provider_endpoint_key(provider_endpoint_key, provider_id))
565}
566
567fn endpoint_id_from_provider_endpoint_key(key: &str, provider_id: &str) -> Option<String> {
568 let mut parts = key.splitn(3, '/');
569 let _service_name = parts.next()?;
570 let provider = parts.next()?;
571 let endpoint = parts.next()?.trim();
572 (provider == provider_id && !endpoint.is_empty()).then(|| endpoint.to_string())
573}
574
575fn record_finished_into_bucket(bucket: &mut UsageBucket, request: &FinishedRequest) {
576 bucket.requests_total = bucket.requests_total.saturating_add(1);
577 if request.status_code >= 400 {
578 bucket.requests_error = bucket.requests_error.saturating_add(1);
579 }
580 bucket.duration_ms_total = bucket.duration_ms_total.saturating_add(request.duration_ms);
581
582 let Some(usage) = request.usage.as_ref() else {
583 return;
584 };
585
586 bucket.usage.add_assign(usage);
587 bucket.cost.record_usage_cost(&request.cost);
588 bucket.requests_with_usage = bucket.requests_with_usage.saturating_add(1);
589 bucket.duration_ms_with_usage_total = bucket
590 .duration_ms_with_usage_total
591 .saturating_add(request.duration_ms);
592 let observability = request.observability_view();
593 let generation_ms = observability.generation_ms.unwrap_or(request.duration_ms);
594 bucket.generation_ms_total = bucket.generation_ms_total.saturating_add(generation_ms);
595 if let Some(ttfb) = observability.ttfb_ms.filter(|value| *value > 0) {
596 bucket.ttfb_ms_total = bucket.ttfb_ms_total.saturating_add(ttfb);
597 bucket.ttfb_samples = bucket.ttfb_samples.saturating_add(1);
598 }
599}
600
601fn group_balances_by_provider(
602 provider_balances: &HashMap<String, Vec<ProviderBalanceSnapshot>>,
603) -> BTreeMap<String, Vec<&ProviderBalanceSnapshot>> {
604 let mut out = BTreeMap::<String, Vec<&ProviderBalanceSnapshot>>::new();
605 for (station_name, snapshots) in provider_balances {
606 for snapshot in snapshots {
607 out.entry(balance_provider_id(station_name, snapshot))
608 .or_default()
609 .push(snapshot);
610 }
611 }
612 out
613}
614
615fn balance_provider_id(station_name: &str, snapshot: &ProviderBalanceSnapshot) -> String {
616 if snapshot.provider_id.trim().is_empty() {
617 station_name.to_string()
618 } else {
619 snapshot.provider_id.trim().to_string()
620 }
621}
622
623fn balance_counts_for_snapshots(
624 snapshots: &[&ProviderBalanceSnapshot],
625 now_ms: u64,
626) -> UsageBalanceStatusCounts {
627 let mut counts = UsageBalanceStatusCounts::default();
628 for snapshot in snapshots {
629 counts.record(classify_snapshot(snapshot, now_ms));
630 }
631 counts
632}
633
634fn classify_snapshot(snapshot: &ProviderBalanceSnapshot, now_ms: u64) -> UsageBalanceStatus {
635 match snapshot.status_at(now_ms) {
636 BalanceSnapshotStatus::Ok if snapshot.unlimited_quota == Some(true) => {
637 UsageBalanceStatus::Unlimited
638 }
639 BalanceSnapshotStatus::Ok => UsageBalanceStatus::Ok,
640 BalanceSnapshotStatus::Exhausted => UsageBalanceStatus::Exhausted,
641 BalanceSnapshotStatus::Stale => UsageBalanceStatus::Stale,
642 BalanceSnapshotStatus::Error => UsageBalanceStatus::Error,
643 BalanceSnapshotStatus::Unknown => UsageBalanceStatus::Unknown,
644 }
645}
646
647fn summarize_snapshot(
648 snapshot: &ProviderBalanceSnapshot,
649 now_ms: u64,
650) -> UsageBalanceSnapshotSummary {
651 UsageBalanceSnapshotSummary {
652 provider_id: snapshot.provider_id.clone(),
653 provider_endpoint_key: snapshot.provider_endpoint_key.clone(),
654 source: snapshot.source.clone(),
655 status: classify_snapshot(snapshot, now_ms),
656 amount_summary: snapshot.amount_summary(),
657 fetched_at_ms: snapshot.fetched_at_ms,
658 stale_after_ms: snapshot.stale_after_ms,
659 age_ms: (now_ms >= snapshot.fetched_at_ms).then(|| now_ms - snapshot.fetched_at_ms),
660 error: snapshot.error.clone(),
661 exhaustion_affects_routing: snapshot.exhaustion_affects_routing,
662 routing_exhausted: snapshot.exhaustion_affects_routing
663 && snapshot.status_at(now_ms) == BalanceSnapshotStatus::Exhausted,
664 routing_ignored_exhaustion: !snapshot.exhaustion_affects_routing
665 && snapshot.status_at(now_ms) == BalanceSnapshotStatus::Exhausted,
666 }
667}
668
669fn primary_balance_snapshot<'a>(
670 snapshots: &'a [&ProviderBalanceSnapshot],
671 now_ms: u64,
672) -> Option<&'a ProviderBalanceSnapshot> {
673 snapshots.iter().copied().min_by(|left, right| {
674 snapshot_display_rank(classify_snapshot(left, now_ms))
675 .cmp(&snapshot_display_rank(classify_snapshot(right, now_ms)))
676 .then_with(|| left.upstream_index.cmp(&right.upstream_index))
677 .then_with(|| right.fetched_at_ms.cmp(&left.fetched_at_ms))
678 })
679}
680
681fn snapshot_display_rank(status: UsageBalanceStatus) -> u8 {
682 match status {
683 UsageBalanceStatus::Ok | UsageBalanceStatus::Unlimited => 0,
684 UsageBalanceStatus::Stale => 1,
685 UsageBalanceStatus::Unknown | UsageBalanceStatus::Error => 2,
686 UsageBalanceStatus::Exhausted => 3,
687 }
688}
689
690fn latest_balance_error(snapshots: &[&ProviderBalanceSnapshot]) -> Option<String> {
691 snapshots
692 .iter()
693 .filter_map(|snapshot| {
694 snapshot
695 .error
696 .as_deref()
697 .map(str::trim)
698 .filter(|value| !value.is_empty())
699 .map(|error| (snapshot.fetched_at_ms, error.to_string()))
700 })
701 .max_by_key(|(fetched_at_ms, _)| *fetched_at_ms)
702 .map(|(_, error)| error)
703}
704
705fn per_mille(num: u64, den: u64) -> Option<u16> {
706 (den > 0).then(|| ((num.saturating_mul(1000) / den).min(1000)) as u16)
707}
708
709fn output_tokens_per_second(bucket: &UsageBucket) -> Option<u64> {
710 let output = bucket.usage.output_tokens.max(0) as u64;
711 if output == 0 || bucket.generation_ms_total == 0 {
712 return None;
713 }
714 Some(output.saturating_mul(1000) / bucket.generation_ms_total)
715}
716
717fn sort_provider_rows(rows: &mut [UsageBalanceProviderRow]) {
718 rows.sort_by(|left, right| {
719 provider_selected_rank(right)
720 .cmp(&provider_selected_rank(left))
721 .then_with(|| {
722 provider_attention_rank(left.balance_status)
723 .cmp(&provider_attention_rank(right.balance_status))
724 })
725 .then_with(|| provider_cost_sort_value(right).cmp(&provider_cost_sort_value(left)))
726 .then_with(|| right.usage.requests_total.cmp(&left.usage.requests_total))
727 .then_with(|| left.provider_id.cmp(&right.provider_id))
728 });
729}
730
731fn provider_selected_rank(row: &UsageBalanceProviderRow) -> u8 {
732 u8::from(row.routing.selected)
733}
734
735fn provider_attention_rank(status: UsageBalanceStatus) -> u8 {
736 match status {
737 UsageBalanceStatus::Error => 0,
738 UsageBalanceStatus::Exhausted => 1,
739 UsageBalanceStatus::Stale => 2,
740 UsageBalanceStatus::Unknown => 3,
741 UsageBalanceStatus::Unlimited => 4,
742 UsageBalanceStatus::Ok => 5,
743 }
744}
745
746fn provider_cost_sort_value(row: &UsageBalanceProviderRow) -> i128 {
747 row.cost_total_usd
748 .as_deref()
749 .and_then(UsdAmount::from_decimal_str)
750 .map(UsdAmount::femto_usd)
751 .unwrap_or(0)
752}
753
754fn sort_endpoint_rows(rows: &mut [UsageBalanceEndpointRow]) {
755 rows.sort_by(|left, right| {
756 u8::from(right.route_selected)
757 .cmp(&u8::from(left.route_selected))
758 .then_with(|| {
759 provider_attention_rank(left.balance_status)
760 .cmp(&provider_attention_rank(right.balance_status))
761 })
762 .then_with(|| right.usage.requests_total.cmp(&left.usage.requests_total))
763 .then_with(|| left.provider_id.cmp(&right.provider_id))
764 .then_with(|| left.endpoint_id.cmp(&right.endpoint_id))
765 });
766}
767
768#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
769struct EndpointKey {
770 provider_id: String,
771 endpoint_id: String,
772}
773
774#[derive(Debug, Clone)]
775struct EndpointAccum {
776 key: EndpointKey,
777 provider_endpoint_key: String,
778 base_url: Option<String>,
779 usage: UsageBucket,
780 balance: Option<UsageBalanceSnapshotSummary>,
781 route_selected: bool,
782 route_skip_reasons: BTreeSet<String>,
783}
784
785impl EndpointAccum {
786 fn new(service_name: &str, key: EndpointKey) -> Self {
787 let provider_endpoint_key =
788 stable_provider_endpoint_key(service_name, &key.provider_id, &key.endpoint_id);
789 Self {
790 key,
791 provider_endpoint_key,
792 base_url: None,
793 usage: UsageBucket::default(),
794 balance: None,
795 route_selected: false,
796 route_skip_reasons: BTreeSet::new(),
797 }
798 }
799
800 fn finish(self) -> UsageBalanceEndpointRow {
801 let balance_status = self
802 .balance
803 .as_ref()
804 .map(|balance| balance.status)
805 .unwrap_or(UsageBalanceStatus::Unknown);
806 UsageBalanceEndpointRow {
807 provider_id: self.key.provider_id,
808 endpoint_id: self.key.endpoint_id,
809 provider_endpoint_key: self.provider_endpoint_key,
810 base_url: self.base_url,
811 usage: self.usage,
812 balance_status,
813 balance: self.balance,
814 route_selected: self.route_selected,
815 route_skip_reasons: self.route_skip_reasons.into_iter().collect(),
816 }
817 }
818}
819
820#[derive(Debug, Clone)]
821struct RouteCandidateView {
822 provider_id: String,
823 endpoint_id: String,
824 provider_endpoint_key: String,
825 base_url: String,
826 selected: bool,
827 skip_reasons: Vec<String>,
828}
829
830#[derive(Debug, Clone, Default)]
831struct RouteIndex {
832 candidates: BTreeMap<String, RouteCandidateView>,
833 provider_impacts: BTreeMap<String, UsageBalanceRouteImpact>,
834 endpoint_by_provider_base_url: BTreeMap<(String, String), String>,
835 endpoint_by_provider_endpoint_key: BTreeMap<String, String>,
836}
837
838impl RouteIndex {
839 fn from_explain(explain: Option<&RoutingExplainResponse>) -> Self {
840 let mut out = Self::default();
841 let Some(explain) = explain else {
842 return out;
843 };
844
845 for candidate in &explain.candidates {
846 out.record_candidate(candidate);
847 }
848 for impact in out.provider_impacts.values_mut() {
849 impact.skip_reasons.sort();
850 impact.skip_reasons.dedup();
851 impact.route_paths.sort();
852 impact.route_paths.dedup();
853 }
854 out
855 }
856
857 fn record_candidate(&mut self, candidate: &RoutingExplainCandidate) {
858 let skip_reasons = candidate
859 .skip_reasons
860 .iter()
861 .map(|reason| reason.code().to_string())
862 .collect::<Vec<_>>();
863 let view = RouteCandidateView {
864 provider_id: candidate.provider_id.clone(),
865 endpoint_id: candidate.endpoint_id.clone(),
866 provider_endpoint_key: candidate.provider_endpoint_key.clone(),
867 base_url: candidate.upstream_base_url.clone(),
868 selected: candidate.selected,
869 skip_reasons: skip_reasons.clone(),
870 };
871
872 self.endpoint_by_provider_base_url.insert(
873 (
874 candidate.provider_id.clone(),
875 candidate.upstream_base_url.clone(),
876 ),
877 candidate.endpoint_id.clone(),
878 );
879 self.endpoint_by_provider_endpoint_key.insert(
880 candidate.provider_endpoint_key.clone(),
881 candidate.endpoint_id.clone(),
882 );
883
884 let impact = self
885 .provider_impacts
886 .entry(candidate.provider_id.clone())
887 .or_insert_with(|| UsageBalanceRouteImpact {
888 provider_id: candidate.provider_id.clone(),
889 ..UsageBalanceRouteImpact::default()
890 });
891 impact.candidate_count += 1;
892 impact.route_paths.push(candidate.route_path.clone());
893 impact.skip_reasons.extend(skip_reasons);
894 if candidate.selected {
895 impact.selected = true;
896 impact.selected_endpoint_id = Some(candidate.endpoint_id.clone());
897 impact.selected_provider_endpoint_key = Some(candidate.provider_endpoint_key.clone());
898 }
899
900 self.candidates
901 .insert(candidate.provider_endpoint_key.clone(), view);
902 }
903}
904
905fn stable_provider_endpoint_key(
906 service_name: &str,
907 provider_id: &str,
908 endpoint_id: &str,
909) -> String {
910 ProviderEndpointKey::new(service_name, provider_id, endpoint_id).stable_key()
911}
912
913fn endpoint_id_from_base_url(base_url: &str) -> String {
914 let trimmed = base_url.trim();
915 let after_scheme = trimmed
916 .split_once("://")
917 .map(|(_, rest)| rest)
918 .unwrap_or(trimmed);
919 after_scheme
920 .split('/')
921 .next()
922 .map(str::trim)
923 .filter(|value| !value.is_empty())
924 .unwrap_or("endpoint")
925 .to_string()
926}
927
928fn non_empty(value: String) -> Option<String> {
929 (!value.trim().is_empty()).then_some(value)
930}
931
932#[cfg(test)]
933mod tests {
934 use super::*;
935 use crate::pricing::CostBreakdown;
936 use crate::routing_explain::{
937 RoutingExplainCandidate, RoutingExplainResponse, RoutingExplainSkipReason,
938 };
939 use crate::usage::UsageMetrics;
940
941 fn build_view(
942 usage_rollup: &UsageRollupView,
943 provider_balances: &HashMap<String, Vec<ProviderBalanceSnapshot>>,
944 recent: &[FinishedRequest],
945 routing_explain: Option<&RoutingExplainResponse>,
946 ) -> UsageBalanceView {
947 UsageBalanceView::build(UsageBalanceBuildInput {
948 service_name: "codex",
949 window_days: 7,
950 generated_at_ms: 1_000,
951 usage_rollup,
952 provider_balances,
953 recent,
954 routing_explain,
955 refresh: UsageBalanceRefreshInput::default(),
956 })
957 }
958
959 #[test]
960 fn balance_status_counts_keep_unknown_stale_exhausted_error_and_unlimited_distinct() {
961 let mut balances = HashMap::new();
962 balances.insert(
963 "station".to_string(),
964 vec![
965 ProviderBalanceSnapshot {
966 provider_id: "ok".to_string(),
967 exhausted: Some(false),
968 total_balance_usd: Some("3".to_string()),
969 fetched_at_ms: 900,
970 ..ProviderBalanceSnapshot::default()
971 },
972 ProviderBalanceSnapshot {
973 provider_id: "unlimited".to_string(),
974 exhausted: Some(false),
975 unlimited_quota: Some(true),
976 fetched_at_ms: 900,
977 ..ProviderBalanceSnapshot::default()
978 },
979 ProviderBalanceSnapshot {
980 provider_id: "stale".to_string(),
981 exhausted: Some(false),
982 stale_after_ms: Some(950),
983 fetched_at_ms: 900,
984 ..ProviderBalanceSnapshot::default()
985 },
986 ProviderBalanceSnapshot {
987 provider_id: "exhausted".to_string(),
988 exhausted: Some(true),
989 fetched_at_ms: 900,
990 ..ProviderBalanceSnapshot::default()
991 },
992 ProviderBalanceSnapshot {
993 provider_id: "error".to_string(),
994 error: Some("lookup failed".to_string()),
995 fetched_at_ms: 900,
996 ..ProviderBalanceSnapshot::default()
997 },
998 ProviderBalanceSnapshot {
999 provider_id: "unknown".to_string(),
1000 fetched_at_ms: 900,
1001 ..ProviderBalanceSnapshot::default()
1002 },
1003 ],
1004 );
1005
1006 let view = build_view(&UsageRollupView::default(), &balances, &[], None);
1007
1008 assert_eq!(view.totals.balance_status_counts.ok, 1);
1009 assert_eq!(view.totals.balance_status_counts.unlimited, 1);
1010 assert_eq!(view.totals.balance_status_counts.stale, 1);
1011 assert_eq!(view.totals.balance_status_counts.exhausted, 1);
1012 assert_eq!(view.totals.balance_status_counts.error, 1);
1013 assert_eq!(view.totals.balance_status_counts.unknown, 1);
1014 assert_eq!(
1015 view.provider_rows
1016 .iter()
1017 .find(|row| row.provider_id == "unlimited")
1018 .map(|row| row.balance_status),
1019 Some(UsageBalanceStatus::Unlimited)
1020 );
1021 assert_eq!(
1022 view.provider_rows
1023 .iter()
1024 .find(|row| row.provider_id == "unknown")
1025 .map(|row| row.balance_status),
1026 Some(UsageBalanceStatus::Unknown)
1027 );
1028 }
1029
1030 #[test]
1031 fn provider_rows_prefer_route_selection_then_attention_and_usage() {
1032 let rollup = UsageRollupView {
1033 by_provider: vec![
1034 (
1035 "cheap".to_string(),
1036 UsageBucket {
1037 requests_total: 2,
1038 ..UsageBucket::default()
1039 },
1040 ),
1041 (
1042 "selected".to_string(),
1043 UsageBucket {
1044 requests_total: 1,
1045 ..UsageBucket::default()
1046 },
1047 ),
1048 ],
1049 ..UsageRollupView::default()
1050 };
1051 let explain = RoutingExplainResponse {
1052 api_version: 1,
1053 service_name: "codex".to_string(),
1054 runtime_loaded_at_ms: None,
1055 request_model: None,
1056 session_id: None,
1057 request_context: Default::default(),
1058 selected_route: None,
1059 candidates: vec![RoutingExplainCandidate {
1060 provider_id: "selected".to_string(),
1061 provider_alias: None,
1062 endpoint_id: "default".to_string(),
1063 provider_endpoint_key: "codex/selected/default".to_string(),
1064 route_path: vec!["main".to_string(), "selected".to_string()],
1065 preference_group: 0,
1066 upstream_base_url: "https://selected.example/v1".to_string(),
1067 capacity: Default::default(),
1068 availability: crate::routing_explain::RoutingExplainAvailability {
1069 available: true,
1070 runtime_available: true,
1071 routable_except_usage: true,
1072 ..Default::default()
1073 },
1074 selected: true,
1075 skip_reasons: Vec::new(),
1076 }],
1077 affinity_policy: "off".to_string(),
1078 affinity: None,
1079 conditional_routes: Vec::new(),
1080 };
1081
1082 let view = build_view(&rollup, &HashMap::new(), &[], Some(&explain));
1083
1084 assert_eq!(view.provider_rows[0].provider_id, "selected");
1085 assert!(view.provider_rows[0].routing.selected);
1086 }
1087
1088 #[test]
1089 fn endpoint_rows_merge_recent_usage_balance_and_route_skip_reasons() {
1090 let recent = vec![FinishedRequest {
1091 id: 1,
1092 trace_id: None,
1093 session_id: None,
1094 session_identity_source: None,
1095 client_name: None,
1096 client_addr: None,
1097 cwd: None,
1098 model: Some("gpt-test".to_string()),
1099 reasoning_effort: None,
1100 service_tier: None,
1101 station_name: Some("station-a".to_string()),
1102 provider_id: Some("right".to_string()),
1103 upstream_base_url: Some("https://right.example/v1".to_string()),
1104 route_decision: None,
1105 usage: Some(UsageMetrics {
1106 input_tokens: 10,
1107 output_tokens: 20,
1108 total_tokens: 30,
1109 ..UsageMetrics::default()
1110 }),
1111 cost: CostBreakdown::unknown(),
1112 retry: None,
1113 provider_signals: Vec::new(),
1114 policy_actions: Vec::new(),
1115 observability: Default::default(),
1116 service: "codex".to_string(),
1117 method: "POST".to_string(),
1118 path: "/v1/responses".to_string(),
1119 status_code: 429,
1120 duration_ms: 120,
1121 ttfb_ms: Some(20),
1122 streaming: false,
1123 ended_at_ms: 950,
1124 }];
1125 let mut balances = HashMap::new();
1126 balances.insert(
1127 "station-a".to_string(),
1128 vec![
1129 ProviderBalanceSnapshot {
1130 provider_id: "right".to_string(),
1131 station_name: Some("station-a".to_string()),
1132 upstream_index: Some(0),
1133 exhausted: Some(true),
1134 fetched_at_ms: 940,
1135 ..ProviderBalanceSnapshot::default()
1136 }
1137 .with_provider_endpoint_key("codex/right/default"),
1138 ],
1139 );
1140 let explain = RoutingExplainResponse {
1141 api_version: 1,
1142 service_name: "codex".to_string(),
1143 runtime_loaded_at_ms: None,
1144 request_model: None,
1145 session_id: None,
1146 request_context: Default::default(),
1147 selected_route: None,
1148 candidates: vec![RoutingExplainCandidate {
1149 provider_id: "right".to_string(),
1150 provider_alias: None,
1151 endpoint_id: "default".to_string(),
1152 provider_endpoint_key: "codex/right/default".to_string(),
1153 route_path: vec!["main".to_string(), "right".to_string()],
1154 preference_group: 0,
1155 upstream_base_url: "https://right.example/v1".to_string(),
1156 capacity: Default::default(),
1157 availability: crate::routing_explain::RoutingExplainAvailability {
1158 available: false,
1159 runtime_available: false,
1160 routable_except_usage: true,
1161 usage_exhausted: true,
1162 dominant_reason: Some(RoutingExplainSkipReason::UsageExhausted),
1163 ..Default::default()
1164 },
1165 selected: false,
1166 skip_reasons: vec![RoutingExplainSkipReason::UsageExhausted],
1167 }],
1168 affinity_policy: "off".to_string(),
1169 affinity: None,
1170 conditional_routes: Vec::new(),
1171 };
1172
1173 let view = build_view(
1174 &UsageRollupView::default(),
1175 &balances,
1176 &recent,
1177 Some(&explain),
1178 );
1179 let endpoint = view
1180 .endpoint_rows
1181 .iter()
1182 .find(|row| row.provider_id == "right")
1183 .expect("right endpoint");
1184
1185 assert_eq!(endpoint.endpoint_id, "default");
1186 assert_eq!(endpoint.provider_endpoint_key, "codex/right/default");
1187 assert_eq!(endpoint.usage.requests_total, 1);
1188 assert_eq!(endpoint.usage.requests_error, 1);
1189 assert_eq!(endpoint.balance_status, UsageBalanceStatus::Exhausted);
1190 assert_eq!(
1191 endpoint
1192 .balance
1193 .as_ref()
1194 .and_then(|balance| balance.provider_endpoint_key.as_deref()),
1195 Some("codex/right/default")
1196 );
1197 assert_eq!(endpoint.route_skip_reasons, vec!["usage_exhausted"]);
1198 }
1199}