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, an estimated `cost_usd` (from
8//! [`agent_core::pricing`]), the persisted conversation id (`session_id`,
9//! the `~/.synaps-cli/sessions/<id>.json` id) and the request-trace
10//! correlation ids (`runtime_session_id` / `turn_id` / `trace_request_id`),
11//! so a record joins both its session and its `request-trace.jsonl` line.
12//! Worker records also carry the worker's identity (`worker_agent`,
13//! `worker_handle`, `parent_session_id`, `parent_runtime_session_id`).
14//! After an automatic context rollover a record names its context window
15//! (`rollover_window`), and the window's first request says `rollover: true`.
16//! `full` additionally records rate-limit headers. The `cache_diag` verdict
17//! is present at either level when `cache_diagnostics` is on.
18//!
19//! Optional fields are omitted when unknown — never written as `0`.
20//!
21//! Writes are best-effort and non-blocking: persistence goes through the
22//! bounded background [`writer`] (Task 11, spec §6.5), so a slow or broken
23//! log path can never delay or fail the request pipeline. Failures are
24//! counted (never silent-and-invisible) with one warning per persistent
25//! failure class.
26
27use serde::Serialize;
28use std::time::{Duration, SystemTime, UNIX_EPOCH};
29
30pub mod writer;
31
32pub use writer::{
33    default_telemetry_log_path, default_trace_log_path, ShutdownOutcome, TelemetryWriter,
34    WriteGate, WriterOptions, WriterStats, WriterTraceSink, DEFAULT_MAX_FILE_BYTES,
35    DEFAULT_MAX_GENERATIONS, DEFAULT_QUEUE_CAPACITY, DEFAULT_SHUTDOWN_FLUSH_TIMEOUT,
36};
37
38/// Telemetry verbosity level, parsed from the `telemetry` config key.
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
40pub enum TelemetryLevel {
41    #[default]
42    Off,
43    Basic,
44    Full,
45}
46
47impl TelemetryLevel {
48    pub fn from_str_key(s: &str) -> Self {
49        match s.trim().to_ascii_lowercase().as_str() {
50            "basic" | "on" | "1" | "true" => Self::Basic,
51            "full" => Self::Full,
52            _ => Self::Off,
53        }
54    }
55
56    pub fn enabled(&self) -> bool {
57        !matches!(self, Self::Off)
58    }
59
60    pub fn as_str(&self) -> &'static str {
61        match self {
62            Self::Off => "off",
63            Self::Basic => "basic",
64            Self::Full => "full",
65        }
66    }
67}
68
69/// The parent session's telemetry, handed to the subagents it spawns.
70///
71/// Workers record at the parent's level and enqueue onto the parent's
72/// session writer instead of starting their own: the writer owns its log
73/// files from a single thread, so a writer per worker would race rotation.
74/// See `Runtime::inherit_telemetry`.
75#[derive(Debug, Clone, Default)]
76pub struct WorkerTelemetry {
77    pub level: TelemetryLevel,
78    /// The parent's session writer. `None` when the parent records nothing.
79    pub writer: Option<TelemetryWriter>,
80    /// The parent's persisted conversation id (the
81    /// `~/.synaps-cli/sessions/<id>.json` id), stamped on the worker's
82    /// records as `parent_session_id`. `None` when the parent is unkeyed.
83    pub parent_session_id: Option<String>,
84    /// The parent's runtime trace session id (`session-<pid>-<ms>-<seq>`),
85    /// stamped as `parent_runtime_session_id`.
86    pub parent_runtime_session_id: Option<String>,
87    /// Set per spawn by [`Self::for_worker`]: the agent name.
88    pub agent: Option<String>,
89    /// Set per spawn by [`Self::for_worker`]: the subagent handle (`sa_N`).
90    pub handle: Option<String>,
91}
92
93impl WorkerTelemetry {
94    /// This telemetry, labelled for one spawned worker.
95    pub fn for_worker(mut self, agent: &str, handle: &str) -> Self {
96        self.agent = Some(agent.to_string());
97        self.handle = Some(handle.to_string());
98        self
99    }
100
101    /// The parent's writer, re-stamped for this worker: every record the
102    /// worker enqueues carries its identity and its parent's ids, and NOT
103    /// the parent's own `session_id`.
104    pub fn worker_writer(&self) -> Option<TelemetryWriter> {
105        self.writer.as_ref().map(|w| {
106            w.with_stamp(RecordStamp {
107                session_id: None,
108                worker_agent: self.agent.clone(),
109                worker_handle: self.handle.clone(),
110                parent_session_id: self.parent_session_id.clone(),
111                parent_runtime_session_id: self.parent_runtime_session_id.clone(),
112            })
113        })
114    }
115}
116
117/// Identity fields a [`TelemetryWriter`] handle stamps onto every record
118/// enqueued through it: the persisted conversation id of the session (see
119/// `Runtime::request_telemetry`) and, on worker handles, who the worker is
120/// (see [`WorkerTelemetry::worker_writer`]).
121#[derive(Debug, Clone, Default, PartialEq, Eq)]
122pub struct RecordStamp {
123    pub session_id: Option<String>,
124    pub worker_agent: Option<String>,
125    pub worker_handle: Option<String>,
126    pub parent_session_id: Option<String>,
127    pub parent_runtime_session_id: Option<String>,
128}
129
130impl RecordStamp {
131    /// Fill the fields a record does not already carry.
132    pub fn apply(&self, record: &mut TelemetryRecord) {
133        fn fill(slot: &mut Option<String>, value: &Option<String>) {
134            if slot.is_none() {
135                slot.clone_from(value);
136            }
137        }
138        fill(&mut record.session_id, &self.session_id);
139        fill(&mut record.worker_agent, &self.worker_agent);
140        fill(&mut record.worker_handle, &self.worker_handle);
141        fill(&mut record.parent_session_id, &self.parent_session_id);
142        fill(
143            &mut record.parent_runtime_session_id,
144            &self.parent_runtime_session_id,
145        );
146    }
147}
148
149/// Token usage for one API call, including the cache-creation TTL breakdown.
150#[derive(Debug, Clone, Default, Serialize)]
151pub struct UsageRecord {
152    pub input: u64,
153    pub output: u64,
154    pub cache_read: u64,
155    pub cache_write: u64,
156    /// Cache writes with 5-minute TTL (from `usage.cache_creation.ephemeral_5m_input_tokens`).
157    #[serde(skip_serializing_if = "Option::is_none")]
158    pub cache_write_5m: Option<u64>,
159    /// Cache writes with 1-hour TTL (from `usage.cache_creation.ephemeral_1h_input_tokens`).
160    #[serde(skip_serializing_if = "Option::is_none")]
161    pub cache_write_1h: Option<u64>,
162    /// Cache hit percentage: cache_read / (input + cache_read + cache_write) * 100.
163    pub hit_pct: f64,
164}
165
166impl UsageRecord {
167    /// Estimated USD cost via [`agent_core::pricing`] (the single pricing
168    /// source), billing 5m and 1h cache writes at their own rates. `None`
169    /// when no usage was reported (all counts zero) — unknown, not free.
170    pub fn cost_usd(&self, model: &str) -> Option<f64> {
171        if self.input == 0 && self.output == 0 && self.cache_read == 0 && self.cache_write == 0 {
172            return None;
173        }
174        Some(agent_core::pricing::calculate_cost_optional_split(
175            model,
176            self.input,
177            self.output,
178            self.cache_read,
179            self.cache_write,
180            self.cache_write_5m,
181            self.cache_write_1h,
182        ))
183    }
184
185    pub fn compute_hit_pct(&mut self) {
186        let total = self.input + self.cache_read + self.cache_write;
187        self.hit_pct = if total > 0 {
188            (self.cache_read as f64 / total as f64 * 1000.0).round() / 10.0
189        } else {
190            0.0
191        };
192    }
193}
194
195/// Rate-limit headroom captured from `anthropic-ratelimit-*` response headers.
196/// Only recorded at `full` level.
197#[derive(Debug, Clone, Default, Serialize)]
198pub struct RateLimitRecord {
199    #[serde(skip_serializing_if = "Option::is_none")]
200    pub requests_limit: Option<u64>,
201    #[serde(skip_serializing_if = "Option::is_none")]
202    pub requests_remaining: Option<u64>,
203    #[serde(skip_serializing_if = "Option::is_none")]
204    pub tokens_limit: Option<u64>,
205    #[serde(skip_serializing_if = "Option::is_none")]
206    pub tokens_remaining: Option<u64>,
207    #[serde(skip_serializing_if = "Option::is_none")]
208    pub input_tokens_remaining: Option<u64>,
209    #[serde(skip_serializing_if = "Option::is_none")]
210    pub output_tokens_remaining: Option<u64>,
211    /// RFC 3339 timestamp when the most restrictive token limit resets.
212    #[serde(skip_serializing_if = "Option::is_none")]
213    pub tokens_reset: Option<String>,
214}
215
216impl RateLimitRecord {
217    pub fn is_empty(&self) -> bool {
218        self.requests_limit.is_none()
219            && self.requests_remaining.is_none()
220            && self.tokens_limit.is_none()
221            && self.tokens_remaining.is_none()
222            && self.input_tokens_remaining.is_none()
223            && self.output_tokens_remaining.is_none()
224            && self.tokens_reset.is_none()
225    }
226}
227
228/// Cache-diagnostics result (beta `cache-diagnosis-2026-04-07`).
229/// Only present when the user opted in via `cache_diagnostics = true`.
230#[derive(Debug, Clone, Serialize)]
231pub struct CacheDiagRecord {
232    /// `cache_miss_reason.type` — e.g. "system_changed", "tools_changed".
233    pub miss_reason: String,
234    /// `cache_missed_input_tokens` — estimated tokens lost after divergence.
235    #[serde(skip_serializing_if = "Option::is_none")]
236    pub missed_tokens: Option<u64>,
237}
238
239/// Request-shape context: what we sent, for correlating cache behavior.
240#[derive(Debug, Clone, Default, Serialize)]
241pub struct ContextRecord {
242    pub messages: usize,
243    pub tools: usize,
244    pub system_bytes: usize,
245    /// Indices of user messages carrying a conversational cache_control marker.
246    pub breakpoints: Vec<usize>,
247}
248
249/// One JSONL record per API call.
250#[derive(Debug, Clone, Default, Serialize)]
251pub struct TelemetryRecord {
252    /// Unix epoch milliseconds at request completion.
253    pub ts: u64,
254    #[serde(skip_serializing_if = "Option::is_none")]
255    pub request_id: Option<String>,
256    /// Anthropic message id (`msg_...`) from message_start.
257    #[serde(skip_serializing_if = "Option::is_none")]
258    pub msg_id: Option<String>,
259    pub model: String,
260    /// 1-based attempt number that succeeded (1 = no retries).
261    pub attempt: u32,
262    /// Number of refusal retries consumed (0 = no refusals). Present only when > 0.
263    #[serde(skip_serializing_if = "Option::is_none")]
264    pub refusal_retries_used: Option<u32>,
265    /// Milliseconds from request send to the first parsed model event — the
266    /// request trace's `outcome.timings.first_model_event_ms`. Absent when no
267    /// event arrived, never `0`.
268    #[serde(skip_serializing_if = "Option::is_none")]
269    pub ttft_ms: Option<u64>,
270    /// Milliseconds from request send to stream close.
271    pub total_ms: u64,
272    #[serde(skip_serializing_if = "Option::is_none")]
273    pub stop_reason: Option<String>,
274    pub usage: UsageRecord,
275    /// Estimated USD cost of this call at API list prices
276    /// ([`UsageRecord::cost_usd`]), also for OAuth/subscription auth.
277    #[serde(skip_serializing_if = "Option::is_none")]
278    pub cost_usd: Option<f64>,
279    #[serde(skip_serializing_if = "Option::is_none")]
280    pub ratelimit: Option<RateLimitRecord>,
281    #[serde(skip_serializing_if = "Option::is_none")]
282    pub cache_diag: Option<CacheDiagRecord>,
283    pub context: ContextRecord,
284    /// `true` only on prompt-cache keep-warm pings (task #444) so telemetry
285    /// can separate pings from real turns. Absent on every turn record.
286    #[serde(skip_serializing_if = "std::ops::Not::not")]
287    pub keepwarm: bool,
288    /// `true` when a subagent (worker runtime) made the request, so reports
289    /// can split delegated traffic from the foreground session. Absent on
290    /// foreground and internal records.
291    #[serde(skip_serializing_if = "std::ops::Not::not")]
292    pub worker: bool,
293    /// Persisted conversation id (`20261028-045932-eb8e`): the id of
294    /// `~/.synaps-cli/sessions/<id>.json` and `synaps daemon sessions`.
295    /// Absent for unkeyed runtimes (workers, tests).
296    #[serde(skip_serializing_if = "Option::is_none")]
297    pub session_id: Option<String>,
298    /// Runtime trace session id (`session-<pid>-<ms>-<seq>`): the `session_id` of
299    /// this call's `request-trace.jsonl` record. Not the persisted id.
300    #[serde(skip_serializing_if = "Option::is_none")]
301    pub runtime_session_id: Option<String>,
302    /// Request-trace turn id (`turn-<pid>-<n>`). Despite the name it is
303    /// minted per logical request (same counter as `trace_request_id`), not
304    /// per user turn.
305    #[serde(skip_serializing_if = "Option::is_none")]
306    pub turn_id: Option<String>,
307    /// Request-trace request id (`req-<pid>-<n>`), shared by every attempt
308    /// of one logical request. Not the provider's `request_id`.
309    #[serde(skip_serializing_if = "Option::is_none")]
310    pub trace_request_id: Option<String>,
311    /// Worker records: the agent name (`spike`), or for an inline subagent
312    /// its sanitized label (model-chosen text, `[A-Za-z0-9_-]`, bounded).
313    #[serde(skip_serializing_if = "Option::is_none")]
314    pub worker_agent: Option<String>,
315    /// Worker records: the subagent handle (`sa_N`).
316    #[serde(skip_serializing_if = "Option::is_none")]
317    pub worker_handle: Option<String>,
318    /// Worker records: the spawning session's persisted `session_id`.
319    #[serde(skip_serializing_if = "Option::is_none")]
320    pub parent_session_id: Option<String>,
321    /// Worker records: the spawning session's `runtime_session_id`.
322    #[serde(skip_serializing_if = "Option::is_none")]
323    pub parent_runtime_session_id: Option<String>,
324    /// The context window (2, 3, …) this request belongs to after an
325    /// automatic context rollover, worker or foreground. Absent before the
326    /// first rollover.
327    #[serde(skip_serializing_if = "Option::is_none")]
328    pub rollover_window: Option<u64>,
329    /// `true` only on the first request of a rolled-over window: the request
330    /// that writes the successor's prompt cache. Absent otherwise.
331    #[serde(skip_serializing_if = "std::ops::Not::not")]
332    pub rollover: bool,
333}
334
335/// What a turn request tells its telemetry record about context rollover
336/// (see [`TelemetryRecord::rollover_window`]).
337#[derive(Debug, Clone, Copy, PartialEq, Eq)]
338pub struct RolloverStamp {
339    /// The context window the request belongs to (2 after one rollover).
340    pub window: u64,
341    /// The first request of that window.
342    pub first_request: bool,
343}
344
345impl TelemetryRecord {
346    /// Fill `runtime_session_id` / `turn_id` / `trace_request_id` from the
347    /// request trace's correlation ids. Without a tracer, `session` alone
348    /// still names the runtime session.
349    pub fn set_correlation(
350        &mut self,
351        session: &crate::runtime::trace::TraceId,
352        ids: Option<(
353            &crate::runtime::trace::TraceId,
354            &crate::runtime::trace::TraceId,
355            &crate::runtime::trace::TraceId,
356        )>,
357    ) {
358        let (session, turn, request) = match ids {
359            Some((s, t, r)) => (s, Some(t), Some(r)),
360            None => (session, None, None),
361        };
362        self.runtime_session_id = Some(session.as_str().to_string());
363        self.turn_id = turn.map(|t| t.as_str().to_string());
364        self.trace_request_id = request.map(|r| r.as_str().to_string());
365    }
366
367    pub fn now_ms() -> u64 {
368        SystemTime::now()
369            .duration_since(UNIX_EPOCH)
370            .map(|d| d.as_millis() as u64)
371            .unwrap_or(0)
372    }
373}
374
375/// Parse a `u64` rate-limit header value.
376fn header_u64(headers: &reqwest::header::HeaderMap, name: &str) -> Option<u64> {
377    headers.get(name)?.to_str().ok()?.parse().ok()
378}
379
380/// Parse a string rate-limit header value.
381fn header_string(headers: &reqwest::header::HeaderMap, name: &str) -> Option<String> {
382    Some(headers.get(name)?.to_str().ok()?.to_string())
383}
384
385/// Extract rate-limit headroom from response headers.
386pub fn ratelimit_from_headers(headers: &reqwest::header::HeaderMap) -> RateLimitRecord {
387    RateLimitRecord {
388        requests_limit: header_u64(headers, "anthropic-ratelimit-requests-limit"),
389        requests_remaining: header_u64(headers, "anthropic-ratelimit-requests-remaining"),
390        tokens_limit: header_u64(headers, "anthropic-ratelimit-tokens-limit"),
391        tokens_remaining: header_u64(headers, "anthropic-ratelimit-tokens-remaining"),
392        input_tokens_remaining: header_u64(headers, "anthropic-ratelimit-input-tokens-remaining"),
393        output_tokens_remaining: header_u64(headers, "anthropic-ratelimit-output-tokens-remaining"),
394        tokens_reset: header_string(headers, "anthropic-ratelimit-tokens-reset"),
395    }
396}
397
398/// Extract the `request-id` response header.
399pub fn request_id_from_headers(headers: &reqwest::header::HeaderMap) -> Option<String> {
400    header_string(headers, "request-id")
401}
402
403/// Maximum delay we'll honour from any rate-limit header. A pathological reset
404/// timestamp far in the future would otherwise stall the turn indefinitely.
405pub const RETRY_DELAY_CAP: Duration = Duration::from_secs(60);
406
407/// Compute how long to wait before the next retry, consulting response headers.
408///
409/// Priority order:
410///   1. `retry-after` (integer seconds, or an HTTP-date — integer is what
411///      Anthropic sends in practice, HTTP-date is handled as a best-effort
412///      fallback).
413///   2. `anthropic-ratelimit-tokens-reset` / `anthropic-ratelimit-requests-reset`
414///      (RFC 3339 UTC timestamp) — take the *minimum* of whichever is present.
415///   3. Classic exponential back-off: `1s * 2^(attempt-1)` (1 s, 2 s, 4 s …).
416///
417/// The returned duration is capped at [`RETRY_DELAY_CAP`] so a far-future
418/// reset timestamp never hangs the turn forever. `attempt` is 1-based (first
419/// retry = 1).
420///
421/// Returns `(delay, from_header)` — the bool is `true` when a header was
422/// used (callers can emit a more informative notice message).
423pub fn retry_delay_from_headers(
424    headers: &reqwest::header::HeaderMap,
425    attempt: u32,
426) -> (Duration, bool) {
427    let now_secs = SystemTime::now()
428        .duration_since(UNIX_EPOCH)
429        .map(|d| d.as_secs())
430        .unwrap_or(0);
431
432    // ── 1. retry-after (integer seconds preferred) ──────────────────────────
433    if let Some(ra) = header_string(headers, "retry-after") {
434        let ra = ra.trim();
435        // Integer form: "30"
436        if let Ok(secs) = ra.parse::<u64>() {
437            let d = Duration::from_secs(secs).min(RETRY_DELAY_CAP);
438            return (d, true);
439        }
440        // HTTP-date form: "Wed, 11 Jun 2025 01:46:00 GMT" — parse via chrono.
441        if let Ok(dt) = chrono::DateTime::parse_from_rfc2822(ra) {
442            let reset_secs = dt.timestamp().max(0) as u64;
443            let wait = reset_secs.saturating_sub(now_secs);
444            let d = Duration::from_secs(wait).min(RETRY_DELAY_CAP);
445            return (d, true);
446        }
447    }
448
449    // ── 2. anthropic-ratelimit-*-reset (RFC 3339) ───────────────────────────
450    let mut min_wait: Option<u64> = None;
451    for name in &[
452        "anthropic-ratelimit-tokens-reset",
453        "anthropic-ratelimit-requests-reset",
454    ] {
455        if let Some(ts) = header_string(headers, name) {
456            if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(ts.trim()) {
457                let reset_secs = dt.timestamp().max(0) as u64;
458                let wait = reset_secs.saturating_sub(now_secs);
459                min_wait = Some(min_wait.map_or(wait, |prev| prev.min(wait)));
460            }
461        }
462    }
463    if let Some(wait) = min_wait {
464        let d = Duration::from_secs(wait).min(RETRY_DELAY_CAP);
465        return (d, true);
466    }
467
468    // ── 3. Exponential back-off fallback ────────────────────────────────────
469    let d = Duration::from_millis(1000 * 2u64.pow(attempt.saturating_sub(1)));
470    (d, false)
471}
472
473#[cfg(test)]
474mod tests {
475    use super::*;
476
477    #[test]
478    fn level_parses_known_values() {
479        assert_eq!(TelemetryLevel::from_str_key("off"), TelemetryLevel::Off);
480        assert_eq!(TelemetryLevel::from_str_key("basic"), TelemetryLevel::Basic);
481        assert_eq!(TelemetryLevel::from_str_key("full"), TelemetryLevel::Full);
482        assert_eq!(TelemetryLevel::from_str_key("FULL"), TelemetryLevel::Full);
483        assert_eq!(TelemetryLevel::from_str_key("true"), TelemetryLevel::Basic);
484        assert_eq!(TelemetryLevel::from_str_key("garbage"), TelemetryLevel::Off);
485        assert_eq!(TelemetryLevel::from_str_key(""), TelemetryLevel::Off);
486    }
487
488    #[test]
489    fn level_enabled() {
490        assert!(!TelemetryLevel::Off.enabled());
491        assert!(TelemetryLevel::Basic.enabled());
492        assert!(TelemetryLevel::Full.enabled());
493    }
494
495    #[test]
496    fn hit_pct_computation() {
497        let mut u = UsageRecord {
498            input: 100,
499            cache_read: 800,
500            cache_write: 100,
501            ..Default::default()
502        };
503        u.compute_hit_pct();
504        assert_eq!(u.hit_pct, 80.0);
505    }
506
507    #[test]
508    fn hit_pct_zero_total() {
509        let mut u = UsageRecord::default();
510        u.compute_hit_pct();
511        assert_eq!(u.hit_pct, 0.0);
512    }
513
514    #[test]
515    fn hit_pct_rounds_to_one_decimal() {
516        let mut u = UsageRecord {
517            input: 1,
518            cache_read: 2,
519            cache_write: 0,
520            ..Default::default()
521        };
522        u.compute_hit_pct();
523        assert_eq!(u.hit_pct, 66.7);
524    }
525
526    #[test]
527    fn record_serializes_skipping_none_fields() {
528        let record = TelemetryRecord {
529            ts: 1,
530            model: "claude-sonnet-4-6".to_string(),
531            attempt: 1,
532            total_ms: 100,
533            ..Default::default()
534        };
535        let json = serde_json::to_string(&record).unwrap();
536        assert!(!json.contains("request_id"));
537        assert!(!json.contains("ratelimit"));
538        assert!(!json.contains("cache_diag"));
539        assert!(!json.contains("keepwarm"), "turn records carry no keepwarm field");
540        assert!(!json.contains("worker"), "foreground records carry no worker field");
541        assert!(json.contains("\"model\":\"claude-sonnet-4-6\""));
542    }
543
544    #[test]
545    fn rollover_fields_appear_only_after_a_rollover() {
546        let plain = serde_json::to_string(&TelemetryRecord::default()).unwrap();
547        assert!(!plain.contains("rollover"), "{plain}");
548        let first = TelemetryRecord {
549            rollover_window: Some(2),
550            rollover: true,
551            worker: true,
552            ..Default::default()
553        };
554        let json = serde_json::to_string(&first).unwrap();
555        assert!(json.contains("\"rollover_window\":2"), "{json}");
556        assert!(json.contains("\"rollover\":true"), "{json}");
557        let later = TelemetryRecord {
558            rollover_window: Some(2),
559            ..Default::default()
560        };
561        let json = serde_json::to_string(&later).unwrap();
562        assert!(json.contains("\"rollover_window\":2"), "{json}");
563        assert!(!json.contains("\"rollover\":"), "{json}");
564    }
565
566    #[test]
567    fn keepwarm_record_is_marked() {
568        let record = TelemetryRecord {
569            ts: 1,
570            model: "claude-opus-4-7".to_string(),
571            keepwarm: true,
572            ..Default::default()
573        };
574        let json = serde_json::to_string(&record).unwrap();
575        assert!(json.contains("\"keepwarm\":true"), "{json}");
576    }
577
578    #[test]
579    fn worker_record_is_marked() {
580        let record = TelemetryRecord {
581            ts: 1,
582            model: "claude-opus-5-5".to_string(),
583            worker: true,
584            ..Default::default()
585        };
586        let json = serde_json::to_string(&record).unwrap();
587        assert!(json.contains("\"worker\":true"), "{json}");
588    }
589
590    #[test]
591    fn ratelimit_empty_detection() {
592        assert!(RateLimitRecord::default().is_empty());
593        let r = RateLimitRecord {
594            requests_remaining: Some(10),
595            ..Default::default()
596        };
597        assert!(!r.is_empty());
598    }
599
600    #[test]
601    fn ratelimit_parses_headers() {
602        let mut headers = reqwest::header::HeaderMap::new();
603        headers.insert(
604            "anthropic-ratelimit-requests-limit",
605            "5000".parse().unwrap(),
606        );
607        headers.insert(
608            "anthropic-ratelimit-requests-remaining",
609            "4900".parse().unwrap(),
610        );
611        headers.insert(
612            "anthropic-ratelimit-tokens-reset",
613            "2026-06-11T01:46:00Z".parse().unwrap(),
614        );
615        let r = ratelimit_from_headers(&headers);
616        assert_eq!(r.requests_limit, Some(5000));
617        assert_eq!(r.requests_remaining, Some(4900));
618        assert_eq!(r.tokens_reset.as_deref(), Some("2026-06-11T01:46:00Z"));
619        assert_eq!(r.tokens_limit, None);
620    }
621
622    #[test]
623    fn ratelimit_ignores_malformed_values() {
624        let mut headers = reqwest::header::HeaderMap::new();
625        headers.insert(
626            "anthropic-ratelimit-requests-limit",
627            "not-a-number".parse().unwrap(),
628        );
629        let r = ratelimit_from_headers(&headers);
630        assert_eq!(r.requests_limit, None);
631    }
632}
633
634#[cfg(test)]
635mod retry_delay_tests {
636    use super::*;
637    use std::time::{Duration, SystemTime, UNIX_EPOCH};
638
639    #[test]
640    fn integer_retry_after() {
641        let mut h = reqwest::header::HeaderMap::new();
642        h.insert("retry-after", "45".parse().unwrap());
643        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
644        assert_eq!(d, Duration::from_secs(45));
645        assert!(from_hdr);
646    }
647
648    #[test]
649    fn integer_retry_after_capped() {
650        let mut h = reqwest::header::HeaderMap::new();
651        h.insert("retry-after", "300".parse().unwrap()); // 5 min — beyond cap
652        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
653        assert_eq!(d, RETRY_DELAY_CAP);
654        assert!(from_hdr);
655    }
656
657    #[test]
658    fn rfc3339_reset_future() {
659        let future_secs = SystemTime::now()
660            .duration_since(UNIX_EPOCH)
661            .unwrap()
662            .as_secs()
663            + 30;
664        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(future_secs as i64, 0).unwrap();
665        let ts = dt.to_rfc3339();
666
667        let mut h = reqwest::header::HeaderMap::new();
668        h.insert("anthropic-ratelimit-tokens-reset", ts.parse().unwrap());
669        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
670        assert!(
671            d.as_secs() >= 28 && d.as_secs() <= 32,
672            "unexpected delay: {:?}",
673            d
674        );
675        assert!(from_hdr);
676    }
677
678    #[test]
679    fn rfc3339_reset_beyond_cap() {
680        let future_secs = SystemTime::now()
681            .duration_since(UNIX_EPOCH)
682            .unwrap()
683            .as_secs()
684            + 600;
685        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(future_secs as i64, 0).unwrap();
686        let ts = dt.to_rfc3339();
687
688        let mut h = reqwest::header::HeaderMap::new();
689        h.insert("anthropic-ratelimit-tokens-reset", ts.parse().unwrap());
690        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
691        assert_eq!(d, RETRY_DELAY_CAP);
692        assert!(from_hdr);
693    }
694
695    #[test]
696    fn no_headers_exponential_fallback() {
697        let h = reqwest::header::HeaderMap::new();
698        let (d1, hdr1) = retry_delay_from_headers(&h, 1);
699        let (d2, hdr2) = retry_delay_from_headers(&h, 2);
700        let (d3, hdr3) = retry_delay_from_headers(&h, 3);
701        assert_eq!(d1, Duration::from_secs(1));
702        assert_eq!(d2, Duration::from_secs(2));
703        assert_eq!(d3, Duration::from_secs(4));
704        assert!(!hdr1 && !hdr2 && !hdr3);
705    }
706
707    #[test]
708    fn prefers_retry_after_over_rfc3339() {
709        let future_secs = SystemTime::now()
710            .duration_since(UNIX_EPOCH)
711            .unwrap()
712            .as_secs()
713            + 30;
714        let dt = chrono::DateTime::<chrono::Utc>::from_timestamp(future_secs as i64, 0).unwrap();
715        let ts = dt.to_rfc3339();
716
717        let mut h = reqwest::header::HeaderMap::new();
718        h.insert("retry-after", "10".parse().unwrap());
719        h.insert("anthropic-ratelimit-tokens-reset", ts.parse().unwrap());
720        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
721        assert_eq!(d, Duration::from_secs(10));
722        assert!(from_hdr);
723    }
724
725    #[test]
726    fn min_of_multiple_ratelimit_reset_headers() {
727        // tokens-reset is sooner; requests-reset is later — should pick tokens.
728        let now = SystemTime::now()
729            .duration_since(UNIX_EPOCH)
730            .unwrap()
731            .as_secs();
732        let tokens_dt =
733            chrono::DateTime::<chrono::Utc>::from_timestamp((now + 15) as i64, 0).unwrap();
734        let requests_dt =
735            chrono::DateTime::<chrono::Utc>::from_timestamp((now + 45) as i64, 0).unwrap();
736
737        let mut h = reqwest::header::HeaderMap::new();
738        h.insert(
739            "anthropic-ratelimit-tokens-reset",
740            tokens_dt.to_rfc3339().parse().unwrap(),
741        );
742        h.insert(
743            "anthropic-ratelimit-requests-reset",
744            requests_dt.to_rfc3339().parse().unwrap(),
745        );
746        let (d, from_hdr) = retry_delay_from_headers(&h, 1);
747        assert!(
748            d.as_secs() >= 13 && d.as_secs() <= 17,
749            "should be ~15s, got {:?}",
750            d
751        );
752        assert!(from_hdr);
753    }
754}
755
756/// Headerless network retry delay. Bounded even for arbitrarily large config
757/// values; matches the streaming error path's 64-second ceiling.
758pub(super) fn network_retry_delay(attempt: u32) -> Duration {
759    Duration::from_secs(1u64 << attempt.saturating_sub(1).min(6))
760}
761
762#[cfg(test)]
763mod network_backoff_tests {
764    #[test]
765    fn configured_large_retry_counts_never_overflow() {
766        let seconds: Vec<_> = [0, 1, 2, 3, 7, 8, u32::MAX]
767            .into_iter()
768            .map(|n| super::network_retry_delay(n).as_secs())
769            .collect();
770        assert_eq!(seconds, [1, 1, 2, 4, 64, 64, 64]);
771    }
772}
agent_engine/runtime/telemetry.rs

agent_engine/runtime/
telemetry.rs
