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

agent_engine/runtime/
telemetry.rs
