diff --git a/crates/buzz-backend-kubernetes/src/sandbox.rs b/crates/buzz-backend-kubernetes/src/sandbox.rs index bdf47163fdf..02e35b0590d 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox.rs +++ b/crates/buzz-backend-kubernetes/src/sandbox.rs @@ -3,21 +3,21 @@ //! //! Lifecycle (`docs/remote-agents.md` §Sandbox lifecycle): a session that ended //! — stopped, completed, replaced, lost, or abandoned while binding — is -//! tombstoned by `buzz-sandbox-reaper` and is never redeployed by an ordinary -//! deploy. +//! tombstoned by `buzz-sandbox-reaper` and is only ever resumed by an explicit +//! `recovery` deploy, never by an ordinary one. use std::collections::BTreeMap; use std::time::Duration; use buzz_backend_kubernetes::lifecycle::{ - annotation, owned_pod, sandbox_resource, EndedReason, Lifecycle, OperatingMode, CHANNEL, - CHECKPOINTING, CHECKPOINTING_DISABLED, CHECKPOINTING_ENABLED, ENDED_REASON, GENERATION, - INITIAL_POD, LIFECYCLE, LIFECYCLE_ENDING, OWNER, RESTORE_CHECKPOINT, SANDBOX_BINDING_VERSION, - SESSION, THREAD, + annotation, is_checkpoint_key, owned_pod, sandbox_resource, EndedReason, Lifecycle, + OperatingMode, CHANNEL, CHECKPOINT, CHECKPOINTING, CHECKPOINTING_DISABLED, + CHECKPOINTING_ENABLED, ENDED_REASON, GENERATION, INITIAL_POD, LIFECYCLE, LIFECYCLE_ENDING, + OWNER, RECOVERED_FROM, RESTORE_CHECKPOINT, SANDBOX_BINDING_VERSION, SESSION, THREAD, }; use k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, ObjectFieldSelector, Pod, Secret}; use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference; -use kube::api::{Api, ListParams, Patch, PatchParams, PostParams}; +use kube::api::{Api, DeleteParams, ListParams, Patch, PatchParams, PostParams, Preconditions}; use kube::core::DynamicObject; use kube::{Client, ResourceExt}; use serde::Deserialize; @@ -26,16 +26,39 @@ use sha2::{Digest, Sha256}; use crate::{config::ProviderConfig, intent::Fingerprint, naming, pod}; +#[path = "sandbox_recovery.rs"] +mod recovery_journal; + /// Secret env key carrying the checkpoint a recovered session restores from. /// Always present (empty unless recovering) so its *key* is part of every /// fingerprint and a later ordinary deploy matches the recovered Sandbox. pub const RESTORE_KEY_ENV: &str = "BUZZ_CHECKPOINT_RESTORE_KEY"; +const SANDBOX_TERMINATION_GRACE_SECONDS: u32 = 300; + +/// How an ended session is resumed. Sent only on explicit user action. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum RecoveryMode { + /// Restore the tombstone's verified checkpoint into a new session. + Checkpoint, + /// Start a new session with an empty workspace. + Fresh, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Recovery { + pub mode: RecoveryMode, +} + #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct Options { pub channel_id: String, pub thread_root: String, + #[serde(default)] + pub recovery: Option, } impl Options { @@ -51,6 +74,12 @@ impl Options { if value == &serde_json::Value::Bool(false) { return Ok(None); } + if let Some(recovery) = value.get("recovery") { + serde_json::from_value::(recovery.clone()).map_err(|_| { + "sandbox recovery must be {\"mode\":\"checkpoint\"} or {\"mode\":\"fresh\"}" + .to_string() + })?; + } let options: Self = serde_json::from_value(value.clone()) .map_err(|_| "sandbox requires channel_id and canonical thread_root".to_string())?; if uuid::Uuid::parse_str(&options.channel_id) @@ -85,6 +114,8 @@ fn reason_is(error: &kube::Error, reason: &str) -> bool { struct NewSession<'a> { generation: &'a str, checkpointing: bool, + restore: Option<&'a str>, + recovered_from: Option<&'a str>, } fn build( @@ -121,10 +152,16 @@ fn build( .into(), ), ]); + if let Some(key) = session.restore { + annotations.insert(RESTORE_CHECKPOINT.into(), key.into()); + } + if let Some(previous) = session.recovered_from { + annotations.insert(RECOVERED_FROM.into(), previous.into()); + } let spec = child.spec.as_mut().ok_or("agent Pod has no spec")?; // SIGTERM must leave time to drain sessions and run the harness's bounded // (120s) workspace checkpoint before the kubelet kills the container. - spec.termination_grace_period_seconds = Some(240); + spec.termination_grace_period_seconds = Some(i64::from(SANDBOX_TERMINATION_GRACE_SECONDS)); // The harness writes its checkpoint receipt here; the reaper reads it // back from the container status (and cross-checks it against S3). spec.containers[0].termination_message_policy = Some("File".into()); @@ -328,8 +365,8 @@ impl<'a> Session<'a> { } /// Reuses a running, identically scoped Sandbox. Ended, replaced, or lost -/// sessions are never redeployed implicitly. Namespace creation remains -/// Terraform's concern. +/// sessions are never redeployed implicitly: only `scope.recovery` replaces a +/// tombstone. Namespace creation remains Terraform's concern. pub async fn deploy( client: Client, identity: &naming::AgentIdentity, @@ -347,18 +384,48 @@ pub async fn deploy( let checkpointing = apply_session_env(scope, &mut env); let fingerprints = Fingerprints::new(cfg, &env); let session = Session::new(client, &cfg.namespace, identity, scope, owner, fingerprints); - let existing = session + let mut existing = session .sandboxes .get_opt(&session.name) .await .map_err(|e| format!("read Sandbox: {e}"))?; - let sandbox = match existing { - Some(existing) => existing, - None => create_new(&session, cfg, checkpointing).await?, + if existing.is_none() { + if let Some(tombstone) = recovery_journal::load(&session).await? { + if scope.recovery.is_none() { + return Err( + "Sandbox recovery was interrupted; explicit recovery is required".into(), + ); + } + existing = Some(tombstone); + } + } + let sandbox = match (existing, scope.recovery) { + (Some(tombstone), Some(recovery)) => { + recover(&session, cfg, tombstone, recovery, checkpointing).await? + } + (Some(existing), None) => existing, + ( + None, + Some(Recovery { + mode: RecoveryMode::Checkpoint, + }), + ) => { + return Err( + "no ended session exists for this thread to recover; start fresh instead".into(), + ) + } + (None, _) => create_new(&session, cfg, checkpointing).await?, }; verify_identity(&sandbox, identity, scope, owner)?; verify_live(&sandbox, &session.fingerprints)?; - bind(&session, cfg, sandbox, env).await + let previous = annotation(&sandbox, RECOVERED_FROM).map(str::to_string); + let result = bind(&session, cfg, sandbox, env).await; + if result.is_ok() { + if let Some(previous) = previous { + recovery_journal::clear(&session, &previous).await; + } + } + result } /// Thread-session env shared by every deploy. Returns whether @@ -427,6 +494,8 @@ async fn create_new( let new = NewSession { generation: &generation, checkpointing, + restore: None, + recovered_from: None, }; let desired = build( identity, @@ -453,6 +522,155 @@ async fn create_new( } } +/// Replace an Ended tombstone, or finish binding its journaled replacement. +/// A live workspace is never discarded. +async fn recover( + session: &Session<'_>, + cfg: &ProviderConfig, + tombstone: DynamicObject, + recovery: Recovery, + checkpointing: bool, +) -> Result { + verify_identity(&tombstone, session.identity, session.scope, session.owner)?; + if Lifecycle::of(&tombstone) == Lifecycle::Live { + if let Some(previous) = annotation(&tombstone, RECOVERED_FROM) { + if recovery_journal::load(session) + .await? + .is_some_and(|recorded| annotation(&recorded, GENERATION) == Some(previous)) + { + // Creation already succeeded. Deploy still validates the live + // session and binds its original Pod before clearing the journal. + return Ok(tombstone); + } + } + } + if Lifecycle::of(&tombstone) != Lifecycle::Ended + || OperatingMode::of(&tombstone) != OperatingMode::Suspended + || tombstone.metadata.deletion_timestamp.is_some() + { + return Err( + "recovery requires an ended session; this Sandbox is still active, so end it first" + .into(), + ); + } + let previous = annotation(&tombstone, GENERATION) + .ok_or("ended Sandbox has no generation")? + .to_string(); + let restore = match recovery.mode { + RecoveryMode::Fresh => None, + RecoveryMode::Checkpoint => Some( + annotation(&tombstone, CHECKPOINT) + .filter(|key| is_checkpoint_key(key)) + .ok_or("this session has no verified checkpoint; start fresh instead")? + .to_string(), + ), + }; + // Persist the recovery pointer before deleting its only durable source. + // A retry can resume even if deletion succeeds and creation never runs. + recovery_journal::preserve(session, &tombstone).await?; + let uid = tombstone.metadata.uid.clone(); + let preconditions = Preconditions { + uid: uid.clone(), + resource_version: tombstone.metadata.resource_version.clone(), + }; + match session + .sandboxes + .delete( + &session.name, + &DeleteParams { + preconditions: Some(preconditions), + ..Default::default() + }, + ) + .await + { + Ok(_) => {} + Err(e) if reason_is(&e, "NotFound") => {} + Err(e) if reason_is(&e, "Conflict") => { + return Err("Sandbox changed during recovery; retry".into()) + } + Err(e) => return Err(format!("delete ended Sandbox: {e}")), + } + // Allow the old Pod its complete checkpoint grace plus controller/GC time. + let deadline = tokio::time::Instant::now() + + Duration::from_secs(u64::from(SANDBOX_TERMINATION_GRACE_SECONDS) + 60); + loop { + match session + .sandboxes + .get_opt(&session.name) + .await + .map_err(|e| format!("observe ended Sandbox deletion: {e}"))? + { + Some(current) if current.metadata.uid == uid => { + if tokio::time::Instant::now() >= deadline { + return Err("ended Sandbox is still being deleted; retry recovery".into()); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + // A concurrent recovery of the same tombstone already won. + Some(current) => return recovered_by_peer(current, &previous), + None => { + // Background garbage collection can leave the old Pod after + // its Sandbox disappears. Do not race it with a new owner. + let child = session + .pods + .get_opt(&session.name) + .await + .map_err(|e| format!("observe ended Sandbox Pod deletion: {e}"))?; + match child { + None => break, + Some(child) if owned_pod(&child, &tombstone) => { + if tokio::time::Instant::now() >= deadline { + return Err("ended Sandbox Pod is still being deleted; retry recovery".into()); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + Some(_) => return Err("a foreign Pod occupies the recovered Sandbox name; retain the recovery journal for operator cleanup".into()), + } + } + } + } + let generation = uuid::Uuid::new_v4().simple().to_string(); + let new = NewSession { + generation: &generation, + checkpointing, + restore: restore.as_deref(), + recovered_from: Some(&previous), + }; + let desired = build( + session.identity, + cfg, + session.scope, + session.owner, + &new, + &session.fingerprints.current, + )?; + match session + .sandboxes + .create(&PostParams::default(), &desired) + .await + { + Ok(created) => Ok(created), + Err(e) if reason_is(&e, "AlreadyExists") => { + let current = session + .sandboxes + .get(&session.name) + .await + .map_err(|e| format!("read concurrent Sandbox: {e}"))?; + recovered_by_peer(current, &previous) + } + Err(e) => Err(format!("create recovered Sandbox: {e}")), + } +} + +fn recovered_by_peer(current: DynamicObject, previous: &str) -> Result { + if annotation(¤t, RECOVERED_FROM) == Some(previous) { + Ok(current) + } else { + Err("another session replaced the ended Sandbox during recovery".into()) + } +} + fn container_started(pod: &Pod) -> bool { pod.status .as_ref() @@ -745,6 +963,9 @@ pub async fn stop( scope: &Options, owner: &str, ) -> Result<(String, StopState), String> { + if scope.recovery.is_some() { + return Err("stop does not accept a recovery mode".into()); + } let sandboxes = api(client, namespace); let name = scope.name(identity); for _ in 0..5 { diff --git a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs index b2e71828fba..f5eeb430d70 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs +++ b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs @@ -2,9 +2,9 @@ //! `kube::Client` request path (`tower::service_fn`, as `cluster.rs` tests do). //! //! It models only what the Sandbox paths touch: Sandbox CRUD with -//! resourceVersion CAS, a controller that creates the Sandbox's Pod while it -//! is Running, Pods whose agent container runs once their `envFrom` Secret -//! exists, and Secret create with failure injection. +//! resourceVersion CAS and delete preconditions, a controller that creates +//! the Sandbox's Pod while it is Running, Pods whose agent container runs once +//! their `envFrom` Secret exists, and Secret create with failure injection. use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; @@ -22,12 +22,20 @@ pub struct State { pub secrets: BTreeMap, /// Pod UIDs by name; Pods are rendered from their Sandbox on read. pub pods: BTreeMap, + /// Old Pods left by background GC, returned once before new controller Pods. + pub lingering_pods: BTreeMap, pub rv: u64, pub uid: u64, /// Fail this many Secret creates with a 500. pub fail_secret_creates: usize, /// Store the Secret, then fail this many create responses with a 500. pub fail_secret_create_responses: usize, + /// Fail Pod-owned credential Secret creates, leaving recovery journals intact. + pub fail_binding_secret_creates: usize, + /// Fail this many Sandbox creates with a 500. + pub fail_sandbox_creates: usize, + /// Store the Sandbox, then fail this many create responses with a 500. + pub fail_sandbox_create_responses: usize, /// Answer this many Sandbox patches with a 409 Conflict. pub conflict_patches: usize, /// Every merge patch body received. @@ -135,10 +143,26 @@ impl State { let items: Vec = names.iter().filter_map(|n| self.pod(n)).collect(); ok(&json!({"apiVersion":"v1","kind":"PodList","metadata":{},"items":items})) } - ["api", "v1", "namespaces", _, "pods", name] => match self.pod(name) { - Some(pod) => ok(&pod), - None => status(404, "NotFound"), - }, + ["api", "v1", "namespaces", _, "pods", name] => { + match self.lingering_pods.remove(*name).or_else(|| self.pod(name)) { + Some(pod) => ok(&pod), + None => status(404, "NotFound"), + } + } + ["api", "v1", "namespaces", _, "secrets", name] if method == Method::DELETE => { + let options: Value = serde_json::from_slice(body).unwrap(); + let Some(secret) = self.secrets.get(*name) else { + return status(404, "NotFound"); + }; + if options["preconditions"]["uid"] != secret["metadata"]["uid"] + || options["preconditions"]["resourceVersion"] + != secret["metadata"]["resourceVersion"] + { + return status(409, "Conflict"); + } + self.secrets.remove(*name); + ok(&json!({"apiVersion":"v1","kind":"Status","status":"Success"})) + } ["api", "v1", "namespaces", _, "secrets", name] => match self.secrets.get(*name) { Some(secret) => ok(secret), None => status(404, "NotFound"), @@ -149,6 +173,16 @@ impl State { return status(500, "InternalError"); } let mut secret: Secret = serde_json::from_slice(body).unwrap(); + if self.fail_binding_secret_creates > 0 + && secret + .metadata + .owner_references + .as_ref() + .is_some_and(|refs| refs.iter().any(|owner| owner.kind == "Pod")) + { + self.fail_binding_secret_creates -= 1; + return status(500, "InternalError"); + } let name = secret.metadata.name.clone().unwrap(); if self.secrets.contains_key(&name) { return status(409, "AlreadyExists"); @@ -179,6 +213,10 @@ impl State { None => status(404, "NotFound"), }, (Method::POST, None) => { + if self.fail_sandbox_creates > 0 { + self.fail_sandbox_creates -= 1; + return status(500, "InternalError"); + } let mut sandbox: Value = serde_json::from_slice(body).unwrap(); let name = sandbox["metadata"]["name"].as_str().unwrap().to_string(); if self.sandboxes.contains_key(&name) { @@ -188,6 +226,10 @@ impl State { sandbox["metadata"]["resourceVersion"] = self.next_rv().into(); self.sandboxes.insert(name.clone(), sandbox.clone()); self.reconcile(&name); + if self.fail_sandbox_create_responses > 0 { + self.fail_sandbox_create_responses -= 1; + return status(500, "InternalError"); + } ok(&sandbox) } (Method::PUT, Some(name)) => { @@ -225,6 +267,21 @@ impl State { self.reconcile(name); ok(&updated) } + (Method::DELETE, Some(name)) => { + let options: Value = serde_json::from_slice(body).unwrap_or(Value::Null); + let Some(current) = self.sandboxes.get(name) else { + return status(404, "NotFound"); + }; + let pre = &options["preconditions"]; + for (field, key) in [("uid", "uid"), ("resourceVersion", "resourceVersion")] { + if !pre[field].is_null() && pre[field] != current["metadata"][key] { + return status(409, "Conflict"); + } + } + let removed = self.sandboxes.remove(name).unwrap(); + self.pods.remove(name); + ok(&removed) + } _ => status(405, "MethodNotAllowed"), } } diff --git a/crates/buzz-backend-kubernetes/src/sandbox_recovery.rs b/crates/buzz-backend-kubernetes/src/sandbox_recovery.rs new file mode 100644 index 00000000000..6a066abf8b6 --- /dev/null +++ b/crates/buzz-backend-kubernetes/src/sandbox_recovery.rs @@ -0,0 +1,136 @@ +//! An interrupted delete/create must not erase the only recovery pointer. +//! Keep one immutable journal Secret per thread until its replacement is bound. + +use std::collections::BTreeMap; + +use k8s_openapi::api::core::v1::Secret; +use kube::api::{DeleteParams, PostParams, Preconditions}; +use kube::core::DynamicObject; + +use super::{ + annotation, reason_is, verify_identity, Lifecycle, OperatingMode, Session, CHECKPOINT, + GENERATION, +}; + +const JOURNAL_TYPE: &str = "buzz.block.xyz/sandbox-recovery-v1"; +const TOMBSTONE: &str = "tombstone.json"; + +fn journal_name(session: &Session<'_>) -> String { + format!("{}-recovery", session.name) +} + +async fn read(session: &Session<'_>) -> Result, String> { + let Some(secret) = session + .secrets + .get_opt(&journal_name(session)) + .await + .map_err(|_| "could not read Sandbox recovery journal")? + else { + return Ok(None); + }; + let tombstone = secret + .data + .as_ref() + .and_then(|data| data.get(TOMBSTONE)) + .and_then(|value| serde_json::from_slice::(&value.0).ok()) + .filter(|object| { + secret.immutable == Some(true) + && secret.type_.as_deref() == Some(JOURNAL_TYPE) + && object + .metadata + .uid + .as_ref() + .is_some_and(|uid| !uid.is_empty()) + && object + .metadata + .resource_version + .as_ref() + .is_some_and(|rv| !rv.is_empty()) + && annotation(object, GENERATION).is_some() + && Lifecycle::of(object) == Lifecycle::Ended + && OperatingMode::of(object) == OperatingMode::Suspended + }) + .ok_or("invalid Sandbox recovery journal; retain it for operator recovery")?; + verify_identity(&tombstone, session.identity, session.scope, session.owner)?; + Ok(Some((secret, tombstone))) +} + +pub(super) async fn load(session: &Session<'_>) -> Result, String> { + Ok(read(session).await?.map(|(_, tombstone)| tombstone)) +} + +fn delete_params(secret: &Secret) -> DeleteParams { + DeleteParams { + preconditions: Some(Preconditions { + uid: secret.metadata.uid.clone(), + resource_version: secret.metadata.resource_version.clone(), + }), + ..Default::default() + } +} + +pub(super) async fn preserve( + session: &Session<'_>, + tombstone: &DynamicObject, +) -> Result<(), String> { + let name = journal_name(session); + for _ in 0..5 { + if let Some((secret, recorded)) = read(session).await? { + if recorded.metadata.uid == tombstone.metadata.uid { + if annotation(&recorded, GENERATION) != annotation(tombstone, GENERATION) + || annotation(&recorded, CHECKPOINT) != annotation(tombstone, CHECKPOINT) + { + return Err("Sandbox recovery journal disagrees with the tombstone".into()); + } + return Ok(()); + } + // A later tombstone is still durable while its older journal is replaced. + match session.secrets.delete(&name, &delete_params(&secret)).await { + Ok(_) => {} + Err(error) if reason_is(&error, "NotFound") || reason_is(&error, "Conflict") => { + continue + } + Err(_) => return Err("could not replace Sandbox recovery journal".into()), + } + } + let encoded = serde_json::to_string(tombstone) + .map_err(|_| "could not encode Sandbox recovery journal")?; + let secret = Secret { + metadata: kube::core::ObjectMeta { + name: Some(name.clone()), + ..Default::default() + }, + immutable: Some(true), + type_: Some(JOURNAL_TYPE.into()), + // No ownerRef to the old Sandbox: its deletion must leave the journal intact. + string_data: Some(BTreeMap::from([(TOMBSTONE.into(), encoded)])), + ..Default::default() + }; + match session + .secrets + .create(&PostParams::default(), &secret) + .await + { + Ok(_) => return Ok(()), + Err(error) if reason_is(&error, "AlreadyExists") => continue, + Err(_) => { + return Err( + "could not preserve Sandbox recovery journal; tombstone retained".into(), + ) + } + } + } + Err("Sandbox recovery journal kept changing; retry recovery".into()) +} + +pub(super) async fn clear(session: &Session<'_>, previous: &str) { + if let Ok(Some((secret, tombstone))) = read(session).await { + if annotation(&tombstone, GENERATION) == Some(previous) { + // Cleanup failure leaves a durable record, never an uncertain recovery. + let _ = session + .secrets + .delete(&journal_name(session), &delete_params(&secret)) + .await; + } + } +} diff --git a/crates/buzz-backend-kubernetes/src/sandbox_recovery_retry_tests.rs b/crates/buzz-backend-kubernetes/src/sandbox_recovery_retry_tests.rs new file mode 100644 index 00000000000..49d516d2f46 --- /dev/null +++ b/crates/buzz-backend-kubernetes/src/sandbox_recovery_retry_tests.rs @@ -0,0 +1,179 @@ +use super::*; + +#[tokio::test(start_paused = true)] +async fn retry_reuses_the_replacement_after_a_lost_create_response() { + assert_recovery_retry(true).await; +} + +#[tokio::test(start_paused = true)] +async fn retry_finishes_binding_the_replacement_after_a_secret_write_failure() { + assert_recovery_retry(false).await; +} + +async fn assert_recovery_retry(lost_create_response: bool) { + for mode in [RecoveryMode::Checkpoint, RecoveryMode::Fresh] { + let state = fake::Shared::default(); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); + let name = deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env(), + ) + .await + .unwrap(); + let previous = sandbox_json(&state, &name)["metadata"]["annotations"][GENERATION] + .as_str() + .unwrap() + .to_string(); + let key = format!("dev/{previous}/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"); + tombstone(&state, &name, &key); + { + let mut guard = state.lock().unwrap(); + guard.secrets.clear(); + if lost_create_response { + guard.fail_sandbox_create_responses = 1; + } else { + guard.fail_binding_secret_creates = 1; + } + } + let recovery = fake_scope(Some(mode)); + let error = deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env(), + ) + .await + .unwrap_err(); + assert!( + error.contains(if lost_create_response { + "create recovered Sandbox" + } else { + "retry deploy" + }), + "{error}" + ); + let replacement = sandbox_json(&state, &name); + let pod_uid = state.lock().unwrap().pods[&name].clone(); + let journal_name = format!("{name}-recovery"); + let journal = state.lock().unwrap().secrets[&journal_name].clone(); + + // A journal alone must not authorize reuse of an unrelated generation. + state.lock().unwrap().sandboxes.get_mut(&name).unwrap()["metadata"]["annotations"] + [RECOVERED_FROM] = "unrelated-generation".into(); + assert!(deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env(), + ) + .await + .unwrap_err() + .contains("still active")); + state + .lock() + .unwrap() + .sandboxes + .insert(name.clone(), replacement.clone()); + assert!(deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "another-owner", + plain_env(), + ) + .await + .is_err()); + let mut changed_cfg = cfg.clone(); + changed_cfg.pod_options.active_deadline_seconds = Some(3601); + assert!(deploy( + fake::client(&state), + &identity, + &changed_cfg, + &recovery, + "owner", + plain_env(), + ) + .await + .unwrap_err() + .contains("launch configuration differs")); + + assert_eq!( + deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env(), + ) + .await + .unwrap(), + name + ); + let resumed = sandbox_json(&state, &name); + for field in ["uid", "name"] { + assert_eq!(resumed["metadata"][field], replacement["metadata"][field]); + } + assert_eq!( + resumed["metadata"]["annotations"][GENERATION], + replacement["metadata"]["annotations"][GENERATION] + ); + assert_eq!(resumed["metadata"]["annotations"][RECOVERED_FROM], previous); + assert_eq!(state.lock().unwrap().pods[&name], pod_uid); + assert!(!state.lock().unwrap().secrets.contains_key(&journal_name)); + assert_eq!( + secret_env(&state, RESTORE_KEY_ENV), + [if mode == RecoveryMode::Checkpoint { + key + } else { + String::new() + }] + ); + assert!( + deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env(), + ) + .await + .unwrap_err() + .contains("still active"), + "a completed recovery must not authorize another replacement" + ); + + // A stale journal must not strand a replacement that has since ended. + state + .lock() + .unwrap() + .secrets + .insert(journal_name.clone(), journal); + tombstone(&state, &name, "missing"); + deploy( + fake::client(&state), + &identity, + &cfg, + &fake_scope(Some(RecoveryMode::Fresh)), + "owner", + plain_env(), + ) + .await + .unwrap(); + assert_ne!( + sandbox_json(&state, &name)["metadata"]["uid"], + resumed["metadata"]["uid"] + ); + assert!(!state.lock().unwrap().secrets.contains_key(&journal_name)); + } +} diff --git a/crates/buzz-backend-kubernetes/src/sandbox_tests.rs b/crates/buzz-backend-kubernetes/src/sandbox_tests.rs index 51092f7ebdb..e82c67b2c1b 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox_tests.rs +++ b/crates/buzz-backend-kubernetes/src/sandbox_tests.rs @@ -6,6 +6,9 @@ use nostr::nips::nip19::ToBech32; #[path = "sandbox_fake.rs"] mod fake; +#[path = "sandbox_recovery_retry_tests.rs"] +mod recovery_retry; + #[test] fn custom_image_is_inside_a_real_sandbox_with_a_downward_uid() { let identity = naming::AgentIdentity::from_nsec( @@ -17,11 +20,14 @@ fn custom_image_is_inside_a_real_sandbox_with_a_downward_uid() { let scope = Options { channel_id: uuid::Uuid::new_v4().to_string(), thread_root: "c".repeat(64), + recovery: None, }; let fps = Fingerprints::new(&cfg, &BTreeMap::new()); let new = NewSession { generation: "generation", checkpointing: false, + restore: None, + recovered_from: None, }; let mut sandbox = build(&identity, &cfg, &scope, "owner", &new, &fps.current).unwrap(); assert_eq!( @@ -111,6 +117,7 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { let scope = Options { channel_id: uuid::Uuid::new_v4().to_string(), thread_root: "a".repeat(64), + recovery: None, }; let (first, concurrent) = tokio::join!( deploy( @@ -192,6 +199,7 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { let second_scope = Options { channel_id: scope.channel_id.clone(), thread_root: "b".repeat(64), + recovery: None, }; let second = deploy( client.clone(), @@ -247,8 +255,30 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { .unwrap(); assert_eq!(sandboxes.items.len(), 2, "one Sandbox per canonical thread"); - // Lifecycle: stop suspends the session; once tombstoned (the reaper's job, - // simulated here) an ordinary deploy is refused. + // Lifecycle: recovery is refused on a live session; stop suspends it; once + // tombstoned (the reaper's job, simulated here) only explicit recovery + // resumes it, under a new generation. + let fresh = Options { + recovery: Some(Recovery { + mode: RecoveryMode::Fresh, + }), + ..Options { + channel_id: second_scope.channel_id.clone(), + thread_root: second_scope.thread_root.clone(), + recovery: None, + } + }; + let refused = deploy( + client.clone(), + &identity, + &cfg, + &fresh, + "owner", + fixture_env(), + ) + .await + .unwrap_err(); + assert!(refused.contains("still active"), "{refused}"); let (_, state) = stop( client.clone(), &identity, @@ -263,7 +293,7 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { let old = sandboxes.get(&second).await.unwrap(); assert_eq!(old.data["spec"]["operatingMode"], "Suspended"); let tombstone = json!({"metadata": {"annotations": { - (LIFECYCLE): LIFECYCLE_ENDED, (lifecycle::CHECKPOINT): lifecycle::CHECKPOINT_MISSING, + (LIFECYCLE): LIFECYCLE_ENDED, (CHECKPOINT): lifecycle::CHECKPOINT_MISSING, (lifecycle::ENDED_AT): "2026-01-01T00:00:00Z"}}}); sandboxes .patch(&second, &PatchParams::default(), &Patch::Merge(&tombstone)) @@ -280,6 +310,46 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { .await .unwrap_err(); assert!(ended.contains("explicit recovery is required"), "{ended}"); + let from_checkpoint = Options { + recovery: Some(Recovery { + mode: RecoveryMode::Checkpoint, + }), + channel_id: second_scope.channel_id.clone(), + thread_root: second_scope.thread_root.clone(), + }; + let no_checkpoint = deploy( + client.clone(), + &identity, + &cfg, + &from_checkpoint, + "owner", + fixture_env(), + ) + .await + .unwrap_err(); + assert!(no_checkpoint.contains("start fresh"), "{no_checkpoint}"); + assert_eq!( + deploy( + client.clone(), + &identity, + &cfg, + &fresh, + "owner", + fixture_env() + ) + .await + .unwrap(), + second + ); + let recovered = sandboxes.get(&second).await.unwrap(); + assert_eq!( + annotation(&recovered, RECOVERED_FROM), + annotation(&old, GENERATION) + ); + assert_ne!( + annotation(&recovered, GENERATION), + annotation(&old, GENERATION) + ); namespaces .delete(&namespace, &DeleteParams::default()) .await @@ -300,10 +370,11 @@ fn fake_cfg() -> ProviderConfig { .unwrap() } -fn fake_scope() -> Options { +fn fake_scope(recovery: Option) -> Options { Options { channel_id: "0b6c1f3e-5a7d-4c2b-9e8f-1a2b3c4d5e6f".into(), thread_root: "d".repeat(64), + recovery: recovery.map(|mode| Recovery { mode }), } } @@ -332,10 +403,22 @@ fn sandbox_json(state: &fake::Shared, name: &str) -> serde_json::Value { state.lock().unwrap().sandboxes[name].clone() } +/// Turn the live session into an Ended tombstone, as the reaper would. +fn tombstone(state: &fake::Shared, name: &str, checkpoint: &str) { + let mut guard = state.lock().unwrap(); + let sandbox = guard.sandboxes.get_mut(name).unwrap(); + sandbox["spec"]["operatingMode"] = "Suspended".into(); + let annotations = &mut sandbox["metadata"]["annotations"]; + annotations[LIFECYCLE] = LIFECYCLE_ENDED.into(); + annotations[ENDED_REASON] = "completed".into(); + annotations[CHECKPOINT] = checkpoint.into(); + guard.pods.remove(name); +} + #[tokio::test(start_paused = true)] async fn fresh_deploy_binds_and_always_carries_an_empty_restore_key() { let state = fake::Shared::default(); - let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); let name = deploy( fake::client(&state), &identity, @@ -378,7 +461,7 @@ async fn fresh_deploy_binds_and_always_carries_an_empty_restore_key() { #[tokio::test(start_paused = true)] async fn a_bound_session_without_its_secret_is_resumed() { let state = fake::Shared::default(); - let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); deploy( fake::client(&state), &identity, @@ -440,7 +523,7 @@ async fn assert_secret_create_retry(stored_before_error: bool) { } else { state.lock().unwrap().fail_secret_creates = 1; } - let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); let error = deploy( fake::client(&state), &identity, @@ -481,10 +564,329 @@ async fn assert_secret_create_retry(stored_before_error: bool) { assert_eq!(secret["metadata"]["ownerReferences"][0]["uid"], pod_uid); } +#[tokio::test(start_paused = true)] +async fn recovery_is_refused_on_a_live_session_and_without_a_tombstone_checkpoint() { + let state = fake::Shared::default(); + let (identity, cfg) = (fake_identity(), fake_cfg()); + let live = fake_scope(None); + for mode in [RecoveryMode::Fresh, RecoveryMode::Checkpoint] { + // No session at all: checkpoint recovery has nothing to restore. + let absent = deploy( + fake::client(&state), + &identity, + &cfg, + &fake_scope(Some(RecoveryMode::Checkpoint)), + "owner", + plain_env(), + ) + .await + .unwrap_err(); + assert!(absent.contains("start fresh"), "{absent}"); + deploy( + fake::client(&state), + &identity, + &cfg, + &live, + "owner", + plain_env(), + ) + .await + .unwrap(); + let refused = deploy( + fake::client(&state), + &identity, + &cfg, + &fake_scope(Some(mode)), + "owner", + plain_env(), + ) + .await + .unwrap_err(); + assert!(refused.contains("still active"), "{mode:?}: {refused}"); + state.lock().unwrap().sandboxes.clear(); + state.lock().unwrap().pods.clear(); + state.lock().unwrap().secrets.clear(); + } + deploy( + fake::client(&state), + &identity, + &cfg, + &live, + "owner", + plain_env(), + ) + .await + .unwrap(); + let name = live.name(&identity); + tombstone(&state, &name, lifecycle::CHECKPOINT_MISSING); + let ended = deploy( + fake::client(&state), + &identity, + &cfg, + &live, + "owner", + plain_env(), + ) + .await + .unwrap_err(); + assert!( + ended.contains("ended (completed); explicit recovery is required"), + "{ended}" + ); + let missing = deploy( + fake::client(&state), + &identity, + &cfg, + &fake_scope(Some(RecoveryMode::Checkpoint)), + "owner", + plain_env(), + ) + .await + .unwrap_err(); + assert!(missing.contains("no verified checkpoint"), "{missing}"); + assert_eq!( + sandbox_json(&state, &name)["metadata"]["annotations"][LIFECYCLE], + LIFECYCLE_ENDED, + "a refused recovery must leave the tombstone in place" + ); +} + +#[tokio::test(start_paused = true)] +async fn interrupted_recovery_keeps_its_checkpoint_and_retries_without_starting_fresh() { + for after_delete in [false, true] { + let state = fake::Shared::default(); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); + let name = deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env(), + ) + .await + .unwrap(); + let previous = sandbox_json(&state, &name)["metadata"]["annotations"][GENERATION] + .as_str() + .unwrap() + .to_string(); + let key = format!("dev/{previous}/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"); + tombstone(&state, &name, &key); + { + let mut guard = state.lock().unwrap(); + guard.secrets.clear(); + if after_delete { + guard.fail_sandbox_creates = 1; + } else { + guard.fail_secret_creates = 1; + } + } + let recovery = fake_scope(Some(RecoveryMode::Checkpoint)); + assert!(deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env() + ) + .await + .is_err()); + let journal_name = format!("{name}-recovery"); + { + let guard = state.lock().unwrap(); + if after_delete { + assert!(!guard.sandboxes.contains_key(&name)); + let journal = &guard.secrets[&journal_name]; + assert_eq!(journal["immutable"], true); + assert!(journal["metadata"]["ownerReferences"].is_null()); + } else { + assert_eq!( + guard.sandboxes[&name]["metadata"]["annotations"][CHECKPOINT], + key + ); + } + } + if after_delete { + let original = state.lock().unwrap().secrets[&journal_name].clone(); + for pointer in [ + "/metadata/uid", + "/metadata/annotations/buzz.block.xyz~1owner-pubkey", + ] { + let mut secret: Secret = serde_json::from_value(original.clone()).unwrap(); + let bytes = &mut secret + .data + .as_mut() + .unwrap() + .get_mut("tombstone.json") + .unwrap() + .0; + let mut recorded: serde_json::Value = serde_json::from_slice(bytes).unwrap(); + *recorded.pointer_mut(pointer).unwrap() = if pointer.ends_with("uid") { + serde_json::Value::Null + } else { + json!("another-owner") + }; + *bytes = serde_json::to_vec(&recorded).unwrap(); + state + .lock() + .unwrap() + .secrets + .insert(journal_name.clone(), serde_json::to_value(secret).unwrap()); + assert!(deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env() + ) + .await + .is_err()); + assert!( + state.lock().unwrap().sandboxes.is_empty(), + "invalid journals must not launch a session" + ); + } + state + .lock() + .unwrap() + .secrets + .insert(journal_name.clone(), original); + } + let ordinary = deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env(), + ) + .await + .unwrap_err(); + assert!( + ordinary.contains("explicit recovery is required"), + "{ordinary}" + ); + deploy( + fake::client(&state), + &identity, + &cfg, + &recovery, + "owner", + plain_env(), + ) + .await + .unwrap(); + assert_eq!(secret_env(&state, RESTORE_KEY_ENV), [key]); + assert_eq!( + sandbox_json(&state, &name)["metadata"]["annotations"][RECOVERED_FROM], + previous + ); + assert!(!state.lock().unwrap().secrets.contains_key(&journal_name)); + } +} + +/// Restore key only in checkpoint recovery; fingerprint stable across a +/// recovery, so the next ordinary deploy reuses the recovered session. +#[tokio::test(start_paused = true)] +async fn checkpoint_recovery_restores_once_under_a_new_generation() { + let state = fake::Shared::default(); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); + let name = deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env(), + ) + .await + .unwrap(); + let old = sandbox_json(&state, &name); + let old_generation = old["metadata"]["annotations"][GENERATION].clone(); + let key = format!( + "dev/{}/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz", + old_generation.as_str().unwrap() + ); + tombstone(&state, &name, &key); + state.lock().unwrap().secrets.clear(); + state.lock().unwrap().lingering_pods.insert( + name.clone(), + json!({ + "apiVersion": "v1", "kind": "Pod", "metadata": { + "name": name, "uid": "old-pod", + "ownerReferences": [{"apiVersion": "agents.x-k8s.io/v1beta1", + "kind": "Sandbox", "name": name, "uid": old["metadata"]["uid"], + "controller": true}] + } + }), + ); + deploy( + fake::client(&state), + &identity, + &cfg, + &fake_scope(Some(RecoveryMode::Checkpoint)), + "owner", + plain_env(), + ) + .await + .unwrap(); + let recovered = sandbox_json(&state, &name); + let annotations = &recovered["metadata"]["annotations"]; + assert_eq!(annotations[RECOVERED_FROM], old_generation); + assert_ne!(annotations[GENERATION], old_generation); + assert_eq!(annotations[RESTORE_CHECKPOINT], key.as_str()); + assert!(annotations.get(LIFECYCLE).is_none()); + assert_ne!(recovered["metadata"]["uid"], old["metadata"]["uid"]); + assert_eq!( + annotations[naming::ANNOTATION_CREATE_INTENT], + old["metadata"]["annotations"][naming::ANNOTATION_CREATE_INTENT], + "fingerprint must survive recovery" + ); + assert_eq!( + secret_env(&state, RESTORE_KEY_ENV), + std::slice::from_ref(&key) + ); + // The next ordinary deploy reuses it and does not re-restore. + assert_eq!( + deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env() + ) + .await + .unwrap(), + name + ); + assert_eq!(secret_env(&state, RESTORE_KEY_ENV), [key]); + + // Fresh recovery of a new tombstone carries no restore key. + tombstone(&state, &name, "missing"); + state.lock().unwrap().secrets.clear(); + deploy( + fake::client(&state), + &identity, + &cfg, + &fake_scope(Some(RecoveryMode::Fresh)), + "owner", + plain_env(), + ) + .await + .unwrap(); + assert!(sandbox_json(&state, &name)["metadata"]["annotations"] + .get(RESTORE_CHECKPOINT) + .is_none()); + assert_eq!(secret_env(&state, RESTORE_KEY_ENV), [""]); +} + #[test] fn the_restore_key_is_fingerprinted_by_name_not_value() { let cfg = fake_cfg(); - let scope = fake_scope(); + let scope = fake_scope(None); let mut plain = plain_env(); apply_session_env(&scope, &mut plain); assert_eq!(plain[RESTORE_KEY_ENV], ""); @@ -508,7 +910,7 @@ fn the_restore_key_is_fingerprinted_by_name_not_value() { #[tokio::test(start_paused = true)] async fn a_sandbox_from_before_the_restore_key_is_still_reused() { let state = fake::Shared::default(); - let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); let name = deploy( fake::client(&state), &identity, @@ -549,7 +951,7 @@ async fn a_sandbox_from_before_the_restore_key_is_still_reused() { #[tokio::test(start_paused = true)] async fn stop_suspends_with_a_cas_patch_and_retries_conflicts() { let state = fake::Shared::default(); - let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope(None)); assert_eq!( stop(fake::client(&state), &identity, "agents", &scope, "owner") .await @@ -615,4 +1017,15 @@ async fn stop_suspends_with_a_cas_patch_and_retries_conflicts() { .await .unwrap_err(); assert!(ending.contains("ending"), "{ending}"); + let recovery = fake_scope(Some(RecoveryMode::Fresh)); + assert!(stop( + fake::client(&state), + &identity, + "agents", + &recovery, + "owner" + ) + .await + .unwrap_err() + .contains("recovery")); } diff --git a/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/README.md b/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/README.md index 12a3d28553a..138ce6892c0 100644 --- a/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/README.md +++ b/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/README.md @@ -41,4 +41,5 @@ covered by the conformance suite, not by a static fixture. `stop-*` fixtures use the same payload as `deploy` with `op: "stop"`; the desktop builds both through one path. `stop` requires an approved `identity_policy`, so the fixture without one is refused before any cluster -contact. +contact. `deploy-sandbox-bad-recovery` pins the refusal of an unknown +`provider_config.sandbox.recovery` mode. diff --git a/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/deploy-sandbox-bad-recovery.request.json b/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/deploy-sandbox-bad-recovery.request.json new file mode 100644 index 00000000000..8ec0e7b93ce --- /dev/null +++ b/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/deploy-sandbox-bad-recovery.request.json @@ -0,0 +1,61 @@ +{ + "op": "deploy", + "request_id": "req-8", + "agent": { + "agent_args": [], + "agent_command": "goose", + "auth_tag": "tag-1", + "env_vars": { + "USER_KEY": "user-value" + }, + "idle_timeout_seconds": null, + "launch": { + "args": [ + "acp" + ], + "command": "goose", + "env": { + "GOOSE_MODEL": "gpt-5", + "GOOSE_PROVIDER": "openai", + "USER_KEY": "user-value" + }, + "owner_pubkey": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "policy_env": { + "BUZZ_ACP_AGENTS": "10", + "BUZZ_ACP_DISPLAY_NAME": "worker", + "BUZZ_ACP_LAZY_POOL": "true", + "BUZZ_ACP_MODEL": "gpt-5", + "BUZZ_ACP_RELAY_OBSERVER": "true", + "BUZZ_ACP_SESSION_POLICY": "channel", + "BUZZ_ACP_SESSION_TITLE": "worker", + "GOOSE_MODE": "auto" + } + }, + "max_turn_duration_seconds": null, + "model": "gpt-5", + "name": "worker", + "parallelism": 10, + "private_key_nsec": "nsec1vl029mgpspedva04g90vltkh6fvh240zqtv9k0t9af8935ke9laqsnlfe5", + "provider": "openai", + "relay_url": "wss://relay.example", + "respond_to": "allowlist", + "respond_to_allowlist": [ + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ], + "system_prompt": null, + "turn_timeout_seconds": 300 + }, + "provider_config": { + "namespace": "buzz-agents-test", + "image": "ghcr.io/block/buzz-sprig@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "inactivity_seconds": 3600, + "sandbox": { + "channel_id": "0b6c1f3e-5a7d-4c2b-9e8f-1a2b3c4d5e6f", + "thread_root": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "recovery": { + "mode": "resurrect" + } + } + } +} diff --git a/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/deploy-sandbox-bad-recovery.response.json b/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/deploy-sandbox-bad-recovery.response.json new file mode 100644 index 00000000000..fa9187ff4de --- /dev/null +++ b/crates/buzz-backend-kubernetes/tests/fixtures/provider-wire/deploy-sandbox-bad-recovery.response.json @@ -0,0 +1 @@ +{"ok":false,"error":"sandbox recovery must be {\"mode\":\"checkpoint\"} or {\"mode\":\"fresh\"}"} diff --git a/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs b/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs index c2fccfd6f35..348ddccac70 100644 --- a/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs +++ b/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs @@ -96,6 +96,7 @@ fn responses_match_their_fixtures() { "deploy-relay-mesh-padded", "deploy-tag-image", "deploy-no-owner", + "deploy-sandbox-bad-recovery", "stop-no-identity-policy", ]; // The list must cover every response fixture on disk. A literal array is diff --git a/docs/remote-agents.md b/docs/remote-agents.md index 0b7f078ae29..9ea6c093c5a 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -1523,6 +1523,33 @@ configuration lists for the namespace. It records a Warning Event for a managed Sandbox that it leaves alone (foreign owner, foreign Pod, inconsistent lifecycle). +**Recovery.** An ordinary deploy to an ended session fails with an error +that contains `explicit recovery is required`. The desktop then offers two +actions in the thread. Each sends `provider_config.sandbox.recovery` once, +for that click only; the desktop never stores it: + +- `{"mode":"checkpoint"}` restores the tombstone's verified checkpoint. It is + refused when the tombstone has no checkpoint key. +- `{"mode":"fresh"}` starts with an empty workspace. + +The provider accepts `recovery` for an Ended tombstone or its live replacement +while a matching recovery journal remains. `fresh` is also accepted when no Sandbox exists. +Before deletion, it preserves the tombstone in an immutable `-recovery` Secret without an owner reference. +This journal contains the recovery pointer, not the agent's private key. +If recovery is interrupted, a later explicit recovery resumes from the journal; an ordinary deploy cannot silently start fresh. +If creation succeeded but its response or credential binding failed, retry reuses that replacement and finishes binding its original Pod. +Its `recovered-from` must match the journal's generation. Identity, lifecycle, and launch-configuration checks still apply. +The provider removes the journal after binding the replacement; failed cleanup retains the journal for a later retry. +It deletes the tombstone +with uid and resourceVersion preconditions, waits until it is gone, and +creates a new Sandbox with a new generation and `recovered-from`. In +checkpoint mode the launch Secret carries +`BUZZ_CHECKPOINT_RESTORE_KEY=`; the image restores it once before the +harness starts. Every launch Secret contains that key (empty when unused), +so it is part of every launch fingerprint and later ordinary deploys reuse +the recovered Sandbox. Sandboxes created before this key existed are still +reused. Recovery of a live session without a matching journal is refused. + **Deploy repair.** A deploy that bound the first Pod but did not write the launch Secret is resumed by the next deploy while the container has not started. The provider writes the Secret with the Sandbox's generation and