Say when a session wants you, and stop calling a stop an error

Three things Bryan asked for, and one the second of them exposed.

**Notifications.** A session that asks a question or finishes a turn now
says so on the phone, per session, switchable from its settings screen.

The switch is stored on the backend rather than the phone, because it is a
fact about the session: one that runs unattended overnight should be quiet
on every device, and answering that question again on each device is how two
of them come to disagree. It is on by default -- a notification nobody
wanted is turned off in one tap, where one that never arrived is not
diagnosable at all.

Which moments count is `notification_for`, and the asymmetry in it is the
point. *Waiting on a person* is worth saying however it was reached.
*Finished* is only worth saying when this server watched the work happen:
sessions settle into idle for several reasons that are not "your work
ended", including every one of them being adopted at startup, and announcing
those would put "finished" on the phone for the whole config on every
backend restart. That is the failure that makes somebody switch the feature
off, so it has a test naming every transition rather than the two that
work.

The stream is `GET /notifications`, live only and with no cursor -- the one
place this server does not offer to catch a client up. A notification is a
claim about now; replaying "your turn" from an hour ago sends somebody to a
session that may have been answered from another device since, and a
notification that is wrong costs the trip *and* the credibility of the next
one. What was missed is still on the session list, which says what is
waiting without claiming to be news.

On the phone it is a foreground service, because Android has had no
long-lived background service since 8.0 -- it is what Syncthing does, and
Discord is not a counter-example since it takes a push from Google, which
would mean this backend talking to Google about somebody's sessions. The
ongoing notification Android charges for it sits on an `IMPORTANCE_MIN`
channel: no sound, no status-bar icon, bottom of the shade. `specialUse`
rather than `dataSync`, which is what it looks like: Android 15 caps
dataSync at six hours a day, and a connection that stops listening after six
hours misses the overnight run it exists for.

**A stop is not an error.** The CLI reports an interrupted turn exactly as
it reports a broken one -- `is_error` on a `result` -- so pressing Stop
showed "the turn ended with an error" for doing what the button says. The
line cannot distinguish them; what does is that this side asked, so the
driver says so before the request goes out and the translator spends that on
the next result. The test's second half is the one that matters: the naive
fix passes the first half and silences every genuine failure after it.

**Every status says which one it is.** The session screen's status row named
only `exited` and left the rest blank, so idle and "nobody could read it"
looked identical -- and a just-stopped turn showed nothing, which reads as
the app having lost the session rather than as the stop having worked. The
words are the session list's own, so a state is not called two things
depending which screen you are on. Red on a quota bar now starts at 90%.

**`GET /sessions/{id}`**, which the notification switch found missing. A
screen opened from a list row carries the row the list last fetched: fine
for a title, wrong for a switch, which is *set to* something. Caught on the
emulator, where the switch read on against a backend that said off, with
nothing on screen to say which was true. The screen now reads the session
when it opens, and until that answers the switch is disabled and says so --
a two-position control cannot say "I do not know", so it does not pretend
to.

Verified on the emulator with the app backgrounded: the service holds the
stream (`isForeground=true types=0x40000000`), a finished turn posts
"Finished" and a question replaces it with "Waiting for you" on the same
tag, turning the switch off silences it with no restart, and turning it back
on from the phone reaches config.ron. The interrupt is a translator test
rather than a live turn, which is where that logic is anyway.

server/src/session/claude.rs

@@ -567,6 +567,10 @@ impl Driver for ClaudeDriver {
     /// typed deliberately, and dropping it would lose a message that never
     /// reached the transcript, with nothing on screen to say so.
     fn interrupt(&self) {
+        // Recorded before the request goes out, so the result it produces is
+        // read as the stop somebody asked for rather than as a failure --
+        // see `Translator::interrupting`.
+        self.state.lock().unwrap().expect_interrupt();
         self.send_control(json!({"subtype": "interrupt"}), None);
     }
 

server/src/session/claude/translate.rs

@@ -82,6 +82,20 @@ pub(super) struct Translator {
     /// out is the response: every entry is removed when one arrives,
     /// whether it succeeded or failed.
     asked: HashMap<String, Setting>,
+    /// Whether this side asked the turn to stop.
+    ///
+    /// The CLI reports an interrupted turn the same way it reports one that
+    /// broke -- a `result` with `is_error` set -- so the line itself cannot
+    /// tell them apart, and a person who pressed Stop was shown "the turn
+    /// ended with an error" for doing exactly what the button says. What
+    /// separates them is not in the message at all: it is that *we* asked.
+    /// So the driver says so before the request goes out, the same way it
+    /// does for a setting, and this remembers it until the result lands.
+    ///
+    /// Its path out is that result -- set by `expect_interrupt`, cleared by
+    /// the next `result` whichever way it went, so a genuine failure in a
+    /// later turn is still reported.
+    interrupting: bool,
     session_dir: PathBuf,
 }
 
@@ -91,6 +105,7 @@ impl Translator {
             session_id: None,
             pending: HashMap::new(),
             asked: HashMap::new(),
+            interrupting: false,
             session_dir,
         }
     }
