From a1eedd7a78b5c37758617ec6d7621cbe83728a13 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Mon, 31 Aug 2026 22:43:24 -0400 Subject: [PATCH] Don't call a turn finished with a message still waiting behind it A message written into the tail of a turn is read the moment that turn's `result` lands: the session reports idle and is running again in the same breath. The phone that sent it got "finished" in between -- seconds before anything it asked for had been done, which is the notification arriving to say the opposite of what is happening. `notification_for` now takes how many messages the session has been given and not started reading, and a turn ending with any of them waiting is not an ending. The count is kept in `pump`, from the recorded events, because that is the one place that sees all of them in transcript order: a `messageQueued` up, and the `userMessage` that resolves it or a `messageDropped` down. Asking the driver instead would answer about the moment the question was asked rather than the moment the status was written, which is the same class of mistake as reading a session's status to decide what a queue contains. It deliberately does not suppress *awaiting input*. A question is worth interrupting somebody for whatever is queued behind it -- the queue is precisely what will not move until it is answered. Tested both halves: the decision on the number, and the number itself, where an echo turn that reads its queued message before going idle still announces its finish. That last is the case a suppression written slightly wrong silences, and it is the common one. --- PLAN.md | 13 +++++ server/src/session/mod.rs | 120 +++++++++++++++++++++++++++++++++----- 2 files changed, 118 insertions(+), 15 deletions(-) diff --git a/PLAN.md b/PLAN.md index 4dd4edf..7e37ceb 100644 --- a/PLAN.md +++ b/PLAN.md @@ -925,6 +925,19 @@ level: the session on screen is registered by the one composable that draws one, and "the app is up" *is* the banner queue being collected, since it collects only while it is on screen. +**What counts as finished** is decided in `notification_for`, and since +2026-08-31 it takes the number of messages the session has been given and +not started reading. With one waiting, a turn ending is not the work +ending: a message written into the tail of a turn is read the moment that +turn's `result` lands, so the session goes idle and immediately runs again +-- and the phone that sent it was told its work had finished, seconds +before any of it was done. The count is kept in `pump` from the recorded +events (`MessageQueued` up, the `UserMessage` that resolves it or a +`MessageDropped` down), because that is the one place that sees every event +in transcript order. It does not suppress *awaiting input*: a question is +worth saying whatever is queued behind it, and the queue is exactly what +will not move until it is answered. + The alternative considered and rejected was giving the app its own connection to `/notifications` while it is in front. That is a second stream per device saying the same thing, and it puts the "which of these two shows diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index cda3e89..c34855c 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -2086,10 +2086,26 @@ fn is_news(event: &Event, shared: &Shared) -> bool { /// adopted at startup, or because a driver announced itself, is not news /// that anything ended, and sending it would put "finished" on the phone for /// every session in the config every time the backend restarts. -fn notification_for(was: SessionStatus, now: SessionStatus) -> Option { +/// +/// `unread` is how many messages the session has been handed and not yet +/// started reading, and it suppresses *Finished* for the same reason: with +/// one waiting, the turn ending is not the work ending. A message written +/// into the tail of a turn is read as soon as that turn's `result` lands, so +/// the session goes idle and immediately runs again -- and the phone that +/// sent it was told its work had finished, seconds before anything of it had +/// been done. It cannot suppress *AwaitingInput*: a question is worth saying +/// whatever else is queued behind it, and the queue is precisely what will +/// not move until it is answered. +fn notification_for( + was: SessionStatus, + now: SessionStatus, + unread: usize, +) -> Option { match (was, now) { (_, SessionStatus::AwaitingInput) => Some(NotificationKind::AwaitingInput), - (SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle) => { + (SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle) + if unread == 0 => + { Some(NotificationKind::Finished) } _ => None, @@ -2105,6 +2121,13 @@ async fn pump( commands: Arc, notifications: broadcast::Sender, ) { + // Messages the session has been given and not started reading, which is + // what makes a turn ending not the same thing as the work ending; see + // `notification_for`. Counted from the recorded events rather than asked + // of the driver, because this is the one place that sees every event in + // the order the transcript has them -- and because the answer has to + // survive being asked a moment later than the driver would have said it. + let mut unread: usize = 0; while let Some(event) = source.recv().await { let ts = now(); // Taking a message is how it enters the conversation, and the @@ -2154,8 +2177,8 @@ async fn pump( // Read before it is overwritten: what makes a status // worth announcing is the transition, not the value. let was = std::mem::replace(&mut *shared.status.lock().unwrap(), *state); - if let Some(kind) = - notification_for(was, *state).filter(|_| *shared.notify.lock().unwrap()) + if let Some(kind) = notification_for(was, *state, unread) + .filter(|_| *shared.notify.lock().unwrap()) { // No subscribers is the ordinary case -- nobody has // the app open -- and it is not an error. @@ -2180,6 +2203,13 @@ async fn pump( Event::Status { state: SessionStatus::Exited, } => commands.abandon("this session's process has exited"), + // The two ends of a message's wait. A `UserMessage` with + // no id never waited -- it is one sent between turns, and + // counting it would take the total below zero. + Event::MessageQueued { .. } => unread += 1, + Event::UserMessage { id: Some(_), .. } | Event::MessageDropped { .. } => { + unread = unread.saturating_sub(1) + } _ => {} } // No subscribers is fine; the transcript already has it. @@ -2400,28 +2430,43 @@ mod tests { // Waiting on a person is worth saying however it was reached: it // will sit unanswered until somebody is told. assert_eq!( - notification_for(Running, SessionStatus::AwaitingInput), + notification_for(Running, SessionStatus::AwaitingInput, 0), Some(AwaitingInput) ); assert_eq!( - notification_for(Idle, SessionStatus::AwaitingInput), + notification_for(Idle, SessionStatus::AwaitingInput, 0), Some(AwaitingInput) ); // A turn this server watched run, ending. - assert_eq!(notification_for(Running, Idle), Some(Finished)); - assert_eq!(notification_for(Compacting, Idle), Some(Finished)); + assert_eq!(notification_for(Running, Idle, 0), Some(Finished)); + assert_eq!(notification_for(Compacting, Idle, 0), Some(Finished)); // Idle arrived at from anywhere else is not an ending. - assert_eq!(notification_for(Idle, Idle), None); - assert_eq!(notification_for(Unknown, Idle), None); - assert_eq!(notification_for(Exited, Idle), None); - assert_eq!(notification_for(SessionStatus::AwaitingInput, Idle), None); + assert_eq!(notification_for(Idle, Idle, 0), None); + assert_eq!(notification_for(Unknown, Idle, 0), None); + assert_eq!(notification_for(Exited, Idle, 0), None); + assert_eq!( + notification_for(SessionStatus::AwaitingInput, Idle, 0), + None + ); // Everything else a session does is progress nobody asked to hear. - assert_eq!(notification_for(Idle, Running), None); - assert_eq!(notification_for(Running, Compacting), None); - assert_eq!(notification_for(Running, Exited), None); + assert_eq!(notification_for(Idle, Running, 0), None); + assert_eq!(notification_for(Running, Compacting, 0), None); + assert_eq!(notification_for(Running, Exited, 0), None); + + // A turn ending with a message the session has not started reading + // is not the work ending: it goes straight back to running, and + // "finished" would arrive seconds before any of that work was done. + assert_eq!(notification_for(Running, Idle, 1), None); + assert_eq!(notification_for(Compacting, Idle, 2), None); + // A question is still worth saying with a queue behind it -- the + // queue is exactly what will not move until it is answered. + assert_eq!( + notification_for(Running, SessionStatus::AwaitingInput, 1), + Some(AwaitingInput) + ); } /// The switch reaches the running pump, not just the config file. @@ -2477,6 +2522,51 @@ mod tests { ); } + /// Counting the wait, rather than only deciding what to do about it. + /// + /// `notification_for` is tested above on the number; this is the number + /// itself, which is kept in `pump` from the recorded events and has no + /// other way to be looked at. Echo takes its queued message *before* + /// going idle -- the same order a real CLI has when the steer lands + /// inside the turn -- so the count is back to zero by the end and the + /// finish is still announced. That is the case a suppression written + /// slightly wrong silences, and it is the common one. + #[tokio::test] + async fn a_turn_that_read_its_queued_message_still_announces_its_finish() { + let dir = tempfile::tempdir().expect("tempdir"); + let config_path = dir.path().join("config.ron"); + let data_dir = dir.path().join("sessions"); + seed_echo_only(&config_path); + let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) + .expect("manager"); + let info = manager.spawn_session(echo_spec()).expect("spawn"); + let session = manager.session(&info.id).expect("live"); + let mut events = session.subscribe(); + let mut notifications = manager.subscribe_notifications(); + + session.send_message("/slow 1".to_string(), Vec::new()); + collect_until(&mut events, |event| { + matches!( + event, + Event::Status { + state: SessionStatus::Running + } + ) + }) + .await; + session.send_message("and this behind it".to_string(), Vec::new()); + collect_until(&mut events, |event| { + matches!(event, Event::MessageQueued { .. }) + }) + .await; + + let announced = tokio::time::timeout(Duration::from_secs(5), notifications.recv()) + .await + .expect("a notification within five seconds") + .expect("channel open"); + assert_eq!(announced.kind, NotificationKind::Finished); + } + /// A session this app *spawned* is one it is driving, and used to look /// like somebody else's. ///