magi_code/sessions/

read.rs

use super::event::SessionEvent;
use super::manager::Session;
use sha2::{Digest, Sha256};
#[cfg(test)]
use std::time::SystemTime;
use std::{
    collections::VecDeque,
    fs,
    io::{BufRead, BufReader, Read, Seek, SeekFrom},
    path::PathBuf,
};
pub fn validate_session_id(id: String) -> anyhow::Result<String> {
    if id.is_empty() {
        anyhow::bail!("session id must not be empty");
    }
    if id == "." || id == ".." || id.contains("..") {
        anyhow::bail!("session id must not contain '..'");
    }
    if id.contains('/') || id.contains('\\') {
        anyhow::bail!("session id must not contain path separators");
    }
    if PathBuf::from(&id).is_absolute() {
        anyhow::bail!("session id must not be an absolute path");
    }
    if !id
        .chars()
        .all(|character| character.is_ascii_alphanumeric() || character == '_' || character == '-')
    {
        anyhow::bail!("session id must match [A-Za-z0-9_-]+");
    }
    Ok(id)
}

fn open_session_file(session: &Session) -> anyhow::Result<fs::File> {
    let root = session
        .path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("session file has no parent"))?;
    super::store::open_existing_primary(root, &session.id)?
        .ok_or_else(|| anyhow::anyhow!("session JSONL is missing"))
}
#[cfg(test)]
pub(crate) fn latest_valid_event_timestamp_streaming(session: &Session) -> Option<SystemTime> {
    let root = session.path.parent()?;
    let file = super::store::open_existing_primary(root, &session.id).ok()??;
    let mut latest = None;
    for line in BufReader::new(file).lines() {
        let Ok(line) = line else {
            continue;
        };
        if line.trim().is_empty() {
            continue;
        }
        let Ok(event) = serde_json::from_str::<SessionEvent>(&line) else {
            continue;
        };
        let timestamp = SystemTime::from(event.timestamp);
        latest = Some(latest.map_or(timestamp, |current: SystemTime| current.max(timestamp)));
    }
    latest
}

#[derive(Debug)]
pub(crate) enum BoundedReadError {
    BudgetExceeded(String),
}

impl std::fmt::Display for BoundedReadError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BudgetExceeded(message) => formatter.write_str(message),
        }
    }
}

impl std::error::Error for BoundedReadError {}

