Fix ACP connection and thread leak (#35670)

When you switched away from an ACP thread, the `AcpThreadView` entity
(and thus thread, and subprocess) was leaked. This happened because we
were using `cx.processor` for the `list` state callback, which uses a
strong reference.

This PR changes the callback so that it holds a weak reference, and adds
some tests and assertions at various levels to make sure we don't
reintroduce the leak in the future.

Release Notes:

- N/A

crates/agent_servers/src/acp/v0.rs

@@ -380,6 +380,7 @@ impl AcpConnection {
 
             let stdin = child.stdin.take().unwrap();
             let stdout = child.stdout.take().unwrap();
+            log::trace!("Spawned (pid: {})", child.id());
 
             let foreground_executor = cx.foreground_executor().clone();
 

crates/agent_servers/src/acp/v1.rs

@@ -19,7 +19,6 @@ pub struct AcpConnection {
     sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
     auth_methods: Vec<acp::AuthMethod>,
     _io_task: Task<Result<()>>,
-    _child: smol::process::Child,
 }
 
 pub struct AcpSession {
@@ -47,6 +46,7 @@ impl AcpConnection {
 
         let stdout = child.stdout.take().expect("Failed to take stdout");
         let stdin = child.stdin.take().expect("Failed to take stdin");
+        log::trace!("Spawned (pid: {})", child.id());
 
         let sessions = Rc::new(RefCell::new(HashMap::default()));
 
@@ -61,7 +61,11 @@ impl AcpConnection {
             }
         });
 
-        let io_task = cx.background_spawn(io_task);
+        let io_task = cx.background_spawn(async move {
+            io_task.await?;
+            drop(child);
+            Ok(())
+        });
 
         let response = connection
             .initialize(acp::InitializeRequest {
@@ -84,7 +88,6 @@ impl AcpConnection {
             connection: connection.into(),
             server_name,
             sessions,
-            _child: child,
             _io_task: io_task,
         })
     }
@@ -155,8 +158,10 @@ impl AgentConnection for AcpConnection {
 
     fn prompt(&self, params: acp::PromptRequest, cx: &mut App) -> Task<Result<()>> {
         let conn = self.connection.clone();
-        cx.foreground_executor()
-            .spawn(async move { Ok(conn.prompt(params).await?) })
+        cx.foreground_executor().spawn(async move {
+            conn.prompt(params).await?;
+            Ok(())
+        })
     }
 
     fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {

crates/agent_servers/src/claude.rs

@@ -125,8 +125,7 @@ impl AgentConnection for ClaudeAgentConnection {
                         session_id.clone(),
                         &mcp_config_path,
                         &cwd,
-                    )
-                    .await?;
+                    )?;
 
                     let pid = child.id();
                     log::trace!("Spawned (pid: {})", pid);
@@ -262,7 +261,7 @@ enum ClaudeSessionMode {
     Resume,
 }
 
-async fn spawn_claude(
+fn spawn_claude(
     command: &AgentServerCommand,
     mode: ClaudeSessionMode,
     session_id: acp::SessionId,

crates/agent_servers/src/e2e_tests.rs

@@ -311,6 +311,27 @@ pub async fn test_cancel(server: impl AgentServer + 'static, cx: &mut TestAppCon
     });
 }
 
+pub async fn test_thread_drop(server: impl AgentServer + 'static, cx: &mut TestAppContext) {
+    let fs = init_test(cx).await;
+    let project = Project::test(fs, [], cx).await;
+    let thread = new_test_thread(server, project.clone(), "/private/tmp", cx).await;
+
+    thread
+        .update(cx, |thread, cx| thread.send_raw("Hello from test!", cx))
+        .await
+        .unwrap();
+
+    thread.read_with(cx, |thread, _| {
+        assert!(thread.entries().len() >= 2, "Expected at least 2 entries");
+    });
+
+    let weak_thread = thread.downgrade();
+    drop(thread);
+
+    cx.executor().run_until_parked();
+    assert!(!weak_thread.is_upgradable());
+}
+
 #[macro_export]
 macro_rules! common_e2e_tests {
     ($server:expr, allow_option_id = $allow_option_id:expr) => {
@@ -351,6 +372,12 @@ macro_rules! common_e2e_tests {
             async fn cancel(cx: &mut ::gpui::TestAppContext) {
                 $crate::e2e_tests::test_cancel($server, cx).await;
             }
+
+            #[::gpui::test]
+            #[cfg_attr(not(feature = "e2e"), ignore)]
+            async fn thread_drop(cx: &mut ::gpui::TestAppContext) {
+                $crate::e2e_tests::test_thread_drop($server, cx).await;
+            }
         }
     };
 }