agent_engine/runtime/
telemetry.rs

1//! Structured per-request API telemetry.
2//!
3//! Opt-in via the `telemetry` config key (`off` | `basic` | `full`).
4//! Writes one JSON record per API call to `~/.cache/synaps/api-log.jsonl`
5//! (mode 0600, O_NOFOLLOW — same hardening as the legacy usage log).
6//!
7//! `basic` records timing + usage + cost. `full` additionally records
8//! rate-limit headers and cache-diagnostics results when available.
9//!
10//! Writes are best-effort: a broken log path must never break the request
11//! pipeline. All errors are silently dropped (matching `log_usage`).
12
13use serde::Serialize;
14use std::time::{Duration, SystemTime, UNIX_EPOCH};
15
16/// Telemetry verbosity level, parsed from the `telemetry` config key.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18pub enum TelemetryLevel {
19    #[default]
20    Off,
21    Basic,
22    Full,
23}
24
25impl TelemetryLevel {
26    pub fn from_str_key(s: &str) -> Self {
27        match s.trim().to_ascii_lowercase().as_str() {
28            "basic" | "on" | "1" | "true" => Self::Basic,
29            "full" => Self::Full,
30            _ => Self::Off,
31        }
32    }
33
34    pub fn enabled(&self) -> bool {
35        !matches!(self, Self::Off)
36    }
37
38    pub fn as_str(&self) -> &'static str {
39        match self {
40            Self::Off => "off",
41            Self::Basic => "basic",
42            Self::Full => "full",
43        }
44    }
45}
46
47/// Token usage for one API call, including the cache-creation TTL breakdown.
48#[derive(Debug, Clone, Default, Serialize)]
49pub struct UsageRecord {
50    pub input: u64,
51    pub output: u64,
52    pub cache_read: u64,
53    pub cache_write: u64,
54    /// Cache writes with 5-minute TTL (from `usage.cache_creation.ephemeral_5m_input_tokens`).
55    #[serde(skip_serializing_if = "Option::is_none")]
56    pub cache_write_5m: Option<u64>,
57    /// Cache writes with 1-hour TTL (from `usage.cache_creation.ephemeral_1h_input_tokens`).
58    #[serde(skip_serializing_if = "Option::is_none")]
59    pub cache_write_1h: Option<u64>,
60    /// Cache hit percentage: cache_read / (input + cache_read + cache_write) * 100.
61    pub hit_pct: f64,
62}
63
64impl UsageRecord {
65    pub fn compute_hit_pct(&mut self) {
66        let total = self.input + self.cache_read + self.cache_write;
67        self.hit_pct = if total > 0 {
68            (self.cache_read as f64 / total as f64 * 1000.0).round() / 10.0
69        } else {
70            0.0
71        };
72    }
73}
74
75/// Rate-limit headroom captured from `anthropic-ratelimit-*` response headers.
76/// Only recorded at `full` level.
77#[derive(Debug, Clone, Default, Serialize)]
78pub struct RateLimitRecord {
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub requests_limit: Option<u64>,
81    #[serde(skip_serializing_if = "Option::is_none")]
82    pub requests_remaining: Option<u64>,
83    #[serde(skip_serializing_if = "Option::is_none")]
84    pub tokens_limit: Option<u64>,
85    #[serde(skip_serializing_if = "Option::is_none")]
86    pub tokens_remaining: Option<u64>,
87    #[serde(skip_serializing_if = "Option::is_none")]
88    pub input_tokens_remaining: Option<u64>,
89    #[serde(skip_serializing_if = "Option::is_none")]
90    pub output_tokens_remaining: Option<u64>,
91    /// RFC 3339 timestamp when the most restrictive token limit resets.
92    #[serde(skip_serializing_if = "Option::is_none")]
93    pub tokens_reset: Option<String>,
94}
95
96impl RateLimitRecord {
97    pub fn is_empty(&self) -> bool {
98        self.requests_limit.is_none()
99            && self.requests_remaining.is_none()
100            && self.tokens_limit.is_none()
101            && self.tokens_remaining.is_none()
102            && self.input_tokens_remaining.is_none()
103            && self.output_tokens_remaining.is_none()
104            && self.tokens_reset.is_none()
105    }
106}
107
108/// Cache-diagnostics result (beta `cache-diagnosis-2026-04-07`).
109/// Only present when the user opted in via `cache_diagnostics = true`.
110#[derive(Debug, Clone, Serialize)]
111pub struct CacheDiagRecord {
112    /// `cache_miss_reason.type` — e.g. "system_changed", "tools_changed".
113    pub miss_reason: String,
114    /// `cache_missed_input_tokens` — estimated tokens lost after divergence.
115    #[serde(skip_serializing_if = "Option::is_none")]
116    pub missed_tokens: Option<u64>,
117}
118
119/// Request-shape context: what we sent, for correlating cache behavior.
120#[derive(Debug, Clone, Default, Serialize)]
121pub struct ContextRecord {
122    pub messages: usize,
123    pub tools: usize,
124    pub system_bytes: usize,
125    /// Indices of user messages carrying a conversational cache_control marker.
126    pub breakpoints: Vec<usize>,
127}
128
129/// One JSONL record per API call.
130#[derive(Debug, Clone, Default, Serialize)]
131pub struct TelemetryRecord {
132    /// Unix epoch milliseconds at request completion.
133    pub ts: u64,
134    #[serde(skip_serializing_if = "Option::is_none")]
135    pub request_id: Option<String>,
136    /// Anthropic message id (`msg_...`) from message_start.
137    #[serde(skip_serializing_if = "Option::is_none")]
138    pub msg_id: Option<String>,
139    pub model: String,
140    /// 1-based attempt number that succeeded (1 = no retries).
141    pub attempt: u32,
142    /// Number of refusal retries consumed (0 = no refusals). Present only when > 0.
143    #[serde(skip_serializing_if = "Option::is_none")]
144    pub refusal_retries_used: Option<u32>,
145    /// Milliseconds from request send to first SSE byte.
146    #[serde(skip_serializing_if = "Option::is_none")]
147    pub ttft_ms: Option<u64>,
148    /// Milliseconds from request send to stream close.
149    pub total_ms: u64,
150    #[serde(skip_serializing_if = "Option::is_none")]
151    pub stop_reason: Option<String>,
152    pub usage: UsageRecord,
153    #[serde(skip_serializing_if = "Option::is_none")]
154    pub ratelimit: Option<RateLimitRecord>,
155    #[serde(skip_serializing_if = "Option::is_none")]
156    pub cache_diag: Option<CacheDiagRecord>,
157    pub context: ContextRecord,
158}
159
160impl TelemetryRecord {
161    pub fn now_ms() -> u64 {
162        SystemTime::now()
163            .duration_since(UNIX_EPOCH)
164            .map(|d| d.as_millis() as u64)
165            .unwrap_or(0)
166    }
167}
168
169/// Parse a `u64` rate-limit header value.
170fn header_u64(headers: &reqwest::header::HeaderMap, name: &str) -> Option<u64> {
171    headers.get(name)?.to_str().ok()?.parse().ok()
172}
173
174/// Parse a string rate-limit header value.
175fn header_string(headers: &reqwest::header::HeaderMap, name: &str) -> Option<String> {
176    Some(headers.get(name)?.to_str().ok()?.to_string())
177}
178
179/// Extract rate-limit headroom from response headers.
180pub fn ratelimit_from_headers(headers: &reqwest::header::HeaderMap) -> RateLimitRecord {
181    RateLimitRecord {
182        requests_limit: header_u64(headers, "anthropic-ratelimit-requests-limit"),
183        requests_remaining: header_u64(headers, "anthropic-ratelimit-requests-remaining"),
184        tokens_limit: header_u64(headers, "anthropic-ratelimit-tokens-limit"),
185        tokens_remaining: header_u64(headers, "anthropic-ratelimit-tokens-remaining"),
186        input_tokens_remaining: header_u64(headers, "anthropic-ratelimit-input-tokens-remaining"),
187        output_tokens_remaining: header_u64(headers, "anthropic-ratelimit-output-tokens-remaining"),
188        tokens_reset: header_string(headers, "anthropic-ratelimit-tokens-reset"),
189    }
190}
191
192/// Extract the `request-id` response header.
193pub fn request_id_from_headers(headers: &reqwest::header::HeaderMap) -> Option<String> {
194    header_string(headers, "request-id")
195}
196
197/// Maximum delay we'll honour from any rate-limit header. A pathological reset
198/// timestamp far in the future would otherwise stall the turn indefinitely.
199pub const RETRY_DELAY_CAP: Duration = Duration::from_secs(60);
200
201/// Compute how long to wait before the next retry, consulting response headers.
202///
203/// Priority order:
204///   1. `retry-after` (integer seconds, or an HTTP-date — integer is what
205///      Anthropic sends in practice, HTTP-date is handled as a best-effort
206///      fallback).
207///   2. `anthropic-ratelimit-tokens-reset` / `anthropic-ratelimit-requests-reset`
208///      (RFC 3339 UTC timestamp) — take the *minimum* of whichever is present.
209///   3. Classic exponential back-off: `1s * 2^(attempt-1)` (1 s, 2 s, 4 s …).
210///
211/// The returned duration is capped at [`RETRY_DELAY_CAP`] so a far-future
212/// reset timestamp never hangs the turn forever. `attempt` is 1-based (first
213/// retry = 1).
214///
215/// Returns `(delay, from_header)` — the bool is `true` when a header was
216/// used (callers can emit a more informative notice message).
217pub fn retry_delay_from_headers(
218    headers: &reqwest::header::HeaderMap,
219    attempt: u32,
220) -> (Duration, bool) {
221    let now_secs = SystemTime::now()
222        .duration_since(UNIX_EPOCH)
223        .map(|d| d.as_secs())
224        .unwrap_or(0);
225
226    // ── 1. retry-after (integer seconds preferred) ──────────────────────────
227    if let Some(ra) = header_string(headers, "retry-after") {
228        let ra = ra.trim();
229        // Integer form: "30"
230        if let Ok(secs) = ra.parse::<u64>() {
231            let d = Duration::from_secs(secs).min(RETRY_DELAY_CAP);
232            return (d, true);
233        }
234        // HTTP-date form: "Wed, 11 Jun 2025 01:46:00 GMT" — parse via chrono.
235        if let Ok(dt) = chrono::DateTime::parse_from_rfc2822(ra) {
236            let reset_secs = dt.timestamp().max(0) as u64;
237            let wait = reset_secs.saturating_sub(now_secs);
238            let d = Duration::from_secs(wait).min(RETRY_DELAY_CAP);
239            return (d, true);
240        }
241    }
242
243    // ── 2. anthropic-ratelimit-*-reset (RFC 3339) ───────────────────────────
244    let mut min_wait: Option<u64> = None;
245    for name in &[
246        "anthropic-ratelimit-tokens-reset",
247        "anthropic-ratelimit-requests-reset",
248    ] {
249        if let Some(ts) = header_string(headers, name) {
250            if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(ts.trim()) {
251                let reset_secs = dt.timestamp().max(0) as u64;
252                let wait = reset_secs.saturating_sub(now_secs);
253                min_wait = Some(min_wait.map_or(wait, |prev| prev.min(wait)));
254            }
255        }
256    }
257    if let Some(wait) = min_wait {
258        let d = Duration::from_secs(wait).min(RETRY_DELAY_CAP);
259        return (d, true);
260    }
261
262    // ── 3. Exponential back-off fallback ────────────────────────────────────
263    let d = Duration::from_millis(1000 * 2u64.pow(attempt.saturating_sub(1)));
264    (d, false)
265}
266
267/// Default telemetry log path: `~/.cache/synaps/api-log.jsonl`.
268fn default_log_path() -> Option<std::path::PathBuf> {
269    let home = std::env::var("HOME").ok()?;
270    Some(std::path::PathBuf::from(home).join(".cache/synaps/api-log.jsonl"))
271}
272
273/// Append a record to the telemetry log. Best-effort — all errors are
274/// silently dropped so a broken log path never breaks the request pipeline.
275///
276/// File is created 0600 with O_NOFOLLOW (CWE-59 hardening, matching
277/// `HelperMethods::log_usage`).
278pub fn write_record(record: &TelemetryRecord) {
279    let Some(path) = default_log_path() else { return };
280    let Ok(line) = serde_json::to_string(record) else { return };
281
282    if let Some(parent) = path.parent() {
283        let _ = std::fs::create_dir_all(parent);
284    }
285
286    // Size-capped rotation: at >50MB, rename to <path>.1 (clobbering any old
287    // .1) before appending. One generation is enough — this is a diagnostic
288    // log, not an audit trail. Errors silently dropped (best-effort contract).
289    const MAX_BYTES: u64 = 50 * 1024 * 1024;
290    if let Ok(meta) = std::fs::metadata(&path) {
291        if meta.len() > MAX_BYTES {
292            let mut rotated = path.as_os_str().to_owned();
293            rotated.push(".1");
294            let _ = std::fs::rename(&path, std::path::PathBuf::from(rotated));
295        }
296    }
297
298    use std::os::unix::fs::OpenOptionsExt;
299
300    let result = std::fs::OpenOptions::new()
301        .create(true)
302        .append(true)
303        .mode(0o600)
304        .custom_flags(libc::O_NOFOLLOW)
305        .open(&path);
306    if let Ok(mut f) = result {
307        use std::io::Write;
308        let _ = writeln!(f, "{}", line);
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315
316    #[test]
317    fn level_parses_known_values() {
318        assert_eq!(TelemetryLevel::from_str_key("off"), TelemetryLevel::Off);
319        assert_eq!(TelemetryLevel::from_str_key("basic"), TelemetryLevel::Basic);
320        assert_eq!(TelemetryLevel::from_str_key("full"), TelemetryLevel::Full);
321        assert_eq!(TelemetryLevel::from_str_key("FULL"), TelemetryLevel::Full);
322        assert_eq!(TelemetryLevel::from_str_key("true"), TelemetryLevel::Basic);
323        assert_eq!(TelemetryLevel::from_str_key("garbage"), TelemetryLevel::Off);
324        assert_eq!(TelemetryLevel::from_str_key(""), TelemetryLevel::Off);
325    }
326
327    #[test]
328    fn level_enabled() {
329        assert!(!TelemetryLevel::Off.enabled());
330        assert!(TelemetryLevel::Basic.enabled());
331        assert!(TelemetryLevel::Full.enabled());
332    }
333
334    #[test]
335    fn hit_pct_computation() {
336        let mut u = UsageRecord {
337            input: 100,
338            cache_read: 800,
339            cache_write: 100,
340            ..Default::default()
341        };
342        u.compute_hit_pct();
343        assert_eq!(u.hit_pct, 80.0);
344    }
345
346    #[test]
347    fn hit_pct_zero_total() {
348        let mut u = UsageRecord::default();
349        u.compute_hit_pct();
350        assert_eq!(u.hit_pct, 0.0);
351    }
352
353    #[test]
354    fn hit_pct_rounds_to_one_decimal() {
355        let mut u = UsageRecord {
356            input: 1,
357            cache_read: 2,
358            cache_write: 0,
359            ..Default::default()
360        };
361        u.compute_hit_pct();
362        assert_eq!(u.hit_pct, 66.7);
363    }
364
365    #[test]
366    fn record_serializes_skipping_none_fields() {
367        let record = TelemetryRecord {
368            ts: 1,
369            model: "claude-sonnet-4-6".to_string(),
370            attempt: 1,
371            total_ms: 100,
372            ..Default::default()
373        };
374        let json = serde_json::to_string(&record).unwrap();
375        assert!(!json.contains("request_id"));
376        assert!(!json.contains("ratelimit"));
377        assert!(!json.contains("cache_diag"));
378        assert!(json.contains("\"model\":\"claude-sonnet-4-6\""));
379    }
380
381    #[test]
382    fn ratelimit_empty_detection() {
383        assert!(RateLimitRecord::default().is_empty());
384        let r = RateLimitRecord {
385            requests_remaining: Some(10),
386            ..Default::default()
387        };
388        assert!(!r.is_empty());
389    }
390
391    #[test]
392    fn ratelimit_parses_headers() {
393        let mut headers = reqwest::header::HeaderMap::new();
394        headers.insert("anthropic-ratelimit-requests-limit", "5000".parse().unwrap());
395        headers.insert("anthropic-ratelimit-requests-remaining", "4900".parse().unwrap());
396        headers.insert("anthropic-ratelimit-tokens-reset", "2026-06-11T01:46:00Z".parse().unwrap());
397        let r = ratelimit_from_headers(&headers);
398        assert_eq!(r.requests_limit, Some(5000));
399        assert_eq!(r.requests_remaining, Some(4900));
400        assert_eq!(r.tokens_reset.as_deref(), Some("2026-06-11T01:46:00Z"));
401        assert_eq!(r.tokens_limit, None);
402    }
403
404    #[test]
405    fn ratelimit_ignores_malformed_values() {
406        let mut headers = reqwest::header::HeaderMap::new();
407        headers.insert("anthropic-ratelimit-requests-limit", "not-a-number".parse().unwrap());
408        let r = ratelimit_from_headers(&headers);
409        assert_eq!(r.requests_limit, None);
410    }
411}
412
413#[cfg(test)]
414mod retry_delay_tests {
415    use super::*;
416    use std::time::{Duration, SystemTime, UNIX_EPOCH};
417
418    #[test]
419    fn integer_retry_after() {
420        let mut h = reqwest::header::HeaderMap::new();
421        h.insert("retry-after", "45".parse().unwrap());
422        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
423        assert_eq!(d, Duration::from_secs(45));
424        assert!(from_hdr);
425    }
426
427    #[test]
428    fn integer_retry_after_capped() {
429        let mut h = reqwest::header::HeaderMap::new();
430        h.insert("retry-after", "300".parse().unwrap()); // 5 min — beyond cap
431        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
432        assert_eq!(d, RETRY_DELAY_CAP);
433        assert!(from_hdr);
434    }
435
436    #[test]
437    fn rfc3339_reset_future() {
438        let future_secs = SystemTime::now()
439            .duration_since(UNIX_EPOCH)
440            .unwrap()
441            .as_secs()
442            + 30;
443        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(future_secs as i64, 0).unwrap();
444        let ts = dt.to_rfc3339();
445
446        let mut h = reqwest::header::HeaderMap::new();
447        h.insert("anthropic-ratelimit-tokens-reset", ts.parse().unwrap());
448        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
449        assert!(d.as_secs() >= 28 && d.as_secs() <= 32, "unexpected delay: {:?}", d);
450        assert!(from_hdr);
451    }
452
453    #[test]
454    fn rfc3339_reset_beyond_cap() {
455        let future_secs = SystemTime::now()
456            .duration_since(UNIX_EPOCH)
457            .unwrap()
458            .as_secs()
459            + 600;
460        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(future_secs as i64, 0).unwrap();
461        let ts = dt.to_rfc3339();
462
463        let mut h = reqwest::header::HeaderMap::new();
464        h.insert("anthropic-ratelimit-tokens-reset", ts.parse().unwrap());
465        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
466        assert_eq!(d, RETRY_DELAY_CAP);
467        assert!(from_hdr);
468    }
469
470    #[test]
471    fn no_headers_exponential_fallback() {
472        let h = reqwest::header::HeaderMap::new();
473        let (d1, hdr1) = retry_delay_from_headers(&h, 1);
474        let (d2, hdr2) = retry_delay_from_headers(&h, 2);
475        let (d3, hdr3) = retry_delay_from_headers(&h, 3);
476        assert_eq!(d1, Duration::from_secs(1));
477        assert_eq!(d2, Duration::from_secs(2));
478        assert_eq!(d3, Duration::from_secs(4));
479        assert!(!hdr1 && !hdr2 && !hdr3);
480    }
481
482    #[test]
483    fn prefers_retry_after_over_rfc3339() {
484        let future_secs = SystemTime::now()
485            .duration_since(UNIX_EPOCH)
486            .unwrap()
487            .as_secs()
488            + 30;
489        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(future_secs as i64, 0).unwrap();
490        let ts = dt.to_rfc3339();
491
492        let mut h = reqwest::header::HeaderMap::new();
493        h.insert("retry-after", "10".parse().unwrap());
494        h.insert("anthropic-ratelimit-tokens-reset", ts.parse().unwrap());
495        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
496        assert_eq!(d, Duration::from_secs(10));
497        assert!(from_hdr);
498    }
499
500    #[test]
501    fn min_of_multiple_ratelimit_reset_headers() {
502        // tokens-reset is sooner; requests-reset is later — should pick tokens.
503        let now = SystemTime::now()
504            .duration_since(UNIX_EPOCH)
505            .unwrap()
506            .as_secs();
507        let tokens_dt = chrono::DateTime::<chrono::Utc>::from_timestamp((now + 15) as i64, 0).unwrap();
508        let requests_dt = chrono::DateTime::<chrono::Utc>::from_timestamp((now + 45) as i64, 0).unwrap();
509
510        let mut h = reqwest::header::HeaderMap::new();
511        h.insert("anthropic-ratelimit-tokens-reset", tokens_dt.to_rfc3339().parse().unwrap());
512        h.insert("anthropic-ratelimit-requests-reset", requests_dt.to_rfc3339().parse().unwrap());
513        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
514        assert!(d.as_secs() >= 13 && d.as_secs() <= 17, "should be ~15s, got {:?}", d);
515        assert!(from_hdr);
516    }
517}
agent_engine/runtime/telemetry.rs

agent_engine/runtime/
telemetry.rs
