Keep sent messages visible until received

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.

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.

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<String>,
+    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<String>) =
+    QueuedMessage("local-${UUID.randomUUID()}", text, attachments, local = true)
+
+private fun QueuedMessage.matches(text: String, attachments: List<String>) =
+    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<QueuedMessage>,
+    event: SessionEvent.MessageQueued,
+): List<QueuedMessage> {
+    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<QueuedMessage>,
+    event: SessionEvent.UserMessage,
+): List<QueuedMessage> {
+    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<QueuedMessage>,
+    id: String,
+    failure: String,
+): List<QueuedMessage> = 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<QueuedMessage>, id: String): List<QueuedMessage> =
+    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<QueuedMessage> {
+    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<QueuedMessage>) {
+    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(),
+            )
+        }
+    }
+}

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<QueuedMessage>()) }
+    // 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<QueuedMessage>) {
+        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<Pair<String, String>>()) }
@@ -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<String>,
-    val refusal: String? = null,
-)
-
 /**
  * Whether a model switch has anything to warn about -- see [ModelSwitchWarning].
  *

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