ai-app/app-rust/src/client/transcript_source.rs

//! Where a session screen gets a transcript from: this phone's copy first,
//! the server for the rest. Ported from `app/.../TranscriptSource.kt`; see
//! `docs/TRANSCRIPT_CACHE.md` for the design this implements and
//! `docs/CLIENT_CORE.md` for how this file corresponds to the Kotlin.
//!
//! One seam rather than a cache the screen has to remember to consult.
//! Everything fetched before is asked of this, and everything the server
//! sends is written into the cache on the way past, so a caller never
//! learns which side answered. The one rule worth keeping in mind: the
//! cache is never load-bearing. Every read here has a network path beside
//! it producing the same result.
//!
//! **Not ported**: `EventStream.kt`'s reconnect-with-backoff loop and the
//! ability to close a live stream from another thread. Both are wall-clock
//! and thread-lifetime concerns that belong to whatever runtime the caller
//! embeds this crate in (a Tokio task, an iris timer, a Kotlin coroutine
//! scope) rather than to this pure logic -- `follow` below is the same
//! decorator shape `iris/desktop-app/src/app.rs` and
//! `iris/android-app/src/transcript_client.rs` already hand-wrote around
//! `event_stream::follow_session_events`, just with the cache write built
//! in so a future caller does not have to repeat it a third time.

use event_model::SeqEvent;

use crate::client::api::{ApiClient, ApiError, Transport};
use crate::client::event_stream::{self, StreamItem};
use crate::client::transcript_cache::SessionCache;

/// How many events a session screen opens with, cached or fetched.
///
/// The server's own default page size, named here because the cached
/// opening has to be the same size as the fetched one -- a reader must not
/// get a shorter first screen for having been here before (`OPENING_WINDOW`
/// in the Kotlin original).
pub const OPENING_WINDOW: u32 = 80;

/// A transcript-line parse failure, told apart from [`ApiError`] so a
/// caller can tell "the server is unreachable" from "the server (or this
/// phone's own disk) sent something this build cannot read" -- the two
/// mean different things to a reader (retry, versus a build that is
/// behind).
#[derive(Debug, Clone)]
pub struct ParseError(pub String);

impl std::fmt::Display for ParseError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}
impl std::error::Error for ParseError {}

/// Either half of what can go wrong asking for a page: the network, or a
/// line neither the cache's nor the server's copy of `parseSeqEvent` could
/// read.
#[derive(Debug, Clone)]
pub enum PageError {
    Api(ApiError),
    Parse(ParseError),
}

impl From<ApiError> for PageError {
    fn from(e: ApiError) -> Self {
        Self::Api(e)
    }
}

impl From<ParseError> for PageError {
    fn from(e: ParseError) -> Self {
        Self::Parse(e)
    }
}

/// What [`TranscriptSource::page`] found, kept as two states rather than
/// one possibly-empty list.
///
/// The difference is the whole of AGENTS.md's `loadOlderPage` incident: an
/// empty [`Self::Events`] means "this conversation has no more history",
/// which a caller is meant to latch, and [`Self::NothingLoaded`] means the
/// question could not be asked yet, which it must not. Collapsing the two
/// into an empty `Vec` puts the bug back, because the caller cannot tell
/// them apart -- and `unwrap_or_default()` on an `Option` would do the
/// same silently.
#[derive(Debug, Clone, PartialEq)]
pub enum OlderPage {
    /// The events before the cursor, oldest first. Empty means the start
    /// of the conversation has been reached.
    Events(Vec<SeqEvent>),
    /// Nothing is loaded, so there was no cursor to page back from
    /// (`before == 0`). Not an answer about the conversation at all.
    NothingLoaded,
}

fn parse_line(line: &str) -> Result<SeqEvent, ParseError> {
    serde_json::from_str(line).map_err(|e| ParseError(format!("{e}")))
}

/// This phone's copy of one session's transcript, plus the server it
/// falls back to. Ported from the Kotlin `TranscriptSource` class.
pub struct TranscriptSource<T: Transport> {
    api: ApiClient<T>,
    session_id: String,
    pub cache: SessionCache,
}

