diff --git a/AGENTS.md b/AGENTS.md index 75d0192..4769ecf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -390,6 +390,11 @@ first if a remote spawn ever mangles an argument. `./ui-sandbox.sh spawn [title]` (an echo session, prints its id), `./ui-sandbox.sh send SID text|@file`, and `./ui-sandbox.sh api /path [curl args]` for everything else. + `./ui-sandbox.sh keep` restarts the server without wiping the sessions and + enrolment already there -- for when the fixture under test was expensive to + build (a long delta-heavy transcript, say) and should survive a rebuild of + the server binary; plain `start` wipes them, which is right for the + list-screen fixtures and wrong for that. It passes `--delay` by default for the reason the next entry gives, and `AI_SANDBOX_BIG_MB` puts one large transcript among the small ones -- `AI_SANDBOX_SPAWN_DELAY` makes the fake CLI slow to start. Both exist diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt index 0f8997f..b0584c5 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -744,10 +744,17 @@ fun fetchTranscript( sessionId: String, before: Long? = null, limit: Int = 80, + // Count [limit] in rows, not events, joining a reply's streamed deltas into one -- so a page + // of a delta-heavy conversation is a page of the screen rather than a fraction of one message. + // The scroll-back pager wants this; the anchor restore does not (it counts events to a known + // seq). Ignored by the server for the newest window, where the live cursor needs real seqs. + // See the server's `read_window`. + coalesce: Boolean = false, ): List { val query = buildString { append("?limit=").append(limit) if (before != null) append("&before=").append(before) + if (coalesce) append("&coalesce=true") } return requestFromServer(settings, "/sessions/$sessionId/transcript$query") { connection -> val body = JSONArray(connection.inputStream.bufferedReader().readText()) diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index d5c4a3c..2ed8c15 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -114,26 +114,25 @@ private val LOADING_SPINNER = 48.dp private const val HISTORY_SCREENS = 6 /** - * How many events a backwards page asks for, which is five times what the opening page takes. + * How many *rows* a backwards page asks for. * - * The floor is that an event is not a row, and the ratio is nothing like one to one. Measured on a - * real transcript (2,426 events, 2026-08-30): the whole conversation is *seven* assistant messages, - * and the median run of consecutive text deltas that fold into one of them is four hundred. A page - * of eighty is therefore a fifth of a single row, and reaching a screenful of fresh rows took about - * thirty sequential round trips inside one collect. Below this number a page can add no visible - * room at all, and the fetch chain degenerates into those round trips again. + * Rows, not events, because the two are nothing alike: a reply is stored a token at a time, and on + * one real transcript (2,426 events, 2026-08-30) the whole conversation was seven assistant + * messages, the median folding four hundred deltas into one row. A page counted in events was a + * fifth of a single row, so reaching a screenful took dozens of sequential round trips and the + * reader stood at the boundary through every one. The server now joins each delta run into the one + * event the fold makes of it (`fetchTranscript(coalesce = true)`), so a page of rows is a page of + * the screen whatever the delta density. * - * At the floor rather than above it, because pages are fetched in the background before the reader - * arrives -- the cushion decides how deep loading runs, and a page that was not enough is followed - * by another without anybody waiting on either. What a *smaller* page buys is hiding: it crosses - * the tunnel in half the time and lands in a smaller frame spike, so the case where the reader - * outruns an in-flight fetch is rarer and cheaper. This was 800 when the reader was the one - * standing at the boundary and each round trip had to be amortized as far as it would go. + * A few screens' worth, so one page clears the cushion below and the reader reaches loaded content + * without a fetch in the way. A page that still falls short is followed by another in the + * background, nobody waiting on either. * - * The opening page stays small: it is the one on the critical path of showing the screen at all, - * and it only has to fill a viewport. + * The opening page stays counted in events and small ([fetchTranscript]'s default): it is on the + * critical path of showing the screen, only has to fill a viewport, and is the newest window, where + * coalescing is unsafe for the live cursor anyway. */ -private const val HISTORY_PAGE = 400 +private const val HISTORY_PAGE = 40 /** * The most events one request of a restore may ask for. @@ -149,6 +148,16 @@ private const val HISTORY_PAGE = 400 */ private const val RESTORE_PAGE_MAX = 4000 +/** + * Events added past the anchor on a restore, so the anchor's row is never the oldest loaded one. + * + * The oldest loaded row is a half-row -- [joinPages] welds its other half on when the page behind + * it arrives, and it grows -- so a restore that stopped exactly at the anchor would put the reader + * a screen out once that growth landed. This is in events, like the rest of the restore span: that + * path counts events to reach a known seq and does not coalesce. + */ +private const val RESTORE_PAGE_CUSHION = 400 + /** * Which row was asked to hold its top edge, and how tall it was when it last measured. * @@ -523,7 +532,7 @@ fun SessionScreen(settings: ServerSettings, summary: SessionSummary, onBack: () * Reads `items` rather than `rows`: this runs in a coroutine, and `rows` is the composition's * value, which does not change under a running one. */ - suspend fun loadOlderPage(limit: Int = HISTORY_PAGE): Boolean { + suspend fun loadOlderPage(limit: Int = HISTORY_PAGE, coalesce: Boolean = true): Boolean { // Nothing is loaded, so there is no "before" to ask about, and asking anyway is not a // harmless no-op: `before = 0` fetches the events before the first one, which is none, // and an empty page is how this function is told it has reached the start of the @@ -556,7 +565,14 @@ fun SessionScreen(settings: ServerSettings, summary: SessionSummary, onBack: () // `items` read below happens back on the caller's thread, where the write does too. val page = withContext(Dispatchers.IO) { - val older = fetchTranscript(settings, summary.id, before = oldestSeq, limit = limit) + val older = + fetchTranscript( + settings, + summary.id, + before = oldestSeq, + limit = limit, + coalesce = coalesce, + ) if (older.isEmpty()) return@withContext null // Folded oldest-first into a list of their own, then put in front: `foldEvent` // merges streaming text into the item before it, so replaying an older page @@ -681,8 +697,17 @@ fun SessionScreen(settings: ServerSettings, summary: SessionSummary, onBack: () // somebody had read a little way back into, and a spinner for all of them. // The bytes are the same either way, since every row between the anchor and // the newest end has to be there for the list to be able to count to it. - val span = oldestSeq - anchor.seq + HISTORY_PAGE - if (!loadOlderPage(span.coerceIn(1L, RESTORE_PAGE_MAX.toLong()).toInt())) break + // Raw, not coalesced: this counts events back to a known seq, and a page + // measured in rows cannot be counted to a seq. `RESTORE_PAGE_MAX` and the loop + // bound it; see [loadOlderPage]. + val span = oldestSeq - anchor.seq + RESTORE_PAGE_CUSHION + if ( + !loadOlderPage( + span.coerceIn(1L, RESTORE_PAGE_MAX.toLong()).toInt(), + coalesce = false, + ) + ) + break } // Resolved to the row that *holds* the saved position rather than passed // straight through, because the two are not always the same seq: the events diff --git a/app/ui-sandbox.sh b/app/ui-sandbox.sh index d307a6e..6382e6f 100755 --- a/app/ui-sandbox.sh +++ b/app/ui-sandbox.sh @@ -127,8 +127,14 @@ print(json.dumps({"text": text}))' "$@" >"$ROOT/send.json" exit 0 ;; start) ;; +# Restart the server but keep the sessions and enrolment already there, so a +# fixture built over minutes (a long delta-heavy transcript, say) survives a +# rebuild of the server binary. Plain `start` wipes them, which is right for +# the list-screen fixtures but wrong when the session under test was expensive +# to make. +keep) KEEP=1 ;; *) - echo "ui-sandbox.sh: unknown command '$1' (start, stop, api, spawn, send)" >&2 + echo "ui-sandbox.sh: unknown command '$1' (start, keep, stop, api, spawn, send)" >&2 exit 2 ;; esac @@ -152,9 +158,13 @@ if [ -f "$ROOT/config.ron" ]; then } END { printf "%s", entries }' h="$hash" "$ROOT/config.ron") fi -rm -rf "$ROOT/home" "$ROOT/sessions" "$ROOT/config.ron" PROJECTS=$ROOT/home/.claude/projects/-home-bob-repos-sandbox +if [ -z "${KEEP:-}" ]; then + rm -rf "$ROOT/home" "$ROOT/sessions" +fi +rm -f "$ROOT/config.ron" mkdir -p "$PROJECTS" "$ROOT/sessions" +if [ -z "${KEEP:-}" ]; then # Eight of them, because the point of the screen is a list long enough that # picking rows one at a time is the annoyance being fixed. Ids are the same @@ -219,6 +229,7 @@ awk -v mb="$BIG_MB" 'BEGIN { } print "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"usage\":{\"input_tokens\":180000,\"output_tokens\":900}}}" }' > "$big" +fi cat >"$ROOT/config.ron" <, #[serde(default = "default_window")] limit: usize, + /// Join each reply's streamed deltas into one event, so a page counts rows rather than + /// tokens. The scroll-back pager asks for this; the anchor-restore path does not, because it + /// counts events to reach a known seq. See `read_window`. + #[serde(default)] + coalesce: bool, } fn default_window() -> usize { @@ -1370,6 +1375,7 @@ async fn transcript( session.transcript_path(), query.before, query.limit, + query.coalesce, ) .map_err(bad_request)?; // How far back a phone has paged, and how much each page cost it to get diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index fdd32f8..1efc2f4 100644 --- a/server/src/session/transcript.rs +++ b/server/src/session/transcript.rs @@ -148,7 +148,12 @@ impl Transcript { /// `before` pages backwards for history somebody actually scrolls to. Only /// the window is parsed; see [`Indexed`] for why that is the whole cost of /// this call. -pub fn read_window(path: &Path, before: Option, limit: usize) -> Result> { +pub fn read_window( + path: &Path, + before: Option, + limit: usize, + coalesce: bool, +) -> Result> { let Some(indexed) = Indexed::read(path)? else { return Ok(Vec::new()); }; @@ -156,7 +161,16 @@ pub fn read_window(path: &Path, before: Option, limit: usize) -> Result indexed.first_at_or_after(before)?, None => indexed.lines.len(), }; - indexed.parse(end.saturating_sub(limit)..end) + // Coalescing counts *rows*, not events, and would misread the newest window: a message still + // streaming there would fold to one event whose seq is its first delta, and the phone resumes + // its live stream from the newest seq it applied -- so the deltas the coalesced event hid + // would replay and double. Only settled history (`before` set) is safe, and it is the only + // place the phone asks for it. See `parse_coalesced`. + if coalesce && before.is_some() { + indexed.parse_coalesced(end, limit) + } else { + indexed.parse(end.saturating_sub(limit)..end) + } } /// How far behind a reconnecting subscriber can be and still be handed the @@ -315,6 +329,74 @@ impl<'a> Indexed<'a> { }) .collect() } + + fn parse_one(&self, index: usize) -> Result { + serde_json::from_str(&self.text[self.lines[index].clone()]) + .with_context(|| format!("bad transcript line in {}", self.path.display())) + } + + /// The newest `limit` *rows* ending at line `end`, with each run of consecutive streamed + /// [`Event::AssistantText`] deltas concatenated into one. + /// + /// A reply is stored a token at a time -- hundreds of `AssistantText` events for one message -- + /// so a window counted in events is a fraction of a row for a reply and a whole row for a tool + /// call, and the phone can neither predict how much a page will show nor fill a screen without + /// folding a page's worth of near-duplicate events. Counted in rows, a page is a page: this + /// walks back from `end`, joining each delta run into the single event the phone would fold it + /// into anyway, and stops once `limit` of them are gathered. + /// + /// A run takes the seq and time of its *oldest* delta, matching the phone's own rule that a + /// streamed message keeps the seq of its first delta -- so anchors, and the `before` cursor the + /// next page pages from, land where they always did. A run cut by the `limit` (its older + /// deltas beyond this page) is emitted as the partial it is; the next page carries the rest and + /// the phone's `healSplitMessage` welds the two, exactly as it does for a run cut by any page + /// boundary. + fn parse_coalesced(&self, end: usize, limit: usize) -> Result> { + // Newest first while walking back, reversed to transcript order at the end. + let mut out: Vec = Vec::new(); + // The run currently being gathered: its oldest seq/ts so far, and its deltas newest-first. + let mut run: Option<(u64, f64, Vec)> = None; + let flush = |run: &mut Option<(u64, f64, Vec)>, out: &mut Vec| { + if let Some((seq, ts, mut deltas)) = run.take() { + deltas.reverse(); + out.push(SeqEvent { + seq, + ts, + event: Event::AssistantText { + delta: deltas.concat(), + }, + }); + } + }; + let mut index = end; + while index > 0 { + // A row is counted when it lands in `out`; an open run is the row being gathered, so + // stopping while one is open would drop the deltas already read. Break only between + // rows, and flush the last run after the loop. + if out.len() >= limit && run.is_none() { + break; + } + index -= 1; + let entry = self.parse_one(index)?; + if let Event::AssistantText { delta } = entry.event { + match run { + Some((ref mut seq, ref mut ts, ref mut deltas)) => { + *seq = entry.seq; + *ts = entry.ts; + deltas.push(delta); + } + None => run = Some((entry.seq, entry.ts, vec![delta])), + } + } else { + // The run above this event (newer) is complete: it is a row, and so is this event. + flush(&mut run, &mut out); + out.push(entry); + } + } + flush(&mut run, &mut out); + out.reverse(); + Ok(out) + } } #[cfg(test)] @@ -446,7 +528,7 @@ mod tests { } // No cursor is the newest page, which is what opening a session asks for. - let newest = read_window(&path, None, 3).expect("window"); + let newest = read_window(&path, None, 3, false).expect("window"); assert_eq!( newest.iter().map(|entry| entry.seq).collect::>(), [8, 9, 10] @@ -454,25 +536,77 @@ mod tests { // Then backwards from the oldest of those, exclusive: the page a phone // scrolling up asks for must not repeat the row it is scrolling from. - let older = read_window(&path, Some(8), 3).expect("window"); + let older = read_window(&path, Some(8), 3, false).expect("window"); assert_eq!( older.iter().map(|entry| entry.seq).collect::>(), [5, 6, 7] ); // Asking for more than there is gives what there is, rather than failing. - assert_eq!(read_window(&path, None, 100).expect("window").len(), 10); + assert_eq!(read_window(&path, None, 100, false).expect("window").len(), 10); // Nothing before the first event, which is how the phone learns to stop // paging. An empty answer here is the end of the history, not a fault. - assert!(read_window(&path, Some(1), 3).expect("window").is_empty()); + assert!(read_window(&path, Some(1), 3, false).expect("window").is_empty()); assert!( - read_window(&dir.path().join("nope.jsonl"), None, 3) + read_window(&dir.path().join("nope.jsonl"), None, 3, false) .expect("window") .is_empty() ); } + #[test] + fn coalescing_counts_rows_and_joins_delta_runs() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("transcript.jsonl"); + let mut transcript = Transcript::open(&path).expect("open"); + // Two replies of three deltas each, split by a tool call: the shape a turn writes, and + // the one an event-counted page cannot see a row of. + for d in ["a", "b", "c"] { + transcript.append(text(d), 0.0).expect("append"); // seq 1..3 + } + transcript + .append( + Event::ToolStart { + id: "t".into(), + tool: "Bash".into(), + input: serde_json::Value::Null, + }, + 0.0, + ) + .expect("append"); // seq 4 + for d in ["d", "e", "f"] { + transcript.append(text(d), 0.0).expect("append"); // seq 5..7 + } + + // Three rows asked for, three rows returned -- each delta run one event -- where a raw + // window of three would have shown one and a half tokens of the newer reply. + let rows = read_window(&path, Some(8), 3, true).expect("window"); + assert_eq!(rows.len(), 3); + // A run keeps its oldest delta's seq, so the phone anchors and pages from where it always + // did. + assert!(matches!( + &rows[0], + SeqEvent { seq: 1, event: Event::AssistantText { delta }, .. } if delta == "abc" + )); + assert!(matches!(&rows[1], SeqEvent { seq: 4, event: Event::ToolStart { .. }, .. })); + assert!(matches!( + &rows[2], + SeqEvent { seq: 5, event: Event::AssistantText { delta }, .. } if delta == "def" + )); + + // The next page pages from the oldest row's seq and returns the rest, no repeat, no gap. + let older = read_window(&path, Some(1), 3, true).expect("window"); + assert!(older.is_empty()); + + // The newest window never coalesces even when asked: the live cursor depends on real seqs. + let newest = read_window(&path, None, 2, true).expect("window"); + assert_eq!( + newest.iter().map(|e| e.seq).collect::>(), + [6, 7] + ); + } + #[test] fn a_missing_file_reads_as_empty() { let dir = tempfile::tempdir().expect("tempdir");