From ffc3a25d6fd36c268c1f330e84629e2955094ce9 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Mon, 19 Oct 2026 14:43:15 -0400 Subject: [PATCH] feat(agents): refuse ended Sandbox sessions on deploy Move the Sandbox deploy onto the shared lifecycle library. A deploy now splits identity checks (agent, owner, thread, management marker) from liveness checks, and reuses a Sandbox only while it is a live, Running session with a matching launch fingerprint. Each refusal names the lifecycle state (ended, ending, unknown, deleting) so the desktop can offer the right explicit action. New Sandboxes record whether checkpoint-on-stop is configured and use a File termination message policy so the harness receipt reaches the container status. Add an in-memory apiserver that drives the real kube::Client request path, and document the Sandbox lifecycle states and tombstones. Signed-off-by: Tee8z --- crates/buzz-backend-kubernetes/Cargo.toml | 1 + crates/buzz-backend-kubernetes/src/main.rs | 6 +- crates/buzz-backend-kubernetes/src/sandbox.rs | 735 +++++++++++------- .../src/sandbox_fake.rs | 201 +++++ .../src/sandbox_tests.rs | 258 +++++- crates/buzz-backend-kubernetes/src/wire.rs | 110 ++- docs/remote-agents.md | 90 ++- 7 files changed, 1107 insertions(+), 294 deletions(-) create mode 100644 crates/buzz-backend-kubernetes/src/sandbox_fake.rs diff --git a/crates/buzz-backend-kubernetes/Cargo.toml b/crates/buzz-backend-kubernetes/Cargo.toml index 98067dce335..3bf4a65863f 100644 --- a/crates/buzz-backend-kubernetes/Cargo.toml +++ b/crates/buzz-backend-kubernetes/Cargo.toml @@ -40,3 +40,4 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std" [dev-dependencies] tower = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-backend-kubernetes/src/main.rs b/crates/buzz-backend-kubernetes/src/main.rs index 292deb5587a..374dc517db3 100644 --- a/crates/buzz-backend-kubernetes/src/main.rs +++ b/crates/buzz-backend-kubernetes/src/main.rs @@ -109,7 +109,7 @@ fn respond(input: &str) -> Response { }; match runtime.block_on(deploy_agent(&deploy)) { Ok(agent_id) => Response::deployed(agent_id), - Err(e) => Response::error(e), + Err(e) => Response::failed(e), } } } @@ -134,7 +134,7 @@ fn refuse_relay_mesh(raw: &serde_json::Value) -> Option { } /// Run one deploy to a terminal outcome. -async fn deploy_agent(request: &wire::DeployRequest) -> Result { +async fn deploy_agent(request: &wire::DeployRequest) -> Result { let cfg = config::parse(&request.provider_config)?; let sandbox = sandbox::Options::parse(&request.provider_config)?; // Identity before any cluster contact: a malformed nsec is a refusal, not @@ -162,7 +162,7 @@ async fn deploy_agent(request: &wire::DeployRequest) -> Result { return sandbox::deploy(client, &identity, &cfg, &scope, owner, env).await; } let substrate = cluster::Cluster::new(client, &cfg.namespace); - reconcile::deploy(&substrate, &identity, &cfg, env).await + Ok(reconcile::deploy(&substrate, &identity, &cfg, env).await?) } #[cfg(test)] diff --git a/crates/buzz-backend-kubernetes/src/sandbox.rs b/crates/buzz-backend-kubernetes/src/sandbox.rs index 0293a546bd0..2f89b65e8f8 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox.rs +++ b/crates/buzz-backend-kubernetes/src/sandbox.rs @@ -1,27 +1,31 @@ //! Agent Sandbox integration. The upstream controller owns the Pod; this //! provider owns its launch contract and binds credentials to the first Pod UID. +//! +//! 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. 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, OWNER, 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, PostParams}; -use kube::core::{ApiResource, DynamicObject, GroupVersionKind}; +use kube::core::DynamicObject; use kube::{Client, ResourceExt}; use serde::Deserialize; use serde_json::json; use sha2::{Digest, Sha256}; +use crate::wire::{ErrorCode, ProviderError}; use crate::{config::ProviderConfig, intent::Fingerprint, naming, pod}; -const INITIAL_POD: &str = "buzz.block.xyz/initial-pod-uid"; -const GENERATION: &str = "buzz.block.xyz/generation"; -const OWNER: &str = "buzz.block.xyz/owner-pubkey"; -const CHANNEL: &str = "buzz.block.xyz/channel-id"; -const THREAD: &str = "buzz.block.xyz/thread-root"; -const SESSION: &str = "buzz.block.xyz/session-id"; - #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct Options { @@ -64,25 +68,18 @@ impl Options { } } -fn resource() -> ApiResource { - ApiResource::from_gvk(&GroupVersionKind::gvk( - "agents.x-k8s.io", - "v1beta1", - "Sandbox", - )) +fn api(client: Client, namespace: &str) -> Api { + Api::namespaced_with(client, namespace, &sandbox_resource()) } -fn api(client: Client, namespace: &str) -> Api { - Api::namespaced_with(client, namespace, &resource()) +fn reason_is(error: &kube::Error, reason: &str) -> bool { + matches!(error, kube::Error::Api(e) if e.reason == reason) } -fn annotation<'a>(object: &'a DynamicObject, key: &str) -> Option<&'a str> { - object - .metadata - .annotations - .as_ref()? - .get(key) - .map(String::as_str) +/// Session-specific inputs to [`build`] beyond the launch contract. +struct NewSession<'a> { + generation: &'a str, + checkpointing: bool, } fn build( @@ -90,16 +87,18 @@ fn build( cfg: &ProviderConfig, scope: &Options, owner: &str, - generation: &str, + session: &NewSession<'_>, fingerprint: &Fingerprint, ) -> Result { + let generation = session.generation; let name = scope.name(identity); let mut child = pod::build_pod(identity, cfg, generation, fingerprint); child.metadata.name = Some(name.clone()); // Older bare-Pod providers must refuse these controller-owned resources. - child - .labels_mut() - .insert(naming::LABEL_BINDING_VERSION.into(), "2".into()); + child.labels_mut().insert( + naming::LABEL_BINDING_VERSION.into(), + SANDBOX_BINDING_VERSION.into(), + ); let annotations = child.metadata.annotations.get_or_insert_with(BTreeMap::new); annotations.extend([ (OWNER.into(), owner.into()), @@ -107,11 +106,23 @@ fn build( (THREAD.into(), scope.thread_root.clone()), (SESSION.into(), generation.into()), (GENERATION.into(), generation.into()), + ( + CHECKPOINTING.into(), + if session.checkpointing { + CHECKPOINTING_ENABLED + } else { + CHECKPOINTING_DISABLED + } + .into(), + ), ]); // The bare-Pod grace (`TERMINATION_GRACE_SECONDS`) already covers the // drain and the harness's 120s checkpoint; keep it so the template // matches the fingerprinted intent. let spec = child.spec.as_mut().ok_or("agent Pod has no spec")?; + // 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()); spec.containers[0] .env .get_or_insert_with(Vec::new) @@ -126,7 +137,7 @@ fn build( }), ..Default::default() }); - let mut sandbox = DynamicObject::new(&name, &resource()); + let mut sandbox = DynamicObject::new(&name, &sandbox_resource()); sandbox.metadata = child.metadata.clone(); sandbox.data = json!({"spec": { "operatingMode": "Running", @@ -139,89 +150,87 @@ fn build( Ok(sandbox) } -fn verify( +/// The Sandbox belongs to this agent, owner, and thread. Recovery cannot fix a +/// mismatch here, so the error does not offer it. +fn verify_identity( sandbox: &DynamicObject, identity: &naming::AgentIdentity, scope: &Options, owner: &str, - fingerprint: &Fingerprint, -) -> Result<(), String> { - for (key, expected) in [ +) -> Result<(), ProviderError> { + let labels = sandbox.labels(); + let marked = labels.get(naming::LABEL_MANAGED_BY).map(String::as_str) + == Some(naming::MANAGED_BY) + && labels + .get(naming::LABEL_BINDING_VERSION) + .map(String::as_str) + == Some(SANDBOX_BINDING_VERSION); + let matches = [ (naming::ANNOTATION_PUBKEY_FULL, identity.pubkey_hex()), - (naming::ANNOTATION_CREATE_INTENT, fingerprint.as_str()), (OWNER, owner), (CHANNEL, scope.channel_id.as_str()), (THREAD, scope.thread_root.as_str()), - ] { - if annotation(sandbox, key) != Some(expected) { - return Err("Sandbox identity, thread, or launch configuration differs; explicit recovery is required".into()); - } - } - if sandbox - .labels() - .get(naming::LABEL_MANAGED_BY) - .map(String::as_str) - != Some(naming::MANAGED_BY) - || sandbox.metadata.deletion_timestamp.is_some() - || sandbox.data["spec"]["operatingMode"] != "Running" - { - return Err( - "Sandbox is not an active managed session; explicit recovery is required".into(), - ); + ] + .into_iter() + .all(|(key, expected)| annotation(sandbox, key) == Some(expected)); + if !marked || !matches { + return Err(ProviderError::refused( + ErrorCode::IdentityMismatch, + "Sandbox belongs to a different agent, owner, or thread", + )); } Ok(()) } -fn owned_pod(pod: &Pod, sandbox: &DynamicObject) -> bool { - sandbox.metadata.uid.as_ref().is_some_and(|uid| { - pod.owner_references().iter().any(|r| { - &r.uid == uid - && r.kind == "Sandbox" - && r.api_version == "agents.x-k8s.io/v1beta1" - && r.controller == Some(true) - }) - }) -} - -/// Opt-out for launches that knowingly run without a stop checkpoint. -const CHECKPOINT_OPT_OUT: &str = "BUZZ_SANDBOX_CHECKPOINT"; - -/// Status reason and code only: an apiserver message can echo object data, -/// and these calls handle the launch Secret. -fn api_failure(error: &kube::Error) -> String { - match error { - kube::Error::Api(status) => format!("{} ({})", status.reason, status.code), - other => other.to_string(), +/// The Sandbox is a live session this deploy may reuse. Every refusal names the +/// lifecycle state so the desktop can offer the right explicit action. +fn verify_live(sandbox: &DynamicObject, fingerprint: &Fingerprint) -> Result<(), ProviderError> { + let refused = |code, message: &str| Err(ProviderError::refused(code, message)); + if sandbox.metadata.deletion_timestamp.is_some() { + return refused( + ErrorCode::SessionDeleting, + "Sandbox is being deleted; retry once it is gone", + ); } -} - -/// Configure the harness for one owner thread. The workspace is an -/// `emptyDir` on a `restartPolicy: Never` Pod, so the stop checkpoint is the -/// only copy that outlives the process; refuse to launch without its storage -/// unless the caller opts out explicitly. -fn apply_session_env(scope: &Options, env: &mut BTreeMap) -> Result<(), String> { - let opted_out = env.get(CHECKPOINT_OPT_OUT).map(String::as_str) == Some("disabled"); - // A blank value is as unusable as a missing one: the save would fail at stop. - let configured = |key: &str| env.get(key).is_some_and(|value| !value.trim().is_empty()); - if !opted_out && configured("BUZZ_CHECKPOINT_BUCKET") && configured("BUZZ_CHECKPOINT_PREFIX") { - env.insert("BUZZ_ACP_CHECKPOINT_ON_STOP".into(), "true".into()); - } else if !opted_out { - return Err(format!( - "Sandbox requires BUZZ_CHECKPOINT_BUCKET and BUZZ_CHECKPOINT_PREFIX in the agent \ - environment, or {CHECKPOINT_OPT_OUT}=disabled to launch without a stop checkpoint" - )); + match Lifecycle::of(sandbox) { + Lifecycle::Ended => { + // The reason comes from the cluster; only a known spelling is + // repeated to the user. + let message = match annotation(sandbox, ENDED_REASON).and_then(EndedReason::parse) { + Some(reason) => format!( + "Sandbox session ended ({}); explicit recovery is required", + reason.as_str() + ), + None => "Sandbox session ended; explicit recovery is required".into(), + }; + return refused(ErrorCode::SessionEnded, &message); + } + Lifecycle::Ending => { + return refused( + ErrorCode::SessionEnding, + "Sandbox session is ending; recover it explicitly once it has ended", + ) + } + Lifecycle::Unrecognized => { + return refused( + ErrorCode::LifecycleUnknown, + "Sandbox has a lifecycle this provider does not know; update Buzz", + ) + } + Lifecycle::Live => {} + } + if OperatingMode::of(sandbox) != OperatingMode::Running { + return refused( + ErrorCode::SessionNotRunning, + "Sandbox is not an active managed session; explicit recovery is required", + ); + } + if annotation(sandbox, naming::ANNOTATION_CREATE_INTENT) != Some(fingerprint.as_str()) { + return refused( + ErrorCode::ConfigDiffers, + "Sandbox launch configuration differs; end the session, then explicit recovery is required", + ); } - env.insert("BUZZ_ACP_ACTIVITY_LOG".into(), "true".into()); - env.insert("BUZZ_ACP_RESPOND_TO".into(), "owner".into()); - env.insert("BUZZ_ACP_ALLOWED_RESPOND_TO".into(), "owner".into()); - env.insert( - "BUZZ_ACP_BOUND_SESSION".into(), - json!({"channel_id":scope.channel_id,"thread_root":scope.thread_root}).to_string(), - ); - env.insert("BUZZ_ACP_SESSION_POLICY".into(), "thread".into()); - env.insert("BUZZ_ACP_SUBSCRIBE".into(), "mentions".into()); - env.insert("BUZZ_ACP_CHANNELS".into(), scope.channel_id.clone()); - env.insert("BUZZ_ACP_HEARTBEAT_INTERVAL".into(), "0".into()); Ok(()) } @@ -276,8 +285,43 @@ fn configure_setup( Ok(()) } -/// Reuses a running, identically scoped Sandbox. Terminal or replaced Pods are -/// never redeployed implicitly. Namespace creation remains Terraform's concern. +/// Per-deploy handles and invariants. +struct Session<'a> { + sandboxes: Api, + pods: Api, + secrets: Api, + name: String, + identity: &'a naming::AgentIdentity, + scope: &'a Options, + owner: &'a str, + fingerprint: Fingerprint, +} + +impl<'a> Session<'a> { + fn new( + client: Client, + namespace: &str, + identity: &'a naming::AgentIdentity, + scope: &'a Options, + owner: &'a str, + fingerprint: Fingerprint, + ) -> Self { + Self { + sandboxes: api(client.clone(), namespace), + pods: Api::namespaced(client.clone(), namespace), + secrets: Api::namespaced(client, namespace), + name: scope.name(identity), + identity, + scope, + owner, + fingerprint, + } + } +} + +/// Reuses a running, identically scoped Sandbox. Ended, replaced, or lost +/// sessions are never redeployed implicitly. Namespace creation remains +/// Terraform's concern. pub async fn deploy( client: Client, identity: &naming::AgentIdentity, @@ -285,225 +329,371 @@ pub async fn deploy( scope: &Options, owner: &str, mut env: BTreeMap, -) -> Result { +) -> Result { let mut effective_cfg = cfg.clone(); configure_setup(&mut effective_cfg, identity, &mut env)?; let cfg = &effective_cfg; if cfg.pod_options.active_deadline_seconds.is_none() { return Err("Sandbox requires pod_options.active_deadline_seconds".into()); } - apply_session_env(scope, &mut env)?; - // Include the UID binding key in the fingerprint without hashing its value. - env.insert("BUZZ_SANDBOX_INITIAL_POD_UID".into(), String::new()); + let checkpointing = apply_session_env(scope, &mut env)?; let fingerprint = pod::intent_template(cfg, env.keys().cloned()).fingerprint(); - let sandboxes = api(client.clone(), &cfg.namespace); - let pods: Api = Api::namespaced(client.clone(), &cfg.namespace); - let secrets: Api = Api::namespaced(client, &cfg.namespace); - let name = scope.name(identity); - let mut sandbox = match sandboxes - .get_opt(&name) + let session = Session::new(client, &cfg.namespace, identity, scope, owner, fingerprint); + let existing = session + .sandboxes + .get_opt(&session.name) .await - .map_err(|e| format!("read Sandbox: {e}"))? - { + .map_err(|e| format!("read Sandbox: {e}"))?; + let sandbox = match existing { Some(existing) => existing, - None => { - if pods - .get_opt(&name) - .await - .map_err(|e| format!("check Pod collision: {e}"))? - .is_some() - || !pods - .list(&ListParams::default().labels(&identity.selector())) - .await - .map_err(|e| format!("check existing agent consumers: {e}"))? - .items - .iter() - .all(|pod| { - pod.labels() - .get(naming::LABEL_BINDING_VERSION) - .map(String::as_str) - == Some("2") - && pod.annotations().get(OWNER).map(String::as_str) == Some(owner) - && pod.owner_references().iter().any(|r| { - r.api_version == "agents.x-k8s.io/v1beta1" - && r.kind == "Sandbox" - && r.controller == Some(true) - }) + None => create_new(&session, cfg, checkpointing).await?, + }; + verify_identity(&sandbox, identity, scope, owner)?; + verify_live(&sandbox, &session.fingerprint)?; + bind(&session, cfg, sandbox, env).await +} + +/// Opt-out for launches that knowingly run without a stop checkpoint. +const CHECKPOINT_OPT_OUT: &str = "BUZZ_SANDBOX_CHECKPOINT"; + +/// Status reason and code only: an apiserver message can echo object data, +/// and these calls handle the launch Secret. +fn api_failure(error: &kube::Error) -> String { + match error { + kube::Error::Api(status) => format!("{} ({})", status.reason, status.code), + other => other.to_string(), + } +} + +/// Thread-session env shared by every deploy. Returns whether +/// checkpoint-on-stop is configured. The workspace is an `emptyDir` on a +/// `restartPolicy: Never` Pod, so the stop checkpoint is the only copy that +/// outlives the process; refuse to launch without its storage unless the +/// caller opts out explicitly. +fn apply_session_env(scope: &Options, env: &mut BTreeMap) -> Result { + let opted_out = env.get(CHECKPOINT_OPT_OUT).map(String::as_str) == Some("disabled"); + // A blank value is as unusable as a missing one: the save would fail at stop. + let configured = |key: &str| env.get(key).is_some_and(|value| !value.trim().is_empty()); + let checkpointing = + !opted_out && configured("BUZZ_CHECKPOINT_BUCKET") && configured("BUZZ_CHECKPOINT_PREFIX"); + if !opted_out && !checkpointing { + return Err(format!( + "Sandbox requires BUZZ_CHECKPOINT_BUCKET and BUZZ_CHECKPOINT_PREFIX in the agent \ + environment, or {CHECKPOINT_OPT_OUT}=disabled to launch without a stop checkpoint" + )); + } + env.insert("BUZZ_ACP_ACTIVITY_LOG".into(), "true".into()); + env.insert("BUZZ_ACP_RESPOND_TO".into(), "owner".into()); + env.insert("BUZZ_ACP_ALLOWED_RESPOND_TO".into(), "owner".into()); + env.insert( + "BUZZ_ACP_BOUND_SESSION".into(), + json!({"channel_id":scope.channel_id,"thread_root":scope.thread_root}).to_string(), + ); + env.insert("BUZZ_ACP_SESSION_POLICY".into(), "thread".into()); + env.insert("BUZZ_ACP_SUBSCRIBE".into(), "mentions".into()); + env.insert("BUZZ_ACP_CHANNELS".into(), scope.channel_id.clone()); + env.insert("BUZZ_ACP_HEARTBEAT_INTERVAL".into(), "0".into()); + if checkpointing { + env.insert("BUZZ_ACP_CHECKPOINT_ON_STOP".into(), "true".into()); + } + // Include the UID binding key in the fingerprint without hashing its value. + env.insert("BUZZ_SANDBOX_INITIAL_POD_UID".into(), String::new()); + Ok(checkpointing) +} + +async fn create_new( + session: &Session<'_>, + cfg: &ProviderConfig, + checkpointing: bool, +) -> Result { + let identity = session.identity; + if session + .pods + .get_opt(&session.name) + .await + .map_err(|e| format!("check Pod collision: {e}"))? + .is_some() + || !session + .pods + .list(&ListParams::default().labels(&identity.selector())) + .await + .map_err(|e| format!("check existing agent consumers: {e}"))? + .items + .iter() + .all(|pod| { + pod.labels() + .get(naming::LABEL_BINDING_VERSION) + .map(String::as_str) + == Some(SANDBOX_BINDING_VERSION) + && pod.annotations().get(OWNER).map(String::as_str) == Some(session.owner) + && pod.owner_references().iter().any(|r| { + r.api_version == "agents.x-k8s.io/v1beta1" + && r.kind == "Sandbox" + && r.controller == Some(true) }) - { - return Err( - "agent already has a Pod; stop and recover it explicitly before using Sandbox" - .into(), - ); - } - let generation = uuid::Uuid::new_v4().simple().to_string(); - let desired = build(identity, cfg, scope, owner, &generation, &fingerprint)?; - match sandboxes.create(&PostParams::default(), &desired).await { - Ok(created) => created, - Err(kube::Error::Api(e)) if e.reason == "AlreadyExists" => sandboxes - .get(&name) - .await - .map_err(|e| format!("read concurrent Sandbox: {e}"))?, - Err(e) => { - return Err(format!( - "create Sandbox (install the Agent Sandbox v1beta1 CRD/controller first): {e}" - )) - } - } - } + }) + { + return Err(ProviderError::refused( + ErrorCode::AgentPodExists, + "agent already has a Pod; stop and recover it explicitly before using Sandbox", + )); + } + let generation = uuid::Uuid::new_v4().simple().to_string(); + let new = NewSession { + generation: &generation, + checkpointing, }; - verify(&sandbox, identity, scope, owner, &fingerprint)?; + let desired = build( + identity, + cfg, + session.scope, + session.owner, + &new, + &session.fingerprint, + )?; + match session + .sandboxes + .create(&PostParams::default(), &desired) + .await + { + Ok(created) => Ok(created), + Err(e) if reason_is(&e, "AlreadyExists") => session + .sandboxes + .get(&session.name) + .await + .map_err(|e| format!("read concurrent Sandbox: {e}").into()), + Err(e) => Err(format!( + "create Sandbox (install the Agent Sandbox v1beta1 CRD/controller first): {e}" + ) + .into()), + } +} + +/// Bind credentials to the Sandbox's first Pod and wait for it to start. +async fn bind( + session: &Session<'_>, + cfg: &ProviderConfig, + mut sandbox: DynamicObject, + env: BTreeMap, +) -> Result { + let identity = session.identity; + let name = &session.name; let generation = annotation(&sandbox, GENERATION) .ok_or("Sandbox generation is missing")? .to_string(); let deadline = tokio::time::Instant::now() + Duration::from_secs(120); loop { - if let Some(child) = pods - .get_opt(&name) + let Some(child) = session + .pods + .get_opt(name) .await .map_err(|e| format!("read Sandbox Pod: {e}"))? - { - if !owned_pod(&child, &sandbox) { - return Err("Pod is not owned by this Sandbox".into()); + else { + if annotation(&sandbox, INITIAL_POD).is_some() { + return Err(ProviderError::refused( + ErrorCode::PodLost, + "Sandbox lost its original Pod; explicit recovery is required", + )); } - let uid = child.metadata.uid.as_deref().ok_or("Pod UID is missing")?; - if child.metadata.deletion_timestamp.is_some() - || child - .status - .as_ref() - .and_then(|s| s.phase.as_deref()) - .is_some_and(|p| matches!(p, "Failed" | "Succeeded")) - { + if tokio::time::Instant::now() >= deadline { return Err( - "Sandbox session ended; save or recover its workspace explicitly".into(), + "Sandbox controller has not created its Pod; inspect the Sandbox and retry" + .into(), ); } - if let Some(initial_uid) = annotation(&sandbox, INITIAL_POD) { - if uid != initial_uid { - return Err("Sandbox Pod was replaced; credentials remain bound to the original session, explicit recovery is required".into()); - } - let secret = secrets - .get_opt(&identity.secret_name(&generation)) - .await - .map_err(|e| { - format!("could not check Sandbox launch Secret: {}", api_failure(&e)) - })?; - let Some(secret) = secret else { + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + }; + if !owned_pod(&child, &sandbox) { + return Err(ProviderError::refused( + ErrorCode::ForeignPod, + "Pod is not owned by this Sandbox", + )); + } + let uid = child.metadata.uid.as_deref().ok_or("Pod UID is missing")?; + if child.metadata.deletion_timestamp.is_some() + || child + .status + .as_ref() + .and_then(|s| s.phase.as_deref()) + .is_some_and(|p| matches!(p, "Failed" | "Succeeded")) + { + return Err(ProviderError::refused( + ErrorCode::PodEnded, + "Sandbox session ended; explicit recovery is required", + )); + } + let Some(initial_uid) = annotation(&sandbox, INITIAL_POD) else { + // resourceVersion makes this an atomic first-Pod binding. A + // concurrent deploy must retry and observe the winner's binding. + sandbox + .annotations_mut() + .insert(INITIAL_POD.into(), uid.into()); + sandbox = match session + .sandboxes + .replace(name, &PostParams::default(), &sandbox) + .await + { + Ok(bound) => bound, + Err(e) if reason_is(&e, "Conflict") => { + sandbox = refresh(session, &generation).await?; if tokio::time::Instant::now() >= deadline { - return Err("Sandbox credential binding is incomplete; explicit recovery is required".into()); + return Err( + "Sandbox kept changing during credential binding; retry deploy".into(), + ); } - tokio::time::sleep(Duration::from_millis(250)).await; continue; - }; - if secret.immutable != Some(true) - || !secret - .owner_references() - .iter() - .any(|owner| owner.uid == uid && owner.kind == "Pod") - || secret - .annotations() - .get(naming::ANNOTATION_PUBKEY_FULL) - .map(String::as_str) - != Some(identity.pubkey_hex()) - { - return Err("Sandbox credentials belong to a different Pod or agent".into()); - } - let saved_relay = secret - .data - .as_ref() - .and_then(|data| data.get("BUZZ_RELAY_URL")) - .map(|value| value.0.as_slice()); - if saved_relay != env.get("BUZZ_RELAY_URL").map(|value| value.as_bytes()) { - return Err("Sandbox belongs to a different relay; use a new thread or explicit recovery".into()); } - } else { - // resourceVersion makes this an atomic first-Pod binding. A - // concurrent deploy must retry and observe the winner's binding. - sandbox - .annotations_mut() - .insert(INITIAL_POD.into(), uid.into()); - sandbox = match sandboxes - .replace(&name, &PostParams::default(), &sandbox) - .await - { - Ok(bound) => bound, - Err(kube::Error::Api(error)) if error.reason == "Conflict" => { - sandbox = sandboxes - .get(&name) - .await - .map_err(|e| format!("refresh Sandbox binding: {e}"))?; - verify(&sandbox, identity, scope, owner, &fingerprint)?; - if annotation(&sandbox, GENERATION) != Some(&generation) { - return Err( - "Sandbox generation changed; explicit recovery is required".into(), - ); - } - if tokio::time::Instant::now() >= deadline { - return Err( - "Sandbox kept changing during credential binding; retry deploy" - .into(), - ); - } - continue; - } - Err(error) => return Err(format!("bind Sandbox to its first Pod: {error}")), - }; - env.insert("BUZZ_MANAGED_AGENT_START_NONCE".into(), generation.clone()); - env.insert("BUZZ_SANDBOX_INITIAL_POD_UID".into(), uid.into()); - let mut secret = pod::build_secret(identity, &cfg.namespace, &generation, env); - secret - .labels_mut() - .insert(naming::LABEL_BINDING_VERSION.into(), "2".into()); - secret.metadata.owner_references = Some(vec![OwnerReference { - api_version: "v1".into(), - kind: "Pod".into(), - name: name.clone(), - uid: uid.into(), - controller: Some(true), - block_owner_deletion: Some(false), - }]); - secrets - .create(&PostParams::default(), &secret) - .await - .map_err(|e| { + Err(e) => return Err(format!("bind Sandbox to its first Pod: {e}").into()), + }; + let secret = bound_secret(identity, cfg, &generation, name, uid, env); + session + .secrets + .create(&PostParams::default(), &secret) + .await + .map_err(|e| { + ProviderError::refused( + ErrorCode::BindingIncomplete, format!( "could not create bound Sandbox credentials ({}); explicit recovery is required", api_failure(&e) - ) - })?; - } - return wait_for_start(&pods, &sandbox, &name, uid).await; - } - if annotation(&sandbox, INITIAL_POD).is_some() { - return Err("Sandbox lost its original Pod; explicit recovery is required".into()); - } - if tokio::time::Instant::now() >= deadline { - return Err( - "Sandbox controller has not created its Pod; inspect the Sandbox and retry".into(), - ); + ), + ) + })?; + return wait_for_start(&session.pods, &sandbox, name, uid).await; + }; + if uid != initial_uid { + return Err(ProviderError::refused( + ErrorCode::PodReplaced, + "Sandbox Pod was replaced; credentials remain bound to the original session, explicit recovery is required", + )); } - tokio::time::sleep(Duration::from_secs(1)).await; + let secret = session + .secrets + .get_opt(&identity.secret_name(&generation)) + .await + .map_err(|e| format!("could not check Sandbox launch Secret: {}", api_failure(&e)))?; + let Some(secret) = secret else { + if tokio::time::Instant::now() >= deadline { + return Err(ProviderError::refused( + ErrorCode::BindingIncomplete, + "Sandbox credential binding is incomplete; explicit recovery is required", + )); + } + tokio::time::sleep(Duration::from_millis(250)).await; + continue; + }; + check_secret(&secret, identity, uid, &env)?; + return wait_for_start(&session.pods, &sandbox, name, uid).await; } } +/// Re-read a Sandbox after a lost CAS and re-check what the binding relies on. +async fn refresh(session: &Session<'_>, generation: &str) -> Result { + let sandbox = session + .sandboxes + .get(&session.name) + .await + .map_err(|e| format!("refresh Sandbox binding: {e}"))?; + verify_identity(&sandbox, session.identity, session.scope, session.owner)?; + verify_live(&sandbox, &session.fingerprint)?; + if annotation(&sandbox, GENERATION) != Some(generation) { + return Err(ProviderError::refused( + ErrorCode::SessionReplaced, + "Sandbox generation changed; explicit recovery is required", + )); + } + Ok(sandbox) +} + +fn bound_secret( + identity: &naming::AgentIdentity, + cfg: &ProviderConfig, + generation: &str, + name: &str, + uid: &str, + mut env: BTreeMap, +) -> Secret { + env.insert("BUZZ_MANAGED_AGENT_START_NONCE".into(), generation.into()); + env.insert("BUZZ_SANDBOX_INITIAL_POD_UID".into(), uid.into()); + let mut secret = pod::build_secret(identity, &cfg.namespace, generation, env); + secret.labels_mut().insert( + naming::LABEL_BINDING_VERSION.into(), + SANDBOX_BINDING_VERSION.into(), + ); + secret.metadata.owner_references = Some(vec![OwnerReference { + api_version: "v1".into(), + kind: "Pod".into(), + name: name.into(), + uid: uid.into(), + controller: Some(true), + block_owner_deletion: Some(false), + }]); + secret +} + +fn check_secret( + secret: &Secret, + identity: &naming::AgentIdentity, + uid: &str, + env: &BTreeMap, +) -> Result<(), ProviderError> { + if secret.immutable != Some(true) + || !secret + .owner_references() + .iter() + .any(|owner| owner.uid == uid && owner.kind == "Pod") + || secret + .annotations() + .get(naming::ANNOTATION_PUBKEY_FULL) + .map(String::as_str) + != Some(identity.pubkey_hex()) + { + return Err(ProviderError::refused( + ErrorCode::CredentialsMismatch, + "Sandbox credentials belong to a different Pod or agent", + )); + } + let saved_relay = secret + .data + .as_ref() + .and_then(|data| data.get("BUZZ_RELAY_URL")) + .map(|value| value.0.as_slice()); + if saved_relay != env.get("BUZZ_RELAY_URL").map(|value| value.as_bytes()) { + return Err(ProviderError::refused( + ErrorCode::RelayDiffers, + "Sandbox belongs to a different relay; use a new thread or explicit recovery", + )); + } + Ok(()) +} + async fn wait_for_start( pods: &Api, sandbox: &DynamicObject, name: &str, uid: &str, -) -> Result { +) -> Result { let deadline = tokio::time::Instant::now() + Duration::from_secs(120); loop { let child = pods .get_opt(name) .await .map_err(|e| format!("observe Sandbox startup: {e}"))? - .ok_or("Sandbox lost its Pod during startup; explicit recovery is required")?; + .ok_or_else(|| { + ProviderError::refused( + ErrorCode::PodLost, + "Sandbox lost its Pod during startup; explicit recovery is required", + ) + })?; if child.metadata.uid.as_deref() != Some(uid) || !owned_pod(&child, sandbox) || child.metadata.deletion_timestamp.is_some() { - return Err("Sandbox Pod changed during startup; explicit recovery is required".into()); + return Err(ProviderError::refused( + ErrorCode::PodReplaced, + "Sandbox Pod changed during startup; explicit recovery is required", + )); } let state = child .status @@ -512,7 +702,10 @@ async fn wait_for_start( .and_then(|containers| containers.iter().find(|c| c.name == "agent")) .and_then(|c| c.state.as_ref()); if state.is_some_and(|s| s.terminated.is_some()) { - return Err("Sandbox process ended during startup; inspect its retained logs".into()); + return Err(ProviderError::refused( + ErrorCode::PodEnded, + "Sandbox process ended during startup; inspect its retained logs", + )); } if state.is_some_and(|s| s.running.is_some()) { return Ok(name.into()); diff --git a/crates/buzz-backend-kubernetes/src/sandbox_fake.rs b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs new file mode 100644 index 00000000000..8928fbd14ee --- /dev/null +++ b/crates/buzz-backend-kubernetes/src/sandbox_fake.rs @@ -0,0 +1,201 @@ +//! A minimal in-memory apiserver for one namespace, driving the real +//! `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. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use http::{Method, Request, Response}; +use k8s_openapi::api::core::v1::Secret; +use k8s_openapi::ByteString; +use kube::client::Body; +use kube::Client; +use serde_json::{json, Value}; + +#[derive(Default)] +pub struct State { + pub sandboxes: BTreeMap, + pub secrets: BTreeMap, + /// Pod UIDs by name; Pods are rendered from their Sandbox on read. + pub pods: BTreeMap, + pub rv: u64, + pub uid: u64, +} + +pub type Shared = Arc>; + +fn status(code: u16, reason: &str) -> Response { + let body = json!({"kind":"Status","apiVersion":"v1","status":"Failure", + "message":reason,"reason":reason,"code":code}); + Response::builder() + .status(code) + .body(Body::from(serde_json::to_vec(&body).unwrap())) + .unwrap() +} + +fn ok(value: &Value) -> Response { + Response::builder() + .status(200) + .body(Body::from(serde_json::to_vec(value).unwrap())) + .unwrap() +} + +impl State { + fn next_rv(&mut self) -> String { + self.rv += 1; + self.rv.to_string() + } + + fn next_uid(&mut self, kind: &str) -> String { + self.uid += 1; + format!("{kind}-uid-{}", self.uid) + } + + /// The controller: a Running Sandbox has a Pod; Suspended has none. + fn reconcile(&mut self, name: &str) { + let running = self + .sandboxes + .get(name) + .is_some_and(|s| s["spec"]["operatingMode"] == "Running"); + if running && !self.pods.contains_key(name) { + let uid = self.next_uid("pod"); + self.pods.insert(name.into(), uid); + } else if !running { + self.pods.remove(name); + } + } + + fn pod(&self, name: &str) -> Option { + let uid = self.pods.get(name)?; + let sandbox = self.sandboxes.get(name)?; + let template = &sandbox["spec"]["podTemplate"]; + let secret = template["spec"]["containers"][0]["envFrom"][0]["secretRef"]["name"] + .as_str() + .unwrap_or_default(); + // `envFrom` names the generation's Secret; the container starts once + // it exists. + let started = self.secrets.contains_key(secret); + let state = if started { + json!({"running": {}}) + } else { + json!({"waiting": {"reason": "CreateContainerConfigError"}}) + }; + Some(json!({ + "apiVersion": "v1", "kind": "Pod", + "metadata": { + "name": name, "uid": uid, + "labels": template["metadata"]["labels"], + "annotations": template["metadata"]["annotations"], + "ownerReferences": [{"apiVersion": "agents.x-k8s.io/v1beta1", + "kind": "Sandbox", "name": name, + "uid": sandbox["metadata"]["uid"], "controller": true}], + }, + "status": {"phase": "Pending", "containerStatuses": [{ + "name": "agent", "image": "i", "imageID": "", "ready": started, + "restartCount": 0, "state": state}]}, + })) + } + + fn handle(&mut self, method: &Method, path: &str, body: &[u8]) -> Response { + let path = path.split('?').next().unwrap_or_default(); + let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect(); + match segments.as_slice() { + ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", _, "sandboxes", rest @ ..] => { + self.sandbox(method, rest.first().copied(), body) + } + ["api", "v1", "namespaces", _, "pods"] => { + let names: Vec = self.pods.keys().cloned().collect(); + 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", _, "secrets", name] => match self.secrets.get(*name) { + Some(secret) => ok(secret), + None => status(404, "NotFound"), + }, + ["api", "v1", "namespaces", _, "secrets"] if method == Method::POST => { + let mut secret: Secret = serde_json::from_slice(body).unwrap(); + let name = secret.metadata.name.clone().unwrap(); + if self.secrets.contains_key(&name) { + return status(409, "AlreadyExists"); + } + secret.data = secret.string_data.take().map(|data| { + data.into_iter() + .map(|(k, v)| (k, ByteString(v.into_bytes()))) + .collect() + }); + secret.metadata.resource_version = Some(self.next_rv()); + secret.metadata.uid = Some(self.next_uid("secret")); + let stored = serde_json::to_value(&secret).unwrap(); + self.secrets.insert(name, stored.clone()); + ok(&stored) + } + _ => status(404, "NotFound"), + } + } + + fn sandbox(&mut self, method: &Method, name: Option<&str>, body: &[u8]) -> Response { + match (method.clone(), name) { + (Method::GET, Some(name)) => match self.sandboxes.get(name) { + Some(sandbox) => ok(sandbox), + None => status(404, "NotFound"), + }, + (Method::POST, None) => { + 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) { + return status(409, "AlreadyExists"); + } + sandbox["metadata"]["uid"] = self.next_uid("sandbox").into(); + sandbox["metadata"]["resourceVersion"] = self.next_rv().into(); + self.sandboxes.insert(name.clone(), sandbox.clone()); + self.reconcile(&name); + ok(&sandbox) + } + (Method::PUT, Some(name)) => { + let mut sandbox: Value = serde_json::from_slice(body).unwrap(); + let Some(current) = self.sandboxes.get(name) else { + return status(404, "NotFound"); + }; + if current["metadata"]["resourceVersion"] != sandbox["metadata"]["resourceVersion"] + { + return status(409, "Conflict"); + } + sandbox["metadata"]["resourceVersion"] = self.next_rv().into(); + self.sandboxes.insert(name.into(), sandbox.clone()); + self.reconcile(name); + ok(&sandbox) + } + _ => status(405, "MethodNotAllowed"), + } + } +} + +/// A client backed by `state`. +pub fn client(state: &Shared) -> Client { + let state = Arc::clone(state); + let service = tower::service_fn(move |request: Request| { + let state = Arc::clone(&state); + async move { + let (parts, body) = request.into_parts(); + let bytes = http_body_util::BodyExt::collect(body) + .await + .map(|b| b.to_bytes().to_vec()) + .unwrap_or_default(); + let response = + state + .lock() + .unwrap() + .handle(&parts.method, &parts.uri.to_string(), &bytes); + Ok::<_, std::convert::Infallible>(response) + } + }); + Client::new(service, "agents") +} diff --git a/crates/buzz-backend-kubernetes/src/sandbox_tests.rs b/crates/buzz-backend-kubernetes/src/sandbox_tests.rs index 01755e580fa..50f43add445 100644 --- a/crates/buzz-backend-kubernetes/src/sandbox_tests.rs +++ b/crates/buzz-backend-kubernetes/src/sandbox_tests.rs @@ -1,6 +1,10 @@ use super::*; +use buzz_backend_kubernetes::lifecycle; use nostr::nips::nip19::ToBech32; +#[path = "sandbox_fake.rs"] +mod fake; + #[test] fn custom_image_is_inside_a_real_sandbox_with_a_downward_uid() { let identity = naming::AgentIdentity::from_nsec( @@ -14,7 +18,11 @@ fn custom_image_is_inside_a_real_sandbox_with_a_downward_uid() { thread_root: "c".repeat(64), }; let fp = pod::intent_template(&cfg, vec![]).fingerprint(); - let mut sandbox = build(&identity, &cfg, &scope, "owner", "generation", &fp).unwrap(); + let new = NewSession { + generation: "generation", + checkpointing: false, + }; + let mut sandbox = build(&identity, &cfg, &scope, "owner", &new, &fp).unwrap(); assert_eq!( sandbox.types.as_ref().unwrap().api_version, "agents.x-k8s.io/v1beta1" @@ -34,11 +42,14 @@ fn custom_image_is_inside_a_real_sandbox_with_a_downward_uid() { crate::config::TERMINATION_GRACE_SECONDS ); assert_eq!(spec["volumes"][0]["emptyDir"]["sizeLimit"], "50Gi"); - verify(&sandbox, &identity, &scope, "owner", &fp).unwrap(); + assert_eq!(spec["containers"][0]["terminationMessagePolicy"], "File"); + assert_eq!(annotation(&sandbox, CHECKPOINTING), Some("disabled")); + verify_identity(&sandbox, &identity, &scope, "owner").unwrap(); + verify_live(&sandbox, &fp).unwrap(); sandbox .annotations_mut() .insert(OWNER.into(), "different-owner".into()); - assert!(verify(&sandbox, &identity, &scope, "owner", &fp).is_err()); + assert!(verify_identity(&sandbox, &identity, &scope, "owner").is_err()); } #[test] @@ -295,17 +306,20 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { .is_err()); let mut other_relay = fixture_env(); other_relay.insert("BUZZ_RELAY_URL".into(), "wss://other.invalid".into()); - assert!(deploy( - client.clone(), - &identity, - &cfg, - &scope, - "owner", - other_relay - ) - .await - .unwrap_err() - .contains("different relay")); + assert_eq!( + deploy( + client.clone(), + &identity, + &cfg, + &scope, + "owner", + other_relay + ) + .await + .unwrap_err() + .code, + Some(ErrorCode::RelayDiffers) + ); let second_scope = Options { channel_id: scope.channel_id.clone(), thread_root: "b".repeat(64), @@ -357,7 +371,7 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { ) .await .unwrap_err(); - assert!(error.contains("replaced"), "{error}"); + assert_eq!(error.code, Some(ErrorCode::PodReplaced), "{error}"); let sandboxes = api(client, &namespace) .list(&ListParams::default()) .await @@ -368,3 +382,217 @@ async fn live_controller_reuses_scopes_and_refuses_lost_workspaces() { .await .unwrap(); } + +// ── Fake-apiserver tests: the real deploy paths over kube::Client ─────────── + +fn fake_identity() -> naming::AgentIdentity { + naming::AgentIdentity::from_nsec(&nostr::Keys::generate().secret_key().to_bech32().unwrap()) + .unwrap() +} + +fn fake_cfg() -> ProviderConfig { + crate::config::parse(&json!({"namespace":"agents", + "image":format!("example.com/agent@sha256:{}", "a".repeat(64)), + "pod_options":{"active_deadline_seconds":3600}})) + .unwrap() +} + +fn fake_scope() -> Options { + Options { + channel_id: "0b6c1f3e-5a7d-4c2b-9e8f-1a2b3c4d5e6f".into(), + thread_root: "d".repeat(64), + } +} + +fn plain_env() -> BTreeMap { + BTreeMap::from([ + ("BUZZ_RELAY_URL".into(), "wss://fixture.invalid".into()), + ("BUZZ_CHECKPOINT_BUCKET".into(), "bucket".into()), + ("BUZZ_CHECKPOINT_PREFIX".into(), "dev/".into()), + ]) +} + +fn secret_env(state: &fake::Shared, key: &str) -> Vec { + state + .lock() + .unwrap() + .secrets + .values() + .map(|secret| { + let secret: Secret = serde_json::from_value(secret.clone()).unwrap(); + String::from_utf8(secret.data.unwrap()[key].0.clone()).unwrap() + }) + .collect() +} + +fn sandbox_json(state: &fake::Shared, name: &str) -> serde_json::Value { + state.lock().unwrap().sandboxes[name].clone() +} + +#[tokio::test(start_paused = true)] +async fn fresh_deploy_binds_and_reuses_its_session() { + let state = fake::Shared::default(); + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let name = deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env(), + ) + .await + .unwrap(); + let sandbox = sandbox_json(&state, &name); + let annotations = &sandbox["metadata"]["annotations"]; + assert_eq!(annotations[CHECKPOINTING], "enabled"); + let initial_pod_uid = annotations[INITIAL_POD].as_str().unwrap(); + assert_eq!( + secret_env(&state, "BUZZ_SANDBOX_INITIAL_POD_UID"), + [initial_pod_uid] + ); + let secret_owner_uids: Vec<_> = state + .lock() + .unwrap() + .secrets + .values() + .map(|secret| { + secret["metadata"]["ownerReferences"][0]["uid"] + .as_str() + .unwrap() + .to_owned() + }) + .collect(); + assert_eq!(secret_owner_uids, [initial_pod_uid]); + assert_eq!( + secret_env(&state, "BUZZ_MANAGED_AGENT_START_NONCE"), + [annotations[GENERATION].as_str().unwrap()] + ); + // A repeat deploy reuses the session. + assert_eq!( + deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env() + ) + .await + .unwrap(), + name + ); + assert_eq!(state.lock().unwrap().secrets.len(), 1); +} + +/// Every identity and liveness refusal for an existing Sandbox, through the +/// real deploy path. A refused deploy writes nothing: no Secret, and the +/// Sandbox (including its resourceVersion) is unchanged. +#[tokio::test(start_paused = true)] +async fn deploy_refuses_a_sandbox_that_is_not_a_reusable_live_session() { + type Mutate = fn(&mut serde_json::Value); + fn set(sandbox: &mut serde_json::Value, key: &str, value: &str) { + sandbox["metadata"]["annotations"][key] = value.into(); + } + let cases: [(&str, Mutate, ErrorCode, &str); 10] = [ + ( + "ended", + |s| { + set(s, lifecycle::LIFECYCLE, lifecycle::LIFECYCLE_ENDED); + set(s, ENDED_REASON, "lost"); + }, + ErrorCode::SessionEnded, + "Sandbox session ended (lost); explicit recovery is required", + ), + ( + "ended with an unknown reason", + |s| { + set(s, lifecycle::LIFECYCLE, lifecycle::LIFECYCLE_ENDED); + set(s, ENDED_REASON, "see https://evil.invalid"); + }, + ErrorCode::SessionEnded, + "Sandbox session ended; explicit recovery is required", + ), + ( + "ending", + |s| set(s, lifecycle::LIFECYCLE, lifecycle::LIFECYCLE_ENDING), + ErrorCode::SessionEnding, + "Sandbox session is ending; recover it explicitly once it has ended", + ), + ( + "unknown lifecycle", + |s| set(s, lifecycle::LIFECYCLE, "paused"), + ErrorCode::LifecycleUnknown, + "Sandbox has a lifecycle this provider does not know; update Buzz", + ), + ( + "deletionTimestamp", + |s| s["metadata"]["deletionTimestamp"] = "2026-01-01T00:00:00Z".into(), + ErrorCode::SessionDeleting, + "Sandbox is being deleted; retry once it is gone", + ), + ( + "Suspended without lifecycle", + |s| s["spec"]["operatingMode"] = "Suspended".into(), + ErrorCode::SessionNotRunning, + "Sandbox is not an active managed session; explicit recovery is required", + ), + ( + "not Running", + |s| s["spec"]["operatingMode"] = "Hibernated".into(), + ErrorCode::SessionNotRunning, + "Sandbox is not an active managed session; explicit recovery is required", + ), + ( + "binding-version mismatch", + |s| s["metadata"]["labels"][naming::LABEL_BINDING_VERSION] = "1".into(), + ErrorCode::IdentityMismatch, + "Sandbox belongs to a different agent, owner, or thread", + ), + ( + "fingerprint mismatch", + |s| set(s, naming::ANNOTATION_CREATE_INTENT, &"0".repeat(64)), + ErrorCode::ConfigDiffers, + "Sandbox launch configuration differs; end the session, then explicit recovery is required", + ), + ( + "owner mismatch", + |s| set(s, OWNER, "other-owner"), + ErrorCode::IdentityMismatch, + "Sandbox belongs to a different agent, owner, or thread", + ), + ]; + let (identity, cfg, scope) = (fake_identity(), fake_cfg(), fake_scope()); + let name = scope.name(&identity); + for (case, mutate, code, message) in cases { + let state = fake::Shared::default(); + let redeploy = || { + deploy( + fake::client(&state), + &identity, + &cfg, + &scope, + "owner", + plain_env(), + ) + }; + redeploy().await.unwrap(); + let before = { + let mut guard = state.lock().unwrap(); + // Without its Secret, an accepted Sandbox would be written to. + guard.secrets.clear(); + let sandbox = guard.sandboxes.get_mut(&name).unwrap(); + mutate(sandbox); + sandbox.clone() + }; + let error = redeploy().await.unwrap_err(); + assert_eq!( + (error.code, error.message.as_str()), + (Some(code), message), + "{case}" + ); + let guard = state.lock().unwrap(); + assert!(guard.secrets.is_empty(), "{case}: wrote a Secret"); + assert_eq!(guard.sandboxes[&name], before, "{case}: wrote the Sandbox"); + } +} diff --git a/crates/buzz-backend-kubernetes/src/wire.rs b/crates/buzz-backend-kubernetes/src/wire.rs index 89a81aece4a..046a2139821 100644 --- a/crates/buzz-backend-kubernetes/src/wire.rs +++ b/crates/buzz-backend-kubernetes/src/wire.rs @@ -126,13 +126,103 @@ pub struct IdentityPreflightResponse { pub struct ErrorResponse { pub ok: bool, pub error: String, + /// Present only for a refusal a client can act on. Additive within + /// protocol version 1: older clients ignore it. + #[serde(skip_serializing_if = "Option::is_none")] + pub code: Option, +} + +/// Stable, machine-readable reason for a refusal (spec §Sandbox error +/// codes). Clients branch on this, never on the `error` text. A code is never +/// renamed or reused; new refusals get new codes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ErrorCode { + /// The session is a tombstone; only an explicit recovery starts another. + SessionEnded, + /// A stop is draining; the session ends once its Pod is gone. + SessionEnding, + /// The lifecycle annotation has a value this provider does not know. + LifecycleUnknown, + /// The Sandbox is being deleted. + SessionDeleting, + /// The Sandbox is not Running and has no lifecycle record. + SessionNotRunning, + /// The live session was launched with a different configuration. + ConfigDiffers, + /// The Sandbox belongs to a different agent, owner, or thread, or lacks + /// the management marker or binding version. + IdentityMismatch, + /// The session's credentials were bound to a different Pod or agent. + CredentialsMismatch, + /// The session was launched against a different relay. + RelayDiffers, + /// The session is bound to its first Pod, but its credentials were not + /// written. + BindingIncomplete, + /// The Pod the session is bound to is gone. + PodLost, + /// The controller replaced the Pod the session is bound to. + PodReplaced, + /// The bound Pod finished, is deleting, or its process exited. + PodEnded, + /// The Pod with the Sandbox's name is not controlled by it. + ForeignPod, + /// A different session replaced this Sandbox during the deploy. + SessionReplaced, + /// The agent already has a Pod outside a Sandbox session. + AgentPodExists, +} + +/// A failed operation: the display message and, for a refusal a client can +/// act on, its [`ErrorCode`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProviderError { + pub code: Option, + pub message: String, +} + +impl ProviderError { + pub fn refused(code: ErrorCode, message: impl Into) -> Self { + Self { + code: Some(code), + message: message.into(), + } + } +} + +impl From for ProviderError { + fn from(message: String) -> Self { + Self { + code: None, + message, + } + } +} + +impl From<&str> for ProviderError { + fn from(message: &str) -> Self { + message.to_string().into() + } +} + +impl std::fmt::Display for ProviderError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } } impl Response { pub fn error(message: impl Into) -> Self { + Self::failed(ProviderError::from(message.into())) + } + + /// The in-band failure for `error`, carrying its code when it has one. + pub fn failed(error: ProviderError) -> Self { Response::Error(ErrorResponse { ok: false, - error: message.into(), + error: error.message, + code: error.code, }) } @@ -269,7 +359,21 @@ mod tests { assert_eq!(v["preflight"], "identity"); let v = serde_json::to_value(Response::error("boom")).unwrap(); - assert_eq!(v["ok"], false); - assert_eq!(v["error"], "boom"); + assert_eq!(v, serde_json::json!({"ok": false, "error": "boom"})); + } + + /// A refusal's code is a stable snake_case string next to `error`; an + /// uncoded failure omits the field, so protocol-v1 clients see no change. + #[test] + fn refusal_codes_serialize_snake_case() { + let v = serde_json::to_value(Response::failed(ProviderError::refused( + ErrorCode::SessionNotRunning, + "not running", + ))) + .unwrap(); + assert_eq!( + v, + serde_json::json!({"ok": false, "error": "not running", "code": "session_not_running"}) + ); } } diff --git a/docs/remote-agents.md b/docs/remote-agents.md index c823402b51d..a493f36868a 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -373,7 +373,10 @@ JSON object to stdout and exits. Requirements on `D` (all implemented): `try_wait`. - **Non-zero exit is failure even if stdout parsed.** Partial output from a crashed operation is never trusted. -- `{"ok": false, "error": …}` is the in-band failure form. +- `{"ok": false, "error": …}` is the in-band failure form. A refusal the + desktop can act on also carries `"code": ""` (§Sandbox error + codes). `D` branches on `code`, never on the `error` text; a response + without `code` is a generic failure. - **Environment**: `P` inherits `D`'s environment. On macOS a GUI launch means launchd's minimal PATH; providers whose substrate credentials invoke helper binaries (kubeconfig `exec` plugins) MUST self-augment their PATH @@ -1444,7 +1447,8 @@ The controller owns the Pod. Concurrent mentions reuse its workspace when the ag The provider binds the launch Secret to the first Pod UID. A replaced or completed Pod requires explicit recovery; a later mention cannot silently create an empty replacement session. The workspace is an `emptyDir` on a `restartPolicy: Never` Pod. When the harness exits, the stop checkpoint is the only copy of its work. -Deleting a Sandbox also removes its Pod and any workspace that remains. Recover or checkpoint authorized work before deletion. +A session that ended is kept as a tombstone, not deleted (§Sandbox lifecycle). +Deleting a Sandbox removes its Pod and any workspace that remains. Recover or checkpoint authorized work before deletion. Sandbox ingress accepts only its owner's signed messages for the assigned channel thread. The provider enables JSONL activity and disables autonomous heartbeat work. @@ -1466,6 +1470,88 @@ Later deploys for that thread wait 120s for the Secret and then fail with "Sandb To recover, delete the Sandbox. This discards the workspace; the container never started, so it contains no agent work. Then mention the agent in the thread again. A later layer of this stack makes the binding resumable, so a retried deploy creates the missing Secret for the same Pod. +### Sandbox lifecycle + +The in-cluster `buzz-sandbox-reaper` (planned; it ships in a later change) +ends Sandbox sessions that stopped, failed, or lost their Pod. It never +starts a session, never sets `operatingMode: Running`, and never creates a +Sandbox. Only an explicit recovery deploy from the desktop starts a new +session for a thread. This section is the contract the reaper implements; +the provider already refuses to reuse a session that is `ending`, `ended`, +or in an unknown lifecycle. + +The provider and the reaper share one classifier and one set of annotation +keys (`buzz-backend-kubernetes` library, `lifecycle.rs`): + +| Annotation | Values | +| --- | --- | +| `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/recovered-from` | the generation of the tombstone that a recovered session replaced | + +The reaper classifies each Sandbox on every pass. All timers use the +apiserver `Date` header, not the reaper's clock. + +| State | Detection | Reaper action | +| --- | --- | --- | +| Binding | Running, no first-Pod binding, Sandbox younger than 1200s | none | +| BindAbandoned | as Binding, Sandbox 1200s or older | tombstone `bind-abandoned` | +| BindingIncomplete | bound to this Pod, container never started | none before the Pod is 1200s old (planned: a deploy completes the binding); then tombstone `binding-failed` | +| 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`. Checkpoint missing: Warning Event, hold 24h, then tombstone with `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 | + +A tombstone is one JSON merge patch that carries the observed +`metadata.resourceVersion`. It sets `spec.operatingMode: Suspended` and the +lifecycle, reason, ended-at, and checkpoint annotations. The controller then +deletes the Pod, and Kubernetes garbage collection deletes the Pod-owned +launch Secret. If the patch conflicts, the reaper reads the Sandbox again +and classifies it again; it does not retry the old decision. Deletion of an +expired tombstone has uid and resourceVersion preconditions. + +The reaper touches only Sandboxes that have the management label, binding +version `2`, and an owner annotation equal to the owner that its +configuration lists for the namespace. It records a Warning Event for a +managed Sandbox that it leaves alone (foreign owner, foreign Pod, +inconsistent lifecycle). + +### Sandbox error codes + +A Sandbox deploy refusal carries a stable `code` next to its `error` text. +The codes are additive within protocol version 1: a code is never renamed +or reused, and a new refusal gets a new code. They are defined in +`crates/buzz-backend-kubernetes/src/wire.rs` (`ErrorCode`). + +| `code` | Meaning | Desktop action | +| --- | --- | --- | +| `session_ended` | The Sandbox is a tombstone (`lifecycle=ended`). | explicit recovery | +| `session_ending` | A stop is draining (`lifecycle=ending`). | wait, then explicit recovery | +| `lifecycle_unknown` | The lifecycle annotation has a value this provider does not know. | update Buzz | +| `session_deleting` | The Sandbox has a `deletionTimestamp`. | retry once it is gone | +| `session_not_running` | `operatingMode` is not `Running` and there is no lifecycle record. | explicit recovery | +| `config_differs` | The live session's launch fingerprint differs from this deploy. | end the session, then explicit recovery | +| `identity_mismatch` | The Sandbox belongs to a different agent, owner, or thread, or lacks the management marker or binding version `2`. | none (use a new thread) | +| `credentials_mismatch` | The launch Secret is bound to a different Pod or agent. | explicit recovery | +| `relay_differs` | The session was launched against a different relay. | new thread or explicit recovery | +| `binding_incomplete` | The session is bound to its first Pod, but its launch Secret was not written. | explicit recovery | +| `pod_lost` | The Pod the session is bound to is gone. | explicit recovery | +| `pod_replaced` | The controller replaced the bound Pod. | explicit recovery | +| `pod_ended` | The bound Pod finished, is deleting, or its process exited during startup. | explicit recovery | +| `foreign_pod` | The Pod with the Sandbox's name is not controlled by it. | operator cleanup | +| `session_replaced` | Another session replaced the Sandbox during this deploy. | explicit recovery | +| `agent_pod_exists` | The agent already has a Pod outside a Sandbox session. | stop it, then explicit recovery | + +The `session_ended` text names the ended reason only when it is one of the +spellings in §Sandbox lifecycle; any other annotation value is not repeated. +Transient and configuration failures (apiserver errors, timeouts) carry no +`code`. + ### Distribution Its own release workflow (macOS arm64/x64 + Linux musl; the sprig workflow's