From e37e90a579734af634dbd3673d81f43128f6ca6d Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Sat, 29 Aug 2026 22:29:12 -0400 Subject: [PATCH 1/2] 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. --- .../main/kotlin/com/example/aiapp/Events.kt | 31 +++++++++++- .../kotlin/com/example/aiapp/SessionScreen.kt | 37 ++++++++------ server/src/session/claude.rs | 50 ++++++++++++++----- server/src/session/driver.rs | 31 ++++++++++++ server/src/session/echo.rs | 37 ++++++++++---- server/src/session/import.rs | 4 +- server/src/session/llama.rs | 26 ++++++++-- server/src/session/mod.rs | 4 +- server/src/session/transcript.rs | 5 +- 9 files changed, 177 insertions(+), 48 deletions(-) diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt index 158dcce..11a6df7 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/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( 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 5933704..0f5bda4 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -270,6 +270,9 @@ fun foldEvent(items: List, entry: SeqEvent): List 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) @@ -375,7 +378,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()) } + // 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>()) } // 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. @@ -428,7 +435,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) { @@ -691,18 +705,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, @@ -816,7 +823,7 @@ fun SessionScreen( waitingCommands.forEach { (_, text) -> CommandBubble(text, waiting = true) } - queued.forEach { text -> UserBubble(text, pending = true) } + queued.forEach { (_, text) -> UserBubble(text, pending = true) } } } } diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index 39b7482..48959e0 100644 --- a/server/src/session/claude.rs +++ b/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, + /// 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 = self.awaiting.drain(..).collect(); + let lost: Vec = 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>, sink: &EventSink) -> bool { - let taken: Vec = { + 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 diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 23e2dd9..553088a 100644 --- a/server/src/session/driver.rs +++ b/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, + 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, text: String, }, /// Streaming assistant text; the phone renders the concatenation as diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index f53e69c..5e7b5a1 100644 --- a/server/src/session/echo.rs +++ b/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, - queued: Arc>>, + /// Held messages with the id of the `MessageQueued` each one announced, + /// so the announcement can say which waiting bubble it resolves. + queued: Arc>>, /// 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>, busy: &AtomicBool) { +fn finish_turn(sink: &EventSink, queued: &Mutex>, 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}"), }); diff --git a/server/src/session/import.rs b/server/src/session/import.rs index b6ebc6f..64194c3 100644 --- a/server/src/session/import.rs +++ b/server/src/session/import.rs @@ -538,7 +538,9 @@ fn push_user(events: &mut Vec, 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 }); } } diff --git a/server/src/session/llama.rs b/server/src/session/llama.rs index b0e5c2b..0378727 100644 --- a/server/src/session/llama.rs +++ b/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 { }; 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(), }, ]); diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index e72ec69..2d19a02 100644 --- a/server/src/session/mod.rs +++ b/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..] diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index a1c0091..ddc7cd7 100644 --- a/server/src/session/transcript.rs +++ b/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(), From 1635fe97c81a1345510b25ca5c48d796ea67fb01 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Sat, 29 Aug 2026 22:33:46 -0400 Subject: [PATCH 2/2] Take the formatter's line wrapping in NerdIcons Left over from running ktfmt across the merge: the comment reflows two lines. No change to what it says. --- .../src/main/kotlin/com/example/aiapp/NerdIcons.kt | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/NerdIcons.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/NerdIcons.kt index 3c21f2d..6ac02fb 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/NerdIcons.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/NerdIcons.kt @@ -34,9 +34,9 @@ import androidx.compose.ui.unit.sp * The same arrangement as dev-updater, down to the cog and the refresh arrow being the same two * Material Design codepoints. Those two must not drift: an icon that means "settings" in one app * and something else in the other is the failure this is worth preventing. The script is copied - * rather than shared because most of what looks like duplication is the `GLYPHS` list, which has - * to differ -- the point of subsetting is to ship only the codepoints one app draws. All Material Design - * bar one, so they read as one family; the exception is noted where it is declared. + * rather than shared because most of what looks like duplication is the `GLYPHS` list, which has to + * differ -- the point of subsetting is to ship only the codepoints one app draws. All Material + * Design bar one, so they read as one family; the exception is noted where it is declared. */ val NerdIcons = FontFamily(Font(R.font.nerd_icons))