fn budget_error(message: impl Into<String>) -> anyhow::Error {
    anyhow::Error::new(BoundedReadError::BudgetExceeded(message.into()))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SessionReadDiagnostic {
    pub(crate) line: usize,
    pub(crate) message: String,
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) struct TolerantSessionEvents {
    pub(crate) events: Vec<SessionEvent>,
    pub(crate) diagnostics: Vec<SessionReadDiagnostic>,
    pub(crate) cutoff_bytes: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct SessionReadStats {
    pub lines_read: usize,
    pub bytes_read: usize,
    pub content_digest: [u8; 32],
}

const MAX_TOLERANT_READ_DIAGNOSTICS: usize = 64;

fn push_tolerant_read_diagnostic(
    diagnostics: &mut Vec<SessionReadDiagnostic>,
    omitted_count: &mut usize,
    diagnostic: SessionReadDiagnostic,
) {
    if diagnostics.len() < MAX_TOLERANT_READ_DIAGNOSTICS {
        diagnostics.push(diagnostic);
    } else {
        *omitted_count = omitted_count.saturating_add(1);
    }
}

fn finalize_tolerant_read_diagnostics(
    mut diagnostics: Vec<SessionReadDiagnostic>,
    omitted_count: usize,
) -> Vec<SessionReadDiagnostic> {
    if omitted_count == 0 {
        return diagnostics;
    }
    if diagnostics.len() == MAX_TOLERANT_READ_DIAGNOSTICS {
        diagnostics.pop();
    }
    diagnostics.push(SessionReadDiagnostic {
        line: 0,
        message: format!(
            "omitted {omitted_count} additional session JSONL diagnostics after cap of {MAX_TOLERANT_READ_DIAGNOSTICS}"
        ),
    });
    diagnostics
}

pub(crate) const MAX_METADATA_VISIT_LINES: usize = 100_000;
pub(crate) const MAX_METADATA_VISIT_BYTES: usize = 64 * 1024 * 1024;

impl Session {
    #[cfg(test)]
    pub(crate) fn read_events(&self) -> anyhow::Result<Vec<SessionEvent>> {
        validate_session_id(self.id.clone())?;
        if !self.path.exists() {
            return Ok(Vec::new());
        }
        self.read_event_lines_streaming()?
            .map(|event_line| event_line.map(|(event, _)| event))
            .collect()
    }

    #[cfg(test)]
    pub(crate) fn read_recent_events(
        &self,
        max_events: usize,
        max_bytes: usize,
    ) -> anyhow::Result<Vec<SessionEvent>> {
        validate_session_id(self.id.clone())?;
        if !self.path.exists() || max_events == 0 || max_bytes == 0 {
            return Ok(Vec::new());
        }
        let mut retained = VecDeque::new();
        let mut retained_bytes = 0usize;
        for event_line in self.read_event_lines_streaming()? {
            let (event, line_bytes) = event_line?;
            // Intentional: use original JSONL line bytes, not reserialized event size; preserve this hardening.
            retained_bytes = retained_bytes.saturating_add(line_bytes);
            retained.push_back((event, line_bytes));
            while retained.len() > max_events || retained_bytes > max_bytes {
                if let Some((_, bytes)) = retained.pop_front() {
                    retained_bytes = retained_bytes.saturating_sub(bytes);
                } else {
                    break;
                }
            }
        }
        Ok(retained.into_iter().map(|(event, _)| event).collect())
    }

    #[cfg(test)]
    pub(crate) fn read_events_tolerant(&self) -> anyhow::Result<TolerantSessionEvents> {
        self.read_events_tolerant_bounded(usize::MAX, usize::MAX)
    }

    pub(crate) fn read_events_tolerant_bounded(
        &self,
        max_lines: usize,
        max_bytes: usize,
    ) -> anyhow::Result<TolerantSessionEvents> {
        Ok(self
            .read_events_tolerant_bounded_with_stats(max_lines, max_bytes)?
            .0)
    }

    pub(crate) fn read_events_tolerant_bounded_with_stats(
        &self,
        max_lines: usize,
        max_bytes: usize,
    ) -> anyhow::Result<(TolerantSessionEvents, SessionReadStats)> {
        let mut events = Vec::new();
        let (diagnostics, cutoff_bytes, stats) =
            self.visit_events_tolerant_bounded_with_stats(max_lines, max_bytes, |event| {
                events.push(event)
            })?;
        Ok((
            TolerantSessionEvents {
                events,
                diagnostics,
                cutoff_bytes: cutoff_bytes as u64,
            },
            stats,
        ))
    }

    pub(crate) fn content_fingerprint_bounded(
        &self,
        max_bytes: usize,
    ) -> anyhow::Result<([u8; 32], usize)> {
        validate_session_id(self.id.clone())?;
        if !self.path.exists() {
            return Ok((Sha256::digest([]).into(), 0));
        }
        let file = open_session_file(self)?;
        let mut reader = file.take(
            u64::try_from(max_bytes)
                .unwrap_or(u64::MAX)
                .saturating_add(1),
        );
        let mut digest = Sha256::new();
        let mut total_bytes = 0usize;
        let mut buffer = [0u8; 8192];
        loop {
            let read = reader.read(&mut buffer)?;
            if read == 0 {
                break;
            }
            total_bytes = total_bytes.saturating_add(read);
            if total_bytes > max_bytes {
                return Err(budget_error(format!(
                    "session JSONL tolerant read limit exceeded: {max_bytes} bytes"
                )));
            }
            digest.update(&buffer[..read]);
        }
        Ok((digest.finalize().into(), total_bytes))
    }

    pub(crate) fn visit_events_tolerant_bounded(
        &self,
        max_lines: usize,
        max_bytes: usize,
        visit: impl FnMut(SessionEvent),
    ) -> anyhow::Result<(Vec<SessionReadDiagnostic>, usize)> {
        let (diagnostics, bytes, _) =
            self.visit_events_tolerant_bounded_with_stats(max_lines, max_bytes, visit)?;
        Ok((diagnostics, bytes))
    }

    fn visit_events_tolerant_bounded_with_stats(
        &self,
        max_lines: usize,
        max_bytes: usize,
        mut visit: impl FnMut(SessionEvent),
    ) -> anyhow::Result<(Vec<SessionReadDiagnostic>, usize, SessionReadStats)> {
        validate_session_id(self.id.clone())?;
        if !self.path.exists() {
            return Ok((
                Vec::new(),
                0,
                SessionReadStats {
                    lines_read: 0,
                    bytes_read: 0,
                    content_digest: Sha256::digest([]).into(),
                },
            ));
        }
        let file = open_session_file(self)?;
        let mut reader = BufReader::new(file);
        let mut diagnostics = Vec::new();
        let mut omitted_diagnostics = 0usize;
        let mut total_bytes = 0usize;
        let mut lines_read = 0usize;
        let mut digest = Sha256::new();
        let mut line = Vec::new();
        for line_number in 1..=max_lines {
            line.clear();
            let remaining = max_bytes.saturating_sub(total_bytes);
            if remaining == 0 {
                if !reader.fill_buf()?.is_empty() {
                    return Err(budget_error(format!(
                        "session JSONL tolerant read limit exceeded: {max_bytes} bytes"
                    )));
                }
                break;
            }
            let read = (&mut reader)
                .take(
                    u64::try_from(remaining)
                        .unwrap_or(u64::MAX)
                        .saturating_add(1),
                )
                .read_until(b'\n', &mut line)?;
            if read == 0 {
                break;
            }
            if read > remaining {
                return Err(budget_error(format!(
                    "session JSONL tolerant read limit exceeded at line {line_number}: {max_bytes} bytes"
                )));
            }
            total_bytes = total_bytes.saturating_add(read);
            lines_read = lines_read.saturating_add(1);
            digest.update(&line);
            if line.last() == Some(&b'\n') {
                line.pop();
                if line.last() == Some(&b'\r') {
                    line.pop();
                }
            }
            match serde_json::from_slice::<SessionEvent>(&line) {
                Ok(event) => visit(event),
                Err(_) => push_tolerant_read_diagnostic(
                    &mut diagnostics,
                    &mut omitted_diagnostics,
                    SessionReadDiagnostic {
                        line: line_number,
                        message: format!(
                            "operation=replay category=session_jsonl failed to parse session JSONL at line {line_number}"
                        ),
                    },
                ),
            }
        }
        if !reader.fill_buf()?.is_empty() {
            return Err(budget_error(format!(
                "session JSONL tolerant read limit exceeded: more than {max_lines} lines or {max_bytes} bytes"
            )));
        }
        let diagnostics = finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics);
        Ok((
            diagnostics,
            total_bytes,
            SessionReadStats {
                lines_read,
                bytes_read: total_bytes,
                content_digest: digest.finalize().into(),
            },
        ))
    }

    pub(crate) fn read_recent_events_tolerant(
        &self,
        max_events: usize,
        max_bytes: usize,
    ) -> anyhow::Result<TolerantSessionEvents> {
        validate_session_id(self.id.clone())?;
        if !self.path.exists() || max_events == 0 || max_bytes == 0 {
            return Ok(TolerantSessionEvents {
                events: Vec::new(),
                diagnostics: Vec::new(),
                cutoff_bytes: 0,
            });
        }
        let file = open_session_file(self)?;
        let lines = BufReader::new(file)
            .split(b'\n')
            .enumerate()
            .map(|(index, line)| (index + 1, line));
        self.collect_recent_events_tolerant_lines(lines, max_events, max_bytes)
    }

    pub(crate) fn read_recent_events_tolerant_tail(
        &self,
        max_events: usize,
        max_retained_bytes: usize,
        max_read_bytes: usize,
    ) -> anyhow::Result<TolerantSessionEvents> {
        validate_session_id(self.id.clone())?;
        if max_events == 0 || max_retained_bytes == 0 || max_read_bytes == 0 {
            anyhow::bail!(
                "session JSONL tolerant tail read limits must be non-zero: max_events={max_events}, max_retained_bytes={max_retained_bytes}, max_read_bytes={max_read_bytes}"
            );
        }
        if !self.path.exists() {
            return Ok(TolerantSessionEvents {
                events: Vec::new(),
                diagnostics: Vec::new(),
                cutoff_bytes: 0,
            });
        }
        // Pin the byte window on the same validated handle. In particular, a file that was
        // small at stat time must not fall back to an unbounded read while a writer appends.
        let mut file = open_session_file(self)?;
        let file_len = file.metadata()?.len();
        let read_bytes = file_len.min(u64::try_from(max_read_bytes).unwrap_or(u64::MAX));
        let start = file_len - read_bytes;
        file.seek(SeekFrom::Start(start))?;
        let mut tail = Vec::with_capacity(read_bytes as usize);
        file.take(read_bytes).read_to_end(&mut tail)?;

        let first_complete_line = if start == 0 {
            0
        } else {
            tail.iter()
                .position(|byte| *byte == b'\n')
                .map_or(tail.len(), |index| index + 1)
        };
        let tail = &tail[first_complete_line..];
        let mut line_count = 0;
        let lines = tail
            .split_inclusive(|byte| *byte == b'\n')
            .enumerate()
            .inspect(|_| line_count += 1)
            .map(|(index, line)| {
                let line = line.strip_suffix(b"\n").unwrap_or(line);
                (index + 1, Ok(line.to_vec()))
            });
        let mut result =
            self.collect_recent_events_tolerant_lines(lines, max_events, max_retained_bytes)?;
        if start > 0 || (line_count > result.events.len() && result.diagnostics.is_empty()) {
            result.diagnostics.insert(
            0,
            SessionReadDiagnostic {
                line: 0,
                message: format!(
                    "operation=recent_context category=session_jsonl bounded tail window; omitted older lines; read final {max_read_bytes} of {file_len} bytes"
                ),
            },
        );
        }
        Ok(result)
    }

    fn collect_recent_events_tolerant_lines(
        &self,
        lines: impl IntoIterator<Item = (usize, Result<Vec<u8>, std::io::Error>)>,
        max_events: usize,
        max_bytes: usize,
    ) -> anyhow::Result<TolerantSessionEvents> {
        let mut retained = VecDeque::new();
        let mut retained_bytes = 0usize;
        let mut malformed_bytes = 0usize;
        let mut diagnostics = Vec::new();
        let mut omitted_diagnostics = 0usize;
        for (line_number, line) in lines {
            match line {
                Ok(line) => {
                    let line_bytes = line.len() + 1;
                    match serde_json::from_slice::<SessionEvent>(&line) {
                        Ok(event) => {
                            retained_bytes = retained_bytes.saturating_add(line_bytes);
                            retained.push_back((event, line_bytes));
                            while retained.len() > max_events || retained_bytes > max_bytes {
                                if let Some((_, bytes)) = retained.pop_front() {
                                    retained_bytes = retained_bytes.saturating_sub(bytes);
                                } else {
                                    break;
                                }
                            }
                        }
                        Err(_) => {
                            malformed_bytes = malformed_bytes.saturating_add(line_bytes);
                            if malformed_bytes > max_bytes {
                                anyhow::bail!(
                                    "operation=recent_context category=session_jsonl tolerant recent read limit exceeded: malformed bytes > {max_bytes}"
                                );
                            }
                            push_tolerant_read_diagnostic(
                                &mut diagnostics,
                                &mut omitted_diagnostics,
                                SessionReadDiagnostic {
                                    line: line_number,
                                    message: format!(
                                        "operation=recent_context category=session_jsonl failed to parse session JSONL at line {line_number}"
                                    ),
                                },
                            );
                        }
                    }
                }
                Err(_) => push_tolerant_read_diagnostic(
                    &mut diagnostics,
                    &mut omitted_diagnostics,
                    SessionReadDiagnostic {
                        line: line_number,
                        message: format!(
                            "operation=recent_context category=session_jsonl read failure at line {line_number}"
                        ),
                    },
                ),
            }
        }
        Ok(TolerantSessionEvents {
            events: retained.into_iter().map(|(event, _)| event).collect(),
            diagnostics: finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics),
            cutoff_bytes: 0,
        })
    }

    #[cfg(test)]
    pub(crate) fn latest_event_timestamp_bounded(&self) -> anyhow::Result<Option<SystemTime>> {
        validate_session_id(self.id.clone())?;
        if !self.path.exists() {
            return Ok(None);
        }
        let mut latest = None;
        for event_line in self.read_event_lines_streaming()? {
            let timestamp = SystemTime::from(event_line?.0.timestamp);
            latest = Some(latest.map_or(timestamp, |current: SystemTime| current.max(timestamp)));
        }
        Ok(latest)
    }

    #[cfg(test)]
    fn read_event_lines_streaming(
        &self,
    ) -> anyhow::Result<impl Iterator<Item = anyhow::Result<(SessionEvent, usize)>> + '_> {
        let file = open_session_file(self)?;
        Ok(BufReader::new(file)
            .lines()
            .enumerate()
            .map(|(index, line)| {
                let line = line.map_err(|error| {
                    anyhow::anyhow!(
                        "failed to read session JSONL at {} line {}: {error}",
                        self.path.display(),
                        index + 1
                    )
                })?;
                // Intentional accounting: retain the original line size to avoid per-event reserialization.
                let line_bytes = line.len() + 1;
                let event = serde_json::from_str::<SessionEvent>(&line).map_err(|error| {
                    anyhow::anyhow!(
                        "failed to parse session JSONL at {} line {}: {error}",
                        self.path.display(),
                        index + 1
                    )
                })?;
                Ok((event, line_bytes))
            }))
    }
}

#[cfg(test)]
mod tests {
    use super::super::manager::SessionManager;
    use super::*;
    use proptest::prelude::*;
    use serde_json::json;
    use tempfile::TempDir;

    fn valid_session_id_strategy() -> impl Strategy<Value = String> {
        proptest::string::string_regex("[A-Za-z0-9_-]{1,64}").unwrap()
    }

    fn invalid_session_id_strategy() -> impl Strategy<Value = String> {
        prop_oneof![
            Just(String::new()),
            Just(".".to_string()),
            any::<String>().prop_map(|value| format!("{value}..")),
            any::<String>().prop_map(|value| format!("{value}/{value}")),
            any::<String>().prop_map(|value| format!("{value}\\{value}")),
            any::<String>().prop_map(|value| format!("{value}.jsonl")),
            any::<String>().prop_map(|value| format!("{value}é")),
        ]
    }

    fn invalid_utf8_event_line(event: &SessionEvent, marker: &[u8]) -> Vec<u8> {
        let mut line = serde_json::to_vec(event).unwrap();
        let marker_start = line
            .windows(marker.len())
            .position(|window| window == marker)
            .expect("test marker must be present in serialized event");
        line[marker_start] = 0xff;
        line
    }

    proptest! {
        #[test]
        fn validate_session_id_accepts_only_non_empty_safe_ascii_ids(id in valid_session_id_strategy()) {
            let validated = validate_session_id(id.clone()).unwrap();

            prop_assert_eq!(&validated, &id);
            prop_assert!(!validated.is_empty());
            prop_assert!(validated
                .chars()
                .all(|character| character.is_ascii_alphanumeric() || character == '_' || character == '-'));
        }

        #[test]
        fn validate_session_id_rejects_generated_unsafe_ids(id in invalid_session_id_strategy()) {
            prop_assert!(validate_session_id(id).is_err());
        }
    }

