magi_code/sessions/
read.rs

1use super::event::SessionEvent;
2use super::manager::Session;
3use std::{
4    collections::VecDeque,
5    fs,
6    io::{BufRead, BufReader, Read, Seek, SeekFrom},
7    path::PathBuf,
8    time::SystemTime,
9};
10
11pub fn validate_session_id(id: String) -> anyhow::Result<String> {
12    if id.is_empty() {
13        anyhow::bail!("session id must not be empty");
14    }
15    if id == "." || id == ".." || id.contains("..") {
16        anyhow::bail!("session id must not contain '..'");
17    }
18    if id.contains('/') || id.contains('\\') {
19        anyhow::bail!("session id must not contain path separators");
20    }
21    if PathBuf::from(&id).is_absolute() {
22        anyhow::bail!("session id must not be an absolute path");
23    }
24    if !id
25        .chars()
26        .all(|character| character.is_ascii_alphanumeric() || character == '_' || character == '-')
27    {
28        anyhow::bail!("session id must match [A-Za-z0-9_-]+");
29    }
30    Ok(id)
31}
32
33fn open_session_file(session: &Session) -> anyhow::Result<fs::File> {
34    let root = session
35        .path
36        .parent()
37        .ok_or_else(|| anyhow::anyhow!("session file has no parent"))?;
38    super::store::open_existing_primary(root, &session.id)?
39        .ok_or_else(|| anyhow::anyhow!("session JSONL is missing"))
40}
41#[cfg(test)]
42pub(crate) fn latest_valid_event_timestamp_streaming(session: &Session) -> Option<SystemTime> {
43    let root = session.path.parent()?;
44    let file = super::store::open_existing_primary(root, &session.id).ok()??;
45    let mut latest = None;
46    for line in BufReader::new(file).lines() {
47        let Ok(line) = line else {
48            continue;
49        };
50        if line.trim().is_empty() {
51            continue;
52        }
53        let Ok(event) = serde_json::from_str::<SessionEvent>(&line) else {
54            continue;
55        };
56        let timestamp = SystemTime::from(event.timestamp);
57        latest = Some(latest.map_or(timestamp, |current: SystemTime| current.max(timestamp)));
58    }
59    latest
60}
61
62#[derive(Debug)]
63pub(crate) enum BoundedReadError {
64    BudgetExceeded(String),
65}
66
67impl std::fmt::Display for BoundedReadError {
68    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
69        match self {
70            Self::BudgetExceeded(message) => formatter.write_str(message),
71        }
72    }
73}
74
75impl std::error::Error for BoundedReadError {}
76
77fn budget_error(message: impl Into<String>) -> anyhow::Error {
78    anyhow::Error::new(BoundedReadError::BudgetExceeded(message.into()))
79}
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct SessionReadDiagnostic {
82    pub line: usize,
83    pub message: String,
84}
85
86#[derive(Debug, Clone, PartialEq)]
87pub struct TolerantSessionEvents {
88    pub events: Vec<SessionEvent>,
89    pub diagnostics: Vec<SessionReadDiagnostic>,
90    pub(crate) cutoff_bytes: u64,
91}
92
93const MAX_TOLERANT_READ_DIAGNOSTICS: usize = 64;
94
95fn push_tolerant_read_diagnostic(
96    diagnostics: &mut Vec<SessionReadDiagnostic>,
97    omitted_count: &mut usize,
98    diagnostic: SessionReadDiagnostic,
99) {
100    if diagnostics.len() < MAX_TOLERANT_READ_DIAGNOSTICS {
101        diagnostics.push(diagnostic);
102    } else {
103        *omitted_count = omitted_count.saturating_add(1);
104    }
105}
106
107fn finalize_tolerant_read_diagnostics(
108    mut diagnostics: Vec<SessionReadDiagnostic>,
109    omitted_count: usize,
110) -> Vec<SessionReadDiagnostic> {
111    if omitted_count == 0 {
112        return diagnostics;
113    }
114    if diagnostics.len() == MAX_TOLERANT_READ_DIAGNOSTICS {
115        diagnostics.pop();
116    }
117    diagnostics.push(SessionReadDiagnostic {
118        line: 0,
119        message: format!(
120            "omitted {omitted_count} additional session JSONL diagnostics after cap of {MAX_TOLERANT_READ_DIAGNOSTICS}"
121        ),
122    });
123    diagnostics
124}
125
126pub(crate) const MAX_METADATA_VISIT_LINES: usize = 100_000;
127pub(crate) const MAX_METADATA_VISIT_BYTES: usize = 64 * 1024 * 1024;
128
129impl Session {
130    pub fn read_events(&self) -> anyhow::Result<Vec<SessionEvent>> {
131        validate_session_id(self.id.clone())?;
132        if !self.path.exists() {
133            return Ok(Vec::new());
134        }
135        self.read_event_lines_streaming()?
136            .map(|event_line| event_line.map(|(event, _)| event))
137            .collect()
138    }
139
140    pub fn read_recent_events(
141        &self,
142        max_events: usize,
143        max_bytes: usize,
144    ) -> anyhow::Result<Vec<SessionEvent>> {
145        validate_session_id(self.id.clone())?;
146        if !self.path.exists() || max_events == 0 || max_bytes == 0 {
147            return Ok(Vec::new());
148        }
149        let mut retained = VecDeque::new();
150        let mut retained_bytes = 0usize;
151        for event_line in self.read_event_lines_streaming()? {
152            let (event, line_bytes) = event_line?;
153            // Intentional: use original JSONL line bytes, not reserialized event size; preserve this hardening.
154            retained_bytes = retained_bytes.saturating_add(line_bytes);
155            retained.push_back((event, line_bytes));
156            while retained.len() > max_events || retained_bytes > max_bytes {
157                if let Some((_, bytes)) = retained.pop_front() {
158                    retained_bytes = retained_bytes.saturating_sub(bytes);
159                } else {
160                    break;
161                }
162            }
163        }
164        Ok(retained.into_iter().map(|(event, _)| event).collect())
165    }
166
167    pub fn read_events_tolerant(&self) -> anyhow::Result<TolerantSessionEvents> {
168        self.read_events_tolerant_bounded(usize::MAX, usize::MAX)
169    }
170
171    pub(crate) fn read_events_tolerant_bounded(
172        &self,
173        max_lines: usize,
174        max_bytes: usize,
175    ) -> anyhow::Result<TolerantSessionEvents> {
176        let mut events = Vec::new();
177        let (diagnostics, cutoff_bytes) =
178            self.visit_events_tolerant_bounded(max_lines, max_bytes, |event| events.push(event))?;
179        Ok(TolerantSessionEvents {
180            events,
181            diagnostics,
182            cutoff_bytes: cutoff_bytes as u64,
183        })
184    }
185
186    pub(crate) fn visit_events_tolerant_bounded(
187        &self,
188        max_lines: usize,
189        max_bytes: usize,
190        mut visit: impl FnMut(SessionEvent),
191    ) -> anyhow::Result<(Vec<SessionReadDiagnostic>, usize)> {
192        validate_session_id(self.id.clone())?;
193        if !self.path.exists() {
194            return Ok((Vec::new(), 0));
195        }
196        let file = open_session_file(self)?;
197        let mut reader = BufReader::new(file);
198        let mut diagnostics = Vec::new();
199        let mut omitted_diagnostics = 0usize;
200        let mut total_bytes = 0usize;
201        let mut line = Vec::new();
202        for line_number in 1..=max_lines {
203            line.clear();
204            let remaining = max_bytes.saturating_sub(total_bytes);
205            if remaining == 0 {
206                if !reader.fill_buf()?.is_empty() {
207                    return Err(budget_error(format!(
208                        "session JSONL tolerant read limit exceeded: {max_bytes} bytes"
209                    )));
210                }
211                break;
212            }
213            let read = (&mut reader)
214                .take(
215                    u64::try_from(remaining)
216                        .unwrap_or(u64::MAX)
217                        .saturating_add(1),
218                )
219                .read_until(b'\n', &mut line)?;
220            if read == 0 {
221                break;
222            }
223            if read > remaining {
224                return Err(budget_error(format!(
225                    "session JSONL tolerant read limit exceeded at line {line_number}: {max_bytes} bytes"
226                )));
227            }
228            total_bytes = total_bytes.saturating_add(read);
229            if line.last() == Some(&b'\n') {
230                line.pop();
231                if line.last() == Some(&b'\r') {
232                    line.pop();
233                }
234            }
235            let text = String::from_utf8_lossy(&line);
236            match serde_json::from_str::<SessionEvent>(&text) {
237                Ok(event) => visit(event),
238                Err(_) => push_tolerant_read_diagnostic(
239                    &mut diagnostics,
240                    &mut omitted_diagnostics,
241                    SessionReadDiagnostic {
242                        line: line_number,
243                        message: format!(
244                            "operation=replay category=session_jsonl failed to parse session JSONL at line {line_number}"
245                        ),
246                    },
247                ),
248            }
249        }
250        if !reader.fill_buf()?.is_empty() {
251            return Err(budget_error(format!(
252                "session JSONL tolerant read limit exceeded: more than {max_lines} lines or {max_bytes} bytes"
253            )));
254        }
255        Ok((
256            finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics),
257            total_bytes,
258        ))
259    }
260
261    pub fn read_recent_events_tolerant(
262        &self,
263        max_events: usize,
264        max_bytes: usize,
265    ) -> anyhow::Result<TolerantSessionEvents> {
266        validate_session_id(self.id.clone())?;
267        if !self.path.exists() || max_events == 0 || max_bytes == 0 {
268            return Ok(TolerantSessionEvents {
269                events: Vec::new(),
270                diagnostics: Vec::new(),
271                cutoff_bytes: 0,
272            });
273        }
274        let file = open_session_file(self)?;
275        let lines = BufReader::new(file)
276            .lines()
277            .enumerate()
278            .map(|(index, line)| (index + 1, line));
279        self.collect_recent_events_tolerant_lines(lines, max_events, max_bytes)
280    }
281
282    pub(crate) fn read_recent_events_tolerant_tail(
283        &self,
284        max_events: usize,
285        max_retained_bytes: usize,
286        max_read_bytes: usize,
287    ) -> anyhow::Result<TolerantSessionEvents> {
288        validate_session_id(self.id.clone())?;
289        if max_events == 0 || max_retained_bytes == 0 || max_read_bytes == 0 {
290            anyhow::bail!(
291                "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}"
292            );
293        }
294        if !self.path.exists() {
295            return Ok(TolerantSessionEvents {
296                events: Vec::new(),
297                diagnostics: Vec::new(),
298                cutoff_bytes: 0,
299            });
300        }
301        let metadata = open_session_file(self)?.metadata()?;
302        let file_len = metadata.len();
303        let max_read_bytes_u64 = u64::try_from(max_read_bytes).unwrap_or(u64::MAX);
304        if file_len <= max_read_bytes_u64 {
305            return self.read_recent_events_tolerant(max_events, max_retained_bytes);
306        }
307        let start = file_len - max_read_bytes_u64;
308        let mut file = open_session_file(self)?;
309        file.seek(SeekFrom::Start(start))?;
310        let mut tail = Vec::with_capacity(max_read_bytes);
311        file.take(max_read_bytes_u64).read_to_end(&mut tail)?;
312
313        let first_complete_line = tail
314            .iter()
315            .position(|byte| *byte == b'\n')
316            .map_or(tail.len(), |index| index + 1);
317        let tail = String::from_utf8_lossy(&tail[first_complete_line..]);
318        let lines = tail
319            .lines()
320            .enumerate()
321            .map(|(index, line)| (index + 1, Ok(line.to_string())));
322        let mut result =
323            self.collect_recent_events_tolerant_lines(lines, max_events, max_retained_bytes)?;
324        result.diagnostics.insert(
325            0,
326            SessionReadDiagnostic {
327                line: 0,
328                message: format!(
329                    "operation=recent_context category=session_jsonl bounded tail window; omitted older lines; read final {max_read_bytes} of {file_len} bytes"
330                ),
331            },
332        );
333        Ok(result)
334    }
335
336    fn collect_recent_events_tolerant_lines(
337        &self,
338        lines: impl IntoIterator<Item = (usize, Result<String, std::io::Error>)>,
339        max_events: usize,
340        max_bytes: usize,
341    ) -> anyhow::Result<TolerantSessionEvents> {
342        let mut retained = VecDeque::new();
343        let mut retained_bytes = 0usize;
344        let mut malformed_bytes = 0usize;
345        let mut diagnostics = Vec::new();
346        let mut omitted_diagnostics = 0usize;
347        for (line_number, line) in lines {
348            match line {
349                Ok(line) => {
350                    let line_bytes = line.len() + 1;
351                    match serde_json::from_str::<SessionEvent>(&line) {
352                        Ok(event) => {
353                            retained_bytes = retained_bytes.saturating_add(line_bytes);
354                            retained.push_back((event, line_bytes));
355                            while retained.len() > max_events || retained_bytes > max_bytes {
356                                if let Some((_, bytes)) = retained.pop_front() {
357                                    retained_bytes = retained_bytes.saturating_sub(bytes);
358                                } else {
359                                    break;
360                                }
361                            }
362                        }
363                        Err(_) => {
364                            malformed_bytes = malformed_bytes.saturating_add(line_bytes);
365                            if malformed_bytes > max_bytes {
366                                anyhow::bail!(
367                                    "operation=recent_context category=session_jsonl tolerant recent read limit exceeded: malformed bytes > {max_bytes}"
368                                );
369                            }
370                            push_tolerant_read_diagnostic(
371                                &mut diagnostics,
372                                &mut omitted_diagnostics,
373                                SessionReadDiagnostic {
374                                    line: line_number,
375                                    message: format!(
376                                        "operation=recent_context category=session_jsonl failed to parse session JSONL at line {line_number}"
377                                    ),
378                                },
379                            );
380                        }
381                    }
382                }
383                Err(_) => push_tolerant_read_diagnostic(
384                    &mut diagnostics,
385                    &mut omitted_diagnostics,
386                    SessionReadDiagnostic {
387                        line: line_number,
388                        message: format!(
389                            "operation=recent_context category=session_jsonl read failure at line {line_number}"
390                        ),
391                    },
392                ),
393            }
394        }
395        Ok(TolerantSessionEvents {
396            events: retained.into_iter().map(|(event, _)| event).collect(),
397            diagnostics: finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics),
398            cutoff_bytes: 0,
399        })
400    }
401
402    pub fn latest_event_timestamp_bounded(&self) -> anyhow::Result<Option<SystemTime>> {
403        validate_session_id(self.id.clone())?;
404        if !self.path.exists() {
405            return Ok(None);
406        }
407        let mut latest = None;
408        for event_line in self.read_event_lines_streaming()? {
409            let timestamp = SystemTime::from(event_line?.0.timestamp);
410            latest = Some(latest.map_or(timestamp, |current: SystemTime| current.max(timestamp)));
411        }
412        Ok(latest)
413    }
414
415    fn read_event_lines_streaming(
416        &self,
417    ) -> anyhow::Result<impl Iterator<Item = anyhow::Result<(SessionEvent, usize)>> + '_> {
418        let file = open_session_file(self)?;
419        Ok(BufReader::new(file)
420            .lines()
421            .enumerate()
422            .map(|(index, line)| {
423                let line = line.map_err(|error| {
424                    anyhow::anyhow!(
425                        "failed to read session JSONL at {} line {}: {error}",
426                        self.path.display(),
427                        index + 1
428                    )
429                })?;
430                // Intentional accounting: retain the original line size to avoid per-event reserialization.
431                let line_bytes = line.len() + 1;
432                let event = serde_json::from_str::<SessionEvent>(&line).map_err(|error| {
433                    anyhow::anyhow!(
434                        "failed to parse session JSONL at {} line {}: {error}",
435                        self.path.display(),
436                        index + 1
437                    )
438                })?;
439                Ok((event, line_bytes))
440            }))
441    }
442}
443
444#[cfg(test)]
445mod tests {
446    use super::super::manager::SessionManager;
447    use super::*;
448    use proptest::prelude::*;
449    use serde_json::json;
450    use tempfile::TempDir;
451
452    fn valid_session_id_strategy() -> impl Strategy<Value = String> {
453        proptest::string::string_regex("[A-Za-z0-9_-]{1,64}").unwrap()
454    }
455
456    fn invalid_session_id_strategy() -> impl Strategy<Value = String> {
457        prop_oneof![
458            Just(String::new()),
459            Just(".".to_string()),
460            any::<String>().prop_map(|value| format!("{value}..")),
461            any::<String>().prop_map(|value| format!("{value}/{value}")),
462            any::<String>().prop_map(|value| format!("{value}\\{value}")),
463            any::<String>().prop_map(|value| format!("{value}.jsonl")),
464            any::<String>().prop_map(|value| format!("{value}é")),
465        ]
466    }
467
468    proptest! {
469        #[test]
470        fn validate_session_id_accepts_only_non_empty_safe_ascii_ids(id in valid_session_id_strategy()) {
471            let validated = validate_session_id(id.clone()).unwrap();
472
473            prop_assert_eq!(&validated, &id);
474            prop_assert!(!validated.is_empty());
475            prop_assert!(validated
476                .chars()
477                .all(|character| character.is_ascii_alphanumeric() || character == '_' || character == '-'));
478        }
479
480        #[test]
481        fn validate_session_id_rejects_generated_unsafe_ids(id in invalid_session_id_strategy()) {
482            prop_assert!(validate_session_id(id).is_err());
483        }
484    }
485
486    #[test]
487    fn recent_and_latest_session_reads_are_bounded_streaming_paths() {
488        let temp = TempDir::new().unwrap();
489        let manager = SessionManager::new(temp.path().join("sessions"));
490        let session = manager.create().unwrap();
491        for index in 0..25 {
492            session
493                .append(&SessionEvent::new(
494                    "event",
495                    session.id().to_string(),
496                    temp.path().to_path_buf(),
497                    json!({"index": index}),
498                ))
499                .unwrap();
500        }
501        let recent = session.read_recent_events(3, 4096).unwrap();
502        assert_eq!(recent.len(), 3);
503        assert_eq!(recent[0].payload["index"], 22);
504        assert!(session.latest_event_timestamp_bounded().unwrap().is_some());
505        assert_eq!(manager.most_recent().unwrap().unwrap().id(), session.id());
506    }
507
508    #[test]
509    fn read_recent_events_uses_original_jsonl_line_bytes() {
510        let temp = TempDir::new().unwrap();
511        let manager = SessionManager::new(temp.path().join("sessions"));
512        let session = manager.create().unwrap();
513        let padded = SessionEvent::new(
514            "event",
515            session.id().to_string(),
516            temp.path().to_path_buf(),
517            json!({"index": 1}),
518        );
519        let recent = SessionEvent::new(
520            "event",
521            session.id().to_string(),
522            temp.path().to_path_buf(),
523            json!({"index": 2}),
524        );
525        let padded_compact_len = serde_json::to_vec(&padded).unwrap().len();
526        let mut padded_line = serde_json::to_value(&padded).unwrap();
527        padded_line["padding"] = json!("x".repeat(2048));
528        let padded_line = serde_json::to_string(&padded_line).unwrap();
529        let recent_line = serde_json::to_string(&recent).unwrap();
530        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
531        fs::write(session.path(), format!("{padded_line}\n{recent_line}\n")).unwrap();
532        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
533
534        let max_bytes = padded_compact_len + recent_line.len() + 2;
535        let events = session.read_recent_events(10, max_bytes).unwrap();
536
537        assert_eq!(events.len(), 1);
538        assert_eq!(events[0].payload["index"], 2);
539    }
540
541    #[test]
542    fn tolerant_read_reports_malformed_lines_without_payload_leak() {
543        let temp = TempDir::new().unwrap();
544        let manager = SessionManager::new(temp.path().join("sessions"));
545        let session = manager.open("safe").unwrap();
546        let first = SessionEvent::new(
547            "user_input",
548            session.id().to_string(),
549            temp.path().to_path_buf(),
550            json!({"text":"before"}),
551        );
552        let second = SessionEvent::new(
553            "assistant_output",
554            session.id().to_string(),
555            temp.path().to_path_buf(),
556            json!({"text":"after"}),
557        );
558        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
559        fs::write(
560            session.path(),
561            format!(
562                "{}\n{{\"access_token\":\"secret-token\",\n{}\n",
563                serde_json::to_string(&first).unwrap(),
564                serde_json::to_string(&second).unwrap()
565            ),
566        )
567        .unwrap();
568        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
569
570        assert!(session.read_events().is_err());
571        let tolerant = session.read_events_tolerant().unwrap();
572
573        assert_eq!(tolerant.events.len(), 2);
574        assert_eq!(tolerant.events[0].payload["text"], "before");
575        assert_eq!(tolerant.events[1].payload["text"], "after");
576        assert_eq!(tolerant.diagnostics.len(), 1);
577        assert_eq!(tolerant.diagnostics[0].line, 2);
578        assert!(tolerant.diagnostics[0].message.contains("line 2"));
579        assert!(!tolerant.diagnostics[0].message.contains("secret-token"));
580    }
581
582    #[test]
583    fn tolerant_session_read_caps_malformed_diagnostics() {
584        let temp = TempDir::new().unwrap();
585        let manager = SessionManager::new(temp.path().join("sessions"));
586        let session = manager.open("safe").unwrap();
587        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
588        let mut lines = Vec::new();
589        for index in 0..(MAX_TOLERANT_READ_DIAGNOSTICS + 10) {
590            lines.push(format!("not json {index}"));
591        }
592        fs::write(session.path(), lines.join("\n")).unwrap();
593        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
594
595        let tolerant = session.read_events_tolerant().unwrap();
596
597        assert!(tolerant.events.is_empty());
598        assert_eq!(tolerant.diagnostics.len(), MAX_TOLERANT_READ_DIAGNOSTICS);
599        let summary = tolerant.diagnostics.last().unwrap();
600        assert_eq!(summary.line, 0);
601        assert!(summary.message.contains("omitted 10 additional"));
602        assert!(summary.message.contains("cap of 64"));
603        assert!(
604            !tolerant
605                .diagnostics
606                .iter()
607                .any(|diagnostic| diagnostic.message.contains("not json"))
608        );
609    }
610
611    #[test]
612    fn read_recent_events_tolerant_preserves_valid_events_after_malformed_line() {
613        let temp = TempDir::new().unwrap();
614        let manager = SessionManager::new(temp.path().join("sessions"));
615        let session = manager.open("safe").unwrap();
616        let old = SessionEvent::new(
617            "user_input",
618            session.id().to_string(),
619            temp.path().to_path_buf(),
620            json!({"text":"old"}),
621        );
622        let recent = SessionEvent::new(
623            "assistant_output",
624            session.id().to_string(),
625            temp.path().to_path_buf(),
626            json!({"text":"recent"}),
627        );
628        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
629        fs::write(
630            session.path(),
631            format!(
632                "{}\nnot json\n{}\n",
633                serde_json::to_string(&old).unwrap(),
634                serde_json::to_string(&recent).unwrap()
635            ),
636        )
637        .unwrap();
638        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
639
640        let tolerant = session.read_recent_events_tolerant(10, 4096).unwrap();
641
642        assert_eq!(tolerant.events.len(), 2);
643        assert_eq!(tolerant.events[1].payload["text"], "recent");
644        assert_eq!(tolerant.diagnostics[0].line, 2);
645        assert!(!tolerant.diagnostics[0].message.contains("not json"));
646    }
647
648    #[test]
649    fn read_recent_events_tolerant_bounds_malformed_input_bytes() {
650        let temp = TempDir::new().unwrap();
651        let manager = SessionManager::new(temp.path().join("sessions"));
652        let session = manager.open("safe").unwrap();
653        let valid = SessionEvent::new(
654            "user_input",
655            session.id().to_string(),
656            temp.path().to_path_buf(),
657            json!({"text":"valid"}),
658        );
659        fs::create_dir_all(session.path().parent().unwrap()).unwrap();
660        let malformed = "{".repeat(128);
661        fs::write(
662            session.path(),
663            format!("{}\n{malformed}\n", serde_json::to_string(&valid).unwrap()),
664        )
665        .unwrap();
666        crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
667
668        let error = session
669            .read_recent_events_tolerant(10, 64)
670            .unwrap_err()
671            .to_string();
672
673        assert!(
674            error.contains("tolerant recent read limit exceeded"),
675            "{error}"
676        );
677        assert!(error.contains("malformed bytes"), "{error}");
678    }
679}
magi_code/sessions/read.rs

magi_code/sessions/
read.rs
