verdant_runtime/
stats.rs

1//! Per-call JSONL statistics.
2//!
3//! One line per cached call, in the shape `verdant stats` and
4//! `verdant audit` already parse: `{ts_us, tool, outcome, duration_us,
5//! args_summary, reason, call_id}`, with `outcome` one of `hit`, `miss`, `uncached`,
6//! `invalidate` or `declined`, plus a `reason` that is null on every line
7//! except an `uncached` one, where it names why the call bypassed the cache,
8//! and a `declined` one, where it names why `bash_match` did not match. A
9//! `declined` line is a query that ran nothing, so the reports keep it out of
10//! every hit and miss rate. `call_id` is a string on every line: the id the
11//! caller attached to the call (see `accept_call_id`), or a generated
12//! `verdant-` plus 32 hex characters when it attached none, so a consumer can
13//! join a line to the call that caused it. Lines written before the field
14//! existed lack it. Both the
15//! MCP tool layer and the LLM proxy emit it, and
16//! the reporting side reads whichever file it is pointed at, so the
17//! shape is one rule and lives here rather than once per producer.
18//!
19//! Recording is best-effort throughout: a statistics write that failed
20//! must never turn a served request into an error, so every failure
21//! path returns quietly. The file is opened per call rather than held,
22//! which keeps the recorder cheap to clone and share across the
23//! proxy's per-request threads; a cached call is already in the
24//! millisecond regime or slower, so the open is not on any hot path.
25
26use std::path::PathBuf;
27use std::sync::atomic::{AtomicU64, Ordering};
28use std::sync::OnceLock;
29
30pub const OUTCOME_HIT: &str = "hit";
31pub const OUTCOME_MISS: &str = "miss";
32pub const OUTCOME_UNCACHED: &str = "uncached";
33pub const OUTCOME_INVALIDATE: &str = "invalidate";
34pub const OUTCOME_DECLINED: &str = "declined";
35
36/// Why a call ran uncached. The set is open (a reader treats an unknown name
37/// as opaque text), but producers use these so a count by reason is stable.
38pub mod reason {
39    pub const NO_RULE: &str = "no_rule";
40    pub const NOT_ALLOWLISTED: &str = "not_allowlisted";
41    pub const OUT_OF_ROOT: &str = "out_of_root";
42    pub const NO_EXEC_CACHE: &str = "no_exec_cache";
43    pub const SANDBOX_UNAVAILABLE: &str = "sandbox_unavailable";
44    pub const SANDBOX_ERROR: &str = "sandbox_error";
45    pub const NETWORK: &str = "network";
46    pub const SYSTEM_SOCKET: &str = "system_socket";
47    pub const FAILED_RUN: &str = "failed_run";
48    pub const TIMEOUT: &str = "timeout";
49    pub const PATH_CHANGED: &str = "path_changed";
50    pub const OVERSIZED: &str = "oversized";
51    pub const NO_INPUTS: &str = "no_inputs";
52    pub const UNHASHABLE_INPUT: &str = "unhashable_input";
53    pub const INPUT_CHANGED: &str = "input_changed";
54    pub const OUTSIDE_WRITE: &str = "outside_write";
55    pub const NOT_CACHEABLE: &str = "not_cacheable";
56    pub const ERROR: &str = "error";
57}
58
59#[derive(Clone, Debug, Default)]
60pub struct StatsRecorder {
61    path: Option<PathBuf>,
62}
63
64impl StatsRecorder {
65    pub fn disabled() -> Self {
66        Self { path: None }
67    }
68
69    pub fn new(path: Option<PathBuf>) -> Self {
70        Self { path }
71    }
72
73    /// Reads the destination from `var`, treating unset or empty as disabled.
74    pub fn from_env(var: &str) -> Self {
75        let path = std::env::var_os(var)
76            .map(PathBuf::from)
77            .filter(|p| !p.as_os_str().is_empty());
78        Self { path }
79    }
80
81    pub fn is_enabled(&self) -> bool {
82        self.path.is_some()
83    }
84
85    /// Appends one line. A disabled recorder returns before generating an id.
86    pub fn write(&self, line: &StatLine<'_>) {
87        let path = match &self.path {
88            Some(p) => p,
89            None => return,
90        };
91        if let Some(parent) = path.parent() {
92            let _ = std::fs::create_dir_all(parent);
93        }
94        let ts = std::time::SystemTime::now()
95            .duration_since(std::time::UNIX_EPOCH)
96            .map(|d| d.as_micros())
97            .unwrap_or(0);
98        let call_id = match line.call_id {
99            Some(id) => id.to_string(),
100            None => generate_call_id(),
101        };
102        let mut record = serde_json::Map::new();
103        for (key, value) in line.extra {
104            record.insert((*key).to_string(), value.clone());
105        }
106        // Shared fields go in last so a producer extra that happens to share a
107        // name cannot replace one of them.
108        let shared = [
109            ("ts_us", serde_json::json!(ts)),
110            ("tool", serde_json::json!(line.tool)),
111            ("outcome", serde_json::json!(line.outcome)),
112            ("duration_us", serde_json::json!(line.duration_us)),
113            ("args_summary", serde_json::json!(line.args_summary)),
114            ("reason", serde_json::json!(line.reason)),
115            ("call_id", serde_json::json!(call_id)),
116        ];
117        for (key, value) in shared {
118            record.insert(key.to_string(), value);
119        }
120        let mut serialized = match serde_json::to_string(&record) {
121            Ok(s) => s,
122            Err(_) => return,
123        };
124        serialized.push('\n');
125        // O_APPEND plus a single write syscall, because the proxy dispatches
126        // each request on its own thread and two concurrent records must not
127        // interleave into one corrupt line.
128        let _ = std::fs::OpenOptions::new()
129            .create(true)
130            .append(true)
131            .open(path)
132            .and_then(|mut f| std::io::Write::write_all(&mut f, serialized.as_bytes()));
133    }
134}
135
136/// One ledger line before it is written. `reason` is only meaningful for an
137/// `uncached` or `declined` outcome. `call_id` is the caller's id for the call,
138/// already checked with `accept_call_id`; `None` makes the writer generate one.
139/// `extra` carries producer-specific fields (the proxy's completion-token
140/// count, say), which the reader ignores when it does not know them.
141#[derive(Clone, Copy, Debug)]
142pub struct StatLine<'a> {
143    pub tool: &'a str,
144    pub outcome: &'a str,
145    pub duration_us: u128,
146    pub args_summary: &'a str,
147    pub reason: Option<&'a str>,
148    pub call_id: Option<&'a str>,
149    pub extra: &'a [(&'a str, serde_json::Value)],
150}
151
152impl<'a> StatLine<'a> {
153    pub fn new(tool: &'a str, outcome: &'a str, duration_us: u128, args_summary: &'a str) -> Self {
154        Self {
155            tool,
156            outcome,
157            duration_us,
158            args_summary,
159            reason: None,
160            call_id: None,
161            extra: &[],
162        }
163    }
164}
165
166pub const MAX_CALL_ID_BYTES: usize = 256;
167
168/// The one rule for a caller-supplied call id, shared by every producer: a
169/// string of 1 to `MAX_CALL_ID_BYTES` bytes is written verbatim, anything else
170/// is treated as absent. Never truncated, so an id on a line is exactly what
171/// the caller sent.
172pub fn accept_call_id(raw: &str) -> Option<&str> {
173    (!raw.is_empty() && raw.len() <= MAX_CALL_ID_BYTES).then_some(raw)
174}
175
176/// `verdant-` and 32 hex characters. Unique per call: the hash input pairs the
177/// process id and the process's first-use wall clock (distinct across
178/// processes, including two containers that both run as pid 1) with a counter
179/// (distinct within the process).
180fn generate_call_id() -> String {
181    static SEQ: AtomicU64 = AtomicU64::new(0);
182    static EPOCH_NS: OnceLock<u128> = OnceLock::new();
183    let epoch = *EPOCH_NS.get_or_init(|| {
184        std::time::SystemTime::now()
185            .duration_since(std::time::UNIX_EPOCH)
186            .map(|d| d.as_nanos())
187            .unwrap_or(0)
188    });
189    let seq = SEQ.fetch_add(1, Ordering::Relaxed);
190    let mut hasher = blake3::Hasher::new();
191    hasher.update(&std::process::id().to_le_bytes());
192    hasher.update(&epoch.to_le_bytes());
193    hasher.update(&seq.to_le_bytes());
194    format!("verdant-{}", &hasher.finalize().to_hex()[..32])
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200    use std::sync::Arc;
201
202    fn lines(path: &std::path::Path) -> Vec<serde_json::Value> {
203        std::fs::read_to_string(path)
204            .unwrap()
205            .lines()
206            .map(|l| serde_json::from_str(l).unwrap())
207            .collect()
208    }
209
210    #[test]
211    fn writes_one_json_line_per_call() {
212        let dir = tempfile::tempdir().unwrap();
213        let path = dir.path().join("stats.jsonl");
214        let recorder = StatsRecorder::new(Some(path.clone()));
215
216        recorder.write(&StatLine::new(
217            "llm_call",
218            OUTCOME_MISS,
219            1_500_000,
220            "qwen3.5-4b",
221        ));
222        recorder.write(&StatLine::new("llm_call", OUTCOME_HIT, 900, "qwen3.5-4b"));
223
224        let records = lines(&path);
225        assert_eq!(records.len(), 2);
226        assert_eq!(records[0]["tool"], "llm_call");
227        assert_eq!(records[0]["outcome"], "miss");
228        assert_eq!(records[0]["duration_us"], 1_500_000u64);
229        assert_eq!(records[0]["args_summary"], "qwen3.5-4b");
230        assert_eq!(records[1]["outcome"], "hit");
231        assert!(records[0]["ts_us"].as_u64().unwrap() > 0);
232    }
233
234    #[test]
235    fn every_line_carries_a_reason_that_is_null_unless_given() {
236        let dir = tempfile::tempdir().unwrap();
237        let path = dir.path().join("stats.jsonl");
238        let recorder = StatsRecorder::new(Some(path.clone()));
239
240        recorder.write(&StatLine::new("read", OUTCOME_HIT, 1, "a.txt"));
241        recorder.write(&StatLine::new("llm_call", OUTCOME_MISS, 1, "m"));
242        recorder.write(&StatLine {
243            reason: Some(reason::NO_RULE),
244            ..StatLine::new("bash", OUTCOME_UNCACHED, 1, "ls")
245        });
246
247        let records = lines(&path);
248        assert!(records[0]["reason"].is_null());
249        assert!(records[0].as_object().unwrap().contains_key("reason"));
250        assert!(records[1]["reason"].is_null());
251        assert_eq!(records[2]["reason"], "no_rule");
252    }
253
254    #[test]
255    fn carries_a_producer_specific_field_alongside_the_shared_shape() {
256        let dir = tempfile::tempdir().unwrap();
257        let path = dir.path().join("stats.jsonl");
258
259        StatsRecorder::new(Some(path.clone())).write(&StatLine {
260            extra: &[("completion_tokens", serde_json::json!(146))],
261            ..StatLine::new("llm_call", OUTCOME_MISS, 1_000, "qwen3.5-4b abc123")
262        });
263
264        let records = lines(&path);
265        assert_eq!(records[0]["completion_tokens"], 146);
266        assert_eq!(records[0]["tool"], "llm_call");
267        assert_eq!(records[0]["outcome"], "miss");
268    }
269
270    #[test]
271    fn creates_the_parent_directory() {
272        let dir = tempfile::tempdir().unwrap();
273        let path = dir.path().join("nested/deeper/stats.jsonl");
274        StatsRecorder::new(Some(path.clone())).write(&StatLine::new(
275            "llm_call",
276            OUTCOME_HIT,
277            1,
278            "",
279        ));
280        assert_eq!(lines(&path).len(), 1);
281    }
282
283    #[test]
284    fn a_disabled_recorder_writes_nothing_and_does_not_fail() {
285        let recorder = StatsRecorder::disabled();
286        assert!(!recorder.is_enabled());
287        recorder.write(&StatLine::new("llm_call", OUTCOME_HIT, 1, ""));
288    }
289
290    #[test]
291    fn an_empty_env_var_disables_rather_than_writing_to_the_working_directory() {
292        // An unset variable often reaches a process as an empty string, and
293        // treating that as a path would drop a stats file wherever the process
294        // happened to start.
295        let var = "VERDANT_STATS_PATH_EMPTY_CASE_TEST";
296        std::env::set_var(var, "");
297        assert!(!StatsRecorder::from_env(var).is_enabled());
298        std::env::remove_var(var);
299        assert!(!StatsRecorder::from_env(var).is_enabled());
300    }
301
302    #[test]
303    fn concurrent_records_do_not_interleave() {
304        let dir = tempfile::tempdir().unwrap();
305        let path = dir.path().join("stats.jsonl");
306        let recorder = Arc::new(StatsRecorder::new(Some(path.clone())));
307
308        let handles: Vec<_> = (0..8)
309            .map(|i| {
310                let recorder = recorder.clone();
311                std::thread::spawn(move || {
312                    for _ in 0..25 {
313                        recorder.write(&StatLine::new(
314                            "llm_call",
315                            OUTCOME_HIT,
316                            i,
317                            "model-with-a-longer-name",
318                        ));
319                    }
320                })
321            })
322            .collect();
323        for h in handles {
324            h.join().unwrap();
325        }
326
327        let records = lines(&path);
328        assert_eq!(records.len(), 200);
329        assert!(records.iter().all(|r| r["tool"] == "llm_call"));
330    }
331
332    fn is_generated_id(id: &str) -> bool {
333        id.strip_prefix("verdant-").is_some_and(|hex| {
334            hex.len() == 32
335                && hex
336                    .bytes()
337                    .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
338        })
339    }
340
341    #[test]
342    fn writes_the_call_id_it_is_given() {
343        let dir = tempfile::tempdir().unwrap();
344        let path = dir.path().join("stats.jsonl");
345        StatsRecorder::new(Some(path.clone())).write(&StatLine {
346            call_id: Some("toolu_01abc"),
347            ..StatLine::new("bash", OUTCOME_MISS, 1, "ls")
348        });
349        assert_eq!(lines(&path)[0]["call_id"], "toolu_01abc");
350    }
351
352    #[test]
353    fn generates_a_call_id_when_none_is_given() {
354        let dir = tempfile::tempdir().unwrap();
355        let path = dir.path().join("stats.jsonl");
356        let recorder = StatsRecorder::new(Some(path.clone()));
357        recorder.write(&StatLine::new("bash", OUTCOME_MISS, 1, "ls"));
358        recorder.write(&StatLine::new("bash", OUTCOME_HIT, 1, "ls"));
359
360        let records = lines(&path);
361        let first = records[0]["call_id"].as_str().unwrap();
362        let second = records[1]["call_id"].as_str().unwrap();
363        assert!(is_generated_id(first), "{first}");
364        assert!(is_generated_id(second), "{second}");
365        assert_ne!(first, second);
366    }
367
368    #[test]
369    fn generated_call_ids_are_unique_across_threads() {
370        let handles: Vec<_> = (0..4)
371            .map(|_| {
372                std::thread::spawn(|| (0..500).map(|_| generate_call_id()).collect::<Vec<_>>())
373            })
374            .collect();
375        let mut all = std::collections::HashSet::new();
376        for h in handles {
377            for id in h.join().unwrap() {
378                assert!(is_generated_id(&id), "{id}");
379                assert!(all.insert(id));
380            }
381        }
382        assert_eq!(all.len(), 2_000);
383    }
384
385    #[test]
386    fn a_caller_id_is_accepted_only_between_1_and_256_bytes() {
387        assert_eq!(accept_call_id("a"), Some("a"));
388        let longest = "x".repeat(MAX_CALL_ID_BYTES);
389        assert_eq!(accept_call_id(&longest), Some(longest.as_str()));
390        assert_eq!(accept_call_id(""), None);
391        assert_eq!(accept_call_id(&"x".repeat(MAX_CALL_ID_BYTES + 1)), None);
392    }
393
394    #[test]
395    fn an_extra_cannot_replace_a_shared_field() {
396        let dir = tempfile::tempdir().unwrap();
397        let path = dir.path().join("stats.jsonl");
398        StatsRecorder::new(Some(path.clone())).write(&StatLine {
399            call_id: Some("real"),
400            extra: &[
401                ("call_id", serde_json::json!("forged")),
402                ("tool", serde_json::json!("forged")),
403            ],
404            ..StatLine::new("llm_call", OUTCOME_MISS, 1, "m")
405        });
406        let record = &lines(&path)[0];
407        assert_eq!(record["call_id"], "real");
408        assert_eq!(record["tool"], "llm_call");
409    }
410}
verdant_runtime/stats.rs

verdant_runtime/
stats.rs