@@ -103,6 +118,13 @@ impl Translator {
     pub(super) fn expect_setting(&mut self, request_id: String, setting: Setting) {
         self.asked.insert(request_id, setting);
     }
+
+    /// Says that the turn about to end was stopped on purpose -- see
+    /// [`Translator::interrupting`]. Called before the request goes out,
+    /// for the reason [`Translator::expect_setting`] gives.
+    pub(super) fn expect_interrupt(&mut self) {
+        self.interrupting = true;
+    }
     pub(super) fn translate(&mut self, message: &Value) -> Vec<Event> {
         // Events from subagents (Task tool internals) carry a
         // parent_tool_use_id; the transcript shows the Task tool's own
@@ -182,10 +204,14 @@ impl Translator {
                         .and_then(Value::as_u64)
                         .unwrap_or(0);
                 let mut events = Vec::new();
-                if message
-                    .get("is_error")
-                    .and_then(Value::as_bool)
-                    .unwrap_or(false)
+                // Whichever way this result went, the interrupt it may have
+                // been answering is now spent.
+                let asked_to_stop = std::mem::take(&mut self.interrupting);
+                if !asked_to_stop
+                    && message
+                        .get("is_error")
+                        .and_then(Value::as_bool)
+                        .unwrap_or(false)
                 {
                     events.push(Event::Error {
                         message: message
@@ -1138,6 +1164,48 @@ mod tests {
         );
     }
 
+    /// Pressing Stop is not a failure, and the CLI cannot tell you which it
+    /// was.
+    ///
+    /// An interrupted turn arrives as exactly the same shape a broken one
+    /// does -- `is_error` set, on a `result` -- so somebody who pressed the
+    /// button was shown "the turn ended with an error" for doing what the
+    /// button says. What separates the two is not in the line: it is that
+    /// this side asked. The second half of this test is the one that
+    /// matters, because the naive fix -- never reporting an error result --
+    /// passes the first half and silences every genuine failure afterwards.
+    #[test]
+    fn a_turn_stopped_on_purpose_is_not_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let mut translator = Translator::new(dir.path().to_path_buf());
+        let stopped_result = r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Interrupted by user","usage":{}}"#;
+
+        translator.expect_interrupt();
+        let events = translate_lines(&mut translator, &[stopped_result]);
+        assert!(
+            !events
+                .iter()
+                .any(|event| matches!(event, Event::Error { .. })),
+            "a stop the driver asked for was reported as a failure: {events:?}"
+        );
+        assert_eq!(
+            *events.last().unwrap(),
+            Event::Status {
+                state: SessionStatus::Idle
+            },
+            "an interrupted turn still has to end the turn"
+        );
+
+        // The interrupt is spent, so the next failure is a failure again.
+        let later = translate_lines(&mut translator, &[stopped_result]);
+        assert!(
+            later
+                .iter()
+                .any(|event| matches!(event, Event::Error { .. })),
+            "a later failure was swallowed by an interrupt that had already been answered"
+        );
+    }
+
     #[test]
     fn replayed_and_synthetic_user_text_is_skipped() {
         let dir = tempfile::tempdir().expect("tempdir");

server/src/session/mod.rs

@@ -42,6 +42,14 @@ use transport::Transport;
 /// the size only bounds memory, not correctness.
 const EVENT_BUFFER: usize = 256;
 
+/// Fan-out buffer for notifications, across every session.
+///
+/// Small, and deliberately: a subscriber that falls this far behind on a
+/// stream carrying two events per turn is not one whose backlog is worth
+/// delivering. Lagging drops the oldest, which is the right end to lose --
+/// the newest "your turn" is the one still true.
+const NOTIFICATION_BUFFER: usize = 64;
+
 pub fn now() -> f64 {
     SystemTime::now()
         .duration_since(UNIX_EPOCH)
@@ -62,6 +70,37 @@ pub struct SpawnSpec {
     pub params: std::collections::BTreeMap<String, String>,
 }
 
+/// A moment worth interrupting somebody for, as `GET /notifications`
+/// sends it.
+///
+/// Two kinds, and the pair is the whole feature: a session that has *asked*
+/// something cannot continue until it is answered, and one that has
+/// *finished* is work somebody walked away from. Everything else a session
+/// does is progress they did not ask to be told about.
+///
+/// Carries the title rather than only the id, so the phone can write the
+/// notification without a round trip -- it may well be showing no screen at
+/// all when this arrives.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Notification {
+    pub session_id: String,
+    pub title: String,
+    pub kind: NotificationKind,
+    /// Epoch seconds, so a phone that was asleep can say how long ago.
+    pub at: f64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub enum NotificationKind {
+    /// The session is waiting on a person: a question, or a permission.
+    AwaitingInput,
+    /// A turn ended without one. Only ever sent for a session that was
+    /// *seen* running -- see `notification_for`.
+    Finished,
+}
+
 /// One row of `GET /sessions`.
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -104,6 +143,10 @@ pub struct SessionInfo {
     pub imported: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub cwd: Option<PathBuf>,
+    /// Whether this session announces itself -- reported for the same
+    /// reason `permission_mode` is: a switch that guesses its own position
+    /// is how you turn something off while believing you are reading it.
+    pub notify: bool,
     pub status: SessionStatus,
     pub last_activity: f64,
     pub created: f64,
@@ -210,6 +253,14 @@ struct Shared {
     /// session was *launched* with, so reporting from it would show the
     /// mode a change had already replaced.
     permission_mode: Mutex<Option<String>>,
+    /// Whether this session's attention-wanting moments are announced.
+    ///
+    /// Mirrored out of the config so the pump can read it without taking
+    /// the manager's lock -- the pump runs underneath the manager and
+    /// reaching back up for a field would invert that. `set_session_notify`
+    /// writes both, in that order, which is the same shape every other
+    /// live-and-persisted setting here uses.
+    notify: Mutex<bool>,
     /// How many events this session has ever recorded.
     ///
     /// Only the import sync reads it, and only to answer one question:
@@ -319,6 +370,7 @@ impl LiveSession {
             title: self.shared.title.lock().unwrap().clone(),
             model: self.shared.model.lock().unwrap().clone(),
             permission_mode: self.shared.permission_mode.lock().unwrap().clone(),
+            notify: *self.shared.notify.lock().unwrap(),
             imported,
             keeps_own_transcript,
             cwd: self.meta.cwd.clone(),
@@ -343,6 +395,10 @@ pub struct SessionManager {
     /// which is why they live beside the session directories rather than
     /// inside one.
     models_dir: PathBuf,
+    /// Where every session's pump sends what a phone should be told about.
+    /// Held here rather than per session for the reason
+    /// [`SessionManager::subscribe_notifications`] gives.
+    notifications: broadcast::Sender<Notification>,
     inner: RwLock<Inner>,
 }
 
@@ -356,6 +412,7 @@ impl SessionManager {
         let config = Config::load(&config_path)?;
         wg_app_link::private::create_dir(&data_dir)?;
 
+        let (notifications, _) = broadcast::channel(NOTIFICATION_BUFFER);
         let mut live = HashMap::new();
         for meta in &config.sessions {
             // One unlaunchable session -- a corrupt transcript, an
@@ -370,6 +427,7 @@ impl SessionManager {
                     &data_dir,
                     &models_dir,
                     None,
+                    notifications.clone(),
                 )
             }) {
                 Ok(session) => {
@@ -384,6 +442,7 @@ impl SessionManager {
             config_path,
             data_dir,
             models_dir,
+            notifications,
             inner: RwLock::new(Inner { config, live }),
         };
         Ok(manager)
@@ -647,6 +706,7 @@ impl SessionManager {
                     title: meta.title.clone(),
                     model: meta.model.clone(),
                     permission_mode: meta.permission_mode.clone(),
+                    notify: meta.notify,
                     imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
                     keeps_own_transcript: keeps_own_transcript(
                         &inner.config,
@@ -662,6 +722,16 @@ impl SessionManager {
             .collect()
     }
 
+    /// Every session's attention-wanting moments, on one stream.
+    ///
+    /// One connection for the whole backend rather than one per session:
+    /// the phone subscribes to this while showing no session at all, and a
+    /// connection per session would mean opening one for every session that
+    /// exists in order to hear about any of them.
+    pub fn subscribe_notifications(&self) -> broadcast::Receiver<Notification> {
+        self.notifications.subscribe()
+    }
+
     pub fn session(&self, id: &str) -> Option<Arc<LiveSession>> {
         self.inner.read().unwrap().live.get(id).cloned()
     }
@@ -739,6 +809,11 @@ impl SessionManager {
             cwd: spec.cwd,
             permission_mode: spec.permission_mode,
             params: spec.params,
+            // On by default -- see `SessionConfig::notify`. Not offered at
+            // spawn: a session's first turn is exactly the one somebody is
+            // waiting for, and a switch on the spawn screen would be a
+            // decision asked before there is anything to decide about.
+            notify: true,
             created: now(),
         };
 
@@ -749,6 +824,7 @@ impl SessionManager {
             &self.data_dir,
             &self.models_dir,
             seed,
+            self.notifications.clone(),
         )?;
         let mut candidate = inner.config.clone();
         candidate.sessions.push(meta);
@@ -804,6 +880,33 @@ impl SessionManager {
         Ok(())
     }
 
+    /// Turns this session's notifications on or off, live and persisted.
+    ///
+    /// Both, in that order, for the reason every setting here writes both:
+    /// the config decides what a restart believes and the live copy decides
+    /// what the running pump does, and a change that lands in one of them is
+    /// a switch that moves back on its own.
+    ///
+    /// Nothing is told to the driver. Unlike the model or the permission
+    /// mode, this changes nothing about how the session runs -- it is about
+    /// who gets told, and the session is not the one being told.
+    pub fn set_session_notify(&self, id: &str, notify: bool) -> Result<()> {
+        let mut inner = self.inner.write().unwrap();
+        if !inner.config.sessions.iter().any(|meta| meta.id == id) {
+            bail!("no session {id}");
+        }
+        let mut candidate = inner.config.clone();
+        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
+            meta.notify = notify;
+        }
+        candidate.save(&self.config_path)?;
+        inner.config = candidate;
+        if let Some(session) = inner.live.get(id) {
+            *session.shared.notify.lock().unwrap() = notify;
+        }
+        Ok(())
+    }
+
     /// Renames a session: persisted, shown, and passed on to whatever is
     /// running it.
     ///
@@ -1084,6 +1187,7 @@ fn launch(
     data_dir: &Path,
     models_dir: &Path,
     seed: Option<Seed>,
+    notifications: broadcast::Sender<Notification>,
 ) -> Result<Arc<LiveSession>> {
     let dir = data_dir.join(&meta.id);
     wg_app_link::private::create_dir(&dir)?;
@@ -1112,6 +1216,7 @@ fn launch(
         last_activity: Mutex::new(now()),
         model: Mutex::new(meta.model.clone()),
         permission_mode: Mutex::new(meta.permission_mode.clone()),
+        notify: Mutex::new(meta.notify),
         written: Mutex::new(0),
     });
 
@@ -1156,11 +1261,13 @@ fn launch(
     });
 
     tokio::spawn(pump(
+        meta.id.clone(),
         transcript,
         source,
         Arc::clone(&shared),
         events.clone(),
         Arc::clone(&commands),
+        notifications,
     ));
 
     Ok(Arc::new(LiveSession {
@@ -1205,12 +1312,33 @@ fn is_news(event: &Event, shared: &Shared) -> bool {
     }
 }
 
+/// Whether moving from `was` to `now` is worth interrupting somebody for.
+///
+/// The asymmetry is the point. *Waiting on a person* is worth saying however
+/// the session got there -- it is a question that will sit unanswered until
+/// somebody sees it. *Finished* is only worth saying when this server
+/// watched the work happen: a session settling into idle because it was
+/// 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> {
+    match (was, now) {
+        (_, SessionStatus::AwaitingInput) => Some(NotificationKind::AwaitingInput),
+        (SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle) => {
+            Some(NotificationKind::Finished)
+        }
+        _ => None,
+    }
+}
+
 async fn pump(
+    id: String,
     mut transcript: Transcript,
     mut source: mpsc::UnboundedReceiver<Event>,
     shared: Arc<Shared>,
     events: broadcast::Sender<SeqEvent>,
     commands: Arc<Commands>,
+    notifications: broadcast::Sender<Notification>,
 ) {
     while let Some(event) = source.recv().await {
         let ts = now();
@@ -1251,7 +1379,21 @@ async fn pump(
         match transcript.append(event, ts) {
             Ok(entry) => {
                 if let Event::Status { state } = &entry.event {
-                    *shared.status.lock().unwrap() = *state;
+                    // 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())
+                    {
+                        // No subscribers is the ordinary case -- nobody has
+                        // the app open -- and it is not an error.
+                        let _ = notifications.send(Notification {
+                            session_id: id.clone(),
+                            title: shared.title.lock().unwrap().clone(),
+                            kind,
+                            at: ts,
+                        });
+                    }
                 }
                 *shared.last_activity.lock().unwrap() = ts;
                 *shared.written.lock().unwrap() += 1;
@@ -1366,6 +1508,101 @@ mod tests {
         assert!(!DriverKind::LlamaCpp.keeps_own_transcript());
     }
 
+    /// The two transitions worth interrupting somebody for, and the ones
+    /// that look like them and are not.
+    ///
+    /// The idle cases are the whole reason this is a function rather than a
+    /// pair of `if`s at the callsite. A session settles into idle for
+    /// several reasons that are not "your work finished": it was adopted at
+    /// startup, its driver announced itself, it came back from a state
+    /// nobody could read. Announcing those would put "finished" on the phone
+    /// for every session in the config every time the backend restarts,
+    /// which is the failure that makes somebody turn the whole feature off.
+    #[test]
+    fn only_a_watched_turn_ending_counts_as_finished() {
+        use NotificationKind::{AwaitingInput, Finished};
+        use SessionStatus::{Compacting, Exited, Idle, Running, Unknown};
+
+        // 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),
+            Some(AwaitingInput)
+        );
+        assert_eq!(
+            notification_for(Idle, SessionStatus::AwaitingInput),
+            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));
+
+        // 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);
+
+        // 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);
+    }
+
+    /// The switch reaches the running pump, not just the config file.
+    ///
+    /// The failure this exists for is silent in the direction that matters:
+    /// a `set_session_notify(false)` that wrote only the config would look
+    /// correct on the settings screen and in the file, and keep notifying
+    /// until the backend was restarted. Nothing on screen would say so, and
+    /// the person who turned it off is by definition not watching.
+    #[tokio::test]
+    async fn turning_notifications_off_stops_them_without_a_restart() {
+        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 notifications = manager.subscribe_notifications();
+        session.send_message("hello".to_string(), Vec::new());
+        let first = tokio::time::timeout(Duration::from_secs(5), notifications.recv())
+            .await
+            .expect("a notification within five seconds")
+            .expect("channel open");
+        assert_eq!(first.kind, NotificationKind::Finished);
+        assert_eq!(first.session_id, info.id);
+        // The title travels with it, because the phone may have no screen
+        // open to look one up on.
+        assert_eq!(first.title, session.info("m", false, false).title);
+
+        manager.set_session_notify(&info.id, false).expect("off");
+        // Subscribed before the message, or the turn can finish in the gap
+        // and leave this waiting for an event that has already gone past.
+        let mut events = session.subscribe();
+        session.send_message("hello again".to_string(), Vec::new());
+        // The turn still happens -- this is a switch about being told, not
+        // about running -- so wait for the turn's own event and then check
+        // that nothing was announced alongside it.
+        collect_until(&mut events, |event| {
+            matches!(
+                event,
+                Event::Status {
+                    state: SessionStatus::Idle
+                }
+            )
+        })
+        .await;
+        assert!(
+            notifications.try_recv().is_err(),
+            "a session with notifications off still announced itself"
+        );
+    }
+
     /// A session this app *spawned* is one it is driving, and used to look
     /// like somebody else's.
     ///