1mod connection;
2mod diff;
3mod mention;
4mod terminal;
5
6use agent_settings::AgentSettings;
7
8/// Key used in ACP ToolCall meta to store the tool's programmatic name.
9/// This is a workaround since ACP's ToolCall doesn't have a dedicated name field.
10pub const TOOL_NAME_META_KEY: &str = "tool_name";
11
12/// The tool name for subagent spawning
13pub const SUBAGENT_TOOL_NAME: &str = "subagent";
14
15/// Helper to extract tool name from ACP meta
16pub fn tool_name_from_meta(meta: &Option<acp::Meta>) -> Option<SharedString> {
17 meta.as_ref()
18 .and_then(|m| m.get(TOOL_NAME_META_KEY))
19 .and_then(|v| v.as_str())
20 .map(|s| SharedString::from(s.to_owned()))
21}
22
23/// Helper to create meta with tool name
24pub fn meta_with_tool_name(tool_name: &str) -> acp::Meta {
25 acp::Meta::from_iter([(TOOL_NAME_META_KEY.into(), tool_name.into())])
26}
27use collections::HashSet;
28pub use connection::*;
29pub use diff::*;
30use language::language_settings::FormatOnSave;
31pub use mention::*;
32use project::lsp_store::{FormatTrigger, LspFormatTarget};
33use serde::{Deserialize, Serialize};
34use serde_json::to_string_pretty;
35use settings::Settings as _;
36use task::{Shell, ShellBuilder};
37pub use terminal::*;
38
39use action_log::{ActionLog, ActionLogTelemetry};
40use agent_client_protocol::{self as acp};
41use anyhow::{Context as _, Result, anyhow};
42use editor::Bias;
43use futures::{FutureExt, channel::oneshot, future::BoxFuture};
44use gpui::{AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Task, WeakEntity};
45use itertools::Itertools;
46use language::{Anchor, Buffer, BufferSnapshot, LanguageRegistry, Point, ToPoint, text_diff};
47use markdown::Markdown;
48use project::{AgentLocation, Project, git_store::GitStoreCheckpoint};
49use std::collections::HashMap;
50use std::error::Error;
51use std::fmt::{Formatter, Write};
52use std::ops::Range;
53use std::process::ExitStatus;
54use std::rc::Rc;
55use std::time::{Duration, Instant};
56use std::{fmt::Display, mem, path::PathBuf, sync::Arc};
57use ui::App;
58use util::{ResultExt, get_default_system_shell_preferring_bash, paths::PathStyle};
59use uuid::Uuid;
60
61#[derive(Debug)]
62pub struct UserMessage {
63 pub id: Option<UserMessageId>,
64 pub content: ContentBlock,
65 pub chunks: Vec<acp::ContentBlock>,
66 pub checkpoint: Option<Checkpoint>,
67 pub indented: bool,
68}
69
70#[derive(Debug)]
71pub struct Checkpoint {
72 git_checkpoint: GitStoreCheckpoint,
73 pub show: bool,
74}
75
76impl UserMessage {
77 fn to_markdown(&self, cx: &App) -> String {
78 let mut markdown = String::new();
79 if self
80 .checkpoint
81 .as_ref()
82 .is_some_and(|checkpoint| checkpoint.show)
83 {
84 writeln!(markdown, "## User (checkpoint)").unwrap();
85 } else {
86 writeln!(markdown, "## User").unwrap();
87 }
88 writeln!(markdown).unwrap();
89 writeln!(markdown, "{}", self.content.to_markdown(cx)).unwrap();
90 writeln!(markdown).unwrap();
91 markdown
92 }
93}
94
95#[derive(Debug, PartialEq)]
96pub struct AssistantMessage {
97 pub chunks: Vec<AssistantMessageChunk>,
98 pub indented: bool,
99}
100
101impl AssistantMessage {
102 pub fn to_markdown(&self, cx: &App) -> String {
103 format!(
104 "## Assistant\n\n{}\n\n",
105 self.chunks
106 .iter()
107 .map(|chunk| chunk.to_markdown(cx))
108 .join("\n\n")
109 )
110 }
111}
112
113#[derive(Debug, PartialEq)]
114pub enum AssistantMessageChunk {
115 Message { block: ContentBlock },
116 Thought { block: ContentBlock },
117}
118
119impl AssistantMessageChunk {
120 pub fn from_str(
121 chunk: &str,
122 language_registry: &Arc<LanguageRegistry>,
123 path_style: PathStyle,
124 cx: &mut App,
125 ) -> Self {
126 Self::Message {
127 block: ContentBlock::new(chunk.into(), language_registry, path_style, cx),
128 }
129 }
130
131 fn to_markdown(&self, cx: &App) -> String {
132 match self {
133 Self::Message { block } => block.to_markdown(cx).to_string(),
134 Self::Thought { block } => {
135 format!("<thinking>\n{}\n</thinking>", block.to_markdown(cx))
136 }
137 }
138 }
139}
140
141#[derive(Debug)]
142pub enum AgentThreadEntry {
143 UserMessage(UserMessage),
144 AssistantMessage(AssistantMessage),
145 ToolCall(ToolCall),
146}
147
148impl AgentThreadEntry {
149 pub fn is_indented(&self) -> bool {
150 match self {
151 Self::UserMessage(message) => message.indented,
152 Self::AssistantMessage(message) => message.indented,
153 Self::ToolCall(_) => false,
154 }
155 }
156
157 pub fn to_markdown(&self, cx: &App) -> String {
158 match self {
159 Self::UserMessage(message) => message.to_markdown(cx),
160 Self::AssistantMessage(message) => message.to_markdown(cx),
161 Self::ToolCall(tool_call) => tool_call.to_markdown(cx),
162 }
163 }
164
165 pub fn user_message(&self) -> Option<&UserMessage> {
166 if let AgentThreadEntry::UserMessage(message) = self {
167 Some(message)
168 } else {
169 None
170 }
171 }
172
173 pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
174 if let AgentThreadEntry::ToolCall(call) = self {
175 itertools::Either::Left(call.diffs())
176 } else {
177 itertools::Either::Right(std::iter::empty())
178 }
179 }
180
181 pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
182 if let AgentThreadEntry::ToolCall(call) = self {
183 itertools::Either::Left(call.terminals())
184 } else {
185 itertools::Either::Right(std::iter::empty())
186 }
187 }
188
189 pub fn location(&self, ix: usize) -> Option<(acp::ToolCallLocation, AgentLocation)> {
190 if let AgentThreadEntry::ToolCall(ToolCall {
191 locations,
192 resolved_locations,
193 ..
194 }) = self
195 {
196 Some((
197 locations.get(ix)?.clone(),
198 resolved_locations.get(ix)?.clone()?,
199 ))
200 } else {
201 None
202 }
203 }
204}
205
206#[derive(Debug)]
207pub struct ToolCall {
208 pub id: acp::ToolCallId,
209 pub label: Entity<Markdown>,
210 pub kind: acp::ToolKind,
211 pub content: Vec<ToolCallContent>,
212 pub status: ToolCallStatus,
213 pub locations: Vec<acp::ToolCallLocation>,
214 pub resolved_locations: Vec<Option<AgentLocation>>,
215 pub raw_input: Option<serde_json::Value>,
216 pub raw_input_markdown: Option<Entity<Markdown>>,
217 pub raw_output: Option<serde_json::Value>,
218 pub tool_name: Option<SharedString>,
219}
220
221impl ToolCall {
222 fn from_acp(
223 tool_call: acp::ToolCall,
224 status: ToolCallStatus,
225 language_registry: Arc<LanguageRegistry>,
226 path_style: PathStyle,
227 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
228 cx: &mut App,
229 ) -> Result<Self> {
230 let title = if tool_call.kind == acp::ToolKind::Execute {
231 tool_call.title
232 } else if let Some((first_line, _)) = tool_call.title.split_once("\n") {
233 first_line.to_owned() + "…"
234 } else {
235 tool_call.title
236 };
237 let mut content = Vec::with_capacity(tool_call.content.len());
238 for item in tool_call.content {
239 if let Some(item) = ToolCallContent::from_acp(
240 item,
241 language_registry.clone(),
242 path_style,
243 terminals,
244 cx,
245 )? {
246 content.push(item);
247 }
248 }
249
250 let raw_input_markdown = tool_call
251 .raw_input
252 .as_ref()
253 .and_then(|input| markdown_for_raw_output(input, &language_registry, cx));
254
255 let tool_name = tool_name_from_meta(&tool_call.meta);
256
257 let result = Self {
258 id: tool_call.tool_call_id,
259 label: cx
260 .new(|cx| Markdown::new(title.into(), Some(language_registry.clone()), None, cx)),
261 kind: tool_call.kind,
262 content,
263 locations: tool_call.locations,
264 resolved_locations: Vec::default(),
265 status,
266 raw_input: tool_call.raw_input,
267 raw_input_markdown,
268 raw_output: tool_call.raw_output,
269 tool_name,
270 };
271 Ok(result)
272 }
273
274 fn update_fields(
275 &mut self,
276 fields: acp::ToolCallUpdateFields,
277 language_registry: Arc<LanguageRegistry>,
278 path_style: PathStyle,
279 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
280 cx: &mut App,
281 ) -> Result<()> {
282 let acp::ToolCallUpdateFields {
283 kind,
284 status,
285 title,
286 content,
287 locations,
288 raw_input,
289 raw_output,
290 ..
291 } = fields;
292
293 if let Some(kind) = kind {
294 self.kind = kind;
295 }
296
297 if let Some(status) = status {
298 self.status = status.into();
299 }
300
301 if let Some(title) = title {
302 self.label.update(cx, |label, cx| {
303 if self.kind == acp::ToolKind::Execute {
304 label.replace(title, cx);
305 } else if let Some((first_line, _)) = title.split_once("\n") {
306 label.replace(first_line.to_owned() + "…", cx);
307 } else {
308 label.replace(title, cx);
309 }
310 });
311 }
312
313 if let Some(content) = content {
314 let mut new_content_len = content.len();
315 let mut content = content.into_iter();
316
317 // Reuse existing content if we can
318 for (old, new) in self.content.iter_mut().zip(content.by_ref()) {
319 let valid_content =
320 old.update_from_acp(new, language_registry.clone(), path_style, terminals, cx)?;
321 if !valid_content {
322 new_content_len -= 1;
323 }
324 }
325 for new in content {
326 if let Some(new) = ToolCallContent::from_acp(
327 new,
328 language_registry.clone(),
329 path_style,
330 terminals,
331 cx,
332 )? {
333 self.content.push(new);
334 } else {
335 new_content_len -= 1;
336 }
337 }
338 self.content.truncate(new_content_len);
339 }
340
341 if let Some(locations) = locations {
342 self.locations = locations;
343 }
344
345 if let Some(raw_input) = raw_input {
346 self.raw_input_markdown = markdown_for_raw_output(&raw_input, &language_registry, cx);
347 self.raw_input = Some(raw_input);
348 }
349
350 if let Some(raw_output) = raw_output {
351 if self.content.is_empty()
352 && let Some(markdown) = markdown_for_raw_output(&raw_output, &language_registry, cx)
353 {
354 self.content
355 .push(ToolCallContent::ContentBlock(ContentBlock::Markdown {
356 markdown,
357 }));
358 }
359 self.raw_output = Some(raw_output);
360 }
361 Ok(())
362 }
363
364 pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
365 self.content.iter().filter_map(|content| match content {
366 ToolCallContent::Diff(diff) => Some(diff),
367 ToolCallContent::ContentBlock(_) => None,
368 ToolCallContent::Terminal(_) => None,
369 ToolCallContent::SubagentThread(_) => None,
370 })
371 }
372
373 pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
374 self.content.iter().filter_map(|content| match content {
375 ToolCallContent::Terminal(terminal) => Some(terminal),
376 ToolCallContent::ContentBlock(_) => None,
377 ToolCallContent::Diff(_) => None,
378 ToolCallContent::SubagentThread(_) => None,
379 })
380 }
381
382 pub fn subagent_thread(&self) -> Option<&Entity<AcpThread>> {
383 self.content.iter().find_map(|content| match content {
384 ToolCallContent::SubagentThread(thread) => Some(thread),
385 _ => None,
386 })
387 }
388
389 pub fn is_subagent(&self) -> bool {
390 matches!(self.kind, acp::ToolKind::Other)
391 && self
392 .tool_name
393 .as_ref()
394 .map(|n| n.as_ref() == SUBAGENT_TOOL_NAME)
395 .unwrap_or(false)
396 }
397
398 pub fn to_markdown(&self, cx: &App) -> String {
399 let mut markdown = format!(
400 "**Tool Call: {}**\nStatus: {}\n\n",
401 self.label.read(cx).source(),
402 self.status
403 );
404 for content in &self.content {
405 markdown.push_str(content.to_markdown(cx).as_str());
406 markdown.push_str("\n\n");
407 }
408 markdown
409 }
410
411 async fn resolve_location(
412 location: acp::ToolCallLocation,
413 project: WeakEntity<Project>,
414 cx: &mut AsyncApp,
415 ) -> Option<ResolvedLocation> {
416 let buffer = project
417 .update(cx, |project, cx| {
418 project
419 .project_path_for_absolute_path(&location.path, cx)
420 .map(|path| project.open_buffer(path, cx))
421 })
422 .ok()??;
423 let buffer = buffer.await.log_err()?;
424 let position = buffer.update(cx, |buffer, _| {
425 let snapshot = buffer.snapshot();
426 if let Some(row) = location.line {
427 let column = snapshot.indent_size_for_line(row).len;
428 let point = snapshot.clip_point(Point::new(row, column), Bias::Left);
429 snapshot.anchor_before(point)
430 } else {
431 Anchor::min_for_buffer(snapshot.remote_id())
432 }
433 });
434
435 Some(ResolvedLocation { buffer, position })
436 }
437
438 fn resolve_locations(
439 &self,
440 project: Entity<Project>,
441 cx: &mut App,
442 ) -> Task<Vec<Option<ResolvedLocation>>> {
443 let locations = self.locations.clone();
444 project.update(cx, |_, cx| {
445 cx.spawn(async move |project, cx| {
446 let mut new_locations = Vec::new();
447 for location in locations {
448 new_locations.push(Self::resolve_location(location, project.clone(), cx).await);
449 }
450 new_locations
451 })
452 })
453 }
454}
455
456// Separate so we can hold a strong reference to the buffer
457// for saving on the thread
458#[derive(Clone, Debug, PartialEq, Eq)]
459struct ResolvedLocation {
460 buffer: Entity<Buffer>,
461 position: Anchor,
462}
463
464impl From<&ResolvedLocation> for AgentLocation {
465 fn from(value: &ResolvedLocation) -> Self {
466 Self {
467 buffer: value.buffer.downgrade(),
468 position: value.position,
469 }
470 }
471}
472
473#[derive(Debug)]
474pub enum ToolCallStatus {
475 /// The tool call hasn't started running yet, but we start showing it to
476 /// the user.
477 Pending,
478 /// The tool call is waiting for confirmation from the user.
479 WaitingForConfirmation {
480 options: PermissionOptions,
481 respond_tx: oneshot::Sender<acp::PermissionOptionId>,
482 },
483 /// The tool call is currently running.
484 InProgress,
485 /// The tool call completed successfully.
486 Completed,
487 /// The tool call failed.
488 Failed,
489 /// The user rejected the tool call.
490 Rejected,
491 /// The user canceled generation so the tool call was canceled.
492 Canceled,
493}
494
495impl From<acp::ToolCallStatus> for ToolCallStatus {
496 fn from(status: acp::ToolCallStatus) -> Self {
497 match status {
498 acp::ToolCallStatus::Pending => Self::Pending,
499 acp::ToolCallStatus::InProgress => Self::InProgress,
500 acp::ToolCallStatus::Completed => Self::Completed,
501 acp::ToolCallStatus::Failed => Self::Failed,
502 _ => Self::Pending,
503 }
504 }
505}
506
507impl Display for ToolCallStatus {
508 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
509 write!(
510 f,
511 "{}",
512 match self {
513 ToolCallStatus::Pending => "Pending",
514 ToolCallStatus::WaitingForConfirmation { .. } => "Waiting for confirmation",
515 ToolCallStatus::InProgress => "In Progress",
516 ToolCallStatus::Completed => "Completed",
517 ToolCallStatus::Failed => "Failed",
518 ToolCallStatus::Rejected => "Rejected",
519 ToolCallStatus::Canceled => "Canceled",
520 }
521 )
522 }
523}
524
525#[derive(Debug, PartialEq, Clone)]
526pub enum ContentBlock {
527 Empty,
528 Markdown { markdown: Entity<Markdown> },
529 ResourceLink { resource_link: acp::ResourceLink },
530 Image { image: Arc<gpui::Image> },
531}
532
533impl ContentBlock {
534 pub fn new(
535 block: acp::ContentBlock,
536 language_registry: &Arc<LanguageRegistry>,
537 path_style: PathStyle,
538 cx: &mut App,
539 ) -> Self {
540 let mut this = Self::Empty;
541 this.append(block, language_registry, path_style, cx);
542 this
543 }
544
545 pub fn new_combined(
546 blocks: impl IntoIterator<Item = acp::ContentBlock>,
547 language_registry: Arc<LanguageRegistry>,
548 path_style: PathStyle,
549 cx: &mut App,
550 ) -> Self {
551 let mut this = Self::Empty;
552 for block in blocks {
553 this.append(block, &language_registry, path_style, cx);
554 }
555 this
556 }
557
558 pub fn append(
559 &mut self,
560 block: acp::ContentBlock,
561 language_registry: &Arc<LanguageRegistry>,
562 path_style: PathStyle,
563 cx: &mut App,
564 ) {
565 match (&mut *self, &block) {
566 (ContentBlock::Empty, acp::ContentBlock::ResourceLink(resource_link)) => {
567 *self = ContentBlock::ResourceLink {
568 resource_link: resource_link.clone(),
569 };
570 }
571 (ContentBlock::Empty, acp::ContentBlock::Image(image_content)) => {
572 if let Some(image) = Self::decode_image(image_content) {
573 *self = ContentBlock::Image { image };
574 } else {
575 let new_content = Self::image_md(image_content);
576 *self = Self::create_markdown_block(new_content, language_registry, cx);
577 }
578 }
579 (ContentBlock::Empty, _) => {
580 let new_content = Self::block_string_contents(&block, path_style);
581 *self = Self::create_markdown_block(new_content, language_registry, cx);
582 }
583 (ContentBlock::Markdown { markdown }, _) => {
584 let new_content = Self::block_string_contents(&block, path_style);
585 markdown.update(cx, |markdown, cx| markdown.append(&new_content, cx));
586 }
587 (ContentBlock::ResourceLink { resource_link }, _) => {
588 let existing_content = Self::resource_link_md(&resource_link.uri, path_style);
589 let new_content = Self::block_string_contents(&block, path_style);
590 let combined = format!("{}\n{}", existing_content, new_content);
591 *self = Self::create_markdown_block(combined, language_registry, cx);
592 }
593 (ContentBlock::Image { .. }, _) => {
594 let new_content = Self::block_string_contents(&block, path_style);
595 let combined = format!("`Image`\n{}", new_content);
596 *self = Self::create_markdown_block(combined, language_registry, cx);
597 }
598 }
599 }
600
601 fn decode_image(image_content: &acp::ImageContent) -> Option<Arc<gpui::Image>> {
602 use base64::Engine as _;
603
604 let bytes = base64::engine::general_purpose::STANDARD
605 .decode(image_content.data.as_bytes())
606 .ok()?;
607 let format = gpui::ImageFormat::from_mime_type(&image_content.mime_type)?;
608 Some(Arc::new(gpui::Image::from_bytes(format, bytes)))
609 }
610
611 fn create_markdown_block(
612 content: String,
613 language_registry: &Arc<LanguageRegistry>,
614 cx: &mut App,
615 ) -> ContentBlock {
616 ContentBlock::Markdown {
617 markdown: cx
618 .new(|cx| Markdown::new(content.into(), Some(language_registry.clone()), None, cx)),
619 }
620 }
621
622 fn block_string_contents(block: &acp::ContentBlock, path_style: PathStyle) -> String {
623 match block {
624 acp::ContentBlock::Text(text_content) => text_content.text.clone(),
625 acp::ContentBlock::ResourceLink(resource_link) => {
626 Self::resource_link_md(&resource_link.uri, path_style)
627 }
628 acp::ContentBlock::Resource(acp::EmbeddedResource {
629 resource:
630 acp::EmbeddedResourceResource::TextResourceContents(acp::TextResourceContents {
631 uri,
632 ..
633 }),
634 ..
635 }) => Self::resource_link_md(uri, path_style),
636 acp::ContentBlock::Image(image) => Self::image_md(image),
637 _ => String::new(),
638 }
639 }
640
641 fn resource_link_md(uri: &str, path_style: PathStyle) -> String {
642 if let Some(uri) = MentionUri::parse(uri, path_style).log_err() {
643 uri.as_link().to_string()
644 } else {
645 uri.to_string()
646 }
647 }
648
649 fn image_md(_image: &acp::ImageContent) -> String {
650 "`Image`".into()
651 }
652
653 pub fn to_markdown<'a>(&'a self, cx: &'a App) -> &'a str {
654 match self {
655 ContentBlock::Empty => "",
656 ContentBlock::Markdown { markdown } => markdown.read(cx).source(),
657 ContentBlock::ResourceLink { resource_link } => &resource_link.uri,
658 ContentBlock::Image { .. } => "`Image`",
659 }
660 }
661
662 pub fn markdown(&self) -> Option<&Entity<Markdown>> {
663 match self {
664 ContentBlock::Empty => None,
665 ContentBlock::Markdown { markdown } => Some(markdown),
666 ContentBlock::ResourceLink { .. } => None,
667 ContentBlock::Image { .. } => None,
668 }
669 }
670
671 pub fn resource_link(&self) -> Option<&acp::ResourceLink> {
672 match self {
673 ContentBlock::ResourceLink { resource_link } => Some(resource_link),
674 _ => None,
675 }
676 }
677
678 pub fn image(&self) -> Option<&Arc<gpui::Image>> {
679 match self {
680 ContentBlock::Image { image } => Some(image),
681 _ => None,
682 }
683 }
684}
685
686#[derive(Debug)]
687pub enum ToolCallContent {
688 ContentBlock(ContentBlock),
689 Diff(Entity<Diff>),
690 Terminal(Entity<Terminal>),
691 SubagentThread(Entity<AcpThread>),
692}
693
694impl ToolCallContent {
695 pub fn from_acp(
696 content: acp::ToolCallContent,
697 language_registry: Arc<LanguageRegistry>,
698 path_style: PathStyle,
699 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
700 cx: &mut App,
701 ) -> Result<Option<Self>> {
702 match content {
703 acp::ToolCallContent::Content(acp::Content { content, .. }) => {
704 Ok(Some(Self::ContentBlock(ContentBlock::new(
705 content,
706 &language_registry,
707 path_style,
708 cx,
709 ))))
710 }
711 acp::ToolCallContent::Diff(diff) => Ok(Some(Self::Diff(cx.new(|cx| {
712 Diff::finalized(
713 diff.path.to_string_lossy().into_owned(),
714 diff.old_text,
715 diff.new_text,
716 language_registry,
717 cx,
718 )
719 })))),
720 acp::ToolCallContent::Terminal(acp::Terminal { terminal_id, .. }) => terminals
721 .get(&terminal_id)
722 .cloned()
723 .map(|terminal| Some(Self::Terminal(terminal)))
724 .ok_or_else(|| anyhow::anyhow!("Terminal with id `{}` not found", terminal_id)),
725 _ => Ok(None),
726 }
727 }
728
729 pub fn update_from_acp(
730 &mut self,
731 new: acp::ToolCallContent,
732 language_registry: Arc<LanguageRegistry>,
733 path_style: PathStyle,
734 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
735 cx: &mut App,
736 ) -> Result<bool> {
737 let needs_update = match (&self, &new) {
738 (Self::Diff(old_diff), acp::ToolCallContent::Diff(new_diff)) => {
739 old_diff.read(cx).needs_update(
740 new_diff.old_text.as_deref().unwrap_or(""),
741 &new_diff.new_text,
742 cx,
743 )
744 }
745 _ => true,
746 };
747
748 if let Some(update) = Self::from_acp(new, language_registry, path_style, terminals, cx)? {
749 if needs_update {
750 *self = update;
751 }
752 Ok(true)
753 } else {
754 Ok(false)
755 }
756 }
757
758 pub fn to_markdown(&self, cx: &App) -> String {
759 match self {
760 Self::ContentBlock(content) => content.to_markdown(cx).to_string(),
761 Self::Diff(diff) => diff.read(cx).to_markdown(cx),
762 Self::Terminal(terminal) => terminal.read(cx).to_markdown(cx),
763 Self::SubagentThread(thread) => thread.read(cx).to_markdown(cx),
764 }
765 }
766
767 pub fn image(&self) -> Option<&Arc<gpui::Image>> {
768 match self {
769 Self::ContentBlock(content) => content.image(),
770 _ => None,
771 }
772 }
773
774 pub fn subagent_thread(&self) -> Option<&Entity<AcpThread>> {
775 match self {
776 Self::SubagentThread(thread) => Some(thread),
777 _ => None,
778 }
779 }
780}
781
782#[derive(Debug, PartialEq)]
783pub enum ToolCallUpdate {
784 UpdateFields(acp::ToolCallUpdate),
785 UpdateDiff(ToolCallUpdateDiff),
786 UpdateTerminal(ToolCallUpdateTerminal),
787 UpdateSubagentThread(ToolCallUpdateSubagentThread),
788}
789
790impl ToolCallUpdate {
791 fn id(&self) -> &acp::ToolCallId {
792 match self {
793 Self::UpdateFields(update) => &update.tool_call_id,
794 Self::UpdateDiff(diff) => &diff.id,
795 Self::UpdateTerminal(terminal) => &terminal.id,
796 Self::UpdateSubagentThread(subagent) => &subagent.id,
797 }
798 }
799}
800
801impl From<acp::ToolCallUpdate> for ToolCallUpdate {
802 fn from(update: acp::ToolCallUpdate) -> Self {
803 Self::UpdateFields(update)
804 }
805}
806
807impl From<ToolCallUpdateDiff> for ToolCallUpdate {
808 fn from(diff: ToolCallUpdateDiff) -> Self {
809 Self::UpdateDiff(diff)
810 }
811}
812
813#[derive(Debug, PartialEq)]
814pub struct ToolCallUpdateDiff {
815 pub id: acp::ToolCallId,
816 pub diff: Entity<Diff>,
817}
818
819impl From<ToolCallUpdateTerminal> for ToolCallUpdate {
820 fn from(terminal: ToolCallUpdateTerminal) -> Self {
821 Self::UpdateTerminal(terminal)
822 }
823}
824
825#[derive(Debug, PartialEq)]
826pub struct ToolCallUpdateTerminal {
827 pub id: acp::ToolCallId,
828 pub terminal: Entity<Terminal>,
829}
830
831impl From<ToolCallUpdateSubagentThread> for ToolCallUpdate {
832 fn from(subagent: ToolCallUpdateSubagentThread) -> Self {
833 Self::UpdateSubagentThread(subagent)
834 }
835}
836
837#[derive(Debug, PartialEq)]
838pub struct ToolCallUpdateSubagentThread {
839 pub id: acp::ToolCallId,
840 pub thread: Entity<AcpThread>,
841}
842
843#[derive(Debug, Default)]
844pub struct Plan {
845 pub entries: Vec<PlanEntry>,
846}
847
848#[derive(Debug)]
849pub struct PlanStats<'a> {
850 pub in_progress_entry: Option<&'a PlanEntry>,
851 pub pending: u32,
852 pub completed: u32,
853}
854
855impl Plan {
856 pub fn is_empty(&self) -> bool {
857 self.entries.is_empty()
858 }
859
860 pub fn stats(&self) -> PlanStats<'_> {
861 let mut stats = PlanStats {
862 in_progress_entry: None,
863 pending: 0,
864 completed: 0,
865 };
866
867 for entry in &self.entries {
868 match &entry.status {
869 acp::PlanEntryStatus::Pending => {
870 stats.pending += 1;
871 }
872 acp::PlanEntryStatus::InProgress => {
873 stats.in_progress_entry = stats.in_progress_entry.or(Some(entry));
874 }
875 acp::PlanEntryStatus::Completed => {
876 stats.completed += 1;
877 }
878 _ => {}
879 }
880 }
881
882 stats
883 }
884}
885
886#[derive(Debug)]
887pub struct PlanEntry {
888 pub content: Entity<Markdown>,
889 pub priority: acp::PlanEntryPriority,
890 pub status: acp::PlanEntryStatus,
891}
892
893impl PlanEntry {
894 pub fn from_acp(entry: acp::PlanEntry, cx: &mut App) -> Self {
895 Self {
896 content: cx.new(|cx| Markdown::new(entry.content.into(), None, None, cx)),
897 priority: entry.priority,
898 status: entry.status,
899 }
900 }
901}
902
903#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
904pub struct TokenUsage {
905 pub max_tokens: u64,
906 pub used_tokens: u64,
907 pub input_tokens: u64,
908 pub output_tokens: u64,
909}
910
911impl TokenUsage {
912 pub fn ratio(&self) -> TokenUsageRatio {
913 #[cfg(debug_assertions)]
914 let warning_threshold: f32 = std::env::var("ZED_THREAD_WARNING_THRESHOLD")
915 .unwrap_or("0.8".to_string())
916 .parse()
917 .unwrap();
918 #[cfg(not(debug_assertions))]
919 let warning_threshold: f32 = 0.8;
920
921 // When the maximum is unknown because there is no selected model,
922 // avoid showing the token limit warning.
923 if self.max_tokens == 0 {
924 TokenUsageRatio::Normal
925 } else if self.used_tokens >= self.max_tokens {
926 TokenUsageRatio::Exceeded
927 } else if self.used_tokens as f32 / self.max_tokens as f32 >= warning_threshold {
928 TokenUsageRatio::Warning
929 } else {
930 TokenUsageRatio::Normal
931 }
932 }
933}
934
935#[derive(Debug, Clone, PartialEq, Eq)]
936pub enum TokenUsageRatio {
937 Normal,
938 Warning,
939 Exceeded,
940}
941
942#[derive(Debug, Clone)]
943pub struct RetryStatus {
944 pub last_error: SharedString,
945 pub attempt: usize,
946 pub max_attempts: usize,
947 pub started_at: Instant,
948 pub duration: Duration,
949}
950
951pub struct AcpThread {
952 title: SharedString,
953 entries: Vec<AgentThreadEntry>,
954 plan: Plan,
955 project: Entity<Project>,
956 action_log: Entity<ActionLog>,
957 shared_buffers: HashMap<Entity<Buffer>, BufferSnapshot>,
958 send_task: Option<Task<()>>,
959 connection: Rc<dyn AgentConnection>,
960 session_id: acp::SessionId,
961 token_usage: Option<TokenUsage>,
962 prompt_capabilities: acp::PromptCapabilities,
963 _observe_prompt_capabilities: Task<anyhow::Result<()>>,
964 terminals: HashMap<acp::TerminalId, Entity<Terminal>>,
965 pending_terminal_output: HashMap<acp::TerminalId, Vec<Vec<u8>>>,
966 pending_terminal_exit: HashMap<acp::TerminalId, acp::TerminalExitStatus>,
967}
968
969impl From<&AcpThread> for ActionLogTelemetry {
970 fn from(value: &AcpThread) -> Self {
971 Self {
972 agent_telemetry_id: value.connection().telemetry_id(),
973 session_id: value.session_id.0.clone(),
974 }
975 }
976}
977
978#[derive(Debug)]
979pub enum AcpThreadEvent {
980 NewEntry,
981 TitleUpdated,
982 TokenUsageUpdated,
983 EntryUpdated(usize),
984 EntriesRemoved(Range<usize>),
985 ToolAuthorizationRequired,
986 Retry(RetryStatus),
987 Stopped,
988 Error,
989 LoadError(LoadError),
990 PromptCapabilitiesUpdated,
991 Refusal,
992 AvailableCommandsUpdated(Vec<acp::AvailableCommand>),
993 ModeUpdated(acp::SessionModeId),
994 ConfigOptionsUpdated(Vec<acp::SessionConfigOption>),
995}
996
997impl EventEmitter<AcpThreadEvent> for AcpThread {}
998
999#[derive(Debug, Clone)]
1000pub enum TerminalProviderEvent {
1001 Created {
1002 terminal_id: acp::TerminalId,
1003 label: String,
1004 cwd: Option<PathBuf>,
1005 output_byte_limit: Option<u64>,
1006 terminal: Entity<::terminal::Terminal>,
1007 },
1008 Output {
1009 terminal_id: acp::TerminalId,
1010 data: Vec<u8>,
1011 },
1012 TitleChanged {
1013 terminal_id: acp::TerminalId,
1014 title: String,
1015 },
1016 Exit {
1017 terminal_id: acp::TerminalId,
1018 status: acp::TerminalExitStatus,
1019 },
1020}
1021
1022#[derive(Debug, Clone)]
1023pub enum TerminalProviderCommand {
1024 WriteInput {
1025 terminal_id: acp::TerminalId,
1026 bytes: Vec<u8>,
1027 },
1028 Resize {
1029 terminal_id: acp::TerminalId,
1030 cols: u16,
1031 rows: u16,
1032 },
1033 Close {
1034 terminal_id: acp::TerminalId,
1035 },
1036}
1037
1038impl AcpThread {
1039 pub fn on_terminal_provider_event(
1040 &mut self,
1041 event: TerminalProviderEvent,
1042 cx: &mut Context<Self>,
1043 ) {
1044 match event {
1045 TerminalProviderEvent::Created {
1046 terminal_id,
1047 label,
1048 cwd,
1049 output_byte_limit,
1050 terminal,
1051 } => {
1052 let entity = self.register_terminal_created(
1053 terminal_id.clone(),
1054 label,
1055 cwd,
1056 output_byte_limit,
1057 terminal,
1058 cx,
1059 );
1060
1061 if let Some(mut chunks) = self.pending_terminal_output.remove(&terminal_id) {
1062 for data in chunks.drain(..) {
1063 entity.update(cx, |term, cx| {
1064 term.inner().update(cx, |inner, cx| {
1065 inner.write_output(&data, cx);
1066 })
1067 });
1068 }
1069 }
1070
1071 if let Some(_status) = self.pending_terminal_exit.remove(&terminal_id) {
1072 entity.update(cx, |_term, cx| {
1073 cx.notify();
1074 });
1075 }
1076
1077 cx.notify();
1078 }
1079 TerminalProviderEvent::Output { terminal_id, data } => {
1080 if let Some(entity) = self.terminals.get(&terminal_id) {
1081 entity.update(cx, |term, cx| {
1082 term.inner().update(cx, |inner, cx| {
1083 inner.write_output(&data, cx);
1084 })
1085 });
1086 } else {
1087 self.pending_terminal_output
1088 .entry(terminal_id)
1089 .or_default()
1090 .push(data);
1091 }
1092 }
1093 TerminalProviderEvent::TitleChanged { terminal_id, title } => {
1094 if let Some(entity) = self.terminals.get(&terminal_id) {
1095 entity.update(cx, |term, cx| {
1096 term.inner().update(cx, |inner, cx| {
1097 inner.breadcrumb_text = title;
1098 cx.emit(::terminal::Event::BreadcrumbsChanged);
1099 })
1100 });
1101 }
1102 }
1103 TerminalProviderEvent::Exit {
1104 terminal_id,
1105 status,
1106 } => {
1107 if let Some(entity) = self.terminals.get(&terminal_id) {
1108 entity.update(cx, |_term, cx| {
1109 cx.notify();
1110 });
1111 } else {
1112 self.pending_terminal_exit.insert(terminal_id, status);
1113 }
1114 }
1115 }
1116 }
1117}
1118
1119#[derive(PartialEq, Eq, Debug)]
1120pub enum ThreadStatus {
1121 Idle,
1122 Generating,
1123}
1124
1125#[derive(Debug, Clone)]
1126pub enum LoadError {
1127 Unsupported {
1128 command: SharedString,
1129 current_version: SharedString,
1130 minimum_version: SharedString,
1131 },
1132 FailedToInstall(SharedString),
1133 Exited {
1134 status: ExitStatus,
1135 },
1136 Other(SharedString),
1137}
1138
1139impl Display for LoadError {
1140 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1141 match self {
1142 LoadError::Unsupported {
1143 command: path,
1144 current_version,
1145 minimum_version,
1146 } => {
1147 write!(
1148 f,
1149 "version {current_version} from {path} is not supported (need at least {minimum_version})"
1150 )
1151 }
1152 LoadError::FailedToInstall(msg) => write!(f, "Failed to install: {msg}"),
1153 LoadError::Exited { status } => write!(f, "Server exited with status {status}"),
1154 LoadError::Other(msg) => write!(f, "{msg}"),
1155 }
1156 }
1157}
1158
1159impl Error for LoadError {}
1160
1161impl AcpThread {
1162 pub fn new(
1163 title: impl Into<SharedString>,
1164 connection: Rc<dyn AgentConnection>,
1165 project: Entity<Project>,
1166 action_log: Entity<ActionLog>,
1167 session_id: acp::SessionId,
1168 mut prompt_capabilities_rx: watch::Receiver<acp::PromptCapabilities>,
1169 cx: &mut Context<Self>,
1170 ) -> Self {
1171 let prompt_capabilities = prompt_capabilities_rx.borrow().clone();
1172 let task = cx.spawn::<_, anyhow::Result<()>>(async move |this, cx| {
1173 loop {
1174 let caps = prompt_capabilities_rx.recv().await?;
1175 this.update(cx, |this, cx| {
1176 this.prompt_capabilities = caps;
1177 cx.emit(AcpThreadEvent::PromptCapabilitiesUpdated);
1178 })?;
1179 }
1180 });
1181
1182 Self {
1183 action_log,
1184 shared_buffers: Default::default(),
1185 entries: Default::default(),
1186 plan: Default::default(),
1187 title: title.into(),
1188 project,
1189 send_task: None,
1190 connection,
1191 session_id,
1192 token_usage: None,
1193 prompt_capabilities,
1194 _observe_prompt_capabilities: task,
1195 terminals: HashMap::default(),
1196 pending_terminal_output: HashMap::default(),
1197 pending_terminal_exit: HashMap::default(),
1198 }
1199 }
1200
1201 pub fn prompt_capabilities(&self) -> acp::PromptCapabilities {
1202 self.prompt_capabilities.clone()
1203 }
1204
1205 pub fn connection(&self) -> &Rc<dyn AgentConnection> {
1206 &self.connection
1207 }
1208
1209 pub fn action_log(&self) -> &Entity<ActionLog> {
1210 &self.action_log
1211 }
1212
1213 pub fn project(&self) -> &Entity<Project> {
1214 &self.project
1215 }
1216
1217 pub fn title(&self) -> SharedString {
1218 self.title.clone()
1219 }
1220
1221 pub fn entries(&self) -> &[AgentThreadEntry] {
1222 &self.entries
1223 }
1224
1225 pub fn session_id(&self) -> &acp::SessionId {
1226 &self.session_id
1227 }
1228
1229 pub fn status(&self) -> ThreadStatus {
1230 if self.send_task.is_some() {
1231 ThreadStatus::Generating
1232 } else {
1233 ThreadStatus::Idle
1234 }
1235 }
1236
1237 pub fn token_usage(&self) -> Option<&TokenUsage> {
1238 self.token_usage.as_ref()
1239 }
1240
1241 pub fn has_pending_edit_tool_calls(&self) -> bool {
1242 for entry in self.entries.iter().rev() {
1243 match entry {
1244 AgentThreadEntry::UserMessage(_) => return false,
1245 AgentThreadEntry::ToolCall(
1246 call @ ToolCall {
1247 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
1248 ..
1249 },
1250 ) if call.diffs().next().is_some() => {
1251 return true;
1252 }
1253 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1254 }
1255 }
1256
1257 false
1258 }
1259
1260 pub fn has_in_progress_tool_calls(&self) -> bool {
1261 for entry in self.entries.iter().rev() {
1262 match entry {
1263 AgentThreadEntry::UserMessage(_) => return false,
1264 AgentThreadEntry::ToolCall(ToolCall {
1265 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
1266 ..
1267 }) => {
1268 return true;
1269 }
1270 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1271 }
1272 }
1273
1274 false
1275 }
1276
1277 pub fn used_tools_since_last_user_message(&self) -> bool {
1278 for entry in self.entries.iter().rev() {
1279 match entry {
1280 AgentThreadEntry::UserMessage(..) => return false,
1281 AgentThreadEntry::AssistantMessage(..) => continue,
1282 AgentThreadEntry::ToolCall(..) => return true,
1283 }
1284 }
1285
1286 false
1287 }
1288
1289 pub fn handle_session_update(
1290 &mut self,
1291 update: acp::SessionUpdate,
1292 cx: &mut Context<Self>,
1293 ) -> Result<(), acp::Error> {
1294 match update {
1295 acp::SessionUpdate::UserMessageChunk(acp::ContentChunk { content, .. }) => {
1296 self.push_user_content_block(None, content, cx);
1297 }
1298 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk { content, .. }) => {
1299 self.push_assistant_content_block(content, false, cx);
1300 }
1301 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk { content, .. }) => {
1302 self.push_assistant_content_block(content, true, cx);
1303 }
1304 acp::SessionUpdate::ToolCall(tool_call) => {
1305 self.upsert_tool_call(tool_call, cx)?;
1306 }
1307 acp::SessionUpdate::ToolCallUpdate(tool_call_update) => {
1308 self.update_tool_call(tool_call_update, cx)?;
1309 }
1310 acp::SessionUpdate::Plan(plan) => {
1311 self.update_plan(plan, cx);
1312 }
1313 acp::SessionUpdate::AvailableCommandsUpdate(acp::AvailableCommandsUpdate {
1314 available_commands,
1315 ..
1316 }) => cx.emit(AcpThreadEvent::AvailableCommandsUpdated(available_commands)),
1317 acp::SessionUpdate::CurrentModeUpdate(acp::CurrentModeUpdate {
1318 current_mode_id,
1319 ..
1320 }) => cx.emit(AcpThreadEvent::ModeUpdated(current_mode_id)),
1321 acp::SessionUpdate::ConfigOptionUpdate(acp::ConfigOptionUpdate {
1322 config_options,
1323 ..
1324 }) => cx.emit(AcpThreadEvent::ConfigOptionsUpdated(config_options)),
1325 _ => {}
1326 }
1327 Ok(())
1328 }
1329
1330 pub fn push_user_content_block(
1331 &mut self,
1332 message_id: Option<UserMessageId>,
1333 chunk: acp::ContentBlock,
1334 cx: &mut Context<Self>,
1335 ) {
1336 self.push_user_content_block_with_indent(message_id, chunk, false, cx)
1337 }
1338
1339 pub fn push_user_content_block_with_indent(
1340 &mut self,
1341 message_id: Option<UserMessageId>,
1342 chunk: acp::ContentBlock,
1343 indented: bool,
1344 cx: &mut Context<Self>,
1345 ) {
1346 let language_registry = self.project.read(cx).languages().clone();
1347 let path_style = self.project.read(cx).path_style(cx);
1348 let entries_len = self.entries.len();
1349
1350 if let Some(last_entry) = self.entries.last_mut()
1351 && let AgentThreadEntry::UserMessage(UserMessage {
1352 id,
1353 content,
1354 chunks,
1355 indented: existing_indented,
1356 ..
1357 }) = last_entry
1358 && *existing_indented == indented
1359 {
1360 *id = message_id.or(id.take());
1361 content.append(chunk.clone(), &language_registry, path_style, cx);
1362 chunks.push(chunk);
1363 let idx = entries_len - 1;
1364 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1365 } else {
1366 let content = ContentBlock::new(chunk.clone(), &language_registry, path_style, cx);
1367 self.push_entry(
1368 AgentThreadEntry::UserMessage(UserMessage {
1369 id: message_id,
1370 content,
1371 chunks: vec![chunk],
1372 checkpoint: None,
1373 indented,
1374 }),
1375 cx,
1376 );
1377 }
1378 }
1379
1380 pub fn push_assistant_content_block(
1381 &mut self,
1382 chunk: acp::ContentBlock,
1383 is_thought: bool,
1384 cx: &mut Context<Self>,
1385 ) {
1386 self.push_assistant_content_block_with_indent(chunk, is_thought, false, cx)
1387 }
1388
1389 pub fn push_assistant_content_block_with_indent(
1390 &mut self,
1391 chunk: acp::ContentBlock,
1392 is_thought: bool,
1393 indented: bool,
1394 cx: &mut Context<Self>,
1395 ) {
1396 let language_registry = self.project.read(cx).languages().clone();
1397 let path_style = self.project.read(cx).path_style(cx);
1398 let entries_len = self.entries.len();
1399 if let Some(last_entry) = self.entries.last_mut()
1400 && let AgentThreadEntry::AssistantMessage(AssistantMessage {
1401 chunks,
1402 indented: existing_indented,
1403 }) = last_entry
1404 && *existing_indented == indented
1405 {
1406 let idx = entries_len - 1;
1407 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1408 match (chunks.last_mut(), is_thought) {
1409 (Some(AssistantMessageChunk::Message { block }), false)
1410 | (Some(AssistantMessageChunk::Thought { block }), true) => {
1411 block.append(chunk, &language_registry, path_style, cx)
1412 }
1413 _ => {
1414 let block = ContentBlock::new(chunk, &language_registry, path_style, cx);
1415 if is_thought {
1416 chunks.push(AssistantMessageChunk::Thought { block })
1417 } else {
1418 chunks.push(AssistantMessageChunk::Message { block })
1419 }
1420 }
1421 }
1422 } else {
1423 let block = ContentBlock::new(chunk, &language_registry, path_style, cx);
1424 let chunk = if is_thought {
1425 AssistantMessageChunk::Thought { block }
1426 } else {
1427 AssistantMessageChunk::Message { block }
1428 };
1429
1430 self.push_entry(
1431 AgentThreadEntry::AssistantMessage(AssistantMessage {
1432 chunks: vec![chunk],
1433 indented,
1434 }),
1435 cx,
1436 );
1437 }
1438 }
1439
1440 fn push_entry(&mut self, entry: AgentThreadEntry, cx: &mut Context<Self>) {
1441 self.entries.push(entry);
1442 cx.emit(AcpThreadEvent::NewEntry);
1443 }
1444
1445 pub fn can_set_title(&mut self, cx: &mut Context<Self>) -> bool {
1446 self.connection.set_title(&self.session_id, cx).is_some()
1447 }
1448
1449 pub fn set_title(&mut self, title: SharedString, cx: &mut Context<Self>) -> Task<Result<()>> {
1450 if title != self.title {
1451 self.title = title.clone();
1452 cx.emit(AcpThreadEvent::TitleUpdated);
1453 if let Some(set_title) = self.connection.set_title(&self.session_id, cx) {
1454 return set_title.run(title, cx);
1455 }
1456 }
1457 Task::ready(Ok(()))
1458 }
1459
1460 pub fn update_token_usage(&mut self, usage: Option<TokenUsage>, cx: &mut Context<Self>) {
1461 self.token_usage = usage;
1462 cx.emit(AcpThreadEvent::TokenUsageUpdated);
1463 }
1464
1465 pub fn update_retry_status(&mut self, status: RetryStatus, cx: &mut Context<Self>) {
1466 cx.emit(AcpThreadEvent::Retry(status));
1467 }
1468
1469 pub fn update_tool_call(
1470 &mut self,
1471 update: impl Into<ToolCallUpdate>,
1472 cx: &mut Context<Self>,
1473 ) -> Result<()> {
1474 let update = update.into();
1475 let languages = self.project.read(cx).languages().clone();
1476 let path_style = self.project.read(cx).path_style(cx);
1477
1478 let ix = match self.index_for_tool_call(update.id()) {
1479 Some(ix) => ix,
1480 None => {
1481 // Tool call not found - create a failed tool call entry
1482 let failed_tool_call = ToolCall {
1483 id: update.id().clone(),
1484 label: cx.new(|cx| Markdown::new("Tool call not found".into(), None, None, cx)),
1485 kind: acp::ToolKind::Fetch,
1486 content: vec![ToolCallContent::ContentBlock(ContentBlock::new(
1487 "Tool call not found".into(),
1488 &languages,
1489 path_style,
1490 cx,
1491 ))],
1492 status: ToolCallStatus::Failed,
1493 locations: Vec::new(),
1494 resolved_locations: Vec::new(),
1495 raw_input: None,
1496 raw_input_markdown: None,
1497 raw_output: None,
1498 tool_name: None,
1499 };
1500 self.push_entry(AgentThreadEntry::ToolCall(failed_tool_call), cx);
1501 return Ok(());
1502 }
1503 };
1504 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1505 unreachable!()
1506 };
1507
1508 match update {
1509 ToolCallUpdate::UpdateFields(update) => {
1510 let location_updated = update.fields.locations.is_some();
1511 call.update_fields(update.fields, languages, path_style, &self.terminals, cx)?;
1512 if location_updated {
1513 self.resolve_locations(update.tool_call_id, cx);
1514 }
1515 }
1516 ToolCallUpdate::UpdateDiff(update) => {
1517 call.content.clear();
1518 call.content.push(ToolCallContent::Diff(update.diff));
1519 }
1520 ToolCallUpdate::UpdateTerminal(update) => {
1521 call.content.clear();
1522 call.content
1523 .push(ToolCallContent::Terminal(update.terminal));
1524 }
1525 ToolCallUpdate::UpdateSubagentThread(update) => {
1526 debug_assert!(
1527 !call.content.iter().any(|c| {
1528 matches!(c, ToolCallContent::SubagentThread(existing) if existing == &update.thread)
1529 }),
1530 "Duplicate SubagentThread update for the same AcpThread entity"
1531 );
1532 call.content
1533 .push(ToolCallContent::SubagentThread(update.thread));
1534 }
1535 }
1536
1537 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1538
1539 Ok(())
1540 }
1541
1542 /// Updates a tool call if id matches an existing entry, otherwise inserts a new one.
1543 pub fn upsert_tool_call(
1544 &mut self,
1545 tool_call: acp::ToolCall,
1546 cx: &mut Context<Self>,
1547 ) -> Result<(), acp::Error> {
1548 let status = tool_call.status.into();
1549 self.upsert_tool_call_inner(tool_call.into(), status, cx)
1550 }
1551
1552 /// Fails if id does not match an existing entry.
1553 pub fn upsert_tool_call_inner(
1554 &mut self,
1555 update: acp::ToolCallUpdate,
1556 status: ToolCallStatus,
1557 cx: &mut Context<Self>,
1558 ) -> Result<(), acp::Error> {
1559 let language_registry = self.project.read(cx).languages().clone();
1560 let path_style = self.project.read(cx).path_style(cx);
1561 let id = update.tool_call_id.clone();
1562
1563 let agent_telemetry_id = self.connection().telemetry_id();
1564 let session = self.session_id();
1565 if let ToolCallStatus::Completed | ToolCallStatus::Failed = status {
1566 let status = if matches!(status, ToolCallStatus::Completed) {
1567 "completed"
1568 } else {
1569 "failed"
1570 };
1571 telemetry::event!(
1572 "Agent Tool Call Completed",
1573 agent_telemetry_id,
1574 session,
1575 status
1576 );
1577 }
1578
1579 if let Some(ix) = self.index_for_tool_call(&id) {
1580 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1581 unreachable!()
1582 };
1583
1584 call.update_fields(
1585 update.fields,
1586 language_registry,
1587 path_style,
1588 &self.terminals,
1589 cx,
1590 )?;
1591 call.status = status;
1592
1593 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1594 } else {
1595 let call = ToolCall::from_acp(
1596 update.try_into()?,
1597 status,
1598 language_registry,
1599 self.project.read(cx).path_style(cx),
1600 &self.terminals,
1601 cx,
1602 )?;
1603 self.push_entry(AgentThreadEntry::ToolCall(call), cx);
1604 };
1605
1606 self.resolve_locations(id, cx);
1607 Ok(())
1608 }
1609
1610 fn index_for_tool_call(&self, id: &acp::ToolCallId) -> Option<usize> {
1611 self.entries
1612 .iter()
1613 .enumerate()
1614 .rev()
1615 .find_map(|(index, entry)| {
1616 if let AgentThreadEntry::ToolCall(tool_call) = entry
1617 && &tool_call.id == id
1618 {
1619 Some(index)
1620 } else {
1621 None
1622 }
1623 })
1624 }
1625
1626 fn tool_call_mut(&mut self, id: &acp::ToolCallId) -> Option<(usize, &mut ToolCall)> {
1627 // The tool call we are looking for is typically the last one, or very close to the end.
1628 // At the moment, it doesn't seem like a hashmap would be a good fit for this use case.
1629 self.entries
1630 .iter_mut()
1631 .enumerate()
1632 .rev()
1633 .find_map(|(index, tool_call)| {
1634 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1635 && &tool_call.id == id
1636 {
1637 Some((index, tool_call))
1638 } else {
1639 None
1640 }
1641 })
1642 }
1643
1644 pub fn tool_call(&mut self, id: &acp::ToolCallId) -> Option<(usize, &ToolCall)> {
1645 self.entries
1646 .iter()
1647 .enumerate()
1648 .rev()
1649 .find_map(|(index, tool_call)| {
1650 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1651 && &tool_call.id == id
1652 {
1653 Some((index, tool_call))
1654 } else {
1655 None
1656 }
1657 })
1658 }
1659
1660 pub fn resolve_locations(&mut self, id: acp::ToolCallId, cx: &mut Context<Self>) {
1661 let project = self.project.clone();
1662 let Some((_, tool_call)) = self.tool_call_mut(&id) else {
1663 return;
1664 };
1665 let task = tool_call.resolve_locations(project, cx);
1666 cx.spawn(async move |this, cx| {
1667 let resolved_locations = task.await;
1668
1669 this.update(cx, |this, cx| {
1670 let project = this.project.clone();
1671
1672 for location in resolved_locations.iter().flatten() {
1673 this.shared_buffers
1674 .insert(location.buffer.clone(), location.buffer.read(cx).snapshot());
1675 }
1676 let Some((ix, tool_call)) = this.tool_call_mut(&id) else {
1677 return;
1678 };
1679
1680 if let Some(Some(location)) = resolved_locations.last() {
1681 project.update(cx, |project, cx| {
1682 let should_ignore = if let Some(agent_location) = project
1683 .agent_location()
1684 .filter(|agent_location| agent_location.buffer == location.buffer)
1685 {
1686 let snapshot = location.buffer.read(cx).snapshot();
1687 let old_position = agent_location.position.to_point(&snapshot);
1688 let new_position = location.position.to_point(&snapshot);
1689
1690 // ignore this so that when we get updates from the edit tool
1691 // the position doesn't reset to the startof line
1692 old_position.row == new_position.row
1693 && old_position.column > new_position.column
1694 } else {
1695 false
1696 };
1697 if !should_ignore {
1698 project.set_agent_location(Some(location.into()), cx);
1699 }
1700 });
1701 }
1702
1703 let resolved_locations = resolved_locations
1704 .iter()
1705 .map(|l| l.as_ref().map(|l| AgentLocation::from(l)))
1706 .collect::<Vec<_>>();
1707
1708 if tool_call.resolved_locations != resolved_locations {
1709 tool_call.resolved_locations = resolved_locations;
1710 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1711 }
1712 })
1713 })
1714 .detach();
1715 }
1716
1717 pub fn request_tool_call_authorization(
1718 &mut self,
1719 tool_call: acp::ToolCallUpdate,
1720 options: PermissionOptions,
1721 respect_always_allow_setting: bool,
1722 cx: &mut Context<Self>,
1723 ) -> Result<BoxFuture<'static, acp::RequestPermissionOutcome>> {
1724 let (tx, rx) = oneshot::channel();
1725
1726 if respect_always_allow_setting && AgentSettings::get_global(cx).always_allow_tool_actions {
1727 // Don't use AllowAlways, because then if you were to turn off always_allow_tool_actions,
1728 // some tools would (incorrectly) continue to auto-accept.
1729 if let Some(allow_once_option) = options.allow_once_option_id() {
1730 self.upsert_tool_call_inner(tool_call, ToolCallStatus::Pending, cx)?;
1731 return Ok(async {
1732 acp::RequestPermissionOutcome::Selected(acp::SelectedPermissionOutcome::new(
1733 allow_once_option,
1734 ))
1735 }
1736 .boxed());
1737 }
1738 }
1739
1740 let status = ToolCallStatus::WaitingForConfirmation {
1741 options,
1742 respond_tx: tx,
1743 };
1744
1745 self.upsert_tool_call_inner(tool_call, status, cx)?;
1746 cx.emit(AcpThreadEvent::ToolAuthorizationRequired);
1747
1748 let fut = async {
1749 match rx.await {
1750 Ok(option) => acp::RequestPermissionOutcome::Selected(
1751 acp::SelectedPermissionOutcome::new(option),
1752 ),
1753 Err(oneshot::Canceled) => acp::RequestPermissionOutcome::Cancelled,
1754 }
1755 }
1756 .boxed();
1757
1758 Ok(fut)
1759 }
1760
1761 pub fn authorize_tool_call(
1762 &mut self,
1763 id: acp::ToolCallId,
1764 option_id: acp::PermissionOptionId,
1765 option_kind: acp::PermissionOptionKind,
1766 cx: &mut Context<Self>,
1767 ) {
1768 let Some((ix, call)) = self.tool_call_mut(&id) else {
1769 return;
1770 };
1771
1772 let new_status = match option_kind {
1773 acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways => {
1774 ToolCallStatus::Rejected
1775 }
1776 acp::PermissionOptionKind::AllowOnce | acp::PermissionOptionKind::AllowAlways => {
1777 ToolCallStatus::InProgress
1778 }
1779 _ => ToolCallStatus::InProgress,
1780 };
1781
1782 let curr_status = mem::replace(&mut call.status, new_status);
1783
1784 if let ToolCallStatus::WaitingForConfirmation { respond_tx, .. } = curr_status {
1785 respond_tx.send(option_id).log_err();
1786 } else if cfg!(debug_assertions) {
1787 panic!("tried to authorize an already authorized tool call");
1788 }
1789
1790 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1791 }
1792
1793 pub fn first_tool_awaiting_confirmation(&self) -> Option<&ToolCall> {
1794 let mut first_tool_call = None;
1795
1796 for entry in self.entries.iter().rev() {
1797 match &entry {
1798 AgentThreadEntry::ToolCall(call) => {
1799 if let ToolCallStatus::WaitingForConfirmation { .. } = call.status {
1800 first_tool_call = Some(call);
1801 } else {
1802 continue;
1803 }
1804 }
1805 AgentThreadEntry::UserMessage(_) | AgentThreadEntry::AssistantMessage(_) => {
1806 // Reached the beginning of the turn.
1807 // If we had pending permission requests in the previous turn, they have been cancelled.
1808 break;
1809 }
1810 }
1811 }
1812
1813 first_tool_call
1814 }
1815
1816 pub fn plan(&self) -> &Plan {
1817 &self.plan
1818 }
1819
1820 pub fn update_plan(&mut self, request: acp::Plan, cx: &mut Context<Self>) {
1821 let new_entries_len = request.entries.len();
1822 let mut new_entries = request.entries.into_iter();
1823
1824 // Reuse existing markdown to prevent flickering
1825 for (old, new) in self.plan.entries.iter_mut().zip(new_entries.by_ref()) {
1826 let PlanEntry {
1827 content,
1828 priority,
1829 status,
1830 } = old;
1831 content.update(cx, |old, cx| {
1832 old.replace(new.content, cx);
1833 });
1834 *priority = new.priority;
1835 *status = new.status;
1836 }
1837 for new in new_entries {
1838 self.plan.entries.push(PlanEntry::from_acp(new, cx))
1839 }
1840 self.plan.entries.truncate(new_entries_len);
1841
1842 cx.notify();
1843 }
1844
1845 fn clear_completed_plan_entries(&mut self, cx: &mut Context<Self>) {
1846 self.plan
1847 .entries
1848 .retain(|entry| !matches!(entry.status, acp::PlanEntryStatus::Completed));
1849 cx.notify();
1850 }
1851
1852 #[cfg(any(test, feature = "test-support"))]
1853 pub fn send_raw(
1854 &mut self,
1855 message: &str,
1856 cx: &mut Context<Self>,
1857 ) -> BoxFuture<'static, Result<()>> {
1858 self.send(vec![message.into()], cx)
1859 }
1860
1861 pub fn send(
1862 &mut self,
1863 message: Vec<acp::ContentBlock>,
1864 cx: &mut Context<Self>,
1865 ) -> BoxFuture<'static, Result<()>> {
1866 let block = ContentBlock::new_combined(
1867 message.clone(),
1868 self.project.read(cx).languages().clone(),
1869 self.project.read(cx).path_style(cx),
1870 cx,
1871 );
1872 let request = acp::PromptRequest::new(self.session_id.clone(), message.clone());
1873 let git_store = self.project.read(cx).git_store().clone();
1874
1875 let message_id = if self.connection.truncate(&self.session_id, cx).is_some() {
1876 Some(UserMessageId::new())
1877 } else {
1878 None
1879 };
1880
1881 self.run_turn(cx, async move |this, cx| {
1882 this.update(cx, |this, cx| {
1883 this.push_entry(
1884 AgentThreadEntry::UserMessage(UserMessage {
1885 id: message_id.clone(),
1886 content: block,
1887 chunks: message,
1888 checkpoint: None,
1889 indented: false,
1890 }),
1891 cx,
1892 );
1893 })
1894 .ok();
1895
1896 let old_checkpoint = git_store
1897 .update(cx, |git, cx| git.checkpoint(cx))
1898 .await
1899 .context("failed to get old checkpoint")
1900 .log_err();
1901 this.update(cx, |this, cx| {
1902 if let Some((_ix, message)) = this.last_user_message() {
1903 message.checkpoint = old_checkpoint.map(|git_checkpoint| Checkpoint {
1904 git_checkpoint,
1905 show: false,
1906 });
1907 }
1908 this.connection.prompt(message_id, request, cx)
1909 })?
1910 .await
1911 })
1912 }
1913
1914 pub fn can_retry(&self, cx: &App) -> bool {
1915 self.connection.retry(&self.session_id, cx).is_some()
1916 }
1917
1918 pub fn retry(&mut self, cx: &mut Context<Self>) -> BoxFuture<'static, Result<()>> {
1919 self.run_turn(cx, async move |this, cx| {
1920 this.update(cx, |this, cx| {
1921 this.connection
1922 .retry(&this.session_id, cx)
1923 .map(|retry| retry.run(cx))
1924 })?
1925 .context("retrying a session is not supported")?
1926 .await
1927 })
1928 }
1929
1930 fn run_turn(
1931 &mut self,
1932 cx: &mut Context<Self>,
1933 f: impl 'static + AsyncFnOnce(WeakEntity<Self>, &mut AsyncApp) -> Result<acp::PromptResponse>,
1934 ) -> BoxFuture<'static, Result<()>> {
1935 self.clear_completed_plan_entries(cx);
1936
1937 let (tx, rx) = oneshot::channel();
1938 let cancel_task = self.cancel(cx);
1939
1940 self.send_task = Some(cx.spawn(async move |this, cx| {
1941 cancel_task.await;
1942 tx.send(f(this, cx).await).ok();
1943 }));
1944
1945 cx.spawn(async move |this, cx| {
1946 let response = rx.await;
1947
1948 this.update(cx, |this, cx| this.update_last_checkpoint(cx))?
1949 .await?;
1950
1951 this.update(cx, |this, cx| {
1952 this.project
1953 .update(cx, |project, cx| project.set_agent_location(None, cx));
1954 match response {
1955 Ok(Err(e)) => {
1956 this.send_task.take();
1957 cx.emit(AcpThreadEvent::Error);
1958 Err(e)
1959 }
1960 result => {
1961 let canceled = matches!(
1962 result,
1963 Ok(Ok(acp::PromptResponse {
1964 stop_reason: acp::StopReason::Cancelled,
1965 ..
1966 }))
1967 );
1968
1969 // We only take the task if the current prompt wasn't canceled.
1970 //
1971 // This prompt may have been canceled because another one was sent
1972 // while it was still generating. In these cases, dropping `send_task`
1973 // would cause the next generation to be canceled.
1974 if !canceled {
1975 this.send_task.take();
1976 }
1977
1978 // Handle refusal - distinguish between user prompt and tool call refusals
1979 if let Ok(Ok(acp::PromptResponse {
1980 stop_reason: acp::StopReason::Refusal,
1981 ..
1982 })) = result
1983 {
1984 if let Some((user_msg_ix, _)) = this.last_user_message() {
1985 // Check if there's a completed tool call with results after the last user message
1986 // This indicates the refusal is in response to tool output, not the user's prompt
1987 let has_completed_tool_call_after_user_msg =
1988 this.entries.iter().skip(user_msg_ix + 1).any(|entry| {
1989 if let AgentThreadEntry::ToolCall(tool_call) = entry {
1990 // Check if the tool call has completed and has output
1991 matches!(tool_call.status, ToolCallStatus::Completed)
1992 && tool_call.raw_output.is_some()
1993 } else {
1994 false
1995 }
1996 });
1997
1998 if has_completed_tool_call_after_user_msg {
1999 // Refusal is due to tool output - don't truncate, just notify
2000 // The model refused based on what the tool returned
2001 cx.emit(AcpThreadEvent::Refusal);
2002 } else {
2003 // User prompt was refused - truncate back to before the user message
2004 let range = user_msg_ix..this.entries.len();
2005 if range.start < range.end {
2006 this.entries.truncate(user_msg_ix);
2007 cx.emit(AcpThreadEvent::EntriesRemoved(range));
2008 }
2009 cx.emit(AcpThreadEvent::Refusal);
2010 }
2011 } else {
2012 // No user message found, treat as general refusal
2013 cx.emit(AcpThreadEvent::Refusal);
2014 }
2015 }
2016
2017 cx.emit(AcpThreadEvent::Stopped);
2018 Ok(())
2019 }
2020 }
2021 })?
2022 })
2023 .boxed()
2024 }
2025
2026 pub fn cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
2027 let Some(send_task) = self.send_task.take() else {
2028 return Task::ready(());
2029 };
2030
2031 for entry in self.entries.iter_mut() {
2032 if let AgentThreadEntry::ToolCall(call) = entry {
2033 let cancel = matches!(
2034 call.status,
2035 ToolCallStatus::Pending
2036 | ToolCallStatus::WaitingForConfirmation { .. }
2037 | ToolCallStatus::InProgress
2038 );
2039
2040 if cancel {
2041 call.status = ToolCallStatus::Canceled;
2042 }
2043 }
2044 }
2045
2046 self.connection.cancel(&self.session_id, cx);
2047
2048 // Wait for the send task to complete
2049 cx.foreground_executor().spawn(send_task)
2050 }
2051
2052 /// Restores the git working tree to the state at the given checkpoint (if one exists)
2053 pub fn restore_checkpoint(
2054 &mut self,
2055 id: UserMessageId,
2056 cx: &mut Context<Self>,
2057 ) -> Task<Result<()>> {
2058 let Some((_, message)) = self.user_message_mut(&id) else {
2059 return Task::ready(Err(anyhow!("message not found")));
2060 };
2061
2062 let checkpoint = message
2063 .checkpoint
2064 .as_ref()
2065 .map(|c| c.git_checkpoint.clone());
2066
2067 // Cancel any in-progress generation before restoring
2068 let cancel_task = self.cancel(cx);
2069 let rewind = self.rewind(id.clone(), cx);
2070 let git_store = self.project.read(cx).git_store().clone();
2071
2072 cx.spawn(async move |_, cx| {
2073 cancel_task.await;
2074 rewind.await?;
2075 if let Some(checkpoint) = checkpoint {
2076 git_store
2077 .update(cx, |git, cx| git.restore_checkpoint(checkpoint, cx))
2078 .await?;
2079 }
2080
2081 Ok(())
2082 })
2083 }
2084
2085 /// Rewinds this thread to before the entry at `index`, removing it and all
2086 /// subsequent entries while rejecting any action_log changes made from that point.
2087 /// Unlike `restore_checkpoint`, this method does not restore from git.
2088 pub fn rewind(&mut self, id: UserMessageId, cx: &mut Context<Self>) -> Task<Result<()>> {
2089 let Some(truncate) = self.connection.truncate(&self.session_id, cx) else {
2090 return Task::ready(Err(anyhow!("not supported")));
2091 };
2092
2093 let telemetry = ActionLogTelemetry::from(&*self);
2094 cx.spawn(async move |this, cx| {
2095 cx.update(|cx| truncate.run(id.clone(), cx)).await?;
2096 this.update(cx, |this, cx| {
2097 if let Some((ix, _)) = this.user_message_mut(&id) {
2098 // Collect all terminals from entries that will be removed
2099 let terminals_to_remove: Vec<acp::TerminalId> = this.entries[ix..]
2100 .iter()
2101 .flat_map(|entry| entry.terminals())
2102 .filter_map(|terminal| terminal.read(cx).id().clone().into())
2103 .collect();
2104
2105 let range = ix..this.entries.len();
2106 this.entries.truncate(ix);
2107 cx.emit(AcpThreadEvent::EntriesRemoved(range));
2108
2109 // Kill and remove the terminals
2110 for terminal_id in terminals_to_remove {
2111 if let Some(terminal) = this.terminals.remove(&terminal_id) {
2112 terminal.update(cx, |terminal, cx| {
2113 terminal.kill(cx);
2114 });
2115 }
2116 }
2117 }
2118 this.action_log().update(cx, |action_log, cx| {
2119 action_log.reject_all_edits(Some(telemetry), cx)
2120 })
2121 })?
2122 .await;
2123 Ok(())
2124 })
2125 }
2126
2127 fn update_last_checkpoint(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
2128 let git_store = self.project.read(cx).git_store().clone();
2129
2130 let Some((_, message)) = self.last_user_message() else {
2131 return Task::ready(Ok(()));
2132 };
2133 let Some(user_message_id) = message.id.clone() else {
2134 return Task::ready(Ok(()));
2135 };
2136 let Some(checkpoint) = message.checkpoint.as_ref() else {
2137 return Task::ready(Ok(()));
2138 };
2139 let old_checkpoint = checkpoint.git_checkpoint.clone();
2140
2141 let new_checkpoint = git_store.update(cx, |git, cx| git.checkpoint(cx));
2142 cx.spawn(async move |this, cx| {
2143 let Some(new_checkpoint) = new_checkpoint
2144 .await
2145 .context("failed to get new checkpoint")
2146 .log_err()
2147 else {
2148 return Ok(());
2149 };
2150
2151 let equal = git_store
2152 .update(cx, |git, cx| {
2153 git.compare_checkpoints(old_checkpoint.clone(), new_checkpoint, cx)
2154 })
2155 .await
2156 .unwrap_or(true);
2157
2158 this.update(cx, |this, cx| {
2159 if let Some((ix, message)) = this.user_message_mut(&user_message_id) {
2160 if let Some(checkpoint) = message.checkpoint.as_mut() {
2161 checkpoint.show = !equal;
2162 cx.emit(AcpThreadEvent::EntryUpdated(ix));
2163 }
2164 }
2165 })?;
2166
2167 Ok(())
2168 })
2169 }
2170
2171 fn last_user_message(&mut self) -> Option<(usize, &mut UserMessage)> {
2172 self.entries
2173 .iter_mut()
2174 .enumerate()
2175 .rev()
2176 .find_map(|(ix, entry)| {
2177 if let AgentThreadEntry::UserMessage(message) = entry {
2178 Some((ix, message))
2179 } else {
2180 None
2181 }
2182 })
2183 }
2184
2185 fn user_message_mut(&mut self, id: &UserMessageId) -> Option<(usize, &mut UserMessage)> {
2186 self.entries.iter_mut().enumerate().find_map(|(ix, entry)| {
2187 if let AgentThreadEntry::UserMessage(message) = entry {
2188 if message.id.as_ref() == Some(id) {
2189 Some((ix, message))
2190 } else {
2191 None
2192 }
2193 } else {
2194 None
2195 }
2196 })
2197 }
2198
2199 pub fn read_text_file(
2200 &self,
2201 path: PathBuf,
2202 line: Option<u32>,
2203 limit: Option<u32>,
2204 reuse_shared_snapshot: bool,
2205 cx: &mut Context<Self>,
2206 ) -> Task<Result<String, acp::Error>> {
2207 // Args are 1-based, move to 0-based
2208 let line = line.unwrap_or_default().saturating_sub(1);
2209 let limit = limit.unwrap_or(u32::MAX);
2210 let project = self.project.clone();
2211 let action_log = self.action_log.clone();
2212 cx.spawn(async move |this, cx| {
2213 let load = project.update(cx, |project, cx| {
2214 let path = project
2215 .project_path_for_absolute_path(&path, cx)
2216 .ok_or_else(|| {
2217 acp::Error::resource_not_found(Some(path.display().to_string()))
2218 })?;
2219 Ok::<_, acp::Error>(project.open_buffer(path, cx))
2220 })?;
2221
2222 let buffer = load.await?;
2223
2224 let snapshot = if reuse_shared_snapshot {
2225 this.read_with(cx, |this, _| {
2226 this.shared_buffers.get(&buffer.clone()).cloned()
2227 })
2228 .log_err()
2229 .flatten()
2230 } else {
2231 None
2232 };
2233
2234 let snapshot = if let Some(snapshot) = snapshot {
2235 snapshot
2236 } else {
2237 action_log.update(cx, |action_log, cx| {
2238 action_log.buffer_read(buffer.clone(), cx);
2239 });
2240
2241 let snapshot = buffer.update(cx, |buffer, _| buffer.snapshot());
2242 this.update(cx, |this, _| {
2243 this.shared_buffers.insert(buffer.clone(), snapshot.clone());
2244 })?;
2245 snapshot
2246 };
2247
2248 let max_point = snapshot.max_point();
2249 let start_position = Point::new(line, 0);
2250
2251 if start_position > max_point {
2252 return Err(acp::Error::invalid_params().data(format!(
2253 "Attempting to read beyond the end of the file, line {}:{}",
2254 max_point.row + 1,
2255 max_point.column
2256 )));
2257 }
2258
2259 let start = snapshot.anchor_before(start_position);
2260 let end = snapshot.anchor_before(Point::new(line.saturating_add(limit), 0));
2261
2262 project.update(cx, |project, cx| {
2263 project.set_agent_location(
2264 Some(AgentLocation {
2265 buffer: buffer.downgrade(),
2266 position: start,
2267 }),
2268 cx,
2269 );
2270 });
2271
2272 Ok(snapshot.text_for_range(start..end).collect::<String>())
2273 })
2274 }
2275
2276 pub fn write_text_file(
2277 &self,
2278 path: PathBuf,
2279 content: String,
2280 cx: &mut Context<Self>,
2281 ) -> Task<Result<()>> {
2282 let project = self.project.clone();
2283 let action_log = self.action_log.clone();
2284 cx.spawn(async move |this, cx| {
2285 let load = project.update(cx, |project, cx| {
2286 let path = project
2287 .project_path_for_absolute_path(&path, cx)
2288 .context("invalid path")?;
2289 anyhow::Ok(project.open_buffer(path, cx))
2290 });
2291 let buffer = load?.await?;
2292 let snapshot = this.update(cx, |this, cx| {
2293 this.shared_buffers
2294 .get(&buffer)
2295 .cloned()
2296 .unwrap_or_else(|| buffer.read(cx).snapshot())
2297 })?;
2298 let edits = cx
2299 .background_executor()
2300 .spawn(async move {
2301 let old_text = snapshot.text();
2302 text_diff(old_text.as_str(), &content)
2303 .into_iter()
2304 .map(|(range, replacement)| {
2305 (
2306 snapshot.anchor_after(range.start)
2307 ..snapshot.anchor_before(range.end),
2308 replacement,
2309 )
2310 })
2311 .collect::<Vec<_>>()
2312 })
2313 .await;
2314
2315 project.update(cx, |project, cx| {
2316 project.set_agent_location(
2317 Some(AgentLocation {
2318 buffer: buffer.downgrade(),
2319 position: edits
2320 .last()
2321 .map(|(range, _)| range.end)
2322 .unwrap_or(Anchor::min_for_buffer(buffer.read(cx).remote_id())),
2323 }),
2324 cx,
2325 );
2326 });
2327
2328 let format_on_save = cx.update(|cx| {
2329 action_log.update(cx, |action_log, cx| {
2330 action_log.buffer_read(buffer.clone(), cx);
2331 });
2332
2333 let format_on_save = buffer.update(cx, |buffer, cx| {
2334 buffer.edit(edits, None, cx);
2335
2336 let settings =
2337 language::language_settings::LanguageSettings::for_buffer(buffer, cx);
2338
2339 settings.format_on_save != FormatOnSave::Off
2340 });
2341 action_log.update(cx, |action_log, cx| {
2342 action_log.buffer_edited(buffer.clone(), cx);
2343 });
2344 format_on_save
2345 });
2346
2347 if format_on_save {
2348 let format_task = project.update(cx, |project, cx| {
2349 project.format(
2350 HashSet::from_iter([buffer.clone()]),
2351 LspFormatTarget::Buffers,
2352 false,
2353 FormatTrigger::Save,
2354 cx,
2355 )
2356 });
2357 format_task.await.log_err();
2358
2359 action_log.update(cx, |action_log, cx| {
2360 action_log.buffer_edited(buffer.clone(), cx);
2361 });
2362 }
2363
2364 project
2365 .update(cx, |project, cx| project.save_buffer(buffer, cx))
2366 .await
2367 })
2368 }
2369
2370 pub fn create_terminal(
2371 &self,
2372 command: String,
2373 args: Vec<String>,
2374 extra_env: Vec<acp::EnvVariable>,
2375 cwd: Option<PathBuf>,
2376 output_byte_limit: Option<u64>,
2377 cx: &mut Context<Self>,
2378 ) -> Task<Result<Entity<Terminal>>> {
2379 let env = match &cwd {
2380 Some(dir) => self.project.update(cx, |project, cx| {
2381 project.environment().update(cx, |env, cx| {
2382 env.directory_environment(dir.as_path().into(), cx)
2383 })
2384 }),
2385 None => Task::ready(None).shared(),
2386 };
2387 let env = cx.spawn(async move |_, _| {
2388 let mut env = env.await.unwrap_or_default();
2389 // Disables paging for `git` and hopefully other commands
2390 env.insert("PAGER".into(), "".into());
2391 for var in extra_env {
2392 env.insert(var.name, var.value);
2393 }
2394 env
2395 });
2396
2397 let project = self.project.clone();
2398 let language_registry = project.read(cx).languages().clone();
2399 let is_windows = project.read(cx).path_style(cx).is_windows();
2400
2401 let terminal_id = acp::TerminalId::new(Uuid::new_v4().to_string());
2402 let terminal_task = cx.spawn({
2403 let terminal_id = terminal_id.clone();
2404 async move |_this, cx| {
2405 let env = env.await;
2406 let shell = project
2407 .update(cx, |project, cx| {
2408 project
2409 .remote_client()
2410 .and_then(|r| r.read(cx).default_system_shell())
2411 })
2412 .unwrap_or_else(|| get_default_system_shell_preferring_bash());
2413 let (task_command, task_args) =
2414 ShellBuilder::new(&Shell::Program(shell), is_windows)
2415 .redirect_stdin_to_dev_null()
2416 .build(Some(command.clone()), &args);
2417 let terminal = project
2418 .update(cx, |project, cx| {
2419 project.create_terminal_task(
2420 task::SpawnInTerminal {
2421 command: Some(task_command),
2422 args: task_args,
2423 cwd: cwd.clone(),
2424 env,
2425 ..Default::default()
2426 },
2427 cx,
2428 )
2429 })
2430 .await?;
2431
2432 anyhow::Ok(cx.new(|cx| {
2433 Terminal::new(
2434 terminal_id,
2435 &format!("{} {}", command, args.join(" ")),
2436 cwd,
2437 output_byte_limit.map(|l| l as usize),
2438 terminal,
2439 language_registry,
2440 cx,
2441 )
2442 }))
2443 }
2444 });
2445
2446 cx.spawn(async move |this, cx| {
2447 let terminal = terminal_task.await?;
2448 this.update(cx, |this, _cx| {
2449 this.terminals.insert(terminal_id, terminal.clone());
2450 terminal
2451 })
2452 })
2453 }
2454
2455 pub fn kill_terminal(
2456 &mut self,
2457 terminal_id: acp::TerminalId,
2458 cx: &mut Context<Self>,
2459 ) -> Result<()> {
2460 self.terminals
2461 .get(&terminal_id)
2462 .context("Terminal not found")?
2463 .update(cx, |terminal, cx| {
2464 terminal.kill(cx);
2465 });
2466
2467 Ok(())
2468 }
2469
2470 pub fn release_terminal(
2471 &mut self,
2472 terminal_id: acp::TerminalId,
2473 cx: &mut Context<Self>,
2474 ) -> Result<()> {
2475 self.terminals
2476 .remove(&terminal_id)
2477 .context("Terminal not found")?
2478 .update(cx, |terminal, cx| {
2479 terminal.kill(cx);
2480 });
2481
2482 Ok(())
2483 }
2484
2485 pub fn terminal(&self, terminal_id: acp::TerminalId) -> Result<Entity<Terminal>> {
2486 self.terminals
2487 .get(&terminal_id)
2488 .context("Terminal not found")
2489 .cloned()
2490 }
2491
2492 pub fn to_markdown(&self, cx: &App) -> String {
2493 self.entries.iter().map(|e| e.to_markdown(cx)).collect()
2494 }
2495
2496 pub fn emit_load_error(&mut self, error: LoadError, cx: &mut Context<Self>) {
2497 cx.emit(AcpThreadEvent::LoadError(error));
2498 }
2499
2500 pub fn register_terminal_created(
2501 &mut self,
2502 terminal_id: acp::TerminalId,
2503 command_label: String,
2504 working_dir: Option<PathBuf>,
2505 output_byte_limit: Option<u64>,
2506 terminal: Entity<::terminal::Terminal>,
2507 cx: &mut Context<Self>,
2508 ) -> Entity<Terminal> {
2509 let language_registry = self.project.read(cx).languages().clone();
2510
2511 let entity = cx.new(|cx| {
2512 Terminal::new(
2513 terminal_id.clone(),
2514 &command_label,
2515 working_dir.clone(),
2516 output_byte_limit.map(|l| l as usize),
2517 terminal,
2518 language_registry,
2519 cx,
2520 )
2521 });
2522 self.terminals.insert(terminal_id.clone(), entity.clone());
2523 entity
2524 }
2525}
2526
2527fn markdown_for_raw_output(
2528 raw_output: &serde_json::Value,
2529 language_registry: &Arc<LanguageRegistry>,
2530 cx: &mut App,
2531) -> Option<Entity<Markdown>> {
2532 match raw_output {
2533 serde_json::Value::Null => None,
2534 serde_json::Value::Bool(value) => Some(cx.new(|cx| {
2535 Markdown::new(
2536 value.to_string().into(),
2537 Some(language_registry.clone()),
2538 None,
2539 cx,
2540 )
2541 })),
2542 serde_json::Value::Number(value) => Some(cx.new(|cx| {
2543 Markdown::new(
2544 value.to_string().into(),
2545 Some(language_registry.clone()),
2546 None,
2547 cx,
2548 )
2549 })),
2550 serde_json::Value::String(value) => Some(cx.new(|cx| {
2551 Markdown::new(
2552 value.clone().into(),
2553 Some(language_registry.clone()),
2554 None,
2555 cx,
2556 )
2557 })),
2558 value => Some(cx.new(|cx| {
2559 let pretty_json = to_string_pretty(value).unwrap_or_else(|_| value.to_string());
2560
2561 Markdown::new(
2562 format!("```json\n{}\n```", pretty_json).into(),
2563 Some(language_registry.clone()),
2564 None,
2565 cx,
2566 )
2567 })),
2568 }
2569}
2570
2571#[cfg(test)]
2572mod tests {
2573 use super::*;
2574 use anyhow::anyhow;
2575 use futures::{channel::mpsc, future::LocalBoxFuture, select};
2576 use gpui::{App, AsyncApp, TestAppContext, WeakEntity};
2577 use indoc::indoc;
2578 use project::{FakeFs, Fs};
2579 use rand::{distr, prelude::*};
2580 use serde_json::json;
2581 use settings::SettingsStore;
2582 use smol::stream::StreamExt as _;
2583 use std::{
2584 any::Any,
2585 cell::RefCell,
2586 path::Path,
2587 rc::Rc,
2588 sync::atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
2589 time::Duration,
2590 };
2591 use util::path;
2592
2593 fn init_test(cx: &mut TestAppContext) {
2594 env_logger::try_init().ok();
2595 cx.update(|cx| {
2596 let settings_store = SettingsStore::test(cx);
2597 cx.set_global(settings_store);
2598 });
2599 }
2600
2601 #[gpui::test]
2602 async fn test_terminal_output_buffered_before_created_renders(cx: &mut gpui::TestAppContext) {
2603 init_test(cx);
2604
2605 let fs = FakeFs::new(cx.executor());
2606 let project = Project::test(fs, [], cx).await;
2607 let connection = Rc::new(FakeAgentConnection::new());
2608 let thread = cx
2609 .update(|cx| connection.new_thread(project, std::path::Path::new(path!("/test")), cx))
2610 .await
2611 .unwrap();
2612
2613 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
2614
2615 // Send Output BEFORE Created - should be buffered by acp_thread
2616 thread.update(cx, |thread, cx| {
2617 thread.on_terminal_provider_event(
2618 TerminalProviderEvent::Output {
2619 terminal_id: terminal_id.clone(),
2620 data: b"hello buffered".to_vec(),
2621 },
2622 cx,
2623 );
2624 });
2625
2626 // Create a display-only terminal and then send Created
2627 let lower = cx.new(|cx| {
2628 let builder = ::terminal::TerminalBuilder::new_display_only(
2629 ::terminal::terminal_settings::CursorShape::default(),
2630 ::terminal::terminal_settings::AlternateScroll::On,
2631 None,
2632 0,
2633 cx.background_executor(),
2634 PathStyle::local(),
2635 )
2636 .unwrap();
2637 builder.subscribe(cx)
2638 });
2639
2640 thread.update(cx, |thread, cx| {
2641 thread.on_terminal_provider_event(
2642 TerminalProviderEvent::Created {
2643 terminal_id: terminal_id.clone(),
2644 label: "Buffered Test".to_string(),
2645 cwd: None,
2646 output_byte_limit: None,
2647 terminal: lower.clone(),
2648 },
2649 cx,
2650 );
2651 });
2652
2653 // After Created, buffered Output should have been flushed into the renderer
2654 let content = thread.read_with(cx, |thread, cx| {
2655 let term = thread.terminal(terminal_id.clone()).unwrap();
2656 term.read_with(cx, |t, cx| t.inner().read(cx).get_content())
2657 });
2658
2659 assert!(
2660 content.contains("hello buffered"),
2661 "expected buffered output to render, got: {content}"
2662 );
2663 }
2664
2665 #[gpui::test]
2666 async fn test_terminal_output_and_exit_buffered_before_created(cx: &mut gpui::TestAppContext) {
2667 init_test(cx);
2668
2669 let fs = FakeFs::new(cx.executor());
2670 let project = Project::test(fs, [], cx).await;
2671 let connection = Rc::new(FakeAgentConnection::new());
2672 let thread = cx
2673 .update(|cx| connection.new_thread(project, std::path::Path::new(path!("/test")), cx))
2674 .await
2675 .unwrap();
2676
2677 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
2678
2679 // Send Output BEFORE Created
2680 thread.update(cx, |thread, cx| {
2681 thread.on_terminal_provider_event(
2682 TerminalProviderEvent::Output {
2683 terminal_id: terminal_id.clone(),
2684 data: b"pre-exit data".to_vec(),
2685 },
2686 cx,
2687 );
2688 });
2689
2690 // Send Exit BEFORE Created
2691 thread.update(cx, |thread, cx| {
2692 thread.on_terminal_provider_event(
2693 TerminalProviderEvent::Exit {
2694 terminal_id: terminal_id.clone(),
2695 status: acp::TerminalExitStatus::new().exit_code(0),
2696 },
2697 cx,
2698 );
2699 });
2700
2701 // Now create a display-only lower-level terminal and send Created
2702 let lower = cx.new(|cx| {
2703 let builder = ::terminal::TerminalBuilder::new_display_only(
2704 ::terminal::terminal_settings::CursorShape::default(),
2705 ::terminal::terminal_settings::AlternateScroll::On,
2706 None,
2707 0,
2708 cx.background_executor(),
2709 PathStyle::local(),
2710 )
2711 .unwrap();
2712 builder.subscribe(cx)
2713 });
2714
2715 thread.update(cx, |thread, cx| {
2716 thread.on_terminal_provider_event(
2717 TerminalProviderEvent::Created {
2718 terminal_id: terminal_id.clone(),
2719 label: "Buffered Exit Test".to_string(),
2720 cwd: None,
2721 output_byte_limit: None,
2722 terminal: lower.clone(),
2723 },
2724 cx,
2725 );
2726 });
2727
2728 // Output should be present after Created (flushed from buffer)
2729 let content = thread.read_with(cx, |thread, cx| {
2730 let term = thread.terminal(terminal_id.clone()).unwrap();
2731 term.read_with(cx, |t, cx| t.inner().read(cx).get_content())
2732 });
2733
2734 assert!(
2735 content.contains("pre-exit data"),
2736 "expected pre-exit data to render, got: {content}"
2737 );
2738 }
2739
2740 /// Test that killing a terminal via Terminal::kill properly:
2741 /// 1. Causes wait_for_exit to complete (doesn't hang forever)
2742 /// 2. The underlying terminal still has the output that was written before the kill
2743 ///
2744 /// This test verifies that the fix to kill_active_task (which now also kills
2745 /// the shell process in addition to the foreground process) properly allows
2746 /// wait_for_exit to complete instead of hanging indefinitely.
2747 #[cfg(unix)]
2748 #[gpui::test]
2749 async fn test_terminal_kill_allows_wait_for_exit_to_complete(cx: &mut gpui::TestAppContext) {
2750 use std::collections::HashMap;
2751 use task::Shell;
2752 use util::shell_builder::ShellBuilder;
2753
2754 init_test(cx);
2755 cx.executor().allow_parking();
2756
2757 let fs = FakeFs::new(cx.executor());
2758 let project = Project::test(fs, [], cx).await;
2759 let connection = Rc::new(FakeAgentConnection::new());
2760 let thread = cx
2761 .update(|cx| connection.new_thread(project.clone(), Path::new(path!("/test")), cx))
2762 .await
2763 .unwrap();
2764
2765 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
2766
2767 // Create a real PTY terminal that runs a command which prints output then sleeps
2768 // We use printf instead of echo and chain with && sleep to ensure proper execution
2769 let (completion_tx, _completion_rx) = smol::channel::unbounded();
2770 let (program, args) = ShellBuilder::new(&Shell::System, false).build(
2771 Some("printf 'output_before_kill\\n' && sleep 60".to_owned()),
2772 &[],
2773 );
2774
2775 let builder = cx
2776 .update(|cx| {
2777 ::terminal::TerminalBuilder::new(
2778 None,
2779 None,
2780 task::Shell::WithArguments {
2781 program,
2782 args,
2783 title_override: None,
2784 },
2785 HashMap::default(),
2786 ::terminal::terminal_settings::CursorShape::default(),
2787 ::terminal::terminal_settings::AlternateScroll::On,
2788 None,
2789 vec![],
2790 0,
2791 false,
2792 0,
2793 Some(completion_tx),
2794 cx,
2795 vec![],
2796 PathStyle::local(),
2797 )
2798 })
2799 .await
2800 .unwrap();
2801
2802 let lower_terminal = cx.new(|cx| builder.subscribe(cx));
2803
2804 // Create the acp_thread Terminal wrapper
2805 thread.update(cx, |thread, cx| {
2806 thread.on_terminal_provider_event(
2807 TerminalProviderEvent::Created {
2808 terminal_id: terminal_id.clone(),
2809 label: "printf output_before_kill && sleep 60".to_string(),
2810 cwd: None,
2811 output_byte_limit: None,
2812 terminal: lower_terminal.clone(),
2813 },
2814 cx,
2815 );
2816 });
2817
2818 // Wait for the printf command to execute and produce output
2819 // Use real time since parking is enabled
2820 cx.executor().timer(Duration::from_millis(500)).await;
2821
2822 // Get the acp_thread Terminal and kill it
2823 let wait_for_exit = thread.update(cx, |thread, cx| {
2824 let term = thread.terminals.get(&terminal_id).unwrap();
2825 let wait_for_exit = term.read(cx).wait_for_exit();
2826 term.update(cx, |term, cx| {
2827 term.kill(cx);
2828 });
2829 wait_for_exit
2830 });
2831
2832 // KEY ASSERTION: wait_for_exit should complete within a reasonable time (not hang).
2833 // Before the fix to kill_active_task, this would hang forever because
2834 // only the foreground process was killed, not the shell, so the PTY
2835 // child never exited and wait_for_completed_task never completed.
2836 let exit_result = futures::select! {
2837 result = futures::FutureExt::fuse(wait_for_exit) => Some(result),
2838 _ = futures::FutureExt::fuse(cx.background_executor.timer(Duration::from_secs(5))) => None,
2839 };
2840
2841 assert!(
2842 exit_result.is_some(),
2843 "wait_for_exit should complete after kill, but it timed out. \
2844 This indicates kill_active_task is not properly killing the shell process."
2845 );
2846
2847 // Give the system a chance to process any pending updates
2848 cx.run_until_parked();
2849
2850 // Verify that the underlying terminal still has the output that was
2851 // written before the kill. This verifies that killing doesn't lose output.
2852 let inner_content = thread.read_with(cx, |thread, cx| {
2853 let term = thread.terminals.get(&terminal_id).unwrap();
2854 term.read(cx).inner().read(cx).get_content()
2855 });
2856
2857 assert!(
2858 inner_content.contains("output_before_kill"),
2859 "Underlying terminal should contain output from before kill, got: {}",
2860 inner_content
2861 );
2862 }
2863
2864 #[gpui::test]
2865 async fn test_push_user_content_block(cx: &mut gpui::TestAppContext) {
2866 init_test(cx);
2867
2868 let fs = FakeFs::new(cx.executor());
2869 let project = Project::test(fs, [], cx).await;
2870 let connection = Rc::new(FakeAgentConnection::new());
2871 let thread = cx
2872 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2873 .await
2874 .unwrap();
2875
2876 // Test creating a new user message
2877 thread.update(cx, |thread, cx| {
2878 thread.push_user_content_block(None, "Hello, ".into(), cx);
2879 });
2880
2881 thread.update(cx, |thread, cx| {
2882 assert_eq!(thread.entries.len(), 1);
2883 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
2884 assert_eq!(user_msg.id, None);
2885 assert_eq!(user_msg.content.to_markdown(cx), "Hello, ");
2886 } else {
2887 panic!("Expected UserMessage");
2888 }
2889 });
2890
2891 // Test appending to existing user message
2892 let message_1_id = UserMessageId::new();
2893 thread.update(cx, |thread, cx| {
2894 thread.push_user_content_block(Some(message_1_id.clone()), "world!".into(), cx);
2895 });
2896
2897 thread.update(cx, |thread, cx| {
2898 assert_eq!(thread.entries.len(), 1);
2899 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
2900 assert_eq!(user_msg.id, Some(message_1_id));
2901 assert_eq!(user_msg.content.to_markdown(cx), "Hello, world!");
2902 } else {
2903 panic!("Expected UserMessage");
2904 }
2905 });
2906
2907 // Test creating new user message after assistant message
2908 thread.update(cx, |thread, cx| {
2909 thread.push_assistant_content_block("Assistant response".into(), false, cx);
2910 });
2911
2912 let message_2_id = UserMessageId::new();
2913 thread.update(cx, |thread, cx| {
2914 thread.push_user_content_block(
2915 Some(message_2_id.clone()),
2916 "New user message".into(),
2917 cx,
2918 );
2919 });
2920
2921 thread.update(cx, |thread, cx| {
2922 assert_eq!(thread.entries.len(), 3);
2923 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[2] {
2924 assert_eq!(user_msg.id, Some(message_2_id));
2925 assert_eq!(user_msg.content.to_markdown(cx), "New user message");
2926 } else {
2927 panic!("Expected UserMessage at index 2");
2928 }
2929 });
2930 }
2931
2932 #[gpui::test]
2933 async fn test_thinking_concatenation(cx: &mut gpui::TestAppContext) {
2934 init_test(cx);
2935
2936 let fs = FakeFs::new(cx.executor());
2937 let project = Project::test(fs, [], cx).await;
2938 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
2939 |_, thread, mut cx| {
2940 async move {
2941 thread.update(&mut cx, |thread, cx| {
2942 thread
2943 .handle_session_update(
2944 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(
2945 "Thinking ".into(),
2946 )),
2947 cx,
2948 )
2949 .unwrap();
2950 thread
2951 .handle_session_update(
2952 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(
2953 "hard!".into(),
2954 )),
2955 cx,
2956 )
2957 .unwrap();
2958 })?;
2959 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
2960 }
2961 .boxed_local()
2962 },
2963 ));
2964
2965 let thread = cx
2966 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2967 .await
2968 .unwrap();
2969
2970 thread
2971 .update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
2972 .await
2973 .unwrap();
2974
2975 let output = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
2976 assert_eq!(
2977 output,
2978 indoc! {r#"
2979 ## User
2980
2981 Hello from Zed!
2982
2983 ## Assistant
2984
2985 <thinking>
2986 Thinking hard!
2987 </thinking>
2988
2989 "#}
2990 );
2991 }
2992
2993 #[gpui::test]
2994 async fn test_edits_concurrently_to_user(cx: &mut TestAppContext) {
2995 init_test(cx);
2996
2997 let fs = FakeFs::new(cx.executor());
2998 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\n"}))
2999 .await;
3000 let project = Project::test(fs.clone(), [], cx).await;
3001 let (read_file_tx, read_file_rx) = oneshot::channel::<()>();
3002 let read_file_tx = Rc::new(RefCell::new(Some(read_file_tx)));
3003 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
3004 move |_, thread, mut cx| {
3005 let read_file_tx = read_file_tx.clone();
3006 async move {
3007 let content = thread
3008 .update(&mut cx, |thread, cx| {
3009 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
3010 })
3011 .unwrap()
3012 .await
3013 .unwrap();
3014 assert_eq!(content, "one\ntwo\nthree\n");
3015 read_file_tx.take().unwrap().send(()).unwrap();
3016 thread
3017 .update(&mut cx, |thread, cx| {
3018 thread.write_text_file(
3019 path!("/tmp/foo").into(),
3020 "one\ntwo\nthree\nfour\nfive\n".to_string(),
3021 cx,
3022 )
3023 })
3024 .unwrap()
3025 .await
3026 .unwrap();
3027 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3028 }
3029 .boxed_local()
3030 },
3031 ));
3032
3033 let (worktree, pathbuf) = project
3034 .update(cx, |project, cx| {
3035 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
3036 })
3037 .await
3038 .unwrap();
3039 let buffer = project
3040 .update(cx, |project, cx| {
3041 project.open_buffer((worktree.read(cx).id(), pathbuf), cx)
3042 })
3043 .await
3044 .unwrap();
3045
3046 let thread = cx
3047 .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
3048 .await
3049 .unwrap();
3050
3051 let request = thread.update(cx, |thread, cx| {
3052 thread.send_raw("Extend the count in /tmp/foo", cx)
3053 });
3054 read_file_rx.await.ok();
3055 buffer.update(cx, |buffer, cx| {
3056 buffer.edit([(0..0, "zero\n".to_string())], None, cx);
3057 });
3058 cx.run_until_parked();
3059 assert_eq!(
3060 buffer.read_with(cx, |buffer, _| buffer.text()),
3061 "zero\none\ntwo\nthree\nfour\nfive\n"
3062 );
3063 assert_eq!(
3064 String::from_utf8(fs.read_file_sync(path!("/tmp/foo")).unwrap()).unwrap(),
3065 "zero\none\ntwo\nthree\nfour\nfive\n"
3066 );
3067 request.await.unwrap();
3068 }
3069
3070 #[gpui::test]
3071 async fn test_reading_from_line(cx: &mut TestAppContext) {
3072 init_test(cx);
3073
3074 let fs = FakeFs::new(cx.executor());
3075 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\nfour\n"}))
3076 .await;
3077 let project = Project::test(fs.clone(), [], cx).await;
3078 project
3079 .update(cx, |project, cx| {
3080 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
3081 })
3082 .await
3083 .unwrap();
3084
3085 let connection = Rc::new(FakeAgentConnection::new());
3086
3087 let thread = cx
3088 .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
3089 .await
3090 .unwrap();
3091
3092 // Whole file
3093 let content = thread
3094 .update(cx, |thread, cx| {
3095 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
3096 })
3097 .await
3098 .unwrap();
3099
3100 assert_eq!(content, "one\ntwo\nthree\nfour\n");
3101
3102 // Only start line
3103 let content = thread
3104 .update(cx, |thread, cx| {
3105 thread.read_text_file(path!("/tmp/foo").into(), Some(3), None, false, cx)
3106 })
3107 .await
3108 .unwrap();
3109
3110 assert_eq!(content, "three\nfour\n");
3111
3112 // Only limit
3113 let content = thread
3114 .update(cx, |thread, cx| {
3115 thread.read_text_file(path!("/tmp/foo").into(), None, Some(2), false, cx)
3116 })
3117 .await
3118 .unwrap();
3119
3120 assert_eq!(content, "one\ntwo\n");
3121
3122 // Range
3123 let content = thread
3124 .update(cx, |thread, cx| {
3125 thread.read_text_file(path!("/tmp/foo").into(), Some(2), Some(2), false, cx)
3126 })
3127 .await
3128 .unwrap();
3129
3130 assert_eq!(content, "two\nthree\n");
3131
3132 // Invalid
3133 let err = thread
3134 .update(cx, |thread, cx| {
3135 thread.read_text_file(path!("/tmp/foo").into(), Some(6), Some(2), false, cx)
3136 })
3137 .await
3138 .unwrap_err();
3139
3140 assert_eq!(
3141 err.to_string(),
3142 "Invalid params: \"Attempting to read beyond the end of the file, line 5:0\""
3143 );
3144 }
3145
3146 #[gpui::test]
3147 async fn test_reading_empty_file(cx: &mut TestAppContext) {
3148 init_test(cx);
3149
3150 let fs = FakeFs::new(cx.executor());
3151 fs.insert_tree(path!("/tmp"), json!({"foo": ""})).await;
3152 let project = Project::test(fs.clone(), [], cx).await;
3153 project
3154 .update(cx, |project, cx| {
3155 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
3156 })
3157 .await
3158 .unwrap();
3159
3160 let connection = Rc::new(FakeAgentConnection::new());
3161
3162 let thread = cx
3163 .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
3164 .await
3165 .unwrap();
3166
3167 // Whole file
3168 let content = thread
3169 .update(cx, |thread, cx| {
3170 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
3171 })
3172 .await
3173 .unwrap();
3174
3175 assert_eq!(content, "");
3176
3177 // Only start line
3178 let content = thread
3179 .update(cx, |thread, cx| {
3180 thread.read_text_file(path!("/tmp/foo").into(), Some(1), None, false, cx)
3181 })
3182 .await
3183 .unwrap();
3184
3185 assert_eq!(content, "");
3186
3187 // Only limit
3188 let content = thread
3189 .update(cx, |thread, cx| {
3190 thread.read_text_file(path!("/tmp/foo").into(), None, Some(2), false, cx)
3191 })
3192 .await
3193 .unwrap();
3194
3195 assert_eq!(content, "");
3196
3197 // Range
3198 let content = thread
3199 .update(cx, |thread, cx| {
3200 thread.read_text_file(path!("/tmp/foo").into(), Some(1), Some(1), false, cx)
3201 })
3202 .await
3203 .unwrap();
3204
3205 assert_eq!(content, "");
3206
3207 // Invalid
3208 let err = thread
3209 .update(cx, |thread, cx| {
3210 thread.read_text_file(path!("/tmp/foo").into(), Some(5), Some(2), false, cx)
3211 })
3212 .await
3213 .unwrap_err();
3214
3215 assert_eq!(
3216 err.to_string(),
3217 "Invalid params: \"Attempting to read beyond the end of the file, line 1:0\""
3218 );
3219 }
3220 #[gpui::test]
3221 async fn test_reading_non_existing_file(cx: &mut TestAppContext) {
3222 init_test(cx);
3223
3224 let fs = FakeFs::new(cx.executor());
3225 fs.insert_tree(path!("/tmp"), json!({})).await;
3226 let project = Project::test(fs.clone(), [], cx).await;
3227 project
3228 .update(cx, |project, cx| {
3229 project.find_or_create_worktree(path!("/tmp"), true, cx)
3230 })
3231 .await
3232 .unwrap();
3233
3234 let connection = Rc::new(FakeAgentConnection::new());
3235
3236 let thread = cx
3237 .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
3238 .await
3239 .unwrap();
3240
3241 // Out of project file
3242 let err = thread
3243 .update(cx, |thread, cx| {
3244 thread.read_text_file(path!("/foo").into(), None, None, false, cx)
3245 })
3246 .await
3247 .unwrap_err();
3248
3249 assert_eq!(err.code, acp::ErrorCode::ResourceNotFound);
3250 }
3251
3252 #[gpui::test]
3253 async fn test_succeeding_canceled_toolcall(cx: &mut TestAppContext) {
3254 init_test(cx);
3255
3256 let fs = FakeFs::new(cx.executor());
3257 let project = Project::test(fs, [], cx).await;
3258 let id = acp::ToolCallId::new("test");
3259
3260 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3261 let id = id.clone();
3262 move |_, thread, mut cx| {
3263 let id = id.clone();
3264 async move {
3265 thread
3266 .update(&mut cx, |thread, cx| {
3267 thread.handle_session_update(
3268 acp::SessionUpdate::ToolCall(
3269 acp::ToolCall::new(id.clone(), "Label")
3270 .kind(acp::ToolKind::Fetch)
3271 .status(acp::ToolCallStatus::InProgress),
3272 ),
3273 cx,
3274 )
3275 })
3276 .unwrap()
3277 .unwrap();
3278 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3279 }
3280 .boxed_local()
3281 }
3282 }));
3283
3284 let thread = cx
3285 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3286 .await
3287 .unwrap();
3288
3289 let request = thread.update(cx, |thread, cx| {
3290 thread.send_raw("Fetch https://example.com", cx)
3291 });
3292
3293 run_until_first_tool_call(&thread, cx).await;
3294
3295 thread.read_with(cx, |thread, _| {
3296 assert!(matches!(
3297 thread.entries[1],
3298 AgentThreadEntry::ToolCall(ToolCall {
3299 status: ToolCallStatus::InProgress,
3300 ..
3301 })
3302 ));
3303 });
3304
3305 thread.update(cx, |thread, cx| thread.cancel(cx)).await;
3306
3307 thread.read_with(cx, |thread, _| {
3308 assert!(matches!(
3309 &thread.entries[1],
3310 AgentThreadEntry::ToolCall(ToolCall {
3311 status: ToolCallStatus::Canceled,
3312 ..
3313 })
3314 ));
3315 });
3316
3317 thread
3318 .update(cx, |thread, cx| {
3319 thread.handle_session_update(
3320 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
3321 id,
3322 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
3323 )),
3324 cx,
3325 )
3326 })
3327 .unwrap();
3328
3329 request.await.unwrap();
3330
3331 thread.read_with(cx, |thread, _| {
3332 assert!(matches!(
3333 thread.entries[1],
3334 AgentThreadEntry::ToolCall(ToolCall {
3335 status: ToolCallStatus::Completed,
3336 ..
3337 })
3338 ));
3339 });
3340 }
3341
3342 #[gpui::test]
3343 async fn test_no_pending_edits_if_tool_calls_are_completed(cx: &mut TestAppContext) {
3344 init_test(cx);
3345 let fs = FakeFs::new(cx.background_executor.clone());
3346 fs.insert_tree(path!("/test"), json!({})).await;
3347 let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
3348
3349 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3350 move |_, thread, mut cx| {
3351 async move {
3352 thread
3353 .update(&mut cx, |thread, cx| {
3354 thread.handle_session_update(
3355 acp::SessionUpdate::ToolCall(
3356 acp::ToolCall::new("test", "Label")
3357 .kind(acp::ToolKind::Edit)
3358 .status(acp::ToolCallStatus::Completed)
3359 .content(vec![acp::ToolCallContent::Diff(acp::Diff::new(
3360 "/test/test.txt",
3361 "foo",
3362 ))]),
3363 ),
3364 cx,
3365 )
3366 })
3367 .unwrap()
3368 .unwrap();
3369 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3370 }
3371 .boxed_local()
3372 }
3373 }));
3374
3375 let thread = cx
3376 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3377 .await
3378 .unwrap();
3379
3380 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Hi".into()], cx)))
3381 .await
3382 .unwrap();
3383
3384 assert!(cx.read(|cx| !thread.read(cx).has_pending_edit_tool_calls()));
3385 }
3386
3387 #[gpui::test(iterations = 10)]
3388 async fn test_checkpoints(cx: &mut TestAppContext) {
3389 init_test(cx);
3390 let fs = FakeFs::new(cx.background_executor.clone());
3391 fs.insert_tree(
3392 path!("/test"),
3393 json!({
3394 ".git": {}
3395 }),
3396 )
3397 .await;
3398 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
3399
3400 let simulate_changes = Arc::new(AtomicBool::new(true));
3401 let next_filename = Arc::new(AtomicUsize::new(0));
3402 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3403 let simulate_changes = simulate_changes.clone();
3404 let next_filename = next_filename.clone();
3405 let fs = fs.clone();
3406 move |request, thread, mut cx| {
3407 let fs = fs.clone();
3408 let simulate_changes = simulate_changes.clone();
3409 let next_filename = next_filename.clone();
3410 async move {
3411 if simulate_changes.load(SeqCst) {
3412 let filename = format!("/test/file-{}", next_filename.fetch_add(1, SeqCst));
3413 fs.write(Path::new(&filename), b"").await?;
3414 }
3415
3416 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
3417 panic!("expected text content block");
3418 };
3419 thread.update(&mut cx, |thread, cx| {
3420 thread
3421 .handle_session_update(
3422 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(
3423 content.text.to_uppercase().into(),
3424 )),
3425 cx,
3426 )
3427 .unwrap();
3428 })?;
3429 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3430 }
3431 .boxed_local()
3432 }
3433 }));
3434 let thread = cx
3435 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3436 .await
3437 .unwrap();
3438
3439 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Lorem".into()], cx)))
3440 .await
3441 .unwrap();
3442 thread.read_with(cx, |thread, cx| {
3443 assert_eq!(
3444 thread.to_markdown(cx),
3445 indoc! {"
3446 ## User (checkpoint)
3447
3448 Lorem
3449
3450 ## Assistant
3451
3452 LOREM
3453
3454 "}
3455 );
3456 });
3457 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
3458
3459 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["ipsum".into()], cx)))
3460 .await
3461 .unwrap();
3462 thread.read_with(cx, |thread, cx| {
3463 assert_eq!(
3464 thread.to_markdown(cx),
3465 indoc! {"
3466 ## User (checkpoint)
3467
3468 Lorem
3469
3470 ## Assistant
3471
3472 LOREM
3473
3474 ## User (checkpoint)
3475
3476 ipsum
3477
3478 ## Assistant
3479
3480 IPSUM
3481
3482 "}
3483 );
3484 });
3485 assert_eq!(
3486 fs.files(),
3487 vec![
3488 Path::new(path!("/test/file-0")),
3489 Path::new(path!("/test/file-1"))
3490 ]
3491 );
3492
3493 // Checkpoint isn't stored when there are no changes.
3494 simulate_changes.store(false, SeqCst);
3495 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["dolor".into()], cx)))
3496 .await
3497 .unwrap();
3498 thread.read_with(cx, |thread, cx| {
3499 assert_eq!(
3500 thread.to_markdown(cx),
3501 indoc! {"
3502 ## User (checkpoint)
3503
3504 Lorem
3505
3506 ## Assistant
3507
3508 LOREM
3509
3510 ## User (checkpoint)
3511
3512 ipsum
3513
3514 ## Assistant
3515
3516 IPSUM
3517
3518 ## User
3519
3520 dolor
3521
3522 ## Assistant
3523
3524 DOLOR
3525
3526 "}
3527 );
3528 });
3529 assert_eq!(
3530 fs.files(),
3531 vec![
3532 Path::new(path!("/test/file-0")),
3533 Path::new(path!("/test/file-1"))
3534 ]
3535 );
3536
3537 // Rewinding the conversation truncates the history and restores the checkpoint.
3538 thread
3539 .update(cx, |thread, cx| {
3540 let AgentThreadEntry::UserMessage(message) = &thread.entries[2] else {
3541 panic!("unexpected entries {:?}", thread.entries)
3542 };
3543 thread.restore_checkpoint(message.id.clone().unwrap(), cx)
3544 })
3545 .await
3546 .unwrap();
3547 thread.read_with(cx, |thread, cx| {
3548 assert_eq!(
3549 thread.to_markdown(cx),
3550 indoc! {"
3551 ## User (checkpoint)
3552
3553 Lorem
3554
3555 ## Assistant
3556
3557 LOREM
3558
3559 "}
3560 );
3561 });
3562 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
3563 }
3564
3565 #[gpui::test]
3566 async fn test_tool_result_refusal(cx: &mut TestAppContext) {
3567 use std::sync::atomic::AtomicUsize;
3568 init_test(cx);
3569
3570 let fs = FakeFs::new(cx.executor());
3571 let project = Project::test(fs, None, cx).await;
3572
3573 // Create a connection that simulates refusal after tool result
3574 let prompt_count = Arc::new(AtomicUsize::new(0));
3575 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3576 let prompt_count = prompt_count.clone();
3577 move |_request, thread, mut cx| {
3578 let count = prompt_count.fetch_add(1, SeqCst);
3579 async move {
3580 if count == 0 {
3581 // First prompt: Generate a tool call with result
3582 thread.update(&mut cx, |thread, cx| {
3583 thread
3584 .handle_session_update(
3585 acp::SessionUpdate::ToolCall(
3586 acp::ToolCall::new("tool1", "Test Tool")
3587 .kind(acp::ToolKind::Fetch)
3588 .status(acp::ToolCallStatus::Completed)
3589 .raw_input(serde_json::json!({"query": "test"}))
3590 .raw_output(serde_json::json!({"result": "inappropriate content"})),
3591 ),
3592 cx,
3593 )
3594 .unwrap();
3595 })?;
3596
3597 // Now return refusal because of the tool result
3598 Ok(acp::PromptResponse::new(acp::StopReason::Refusal))
3599 } else {
3600 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3601 }
3602 }
3603 .boxed_local()
3604 }
3605 }));
3606
3607 let thread = cx
3608 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3609 .await
3610 .unwrap();
3611
3612 // Track if we see a Refusal event
3613 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
3614 let saw_refusal_event_captured = saw_refusal_event.clone();
3615 thread.update(cx, |_thread, cx| {
3616 cx.subscribe(
3617 &thread,
3618 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
3619 if matches!(event, AcpThreadEvent::Refusal) {
3620 *saw_refusal_event_captured.lock().unwrap() = true;
3621 }
3622 },
3623 )
3624 .detach();
3625 });
3626
3627 // Send a user message - this will trigger tool call and then refusal
3628 let send_task = thread.update(cx, |thread, cx| thread.send(vec!["Hello".into()], cx));
3629 cx.background_executor.spawn(send_task).detach();
3630 cx.run_until_parked();
3631
3632 // Verify that:
3633 // 1. A Refusal event WAS emitted (because it's a tool result refusal, not user prompt)
3634 // 2. The user message was NOT truncated
3635 assert!(
3636 *saw_refusal_event.lock().unwrap(),
3637 "Refusal event should be emitted for tool result refusals"
3638 );
3639
3640 thread.read_with(cx, |thread, _| {
3641 let entries = thread.entries();
3642 assert!(entries.len() >= 2, "Should have user message and tool call");
3643
3644 // Verify user message is still there
3645 assert!(
3646 matches!(entries[0], AgentThreadEntry::UserMessage(_)),
3647 "User message should not be truncated"
3648 );
3649
3650 // Verify tool call is there with result
3651 if let AgentThreadEntry::ToolCall(tool_call) = &entries[1] {
3652 assert!(
3653 tool_call.raw_output.is_some(),
3654 "Tool call should have output"
3655 );
3656 } else {
3657 panic!("Expected tool call at index 1");
3658 }
3659 });
3660 }
3661
3662 #[gpui::test]
3663 async fn test_user_prompt_refusal_emits_event(cx: &mut TestAppContext) {
3664 init_test(cx);
3665
3666 let fs = FakeFs::new(cx.executor());
3667 let project = Project::test(fs, None, cx).await;
3668
3669 let refuse_next = Arc::new(AtomicBool::new(false));
3670 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3671 let refuse_next = refuse_next.clone();
3672 move |_request, _thread, _cx| {
3673 if refuse_next.load(SeqCst) {
3674 async move { Ok(acp::PromptResponse::new(acp::StopReason::Refusal)) }
3675 .boxed_local()
3676 } else {
3677 async move { Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)) }
3678 .boxed_local()
3679 }
3680 }
3681 }));
3682
3683 let thread = cx
3684 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3685 .await
3686 .unwrap();
3687
3688 // Track if we see a Refusal event
3689 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
3690 let saw_refusal_event_captured = saw_refusal_event.clone();
3691 thread.update(cx, |_thread, cx| {
3692 cx.subscribe(
3693 &thread,
3694 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
3695 if matches!(event, AcpThreadEvent::Refusal) {
3696 *saw_refusal_event_captured.lock().unwrap() = true;
3697 }
3698 },
3699 )
3700 .detach();
3701 });
3702
3703 // Send a message that will be refused
3704 refuse_next.store(true, SeqCst);
3705 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
3706 .await
3707 .unwrap();
3708
3709 // Verify that a Refusal event WAS emitted for user prompt refusal
3710 assert!(
3711 *saw_refusal_event.lock().unwrap(),
3712 "Refusal event should be emitted for user prompt refusals"
3713 );
3714
3715 // Verify the message was truncated (user prompt refusal)
3716 thread.read_with(cx, |thread, cx| {
3717 assert_eq!(thread.to_markdown(cx), "");
3718 });
3719 }
3720
3721 #[gpui::test]
3722 async fn test_refusal(cx: &mut TestAppContext) {
3723 init_test(cx);
3724 let fs = FakeFs::new(cx.background_executor.clone());
3725 fs.insert_tree(path!("/"), json!({})).await;
3726 let project = Project::test(fs.clone(), [path!("/").as_ref()], cx).await;
3727
3728 let refuse_next = Arc::new(AtomicBool::new(false));
3729 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3730 let refuse_next = refuse_next.clone();
3731 move |request, thread, mut cx| {
3732 let refuse_next = refuse_next.clone();
3733 async move {
3734 if refuse_next.load(SeqCst) {
3735 return Ok(acp::PromptResponse::new(acp::StopReason::Refusal));
3736 }
3737
3738 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
3739 panic!("expected text content block");
3740 };
3741 thread.update(&mut cx, |thread, cx| {
3742 thread
3743 .handle_session_update(
3744 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(
3745 content.text.to_uppercase().into(),
3746 )),
3747 cx,
3748 )
3749 .unwrap();
3750 })?;
3751 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3752 }
3753 .boxed_local()
3754 }
3755 }));
3756 let thread = cx
3757 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3758 .await
3759 .unwrap();
3760
3761 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
3762 .await
3763 .unwrap();
3764 thread.read_with(cx, |thread, cx| {
3765 assert_eq!(
3766 thread.to_markdown(cx),
3767 indoc! {"
3768 ## User
3769
3770 hello
3771
3772 ## Assistant
3773
3774 HELLO
3775
3776 "}
3777 );
3778 });
3779
3780 // Simulate refusing the second message. The message should be truncated
3781 // when a user prompt is refused.
3782 refuse_next.store(true, SeqCst);
3783 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["world".into()], cx)))
3784 .await
3785 .unwrap();
3786 thread.read_with(cx, |thread, cx| {
3787 assert_eq!(
3788 thread.to_markdown(cx),
3789 indoc! {"
3790 ## User
3791
3792 hello
3793
3794 ## Assistant
3795
3796 HELLO
3797
3798 "}
3799 );
3800 });
3801 }
3802
3803 async fn run_until_first_tool_call(
3804 thread: &Entity<AcpThread>,
3805 cx: &mut TestAppContext,
3806 ) -> usize {
3807 let (mut tx, mut rx) = mpsc::channel::<usize>(1);
3808
3809 let subscription = cx.update(|cx| {
3810 cx.subscribe(thread, move |thread, _, cx| {
3811 for (ix, entry) in thread.read(cx).entries.iter().enumerate() {
3812 if matches!(entry, AgentThreadEntry::ToolCall(_)) {
3813 return tx.try_send(ix).unwrap();
3814 }
3815 }
3816 })
3817 });
3818
3819 select! {
3820 _ = futures::FutureExt::fuse(cx.background_executor.timer(Duration::from_secs(10))) => {
3821 panic!("Timeout waiting for tool call")
3822 }
3823 ix = rx.next().fuse() => {
3824 drop(subscription);
3825 ix.unwrap()
3826 }
3827 }
3828 }
3829
3830 #[derive(Clone, Default)]
3831 struct FakeAgentConnection {
3832 auth_methods: Vec<acp::AuthMethod>,
3833 sessions: Arc<parking_lot::Mutex<HashMap<acp::SessionId, WeakEntity<AcpThread>>>>,
3834 on_user_message: Option<
3835 Rc<
3836 dyn Fn(
3837 acp::PromptRequest,
3838 WeakEntity<AcpThread>,
3839 AsyncApp,
3840 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
3841 + 'static,
3842 >,
3843 >,
3844 }
3845
3846 impl FakeAgentConnection {
3847 fn new() -> Self {
3848 Self {
3849 auth_methods: Vec::new(),
3850 on_user_message: None,
3851 sessions: Arc::default(),
3852 }
3853 }
3854
3855 #[expect(unused)]
3856 fn with_auth_methods(mut self, auth_methods: Vec<acp::AuthMethod>) -> Self {
3857 self.auth_methods = auth_methods;
3858 self
3859 }
3860
3861 fn on_user_message(
3862 mut self,
3863 handler: impl Fn(
3864 acp::PromptRequest,
3865 WeakEntity<AcpThread>,
3866 AsyncApp,
3867 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
3868 + 'static,
3869 ) -> Self {
3870 self.on_user_message.replace(Rc::new(handler));
3871 self
3872 }
3873 }
3874
3875 impl AgentConnection for FakeAgentConnection {
3876 fn telemetry_id(&self) -> SharedString {
3877 "fake".into()
3878 }
3879
3880 fn auth_methods(&self) -> &[acp::AuthMethod] {
3881 &self.auth_methods
3882 }
3883
3884 fn new_thread(
3885 self: Rc<Self>,
3886 project: Entity<Project>,
3887 _cwd: &Path,
3888 cx: &mut App,
3889 ) -> Task<gpui::Result<Entity<AcpThread>>> {
3890 let session_id = acp::SessionId::new(
3891 rand::rng()
3892 .sample_iter(&distr::Alphanumeric)
3893 .take(7)
3894 .map(char::from)
3895 .collect::<String>(),
3896 );
3897 let action_log = cx.new(|_| ActionLog::new(project.clone()));
3898 let thread = cx.new(|cx| {
3899 AcpThread::new(
3900 "Test",
3901 self.clone(),
3902 project,
3903 action_log,
3904 session_id.clone(),
3905 watch::Receiver::constant(
3906 acp::PromptCapabilities::new()
3907 .image(true)
3908 .audio(true)
3909 .embedded_context(true),
3910 ),
3911 cx,
3912 )
3913 });
3914 self.sessions.lock().insert(session_id, thread.downgrade());
3915 Task::ready(Ok(thread))
3916 }
3917
3918 fn authenticate(&self, method: acp::AuthMethodId, _cx: &mut App) -> Task<gpui::Result<()>> {
3919 if self.auth_methods().iter().any(|m| m.id == method) {
3920 Task::ready(Ok(()))
3921 } else {
3922 Task::ready(Err(anyhow!("Invalid Auth Method")))
3923 }
3924 }
3925
3926 fn prompt(
3927 &self,
3928 _id: Option<UserMessageId>,
3929 params: acp::PromptRequest,
3930 cx: &mut App,
3931 ) -> Task<gpui::Result<acp::PromptResponse>> {
3932 let sessions = self.sessions.lock();
3933 let thread = sessions.get(¶ms.session_id).unwrap();
3934 if let Some(handler) = &self.on_user_message {
3935 let handler = handler.clone();
3936 let thread = thread.clone();
3937 cx.spawn(async move |cx| handler(params, thread, cx.clone()).await)
3938 } else {
3939 Task::ready(Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)))
3940 }
3941 }
3942
3943 fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
3944 let sessions = self.sessions.lock();
3945 let thread = sessions.get(session_id).unwrap().clone();
3946
3947 cx.spawn(async move |cx| {
3948 thread
3949 .update(cx, |thread, cx| thread.cancel(cx))
3950 .unwrap()
3951 .await
3952 })
3953 .detach();
3954 }
3955
3956 fn truncate(
3957 &self,
3958 session_id: &acp::SessionId,
3959 _cx: &App,
3960 ) -> Option<Rc<dyn AgentSessionTruncate>> {
3961 Some(Rc::new(FakeAgentSessionEditor {
3962 _session_id: session_id.clone(),
3963 }))
3964 }
3965
3966 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
3967 self
3968 }
3969 }
3970
3971 struct FakeAgentSessionEditor {
3972 _session_id: acp::SessionId,
3973 }
3974
3975 impl AgentSessionTruncate for FakeAgentSessionEditor {
3976 fn run(&self, _message_id: UserMessageId, _cx: &mut App) -> Task<Result<()>> {
3977 Task::ready(Ok(()))
3978 }
3979 }
3980
3981 #[gpui::test]
3982 async fn test_tool_call_not_found_creates_failed_entry(cx: &mut TestAppContext) {
3983 init_test(cx);
3984
3985 let fs = FakeFs::new(cx.executor());
3986 let project = Project::test(fs, [], cx).await;
3987 let connection = Rc::new(FakeAgentConnection::new());
3988 let thread = cx
3989 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
3990 .await
3991 .unwrap();
3992
3993 // Try to update a tool call that doesn't exist
3994 let nonexistent_id = acp::ToolCallId::new("nonexistent-tool-call");
3995 thread.update(cx, |thread, cx| {
3996 let result = thread.handle_session_update(
3997 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
3998 nonexistent_id.clone(),
3999 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
4000 )),
4001 cx,
4002 );
4003
4004 // The update should succeed (not return an error)
4005 assert!(result.is_ok());
4006
4007 // There should now be exactly one entry in the thread
4008 assert_eq!(thread.entries.len(), 1);
4009
4010 // The entry should be a failed tool call
4011 if let AgentThreadEntry::ToolCall(tool_call) = &thread.entries[0] {
4012 assert_eq!(tool_call.id, nonexistent_id);
4013 assert!(matches!(tool_call.status, ToolCallStatus::Failed));
4014 assert_eq!(tool_call.kind, acp::ToolKind::Fetch);
4015
4016 // Check that the content contains the error message
4017 assert_eq!(tool_call.content.len(), 1);
4018 if let ToolCallContent::ContentBlock(content_block) = &tool_call.content[0] {
4019 match content_block {
4020 ContentBlock::Markdown { markdown } => {
4021 let markdown_text = markdown.read(cx).source();
4022 assert!(markdown_text.contains("Tool call not found"));
4023 }
4024 ContentBlock::Empty => panic!("Expected markdown content, got empty"),
4025 ContentBlock::ResourceLink { .. } => {
4026 panic!("Expected markdown content, got resource link")
4027 }
4028 ContentBlock::Image { .. } => {
4029 panic!("Expected markdown content, got image")
4030 }
4031 }
4032 } else {
4033 panic!("Expected ContentBlock, got: {:?}", tool_call.content[0]);
4034 }
4035 } else {
4036 panic!("Expected ToolCall entry, got: {:?}", thread.entries[0]);
4037 }
4038 });
4039 }
4040
4041 /// Tests that restoring a checkpoint properly cleans up terminals that were
4042 /// created after that checkpoint, and cancels any in-progress generation.
4043 ///
4044 /// Reproduces issue #35142: When a checkpoint is restored, any terminal processes
4045 /// that were started after that checkpoint should be terminated, and any in-progress
4046 /// AI generation should be canceled.
4047 #[gpui::test]
4048 async fn test_restore_checkpoint_kills_terminal(cx: &mut TestAppContext) {
4049 init_test(cx);
4050
4051 let fs = FakeFs::new(cx.executor());
4052 let project = Project::test(fs, [], cx).await;
4053 let connection = Rc::new(FakeAgentConnection::new());
4054 let thread = cx
4055 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
4056 .await
4057 .unwrap();
4058
4059 // Send first user message to create a checkpoint
4060 cx.update(|cx| {
4061 thread.update(cx, |thread, cx| {
4062 thread.send(vec!["first message".into()], cx)
4063 })
4064 })
4065 .await
4066 .unwrap();
4067
4068 // Send second message (creates another checkpoint) - we'll restore to this one
4069 cx.update(|cx| {
4070 thread.update(cx, |thread, cx| {
4071 thread.send(vec!["second message".into()], cx)
4072 })
4073 })
4074 .await
4075 .unwrap();
4076
4077 // Create 2 terminals BEFORE the checkpoint that have completed running
4078 let terminal_id_1 = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
4079 let mock_terminal_1 = cx.new(|cx| {
4080 let builder = ::terminal::TerminalBuilder::new_display_only(
4081 ::terminal::terminal_settings::CursorShape::default(),
4082 ::terminal::terminal_settings::AlternateScroll::On,
4083 None,
4084 0,
4085 cx.background_executor(),
4086 PathStyle::local(),
4087 )
4088 .unwrap();
4089 builder.subscribe(cx)
4090 });
4091
4092 thread.update(cx, |thread, cx| {
4093 thread.on_terminal_provider_event(
4094 TerminalProviderEvent::Created {
4095 terminal_id: terminal_id_1.clone(),
4096 label: "echo 'first'".to_string(),
4097 cwd: Some(PathBuf::from("/test")),
4098 output_byte_limit: None,
4099 terminal: mock_terminal_1.clone(),
4100 },
4101 cx,
4102 );
4103 });
4104
4105 thread.update(cx, |thread, cx| {
4106 thread.on_terminal_provider_event(
4107 TerminalProviderEvent::Output {
4108 terminal_id: terminal_id_1.clone(),
4109 data: b"first\n".to_vec(),
4110 },
4111 cx,
4112 );
4113 });
4114
4115 thread.update(cx, |thread, cx| {
4116 thread.on_terminal_provider_event(
4117 TerminalProviderEvent::Exit {
4118 terminal_id: terminal_id_1.clone(),
4119 status: acp::TerminalExitStatus::new().exit_code(0),
4120 },
4121 cx,
4122 );
4123 });
4124
4125 let terminal_id_2 = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
4126 let mock_terminal_2 = cx.new(|cx| {
4127 let builder = ::terminal::TerminalBuilder::new_display_only(
4128 ::terminal::terminal_settings::CursorShape::default(),
4129 ::terminal::terminal_settings::AlternateScroll::On,
4130 None,
4131 0,
4132 cx.background_executor(),
4133 PathStyle::local(),
4134 )
4135 .unwrap();
4136 builder.subscribe(cx)
4137 });
4138
4139 thread.update(cx, |thread, cx| {
4140 thread.on_terminal_provider_event(
4141 TerminalProviderEvent::Created {
4142 terminal_id: terminal_id_2.clone(),
4143 label: "echo 'second'".to_string(),
4144 cwd: Some(PathBuf::from("/test")),
4145 output_byte_limit: None,
4146 terminal: mock_terminal_2.clone(),
4147 },
4148 cx,
4149 );
4150 });
4151
4152 thread.update(cx, |thread, cx| {
4153 thread.on_terminal_provider_event(
4154 TerminalProviderEvent::Output {
4155 terminal_id: terminal_id_2.clone(),
4156 data: b"second\n".to_vec(),
4157 },
4158 cx,
4159 );
4160 });
4161
4162 thread.update(cx, |thread, cx| {
4163 thread.on_terminal_provider_event(
4164 TerminalProviderEvent::Exit {
4165 terminal_id: terminal_id_2.clone(),
4166 status: acp::TerminalExitStatus::new().exit_code(0),
4167 },
4168 cx,
4169 );
4170 });
4171
4172 // Get the second message ID to restore to
4173 let second_message_id = thread.read_with(cx, |thread, _| {
4174 // At this point we have:
4175 // - Index 0: First user message (with checkpoint)
4176 // - Index 1: Second user message (with checkpoint)
4177 // No assistant responses because FakeAgentConnection just returns EndTurn
4178 let AgentThreadEntry::UserMessage(message) = &thread.entries[1] else {
4179 panic!("expected user message at index 1");
4180 };
4181 message.id.clone().unwrap()
4182 });
4183
4184 // Create a terminal AFTER the checkpoint we'll restore to.
4185 // This simulates the AI agent starting a long-running terminal command.
4186 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
4187 let mock_terminal = cx.new(|cx| {
4188 let builder = ::terminal::TerminalBuilder::new_display_only(
4189 ::terminal::terminal_settings::CursorShape::default(),
4190 ::terminal::terminal_settings::AlternateScroll::On,
4191 None,
4192 0,
4193 cx.background_executor(),
4194 PathStyle::local(),
4195 )
4196 .unwrap();
4197 builder.subscribe(cx)
4198 });
4199
4200 // Register the terminal as created
4201 thread.update(cx, |thread, cx| {
4202 thread.on_terminal_provider_event(
4203 TerminalProviderEvent::Created {
4204 terminal_id: terminal_id.clone(),
4205 label: "sleep 1000".to_string(),
4206 cwd: Some(PathBuf::from("/test")),
4207 output_byte_limit: None,
4208 terminal: mock_terminal.clone(),
4209 },
4210 cx,
4211 );
4212 });
4213
4214 // Simulate the terminal producing output (still running)
4215 thread.update(cx, |thread, cx| {
4216 thread.on_terminal_provider_event(
4217 TerminalProviderEvent::Output {
4218 terminal_id: terminal_id.clone(),
4219 data: b"terminal is running...\n".to_vec(),
4220 },
4221 cx,
4222 );
4223 });
4224
4225 // Create a tool call entry that references this terminal
4226 // This represents the agent requesting a terminal command
4227 thread.update(cx, |thread, cx| {
4228 thread
4229 .handle_session_update(
4230 acp::SessionUpdate::ToolCall(
4231 acp::ToolCall::new("terminal-tool-1", "Running command")
4232 .kind(acp::ToolKind::Execute)
4233 .status(acp::ToolCallStatus::InProgress)
4234 .content(vec![acp::ToolCallContent::Terminal(acp::Terminal::new(
4235 terminal_id.clone(),
4236 ))])
4237 .raw_input(serde_json::json!({"command": "sleep 1000", "cd": "/test"})),
4238 ),
4239 cx,
4240 )
4241 .unwrap();
4242 });
4243
4244 // Verify terminal exists and is in the thread
4245 let terminal_exists_before =
4246 thread.read_with(cx, |thread, _| thread.terminals.contains_key(&terminal_id));
4247 assert!(
4248 terminal_exists_before,
4249 "Terminal should exist before checkpoint restore"
4250 );
4251
4252 // Verify the terminal's underlying task is still running (not completed)
4253 let terminal_running_before = thread.read_with(cx, |thread, _cx| {
4254 let terminal_entity = thread.terminals.get(&terminal_id).unwrap();
4255 terminal_entity.read_with(cx, |term, _cx| {
4256 term.output().is_none() // output is None means it's still running
4257 })
4258 });
4259 assert!(
4260 terminal_running_before,
4261 "Terminal should be running before checkpoint restore"
4262 );
4263
4264 // Verify we have the expected entries before restore
4265 let entry_count_before = thread.read_with(cx, |thread, _| thread.entries.len());
4266 assert!(
4267 entry_count_before > 1,
4268 "Should have multiple entries before restore"
4269 );
4270
4271 // Restore the checkpoint to the second message.
4272 // This should:
4273 // 1. Cancel any in-progress generation (via the cancel() call)
4274 // 2. Remove the terminal that was created after that point
4275 thread
4276 .update(cx, |thread, cx| {
4277 thread.restore_checkpoint(second_message_id, cx)
4278 })
4279 .await
4280 .unwrap();
4281
4282 // Verify that no send_task is in progress after restore
4283 // (cancel() clears the send_task)
4284 let has_send_task_after = thread.read_with(cx, |thread, _| thread.send_task.is_some());
4285 assert!(
4286 !has_send_task_after,
4287 "Should not have a send_task after restore (cancel should have cleared it)"
4288 );
4289
4290 // Verify the entries were truncated (restoring to index 1 truncates at 1, keeping only index 0)
4291 let entry_count = thread.read_with(cx, |thread, _| thread.entries.len());
4292 assert_eq!(
4293 entry_count, 1,
4294 "Should have 1 entry after restore (only the first user message)"
4295 );
4296
4297 // Verify the 2 completed terminals from before the checkpoint still exist
4298 let terminal_1_exists = thread.read_with(cx, |thread, _| {
4299 thread.terminals.contains_key(&terminal_id_1)
4300 });
4301 assert!(
4302 terminal_1_exists,
4303 "Terminal 1 (from before checkpoint) should still exist"
4304 );
4305
4306 let terminal_2_exists = thread.read_with(cx, |thread, _| {
4307 thread.terminals.contains_key(&terminal_id_2)
4308 });
4309 assert!(
4310 terminal_2_exists,
4311 "Terminal 2 (from before checkpoint) should still exist"
4312 );
4313
4314 // Verify they're still in completed state
4315 let terminal_1_completed = thread.read_with(cx, |thread, _cx| {
4316 let terminal_entity = thread.terminals.get(&terminal_id_1).unwrap();
4317 terminal_entity.read_with(cx, |term, _cx| term.output().is_some())
4318 });
4319 assert!(terminal_1_completed, "Terminal 1 should still be completed");
4320
4321 let terminal_2_completed = thread.read_with(cx, |thread, _cx| {
4322 let terminal_entity = thread.terminals.get(&terminal_id_2).unwrap();
4323 terminal_entity.read_with(cx, |term, _cx| term.output().is_some())
4324 });
4325 assert!(terminal_2_completed, "Terminal 2 should still be completed");
4326
4327 // Verify the running terminal (created after checkpoint) was removed
4328 let terminal_3_exists =
4329 thread.read_with(cx, |thread, _| thread.terminals.contains_key(&terminal_id));
4330 assert!(
4331 !terminal_3_exists,
4332 "Terminal 3 (created after checkpoint) should have been removed"
4333 );
4334
4335 // Verify total count is 2 (the two from before the checkpoint)
4336 let terminal_count = thread.read_with(cx, |thread, _| thread.terminals.len());
4337 assert_eq!(
4338 terminal_count, 2,
4339 "Should have exactly 2 terminals (the completed ones from before checkpoint)"
4340 );
4341 }
4342
4343 /// Tests that update_last_checkpoint correctly updates the original message's checkpoint
4344 /// even when a new user message is added while the async checkpoint comparison is in progress.
4345 ///
4346 /// This is a regression test for a bug where update_last_checkpoint would fail with
4347 /// "no checkpoint" if a new user message (without a checkpoint) was added between when
4348 /// update_last_checkpoint started and when its async closure ran.
4349 #[gpui::test]
4350 async fn test_update_last_checkpoint_with_new_message_added(cx: &mut TestAppContext) {
4351 init_test(cx);
4352
4353 let fs = FakeFs::new(cx.executor());
4354 fs.insert_tree(path!("/test"), json!({".git": {}, "file.txt": "content"}))
4355 .await;
4356 let project = Project::test(fs.clone(), [Path::new(path!("/test"))], cx).await;
4357
4358 let handler_done = Arc::new(AtomicBool::new(false));
4359 let handler_done_clone = handler_done.clone();
4360 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
4361 move |_, _thread, _cx| {
4362 handler_done_clone.store(true, SeqCst);
4363 async move { Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)) }.boxed_local()
4364 },
4365 ));
4366
4367 let thread = cx
4368 .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
4369 .await
4370 .unwrap();
4371
4372 let send_future = thread.update(cx, |thread, cx| thread.send_raw("First message", cx));
4373 let send_task = cx.background_executor.spawn(send_future);
4374
4375 // Tick until handler completes, then a few more to let update_last_checkpoint start
4376 while !handler_done.load(SeqCst) {
4377 cx.executor().tick();
4378 }
4379 for _ in 0..5 {
4380 cx.executor().tick();
4381 }
4382
4383 thread.update(cx, |thread, cx| {
4384 thread.push_entry(
4385 AgentThreadEntry::UserMessage(UserMessage {
4386 id: Some(UserMessageId::new()),
4387 content: ContentBlock::Empty,
4388 chunks: vec!["Injected message (no checkpoint)".into()],
4389 checkpoint: None,
4390 indented: false,
4391 }),
4392 cx,
4393 );
4394 });
4395
4396 cx.run_until_parked();
4397 let result = send_task.await;
4398
4399 assert!(
4400 result.is_ok(),
4401 "send should succeed even when new message added during update_last_checkpoint: {:?}",
4402 result.err()
4403 );
4404 }
4405}