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..8c1fbe6e4e7 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; @@ -37,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}; @@ -1143,7 +1145,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 +1189,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 +1562,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 +1795,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 +1806,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 +2676,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"); } @@ -3068,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(_) ) { @@ -3182,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", @@ -3343,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), @@ -3792,6 +3835,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 +3847,7 @@ async fn initialize_agent_pool( agent_name, goose_system_prompt_supported: None, protocol_version, + supports_load_session, })); } Ok(Err(e)) => { @@ -3852,7 +3898,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 +3908,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 +3917,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 +5234,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..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). @@ -168,6 +169,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( @@ -401,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. @@ -423,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 @@ -529,6 +572,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,18 +844,191 @@ 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`. +/// +/// 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>, +) -> 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 { + 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); + } + 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), + 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 — leaving restore guard in place" + ); + ctx.session_store.block_channel( + &ctx.agent_command, + &ctx.agent_args, + channel_id, + ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Load(e)); + } + } + } + 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 — leaving restore guard in place" + ); + ctx.session_store.block_channel( + &ctx.agent_command, + &ctx.agent_args, + channel_id, + ); + return SessionResolution::Indeterminate(SessionRestoreFailure::Load(e)); + } + } + if !ctx.session_store.confirm_restore( + &ctx.agent_command, + &ctx.agent_args, + channel_id, + &stored, + ) { + 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(error) => { + tracing::warn!( + target: "pool::session", + session_id = %stored, + channel_id = %channel_id, + error = %error, + "session/load outcome indeterminate — write-ahead quarantine remains and new session creation is blocked" + ); + session_resolution_after_load_failure(error) + } + } +} + /// 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 @@ -867,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 { @@ -884,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 } } @@ -904,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), @@ -915,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 @@ -927,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. @@ -972,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. @@ -1005,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. @@ -1025,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, @@ -1048,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!( @@ -1467,54 +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) + 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()); - // 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; + } + } } } } @@ -1524,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 {}", @@ -3650,6 +3934,68 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec) #[cfg(test)] mod tests { use super::*; + + /// 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 every_session_load_failure_blocks_new_session_fallback() { + use std::time::Duration; + + 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)), + AcpError::CancelDrainTimeout(Duration::from_secs(1)), + AcpError::HardTimeout { + silence: Duration::from_secs(1), + }, + AcpError::AgentExited, + AcpError::Protocol("truncated frame".into()), + ] { + assert!( + matches!( + super::session_resolution_after_load_failure(error), + SessionResolution::Indeterminate(_) + ), + "a failed load must be indeterminate until ACP defines a narrow not-found contract" + ); + } + } + + /// 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; @@ -4510,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", }; @@ -4998,6 +5345,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 +5404,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 +5656,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..10f28619b11 --- /dev/null +++ b/crates/buzz-acp/src/session_store.rs @@ -0,0 +1,1356 @@ +//! 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 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, 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; + +/// 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, +} + +#[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. +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 { + // 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, + 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. + 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 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 { + let key = binding_key(agent_command, agent_args, channel_id); + let _lock = self.acquire_lock(false)?; + match load_store(&self.path) { + 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 + } + } + } + + /// 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 { + self.block_key(&key); + return false; + }; + let mut data = match load_store(&self.path) { + Ok(data) => data, + Err(e) => { + self.warn_io( + "failed to read ACP session bindings before creation commit", + &e, + ); + 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.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`. + /// + /// 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, + 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; + } + 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) => { + 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) = create_dir_all_durable(parent) { + self.warn_io("failed to durably 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 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(" "); + 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() { + 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))?; + 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)?; + 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)] +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(); + 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 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(); + 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")); + } +}