Merge remote-tracking branch 'origin/main'

app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt

@@ -20,7 +20,28 @@ data class SeqEvent(val seq: Long, val ts: Double, val event: SessionEvent)
 data class QuestionOption(val label: String, val description: String?, val preview: String?)
 
 sealed class SessionEvent {
-    data class UserMessage(val text: String) : SessionEvent()
+    data class UserMessage(
+        val text: String,
+        /**
+         * The [MessageQueued] this resolves, or null when it never waited.
+         *
+         * Matched on rather than the text, because the same message sent twice is two waiting
+         * bubbles and clearing whichever one matched first would leave the wrong one on screen.
+         */
+        val id: String?,
+    ) : SessionEvent()
+
+    /**
+     * A message the server has accepted and the session has not read yet.
+     *
+     * From the server, not from this app's memory of what it sent. The pending bubble used to be
+     * screen state, so leaving the session or restarting the app drew nothing waiting while the
+     * message was still queued -- and nothing waiting is what "there is nothing" looks like.
+     *
+     * Resolved by the [UserMessage] carrying the same id, exactly as [CommandQueued] is resolved by
+     * [CommandSent].
+     */
+    data class MessageQueued(val id: String, val text: String) : SessionEvent()
 
     data class AssistantText(val delta: String) : SessionEvent()
 
@@ -119,7 +140,13 @@ fun parseSeqEvent(json: String): SeqEvent {
     val body = JSONObject(json)
     val event =
         when (val type = body.getString("type")) {
-            "userMessage" -> SessionEvent.UserMessage(body.getString("text"))
+            "userMessage" ->
+                SessionEvent.UserMessage(
+                    body.getString("text"),
+                    body.optString("id").ifEmpty { null },
+                )
+            "messageQueued" ->
+                SessionEvent.MessageQueued(body.getString("id"), body.getString("text"))
             "assistantText" -> SessionEvent.AssistantText(body.getString("delta"))
             "toolStart" ->
                 SessionEvent.ToolStart(

app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt

@@ -269,6 +269,9 @@ fun foldEvent(items: List<TranscriptItem>, entry: SeqEvent): List<TranscriptItem
         is SessionEvent.CommandSent -> items + TranscriptItem.CommandRow(entry.seq, event.text)
         // Screen-level state, not transcript rows -- see SessionScreen.
         is SessionEvent.CommandQueued -> items
+        // No row of its own: a message that is still waiting is drawn as a pending bubble below
+        // the transcript, and becomes an ordinary one where the session read it.
+        is SessionEvent.MessageQueued -> items
         is SessionEvent.Settings -> items
         is SessionEvent.Status -> items
         is SessionEvent.Error -> items + TranscriptItem.ErrorMsg(entry.seq, event.message)
@@ -373,7 +376,11 @@ fun SessionScreen(
     // the working indicator, because that is where it is in the session's
     // reading of events: after everything taken in, not yet taken in
     // itself.
-    var queued by remember { mutableStateOf(listOf<String>()) }
+    // 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 are still here
+    // after leaving the session or restarting the app -- and so a message sent from another device
+    // is drawn waiting on this one too.
+    var queued by remember { mutableStateOf(listOf<Pair<String, String>>()) }
     // Commands the session has been asked to run and cannot yet, by the id that will resolve
     // them. From the server rather than from this screen, so a rename sent from the settings
     // screen -- or from another device -- is drawn waiting here too.
@@ -426,7 +433,14 @@ fun SessionScreen(
                 // all that distinguishes one message from an identical
                 // earlier one -- and only the first match, so two
                 // identical messages wait twice.
-                if (event is SessionEvent.UserMessage) queued = queued - event.text
+                // 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 + (event.id to event.text)
+                }
+                if (event is SessionEvent.UserMessage) {
+                    queued = queued.filterNot { it.first == 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.
                 if (event is SessionEvent.CommandQueued) {
@@ -689,18 +703,11 @@ fun SessionScreen(
         input = ""
         saveDraft(context, summary.id, "")
         pendingAttachments = emptyList()
-        if (running && text.isNotEmpty()) queued = queued + text
-        // A held message leaves this list exactly two ways: the session
-        // reads it, which comes back as a UserMessage (see `apply`), or the
-        // send itself failed and there is nothing to wait for. Clearing the
-        // whole list when a turn ended was neither -- the server holds a
-        // queue of its own and takes one message per turn, so ending a turn
-        // is precisely when the *rest* are still waiting. It wiped them off
-        // the screen while they were on their way, which reads as messages
-        // two and three having been dropped.
-        act(onFailure = { queued = queued - text }) {
-            sendMessage(settings, summary.id, text, attachments)
-        }
+        // 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 or the
+        // session left, and then the screen showed nothing pending while the queue was full.
+        act { sendMessage(settings, summary.id, text, attachments) }
     }
 
     // The system photo picker; the image uploads as soon as it's chosen,
@@ -827,7 +834,7 @@ fun SessionScreen(
                             waitingCommands.forEach { (_, text) ->
                                 CommandBubble(text, waiting = true)
                             }
-                            queued.forEach { text -> UserBubble(text, pending = true) }
+                            queued.forEach { (_, text) -> UserBubble(text, pending = true) }
                         }
                     }
                 }

server/src/session/claude.rs

@@ -136,8 +136,10 @@ const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5);
 struct Queue {
     /// A turn is in flight, so a message sent now is a steer into it.
     running: bool,
-    /// Written, not yet announced, oldest first.
-    awaiting: VecDeque<String>,
+    /// Written, not yet announced, oldest first, each with the id of the
+    /// `MessageQueued` that told the phone it was waiting -- so the
+    /// announcement can name which bubble it resolves.
+    awaiting: VecDeque<(String, String)>,
     /// The process is gone, so nothing can be taken up any more.
     ///
     /// Needed because every other way out of a turn is an `Idle` this
@@ -157,7 +159,7 @@ impl Queue {
     fn close(&mut self, sink: &EventSink, why: &str) {
         self.closed = true;
         self.running = false;
-        let lost: Vec<String> = self.awaiting.drain(..).collect();
+        let lost: Vec<String> = self.awaiting.drain(..).map(|(_, text)| text).collect();
         if lost.is_empty() {
             return;
         }
@@ -514,16 +516,25 @@ impl Driver for ClaudeDriver {
         if queue.running {
             // Into the running turn, now. Announced when the CLI shows it
             // has been round the model again -- see `Queue`.
-            queue.awaiting.push_back(text);
+            //
+            // The *waiting* is recorded here, though, which is the one
+            // thing that must not be left to the phone to remember: it put
+            // the bubble on screen from its own state, so leaving the
+            // session or restarting the app drew nothing pending while a
+            // message was still in the queue.
+            let id = super::random_hex();
+            queue.awaiting.push_back((id.clone(), text.clone()));
             drop(queue);
+            let _ = self.sink.send(Event::MessageQueued { id, text });
             self.send_line(line);
             return;
         }
         queue.running = true;
         drop(queue);
         // Nothing is in flight, so there is nothing to wait for: this
-        // message *is* the turn about to start.
-        let _ = self.sink.send(Event::MessageTaken { text });
+        // message *is* the turn about to start, and it never had a
+        // `MessageQueued` to resolve.
+        let _ = self.sink.send(Event::MessageTaken { id: None, text });
         let _ = self.sink.send(Event::Status {
             state: SessionStatus::Running,
         });
@@ -930,12 +941,15 @@ fn proves_a_turn(event: &Event) -> bool {
 /// [`translate_line`], and the pair is the whole of the rule -- a steer
 /// announced anywhere else lands above output that predates it.
 fn announce_steers(queue: &Arc<Mutex<Queue>>, sink: &EventSink) -> bool {
-    let taken: Vec<String> = {
+    let taken: Vec<(String, String)> = {
         let mut queue = queue.lock().unwrap();
         queue.awaiting.drain(..).collect()
     };
-    for text in taken {
-        if sink.send(Event::MessageTaken { text }).is_err() {
+    for (id, text) in taken {
+        if sink
+            .send(Event::MessageTaken { id: Some(id), text })
+            .is_err()
+        {
             return false;
         }
     }
@@ -1137,7 +1151,7 @@ mod tests {
                 .lock()
                 .unwrap()
                 .awaiting
-                .push_back("do the other one instead".into())
+                .push_back(("q1".into(), "do the other one instead".into()))
         });
 
         let at = |find: fn(&Event) -> bool| {
@@ -1147,6 +1161,16 @@ mod tests {
                 .unwrap_or_else(|| panic!("nothing matched in {events:?}"))
         };
         let taken = at(|e| matches!(e, Event::MessageTaken { .. }));
+        // Named, not just announced: the phone has a waiting bubble on screen for this message
+        // and clears the one with this id. Matching on the text instead would clear the wrong
+        // bubble whenever the same thing was sent twice.
+        assert!(
+            matches!(
+                &events[taken],
+                Event::MessageTaken { id: Some(id), .. } if id == "q1"
+            ),
+            "an announcement must name the queue entry it resolves: {events:?}"
+        );
         assert!(
             taken > at(|e| matches!(e, Event::ToolStart { .. })),
             "a steer must not sit above a call the model had already made: {events:?}"
@@ -1184,7 +1208,7 @@ mod tests {
                 .lock()
                 .unwrap()
                 .awaiting
-                .push_back("never mind".into())
+                .push_back(("q2".into(), "never mind".into()))
         });
 
         let taken = events
@@ -1347,8 +1371,8 @@ mod tests {
             running: true,
             ..Queue::default()
         };
-        queue.awaiting.push_back("first".into());
-        queue.awaiting.push_back("second".into());
+        queue.awaiting.push_back(("q1".into(), "first".into()));
+        queue.awaiting.push_back(("q2".into(), "second".into()));
         queue.close(&sink, "the session ended");
 
         // Named rather than counted, because these never reached the

server/src/session/driver.rs

@@ -69,6 +69,34 @@ pub enum Event {
     /// one stream. Recorded when the session reads the message, which is
     /// what `MessageTaken` reports.
     UserMessage {
+        /// The [`Event::MessageQueued`] this resolves, when it waited.
+        ///
+        /// A message sent between turns is read at once and never queued,
+        /// so this is `None` for most of them. It is the pair to the id on
+        /// `MessageQueued` and exists for the same reason `CommandSent`
+        /// carries one: the phone has a bubble on screen for the waiting
+        /// message and needs to know *which* one this is, rather than
+        /// matching on the text and clearing the wrong one when the same
+        /// thing was sent twice.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        id: Option<String>,
+        text: String,
+    },
+    /// A message accepted from the phone that the session cannot read yet.
+    ///
+    /// Recorded, unlike the message itself, and that difference is the
+    /// point. The *message* belongs in the transcript where the session
+    /// read it -- see `MessageTaken` -- but something has to say it is
+    /// waiting, and it has to be the server that says it: the phone used
+    /// to remember its own outgoing messages, so leaving the session
+    /// screen or restarting the app showed nothing pending when something
+    /// was, which reads as "nothing queued" rather than "I have forgotten".
+    ///
+    /// Carries no row of its own. It is resolved by the `UserMessage`
+    /// bearing the same id, exactly as `CommandQueued` is resolved by
+    /// `CommandSent`.
+    MessageQueued {
+        id: String,
         text: String,
     },
     /// A driver has taken one of the user's messages and started reading
@@ -82,6 +110,9 @@ pub enum Event {
     /// predates it, and leaves a phone drawing it as still waiting with
     /// nothing coming to say otherwise.
     MessageTaken {
+        /// The `MessageQueued` this answers, or `None` when it never
+        /// waited. Carried through onto the `UserMessage`.
+        id: Option<String>,
         text: String,
     },
     /// Streaming assistant text; the phone renders the concatenation as

server/src/session/echo.rs

@@ -73,7 +73,9 @@ pub struct EchoDriver {
     /// dropped to idle immediately, so a phone had nothing to show as
     /// pending. Holding it here is what makes echo able to stand in.
     busy: Arc<AtomicBool>,
-    queued: Arc<Mutex<Vec<String>>>,
+    /// Held messages with the id of the `MessageQueued` each one announced,
+    /// so the announcement can say which waiting bubble it resolves.
+    queued: Arc<Mutex<Vec<(String, String)>>>,
     /// Ids of the questions awaiting an answer, in the order they were
     /// asked. A list because `/ask` puts up to four on one tool call, the
     /// way AskUserQuestion does, and the turn resumes when the last of
@@ -207,7 +209,15 @@ impl EchoDriver {
         // session went idle the instant one arrived, and every state that
         // only exists while something is queued was untestable.
         if self.busy.load(Ordering::SeqCst) {
-            self.queued.lock().unwrap().push(text);
+            // The waiting is recorded, exactly as the real driver records
+            // it: the phone draws its pending bubbles from the server, so
+            // an echo session has to produce the same events or the states
+            // it exists to exercise are not the app's real ones.
+            let id = super::random_hex();
+            self.queued.lock().unwrap().push((id.clone(), text.clone()));
+            if announce {
+                self.emit(Event::MessageQueued { id, text });
+            }
             return;
         }
 
@@ -219,7 +229,10 @@ impl EchoDriver {
         // from the transcript is the one thing echo must not model.
         if let Some(rest) = text.strip_prefix("/peer") {
             if announce {
-                self.emit(Event::MessageTaken { text: text.clone() });
+                self.emit(Event::MessageTaken {
+                    id: None,
+                    text: text.clone(),
+                });
             }
             self.emit(Event::PeerMessage {
                 from: "dev-updater-f5".to_string(),
@@ -240,7 +253,7 @@ impl EchoDriver {
         // this is the typed path onto it.
         if text.trim() == "/compact" {
             if announce {
-                self.emit(Event::MessageTaken { text });
+                self.emit(Event::MessageTaken { id: None, text });
             }
             self.compact();
             return;
@@ -248,7 +261,7 @@ impl EchoDriver {
 
         if text.trim() == "/ask" {
             if announce {
-                self.emit(Event::MessageTaken { text });
+                self.emit(Event::MessageTaken { id: None, text });
             }
             self.ask_user_question();
             return;
@@ -322,7 +335,10 @@ impl EchoDriver {
             // message it thinks is still queued, and the point of an echo
             // provider is that it behaves like the real ones.
             if announce {
-                send(Event::MessageTaken { text: text.clone() });
+                send(Event::MessageTaken {
+                    id: None,
+                    text: text.clone(),
+                });
             }
             send(Event::Status {
                 state: SessionStatus::Running,
@@ -438,14 +454,17 @@ impl EchoDriver {
 /// moment a real CLI would have injected it. One place, because a turn has
 /// several ways to end (a reply, an interrupt, a compaction) and every one
 /// of them owes the same answer.
-fn finish_turn(sink: &EventSink, queued: &Mutex<Vec<String>>, busy: &AtomicBool) {
+fn finish_turn(sink: &EventSink, queued: &Mutex<Vec<(String, String)>>, busy: &AtomicBool) {
     let held = std::mem::take(&mut *queued.lock().unwrap());
-    for text in held {
+    for (id, text) in held {
         // Announced before it is answered, in that order: a phone showing
         // the message as pending needs the signal that it has been read,
         // and the answer is meaningless above a message still drawn as
         // waiting.
-        let _ = sink.send(Event::MessageTaken { text: text.clone() });
+        let _ = sink.send(Event::MessageTaken {
+            id: Some(id),
+            text: text.clone(),
+        });
         let _ = sink.send(Event::AssistantText {
             delta: format!("\n(taken from the queue) You said: {text}"),
         });

server/src/session/import.rs

@@ -538,7 +538,9 @@ fn push_user(events: &mut Vec<Event>, content: &Value, session_dir: &std::path::
     }
     let text = text_of(content);
     if !text.trim().is_empty() {
-        events.push(Event::UserMessage { text });
+        // Replayed from the CLI's own file: it was read long ago, so
+        // there is no waiting bubble for it to resolve.
+        events.push(Event::UserMessage { id: None, text });
     }
 }
 

server/src/session/llama.rs

@@ -344,7 +344,10 @@ impl Driver for LlamaDriver {
             // Nothing is ever held back here -- there is no queue to wait
             // in -- so the message is taken the moment it arrives. Said
             // anyway, because this is what records it: see `MessageTaken`.
-            let _ = sink.send(Event::MessageTaken { text: text.clone() });
+            let _ = sink.send(Event::MessageTaken {
+                id: None,
+                text: text.clone(),
+            });
             let _ = sink.send(Event::Status {
                 state: SessionStatus::Running,
             });
@@ -474,7 +477,7 @@ fn conversation(path: &Path) -> Vec<Message> {
     };
     for event in events.iter().cloned() {
         match event.event {
-            Event::UserMessage { text } => {
+            Event::UserMessage { text, .. } => {
                 if !pending.is_empty() {
                     messages.push(Message {
                         role: "assistant".into(),
@@ -631,6 +634,7 @@ mod tests {
     fn deltas_between_user_messages_are_one_assistant_turn() {
         let (_dir, path) = transcript_with(&[
             Event::UserMessage {
+                id: None,
                 text: "hello".into(),
             },
             Event::AssistantText {
@@ -643,6 +647,7 @@ mod tests {
                 state: SessionStatus::Idle,
             },
             Event::UserMessage {
+                id: None,
                 text: "again".into(),
             },
             Event::AssistantText {
@@ -673,6 +678,7 @@ mod tests {
     fn an_interrupted_reply_stays_in_the_conversation() {
         let (_dir, path) = transcript_with(&[
             Event::UserMessage {
+                id: None,
                 text: "count".into(),
             },
             Event::AssistantText {
@@ -697,6 +703,7 @@ mod tests {
                 state: SessionStatus::Running,
             },
             Event::UserMessage {
+                id: None,
                 text: "hello".into(),
             },
             Event::Error {
@@ -720,6 +727,7 @@ mod tests {
     fn the_conversation_starts_after_the_last_clear() {
         let (_dir, path) = transcript_with(&[
             Event::UserMessage {
+                id: None,
                 text: "the long expensive conversation".into(),
             },
             Event::AssistantText {
@@ -727,6 +735,7 @@ mod tests {
             },
             Event::Cleared,
             Event::UserMessage {
+                id: None,
                 text: "a fresh start".into(),
             },
             Event::AssistantText {
@@ -744,11 +753,18 @@ mod tests {
     /// first clear dropped.
     fn only_the_newest_clear_counts() {
         let (_dir, path) = transcript_with(&[
-            Event::UserMessage { text: "one".into() },
-            Event::Cleared,
-            Event::UserMessage { text: "two".into() },
+            Event::UserMessage {
+                id: None,
+                text: "one".into(),
+            },
             Event::Cleared,
             Event::UserMessage {
+                id: None,
+                text: "two".into(),
+            },
+            Event::Cleared,
+            Event::UserMessage {
+                id: None,
                 text: "three".into(),
             },
         ]);

server/src/session/mod.rs

@@ -1219,7 +1219,7 @@ async fn pump(
         // message here rather than being carried alongside it. One rule
         // for where a user's message sits: where the session read it.
         let event = match event {
-            Event::MessageTaken { text } => Event::UserMessage { text },
+            Event::MessageTaken { id, text } => Event::UserMessage { id, text },
             other => other,
         };
         // Nothing changed, so there is nothing to record. Both of these
@@ -1464,7 +1464,7 @@ mod tests {
         let user_at = seen
             .iter()
             .position(|entry| {
-                matches!(&entry.event, Event::UserMessage { text } if text == "hello there")
+                matches!(&entry.event, Event::UserMessage { text, .. } if text == "hello there")
             })
             .expect("user message in the stream");
         let echoed: String = seen[user_at..]

server/src/session/transcript.rs

@@ -315,7 +315,10 @@ mod tests {
         let dir = tempfile::tempdir().expect("tempdir");
         let path = dir.path().join("transcript.jsonl");
         let events = vec![
-            Event::UserMessage { text: "hi".into() },
+            Event::UserMessage {
+                id: None,
+                text: "hi".into(),
+            },
             text("hello"),
             Event::ToolStart {
                 id: "t1".into(),