impl<T: Transport> TranscriptSource<T> {
    pub fn new(api: ApiClient<T>, session_id: impl Into<String>, cache: SessionCache) -> Self {
        Self {
            api,
            session_id: session_id.into(),
            cache,
        }
    }

    /// The cached opening window, or `None` when there is nothing usable
    /// to draw.
    ///
    /// Meant to be drawn *before* [`Self::probe`] returns, which is the
    /// whole point of the feature: the rows are on screen while the check
    /// that they are still the server's rows is in flight, and a failed
    /// check replaces them exactly as a reset does.
    pub fn cached_opening(&self, limit: usize) -> Option<Vec<SeqEvent>> {
        self.cache.tail()?;
        let lines = self.cache.newest(limit);
        if lines.is_empty() {
            return None;
        }
        match lines.iter().map(|l| parse_line(l)).collect() {
            Ok(events) => Some(events),
            // A line this build cannot read at all, which the cache's own checks cannot
            // see: it reads a seq off a line, not an event. Nothing to serve, so a cold
            // open.
            Err(ParseError(_)) => {
                self.cache.purge();
                None
            }
        }
    }

    /// Whether the server's event at the cached cursor is still the cached
    /// one.
    ///
    /// A caller must not resume a live stream from a cached seq unless it
    /// is the same conversation: a transcript is append-only in ordinary
    /// use, but the file backing it can be replaced or truncated (a
    /// sandbox re-seeded with the same ids, a backup restored, a session
    /// re-imported), and the server's catch-up on such a file would hand
    /// this phone a continuation of a *different* conversation, spliced
    /// onto the cached one with no seam. Caught with one request of a few
    /// hundred bytes.
    ///
    /// `Ok(false)` purges the cache and means "open cold". `Err` is the
    /// server not being askable, which is neither: the cached rows stay
    /// on screen and the caller tries again on its own reconnect schedule.
    ///
    /// What this cannot see is a line changed in the middle of the file
    /// with the tail intact -- that is what a full reload is for.
    pub fn probe(&self) -> Result<bool, ApiError> {
        let Some(tail) = self.cache.tail() else {
            return Ok(false);
        };
        // `before = seq + 1` is the newest event with seq <= the cursor, which is the
        // event *at* the cursor when the server still has one there.
        let page = self.api.fetch_transcript_lines(
            &self.session_id,
            Some(tail.seq + 1),
            1,
            false,
            None,
        )?;
        let matches = page.len() == 1
            && parse_line(&tail.line)
                .map(|cached| cached == page[0].1)
                .unwrap_or(false);
        if !matches {
            self.cache.purge();
        }
        Ok(matches)
    }

    /// Today's opening fetch, kept as the start of the live run. Only
    /// called when the cache has nothing to open with, or when
    /// [`Self::probe`] said what it had was not the server's.
    pub fn fetch_opening(&self) -> Result<Vec<SeqEvent>, ApiError> {
        let page =
            self.api
                .fetch_transcript_lines(&self.session_id, None, OPENING_WINDOW, false, None)?;
        for (line, event) in &page {
            self.cache.append(line, event.seq);
        }
        self.cache.flush();
        Ok(page.into_iter().map(|(_, event)| event).collect())
    }

