acp: Clean up entry views on rewind (#36197)

We were leaking diffs and terminals on rewind, we'll now clean them up.
This PR also introduces a refactor of how we mantain the entry view
state to use a `Vec` that's kept in sync with the thread entries.

Release Notes:

- N/A

crates/acp_thread/Cargo.toml

@@ -13,7 +13,7 @@ path = "src/acp_thread.rs"
 doctest = false
 
 [features]
-test-support = ["gpui/test-support", "project/test-support"]
+test-support = ["gpui/test-support", "project/test-support", "dep:parking_lot"]
 
 [dependencies]
 action_log.workspace = true
@@ -29,6 +29,7 @@ gpui.workspace = true
 itertools.workspace = true
 language.workspace = true
 markdown.workspace = true
+parking_lot = { workspace = true, optional = true }
 project.workspace = true
 prompt_store.workspace = true
 serde.workspace = true

crates/acp_thread/src/acp_thread.rs

@@ -1575,11 +1575,7 @@ mod tests {
         let project = Project::test(fs, [], cx).await;
         let connection = Rc::new(FakeAgentConnection::new());
         let thread = cx
-            .spawn(async move |mut cx| {
-                connection
-                    .new_thread(project, Path::new(path!("/test")), &mut cx)
-                    .await
-            })
+            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
             .await
             .unwrap();
 
@@ -1699,11 +1695,7 @@ mod tests {
         ));
 
         let thread = cx
-            .spawn(async move |mut cx| {
-                connection
-                    .new_thread(project, Path::new(path!("/test")), &mut cx)
-                    .await
-            })
+            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
             .await
             .unwrap();
 
@@ -1786,7 +1778,7 @@ mod tests {
             .unwrap();
 
         let thread = cx
-            .spawn(|mut cx| connection.new_thread(project, Path::new(path!("/tmp")), &mut cx))
+            .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
             .await
             .unwrap();
 
@@ -1849,11 +1841,7 @@ mod tests {
         }));
 
         let thread = cx
-            .spawn(async move |mut cx| {
-                connection
-                    .new_thread(project, Path::new(path!("/test")), &mut cx)
-                    .await
-            })
+            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
             .await
             .unwrap();
 
@@ -1961,10 +1949,11 @@ mod tests {
             }
         }));
 
-        let thread = connection
-            .new_thread(project, Path::new(path!("/test")), &mut cx.to_async())
+        let thread = cx
+            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
             .await
             .unwrap();
+
         cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Hi".into()], cx)))
             .await
             .unwrap();
@@ -2021,8 +2010,8 @@ mod tests {
                 .boxed_local()
             }
         }));
-        let thread = connection
-            .new_thread(project, Path::new(path!("/test")), &mut cx.to_async())
+        let thread = cx
+            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
             .await
             .unwrap();
 
@@ -2227,7 +2216,7 @@ mod tests {
             self: Rc<Self>,
             project: Entity<Project>,
             _cwd: &Path,
-            cx: &mut gpui::AsyncApp,
+            cx: &mut gpui::App,
         ) -> Task<gpui::Result<Entity<AcpThread>>> {
             let session_id = acp::SessionId(
                 rand::thread_rng()
@@ -2237,9 +2226,8 @@ mod tests {
                     .collect::<String>()
                     .into(),
             );
-            let thread = cx
-                .new(|cx| AcpThread::new("Test", self.clone(), project, session_id.clone(), cx))
-                .unwrap();
+            let thread =
+                cx.new(|cx| AcpThread::new("Test", self.clone(), project, session_id.clone(), cx));
             self.sessions.lock().insert(session_id, thread.downgrade());
             Task::ready(Ok(thread))
         }

crates/acp_thread/src/connection.rs

@@ -2,7 +2,7 @@ use crate::AcpThread;
 use agent_client_protocol::{self as acp};
 use anyhow::Result;
 use collections::IndexMap;
-use gpui::{AsyncApp, Entity, SharedString, Task};
+use gpui::{Entity, SharedString, Task};
 use project::Project;
 use std::{error::Error, fmt, path::Path, rc::Rc, sync::Arc};
 use ui::{App, IconName};
@@ -22,7 +22,7 @@ pub trait AgentConnection {
         self: Rc<Self>,
         project: Entity<Project>,
         cwd: &Path,
-        cx: &mut AsyncApp,
+        cx: &mut App,
     ) -> Task<Result<Entity<AcpThread>>>;
 
     fn auth_methods(&self) -> &[acp::AuthMethod];
@@ -160,3 +160,155 @@ impl AgentModelList {
         }
     }
 }
