diff --git a/Cargo.lock b/Cargo.lock index ea5b02aaab..03d0199cc1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -781,6 +781,7 @@ dependencies = [ "base64", "buzz-core", "buzz-persona", + "buzz-publication-fence", "buzz-sdk", "chrono", "clap", @@ -795,6 +796,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", @@ -900,6 +902,7 @@ dependencies = [ "base64", "buzz-core", "buzz-persona", + "buzz-publication-fence", "buzz-sdk", "buzz-ws-client", "bytes", @@ -1080,6 +1083,18 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "buzz-publication-fence" +version = "0.1.0" +dependencies = [ + "fs2", + "serde", + "serde_json", + "tempfile", + "thiserror 2.0.18", + "uuid", +] + [[package]] name = "buzz-pubsub" version = "0.1.0" @@ -2871,6 +2886,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" diff --git a/Cargo.toml b/Cargo.toml index 3ac7ee4cce..564f70bf4e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ members = [ "crates/buzz-relay", "crates/buzz-core", + "crates/buzz-publication-fence", "crates/buzz-conformance", "crates/buzz-push-gateway", "crates/buzz-db", @@ -88,6 +89,7 @@ anyhow = "1" # Utilities uuid = { version = "1", features = ["v4", "serde"] } chrono = { version = "0.4", features = ["serde"] } +fs2 = "0.4" # HTTP client (webhook delivery) reqwest = { version = "0.13", features = ["json", "rustls"], default-features = false } @@ -123,6 +125,7 @@ schemars = { version = "1", default-features = false } # Internal crates buzz-core = { path = "crates/buzz-core" } +buzz-publication-fence = { path = "crates/buzz-publication-fence" } buzz-conformance = { path = "crates/buzz-conformance" } buzz-db = { path = "crates/buzz-db" } buzz-auth = { path = "crates/buzz-auth" } diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806..0c8bd7b54b 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -19,6 +19,7 @@ path = "src/main.rs" # Internal buzz-core = { workspace = true } buzz-sdk = { workspace = true } +buzz-publication-fence = { workspace = true } buzz-persona = { path = "../buzz-persona" } # Nostr @@ -61,6 +62,7 @@ tracing-subscriber = { workspace = true } # Error handling thiserror = { workspace = true } anyhow = { workspace = true } +tempfile = "3" # CLI clap = { version = "4", features = ["derive", "env"] } @@ -74,7 +76,7 @@ evalexpr = { workspace = true } # Process-group kill (safe wrapper around killpg) — Unix-only; kill_process_group # has a #[cfg(not(unix))] fallback in acp.rs. [target.'cfg(unix)'.dependencies] -nix = { version = "0.31", default-features = false, features = ["signal"] } +nix = { version = "0.31", default-features = false, features = ["signal", "user"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8a698954a0..238ee662df 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -8,6 +8,10 @@ //! 4. [`AcpClient::session_prompt_with_idle_timeout`] — send prompt with idle/hard deadline, return stop reason //! 5. [`AcpClient::session_cancel`] / [`AcpClient::cancel_with_cleanup`] — cancel in-flight turn +use buzz_publication_fence::{ + FenceError, PublicationFence, PublicationScope, PUBLICATION_FENCE_CAPABILITY_ARG, + PUBLICATION_FENCE_CAPABILITY_RESPONSE, PUBLICATION_FENCE_ENV, +}; use futures_util::StreamExt; use tokio::io::AsyncWriteExt; use tokio::process::{Child, ChildStdin, ChildStdout}; @@ -106,6 +110,9 @@ pub enum AcpError { #[error("Agent reported error (code {code}): {message}")] AgentError { code: i64, message: String }, + + #[error("Publication fence error: {0}")] + PublicationFence(#[from] FenceError), } /// Build an [`AcpError::AgentError`] from a JSON-RPC error object, @@ -211,6 +218,70 @@ pub struct AcpClient { /// deltas. Both goose and buzz-agent emit this notification; goose gates /// on client capability advertisement, buzz-agent emits unconditionally. goose_usage: UsageTracker, + /// Cross-process publication fence inherited by CLI and MCP descendants. + publication_fence: PublicationFence, + /// Private, process-owned directory containing the fence file. + publication_fence_dir: std::path::PathBuf, + /// Verified PATH value inherited by the ACP child and forced into MCP servers. + managed_cli_path_env: Option, +} + +/// Active managed-turn publication generation. +/// +/// Dropping or explicitly closing the guard marks only its own generation +/// terminal. A stale guard cannot close a newer turn. +pub struct PublicationTurnGuard { + fence: PublicationFence, + generation: Option, +} + +impl PublicationTurnGuard { + /// Mark this turn generation terminal without blocking a Tokio worker. + pub async fn close_with_timeout( + &mut self, + timeout: std::time::Duration, + ) -> Result { + let Some(generation) = self.generation else { + return Ok(true); + }; + let fence = self.fence.clone(); + let terminated = + tokio::task::spawn_blocking(move || fence.terminate_with_timeout(generation, timeout)) + .await + .map_err(|error| { + FenceError::Io(std::io::Error::other(format!( + "publication fence worker failed: {error}" + ))) + })??; + self.generation = None; + Ok(terminated) + } +} + +impl Drop for PublicationTurnGuard { + fn drop(&mut self) { + let Some(generation) = self.generation.take() else { + return; + }; + let fence = self.fence.clone(); + // Panic/task-abort fallback only. Ordinary lifecycle paths await + // `settle_publication_turn`, which can kill and reap a lease holder. + // A detached OS thread keeps this blocking fallback off Tokio workers. + let _ = std::thread::Builder::new() + .name("buzz-publication-fence-close".into()) + .spawn(move || { + let _ = fence.terminate_with_timeout(generation, std::time::Duration::from_secs(5)); + }); + } +} + +/// Result of draining and terminalizing a managed turn's publication fence. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum PublicationCloseOutcome { + /// Existing publication leases drained inside the grace window. + Closed, + /// A lease stalled, so the ACP process group was killed and reaped first. + AgentKilled, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -440,22 +511,62 @@ impl AcpClient { } } - /// Spawn the agent binary as a subprocess and connect to its stdio pipes. + /// Spawn a managed agent after verifying the co-versioned sibling Buzz CLI. /// - /// `has_generated_codex_config` must be true when `codex_network_env()` successfully - /// injected a `CODEX_CONFIG` entry into `extra_env`. The spawn path uses it to - /// trigger the recursive merge + forced `network_access=true` in - /// `build_codex_config_env`. Pass `false` for test spawns and non-Codex agents. + /// The verified sibling directory is prepended to the child's `PATH`, so + /// ordinary `buzz` tool execution cannot resolve an older installation. + pub async fn spawn_managed( + command: &str, + args: &[String], + extra_env: &[(String, String)], + has_generated_codex_config: bool, + ) -> Result { + let managed_cli = verify_sibling_buzz_cli().await?; + Self::spawn_inner( + command, + args, + extra_env, + has_generated_codex_config, + Some(&managed_cli), + ) + .await + } + + /// Spawn the agent binary as a subprocess and connect to its stdio pipes. /// - /// After spawning, call [`initialize`](Self::initialize) before any other method. + /// This unfenced constructor is retained for auth probes and tests that do + /// not execute managed turns. Runtime pools must use [`Self::spawn_managed`]. pub async fn spawn( command: &str, args: &[String], extra_env: &[(String, String)], has_generated_codex_config: bool, + ) -> Result { + Self::spawn_inner(command, args, extra_env, has_generated_codex_config, None).await + } + + async fn spawn_inner( + command: &str, + args: &[String], + extra_env: &[(String, String)], + has_generated_codex_config: bool, + managed_cli: Option<&std::path::Path>, ) -> Result { use std::process::Stdio; + scavenge_stale_fence_dirs(); + let fence_dir = create_private_fence_dir()?; + let fence_path = fence_dir.join("state.json"); + let publication_fence = match PublicationFence::create(&fence_path) { + Ok(fence) => fence, + Err(error) => { + let _ = std::fs::remove_dir_all(&fence_dir); + return Err(error.into()); + } + }; + let mut spawn_artifacts = + SpawnFenceArtifacts::new(publication_fence.clone(), fence_dir.clone()); + let mut cmd = tokio::process::Command::new(command); cmd.args(args) .stdin(Stdio::piped()) @@ -464,7 +575,27 @@ impl AcpClient { .stderr(Stdio::inherit()) // Ensure the child is killed when the AcpClient is dropped (best-effort). // Callers MUST still call shutdown().await for guaranteed cleanup. - .kill_on_drop(true); + .kill_on_drop(true) + // Descendant `buzz messages send` processes consult this file before + // starting and immediately before submitting a managed reply. + .env(PUBLICATION_FENCE_ENV, publication_fence.path()); + + let mut managed_cli_path_env = None; + if let Some(managed_cli) = managed_cli { + let cli_dir = managed_cli.parent().ok_or_else(|| { + AcpError::Protocol("managed Buzz CLI has no parent directory".into()) + })?; + let inherited_path = std::env::var_os("PATH").unwrap_or_default(); + let path = std::env::join_paths( + std::iter::once(cli_dir.to_path_buf()) + .chain(std::env::split_paths(&inherited_path)), + ) + .map_err(|error| { + AcpError::Protocol(format!("failed to construct managed CLI PATH: {error}")) + })?; + managed_cli_path_env = Some(path.to_string_lossy().into_owned()); + cmd.env("PATH", path); + } // Per-persona env vars (e.g., GOOSE_PROVIDER, BUZZ_AGENT_PROVIDER). // For most keys, operator precedence wins: skip injection if already set @@ -513,6 +644,14 @@ impl AcpClient { cmd.env("CODEX_CONFIG", merged); } + // Publication fencing and managed CLI resolution are harness-owned + // invariants. Reapply them after all persona/runtime overlays so a + // reserved extra_env entry cannot replace either value. + cmd.env(PUBLICATION_FENCE_ENV, publication_fence.path()); + if let Some(path) = managed_cli_path_env.as_deref() { + cmd.env("PATH", path); + } + // Spawn the agent in its own process group so SIGKILL doesn't propagate // to the harness's own process group on Unix. // tokio::process::Command::process_group is a stable tokio API (no extra imports needed). @@ -525,15 +664,18 @@ impl AcpClient { let mut child = cmd.spawn()?; - let stdin = child - .stdin - .take() - .ok_or_else(|| AcpError::Protocol("failed to open agent stdin".into()))?; - let stdout = child - .stdout - .take() - .ok_or_else(|| AcpError::Protocol("failed to open agent stdout".into()))?; + let (stdin, stdout) = match (child.stdin.take(), child.stdout.take()) { + (Some(stdin), Some(stdout)) => (stdin, stdout), + _ => { + let _ = child.start_kill(); + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), child.wait()).await; + return Err(AcpError::Protocol( + "failed to open agent stdio pipes".into(), + )); + } + }; + spawn_artifacts.disarm(); Ok(Self { child, stdin, @@ -550,9 +692,67 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), + publication_fence, + publication_fence_dir: fence_dir, + managed_cli_path_env, + }) + } + + /// Return the fence path for per-session MCP environment injection. + pub fn publication_fence_path(&self) -> &std::path::Path { + self.publication_fence.path() + } + + /// Return the verified managed PATH forced into MCP server environments. + pub fn managed_cli_path_env(&self) -> Option<&str> { + self.managed_cli_path_env.as_deref() + } + + /// Open a publication generation for one managed turn without blocking a Tokio worker. + pub async fn begin_publication_turn( + &self, + scope: PublicationScope, + timeout: std::time::Duration, + ) -> Result { + let fence = self.publication_fence.clone(); + let generation = + tokio::task::spawn_blocking(move || fence.begin_with_timeout(scope, timeout)) + .await + .map_err(|error| { + AcpError::PublicationFence(FenceError::Io(std::io::Error::other(format!( + "publication fence worker failed: {error}" + )))) + })??; + Ok(PublicationTurnGuard { + fence: self.publication_fence.clone(), + generation: Some(generation), }) } + /// Drain and terminalize a publication generation, killing the ACP process + /// group first when a descendant holds its lease past `timeout`. + pub async fn settle_publication_turn( + &mut self, + guard: &mut PublicationTurnGuard, + timeout: std::time::Duration, + ) -> Result { + match guard.close_with_timeout(timeout).await { + Ok(_) => Ok(PublicationCloseOutcome::Closed), + Err(FenceError::LockTimeout(_)) => { + tracing::warn!( + timeout = ?timeout, + "publication lease did not drain; killing ACP process group" + ); + self.shutdown().await; + guard + .close_with_timeout(std::time::Duration::from_secs(1)) + .await?; + Ok(PublicationCloseOutcome::AgentKilled) + } + Err(error) => Err(error.into()), + } + } + /// Attach a local observer feed to this ACP client. pub fn set_observer(&mut self, observer: Option, agent_index: usize) { self.observer = observer; @@ -2163,11 +2363,211 @@ pub fn model_in_catalog( }) } +// ─── Fence filesystem lifecycle ─────────────────────────────────────────────── + +const FENCE_DIR_PREFIX: &str = "buzz-acp-publication-fence-"; + +struct SpawnFenceArtifacts { + fence: PublicationFence, + directory: std::path::PathBuf, + armed: bool, +} + +impl SpawnFenceArtifacts { + fn new(fence: PublicationFence, directory: std::path::PathBuf) -> Self { + Self { + fence, + directory, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for SpawnFenceArtifacts { + fn drop(&mut self) { + if self.armed { + cleanup_fence_artifacts(&self.fence, &self.directory); + } + } +} + +fn create_private_fence_dir() -> Result { + let prefix = format!("{FENCE_DIR_PREFIX}{}-", std::process::id()); + let directory = tempfile::Builder::new() + .prefix(&prefix) + .tempdir_in(std::env::temp_dir())? + .keep(); + #[cfg(unix)] + { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o700))?; + let metadata = std::fs::symlink_metadata(&directory)?; + if !metadata.is_dir() + || metadata.file_type().is_symlink() + || metadata.mode() & 0o777 != 0o700 + || metadata.uid() != nix::unistd::Uid::current().as_raw() + { + let _ = std::fs::remove_dir_all(&directory); + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "publication fence directory is not private and process-owned", + )); + } + } + Ok(directory) +} + +fn cleanup_fence_artifacts(fence: &PublicationFence, directory: &std::path::Path) { + let _ = fence.remove(); + let _ = std::fs::remove_dir(directory); +} + +#[cfg(unix)] +fn cleanup_fence_after_process_exit( + fence: PublicationFence, + directory: std::path::PathBuf, + pid: u32, +) { + let _ = std::thread::Builder::new() + .name("buzz-publication-fence-reap".into()) + .spawn(move || { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let process_is_gone = matches!( + nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid as i32), None), + Err(nix::errno::Errno::ESRCH) + ); + if process_is_gone { + cleanup_fence_artifacts(&fence, &directory); + return; + } + if std::time::Instant::now() >= deadline { + return; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + }); +} + +#[cfg(unix)] +fn scavenge_stale_fence_dirs() { + use std::os::unix::fs::MetadataExt; + + let Ok(entries) = std::fs::read_dir(std::env::temp_dir()) else { + return; + }; + let current_uid = nix::unistd::Uid::current().as_raw(); + for entry in entries.flatten() { + let name = entry.file_name(); + let Some(name) = name.to_str() else { + continue; + }; + let Some(rest) = name.strip_prefix(FENCE_DIR_PREFIX) else { + continue; + }; + let Some((pid_text, _random)) = rest.split_once('-') else { + continue; + }; + let Ok(pid) = pid_text.parse::() else { + continue; + }; + let Ok(metadata) = std::fs::symlink_metadata(entry.path()) else { + continue; + }; + if !metadata.is_dir() + || metadata.file_type().is_symlink() + || metadata.uid() != current_uid + || metadata.mode() & 0o777 != 0o700 + { + continue; + } + let process_is_gone = matches!( + nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None), + Err(nix::errno::Errno::ESRCH) + ); + if process_is_gone { + let _ = std::fs::remove_dir_all(entry.path()); + } + } +} + +#[cfg(not(unix))] +fn scavenge_stale_fence_dirs() {} + +// ─── Managed CLI capability ─────────────────────────────────────────────────── + +fn sibling_buzz_cli_name(acp_executable: &std::path::Path) -> std::ffi::OsString { + let file_name = acp_executable + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or_default(); + let stem = file_name.strip_suffix(".exe").unwrap_or(file_name); + let suffix = stem.strip_prefix("buzz-acp-"); + let name = match suffix { + Some(target) => format!("buzz-{target}"), + None => "buzz".to_string(), + }; + if cfg!(windows) { + format!("{name}.exe").into() + } else { + name.into() + } +} + +async fn verify_sibling_buzz_cli() -> Result { + let executable = std::env::current_exe()?; + let directory = executable + .parent() + .ok_or_else(|| AcpError::Protocol("ACP executable has no parent directory".into()))?; + let candidate = directory.join(sibling_buzz_cli_name(&executable)); + verify_buzz_cli_candidate(&candidate).await?; + Ok(candidate) +} + +async fn verify_buzz_cli_candidate(path: &std::path::Path) -> Result<(), AcpError> { + if !path.is_file() { + return Err(AcpError::Protocol(format!( + "co-versioned managed Buzz CLI is missing: {}", + path.display() + ))); + } + + let mut command = tokio::process::Command::new(path); + command + .arg(PUBLICATION_FENCE_CAPABILITY_ARG) + .stdin(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .kill_on_drop(true); + let output = tokio::time::timeout(std::time::Duration::from_secs(2), command.output()) + .await + .map_err(|_| { + AcpError::Protocol(format!( + "managed Buzz CLI capability probe timed out: {}", + path.display() + )) + })??; + let response = String::from_utf8_lossy(&output.stdout); + if !output.status.success() + || response.trim_end_matches(['\r', '\n']) != PUBLICATION_FENCE_CAPABILITY_RESPONSE + { + return Err(AcpError::Protocol(format!( + "managed Buzz CLI lacks publication-fence capability: {}", + path.display() + ))); + } + Ok(()) +} + // ─── Drop: kill child process ───────────────────────────────────────────────── impl Drop for AcpClient { fn drop(&mut self) { // Best-effort SIGKILL + reap. We cannot `await` in Drop (sync context). + let child_pid = self.child.id(); // Kill the process group when possible so subprocesses don't leak. // Callers SHOULD still call `shutdown().await` for guaranteed reaping. match self.child.id() { @@ -2176,9 +2576,21 @@ impl Drop for AcpClient { let _ = self.child.start_kill(); } } - // Non-blocking reap attempt — prevents zombie accumulation in the - // common case where SIGKILL takes effect before Drop returns. - let _ = self.child.try_wait(); + // Remove the fence path only after confirmed reap. If the child has + // not exited yet, a later harness startup scavenges this private, + // ownership-checked directory once its PID is gone. + if matches!(self.child.try_wait(), Ok(Some(_))) { + cleanup_fence_artifacts(&self.publication_fence, &self.publication_fence_dir); + } else { + #[cfg(unix)] + if let Some(pid) = child_pid { + cleanup_fence_after_process_exit( + self.publication_fence.clone(), + self.publication_fence_dir.clone(), + pid, + ); + } + } } } @@ -2199,9 +2611,25 @@ fn kill_process_group(pid: u32) -> bool { killpg(Pid::from_raw(pid as i32), Signal::SIGKILL).is_ok() } -/// Fallback for non-Unix: process-group kill not available. -/// Returns `false` so the caller falls back to `child.start_kill()`. -#[cfg(not(unix))] +/// Kill the Windows process tree rooted at `pid`. `taskkill /T /F` is the +/// safe equivalent of Unix process-group termination and is already used by +/// the desktop managed-agent lifecycle. A failure returns false so callers +/// kill the direct child and, critically, fence settlement still fails closed +/// if a descendant retains the publication lease. +#[cfg(windows)] +fn kill_process_group(pid: u32) -> bool { + use std::os::windows::process::CommandExt; + + const CREATE_NO_WINDOW: u32 = 0x0800_0000; + std::process::Command::new("taskkill") + .args(["/T", "/F", "/PID", &pid.to_string()]) + .creation_flags(CREATE_NO_WINDOW) + .status() + .is_ok_and(|status| status.success()) +} + +/// Fallback for targets without Unix process groups or Windows `taskkill`. +#[cfg(not(any(unix, windows)))] fn kill_process_group(_pid: u32) -> bool { false } @@ -2248,6 +2676,340 @@ mod tests { assert_eq!(StopReason::from_str("endturn"), None); // no camelCase — still unknown } + #[cfg(unix)] + #[tokio::test] + async fn fence_directory_is_private_and_removed_after_confirmed_reap() { + use std::os::unix::fs::MetadataExt; + + let mut client = spawn_inert_client().await; + let directory = client.publication_fence_dir.clone(); + let metadata = std::fs::symlink_metadata(&directory).expect("fence directory metadata"); + assert_eq!(metadata.mode() & 0o777, 0o700); + assert!(!metadata.file_type().is_symlink()); + + client.shutdown().await; + drop(client); + assert!(!directory.exists()); + } + + #[cfg(unix)] + #[test] + fn scavenger_removes_only_private_owned_dead_pid_directory() { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::Builder::new() + .prefix(&format!("{FENCE_DIR_PREFIX}2000000000-")) + .tempdir_in(std::env::temp_dir()) + .expect("stale tempdir") + .keep(); + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o700)) + .expect("private stale dir"); + std::fs::write(directory.join("state.json"), b"stale").expect("stale state"); + + scavenge_stale_fence_dirs(); + assert!(!directory.exists()); + } + + #[tokio::test] + async fn acp_child_inherits_publication_fence_path() { + let mut client = AcpClient::spawn( + "bash", + &[ + "-c".to_string(), + format!("printf '%s\\n' \"${}\"", PUBLICATION_FENCE_ENV), + ], + &[], + false, + ) + .await + .expect("spawn env probe"); + let expected = client + .publication_fence_path() + .to_string_lossy() + .into_owned(); + let observed = client + .reader + .next() + .await + .expect("probe line") + .expect("valid probe line"); + assert_eq!(observed, expected); + client.shutdown().await; + } + + #[test] + fn sibling_cli_name_supports_plain_and_tauri_sidecar_binaries() { + let extension = if cfg!(windows) { ".exe" } else { "" }; + assert_eq!( + sibling_buzz_cli_name(std::path::Path::new(&format!("buzz-acp{extension}"))), + std::ffi::OsString::from(format!("buzz{extension}")) + ); + assert_eq!( + sibling_buzz_cli_name(std::path::Path::new(&format!( + "buzz-acp-aarch64-apple-darwin{extension}" + ))), + std::ffi::OsString::from(format!("buzz-aarch64-apple-darwin{extension}")) + ); + } + + #[cfg(unix)] + fn write_fake_buzz_cli(path: &std::path::Path, response: &str, success: bool) { + use std::os::unix::fs::PermissionsExt; + + let exit = if success { 0 } else { 1 }; + std::fs::write( + path, + format!( + "#!/bin/sh\nif [ \"$1\" = \"{PUBLICATION_FENCE_CAPABILITY_ARG}\" ]; then printf '%s\\n' '{response}'; exit {exit}; fi\nexit 64\n" + ), + ) + .expect("write fake buzz"); + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)) + .expect("make fake buzz executable"); + } + + #[cfg(unix)] + #[tokio::test] + async fn managed_cli_probe_rejects_older_binary_and_pins_child_path() { + let temp = tempfile::tempdir().expect("tempdir"); + let old_dir = temp.path().join("old"); + let current_dir = temp.path().join("current"); + std::fs::create_dir_all(&old_dir).expect("old dir"); + std::fs::create_dir_all(¤t_dir).expect("current dir"); + let old = old_dir.join("buzz"); + let current = current_dir.join("buzz"); + write_fake_buzz_cli(&old, "older-buzz", false); + write_fake_buzz_cli(¤t, PUBLICATION_FENCE_CAPABILITY_RESPONSE, true); + + assert!(matches!( + verify_buzz_cli_candidate(&old).await, + Err(AcpError::Protocol(message)) if message.contains("lacks publication-fence capability") + )); + verify_buzz_cli_candidate(¤t) + .await + .expect("current CLI capability"); + + let mut client = AcpClient::spawn_inner( + "sh", + &["-c".into(), "command -v buzz".into()], + &[], + false, + Some(¤t), + ) + .await + .expect("spawn PATH probe"); + let resolved = client + .reader + .next() + .await + .expect("PATH probe line") + .expect("valid PATH probe line"); + assert_eq!(std::path::Path::new(&resolved), current); + client.shutdown().await; + } + + #[tokio::test] + async fn acp_turn_guard_terminalizes_descendant_publication_attempt() { + let mut client = spawn_inert_client().await; + let path = client.publication_fence_path().to_path_buf(); + let channel_id = uuid::Uuid::new_v4(); + let mut guard = client + .begin_publication_turn( + PublicationScope { + turn_id: "turn-a".to_string(), + channel_id: Some(channel_id), + reply_to: Some("root-a".to_string()), + }, + std::time::Duration::from_secs(1), + ) + .await + .expect("begin publication turn"); + let attempt = + buzz_publication_fence::PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture active publication"); + + assert!(guard + .close_with_timeout(std::time::Duration::from_secs(1)) + .await + .expect("close publication turn")); + assert!(matches!( + attempt.acquire(), + Err(buzz_publication_fence::FenceError::Terminal) + )); + client.shutdown().await; + } + + #[tokio::test] + async fn aborted_turn_task_terminalizes_generation_via_drop_fallback() { + let mut client = spawn_inert_client().await; + let path = client.publication_fence_path().to_path_buf(); + let channel_id = uuid::Uuid::new_v4(); + let guard = client + .begin_publication_turn( + PublicationScope { + turn_id: "turn-abort".into(), + channel_id: Some(channel_id), + reply_to: Some("root-a".into()), + }, + std::time::Duration::from_secs(1), + ) + .await + .expect("begin publication turn"); + let attempt = + buzz_publication_fence::PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture active attempt"); + let task = tokio::spawn(async move { + let _guard = guard; + std::future::pending::<()>().await; + }); + + task.abort(); + let _ = task.await; + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if matches!( + buzz_publication_fence::PublicationAttempt::capture( + &path, + channel_id, + Some("root-a"), + ), + Err(buzz_publication_fence::FenceError::Terminal) + ) { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("drop fallback terminalizes boundedly"); + assert!(matches!( + attempt.acquire(), + Err(buzz_publication_fence::FenceError::Terminal) + )); + client.shutdown().await; + } + + #[test] + fn publication_lock_holder_child() { + const PARENT_ENV: &str = "BUZZ_TEST_HOLD_PUBLICATION_LEASE"; + const GRANDCHILD_ENV: &str = "BUZZ_TEST_HOLD_PUBLICATION_LEASE_GRANDCHILD"; + let is_parent = std::env::var_os(PARENT_ENV).is_some(); + let is_grandchild = std::env::var_os(GRANDCHILD_ENV).is_some(); + if !is_parent && !is_grandchild { + return; + } + use std::io::{BufRead, Write}; + + if is_parent && !is_grandchild { + let mut line = String::new(); + std::io::BufReader::new(std::io::stdin()) + .read_line(&mut line) + .expect("read parent signal"); + let executable = std::env::current_exe().expect("test executable"); + let mut grandchild = std::process::Command::new(executable) + .args([ + "--exact", + "acp::tests::publication_lock_holder_child", + "--nocapture", + ]) + .env(GRANDCHILD_ENV, "1") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::inherit()) + .stderr(std::process::Stdio::inherit()) + .spawn() + .expect("spawn publication-lease grandchild"); + std::thread::sleep(std::time::Duration::from_secs(60)); + let _ = grandchild.kill(); + let _ = grandchild.wait(); + return; + } + + let channel_id = uuid::Uuid::parse_str( + &std::env::var("BUZZ_TEST_PUBLICATION_CHANNEL").expect("channel env"), + ) + .expect("valid channel"); + let _lease = buzz_publication_fence::PublicationAttempt::capture_from_env( + channel_id, + Some("root-a"), + ) + .expect("capture publication") + .expect("managed fence") + .acquire() + .expect("hold publication lease"); + println!("publication-lease-held"); + std::io::stdout().flush().expect("flush ready line"); + std::thread::sleep(std::time::Duration::from_secs(60)); + } + + #[tokio::test] + async fn stalled_descendant_lease_kills_process_and_terminalizes_boundedly() { + let channel_id = uuid::Uuid::new_v4(); + let executable = std::env::current_exe().expect("test executable"); + let mut client = AcpClient::spawn( + executable.to_str().expect("utf-8 test path"), + &[ + "--exact".into(), + "acp::tests::publication_lock_holder_child".into(), + "--nocapture".into(), + ], + &[ + ("BUZZ_TEST_HOLD_PUBLICATION_LEASE".into(), "1".into()), + ( + "BUZZ_TEST_PUBLICATION_CHANNEL".into(), + channel_id.to_string(), + ), + ], + false, + ) + .await + .expect("spawn lease holder"); + let path = client.publication_fence_path().to_path_buf(); + let mut guard = client + .begin_publication_turn( + PublicationScope { + turn_id: "turn-a".into(), + channel_id: Some(channel_id), + reply_to: Some("root-a".into()), + }, + std::time::Duration::from_secs(1), + ) + .await + .expect("begin publication turn"); + let late_attempt = + buzz_publication_fence::PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture attempt before terminal transition"); + + client + .stdin + .write_all(b"hold\n") + .await + .expect("release child to lock"); + loop { + let line = client + .reader + .next() + .await + .expect("child output") + .expect("valid child output"); + if line == "publication-lease-held" { + break; + } + } + + let started = std::time::Instant::now(); + let outcome = client + .settle_publication_turn(&mut guard, std::time::Duration::from_millis(50)) + .await + .expect("kill and terminalize"); + + assert_eq!(outcome, PublicationCloseOutcome::AgentKilled); + assert!(started.elapsed() < std::time::Duration::from_secs(3)); + assert!(matches!( + late_attempt.acquire(), + Err(buzz_publication_fence::FenceError::Terminal) + )); + } + #[test] fn stop_reason_is_case_insensitive() { // Agents may send uppercase or mixed-case variants — all should parse correctly. diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 403512a322..dde5e88362 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2736,6 +2736,30 @@ fn event_mentions_agent(event: &nostr::Event, agent_pubkey_hex: &str) -> bool { }) } +fn control_command_content_matches(content: &str, command: &str) -> bool { + let content = content.trim(); + if content == command { + return true; + } + + // The desktop composer keeps a selected mention in visible content so it + // can derive the event's `p` tag. When the toolbar inserts that mention at + // the end of an already-authored command, it serializes as + // ` @`. Accept only that single-line suffix form; + // author and exact-agent `p`-tag checks remain separate hard gates. + let Some(mention) = content + .strip_prefix(command) + .and_then(|suffix| suffix.strip_prefix(' ')) + else { + return false; + }; + + mention.starts_with('@') + && mention.len() > 1 + && !mention[1..].contains('@') + && !mention.contains(['\r', '\n']) +} + fn is_owner_control_command( event: &nostr::Event, kind_u32: u32, @@ -2743,7 +2767,7 @@ fn is_owner_control_command( agent_pubkey_hex: &str, ) -> bool { kind_u32 == KIND_STREAM_MESSAGE - && event.content.trim() == command + && control_command_content_matches(&event.content, command) && event_mentions_agent(event, agent_pubkey_hex) } @@ -3371,6 +3395,7 @@ fn handle_prompt_result( | acp::AcpError::WriteTimeout(_) | acp::AcpError::Timeout(_) | acp::AcpError::Protocol(_) + | acp::AcpError::PublicationFence(_) ); let error_code = match &e { acp::AcpError::AgentError { code, .. } => Some(*code), @@ -3783,7 +3808,7 @@ async fn initialize_agent_pool( // Attempt each spawn under a 60-second timeout; a partial pool is valid. let mut agent_slots: Vec> = Vec::with_capacity(startup.agents as usize); for i in 0..startup.agents as usize { - let spawn_result = AcpClient::spawn( + let spawn_result = AcpClient::spawn_managed( &startup.command, &startup.args, &startup.extra_env, @@ -3891,7 +3916,7 @@ async fn spawn_and_init( agent_index: usize, observer: Option, ) -> Result<(AcpClient, u32, String)> { - let mut acp = AcpClient::spawn(command, args, extra_env, has_generated_codex_config) + let mut acp = AcpClient::spawn_managed(command, args, extra_env, has_generated_codex_config) .await .map_err(|e| anyhow::anyhow!("failed to spawn agent: {e}"))?; acp.set_observer(observer, agent_index); @@ -4314,6 +4339,59 @@ mod owner_control_command_tests { )); } + #[test] + fn owner_control_command_accepts_ui_suffix_mention() { + let agent = "ab".repeat(32); + let event = make_event( + KIND_STREAM_MESSAGE, + "!cancel @Unitus SEO OS Representative", + Some(&agent), + ); + + assert!(is_owner_control_command( + &event, + KIND_STREAM_MESSAGE, + "!cancel", + &agent + )); + } + + #[test] + fn owner_control_command_rejects_non_ui_content_shapes() { + let agent = "ab".repeat(32); + for content in [ + "please !cancel @Agent", + "!cancel please @Agent", + "@Agent !cancel", + "!cancel @Agent", + "!cancel @Agent @Other", + "!cancel @Agent\nmore", + ] { + let event = make_event(KIND_STREAM_MESSAGE, content, Some(&agent)); + assert!( + !is_owner_control_command(&event, KIND_STREAM_MESSAGE, "!cancel", &agent), + "unexpected match for {content:?}" + ); + } + + let no_mention_tag = make_event(KIND_STREAM_MESSAGE, "!cancel @Agent", None); + assert!(!is_owner_control_command( + &no_mention_tag, + KIND_STREAM_MESSAGE, + "!cancel", + &agent + )); + + let other_agent = "cd".repeat(32); + let wrong_agent_tag = make_event(KIND_STREAM_MESSAGE, "!cancel @Agent", Some(&other_agent)); + assert!(!is_owner_control_command( + &wrong_agent_tag, + KIND_STREAM_MESSAGE, + "!cancel", + &agent + )); + } + #[test] fn mode_gate_signal_maps_handling_to_control_signal() { let owner = "a".repeat(64); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 158477c0af..a0e5747070 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -31,7 +31,8 @@ use uuid::Uuid; use crate::acp::{ extract_model_config_options, extract_model_state, model_in_catalog, - resolve_model_switch_method, AcpClient, AcpError, McpServer, ModelSwitchMethod, StopReason, + resolve_model_switch_method, AcpClient, AcpError, McpServer, ModelSwitchMethod, + PublicationCloseOutcome, PublicationTurnGuard, StopReason, }; use crate::config::{compose_session_title, DedupMode, PermissionMode}; use crate::observer; @@ -817,6 +818,46 @@ const MODEL_SWITCH_TIMEOUT: Duration = Duration::from_secs(5); /// [`classify_control_cancel_failure`]. const CONTROL_CANCEL_GRACE: Duration = Duration::from_secs(5); +/// Maximum time a terminal transition waits for an in-flight publication. +/// On expiry, the ACP process group is killed and reaped before a final retry. +const PUBLICATION_LEASE_DRAIN_GRACE: Duration = Duration::from_secs(1); + +async fn begin_publication_turn_bounded( + agent: &mut OwnedAgent, + scope: buzz_publication_fence::PublicationScope, +) -> Result { + match agent + .acp + .begin_publication_turn(scope, PUBLICATION_LEASE_DRAIN_GRACE) + .await + { + Ok(guard) => Ok(guard), + Err( + error @ AcpError::PublicationFence(buzz_publication_fence::FenceError::LockTimeout(_)), + ) => { + agent.acp.shutdown().await; + Err(error) + } + Err(error) => Err(error), + } +} + +async fn settle_publication_turn_bounded( + agent: &mut OwnedAgent, + guard: &mut PublicationTurnGuard, +) -> Result<(), AcpError> { + match agent + .acp + .settle_publication_turn(guard, PUBLICATION_LEASE_DRAIN_GRACE) + .await? + { + PublicationCloseOutcome::Closed => Ok(()), + PublicationCloseOutcome::AgentKilled => Err(AcpError::PublicationFence( + buzz_publication_fence::FenceError::LockTimeout(PUBLICATION_LEASE_DRAIN_GRACE), + )), + } +} + /// Timeout for permission-mode requests (`session/set_config_option` with `configId: "mode"`). const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5); @@ -863,6 +904,30 @@ async fn resolve_new_session_channel_context( (is_dm, title_channel) } +fn inject_managed_publication_env( + servers: &mut [McpServer], + fence_path: &std::path::Path, + managed_path: Option<&str>, +) { + let fence_value = fence_path.to_string_lossy().into_owned(); + for server in servers { + server.env.retain(|entry| { + entry.name != buzz_publication_fence::PUBLICATION_FENCE_ENV + && (managed_path.is_none() || entry.name != "PATH") + }); + server.env.push(crate::acp::EnvVar { + name: buzz_publication_fence::PUBLICATION_FENCE_ENV.to_string(), + value: fence_value.clone(), + }); + if let Some(value) = managed_path { + server.env.push(crate::acp::EnvVar { + name: "PATH".to_string(), + value: value.to_string(), + }); + } + } +} + /// Create a new ACP session via `session_new_full()`, populate model capabilities /// on the agent (first session only), and apply `desired_model` if set. /// @@ -899,11 +964,18 @@ async fn create_session_and_apply_model( .as_deref() .map(|agent_name| compose_session_title(agent_name, channel_name)); + let mut mcp_servers = ctx.mcp_servers.clone(); + inject_managed_publication_env( + &mut mcp_servers, + agent.acp.publication_fence_path(), + agent.acp.managed_cli_path_env(), + ); + let resp = agent .acp .session_new_full( &ctx.cwd, - ctx.mcp_servers.clone(), + mcp_servers, session_new_system_prompt( is_goose, agent.protocol_version, @@ -1658,6 +1730,54 @@ pub async fn run_prompt_task( }), ); + // Open the managed publication generation before any initial-message or + // turn prompt can launch a descendant `buzz messages send`. The ordinary + // reply anchor is refined after channel/profile context is resolved below. + let mut publication_guard = match begin_publication_turn_bounded( + &mut agent, + buzz_publication_fence::PublicationScope { + turn_id: turn_id.clone(), + channel_id: match &source { + PromptSource::Channel(channel_id) => Some(*channel_id), + PromptSource::Heartbeat => None, + }, + reply_to: None, + }, + ) + .await + { + Ok(guard) => guard, + Err(error) => { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(error), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + }; + + macro_rules! settle_publication_or_return_error { + () => { + if let Err(error) = + settle_publication_turn_bounded(&mut agent, &mut publication_guard).await + { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(error), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + }; + } + if is_new_session { if let (PromptSource::Channel(cid), Some(ref initial_msg)) = (&source, &ctx.initial_message) { @@ -1701,12 +1821,14 @@ pub async fn run_prompt_task( match init_result { Ok(stop_reason) => { + settle_publication_or_return_error!(); tracing::info!( target: "pool::session", "initial_message complete for channel {cid}: {stop_reason:?}" ); } Err(AcpError::AgentExited) => { + settle_publication_or_return_error!(); agent.state.invalidate_all(); send_prompt_result( &result_tx, @@ -1719,6 +1841,7 @@ pub async fn run_prompt_task( return; } Err(AcpError::IdleTimeout(_)) => { + settle_publication_or_return_error!(); tracing::warn!( target: "pool::session", "initial_message idle timeout ({}s) for channel {cid} — cancelling", @@ -1763,6 +1886,7 @@ pub async fn run_prompt_task( return; } Err(AcpError::HardTimeout { silence }) => { + settle_publication_or_return_error!(); let recently_active = silence < RECENT_ACTIVITY_WINDOW; tracing::error!( target: "pool::session", @@ -1781,6 +1905,7 @@ pub async fn run_prompt_task( return; } Err(e) => { + settle_publication_or_return_error!(); tracing::error!( target: "pool::session", "initial_message failed for channel {cid}: {e} — invalidating session" @@ -1833,6 +1958,48 @@ pub async fn run_prompt_task( let profile_lookup = fetch_prompt_profile_lookup(b, conversation_context.as_ref(), &ctx.rest_client).await; + let reply_anchor = crate::queue::publication_reply_anchor( + b, + channel_info.as_ref(), + profile_lookup.as_ref(), + ); + if let Err(error) = + settle_publication_turn_bounded(&mut agent, &mut publication_guard).await + { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(error), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + publication_guard = match begin_publication_turn_bounded( + &mut agent, + buzz_publication_fence::PublicationScope { + turn_id: turn_id.clone(), + channel_id: Some(b.channel_id), + reply_to: reply_anchor, + }, + ) + .await + { + Ok(guard) => guard, + Err(error) => { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(error), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + }; + let known_names: Vec<&str> = profile_lookup .iter() .flat_map(|lookup| lookup.values()) @@ -1866,6 +2033,7 @@ pub async fn run_prompt_task( } else { // Should not happen — batch is None only for heartbeats which have prompt_text. // Return the agent to the pool to prevent a permanent slot leak. + settle_publication_or_return_error!(); tracing::error!("run_prompt_task: no batch and no prompt_text — returning agent"); send_prompt_result( &result_tx, @@ -1942,13 +2110,64 @@ pub async fn run_prompt_task( mode = rx => { let control_signal = mode.unwrap_or(ControlSignal::Cancel); // Land the model switch before any cancel/requeue work: setting - // `desired_model` here means the fresh session created by the - // requeued turn (busy) or the next turn (already-completed) - // applies the new model. Runtime-only — never persisted. + // `desired_model` here means a replacement process preserves the + // requested runtime-only switch too. if let ControlSignal::SwitchModel(ref model_id) = control_signal { agent.desired_model = Some(model_id.clone()); agent.model_overridden = true; } + // Terminalize publication before asking ACP to cancel. Lock + // waiting runs off executor workers and is bounded. If a + // descendant submit stalls, kill/reap the process group, retry + // terminalization, and classify the turn as poisoned. + match agent + .acp + .settle_publication_turn( + &mut publication_guard, + PUBLICATION_LEASE_DRAIN_GRACE, + ) + .await + { + Ok(PublicationCloseOutcome::Closed) => {} + Ok(PublicationCloseOutcome::AgentKilled) => { + agent.state.invalidate_all(); + let retry_batch = + requeue_cancelled_batch(&ctx, control_signal, batch); + let usage = agent.acp.take_turn_usage(); + publish_agent_turn_metric( + &ctx, + usage, + observer_channel_id, + &session_id, + &turn_id, + Some(buzz_core::agent_turn_metric::StopReason::Error), + ) + .await; + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::CancelDrainTimeout( + PUBLICATION_LEASE_DRAIN_GRACE, + ), + retry_batch, + ); + return; + } + Err(error) => { + agent.acp.shutdown().await; + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(error), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + } // Control signal received. Guard against Race 1: the turn may // have completed naturally just as cancel fired. if agent.acp.has_in_flight_prompt() { @@ -2080,6 +2299,10 @@ pub async fn run_prompt_task( } }; + // Normal completion and timeout paths cross the same terminal boundary + // before metrics, retries, respawn, or pool return can advance the slot. + settle_publication_or_return_error!(); + match prompt_result { Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); @@ -3965,6 +4188,51 @@ mod tests { use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; + #[test] + fn mcp_servers_receive_exact_fence_and_managed_path_without_override() { + let mut servers = vec![McpServer { + name: "tools".into(), + command: "tools".into(), + args: vec![], + env: vec![ + crate::acp::EnvVar { + name: "KEEP".into(), + value: "yes".into(), + }, + crate::acp::EnvVar { + name: buzz_publication_fence::PUBLICATION_FENCE_ENV.into(), + value: "/attacker/old-fence".into(), + }, + crate::acp::EnvVar { + name: "PATH".into(), + value: "/attacker/bin".into(), + }, + ], + }]; + let expected = std::path::Path::new("/private/fence/state.json"); + + inject_managed_publication_env(&mut servers, expected, Some("/co-versioned/bin:/usr/bin")); + + assert!(servers[0] + .env + .iter() + .any(|entry| entry.name == "KEEP" && entry.value == "yes")); + let fence_entries: Vec<_> = servers[0] + .env + .iter() + .filter(|entry| entry.name == buzz_publication_fence::PUBLICATION_FENCE_ENV) + .collect(); + assert_eq!(fence_entries.len(), 1); + assert_eq!(fence_entries[0].value, expected.to_string_lossy()); + let path_entries: Vec<_> = servers[0] + .env + .iter() + .filter(|entry| entry.name == "PATH") + .collect(); + assert_eq!(path_entries.len(), 1); + assert_eq!(path_entries[0].value, "/co-versioned/bin:/usr/bin"); + } + // These pin the initial_message dispatch path (run_prompt_task, ~line 855): // a legacy agent WITH a base_prompt must get [Base] prepended to the user // message. This is the exact regression that shipped in the round-2 bug. diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 029bf86dbf..801a292cef 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1383,6 +1383,35 @@ pub(crate) fn base_section(base_prompt: &str) -> String { format!("[Base]\n{}", base_prompt.trim_end()) } +/// Resolve the ordinary managed-reply anchor for publication fencing. +/// +/// This is the same destination rendered into the `[Context]` instructions. +/// Keeping one resolver prevents the CLI fence and the prompt from disagreeing. +pub(crate) fn publication_reply_anchor( + batch: &FlushBatch, + channel_info: Option<&PromptChannelInfo>, + profile_lookup: Option<&PromptProfileLookup>, +) -> Option { + let last_event = batch.events.last()?; + let thread_tags = parse_thread_tags(&last_event.event); + let is_dm = channel_info + .map(|info| info.channel_type == "dm") + .unwrap_or(false); + if is_dm { + thread_tags + .root_event_id + .is_some() + .then(|| last_event.event.id.to_hex()) + } else { + resolve_reply_anchor( + &last_event.event.pubkey.to_hex(), + &thread_tags, + &last_event.event.id.to_hex(), + profile_lookup, + ) + } +} + /// Format a [`FlushBatch`] into the per-section prompt blocks for the agent. /// /// Produces a stable prompt with these sections (in order): @@ -1464,20 +1493,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec, } +async fn submit_fenced_message( + client: &BuzzClient, + event: nostr::Event, + publication_attempt: Option, +) -> Result { + // Hold the shared lease through the relay submission. A terminal transition + // takes the exclusive lock, so once cancellation settles no later managed + // publication can cross this boundary. + let _publication_lease = publication_attempt + .map(|attempt| attempt.acquire()) + .transpose() + .map_err(|error| CliError::Other(error.to_string()))?; + client.submit_event(event).await +} + pub async fn cmd_send_message( client: &BuzzClient, mut p: SendMessageParams, ) -> Result<(), CliError> { + if let Some(ref r) = p.reply_to { + validate_hex64(r)?; + } + let channel_uuid = parse_uuid(&p.channel_id)?; + + // Managed ACP subprocesses capture their active publication generation + // before reading stdin, uploading files, or resolving context. Standalone + // CLI use has no fence env and remains unchanged. + let publication_attempt = buzz_publication_fence::PublicationAttempt::capture_from_env( + channel_uuid, + p.reply_to.as_deref(), + ) + .map_err(|error| CliError::Other(error.to_string()))?; + // Allow '-' to read content from stdin. This keeps callers from having to // jam shell-metacharacter-heavy text (backticks, $vars, etc.) through argv // quoting — the source of countless self-inflicted command-substitution // bugs for agent and human users alike. p.content = read_or_stdin(&p.content)?; validate_content_size(&p.content)?; - if let Some(ref r) = p.reply_to { - validate_hex64(r)?; - } - let channel_uuid = parse_uuid(&p.channel_id)?; let explicit_mentions = normalize_explicit_mentions(&p.mentions)?; let stripped = strip_code_regions(&p.content); @@ -679,7 +704,7 @@ pub async fn cmd_send_message( let event = client.sign_event(builder)?; let emitted_mentions = event_mention_pubkeys(&event); - let resp = client.submit_event(event).await?; + let resp = submit_fenced_message(client, event, publication_attempt).await?; let mut output: serde_json::Value = serde_json::from_str(&normalize_write_response(&resp)) .unwrap_or_else(|_| serde_json::json!({ "response": resp })); if let Some(object) = output.as_object_mut() { @@ -724,6 +749,24 @@ pub async fn cmd_send_diff_message(client: &BuzzClient, p: SendDiffParams) -> Re let channel_uuid = parse_uuid(&p.channel_id)?; + // Match ordinary message publication: capture the managed generation before + // reading stdin or resolving thread context, then hold its lease across the + // final relay submission. Standalone CLI use has no fence env. + let publication_attempt = buzz_publication_fence::PublicationAttempt::capture_from_env( + channel_uuid, + p.reply_to.as_deref(), + ) + .map_err(|error| CliError::Other(error.to_string()))?; + + send_diff_message_with_attempt(client, p, channel_uuid, publication_attempt).await +} + +async fn send_diff_message_with_attempt( + client: &BuzzClient, + p: SendDiffParams, + channel_uuid: uuid::Uuid, + publication_attempt: Option, +) -> Result<(), CliError> { // Read diff from stdin if "--diff -" let diff_content = read_or_stdin(&p.diff)?; @@ -775,7 +818,7 @@ pub async fn cmd_send_diff_message(client: &BuzzClient, p: SendDiffParams) -> Re let event = client.sign_event(builder)?; - let resp = client.submit_event(event).await?; + let resp = submit_fenced_message(client, event, publication_attempt).await?; println!("{}", normalize_write_response(&resp)); Ok(()) } @@ -995,8 +1038,10 @@ mod tests { use super::{ event_mention_pubkeys, find_root_from_tags, match_profiles_by_name, merge_message_mentions, missing_members, normalize_explicit_mentions, parse_member_pubkeys, - resolve_names_to_pubkeys, + resolve_names_to_pubkeys, send_diff_message_with_attempt, submit_fenced_message, + SendDiffParams, }; + use buzz_publication_fence::{PublicationAttempt, PublicationFence, PublicationScope}; use buzz_sdk::mentions::{ extract_at_mentions_with_known, extract_at_names, match_names_to_profiles, MentionProfile, }; @@ -1012,6 +1057,130 @@ mod tests { const PK_VALID_B: &str = "c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05"; const PK_VALID_C: &str = "f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68"; + #[tokio::test] + async fn terminal_fence_rejects_message_at_final_submit_boundary() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = uuid::Uuid::new_v4(); + let generation = fence + .begin(PublicationScope { + turn_id: "turn-a".to_string(), + channel_id: Some(channel_id), + reply_to: Some(ID_A.to_string()), + }) + .expect("begin turn"); + let attempt = PublicationAttempt::capture(&path, channel_id, Some(ID_A)) + .expect("capture active attempt"); + assert!(fence.terminate(generation).expect("terminate turn")); + + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "late reply") + .sign_with_keys(&keys) + .expect("sign event"); + let client = + crate::client::BuzzClient::new("http://127.0.0.1:1".to_string(), keys, None, None) + .expect("construct client"); + + let error = submit_fenced_message(&client, event, Some(attempt)) + .await + .expect_err("terminal fence must reject before network submit"); + assert!(matches!( + error, + crate::error::CliError::Other(ref message) + if message.contains("terminal") && message.contains("rejected") + )); + } + + #[tokio::test] + async fn active_matching_fence_reaches_network_submit() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = uuid::Uuid::new_v4(); + fence + .begin(PublicationScope { + turn_id: "turn-a".into(), + channel_id: Some(channel_id), + reply_to: Some(ID_A.into()), + }) + .expect("begin turn"); + let attempt = PublicationAttempt::capture(&path, channel_id, Some(ID_A)) + .expect("capture active attempt"); + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "active reply") + .sign_with_keys(&keys) + .expect("sign event"); + let client = crate::client::BuzzClient::new("http://127.0.0.1:1".into(), keys, None, None) + .expect("construct client"); + + let error = submit_fenced_message(&client, event, Some(attempt)) + .await + .expect_err("unreachable relay proves submit was attempted"); + assert!(!error.to_string().contains("publication rejected")); + } + + #[tokio::test] + async fn standalone_send_without_fence_reaches_network_submit() { + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "standalone reply") + .sign_with_keys(&keys) + .expect("sign event"); + let client = crate::client::BuzzClient::new("http://127.0.0.1:1".into(), keys, None, None) + .expect("construct client"); + + let error = submit_fenced_message(&client, event, None) + .await + .expect_err("unreachable relay proves submit was attempted"); + assert!(!error.to_string().contains("publication rejected")); + } + + #[tokio::test] + async fn terminal_fence_rejects_send_diff_at_final_submit_boundary() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = uuid::Uuid::new_v4(); + let generation = fence + .begin(PublicationScope { + turn_id: "turn-diff".to_string(), + channel_id: Some(channel_id), + reply_to: None, + }) + .expect("begin turn"); + let attempt = PublicationAttempt::capture(&path, channel_id, None) + .expect("capture active diff publication"); + assert!(fence.terminate(generation).expect("terminate turn")); + + let keys = nostr::Keys::generate(); + let client = + crate::client::BuzzClient::new("http://127.0.0.1:1".to_string(), keys, None, None) + .expect("construct client"); + let params = SendDiffParams { + channel_id: channel_id.to_string(), + diff: "@@ -1 +1 @@\n-old\n+new\n".to_string(), + repo_url: "https://example.invalid/repo".to_string(), + commit_sha: ID_A.to_string(), + file_path: Some("src/lib.rs".to_string()), + parent_commit_sha: None, + source_branch: None, + target_branch: None, + pr_number: None, + language: Some("rust".to_string()), + description: Some("fenced diff".to_string()), + reply_to: None, + }; + + let error = send_diff_message_with_attempt(&client, params, channel_id, Some(attempt)) + .await + .expect_err("terminal fence must reject diff before network submit"); + assert!(matches!( + error, + crate::error::CliError::Other(ref message) + if message.contains("terminal") && message.contains("rejected") + )); + } + #[test] fn root_marker_wins_over_reply_marker() { let tags = json!([ diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 0726406d29..87a2f47d12 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -173,6 +173,9 @@ pub enum OutputFormat { #[derive(Subcommand)] enum Cmd { + /// Internal harness capability probe. + #[command(name = "__publication-fence-capability", hide = true)] + PublicationFenceCapability, /// Draft owner-reviewed agent creation and updates #[command(subcommand)] Agents(AgentsCmd), @@ -1769,6 +1772,14 @@ pub enum ModerationCmd { } async fn run(cli: Cli) -> Result<(), CliError> { + if matches!(&cli.command, Cmd::PublicationFenceCapability) { + println!( + "{}", + buzz_publication_fence::PUBLICATION_FENCE_CAPABILITY_RESPONSE + ); + return Ok(()); + } + let relay_url = client::normalize_relay_url(&cli.relay); // Pack commands are local-only — no relay connection needed. @@ -1826,7 +1837,7 @@ async fn run(cli: Cli) -> Result<(), CliError> { Cmd::Upload(sub) => commands::upload::dispatch(sub, &client).await, Cmd::Mem(sub) => commands::mem::dispatch(sub, &client).await, Cmd::Moderation(sub) => commands::moderation::dispatch(sub, &client, &cli.format).await, - Cmd::Pack(_) => unreachable!("handled above"), + Cmd::Pack(_) | Cmd::PublicationFenceCapability => unreachable!("handled above"), } } @@ -1835,6 +1846,18 @@ mod tests { use super::*; use clap::CommandFactory; + #[tokio::test] + async fn publication_fence_capability_needs_no_identity_or_relay() { + assert_eq!( + run_from_args([ + "buzz", + buzz_publication_fence::PUBLICATION_FENCE_CAPABILITY_ARG, + ]) + .await, + 0 + ); + } + /// Smoke test: CLI definition is valid and parseable. #[test] fn cli_definition_is_valid() { @@ -1894,6 +1917,7 @@ mod tests { let cmd = Cli::command(); let mut actual: Vec = cmd .get_subcommands() + .filter(|s| !s.is_hide_set()) .map(|s| s.get_name().to_string()) .filter(|n| n != "help") .collect(); diff --git a/crates/buzz-publication-fence/Cargo.toml b/crates/buzz-publication-fence/Cargo.toml new file mode 100644 index 0000000000..d8a97bc7e4 --- /dev/null +++ b/crates/buzz-publication-fence/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "buzz-publication-fence" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Cross-process publication fence for managed Buzz agent turns" + +[dependencies] +fs2 = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +uuid = { workspace = true } + +[dev-dependencies] +tempfile = "3" diff --git a/crates/buzz-publication-fence/src/lib.rs b/crates/buzz-publication-fence/src/lib.rs new file mode 100644 index 0000000000..af8fd65b08 --- /dev/null +++ b/crates/buzz-publication-fence/src/lib.rs @@ -0,0 +1,503 @@ +#![deny(unsafe_code)] +#![warn(missing_docs)] +//! Cross-process publication fencing for managed Buzz agent turns. +//! +//! A harness owns a fence file for each ACP process. A publishing subprocess +//! captures the active generation when its command starts, then reacquires a +//! shared lease immediately before submitting the event. Terminal transitions +//! take an exclusive lock, so a completed transition rejects every later lease. + +use std::fs::{File, OpenOptions}; +use std::io::{Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use fs2::FileExt; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use uuid::Uuid; + +/// Environment variable containing the managed turn's publication fence path. +pub const PUBLICATION_FENCE_ENV: &str = "BUZZ_ACP_PUBLICATION_FENCE"; + +/// Hidden CLI argument used by the harness to verify fence-capable tooling. +pub const PUBLICATION_FENCE_CAPABILITY_ARG: &str = "__publication-fence-capability"; + +/// Exact response emitted by a fence-capable Buzz CLI. +pub const PUBLICATION_FENCE_CAPABILITY_RESPONSE: &str = "buzz-publication-fence-v1"; + +/// Errors returned by publication fence operations. +#[derive(Debug, Error)] +pub enum FenceError { + /// The fence file could not be opened, locked, read, or written. + #[error("publication fence I/O failed: {0}")] + Io(#[from] std::io::Error), + /// The fence file contained malformed state. + #[error("publication fence state is invalid: {0}")] + Json(#[from] serde_json::Error), + /// No active managed turn may publish. + #[error("managed turn is terminal; publication rejected")] + Terminal, + /// The command belongs to an earlier turn generation. + #[error("managed turn generation changed; stale publication rejected")] + StaleGeneration, + /// The publication destination is outside the active turn's scope. + #[error("publication destination does not match the active managed turn")] + ScopeMismatch, + /// The generation counter cannot be advanced safely. + #[error("publication fence generation exhausted")] + GenerationExhausted, + /// An exclusive transition could not drain active publication leases in time. + #[error("publication fence lease did not drain within {0:?}")] + LockTimeout(Duration), +} + +/// Channel and ordinary reply destination authorized for one managed turn. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct PublicationScope { + /// Harness turn identifier used for diagnostics. + pub turn_id: String, + /// Channel the turn may publish into. `None` allows any channel. + pub channel_id: Option, + /// Ordinary reply anchor. A different explicit reply target is rejected. + pub reply_to: Option, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +enum FenceStatus { + Active, + Terminal, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +struct FenceState { + version: u8, + generation: u64, + status: FenceStatus, + turn_id: String, + channel_id: Option, + reply_to: Option, +} + +/// Harness-owned writer for a single ACP process's publication fence. +#[derive(Clone, Debug)] +pub struct PublicationFence { + path: PathBuf, +} + +impl PublicationFence { + /// Create a terminal fence at `path`. + pub fn create(path: impl AsRef) -> Result { + let path = path.as_ref().to_path_buf(); + let fence = Self { path }; + let file = open_fence(&fence.path, true)?; + FileExt::lock_exclusive(&file)?; + write_state( + &file, + &FenceState { + version: 1, + generation: 0, + status: FenceStatus::Terminal, + turn_id: String::new(), + channel_id: None, + reply_to: None, + }, + )?; + FileExt::unlock(&file)?; + Ok(fence) + } + + /// Return the backing fence path for child-process environment injection. + pub fn path(&self) -> &Path { + &self.path + } + + /// Remove the fence file after the owning ACP process has been reaped. + pub fn remove(&self) -> Result<(), FenceError> { + match std::fs::remove_file(&self.path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } + } + + /// Open a new active generation and return its generation number. + pub fn begin(&self, scope: PublicationScope) -> Result { + let file = open_fence(&self.path, false)?; + FileExt::lock_exclusive(&file)?; + begin_locked(&file, scope) + } + + /// Open a new active generation after waiting at most `timeout` for leases. + pub fn begin_with_timeout( + &self, + scope: PublicationScope, + timeout: Duration, + ) -> Result { + let file = open_fence(&self.path, false)?; + lock_exclusive_with_timeout(&file, timeout)?; + begin_locked(&file, scope) + } + + /// Mark `generation` terminal. Returns false when a newer generation won. + pub fn terminate(&self, generation: u64) -> Result { + let file = open_fence(&self.path, false)?; + FileExt::lock_exclusive(&file)?; + terminate_locked(&file, generation) + } + + /// Mark `generation` terminal after waiting at most `timeout` for leases. + pub fn terminate_with_timeout( + &self, + generation: u64, + timeout: Duration, + ) -> Result { + let file = open_fence(&self.path, false)?; + lock_exclusive_with_timeout(&file, timeout)?; + terminate_locked(&file, generation) + } +} + +/// A publication attempt captured while a managed turn was active. +#[derive(Debug)] +pub struct PublicationAttempt { + path: PathBuf, + generation: u64, + channel_id: Uuid, + reply_to: Option, +} + +impl PublicationAttempt { + /// Capture from [`PUBLICATION_FENCE_ENV`] when managed fencing is enabled. + /// + /// Standalone and human-invoked CLI processes normally have no fence + /// variable and therefore return `Ok(None)` without changing behavior. + pub fn capture_from_env( + channel_id: Uuid, + reply_to: Option<&str>, + ) -> Result, FenceError> { + let Some(path) = std::env::var_os(PUBLICATION_FENCE_ENV) else { + return Ok(None); + }; + Self::capture(PathBuf::from(path), channel_id, reply_to).map(Some) + } + + /// Capture the active generation and destination scope at command start. + pub fn capture( + path: impl AsRef, + channel_id: Uuid, + reply_to: Option<&str>, + ) -> Result { + let path = path.as_ref().to_path_buf(); + let file = open_fence(&path, false)?; + FileExt::lock_shared(&file)?; + let state = read_state(&file)?; + validate_scope(&state, channel_id, reply_to)?; + FileExt::unlock(&file)?; + Ok(Self { + path, + generation: state.generation, + channel_id, + reply_to: reply_to.map(str::to_owned), + }) + } + + /// Acquire a shared publication lease immediately before network submit. + pub fn acquire(self) -> Result { + let file = open_fence(&self.path, false)?; + FileExt::lock_shared(&file)?; + let state = read_state(&file)?; + if state.generation != self.generation { + FileExt::unlock(&file)?; + return Err(FenceError::StaleGeneration); + } + if let Err(error) = validate_scope(&state, self.channel_id, self.reply_to.as_deref()) { + FileExt::unlock(&file)?; + return Err(error); + } + Ok(PublicationLease { file }) + } +} + +/// Shared lock held across the final network submission. +#[derive(Debug)] +pub struct PublicationLease { + file: File, +} + +impl Drop for PublicationLease { + fn drop(&mut self) { + let _ = FileExt::unlock(&self.file); + } +} + +fn begin_locked(file: &File, scope: PublicationScope) -> Result { + let current = read_state(file)?; + let generation = current + .generation + .checked_add(1) + .ok_or(FenceError::GenerationExhausted)?; + write_state( + file, + &FenceState { + version: 1, + generation, + status: FenceStatus::Active, + turn_id: scope.turn_id, + channel_id: scope.channel_id, + reply_to: scope.reply_to, + }, + )?; + FileExt::unlock(file)?; + Ok(generation) +} + +fn terminate_locked(file: &File, generation: u64) -> Result { + let mut state = read_state(file)?; + if state.generation != generation { + FileExt::unlock(file)?; + return Ok(false); + } + if state.status != FenceStatus::Terminal { + state.status = FenceStatus::Terminal; + write_state(file, &state)?; + } + FileExt::unlock(file)?; + Ok(true) +} + +fn lock_exclusive_with_timeout(file: &File, timeout: Duration) -> Result<(), FenceError> { + let deadline = Instant::now() + .checked_add(timeout) + .unwrap_or_else(Instant::now); + loop { + match FileExt::try_lock_exclusive(file) { + Ok(()) => return Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + let now = Instant::now(); + if now >= deadline { + return Err(FenceError::LockTimeout(timeout)); + } + std::thread::sleep( + deadline + .saturating_duration_since(now) + .min(Duration::from_millis(10)), + ); + } + Err(error) => return Err(error.into()), + } + } +} + +fn validate_scope( + state: &FenceState, + channel_id: Uuid, + reply_to: Option<&str>, +) -> Result<(), FenceError> { + if state.status != FenceStatus::Active { + return Err(FenceError::Terminal); + } + if state + .channel_id + .is_some_and(|expected| expected != channel_id) + { + return Err(FenceError::ScopeMismatch); + } + if let (Some(expected), Some(actual)) = (state.reply_to.as_deref(), reply_to) { + if expected != actual { + return Err(FenceError::ScopeMismatch); + } + } + Ok(()) +} + +fn open_fence(path: &Path, create: bool) -> Result { + let mut options = OpenOptions::new(); + options.read(true).write(true).create(create); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + options.open(path) +} + +fn read_state(file: &File) -> Result { + let mut reader = file.try_clone()?; + reader.seek(SeekFrom::Start(0))?; + Ok(serde_json::from_reader(reader)?) +} + +fn write_state(file: &File, state: &FenceState) -> Result<(), FenceError> { + let mut writer = file.try_clone()?; + writer.set_len(0)?; + writer.seek(SeekFrom::Start(0))?; + serde_json::to_writer(&mut writer, state)?; + writer.write_all(b"\n")?; + writer.flush()?; + writer.sync_data()?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scope(channel_id: Uuid, reply_to: &str, turn_id: &str) -> PublicationScope { + PublicationScope { + turn_id: turn_id.to_string(), + channel_id: Some(channel_id), + reply_to: Some(reply_to.to_string()), + } + } + + #[test] + fn terminal_transition_rejects_attempt_captured_while_active() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + let generation = fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin turn"); + let attempt = PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture active attempt"); + + assert!(fence.terminate(generation).expect("terminate turn")); + assert!(matches!(attempt.acquire(), Err(FenceError::Terminal))); + } + + #[test] + fn active_matching_generation_receives_publication_lease() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin turn"); + + let attempt = PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture active attempt"); + let _lease = attempt.acquire().expect("acquire active lease"); + } + + #[test] + fn newer_generation_rejects_attempt_from_prior_turn() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin first turn"); + let attempt = PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture first turn"); + + fence + .begin(scope(channel_id, "root-b", "turn-b")) + .expect("begin second turn"); + + assert!(matches!( + attempt.acquire(), + Err(FenceError::StaleGeneration) + )); + } + + #[test] + fn capture_rejects_wrong_channel_or_explicit_reply_target() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin turn"); + + assert!(matches!( + PublicationAttempt::capture(&path, Uuid::new_v4(), Some("root-a")), + Err(FenceError::ScopeMismatch) + )); + assert!(matches!( + PublicationAttempt::capture(&path, channel_id, Some("root-b")), + Err(FenceError::ScopeMismatch) + )); + } + + #[test] + fn terminal_transition_waits_for_in_flight_publication_lease() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + let generation = fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin turn"); + let lease = PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture active attempt") + .acquire() + .expect("acquire publication lease"); + let (tx, rx) = std::sync::mpsc::channel(); + let closer = fence.clone(); + let thread = std::thread::spawn(move || { + tx.send(closer.terminate(generation)) + .expect("send close result"); + }); + + std::thread::sleep(std::time::Duration::from_millis(50)); + assert!(matches!( + rx.try_recv(), + Err(std::sync::mpsc::TryRecvError::Empty) + )); + drop(lease); + assert!(rx + .recv_timeout(std::time::Duration::from_secs(1)) + .expect("terminal transition completes") + .expect("terminal transition succeeds")); + thread.join().expect("join closer"); + } + + #[test] + fn bounded_terminal_transition_times_out_without_blocking_forever() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + let generation = fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin turn"); + let _lease = PublicationAttempt::capture(&path, channel_id, Some("root-a")) + .expect("capture active attempt") + .acquire() + .expect("acquire publication lease"); + let timeout = Duration::from_millis(30); + let started = Instant::now(); + + assert!(matches!( + fence.terminate_with_timeout(generation, timeout), + Err(FenceError::LockTimeout(actual)) if actual == timeout + )); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[test] + fn stale_termination_cannot_close_new_generation() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("publication-fence.json"); + let fence = PublicationFence::create(&path).expect("create fence"); + let channel_id = Uuid::new_v4(); + let old_generation = fence + .begin(scope(channel_id, "root-a", "turn-a")) + .expect("begin first turn"); + fence + .begin(scope(channel_id, "root-b", "turn-b")) + .expect("begin second turn"); + + assert!(!fence + .terminate(old_generation) + .expect("ignore stale termination")); + PublicationAttempt::capture(&path, channel_id, Some("root-b")) + .expect("new turn remains active"); + } +} diff --git a/crates/sprig/src/main.rs b/crates/sprig/src/main.rs index 672a5a5f37..0e2651c80b 100644 --- a/crates/sprig/src/main.rs +++ b/crates/sprig/src/main.rs @@ -46,8 +46,8 @@ fn print_usage() { println!( "Sprig — all-in-one Buzz ACP harness, agent, and developer MCP\n\n\ Sprig is a multicall binary. Invoke it through one of the personality names:\n\n\ - buzz-acp ACP harness\n buzz-agent ACP-compliant agent\n buzz-dev-mcp Developer MCP server\n\n\ -Developer MCP helper names are also supported: rg, tree, buzz, git-credential-nostr, git-sign-nostr.\n\n\ -Installers can create links with:\n ln -s sprig buzz-acp\n ln -s sprig buzz-agent\n ln -s sprig buzz-dev-mcp" + buzz-acp ACP harness\n buzz-agent ACP-compliant agent\n buzz-dev-mcp Developer MCP server\n buzz Fence-capable Buzz CLI\n\n\ +Developer MCP helper names are also supported: rg, tree, git-credential-nostr, git-sign-nostr.\n\n\ +Installers can create links with:\n ln -s sprig buzz-acp\n ln -s sprig buzz-agent\n ln -s sprig buzz-dev-mcp\n ln -s sprig buzz" ); } diff --git a/scripts/build-sprig.sh b/scripts/build-sprig.sh index 77ee6e5832..24aebac8c8 100755 --- a/scripts/build-sprig.sh +++ b/scripts/build-sprig.sh @@ -5,8 +5,8 @@ # sprig implementation binary # buzz-acp link to sprig (ACP harness) # buzz-agent link to sprig (ACP-compliant agent) -# buzz-dev-mcp link to sprig (developer MCP server; also dispatches -# rg/tree/buzz/git-credential-nostr/git-sign-nostr) +# buzz-dev-mcp link to sprig (developer MCP server) +# buzz link to sprig (fence-capable CLI used by managed turns) # # Usage: # ./scripts/build-sprig.sh [version] [target] @@ -25,6 +25,7 @@ # across builds (e.g. `sprig-`). Defaults to # `sprig--`. # DIST_DIR output directory (default: dist) +# SPRIG_BIN prebuilt source binary override (package tests only) # # Output: # ${DIST_DIR}/${ARCHIVE_BASENAME}.tar.gz @@ -35,6 +36,7 @@ # buzz-acp # buzz-agent # buzz-dev-mcp +# buzz # README.md # sprig.json { version, git_sha, target, binaries: [{name, sha256, size}] } @@ -59,7 +61,7 @@ else fi BUNDLE_BIN="sprig" -COMMANDS=(buzz-acp buzz-agent buzz-dev-mcp) +COMMANDS=(buzz-acp buzz-agent buzz-dev-mcp buzz) echo "==> Building Sprig v${VERSION} for ${TARGET}" echo " git_sha=${GIT_SHA}" @@ -85,8 +87,9 @@ else "${BUILDER[@]}" -p "$BUNDLE_BIN" fi -if [[ ! -f "${BIN_DIR}/${BUNDLE_BIN}" ]]; then - echo "error: ${BIN_DIR}/${BUNDLE_BIN} not found after build" >&2 +SOURCE_BIN="${SPRIG_BIN:-${BIN_DIR}/${BUNDLE_BIN}}" +if [[ ! -f "${SOURCE_BIN}" ]]; then + echo "error: ${SOURCE_BIN} not found after build" >&2 exit 1 fi @@ -102,7 +105,7 @@ sha256_of() { fi } -cp "${BIN_DIR}/${BUNDLE_BIN}" "${STAGING}/${BUNDLE_BIN}" +cp "${SOURCE_BIN}" "${STAGING}/${BUNDLE_BIN}" chmod 0755 "${STAGING}/${BUNDLE_BIN}" if command -v strip >/dev/null 2>&1; then strip "${STAGING}/${BUNDLE_BIN}" 2>/dev/null || true @@ -143,9 +146,9 @@ Commands: - `buzz-acp` — ACP harness that bridges Buzz channel events to an ACP-compliant agent over stdio. - `buzz-agent` — ACP-compliant agent (spawns MCP servers, calls LLMs). -- `buzz-dev-mcp` — Developer MCP server (shell, str_replace, todo) and - multicall entrypoint for `rg`, `tree`, `buzz`, `git-credential-nostr`, - `git-sign-nostr`. +- `buzz-dev-mcp` — Developer MCP server (shell, str_replace, todo). +- `buzz` — co-versioned CLI used by managed turns. The harness verifies its + publication-fence capability before starting an agent pool. See `sprig.json` for SHA-256s, sizes, target, and source git SHA. diff --git a/scripts/test-build-sprig-package.sh b/scripts/test-build-sprig-package.sh new file mode 100755 index 0000000000..db4469a927 --- /dev/null +++ b/scripts/test-build-sprig-package.sh @@ -0,0 +1,38 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TMP="$(mktemp -d)" +trap 'rm -rf "${TMP}"' EXIT + +FAKE="${TMP}/sprig" +cat > "${FAKE}" <<'SH' +#!/usr/bin/env bash +name="$(basename "$0")" +if [[ "$name" == "buzz" && "${1:-}" == "__publication-fence-capability" ]]; then + printf '%s\n' 'buzz-publication-fence-v1' + exit 0 +fi +if [[ "$name" == "sprig" && "${1:-}" == "--version" ]]; then + printf '%s\n' 'sprig package-test' + exit 0 +fi +exit 64 +SH +chmod 0755 "${FAKE}" + +cd "${ROOT}" +SKIP_BUILD=1 \ +SPRIG_BIN="${FAKE}" \ +DIST_DIR="${TMP}/dist" \ +ARCHIVE_BASENAME="sprig-package-test" \ +./scripts/build-sprig.sh 0.1.0-test >/dev/null + +mkdir "${TMP}/unpacked" +tar -xzf "${TMP}/dist/sprig-package-test.tar.gz" -C "${TMP}/unpacked" +test -L "${TMP}/unpacked/buzz" +test "$(readlink "${TMP}/unpacked/buzz")" = "sprig" +test "$("${TMP}/unpacked/buzz" __publication-fence-capability)" = "buzz-publication-fence-v1" +grep -q '"name":"buzz"' "${TMP}/unpacked/sprig.json" + +echo "sprig package capability: PASS" diff --git a/scripts/test-tauri-sidecar-set.sh b/scripts/test-tauri-sidecar-set.sh new file mode 100755 index 0000000000..dde577751b --- /dev/null +++ b/scripts/test-tauri-sidecar-set.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TMP="$(mktemp -d)" +trap 'rm -rf "${TMP}"' EXIT +TARGET="aarch64-apple-darwin" + +python3 - "${ROOT}/desktop/src-tauri/tauri.conf.json" <<'PY' +import json, sys +with open(sys.argv[1], encoding="utf-8") as handle: + external = json.load(handle)["bundle"]["externalBin"] +assert "binaries/buzz-acp" in external +assert "binaries/buzz" in external +PY + +grep -q 'for bin in buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz' "${ROOT}/Justfile" +grep -q 'for bin in buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz' "${ROOT}/.github/workflows/ci.yml" + +mkdir -p "${TMP}/bin" "${TMP}/target/${TARGET}/release" +cat > "${TMP}/bin/rustc" <<'SH' +#!/usr/bin/env bash +printf '%s\n' 'host: aarch64-apple-darwin' +SH +chmod 0755 "${TMP}/bin/rustc" +for bin in buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr; do + printf '#!/usr/bin/env bash\nexit 0\n' > "${TMP}/target/${TARGET}/release/${bin}" + chmod 0755 "${TMP}/target/${TARGET}/release/${bin}" +done +cat > "${TMP}/target/${TARGET}/release/buzz" <<'SH' +#!/usr/bin/env bash +if [[ "${1:-}" == "__publication-fence-capability" ]]; then + printf '%s\n' 'buzz-publication-fence-v1' + exit 0 +fi +exit 64 +SH +chmod 0755 "${TMP}/target/${TARGET}/release/buzz" + +cd "${TMP}" +PATH="${TMP}/bin:${PATH}" "${ROOT}/scripts/bundle-sidecars.sh" "${TARGET}" >/dev/null +ACP="desktop/src-tauri/binaries/buzz-acp-${TARGET}" +CLI="desktop/src-tauri/binaries/buzz-${TARGET}" +test -x "${ACP}" +test -x "${CLI}" +test "$("${CLI}" __publication-fence-capability)" = "buzz-publication-fence-v1" + +echo "tauri sidecar capability set: PASS"