diff --git a/crates/buzz-acp/TESTING.md b/crates/buzz-acp/TESTING.md index 528bac21384..6d63ff3adb4 100644 --- a/crates/buzz-acp/TESTING.md +++ b/crates/buzz-acp/TESTING.md @@ -21,6 +21,21 @@ Make sure `pi` and `buzz-pi-acp` are on PATH, then restart Buzz. cargo test -p buzz-acp ``` +The package suite also checks owner-only thread routing, ACP full-access negotiation, +redacted JSONL activity, and checkpoint failure propagation. Activity tests use the +real ACP client with a synthetic adapter and assert that test secrets never enter +the platform stream. Checkpoint tests cover validated receipts, command failures, +invalid and oversized output, missing helpers, and timeouts, and check that each +failure records its reason. `tests/checkpoint_stop.rs` stops the real harness +binary with a stub helper on `PATH` and a local fake relay. It checks that a +normal stop checkpoints after adapters stop and presence is `offline`, that +cancelling or failing startup still checkpoints once, and that a failed +checkpoint exits non-zero. Startup failure remains the primary error even when +the checkpoint also fails; activity reports the distinct `startup_failed` reason. +The image repository's `buzz-agent/test-checkpoint-storage.py` also stops the +packaged harness during startup while an adapter writes files. It checks that +recovery preserves the final edit and rejected uploads make shutdown fail. + Managed agent sessions may already export harness options. Clear them when running the package suite: three CLI parsing tests assert the unset defaults, and inherited values would change the inputs those tests exercise. Running the diff --git a/crates/buzz-acp/src/activity.rs b/crates/buzz-acp/src/activity.rs new file mode 100644 index 00000000000..4bf352c298b --- /dev/null +++ b/crates/buzz-acp/src/activity.rs @@ -0,0 +1,399 @@ +//! Bounded, content-minimized activity records for platform log collection. +//! +//! The encrypted observer feed retains its original payload. The platform feed +//! records operation metadata and output sizes; prompts, raw results, credentials, +//! and arbitrary adapter-provided titles never enter this stream. + +use std::collections::HashMap; +use std::time::Instant; + +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use crate::observer::ObserverEvent; + +const MAX_ACTIVE_TOOLS: usize = 256; + +pub(crate) struct ActivityRecorder { + source_id: String, + sequence: u64, + tools: HashMap<(String, String), (Instant, Option)>, +} + +impl ActivityRecorder { + pub(crate) fn new() -> Self { + Self { + source_id: uuid::Uuid::new_v4().to_string(), + sequence: 0, + tools: HashMap::new(), + } + } + + pub(crate) fn record(&mut self, event: &ObserverEvent) -> Option { + let mut details = json!({}); + let kind = match event.kind.as_str() { + "checkpoint_started" => event.kind.as_str(), + "checkpoint_failed" => { + details["reason"] = event.payload["reason"].as_str().into(); + details["exit_code"] = event.payload["exit_code"].as_i64().into(); + event.kind.as_str() + } + "checkpoint_saved" => { + details = event.payload.clone(); + event.kind.as_str() + } + "harness_started" | "harness_stopped" | "turn_started" | "turn_completed" => { + if event.kind == "harness_stopped" { + if let Some(reason @ ("startup_failed" | "startup_cancelled")) = + event.payload["reason"].as_str() + { + details["reason"] = reason.into(); + } + details["unfinished_tools"] = self.tools.len().into(); + self.tools.clear(); + } else if event.kind == "turn_completed" { + let before = self.tools.len(); + self.tools.retain(|_, (_, turn)| turn != &event.turn_id); + details["unfinished_tools"] = (before - self.tools.len()).into(); + } + event.kind.as_str() + } + "acp_read" if event.payload["method"] == "session/update" => { + let update = &event.payload["params"]["update"]; + let update_kind = update["sessionUpdate"].as_str()?; + if !matches!(update_kind, "tool_call" | "tool_call_update") { + return None; + } + let tool_id = opaque_id(update["toolCallId"].as_str().unwrap_or("missing")); + let key = ( + opaque_id(session_id(event).unwrap_or("missing")), + tool_id.clone(), + ); + details["tool_id"] = tool_id.into(); + details["kind"] = update["kind"] + .as_str() + .filter(|value| { + matches!( + *value, + "read" + | "edit" + | "delete" + | "move" + | "search" + | "execute" + | "think" + | "fetch" + | "switch_mode" + ) + }) + .unwrap_or("other") + .into(); + let status = update["status"] + .as_str() + .filter(|value| { + matches!(*value, "pending" | "in_progress" | "completed" | "failed") + }) + .unwrap_or("unknown"); + details["status"] = status.into(); + details["arguments"] = command_summary(&update["rawInput"]); + details["output"] = json!({ + "redacted": true, + "bytes": update.get("content").map(serialized_len).unwrap_or(0), + "exit_code": update["rawOutput"]["exitCode"].as_i64(), + }); + if update_kind == "tool_call" { + if self.tools.len() < MAX_ACTIVE_TOOLS { + self.tools + .entry(key.clone()) + .or_insert_with(|| (Instant::now(), event.turn_id.clone())); + } else { + details["timing_capacity_exceeded"] = true.into(); + } + } + if matches!(status, "completed" | "failed") { + if let Some((start, _)) = self.tools.remove(&key) { + details["duration_ms"] = json!(start.elapsed().as_millis()); + } + "tool_completed" + } else if update_kind == "tool_call" { + "tool_started" + } else { + "tool_updated" + } + } + _ => return None, + }; + self.sequence += 1; + Some(json!({ + "schema": "buzz.activity.v1", + "source_id": self.source_id, + "event_id": format!("{}:{}", self.source_id, self.sequence), + "sequence": self.sequence, + "timestamp": event.timestamp, + "event": kind, + "agent_index": event.agent_index, + "channel_id": event.channel_id.as_deref().map(opaque_id), + "acp_session_id": session_id(event).map(opaque_id), + "turn_id": event.turn_id.as_deref().map(opaque_id), + "details": details, + })) + } +} + +fn session_id(event: &ObserverEvent) -> Option<&str> { + event + .session_id + .as_deref() + .or_else(|| event.payload["params"]["sessionId"].as_str()) +} + +fn opaque_id(value: &str) -> String { + hex::encode(Sha256::digest(value.as_bytes())) +} + +/// Serialized size without materializing a copy of possibly large output. +fn serialized_len(value: &Value) -> usize { + struct Count(usize); + impl std::io::Write for Count { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 += bytes.len(); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let mut count = Count(0); + // Writing a `Value` to an infallible sink cannot fail. + let _ = serde_json::to_writer(&mut count, value); + count.0 +} + +fn command_summary(input: &Value) -> Value { + // Log only known command names. Arbitrary arguments may contain secrets, + // SQL statements, environment assignments, or embedded shell expansions. + let command = input["command"].as_str().unwrap_or(""); + let mut words = command.split_whitespace(); + let program = words + .next() + .filter(|value| { + matches!( + *value, + "cargo" + | "git" + | "go" + | "python" + | "python3" + | "node" + | "npm" + | "pnpm" + | "yarn" + | "just" + | "make" + | "via" + | "gh" + | "psql" + | "sqlite3" + ) + }) + .unwrap_or("other"); + let operation = words + .next() + .filter(|value| { + matches!( + *value, + "build" + | "test" + | "check" + | "clippy" + | "fmt" + | "clone" + | "fetch" + | "commit" + | "push" + | "status" + | "diff" + | "install" + | "run" + | "api" + ) + }) + .unwrap_or("other"); + json!({"program": program, "operation": operation, "remaining_arguments": "[redacted]"}) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event(payload: Value) -> ObserverEvent { + ObserverEvent { + seq: 1, + timestamp: "2026-10-08T12:00:00Z".into(), + kind: "acp_read".into(), + agent_index: Some(0), + channel_id: Some("channel".into()), + session_id: Some("session".into()), + turn_id: Some("turn".into()), + started_at: None, + payload, + } + } + + #[test] + fn tool_records_are_correlated_and_exclude_raw_content() { + let mut recorder = ActivityRecorder::new(); + let mut e = event(json!({"method":"session/update","params":{"update":{ + "sessionUpdate":"tool_call", "toolCallId":"secret-id", "kind":"execute", + "title":"SECRET-TITLE", "status":"in_progress", + "rawInput":{"command":"cargo test SECRET-ARG", "password":"SECRET-PASSWORD"}, + "content":[{"text":"SECRET-RESULT"}] + }}})); + let start = recorder.record(&e).unwrap(); + e.payload["params"]["update"]["sessionUpdate"] = "tool_call_update".into(); + e.payload["params"]["update"]["status"] = "completed".into(); + e.payload["params"]["update"]["rawOutput"] = json!({"exitCode":0}); + let finish = recorder.record(&e).unwrap(); + assert_eq!(start["event"], "tool_started"); + assert_eq!(finish["event"], "tool_completed"); + assert_eq!(start["details"]["tool_id"], finish["details"]["tool_id"]); + assert_eq!(finish["sequence"], 2); + assert!(finish["details"]["duration_ms"].is_number()); + assert_eq!(finish["details"]["output"]["exit_code"], 0); + assert_eq!(start["details"]["arguments"]["program"], "cargo"); + for value in [start, finish] { + let serialized = value.to_string(); + assert!(!serialized.contains("SECRET")); + assert!(!serialized.contains("secret-id")); + } + } + + #[test] + fn checkpoint_failure_records_only_its_reason_and_exit_code() { + let mut recorder = ActivityRecorder::new(); + let mut e = event(json!({"reason":"exit_status","exit_code":3,"stderr":"SECRET"})); + e.kind = "checkpoint_failed".into(); + let record = recorder.record(&e).unwrap(); + assert_eq!(record["event"], "checkpoint_failed"); + assert_eq!( + record["details"], + json!({"reason":"exit_status","exit_code":3}) + ); + } + + #[test] + fn output_bytes_match_the_serialized_tool_content() { + let content = json!([{"text":"0123456789", "nested":{"quote":"\""}}]); + let mut recorder = ActivityRecorder::new(); + let record = recorder + .record(&event( + json!({"method":"session/update","params":{"update":{ + "sessionUpdate":"tool_call", "toolCallId":"id", "content": content + }}}), + )) + .unwrap(); + assert_eq!( + record["details"]["output"]["bytes"], + content.to_string().len() + ); + } + + #[test] + fn prompt_and_agent_text_are_never_platform_records() { + let mut recorder = ActivityRecorder::new(); + let mut e = event(json!({"method":"session/update","params":{"update":{ + "sessionUpdate":"agent_message_chunk", "content":{"text":"SECRET"} + }}})); + assert!(recorder.record(&e).is_none()); + e.kind = "acp_write".into(); + e.payload = json!({"method":"session/prompt","params":{"prompt":"SECRET"}}); + assert!(recorder.record(&e).is_none()); + } + + #[test] + fn active_tool_tracking_is_bounded() { + let mut recorder = ActivityRecorder::new(); + let mut last = Value::Null; + for id in 0..=MAX_ACTIVE_TOOLS { + last = recorder + .record(&event( + json!({"method":"session/update","params":{"update":{ + "sessionUpdate":"tool_call", "toolCallId":id.to_string() + }}}), + )) + .unwrap(); + } + assert_eq!(recorder.tools.len(), MAX_ACTIVE_TOOLS); + assert_eq!(last["details"]["timing_capacity_exceeded"], true); + } + + #[cfg(unix)] + #[derive(Clone)] + struct Capture(std::sync::Arc>>); + + #[cfg(unix)] + impl std::io::Write for Capture { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + #[cfg(unix)] + #[tokio::test] + async fn real_acp_read_loop_emits_only_sanitized_activity_when_enabled() { + use tracing::instrument::WithSubscriber; + let output = Capture(Default::default()); + let writer = output.clone(); + let subscriber = tracing_subscriber::fmt() + .json() + .with_max_level(tracing::Level::INFO) + .with_writer(move || writer.clone()) + .finish(); + async { + for enabled in [false, true] { + let script = r#" +read -r request +printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"fixture","update":{"sessionUpdate":"tool_call","toolCallId":"tool-1","kind":"execute","status":"in_progress","title":"DO-NOT-LOG-TITLE","rawInput":{"command":"cargo test DO-NOT-LOG-ARG"}}}}' +printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"fixture","update":{"sessionUpdate":"tool_call_update","toolCallId":"tool-1","status":"completed","content":[{"text":"DO-NOT-LOG-RESULT"}],"rawOutput":{"exitCode":0}}}}' +printf '%s\n' '{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":2,"agentCapabilities":{},"authMethods":[]}}' +"#; + let mut client = crate::acp::AcpClient::spawn( + "bash", &["-c".into(), script.into()], &[], false, + ).await.unwrap(); + let observer = if enabled { + crate::observer::ObserverHandle::with_activity_log() + } else { + crate::observer::ObserverHandle::in_process() + }; + client.set_observer(Some(observer), 0); + client.initialize().await.unwrap(); + client.shutdown().await; + } + }.with_subscriber(subscriber).await; + let text = String::from_utf8(output.0.lock().unwrap().clone()).unwrap(); + let records: Vec = text + .lines() + .filter_map(|line| { + let envelope: Value = serde_json::from_str(line).ok()?; + if envelope["fields"]["schema"] != "buzz.activity.v1" { + return None; + } + Some(serde_json::from_str(envelope["fields"]["record"].as_str()?).unwrap()) + }) + .collect(); + assert_eq!( + records.len(), + 2, + "only the enabled adapter emits platform records" + ); + assert_eq!(records[0]["event"], "tool_started"); + assert_eq!(records[1]["event"], "tool_completed"); + assert!(!serde_json::to_string(&records) + .unwrap() + .contains("DO-NOT-LOG")); + } +} diff --git a/crates/buzz-acp/src/checkpoint.rs b/crates/buzz-acp/src/checkpoint.rs new file mode 100644 index 00000000000..ed347739e7f --- /dev/null +++ b/crates/buzz-acp/src/checkpoint.rs @@ -0,0 +1,317 @@ +//! Checkpoint the workspace after all adapter processes have stopped. +use std::time::Duration; + +use serde::Deserialize; +use tokio::io::AsyncReadExt; +use tokio::process::Command; + +use crate::observer::{ObserverContext, ObserverHandle}; + +const RECEIPT_BUDGET: usize = 2048; + +/// Why a stop checkpoint failed. Helper stderr is discarded because it may +/// carry storage credentials, so these variants are the whole diagnosis. +#[derive(Debug, thiserror::Error)] +pub(crate) enum CheckpointError { + #[error("could not start buzz-agent-checkpoint: {0}")] + Spawn(std::io::Error), + #[error("could not read the checkpoint helper: {0}")] + Io(std::io::Error), + #[error("checkpoint helper did not finish within {0:?}")] + Timeout(Duration), + #[error("checkpoint helper failed ({0})")] + ExitStatus(std::process::ExitStatus), + #[error("checkpoint receipt exceeds {RECEIPT_BUDGET} bytes")] + ReceiptOversized, + #[error("invalid checkpoint receipt: {0}")] + ReceiptInvalid(&'static str), +} + +impl CheckpointError { + fn reason(&self) -> &'static str { + match self { + Self::Spawn(_) => "spawn", + Self::Io(_) => "io", + Self::Timeout(_) => "timeout", + Self::ExitStatus(_) => "exit_status", + Self::ReceiptOversized => "receipt_oversized", + Self::ReceiptInvalid(_) => "receipt_invalid", + } + } + + fn details(&self) -> serde_json::Value { + let mut details = serde_json::json!({"reason": self.reason()}); + if let Self::ExitStatus(status) = self { + details["exit_code"] = status.code().into(); + } + details + } +} + +/// Only these fields reach activity; anything else the helper adds is ignored. +#[derive(Deserialize)] +struct Receipt { + version: u32, + key: String, + sha256: String, + repositories: u64, + bytes: u64, +} + +fn receipt(bytes: &[u8]) -> Result { + let saved: Receipt = serde_json::from_slice(bytes) + .map_err(|_| CheckpointError::ReceiptInvalid("malformed receipt"))?; + if saved.version != 1 { + return Err(CheckpointError::ReceiptInvalid("unsupported version")); + } + if saved.sha256.len() != 64 || !saved.sha256.bytes().all(|b| b.is_ascii_hexdigit()) { + return Err(CheckpointError::ReceiptInvalid("invalid sha256")); + } + let key = saved.key.as_str(); + if key.is_empty() + || key.len() > 200 + || key.starts_with('/') + || key.contains("..") + || !key + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"-/._".contains(&b)) + { + return Err(CheckpointError::ReceiptInvalid("invalid key")); + } + Ok(serde_json::json!({"key":saved.key,"sha256":saved.sha256, + "repositories":saved.repositories,"bytes":saved.bytes})) +} + +pub(crate) async fn save(observer: Option<&ObserverHandle>) -> anyhow::Result<()> { + let mut command = Command::new("buzz-agent-checkpoint"); + command.arg("save"); + save_with_command(observer, command, Duration::from_secs(120)) + .await + .map_err(|error| { + anyhow::Error::new(error) + .context("checkpoint failed; retain the Sandbox workspace for explicit recovery") + }) +} + +async fn save_with_command( + observer: Option<&ObserverHandle>, + mut command: Command, + timeout: Duration, +) -> Result<(), CheckpointError> { + let emit = |kind, details| { + if let Some(observer) = observer { + observer.emit(kind, None, &ObserverContext::default(), details); + } + }; + emit("checkpoint_started", serde_json::json!({})); + command + .kill_on_drop(true) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::null()); + #[cfg(unix)] + command.process_group(0); + match run_checkpoint(command, timeout).await { + Ok(details) => { + emit("checkpoint_saved", details); + Ok(()) + } + Err(error) => { + tracing::error!( + reason = error.reason(), + "workspace checkpoint failed: {error}" + ); + emit("checkpoint_failed", error.details()); + Err(error) + } + } +} + +async fn run_checkpoint( + mut command: Command, + timeout: Duration, +) -> Result { + let mut child = command.spawn().map_err(CheckpointError::Spawn)?; + // `id()` is `None` once the helper is reaped. Capture the group now so a + // failed helper's surviving descendants are still killed. + #[cfg(unix)] + let group = child.id().and_then(|id| i32::try_from(id).ok()); + let stdout = child.stdout.take().ok_or_else(|| { + CheckpointError::Io(std::io::Error::other("checkpoint output is missing")) + })?; + let result = tokio::time::timeout(timeout, async { + let mut output = Vec::new(); + stdout + .take(RECEIPT_BUDGET as u64 + 1) + .read_to_end(&mut output) + .await + .map_err(CheckpointError::Io)?; + if output.len() > RECEIPT_BUDGET { + return Err(CheckpointError::ReceiptOversized); + } + let status = child.wait().await.map_err(CheckpointError::Io)?; + if !status.success() { + return Err(CheckpointError::ExitStatus(status)); + } + receipt(&output) + }) + .await + .unwrap_or(Err(CheckpointError::Timeout(timeout))); + #[cfg(unix)] + if let (Err(_), Some(group)) = (&result, group) { + let _ = nix::sys::signal::killpg( + nix::unistd::Pid::from_raw(group), + nix::sys::signal::Signal::SIGKILL, + ); + } + result +} + +#[cfg(test)] +mod tests { + use super::*; + + fn valid_receipt() -> serde_json::Value { + serde_json::json!({"version":1,"key":"developer/session/id.tar.gz", + "sha256":"a".repeat(64),"repositories":2,"bytes":100}) + } + + #[cfg(unix)] + #[tokio::test] + async fn oversized_receipt_is_rejected_before_the_helper_finishes() { + let mut command = Command::new("sh"); + command.args(["-c", "head -c 2049 /dev/zero; sleep 60"]); + let started = std::time::Instant::now(); + let error = save_with_command(None, command, Duration::from_secs(10)) + .await + .unwrap_err(); + assert!(matches!(error, CheckpointError::ReceiptOversized)); + assert!(started.elapsed() < Duration::from_secs(2)); + } + + #[test] + fn receipt_cannot_smuggle_raw_output_into_activity() { + assert!(receipt(valid_receipt().to_string().as_bytes()).is_ok()); + // Unknown fields from a newer helper are accepted but never projected. + let mut extra = valid_receipt(); + extra["raw_output"] = "synthetic secret".into(); + let projected = receipt(extra.to_string().as_bytes()).unwrap(); + assert_eq!( + projected, + serde_json::json!({"key":"developer/session/id.tar.gz", + "sha256":"a".repeat(64),"repositories":2,"bytes":100}) + ); + for key in ["../another-developer", "", "/etc/passwd", "key with space"] { + let mut invalid = valid_receipt(); + invalid["key"] = key.into(); + assert!( + matches!( + receipt(invalid.to_string().as_bytes()), + Err(CheckpointError::ReceiptInvalid("invalid key")) + ), + "{key:?} must be rejected" + ); + } + let mut future = valid_receipt(); + future["version"] = 2.into(); + assert!(matches!( + receipt(future.to_string().as_bytes()), + Err(CheckpointError::ReceiptInvalid("unsupported version")) + )); + } + + #[cfg(unix)] + #[tokio::test] + async fn helper_failures_record_their_reason_and_never_a_saved_checkpoint() { + let cases = [ + ( + "exit 3", + serde_json::json!({"reason":"exit_status","exit_code":3}), + ), + ( + "printf 'not a receipt'", + serde_json::json!({"reason":"receipt_invalid"}), + ), + ("sleep 60 & wait", serde_json::json!({"reason":"timeout"})), + ]; + for (script, details) in cases { + let observer = ObserverHandle::in_process(); + let mut command = Command::new("sh"); + command.args(["-c", script]); + assert!( + save_with_command(Some(&observer), command, Duration::from_millis(100)) + .await + .is_err() + ); + let events = observer.snapshot(); + assert_eq!( + events + .iter() + .map(|event| event.kind.as_str()) + .collect::>(), + ["checkpoint_started", "checkpoint_failed"] + ); + assert_eq!(events[1].payload, details, "{script}"); + } + let observer = ObserverHandle::in_process(); + let command = Command::new("/nonexistent/checkpoint-helper"); + assert!(matches!( + save_with_command(Some(&observer), command, Duration::from_secs(1)).await, + Err(CheckpointError::Spawn(_)) + )); + let failed = observer.snapshot().last().unwrap().clone(); + assert_eq!(failed.kind, "checkpoint_failed"); + assert_eq!(failed.payload, serde_json::json!({"reason":"spawn"})); + } + + /// The helper is already reaped when its exit status is known, so the + /// group kill must use the PID captured at spawn to reach descendants. + #[cfg(unix)] + #[tokio::test] + async fn failed_helper_descendants_are_killed_after_the_helper_exits() { + let dir = tempfile::tempdir().unwrap(); + let pid_file = dir.path().join("descendant.pid"); + let mut command = Command::new("sh"); + command.args([ + "-c", + "sleep 60 >/dev/null 2>&1 & echo $! > \"$1\"; exit 1", + "checkpoint-test", + pid_file.to_str().unwrap(), + ]); + assert!(matches!( + save_with_command(None, command, Duration::from_secs(5)).await, + Err(CheckpointError::ExitStatus(_)) + )); + let pid: i32 = std::fs::read_to_string(&pid_file) + .unwrap() + .trim() + .parse() + .unwrap(); + let pid = nix::unistd::Pid::from_raw(pid); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + // The orphan is reaped by init (or a subreaper) once killed. + while nix::sys::signal::kill(pid, None).is_ok() { + assert!( + std::time::Instant::now() < deadline, + "helper descendant survived a failed checkpoint" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + #[cfg(unix)] + #[tokio::test] + async fn helper_success_emits_verified_receipt_after_exit() { + let observer = ObserverHandle::in_process(); + let valid = valid_receipt().to_string(); + let mut command = Command::new("sh"); + command.args(["-c", "printf '%s' \"$1\"", "checkpoint-test", &valid]); + save_with_command(Some(&observer), command, Duration::from_secs(1)) + .await + .unwrap(); + let events = observer.snapshot(); + assert_eq!(events.len(), 2); + assert_eq!(events[0].kind, "checkpoint_started"); + assert_eq!(events[1].kind, "checkpoint_saved"); + assert_eq!(events[1].payload["key"], "developer/session/id.tar.gz"); + } +} diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 4dbaee4b19a..e9207bb469e 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -212,7 +212,7 @@ impl std::fmt::Display for PermissionMode { /// CLI args for `buzz-acp models` — query available models from an agent. /// /// This is a standalone `Parser` (not a subcommand variant) because the -/// `models` path must bypass `Config::from_cli()` entirely — no relay, +/// `models` path must bypass `Config::from_args()` entirely — no relay, /// no private key, no harness setup. #[derive(Debug, Parser)] #[command( @@ -541,10 +541,18 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_RELAY_OBSERVER", default_value_t = false)] pub relay_observer: bool, + /// Emit bounded tool and lifecycle activity as JSONL diagnostics. + #[arg(long, env = "BUZZ_ACP_ACTIVITY_LOG", default_value_t = false)] + pub activity_log: bool, + /// Restrict this remote workspace to one channel and canonical parent thread. #[arg(long, env = "BUZZ_ACP_BOUND_SESSION", value_parser = parse_bound_session)] pub bound_session: Option, + /// Save and verify a private workspace checkpoint after draining the agents. + #[arg(long, env = "BUZZ_ACP_CHECKPOINT_ON_STOP", default_value_t = false)] + pub checkpoint_on_stop: bool, + /// Exit after this many seconds with no dispatched events and no turn in flight. /// 0 disables inactivity self-termination. #[arg(long, env = "BUZZ_ACP_EXIT_AFTER_INACTIVITY", default_value_t = 0)] @@ -650,7 +658,12 @@ pub struct Config { pub has_generated_codex_config: bool, /// Whether to publish encrypted observer frames through the relay. pub relay_observer: bool, + /// Whether to emit bounded JSONL activity for platform collection. + pub activity_log: bool, pub bound_session: Option, + /// Whether to run `buzz-agent-checkpoint save` after draining on stop and + /// fail shutdown when it does not produce a verified receipt. + pub checkpoint_on_stop: bool, /// Seconds without dispatched events before an idle harness exits. 0 = disabled. pub exit_after_inactivity_secs: u64, /// Whether ACP/LLM subprocess initialization is deferred until accepted work arrives. @@ -671,7 +684,7 @@ pub struct Config { /// Disable the `` platform-context section prepended to every prompt. pub no_base_prompt: bool, /// Resolved content from `--base-prompt-file`, read and validated in - /// `from_cli()`. `None` when using the compiled-in default or when + /// `from_args()`. `None` when using the compiled-in default or when /// `--no-base-prompt` is set. pub base_prompt_content: Option, } @@ -963,17 +976,13 @@ pub fn propagate_legacy_env_vars() { } impl Config { - pub fn from_cli() -> Result { - // Legacy env-var propagation is intentionally NOT done here. - // Call `propagate_legacy_env_vars()` before the tokio runtime starts - // (in the sync `fn main()` wrapper) — see Rust 2024 edition safety. - let args = CliArgs::parse(); - Self::from_args(args) - } - - /// Build a `Config` from already-parsed `CliArgs`. Separated from `from_cli()` so - /// tests can construct `CliArgs` via `CliArgs::try_parse_from` and exercise the full - /// validation path without going through process args. + /// Build a `Config` from already-parsed `CliArgs`. Taking parsed args lets + /// the binary install logging before validation and lets tests construct + /// `CliArgs` via `CliArgs::try_parse_from` to exercise the full validation path. + /// + /// Legacy env-var propagation is intentionally NOT done here. Call + /// `propagate_legacy_env_vars()` before the tokio runtime starts (in the + /// sync `fn main()` wrapper) — see Rust 2024 edition safety. pub fn from_args(mut args: CliArgs) -> Result { if args.bound_session.is_some() { let refusal = if args.respond_to != RespondTo::Owner { @@ -1264,7 +1273,9 @@ impl Config { persona_env_vars, has_generated_codex_config, relay_observer: args.relay_observer, + activity_log: args.activity_log, bound_session: args.bound_session, + checkpoint_on_stop: args.checkpoint_on_stop, exit_after_inactivity_secs: args.exit_after_inactivity, lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, @@ -1761,7 +1772,9 @@ mod tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + activity_log: false, bound_session: None, + checkpoint_on_stop: false, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index a9be976586f..81ce7c103ea 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -5,6 +5,8 @@ mod git; mod git_runtime_tests; mod acp; +mod activity; +mod checkpoint; mod config; mod edit_routing; mod engram_fetch; @@ -2800,17 +2802,28 @@ async fn tokio_main() -> Result<()> { return run_authenticate(args).await; } - // Stdout is the ACP transport when buzz-acp is launched as an agent command. - // Keep every harness diagnostic on stderr so logging can never corrupt NDJSON. - tracing_subscriber::fmt() - .with_writer(std::io::stderr) - .with_env_filter( - EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("buzz_acp=info")), - ) - .compact() - .init(); - - let config = Config::from_cli().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?; + // Parse before installing the subscriber (its format depends on + // `--activity-log`), but validate after, so config warnings are not lost. + let args = config::CliArgs::parse(); + let filter = + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("buzz_acp=info")); + // Stdout is the ACP transport when buzz-acp is launched as an agent + // command; every diagnostic, including platform activity, uses stderr. + if args.activity_log { + tracing_subscriber::fmt() + .with_writer(std::io::stderr) + .with_env_filter(filter.add_directive("buzz_acp::activity=info".parse()?)) + .json() + .init(); + } else { + tracing_subscriber::fmt() + .with_writer(std::io::stderr) + .with_env_filter(filter) + .compact() + .init(); + } + let config = + Config::from_args(args).map_err(|e| anyhow::anyhow!("configuration error: {e}"))?; // ── Setup-mode early branch ─────────────────────────────────────────────── // @@ -2847,17 +2860,52 @@ async fn tokio_main() -> Result<()> { let _ = tokio::signal::ctrl_c().await; let _ = tx.send(()); }); - let (ready_tx, mut ready_rx) = tokio::sync::oneshot::channel(); - let harness = run_harness(config, shutdown_tx, shutdown_rx.clone(), ready_tx); - tokio::pin!(harness); - let result = tokio::select! { - biased; - _ = shutdown_rx.changed() => Ok(()), - result = &mut harness => result, - _ = &mut ready_rx => harness.await, + let checkpoint_on_stop = config.checkpoint_on_stop; + let observer = if config.activity_log { + Some(observer::ObserverHandle::with_activity_log()) + } else { + config + .relay_observer + .then(observer::ObserverHandle::in_process) + }; + let (startup_finished, result) = { + let (ready_tx, mut ready_rx) = tokio::sync::oneshot::channel(); + let harness = run_harness( + config, + shutdown_tx, + shutdown_rx.clone(), + ready_tx, + observer.clone(), + ); + tokio::pin!(harness); + tokio::select! { + biased; + _ = shutdown_rx.changed() => (false, Ok(())), + result = &mut harness => (ready_rx.try_recv().is_ok(), result), + ready = &mut ready_rx => (ready.is_ok(), harness.await), + } }; signal_task.abort(); - result + if startup_finished { + return result; + } + // The startup future has now dropped, killing adapter process groups and + // removing temporary keys. Existing work still needs a checkpoint even + // when startup failed or was cancelled before the relay became available. + let checkpoint_result = if checkpoint_on_stop { + checkpoint::save(observer.as_ref()).await + } else { + Ok(()) + }; + if let Some(handle) = observer { + handle.emit( + "harness_stopped", + None, + &observer::ObserverContext::default(), + serde_json::json!({"reason": if result.is_err() { "startup_failed" } else { "startup_cancelled" }}), + ); + } + result.and(checkpoint_result) } async fn run_harness( @@ -2865,15 +2913,13 @@ async fn run_harness( shutdown_tx: watch::Sender<()>, mut shutdown_rx: watch::Receiver<()>, startup_ready: tokio::sync::oneshot::Sender<()>, + observer: Option, ) -> Result<()> { let runtime = AgentRuntime::prepare(config)?; let config = runtime.config(); tracing::info!("buzz-acp starting: {}", config.summary()); - let observer = config - .relay_observer - .then(observer::ObserverHandle::in_process); if let Some(handle) = &observer { handle.emit( "harness_started", @@ -4428,6 +4474,14 @@ async fn run_harness( } } + // Offline goes first: the agent stopped accepting work at drain, and the + // checkpoint may take up to its full timeout. + let checkpoint_result = if config.checkpoint_on_stop { + checkpoint::save(observer.as_ref()).await + } else { + Ok(()) + }; + if let Some(handle) = relay_observer_publisher_task.take() { handle.abort(); } @@ -4436,8 +4490,16 @@ async fn run_harness( // for the background task to finish, rather than aborting immediately (#40). relay.shutdown().await; + if let Some(handle) = &observer { + handle.emit( + "harness_stopped", + None, + &observer::ObserverContext::default(), + serde_json::json!({}), + ); + } tracing::info!("buzz-acp stopped"); - Ok(()) + checkpoint_result } #[derive(PartialEq)] @@ -10460,7 +10522,9 @@ mod build_mcp_servers_tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + activity_log: false, bound_session: None, + checkpoint_on_stop: false, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, @@ -11558,7 +11622,9 @@ mod error_outcome_emission_tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + activity_log: false, bound_session: None, + checkpoint_on_stop: false, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, diff --git a/crates/buzz-acp/src/observer.rs b/crates/buzz-acp/src/observer.rs index 7029e5af6d5..d1727d66b3b 100644 --- a/crates/buzz-acp/src/observer.rs +++ b/crates/buzz-acp/src/observer.rs @@ -40,15 +40,17 @@ struct ObserverInner { tx: broadcast::Sender, buffer: Mutex>, seq: AtomicU64, + activity: Option>, } -fn new_observer_handle() -> ObserverHandle { +fn new_observer_handle(activity: bool) -> ObserverHandle { let (tx, _) = broadcast::channel(OBSERVER_BUFFER_CAP); ObserverHandle { inner: Arc::new(ObserverInner { tx, buffer: Mutex::new(VecDeque::with_capacity(OBSERVER_BUFFER_CAP)), seq: AtomicU64::new(1), + activity: activity.then(|| Mutex::new(crate::activity::ActivityRecorder::new())), }), } } @@ -81,7 +83,12 @@ pub struct ObserverEvent { impl ObserverHandle { /// Create an in-process observer feed. pub fn in_process() -> Self { - new_observer_handle() + new_observer_handle(false) + } + + /// Create an observer feed with bounded JSONL activity through tracing. + pub(crate) fn with_activity_log() -> Self { + new_observer_handle(true) } /// Subscribe to live observer events. @@ -120,6 +127,18 @@ impl ObserverHandle { payload, }; + // Release the recorder before logging so stderr never blocks other emitters. + let record = self.inner.activity.as_ref().and_then(|activity| { + match activity.lock() { + Ok(recorder) => recorder, + Err(poisoned) => poisoned.into_inner(), + } + .record(&event) + }); + if let Some(record) = record { + tracing::info!(target: "buzz_acp::activity", schema = "buzz.activity.v1", record = %record); + } + match self.inner.buffer.lock() { Ok(mut buffer) => { if buffer.len() >= OBSERVER_BUFFER_CAP { diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index 827d297ac9c..620038ce908 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -6,7 +6,7 @@ //! early-branch path: //! //! ```text -//! Config::from_cli() +//! Config::from_args(CliArgs::parse()) //! └─ SetupPayload::from_env()? //! ├─ Some(payload) → run_setup_listener(config, payload) [this module] //! └─ None → normal pool path (unchanged) diff --git a/crates/buzz-acp/tests/checkpoint_stop.rs b/crates/buzz-acp/tests/checkpoint_stop.rs new file mode 100644 index 00000000000..a958d1c2a37 --- /dev/null +++ b/crates/buzz-acp/tests/checkpoint_stop.rs @@ -0,0 +1,312 @@ +//! Stop the real harness binary with `--checkpoint-on-stop` and a stub +//! `buzz-agent-checkpoint` on PATH, on shutdown and startup failure paths. +#![cfg(unix)] + +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::process::{ExitStatus, Stdio}; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_tungstenite::tungstenite::Message; + +const ADAPTER: &str = r#"#!/bin/sh +echo $$ > "$PROBE_DIR/adapter.pid" +# Never answer initialize, so SIGTERM arrives while startup is still running. +[ "$ADAPTER_MODE" = hang ] && exec sleep 60 +[ "$ADAPTER_MODE" = fail ] && exit 17 +read -r request +printf '%s\n' '{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":2,"agentCapabilities":{},"authMethods":[]}}' +exec cat >/dev/null +"#; + +// Records what had already happened when the harness started the checkpoint. +const HELPER: &str = r#"#!/bin/sh +test "$1" = save || exit 64 +echo save >> "$PROBE_DIR/checkpoint-runs" +{ + if kill -0 "$(cat "$PROBE_DIR/adapter.pid")" 2>/dev/null; then + echo adapter=running + else + echo adapter=stopped + fi + if grep -q '"content":"offline"' "$PROBE_DIR/relay-events" 2>/dev/null; then + echo presence=offline + else + echo presence=online + fi +} > "$PROBE_DIR/checkpoint" +test "$CHECKPOINT_FAIL" = 1 && exit 3 +printf '{"version":1,"key":"developer/session/id.tar.gz","sha256":"%064d","repositories":0,"bytes":0}' 0 +"#; + +#[derive(Clone, Copy, PartialEq)] +enum Stop { + /// SIGTERM after startup: the main loop drains, then checkpoints. + AfterReady, + /// SIGTERM during adapter initialization: shutdown cancels startup. + DuringStartup, + /// Initialization fails before the harness can announce readiness. + StartupFailure, +} + +struct Stopped { + status: ExitStatus, + /// Stub output, or `None` when the harness never ran the helper. + checkpoint: Option, + logs: String, +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn normal_stop_checkpoints_after_drain_and_offline_presence() { + let stopped = stop(Stop::AfterReady, false).await; + assert_eq!(stopped.status.code(), Some(0), "{}", stopped.logs); + assert_eq!( + stopped.checkpoint.as_deref(), + Some("adapter=stopped\npresence=offline\n"), + "{}", + stopped.logs + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn startup_cancellation_still_checkpoints() { + let stopped = stop(Stop::DuringStartup, false).await; + assert_eq!(stopped.status.code(), Some(0), "{}", stopped.logs); + assert!(stopped.checkpoint.is_some(), "{}", stopped.logs); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_checkpoint_fails_either_shutdown_path() { + for path in [Stop::AfterReady, Stop::DuringStartup] { + let stopped = stop(path, true).await; + assert!(stopped.checkpoint.is_some(), "{}", stopped.logs); + assert!(!stopped.status.success(), "{}", stopped.logs); + assert!(stopped.logs.contains("exit_status"), "{}", stopped.logs); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn startup_failure_checkpoints_and_preserves_the_original_error() { + for fail_checkpoint in [false, true] { + let stopped = stop(Stop::StartupFailure, fail_checkpoint).await; + assert!(!stopped.status.success(), "{}", stopped.logs); + assert_eq!( + stopped.checkpoint.as_deref(), + Some("adapter=stopped\npresence=online\n") + ); + assert!( + stopped.logs.contains("all 1 agents failed to start"), + "{}", + stopped.logs + ); + let records: Vec = stopped + .logs + .lines() + .filter_map(|line| { + let envelope: serde_json::Value = serde_json::from_str(line).ok()?; + serde_json::from_str(envelope["fields"]["record"].as_str()?).ok() + }) + .collect(); + assert!( + records + .iter() + .any(|record| record["event"] == "harness_stopped" + && record["details"]["reason"] == "startup_failed"), + "{}", + stopped.logs + ); + let expected = if fail_checkpoint { + "checkpoint_failed" + } else { + "checkpoint_saved" + }; + assert!( + records.iter().any(|record| record["event"] == expected), + "{}", + stopped.logs + ); + } +} + +async fn stop(path: Stop, fail: bool) -> Stopped { + let temp = tempfile::tempdir().unwrap(); + let dir = temp.path(); + let bin = dir.join("bin"); + std::fs::create_dir(&bin).unwrap(); + write_script(&dir.join("adapter"), ADAPTER); + write_script(&bin.join("buzz-agent-checkpoint"), HELPER); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let relay_url = format!("ws://{}", listener.local_addr().unwrap()); + let relay = tokio::spawn(fake_relay(listener, dir.join("relay-events"))); + + let mut search_path = std::ffi::OsString::from(&bin); + search_path.push(":"); + search_path.push(std::env::var_os("PATH").unwrap_or_default()); + let log = std::fs::File::create(dir.join("harness.log")).unwrap(); + let keys = nostr::Keys::generate(); + let mut child = tokio::process::Command::new(env!("CARGO_BIN_EXE_buzz-acp")) + .env_clear() + .env("PATH", search_path) + .env("PROBE_DIR", dir) + .env( + "ADAPTER_MODE", + match path { + Stop::DuringStartup => "hang", + Stop::StartupFailure => "fail", + Stop::AfterReady => "", + }, + ) + .env("CHECKPOINT_FAIL", if fail { "1" } else { "0" }) + .args([ + "--private-key", + &keys.secret_key().to_secret_hex(), + "--relay-url", + &relay_url, + "--agent-command", + dir.join("adapter").to_str().unwrap(), + "--agent-args", + "", + "--checkpoint-on-stop", + "--activity-log", + ]) + .stdin(Stdio::null()) + .stdout(log.try_clone().unwrap()) + .stderr(log) + .kill_on_drop(true) + .spawn() + .unwrap(); + + // "presence set to online" is logged in the same poll that signals + // readiness (no await in between), so after it SIGTERM takes the + // main-loop path rather than startup cancellation. + let ready = || match path { + Stop::AfterReady => std::fs::read_to_string(dir.join("harness.log")) + .unwrap_or_default() + .contains("presence set to online"), + Stop::DuringStartup => dir.join("adapter.pid").exists(), + Stop::StartupFailure => true, + }; + tokio::time::timeout(Duration::from_secs(20), async { + while !ready() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("harness never reached the stop point: {}", logs(dir))); + + if path != Stop::StartupFailure { + nix::sys::signal::kill( + nix::unistd::Pid::from_raw(child.id().unwrap() as i32), + nix::sys::signal::Signal::SIGTERM, + ) + .unwrap(); + } + let status = tokio::time::timeout(Duration::from_secs(30), child.wait()) + .await + .unwrap_or_else(|_| panic!("harness did not exit: {}", logs(dir))) + .unwrap(); + relay.abort(); + assert_eq!( + std::fs::read_to_string(dir.join("checkpoint-runs")).unwrap_or_default(), + "save\n", + "the stop hook must run exactly once: {}", + logs(dir) + ); + Stopped { + status, + checkpoint: std::fs::read_to_string(dir.join("checkpoint")).ok(), + logs: logs(dir), + } +} + +fn write_script(path: &Path, body: &str) { + std::fs::write(path, body).unwrap(); + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)).unwrap(); +} + +fn logs(dir: &Path) -> String { + std::fs::read_to_string(dir.join("harness.log")).unwrap_or_default() +} + +/// Minimal relay: NIP-42 challenge, `OK` for every event, `EOSE` for every +/// subscription, and an empty result for every HTTP query. Published events +/// are appended to `events` before they are acknowledged. +async fn fake_relay(listener: TcpListener, events: PathBuf) { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + tokio::spawn(serve(stream, events.clone())); + } +} + +async fn serve(mut stream: TcpStream, events: PathBuf) { + let mut head = [0; 4096]; + let peeked = loop { + let n = stream.peek(&mut head).await.unwrap_or(0); + if n == 0 || n == head.len() || head[..n].windows(4).any(|w| w == b"\r\n\r\n") { + break n; + } + tokio::time::sleep(Duration::from_millis(5)).await; + }; + let head = String::from_utf8_lossy(&head[..peeked]).to_ascii_lowercase(); + if !head.contains("upgrade: websocket") { + let length: usize = head + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .and_then(|value| value.trim().parse().ok()) + .unwrap_or(0); + let mut request = Vec::new(); + let mut chunk = [0; 4096]; + while request + .windows(4) + .position(|w| w == b"\r\n\r\n") + .is_none_or(|end| request.len() < end + 4 + length) + { + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(n) => request.extend_from_slice(&chunk[..n]), + } + } + let response = "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Content-Length: 2\r\nConnection: close\r\n\r\n[]"; + let _ = stream.write_all(response.as_bytes()).await; + return; + } + let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else { + return; + }; + let _ = ws + .send(Message::Text(r#"["AUTH","challenge"]"#.into())) + .await; + while let Some(Ok(message)) = ws.next().await { + let Ok(frame) = serde_json::from_str::(message.to_text().unwrap_or("")) + else { + continue; + }; + let reply = match frame[0].as_str() { + Some("AUTH" | "EVENT") => { + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&events) + .unwrap(); + std::io::Write::write_all(&mut file, format!("{}\n", frame[1]).as_bytes()).unwrap(); + serde_json::json!(["OK", frame[1]["id"], true, ""]) + } + Some("REQ") => serde_json::json!(["EOSE", frame[1]]), + _ => continue, + }; + if ws + .send(Message::Text(reply.to_string().into())) + .await + .is_err() + { + return; + } + } +} diff --git a/crates/buzz-backend-kubernetes/src/cluster.rs b/crates/buzz-backend-kubernetes/src/cluster.rs index 755f41f6c8a..3343cc1a387 100644 --- a/crates/buzz-backend-kubernetes/src/cluster.rs +++ b/crates/buzz-backend-kubernetes/src/cluster.rs @@ -11,7 +11,7 @@ //! * **Reads leave `resourceVersion` unset**, which is the quorum read. `"0"` //! is the cache read, and a confirmed absence from a cache is proof of //! nothing (`:761-769`). -//! * **Deletes never set `grace_period_seconds`**, so the object's own 60s +//! * **Deletes never set `grace_period_seconds`**, so the object's own declared //! budget applies. Passing `0` is a force-kill that discards the shutdown //! window the pod declares (`:1185-1189`). @@ -216,7 +216,7 @@ impl Substrate for Cluster { uid: Some(fence.uid.clone()), resource_version: Some(fence.resource_version.clone()), }), - // grace_period_seconds deliberately unset: the pod's own 60s + // grace_period_seconds deliberately unset: the pod's own declared // budget applies. ..Default::default() }; diff --git a/crates/buzz-backend-kubernetes/src/config.rs b/crates/buzz-backend-kubernetes/src/config.rs index df8e94adc6e..3937576ce6a 100644 --- a/crates/buzz-backend-kubernetes/src/config.rs +++ b/crates/buzz-backend-kubernetes/src/config.rs @@ -55,9 +55,21 @@ pub const RUN_AS_GID: i64 = 10001; /// (§Working directory). pub const WORKSPACE_PATH: &str = "/home/agent"; -/// `terminationGracePeriodSeconds` — a declared budget, not a derived sum -/// (§Pod shape). Kubernetes' default 30s would SIGKILL the harness mid-drain. -pub const TERMINATION_GRACE_SECONDS: i64 = 60; +/// Declared shutdown budget, including the 120s checkpoint timeout. +/// Child reaping is variable; this is operational margin, not a proven bound +/// on the harness's whole shutdown path (§K8s Grace). +pub const TERMINATION_GRACE_SECONDS: i64 = 300; + +/// `buzz-acp`'s stop checkpoint timeout (`checkpoint::save`). +const CHECKPOINT_TIMEOUT_SECONDS: i64 = 120; + +/// Known stop-path segments at the desktop's default parallelism of 10 +/// (§Stop): wake-task drain and awakened-pool reap, prompt drain and slot +/// reap (5s post-SIGKILL wait per slot each), presence `offline` (2s), and +/// relay close (5s). +const STOP_TAIL_SECONDS: i64 = (30 + 5 * 10) + (30 + 5 * 10) + 2 + 5; + +const _: () = assert!(TERMINATION_GRACE_SECONDS >= CHECKPOINT_TIMEOUT_SECONDS + STOP_TAIL_SECONDS); /// The only restart policy v1 ships. `OnFailure` is double-gated on the /// harness exit-code contract *and* a crash-loop classification row the state diff --git a/crates/buzz-backend-kubernetes/src/pod.rs b/crates/buzz-backend-kubernetes/src/pod.rs index bb78816c8e0..67b462bfc43 100644 --- a/crates/buzz-backend-kubernetes/src/pod.rs +++ b/crates/buzz-backend-kubernetes/src/pod.rs @@ -319,11 +319,10 @@ mod tests { assert_eq!(spec(&pod()).restart_policy.as_deref(), Some("Never")); } - /// 60s, not Kubernetes' default 30s — which would SIGKILL the harness - /// mid-drain and leave presence stale-online (§Pod shape). + /// Allow the harness to drain and run its 120s shutdown checkpoint. #[test] - fn declares_the_sixty_second_grace_budget() { - assert_eq!(spec(&pod()).termination_grace_period_seconds, Some(60)); + fn declares_the_checkpoint_shutdown_budget() { + assert_eq!(spec(&pod()).termination_grace_period_seconds, Some(300)); } /// The pod must not override the image's entrypoint: the image execs the diff --git a/docs/remote-agents.md b/docs/remote-agents.md index 595b0b48196..b997e576ebc 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -917,9 +917,9 @@ can yield two live instances in one scope. declared bounds — currently 2s + 5s = 7s** — which child cleanup may never consume: child reaping degrades first (skip remaining per-slot waits, force-kill), because a shared deadline without the reservation - can legally spend all 60s reaping children and hit SIGKILL before the + can legally spend the entire budget reaping children and hit SIGKILL before the one action the grace period exists to protect. The binding declares the - budget (§K8s Grace: 60s); anyone re-deriving "~37s" from the segment + budget (§K8s Grace: 300s); anyone re-deriving "~37s" from the segment constants is reading numbers without their variables. The desktop's local stop command rejects remote agents. - **Delete** with a live `backend_agent_id` requires `force_remote_delete: @@ -1073,7 +1073,7 @@ PID 1, and a PID-1 bash with no trap never delivers SIGTERM to the harness (PID 1 receives kernel-level default-handler signal immunity), so the pod rides out the entire grace period and is SIGKILLed with presence still online — voiding I5's substrate half and the very staleness window -`terminationGracePeriodSeconds: 60` was sized to close. The entrypoint +`terminationGracePeriodSeconds` is intended to close. The entrypoint shape and the grace period are one requirement, not two. **Payload → environment mapping.** The provider builds the pod environment @@ -1200,7 +1200,7 @@ regardless of `HOME`. 404) before creating. The delete call MUST use the object's own grace period (kube-rs: `DeleteParams { grace_period_seconds: None, .. }`); the tempting shortcut of passing `0` to skip the poll is a **force-kill** that - discards the 60s shutdown grace pinned below — the poll is mandatory + discards the declared shutdown grace pinned below — the poll is mandatory precisely because the fast path is wrong. For *terminal* (Succeeded/Failed) pods — and for **unscheduled** pods (no assigned node: unschedulable or quota-blocked `Pending`, a state users hit while setting @@ -1208,10 +1208,11 @@ regardless of `HOME`. zeroes the grace period and deletes immediately, so the normal restart path needs no meaningful wait — do not add a fixed sleep, and do not use zero-grace cleanup as a reason to skip the poll in the live-pod arm. -- **`terminationGracePeriodSeconds: 60` — a declared budget the harness +- **`terminationGracePeriodSeconds: 300` — a declared budget the harness MUST honor, not a sum the spec derived.** Kubernetes' default 30s grace would SIGKILL the harness mid-drain, leaving presence stale-online — the - avoidable half of I3's staleness window — so the binding declares 60s. + avoidable half of I3's staleness window — so the binding declares 300s, including room for the 120s shutdown + checkpoint. But the shutdown tail is *variable*, not constant (§Stop: the post-drain reap segment alone reaches ~87s at default parallelism at `28ae6cd21`, and earlier untimed segments precede it — the total is not bounded by @@ -1222,7 +1223,7 @@ regardless of `HOME`. finalization slice (≥ the finalizers' declared bounds, currently 7s) for presence `offline` and relay close, child cleanup degrading first (§Stop, Known Defect 7). Until the harness enforcement - lands, 60s is an operational margin that the tail can exceed. + lands, 300s is an operational margin that the tail can exceed. - **Hardening defaults (normative).** The workload is a prompted coding agent running repository and tool code while holding an nsec; the pod MUST NOT hand it ambient cluster credentials or kernel privilege on top: