1use crate::{AcpThread, ThreadEntryId, ThreadId, ToolCallId, ToolCallRequest};
2use agentic_coding_protocol as acp;
3use anyhow::{Context as _, Result};
4use async_trait::async_trait;
5use collections::HashMap;
6use gpui::{App, AppContext, AsyncApp, Context, Entity, Task, WeakEntity};
7use parking_lot::Mutex;
8use project::Project;
9use smol::process::Child;
10use std::{process::ExitStatus, sync::Arc};
11use util::ResultExt;
12
13pub struct AcpServer {
14 connection: Arc<acp::AgentConnection>,
15 threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<AcpThread>>>>,
16 project: Entity<Project>,
17 exit_status: Arc<Mutex<Option<ExitStatus>>>,
18 _handler_task: Task<()>,
19 _io_task: Task<()>,
20}
21
22struct AcpClientDelegate {
23 project: Entity<Project>,
24 threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<AcpThread>>>>,
25 cx: AsyncApp,
26 // sent_buffer_versions: HashMap<Entity<Buffer>, HashMap<u64, BufferSnapshot>>,
27}
28
29impl AcpClientDelegate {
30 fn new(
31 project: Entity<Project>,
32 threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<AcpThread>>>>,
33 cx: AsyncApp,
34 ) -> Self {
35 Self {
36 project,
37 threads,
38 cx: cx,
39 }
40 }
41
42 fn update_thread<R>(
43 &self,
44 thread_id: &ThreadId,
45 cx: &mut App,
46 callback: impl FnOnce(&mut AcpThread, &mut Context<AcpThread>) -> R,
47 ) -> Option<R> {
48 let thread = self.threads.lock().get(&thread_id)?.clone();
49 let Some(thread) = thread.upgrade() else {
50 self.threads.lock().remove(&thread_id);
51 return None;
52 };
53 Some(thread.update(cx, callback))
54 }
55}
56
57#[async_trait(?Send)]
58impl acp::Client for AcpClientDelegate {
59 async fn stream_message_chunk(
60 &self,
61 params: acp::StreamMessageChunkParams,
62 ) -> Result<acp::StreamMessageChunkResponse> {
63 let cx = &mut self.cx.clone();
64
65 cx.update(|cx| {
66 self.update_thread(¶ms.thread_id.into(), cx, |thread, cx| {
67 thread.push_assistant_chunk(params.chunk, cx)
68 });
69 })?;
70
71 Ok(acp::StreamMessageChunkResponse)
72 }
73
74 async fn request_tool_call_confirmation(
75 &self,
76 request: acp::RequestToolCallConfirmationParams,
77 ) -> Result<acp::RequestToolCallConfirmationResponse> {
78 let cx = &mut self.cx.clone();
79 let ToolCallRequest { id, outcome } = cx
80 .update(|cx| {
81 self.update_thread(&request.thread_id.into(), cx, |thread, cx| {
82 thread.request_tool_call(request.label, request.icon, request.content, request.confirmation, cx)
83 })
84 })?
85 .context("Failed to update thread")?;
86
87 Ok(acp::RequestToolCallConfirmationResponse {
88 id: id.into(),
89 outcome: outcome.await?,
90 })
91 }
92
93 async fn push_tool_call(
94 &self,
95 request: acp::PushToolCallParams,
96 ) -> Result<acp::PushToolCallResponse> {
97 let cx = &mut self.cx.clone();
98 let entry_id = cx
99 .update(|cx| {
100 self.update_thread(&request.thread_id.into(), cx, |thread, cx| {
101 thread.push_tool_call(request.label, request.icon, request.content, cx)
102 })
103 })?
104 .context("Failed to update thread")?;
105
106 Ok(acp::PushToolCallResponse {
107 id: entry_id.into(),
108 })
109 }
110
111 async fn update_tool_call(
112 &self,
113 request: acp::UpdateToolCallParams,
114 ) -> Result<acp::UpdateToolCallResponse> {
115 let cx = &mut self.cx.clone();
116
117 cx.update(|cx| {
118 self.update_thread(&request.thread_id.into(), cx, |thread, cx| {
119 thread.update_tool_call(
120 request.tool_call_id.into(),
121 request.status,
122 request.content,
123 cx,
124 )
125 })
126 })?
127 .context("Failed to update thread")??;
128
129 Ok(acp::UpdateToolCallResponse)
130 }
131}
132
133impl AcpServer {
134 pub fn stdio(mut process: Child, project: Entity<Project>, cx: &mut App) -> Arc<Self> {
135 let stdin = process.stdin.take().expect("process didn't have stdin");
136 let stdout = process.stdout.take().expect("process didn't have stdout");
137
138 let threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<AcpThread>>>> = Default::default();
139 let (connection, handler_fut, io_fut) = acp::AgentConnection::connect_to_agent(
140 AcpClientDelegate::new(project.clone(), threads.clone(), cx.to_async()),
141 stdin,
142 stdout,
143 );
144
145 let exit_status: Arc<Mutex<Option<ExitStatus>>> = Default::default();
146 let io_task = cx.background_spawn({
147 let exit_status = exit_status.clone();
148 async move {
149 io_fut.await.log_err();
150 let result = process.status().await.log_err();
151 *exit_status.lock() = result;
152 }
153 });
154
155 Arc::new(Self {
156 project,
157 connection: Arc::new(connection),
158 threads,
159 exit_status,
160 _handler_task: cx.foreground_executor().spawn(handler_fut),
161 _io_task: io_task,
162 })
163 }
164
165 pub async fn initialize(&self) -> Result<acp::InitializeResponse> {
166 self.connection
167 .request(acp::InitializeParams)
168 .await
169 .map_err(to_anyhow)
170 }
171
172 pub async fn authenticate(&self) -> Result<()> {
173 self.connection
174 .request(acp::AuthenticateParams)
175 .await
176 .map_err(to_anyhow)?;
177
178 Ok(())
179 }
180
181 pub async fn create_thread(self: Arc<Self>, cx: &mut AsyncApp) -> Result<Entity<AcpThread>> {
182 let response = self
183 .connection
184 .request(acp::CreateThreadParams)
185 .await
186 .map_err(to_anyhow)?;
187
188 let thread_id: ThreadId = response.thread_id.into();
189 let server = self.clone();
190 let thread = cx.new(|_| AcpThread {
191 // todo!
192 title: "ACP Thread".into(),
193 id: thread_id.clone(), // Either<ErrorState, Id>
194 next_entry_id: ThreadEntryId(0),
195 entries: Vec::default(),
196 project: self.project.clone(),
197 server,
198 })?;
199 self.threads.lock().insert(thread_id, thread.downgrade());
200 Ok(thread)
201 }
202
203 pub async fn send_message(
204 &self,
205 thread_id: ThreadId,
206 message: acp::Message,
207 _cx: &mut AsyncApp,
208 ) -> Result<()> {
209 self.connection
210 .request(acp::SendMessageParams {
211 thread_id: thread_id.clone().into(),
212 message,
213 })
214 .await
215 .map_err(to_anyhow)?;
216 Ok(())
217 }
218
219 pub fn exit_status(&self) -> Option<ExitStatus> {
220 self.exit_status.lock().clone()
221 }
222}
223
224#[track_caller]
225fn to_anyhow(e: acp::Error) -> anyhow::Error {
226 log::error!(
227 "failed to send message: {code}: {message}",
228 code = e.code,
229 message = e.message
230 );
231 anyhow::anyhow!(e.message)
232}
233
234impl From<acp::ThreadId> for ThreadId {
235 fn from(thread_id: acp::ThreadId) -> Self {
236 Self(thread_id.0.into())
237 }
238}
239
240impl From<ThreadId> for acp::ThreadId {
241 fn from(thread_id: ThreadId) -> Self {
242 acp::ThreadId(thread_id.0.to_string())
243 }
244}
245
246impl From<acp::ToolCallId> for ToolCallId {
247 fn from(tool_call_id: acp::ToolCallId) -> Self {
248 Self(ThreadEntryId(tool_call_id.0))
249 }
250}
251
252impl From<ToolCallId> for acp::ToolCallId {
253 fn from(tool_call_id: ToolCallId) -> Self {
254 acp::ToolCallId(tool_call_id.as_u64())
255 }
256}