    /// The page before `before`: from the cache when it holds it,
    /// otherwise from the server bounded by what the cache already has.
    ///
    /// The server bound (`after`) is what keeps the cache worth having. A
    /// coalesced page reaches back as far as its row count takes it -- a
    /// single reply is hundreds of lines -- so a page fetched after the
    /// reader has been away could run straight past the cached run and
    /// overlap it, and an overlapping page cannot be stored. Told where
    /// this phone's copy starts, the server stops there instead.
    ///
    /// `before == 0` answers [`OlderPage::NothingLoaded`] without asking
    /// the cache or the server anything -- see AGENTS.md's "things that
    /// have bitten": there is no event before the first one, so the
    /// request is not a harmless no-op, and its empty answer is
    /// indistinguishable from having reached the start of history.
    /// Guarded here rather than left to every caller, because it is a fact
    /// about the question, not about who is asking it.
    pub fn page(&self, before: u64, limit: u32, coalesce: bool) -> Result<OlderPage, PageError> {
        if before == 0 {
            return Ok(OlderPage::NothingLoaded);
        }
        if let Some(lines) = self.cache.page(before, limit as usize, coalesce) {
            let events: Vec<SeqEvent> = lines
                .iter()
                .map(|l| parse_line(l).map_err(PageError::from))
                .collect::<Result<_, _>>()?;
            return Ok(OlderPage::Events(events));
        }
        let after = self.cache.covered_up_to(before).map(|v| v - 1);
        let page = self.api.fetch_transcript_lines(
            &self.session_id,
            Some(before),
            limit,
            coalesce,
            after,
        )?;
        if let Some((_, first_event)) = page.first() {
            // `before` rather than the newest line's seq: a coalesced page covers
            // everything up to the cursor it was asked with, and nothing in its lines
            // says so.
            let lines: Vec<String> = page.iter().map(|(line, _)| line.clone()).collect();
            self.cache
                .store_page(&lines, first_event.seq, before, coalesce);
        }
        Ok(OlderPage::Events(
            page.into_iter().map(|(_, event)| event).collect(),
        ))
    }

    /// [`event_stream::follow_session_events`], with every frame written to
    /// the cache before `on_item` sees it.
    ///
    /// Before, so that an event held back for a reader who is scrolled
    /// away is already on disk -- what the cache holds is what the server
    /// sent, not what a screen has got round to drawing. Flushed on each
    /// status change, which is a turn's boundary and the granularity a
    /// crash may as well lose, and once more when the stream ends.
    pub fn follow(
        &self,
        after: u64,
        mut on_item: impl FnMut(StreamItem) -> bool,
    ) -> Result<(), ApiError> {
        let cache = &self.cache;
        let result = event_stream::follow_session_events(
            self.api.transport(),
            &self.session_id,
            after,
            |item| {
                if let StreamItem::Event { raw, event } = &item {
                    cache.append(raw, event.seq);
                    if matches!(event.event, event_model::Event::Status { .. }) {
                        cache.flush();
                    }
                }
                on_item(item)
            },
        );
        cache.flush();
        result
    }

    /// Leaves the cache with everything it was given -- called once a
    /// caller is done with this source, mirroring the Kotlin `close`'s
    /// final flush (that method's stream cancellation itself is the
    /// runtime concern the module doc says is not ported here).
    pub fn close(&self) {
        self.cache.flush();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::client::api::{Body, RawResponse};
    use std::collections::VecDeque;
    use std::io::Read;
    use std::sync::Mutex;

    /// A transport that answers fixed bodies in call order, and records
    /// every path it was asked for -- so a test can assert *how many*
    /// requests a method made, which is the point for the `before == 0`
    /// guard (AGENTS.md's regression: the guard must stop the request
    /// before it happens, not merely tolerate the empty answer).
    #[derive(Default)]
    struct ScriptedTransport {
        responses: Mutex<VecDeque<(u16, String)>>,
        calls: Mutex<Vec<String>>,
    }

    impl ScriptedTransport {
        fn respond(&self, status: u16, body: impl Into<String>) {
            self.responses
                .lock()
                .unwrap()
                .push_back((status, body.into()));
        }

        fn call_count(&self) -> usize {
            self.calls.lock().unwrap().len()
        }
    }

    impl Transport for ScriptedTransport {
        fn request(
            &self,
            _method: &str,
            path: &str,
            _body: Option<Body>,
        ) -> Result<RawResponse, ApiError> {
            self.calls.lock().unwrap().push(path.to_string());
            let (status, body) = self
                .responses
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or_else(|| panic!("ScriptedTransport got an unscripted request: {path}"));
            Ok(RawResponse {
                status,
                body: body.into_bytes(),
            })
        }

        fn stream(&self, path: &str) -> Result<Box<dyn Read + Send>, ApiError> {
            self.calls.lock().unwrap().push(path.to_string());
            let (_, body) = self
                .responses
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or_else(|| {
                    panic!("ScriptedTransport got an unscripted stream request: {path}")
                });
            Ok(Box::new(std::io::Cursor::new(body.into_bytes())))
        }
    }

    fn source(
        transport: ScriptedTransport,
        cache_root: &std::path::Path,
    ) -> TranscriptSource<ScriptedTransport> {
        let api = ApiClient::new(transport);
        let cache = crate::client::transcript_cache::TranscriptCache::new(cache_root).session("s1");
        TranscriptSource::new(api, "s1", cache)
    }

    fn status_line(seq: u64) -> String {
        format!(r#"{{"seq":{seq},"ts":1.0,"type":"status","state":"idle"}}"#)
    }

    #[test]
    fn a_cold_cache_has_no_opening_and_fetches_from_the_server() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{}]", status_line(1)));
        let source = source(transport, dir.path());

        assert_eq!(source.cached_opening(80), None);
        let opening = source.fetch_opening().unwrap();
        assert_eq!(opening.len(), 1);
        assert_eq!(opening[0].seq, 1);
        // The fetch wrote through: reopening the same cache now has something to show.
        assert!(source.cache.tail().is_some());
    }

