1use std::{io::Write as _, path::Path, sync::Arc};
2
3use crate::{
4 Agent, AgentThreadEntryContent, AgentThreadSummary, Message, MessageChunk, Role, Thread,
5 ThreadEntryId, ThreadId,
6};
7use agentic_coding_protocol as acp;
8use anyhow::{Context as _, Result};
9use async_trait::async_trait;
10use collections::HashMap;
11use gpui::{App, AppContext, AsyncApp, Context, Entity, Task, WeakEntity};
12use parking_lot::Mutex;
13use project::Project;
14use smol::process::Child;
15use util::ResultExt;
16
17pub struct AcpAgent {
18 connection: Arc<acp::AgentConnection>,
19 threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<Thread>>>>,
20 project: Entity<Project>,
21 _handler_task: Task<()>,
22 _io_task: Task<()>,
23}
24
25struct AcpClientDelegate {
26 project: Entity<Project>,
27 threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<Thread>>>>,
28 cx: AsyncApp,
29 // sent_buffer_versions: HashMap<Entity<Buffer>, HashMap<u64, BufferSnapshot>>,
30}
31
32impl AcpClientDelegate {
33 fn new(
34 project: Entity<Project>,
35 threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<Thread>>>>,
36 cx: AsyncApp,
37 ) -> Self {
38 Self {
39 project,
40 threads,
41 cx: cx,
42 }
43 }
44
45 fn update_thread<R>(
46 &self,
47 thread_id: &ThreadId,
48 cx: &mut App,
49 callback: impl FnMut(&mut Thread, &mut Context<Thread>) -> R,
50 ) -> Option<R> {
51 let thread = self.threads.lock().get(&thread_id)?.clone();
52 let Some(thread) = thread.upgrade() else {
53 self.threads.lock().remove(&thread_id);
54 return None;
55 };
56 Some(thread.update(cx, callback))
57 }
58}
59
60#[async_trait(?Send)]
61impl acp::Client for AcpClientDelegate {
62 async fn stat(&self, params: acp::StatParams) -> Result<acp::StatResponse> {
63 let cx = &mut self.cx.clone();
64 self.project.update(cx, |project, cx| {
65 let path = project
66 .project_path_for_absolute_path(Path::new(¶ms.path), cx)
67 .context("Failed to get project path")?;
68
69 match project.entry_for_path(&path, cx) {
70 // todo! refresh entry?
71 None => Ok(acp::StatResponse {
72 exists: false,
73 is_directory: false,
74 }),
75 Some(entry) => Ok(acp::StatResponse {
76 exists: entry.is_created(),
77 is_directory: entry.is_dir(),
78 }),
79 }
80 })?
81 }
82
83 async fn stream_message_chunk(
84 &self,
85 chunk: acp::StreamMessageChunkParams,
86 ) -> Result<acp::StreamMessageChunkResponse> {
87 Ok(acp::StreamMessageChunkResponse)
88 }
89
90 async fn read_text_file(
91 &self,
92 request: acp::ReadTextFileParams,
93 ) -> Result<acp::ReadTextFileResponse> {
94 let cx = &mut self.cx.clone();
95 let buffer = self
96 .project
97 .update(cx, |project, cx| {
98 let path = project
99 .project_path_for_absolute_path(Path::new(&request.path), cx)
100 .context("Failed to get project path")?;
101 anyhow::Ok(project.open_buffer(path, cx))
102 })??
103 .await?;
104
105 buffer.update(cx, |buffer, cx| {
106 let start = language::Point::new(request.line_offset.unwrap_or(0), 0);
107 let end = match request.line_limit {
108 None => buffer.max_point(),
109 Some(limit) => start + language::Point::new(limit + 1, 0),
110 };
111
112 let content: String = buffer.text_for_range(start..end).collect();
113 self.update_thread(&request.thread_id.into(), cx, |thread, cx| {
114 thread.push_entry(
115 AgentThreadEntryContent::ReadFile {
116 path: request.path.clone(),
117 content: content.clone(),
118 },
119 cx,
120 );
121 });
122
123 acp::ReadTextFileResponse {
124 content,
125 version: acp::FileVersion(0),
126 }
127 })
128 }
129
130 async fn read_binary_file(
131 &self,
132 request: acp::ReadBinaryFileParams,
133 ) -> Result<acp::ReadBinaryFileResponse> {
134 let cx = &mut self.cx.clone();
135 let file = self
136 .project
137 .update(cx, |project, cx| {
138 let (worktree, path) = project
139 .find_worktree(Path::new(&request.path), cx)
140 .context("Failed to get project path")?;
141
142 let task = worktree.update(cx, |worktree, cx| worktree.load_binary_file(&path, cx));
143 anyhow::Ok(task)
144 })??
145 .await?;
146
147 // todo! test
148 let content = cx
149 .background_spawn(async move {
150 let start = request.byte_offset.unwrap_or(0) as usize;
151 let end = request
152 .byte_limit
153 .map(|limit| (start + limit as usize).min(file.content.len()))
154 .unwrap_or(file.content.len());
155
156 let range_content = &file.content[start..end];
157
158 let mut base64_content = Vec::new();
159 let mut base64_encoder = base64::write::EncoderWriter::new(
160 std::io::Cursor::new(&mut base64_content),
161 &base64::engine::general_purpose::STANDARD,
162 );
163 base64_encoder.write_all(range_content)?;
164 drop(base64_encoder);
165
166 // SAFETY: The base64 encoder should not produce non-UTF8.
167 unsafe { anyhow::Ok(String::from_utf8_unchecked(base64_content)) }
168 })
169 .await?;
170
171 Ok(acp::ReadBinaryFileResponse {
172 content,
173 // todo!
174 version: acp::FileVersion(0),
175 })
176 }
177
178 async fn glob_search(&self, request: acp::GlobSearchParams) -> Result<acp::GlobSearchResponse> {
179 todo!()
180 }
181}
182
183impl AcpAgent {
184 pub fn stdio(mut process: Child, project: Entity<Project>, cx: &mut AsyncApp) -> Arc<Self> {
185 let stdin = process.stdin.take().expect("process didn't have stdin");
186 let stdout = process.stdout.take().expect("process didn't have stdout");
187
188 let threads: Arc<Mutex<HashMap<ThreadId, WeakEntity<Thread>>>> = Default::default();
189 let (connection, handler_fut, io_fut) = acp::AgentConnection::connect_to_agent(
190 AcpClientDelegate::new(project.clone(), threads.clone(), cx.clone()),
191 stdin,
192 stdout,
193 );
194
195 let io_task = cx.background_spawn(async move {
196 io_fut.await.log_err();
197 process.status().await.log_err();
198 });
199
200 Arc::new(Self {
201 project,
202 connection: Arc::new(connection),
203 threads,
204 _handler_task: cx.foreground_executor().spawn(handler_fut),
205 _io_task: io_task,
206 })
207 }
208}
209
210#[async_trait(?Send)]
211impl Agent for AcpAgent {
212 async fn threads(&self, cx: &mut AsyncApp) -> Result<Vec<AgentThreadSummary>> {
213 let response = self.connection.request(acp::GetThreadsParams).await?;
214 response
215 .threads
216 .into_iter()
217 .map(|thread| {
218 Ok(AgentThreadSummary {
219 id: thread.id.into(),
220 title: thread.title,
221 created_at: thread.modified_at,
222 })
223 })
224 .collect()
225 }
226
227 async fn create_thread(self: Arc<Self>, cx: &mut AsyncApp) -> Result<Entity<Thread>> {
228 let response = self.connection.request(acp::CreateThreadParams).await?;
229 let thread_id: ThreadId = response.thread_id.into();
230 let agent = self.clone();
231 let thread = cx.new(|_| Thread {
232 id: thread_id.clone(),
233 next_entry_id: ThreadEntryId(0),
234 entries: Vec::default(),
235 project: self.project.clone(),
236 agent,
237 })?;
238 self.threads.lock().insert(thread_id, thread.downgrade());
239 Ok(thread)
240 }
241
242 async fn open_thread(&self, id: ThreadId, cx: &mut AsyncApp) -> Result<Entity<Thread>> {
243 todo!()
244 }
245
246 async fn thread_entries(
247 &self,
248 thread_id: ThreadId,
249 cx: &mut AsyncApp,
250 ) -> Result<Vec<AgentThreadEntryContent>> {
251 let response = self
252 .connection
253 .request(acp::GetThreadEntriesParams {
254 thread_id: thread_id.clone().into(),
255 })
256 .await?;
257
258 Ok(response
259 .entries
260 .into_iter()
261 .map(|entry| match entry {
262 acp::ThreadEntry::Message { message } => {
263 AgentThreadEntryContent::Message(Message {
264 role: match message.role {
265 acp::Role::User => Role::User,
266 acp::Role::Assistant => Role::Assistant,
267 },
268 chunks: message
269 .chunks
270 .into_iter()
271 .map(|chunk| match chunk {
272 acp::MessageChunk::Text { chunk } => MessageChunk::Text { chunk },
273 })
274 .collect(),
275 })
276 }
277 acp::ThreadEntry::ReadFile { path, content } => {
278 AgentThreadEntryContent::ReadFile { path, content }
279 }
280 })
281 .collect())
282 }
283
284 async fn send_thread_message(
285 &self,
286 thread_id: ThreadId,
287 message: crate::Message,
288 cx: &mut AsyncApp,
289 ) -> Result<()> {
290 self.connection
291 .request(acp::SendMessageParams {
292 thread_id: thread_id.clone().into(),
293 message: acp::Message {
294 role: match message.role {
295 Role::User => acp::Role::User,
296 Role::Assistant => acp::Role::Assistant,
297 },
298 chunks: message
299 .chunks
300 .into_iter()
301 .map(|chunk| match chunk {
302 MessageChunk::Text { chunk } => acp::MessageChunk::Text { chunk },
303 MessageChunk::File { .. } => todo!(),
304 MessageChunk::Directory { .. } => todo!(),
305 MessageChunk::Symbol { .. } => todo!(),
306 MessageChunk::Thread { .. } => todo!(),
307 MessageChunk::Fetch { .. } => todo!(),
308 })
309 .collect(),
310 },
311 })
312 .await?;
313 Ok(())
314 }
315}
316
317impl From<acp::ThreadId> for ThreadId {
318 fn from(thread_id: acp::ThreadId) -> Self {
319 Self(thread_id.0.into())
320 }
321}
322
323impl From<ThreadId> for acp::ThreadId {
324 fn from(thread_id: ThreadId) -> Self {
325 acp::ThreadId(thread_id.0.to_string())
326 }
327}