iris/ai-app
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Page history by rows, coalescing a reply's deltas server-side

Browse Source 
The scroll-up freeze-then-skip was the history pager fighting the transcript's
own storage. A streamed reply is stored one token per event -- hundreds of
AssistantText events for one message -- but a page was counted in events, so
on a delta-heavy conversation a page was a fraction of one row: the opening
80-event load was less than a screen, "scroll up a bit" hit the unloaded
boundary at once, and each page the client did fetch cost a 400-event fold
(hundreds of thousands of list copies) that landed as one jarring insertion.

The server now joins each run of consecutive AssistantText deltas into the one
event the client's fold makes of it, and counts a page's limit in these
coalesced rows -- so a page is a page of the screen whatever the delta density.
Measured against a 3,500-event / 100-row echo session on the emulator: a raw
limit-20 page returns 20 tokens of one reply; the coalesced limit-20 returns
five whole replies. Scrolling the whole thing showed waited p99 51.7ms -> 0.6ms
and the worst whole-transcript measure 59ms -> ~0, with the client folding ~100
row-events instead of 3,500 token-events. No duplicate keys; the first reply
still reconstructs whole from token zero, so healSplitMessage welds the raw
newest window to the coalesced older pages exactly as before.

Coalescing is opt-in per request (`?coalesce=true`) and applied only to older
pages (`before` set): the newest window keeps real seqs because the live stream
resumes from the newest seq the phone applied, and a coalesced newest event
would hide the deltas after its first seq and replay them. The anchor-restore
path also stays raw -- it counts events to reach a known seq, which a page
measured in rows cannot do -- so HISTORY_PAGE is now rows while the restore
span and its cushion stay in events.

Also: ui-sandbox.sh gains a `keep` verb that restarts the server without wiping
sessions, so a fixture that costs minutes to build (a long delta-heavy
transcript) survives a server rebuild.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
irisandClaude Fable 5 committed 2026-09-01 21:43:18 -04:00
1 parent 356dee65f1
commit 63c0bb9e4a
6 files changed
+220 -29
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/session/transcript.rs
+141 -7
View File
@@ -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<u64>, limit: usize) -> Result<Vec<SeqEvent>> {
pub fn read_window(
path: &Path,
before: Option<u64>,
limit: usize,
coalesce: bool,
) -> Result<Vec<SeqEvent>> {
let Some(indexed) = Indexed::read(path)? else {
return Ok(Vec::new());
};
@@ -156,7 +161,16 @@ pub fn read_window(path: &Path, before: Option<u64>, limit: usize) -> Result<Vec
Some(before) => 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<SeqEvent> {
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<Vec<SeqEvent>> {
// Newest first while walking back, reversed to transcript order at the end.
let mut out: Vec<SeqEvent> = 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<String>)> = None;
let flush = |run: &mut Option<(u64, f64, Vec<String>)>, out: &mut Vec<SeqEvent>| {
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::<Vec<_>>(),
[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::<Vec<_>>(),
[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::<Vec<_>>(),
[6, 7]
);
}
#[test]
fn a_missing_file_reads_as_empty() {
let dir = tempfile::tempdir().expect("tempdir");