agent_sessions/
history.rs

1mod entry;
2use crate::*;
3pub use entry::*;
4use serde_json::Value;
5use std::{
6    fs::File,
7    io::{BufRead, BufReader},
8    path::{Path, PathBuf},
9};
10
11#[derive(Debug, Clone)]
12pub struct HistoryOptions {
13    pub read: RawReadOptions,
14    pub tail: TailMode,
15    /// False preserves legacy record-counting behavior for valid JSON while
16    /// exposing invalid optional fields in HistoryEntry.invalid_fields.
17    pub strict_fields: bool,
18}
19impl Default for HistoryOptions {
20    fn default() -> Self {
21        Self {
22            read: RawReadOptions::default(),
23            tail: TailMode::Strict,
24            strict_fields: true,
25        }
26    }
27}
28pub fn history_files(roots: &Roots) -> Vec<(Agent, PathBuf)> {
29    [
30        (Agent::ClaudeCode, &roots.claude),
31        (Agent::Codex, &roots.codex),
32    ]
33    .into_iter()
34    .filter_map(|(agent, root)| root.as_ref().map(|p| (agent, p.join("history.jsonl"))))
35    .collect()
36}
37pub struct HistoryReader<R> {
38    raw: RawReader<R>,
39    agent: Agent,
40    options: HistoryOptions,
41    errors: u64,
42    checkpoint: u64,
43    incomplete: bool,
44    ended: bool,
45}
46pub fn read_history(
47    agent: Agent,
48    path: &Path,
49    options: &HistoryOptions,
50) -> Result<HistoryReader<BufReader<File>>, ReadError> {
51    Ok(HistoryReader::new(
52        agent,
53        read_raw_file(path, &options.read)?,
54        options,
55    ))
56}
57pub fn read_history_from<R: BufRead>(
58    agent: Agent,
59    source: R,
60    options: &HistoryOptions,
61) -> Result<HistoryReader<R>, ReadError> {
62    Ok(HistoryReader::new(
63        agent,
64        read_raw_from(source, &options.read)?,
65        options,
66    ))
67}
68impl<R: BufRead> HistoryReader<R> {
69    fn new(agent: Agent, raw: RawReader<R>, options: &HistoryOptions) -> Self {
70        Self {
71            raw,
72            agent,
73            options: options.clone(),
74            errors: 0,
75            checkpoint: options.read.start_offset,
76            incomplete: false,
77            ended: false,
78        }
79    }
80    /// Next physical line, or the failed line after a stream I/O error.
81    pub fn next_line_no(&self) -> u64 {
82        self.raw.next_line_no()
83    }
84    pub fn finish(self) -> ReadSummary {
85        let raw = self.raw.finish();
86        let status = if self.incomplete {
87            ReadStatus::IncompleteTail
88        } else if raw.status == ReadStatus::Complete && self.errors > 0 {
89            ReadStatus::CompleteWithErrors
90        } else {
91            raw.status
92        };
93        ReadSummary {
94            status,
95            lines: raw.records,
96            bytes_read: raw.bytes_read,
97            last_complete_byte: self.checkpoint,
98            truncated_tail: self.incomplete,
99            line_errors: self.errors,
100            ..Default::default()
101        }
102    }
103}
104impl<R: BufRead> Iterator for HistoryReader<R> {
105    type Item = Result<Located<HistoryEntry>, StreamError>;
106    fn next(&mut self) -> Option<Self::Item> {
107        if self.ended {
108            return None;
109        }
110        loop {
111            let raw = match self.raw.next()? {
112                Ok(r) => r,
113                Err(e) => {
114                    if matches!(e, StreamError::Line { .. }) {
115                        self.errors += 1;
116                    }
117                    return Some(Err(e));
118                }
119            };
120            let utf8 = std::str::from_utf8(&raw.bytes);
121            if utf8.is_ok_and(|s| s.trim().is_empty()) {
122                if self.errors == 0 {
123                    self.checkpoint = raw.byte_end;
124                }
125                continue;
126            }
127            let entry = if utf8.is_err() {
128                Err(LineErrorKind::InvalidUtf8)
129            } else {
130                match serde_json::from_slice::<Value>(&raw.bytes) {
131                    Ok(v) => history_entry(self.agent, &v, self.options.strict_fields),
132                    Err(e)
133                        if !raw.terminated
134                            && e.is_eof()
135                            && self.options.tail == TailMode::AllowIncomplete =>
136                    {
137                        self.ended = true;
138                        self.incomplete = true;
139                        return None;
140                    }
141                    Err(_) => Err(LineErrorKind::InvalidJson),
142                }
143            };
144            return Some(match entry {
145                Ok(entry) => {
146                    if self.errors == 0 {
147                        self.checkpoint = raw.byte_end;
148                    }
149                    Ok(Located {
150                        location: Location {
151                            record_index: raw.line_no - 1,
152                            line_no: raw.line_no,
153                            byte_start: raw.byte_start,
154                            byte_end: raw.byte_end,
155                            event_index: 0,
156                        },
157                        at: entry.at,
158                        timestamp_text: None,
159                        record_id: None,
160                        record_type: None,
161                        session_id: entry.session_id.clone(),
162                        message_id: None,
163                        value: entry,
164                    })
165                }
166                Err(kind) => {
167                    self.errors += 1;
168                    Err(StreamError::Line {
169                        line_no: raw.line_no,
170                        byte_start: raw.byte_start,
171                        kind,
172                    })
173                }
174            });
175        }
176    }
177}
agent_sessions/history.rs

agent_sessions/
history.rs
