verdant_runtime/

stats.rs

//! Per-call JSONL statistics.
//!
//! One line per cached call, in the shape `verdant stats` and
//! `verdant audit` already parse: `{ts_us, tool, outcome, duration_us,
//! args_summary}`, with `outcome` one of `hit`, `miss`, `uncached` or
//! `invalidate`. Both the MCP tool layer and the LLM proxy emit it, and
//! the reporting side reads whichever file it is pointed at, so the
//! shape is one rule and lives here rather than once per producer.
//!
//! Recording is best-effort throughout: a statistics write that failed
//! must never turn a served request into an error, so every failure
//! path returns quietly. The file is opened per call rather than held,
//! which keeps the recorder cheap to clone and share across the
//! proxy's per-request threads; a cached call is already in the
//! millisecond regime or slower, so the open is not on any hot path.

use std::path::PathBuf;

pub const OUTCOME_HIT: &str = "hit";
pub const OUTCOME_MISS: &str = "miss";
pub const OUTCOME_UNCACHED: &str = "uncached";
pub const OUTCOME_INVALIDATE: &str = "invalidate";

#[derive(Clone, Debug, Default)]
pub struct StatsRecorder {
    path: Option<PathBuf>,
}

impl StatsRecorder {
    pub fn disabled() -> Self {
        Self { path: None }
    }

    pub fn new(path: Option<PathBuf>) -> Self {
        Self { path }
    }

    /// Reads the destination from `var`, treating unset or empty as disabled.
    pub fn from_env(var: &str) -> Self {
        let path = std::env::var_os(var)
            .map(PathBuf::from)
            .filter(|p| !p.as_os_str().is_empty());
        Self { path }
    }

    pub fn is_enabled(&self) -> bool {
        self.path.is_some()
    }

    pub fn record(&self, tool: &str, outcome: &str, duration_us: u128, args_summary: &str) {
        self.record_with(tool, outcome, duration_us, args_summary, &[]);
    }

    /// The same record with producer-specific fields alongside it. The reader
    /// ignores names it does not know, so a producer can carry what only it can
    /// measure (the proxy's completion-token count, say) without every producer
    /// having to know about it.
    pub fn record_with(
        &self,
        tool: &str,
        outcome: &str,
        duration_us: u128,
        args_summary: &str,
        extra: &[(&str, serde_json::Value)],
    ) {
        let path = match &self.path {
            Some(p) => p,
            None => return,
        };
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let ts = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_micros())
            .unwrap_or(0);
        let mut line = serde_json::json!({
            "ts_us": ts,
            "tool": tool,
            "outcome": outcome,
            "duration_us": duration_us,
            "args_summary": args_summary,
        });
        if let Some(obj) = line.as_object_mut() {
            for (key, value) in extra {
                obj.insert((*key).to_string(), value.clone());
            }
        }
        let mut serialized = match serde_json::to_string(&line) {
            Ok(s) => s,
            Err(_) => return,
        };
        serialized.push('\n');
        // O_APPEND plus a single write syscall, because the proxy dispatches
        // each request on its own thread and two concurrent records must not
        // interleave into one corrupt line.
        let _ = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
            .and_then(|mut f| std::io::Write::write_all(&mut f, serialized.as_bytes()));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;

    fn lines(path: &std::path::Path) -> Vec<serde_json::Value> {
        std::fs::read_to_string(path)
            .unwrap()
            .lines()
            .map(|l| serde_json::from_str(l).unwrap())
            .collect()
    }

    #[test]
    fn writes_one_json_line_per_call() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");
        let recorder = StatsRecorder::new(Some(path.clone()));

        recorder.record("llm_call", OUTCOME_MISS, 1_500_000, "qwen3.5-4b");
        recorder.record("llm_call", OUTCOME_HIT, 900, "qwen3.5-4b");

        let records = lines(&path);
        assert_eq!(records.len(), 2);
        assert_eq!(records[0]["tool"], "llm_call");
        assert_eq!(records[0]["outcome"], "miss");
        assert_eq!(records[0]["duration_us"], 1_500_000u64);
        assert_eq!(records[0]["args_summary"], "qwen3.5-4b");
        assert_eq!(records[1]["outcome"], "hit");
        assert!(records[0]["ts_us"].as_u64().unwrap() > 0);
    }

    #[test]
    fn carries_a_producer_specific_field_alongside_the_shared_shape() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");

        StatsRecorder::new(Some(path.clone())).record_with(
            "llm_call",
            OUTCOME_MISS,
            1_000,
            "qwen3.5-4b abc123",
            &[("completion_tokens", serde_json::json!(146))],
        );

        let records = lines(&path);
        assert_eq!(records[0]["completion_tokens"], 146);
        assert_eq!(records[0]["tool"], "llm_call");
        assert_eq!(records[0]["outcome"], "miss");
    }

    #[test]
    fn creates_the_parent_directory() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nested/deeper/stats.jsonl");
        StatsRecorder::new(Some(path.clone())).record("llm_call", OUTCOME_HIT, 1, "");
        assert_eq!(lines(&path).len(), 1);
    }

    #[test]
    fn a_disabled_recorder_writes_nothing_and_does_not_fail() {
        let recorder = StatsRecorder::disabled();
        assert!(!recorder.is_enabled());
        recorder.record("llm_call", OUTCOME_HIT, 1, "");
    }

    #[test]
    fn an_empty_env_var_disables_rather_than_writing_to_the_working_directory() {
        // An unset variable often reaches a process as an empty string, and
        // treating that as a path would drop a stats file wherever the process
        // happened to start.
        let var = "VERDANT_STATS_PATH_EMPTY_CASE_TEST";
        std::env::set_var(var, "");
        assert!(!StatsRecorder::from_env(var).is_enabled());
        std::env::remove_var(var);
        assert!(!StatsRecorder::from_env(var).is_enabled());
    }

    #[test]
    fn concurrent_records_do_not_interleave() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");
        let recorder = Arc::new(StatsRecorder::new(Some(path.clone())));

        let handles: Vec<_> = (0..8)
            .map(|i| {
                let recorder = recorder.clone();
                std::thread::spawn(move || {
                    for _ in 0..25 {
                        recorder.record("llm_call", OUTCOME_HIT, i, "model-with-a-longer-name");
                    }
                })
            })
            .collect();
        for h in handles {
            h.join().unwrap();
        }

        let records = lines(&path);
        assert_eq!(records.len(), 200);
        assert!(records.iter().all(|r| r["tool"] == "llm_call"));
    }
}