From 0dfbbf90bf837df43285c73d695ec7c523ee6be7 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 26 Oct 2026 22:58:07 +0000 Subject: [PATCH 1/5] Mark runs lost when their SSH host stays unreachable and batch pending wake-ups per session (OR-628) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/commands/supervise.rs | 18 ++++++ src/local/chat/mod.rs | 130 ++++++++++++++++++++++---------------- 2 files changed, 95 insertions(+), 53 deletions(-) diff --git a/src/commands/supervise.rs b/src/commands/supervise.rs index de8314ef2..aabe792be 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) @@ -706,6 +708,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 +771,9 @@ async fn watch_ssh_job( Err(err) => eprintln!("supervise {run_id}: could not save monitoring state: {err}"), } } + unreachable_since = observed + .is_err() + .then(|| unreachable_since.unwrap_or_else(std::time::Instant::now)); let job = match observed { Ok(job) => job, Err(_) => match sandbox { @@ -780,6 +786,18 @@ async fn watch_ssh_job( .into(), message: Some(format!("box {id} was deleted while the run was active")), }, + _ if unreachable_since + .is_some_and(|since| since.elapsed() >= UNREACHABLE_LIMIT) => + { + 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; diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 7cb2fce99..68eb67abe 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -8376,16 +8376,31 @@ fn run_monitoring_text(runs: &[crate::store::StoredRun]) -> String { format!( "[orx] orx can no longer monitor these live runs:\n{lines}\nRun `orx exp status ` \ for the reason and tell the user so they can fix it. orx wakes you once it can see a run \ - finish; until then each run keeps its current status." + finish, or marks it lost if its host stays unreachable for 15 minutes." ) } -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,37 @@ 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)? { - guard.release().await; - continue; + let mut claimed = Vec::new(); + let mut texts = Vec::new(); + 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)); } - 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 +8463,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 +10619,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![ - crate::store::RunWakeup { - run: first, - chat_session_id: "owner".into(), - state: "pending".into(), - monitoring_alerted: false, - }, + 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: second, - chat_session_id: "owner".into(), - state: "pending".into(), + run, + chat_session_id: session.into(), + state: state.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] @@ -10809,7 +10833,7 @@ with other project runs using `orx runs p1` and inspect the file located by `orx - run `run_x` of experiment `exp_1` (still **starting**)\n\ - run `run_y` of experiment `exp_1` (still **running**)\n\ Run `orx exp status ` for the reason and tell the user so they can fix it. orx wakes you \ -once it can see a run finish; until then each run keeps its current status." +once it can see a run finish, or marks it lost if its host stays unreachable for 15 minutes." ); } From f4cf92a2d5bb158d5e37311cba12e1d44dd8012b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:02:28 +0000 Subject: [PATCH 2/5] Test the unreachable-host limit that declares SSH runs lost (OR-628) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/commands/supervise.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/src/commands/supervise.rs b/src/commands/supervise.rs index aabe792be..4f125da2d 100644 --- a/src/commands/supervise.rs +++ b/src/commands/supervise.rs @@ -271,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 { @@ -786,9 +790,7 @@ async fn watch_ssh_job( .into(), message: Some(format!("box {id} was deleted while the run was active")), }, - _ if unreachable_since - .is_some_and(|since| since.elapsed() >= UNREACHABLE_LIMIT) => - { + _ if host_declared_lost(unreachable_since.map(|since| since.elapsed())) => { ssh::JobState { stage: "LOST".into(), message: Some(format!( @@ -1729,6 +1731,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!( From 7b2c0ffbb03727a04340e1bc85916322e91ff7a2 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:08:29 +0000 Subject: [PATCH 3/5] Only declare SSH runs lost when the host cannot run a trivial command; keep the outage alert backend-neutral (OR-628) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/commands/supervise.rs | 6 +++++- src/local/chat/mod.rs | 6 +++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/src/commands/supervise.rs b/src/commands/supervise.rs index 4f125da2d..0b4ad000d 100644 --- a/src/commands/supervise.rs +++ b/src/commands/supervise.rs @@ -790,7 +790,11 @@ 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())) => { + // Inspection can also fail on a reachable host (e.g. Docker errors); only a host + // that cannot run a trivial command is treated as gone. + _ if host_declared_lost(unreachable_since.map(|since| since.elapsed())) + && ssh::ssh_run(&target, "true", None).await.is_err() => + { ssh::JobState { stage: "LOST".into(), message: Some(format!( diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 68eb67abe..02e06996d 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -8375,8 +8375,8 @@ fn run_monitoring_text(runs: &[crate::store::StoredRun]) -> String { .join("\n"); format!( "[orx] orx can no longer monitor these live runs:\n{lines}\nRun `orx exp status ` \ - for the reason and tell the user so they can fix it. orx wakes you once it can see a run \ - finish, or marks it lost if its host stays unreachable for 15 minutes." + for the reason and tell the user so they can fix it. orx wakes you once a run \ + reaches a final status." ) } @@ -10833,7 +10833,7 @@ with other project runs using `orx runs p1` and inspect the file located by `orx - run `run_x` of experiment `exp_1` (still **starting**)\n\ - run `run_y` of experiment `exp_1` (still **running**)\n\ Run `orx exp status ` for the reason and tell the user so they can fix it. orx wakes you \ -once it can see a run finish, or marks it lost if its host stays unreachable for 15 minutes." +once a run reaches a final status." ); } From ab03c7f908a3b0fc356ae67113eec018688f3be9 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:11:47 +0000 Subject: [PATCH 4/5] Release claimed wake-ups if collecting a batch fails; keep the original outage alert text (OR-628) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/local/chat/mod.rs | 40 +++++++++++++++++++++++++--------------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 02e06996d..4214a93ac 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -8375,8 +8375,8 @@ fn run_monitoring_text(runs: &[crate::store::StoredRun]) -> String { .join("\n"); format!( "[orx] orx can no longer monitor these live runs:\n{lines}\nRun `orx exp status ` \ - for the reason and tell the user so they can fix it. orx wakes you once a run \ - reaches a final status." + for the reason and tell the user so they can fix it. orx wakes you once it can see a run \ + finish; until then each run keeps its current status." ) } @@ -8426,19 +8426,29 @@ async fn process_run_wakeups( } let mut claimed = Vec::new(); let mut texts = Vec::new(); - 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 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)); } - 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; + return Err(err); } if claimed.is_empty() { guard.release().await; @@ -10833,7 +10843,7 @@ with other project runs using `orx runs p1` and inspect the file located by `orx - run `run_x` of experiment `exp_1` (still **starting**)\n\ - run `run_y` of experiment `exp_1` (still **running**)\n\ Run `orx exp status ` for the reason and tell the user so they can fix it. orx wakes you \ -once a run reaches a final status." +once it can see a run finish; until then each run keeps its current status." ); } From 6506ea8d99c8ba58572132a5f808249c83115bd8 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:22:45 +0000 Subject: [PATCH 5/5] Count only polls where the SSH host fails a trivial command toward the lost deadline (OR-628) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/commands/supervise.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/src/commands/supervise.rs b/src/commands/supervise.rs index 0b4ad000d..d429524db 100644 --- a/src/commands/supervise.rs +++ b/src/commands/supervise.rs @@ -775,9 +775,12 @@ async fn watch_ssh_job( Err(err) => eprintln!("supervise {run_id}: could not save monitoring state: {err}"), } } - unreachable_since = observed - .is_err() - .then(|| unreachable_since.unwrap_or_else(std::time::Instant::now)); + // 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 { @@ -790,11 +793,7 @@ async fn watch_ssh_job( .into(), message: Some(format!("box {id} was deleted while the run was active")), }, - // Inspection can also fail on a reachable host (e.g. Docker errors); only a host - // that cannot run a trivial command is treated as gone. - _ if host_declared_lost(unreachable_since.map(|since| since.elapsed())) - && ssh::ssh_run(&target, "true", None).await.is_err() => - { + _ if host_declared_lost(unreachable_since.map(|since| since.elapsed())) => { ssh::JobState { stage: "LOST".into(), message: Some(format!(