1mod connection;
2pub use connection::*;
3
4use agent_client_protocol as acp;
5use agentic_coding_protocol as acp_old;
6use anyhow::{Context as _, Result};
7use assistant_tool::ActionLog;
8use buffer_diff::BufferDiff;
9use editor::{Bias, MultiBuffer, PathKey};
10use futures::{FutureExt, channel::oneshot, future::BoxFuture};
11use gpui::{AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Task, WeakEntity};
12use itertools::{Itertools, PeekingNext};
13use language::{
14 Anchor, Buffer, BufferSnapshot, Capability, LanguageRegistry, OffsetRangeExt as _, Point,
15 text_diff,
16};
17use markdown::Markdown;
18use project::{AgentLocation, Project};
19use std::cell::RefCell;
20use std::collections::HashMap;
21use std::error::Error;
22use std::fmt::Formatter;
23use std::rc::Rc;
24use std::{
25 fmt::Display,
26 mem,
27 path::{Path, PathBuf},
28 sync::Arc,
29};
30use ui::App;
31use util::ResultExt;
32
33#[derive(Debug)]
34pub struct UserMessage {
35 pub content: ContentBlock,
36}
37
38impl UserMessage {
39 pub fn from_acp(
40 message: impl IntoIterator<Item = acp::ContentBlock>,
41 language_registry: Arc<LanguageRegistry>,
42 cx: &mut App,
43 ) -> Self {
44 let mut content = ContentBlock::Empty;
45 for chunk in message {
46 content.append(chunk, &language_registry, cx)
47 }
48 Self { content: content }
49 }
50
51 fn to_markdown(&self, cx: &App) -> String {
52 format!("## User\n{}\n", self.content.to_markdown(cx))
53 }
54}
55
56#[derive(Debug)]
57pub struct MentionPath<'a>(&'a Path);
58
59impl<'a> MentionPath<'a> {
60 const PREFIX: &'static str = "@file:";
61
62 pub fn new(path: &'a Path) -> Self {
63 MentionPath(path)
64 }
65
66 pub fn try_parse(url: &'a str) -> Option<Self> {
67 let path = url.strip_prefix(Self::PREFIX)?;
68 Some(MentionPath(Path::new(path)))
69 }
70
71 pub fn path(&self) -> &Path {
72 self.0
73 }
74}
75
76impl Display for MentionPath<'_> {
77 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78 write!(
79 f,
80 "[@{}]({}{})",
81 self.0.file_name().unwrap_or_default().display(),
82 Self::PREFIX,
83 self.0.display()
84 )
85 }
86}
87
88#[derive(Debug, PartialEq)]
89pub struct AssistantMessage {
90 pub chunks: Vec<AssistantMessageChunk>,
91}
92
93impl AssistantMessage {
94 pub fn to_markdown(&self, cx: &App) -> String {
95 format!(
96 "## Assistant\n\n{}\n\n",
97 self.chunks
98 .iter()
99 .map(|chunk| chunk.to_markdown(cx))
100 .join("\n\n")
101 )
102 }
103}
104
105#[derive(Debug, PartialEq)]
106pub enum AssistantMessageChunk {
107 Message { block: ContentBlock },
108 Thought { block: ContentBlock },
109}
110
111impl AssistantMessageChunk {
112 pub fn from_str(chunk: &str, language_registry: &Arc<LanguageRegistry>, cx: &mut App) -> Self {
113 Self::Message {
114 block: ContentBlock::new(
115 acp::ContentBlock::Text(acp::TextContent {
116 text: chunk.to_owned().into(),
117 annotations: None,
118 }),
119 language_registry,
120 cx,
121 ),
122 }
123 }
124
125 fn to_markdown(&self, cx: &App) -> String {
126 match self {
127 Self::Message { block } => block.to_markdown(cx).to_string(),
128 Self::Thought { block } => {
129 format!("<thinking>\n{}\n</thinking>", block.to_markdown(cx))
130 }
131 }
132 }
133}
134
135#[derive(Debug)]
136pub enum AgentThreadEntry {
137 UserMessage(UserMessage),
138 AssistantMessage(AssistantMessage),
139 ToolCall(ToolCall),
140}
141
142impl AgentThreadEntry {
143 fn to_markdown(&self, cx: &App) -> String {
144 match self {
145 Self::UserMessage(message) => message.to_markdown(cx),
146 Self::AssistantMessage(message) => message.to_markdown(cx),
147 Self::ToolCall(too_call) => too_call.to_markdown(cx),
148 }
149 }
150
151 // todo! return all diffs?
152 pub fn first_diff(&self) -> Option<&Diff> {
153 if let AgentThreadEntry::ToolCall(call) = self {
154 call.first_diff()
155 } else {
156 None
157 }
158 }
159
160 pub fn locations(&self) -> Option<&[acp::ToolCallLocation]> {
161 if let AgentThreadEntry::ToolCall(ToolCall { locations, .. }) = self {
162 Some(locations)
163 } else {
164 None
165 }
166 }
167}
168
169#[derive(Debug)]
170pub struct ToolCall {
171 pub id: acp::ToolCallId,
172 pub label: Entity<Markdown>,
173 pub kind: acp::ToolKind,
174 pub content: Vec<ToolCallContent>,
175 pub status: ToolCallStatus,
176 pub locations: Vec<acp::ToolCallLocation>,
177}
178
179impl ToolCall {
180 fn from_acp(
181 tool_call: acp::ToolCall,
182 status: ToolCallStatus,
183 language_registry: Arc<LanguageRegistry>,
184 cx: &mut App,
185 ) -> Self {
186 Self {
187 id: tool_call.id,
188 label: cx.new(|cx| {
189 Markdown::new(
190 tool_call.label.into(),
191 Some(language_registry.clone()),
192 None,
193 cx,
194 )
195 }),
196 kind: tool_call.kind,
197 content: tool_call
198 .content
199 .into_iter()
200 .map(|content| ToolCallContent::from_acp(content, language_registry.clone(), cx))
201 .collect(),
202 locations: tool_call.locations,
203 status,
204 }
205 }
206
207 pub fn first_diff(&self) -> Option<&Diff> {
208 self.content.iter().find_map(|content| match content {
209 ToolCallContent::ContentBlock { .. } => None,
210 ToolCallContent::Diff { diff } => Some(diff),
211 })
212 }
213
214 fn to_markdown(&self, cx: &App) -> String {
215 let mut markdown = format!(
216 "**Tool Call: {}**\nStatus: {}\n\n",
217 self.label.read(cx).source(),
218 self.status
219 );
220 for content in &self.content {
221 markdown.push_str(content.to_markdown(cx).as_str());
222 markdown.push_str("\n\n");
223 }
224 markdown
225 }
226}
227
228#[derive(Debug)]
229pub enum ToolCallStatus {
230 WaitingForConfirmation {
231 options: Vec<acp::PermissionOption>,
232 respond_tx: oneshot::Sender<acp::PermissionOptionId>,
233 },
234 Allowed {
235 status: acp::ToolCallStatus,
236 },
237 Rejected,
238 Canceled,
239}
240
241impl Display for ToolCallStatus {
242 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
243 write!(
244 f,
245 "{}",
246 match self {
247 ToolCallStatus::WaitingForConfirmation { .. } => "Waiting for confirmation",
248 ToolCallStatus::Allowed { status } => match status {
249 acp::ToolCallStatus::InProgress => "In Progress",
250 acp::ToolCallStatus::Completed => "Completed",
251 acp::ToolCallStatus::Failed => "Failed",
252 },
253 ToolCallStatus::Rejected => "Rejected",
254 ToolCallStatus::Canceled => "Canceled",
255 }
256 )
257 }
258}
259
260#[derive(Debug, PartialEq, Clone)]
261pub enum ContentBlock {
262 Empty,
263 Markdown { markdown: Entity<Markdown> },
264}
265
266impl ContentBlock {
267 pub fn new(
268 block: acp::ContentBlock,
269 language_registry: &Arc<LanguageRegistry>,
270 cx: &mut App,
271 ) -> Self {
272 let mut this = Self::Empty;
273 this.append(block, language_registry, cx);
274 this
275 }
276
277 pub fn new_combined(
278 blocks: impl IntoIterator<Item = acp::ContentBlock>,
279 language_registry: Arc<LanguageRegistry>,
280 cx: &mut App,
281 ) -> Self {
282 let mut this = Self::Empty;
283 for block in blocks {
284 this.append(block, &language_registry, cx);
285 }
286 this
287 }
288
289 pub fn append(
290 &mut self,
291 block: acp::ContentBlock,
292 language_registry: &Arc<LanguageRegistry>,
293 cx: &mut App,
294 ) {
295 let new_content = match block {
296 acp::ContentBlock::Text(text_content) => text_content.text.clone(),
297 acp::ContentBlock::ResourceLink(resource_link) => {
298 if let Some(path) = resource_link.uri.strip_prefix("file://") {
299 format!("{}", MentionPath(path.as_ref()))
300 } else {
301 resource_link.uri.clone()
302 }
303 }
304 acp::ContentBlock::Image(_)
305 | acp::ContentBlock::Audio(_)
306 | acp::ContentBlock::Resource(_) => String::new(),
307 };
308
309 match self {
310 ContentBlock::Empty => {
311 *self = ContentBlock::Markdown {
312 markdown: cx.new(|cx| {
313 Markdown::new(
314 new_content.into(),
315 Some(language_registry.clone()),
316 None,
317 cx,
318 )
319 }),
320 };
321 }
322 ContentBlock::Markdown { markdown } => {
323 markdown.update(cx, |markdown, cx| markdown.append(&new_content, cx));
324 }
325 }
326 }
327
328 fn to_markdown<'a>(&'a self, cx: &'a App) -> &'a str {
329 match self {
330 ContentBlock::Empty => "",
331 ContentBlock::Markdown { markdown } => markdown.read(cx).source(),
332 }
333 }
334}
335
336#[derive(Debug)]
337pub enum ToolCallContent {
338 ContentBlock { content: ContentBlock },
339 Diff { diff: Diff },
340}
341
342impl ToolCallContent {
343 pub fn from_acp(
344 content: acp::ToolCallContent,
345 language_registry: Arc<LanguageRegistry>,
346 cx: &mut App,
347 ) -> Self {
348 match content {
349 acp::ToolCallContent::ContentBlock { content } => Self::ContentBlock {
350 content: ContentBlock::new(content, &language_registry, cx),
351 },
352 acp::ToolCallContent::Diff { diff } => Self::Diff {
353 diff: Diff::from_acp(diff, language_registry, cx),
354 },
355 }
356 }
357
358 pub fn from_acp_contents(
359 content: Vec<acp::ToolCallContent>,
360 language_registry: Arc<LanguageRegistry>,
361 cx: &mut App,
362 ) -> Vec<Self> {
363 content
364 .into_iter()
365 .peekable()
366 .batching(|it| match it.next()? {
367 acp::ToolCallContent::ContentBlock { content } => {
368 let mut block = ContentBlock::new(content, &language_registry, cx);
369 while let Some(acp::ToolCallContent::ContentBlock { content }) =
370 it.peeking_next(|c| matches!(c, acp::ToolCallContent::ContentBlock { .. }))
371 {
372 block.append(content, &language_registry, cx);
373 }
374 Some(ToolCallContent::ContentBlock { content: block })
375 }
376 content @ acp::ToolCallContent::Diff { .. } => Some(ToolCallContent::from_acp(
377 content,
378 language_registry.clone(),
379 cx,
380 )),
381 })
382 .collect()
383 }
384
385 pub fn to_markdown(&self, cx: &App) -> String {
386 match self {
387 Self::ContentBlock { content } => content.to_markdown(cx).to_string(),
388 Self::Diff { diff } => diff.to_markdown(cx),
389 }
390 }
391}
392
393#[derive(Debug)]
394pub struct Diff {
395 pub multibuffer: Entity<MultiBuffer>,
396 pub path: PathBuf,
397 pub new_buffer: Entity<Buffer>,
398 pub old_buffer: Entity<Buffer>,
399 _task: Task<Result<()>>,
400}
401
402impl Diff {
403 pub fn from_acp(
404 diff: acp::Diff,
405 language_registry: Arc<LanguageRegistry>,
406 cx: &mut App,
407 ) -> Self {
408 let acp::Diff {
409 path,
410 old_text,
411 new_text,
412 } = diff;
413
414 let multibuffer = cx.new(|_cx| MultiBuffer::without_headers(Capability::ReadOnly));
415
416 let new_buffer = cx.new(|cx| Buffer::local(new_text, cx));
417 let old_buffer = cx.new(|cx| Buffer::local(old_text.unwrap_or("".into()), cx));
418 let new_buffer_snapshot = new_buffer.read(cx).text_snapshot();
419 let old_buffer_snapshot = old_buffer.read(cx).snapshot();
420 let buffer_diff = cx.new(|cx| BufferDiff::new(&new_buffer_snapshot, cx));
421 let diff_task = buffer_diff.update(cx, |diff, cx| {
422 diff.set_base_text(
423 old_buffer_snapshot,
424 Some(language_registry.clone()),
425 new_buffer_snapshot,
426 cx,
427 )
428 });
429
430 let task = cx.spawn({
431 let multibuffer = multibuffer.clone();
432 let path = path.clone();
433 let new_buffer = new_buffer.clone();
434 async move |cx| {
435 diff_task.await?;
436
437 multibuffer
438 .update(cx, |multibuffer, cx| {
439 let hunk_ranges = {
440 let buffer = new_buffer.read(cx);
441 let diff = buffer_diff.read(cx);
442 diff.hunks_intersecting_range(Anchor::MIN..Anchor::MAX, &buffer, cx)
443 .map(|diff_hunk| diff_hunk.buffer_range.to_point(&buffer))
444 .collect::<Vec<_>>()
445 };
446
447 multibuffer.set_excerpts_for_path(
448 PathKey::for_buffer(&new_buffer, cx),
449 new_buffer.clone(),
450 hunk_ranges,
451 editor::DEFAULT_MULTIBUFFER_CONTEXT,
452 cx,
453 );
454 multibuffer.add_diff(buffer_diff.clone(), cx);
455 })
456 .log_err();
457
458 if let Some(language) = language_registry
459 .language_for_file_path(&path)
460 .await
461 .log_err()
462 {
463 new_buffer.update(cx, |buffer, cx| buffer.set_language(Some(language), cx))?;
464 }
465
466 anyhow::Ok(())
467 }
468 });
469
470 Self {
471 multibuffer,
472 path,
473 new_buffer,
474 old_buffer,
475 _task: task,
476 }
477 }
478
479 fn to_markdown(&self, cx: &App) -> String {
480 let buffer_text = self
481 .multibuffer
482 .read(cx)
483 .all_buffers()
484 .iter()
485 .map(|buffer| buffer.read(cx).text())
486 .join("\n");
487 format!("Diff: {}\n```\n{}\n```\n", self.path.display(), buffer_text)
488 }
489}
490
491#[derive(Debug, Default)]
492pub struct Plan {
493 pub entries: Vec<PlanEntry>,
494}
495
496#[derive(Debug)]
497pub struct PlanStats<'a> {
498 pub in_progress_entry: Option<&'a PlanEntry>,
499 pub pending: u32,
500 pub completed: u32,
501}
502
503impl Plan {
504 pub fn is_empty(&self) -> bool {
505 self.entries.is_empty()
506 }
507
508 pub fn stats(&self) -> PlanStats<'_> {
509 let mut stats = PlanStats {
510 in_progress_entry: None,
511 pending: 0,
512 completed: 0,
513 };
514
515 for entry in &self.entries {
516 match &entry.status {
517 acp::PlanEntryStatus::Pending => {
518 stats.pending += 1;
519 }
520 acp::PlanEntryStatus::InProgress => {
521 stats.in_progress_entry = stats.in_progress_entry.or(Some(entry));
522 }
523 acp::PlanEntryStatus::Completed => {
524 stats.completed += 1;
525 }
526 }
527 }
528
529 stats
530 }
531}
532
533#[derive(Debug)]
534pub struct PlanEntry {
535 pub content: Entity<Markdown>,
536 pub priority: acp::PlanEntryPriority,
537 pub status: acp::PlanEntryStatus,
538}
539
540impl PlanEntry {
541 pub fn from_acp(entry: acp::PlanEntry, cx: &mut App) -> Self {
542 Self {
543 content: cx.new(|cx| Markdown::new_text(entry.content.into(), cx)),
544 priority: entry.priority,
545 status: entry.status,
546 }
547 }
548}
549
550pub struct AcpThread {
551 title: SharedString,
552 entries: Vec<AgentThreadEntry>,
553 plan: Plan,
554 project: Entity<Project>,
555 action_log: Entity<ActionLog>,
556 shared_buffers: HashMap<Entity<Buffer>, BufferSnapshot>,
557 send_task: Option<Task<()>>,
558 connection: Arc<dyn AgentConnection>,
559 child_status: Option<Task<Result<()>>>,
560 session_id: acp::SessionId,
561}
562
563pub enum AcpThreadEvent {
564 NewEntry,
565 EntryUpdated(usize),
566}
567
568impl EventEmitter<AcpThreadEvent> for AcpThread {}
569
570#[derive(PartialEq, Eq)]
571pub enum ThreadStatus {
572 Idle,
573 WaitingForToolConfirmation,
574 Generating,
575}
576
577#[derive(Debug, Clone)]
578pub enum LoadError {
579 Unsupported {
580 error_message: SharedString,
581 upgrade_message: SharedString,
582 upgrade_command: String,
583 },
584 Exited(i32),
585 Other(SharedString),
586}
587
588impl Display for LoadError {
589 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
590 match self {
591 LoadError::Unsupported { error_message, .. } => write!(f, "{}", error_message),
592 LoadError::Exited(status) => write!(f, "Server exited with status {}", status),
593 LoadError::Other(msg) => write!(f, "{}", msg),
594 }
595 }
596}
597
598impl Error for LoadError {}
599
600impl AcpThread {
601 pub fn new(
602 connection: Arc<dyn AgentConnection>,
603 // todo! remove me
604 title: SharedString,
605 // todo! remove this?
606 child_status: Option<Task<Result<()>>>,
607 project: Entity<Project>,
608 session_id: acp::SessionId,
609 cx: &mut Context<Self>,
610 ) -> Self {
611 let action_log = cx.new(|_| ActionLog::new(project.clone()));
612
613 Self {
614 action_log,
615 shared_buffers: Default::default(),
616 entries: Default::default(),
617 plan: Default::default(),
618 title,
619 project,
620 send_task: None,
621 connection,
622 child_status,
623 session_id,
624 }
625 }
626
627 pub fn action_log(&self) -> &Entity<ActionLog> {
628 &self.action_log
629 }
630
631 pub fn project(&self) -> &Entity<Project> {
632 &self.project
633 }
634
635 pub fn title(&self) -> SharedString {
636 self.title.clone()
637 }
638
639 pub fn entries(&self) -> &[AgentThreadEntry] {
640 &self.entries
641 }
642
643 pub fn status(&self) -> ThreadStatus {
644 if self.send_task.is_some() {
645 if self.waiting_for_tool_confirmation() {
646 ThreadStatus::WaitingForToolConfirmation
647 } else {
648 ThreadStatus::Generating
649 }
650 } else {
651 ThreadStatus::Idle
652 }
653 }
654
655 pub fn has_pending_edit_tool_calls(&self) -> bool {
656 for entry in self.entries.iter().rev() {
657 match entry {
658 AgentThreadEntry::UserMessage(_) => return false,
659 AgentThreadEntry::ToolCall(call) if call.first_diff().is_some() => return true,
660 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
661 }
662 }
663
664 false
665 }
666
667 pub fn push_entry(&mut self, entry: AgentThreadEntry, cx: &mut Context<Self>) {
668 self.entries.push(entry);
669 cx.emit(AcpThreadEvent::NewEntry);
670 }
671
672 pub fn push_assistant_chunk(
673 &mut self,
674 chunk: acp::ContentBlock,
675 is_thought: bool,
676 cx: &mut Context<Self>,
677 ) {
678 let language_registry = self.project.read(cx).languages().clone();
679 let entries_len = self.entries.len();
680 if let Some(last_entry) = self.entries.last_mut()
681 && let AgentThreadEntry::AssistantMessage(AssistantMessage { chunks }) = last_entry
682 {
683 cx.emit(AcpThreadEvent::EntryUpdated(entries_len - 1));
684 match (chunks.last_mut(), is_thought) {
685 (Some(AssistantMessageChunk::Message { block }), false)
686 | (Some(AssistantMessageChunk::Thought { block }), true) => {
687 block.append(chunk, &language_registry, cx)
688 }
689 _ => {
690 let block = ContentBlock::new(chunk, &language_registry, cx);
691 if is_thought {
692 chunks.push(AssistantMessageChunk::Thought { block })
693 } else {
694 chunks.push(AssistantMessageChunk::Message { block })
695 }
696 }
697 }
698 } else {
699 let block = ContentBlock::new(chunk, &language_registry, cx);
700 let chunk = if is_thought {
701 AssistantMessageChunk::Thought { block }
702 } else {
703 AssistantMessageChunk::Message { block }
704 };
705
706 self.push_entry(
707 AgentThreadEntry::AssistantMessage(AssistantMessage {
708 chunks: vec![chunk],
709 }),
710 cx,
711 );
712 }
713 }
714
715 pub fn update_tool_call(
716 &mut self,
717 id: acp::ToolCallId,
718 status: acp::ToolCallStatus,
719 content: Option<Vec<acp::ToolCallContent>>,
720 cx: &mut Context<Self>,
721 ) -> Result<()> {
722 let languages = self.project.read(cx).languages().clone();
723 let (ix, current_call) = self.tool_call_mut(&id).context("Tool call not found")?;
724
725 if let Some(content) = content {
726 current_call.content = content
727 .into_iter()
728 .map(|chunk| ToolCallContent::from_acp(chunk, languages.clone(), cx))
729 .collect();
730 }
731 current_call.status = ToolCallStatus::Allowed { status };
732
733 cx.emit(AcpThreadEvent::EntryUpdated(ix));
734
735 Ok(())
736 }
737
738 /// Updates a tool call if id matches an existing entry, otherwise inserts a new one.
739 pub fn upsert_tool_call(&mut self, tool_call: acp::ToolCall, cx: &mut Context<Self>) {
740 let status = ToolCallStatus::Allowed {
741 status: tool_call.status,
742 };
743 self.upsert_tool_call_inner(tool_call, status, cx)
744 }
745
746 pub fn upsert_tool_call_inner(
747 &mut self,
748 tool_call: acp::ToolCall,
749 status: ToolCallStatus,
750 cx: &mut Context<Self>,
751 ) {
752 let language_registry = self.project.read(cx).languages().clone();
753 let call = ToolCall::from_acp(tool_call, status, language_registry, cx);
754
755 let location = call.locations.last().cloned();
756
757 if let Some((ix, current_call)) = self.tool_call_mut(&call.id) {
758 *current_call = call;
759
760 cx.emit(AcpThreadEvent::EntryUpdated(ix));
761 } else {
762 self.push_entry(AgentThreadEntry::ToolCall(call), cx);
763 }
764
765 if let Some(location) = location {
766 self.set_project_location(location, cx)
767 }
768 }
769
770 fn tool_call_mut(&mut self, id: &acp::ToolCallId) -> Option<(usize, &mut ToolCall)> {
771 // todo! use map
772 self.entries
773 .iter_mut()
774 .enumerate()
775 .rev()
776 .find_map(|(index, tool_call)| {
777 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
778 && &tool_call.id == id
779 {
780 Some((index, tool_call))
781 } else {
782 None
783 }
784 })
785 }
786
787 pub fn request_tool_call_permission(
788 &mut self,
789 tool_call: acp::ToolCall,
790 options: Vec<acp::PermissionOption>,
791 cx: &mut Context<Self>,
792 ) -> oneshot::Receiver<acp::PermissionOptionId> {
793 let (tx, rx) = oneshot::channel();
794
795 let status = ToolCallStatus::WaitingForConfirmation {
796 options,
797 respond_tx: tx,
798 };
799
800 self.upsert_tool_call_inner(tool_call, status, cx);
801 rx
802 }
803
804 pub fn authorize_tool_call(
805 &mut self,
806 id: acp::ToolCallId,
807 option: acp::PermissionOption,
808 cx: &mut Context<Self>,
809 ) {
810 let Some((ix, call)) = self.tool_call_mut(&id) else {
811 return;
812 };
813
814 let new_status = match option.kind {
815 acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways => {
816 ToolCallStatus::Rejected
817 }
818 acp::PermissionOptionKind::AllowOnce | acp::PermissionOptionKind::AllowAlways => {
819 ToolCallStatus::Allowed {
820 status: acp::ToolCallStatus::InProgress,
821 }
822 }
823 };
824
825 let curr_status = mem::replace(&mut call.status, new_status);
826
827 if let ToolCallStatus::WaitingForConfirmation { respond_tx, .. } = curr_status {
828 respond_tx.send(option.id).log_err();
829 } else if cfg!(debug_assertions) {
830 panic!("tried to authorize an already authorized tool call");
831 }
832
833 cx.emit(AcpThreadEvent::EntryUpdated(ix));
834 }
835
836 pub fn plan(&self) -> &Plan {
837 &self.plan
838 }
839
840 pub fn update_plan(&mut self, request: acp::Plan, cx: &mut Context<Self>) {
841 self.plan = Plan {
842 entries: request
843 .entries
844 .into_iter()
845 .map(|entry| PlanEntry::from_acp(entry, cx))
846 .collect(),
847 };
848
849 cx.notify();
850 }
851
852 pub fn clear_completed_plan_entries(&mut self, cx: &mut Context<Self>) {
853 self.plan
854 .entries
855 .retain(|entry| !matches!(entry.status, acp::PlanEntryStatus::Completed));
856 cx.notify();
857 }
858
859 pub fn set_project_location(&self, location: acp::ToolCallLocation, cx: &mut Context<Self>) {
860 self.project.update(cx, |project, cx| {
861 let Some(path) = project.project_path_for_absolute_path(&location.path, cx) else {
862 return;
863 };
864 let buffer = project.open_buffer(path, cx);
865 cx.spawn(async move |project, cx| {
866 let buffer = buffer.await?;
867
868 project.update(cx, |project, cx| {
869 let position = if let Some(line) = location.line {
870 let snapshot = buffer.read(cx).snapshot();
871 let point = snapshot.clip_point(Point::new(line, 0), Bias::Left);
872 snapshot.anchor_before(point)
873 } else {
874 Anchor::MIN
875 };
876
877 project.set_agent_location(
878 Some(AgentLocation {
879 buffer: buffer.downgrade(),
880 position,
881 }),
882 cx,
883 );
884 })
885 })
886 .detach_and_log_err(cx);
887 });
888 }
889
890 /// Returns true if the last turn is awaiting tool authorization
891 pub fn waiting_for_tool_confirmation(&self) -> bool {
892 for entry in self.entries.iter().rev() {
893 match &entry {
894 AgentThreadEntry::ToolCall(call) => match call.status {
895 ToolCallStatus::WaitingForConfirmation { .. } => return true,
896 ToolCallStatus::Allowed { .. }
897 | ToolCallStatus::Rejected
898 | ToolCallStatus::Canceled => continue,
899 },
900 AgentThreadEntry::UserMessage(_) | AgentThreadEntry::AssistantMessage(_) => {
901 // Reached the beginning of the turn
902 return false;
903 }
904 }
905 }
906 false
907 }
908
909 pub fn authenticate(&self, cx: &mut App) -> impl use<> + Future<Output = Result<()>> {
910 self.connection.authenticate(cx)
911 }
912
913 #[cfg(any(test, feature = "test-support"))]
914 pub fn send_raw(
915 &mut self,
916 message: &str,
917 cx: &mut Context<Self>,
918 ) -> BoxFuture<'static, Result<(), acp_old::Error>> {
919 self.send(
920 vec![acp::ContentBlock::Text(acp::TextContent {
921 text: message.to_string(),
922 annotations: None,
923 })],
924 cx,
925 )
926 }
927
928 pub fn send(
929 &mut self,
930 message: Vec<acp::ContentBlock>,
931 cx: &mut Context<Self>,
932 ) -> BoxFuture<'static, Result<(), acp_old::Error>> {
933 let block = ContentBlock::new_combined(
934 message.clone(),
935 self.project.read(cx).languages().clone(),
936 cx,
937 );
938 self.push_entry(
939 AgentThreadEntry::UserMessage(UserMessage { content: block }),
940 cx,
941 );
942
943 let (tx, rx) = oneshot::channel();
944 self.cancel(cx);
945
946 let old_send = self.send_task.take();
947 self.send_task = Some(cx.spawn(async move |this, cx| {
948 async {
949 if let Some(old_send) = old_send {
950 old_send.await;
951 }
952 let result = this
953 .update(cx, |this, cx| {
954 this.connection.prompt(
955 acp::PromptToolArguments {
956 prompt: message,
957 session_id: this.session_id.clone(),
958 },
959 cx,
960 )
961 })?
962 .await;
963 tx.send(result).log_err();
964 this.update(cx, |this, _cx| this.send_task.take())?;
965 anyhow::Ok(())
966 }
967 .await
968 .log_err();
969 }));
970
971 async move {
972 match rx.await {
973 Ok(Err(e)) => Err(e)?,
974 _ => Ok(()),
975 }
976 }
977 .boxed()
978 }
979
980 pub fn cancel(&mut self, cx: &mut Context<Self>) {
981 if self.send_task.take().is_none() {
982 return;
983 }
984 self.connection.cancel(cx);
985 for entry in self.entries.iter_mut() {
986 if let AgentThreadEntry::ToolCall(call) = entry {
987 let cancel = matches!(
988 call.status,
989 ToolCallStatus::WaitingForConfirmation { .. }
990 | ToolCallStatus::Allowed {
991 status: acp::ToolCallStatus::InProgress
992 }
993 );
994
995 if cancel {
996 call.status = ToolCallStatus::Canceled;
997 }
998 }
999 }
1000 }
1001
1002 pub fn read_text_file(
1003 &self,
1004 request: acp::ReadTextFileArguments,
1005 reuse_shared_snapshot: bool,
1006 cx: &mut Context<Self>,
1007 ) -> Task<Result<String>> {
1008 let project = self.project.clone();
1009 let action_log = self.action_log.clone();
1010 cx.spawn(async move |this, cx| {
1011 let load = project.update(cx, |project, cx| {
1012 let path = project
1013 .project_path_for_absolute_path(&request.path, cx)
1014 .context("invalid path")?;
1015 anyhow::Ok(project.open_buffer(path, cx))
1016 });
1017 let buffer = load??.await?;
1018
1019 let snapshot = if reuse_shared_snapshot {
1020 this.read_with(cx, |this, _| {
1021 this.shared_buffers.get(&buffer.clone()).cloned()
1022 })
1023 .log_err()
1024 .flatten()
1025 } else {
1026 None
1027 };
1028
1029 let snapshot = if let Some(snapshot) = snapshot {
1030 snapshot
1031 } else {
1032 action_log.update(cx, |action_log, cx| {
1033 action_log.buffer_read(buffer.clone(), cx);
1034 })?;
1035 project.update(cx, |project, cx| {
1036 let position = buffer
1037 .read(cx)
1038 .snapshot()
1039 .anchor_before(Point::new(request.line.unwrap_or_default(), 0));
1040 project.set_agent_location(
1041 Some(AgentLocation {
1042 buffer: buffer.downgrade(),
1043 position,
1044 }),
1045 cx,
1046 );
1047 })?;
1048
1049 buffer.update(cx, |buffer, _| buffer.snapshot())?
1050 };
1051
1052 this.update(cx, |this, _| {
1053 let text = snapshot.text();
1054 this.shared_buffers.insert(buffer.clone(), snapshot);
1055 if request.line.is_none() && request.limit.is_none() {
1056 return Ok(text);
1057 }
1058 let limit = request.limit.unwrap_or(u32::MAX) as usize;
1059 let Some(line) = request.line else {
1060 return Ok(text.lines().take(limit).collect::<String>());
1061 };
1062
1063 let count = text.lines().count();
1064 if count < line as usize {
1065 anyhow::bail!("There are only {} lines", count);
1066 }
1067 Ok(text
1068 .lines()
1069 .skip(line as usize + 1)
1070 .take(limit)
1071 .collect::<String>())
1072 })?
1073 })
1074 }
1075
1076 pub fn write_text_file(
1077 &self,
1078 request: acp::WriteTextFileToolArguments,
1079 cx: &mut Context<Self>,
1080 ) -> Task<Result<()>> {
1081 let project = self.project.clone();
1082 let action_log = self.action_log.clone();
1083 cx.spawn(async move |this, cx| {
1084 let load = project.update(cx, |project, cx| {
1085 let path = project
1086 .project_path_for_absolute_path(&request.path, cx)
1087 .context("invalid path")?;
1088 anyhow::Ok(project.open_buffer(path, cx))
1089 });
1090 let buffer = load??.await?;
1091 let snapshot = this.update(cx, |this, cx| {
1092 this.shared_buffers
1093 .get(&buffer)
1094 .cloned()
1095 .unwrap_or_else(|| buffer.read(cx).snapshot())
1096 })?;
1097 let edits = cx
1098 .background_executor()
1099 .spawn(async move {
1100 let old_text = snapshot.text();
1101 text_diff(old_text.as_str(), &request.content)
1102 .into_iter()
1103 .map(|(range, replacement)| {
1104 (
1105 snapshot.anchor_after(range.start)
1106 ..snapshot.anchor_before(range.end),
1107 replacement,
1108 )
1109 })
1110 .collect::<Vec<_>>()
1111 })
1112 .await;
1113 cx.update(|cx| {
1114 project.update(cx, |project, cx| {
1115 project.set_agent_location(
1116 Some(AgentLocation {
1117 buffer: buffer.downgrade(),
1118 position: edits
1119 .last()
1120 .map(|(range, _)| range.end)
1121 .unwrap_or(Anchor::MIN),
1122 }),
1123 cx,
1124 );
1125 });
1126
1127 action_log.update(cx, |action_log, cx| {
1128 action_log.buffer_read(buffer.clone(), cx);
1129 });
1130 buffer.update(cx, |buffer, cx| {
1131 buffer.edit(edits, None, cx);
1132 });
1133 action_log.update(cx, |action_log, cx| {
1134 action_log.buffer_edited(buffer.clone(), cx);
1135 });
1136 })?;
1137 project
1138 .update(cx, |project, cx| project.save_buffer(buffer, cx))?
1139 .await
1140 })
1141 }
1142
1143 pub fn child_status(&mut self) -> Option<Task<Result<()>>> {
1144 self.child_status.take()
1145 }
1146
1147 pub fn to_markdown(&self, cx: &App) -> String {
1148 self.entries.iter().map(|e| e.to_markdown(cx)).collect()
1149 }
1150}
1151
1152#[derive(Clone)]
1153pub struct OldAcpClientDelegate {
1154 thread: Rc<RefCell<WeakEntity<AcpThread>>>,
1155 cx: AsyncApp,
1156 next_tool_call_id: Rc<RefCell<u64>>,
1157 // sent_buffer_versions: HashMap<Entity<Buffer>, HashMap<u64, BufferSnapshot>>,
1158}
1159
1160impl OldAcpClientDelegate {
1161 pub fn new(thread: Rc<RefCell<WeakEntity<AcpThread>>>, cx: AsyncApp) -> Self {
1162 Self {
1163 thread,
1164 cx,
1165 next_tool_call_id: Rc::new(RefCell::new(0)),
1166 }
1167 }
1168
1169 pub async fn clear_completed_plan_entries(&self) -> Result<()> {
1170 let cx = &mut self.cx.clone();
1171 cx.update(|cx| {
1172 self.thread
1173 .borrow()
1174 .update(cx, |thread, cx| thread.clear_completed_plan_entries(cx))
1175 })?
1176 .context("Failed to update thread")?;
1177
1178 Ok(())
1179 }
1180
1181 pub async fn read_text_file_reusing_snapshot(
1182 &self,
1183 request: acp_old::ReadTextFileParams,
1184 ) -> Result<acp_old::ReadTextFileResponse, acp_old::Error> {
1185 let content = self
1186 .cx
1187 .update(|cx| {
1188 self.thread.borrow().update(cx, |thread, cx| {
1189 thread.read_text_file(
1190 acp::ReadTextFileArguments {
1191 path: request.path,
1192 line: request.line,
1193 limit: request.limit,
1194 },
1195 true,
1196 cx,
1197 )
1198 })
1199 })?
1200 .context("Failed to update thread")?
1201 .await?;
1202 Ok(acp_old::ReadTextFileResponse { content })
1203 }
1204}
1205impl acp_old::Client for OldAcpClientDelegate {
1206 async fn stream_assistant_message_chunk(
1207 &self,
1208 params: acp_old::StreamAssistantMessageChunkParams,
1209 ) -> Result<(), acp_old::Error> {
1210 let cx = &mut self.cx.clone();
1211
1212 cx.update(|cx| {
1213 self.thread
1214 .borrow()
1215 .update(cx, |thread, cx| match params.chunk {
1216 acp_old::AssistantMessageChunk::Text { text } => {
1217 thread.push_assistant_chunk(text.into(), false, cx)
1218 }
1219 acp_old::AssistantMessageChunk::Thought { thought } => {
1220 thread.push_assistant_chunk(thought.into(), true, cx)
1221 }
1222 })
1223 .ok();
1224 })?;
1225
1226 Ok(())
1227 }
1228
1229 async fn request_tool_call_confirmation(
1230 &self,
1231 request: acp_old::RequestToolCallConfirmationParams,
1232 ) -> Result<acp_old::RequestToolCallConfirmationResponse, acp_old::Error> {
1233 let cx = &mut self.cx.clone();
1234
1235 let old_acp_id = *self.next_tool_call_id.borrow() + 1;
1236 self.next_tool_call_id.replace(old_acp_id);
1237
1238 let tool_call = into_new_tool_call(
1239 acp::ToolCallId(old_acp_id.to_string().into()),
1240 request.tool_call,
1241 );
1242
1243 let mut options = match request.confirmation {
1244 acp_old::ToolCallConfirmation::Edit { .. } => vec![(
1245 acp_old::ToolCallConfirmationOutcome::AlwaysAllow,
1246 acp::PermissionOptionKind::AllowAlways,
1247 "Always Allow Edits".to_string(),
1248 )],
1249 acp_old::ToolCallConfirmation::Execute { root_command, .. } => vec![(
1250 acp_old::ToolCallConfirmationOutcome::AlwaysAllow,
1251 acp::PermissionOptionKind::AllowAlways,
1252 format!("Always Allow {}", root_command),
1253 )],
1254 acp_old::ToolCallConfirmation::Mcp {
1255 server_name,
1256 tool_name,
1257 ..
1258 } => vec![
1259 (
1260 acp_old::ToolCallConfirmationOutcome::AlwaysAllowMcpServer,
1261 acp::PermissionOptionKind::AllowAlways,
1262 format!("Always Allow {}", server_name),
1263 ),
1264 (
1265 acp_old::ToolCallConfirmationOutcome::AlwaysAllowTool,
1266 acp::PermissionOptionKind::AllowAlways,
1267 format!("Always Allow {}", tool_name),
1268 ),
1269 ],
1270 acp_old::ToolCallConfirmation::Fetch { .. } => vec![(
1271 acp_old::ToolCallConfirmationOutcome::AlwaysAllow,
1272 acp::PermissionOptionKind::AllowAlways,
1273 "Always Allow".to_string(),
1274 )],
1275 acp_old::ToolCallConfirmation::Other { .. } => vec![(
1276 acp_old::ToolCallConfirmationOutcome::AlwaysAllow,
1277 acp::PermissionOptionKind::AllowAlways,
1278 "Always Allow".to_string(),
1279 )],
1280 };
1281
1282 options.extend([
1283 (
1284 acp_old::ToolCallConfirmationOutcome::Allow,
1285 acp::PermissionOptionKind::AllowOnce,
1286 "Allow".to_string(),
1287 ),
1288 (
1289 acp_old::ToolCallConfirmationOutcome::Reject,
1290 acp::PermissionOptionKind::RejectOnce,
1291 "Reject".to_string(),
1292 ),
1293 ]);
1294
1295 let mut outcomes = Vec::with_capacity(options.len());
1296 let mut acp_options = Vec::with_capacity(options.len());
1297
1298 for (index, (outcome, kind, label)) in options.into_iter().enumerate() {
1299 outcomes.push(outcome);
1300 acp_options.push(acp::PermissionOption {
1301 id: acp::PermissionOptionId(index.to_string().into()),
1302 label,
1303 kind,
1304 })
1305 }
1306
1307 let response = cx
1308 .update(|cx| {
1309 self.thread.borrow().update(cx, |thread, cx| {
1310 thread.request_tool_call_permission(tool_call, acp_options, cx)
1311 })
1312 })?
1313 .context("Failed to update thread")?
1314 .await;
1315
1316 let outcome = match response {
1317 Ok(option_id) => outcomes[option_id.0.parse::<usize>().unwrap_or(0)].clone(),
1318 Err(oneshot::Canceled) => acp_old::ToolCallConfirmationOutcome::Cancel,
1319 };
1320
1321 Ok(acp_old::RequestToolCallConfirmationResponse {
1322 id: acp_old::ToolCallId(old_acp_id),
1323 outcome: outcome,
1324 })
1325 }
1326
1327 async fn push_tool_call(
1328 &self,
1329 request: acp_old::PushToolCallParams,
1330 ) -> Result<acp_old::PushToolCallResponse, acp_old::Error> {
1331 let cx = &mut self.cx.clone();
1332
1333 let old_acp_id = *self.next_tool_call_id.borrow() + 1;
1334 self.next_tool_call_id.replace(old_acp_id);
1335
1336 cx.update(|cx| {
1337 self.thread.borrow().update(cx, |thread, cx| {
1338 thread.upsert_tool_call(
1339 into_new_tool_call(acp::ToolCallId(old_acp_id.to_string().into()), request),
1340 cx,
1341 )
1342 })
1343 })?
1344 .context("Failed to update thread")?;
1345
1346 Ok(acp_old::PushToolCallResponse {
1347 id: acp_old::ToolCallId(old_acp_id),
1348 })
1349 }
1350
1351 async fn update_tool_call(
1352 &self,
1353 request: acp_old::UpdateToolCallParams,
1354 ) -> Result<(), acp_old::Error> {
1355 let cx = &mut self.cx.clone();
1356
1357 cx.update(|cx| {
1358 self.thread.borrow().update(cx, |thread, cx| {
1359 let languages = thread.project.read(cx).languages().clone();
1360
1361 if let Some((ix, tool_call)) = thread
1362 .tool_call_mut(&acp::ToolCallId(request.tool_call_id.0.to_string().into()))
1363 {
1364 tool_call.status = ToolCallStatus::Allowed {
1365 status: into_new_tool_call_status(request.status),
1366 };
1367 tool_call.content = request
1368 .content
1369 .into_iter()
1370 .map(|content| {
1371 ToolCallContent::from_acp(
1372 into_new_tool_call_content(content),
1373 languages.clone(),
1374 cx,
1375 )
1376 })
1377 .collect();
1378
1379 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1380 anyhow::Ok(())
1381 } else {
1382 anyhow::bail!("Tool call not found")
1383 }
1384 })
1385 })?
1386 .context("Failed to update thread")??;
1387
1388 Ok(())
1389 }
1390
1391 async fn update_plan(&self, request: acp_old::UpdatePlanParams) -> Result<(), acp_old::Error> {
1392 let cx = &mut self.cx.clone();
1393
1394 cx.update(|cx| {
1395 self.thread.borrow().update(cx, |thread, cx| {
1396 thread.update_plan(
1397 acp::Plan {
1398 entries: request
1399 .entries
1400 .into_iter()
1401 .map(into_new_plan_entry)
1402 .collect(),
1403 },
1404 cx,
1405 )
1406 })
1407 })?
1408 .context("Failed to update thread")?;
1409
1410 Ok(())
1411 }
1412
1413 async fn read_text_file(
1414 &self,
1415 request: acp_old::ReadTextFileParams,
1416 ) -> Result<acp_old::ReadTextFileResponse, acp_old::Error> {
1417 let content = self
1418 .cx
1419 .update(|cx| {
1420 self.thread.borrow().update(cx, |thread, cx| {
1421 thread.read_text_file(
1422 acp::ReadTextFileArguments {
1423 path: request.path,
1424 line: request.line,
1425 limit: request.limit,
1426 },
1427 false,
1428 cx,
1429 )
1430 })
1431 })?
1432 .context("Failed to update thread")?
1433 .await?;
1434 Ok(acp_old::ReadTextFileResponse { content })
1435 }
1436
1437 async fn write_text_file(
1438 &self,
1439 request: acp_old::WriteTextFileParams,
1440 ) -> Result<(), acp_old::Error> {
1441 self.cx
1442 .update(|cx| {
1443 self.thread.borrow().update(cx, |thread, cx| {
1444 thread.write_text_file(
1445 acp::WriteTextFileToolArguments {
1446 path: request.path,
1447 content: request.content,
1448 },
1449 cx,
1450 )
1451 })
1452 })?
1453 .context("Failed to update thread")?
1454 .await?;
1455
1456 Ok(())
1457 }
1458}
1459
1460fn into_new_tool_call(id: acp::ToolCallId, request: acp_old::PushToolCallParams) -> acp::ToolCall {
1461 acp::ToolCall {
1462 id: id,
1463 label: request.label,
1464 kind: acp_kind_from_old_icon(request.icon),
1465 status: acp::ToolCallStatus::InProgress,
1466 content: request
1467 .content
1468 .into_iter()
1469 .map(into_new_tool_call_content)
1470 .collect(),
1471 locations: request
1472 .locations
1473 .into_iter()
1474 .map(into_new_tool_call_location)
1475 .collect(),
1476 }
1477}
1478
1479fn acp_kind_from_old_icon(icon: acp_old::Icon) -> acp::ToolKind {
1480 match icon {
1481 acp_old::Icon::FileSearch => acp::ToolKind::Search,
1482 acp_old::Icon::Folder => acp::ToolKind::Search,
1483 acp_old::Icon::Globe => acp::ToolKind::Search,
1484 acp_old::Icon::Hammer => acp::ToolKind::Other,
1485 acp_old::Icon::LightBulb => acp::ToolKind::Think,
1486 acp_old::Icon::Pencil => acp::ToolKind::Edit,
1487 acp_old::Icon::Regex => acp::ToolKind::Search,
1488 acp_old::Icon::Terminal => acp::ToolKind::Execute,
1489 }
1490}
1491
1492fn into_new_tool_call_status(status: acp_old::ToolCallStatus) -> acp::ToolCallStatus {
1493 match status {
1494 acp_old::ToolCallStatus::Running => acp::ToolCallStatus::InProgress,
1495 acp_old::ToolCallStatus::Finished => acp::ToolCallStatus::Completed,
1496 acp_old::ToolCallStatus::Error => acp::ToolCallStatus::Failed,
1497 }
1498}
1499
1500fn into_new_tool_call_content(content: acp_old::ToolCallContent) -> acp::ToolCallContent {
1501 match content {
1502 acp_old::ToolCallContent::Markdown { markdown } => acp::ToolCallContent::ContentBlock {
1503 content: acp::ContentBlock::Text(acp::TextContent {
1504 annotations: None,
1505 text: markdown,
1506 }),
1507 },
1508 acp_old::ToolCallContent::Diff { diff } => acp::ToolCallContent::Diff {
1509 diff: into_new_diff(diff),
1510 },
1511 }
1512}
1513
1514fn into_new_diff(diff: acp_old::Diff) -> acp::Diff {
1515 acp::Diff {
1516 path: diff.path,
1517 old_text: diff.old_text,
1518 new_text: diff.new_text,
1519 }
1520}
1521
1522fn into_new_tool_call_location(location: acp_old::ToolCallLocation) -> acp::ToolCallLocation {
1523 acp::ToolCallLocation {
1524 path: location.path,
1525 line: location.line,
1526 }
1527}
1528
1529fn into_new_plan(request: acp_old::UpdatePlanParams) -> acp::Plan {
1530 acp::Plan {
1531 entries: request
1532 .entries
1533 .into_iter()
1534 .map(into_new_plan_entry)
1535 .collect(),
1536 }
1537}
1538
1539fn into_new_plan_entry(entry: acp_old::PlanEntry) -> acp::PlanEntry {
1540 acp::PlanEntry {
1541 content: entry.content,
1542 priority: into_new_plan_priority(entry.priority),
1543 status: into_new_plan_status(entry.status),
1544 }
1545}
1546
1547fn into_new_plan_priority(priority: acp_old::PlanEntryPriority) -> acp::PlanEntryPriority {
1548 match priority {
1549 acp_old::PlanEntryPriority::Low => acp::PlanEntryPriority::Low,
1550 acp_old::PlanEntryPriority::Medium => acp::PlanEntryPriority::Medium,
1551 acp_old::PlanEntryPriority::High => acp::PlanEntryPriority::High,
1552 }
1553}
1554
1555fn into_new_plan_status(status: acp_old::PlanEntryStatus) -> acp::PlanEntryStatus {
1556 match status {
1557 acp_old::PlanEntryStatus::Pending => acp::PlanEntryStatus::Pending,
1558 acp_old::PlanEntryStatus::InProgress => acp::PlanEntryStatus::InProgress,
1559 acp_old::PlanEntryStatus::Completed => acp::PlanEntryStatus::Completed,
1560 }
1561}
1562
1563#[cfg(test)]
1564mod tests {
1565 use super::*;
1566 use anyhow::anyhow;
1567 use async_pipe::{PipeReader, PipeWriter};
1568 use futures::{channel::mpsc, future::LocalBoxFuture, select};
1569 use gpui::{AsyncApp, TestAppContext};
1570 use indoc::indoc;
1571 use project::FakeFs;
1572 use serde_json::json;
1573 use settings::SettingsStore;
1574 use smol::{future::BoxedLocal, stream::StreamExt as _};
1575 use std::{cell::RefCell, rc::Rc, time::Duration};
1576 use util::path;
1577
1578 fn init_test(cx: &mut TestAppContext) {
1579 env_logger::try_init().ok();
1580 cx.update(|cx| {
1581 let settings_store = SettingsStore::test(cx);
1582 cx.set_global(settings_store);
1583 Project::init_settings(cx);
1584 language::init(cx);
1585 });
1586 }
1587
1588 #[gpui::test]
1589 async fn test_thinking_concatenation(cx: &mut TestAppContext) {
1590 init_test(cx);
1591
1592 let fs = FakeFs::new(cx.executor());
1593 let project = Project::test(fs, [], cx).await;
1594 let (thread, fake_server) = fake_acp_thread(project, cx);
1595
1596 fake_server.update(cx, |fake_server, _| {
1597 fake_server.on_user_message(move |_, server, mut cx| async move {
1598 server
1599 .update(&mut cx, |server, _| {
1600 server.send_to_zed(acp_old::StreamAssistantMessageChunkParams {
1601 chunk: acp_old::AssistantMessageChunk::Thought {
1602 thought: "Thinking ".into(),
1603 },
1604 })
1605 })?
1606 .await
1607 .unwrap();
1608 server
1609 .update(&mut cx, |server, _| {
1610 server.send_to_zed(acp_old::StreamAssistantMessageChunkParams {
1611 chunk: acp_old::AssistantMessageChunk::Thought {
1612 thought: "hard!".into(),
1613 },
1614 })
1615 })?
1616 .await
1617 .unwrap();
1618
1619 Ok(())
1620 })
1621 });
1622
1623 thread
1624 .update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
1625 .await
1626 .unwrap();
1627
1628 let output = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
1629 assert_eq!(
1630 output,
1631 indoc! {r#"
1632 ## User
1633
1634 Hello from Zed!
1635
1636 ## Assistant
1637
1638 <thinking>
1639 Thinking hard!
1640 </thinking>
1641
1642 "#}
1643 );
1644 }
1645
1646 #[gpui::test]
1647 async fn test_edits_concurrently_to_user(cx: &mut TestAppContext) {
1648 init_test(cx);
1649
1650 let fs = FakeFs::new(cx.executor());
1651 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\n"}))
1652 .await;
1653 let project = Project::test(fs.clone(), [], cx).await;
1654 let (thread, fake_server) = fake_acp_thread(project.clone(), cx);
1655 let (worktree, pathbuf) = project
1656 .update(cx, |project, cx| {
1657 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
1658 })
1659 .await
1660 .unwrap();
1661 let buffer = project
1662 .update(cx, |project, cx| {
1663 project.open_buffer((worktree.read(cx).id(), pathbuf), cx)
1664 })
1665 .await
1666 .unwrap();
1667
1668 let (read_file_tx, read_file_rx) = oneshot::channel::<()>();
1669 let read_file_tx = Rc::new(RefCell::new(Some(read_file_tx)));
1670
1671 fake_server.update(cx, |fake_server, _| {
1672 fake_server.on_user_message(move |_, server, mut cx| {
1673 let read_file_tx = read_file_tx.clone();
1674 async move {
1675 let content = server
1676 .update(&mut cx, |server, _| {
1677 server.send_to_zed(acp_old::ReadTextFileParams {
1678 path: path!("/tmp/foo").into(),
1679 line: None,
1680 limit: None,
1681 })
1682 })?
1683 .await
1684 .unwrap();
1685 assert_eq!(content.content, "one\ntwo\nthree\n");
1686 read_file_tx.take().unwrap().send(()).unwrap();
1687 server
1688 .update(&mut cx, |server, _| {
1689 server.send_to_zed(acp_old::WriteTextFileParams {
1690 path: path!("/tmp/foo").into(),
1691 content: "one\ntwo\nthree\nfour\nfive\n".to_string(),
1692 })
1693 })?
1694 .await
1695 .unwrap();
1696 Ok(())
1697 }
1698 })
1699 });
1700
1701 let request = thread.update(cx, |thread, cx| {
1702 thread.send_raw("Extend the count in /tmp/foo", cx)
1703 });
1704 read_file_rx.await.ok();
1705 buffer.update(cx, |buffer, cx| {
1706 buffer.edit([(0..0, "zero\n".to_string())], None, cx);
1707 });
1708 cx.run_until_parked();
1709 assert_eq!(
1710 buffer.read_with(cx, |buffer, _| buffer.text()),
1711 "zero\none\ntwo\nthree\nfour\nfive\n"
1712 );
1713 assert_eq!(
1714 String::from_utf8(fs.read_file_sync(path!("/tmp/foo")).unwrap()).unwrap(),
1715 "zero\none\ntwo\nthree\nfour\nfive\n"
1716 );
1717 request.await.unwrap();
1718 }
1719
1720 #[gpui::test]
1721 async fn test_succeeding_canceled_toolcall(cx: &mut TestAppContext) {
1722 init_test(cx);
1723
1724 let fs = FakeFs::new(cx.executor());
1725 let project = Project::test(fs, [], cx).await;
1726 let (thread, fake_server) = fake_acp_thread(project, cx);
1727
1728 let (end_turn_tx, end_turn_rx) = oneshot::channel::<()>();
1729
1730 let tool_call_id = Rc::new(RefCell::new(None));
1731 let end_turn_rx = Rc::new(RefCell::new(Some(end_turn_rx)));
1732 fake_server.update(cx, |fake_server, _| {
1733 let tool_call_id = tool_call_id.clone();
1734 fake_server.on_user_message(move |_, server, mut cx| {
1735 let end_turn_rx = end_turn_rx.clone();
1736 let tool_call_id = tool_call_id.clone();
1737 async move {
1738 let tool_call_result = server
1739 .update(&mut cx, |server, _| {
1740 server.send_to_zed(acp_old::PushToolCallParams {
1741 label: "Fetch".to_string(),
1742 icon: acp_old::Icon::Globe,
1743 content: None,
1744 locations: vec![],
1745 })
1746 })?
1747 .await
1748 .unwrap();
1749 *tool_call_id.clone().borrow_mut() = Some(tool_call_result.id);
1750 end_turn_rx.take().unwrap().await.ok();
1751
1752 Ok(())
1753 }
1754 })
1755 });
1756
1757 let request = thread.update(cx, |thread, cx| {
1758 thread.send_raw("Fetch https://example.com", cx)
1759 });
1760
1761 run_until_first_tool_call(&thread, cx).await;
1762
1763 thread.read_with(cx, |thread, _| {
1764 assert!(matches!(
1765 thread.entries[1],
1766 AgentThreadEntry::ToolCall(ToolCall {
1767 status: ToolCallStatus::Allowed {
1768 status: acp::ToolCallStatus::InProgress,
1769 ..
1770 },
1771 ..
1772 })
1773 ));
1774 });
1775
1776 cx.run_until_parked();
1777
1778 thread.update(cx, |thread, cx| thread.cancel(cx));
1779
1780 thread.read_with(cx, |thread, _| {
1781 assert!(matches!(
1782 &thread.entries[1],
1783 AgentThreadEntry::ToolCall(ToolCall {
1784 status: ToolCallStatus::Canceled,
1785 ..
1786 })
1787 ));
1788 });
1789
1790 fake_server
1791 .update(cx, |fake_server, _| {
1792 fake_server.send_to_zed(acp_old::UpdateToolCallParams {
1793 tool_call_id: tool_call_id.borrow().unwrap(),
1794 status: acp_old::ToolCallStatus::Finished,
1795 content: None,
1796 })
1797 })
1798 .await
1799 .unwrap();
1800
1801 drop(end_turn_tx);
1802 request.await.unwrap();
1803
1804 thread.read_with(cx, |thread, _| {
1805 assert!(matches!(
1806 thread.entries[1],
1807 AgentThreadEntry::ToolCall(ToolCall {
1808 status: ToolCallStatus::Allowed {
1809 status: acp::ToolCallStatus::Completed,
1810 ..
1811 },
1812 ..
1813 })
1814 ));
1815 });
1816 }
1817
1818 async fn run_until_first_tool_call(
1819 thread: &Entity<AcpThread>,
1820 cx: &mut TestAppContext,
1821 ) -> usize {
1822 let (mut tx, mut rx) = mpsc::channel::<usize>(1);
1823
1824 let subscription = cx.update(|cx| {
1825 cx.subscribe(thread, move |thread, _, cx| {
1826 for (ix, entry) in thread.read(cx).entries.iter().enumerate() {
1827 if matches!(entry, AgentThreadEntry::ToolCall(_)) {
1828 return tx.try_send(ix).unwrap();
1829 }
1830 }
1831 })
1832 });
1833
1834 select! {
1835 _ = futures::FutureExt::fuse(smol::Timer::after(Duration::from_secs(10))) => {
1836 panic!("Timeout waiting for tool call")
1837 }
1838 ix = rx.next().fuse() => {
1839 drop(subscription);
1840 ix.unwrap()
1841 }
1842 }
1843 }
1844
1845 pub fn fake_acp_thread(
1846 project: Entity<Project>,
1847 cx: &mut TestAppContext,
1848 ) -> (Entity<AcpThread>, Entity<FakeAcpServer>) {
1849 let (stdin_tx, stdin_rx) = async_pipe::pipe();
1850 let (stdout_tx, stdout_rx) = async_pipe::pipe();
1851
1852 let thread = cx.new(|cx| {
1853 let foreground_executor = cx.foreground_executor().clone();
1854 let thread_rc = Rc::new(RefCell::new(cx.entity().downgrade()));
1855
1856 let (connection, io_fut) = acp_old::AgentConnection::connect_to_agent(
1857 OldAcpClientDelegate::new(thread_rc.clone(), cx.to_async()),
1858 stdin_tx,
1859 stdout_rx,
1860 move |fut| {
1861 foreground_executor.spawn(fut).detach();
1862 },
1863 );
1864
1865 let io_task = cx.background_spawn({
1866 async move {
1867 io_fut.await.log_err();
1868 Ok(())
1869 }
1870 });
1871 let connection = OldAcpAgentConnection {
1872 connection,
1873 child_status: io_task,
1874 thread: thread_rc,
1875 };
1876
1877 AcpThread::new(
1878 Arc::new(connection),
1879 "Test".into(),
1880 None,
1881 project,
1882 acp::SessionId("test".into()),
1883 cx,
1884 )
1885 });
1886 let agent = cx.update(|cx| cx.new(|cx| FakeAcpServer::new(stdin_rx, stdout_tx, cx)));
1887 (thread, agent)
1888 }
1889
1890 pub struct FakeAcpServer {
1891 connection: acp_old::ClientConnection,
1892
1893 _io_task: Task<()>,
1894 on_user_message: Option<
1895 Rc<
1896 dyn Fn(
1897 acp_old::SendUserMessageParams,
1898 Entity<FakeAcpServer>,
1899 AsyncApp,
1900 ) -> LocalBoxFuture<'static, Result<(), acp_old::Error>>,
1901 >,
1902 >,
1903 }
1904
1905 #[derive(Clone)]
1906 struct FakeAgent {
1907 server: Entity<FakeAcpServer>,
1908 cx: AsyncApp,
1909 }
1910
1911 impl acp_old::Agent for FakeAgent {
1912 async fn initialize(
1913 &self,
1914 params: acp_old::InitializeParams,
1915 ) -> Result<acp_old::InitializeResponse, acp_old::Error> {
1916 Ok(acp_old::InitializeResponse {
1917 protocol_version: params.protocol_version,
1918 is_authenticated: true,
1919 })
1920 }
1921
1922 async fn authenticate(&self) -> Result<(), acp_old::Error> {
1923 Ok(())
1924 }
1925
1926 async fn cancel_send_message(&self) -> Result<(), acp_old::Error> {
1927 Ok(())
1928 }
1929
1930 async fn send_user_message(
1931 &self,
1932 request: acp_old::SendUserMessageParams,
1933 ) -> Result<(), acp_old::Error> {
1934 let mut cx = self.cx.clone();
1935 let handler = self
1936 .server
1937 .update(&mut cx, |server, _| server.on_user_message.clone())
1938 .ok()
1939 .flatten();
1940 if let Some(handler) = handler {
1941 handler(request, self.server.clone(), self.cx.clone()).await
1942 } else {
1943 Err(anyhow::anyhow!("No handler for on_user_message").into())
1944 }
1945 }
1946 }
1947
1948 impl FakeAcpServer {
1949 fn new(stdin: PipeReader, stdout: PipeWriter, cx: &Context<Self>) -> Self {
1950 let agent = FakeAgent {
1951 server: cx.entity(),
1952 cx: cx.to_async(),
1953 };
1954 let foreground_executor = cx.foreground_executor().clone();
1955
1956 let (connection, io_fut) = acp_old::ClientConnection::connect_to_client(
1957 agent.clone(),
1958 stdout,
1959 stdin,
1960 move |fut| {
1961 foreground_executor.spawn(fut).detach();
1962 },
1963 );
1964 FakeAcpServer {
1965 connection: connection,
1966 on_user_message: None,
1967 _io_task: cx.background_spawn(async move {
1968 io_fut.await.log_err();
1969 }),
1970 }
1971 }
1972
1973 fn on_user_message<F>(
1974 &mut self,
1975 handler: impl for<'a> Fn(
1976 acp_old::SendUserMessageParams,
1977 Entity<FakeAcpServer>,
1978 AsyncApp,
1979 ) -> F
1980 + 'static,
1981 ) where
1982 F: Future<Output = Result<(), acp_old::Error>> + 'static,
1983 {
1984 self.on_user_message
1985 .replace(Rc::new(move |request, server, cx| {
1986 handler(request, server, cx).boxed_local()
1987 }));
1988 }
1989
1990 fn send_to_zed<T: acp_old::ClientRequest + 'static>(
1991 &self,
1992 message: T,
1993 ) -> BoxedLocal<Result<T::Response>> {
1994 self.connection
1995 .request(message)
1996 .map(|f| f.map_err(|err| anyhow!(err)))
1997 .boxed_local()
1998 }
1999 }
2000}