fix(sandbox): quarantine the workspace when the final checkpoint fails

crates/buzz-acp/src/activity.rs

@@ -33,12 +33,12 @@ impl ActivityRecorder {
         let mut details = json!({});
         let kind = match event.kind.as_str() {
             "checkpoint_started" | "checkpoint_receipt_failed" => event.kind.as_str(),
-            "checkpoint_failed" => {
+            "checkpoint_failed" | "checkpoint_quarantine_failed" => {
                 details["reason"] = event.payload["reason"].as_str().into();
                 details["exit_code"] = event.payload["exit_code"].as_i64().into();
                 event.kind.as_str()
             }
-            "checkpoint_saved" => {
+            "checkpoint_saved" | "checkpoint_quarantined" => {
                 details = event.payload.clone();
                 event.kind.as_str()
             }
@@ -273,15 +273,18 @@ mod tests {
 
     #[test]
     fn checkpoint_failure_records_only_its_reason_and_exit_code() {
-        let mut recorder = ActivityRecorder::new();
-        let mut e = event(json!({"reason":"exit_status","exit_code":3,"stderr":"SECRET"}));
-        e.kind = "checkpoint_failed".into();
-        let record = recorder.record(&e).unwrap();
-        assert_eq!(record["event"], "checkpoint_failed");
-        assert_eq!(
-            record["details"],
-            json!({"reason":"exit_status","exit_code":3})
-        );
+        for kind in ["checkpoint_failed", "checkpoint_quarantine_failed"] {
+            let mut recorder = ActivityRecorder::new();
+            let mut e = event(json!({"reason":"exit_status","exit_code":3,"stderr":"SECRET"}));
+            e.kind = kind.into();
+            let record = recorder.record(&e).unwrap();
+            assert_eq!(record["event"], kind);
+            assert_eq!(
+                record["details"],
+                json!({"reason":"exit_status","exit_code":3}),
+                "{kind}"
+            );
+        }
     }
 
     #[test]

crates/buzz-acp/src/checkpoint.rs

@@ -13,6 +13,31 @@ use crate::observer::{ObserverContext, ObserverHandle};
 
 const RECEIPT_BUDGET: usize = 2048;
 
+/// Bound of one `buzz-agent-checkpoint save`.
+const SAVE_TIMEOUT: Duration = Duration::from_secs(120);
+
+/// How long a running background save may continue after the stop path
+/// starts, overlapping the drain, before the final save cancels it.
+pub(crate) const TURN_SAVE_SETTLE: Duration = Duration::from_secs(60);
+
+/// Longest quarantine that follows a failed final save.
+const QUARANTINE_TIMEOUT: Duration = Duration::from_secs(90);
+
+/// A quarantine with less time than this left before the stop deadline is
+/// skipped (reported as `checkpoint_quarantine_failed`, reason `deadline`).
+const QUARANTINE_MINIMUM: Duration = Duration::from_secs(10);
+
+/// Time from the start of the stop path by which the quarantine must end:
+/// the Sandbox Pod's 300s termination grace period
+/// (`config::TERMINATION_GRACE_SECONDS` in buzz-backend-kubernetes), less the
+/// 5s relay close that follows the checkpoint and a 5s margin. The drains
+/// before the checkpoint have no shared bound (docs/remote-agents.md
+/// §Stop and Delete), so the quarantine takes what remains of this budget
+/// rather than adding a fixed term to it.
+pub(crate) const STOP_BUDGET: Duration = Duration::from_secs(290);
+
+const _: () = assert!(TURN_SAVE_SETTLE.as_secs() + SAVE_TIMEOUT.as_secs() < STOP_BUDGET.as_secs());
+
 /// Why a stop checkpoint failed. Helper stderr is discarded because it may
 /// carry storage credentials, so these variants are the whole diagnosis.
 #[derive(Debug, thiserror::Error)]
@@ -128,22 +153,107 @@ struct TerminationReceipt<'a> {
     checkpoint: &'a str,
 }
 
-pub(crate) async fn save(observer: Option<&ObserverHandle>) -> anyhow::Result<()> {
-    save_to(observer, Some(Path::new(TERMINATION_LOG))).await
+fn helper(action: &str) -> Command {
+    let mut command = Command::new("buzz-agent-checkpoint");
+    command.arg(action);
+    command
 }
 
-async fn save_to(
+/// The final save, after every adapter has stopped. When it fails, the
+/// workspace is quarantined: stored without the secret scan under the
+/// session's `quarantine/` prefix, so the work survives the Pod's node. The
+/// quarantine ends by `stop_deadline` ([`STOP_BUDGET`] after the stop path
+/// started). The failed save is still the result, so the harness exits
+/// non-zero.
+pub(crate) async fn save(
     observer: Option<&ObserverHandle>,
+    stop_deadline: tokio::time::Instant,
+) -> anyhow::Result<()> {
+    final_save(
+        observer,
+        helper("save"),
+        helper("quarantine"),
+        Some(Path::new(TERMINATION_LOG)),
+        stop_deadline,
+    )
+    .await
+}
+
+async fn final_save(
+    observer: Option<&ObserverHandle>,
+    save: Command,
+    quarantine: Command,
     termination_log: Option<&Path>,
+    stop_deadline: tokio::time::Instant,
 ) -> anyhow::Result<()> {
-    let mut command = Command::new("buzz-agent-checkpoint");
-    command.arg("save");
-    save_with_command(observer, command, Duration::from_secs(120), termination_log)
+    let saved = save_with_command(observer, save, SAVE_TIMEOUT, termination_log).await;
+    if saved.is_err() {
+        let left = stop_deadline.saturating_duration_since(tokio::time::Instant::now());
+        if left < QUARANTINE_MINIMUM {
+            tracing::error!("no time left before the stop deadline to quarantine the workspace");
+            if let Some(observer) = observer {
+                observer.emit(
+                    "checkpoint_quarantine_failed",
+                    None,
+                    &ObserverContext::default(),
+                    serde_json::json!({"reason": "deadline"}),
+                );
+            }
+        } else {
+            quarantine_with_command(observer, quarantine, QUARANTINE_TIMEOUT.min(left)).await;
+        }
+    }
+    saved.map_err(failed)
+}
+
+/// A background save after a completed turn. Its failure only logs: the
+/// final save still runs, and quarantines the workspace if it fails too.
+async fn turn_save(observer: Option<&ObserverHandle>) -> anyhow::Result<()> {
+    save_with_command(observer, helper("save"), SAVE_TIMEOUT, None)
         .await
-        .map_err(|error| {
-            anyhow::Error::new(error)
-                .context("checkpoint failed; retain the Sandbox workspace for explicit recovery")
-        })
+        .map_err(failed)
+}
+
+fn failed(error: CheckpointError) -> anyhow::Error {
+    anyhow::Error::new(error).context("workspace checkpoint failed")
+}
+
+/// Store the workspace unscanned after a failed final save. It writes no
+/// termination receipt: the reaper finds the archive by listing the
+/// session's quarantine prefix. Reports `checkpoint_quarantined` with the
+/// receipt, or `checkpoint_quarantine_failed` with the reason.
+async fn quarantine_with_command(
+    observer: Option<&ObserverHandle>,
+    mut command: Command,
+    timeout: Duration,
+) {
+    let emit = |kind, details| {
+        if let Some(observer) = observer {
+            observer.emit(kind, None, &ObserverContext::default(), details);
+        }
+    };
+    command
+        .kill_on_drop(true)
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::null());
+    #[cfg(unix)]
+    command.process_group(0);
+    match run_checkpoint(command, timeout).await {
+        Ok(details) => {
+            tracing::warn!(
+                "final checkpoint failed; quarantined the workspace without the secret scan"
+            );
+            emit("checkpoint_quarantined", details);
+        }
+        Err(error) => {
+            tracing::error!(
+                reason = error.reason(),
+                step = error.step(),
+                "workspace quarantine failed: {error}"
+            );
+            emit("checkpoint_quarantine_failed", error.details());
+        }
+    }
 }
 
 type SaveFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
@@ -187,7 +297,7 @@ impl TurnCheckpoints {
             Box::new(move || {
                 let observer = observer.clone();
                 Box::pin(async move {
-                    if let Err(error) = save_to(observer.as_ref(), None).await {
+                    if let Err(error) = turn_save(observer.as_ref()).await {
                         tracing::warn!(%error, "turn checkpoint failed");
                     }
                 })
@@ -670,6 +780,111 @@ mod tests {
         assert!(std::fs::read_to_string(&log).unwrap().is_empty());
     }
 
+    /// A failed final save quarantines the workspace and still fails the
+    /// stop; a successful one never runs the quarantine. The quarantine
+    /// writes no termination receipt (the reaper lists its prefix instead).
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn a_failed_final_save_quarantines_the_workspace() {
+        let quarantined = serde_json::json!({"version":1,
+            "key":"dev/g/quarantine/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz",
+            "sha256":"b".repeat(64),"repositories":1,"bytes":100});
+        let receipt = valid_receipt().to_string();
+        let quarantine_receipt = quarantined.to_string();
+        let cases = [
+            (
+                format!("printf '%s' '{receipt}'"),
+                "touch \"$1\"; exit 0".to_string(),
+                true,
+                vec!["checkpoint_started", "checkpoint_saved"],
+                serde_json::Value::Null,
+            ),
+            (
+                "exit 11".to_string(),
+                format!("touch \"$1\"; printf '%s' '{quarantine_receipt}'"),
+                false,
+                vec![
+                    "checkpoint_started",
+                    "checkpoint_failed",
+                    "checkpoint_quarantined",
+                ],
+                // Only the validated receipt fields reach activity.
+                serde_json::json!({
+                    "key":"dev/g/quarantine/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz",
+                    "sha256":"b".repeat(64),"repositories":1,"bytes":100}),
+            ),
+            (
+                "exit 11".to_string(),
+                "touch \"$1\"; exit 13".to_string(),
+                false,
+                vec![
+                    "checkpoint_started",
+                    "checkpoint_failed",
+                    "checkpoint_quarantine_failed",
+                ],
+                serde_json::json!({"reason":"exit_status","exit_code":13,
+                    "step":"uploading the checkpoint"}),
+            ),
+        ];
+        for (save_script, quarantine_script, saved, kinds, last) in cases {
+            let dir = tempfile::tempdir().unwrap();
+            let log = dir.path().join("termination-log");
+            std::fs::write(&log, "").unwrap();
+            let ran = dir.path().join("quarantine-ran");
+            let observer = ObserverHandle::in_process();
+            let mut save = Command::new("sh");
+            save.args(["-c", &save_script]);
+            let mut quarantine = Command::new("sh");
+            quarantine.args([
+                "-c",
+                &quarantine_script,
+                "checkpoint-test",
+                ran.to_str().unwrap(),
+            ]);
+            let deadline = tokio::time::Instant::now() + STOP_BUDGET;
+            let result = final_save(Some(&observer), save, quarantine, Some(&log), deadline).await;
+            assert_eq!(result.is_ok(), saved, "{save_script}");
+            assert_eq!(
+                ran.exists(),
+                !saved,
+                "quarantine runs only after a failed save"
+            );
+            let events = observer.snapshot();
+            assert_eq!(
+                events.iter().map(|e| e.kind.as_str()).collect::<Vec<_>>(),
+                kinds
+            );
+            if !saved {
+                assert_eq!(events.last().unwrap().payload, last);
+                assert!(std::fs::read_to_string(&log).unwrap().is_empty());
+            }
+        }
+        // With the stop deadline (nearly) spent, the quarantine is skipped
+        // and reported rather than cut off by the kubelet's SIGKILL.
+        let dir = tempfile::tempdir().unwrap();
+        let ran = dir.path().join("quarantine-ran");
+        let observer = ObserverHandle::in_process();
+        let mut save = Command::new("sh");
+        save.args(["-c", "exit 11"]);
+        let mut quarantine = Command::new("sh");
+        quarantine.args([
+            "-c",
+            "touch \"$1\"",
+            "checkpoint-test",
+            ran.to_str().unwrap(),
+        ]);
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+        assert!(
+            final_save(Some(&observer), save, quarantine, None, deadline)
+                .await
+                .is_err()
+        );
+        assert!(!ran.exists());
+        let last = observer.snapshot().last().unwrap().clone();
+        assert_eq!(last.kind, "checkpoint_quarantine_failed");
+        assert_eq!(last.payload, serde_json::json!({"reason":"deadline"}));
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn a_turn_save_emits_a_checkpoint_without_a_termination_receipt() {

crates/buzz-acp/src/lib.rs

@@ -2909,7 +2909,12 @@ async fn tokio_main() -> Result<()> {
     // removing temporary keys. Existing work still needs a checkpoint even
     // when startup failed or was cancelled before the relay became available.
     let checkpoint_result = if checkpoint_on_stop {
-        checkpoint::save(observer.as_ref()).await
+        // Startup has already been torn down; the stop budget starts here.
+        checkpoint::save(
+            observer.as_ref(),
+            tokio::time::Instant::now() + checkpoint::STOP_BUDGET,
+        )
+        .await
     } else {
         Ok(())
     };
@@ -4436,7 +4441,8 @@ async fn run_harness(
     // A running background turn checkpoint gets 60s from the start of the
     // stop path, overlapping the drains below rather than adding to the
     // shutdown budget; past that the save on stop cancels it.
-    let turn_checkpoint_deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    let stop_started = tokio::time::Instant::now();
+    let turn_checkpoint_deadline = stop_started + checkpoint::TURN_SAVE_SETTLE;
 
     // Drain wake tasks gracefully rather than aborting: an in-flight
     // initialize_agent_pool observes the shutdown watch at its biased per-slot
@@ -4555,7 +4561,7 @@ async fn run_harness(
     // before starting the final save, using the deadline captured at drain.
     turn_checkpoints.settle(turn_checkpoint_deadline).await;
     let checkpoint_result = if config.checkpoint_on_stop {
-        checkpoint::save(observer.as_ref()).await
+        checkpoint::save(observer.as_ref(), stop_started + checkpoint::STOP_BUDGET).await
     } else {
         Ok(())
     };

crates/buzz-acp/tests/checkpoint_stop.rs

@@ -23,7 +23,13 @@ exec cat >/dev/null
 "#;
 
 // Records what had already happened when the harness started the checkpoint.
+// A quarantine (after a failed save) records only that it ran.
 const HELPER: &str = r#"#!/bin/sh
+if [ "$1" = quarantine ]; then
+  echo quarantine >> "$PROBE_DIR/checkpoint-runs"
+  printf '{"version":1,"key":"developer/session/quarantine/id.tar.gz","sha256":"%064d","repositories":0,"bytes":0}' 0
+  exit 0
+fi
 test "$1" = save || exit 64
 echo save >> "$PROBE_DIR/checkpoint-runs"
 {
@@ -85,6 +91,11 @@ async fn failed_checkpoint_fails_either_shutdown_path() {
         assert!(stopped.checkpoint.is_some(), "{}", stopped.logs);
         assert!(!stopped.status.success(), "{}", stopped.logs);
         assert!(stopped.logs.contains("exit_status"), "{}", stopped.logs);
+        assert!(
+            stopped.logs.contains("checkpoint_quarantined"),
+            "{}",
+            stopped.logs
+        );
     }
 }
 
@@ -210,10 +221,11 @@ async fn stop(path: Stop, fail: bool) -> Stopped {
         .unwrap_or_else(|_| panic!("harness did not exit: {}", logs(dir)))
         .unwrap();
     relay.abort();
+    // The stop hook runs exactly once; only a failed save is quarantined.
     assert_eq!(
         std::fs::read_to_string(dir.join("checkpoint-runs")).unwrap_or_default(),
-        "save\n",
-        "the stop hook must run exactly once: {}",
+        if fail { "save\nquarantine\n" } else { "save\n" },
+        "{}",
         logs(dir)
     );
     Stopped {

crates/buzz-backend-kubernetes/src/config.rs

@@ -57,7 +57,10 @@ pub const WORKSPACE_PATH: &str = "/home/agent";
 
 /// Declared shutdown budget, including the 120s checkpoint timeout. The up to
 /// 60s wait for a running background turn checkpoint counts from the start of
-/// the stop path, so it overlaps the drains instead of adding to them.
+/// the stop path, so it overlaps the drains instead of adding to them. After a
+/// failed checkpoint, `buzz-acp` quarantines the workspace only within what
+/// remains of a 290s deadline from the start of the stop path
+/// (`checkpoint::STOP_BUDGET`), so the quarantine adds no term here.
 /// Child reaping is variable; this is operational margin, not a proven bound
 /// on the harness's whole shutdown path (§K8s Grace).
 pub const TERMINATION_GRACE_SECONDS: i64 = 300;