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

@@ -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

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<NotificationKind> {
+///
+/// `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<NotificationKind> {
     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<Commands>,
     notifications: broadcast::Sender<Notification>,
 ) {
+    // 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.
     ///