agent_sessions/reader/
mod.rs

1pub(crate) mod input;
2mod options;
3mod stream;
4use crate::parser::State;
5use crate::{Agent, Event, FileKind, Located, ReadError, SessionFile};
6pub use options::*;
7use std::{
8    collections::VecDeque,
9    fs::File,
10    io::{BufRead, BufReader},
11};
12
13/// Stateful forward-only reader. Always inspect finish() after consuming events.
14pub struct SessionReader<R = BufReader<File>> {
15    source: R,
16    agent: Agent,
17    opts: ReadOptions,
18    state: State,
19    summary: ReadSummary,
20    pending: VecDeque<Located<Event>>,
21    ended: bool,
22    checkpoint: Option<u64>,
23    file_sidechain: bool,
24    scratch: Vec<u8>,
25}
26
27pub fn read(file: &SessionFile, opts: &ReadOptions) -> Result<SessionReader, ReadError> {
28    opts.validate()?;
29    let source = File::open(&file.path).map_err(ReadError::Io)?;
30    let metadata = source.metadata().map_err(ReadError::Io)?;
31    if let Some(limit) = opts.max_file_bytes
32        && opts.stop_at_byte.unwrap_or(metadata.len()) > limit
33    {
34        return Err(ReadError::TooLarge { limit });
35    }
36    let mut reader = read_from(file.agent, BufReader::new(source), opts)?;
37    reader.file_sidechain = file.kind == FileKind::Subagent;
38    Ok(reader)
39}
40
41/// The supplied reader must start at record zero. Cumulative-only usage needs
42/// the whole prefix; this API does not infer a baseline from an arbitrary seek.
43pub fn read_from<R: BufRead>(
44    agent: Agent,
45    source: R,
46    opts: &ReadOptions,
47) -> Result<SessionReader<R>, ReadError> {
48    opts.validate()?;
49    if !agent.streaming() {
50        return Err(ReadError::InvalidOptions(
51            "use import_session for this source format",
52        ));
53    }
54    Ok(SessionReader {
55        source,
56        agent,
57        opts: opts.clone(),
58        state: State {
59            codex_content: opts.codex_content,
60            ..Default::default()
61        },
62        summary: ReadSummary::default(),
63        pending: VecDeque::new(),
64        ended: false,
65        checkpoint: None,
66        file_sidechain: false,
67        scratch: Vec::new(),
68    })
69}
70
71impl<R: BufRead> SessionReader<R> {
72    /// Does not drain unread data. StoppedEarly is retained unless next() reached
73    /// EOF or yielded a fatal error / explicit incomplete-tail result.
74    pub fn finish(self) -> ReadSummary {
75        self.summary
76    }
77    pub fn summary(&self) -> &ReadSummary {
78        &self.summary
79    }
80    fn deliver(&mut self) -> Option<Located<Event>> {
81        let event = self.pending.pop_front();
82        if self.pending.is_empty()
83            && let Some(end) = self.checkpoint.take()
84        {
85            self.summary.last_complete_byte = end;
86        }
87        event
88    }
89}
agent_sessions/reader/mod.rs

agent_sessions/reader/
mod.rs
