Add ACP session resume support

Author: vipmax

anycode-backend/Cargo.toml

@@ -46,8 +46,8 @@ dirs = "6.0.0"
 shell-words = "1.1.0"
 lsp-types = "0.97.0"
 similar = "2.7.0"
-agent-client-protocol = "0.9.0"
-agent-client-protocol-schema = "0.10.1"
+agent-client-protocol = { version = "0.9.0", features = ["unstable_session_list", "unstable_session_resume"] }
+agent-client-protocol-schema = { version = "0.10.1", features = ["unstable_session_list", "unstable_session_resume"] }
 async-trait = "0.1"
 tokio-util = { version = "0.7.13", features = ["compat"] }
 unicode-segmentation = "1.12.0"

anycode-backend/src/acp.rs

@@ -1,57 +1,79 @@
-use serde::{Deserialize, Serialize};
-use serde_json::{self, json};
-use socketioxide::{extract::{AckSender, Data, SocketRef, State}};
-use tracing::{info, error};
-use anyhow::anyhow;
+use crate::acp::{AcpMessage, AcpPermissionMode};
 use crate::app_state::AppState;
 use crate::error_ack;
-use crate::acp::{AcpMessage, AcpPermissionMode};
+use serde::{Deserialize, Serialize};
+use serde_json::{self, json};
+use socketioxide::extract::{AckSender, Data, SocketRef, State};
 use tokio::sync::broadcast;
