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