feat(agents): add an explicit stop op for Sandbox sessions

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

@@ -99,22 +99,49 @@ fn respond(input: &str) -> Response {
 
     match request {
         Request::Info => Response::info(),
-        Request::Deploy(deploy) => {
-            let runtime = match tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-            {
-                Ok(runtime) => runtime,
-                Err(e) => return Response::error(format!("could not start the runtime: {e}")),
-            };
-            match runtime.block_on(deploy_agent(&deploy)) {
+        Request::Deploy(deploy) => match runtime() {
+            Ok(runtime) => match runtime.block_on(deploy_agent(&deploy)) {
                 Ok(agent_id) => Response::deployed(agent_id),
                 Err(e) => Response::failed(e),
-            }
-        }
+            },
+            Err(e) => Response::error(e),
+        },
+        Request::Stop(stop) => match runtime() {
+            Ok(runtime) => match runtime.block_on(stop_session(&stop)) {
+                Ok((agent_id, state)) => Response::stopped(agent_id, state),
+                Err(e) => Response::failed(e),
+            },
+            Err(e) => Response::error(e),
+        },
     }
 }
 
+fn runtime() -> Result<tokio::runtime::Runtime, String> {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .map_err(|e| format!("could not start the runtime: {e}"))
+}
+
+/// End one Sandbox session. Refusals (identity, scope) precede any cluster
+/// contact, exactly as for `deploy`.
+async fn stop_session(
+    request: &wire::DeployRequest,
+) -> Result<(String, sandbox::StopState), wire::ProviderError> {
+    let cfg = config::parse(&request.provider_config)?;
+    let scope = sandbox::Options::parse(&request.provider_config)?
+        .ok_or("stop requires provider_config.sandbox")?;
+    let identity = identity::check(request, true)?;
+    let owner = request
+        .agent
+        .launch
+        .as_ref()
+        .and_then(|launch| launch.owner_pubkey.as_deref())
+        .ok_or("Sandbox requires a verified launch owner")?;
+    let client = client::connect(cfg.context.as_deref()).await?;
+    sandbox::stop(client, &identity, &cfg.namespace, &scope, owner).await
+}
+
 /// Refuse a shared-compute agent, reading the **raw wire value**.
 ///
 /// Trimmed before comparing: the desktop's own layers disagree about padding
@@ -226,6 +253,27 @@ mod tests {
         assert_eq!(json["ok"], true);
         assert_eq!(json["protocol_version"], wire::PROTOCOL_VERSION);
         assert!(json["config_schema"]["properties"]["namespace"].is_object());
+        assert!(json.get("ops").is_none());
+    }
+
+    /// `stop` is gated like a preflight: no approved identity policy, no
+    /// cluster contact.
+    #[test]
+    fn stop_requires_sandbox_scope_and_identity_policy_before_cluster_contact() {
+        use nostr::nips::nip19::ToBech32;
+        let agent = nostr::Keys::generate();
+        let mut request = serde_json::json!({
+            "op": "stop",
+            "agent": {"relay_url": "wss://example.invalid",
+                "private_key_nsec": agent.secret_key().to_bech32().unwrap()},
+            "provider_config": {"namespace": "test",
+                "image": format!("example.invalid/pilot@sha256:{}", "a".repeat(64))}
+        });
+        assert!(error_of(&respond(&request.to_string())).contains("provider_config.sandbox"));
+        request["provider_config"]["sandbox"] = serde_json::json!({
+            "channel_id": uuid::Uuid::new_v4().to_string(), "thread_root": "a".repeat(64)
+        });
+        assert!(error_of(&respond(&request.to_string())).contains("identity_policy"));
     }
 
     #[test]

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

@@ -12,16 +12,17 @@ use std::time::Duration;
 use buzz_backend_kubernetes::lifecycle::{
     annotation, owned_pod, sandbox_resource, EndedReason, Lifecycle, OperatingMode,
     BIND_TIMEOUT_SECS, CHANNEL, CHECKPOINTING, CHECKPOINTING_DISABLED, CHECKPOINTING_ENABLED,
-    ENDED_REASON, GENERATION, INITIAL_POD, OWNER, RESTORE_CHECKPOINT, SANDBOX_BINDING_VERSION,
-    SESSION, THREAD,
+    ENDED_REASON, GENERATION, INITIAL_POD, LIFECYCLE, LIFECYCLE_ENDING, OWNER, RESTORE_CHECKPOINT,
+    SANDBOX_BINDING_VERSION, SESSION, THREAD,
 };
 use chrono::{DateTime, Utc};
 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::api::{Api, ListParams, Patch, PatchParams, PostParams};
 use kube::core::DynamicObject;
 use kube::{Client, ResourceExt};
-use serde::Deserialize;
+use rand::RngExt;
+use serde::{Deserialize, Serialize};
 use serde_json::json;
 use sha2::{Digest, Sha256};
 
@@ -817,6 +818,86 @@ async fn wait_for_start(
     }
 }
 
