//! Append-only JSONL event log, one per session, with monotonically //! increasing sequence numbers -- the phone's resume cursor. //! //! One line per event: `{"seq":N,"ts":...,"type":...,...}`. The writer //! assigns sequence numbers; readers replay everything after a cursor. //! Reopening an existing file continues the numbering, which is what makes //! a backend restart invisible to a phone holding a cursor. use std::fs::{File, OpenOptions}; use std::io::{BufRead, BufReader, Write}; use std::os::unix::fs::OpenOptionsExt; use std::path::Path; use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use super::driver::Event; /// One transcript line: an [`Event`] plus its position and time. The event /// is flattened so the wire shape stays one flat object. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct SeqEvent { pub seq: u64, /// Epoch seconds. pub ts: f64, #[serde(flatten)] pub event: Event, } pub struct Transcript { file: File, next_seq: u64, } impl Transcript { /// Opens (or creates) the log at `path`, continuing the sequence from /// the last line if one exists. pub fn open(path: &Path) -> Result { let last_seq = last_seq(path)?; // Owner-only: a transcript is the whole conversation, including // whatever the session read, wrote, or was told. let file = OpenOptions::new() .create(true) .append(true) .mode(0o600) .open(path) .with_context(|| format!("open transcript {}", path.display()))?; Ok(Self { file, next_seq: last_seq + 1 }) } /// Appends `event`, assigning it the next sequence number. Flushed per /// event: each line is tiny, and the transcript is the source of truth /// a crash must not lose the tail of. pub fn append(&mut self, event: Event, ts: f64) -> Result { let entry = SeqEvent { seq: self.next_seq, ts, event }; let mut line = serde_json::to_string(&entry).context("serialize event")?; line.push('\n'); self.file.write_all(line.as_bytes()).context("append to transcript")?; self.next_seq += 1; Ok(entry) } } /// Replays every event with `seq > after`, oldest first. A missing file is /// an empty transcript, not an error -- the session just hasn't produced an /// event yet. pub fn read_after(path: &Path, after: u64) -> Result> { let file = match File::open(path) { Ok(file) => file, Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), Err(err) => return Err(err).with_context(|| format!("read transcript {}", path.display())), }; let mut events = Vec::new(); for line in BufReader::new(file).lines() { let line = line.context("read transcript line")?; if line.trim().is_empty() { continue; } let entry: SeqEvent = serde_json::from_str(&line) .with_context(|| format!("bad transcript line in {}", path.display()))?; if entry.seq > after { events.push(entry); } } Ok(events) } fn last_seq(path: &Path) -> Result { Ok(read_after(path, 0)?.last().map(|entry| entry.seq).unwrap_or(0)) } #[cfg(test)] mod tests { use super::*; use crate::session::driver::SessionStatus; fn text(delta: &str) -> Event { Event::AssistantText { delta: delta.to_string() } } #[test] fn assigns_increasing_seqs_and_replays_after_a_cursor() { let dir = tempfile::tempdir().expect("tempdir"); let path = dir.path().join("transcript.jsonl"); let mut transcript = Transcript::open(&path).expect("open"); assert_eq!(transcript.append(text("a"), 1.0).expect("append").seq, 1); assert_eq!(transcript.append(text("b"), 2.0).expect("append").seq, 2); assert_eq!(transcript.append(text("c"), 3.0).expect("append").seq, 3); let replay = read_after(&path, 1).expect("read"); assert_eq!(replay.len(), 2); assert_eq!(replay[0].seq, 2); assert_eq!(replay[0].event, text("b")); assert_eq!(replay[1].seq, 3); // A cursor at or past the end replays nothing. assert!(read_after(&path, 3).expect("read").is_empty()); } #[test] fn reopening_continues_the_numbering() { let dir = tempfile::tempdir().expect("tempdir"); let path = dir.path().join("transcript.jsonl"); let mut transcript = Transcript::open(&path).expect("open"); transcript.append(text("a"), 1.0).expect("append"); transcript.append(text("b"), 2.0).expect("append"); drop(transcript); let mut reopened = Transcript::open(&path).expect("reopen"); assert_eq!(reopened.append(text("c"), 3.0).expect("append").seq, 3); } #[test] fn a_missing_file_reads_as_empty() { let dir = tempfile::tempdir().expect("tempdir"); assert!(read_after(&dir.path().join("nope.jsonl"), 0).expect("read").is_empty()); } #[test] fn round_trips_every_event_shape() { let dir = tempfile::tempdir().expect("tempdir"); let path = dir.path().join("transcript.jsonl"); let events = vec![ Event::UserMessage { text: "hi".into() }, text("hello"), Event::ToolStart { id: "t1".into(), tool: "bash".into(), input: serde_json::json!({"command": "ls"}), }, Event::ToolUpdate { id: "t1".into(), output: "partial".into() }, Event::ToolEnd { id: "t1".into(), output: "done".into() }, Event::Image { image: "img1".into() }, Event::Question { id: "q1".into(), prompt: "Allow?".into(), options: vec!["Yes".into(), "No".into()], }, Event::Answered { id: "q1".into(), answer: "Yes".into() }, Event::Status { state: SessionStatus::Idle }, Event::UsageDelta { tokens: 42 }, Event::Error { message: "boom".into() }, ]; let mut transcript = Transcript::open(&path).expect("open"); for event in &events { transcript.append(event.clone(), 0.0).expect("append"); } let replayed: Vec = read_after(&path, 0) .expect("read") .into_iter() .map(|entry| entry.event) .collect(); assert_eq!(replayed, events); } }