    #[test]
    fn probe_matching_the_cached_tail_leaves_the_cache_alone() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{}]", status_line(1)));
        let source = source(transport, dir.path());
        source.fetch_opening().unwrap();

        let transport2 = ScriptedTransport::default();
        transport2.respond(200, format!("[{}]", status_line(1)));
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        let source2 = TranscriptSource::new(ApiClient::new(transport2), "s1", cache);
        assert!(source2.probe().unwrap());
        assert!(source2.cache.tail().is_some());
    }

    #[test]
    fn probe_mismatching_the_cached_tail_purges_the_cache() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{}]", status_line(1)));
        let source = source(transport, dir.path());
        source.fetch_opening().unwrap();

        // The server now answers with a different event at the same seq -- the file
        // behind this session was replaced.
        let transport2 = ScriptedTransport::default();
        let different = r#"{"seq":1,"ts":1.0,"type":"status","state":"running"}"#.to_string();
        transport2.respond(200, format!("[{different}]"));
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        let source2 = TranscriptSource::new(ApiClient::new(transport2), "s1", cache);
        assert!(!source2.probe().unwrap());
        assert!(source2.cache.tail().is_none());
    }

    #[test]
    fn probe_finding_no_server_leaves_the_cache_untouched() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{}]", status_line(1)));
        let source = source(transport, dir.path());
        source.fetch_opening().unwrap();

        let transport2 = ScriptedTransport::default();
        transport2.respond(500, "server on fire");
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        let source2 = TranscriptSource::new(ApiClient::new(transport2), "s1", cache);
        assert!(source2.probe().is_err());
        assert!(
            source2.cache.tail().is_some(),
            "an unreachable server must not be treated as a mismatch"
        );
    }

    /// The regression this module exists to close: `before == 0` must
    /// never reach the network or the cache, because an empty answer there
    /// is indistinguishable from "there is genuinely no more history" --
    /// AGENTS.md's `loadOlderPage` incident.
    #[test]
    fn paging_before_the_first_event_makes_no_request_at_all() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        let source = source(transport, dir.path());
        assert_eq!(source.page(0, 80, true).unwrap(), OlderPage::NothingLoaded);
        assert_eq!(source.api.transport().call_count(), 0);
    }

    #[test]
    fn a_page_already_covered_by_the_cache_never_reaches_the_server() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{},{}]", status_line(1), status_line(2)));
        let source = source(transport, dir.path());
        source.fetch_opening().unwrap();

        let calls_before = source.api.transport().call_count();
        let OlderPage::Events(page) = source.page(2, 10, true).unwrap() else {
            panic!("a cursor of 2 is a real question about the conversation");
        };
        assert_eq!(page.len(), 1);
        assert_eq!(page[0].seq, 1);
        assert_eq!(
            source.api.transport().call_count(),
            calls_before,
            "a cache hit must not touch the network"
        );
    }

    /// With nothing older cached there is no floor to give the server, so
    /// the request carries no `after` at all.
    #[test]
    fn a_server_page_with_nothing_older_cached_carries_no_bound() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{}]", status_line(5)));
        let source = source(transport, dir.path());
        source.fetch_opening().unwrap();

        let transport2 = ScriptedTransport::default();
        transport2.respond(200, format!("[{}]", status_line(3)));
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        let source2 = TranscriptSource::new(ApiClient::new(transport2), "s1", cache);
        source2.page(5, 10, true).unwrap();
        assert_eq!(
            source2.api.transport().calls.lock().unwrap()[0],
            "/sessions/s1/transcript?limit=10&before=5&coalesce=true"
        );
    }

    /// The half the test above cannot show: when the cache *does* hold an
    /// older run, the fetch is floored at its end, or the page would run
    /// straight past it and overlap -- which `store_page` then refuses,
    /// silently costing the phone the page it just paid for.
    #[test]
    fn a_server_page_is_floored_at_the_end_of_the_cached_run() {
        let dir = tempfile::tempdir().unwrap();
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        // A stored page covering [3, 6) and two live events above it, so the run this
        // phone holds is [3, 8) -- the newest chunk has to be an appended one, or the
        // cache reads the directory as damaged and discards it.
        let lines: Vec<String> = (3..6).map(status_line).collect();
        assert!(cache.store_page(&lines, 3, 6, true));
        cache.append(&status_line(6), 6);
        cache.append(&status_line(7), 7);
        cache.flush();

        let transport = ScriptedTransport::default();
        transport.respond(200, format!("[{}]", status_line(9)));
        let source = TranscriptSource::new(ApiClient::new(transport), "s1", cache);
        source.page(10, 10, true).unwrap();
        assert_eq!(
            source.api.transport().calls.lock().unwrap()[0],
            "/sessions/s1/transcript?limit=10&before=10&coalesce=true&after=7",
            "the fetch must stop one seq below where this phone's copy ends"
        );
    }

    /// A page the server could not answer is an error, never an empty
    /// page: the caller would read the second as "this conversation has no
    /// more history" and stop paging for good.
    #[test]
    fn a_failing_server_page_is_an_error_rather_than_an_empty_one() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(500, "server on fire");
        let source = source(transport, dir.path());
        assert!(matches!(source.page(9, 10, true), Err(PageError::Api(_)),));
    }

    /// A cached line this build cannot read is told apart from the network
    /// failing, for the same reason: neither is "no more history".
    #[test]
    fn an_unreadable_cached_page_is_a_parse_error_rather_than_an_empty_one() {
        let dir = tempfile::tempdir().unwrap();
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        cache.store_page(
            &[r#"{"seq":3,"but":"not an event"}"#.to_string()],
            3,
            4,
            true,
        );
        cache.append(&status_line(4), 4);
        cache.flush();
        let transport = ScriptedTransport::default();
        let source = TranscriptSource::new(ApiClient::new(transport), "s1", cache);
        assert!(matches!(source.page(4, 10, true), Err(PageError::Parse(_)),));
        assert_eq!(
            source.api.transport().call_count(),
            0,
            "a cache hit that cannot be read must not fall through to the server unnoticed"
        );
    }

    #[test]
    fn a_bad_cached_opening_line_purges_rather_than_panicking() {
        let dir = tempfile::tempdir().unwrap();
        let cache = crate::client::transcript_cache::TranscriptCache::new(dir.path()).session("s1");
        cache.append("not json at all", 1);
        cache.flush();
        let transport = ScriptedTransport::default();
        let source = TranscriptSource::new(ApiClient::new(transport), "s1", cache);
        assert_eq!(source.cached_opening(80), None);
        assert!(
            source.cache.tail().is_none(),
            "a damaged line purges the cache"
        );
    }

    #[test]
    fn follow_writes_events_to_the_cache_before_the_caller_sees_them() {
        let dir = tempfile::tempdir().unwrap();
        let transport = ScriptedTransport::default();
        transport.respond(200, format!("{}\n\n", sse_frame(&status_line(1))));
        let source = source(transport, dir.path());
        let mut seen = Vec::new();
        source
            .follow(0, |item| {
                if let StreamItem::Event { event, .. } = item {
                    seen.push(event.seq);
                }
                true
            })
            .unwrap();
        assert_eq!(seen, vec![1]);
        assert_eq!(source.cache.tail().unwrap().seq, 1);
    }

    fn sse_frame(data: &str) -> String {
        format!("data:{data}")
    }
}