feat(android): resume from the seq, and walk back through the session

The chat screen asked the claw for everything the cache did not hold by
counting events, and a session that ran for weeks cost the whole record
the first time it opened here. The stream now names each frame's seq and
the cache stores it, so a reopening screen sends since=<seq> and receives
only the gap. The cache migrates in place: the old count of events is
exactly the last seq plus one, because the record numbers densely.

A screen with no cache opens on the newest turn and walks back through
the new history route when the reader reaches the top, one whole turn
per page, so the phone does not hold a weeks-long transcript to show the
last few messages. Pages are folded through the same reducer the stream
uses and prepended, since the list runs oldest first: appended, they
would land at the newest end. A reset comment (the record no longer
holds the cursor) drops the mirror, and the tail that follows rebuilds
it. The outcome tag colours a steered turn as an accent rather than a
failure, matching the web UI.
This commit is contained in:
vhaudiquet committed 2026-11-08 18:01:52 +01:00
1 parent 70d9bb0d34
commit fa0e3d4470
7 files changed
+277 -51

No files matched your search

+22 -12
View File
@@ -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=<count>` 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=<seq>` 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 <token>` 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=<seq>` 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
@@ -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<ClawEvent> = 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<Applied> = 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
}
}
@@ -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<ClawEvent> = 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
@@ -32,7 +32,16 @@ sealed interface ClawResult<out T> {
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<SseFrame> = 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<ClawEvent>(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=<seq>&limit=<n>: 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<HistoryPage> {
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<HistoryPage>(response.body?.string() ?: "{}")
}
}
private suspend fun baseOrNull(): String? =
settings.current().baseUrl.trim().trimEnd('/').takeIf { it.isNotBlank() }
@@ -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<HistoryEvent> = 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
@@ -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,
)
}
@@ -35,6 +35,7 @@
<string name="chat_tool_running">tool running: %1$s</string>
<string name="chat_tool_done">tool: %1$s</string>
<string name="chat_attachment_download">Download and open</string>
<string name="chat_loading_earlier">loading earlier turns</string>
<string name="chat_attachment_downloading">downloading…</string>
<string name="chat_attachment_open">Open</string>
<string name="chat_attachment_open_failed">No app on this phone opens this file type.</string>