Mark runs lost when their SSH host stays unreachable and deliver pending wake-ups together (OR-628)

src/commands/supervise.rs

@@ -26,6 +26,8 @@ const POLL_INTERVAL: Duration = Duration::from_secs(5);
 const LOG_IDLE: Duration = Duration::from_secs(30);
 /// How long monitoring must keep failing before it is reported on the run.
 const MONITORING_GRACE: Duration = Duration::from_secs(60);
+/// How long an SSH host may stay unreachable before its run is declared lost.
+const UNREACHABLE_LIMIT: Duration = Duration::from_secs(15 * 60);
 
 fn should_report_monitoring_error(already_reported: bool, elapsed: Option<Duration>) -> bool {
     already_reported || elapsed.is_some_and(|elapsed| elapsed >= MONITORING_GRACE)
@@ -269,6 +271,10 @@ fn should_report_cancelled(store: &Store, run_id: &str, cancel_sent: bool) -> bo
     cancel_sent || local_cancel_requested(store, run_id)
 }
 
+fn host_declared_lost(unreachable_for: Option<Duration>) -> bool {
+    unreachable_for.is_some_and(|elapsed| elapsed >= UNREACHABLE_LIMIT)
+}
+
 fn run_status_for_stage(store: &Store, run_id: &str, cancel_sent: bool, stage: &str) -> RunStatus {
     if stage == "LOST" {
         return if cancel_sent {
@@ -706,6 +712,7 @@ async fn watch_ssh_job(
     let mut cancel_sent = descriptor.cancellation_accepted;
     let mut last_message = None;
     let mut failing_since = None;
+    let mut unreachable_since = None;
     let mut last_error = None;
     let mut missing_polls = 0;
 
@@ -768,6 +775,12 @@ async fn watch_ssh_job(
                 Err(err) => eprintln!("supervise {run_id}: could not save monitoring state: {err}"),
             }
         }
+        // Inspection can also fail on a reachable host (e.g. Docker errors); only polls where the
+        // host cannot run a trivial command count toward declaring it gone.
+        let host_unreachable =
+            observed.is_err() && ssh::ssh_run(&target, "true", None).await.is_err();
+        unreachable_since =
+            host_unreachable.then(|| unreachable_since.unwrap_or_else(std::time::Instant::now));
         let job = match observed {
             Ok(job) => job,
             Err(_) => match sandbox {
@@ -780,6 +793,16 @@ async fn watch_ssh_job(
                     .into(),
                     message: Some(format!("box {id} was deleted while the run was active")),
                 },
+                _ if host_declared_lost(unreachable_since.map(|since| since.elapsed())) => {
+                    ssh::JobState {
+                        stage: "LOST".into(),
+                        message: Some(format!(
+                            "{} was unreachable for {} minutes while the run was active",
+                            target.dest,
+                            UNREACHABLE_LIMIT.as_secs() / 60
+                        )),
+                    }
+                }
                 _ => {
                     missing_polls = 0;
                     tokio::time::sleep(POLL_INTERVAL).await;
@@ -1711,6 +1734,15 @@ mod tests {
         assert!(should_report_monitoring_error(true, Some(Duration::ZERO)));
     }
 
+    #[test]
+    fn unreachable_hosts_are_declared_lost_only_after_the_limit() {
+        assert!(!host_declared_lost(None));
+        assert!(!host_declared_lost(Some(
+            UNREACHABLE_LIMIT - Duration::from_nanos(1)
+        )));
+        assert!(host_declared_lost(Some(UNREACHABLE_LIMIT)));
+    }
+
     #[test]
     fn recovered_cancel_intent_only_overrides_failed_terminal_state() {
         let dir = std::env::temp_dir().join(format!(

src/local/chat/mod.rs

@@ -8380,12 +8380,27 @@ fn run_monitoring_text(runs: &[crate::store::StoredRun]) -> String {
     )
 }
 
-fn first_wakeup_per_session(wakeups: Vec<crate::store::RunWakeup>) -> Vec<crate::store::RunWakeup> {
-    let mut seen_sessions = HashSet::new();
-    wakeups
-        .into_iter()
-        .filter(|wakeup| seen_sessions.insert(wakeup.chat_session_id.clone()))
-        .collect()
+/// Groups each session's ready wake-ups so they arrive together in one turn. Sessions with a
+/// wake-up claimed elsewhere wait until that claim settles.
+fn pending_wakeups_by_session(
+    wakeups: Vec<crate::store::RunWakeup>,
+) -> Vec<Vec<crate::store::RunWakeup>> {
+    let mut claimed = HashSet::new();
+    let mut groups: Vec<Vec<crate::store::RunWakeup>> = Vec::new();
+    for wakeup in wakeups {
+        if wakeup.state != "pending" {
+            claimed.insert(wakeup.chat_session_id.clone());
+        } else if let Some(group) = groups
+            .iter_mut()
+            .find(|group| group[0].chat_session_id == wakeup.chat_session_id)
+        {
+            group.push(wakeup);
+        } else {
+            groups.push(vec![wakeup]);
+        }
+    }
+    groups.retain(|group| !claimed.contains(&group[0].chat_session_id));
+    groups
 }
 
 async fn process_run_wakeups(
@@ -8398,11 +8413,9 @@ async fn process_run_wakeups(
         return Ok(());
     }
     store.prune_run_wakeups()?;
-    for wakeup in first_wakeup_per_session(store.list_ready_run_wakeups()?) {
-        if wakeup.state != "pending" {
-            continue;
-        }
-        let Some(mut guard) = TurnGuard::claim_hidden(chat, &wakeup.chat_session_id).await else {
+    for wakeups in pending_wakeups_by_session(store.list_ready_run_wakeups()?) {
+        let session_id = wakeups[0].chat_session_id.clone();
+        let Some(mut guard) = TurnGuard::claim_hidden(chat, &session_id).await else {
             continue;
         };
         if data_dir_move_in_progress
@@ -8411,32 +8424,47 @@ async fn process_run_wakeups(
             guard.release().await;
             return Ok(());
         }
-        let Some(token) = store.claim_run_wakeup(&wakeup.run.id, &wakeup.chat_session_id)? else {
-            guard.release().await;
-            continue;
-        };
-        if !store.renew_run_wakeup_claim(&wakeup.run.id, &wakeup.chat_session_id, &token)? {
+        let mut claimed = Vec::new();
+        let mut texts = Vec::new();
+        let collected = (|| -> Result<()> {
+            for wakeup in wakeups {
+                let Some(token) = store.claim_run_wakeup(&wakeup.run.id, &session_id)? else {
+                    continue;
+                };
+                if !store.renew_run_wakeup_claim(&wakeup.run.id, &session_id, &token)? {
+                    continue;
+                }
+                let Some(text) = run_wakeup_text(&wakeup.run) else {
+                    store.release_run_wakeup(&wakeup.run.id, &session_id, &token)?;
+                    continue;
+                };
+                texts.push(text);
+                claimed.push((wakeup.run.id, token));
+            }
+            Ok(())
+        })();
+        if let Err(err) = collected {
+            for (run_id, token) in &claimed {
+                let _ = store.release_run_wakeup(run_id, &session_id, token);
+            }
             guard.release().await;
-            continue;
+            return Err(err);
         }
-        let Some(text) = run_wakeup_text(&wakeup.run) else {
-            store.release_run_wakeup(&wakeup.run.id, &wakeup.chat_session_id, &token)?;
+        if claimed.is_empty() {
             guard.release().await;
             continue;
-        };
+        }
         match chat
-            .send_hidden_message(&wakeup.chat_session_id, text, guard)
+            .send_hidden_message(&session_id, texts.join("\n\n"), guard)
             .await
         {
             Ok(TurnSubmission::Started(_)) => {
-                if !store.mark_run_wakeup_delivered(
-                    &wakeup.run.id,
-                    &wakeup.chat_session_id,
-                    &token,
-                )? {
-                    return Err(anyhow!(
-                        "run wake-up claim expired before delivery was recorded"
-                    ));
+                for (run_id, token) in &claimed {
+                    if !store.mark_run_wakeup_delivered(run_id, &session_id, token)? {
+                        return Err(anyhow!(
+                            "run wake-up claim expired before delivery was recorded"
+                        ));
+                    }
                 }
             }
             Ok(
@@ -8445,11 +8473,15 @@ async fn process_run_wakeups(
                 | TurnSubmission::Existing(_)
                 | TurnSubmission::NotStarted,
             ) => {
-                store.release_run_wakeup(&wakeup.run.id, &wakeup.chat_session_id, &token)?;
+                for (run_id, token) in &claimed {
+                    store.release_run_wakeup(run_id, &session_id, token)?;
+                }
             }
             Err(err) => {
-                store.release_run_wakeup(&wakeup.run.id, &wakeup.chat_session_id, &token)?;
-                if !chat.is_busy(&wakeup.chat_session_id).await {
+                for (run_id, token) in &claimed {
+                    store.release_run_wakeup(run_id, &session_id, token)?;
+                }
+                if !chat.is_busy(&session_id).await {
                     eprintln!("orx up: run watcher: {err}");
                 }
             }
@@ -10597,28 +10629,30 @@ with other project runs using `orx runs p1` and inspect the file located by `orx
     }
 
     #[test]
-    fn only_earliest_wakeup_per_session_is_attempted_each_tick() {
-        let mut first = run("done");
-        first.id = "run_first".into();
-        let mut second = run("done");
-        second.id = "run_second".into();
-        let selected = first_wakeup_per_session(vec![
+    fn ready_wakeups_are_grouped_per_session_unless_one_is_claimed() {
+        let wakeup = |id: &str, session: &str, state: &str| {
+            let mut run = run("done");
+            run.id = id.into();
             crate::store::RunWakeup {
-                run: first,
-                chat_session_id: "owner".into(),
-                state: "pending".into(),
+                run,
+                chat_session_id: session.into(),
+                state: state.into(),
                 monitoring_alerted: false,
-            },
-            crate::store::RunWakeup {
-                run: second,
-                chat_session_id: "owner".into(),
-                state: "pending".into(),
-                monitoring_alerted: false,
-            },
+            }
+        };
+        let groups = pending_wakeups_by_session(vec![
+            wakeup("a1", "a", "pending"),
+            wakeup("b1", "b", "pending"),
+            wakeup("a2", "a", "pending"),
+            wakeup("c1", "c", "pending"),
+            wakeup("c2", "c", "claimed"),
         ]);
 
-        assert_eq!(selected.len(), 1);
-        assert_eq!(selected[0].run.id, "run_first");
+        let ids = groups
+            .iter()
+            .map(|group| group.iter().map(|w| w.run.id.as_str()).collect::<Vec<_>>())
+            .collect::<Vec<_>>();
+        assert_eq!(ids, vec![vec!["a1", "a2"], vec!["b1"]]);
     }
 
     #[test]