From cab8acc7163ae12ccba67f44d5862d989628fa35 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Fri, 9 Oct 2026 13:29:28 -0400 Subject: [PATCH] feat(agents): checkpoint after turns and recover lost sessions from them A node lost without warning sends no SIGTERM, so the checkpoint on stop never runs. The harness now also saves in the background after completed turns (one at a time, at most once per BUZZ_ACP_CHECKPOINT_AFTER_TURN_SECS, default 600), and the save on stop waits for or cancels a running one. A cancelled or failed save kills its helper's whole process group. When a Pod left no termination receipt, the reaper records the newest checkpoint directly under the session's own prefix instead of `missing`. turn_completed activity records now carry duration_ms for long-turn alerts. Signed-off-by: Tee8z --- crates/buzz-acp/src/activity.rs | 19 ++ crates/buzz-acp/src/checkpoint.rs | 286 ++++++++++++++++-- crates/buzz-acp/src/config.rs | 14 + crates/buzz-acp/src/lib.rs | 13 + crates/buzz-acp/src/pool.rs | 5 +- crates/buzz-sandbox-reaper/src/evidence.rs | 97 +++++- crates/buzz-sandbox-reaper/src/plan.rs | 8 +- crates/buzz-sandbox-reaper/src/reconcile.rs | 13 + .../src/reconcile_tests.rs | 21 ++ docs/remote-agents.md | 24 +- 10 files changed, 463 insertions(+), 37 deletions(-) diff --git a/crates/buzz-acp/src/activity.rs b/crates/buzz-acp/src/activity.rs index cd63a6f86dd..6200a36c086 100644 --- a/crates/buzz-acp/src/activity.rs +++ b/crates/buzz-acp/src/activity.rs @@ -45,6 +45,9 @@ impl ActivityRecorder { let before = self.tools.len(); self.tools.retain(|_, (_, turn)| turn != &event.turn_id); details["unfinished_tools"] = (before - self.tools.len()).into(); + if let Some(duration_ms) = event.payload["duration_ms"].as_u64() { + details["duration_ms"] = duration_ms.into(); + } } event.kind.as_str() } @@ -252,6 +255,22 @@ mod tests { assert!(recorder.record(&e).is_none()); } + #[test] + fn turn_completion_reports_only_its_duration() { + let mut recorder = ActivityRecorder::new(); + let mut completed = event(json!({"duration_ms": 4200, "prompt": "synthetic secret"})); + completed.kind = "turn_completed".into(); + let record = recorder.record(&completed).unwrap(); + assert_eq!(record["details"]["duration_ms"], 4200); + assert_eq!(record["details"]["unfinished_tools"], 0); + assert!(!record.to_string().contains("synthetic secret")); + let mut legacy = event(json!({})); + legacy.kind = "turn_completed".into(); + assert!(recorder.record(&legacy).unwrap()["details"] + .get("duration_ms") + .is_none()); + } + #[test] fn active_tool_tracking_is_bounded() { let mut recorder = ActivityRecorder::new(); diff --git a/crates/buzz-acp/src/checkpoint.rs b/crates/buzz-acp/src/checkpoint.rs index aa9cda7e6b8..68a8d41ffb9 100644 --- a/crates/buzz-acp/src/checkpoint.rs +++ b/crates/buzz-acp/src/checkpoint.rs @@ -1,5 +1,8 @@ -//! Checkpoint the workspace after all adapter processes have stopped. +//! Checkpoint the workspace after all adapter processes have stopped, and in +//! the background after completed turns. +use std::future::Future; use std::path::Path; +use std::pin::Pin; use std::time::Duration; use anyhow::{ensure, Context, Result}; @@ -44,15 +47,80 @@ fn receipt(bytes: &[u8]) -> Result { const TERMINATION_LOG: &str = "/dev/termination-log"; pub(crate) async fn save(observer: Option<&ObserverHandle>) -> Result<()> { + save_to(observer, Some(Path::new(TERMINATION_LOG))).await +} + +async fn save_to(observer: Option<&ObserverHandle>, termination_log: Option<&Path>) -> Result<()> { let mut command = Command::new("buzz-agent-checkpoint"); command.arg("save"); - save_with_command( - observer, - command, - Duration::from_secs(120), - Path::new(TERMINATION_LOG), - ) - .await + save_with_command(observer, command, Duration::from_secs(120), termination_log).await +} + +type SaveFuture = Pin + Send>>; + +/// Background checkpoints after completed turns. A node lost without warning +/// sends no SIGTERM, so the save on stop never runs; these bound the loss to +/// the work since the last one. One save runs at a time, at most one starts +/// per `interval`, and the save on stop still runs after the adapters exit. +pub(crate) struct TurnCheckpoints { + interval: Duration, + start: Box SaveFuture + Send>, + last_started: Option, + running: Option>, +} + +impl TurnCheckpoints { + /// Saves through `buzz-agent-checkpoint save`; `interval` zero disables. + pub(crate) fn new(interval: Duration, observer: Option) -> Self { + Self::with_start( + interval, + Box::new(move || { + let observer = observer.clone(); + Box::pin(async move { + if let Err(error) = save_to(observer.as_ref(), None).await { + tracing::warn!(%error, "turn checkpoint failed"); + } + }) + }), + ) + } + + fn with_start(interval: Duration, start: Box SaveFuture + Send>) -> Self { + Self { + interval, + start, + last_started: None, + running: None, + } + } + + /// Start a background save after a completed turn, unless one is running + /// or the last one started less than `interval` ago. + pub(crate) fn after_turn(&mut self) { + if self.interval.is_zero() || self.running.as_ref().is_some_and(|t| !t.is_finished()) { + return; + } + let now = tokio::time::Instant::now(); + if self + .last_started + .is_some_and(|at| now.duration_since(at) < self.interval) + { + return; + } + self.last_started = Some(now); + self.running = Some(tokio::spawn((self.start)())); + } + + /// Before the save on stop: let a running save finish within `bound`, + /// otherwise cancel it (its helper process group is killed). + pub(crate) async fn settle(&mut self, bound: Duration) { + if let Some(mut task) = self.running.take() { + if tokio::time::timeout(bound, &mut task).await.is_err() { + task.abort(); + let _ = task.await; + } + } + } } /// Record the saved checkpoint key as the container's termination message: @@ -74,8 +142,16 @@ async fn save_with_command( observer: Option<&ObserverHandle>, mut command: Command, timeout: Duration, - termination_log: &Path, + termination_log: Option<&Path>, ) -> Result<()> { + // Only shutdown may attest that all workspace writes have stopped. Clear + // any earlier receipt before attempting the final save, including failures. + if let Some(path) = termination_log { + let _ = std::fs::OpenOptions::new() + .write(true) + .truncate(true) + .open(path); + } let emit = |kind, details| { if let Some(observer) = observer { observer.emit(kind, None, &ObserverContext::default(), details); @@ -91,8 +167,8 @@ async fn save_with_command( let saved = run_checkpoint(command, timeout).await; match saved { Ok(details) => { - if let Some(key) = details["key"].as_str() { - write_termination_receipt(termination_log, key); + if let (Some(path), Some(key)) = (termination_log, details["key"].as_str()) { + write_termination_receipt(path, key); } emit("checkpoint_saved", details); Ok(()) @@ -112,6 +188,10 @@ async fn run_checkpoint(mut command: Command, timeout: Duration) -> Result Result Ok(details), @@ -137,6 +212,21 @@ async fn run_checkpoint(mut command: Command, timeout: Duration) -> Result); + +#[cfg(unix)] +impl Drop for KillGroupOnDrop { + fn drop(&mut self) { + if let Some(pid) = self.0 { + let _ = nix::sys::signal::killpg( + nix::unistd::Pid::from_raw(pid), + nix::sys::signal::Signal::SIGKILL, + ); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -154,7 +244,7 @@ mod tests { None, command, Duration::from_secs(10), - Path::new(NO_LOG_PATH) + Some(Path::new(NO_LOG_PATH)) ) .await .is_err()); @@ -185,7 +275,7 @@ mod tests { Some(&observer), command, Duration::from_millis(100), - Path::new(NO_LOG_PATH) + Some(Path::new(NO_LOG_PATH)) ) .await .is_err()); @@ -204,7 +294,7 @@ mod tests { Some(&observer), command, Duration::from_secs(1), - Path::new(NO_LOG_PATH) + Some(Path::new(NO_LOG_PATH)) ) .await .is_err()); @@ -226,7 +316,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let log = dir.path().join("termination-log"); std::fs::write(&log, "stale").unwrap(); - save_with_command(Some(&observer), command, Duration::from_secs(1), &log) + save_with_command(Some(&observer), command, Duration::from_secs(1), Some(&log)) .await .unwrap(); let events = observer.snapshot(); @@ -253,7 +343,7 @@ mod tests { .to_string(); let mut command = Command::new("sh"); command.args(["-c", "printf '%s' \"$1\"", "checkpoint-test", &valid]); - save_with_command(None, command, Duration::from_secs(1), &absent) + save_with_command(None, command, Duration::from_secs(1), Some(&absent)) .await .unwrap(); assert!(!absent.exists()); @@ -262,10 +352,158 @@ mod tests { let mut failing = Command::new("sh"); failing.args(["-c", "exit 1"]); assert!( - save_with_command(None, failing, Duration::from_secs(1), &present) + save_with_command(None, failing, Duration::from_secs(1), Some(&present)) .await .is_err() ); assert_eq!(std::fs::read_to_string(&present).unwrap(), ""); } + + #[cfg(unix)] + #[tokio::test] + async fn failed_final_save_invalidates_an_earlier_receipt() { + let dir = tempfile::tempdir().unwrap(); + let log = dir.path().join("termination-log"); + std::fs::write(&log, r#"{"version":1,"checkpoint":"dev/g/old.tar.gz"}"#).unwrap(); + let mut command = Command::new("sh"); + command.args(["-c", "exit 1"]); + assert!( + save_with_command(None, command, Duration::from_secs(1), Some(&log)) + .await + .is_err() + ); + assert!(std::fs::read_to_string(&log).unwrap().is_empty()); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_turn_save_emits_a_checkpoint_without_a_termination_receipt() { + let observer = ObserverHandle::in_process(); + let valid = serde_json::json!({"version":1,"key":"dev/g/turn.tar.gz", + "sha256":"a".repeat(64),"repositories":1,"bytes":100}) + .to_string(); + let mut command = Command::new("sh"); + command.args(["-c", "printf '%s' \"$1\"", "checkpoint-test", &valid]); + save_with_command(Some(&observer), command, Duration::from_secs(1), None) + .await + .unwrap(); + let events = observer.snapshot(); + assert_eq!(events[0].kind, "checkpoint_started"); + let saved = &events[1]; + assert_eq!(saved.kind, "checkpoint_saved"); + assert_eq!(saved.payload["key"], "dev/g/turn.tar.gz"); + } + + fn counting( + interval: Duration, + hold: Duration, + ) -> ( + TurnCheckpoints, + std::sync::Arc, + ) { + let started = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let count = started.clone(); + let checkpoints = TurnCheckpoints::with_start( + interval, + Box::new(move || { + count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Box::pin(tokio::time::sleep(hold)) + }), + ); + (checkpoints, started) + } + + async fn settle_tasks() { + for _ in 0..4 { + tokio::task::yield_now().await; + } + } + + #[tokio::test(start_paused = true)] + async fn turn_checkpoints_are_single_flight_and_rate_limited() { + use std::sync::atomic::Ordering::SeqCst; + let (mut checkpoints, started) = + counting(Duration::from_secs(600), Duration::from_secs(30)); + checkpoints.after_turn(); + settle_tasks().await; + assert_eq!(started.load(SeqCst), 1); + // Within the interval: no new save. + checkpoints.after_turn(); + assert_eq!(started.load(SeqCst), 1); + // Past the interval but a save still running: no second save. + let (mut slow, slow_started) = counting(Duration::from_secs(10), Duration::from_secs(60)); + slow.after_turn(); + settle_tasks().await; + tokio::time::advance(Duration::from_secs(20)).await; + slow.after_turn(); + assert_eq!(slow_started.load(SeqCst), 1); + // Finished and past the interval: the next turn saves again. + tokio::time::advance(Duration::from_secs(600)).await; + settle_tasks().await; + checkpoints.after_turn(); + settle_tasks().await; + assert_eq!(started.load(SeqCst), 2); + } + + #[tokio::test(start_paused = true)] + async fn zero_interval_disables_turn_checkpoints() { + let (mut checkpoints, started) = counting(Duration::ZERO, Duration::ZERO); + checkpoints.after_turn(); + settle_tasks().await; + assert_eq!(started.load(std::sync::atomic::Ordering::SeqCst), 0); + } + + #[tokio::test(start_paused = true)] + async fn settle_waits_for_a_running_save_then_cancels_past_the_bound() { + let (mut quick, _) = counting(Duration::from_secs(600), Duration::from_secs(5)); + quick.after_turn(); + settle_tasks().await; + let before = tokio::time::Instant::now(); + quick.settle(Duration::from_secs(60)).await; + assert_eq!(before.elapsed(), Duration::from_secs(5)); + let (mut stuck, _) = counting(Duration::from_secs(600), Duration::from_secs(3600)); + stuck.after_turn(); + settle_tasks().await; + let before = tokio::time::Instant::now(); + stuck.settle(Duration::from_secs(60)).await; + assert_eq!(before.elapsed(), Duration::from_secs(60)); + assert!(stuck.running.is_none()); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_cancelled_save_kills_its_helper_process_group() { + let dir = tempfile::tempdir().unwrap(); + let pid_file = dir.path().join("child.pid"); + let mut command = Command::new("sh"); + command.args([ + "-c", + &format!("sleep 300 & echo $! > {}; wait", pid_file.display()), + ]); + command + .kill_on_drop(true) + .stdout(std::process::Stdio::piped()); + command.process_group(0); + let save = tokio::spawn(run_checkpoint(command, Duration::from_secs(600))); + let child_pid = loop { + if let Ok(text) = std::fs::read_to_string(&pid_file) { + if let Ok(pid) = text.trim().parse::() { + break pid; + } + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + save.abort(); + let _ = save.await; + let pid = nix::unistd::Pid::from_raw(child_pid); + let mut gone = false; + for _ in 0..100 { + if nix::sys::signal::kill(pid, None).is_err() { + gone = true; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(gone, "background helper survived cancellation"); + } } diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index a23b469a89b..96fd1bb912f 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -553,6 +553,15 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_CHECKPOINT_ON_STOP", default_value_t = false)] pub checkpoint_on_stop: bool, + /// With --checkpoint-on-stop, also save in the background after completed + /// turns, at most once per this many seconds. 0 disables. + #[arg( + long, + env = "BUZZ_ACP_CHECKPOINT_AFTER_TURN_SECS", + default_value_t = 600 + )] + pub checkpoint_after_turn_secs: u64, + /// Exit after this many seconds with no dispatched events and no turn in flight. /// 0 disables inactivity self-termination. #[arg(long, env = "BUZZ_ACP_EXIT_AFTER_INACTIVITY", default_value_t = 0)] @@ -662,6 +671,9 @@ pub struct Config { pub activity_log: bool, pub bound_session: Option, pub checkpoint_on_stop: bool, + /// Minimum seconds between background checkpoints after turns; 0 = disabled. + /// Applies only with `checkpoint_on_stop`. + pub checkpoint_after_turn_secs: u64, /// Seconds without dispatched events before an idle harness exits. 0 = disabled. pub exit_after_inactivity_secs: u64, /// Whether ACP/LLM subprocess initialization is deferred until accepted work arrives. @@ -1265,6 +1277,7 @@ impl Config { activity_log: args.activity_log, bound_session: args.bound_session, checkpoint_on_stop: args.checkpoint_on_stop, + checkpoint_after_turn_secs: args.checkpoint_after_turn_secs, exit_after_inactivity_secs: args.exit_after_inactivity, lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, @@ -1754,6 +1767,7 @@ mod tests { activity_log: false, bound_session: None, checkpoint_on_stop: false, + checkpoint_after_turn_secs: 0, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 411ece5576f..73ad7e81a1b 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3157,6 +3157,14 @@ async fn run_harness( }; let mut typing_channels: HashMap = HashMap::new(); let mut presence_task: Option> = None; + let mut turn_checkpoints = checkpoint::TurnCheckpoints::new( + if config.checkpoint_on_stop { + Duration::from_secs(config.checkpoint_after_turn_secs) + } else { + Duration::ZERO + }, + observer.clone(), + ); // Independent of pool readiness: a never-mentioned lazy agent must still // self-terminate. The watch interval is capped so small configured bounds @@ -4047,6 +4055,7 @@ async fn run_harness( } } Some(PoolEvent::Result(result)) => { + turn_checkpoints.after_turn(); // Stop the typing indicator for the completed turn's exact scope, // not the whole channel — a sibling thread still running in the // same channel must keep its indicator. @@ -4438,6 +4447,8 @@ async fn run_harness( } } + // Never overlap the save on stop with a background turn checkpoint. + turn_checkpoints.settle(Duration::from_secs(60)).await; let checkpoint_result = if config.checkpoint_on_stop { checkpoint::save(observer.as_ref()).await } else { @@ -10270,6 +10281,7 @@ mod build_mcp_servers_tests { activity_log: false, bound_session: None, checkpoint_on_stop: false, + checkpoint_after_turn_secs: 0, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, @@ -11370,6 +11382,7 @@ mod error_outcome_emission_tests { activity_log: false, bound_session: None, checkpoint_on_stop: false, + checkpoint_after_turn_secs: 0, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 079de1340c6..3df2e62e430 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -5266,6 +5266,7 @@ struct TurnCompletionGuard { agent_index: Option, channel_id: Option, turn_id: String, + started: std::time::Instant, } impl TurnCompletionGuard { @@ -5280,6 +5281,7 @@ impl TurnCompletionGuard { agent_index, channel_id, turn_id, + started: std::time::Instant::now(), } } } @@ -5288,11 +5290,12 @@ impl Drop for TurnCompletionGuard { fn drop(&mut self) { if let Some(observer) = self.observer.take() { let context = observer::context_for(self.channel_id, None, Some(self.turn_id.clone())); + let duration_ms = u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX); observer.emit( "turn_completed", self.agent_index, &context, - serde_json::json!({}), + serde_json::json!({ "duration_ms": duration_ms }), ); } } diff --git a/crates/buzz-sandbox-reaper/src/evidence.rs b/crates/buzz-sandbox-reaper/src/evidence.rs index 506f31900dd..0f0acef57e1 100644 --- a/crates/buzz-sandbox-reaper/src/evidence.rs +++ b/crates/buzz-sandbox-reaper/src/evidence.rs @@ -27,6 +27,9 @@ pub trait CheckpointStore: Send + Sync { prefix: &str, key: &str, ) -> impl Future> + Send; + + /// The most recently written checkpoint object directly under `prefix`. + fn newest(&self, prefix: &str) -> impl Future, String>> + Send; } /// S3 store using `ListObjectsV2` only (IAM: `s3:ListBucket`). @@ -57,8 +60,14 @@ impl S3Store { } } -impl CheckpointStore for S3Store { - async fn contains(&self, prefix: &str, key: &str) -> Result { +impl S3Store { + /// Visit every listed object under `prefix` until `visit` returns true. + /// Returns whether a visit stopped the walk. + async fn walk( + &self, + prefix: &str, + mut visit: impl FnMut(&aws_sdk_s3::types::Object) -> bool, + ) -> Result { let mut token: Option = None; for _ in 0..MAX_PAGES { let page = self @@ -75,11 +84,7 @@ impl CheckpointStore for S3Store { aws_sdk_s3::error::DisplayErrorContext(e) ) })?; - if page - .contents() - .iter() - .any(|object| object.key() == Some(key)) - { + if page.contents().iter().any(&mut visit) { return Ok(true); } match page.next_continuation_token() { @@ -93,6 +98,31 @@ impl CheckpointStore for S3Store { } } +impl CheckpointStore for S3Store { + async fn contains(&self, prefix: &str, key: &str) -> Result { + self.walk(prefix, |object| object.key() == Some(key)).await + } + + async fn newest(&self, prefix: &str) -> Result, String> { + let mut newest: Option<((i64, u32), String)> = None; + self.walk(prefix, |object| { + if let (Some(key), Some(at)) = (object.key(), object.last_modified()) { + let at = (at.secs(), at.subsec_nanos()); + if is_session_key(key, prefix) + && newest + .as_ref() + .is_none_or(|(best, best_key)| (at, key) > (*best, best_key.as_str())) + { + newest = Some((at, key.to_string())); + } + } + false + }) + .await?; + Ok(newest.map(|(_, key)| key)) + } +} + /// Evidence for a terminated container's `message` within `session_prefix` /// (`/`). pub async fn from_receipt( @@ -106,6 +136,22 @@ pub async fn from_receipt( probe(store, session_prefix, key, Evidence::Verified).await } +/// Evidence for a session that left no receipt, as a node lost without +/// warning never writes its termination message: the newest checkpoint in the +/// session's own prefix, from the harness's background saves after turns. +/// Only objects directly under `/` qualify. +pub async fn latest(store: &S, session_prefix: &str) -> Evidence { + match store.newest(session_prefix).await { + Ok(Some(key)) => Evidence::Listed(key), + Ok(None) => Evidence::Missing, + Err(error) => { + tracing::warn!(%error, "checkpoint listing failed; deferring decision"); + metrics::counter!("buzz_sandbox_reaper_s3_errors_total").increment(1); + Evidence::Unavailable + } + } +} + /// Evidence for a recovered session that left no verified checkpoint of its /// own: the checkpoint it was restored from, if that object still exists /// under the configured developer prefix. The key can belong to an older @@ -181,6 +227,20 @@ pub mod tests { } Ok(self.keys.iter().any(|k| k == key && k.starts_with(prefix))) } + + /// Insertion order stands in for LastModified. + async fn newest(&self, prefix: &str) -> Result, String> { + self.probes.lock().unwrap().push(prefix.to_string()); + if self.fail { + return Err("synthetic outage".into()); + } + Ok(self + .keys + .iter() + .rev() + .find(|k| is_session_key(k, prefix)) + .cloned()) + } } const PREFIX: &str = "dev/0123456789abcdef0123456789abcdef/"; @@ -273,4 +333,27 @@ pub mod tests { Evidence::Unavailable ); } + + #[tokio::test] + async fn the_newest_object_in_the_session_prefix_is_listed() { + let older = format!("{PREFIX}00000000-0000-0000-0000-000000000001.tar.gz"); + let newer = format!("{PREFIX}00000000-0000-0000-0000-000000000002.tar.gz"); + let store = FakeStore { + keys: vec![ + older, + newer.clone(), + "dev/ffffffffffffffffffffffffffffffff/x.tar.gz".into(), + format!("{PREFIX}nested/x.tar.gz"), + ], + ..Default::default() + }; + assert_eq!(latest(&store, PREFIX).await, Evidence::Listed(newer)); + let empty = FakeStore::default(); + assert_eq!(latest(&empty, PREFIX).await, Evidence::Missing); + let outage = FakeStore { + fail: true, + ..Default::default() + }; + assert_eq!(latest(&outage, PREFIX).await, Evidence::Unavailable); + } } diff --git a/crates/buzz-sandbox-reaper/src/plan.rs b/crates/buzz-sandbox-reaper/src/plan.rs index 8494b3cc61b..813959b295d 100644 --- a/crates/buzz-sandbox-reaper/src/plan.rs +++ b/crates/buzz-sandbox-reaper/src/plan.rs @@ -21,6 +21,9 @@ pub enum Evidence { Unconfigured, /// No receipt, or the receipt and S3 disagree. Missing, + /// No receipt (for example, a lost node), but this is the newest + /// checkpoint in the session's own prefix, from saves after turns. + Listed(String), /// Missing, but the session was recovered from this (still present) key, /// which the tombstone carries forward. Inherited(String), @@ -32,7 +35,7 @@ impl Evidence { /// Tombstone `checkpoint` annotation value; `None` while undecidable. fn annotation(&self) -> Option { match self { - Self::Verified(key) | Self::Inherited(key) => Some(key.clone()), + Self::Verified(key) | Self::Listed(key) | Self::Inherited(key) => Some(key.clone()), Self::Unconfigured => Some(CHECKPOINT_UNCONFIGURED.into()), Self::Missing => Some(CHECKPOINT_MISSING.into()), Self::Unavailable => None, @@ -98,7 +101,7 @@ pub fn plan( State::Replaced => tombstone(EndedReason::Replaced, evidence), State::Lost { confirmed: true } => tombstone(EndedReason::Lost, evidence), State::Completed { finished_at, .. } => match evidence { - Evidence::Missing | Evidence::Inherited(_) => { + Evidence::Missing | Evidence::Listed(_) | Evidence::Inherited(_) => { let until = finished_at.unwrap_or(hold_since) + Duration::seconds(MISSING_CHECKPOINT_HOLD_SECS); if now >= until { @@ -321,6 +324,7 @@ mod tests { prop_oneof![ "[a-z]{1,8}".prop_map(Evidence::Verified), "[a-z]{1,8}".prop_map(Evidence::Inherited), + "[a-z]{1,8}".prop_map(Evidence::Listed), Just(Evidence::Unconfigured), Just(Evidence::Missing), Just(Evidence::Unavailable), diff --git a/crates/buzz-sandbox-reaper/src/reconcile.rs b/crates/buzz-sandbox-reaper/src/reconcile.rs index f469f94c024..2114f259151 100644 --- a/crates/buzz-sandbox-reaper/src/reconcile.rs +++ b/crates/buzz-sandbox-reaper/src/reconcile.rs @@ -428,6 +428,19 @@ impl Reconciler { let own = self .own_evidence(state, sandbox, observation, config, memory) .await; + // A session that ran but left no receipt (a lost node never writes its + // termination message) falls back to its newest saved checkpoint. + let ran = !matches!( + state, + State::Binding { .. } | State::BindingIncomplete { .. } + ); + let own = match (own, annotation(sandbox, GENERATION)) { + (Evidence::Missing, Some(generation)) if ran => { + let session_prefix = format!("{}{generation}/", config.checkpoint_prefix); + evidence::latest(&self.store, &session_prefix).await + } + (own, _) => own, + }; match ( own, annotation(sandbox, RESTORE_CHECKPOINT), diff --git a/crates/buzz-sandbox-reaper/src/reconcile_tests.rs b/crates/buzz-sandbox-reaper/src/reconcile_tests.rs index 860ac2c086b..7a2af7a8c93 100644 --- a/crates/buzz-sandbox-reaper/src/reconcile_tests.rs +++ b/crates/buzz-sandbox-reaper/src/reconcile_tests.rs @@ -361,6 +361,27 @@ async fn a_lost_pod_is_tombstoned_only_after_the_grace() { ); } +/// A node lost without warning leaves no termination receipt. The session +/// still ends with the newest checkpoint the harness saved after a turn. +#[tokio::test] +async fn a_lost_session_records_its_newest_turn_checkpoint() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + s.sandboxes.insert("s".into(), sb); + } + let older = format!("dev/{GENERATION_ID}/00000000-0000-0000-0000-000000000001.tar.gz"); + let r = reconciler(&state, store_with(&[older, key()])); + let mut memory = Memory::default(); + pass(&r, &mut memory).await.unwrap(); + state.lock().unwrap().now += chrono::Duration::seconds(60); + pass(&r, &mut memory).await.unwrap(); + let annotations = &stored(&state, "s")["metadata"]["annotations"]; + assert_eq!(annotations[ENDED_REASON], "lost"); + assert_eq!(annotations[CHECKPOINT], key()); +} + #[tokio::test] async fn abandoned_bindings_are_tombstoned_without_probing_s3() { let state = fake::Shared::default(); diff --git a/docs/remote-agents.md b/docs/remote-agents.md index b7492e8ee55..292988b3643 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -1530,9 +1530,27 @@ Pod's termination message. A Pod watch keeps the message after a stop or a replacement deletes the Pod. The reaper accepts a key only if it is under `/` (prefix from the reaper configuration, generation from the Sandbox) and an S3 `ListObjectsV2` of that prefix finds -it. Otherwise it records `missing`. A compromised Pod therefore cannot point -a tombstone at the checkpoint of a different session. If S3 cannot be -reached, the reaper waits for a later pass. +it. A compromised Pod therefore cannot point a tombstone at the checkpoint of +a different session. If S3 cannot be reached, the reaper waits for a later +pass. + +**Checkpoints after turns.** With `BUZZ_ACP_CHECKPOINT_ON_STOP`, the harness +also saves in the background after a completed turn: one save at a time, at +most one per `BUZZ_ACP_CHECKPOINT_AFTER_TURN_SECS` (default 600; 0 disables). +Background saves never write a termination receipt. Only a successful final save +after draining can attest the final workspace; failed exits cannot supply that +receipt. The final save clears any stale receipt before it starts. The save on stop waits +up to 60s for a running background save, then cancels it and runs. A +background save does not stop the adapters, so another agent slot can be +mid-turn while it reads the workspace. + +A node lost without warning never writes a termination message. When the +bound Pod left no receipt, the reaper uses the newest object directly under +the session's own `/` prefix, by S3 +`LastModified`. All sessions of one developer share one IAM role, so a +compromised session of the same developer could write into that prefix; it +cannot reach another developer's prefix. With neither a receipt nor a listed +object, the reaper records `missing`. A recovered session that has no verified checkpoint of its own (for example, its restore failed) keeps the checkpoint it was recovered from. The reaper