From cefb9aa0e87e7affa87248f8741f3734cc39127e Mon Sep 17 00:00:00 2001 From: Tee8z Date: Mon, 19 Oct 2026 15:05:15 -0400 Subject: [PATCH] feat(acp): emit activity records and checkpoint on stop Add an opt-in JSONL activity log that records bounded, redacted tool and lifecycle activity from the observer feed on stderr, so prompts, agent text, and secrets never enter the platform stream. The binary now parses CLI arguments before installing logging because the log format depends on the flag. Add an opt-in stop checkpoint that runs buzz-agent-checkpoint save after sessions drain, including when shutdown cancels startup. It validates the receipt and makes shutdown fail when the checkpoint fails. Signed-off-by: Tee8z --- crates/buzz-acp/TESTING.md | 15 + crates/buzz-acp/src/activity.rs | 399 ++++++++++++++++++ crates/buzz-acp/src/checkpoint.rs | 317 ++++++++++++++ crates/buzz-acp/src/config.rs | 39 +- crates/buzz-acp/src/lib.rs | 114 +++-- crates/buzz-acp/src/observer.rs | 23 +- crates/buzz-acp/src/setup_mode.rs | 2 +- crates/buzz-acp/tests/checkpoint_stop.rs | 312 ++++++++++++++ crates/buzz-backend-kubernetes/src/cluster.rs | 4 +- crates/buzz-backend-kubernetes/src/config.rs | 18 +- crates/buzz-backend-kubernetes/src/pod.rs | 7 +- docs/remote-agents.md | 15 +- 12 files changed, 1209 insertions(+), 56 deletions(-) create mode 100644 crates/buzz-acp/src/activity.rs create mode 100644 crates/buzz-acp/src/checkpoint.rs create mode 100644 crates/buzz-acp/tests/checkpoint_stop.rs 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: