diff --git a/android/README.md b/android/README.md index af45568..c78e777 100644 --- a/android/README.md +++ b/android/README.md @@ -46,15 +46,19 @@ manages the device fleet the claw keeps. - **Chat**: session list (`GET /v1/sessions`, interactive sessions first, the phone's own `phone:*` status keys hidden) and one chat screen per session. The applied events persist in a - per-session cache (DataStore, newest 500 with the full count as the - cursor), so a screen opens on its cached transcript at once and the - stream connects with `?since=` to receive only what is new - instead of replaying the whole history; the cache flushes at every - `turn_done` and when the screen closes. While a chat screen is open the - app holds one SSE connection (`GET /v1/sessions/{key}/events`, OkHttp - streaming GET with the auth header, hand-parsed frames: blank-line - delimited `data:` lines, `:` comment keep-alives skipped, unknown event - types skipped). The reader's own messages render as right-anchored + per-session cache (DataStore, the newest 500 events, the seq of the + last one as its cursor), so a screen opens on its cached transcript at + once and the stream connects with `?since=` to receive only the + gap instead of replaying the session again; the cache flushes at every + `turn_done` and when the screen closes. A screen with no cache opens on + the claw's newest turn and walks back through `GET + /v1/sessions/{key}/history` as the reader scrolls up, one whole turn + per page. While a chat screen is open the app holds one SSE connection + (`GET /v1/sessions/{key}/events`, OkHttp streaming GET with the auth + header, hand-parsed frames: blank-line delimited `data:` lines with the + event's seq on the `id:` line, `:` comment keep-alives skipped, the + claw's `reset` comment dropping the mirror, unknown event types + skipped). The reader's own messages render as right-anchored bubbles (elevated surface, accent bar on the trailing edge); the agent's turns keep the avatar on the left. The composer's (+) door stages images for the next send — a photo-picker @@ -215,7 +219,8 @@ All under one base URL; `Authorization: Bearer ` on `/v1`, never on | `PUT /v1/providers/active` | switch the backend instance, persisted | | `GET /v1/sessions` | session list | | `POST /v1/sessions/{key}/messages` | `{"text"}`, optional `images` (≤ 8, ≤ 5 MiB decoded each, bare base64) → 202; reply via stream | -| `GET /v1/sessions/{key}/events` | SSE, replay-then-live: `user_message`, `thinking_delta`, `text_delta`, `tool_start`, `tool_result`, `turn_done`, `attachment`, `consent_request`, `consent_resolved`. `?since=N` skips the replay's oldest N events — the transcript cache's cursor | +| `GET /v1/sessions/{key}/events` | SSE, replay-then-live: `user_message`, `thinking_delta`, `text_delta`, `tool_start`, `tool_result`, `turn_done`, `attachment`, `consent_request`, `consent_resolved`. Every frame carries the record's seq on its `id:` line; `?since=` replays only the events past it, and no cursor replays the newest turn only — the transcript cache's cursor | +| `GET /v1/sessions/{key}/history?before=&limit=` | one page of whole turns older than `before`, ending exactly there, with `more` saying whether anything precedes it — the chat screen's walk back | | `GET /v1/attachments/{id}` | an attachment event's bytes (bearer) | | `GET /v1/consents` | pending consent questions, oldest first | | `POST /v1/consents/{id}` | answer one: `{"secret"}` or `{"grant"}` | @@ -261,11 +266,16 @@ posts for that topic. notifications are as fresh as the WorkManager interval (15 min). The chat screen is always live via SSE. - **Transcript cache**: the applied event log persists per session (the - newest 500 events plus the full count), so a screen opens on its cached + newest 500 events plus their seqs), so a screen opens on its cached transcript and the stream resumes past it. The cache is a cursor over the claw's record, not a copy of it: a claw database wiped behind the phone's back leaves the cache pointing past a history that no longer - exists, and that session replays nothing until it catches up. + exists, which the claw answers with a `reset` comment and the newest + turn — the screen drops its stale rows and rebuilds from those. +- **Older turns are fetched, not cached**: the screen holds the cached + window plus whatever the reader scrolled back to; reopening a session + starts from the cache again, and the walk back re-reads the record. + Nothing older is persisted, so the cache stays a fixed size. - **First launch**: nothing works until Settings validates `/healthz`. ## One-consumer principle diff --git a/android/app/src/main/java/rs/rhema/phone/chat/ChatController.kt b/android/app/src/main/java/rs/rhema/phone/chat/ChatController.kt index 21c45c2..a115f96 100644 --- a/android/app/src/main/java/rs/rhema/phone/chat/ChatController.kt +++ b/android/app/src/main/java/rs/rhema/phone/chat/ChatController.kt @@ -34,8 +34,10 @@ const val MAX_IMAGE_BYTES = 5L * 1024 * 1024 // seeded from the session cache before the first frame and kept in the // cache as events stream, so opening a session paints it at once and the // claw is asked only for the events past the cache (the stream's `since` -// cursor). Rotation re-seeds from the cache; process death no longer -// costs the transcript. +// cursor, which is the seq of the last event applied). A screen with no +// cache opens on the newest turn and walks back through the history +// route as the reader scrolls up. Rotation re-seeds from the cache; +// process death no longer costs the transcript. class ChatController( private val api: ClawApi, private val cache: SessionCache, @@ -101,6 +103,11 @@ class ChatController( // reply carries the text, this carries the outcome. data class Outcome(override val id: Long, val outcome: String) : Item + // One applied event with the seq the claw gave it: the cache window + // needs the seqs to resume from, and a lagged drop leaves a gap the + // window's arithmetic must not paper over. + private data class Applied(val seq: Long?, val event: ClawEvent) + sealed interface Conn { data object Connecting : Conn data object Live : Conn @@ -120,11 +127,24 @@ class ChatController( private var nextId = 0L private fun newId() = nextId++ - // The applied event log: the cache's tail plus what this connection - // delivered, and [appliedTotal] its count from the session's first - // event — the cursor every (re)connection opens with. - private var applied: MutableList = mutableListOf() - private var appliedTotal = 0L + // The applied event log the cache window is taken from: the cached + // tail plus what this connection delivered. Pages folded in above it + // by [loadOlder] are transcript rows only — the walk back re-reads + // them, so they never bloat what is persisted. + private var applied: MutableList = mutableListOf() + // The seq of the newest event applied: the cursor every + // (re)connection opens with. -1 asks for the tail. + private var cursor = -1L + // The seq of the oldest event on screen: the anchor [loadOlder] + // ends its page at. -1 when the screen holds nothing. + private var oldestSeq = -1L + + // The walk back through history: whether a page is in flight, and + // whether any more exist below what the screen holds. + var loadingOlder by mutableStateOf(false) + private set + var moreOlder by mutableStateOf(false) + private set private var streamJob: Job? = null @@ -134,18 +154,30 @@ class ChatController( // Seed before the first frame: the screen shows the cached // transcript at once, and the stream opens past it. val cached = cache.get(sessionKey) - appliedTotal = cached.total - applied = cached.events.toMutableList() - items = applyEvents(emptyList(), applied) + cursor = cached.resumeCursor + oldestSeq = cached.oldestSeq + applied = cached.events.map { Applied(null, it) }.toMutableList() + items = applyEvents(emptyList(), cached.events) + // A cache holds a window, not the record: the walk back has + // somewhere to go unless the session held nothing. (A seed + // from an empty cache leaves this false; the tail's first + // event re-arms it — see [record].) + moreOlder = items.isNotEmpty() try { while (isActive) { - api.eventStream(sessionKey, appliedTotal).collect { frame -> + api.eventStream(sessionKey, cursor).collect { frame -> when (frame) { // The replay was cut at the cursor, so nothing // already on screen arrives twice; the feed just - // continues from where the cache left off. + // continues from where the cache left off. With + // no cursor it is the newest turn — the screen + // paints it and the reader walks back. SseFrame.Opened -> connection = Conn.Live - is SseFrame.Event -> record(frame.event) + is SseFrame.Event -> record(frame.event, frame.seq) + // The cursor named events the claw no longer + // holds: drop the mirror, take the tail that + // follows. + SseFrame.Reset -> resetMirror() SseFrame.Closed -> Unit is SseFrame.Failed -> connection = Conn.Failed(frame.code, frame.message) } @@ -166,16 +198,79 @@ class ChatController( // Appends one delivered event to the log and the transcript. The // turn's end persists the cache — once per turn, not per delta. - private suspend fun record(event: ClawEvent) { - applied.add(event) - appliedTotal += 1 + private suspend fun record(event: ClawEvent, seq: Long?) { + if (seq != null) cursor = seq + if (seq != null && oldestSeq < 0) { + // The first event on an empty mirror: this is the tail the + // claw opened with, and the record may hold more above it — + // only a page that comes back empty (or `more` false) says + // otherwise. + oldestSeq = seq + moreOlder = true + } + applied.add(Applied(seq, event)) items = applyEvent(items, event) if (event is ClawEvent.TurnDone) persist() } + // The claw no longer holds what this screen's cursor named: the + // transcript goes, and the tail that follows the notice rebuilds it. + private suspend fun resetMirror() { + applied = mutableListOf() + cursor = -1 + oldestSeq = -1 + items = emptyList() + loadingOlder = false + moreOlder = false + cache.put(sessionKey, CachedSession()) + } + + // One page of turns older than the screen's oldest, appended above it. + // Whole turns only, so the page never duplicates or strands a row; the + // page ends exactly where the screen begins, and [moreOlder] says + // whether another exists. A page folded from empty goes through the + // same reducer the stream uses, so a replayed consent or attachment + // card folds exactly as it would have live. + suspend fun loadOlder() { + if (loadingOlder || !moreOlder || oldestSeq < 0) return + loadingOlder = true + try { + when (val result = api.history(sessionKey, oldestSeq, HISTORY_PAGE)) { + is ClawResult.Ok -> { + val page = result.value.events.mapNotNull { entry -> + entry.event?.let { entry.seq to it } + } + val first = page.firstOrNull() ?: return + // The page is older than everything on screen, and the + // list runs oldest first: it goes above, never below — + // appended, it would land at the newest end. + items = applyEvents(emptyList(), page.map { it.second }) + items + oldestSeq = first.first + moreOlder = result.value.more + } + is ClawResult.HttpError, is ClawResult.Network -> Unit + } + } finally { + loadingOlder = false + } + } + private suspend fun persist() { + // The window starts on a turn boundary: a page cut mid-turn would + // paint a reply with no question above it on the next open, and + // the walk back from the window's first seq would then have to + // start inside a turn instead of at its top. val window = applied.takeLast(SESSION_WINDOW) - cache.put(sessionKey, CachedSession(total = appliedTotal, events = window)) + val start = window.indexOfFirst { it.event is ClawEvent.UserMessage }.takeIf { it > 0 } ?: 0 + val kept = window.drop(start) + cache.put( + sessionKey, + CachedSession( + cursor = cursor, + first = kept.firstOrNull()?.seq ?: -1, + events = kept.map { it.event }, + ), + ) } fun stop() { @@ -381,5 +476,9 @@ class ChatController( private companion object { const val RECONNECT_DELAY_MS = 2_000L + + // The most events one history page may hold: the claw widens the + // page back to a turn boundary, so this is a floor, not a cap. + const val HISTORY_PAGE = 400 } } diff --git a/android/app/src/main/java/rs/rhema/phone/data/SessionCache.kt b/android/app/src/main/java/rs/rhema/phone/data/SessionCache.kt index 5bb8159..690eef1 100644 --- a/android/app/src/main/java/rs/rhema/phone/data/SessionCache.kt +++ b/android/app/src/main/java/rs/rhema/phone/data/SessionCache.kt @@ -12,17 +12,39 @@ import rs.rhema.phone.net.ClawEvent // One session's cached transcript tail: the events this device has // applied, persisted as they stream. Opening a session seeds the screen -// from here instead of watching the claw's whole replay again, and the -// stream opens with `total` as its `since` cursor so the claw sends only +// from here instead of watching the claw replay the session again, and +// the stream opens with [cursor] as its `since` so the claw sends only // the events past the cache. @Serializable data class CachedSession( - // The session's full event count — the `since` cursor. Can exceed - // [events]: the list keeps only the newest [SESSION_WINDOW] events, - // the count counts from the session's first. - val total: Long = 0, + // The seq of the newest event in [events] — the `since` cursor. -1 + // when nothing is cached. + val cursor: Long = -1, + // The seq of the oldest event in [events]: the anchor the walk back + // through /history ends at. + val first: Long = -1, val events: List = emptyList(), -) + // The cursor of a cache written before the stream named seqs: a + // count of the session's events. Read only to migrate one, see + // [resumeCursor]; -1 for every cache written since. + val total: Long = -1, +) { + // A cache from before seqs still resumes exactly: the record numbers + // events densely from zero, so the last one it applied is `total - 1`, + // and the window above it is contiguous. The first open after the + // upgrade costs nothing. + val resumeCursor: Long + get() = if (cursor >= 0) cursor else if (total > 0) total - 1 else -1 + + val oldestSeq: Long + get() = if (first >= 0) { + first + } else if (events.isNotEmpty() && resumeCursor >= 0) { + resumeCursor - events.size + 1 + } else { + -1 + } +} // How many events the cache keeps per session. A resident session runs // for weeks; a bounded window keeps the DataStore write flat, and a tail diff --git a/android/app/src/main/java/rs/rhema/phone/net/ClawApi.kt b/android/app/src/main/java/rs/rhema/phone/net/ClawApi.kt index 653807c..62a2a27 100644 --- a/android/app/src/main/java/rs/rhema/phone/net/ClawApi.kt +++ b/android/app/src/main/java/rs/rhema/phone/net/ClawApi.kt @@ -32,7 +32,16 @@ sealed interface ClawResult { sealed interface SseFrame { data object Opened : SseFrame - data class Event(val event: ClawEvent) : SseFrame + + // seq is the record's number for the event, from the frame's `id:` + // line: the cursor a reconnect names as `since`. Null only when the + // frame carried no id. + data class Event(val event: ClawEvent, val seq: Long?) : SseFrame + + // The claw no longer holds the events our cursor named (a wiped or + // reset database): the mirror is invalid, and the tail that follows + // this notice is what replaces it. + data object Reset : SseFrame // code != null carries an HTTP status; 401 means the token is wrong. data class Failed(val code: Int?, val message: String) : SseFrame @@ -263,18 +272,20 @@ class ClawApi( return execute(builder.post(body).build()) { } } - // GET /v1/sessions/{key}/events?since=N: the claw replays the - // session's record minus its oldest N events, then goes live. `since` - // is the transcript cache's cursor — how many events this device has - // already applied — so a reopening screen receives only what is new - // instead of the whole history. Built by hand like notifications(): - // v1() bakes the URL into the builder, leaving no room for a query. + // GET /v1/sessions/{key}/events?since=N: the claw replays the events + // past seq N, then goes live. `since` is the transcript cache's + // cursor — the seq of the last event this device applied — so a + // reopening screen receives only the gap instead of the whole + // history. A negative `since` asks for nothing: the claw answers + // with the newest turn, which is what a screen with no cache opens + // on. Built by hand like notifications(): v1() bakes the URL into + // the builder, leaving no room for a query. fun eventStream(sessionKey: String, since: Long): Flow = callbackFlow { val url = baseOrNull()?.toHttpUrlOrNull()?.newBuilder() ?.addPathSegments("v1/sessions") ?.addPathSegment(sessionKey) ?.addPathSegment("events") - ?.apply { if (since > 0) addQueryParameter("since", since.toString()) } + ?.apply { if (since >= 0) addQueryParameter("since", since.toString()) } ?.build() if (url == null) { trySend(SseFrame.Failed(null, "no base URL configured")) @@ -311,27 +322,38 @@ class ClawApi( close() return } - // SSE framing per contract: blank-line-delimited data lines, - // ":" comment lines are keep-alives. No event:/id: fields. + // SSE framing per contract: blank-line-delimited + // frames, `data:` the payload, `id:` the event's seq, + // ":" comment lines the keep-alives (plus the claw's + // lag and reset notices). `event:` lines do not occur. val data = StringBuilder() + var id: String? = null try { while (true) { val line = source.readUtf8Line() ?: break when { - line.startsWith(":") -> Unit + line.startsWith(":") -> { + if (line.removePrefix(":").trimStart().startsWith("reset")) { + trySend(SseFrame.Reset) + } + } line.isEmpty() -> { if (data.isNotEmpty()) { val payload = data.toString().trim() + val seq = id?.trim()?.toLongOrNull() + id = null data.setLength(0) runCatching { json.decodeFromString(payload) } .getOrNull() - ?.let { trySend(SseFrame.Event(it)) } + ?.let { trySend(SseFrame.Event(it, seq)) } } } line.startsWith("data:") -> { data.append(line.removePrefix("data:").trimStart()) data.append('\n') } + line.startsWith("id:") -> + id = line.removePrefix("id:").trim() } } trySend(SseFrame.Closed) @@ -346,6 +368,25 @@ class ClawApi( awaitClose { call.cancel() } }.buffer(Channel.UNLIMITED) + // GET /v1/sessions/{key}/history?before=&limit=: one page of + // whole turns older than the seq this device holds oldest, for the + // walk back through a session's history. `more` false means the page + // reached the session's first event. Built by hand like + // eventStream(): the query is not expressible through v1(). + suspend fun history(sessionKey: String, before: Long, limit: Int): ClawResult { + val base = baseOrNull() ?: return ClawResult.Network("no base URL configured") + val url = base.toHttpUrlOrNull()?.newBuilder() + ?.addPathSegments("v1/sessions") + ?.addPathSegment(sessionKey) + ?.addPathSegment("history") + ?.addQueryParameter("before", before.toString()) + ?.addQueryParameter("limit", limit.toString()) + ?.build() ?: return ClawResult.Network("base URL is not a URL") + return execute(Request.Builder().url(url).header("Authorization", bearer()).build()) { response -> + json.decodeFromString(response.body?.string() ?: "{}") + } + } + private suspend fun baseOrNull(): String? = settings.current().baseUrl.trim().trimEnd('/').takeIf { it.isNotBlank() } diff --git a/android/app/src/main/java/rs/rhema/phone/net/Models.kt b/android/app/src/main/java/rs/rhema/phone/net/Models.kt index c20fed4..29e86d7 100644 --- a/android/app/src/main/java/rs/rhema/phone/net/Models.kt +++ b/android/app/src/main/java/rs/rhema/phone/net/Models.kt @@ -117,9 +117,21 @@ data class RhemaNotification( @Serializable data class UserImage(val id: String, @SerialName("media_type") val mediaType: String) +// One event of GET /v1/sessions/{key}/history: the record's seq for it, +// beside the event itself. Seq order is the wire order. +@Serializable +data class HistoryEvent(val seq: Long = -1, val event: ClawEvent? = null) + +// One page of whole turns older than the seq the screen holds oldest +// (`before`). `more` false means the page reached the session's first +// event, so the walk back can stop. +@Serializable +data class HistoryPage(val events: List = emptyList(), val more: Boolean = false) + // Frames of GET /v1/sessions/{key}/events; "type" is the discriminator. -// The stream opens with the session's recorded turns replayed before the -// live feed, so a screen (re)builds its transcript from the replay. +// Every frame also names the record's seq in its `id:` line, which is the +// stream's resume cursor. A screen with no cache opens on the newest turn +// and walks back through /history; one with a cache resumes past it. // Unknown types fail the per-frame parse and are skipped, so the API can grow // without breaking older clients. @Serializable diff --git a/android/app/src/main/java/rs/rhema/phone/ui/ChatScreen.kt b/android/app/src/main/java/rs/rhema/phone/ui/ChatScreen.kt index c8462c2..8692b04 100644 --- a/android/app/src/main/java/rs/rhema/phone/ui/ChatScreen.kt +++ b/android/app/src/main/java/rs/rhema/phone/ui/ChatScreen.kt @@ -31,6 +31,7 @@ import androidx.compose.foundation.layout.size import androidx.compose.foundation.layout.widthIn import androidx.compose.foundation.lazy.LazyColumn import androidx.compose.foundation.lazy.items +import androidx.compose.foundation.lazy.rememberLazyListState import androidx.compose.foundation.shape.CircleShape import androidx.compose.material.icons.Icons import androidx.compose.material.icons.automirrored.filled.ArrowBack @@ -58,6 +59,7 @@ import androidx.compose.runtime.remember import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.runtime.saveable.rememberSaveable import androidx.compose.runtime.setValue +import androidx.compose.runtime.snapshotFlow import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.draw.clip @@ -81,6 +83,7 @@ import rs.rhema.phone.data.AgentIdentityCache import rs.rhema.phone.di.AppContainer import rs.rhema.phone.net.ClawResult import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.launch import kotlinx.coroutines.withContext import java.io.File @@ -178,7 +181,25 @@ fun ChatScreen(container: AppContainer, sessionKey: String, onBack: () -> Unit) // bottom edge. A streaming reply grows against that anchor, so // the newest text stays on screen without scroll bookkeeping, // and a reader scrolled into history is never yanked back. + val listState = rememberLazyListState() + // Older turns are fetched, not cached: the claw holds the record + // and the screen opens on its newest turn, so the walk back + // starts when the reader reaches the top (higher indices are + // older here) or when the transcript is too short to fill the + // screen — which reads as "the last visible item is the last + // item" either way. + LaunchedEffect(listState, controller) { + snapshotFlow { + val info = listState.layoutInfo + info.visibleItemsInfo.lastOrNull()?.index to info.totalItemsCount + } + .distinctUntilChanged() + .collect { (last, total) -> + if (last != null && total > 0 && last >= total - 2) controller.loadOlder() + } + } LazyColumn( + state = listState, modifier = Modifier .weight(1f) .fillMaxWidth(), @@ -189,6 +210,18 @@ fun ChatScreen(container: AppContainer, sessionKey: String, onBack: () -> Unit) items(turnGroups(controller.items).asReversed(), key = { it.key }) { group -> TurnGroupRow(group, container, controller, face) } + // The last declared item sits at the top of a reversed list. + if (controller.loadingOlder) { + item(key = "older") { + Text( + stringResource(R.string.chat_loading_earlier), + style = MaterialTheme.typography.labelSmall, + fontFamily = JadeiteMono, + color = JadeiteText3, + modifier = Modifier.fillMaxWidth(), + ) + } + } } controller.sendError?.let { err -> @@ -678,11 +711,19 @@ private fun ReplyBlock(item: ChatController.Reply) { @Composable private fun OutcomeTag(outcome: String) { + // A steered turn was cut short by the next message, not failed: the + // reply that replaced it streams right below, so the tag reads as an + // accent, not an error — the same distinction the web UI draws. + val color = when { + outcome.equals("Failed", ignoreCase = true) -> JadeiteError + outcome.equals("Steered", ignoreCase = true) -> JadeiteAccentText + else -> JadeiteText3 + } Text( outcome, style = MaterialTheme.typography.labelSmall, fontFamily = JadeiteMono, - color = if (outcome.equals("Failed", ignoreCase = true)) JadeiteError else JadeiteText3, + color = color, ) } diff --git a/android/app/src/main/res/values/strings.xml b/android/app/src/main/res/values/strings.xml index 300754d..4e85894 100644 --- a/android/app/src/main/res/values/strings.xml +++ b/android/app/src/main/res/values/strings.xml @@ -35,6 +35,7 @@ tool running: %1$s tool: %1$s Download and open + loading earlier turns downloading… Open No app on this phone opens this file type.