diff --git a/crates/buzz-acp/src/activity.rs b/crates/buzz-acp/src/activity.rs index cbe1e2e9457..c63047ad524 100644 --- a/crates/buzz-acp/src/activity.rs +++ b/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] diff --git a/crates/buzz-acp/src/checkpoint.rs b/crates/buzz-acp/src/checkpoint.rs index b593b57c8c4..457553c26eb 100644 --- a/crates/buzz-acp/src/checkpoint.rs +++ b/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 + 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::>(), + 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() { diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 368e72e7f00..876969f3e9a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/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(()) }; diff --git a/crates/buzz-acp/tests/checkpoint_stop.rs b/crates/buzz-acp/tests/checkpoint_stop.rs index a958d1c2a37..ed0cd76fea1 100644 --- a/crates/buzz-acp/tests/checkpoint_stop.rs +++ b/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 { diff --git a/crates/buzz-backend-kubernetes/src/config.rs b/crates/buzz-backend-kubernetes/src/config.rs index 98058495109..3aefe12ba8c 100644 --- a/crates/buzz-backend-kubernetes/src/config.rs +++ b/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; diff --git a/crates/buzz-backend-kubernetes/src/lifecycle.rs b/crates/buzz-backend-kubernetes/src/lifecycle.rs index c9b2407737b..14670e0e253 100644 --- a/crates/buzz-backend-kubernetes/src/lifecycle.rs +++ b/crates/buzz-backend-kubernetes/src/lifecycle.rs @@ -49,8 +49,9 @@ pub const ENDED_AT: &str = "buzz.block.xyz/ended-at"; /// `` | `missing` | `unconfigured`. pub const CHECKPOINT: &str = "buzz.block.xyz/checkpoint"; /// Where the tombstone's [`CHECKPOINT`] key came from: -/// [`CHECKPOINT_SOURCE_RECEIPT`] | [`CHECKPOINT_SOURCE_LISTED`] | -/// [`CHECKPOINT_SOURCE_INHERITED`]. Absent when [`CHECKPOINT`] is not a key. +/// [`CHECKPOINT_SOURCE_RECEIPT`] | [`CHECKPOINT_SOURCE_QUARANTINE`] | +/// [`CHECKPOINT_SOURCE_LISTED`] | [`CHECKPOINT_SOURCE_INHERITED`]. Absent when +/// [`CHECKPOINT`] is not a key. pub const CHECKPOINT_SOURCE: &str = "buzz.block.xyz/checkpoint-source"; /// Generation of the tombstone a recovered session replaced. pub const RECOVERED_FROM: &str = "buzz.block.xyz/recovered-from"; @@ -71,6 +72,15 @@ pub const CHECKPOINT_SOURCE_RECEIPT: &str = "receipt"; pub const CHECKPOINT_SOURCE_LISTED: &str = "listed"; /// [`CHECKPOINT_SOURCE`]: the checkpoint a recovered session was restored from. pub const CHECKPOINT_SOURCE_INHERITED: &str = "inherited"; +/// [`CHECKPOINT_SOURCE`]: the archive the harness stored under the session's +/// [`QUARANTINE_DIR`] after its final verified save failed. It holds the final +/// workspace without the secret scan; excluded credential paths stay out. +pub const CHECKPOINT_SOURCE_QUARANTINE: &str = "quarantine"; +/// Subdirectory of a session prefix (`/`) that +/// holds quarantine archives. Its objects are not session keys +/// ([`is_session_key`]), so a listing of the session prefix never picks one. +/// Matches `QUARANTINE` in `voltage/images/buzz-agent/checkpoint.py`. +pub const QUARANTINE_DIR: &str = "quarantine/"; /// [`CHECKPOINTING`] value written when checkpoint-on-stop is configured. pub const CHECKPOINTING_ENABLED: &str = "enabled"; /// [`CHECKPOINTING`] value written when checkpoint-on-stop is not configured. @@ -510,6 +520,14 @@ pub fn is_session_key(key: &str, session_prefix: &str) -> bool { .all(|b| b.is_ascii_alphanumeric() || b"-/._".contains(&b)) } +/// Is `key` one checkpoint object of the session: a session key +/// ([`is_session_key`]) or a quarantine archive directly under +/// `quarantine/`? +pub fn is_restorable_key(key: &str, session_prefix: &str) -> bool { + is_session_key(key, session_prefix) + || is_session_key(key, &format!("{session_prefix}{QUARANTINE_DIR}")) +} + /// Is a tombstone `checkpoint` value an actual key? pub fn is_checkpoint_key(value: &str) -> bool { !value.is_empty() && value != CHECKPOINT_MISSING && value != CHECKPOINT_UNCONFIGURED diff --git a/crates/buzz-sandbox-reaper/src/evidence.rs b/crates/buzz-sandbox-reaper/src/evidence.rs index 4eed891e909..a4ca4278053 100644 --- a/crates/buzz-sandbox-reaper/src/evidence.rs +++ b/crates/buzz-sandbox-reaper/src/evidence.rs @@ -8,6 +8,9 @@ //! //! - [`from_receipt`]: the harness's termination-message receipt. The receipt //! alone is written by the (untrusted) Pod, so S3 must also list it. +//! - [`quarantined`]: the newest archive under the session's `quarantine/` +//! prefix, which the harness writes without the secret scan after its +//! final verified save fails. It holds the final workspace. //! - [`latest`]: the newest object in the session prefix, for a session //! without a verified receipt (the reconcile engine decides when). It may be //! a turn checkpoint older than the final workspace. @@ -22,7 +25,9 @@ use std::future::Future; use std::time::Duration; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; -use buzz_backend_kubernetes::lifecycle::{is_session_key, receipt_key}; +use buzz_backend_kubernetes::lifecycle::{ + is_restorable_key, is_session_key, receipt_key, QUARANTINE_DIR, +}; use crate::plan::Evidence; @@ -254,9 +259,23 @@ pub async fn latest(store: &S, session_prefix: &str) -> Evid } } +/// Evidence for a session whose final verified save failed: the newest +/// quarantine archive directly under `quarantine/`, as +/// [`Evidence::Quarantined`]. Nothing listed is [`Evidence::Missing`]; a +/// failed listing is [`Evidence::Unavailable`], never `Missing`. +pub async fn quarantined(store: &S, session_prefix: &str) -> Evidence { + let prefix = format!("{session_prefix}{QUARANTINE_DIR}"); + match store.newest(&prefix).await { + Ok(Some(key)) if is_session_key(&key, &prefix) => Evidence::Quarantined(key), + Ok(_) => Evidence::Missing, + Err(error) => unavailable(&error), + } +} + /// Evidence for a recovered session that left no verified checkpoint of its /// own: the checkpoint it was restored from, if that object still exists -/// directly under `/` for a valid generation. +/// directly under `/` (or its quarantine +/// subdirectory, after a recovery from a quarantine) for a valid generation. /// The generation can be an older ancestor than `recovered-from` after /// repeated failed recovery. `restore` comes from the provider-written /// Sandbox annotation, never the Pod, so carrying it forward keeps the @@ -271,7 +290,7 @@ pub async fn inherited( .and_then(|rest| rest.split_once('/')) .and_then(|(generation, _)| session_prefix(developer_prefix, generation)); match prefix { - Some(prefix) if is_session_key(restore, &prefix) => { + Some(prefix) if is_restorable_key(restore, &prefix) => { probe(store, restore.to_string(), Evidence::Inherited).await } _ => Evidence::Missing, @@ -425,6 +444,41 @@ pub mod tests { ); } + /// Quarantine archives live one level below the session prefix: a + /// listing of the session never picks one, and the quarantine probe + /// picks only its newest direct child. + #[tokio::test] + async fn quarantine_archives_are_listed_apart_from_session_checkpoints() { + let turn = format!("{PREFIX}00000000-0000-0000-0000-000000000001.tar.gz"); + let older = format!("{PREFIX}quarantine/00000000-0000-0000-0000-000000000002.tar.gz"); + let newer = format!("{PREFIX}quarantine/00000000-0000-0000-0000-000000000003.tar.gz"); + let store = FakeStore { + keys: vec![ + turn.clone(), + older, + newer.clone(), + format!("{PREFIX}quarantine/nested/x.tar.gz"), + "dev/ffffffffffffffffffffffffffffffff/quarantine/x.tar.gz".into(), + ], + ..Default::default() + }; + assert_eq!(latest(&store, PREFIX).await, Evidence::Listed(turn.clone())); + assert_eq!( + quarantined(&store, PREFIX).await, + Evidence::Quarantined(newer) + ); + let none = FakeStore { + keys: vec![turn], + ..Default::default() + }; + assert_eq!(quarantined(&none, PREFIX).await, Evidence::Missing); + let outage = FakeStore { + fail: true, + ..Default::default() + }; + assert_eq!(quarantined(&outage, PREFIX).await, Evidence::Unavailable); + } + #[test] fn session_prefixes_require_a_provider_generation() { let generation = "0123456789abcdef0123456789abcdef"; @@ -453,6 +507,16 @@ pub mod tests { inherited(&store, &key, "dev/").await, Evidence::Inherited(key.clone()) ); + // A recovery from a quarantine archive carries that archive forward. + let quarantine = format!("{PREFIX}quarantine/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"); + let quarantined_store = FakeStore { + keys: vec![quarantine.clone()], + ..Default::default() + }; + assert_eq!( + inherited(&quarantined_store, &quarantine, "dev/").await, + Evidence::Inherited(quarantine) + ); // Expired or deleted since the recovery. let gone = format!("{PREFIX}00000000-0000-0000-0000-000000000000.tar.gz"); assert_eq!(inherited(&store, &gone, "dev/").await, Evidence::Missing); @@ -466,6 +530,7 @@ pub mod tests { for bad in [ "dev/NOT-A-GENERATION/x.tar.gz", &format!("{PREFIX}a/x.tar.gz"), + &format!("{PREFIX}quarantine/a/x.tar.gz"), ] { let store = FakeStore { keys: vec![bad.to_string()], diff --git a/crates/buzz-sandbox-reaper/src/plan.rs b/crates/buzz-sandbox-reaper/src/plan.rs index 54100157100..01ff1b19a60 100644 --- a/crates/buzz-sandbox-reaper/src/plan.rs +++ b/crates/buzz-sandbox-reaper/src/plan.rs @@ -8,8 +8,8 @@ use buzz_backend_kubernetes::lifecycle::{ EndedReason, Ignored, State, CHECKPOINT_MISSING, CHECKPOINT_SOURCE_INHERITED, - CHECKPOINT_SOURCE_LISTED, CHECKPOINT_SOURCE_RECEIPT, CHECKPOINT_UNCONFIGURED, - MISSING_CHECKPOINT_HOLD_SECS, + CHECKPOINT_SOURCE_LISTED, CHECKPOINT_SOURCE_QUARANTINE, CHECKPOINT_SOURCE_RECEIPT, + CHECKPOINT_UNCONFIGURED, MISSING_CHECKPOINT_HOLD_SECS, }; use chrono::{DateTime, Duration, Utc}; @@ -22,6 +22,12 @@ pub enum Evidence { Unconfigured, /// No receipt, or the receipt and S3 disagree. Missing, + /// The final verified save failed, and the harness stored the final + /// workspace without the secret scan under the session's quarantine + /// prefix. S3 lists this newest archive there. It is durable, so a + /// Completed session records it at once instead of holding a workspace + /// that only lives on the Pod's node. + Quarantined(String), /// No verified receipt (none observed, or a failed final save), but S3 /// lists this newest object in the session's own prefix. Not attested /// by the harness and possibly an older turn checkpoint, so a Completed @@ -40,6 +46,7 @@ impl Evidence { fn annotation(&self) -> Option<(String, Option<&'static str>)> { match self { Self::Verified(key) => Some((key.clone(), Some(CHECKPOINT_SOURCE_RECEIPT))), + Self::Quarantined(key) => Some((key.clone(), Some(CHECKPOINT_SOURCE_QUARANTINE))), Self::Listed(key) => Some((key.clone(), Some(CHECKPOINT_SOURCE_LISTED))), Self::Inherited(key) => Some((key.clone(), Some(CHECKPOINT_SOURCE_INHERITED))), Self::Unconfigured => Some((CHECKPOINT_UNCONFIGURED.into(), None)), @@ -56,7 +63,8 @@ pub enum Action { None, /// Leave it alone and say why (Event). Ignore(Ignored), - /// Completed without a checkpoint: hold until `until` (Event + metric). + /// Completed without a durable copy of its final workspace: hold until + /// `until` (Event + metric). Hold { until: DateTime }, /// One CAS merge patch: Suspended + lifecycle annotations. Tombstone { @@ -124,7 +132,13 @@ pub fn plan( Action::Hold { until } } } - _ => tombstone(EndedReason::Completed, evidence), + // A quarantine archive is a durable copy of the final workspace, + // so nothing is gained by holding the Pod. `Unavailable` plans + // nothing (see `tombstone`). + Evidence::Verified(_) + | Evidence::Quarantined(_) + | Evidence::Unconfigured + | Evidence::Unavailable => tombstone(EndedReason::Completed, evidence), }, State::Ended { expired: true } => Action::Delete, State::Binding { abandoned: false } @@ -208,6 +222,22 @@ mod tests { source: Some(CHECKPOINT_SOURCE_RECEIPT), } ); + // A failed final save that left a quarantine archive does not hold: + // the archive is the durable copy of the final workspace, while the + // Pod's emptyDir dies with a spot interruption or a scale-down. + assert_eq!( + plan( + &completed, + &Evidence::Quarantined("dev/g/quarantine/x.tar.gz".into()), + t(1), + t(1) + ), + Action::Tombstone { + reason: EndedReason::Completed, + checkpoint: "dev/g/quarantine/x.tar.gz".into(), + source: Some(CHECKPOINT_SOURCE_QUARANTINE), + } + ); assert_eq!( plan(&completed, &Evidence::Unconfigured, t(1), t(1)), Action::Tombstone { @@ -349,6 +379,7 @@ mod tests { fn any_evidence() -> impl Strategy { prop_oneof![ "[a-z]{1,8}".prop_map(Evidence::Verified), + "[a-z]{1,8}".prop_map(Evidence::Quarantined), "[a-z]{1,8}".prop_map(Evidence::Inherited), "[a-z]{1,8}".prop_map(Evidence::Listed), Just(Evidence::Unconfigured), @@ -385,11 +416,18 @@ mod tests { *source == Some(CHECKPOINT_SOURCE_RECEIPT), matches!(evidence, Evidence::Verified(_)) ); + prop_assert_eq!( + *source == Some(CHECKPOINT_SOURCE_QUARANTINE), + matches!(evidence, Evidence::Quarantined(_)) + ); prop_assert_eq!( source.is_some(), matches!( evidence, - Evidence::Verified(_) | Evidence::Listed(_) | Evidence::Inherited(_) + Evidence::Verified(_) + | Evidence::Quarantined(_) + | Evidence::Listed(_) + | Evidence::Inherited(_) ) ); } diff --git a/crates/buzz-sandbox-reaper/src/reconcile.rs b/crates/buzz-sandbox-reaper/src/reconcile.rs index 53013ccefdd..b26d357b7c1 100644 --- a/crates/buzz-sandbox-reaper/src/reconcile.rs +++ b/crates/buzz-sandbox-reaper/src/reconcile.rs @@ -495,8 +495,11 @@ impl Reconciler { /// otherwise from what the watch captured. The session prefix comes from /// config and the Sandbox's (validated) generation, never the Pod. /// - /// Without a verified receipt, the newest object S3 lists in the session - /// prefix stands in for it ([`evidence::latest`], recorded as `listed`). + /// Without a verified receipt, a quarantine archive stands in for it + /// ([`evidence::quarantined`], recorded as `quarantine`): the harness + /// writes one only after its final verified save failed, so it holds the + /// final workspace. Otherwise the newest object S3 lists in the session + /// prefix does ([`evidence::latest`], recorded as `listed`). /// The receipt may be unknown (a reaper restart or a watch gap lost it), /// never written (a node lost without warning), or invalid (a failed final /// save exits non-zero, so only the harness's background turn saves @@ -534,7 +537,10 @@ impl Reconciler { _ => memory.receipts.get(initial).map(String::as_str), }; match evidence::from_receipt(&self.store, message, &session_prefix).await { - Evidence::Missing => evidence::latest(&self.store, &session_prefix).await, + Evidence::Missing => match evidence::quarantined(&self.store, &session_prefix).await { + Evidence::Missing => evidence::latest(&self.store, &session_prefix).await, + quarantined_or_unavailable => quarantined_or_unavailable, + }, verified_or_unavailable => verified_or_unavailable, } } diff --git a/crates/buzz-sandbox-reaper/src/reconcile_tests.rs b/crates/buzz-sandbox-reaper/src/reconcile_tests.rs index db105d901a4..3af49c3f27b 100644 --- a/crates/buzz-sandbox-reaper/src/reconcile_tests.rs +++ b/crates/buzz-sandbox-reaper/src/reconcile_tests.rs @@ -289,11 +289,49 @@ async fn a_recovered_session_prefers_its_own_listed_checkpoint_to_the_inherited_ assert_eq!(annotations[CHECKPOINT_SOURCE], "listed"); assert_eq!( r.store.probes.lock().unwrap().as_slice(), - [format!("dev/{GENERATION_ID}/")], + [ + format!("dev/{GENERATION_ID}/quarantine/"), + format!("dev/{GENERATION_ID}/"), + ], "the inherited key is not consulted" ); } +/// The staging incident of 2026-11-09: the final save failed, the harness +/// quarantined the workspace, and its spot node was reclaimed minutes later. +/// The session ends at once on its quarantine archive, which outranks an +/// older turn checkpoint, instead of holding a workspace that lives only on +/// the Pod's node. Without an archive the 24-hour hold still applies (see +/// `a_completed_session_without_a_checkpoint_is_held_for_24_hours`). +#[tokio::test] +async fn a_failed_final_save_ends_at_once_on_its_quarantine_archive() { + let quarantine = + format!("dev/{GENERATION_ID}/quarantine/00000000-0000-0000-0000-000000000009.tar.gz"); + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + let mut failed = terminated(None); + failed["terminated"]["exitCode"] = 1.into(); + let p = pod(&s, "s", "pod-1", 600, failed); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, store_with(&[key(), quarantine.clone()])); + let summary = pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!((summary.held, summary.tombstoned), (0, 1)); + let annotations = &stored(&state, "s")["metadata"]["annotations"]; + assert_eq!(annotations[ENDED_REASON], "completed"); + assert_eq!(annotations[CHECKPOINT], quarantine); + assert_eq!(annotations[CHECKPOINT_SOURCE], "quarantine"); + let sent = events(&state); + assert_eq!(sent.len(), 1, "no CheckpointMissing hold"); + assert_eq!( + sent[0]["message"], + format!("session ended (completed); checkpoint: {quarantine} (source: quarantine)") + ); +} + /// `stop` suspends the Sandbox; the controller deletes the Pod after the /// harness wrote its receipt. The watch captured it, so the tombstone /// records the checkpoint even though the Pod is gone. @@ -933,7 +971,9 @@ async fn failed_final_save_does_not_verify_an_earlier_turn_receipt() { assert_eq!( r.store.probes.lock().unwrap().as_slice(), [ + format!("dev/{GENERATION_ID}/quarantine/"), format!("dev/{GENERATION_ID}/"), + format!("dev/{GENERATION_ID}/quarantine/"), format!("dev/{GENERATION_ID}/") ], "the receipt is never probed as if it were attested" @@ -997,7 +1037,10 @@ async fn a_restarted_reaper_lists_the_checkpoint_of_a_deleted_pod() { assert_eq!(annotations[CHECKPOINT_SOURCE], "listed"); assert_eq!( r.store.probes.lock().unwrap().as_slice(), - [format!("dev/{GENERATION_ID}/")] + [ + format!("dev/{GENERATION_ID}/quarantine/"), + format!("dev/{GENERATION_ID}/") + ] ); // S3 down: no tombstone this pass, rather than a false `missing`. diff --git a/docs/remote-agents.md b/docs/remote-agents.md index bdea3330758..3af1b0e8395 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -1487,10 +1487,10 @@ Deleting a Sandbox removes its Pod and any workspace that remains. Recover or ch Sandbox ingress accepts only its owner's signed messages for the assigned channel thread. The provider enables JSONL activity and disables autonomous heartbeat work. -The agent environment must set `BUZZ_CHECKPOINT_BUCKET` and `BUZZ_CHECKPOINT_PREFIX`; the deployment's launch environment supplies them. Shutdown then invokes `buzz-agent-checkpoint save` after sessions drain, and a failed save makes the harness exit non-zero. +The agent environment must set `BUZZ_CHECKPOINT_BUCKET` and `BUZZ_CHECKPOINT_PREFIX`; the deployment's launch environment supplies them. Shutdown then invokes `buzz-agent-checkpoint save` after sessions drain, and a failed save makes the harness exit non-zero. A failed final save is followed by `buzz-agent-checkpoint quarantine`, which stores the workspace without the secret scan (see **Quarantine** under §Sandbox lifecycle). Deploy refuses Sandbox mode when either value is missing. To launch without a stop checkpoint, for example in a disposable test cluster, set `BUZZ_SANDBOX_CHECKPOINT=disabled`. Such a workspace is lost when its process exits. -Sandbox Pods use the same `terminationGracePeriodSeconds` as bare Pods (300s), which covers the drain and the 120s checkpoint. +Sandbox Pods use the same `terminationGracePeriodSeconds` as bare Pods (300s), which covers the drain and the 120s checkpoint. The quarantine after a failed checkpoint gets at most 90s and must end within 290s of the stop's start. When the drains leave less than 10s, it is skipped (`checkpoint_quarantine_failed`, reason `deadline`) and the session holds as before. The image supplies that helper and account setup. Infrastructure supplies storage, resource quotas, and log collection. Use channel thread mentions to launch these workspaces. A generic Start action has no thread binding and is refused. @@ -1521,7 +1521,7 @@ keys (`buzz-backend-kubernetes` library, `lifecycle.rs`): | `buzz.block.xyz/lifecycle` | `ending` (a stop is draining), `ended` (a tombstone) | | `buzz.block.xyz/ended-reason` | `stopped`, `completed`, `replaced`, `lost`, `binding-failed`, `bind-abandoned` | | `buzz.block.xyz/ended-at` | RFC 3339 apiserver time at which the tombstone was written | -| `buzz.block.xyz/checkpoint` | a verified S3 key, `missing`, or `unconfigured` | +| `buzz.block.xyz/checkpoint` | an S3 key that the reaper confirmed exists (see `checkpoint-source`), `missing`, or `unconfigured` | | `buzz.block.xyz/recovered-from` | the generation of the tombstone that a recovered session replaced | The reaper classifies each Sandbox on every pass. All timers use the @@ -1538,7 +1538,7 @@ starts the hold again; this only delays the tombstone. | Active | bound to this Pod, container running | none | | Draining | Pod deleting, or `lifecycle=ending` with the Pod present | none | | Ending | `lifecycle=ending`, Pod gone | tombstone with the recorded reason (`stopped`) | -| Completed | bound to this Pod, container terminated | checkpoint verified or unconfigured: tombstone `completed`. No verified receipt: Warning Event, hold 24h, then tombstone with the `listed` or `inherited` key, or `checkpoint=missing` | +| Completed | bound to this Pod, container terminated | checkpoint verified, quarantined, or unconfigured: tombstone `completed`. No verified receipt and no quarantine: Warning Event, hold 24h, then tombstone with the `listed` or `inherited` key, or `checkpoint=missing` | | Replaced | Pod UID differs from the binding | tombstone `replaced` | | Lost | bound, no Pod for 60s or more | tombstone `lost` | | Ended | Suspended and `lifecycle=ended` | delete the Sandbox 30 days after `ended-at` (the checkpoint object expiry); before that, none | @@ -1571,10 +1571,11 @@ records `missing` because S3 failed. `buzz_sandbox_reaper_s3_errors_total{kind}` counts these failures by kind: `timeout` or `other` for an outage, `access_denied` or `not_found_bucket` for a misconfigured role or bucket. -Without a verified receipt, the reaper records the newest object directly -under the session's own `/` prefix, by S3 -`LastModified` (the greater key breaks a tie), and records `missing` only when -that listing is empty. A session can lack a verified receipt for these +Without a verified receipt, the reaper first lists the session's +`quarantine/` prefix (see **Quarantine**). Without a quarantine archive, it +records the newest object directly under the session's own +`/` prefix, by S3 `LastModified` (the greater +key breaks a tie), and records `missing` only when both listings are empty. A session can lack a verified receipt for these reasons: - The reaper keeps observed receipts in memory only. A reaper restart, or a @@ -1597,13 +1598,50 @@ repeats it. The annotation is absent when `checkpoint` is `missing` or | `checkpoint-source` | Key | What it holds | | --- | --- | --- | | `receipt` | the final save's receipt, confirmed by S3 | the final workspace | +| `quarantine` | the newest object directly under `/quarantine/` | the final workspace, not secret-scanned | | `listed` | the newest object in the session's own prefix | possibly an older background save: work after it is not included | | `inherited` | the key the recovered session was restored from | the workspace at the earlier tombstone | -A session's own key (`receipt`, then `listed`) has precedence over the -inherited one. The image restores the inherited checkpoint before the harness +A session's own key (`receipt`, then `quarantine`, then `listed`) has +precedence over the inherited one. The image restores the inherited checkpoint before the harness starts, so a background save of a recovered session already contains that work. +**Quarantine.** When the final save fails (the secret scan refuses the +workspace, the helper times out, or anything else), the harness runs +`buzz-agent-checkpoint quarantine` with a 90s limit. It packs the workspace +like a save but without the secret scan or bundle validation, and uploads it +to `/quarantine/.tar.gz`. It drops no +repository: one whose HEAD has no commit keeps its branches and files. It still +fails on a special file, an external symlink, or the size and file budgets; +the session then holds as before. Excluded paths +(private runtime files, `.env*`, and named credential files) stay out; a +credential the agent wrote into other files does not. The harness still exits +non-zero and writes no receipt. Quarantine keys contain one more `/` than +session keys, so the session listing never picks one; the reaper lists the +quarantine prefix separately, before the session prefix. A Completed session +with a quarantine archive is tombstoned at once: the 24-hour hold exists to +keep the Pod's workspace for recovery, but an `emptyDir` lives only as long as +its node, and spot interruptions and autoscaler scale-down (which ignores +terminated Pods) remove it within minutes. Only a session with neither a +receipt nor a quarantine still holds. + +A quarantine key is trusted as far as a `listed` key: every session of the +developer shares one IAM role, so a compromised session of that developer +could plant an object under another of their sessions' `quarantine/` prefix. +Such an object would outrank `listed`, end that session without the hold, and +restore without the scan; it cannot reach another developer's prefix. + +A session recovered from a quarantine still contains what the scan refused, +so its own saves keep failing until the agent or developer removes it. Each +later stop then quarantines again, but turn checkpoints add nothing, and a +node lost without SIGTERM loses the work since the recovery. + +The quarantine prefix needs no infrastructure change (voltage-terraform +`v2/environments/staging-agent`): the Sandbox role may put and get any object +under `/`, the reaper may list the bucket, and the bucket's 30-day +expiration rule has no filter, so quarantine archives expire with the +session's other checkpoints. + **Checkpoints after turns.** With `BUZZ_ACP_CHECKPOINT_ON_STOP`, the harness also saves in the background after completed turns, without a termination receipt. A save starts only at an idle point: after a turn's result, when no @@ -1648,7 +1686,9 @@ click only; the desktop never stores it: refused with `no_checkpoint` when the tombstone has no checkpoint key, and with `no_ended_session` when the thread has no Sandbox. With `checkpoint-source=listed`, that key can be a background save older than - the session's last work (see **Checkpoints after turns**). + the session's last work (see **Checkpoints after turns**). With + `checkpoint-source=quarantine`, it restores the unscanned final workspace; + the image skips its restore-time secret scan for quarantine keys only. - `{"mode":"fresh"}` starts with an empty workspace. It is also accepted when the thread has no Sandbox. diff --git a/voltage/images/buzz-agent/README.md b/voltage/images/buzz-agent/README.md index e6ae73c858c..59e29b3747c 100644 --- a/voltage/images/buzz-agent/README.md +++ b/voltage/images/buzz-agent/README.md @@ -93,7 +93,15 @@ For older full snapshots without boundary metadata, restore reconstructs missing Missing trees, missing blobs, and other integrity failures stop recovery and remove the staged destination. Full snapshots retain manifest version 1. Older images restore complete histories, but cannot restore shallow bundles. Publish and promote the updated agent image before recovering a session with shallow repositories. -A failed save retains the Sandbox for recovery; do not delete its workspace. +A failed final save is quarantined (below). Only if the quarantine also fails does the reaper hold the Sandbox for 24 hours; do not delete its workspace then. + +**Quarantine.** When the final save on stop fails, the harness runs `buzz-agent-checkpoint quarantine` (90-second limit). +It packs the workspace like a save, but runs no secret scan and validates no bundle, and stores it as `/quarantine/.tar.gz`. +Excluded paths (private runtime files, `.env*`, and the credential names in `PRIVATE_NAMES`) stay out. A credential the agent wrote into other files is kept. +A repository without a usable origin is saved whole. One whose HEAD has no commit (a new `git init`, an orphan checkout) keeps its other branches and every file, unstaged. +Restoring a repository without a usable origin clears the origin URL but keeps its remote-tracking refs. +The harness still exits non-zero, writes no termination receipt, and reports `checkpoint_quarantined` or `checkpoint_quarantine_failed` in activity. +The Pod's `emptyDir` lives only as long as its node: spot interruptions and autoscaler scale-down remove it within minutes. The quarantine archive is the copy that survives. A checkpoint holds at most 256 MiB of source and 50,000 files. A repository whose HTTPS `origin` shares a branch tip with it, as `git ls-remote` reports at save time, is saved against origin: only commits beyond origin's branches, changed and untracked files, deletions, local branches, and tags origin lacks (annotated tag objects by content). The save asks every origin at once, using the normalized URL and global Git configuration that restore uses, with one 20-second deadline for all of them. @@ -107,6 +115,7 @@ Origin-based snapshots do not save Git notes (`refs/notes`) or stash entries oth Explicit checkpoint recovery sets `BUZZ_CHECKPOINT_RESTORE_KEY` in the launch Secret; it is empty otherwise. After the developer profile bootstrap and before the harness starts, the entrypoint runs `buzz-agent-checkpoint restore-once`. The key must be under the developer's `BUZZ_CHECKPOINT_PREFIX`. A failed restore stops startup; the harness never starts with an empty workspace. +A key under a session's `quarantine/` prefix restores without the restore-time secret scan, which would refuse it again. The key decides this, not the archive's manifest. The restore moves repositories into `/home/agent` without overwriting existing paths, then writes `.config/buzz-agent/checkpoint-restored.json`. A container restart with that marker skips the restore and keeps newer edits. After a successful save, the harness writes `{"version":1,"checkpoint":""}` to `/dev/termination-log` for the reaper. @@ -167,7 +176,7 @@ The manager sets `BUZZ_DEVELOPER_ID`, `BUZZ_DEVELOPER_PROFILE_ARN`, `BUZZ_CHECKP The bootstrap checks the profile against Buzz's verified owner, then writes Git identity, GitHub, via, the agent signing key, and the vault's Depot credential to private workspace files. Missing profiles, owner mismatches, and unsafe paths refuse startup. A new enrollment takes effect for new sessions; running sessions keep their credentials, so revoke upstream tokens when required. -Credentials are excluded from checkpoints. The image test uses a Secrets Manager emulator. +Credential files are excluded from checkpoints, including quarantines. The image test uses a Secrets Manager emulator. `models` accepts these optional credentials: diff --git a/voltage/images/buzz-agent/checkpoint.py b/voltage/images/buzz-agent/checkpoint.py index a358a572fad..1d1e76023fb 100644 --- a/voltage/images/buzz-agent/checkpoint.py +++ b/voltage/images/buzz-agent/checkpoint.py @@ -21,6 +21,8 @@ from urllib.parse import urlsplit, urlunsplit MAX_BYTES = 256 * 1024 * 1024 +# Session subdirectory for unscanned archives (lifecycle::QUARANTINE_DIR). +QUARANTINE = 'quarantine/' MAX_FILES = 50000 # All origins are asked at once; this bounds the whole save's network wait. ORIGIN_SECONDS = 20 @@ -318,7 +320,33 @@ def local_refs(repo, origin_refs): return refs, tags -def pack(workspace, output): +def has_commit(repo): + return subprocess.run(['git', '-c', 'core.hooksPath=/dev/null', '-C', str(repo), + 'rev-parse', '--verify', '--quiet', 'HEAD'], + capture_output=True, timeout=60).returncode == 0 + + +def quarantine_origin(repo): + """The repository's origin for a quarantine, or '' where it has none usable. + + A verified save refuses such a repository; a quarantine keeps its history + in a self-contained bundle instead.""" + try: + return public_origin(repo) + except ValueError: + return '' + + +def pack(workspace, output, *, scanned=True): + """Pack the workspace's repositories into a checkpoint archive. + + With `scanned=False` (a quarantine, after a verified save failed) nothing + is scanned for secrets and no bundle is integrity-checked: the archive + holds the agent's work as found. Excluded paths stay excluded, so private + runtime files and credential files are never packed. A repository with no + usable origin is bundled whole. One whose HEAD has no commit (a new + `git init`, an orphan checkout) keeps its branches and every file as + base `unborn`, so a quarantine never drops a repository.""" workspace = workspace.resolve(strict=True) manifest = {'version': 1, 'repositories': []} total = 0 @@ -331,17 +359,21 @@ def pack(workspace, output): repos = repositories(workspace) if any(repo == workspace for repo in repos): raise ValueError('Use a workspace directory containing repositories') - origins = [public_origin(repo) for repo in repos] + origins = [public_origin(repo) if scanned else quarantine_origin(repo) for repo in repos] listings = origin_listings(origins) for index, (repo, origin, listing) in enumerate(zip(repos, origins, listings)): relative = repo.relative_to(workspace).as_posix() tips, origin_refs = origin_base(repo, listing) - base = 'origin' if tips else 'bundle' + # A verified save refuses an unborn HEAD at rev-parse below. + unborn = not scanned and not has_commit(repo) + base = 'unborn' if unborn else 'origin' if tips else 'bundle' record = {'path': relative, 'origin': origin, 'base': base, - 'head': git(repo, 'rev-parse', 'HEAD').decode().strip(), + 'head': None if unborn else git(repo, 'rev-parse', 'HEAD').decode().strip(), 'branch': git(repo, 'branch', '--show-current').decode().strip(), 'shallow': shallow_commits(repo), 'excluded': [], 'files': [], 'deleted': []} + if unborn: + record['object_format'] = git(repo, 'rev-parse', '--show-object-format').decode().strip() if base == 'origin': # Older images refuse version 2, so it marks only what needs it. manifest['version'] = 2 @@ -361,7 +393,7 @@ def pack(workspace, output): beyond = [name for name in owned if git( repo, 'rev-list', '--max-count=1', '--stdin', input=(name + '\n').encode() + tips).strip()] revisions = ''.join(name + '\n' for name in beyond).encode() + tips - if beyond: + if beyond and scanned: # The same commits as the bundle, bounded by their boundary # rather than by every origin tip, which can exceed one # argument's size limit. @@ -369,14 +401,20 @@ def pack(workspace, output): input=revisions).decode().splitlines() if line.startswith('-')] scan(repo, history=True, revisions=beyond + ['^' + value for value in boundary]) - else: + elif scanned: scan(repo, history=True) - review_changes(repo, review / str(index)) + if scanned: + review_changes(repo, review / str(index)) bundle = destination / 'history.bundle' if base == 'bundle': # The whole history, and every file: restore needs nothing else. git(repo, 'bundle', 'create', str(bundle), '--all', 'HEAD') names = git(repo, 'ls-files', '-z', '--cached', '--others', '--exclude-standard') + elif base == 'unborn': + # Branches other than the unborn HEAD, if any, and every file. + if git(repo, 'for-each-ref', '--count=1', '--format=%(refname)').strip(): + git(repo, 'bundle', 'create', str(bundle), '--all') + names = git(repo, 'ls-files', '-z', '--cached', '--others', '--exclude-standard') else: # Only commits origin lacks, and only files that differ from HEAD. # git refuses an empty bundle, so a fully pushed repository has none. @@ -388,9 +426,11 @@ def pack(workspace, output): total += bundle.stat().st_size if total > MAX_BYTES: raise ValueError('Checkpoint history exceeds the 256 MiB source budget') - if base == 'bundle': + if base == 'bundle' and scanned: # Preserve independent validation of self-contained bundles. # Origin deltas depend on prerequisites supplied by the forge. + # A quarantine skips it to stay inside its deadline; restore + # validates the bundle again either way. with tempfile.TemporaryDirectory(prefix='agent-checkpoint-history-') as history_temp: restore_history(bundle, record, Path(history_temp) / 'repo') for raw in sorted(set(names.split(b'\0')) - {b''}): @@ -440,8 +480,9 @@ def pack(workspace, output): if total > MAX_BYTES: raise ValueError('Checkpoint history exceeds the 256 MiB source budget') (stage / 'manifest.json').write_text(json.dumps(manifest)) - shutil.copyfile(stage / 'manifest.json', review / 'manifest.json') - scan(review) + if scanned: + shutil.copyfile(stage / 'manifest.json', review / 'manifest.json') + scan(review) with tarfile.open(output, 'w:gz', dereference=False) as archive: archive.add(stage, arcname='snapshot') os.chmod(output, 0o600) @@ -453,7 +494,11 @@ def digest(path): return hashlib.file_digest(stream, 'sha256').digest() -def restore(archive_path, destination): +def restore(archive_path, destination, *, scanned=True): + """Restore an archive into a new destination. + + `scanned=False` restores a quarantine, which pack() saved without the + secret scan; it skips the restore-time scan, which would refuse it again.""" if destination.exists(): raise ValueError('Restore requires a new destination; existing work is never overwritten') with tempfile.TemporaryDirectory(prefix='agent-restore-') as temp: @@ -472,7 +517,7 @@ def restore(archive_path, destination): raise ValueError('Unsupported checkpoint version') destination.mkdir(mode=0o700, parents=True) try: - restore_repositories(stage, manifest, destination) + restore_repositories(stage, manifest, destination, scanned=scanned) except BaseException: # The destination is new, so a refused restore leaves nothing behind. shutil.rmtree(destination, ignore_errors=True) @@ -480,7 +525,7 @@ def restore(archive_path, destination): return manifest -def restore_repositories(stage, manifest, destination): +def restore_repositories(stage, manifest, destination, *, scanned=True): with tempfile.TemporaryDirectory(prefix='agent-restore-review-') as review_temp: # The same scope as pack(): what each repository adds to its HEAD. review = Path(review_temp) @@ -491,10 +536,14 @@ def restore_repositories(stage, manifest, destination): raise ValueError('Checkpoint repository path is reserved') source = stage / 'repositories' / str(index) repo = destination / relative - if not re.fullmatch('[0-9a-f]{40,64}', record['head']): - raise ValueError('Invalid checkpoint revision') # Version 1 checkpoints are all bundle snapshots. base = record.get('base', 'bundle') + if base == 'unborn' and not scanned: + restore_unborn(source, record, repo) + continue + # A scanned restore never accepts `unborn`: it has no head. + if not isinstance(record['head'], str) or not re.fullmatch('[0-9a-f]{40,64}', record['head']): + raise ValueError('Invalid checkpoint revision') if base == 'bundle': restore_history(source / 'history.bundle', record, repo) elif base == 'origin': @@ -515,7 +564,14 @@ def restore_repositories(stage, manifest, destination): if record['branch']: git(repo, 'check-ref-format', '--branch', record['branch']) git(repo, 'checkout', '-B', record['branch'], timeout=600) - git(repo, 'remote', 'set-url', 'origin', record['origin']) + if record['origin']: + git(repo, 'remote', 'set-url', 'origin', record['origin']) + elif scanned: + raise ValueError('Checkpoint repository has no origin') + else: + # A quarantined repository without a usable origin: drop the + # temporary bundle URL, but keep its remote-tracking refs. + git(repo, 'config', '--unset', 'remote.origin.url') if base == 'bundle': # Absence from the snapshot means deletion, except for excluded # paths: they were never saved, so they keep their committed content. @@ -531,8 +587,38 @@ def restore_repositories(stage, manifest, destination): if excluded(path): raise ValueError('Checkpoint file path is reserved') place(repo, source / 'files', path) - review_changes(repo, review / str(index)) - scan(review) + if scanned: + review_changes(repo, review / str(index)) + if scanned: + scan(review) + + +def restore_unborn(source, record, repo): + """Restore a quarantined repository whose HEAD had no commit. + + Its saved branches come from the bundle, if it has one; HEAD names the + unborn branch again, and the saved files return to the working tree.""" + if record.get('object_format') not in ('sha1', 'sha256') or not isinstance(record.get('files'), list): + raise ValueError('Invalid checkpoint repository record') + run('git', '-c', 'core.hooksPath=/dev/null', 'init', + '--object-format=' + record['object_format'], str(repo)) + bundle = source / 'history.bundle' + if bundle.exists(): + # Branches, tags, and remote-tracking refs only; never replacements. + # --update-head-ok: the new repository's default branch can share a + # saved branch's name; HEAD is pointed at the unborn branch below. + git(repo, 'fetch', '--no-tags', '--update-head-ok', str(bundle), '+refs/heads/*:refs/heads/*', + '+refs/tags/*:refs/tags/*', '+refs/remotes/*:refs/remotes/*', timeout=600) + if record['branch']: + git(repo, 'check-ref-format', '--branch', record['branch']) + git(repo, 'symbolic-ref', 'HEAD', 'refs/heads/' + record['branch']) + if record['origin']: + git(repo, 'remote', 'add', 'origin', record['origin']) + for value in record['files']: + path = safe_path(value) + if excluded(path): + raise ValueError('Checkpoint file path is reserved') + place(repo, source / 'files', path) def clone_origin(origin, repo): @@ -691,16 +777,17 @@ def storage(): return bucket, prefix + session + '/' -def save(workspace): +def upload(workspace, prefix, *, scanned): + """Pack the workspace, store it under `prefix`, and return its receipt.""" step('reading the checkpoint settings') - bucket, prefix = storage() + bucket, session = storage() with tempfile.TemporaryDirectory(prefix='agent-upload-') as temp: artifact = Path(temp) / 'checkpoint.tar.gz' step('packing the workspace') - manifest = pack(workspace, artifact) + manifest = pack(workspace, artifact, scanned=scanned) step('uploading the checkpoint') checksum = base64.b64encode(digest(artifact)).decode() - key = prefix + str(uuid.uuid4()) + '.tar.gz' + key = session + prefix + str(uuid.uuid4()) + '.tar.gz' run('aws', 's3api', 'put-object', '--bucket', bucket, '--key', key, '--body', str(artifact), '--if-none-match', '*', '--server-side-encryption', 'AES256', '--checksum-algorithm', 'SHA256', '--checksum-sha256', checksum) @@ -708,22 +795,40 @@ def save(workspace): '--key', key, '--checksum-mode', 'ENABLED')) if head.get('ChecksumSHA256') != checksum or head.get('ContentLength') != artifact.stat().st_size: raise ValueError('Checkpoint upload could not be verified') - receipt = {'version': 1, 'key': key, 'sha256': digest(artifact).hex(), - 'repositories': len(manifest['repositories']), 'bytes': artifact.stat().st_size} - private = workspace / '.config/buzz-agent' - private.mkdir(parents=True, exist_ok=True, mode=0o700) - receipt_path = private / 'checkpoint.json' - fd = os.open(receipt_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600) - with os.fdopen(fd, 'w') as stream: - json.dump(receipt, stream) - return receipt + return {'version': 1, 'key': key, 'sha256': digest(artifact).hex(), + 'repositories': len(manifest['repositories']), 'bytes': artifact.stat().st_size} + + +def quarantine(workspace): + """Store the workspace without the secret scan, after a verified save failed. + + The archive goes to the session's `quarantine/` prefix, which the reaper + reads separately and restore accepts only through explicit recovery. It + writes no verified receipt.""" + return upload(workspace, QUARANTINE, scanned=False) + + +def save(workspace): + receipt = upload(workspace, '', scanned=True) + private = workspace / '.config/buzz-agent' + private.mkdir(parents=True, exist_ok=True, mode=0o700) + receipt_path = private / 'checkpoint.json' + fd = os.open(receipt_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600) + with os.fdopen(fd, 'w') as stream: + json.dump(receipt, stream) + return receipt def recover(key, destination): bucket, _ = storage() - if not key.startswith(os.environ['BUZZ_CHECKPOINT_PREFIX']) or not re.fullmatch( - '[a-z0-9-]+/[a-z0-9-]{8,64}/[a-f0-9-]{36}\\.tar\\.gz', key): + match = re.fullmatch('[a-z0-9-]+/[a-z0-9-]{8,64}/(' + QUARANTINE + ')?[a-f0-9-]{36}\\.tar\\.gz', key) + if not key.startswith(os.environ['BUZZ_CHECKPOINT_PREFIX']) or not match: raise ValueError('Checkpoint does not belong to the configured developer prefix') + # The key, not the archive's manifest, selects the unscanned restore. The + # reaper records quarantine keys from its own listing, but every session + # of the developer can write under the prefix, so the listing is only as + # trustworthy as that developer's Sandboxes (as for `listed` keys). + scanned = match.group(1) is None head = json.loads(run('aws', 's3api', 'head-object', '--bucket', bucket, '--key', key, '--checksum-mode', 'ENABLED')) if not 0 < head.get('ContentLength', 0) <= MAX_BYTES * 2 or not head.get('ChecksumSHA256'): @@ -736,7 +841,7 @@ def recover(key, destination): '--bucket', bucket, '--key', key, str(artifact), timeout=None) if artifact.stat().st_size != head['ContentLength'] or base64.b64encode(digest(artifact)).decode() != head['ChecksumSHA256']: raise ValueError('Downloaded checkpoint checksum differs') - return restore(artifact, destination) + return restore(artifact, destination, scanned=scanned) def restore_once(key, workspace): @@ -782,7 +887,7 @@ def restore_once(key, workspace): def main(): parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('action', choices=['save', 'pack', 'restore', 'recover', 'restore-once']) + parser.add_argument('action', choices=['save', 'quarantine', 'pack', 'restore', 'recover', 'restore-once']) parser.add_argument('--workspace', type=Path, default=Path('/home/agent')) parser.add_argument('--archive', type=Path) parser.add_argument('--destination', type=Path) @@ -792,6 +897,8 @@ def main(): step('restoring the checkpoint') if args.action == 'save': result = save(args.workspace) + elif args.action == 'quarantine': + result = quarantine(args.workspace) elif args.action == 'pack' and args.archive: result = pack(args.workspace, args.archive) elif args.action == 'restore' and args.archive and args.destination: diff --git a/voltage/images/buzz-agent/test-checkpoint-storage.py b/voltage/images/buzz-agent/test-checkpoint-storage.py index 680c07a8b0e..a10ba6e3ddb 100644 --- a/voltage/images/buzz-agent/test-checkpoint-storage.py +++ b/voltage/images/buzz-agent/test-checkpoint-storage.py @@ -98,11 +98,14 @@ def stop_harness(bucket, expected): if (code == 0) != expected: raise AssertionError('checkpoint shutdown returned the wrong status') log.seek(0) events = [] + details = {} for line in log: try: fields = json.loads(line).get('fields', {}) if fields.get('schema') == 'buzz.activity.v1': - events.append(json.loads(fields['record'])['event']) + record = json.loads(fields['record']) + events.append(record['event']) + details[record['event']] = record.get('details', {}) except json.JSONDecodeError: pass if expected: @@ -110,6 +113,7 @@ def stop_harness(bucket, expected): return json.loads(receipt_path.read_text()) assert 'checkpoint_failed' in events and 'checkpoint_saved' not in events assert not receipt_path.exists() + return events, details finally: if process.poll() is None: process.kill() @@ -121,8 +125,63 @@ def stop_harness(bucket, expected): assert final_edit == (repo/'source.txt').read_text(), 'adapter survived harness shutdown' run('buzz-agent-checkpoint','recover','--key',receipt['key'],'--destination','/home/agent/after-stop') assert Path('/home/agent/after-stop/example/source.txt').read_text() == final_edit -stop_harness('nonexistent-checkpoint-fixture', False) -print('PASS: SIGTERM drains a writing adapter before verified storage; recovery preserves its final edit; upload failure exits nonzero and emits checkpoint_failed') +events, _ = stop_harness('nonexistent-checkpoint-fixture', False) +# The quarantine uploads to the same missing bucket, so it fails too. +assert events.index('checkpoint_failed') < events.index('checkpoint_quarantine_failed') +print('PASS: SIGTERM drains a writing adapter before verified storage; recovery preserves its final edit; upload failure exits nonzero and emits checkpoint_failed and checkpoint_quarantine_failed') + +# A final save the secret scan refuses quarantines the workspace instead of +# leaving it only on the Pod's node. A repository without an origin, which a +# verified save also refuses, is bundled whole. Credential paths stay out. +scratch = Path('/home/agent/scratch') +scratch.mkdir() +run('git','-C',str(scratch),'init') +run('git','-C',str(scratch),'config','user.name','Checkpoint Fixture') +run('git','-C',str(scratch),'config','user.email','fixture@example.invalid') +(scratch/'notes.txt').write_text('local only\n') +run('git','-C',str(scratch),'add','.') +run('git','-C',str(scratch),'-c','commit.gpgsign=false','commit','-m','test: local-only fixture') +fresh = Path('/home/agent/fresh') +fresh.mkdir() +run('git','-C',str(fresh),'init','-b','main') +(fresh/'draft.txt').write_text('never committed\n') +leak = 'github_token = "ghp_'+'aB3dE6gH9jK2mN5pQ8sT1vW4yZ7cF0iL3oR6'+'"\n' +(repo/'leak.txt').write_text(leak) +(repo/'.env').write_text('FIXTURE_TOKEN=never-packed\n') +refused = subprocess.run(['buzz-agent-checkpoint','save'],capture_output=True) +assert refused.returncode != 0, 'a verified save admitted a credential' +events, details = stop_harness(os.environ['BUZZ_CHECKPOINT_BUCKET'], False) +assert events.index('checkpoint_failed') < events.index('checkpoint_quarantined') < events.index('harness_stopped') +key = details['checkpoint_quarantined']['key'] +session = os.environ['BUZZ_CHECKPOINT_PREFIX'] + os.environ['BUZZ_MANAGED_AGENT_START_NONCE'] + '/' +assert key.startswith(session + 'quarantine/') and key.count('/') == 3, key +assert not Path('/home/agent/.config/buzz-agent/checkpoint.json').exists(), 'a quarantine is never a verified receipt' +run('buzz-agent-checkpoint','recover','--key',key,'--destination','/home/agent/from-quarantine') +recovered = Path('/home/agent/from-quarantine') +assert (recovered/'example/leak.txt').read_text() == leak +assert not (recovered/'example/.env').exists() +assert (recovered/'scratch/notes.txt').read_text() == 'local only\n' +leftover = subprocess.run(['git','-C',str(recovered/'scratch'),'config','--get','remote.origin.url'],capture_output=True) +assert leftover.returncode != 0, 'the temporary bundle URL was left behind' +assert (recovered/'fresh/draft.txt').read_text() == 'never committed\n' +assert run('git','-C',str(recovered/'fresh'),'symbolic-ref','HEAD').strip() == b'refs/heads/main' +print('PASS: a refused final save quarantines the workspace outside the session prefix; explicit recovery restores the unscanned work, a repository without an origin, and one without a commit, never .env') + +# Only the key selects the unscanned restore: the same archive stored under an +# ordinary session key takes the verified path, which refuses it (the unborn +# repository, exit 14, or the restore-time scan, exit 11) and leaves nothing. +import base64, hashlib, uuid +archive = Path('/tmp/quarantine.tar.gz') +run('aws','s3api','get-object','--bucket',os.environ['BUZZ_CHECKPOINT_BUCKET'],'--key',key,str(archive)) +checksum = base64.b64encode(hashlib.sha256(archive.read_bytes()).digest()).decode() +ordinary = session + str(uuid.uuid4()) + '.tar.gz' +run('aws','s3api','put-object','--bucket',os.environ['BUZZ_CHECKPOINT_BUCKET'],'--key',ordinary, + '--body',str(archive),'--checksum-algorithm','SHA256','--checksum-sha256',checksum) +refused = subprocess.run(['buzz-agent-checkpoint','recover','--key',ordinary,'--destination','/home/agent/ordinary'], + capture_output=True, text=True) +assert refused.returncode in (11, 14), refused.returncode +assert not Path('/home/agent/ordinary').exists() +print('PASS: an ordinary checkpoint key takes the verified restore, which refuses the unscanned archive') ''' name = 'checkpoint-test-' + uuid.uuid4().hex[:8] diff --git a/voltage/remote-agents.md b/voltage/remote-agents.md index 7c4184de05c..701e432fb6f 100644 --- a/voltage/remote-agents.md +++ b/voltage/remote-agents.md @@ -118,7 +118,7 @@ sequenceDiagram | ChatGPT refresh token | Dedicated sign-in button | Manager-only Secrets Manager secret | Manager refresh broker | | Depot, Grafana, Linear, Argo CD tokens | Nobody: admins keep them in the vault | 1Password Buzz Agents Dev | The Sandbox, through `via` | -The app removes the dedicated ChatGPT login cache after enrollment or cancellation. Other enrollment credentials are held in memory only. The profile is encrypted in transit and at rest. Checkpoints never include credentials. +The app removes the dedicated ChatGPT login cache after enrollment or cancellation. Other enrollment credentials are held in memory only. The profile is encrypted in transit and at rest. Checkpoints never include the credential files the Sandbox writes. If a final checkpoint fails, for example because the secret scan found a token the agent wrote into the repository, the Sandbox stores an unscanned *quarantine* copy in your checkpoint prefix so the work is not lost; **Recover from checkpoint** restores it. ## Onboard a developer