diff --git a/client-core/Cargo.toml b/client-core/Cargo.toml index 953bf55..2cff592 100644 --- a/client-core/Cargo.toml +++ b/client-core/Cargo.toml @@ -17,7 +17,12 @@ edition = "2024" [dependencies] event-model = { path = "../event-model" } serde = { version = "1", features = ["derive"] } -serde_json = { version = "1", features = ["float_roundtrip"] } +# "raw_value" is `fetch_transcript_lines`'s reason -- it needs the exact +# bytes the server sent, not this crate's own re-serialization of a parsed +# `Value`, so a cached line and a live SSE frame for the same event agree +# byte-for-byte (see that method's doc). "float_roundtrip" is why they +# agree on a `ts` at all -- see server/Cargo.toml's identical comment. +serde_json = { version = "1", features = ["float_roundtrip", "raw_value"] } # The blocking HTTP client for the REST calls and the long-lived SSE GETs. # `server/` already depends on ureq for its own outbound HTTPS (the usage # poll in usage.rs) and it is rustls-backed like the rest of this project's diff --git a/client-core/src/api.rs b/client-core/src/api.rs index 72fd047..22ab9e5 100644 --- a/client-core/src/api.rs +++ b/client-core/src/api.rs @@ -10,6 +10,7 @@ use std::io::Read; +use event_model::SeqEvent; use serde::Deserialize; use serde_json::Value; @@ -116,6 +117,14 @@ impl ApiClient { Self { transport } } + /// The transport underneath, for a caller that needs the raw SSE + /// stream (`event_stream::follow_session_events`) rather than one of + /// this client's typed REST calls -- `transcript_source::TranscriptSource` + /// is the one that does. + pub fn transport(&self) -> &T { + &self.transport + } + fn json_request Deserialize<'de>>( &self, method: &str, @@ -266,15 +275,72 @@ impl ApiClient { limit: u32, coalesce: bool, ) -> Result, ApiError> { - let mut path = format!("/sessions/{session_id}/transcript?limit={limit}"); - if let Some(before) = before { - path.push_str(&format!("&before={before}")); - } - if coalesce { - path.push_str("&coalesce=true"); - } - self.json_request("GET", &path, None) + self.json_request( + "GET", + &transcript_path(session_id, before, limit, coalesce, None), + None, + ) } + + /// A page of transcript history, each line handed back paired with the + /// exact text it came from, and bounded below by `after` -- the shape + /// `crate::transcript_source::TranscriptSource` needs to store what it + /// fetched in the transcript cache without a second round trip to fetch + /// the raw text separately. Ported from `Api.kt`'s `fetchTranscript`. + /// + /// Uses [`serde_json::value::RawValue`] rather than re-serializing a + /// parsed [`Value`], so the stored line is the exact bytes the server + /// sent (key order and float literal included) rather than this + /// crate's own idea of how to write them back out -- the cache and a + /// live SSE frame must agree byte-for-byte on the same event, which is + /// exactly what caught the `serde_json` float-rounding bug this + /// project's `AGENTS.md` records. + pub fn fetch_transcript_lines( + &self, + session_id: &str, + before: Option, + limit: u32, + coalesce: bool, + after: Option, + ) -> Result, ApiError> { + let path = transcript_path(session_id, before, limit, coalesce, after); + let raw: Vec> = self.json_request("GET", &path, None)?; + raw.into_iter() + .map(|value| { + let line = value.get().to_string(); + let event: SeqEvent = serde_json::from_str(&line).map_err(|e| ApiError { + message: format!( + "the server sent a transcript line this build couldn't parse: {e}" + ), + status: None, + })?; + Ok((line, event)) + }) + .collect() + } +} + +/// The query string shared by [`ApiClient::fetch_transcript_page`] and +/// [`ApiClient::fetch_transcript_lines`], so the two agree on how each +/// parameter is written rather than keeping two copies to drift. +fn transcript_path( + session_id: &str, + before: Option, + limit: u32, + coalesce: bool, + after: Option, +) -> String { + let mut path = format!("/sessions/{session_id}/transcript?limit={limit}"); + if let Some(before) = before { + path.push_str(&format!("&before={before}")); + } + if coalesce { + path.push_str("&coalesce=true"); + } + if let Some(after) = after { + path.push_str(&format!("&after={after}")); + } + path } /// The blocking [`Transport`] backed by `ureq`, the same crate `server/` diff --git a/client-core/src/lib.rs b/client-core/src/lib.rs index 852cc1e..a8458c5 100644 --- a/client-core/src/lib.rs +++ b/client-core/src/lib.rs @@ -11,5 +11,6 @@ pub mod notifications; pub mod sse; pub mod transcript_cache; pub mod transcript_fold; +pub mod transcript_source; pub use event_model::*; diff --git a/client-core/src/transcript_cache.rs b/client-core/src/transcript_cache.rs index a0ec342..7702989 100644 --- a/client-core/src/transcript_cache.rs +++ b/client-core/src/transcript_cache.rs @@ -361,6 +361,10 @@ impl SessionCache { { return Ok(false); } + debug_assert!( + lines.iter().all(|l| !l.contains('\n')), + "a stored page's lines must each be one line" + ); fs::create_dir_all(&this.dir)?; let kind = if rows { "rows" } else { "raw" }; let mut content = lines.join("\n"); @@ -389,8 +393,16 @@ impl SessionCache { return Ok(()); }; // Written as it arrived. A newline inside it would split one - // event into two unreadable halves, but neither source can - // produce one. + // event into two unreadable halves. No source here can produce + // one -- an SSE `data:` field cannot hold a raw newline, and a + // fetched line is one element of a compact JSON array -- but + // that is a fact about the *server's* serializer rather than + // anything this file controls, so it is checked rather than + // trusted. + debug_assert!( + !line.contains('\n'), + "a cached transcript line must be one line: {line}" + ); use std::io::Write; writer.write_all(line.as_bytes())?; writer.write_all(b"\n")?; diff --git a/client-core/src/transcript_fold.rs b/client-core/src/transcript_fold.rs index cc76316..a4cf03c 100644 --- a/client-core/src/transcript_fold.rs +++ b/client-core/src/transcript_fold.rs @@ -294,6 +294,199 @@ fn split_run(tail: &[TranscriptItem], behind: Option<&str>) -> Vec Vec { + let (older, newer) = heal_split_message(earlier, later); + let started_earlier: std::collections::HashSet<&str> = older + .iter() + .filter_map(TranscriptItem::as_tool_run) + .collect(); + // Owned rather than borrowed from `newer`: `kept` below needs to consume `newer` by + // value, and a map borrowing it would keep that alive. + let ended_later: std::collections::HashMap = newer + .iter() + .filter_map(|item| item.as_tool_run().map(|id| (id.to_string(), item.clone()))) + .filter(|(id, _)| started_earlier.contains(id.as_str())) + .collect(); + let healed: Vec = older + .into_iter() + .map(|row| match row { + TranscriptItem::ToolRun { + seq, + id, + run_id, + tool, + input, + asks: row_asks, + images: row_images, + .. + } if ended_later.contains_key(id.as_str()) => { + let &TranscriptItem::ToolRun { + ref output, + done, + asks: ref half_asks, + images: ref half_images, + .. + } = &ended_later[id.as_str()] + else { + unreachable!("filtered to ToolRun above"); + }; + TranscriptItem::ToolRun { + seq, + id, + run_id, + tool, + input, + output: output.clone(), + done, + // Kept from both halves: a question or an image can be + // attached to either, depending on which side of the + // boundary its event fell. + asks: row_asks.into_iter().chain(half_asks.clone()).collect(), + images: row_images.into_iter().chain(half_images.clone()).collect(), + } + } + other => other, + }) + .collect(); + let kept: Vec = newer + .into_iter() + .filter(|item| match item.as_tool_run() { + Some(id) => !ended_later.contains_key(id), + None => true, + }) + .collect(); + let mut out = adopt_run(&healed, &kept); + out.extend(kept); + // What this function exists to prevent, checked rather than assumed: the same + // call drawn twice, once from the page that saw its start and once from the page + // that saw its end. Not a seq-ordering check -- a peer note is stamped with the + // seq its turn began at, which can be older than the page it arrived in, so the + // two pages' seqs legitimately interleave at the boundary. + debug_assert!( + { + let mut ids: Vec<&str> = out.iter().filter_map(TranscriptItem::as_tool_run).collect(); + let before = ids.len(); + ids.sort_unstable(); + ids.dedup(); + ids.len() == before + }, + "join_pages left the same tool call in both halves" + ); + out +} + +/// Rejoins a message the page boundary cut, and hands back the two pages to +/// concatenate. Ported from `TranscriptItems.kt`'s `healSplitMessage`. +/// +/// `fold_event` never leaves two assistant messages next to each other +/// inside one page, so two meeting at a join are always the two halves of +/// one reply, and leaving them apart drew a single answer as two with a +/// paragraph break through the middle of a sentence. +/// +/// The newer half keeps its identity, for the reason `adopt_run`'s doc +/// gives. It grows by what the older half brings, which is safe here and +/// nowhere else -- the join is at the oldest end of what is loaded, so the +/// growth extends off the top of the screen. +fn heal_split_message( + earlier: &[TranscriptItem], + later: &[TranscriptItem], +) -> (Vec, Vec) { + let ( + Some(TranscriptItem::AssistantMsg { + text: head_text, .. + }), + Some(TranscriptItem::AssistantMsg { + seq: tail_seq, + text: tail_text, + settled: tail_settled, + }), + ) = (earlier.last(), later.first()) + else { + return (earlier.to_vec(), later.to_vec()); + }; + let merged = TranscriptItem::AssistantMsg { + seq: *tail_seq, + text: format!("{head_text}{tail_text}"), + settled: *tail_settled, + }; + let mut newer = vec![merged]; + newer.extend(later[1..].iter().cloned()); + (earlier[..earlier.len() - 1].to_vec(), newer) +} + +/// Hands the older calls at the join the name of the run they are joining. +/// Ported from `TranscriptItems.kt`'s `adoptRun`. +/// +/// The two pages were folded separately, so a run split by the boundary +/// came back as two runs with two names. Naming the joined run after the +/// *older* half would be the obvious way round and is wrong: the newer half +/// is the part already on screen, and renaming it is renaming the row the +/// reader is looking at, which is how a list loses its anchor. +fn adopt_run(earlier: &[TranscriptItem], later: &[TranscriptItem]) -> Vec { + let Some(TranscriptItem::ToolRun { run_id, tool, .. }) = later.first() else { + return earlier.to_vec(); + }; + // A question is in a run of its own on both sides of the join, the same as it would be + // had the two pages been folded as one. Without this the heal would merge a group + // straight through the row the reader was asked something on. + if tool == ASK_USER_QUESTION { + return earlier.to_vec(); + } + let joining = run_id.clone(); + let tail_len = earlier + .iter() + .rev() + .take_while(|item| matches!(item, TranscriptItem::ToolRun { tool, .. } if tool != ASK_USER_QUESTION)) + .count(); + if tail_len == 0 { + return earlier.to_vec(); + } + let split = earlier.len() - tail_len; + let mut out = earlier[..split].to_vec(); + out.extend(earlier[split..].iter().cloned().map(|mut item| { + // `take_while` above already restricted this slice to non-question tool calls; + // this just guards the invariant rather than trusting it silently. + debug_assert!( + matches!(&item, TranscriptItem::ToolRun { tool, .. } if tool != ASK_USER_QUESTION), + "adopt_run must never rename a question's own run" + ); + if let TranscriptItem::ToolRun { run_id, .. } = &mut item { + *run_id = joining.clone(); + } + item + })); + out +} + /// Folds one transcript event onto `items`, the way `foldEvent` does in /// `TranscriptItems.kt`. Every wire event has a case; see the module doc /// for the one difference from the Kotlin original (no `Unknown` fallback @@ -955,4 +1148,134 @@ mod tests { let err = fold_page(&values).unwrap_err(); assert!(err.contains("couldn't parse")); } + + fn tool_start(seq: u64, id: &str, tool: &str) -> SeqEvent { + event( + seq, + Event::ToolStart { + id: id.to_string(), + tool: tool.to_string(), + input: serde_json::json!({}), + }, + ) + } + + fn tool_end(seq: u64, id: &str, output: &str) -> SeqEvent { + event( + seq, + Event::ToolEnd { + id: id.to_string(), + output: output.to_string(), + }, + ) + } + + /// AGENTS.md's "things that have bitten": `joinPages` used to run + /// `adoptRun` only on the path where a *split* call was found, so a + /// boundary landing cleanly between two already-finished calls -- most + /// of them -- left the older page's calls under the run name they were + /// folded with, drawing one run of tool calls as two groups. Two + /// finished, unrelated calls (no id in common) must still end up under + /// one run name after the join. + #[test] + fn a_clean_boundary_between_two_finished_runs_is_still_healed_into_one_run() { + let older = fold_all(&[tool_start(1, "a", "Bash"), tool_end(2, "a", "old output")]); + let newer = fold_all(&[tool_start(3, "b", "Bash"), tool_end(4, "b", "new output")]); + let joined = join_pages(&older, &newer); + let run_ids: Vec<_> = joined + .iter() + .map(|item| match item { + TranscriptItem::ToolRun { run_id, .. } => run_id.as_str(), + other => panic!("expected only ToolRun items, got {other:?}"), + }) + .collect(); + assert_eq!( + run_ids, + vec!["b", "b"], + "the older call must adopt the newer, already-on-screen run's name" + ); + } + + #[test] + fn a_call_split_across_the_boundary_merges_into_one_row() { + let older = fold_all(&[tool_start(1, "x", "Bash")]); + let newer = fold_all(&[tool_end(2, "x", "the result")]); + let joined = join_pages(&older, &newer); + assert_eq!( + joined, + vec![TranscriptItem::ToolRun { + seq: 1, + id: "x".to_string(), + run_id: "x".to_string(), + tool: "Bash".to_string(), + input: "{}".to_string(), + output: "the result".to_string(), + done: true, + asks: Vec::new(), + images: Vec::new(), + }], + "the older half's tool/input and the newer half's output/done must both survive" + ); + } + + #[test] + fn a_message_split_across_the_boundary_is_rejoined_with_the_newer_halfs_identity() { + let older = vec![TranscriptItem::AssistantMsg { + seq: 1, + text: "Hel".to_string(), + settled: false, + }]; + let newer = vec![ + TranscriptItem::AssistantMsg { + seq: 2, + text: "lo".to_string(), + settled: true, + }, + TranscriptItem::UserMsg { + seq: 3, + text: "next".to_string(), + attachments: Vec::new(), + }, + ]; + let joined = join_pages(&older, &newer); + assert_eq!( + joined, + vec![ + TranscriptItem::AssistantMsg { + seq: 2, + text: "Hello".to_string(), + settled: true, + }, + TranscriptItem::UserMsg { + seq: 3, + text: "next".to_string(), + attachments: Vec::new(), + }, + ] + ); + } + + /// A question is in a run of its own on both sides of a join -- healing + /// must never rename the run of calls the reader was asked something + /// on, the same rule `splitRun` enforces for a live turn boundary. + #[test] + fn adopt_run_never_renames_into_a_question_row() { + let older = fold_all(&[tool_start(1, "a", "Bash"), tool_end(2, "a", "done")]); + let newer = vec![TranscriptItem::ToolRun { + seq: 3, + id: "q".to_string(), + run_id: "q".to_string(), + tool: ASK_USER_QUESTION.to_string(), + input: "{}".to_string(), + output: String::new(), + done: false, + asks: Vec::new(), + images: Vec::new(), + }]; + let joined = join_pages(&older, &newer); + match &joined[0] { + TranscriptItem::ToolRun { run_id, .. } => assert_eq!(run_id, "a"), + other => panic!("expected a ToolRun, got {other:?}"), + } + } } diff --git a/client-core/src/transcript_source.rs b/client-core/src/transcript_source.rs new file mode 100644 index 0000000..1bb0ee9 --- /dev/null +++ b/client-core/src/transcript_source.rs @@ -0,0 +1,588 @@ +//! 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::api::{ApiClient, ApiError, Transport}; +use crate::event_stream::{self, StreamItem}; +use crate::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 for PageError { + fn from(e: ApiError) -> Self { + Self::Api(e) + } +} + +impl From 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), + /// 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 { + 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 { + api: ApiClient, + session_id: String, + pub cache: SessionCache, +} + +impl TranscriptSource { + pub fn new(api: ApiClient, session_id: impl Into, 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> { + 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 { + 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, 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 { + if before == 0 { + return Ok(OlderPage::NothingLoaded); + } + if let Some(lines) = self.cache.page(before, limit as usize, coalesce) { + let events: Vec = lines + .iter() + .map(|l| parse_line(l).map_err(PageError::from)) + .collect::>()?; + 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 = 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::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>, + calls: Mutex>, + } + + impl ScriptedTransport { + fn respond(&self, status: u16, body: impl Into) { + 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, + ) -> Result { + 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, 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 { + let api = ApiClient::new(transport); + let cache = crate::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::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::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::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::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::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 = (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::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::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}") + } +} diff --git a/docs/CLIENT_CORE.md b/docs/CLIENT_CORE.md index bc07d25..d2bc7eb 100644 --- a/docs/CLIENT_CORE.md +++ b/docs/CLIENT_CORE.md @@ -25,16 +25,19 @@ next (a Masonry or iris transcript screen, most likely). | `sse.rs` | `Sse.kt` (the framing half) | Done, new tests (Kotlin had none of its own beyond integration) | | `api.rs` | `Api.kt` | Partial -- see below | | `event_stream.rs` | `EventStream.kt` | Done | -| `transcript_fold.rs` | `TranscriptItems.kt`, `ToolRows.kt` | Partial -- see below | +| `transcript_fold.rs` | `TranscriptItems.kt`, `ToolRows.kt` | Done -- see below | | `config.rs` | `ServerConfig.kt`'s `handleEnrollment` | New, desktop-only so far -- see below | -| *(not started)* | `TranscriptSource.kt` | Not started | +| `transcript_source.rs` | `TranscriptSource.kt` | Done -- see below | | *(not ported, and may never be)* | `TranscriptUnits.kt` | Out of scope -- see below | Every file above whose Kotlin counterpart had a JVM unit test (`AnsiTest`, `HighlighterTest`, `TranscriptCacheTest`) has had every one of those test cases ported alongside it, plus new tests for the pieces that had none -(`sse.rs`, `api.rs`, `event_stream.rs`, `transcript_fold.rs`). Test count by -crate as of this writing: **85 in `client-core`**, 0 in `event-model` (its +(`sse.rs`, `api.rs`, `event_stream.rs`, `transcript_fold.rs`, +`transcript_source.rs` -- the Kotlin `TranscriptSource.kt`/`TranscriptItems.kt` +had no JVM unit tests of their own, so these were written fresh against the +Kotlin source and AGENTS.md's paging incidents as the spec). Test count by +crate as of this writing: **109 in `client-core`**, 0 in `event-model` (its types carry no logic of their own to test -- `server/`'s own tests exercise them via `session::transcript`'s round-trip coverage). @@ -88,13 +91,27 @@ the full table to work from when one of these is next. including tool-call/question/image attachment and peer-message placement. `group_tool_runs` groups adjacent calls into `TranscriptRow::Tools`. -**Not ported:** `TranscriptItems.kt`'s `joinPages` (and its -`healSplitMessage`/`adoptRun` helpers) -- the page-boundary healing that -merges a tool call split across two fetched pages and re-merges a run a -boundary cut through. This matters the moment paging backward through -history is exercised; it is deliberately left rather than rushed, since -it is exactly the kind of boundary logic this project's own "things that -have bitten" section warns reads fine and is wrong at the edges. +`join_pages` (with `heal_split_message` and `adopt_run`, both private) is +now ported too, 2026-09-06 -- the page-boundary healing that merges a tool +call split across two fetched pages, rejoins a message a boundary cut +through, and renames a run of tool calls onto whichever name is already on +screen. Ported with AGENTS.md's "things that have bitten" incidents as the +spec rather than a JVM test file (`TranscriptItems.kt` had none of its +own): `a_clean_boundary_between_two_finished_runs_is_still_healed_into_one_run` +is the regression test for the bug that shipped -- `adopt_run` must run on +*every* join, not only the one where a split call was found, or a boundary +landing cleanly between two already-finished calls (most of them) leaves +one run drawn as two. `a_call_split_across_the_boundary_merges_into_one_row`, +`a_message_split_across_the_boundary_is_rejoined_with_the_newer_halfs_identity`, +and `adopt_run_never_renames_into_a_question_row` cover the other three +edges the Kotlin doc calls out. `join_pages` ends in a `debug_assert!` +that no tool id survives in both halves -- the duplicate row it exists to +prevent, checked rather than assumed. What it deliberately does *not* +assert is seq ordering across the boundary: a peer note carries the seq +its turn began at (`place_peer_note`), which can be older than the page +it arrived in, so the two pages' seqs legitimately interleave there. An +earlier draft asserted it and would have panicked in debug builds on an +ordinary transcript. **Known gap, and a decision for whoever closes it:** `event_model::Event` has no `Unknown`/catch-all variant, unlike `Events.kt`'s hand-kept mirror. @@ -119,12 +136,61 @@ caller-specific (the code rules' "ask for the least you need"). Its only caller today is `desktop-app`; a future Android build of this crate would be a second one, not a reason to move the type. +## What `transcript_source.rs` covers, and what it does not + +`TranscriptSource` is the seam a session screen asks for a +page, ported test-for-test against the Kotlin doc rather than a JVM test +file (there wasn't one): `cached_opening`, `probe`, `fetch_opening`, +`page` and `follow`, each matching its Kotlin namesake's contract -- +including `probe`'s three-way outcome (matches / cache purged / +unreachable, told apart so a caller never treats "couldn't ask" as "was +wrong") and `page`'s cache-vs-server split bounded by `covered_up_to`. + +Two additions beyond a literal port, both load-bearing: + +- **`page(before, ..)` refuses `before == 0` before touching the cache or + the network**, answering `OlderPage::NothingLoaded`. This is AGENTS.md's + `loadOlderPage` incident (`before = 0` is "no event before the first + one," indistinguishable from "reached the start of history" if a caller + ever asks it) moved out of the Kotlin screen and into this layer, so + every future caller gets the guard rather than having to remember it. + **The return type is `OlderPage`, not a `Vec`, and that is the guard.** + The Kotlin's two falses are different answers -- `oldestSeq == 0` + returns without touching `moreHistory`, an empty page latches it false + -- so a port that answered both with an empty list would have moved the + bug rather than fixed it, one layer down and out of sight of the screen + that used to hold the check. `OlderPage::Events(vec![])` means the start + of the conversation; `OlderPage::NothingLoaded` is not an answer about + the conversation at all. Reviewed 2026-09-06. + `paging_before_the_first_event_makes_no_request_at_all` asserts zero + transport calls, not just the variant, since a request that happens to + answer empty is exactly what caused the original bug, and + `a_failing_server_page_is_an_error_rather_than_an_empty_one` plus + `an_unreadable_cached_page_is_a_parse_error_rather_than_an_empty_one` + are the same rule for the two ways a page can fail. +- **`fetch_transcript_lines`** (new in `api.rs`) hands back each line + paired with the exact server bytes it came from, via + `serde_json::value::RawValue` rather than re-serializing a parsed + `Value` -- the cache and a live SSE frame for the same event have to + agree byte-for-byte, which is exactly what the `serde_json` + float-rounding bug (AGENTS.md) was about. The existing + `fetch_transcript_page` is untouched (other callers under `iris/` + depend on its signature); the two share a `transcript_path` helper so + the query string is written in one place. + +**Not ported:** `EventStream.kt`'s reconnect-with-backoff loop, and +`TranscriptSource.close`'s ability to cancel a live stream from another +thread. Both are wall-clock/thread-lifetime policy that belongs to +whichever runtime embeds this crate (iris's own timers, a Tokio task, a +Kotlin coroutine scope), not to this pure logic -- `follow` is the same +"write to the cache, then hand the frame to the caller" decorator +`iris/desktop-app/src/app.rs` and `iris/android-app/src/transcript_client.rs` +already hand-wrote around `event_stream::follow_session_events` before this +existed; the cache write moved into one shared place so a third caller +does not repeat it again by hand. + ## What is not started at all -- **`TranscriptSource.kt`** -- the layer that decides whether a page comes - from the transcript cache or the server, and stitches the two. Needs - `transcript_cache.rs` and `api.rs`'s transcript-page method, both of - which exist now, so this is unblocked whenever picked up. - **The markdown *block* model beyond syntax spans** -- `highlight/markdown.rs` colours a `.md` file or fence for the highlighter, but does not build the block tree (headings, lists, tables, fences as distinct nodes) that a @@ -142,5 +208,6 @@ be a second one, not a reason to move the type. `./run-tests.sh` from the repo root now runs `event-model`, `client-core` and `server` in that order (each `cargo test`, forwarding arguments the -same way it always has). From `client-core/` directly: `cargo test`, -`cargo clippy --all-targets`, `cargo fmt` -- all clean as of this writing. +same way it always has). From `client-core/` directly: `cargo test` +(109 tests), `cargo clippy --all-targets`, `cargo fmt` -- all clean as of +this writing (2026-09-06). diff --git a/docs/RUST.md b/docs/RUST.md index c3c5d9b..986acd8 100644 --- a/docs/RUST.md +++ b/docs/RUST.md @@ -177,12 +177,32 @@ closes it. touch pan from the same mechanism `List` uses, not a copy. - [ ] **Streaming re-layout** (IRIS_TODO.md's last section) — after the above, since they make the stream phase unrepresentative today. -- [ ] **client-core prerequisites for P1, in parallel** (pure Rust, - disjoint from `iris/`): `TranscriptSource`'s cache-vs-server - stitching and `joinPages`/`healSplitMessage`/`adoptRun` page-boundary - healing, per `CLIENT_CORE.md`. Ported with the Kotlin tests as the - spec. Cheap to have ready if P0 passes, and no rendering risk if - it does not. +- [x] **client-core prerequisites for P1, in parallel** (pure Rust, + disjoint from `iris/`), closed 2026-09-06: `TranscriptSource`'s + cache-vs-server stitching (new `client-core/src/transcript_source.rs`) + and `joinPages`/`healSplitMessage`/`adoptRun` page-boundary healing + (new functions in `transcript_fold.rs`), per `CLIENT_CORE.md`. Ported + against the Kotlin source and AGENTS.md's paging incidents as the + spec (`TranscriptSource.kt`/`TranscriptItems.kt` had no JVM unit + tests of their own to port test-for-test). `client-core` goes from + 85 to 109 tests; `cargo test`/`clippy --all-targets`/`fmt` all clean. + Both AGENTS.md regressions have a dedicated test: `loadOlderPage`'s + `before == 0` guard moved into `TranscriptSource::page` itself + (`paging_before_the_first_event_makes_no_request_at_all` asserts + zero transport calls, not just an empty result), and + `a_clean_boundary_between_two_finished_runs_is_still_healed_into_one_run` + pins `adopt_run` running on *every* join rather than only the + split-call path. One incidental fix needed to port `TranscriptSource` + faithfully: `api.rs` gained `fetch_transcript_lines` (additive, the + existing `fetch_transcript_page` untouched since `iris/` depends on + its signature), which pairs each event with the exact server bytes + it came from via `serde_json::value::RawValue` rather than + re-serializing a parsed `Value` -- needed so the cache and a live SSE + frame agree byte-for-byte, the same class of bug as the + `float_roundtrip` fix. Deliberately not ported: `EventStream.kt`'s + reconnect/backoff and cross-thread stream cancellation, which are + runtime policy for whichever framework embeds this crate, not pure + logic -- see `CLIENT_CORE.md`'s new section for the full account. - **Then**: redeliver `~/host/bench/iris-bench-arm64.apk` for Iris with its README saying what changed, and record any choice she should see in `DECISIONS.md`.