Let the server say what is waiting, instead of the phone remembering

A message sent into a running turn was drawn as a pending bubble from
screen state, so leaving the session or restarting the app showed nothing
waiting while the queue was full. Nothing waiting is what "there is
nothing" looks like -- the reader had no way to tell it from a queue that
had already drained, and Bryan hit exactly that: a message he sent
arrived, and his phone stopped showing it after a restart.

The server now records the waiting. `MessageQueued { id, text }` goes into
the transcript when a driver takes a message it cannot deliver yet, and
is resolved by the `UserMessage` carrying the same id -- the same shape
`CommandQueued` and `CommandSent` already had, so this is one more
instance of a mechanism rather than a second one beside it.

The message itself still lands where the session read it, which is what
the last change was about; only the *waiting* is recorded early. The two
are different facts and now have different events.

Paired by id rather than by text. The old code removed the bubble whose
text matched, so sending the same thing twice cleared the wrong one and
left a message on screen that had already been read.

Both drivers that can queue do it: the echo driver too, because the phone
now draws pending bubbles from the stream and a rig that skipped the
event would exercise a state the real app never sees.

Checked on the emulator: two messages sent into a `/slow` turn, then the
app force-stopped and relaunched -- both still drawn as waiting, in the
pending style, and both resolved into ordinary bubbles when the turn
ended and the session read them.

Still outstanding, and worth knowing: an entry outlives a *server*
restart in the transcript but not in the driver's memory, so a backend
restarted mid-queue would leave the bubble drawn with nothing coming to
resolve it. Before this change that message vanished from the transcript
entirely, so the failure is now visible rather than silent -- but it is
not yet right.

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(),