diff --git a/client-core/src/lib.rs b/client-core/src/lib.rs index e3316ae..db0bc34 100644 --- a/client-core/src/lib.rs +++ b/client-core/src/lib.rs @@ -4,5 +4,6 @@ pub mod ansi; pub mod highlight; +pub mod transcript_cache; pub use event_model::*; diff --git a/client-core/src/transcript_cache.rs b/client-core/src/transcript_cache.rs new file mode 100644 index 0000000..9196d40 --- /dev/null +++ b/client-core/src/transcript_cache.rs @@ -0,0 +1,1391 @@ +//! This phone's copy of the transcripts it has already been sent, so +//! reopening a session does not download it again. Ported from +//! `app/.../TranscriptCache.kt`; see `TRANSCRIPT_CACHE.md` at the repo root +//! for the design and `CLIENT_CORE.md` for how this file corresponds to it. +//! +//! What is stored is the server's own JSON for one event per line, in +//! transcript order. Reading the cache means running the same [`seq_of`] +//! the network path runs, so a cached transcript and a fetched one cannot +//! draw differently, and an event type this build does not know keeps +//! every field it arrived with for the build that will. Rows are +//! deliberately *not* what is stored: a row is a rendering, and a cache of +//! rows would need throwing away on every update that touched the fold. +//! +//! Four rules run through all of it: +//! 1. what is on screen is what the server's transcript says, in order, +//! with nothing missing -- the cache is a copy and is never inferred, +//! folded or edited here; +//! 2. a cached line is never ahead of the live cursor, and the cursor never +//! ahead of the cache; +//! 3. the cache is never load-bearing -- missing, evicted, damaged or +//! unwritable all degrade to a cold open, never to a blank or a wrong +//! screen; +//! 4. a line already on the phone is not fetched again. +//! +//! No JSON parser here: what it needs off a line is the sequence number and +//! whether the line is a streamed delta, both read with a regex-free scan +//! (see [`seq_of`] and [`is_delta`]). A line it cannot read that way is +//! treated as damage. + +use std::collections::VecDeque; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +/// How much of this phone's cache directory all of one server's transcripts +/// may take. A dozen of the largest transcripts seen in the dev VM (21 MB +/// for 24,000 events) and a small fraction of a phone. A number to revisit +/// against real use rather than a measurement of anything. +pub const CACHE_BUDGET_BYTES: u64 = 256_000_000; + +/// What the newest cached line says, which is what the probe checks against +/// the server. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CachedTail { + pub seq: u64, + pub line: String, +} + +/// This phone's cache root for one server, holding one directory per session. +pub struct TranscriptCache { + root: PathBuf, + warn: std::sync::Arc, +} + +impl TranscriptCache { + pub fn new(root: impl Into) -> Self { + Self::with_warn(root, |msg| eprintln!("ai-app: {msg}")) + } + + pub fn with_warn( + root: impl Into, + warn: impl Fn(&str) + Send + Sync + 'static, + ) -> Self { + Self { + root: root.into(), + warn: std::sync::Arc::new(warn), + } + } + + /// The cache for one session, whether or not anything has been stored + /// for it yet. + pub fn session(&self, id: &str) -> SessionCache { + SessionCache::new(self.root.join(id), self.warn.clone()) + } + + /// Deletes every session directory not in `ids`, called after a + /// successful list fetch. The path out for a session deleted on + /// another device: nothing here would otherwise hear about it, and + /// unlike a draft's few bytes what it leaves behind is megabytes. + pub fn retain_only(&self, ids: &std::collections::HashSet) { + guard_io((), self.warn.as_ref(), || { + for dir in session_dirs(&self.root)? { + if let Some(name) = dir.file_name().and_then(|n| n.to_str()) + && !ids.contains(name) + { + fs::remove_dir_all(&dir)?; + } + } + Ok(()) + }); + } + + /// Deletes least-recently-touched session directories, never `keep`, + /// until the whole of this server's cache is under `budget`. + /// Least-recently-touched rather than largest: what a reader is likely + /// to open again is what they opened last, and evicting the big ones + /// first would empty the cache for exactly the conversations it exists + /// for. + pub fn evict_to_budget(&self, keep: &str, budget: u64) { + guard_io((), self.warn.as_ref(), || { + let mut dirs: Vec<(PathBuf, std::time::SystemTime)> = session_dirs(&self.root)? + .into_iter() + .map(|d| { + let modified = fs::metadata(&d) + .and_then(|m| m.modified()) + .unwrap_or(std::time::UNIX_EPOCH); + (d, modified) + }) + .collect(); + dirs.sort_by_key(|(_, modified)| *modified); + let mut total: u64 = dirs.iter().map(|(d, _)| dir_size(d)).sum(); + for (dir, _) in dirs { + if total <= budget { + break; + } + if dir.file_name().and_then(|n| n.to_str()) == Some(keep) { + continue; + } + let was = dir_size(&dir); + if fs::remove_dir_all(&dir).is_ok() { + total -= was; + } + } + Ok(()) + }); + } + + pub fn purge_all(&self) { + guard_io((), self.warn.as_ref(), || { + if self.root.is_dir() { + fs::remove_dir_all(&self.root)?; + } + Ok(()) + }); + } +} + +fn session_dirs(root: &Path) -> io::Result> { + if !root.is_dir() { + return Ok(Vec::new()); + } + Ok(fs::read_dir(root)? + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .filter(|p| p.is_dir()) + .collect()) +} + +fn dir_size(path: &Path) -> u64 { + let Ok(entries) = fs::read_dir(path) else { + return 0; + }; + entries + .filter_map(|e| e.ok()) + .map(|e| { + let p = e.path(); + if p.is_dir() { + dir_size(&p) + } else { + fs::metadata(&p).map(|m| m.len()).unwrap_or(0) + } + }) + .sum() +} + +/// One session's cached lines, as a directory of chunks. +/// +/// A chunk is a set of lines *and a claim about what they cover*, and the +/// two are not the same thing: a coalesced page joins each run of streamed +/// deltas into one event carrying the seq of the run's oldest delta, so a +/// page whose newest event is seq 1,200 may cover everything up to the +/// 1,650 it was fetched with, and nothing in the lines says so. So coverage +/// is the half-open range in the file's name: +/// `-.rows.jsonl` (a coalesced page; `end` is the `before` it +/// was fetched with) or `-.raw.jsonl` (an uncoalesced page, or a +/// closed live run); `-open.raw.jsonl` is the live run, whose end is +/// its last line's seq + 1. +/// +/// Two chunks are adjacent when one's `end` is the other's `first`. Only +/// the contiguous run ending at the newest chunk -- the **suffix** -- is +/// ever served: chunks behind a gap are kept, because the gap is usually +/// closed by paging back through it, but nothing is served across one. +/// +/// **The newest chunk is always raw**, which is what makes the stream +/// cursor and the probe well defined. +/// +/// Nothing here is load-bearing. Every operation that touches the disk +/// answers as though the cache were empty when it cannot, and a write +/// failure disables writing for the rest of this instance's life so that a +/// full disk costs one log line rather than one per delta. +/// +/// A `Mutex` around the writer state stands in for Kotlin's `@Synchronized`: +/// the stream appends live events from its own thread while a reader +/// scrolling back reads pages from another, and this is what keeps the open +/// chunk's name, its end and its writer from being read half-rotated. +pub struct SessionCache { + dir: PathBuf, + warn: std::sync::Arc, + state: Mutex, +} + +#[derive(Default)] +struct WriterState { + /// Set by the first write that fails: a second would fail the same way, + /// once per delta. + disabled: bool, + writer: Option, + open_file: Option, + open_end: u64, +} + +impl SessionCache { + fn new(dir: PathBuf, warn: std::sync::Arc) -> Self { + Self { + dir, + warn, + state: Mutex::new(WriterState::default()), + } + } + + /// The newest line of the suffix, or `None` when there is none or the + /// newest chunk is not raw. + pub fn tail(&self) -> Option { + self.guard(None, |this, state| { + let Some(newest) = this.suffix(state)?.last().cloned() else { + return Ok(None); + }; + let mut found = None; + this.each_line(state, &newest, |line| { + found = Some(CachedTail { + seq: seq_of(line).expect("chunk lines are checked in each_line"), + line: line.to_string(), + }); + false + }); + Ok(found) + }) + } + + /// The newest `limit` lines of the suffix, oldest first -- the opening window. + pub fn newest(&self, limit: usize) -> Vec { + self.guard(Vec::new(), |this, state| { + let mut taken: VecDeque = VecDeque::new(); + for chunk in this.suffix(state)?.iter().rev() { + if taken.len() >= limit { + break; + } + this.each_line(state, chunk, |line| { + taken.push_front(line.to_string()); + taken.len() < limit + }); + } + Ok(taken.into_iter().collect()) + }) + } + + /// The page of lines before `before`, oldest first, or `None` when the + /// cache cannot answer. + /// + /// `None` is a miss -- the suffix does not cover the ground immediately + /// below `before` -- and means the server has to be asked. Deliberately + /// not an empty list: an empty page is how the screen is told it has + /// reached the start of the conversation. + /// + /// With `rows` the count is rows rather than lines, mirroring the + /// server's `parse_coalesced`. The deltas are not joined here -- the + /// fold does that, and the joined row keeps the seq of its first delta + /// either way. + pub fn page(&self, before: u64, limit: usize, rows: bool) -> Option> { + self.guard(None, |this, state| { + let suffix = this.suffix(state)?; + let Some(newest) = suffix.last().cloned() else { + return Ok(None); + }; + // Above what is held, or at or below where it starts: either + // way the run the caller is scrolling into is not continuous + // with this one, and only the server has it. + if before > newest.end || before <= suffix.first().unwrap().first { + return Ok(None); + } + let mut taken: VecDeque = VecDeque::new(); + let mut counted = 0usize; + let mut in_run = false; + let mut wanting = true; + for chunk in suffix.iter().rev() { + if !wanting { + break; + } + if chunk.first >= before { + continue; + } + this.each_line(state, chunk, |line| { + // The page is what is *before* the cursor; the rows at + // or above it are already on screen. + let seq = seq_of(line).expect("chunk lines are checked in each_line"); + if seq >= before { + return true; + } + if rows { + let delta = is_delta(line); + // Stop only between rows: a delta continuing the + // run being gathered is part of a row already + // counted, and breaking on it would drop the half + // of that row already taken. + if counted >= limit && !(delta && in_run) { + wanting = false; + } else { + if !delta || !in_run { + counted += 1; + } + in_run = delta; + } + } else if taken.len() >= limit { + wanting = false; + } + if wanting { + taken.push_front(line.to_string()); + } + wanting + }); + } + Ok(Some(taken.into_iter().collect())) + }) + } + + /// The `end` of the nearest chunk at or below `before`, which is the + /// floor a fetched page is asked with so that it stops where this + /// phone's copy starts. `None` when there is no such chunk. + pub fn covered_up_to(&self, before: u64) -> Option { + self.guard(None, |this, state| { + Ok(this + .chunks(state)? + .iter() + .map(|c| c.end) + .filter(|&end| end <= before) + .max()) + }) + } + + /// Stores a fetched page covering `[first, end)`; `false` when it was + /// not stored. + /// + /// Refused when it overlaps a chunk already here, because there is no + /// clean cut: a coalesced event cannot be split at a seq inside its own + /// delta run. The caller keeps that from arising by bounding what it + /// fetches, and this is the guard for a page that arrives anyway. + /// + /// The newest chunk is never stored through here: the opening window + /// and every live frame go through [`Self::append`], which is what + /// keeps the newest chunk raw and open. + pub fn store_page(&self, lines: &[String], first: u64, end: u64, rows: bool) -> bool { + self.guard(false, |this, state| { + if state.disabled || lines.is_empty() || end <= first { + return Ok(false); + } + if this + .chunks(state)? + .iter() + .any(|c| first < c.end && c.first < end) + { + return Ok(false); + } + fs::create_dir_all(&this.dir)?; + let kind = if rows { "rows" } else { "raw" }; + let mut content = lines.join("\n"); + content.push('\n'); + fs::write( + this.dir.join(format!("{first}-{end}.{kind}.jsonl")), + content, + )?; + Ok(true) + }) + } + + /// Appends one live event, which is also how a freshly fetched opening + /// window is stored. + /// + /// A seq equal to the open chunk's end extends it. A larger one is a + /// gap -- which is what a `reset` looks like from here -- and closes + /// the open chunk under the end it turned out to have. A smaller one is + /// already covered and is ignored; the SSE contract is `seq > after`. + pub fn append(&self, line: &str, seq: u64) { + self.guard((), |this, state| { + if state.disabled { + return Ok(()); + } + let Some(writer) = this.writer_for(state, seq)? else { + return Ok(()); + }; + // Written as it arrived. A newline inside it would split one + // event into two unreadable halves, but neither source can + // produce one. + use std::io::Write; + writer.write_all(line.as_bytes())?; + writer.write_all(b"\n")?; + state.open_end = seq + 1; + Ok(()) + }); + } + + /// Flushes what [`Self::append`] has buffered. + pub fn flush(&self) { + self.guard((), |_this, state| { + if let Some(writer) = state.writer.as_mut() { + use std::io::Write; + writer.flush()?; + } + Ok(()) + }); + } + + /// What [`Self::purge`] would discard, for the reload row in session settings. + pub fn bytes(&self) -> u64 { + self.guard(0, |this, _state| Ok(dir_size(&this.dir))) + } + + /// Marks this session as visited, which is what eviction ranks by. + pub fn touch(&self) { + self.guard((), |this, _state| { + if this.dir.is_dir() { + let now = std::time::SystemTime::now(); + filetime_set_modified(&this.dir, now)?; + } + Ok(()) + }); + } + + pub fn purge(&self) { + self.guard((), |this, state| { + close_writer(state); + if this.dir.is_dir() { + fs::remove_dir_all(&this.dir)?; + } + Ok(()) + }); + } + + // -- chunks ------------------------------------------------------------------------------ + + /// Every chunk on disk, oldest first. A name this does not recognise is + /// not ours and is ignored. Recomputed per operation rather than kept: + /// another operation may have changed the directory. + fn chunks(&self, state: &mut WriterState) -> io::Result> { + if let Some(writer) = state.writer.as_mut() { + use std::io::Write; + let _ = writer.flush(); + } + if !self.dir.is_dir() { + return Ok(Vec::new()); + } + let mut out = Vec::new(); + for entry in fs::read_dir(&self.dir)? { + let entry = entry?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + let Some((first, end_str, kind)) = parse_chunk_name(name) else { + continue; + }; + let open = end_str == "open"; + let end = if open { + self.open_end_of(state, &entry.path(), first) + } else { + end_str.parse::().ok() + }; + // A chunk covering nothing is one that was created and never + // written to -- an append whose very first write failed. It + // says nothing, so it is not a chunk. + if let Some(end) = end + && end > first + { + out.push(Chunk { + file: entry.path(), + first, + end, + open, + rows: kind == "rows", + }); + } + } + out.sort_by_key(|c| c.first); + Ok(out) + } + + /// The open chunk's end: its last line's seq plus one, or the in-memory + /// end while this instance is the one writing it. + /// + /// An open chunk whose last line cannot be read is this app having died + /// mid-write. That line is dropped and the file truncated to the last + /// good one, which is the one place damage is repaired rather than + /// discarded. + fn open_end_of(&self, state: &WriterState, file: &Path, first: u64) -> Option { + if state.open_file.as_deref() == Some(file) && state.open_end > 0 { + return Some(state.open_end); + } + let _ = repair_tail(file); + let mut end = first; + each_line_backwards(file, |_offset, line| { + if let Some(seq) = seq_of(line) { + end = seq + 1; + } + false + }); + Some(end) + } + + /// The contiguous run of adjacent chunks ending at the newest one, + /// oldest first. + /// + /// A newest chunk that is not raw cannot happen while this code is the + /// only writer, and means the directory is not to be trusted -- so the + /// session is discarded. + fn suffix(&self, state: &mut WriterState) -> io::Result> { + let all = self.chunks(state)?; + let Some(newest) = all.last() else { + return Ok(Vec::new()); + }; + if newest.rows { + return Err(damaged(&newest.file)); + } + let mut run: VecDeque = VecDeque::new(); + run.push_front(newest.clone()); + let mut index = all.len() - 1; + while index > 0 && all[index - 1].end == run.front().unwrap().first { + index -= 1; + run.push_front(all[index].clone()); + } + Ok(run.into_iter().collect()) + } + + /// Each line of `chunk`, newest first, until `take` says stop. + /// + /// Damage anywhere but at the tail of the open chunk was not written by + /// this code, and there is no honest way to say what a chunk covers + /// with a line of it unreadable -- so it is treated as damage rather + /// than served partially. Propagated as an `Err` from the caller's + /// `guard` closure via a panic-free sentinel: callers pass a `take` + /// that never sees a non-seq line reach it, because `each_line_backwards` + /// is stopped the moment one does and the damage is reported by the + /// `Result` this returns. + fn each_line( + &self, + _state: &mut WriterState, + chunk: &Chunk, + mut take: impl FnMut(&str) -> bool, + ) { + each_line_backwards(&chunk.file, |_offset, line| { + if seq_of(line).is_none() { + DAMAGED.with(|cell| *cell.borrow_mut() = Some(chunk.file.clone())); + return false; + } + take(line) + }); + } + + // -- writing ----------------------------------------------------------------------------- + + /// The writer for the chunk `seq` belongs in, opening or rotating one + /// as it has to. + fn writer_for<'s>( + &self, + state: &'s mut WriterState, + seq: u64, + ) -> io::Result> { + if state.writer.is_some() { + if seq == state.open_end { + return Ok(state.writer.as_mut()); + } + if seq < state.open_end { + return Ok(None); + } + // A gap: what this instance has written covers up to `open_end`, + // and that is the name the chunk gets before a new one starts + // at the arriving seq. + let end = state.open_end; + self.close_open_chunk(state, end); + } + fs::create_dir_all(&self.dir)?; + // An open chunk left by an earlier instance, or by an earlier screen. + let existing = self.chunks(state)?.into_iter().rfind(|c| c.open); + if let Some(existing) = existing { + if seq < existing.end { + return Ok(None); + } + if seq == existing.end { + state.open_file = Some(existing.file.clone()); + state.open_end = existing.end; + let file = fs::OpenOptions::new().append(true).open(&existing.file)?; + state.writer = Some(file); + return Ok(state.writer.as_mut()); + } + let first = existing.first; + let end = existing.end; + rename_chunk(&existing.file, &self.dir, first, end); + } + // A chunk that was created and never written to would otherwise be + // left behind under a name a second one is about to want; it + // covers nothing, so nothing is lost with it. + if let Ok(entries) = fs::read_dir(&self.dir) { + for entry in entries.flatten() { + let name = entry.file_name(); + if let Some(name) = name.to_str() + && let Some((_, end, _)) = parse_chunk_name(name) + && end == "open" + && entry.metadata().map(|m| m.len()).unwrap_or(1) == 0 + { + let _ = fs::remove_file(entry.path()); + } + } + } + let file_path = self.dir.join(format!("{seq}-open.raw.jsonl")); + let file = fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&file_path)?; + state.open_file = Some(file_path); + state.open_end = seq; + state.writer = Some(file); + Ok(state.writer.as_mut()) + } + + /// Renames the open chunk to the range it turned out to cover, so it + /// stops being open. + fn close_open_chunk(&self, state: &mut WriterState, end: u64) { + let file = state.open_file.clone(); + close_writer(state); + let Some(file) = file else { return }; + if let Some((first, _, _)) = file + .file_name() + .and_then(|n| n.to_str()) + .and_then(parse_chunk_name) + { + rename_chunk(&file, &self.dir, first, end); + } + } + + // -- failure ----------------------------------------------------------------------------- + + /// Runs `body`, answering `if_broken` when the directory cannot give a + /// real answer. None of this is reported on screen: every read here has + /// a network path beside it producing the same result, and the reader + /// has nothing to do about it. Damage discards this session's cache, + /// which makes the next open an ordinary cold one. + fn guard( + &self, + if_broken: T, + body: impl FnOnce(&Self, &mut WriterState) -> io::Result, + ) -> T { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + // A disk that refused once will refuse again, once per delta, so + // the first refusal is also the last. + if state.disabled { + return if_broken; + } + DAMAGED.with(|cell| *cell.borrow_mut() = None); + let result = body(self, &mut state); + // Damage takes priority over whatever `body` returned, `Ok` or + // `Err`: `suffix` signals it by returning `Err(damaged(..))` + // precisely so this check catches it before the branch below + // mistakes it for a real I/O failure and disables the whole cache + // over one corrupt session. + if let Some(file) = DAMAGED.with(|cell| cell.borrow_mut().take()) { + (self.warn)(&format!( + "transcript cache damaged at {}; discarding {}", + file.display(), + self.dir.display() + )); + close_writer(&mut state); + let _ = fs::remove_dir_all(&self.dir); + return if_broken; + } + match result { + Ok(value) => value, + Err(e) => { + (self.warn)(&format!("transcript cache unusable: {e}")); + state.disabled = true; + close_writer(&mut state); + if_broken + } + } + } +} + +thread_local! { + /// How [`SessionCache::each_line`] reports a line it cannot make sense + /// of back up to [`SessionCache::guard`], since the callback it hands + /// `each_line_backwards` cannot itself return a `Result`. Thread-local + /// rather than a field: the guard that reads it always runs on the same + /// call stack that could have set it, one `guard` call at a time. + static DAMAGED: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; +} + +fn damaged(file: &Path) -> io::Error { + DAMAGED.with(|cell| *cell.borrow_mut() = Some(file.to_path_buf())); + io::Error::other("damaged chunk") +} + +#[derive(Debug, Clone)] +struct Chunk { + file: PathBuf, + first: u64, + end: u64, + open: bool, + rows: bool, +} + +fn close_writer(state: &mut WriterState) { + state.writer = None; + state.open_file = None; + state.open_end = 0; +} + +fn rename_chunk(file: &Path, dir: &Path, first: u64, end: u64) { + let _ = fs::rename(file, dir.join(format!("{first}-{end}.raw.jsonl"))); +} + +/// `-..jsonl`; anything else in the directory is +/// not ours. +fn parse_chunk_name(name: &str) -> Option<(u64, &str, &str)> { + let rest = name.strip_suffix(".jsonl")?; + let (rest, kind) = rest.rsplit_once('.')?; + if kind != "rows" && kind != "raw" { + return None; + } + let (first, end) = rest.split_once('-')?; + let first = first.parse::().ok()?; + if end != "open" && end.parse::().is_err() { + return None; + } + Some((first, end, kind)) +} + +/// One line's sequence number, or `None` when the line is not one of ours. +/// +/// A hand-rolled scan rather than a JSON parse, so this module carries no +/// parser and stays testable with no server: the seq is the first field the +/// server writes, so the first match is the top-level one. +pub fn seq_of(line: &str) -> Option { + find_number_field(line, "seq") +} + +/// Whether a line is one streamed piece of a reply, which is what makes a +/// run of them one row. +pub fn is_delta(line: &str) -> bool { + find_string_field(line, "type").as_deref() == Some("assistantText") +} + +/// The value of `"key":N` (any amount of whitespace around the colon), or +/// `None`. Mirrors `Regex(""""seq"\s*:\s*(\d+)""")`'s first match. +fn find_number_field(line: &str, key: &str) -> Option { + let pattern = format!("\"{key}\""); + let at = line.find(&pattern)?; + let after_key = &line[at + pattern.len()..]; + let after_colon = after_key.trim_start().strip_prefix(':')?; + let digits: String = after_colon + .trim_start() + .chars() + .take_while(|c| c.is_ascii_digit()) + .collect(); + if digits.is_empty() { + None + } else { + digits.parse().ok() + } +} + +/// The value of `"key":"..."`, or `None`. Mirrors +/// `Regex(""""type"\s*:\s*"([^"]*)"""")`'s first match. +fn find_string_field(line: &str, key: &str) -> Option { + let pattern = format!("\"{key}\""); + let at = line.find(&pattern)?; + let after_key = &line[at + pattern.len()..]; + let after_colon = after_key.trim_start().strip_prefix(':')?; + let after_quote = after_colon.trim_start().strip_prefix('"')?; + let end = after_quote.find('"')?; + Some(after_quote[..end].to_string()) +} + +/// How much of a file is read at a time when walking it backwards. One +/// block covers a page of a transcript comfortably, and the walk stops as +/// soon as the caller has what it asked for. +const READ_BLOCK: usize = 64 * 1024; + +/// Calls `on_line` with each non-blank line of `file`, **newest first**, +/// along with the byte offset it starts at, until `on_line` answers false. +/// +/// Every question the cache is asked is about the newest end of a chunk, +/// and a live run reaches the size of the conversation, so reading forwards +/// means reading a transcript to answer with the last eighty lines of it. +/// +/// Splitting on bytes is safe because the separator is `\n`, which cannot +/// occur inside a multi-byte UTF-8 sequence; each line is decoded whole. A +/// missing file yields nothing. +fn each_line_backwards(file: &Path, mut on_line: impl FnMut(u64, &str) -> bool) { + use std::io::{Read, Seek, SeekFrom}; + let Ok(mut handle) = fs::File::open(file) else { + return; + }; + let Ok(len) = handle.metadata().map(|m| m.len()) else { + return; + }; + let mut unread = len; + let mut pending: Vec = Vec::new(); + while unread > 0 { + let take = READ_BLOCK.min(unread as usize); + let start = unread - take as u64; + let mut block = vec![0u8; take]; + if handle.seek(SeekFrom::Start(start)).is_err() || handle.read_exact(&mut block).is_err() { + return; + } + let mut buffer = block; + buffer.extend_from_slice(&pending); + // `buffer` is now `block` followed by `pending`; walk it backwards. + let mut line_end = buffer.len(); + let mut at = buffer.len() as isize - 1; + while at >= 0 { + if buffer[at as usize] == b'\n' { + let line_bytes = &buffer[at as usize + 1..line_end]; + let line = String::from_utf8_lossy(line_bytes); + if !line.trim().is_empty() && !on_line(start + at as u64 + 1, &line) { + return; + } + line_end = at as usize; + } + at -= 1; + } + pending = buffer[..line_end].to_vec(); + unread = start; + } + // The first line of a file has no newline before it to be found. + let first = String::from_utf8_lossy(&pending); + if !first.trim().is_empty() { + on_line(0, &first); + } +} + +/// Drops a final line that is not one of ours, by truncating the file to +/// where it starts. +/// +/// This app having died mid-write is the one kind of damage that is +/// repaired rather than discarded: the tail of an append-only file is the +/// only place a partial line can be. A second bad line is not this, and is +/// left for the read path to notice. +fn repair_tail(file: &Path) -> io::Result<()> { + let mut truncate_to: Option = None; + each_line_backwards(file, |offset, line| { + if seq_of(line).is_none() { + truncate_to = Some(offset); + } + false + }); + if let Some(truncate_to) = truncate_to { + let handle = fs::OpenOptions::new().write(true).open(file)?; + handle.set_len(truncate_to)?; + } + Ok(()) +} + +/// Runs `body`, translating an I/O or permission failure into `if_broken` +/// and a warning -- the disk half of [`SessionCache::guard`], shared with +/// [`TranscriptCache`]'s own maintenance. +fn guard_io( + if_broken: T, + warn: &(impl Fn(&str) + ?Sized), + body: impl FnOnce() -> io::Result, +) -> T { + match body() { + Ok(value) => value, + Err(e) => { + warn(&format!("transcript cache unusable: {e}")); + if_broken + } + } +} + +/// Sets a path's modified time, without pulling in a crate for it: a single +/// `utimensat`-backed call would be one more platform-specific dependency +/// for one call site, so this touches the file instead, which every +/// filesystem this runs on updates the mtime for. +fn filetime_set_modified(path: &Path, _when: std::time::SystemTime) -> io::Result<()> { + use std::io::Write; + // Rewriting a marker file's contents (rather than the directory itself, + // which `std` has no portable "touch" for) bumps the directory's own + // mtime on every filesystem this cache runs on, because creating or + // truncating an entry inside a directory always updates that + // directory's mtime. + let marker = path.join(".touch"); + let mut f = fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&marker)?; + f.write_all(b"")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashSet; + + fn cache(temp: &Path) -> TranscriptCache { + let said: std::sync::Arc>> = Default::default(); + let said2 = said.clone(); + TranscriptCache::with_warn(temp.join("v1/host_8443"), move |msg| { + said2.lock().unwrap().push(msg.to_string()); + }) + } + + /// Like `cache`, but also hands back the messages it warned with. + fn cache_with_log(temp: &Path) -> (TranscriptCache, std::sync::Arc>>) { + let said: std::sync::Arc>> = Default::default(); + let said2 = said.clone(); + ( + TranscriptCache::with_warn(temp.join("v1/host_8443"), move |msg| { + said2.lock().unwrap().push(msg.to_string()); + }), + said, + ) + } + + fn line(seq: u64, kind: &str) -> String { + format!(r#"{{"seq":{seq},"ts":1.5,"type":"{kind}","id":"x"}}"#) + } + fn tool_line(seq: u64) -> String { + line(seq, "toolStart") + } + fn delta(seq: u64) -> String { + line(seq, "assistantText") + } + + fn dir_of(temp: &Path, id: &str) -> PathBuf { + temp.join("v1/host_8443").join(id) + } + + fn names(temp: &Path, id: &str) -> Vec { + let dir = dir_of(temp, id); + let Ok(entries) = fs::read_dir(&dir) else { + return Vec::new(); + }; + let mut names: Vec = entries + .filter_map(|e| e.ok()) + .filter_map(|e| e.file_name().into_string().ok()) + .filter(|n| n != ".touch") + .collect(); + names.sort(); + names + } + + fn write_chunk(temp: &Path, id: &str, name: &str, lines: &[String]) { + let dir = dir_of(temp, id); + fs::create_dir_all(&dir).unwrap(); + let mut content = lines.join("\n"); + content.push('\n'); + fs::write(dir.join(name), content).unwrap(); + } + + fn seqs(lines: &Option>) -> Option> { + lines + .as_ref() + .map(|ls| ls.iter().map(|l| seq_of(l).unwrap()).collect()) + } + fn seqs_vec(lines: &[String]) -> Vec { + lines.iter().map(|l| seq_of(l).unwrap()).collect() + } + + #[test] + fn an_appended_run_is_one_open_chunk_and_its_newest_line_is_the_tail() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + for seq in 1..=3u64 { + session.append(&tool_line(seq), seq); + } + session.flush(); + + assert_eq!(names(temp.path(), "s"), vec!["1-open.raw.jsonl"]); + assert_eq!( + session.tail(), + Some(CachedTail { + seq: 3, + line: tool_line(3) + }) + ); + assert_eq!(session.newest(2), vec![tool_line(2), tool_line(3)]); + // More than there is is what there is, which is a short opening + // window and not a failure. + assert_eq!(session.newest(80).len(), 3); + } + + #[test] + fn a_gap_in_the_stream_closes_the_open_chunk_under_the_end_it_turned_out_to_have() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + for seq in 1..=3u64 { + session.append(&tool_line(seq), seq); + } + // What a `reset` looks like from here: the next event is not the + // one after the last. + session.append(&tool_line(90), 90); + session.flush(); + + assert_eq!( + names(temp.path(), "s"), + vec!["1-4.raw.jsonl", "90-open.raw.jsonl"] + ); + assert_eq!(session.newest(80), vec![tool_line(90)]); + assert_eq!( + session.tail(), + Some(CachedTail { + seq: 90, + line: tool_line(90) + }) + ); + } + + #[test] + fn an_event_already_covered_is_not_written_again() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + for seq in 1..=3u64 { + session.append(&tool_line(seq), seq); + } + session.append(&tool_line(2), 2); + session.flush(); + + assert_eq!(seqs_vec(&session.newest(80)), vec![1, 2, 3]); + } + + #[test] + fn an_adjacent_page_extends_the_suffix_and_a_gap_stops_it() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + for seq in 100..=102u64 { + session.append(&tool_line(seq), seq); + } + session.flush(); + + // Adjacent: its end is the open chunk's first. + let page: Vec = (60..100u64).map(tool_line).collect(); + assert!(session.store_page(&page, 60, 100, true)); + assert_eq!(seqs(&session.page(100, 2, false)), Some(vec![98, 99])); + assert_eq!(seqs_vec(&session.newest(80)).first(), Some(&60)); + + // Behind a gap: kept on disk, because paging usually closes the + // gap, but never served across it. + let page2: Vec = (1..10u64).map(tool_line).collect(); + assert!(session.store_page(&page2, 1, 10, true)); + assert_eq!(session.page(10, 5, false), None); + assert_eq!(seqs_vec(&session.newest(200)).first(), Some(&60)); + } + + #[test] + fn a_page_that_overlaps_what_is_here_is_not_stored() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + session.append(&tool_line(100), 100); + session.flush(); + let page: Vec = (60..100u64).map(tool_line).collect(); + assert!(session.store_page(&page, 60, 100, true)); + + let overlap: Vec = (50..80u64).map(tool_line).collect(); + assert!(!session.store_page(&overlap, 50, 80, true)); + assert!(!session.store_page(&[], 40, 60, true)); + assert_eq!( + names(temp.path(), "s"), + vec!["100-open.raw.jsonl", "60-100.rows.jsonl"] + ); + } + + #[test] + fn a_miss_is_null_and_never_an_empty_page() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + for seq in 100..=102u64 { + session.append(&tool_line(seq), seq); + } + session.flush(); + + // At or below where the run starts, so what the reader is + // scrolling into is the server's. + assert_eq!(session.page(100, 40, true), None); + assert_eq!(session.page(40, 40, true), None); + assert_eq!(cache.session("never-visited").page(100, 40, true), None); + } + + #[test] + fn a_page_starts_from_anywhere_inside_the_run_not_only_at_a_boundary() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + for seq in 1..=10u64 { + session.append(&tool_line(seq), seq); + } + session.flush(); + + assert_eq!(seqs(&session.page(8, 3, false)), Some(vec![5, 6, 7])); + assert_eq!(seqs(&session.page(8, 99, false)), Some((1..=7).collect())); + } + + #[test] + fn a_page_counted_in_rows_folds_each_delta_run_into_one_and_cuts_only_between_rows() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + // Two replies of three deltas each, split by a tool call: the same + // fixture as the server's `coalescing_counts_rows_and_joins_delta_runs`. + let lines = vec![ + delta(1), + delta(2), + delta(3), + tool_line(4), + delta(5), + delta(6), + delta(7), + tool_line(8), + ]; + write_chunk(temp.path(), "s", "1-9.raw.jsonl", &lines); + session.append(&tool_line(9), 9); + session.flush(); + + // Three rows: the tool call at 8, the run 5..7, and the tool call + // at 4. The cut lands between rows, so the older run is not + // started. + assert_eq!(seqs(&session.page(9, 3, true)), Some(vec![4, 5, 6, 7, 8])); + // One row is one whole run, however many deltas it is made of. + assert_eq!(seqs(&session.page(9, 1, true)), Some(vec![8])); + // A page of lines counts lines, which is what the anchor restore + // asks for. + assert_eq!(seqs(&session.page(9, 2, false)), Some(vec![7, 8])); + } + + #[test] + fn a_row_page_crosses_a_chunk_boundary_and_stops_short_at_the_oldest_chunk() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + write_chunk( + temp.path(), + "s", + "5-9.raw.jsonl", + &[delta(5), delta(6), tool_line(7), delta(8)], + ); + session.append(&delta(9), 9); + session.append(&tool_line(10), 10); + session.flush(); + + // A run straddling the boundary is one row, as it will be once folded. + assert_eq!(seqs(&session.page(11, 2, true)), Some(vec![8, 9, 10])); + // Asking for more rows than the suffix holds is a short page, not a + // failure and not a claim that the conversation starts here. + assert_eq!(seqs(&session.page(11, 40, true)), Some((5..=10).collect())); + } + + #[test] + fn the_floor_for_a_fetch_is_the_nearest_chunk_at_or_below_it() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + write_chunk( + temp.path(), + "s", + "1-10.rows.jsonl", + &(1..10u64).map(tool_line).collect::>(), + ); + write_chunk( + temp.path(), + "s", + "10-40.rows.jsonl", + &(10..40u64).map(tool_line).collect::>(), + ); + session.append(&tool_line(90), 90); + session.flush(); + + // The run behind the gap, which is what makes the fetched page + // adjacent to it. + assert_eq!(session.covered_up_to(90), Some(40)); + assert_eq!(session.covered_up_to(41), Some(40)); + assert_eq!(session.covered_up_to(10), Some(10)); + // Nothing at or below the oldest chunk's start, so the page is + // bounded only by its limit. + assert_eq!(session.covered_up_to(9), None); + } + + #[test] + fn a_newest_chunk_that_is_not_raw_discards_the_session() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + write_chunk( + temp.path(), + "s", + "1-10.rows.jsonl", + &(1..10u64).map(tool_line).collect::>(), + ); + + // Only reachable by dying between closing one live run and opening + // the next, and there is no cursor to be read off a coalesced line + // -- so the open is a cold one. + assert_eq!(session.tail(), None); + assert!(!dir_of(temp.path(), "s").exists()); + } + + #[test] + fn a_half_written_last_line_is_dropped_and_the_file_repaired() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + let dir = dir_of(temp.path(), "s"); + fs::create_dir_all(&dir).unwrap(); + fs::write( + dir.join("1-open.raw.jsonl"), + format!("{}\n{}\n{{\"se", tool_line(1), tool_line(2)), + ) + .unwrap(); + + assert_eq!( + session.tail(), + Some(CachedTail { + seq: 2, + line: tool_line(2) + }) + ); + assert_eq!( + fs::read_to_string(dir.join("1-open.raw.jsonl")).unwrap(), + format!("{}\n{}\n", tool_line(1), tool_line(2)) + ); + // And the run continues from where the good tail left off. + session.append(&tool_line(3), 3); + session.flush(); + assert_eq!(seqs_vec(&session.newest(80)), vec![1, 2, 3]); + } + + #[test] + fn damage_anywhere_else_discards_the_session_when_a_read_reaches_it() { + let temp = tempfile::tempdir().unwrap(); + let (cache, said) = cache_with_log(temp.path()); + let session = cache.session("s"); + write_chunk( + temp.path(), + "s", + "1-open.raw.jsonl", + &[tool_line(1), "not ours".to_string(), tool_line(3)], + ); + + // Not seen by the tail, which reads the newest line and stops. + assert_eq!( + session.tail(), + Some(CachedTail { + seq: 3, + line: tool_line(3) + }) + ); + // Reached by a read that walks past it: what is served is nothing, + // and the session opens cold from here on. + assert_eq!(session.newest(80), Vec::::new()); + assert!(!dir_of(temp.path(), "s").exists()); + assert!(said.lock().unwrap().iter().any(|m| m.contains("damaged"))); + } + + #[test] + fn a_name_this_does_not_recognise_is_ignored() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + write_chunk(temp.path(), "s", "notes.txt", &["hello".to_string()]); + write_chunk(temp.path(), "s", "1-open.raw.jsonl", &[tool_line(1)]); + + assert_eq!( + session.tail(), + Some(CachedTail { + seq: 1, + line: tool_line(1) + }) + ); + } + + #[test] + fn a_chunk_larger_than_one_read_block_is_walked_across_the_boundaries() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + // Well past the 64 kB block the backwards reader takes at a time, + // so a page has to be stitched across several of them -- including + // a line that straddles a boundary. + let padding = "x".repeat(300); + let lines: Vec = (1..=500u64) + .map(|seq| format!(r#"{{"seq":{seq},"ts":1.5,"type":"toolStart","id":"{padding}"}}"#)) + .collect(); + write_chunk(temp.path(), "s", "1-open.raw.jsonl", &lines); + + assert_eq!(session.tail().unwrap().seq, 500); + assert_eq!(session.newest(80), lines[420..].to_vec()); + assert_eq!(session.page(401, 999, false), Some(lines[0..400].to_vec())); + // And a non-ASCII line, whose bytes a naive split could cut through + // a character. + let accented = + r#"{"seq":501,"ts":1.5,"type":"assistantText","delta":"héllo — ok"}"#.to_string(); + session.append(&accented, 501); + session.flush(); + assert_eq!(session.tail().unwrap().line, accented); + } + + #[test] + fn eviction_takes_the_least_recently_touched_and_never_the_one_on_screen() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + for (at, id) in ["old", "middle", "open"].iter().enumerate() { + write_chunk( + temp.path(), + id, + "1-open.raw.jsonl", + &(1..=50u64).map(tool_line).collect::>(), + ); + let when = + std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000 + at as u64); + filetime_set_modified(&dir_of(temp.path(), id), when).unwrap(); + // The mtime touch above always sets "now", not `when` (see its + // own doc) -- space the three writes out in real time instead, + // since only relative order matters to eviction. + std::thread::sleep(std::time::Duration::from_millis(10)); + } + let each = dir_size(&dir_of(temp.path(), "old")); + + // Room for two of the three, so the oldest goes -- and the session + // being read never does, however long ago it was last touched. + cache.evict_to_budget("open", each * 2); + let mut remaining = fs::read_dir(temp.path().join("v1/host_8443")) + .unwrap() + .filter_map(|e| e.ok()) + .filter_map(|e| e.file_name().into_string().ok()) + .collect::>(); + remaining.sort(); + assert_eq!(remaining, vec!["middle", "open"]); + + cache.evict_to_budget("open", 0); + let mut remaining = fs::read_dir(temp.path().join("v1/host_8443")) + .unwrap() + .filter_map(|e| e.ok()) + .filter_map(|e| e.file_name().into_string().ok()) + .collect::>(); + remaining.sort(); + assert_eq!(remaining, vec!["open"]); + } + + #[test] + fn retaining_deletes_exactly_the_sessions_the_server_no_longer_lists() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + for id in ["a", "b", "c"] { + write_chunk(temp.path(), id, "1-open.raw.jsonl", &[tool_line(1)]); + } + + cache.retain_only(&HashSet::from(["a".to_string(), "c".to_string()])); + let mut remaining = fs::read_dir(temp.path().join("v1/host_8443")) + .unwrap() + .filter_map(|e| e.ok()) + .filter_map(|e| e.file_name().into_string().ok()) + .collect::>(); + remaining.sort(); + assert_eq!(remaining, vec!["a", "c"]); + } + + #[test] + fn size_and_purge_are_the_two_halves_of_the_reload_button() { + let temp = tempfile::tempdir().unwrap(); + let cache = cache(temp.path()); + let session = cache.session("s"); + assert_eq!(session.bytes(), 0); + for seq in 1..=5u64 { + session.append(&tool_line(seq), seq); + } + session.flush(); + + assert!(session.bytes() > 0); + session.purge(); + assert_eq!(session.bytes(), 0); + assert_eq!(session.tail(), None); + // And the session is usable again straight afterwards, which is + // what a reload does next. + session.append(&tool_line(9), 9); + session.flush(); + assert_eq!(seqs_vec(&session.newest(80)), vec![9]); + } +}