use super::event::SessionEvent;
use super::manager::Session;
use std::{
collections::VecDeque,
fs,
io::{BufRead, BufReader, Read, Seek, SeekFrom},
path::PathBuf,
time::SystemTime,
};
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 struct SessionReadDiagnostic {
pub line: usize,
pub message: String,
}
#[derive(Debug, Clone, PartialEq)]
pub struct TolerantSessionEvents {
pub events: Vec<SessionEvent>,
pub diagnostics: Vec<SessionReadDiagnostic>,
pub(crate) cutoff_bytes: u64,
}
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 {
pub 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()
}
pub 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?;
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())
}
pub 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> {
let mut events = Vec::new();
let (diagnostics, cutoff_bytes) =
self.visit_events_tolerant_bounded(max_lines, max_bytes, |event| events.push(event))?;
Ok(TolerantSessionEvents {
events,
diagnostics,
cutoff_bytes: cutoff_bytes as u64,
})
}
pub(crate) fn visit_events_tolerant_bounded(
&self,
max_lines: usize,
max_bytes: usize,
mut visit: impl FnMut(SessionEvent),
) -> anyhow::Result<(Vec<SessionReadDiagnostic>, usize)> {
validate_session_id(self.id.clone())?;
if !self.path.exists() {
return Ok((Vec::new(), 0));
}
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 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);
if line.last() == Some(&b'\n') {
line.pop();
if line.last() == Some(&b'\r') {
line.pop();
}
}
let text = String::from_utf8_lossy(&line);
match serde_json::from_str::<SessionEvent>(&text) {
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"
)));
}
Ok((
finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics),
total_bytes,
))
}
pub 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)
.lines()
.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,
});
}
let metadata = open_session_file(self)?.metadata()?;
let file_len = metadata.len();
let max_read_bytes_u64 = u64::try_from(max_read_bytes).unwrap_or(u64::MAX);
if file_len <= max_read_bytes_u64 {
return self.read_recent_events_tolerant(max_events, max_retained_bytes);
}
let start = file_len - max_read_bytes_u64;
let mut file = open_session_file(self)?;
file.seek(SeekFrom::Start(start))?;
let mut tail = Vec::with_capacity(max_read_bytes);
file.take(max_read_bytes_u64).read_to_end(&mut tail)?;
let first_complete_line = tail
.iter()
.position(|byte| *byte == b'\n')
.map_or(tail.len(), |index| index + 1);
let tail = String::from_utf8_lossy(&tail[first_complete_line..]);
let lines = tail
.lines()
.enumerate()
.map(|(index, line)| (index + 1, Ok(line.to_string())));
let mut result =
self.collect_recent_events_tolerant_lines(lines, max_events, max_retained_bytes)?;
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<String, 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_str::<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,
})
}
pub 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)
}
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
)
})?;
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}é")),
]
}
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_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}");
}
}