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    Ok(SessionReader {
50        source,
51        agent,
52        opts: opts.clone(),
53        state: State::default(),
54        summary: ReadSummary::default(),
55        pending: VecDeque::new(),
56        ended: false,
57        checkpoint: None,
58        file_sidechain: false,
59        scratch: Vec::new(),
60    })
61}
62
63impl<R: BufRead> SessionReader<R> {
64    /// Does not drain unread data. StoppedEarly is retained unless next() reached
65    /// EOF or yielded a fatal error / explicit incomplete-tail result.
66    pub fn finish(self) -> ReadSummary {
67        self.summary
68    }
69    pub fn summary(&self) -> &ReadSummary {
70        &self.summary
71    }
72    fn deliver(&mut self) -> Option<Located<Event>> {
73        let event = self.pending.pop_front();
74        if self.pending.is_empty()
75            && let Some(end) = self.checkpoint.take()
76        {
77            self.summary.last_complete_byte = end;
78        }
79        event
80    }
81}
agent_sessions/reader/mod.rs

agent_sessions/reader/
mod.rs
