From 99df45e99af94f3bc06fd363a33070b8c96fa9c7 Mon Sep 17 00:00:00 2001 From: weishu Date: Mon, 17 Aug 2026 15:50:47 +0800 Subject: [PATCH] feat(android): message window store port, pagination fixtures green (B-M2c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit :core:protocol window/ — pure state machine ported function-for-function from web/src/lib/message-window-store.ts + messages.ts: merge-by-(id|localId) with optimistic echo reconciliation, position ordering (invokedAt ?? createdAt, seq, ASCII id tie-break), trim-preserving-queued with the codex agent-run budget, epoch reset handling, latest-replace with request-baseline identity preservation, consumed/cancelled/queued-reconcile transitions, tail/history modes, hydrate/persist shapes. MessageRetention ports the null-decision tree of normalizeDecryptedMessage (dedup with the B-M2a pipeline port flagged). :core:data store/ — per-session MessageWindowStore (StateFlow + Mutex, single-flight tail sync with trailing drain, fetchOlder with epoch-reset resync, SSE ingest hooks, optimistic sends, queued-state reconciliation, seedFrom for resume id changes) + WindowSnapshots (atomic JSON files, LRU 10; JsonSnapshotStore dedup TODO) + MessageWindowStores registry. Minimal MessagesApi interface (sealed MessagesQuery) extracted over the two message endpoints, implemented by HapiApi. Gate: PaginationFixtureTest replays all 11 shared/fixtures/pagination scripts against the real store — requests, older-load outcomes, reconcile candidates and the final window projection all exact — 11/11 green; plus targeted concurrency/snapshot/seed unit tests. :app:assembleDebug green. --- android/core/data/build.gradle.kts | 12 + .../main/kotlin/app/hapi/data/api/HapiApi.kt | 24 +- .../kotlin/app/hapi/data/api/MessagesApi.kt | 51 ++ .../app/hapi/data/store/MessageWindowStore.kt | 582 ++++++++++++++++ .../hapi/data/store/MessageWindowStores.kt | 59 ++ .../app/hapi/data/store/WindowSnapshots.kt | 90 +++ .../hapi/data/store/MessageWindowStoreTest.kt | 332 +++++++++ .../hapi/data/store/PaginationFixtureTest.kt | 314 +++++++++ .../app/hapi/protocol/window/MessageMerge.kt | 129 ++++ .../hapi/protocol/window/MessageRetention.kt | 213 ++++++ .../protocol/window/MessageWindowLogic.kt | 659 ++++++++++++++++++ .../protocol/window/MessageWindowState.kt | 113 +++ .../app/hapi/protocol/window/WindowMessage.kt | 160 +++++ .../app/hapi/protocol/wire/MessageFlags.kt | 23 + .../hapi/protocol/window/MessageMergeTest.kt | 121 ++++ 15 files changed, 2880 insertions(+), 2 deletions(-) create mode 100644 android/core/data/src/main/kotlin/app/hapi/data/api/MessagesApi.kt create mode 100644 android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStore.kt create mode 100644 android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStores.kt create mode 100644 android/core/data/src/main/kotlin/app/hapi/data/store/WindowSnapshots.kt create mode 100644 android/core/data/src/test/kotlin/app/hapi/data/store/MessageWindowStoreTest.kt create mode 100644 android/core/data/src/test/kotlin/app/hapi/data/store/PaginationFixtureTest.kt create mode 100644 android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageMerge.kt create mode 100644 android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageRetention.kt create mode 100644 android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowLogic.kt create mode 100644 android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowState.kt create mode 100644 android/core/protocol/src/main/kotlin/app/hapi/protocol/window/WindowMessage.kt create mode 100644 android/core/protocol/src/main/kotlin/app/hapi/protocol/wire/MessageFlags.kt create mode 100644 android/core/protocol/src/test/kotlin/app/hapi/protocol/window/MessageMergeTest.kt diff --git a/android/core/data/build.gradle.kts b/android/core/data/build.gradle.kts index b00fe0e3..817cb389 100644 --- a/android/core/data/build.gradle.kts +++ b/android/core/data/build.gradle.kts @@ -19,6 +19,18 @@ android { sourceCompatibility = JavaVersion.VERSION_17 targetCompatibility = JavaVersion.VERSION_17 } + + testOptions { + unitTests.all { test -> + // Golden fixtures live at the repo root (outside the Gradle root + // dir); the pagination conformance suite replays them against the + // real MessageWindowStore (same wiring as :core:protocol). + test.systemProperty( + "hapi.fixtures.dir", + rootDir.parentFile.resolve("shared/fixtures").absolutePath, + ) + } + } } kotlin { diff --git a/android/core/data/src/main/kotlin/app/hapi/data/api/HapiApi.kt b/android/core/data/src/main/kotlin/app/hapi/data/api/HapiApi.kt index 97768be3..9d2aac86 100644 --- a/android/core/data/src/main/kotlin/app/hapi/data/api/HapiApi.kt +++ b/android/core/data/src/main/kotlin/app/hapi/data/api/HapiApi.kt @@ -93,7 +93,7 @@ class HapiApi( private val client: OkHttpClient, private val imageClient: OkHttpClient = client, private val authClient: OkHttpClient = client, -) { +) : MessagesApi { /** Normalized hub origin this instance talks to. */ val hubUrl: String = HubUrls.normalize(hubUrl) ?: throw IllegalArgumentException("Invalid hub URL: $hubUrl") @@ -193,6 +193,26 @@ class HapiApi( return request("GET", target.build()) } + /** [MessagesApi] seam over [getMessages] — the window store's transport. */ + override suspend fun getMessages(sessionId: String, query: MessagesQuery): MessagesResponse = when (query) { + is MessagesQuery.Latest -> getMessages(sessionId, limit = query.limit) + is MessagesQuery.Before -> getMessages( + sessionId, + limit = query.limit, + beforeSeq = query.beforeSeq, + beforeAt = query.beforeAt, + ) + is MessagesQuery.After -> getMessages( + sessionId, + limit = query.limit, + afterSeq = query.afterSeq, + afterAt = query.afterAt, + untilSeq = query.untilSeq, + untilAt = query.untilAt, + epoch = query.epoch, + ) + } + /** * `POST /api/sessions/:id/messages` — responds `{ok: true}` only; the * message itself arrives via SSE (`message-received`), reconciled by @@ -211,7 +231,7 @@ class HapiApi( request("POST", url("api", "sessions", sessionId, "messages", messageId, "steer").build(), EMPTY_JSON) /** `POST /api/sessions/:id/messages/queued-state` — resync optimistic sends after reconnect. */ - suspend fun getQueuedState(sessionId: String, localIds: List): QueuedStateResponse = + override suspend fun getQueuedState(sessionId: String, localIds: List): QueuedStateResponse = request( "POST", url("api", "sessions", sessionId, "messages", "queued-state").build(), diff --git a/android/core/data/src/main/kotlin/app/hapi/data/api/MessagesApi.kt b/android/core/data/src/main/kotlin/app/hapi/data/api/MessagesApi.kt new file mode 100644 index 00000000..84624792 --- /dev/null +++ b/android/core/data/src/main/kotlin/app/hapi/data/api/MessagesApi.kt @@ -0,0 +1,51 @@ +package app.hapi.data.api + +import app.hapi.protocol.wire.MessagesResponse +import app.hapi.protocol.wire.QueuedStateResponse + +/** + * One `GET /api/sessions/:id/messages` request. The three variants are the + * only shapes the reference client ever issues + * (`docs/api/client-contract/pagination.md`): the store cannot produce an + * invalid parameter combination by construction, and the pagination fixture + * harness asserts these against the recorded `expectedRequests` — `After` + * always carries the `until` keys (explicitly null on the first catch-up + * request), the other variants never do. + */ +sealed interface MessagesQuery { + val limit: Int + + /** Newest page, no cursor. */ + data class Latest(override val limit: Int) : MessagesQuery + + /** Rows strictly older than the compound cursor. */ + data class Before( + val beforeAt: Long, + val beforeSeq: Long, + override val limit: Int, + ) : MessagesQuery + + /** Tail catch-up: rows strictly newer than the cursor, snapshot-bounded. */ + data class After( + val afterAt: Long, + val afterSeq: Long, + val untilAt: Long?, + val untilSeq: Long?, + val epoch: Long, + override val limit: Int, + ) : MessagesQuery +} + +/** + * Minimal seam over the two message endpoints the window store drives — + * implemented by [HapiApi] in production and by a scripted fake in the + * pagination fixture harness. Extracted (instead of depending on the concrete + * [HapiApi]) so the REAL orchestration is what the fixtures replay. + */ +interface MessagesApi { + /** `GET /api/sessions/:id/messages`. */ + suspend fun getMessages(sessionId: String, query: MessagesQuery): MessagesResponse + + /** `POST /api/sessions/:id/messages/queued-state`. */ + suspend fun getQueuedState(sessionId: String, localIds: List): QueuedStateResponse +} diff --git a/android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStore.kt b/android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStore.kt new file mode 100644 index 00000000..97ec8740 --- /dev/null +++ b/android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStore.kt @@ -0,0 +1,582 @@ +package app.hapi.data.store + +import app.hapi.data.api.MessagesApi +import app.hapi.data.api.MessagesQuery +import app.hapi.protocol.window.MessagePosition +import app.hapi.protocol.window.MessageStatus +import app.hapi.protocol.window.MessageViewMode +import app.hapi.protocol.window.MessageWindowLogic +import app.hapi.protocol.window.MessageWindowState +import app.hapi.protocol.window.OlderLoadOutcome +import app.hapi.protocol.window.PAGE_SIZE +import app.hapi.protocol.window.WindowMessage +import app.hapi.protocol.window.asWindowMessage +import app.hapi.protocol.window.buildOptimisticMessage +import app.hapi.protocol.wire.AttachmentMetadata +import app.hapi.protocol.wire.MessagesResponse +import app.hapi.protocol.wire.SyncEvent +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.async +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** What the chat screen renders — mirror of the web `useMessages` return shape. */ +data class MessageWindowUiState( + val messages: List, + val hasMore: Boolean, + val isSyncingTail: Boolean, + val isLoadingMore: Boolean, + val warning: String?, + val viewMode: MessageViewMode, + val messagesVersion: Long, + val historyVersion: Long, + val tailRevision: Long, +) + +/** + * Per-session message window orchestration — the async half of the web + * reference `web/src/lib/message-window-store.ts`, driving the pure + * transitions in `MessageWindowLogic` over a [StateFlow]. + * + * Concurrency model: the web store is single-threaded JS whose interleaving + * points are its `await`s; here every synchronous read-modify-write segment + * between transport calls runs under [stateMutex], and the generation + * counters + request-baseline identity checks (ported as-is) handle whatever + * interleaves across the api calls — which are deliberately made **outside** + * the lock. Tail syncs are single-flight per session with an optional + * trailing re-run, exactly like the web `TailSyncController`. + */ +class MessageWindowStore( + val sessionId: String, + private val api: MessagesApi, + private val scope: CoroutineScope, + private val snapshots: WindowSnapshots? = null, + initialState: MessageWindowState? = null, +) { + private val stateMutex = Mutex() + private val _state = MutableStateFlow(initialState ?: MessageWindowLogic.createState(sessionId)) + + /** The full window state (UI projects what it needs; see [uiState]). */ + val state: StateFlow = _state.asStateFlow() + + /** + * UI-facing projection — the fields the web `useMessages` hook returns; + * cursor/generation internals stay off the render path. + */ + val uiState: Flow = state + .map { current -> + MessageWindowUiState( + messages = current.messages, + hasMore = current.hasMore, + isSyncingTail = current.isSyncingTail, + isLoadingMore = current.isLoadingMore, + warning = current.warning, + viewMode = current.viewMode, + messagesVersion = current.messagesVersion, + historyVersion = current.historyVersion, + tailRevision = current.tailRevision, + ) + } + .distinctUntilChanged() + + // ------------------------------------------------------- tail controller -- + + private val controllerLock = Any() + private var running: TailRun? = null + private var trailingRequested = false + + /** Bumped by [clear]/[seedFrom] so a finished run stops chaining trailing runs. */ + private var controllerEpoch = 0 + + private class TailRun(val prefersLatest: Boolean) { + lateinit var job: Deferred + } + + private sealed interface SyncDecision { + data class Await(val job: Deferred) : SyncDecision + data class Drain(val observed: Deferred) : SyncDecision + } + + /** + * Run (or join) a tail sync (web `syncTailMessages`). With + * [ensureAfterCurrent] the call drains: if a run is already in flight, a + * trailing run is requested and awaited, so the caller returns only after + * a sync that STARTED at or after this call. + */ + suspend fun syncTail(ensureAfterCurrent: Boolean = false) { + val decision: SyncDecision = synchronized(controllerLock) { + val current = running + when { + current == null -> SyncDecision.Await(startTailSyncLocked()) + _state.value.preferLatestOnActivation -> + if (current.prefersLatest) { + SyncDecision.Await(current.job) + } else { + trailingRequested = false + SyncDecision.Await(startTailSyncLocked()) + } + !ensureAfterCurrent -> SyncDecision.Await(current.job) + else -> { + trailingRequested = true + SyncDecision.Drain(current.job) + } + } + } + when (decision) { + is SyncDecision.Await -> decision.job.join() + is SyncDecision.Drain -> drainTailSync(decision.observed) + } + } + + private fun startTailSyncLocked(): Deferred { + val epoch = controllerEpoch + val run = TailRun(prefersLatest = _state.value.preferLatestOnActivation) + // UNDISPATCHED mirrors the web: `runTailSync` executes to its first true + // suspension (normally the api call) before `startTailSync` returns, so + // `beginTailSync`'s generation bump lands synchronously and an in-flight + // run it replaces cannot commit another page in between. + run.job = scope.async(start = CoroutineStart.UNDISPATCHED) { runTailSync() } + running = run + run.job.invokeOnCompletion { + synchronized(controllerLock) { + if (controllerEpoch != epoch || running !== run) return@invokeOnCompletion + running = null + if (!trailingRequested) return@invokeOnCompletion + trailingRequested = false + startTailSyncLocked() + } + } + return run.job + } + + /** Web `waitForTailSyncDrain`: follow the chain until no newer run exists. */ + private suspend fun drainTailSync(observed: Deferred) { + val epoch = synchronized(controllerLock) { controllerEpoch } + var current: Deferred? = observed + while (current != null) { + current.join() + current = synchronized(controllerLock) { + if (controllerEpoch != epoch) return + val next = running?.job + if (next != null && next !== current) next else null + } + } + } + + private fun isCurrentTailSync(generation: Long): Boolean = + _state.value.syncGeneration == generation + + private suspend fun runTailSync() { + val generation = update { MessageWindowLogic.beginTailSync(it) }.syncGeneration + try { + val initial = _state.value + val initialCursor = initial.newestPosition + val preferLatestOnActivation = initial.preferLatestOnActivation + val canIncrement = initialCursor != null + && initial.epoch != null + && !initial.requiresLatestReset + && !preferLatestOnActivation + + if (!canIncrement) { + val requestBaseline = baseline() + val response = api.getMessages(sessionId, MessagesQuery.Latest(limit = PAGE_SIZE)) + if (!isCurrentTailSync(generation)) return + update { previous -> + if (previous.syncGeneration != generation) return@update previous + MessageWindowLogic.applyLatestResponse( + previous, + response.windowMessages(), + response.page, + replaceServerRows = initial.requiresLatestReset + || preferLatestOnActivation + || response.page.reset, + requestBaseline = requestBaseline, + ).copy(preferLatestOnActivation = false) + } + finishTailSync(generation, warning = null) + return + } + + var after: MessagePosition = initialCursor!! + var until: MessagePosition? = null + while (true) { + val requestBaseline = baseline() + val response = api.getMessages( + sessionId, + MessagesQuery.After( + afterAt = after.at, + afterSeq = after.seq, + untilAt = until?.at, + untilSeq = until?.seq, + epoch = initial.epoch!!, + limit = PAGE_SIZE, + ), + ) + if (!isCurrentTailSync(generation)) return + + if (response.page.reset || response.page.direction == "latest") { + update { previous -> + if (previous.syncGeneration != generation) return@update previous + MessageWindowLogic.applyLatestResponse( + previous, + response.windowMessages(), + response.page, + replaceServerRows = true, + requestBaseline = requestBaseline, + ) + } + break + } + + val nextAfter = MessageWindowLogic.pagePosition(response.page.nextAfterAt, response.page.nextAfterSeq) + val snapshotHead = + MessageWindowLogic.pagePosition(response.page.snapshotHeadAt, response.page.snapshotHeadSeq) + if (until == null) { + until = snapshotHead + } + + update { previous -> + if (previous.syncGeneration != generation) return@update previous + MessageWindowLogic.applyAfterPage(previous, response.windowMessages(), response.page, nextAfter) + } + + val current = _state.value + if ( + current.requiresLatestReset + || current.preferLatestOnActivation + || !response.page.hasMore + || nextAfter == null + ) { + break + } + if (nextAfter <= after) { + throw IllegalStateException("Message tail cursor did not advance") + } + after = nextAfter + } + + finishTailSync(generation, warning = null) + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + if (!isCurrentTailSync(generation)) return + finishTailSync(generation, error.message ?: "Failed to synchronize messages") + } + } + + private suspend fun finishTailSync(generation: Long, warning: String?) { + update { MessageWindowLogic.finishTailSync(it, generation, warning) } + persist() + } + + // ---------------------------------------------------------- older pages -- + + /** + * Load one older page (web `fetchOlderMessages`). On an epoch mismatch + * the window is invalidated and a full tail sync (`ensureAfterCurrent`) + * runs before the `stopped/epoch-reset` outcome is returned. + * [onBeforeApply] runs while the state lock is held (mirror of the web's + * synchronous updater) — it must not call back into this store. + */ + suspend fun fetchOlder(onBeforeApply: ((Long) -> Boolean)? = null): OlderLoadOutcome { + val initial = _state.value + when (val precheck = MessageWindowLogic.olderLoadPrecheck(initial)) { + is MessageWindowLogic.OlderPrecheck.Stop -> return OlderLoadOutcome.Stopped(precheck.reason) + is MessageWindowLogic.OlderPrecheck.Proceed -> { + val before = precheck.before + val generation = initial.olderGeneration + 1 + update { MessageWindowLogic.beginOlderLoad(it, generation) } + return try { + val response = api.getMessages( + sessionId, + MessagesQuery.Before(beforeAt = before.at, beforeSeq = before.seq, limit = PAGE_SIZE), + ) + if (_state.value.olderGeneration != generation) { + return OlderLoadOutcome.Stopped(OlderLoadOutcome.StopReason.Invalidated) + } + + if (initial.epoch != null && response.page.epoch != initial.epoch) { + update { MessageWindowLogic.applyOlderEpochMismatch(it, generation) } + syncTail(ensureAfterCurrent = true) + return OlderLoadOutcome.Stopped(OlderLoadOutcome.StopReason.EpochReset) + } + + var historyVersion = 0L + var addedRenderableCount = 0 + var applyRejected = false + update { previous -> + if (previous.olderGeneration != generation) return@update previous + val incoming = response.windowMessages() + addedRenderableCount = MessageWindowLogic.countNewRenderableMessages(previous, incoming) + val nextHistoryVersion = previous.historyVersion + 1 + if (onBeforeApply != null && !onBeforeApply(nextHistoryVersion)) { + applyRejected = true + return@update MessageWindowLogic.rejectOlderApply(previous) + } + historyVersion = nextHistoryVersion + MessageWindowLogic.applyOlderResponse(previous, incoming, response.page, nextHistoryVersion) + } + if (applyRejected || historyVersion == 0L) { + return OlderLoadOutcome.Stopped(OlderLoadOutcome.StopReason.Invalidated) + } + persist() + OlderLoadOutcome.Applied( + historyVersion = historyVersion, + hasMore = response.page.hasMore, + addedRenderableCount = addedRenderableCount, + ) + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + if (_state.value.olderGeneration != generation) { + return OlderLoadOutcome.Stopped(OlderLoadOutcome.StopReason.Invalidated) + } + update { + MessageWindowLogic.failOlderLoad( + it, + generation, + error.message ?: "Failed to load older messages", + ) + } + OlderLoadOutcome.Failed(error) + } + } + } + } + + suspend fun cancelOlderLoad() { + update { MessageWindowLogic.cancelOlderLoad(it) } + } + + // ----------------------------------------------------------- SSE ingest -- + + /** + * Route one message-stream SSE event into this window (the + * `SyncTargets.onMessageEvent` hook). `messages-invalidated` clears the + * window and starts a fresh tail sync; `scheduled-matured` re-syncs so + * the released row (and its consumption) lands even if the + * `message-received` frame was missed. + */ + suspend fun onMessageEvent(event: SyncEvent) { + when (event) { + is SyncEvent.MessageReceived -> ingestSseMessages(listOf(event.message.asWindowMessage())) + is SyncEvent.MessagesConsumed -> markConsumed(event.localIds, event.invokedAt) + is SyncEvent.MessageCancelled -> removeMessage(event.messageId) + is SyncEvent.MessagesInvalidated -> { + clear() + scope.launch { syncTail() } + } + is SyncEvent.ScheduledMatured -> scope.launch { syncTail(ensureAfterCurrent = true) } + else -> Unit + } + } + + /** SSE `message-received` ingest (web `ingestIncomingMessages`). */ + suspend fun ingestSseMessages(messages: List) { + if (messages.isEmpty()) return + update { MessageWindowLogic.ingestIncoming(it, messages) } + persist() + } + + /** SSE `messages-consumed` (web `markMessagesConsumed`). */ + suspend fun markConsumed(localIds: List, invokedAt: Long) { + if (localIds.isEmpty()) return + update { MessageWindowLogic.markConsumed(it, localIds, invokedAt) } + persist() + } + + /** SSE `message-cancelled` / optimistic DELETE removal (web `removeOptimisticMessage`). */ + suspend fun removeMessage(localIdOrId: String) { + update { MessageWindowLogic.removeByLocalIdOrId(it, localIdOrId) } + persist() + } + + // ------------------------------------------------------ optimistic sends -- + + /** Append a pre-built optimistic row (web `appendOptimisticMessage`). */ + suspend fun appendOptimistic(message: WindowMessage) { + update { MessageWindowLogic.appendOptimistic(it, message) } + persist() + } + + /** + * Append the standard optimistic row for a send + * (`useSendMessage.createOptimisticMessage`): status `sending` until the + * POST settles, then [updateStatus] to `queued`/`sent`/`failed`. + */ + suspend fun appendOptimistic( + localId: String, + text: String, + attachments: List? = null, + scheduledAt: Long? = null, + deliveryMode: String = "queue", + createdAt: Long = System.currentTimeMillis(), + ) { + appendOptimistic( + buildOptimisticMessage( + localId = localId, + text = text, + createdAt = createdAt, + attachments = attachments, + scheduledAt = scheduledAt, + deliveryMode = deliveryMode, + status = MessageStatus.Sending, + ), + ) + } + + suspend fun updateStatus(localId: String, status: MessageStatus) { + update { MessageWindowLogic.updateStatus(it, localId, status) } + persist() + } + + /** + * A cancel DELETE answered `{"status":"invoked"}` — too late, the agent + * consumed the row. Remove the queued snapshot and ingest the returned + * authoritative row as `sent` (web `useCancelQueuedMessage`). + */ + suspend fun applyCancelInvoked(localId: String, message: WindowMessage) { + removeMessage(localId) + appendOptimistic(message.copy(status = MessageStatus.Sent)) + } + + // ------------------------------------------------- queued reconciliation -- + + fun queuedReconcileCandidateLocalIds(): List = + MessageWindowLogic.queuedReconcileCandidateLocalIds(_state.value) + + suspend fun reconcileQueuedLocalIds(candidateLocalIds: List, queuedLocalIds: List) { + update { MessageWindowLogic.reconcileQueuedLocalIds(it, candidateLocalIds, queuedLocalIds) } + persist() + } + + /** + * Queued-state recovery after a `resume: 'gap'` reconnect (port of + * `web/src/lib/queued-state-reconciliation.ts`): tail-sync to the drain, + * collect candidates, ask the hub for the verdict in ≤1000-id batches, + * stamp invoked rows like `messages-consumed`, drop deleted candidates. + */ + suspend fun reconcileQueuedState() { + syncTail(ensureAfterCurrent = true) + val candidateLocalIds = queuedReconcileCandidateLocalIds() + if (candidateLocalIds.isEmpty()) return + val queuedLocalIds = mutableListOf() + val invokedLocalMessages = mutableListOf>() + candidateLocalIds.chunked(QUEUED_STATE_BATCH_SIZE).forEach { batch -> + val response = api.getQueuedState(sessionId, batch) + queuedLocalIds += response.queuedLocalIds + invokedLocalMessages += response.invokedLocalMessages.map { it.localId to it.invokedAt } + } + val invokedByTimestamp = LinkedHashMap>() + for ((localId, invokedAt) in invokedLocalMessages) { + invokedByTimestamp.getOrPut(invokedAt) { mutableListOf() }.add(localId) + } + for ((invokedAt, localIds) in invokedByTimestamp) { + markConsumed(localIds, invokedAt) + } + reconcileQueuedLocalIds(candidateLocalIds, queuedLocalIds) + } + + // ------------------------------------------------------------ lifecycle -- + + /** Switch view mode (web `setMessageViewMode`); tail re-entry trims. */ + suspend fun setViewMode(mode: MessageViewMode) { + update { MessageWindowLogic.setViewMode(it, mode) } + persist() + } + + /** + * Session (re-)activation (web `activateMessageWindow`): trims back to + * the visible window and, with usable persisted state, requests a + * fresh-latest trailing run from an in-flight sync. + */ + suspend fun activate() { + var requestedLatest = false + update { previous -> + val activation = MessageWindowLogic.activate(previous) + requestedLatest = activation.requestedLatest + activation.state + } + if (requestedLatest) { + synchronized(controllerLock) { + if (running != null) { + trailingRequested = true + } + } + } + } + + /** Web `clearMessageWindow`: forget everything but keep generations poisoned. */ + suspend fun clear() { + synchronized(controllerLock) { + controllerEpoch += 1 + running = null + trailingRequested = false + } + snapshots?.delete(sessionId) + update { previous -> + MessageWindowLogic.createState(sessionId).copy( + syncGeneration = previous.syncGeneration + 1, + olderGeneration = previous.olderGeneration + 1, + ) + } + } + + /** + * Seed this window from another session's (web + * `seedMessageWindowFromSession`) — resume/reopen can hand back a new + * session id; the old rows render instantly and `requiresLatestReset` + * forces a fresh latest page underneath. + */ + suspend fun seedFrom(source: MessageWindowStore) { + if (source.sessionId == sessionId) return + synchronized(controllerLock) { + controllerEpoch += 1 + running = null + trailingRequested = false + } + val sourceState = source.state.value + update { target -> MessageWindowLogic.seededState(sourceState, target) } + persist() + } + + // -------------------------------------------------------------- internals -- + + private suspend fun update(transform: (MessageWindowState) -> MessageWindowState): MessageWindowState = + stateMutex.withLock { + val next = transform(_state.value) + _state.value = next + next + } + + /** Identity baseline of the current rows, keyed by id (web `requestBaseline`). */ + private fun baseline(): Map = + _state.value.messages.associateBy { it.id } + + private fun MessagesResponse.windowMessages(): List = + messages.map { it.asWindowMessage() } + + private suspend fun persist() { + val snapshotStore = snapshots ?: return + val state = _state.value + if (MessageWindowLogic.shouldPersist(state)) { + snapshotStore.save(sessionId, MessageWindowLogic.toPersisted(state)) + } else { + snapshotStore.delete(sessionId) + } + } + + private companion object { + const val QUEUED_STATE_BATCH_SIZE = 1000 + } +} diff --git a/android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStores.kt b/android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStores.kt new file mode 100644 index 00000000..250138f6 --- /dev/null +++ b/android/core/data/src/main/kotlin/app/hapi/data/store/MessageWindowStores.kt @@ -0,0 +1,59 @@ +package app.hapi.data.store + +import app.hapi.data.api.MessagesApi +import app.hapi.protocol.window.MessageWindowLogic +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * Per-hub registry of per-session [MessageWindowStore]s — the analogue of the + * web module's session-keyed maps. [open] hydrates a cold session from its + * disk snapshot (interrupted `sending` rows restore, stale snapshots start + * flagged for a latest reset); [seed] carries a window across a + * resume/reopen id change. Snapshot files themselves are LRU-capped by + * [WindowSnapshots]. + */ +class MessageWindowStores( + private val api: MessagesApi, + private val scope: CoroutineScope, + private val snapshots: WindowSnapshots? = null, +) { + private val mutex = Mutex() + private val stores = HashMap() + + /** The session's store, hydrating from its snapshot on first open. */ + suspend fun open(sessionId: String): MessageWindowStore { + mutex.withLock { stores[sessionId] }?.let { return it } + // Hydrate outside the lock (disk I/O), then race-tolerantly install. + val hydrated = snapshots?.load(sessionId)?.let { MessageWindowLogic.hydrate(sessionId, it) } + return mutex.withLock { + stores.getOrPut(sessionId) { + MessageWindowStore( + sessionId = sessionId, + api = api, + scope = scope, + snapshots = snapshots, + initialState = hydrated, + ) + } + } + } + + /** Already-open store, if any (no hydration). */ + suspend fun peek(sessionId: String): MessageWindowStore? = + mutex.withLock { stores[sessionId] } + + /** Web `seedMessageWindowFromSession` for resume/reopen id migration. */ + suspend fun seed(fromSessionId: String, toSessionId: String) { + if (fromSessionId.isEmpty() || toSessionId.isEmpty() || fromSessionId == toSessionId) return + val source = open(fromSessionId) + val target = open(toSessionId) + target.seedFrom(source) + } + + /** Web `clearMessageWindow` (e.g. on `session-removed`). */ + suspend fun clear(sessionId: String) { + mutex.withLock { stores[sessionId] }?.clear() + } +} diff --git a/android/core/data/src/main/kotlin/app/hapi/data/store/WindowSnapshots.kt b/android/core/data/src/main/kotlin/app/hapi/data/store/WindowSnapshots.kt new file mode 100644 index 00000000..60c01965 --- /dev/null +++ b/android/core/data/src/main/kotlin/app/hapi/data/store/WindowSnapshots.kt @@ -0,0 +1,90 @@ +package app.hapi.data.store + +import app.hapi.protocol.window.PersistedMessageWindow +import app.hapi.protocol.wire.HapiJson +import java.io.File +import java.io.FileOutputStream +import java.security.MessageDigest +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import kotlinx.serialization.json.Json + +/** + * Per-session message-window snapshots on disk: one JSON file per session in + * [dir], written atomically (temp file + fsync + rename — the same guarantee + * `android.util.AtomicFile` provides, hand-rolled here so plain JVM unit + * tests can exercise it), pruned LRU to [maxSessions] files. + * + * Mirrors the web's `sessionStorage` persistence of + * `PersistedMessageWindowState` (key `hapi:message-window:v2:`): messages + + * cursors + epoch for instant cold-start rendering; `hydrate` in + * `MessageWindowLogic` restores interrupted send states and flags stale + * snapshots for a latest reset. + * + * TODO(B-M2b dedup): fold into the shared `JsonSnapshotStore` once that + * lands — this class predates it in a parallel work package. + */ +class WindowSnapshots( + private val dir: File, + private val maxSessions: Int = DEFAULT_MAX_SESSIONS, + private val json: Json = HapiJson, + private val io: CoroutineDispatcher = Dispatchers.IO, +) { + + suspend fun save(sessionId: String, snapshot: PersistedMessageWindow): Unit = withContext(io) { + dir.mkdirs() + val target = fileFor(sessionId) + val temp = File(dir, target.name + ".tmp") + FileOutputStream(temp).use { stream -> + stream.write(json.encodeToString(PersistedMessageWindow.serializer(), snapshot).toByteArray()) + stream.fd.sync() + } + if (!temp.renameTo(target)) { + // Windows-style rename-over-existing failure; best effort swap. + target.delete() + temp.renameTo(target) + } + prune() + } + + suspend fun load(sessionId: String): PersistedMessageWindow? = withContext(io) { + val file = fileFor(sessionId) + if (!file.isFile) return@withContext null + try { + val snapshot = json.decodeFromString(PersistedMessageWindow.serializer(), file.readText()) + // Touch for LRU recency: reading a snapshot marks the session live. + file.setLastModified(System.currentTimeMillis()) + snapshot + } catch (_: Exception) { + file.delete() + null + } + } + + suspend fun delete(sessionId: String): Unit = withContext(io) { + fileFor(sessionId).delete() + Unit + } + + /** Drop the least-recently-written files beyond [maxSessions]. */ + private fun prune() { + val files = dir.listFiles { file -> file.isFile && file.name.endsWith(SUFFIX) } ?: return + if (files.size <= maxSessions) return + files.sortedByDescending { it.lastModified() } + .drop(maxSessions) + .forEach { it.delete() } + } + + /** Session ids are arbitrary strings — file names come from a digest. */ + private fun fileFor(sessionId: String): File { + val digest = MessageDigest.getInstance("SHA-256").digest(sessionId.toByteArray()) + val name = digest.joinToString("") { "%02x".format(it) }.take(32) + return File(dir, name + SUFFIX) + } + + private companion object { + const val DEFAULT_MAX_SESSIONS = 10 + const val SUFFIX = ".window.json" + } +} diff --git a/android/core/data/src/test/kotlin/app/hapi/data/store/MessageWindowStoreTest.kt b/android/core/data/src/test/kotlin/app/hapi/data/store/MessageWindowStoreTest.kt new file mode 100644 index 00000000..9818d26c --- /dev/null +++ b/android/core/data/src/test/kotlin/app/hapi/data/store/MessageWindowStoreTest.kt @@ -0,0 +1,332 @@ +package app.hapi.data.store + +import app.hapi.data.api.MessagesApi +import app.hapi.data.api.MessagesQuery +import app.hapi.protocol.window.MessageStatus +import app.hapi.protocol.window.MessageWindowLogic +import app.hapi.protocol.window.PersistedMessageWindow +import app.hapi.protocol.window.asWindowMessage +import app.hapi.protocol.window.buildOptimisticMessage +import app.hapi.protocol.wire.DecryptedMessage +import app.hapi.protocol.wire.MessagesPage +import app.hapi.protocol.wire.MessagesResponse +import app.hapi.protocol.wire.OptionalField +import app.hapi.protocol.wire.QueuedStateResponse +import java.io.File +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withContext +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder + +/** + * Targeted coverage for what the pagination fixtures cannot reach: tail-sync + * single-flight/trailing behavior, concurrent-ingest preservation across a + * reset replace (identity baseline), the cursor-no-advance guard, snapshot + * round-trips + LRU, seed-on-resume, and Mutex discipline under parallelism. + */ +class MessageWindowStoreTest { + + @get:Rule + val temp = TemporaryFolder() + + // ------------------------------------------------------------- fixtures -- + + private fun agentRow(id: String, seq: Long, at: Long): DecryptedMessage = DecryptedMessage( + id = id, + seq = seq, + createdAt = at, + invokedAt = OptionalField.Present(at), + content = buildJsonObject { + put("role", "agent") + put("content", buildJsonObject { + put("type", "codex") + put("data", buildJsonObject { + put("type", "message") + put("message", id) + }) + }) + }, + ) + + private fun latestPage( + messages: List, + epoch: Long, + reset: Boolean = false, + hasMore: Boolean = false, + ): MessagesResponse { + val newest = messages.lastOrNull() + val oldest = messages.firstOrNull() + return MessagesResponse( + messages = messages, + page = MessagesPage( + direction = "latest", + limit = 200, + epoch = epoch, + reset = reset, + nextBeforeSeq = oldest?.seq, + nextBeforeAt = oldest?.positionAt, + nextAfterSeq = null, + nextAfterAt = null, + snapshotHeadSeq = newest?.seq, + snapshotHeadAt = newest?.positionAt, + hasMore = hasMore, + ), + ) + } + + private fun afterPage( + messages: List, + epoch: Long, + nextAfter: Pair?, + hasMore: Boolean, + ): MessagesResponse = MessagesResponse( + messages = messages, + page = MessagesPage( + direction = "after", + limit = 200, + epoch = epoch, + reset = false, + nextBeforeSeq = null, + nextBeforeAt = null, + nextAfterSeq = nextAfter?.second, + nextAfterAt = nextAfter?.first, + snapshotHeadSeq = nextAfter?.second, + snapshotHeadAt = nextAfter?.first, + hasMore = hasMore, + ), + ) + + /** Api whose responses are released manually, so tests control interleaving. */ + private class GatedMessagesApi : MessagesApi { + val requests = mutableListOf() + private val responses = Channel(Channel.UNLIMITED) + + fun release(response: MessagesResponse) { + check(responses.trySend(response).isSuccess) + } + + override suspend fun getMessages(sessionId: String, query: MessagesQuery): MessagesResponse { + requests += query + return responses.receive() + } + + override suspend fun getQueuedState(sessionId: String, localIds: List): QueuedStateResponse = + QueuedStateResponse(emptyList(), emptyList()) + } + + // ------------------------------------------------------------ tail sync -- + + @Test + fun `concurrent syncTail calls coalesce into a single run`() = runTest { + val api = GatedMessagesApi() + val store = MessageWindowStore("s", api, backgroundScope) + + val first = launch { store.syncTail() } + runCurrent() + assertEquals(1, api.requests.size) + + val second = launch { store.syncTail() } + runCurrent() + // The second caller joined the in-flight run instead of issuing its own. + assertEquals(1, api.requests.size) + + api.release(latestPage(listOf(agentRow("a-1", 1, 1000)), epoch = 0)) + first.join() + second.join() + assertEquals(1, api.requests.size) + assertEquals(listOf("a-1"), store.state.value.messages.map { it.id }) + } + + @Test + fun `ensureAfterCurrent requests a trailing run and drains it`() = runTest { + val api = GatedMessagesApi() + val store = MessageWindowStore("s", api, backgroundScope) + + val first = launch { store.syncTail() } + runCurrent() + assertEquals(1, api.requests.size) + + val drain = launch { store.syncTail(ensureAfterCurrent = true) } + runCurrent() + assertEquals(1, api.requests.size) + + api.release(latestPage(listOf(agentRow("a-1", 1, 1000)), epoch = 0)) + first.join() + runCurrent() + // The trailing run started and issued its own (after-cursor) request. + assertEquals(2, api.requests.size) + assertTrue(api.requests[1] is MessagesQuery.After) + assertTrue(drain.isActive) + + api.release(afterPage(emptyList(), epoch = 0, nextAfter = 1000L to 1L, hasMore = false)) + drain.join() + assertEquals(2, api.requests.size) + } + + @Test + fun `reset replace preserves rows that arrived while the request was in flight`() = runTest { + val api = GatedMessagesApi() + val store = MessageWindowStore("s", api, backgroundScope) + + // Seed epoch + cursor with a completed latest sync. + val seed = launch { store.syncTail() } + runCurrent() + api.release(latestPage(listOf(agentRow("a-1", 1, 1000)), epoch = 0)) + seed.join() + + // Second sync goes down the after-cursor path and hangs on the api. + val sync = launch { store.syncTail() } + runCurrent() + assertTrue(api.requests.last() is MessagesQuery.After) + + // A live SSE row lands while the request is in flight. + store.ingestSseMessages(listOf(agentRow("b-2", 2, 2000).asWindowMessage())) + + // The server answers with a reset page that does not contain b-2. + api.release(latestPage(listOf(agentRow("c-3", 3, 3000)), epoch = 1, reset = true)) + sync.join() + + val state = store.state.value + // a-1 (captured in the request baseline) was replaced; the concurrent + // b-2 and the authoritative c-3 both survive. + assertEquals(listOf("b-2", "c-3"), state.messages.map { it.id }) + assertEquals(1L, state.epoch) + assertEquals(3000L to 3L, state.newestPosition!!.let { it.at to it.seq }) + } + + @Test + fun `a tail cursor that does not advance aborts with a warning instead of spinning`() = runTest { + val api = GatedMessagesApi() + val store = MessageWindowStore("s", api, backgroundScope) + + val seed = launch { store.syncTail() } + runCurrent() + api.release(latestPage(listOf(agentRow("a-1", 1, 1000)), epoch = 0)) + seed.join() + + val sync = launch { store.syncTail() } + runCurrent() + // hasMore with a non-advancing nextAfter is a protocol violation. + api.release(afterPage(listOf(agentRow("b-2", 2, 2000)), epoch = 0, nextAfter = 1000L to 1L, hasMore = true)) + sync.join() + + val state = store.state.value + assertEquals("Message tail cursor did not advance", state.warning) + assertTrue(!state.isSyncingTail) + assertEquals(2, api.requests.size) + } + + // ------------------------------------------------------------ snapshots -- + + @Test + fun `snapshot round-trip restores interrupted sends and flags stale snapshots`() = runBlocking { + val snapshots = WindowSnapshots(temp.newFolder(), io = Dispatchers.Unconfined) + + val queuedSending = buildOptimisticMessage( + localId = "local-1", + text = "hello", + createdAt = 1000, + status = MessageStatus.Sending, + ) + val fresh = PersistedMessageWindow( + messages = listOf(queuedSending, agentRow("a-1", 1, 500).asWindowMessage()), + hasMore = true, + oldestPositionAt = 500, + oldestPositionSeq = 1, + newestPositionAt = 500, + newestPositionSeq = 1, + epoch = 3, + ) + snapshots.save("session-a", fresh) + + val loaded = snapshots.load("session-a")!! + val hydrated = MessageWindowLogic.hydrate("session-a", loaded) + // Interrupted `sending` on a queued row restores to `queued`. + assertEquals( + MessageStatus.Queued, + hydrated.messages.first { it.id == "local-1" }.status, + ) + assertEquals(3L, hydrated.epoch) + assertTrue(!hydrated.requiresLatestReset) + assertEquals(500L to 1L, hydrated.newestPosition!!.let { it.at to it.seq }) + + // A snapshot without a usable epoch hydrates flagged for a latest reset. + val stale = fresh.copy(epoch = null) + snapshots.save("session-b", stale) + val staleHydrated = MessageWindowLogic.hydrate("session-b", snapshots.load("session-b")!!) + assertTrue(staleHydrated.requiresLatestReset) + assertNull(staleHydrated.epoch) + } + + @Test + fun `snapshots are LRU-capped at ten sessions`() = runBlocking { + val dir = temp.newFolder() + val snapshots = WindowSnapshots(dir, io = Dispatchers.Unconfined) + repeat(12) { index -> + snapshots.save( + "session-$index", + PersistedMessageWindow(messages = listOf(agentRow("a-$index", 1, 1000).asWindowMessage())), + ) + } + val files = dir.listFiles { file: File -> file.name.endsWith(".window.json") }.orEmpty() + assertEquals(10, files.size) + } + + // ----------------------------------------------------------------- seed -- + + @Test + fun `seedFrom carries rows and older cursor but forces a latest reset`() = runTest { + val api = GatedMessagesApi() + val source = MessageWindowStore("old-session", api, backgroundScope) + val seed = launch { source.syncTail() } + runCurrent() + api.release(latestPage(listOf(agentRow("a-1", 1, 1000)), epoch = 5, hasMore = true)) + seed.join() + + val target = MessageWindowStore("new-session", api, backgroundScope) + target.seedFrom(source) + + val state = target.state.value + assertEquals("new-session", state.sessionId) + assertEquals(listOf("a-1"), state.messages.map { it.id }) + assertTrue(state.hasMore) + assertTrue(state.requiresLatestReset) + assertNull(state.epoch) + assertNull(state.newestPosition) + assertEquals(1000L to 1L, state.oldestPosition!!.let { it.at to it.seq }) + } + + // ------------------------------------------------------ mutex discipline -- + + @Test + fun `parallel mutations from real threads serialize through the state mutex`() = runBlocking { + val api = GatedMessagesApi() + val store = MessageWindowStore("s", api, this) + withContext(Dispatchers.Default) { + (0 until 64).map { index -> + launch { + store.appendOptimistic( + localId = "local-$index", + text = "message $index", + createdAt = 1000L + index, + ) + store.updateStatus("local-$index", MessageStatus.Queued) + } + }.forEach { it.join() } + } + val state = store.state.value + assertEquals(64, state.messages.size) + assertTrue(state.messages.all { it.status == MessageStatus.Queued && it.isQueuedForInvocation }) + } +} diff --git a/android/core/data/src/test/kotlin/app/hapi/data/store/PaginationFixtureTest.kt b/android/core/data/src/test/kotlin/app/hapi/data/store/PaginationFixtureTest.kt new file mode 100644 index 00000000..f0f7e84b --- /dev/null +++ b/android/core/data/src/test/kotlin/app/hapi/data/store/PaginationFixtureTest.kt @@ -0,0 +1,314 @@ +package app.hapi.data.store + +import app.hapi.data.api.MessagesApi +import app.hapi.data.api.MessagesQuery +import app.hapi.protocol.window.MessageStatus +import app.hapi.protocol.window.MessageViewMode +import app.hapi.protocol.window.MessageWindowState +import app.hapi.protocol.window.OlderLoadOutcome +import app.hapi.protocol.window.WindowMessage +import app.hapi.protocol.wire.HapiJson +import app.hapi.protocol.wire.MessagesResponse +import app.hapi.protocol.wire.OptionalField +import app.hapi.protocol.wire.QueuedStateResponse +import app.hapi.protocol.wire.longOrNull +import app.hapi.protocol.wire.stringOrNull +import java.io.File +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlin.test.fail +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.builtins.ListSerializer +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonNull +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.put +import org.junit.Test +import org.junit.runner.RunWith +import org.junit.runners.Parameterized + +/** + * Pagination conformance suite: replays every op script in + * `shared/fixtures/pagination/` against the REAL [MessageWindowStore] + * (driven by a scripted [MessagesApi]) and requires + * + * - the exact same `GET /messages` requests, in order (`expectedRequests`), + * - the same older-load outcomes (`expectedOutcome`) and queued-state + * reconcile candidates (`expectedCandidates`), + * - the exact final window projection (`expectedState`). + * + * The twin of `web/src/lib/message-window-store.fixtures.test.ts` / + * `web/scripts/fixtures/pagination/runner.ts`. A failure here means this port + * drifted from the web reference (or the fixtures were regenerated after a + * web behavior change and the port must catch up). + * + * Determinism: ops run strictly sequentially; the store's scope uses + * [Dispatchers.Unconfined], so internal tail-sync jobs run to completion + * within each op (the scripted api never suspends) — mirroring the web + * harness awaiting every op. + */ +@RunWith(Parameterized::class) +class PaginationFixtureTest(private val fixtureName: String) { + + companion object { + private val fixturesDir: File by lazy { + val path = checkNotNull(System.getProperty("hapi.fixtures.dir")) { + "hapi.fixtures.dir system property not set (see core/data/build.gradle.kts)" + } + File(path).also { + check(it.isDirectory) { "fixtures dir does not exist: $it" } + } + } + + @JvmStatic + @Parameterized.Parameters(name = "{0}") + fun fixtures(): List { + val files = File(fixturesDir, "pagination") + .listFiles { file -> file.name.endsWith(".json") } + ?.map { it.name } + ?.sorted() + .orEmpty() + check(files.isNotEmpty()) { + "no pagination fixtures found under $fixturesDir/pagination — refusing to pass on zero files" + } + return files + } + } + + @Test + fun `replaying the ops against the real store matches the stored expectations`() { + val document = HapiJson.parseToJsonElement( + File(File(fixturesDir, "pagination"), fixtureName).readText() + ).jsonObject + + val version = File(fixturesDir, "VERSION").readText().trim().toInt() + assertEquals(version, document.getValue("fixtureVersion").longOrNull?.toInt(), "fixtureVersion") + assertEquals(fixtureName, "${document.getValue("name").stringOrNull}.json", "name matches file") + + val scripted = ScriptedMessagesApi() + val store = MessageWindowStore( + sessionId = "fixture-pagination-${document.getValue("name").stringOrNull}", + api = scripted, + scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined), + ) + + runBlocking { + for ((opIndex, opElement) in document.getValue("ops").jsonArray.withIndex()) { + executeOp(store, scripted, opElement.jsonObject, opIndex) + } + } + + assertEquals( + document.getValue("expectedState"), + projectWindowState(store.state.value), + "final window projection ($fixtureName)", + ) + } + + // ------------------------------------------------------------ op replay -- + + private suspend fun executeOp( + store: MessageWindowStore, + scripted: ScriptedMessagesApi, + op: JsonObject, + opIndex: Int, + ) { + val kind = op.getValue("op").stringOrNull + when (kind) { + "sync-tail" -> { + scripted.begin(decodeResponses(op)) + store.syncTail() + assertSettled(store, scripted, opIndex) + assertRequests(op, scripted, opIndex) + } + "fetch-older" -> { + scripted.begin(decodeResponses(op)) + val outcome = store.fetchOlder() + assertSettled(store, scripted, opIndex) + assertRequests(op, scripted, opIndex) + op["expectedOutcome"]?.let { expected -> + assertEquals(expected, projectOutcome(outcome), "ops[$opIndex] outcome") + } + } + "sse-messages" -> store.ingestSseMessages( + op.getValue("messages").jsonArray.map(::decodeWindowMessage) + ) + "append-optimistic" -> store.appendOptimistic(decodeWindowMessage(op.getValue("message"))) + "update-status" -> store.updateStatus( + localId = op.getValue("localId").stringOrNull!!, + status = MessageStatus.fromWire(op.getValue("status").stringOrNull)!!, + ) + "messages-consumed" -> store.markConsumed( + localIds = op.getValue("localIds").jsonArray.map { it.stringOrNull!! }, + invokedAt = op.getValue("invokedAt").longOrNull!!, + ) + "message-cancelled" -> store.removeMessage(op.getValue("localId").stringOrNull!!) + "cancel-invoked" -> store.applyCancelInvoked( + localId = op.getValue("localId").stringOrNull!!, + message = decodeWindowMessage(op.getValue("message")), + ) + "set-view-mode" -> store.setViewMode( + when (op.getValue("mode").stringOrNull) { + "history" -> MessageViewMode.History + else -> MessageViewMode.Tail + } + ) + "queued-state" -> { + // Mirrors the web runner (reconcileQueuedStateAfterConnect's + // post tail-sync half): collect candidates, apply invoked + // verdicts grouped by timestamp, drop deleted candidates. + val candidates = store.queuedReconcileCandidateLocalIds() + op["expectedCandidates"]?.let { expected -> + assertEquals( + expected, + JsonArray(candidates.map(::JsonPrimitive)), + "ops[$opIndex] candidates", + ) + } + val invokedByTimestamp = LinkedHashMap>() + for (entry in op.getValue("invoked").jsonArray) { + val invoked = entry.jsonObject + invokedByTimestamp + .getOrPut(invoked.getValue("invokedAt").longOrNull!!) { mutableListOf() } + .add(invoked.getValue("localId").stringOrNull!!) + } + for ((invokedAt, localIds) in invokedByTimestamp) { + store.markConsumed(localIds, invokedAt) + } + store.reconcileQueuedLocalIds( + candidateLocalIds = candidates, + queuedLocalIds = op.getValue("queuedLocalIds").jsonArray.map { it.stringOrNull!! }, + ) + } + else -> fail("ops[$opIndex]: unknown op '$kind'") + } + } + + private fun decodeResponses(op: JsonObject): List = + HapiJson.decodeFromJsonElement( + ListSerializer(MessagesResponse.serializer()), + op.getValue("responses"), + ) + + private fun decodeWindowMessage(element: JsonElement): WindowMessage = + HapiJson.decodeFromJsonElement(WindowMessage.serializer(), element) + + private fun assertSettled(store: MessageWindowStore, scripted: ScriptedMessagesApi, opIndex: Int) { + assertEquals(0, scripted.remaining, "ops[$opIndex]: scripted response(s) left unconsumed") + val state = store.state.value + assertNull(state.warning, "ops[$opIndex]: store reported a warning") + assertTrue(!state.isSyncingTail && !state.isLoadingMore, "ops[$opIndex]: store still busy after the op settled") + } + + private fun assertRequests(op: JsonObject, scripted: ScriptedMessagesApi, opIndex: Int) { + val expected = op["expectedRequests"] ?: return + assertEquals(expected, JsonArray(scripted.requests.toList()), "ops[$opIndex] requests") + } + + // ----------------------------------------------------------- projection -- + + private fun projectOutcome(outcome: OlderLoadOutcome): JsonObject = when (outcome) { + is OlderLoadOutcome.Applied -> buildJsonObject { + put("kind", "applied") + put("hasMore", outcome.hasMore) + put("addedRenderableCount", outcome.addedRenderableCount) + } + is OlderLoadOutcome.Stopped -> buildJsonObject { + put("kind", "stopped") + put("reason", outcome.reason.wire) + } + is OlderLoadOutcome.Failed -> fail("older-page load failed: ${outcome.error.message}") + } + + private fun projectWindowState(state: MessageWindowState): JsonObject = buildJsonObject { + put("messages", JsonArray(state.messages.map(::projectMessage))) + put("hasMore", state.hasMore) + put("epoch", state.epoch?.let(::JsonPrimitive) ?: JsonNull) + put("viewMode", state.viewMode.wire) + put("olderCursor", projectCursor(state.oldestPosition?.at, state.oldestPosition?.seq)) + put("newestCursor", projectCursor(state.newestPosition?.at, state.newestPosition?.seq)) + } + + private fun projectMessage(message: WindowMessage): JsonObject = buildJsonObject { + put("id", message.id) + put("localId", message.localId?.let(::JsonPrimitive) ?: JsonNull) + put("seq", message.seq?.let(::JsonPrimitive) ?: JsonNull) + put("createdAt", message.createdAt) + // Wire tri-state: the key appears only when the wire carried it. + when (val invokedAt = message.wire.invokedAt) { + is OptionalField.Present -> put("invokedAt", invokedAt.value?.let(::JsonPrimitive) ?: JsonNull) + OptionalField.Absent -> Unit + } + message.wire.scheduledAt?.let { put("scheduledAt", it) } + message.status?.let { put("status", it.wire) } + put("queued", message.isQueuedForInvocation) + put("optimistic", message.isOptimistic) + } + + private fun projectCursor(at: Long?, seq: Long?): JsonElement = + if (at != null && seq != null) { + buildJsonObject { + put("at", at) + put("seq", seq) + } + } else { + JsonNull + } +} + +/** + * The scripted transport: serves queued [MessagesResponse]s FIFO and records + * each request in the canonical shape the web harness pinned — `limit` only + * for a latest page; `beforeAt`+`beforeSeq`+`limit` for older pages; + * `afterAt`+`afterSeq`+`untilAt`+`untilSeq`+`epoch`+`limit` (untils + * explicitly null on the first loop request) for tail catch-up. + */ +internal class ScriptedMessagesApi : MessagesApi { + private val queue = ArrayDeque() + val requests = mutableListOf() + + val remaining: Int get() = queue.size + + fun begin(responses: List) { + queue.clear() + queue.addAll(responses) + requests.clear() + } + + override suspend fun getMessages(sessionId: String, query: MessagesQuery): MessagesResponse { + requests += query.toRequestJson() + return queue.removeFirstOrNull() + ?: throw IllegalStateException("scripted MessagesApi exhausted: unexpected getMessages request") + } + + override suspend fun getQueuedState(sessionId: String, localIds: List): QueuedStateResponse = + throw UnsupportedOperationException("queued-state ops are replayed via store primitives, like the web runner") + + private fun MessagesQuery.toRequestJson(): JsonObject = when (this) { + is MessagesQuery.Latest -> buildJsonObject { + put("limit", limit) + } + is MessagesQuery.Before -> buildJsonObject { + put("beforeAt", beforeAt) + put("beforeSeq", beforeSeq) + put("limit", limit) + } + is MessagesQuery.After -> buildJsonObject { + put("afterAt", afterAt) + put("afterSeq", afterSeq) + put("epoch", epoch) + put("limit", limit) + put("untilAt", untilAt?.let(::JsonPrimitive) ?: JsonNull) + put("untilSeq", untilSeq?.let(::JsonPrimitive) ?: JsonNull) + } + } +} diff --git a/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageMerge.kt b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageMerge.kt new file mode 100644 index 00000000..ff5223c3 --- /dev/null +++ b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageMerge.kt @@ -0,0 +1,129 @@ +package app.hapi.protocol.window + +import app.hapi.protocol.wire.OptionalField +import app.hapi.protocol.wire.isUserMessage +import kotlin.math.abs + +/** + * Merge/order semantics for window rows — a faithful port of + * `web/src/lib/messages.ts` (`compareMessages`, `mergeMessages`). + */ +object MessageMerge { + + /** + * Position comparator: ascending `at = invokedAt ?? createdAt`, ties by + * `seq`, full ties by ASCII id comparison. The web uses `localeCompare` + * for the final tie-break, but the fixtures keep every position pair + * distinct and the fixtures README pins ASCII comparison for native + * ports (`shared/fixtures/README.md`, pagination determinism notes). + */ + val comparator: Comparator = Comparator { a, b -> + val aTime = a.invokedAtOrNull ?: a.createdAt + val bTime = b.invokedAtOrNull ?: b.createdAt + if (aTime != bTime) return@Comparator aTime.compareTo(bTime) + val aSeq = a.seq + val bSeq = b.seq + if (aSeq != null && bSeq != null && aSeq != bSeq) return@Comparator aSeq.compareTo(bSeq) + a.id.compareTo(b.id) + } + + /** + * Merge [incoming] into [existing]: + * - by-id upsert (incoming wins; a numeric `invokedAt` already known + * locally is preserved when the incoming copy lacks one); + * - stored rows carrying a `localId` evict the optimistic bubble with the + * same `localId`, inheriting its client `status` (when the server row + * has none) and any locally-known numeric `invokedAt`; + * - fallback: an optimistic row already `sent` is dropped when a server + * user message sits within 10 s of the same position; + * - stable sort by [comparator]. + */ + fun mergeMessages(existing: List, incoming: List): List { + if (existing.isEmpty()) { + return incoming.sortedWith(comparator) + } + if (incoming.isEmpty()) { + return existing.sortedWith(comparator) + } + + val byId = LinkedHashMap() + for (message in existing) { + byId[message.id] = message + } + for (message in incoming) { + val current = byId[message.id] + // JS `existing.invokedAt != null && msg.invokedAt == null` — keep a + // locally-known invocation stamp when the incoming copy has none + // (explicit null or absent). + byId[message.id] = if (current != null && current.invokedAtOrNull != null && message.invokedAtOrNull == null) { + message.withInvokedAt(current.invokedAtOrNull!!) + } else { + message + } + } + + var merged: List = byId.values.toList() + + val incomingStoredLocalIds = buildSet { + for (message in incoming) { + val localId = message.localId + if (localId != null && !message.isOptimistic) add(localId) + } + } + + if (incomingStoredLocalIds.isNotEmpty()) { + val optimisticStatusByLocalId = HashMap() + val optimisticInvokedAtByLocalId = HashMap() + for (message in merged) { + val localId = message.localId ?: continue + if (!message.isOptimistic || localId !in incomingStoredLocalIds) continue + message.status?.let { optimisticStatusByLocalId[localId] = it } + if (message.wire.invokedAt is OptionalField.Present) { + optimisticInvokedAtByLocalId[localId] = message.invokedAtOrNull + } + } + merged = merged.filter { message -> + val localId = message.localId + localId == null || localId !in incomingStoredLocalIds || !message.isOptimistic + } + if (optimisticStatusByLocalId.isNotEmpty() || optimisticInvokedAtByLocalId.isNotEmpty()) { + merged = merged.map { message -> + val localId = message.localId ?: return@map message + var updated = message + val preservedStatus = optimisticStatusByLocalId[localId] + if (preservedStatus != null && message.status == null) { + updated = updated.copy(status = preservedStatus) + } + if (optimisticInvokedAtByLocalId.containsKey(localId) && message.invokedAtOrNull == null) { + val optimisticInvokedAt = optimisticInvokedAtByLocalId[localId] + if (optimisticInvokedAt != null) { + updated = updated.withInvokedAt(optimisticInvokedAt) + } + } + updated + } + } + } + + val optimisticMessages = merged.filter { it.isOptimistic } + val nonOptimisticMessages = merged.filter { !it.isOptimistic } + val result = nonOptimisticMessages.toMutableList() + + for (optimistic in optimisticMessages) { + if (optimistic.status == MessageStatus.Sent) { + // Compare by the position key (invokedAt ?? createdAt): a late + // ack can attach invokedAt long after createdAt, and matching + // on createdAt alone would render duplicates. + val optimisticTime = optimistic.invokedAtOrNull ?: optimistic.createdAt + val hasServerUserMessage = nonOptimisticMessages.any { message -> + message.wire.isUserMessage + && abs((message.invokedAtOrNull ?: message.createdAt) - optimisticTime) < 10_000 + } + if (hasServerUserMessage) continue + } + result.add(optimistic) + } + + return result.sortedWith(comparator) + } +} diff --git a/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageRetention.kt b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageRetention.kt new file mode 100644 index 00000000..25af96d4 --- /dev/null +++ b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageRetention.kt @@ -0,0 +1,213 @@ +package app.hapi.protocol.window + +import app.hapi.protocol.wire.stringOrNull +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonNull +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive + +/** + * Renderability predicate for window retention — a faithful port of the + * **null-decision tree** of `normalizeDecryptedMessage` + * (`web/src/chat/normalize.ts` + `web/src/chat/normalizeAgent.ts` + + * `shared/src/messages.ts`), used by the web store's + * `shouldRetainWindowMessage`: rows the chat pipeline would hide never enter + * the window (but still advance cursors). + * + * The web returns `null` (⇒ hidden) in exactly two situations: + * 1. agent rows whose content `isSkippableAgentContent` (meta/compact-summary + * Claude output, empty recaps/agy steps, non-chat-visible system rows); + * 2. codex-envelope agent rows (`content.type === 'codex'`) whose payload + * the codex switch of `normalizeAgentRecord` cannot render. + * Everything else — user rows, unknown roles, unparseable envelopes — falls + * back to a raw-stringify bubble and is renderable. + * + * TODO(B-M2a dedup): the full chat-pipeline port owns the complete + * `normalizeAgentRecord`; once it lands, this predicate should be re-expressed + * as `normalize(...) != null` so the two cannot drift. + */ +object MessageRetention { + + fun isRenderable(content: JsonElement): Boolean { + val record = unwrapRoleWrappedRecordEnvelope(content) ?: return true + if (record.role != "agent") return true + if (isSkippableAgentContent(record.content)) return false + if (isCodexContent(record.content) && !isRenderableCodexContent(record.content)) return false + return true + } + + // ------------------------------------------------------------ envelope -- + + internal data class RoleWrappedRecord(val role: String, val content: JsonElement) + + /** `shared/src/messages.ts` `isRoleWrappedRecord`: object with a string `role` and a `content` key. */ + private fun asRoleWrappedRecord(value: JsonElement?): RoleWrappedRecord? { + val obj = value as? JsonObject ?: return null + val role = obj["role"].stringOrNull ?: return null + if (!obj.containsKey("content")) return null + return RoleWrappedRecord(role, obj["content"] ?: JsonNull) + } + + /** `shared/src/messages.ts` `unwrapRoleWrappedRecordEnvelope`. */ + internal fun unwrapRoleWrappedRecordEnvelope(value: JsonElement?): RoleWrappedRecord? { + asRoleWrappedRecord(value)?.let { return it } + val obj = value as? JsonObject ?: return null + asRoleWrappedRecord(obj["message"])?.let { return it } + (obj["data"] as? JsonObject)?.let { data -> asRoleWrappedRecord(data["message"])?.let { return it } } + (obj["payload"] as? JsonObject)?.let { payload -> asRoleWrappedRecord(payload["message"])?.let { return it } } + return null + } + + // ------------------------------------------------- JS-semantics helpers -- + + /** JS `isObject` (`shared/src/utils.ts`): non-null `typeof 'object'` — arrays included. */ + private fun jsIsObject(value: JsonElement?): Boolean = + value is JsonObject || value is JsonArray + + /** JS property access: arrays and primitives have no string keys (→ undefined). */ + private fun prop(value: JsonElement?, key: String): JsonElement? = + (value as? JsonObject)?.get(key) + + /** JS `asString`: the value iff it is a string. */ + private fun asString(value: JsonElement?): String? = value.stringOrNull + + /** JS `asNumber`: the value iff it is a finite number. */ + private fun asNumber(value: JsonElement?): Double? = + (value as? JsonPrimitive)?.takeUnless { it.isString }?.content?.toDoubleOrNull() + ?.takeIf { it.isFinite() } + + /** JS `Boolean(x)` truthiness over a JSON value. */ + private fun jsTruthy(value: JsonElement?): Boolean = when (value) { + null, is JsonNull -> false + is JsonObject, is JsonArray -> true + is JsonPrimitive -> when { + value.isString -> value.content.isNotEmpty() + value.content == "true" -> true + value.content == "false" -> false + else -> value.content.toDoubleOrNull()?.let { it != 0.0 && !it.isNaN() } ?: false + } + } + + /** JS `a ?? b` over JSON lookups (absent and `null` fall through). */ + private fun coalesce(vararg values: JsonElement?): JsonElement? = + values.firstOrNull { it != null && it !is JsonNull } + + // --------------------------------------------------- agent skippability -- + + private val VISIBLE_CLAUDE_SYSTEM_SUBTYPES = setOf( + "api_error", + "turn_duration", + "microcompact_boundary", + "compact_boundary", + "away_summary", + ) + + /** `shared/src/messages.ts` `isClaudeChatVisibleMessage`. */ + private fun isClaudeChatVisibleMessage(type: JsonElement?, subtype: JsonElement?): Boolean { + val typeString = type.stringOrNull + if (typeString == "rate_limit_event") return false + if (typeString == "tool_progress") return false + if (typeString != "system") return true + return subtype.stringOrNull in VISIBLE_CLAUDE_SYSTEM_SUBTYPES + } + + /** `web/src/chat/normalizeAgent.ts` `isSkippableAgentContent`. */ + internal fun isSkippableAgentContent(content: JsonElement?): Boolean { + if (prop(content, "type").stringOrNull != "output") return false + val data = prop(content, "data")?.takeIf { jsIsObject(it) } ?: return false + if (jsTruthy(prop(data, "isMeta")) || jsTruthy(prop(data, "isCompactSummary"))) return true + val dataType = prop(data, "type") + val subtype = prop(data, "subtype") + if ( + dataType.stringOrNull == "system" + && subtype.stringOrNull == "away_summary" + && asString(prop(data, "content"))?.trim().isNullOrEmpty() + ) { + return true + } + if ( + dataType.stringOrNull == "agy_message" + && (asString(prop(data, "content")) ?: "").trim().isEmpty() + ) { + return true + } + return !isClaudeChatVisibleMessage(dataType, subtype) + } + + /** `content.type === AGENT_MESSAGE_PAYLOAD_TYPE` (`'codex'`, `shared/src/modes.ts`). */ + internal fun isCodexContent(content: JsonElement?): Boolean = + prop(content, "type").stringOrNull == "codex" + + // ------------------------------------------------- codex renderability -- + + /** + * Whether the codex branch of `normalizeAgentRecord` yields a non-null + * message for this content (`web/src/chat/normalizeAgent.ts:923-1241`). + */ + internal fun isRenderableCodexContent(content: JsonElement?): Boolean { + val data = prop(content, "data")?.takeIf { jsIsObject(it) } ?: return false + val type = prop(data, "type").stringOrNull ?: return false + return when (type) { + "agent-run-start", "agent-run-update", "agent-run-trace" -> true + "generated-image" -> + !asString(coalesce(prop(data, "imageId"), prop(data, "image_id"))).isNullOrEmpty() + "error" -> prop(data, "message").stringOrNull != null + "message" -> prop(data, "message").stringOrNull != null + "reasoning" -> prop(data, "message").stringOrNull != null + "context_compacted" -> true + "compact-summary" -> prop(data, "summary").stringOrNull != null + "token_count" -> hasCodexTokenUsage(data) + "thread_goal_updated" -> hasValidThreadGoal(prop(data, "goal")) + "thread_goal_cleared" -> true + "tool-call" -> prop(data, "callId").stringOrNull != null + "tool-call-result" -> prop(data, "callId").stringOrNull != null + "plan" -> hasPlanEntries(coalesce(prop(data, "entries"), prop(data, "items")) ?: data) + "plan_update" -> hasPlanEntries( + coalesce(prop(data, "plan"), prop(data, "update"), prop(data, "items"), prop(data, "steps")) ?: data + ) + else -> false + } + } + + /** Null-condition of `normalizeCodexTokenUsage`: `info` object with numeric input+output tokens. */ + private fun hasCodexTokenUsage(data: JsonElement): Boolean { + val info = prop(data, "info")?.takeIf { jsIsObject(it) } ?: return false + val usageSource = sequenceOf("last", "lastTokenUsage", "last_token_usage", "total", "totalTokenUsage", "total_token_usage") + .map { prop(info, it) } + .firstOrNull { jsIsObject(it) } + ?: info + val input = asNumber(coalesce(prop(usageSource, "inputTokens"), prop(usageSource, "input_tokens"))) + val output = asNumber(coalesce(prop(usageSource, "outputTokens"), prop(usageSource, "output_tokens"))) + return input != null && output != null + } + + /** Null-condition of `normalizeThreadGoal`. */ + private fun hasValidThreadGoal(goal: JsonElement?): Boolean { + if (!jsIsObject(goal)) return false + val threadId = asString(coalesce(prop(goal, "threadId"), prop(goal, "thread_id"))) + val objective = asString(prop(goal, "objective")) + val status = asString(prop(goal, "status")) + if (threadId.isNullOrEmpty() || objective.isNullOrEmpty() || status.isNullOrEmpty()) return false + return status in setOf("active", "paused", "budgetLimited", "usageLimited", "blocked", "complete") + } + + /** Non-empty condition of `normalizePlanEntries`. */ + private fun hasPlanEntries(value: JsonElement?): Boolean { + val entries: List = when { + value is JsonArray -> value + jsIsObject(value) -> sequenceOf("plan", "items", "steps") + .map { prop(value, it) } + .firstOrNull { it is JsonArray } + ?.let { (it as JsonArray).toList() } + ?: emptyList() + else -> emptyList() + } + return entries.any { entry -> + entry.stringOrNull != null || ( + entry is JsonObject && sequenceOf("step", "content", "text", "title", "description") + .any { key -> !asString(prop(entry, key)).isNullOrEmpty() } + ) + } + } +} diff --git a/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowLogic.kt b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowLogic.kt new file mode 100644 index 00000000..2996ebec --- /dev/null +++ b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowLogic.kt @@ -0,0 +1,659 @@ +package app.hapi.protocol.window + +import app.hapi.protocol.wire.MessagesPage +import app.hapi.protocol.wire.hasExplicitNullInvokedAt +import app.hapi.protocol.wire.stringOrNull +import kotlinx.serialization.json.JsonObject + +/** + * Pure state transitions of the message window — a function-for-function port + * of `web/src/lib/message-window-store.ts` (with the async orchestration + * stripped out; `:core:data`'s `MessageWindowStore` re-adds it). Every + * function takes the previous [MessageWindowState] and returns the next one; + * nothing here touches clocks, I/O, or coroutines, so the whole surface is + * fixture-comparable. + */ +object MessageWindowLogic { + + enum class TrimMode { Append, Prepend } + + fun createState(sessionId: String): MessageWindowState = MessageWindowState(sessionId = sessionId) + + // -------------------------------------------------------------- helpers -- + + /** Web `buildState`: swap the messages list, re-derive seq bounds, bump the version on identity change. */ + private fun MessageWindowState.withMessages(messages: List): MessageWindowState { + if (messages === this.messages) return this + var oldestSeq: Long? = null + var newestSeq: Long? = null + for (message in messages) { + val seq = message.seq ?: continue + oldestSeq = if (oldestSeq == null) seq else minOf(oldestSeq, seq) + newestSeq = if (newestSeq == null) seq else maxOf(newestSeq, seq) + } + return copy( + messages = messages, + oldestSeq = oldestSeq, + newestSeq = newestSeq, + messagesVersion = messagesVersion + 1, + ) + } + + /** A row's compound position; rows without a server `seq` have none. */ + fun messagePosition(message: WindowMessage): MessagePosition? { + val seq = message.seq ?: return null + return MessagePosition(at = message.positionAt, seq = seq) + } + + enum class PositionEnd { Oldest, Newest } + + fun derivePosition(messages: List, end: PositionEnd): MessagePosition? { + var selected: MessagePosition? = null + for (message in messages) { + val candidate = messagePosition(message) ?: continue + val current = selected + if (current == null) { + selected = candidate + continue + } + val comparison = candidate.compareTo(current) + if ((end == PositionEnd.Oldest && comparison < 0) || (end == PositionEnd.Newest && comparison > 0)) { + selected = candidate + } + } + return selected + } + + /** Pairwise page cursor (`pagePosition` in the web): both halves or nothing. */ + fun pagePosition(at: Long?, seq: Long?): MessagePosition? = + if (at != null && seq != null) MessagePosition(at, seq) else null + + private fun maxPosition(a: MessagePosition?, b: MessagePosition?): MessagePosition? = when { + a == null -> b + b == null -> a + a >= b -> a + else -> b + } + + // ---------------------------------------------------------------- trims -- + + private data class Trim(val kept: List, val dropped: List) + + private fun sliceForTrim(items: List, limit: Int, mode: TrimMode): Trim { + if (items.size <= limit) return Trim(items, emptyList()) + if (limit <= 0) return Trim(emptyList(), items) + return when (mode) { + TrimMode.Prepend -> Trim(items.subList(0, limit).toList(), items.subList(limit, items.size).toList()) + TrimMode.Append -> Trim( + items.subList(items.size - limit, items.size).toList(), + items.subList(0, items.size - limit).toList(), + ) + } + } + + /** + * Codex background-agent trace rows (`agent-run-*`) get their own trim + * bucket so long traces do not evict chat (web `isCodexAgentRunMessage`). + */ + fun isCodexAgentRunMessage(message: WindowMessage): Boolean { + val outer = message.wire.content as? JsonObject ?: return false + if (outer["role"].stringOrNull != "agent") return false + val content = outer["content"] as? JsonObject ?: return false + if (content["type"].stringOrNull != "codex") return false + val data = content["data"] as? JsonObject ?: return false + val type = data["type"].stringOrNull + return type == "agent-run-start" || type == "agent-run-update" || type == "agent-run-trace" + } + + /** + * Trim to [regularLimit] while never dropping queued rows: the regular + * budget shrinks by the queued count, `agent-run-*` rows trim against + * their own [AGENT_RUN_WINDOW_SIZE] bucket, and queued rows are re-merged + * afterwards (web `trimPreservingQueued`). + */ + private fun trimPreservingQueued(messages: List, regularLimit: Int, mode: TrimMode): Trim { + val queued = messages.filter { it.isQueuedForInvocation } + val queuedIds = queued.mapTo(HashSet()) { it.id } + val nonQueued = messages.filter { it.id !in queuedIds } + val agentRuns = nonQueued.filter { isCodexAgentRunMessage(it) } + val regular = nonQueued.filter { !isCodexAgentRunMessage(it) } + val regularTrim = sliceForTrim(regular, maxOf(0, regularLimit - queued.size), mode) + val agentRunTrim = sliceForTrim(agentRuns, AGENT_RUN_WINDOW_SIZE, mode) + return Trim( + kept = MessageMerge.mergeMessages(regularTrim.kept + agentRunTrim.kept, queued), + dropped = regularTrim.dropped + agentRunTrim.dropped, + ) + } + + // ------------------------------------------------------------ retention -- + + /** Web `shouldRetainWindowMessage`: queued, or renderable by the chat pipeline. */ + fun shouldRetainWindowMessage(message: WindowMessage): Boolean = + message.isQueuedForInvocation || MessageRetention.isRenderable(message.wire.content) + + /** + * How many of [incoming] would add a NEW renderable row (not already + * represented by id or localId) — feeds the `applied` older-load outcome. + */ + fun countNewRenderableMessages(previous: MessageWindowState, incoming: List): Int { + val representedIds = previous.messages.mapTo(HashSet()) { it.id } + val representedLocalIds = previous.messages.mapNotNullTo(HashSet()) { it.localId } + var count = 0 + for (message in incoming) { + if (!shouldRetainWindowMessage(message)) continue + if (message.id in representedIds) continue + val localId = message.localId + if (localId != null && localId in representedLocalIds) continue + count += 1 + representedIds.add(message.id) + if (localId != null) representedLocalIds.add(localId) + } + return count + } + + // ---------------------------------------------------------------- merge -- + + /** + * Merge [incoming] into the window, trim per mode/limit, and repair + * cursors/flags on overflow (web `mergeIntoWindow`): an append-side trim + * flips `hasMore` and recomputes the older cursor from the oldest kept + * row; a prepend-side trim drops tail rows, so the window no longer + * reaches the live bottom — flag `requiresLatestReset` and pull the + * newest cursor back to the newest kept row. + */ + fun mergeIntoWindow( + previous: MessageWindowState, + incoming: List, + mode: TrimMode? = null, + regularLimit: Int? = null, + advanceTailRevision: Boolean = false, + ): MessageWindowState { + val retainedIncoming = incoming.filter { shouldRetainWindowMessage(it) } + if (retainedIncoming.isEmpty()) { + return previous + } + val effectiveMode = mode + ?: if (previous.viewMode == MessageViewMode.History) TrimMode.Prepend else TrimMode.Append + val effectiveLimit = regularLimit + ?: if (previous.viewMode == MessageViewMode.History) HISTORY_WINDOW_SIZE else VISIBLE_WINDOW_SIZE + val merged = MessageMerge.mergeMessages(previous.messages, retainedIncoming) + val (kept, dropped) = trimPreservingQueued(merged, effectiveLimit, effectiveMode).let { it.kept to it.dropped } + var next = previous.withMessages(kept) + if (advanceTailRevision) { + next = next.copy(tailRevision = previous.tailRevision + 1) + } + if (dropped.isEmpty()) { + return next + } + if (effectiveMode == TrimMode.Append) { + val oldest = derivePosition(kept, PositionEnd.Oldest) + return next.copy( + hasMore = true, + oldestPosition = oldest ?: next.oldestPosition, + ) + } + val newest = derivePosition(kept, PositionEnd.Newest) + return next.copy( + requiresLatestReset = true, + newestPosition = newest, + ) + } + + // -------------------------------------------------------- latest replace -- + + /** + * Apply a `latest` page (cold start, activation refresh, or a reset + * response). With [replaceServerRows] the page is authoritative: every + * server row captured in [requestBaseline] (by **reference**) is + * discarded, while optimistic rows and rows that changed since the + * request left (concurrent SSE) survive the swap (web + * `applyLatestResponse`). + */ + fun applyLatestResponse( + previous: MessageWindowState, + responseMessages: List, + page: MessagesPage, + replaceServerRows: Boolean, + requestBaseline: Map, + ): MessageWindowState { + val retainedResponseMessages = responseMessages.filter { shouldRetainWindowMessage(it) } + val concurrentServerRows = previous.messages.filter { message -> + !message.isOptimistic && requestBaseline[message.id] !== message + } + val preserved = if (replaceServerRows) { + previous.messages.filter { message -> + message.isOptimistic || requestBaseline[message.id] !== message + } + } else { + previous.messages + } + val authoritative = MessageMerge.mergeMessages(preserved, retainedResponseMessages) + val incoming = MessageMerge.mergeMessages(authoritative, concurrentServerRows) + val (kept, dropped) = trimPreservingQueued(incoming, VISIBLE_WINDOW_SIZE, TrimMode.Append) + .let { it.kept to it.dropped } + val snapshotHead = pagePosition(page.snapshotHeadAt, page.snapshotHeadSeq) + ?: derivePosition(responseMessages, PositionEnd.Newest) + val newestKept = derivePosition(kept, PositionEnd.Newest) + val newest = maxPosition(snapshotHead, newestKept) + val responseOldest = pagePosition(page.nextBeforeAt, page.nextBeforeSeq) + val oldest = when { + dropped.isNotEmpty() -> derivePosition(kept, PositionEnd.Oldest) + replaceServerRows -> responseOldest + else -> responseOldest ?: previous.oldestPosition + } + return previous.withMessages(kept).copy( + hasMore = page.hasMore || (!replaceServerRows && previous.hasMore) || dropped.isNotEmpty(), + epoch = page.epoch, + oldestPosition = oldest, + newestPosition = newest, + tailRevision = previous.tailRevision + 1, + requiresLatestReset = false, + isLoadingMore = if (replaceServerRows) false else previous.isLoadingMore, + olderGeneration = if (replaceServerRows) previous.olderGeneration + 1 else previous.olderGeneration, + warning = null, + ) + } + + // ------------------------------------------------------------ tail sync -- + + /** + * Web `beginTailSync`: claim a new sync generation. The older generation + * bumps too — an older-page response captured before this point must not + * commit while the tail request is in flight, or a reset can mistake it + * for concurrent SSE. + */ + fun beginTailSync(previous: MessageWindowState): MessageWindowState = previous.copy( + syncGeneration = previous.syncGeneration + 1, + olderGeneration = previous.olderGeneration + 1, + isSyncingTail = true, + isLoadingMore = false, + warning = null, + ) + + fun finishTailSync(previous: MessageWindowState, generation: Long, warning: String?): MessageWindowState = + if (previous.syncGeneration != generation) { + previous + } else { + previous.copy(isSyncingTail = false, warning = warning) + } + + /** + * Loop body of the after-cursor catch-up (web `runTailSync` inner + * updater): merge the page, adopt its epoch, and advance the newest + * cursor to `max(current, nextAfter)` — unless a prepend-side trim just + * flagged a latest reset, in which case the cursor stays pulled back. + */ + fun applyAfterPage( + previous: MessageWindowState, + responseMessages: List, + page: MessagesPage, + nextAfter: MessagePosition?, + ): MessageWindowState { + val merged = mergeIntoWindow(previous, responseMessages, advanceTailRevision = true) + if (merged.requiresLatestReset) { + return merged.copy(epoch = page.epoch, warning = null) + } + val newest = maxPosition(nextAfter, merged.newestPosition) + return merged.copy( + epoch = page.epoch, + newestPosition = newest, + warning = null, + ) + } + + // ------------------------------------------------------------ view mode -- + + /** + * Re-enter tail mode (web `enterTailMode`): trim to the visible window; + * when a history overflow flagged `requiresLatestReset`, drop epoch and + * newest cursor so the next tail sync fetches a fresh latest page. + */ + fun enterTailMode(previous: MessageWindowState): MessageWindowState { + val trim = trimPreservingQueued(previous.messages, VISIBLE_WINDOW_SIZE, TrimMode.Append) + val forceLatest = previous.requiresLatestReset + val oldest = if (trim.dropped.isNotEmpty()) { + derivePosition(trim.kept, PositionEnd.Oldest) + } else { + previous.oldestPosition + } + return previous.withMessages(trim.kept).copy( + hasMore = previous.hasMore || trim.dropped.isNotEmpty(), + viewMode = MessageViewMode.Tail, + epoch = if (forceLatest) null else previous.epoch, + oldestPosition = oldest, + newestPosition = if (forceLatest) null else previous.newestPosition, + ) + } + + fun setViewMode(previous: MessageWindowState, mode: MessageViewMode): MessageWindowState = when { + previous.viewMode == mode -> previous + mode == MessageViewMode.History -> previous.copy(viewMode = MessageViewMode.History) + else -> enterTailMode(previous) + } + + data class Activation(val state: MessageWindowState, val requestedLatest: Boolean) + + /** + * Session (re-)activation (web `activateMessageWindow`): trim back to the + * visible window, and when a usable persisted cursor exists, flag + * `preferLatestOnActivation` — another client may have advanced the + * session by many pages, so the next tail sync fetches the current tail + * first and reconciles it through the reset-preservation path. + */ + fun activate(previous: MessageWindowState): Activation { + val trim = trimPreservingQueued(previous.messages, VISIBLE_WINDOW_SIZE, TrimMode.Append) + val forceLatest = previous.requiresLatestReset + val hasUsableCursor = previous.newestPosition != null && previous.epoch != null && !forceLatest + val preferLatestOnActivation = hasUsableCursor && trim.kept.isNotEmpty() + val requestedLatest = preferLatestOnActivation && !previous.preferLatestOnActivation + val invalidateRunningSync = requestedLatest && previous.isSyncingTail + + fun MessageWindowState.withActivationUpdates(): MessageWindowState { + if (!preferLatestOnActivation) return this + val flagged = copy(preferLatestOnActivation = true) + return if (invalidateRunningSync) { + flagged.copy( + syncGeneration = flagged.syncGeneration + 1, + olderGeneration = flagged.olderGeneration + 1, + ) + } else { + flagged + } + } + + if ( + previous.viewMode == MessageViewMode.Tail + && trim.kept.size == previous.messages.size + && !forceLatest + ) { + val state = if (preferLatestOnActivation) previous.withActivationUpdates() else previous + return Activation(state, requestedLatest) + } + val next = enterTailMode(previous) + val state = if (preferLatestOnActivation) next.withActivationUpdates() else next + return Activation(state, requestedLatest) + } + + // ---------------------------------------------------------- older pages -- + + sealed interface OlderPrecheck { + data class Proceed(val before: MessagePosition) : OlderPrecheck + data class Stop(val reason: OlderLoadOutcome.StopReason) : OlderPrecheck + } + + fun olderLoadPrecheck(state: MessageWindowState): OlderPrecheck { + val before = state.oldestPosition + if (state.isSyncingTail || state.isLoadingMore) { + return OlderPrecheck.Stop(OlderLoadOutcome.StopReason.Busy) + } + if (!state.hasMore) { + return OlderPrecheck.Stop(OlderLoadOutcome.StopReason.Exhausted) + } + if (before == null) { + return OlderPrecheck.Stop(OlderLoadOutcome.StopReason.Unavailable) + } + return OlderPrecheck.Proceed(before) + } + + fun beginOlderLoad(previous: MessageWindowState, generation: Long): MessageWindowState = + previous.copy(olderGeneration = generation, isLoadingMore = true, warning = null) + + /** + * An older page answered with a different epoch: every cursor is + * meaningless. Drop epoch + newest cursor and flag the latest reset; the + * caller then runs a tail sync and reports `stopped/epoch-reset`. + */ + fun applyOlderEpochMismatch(previous: MessageWindowState, generation: Long): MessageWindowState = + if (previous.olderGeneration != generation) { + previous + } else { + previous.copy( + isLoadingMore = false, + epoch = null, + newestPosition = null, + requiresLatestReset = true, + ) + } + + /** The `onBeforeApply` veto path: invalidate this load, keep the window. */ + fun rejectOlderApply(previous: MessageWindowState): MessageWindowState = previous.copy( + olderGeneration = previous.olderGeneration + 1, + isLoadingMore = false, + warning = null, + ) + + fun applyOlderResponse( + previous: MessageWindowState, + responseMessages: List, + page: MessagesPage, + historyVersion: Long, + ): MessageWindowState { + val merged = mergeIntoWindow( + previous, + responseMessages, + mode = TrimMode.Prepend, + regularLimit = OLDER_LOAD_WINDOW_SIZE, + ) + return merged.copy( + hasMore = page.hasMore, + epoch = page.epoch, + oldestPosition = pagePosition(page.nextBeforeAt, page.nextBeforeSeq), + isLoadingMore = false, + historyVersion = historyVersion, + warning = null, + ) + } + + fun failOlderLoad(previous: MessageWindowState, generation: Long, warning: String): MessageWindowState = + if (previous.olderGeneration != generation) { + previous + } else { + previous.copy(isLoadingMore = false, warning = warning) + } + + fun cancelOlderLoad(previous: MessageWindowState): MessageWindowState = + if (!previous.isLoadingMore) { + previous + } else { + previous.copy( + olderGeneration = previous.olderGeneration + 1, + isLoadingMore = false, + warning = null, + ) + } + + // ------------------------------------------------------------ live rows -- + + /** + * SSE `message-received` ingest (web `ingestIncomingMessages`): merge, + * then — when an epoch is cached and no reset is pending — advance the + * newest cursor past the incoming rows' positions, **including rows the + * pipeline hides**, so a later tail sync does not refetch them. + */ + fun ingestIncoming(previous: MessageWindowState, incoming: List): MessageWindowState { + if (incoming.isEmpty()) return previous + val merged = mergeIntoWindow(previous, incoming, advanceTailRevision = true) + if (merged.epoch == null || merged.requiresLatestReset) { + return merged + } + val incomingNewest = derivePosition(incoming, PositionEnd.Newest) + val currentNewest = merged.newestPosition + val newest = if (incomingNewest != null && (currentNewest == null || incomingNewest > currentNewest)) { + incomingNewest + } else { + currentNewest + } + return merged.copy(newestPosition = newest) + } + + /** Optimistic append (web `appendOptimisticMessage`): never advances the newest cursor. */ + fun appendOptimistic(previous: MessageWindowState, message: WindowMessage): MessageWindowState = + mergeIntoWindow( + previous, + listOf(message), + mode = if (previous.viewMode == MessageViewMode.History) TrimMode.Prepend else TrimMode.Append, + advanceTailRevision = true, + ) + + fun updateStatus(previous: MessageWindowState, localId: String, status: MessageStatus): MessageWindowState { + if (localId.isEmpty()) return previous + var changed = false + val messages = previous.messages.map { message -> + if (message.localId != localId || message.status == status) { + message + } else { + changed = true + message.copy(status = status) + } + } + return if (changed) previous.withMessages(messages) else previous + } + + /** `message-cancelled` / optimistic DELETE removal: matches localId OR id; idempotent. */ + fun removeByLocalIdOrId(previous: MessageWindowState, localId: String): MessageWindowState { + if (localId.isEmpty()) return previous + val messages = previous.messages.filter { it.localId != localId && it.id != localId } + return if (messages.size == previous.messages.size) previous else previous.withMessages(messages) + } + + /** + * SSE `messages-consumed` (web `markMessagesConsumed`): stamp `invokedAt` + * and flip status to `sent` on matching rows (skip `failed`), then + * re-sort — the row moves from its enqueue position to its invocation + * position. Never advances the newest cursor. + */ + fun markConsumed(previous: MessageWindowState, localIds: List, invokedAt: Long): MessageWindowState { + if (localIds.isEmpty()) return previous + val idSet = localIds.toSet() + var changed = false + val updated = previous.messages.map { message -> + val localId = message.localId + if (localId == null || localId !in idSet || message.status == MessageStatus.Failed) { + return@map message + } + val needsStatus = message.status != MessageStatus.Sent + val needsInvokedAt = message.wire.hasExplicitNullInvokedAt + if (!needsStatus && !needsInvokedAt) return@map message + changed = true + var next = message + if (needsStatus) next = next.copy(status = MessageStatus.Sent) + if (needsInvokedAt) next = next.withInvokedAt(invokedAt) + next + } + if (!changed) return previous + return previous + .withMessages(MessageMerge.mergeMessages(emptyList(), updated)) + .copy(tailRevision = previous.tailRevision + 1) + } + + // -------------------------------------------------- queued reconciliation -- + + /** Web `isQueuedReconcileCandidate`. */ + private fun isQueuedReconcileCandidate(message: WindowMessage): Boolean { + if (message.localId == null || !message.isQueuedForInvocation) return false + if (!message.isOptimistic) return true + return message.status == MessageStatus.Queued || message.status == MessageStatus.Sent + } + + /** Candidate localIds for the queued-state round trip, in window order. */ + fun queuedReconcileCandidateLocalIds(state: MessageWindowState): List { + val localIds = LinkedHashSet() + for (message in state.messages) { + if (isQueuedReconcileCandidate(message)) { + localIds.add(message.localId!!) + } + } + return localIds.toList() + } + + /** + * Apply the queued-state verdict (web `reconcileQueuedLocalIds`): drop + * candidates that are in neither the still-queued list nor (post + * `markConsumed`) invoked — they were deleted server-side. + */ + fun reconcileQueuedLocalIds( + previous: MessageWindowState, + candidateLocalIds: List, + queuedLocalIds: List, + ): MessageWindowState { + if (candidateLocalIds.isEmpty()) return previous + val candidates = candidateLocalIds.toSet() + val queued = queuedLocalIds.toSet() + val messages = previous.messages.filter { message -> + val localId = message.localId + if (localId == null || localId !in candidates) return@filter true + localId in queued || !isQueuedReconcileCandidate(message) + } + return if (messages.size == previous.messages.size) previous else previous.withMessages(messages) + } + + // ---------------------------------------------------------- persistence -- + + /** Web `shouldPersistState`. */ + fun shouldPersist(state: MessageWindowState): Boolean = + state.messages.isNotEmpty() + || state.hasMore + || state.epoch != null + || state.oldestPosition != null + || state.newestPosition != null + + fun toPersisted(state: MessageWindowState): PersistedMessageWindow = PersistedMessageWindow( + messages = state.messages, + hasMore = state.hasMore, + oldestPositionAt = state.oldestPosition?.at, + oldestPositionSeq = state.oldestPosition?.seq, + newestPositionAt = state.newestPosition?.at, + newestPositionSeq = state.newestPosition?.seq, + epoch = state.epoch, + ) + + /** + * Rebuild a state from a persisted snapshot (web `hydrateState`): + * interrupted `sending` rows restore to `queued`/`sent`, and a snapshot + * with rows but no usable cursor/epoch starts flagged for a latest reset. + */ + fun hydrate(sessionId: String, persisted: PersistedMessageWindow): MessageWindowState { + val restored = persisted.messages.map { message -> + if (message.status != MessageStatus.Sending) { + message + } else { + message.copy( + status = if (message.wire.hasExplicitNullInvokedAt) MessageStatus.Queued else MessageStatus.Sent, + ) + } + } + val oldest = pagePosition(persisted.oldestPositionAt, persisted.oldestPositionSeq) + val newest = pagePosition(persisted.newestPositionAt, persisted.newestPositionSeq) + val epoch = persisted.epoch?.takeIf { it >= 0 } + return createState(sessionId) + .withMessages(MessageMerge.mergeMessages(emptyList(), restored)) + .copy( + hasMore = persisted.hasMore, + oldestPosition = oldest, + newestPosition = newest, + epoch = epoch, + requiresLatestReset = persisted.messages.isNotEmpty() && (newest == null || epoch == null), + ) + } + + // ---------------------------------------------------------------- seed -- + + /** + * Seed a fresh window from another session's (web + * `seedMessageWindowFromSession`) — resume/reopen may hand back a + * different session id; the old rows render instantly while + * `requiresLatestReset` forces a fresh latest page underneath. Cursor + * epoch and newest position are NOT carried over. + */ + fun seededState(source: MessageWindowState, target: MessageWindowState): MessageWindowState = + createState(target.sessionId) + .withMessages(source.messages.toList()) + .copy( + hasMore = source.hasMore, + tailRevision = source.tailRevision, + oldestPosition = source.oldestPosition, + requiresLatestReset = true, + syncGeneration = target.syncGeneration + 1, + olderGeneration = target.olderGeneration + 1, + ) +} diff --git a/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowState.kt b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowState.kt new file mode 100644 index 00000000..8ba31330 --- /dev/null +++ b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/MessageWindowState.kt @@ -0,0 +1,113 @@ +package app.hapi.protocol.window + +import kotlinx.serialization.Serializable + +/** + * State machine model for one session's message window — the faithful port of + * the web reference `web/src/lib/message-window-store.ts` (its + * `InternalState`), per `docs/api/client-contract/pagination.md`. The pure + * transitions over this state live in [MessageWindowLogic]; the async + * orchestration (`:core:data` `MessageWindowStore`) drives them. + */ + +/** Windowing constants — normative, from the web reference. */ +const val VISIBLE_WINDOW_SIZE: Int = 400 +const val HISTORY_WINDOW_SIZE: Int = 600 +const val OLDER_LOAD_WINDOW_SIZE: Int = 800 +const val AGENT_RUN_WINDOW_SIZE: Int = 800 +const val PAGE_SIZE: Int = 200 + +enum class MessageViewMode(val wire: String) { + /** Following the live bottom; trims from the top. */ + Tail("tail"), + + /** Scrolled back; trims from the bottom and may force a latest reset. */ + History("history"), +} + +/** + * Compound paging position `(at, seq)` where `at = invokedAt ?? createdAt` — + * both halves always travel together (`pagination.md` "Position key"). + */ +data class MessagePosition(val at: Long, val seq: Long) : Comparable { + override fun compareTo(other: MessagePosition): Int = + if (at != other.at) at.compareTo(other.at) else seq.compareTo(other.seq) +} + +/** Outcome of one older-page load (`fetchOlderMessages` in the web). */ +sealed interface OlderLoadOutcome { + data class Applied( + val historyVersion: Long, + val hasMore: Boolean, + val addedRenderableCount: Int, + ) : OlderLoadOutcome + + data class Stopped(val reason: StopReason) : OlderLoadOutcome + + data class Failed(val error: Throwable) : OlderLoadOutcome + + enum class StopReason(val wire: String) { + Unavailable("unavailable"), + Busy("busy"), + Invalidated("invalidated"), + EpochReset("epoch-reset"), + Exhausted("exhausted"), + } +} + +/** + * The full window state. Field-for-field port of the web `InternalState` + * (public `MessageWindowState` + internal cursor/generation fields), with the + * two `(at, seq)` half-pairs folded into nullable [MessagePosition]s — the web + * only ever reads/writes them pairwise (`readPosition`). + */ +data class MessageWindowState( + val sessionId: String, + /** Window rows in position order (queued rows re-merged after each trim). */ + val messages: List = emptyList(), + /** Older history exists (server flag, or rows were trimmed away). */ + val hasMore: Boolean = false, + /** min/max `seq` over [messages] — derived, kept for UI parity. */ + val oldestSeq: Long? = null, + val newestSeq: Long? = null, + /** Cached server epoch; null until the first page (or after invalidation). */ + val epoch: Long? = null, + val isSyncingTail: Boolean = false, + val isLoadingMore: Boolean = false, + val warning: String? = null, + val viewMode: MessageViewMode = MessageViewMode.Tail, + /** Bumped whenever the [messages] list instance changes. */ + val messagesVersion: Long = 0, + /** Bumped per applied older page (scroll-anchoring handle). */ + val historyVersion: Long = 0, + /** Bumped on tail-side content changes (auto-scroll handle). */ + val tailRevision: Long = 0, + /** Next `before` request position (web `oldestPositionAt/Seq`). */ + val oldestPosition: MessagePosition? = null, + /** Next `after` request position (web `newestPositionAt/Seq`). */ + val newestPosition: MessagePosition? = null, + /** Cursors are unusable — the next tail sync must fetch a fresh latest page. */ + val requiresLatestReset: Boolean = false, + /** Re-activation with persisted state: fetch the current tail first. */ + val preferLatestOnActivation: Boolean = false, + /** Invalidates in-flight tail syncs (compare-and-ignore). */ + val syncGeneration: Long = 0, + /** Invalidates in-flight older-page loads. */ + val olderGeneration: Long = 0, +) + +/** + * Snapshot shape persisted per session (web `PersistedMessageWindowState`, + * storage key `hapi:message-window:v2:`): messages + cursors + epoch only — + * transient flags and counters never survive a restart. + */ +@Serializable +data class PersistedMessageWindow( + val messages: List = emptyList(), + val hasMore: Boolean = false, + val oldestPositionAt: Long? = null, + val oldestPositionSeq: Long? = null, + val newestPositionAt: Long? = null, + val newestPositionSeq: Long? = null, + val epoch: Long? = null, +) diff --git a/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/WindowMessage.kt b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/WindowMessage.kt new file mode 100644 index 00000000..00adb284 --- /dev/null +++ b/android/core/protocol/src/main/kotlin/app/hapi/protocol/window/WindowMessage.kt @@ -0,0 +1,160 @@ +package app.hapi.protocol.window + +import app.hapi.protocol.wire.AttachmentMetadata +import app.hapi.protocol.wire.DecryptedMessage +import app.hapi.protocol.wire.OptionalField +import app.hapi.protocol.wire.hasExplicitNullInvokedAt +import app.hapi.protocol.wire.isUserMessage +import app.hapi.protocol.wire.stringOrNull +import kotlinx.serialization.KSerializer +import kotlinx.serialization.Serializable +import kotlinx.serialization.SerializationException +import kotlinx.serialization.builtins.ListSerializer +import kotlinx.serialization.descriptors.SerialDescriptor +import kotlinx.serialization.encoding.Decoder +import kotlinx.serialization.encoding.Encoder +import kotlinx.serialization.json.JsonDecoder +import kotlinx.serialization.json.JsonEncoder +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put + +/** + * Client-side send state of an optimistic user message. The web reference + * extends the wire `DecryptedMessage` with `status?: MessageStatus` + * (`web/src/types/api.ts`); the wire never carries it — servers echo rows + * without a status and the client re-attaches it during merge. + */ +enum class MessageStatus(val wire: String) { + Queued("queued"), + Sending("sending"), + Sent("sent"), + Failed("failed"), + ; + + companion object { + fun fromWire(value: String?): MessageStatus? = entries.firstOrNull { it.wire == value } + } +} + +/** + * One row of the message window: the wire [DecryptedMessage] plus the + * client-side [status]. Mirrors the web's `DecryptedMessage & {status?}`. + * + * Identity matters: `applyLatestResponse`'s request-baseline comparison is by + * **reference** (web `!==`), so transitions must only create new instances for + * rows they actually change — which every function in this package does. + */ +@Serializable(with = WindowMessageSerializer::class) +data class WindowMessage( + val wire: DecryptedMessage, + val status: MessageStatus? = null, +) { + val id: String get() = wire.id + val seq: Long? get() = wire.seq + val localId: String? get() = wire.localId + val createdAt: Long get() = wire.createdAt + + /** Position time `invokedAt ?? createdAt` (collapsed tri-state). */ + val positionAt: Long get() = wire.positionAt + + val invokedAtOrNull: Long? get() = wire.invokedAtOrNull + + /** A row is optimistic iff it has a localId and `id === localId`. */ + val isOptimistic: Boolean + get() = wire.localId != null && wire.id == wire.localId + + /** + * Web `isQueuedForInvocation`: a user message whose `invokedAt` is an + * explicit `null` and whose send did not fail. Only these rows sit in the + * queued bar and survive window trims. + */ + val isQueuedForInvocation: Boolean + get() = wire.isUserMessage && wire.hasExplicitNullInvokedAt && status != MessageStatus.Failed + + /** Copy with `invokedAt` stamped to an explicit number. */ + fun withInvokedAt(invokedAt: Long): WindowMessage = + copy(wire = wire.copy(invokedAt = OptionalField.Present(invokedAt))) +} + +fun DecryptedMessage.asWindowMessage(status: MessageStatus? = null): WindowMessage = + WindowMessage(this, status) + +/** + * Serializes as the wire object plus an optional `status` key — the same + * shape the web persists (and the pagination fixtures use for op inputs). + */ +internal object WindowMessageSerializer : KSerializer { + override val descriptor: SerialDescriptor = + SerialDescriptor("app.hapi.protocol.window.WindowMessage", JsonObject.serializer().descriptor) + + override fun deserialize(decoder: Decoder): WindowMessage { + val input = decoder as? JsonDecoder + ?: throw SerializationException("WindowMessage supports JSON only") + val obj = input.decodeJsonElement() as? JsonObject + ?: throw SerializationException("WindowMessage must be a JSON object") + return WindowMessage( + wire = input.json.decodeFromJsonElement(DecryptedMessage.serializer(), obj), + status = MessageStatus.fromWire(obj["status"].stringOrNull), + ) + } + + override fun serialize(encoder: Encoder, value: WindowMessage) { + val output = encoder as? JsonEncoder + ?: throw SerializationException("WindowMessage supports JSON only") + val wireObject = output.json.encodeToJsonElement(DecryptedMessage.serializer(), value.wire) as JsonObject + output.encodeJsonElement(buildJsonObject { + wireObject.forEach { (key, element) -> put(key, element) } + value.status?.let { put("status", JsonPrimitive(it.wire)) } + }) + } +} + +/** + * Builds the optimistic row appended on send, mirroring + * `createOptimisticMessage` in `web/src/hooks/mutations/useSendMessage.ts` + * and the contract's "Optimistic sends" lifecycle: `id = localId`, + * `seq = null`, explicit `invokedAt: null` (so the strict-null queued check + * matches), content `{role:'user', content:{type:'text', text, attachments?}, + * meta:{deliveryMode}}`. + */ +fun buildOptimisticMessage( + localId: String, + text: String, + createdAt: Long, + attachments: List? = null, + scheduledAt: Long? = null, + deliveryMode: String = "queue", + status: MessageStatus = MessageStatus.Sending, +): WindowMessage { + val content = buildJsonObject { + put("role", "user") + put("content", buildJsonObject { + put("type", "text") + put("text", text) + if (attachments != null) { + put( + "attachments", + app.hapi.protocol.wire.HapiJson.encodeToJsonElement( + ListSerializer(AttachmentMetadata.serializer()), + attachments, + ), + ) + } + }) + put("meta", buildJsonObject { put("deliveryMode", deliveryMode) }) + } + return WindowMessage( + wire = DecryptedMessage( + id = localId, + seq = null, + localId = localId, + content = content, + createdAt = createdAt, + invokedAt = OptionalField.Present(null), + scheduledAt = scheduledAt, + ), + status = status, + ) +} diff --git a/android/core/protocol/src/main/kotlin/app/hapi/protocol/wire/MessageFlags.kt b/android/core/protocol/src/main/kotlin/app/hapi/protocol/wire/MessageFlags.kt new file mode 100644 index 00000000..f5dc647f --- /dev/null +++ b/android/core/protocol/src/main/kotlin/app/hapi/protocol/wire/MessageFlags.kt @@ -0,0 +1,23 @@ +package app.hapi.protocol.wire + +/** + * Wire-level message predicates shared by the window state machine and (later) + * the chat pipeline. Ported from `web/src/lib/messages.ts`. + */ + +/** + * `true` when [DecryptedMessage.content] is a role-wrapped envelope with + * `role: "user"` (web `isUserMessage`). Arrays and primitives have no `role` + * key and are not user messages. + */ +val DecryptedMessage.isUserMessage: Boolean + get() = content.objOrNull?.get("role").stringOrNull == "user" + +/** + * `true` when `invokedAt` is an **explicit** wire `null` — the strict-null + * half of the queued predicate (`docs/api/client-contract/pagination.md` + * "Queued semantics"). An absent field (pre-V8 hubs) means already-invoked + * and returns `false`. + */ +val DecryptedMessage.hasExplicitNullInvokedAt: Boolean + get() = invokedAt is OptionalField.Present && invokedAt.value == null diff --git a/android/core/protocol/src/test/kotlin/app/hapi/protocol/window/MessageMergeTest.kt b/android/core/protocol/src/test/kotlin/app/hapi/protocol/window/MessageMergeTest.kt new file mode 100644 index 00000000..1cc8f4f0 --- /dev/null +++ b/android/core/protocol/src/test/kotlin/app/hapi/protocol/window/MessageMergeTest.kt @@ -0,0 +1,121 @@ +package app.hapi.protocol.window + +import app.hapi.protocol.wire.DecryptedMessage +import app.hapi.protocol.wire.OptionalField +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put + +/** + * Merge/retention edges the pagination fixtures deliberately avoid: full + * position ties (the fixtures keep every `(at, seq)` pair distinct so the + * web's `localeCompare` never decides an order — natives pin ASCII), the 10s + * no-echo dedup fallback, invokedAt preservation across a stale refetch, and + * the skippable/codex retention predicate. + */ +class MessageMergeTest { + + private fun userRow( + id: String, + createdAt: Long, + localId: String? = null, + seq: Long? = null, + invokedAt: OptionalField = OptionalField.Present(null), + status: MessageStatus? = null, + ): WindowMessage = WindowMessage( + wire = DecryptedMessage( + id = id, + seq = seq, + localId = localId, + createdAt = createdAt, + invokedAt = invokedAt, + content = buildJsonObject { + put("role", "user") + put("content", buildJsonObject { + put("type", "text") + put("text", id) + }) + }, + ), + status = status, + ) + + @Test + fun `full position ties break by ASCII id comparison`() { + // 'B' (0x42) sorts before 'a' (0x61) in ASCII; locale-aware collation + // (the web's localeCompare) would typically order them the other way. + val rows = listOf( + userRow("row-a", createdAt = 1000), + userRow("row-B", createdAt = 1000), + ) + val merged = MessageMerge.mergeMessages(emptyList(), rows) + assertEquals(listOf("row-B", "row-a"), merged.map { it.id }) + } + + @Test + fun `an optimistic sent row is dropped when a server user row lands within ten seconds`() { + val optimistic = userRow( + "local-1", + createdAt = 10_000, + localId = "local-1", + status = MessageStatus.Sent, + ) + // Server row WITHOUT a localId echo, 9 999 ms away by position. + val server = userRow("srv-9", createdAt = 19_999, seq = 9, invokedAt = OptionalField.Absent) + val merged = MessageMerge.mergeMessages(listOf(optimistic), listOf(server)) + assertEquals(listOf("srv-9"), merged.map { it.id }) + + // At exactly 10 000 ms the fallback does not fire. + val far = userRow("srv-far", createdAt = 20_000, seq = 10, invokedAt = OptionalField.Absent) + val kept = MessageMerge.mergeMessages(listOf(optimistic), listOf(far)) + assertEquals(listOf("local-1", "srv-far"), kept.map { it.id }) + } + + @Test + fun `a locally-known invokedAt survives a stale refetch of the same row`() { + val stamped = userRow("srv-1", createdAt = 1000, seq = 1, invokedAt = OptionalField.Present(5000)) + val stale = userRow("srv-1", createdAt = 1000, seq = 1, invokedAt = OptionalField.Present(null)) + val merged = MessageMerge.mergeMessages(listOf(stamped), listOf(stale)) + assertEquals(5000L, merged.single().invokedAtOrNull) + } + + @Test + fun `retention hides skippable output and unrenderable codex payloads only`() { + fun agentContent(inner: kotlinx.serialization.json.JsonObject) = buildJsonObject { + put("role", "agent") + put("content", inner) + } + + // Meta system output — hidden. + assertTrue( + !MessageRetention.isRenderable(agentContent(buildJsonObject { + put("type", "output") + put("data", buildJsonObject { + put("type", "system") + put("isMeta", true) + }) + })) + ) + // codex row with no message payload — hidden. + assertTrue( + !MessageRetention.isRenderable(agentContent(buildJsonObject { + put("type", "codex") + put("data", buildJsonObject { put("type", "unknown-kind") }) + })) + ) + // codex text message — renderable. + assertTrue( + MessageRetention.isRenderable(agentContent(buildJsonObject { + put("type", "codex") + put("data", buildJsonObject { + put("type", "message") + put("message", "hi") + }) + })) + ) + // Unparseable envelope falls back to a stringify bubble — renderable. + assertTrue(MessageRetention.isRenderable(buildJsonObject { put("whatever", 1) })) + } +}