+use tracing::{error, info};
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct AcpStartRequest {
     pub agent_id: String,
     pub agent_name: String,
     pub command: String,
     pub args: Vec<String>,
+    pub resume_session_id: Option<String>,
 }
 
 pub async fn handle_acp_start(
     socket: SocketRef,
     Data(request): Data<AcpStartRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_start {:?}", request);
-    let AcpStartRequest { agent_id, agent_name, command, args } = request;
-    let mut acp_manager = state.acp_manager.lock().await;
-
-    // Check if agent already exists
-    if acp_manager.get_agent(&agent_id).is_some() {
-        error_ack!(ack, &agent_id, "Agent {} already running", agent_id);
-    }
+    let AcpStartRequest {
+        agent_id,
+        agent_name,
+        command,
+        args,
+        resume_session_id,
+    } = request;
+    let (start_result, msg_rx) = {
+        let mut acp_manager = state.acp_manager.lock().await;
+
+        // Check if agent already exists
+        if acp_manager.get_agent(&agent_id).is_some() {
+            error_ack!(ack, &agent_id, "Agent {} already running", agent_id);
+        }
 
-    // Start the agent
-    let start_result = acp_manager.start_agent(
-        agent_id.clone(), agent_name, &command, &args
-    ).await;
+        let start_result = acp_manager
+            .start_agent(
+                agent_id.clone(),
+                agent_name,
+                &command,
+                &args,
+                resume_session_id,
+            )
+            .await;
+
+        let msg_rx = if start_result.is_ok() {
+            acp_manager.subscribe(&agent_id)
+        } else {
+            None
+        };
+
+        (start_result, msg_rx)
+    };
 
     match start_result {
-        Ok(_) => {
+        Ok(session_id) => {
             info!("ACP agent {} started successfully", agent_id);
-            
-            // Subscribe to agent messages
-            let msg_rx = acp_manager.subscribe(&agent_id);
-            
+
+            send_agent_history(&socket, &state, &agent_id).await;
+
             // Set up message forwarding
             if let Some(msg_rx) = msg_rx {
                 let agent_id = agent_id.clone();
+                let state = state.0.clone();
                 tokio::spawn(async move {
-                    forward_agent_messages(socket, agent_id, msg_rx).await;
+                    forward_agent_messages(socket, state, agent_id, msg_rx).await;
                 });
             }
 
-            ack.send(&json!({ "success": true, "agent_id": agent_id })).ok();
+            ack.send(&json!({ "success": true, "agent_id": agent_id, "session_id": session_id }))
+                .ok();
         }
         Err(e) => {
             error!("Failed to start ACP agent {}: {}", agent_id, e);
@@ -62,7 +84,10 @@ pub async fn handle_acp_start(
 
 /// Forward new agent messages to the socket
 pub(crate) async fn forward_agent_messages(
-    socket: SocketRef, agent_id: String, mut msg_rx: broadcast::Receiver<AcpMessage>,
+    socket: SocketRef,
+    state: AppState,
+    agent_id: String,
+    mut msg_rx: broadcast::Receiver<AcpMessage>,
 ) {
     loop {
         match msg_rx.recv().await {
@@ -71,30 +96,43 @@ pub(crate) async fn forward_agent_messages(
                 let result = socket.emit("acp:message", &data);
                 if result.is_err() {
                     error!(
-                        "Failed to forward agent message for agent {}: {} to socket {}", 
-                        agent_id, result.err().unwrap(), socket.id
+                        "Failed to forward agent message for agent {}: {} to socket {}",
+                        agent_id,
+                        result.err().unwrap(),
+                        socket.id
                     );
                     break; // Socket disconnected
                 }
             }
             Err(broadcast::error::RecvError::Closed) => {
-                error!(
-                    "Channel closed for agent {}: {}", 
-                    agent_id, socket.id
-                );
+                error!("Channel closed for agent {}: {}", agent_id, socket.id);
                 break; // Channel closed
             }
-            Err(broadcast::error::RecvError::Lagged(_)) => {
+            Err(broadcast::error::RecvError::Lagged(skipped)) => {
                 error!(
-                    "Lagged behind for agent {}: {}", 
-                    agent_id, socket.id
+                    "Lagged behind for agent {}: {} (skipped {} messages), sending full history resync",
+                    agent_id, socket.id, skipped
                 );
+                send_agent_history(&socket, &State(state.clone()), &agent_id).await;
                 continue; // Lagged behind, continue
             }
         }
     }
 }
 
+async fn send_agent_history(socket: &SocketRef, state: &State<AppState>, agent_id: &str) {
+    let mut acp_manager = state.acp_manager.lock().await;
+    if let Some(history) = acp_manager.get_agent_history(agent_id).await {
+        let data = json!({ "agent_id": agent_id, "history": history });
+        if let Err(err) = socket.emit("acp:history", &data) {
+            error!(
+                "Failed to send ACP history for agent {} to socket {}: {}",
+                agent_id, socket.id, err
+            );
+        }
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct AcpPromptRequest {
     pub agent_id: String,
@@ -104,7 +142,7 @@ pub struct AcpPromptRequest {
 pub async fn handle_acp_prompt(
     Data(request): Data<AcpPromptRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_prompt {:?}", request);
     let AcpPromptRequest { agent_id, prompt } = request;
@@ -138,7 +176,7 @@ pub struct AcpStopRequest {
 pub async fn handle_acp_stop(
     Data(request): Data<AcpStopRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_stop {:?}", request);
     let AcpStopRequest { agent_id } = request;
@@ -157,7 +195,7 @@ pub struct AcpCancelRequest {
 pub async fn handle_acp_cancel(
     Data(request): Data<AcpCancelRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_cancel {:?}", request);
     let AcpCancelRequest { agent_id } = request;
@@ -175,22 +213,48 @@ pub async fn handle_acp_cancel(
     }
 }
 
-pub async fn handle_acp_list(
-    ack: AckSender,
-    state: State<AppState>
-) {
+pub async fn handle_acp_list(ack: AckSender, state: State<AppState>) {
     info!("handle_acp_list");
 
     let acp_manager = state.acp_manager.lock().await;
     let agents = acp_manager.list_agents();
-    
-    let agents_json: Vec<serde_json::Value> = agents.iter()
+
+    let agents_json: Vec<serde_json::Value> = agents
+        .iter()
         .map(|(id, name)| json!({ "id": id, "name": name }))
         .collect();
 
-    ack.send(&json!({ "success": true, "agents": agents_json })).ok();
+    ack.send(&json!({ "success": true, "agents": agents_json }))
+        .ok();
 }
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct AcpSessionsListRequest {
+    pub command: String,
+    pub args: Vec<String>,
+}
+
+pub async fn handle_acp_sessions_list(
+    Data(request): Data<AcpSessionsListRequest>,
+    ack: AckSender,
+    state: State<AppState>,
+) {
+    info!("handle_acp_sessions_list {:?}", request);
+    let AcpSessionsListRequest { command, args } = request;
+    let cwd = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
+
+    let acp_manager = state.acp_manager.lock().await;
+    match acp_manager.list_sessions(&command, &args, cwd).await {
+        Ok(sessions) => {
+            ack.send(&json!({ "success": true, "sessions": sessions }))
+                .ok();
+        }
+        Err(err) => {
+            error!("Failed to list ACP sessions: {}", err);
+            error_ack!(ack, &command, "Failed to list ACP sessions: {}", err);
+        }
+    }
+}
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct AcpPermissionResponseRequest {
@@ -207,10 +271,14 @@ pub struct AcpPermissionModeRequest {
 pub async fn handle_acp_permission_response(
     Data(request): Data<AcpPermissionResponseRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_permission_response {:?}", request);
-    let AcpPermissionResponseRequest { agent_id, permission_id, option_id } = request;
+    let AcpPermissionResponseRequest {
+        agent_id,
+        permission_id,
+        option_id,
+    } = request;
 
     let mut acp_manager = state.acp_manager.lock().await;
 
@@ -221,13 +289,22 @@ pub async fn handle_acp_permission_response(
         }
     };
 
-    match agent.send_permission_response(&permission_id, option_id).await {
+    match agent
+        .send_permission_response(&permission_id, option_id)
+        .await
+    {
         Ok(true) => {
-            info!("Permission response sent for agent {} permission {}", agent_id, permission_id);
+            info!(
+                "Permission response sent for agent {} permission {}",
+                agent_id, permission_id
+            );
             ack.send(&json!({ "success": true })).ok();
         }
         Ok(false) => {
-            error!("Permission {} not found for agent {}", permission_id, agent_id);
+            error!(
+                "Permission {} not found for agent {}",
+                permission_id, agent_id
+            );
             error_ack!(ack, &agent_id, "Permission {} not found", permission_id);
         }
         Err(e) => {
@@ -240,7 +317,7 @@ pub async fn handle_acp_permission_response(
 pub async fn handle_acp_permission_mode(
     Data(request): Data<AcpPermissionModeRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_permission_mode {:?}", request);
     let AcpPermissionModeRequest { mode } = request;
@@ -256,19 +333,16 @@ pub async fn handle_acp_permission_mode(
     let acp_manager = state.acp_manager.lock().await;
     acp_manager.set_permission_mode(permission_mode);
 
-    ack.send(&json!({ "success": true, "mode": permission_mode.as_str() })).ok();
+    ack.send(&json!({ "success": true, "mode": permission_mode.as_str() }))
+        .ok();
 }
 
-pub async fn handle_acp_reconnect(
-    socket: SocketRef,
-    ack: AckSender,
-    state: State<AppState>
-) {
+pub async fn handle_acp_reconnect(socket: SocketRef, ack: AckSender, state: State<AppState>) {
     info!("handle_acp_reconnect for socket {}", socket.id);
-    
+
     let mut acp_manager = state.acp_manager.lock().await;
     let agents = acp_manager.list_agents();
-    
+
     // Re-subscribe to all running agents
     for (agent_id, _) in &agents {
         // Get history and send it if not empty
@@ -282,22 +356,25 @@ pub async fn handle_acp_reconnect(
                 }
             }
         }
-        
+
         // Subscribe to new messages in a separate task
         if let Some(msg_rx) = acp_manager.subscribe(agent_id) {
             let socket_clone = socket.clone();
             let agent_id_clone = agent_id.clone();
+            let state_clone = state.0.clone();
             tokio::spawn(async move {
-                forward_agent_messages(socket_clone, agent_id_clone, msg_rx).await;
+                forward_agent_messages(socket_clone, state_clone, agent_id_clone, msg_rx).await;
             });
         }
     }
-        
-    let agents_json: Vec<serde_json::Value> = agents.iter()
+
+    let agents_json: Vec<serde_json::Value> = agents
+        .iter()
         .map(|(id, name)| json!({ "id": id, "name": name }))
         .collect();
 
-    ack.send(&json!({ "success": true, "agents": agents_json })).ok();
+    ack.send(&json!({ "success": true, "agents": agents_json }))
+        .ok();
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -312,15 +389,21 @@ pub struct AcpUndoRequest {
 pub async fn handle_acp_undo(
     Data(request): Data<AcpUndoRequest>,
     ack: AckSender,
-    state: State<AppState>
+    state: State<AppState>,
 ) {
     info!("handle_acp_undo {:?}", request);
-    let AcpUndoRequest { agent_id, checkpoint_id, prompt } = request;
+    let AcpUndoRequest {
+        agent_id,
+        checkpoint_id,
+        prompt,
+    } = request;
 
     let acp_manager = state.acp_manager.lock().await;
 
     let result = if let Some(checkpoint_id) = checkpoint_id {
-        acp_manager.restore_to_checkpoint_id(&agent_id, &checkpoint_id).await
+        acp_manager
+            .restore_to_checkpoint_id(&agent_id, &checkpoint_id)
+            .await
     } else if let Some(prompt) = prompt {
         acp_manager.restore_to_prompt(&agent_id, &prompt).await
     } else {