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

agent_engine/runtime/
telemetry.rs
