diff --git a/AGENTS.md b/AGENTS.md index b28aa86..8432cf9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -415,3 +415,12 @@ belongs in `~/.claude/TOOLCHAIN.md` or `~/.claude/MACHINE.md` instead. shape: the `markdownIn` scan that decides *what* to parse ran before the hop to `Dispatchers.Default`. The shape to watch for is a `withContext` that wraps the *fetch* and leaves the work done with the result outside it. +- **A transcript snapshot cannot survive a suspension and then be assigned.** + `loadOlderPage` joined its page to `items`, suspended while `warm` parsed + markdown, and then assigned the joined snapshot. An SSE event arriving in + that gap appeared and vanished; reopening brought it back because the + transcript and cache had it all along. Warm against a candidate if needed, + then join against the current `items` and assign without another suspension. + Also keep the page's original `oldestSeq`: a stream reset while the fetch or + warm is suspended makes the page stale, and it must be discarded rather + than joined into the reset window. diff --git a/PLAN.md b/PLAN.md index 8f1cacb..28519f0 100644 --- a/PLAN.md +++ b/PLAN.md @@ -397,6 +397,22 @@ in the right place. The echo driver models both shapes: `/peer` and ### Taking a queued message back (2026-08-31) +**Pressing Send always makes a quiet local bubble first (2026-09-10).** It remains until the +provider's `UserMessage` records that the message was received. A server `MessageQueued` replaces +the local bridge with its durable queue entry rather than adding a second bubble; an immediate +`UserMessage` removes it directly. If the request cannot reach the server, the local bubble stays +and carries that network failure underneath the message. Local bridges without a successful server +response are stored per server and session on the phone, so leaving and reopening the screen cannot +eat the only copy. Once the server accepts the request, the bubble remains in memory until the +provider event but the phone stops storing it: ownership has crossed to the server, whose transcript +and driver state survive the screen. Accepted queued messages remain the server transcript's fact +and are replayed from it on every device. + +Reconciliation uses the first local message with the same text and attachments because the current +message route has no caller-supplied id. Identical sends are therefore consumed in wire order. A +client id on the route and events would make cross-device identical simultaneous sends unambiguous, +but expanding the protocol solely for a transient display bridge was rejected. + `POST /sessions/{id}/unqueue`, answered by `Driver::unqueue` and recorded as `Event::MessageDropped` so every device loses the bubble and a reconnect does not replay it. diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/PendingMessages.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/PendingMessages.kt new file mode 100644 index 0000000..56584db --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/PendingMessages.kt @@ -0,0 +1,116 @@ +package com.example.aiapp + +import android.content.Context +import androidx.core.content.edit +import java.util.UUID +import org.json.JSONArray +import org.json.JSONObject + +private const val PENDING_MESSAGES = "pending-messages" + +/** A quiet user bubble below the durable transcript. */ +internal data class QueuedMessage( + val id: String, + val text: String, + val attachments: List, + val refusal: String? = null, + /** This phone is still waiting for any durable event that says the server accepted it. */ + val local: Boolean = false, + /** The HTTP request returned successfully; the provider event is still outstanding. */ + val serverAccepted: Boolean = false, +) + +internal fun localPendingMessage(text: String, attachments: List) = + QueuedMessage("local-${UUID.randomUUID()}", text, attachments, local = true) + +private fun QueuedMessage.matches(text: String, attachments: List) = + this.text == text && this.attachments == attachments + +/** Replaces the local bridge with the server's durable waiting message, without drawing both. */ +internal fun reconcileQueuedMessage( + queued: List, + event: SessionEvent.MessageQueued, +): List { + if (queued.any { !it.local && it.id == event.id }) return queued + val at = queued.indexOfFirst { it.local && it.matches(event.text, event.attachments) } + if (at < 0) return queued + QueuedMessage(event.id, event.text, event.attachments) + return queued.mapIndexed { index, message -> + if (index == at) QueuedMessage(event.id, event.text, event.attachments) else message + } +} + +/** Removes exactly the pending bubble that became a provider-received user message. */ +internal fun reconcileUserMessage( + queued: List, + event: SessionEvent.UserMessage, +): List { + val at = + event.id?.let { id -> queued.indexOfFirst { !it.local && it.id == id }.takeIf { it >= 0 } } + ?: queued.indexOfFirst { it.local && it.matches(event.text, event.attachments) } + return if (at < 0) queued else queued.filterIndexed { index, _ -> index != at } +} + +/** Keeps a failed send in place and puts its actionable failure in that message's bubble. */ +internal fun markPendingFailure( + queued: List, + id: String, + failure: String, +): List = queued.map { message -> + if (message.local && message.id == id) message.copy(refusal = failure) else message +} + +/** Stops persisting a send once the server owns it, while its bubble awaits the provider event. */ +internal fun markPendingAccepted(queued: List, id: String): List = + queued.map { message -> + if (message.local && message.id == id) message.copy(serverAccepted = true) else message + } + +/** Restores sends for which this phone has not yet seen a durable server event. */ +internal fun loadPendingMessages(context: Context, key: String): List { + val encoded = + context.getSharedPreferences(PENDING_MESSAGES, Context.MODE_PRIVATE).getString(key, null) + ?: return emptyList() + return try { + val messages = JSONArray(encoded) + List(messages.length()) { index -> + val message = messages.getJSONObject(index) + val attachments = message.optJSONArray("attachments") ?: JSONArray() + QueuedMessage( + id = message.getString("id"), + text = message.getString("text"), + attachments = List(attachments.length()) { attachments.getString(it) }, + refusal = message.optString("refusal").takeIf { it.isNotEmpty() }, + local = true, + ) + } + } catch (_: org.json.JSONException) { + // A corrupt local outbox is not useful on the next open either. Remove it rather than + // repeatedly pretending it decoded to an intentionally empty one. + context.getSharedPreferences(PENDING_MESSAGES, Context.MODE_PRIVATE).edit { remove(key) } + emptyList() + } +} + +/** Stores only sends the server has not confirmed; everything accepted is the server's to keep. */ +internal fun savePendingMessages(context: Context, key: String, queued: List) { + val local = queued.filter { it.local && !it.serverAccepted } + context.getSharedPreferences(PENDING_MESSAGES, Context.MODE_PRIVATE).edit { + if (local.isEmpty()) { + remove(key) + } else { + putString( + key, + JSONArray( + local.map { message -> + JSONObject() + .put("id", message.id) + .put("text", message.text) + .put("attachments", JSONArray(message.attachments)) + .put("refusal", message.refusal ?: "") + } + ) + .toString(), + ) + } + } +} diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index a387a6b..2d2723b 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -349,10 +349,20 @@ fun SessionScreen( // the newest end and then travelling to the anchor is exactly the journey a reader must never // see. var restoring by remember(address, epoch) { mutableStateOf(savedAnchor != null) } - // Messages the server has taken and the session has not read yet, by the id that will resolve - // them. From the event stream rather than from what this screen sent, so they survive leaving - // the session -- and a message sent from another device is drawn waiting on this one too. - var queued by remember { mutableStateOf(listOf()) } + // Messages not yet recorded as received by the provider. The local ones bridge Send to the + // server's first durable event (and hold a network failure); the rest are reconstructed from + // that event stream, so another device's queued message is drawn here too. + val pendingKey = + remember(settings, address) { "${settings.host}:${settings.port}/${address.cachePath}" } + var queued by remember(pendingKey) { mutableStateOf(loadPendingMessages(context, pendingKey)) } + + /** + * Changes the visible outbox and keeps only its not-yet-durable part across a screen reopen. + */ + fun replaceQueued(messages: List) { + queued = messages + savePendingMessages(context, pendingKey, messages) + } // Commands the session has been asked to run and cannot yet, by the id that will resolve them. // From the server, so a rename sent from another device is drawn waiting here too. var waitingCommands by remember { mutableStateOf(listOf>()) } @@ -419,7 +429,9 @@ fun SessionScreen( held = listOf() oldestSeq = 0L moreHistory = true - queued = listOf() + // A send the server could not accept is this phone's only copy. A reset replaces server + // state, not that local outbox, so dropping it here would eat the message a second time. + replaceQueued(queued.filter { it.local }) waitingCommands = listOf() } @@ -449,15 +461,15 @@ fun SessionScreen( // Waiting, then read. Matched by id: the same message sent twice is two bubbles, and // clearing by text would take away whichever matched first. if (event is SessionEvent.MessageQueued) { - queued = queued + QueuedMessage(event.id, event.text, event.attachments) + replaceQueued(reconcileQueuedMessage(queued, event)) } if (event is SessionEvent.UserMessage) { - queued = queued.filterNot { it.id == event.id } + replaceQueued(reconcileUserMessage(queued, event)) } // Waiting, then taken back. From the server rather than from the tap, so every device drops // the bubble and a reconnect does not put back one that was cancelled. if (event is SessionEvent.MessageDropped) { - queued = queued.filterNot { it.id == event.id } + replaceQueued(queued.filterNot { it.id == event.id }) } // Waiting, then gone: a command leaves this list when the session takes it, and the row it // becomes is added by `foldEvent` in the same pass. @@ -597,7 +609,8 @@ fun SessionScreen( // what a phone over the tunnel costs, it won the race. // // Guarded here rather than at the two callers because it is a fact about the question. - if (oldestSeq == 0L) return false + val before = oldestSeq + if (before == 0L) return false // The fetch *and* the fold, both off the thread that draws. Only the fetch used to be, and // the fold is the expensive half: `foldEvent` returns a new list per event, so a page is // that many copies of a growing list -- around three hundred thousand element copies, run @@ -607,7 +620,7 @@ fun SessionScreen( // errand. Neither half touches anything the composition owns. val page = withContext(Dispatchers.IO) { - val older = source.page(before = oldestSeq, limit = limit, coalesce = coalesce) + val older = source.page(before = before, limit = limit, coalesce = coalesce) if (older.isEmpty()) return@withContext null // Folded oldest-first into a list of their own, then put in front: `foldEvent` // merges streaming text into the item before it, so replaying an older page through @@ -620,21 +633,29 @@ fun SessionScreen( } older.first().seq to earlier } + // A stream reset can replace the transcript while the page request is out. Its answer is a + // page of the view that no longer exists, so it must not decide anything about the new one. + if (oldestSeq != before) return false if (page == null) { moreHistory = false return false } val (oldest, earlier) = page - oldestSeq = oldest - moreHistory = oldestSeq > 1L // Joined here rather than above, because it is the one step that reads what is already // loaded: `items` must be read where it is written. - val joined = joinPages(earlier, items) + val joinedForWarm = joinPages(earlier, items) // After the join rather than on the page alone: a boundary that fell through a reply leaves // `joinPages` holding a message made of both halves, and that text has existed for no time // at all. Warming the page by itself warmed the two halves and missed the one thing drawn. - warm(replies, joined) - items = joined + warm(replies, joinedForWarm) + // `warm` suspends. A live event can be recorded while it parses, so joining the page before + // it and then assigning that snapshot afterward used to erase the event from this visit -- + // the durable transcript still held it, which is why reopening made a sent message return. + // Read and write `items` together after the suspension instead. + if (oldestSeq != before) return false + oldestSeq = oldest + moreHistory = oldestSeq > 1L + items = joinPages(earlier, items) return true } @@ -1104,7 +1125,7 @@ fun SessionScreen( } catch (e: ApiException) { e.message ?: "this message could not be taken back" } - queued = queued.map { if (it.id == messageId) it.copy(refusal = refusal) else it } + replaceQueued(queued.map { if (it.id == messageId) it.copy(refusal = refusal) else it }) } } @@ -1184,10 +1205,23 @@ fun SessionScreen( input = atEnd("") saveDraft(context, summary.id, "") pendingAttachments = emptyList() - // Nothing is added here. The server says what is waiting -- it emits `messageQueued` when - // it takes a message it cannot deliver yet -- and this screen draws that. Holding a local - // copy as well was the bug: the two agreed only until the app was restarted. - act { sendMessage(settings, summary.id, text, attachments) } + // Clearing the field makes one promise: its contents are either waiting in this quiet + // bubble or recorded as a user message. The server's `messageQueued` replaces this local + // bridge when it arrives; an immediate `userMessage` removes it. Until one does, even a + // network failure leaves the words visible with the failure attached to them. + val pending = localPendingMessage(text, attachments) + replaceQueued(queued + pending) + scope.launch { + try { + withContext(Dispatchers.IO) { sendMessage(settings, summary.id, text, attachments) } + replaceQueued(markPendingAccepted(queued, pending.id)) + actionError = null + } catch (e: ApiException) { + replaceQueued( + markPendingFailure(queued, pending.id, e.message ?: "message not sent") + ) + } + } } // One path for everything attached, however it arrived: the photo picker, the file chooser or @@ -1475,7 +1509,9 @@ fun SessionScreen( // produces, not here: the server knows whether the // message was still its to take back, and the other // devices have to be told by the same event. - onTakeBack = { takeBack(waiting.id) }, + onTakeBack = + if (waiting.local) null + else ({ takeBack(waiting.id) }), ) } } @@ -2082,12 +2118,13 @@ private fun UserChunkRow( /** * A message the person holding the phone sent, in a bubble at their end of the conversation. * - * [pending] is one the server has taken and the session has not read yet -- drawn quieter, because - * "said" and "heard" are different claims and the transcript must not merge them. + * [pending] has not yet been recorded as received by the provider -- drawn quieter, because + * "pressed Send" and "heard" are different claims and the transcript must not merge them. * - * A pending bubble is tappable: [onTakeBack] asks the server to drop the message before the session - * reads it, and [refusal] is what came back when it would not. The refusal is drawn here rather - * than with the screen's other errors because this is where the reader pressed. + * A server-accepted pending bubble is tappable: [onTakeBack] asks the server to drop the message + * before the session reads it. A local one is not, because its request may still be in flight. + * [refusal] is either that take-back refusal or the send's network failure. It is drawn here rather + * than with the screen's other errors because this is the message the failure belongs to. * * A settled message longer than [USER_SPLIT_CHARS] is drawn as [UserChunkRow] slices instead -- one * `Text` holding a pasted log is a hundred-thousand-pixel layout in the frame the row scrolls into. @@ -2165,13 +2202,6 @@ private fun UserBubble( * [refusal] is why taking it back did not work, kept per message rather than on the screen: two * bubbles can be waiting at once, and an error above them both would not say which. */ -private data class QueuedMessage( - val id: String, - val text: String, - val attachments: List, - val refusal: String? = null, -) - /** * Whether a model switch has anything to warn about -- see [ModelSwitchWarning]. * diff --git a/app/androidApp/src/test/kotlin/com/example/aiapp/PendingMessagesTest.kt b/app/androidApp/src/test/kotlin/com/example/aiapp/PendingMessagesTest.kt new file mode 100644 index 0000000..4f3558f --- /dev/null +++ b/app/androidApp/src/test/kotlin/com/example/aiapp/PendingMessagesTest.kt @@ -0,0 +1,62 @@ +package com.example.aiapp + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class PendingMessagesTest { + private fun local(text: String = "keep this") = + QueuedMessage("local-1", text, emptyList(), local = true) + + @Test + fun a_server_queue_replaces_the_local_bridge_instead_of_duplicating_it() { + val queued = + reconcileQueuedMessage( + listOf(local()), + SessionEvent.MessageQueued("server-1", "keep this", emptyList()), + ) + + assertEquals(1, queued.size) + assertEquals("server-1", queued.single().id) + assertTrue(!queued.single().local) + } + + @Test + fun an_immediately_received_message_removes_its_local_bridge() { + val queued = + reconcileUserMessage( + listOf(local()), + SessionEvent.UserMessage("keep this", id = null, attachments = emptyList()), + ) + + assertTrue(queued.isEmpty()) + } + + @Test + fun a_transport_failure_stays_on_its_message() { + val queued = markPendingFailure(listOf(local()), "local-1", "Can't reach the server") + + assertEquals("Can't reach the server", queued.single().refusal) + assertTrue(queued.single().local) + } + + @Test + fun server_acceptance_keeps_the_bubble_until_the_provider_event() { + val queued = markPendingAccepted(listOf(local()), "local-1") + + assertEquals(1, queued.size) + assertTrue(queued.single().serverAccepted) + } + + @Test + fun identical_messages_are_reconciled_one_at_a_time() { + val queued = listOf(local(), local().copy(id = "local-2")) + val afterFirst = + reconcileUserMessage( + queued, + SessionEvent.UserMessage("keep this", id = null, attachments = emptyList()), + ) + + assertEquals(listOf("local-2"), afterFirst.map { it.id }) + } +}