use super::replay::{
ConversationReplay, REPLAY_JSONL_MAX_BYTES, build_conversation_replay,
build_conversation_replay_with_read_stats, map_replay_read_error,
};
use crate::sessions::Session;
use std::{
fs,
path::{Path, PathBuf},
time::{SystemTime, UNIX_EPOCH},
};
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct ReplayCacheMetrics {
pub full_scans: u64,
pub bytes_read: u64,
pub lines_parsed: u64,
pub events_parsed: u64,
pub replay_builds: u64,
pub events_before_cutoff: u64,
pub events_after_cutoff: u64,
pub cache_hits: u64,
pub invalidations: u64,
}
#[derive(Debug, Default)]
pub(crate) struct ConversationReplayCache {
entry: Option<ConversationReplayCacheEntry>,
metrics: ReplayCacheMetrics,
}
#[derive(Debug)]
struct ConversationReplayCacheEntry {
session_id: String,
path: PathBuf,
generation: u64,
stamp: SessionFileStamp,
content_digest: [u8; 32],
replay: ConversationReplay,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SessionFileStamp {
Missing,
Present {
len: u64,
modified_ns: i128,
#[cfg(unix)]
device: u64,
#[cfg(unix)]
inode: u64,
#[cfg(unix)]
changed_seconds: i64,
#[cfg(unix)]
changed_nanoseconds: i64,
},
}
impl ConversationReplayCache {
pub(crate) fn replay(
&mut self,
session: Option<&Session>,
) -> anyhow::Result<ConversationReplay> {
let Some(session) = session else {
return build_conversation_replay(None);
};
let generation = session.replay_generation();
let stamp = session_file_stamp(session.path())?;
let same_cached_session = self.entry.as_ref().is_some_and(|entry| {
entry.session_id == session.id()
&& entry.path == session.path()
&& entry.generation == generation
});
if same_cached_session {
let fast_hit = self
.entry
.as_ref()
.is_some_and(|entry| stamp_supports_fast_reuse(stamp) && entry.stamp == stamp);
if fast_hit {
self.metrics.cache_hits = self.metrics.cache_hits.saturating_add(1);
return Ok(self
.entry
.as_ref()
.expect("cached session checked")
.replay
.clone());
}
let (digest, bytes_read) = session
.content_fingerprint_bounded(REPLAY_JSONL_MAX_BYTES)
.map_err(map_replay_read_error)?;
self.metrics.bytes_read = self.metrics.bytes_read.saturating_add(bytes_read as u64);
let entry = self.entry.as_mut().expect("cached session checked");
if entry.content_digest == digest {
entry.stamp = stamp;
self.metrics.cache_hits = self.metrics.cache_hits.saturating_add(1);
return Ok(entry.replay.clone());
}
}
if self.entry.is_some() {
self.metrics.invalidations = self.metrics.invalidations.saturating_add(1);
}
let (entry, load) = load_stable_replay_snapshot(session)?;
self.metrics.full_scans = self.metrics.full_scans.saturating_add(1);
self.metrics.bytes_read = self
.metrics
.bytes_read
.saturating_add(load.bytes_read as u64);
self.metrics.lines_parsed = self
.metrics
.lines_parsed
.saturating_add(load.lines_read as u64);
self.metrics.events_parsed = self
.metrics
.events_parsed
.saturating_add(load.events_parsed as u64);
self.metrics.replay_builds = self.metrics.replay_builds.saturating_add(1);
self.metrics.events_before_cutoff = self
.metrics
.events_before_cutoff
.saturating_add(load.events_before_cutoff as u64);
self.metrics.events_after_cutoff = self
.metrics
.events_after_cutoff
.saturating_add(load.events_after_cutoff as u64);
let replay = entry.replay.clone();
self.entry = Some(entry);
Ok(replay)
}
#[cfg(test)]
pub(crate) fn metrics(&self) -> ReplayCacheMetrics {
self.metrics
}
}
#[derive(Debug, Clone, Copy)]
struct ReplayLoadStats {
bytes_read: usize,
lines_read: usize,
events_parsed: usize,
events_before_cutoff: usize,
events_after_cutoff: usize,
}
fn load_stable_replay_snapshot(
session: &Session,
) -> anyhow::Result<(ConversationReplayCacheEntry, ReplayLoadStats)> {
const MAX_STABILITY_ATTEMPTS: usize = 3;
for _ in 0..MAX_STABILITY_ATTEMPTS {
let generation_before = session.replay_generation();
let stamp_before = session_file_stamp(session.path())?;
let load = build_conversation_replay_with_read_stats(session)?;
let stamp_after = session_file_stamp(session.path())?;
let generation_after = session.replay_generation();
if generation_before == generation_after && stamp_before == stamp_after {
return Ok((
ConversationReplayCacheEntry {
session_id: session.id().to_string(),
path: session.path().to_path_buf(),
generation: generation_after,
stamp: stamp_after,
content_digest: load.read.content_digest,
replay: load.replay,
},
ReplayLoadStats {
bytes_read: load.read.bytes_read,
lines_read: load.read.lines_read,
events_parsed: load.events_parsed,
events_before_cutoff: load.events_before_cutoff,
events_after_cutoff: load.events_after_cutoff,
},
));
}
}
anyhow::bail!("session changed repeatedly while preparing replay; retry the prompt")
}
fn session_file_stamp(path: &Path) -> anyhow::Result<SessionFileStamp> {
let metadata = match fs::symlink_metadata(path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Ok(SessionFileStamp::Missing);
}
Err(error) => return Err(error.into()),
};
if !metadata.file_type().is_file() {
anyhow::bail!("session JSONL is not a regular file")
}
let modified_ns = system_time_ns(metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH));
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
Ok(SessionFileStamp::Present {
len: metadata.len(),
modified_ns,
device: metadata.dev(),
inode: metadata.ino(),
changed_seconds: metadata.ctime(),
changed_nanoseconds: metadata.ctime_nsec(),
})
}
#[cfg(not(unix))]
{
Ok(SessionFileStamp::Present {
len: metadata.len(),
modified_ns,
})
}
}
fn system_time_ns(time: SystemTime) -> i128 {
match time.duration_since(UNIX_EPOCH) {
Ok(duration) => i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX),
Err(error) => -i128::try_from(error.duration().as_nanos()).unwrap_or(i128::MAX),
}
}
fn stamp_supports_fast_reuse(stamp: SessionFileStamp) -> bool {
#[cfg(unix)]
{
match stamp {
SessionFileStamp::Missing => true,
SessionFileStamp::Present {
modified_ns,
changed_nanoseconds,
..
} => {
modified_ns.rem_euclid(1_000_000_000) != 0 || changed_nanoseconds != 0
}
}
}
#[cfg(not(unix))]
{
let _ = stamp;
false
}
}