watch: scheduled results from the server's /events stream, told by the task's notify policy

The app follows GET /api/v2/events instead of asking /schedules every 15 s; each
`schedules` frame goes to the page (the Scheduled page redraws from it) and a
schedule's newly finished last run is told by the server's decision: `notified`
when it already told a device, else its `notify` policy (matters: a result, not
a silent one; always: silent too; never: nothing), and a failure the first time
in a row and on the third, which asks whether to pause. A run is told once, also
when it finished under the id it had while running. Signing out or another
server ends the stream; a server without it is asked again every 5 minutes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src-tauri/src/lib.rs

@@ -1368,8 +1368,7 @@ fn on_agent_state(handle: &AppHandle, state: &AgentState) {
     announce(handle);
 }
 
-/// Look at the chat list every so often, and say when a reply finished out of sight; and at the
+/// Look at the chat list every so often, and say when a reply finished out of sight.
-/// scheduled tasks, and say when one has a result the server tells of.
 fn watch_chats(handle: AppHandle) {
     std::thread::spawn(move || loop {
         std::thread::sleep(watch::EVERY);
@@ -1387,11 +1386,7 @@ fn watch_chats(handle: AppHandle) {
             }
         };
         let done = app.watch.look(&list);
-        // Scheduled tasks: a server without the routes (404) has none to tell of.
+        if done.is_empty() || !app.settings().notify {
-        let ran = server::request(&app_url, &token, "GET", "/schedules", None)
-            .map(|v| app.watch.look_schedules(&v))
-            .unwrap_or_default();
-        if (done.is_empty() && ran.is_empty()) || !app.settings().notify {
             continue;
         }
         let in_front = main_window(&handle)
@@ -1414,23 +1409,68 @@ fn watch_chats(handle: AppHandle) {
                 .body(if d.agent { &words.agent } else { &words.reply })
                 .show();
         }
-        // A scheduled result is said even with the window in front: it ran on its own.
+    });
-        for r in ran {
+}
-            let title = if r.title.is_empty() {
+
-                "LLeMbas".to_string()
+/// How long to wait before following the event stream again: after it closed or broke off, and
-            } else {
+/// (longer) on a server that has none.
-                r.title
+const EVENTS_AGAIN: std::time::Duration = std::time::Duration::from_secs(5);
+const EVENTS_NONE: std::time::Duration = std::time::Duration::from_secs(300);
+
+/// Follow the server's event stream for the signed-in person (`GET /events`): each `schedules`
+/// frame goes to the page as it is (the Scheduled page redraws from it), and a scheduled task's
+/// newly finished run is told by its notify decision (`watch::decide`). Said even with the window
+/// in front: it ran on its own. Signing out or another server ends the stream (`Watch::reset`).
+fn watch_events(handle: AppHandle) {
+    std::thread::spawn(move || loop {
+        let app = handle.state::<App>();
+        let Some((app_url, token)) = app.door() else {
+            std::thread::sleep(EVENTS_AGAIN);
+            continue;
+        };
+        let stop = app.watch.stop_flag();
+        let result = server::stream(&app_url, &token, "/events", &stop, |event, data| {
+            if event != "schedules" {
+                return true;
+            }
+            let Ok(list) = serde_json::from_str::<Value>(data) else {
+                return true;
             };
-            let _ = handle
+            let app = handle.state::<App>();
-                .notification()
+            let ran = app.watch.look_schedules(&list);
-                .builder()
+            let _ = handle.emit("schedules", &list);
-                .title(title)
+            if ran.is_empty() || !app.settings().notify {
-                .body(if r.failed {
+                return true;
-                    &words.scheduled_failed
+            }
+            let words = app.watch.words.lock().unwrap().clone();
+            for r in ran {
+                let title = if r.title.is_empty() {
+                    "LLeMbas".to_string()
                 } else {
-                    &words.scheduled
+                    r.title
-                })
+                };
-                .show();
+                let body = match r.told {
+                    watch::Told::Result => &words.scheduled,
+                    watch::Told::Failed => &words.scheduled_failed,
+                    watch::Told::Pause => &words.scheduled_pause,
+                };
+                let _ = handle
+                    .notification()
+                    .builder()
+                    .title(title)
+                    .body(body)
+                    .show();
+            }
+            true
+        });
+        match result {
+            Err(e) if e.status == 401 => {
+                token_refused(&handle);
+                std::thread::sleep(EVENTS_AGAIN);
+            }
+            // A server without the stream: look again now and then (it may be updated).
+            Err(e) if e.status == 404 || e.status == 405 => std::thread::sleep(EVENTS_NONE),
+            _ => std::thread::sleep(EVENTS_AGAIN),
         }
     });
 }
@@ -1594,6 +1634,7 @@ pub fn run() {
             build_main(&handle)?;
             watch_updates(handle.clone());
             watch_chats(handle.clone());
+            watch_events(handle.clone());
             {
                 let h = handle.clone();
                 std::thread::spawn(move || {

src-tauri/src/watch.rs

@@ -8,13 +8,19 @@
 //! the page's language; the chat's title is the person's own. Nothing from the reply itself is
 //! put on screen outside the app.
 //!
-//! Scheduled tasks: the app asks `/api/v2/schedules` as often, and a schedule whose last run is a
+//! Scheduled tasks: the app follows the server's own event stream (`GET /api/v2/events`), whose
-//! new one, with the server's word that it is worth telling (`notified`: its notify policy, and a
+//! `schedules` frame carries the whole list (`{"version", "schedules": [schedule]}`) whenever it
-//! `[SILENT]` run not), is said once -- by the schedule's title, never the result's words. A
+//! changes -- a run starting, waiting, finishing. A schedule's last run that has newly finished is
-//! server without the routes answers 404, and nothing is said.
+//! told by the server's notify decision for it: `notified` when the server already told a device,
+//! else the schedule's `notify` policy as the server applies it (`services/schedule/notify.py`:
+//! `always` -- a result, even "nothing to report"; `matters` -- a result, never a `[SILENT]` one;
+//! `never` -- nothing), and a failure the first time in a row and again on the third, which asks
+//! whether to pause it. Said by the schedule's title, never the result's words. A server without
+//! the stream answers 404, and nothing is said.
 
 use std::collections::HashMap;
-use std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use serde::Deserialize;
@@ -35,6 +41,9 @@ pub struct Words {
     /// A scheduled task's run failed.
     #[serde(default = "scheduled_failed_word")]
     pub scheduled_failed: String,
+    /// A scheduled task failed three times in a row: pause it?
+    #[serde(default = "scheduled_pause_word")]
+    pub scheduled_pause: String,
 }
 
 fn scheduled_word() -> String {
@@ -45,6 +54,10 @@ fn scheduled_failed_word() -> String {
     "A scheduled task failed.".into()
 }
 
+fn scheduled_pause_word() -> String {
+    "A scheduled task failed 3 times in a row. Pause it?".into()
+}
+
 impl Default for Words {
     fn default() -> Self {
         Words {
@@ -52,6 +65,7 @@ impl Default for Words {
             agent: "The agent has finished.".into(),
             scheduled: scheduled_word(),
             scheduled_failed: scheduled_failed_word(),
+            scheduled_pause: scheduled_pause_word(),
         }
     }
 }
@@ -71,15 +85,56 @@ pub struct Watch {
     pub words: Mutex<Words>,
     /// Which chats were being answered at the last look (None before the first look).
     running: Mutex<Option<HashMap<String, bool>>>,
-    /// Each schedule's last run at the last look (None before the first look).
+    /// Each schedule's last run at the last look, and whether it had finished (None before the
-    runs: Mutex<Option<HashMap<String, String>>>,
+    /// first look).
+    runs: Mutex<Option<HashMap<String, (String, bool)>>>,
+    /// Set to end the event stream that is followed now; `reset` sets it and puts a new one here.
+    stop: Mutex<Arc<AtomicBool>>,
 }
 
-/// A scheduled task's new run that the server says is worth telling of.
+/// How a scheduled task's run is told.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum Told {
+    /// It has a result.
+    Result,
+    /// It failed.
+    Failed,
+    /// It failed for the third time in a row: pause it?
+    Pause,
+}
+
+/// A scheduled task's newly finished run that is worth telling of.
 #[derive(Clone, Debug, PartialEq)]
 pub struct Ran {
     pub title: String,
-    pub failed: bool,
+    pub told: Told,
+}
+
+/// The run statuses that are an end (`db/models/schedule.py`): `running` and `waiting` are not.
+const FINISHED: [&str; 5] = ["ok", "silent", "failed", "skipped", "missed"];
+/// The failure in a row that asks whether to pause (`notify.PAUSE_AFTER`).
+const PAUSE_AFTER: i64 = 3;
+
+/// Whether a finished run is told, and how: the server's word when it already told (`notified`),
+/// else its notify policy for the schedule as the server applies it.
+pub fn decide(policy: &str, status: &str, notified: bool, streak: i64) -> Option<Told> {
+    let failed = matches!(status, "failed" | "skipped");
+    let kind = if failed && streak == PAUSE_AFTER {
+        Told::Pause
+    } else if failed {
+        Told::Failed
+    } else {
+        Told::Result
+    };
+    if notified {
+        return Some(kind);
+    }
+    match (policy, status) {
+        ("never", _) => None,
+        (_, "ok") | ("always", "silent") => Some(Told::Result),
+        _ if failed && (streak <= 1 || streak == PAUSE_AFTER) => Some(kind),
+        _ => None,
+    }
 }
 
 impl Watch {
@@ -120,8 +175,9 @@ impl Watch {
         done
     }
 
-    /// Note the schedules as the server gave them; the new runs to tell of since the last look.
+    /// Note the schedules as the event stream gave them (`{"schedules": [...]}`); the runs that
-    /// The first look only notes, as with the chats.
+    /// finished since the last look and are told. The first look only notes, as with the chats; a
+    /// run is told once, however often the list comes again.
     pub fn look_schedules(&self, list: &Value) -> Vec<Ran> {
         let rows = list
             .get("schedules")
@@ -142,28 +198,48 @@ impl Watch {
             let Some(run_id) = run.get("id").and_then(Value::as_str) else {
                 continue;
             };
+            let status = run.get("status").and_then(Value::as_str).unwrap_or("");
+            let finished = FINISHED.contains(&status);
             let before = seen.as_ref().and_then(|s| s.get(id));
-            let told = run.get("notified").and_then(Value::as_bool) == Some(true);
+            let known = before.is_some_and(|(r, done)| r == run_id && *done);
-            if !first && before.map(String::as_str) != Some(run_id) && told {
+            if !first && finished && !known {
-                out.push(Ran {
+                let told = decide(
-                    title: row
+                    row.get("notify").and_then(Value::as_str).unwrap_or(""),
-                        .get("title")
+                    status,
-                        .and_then(Value::as_str)
+                    run.get("notified").and_then(Value::as_bool) == Some(true),
-                        .unwrap_or("")
+                    row.get("failure_streak")
-                        .to_string(),
+                        .and_then(Value::as_i64)
-                    failed: run.get("status").and_then(Value::as_str) == Some("failed"),
+                        .unwrap_or(0),
-                });
+                );
+                if let Some(told) = told {
+                    out.push(Ran {
+                        title: row
+                            .get("title")
+                            .and_then(Value::as_str)
+                            .unwrap_or("")
+                            .to_string(),
+                        told,
+                    });
+                }
             }
-            now.insert(id.to_string(), run_id.to_string());
+            now.insert(id.to_string(), (run_id.to_string(), finished));
         }
         *seen = Some(now);
         out
     }
 
+    /// The flag that ends the event stream followed now.
+    pub fn stop_flag(&self) -> Arc<AtomicBool> {
+        self.stop.lock().unwrap().clone()
+    }
+
     /// Start again from nothing: another server, or signed out.
     pub fn reset(&self) {
         *self.running.lock().unwrap() = None;
         *self.runs.lock().unwrap() = None;
+        let mut stop = self.stop.lock().unwrap();
+        stop.store(true, Ordering::SeqCst);
+        *stop = Arc::new(AtomicBool::new(false));
     }
 }
 
@@ -172,38 +248,88 @@ mod tests {
     use super::*;
     use serde_json::json;
 
-    /// A schedule's last run in a test: its id, status, and whether the server tells of it.
+    /// A schedule in a test: its id, notify policy, failure streak, and last run (id, status,
-    type LastRun<'a> = Option<(&'a str, &'a str, bool)>;
+    /// whether the server told of it).
+    type Row<'a> = (&'a str, &'a str, i64, Option<(&'a str, &'a str, bool)>);
 
-    fn schedules(rows: &[(&str, LastRun)]) -> Value {
+    fn schedules(rows: &[Row]) -> Value {
-        json!({"schedules": rows.iter().map(|(id, run)| json!({
+        json!({"version": "v", "schedules": rows.iter().map(|(id, notify, streak, run)| json!({
-            "id": id, "title": format!("Task {id}"),
+            "id": id, "title": format!("Task {id}"), "notify": notify, "failure_streak": streak,
             "last_run": run.map(|(r, status, told)| json!({"id": r, "status": status, "notified": told})),
         })).collect::<Vec<_>>()})
     }
 
+    fn ran(id: &str, told: Told) -> Ran {
+        Ran {
+            title: format!("Task {id}"),
+            told,
+        }
+    }
+
     #[test]
-    fn a_new_run_the_server_tells_of_is_said_once_and_a_silent_one_never() {
+    fn a_finished_run_is_said_once_by_its_policy_and_a_silent_one_only_when_always() {
         let w = Watch::default();
         assert!(w
-            .look_schedules(&schedules(&[("a", Some(("r1", "ok", true))), ("b", None)]))
+            .look_schedules(&schedules(&[
+                ("a", "matters", 0, Some(("r1", "ok", false))),
+                ("b", "always", 0, None)
+            ]))
             .is_empty());
-        let ran = w.look_schedules(&schedules(&[
+        // Running is not news; finishing is, even under the same run id.
-            ("a", Some(("r2", "ok", true))),
-            ("b", Some(("r3", "silent", false))),
-        ]));
-        assert_eq!(
-            ran,
-            vec![Ran {
-                title: "Task a".into(),
-                failed: false
-            }]
-        );
         assert!(w
-            .look_schedules(&schedules(&[("a", Some(("r2", "ok", true)))]))
+            .look_schedules(&schedules(&[(
+                "a",
+                "matters",
+                0,
+                Some(("r2", "running", false))
+            )]))
+            .is_empty());
+        let out = w.look_schedules(&schedules(&[
+            ("a", "matters", 0, Some(("r2", "ok", false))),
+            ("b", "always", 0, Some(("r3", "silent", false))),
+            ("c", "matters", 0, Some(("r4", "silent", false))),
+            ("d", "never", 0, Some(("r5", "ok", false))),
+        ]));
+        assert_eq!(out, vec![ran("a", Told::Result), ran("b", Told::Result)]);
+        // The same list again (the server told a device meanwhile): not said twice.
+        assert!(w
+            .look_schedules(&schedules(&[("a", "matters", 0, Some(("r2", "ok", true)))]))
+            .is_empty());
+    }
+
+    #[test]
+    fn failures_are_said_the_first_time_and_the_third_asks_to_pause() {
+        assert_eq!(decide("matters", "failed", false, 1), Some(Told::Failed));
+        assert_eq!(decide("matters", "failed", false, 2), None);
+        assert_eq!(decide("always", "skipped", false, 3), Some(Told::Pause));
+        assert_eq!(decide("never", "failed", false, 1), None);
+        assert_eq!(decide("matters", "missed", false, 0), None);
+        // The server's own word wins: it told a device.
+        assert_eq!(decide("matters", "failed", true, 2), Some(Told::Failed));
+        assert_eq!(decide("matters", "silent", true, 0), Some(Told::Result));
+    }
+
+    #[test]
+    fn a_reset_ends_the_stream_and_starts_quiet() {
+        let w = Watch::default();
+        let old = w.stop_flag();
+        w.look_schedules(&schedules(&[(
+            "a",
+            "matters",
+            0,
+            Some(("r1", "running", false)),
+        )]));
+        w.reset();
+        assert!(old.load(Ordering::SeqCst));
+        assert!(!w.stop_flag().load(Ordering::SeqCst));
+        assert!(w
+            .look_schedules(&schedules(&[(
+                "a",
+                "matters",
+                0,
+                Some(("r1", "ok", false))
+            )]))
             .is_empty());
-        let failed = w.look_schedules(&schedules(&[("a", Some(("r4", "failed", true)))]));
-        assert!(failed[0].failed);
     }
 
     fn list(rows: &[(&str, bool, &str)]) -> Value {