agent_top_core/
jsonl.rs

1//! Incremental line reader for append-only JSONL transcripts.
2//!
3//! Harnesses append to their transcript on every event, and a busy session can
4//! reach tens of megabytes. Re-parsing the whole file each second is not an
5//! option, so the reader remembers its byte offset and only returns lines that
6//! arrived since the last call. A partial trailing line (a write in progress)
7//! is held back until its newline shows up.
8
9use std::fs::File;
10use std::io::{self, Read, Seek, SeekFrom};
11use std::path::{Path, PathBuf};
12
13pub struct TailReader {
14    path: PathBuf,
15    offset: u64,
16    partial: Vec<u8>,
17}
18
19impl TailReader {
20    pub fn new(path: impl Into<PathBuf>) -> Self {
21        TailReader { path: path.into(), offset: 0, partial: Vec::new() }
22    }
23
24    pub fn path(&self) -> &Path {
25        &self.path
26    }
27
28    pub fn offset(&self) -> u64 {
29        self.offset
30    }
31
32    /// Read up to `budget` new bytes and return the complete lines in them.
33    /// Returns `(lines, more_pending)`; `more_pending` is true when the file
34    /// still has unread bytes after the budget was spent.
35    pub fn read_new_lines(&mut self, budget: usize) -> io::Result<(Vec<String>, bool)> {
36        let mut file = File::open(&self.path)?;
37        let len = file.metadata()?.len();
38        if len < self.offset {
39            // Truncated or rotated: start over.
40            self.offset = 0;
41            self.partial.clear();
42        }
43        if len == self.offset {
44            return Ok((Vec::new(), false));
45        }
46        file.seek(SeekFrom::Start(self.offset))?;
47        let want = (len - self.offset).min(budget as u64) as usize;
48        let mut buf = vec![0u8; want];
49        let mut read = 0;
50        while read < want {
51            let n = file.read(&mut buf[read..])?;
52            if n == 0 {
53                break;
54            }
55            read += n;
56        }
57        buf.truncate(read);
58        self.offset += read as u64;
59
60        let mut lines = Vec::new();
61        let mut start = 0;
62        for (i, b) in buf.iter().enumerate() {
63            if *b == b'\n' {
64                let mut line = std::mem::take(&mut self.partial);
65                line.extend_from_slice(&buf[start..i]);
66                if !line.is_empty() {
67                    lines.push(String::from_utf8_lossy(&line).into_owned());
68                }
69                start = i + 1;
70            }
71        }
72        self.partial.extend_from_slice(&buf[start..]);
73        Ok((lines, self.offset < len))
74    }
75}
76
77#[cfg(test)]
78mod tests {
79    use super::*;
80    use std::io::Write;
81
82    #[test]
83    fn reads_incrementally_and_holds_partial_lines() {
84        let dir = std::env::temp_dir().join(format!("agent-top-jsonl-{}", std::process::id()));
85        std::fs::create_dir_all(&dir).unwrap();
86        let path = dir.join("t.jsonl");
87        let mut f = File::create(&path).unwrap();
88        write!(f, "{{\"a\":1}}\n{{\"b\":2").unwrap();
89        let mut r = TailReader::new(&path);
90        let (lines, more) = r.read_new_lines(1 << 20).unwrap();
91        assert_eq!(lines, vec!["{\"a\":1}"]);
92        assert!(!more);
93        writeln!(f, "}}").unwrap();
94        let (lines, _) = r.read_new_lines(1 << 20).unwrap();
95        assert_eq!(lines, vec!["{\"b\":2}"]);
96        let _ = std::fs::remove_dir_all(&dir);
97    }
98}
agent_top_core/jsonl.rs

agent_top_core/
jsonl.rs
