diff --git a/src/commands/supervise.rs b/src/commands/supervise.rs index de8314ef2..d429524db 100644 --- a/src/commands/supervise.rs +++ b/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) -> 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) -> 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!( diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 7cb2fce99..4214a93ac 100644 --- a/src/local/chat/mod.rs +++ b/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) -> Vec { - 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, +) -> Vec> { + let mut claimed = HashSet::new(); + let mut groups: Vec> = 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::>()) + .collect::>(); + assert_eq!(ids, vec![vec!["a1", "a2"], vec!["b1"]]); } #[test]