ai-app/server/src/session/transcript.rs

//! 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<Self> {
        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<SeqEvent> {
        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<Vec<SeqEvent>> {
    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<u64> {
    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<Event> = read_after(&path, 0)
            .expect("read")
            .into_iter()
            .map(|entry| entry.event)
            .collect();
        assert_eq!(replayed, events);
    }
}