From a802522039d713e96209effd446812031d36be79 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Fri, 4 Sep 2026 15:00:25 -0400 Subject: [PATCH] Keep visited transcripts on the phone Reopening a session downloaded the conversation again, every time, over the tunnel. It now draws from a copy of what the server has already sent and asks for one event to check that copy is still current. Per session, under cacheDir, the server's own event lines in chunks named for the range they cover -- so a coalesced page, whose lines do not say what they cover, still records it. Only the contiguous run ending at the newest chunk is served; a gap is closed by paging through it, bounded by `after` on /transcript so the page stops where the phone's copy starts and can therefore be kept. Nothing is derived and stored: rows are a rendering, and a cache of them would need throwing away on every change to the fold. Nothing here is load-bearing. Missing, evicted, damaged or unwritable all degrade to the cold open this screen did before, and the check before the stream resumes -- one request, one event -- is what stops a replaced or truncated file being spliced onto a copy of a different conversation. What that check cannot see, a line changed mid-file with the tail intact, is what Reload in session settings is for. Measured on the emulator against ui-sandbox, on a 505-event session: reopening it costs one request for one event, including scrolling the whole conversation back; a cold open is two requests and 100 events. A reset after falling 300 behind fetched the gap as four coalesced rows rather than re-fetching 104 events and discarding them. Every chunk was checked line by line against what the server says for the range its name claims, across the reset and the gap-fill. transcript-bench.sh, same viewport content and gestures, before and after: p50 16.9ms both, p90 25.6 -> 23.2ms, p99 33.5 -> 36.7ms, and the transcript's own draw accounting 0.33ms -> 0.32ms with place 0.31ms either way. Within the emulator's noise, which is what a cache must be: it changes what is fetched, not what is drawn. Building it also found that the server handed out the same transcript line two different ways. serde_json's default float parser is not correctly rounded, so a ts written as ...0757 came back from /transcript as ...0755 while the SSE stream sent the original -- invisible on screen, since a ts is drawn as a relative time, and visible here only because the cache compares a line it holds against the server's answer. Fixed with float_roundtrip, with a test that fails the moment it is dropped. Co-Authored-By: Claude Opus 5 --- AGENTS.md | 48 ++ PLAN.md | 13 +- TRANSCRIPT_CACHE.md | 510 ++++++++++++++ .../src/main/kotlin/com/example/aiapp/Api.kt | 16 +- .../kotlin/com/example/aiapp/EventStream.kt | 11 +- .../kotlin/com/example/aiapp/ImportScreen.kt | 9 - .../com/example/aiapp/SessionListScreen.kt | 17 + .../kotlin/com/example/aiapp/SessionScreen.kt | 220 +++++-- .../example/aiapp/SessionSettingsDialog.kt | 52 +- .../main/kotlin/com/example/aiapp/Sizes.kt | 20 + .../com/example/aiapp/TranscriptCache.kt | 621 ++++++++++++++++++ .../com/example/aiapp/TranscriptItems.kt | 11 +- .../com/example/aiapp/TranscriptSource.kt | 183 ++++++ .../com/example/aiapp/TranscriptCacheTest.kt | 310 +++++++++ server/Cargo.toml | 8 +- server/src/routes.rs | 10 + server/src/session/transcript.rs | 155 ++++- 17 files changed, 2140 insertions(+), 74 deletions(-) create mode 100644 TRANSCRIPT_CACHE.md create mode 100644 app/androidApp/src/main/kotlin/com/example/aiapp/Sizes.kt create mode 100644 app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptCache.kt create mode 100644 app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptSource.kt create mode 100644 app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptCacheTest.kt 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");