verdant-cache-runtime 0.6.1

//! 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, reason, call_id}`, with `outcome` one of `hit`, `miss`, `uncached`,
//! `invalidate` or `declined`, plus a `reason` that is null on every line
//! except an `uncached` one, where it names why the call bypassed the cache,
//! and a `declined` one, where it names why `bash_match` did not match. A
//! `declined` line is a query that ran nothing, so the reports keep it out of
//! every hit and miss rate. `call_id` is a string on every line: the id the
//! caller attached to the call (see `accept_call_id`), or a generated
//! `verdant-` plus 32 hex characters when it attached none, so a consumer can
//! join a line to the call that caused it. Lines written before the field
//! existed lack it. 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;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;

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

/// Why a call ran uncached. The set is open (a reader treats an unknown name
/// as opaque text), but producers use these so a count by reason is stable.
pub mod reason {
    pub const NO_RULE: &str = "no_rule";
    pub const NOT_ALLOWLISTED: &str = "not_allowlisted";
    pub const OUT_OF_ROOT: &str = "out_of_root";
    pub const NO_EXEC_CACHE: &str = "no_exec_cache";
    pub const SANDBOX_UNAVAILABLE: &str = "sandbox_unavailable";
    pub const SANDBOX_ERROR: &str = "sandbox_error";
    pub const NETWORK: &str = "network";
    pub const SYSTEM_SOCKET: &str = "system_socket";
    pub const FAILED_RUN: &str = "failed_run";
    pub const TIMEOUT: &str = "timeout";
    pub const PATH_CHANGED: &str = "path_changed";
    pub const OVERSIZED: &str = "oversized";
    pub const NO_INPUTS: &str = "no_inputs";
    pub const UNHASHABLE_INPUT: &str = "unhashable_input";
    pub const INPUT_CHANGED: &str = "input_changed";
    pub const OUTSIDE_WRITE: &str = "outside_write";
    pub const NOT_CACHEABLE: &str = "not_cacheable";
    pub const ERROR: &str = "error";
}

#[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()
    }

    /// Appends one line. A disabled recorder returns before generating an id.
    pub fn write(&self, line: &StatLine<'_>) {
        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 call_id = match line.call_id {
            Some(id) => id.to_string(),
            None => generate_call_id(),
        };
        let mut record = serde_json::Map::new();
        for (key, value) in line.extra {
            record.insert((*key).to_string(), value.clone());
        }
        // Shared fields go in last so a producer extra that happens to share a
        // name cannot replace one of them.
        let shared = [
            ("ts_us", serde_json::json!(ts)),
            ("tool", serde_json::json!(line.tool)),
            ("outcome", serde_json::json!(line.outcome)),
            ("duration_us", serde_json::json!(line.duration_us)),
            ("args_summary", serde_json::json!(line.args_summary)),
            ("reason", serde_json::json!(line.reason)),
            ("call_id", serde_json::json!(call_id)),
        ];
        for (key, value) in shared {
            record.insert(key.to_string(), value);
        }
        let mut serialized = match serde_json::to_string(&record) {
            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()));
    }
}

/// One ledger line before it is written. `reason` is only meaningful for an
/// `uncached` or `declined` outcome. `call_id` is the caller's id for the call,
/// already checked with `accept_call_id`; `None` makes the writer generate one.
/// `extra` carries producer-specific fields (the proxy's completion-token
/// count, say), which the reader ignores when it does not know them.
#[derive(Clone, Copy, Debug)]
pub struct StatLine<'a> {
    pub tool: &'a str,
    pub outcome: &'a str,
    pub duration_us: u128,
    pub args_summary: &'a str,
    pub reason: Option<&'a str>,
    pub call_id: Option<&'a str>,
    pub extra: &'a [(&'a str, serde_json::Value)],
}

impl<'a> StatLine<'a> {
    pub fn new(tool: &'a str, outcome: &'a str, duration_us: u128, args_summary: &'a str) -> Self {
        Self {
            tool,
            outcome,
            duration_us,
            args_summary,
            reason: None,
            call_id: None,
            extra: &[],
        }
    }
}

pub const MAX_CALL_ID_BYTES: usize = 256;

/// The one rule for a caller-supplied call id, shared by every producer: a
/// string of 1 to `MAX_CALL_ID_BYTES` bytes is written verbatim, anything else
/// is treated as absent. Never truncated, so an id on a line is exactly what
/// the caller sent.
pub fn accept_call_id(raw: &str) -> Option<&str> {
    (!raw.is_empty() && raw.len() <= MAX_CALL_ID_BYTES).then_some(raw)
}

