merge: B-M2c Android message window store, pagination fixtures green

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 {

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<String>): QueuedStateResponse =
+    override suspend fun getQueuedState(sessionId: String, localIds: List<String>): QueuedStateResponse =
         request(
             "POST",
             url("api", "sessions", sessionId, "messages", "queued-state").build(),

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<String>): QueuedStateResponse
+}

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<WindowMessage>,
+    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<MessageWindowState> = _state.asStateFlow()
+
+    /**
+     * UI-facing projection — the fields the web `useMessages` hook returns;
+     * cursor/generation internals stay off the render path.
+     */
+    val uiState: Flow<MessageWindowUiState> = 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<Unit>
+    }
+
+    private sealed interface SyncDecision {
+        data class Await(val job: Deferred<Unit>) : SyncDecision
+        data class Drain(val observed: Deferred<Unit>) : 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<Unit> {
+        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<Unit>) {
+        val epoch = synchronized(controllerLock) { controllerEpoch }
+        var current: Deferred<Unit>? = 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<WindowMessage>) {
+        if (messages.isEmpty()) return
+        update { MessageWindowLogic.ingestIncoming(it, messages) }
+        persist()
+    }
+
+    /** SSE `messages-consumed` (web `markMessagesConsumed`). */
+    suspend fun markConsumed(localIds: List<String>, 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<AttachmentMetadata>? = 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<String> =
+        MessageWindowLogic.queuedReconcileCandidateLocalIds(_state.value)
+
+    suspend fun reconcileQueuedLocalIds(candidateLocalIds: List<String>, queuedLocalIds: List<String>) {
+        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<String>()
+        val invokedLocalMessages = mutableListOf<Pair<String, Long>>()
+        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<Long, MutableList<String>>()
+        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<String, WindowMessage> =
+        _state.value.messages.associateBy { it.id }
+
+    private fun MessagesResponse.windowMessages(): List<WindowMessage> =
+        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
+    }
+}

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<String, MessageWindowStore>()
+
+    /** 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()
+    }
+}

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"
+    }
+}

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<DecryptedMessage>,
+        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<DecryptedMessage>,
+        epoch: Long,
+        nextAfter: Pair<Long, Long>?,
+        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<MessagesQuery>()
+        private val responses = Channel<MessagesResponse>(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<String>): 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 })
+    }
+}

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<String> {
+            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<Long, MutableList<String>>()
+                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<MessagesResponse> =
+        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<MessagesResponse>()
+    val requests = mutableListOf<JsonObject>()
+
+    val remaining: Int get() = queue.size
+
+    fun begin(responses: List<MessagesResponse>) {
+        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<String>): 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)
+        }
+    }
+}

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<WindowMessage> = 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<WindowMessage>, incoming: List<WindowMessage>): List<WindowMessage> {
+        if (existing.isEmpty()) {
+            return incoming.sortedWith(comparator)
+        }
+        if (incoming.isEmpty()) {
+            return existing.sortedWith(comparator)
+        }
+
+        val byId = LinkedHashMap<String, WindowMessage>()
+        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<WindowMessage> = 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<String, MessageStatus>()
+            val optimisticInvokedAtByLocalId = HashMap<String, Long?>()
+            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)
+    }
+}

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<JsonElement> = 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() }
+                )
+        }
+    }
+}

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<WindowMessage>): 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<WindowMessage>, 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<WindowMessage>, val dropped: List<WindowMessage>)
+
+    private fun sliceForTrim(items: List<WindowMessage>, 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<WindowMessage>, 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<WindowMessage>): 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<WindowMessage>,
+        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<WindowMessage>,
+        page: MessagesPage,
+        replaceServerRows: Boolean,
+        requestBaseline: Map<String, WindowMessage>,
+    ): 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<WindowMessage>,
+        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<WindowMessage>,
+        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<WindowMessage>): 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<String>, 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<String> {
+        val localIds = LinkedHashSet<String>()
+        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<String>,
+        queuedLocalIds: List<String>,
+    ): 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,
+            )
+}

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<MessagePosition> {
+    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<WindowMessage> = 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<WindowMessage> = 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,
+)

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<WindowMessage> {
+    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<AttachmentMetadata>? = 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,
+    )
+}

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

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<Long?> = 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) }))
+    }
+}