    #[test]
    fn recent_and_latest_session_reads_are_bounded_streaming_paths() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.create().unwrap();
        for index in 0..25 {
            session
                .append(&SessionEvent::new(
                    "event",
                    session.id().to_string(),
                    temp.path().to_path_buf(),
                    json!({"index": index}),
                ))
                .unwrap();
        }
        let recent = session.read_recent_events(3, 4096).unwrap();
        assert_eq!(recent.len(), 3);
        assert_eq!(recent[0].payload["index"], 22);
        assert!(session.latest_event_timestamp_bounded().unwrap().is_some());
        assert_eq!(manager.most_recent().unwrap().unwrap().id(), session.id());
    }

    #[test]
    fn read_recent_events_uses_original_jsonl_line_bytes() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.create().unwrap();
        let padded = SessionEvent::new(
            "event",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"index": 1}),
        );
        let recent = SessionEvent::new(
            "event",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"index": 2}),
        );
        let padded_compact_len = serde_json::to_vec(&padded).unwrap().len();
        let mut padded_line = serde_json::to_value(&padded).unwrap();
        padded_line["padding"] = json!("x".repeat(2048));
        let padded_line = serde_json::to_string(&padded_line).unwrap();
        let recent_line = serde_json::to_string(&recent).unwrap();
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        fs::write(session.path(), format!("{padded_line}\n{recent_line}\n")).unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        let max_bytes = padded_compact_len + recent_line.len() + 2;
        let events = session.read_recent_events(10, max_bytes).unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].payload["index"], 2);
    }

    #[test]
    fn tolerant_read_reports_malformed_lines_without_payload_leak() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.open("safe").unwrap();
        let first = SessionEvent::new(
            "user_input",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"before"}),
        );
        let second = SessionEvent::new(
            "assistant_output",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"after"}),
        );
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        fs::write(
            session.path(),
            format!(
                "{}\n{{\"access_token\":\"secret-token\",\n{}\n",
                serde_json::to_string(&first).unwrap(),
                serde_json::to_string(&second).unwrap()
            ),
        )
        .unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        assert!(session.read_events().is_err());
        let tolerant = session.read_events_tolerant().unwrap();

        assert_eq!(tolerant.events.len(), 2);
        assert_eq!(tolerant.events[0].payload["text"], "before");
        assert_eq!(tolerant.events[1].payload["text"], "after");
        assert_eq!(tolerant.diagnostics.len(), 1);
        assert_eq!(tolerant.diagnostics[0].line, 2);
        assert!(tolerant.diagnostics[0].message.contains("line 2"));
        assert!(!tolerant.diagnostics[0].message.contains("secret-token"));
    }

    #[test]
    fn tolerant_bounded_read_rejects_invalid_utf8_without_replacement_history() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.open("safe").unwrap();
        let before = SessionEvent::new(
            "user_input",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"before"}),
        );
        let invalid = SessionEvent::new(
            "assistant_output",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"invalid-secret"}),
        );
        let after = SessionEvent::new(
            "assistant_output",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"after"}),
        );
        let mut content = Vec::new();
        content.extend(serde_json::to_vec(&before).unwrap());
        content.push(b'\n');
        content.extend(invalid_utf8_event_line(&invalid, b"invalid-secret"));
        content.push(b'\n');
        content.extend(serde_json::to_vec(&after).unwrap());
        content.push(b'\n');
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        fs::write(session.path(), &content).unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        let tolerant = session
            .read_events_tolerant_bounded(10, content.len())
            .unwrap();

        assert_eq!(tolerant.events.len(), 2);
        assert_eq!(tolerant.events[0].payload["text"], "before");
        assert_eq!(tolerant.events[1].payload["text"], "after");
        assert_eq!(tolerant.diagnostics.len(), 1);
        assert_eq!(tolerant.diagnostics[0].line, 2);
        assert!(!tolerant.diagnostics[0].message.contains("invalid-secret"));
        assert!(!tolerant.diagnostics[0].message.contains('\u{fffd}'));
    }

    #[test]
    fn tolerant_recent_paths_reject_invalid_utf8_without_replacement_history() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.open("safe").unwrap();
        let prefix = SessionEvent::new(
            "user_input",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"prefix"}),
        );
        let invalid = SessionEvent::new(
            "assistant_output",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"invalid-secret"}),
        );
        let after = SessionEvent::new(
            "assistant_output",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"after"}),
        );
        let prefix_line = serde_json::to_vec(&prefix).unwrap();
        let invalid_line = invalid_utf8_event_line(&invalid, b"invalid-secret");
        let after_line = serde_json::to_vec(&after).unwrap();
        let mut content = Vec::new();
        content.extend(&prefix_line);
        content.push(b'\n');
        content.extend(&invalid_line);
        content.push(b'\n');
        content.extend(&after_line);
        content.push(b'\n');
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        fs::write(session.path(), &content).unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        let recent = session
            .read_recent_events_tolerant(10, content.len())
            .unwrap();
        assert_eq!(recent.events.len(), 2);
        assert_eq!(recent.events[0].payload["text"], "prefix");
        assert_eq!(recent.events[1].payload["text"], "after");
        assert!(
            recent
                .diagnostics
                .iter()
                .any(|diagnostic| diagnostic.line == 2)
        );

        let max_read_bytes = invalid_line.len() + after_line.len() + 4;
        let tail = session
            .read_recent_events_tolerant_tail(10, content.len(), max_read_bytes)
            .unwrap();
        assert_eq!(tail.events.len(), 1);
        assert_eq!(tail.events[0].payload["text"], "after");
        assert!(
            tail.diagnostics
                .iter()
                .any(|diagnostic| diagnostic.line == 1)
        );
        assert!(
            !tail
                .diagnostics
                .iter()
                .any(|diagnostic| diagnostic.message.contains("invalid-secret"))
        );
        assert!(
            !tail
                .diagnostics
                .iter()
                .any(|diagnostic| diagnostic.message.contains('\u{fffd}'))
        );
    }

    #[test]
    fn tolerant_session_read_caps_malformed_diagnostics() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.open("safe").unwrap();
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        let mut lines = Vec::new();
        for index in 0..(MAX_TOLERANT_READ_DIAGNOSTICS + 10) {
            lines.push(format!("not json {index}"));
        }
        fs::write(session.path(), lines.join("\n")).unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        let tolerant = session.read_events_tolerant().unwrap();

        assert!(tolerant.events.is_empty());
        assert_eq!(tolerant.diagnostics.len(), MAX_TOLERANT_READ_DIAGNOSTICS);
        let summary = tolerant.diagnostics.last().unwrap();
        assert_eq!(summary.line, 0);
        assert!(summary.message.contains("omitted 10 additional"));
        assert!(summary.message.contains("cap of 64"));
        assert!(
            !tolerant
                .diagnostics
                .iter()
                .any(|diagnostic| diagnostic.message.contains("not json"))
        );
    }

    #[test]
    fn read_recent_events_tolerant_preserves_valid_events_after_malformed_line() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.open("safe").unwrap();
        let old = SessionEvent::new(
            "user_input",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"old"}),
        );
        let recent = SessionEvent::new(
            "assistant_output",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"recent"}),
        );
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        fs::write(
            session.path(),
            format!(
                "{}\nnot json\n{}\n",
                serde_json::to_string(&old).unwrap(),
                serde_json::to_string(&recent).unwrap()
            ),
        )
        .unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        let tolerant = session.read_recent_events_tolerant(10, 4096).unwrap();

        assert_eq!(tolerant.events.len(), 2);
        assert_eq!(tolerant.events[1].payload["text"], "recent");
        assert_eq!(tolerant.diagnostics[0].line, 2);
        assert!(!tolerant.diagnostics[0].message.contains("not json"));
    }

    #[test]
    fn read_recent_events_tolerant_bounds_malformed_input_bytes() {
        let temp = TempDir::new().unwrap();
        let manager = SessionManager::new(temp.path().join("sessions"));
        let session = manager.open("safe").unwrap();
        let valid = SessionEvent::new(
            "user_input",
            session.id().to_string(),
            temp.path().to_path_buf(),
            json!({"text":"valid"}),
        );
        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
        let malformed = "{".repeat(128);
        fs::write(
            session.path(),
            format!("{}\n{malformed}\n", serde_json::to_string(&valid).unwrap()),
        )
        .unwrap();
        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());

        let error = session
            .read_recent_events_tolerant(10, 64)
            .unwrap_err()
            .to_string();

        assert!(
            error.contains("tolerant recent read limit exceeded"),
            "{error}"
        );
        assert!(error.contains("malformed bytes"), "{error}");
    }
}