+
+#[cfg(feature = "test-support")]
+mod test_support {
+    use std::sync::Arc;
+
+    use collections::HashMap;
+    use futures::future::try_join_all;
+    use gpui::{AppContext as _, WeakEntity};
+    use parking_lot::Mutex;
+
+    use super::*;
+
+    #[derive(Clone, Default)]
+    pub struct StubAgentConnection {
+        sessions: Arc<Mutex<HashMap<acp::SessionId, WeakEntity<AcpThread>>>>,
+        permission_requests: HashMap<acp::ToolCallId, Vec<acp::PermissionOption>>,
+        next_prompt_updates: Arc<Mutex<Vec<acp::SessionUpdate>>>,
+    }
+
+    impl StubAgentConnection {
+        pub fn new() -> Self {
+            Self {
+                next_prompt_updates: Default::default(),
+                permission_requests: HashMap::default(),
+                sessions: Arc::default(),
+            }
+        }
+
+        pub fn set_next_prompt_updates(&self, updates: Vec<acp::SessionUpdate>) {
+            *self.next_prompt_updates.lock() = updates;
+        }
+
+        pub fn with_permission_requests(
+            mut self,
+            permission_requests: HashMap<acp::ToolCallId, Vec<acp::PermissionOption>>,
+        ) -> Self {
+            self.permission_requests = permission_requests;
+            self
+        }
+
+        pub fn send_update(
+            &self,
+            session_id: acp::SessionId,
+            update: acp::SessionUpdate,
+            cx: &mut App,
+        ) {
+            self.sessions
+                .lock()
+                .get(&session_id)
+                .unwrap()
+                .update(cx, |thread, cx| {
+                    thread.handle_session_update(update.clone(), cx).unwrap();
+                })
+                .unwrap();
+        }
+    }
+
+    impl AgentConnection for StubAgentConnection {
+        fn auth_methods(&self) -> &[acp::AuthMethod] {
+            &[]
+        }
+
+        fn new_thread(
+            self: Rc<Self>,
+            project: Entity<Project>,
+            _cwd: &Path,
+            cx: &mut gpui::App,
+        ) -> Task<gpui::Result<Entity<AcpThread>>> {
+            let session_id = acp::SessionId(self.sessions.lock().len().to_string().into());
+            let thread =
+                cx.new(|cx| AcpThread::new("Test", self.clone(), project, session_id.clone(), cx));
+            self.sessions.lock().insert(session_id, thread.downgrade());
+            Task::ready(Ok(thread))
+        }
+
+        fn authenticate(
+            &self,
+            _method_id: acp::AuthMethodId,
+            _cx: &mut App,
+        ) -> Task<gpui::Result<()>> {
+            unimplemented!()
+        }
+
+        fn prompt(
+            &self,
+            _id: Option<UserMessageId>,
+            params: acp::PromptRequest,
+            cx: &mut App,
+        ) -> Task<gpui::Result<acp::PromptResponse>> {
+            let sessions = self.sessions.lock();
+            let thread = sessions.get(&params.session_id).unwrap();
+            let mut tasks = vec![];
+            for update in self.next_prompt_updates.lock().drain(..) {
+                let thread = thread.clone();
+                let update = update.clone();
+                let permission_request = if let acp::SessionUpdate::ToolCall(tool_call) = &update
+                    && let Some(options) = self.permission_requests.get(&tool_call.id)
+                {
+                    Some((tool_call.clone(), options.clone()))
+                } else {
+                    None
+                };
+                let task = cx.spawn(async move |cx| {
+                    if let Some((tool_call, options)) = permission_request {
+                        let permission = thread.update(cx, |thread, cx| {
+                            thread.request_tool_call_authorization(
+                                tool_call.clone(),
+                                options.clone(),
+                                cx,
+                            )
+                        })?;
+                        permission.await?;
+                    }
+                    thread.update(cx, |thread, cx| {
+                        thread.handle_session_update(update.clone(), cx).unwrap();
+                    })?;
+                    anyhow::Ok(())
+                });
+                tasks.push(task);
+            }
+            cx.spawn(async move |_| {
+                try_join_all(tasks).await?;
+                Ok(acp::PromptResponse {
+                    stop_reason: acp::StopReason::EndTurn,
+                })
+            })
+        }
+
+        fn cancel(&self, _session_id: &acp::SessionId, _cx: &mut App) {
+            unimplemented!()
+        }
+
+        fn session_editor(
+            &self,
+            _session_id: &agent_client_protocol::SessionId,
+            _cx: &mut App,
+        ) -> Option<Rc<dyn AgentSessionEditor>> {
+            Some(Rc::new(StubAgentSessionEditor))
+        }
+    }
+
+    struct StubAgentSessionEditor;
+
+    impl AgentSessionEditor for StubAgentSessionEditor {
+        fn truncate(&self, _: UserMessageId, _: &mut App) -> Task<Result<()>> {
+            Task::ready(Ok(()))
+        }
+    }
+}
+
+#[cfg(feature = "test-support")]
+pub use test_support::*;