From bfd100939f46eb1d6a0b3b4644a80b4c5280861b Mon Sep 17 00:00:00 2001 From: joelbrilliant Date: Sat, 25 Jul 2026 09:15:22 +1000 Subject: [PATCH 1/3] feat(acp): durable ACP session store and pool Channel to ACP session bindings are in-memory only, so every harness restart starts a cold conversation even against an agent that advertises `loadSession`. This persists the mapping as a small JSON sidecar under the process data directory, keyed by (agent command identity, agent args, channel id) so different agent binaries and profiles never share bindings. Heartbeats stay ephemeral and are never stored. The store is shared between processes, so a process-local cache alone is unsafe: two buzz-acp processes with the same agent identity would clobber each other. Every read and mutation takes a sibling lockfile, reloads the on-disk map under that lock, then writes atomically. Removal is conditional on purpose. After a failed `session/load` another process may already have written a newer session for the same channel, so `remove_if_equals` only drops the binding it actually failed to load. A key-only remove would delete the fresher one. Signed-off-by: joelbrilliant --- Cargo.lock | 15 +- crates/buzz-acp/Cargo.toml | 6 + crates/buzz-acp/src/acp.rs | 41 +++ crates/buzz-acp/src/lib.rs | 27 +- crates/buzz-acp/src/pool.rs | 119 ++++++++ crates/buzz-acp/src/session_store.rs | 442 +++++++++++++++++++++++++++ 6 files changed, 643 insertions(+), 7 deletions(-) create mode 100644 crates/buzz-acp/src/session_store.rs diff --git a/Cargo.lock b/Cargo.lock index 9d0190868de..d8816ff17f3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -771,7 +771,9 @@ dependencies = [ "buzz-sdk", "chrono", "clap", + "dirs", "evalexpr", + "fs2", "futures-util", "hex", "httparse", @@ -782,6 +784,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", @@ -2172,7 +2175,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.117", ] [[package]] @@ -2853,6 +2856,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/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..d573f3e5224 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -68,6 +68,11 @@ clap = { version = "4", features = ["derive", "env"] } # Config file toml = "1.0" +# Durable ACP session binding store location +dirs = "6" +# Cross-process flock for shared session bindings +fs2 = "0.4" + # Filter expressions evalexpr = { workspace = true } @@ -78,4 +83,5 @@ nix = { version = "0.31", default-features = false, features = ["signal"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +tempfile = "3" httparse = "1" diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 78db7ff718b..59db3c9f4bf 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -601,6 +601,47 @@ impl AcpClient { .session_id) } + /// Send `session/load` for an existing ACP session id. + /// + /// Used after harness restart when a durable channel→session binding is + /// known and the agent advertised `agentCapabilities.loadSession`. + /// History-replay `session/update` notifications are consumed by the + /// request loop and logged only — they are not re-published to Buzz. + pub async fn session_load_full( + &mut self, + cwd: &str, + session_id: &str, + mcp_servers: Vec, + ) -> Result { + let params = serde_json::json!({ + "cwd": cwd, + "sessionId": session_id, + "mcpServers": mcp_servers, + }); + let result = self.send_request("session/load", params).await?; + // Spec-compliant agents may omit sessionId on load (it is implied). + // Prefer the request id so callers always have a concrete binding. + let resolved_id = result + .get("sessionId") + .and_then(|v| v.as_str()) + .unwrap_or(session_id) + .to_owned(); + tracing::info!(target: "acp::session", "session loaded: {resolved_id}"); + Ok(SessionNewResponse { + session_id: resolved_id, + raw: result, + }) + } + + /// Returns true when an initialize result advertises `loadSession`. + pub fn agent_supports_load_session(init_result: &serde_json::Value) -> bool { + init_result + .get("agentCapabilities") + .and_then(|caps| caps.get("loadSession")) + .and_then(|v| v.as_bool()) + .unwrap_or(false) + } + /// Send Goose's custom system-prompt request after `session/new`. pub async fn session_set_goose_system_prompt( &mut self, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 0230ea0875f..16ddc7dc299 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -9,6 +9,7 @@ mod pool; mod pool_lifecycle; mod queue; mod relay; +mod session_store; mod setup_mode; mod usage; @@ -1143,7 +1144,7 @@ struct RespawnResult { /// Tuple: (initialized client, protocol version, supports_goose_steer). /// The third element is always `true` — the supervisor uses /// try-and-tolerate for the steer extension. - result: Result<(AcpClient, u32, String)>, + result: Result<(AcpClient, u32, String, bool)>, } /// Outcome of a non-cancelling steer attempt, forwarded from a per-attempt @@ -1187,7 +1188,7 @@ impl RespawnGuard { /// Send the result and disarm the guard. Uses `try_send` (sync) so there /// is no await boundary between marking `sent` and actually enqueueing — /// cancellation cannot slip between the two. - fn send(mut self, result: Result<(AcpClient, u32, String)>) { + fn send(mut self, result: Result<(AcpClient, u32, String, bool)>) { // Invariant: try_send succeeds because the channel capacity equals the // slot count, and respawn_in_flight guarantees at most one outstanding // result per slot. If this ever fails, the channel sizing or the @@ -1560,6 +1561,14 @@ async fn tokio_main() -> Result<()> { memory_enabled: config.memory_enabled, harness_name: crate::config::normalize_agent_command_identity(&config.agent_command), relay_url: config.relay_url.clone(), + agent_command: config.agent_command.clone(), + agent_args: config.agent_args.clone(), + session_store: std::sync::Arc::new(crate::session_store::SessionStore::open( + crate::session_store::SessionStore::default_path( + &config.agent_command, + &config.agent_args, + ), + )), }); if !config.memory_enabled { @@ -1785,7 +1794,7 @@ async fn tokio_main() -> Result<()> { while let Ok(rr) = respawn_rx.try_recv() { crash_history[rr.index].respawn_in_flight = false; match rr.result { - Ok((acp, protocol_version, agent_name)) => { + Ok((acp, protocol_version, agent_name, supports_load_session)) => { let agent = OwnedAgent { index: rr.index, acp, @@ -1796,6 +1805,7 @@ async fn tokio_main() -> Result<()> { agent_name, goose_system_prompt_supported: None, protocol_version, + supports_load_session, }; pool.return_agent(agent); tracing::info!(agent = rr.index, "respawn complete"); @@ -2665,7 +2675,7 @@ async fn tokio_main() -> Result<()> { // Drain any respawn results that completed before the abort. Explicitly // shut down returned agents instead of relying on AcpClient::Drop. while let Ok(rr) = respawn_rx.try_recv() { - if let Ok((mut acp, _, _)) = rr.result { + if let Ok((mut acp, _, _, _)) = rr.result { acp.shutdown().await; tracing::debug!(agent = rr.index, "reaped respawned agent on shutdown"); } @@ -3792,6 +3802,8 @@ async fn initialize_agent_pool( }), ); let agent_name = normalized_agent_name(&init_result); + let supports_load_session = + AcpClient::agent_supports_load_session(&init_result); agent_slots.push(Some(OwnedAgent { index: i, acp, @@ -3802,6 +3814,7 @@ async fn initialize_agent_pool( agent_name, goose_system_prompt_supported: None, protocol_version, + supports_load_session, })); } Ok(Err(e)) => { @@ -3852,7 +3865,7 @@ async fn spawn_and_init( has_generated_codex_config: bool, agent_index: usize, observer: Option, -) -> Result<(AcpClient, u32, String)> { +) -> Result<(AcpClient, u32, String, bool)> { let mut acp = AcpClient::spawn(command, args, extra_env, has_generated_codex_config) .await .map_err(|e| anyhow::anyhow!("failed to spawn agent: {e}"))?; @@ -3862,6 +3875,7 @@ async fn spawn_and_init( Ok(init_result) => { tracing::info!("agent initialized: {init_result}"); let protocol_version = init_result["protocolVersion"].as_u64().unwrap_or(1) as u32; + let supports_load_session = AcpClient::agent_supports_load_session(&init_result); acp.observe( "agent_initialized", serde_json::json!({ @@ -3870,7 +3884,7 @@ async fn spawn_and_init( }), ); let agent_name = normalized_agent_name(&init_result); - Ok((acp, protocol_version, agent_name)) + Ok((acp, protocol_version, agent_name, supports_load_session)) } Err(e) => { // Explicitly shut down the spawned child to prevent zombie/leak. @@ -5187,6 +5201,7 @@ mod error_outcome_emission_tests { // Error branches under test never read this; 1 is the legacy // non-systemPrompt path, the simplest valid value. protocol_version: 1, + supports_load_session: false, } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index cc537f86830..451724eb95a 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -168,6 +168,8 @@ pub struct OwnedAgent { pub goose_system_prompt_supported: Option, /// Protocol version reported by the agent in its initialize response. pub protocol_version: u32, + /// Whether the agent advertised `agentCapabilities.loadSession` at init. + pub supports_load_session: bool, } fn has_system_prompt_support( @@ -529,6 +531,12 @@ pub struct PromptContext { /// the desktop keys per (agent, relay) pair, e.g. `session_config_captured`, /// mirroring the `managed_agent_runtime_lifecycle` frames. pub relay_url: String, + /// Agent binary as configured (for durable session binding identity). + pub agent_command: String, + /// Agent args as configured (for durable session binding identity). + pub agent_args: Vec, + /// Durable channel→session bindings surviving harness restarts. + pub session_store: std::sync::Arc, } impl AgentPool { @@ -795,6 +803,87 @@ const CONTROL_CANCEL_GRACE: Duration = Duration::from_secs(5); /// Timeout for permission-mode requests (`session/set_config_option` with `configId: "mode"`). const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5); +/// Try to restore a durable channel session via `session/load`. +/// +/// Returns `Some(session_id)` on success. On miss, capability absence, or load +/// failure, clears the stale binding (when present) and returns `None` so the +/// caller can fall through to `session/new`. +async fn try_load_persisted_session( + agent: &mut OwnedAgent, + ctx: &PromptContext, + channel_id: &Uuid, + _agent_core: Option<&str>, + _agent_canvas: Option<&str>, +) -> Option { + if !agent.supports_load_session { + return None; + } + let stored = ctx + .session_store + .get(&ctx.agent_command, &ctx.agent_args, channel_id)?; + match agent + .acp + .session_load_full(&ctx.cwd, &stored, ctx.mcp_servers.clone()) + .await + { + Ok(resp) => { + if agent.model_capabilities.is_none() { + agent.model_capabilities = Some(AgentModelCapabilities { + config_options_raw: extract_model_config_options(&resp.raw), + available_models_raw: extract_model_state(&resp.raw), + }); + } + // Re-apply desired model after load when present. + if let Some(ref desired) = agent.desired_model { + if let Some(method) = resolve_model_switch_method(&resp.raw, desired) { + if let Err(e) = + apply_model_switch(&mut agent.acp, &resp.session_id, desired, &method).await + { + tracing::warn!( + target: "pool::session", + error = %e, + "model re-apply after session/load failed — continuing with loaded session" + ); + } + } + } + if !ctx.permission_mode.is_default() + && agent_supports_mode(&resp.raw, ctx.permission_mode.as_wire_str()) + { + if let Err(e) = + apply_permission_mode(&mut agent.acp, &resp.session_id, &ctx.permission_mode) + .await + { + tracing::warn!( + target: "pool::session", + error = %e, + "permission mode after session/load failed — continuing" + ); + } + } + Some(resp.session_id) + } + Err(e) => { + tracing::warn!( + target: "pool::session", + session_id = %stored, + channel_id = %channel_id, + error = %e, + "session/load failed — clearing stale binding (if unchanged) and creating a new session" + ); + // Only drop the binding we failed to load. A concurrent process may + // already have written a newer session for this channel. + let _ = ctx.session_store.remove_if_equals( + &ctx.agent_command, + &ctx.agent_args, + channel_id, + &stored, + ); + None + } + } +} + /// Create a new ACP session via `session_new_full()`, populate model capabilities /// on the agent (first session only), and apply `desired_model` if set. /// @@ -1469,6 +1558,24 @@ pub async fn run_prompt_task( PromptSource::Channel(cid) => { if let Some(sid) = agent.state.sessions.get(cid) { (sid.clone(), false) + } else if let Some(sid) = try_load_persisted_session( + &mut agent, + &ctx, + cid, + agent_core.as_deref(), + agent_canvas.as_deref(), + ) + .await + { + tracing::info!( + target: "pool::session", + "loaded session {sid} for channel {cid}" + ); + agent.state.sessions.insert(*cid, sid.clone()); + if let Some((pending_cid, section)) = pending_canvas.take() { + agent.state.canvas_sections.insert(pending_cid, section); + } + (sid, false) } else { // Create new session with model application. match create_session_and_apply_model( @@ -1485,6 +1592,8 @@ pub async fn run_prompt_task( "created session {sid} for channel {cid}" ); agent.state.sessions.insert(*cid, sid.clone()); + ctx.session_store + .put(&ctx.agent_command, &ctx.agent_args, cid, &sid); // Commit canvas only after session creation succeeds (I3). if let Some((pending_cid, section)) = pending_canvas.take() { agent.state.canvas_sections.insert(pending_cid, section); @@ -4998,6 +5107,7 @@ mod tests { agent_name: "unknown".into(), goose_system_prompt_supported: None, protocol_version: 2, + supports_load_session: false, }; // Simulate dispatch: install a steer receiver (normally done by @@ -5056,6 +5166,7 @@ mod tests { agent_name: "unknown".into(), goose_system_prompt_supported: None, protocol_version: 2, + supports_load_session: false, }; // Simulate a completed turn: `steer_rx` was consumed by the read loop @@ -5307,6 +5418,14 @@ mod tests { memory_enabled: false, harness_name: "goose".to_string(), relay_url: "ws://127.0.0.1:3000".to_string(), + agent_command: "goose".to_string(), + agent_args: vec!["acp".to_string()], + session_store: std::sync::Arc::new(crate::session_store::SessionStore::open( + std::env::temp_dir().join(format!( + "buzz-acp-test-sessions-{}.json", + uuid::Uuid::new_v4() + )), + )), } } diff --git a/crates/buzz-acp/src/session_store.rs b/crates/buzz-acp/src/session_store.rs new file mode 100644 index 00000000000..1617213a894 --- /dev/null +++ b/crates/buzz-acp/src/session_store.rs @@ -0,0 +1,442 @@ +//! Durable channel → ACP session bindings for harness restarts. +//! +//! `SessionState` is in-memory only. Agents that advertise `loadSession` (e.g. +//! Hermes) can restore a prior ACP conversation after the harness respawns if +//! the channel→session mapping survives. This module persists that mapping as +//! a small JSON sidecar under the process data directory. +//! +//! Keyed by `(agent_command_identity, agent_args, channel_id)` so different +//! agent binaries / profiles do not share bindings. Heartbeats are never +//! stored — they stay ephemeral. +//! +//! Cross-process safety: the store is a shared file. Every read and mutation +//! takes a sibling lockfile, reloads the on-disk map under that lock, then +//! writes atomically. A process-local cache alone is unsafe when two +//! `buzz-acp` processes share the same agent command/args identity. + +use std::collections::HashMap; +use std::ffi::OsString; +use std::fs::{self, File, OpenOptions}; +use std::path::{Path, PathBuf}; + +use fs2::FileExt; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::config::normalize_agent_command_identity; + +/// Environment override for the session store path (tests / operators). +pub const SESSION_STORE_ENV: &str = "BUZZ_ACP_SESSION_STORE"; + +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)] +struct StoreFile { + /// version for future migrations + version: u32, + /// map key → ACP session id + sessions: HashMap, +} + +/// Durable session binding store shared across buzz-acp processes. +pub struct SessionStore { + path: PathBuf, + lock_path: PathBuf, +} + +/// RAII wrapper that unlocks the OS file lock on drop. +struct StoreLock { + file: File, +} + +impl Drop for StoreLock { + fn drop(&mut self) { + let _ = FileExt::unlock(&self.file); + } +} + +impl SessionStore { + /// Open or create the store at the resolved path. + /// + /// Does not cache file contents; each operation reloads under lock. + pub fn open(path: PathBuf) -> Self { + let lock_path = sibling_lock_path(&path); + Self { path, lock_path } + } + + /// Resolve the default store path for this agent identity. + pub fn default_path(agent_command: &str, agent_args: &[String]) -> PathBuf { + if let Ok(override_path) = std::env::var(SESSION_STORE_ENV) { + if !override_path.trim().is_empty() { + return PathBuf::from(override_path); + } + } + let identity = store_identity(agent_command, agent_args); + let base = dirs::data_local_dir() + .or_else(dirs::data_dir) + .unwrap_or_else(|| PathBuf::from(".")); + base.join("buzz-acp") + .join("sessions") + .join(format!("{identity}.json")) + } + + /// Look up a stored ACP session id for a channel. + pub fn get( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + ) -> Option { + let key = binding_key(agent_command, agent_args, channel_id); + let _lock = self.acquire_lock(false)?; + match load_store(&self.path) { + Ok(data) => data.sessions.get(&key).cloned(), + Err(e) => { + self.warn_io("failed to read ACP session bindings", &e); + None + } + } + } + + /// Persist a channel → session binding. + pub fn put( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + session_id: &str, + ) { + let key = binding_key(agent_command, agent_args, channel_id); + let Some(_lock) = self.acquire_lock(true) else { + return; + }; + let mut data = match load_store(&self.path) { + Ok(data) => data, + Err(e) if e.kind() == std::io::ErrorKind::InvalidData => { + // Corrupt sidecar: log and recover empty rather than wedging puts forever. + self.warn_io( + "corrupt ACP session store on update — rewriting from empty map", + &e, + ); + StoreFile::default() + } + Err(e) => { + self.warn_io("failed to read ACP session bindings before update", &e); + return; + } + }; + data.version = 1; + data.sessions.insert(key, session_id.to_owned()); + if let Err(e) = save_store(&self.path, &data) { + self.warn_io("failed to persist ACP session binding", &e); + } + } + + /// Remove a binding only if it still points at `expected_session_id`. + /// + /// Used after a failed `session/load`: another process may have already + /// written a newer session for the same channel, and a key-only remove + /// would delete that fresher binding. + /// + /// Returns `true` when a matching binding was removed. + pub fn remove_if_equals( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + expected_session_id: &str, + ) -> bool { + let key = binding_key(agent_command, agent_args, channel_id); + let Some(_lock) = self.acquire_lock(true) else { + return false; + }; + match load_store(&self.path) { + Ok(mut data) => { + let matches = data + .sessions + .get(&key) + .is_some_and(|current| current == expected_session_id); + if !matches { + return false; + } + data.sessions.remove(&key); + if let Err(e) = save_store(&self.path, &data) { + self.warn_io( + "failed to persist conditional ACP session binding removal", + &e, + ); + return false; + } + true + } + Err(e) => { + self.warn_io( + "failed to read ACP session bindings before conditional removal", + &e, + ); + false + } + } + } + + fn acquire_lock(&self, exclusive: bool) -> Option { + if let Some(parent) = self.lock_path.parent() { + if let Err(e) = fs::create_dir_all(parent) { + self.warn_io("failed to create ACP session store directory", &e); + return None; + } + } + let file = match OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&self.lock_path) + { + Ok(file) => file, + Err(e) => { + self.warn_io("failed to open ACP session store lock", &e); + return None; + } + }; + let result = if exclusive { + FileExt::lock_exclusive(&file) + } else { + FileExt::lock_shared(&file) + }; + if let Err(e) = result { + self.warn_io("failed to lock ACP session store", &e); + return None; + } + Some(StoreLock { file }) + } + + fn warn_io(&self, message: &'static str, error: &std::io::Error) { + tracing::warn!( + target: "session_store", + path = %self.path.display(), + lock_path = %self.lock_path.display(), + error = %error, + "{message}" + ); + } +} + +fn sibling_lock_path(path: &Path) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(OsString::from(".lock")); + PathBuf::from(name) +} + +fn store_identity(agent_command: &str, agent_args: &[String]) -> String { + let cmd = normalize_agent_command_identity(agent_command); + let args = agent_args.join(" "); + let raw = if args.is_empty() { + cmd + } else { + format!("{cmd} {args}") + }; + // Keep the filename filesystem-safe and short. + let mut out = String::with_capacity(raw.len()); + for ch in raw.chars() { + if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' { + out.push(ch); + } else { + out.push('_'); + } + } + if out.is_empty() { + "agent".into() + } else { + out + } +} + +fn binding_key(agent_command: &str, agent_args: &[String], channel_id: &Uuid) -> String { + format!( + "{}|{}|{}", + normalize_agent_command_identity(agent_command), + agent_args.join("\u{1f}"), + channel_id + ) +} + +fn load_store(path: &Path) -> std::io::Result { + match fs::read_to_string(path) { + Ok(text) => serde_json::from_str(&text) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(StoreFile::default()), + Err(e) => Err(e), + } +} + +fn save_store(path: &Path, data: &StoreFile) -> std::io::Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let tmp = path.with_extension("json.tmp"); + let json = serde_json::to_string_pretty(data) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + fs::write(&tmp, json)?; + fs::rename(&tmp, path)?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + #[test] + fn round_trip_binding() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let store = SessionStore::open(path); + let channel = Uuid::new_v4(); + assert!(store.get("hermes", &["acp".into()], &channel).is_none()); + store.put("hermes", &["acp".into()], &channel, "sess-1"); + assert_eq!( + store.get("hermes", &["acp".into()], &channel).as_deref(), + Some("sess-1") + ); + // Re-open from disk. + let store2 = SessionStore::open(store.path.clone()); + assert_eq!( + store2.get("hermes", &["acp".into()], &channel).as_deref(), + Some("sess-1") + ); + assert!(store2.remove_if_equals("hermes", &["acp".into()], &channel, "sess-1")); + assert!(store2.get("hermes", &["acp".into()], &channel).is_none()); + } + + #[test] + fn different_args_are_isolated() { + let dir = tempdir().unwrap(); + let store = SessionStore::open(dir.path().join("s.json")); + let channel = Uuid::new_v4(); + store.put("hermes", &["acp".into()], &channel, "a"); + store.put( + "hermes", + &["-p".into(), "chad".into(), "acp".into()], + &channel, + "b", + ); + assert_eq!( + store.get("hermes", &["acp".into()], &channel).as_deref(), + Some("a") + ); + assert_eq!( + store + .get( + "hermes", + &["-p".into(), "chad".into(), "acp".into()], + &channel + ) + .as_deref(), + Some("b") + ); + } + + #[test] + fn independently_opened_stores_do_not_lose_updates() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let store_a = SessionStore::open(path.clone()); + let store_b = SessionStore::open(path.clone()); + let channel_a = Uuid::new_v4(); + let channel_b = Uuid::new_v4(); + let channel_c = Uuid::new_v4(); + let args = ["acp".into()]; + + store_a.put("hermes", &args, &channel_a, "session-a"); + store_b.put("hermes", &args, &channel_b, "session-b"); + + let reopened = SessionStore::open(path.clone()); + assert_eq!( + reopened.get("hermes", &args, &channel_a).as_deref(), + Some("session-a") + ); + assert_eq!( + reopened.get("hermes", &args, &channel_b).as_deref(), + Some("session-b") + ); + + // Open both before either mutation. A stale process-local snapshot would + // resurrect channel A when the second store writes channel C. + let remover = SessionStore::open(path.clone()); + let writer = SessionStore::open(path.clone()); + assert!(remover.remove_if_equals("hermes", &args, &channel_a, "session-a")); + writer.put("hermes", &args, &channel_c, "session-c"); + + let final_store = SessionStore::open(path); + assert!(final_store.get("hermes", &args, &channel_a).is_none()); + assert_eq!( + final_store.get("hermes", &args, &channel_b).as_deref(), + Some("session-b") + ); + assert_eq!( + final_store.get("hermes", &args, &channel_c).as_deref(), + Some("session-c") + ); + } + + #[test] + fn put_recovers_from_corrupt_store() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + fs::write(&path, "{not-json").unwrap(); + let store = SessionStore::open(path.clone()); + let channel = Uuid::new_v4(); + store.put("hermes", &["acp".into()], &channel, "recovered"); + assert_eq!( + store.get("hermes", &["acp".into()], &channel).as_deref(), + Some("recovered") + ); + } + + #[test] + fn remove_if_equals_does_not_delete_newer_binding() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let args = ["acp".into()]; + let channel = Uuid::new_v4(); + + // Process A reads X. + let process_a = SessionStore::open(path.clone()); + process_a.put("hermes", &args, &channel, "session-x"); + let read_x = process_a + .get("hermes", &args, &channel) + .expect("process A read X"); + assert_eq!(read_x, "session-x"); + + // Process B writes Y for the same channel. + let process_b = SessionStore::open(path.clone()); + process_b.put("hermes", &args, &channel, "session-y"); + assert_eq!( + process_b.get("hermes", &args, &channel).as_deref(), + Some("session-y") + ); + + // Process A's failed load of X must not delete Y. + let removed = process_a.remove_if_equals("hermes", &args, &channel, &read_x); + assert!(!removed); + + let final_store = SessionStore::open(path); + assert_eq!( + final_store.get("hermes", &args, &channel).as_deref(), + Some("session-y") + ); + } + + #[test] + fn remove_if_equals_clears_matching_stale_binding() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let args = ["acp".into()]; + let channel = Uuid::new_v4(); + let store = SessionStore::open(path.clone()); + store.put("hermes", &args, &channel, "session-x"); + assert!(store.remove_if_equals("hermes", &args, &channel, "session-x")); + assert!(store.get("hermes", &args, &channel).is_none()); + // No-op when already gone. + assert!(!store.remove_if_equals("hermes", &args, &channel, "session-x")); + } +} From 7b76647e7dff22b59896f56e22b6ec1eba85a5bf Mon Sep 17 00:00:00 2001 From: joelbrilliant Date: Mon, 27 Jul 2026 13:24:00 +1000 Subject: [PATCH 2/3] fix(acp): don't clear a session binding on an indeterminate load failure `session/load` failure handling treated every error the same: drop the stored binding and fall through to `session/new`. Only a JSON-RPC error response actually proves the session is gone. A timeout, transport failure or malformed response does not. The provider may have loaded the session and still hold it open, so clearing the binding and creating a new session forks hidden provider state: two live sessions for one channel, one of them unreachable and never cleaned up. Clear only on `AcpError::AgentError`. Everything else keeps the mapping and logs the indeterminate outcome. That is self-healing rather than sticky: a provider that has genuinely lost the session answers `AgentError` on a later attempt, and that clears it then. Raised by @OpenSpek while validating restart-durable sessions on Windows. Signed-off-by: joelbrilliant --- crates/buzz-acp/src/pool.rs | 62 +++++++++++++++++++++++++++++++++++-- 1 file changed, 60 insertions(+), 2 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 451724eb95a..53a9775ebaf 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -863,13 +863,13 @@ async fn try_load_persisted_session( } Some(resp.session_id) } - Err(e) => { + Err(e) if load_failure_is_definitive(&e) => { tracing::warn!( target: "pool::session", session_id = %stored, channel_id = %channel_id, error = %e, - "session/load failed — clearing stale binding (if unchanged) and creating a new session" + "session/load rejected by agent — clearing stale binding (if unchanged) and creating a new session" ); // Only drop the binding we failed to load. A concurrent process may // already have written a newer session for this channel. @@ -881,9 +881,35 @@ async fn try_load_persisted_session( ); None } + Err(e) => { + tracing::warn!( + target: "pool::session", + session_id = %stored, + channel_id = %channel_id, + error = %e, + "session/load outcome indeterminate — keeping binding and creating a new session; \ + the stored session may still be live on the provider" + ); + None + } } } +/// Whether a failed `session/load` proves the stored binding is dead. +/// +/// Only a JSON-RPC error response is definitive: the provider answered and +/// refused, so the session is genuinely gone and the binding is safe to drop. +/// +/// Everything else is indeterminate. A timeout, transport failure or malformed +/// response does NOT prove the provider failed to load — it may hold the session +/// open. Dropping the binding on those and falling through to `session/new` +/// would fork hidden provider state: two live sessions, one unreachable. Keeping +/// the mapping is self-healing, because a provider that has genuinely lost the +/// session answers `AgentError` on a later attempt and that clears it then. +fn load_failure_is_definitive(error: &AcpError) -> bool { + matches!(error, AcpError::AgentError { .. }) +} + /// Create a new ACP session via `session_new_full()`, populate model capabilities /// on the agent (first session only), and apply `desired_model` if set. /// @@ -3758,6 +3784,38 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec) #[cfg(test)] mod tests { + + /// A `session/load` failure only clears the durable binding when the + /// provider actually answered and refused. Timeouts, transport failures and + /// malformed responses are indeterminate: the provider may hold the session + /// open, and clearing the binding there would fork hidden state into two + /// live sessions with one unreachable. + #[test] + fn only_an_agent_error_is_a_definitive_session_load_failure() { + use std::time::Duration; + + assert!(super::load_failure_is_definitive(&AcpError::AgentError { + code: -32602, + message: "no such session".into(), + })); + + for indeterminate in [ + AcpError::Timeout(Duration::from_secs(1)), + AcpError::IdleTimeout(Duration::from_secs(1)), + AcpError::WriteTimeout(Duration::from_secs(1)), + AcpError::CancelDrainTimeout(Duration::from_secs(1)), + AcpError::HardTimeout { + silence: Duration::from_secs(1), + }, + AcpError::AgentExited, + AcpError::Protocol("truncated frame".into()), + ] { + assert!( + !super::load_failure_is_definitive(&indeterminate), + "{indeterminate:?} must not clear the binding" + ); + } + } use super::*; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; From 0b8e2452e92c80867ae0d6ff8a61ef022699ba50 Mon Sep 17 00:00:00 2001 From: Luke Westlake <35435298+LukeWestlake@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:17:35 -0400 Subject: [PATCH 3/3] fix(acp): fail closed on indeterminate sessions Signed-off-by: Luke Westlake <35435298+LukeWestlake@users.noreply.github.com> --- crates/buzz-acp/src/lib.rs | 41 +- crates/buzz-acp/src/pool.rs | 472 +++++++++---- crates/buzz-acp/src/session_store.rs | 976 ++++++++++++++++++++++++++- 3 files changed, 1308 insertions(+), 181 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 16ddc7dc299..8c1fbe6e4e7 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -38,8 +38,9 @@ use filter::SubscriptionRule; use futures_util::FutureExt; use nostr::{PublicKey, ToBech32}; use pool::{ - AgentPool, ControlSignal, IdleSwitchResult, OwnedAgent, PromptContext, PromptOutcome, - PromptResult, PromptSource, SessionState, TimeoutKind, + failure_batch_disposition, AgentPool, ControlSignal, FailureBatchDisposition, IdleSwitchResult, + OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource, SessionRestoreFailure, + SessionState, TimeoutKind, SESSION_RESTORE_INDETERMINATE_NOTICE, }; use pool_lifecycle::PoolLifecycle; use queue::{CancelReason, EventQueue, FlushBatch, QueuedEvent, ThreadTags}; @@ -3078,7 +3079,23 @@ fn handle_prompt_result( // Don't requeue batches for channels the agent was removed from — // those events are stale and should be silently dropped. if !removed_channels.contains(&batch.channel_id) { - if matches!( + if failure_batch_disposition(&result.outcome) + == FailureBatchDisposition::BestEffortNoticeAndDrop + { + tracing::warn!( + channel_id = %batch.channel_id, + events = batch.events.len(), + "session restore blocked — attempting best-effort notice and dropping batch without retry" + ); + // One detached publication attempt for this blocked result. The + // batch is dropped regardless of success; this is neither + // guaranteed delivery nor an exactly-once incident outbox. + spawn_failure_notice( + rest_client, + &batch, + SESSION_RESTORE_INDETERMINATE_NOTICE.to_string(), + ); + } else if matches!( result.outcome, PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_) ) { @@ -3192,6 +3209,7 @@ fn handle_prompt_result( let outcome_label = match &result.outcome { PromptOutcome::Ok(_) => "ok", PromptOutcome::Error(_) => "error", + PromptOutcome::SessionRestoreIndeterminate(_) => "session_restore_indeterminate", PromptOutcome::Timeout(TimeoutKind::Idle) => "idle_timeout", PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => "hard_timeout", PromptOutcome::AgentExited => "exited", @@ -3353,13 +3371,28 @@ fn handle_prompt_result( ); pool.return_agent(result.agent); } - PromptOutcome::Error(ref e) => { + PromptOutcome::SessionRestoreIndeterminate(SessionRestoreFailure::Quarantined) + | PromptOutcome::SessionRestoreIndeterminate(SessionRestoreFailure::Unavailable) => { + tracing::warn!( + agent = agent_index, + outcome = outcome_label, + configured_model = %harness_configured_model, + pid = harness_pid, + "agent_returned (session restore remains safely blocked)" + ); + emit_turn_error(SESSION_RESTORE_INDETERMINATE_NOTICE, None); + pool.return_agent(result.agent); + } + PromptOutcome::Error(ref e) + | PromptOutcome::SessionRestoreIndeterminate(SessionRestoreFailure::Load(ref e)) + | PromptOutcome::SessionRestoreIndeterminate(SessionRestoreFailure::Create(ref e)) => { let is_transport_error = matches!( e, acp::AcpError::Io(_) | acp::AcpError::WriteTimeout(_) | acp::AcpError::Timeout(_) | acp::AcpError::Protocol(_) + | acp::AcpError::AgentExited ); let error_code = match &e { acp::AcpError::AgentError { code, .. } => Some(*code), diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 53a9775ebaf..2f9f4a89143 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -39,6 +39,7 @@ use crate::queue::{ PromptProfile, PromptProfileLookup, ThreadTags, }; use crate::relay::{ChannelInfo, RestClient}; +use crate::session_store::RestoreClaim; /// Window within which agent activity before a hard-cap death qualifies /// the turn as "recently active" (eligible for requeue instead of dead-letter). @@ -403,11 +404,30 @@ pub enum TimeoutKind { Hard { recently_active: bool }, } +/// Why a durable session restore is blocked. +pub enum SessionRestoreFailure { + /// A `session/load` request was attempted, but its outcome is ambiguous. + Load(AcpError), + /// A write-ahead-protected `session/new` request failed after creation + /// intent was committed, so replacement creation is unsafe. + Create(AcpError), + /// An earlier ambiguous attempt is already durably quarantined. No wire + /// request was sent for this turn. + Quarantined, + /// Buzz could not safely read, reserve, or commit the durable binding. No + /// stateful ACP wire request was sent for this turn. + Unavailable, +} + /// Outcome of a prompt task. #[allow(dead_code)] pub enum PromptOutcome { Ok(StopReason), Error(AcpError), + /// A durable `session/load` may have succeeded remotely, but Buzz could not + /// prove the outcome. The stored binding is retained and the prompt is + /// blocked rather than retried or redirected into a new session. + SessionRestoreIndeterminate(SessionRestoreFailure), AgentExited, Timeout(TimeoutKind), /// Intentional cancel via `!cancel` command or interrupt mode. @@ -425,6 +445,27 @@ pub enum PromptOutcome { CancelDrainTimeout(Duration), } +pub(crate) const SESSION_RESTORE_INDETERMINATE_NOTICE: &str = + "⚠️ Durable ACP session resolution is safety-blocked. Buzz dropped this request and will not retry `session/load` or `session/new`, deliver the prompt, or create another session. A stateful ACP request may have reached the provider. Ask an operator to stop every Buzz ACP process using this store, locate the sidecar via `BUZZ_ACP_SESSION_STORE` (or its default location), and reconcile this channel's binding and restore guard before restarting. Restart alone does not clear the durable block."; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum FailureBatchDisposition { + RetryNormally, + BestEffortNoticeAndDrop, +} + +/// Decide whether a failed prompt batch follows the ordinary retry policy or +/// is dropped after one best-effort notice attempt for that blocked result. An +/// indeterminate restore must never be requeued: retrying could issue another stateful load or fall through +/// to creation before the stored binding has been reconciled. +pub(crate) fn failure_batch_disposition(outcome: &PromptOutcome) -> FailureBatchDisposition { + if matches!(outcome, PromptOutcome::SessionRestoreIndeterminate(_)) { + FailureBatchDisposition::BestEffortNoticeAndDrop + } else { + FailureBatchDisposition::RetryNormally + } +} + /// Immutable config subset shared (via `Arc`) by all spawned prompt tasks. /// /// Built once from `Config` at startup. Avoids cloning the full config @@ -803,30 +844,90 @@ const CONTROL_CANCEL_GRACE: Duration = Duration::from_secs(5); /// Timeout for permission-mode requests (`session/set_config_option` with `configId: "mode"`). const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5); +/// Result of resolving a durable channel binding before the first prompt. +/// +/// `Indeterminate` is intentionally distinct from `NoBinding`: collapsing the +/// two would let the caller fall through to `session/new` after a load may have +/// succeeded remotely, forking hidden provider state. +enum SessionResolution { + NoBinding, + Loaded(String), + Indeterminate(SessionRestoreFailure), +} + +/// No current ACP error variant is a narrow, protocol-defined proof that a +/// stored session no longer exists. `AgentError` also carries authentication, +/// parameter, and other application failures, so every failed load is +/// indeterminate until ACP defines a specific not-found contract. +fn session_resolution_after_load_failure(error: AcpError) -> SessionResolution { + SessionResolution::Indeterminate(SessionRestoreFailure::Load(error)) +} + /// Try to restore a durable channel session via `session/load`. /// -/// Returns `Some(session_id)` on success. On miss, capability absence, or load -/// failure, clears the stale binding (when present) and returns `None` so the -/// caller can fall through to `session/new`. +/// Only an authoritative missing binding permits `session/new`. A present +/// binding with no load capability is blocked rather than silently replaced. +/// Success returns the loaded session. Any load failure is indeterminate and +/// must block creation, retry, deletion, overwrite, and prompt publication. async fn try_load_persisted_session( agent: &mut OwnedAgent, ctx: &PromptContext, channel_id: &Uuid, _agent_core: Option<&str>, _agent_canvas: Option<&str>, -) -> Option { +) -> SessionResolution { + let stored = match ctx.session_store.claim_restore( + &ctx.agent_command, + &ctx.agent_args, + channel_id, + ) { + RestoreClaim::NoBinding => return SessionResolution::NoBinding, + RestoreClaim::Claimed(session_id) => session_id, + RestoreClaim::Indeterminate(session_id) => { + tracing::warn!( + target: "pool::session", + session_id = %session_id, + channel_id = %channel_id, + "session binding is quarantined after an earlier indeterminate load — blocking without retry" + ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Quarantined); + } + RestoreClaim::Unavailable => { + tracing::warn!( + target: "pool::session", + channel_id = %channel_id, + "session binding could not be reserved safely — blocking without a wire request" + ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Unavailable); + } + }; if !agent.supports_load_session { - return None; + tracing::warn!( + target: "pool::session", + session_id = %stored, + channel_id = %channel_id, + "stored binding exists but agent does not support session/load — blocking replacement session creation" + ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Unavailable); } - let stored = ctx - .session_store - .get(&ctx.agent_command, &ctx.agent_args, channel_id)?; match agent .acp .session_load_full(&ctx.cwd, &stored, ctx.mcp_servers.clone()) .await { Ok(resp) => { + if resp.session_id != stored { + tracing::warn!( + target: "pool::session", + expected_session_id = %stored, + returned_session_id = %resp.session_id, + channel_id = %channel_id, + "session/load returned a different session id — leaving restore guard in place and blocking publication" + ); + ctx.session_store + .block_channel(&ctx.agent_command, &ctx.agent_args, channel_id); + return SessionResolution::Indeterminate(SessionRestoreFailure::Unavailable); + } if agent.model_capabilities.is_none() { agent.model_capabilities = Some(AgentModelCapabilities { config_options_raw: extract_model_config_options(&resp.raw), @@ -842,8 +943,14 @@ async fn try_load_persisted_session( tracing::warn!( target: "pool::session", error = %e, - "model re-apply after session/load failed — continuing with loaded session" + "model re-apply after session/load failed — leaving restore guard in place" + ); + ctx.session_store.block_channel( + &ctx.agent_command, + &ctx.agent_args, + channel_id, ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Load(e)); } } } @@ -857,71 +964,71 @@ async fn try_load_persisted_session( tracing::warn!( target: "pool::session", error = %e, - "permission mode after session/load failed — continuing" + "permission mode after session/load failed — leaving restore guard in place" + ); + ctx.session_store.block_channel( + &ctx.agent_command, + &ctx.agent_args, + channel_id, ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Load(e)); } } - Some(resp.session_id) - } - Err(e) if load_failure_is_definitive(&e) => { - tracing::warn!( - target: "pool::session", - session_id = %stored, - channel_id = %channel_id, - error = %e, - "session/load rejected by agent — clearing stale binding (if unchanged) and creating a new session" - ); - // Only drop the binding we failed to load. A concurrent process may - // already have written a newer session for this channel. - let _ = ctx.session_store.remove_if_equals( + if !ctx.session_store.confirm_restore( &ctx.agent_command, &ctx.agent_args, channel_id, &stored, - ); - None + ) { + tracing::warn!( + target: "pool::session", + session_id = %stored, + channel_id = %channel_id, + "session/load succeeded but durable restore commit could not be confirmed — blocking prompt publication" + ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Unavailable); + } + SessionResolution::Loaded(resp.session_id) } - Err(e) => { + Err(error) => { tracing::warn!( target: "pool::session", session_id = %stored, channel_id = %channel_id, - error = %e, - "session/load outcome indeterminate — keeping binding and creating a new session; \ - the stored session may still be live on the provider" + error = %error, + "session/load outcome indeterminate — write-ahead quarantine remains and new session creation is blocked" ); - None + session_resolution_after_load_failure(error) } } } -/// Whether a failed `session/load` proves the stored binding is dead. -/// -/// Only a JSON-RPC error response is definitive: the provider answered and -/// refused, so the session is genuinely gone and the binding is safe to drop. -/// -/// Everything else is indeterminate. A timeout, transport failure or malformed -/// response does NOT prove the provider failed to load — it may hold the session -/// open. Dropping the binding on those and falling through to `session/new` -/// would fork hidden provider state: two live sessions, one unreachable. Keeping -/// the mapping is self-healing, because a provider that has genuinely lost the -/// session answers `AgentError` on a later attempt and that clears it then. -fn load_failure_is_definitive(error: &AcpError) -> bool { - matches!(error, AcpError::AgentError { .. }) -} - /// Create a new ACP session via `session_new_full()`, populate model capabilities /// on the agent (first session only), and apply `desired_model` if set. /// /// On error from `session_new_full()`, returns the `AcpError` — caller handles /// error reporting. Model-switch failures are logged and gracefully ignored /// (the agent proceeds with its default model). +struct CreatedSession { + session_id: String, + observer_frames: Vec<(&'static str, serde_json::Value)>, +} + +impl CreatedSession { + fn publish_observer_frames(self, acp: &mut AcpClient) -> String { + for (kind, payload) in self.observer_frames { + acp.observe(kind, payload); + } + self.session_id + } +} + async fn create_session_and_apply_model( agent: &mut OwnedAgent, ctx: &PromptContext, agent_core: Option<&str>, agent_canvas: Option<&str>, -) -> Result { +) -> Result { // Build base_prompt + system_prompt + agent core + canvas metadata into a // single prompt. Standard protocol-v2 agents receive it in `session/new`; // Goose receives it through the custom request below. Legacy agents receive @@ -982,6 +1089,9 @@ async fn create_session_and_apply_model( } // Apply desired_model if set, matching against the fresh session/new response. + // Observer frames are accumulated and published only after the caller's + // durable channel-binding commit succeeds. + let mut observer_frames = Vec::new(); // Track whether the switch succeeded so session_config_captured reflects // the post-switch state (not the pre-switch desired state). let switch_succeeded = if let Some(ref desired) = agent.desired_model { @@ -999,14 +1109,14 @@ async fn create_session_and_apply_model( // pick rather than silently no-op. On the busy path the turn has // already been cancelled+requeued by the time we get here, so the // turn restarts on the unchanged model and the user is told no. - agent.acp.observe( + observer_frames.push(( "control_result", serde_json::json!({ "type": "switch_model", "status": "unsupported_model", "modelId": desired, }), - ); + )); false } } @@ -1019,7 +1129,7 @@ async fn create_session_and_apply_model( // post-switch state. modelOverridden reflects whether the switch actually // applied — false on the unsupported arm so the panel doesn't show a // stale override badge. - agent.acp.observe( + observer_frames.push(( "session_config_captured", serde_json::json!({ "configOptions": resp.raw.get("configOptions").cloned().unwrap_or(serde_json::Value::Null), @@ -1030,7 +1140,7 @@ async fn create_session_and_apply_model( // keyed by (agent, relay) like the lifecycle frames. "relayUrl": ctx.relay_url, }), - ); + )); // Apply permission mode if not the agent's built-in default AND the agent // advertises the requested mode in session/new. Agents that don't support @@ -1042,7 +1152,10 @@ async fn create_session_and_apply_model( apply_permission_mode(&mut agent.acp, &resp.session_id, &ctx.permission_mode).await?; } - Ok(resp.session_id) + Ok(CreatedSession { + session_id: resp.session_id, + observer_frames, + }) } /// Send the appropriate ACP model-switch request with a timeout. @@ -1087,26 +1200,15 @@ async fn apply_model_switch( "applied model {desired} via {method_label} on session {session_id}" ); } - // Transport-class errors may have corrupted the stdio stream — propagate - // so the caller can respawn the agent instead of reusing a poisoned one. - Ok(Err(e @ AcpError::Io(_))) - | Ok(Err(e @ AcpError::WriteTimeout(_))) - | Ok(Err(e @ AcpError::Timeout(_))) - | Ok(Err(e @ AcpError::Protocol(_))) - | Ok(Err(e @ AcpError::AgentExited)) => { + // Any rejected or malformed config subrequest leaves the requested + // fresh/restored session transaction unauthorized for publication. + Ok(Err(e)) => { tracing::error!( target: "pool::model", - "fatal error setting model {desired} via {method_label}: {e}" + "failed to authorize model {desired} via {method_label}: {e}" ); return Err(e); } - // Application-level errors (Json, etc.) — agent is fine, just uses default model. - Ok(Err(e)) => { - tracing::warn!( - target: "pool::model", - "failed to set model {desired} via {method_label}: {e} — proceeding with agent default" - ); - } Err(_) => { // Outer timeout fired — the inner send_request may have left the // stream in an unknown state. Treat as transport error. @@ -1120,10 +1222,6 @@ async fn apply_model_switch( Ok(()) } -/// Set the session permission mode via `session/set_config_option`. -/// -/// Non-fatal for most errors: logs and proceeds. The agent falls back -/// to its default permission mode (`"default"`), which still works via /// Check if the agent's `session/new` response advertises a given mode ID /// in `result.modes.availableModes[].id`. Returns `false` if the modes /// field is absent or the mode isn't listed. @@ -1140,10 +1238,10 @@ fn agent_supports_mode(session_new_result: &serde_json::Value, mode_wire: &str) .unwrap_or(false) } -/// per-tool auto-approval in `handle_permission_request`. +/// Set the session permission mode via `session/set_config_option`. /// -/// **Fatal exception:** if the agent process exits (e.g., goose crashes on -/// unrecognized methods), returns `Err(AgentExited)` so the caller can respawn. +/// The requested mode is part of session authorization. Any ACP error is +/// returned so callers can keep durable session resolution blocked. async fn apply_permission_mode( acp: &mut AcpClient, session_id: &str, @@ -1163,26 +1261,15 @@ async fn apply_permission_mode( "applied permission mode {wire:?} on session {session_id}" ); } - // Transport-class errors may have corrupted the stdio stream — propagate - // so the caller can respawn the agent. - Ok(Err(e @ AcpError::Io(_))) - | Ok(Err(e @ AcpError::WriteTimeout(_))) - | Ok(Err(e @ AcpError::Timeout(_))) - | Ok(Err(e @ AcpError::Protocol(_))) - | Ok(Err(e @ AcpError::AgentExited)) => { + // Permission configuration is part of session authorization. Any ACP + // error blocks publication rather than silently changing the policy. + Ok(Err(e)) => { tracing::error!( target: "pool::permission", - "fatal error setting permission mode {wire:?}: {e}" + "failed to authorize permission mode {wire:?}: {e}" ); return Err(e); } - // Application-level errors — agent is fine, just uses default permission mode. - Ok(Err(e)) => { - tracing::warn!( - target: "pool::permission", - "failed to set permission mode {wire:?}: {e} — falling back to per-tool auto-approval" - ); - } Err(_) => { // Outer timeout fired — stream may be in unknown state. tracing::error!( @@ -1582,74 +1669,135 @@ pub async fn run_prompt_task( let (session_id, is_new_session) = match &source { PromptSource::Channel(cid) => { - if let Some(sid) = agent.state.sessions.get(cid) { - (sid.clone(), false) - } else if let Some(sid) = try_load_persisted_session( - &mut agent, - &ctx, - cid, - agent_core.as_deref(), - agent_canvas.as_deref(), - ) - .await - { - tracing::info!( - target: "pool::session", - "loaded session {sid} for channel {cid}" - ); - agent.state.sessions.insert(*cid, sid.clone()); - if let Some((pending_cid, section)) = pending_canvas.take() { - agent.state.canvas_sections.insert(pending_cid, section); + if let Some(sid) = agent.state.sessions.get(cid).cloned() { + if !ctx.session_store.retains_channel_lease( + &ctx.agent_command, + &ctx.agent_args, + cid, + ) { + tracing::warn!( + target: "pool::session", + channel_id = %cid, + "in-memory session has no retained durable binding lease — blocking prompt publication" + ); + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::SessionRestoreIndeterminate( + SessionRestoreFailure::Unavailable, + ), + batch, + ); + return; } (sid, false) } else { - // Create new session with model application. - match create_session_and_apply_model( + match try_load_persisted_session( &mut agent, &ctx, + cid, agent_core.as_deref(), agent_canvas.as_deref(), ) .await { - Ok(sid) => { + SessionResolution::Loaded(sid) => { tracing::info!( target: "pool::session", - "created session {sid} for channel {cid}" + "loaded session {sid} for channel {cid}" ); agent.state.sessions.insert(*cid, sid.clone()); - ctx.session_store - .put(&ctx.agent_command, &ctx.agent_args, cid, &sid); - // Commit canvas only after session creation succeeds (I3). if let Some((pending_cid, section)) = pending_canvas.take() { agent.state.canvas_sections.insert(pending_cid, section); } - (sid, true) + (sid, false) } - Err(AcpError::AgentExited) => { - agent.state.invalidate_all(); + SessionResolution::Indeterminate(error) => { send_prompt_result( &result_tx, &turn_id, agent, source, - PromptOutcome::AgentExited, - requeue_batch_if_queue(&ctx, batch), + PromptOutcome::SessionRestoreIndeterminate(error), + batch, ); return; } - Err(e) => { - // Session creation failed; pending canvas was never committed, - // so the next retry will re-fetch a fresh revision. - send_prompt_result( - &result_tx, - &turn_id, - agent, - source, - PromptOutcome::Error(e), - requeue_batch_if_queue(&ctx, batch), - ); - return; + SessionResolution::NoBinding => { + // Create new session with model application. + match create_session_and_apply_model( + &mut agent, + &ctx, + agent_core.as_deref(), + agent_canvas.as_deref(), + ) + .await + { + Ok(created) => { + let sid = &created.session_id; + tracing::info!( + target: "pool::session", + "created session {sid} for channel {cid}" + ); + if !ctx.session_store.commit_new_binding( + &ctx.agent_command, + &ctx.agent_args, + cid, + sid, + ) { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::SessionRestoreIndeterminate( + SessionRestoreFailure::Unavailable, + ), + batch, + ); + return; + } + let sid = created.publish_observer_frames(&mut agent.acp); + agent.state.sessions.insert(*cid, sid.clone()); + // Commit canvas only after session creation succeeds (I3). + if let Some((pending_cid, section)) = pending_canvas.take() { + agent.state.canvas_sections.insert(pending_cid, section); + } + (sid, true) + } + Err(AcpError::AgentExited) => { + agent.state.invalidate_all(); + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::SessionRestoreIndeterminate( + SessionRestoreFailure::Create(AcpError::AgentExited), + ), + batch, + ); + return; + } + Err(e) => { + // Creation intent was durably committed before + // session/new, so a failed response is blocked + // rather than retried or replaced. + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::SessionRestoreIndeterminate( + SessionRestoreFailure::Create(e), + ), + batch, + ); + return; + } + } } } } @@ -1659,7 +1807,8 @@ pub async fn run_prompt_task( (sid.clone(), false) } else { match create_session_and_apply_model(&mut agent, &ctx, None, None).await { - Ok(sid) => { + Ok(created) => { + let sid = created.publish_observer_frames(&mut agent.acp); tracing::info!( target: "pool::session", "created heartbeat session {sid} for agent {}", @@ -3784,22 +3933,21 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec) #[cfg(test)] mod tests { + use super::*; - /// A `session/load` failure only clears the durable binding when the - /// provider actually answered and refused. Timeouts, transport failures and - /// malformed responses are indeterminate: the provider may hold the session - /// open, and clearing the binding there would fork hidden state into two - /// live sessions with one unreachable. + /// No current ACP error variant proves that a stored session is gone. + /// In particular, `AgentError` also carries authentication and parameter + /// failures. Every failed `session/load` must therefore block fallback to + /// `session/new` rather than deleting or overwriting the durable binding. #[test] - fn only_an_agent_error_is_a_definitive_session_load_failure() { + fn every_session_load_failure_blocks_new_session_fallback() { use std::time::Duration; - assert!(super::load_failure_is_definitive(&AcpError::AgentError { - code: -32602, - message: "no such session".into(), - })); - - for indeterminate in [ + for error in [ + AcpError::AgentError { + code: -32602, + message: "authentication required".into(), + }, AcpError::Timeout(Duration::from_secs(1)), AcpError::IdleTimeout(Duration::from_secs(1)), AcpError::WriteTimeout(Duration::from_secs(1)), @@ -3811,12 +3959,43 @@ mod tests { AcpError::Protocol("truncated frame".into()), ] { assert!( - !super::load_failure_is_definitive(&indeterminate), - "{indeterminate:?} must not clear the binding" + matches!( + super::session_resolution_after_load_failure(error), + SessionResolution::Indeterminate(_) + ), + "a failed load must be indeterminate until ACP defines a narrow not-found contract" ); } } - use super::*; + + /// The indeterminate path is deliberately visible and non-retryable: the + /// triggering batch is used for a best-effort actionable notice, then + /// dropped. Later blocked prompts may emit the notice again; this is not an + /// exactly-once delivery contract or a durable outbox. + #[test] + fn indeterminate_restore_requests_best_effort_notice_without_prompt_retry() { + let outcome = PromptOutcome::SessionRestoreIndeterminate(SessionRestoreFailure::Load( + AcpError::Protocol("load response was malformed".into()), + )); + + assert_eq!( + failure_batch_disposition(&outcome), + FailureBatchDisposition::BestEffortNoticeAndDrop + ); + assert_eq!( + SESSION_RESTORE_INDETERMINATE_NOTICE, + "⚠️ Durable ACP session resolution is safety-blocked. Buzz dropped this request and will not retry `session/load` or `session/new`, deliver the prompt, or create another session. A stateful ACP request may have reached the provider. Ask an operator to stop every Buzz ACP process using this store, locate the sidecar via `BUZZ_ACP_SESSION_STORE` (or its default location), and reconcile this channel's binding and restore guard before restarting. Restart alone does not clear the durable block." + ); + + let create_outcome = PromptOutcome::SessionRestoreIndeterminate( + SessionRestoreFailure::Create(AcpError::Protocol("new response was malformed".into())), + ); + assert_eq!( + failure_batch_disposition(&create_outcome), + FailureBatchDisposition::BestEffortNoticeAndDrop + ); + } + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; @@ -4677,6 +4856,7 @@ mod tests { PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => "Timeout(Hard)", PromptOutcome::CancelDrainTimeout(_) => "CancelDrainTimeout", PromptOutcome::Error(_) => "Error", + PromptOutcome::SessionRestoreIndeterminate(_) => "SessionRestoreIndeterminate", PromptOutcome::Cancelled => "Cancelled", PromptOutcome::Ok(_) => "Ok", }; diff --git a/crates/buzz-acp/src/session_store.rs b/crates/buzz-acp/src/session_store.rs index 1617213a894..10f28619b11 100644 --- a/crates/buzz-acp/src/session_store.rs +++ b/crates/buzz-acp/src/session_store.rs @@ -9,18 +9,26 @@ //! agent binaries / profiles do not share bindings. Heartbeats are never //! stored — they stay ephemeral. //! -//! Cross-process safety: the store is a shared file. Every read and mutation -//! takes a sibling lockfile, reloads the on-disk map under that lock, then -//! writes atomically. A process-local cache alone is unsafe when two -//! `buzz-acp` processes share the same agent command/args identity. +//! Cross-process safety: the v1 binding map remains backward-readable, while +//! restore intent is written to a separate per-binding guard that legacy +//! whole-file rewrites cannot erase. Corrected processes also retain a +//! per-binding OS lease for the active channel lifetime. Already-running older +//! binaries do not honor either mechanism and must be stopped before cutover; +//! rolling mixed-version safety cannot be enforced by this module. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::ffi::OsString; use std::fs::{self, File, OpenOptions}; +use std::io::Write; use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +#[cfg(windows)] +use std::os::windows::fs::OpenOptionsExt; use fs2::FileExt; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use uuid::Uuid; use crate::config::normalize_agent_command_identity; @@ -36,10 +44,45 @@ struct StoreFile { sessions: HashMap, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +struct RestoreGuardFile { + version: u32, + state: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + session_id: Option, +} + +const RESTORE_GUARD_VERSION: u32 = 1; + +#[cfg(test)] +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum StoredSessionBinding { + Bound(String), + Indeterminate(String), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum RestoreClaim { + NoBinding, + Claimed(String), + Indeterminate(String), + Unavailable, +} + +#[derive(Debug, PartialEq, Eq)] +enum RestoreGuard { + Missing, + Creating, + Ready(String), + Restoring(String), +} + /// Durable session binding store shared across buzz-acp processes. pub struct SessionStore { path: PathBuf, lock_path: PathBuf, + channel_leases: Mutex>, + volatile_blocks: Mutex>, } /// RAII wrapper that unlocks the OS file lock on drop. @@ -58,8 +101,246 @@ impl SessionStore { /// /// Does not cache file contents; each operation reloads under lock. pub fn open(path: PathBuf) -> Self { + // Pin relative BUZZ_ACP_SESSION_STORE overrides to this process's + // current directory up front. A bare filename otherwise has an empty + // parent, which cannot be opened for the final durability sync. + let path = if path.is_absolute() { + path + } else { + std::env::current_dir() + .unwrap_or_else(|_| PathBuf::from(".")) + .join(path) + }; let lock_path = sibling_lock_path(&path); - Self { path, lock_path } + Self { + path, + lock_path, + channel_leases: Mutex::new(HashMap::new()), + volatile_blocks: Mutex::new(HashSet::new()), + } + } + + /// Atomically reserve a channel for this process and record restore intent + /// before any stateful ACP wire request is allowed. + /// + /// The process-local map retains an OS lock for the lifetime of this store, + /// fencing same-version peers even after a successful load clears the + /// durable in-progress marker. A lock/read/serialization/commit failure is + /// distinct from a genuinely absent binding and therefore fails closed. + pub(crate) fn claim_restore( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + ) -> RestoreClaim { + let key = binding_key(agent_command, agent_args, channel_id); + if self.is_key_blocked(&key) { + return RestoreClaim::Unavailable; + } + if !self.retain_channel_lease(&key) { + self.block_key(&key); + return RestoreClaim::Unavailable; + } + + let Some(_lock) = self.acquire_lock(true) else { + self.block_key(&key); + return RestoreClaim::Unavailable; + }; + let data = match load_store(&self.path) { + Ok(data) => data, + Err(e) => { + self.warn_io("failed to read ACP session bindings before restore", &e); + self.block_key(&key); + return RestoreClaim::Unavailable; + } + }; + let guard = match read_restore_guard(&self.path, &key) { + Ok(guard) => guard, + Err(e) => { + self.warn_io("failed to read ACP session restore guard", &e); + self.block_key(&key); + return RestoreClaim::Unavailable; + } + }; + let Some(session_id) = data.sessions.get(&key).cloned() else { + return match guard { + RestoreGuard::Missing => { + if let Err(e) = persist_creating_guard(&self.path, &key) { + self.warn_io("failed to persist ACP session creation intent", &e); + self.block_key(&key); + RestoreClaim::Unavailable + } else { + RestoreClaim::NoBinding + } + } + RestoreGuard::Creating | RestoreGuard::Ready(_) | RestoreGuard::Restoring(_) => { + RestoreClaim::Unavailable + } + }; + }; + match guard { + RestoreGuard::Restoring(ref guarded) if guarded == &session_id => { + return RestoreClaim::Indeterminate(session_id); + } + RestoreGuard::Ready(ref guarded) if guarded == &session_id => { + if let Err(e) = replace_restore_guard( + &self.path, + &key, + &RestoreGuard::Ready(session_id.clone()), + "restoring", + Some(&session_id), + ) { + self.warn_io("failed to persist ACP session restore intent", &e); + self.block_key(&key); + return RestoreClaim::Unavailable; + } + return RestoreClaim::Claimed(session_id); + } + RestoreGuard::Creating | RestoreGuard::Ready(_) | RestoreGuard::Restoring(_) => { + self.block_key(&key); + return RestoreClaim::Unavailable; + } + RestoreGuard::Missing => {} + } + + if let Err(e) = persist_restore_guard(&self.path, &key, &session_id) { + self.warn_io("failed to persist ACP session restore intent", &e); + self.block_key(&key); + return RestoreClaim::Unavailable; + } + RestoreClaim::Claimed(session_id) + } + + /// Commit a successful restore by atomically transitioning the exact + /// write-ahead guard to `Ready`. Either the old `Restoring` state or the + /// new `Ready` state is safe across a crash; no guard deletion is used. + /// Failure blocks publication for the current process. + pub(crate) fn confirm_restore( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + expected_session_id: &str, + ) -> bool { + let key = binding_key(agent_command, agent_args, channel_id); + let Some(_lock) = self.acquire_lock(true) else { + self.block_key(&key); + return false; + }; + let data = match load_store(&self.path) { + Ok(data) => data, + Err(e) => { + self.warn_io("failed to read ACP session bindings after restore", &e); + self.block_key(&key); + return false; + } + }; + let binding_matches = data + .sessions + .get(&key) + .is_some_and(|current| current == expected_session_id); + if !binding_matches { + self.block_key(&key); + return false; + } + let guard_matches = matches!( + read_restore_guard(&self.path, &key), + Ok(RestoreGuard::Restoring(ref current)) if current == expected_session_id + ); + if !guard_matches { + self.block_key(&key); + return false; + } + if let Err(e) = replace_restore_guard( + &self.path, + &key, + &RestoreGuard::Restoring(expected_session_id.to_owned()), + "ready", + Some(expected_session_id), + ) { + self.warn_io("failed to commit ACP session restore", &e); + self.block_key(&key); + return false; + } + true + } + + pub(crate) fn retains_channel_lease( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + ) -> bool { + let key = binding_key(agent_command, agent_args, channel_id); + let leases = match self.channel_leases.lock() { + Ok(leases) => leases, + Err(poisoned) => poisoned.into_inner(), + }; + let retained = leases.contains_key(&key); + drop(leases); + retained && !self.is_key_blocked(&key) + } + + pub(crate) fn block_channel( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + ) { + self.block_key(&binding_key(agent_command, agent_args, channel_id)); + } + + fn is_key_blocked(&self, key: &str) -> bool { + let blocks = match self.volatile_blocks.lock() { + Ok(blocks) => blocks, + Err(poisoned) => poisoned.into_inner(), + }; + blocks.contains(key) + } + + fn block_key(&self, key: &str) { + let mut blocks = match self.volatile_blocks.lock() { + Ok(blocks) => blocks, + Err(poisoned) => poisoned.into_inner(), + }; + blocks.insert(key.to_owned()); + } + + fn retain_channel_lease(&self, key: &str) -> bool { + let mut leases = match self.channel_leases.lock() { + Ok(leases) => leases, + Err(poisoned) => poisoned.into_inner(), + }; + if leases.contains_key(key) { + return true; + } + + let lease_path = channel_lease_path(&self.path, key); + if let Some(parent) = lease_path.parent() { + if let Err(e) = create_dir_all_durable(parent) { + self.warn_io("failed to commit ACP session lease directory", &e); + return false; + } + } + let file = match OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(&lease_path) + { + Ok(file) => file, + Err(e) => { + self.warn_io("failed to open ACP session channel lease", &e); + return false; + } + }; + if let Err(e) = FileExt::try_lock_exclusive(&file) { + self.warn_io("ACP session channel is owned by another process", &e); + return false; + } + leases.insert(key.to_owned(), StoreLock { file }); + true } /// Resolve the default store path for this agent identity. @@ -78,17 +359,35 @@ impl SessionStore { .join(format!("{identity}.json")) } - /// Look up a stored ACP session id for a channel. - pub fn get( + /// Look up the stored ACP session binding and its durable restore state. + #[cfg(test)] + pub(crate) fn get_binding( &self, agent_command: &str, agent_args: &[String], channel_id: &Uuid, - ) -> Option { + ) -> Option { let key = binding_key(agent_command, agent_args, channel_id); let _lock = self.acquire_lock(false)?; match load_store(&self.path) { - Ok(data) => data.sessions.get(&key).cloned(), + Ok(data) => { + let session_id = data.sessions.get(&key)?.clone(); + match read_restore_guard(&self.path, &key) { + Ok(RestoreGuard::Restoring(guarded)) if guarded == session_id => { + Some(StoredSessionBinding::Indeterminate(session_id)) + } + Ok(RestoreGuard::Missing) => Some(StoredSessionBinding::Bound(session_id)), + Ok(RestoreGuard::Ready(guarded)) if guarded == session_id => { + Some(StoredSessionBinding::Bound(session_id)) + } + Ok( + RestoreGuard::Creating + | RestoreGuard::Ready(_) + | RestoreGuard::Restoring(_), + ) + | Err(_) => None, + } + } Err(e) => { self.warn_io("failed to read ACP session bindings", &e); None @@ -96,47 +395,151 @@ impl SessionStore { } } - /// Persist a channel → session binding. - pub fn put( + /// Look up a stored ACP session id without discarding its durable state. + /// Callers that decide whether to load must use [`Self::get_binding`]. + #[cfg(test)] + pub fn get( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + ) -> Option { + self.get_binding(agent_command, agent_args, channel_id) + .map(|binding| match binding { + StoredSessionBinding::Bound(id) | StoredSessionBinding::Indeterminate(id) => id, + }) + } + + /// Commit a newly-created session binding before any session publication. + pub(crate) fn commit_new_binding( &self, agent_command: &str, agent_args: &[String], channel_id: &Uuid, session_id: &str, - ) { + ) -> bool { let key = binding_key(agent_command, agent_args, channel_id); + if self.is_key_blocked(&key) + || !self.retains_channel_lease(agent_command, agent_args, channel_id) + { + self.block_key(&key); + return false; + } let Some(_lock) = self.acquire_lock(true) else { - return; + self.block_key(&key); + return false; }; let mut data = match load_store(&self.path) { Ok(data) => data, - Err(e) if e.kind() == std::io::ErrorKind::InvalidData => { - // Corrupt sidecar: log and recover empty rather than wedging puts forever. + Err(e) => { self.warn_io( - "corrupt ACP session store on update — rewriting from empty map", + "failed to read ACP session bindings before creation commit", &e, ); - StoreFile::default() - } - Err(e) => { - self.warn_io("failed to read ACP session bindings before update", &e); - return; + self.block_key(&key); + return false; } }; + if !matches!( + read_restore_guard(&self.path, &key), + Ok(RestoreGuard::Creating) + ) { + self.block_key(&key); + return false; + } data.version = 1; - data.sessions.insert(key, session_id.to_owned()); + data.sessions.insert(key.clone(), session_id.to_owned()); if let Err(e) = save_store(&self.path, &data) { self.warn_io("failed to persist ACP session binding", &e); + self.block_key(&key); + return false; + } + if let Err(e) = replace_restore_guard( + &self.path, + &key, + &RestoreGuard::Creating, + "ready", + Some(session_id), + ) { + self.warn_io("failed to commit ACP session creation guard", &e); + self.block_key(&key); + return false; + } + true + } + + /// Test-only direct binding writer for storage/CAS fixtures. + #[cfg(test)] + pub fn put( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + session_id: &str, + ) { + let key = binding_key(agent_command, agent_args, channel_id); + let _lock = self.acquire_lock(true).expect("test store lock"); + let mut data = load_store(&self.path).unwrap_or_default(); + data.version = 1; + data.sessions.insert(key.clone(), session_id.to_owned()); + save_store(&self.path, &data).expect("test binding save"); + let _ = remove_restore_guard(&self.path, &key); + } + + /// Quarantine a binding only if it still points at `expected_session_id`. + /// + /// The compare-and-set keeps an old process from quarantining a fresher + /// binding written by another process. Quarantine is durable so a harness + /// restart cannot silently retry an ambiguous stateful load. + #[cfg(test)] + pub fn mark_indeterminate_if_equals( + &self, + agent_command: &str, + agent_args: &[String], + channel_id: &Uuid, + expected_session_id: &str, + ) -> bool { + let key = binding_key(agent_command, agent_args, channel_id); + let Some(_lock) = self.acquire_lock(true) else { + return false; + }; + match load_store(&self.path) { + Ok(data) => { + let matches = data + .sessions + .get(&key) + .is_some_and(|current| current == expected_session_id); + if !matches { + return false; + } + if let Err(e) = persist_restore_guard(&self.path, &key, expected_session_id) { + if e.kind() == std::io::ErrorKind::AlreadyExists + && matches!( + read_restore_guard(&self.path, &key), + Ok(RestoreGuard::Restoring(ref current)) if current == expected_session_id + ) + { + return true; + } + self.warn_io("failed to persist indeterminate ACP session binding", &e); + return false; + } + true + } + Err(e) => { + self.warn_io("failed to read ACP session bindings before quarantine", &e); + false + } } } /// Remove a binding only if it still points at `expected_session_id`. /// - /// Used after a failed `session/load`: another process may have already - /// written a newer session for the same channel, and a key-only remove - /// would delete that fresher binding. + /// Retained as a test-only primitive for compare-and-set coverage; failed + /// loads are quarantined rather than removed. /// /// Returns `true` when a matching binding was removed. + #[cfg(test)] pub fn remove_if_equals( &self, agent_command: &str, @@ -165,6 +568,15 @@ impl SessionStore { ); return false; } + if let Err(e) = remove_restore_guard(&self.path, &key) { + if e.kind() != std::io::ErrorKind::NotFound { + self.warn_io( + "failed to remove ACP session restore guard with binding", + &e, + ); + return false; + } + } true } Err(e) => { @@ -179,8 +591,8 @@ impl SessionStore { fn acquire_lock(&self, exclusive: bool) -> Option { if let Some(parent) = self.lock_path.parent() { - if let Err(e) = fs::create_dir_all(parent) { - self.warn_io("failed to create ACP session store directory", &e); + if let Err(e) = create_dir_all_durable(parent) { + self.warn_io("failed to durably create ACP session store directory", &e); return None; } } @@ -226,6 +638,181 @@ fn sibling_lock_path(path: &Path) -> PathBuf { PathBuf::from(name) } +fn binding_digest(binding_key: &str) -> String { + hex::encode(Sha256::digest(binding_key.as_bytes())) +} + +fn restore_state_dir(path: &Path) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(OsString::from(".restore")); + PathBuf::from(name) +} + +fn channel_lease_path(path: &Path, binding_key: &str) -> PathBuf { + restore_state_dir(path).join(format!("{}.lock", binding_digest(binding_key))) +} + +fn restore_guard_path(path: &Path, binding_key: &str) -> PathBuf { + restore_state_dir(path).join(format!("{}.json", binding_digest(binding_key))) +} + +fn read_restore_guard(path: &Path, binding_key: &str) -> std::io::Result { + match fs::read_to_string(restore_guard_path(path, binding_key)) { + Ok(text) => { + let guard: RestoreGuardFile = serde_json::from_str(&text) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + if guard.version != RESTORE_GUARD_VERSION { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "unsupported ACP session restore guard", + )); + } + match (guard.state.as_str(), guard.session_id) { + ("creating", None) => Ok(RestoreGuard::Creating), + ("ready", Some(session_id)) if !session_id.is_empty() => { + Ok(RestoreGuard::Ready(session_id)) + } + ("restoring", Some(session_id)) if !session_id.is_empty() => { + Ok(RestoreGuard::Restoring(session_id)) + } + _ => Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "invalid ACP session restore guard state", + )), + } + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(RestoreGuard::Missing), + Err(e) => Err(e), + } +} + +fn persist_creating_guard(path: &Path, binding_key: &str) -> std::io::Result<()> { + persist_guard(path, binding_key, "creating", None) +} + +fn persist_restore_guard(path: &Path, binding_key: &str, session_id: &str) -> std::io::Result<()> { + persist_guard(path, binding_key, "restoring", Some(session_id)) +} + +fn persist_guard( + path: &Path, + binding_key: &str, + state: &str, + session_id: Option<&str>, +) -> std::io::Result<()> { + let guard_path = restore_guard_path(path, binding_key); + if let Some(parent) = guard_path.parent() { + create_dir_all_durable(parent)?; + } + let encoded = serde_json::to_vec(&RestoreGuardFile { + version: RESTORE_GUARD_VERSION, + state: state.to_owned(), + session_id: session_id.map(str::to_owned), + }) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + let mut file = OpenOptions::new() + .create_new(true) + .write(true) + .open(&guard_path)?; + file.write_all(&encoded)?; + file.sync_all()?; + let parent = guard_path.parent().ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "guard path has no parent") + })?; + sync_directory(parent) +} + +fn replace_restore_guard( + path: &Path, + binding_key: &str, + expected: &RestoreGuard, + state: &str, + session_id: Option<&str>, +) -> std::io::Result<()> { + if &read_restore_guard(path, binding_key)? != expected { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "ACP session restore guard changed before commit", + )); + } + let guard_path = restore_guard_path(path, binding_key); + let parent = guard_path.parent().ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "guard path has no parent") + })?; + let tmp = guard_path.with_extension("json.tmp"); + let encoded = serde_json::to_vec(&RestoreGuardFile { + version: RESTORE_GUARD_VERSION, + state: state.to_owned(), + session_id: session_id.map(str::to_owned), + }) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(&tmp)?; + file.write_all(&encoded)?; + file.sync_all()?; + drop(file); + fs::rename(&tmp, &guard_path)?; + sync_directory(parent) +} + +#[cfg(test)] +fn remove_restore_guard(path: &Path, binding_key: &str) -> std::io::Result<()> { + let guard_path = restore_guard_path(path, binding_key); + fs::remove_file(&guard_path)?; + let parent = guard_path.parent().ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "guard path has no parent") + })?; + sync_directory(parent) +} + +fn create_dir_all_durable(path: &Path) -> std::io::Result<()> { + let path = if path.is_absolute() { + path.to_owned() + } else { + std::env::current_dir()?.join(path) + }; + let mut missing = Vec::new(); + let mut cursor = Some(path.as_path()); + while let Some(directory) = cursor { + if directory.exists() { + break; + } + missing.push(directory.to_owned()); + cursor = directory.parent(); + } + fs::create_dir_all(&path)?; + // Commit every newly-created directory entry from the highest missing + // ancestor down. Syncing only the leaf parent can lose an entire nested + // BUZZ_ACP_SESSION_STORE subtree after an acknowledged first-run commit. + for directory in missing.iter().rev() { + let parent = directory.parent().ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "new session-store directory has no parent", + ) + })?; + sync_directory(parent)?; + } + Ok(()) +} + +fn sync_directory(path: &Path) -> std::io::Result<()> { + #[cfg(windows)] + const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000; + + let mut options = OpenOptions::new(); + #[cfg(windows)] + options.read(true).write(true); + #[cfg(not(windows))] + options.read(true); + #[cfg(windows)] + options.custom_flags(FILE_FLAG_BACKUP_SEMANTICS); + options.open(path)?.sync_all() +} + fn store_identity(agent_command: &str, agent_args: &[String]) -> String { let cmd = normalize_agent_command_identity(agent_command); let args = agent_args.join(" "); @@ -270,14 +857,24 @@ fn load_store(path: &Path) -> std::io::Result { fn save_store(path: &Path, data: &StoreFile) -> std::io::Result<()> { if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; + create_dir_all_durable(parent)?; } let tmp = path.with_extension("json.tmp"); let json = serde_json::to_string_pretty(data) .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; - fs::write(&tmp, json)?; + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(&tmp)?; + file.write_all(json.as_bytes())?; + file.sync_all()?; + drop(file); fs::rename(&tmp, path)?; - Ok(()) + let parent = path.parent().ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "store path has no parent") + })?; + sync_directory(parent) } #[cfg(test)] @@ -285,6 +882,55 @@ mod tests { use super::*; use tempfile::tempdir; + #[test] + fn first_run_nested_store_creates_complete_durable_state() { + let dir = tempdir().unwrap(); + let path = dir + .path() + .join("nested") + .join("operator") + .join("sessions.json"); + let channel = Uuid::new_v4(); + let args = vec!["chat".to_string()]; + let key = binding_key("hermes", &args, &channel); + let store = SessionStore::open(path.clone()); + + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::NoBinding + )); + assert!(store.commit_new_binding("hermes", &args, &channel, "sess-nested")); + assert!(path.exists()); + assert!(matches!( + read_restore_guard(&path, &key), + Ok(RestoreGuard::Ready(ref id)) if id == "sess-nested" + )); + } + + #[test] + fn bare_relative_store_commits_first_session_without_empty_parent() { + let filename = format!(".buzz-session-store-test-{}.json", Uuid::new_v4()); + let channel = Uuid::new_v4(); + let args = vec!["chat".to_string()]; + let store = SessionStore::open(PathBuf::from(&filename)); + let absolute_path = store.path.clone(); + + assert!(absolute_path.is_absolute()); + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::NoBinding + )); + assert!(store.commit_new_binding("hermes", &args, &channel, "sess-relative")); + assert!(absolute_path.exists()); + + let lock_path = store.lock_path.clone(); + let restore_dir = restore_state_dir(&absolute_path); + drop(store); + fs::remove_file(absolute_path).unwrap(); + fs::remove_file(lock_path).unwrap(); + fs::remove_dir_all(restore_dir).unwrap(); + } + #[test] fn round_trip_binding() { let dir = tempdir().unwrap(); @@ -307,6 +953,274 @@ mod tests { assert!(store2.get("hermes", &["acp".into()], &channel).is_none()); } + #[test] + fn indeterminate_binding_survives_reopen_and_blocks_implicit_reload() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let channel = Uuid::new_v4(); + let args = ["acp".into()]; + let store = SessionStore::open(path.clone()); + + store.put("hermes", &args, &channel, "sess-live"); + assert!(store.mark_indeterminate_if_equals("hermes", &args, &channel, "sess-live")); + + let reopened = SessionStore::open(path); + assert!(matches!( + reopened.get_binding("hermes", &args, &channel), + Some(StoredSessionBinding::Indeterminate(ref id)) if id == "sess-live" + )); + assert!(!reopened.mark_indeterminate_if_equals( + "hermes", + &args, + &channel, + "different-session" + )); + + // Reconciliation is explicit: remove the quarantined mapping and guard + // before writing a replacement. A normal put never erases a guard. + assert!(reopened.remove_if_equals("hermes", &args, &channel, "sess-live")); + reopened.put("hermes", &args, &channel, "sess-reconciled"); + assert!(matches!( + reopened.get_binding("hermes", &args, &channel), + Some(StoredSessionBinding::Bound(ref id)) if id == "sess-reconciled" + )); + } + + #[test] + fn restore_claim_is_durable_before_wire_and_exclusive_across_store_instances() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let channel = Uuid::new_v4(); + let args = ["acp".into()]; + let process_a = SessionStore::open(path.clone()); + let process_b = SessionStore::open(path.clone()); + + process_a.put("hermes", &args, &channel, "sess-live"); + assert!(matches!( + process_a.claim_restore("hermes", &args, &channel), + RestoreClaim::Claimed(ref id) if id == "sess-live" + )); + + // The durable marker is committed before any ACP request is allowed. + let observer = SessionStore::open(path.clone()); + assert!(matches!( + observer.get_binding("hermes", &args, &channel), + Some(StoredSessionBinding::Indeterminate(ref id)) if id == "sess-live" + )); + + // A peer cannot race a second restore while A owns this channel. + assert!(matches!( + process_b.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + + // Confirming a successful load clears only the durable intent. The + // process-local OS lease remains held for the active channel session. + assert!(process_a.confirm_restore("hermes", &args, &channel, "sess-live")); + assert!(matches!( + observer.get_binding("hermes", &args, &channel), + Some(StoredSessionBinding::Bound(ref id)) if id == "sess-live" + )); + assert!(matches!( + process_b.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + + drop(process_a); + // A process that observed a competing owner remains fail-closed. A + // fresh process may claim only after the prior OS lease is released. + assert!(matches!( + process_b.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + let process_c = SessionStore::open(path.clone()); + assert!(matches!( + process_c.claim_restore("hermes", &args, &channel), + RestoreClaim::Claimed(ref id) if id == "sess-live" + )); + + let empty_channel = Uuid::new_v4(); + let empty_a = SessionStore::open(path.clone()); + let empty_b = SessionStore::open(path); + assert!(matches!( + empty_a.claim_restore("hermes", &args, &empty_channel), + RestoreClaim::NoBinding + )); + assert!(matches!( + empty_b.claim_restore("hermes", &args, &empty_channel), + RestoreClaim::Unavailable + )); + } + + #[test] + fn creation_intent_is_durable_before_wire_and_commit_precedes_publication() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let channel = Uuid::new_v4(); + let args = ["acp".into()]; + let store = SessionStore::open(path.clone()); + + // NoBinding is returned only after a durable Creating guard exists. + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::NoBinding + )); + let key = binding_key("hermes", &args, &channel); + assert!(matches!( + read_restore_guard(&path, &key), + Ok(RestoreGuard::Creating) + )); + + // A confirmed session/new is publishable only after the binding file + // is durable and the Creating guard is atomically transitioned to Ready. + assert!(store.commit_new_binding("hermes", &args, &channel, "sess-created")); + assert!(matches!( + store.get_binding("hermes", &args, &channel), + Some(StoredSessionBinding::Bound(ref id)) if id == "sess-created" + )); + assert!(matches!( + read_restore_guard(&path, &key), + Ok(RestoreGuard::Ready(ref id)) if id == "sess-created" + )); + } + + #[test] + fn failed_creation_commit_blocks_same_process_and_restart() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let channel = Uuid::new_v4(); + let args = ["acp".into()]; + let store = SessionStore::open(path.clone()); + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::NoBinding + )); + + // Force the main binding commit to fail after the write-ahead guard, + // as if session/new had already returned an exact session id. + fs::create_dir(&path).unwrap(); + assert!(!store.commit_new_binding("hermes", &args, &channel, "sess-uncommitted")); + fs::remove_dir(&path).unwrap(); + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + drop(store); + + // Repairing/restarting does not turn the missing main binding into + // permission for another session/new; the Creating guard survives. + let reopened = SessionStore::open(path); + assert!(matches!( + reopened.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + } + + #[test] + fn restore_claim_storage_failures_never_become_no_binding() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let channel = Uuid::new_v4(); + let args = ["acp".into()]; + let store = SessionStore::open(path.clone()); + store.put("hermes", &args, &channel, "sess-live"); + + // Make the per-binding guard unreadable. A storage failure must not be + // collapsed into absence or permission for an ACP wire request. + let key = binding_key("hermes", &args, &channel); + fs::create_dir_all(restore_guard_path(&path, &key)).unwrap(); + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + assert_eq!( + load_store(&path) + .unwrap() + .sessions + .get(&key) + .map(String::as_str), + Some("sess-live") + ); + fs::remove_dir(restore_guard_path(&path, &key)).unwrap(); + assert!(matches!( + store.claim_restore("hermes", &args, &channel), + RestoreClaim::Unavailable + )); + + let corrupt_path = dir.path().join("corrupt.json"); + fs::write(&corrupt_path, "{not-json").unwrap(); + let corrupt = SessionStore::open(corrupt_path); + assert!(matches!( + corrupt.claim_restore("hermes", &args, &Uuid::new_v4()), + RestoreClaim::Unavailable + )); + + let unsupported_path = dir.path().join("unsupported.json"); + let unsupported_channel = Uuid::new_v4(); + let unsupported = SessionStore::open(unsupported_path.clone()); + unsupported.put("hermes", &args, &unsupported_channel, "sess-versioned"); + let unsupported_key = binding_key("hermes", &args, &unsupported_channel); + let guard_path = restore_guard_path(&unsupported_path, &unsupported_key); + fs::create_dir_all(guard_path.parent().unwrap()).unwrap(); + fs::write( + guard_path, + r#"{"version":99,"session_id":"sess-versioned"}"#, + ) + .unwrap(); + assert!(matches!( + unsupported.claim_restore("hermes", &args, &unsupported_channel), + RestoreClaim::Unavailable + )); + } + + #[test] + fn legacy_main_store_rewrite_cannot_erase_separate_restore_guard() { + let dir = tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let args = ["acp".into()]; + let guarded_channel = Uuid::new_v4(); + let unrelated_channel = Uuid::new_v4(); + let mismatch_channel = Uuid::new_v4(); + let store = SessionStore::open(path.clone()); + + store.put("hermes", &args, &guarded_channel, "session-x"); + store.put("hermes", &args, &mismatch_channel, "session-old"); + assert!(store.mark_indeterminate_if_equals("hermes", &args, &guarded_channel, "session-x")); + assert!(store.mark_indeterminate_if_equals( + "hermes", + &args, + &mismatch_channel, + "session-old" + )); + + // Simulate the base/v1 writer rewriting the complete main JSON map. + // Separate restore guards are outside that writer's schema and survive. + let _lock = store.acquire_lock(true).unwrap(); + let mut legacy = load_store(&path).unwrap(); + legacy.version = 1; + legacy.sessions.insert( + binding_key("hermes", &args, &unrelated_channel), + "session-other".into(), + ); + legacy.sessions.insert( + binding_key("hermes", &args, &mismatch_channel), + "session-new".into(), + ); + save_store(&path, &legacy).unwrap(); + drop(_lock); + + let corrected = SessionStore::open(path.clone()); + assert!(matches!( + corrected.claim_restore("hermes", &args, &guarded_channel), + RestoreClaim::Indeterminate(ref id) if id == "session-x" + )); + let conflict = SessionStore::open(path); + assert!(matches!( + conflict.claim_restore("hermes", &args, &mismatch_channel), + RestoreClaim::Unavailable + )); + } + #[test] fn different_args_are_isolated() { let dir = tempdir().unwrap();