codetether_agent/telemetry/provider/
metrics.rs

1//! Rolling 1000-entry buffer of provider requests, plus per-provider aggregation.
2
3use std::collections::HashMap;
4use tokio::sync::Mutex;
5
6use super::request::ProviderRequestRecord;
7use super::snapshot::ProviderSnapshot;
8use super::stats::{mean, percentile, sort_f64};
9
10/// Bounded history of provider requests. Oldest entries are evicted once
11/// the buffer exceeds 1000 records.
12///
13/// Prefer the [`crate::telemetry::PROVIDER_METRICS`] singleton. Constructing
14/// your own is only useful in tests.
15#[derive(Debug, Default)]
16pub struct ProviderMetrics {
17    requests: Mutex<Vec<ProviderRequestRecord>>,
18}
19
20impl ProviderMetrics {
21    /// Build an empty buffer.
22    pub fn new() -> Self {
23        Self::default()
24    }
25
26    /// Append a request, evicting the oldest if the buffer exceeds 1000.
27    pub async fn record(&self, record: ProviderRequestRecord) {
28        let mut requests = self.requests.lock().await;
29        requests.push(record);
30        if requests.len() > 1000 {
31            requests.remove(0);
32        }
33    }
34
35    /// Return up to `limit` most-recent records (newest first).
36    pub async fn get_recent(&self, limit: usize) -> Vec<ProviderRequestRecord> {
37        let requests = self.requests.lock().await;
38        requests.iter().rev().take(limit).cloned().collect()
39    }
40
41    /// Aggregate the buffer into one [`ProviderSnapshot`] per provider.
42    /// Returns an empty `Vec` if the buffer lock is contended.
43    pub fn all_snapshots(&self) -> Vec<ProviderSnapshot> {
44        let Ok(guard) = self.requests.try_lock() else {
45            return Vec::new();
46        };
47        let requests = guard.clone();
48        drop(guard);
49        aggregate_by_provider(requests)
50    }
51}
52
53/// Group `requests` by provider and fold each group into a [`ProviderSnapshot`].
54fn aggregate_by_provider(requests: Vec<ProviderRequestRecord>) -> Vec<ProviderSnapshot> {
55    let mut by_provider: HashMap<String, Vec<ProviderRequestRecord>> = HashMap::new();
56    for req in requests {
57        by_provider
58            .entry(req.provider.clone())
59            .or_default()
60            .push(req);
61    }
62    by_provider
63        .into_iter()
64        .filter(|(_, reqs)| !reqs.is_empty())
65        .map(|(provider, reqs)| snapshot_for(provider, &reqs))
66        .collect()
67}
68
69/// Build one [`ProviderSnapshot`] from an already-grouped `reqs` slice.
70fn snapshot_for(provider: String, reqs: &[ProviderRequestRecord]) -> ProviderSnapshot {
71    let request_count = reqs.len();
72    let total_input_tokens: u64 = reqs.iter().map(|r| r.input_tokens).sum();
73    let total_output_tokens: u64 = reqs.iter().map(|r| r.output_tokens).sum();
74    let avg_latency_ms = mean(&reqs.iter().map(|r| r.latency_ms as f64).collect::<Vec<_>>());
75
76    let mut tps: Vec<f64> = reqs.iter().map(|r| r.tokens_per_second()).collect();
77    sort_f64(&mut tps);
78    let mut lat: Vec<f64> = reqs.iter().map(|r| r.latency_ms as f64).collect();
79    sort_f64(&mut lat);
80
81    ProviderSnapshot {
82        provider,
83        request_count,
84        total_input_tokens,
85        total_output_tokens,
86        avg_tps: mean(&tps),
87        avg_latency_ms,
88        p50_tps: percentile(&tps, 0.50),
89        p50_latency_ms: percentile(&lat, 0.50),
90        p95_tps: percentile(&tps, 0.95),
91        p95_latency_ms: percentile(&lat, 0.95),
92    }
93}
codetether_agent/telemetry/provider/metrics.rs

codetether_agent/telemetry/provider/
metrics.rs