+/// What `stop` left behind; the `state` of its response.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum StopState {
+    /// Suspended and draining; the reaper writes the tombstone.
+    Ending,
+    /// Already a tombstone.
+    Ended,
+    /// No Sandbox exists for this thread.
+    Absent,
+}
+
+/// Read-and-patch attempts before `stop` gives up on a changing Sandbox.
+const STOP_ATTEMPTS: u32 = 5;
+/// Base pause after a conflicting `stop` patch; grows per attempt.
+const STOP_BACKOFF_MS: u64 = 100;
+
+/// End a (possibly wedged) session: one CAS merge patch sets
+/// `lifecycle=ending`, reason `stopped`, and `operatingMode: Suspended`. The
+/// controller then deletes the Pod (SIGTERM, so the harness still gets its
+/// checkpoint window). The normal stop remains `!shutdown`.
+pub async fn stop(
+    client: Client,
+    identity: &naming::AgentIdentity,
+    namespace: &str,
+    scope: &Options,
+    owner: &str,
+) -> Result<(String, StopState), ProviderError> {
+    let sandboxes = api(client, namespace);
+    let name = scope.name(identity);
+    for attempt in 1..=STOP_ATTEMPTS {
+        let Some(sandbox) = sandboxes
+            .get_opt(&name)
+            .await
+            .map_err(|e| format!("read Sandbox: {e}"))?
+        else {
+            return Ok((name, StopState::Absent));
+        };
+        verify_identity(&sandbox, identity, scope, owner)?;
+        match Lifecycle::of(&sandbox) {
+            Lifecycle::Ended => return Ok((name, StopState::Ended)),
+            Lifecycle::Ending => return Ok((name, StopState::Ending)),
+            // Never overwrite a record this version does not understand.
+            Lifecycle::Unrecognized => {
+                return Err(ProviderError::refused(
+                    ErrorCode::LifecycleUnknown,
+                    "Sandbox has a lifecycle this provider does not know; update Buzz",
+                ))
+            }
+            Lifecycle::Live => {}
+        }
+        let patch = json!({
+            "metadata": {
+                "resourceVersion": sandbox.metadata.resource_version,
+                "annotations": {
+                    (LIFECYCLE): LIFECYCLE_ENDING,
+                    (ENDED_REASON): EndedReason::Stopped.as_str(),
+                },
+            },
+            "spec": {"operatingMode": "Suspended"},
+        });
+        match sandboxes
+            .patch(&name, &PatchParams::default(), &Patch::Merge(&patch))
+            .await
+        {
+            Ok(_) => return Ok((name, StopState::Ending)),
+            Err(e) if reason_is(&e, "Conflict") => {
+                if attempt < STOP_ATTEMPTS {
+                    // Jittered, so concurrent writers stop colliding.
+                    let jitter = rand::rng().random_range(0..=STOP_BACKOFF_MS);
+                    let pause = STOP_BACKOFF_MS * u64::from(attempt) + jitter;
+                    tokio::time::sleep(Duration::from_millis(pause)).await;
+                }
+            }
+            Err(e) => return Err(format!("stop Sandbox: {e}").into()),
+        }
+    }
+    Err("Sandbox kept changing while stopping; retry".into())
+}
+
 #[cfg(test)]
 #[path = "sandbox_tests.rs"]
 mod tests;

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

@@ -35,6 +35,10 @@ pub struct State {
     pub fail_secret_creates: usize,
     /// Store the Secret, then fail this many create responses with a 500.
     pub fail_secret_create_responses: usize,
+    /// Answer this many Sandbox patches with a 409 Conflict.
+    pub conflict_patches: usize,
+    /// Every merge patch body received.
+    pub patches: Vec<Value>,
 }
 
 pub type Shared = Arc<Mutex<State>>;
@@ -55,6 +59,21 @@ fn ok(value: &Value) -> Response<Body> {
         .unwrap()
 }
 
+fn merge(target: &mut Value, patch: &Value) {
+    match (target, patch) {
+        (Value::Object(target), Value::Object(patch)) => {
+            for (key, value) in patch {
+                if value.is_null() {
+                    target.remove(key);
+                } else {
+                    merge(target.entry(key.clone()).or_insert(Value::Null), value);
+                }
+            }
+        }
+        (target, patch) => *target = patch.clone(),
+    }
+}
+
 impl State {
     fn next_rv(&mut self) -> String {
         self.rv += 1;
@@ -202,6 +221,27 @@ impl State {
                 self.reconcile(name);
                 ok(&sandbox)
             }
+            (Method::PATCH, Some(name)) => {
+                let patch: Value = serde_json::from_slice(body).unwrap();
+                self.patches.push(patch.clone());
+                if self.conflict_patches > 0 {
+                    self.conflict_patches -= 1;
+                    return status(409, "Conflict");
+                }
+                let rv = self.next_rv();
+                let Some(current) = self.sandboxes.get_mut(name) else {
+                    return status(404, "NotFound");
+                };
+                let expected = &patch["metadata"]["resourceVersion"];
+                if !expected.is_null() && *expected != current["metadata"]["resourceVersion"] {
+                    return status(409, "Conflict");
+                }
+                merge(current, &patch);
+                current["metadata"]["resourceVersion"] = rv.into();
+                let updated = current.clone();
+                self.reconcile(name);
+                ok(&updated)
+            }
             _ => status(405, "MethodNotAllowed"),
         }
     }