From 957c310456cdf631873ed35ced3a8eacebad7fdc Mon Sep 17 00:00:00 2001 From: JARVIS-coding-Agent Date: Sun, 19 Apr 2026 07:13:05 +0000 Subject: [PATCH 1/4] =?UTF-8?q?fix(acp):=20persist=20thread=E2=86=92sessio?= =?UTF-8?q?n=20mapping=20to=20survive=20pod=20restarts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/acp/pool.rs | 36 ++++++++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/src/acp/pool.rs b/src/acp/pool.rs index 0c7dc93..fe0750c 100644 --- a/src/acp/pool.rs +++ b/src/acp/pool.rs @@ -2,6 +2,7 @@ use crate::acp::connection::AcpConnection; use crate::config::AgentConfig; use anyhow::{anyhow, Result}; use std::collections::HashMap; +use std::path::PathBuf; use std::sync::Arc; use tokio::sync::{Mutex, RwLock}; use tokio::time::Instant; @@ -24,6 +25,7 @@ pub struct SessionPool { state: RwLock, config: AgentConfig, max_sessions: usize, + mapping_path: PathBuf, } type EvictionCandidate = ( @@ -59,14 +61,33 @@ fn get_or_insert_gate( impl SessionPool { pub fn new(config: AgentConfig, max_sessions: usize) -> Self { + let mapping_path = PathBuf::from(&config.working_dir).join("thread_map.json"); + let suspended = Self::load_mapping(&mapping_path); Self { state: RwLock::new(PoolState { active: HashMap::new(), - suspended: HashMap::new(), + suspended, creating: HashMap::new(), }), config, max_sessions, + mapping_path, + } + } + + fn load_mapping(path: &PathBuf) -> HashMap { + match std::fs::read_to_string(path) { + Ok(data) => serde_json::from_str(&data).unwrap_or_else(|e| { + warn!(path = %path.display(), error = %e, "corrupt thread_map.json, starting fresh"); + HashMap::new() + }), + Err(_) => HashMap::new(), + } + } + + fn save_mapping(&self, suspended: &HashMap) { + if let Err(e) = std::fs::write(&self.mapping_path, serde_json::to_string(suspended).unwrap_or_default()) { + warn!(path = %self.mapping_path.display(), error = %e, "failed to persist thread mapping"); } } @@ -206,6 +227,7 @@ impl SessionPool { state.suspended.remove(thread_id); state.active.insert(thread_id.to_string(), new_conn); + self.save_mapping(&state.suspended); Ok(()) } @@ -267,12 +289,22 @@ impl SessionPool { } } } + self.save_mapping(&state.suspended); } pub async fn shutdown(&self) { let mut state = self.state.write().await; + // Persist active sessions so they can be resumed after restart. + for (key, conn) in state.active.iter() { + if let Ok(conn) = conn.try_lock() { + if let Some(sid) = conn.acp_session_id.clone() { + state.suspended.insert(key.clone(), sid); + } + } + } + self.save_mapping(&state.suspended); let count = state.active.len(); - state.active.clear(); // Drop impl kills process groups + state.active.clear(); info!(count, "pool shutdown complete"); } } From eeb9a118dfdc704046725287c5b46b70205cecdc Mon Sep 17 00:00:00 2001 From: JARVIS-coding-Agent Date: Sun, 19 Apr 2026 07:16:44 +0000 Subject: [PATCH 2/4] fix(acp): resolve borrow checker conflict in shutdown() --- src/acp/pool.rs | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/src/acp/pool.rs b/src/acp/pool.rs index fe0750c..b23c9ad 100644 --- a/src/acp/pool.rs +++ b/src/acp/pool.rs @@ -295,12 +295,17 @@ impl SessionPool { pub async fn shutdown(&self) { let mut state = self.state.write().await; // Persist active sessions so they can be resumed after restart. - for (key, conn) in state.active.iter() { - if let Ok(conn) = conn.try_lock() { - if let Some(sid) = conn.acp_session_id.clone() { - state.suspended.insert(key.clone(), sid); - } - } + let to_save: Vec<(String, String)> = state + .active + .iter() + .filter_map(|(key, conn)| { + let conn = conn.try_lock().ok()?; + let sid = conn.acp_session_id.clone()?; + Some((key.clone(), sid)) + }) + .collect(); + for (key, sid) in to_save { + state.suspended.insert(key, sid); } self.save_mapping(&state.suspended); let count = state.active.len(); From 5a8901402076d46fcf4b02ce53718c4160eb0ae0 Mon Sep 17 00:00:00 2001 From: JARVIS-coding-Agent Date: Sun, 19 Apr 2026 07:30:30 +0000 Subject: [PATCH 3/4] =?UTF-8?q?fix(acp):=20address=20review=20feedback=20?= =?UTF-8?q?=E2=80=94=20atomic=20write,=20idiomatic=20Path,=20pretty=20prin?= =?UTF-8?q?t,=20await=20in=20shutdown?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/acp/pool.rs | 33 +++++++++++++++++++-------------- 1 file changed, 19 insertions(+), 14 deletions(-) diff --git a/src/acp/pool.rs b/src/acp/pool.rs index b23c9ad..5be6d7c 100644 --- a/src/acp/pool.rs +++ b/src/acp/pool.rs @@ -2,7 +2,7 @@ use crate::acp::connection::AcpConnection; use crate::config::AgentConfig; use anyhow::{anyhow, Result}; use std::collections::HashMap; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::sync::{Mutex, RwLock}; use tokio::time::Instant; @@ -75,7 +75,7 @@ impl SessionPool { } } - fn load_mapping(path: &PathBuf) -> HashMap { + fn load_mapping(path: &Path) -> HashMap { match std::fs::read_to_string(path) { Ok(data) => serde_json::from_str(&data).unwrap_or_else(|e| { warn!(path = %path.display(), error = %e, "corrupt thread_map.json, starting fresh"); @@ -86,7 +86,15 @@ impl SessionPool { } fn save_mapping(&self, suspended: &HashMap) { - if let Err(e) = std::fs::write(&self.mapping_path, serde_json::to_string(suspended).unwrap_or_default()) { + let data = match serde_json::to_string_pretty(suspended) { + Ok(d) => d, + Err(e) => { + warn!(error = %e, "failed to serialize thread mapping"); + return; + } + }; + let tmp = self.mapping_path.with_extension("json.tmp"); + if let Err(e) = std::fs::write(&tmp, &data).and_then(|_| std::fs::rename(&tmp, &self.mapping_path)) { warn!(path = %self.mapping_path.display(), error = %e, "failed to persist thread mapping"); } } @@ -295,17 +303,14 @@ impl SessionPool { pub async fn shutdown(&self) { let mut state = self.state.write().await; // Persist active sessions so they can be resumed after restart. - let to_save: Vec<(String, String)> = state - .active - .iter() - .filter_map(|(key, conn)| { - let conn = conn.try_lock().ok()?; - let sid = conn.acp_session_id.clone()?; - Some((key.clone(), sid)) - }) - .collect(); - for (key, sid) in to_save { - state.suspended.insert(key, sid); + let keys: Vec = state.active.keys().cloned().collect(); + for key in keys { + if let Some(conn) = state.active.get(&key) { + let conn = conn.lock().await; + if let Some(sid) = conn.acp_session_id.clone() { + state.suspended.insert(key, sid); + } + } } self.save_mapping(&state.suspended); let count = state.active.len(); From c28ab44c0374f23acaac0b9710d2661d9c128c33 Mon Sep 17 00:00:00 2001 From: JARVIS-coding-Agent Date: Sun, 19 Apr 2026 07:34:53 +0000 Subject: [PATCH 4/4] fix(acp): resolve borrow conflict in shutdown by collecting handles first --- src/acp/pool.rs | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/src/acp/pool.rs b/src/acp/pool.rs index 5be6d7c..7fd0b74 100644 --- a/src/acp/pool.rs +++ b/src/acp/pool.rs @@ -302,14 +302,16 @@ impl SessionPool { pub async fn shutdown(&self) { let mut state = self.state.write().await; - // Persist active sessions so they can be resumed after restart. - let keys: Vec = state.active.keys().cloned().collect(); - for key in keys { - if let Some(conn) = state.active.get(&key) { - let conn = conn.lock().await; - if let Some(sid) = conn.acp_session_id.clone() { - state.suspended.insert(key, sid); - } + // Collect handles before borrowing suspended mutably. + let handles: Vec<(String, Arc>)> = state + .active + .iter() + .map(|(k, v)| (k.clone(), Arc::clone(v))) + .collect(); + for (key, conn) in handles { + let conn = conn.lock().await; + if let Some(sid) = conn.acp_session_id.clone() { + state.suspended.insert(key, sid); } } self.save_mapping(&state.suspended);