kaizen/telemetry/
file.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2//! Append-only NDJSON file sink (local telemetry).
3
4use super::TelemetryExporter;
5use super::batch_metadata;
6use crate::sync::IngestExportBatch;
7use anyhow::{Context, Result};
8use std::fs::OpenOptions;
9use std::io::Write;
10use std::path::{Path, PathBuf};
11
12/// Default: `~/.kaizen/projects/<slug>/telemetry.ndjson`.
13pub fn default_ndjson_path(workspace: &Path) -> PathBuf {
14    crate::core::paths::project_data_dir(workspace)
15        .map(|d| d.join("telemetry.ndjson"))
16        .unwrap_or_else(|_| workspace.join("telemetry.ndjson"))
17}
18
19pub fn resolve_file_exporter_path(path_opt: Option<&str>, workspace: &Path) -> PathBuf {
20    let p: PathBuf = path_opt
21        .map(PathBuf::from)
22        .unwrap_or_else(|| default_ndjson_path(workspace));
23    if p.is_absolute() {
24        p
25    } else {
26        workspace.join(p)
27    }
28}
29
30pub struct FileExporter {
31    path: PathBuf,
32}
33
34impl FileExporter {
35    pub fn new(path: PathBuf) -> Self {
36        Self { path }
37    }
38
39    pub fn path(&self) -> &Path {
40        &self.path
41    }
42}
43
44impl TelemetryExporter for FileExporter {
45    fn name(&self) -> &str {
46        "file"
47    }
48
49    fn export(&self, batch: &IngestExportBatch) -> Result<()> {
50        let t = now_ms();
51        let v = batch_metadata::telemetry_file_line(batch, t);
52        append_json_line(&self.path, &v)
53    }
54}
55
56fn now_ms() -> i64 {
57    std::time::SystemTime::now()
58        .duration_since(std::time::UNIX_EPOCH)
59        .map(|d| d.as_millis() as i64)
60        .unwrap_or(0)
61}
62
63fn append_json_line(path: &Path, v: &serde_json::Value) -> Result<()> {
64    if let Some(d) = path.parent() {
65        std::fs::create_dir_all(d).with_context(|| format!("create {}", d.display()))?;
66    }
67    let mut f = OpenOptions::new()
68        .create(true)
69        .append(true)
70        .open(path)
71        .with_context(|| format!("open {}", path.display()))?;
72    serde_json::to_writer(&mut f, v)?;
73    f.write_all(b"\n")?;
74    f.flush()?;
75    Ok(())
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81    use crate::sync::IngestExportBatch;
82    use crate::sync::export_batch::SessionEvalsBatchBody;
83
84    #[test]
85    fn file_export_writes_line() {
86        let dir = tempfile::tempdir().unwrap();
87        let p = dir.path().join("t.ndjson");
88        let e = FileExporter::new(p.clone());
89        let b = IngestExportBatch::SessionEvals(SessionEvalsBatchBody { evals: vec![] });
90        e.export(&b).unwrap();
91        let s = std::fs::read_to_string(&p).unwrap();
92        let v: serde_json::Value = serde_json::from_str(s.lines().next().unwrap()).unwrap();
93        assert_eq!(
94            v.get("batch_kind").and_then(|x| x.as_str()),
95            Some("session_evals")
96        );
97    }
98}
kaizen/telemetry/file.rs

kaizen/telemetry/
file.rs