/// `verdant-` and 32 hex characters. Unique per call: the hash input pairs the
/// process id and the process's first-use wall clock (distinct across
/// processes, including two containers that both run as pid 1) with a counter
/// (distinct within the process).
fn generate_call_id() -> String {
    static SEQ: AtomicU64 = AtomicU64::new(0);
    static EPOCH_NS: OnceLock<u128> = OnceLock::new();
    let epoch = *EPOCH_NS.get_or_init(|| {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0)
    });
    let seq = SEQ.fetch_add(1, Ordering::Relaxed);
    let mut hasher = blake3::Hasher::new();
    hasher.update(&std::process::id().to_le_bytes());
    hasher.update(&epoch.to_le_bytes());
    hasher.update(&seq.to_le_bytes());
    format!("verdant-{}", &hasher.finalize().to_hex()[..32])
}

#[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.write(&StatLine::new(
            "llm_call",
            OUTCOME_MISS,
            1_500_000,
            "qwen3.5-4b",
        ));
        recorder.write(&StatLine::new("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 every_line_carries_a_reason_that_is_null_unless_given() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");
        let recorder = StatsRecorder::new(Some(path.clone()));

        recorder.write(&StatLine::new("read", OUTCOME_HIT, 1, "a.txt"));
        recorder.write(&StatLine::new("llm_call", OUTCOME_MISS, 1, "m"));
        recorder.write(&StatLine {
            reason: Some(reason::NO_RULE),
            ..StatLine::new("bash", OUTCOME_UNCACHED, 1, "ls")
        });

        let records = lines(&path);
        assert!(records[0]["reason"].is_null());
        assert!(records[0].as_object().unwrap().contains_key("reason"));
        assert!(records[1]["reason"].is_null());
        assert_eq!(records[2]["reason"], "no_rule");
    }

    #[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())).write(&StatLine {
            extra: &[("completion_tokens", serde_json::json!(146))],
            ..StatLine::new("llm_call", OUTCOME_MISS, 1_000, "qwen3.5-4b abc123")
        });

        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())).write(&StatLine::new(
            "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.write(&StatLine::new("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.write(&StatLine::new(
                            "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"));
    }

    fn is_generated_id(id: &str) -> bool {
        id.strip_prefix("verdant-").is_some_and(|hex| {
            hex.len() == 32
                && hex
                    .bytes()
                    .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
        })
    }

    #[test]
    fn writes_the_call_id_it_is_given() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");
        StatsRecorder::new(Some(path.clone())).write(&StatLine {
            call_id: Some("toolu_01abc"),
            ..StatLine::new("bash", OUTCOME_MISS, 1, "ls")
        });
        assert_eq!(lines(&path)[0]["call_id"], "toolu_01abc");
    }

    #[test]
    fn generates_a_call_id_when_none_is_given() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");
        let recorder = StatsRecorder::new(Some(path.clone()));
        recorder.write(&StatLine::new("bash", OUTCOME_MISS, 1, "ls"));
        recorder.write(&StatLine::new("bash", OUTCOME_HIT, 1, "ls"));

        let records = lines(&path);
        let first = records[0]["call_id"].as_str().unwrap();
        let second = records[1]["call_id"].as_str().unwrap();
        assert!(is_generated_id(first), "{first}");
        assert!(is_generated_id(second), "{second}");
        assert_ne!(first, second);
    }

    #[test]
    fn generated_call_ids_are_unique_across_threads() {
        let handles: Vec<_> = (0..4)
            .map(|_| {
                std::thread::spawn(|| (0..500).map(|_| generate_call_id()).collect::<Vec<_>>())
            })
            .collect();
        let mut all = std::collections::HashSet::new();
        for h in handles {
            for id in h.join().unwrap() {
                assert!(is_generated_id(&id), "{id}");
                assert!(all.insert(id));
            }
        }
        assert_eq!(all.len(), 2_000);
    }

    #[test]
    fn a_caller_id_is_accepted_only_between_1_and_256_bytes() {
        assert_eq!(accept_call_id("a"), Some("a"));
        let longest = "x".repeat(MAX_CALL_ID_BYTES);
        assert_eq!(accept_call_id(&longest), Some(longest.as_str()));
        assert_eq!(accept_call_id(""), None);
        assert_eq!(accept_call_id(&"x".repeat(MAX_CALL_ID_BYTES + 1)), None);
    }

    #[test]
    fn an_extra_cannot_replace_a_shared_field() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stats.jsonl");
        StatsRecorder::new(Some(path.clone())).write(&StatLine {
            call_id: Some("real"),
            extra: &[
                ("call_id", serde_json::json!("forged")),
                ("tool", serde_json::json!("forged")),
            ],
            ..StatLine::new("llm_call", OUTCOME_MISS, 1, "m")
        });
        let record = &lines(&path)[0];
        assert_eq!(record["call_id"], "real");
        assert_eq!(record["tool"], "llm_call");
    }
}