diff --git a/AGENTS.md b/AGENTS.md index 718aa0d..e7f3962 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -829,6 +829,54 @@ machine belongs in `~/.claude/TOOLCHAIN.md` (toolchain versions) or every page. Both are off it now. The shape to watch for is a `withContext` that wraps the *fetch* and leaves the work done with the result outside it. +- **The phone keeps the transcripts it has been sent, and the design is + `TRANSCRIPT_CACHE.md`** -- read that before touching `TranscriptCache.kt`, + `TranscriptSource.kt`, or the opening and stream effects in + `SessionScreen.kt`. What the day-to-day work needs to know: + `/transcripts/v1/_//` holds the server's + own event lines in chunks named for the range they cover + (`-.rows.jsonl`, `.raw.jsonl`, and one `-open.raw.jsonl` + the live stream appends to), and only the contiguous run ending at the + newest chunk is ever served. Measured on the emulator 2026-09-04 against + the sandbox: reopening a 500-event session costs **one request for one + event** -- the probe -- and scrolling the whole conversation back costs + nothing more. A cold open of the same session is two pages, 100 events. + Four things are easy to undo by accident. + **The probe is not optional**: before a stream is resumed from a cached + cursor, `GET /transcript?before=&limit=1` has to come back as the + line the cache holds, or the cache is thrown away and the open is cold. It + is what stops a replaced or truncated file being spliced onto this phone's + copy of a different conversation, with no seam to see. + **A page fetched for the gap passes `after`** (the transcript route's own + parameter, added for this), so it stops where the phone's copy starts. A + page that overlaps a chunk cannot be stored -- a coalesced event has no + clean cut inside its delta run -- so without the bound the first scroll + back after a reset throws away everything behind it. + **Nothing here is load-bearing.** Every read has a network path beside it + giving the same answer, and a missing, evicted, damaged or unwritable cache + degrades to a cold open. Keep it that way: a cache that can blank the + screen is worse than no cache. + **Reload, in session settings, is the answer to what the probe cannot + see** -- a line changed in the middle of the file with the tail intact. + It purges and rebuilds the screen as a cold open, putting the reader back + where they were. + Exercise all of it with `./ui-sandbox.sh` and + `RUST_LOG=ai_server=debug`, which logs every page with its `before`, + `after` and what came back; `adb shell run-as com.example.aiapp ls + cache/transcripts/v1/*/` shows whether the chunk names are adjacent, + which is the one thing the screen cannot tell you. +- **The server used to hand out the same transcript line two different + ways.** `serde_json`'s default float parser is not correctly rounded, so a + `ts` of `1788546972.6030757` in the transcript came back from + `/transcript` as `...0755` while the SSE stream, serializing the same + struct, sent the original. Nothing on screen could show it -- a `ts` is + drawn as a relative time -- and what found it was the phone's cache + comparing a line it held against the server's own answer, which turned an + invisible last-bit difference into a cache silently thrown away and a + transcript downloaded again. The `float_roundtrip` feature in + `server/Cargo.toml` is the fix and + `a_line_read_back_is_the_line_that_was_written` is what keeps it; that test + fails within a second of the feature being dropped. - **ZXing only looks for a dark code on a light ground.** The enrollment QR is block characters in the terminal's foreground colour, so a dark-themed terminal renders it as a negative and the in-app scanner diff --git a/PLAN.md b/PLAN.md index f83f603..ab7427d 100644 --- a/PLAN.md +++ b/PLAN.md @@ -1016,9 +1016,16 @@ dev-updater (Kotlin 2.4.x, CMP 1.11.x, JDK 21). Screens: Networking mirrors dev-updater's app layer (`AppsApi.kt` style thin client + pinned transport), plus an SSE client with `after=` resume driven by -connectivity/lifecycle. The app keeps no persistent transcript store — the -backend's transcript is the source of truth; the app caches only for the -screen it's showing. +connectivity/lifecycle. The backend's transcript is the source of truth, and +the app keeps a **copy of what it has already been sent** — see +`TRANSCRIPT_CACHE.md`, added 2026-09-04, because reopening a session over the +tunnel was re-downloading a conversation the phone had just read. The copy is +the server's own event lines, per session, under `cacheDir`; it is checked +against the server before a stream is resumed from it, thrown away rather than +patched when that check fails, and never load-bearing — every path that reads +it has a network path beside it giving the same answer. What the app still +does not keep is anything *derived*: the folded rows are rebuilt from events +every time. ### Notifications: two places, never both (decided 2026-08-30) diff --git a/TRANSCRIPT_CACHE.md b/TRANSCRIPT_CACHE.md new file mode 100644 index 0000000..5ab6cb9 --- /dev/null +++ b/TRANSCRIPT_CACHE.md @@ -0,0 +1,510 @@ +# The transcript cache + +Asked for by Iris on 2026-09-04: keep the transcripts of recently visited +sessions on the phone, so reopening one does not download it again. It has +to save data over the tunnel, it must not disturb a reply that is streaming +when the screen is reopened, it must never skip an event, and session +settings needs a manual reload for when the file on the machine has +changed under it. + +Built 2026-09-04. Like EXPLORER.md this records each decision with its +reason and what was rejected, so that when one changes it is changed here +rather than re-argued -- three of them changed during the building, and +"What building it changed" at the foot says which and why. What it is *not* +is the operational half: how to exercise it, and what has bitten, are in +AGENTS.md with the rest of the working notes. + +## What it is, in one paragraph + +A per-session file on the phone holding the exact JSON lines the server has +already sent, in transcript order, with a record of which sequence numbers +each run of lines covers. Everything the session screen fetches today -- +the opening window, the pages it scrolls back through, the span an anchor +restore reaches for -- is asked of the cache first and of the server only +for what the cache does not hold, and everything that arrives from the +server is written into it. The live stream then resumes from the newest +cached event, exactly as it resumes today from the newest event on screen, +so the server sends only what happened since. One tiny request checks that +the cached tail is still what the server has before the stream is opened +from it, and a button in session settings throws the cache away and +rebuilds the screen as a cold open for the cases that check cannot see. + +## The invariants + +Everything below is in service of four rules. When a decision looks +arbitrary, it is one of these forcing it. + +1. **What is on screen is what the server's transcript says, in order, + with nothing missing, for every sequence number the screen claims to + show.** The cache is a copy of server output and is never inferred, + folded, or edited on the phone. Where the copy cannot be shown to be + current, it is thrown away, not patched. +2. **A cached line is never ahead of the live cursor, and the live cursor + is never ahead of the cache.** The stream resumes from the newest cached + event, so a reply that was mid-stream when the screen closed picks up at + its next delta and folds into the same row, as it does today when the + phone merely lost the tunnel for a second. +3. **The cache is never load-bearing.** A missing, evicted, corrupt or + unwritable cache degrades to today's behaviour -- a cold open -- and + never to a blank or wrong screen. Every path that reads it has a + network path beside it that produces the same result. +4. **Data crosses the tunnel once.** A line already on the phone is not + fetched again unless the reader asks for that (the reload button) or the + check in decision 3 says it must be. + +## Decisions + +### 1. Raw server lines, on the phone, keyed by server and session + +The cache stores the server's own JSON, one event per line, byte-for-byte +as it arrived: the elements of the `/transcript` array and the `data:` +payload of each SSE frame. Reading the cache means running the same +`parseSeqEvent` the network path runs, so a cached transcript and a fetched +one cannot draw differently, and an event type this build does not know +(`SessionEvent.Unknown`) survives on disk for the build that will. + +It lives under `context.cacheDir` -- `/transcripts/v1/_//` +-- because it is exactly what that directory is for: bytes the phone can +regenerate from the server, which Android may delete under storage +pressure without asking. Keyed by the server's host and port because two +servers can hold a session with the same id (the sandbox and the real +server, or a re-enrolment), and a line from one shown against the other +is invariant 1 broken. `ServerSettings` has both fields; the key is +`"${settings.host}_${settings.port}"` with `:` never appearing in it. +The `v1` segment is the format version: any change to the layout below +bumps it, and a directory of another version is deleted on first use. + +Rejected: a database (Room, SQLite). The access pattern is "the newest N +lines" and "the lines before seq X", on files of tens of megabytes at most, +and a JSONL file per contiguous run answers both by reading from its end. +A database would be a new dependency for an index the file layout already +provides. + +Rejected: caching folded `TranscriptItem` rows instead of events. Rows are +a *rendering* of events, and their shape changes when the fold changes; +the cache would need invalidating on every app update that touched +`foldEvent`, and would still have to keep raw seqs for the stream cursor. +Events are the server's contract and the only thing that is stable. + +### 2. Chunks with explicit coverage; one contiguous run behind the cursor + +A page from the server is a set of lines *and a claim about what they +cover*, and the two are not the same thing. A coalesced page +(`coalesce=true`, which the scroll-back pager asks for) joins each run of +`assistantText` deltas into one event carrying the seq of its *oldest* +delta, so a page whose newest event has seq 1,200 may in fact cover every +line up to the `before` it was asked with, say 1,650. Nothing in the lines +themselves says so. So each stored chunk records its coverage as a +half-open range `[first, end)`, where `first` is the seq of its oldest +event and `end` is the `before` the request was made with -- or, for a +raw chunk, its newest seq plus one. + +Chunks are files named by their coverage: + + -.rows.jsonl a coalesced page; end is the `before` it was fetched with + -.raw.jsonl an uncoalesced page or a closed live run + -open.raw.jsonl the live run: appended to by the stream; end = last line's seq + 1 + +Two chunks are **adjacent** when one's `end` equals the other's `first`. +The cache serves only the contiguous run of adjacent chunks that ends at +the newest raw chunk (the **suffix**); chunks behind a gap are kept on +disk, because the gap is usually filled (decision 4), but are never served +across the gap. + +**The newest chunk is always raw.** That is what makes the stream cursor +and the check in decision 3 well defined: a raw chunk's last line is a real +event at a real seq, and the server never coalesces the newest window +("the live cursor depends on real seqs", `read_window`). It holds by +construction -- the opening window is fetched with no `before`, stream +frames are raw, and a `reset` window is raw -- and is *checked* on read: +if the newest chunk on disk is a `.rows` chunk (which can only happen if +the app died between closing one live run and appending to the next), the +session's cache is purged and the open is cold. + +There is at most one open chunk. When a stream event arrives whose seq is +not the open chunk's `end` -- which is what a `reset` looks like from +here, see decision 6 -- the open chunk is closed by renaming it with its +real end, and a new open chunk starts at the arriving seq. An event whose +seq is below the open chunk's `end` is already covered and is not written +(the SSE contract is `seq > after`, so this is a guard, not a path). + +Rejected: one file per session, rewritten to prepend older pages. A +20 MB transcript would be rewritten on every page scrolled back to. The +chunk directory costs a directory listing per open instead. + +Rejected: trimming chunks to resolve overlaps. A coalesced event cannot be +split at a seq inside its run, so an overlap between a coalesced page and +an existing chunk has no clean cut. The cache therefore **never stores a +page that overlaps an existing chunk**; decision 4 makes sure such a page +is never fetched in the first place, and if one arrives anyway (a server +without decision 4's change) it is used for display and not stored. + +### 3. The cached tail is checked against the server before the stream opens from it + +The screen must not resume a stream from a cached seq unless the server's +event at that seq is the one in the cache. The transcript file on the +machine is append-only in ordinary use, but it can be replaced or +truncated -- a sandbox re-seeded with the same ids, a backup restored, a +directory deleted and the session re-imported under the same name -- and +`catch_up` on such a file would hand the phone a continuation of a +different conversation, spliced onto the cached one with no seam. That is +the worst thing this feature can do, and it is caught with one request. + +**The probe:** `GET /sessions/{id}/transcript?before=&limit=1`, +where `cursor` is the seq of the cache's newest line. `read_window` with +that `before` returns the single newest event with seq ≤ cursor, which is +the event *at* the cursor when it exists. The probe passes when that +response, parsed with `parseSeqEvent`, is `==` to the cached line parsed +the same way -- data-class equality over seq, ts, and the whole event. It +fails when the response is empty, is a different seq, or differs in any +field. + +That equality rested on an assumption this plan stated and did not check: +that the two ways the server hands out a line agree bit for bit. **They did +not.** `serde_json`'s default float parser is not correctly rounded, so a +`ts` of `1788546972.6030757` written to the transcript came back from +`/transcript` as `...0755`, while the SSE stream -- serializing the same +struct -- sent the original. Measured on the emulator 2026-09-04: 23 of 330 +cached lines differed from the server's answer in the last bit, so the probe +would have failed on any session whose cached tail happened to be one of +them, silently and only sometimes. That is a defect in the server +independent of this feature -- two answers to "what is line 30" -- and it is +fixed there, with `float_roundtrip` and a test +(`a_line_read_back_is_the_line_that_was_written`) that fails the moment the +feature is dropped. Comparing everything *except* `ts` was the other option +and was rejected: a re-seeded fixture is identical in content and differs +only in when it happened, which is exactly the case the probe exists for. A failed probe **purges the session's cache +and proceeds as a cold open**. A probe that cannot be made (no route to +the server) leaves the cached transcript on screen, shows the request's +error on the stream banner where a connection failure shows today, and is +retried on the stream loop's schedule (`RECONNECT_DELAY_MS`); the stream +is never opened until a probe has passed once for this screen instance. + +What the probe does *not* catch: a line changed in the middle of the file +with the tail intact, or a file rewritten so that the event at the cursor +happens to be identical. Those are what the reload button is for, and the +button's caption says so. + +Cost: one request of a few hundred bytes, one round trip, in the slot +where the opening page's request is today -- so the round trips before +the stream is live are unchanged at two, and the bytes fall from a page to +a line. The cached rows are drawn *before* the probe returns, which is the +whole point of the feature; a failed probe replaces them, the same +appearance as a `reset`. + +Rejected: a server-side check on the stream (`events?after=N&ts=T`, +answered with a distinct frame when the event at N is not what the phone +thinks). Strictly better coverage -- it would run on every reconnect, not +only on open -- and no extra round trip. Not chosen because it puts a +cache's validation into a protocol that otherwise knows nothing about +caching, and because the reset frame already has to keep meaning "you are +behind, your history is fine" (decision 6), so a second frame would be +needed. Worth revisiting if the probe's round trip is ever measured as the +thing making reopen slow; note it as the alternative here and in PLAN.md. + +Rejected: trusting the cache without a check and relying on the reload +button. Invariant 1 is not something a button restores after the fact. + +Rejected: fetching the newest page as today and using it to validate the +overlap. Zero saving on the opening page, which is the request paid on +every open. + +### 4. Pages ask the server only for the gap: `after` on `/transcript` + +After a reader has been away, the cache holds `[a, b)` and the screen +holds the newest window `[W, …)` with a gap between `b` and `W`. Paging +back from `W` asks the server for a coalesced page before `W`, and that +page may reach back past `b` -- a single reply is hundreds of lines, so +forty rows can be thousands of seqs -- producing exactly the overlap +decision 2 refuses to store. Left like that, every cached chunk would be +overlapped and dropped in turn as the reader paged back through the gap, +and the cache would save nothing for the sessions it exists for. + +So the transcript route gains a lower bound. `TranscriptQuery` in +`server/src/routes.rs` gets + + /// Return nothing at or below this seq; the page stops here instead of at `limit`. + /// The phone passes the end of what it already holds, so a page never overlaps it. + #[serde(default)] + after: Option, + +named to match the SSE route's `after` (exclusive, `seq > after`). +`read_window(path, before, after, limit, coalesce)` in +`server/src/session/transcript.rs` computes +`start = first_at_or_after(after + 1)` and stops the walk there: the raw +branch parses `max(start, end - limit)..end`; `parse_coalesced` takes a +`start` and its `while index > 0` becomes `while index > start`. A delta +run cut at `start` is emitted as the partial it is, exactly as one cut by +`limit` already is, and `healSplitMessage` welds it on the phone -- no new +mechanism. The route's table comment in `routes.rs` gains the parameter, +and `transcript.rs` gets a test beside +`a_window_is_the_events_before_a_cursor_and_nothing_else`: with `after` +set, the page's oldest seq is greater than `after`, and with `after` set +inside a delta run the partial run's seq is the first delta above `after`. + +The phone passes `after = b - 1` where `b` is the `end` of the nearest +chunk whose `end ≤ before`, and nothing when there is none. A page that +comes back with `first == b` is adjacent, and the suffix now runs through +the old chunks: the gap is closed with exactly the bytes it was wide, and +the history behind it is served locally from then on. + +Rejected: fetching the gap raw in one request (`before=W&limit=W-b`, +which is what the anchor restore already does). Exact, but a gap of ten +thousand lines is several megabytes downloaded to save re-downloading +history the reader may never scroll to; the feature exists to save data. +Paging as today with a bound saves the same bytes and fetches only what +is read. + +Rejected: dropping the cached run whenever a gap opens. Being more than +`CATCH_UP_LIMIT` (200) events behind is the *ordinary* state of an active +session revisited -- 200 raw events is one reply -- so this would empty +the cache for exactly the sessions that are opened most. + +### 5. A page is served locally in rows, mirroring the server's count + +`loadOlderPage` asks for `HISTORY_PAGE` (40) **rows** when +`coalesce = true`, and for a number of **events** otherwise (the anchor +restore). Served from the cache, the events branch is the `limit` lines +before `before`. The rows branch walks back from the line before `before` +counting rows the way `parse_coalesced` does: every event that is not an +`assistantText` is a row, and each maximal run of `assistantText` lines is +one row; it stops only between rows, once `limit` rows are complete, and +returns the raw lines oldest-first. It does not join the deltas -- the +fold does that (`foldEvent` appends a delta to a preceding +`AssistantMsg`), and the joined row keeps the seq of its first delta either +way, so anchors and the next `before` land where they do today. + +A cached page is allowed to be **short**: the suffix's oldest chunk starts +at some `first`, and a walk that reaches it returns what it found. The +caller already treats a short page as a page; only an *empty* page means +"start of the conversation" (`moreHistory = false`), and the cache never +returns an empty page -- it returns `null` (a miss) and the network is +asked. The walk may cross a chunk boundary inside the suffix, since adjacent +chunks are one run; a delta run straddling a boundary counts as one row, as +it should. + +A miss is `before` **outside what the suffix covers continuously** -- above +its newest `end`, or at or below its oldest `first`. This plan first said a +miss was "no chunk of the suffix ends at `before`", which is wrong in the +commonest case there is: a warm open draws the newest eighty lines of the +live run, so the cursor the reader then scrolls back from is in the *middle* +of a chunk, not at a boundary. Under the narrower rule every warm open sent +its first backwards page to the server, and that page -- reaching back past +the run the phone already held -- overlapped it and could not be stored, so +the same history was fetched again on every visit. The feature would have +saved the opening window and nothing else. + +The row rule is a copy of the server's, and copies drift. It is short +(one comparison), it is pure, and it goes under a JVM unit test with the +same fixture as the server's `coalescing_counts_rows_and_joins_delta_runs` +-- the three cases are a run cut by the limit, a `usageDelta` inside a run +(the server flushes the run there, so it is two rows), and a page that is +all one run. + +### 6. What a `reset` means for the cache: behind, not wrong + +The server sends `reset` when the cursor is more than `CATCH_UP_LIMIT` +events behind, then the newest 200 raw events. The screen already drops +everything and rebuilds from that window. For the cache, a reset means +**the history is intact and there is a gap**: the probe passed, the file +is append-only, and the window's first seq is above the open chunk's end. +The store learns this from the first window event's seq (decision 2: +a seq that is not the open chunk's `end` closes it and opens a new chunk) +and needs no signal from the screen; the gap is filled by paging +(decision 4). + +Two things the reset handler in `SessionScreen` does not clear today and +must: `queued` and `waitingCommands`. Both are folded from events, and a +`messageQueued` whose resolving `userMessage` fell in the gap would +otherwise draw a waiting bubble for a message the session has long since +read. This is a latent bug today, made likely by the cache because a +cached tail is older than a fetched one. `contextTokens` needs no change: +`UsageDelta.context` is absolute, so the window's first one corrects it. + +### 7. Session state that is not the transcript comes from the list, not the cache + +`apply` derives `status`, `model`, `permissionMode` and `compactingSince` +from `Status` and `Settings` events. Replayed from a fetched page those are +current; replayed from the cache they are as old as the last visit, while +`summary.status`, `summary.model` and `summary.permissionMode` -- the row +the reader just tapped -- were fetched moments ago. So the cache replay +runs through `apply` for the transcript's sake (queued bubbles, context, +rows) and then **reassigns those four from `summary`**, which is the newer +of the two measurements; the stream's catch-up then makes them current. +Without this a session that finished an hour ago would open saying +"working" until the stream connected, which is a status row lying for a +round trip. + +### 8. Reload, in session settings + +`SessionSettingsDialog` gains a row under the working directory: + + [ Transcript ] 2.3 MB cached [ Reload ] + +The size is what the button discards, and it is the unknown state made +visible: `null` while the directory is being measured (spinner, as the +notifications switch does), "nothing cached" when the directory is absent +or empty, else the size. A caption in the style of Move's, because the +button costs something the reader cannot see: + + Reload throws away this phone's copy and fetches the transcript from the + server again. Use it when what is shown here disagrees with the file on + the machine. + +Pressing it: purge the session's cache directory, close the dialog, and +rebuild the screen as a cold open -- the same sequence as `reset` plus a +fresh opening fetch, with the reader put back where they were. The +mechanism is an `epoch` counter (`mutableIntStateOf(0)`) added to the key +of the opening effect and the stream effect; incrementing it cancels both +(the stream's `finally` closes the socket) and relaunches them. State the +relaunch must see cleared: `items`, `replies.clear()`, `held`, `oldestSeq += 0`, `moreHistory = true`, `queued`, `waitingCommands`, `lastSeq.set(0)`, +`ready = false`. `savedAnchor` becomes `remember(summary.id, epoch)` so +the restore path reads the anchor saved at the reader's *current* +position (the anchor saver writes on every settle, so it is there), and +`restoring` is re-derived from it. The button is enabled whether or not +anything is cached: "what I see disagrees with the machine" is a state an +empty cache can also be in, and a control that comes and goes makes its +own presence the signal. + +Nothing is announced on success. The transcript shows the opening spinner +and then the rows, which is what the screen already says about a reload. +A failure is the opening fetch's, and lands on the stream banner where +that failure lands today. + +Rejected: a global "clear transcript cache" in the app's settings screen. +Not asked for; eviction (decision 9) bounds the total, and the per-session +button is where the reader is when they notice a problem. Easy to add as +one more caller of `TranscriptCache.purgeAll` if wanted. + +### 9. Budget, eviction, pruning + +The cache is bounded three ways, each with its path out written beside +the path in: + +- **Budget.** `CACHE_BUDGET_BYTES = 256 MB` across all sessions of one + server. Each open touches the session directory's mtime; after the + opening replay, on `Dispatchers.IO`, the store sums the server's + directories and deletes least-recently-touched session directories + (never the one on screen) until under budget. 256 MB is a dozen of the + largest transcripts seen in this VM (21 MB for 24,000 events) and a + small fraction of a phone; it is a number to revisit against real use, + not a measurement. +- **Deleted sessions.** `SessionListScreen`'s delete calls + `cache.session(id).purge()` after `deleteSession` succeeds, and every + successful list fetch calls `cache.retainOnly(ids)` for that server, so + a session deleted from another device or from the backend is pruned on + the next visit to the list. `Drafts.kt` chose not to prune because its + residue is bytes; here it is megabytes, so the pass is worth having. +- **Android.** `cacheDir` may be emptied under pressure at any moment, + including while a screen is open. Every read tolerates a missing + directory (cold open) and every write failure is swallowed once and + disables writing for that screen instance (decision 10). + +### 10. The cache never breaks the screen + +Every store operation that touches the disk catches `IOException` and +answers as if the cache were empty: `null` from a read, no-op from a +write, with the failure logged once at `Log.w("ai-app", …)`. After a +write failure the `SessionCache` instance sets `disabled = true` and +writes nothing more, so a full disk costs one log line rather than one +per delta. A line at the end of an open chunk that does not parse -- the +app died mid-write -- is dropped and the file truncated to the last +good line before anything is served from it; a line that does not parse +anywhere else purges the session's cache (that file was not written by +this code). None of this is reported on screen: none of it changes what +the screen shows, and the reader has nothing to do about it. + +## Layout on disk + + /transcripts/ + v1/ + 10.0.2.2_8443/ one directory per server (host_port) + 3f2c…/ one per session id + 1-1650.rows.jsonl coalesced page: covers seqs 1..1649 + 1650-2001.rows.jsonl + 2001-2400.raw.jsonl a closed live run + 2600-open.raw.jsonl the live run; end = last line's seq + 1 + +Here 2400..2599 is a gap: the reader was away for two hundred events and +the stream reset. The suffix is the single chunk `2600-open`; the first +backwards page asks the server for `before=2600&after=2399&coalesce=true`, +and once a page comes back with `first == 2400` the suffix runs to seq 1. + +Each `.jsonl` is one JSON object per line, oldest first, exactly as the +server sent it. No header, no index: coverage is in the name, order is the +file's, and the seq is in every line. + +## What building it changed + +Each of these contradicted something written above, and each was found by +running it rather than by reading it. The decisions themselves are amended +in place; this is the list of what moved, so that a reader who remembers the +first version knows what to re-read. + +- **The probe's equality had a false premise** -- decision 3. The server did + not hand out the same line twice the same way. Fixed on the server. +- **A cached page starts anywhere inside the run** -- decision 5. Requiring + a chunk boundary would have made the cache save the opening window and + nothing else. +- **The opening window is stored by `append`, not by `storePage`.** The + sketch below had `storePage` grow a special case for "this page is the new + open chunk", decided by an implicit condition that a raw history page also + satisfies. Appending each line instead is the mechanism that already + exists, and the open chunk stays the one thing that grows. +- **Chunks are read backwards, in blocks, and never whole.** Every question + the cache is asked is about the newest end, and a live run reaches the size + of the conversation -- so reading a chunk to answer with eighty lines of it + is the cost the server's own reader was rewritten to stop paying, arriving + on the phone. Damage is therefore noticed when a read reaches it rather + than up front, which is the better time: what is not read cannot be wrong. +- **The stream waits for the opening effect's probe.** The screen lifts + `ready` before the probe returns -- that is the point of the cache -- so + `ready` stopped being the whole gate, and the stream loop asked the same + question a second time and raced its own answer. Two probes per warm open, + visible in the server's log. +- **`SessionCache` is synchronized.** The stream appends live events from + one IO thread while a reader scrolling back reads pages from another; the + open chunk's name, its end and its writer must never be seen + half-rotated. + +## What it cost, measured + +On the emulator against `app/ui-sandbox.sh`, 2026-09-04, on a session of +505 events (three short exchanges and two 300-delta replies): + +- **Reopening it: one request, for one event.** The probe, and nothing else + -- including scrolling the whole conversation back to its first line. A + cold open of the same session is two requests and 100 events. +- **A reset after falling 300 events behind costs the gap and no more.** + The window arrived at seq 306, the phone held up to 202, and the first + backwards page asked `before=306&after=201` and came back with **four + coalesced rows** covering 202..305 -- against the 104 raw events an + unbounded page would have re-fetched and then thrown away. +- **Every chunk is exactly what the server says for the range its name + claims**, checked line by line against `/transcript` for each chunk's own + `before`/`after`/`coalesce`, across a reset and a gap-fill. +- **Nothing about drawing changed**, which is what a cache must not do: + `transcript-bench.sh` before and after, same viewport content and the same + gestures, reported p50 16.9ms both times and the transcript's own draw + accounting at 0.33ms against 0.32ms. + +Still to measure, in real use rather than here: the size the cache reaches +against `CACHE_BUDGET_BYTES`, and whether the probe's round trip is ever +what a reader waits on. + +## Open questions + +- **The probe on every reconnect, not only on open?** Decision 3 probes + once per screen instance. A file replaced *while* the screen is open is + today's behaviour and not made worse, but the server-side check it + rejects would close it. Decide after measuring how often the probe's + round trip is what the reader waits on. +- **TRANSCRIPT_RENDERING.md's item 1** (a reset arriving during an anchor + restore left the screen reconnecting). The cache makes the restore + cheaper and therefore shorter, which narrows the window without closing + it; the fix there is still owed and is unrelated to this. +- **Images.** `SessionImage` fetches bytes from the files route on draw; + they are not part of this cache and are re-downloaded per view. A + separate, simpler cache (a directory of refs, no ordering) if the + measurement above says the images are where the data goes. 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 051cc71..c3e39b6 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -775,15 +775,27 @@ fun fetchTranscript( // 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 { + // Return nothing at or below this seq, stopping the page here instead of at [limit]. The + // phone passes the end of the run it already holds cached, so a page never overlaps that copy + // -- an overlap it cannot store, since a coalesced event cannot be cut at a seq inside its own + // delta run. Exclusive, like the SSE route's cursor. See TranscriptCache and `read_window`. + after: Long? = null, +): List> { val query = buildString { append("?limit=").append(limit) if (before != null) append("&before=").append(before) if (coalesce) append("&coalesce=true") + if (after != null) append("&after=").append(after) } return requestFromServer(settings, "/sessions/$sessionId/transcript$query") { connection -> val body = JSONArray(connection.inputStream.bufferedReader().readText()) - (0 until body.length()).map { parseSeqEvent(body.getJSONObject(it).toString()) } + // The text as well as the event: the transcript cache stores the one and the fold needs + // the other, and they have to be the same line -- a second entry point differing only in + // return type would be two answers to one question. + (0 until body.length()).map { + val line = body.getJSONObject(it).toString() + line to parseSeqEvent(line) + } } } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt index c031e1a..4176411 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt @@ -27,12 +27,19 @@ class EventStream(settings: ServerSettings, private val sessionId: String) { * caller drops what it holds and rebuilds -- the same thing it does when the screen opens. It * arrives before those events, so a caller that clears on it stays in order. */ - fun run(after: Long, onOpen: () -> Unit, onReset: () -> Unit, onEvent: (SeqEvent) -> Unit) { + fun run( + after: Long, + onOpen: () -> Unit, + onReset: () -> Unit, + // The frame's own text as well as the event parsed from it: the transcript cache stores + // the one and the screen folds the other, and they have to be the same line. + onEvent: (raw: String, event: SeqEvent) -> Unit, + ) { stream.run("/sessions/$sessionId/events?after=$after", onOpen) { name, data -> // A named frame carries no payload and a data frame has no name, so this is one or // the other. if (name == RESET_EVENT) onReset() - else if (data.isNotEmpty()) onEvent(parseSeqEvent(data)) + else if (data.isNotEmpty()) onEvent(data, parseSeqEvent(data)) } } } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt index d1cfd65..6f817e9 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt @@ -667,15 +667,6 @@ private fun ImportableList( } } -/** A byte count at the coarsest unit that still says something, so rows stay comparable. */ -private fun humanSize(bytes: Long): String? = - when { - bytes <= 0L -> null - bytes >= 1_000_000L -> "${bytes / 1_000_000L} MB" - bytes >= 1_000L -> "${bytes / 1_000L} kB" - else -> "$bytes B" - } - /** What this session is: the measurements, in the order they are worth knowing. */ private fun statsOf(session: Importable): String = listOfNotNull( diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt index 832b087..5f63f55 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt @@ -29,6 +29,7 @@ import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.runtime.setValue import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier +import androidx.compose.ui.platform.LocalContext import androidx.compose.ui.unit.dp import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch @@ -68,6 +69,11 @@ fun SessionListScreen( // request rather than to the session. var deleting by remember { mutableStateOf>(emptySet()) } + // This phone's copies of these sessions' transcripts, pruned from here because this is where + // a session stops existing. See TranscriptCache. + val context = LocalContext.current + val transcriptCache = remember(settings) { TranscriptCache(cacheRoot(context, settings)) } + fun refresh() { listState = LoadState.Loading scope.launch { @@ -76,6 +82,14 @@ fun SessionListScreen( val loaded = withContext(Dispatchers.IO) { LoadState.Loaded(fetchSessions(settings)) } deleteErrors = emptyMap() + // The path out for a cached transcript whose session was deleted somewhere + // else -- from another device, or at the backend. This list is the only place + // that ever learns the full set, and what the residue costs here is megabytes + // rather than a draft's few bytes. On the answer rather than in `finally`: a + // list that failed to arrive says nothing about which sessions exist. + withContext(Dispatchers.IO) { + transcriptCache.retainOnly(loaded.value.map { it.id }.toSet()) + } loaded } catch (e: ApiException) { LoadState.failed(e) @@ -223,6 +237,9 @@ fun SessionListScreen( try { withContext(Dispatchers.IO) { deleteSession(settings, session.id, alsoDeleteForeign) + // After it succeeded, not before: a refused delete leaves the + // session exactly as it was, and its transcript with it. + transcriptCache.session(session.id).purge() } // Only this row, and only what changed. Refetching the list // instead put every other session back through loading and 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 78834f6..9722eb8 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -81,7 +81,6 @@ import androidx.lifecycle.Lifecycle import androidx.lifecycle.compose.LocalLifecycleOwner import androidx.lifecycle.repeatOnLifecycle import java.util.concurrent.atomic.AtomicLong -import java.util.concurrent.atomic.AtomicReference import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.delay @@ -324,7 +323,27 @@ fun SessionScreen( val lifecycleOwner = LocalLifecycleOwner.current // The resume cursor, written from the stream's IO thread. val lastSeq = remember { AtomicLong(0) } - val activeStream = remember { AtomicReference(null) } + // Bumped to rebuild this screen from nothing -- what Reload in the settings dialog does. It + // keys everything that describes one visit to this session: the source below, the opening + // effect, the stream, and the anchor being put back. See TRANSCRIPT_CACHE.md's decision 8. + var epoch by remember(summary.id) { mutableIntStateOf(0) } + // This server's cached transcripts, and this session's half of them. The cache is per server + // because two servers can hold a session with the same id; the source is per visit because + // Reload throws away what it was reading from. + val cache = remember(settings) { TranscriptCache(cacheRoot(context, settings)) } + val source = + remember(summary.id, epoch) { + TranscriptSource(settings, summary.id, cache.session(summary.id)) + } + // Whether the cached tail has been shown to still be the server's own line. Nothing is + // resumed from a cached cursor until it has -- see [TranscriptSource.probe] -- and a probe + // that could not be made leaves this false for the stream loop to try again. + var probePassed by remember(summary.id, epoch) { mutableStateOf(false) } + // Whether the opening effect is still settling that question. It draws the cached rows and + // lifts [ready] before the answer arrives, which is the point of the cache -- so the stream + // below has to wait for this rather than for `ready`, or it asks the very same question a + // second time and races the answer. + var probing by remember(summary.id, epoch) { mutableStateOf(true) } // The oldest sequence number loaded, and whether there is more behind // it. Paging backwards is what keeps opening a long session cheap: the // screen starts with the end of the conversation and fetches earlier @@ -333,12 +352,12 @@ fun SessionScreen( // Where this session was last being read, from this device's own store. Read once, because // it is the question "where did I leave off" and the answer stops being interesting the // moment the list is on screen. - val savedAnchor = remember(summary.id) { loadScrollAnchor(context, summary.id) } + val savedAnchor = remember(summary.id, epoch) { loadScrollAnchor(context, summary.id) } // Whether the saved position is still being put back -- the history it needs fetched, and the // scroll applied. Nothing is drawn while it is: opening at the newest end and then travelling // to the anchor is exactly the journey a reader must never see, and this transcript is not // allowed to move under one. - var restoring by remember(summary.id) { mutableStateOf(savedAnchor != null) } + var restoring by remember(summary.id, epoch) { mutableStateOf(savedAnchor != null) } // Sent, but not yet read by the session -- which is when the backend // records it and it comes back as a row. Until then it is drawn below // the working indicator, because that is where it is in the session's @@ -400,6 +419,31 @@ fun SessionScreen( val currentUnits by rememberUpdatedState(units) val lastTouch = remember { LastTouch() } + /** + * Drops everything loaded, so the screen can be rebuilt from a window that is not adjacent to + * it. + * + * One function rather than a clearing written at each of the three places that need it -- a + * stream reset, a cached transcript the server turns out not to have, and Reload -- because + * what has to go is a property of "these rows are no longer continuous with what comes next", + * not of who noticed. The two easy ones to leave out are [queued] and [waitingCommands]: both + * are folded from events, so a `messageQueued` whose resolving `userMessage` fell in the gap + * draws a bubble waiting for a message the session read long ago. [contextTokens] needs no + * clearing, because `UsageDelta.context` is absolute and the next one corrects it. + * + * The resume cursor is deliberately *not* cleared here: a reset continues from where it was, + * and only a caller that is starting the conversation again from the server says so itself. + */ + fun dropLoadedTranscript() { + items = listOf() + replies.clear() + held = listOf() + oldestSeq = 0L + moreHistory = true + queued = listOf() + waitingCommands = listOf() + } + /** * Everything the transcript list draws, from one event. * @@ -614,14 +658,7 @@ fun SessionScreen( // `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, - coalesce = coalesce, - ) + val older = source.page(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 @@ -694,17 +731,22 @@ fun SessionScreen( // The stream lifecycle: connect, follow, and on any drop reconnect // from the cursor -- so a flaky link (or a backend restart) costs // nothing but the gap's latency. - // The newest page first, in one request, before the stream opens. The - // stream then starts from where that page ended, so it carries live - // events only -- which is what it is good at. - LaunchedEffect(summary.id) { - try { - val page = withContext(Dispatchers.IO) { fetchTranscript(settings, summary.id) } - // Warmed before the fold lands rather than after: flattening the rows into units - // splits every settled reply ([transcriptUnits]), and the flatten runs in the - // composition that first sees the rows. Folded into a scratch list off this thread - // to find out what needs warming; the real fold below also maintains the queue and - // the cursor, so it cannot be reused here. + // The newest window first, before the stream opens, so the stream starts from where that + // window ended and carries live events only -- which is what it is good at. The window comes + // from this phone's own copy when there is one, and then costs a single request to check + // that the server's transcript is still the one it came from; otherwise it is a page fetched + // as it always was. See TRANSCRIPT_CACHE.md. + LaunchedEffect(summary.id, epoch) { + /** + * One opening window onto the screen, whichever side it came from. + * + * Warmed before the fold lands rather than after: flattening the rows into units splits + * every settled reply ([transcriptUnits]), and the flatten runs in the composition that + * first sees the rows. Folded into a scratch list off this thread to find out what needs + * warming; the real fold below also maintains the queue and the cursor, so it cannot be + * reused here. + */ + suspend fun open(page: List) { withContext(Dispatchers.IO) { var scratch = listOf() page.forEach { entry -> @@ -715,6 +757,58 @@ fun SessionScreen( warm(replies, scratch) } page.forEach { apply(it) } + } + + try { + // This phone's own copy first, drawn before anything is asked of the server -- which + // is the whole point of the cache. What makes it safe to draw before it is checked is + // that a failed check replaces these rows, with the same appearance as a reset. + val cached = withContext(Dispatchers.IO) { source.cachedOpening() } + if (cached != null) { + open(cached) + // A replay is as old as the last visit; the row this screen was opened from was + // fetched moments ago. So the transcript comes from the cache and everything that + // is not the transcript comes from the summary, which is the newer measurement of + // the same thing -- otherwise a session that finished an hour ago opens saying + // "working" until the stream connects, which is a status row lying for a round + // trip. + status = summary.status + model = summary.model + permissionMode = summary.permissionMode ?: "auto" + if (summary.status != "compacting") compactingSince = null + // Nothing to put back, so these rows are the screen and the probe can return + // under them. A restore still has history to fetch and is gated below. + if (savedAnchor == null) ready = true + } + // The one thing a cached cursor has to be shown before the stream resumes from it. + val usable = cached != null && withContext(Dispatchers.IO) { source.probe() } + if (usable) probePassed = true + if (!usable) { + // Either there was nothing cached, or what was cached is not what the server + // has -- the file was replaced or truncated under it. Same clearing as a reset, + // then an ordinary cold open. + if (cached != null) { + dropLoadedTranscript() + lastSeq.set(0) + } + open(withContext(Dispatchers.IO) { source.fetchOpening() }) + // Refilled from the server, so the tail is the server's by construction. + probePassed = true + } + } catch (e: ApiException) { + // Not fatal: the stream below still replays from zero, which is slow but complete. + // Saying so beats silently showing nothing. + // + // It is also where a probe that could not be *made* lands -- a phone with no route to + // the server. Whatever was cached stays on screen and [probePassed] stays false, so + // the stream loop asks again before it resumes from that cursor. + streamError = e.message + } finally { + // However that went, the stream is free to take it from here. + probing = false + } + + try { // Then back where reading stopped. An anchor deeper than the newest page is exactly // the one worth restoring -- somebody who read to the bottom has no anchor at all -- // and the cost was already paid on the way down there. @@ -792,8 +886,8 @@ fun SessionScreen( } } } catch (e: ApiException) { - // Not fatal: the stream below still replays from zero, which is - // slow but complete. Saying so beats silently showing nothing. + // A page of history that never arrived. The reader is left at the newest end rather + // than where they were, which is the state this screen opens in anyway. streamError = e.message } // Whatever happened above, including a page that never arrived: an empty transcript is a @@ -816,6 +910,13 @@ fun SessionScreen( loadingHistory = false } } + // Last, and off this thread: this session is what must not be evicted, so it is marked + // as visited before the budget is applied, and both are a walk of the cache directory + // that nothing on screen is waiting for. + withContext(Dispatchers.IO) { + source.cache.touch() + cache.evictToBudget(keep = summary.id) + } } // Only while the screen is actually on screen. Android stops the @@ -826,16 +927,37 @@ fun SessionScreen( // that has been backgrounded is work nobody is watching. Stopping the // stream deliberately makes the drop a close rather than an error (see // EventStream.close), and resuming reconnects from the same cursor. - LaunchedEffect(summary.id, ready, lifecycleOwner) { + LaunchedEffect(summary.id, ready, epoch, lifecycleOwner) { if (!ready) return@LaunchedEffect + // The opening effect draws cached rows and lifts `ready` *before* it has checked that the + // cursor under them is still the server's, so `ready` is no longer the whole gate: this + // waits for that check to settle. Without it the two run at once, ask the same question + // twice, and race each other's answer -- two probes per warm open in the server's log. + snapshotFlow { probing }.first { !it } lifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) { try { while (true) { - val stream = EventStream(settings, summary.id) - activeStream.set(stream) try { + // A cached cursor whose probe never got an answer, because the server + // could not be reached when the screen opened. Resuming a stream from an + // unchecked cursor is the one thing this must not do, so it is asked + // again here, on the reconnect schedule, with the cached rows still on + // screen meanwhile. False covers both answers that mean "open cold": + // the file is not the one these rows came from, and there was nothing + // cached to check. + if (!probePassed) { + if (withContext(Dispatchers.IO) { source.probe() }) { + probePassed = true + } else { + dropLoadedTranscript() + lastSeq.set(0) + withContext(Dispatchers.IO) { source.fetchOpening() } + .forEach { apply(it) } + probePassed = true + } + } withContext(Dispatchers.IO) { - stream.run( + source.follow( after = lastSeq.get(), // Connected, measured rather than inferred: this is what // takes a failure off the screen, and nothing else does. @@ -850,11 +972,10 @@ fun SessionScreen( // is what makes this the same as opening the // screen -- `apply` refills them, and scrolling // up pages the rest back in as it always does. - items = listOf() - replies.clear() - held = listOf() - oldestSeq = 0L - moreHistory = true + // The cache needs no telling: the window's first + // seq is not the seq it was expecting, which is + // what closes its live run and starts another. + dropLoadedTranscript() }, ) { entry -> apply(entry) @@ -870,7 +991,7 @@ fun SessionScreen( // closing the app over. Reported on the screen either way. streamError = e.message ?: e::class.simpleName } finally { - stream.close() + source.close() } delay(RECONNECT_DELAY_MS) } @@ -878,14 +999,15 @@ fun SessionScreen( // Cancellation -- going below STARTED, or leaving the screen -- // cannot interrupt a blocking socket read. Closing is what // unblocks it, and what marks the drop deliberate. - activeStream.getAndSet(null)?.close() + source.close() } } } // The screen going away entirely, which the lifecycle scope above does // not cover: a composable can leave the composition while the activity - // stays started. - DisposableEffect(summary.id) { onDispose { activeStream.get()?.close() } } + // stays started. Keyed on the epoch as well, so that Reload's replacement + // source is the one a later disposal closes. + DisposableEffect(summary.id, epoch) { onDispose { source.close() } } // Nothing gets announced about the session somebody is reading; see NotificationService. // RESUMED rather than STARTED because "looking at it" means the foreground -- a session left @@ -1936,10 +2058,32 @@ fun SessionScreen( UsageDialog(feed = usageFeed, onDismiss = { usageOpen = false }) } if (settingsOpen) { + // Measured when the dialog opens rather than kept up to date: what the reader is being + // told is what pressing the button now would discard, and null until the walk of the + // directory returns is what not knowing looks like. + var cachedBytes by remember(summary.id, epoch) { mutableStateOf(null) } + LaunchedEffect(summary.id, epoch) { + cachedBytes = withContext(Dispatchers.IO) { source.cache.bytes() } + } SessionSettingsDialog( settings = settings, sessionId = summary.id, title = title, + cachedBytes = cachedBytes, + // The purge finishes before the epoch moves, because the relaunched opening effect + // reads the same directory and would otherwise draw what is about to be deleted. + // Everything else here is the clearing a cold open needs; the epoch is what makes it + // one, by rebuilding the opening effect, the stream, and the anchor being put back. + onReload = { + settingsOpen = false + scope.launch { + withContext(Dispatchers.IO) { source.cache.purge() } + dropLoadedTranscript() + lastSeq.set(0) + ready = false + epoch++ + } + }, // The header takes the new name at once and the dialog closes on it, because the // rename has already been accepted by the server -- see [title], which is this app's // own datum. The list behind this refetches on the way out of the session anyway. diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt index d04f52b..bccbe09 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt @@ -42,9 +42,11 @@ import kotlinx.coroutines.withContext * are changed *while* reading a turn -- "not this model, try that one" -- and a control belongs * with the thing it acts on. * - * Nothing here is captioned. Each control is a labelled noun with a switch or a field beside it, - * and a paragraph under every one of them made the dialog longer than the conversation it covers. - * Failures still get their words: those are what the reader cannot work out by looking. + * Captions are for what a control costs rather than for what it is. Each control is a labelled noun + * with a switch or a field beside it, and a paragraph under every one of them made the dialog + * longer than the conversation it covers -- so Notifications has none, while Move and Reload do, + * because what those two take away is not visible from here. Failures get their words for the same + * reason: they are what the reader cannot work out by looking. */ @Composable fun SessionSettingsDialog( @@ -55,6 +57,12 @@ fun SessionSettingsDialog( */ title: String, onRenamed: (String) -> Unit, + /** + * What this phone is holding of the conversation, or null while that is being measured -- see + * the Reload row below, which is what would discard it. + */ + cachedBytes: Long?, + onReload: () -> Unit, onDismiss: () -> Unit, ) { val scope = rememberCoroutineScope() @@ -250,6 +258,44 @@ fun SessionSettingsDialog( style = MaterialTheme.typography.bodySmall, ) } + Spacer(Modifier.height(8.dp)) + Row( + verticalAlignment = Alignment.CenterVertically, + modifier = Modifier.fillMaxWidth(), + ) { + Text("Transcript", modifier = Modifier.weight(1f)) + // The size is what the button discards, and the unknown state is drawn + // rather than guessed: a spinner while the directory is being measured, and + // words when there is nothing there, because "nothing cached" and "0 B" read + // as different claims. + when { + cachedBytes == null -> + CircularProgressIndicator( + modifier = Modifier.width(16.dp).height(16.dp), + strokeWidth = 2.dp, + ) + else -> + Text( + humanSize(cachedBytes)?.let { "$it cached" } ?: "nothing cached", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + Spacer(Modifier.width(12.dp)) + // Enabled whether or not anything is cached: "what I see disagrees with the + // machine" is a state an empty cache can be in too, and a control that comes + // and goes makes its own presence the signal. + TextButton(onClick = onReload) { Text("Reload") } + } + // Captioned, unlike the controls above it, for the same reason Move is: what it + // costs is not visible, and neither is the case it exists for. + Text( + "Reload throws away this phone's copy and fetches the transcript from the " + + "server again. Use it when what is shown here disagrees with the file " + + "on the machine.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) error?.let { Spacer(Modifier.height(8.dp)) Text( diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Sizes.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Sizes.kt new file mode 100644 index 0000000..1255f1b --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Sizes.kt @@ -0,0 +1,20 @@ +package com.example.aiapp + +/** + * A byte count at the coarsest unit that still says something, so two of them stay comparable. + * + * Null for nothing at all, which is a different answer from a small number and is drawn with words + * rather than a figure: an import row with no size says nothing about size, and a transcript cache + * holding nothing says "nothing cached". + * + * Here rather than beside either caller because a second copy of it would drift, and there is + * already one variant too many -- `ModelsScreen`'s `gigabytes` writes a download's size to two + * decimal places, which is a different question about a much larger number. + */ +fun humanSize(bytes: Long): String? = + when { + bytes <= 0L -> null + bytes >= 1_000_000L -> "${bytes / 1_000_000L} MB" + bytes >= 1_000L -> "${bytes / 1_000L} kB" + else -> "$bytes B" + } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptCache.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptCache.kt new file mode 100644 index 0000000..d985416 --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptCache.kt @@ -0,0 +1,621 @@ +package com.example.aiapp + +import android.util.Log +import java.io.BufferedWriter +import java.io.File +import java.io.FileWriter +import java.io.IOException +import java.io.RandomAccessFile + +/** + * This phone's copy of the transcripts it has already been sent, so reopening a session does not + * download it again. + * + * What is stored is the server's own JSON for one event per line, in transcript order -- the + * elements of a `/transcript` page and the payload of each SSE frame. Reading the cache means + * running the same [parseSeqEvent] the network path runs, so a cached transcript and a fetched one + * cannot draw differently, and an event type this build does not know ([SessionEvent.Unknown]) + * keeps every field it arrived with, on disk, for the build that will. Rows are deliberately *not* + * what is stored: a row is a rendering of events, its shape changes whenever the fold does, and a + * cache of rows would need throwing away on every app update that touched `foldEvent`. + * + * See TRANSCRIPT_CACHE.md for the design. Four rules run through all of it: + * 1. what is on screen is what the server's transcript says, in order, with nothing missing -- the + * cache is a copy and is never inferred, folded or edited here; + * 2. a cached line is never ahead of the live cursor, and the cursor never ahead of the cache; + * 3. the cache is never load-bearing -- missing, evicted, damaged or unwritable all degrade to a + * cold open, never to a blank or a wrong screen; + * 4. a line already on the phone is not fetched again. + * + * A plain [File] root and no Compose, `Context` or network, so the whole of the file logic runs + * under the JVM unit tests. It is also why there is no JSON parser in here: what it needs off a + * line is the sequence number and whether the line is a streamed delta, and both are read with a + * regex over text the server wrote. A line it cannot read that way is treated as damage, which + * gives the same answer as having no cache at all. + * + * [warn] is where failures are said, for the same reason -- `android.util.Log` is a stub that + * throws under the JVM tests, and this file has to be exercisable there. + */ +class TranscriptCache( + private val root: File, + private val warn: (String) -> Unit = { Log.w("ai-app", it) }, +) { + /** The cache for one session, whether or not anything has been stored for it yet. */ + fun session(id: String): SessionCache = SessionCache(File(root, id), warn) + + /** + * Deletes every session directory not in [ids], called after a successful list fetch. + * + * The path out for a session deleted on another device or at the backend: nothing here would + * otherwise ever hear about it, and unlike a draft's few bytes what it leaves behind is + * megabytes. + */ + fun retainOnly(ids: Set) = + guardIo(Unit, warn) { + sessionDirs().forEach { if (it.name !in ids) it.deleteRecursively() } + } + + /** + * Deletes least-recently-touched session directories, never [keep], until the whole of this + * server's cache is under [budget]. + * + * Least-recently-touched rather than largest: what a reader is likely to open again is what + * they opened last, and evicting the big ones first would empty the cache for exactly the + * conversations it exists for. + */ + fun evictToBudget(keep: String, budget: Long = CACHE_BUDGET_BYTES) = + guardIo(Unit, warn) { + val dirs = sessionDirs().sortedBy { it.lastModified() } + var total = dirs.sumOf { sizeOf(it) } + for (dir in dirs) { + if (total <= budget) break + if (dir.name == keep) continue + val was = sizeOf(dir) + if (dir.deleteRecursively()) total -= was + } + } + + fun purgeAll() = guardIo(Unit, warn) { root.deleteRecursively() } + + private fun sessionDirs(): List = root.listFiles()?.filter { it.isDirectory }.orEmpty() +} + +/** + * How much of this phone's cache directory all of one server's transcripts may take. + * + * A dozen of the largest transcripts seen in the dev VM (21 MB for 24,000 events) and a small + * fraction of a phone. A number to revisit against real use rather than a measurement of anything. + */ +const val CACHE_BUDGET_BYTES: Long = 256L * 1000 * 1000 + +/** + * What the newest cached line says, which is what the probe checks against the server. + * + * Both halves are wanted together and by the same caller: the seq is what the request asks about, + * and the line is what its answer is compared with. + */ +data class CachedTail(val seq: Long, val line: String) + +/** + * One session's cached lines, as a directory of chunks. + * + * A chunk is a set of lines *and a claim about what they cover*, and the two are not the same + * thing: a coalesced page joins each run of streamed deltas into one event carrying the seq of the + * run's oldest delta, so a page whose newest event is seq 1,200 may in fact cover everything up to + * the 1,650 it was fetched with, and nothing in the lines says so. So coverage is the half-open + * range in the file's name: + * ``` + * -.rows.jsonl a coalesced page; end is the `before` it was fetched with + * -.raw.jsonl an uncoalesced page, or a closed live run + * -open.raw.jsonl the live run; end is its last line's seq + 1 + * ``` + * + * Two chunks are adjacent when one's `end` is the other's `first`. Only the contiguous run of + * adjacent chunks ending at the newest chunk -- the **suffix** -- is ever served: chunks behind a + * gap are kept, because the gap is usually closed by paging back through it, but nothing is served + * across one. + * + * **The newest chunk is always raw**, which is what makes the stream cursor and the probe well + * defined -- a raw chunk's last line is a real event at a real seq, and the server never coalesces + * the newest window. It holds by construction (the opening window and every stream frame are raw) + * and is checked on read: a `.rows` chunk at the newest end can only mean this app died between + * closing one live run and opening the next, and it discards the session. + * + * Nothing here is load-bearing. Every operation that touches the disk answers as though the cache + * were empty when it cannot, and a write failure disables writing for the rest of this instance's + * life so that a full disk costs one log line rather than one per delta. + * + * Every operation is synchronized, because two of them really do run at once: the stream appends + * live events from its own IO thread while a reader scrolling back reads pages from another. The + * lock is uncontended in the ordinary case and what it buys is that the open chunk's name, its end + * and its writer are never read half-rotated -- which would show up as a page silently fetched + * again, or as a stored chunk overlapping the run it was written beside. + */ +class SessionCache( + private val dir: File, + private val warn: (String) -> Unit = { Log.w("ai-app", it) }, +) { + /** Set by the first write that fails: a second would fail the same way, once per delta. */ + private var disabled = false + /** + * The open chunk's writer, its file, and the seq that chunk now ends at. + * + * Buffered, and flushed on [flush], because a delta is a hundred bytes and arrives dozens of + * times a second while a reply streams -- a syscall each is the thing to avoid. What that costs + * is the unflushed tail on a crash, which is safe: a shorter cache is a longer catch-up, never + * a wrong one. + */ + private var writer: BufferedWriter? = null + private var openFile: File? = null + private var openEnd: Long = 0 + + /** + * The newest line of the suffix, or null when there is none or the newest chunk is not raw. + * + * This is the cursor the live stream would resume from, so it is also what has to be shown to + * still be the server's own line before anything is resumed from it -- see + * `TranscriptSource.probe`. + */ + @Synchronized + fun tail(): CachedTail? = + guard(null) { + val newest = suffix().lastOrNull() ?: return@guard null + var found: CachedTail? = null + eachLine(newest) { line -> + found = CachedTail(seqOf(line)!!, line) + false + } + found + } + + /** The newest [limit] lines of the suffix, oldest first -- the opening window. */ + @Synchronized + fun newest(limit: Int): List = + guard(emptyList()) { + val taken = ArrayDeque() + for (chunk in suffix().asReversed()) { + if (taken.size >= limit) break + eachLine(chunk) { line -> + taken.addFirst(line) + taken.size < limit + } + } + taken.toList() + } + + /** + * The page of lines before [before], oldest first, or null when the cache cannot answer. + * + * Null is a miss -- the suffix does not cover the ground immediately below [before] -- and + * means the server has to be asked. It is deliberately not an empty list: an empty page is how + * the screen is told it has reached the start of the conversation, and a cache saying that of + * history it merely does not hold would stop the transcript scrolling back for good. + * + * [before] is anywhere inside the suffix, not only at a chunk boundary. The cursor a warm open + * leaves behind is in the middle of the live run -- the screen draws the newest eighty lines of + * it -- so a cache that could only answer at a boundary would send the very first backwards + * page to the server and, since that page would overlap the run, keep none of it. + * + * A short page is fine, and is what a walk that reaches the oldest chunk of the suffix returns: + * the caller already treats a short page as a page. + * + * With [rows] the count is rows rather than lines, mirroring the server's `parse_coalesced`: + * every event that is not a streamed delta is a row, and each maximal run of deltas is one row. + * The deltas are not joined here -- `foldEvent` does that, and the joined row keeps the seq of + * its first delta either way, so anchors and the next `before` land where they do today. + */ + @Synchronized + fun page(before: Long, limit: Int, rows: Boolean): List? = + guard(null) { + val suffix = suffix() + val newest = suffix.lastOrNull() ?: return@guard null + // Above what is held, or at or below where it starts: either way the run the caller + // is scrolling into is not continuous with this one, and only the server has it. + if (before > newest.end || before <= suffix.first().first) return@guard null + val taken = ArrayDeque() + var counted = 0 + var inRun = false + var wanting = true + for (chunk in suffix.asReversed()) { + if (!wanting) break + if (chunk.first >= before) continue + eachLine(chunk) { line -> + // The page is what is *before* the cursor; the rows at or above it are the + // ones already on screen. + if (seqOf(line)!! >= before) return@eachLine true + if (rows) { + val delta = isDelta(line) + // Stop only between rows: a delta continuing the run being gathered is + // part of a row already counted, and breaking on it would drop the half + // of that row already taken. + if (counted >= limit && !(delta && inRun)) wanting = false + else { + if (!delta || !inRun) counted++ + inRun = delta + } + } else if (taken.size >= limit) { + wanting = false + } + if (wanting) taken.addFirst(line) + wanting + } + } + taken.toList() + } + + /** + * The `end` of the nearest chunk at or below [before], which is the floor a fetched page is + * asked with so that it stops where this phone's copy starts. Null when there is no such chunk. + * + * Any chunk, not only the suffix's: the whole point is to reach the run behind a gap, so that + * the gap is closed with exactly the bytes it is wide and the history behind it is served + * locally from then on. + */ + @Synchronized + fun coveredUpTo(before: Long): Long? = + guard(null) { chunks().map { it.end }.filter { it <= before }.maxOrNull() } + + /** + * Stores a fetched page covering `[first, end)`; false when it was not stored. + * + * Refused when it overlaps a chunk already here, because there is no clean cut: a coalesced + * event cannot be split at a seq inside its own delta run. `TranscriptSource` keeps that from + * arising by bounding what it fetches, and this is the guard for a page that arrives anyway -- + * from a server without the `after` parameter, say. Such a page is still drawn; it is only not + * kept. + * + * The newest chunk is never stored through here: the opening window and every live frame go + * through [append], which is what keeps the newest chunk raw and open. + */ + @Synchronized + fun storePage(lines: List, first: Long, end: Long, rows: Boolean): Boolean = + guard(false) { + if (disabled || lines.isEmpty() || end <= first) return@guard false + if (chunks().any { first < it.end && it.first < end }) return@guard false + dir.mkdirs() + val kind = if (rows) "rows" else "raw" + File(dir, "$first-$end.$kind.jsonl").writeText(lines.joinToString("\n", postfix = "\n")) + true + } + + /** + * Appends one live event, which is also how a freshly fetched opening window is stored. + * + * A seq equal to the open chunk's end extends it. A larger one is a gap -- which is what a + * `reset` looks like from here -- and closes the open chunk under the end it turned out to have + * before starting a new one at [seq]. A smaller one is already covered and is ignored; the SSE + * contract is `seq > after`, so that is a guard rather than a path. + */ + @Synchronized + fun append(line: String, seq: Long) = + guard(Unit) { + if (disabled) return@guard + val writer = writerFor(seq) ?: return@guard + // Written as it arrived. A newline inside it would split one event into two + // unreadable halves, but neither source can produce one: SSE framing forbids it, and + // a page's elements are re-serialized compactly, which escapes it. + writer.write(line) + writer.write("\n") + openEnd = seq + 1 + } + + /** + * Flushes what [append] has buffered. Called on each `Status` event -- the boundaries of a + * turn, which is the granularity a crash may as well lose -- and when the stream closes. + */ + @Synchronized fun flush() = guard(Unit) { writer?.flush() } + + /** What [purge] would discard, for the reload row in session settings. */ + @Synchronized fun bytes(): Long = guard(0L) { sizeOf(dir) } + + /** Marks this session as visited, which is what eviction ranks by. */ + @Synchronized + fun touch() = + guard(Unit) { if (dir.isDirectory) dir.setLastModified(System.currentTimeMillis()) } + + @Synchronized + fun purge() = + guard(Unit) { + closeWriter() + dir.deleteRecursively() + } + + // -- chunks ------------------------------------------------------------------------------ + + private data class Chunk(val file: File, val first: Long, val end: Long, val open: Boolean) { + val rows: Boolean + get() = file.name.endsWith(".rows.jsonl") + } + + /** + * Every chunk on disk, oldest first. A name this does not recognise is not ours and is ignored. + * + * Recomputed per operation rather than kept: another operation may have changed the directory, + * and a hundred names is a directory listing. + */ + private fun chunks(): List { + writer?.flush() + return dir.listFiles() + .orEmpty() + .mapNotNull { file -> + val match = CHUNK_NAME.matchEntire(file.name) ?: return@mapNotNull null + val first = match.groupValues[1].toLongOrNull() ?: return@mapNotNull null + val open = match.groupValues[2] == "open" + val end = if (open) openEndOf(file, first) else match.groupValues[2].toLongOrNull() + // A chunk covering nothing is one that was created and never written to -- an + // append whose very first write failed. It says nothing, so it is not a chunk. + if (end == null || end <= first) null else Chunk(file, first, end, open) + } + .sortedBy { it.first } + } + + /** + * The open chunk's end: its last line's seq plus one, or the in-memory end while this instance + * is the one writing it. + * + * An open chunk whose last line cannot be read is this app having died mid-write. That line is + * dropped and the file truncated to the last good one before anything is served from it, which + * is the one place damage is repaired rather than discarded: the tail of an append-only file is + * the only place a partial line can be. + */ + private fun openEndOf(file: File, first: Long): Long { + if (openFile == file && openEnd > 0) return openEnd + repairTail(file) + var end = first + eachLineBackwards(file) { _, line -> + seqOf(line)?.let { end = it + 1 } + false + } + return end + } + + /** + * The contiguous run of adjacent chunks ending at the newest one, oldest first. + * + * A newest chunk that is not raw cannot happen while this code is the only writer, and means + * the directory is not to be trusted -- so the session is discarded rather than served across + * whatever else is wrong with it. + */ + private fun suffix(): List { + val all = chunks() + var index = all.size - 1 + val newest = all.lastOrNull() ?: return emptyList() + if (newest.rows) throw Damaged(newest.file) + val run = ArrayDeque() + run.addFirst(newest) + while (index > 0 && all[index - 1].end == run.first().first) { + index-- + run.addFirst(all[index]) + } + return run.toList() + } + + /** + * Each line of [chunk], newest first, until [take] says stop. + * + * Backwards and lazily, because every question this cache is asked is about the newest end -- + * the tail, the opening window, the page before a cursor -- and a live run grows to the size of + * the conversation. Reading the file whole to answer with eighty lines of it is the cost the + * server's own reader was rewritten to stop paying. + * + * Damage anywhere but at the tail of the open chunk was not written by this code, and there is + * no honest way to say what a chunk covers with a line of it unreadable -- so it discards the + * session rather than serving what it can read. + */ + private fun eachLine(chunk: Chunk, take: (String) -> Boolean) { + eachLineBackwards(chunk.file) { _, line -> + if (seqOf(line) == null) throw Damaged(chunk.file) + take(line) + } + } + + // -- writing ----------------------------------------------------------------------------- + + /** The writer for the chunk [seq] belongs in, opening or rotating one as it has to. */ + private fun writerFor(seq: Long): BufferedWriter? { + writer?.let { held -> + if (seq == openEnd) return held + if (seq < openEnd) return null + // A gap: what this instance has written covers up to `openEnd`, and that is the name + // the chunk gets before a new one starts at the arriving seq. + closeOpenChunk(openEnd) + } + dir.mkdirs() + // An open chunk left by an earlier instance, or by an earlier screen. + chunks() + .lastOrNull { it.open } + ?.let { existing -> + if (seq < existing.end) return null + if (seq == existing.end) { + openFile = existing.file + openEnd = existing.end + return FileWriter(existing.file, true).buffered().also { writer = it } + } + rename(existing.file, existing.first, existing.end) + } + // A chunk that was created and never written to would otherwise be left behind under a + // name a second one is about to want; it covers nothing, so nothing is lost with it. + dir.listFiles().orEmpty().forEach { + if (CHUNK_NAME.matchEntire(it.name)?.groupValues?.get(2) == "open" && it.length() == 0L) + it.delete() + } + val file = File(dir, "$seq-open.raw.jsonl") + openFile = file + openEnd = seq + return FileWriter(file, false).buffered().also { writer = it } + } + + /** Renames the open chunk to the range it turned out to cover, so it stops being open. */ + private fun closeOpenChunk(end: Long) { + val file = openFile + closeWriter() + if (file == null) return + val first = CHUNK_NAME.matchEntire(file.name)?.groupValues?.get(1)?.toLongOrNull() + if (first != null) rename(file, first, end) + } + + private fun rename(file: File, first: Long, end: Long) { + file.renameTo(File(dir, "$first-$end.raw.jsonl")) + } + + private fun closeWriter() { + try { + writer?.close() + } catch (_: IOException) { + // Nothing left to do about it: the file is what it is, and the read path repairs a + // half-written tail. + } + writer = null + openFile = null + openEnd = 0 + } + + // -- failure ----------------------------------------------------------------------------- + + /** A chunk that cannot be read as what its name claims. */ + private class Damaged(val file: File) : RuntimeException() + + /** + * Runs [body], answering [ifBroken] when the directory cannot give a real answer. + * + * None of this is reported on screen: none of it changes what the screen shows -- every read + * here has a network path beside it producing the same result -- and the reader has nothing to + * do about it. It is logged, and damage discards this session's cache, which is what makes the + * next open an ordinary cold one. + */ + private fun guard(ifBroken: T, body: () -> T): T = + // A disk that refused once will refuse again, once per delta, so the first refusal is + // also the last: this instance stops writing rather than logging a line a token. + guardIo( + ifBroken, + warn, + onFailure = { + disabled = true + closeWriter() + }, + ) { + try { + body() + } catch (e: Damaged) { + warn("transcript cache damaged at ${e.file}; discarding ${dir.name}") + closeWriter() + dir.deleteRecursively() + ifBroken + } + } +} + +/** `-..jsonl`; anything else in the directory is not ours. */ +private val CHUNK_NAME = Regex("""^(\d+)-(\d+|open)\.(rows|raw)\.jsonl$""") + +private val SEQ_IN_LINE = Regex(""""seq"\s*:\s*(\d+)""") +private val TYPE_IN_LINE = Regex(""""type"\s*:\s*"([^"]*)"""") + +/** + * One line's sequence number, or null when the line is not one of ours. + * + * A regex rather than a JSON parse, so that this file carries no parser and runs under the JVM + * tests: the seq is the first field the server writes (`SeqEvent`'s declaration order, with the + * event flattened after it), so the first match is the top-level one. + */ +private fun seqOf(line: String): Long? = SEQ_IN_LINE.find(line)?.groupValues?.get(1)?.toLongOrNull() + +/** Whether a line is one streamed piece of a reply, which is what makes a run of them one row. */ +private fun isDelta(line: String): Boolean = + TYPE_IN_LINE.find(line)?.groupValues?.get(1) == "assistantText" + +/** + * How much of a file is read at a time when walking it backwards. One block covers a page of a + * transcript comfortably, and the walk stops as soon as the caller has what it asked for. + */ +private const val READ_BLOCK = 64 * 1024 + +/** + * Calls [onLine] with each non-blank line of [file], **newest first**, along with the byte offset + * it starts at, until [onLine] answers false. + * + * Every question the cache is asked is about the newest end of a chunk, and a live run reaches the + * size of the conversation, so reading forwards means reading a transcript to answer with the last + * eighty lines of it. This reads blocks from the end and stops where the caller stops. + * + * Splitting on bytes is safe because the separator is `\n`, which cannot occur inside a multi-byte + * UTF-8 sequence; each line is decoded whole, so nothing is cut through a character. A missing file + * yields nothing, which is the same answer as an empty one. + */ +private fun eachLineBackwards(file: File, onLine: (offset: Long, line: String) -> Boolean) { + if (!file.isFile) return + RandomAccessFile(file, "r").use { handle -> + // Bytes below `unread` have not been looked at; `pending` is the oldest line so far, which + // is incomplete until a newline is found before it in an older block. + var unread = handle.length() + var pending = ByteArray(0) + while (unread > 0) { + val take = minOf(READ_BLOCK.toLong(), unread).toInt() + val start = unread - take + val block = ByteArray(take) + handle.seek(start) + handle.readFully(block) + val buffer = if (pending.isEmpty()) block else block + pending + var lineEnd = buffer.size + var at = buffer.size - 1 + while (at >= 0) { + if (buffer[at] == NEWLINE) { + val line = String(buffer, at + 1, lineEnd - at - 1, Charsets.UTF_8) + if (line.isNotBlank() && !onLine(start + at + 1, line)) return + lineEnd = at + } + at-- + } + pending = buffer.copyOfRange(0, lineEnd) + unread = start + } + // The first line of a file has no newline before it to be found. + val first = String(pending, Charsets.UTF_8) + if (first.isNotBlank()) onLine(0, first) + } +} + +private const val NEWLINE = '\n'.code.toByte() + +/** + * Drops a final line that is not one of ours, by truncating the file to where it starts. + * + * This app having died mid-write is the one kind of damage that is repaired rather than discarded: + * the tail of an append-only file is the only place a partial line can be, and everything before it + * is intact. A second bad line is not this, and is left for the read path to notice. + */ +private fun repairTail(file: File) { + var truncateTo = -1L + eachLineBackwards(file) { offset, line -> + if (seqOf(line) == null) truncateTo = offset + false + } + if (truncateTo >= 0) RandomAccessFile(file, "rw").use { it.setLength(truncateTo) } +} + +private fun sizeOf(file: File): Long = + if (file.isDirectory) file.listFiles().orEmpty().sumOf { sizeOf(it) } else file.length() + +/** + * The disk half of [SessionCache.guard], shared with [TranscriptCache]'s own maintenance. + * + * [onFailure] is what the caller does about it beyond answering [ifBroken] -- for a session's + * cache, giving up on writing. + */ +private fun guardIo( + ifBroken: T, + warn: (String) -> Unit, + onFailure: () -> Unit = {}, + body: () -> T, +): T = + try { + body() + } catch (e: IOException) { + warn("transcript cache unusable: ${e.message}") + onFailure() + ifBroken + } catch (e: SecurityException) { + warn("transcript cache unreadable: ${e.message}") + onFailure() + ifBroken + } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt index 867bec6..cb91df9 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt @@ -5,9 +5,14 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext /** - * What the transcript renders: the event stream folded into displayable rows (see [foldEvent]). The - * stream is the only data source -- opening a session screen replays from seq 0, and a reconnect - * resumes from the last seq seen, so there is no separate history fetch to drift from it. + * What the transcript renders: the event stream folded into displayable rows (see [foldEvent]). + * + * Events are the only data source, and there is deliberately no second shape for history to drift + * from: a page fetched backwards, a live frame, and a line read out of this phone's own cache are + * all the same events through the same parser. Since 2026-09-04 the cache is where most of them + * come from on a session opened again -- see [TranscriptCache], which stores the server's lines + * rather than these rows for exactly that reason: a row is a rendering, and its shape changes + * whenever this file does. */ @Immutable sealed class TranscriptItem { diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptSource.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptSource.kt new file mode 100644 index 0000000..b716009 --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptSource.kt @@ -0,0 +1,183 @@ +package com.example.aiapp + +import android.content.Context +import java.io.File +import java.util.concurrent.atomic.AtomicReference + +/** + * Where the session screen gets a transcript from: this phone's copy first, the server for the + * rest. + * + * One seam rather than a cache the screen has to remember to consult. Everything it fetched before + * -- the opening window, the pages it scrolls back through, the span an anchor restore reaches for + * -- is asked of this, and everything the server sends is written into the cache on the way past, + * so the screen never learns which side answered. What it does learn, through [DebugStats], is how + * often each one did, which is how the saving is measured. + * + * See TRANSCRIPT_CACHE.md. The one rule worth keeping in mind here: the cache is never + * load-bearing. Every read has a network path beside it producing the same result, so a missing, + * evicted or damaged cache degrades to exactly what this screen did before it existed. + */ +class TranscriptSource( + private val settings: ServerSettings, + private val sessionId: String, + val cache: SessionCache, +) { + private val stream = AtomicReference(null) + + /** + * The cached opening window, or null when there is nothing usable to draw. + * + * Drawn *before* [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. + */ + fun cachedOpening(limit: Int = OPENING_WINDOW): List? { + if (cache.tail() == null) return null + val lines = cache.newest(limit) + if (lines.isEmpty()) return null + return try { + lines.map { parseSeqEvent(it) } + } catch (e: org.json.JSONException) { + // Lines 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. + cache.purge() + null + } + } + + /** + * Whether the server's event at the cached cursor is still the cached one. + * + * The screen must not resume a stream from a cached seq unless it is the same conversation. A + * transcript is append-only in ordinary use, but the file can be replaced or truncated -- a + * sandbox re-seeded with the same ids, a backup restored, a directory deleted and the 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. That is the worst + * thing this feature can do, and it is caught with one request of a few hundred bytes, in the + * slot the opening page's request used to be in. + * + * False purges the cache and means "open cold". A throw is the server not being askable, which + * is neither: the cached rows stay on screen, the failure goes on the stream banner, and the + * caller tries again on the stream's own reconnect schedule. + * + * What this cannot see is a line changed in the middle of the file with the tail intact. That + * is what the Reload button in session settings is for, and its caption says so. + */ + suspend fun probe(): Boolean { + val tail = cache.tail() ?: return 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. + val answer = fetchTranscript(settings, sessionId, before = tail.seq + 1, limit = 1) + val matches = + answer.size == 1 && + try { + answer[0].second == parseSeqEvent(tail.line) + } catch (e: org.json.JSONException) { + false + } + if (!matches) cache.purge() + return 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 [probe] said what it had was not the server's. + */ + suspend fun fetchOpening(): List { + DebugStats.count("transcript page from server") + val page = fetchTranscript(settings, sessionId, limit = OPENING_WINDOW) + page.forEach { (line, entry) -> cache.append(line, entry.seq) } + cache.flush() + return page.map { it.second } + } + + /** + * The page before [before]: from the cache when it holds it, otherwise from the server bounded + * by what the cache already has. + * + * The bound 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 would 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, the gap is closed with exactly the bytes it was wide, and the history behind + * it is served locally from then on. + */ + suspend fun page(before: Long, limit: Int, coalesce: Boolean): List { + cache.page(before, limit, rows = coalesce)?.let { lines -> + DebugStats.count("transcript page from cache") + return lines.map { parseSeqEvent(it) } + } + DebugStats.count("transcript page from server") + val page = + fetchTranscript( + settings, + sessionId, + before = before, + limit = limit, + coalesce = coalesce, + after = cache.coveredUpTo(before)?.minus(1), + ) + if (page.isNotEmpty()) { + // `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. + cache.storePage(page.map { it.first }, page.first().second.seq, before, rows = coalesce) + } + return page.map { it.second } + } + + /** + * [EventStream.run], with every frame written to the cache before [onEvent] 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 the 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. + */ + fun follow(after: Long, onOpen: () -> Unit, onReset: () -> Unit, onEvent: (SeqEvent) -> Unit) { + val opened = EventStream(settings, sessionId) + stream.getAndSet(opened)?.close() + try { + opened.run(after, onOpen, onReset) { raw, entry -> + cache.append(raw, entry.seq) + if (entry.event is SessionEvent.Status) cache.flush() + onEvent(entry) + } + } finally { + cache.flush() + } + } + + /** Ends the stream, from any thread, and leaves the cache with everything it was given. */ + fun close() { + stream.getAndSet(null)?.close() + cache.flush() + } +} + +/** + * How many events the screen opens with, cached or fetched. + * + * The server's own default for a page, 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. + */ +private const val OPENING_WINDOW = 80 + +/** + * Where this server's cached transcripts live. + * + * Under `cacheDir` because that is exactly what it is for: bytes the phone can regenerate from the + * server, which Android may delete under storage pressure without asking. Keyed by host and port + * because two servers can hold a session with the same id -- the sandbox and the real server, or a + * re-enrolment -- and a line from one shown against the other is the whole invariant broken. `v1` + * is the layout's version: a change to it bumps the segment, and a directory of another version is + * deleted the first time this is called. + */ +fun cacheRoot(context: Context, settings: ServerSettings): File { + val transcripts = File(context.cacheDir, "transcripts") + transcripts.listFiles()?.forEach { if (it.name != CACHE_VERSION) it.deleteRecursively() } + return File(transcripts, "$CACHE_VERSION/${settings.host}_${settings.port}") +} + +private const val CACHE_VERSION = "v1" diff --git a/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptCacheTest.kt b/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptCacheTest.kt new file mode 100644 index 0000000..04b4db9 --- /dev/null +++ b/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptCacheTest.kt @@ -0,0 +1,310 @@ +package com.example.aiapp + +import java.io.File +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNull +import kotlin.test.assertTrue +import org.junit.jupiter.api.io.TempDir + +/** + * The cache's file logic, which is the half of the transcript cache that can be wrong without + * anything on screen saying so: a page served short, a chunk served across a gap, or a run of lines + * whose recorded coverage does not match what is in it. + * + * Lines here are the shape the server writes -- `{"seq":N,"ts":T,"type":...}` -- because that is + * what the cache reads its two facts off. Nothing parses JSON on either side. + */ +class TranscriptCacheTest { + @field:TempDir lateinit var temp: File + + private val said = mutableListOf() + + private fun cache() = TranscriptCache(File(temp, "v1/host_8443")) { said += it } + + private fun session(id: String = "s") = cache().session(id) + + private fun line(seq: Long, type: String = "toolStart") = + """{"seq":$seq,"ts":1.5,"type":"$type","id":"x"}""" + + private fun delta(seq: Long) = line(seq, "assistantText") + + private fun dirOf(id: String = "s") = File(temp, "v1/host_8443/$id") + + private fun names(id: String = "s") = dirOf(id).list().orEmpty().sorted() + + private fun write(name: String, lines: List, id: String = "s") { + dirOf(id).mkdirs() + File(dirOf(id), name).writeText(lines.joinToString("\n", postfix = "\n")) + } + + private fun seqs(lines: List?) = lines?.map { + Regex("\"seq\":(\\d+)").find(it)!!.groupValues[1].toLong() + } + + @Test + fun an_appended_run_is_one_open_chunk_and_its_newest_line_is_the_tail() { + val cache = session() + (1L..3L).forEach { cache.append(line(it), it) } + cache.flush() + + assertEquals(listOf("1-open.raw.jsonl"), names()) + assertEquals(CachedTail(3, line(3)), cache.tail()) + assertEquals(listOf(line(2), line(3)), cache.newest(2)) + // More than there is is what there is, which is a short opening window and not a failure. + assertEquals(3, cache.newest(80).size) + } + + @Test + fun a_gap_in_the_stream_closes_the_open_chunk_under_the_end_it_turned_out_to_have() { + val cache = session() + (1L..3L).forEach { cache.append(line(it), it) } + // What a `reset` looks like from here: the next event is not the one after the last. + cache.append(line(90), 90) + cache.flush() + + assertEquals(listOf("1-4.raw.jsonl", "90-open.raw.jsonl"), names()) + // Nothing is served across the gap: the suffix is the newest chunk alone. + assertEquals(listOf(line(90)), cache.newest(80)) + assertEquals(CachedTail(90, line(90)), cache.tail()) + } + + @Test + fun an_event_already_covered_is_not_written_again() { + val cache = session() + (1L..3L).forEach { cache.append(line(it), it) } + cache.append(line(2), 2) + cache.flush() + + assertEquals(listOf(1L, 2L, 3L), seqs(cache.newest(80))) + } + + @Test + fun an_adjacent_page_extends_the_suffix_and_a_gap_stops_it() { + val cache = session() + (100L..102L).forEach { cache.append(line(it), it) } + cache.flush() + + // Adjacent: its end is the open chunk's first. + assertTrue(cache.storePage((60L..99L).map { line(it) }, 60, 100, rows = true)) + assertEquals(listOf(98L, 99L), seqs(cache.page(before = 100, limit = 2, rows = false))) + assertEquals(60L, seqs(cache.newest(80))?.first()) + + // Behind a gap: kept on disk, because paging usually closes the gap, but never served + // across it. + assertTrue(cache.storePage((1L..9L).map { line(it) }, 1, 10, rows = true)) + assertNull(cache.page(before = 10, limit = 5, rows = false)) + assertEquals(60L, seqs(cache.newest(200))?.first()) + } + + @Test + fun a_page_that_overlaps_what_is_here_is_not_stored() { + val cache = session() + cache.append(line(100), 100) + cache.flush() + assertTrue(cache.storePage((60L..99L).map { line(it) }, 60, 100, rows = true)) + + assertFalse(cache.storePage((50L..79L).map { line(it) }, 50, 80, rows = true)) + assertFalse(cache.storePage(emptyList(), 40, 60, rows = true)) + assertEquals(listOf("100-open.raw.jsonl", "60-100.rows.jsonl"), names()) + } + + @Test + fun a_miss_is_null_and_never_an_empty_page() { + val cache = session() + (100L..102L).forEach { cache.append(line(it), it) } + cache.flush() + + // At or below where the run starts, so what the reader is scrolling into is the server's. + // An empty list here would be read as the start of the conversation and would stop the + // transcript scrolling back at all. + assertNull(cache.page(before = 100, limit = 40, rows = true)) + assertNull(cache.page(before = 40, limit = 40, rows = true)) + assertNull(session("never-visited").page(before = 100, limit = 40, rows = true)) + } + + @Test + fun a_page_starts_from_anywhere_inside_the_run_not_only_at_a_boundary() { + val cache = session() + (1L..10L).forEach { cache.append(line(it), it) } + cache.flush() + + // Where a warm open leaves the cursor: in the middle of the live run, because the screen + // drew the newest lines of it. A cache that could only answer at a chunk boundary would + // send this to the server -- and the page that came back would overlap the run and be + // thrown away, so the whole of the scroll back would be fetched again on every visit. + assertEquals(listOf(5L, 6L, 7L), seqs(cache.page(before = 8, limit = 3, rows = false))) + assertEquals((1L..7L).toList(), seqs(cache.page(before = 8, limit = 99, rows = false))) + } + + @Test + fun a_page_counted_in_rows_folds_each_delta_run_into_one_and_cuts_only_between_rows() { + val cache = session() + // Two replies of three deltas each, split by a tool call: the same fixture as the + // server's `coalescing_counts_rows_and_joins_delta_runs`. + val lines = + listOf(delta(1), delta(2), delta(3), line(4), delta(5), delta(6), delta(7), line(8)) + write("1-9.raw.jsonl", lines) + cache.append(line(9), 9) + cache.flush() + + // Three rows: the tool call at 8, the run 5..7, and the tool call at 4. The cut lands + // between rows, so the older run is not started. + assertEquals( + listOf(4L, 5L, 6L, 7L, 8L), + seqs(cache.page(before = 9, limit = 3, rows = true)), + ) + // One row is one whole run, however many deltas it is made of. + assertEquals(listOf(8L), seqs(cache.page(before = 9, limit = 1, rows = true))) + // A page of lines counts lines, which is what the anchor restore asks for. + assertEquals(listOf(7L, 8L), seqs(cache.page(before = 9, limit = 2, rows = false))) + } + + @Test + fun a_row_page_crosses_a_chunk_boundary_and_stops_short_at_the_oldest_chunk() { + val cache = session() + write("5-9.raw.jsonl", listOf(delta(5), delta(6), line(7), delta(8))) + cache.append(delta(9), 9) + cache.append(line(10), 10) + cache.flush() + + // A run straddling the boundary is one row, as it will be once folded. + assertEquals(listOf(8L, 9L, 10L), seqs(cache.page(before = 11, limit = 2, rows = true))) + // Asking for more rows than the suffix holds is a short page, not a failure and not a + // claim that the conversation starts here. + assertEquals((5L..10L).toList(), seqs(cache.page(before = 11, limit = 40, rows = true))) + } + + @Test + fun the_floor_for_a_fetch_is_the_nearest_chunk_at_or_below_it() { + val cache = session() + write("1-10.rows.jsonl", (1L..9L).map { line(it) }) + write("10-40.rows.jsonl", (10L..39L).map { line(it) }) + cache.append(line(90), 90) + cache.flush() + + // The run behind the gap, which is what makes the fetched page adjacent to it: a page + // fetched before 90 with a floor of 39 stops at 40 and closes the gap exactly. + assertEquals(40L, cache.coveredUpTo(90)) + assertEquals(40L, cache.coveredUpTo(41)) + assertEquals(10L, cache.coveredUpTo(10)) + // Nothing at or below the oldest chunk's start, so the page is bounded only by its limit. + assertNull(cache.coveredUpTo(9)) + } + + @Test + fun a_newest_chunk_that_is_not_raw_discards_the_session() { + val cache = session() + write("1-10.rows.jsonl", (1L..9L).map { line(it) }) + + // Only reachable by dying between closing one live run and opening the next, and there is + // no cursor to be read off a coalesced line -- so the open is a cold one. + assertNull(cache.tail()) + assertFalse(dirOf().exists()) + } + + @Test + fun a_half_written_last_line_is_dropped_and_the_file_repaired() { + val cache = session() + dirOf().mkdirs() + File(dirOf(), "1-open.raw.jsonl").writeText(line(1) + "\n" + line(2) + "\n" + """{"se""") + + assertEquals(CachedTail(2, line(2)), cache.tail()) + assertEquals(line(1) + "\n" + line(2) + "\n", File(dirOf(), "1-open.raw.jsonl").readText()) + // And the run continues from where the good tail left off. + cache.append(line(3), 3) + cache.flush() + assertEquals(listOf(1L, 2L, 3L), seqs(cache.newest(80))) + } + + @Test + fun damage_anywhere_else_discards_the_session_when_a_read_reaches_it() { + val cache = session() + write("1-open.raw.jsonl", listOf(line(1), "not ours", line(3))) + + // Not seen by the tail, which reads the newest line and stops -- reading a chunk from its + // end is exactly not reading the rest of it, and that is what keeps a warm open cheap on + // a conversation of tens of megabytes. + assertEquals(CachedTail(3, line(3)), cache.tail()) + // Reached by a read that walks past it, and there is no honest way to say what a chunk + // covers with a line of it unreadable -- so what is served is nothing, and the session + // opens cold from here on. + assertEquals(emptyList(), cache.newest(80)) + assertFalse(dirOf().exists()) + assertTrue(said.any { it.contains("damaged") }) + } + + @Test + fun a_name_this_does_not_recognise_is_ignored() { + val cache = session() + write("notes.txt", listOf("hello")) + write("1-open.raw.jsonl", listOf(line(1))) + + assertEquals(CachedTail(1, line(1)), cache.tail()) + } + + @Test + fun a_chunk_larger_than_one_read_block_is_walked_across_the_boundaries() { + val cache = session() + // Well past the 64 kB block the backwards reader takes at a time, so a page has to be + // stitched across several of them -- including a line that straddles a boundary, which + // is the case nothing else here would notice going wrong. + val padding = "x".repeat(300) + val lines = (1L..500L).map { """{"seq":$it,"ts":1.5,"type":"toolStart","id":"$padding"}""" } + write("1-open.raw.jsonl", lines) + + assertEquals(500L, cache.tail()!!.seq) + assertEquals(lines.takeLast(80), cache.newest(80)) + assertEquals(lines.subList(0, 400), cache.page(before = 401, limit = 999, rows = false)) + // And a non-ASCII line, whose bytes a naive split could cut through a character. + val accented = """{"seq":501,"ts":1.5,"type":"assistantText","delta":"héllo — ok"}""" + cache.append(accented, 501) + cache.flush() + assertEquals(accented, cache.tail()!!.line) + } + + @Test + fun eviction_takes_the_least_recently_touched_and_never_the_one_on_screen() { + val cache = cache() + listOf("old", "middle", "open").forEachIndexed { at, id -> + write("1-open.raw.jsonl", List(50) { line(it + 1L) }, id = id) + dirOf(id).setLastModified(1_000_000L + at * 1000L) + } + val each = dirOf("old").walkTopDown().filter { it.isFile }.sumOf { it.length() } + + // Room for two of the three, so the oldest goes -- and the session being read never does, + // however long ago it was last touched. + cache.evictToBudget(keep = "open", budget = each * 2) + assertEquals(listOf("middle", "open"), File(temp, "v1/host_8443").list()!!.sorted()) + + cache.evictToBudget(keep = "open", budget = 0) + assertEquals(listOf("open"), File(temp, "v1/host_8443").list()!!.sorted()) + } + + @Test + fun retaining_deletes_exactly_the_sessions_the_server_no_longer_lists() { + val cache = cache() + listOf("a", "b", "c").forEach { write("1-open.raw.jsonl", listOf(line(1)), id = it) } + + cache.retainOnly(setOf("a", "c")) + assertEquals(listOf("a", "c"), File(temp, "v1/host_8443").list()!!.sorted()) + } + + @Test + fun size_and_purge_are_the_two_halves_of_the_reload_button() { + val cache = session() + assertEquals(0L, cache.bytes()) + (1L..5L).forEach { cache.append(line(it), it) } + cache.flush() + + assertTrue(cache.bytes() > 0) + cache.purge() + assertEquals(0L, cache.bytes()) + assertNull(cache.tail()) + // And the session is usable again straight afterwards, which is what a reload does next. + cache.append(line(9), 9) + cache.flush() + assertEquals(listOf(9L), seqs(cache.newest(80))) + } +} diff --git a/server/Cargo.toml b/server/Cargo.toml index 6c941ef..2e42672 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -23,7 +23,13 @@ tokio-stream = { version = "0.1", features = ["sync"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } serde = { version = "1", features = ["derive"] } -serde_json = "1" +# `float_roundtrip` because this server hands out the same transcript line two ways -- the +# `/transcript` page and the SSE backlog both parse it out of the file and serialize it again -- +# and serde_json's default float parser is not correctly rounded. Measured 2026-09-04: a `ts` of +# 1788546972.6030757 in the file came back as ...0755, so the two answers to "what is line 30" +# differed in the last bit while looking identical. What made that visible was the phone's +# transcript cache, which compares a line it already holds against the server's own answer. +serde_json = { version = "1", features = ["float_roundtrip"] } # The config file's format. Not JSON, because this file is written and read # by hand and RON says a sum type as syntax. The two house rules both # projects write it under live in wg-app-link; this is here for the error diff --git a/server/src/routes.rs b/server/src/routes.rs index f2b5c51..22e7dde 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -21,6 +21,9 @@ //! GET /sessions/{id}/events?after=N SSE: backlog after N, then live //! (a backlog past CATCH_UP_LIMIT arrives as a //! `reset` frame plus the newest window) +//! GET /sessions/{id}/transcript a page of history: ?before=N (newest when absent), +//! ?limit=N, ?coalesce=true to count rows not deltas, +//! ?after=N to floor it at what the caller already holds //! POST /sessions/{id}/message {text, attachmentIds?} //! (starts the process first if it has exited) //! POST /sessions/{id}/unqueue {messageId} -- take back one not read yet @@ -1647,6 +1650,11 @@ struct TranscriptQuery { /// counts events to reach a known seq. See `read_window`. #[serde(default)] coalesce: bool, + /// Return nothing at or below this seq; the page stops here instead of at `limit`. + /// The phone passes the end of what it already holds, so a page never overlaps it. + /// Named to match the SSE route's `after`, and exclusive in the same way. + #[serde(default)] + after: Option, } fn default_window() -> usize { @@ -1668,6 +1676,7 @@ async fn transcript( let events = crate::session::transcript::read_window( session.transcript_path(), query.before, + query.after, query.limit, query.coalesce, ) @@ -1679,6 +1688,7 @@ async fn transcript( tracing::debug!( session = %id, before = ?query.before, + after = ?query.after, limit = query.limit, got = events.len(), oldest = ?events.first().map(|entry| entry.seq), diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index 36a2d92..019ba04 100644 --- a/server/src/session/transcript.rs +++ b/server/src/session/transcript.rs @@ -148,9 +148,18 @@ 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. +/// +/// `after` is a floor: nothing at or below it is returned, and the page +/// stops there rather than at `limit`. A phone holding a cached run of the +/// transcript passes the end of what it already has, so the page it gets +/// back is exactly the gap and never overlaps its copy -- an overlap it +/// cannot store, since a coalesced event cannot be cut at a seq inside its +/// own delta run. A run cut by this floor is emitted as the partial it is, +/// exactly as one cut by `limit` already is. pub fn read_window( path: &Path, before: Option, + after: Option, limit: usize, coalesce: bool, ) -> Result> { @@ -161,15 +170,21 @@ pub fn read_window( Some(before) => indexed.first_at_or_after(before)?, None => indexed.lines.len(), }; + let start = match after { + Some(after) => indexed.first_at_or_after(after.saturating_add(1))?, + None => 0, + }; + // A floor above the window is an empty page, not a walk backwards past it. + let start = start.min(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) + indexed.parse_coalesced(start, end, limit) } else { - indexed.parse(end.saturating_sub(limit)..end) + indexed.parse(start.max(end.saturating_sub(limit))..end) } } @@ -350,8 +365,9 @@ impl<'a> Indexed<'a> { /// 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> { + /// boundary. `start` is the same kind of cut from the other end -- the floor `read_window`'s + /// `after` computes -- and a run reaching it is partial in the same way. + fn parse_coalesced(&self, start: usize, 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. @@ -369,7 +385,7 @@ impl<'a> Indexed<'a> { } }; let mut index = end; - while index > 0 { + while index > start { // 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. @@ -528,7 +544,7 @@ mod tests { } // No cursor is the newest page, which is what opening a session asks for. - let newest = read_window(&path, None, 3, false).expect("window"); + let newest = read_window(&path, None, None, 3, false).expect("window"); assert_eq!( newest.iter().map(|entry| entry.seq).collect::>(), [8, 9, 10] @@ -536,7 +552,7 @@ 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, false).expect("window"); + let older = read_window(&path, Some(8), None, 3, false).expect("window"); assert_eq!( older.iter().map(|entry| entry.seq).collect::>(), [5, 6, 7] @@ -544,24 +560,108 @@ mod tests { // Asking for more than there is gives what there is, rather than failing. assert_eq!( - read_window(&path, None, 100, false).expect("window").len(), + read_window(&path, None, 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, false) + read_window(&path, Some(1), None, 3, false) .expect("window") .is_empty() ); assert!( - read_window(&dir.path().join("nope.jsonl"), None, 3, false) + read_window(&dir.path().join("nope.jsonl"), None, None, 3, false) .expect("window") .is_empty() ); } + #[test] + fn a_floor_stops_a_page_at_what_the_caller_already_holds() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("transcript.jsonl"); + let mut transcript = Transcript::open(&path).expect("open"); + for n in 1..=10 { + transcript + .append(text(&n.to_string()), 0.0) + .expect("append"); + } + + // The floor is exclusive, like the SSE route's `after`, and it -- not the + // limit -- is what the page stops at. This is the gap between a phone's + // cached run and the window on its screen, fetched exactly. + let page = read_window(&path, Some(9), Some(5), 100, false).expect("window"); + assert_eq!( + page.iter().map(|entry| entry.seq).collect::>(), + [6, 7, 8] + ); + + // A limit smaller than the gap still bites; the floor is a bound, not a + // replacement for one. + let page = read_window(&path, Some(9), Some(2), 3, false).expect("window"); + assert_eq!( + page.iter().map(|entry| entry.seq).collect::>(), + [6, 7, 8] + ); + + // A floor at or above the window is an empty page, not a walk past it. + assert!( + read_window(&path, Some(4), Some(9), 10, false) + .expect("window") + .is_empty() + ); + } + + #[test] + fn a_floor_inside_a_delta_run_leaves_the_partial_run_it_cuts() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("transcript.jsonl"); + let mut transcript = Transcript::open(&path).expect("open"); + for d in ["a", "b", "c", "d"] { + transcript.append(text(d), 0.0).expect("append"); // seq 1..4 + } + transcript + .append( + Event::ToolStart { + id: "t".into(), + tool: "Bash".into(), + input: serde_json::Value::Null, + }, + 0.0, + ) + .expect("append"); // seq 5 + + // Cut inside the run: what comes back is the deltas above the floor, seq'd + // at the first of them -- the partial the phone's `healSplitMessage` welds + // onto the rest, the same as a run cut by the limit. + let rows = read_window(&path, Some(6), Some(2), 10, true).expect("window"); + assert_eq!(rows.len(), 2); + assert!(matches!( + &rows[0], + SeqEvent { seq: 3, event: Event::AssistantText { delta }, .. } if delta == "cd" + )); + assert!(matches!( + &rows[1], + SeqEvent { + seq: 5, + event: Event::ToolStart { .. }, + .. + } + )); + + // And with no floor the whole run is one row, as before. + let rows = read_window(&path, Some(6), None, 10, true).expect("window"); + assert_eq!(rows.len(), 2); + assert!(matches!( + &rows[0], + SeqEvent { seq: 1, event: Event::AssistantText { delta }, .. } if delta == "abcd" + )); + } + #[test] fn coalescing_counts_rows_and_joins_delta_runs() { let dir = tempfile::tempdir().expect("tempdir"); @@ -588,7 +688,7 @@ mod tests { // 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"); + let rows = read_window(&path, Some(8), None, 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. @@ -610,14 +710,43 @@ mod tests { )); // 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"); + let older = read_window(&path, Some(1), None, 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"); + let newest = read_window(&path, None, None, 2, true).expect("window"); assert_eq!(newest.iter().map(|e| e.seq).collect::>(), [6, 7]); } + #[test] + fn a_line_read_back_is_the_line_that_was_written() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("transcript.jsonl"); + // A timestamp with enough digits to be lost: the clock produces these all day, and this + // one is real (2026-09-04). serde_json's default float parser is not correctly rounded, + // so it read this back as ...0755 and every reader got a line one bit different from the + // one in the file -- while the SSE stream, which serializes the same struct, had already + // sent the original. Two answers to "what is line 1", indistinguishable by eye. + // + // Nothing on screen showed it: a `ts` is drawn as a relative time. What found it was the + // phone's transcript cache, which keeps the line it was sent and checks it against the + // server's own answer before resuming a stream from it -- so the mismatch turned into a + // cache thrown away and a transcript downloaded again, silently and only sometimes. The + // `float_roundtrip` feature in Cargo.toml is the fix; this is what keeps it. + let mut transcript = Transcript::open(&path).expect("open"); + transcript + .append(text("hello"), 1788546972.6030757) + .expect("append"); + drop(transcript); + + let written = std::fs::read_to_string(&path).expect("read"); + let entry = read_window(&path, None, None, 10, false).expect("window"); + assert_eq!( + serde_json::to_string(&entry[0]).expect("serialize"), + written.trim() + ); + } + #[test] fn a_missing_file_reads_as_empty() { let dir = tempfile::tempdir().expect("tempdir");