From 52b4ac3f5ae1f06be01ca824911d11e93c958fc2 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 | 13 + crates/buzz-acp/Cargo.toml | 6 + crates/buzz-acp/src/acp.rs | 41 +++ crates/buzz-acp/src/lib.rs | 30 +- crates/buzz-acp/src/pool.rs | 119 ++++++++ crates/buzz-acp/src/session_store.rs | 442 +++++++++++++++++++++++++++ 6 files changed, 644 insertions(+), 7 deletions(-) create mode 100644 crates/buzz-acp/src/session_store.rs diff --git a/Cargo.lock b/Cargo.lock index 937ead564a..7a2d3536c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -807,7 +807,9 @@ dependencies = [ "buzz-sdk", "chrono", "clap", + "dirs", "evalexpr", + "fs2", "futures-util", "hex", "httparse", @@ -818,6 +820,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", @@ -2986,6 +2989,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 d047849806..d573f3e522 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 93109fa94d..52c01e2364 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -689,6 +689,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 0c4e5f158c..6466a88fef 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; @@ -1140,8 +1141,9 @@ fn any_respawn_in_flight(crash_history: &[SlotCircuit]) -> bool { /// Result of a background respawn task. struct RespawnResult { index: usize, - /// Tuple: (initialized client, protocol version, agent name). - result: Result<(AcpClient, u32, String)>, + /// Tuple: (initialized client, protocol version, agent name, + /// supports session/load). + result: Result<(AcpClient, u32, String, bool)>, } /// Outcome of a non-cancelling steer attempt, forwarded from a per-attempt @@ -1185,7 +1187,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 @@ -1612,6 +1614,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 { @@ -1854,7 +1864,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, @@ -1865,6 +1875,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"); @@ -2785,7 +2796,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"); } @@ -3931,6 +3942,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, @@ -3941,6 +3954,7 @@ async fn initialize_agent_pool( agent_name, goose_system_prompt_supported: None, protocol_version, + supports_load_session, })); } Ok(Err(e)) => { @@ -3991,7 +4005,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}"))?; @@ -4001,6 +4015,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!({ @@ -4009,7 +4024,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. @@ -5396,6 +5411,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 8430307d9c..3c51806cbf 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -170,6 +170,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, } /// Package name reported by `claude-agent-acp` in its `initialize` response. @@ -564,6 +566,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 { @@ -876,6 +884,87 @@ async fn resolve_new_session_channel_context( (is_dm, title_channel, Some(info.channel_type)) } +/// 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. /// @@ -1597,6 +1686,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 { // The title is channel-qualified (`Agent · #channel`) so one // agent in several channels doesn't produce identical session @@ -1619,6 +1726,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); @@ -6018,6 +6127,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 @@ -6076,6 +6186,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 @@ -6542,6 +6653,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 0000000000..1617213a89 --- /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 62892c166d8aa173c8f920931fd00f40298e6784 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 3c51806cbf..89e3c71be8 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -944,13 +944,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. @@ -962,9 +962,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. /// @@ -4135,6 +4161,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 e236fa4a0100f7d095944c1552f98c2577dd2854 Mon Sep 17 00:00:00 2001 From: joelbrilliant Date: Thu, 30 Jul 2026 21:08:05 +1000 Subject: [PATCH 3/3] fix(acp): preserve durable session rotation semantics Signed-off-by: joelbrilliant --- crates/buzz-acp/src/acp.rs | 90 +++++++++++-- crates/buzz-acp/src/lib.rs | 7 + crates/buzz-acp/src/pool.rs | 194 +++++++++++++++++++++++++-- crates/buzz-acp/src/session_store.rs | 45 +++++++ 4 files changed, 317 insertions(+), 19 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 52c01e2364..46e683139f 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -694,7 +694,8 @@ impl AcpClient { /// 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. + /// request loop without entering the observer feed, so relay observers do + /// not republish the loaded transcript. pub async fn session_load_full( &mut self, cwd: &str, @@ -706,7 +707,9 @@ impl AcpClient { "sessionId": session_id, "mcpServers": mcp_servers, }); - let result = self.send_request("session/load", params).await?; + let result = self + .send_request_with_session_update_observer("session/load", params, false) + .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 @@ -1108,7 +1111,7 @@ impl AcpClient { /// Send a JSON-RPC request and wait for the matching response. /// /// Assigns the next available id, writes the NDJSON line to stdin, - /// then calls [`read_until_response`](Self::read_until_response). + /// then reads until the matching response arrives. /// /// The write phase is bounded by `WRITE_TIMEOUT` (30s) and the read phase /// by `REQUEST_TIMEOUT` (60s), so worst-case wall clock is ~90s. Non-prompt @@ -1118,6 +1121,16 @@ impl AcpClient { &mut self, method: &str, params: serde_json::Value, + ) -> Result { + self.send_request_with_session_update_observer(method, params, true) + .await + } + + async fn send_request_with_session_update_observer( + &mut self, + method: &str, + params: serde_json::Value, + observe_session_updates: bool, ) -> Result { let id = self.next_id; self.next_id += 1; @@ -1140,7 +1153,12 @@ impl AcpClient { Err(_) => return Err(AcpError::Timeout(timeout)), } - match tokio::time::timeout(timeout, self.read_until_response(id)).await { + match tokio::time::timeout( + timeout, + self.read_until_response_with_session_update_observer(id, observe_session_updates), + ) + .await + { Ok(result) => result, Err(_) => Err(AcpError::Timeout(timeout)), } @@ -1150,7 +1168,7 @@ impl AcpClient { /// /// After a [`AcpError::Timeout`] from [`send_request`], the agent may /// eventually send the late response. That stale message will sit in the - /// `BufReader` buffer and be silently skipped by the next `read_until_response` + /// `BufReader` buffer and be silently skipped by the next response-read /// call (ID mismatch). However, if the caller wants a clean slate — e.g. /// before retrying the same method — they can call this to consume any /// buffered data with a short deadline. @@ -1210,9 +1228,10 @@ impl AcpClient { /// /// Compares the incoming `id` field as a `serde_json::Value` against /// `json!(expected_id)` so that both numeric and string IDs work correctly. - async fn read_until_response( + async fn read_until_response_with_session_update_observer( &mut self, expected_id: u64, + observe_session_updates: bool, ) -> Result { loop { // LinesCodec::new_with_max_length enforces MAX_LINE_SIZE at the @@ -1256,7 +1275,11 @@ impl AcpClient { continue; } }; - self.observe("acp_read", msg.clone()); + let is_session_update = + msg.get("method").and_then(|v| v.as_str()) == Some("session/update"); + if observe_session_updates || !is_session_update { + self.observe("acp_read", msg.clone()); + } // Check if this is a response to our expected request (has matching id // AND no `method` field — a `method` field means it's an agent-initiated @@ -1303,7 +1326,7 @@ impl AcpClient { } } - /// Idle-aware message loop: like [`read_until_response`] but resets an idle + /// Idle-aware message loop: like the regular response-read path but resets an idle /// deadline on every stdout line. Fires [`AcpError::IdleTimeout`] on silence /// or [`AcpError::HardTimeout`] on absolute wall-clock cap. /// @@ -3294,6 +3317,57 @@ mod tests { assert_eq!(result.unwrap()["worked"], serde_json::json!(true)); } + #[tokio::test] + async fn session_load_suppresses_replayed_updates_from_observer_only() { + let script = r#" + read -t 2 _load + echo '{"jsonrpc":"2.0","method":"session/update","params":{"marker":"replayed"}}' + echo '{"jsonrpc":"2.0","id":0,"result":{}}' + read -t 2 _next + echo '{"jsonrpc":"2.0","method":"session/update","params":{"marker":"live"}}' + echo '{"jsonrpc":"2.0","id":1,"result":{"worked":true}}' + sleep 1 + "#; + let mut client = spawn_script(script).await; + let observer = crate::observer::ObserverHandle::in_process(); + client.set_observer(Some(observer.clone()), 0); + + let loaded = client + .session_load_full("/", "sess-existing", Vec::new()) + .await + .expect("session/load should succeed"); + assert_eq!(loaded.session_id, "sess-existing"); + + let next = client + .send_request("test/echo", serde_json::json!({})) + .await + .expect("follow-up request should succeed"); + assert_eq!(next["worked"], serde_json::json!(true)); + + let observed_reads: Vec<_> = observer + .snapshot() + .into_iter() + .filter(|event| event.kind == "acp_read") + .map(|event| event.payload) + .collect(); + assert!( + !observed_reads + .iter() + .any(|payload| payload["params"]["marker"] == "replayed"), + "session/load replay updates must not enter the observer feed" + ); + assert!( + observed_reads + .iter() + .any(|payload| payload["params"]["marker"] == "live"), + "normal session updates must remain observable after load" + ); + assert!( + observed_reads.iter().any(|payload| payload["id"] == 0), + "the session/load response itself must remain observable" + ); + } + #[tokio::test] async fn keepalive_resets_idle_past_deadline() { // Keepalive session/update lines every 50ms against a 100ms idle deadline. diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 6466a88fef..b4d3bd2aa1 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2190,6 +2190,11 @@ async fn tokio_main() -> Result<()> { if is_rotate { if let Some(owner) = owner_cache.get() { if buzz_event.event.pubkey.to_hex() == *owner { + let durable_cleared = + pool::clear_durable_channel_binding( + &ctx, + &buzz_event.channel_id, + ); let fired = signal_in_flight_task( &mut pool, buzz_event.channel_id, @@ -2198,6 +2203,7 @@ async fn tokio_main() -> Result<()> { if fired { tracing::info!( channel_id = %buzz_event.channel_id, + durable_cleared, "!rotate received — cancelling in-flight turn and rotating session" ); } else { @@ -2205,6 +2211,7 @@ async fn tokio_main() -> Result<()> { tracing::info!( channel_id = %buzz_event.channel_id, invalidated, + durable_cleared, "!rotate received — invalidated idle channel session(s)" ); } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 89e3c71be8..0975fe84a1 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -80,6 +80,54 @@ pub struct AgentModelCapabilities { pub available_models_raw: Option, } +fn model_capabilities_from_response( + response: &serde_json::Value, +) -> Option { + let config_options_raw = extract_model_config_options(response); + let available_models_raw = response + .get("models") + .filter(|models| models.is_object()) + .cloned(); + if config_options_raw.is_empty() && available_models_raw.is_none() { + None + } else { + Some(AgentModelCapabilities { + config_options_raw, + available_models_raw, + }) + } +} + +#[derive(Debug, PartialEq)] +enum LoadModelResolution { + Method(ModelSwitchMethod), + Unsupported, + Unverifiable, +} + +fn resolve_load_model_switch( + load_response: &serde_json::Value, + cached: Option<&AgentModelCapabilities>, + desired_model: &str, +) -> LoadModelResolution { + if model_capabilities_from_response(load_response).is_some() { + return resolve_model_switch_method(load_response, desired_model) + .map(LoadModelResolution::Method) + .unwrap_or(LoadModelResolution::Unsupported); + } + + let Some(cached) = cached else { + return LoadModelResolution::Unverifiable; + }; + let cached_response = serde_json::json!({ + "configOptions": cached.config_options_raw, + "models": cached.available_models_raw, + }); + resolve_model_switch_method(&cached_response, desired_model) + .map(LoadModelResolution::Method) + .unwrap_or(LoadModelResolution::Unsupported) +} + /// Per-channel session IDs and turn counters. /// /// Separated from `OwnedAgent` so the state machine is testable without @@ -271,6 +319,25 @@ fn apply_completed_before_control_signal( } } +fn control_signal_discards_session(control_signal: &ControlSignal) -> bool { + matches!( + control_signal, + ControlSignal::Rotate | ControlSignal::SwitchModel(_) + ) +} + +pub(crate) fn clear_durable_channel_binding(ctx: &PromptContext, channel_id: &Uuid) -> bool { + ctx.session_store + .remove(&ctx.agent_command, &ctx.agent_args, channel_id) +} + +fn clear_durable_source_binding(ctx: &PromptContext, source: &PromptSource) -> bool { + match source { + PromptSource::Channel(channel_id) => clear_durable_channel_binding(ctx, channel_id), + PromptSource::Heartbeat => false, + } +} + /// Control signal for an in-flight channel turn. /// /// Not `Copy`: `SwitchModel` carries an owned `String`. Callers must clone when @@ -909,21 +976,37 @@ async fn try_load_persisted_session( { 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), - }); + agent.model_capabilities = model_capabilities_from_response(&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 - { + match resolve_load_model_switch( + &resp.raw, + agent.model_capabilities.as_ref(), + desired, + ) { + LoadModelResolution::Method(method) => { + 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" + ); + } + } + LoadModelResolution::Unsupported => { tracing::warn!( - target: "pool::session", - error = %e, - "model re-apply after session/load failed — continuing with loaded session" + target: "pool::model", + "desired model {desired} is absent from the advertised model catalog after session/load" + ); + } + LoadModelResolution::Unverifiable => { + tracing::debug!( + target: "pool::model", + "session/load returned no model catalog and none is cached — leaving desired model unverifiable" ); } } @@ -2131,6 +2214,8 @@ pub async fn run_prompt_task( ) => result, mode = rx => { let control_signal = mode.unwrap_or(ControlSignal::Cancel); + let discard_durable_session = + control_signal_discards_session(&control_signal); // Land the model switch before any cancel/requeue work: setting // `desired_model` here means the fresh session created by the // requeued turn (busy) or the next turn (already-completed) @@ -2151,6 +2236,9 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); agent.state.invalidate(&source); + if discard_durable_session { + clear_durable_source_binding(&ctx, &source); + } let retry_batch = requeue_cancelled_batch(&ctx, control_signal, batch); @@ -2189,6 +2277,9 @@ pub async fn run_prompt_task( } else { agent.state.invalidate(&source); } + if discard_durable_session { + clear_durable_source_binding(&ctx, &source); + } let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( @@ -2245,6 +2336,9 @@ pub async fn run_prompt_task( &source, &control_signal, ); + if discard_durable_session { + clear_durable_source_binding(&ctx, &source); + } let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( &ctx, @@ -2304,6 +2398,7 @@ pub async fn run_prompt_task( "rotating session for {source:?} after {stop_reason:?}", ); agent.state.invalidate(&source); + clear_durable_source_binding(&ctx, &source); } let core_stop = acp_stop_to_core(&stop_reason); @@ -5446,6 +5541,62 @@ mod tests { (s, ch_a, ch_b) } + #[test] + fn load_without_catalog_does_not_poison_model_capabilities() { + let response = json!({"sessionId": "loaded"}); + assert!(model_capabilities_from_response(&response).is_none()); + } + + #[test] + fn load_uses_cached_catalog_when_response_omits_it() { + let cached_response = json!({ + "configOptions": [{ + "configId": "model", + "category": "model", + "options": [{"value": "model-a"}] + }] + }); + let cached = model_capabilities_from_response(&cached_response) + .expect("model catalog should be captured"); + + assert_eq!( + resolve_load_model_switch(&json!({}), Some(&cached), "model-a"), + LoadModelResolution::Method(ModelSwitchMethod::ConfigOption { + config_id: "model".into(), + option_value: "model-a".into(), + }) + ); + } + + #[test] + fn advertised_load_catalog_is_authoritative_even_when_empty() { + let cached_response = json!({ + "models": { + "availableModels": [{"modelId": "model-a"}] + } + }); + let cached = model_capabilities_from_response(&cached_response) + .expect("model catalog should be captured"); + let load_response = json!({ + "models": { + "availableModels": [] + } + }); + + assert_eq!( + resolve_load_model_switch(&load_response, Some(&cached), "model-a"), + LoadModelResolution::Unsupported + ); + } + + #[test] + fn load_without_any_catalog_is_unverifiable() { + assert_eq!( + resolve_load_model_switch(&json!({}), None, "model-a"), + LoadModelResolution::Unverifiable + ); + } + #[test] fn test_rotate_after_natural_completion_invalidates_channel_state() { let (mut s, ch_a, ch_b) = make_state(); @@ -6722,6 +6873,27 @@ mod tests { } } + #[test] + fn explicit_rotation_clears_the_durable_channel_binding() { + let ctx = make_prompt_context_no_owner(); + let channel_id = Uuid::new_v4(); + ctx.session_store.put( + &ctx.agent_command, + &ctx.agent_args, + &channel_id, + "session-before-rotate", + ); + + assert!(clear_durable_source_binding( + &ctx, + &PromptSource::Channel(channel_id) + )); + assert!(ctx + .session_store + .get(&ctx.agent_command, &ctx.agent_args, &channel_id) + .is_none()); + } + // ── render_canvas_section ──────────────────────────────────────────────── #[test] diff --git a/crates/buzz-acp/src/session_store.rs b/crates/buzz-acp/src/session_store.rs index 1617213a89..d663524fdc 100644 --- a/crates/buzz-acp/src/session_store.rs +++ b/crates/buzz-acp/src/session_store.rs @@ -130,6 +130,36 @@ impl SessionStore { } } + /// Remove the current binding for a channel, regardless of session id. + /// + /// This is reserved for explicit discard semantics such as owner-requested + /// rotation. Failed loads must use [`Self::remove_if_equals`] so they cannot + /// delete a newer binding written by another process. + /// + /// Returns `true` when a binding was removed. + pub fn remove(&self, agent_command: &str, agent_args: &[String], channel_id: &Uuid) -> 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) => { + if data.sessions.remove(&key).is_none() { + return false; + } + if let Err(e) = save_store(&self.path, &data) { + self.warn_io("failed to persist ACP session binding removal", &e); + return false; + } + true + } + Err(e) => { + self.warn_io("failed to read ACP session bindings before removal", &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 @@ -439,4 +469,19 @@ mod tests { // No-op when already gone. assert!(!store.remove_if_equals("hermes", &args, &channel, "session-x")); } + + #[test] + fn remove_clears_whichever_binding_is_current() { + 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); + store.put("hermes", &args, &channel, "session-x"); + store.put("hermes", &args, &channel, "session-y"); + + assert!(store.remove("hermes", &args, &channel)); + assert!(store.get("hermes", &args, &channel).is_none()); + assert!(!store.remove("hermes", &args, &channel)); + } }