toolpath_codex/
reader.rs

1//! Parse Codex rollout JSONL files.
2//!
3//! The writer is append-only but backgrounded — a crashed Codex
4//! process may leave the final line mid-write. We skip unparseable
5//! lines by default and surface them as warnings rather than failing
6//! the whole read.
7
8use crate::error::{ConvoError, Result};
9use crate::types::{RolloutLine, Session};
10use std::fs::File;
11use std::io::{BufRead, BufReader};
12use std::path::{Path, PathBuf};
13
14pub struct RolloutReader;
15
16impl RolloutReader {
17    /// Read every line of a rollout file into a [`Session`].
18    ///
19    /// The session id is taken from the first line's `session_meta`
20    /// payload if present; otherwise from the filename stem.
21    pub fn read_session<P: AsRef<Path>>(path: P) -> Result<Session> {
22        let path = path.as_ref();
23        if !path.exists() {
24            return Err(ConvoError::SessionNotFound(path.display().to_string()));
25        }
26
27        let file = File::open(path)?;
28        let reader = BufReader::new(file);
29        let mut lines: Vec<RolloutLine> = Vec::new();
30        for (idx, raw) in reader.lines().enumerate() {
31            let raw = match raw {
32                Ok(s) => s,
33                Err(e) => {
34                    eprintln!(
35                        "Warning: IO error reading {} line {}: {}",
36                        path.display(),
37                        idx + 1,
38                        e
39                    );
40                    continue;
41                }
42            };
43            if raw.trim().is_empty() {
44                continue;
45            }
46            match serde_json::from_str::<RolloutLine>(&raw) {
47                Ok(line) => lines.push(line),
48                Err(e) => {
49                    // Tolerate a single truncated last line (common after crashes);
50                    // warn about anything else.
51                    if std::env::var("CODEX_ROLLOUT_STRICT").is_ok() {
52                        return Err(ConvoError::Json(e));
53                    }
54                    eprintln!(
55                        "Warning: unparseable rollout line {} in {}: {}",
56                        idx + 1,
57                        path.file_name().and_then(|n| n.to_str()).unwrap_or("<?>"),
58                        e
59                    );
60                }
61            }
62        }
63
64        let id = Self::derive_session_id(&lines, path);
65        Ok(Session {
66            id,
67            file_path: path.to_path_buf(),
68            lines,
69        })
70    }
71
72    /// Peek just the first `session_meta` payload without fully parsing
73    /// the rest of the file. Returns the session id if found.
74    pub fn peek_session_id<P: AsRef<Path>>(path: P) -> Option<String> {
75        let file = File::open(path).ok()?;
76        let mut reader = BufReader::new(file);
77        let mut first = String::new();
78        reader.read_line(&mut first).ok()?;
79        let line: RolloutLine = serde_json::from_str(first.trim()).ok()?;
80        if line.kind != "session_meta" {
81            return None;
82        }
83        line.payload
84            .get("id")
85            .and_then(|v| v.as_str())
86            .map(str::to_string)
87    }
88
89    /// Return the byte-length of a rollout file.
90    pub fn file_size<P: AsRef<Path>>(path: P) -> Result<u64> {
91        let path = path.as_ref();
92        if !path.exists() {
93            return Err(ConvoError::SessionNotFound(path.display().to_string()));
94        }
95        Ok(std::fs::metadata(path)?.len())
96    }
97
98    fn derive_session_id(lines: &[RolloutLine], path: &Path) -> String {
99        // Prefer the session_meta payload.
100        if let Some(first) = lines.first()
101            && first.kind == "session_meta"
102            && let Some(id) = first.payload.get("id").and_then(|v| v.as_str())
103        {
104            return id.to_string();
105        }
106        // Fall back to the UUID suffix of the filename stem.
107        if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
108            return crate::paths::session_id_from_stem(stem).to_string();
109        }
110        "unknown".to_string()
111    }
112}
113
114/// Type alias exposed for consumers to avoid re-importing `PathBuf`.
115pub type RolloutPath = PathBuf;
116
117#[cfg(test)]
118mod tests {
119    use super::*;
120    use std::io::Write;
121    use tempfile::NamedTempFile;
122
123    fn sample_rollout() -> String {
124        [
125            r#"{"timestamp":"2026-04-20T16:44:37.772Z","type":"session_meta","payload":{"id":"019dabc6-8fef-7681-a054-b5bb75fcb97d","timestamp":"2026-04-20T16:43:30.171Z","cwd":"/tmp/proj","originator":"codex-tui","cli_version":"0.118.0","source":"cli"}}"#,
126            r#"{"timestamp":"2026-04-20T16:44:37.773Z","type":"turn_context","payload":{"turn_id":"019dabc7","cwd":"/tmp/proj"}}"#,
127            r#"{"timestamp":"2026-04-20T16:44:37.775Z","type":"event_msg","payload":{"type":"task_started","turn_id":"019dabc7"}}"#,
128            r#"{"timestamp":"2026-04-20T16:44:38.000Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"hello"}]}}"#,
129        ]
130        .join("\n")
131    }
132
133    fn write_fixture(body: &str) -> NamedTempFile {
134        let mut f = NamedTempFile::new().unwrap();
135        f.write_all(body.as_bytes()).unwrap();
136        f.flush().unwrap();
137        f
138    }
139
140    #[test]
141    fn read_session_basic() {
142        let f = write_fixture(&sample_rollout());
143        let s = RolloutReader::read_session(f.path()).unwrap();
144        assert_eq!(s.id, "019dabc6-8fef-7681-a054-b5bb75fcb97d");
145        assert_eq!(s.lines.len(), 4);
146        assert!(s.meta().is_some());
147    }
148
149    #[test]
150    fn read_session_nonexistent_errors() {
151        let err = RolloutReader::read_session("/nonexistent").unwrap_err();
152        assert!(matches!(err, ConvoError::SessionNotFound(_)));
153    }
154
155    /// Serializes access to `CODEX_ROLLOUT_STRICT` across tests in this
156    /// module. Two tests probe `read_session` with opposing strictness
157    /// expectations; without serialization, cargo test's threaded
158    /// runner can observe the env var set by one test during another.
159    fn strict_env_lock() -> std::sync::MutexGuard<'static, ()> {
160        use std::sync::{Mutex, OnceLock};
161        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
162        LOCK.get_or_init(|| Mutex::new(()))
163            .lock()
164            .unwrap_or_else(|p| p.into_inner())
165    }
166
167    #[test]
168    fn read_session_handles_truncated_last_line() {
169        let _g = strict_env_lock();
170        // Belt-and-braces: even under the lock, make sure the env var
171        // is clear before we observe lenient behavior.
172        unsafe { std::env::remove_var("CODEX_ROLLOUT_STRICT") };
173        // Good first line, garbage second — reader skips and warns.
174        let body = sample_rollout() + "\n{\"timestamp\":\"broken"; // truncated
175        let f = write_fixture(&body);
176        let s = RolloutReader::read_session(f.path()).unwrap();
177        assert_eq!(s.lines.len(), 4, "truncated line dropped, others kept");
178    }
179
180    #[test]
181    fn read_session_respects_strict_env() {
182        let _g = strict_env_lock();
183        let body = sample_rollout() + "\n{\"timestamp\":\"broken";
184        let f = write_fixture(&body);
185        unsafe { std::env::set_var("CODEX_ROLLOUT_STRICT", "1") };
186        let err = RolloutReader::read_session(f.path()).unwrap_err();
187        unsafe { std::env::remove_var("CODEX_ROLLOUT_STRICT") };
188        assert!(matches!(err, ConvoError::Json(_)));
189    }
190
191    #[test]
192    fn peek_session_id_reads_first_line_only() {
193        let f = write_fixture(&sample_rollout());
194        let id = RolloutReader::peek_session_id(f.path()).unwrap();
195        assert_eq!(id, "019dabc6-8fef-7681-a054-b5bb75fcb97d");
196    }
197
198    #[test]
199    fn peek_session_id_missing_when_first_line_not_meta() {
200        let body = r#"{"timestamp":"t","type":"event_msg","payload":{"type":"x"}}"#;
201        let f = write_fixture(body);
202        assert!(RolloutReader::peek_session_id(f.path()).is_none());
203    }
204
205    #[test]
206    fn session_started_at_and_last_activity() {
207        let f = write_fixture(&sample_rollout());
208        let s = RolloutReader::read_session(f.path()).unwrap();
209        assert!(s.started_at().is_some());
210        assert!(s.last_activity() >= s.started_at());
211    }
212
213    #[test]
214    fn session_first_user_text() {
215        let f = write_fixture(&sample_rollout());
216        let s = RolloutReader::read_session(f.path()).unwrap();
217        assert_eq!(s.first_user_text().as_deref(), Some("hello"));
218    }
219
220    #[test]
221    fn file_size_works() {
222        let f = write_fixture(&sample_rollout());
223        let size = RolloutReader::file_size(f.path()).unwrap();
224        assert!(size > 0);
225    }
226
227    #[test]
228    fn derive_session_id_falls_back_to_stem_uuid() {
229        let body = r#"{"timestamp":"t","type":"event_msg","payload":{"type":"x"}}"#;
230        let f = NamedTempFile::new().unwrap();
231        let path = f
232            .path()
233            .parent()
234            .unwrap()
235            .join("rollout-2026-04-20T10-00-00-019dabc6-8fef-7681-a054-b5bb75fcb97d.jsonl");
236        std::fs::write(&path, body).unwrap();
237        let s = RolloutReader::read_session(&path).unwrap();
238        assert_eq!(s.id, "019dabc6-8fef-7681-a054-b5bb75fcb97d");
239        // Clean up
240        drop(f);
241        let _ = std::fs::remove_file(path);
242    }
243}
toolpath_codex/reader.rs

toolpath_codex/
reader.rs
