nyx_agent_core/
event_log.rs

1//! Durable per-run event logs for the UI's live stream.
2
3use std::collections::HashMap;
4use std::path::{Path, PathBuf};
5
6use serde::Serialize;
7use tokio::fs::{self, File, OpenOptions};
8use tokio::io::AsyncWriteExt;
9
10use nyx_agent_types::event::AgentEvent;
11
12use crate::now_epoch_ms;
13
14#[derive(Serialize)]
15struct BorrowedRunEventLogEntry<'a> {
16    ts_ms: i64,
17    event: &'a AgentEvent,
18}
19
20/// Stable path for the event-log artifact belonging to one run.
21pub fn run_event_log_path(logs_dir: impl AsRef<Path>, run_id: &str) -> PathBuf {
22    logs_dir.as_ref().join("runs").join(format!("{}.events.jsonl", safe_run_log_segment(run_id)))
23}
24
25/// Filesystem-safe segment for a run-scoped log artifact.
26pub fn safe_run_log_segment(run_id: &str) -> String {
27    let mut out = String::with_capacity(run_id.len());
28    for ch in run_id.chars() {
29        if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
30            out.push(ch);
31        } else {
32            out.push('_');
33        }
34    }
35    let mut safe = out.trim_matches(['.', '_', '-']).to_string();
36    if safe.is_empty() {
37        safe = "run".to_string();
38    }
39    if safe.len() > 80 {
40        let mut end = 80;
41        while end > 0 && !safe.is_char_boundary(end) {
42            end -= 1;
43        }
44        safe.truncate(end);
45        safe = safe.trim_matches(['.', '_', '-']).to_string();
46        if safe.is_empty() {
47            safe = "run".to_string();
48        }
49    }
50    let hash = blake3::hash(run_id.as_bytes());
51    let hex = hash.to_hex();
52    format!("{safe}-{}", &hex.as_str()[..8])
53}
54
55/// Append-only writer that keeps file handles open while a run is active.
56#[derive(Debug)]
57pub struct RunEventLogWriter {
58    logs_dir: PathBuf,
59    files: HashMap<String, File>,
60}
61
62impl RunEventLogWriter {
63    pub fn new(logs_dir: impl Into<PathBuf>) -> Self {
64        Self { logs_dir: logs_dir.into(), files: HashMap::new() }
65    }
66
67    pub async fn append(&mut self, run_id: &str, event: &AgentEvent) -> anyhow::Result<()> {
68        let path = run_event_log_path(&self.logs_dir, run_id);
69        if !self.files.contains_key(run_id) {
70            if let Some(parent) = path.parent() {
71                fs::create_dir_all(parent).await?;
72            }
73            let file = OpenOptions::new().create(true).append(true).open(&path).await?;
74            self.files.insert(run_id.to_string(), file);
75        }
76
77        let entry = BorrowedRunEventLogEntry { ts_ms: now_epoch_ms(), event };
78        let mut line = serde_json::to_vec(&entry)?;
79        line.push(b'\n');
80        if let Some(file) = self.files.get_mut(run_id) {
81            file.write_all(&line).await?;
82        }
83        Ok(())
84    }
85
86    pub async fn finish_run(&mut self, run_id: &str) -> anyhow::Result<()> {
87        if let Some(mut file) = self.files.remove(run_id) {
88            file.flush().await?;
89        }
90        Ok(())
91    }
92}
93
94#[cfg(test)]
95mod tests {
96    use super::*;
97    use nyx_agent_types::event::RunEvent;
98
99    #[test]
100    fn safe_run_log_segment_is_stable_and_filesystem_safe() {
101        let got = safe_run_log_segment("../run one");
102        assert!(got.starts_with("run_one-"));
103        assert!(!got.contains('/'));
104        assert!(!got.contains(' '));
105    }
106
107    #[tokio::test]
108    async fn writer_appends_jsonl_entries() {
109        let tmp = tempfile::tempdir().unwrap();
110        let mut writer = RunEventLogWriter::new(tmp.path());
111        let event = AgentEvent::Run {
112            data: RunEvent::RunStarted {
113                run_id: "run-1".to_string(),
114                project_id: "project-1".to_string(),
115                repos: vec!["repo".to_string()],
116                started_at_ms: 1,
117            },
118        };
119        writer.append("run-1", &event).await.unwrap();
120        writer.finish_run("run-1").await.unwrap();
121
122        let body = fs::read_to_string(run_event_log_path(tmp.path(), "run-1")).await.unwrap();
123        assert!(body.contains("\"ts_ms\""));
124        assert!(body.contains("\"RunStarted\""));
125    }
126}
nyx_agent_core/event_log.rs

nyx_agent_core/
event_log.rs
