1mod connection;
2mod diff;
3mod mention;
4mod terminal;
5
6use agent_settings::AgentSettings;
7use collections::HashSet;
8pub use connection::*;
9pub use diff::*;
10use futures::future::Shared;
11use language::language_settings::FormatOnSave;
12pub use mention::*;
13use project::lsp_store::{FormatTrigger, LspFormatTarget};
14use serde::{Deserialize, Serialize};
15use settings::Settings as _;
16pub use terminal::*;
17
18use action_log::ActionLog;
19use agent_client_protocol::{self as acp};
20use anyhow::{Context as _, Result, anyhow};
21use editor::Bias;
22use futures::{FutureExt, channel::oneshot, future::BoxFuture};
23use gpui::{AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Task, WeakEntity};
24use itertools::Itertools;
25use language::{Anchor, Buffer, BufferSnapshot, LanguageRegistry, Point, ToPoint, text_diff};
26use markdown::Markdown;
27use project::{AgentLocation, Project, git_store::GitStoreCheckpoint};
28use std::collections::HashMap;
29use std::error::Error;
30use std::fmt::{Formatter, Write};
31use std::ops::Range;
32use std::process::ExitStatus;
33use std::rc::Rc;
34use std::time::{Duration, Instant};
35use std::{fmt::Display, mem, path::PathBuf, sync::Arc};
36use ui::App;
37use util::{ResultExt, get_system_shell};
38use uuid::Uuid;
39
40#[derive(Debug)]
41pub struct UserMessage {
42 pub id: Option<UserMessageId>,
43 pub content: ContentBlock,
44 pub chunks: Vec<acp::ContentBlock>,
45 pub checkpoint: Option<Checkpoint>,
46}
47
48#[derive(Debug)]
49pub struct Checkpoint {
50 git_checkpoint: GitStoreCheckpoint,
51 pub show: bool,
52}
53
54impl UserMessage {
55 fn to_markdown(&self, cx: &App) -> String {
56 let mut markdown = String::new();
57 if self
58 .checkpoint
59 .as_ref()
60 .is_some_and(|checkpoint| checkpoint.show)
61 {
62 writeln!(markdown, "## User (checkpoint)").unwrap();
63 } else {
64 writeln!(markdown, "## User").unwrap();
65 }
66 writeln!(markdown).unwrap();
67 writeln!(markdown, "{}", self.content.to_markdown(cx)).unwrap();
68 writeln!(markdown).unwrap();
69 markdown
70 }
71}
72
73#[derive(Debug, PartialEq)]
74pub struct AssistantMessage {
75 pub chunks: Vec<AssistantMessageChunk>,
76}
77
78impl AssistantMessage {
79 pub fn to_markdown(&self, cx: &App) -> String {
80 format!(
81 "## Assistant\n\n{}\n\n",
82 self.chunks
83 .iter()
84 .map(|chunk| chunk.to_markdown(cx))
85 .join("\n\n")
86 )
87 }
88}
89
90#[derive(Debug, PartialEq)]
91pub enum AssistantMessageChunk {
92 Message { block: ContentBlock },
93 Thought { block: ContentBlock },
94}
95
96impl AssistantMessageChunk {
97 pub fn from_str(chunk: &str, language_registry: &Arc<LanguageRegistry>, cx: &mut App) -> Self {
98 Self::Message {
99 block: ContentBlock::new(chunk.into(), language_registry, cx),
100 }
101 }
102
103 fn to_markdown(&self, cx: &App) -> String {
104 match self {
105 Self::Message { block } => block.to_markdown(cx).to_string(),
106 Self::Thought { block } => {
107 format!("<thinking>\n{}\n</thinking>", block.to_markdown(cx))
108 }
109 }
110 }
111}
112
113#[derive(Debug)]
114pub enum AgentThreadEntry {
115 UserMessage(UserMessage),
116 AssistantMessage(AssistantMessage),
117 ToolCall(ToolCall),
118}
119
120impl AgentThreadEntry {
121 pub fn to_markdown(&self, cx: &App) -> String {
122 match self {
123 Self::UserMessage(message) => message.to_markdown(cx),
124 Self::AssistantMessage(message) => message.to_markdown(cx),
125 Self::ToolCall(tool_call) => tool_call.to_markdown(cx),
126 }
127 }
128
129 pub fn user_message(&self) -> Option<&UserMessage> {
130 if let AgentThreadEntry::UserMessage(message) = self {
131 Some(message)
132 } else {
133 None
134 }
135 }
136
137 pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
138 if let AgentThreadEntry::ToolCall(call) = self {
139 itertools::Either::Left(call.diffs())
140 } else {
141 itertools::Either::Right(std::iter::empty())
142 }
143 }
144
145 pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
146 if let AgentThreadEntry::ToolCall(call) = self {
147 itertools::Either::Left(call.terminals())
148 } else {
149 itertools::Either::Right(std::iter::empty())
150 }
151 }
152
153 pub fn location(&self, ix: usize) -> Option<(acp::ToolCallLocation, AgentLocation)> {
154 if let AgentThreadEntry::ToolCall(ToolCall {
155 locations,
156 resolved_locations,
157 ..
158 }) = self
159 {
160 Some((
161 locations.get(ix)?.clone(),
162 resolved_locations.get(ix)?.clone()?,
163 ))
164 } else {
165 None
166 }
167 }
168}
169
170#[derive(Debug)]
171pub struct ToolCall {
172 pub id: acp::ToolCallId,
173 pub label: Entity<Markdown>,
174 pub kind: acp::ToolKind,
175 pub content: Vec<ToolCallContent>,
176 pub status: ToolCallStatus,
177 pub locations: Vec<acp::ToolCallLocation>,
178 pub resolved_locations: Vec<Option<AgentLocation>>,
179 pub raw_input: Option<serde_json::Value>,
180 pub raw_output: Option<serde_json::Value>,
181}
182
183impl ToolCall {
184 fn from_acp(
185 tool_call: acp::ToolCall,
186 status: ToolCallStatus,
187 language_registry: Arc<LanguageRegistry>,
188 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
189 cx: &mut App,
190 ) -> Result<Self> {
191 let title = if let Some((first_line, _)) = tool_call.title.split_once("\n") {
192 first_line.to_owned() + "…"
193 } else {
194 tool_call.title
195 };
196 let mut content = Vec::with_capacity(tool_call.content.len());
197 for item in tool_call.content {
198 content.push(ToolCallContent::from_acp(
199 item,
200 language_registry.clone(),
201 terminals,
202 cx,
203 )?);
204 }
205
206 let result = Self {
207 id: tool_call.id,
208 label: cx
209 .new(|cx| Markdown::new(title.into(), Some(language_registry.clone()), None, cx)),
210 kind: tool_call.kind,
211 content,
212 locations: tool_call.locations,
213 resolved_locations: Vec::default(),
214 status,
215 raw_input: tool_call.raw_input,
216 raw_output: tool_call.raw_output,
217 };
218 Ok(result)
219 }
220
221 fn update_fields(
222 &mut self,
223 fields: acp::ToolCallUpdateFields,
224 language_registry: Arc<LanguageRegistry>,
225 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
226 cx: &mut App,
227 ) -> Result<()> {
228 let acp::ToolCallUpdateFields {
229 kind,
230 status,
231 title,
232 content,
233 locations,
234 raw_input,
235 raw_output,
236 } = fields;
237
238 if let Some(kind) = kind {
239 self.kind = kind;
240 }
241
242 if let Some(status) = status {
243 self.status = status.into();
244 }
245
246 if let Some(title) = title {
247 self.label.update(cx, |label, cx| {
248 if let Some((first_line, _)) = title.split_once("\n") {
249 label.replace(first_line.to_owned() + "…", cx)
250 } else {
251 label.replace(title, cx);
252 }
253 });
254 }
255
256 if let Some(content) = content {
257 let new_content_len = content.len();
258 let mut content = content.into_iter();
259
260 // Reuse existing content if we can
261 for (old, new) in self.content.iter_mut().zip(content.by_ref()) {
262 old.update_from_acp(new, language_registry.clone(), terminals, cx)?;
263 }
264 for new in content {
265 self.content.push(ToolCallContent::from_acp(
266 new,
267 language_registry.clone(),
268 terminals,
269 cx,
270 )?)
271 }
272 self.content.truncate(new_content_len);
273 }
274
275 if let Some(locations) = locations {
276 self.locations = locations;
277 }
278
279 if let Some(raw_input) = raw_input {
280 self.raw_input = Some(raw_input);
281 }
282
283 if let Some(raw_output) = raw_output {
284 if self.content.is_empty()
285 && let Some(markdown) = markdown_for_raw_output(&raw_output, &language_registry, cx)
286 {
287 self.content
288 .push(ToolCallContent::ContentBlock(ContentBlock::Markdown {
289 markdown,
290 }));
291 }
292 self.raw_output = Some(raw_output);
293 }
294 Ok(())
295 }
296
297 pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
298 self.content.iter().filter_map(|content| match content {
299 ToolCallContent::Diff(diff) => Some(diff),
300 ToolCallContent::ContentBlock(_) => None,
301 ToolCallContent::Terminal(_) => None,
302 })
303 }
304
305 pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
306 self.content.iter().filter_map(|content| match content {
307 ToolCallContent::Terminal(terminal) => Some(terminal),
308 ToolCallContent::ContentBlock(_) => None,
309 ToolCallContent::Diff(_) => None,
310 })
311 }
312
313 fn to_markdown(&self, cx: &App) -> String {
314 let mut markdown = format!(
315 "**Tool Call: {}**\nStatus: {}\n\n",
316 self.label.read(cx).source(),
317 self.status
318 );
319 for content in &self.content {
320 markdown.push_str(content.to_markdown(cx).as_str());
321 markdown.push_str("\n\n");
322 }
323 markdown
324 }
325
326 async fn resolve_location(
327 location: acp::ToolCallLocation,
328 project: WeakEntity<Project>,
329 cx: &mut AsyncApp,
330 ) -> Option<AgentLocation> {
331 let buffer = project
332 .update(cx, |project, cx| {
333 project
334 .project_path_for_absolute_path(&location.path, cx)
335 .map(|path| project.open_buffer(path, cx))
336 })
337 .ok()??;
338 let buffer = buffer.await.log_err()?;
339 let position = buffer
340 .update(cx, |buffer, _| {
341 if let Some(row) = location.line {
342 let snapshot = buffer.snapshot();
343 let column = snapshot.indent_size_for_line(row).len;
344 let point = snapshot.clip_point(Point::new(row, column), Bias::Left);
345 snapshot.anchor_before(point)
346 } else {
347 Anchor::MIN
348 }
349 })
350 .ok()?;
351
352 Some(AgentLocation {
353 buffer: buffer.downgrade(),
354 position,
355 })
356 }
357
358 fn resolve_locations(
359 &self,
360 project: Entity<Project>,
361 cx: &mut App,
362 ) -> Task<Vec<Option<AgentLocation>>> {
363 let locations = self.locations.clone();
364 project.update(cx, |_, cx| {
365 cx.spawn(async move |project, cx| {
366 let mut new_locations = Vec::new();
367 for location in locations {
368 new_locations.push(Self::resolve_location(location, project.clone(), cx).await);
369 }
370 new_locations
371 })
372 })
373 }
374}
375
376#[derive(Debug)]
377pub enum ToolCallStatus {
378 /// The tool call hasn't started running yet, but we start showing it to
379 /// the user.
380 Pending,
381 /// The tool call is waiting for confirmation from the user.
382 WaitingForConfirmation {
383 options: Vec<acp::PermissionOption>,
384 respond_tx: oneshot::Sender<acp::PermissionOptionId>,
385 },
386 /// The tool call is currently running.
387 InProgress,
388 /// The tool call completed successfully.
389 Completed,
390 /// The tool call failed.
391 Failed,
392 /// The user rejected the tool call.
393 Rejected,
394 /// The user canceled generation so the tool call was canceled.
395 Canceled,
396}
397
398impl From<acp::ToolCallStatus> for ToolCallStatus {
399 fn from(status: acp::ToolCallStatus) -> Self {
400 match status {
401 acp::ToolCallStatus::Pending => Self::Pending,
402 acp::ToolCallStatus::InProgress => Self::InProgress,
403 acp::ToolCallStatus::Completed => Self::Completed,
404 acp::ToolCallStatus::Failed => Self::Failed,
405 }
406 }
407}
408
409impl Display for ToolCallStatus {
410 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
411 write!(
412 f,
413 "{}",
414 match self {
415 ToolCallStatus::Pending => "Pending",
416 ToolCallStatus::WaitingForConfirmation { .. } => "Waiting for confirmation",
417 ToolCallStatus::InProgress => "In Progress",
418 ToolCallStatus::Completed => "Completed",
419 ToolCallStatus::Failed => "Failed",
420 ToolCallStatus::Rejected => "Rejected",
421 ToolCallStatus::Canceled => "Canceled",
422 }
423 )
424 }
425}
426
427#[derive(Debug, PartialEq, Clone)]
428pub enum ContentBlock {
429 Empty,
430 Markdown { markdown: Entity<Markdown> },
431 ResourceLink { resource_link: acp::ResourceLink },
432}
433
434impl ContentBlock {
435 pub fn new(
436 block: acp::ContentBlock,
437 language_registry: &Arc<LanguageRegistry>,
438 cx: &mut App,
439 ) -> Self {
440 let mut this = Self::Empty;
441 this.append(block, language_registry, cx);
442 this
443 }
444
445 pub fn new_combined(
446 blocks: impl IntoIterator<Item = acp::ContentBlock>,
447 language_registry: Arc<LanguageRegistry>,
448 cx: &mut App,
449 ) -> Self {
450 let mut this = Self::Empty;
451 for block in blocks {
452 this.append(block, &language_registry, cx);
453 }
454 this
455 }
456
457 pub fn append(
458 &mut self,
459 block: acp::ContentBlock,
460 language_registry: &Arc<LanguageRegistry>,
461 cx: &mut App,
462 ) {
463 if matches!(self, ContentBlock::Empty)
464 && let acp::ContentBlock::ResourceLink(resource_link) = block
465 {
466 *self = ContentBlock::ResourceLink { resource_link };
467 return;
468 }
469
470 let new_content = self.block_string_contents(block);
471
472 match self {
473 ContentBlock::Empty => {
474 *self = Self::create_markdown_block(new_content, language_registry, cx);
475 }
476 ContentBlock::Markdown { markdown } => {
477 markdown.update(cx, |markdown, cx| markdown.append(&new_content, cx));
478 }
479 ContentBlock::ResourceLink { resource_link } => {
480 let existing_content = Self::resource_link_md(&resource_link.uri);
481 let combined = format!("{}\n{}", existing_content, new_content);
482
483 *self = Self::create_markdown_block(combined, language_registry, cx);
484 }
485 }
486 }
487
488 fn create_markdown_block(
489 content: String,
490 language_registry: &Arc<LanguageRegistry>,
491 cx: &mut App,
492 ) -> ContentBlock {
493 ContentBlock::Markdown {
494 markdown: cx
495 .new(|cx| Markdown::new(content.into(), Some(language_registry.clone()), None, cx)),
496 }
497 }
498
499 fn block_string_contents(&self, block: acp::ContentBlock) -> String {
500 match block {
501 acp::ContentBlock::Text(text_content) => text_content.text,
502 acp::ContentBlock::ResourceLink(resource_link) => {
503 Self::resource_link_md(&resource_link.uri)
504 }
505 acp::ContentBlock::Resource(acp::EmbeddedResource {
506 resource:
507 acp::EmbeddedResourceResource::TextResourceContents(acp::TextResourceContents {
508 uri,
509 ..
510 }),
511 ..
512 }) => Self::resource_link_md(&uri),
513 acp::ContentBlock::Image(image) => Self::image_md(&image),
514 acp::ContentBlock::Audio(_) | acp::ContentBlock::Resource(_) => String::new(),
515 }
516 }
517
518 fn resource_link_md(uri: &str) -> String {
519 if let Some(uri) = MentionUri::parse(uri).log_err() {
520 uri.as_link().to_string()
521 } else {
522 uri.to_string()
523 }
524 }
525
526 fn image_md(_image: &acp::ImageContent) -> String {
527 "`Image`".into()
528 }
529
530 pub fn to_markdown<'a>(&'a self, cx: &'a App) -> &'a str {
531 match self {
532 ContentBlock::Empty => "",
533 ContentBlock::Markdown { markdown } => markdown.read(cx).source(),
534 ContentBlock::ResourceLink { resource_link } => &resource_link.uri,
535 }
536 }
537
538 pub fn markdown(&self) -> Option<&Entity<Markdown>> {
539 match self {
540 ContentBlock::Empty => None,
541 ContentBlock::Markdown { markdown } => Some(markdown),
542 ContentBlock::ResourceLink { .. } => None,
543 }
544 }
545
546 pub fn resource_link(&self) -> Option<&acp::ResourceLink> {
547 match self {
548 ContentBlock::ResourceLink { resource_link } => Some(resource_link),
549 _ => None,
550 }
551 }
552}
553
554#[derive(Debug)]
555pub enum ToolCallContent {
556 ContentBlock(ContentBlock),
557 Diff(Entity<Diff>),
558 Terminal(Entity<Terminal>),
559}
560
561impl ToolCallContent {
562 pub fn from_acp(
563 content: acp::ToolCallContent,
564 language_registry: Arc<LanguageRegistry>,
565 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
566 cx: &mut App,
567 ) -> Result<Self> {
568 match content {
569 acp::ToolCallContent::Content { content } => Ok(Self::ContentBlock(ContentBlock::new(
570 content,
571 &language_registry,
572 cx,
573 ))),
574 acp::ToolCallContent::Diff { diff } => Ok(Self::Diff(cx.new(|cx| {
575 Diff::finalized(
576 diff.path,
577 diff.old_text,
578 diff.new_text,
579 language_registry,
580 cx,
581 )
582 }))),
583 acp::ToolCallContent::Terminal { terminal_id } => terminals
584 .get(&terminal_id)
585 .cloned()
586 .map(Self::Terminal)
587 .ok_or_else(|| anyhow::anyhow!("Terminal with id `{}` not found", terminal_id)),
588 }
589 }
590
591 pub fn update_from_acp(
592 &mut self,
593 new: acp::ToolCallContent,
594 language_registry: Arc<LanguageRegistry>,
595 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
596 cx: &mut App,
597 ) -> Result<()> {
598 let needs_update = match (&self, &new) {
599 (Self::Diff(old_diff), acp::ToolCallContent::Diff { diff: new_diff }) => {
600 old_diff.read(cx).needs_update(
601 new_diff.old_text.as_deref().unwrap_or(""),
602 &new_diff.new_text,
603 cx,
604 )
605 }
606 _ => true,
607 };
608
609 if needs_update {
610 *self = Self::from_acp(new, language_registry, terminals, cx)?;
611 }
612 Ok(())
613 }
614
615 pub fn to_markdown(&self, cx: &App) -> String {
616 match self {
617 Self::ContentBlock(content) => content.to_markdown(cx).to_string(),
618 Self::Diff(diff) => diff.read(cx).to_markdown(cx),
619 Self::Terminal(terminal) => terminal.read(cx).to_markdown(cx),
620 }
621 }
622}
623
624#[derive(Debug, PartialEq)]
625pub enum ToolCallUpdate {
626 UpdateFields(acp::ToolCallUpdate),
627 UpdateDiff(ToolCallUpdateDiff),
628 UpdateTerminal(ToolCallUpdateTerminal),
629}
630
631impl ToolCallUpdate {
632 fn id(&self) -> &acp::ToolCallId {
633 match self {
634 Self::UpdateFields(update) => &update.id,
635 Self::UpdateDiff(diff) => &diff.id,
636 Self::UpdateTerminal(terminal) => &terminal.id,
637 }
638 }
639}
640
641impl From<acp::ToolCallUpdate> for ToolCallUpdate {
642 fn from(update: acp::ToolCallUpdate) -> Self {
643 Self::UpdateFields(update)
644 }
645}
646
647impl From<ToolCallUpdateDiff> for ToolCallUpdate {
648 fn from(diff: ToolCallUpdateDiff) -> Self {
649 Self::UpdateDiff(diff)
650 }
651}
652
653#[derive(Debug, PartialEq)]
654pub struct ToolCallUpdateDiff {
655 pub id: acp::ToolCallId,
656 pub diff: Entity<Diff>,
657}
658
659impl From<ToolCallUpdateTerminal> for ToolCallUpdate {
660 fn from(terminal: ToolCallUpdateTerminal) -> Self {
661 Self::UpdateTerminal(terminal)
662 }
663}
664
665#[derive(Debug, PartialEq)]
666pub struct ToolCallUpdateTerminal {
667 pub id: acp::ToolCallId,
668 pub terminal: Entity<Terminal>,
669}
670
671#[derive(Debug, Default)]
672pub struct Plan {
673 pub entries: Vec<PlanEntry>,
674}
675
676#[derive(Debug)]
677pub struct PlanStats<'a> {
678 pub in_progress_entry: Option<&'a PlanEntry>,
679 pub pending: u32,
680 pub completed: u32,
681}
682
683impl Plan {
684 pub fn is_empty(&self) -> bool {
685 self.entries.is_empty()
686 }
687
688 pub fn stats(&self) -> PlanStats<'_> {
689 let mut stats = PlanStats {
690 in_progress_entry: None,
691 pending: 0,
692 completed: 0,
693 };
694
695 for entry in &self.entries {
696 match &entry.status {
697 acp::PlanEntryStatus::Pending => {
698 stats.pending += 1;
699 }
700 acp::PlanEntryStatus::InProgress => {
701 stats.in_progress_entry = stats.in_progress_entry.or(Some(entry));
702 }
703 acp::PlanEntryStatus::Completed => {
704 stats.completed += 1;
705 }
706 }
707 }
708
709 stats
710 }
711}
712
713#[derive(Debug)]
714pub struct PlanEntry {
715 pub content: Entity<Markdown>,
716 pub priority: acp::PlanEntryPriority,
717 pub status: acp::PlanEntryStatus,
718}
719
720impl PlanEntry {
721 pub fn from_acp(entry: acp::PlanEntry, cx: &mut App) -> Self {
722 Self {
723 content: cx.new(|cx| Markdown::new(entry.content.into(), None, None, cx)),
724 priority: entry.priority,
725 status: entry.status,
726 }
727 }
728}
729
730#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
731pub struct TokenUsage {
732 pub max_tokens: u64,
733 pub used_tokens: u64,
734}
735
736impl TokenUsage {
737 pub fn ratio(&self) -> TokenUsageRatio {
738 #[cfg(debug_assertions)]
739 let warning_threshold: f32 = std::env::var("ZED_THREAD_WARNING_THRESHOLD")
740 .unwrap_or("0.8".to_string())
741 .parse()
742 .unwrap();
743 #[cfg(not(debug_assertions))]
744 let warning_threshold: f32 = 0.8;
745
746 // When the maximum is unknown because there is no selected model,
747 // avoid showing the token limit warning.
748 if self.max_tokens == 0 {
749 TokenUsageRatio::Normal
750 } else if self.used_tokens >= self.max_tokens {
751 TokenUsageRatio::Exceeded
752 } else if self.used_tokens as f32 / self.max_tokens as f32 >= warning_threshold {
753 TokenUsageRatio::Warning
754 } else {
755 TokenUsageRatio::Normal
756 }
757 }
758}
759
760#[derive(Debug, Clone, PartialEq, Eq)]
761pub enum TokenUsageRatio {
762 Normal,
763 Warning,
764 Exceeded,
765}
766
767#[derive(Debug, Clone)]
768pub struct RetryStatus {
769 pub last_error: SharedString,
770 pub attempt: usize,
771 pub max_attempts: usize,
772 pub started_at: Instant,
773 pub duration: Duration,
774}
775
776pub struct AcpThread {
777 title: SharedString,
778 entries: Vec<AgentThreadEntry>,
779 plan: Plan,
780 project: Entity<Project>,
781 action_log: Entity<ActionLog>,
782 shared_buffers: HashMap<Entity<Buffer>, BufferSnapshot>,
783 send_task: Option<Task<()>>,
784 connection: Rc<dyn AgentConnection>,
785 session_id: acp::SessionId,
786 token_usage: Option<TokenUsage>,
787 prompt_capabilities: acp::PromptCapabilities,
788 _observe_prompt_capabilities: Task<anyhow::Result<()>>,
789 determine_shell: Shared<Task<String>>,
790 terminals: HashMap<acp::TerminalId, Entity<Terminal>>,
791}
792
793#[derive(Debug)]
794pub enum AcpThreadEvent {
795 NewEntry,
796 TitleUpdated,
797 TokenUsageUpdated,
798 EntryUpdated(usize),
799 EntriesRemoved(Range<usize>),
800 ToolAuthorizationRequired,
801 Retry(RetryStatus),
802 Stopped,
803 Error,
804 LoadError(LoadError),
805 PromptCapabilitiesUpdated,
806 Refusal,
807 AvailableCommandsUpdated(Vec<acp::AvailableCommand>),
808 ModeUpdated(acp::SessionModeId),
809}
810
811impl EventEmitter<AcpThreadEvent> for AcpThread {}
812
813#[derive(PartialEq, Eq, Debug)]
814pub enum ThreadStatus {
815 Idle,
816 Generating,
817}
818
819#[derive(Debug, Clone)]
820pub enum LoadError {
821 Unsupported {
822 command: SharedString,
823 current_version: SharedString,
824 minimum_version: SharedString,
825 },
826 FailedToInstall(SharedString),
827 Exited {
828 status: ExitStatus,
829 },
830 Other(SharedString),
831}
832
833impl Display for LoadError {
834 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
835 match self {
836 LoadError::Unsupported {
837 command: path,
838 current_version,
839 minimum_version,
840 } => {
841 write!(
842 f,
843 "version {current_version} from {path} is not supported (need at least {minimum_version})"
844 )
845 }
846 LoadError::FailedToInstall(msg) => write!(f, "Failed to install: {msg}"),
847 LoadError::Exited { status } => write!(f, "Server exited with status {status}"),
848 LoadError::Other(msg) => write!(f, "{msg}"),
849 }
850 }
851}
852
853impl Error for LoadError {}
854
855impl AcpThread {
856 pub fn new(
857 title: impl Into<SharedString>,
858 connection: Rc<dyn AgentConnection>,
859 project: Entity<Project>,
860 action_log: Entity<ActionLog>,
861 session_id: acp::SessionId,
862 mut prompt_capabilities_rx: watch::Receiver<acp::PromptCapabilities>,
863 cx: &mut Context<Self>,
864 ) -> Self {
865 let prompt_capabilities = *prompt_capabilities_rx.borrow();
866 let task = cx.spawn::<_, anyhow::Result<()>>(async move |this, cx| {
867 loop {
868 let caps = prompt_capabilities_rx.recv().await?;
869 this.update(cx, |this, cx| {
870 this.prompt_capabilities = caps;
871 cx.emit(AcpThreadEvent::PromptCapabilitiesUpdated);
872 })?;
873 }
874 });
875
876 let determine_shell = cx
877 .background_spawn(async move {
878 if cfg!(windows) {
879 return get_system_shell();
880 }
881
882 if which::which("bash").is_ok() {
883 "bash".into()
884 } else {
885 get_system_shell()
886 }
887 })
888 .shared();
889
890 Self {
891 action_log,
892 shared_buffers: Default::default(),
893 entries: Default::default(),
894 plan: Default::default(),
895 title: title.into(),
896 project,
897 send_task: None,
898 connection,
899 session_id,
900 token_usage: None,
901 prompt_capabilities,
902 _observe_prompt_capabilities: task,
903 terminals: HashMap::default(),
904 determine_shell,
905 }
906 }
907
908 pub fn prompt_capabilities(&self) -> acp::PromptCapabilities {
909 self.prompt_capabilities
910 }
911
912 pub fn connection(&self) -> &Rc<dyn AgentConnection> {
913 &self.connection
914 }
915
916 pub fn action_log(&self) -> &Entity<ActionLog> {
917 &self.action_log
918 }
919
920 pub fn project(&self) -> &Entity<Project> {
921 &self.project
922 }
923
924 pub fn title(&self) -> SharedString {
925 self.title.clone()
926 }
927
928 pub fn entries(&self) -> &[AgentThreadEntry] {
929 &self.entries
930 }
931
932 pub fn session_id(&self) -> &acp::SessionId {
933 &self.session_id
934 }
935
936 pub fn status(&self) -> ThreadStatus {
937 if self.send_task.is_some() {
938 ThreadStatus::Generating
939 } else {
940 ThreadStatus::Idle
941 }
942 }
943
944 pub fn token_usage(&self) -> Option<&TokenUsage> {
945 self.token_usage.as_ref()
946 }
947
948 pub fn has_pending_edit_tool_calls(&self) -> bool {
949 for entry in self.entries.iter().rev() {
950 match entry {
951 AgentThreadEntry::UserMessage(_) => return false,
952 AgentThreadEntry::ToolCall(
953 call @ ToolCall {
954 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
955 ..
956 },
957 ) if call.diffs().next().is_some() => {
958 return true;
959 }
960 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
961 }
962 }
963
964 false
965 }
966
967 pub fn used_tools_since_last_user_message(&self) -> bool {
968 for entry in self.entries.iter().rev() {
969 match entry {
970 AgentThreadEntry::UserMessage(..) => return false,
971 AgentThreadEntry::AssistantMessage(..) => continue,
972 AgentThreadEntry::ToolCall(..) => return true,
973 }
974 }
975
976 false
977 }
978
979 pub fn handle_session_update(
980 &mut self,
981 update: acp::SessionUpdate,
982 cx: &mut Context<Self>,
983 ) -> Result<(), acp::Error> {
984 match update {
985 acp::SessionUpdate::UserMessageChunk { content } => {
986 self.push_user_content_block(None, content, cx);
987 }
988 acp::SessionUpdate::AgentMessageChunk { content } => {
989 self.push_assistant_content_block(content, false, cx);
990 }
991 acp::SessionUpdate::AgentThoughtChunk { content } => {
992 self.push_assistant_content_block(content, true, cx);
993 }
994 acp::SessionUpdate::ToolCall(tool_call) => {
995 self.upsert_tool_call(tool_call, cx)?;
996 }
997 acp::SessionUpdate::ToolCallUpdate(tool_call_update) => {
998 self.update_tool_call(tool_call_update, cx)?;
999 }
1000 acp::SessionUpdate::Plan(plan) => {
1001 self.update_plan(plan, cx);
1002 }
1003 acp::SessionUpdate::AvailableCommandsUpdate { available_commands } => {
1004 cx.emit(AcpThreadEvent::AvailableCommandsUpdated(available_commands))
1005 }
1006 acp::SessionUpdate::CurrentModeUpdate { current_mode_id } => {
1007 cx.emit(AcpThreadEvent::ModeUpdated(current_mode_id))
1008 }
1009 }
1010 Ok(())
1011 }
1012
1013 pub fn push_user_content_block(
1014 &mut self,
1015 message_id: Option<UserMessageId>,
1016 chunk: acp::ContentBlock,
1017 cx: &mut Context<Self>,
1018 ) {
1019 let language_registry = self.project.read(cx).languages().clone();
1020 let entries_len = self.entries.len();
1021
1022 if let Some(last_entry) = self.entries.last_mut()
1023 && let AgentThreadEntry::UserMessage(UserMessage {
1024 id,
1025 content,
1026 chunks,
1027 ..
1028 }) = last_entry
1029 {
1030 *id = message_id.or(id.take());
1031 content.append(chunk.clone(), &language_registry, cx);
1032 chunks.push(chunk);
1033 let idx = entries_len - 1;
1034 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1035 } else {
1036 let content = ContentBlock::new(chunk.clone(), &language_registry, cx);
1037 self.push_entry(
1038 AgentThreadEntry::UserMessage(UserMessage {
1039 id: message_id,
1040 content,
1041 chunks: vec![chunk],
1042 checkpoint: None,
1043 }),
1044 cx,
1045 );
1046 }
1047 }
1048
1049 pub fn push_assistant_content_block(
1050 &mut self,
1051 chunk: acp::ContentBlock,
1052 is_thought: bool,
1053 cx: &mut Context<Self>,
1054 ) {
1055 let language_registry = self.project.read(cx).languages().clone();
1056 let entries_len = self.entries.len();
1057 if let Some(last_entry) = self.entries.last_mut()
1058 && let AgentThreadEntry::AssistantMessage(AssistantMessage { chunks }) = last_entry
1059 {
1060 let idx = entries_len - 1;
1061 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1062 match (chunks.last_mut(), is_thought) {
1063 (Some(AssistantMessageChunk::Message { block }), false)
1064 | (Some(AssistantMessageChunk::Thought { block }), true) => {
1065 block.append(chunk, &language_registry, cx)
1066 }
1067 _ => {
1068 let block = ContentBlock::new(chunk, &language_registry, cx);
1069 if is_thought {
1070 chunks.push(AssistantMessageChunk::Thought { block })
1071 } else {
1072 chunks.push(AssistantMessageChunk::Message { block })
1073 }
1074 }
1075 }
1076 } else {
1077 let block = ContentBlock::new(chunk, &language_registry, cx);
1078 let chunk = if is_thought {
1079 AssistantMessageChunk::Thought { block }
1080 } else {
1081 AssistantMessageChunk::Message { block }
1082 };
1083
1084 self.push_entry(
1085 AgentThreadEntry::AssistantMessage(AssistantMessage {
1086 chunks: vec![chunk],
1087 }),
1088 cx,
1089 );
1090 }
1091 }
1092
1093 fn push_entry(&mut self, entry: AgentThreadEntry, cx: &mut Context<Self>) {
1094 self.entries.push(entry);
1095 cx.emit(AcpThreadEvent::NewEntry);
1096 }
1097
1098 pub fn can_set_title(&mut self, cx: &mut Context<Self>) -> bool {
1099 self.connection.set_title(&self.session_id, cx).is_some()
1100 }
1101
1102 pub fn set_title(&mut self, title: SharedString, cx: &mut Context<Self>) -> Task<Result<()>> {
1103 if title != self.title {
1104 self.title = title.clone();
1105 cx.emit(AcpThreadEvent::TitleUpdated);
1106 if let Some(set_title) = self.connection.set_title(&self.session_id, cx) {
1107 return set_title.run(title, cx);
1108 }
1109 }
1110 Task::ready(Ok(()))
1111 }
1112
1113 pub fn update_token_usage(&mut self, usage: Option<TokenUsage>, cx: &mut Context<Self>) {
1114 self.token_usage = usage;
1115 cx.emit(AcpThreadEvent::TokenUsageUpdated);
1116 }
1117
1118 pub fn update_retry_status(&mut self, status: RetryStatus, cx: &mut Context<Self>) {
1119 cx.emit(AcpThreadEvent::Retry(status));
1120 }
1121
1122 pub fn update_tool_call(
1123 &mut self,
1124 update: impl Into<ToolCallUpdate>,
1125 cx: &mut Context<Self>,
1126 ) -> Result<()> {
1127 let update = update.into();
1128 let languages = self.project.read(cx).languages().clone();
1129
1130 let ix = self
1131 .index_for_tool_call(update.id())
1132 .context("Tool call not found")?;
1133 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1134 unreachable!()
1135 };
1136
1137 match update {
1138 ToolCallUpdate::UpdateFields(update) => {
1139 let location_updated = update.fields.locations.is_some();
1140 call.update_fields(update.fields, languages, &self.terminals, cx)?;
1141 if location_updated {
1142 self.resolve_locations(update.id, cx);
1143 }
1144 }
1145 ToolCallUpdate::UpdateDiff(update) => {
1146 call.content.clear();
1147 call.content.push(ToolCallContent::Diff(update.diff));
1148 }
1149 ToolCallUpdate::UpdateTerminal(update) => {
1150 call.content.clear();
1151 call.content
1152 .push(ToolCallContent::Terminal(update.terminal));
1153 }
1154 }
1155
1156 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1157
1158 Ok(())
1159 }
1160
1161 /// Updates a tool call if id matches an existing entry, otherwise inserts a new one.
1162 pub fn upsert_tool_call(
1163 &mut self,
1164 tool_call: acp::ToolCall,
1165 cx: &mut Context<Self>,
1166 ) -> Result<(), acp::Error> {
1167 let status = tool_call.status.into();
1168 self.upsert_tool_call_inner(tool_call.into(), status, cx)
1169 }
1170
1171 /// Fails if id does not match an existing entry.
1172 pub fn upsert_tool_call_inner(
1173 &mut self,
1174 update: acp::ToolCallUpdate,
1175 status: ToolCallStatus,
1176 cx: &mut Context<Self>,
1177 ) -> Result<(), acp::Error> {
1178 let language_registry = self.project.read(cx).languages().clone();
1179 let id = update.id.clone();
1180
1181 if let Some(ix) = self.index_for_tool_call(&id) {
1182 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1183 unreachable!()
1184 };
1185
1186 call.update_fields(update.fields, language_registry, &self.terminals, cx)?;
1187 call.status = status;
1188
1189 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1190 } else {
1191 let call = ToolCall::from_acp(
1192 update.try_into()?,
1193 status,
1194 language_registry,
1195 &self.terminals,
1196 cx,
1197 )?;
1198 self.push_entry(AgentThreadEntry::ToolCall(call), cx);
1199 };
1200
1201 self.resolve_locations(id, cx);
1202 Ok(())
1203 }
1204
1205 fn index_for_tool_call(&self, id: &acp::ToolCallId) -> Option<usize> {
1206 self.entries
1207 .iter()
1208 .enumerate()
1209 .rev()
1210 .find_map(|(index, entry)| {
1211 if let AgentThreadEntry::ToolCall(tool_call) = entry
1212 && &tool_call.id == id
1213 {
1214 Some(index)
1215 } else {
1216 None
1217 }
1218 })
1219 }
1220
1221 fn tool_call_mut(&mut self, id: &acp::ToolCallId) -> Option<(usize, &mut ToolCall)> {
1222 // The tool call we are looking for is typically the last one, or very close to the end.
1223 // At the moment, it doesn't seem like a hashmap would be a good fit for this use case.
1224 self.entries
1225 .iter_mut()
1226 .enumerate()
1227 .rev()
1228 .find_map(|(index, tool_call)| {
1229 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1230 && &tool_call.id == id
1231 {
1232 Some((index, tool_call))
1233 } else {
1234 None
1235 }
1236 })
1237 }
1238
1239 pub fn tool_call(&mut self, id: &acp::ToolCallId) -> Option<(usize, &ToolCall)> {
1240 self.entries
1241 .iter()
1242 .enumerate()
1243 .rev()
1244 .find_map(|(index, tool_call)| {
1245 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1246 && &tool_call.id == id
1247 {
1248 Some((index, tool_call))
1249 } else {
1250 None
1251 }
1252 })
1253 }
1254
1255 pub fn resolve_locations(&mut self, id: acp::ToolCallId, cx: &mut Context<Self>) {
1256 let project = self.project.clone();
1257 let Some((_, tool_call)) = self.tool_call_mut(&id) else {
1258 return;
1259 };
1260 let task = tool_call.resolve_locations(project, cx);
1261 cx.spawn(async move |this, cx| {
1262 let resolved_locations = task.await;
1263 this.update(cx, |this, cx| {
1264 let project = this.project.clone();
1265 let Some((ix, tool_call)) = this.tool_call_mut(&id) else {
1266 return;
1267 };
1268 if let Some(Some(location)) = resolved_locations.last() {
1269 project.update(cx, |project, cx| {
1270 if let Some(agent_location) = project.agent_location() {
1271 let should_ignore = agent_location.buffer == location.buffer
1272 && location
1273 .buffer
1274 .update(cx, |buffer, _| {
1275 let snapshot = buffer.snapshot();
1276 let old_position =
1277 agent_location.position.to_point(&snapshot);
1278 let new_position = location.position.to_point(&snapshot);
1279 // ignore this so that when we get updates from the edit tool
1280 // the position doesn't reset to the startof line
1281 old_position.row == new_position.row
1282 && old_position.column > new_position.column
1283 })
1284 .ok()
1285 .unwrap_or_default();
1286 if !should_ignore {
1287 project.set_agent_location(Some(location.clone()), cx);
1288 }
1289 }
1290 });
1291 }
1292 if tool_call.resolved_locations != resolved_locations {
1293 tool_call.resolved_locations = resolved_locations;
1294 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1295 }
1296 })
1297 })
1298 .detach();
1299 }
1300
1301 pub fn request_tool_call_authorization(
1302 &mut self,
1303 tool_call: acp::ToolCallUpdate,
1304 options: Vec<acp::PermissionOption>,
1305 respect_always_allow_setting: bool,
1306 cx: &mut Context<Self>,
1307 ) -> Result<BoxFuture<'static, acp::RequestPermissionOutcome>> {
1308 let (tx, rx) = oneshot::channel();
1309
1310 if respect_always_allow_setting && AgentSettings::get_global(cx).always_allow_tool_actions {
1311 // Don't use AllowAlways, because then if you were to turn off always_allow_tool_actions,
1312 // some tools would (incorrectly) continue to auto-accept.
1313 if let Some(allow_once_option) = options.iter().find_map(|option| {
1314 if matches!(option.kind, acp::PermissionOptionKind::AllowOnce) {
1315 Some(option.id.clone())
1316 } else {
1317 None
1318 }
1319 }) {
1320 self.upsert_tool_call_inner(tool_call, ToolCallStatus::Pending, cx)?;
1321 return Ok(async {
1322 acp::RequestPermissionOutcome::Selected {
1323 option_id: allow_once_option,
1324 }
1325 }
1326 .boxed());
1327 }
1328 }
1329
1330 let status = ToolCallStatus::WaitingForConfirmation {
1331 options,
1332 respond_tx: tx,
1333 };
1334
1335 self.upsert_tool_call_inner(tool_call, status, cx)?;
1336 cx.emit(AcpThreadEvent::ToolAuthorizationRequired);
1337
1338 let fut = async {
1339 match rx.await {
1340 Ok(option) => acp::RequestPermissionOutcome::Selected { option_id: option },
1341 Err(oneshot::Canceled) => acp::RequestPermissionOutcome::Cancelled,
1342 }
1343 }
1344 .boxed();
1345
1346 Ok(fut)
1347 }
1348
1349 pub fn authorize_tool_call(
1350 &mut self,
1351 id: acp::ToolCallId,
1352 option_id: acp::PermissionOptionId,
1353 option_kind: acp::PermissionOptionKind,
1354 cx: &mut Context<Self>,
1355 ) {
1356 let Some((ix, call)) = self.tool_call_mut(&id) else {
1357 return;
1358 };
1359
1360 let new_status = match option_kind {
1361 acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways => {
1362 ToolCallStatus::Rejected
1363 }
1364 acp::PermissionOptionKind::AllowOnce | acp::PermissionOptionKind::AllowAlways => {
1365 ToolCallStatus::InProgress
1366 }
1367 };
1368
1369 let curr_status = mem::replace(&mut call.status, new_status);
1370
1371 if let ToolCallStatus::WaitingForConfirmation { respond_tx, .. } = curr_status {
1372 respond_tx.send(option_id).log_err();
1373 } else if cfg!(debug_assertions) {
1374 panic!("tried to authorize an already authorized tool call");
1375 }
1376
1377 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1378 }
1379
1380 pub fn first_tool_awaiting_confirmation(&self) -> Option<&ToolCall> {
1381 let mut first_tool_call = None;
1382
1383 for entry in self.entries.iter().rev() {
1384 match &entry {
1385 AgentThreadEntry::ToolCall(call) => {
1386 if let ToolCallStatus::WaitingForConfirmation { .. } = call.status {
1387 first_tool_call = Some(call);
1388 } else {
1389 continue;
1390 }
1391 }
1392 AgentThreadEntry::UserMessage(_) | AgentThreadEntry::AssistantMessage(_) => {
1393 // Reached the beginning of the turn.
1394 // If we had pending permission requests in the previous turn, they have been cancelled.
1395 break;
1396 }
1397 }
1398 }
1399
1400 first_tool_call
1401 }
1402
1403 pub fn plan(&self) -> &Plan {
1404 &self.plan
1405 }
1406
1407 pub fn update_plan(&mut self, request: acp::Plan, cx: &mut Context<Self>) {
1408 let new_entries_len = request.entries.len();
1409 let mut new_entries = request.entries.into_iter();
1410
1411 // Reuse existing markdown to prevent flickering
1412 for (old, new) in self.plan.entries.iter_mut().zip(new_entries.by_ref()) {
1413 let PlanEntry {
1414 content,
1415 priority,
1416 status,
1417 } = old;
1418 content.update(cx, |old, cx| {
1419 old.replace(new.content, cx);
1420 });
1421 *priority = new.priority;
1422 *status = new.status;
1423 }
1424 for new in new_entries {
1425 self.plan.entries.push(PlanEntry::from_acp(new, cx))
1426 }
1427 self.plan.entries.truncate(new_entries_len);
1428
1429 cx.notify();
1430 }
1431
1432 fn clear_completed_plan_entries(&mut self, cx: &mut Context<Self>) {
1433 self.plan
1434 .entries
1435 .retain(|entry| !matches!(entry.status, acp::PlanEntryStatus::Completed));
1436 cx.notify();
1437 }
1438
1439 #[cfg(any(test, feature = "test-support"))]
1440 pub fn send_raw(
1441 &mut self,
1442 message: &str,
1443 cx: &mut Context<Self>,
1444 ) -> BoxFuture<'static, Result<()>> {
1445 self.send(
1446 vec![acp::ContentBlock::Text(acp::TextContent {
1447 text: message.to_string(),
1448 annotations: None,
1449 })],
1450 cx,
1451 )
1452 }
1453
1454 pub fn send(
1455 &mut self,
1456 message: Vec<acp::ContentBlock>,
1457 cx: &mut Context<Self>,
1458 ) -> BoxFuture<'static, Result<()>> {
1459 let block = ContentBlock::new_combined(
1460 message.clone(),
1461 self.project.read(cx).languages().clone(),
1462 cx,
1463 );
1464 let request = acp::PromptRequest {
1465 prompt: message.clone(),
1466 session_id: self.session_id.clone(),
1467 };
1468 let git_store = self.project.read(cx).git_store().clone();
1469
1470 let message_id = if self.connection.truncate(&self.session_id, cx).is_some() {
1471 Some(UserMessageId::new())
1472 } else {
1473 None
1474 };
1475
1476 self.run_turn(cx, async move |this, cx| {
1477 this.update(cx, |this, cx| {
1478 this.push_entry(
1479 AgentThreadEntry::UserMessage(UserMessage {
1480 id: message_id.clone(),
1481 content: block,
1482 chunks: message,
1483 checkpoint: None,
1484 }),
1485 cx,
1486 );
1487 })
1488 .ok();
1489
1490 let old_checkpoint = git_store
1491 .update(cx, |git, cx| git.checkpoint(cx))?
1492 .await
1493 .context("failed to get old checkpoint")
1494 .log_err();
1495 this.update(cx, |this, cx| {
1496 if let Some((_ix, message)) = this.last_user_message() {
1497 message.checkpoint = old_checkpoint.map(|git_checkpoint| Checkpoint {
1498 git_checkpoint,
1499 show: false,
1500 });
1501 }
1502 this.connection.prompt(message_id, request, cx)
1503 })?
1504 .await
1505 })
1506 }
1507
1508 pub fn can_resume(&self, cx: &App) -> bool {
1509 self.connection.resume(&self.session_id, cx).is_some()
1510 }
1511
1512 pub fn resume(&mut self, cx: &mut Context<Self>) -> BoxFuture<'static, Result<()>> {
1513 self.run_turn(cx, async move |this, cx| {
1514 this.update(cx, |this, cx| {
1515 this.connection
1516 .resume(&this.session_id, cx)
1517 .map(|resume| resume.run(cx))
1518 })?
1519 .context("resuming a session is not supported")?
1520 .await
1521 })
1522 }
1523
1524 fn run_turn(
1525 &mut self,
1526 cx: &mut Context<Self>,
1527 f: impl 'static + AsyncFnOnce(WeakEntity<Self>, &mut AsyncApp) -> Result<acp::PromptResponse>,
1528 ) -> BoxFuture<'static, Result<()>> {
1529 self.clear_completed_plan_entries(cx);
1530
1531 let (tx, rx) = oneshot::channel();
1532 let cancel_task = self.cancel(cx);
1533
1534 self.send_task = Some(cx.spawn(async move |this, cx| {
1535 cancel_task.await;
1536 tx.send(f(this, cx).await).ok();
1537 }));
1538
1539 cx.spawn(async move |this, cx| {
1540 let response = rx.await;
1541
1542 this.update(cx, |this, cx| this.update_last_checkpoint(cx))?
1543 .await?;
1544
1545 this.update(cx, |this, cx| {
1546 this.project
1547 .update(cx, |project, cx| project.set_agent_location(None, cx));
1548 match response {
1549 Ok(Err(e)) => {
1550 this.send_task.take();
1551 cx.emit(AcpThreadEvent::Error);
1552 Err(e)
1553 }
1554 result => {
1555 let canceled = matches!(
1556 result,
1557 Ok(Ok(acp::PromptResponse {
1558 stop_reason: acp::StopReason::Cancelled
1559 }))
1560 );
1561
1562 // We only take the task if the current prompt wasn't canceled.
1563 //
1564 // This prompt may have been canceled because another one was sent
1565 // while it was still generating. In these cases, dropping `send_task`
1566 // would cause the next generation to be canceled.
1567 if !canceled {
1568 this.send_task.take();
1569 }
1570
1571 // Handle refusal - distinguish between user prompt and tool call refusals
1572 if let Ok(Ok(acp::PromptResponse {
1573 stop_reason: acp::StopReason::Refusal,
1574 })) = result
1575 {
1576 if let Some((user_msg_ix, _)) = this.last_user_message() {
1577 // Check if there's a completed tool call with results after the last user message
1578 // This indicates the refusal is in response to tool output, not the user's prompt
1579 let has_completed_tool_call_after_user_msg =
1580 this.entries.iter().skip(user_msg_ix + 1).any(|entry| {
1581 if let AgentThreadEntry::ToolCall(tool_call) = entry {
1582 // Check if the tool call has completed and has output
1583 matches!(tool_call.status, ToolCallStatus::Completed)
1584 && tool_call.raw_output.is_some()
1585 } else {
1586 false
1587 }
1588 });
1589
1590 if has_completed_tool_call_after_user_msg {
1591 // Refusal is due to tool output - don't truncate, just notify
1592 // The model refused based on what the tool returned
1593 cx.emit(AcpThreadEvent::Refusal);
1594 } else {
1595 // User prompt was refused - truncate back to before the user message
1596 let range = user_msg_ix..this.entries.len();
1597 if range.start < range.end {
1598 this.entries.truncate(user_msg_ix);
1599 cx.emit(AcpThreadEvent::EntriesRemoved(range));
1600 }
1601 cx.emit(AcpThreadEvent::Refusal);
1602 }
1603 } else {
1604 // No user message found, treat as general refusal
1605 cx.emit(AcpThreadEvent::Refusal);
1606 }
1607 }
1608
1609 cx.emit(AcpThreadEvent::Stopped);
1610 Ok(())
1611 }
1612 }
1613 })?
1614 })
1615 .boxed()
1616 }
1617
1618 pub fn cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
1619 let Some(send_task) = self.send_task.take() else {
1620 return Task::ready(());
1621 };
1622
1623 for entry in self.entries.iter_mut() {
1624 if let AgentThreadEntry::ToolCall(call) = entry {
1625 let cancel = matches!(
1626 call.status,
1627 ToolCallStatus::Pending
1628 | ToolCallStatus::WaitingForConfirmation { .. }
1629 | ToolCallStatus::InProgress
1630 );
1631
1632 if cancel {
1633 call.status = ToolCallStatus::Canceled;
1634 }
1635 }
1636 }
1637
1638 self.connection.cancel(&self.session_id, cx);
1639
1640 // Wait for the send task to complete
1641 cx.foreground_executor().spawn(send_task)
1642 }
1643
1644 /// Restores the git working tree to the state at the given checkpoint (if one exists)
1645 pub fn restore_checkpoint(
1646 &mut self,
1647 id: UserMessageId,
1648 cx: &mut Context<Self>,
1649 ) -> Task<Result<()>> {
1650 let Some((_, message)) = self.user_message_mut(&id) else {
1651 return Task::ready(Err(anyhow!("message not found")));
1652 };
1653
1654 let checkpoint = message
1655 .checkpoint
1656 .as_ref()
1657 .map(|c| c.git_checkpoint.clone());
1658 let rewind = self.rewind(id.clone(), cx);
1659 let git_store = self.project.read(cx).git_store().clone();
1660
1661 cx.spawn(async move |_, cx| {
1662 rewind.await?;
1663 if let Some(checkpoint) = checkpoint {
1664 git_store
1665 .update(cx, |git, cx| git.restore_checkpoint(checkpoint, cx))?
1666 .await?;
1667 }
1668
1669 Ok(())
1670 })
1671 }
1672
1673 /// Rewinds this thread to before the entry at `index`, removing it and all
1674 /// subsequent entries while rejecting any action_log changes made from that point.
1675 /// Unlike `restore_checkpoint`, this method does not restore from git.
1676 pub fn rewind(&mut self, id: UserMessageId, cx: &mut Context<Self>) -> Task<Result<()>> {
1677 let Some(truncate) = self.connection.truncate(&self.session_id, cx) else {
1678 return Task::ready(Err(anyhow!("not supported")));
1679 };
1680
1681 cx.spawn(async move |this, cx| {
1682 cx.update(|cx| truncate.run(id.clone(), cx))?.await?;
1683 this.update(cx, |this, cx| {
1684 if let Some((ix, _)) = this.user_message_mut(&id) {
1685 let range = ix..this.entries.len();
1686 this.entries.truncate(ix);
1687 cx.emit(AcpThreadEvent::EntriesRemoved(range));
1688 }
1689 this.action_log()
1690 .update(cx, |action_log, cx| action_log.reject_all_edits(cx))
1691 })?
1692 .await;
1693 Ok(())
1694 })
1695 }
1696
1697 fn update_last_checkpoint(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1698 let git_store = self.project.read(cx).git_store().clone();
1699
1700 let old_checkpoint = if let Some((_, message)) = self.last_user_message() {
1701 if let Some(checkpoint) = message.checkpoint.as_ref() {
1702 checkpoint.git_checkpoint.clone()
1703 } else {
1704 return Task::ready(Ok(()));
1705 }
1706 } else {
1707 return Task::ready(Ok(()));
1708 };
1709
1710 let new_checkpoint = git_store.update(cx, |git, cx| git.checkpoint(cx));
1711 cx.spawn(async move |this, cx| {
1712 let new_checkpoint = new_checkpoint
1713 .await
1714 .context("failed to get new checkpoint")
1715 .log_err();
1716 if let Some(new_checkpoint) = new_checkpoint {
1717 let equal = git_store
1718 .update(cx, |git, cx| {
1719 git.compare_checkpoints(old_checkpoint.clone(), new_checkpoint, cx)
1720 })?
1721 .await
1722 .unwrap_or(true);
1723 this.update(cx, |this, cx| {
1724 let (ix, message) = this.last_user_message().context("no user message")?;
1725 let checkpoint = message.checkpoint.as_mut().context("no checkpoint")?;
1726 checkpoint.show = !equal;
1727 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1728 anyhow::Ok(())
1729 })??;
1730 }
1731
1732 Ok(())
1733 })
1734 }
1735
1736 fn last_user_message(&mut self) -> Option<(usize, &mut UserMessage)> {
1737 self.entries
1738 .iter_mut()
1739 .enumerate()
1740 .rev()
1741 .find_map(|(ix, entry)| {
1742 if let AgentThreadEntry::UserMessage(message) = entry {
1743 Some((ix, message))
1744 } else {
1745 None
1746 }
1747 })
1748 }
1749
1750 fn user_message_mut(&mut self, id: &UserMessageId) -> Option<(usize, &mut UserMessage)> {
1751 self.entries.iter_mut().enumerate().find_map(|(ix, entry)| {
1752 if let AgentThreadEntry::UserMessage(message) = entry {
1753 if message.id.as_ref() == Some(id) {
1754 Some((ix, message))
1755 } else {
1756 None
1757 }
1758 } else {
1759 None
1760 }
1761 })
1762 }
1763
1764 pub fn read_text_file(
1765 &self,
1766 path: PathBuf,
1767 line: Option<u32>,
1768 limit: Option<u32>,
1769 reuse_shared_snapshot: bool,
1770 cx: &mut Context<Self>,
1771 ) -> Task<Result<String>> {
1772 let project = self.project.clone();
1773 let action_log = self.action_log.clone();
1774 cx.spawn(async move |this, cx| {
1775 let load = project.update(cx, |project, cx| {
1776 let path = project
1777 .project_path_for_absolute_path(&path, cx)
1778 .context("invalid path")?;
1779 anyhow::Ok(project.open_buffer(path, cx))
1780 });
1781 let buffer = load??.await?;
1782
1783 let snapshot = if reuse_shared_snapshot {
1784 this.read_with(cx, |this, _| {
1785 this.shared_buffers.get(&buffer.clone()).cloned()
1786 })
1787 .log_err()
1788 .flatten()
1789 } else {
1790 None
1791 };
1792
1793 let snapshot = if let Some(snapshot) = snapshot {
1794 snapshot
1795 } else {
1796 action_log.update(cx, |action_log, cx| {
1797 action_log.buffer_read(buffer.clone(), cx);
1798 })?;
1799 project.update(cx, |project, cx| {
1800 let position = buffer
1801 .read(cx)
1802 .snapshot()
1803 .anchor_before(Point::new(line.unwrap_or_default(), 0));
1804 project.set_agent_location(
1805 Some(AgentLocation {
1806 buffer: buffer.downgrade(),
1807 position,
1808 }),
1809 cx,
1810 );
1811 })?;
1812
1813 buffer.update(cx, |buffer, _| buffer.snapshot())?
1814 };
1815
1816 this.update(cx, |this, _| {
1817 let text = snapshot.text();
1818 this.shared_buffers.insert(buffer.clone(), snapshot);
1819 if line.is_none() && limit.is_none() {
1820 return Ok(text);
1821 }
1822 let limit = limit.unwrap_or(u32::MAX) as usize;
1823 let Some(line) = line else {
1824 return Ok(text.lines().take(limit).collect::<String>());
1825 };
1826
1827 let count = text.lines().count();
1828 if count < line as usize {
1829 anyhow::bail!("There are only {} lines", count);
1830 }
1831 Ok(text
1832 .lines()
1833 .skip(line as usize + 1)
1834 .take(limit)
1835 .collect::<String>())
1836 })?
1837 })
1838 }
1839
1840 pub fn write_text_file(
1841 &self,
1842 path: PathBuf,
1843 content: String,
1844 cx: &mut Context<Self>,
1845 ) -> Task<Result<()>> {
1846 let project = self.project.clone();
1847 let action_log = self.action_log.clone();
1848 cx.spawn(async move |this, cx| {
1849 let load = project.update(cx, |project, cx| {
1850 let path = project
1851 .project_path_for_absolute_path(&path, cx)
1852 .context("invalid path")?;
1853 anyhow::Ok(project.open_buffer(path, cx))
1854 });
1855 let buffer = load??.await?;
1856 let snapshot = this.update(cx, |this, cx| {
1857 this.shared_buffers
1858 .get(&buffer)
1859 .cloned()
1860 .unwrap_or_else(|| buffer.read(cx).snapshot())
1861 })?;
1862 let edits = cx
1863 .background_executor()
1864 .spawn(async move {
1865 let old_text = snapshot.text();
1866 text_diff(old_text.as_str(), &content)
1867 .into_iter()
1868 .map(|(range, replacement)| {
1869 (
1870 snapshot.anchor_after(range.start)
1871 ..snapshot.anchor_before(range.end),
1872 replacement,
1873 )
1874 })
1875 .collect::<Vec<_>>()
1876 })
1877 .await;
1878
1879 project.update(cx, |project, cx| {
1880 project.set_agent_location(
1881 Some(AgentLocation {
1882 buffer: buffer.downgrade(),
1883 position: edits
1884 .last()
1885 .map(|(range, _)| range.end)
1886 .unwrap_or(Anchor::MIN),
1887 }),
1888 cx,
1889 );
1890 })?;
1891
1892 let format_on_save = cx.update(|cx| {
1893 action_log.update(cx, |action_log, cx| {
1894 action_log.buffer_read(buffer.clone(), cx);
1895 });
1896
1897 let format_on_save = buffer.update(cx, |buffer, cx| {
1898 buffer.edit(edits, None, cx);
1899
1900 let settings = language::language_settings::language_settings(
1901 buffer.language().map(|l| l.name()),
1902 buffer.file(),
1903 cx,
1904 );
1905
1906 settings.format_on_save != FormatOnSave::Off
1907 });
1908 action_log.update(cx, |action_log, cx| {
1909 action_log.buffer_edited(buffer.clone(), cx);
1910 });
1911 format_on_save
1912 })?;
1913
1914 if format_on_save {
1915 let format_task = project.update(cx, |project, cx| {
1916 project.format(
1917 HashSet::from_iter([buffer.clone()]),
1918 LspFormatTarget::Buffers,
1919 false,
1920 FormatTrigger::Save,
1921 cx,
1922 )
1923 })?;
1924 format_task.await.log_err();
1925
1926 action_log.update(cx, |action_log, cx| {
1927 action_log.buffer_edited(buffer.clone(), cx);
1928 })?;
1929 }
1930
1931 project
1932 .update(cx, |project, cx| project.save_buffer(buffer, cx))?
1933 .await
1934 })
1935 }
1936
1937 pub fn create_terminal(
1938 &self,
1939 mut command: String,
1940 args: Vec<String>,
1941 extra_env: Vec<acp::EnvVariable>,
1942 cwd: Option<PathBuf>,
1943 output_byte_limit: Option<u64>,
1944 cx: &mut Context<Self>,
1945 ) -> Task<Result<Entity<Terminal>>> {
1946 for arg in args {
1947 command.push(' ');
1948 command.push_str(&arg);
1949 }
1950
1951 let shell_command = if cfg!(windows) {
1952 format!("$null | & {{{}}}", command.replace("\"", "'"))
1953 } else if let Some(cwd) = cwd.as_ref().and_then(|cwd| cwd.as_os_str().to_str()) {
1954 // Make sure once we're *inside* the shell, we cd into `cwd`
1955 format!("(cd {cwd}; {}) </dev/null", command)
1956 } else {
1957 format!("({}) </dev/null", command)
1958 };
1959 let args = vec!["-c".into(), shell_command];
1960
1961 let env = match &cwd {
1962 Some(dir) => self.project.update(cx, |project, cx| {
1963 project.directory_environment(dir.as_path().into(), cx)
1964 }),
1965 None => Task::ready(None).shared(),
1966 };
1967
1968 let env = cx.spawn(async move |_, _| {
1969 let mut env = env.await.unwrap_or_default();
1970 if cfg!(unix) {
1971 env.insert("PAGER".into(), "cat".into());
1972 }
1973 for var in extra_env {
1974 env.insert(var.name, var.value);
1975 }
1976 env
1977 });
1978
1979 let project = self.project.clone();
1980 let language_registry = project.read(cx).languages().clone();
1981 let determine_shell = self.determine_shell.clone();
1982
1983 let terminal_id = acp::TerminalId(Uuid::new_v4().to_string().into());
1984 let terminal_task = cx.spawn({
1985 let terminal_id = terminal_id.clone();
1986 async move |_this, cx| {
1987 let program = determine_shell.await;
1988 let env = env.await;
1989 let terminal = project
1990 .update(cx, |project, cx| {
1991 project.create_terminal_task(
1992 task::SpawnInTerminal {
1993 command: Some(program),
1994 args,
1995 cwd: cwd.clone(),
1996 env,
1997 ..Default::default()
1998 },
1999 cx,
2000 )
2001 })?
2002 .await?;
2003
2004 cx.new(|cx| {
2005 Terminal::new(
2006 terminal_id,
2007 command,
2008 cwd,
2009 output_byte_limit.map(|l| l as usize),
2010 terminal,
2011 language_registry,
2012 cx,
2013 )
2014 })
2015 }
2016 });
2017
2018 cx.spawn(async move |this, cx| {
2019 let terminal = terminal_task.await?;
2020 this.update(cx, |this, _cx| {
2021 this.terminals.insert(terminal_id, terminal.clone());
2022 terminal
2023 })
2024 })
2025 }
2026
2027 pub fn kill_terminal(
2028 &mut self,
2029 terminal_id: acp::TerminalId,
2030 cx: &mut Context<Self>,
2031 ) -> Result<()> {
2032 self.terminals
2033 .get(&terminal_id)
2034 .context("Terminal not found")?
2035 .update(cx, |terminal, cx| {
2036 terminal.kill(cx);
2037 });
2038
2039 Ok(())
2040 }
2041
2042 pub fn release_terminal(
2043 &mut self,
2044 terminal_id: acp::TerminalId,
2045 cx: &mut Context<Self>,
2046 ) -> Result<()> {
2047 self.terminals
2048 .remove(&terminal_id)
2049 .context("Terminal not found")?
2050 .update(cx, |terminal, cx| {
2051 terminal.kill(cx);
2052 });
2053
2054 Ok(())
2055 }
2056
2057 pub fn terminal(&self, terminal_id: acp::TerminalId) -> Result<Entity<Terminal>> {
2058 self.terminals
2059 .get(&terminal_id)
2060 .context("Terminal not found")
2061 .cloned()
2062 }
2063
2064 pub fn to_markdown(&self, cx: &App) -> String {
2065 self.entries.iter().map(|e| e.to_markdown(cx)).collect()
2066 }
2067
2068 pub fn emit_load_error(&mut self, error: LoadError, cx: &mut Context<Self>) {
2069 cx.emit(AcpThreadEvent::LoadError(error));
2070 }
2071}
2072
2073fn markdown_for_raw_output(
2074 raw_output: &serde_json::Value,
2075 language_registry: &Arc<LanguageRegistry>,
2076 cx: &mut App,
2077) -> Option<Entity<Markdown>> {
2078 match raw_output {
2079 serde_json::Value::Null => None,
2080 serde_json::Value::Bool(value) => Some(cx.new(|cx| {
2081 Markdown::new(
2082 value.to_string().into(),
2083 Some(language_registry.clone()),
2084 None,
2085 cx,
2086 )
2087 })),
2088 serde_json::Value::Number(value) => Some(cx.new(|cx| {
2089 Markdown::new(
2090 value.to_string().into(),
2091 Some(language_registry.clone()),
2092 None,
2093 cx,
2094 )
2095 })),
2096 serde_json::Value::String(value) => Some(cx.new(|cx| {
2097 Markdown::new(
2098 value.clone().into(),
2099 Some(language_registry.clone()),
2100 None,
2101 cx,
2102 )
2103 })),
2104 value => Some(cx.new(|cx| {
2105 Markdown::new(
2106 format!("```json\n{}\n```", value).into(),
2107 Some(language_registry.clone()),
2108 None,
2109 cx,
2110 )
2111 })),
2112 }
2113}
2114
2115#[cfg(test)]
2116mod tests {
2117 use super::*;
2118 use anyhow::anyhow;
2119 use futures::{channel::mpsc, future::LocalBoxFuture, select};
2120 use gpui::{App, AsyncApp, TestAppContext, WeakEntity};
2121 use indoc::indoc;
2122 use project::{FakeFs, Fs};
2123 use rand::{distr, prelude::*};
2124 use serde_json::json;
2125 use settings::SettingsStore;
2126 use smol::stream::StreamExt as _;
2127 use std::{
2128 any::Any,
2129 cell::RefCell,
2130 path::Path,
2131 rc::Rc,
2132 sync::atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
2133 time::Duration,
2134 };
2135 use util::path;
2136
2137 fn init_test(cx: &mut TestAppContext) {
2138 env_logger::try_init().ok();
2139 cx.update(|cx| {
2140 let settings_store = SettingsStore::test(cx);
2141 cx.set_global(settings_store);
2142 Project::init_settings(cx);
2143 language::init(cx);
2144 });
2145 }
2146
2147 #[gpui::test]
2148 async fn test_push_user_content_block(cx: &mut gpui::TestAppContext) {
2149 init_test(cx);
2150
2151 let fs = FakeFs::new(cx.executor());
2152 let project = Project::test(fs, [], cx).await;
2153 let connection = Rc::new(FakeAgentConnection::new());
2154 let thread = cx
2155 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2156 .await
2157 .unwrap();
2158
2159 // Test creating a new user message
2160 thread.update(cx, |thread, cx| {
2161 thread.push_user_content_block(
2162 None,
2163 acp::ContentBlock::Text(acp::TextContent {
2164 annotations: None,
2165 text: "Hello, ".to_string(),
2166 }),
2167 cx,
2168 );
2169 });
2170
2171 thread.update(cx, |thread, cx| {
2172 assert_eq!(thread.entries.len(), 1);
2173 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
2174 assert_eq!(user_msg.id, None);
2175 assert_eq!(user_msg.content.to_markdown(cx), "Hello, ");
2176 } else {
2177 panic!("Expected UserMessage");
2178 }
2179 });
2180
2181 // Test appending to existing user message
2182 let message_1_id = UserMessageId::new();
2183 thread.update(cx, |thread, cx| {
2184 thread.push_user_content_block(
2185 Some(message_1_id.clone()),
2186 acp::ContentBlock::Text(acp::TextContent {
2187 annotations: None,
2188 text: "world!".to_string(),
2189 }),
2190 cx,
2191 );
2192 });
2193
2194 thread.update(cx, |thread, cx| {
2195 assert_eq!(thread.entries.len(), 1);
2196 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
2197 assert_eq!(user_msg.id, Some(message_1_id));
2198 assert_eq!(user_msg.content.to_markdown(cx), "Hello, world!");
2199 } else {
2200 panic!("Expected UserMessage");
2201 }
2202 });
2203
2204 // Test creating new user message after assistant message
2205 thread.update(cx, |thread, cx| {
2206 thread.push_assistant_content_block(
2207 acp::ContentBlock::Text(acp::TextContent {
2208 annotations: None,
2209 text: "Assistant response".to_string(),
2210 }),
2211 false,
2212 cx,
2213 );
2214 });
2215
2216 let message_2_id = UserMessageId::new();
2217 thread.update(cx, |thread, cx| {
2218 thread.push_user_content_block(
2219 Some(message_2_id.clone()),
2220 acp::ContentBlock::Text(acp::TextContent {
2221 annotations: None,
2222 text: "New user message".to_string(),
2223 }),
2224 cx,
2225 );
2226 });
2227
2228 thread.update(cx, |thread, cx| {
2229 assert_eq!(thread.entries.len(), 3);
2230 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[2] {
2231 assert_eq!(user_msg.id, Some(message_2_id));
2232 assert_eq!(user_msg.content.to_markdown(cx), "New user message");
2233 } else {
2234 panic!("Expected UserMessage at index 2");
2235 }
2236 });
2237 }
2238
2239 #[gpui::test]
2240 async fn test_thinking_concatenation(cx: &mut gpui::TestAppContext) {
2241 init_test(cx);
2242
2243 let fs = FakeFs::new(cx.executor());
2244 let project = Project::test(fs, [], cx).await;
2245 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
2246 |_, thread, mut cx| {
2247 async move {
2248 thread.update(&mut cx, |thread, cx| {
2249 thread
2250 .handle_session_update(
2251 acp::SessionUpdate::AgentThoughtChunk {
2252 content: "Thinking ".into(),
2253 },
2254 cx,
2255 )
2256 .unwrap();
2257 thread
2258 .handle_session_update(
2259 acp::SessionUpdate::AgentThoughtChunk {
2260 content: "hard!".into(),
2261 },
2262 cx,
2263 )
2264 .unwrap();
2265 })?;
2266 Ok(acp::PromptResponse {
2267 stop_reason: acp::StopReason::EndTurn,
2268 })
2269 }
2270 .boxed_local()
2271 },
2272 ));
2273
2274 let thread = cx
2275 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2276 .await
2277 .unwrap();
2278
2279 thread
2280 .update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
2281 .await
2282 .unwrap();
2283
2284 let output = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
2285 assert_eq!(
2286 output,
2287 indoc! {r#"
2288 ## User
2289
2290 Hello from Zed!
2291
2292 ## Assistant
2293
2294 <thinking>
2295 Thinking hard!
2296 </thinking>
2297
2298 "#}
2299 );
2300 }
2301
2302 #[gpui::test]
2303 async fn test_edits_concurrently_to_user(cx: &mut TestAppContext) {
2304 init_test(cx);
2305
2306 let fs = FakeFs::new(cx.executor());
2307 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\n"}))
2308 .await;
2309 let project = Project::test(fs.clone(), [], cx).await;
2310 let (read_file_tx, read_file_rx) = oneshot::channel::<()>();
2311 let read_file_tx = Rc::new(RefCell::new(Some(read_file_tx)));
2312 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
2313 move |_, thread, mut cx| {
2314 let read_file_tx = read_file_tx.clone();
2315 async move {
2316 let content = thread
2317 .update(&mut cx, |thread, cx| {
2318 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
2319 })
2320 .unwrap()
2321 .await
2322 .unwrap();
2323 assert_eq!(content, "one\ntwo\nthree\n");
2324 read_file_tx.take().unwrap().send(()).unwrap();
2325 thread
2326 .update(&mut cx, |thread, cx| {
2327 thread.write_text_file(
2328 path!("/tmp/foo").into(),
2329 "one\ntwo\nthree\nfour\nfive\n".to_string(),
2330 cx,
2331 )
2332 })
2333 .unwrap()
2334 .await
2335 .unwrap();
2336 Ok(acp::PromptResponse {
2337 stop_reason: acp::StopReason::EndTurn,
2338 })
2339 }
2340 .boxed_local()
2341 },
2342 ));
2343
2344 let (worktree, pathbuf) = project
2345 .update(cx, |project, cx| {
2346 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
2347 })
2348 .await
2349 .unwrap();
2350 let buffer = project
2351 .update(cx, |project, cx| {
2352 project.open_buffer((worktree.read(cx).id(), pathbuf), cx)
2353 })
2354 .await
2355 .unwrap();
2356
2357 let thread = cx
2358 .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
2359 .await
2360 .unwrap();
2361
2362 let request = thread.update(cx, |thread, cx| {
2363 thread.send_raw("Extend the count in /tmp/foo", cx)
2364 });
2365 read_file_rx.await.ok();
2366 buffer.update(cx, |buffer, cx| {
2367 buffer.edit([(0..0, "zero\n".to_string())], None, cx);
2368 });
2369 cx.run_until_parked();
2370 assert_eq!(
2371 buffer.read_with(cx, |buffer, _| buffer.text()),
2372 "zero\none\ntwo\nthree\nfour\nfive\n"
2373 );
2374 assert_eq!(
2375 String::from_utf8(fs.read_file_sync(path!("/tmp/foo")).unwrap()).unwrap(),
2376 "zero\none\ntwo\nthree\nfour\nfive\n"
2377 );
2378 request.await.unwrap();
2379 }
2380
2381 #[gpui::test]
2382 async fn test_succeeding_canceled_toolcall(cx: &mut TestAppContext) {
2383 init_test(cx);
2384
2385 let fs = FakeFs::new(cx.executor());
2386 let project = Project::test(fs, [], cx).await;
2387 let id = acp::ToolCallId("test".into());
2388
2389 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2390 let id = id.clone();
2391 move |_, thread, mut cx| {
2392 let id = id.clone();
2393 async move {
2394 thread
2395 .update(&mut cx, |thread, cx| {
2396 thread.handle_session_update(
2397 acp::SessionUpdate::ToolCall(acp::ToolCall {
2398 id: id.clone(),
2399 title: "Label".into(),
2400 kind: acp::ToolKind::Fetch,
2401 status: acp::ToolCallStatus::InProgress,
2402 content: vec![],
2403 locations: vec![],
2404 raw_input: None,
2405 raw_output: None,
2406 }),
2407 cx,
2408 )
2409 })
2410 .unwrap()
2411 .unwrap();
2412 Ok(acp::PromptResponse {
2413 stop_reason: acp::StopReason::EndTurn,
2414 })
2415 }
2416 .boxed_local()
2417 }
2418 }));
2419
2420 let thread = cx
2421 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2422 .await
2423 .unwrap();
2424
2425 let request = thread.update(cx, |thread, cx| {
2426 thread.send_raw("Fetch https://example.com", cx)
2427 });
2428
2429 run_until_first_tool_call(&thread, cx).await;
2430
2431 thread.read_with(cx, |thread, _| {
2432 assert!(matches!(
2433 thread.entries[1],
2434 AgentThreadEntry::ToolCall(ToolCall {
2435 status: ToolCallStatus::InProgress,
2436 ..
2437 })
2438 ));
2439 });
2440
2441 thread.update(cx, |thread, cx| thread.cancel(cx)).await;
2442
2443 thread.read_with(cx, |thread, _| {
2444 assert!(matches!(
2445 &thread.entries[1],
2446 AgentThreadEntry::ToolCall(ToolCall {
2447 status: ToolCallStatus::Canceled,
2448 ..
2449 })
2450 ));
2451 });
2452
2453 thread
2454 .update(cx, |thread, cx| {
2455 thread.handle_session_update(
2456 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate {
2457 id,
2458 fields: acp::ToolCallUpdateFields {
2459 status: Some(acp::ToolCallStatus::Completed),
2460 ..Default::default()
2461 },
2462 }),
2463 cx,
2464 )
2465 })
2466 .unwrap();
2467
2468 request.await.unwrap();
2469
2470 thread.read_with(cx, |thread, _| {
2471 assert!(matches!(
2472 thread.entries[1],
2473 AgentThreadEntry::ToolCall(ToolCall {
2474 status: ToolCallStatus::Completed,
2475 ..
2476 })
2477 ));
2478 });
2479 }
2480
2481 #[gpui::test]
2482 async fn test_no_pending_edits_if_tool_calls_are_completed(cx: &mut TestAppContext) {
2483 init_test(cx);
2484 let fs = FakeFs::new(cx.background_executor.clone());
2485 fs.insert_tree(path!("/test"), json!({})).await;
2486 let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
2487
2488 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2489 move |_, thread, mut cx| {
2490 async move {
2491 thread
2492 .update(&mut cx, |thread, cx| {
2493 thread.handle_session_update(
2494 acp::SessionUpdate::ToolCall(acp::ToolCall {
2495 id: acp::ToolCallId("test".into()),
2496 title: "Label".into(),
2497 kind: acp::ToolKind::Edit,
2498 status: acp::ToolCallStatus::Completed,
2499 content: vec![acp::ToolCallContent::Diff {
2500 diff: acp::Diff {
2501 path: "/test/test.txt".into(),
2502 old_text: None,
2503 new_text: "foo".into(),
2504 },
2505 }],
2506 locations: vec![],
2507 raw_input: None,
2508 raw_output: None,
2509 }),
2510 cx,
2511 )
2512 })
2513 .unwrap()
2514 .unwrap();
2515 Ok(acp::PromptResponse {
2516 stop_reason: acp::StopReason::EndTurn,
2517 })
2518 }
2519 .boxed_local()
2520 }
2521 }));
2522
2523 let thread = cx
2524 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2525 .await
2526 .unwrap();
2527
2528 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Hi".into()], cx)))
2529 .await
2530 .unwrap();
2531
2532 assert!(cx.read(|cx| !thread.read(cx).has_pending_edit_tool_calls()));
2533 }
2534
2535 #[gpui::test(iterations = 10)]
2536 async fn test_checkpoints(cx: &mut TestAppContext) {
2537 init_test(cx);
2538 let fs = FakeFs::new(cx.background_executor.clone());
2539 fs.insert_tree(
2540 path!("/test"),
2541 json!({
2542 ".git": {}
2543 }),
2544 )
2545 .await;
2546 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
2547
2548 let simulate_changes = Arc::new(AtomicBool::new(true));
2549 let next_filename = Arc::new(AtomicUsize::new(0));
2550 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2551 let simulate_changes = simulate_changes.clone();
2552 let next_filename = next_filename.clone();
2553 let fs = fs.clone();
2554 move |request, thread, mut cx| {
2555 let fs = fs.clone();
2556 let simulate_changes = simulate_changes.clone();
2557 let next_filename = next_filename.clone();
2558 async move {
2559 if simulate_changes.load(SeqCst) {
2560 let filename = format!("/test/file-{}", next_filename.fetch_add(1, SeqCst));
2561 fs.write(Path::new(&filename), b"").await?;
2562 }
2563
2564 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
2565 panic!("expected text content block");
2566 };
2567 thread.update(&mut cx, |thread, cx| {
2568 thread
2569 .handle_session_update(
2570 acp::SessionUpdate::AgentMessageChunk {
2571 content: content.text.to_uppercase().into(),
2572 },
2573 cx,
2574 )
2575 .unwrap();
2576 })?;
2577 Ok(acp::PromptResponse {
2578 stop_reason: acp::StopReason::EndTurn,
2579 })
2580 }
2581 .boxed_local()
2582 }
2583 }));
2584 let thread = cx
2585 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2586 .await
2587 .unwrap();
2588
2589 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Lorem".into()], cx)))
2590 .await
2591 .unwrap();
2592 thread.read_with(cx, |thread, cx| {
2593 assert_eq!(
2594 thread.to_markdown(cx),
2595 indoc! {"
2596 ## User (checkpoint)
2597
2598 Lorem
2599
2600 ## Assistant
2601
2602 LOREM
2603
2604 "}
2605 );
2606 });
2607 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
2608
2609 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["ipsum".into()], cx)))
2610 .await
2611 .unwrap();
2612 thread.read_with(cx, |thread, cx| {
2613 assert_eq!(
2614 thread.to_markdown(cx),
2615 indoc! {"
2616 ## User (checkpoint)
2617
2618 Lorem
2619
2620 ## Assistant
2621
2622 LOREM
2623
2624 ## User (checkpoint)
2625
2626 ipsum
2627
2628 ## Assistant
2629
2630 IPSUM
2631
2632 "}
2633 );
2634 });
2635 assert_eq!(
2636 fs.files(),
2637 vec![
2638 Path::new(path!("/test/file-0")),
2639 Path::new(path!("/test/file-1"))
2640 ]
2641 );
2642
2643 // Checkpoint isn't stored when there are no changes.
2644 simulate_changes.store(false, SeqCst);
2645 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["dolor".into()], cx)))
2646 .await
2647 .unwrap();
2648 thread.read_with(cx, |thread, cx| {
2649 assert_eq!(
2650 thread.to_markdown(cx),
2651 indoc! {"
2652 ## User (checkpoint)
2653
2654 Lorem
2655
2656 ## Assistant
2657
2658 LOREM
2659
2660 ## User (checkpoint)
2661
2662 ipsum
2663
2664 ## Assistant
2665
2666 IPSUM
2667
2668 ## User
2669
2670 dolor
2671
2672 ## Assistant
2673
2674 DOLOR
2675
2676 "}
2677 );
2678 });
2679 assert_eq!(
2680 fs.files(),
2681 vec![
2682 Path::new(path!("/test/file-0")),
2683 Path::new(path!("/test/file-1"))
2684 ]
2685 );
2686
2687 // Rewinding the conversation truncates the history and restores the checkpoint.
2688 thread
2689 .update(cx, |thread, cx| {
2690 let AgentThreadEntry::UserMessage(message) = &thread.entries[2] else {
2691 panic!("unexpected entries {:?}", thread.entries)
2692 };
2693 thread.restore_checkpoint(message.id.clone().unwrap(), cx)
2694 })
2695 .await
2696 .unwrap();
2697 thread.read_with(cx, |thread, cx| {
2698 assert_eq!(
2699 thread.to_markdown(cx),
2700 indoc! {"
2701 ## User (checkpoint)
2702
2703 Lorem
2704
2705 ## Assistant
2706
2707 LOREM
2708
2709 "}
2710 );
2711 });
2712 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
2713 }
2714
2715 #[gpui::test]
2716 async fn test_tool_result_refusal(cx: &mut TestAppContext) {
2717 use std::sync::atomic::AtomicUsize;
2718 init_test(cx);
2719
2720 let fs = FakeFs::new(cx.executor());
2721 let project = Project::test(fs, None, cx).await;
2722
2723 // Create a connection that simulates refusal after tool result
2724 let prompt_count = Arc::new(AtomicUsize::new(0));
2725 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2726 let prompt_count = prompt_count.clone();
2727 move |_request, thread, mut cx| {
2728 let count = prompt_count.fetch_add(1, SeqCst);
2729 async move {
2730 if count == 0 {
2731 // First prompt: Generate a tool call with result
2732 thread.update(&mut cx, |thread, cx| {
2733 thread
2734 .handle_session_update(
2735 acp::SessionUpdate::ToolCall(acp::ToolCall {
2736 id: acp::ToolCallId("tool1".into()),
2737 title: "Test Tool".into(),
2738 kind: acp::ToolKind::Fetch,
2739 status: acp::ToolCallStatus::Completed,
2740 content: vec![],
2741 locations: vec![],
2742 raw_input: Some(serde_json::json!({"query": "test"})),
2743 raw_output: Some(
2744 serde_json::json!({"result": "inappropriate content"}),
2745 ),
2746 }),
2747 cx,
2748 )
2749 .unwrap();
2750 })?;
2751
2752 // Now return refusal because of the tool result
2753 Ok(acp::PromptResponse {
2754 stop_reason: acp::StopReason::Refusal,
2755 })
2756 } else {
2757 Ok(acp::PromptResponse {
2758 stop_reason: acp::StopReason::EndTurn,
2759 })
2760 }
2761 }
2762 .boxed_local()
2763 }
2764 }));
2765
2766 let thread = cx
2767 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2768 .await
2769 .unwrap();
2770
2771 // Track if we see a Refusal event
2772 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
2773 let saw_refusal_event_captured = saw_refusal_event.clone();
2774 thread.update(cx, |_thread, cx| {
2775 cx.subscribe(
2776 &thread,
2777 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
2778 if matches!(event, AcpThreadEvent::Refusal) {
2779 *saw_refusal_event_captured.lock().unwrap() = true;
2780 }
2781 },
2782 )
2783 .detach();
2784 });
2785
2786 // Send a user message - this will trigger tool call and then refusal
2787 let send_task = thread.update(cx, |thread, cx| {
2788 thread.send(
2789 vec![acp::ContentBlock::Text(acp::TextContent {
2790 text: "Hello".into(),
2791 annotations: None,
2792 })],
2793 cx,
2794 )
2795 });
2796 cx.background_executor.spawn(send_task).detach();
2797 cx.run_until_parked();
2798
2799 // Verify that:
2800 // 1. A Refusal event WAS emitted (because it's a tool result refusal, not user prompt)
2801 // 2. The user message was NOT truncated
2802 assert!(
2803 *saw_refusal_event.lock().unwrap(),
2804 "Refusal event should be emitted for tool result refusals"
2805 );
2806
2807 thread.read_with(cx, |thread, _| {
2808 let entries = thread.entries();
2809 assert!(entries.len() >= 2, "Should have user message and tool call");
2810
2811 // Verify user message is still there
2812 assert!(
2813 matches!(entries[0], AgentThreadEntry::UserMessage(_)),
2814 "User message should not be truncated"
2815 );
2816
2817 // Verify tool call is there with result
2818 if let AgentThreadEntry::ToolCall(tool_call) = &entries[1] {
2819 assert!(
2820 tool_call.raw_output.is_some(),
2821 "Tool call should have output"
2822 );
2823 } else {
2824 panic!("Expected tool call at index 1");
2825 }
2826 });
2827 }
2828
2829 #[gpui::test]
2830 async fn test_user_prompt_refusal_emits_event(cx: &mut TestAppContext) {
2831 init_test(cx);
2832
2833 let fs = FakeFs::new(cx.executor());
2834 let project = Project::test(fs, None, cx).await;
2835
2836 let refuse_next = Arc::new(AtomicBool::new(false));
2837 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2838 let refuse_next = refuse_next.clone();
2839 move |_request, _thread, _cx| {
2840 if refuse_next.load(SeqCst) {
2841 async move {
2842 Ok(acp::PromptResponse {
2843 stop_reason: acp::StopReason::Refusal,
2844 })
2845 }
2846 .boxed_local()
2847 } else {
2848 async move {
2849 Ok(acp::PromptResponse {
2850 stop_reason: acp::StopReason::EndTurn,
2851 })
2852 }
2853 .boxed_local()
2854 }
2855 }
2856 }));
2857
2858 let thread = cx
2859 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2860 .await
2861 .unwrap();
2862
2863 // Track if we see a Refusal event
2864 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
2865 let saw_refusal_event_captured = saw_refusal_event.clone();
2866 thread.update(cx, |_thread, cx| {
2867 cx.subscribe(
2868 &thread,
2869 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
2870 if matches!(event, AcpThreadEvent::Refusal) {
2871 *saw_refusal_event_captured.lock().unwrap() = true;
2872 }
2873 },
2874 )
2875 .detach();
2876 });
2877
2878 // Send a message that will be refused
2879 refuse_next.store(true, SeqCst);
2880 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
2881 .await
2882 .unwrap();
2883
2884 // Verify that a Refusal event WAS emitted for user prompt refusal
2885 assert!(
2886 *saw_refusal_event.lock().unwrap(),
2887 "Refusal event should be emitted for user prompt refusals"
2888 );
2889
2890 // Verify the message was truncated (user prompt refusal)
2891 thread.read_with(cx, |thread, cx| {
2892 assert_eq!(thread.to_markdown(cx), "");
2893 });
2894 }
2895
2896 #[gpui::test]
2897 async fn test_refusal(cx: &mut TestAppContext) {
2898 init_test(cx);
2899 let fs = FakeFs::new(cx.background_executor.clone());
2900 fs.insert_tree(path!("/"), json!({})).await;
2901 let project = Project::test(fs.clone(), [path!("/").as_ref()], cx).await;
2902
2903 let refuse_next = Arc::new(AtomicBool::new(false));
2904 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2905 let refuse_next = refuse_next.clone();
2906 move |request, thread, mut cx| {
2907 let refuse_next = refuse_next.clone();
2908 async move {
2909 if refuse_next.load(SeqCst) {
2910 return Ok(acp::PromptResponse {
2911 stop_reason: acp::StopReason::Refusal,
2912 });
2913 }
2914
2915 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
2916 panic!("expected text content block");
2917 };
2918 thread.update(&mut cx, |thread, cx| {
2919 thread
2920 .handle_session_update(
2921 acp::SessionUpdate::AgentMessageChunk {
2922 content: content.text.to_uppercase().into(),
2923 },
2924 cx,
2925 )
2926 .unwrap();
2927 })?;
2928 Ok(acp::PromptResponse {
2929 stop_reason: acp::StopReason::EndTurn,
2930 })
2931 }
2932 .boxed_local()
2933 }
2934 }));
2935 let thread = cx
2936 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2937 .await
2938 .unwrap();
2939
2940 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
2941 .await
2942 .unwrap();
2943 thread.read_with(cx, |thread, cx| {
2944 assert_eq!(
2945 thread.to_markdown(cx),
2946 indoc! {"
2947 ## User
2948
2949 hello
2950
2951 ## Assistant
2952
2953 HELLO
2954
2955 "}
2956 );
2957 });
2958
2959 // Simulate refusing the second message. The message should be truncated
2960 // when a user prompt is refused.
2961 refuse_next.store(true, SeqCst);
2962 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["world".into()], cx)))
2963 .await
2964 .unwrap();
2965 thread.read_with(cx, |thread, cx| {
2966 assert_eq!(
2967 thread.to_markdown(cx),
2968 indoc! {"
2969 ## User
2970
2971 hello
2972
2973 ## Assistant
2974
2975 HELLO
2976
2977 "}
2978 );
2979 });
2980 }
2981
2982 async fn run_until_first_tool_call(
2983 thread: &Entity<AcpThread>,
2984 cx: &mut TestAppContext,
2985 ) -> usize {
2986 let (mut tx, mut rx) = mpsc::channel::<usize>(1);
2987
2988 let subscription = cx.update(|cx| {
2989 cx.subscribe(thread, move |thread, _, cx| {
2990 for (ix, entry) in thread.read(cx).entries.iter().enumerate() {
2991 if matches!(entry, AgentThreadEntry::ToolCall(_)) {
2992 return tx.try_send(ix).unwrap();
2993 }
2994 }
2995 })
2996 });
2997
2998 select! {
2999 _ = futures::FutureExt::fuse(smol::Timer::after(Duration::from_secs(10))) => {
3000 panic!("Timeout waiting for tool call")
3001 }
3002 ix = rx.next().fuse() => {
3003 drop(subscription);
3004 ix.unwrap()
3005 }
3006 }
3007 }
3008
3009 #[derive(Clone, Default)]
3010 struct FakeAgentConnection {
3011 auth_methods: Vec<acp::AuthMethod>,
3012 sessions: Arc<parking_lot::Mutex<HashMap<acp::SessionId, WeakEntity<AcpThread>>>>,
3013 on_user_message: Option<
3014 Rc<
3015 dyn Fn(
3016 acp::PromptRequest,
3017 WeakEntity<AcpThread>,
3018 AsyncApp,
3019 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
3020 + 'static,
3021 >,
3022 >,
3023 }
3024
3025 impl FakeAgentConnection {
3026 fn new() -> Self {
3027 Self {
3028 auth_methods: Vec::new(),
3029 on_user_message: None,
3030 sessions: Arc::default(),
3031 }
3032 }
3033
3034 #[expect(unused)]
3035 fn with_auth_methods(mut self, auth_methods: Vec<acp::AuthMethod>) -> Self {
3036 self.auth_methods = auth_methods;
3037 self
3038 }
3039
3040 fn on_user_message(
3041 mut self,
3042 handler: impl Fn(
3043 acp::PromptRequest,
3044 WeakEntity<AcpThread>,
3045 AsyncApp,
3046 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
3047 + 'static,
3048 ) -> Self {
3049 self.on_user_message.replace(Rc::new(handler));
3050 self
3051 }
3052 }
3053
3054 impl AgentConnection for FakeAgentConnection {
3055 fn auth_methods(&self) -> &[acp::AuthMethod] {
3056 &self.auth_methods
3057 }
3058
3059 fn new_thread(
3060 self: Rc<Self>,
3061 project: Entity<Project>,
3062 _cwd: &Path,
3063 cx: &mut App,
3064 ) -> Task<gpui::Result<Entity<AcpThread>>> {
3065 let session_id = acp::SessionId(
3066 rand::rng()
3067 .sample_iter(&distr::Alphanumeric)
3068 .take(7)
3069 .map(char::from)
3070 .collect::<String>()
3071 .into(),
3072 );
3073 let action_log = cx.new(|_| ActionLog::new(project.clone()));
3074 let thread = cx.new(|cx| {
3075 AcpThread::new(
3076 "Test",
3077 self.clone(),
3078 project,
3079 action_log,
3080 session_id.clone(),
3081 watch::Receiver::constant(acp::PromptCapabilities {
3082 image: true,
3083 audio: true,
3084 embedded_context: true,
3085 }),
3086 cx,
3087 )
3088 });
3089 self.sessions.lock().insert(session_id, thread.downgrade());
3090 Task::ready(Ok(thread))
3091 }
3092
3093 fn authenticate(&self, method: acp::AuthMethodId, _cx: &mut App) -> Task<gpui::Result<()>> {
3094 if self.auth_methods().iter().any(|m| m.id == method) {
3095 Task::ready(Ok(()))
3096 } else {
3097 Task::ready(Err(anyhow!("Invalid Auth Method")))
3098 }
3099 }
3100
3101 fn prompt(
3102 &self,
3103 _id: Option<UserMessageId>,
3104 params: acp::PromptRequest,
3105 cx: &mut App,
3106 ) -> Task<gpui::Result<acp::PromptResponse>> {
3107 let sessions = self.sessions.lock();
3108 let thread = sessions.get(¶ms.session_id).unwrap();
3109 if let Some(handler) = &self.on_user_message {
3110 let handler = handler.clone();
3111 let thread = thread.clone();
3112 cx.spawn(async move |cx| handler(params, thread, cx.clone()).await)
3113 } else {
3114 Task::ready(Ok(acp::PromptResponse {
3115 stop_reason: acp::StopReason::EndTurn,
3116 }))
3117 }
3118 }
3119
3120 fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
3121 let sessions = self.sessions.lock();
3122 let thread = sessions.get(session_id).unwrap().clone();
3123
3124 cx.spawn(async move |cx| {
3125 thread
3126 .update(cx, |thread, cx| thread.cancel(cx))
3127 .unwrap()
3128 .await
3129 })
3130 .detach();
3131 }
3132
3133 fn truncate(
3134 &self,
3135 session_id: &acp::SessionId,
3136 _cx: &App,
3137 ) -> Option<Rc<dyn AgentSessionTruncate>> {
3138 Some(Rc::new(FakeAgentSessionEditor {
3139 _session_id: session_id.clone(),
3140 }))
3141 }
3142
3143 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
3144 self
3145 }
3146 }
3147
3148 struct FakeAgentSessionEditor {
3149 _session_id: acp::SessionId,
3150 }
3151
3152 impl AgentSessionTruncate for FakeAgentSessionEditor {
3153 fn run(&self, _message_id: UserMessageId, _cx: &mut App) -> Task<Result<()>> {
3154 Task::ready(Ok(()))
3155 }
3156 }
3157}