1mod connection;
2mod diff;
3mod mention;
4mod terminal;
5use action_log::{ActionLog, ActionLogTelemetry};
6use agent_client_protocol::{self as acp};
7use anyhow::{Context as _, Result, anyhow};
8use collections::HashSet;
9pub use connection::*;
10pub use diff::*;
11use futures::{FutureExt, channel::oneshot, future::BoxFuture};
12use gpui::{
13 AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Subscription, Task,
14 WeakEntity,
15};
16use itertools::Itertools;
17use language::language_settings::FormatOnSave;
18use language::{
19 Anchor, Buffer, BufferEvent, BufferSnapshot, LanguageRegistry, Point, ToPoint, text_diff,
20};
21use markdown::Markdown;
22pub use mention::*;
23use project::lsp_store::{FormatTrigger, LspFormatTarget};
24use project::{AgentLocation, Project, git_store::GitStoreCheckpoint};
25use serde::{Deserialize, Serialize};
26use serde_json::to_string_pretty;
27use std::collections::HashMap;
28use std::error::Error;
29use std::fmt::{Formatter, Write};
30use std::ops::Range;
31use std::process::ExitStatus;
32use std::rc::Rc;
33use std::time::{Duration, Instant};
34use std::{fmt::Display, mem, path::PathBuf, sync::Arc};
35use task::{Shell, ShellBuilder};
36pub use terminal::*;
37use text::Bias;
38use ui::App;
39use util::markdown::MarkdownEscaped;
40use util::path_list::PathList;
41use util::{ResultExt, get_default_system_shell_preferring_bash, paths::PathStyle};
42use uuid::Uuid;
43
44/// Key used in ACP ToolCall meta to store the tool's programmatic name.
45/// This is a workaround since ACP's ToolCall doesn't have a dedicated name field.
46pub const TOOL_NAME_META_KEY: &str = "tool_name";
47
48/// Helper to extract tool name from ACP meta
49pub fn tool_name_from_meta(meta: &Option<acp::Meta>) -> Option<SharedString> {
50 meta.as_ref()
51 .and_then(|m| m.get(TOOL_NAME_META_KEY))
52 .and_then(|v| v.as_str())
53 .map(|s| SharedString::from(s.to_owned()))
54}
55
56/// Helper to create meta with tool name
57pub fn meta_with_tool_name(tool_name: &str) -> acp::Meta {
58 acp::Meta::from_iter([(TOOL_NAME_META_KEY.into(), tool_name.into())])
59}
60
61/// Key used in ACP ToolCall meta to store the session id and message indexes
62pub const SUBAGENT_SESSION_INFO_META_KEY: &str = "subagent_session_info";
63
64#[derive(Clone, Debug, Deserialize, Serialize)]
65pub struct SubagentSessionInfo {
66 /// The session id of the subagent sessiont that was spawned
67 pub session_id: acp::SessionId,
68 /// The index of the message of the start of the "turn" run by this tool call
69 pub message_start_index: usize,
70 /// The index of the output of the message that the subagent has returned
71 #[serde(skip_serializing_if = "Option::is_none")]
72 pub message_end_index: Option<usize>,
73}
74
75/// Helper to extract subagent session id from ACP meta
76pub fn subagent_session_info_from_meta(meta: &Option<acp::Meta>) -> Option<SubagentSessionInfo> {
77 meta.as_ref()
78 .and_then(|m| m.get(SUBAGENT_SESSION_INFO_META_KEY))
79 .and_then(|v| serde_json::from_value(v.clone()).ok())
80}
81
82#[derive(Debug)]
83pub struct UserMessage {
84 pub id: Option<UserMessageId>,
85 pub content: ContentBlock,
86 pub chunks: Vec<acp::ContentBlock>,
87 pub checkpoint: Option<Checkpoint>,
88 pub indented: bool,
89}
90
91#[derive(Debug)]
92pub struct Checkpoint {
93 git_checkpoint: GitStoreCheckpoint,
94 pub show: bool,
95}
96
97impl UserMessage {
98 fn to_markdown(&self, cx: &App) -> String {
99 let mut markdown = String::new();
100 if self
101 .checkpoint
102 .as_ref()
103 .is_some_and(|checkpoint| checkpoint.show)
104 {
105 writeln!(markdown, "## User (checkpoint)").unwrap();
106 } else {
107 writeln!(markdown, "## User").unwrap();
108 }
109 writeln!(markdown).unwrap();
110 writeln!(markdown, "{}", self.content.to_markdown(cx)).unwrap();
111 writeln!(markdown).unwrap();
112 markdown
113 }
114}
115
116#[derive(Debug, PartialEq)]
117pub struct AssistantMessage {
118 pub chunks: Vec<AssistantMessageChunk>,
119 pub indented: bool,
120 pub is_subagent_output: bool,
121}
122
123impl AssistantMessage {
124 pub fn to_markdown(&self, cx: &App) -> String {
125 format!(
126 "## Assistant\n\n{}\n\n",
127 self.chunks
128 .iter()
129 .map(|chunk| chunk.to_markdown(cx))
130 .join("\n\n")
131 )
132 }
133}
134
135#[derive(Debug, PartialEq)]
136pub enum AssistantMessageChunk {
137 Message { block: ContentBlock },
138 Thought { block: ContentBlock },
139}
140
141impl AssistantMessageChunk {
142 pub fn from_str(
143 chunk: &str,
144 language_registry: &Arc<LanguageRegistry>,
145 path_style: PathStyle,
146 cx: &mut App,
147 ) -> Self {
148 Self::Message {
149 block: ContentBlock::new(chunk.into(), language_registry, path_style, cx),
150 }
151 }
152
153 fn to_markdown(&self, cx: &App) -> String {
154 match self {
155 Self::Message { block } => block.to_markdown(cx).to_string(),
156 Self::Thought { block } => {
157 format!("<thinking>\n{}\n</thinking>", block.to_markdown(cx))
158 }
159 }
160 }
161}
162
163#[derive(Debug)]
164pub enum AgentThreadEntry {
165 UserMessage(UserMessage),
166 AssistantMessage(AssistantMessage),
167 ToolCall(ToolCall),
168}
169
170impl AgentThreadEntry {
171 pub fn is_indented(&self) -> bool {
172 match self {
173 Self::UserMessage(message) => message.indented,
174 Self::AssistantMessage(message) => message.indented,
175 Self::ToolCall(_) => false,
176 }
177 }
178
179 pub fn to_markdown(&self, cx: &App) -> String {
180 match self {
181 Self::UserMessage(message) => message.to_markdown(cx),
182 Self::AssistantMessage(message) => message.to_markdown(cx),
183 Self::ToolCall(tool_call) => tool_call.to_markdown(cx),
184 }
185 }
186
187 pub fn user_message(&self) -> Option<&UserMessage> {
188 if let AgentThreadEntry::UserMessage(message) = self {
189 Some(message)
190 } else {
191 None
192 }
193 }
194
195 pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
196 if let AgentThreadEntry::ToolCall(call) = self {
197 itertools::Either::Left(call.diffs())
198 } else {
199 itertools::Either::Right(std::iter::empty())
200 }
201 }
202
203 pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
204 if let AgentThreadEntry::ToolCall(call) = self {
205 itertools::Either::Left(call.terminals())
206 } else {
207 itertools::Either::Right(std::iter::empty())
208 }
209 }
210
211 pub fn location(&self, ix: usize) -> Option<(acp::ToolCallLocation, AgentLocation)> {
212 if let AgentThreadEntry::ToolCall(ToolCall {
213 locations,
214 resolved_locations,
215 ..
216 }) = self
217 {
218 Some((
219 locations.get(ix)?.clone(),
220 resolved_locations.get(ix)?.clone()?,
221 ))
222 } else {
223 None
224 }
225 }
226}
227
228#[derive(Debug)]
229pub struct ToolCall {
230 pub id: acp::ToolCallId,
231 pub label: Entity<Markdown>,
232 pub kind: acp::ToolKind,
233 pub content: Vec<ToolCallContent>,
234 pub status: ToolCallStatus,
235 pub locations: Vec<acp::ToolCallLocation>,
236 pub resolved_locations: Vec<Option<AgentLocation>>,
237 pub raw_input: Option<serde_json::Value>,
238 pub raw_input_markdown: Option<Entity<Markdown>>,
239 pub raw_output: Option<serde_json::Value>,
240 pub tool_name: Option<SharedString>,
241 pub subagent_session_info: Option<SubagentSessionInfo>,
242}
243
244impl ToolCall {
245 fn from_acp(
246 tool_call: acp::ToolCall,
247 status: ToolCallStatus,
248 language_registry: Arc<LanguageRegistry>,
249 path_style: PathStyle,
250 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
251 cx: &mut App,
252 ) -> Result<Self> {
253 let title = if tool_call.kind == acp::ToolKind::Execute {
254 tool_call.title
255 } else if tool_call.kind == acp::ToolKind::Edit {
256 MarkdownEscaped(tool_call.title.as_str()).to_string()
257 } else if let Some((first_line, _)) = tool_call.title.split_once("\n") {
258 first_line.to_owned() + "…"
259 } else {
260 tool_call.title
261 };
262 let mut content = Vec::with_capacity(tool_call.content.len());
263 for item in tool_call.content {
264 if let Some(item) = ToolCallContent::from_acp(
265 item,
266 language_registry.clone(),
267 path_style,
268 terminals,
269 cx,
270 )? {
271 content.push(item);
272 }
273 }
274
275 let raw_input_markdown = tool_call
276 .raw_input
277 .as_ref()
278 .and_then(|input| markdown_for_raw_output(input, &language_registry, cx));
279
280 let tool_name = tool_name_from_meta(&tool_call.meta);
281
282 let subagent_session_info = subagent_session_info_from_meta(&tool_call.meta);
283
284 let result = Self {
285 id: tool_call.tool_call_id,
286 label: cx
287 .new(|cx| Markdown::new(title.into(), Some(language_registry.clone()), None, cx)),
288 kind: tool_call.kind,
289 content,
290 locations: tool_call.locations,
291 resolved_locations: Vec::default(),
292 status,
293 raw_input: tool_call.raw_input,
294 raw_input_markdown,
295 raw_output: tool_call.raw_output,
296 tool_name,
297 subagent_session_info,
298 };
299 Ok(result)
300 }
301
302 fn update_fields(
303 &mut self,
304 fields: acp::ToolCallUpdateFields,
305 meta: Option<acp::Meta>,
306 language_registry: Arc<LanguageRegistry>,
307 path_style: PathStyle,
308 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
309 cx: &mut App,
310 ) -> Result<()> {
311 let acp::ToolCallUpdateFields {
312 kind,
313 status,
314 title,
315 content,
316 locations,
317 raw_input,
318 raw_output,
319 ..
320 } = fields;
321
322 if let Some(kind) = kind {
323 self.kind = kind;
324 }
325
326 if let Some(status) = status {
327 self.status = status.into();
328 }
329
330 if let Some(tool_name) = tool_name_from_meta(&meta) {
331 self.tool_name = Some(tool_name);
332 }
333
334 if let Some(subagent_session_info) = subagent_session_info_from_meta(&meta) {
335 self.subagent_session_info = Some(subagent_session_info);
336 }
337
338 if let Some(title) = title {
339 if self.kind == acp::ToolKind::Execute {
340 for terminal in self.terminals() {
341 terminal.update(cx, |terminal, cx| {
342 terminal.update_command_label(&title, cx);
343 });
344 }
345 }
346 self.label.update(cx, |label, cx| {
347 if self.kind == acp::ToolKind::Execute {
348 label.replace(title, cx);
349 } else if self.kind == acp::ToolKind::Edit {
350 label.replace(MarkdownEscaped(&title).to_string(), cx)
351 } else if let Some((first_line, _)) = title.split_once("\n") {
352 label.replace(first_line.to_owned() + "…", cx);
353 } else {
354 label.replace(title, cx);
355 }
356 });
357 }
358
359 if let Some(content) = content {
360 let mut new_content_len = content.len();
361 let mut content = content.into_iter();
362
363 // Reuse existing content if we can
364 for (old, new) in self.content.iter_mut().zip(content.by_ref()) {
365 let valid_content =
366 old.update_from_acp(new, language_registry.clone(), path_style, terminals, cx)?;
367 if !valid_content {
368 new_content_len -= 1;
369 }
370 }
371 for new in content {
372 if let Some(new) = ToolCallContent::from_acp(
373 new,
374 language_registry.clone(),
375 path_style,
376 terminals,
377 cx,
378 )? {
379 self.content.push(new);
380 } else {
381 new_content_len -= 1;
382 }
383 }
384 self.content.truncate(new_content_len);
385 }
386
387 if let Some(locations) = locations {
388 self.locations = locations;
389 }
390
391 if let Some(raw_input) = raw_input {
392 self.raw_input_markdown = markdown_for_raw_output(&raw_input, &language_registry, cx);
393 self.raw_input = Some(raw_input);
394 }
395
396 if let Some(raw_output) = raw_output {
397 if self.content.is_empty()
398 && let Some(markdown) = markdown_for_raw_output(&raw_output, &language_registry, cx)
399 {
400 self.content
401 .push(ToolCallContent::ContentBlock(ContentBlock::Markdown {
402 markdown,
403 }));
404 }
405 self.raw_output = Some(raw_output);
406 }
407 Ok(())
408 }
409
410 pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
411 self.content.iter().filter_map(|content| match content {
412 ToolCallContent::Diff(diff) => Some(diff),
413 ToolCallContent::ContentBlock(_) => None,
414 ToolCallContent::Terminal(_) => None,
415 })
416 }
417
418 pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
419 self.content.iter().filter_map(|content| match content {
420 ToolCallContent::Terminal(terminal) => Some(terminal),
421 ToolCallContent::ContentBlock(_) => None,
422 ToolCallContent::Diff(_) => None,
423 })
424 }
425
426 pub fn is_subagent(&self) -> bool {
427 self.tool_name.as_ref().is_some_and(|s| s == "spawn_agent")
428 || self.subagent_session_info.is_some()
429 }
430
431 pub fn to_markdown(&self, cx: &App) -> String {
432 let mut markdown = format!(
433 "**Tool Call: {}**\nStatus: {}\n\n",
434 self.label.read(cx).source(),
435 self.status
436 );
437 for content in &self.content {
438 markdown.push_str(content.to_markdown(cx).as_str());
439 markdown.push_str("\n\n");
440 }
441 markdown
442 }
443
444 async fn resolve_location(
445 location: acp::ToolCallLocation,
446 project: WeakEntity<Project>,
447 cx: &mut AsyncApp,
448 ) -> Option<ResolvedLocation> {
449 let buffer = project
450 .update(cx, |project, cx| {
451 project
452 .project_path_for_absolute_path(&location.path, cx)
453 .map(|path| project.open_buffer(path, cx))
454 })
455 .ok()??;
456 let buffer = buffer.await.log_err()?;
457 let position = buffer.update(cx, |buffer, _| {
458 let snapshot = buffer.snapshot();
459 if let Some(row) = location.line {
460 let column = snapshot.indent_size_for_line(row).len;
461 let point = snapshot.clip_point(Point::new(row, column), Bias::Left);
462 snapshot.anchor_before(point)
463 } else {
464 Anchor::min_for_buffer(snapshot.remote_id())
465 }
466 });
467
468 Some(ResolvedLocation { buffer, position })
469 }
470
471 fn resolve_locations(
472 &self,
473 project: Entity<Project>,
474 cx: &mut App,
475 ) -> Task<Vec<Option<ResolvedLocation>>> {
476 let locations = self.locations.clone();
477 project.update(cx, |_, cx| {
478 cx.spawn(async move |project, cx| {
479 let mut new_locations = Vec::new();
480 for location in locations {
481 new_locations.push(Self::resolve_location(location, project.clone(), cx).await);
482 }
483 new_locations
484 })
485 })
486 }
487}
488
489// Separate so we can hold a strong reference to the buffer
490// for saving on the thread
491#[derive(Clone, Debug, PartialEq, Eq)]
492struct ResolvedLocation {
493 buffer: Entity<Buffer>,
494 position: Anchor,
495}
496
497impl From<&ResolvedLocation> for AgentLocation {
498 fn from(value: &ResolvedLocation) -> Self {
499 Self {
500 buffer: value.buffer.downgrade(),
501 position: value.position,
502 }
503 }
504}
505
506#[derive(Debug, Clone)]
507pub enum SelectedPermissionParams {
508 Terminal { patterns: Vec<String> },
509}
510
511#[derive(Debug)]
512pub struct SelectedPermissionOutcome {
513 pub option_id: acp::PermissionOptionId,
514 pub params: Option<SelectedPermissionParams>,
515}
516
517impl SelectedPermissionOutcome {
518 pub fn new(option_id: acp::PermissionOptionId) -> Self {
519 Self {
520 option_id,
521 params: None,
522 }
523 }
524
525 pub fn params(mut self, params: Option<SelectedPermissionParams>) -> Self {
526 self.params = params;
527 self
528 }
529}
530
531impl From<acp::PermissionOptionId> for SelectedPermissionOutcome {
532 fn from(option_id: acp::PermissionOptionId) -> Self {
533 Self::new(option_id)
534 }
535}
536
537impl From<SelectedPermissionOutcome> for acp::SelectedPermissionOutcome {
538 fn from(value: SelectedPermissionOutcome) -> Self {
539 Self::new(value.option_id)
540 }
541}
542
543#[derive(Debug)]
544pub enum RequestPermissionOutcome {
545 Cancelled,
546 Selected(SelectedPermissionOutcome),
547}
548
549impl From<RequestPermissionOutcome> for acp::RequestPermissionOutcome {
550 fn from(value: RequestPermissionOutcome) -> Self {
551 match value {
552 RequestPermissionOutcome::Cancelled => Self::Cancelled,
553 RequestPermissionOutcome::Selected(outcome) => Self::Selected(outcome.into()),
554 }
555 }
556}
557
558#[derive(Debug)]
559pub enum ToolCallStatus {
560 /// The tool call hasn't started running yet, but we start showing it to
561 /// the user.
562 Pending,
563 /// The tool call is waiting for confirmation from the user.
564 WaitingForConfirmation {
565 options: PermissionOptions,
566 respond_tx: oneshot::Sender<SelectedPermissionOutcome>,
567 },
568 /// The tool call is currently running.
569 InProgress,
570 /// The tool call completed successfully.
571 Completed,
572 /// The tool call failed.
573 Failed,
574 /// The user rejected the tool call.
575 Rejected,
576 /// The user canceled generation so the tool call was canceled.
577 Canceled,
578}
579
580impl From<acp::ToolCallStatus> for ToolCallStatus {
581 fn from(status: acp::ToolCallStatus) -> Self {
582 match status {
583 acp::ToolCallStatus::Pending => Self::Pending,
584 acp::ToolCallStatus::InProgress => Self::InProgress,
585 acp::ToolCallStatus::Completed => Self::Completed,
586 acp::ToolCallStatus::Failed => Self::Failed,
587 _ => Self::Pending,
588 }
589 }
590}
591
592impl Display for ToolCallStatus {
593 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
594 write!(
595 f,
596 "{}",
597 match self {
598 ToolCallStatus::Pending => "Pending",
599 ToolCallStatus::WaitingForConfirmation { .. } => "Waiting for confirmation",
600 ToolCallStatus::InProgress => "In Progress",
601 ToolCallStatus::Completed => "Completed",
602 ToolCallStatus::Failed => "Failed",
603 ToolCallStatus::Rejected => "Rejected",
604 ToolCallStatus::Canceled => "Canceled",
605 }
606 )
607 }
608}
609
610#[derive(Debug, PartialEq, Clone)]
611pub enum ContentBlock {
612 Empty,
613 Markdown { markdown: Entity<Markdown> },
614 ResourceLink { resource_link: acp::ResourceLink },
615 Image { image: Arc<gpui::Image> },
616}
617
618impl ContentBlock {
619 pub fn new(
620 block: acp::ContentBlock,
621 language_registry: &Arc<LanguageRegistry>,
622 path_style: PathStyle,
623 cx: &mut App,
624 ) -> Self {
625 let mut this = Self::Empty;
626 this.append(block, language_registry, path_style, cx);
627 this
628 }
629
630 pub fn new_combined(
631 blocks: impl IntoIterator<Item = acp::ContentBlock>,
632 language_registry: Arc<LanguageRegistry>,
633 path_style: PathStyle,
634 cx: &mut App,
635 ) -> Self {
636 let mut this = Self::Empty;
637 for block in blocks {
638 this.append(block, &language_registry, path_style, cx);
639 }
640 this
641 }
642
643 pub fn append(
644 &mut self,
645 block: acp::ContentBlock,
646 language_registry: &Arc<LanguageRegistry>,
647 path_style: PathStyle,
648 cx: &mut App,
649 ) {
650 match (&mut *self, &block) {
651 (ContentBlock::Empty, acp::ContentBlock::ResourceLink(resource_link)) => {
652 *self = ContentBlock::ResourceLink {
653 resource_link: resource_link.clone(),
654 };
655 }
656 (ContentBlock::Empty, acp::ContentBlock::Image(image_content)) => {
657 if let Some(image) = Self::decode_image(image_content) {
658 *self = ContentBlock::Image { image };
659 } else {
660 let new_content = Self::image_md(image_content);
661 *self = Self::create_markdown_block(new_content, language_registry, cx);
662 }
663 }
664 (ContentBlock::Empty, _) => {
665 let new_content = Self::block_string_contents(&block, path_style);
666 *self = Self::create_markdown_block(new_content, language_registry, cx);
667 }
668 (ContentBlock::Markdown { markdown }, _) => {
669 let new_content = Self::block_string_contents(&block, path_style);
670 markdown.update(cx, |markdown, cx| markdown.append(&new_content, cx));
671 }
672 (ContentBlock::ResourceLink { resource_link }, _) => {
673 let existing_content = Self::resource_link_md(&resource_link.uri, path_style);
674 let new_content = Self::block_string_contents(&block, path_style);
675 let combined = format!("{}\n{}", existing_content, new_content);
676 *self = Self::create_markdown_block(combined, language_registry, cx);
677 }
678 (ContentBlock::Image { .. }, _) => {
679 let new_content = Self::block_string_contents(&block, path_style);
680 let combined = format!("`Image`\n{}", new_content);
681 *self = Self::create_markdown_block(combined, language_registry, cx);
682 }
683 }
684 }
685
686 fn decode_image(image_content: &acp::ImageContent) -> Option<Arc<gpui::Image>> {
687 use base64::Engine as _;
688
689 let bytes = base64::engine::general_purpose::STANDARD
690 .decode(image_content.data.as_bytes())
691 .ok()?;
692 let format = gpui::ImageFormat::from_mime_type(&image_content.mime_type)?;
693 Some(Arc::new(gpui::Image::from_bytes(format, bytes)))
694 }
695
696 fn create_markdown_block(
697 content: String,
698 language_registry: &Arc<LanguageRegistry>,
699 cx: &mut App,
700 ) -> ContentBlock {
701 ContentBlock::Markdown {
702 markdown: cx
703 .new(|cx| Markdown::new(content.into(), Some(language_registry.clone()), None, cx)),
704 }
705 }
706
707 fn block_string_contents(block: &acp::ContentBlock, path_style: PathStyle) -> String {
708 match block {
709 acp::ContentBlock::Text(text_content) => text_content.text.clone(),
710 acp::ContentBlock::ResourceLink(resource_link) => {
711 Self::resource_link_md(&resource_link.uri, path_style)
712 }
713 acp::ContentBlock::Resource(acp::EmbeddedResource {
714 resource:
715 acp::EmbeddedResourceResource::TextResourceContents(acp::TextResourceContents {
716 uri,
717 ..
718 }),
719 ..
720 }) => Self::resource_link_md(uri, path_style),
721 acp::ContentBlock::Image(image) => Self::image_md(image),
722 _ => String::new(),
723 }
724 }
725
726 fn resource_link_md(uri: &str, path_style: PathStyle) -> String {
727 if let Some(uri) = MentionUri::parse(uri, path_style).log_err() {
728 uri.as_link().to_string()
729 } else {
730 uri.to_string()
731 }
732 }
733
734 fn image_md(_image: &acp::ImageContent) -> String {
735 "`Image`".into()
736 }
737
738 pub fn to_markdown<'a>(&'a self, cx: &'a App) -> &'a str {
739 match self {
740 ContentBlock::Empty => "",
741 ContentBlock::Markdown { markdown } => markdown.read(cx).source(),
742 ContentBlock::ResourceLink { resource_link } => &resource_link.uri,
743 ContentBlock::Image { .. } => "`Image`",
744 }
745 }
746
747 pub fn markdown(&self) -> Option<&Entity<Markdown>> {
748 match self {
749 ContentBlock::Empty => None,
750 ContentBlock::Markdown { markdown } => Some(markdown),
751 ContentBlock::ResourceLink { .. } => None,
752 ContentBlock::Image { .. } => None,
753 }
754 }
755
756 pub fn resource_link(&self) -> Option<&acp::ResourceLink> {
757 match self {
758 ContentBlock::ResourceLink { resource_link } => Some(resource_link),
759 _ => None,
760 }
761 }
762
763 pub fn image(&self) -> Option<&Arc<gpui::Image>> {
764 match self {
765 ContentBlock::Image { image } => Some(image),
766 _ => None,
767 }
768 }
769}
770
771#[derive(Debug)]
772pub enum ToolCallContent {
773 ContentBlock(ContentBlock),
774 Diff(Entity<Diff>),
775 Terminal(Entity<Terminal>),
776}
777
778impl ToolCallContent {
779 pub fn from_acp(
780 content: acp::ToolCallContent,
781 language_registry: Arc<LanguageRegistry>,
782 path_style: PathStyle,
783 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
784 cx: &mut App,
785 ) -> Result<Option<Self>> {
786 match content {
787 acp::ToolCallContent::Content(acp::Content { content, .. }) => {
788 Ok(Some(Self::ContentBlock(ContentBlock::new(
789 content,
790 &language_registry,
791 path_style,
792 cx,
793 ))))
794 }
795 acp::ToolCallContent::Diff(diff) => Ok(Some(Self::Diff(cx.new(|cx| {
796 Diff::finalized(
797 diff.path.to_string_lossy().into_owned(),
798 diff.old_text,
799 diff.new_text,
800 language_registry,
801 cx,
802 )
803 })))),
804 acp::ToolCallContent::Terminal(acp::Terminal { terminal_id, .. }) => terminals
805 .get(&terminal_id)
806 .cloned()
807 .map(|terminal| Some(Self::Terminal(terminal)))
808 .ok_or_else(|| anyhow::anyhow!("Terminal with id `{}` not found", terminal_id)),
809 _ => Ok(None),
810 }
811 }
812
813 pub fn update_from_acp(
814 &mut self,
815 new: acp::ToolCallContent,
816 language_registry: Arc<LanguageRegistry>,
817 path_style: PathStyle,
818 terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
819 cx: &mut App,
820 ) -> Result<bool> {
821 let needs_update = match (&self, &new) {
822 (Self::Diff(old_diff), acp::ToolCallContent::Diff(new_diff)) => {
823 old_diff.read(cx).needs_update(
824 new_diff.old_text.as_deref().unwrap_or(""),
825 &new_diff.new_text,
826 cx,
827 )
828 }
829 _ => true,
830 };
831
832 if let Some(update) = Self::from_acp(new, language_registry, path_style, terminals, cx)? {
833 if needs_update {
834 *self = update;
835 }
836 Ok(true)
837 } else {
838 Ok(false)
839 }
840 }
841
842 pub fn to_markdown(&self, cx: &App) -> String {
843 match self {
844 Self::ContentBlock(content) => content.to_markdown(cx).to_string(),
845 Self::Diff(diff) => diff.read(cx).to_markdown(cx),
846 Self::Terminal(terminal) => terminal.read(cx).to_markdown(cx),
847 }
848 }
849
850 pub fn image(&self) -> Option<&Arc<gpui::Image>> {
851 match self {
852 Self::ContentBlock(content) => content.image(),
853 _ => None,
854 }
855 }
856}
857
858#[derive(Debug, PartialEq)]
859pub enum ToolCallUpdate {
860 UpdateFields(acp::ToolCallUpdate),
861 UpdateDiff(ToolCallUpdateDiff),
862 UpdateTerminal(ToolCallUpdateTerminal),
863}
864
865impl ToolCallUpdate {
866 fn id(&self) -> &acp::ToolCallId {
867 match self {
868 Self::UpdateFields(update) => &update.tool_call_id,
869 Self::UpdateDiff(diff) => &diff.id,
870 Self::UpdateTerminal(terminal) => &terminal.id,
871 }
872 }
873}
874
875impl From<acp::ToolCallUpdate> for ToolCallUpdate {
876 fn from(update: acp::ToolCallUpdate) -> Self {
877 Self::UpdateFields(update)
878 }
879}
880
881impl From<ToolCallUpdateDiff> for ToolCallUpdate {
882 fn from(diff: ToolCallUpdateDiff) -> Self {
883 Self::UpdateDiff(diff)
884 }
885}
886
887#[derive(Debug, PartialEq)]
888pub struct ToolCallUpdateDiff {
889 pub id: acp::ToolCallId,
890 pub diff: Entity<Diff>,
891}
892
893impl From<ToolCallUpdateTerminal> for ToolCallUpdate {
894 fn from(terminal: ToolCallUpdateTerminal) -> Self {
895 Self::UpdateTerminal(terminal)
896 }
897}
898
899#[derive(Debug, PartialEq)]
900pub struct ToolCallUpdateTerminal {
901 pub id: acp::ToolCallId,
902 pub terminal: Entity<Terminal>,
903}
904
905#[derive(Debug, Default)]
906pub struct Plan {
907 pub entries: Vec<PlanEntry>,
908}
909
910#[derive(Debug)]
911pub struct PlanStats<'a> {
912 pub in_progress_entry: Option<&'a PlanEntry>,
913 pub pending: u32,
914 pub completed: u32,
915}
916
917impl Plan {
918 pub fn is_empty(&self) -> bool {
919 self.entries.is_empty()
920 }
921
922 pub fn stats(&self) -> PlanStats<'_> {
923 let mut stats = PlanStats {
924 in_progress_entry: None,
925 pending: 0,
926 completed: 0,
927 };
928
929 for entry in &self.entries {
930 match &entry.status {
931 acp::PlanEntryStatus::Pending => {
932 stats.pending += 1;
933 }
934 acp::PlanEntryStatus::InProgress => {
935 stats.in_progress_entry = stats.in_progress_entry.or(Some(entry));
936 }
937 acp::PlanEntryStatus::Completed => {
938 stats.completed += 1;
939 }
940 _ => {}
941 }
942 }
943
944 stats
945 }
946}
947
948#[derive(Debug)]
949pub struct PlanEntry {
950 pub content: Entity<Markdown>,
951 pub priority: acp::PlanEntryPriority,
952 pub status: acp::PlanEntryStatus,
953}
954
955impl PlanEntry {
956 pub fn from_acp(entry: acp::PlanEntry, cx: &mut App) -> Self {
957 Self {
958 content: cx.new(|cx| Markdown::new(entry.content.into(), None, None, cx)),
959 priority: entry.priority,
960 status: entry.status,
961 }
962 }
963}
964
965#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
966pub struct TokenUsage {
967 pub max_tokens: u64,
968 pub used_tokens: u64,
969 pub input_tokens: u64,
970 pub output_tokens: u64,
971 pub max_output_tokens: Option<u64>,
972}
973
974pub const TOKEN_USAGE_WARNING_THRESHOLD: f32 = 0.8;
975
976impl TokenUsage {
977 pub fn ratio(&self) -> TokenUsageRatio {
978 #[cfg(debug_assertions)]
979 let warning_threshold: f32 = std::env::var("ZED_THREAD_WARNING_THRESHOLD")
980 .unwrap_or(TOKEN_USAGE_WARNING_THRESHOLD.to_string())
981 .parse()
982 .unwrap();
983 #[cfg(not(debug_assertions))]
984 let warning_threshold: f32 = TOKEN_USAGE_WARNING_THRESHOLD;
985
986 // When the maximum is unknown because there is no selected model,
987 // avoid showing the token limit warning.
988 if self.max_tokens == 0 {
989 TokenUsageRatio::Normal
990 } else if self.used_tokens >= self.max_tokens {
991 TokenUsageRatio::Exceeded
992 } else if self.used_tokens as f32 / self.max_tokens as f32 >= warning_threshold {
993 TokenUsageRatio::Warning
994 } else {
995 TokenUsageRatio::Normal
996 }
997 }
998}
999
1000#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
1001pub enum TokenUsageRatio {
1002 Normal,
1003 Warning,
1004 Exceeded,
1005}
1006
1007#[derive(Debug, Clone)]
1008pub struct RetryStatus {
1009 pub last_error: SharedString,
1010 pub attempt: usize,
1011 pub max_attempts: usize,
1012 pub started_at: Instant,
1013 pub duration: Duration,
1014}
1015
1016struct RunningTurn {
1017 id: u32,
1018 send_task: Task<()>,
1019}
1020
1021struct InferredEditCandidateReady {
1022 nonce: u64,
1023 buffer: Entity<Buffer>,
1024 baseline_snapshot: text::BufferSnapshot,
1025 existed_on_disk: bool,
1026 was_dirty: bool,
1027 observed_external_file_change: bool,
1028 has_inferred_edit: bool,
1029 _subscription: Subscription,
1030}
1031
1032enum InferredEditCandidateState {
1033 Pending { nonce: u64 },
1034 Ready(InferredEditCandidateReady),
1035}
1036
1037impl InferredEditCandidateState {
1038 fn nonce(&self) -> u64 {
1039 match self {
1040 Self::Pending { nonce } => *nonce,
1041 Self::Ready(candidate) => candidate.nonce,
1042 }
1043 }
1044}
1045
1046pub struct AcpThread {
1047 session_id: acp::SessionId,
1048 work_dirs: Option<PathList>,
1049 parent_session_id: Option<acp::SessionId>,
1050 title: SharedString,
1051 provisional_title: Option<SharedString>,
1052 entries: Vec<AgentThreadEntry>,
1053 plan: Plan,
1054 project: Entity<Project>,
1055 action_log: Entity<ActionLog>,
1056 shared_buffers: HashMap<Entity<Buffer>, BufferSnapshot>,
1057 turn_id: u32,
1058 running_turn: Option<RunningTurn>,
1059 connection: Rc<dyn AgentConnection>,
1060 token_usage: Option<TokenUsage>,
1061 prompt_capabilities: acp::PromptCapabilities,
1062 _observe_prompt_capabilities: Task<anyhow::Result<()>>,
1063 terminals: HashMap<acp::TerminalId, Entity<Terminal>>,
1064 pending_terminal_output: HashMap<acp::TerminalId, Vec<Vec<u8>>>,
1065 pending_terminal_exit: HashMap<acp::TerminalId, acp::TerminalExitStatus>,
1066 inferred_edit_candidates:
1067 HashMap<acp::ToolCallId, HashMap<PathBuf, InferredEditCandidateState>>,
1068 finalizing_inferred_edit_tool_calls: HashSet<acp::ToolCallId>,
1069 next_inferred_edit_candidate_nonce: u64,
1070 had_error: bool,
1071 /// The user's unsent prompt text, persisted so it can be restored when reloading the thread.
1072 draft_prompt: Option<Vec<acp::ContentBlock>>,
1073 /// The initial scroll position for the thread view, set during session registration.
1074 ui_scroll_position: Option<gpui::ListOffset>,
1075 /// Buffer for smooth text streaming. Holds text that has been received from
1076 /// the model but not yet revealed in the UI. A timer task drains this buffer
1077 /// gradually to create a fluid typing effect instead of choppy chunk-at-a-time
1078 /// updates.
1079 streaming_text_buffer: Option<StreamingTextBuffer>,
1080}
1081
1082struct StreamingTextBuffer {
1083 /// Text received from the model but not yet appended to the Markdown source.
1084 pending: String,
1085 /// The number of bytes to reveal per timer turn.
1086 bytes_to_reveal_per_tick: usize,
1087 /// The Markdown entity being streamed into.
1088 target: Entity<Markdown>,
1089 /// Timer task that periodically moves text from `pending` into `source`.
1090 _reveal_task: Task<()>,
1091}
1092
1093impl StreamingTextBuffer {
1094 /// The number of milliseconds between each timer tick, controlling how quickly
1095 /// text is revealed.
1096 const TASK_UPDATE_MS: u64 = 16;
1097 /// The time in milliseconds to reveal the entire pending text.
1098 const REVEAL_TARGET: f32 = 200.0;
1099}
1100
1101impl From<&AcpThread> for ActionLogTelemetry {
1102 fn from(value: &AcpThread) -> Self {
1103 Self {
1104 agent_telemetry_id: value.connection().telemetry_id(),
1105 session_id: value.session_id.0.clone(),
1106 }
1107 }
1108}
1109
1110#[derive(Debug)]
1111pub enum AcpThreadEvent {
1112 NewEntry,
1113 TitleUpdated,
1114 TokenUsageUpdated,
1115 EntryUpdated(usize),
1116 EntriesRemoved(Range<usize>),
1117 ToolAuthorizationRequested(acp::ToolCallId),
1118 ToolAuthorizationReceived(acp::ToolCallId),
1119 Retry(RetryStatus),
1120 SubagentSpawned(acp::SessionId),
1121 Stopped(acp::StopReason),
1122 Error,
1123 LoadError(LoadError),
1124 PromptCapabilitiesUpdated,
1125 Refusal,
1126 AvailableCommandsUpdated(Vec<acp::AvailableCommand>),
1127 ModeUpdated(acp::SessionModeId),
1128 ConfigOptionsUpdated(Vec<acp::SessionConfigOption>),
1129}
1130
1131impl EventEmitter<AcpThreadEvent> for AcpThread {}
1132
1133#[derive(Debug, Clone)]
1134pub enum TerminalProviderEvent {
1135 Created {
1136 terminal_id: acp::TerminalId,
1137 label: String,
1138 cwd: Option<PathBuf>,
1139 output_byte_limit: Option<u64>,
1140 terminal: Entity<::terminal::Terminal>,
1141 },
1142 Output {
1143 terminal_id: acp::TerminalId,
1144 data: Vec<u8>,
1145 },
1146 TitleChanged {
1147 terminal_id: acp::TerminalId,
1148 title: String,
1149 },
1150 Exit {
1151 terminal_id: acp::TerminalId,
1152 status: acp::TerminalExitStatus,
1153 },
1154}
1155
1156#[derive(Debug, Clone)]
1157pub enum TerminalProviderCommand {
1158 WriteInput {
1159 terminal_id: acp::TerminalId,
1160 bytes: Vec<u8>,
1161 },
1162 Resize {
1163 terminal_id: acp::TerminalId,
1164 cols: u16,
1165 rows: u16,
1166 },
1167 Close {
1168 terminal_id: acp::TerminalId,
1169 },
1170}
1171
1172#[derive(PartialEq, Eq, Debug)]
1173pub enum ThreadStatus {
1174 Idle,
1175 Generating,
1176}
1177
1178#[derive(Debug, Clone)]
1179pub enum LoadError {
1180 Unsupported {
1181 command: SharedString,
1182 current_version: SharedString,
1183 minimum_version: SharedString,
1184 },
1185 FailedToInstall(SharedString),
1186 Exited {
1187 status: ExitStatus,
1188 },
1189 Other(SharedString),
1190}
1191
1192impl Display for LoadError {
1193 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1194 match self {
1195 LoadError::Unsupported {
1196 command: path,
1197 current_version,
1198 minimum_version,
1199 } => {
1200 write!(
1201 f,
1202 "version {current_version} from {path} is not supported (need at least {minimum_version})"
1203 )
1204 }
1205 LoadError::FailedToInstall(msg) => write!(f, "Failed to install: {msg}"),
1206 LoadError::Exited { status } => write!(f, "Server exited with status {status}"),
1207 LoadError::Other(msg) => write!(f, "{msg}"),
1208 }
1209 }
1210}
1211
1212impl Error for LoadError {}
1213
1214impl AcpThread {
1215 pub fn new(
1216 parent_session_id: Option<acp::SessionId>,
1217 title: impl Into<SharedString>,
1218 work_dirs: Option<PathList>,
1219 connection: Rc<dyn AgentConnection>,
1220 project: Entity<Project>,
1221 action_log: Entity<ActionLog>,
1222 session_id: acp::SessionId,
1223 mut prompt_capabilities_rx: watch::Receiver<acp::PromptCapabilities>,
1224 cx: &mut Context<Self>,
1225 ) -> Self {
1226 let prompt_capabilities = prompt_capabilities_rx.borrow().clone();
1227 let task = cx.spawn::<_, anyhow::Result<()>>(async move |this, cx| {
1228 loop {
1229 let caps = prompt_capabilities_rx.recv().await?;
1230 this.update(cx, |this, cx| {
1231 this.prompt_capabilities = caps;
1232 cx.emit(AcpThreadEvent::PromptCapabilitiesUpdated);
1233 })?;
1234 }
1235 });
1236
1237 Self {
1238 parent_session_id,
1239 work_dirs,
1240 action_log,
1241 shared_buffers: Default::default(),
1242 entries: Default::default(),
1243 plan: Default::default(),
1244 title: title.into(),
1245 provisional_title: None,
1246 project,
1247 running_turn: None,
1248 turn_id: 0,
1249 connection,
1250 session_id,
1251 token_usage: None,
1252 prompt_capabilities,
1253 _observe_prompt_capabilities: task,
1254 terminals: HashMap::default(),
1255 pending_terminal_output: HashMap::default(),
1256 pending_terminal_exit: HashMap::default(),
1257 inferred_edit_candidates: HashMap::default(),
1258 finalizing_inferred_edit_tool_calls: HashSet::default(),
1259 next_inferred_edit_candidate_nonce: 0,
1260 had_error: false,
1261 draft_prompt: None,
1262 ui_scroll_position: None,
1263 streaming_text_buffer: None,
1264 }
1265 }
1266
1267 pub fn parent_session_id(&self) -> Option<&acp::SessionId> {
1268 self.parent_session_id.as_ref()
1269 }
1270
1271 pub fn prompt_capabilities(&self) -> acp::PromptCapabilities {
1272 self.prompt_capabilities.clone()
1273 }
1274
1275 pub fn draft_prompt(&self) -> Option<&[acp::ContentBlock]> {
1276 self.draft_prompt.as_deref()
1277 }
1278
1279 pub fn set_draft_prompt(&mut self, prompt: Option<Vec<acp::ContentBlock>>) {
1280 self.draft_prompt = prompt;
1281 }
1282
1283 pub fn ui_scroll_position(&self) -> Option<gpui::ListOffset> {
1284 self.ui_scroll_position
1285 }
1286
1287 pub fn set_ui_scroll_position(&mut self, position: Option<gpui::ListOffset>) {
1288 self.ui_scroll_position = position;
1289 }
1290
1291 pub fn connection(&self) -> &Rc<dyn AgentConnection> {
1292 &self.connection
1293 }
1294
1295 pub fn action_log(&self) -> &Entity<ActionLog> {
1296 &self.action_log
1297 }
1298
1299 pub fn project(&self) -> &Entity<Project> {
1300 &self.project
1301 }
1302
1303 pub fn title(&self) -> SharedString {
1304 self.provisional_title
1305 .clone()
1306 .unwrap_or_else(|| self.title.clone())
1307 }
1308
1309 pub fn has_provisional_title(&self) -> bool {
1310 self.provisional_title.is_some()
1311 }
1312
1313 pub fn entries(&self) -> &[AgentThreadEntry] {
1314 &self.entries
1315 }
1316
1317 pub fn session_id(&self) -> &acp::SessionId {
1318 &self.session_id
1319 }
1320
1321 pub fn work_dirs(&self) -> Option<&PathList> {
1322 self.work_dirs.as_ref()
1323 }
1324
1325 pub fn status(&self) -> ThreadStatus {
1326 if self.running_turn.is_some() {
1327 ThreadStatus::Generating
1328 } else {
1329 ThreadStatus::Idle
1330 }
1331 }
1332
1333 pub fn had_error(&self) -> bool {
1334 self.had_error
1335 }
1336
1337 pub fn is_waiting_for_confirmation(&self) -> bool {
1338 for entry in self.entries.iter().rev() {
1339 match entry {
1340 AgentThreadEntry::UserMessage(_) => return false,
1341 AgentThreadEntry::ToolCall(ToolCall {
1342 status: ToolCallStatus::WaitingForConfirmation { .. },
1343 ..
1344 }) => return true,
1345 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1346 }
1347 }
1348 false
1349 }
1350
1351 pub fn token_usage(&self) -> Option<&TokenUsage> {
1352 self.token_usage.as_ref()
1353 }
1354
1355 pub fn has_pending_edit_tool_calls(&self) -> bool {
1356 for entry in self.entries.iter().rev() {
1357 match entry {
1358 AgentThreadEntry::UserMessage(_) => return false,
1359 AgentThreadEntry::ToolCall(
1360 call @ ToolCall {
1361 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
1362 ..
1363 },
1364 ) if call.diffs().next().is_some() || Self::should_infer_external_edits(call) => {
1365 return true;
1366 }
1367 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1368 }
1369 }
1370
1371 false
1372 }
1373
1374 pub fn has_in_progress_tool_calls(&self) -> bool {
1375 for entry in self.entries.iter().rev() {
1376 match entry {
1377 AgentThreadEntry::UserMessage(_) => return false,
1378 AgentThreadEntry::ToolCall(ToolCall {
1379 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
1380 ..
1381 }) => {
1382 return true;
1383 }
1384 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1385 }
1386 }
1387
1388 false
1389 }
1390
1391 pub fn used_tools_since_last_user_message(&self) -> bool {
1392 for entry in self.entries.iter().rev() {
1393 match entry {
1394 AgentThreadEntry::UserMessage(..) => return false,
1395 AgentThreadEntry::AssistantMessage(..) => continue,
1396 AgentThreadEntry::ToolCall(..) => return true,
1397 }
1398 }
1399
1400 false
1401 }
1402
1403 pub fn handle_session_update(
1404 &mut self,
1405 update: acp::SessionUpdate,
1406 cx: &mut Context<Self>,
1407 ) -> Result<(), acp::Error> {
1408 match update {
1409 acp::SessionUpdate::UserMessageChunk(acp::ContentChunk { content, .. }) => {
1410 self.push_user_content_block(None, content, cx);
1411 }
1412 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk { content, .. }) => {
1413 self.push_assistant_content_block(content, false, cx);
1414 }
1415 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk { content, .. }) => {
1416 self.push_assistant_content_block(content, true, cx);
1417 }
1418 acp::SessionUpdate::ToolCall(tool_call) => {
1419 self.upsert_tool_call(tool_call, cx)?;
1420 }
1421 acp::SessionUpdate::ToolCallUpdate(tool_call_update) => {
1422 self.update_tool_call(tool_call_update, cx)?;
1423 }
1424 acp::SessionUpdate::Plan(plan) => {
1425 self.update_plan(plan, cx);
1426 }
1427 acp::SessionUpdate::SessionInfoUpdate(info_update) => {
1428 if let acp::MaybeUndefined::Value(title) = info_update.title {
1429 let had_provisional = self.provisional_title.take().is_some();
1430 let title: SharedString = title.into();
1431 if title != self.title {
1432 self.title = title;
1433 cx.emit(AcpThreadEvent::TitleUpdated);
1434 } else if had_provisional {
1435 cx.emit(AcpThreadEvent::TitleUpdated);
1436 }
1437 }
1438 }
1439 acp::SessionUpdate::AvailableCommandsUpdate(acp::AvailableCommandsUpdate {
1440 available_commands,
1441 ..
1442 }) => cx.emit(AcpThreadEvent::AvailableCommandsUpdated(available_commands)),
1443 acp::SessionUpdate::CurrentModeUpdate(acp::CurrentModeUpdate {
1444 current_mode_id,
1445 ..
1446 }) => cx.emit(AcpThreadEvent::ModeUpdated(current_mode_id)),
1447 acp::SessionUpdate::ConfigOptionUpdate(acp::ConfigOptionUpdate {
1448 config_options,
1449 ..
1450 }) => cx.emit(AcpThreadEvent::ConfigOptionsUpdated(config_options)),
1451 _ => {}
1452 }
1453 Ok(())
1454 }
1455
1456 pub fn push_user_content_block(
1457 &mut self,
1458 message_id: Option<UserMessageId>,
1459 chunk: acp::ContentBlock,
1460 cx: &mut Context<Self>,
1461 ) {
1462 self.push_user_content_block_with_indent(message_id, chunk, false, cx)
1463 }
1464
1465 pub fn push_user_content_block_with_indent(
1466 &mut self,
1467 message_id: Option<UserMessageId>,
1468 chunk: acp::ContentBlock,
1469 indented: bool,
1470 cx: &mut Context<Self>,
1471 ) {
1472 let language_registry = self.project.read(cx).languages().clone();
1473 let path_style = self.project.read(cx).path_style(cx);
1474 let entries_len = self.entries.len();
1475
1476 if let Some(last_entry) = self.entries.last_mut()
1477 && let AgentThreadEntry::UserMessage(UserMessage {
1478 id,
1479 content,
1480 chunks,
1481 indented: existing_indented,
1482 ..
1483 }) = last_entry
1484 && *existing_indented == indented
1485 {
1486 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1487 *id = message_id.or(id.take());
1488 content.append(chunk.clone(), &language_registry, path_style, cx);
1489 chunks.push(chunk);
1490 let idx = entries_len - 1;
1491 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1492 } else {
1493 let content = ContentBlock::new(chunk.clone(), &language_registry, path_style, cx);
1494 self.push_entry(
1495 AgentThreadEntry::UserMessage(UserMessage {
1496 id: message_id,
1497 content,
1498 chunks: vec![chunk],
1499 checkpoint: None,
1500 indented,
1501 }),
1502 cx,
1503 );
1504 }
1505 }
1506
1507 pub fn push_assistant_content_block(
1508 &mut self,
1509 chunk: acp::ContentBlock,
1510 is_thought: bool,
1511 cx: &mut Context<Self>,
1512 ) {
1513 self.push_assistant_content_block_with_indent(chunk, is_thought, false, cx)
1514 }
1515
1516 pub fn push_assistant_content_block_with_indent(
1517 &mut self,
1518 chunk: acp::ContentBlock,
1519 is_thought: bool,
1520 indented: bool,
1521 cx: &mut Context<Self>,
1522 ) {
1523 let path_style = self.project.read(cx).path_style(cx);
1524
1525 // For text chunks going to an existing Markdown block, buffer for smooth
1526 // streaming instead of appending all at once which may feel more choppy.
1527 if let acp::ContentBlock::Text(text_content) = &chunk {
1528 if let Some(markdown) = self.streaming_markdown_target(is_thought, indented) {
1529 let entries_len = self.entries.len();
1530 cx.emit(AcpThreadEvent::EntryUpdated(entries_len - 1));
1531 self.buffer_streaming_text(&markdown, text_content.text.clone(), cx);
1532 return;
1533 }
1534 }
1535
1536 let language_registry = self.project.read(cx).languages().clone();
1537 let entries_len = self.entries.len();
1538 if let Some(last_entry) = self.entries.last_mut()
1539 && let AgentThreadEntry::AssistantMessage(AssistantMessage {
1540 chunks,
1541 indented: existing_indented,
1542 is_subagent_output: _,
1543 }) = last_entry
1544 && *existing_indented == indented
1545 {
1546 let idx = entries_len - 1;
1547 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1548 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1549 match (chunks.last_mut(), is_thought) {
1550 (Some(AssistantMessageChunk::Message { block }), false)
1551 | (Some(AssistantMessageChunk::Thought { block }), true) => {
1552 block.append(chunk, &language_registry, path_style, cx)
1553 }
1554 _ => {
1555 let block = ContentBlock::new(chunk, &language_registry, path_style, cx);
1556 if is_thought {
1557 chunks.push(AssistantMessageChunk::Thought { block })
1558 } else {
1559 chunks.push(AssistantMessageChunk::Message { block })
1560 }
1561 }
1562 }
1563 } else {
1564 let block = ContentBlock::new(chunk, &language_registry, path_style, cx);
1565 let chunk = if is_thought {
1566 AssistantMessageChunk::Thought { block }
1567 } else {
1568 AssistantMessageChunk::Message { block }
1569 };
1570
1571 self.push_entry(
1572 AgentThreadEntry::AssistantMessage(AssistantMessage {
1573 chunks: vec![chunk],
1574 indented,
1575 is_subagent_output: false,
1576 }),
1577 cx,
1578 );
1579 }
1580 }
1581
1582 fn streaming_markdown_target(
1583 &self,
1584 is_thought: bool,
1585 indented: bool,
1586 ) -> Option<Entity<Markdown>> {
1587 let last_entry = self.entries.last()?;
1588 if let AgentThreadEntry::AssistantMessage(AssistantMessage {
1589 chunks,
1590 indented: existing_indented,
1591 ..
1592 }) = last_entry
1593 && *existing_indented == indented
1594 && let [.., chunk] = chunks.as_slice()
1595 {
1596 match (chunk, is_thought) {
1597 (
1598 AssistantMessageChunk::Message {
1599 block: ContentBlock::Markdown { markdown },
1600 },
1601 false,
1602 )
1603 | (
1604 AssistantMessageChunk::Thought {
1605 block: ContentBlock::Markdown { markdown },
1606 },
1607 true,
1608 ) => Some(markdown.clone()),
1609 _ => None,
1610 }
1611 } else {
1612 None
1613 }
1614 }
1615
1616 /// Add text to the streaming buffer. If the target changed (e.g. switching
1617 /// from thoughts to message text), flush the old buffer first.
1618 fn buffer_streaming_text(
1619 &mut self,
1620 markdown: &Entity<Markdown>,
1621 text: String,
1622 cx: &mut Context<Self>,
1623 ) {
1624 if let Some(buffer) = &mut self.streaming_text_buffer {
1625 if buffer.target.entity_id() == markdown.entity_id() {
1626 buffer.pending.push_str(&text);
1627
1628 buffer.bytes_to_reveal_per_tick = (buffer.pending.len() as f32
1629 / StreamingTextBuffer::REVEAL_TARGET
1630 * StreamingTextBuffer::TASK_UPDATE_MS as f32)
1631 .ceil() as usize;
1632 return;
1633 }
1634 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1635 }
1636
1637 let target = markdown.clone();
1638 let _reveal_task = self.start_streaming_reveal(cx);
1639 let pending_len = text.len();
1640 let bytes_to_reveal = (pending_len as f32 / StreamingTextBuffer::REVEAL_TARGET
1641 * StreamingTextBuffer::TASK_UPDATE_MS as f32)
1642 .ceil() as usize;
1643 self.streaming_text_buffer = Some(StreamingTextBuffer {
1644 pending: text,
1645 bytes_to_reveal_per_tick: bytes_to_reveal,
1646 target,
1647 _reveal_task,
1648 });
1649 }
1650
1651 /// Flush all buffered streaming text into the Markdown entity immediately.
1652 fn flush_streaming_text(
1653 streaming_text_buffer: &mut Option<StreamingTextBuffer>,
1654 cx: &mut Context<Self>,
1655 ) {
1656 if let Some(buffer) = streaming_text_buffer.take() {
1657 if !buffer.pending.is_empty() {
1658 buffer
1659 .target
1660 .update(cx, |markdown, cx| markdown.append(&buffer.pending, cx));
1661 }
1662 }
1663 }
1664
1665 /// Spawns a foreground task that periodically drains
1666 /// `streaming_text_buffer.pending` into the target `Markdown` entity,
1667 /// producing smooth, continuous text output.
1668 fn start_streaming_reveal(&self, cx: &mut Context<Self>) -> Task<()> {
1669 cx.spawn(async move |this, cx| {
1670 loop {
1671 cx.background_executor()
1672 .timer(Duration::from_millis(StreamingTextBuffer::TASK_UPDATE_MS))
1673 .await;
1674
1675 let should_continue = this
1676 .update(cx, |this, cx| {
1677 let Some(buffer) = &mut this.streaming_text_buffer else {
1678 return false;
1679 };
1680
1681 if buffer.pending.is_empty() {
1682 return true;
1683 }
1684
1685 let pending_len = buffer.pending.len();
1686
1687 let byte_boundary = buffer
1688 .pending
1689 .ceil_char_boundary(buffer.bytes_to_reveal_per_tick)
1690 .min(pending_len);
1691
1692 buffer.target.update(cx, |markdown: &mut Markdown, cx| {
1693 markdown.append(&buffer.pending[..byte_boundary], cx);
1694 buffer.pending.drain(..byte_boundary);
1695 });
1696
1697 true
1698 })
1699 .unwrap_or(false);
1700
1701 if !should_continue {
1702 break;
1703 }
1704 }
1705 })
1706 }
1707
1708 fn push_entry(&mut self, entry: AgentThreadEntry, cx: &mut Context<Self>) {
1709 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1710 self.entries.push(entry);
1711 cx.emit(AcpThreadEvent::NewEntry);
1712 }
1713
1714 pub fn can_set_title(&mut self, cx: &mut Context<Self>) -> bool {
1715 self.connection.set_title(&self.session_id, cx).is_some()
1716 }
1717
1718 pub fn set_title(&mut self, title: SharedString, cx: &mut Context<Self>) -> Task<Result<()>> {
1719 let had_provisional = self.provisional_title.take().is_some();
1720 if title != self.title {
1721 self.title = title.clone();
1722 cx.emit(AcpThreadEvent::TitleUpdated);
1723 if let Some(set_title) = self.connection.set_title(&self.session_id, cx) {
1724 return set_title.run(title, cx);
1725 }
1726 } else if had_provisional {
1727 cx.emit(AcpThreadEvent::TitleUpdated);
1728 }
1729 Task::ready(Ok(()))
1730 }
1731
1732 /// Sets a provisional display title without propagating back to the
1733 /// underlying agent connection. This is used for quick preview titles
1734 /// (e.g. first 20 chars of the user message) that should be shown
1735 /// immediately but replaced once the LLM generates a proper title via
1736 /// `set_title`.
1737 pub fn set_provisional_title(&mut self, title: SharedString, cx: &mut Context<Self>) {
1738 self.provisional_title = Some(title);
1739 cx.emit(AcpThreadEvent::TitleUpdated);
1740 }
1741
1742 pub fn subagent_spawned(&mut self, session_id: acp::SessionId, cx: &mut Context<Self>) {
1743 cx.emit(AcpThreadEvent::SubagentSpawned(session_id));
1744 }
1745
1746 pub fn update_token_usage(&mut self, usage: Option<TokenUsage>, cx: &mut Context<Self>) {
1747 self.token_usage = usage;
1748 cx.emit(AcpThreadEvent::TokenUsageUpdated);
1749 }
1750
1751 pub fn update_retry_status(&mut self, status: RetryStatus, cx: &mut Context<Self>) {
1752 cx.emit(AcpThreadEvent::Retry(status));
1753 }
1754
1755 fn should_infer_external_edits(tool_call: &ToolCall) -> bool {
1756 tool_call.tool_name.is_none()
1757 && tool_call.kind == acp::ToolKind::Edit
1758 && !tool_call.locations.is_empty()
1759 }
1760
1761 fn is_inferred_edit_terminal_status(status: &ToolCallStatus) -> bool {
1762 matches!(status, ToolCallStatus::Completed | ToolCallStatus::Failed)
1763 }
1764
1765 fn allocate_inferred_edit_candidate_nonce(&mut self) -> u64 {
1766 let nonce = self.next_inferred_edit_candidate_nonce;
1767 self.next_inferred_edit_candidate_nonce =
1768 self.next_inferred_edit_candidate_nonce.wrapping_add(1);
1769 nonce
1770 }
1771
1772 fn remove_inferred_edit_candidate_if_matching(
1773 &mut self,
1774 tool_call_id: &acp::ToolCallId,
1775 abs_path: &PathBuf,
1776 nonce: u64,
1777 ) {
1778 let remove_tool_call =
1779 if let Some(candidates) = self.inferred_edit_candidates.get_mut(tool_call_id) {
1780 let should_remove = candidates
1781 .get(abs_path)
1782 .is_some_and(|candidate_state| candidate_state.nonce() == nonce);
1783 if should_remove {
1784 candidates.remove(abs_path);
1785 }
1786 candidates.is_empty()
1787 } else {
1788 false
1789 };
1790
1791 if remove_tool_call {
1792 self.inferred_edit_candidates.remove(tool_call_id);
1793 self.finalizing_inferred_edit_tool_calls
1794 .remove(tool_call_id);
1795 }
1796 }
1797
1798 fn clear_inferred_edit_candidates_for_tool_calls(
1799 &mut self,
1800 tool_call_ids: impl IntoIterator<Item = acp::ToolCallId>,
1801 ) {
1802 for tool_call_id in tool_call_ids {
1803 self.inferred_edit_candidates.remove(&tool_call_id);
1804 self.finalizing_inferred_edit_tool_calls
1805 .remove(&tool_call_id);
1806 }
1807 }
1808
1809 fn finalize_all_inferred_edit_tool_calls(&mut self, cx: &mut Context<Self>) {
1810 let tool_call_ids = self
1811 .inferred_edit_candidates
1812 .keys()
1813 .filter(|tool_call_id| {
1814 self.tool_call(tool_call_id).is_some_and(|(_, tool_call)| {
1815 Self::should_infer_external_edits(tool_call)
1816 && Self::is_inferred_edit_terminal_status(&tool_call.status)
1817 })
1818 })
1819 .cloned()
1820 .collect::<Vec<_>>();
1821 for tool_call_id in tool_call_ids {
1822 self.finalize_inferred_edit_tool_call(tool_call_id, cx);
1823 }
1824 }
1825
1826 fn sync_inferred_edit_candidate_paths(
1827 &mut self,
1828 tool_call_id: &acp::ToolCallId,
1829 locations: &[acp::ToolCallLocation],
1830 ) {
1831 let mut current_paths = HashSet::default();
1832 for location in locations {
1833 current_paths.insert(location.path.clone());
1834 }
1835
1836 let remove_tool_call =
1837 if let Some(candidates) = self.inferred_edit_candidates.get_mut(tool_call_id) {
1838 candidates.retain(|path, _| current_paths.contains(path));
1839 candidates.is_empty()
1840 } else {
1841 false
1842 };
1843
1844 if remove_tool_call {
1845 self.inferred_edit_candidates.remove(tool_call_id);
1846 self.finalizing_inferred_edit_tool_calls
1847 .remove(tool_call_id);
1848 }
1849 }
1850
1851 fn register_inferred_edit_locations(
1852 &mut self,
1853 tool_call_id: acp::ToolCallId,
1854 locations: &[acp::ToolCallLocation],
1855 cx: &mut Context<Self>,
1856 ) {
1857 self.sync_inferred_edit_candidate_paths(&tool_call_id, locations);
1858
1859 let mut unique_paths = HashSet::default();
1860 for location in locations {
1861 let abs_path = location.path.clone();
1862 if !unique_paths.insert(abs_path.clone()) {
1863 continue;
1864 }
1865
1866 let nonce = self.allocate_inferred_edit_candidate_nonce();
1867 let candidates = self
1868 .inferred_edit_candidates
1869 .entry(tool_call_id.clone())
1870 .or_default();
1871 if candidates.contains_key(&abs_path) {
1872 continue;
1873 }
1874
1875 candidates.insert(
1876 abs_path.clone(),
1877 InferredEditCandidateState::Pending { nonce },
1878 );
1879
1880 let open_buffer = self.project.update(cx, |project, cx| {
1881 let project_path = project.project_path_for_absolute_path(&abs_path, cx)?;
1882 if let Some(buffer) = project.get_open_buffer(&project_path, cx) {
1883 let (baseline_snapshot, existed_on_disk, was_dirty) =
1884 Self::inferred_edit_candidate_baseline(&buffer, cx);
1885 return Some((
1886 Some((buffer, baseline_snapshot, existed_on_disk, was_dirty)),
1887 None,
1888 ));
1889 }
1890
1891 Some((None, Some(project.open_buffer(project_path, cx))))
1892 });
1893
1894 let Some((ready_candidate, open_buffer)) = open_buffer else {
1895 self.remove_inferred_edit_candidate_if_matching(&tool_call_id, &abs_path, nonce);
1896 continue;
1897 };
1898
1899 if let Some((buffer, baseline_snapshot, existed_on_disk, was_dirty)) = ready_candidate {
1900 self.set_inferred_edit_candidate_ready(
1901 tool_call_id.clone(),
1902 abs_path,
1903 nonce,
1904 buffer,
1905 baseline_snapshot,
1906 existed_on_disk,
1907 was_dirty,
1908 cx,
1909 );
1910 continue;
1911 }
1912
1913 let Some(open_buffer) = open_buffer else {
1914 self.remove_inferred_edit_candidate_if_matching(&tool_call_id, &abs_path, nonce);
1915 continue;
1916 };
1917
1918 let tool_call_id = tool_call_id.clone();
1919 cx.spawn::<_, anyhow::Result<()>>(async move |this, cx| {
1920 let buffer = match open_buffer.await {
1921 Ok(buffer) => buffer,
1922 Err(_) => {
1923 this.update(cx, |this, _| {
1924 this.remove_inferred_edit_candidate_if_matching(
1925 &tool_call_id,
1926 &abs_path,
1927 nonce,
1928 );
1929 })
1930 .ok();
1931 return Ok(());
1932 }
1933 };
1934
1935 let (baseline_snapshot, existed_on_disk, was_dirty) =
1936 Self::inferred_edit_candidate_baseline(&buffer, cx);
1937
1938 this.update(cx, |this, cx| {
1939 this.set_inferred_edit_candidate_ready(
1940 tool_call_id.clone(),
1941 abs_path.clone(),
1942 nonce,
1943 buffer,
1944 baseline_snapshot,
1945 existed_on_disk,
1946 was_dirty,
1947 cx,
1948 );
1949 })
1950 .ok();
1951
1952 Ok(())
1953 })
1954 .detach_and_log_err(cx);
1955 }
1956 }
1957
1958 fn inferred_edit_candidate_baseline(
1959 buffer: &Entity<Buffer>,
1960 cx: &mut impl AppContext,
1961 ) -> (text::BufferSnapshot, bool, bool) {
1962 buffer.read_with(cx, |buffer, _| {
1963 (
1964 buffer.text_snapshot(),
1965 buffer.file().is_some_and(|file| file.disk_state().exists()),
1966 buffer.is_dirty(),
1967 )
1968 })
1969 }
1970
1971 fn update_ready_inferred_edit_candidate_if_matching(
1972 &mut self,
1973 tool_call_id: &acp::ToolCallId,
1974 abs_path: &PathBuf,
1975 nonce: u64,
1976 update: impl FnOnce(&mut InferredEditCandidateReady),
1977 ) -> bool {
1978 let Some(candidates) = self.inferred_edit_candidates.get_mut(tool_call_id) else {
1979 return false;
1980 };
1981 let Some(InferredEditCandidateState::Ready(candidate)) = candidates.get_mut(abs_path)
1982 else {
1983 return false;
1984 };
1985 if candidate.nonce != nonce {
1986 return false;
1987 }
1988
1989 update(candidate);
1990 true
1991 }
1992
1993 fn set_inferred_edit_candidate_ready(
1994 &mut self,
1995 tool_call_id: acp::ToolCallId,
1996 abs_path: PathBuf,
1997 nonce: u64,
1998 buffer: Entity<Buffer>,
1999 baseline_snapshot: text::BufferSnapshot,
2000 existed_on_disk: bool,
2001 was_dirty: bool,
2002 cx: &mut Context<Self>,
2003 ) {
2004 let subscription = cx.subscribe(&buffer, {
2005 let tool_call_id = tool_call_id.clone();
2006 let abs_path = abs_path.clone();
2007 move |this, buffer, event: &BufferEvent, cx| {
2008 this.handle_inferred_edit_candidate_buffer_event(
2009 tool_call_id.clone(),
2010 abs_path.clone(),
2011 nonce,
2012 buffer,
2013 event,
2014 cx,
2015 );
2016 }
2017 });
2018
2019 let Some(candidates) = self.inferred_edit_candidates.get_mut(&tool_call_id) else {
2020 return;
2021 };
2022 let Some(candidate_state) = candidates.get_mut(&abs_path) else {
2023 return;
2024 };
2025 if candidate_state.nonce() != nonce {
2026 return;
2027 }
2028
2029 *candidate_state = InferredEditCandidateState::Ready(InferredEditCandidateReady {
2030 nonce,
2031 buffer,
2032 baseline_snapshot,
2033 existed_on_disk,
2034 was_dirty,
2035 observed_external_file_change: false,
2036 has_inferred_edit: false,
2037 _subscription: subscription,
2038 });
2039 }
2040
2041 fn can_infer_initial_external_edit_for_candidate(
2042 &self,
2043 buffer: &Entity<Buffer>,
2044 was_dirty: bool,
2045 cx: &App,
2046 ) -> bool {
2047 if was_dirty {
2048 return false;
2049 }
2050
2051 if buffer.read_with(cx, |buffer, _| buffer.is_dirty()) {
2052 return false;
2053 }
2054
2055 !self.action_log.read_with(cx, |action_log, cx| {
2056 action_log.has_changed_buffer(buffer, cx)
2057 })
2058 }
2059
2060 fn inferred_edit_candidate_has_meaningful_change(
2061 existed_on_disk: bool,
2062 baseline_snapshot: &text::BufferSnapshot,
2063 buffer: &Entity<Buffer>,
2064 cx: &App,
2065 ) -> bool {
2066 let (current_snapshot, current_exists) = buffer.read_with(cx, |buffer, _| {
2067 (
2068 buffer.text_snapshot(),
2069 buffer.file().is_some_and(|file| file.disk_state().exists()),
2070 )
2071 });
2072
2073 if !existed_on_disk {
2074 current_exists || current_snapshot.text() != baseline_snapshot.text()
2075 } else {
2076 current_snapshot.text() != baseline_snapshot.text()
2077 }
2078 }
2079
2080 fn infer_inferred_edit_candidate_from_baseline(
2081 &mut self,
2082 buffer: Entity<Buffer>,
2083 baseline_snapshot: text::BufferSnapshot,
2084 existed_on_disk: bool,
2085 cx: &mut Context<Self>,
2086 ) {
2087 if existed_on_disk {
2088 self.action_log.update(cx, |action_log, cx| {
2089 action_log.infer_buffer_edited_from_snapshot(buffer.clone(), baseline_snapshot, cx);
2090 });
2091 } else {
2092 self.action_log.update(cx, |action_log, cx| {
2093 action_log.infer_buffer_created(buffer.clone(), baseline_snapshot, cx);
2094 });
2095 }
2096 }
2097
2098 fn handle_inferred_edit_candidate_buffer_event(
2099 &mut self,
2100 tool_call_id: acp::ToolCallId,
2101 abs_path: PathBuf,
2102 nonce: u64,
2103 buffer: Entity<Buffer>,
2104 event: &BufferEvent,
2105 cx: &mut Context<Self>,
2106 ) {
2107 let Some(InferredEditCandidateState::Ready(candidate)) = self
2108 .inferred_edit_candidates
2109 .get(&tool_call_id)
2110 .and_then(|candidates| candidates.get(&abs_path))
2111 else {
2112 return;
2113 };
2114 if candidate.nonce != nonce {
2115 return;
2116 }
2117
2118 let baseline_snapshot = candidate.baseline_snapshot.clone();
2119 let existed_on_disk = candidate.existed_on_disk;
2120 let was_dirty = candidate.was_dirty;
2121 let observed_external_file_change = candidate.observed_external_file_change;
2122 let has_inferred_edit = candidate.has_inferred_edit;
2123
2124 enum Action {
2125 None,
2126 Remove,
2127 SetObservedExternalFileChange,
2128 ClearObservedExternalFileChange,
2129 ClearObservedAndTrackEdited,
2130 ClearObservedAndInferCreatedOrEdited,
2131 RemoveAndWillDeleteBuffer,
2132 RemoveAndInferDeleted,
2133 }
2134
2135 let action = match event {
2136 BufferEvent::Edited { is_local: true }
2137 if !has_inferred_edit && !observed_external_file_change =>
2138 {
2139 Action::Remove
2140 }
2141 BufferEvent::Saved if !has_inferred_edit => Action::Remove,
2142 BufferEvent::FileHandleChanged => {
2143 let is_deleted = buffer.read_with(cx, |buffer, _| {
2144 buffer
2145 .file()
2146 .is_some_and(|file| file.disk_state().is_deleted())
2147 });
2148
2149 if is_deleted {
2150 if has_inferred_edit {
2151 Action::RemoveAndWillDeleteBuffer
2152 } else if self
2153 .can_infer_initial_external_edit_for_candidate(&buffer, was_dirty, cx)
2154 {
2155 Action::RemoveAndInferDeleted
2156 } else {
2157 Action::Remove
2158 }
2159 } else {
2160 Action::SetObservedExternalFileChange
2161 }
2162 }
2163 BufferEvent::Reloaded if observed_external_file_change => {
2164 if has_inferred_edit {
2165 Action::ClearObservedAndTrackEdited
2166 } else if self.can_infer_initial_external_edit_for_candidate(&buffer, was_dirty, cx)
2167 && Self::inferred_edit_candidate_has_meaningful_change(
2168 existed_on_disk,
2169 &baseline_snapshot,
2170 &buffer,
2171 cx,
2172 )
2173 {
2174 Action::ClearObservedAndInferCreatedOrEdited
2175 } else {
2176 Action::ClearObservedExternalFileChange
2177 }
2178 }
2179 _ => Action::None,
2180 };
2181
2182 match action {
2183 Action::None => {}
2184 Action::Remove => {
2185 self.remove_inferred_edit_candidate_if_matching(&tool_call_id, &abs_path, nonce);
2186 }
2187 Action::SetObservedExternalFileChange => {
2188 self.update_ready_inferred_edit_candidate_if_matching(
2189 &tool_call_id,
2190 &abs_path,
2191 nonce,
2192 |candidate| {
2193 candidate.observed_external_file_change = true;
2194 },
2195 );
2196 }
2197 Action::ClearObservedExternalFileChange => {
2198 self.update_ready_inferred_edit_candidate_if_matching(
2199 &tool_call_id,
2200 &abs_path,
2201 nonce,
2202 |candidate| {
2203 candidate.observed_external_file_change = false;
2204 },
2205 );
2206 }
2207 Action::ClearObservedAndTrackEdited => {
2208 let updated = self.update_ready_inferred_edit_candidate_if_matching(
2209 &tool_call_id,
2210 &abs_path,
2211 nonce,
2212 |candidate| {
2213 candidate.observed_external_file_change = false;
2214 },
2215 );
2216 if updated {
2217 self.action_log.update(cx, |action_log, cx| {
2218 action_log.buffer_edited(buffer.clone(), cx);
2219 });
2220 }
2221 }
2222 Action::ClearObservedAndInferCreatedOrEdited => {
2223 let updated = self.update_ready_inferred_edit_candidate_if_matching(
2224 &tool_call_id,
2225 &abs_path,
2226 nonce,
2227 |candidate| {
2228 candidate.observed_external_file_change = false;
2229 candidate.has_inferred_edit = true;
2230 },
2231 );
2232 if updated {
2233 self.infer_inferred_edit_candidate_from_baseline(
2234 buffer,
2235 baseline_snapshot,
2236 existed_on_disk,
2237 cx,
2238 );
2239 }
2240 }
2241 Action::RemoveAndWillDeleteBuffer => {
2242 self.remove_inferred_edit_candidate_if_matching(&tool_call_id, &abs_path, nonce);
2243 self.action_log.update(cx, |action_log, cx| {
2244 action_log.will_delete_buffer(buffer.clone(), cx);
2245 });
2246 }
2247 Action::RemoveAndInferDeleted => {
2248 self.remove_inferred_edit_candidate_if_matching(&tool_call_id, &abs_path, nonce);
2249 self.action_log.update(cx, |action_log, cx| {
2250 action_log.infer_buffer_deleted_from_snapshot(
2251 buffer.clone(),
2252 baseline_snapshot,
2253 cx,
2254 );
2255 });
2256 }
2257 }
2258 }
2259
2260 fn finalize_inferred_edit_tool_call(
2261 &mut self,
2262 tool_call_id: acp::ToolCallId,
2263 cx: &mut Context<Self>,
2264 ) {
2265 let should_finalize = self.tool_call(&tool_call_id).is_some_and(|(_, tool_call)| {
2266 Self::should_infer_external_edits(tool_call)
2267 && Self::is_inferred_edit_terminal_status(&tool_call.status)
2268 });
2269 if !should_finalize {
2270 self.clear_inferred_edit_candidates_for_tool_calls([tool_call_id]);
2271 return;
2272 }
2273
2274 if !self
2275 .finalizing_inferred_edit_tool_calls
2276 .insert(tool_call_id.clone())
2277 {
2278 return;
2279 }
2280
2281 let project = self.project.clone();
2282 cx.spawn::<_, anyhow::Result<()>>(async move |this, cx| {
2283 const MAX_ATTEMPTS: usize = 3;
2284 const ATTEMPT_DELAY: Duration = Duration::from_millis(50);
2285
2286 for attempt in 0..MAX_ATTEMPTS {
2287 let (buffers_to_reload, has_pending, has_observed_external_file_change) = this
2288 .read_with(cx, |this, _| {
2289 let Some(candidates) = this.inferred_edit_candidates.get(&tool_call_id)
2290 else {
2291 return (Vec::new(), false, false);
2292 };
2293
2294 let mut buffers_to_reload = HashSet::default();
2295 let mut has_pending = false;
2296 let mut has_observed_external_file_change = false;
2297
2298 for candidate_state in candidates.values() {
2299 match candidate_state {
2300 InferredEditCandidateState::Pending { .. } => has_pending = true,
2301 InferredEditCandidateState::Ready(candidate) => {
2302 if candidate.observed_external_file_change {
2303 has_observed_external_file_change = true;
2304 buffers_to_reload.insert(candidate.buffer.clone());
2305 }
2306 }
2307 }
2308 }
2309
2310 (
2311 buffers_to_reload.into_iter().collect::<Vec<_>>(),
2312 has_pending,
2313 has_observed_external_file_change,
2314 )
2315 })
2316 .unwrap_or((Vec::new(), false, false));
2317
2318 if buffers_to_reload.is_empty() && !has_pending {
2319 break;
2320 }
2321
2322 for buffer in buffers_to_reload {
2323 let should_reload = buffer.read_with(cx, |buffer, _| !buffer.is_dirty());
2324 if !should_reload {
2325 continue;
2326 }
2327
2328 let reload = project.update(cx, |project, cx| {
2329 let mut buffers = HashSet::default();
2330 buffers.insert(buffer.clone());
2331 project.reload_buffers(buffers, false, cx)
2332 });
2333 reload.await.log_err();
2334 }
2335
2336 if (!has_pending && !has_observed_external_file_change)
2337 || attempt + 1 == MAX_ATTEMPTS
2338 {
2339 break;
2340 }
2341
2342 cx.background_executor().timer(ATTEMPT_DELAY).await;
2343 }
2344
2345 this.update(cx, |this, _| {
2346 this.clear_inferred_edit_candidates_for_tool_calls([tool_call_id.clone()]);
2347 })
2348 .ok();
2349
2350 Ok(())
2351 })
2352 .detach_and_log_err(cx);
2353 }
2354
2355 fn refresh_inferred_edit_tool_call(
2356 &mut self,
2357 tool_call_id: acp::ToolCallId,
2358 cx: &mut Context<Self>,
2359 ) {
2360 let Some((_, tool_call)) = self.tool_call(&tool_call_id) else {
2361 return;
2362 };
2363
2364 let should_track = Self::should_infer_external_edits(tool_call);
2365 let should_finalize = Self::is_inferred_edit_terminal_status(&tool_call.status);
2366 let locations = tool_call.locations.clone();
2367
2368 if !should_track {
2369 self.clear_inferred_edit_candidates_for_tool_calls([tool_call_id]);
2370 return;
2371 }
2372
2373 self.register_inferred_edit_locations(tool_call_id.clone(), &locations, cx);
2374
2375 if should_finalize {
2376 self.finalize_inferred_edit_tool_call(tool_call_id, cx);
2377 }
2378 }
2379
2380 pub fn update_tool_call(
2381 &mut self,
2382 update: impl Into<ToolCallUpdate>,
2383 cx: &mut Context<Self>,
2384 ) -> Result<()> {
2385 let update = update.into();
2386 let tool_call_id = update.id().clone();
2387 let languages = self.project.read(cx).languages().clone();
2388 let path_style = self.project.read(cx).path_style(cx);
2389
2390 let ix = match self.index_for_tool_call(&tool_call_id) {
2391 Some(ix) => ix,
2392 None => {
2393 // Tool call not found - create a failed tool call entry
2394 let failed_tool_call = ToolCall {
2395 id: tool_call_id,
2396 label: cx.new(|cx| Markdown::new("Tool call not found".into(), None, None, cx)),
2397 kind: acp::ToolKind::Fetch,
2398 content: vec![ToolCallContent::ContentBlock(ContentBlock::new(
2399 "Tool call not found".into(),
2400 &languages,
2401 path_style,
2402 cx,
2403 ))],
2404 status: ToolCallStatus::Failed,
2405 locations: Vec::new(),
2406 resolved_locations: Vec::new(),
2407 raw_input: None,
2408 raw_input_markdown: None,
2409 raw_output: None,
2410 tool_name: None,
2411 subagent_session_info: None,
2412 };
2413 self.push_entry(AgentThreadEntry::ToolCall(failed_tool_call), cx);
2414 return Ok(());
2415 }
2416 };
2417 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
2418 unreachable!()
2419 };
2420
2421 match update {
2422 ToolCallUpdate::UpdateFields(update) => {
2423 let location_updated = update.fields.locations.is_some();
2424 call.update_fields(
2425 update.fields,
2426 update.meta,
2427 languages,
2428 path_style,
2429 &self.terminals,
2430 cx,
2431 )?;
2432 if location_updated {
2433 self.resolve_locations(update.tool_call_id, cx);
2434 }
2435 }
2436 ToolCallUpdate::UpdateDiff(update) => {
2437 call.content.clear();
2438 call.content.push(ToolCallContent::Diff(update.diff));
2439 }
2440 ToolCallUpdate::UpdateTerminal(update) => {
2441 call.content.clear();
2442 call.content
2443 .push(ToolCallContent::Terminal(update.terminal));
2444 }
2445 }
2446
2447 cx.emit(AcpThreadEvent::EntryUpdated(ix));
2448 self.refresh_inferred_edit_tool_call(tool_call_id, cx);
2449
2450 Ok(())
2451 }
2452
2453 /// Updates a tool call if id matches an existing entry, otherwise inserts a new one.
2454 pub fn upsert_tool_call(
2455 &mut self,
2456 tool_call: acp::ToolCall,
2457 cx: &mut Context<Self>,
2458 ) -> Result<(), acp::Error> {
2459 let status = tool_call.status.into();
2460 self.upsert_tool_call_inner(tool_call.into(), status, cx)
2461 }
2462
2463 /// Fails if id does not match an existing entry.
2464 pub fn upsert_tool_call_inner(
2465 &mut self,
2466 update: acp::ToolCallUpdate,
2467 status: ToolCallStatus,
2468 cx: &mut Context<Self>,
2469 ) -> Result<(), acp::Error> {
2470 let language_registry = self.project.read(cx).languages().clone();
2471 let path_style = self.project.read(cx).path_style(cx);
2472 let id = update.tool_call_id.clone();
2473
2474 let agent_telemetry_id = self.connection().telemetry_id();
2475 let session = self.session_id();
2476 let parent_session_id = self.parent_session_id();
2477 if let ToolCallStatus::Completed | ToolCallStatus::Failed = status {
2478 let status = if matches!(status, ToolCallStatus::Completed) {
2479 "completed"
2480 } else {
2481 "failed"
2482 };
2483 telemetry::event!(
2484 "Agent Tool Call Completed",
2485 agent_telemetry_id,
2486 session,
2487 parent_session_id,
2488 status
2489 );
2490 }
2491
2492 if let Some(ix) = self.index_for_tool_call(&id) {
2493 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
2494 unreachable!()
2495 };
2496
2497 call.update_fields(
2498 update.fields,
2499 update.meta,
2500 language_registry,
2501 path_style,
2502 &self.terminals,
2503 cx,
2504 )?;
2505 call.status = status;
2506
2507 cx.emit(AcpThreadEvent::EntryUpdated(ix));
2508 } else {
2509 let call = ToolCall::from_acp(
2510 update.try_into()?,
2511 status,
2512 language_registry,
2513 self.project.read(cx).path_style(cx),
2514 &self.terminals,
2515 cx,
2516 )?;
2517 self.push_entry(AgentThreadEntry::ToolCall(call), cx);
2518 };
2519
2520 self.resolve_locations(id.clone(), cx);
2521 self.refresh_inferred_edit_tool_call(id, cx);
2522 Ok(())
2523 }
2524
2525 fn index_for_tool_call(&self, id: &acp::ToolCallId) -> Option<usize> {
2526 self.entries
2527 .iter()
2528 .enumerate()
2529 .rev()
2530 .find_map(|(index, entry)| {
2531 if let AgentThreadEntry::ToolCall(tool_call) = entry
2532 && &tool_call.id == id
2533 {
2534 Some(index)
2535 } else {
2536 None
2537 }
2538 })
2539 }
2540
2541 fn tool_call_mut(&mut self, id: &acp::ToolCallId) -> Option<(usize, &mut ToolCall)> {
2542 // The tool call we are looking for is typically the last one, or very close to the end.
2543 // At the moment, it doesn't seem like a hashmap would be a good fit for this use case.
2544 self.entries
2545 .iter_mut()
2546 .enumerate()
2547 .rev()
2548 .find_map(|(index, tool_call)| {
2549 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
2550 && &tool_call.id == id
2551 {
2552 Some((index, tool_call))
2553 } else {
2554 None
2555 }
2556 })
2557 }
2558
2559 pub fn tool_call(&self, id: &acp::ToolCallId) -> Option<(usize, &ToolCall)> {
2560 self.entries
2561 .iter()
2562 .enumerate()
2563 .rev()
2564 .find_map(|(index, tool_call)| {
2565 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
2566 && &tool_call.id == id
2567 {
2568 Some((index, tool_call))
2569 } else {
2570 None
2571 }
2572 })
2573 }
2574
2575 pub fn tool_call_for_subagent(&self, session_id: &acp::SessionId) -> Option<&ToolCall> {
2576 self.entries.iter().find_map(|entry| match entry {
2577 AgentThreadEntry::ToolCall(tool_call) => {
2578 if let Some(subagent_session_info) = &tool_call.subagent_session_info
2579 && &subagent_session_info.session_id == session_id
2580 {
2581 Some(tool_call)
2582 } else {
2583 None
2584 }
2585 }
2586 _ => None,
2587 })
2588 }
2589
2590 pub fn resolve_locations(&mut self, id: acp::ToolCallId, cx: &mut Context<Self>) {
2591 let project = self.project.clone();
2592 let should_update_agent_location = self.parent_session_id.is_none();
2593 let Some((_, tool_call)) = self.tool_call_mut(&id) else {
2594 return;
2595 };
2596 let task = tool_call.resolve_locations(project, cx);
2597 cx.spawn(async move |this, cx| {
2598 let resolved_locations = task.await;
2599
2600 this.update(cx, |this, cx| {
2601 let project = this.project.clone();
2602
2603 for location in resolved_locations.iter().flatten() {
2604 this.shared_buffers
2605 .insert(location.buffer.clone(), location.buffer.read(cx).snapshot());
2606 }
2607 let Some((ix, tool_call)) = this.tool_call_mut(&id) else {
2608 return;
2609 };
2610
2611 if let Some(Some(location)) = resolved_locations.last() {
2612 project.update(cx, |project, cx| {
2613 let should_ignore = if let Some(agent_location) = project
2614 .agent_location()
2615 .filter(|agent_location| agent_location.buffer == location.buffer)
2616 {
2617 let snapshot = location.buffer.read(cx).snapshot();
2618 let old_position = agent_location.position.to_point(&snapshot);
2619 let new_position = location.position.to_point(&snapshot);
2620
2621 // ignore this so that when we get updates from the edit tool
2622 // the position doesn't reset to the startof line
2623 old_position.row == new_position.row
2624 && old_position.column > new_position.column
2625 } else {
2626 false
2627 };
2628 if !should_ignore && should_update_agent_location {
2629 project.set_agent_location(Some(location.into()), cx);
2630 }
2631 });
2632 }
2633
2634 let resolved_locations = resolved_locations
2635 .iter()
2636 .map(|l| l.as_ref().map(|l| AgentLocation::from(l)))
2637 .collect::<Vec<_>>();
2638
2639 if tool_call.resolved_locations != resolved_locations {
2640 tool_call.resolved_locations = resolved_locations;
2641 cx.emit(AcpThreadEvent::EntryUpdated(ix));
2642 }
2643 })
2644 })
2645 .detach();
2646 }
2647
2648 pub fn request_tool_call_authorization(
2649 &mut self,
2650 tool_call: acp::ToolCallUpdate,
2651 options: PermissionOptions,
2652 cx: &mut Context<Self>,
2653 ) -> Result<Task<RequestPermissionOutcome>> {
2654 let (tx, rx) = oneshot::channel();
2655
2656 let status = ToolCallStatus::WaitingForConfirmation {
2657 options,
2658 respond_tx: tx,
2659 };
2660
2661 let tool_call_id = tool_call.tool_call_id.clone();
2662 self.upsert_tool_call_inner(tool_call, status, cx)?;
2663 cx.emit(AcpThreadEvent::ToolAuthorizationRequested(
2664 tool_call_id.clone(),
2665 ));
2666
2667 Ok(cx.spawn(async move |this, cx| {
2668 let outcome = match rx.await {
2669 Ok(outcome) => RequestPermissionOutcome::Selected(outcome),
2670 Err(oneshot::Canceled) => RequestPermissionOutcome::Cancelled,
2671 };
2672 this.update(cx, |_this, cx| {
2673 cx.emit(AcpThreadEvent::ToolAuthorizationReceived(tool_call_id))
2674 })
2675 .ok();
2676 outcome
2677 }))
2678 }
2679
2680 pub fn authorize_tool_call(
2681 &mut self,
2682 id: acp::ToolCallId,
2683 outcome: SelectedPermissionOutcome,
2684 option_kind: acp::PermissionOptionKind,
2685 cx: &mut Context<Self>,
2686 ) {
2687 let Some((ix, call)) = self.tool_call_mut(&id) else {
2688 return;
2689 };
2690
2691 let new_status = match option_kind {
2692 acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways => {
2693 ToolCallStatus::Rejected
2694 }
2695 acp::PermissionOptionKind::AllowOnce | acp::PermissionOptionKind::AllowAlways => {
2696 ToolCallStatus::InProgress
2697 }
2698 _ => ToolCallStatus::InProgress,
2699 };
2700
2701 let curr_status = mem::replace(&mut call.status, new_status);
2702
2703 if let ToolCallStatus::WaitingForConfirmation { respond_tx, .. } = curr_status {
2704 respond_tx.send(outcome).log_err();
2705 } else if cfg!(debug_assertions) {
2706 panic!("tried to authorize an already authorized tool call");
2707 }
2708
2709 cx.emit(AcpThreadEvent::EntryUpdated(ix));
2710 }
2711
2712 pub fn plan(&self) -> &Plan {
2713 &self.plan
2714 }
2715
2716 pub fn update_plan(&mut self, request: acp::Plan, cx: &mut Context<Self>) {
2717 let new_entries_len = request.entries.len();
2718 let mut new_entries = request.entries.into_iter();
2719
2720 // Reuse existing markdown to prevent flickering
2721 for (old, new) in self.plan.entries.iter_mut().zip(new_entries.by_ref()) {
2722 let PlanEntry {
2723 content,
2724 priority,
2725 status,
2726 } = old;
2727 content.update(cx, |old, cx| {
2728 old.replace(new.content, cx);
2729 });
2730 *priority = new.priority;
2731 *status = new.status;
2732 }
2733 for new in new_entries {
2734 self.plan.entries.push(PlanEntry::from_acp(new, cx))
2735 }
2736 self.plan.entries.truncate(new_entries_len);
2737
2738 cx.notify();
2739 }
2740
2741 fn clear_completed_plan_entries(&mut self, cx: &mut Context<Self>) {
2742 self.plan
2743 .entries
2744 .retain(|entry| !matches!(entry.status, acp::PlanEntryStatus::Completed));
2745 cx.notify();
2746 }
2747
2748 #[cfg(any(test, feature = "test-support"))]
2749 pub fn send_raw(
2750 &mut self,
2751 message: &str,
2752 cx: &mut Context<Self>,
2753 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2754 self.send(vec![message.into()], cx)
2755 }
2756
2757 pub fn send(
2758 &mut self,
2759 message: Vec<acp::ContentBlock>,
2760 cx: &mut Context<Self>,
2761 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2762 let block = ContentBlock::new_combined(
2763 message.clone(),
2764 self.project.read(cx).languages().clone(),
2765 self.project.read(cx).path_style(cx),
2766 cx,
2767 );
2768 let request = acp::PromptRequest::new(self.session_id.clone(), message.clone());
2769 let git_store = self.project.read(cx).git_store().clone();
2770
2771 let message_id = if self.connection.truncate(&self.session_id, cx).is_some() {
2772 Some(UserMessageId::new())
2773 } else {
2774 None
2775 };
2776
2777 self.run_turn(cx, async move |this, cx| {
2778 this.update(cx, |this, cx| {
2779 this.push_entry(
2780 AgentThreadEntry::UserMessage(UserMessage {
2781 id: message_id.clone(),
2782 content: block,
2783 chunks: message,
2784 checkpoint: None,
2785 indented: false,
2786 }),
2787 cx,
2788 );
2789 })
2790 .ok();
2791
2792 let old_checkpoint = git_store
2793 .update(cx, |git, cx| git.checkpoint(cx))
2794 .await
2795 .context("failed to get old checkpoint")
2796 .log_err();
2797 this.update(cx, |this, cx| {
2798 if let Some((_ix, message)) = this.last_user_message() {
2799 message.checkpoint = old_checkpoint.map(|git_checkpoint| Checkpoint {
2800 git_checkpoint,
2801 show: false,
2802 });
2803 }
2804 this.connection.prompt(message_id, request, cx)
2805 })?
2806 .await
2807 })
2808 }
2809
2810 pub fn can_retry(&self, cx: &App) -> bool {
2811 self.connection.retry(&self.session_id, cx).is_some()
2812 }
2813
2814 pub fn retry(
2815 &mut self,
2816 cx: &mut Context<Self>,
2817 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2818 self.run_turn(cx, async move |this, cx| {
2819 this.update(cx, |this, cx| {
2820 this.connection
2821 .retry(&this.session_id, cx)
2822 .map(|retry| retry.run(cx))
2823 })?
2824 .context("retrying a session is not supported")?
2825 .await
2826 })
2827 }
2828
2829 fn run_turn(
2830 &mut self,
2831 cx: &mut Context<Self>,
2832 f: impl 'static + AsyncFnOnce(WeakEntity<Self>, &mut AsyncApp) -> Result<acp::PromptResponse>,
2833 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2834 self.clear_completed_plan_entries(cx);
2835 self.had_error = false;
2836
2837 let (tx, rx) = oneshot::channel();
2838 let cancel_task = self.cancel(cx);
2839
2840 self.turn_id += 1;
2841 let turn_id = self.turn_id;
2842 self.running_turn = Some(RunningTurn {
2843 id: turn_id,
2844 send_task: cx.spawn(async move |this, cx| {
2845 cancel_task.await;
2846 tx.send(f(this, cx).await).ok();
2847 }),
2848 });
2849
2850 cx.spawn(async move |this, cx| {
2851 let response = rx.await;
2852
2853 this.update(cx, |this, cx| this.update_last_checkpoint(cx))?
2854 .await?;
2855
2856 this.update(cx, |this, cx| {
2857 if this.parent_session_id.is_none() {
2858 this.project
2859 .update(cx, |project, cx| project.set_agent_location(None, cx));
2860 }
2861 let Ok(response) = response else {
2862 // tx dropped, just return
2863 return Ok(None);
2864 };
2865
2866 let is_same_turn = this
2867 .running_turn
2868 .as_ref()
2869 .is_some_and(|turn| turn_id == turn.id);
2870
2871 // If the user submitted a follow up message, running_turn might
2872 // already point to a different turn. Therefore we only want to
2873 // take the task if it's the same turn.
2874 if is_same_turn {
2875 this.running_turn.take();
2876 }
2877
2878 match response {
2879 Ok(r) => {
2880 Self::flush_streaming_text(&mut this.streaming_text_buffer, cx);
2881
2882 if r.stop_reason == acp::StopReason::MaxTokens {
2883 this.finalize_all_inferred_edit_tool_calls(cx);
2884 this.had_error = true;
2885 cx.emit(AcpThreadEvent::Error);
2886 log::error!("Max tokens reached. Usage: {:?}", this.token_usage);
2887 return Err(anyhow!("Max tokens reached"));
2888 }
2889
2890 let canceled = matches!(r.stop_reason, acp::StopReason::Cancelled);
2891 if canceled {
2892 let canceled_tool_call_ids = this.mark_pending_tools_as_canceled();
2893 this.clear_inferred_edit_candidates_for_tool_calls(
2894 canceled_tool_call_ids,
2895 );
2896 this.finalize_all_inferred_edit_tool_calls(cx);
2897 }
2898
2899 // Handle refusal - distinguish between user prompt and tool call refusals
2900 if let acp::StopReason::Refusal = r.stop_reason {
2901 this.had_error = true;
2902 if let Some((user_msg_ix, _)) = this.last_user_message() {
2903 // Check if there's a completed tool call with results after the last user message
2904 // This indicates the refusal is in response to tool output, not the user's prompt
2905 let has_completed_tool_call_after_user_msg =
2906 this.entries.iter().skip(user_msg_ix + 1).any(|entry| {
2907 if let AgentThreadEntry::ToolCall(tool_call) = entry {
2908 // Check if the tool call has completed and has output
2909 matches!(tool_call.status, ToolCallStatus::Completed)
2910 && tool_call.raw_output.is_some()
2911 } else {
2912 false
2913 }
2914 });
2915
2916 if has_completed_tool_call_after_user_msg {
2917 // Refusal is due to tool output - don't truncate, just notify
2918 // The model refused based on what the tool returned
2919 cx.emit(AcpThreadEvent::Refusal);
2920 } else {
2921 // User prompt was refused - truncate back to before the user message
2922 let range = user_msg_ix..this.entries.len();
2923 if range.start < range.end {
2924 let removed_tool_call_ids = this.entries[user_msg_ix..]
2925 .iter()
2926 .filter_map(|entry| {
2927 if let AgentThreadEntry::ToolCall(tool_call) = entry
2928 {
2929 Some(tool_call.id.clone())
2930 } else {
2931 None
2932 }
2933 })
2934 .collect::<Vec<_>>();
2935 this.clear_inferred_edit_candidates_for_tool_calls(
2936 removed_tool_call_ids,
2937 );
2938 this.entries.truncate(user_msg_ix);
2939 cx.emit(AcpThreadEvent::EntriesRemoved(range));
2940 }
2941 cx.emit(AcpThreadEvent::Refusal);
2942 }
2943 } else {
2944 // No user message found, treat as general refusal
2945 cx.emit(AcpThreadEvent::Refusal);
2946 }
2947 }
2948
2949 this.finalize_all_inferred_edit_tool_calls(cx);
2950 cx.emit(AcpThreadEvent::Stopped(r.stop_reason));
2951 Ok(Some(r))
2952 }
2953 Err(e) => {
2954 Self::flush_streaming_text(&mut this.streaming_text_buffer, cx);
2955
2956 this.finalize_all_inferred_edit_tool_calls(cx);
2957 this.had_error = true;
2958 cx.emit(AcpThreadEvent::Error);
2959 log::error!("Error in run turn: {:?}", e);
2960 Err(e)
2961 }
2962 }
2963 })?
2964 })
2965 .boxed()
2966 }
2967
2968 pub fn cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
2969 let Some(turn) = self.running_turn.take() else {
2970 return Task::ready(());
2971 };
2972 self.connection.cancel(&self.session_id, cx);
2973
2974 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
2975 let canceled_tool_call_ids = self.mark_pending_tools_as_canceled();
2976 self.clear_inferred_edit_candidates_for_tool_calls(canceled_tool_call_ids);
2977 self.finalize_all_inferred_edit_tool_calls(cx);
2978
2979 cx.background_spawn(turn.send_task)
2980 }
2981
2982 fn mark_pending_tools_as_canceled(&mut self) -> Vec<acp::ToolCallId> {
2983 let mut canceled_tool_call_ids = Vec::new();
2984
2985 for entry in self.entries.iter_mut() {
2986 if let AgentThreadEntry::ToolCall(call) = entry {
2987 let cancel = matches!(
2988 call.status,
2989 ToolCallStatus::Pending
2990 | ToolCallStatus::WaitingForConfirmation { .. }
2991 | ToolCallStatus::InProgress
2992 );
2993
2994 if cancel {
2995 call.status = ToolCallStatus::Canceled;
2996 canceled_tool_call_ids.push(call.id.clone());
2997 }
2998 }
2999 }
3000
3001 canceled_tool_call_ids
3002 }
3003
3004 /// Restores the git working tree to the state at the given checkpoint (if one exists)
3005 pub fn restore_checkpoint(
3006 &mut self,
3007 id: UserMessageId,
3008 cx: &mut Context<Self>,
3009 ) -> Task<Result<()>> {
3010 let Some((_, message)) = self.user_message_mut(&id) else {
3011 return Task::ready(Err(anyhow!("message not found")));
3012 };
3013
3014 let checkpoint = message
3015 .checkpoint
3016 .as_ref()
3017 .map(|c| c.git_checkpoint.clone());
3018
3019 // Cancel any in-progress generation before restoring
3020 let cancel_task = self.cancel(cx);
3021 let rewind = self.rewind(id.clone(), cx);
3022 let git_store = self.project.read(cx).git_store().clone();
3023
3024 cx.spawn(async move |_, cx| {
3025 cancel_task.await;
3026 rewind.await?;
3027 if let Some(checkpoint) = checkpoint {
3028 git_store
3029 .update(cx, |git, cx| git.restore_checkpoint(checkpoint, cx))
3030 .await?;
3031 }
3032
3033 Ok(())
3034 })
3035 }
3036
3037 /// Rewinds this thread to before the entry at `index`, removing it and all
3038 /// subsequent entries while rejecting any action_log changes made from that point.
3039 /// Unlike `restore_checkpoint`, this method does not restore from git.
3040 pub fn rewind(&mut self, id: UserMessageId, cx: &mut Context<Self>) -> Task<Result<()>> {
3041 let Some(truncate) = self.connection.truncate(&self.session_id, cx) else {
3042 return Task::ready(Err(anyhow!("not supported")));
3043 };
3044
3045 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
3046 let telemetry = ActionLogTelemetry::from(&*self);
3047 cx.spawn(async move |this, cx| {
3048 cx.update(|cx| truncate.run(id.clone(), cx)).await?;
3049 this.update(cx, |this, cx| {
3050 if let Some((ix, _)) = this.user_message_mut(&id) {
3051 // Collect all terminals from entries that will be removed
3052 let terminals_to_remove: Vec<acp::TerminalId> = this.entries[ix..]
3053 .iter()
3054 .flat_map(|entry| entry.terminals())
3055 .filter_map(|terminal| terminal.read(cx).id().clone().into())
3056 .collect();
3057 let removed_tool_call_ids = this.entries[ix..]
3058 .iter()
3059 .filter_map(|entry| {
3060 if let AgentThreadEntry::ToolCall(tool_call) = entry {
3061 Some(tool_call.id.clone())
3062 } else {
3063 None
3064 }
3065 })
3066 .collect::<Vec<_>>();
3067
3068 let range = ix..this.entries.len();
3069 this.clear_inferred_edit_candidates_for_tool_calls(removed_tool_call_ids);
3070 this.entries.truncate(ix);
3071 cx.emit(AcpThreadEvent::EntriesRemoved(range));
3072
3073 // Kill and remove the terminals
3074 for terminal_id in terminals_to_remove {
3075 if let Some(terminal) = this.terminals.remove(&terminal_id) {
3076 terminal.update(cx, |terminal, cx| {
3077 terminal.kill(cx);
3078 });
3079 }
3080 }
3081 }
3082 this.action_log().update(cx, |action_log, cx| {
3083 action_log.reject_all_edits(Some(telemetry), cx)
3084 })
3085 })?
3086 .await;
3087 Ok(())
3088 })
3089 }
3090
3091 fn update_last_checkpoint(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
3092 let git_store = self.project.read(cx).git_store().clone();
3093
3094 let Some((_, message)) = self.last_user_message() else {
3095 return Task::ready(Ok(()));
3096 };
3097 let Some(user_message_id) = message.id.clone() else {
3098 return Task::ready(Ok(()));
3099 };
3100 let Some(checkpoint) = message.checkpoint.as_ref() else {
3101 return Task::ready(Ok(()));
3102 };
3103 let old_checkpoint = checkpoint.git_checkpoint.clone();
3104
3105 let new_checkpoint = git_store.update(cx, |git, cx| git.checkpoint(cx));
3106 cx.spawn(async move |this, cx| {
3107 let Some(new_checkpoint) = new_checkpoint
3108 .await
3109 .context("failed to get new checkpoint")
3110 .log_err()
3111 else {
3112 return Ok(());
3113 };
3114
3115 let equal = git_store
3116 .update(cx, |git, cx| {
3117 git.compare_checkpoints(old_checkpoint.clone(), new_checkpoint, cx)
3118 })
3119 .await
3120 .unwrap_or(true);
3121
3122 this.update(cx, |this, cx| {
3123 if let Some((ix, message)) = this.user_message_mut(&user_message_id) {
3124 if let Some(checkpoint) = message.checkpoint.as_mut() {
3125 checkpoint.show = !equal;
3126 cx.emit(AcpThreadEvent::EntryUpdated(ix));
3127 }
3128 }
3129 })?;
3130
3131 Ok(())
3132 })
3133 }
3134
3135 fn last_user_message(&mut self) -> Option<(usize, &mut UserMessage)> {
3136 self.entries
3137 .iter_mut()
3138 .enumerate()
3139 .rev()
3140 .find_map(|(ix, entry)| {
3141 if let AgentThreadEntry::UserMessage(message) = entry {
3142 Some((ix, message))
3143 } else {
3144 None
3145 }
3146 })
3147 }
3148
3149 fn user_message_mut(&mut self, id: &UserMessageId) -> Option<(usize, &mut UserMessage)> {
3150 self.entries.iter_mut().enumerate().find_map(|(ix, entry)| {
3151 if let AgentThreadEntry::UserMessage(message) = entry {
3152 if message.id.as_ref() == Some(id) {
3153 Some((ix, message))
3154 } else {
3155 None
3156 }
3157 } else {
3158 None
3159 }
3160 })
3161 }
3162
3163 pub fn read_text_file(
3164 &self,
3165 path: PathBuf,
3166 line: Option<u32>,
3167 limit: Option<u32>,
3168 reuse_shared_snapshot: bool,
3169 cx: &mut Context<Self>,
3170 ) -> Task<Result<String, acp::Error>> {
3171 // Args are 1-based, move to 0-based
3172 let line = line.unwrap_or_default().saturating_sub(1);
3173 let limit = limit.unwrap_or(u32::MAX);
3174 let project = self.project.clone();
3175 let action_log = self.action_log.clone();
3176 let should_update_agent_location = self.parent_session_id.is_none();
3177 cx.spawn(async move |this, cx| {
3178 let load = project.update(cx, |project, cx| {
3179 let path = project
3180 .project_path_for_absolute_path(&path, cx)
3181 .ok_or_else(|| {
3182 acp::Error::resource_not_found(Some(path.display().to_string()))
3183 })?;
3184 Ok::<_, acp::Error>(project.open_buffer(path, cx))
3185 })?;
3186
3187 let buffer = load.await?;
3188
3189 let snapshot = if reuse_shared_snapshot {
3190 this.read_with(cx, |this, _| {
3191 this.shared_buffers.get(&buffer.clone()).cloned()
3192 })
3193 .log_err()
3194 .flatten()
3195 } else {
3196 None
3197 };
3198
3199 let snapshot = if let Some(snapshot) = snapshot {
3200 snapshot
3201 } else {
3202 action_log.update(cx, |action_log, cx| {
3203 action_log.buffer_read(buffer.clone(), cx);
3204 });
3205
3206 let snapshot = buffer.update(cx, |buffer, _| buffer.snapshot());
3207 this.update(cx, |this, _| {
3208 this.shared_buffers.insert(buffer.clone(), snapshot.clone());
3209 })?;
3210 snapshot
3211 };
3212
3213 let max_point = snapshot.max_point();
3214 let start_position = Point::new(line, 0);
3215
3216 if start_position > max_point {
3217 return Err(acp::Error::invalid_params().data(format!(
3218 "Attempting to read beyond the end of the file, line {}:{}",
3219 max_point.row + 1,
3220 max_point.column
3221 )));
3222 }
3223
3224 let start = snapshot.anchor_before(start_position);
3225 let end = snapshot.anchor_before(Point::new(line.saturating_add(limit), 0));
3226
3227 if should_update_agent_location {
3228 project.update(cx, |project, cx| {
3229 project.set_agent_location(
3230 Some(AgentLocation {
3231 buffer: buffer.downgrade(),
3232 position: start,
3233 }),
3234 cx,
3235 );
3236 });
3237 }
3238
3239 Ok(snapshot.text_for_range(start..end).collect::<String>())
3240 })
3241 }
3242
3243 pub fn write_text_file(
3244 &self,
3245 path: PathBuf,
3246 content: String,
3247 cx: &mut Context<Self>,
3248 ) -> Task<Result<()>> {
3249 let project = self.project.clone();
3250 let action_log = self.action_log.clone();
3251 let should_update_agent_location = self.parent_session_id.is_none();
3252 cx.spawn(async move |this, cx| {
3253 let load = project.update(cx, |project, cx| {
3254 let path = project
3255 .project_path_for_absolute_path(&path, cx)
3256 .context("invalid path")?;
3257 anyhow::Ok(project.open_buffer(path, cx))
3258 });
3259 let buffer = load?.await?;
3260 let snapshot = this.update(cx, |this, cx| {
3261 this.shared_buffers
3262 .get(&buffer)
3263 .cloned()
3264 .unwrap_or_else(|| buffer.read(cx).snapshot())
3265 })?;
3266 let edits = cx
3267 .background_executor()
3268 .spawn(async move {
3269 let old_text = snapshot.text();
3270 text_diff(old_text.as_str(), &content)
3271 .into_iter()
3272 .map(|(range, replacement)| {
3273 (snapshot.anchor_range_around(range), replacement)
3274 })
3275 .collect::<Vec<_>>()
3276 })
3277 .await;
3278
3279 if should_update_agent_location {
3280 project.update(cx, |project, cx| {
3281 project.set_agent_location(
3282 Some(AgentLocation {
3283 buffer: buffer.downgrade(),
3284 position: edits
3285 .last()
3286 .map(|(range, _)| range.end)
3287 .unwrap_or(Anchor::min_for_buffer(buffer.read(cx).remote_id())),
3288 }),
3289 cx,
3290 );
3291 });
3292 }
3293
3294 let format_on_save = cx.update(|cx| {
3295 action_log.update(cx, |action_log, cx| {
3296 action_log.buffer_read(buffer.clone(), cx);
3297 });
3298
3299 let format_on_save = buffer.update(cx, |buffer, cx| {
3300 buffer.edit(edits, None, cx);
3301
3302 let settings = language::language_settings::language_settings(
3303 buffer.language().map(|l| l.name()),
3304 buffer.file(),
3305 cx,
3306 );
3307
3308 settings.format_on_save != FormatOnSave::Off
3309 });
3310 action_log.update(cx, |action_log, cx| {
3311 action_log.buffer_edited(buffer.clone(), cx);
3312 });
3313 format_on_save
3314 });
3315
3316 if format_on_save {
3317 let format_task = project.update(cx, |project, cx| {
3318 project.format(
3319 HashSet::from_iter([buffer.clone()]),
3320 LspFormatTarget::Buffers,
3321 false,
3322 FormatTrigger::Save,
3323 cx,
3324 )
3325 });
3326 format_task.await.log_err();
3327
3328 action_log.update(cx, |action_log, cx| {
3329 action_log.buffer_edited(buffer.clone(), cx);
3330 });
3331 }
3332
3333 project
3334 .update(cx, |project, cx| project.save_buffer(buffer, cx))
3335 .await
3336 })
3337 }
3338
3339 pub fn create_terminal(
3340 &self,
3341 command: String,
3342 args: Vec<String>,
3343 extra_env: Vec<acp::EnvVariable>,
3344 cwd: Option<PathBuf>,
3345 output_byte_limit: Option<u64>,
3346 cx: &mut Context<Self>,
3347 ) -> Task<Result<Entity<Terminal>>> {
3348 let env = match &cwd {
3349 Some(dir) => self.project.update(cx, |project, cx| {
3350 project.environment().update(cx, |env, cx| {
3351 env.directory_environment(dir.as_path().into(), cx)
3352 })
3353 }),
3354 None => Task::ready(None).shared(),
3355 };
3356 let env = cx.spawn(async move |_, _| {
3357 let mut env = env.await.unwrap_or_default();
3358 // Disables paging for `git` and hopefully other commands
3359 env.insert("PAGER".into(), "".into());
3360 for var in extra_env {
3361 env.insert(var.name, var.value);
3362 }
3363 env
3364 });
3365
3366 let project = self.project.clone();
3367 let language_registry = project.read(cx).languages().clone();
3368 let is_windows = project.read(cx).path_style(cx).is_windows();
3369
3370 let terminal_id = acp::TerminalId::new(Uuid::new_v4().to_string());
3371 let terminal_task = cx.spawn({
3372 let terminal_id = terminal_id.clone();
3373 async move |_this, cx| {
3374 let env = env.await;
3375 let shell = project
3376 .update(cx, |project, cx| {
3377 project
3378 .remote_client()
3379 .and_then(|r| r.read(cx).default_system_shell())
3380 })
3381 .unwrap_or_else(|| get_default_system_shell_preferring_bash());
3382 let (task_command, task_args) =
3383 ShellBuilder::new(&Shell::Program(shell), is_windows)
3384 .redirect_stdin_to_dev_null()
3385 .build(Some(command.clone()), &args);
3386 let terminal = project
3387 .update(cx, |project, cx| {
3388 project.create_terminal_task(
3389 task::SpawnInTerminal {
3390 command: Some(task_command),
3391 args: task_args,
3392 cwd: cwd.clone(),
3393 env,
3394 ..Default::default()
3395 },
3396 cx,
3397 )
3398 })
3399 .await?;
3400
3401 anyhow::Ok(cx.new(|cx| {
3402 Terminal::new(
3403 terminal_id,
3404 &format!("{} {}", command, args.join(" ")),
3405 cwd,
3406 output_byte_limit.map(|l| l as usize),
3407 terminal,
3408 language_registry,
3409 cx,
3410 )
3411 }))
3412 }
3413 });
3414
3415 cx.spawn(async move |this, cx| {
3416 let terminal = terminal_task.await?;
3417 this.update(cx, |this, _cx| {
3418 this.terminals.insert(terminal_id, terminal.clone());
3419 terminal
3420 })
3421 })
3422 }
3423
3424 pub fn kill_terminal(
3425 &mut self,
3426 terminal_id: acp::TerminalId,
3427 cx: &mut Context<Self>,
3428 ) -> Result<()> {
3429 self.terminals
3430 .get(&terminal_id)
3431 .context("Terminal not found")?
3432 .update(cx, |terminal, cx| {
3433 terminal.kill(cx);
3434 });
3435
3436 Ok(())
3437 }
3438
3439 pub fn release_terminal(
3440 &mut self,
3441 terminal_id: acp::TerminalId,
3442 cx: &mut Context<Self>,
3443 ) -> Result<()> {
3444 self.terminals
3445 .remove(&terminal_id)
3446 .context("Terminal not found")?
3447 .update(cx, |terminal, cx| {
3448 terminal.kill(cx);
3449 });
3450
3451 Ok(())
3452 }
3453
3454 pub fn terminal(&self, terminal_id: acp::TerminalId) -> Result<Entity<Terminal>> {
3455 self.terminals
3456 .get(&terminal_id)
3457 .context("Terminal not found")
3458 .cloned()
3459 }
3460
3461 pub fn to_markdown(&self, cx: &App) -> String {
3462 self.entries.iter().map(|e| e.to_markdown(cx)).collect()
3463 }
3464
3465 pub fn emit_load_error(&mut self, error: LoadError, cx: &mut Context<Self>) {
3466 cx.emit(AcpThreadEvent::LoadError(error));
3467 }
3468
3469 pub fn register_terminal_created(
3470 &mut self,
3471 terminal_id: acp::TerminalId,
3472 command_label: String,
3473 working_dir: Option<PathBuf>,
3474 output_byte_limit: Option<u64>,
3475 terminal: Entity<::terminal::Terminal>,
3476 cx: &mut Context<Self>,
3477 ) -> Entity<Terminal> {
3478 let language_registry = self.project.read(cx).languages().clone();
3479
3480 let entity = cx.new(|cx| {
3481 Terminal::new(
3482 terminal_id.clone(),
3483 &command_label,
3484 working_dir.clone(),
3485 output_byte_limit.map(|l| l as usize),
3486 terminal,
3487 language_registry,
3488 cx,
3489 )
3490 });
3491 self.terminals.insert(terminal_id.clone(), entity.clone());
3492 entity
3493 }
3494
3495 pub fn mark_as_subagent_output(&mut self, cx: &mut Context<Self>) {
3496 for entry in self.entries.iter_mut().rev() {
3497 if let AgentThreadEntry::AssistantMessage(assistant_message) = entry {
3498 assistant_message.is_subagent_output = true;
3499 cx.notify();
3500 return;
3501 }
3502 }
3503 }
3504
3505 pub fn on_terminal_provider_event(
3506 &mut self,
3507 event: TerminalProviderEvent,
3508 cx: &mut Context<Self>,
3509 ) {
3510 match event {
3511 TerminalProviderEvent::Created {
3512 terminal_id,
3513 label,
3514 cwd,
3515 output_byte_limit,
3516 terminal,
3517 } => {
3518 let entity = self.register_terminal_created(
3519 terminal_id.clone(),
3520 label,
3521 cwd,
3522 output_byte_limit,
3523 terminal,
3524 cx,
3525 );
3526
3527 if let Some(mut chunks) = self.pending_terminal_output.remove(&terminal_id) {
3528 for data in chunks.drain(..) {
3529 entity.update(cx, |term, cx| {
3530 term.inner().update(cx, |inner, cx| {
3531 inner.write_output(&data, cx);
3532 })
3533 });
3534 }
3535 }
3536
3537 if let Some(_status) = self.pending_terminal_exit.remove(&terminal_id) {
3538 entity.update(cx, |_term, cx| {
3539 cx.notify();
3540 });
3541 }
3542
3543 cx.notify();
3544 }
3545 TerminalProviderEvent::Output { terminal_id, data } => {
3546 if let Some(entity) = self.terminals.get(&terminal_id) {
3547 entity.update(cx, |term, cx| {
3548 term.inner().update(cx, |inner, cx| {
3549 inner.write_output(&data, cx);
3550 })
3551 });
3552 } else {
3553 self.pending_terminal_output
3554 .entry(terminal_id)
3555 .or_default()
3556 .push(data);
3557 }
3558 }
3559 TerminalProviderEvent::TitleChanged { terminal_id, title } => {
3560 if let Some(entity) = self.terminals.get(&terminal_id) {
3561 entity.update(cx, |term, cx| {
3562 term.inner().update(cx, |inner, cx| {
3563 inner.breadcrumb_text = title;
3564 cx.emit(::terminal::Event::BreadcrumbsChanged);
3565 })
3566 });
3567 }
3568 }
3569 TerminalProviderEvent::Exit {
3570 terminal_id,
3571 status,
3572 } => {
3573 if let Some(entity) = self.terminals.get(&terminal_id) {
3574 entity.update(cx, |_term, cx| {
3575 cx.notify();
3576 });
3577 } else {
3578 self.pending_terminal_exit.insert(terminal_id, status);
3579 }
3580 }
3581 }
3582 }
3583}
3584
3585fn markdown_for_raw_output(
3586 raw_output: &serde_json::Value,
3587 language_registry: &Arc<LanguageRegistry>,
3588 cx: &mut App,
3589) -> Option<Entity<Markdown>> {
3590 match raw_output {
3591 serde_json::Value::Null => None,
3592 serde_json::Value::Bool(value) => Some(cx.new(|cx| {
3593 Markdown::new(
3594 value.to_string().into(),
3595 Some(language_registry.clone()),
3596 None,
3597 cx,
3598 )
3599 })),
3600 serde_json::Value::Number(value) => Some(cx.new(|cx| {
3601 Markdown::new(
3602 value.to_string().into(),
3603 Some(language_registry.clone()),
3604 None,
3605 cx,
3606 )
3607 })),
3608 serde_json::Value::String(value) => Some(cx.new(|cx| {
3609 Markdown::new(
3610 value.clone().into(),
3611 Some(language_registry.clone()),
3612 None,
3613 cx,
3614 )
3615 })),
3616 value => Some(cx.new(|cx| {
3617 let pretty_json = to_string_pretty(value).unwrap_or_else(|_| value.to_string());
3618
3619 Markdown::new(
3620 format!("```json\n{}\n```", pretty_json).into(),
3621 Some(language_registry.clone()),
3622 None,
3623 cx,
3624 )
3625 })),
3626 }
3627}
3628
3629#[cfg(test)]
3630mod tests {
3631 use super::*;
3632 use anyhow::anyhow;
3633 use futures::{channel::mpsc, future::LocalBoxFuture, select};
3634 use gpui::{App, AsyncApp, TestAppContext, WeakEntity};
3635 use indoc::indoc;
3636 use project::{AgentId, FakeFs, Fs};
3637 use rand::{distr, prelude::*};
3638 use serde_json::json;
3639 use settings::SettingsStore;
3640 use smol::stream::StreamExt as _;
3641 use std::{
3642 any::Any,
3643 cell::RefCell,
3644 path::Path,
3645 rc::Rc,
3646 sync::atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
3647 time::Duration,
3648 };
3649 use util::{path, path_list::PathList};
3650
3651 fn init_test(cx: &mut TestAppContext) {
3652 env_logger::try_init().ok();
3653 cx.update(|cx| {
3654 let settings_store = SettingsStore::test(cx);
3655 cx.set_global(settings_store);
3656 });
3657 }
3658
3659 #[gpui::test]
3660 async fn test_terminal_output_buffered_before_created_renders(cx: &mut gpui::TestAppContext) {
3661 init_test(cx);
3662
3663 let fs = FakeFs::new(cx.executor());
3664 let project = Project::test(fs, [], cx).await;
3665 let connection = Rc::new(FakeAgentConnection::new());
3666 let thread = cx
3667 .update(|cx| {
3668 connection.new_session(
3669 project,
3670 PathList::new(&[std::path::Path::new(path!("/test"))]),
3671 cx,
3672 )
3673 })
3674 .await
3675 .unwrap();
3676
3677 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
3678
3679 // Send Output BEFORE Created - should be buffered by acp_thread
3680 thread.update(cx, |thread, cx| {
3681 thread.on_terminal_provider_event(
3682 TerminalProviderEvent::Output {
3683 terminal_id: terminal_id.clone(),
3684 data: b"hello buffered".to_vec(),
3685 },
3686 cx,
3687 );
3688 });
3689
3690 // Create a display-only terminal and then send Created
3691 let lower = cx.new(|cx| {
3692 let builder = ::terminal::TerminalBuilder::new_display_only(
3693 ::terminal::terminal_settings::CursorShape::default(),
3694 ::terminal::terminal_settings::AlternateScroll::On,
3695 None,
3696 0,
3697 cx.background_executor(),
3698 PathStyle::local(),
3699 )
3700 .unwrap();
3701 builder.subscribe(cx)
3702 });
3703
3704 thread.update(cx, |thread, cx| {
3705 thread.on_terminal_provider_event(
3706 TerminalProviderEvent::Created {
3707 terminal_id: terminal_id.clone(),
3708 label: "Buffered Test".to_string(),
3709 cwd: None,
3710 output_byte_limit: None,
3711 terminal: lower.clone(),
3712 },
3713 cx,
3714 );
3715 });
3716
3717 // After Created, buffered Output should have been flushed into the renderer
3718 let content = thread.read_with(cx, |thread, cx| {
3719 let term = thread.terminal(terminal_id.clone()).unwrap();
3720 term.read_with(cx, |t, cx| t.inner().read(cx).get_content())
3721 });
3722
3723 assert!(
3724 content.contains("hello buffered"),
3725 "expected buffered output to render, got: {content}"
3726 );
3727 }
3728
3729 #[gpui::test]
3730 async fn test_terminal_output_and_exit_buffered_before_created(cx: &mut gpui::TestAppContext) {
3731 init_test(cx);
3732
3733 let fs = FakeFs::new(cx.executor());
3734 let project = Project::test(fs, [], cx).await;
3735 let connection = Rc::new(FakeAgentConnection::new());
3736 let thread = cx
3737 .update(|cx| {
3738 connection.new_session(
3739 project,
3740 PathList::new(&[std::path::Path::new(path!("/test"))]),
3741 cx,
3742 )
3743 })
3744 .await
3745 .unwrap();
3746
3747 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
3748
3749 // Send Output BEFORE Created
3750 thread.update(cx, |thread, cx| {
3751 thread.on_terminal_provider_event(
3752 TerminalProviderEvent::Output {
3753 terminal_id: terminal_id.clone(),
3754 data: b"pre-exit data".to_vec(),
3755 },
3756 cx,
3757 );
3758 });
3759
3760 // Send Exit BEFORE Created
3761 thread.update(cx, |thread, cx| {
3762 thread.on_terminal_provider_event(
3763 TerminalProviderEvent::Exit {
3764 terminal_id: terminal_id.clone(),
3765 status: acp::TerminalExitStatus::new().exit_code(0),
3766 },
3767 cx,
3768 );
3769 });
3770
3771 // Now create a display-only lower-level terminal and send Created
3772 let lower = cx.new(|cx| {
3773 let builder = ::terminal::TerminalBuilder::new_display_only(
3774 ::terminal::terminal_settings::CursorShape::default(),
3775 ::terminal::terminal_settings::AlternateScroll::On,
3776 None,
3777 0,
3778 cx.background_executor(),
3779 PathStyle::local(),
3780 )
3781 .unwrap();
3782 builder.subscribe(cx)
3783 });
3784
3785 thread.update(cx, |thread, cx| {
3786 thread.on_terminal_provider_event(
3787 TerminalProviderEvent::Created {
3788 terminal_id: terminal_id.clone(),
3789 label: "Buffered Exit Test".to_string(),
3790 cwd: None,
3791 output_byte_limit: None,
3792 terminal: lower.clone(),
3793 },
3794 cx,
3795 );
3796 });
3797
3798 // Output should be present after Created (flushed from buffer)
3799 let content = thread.read_with(cx, |thread, cx| {
3800 let term = thread.terminal(terminal_id.clone()).unwrap();
3801 term.read_with(cx, |t, cx| t.inner().read(cx).get_content())
3802 });
3803
3804 assert!(
3805 content.contains("pre-exit data"),
3806 "expected pre-exit data to render, got: {content}"
3807 );
3808 }
3809
3810 /// Test that killing a terminal via Terminal::kill properly:
3811 /// 1. Causes wait_for_exit to complete (doesn't hang forever)
3812 /// 2. The underlying terminal still has the output that was written before the kill
3813 ///
3814 /// This test verifies that the fix to kill_active_task (which now also kills
3815 /// the shell process in addition to the foreground process) properly allows
3816 /// wait_for_exit to complete instead of hanging indefinitely.
3817 #[cfg(unix)]
3818 #[gpui::test]
3819 async fn test_terminal_kill_allows_wait_for_exit_to_complete(cx: &mut gpui::TestAppContext) {
3820 use std::collections::HashMap;
3821 use task::Shell;
3822 use util::shell_builder::ShellBuilder;
3823
3824 init_test(cx);
3825 cx.executor().allow_parking();
3826
3827 let fs = FakeFs::new(cx.executor());
3828 let project = Project::test(fs, [], cx).await;
3829 let connection = Rc::new(FakeAgentConnection::new());
3830 let thread = cx
3831 .update(|cx| {
3832 connection.new_session(
3833 project.clone(),
3834 PathList::new(&[Path::new(path!("/test"))]),
3835 cx,
3836 )
3837 })
3838 .await
3839 .unwrap();
3840
3841 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
3842
3843 // Create a real PTY terminal that runs a command which prints output then sleeps
3844 // We use printf instead of echo and chain with && sleep to ensure proper execution
3845 let (completion_tx, _completion_rx) = smol::channel::unbounded();
3846 let (program, args) = ShellBuilder::new(&Shell::System, false).build(
3847 Some("printf 'output_before_kill\\n' && sleep 60".to_owned()),
3848 &[],
3849 );
3850
3851 let builder = cx
3852 .update(|cx| {
3853 ::terminal::TerminalBuilder::new(
3854 None,
3855 None,
3856 task::Shell::WithArguments {
3857 program,
3858 args,
3859 title_override: None,
3860 },
3861 HashMap::default(),
3862 ::terminal::terminal_settings::CursorShape::default(),
3863 ::terminal::terminal_settings::AlternateScroll::On,
3864 None,
3865 vec![],
3866 0,
3867 false,
3868 0,
3869 Some(completion_tx),
3870 cx,
3871 vec![],
3872 PathStyle::local(),
3873 )
3874 })
3875 .await
3876 .unwrap();
3877
3878 let lower_terminal = cx.new(|cx| builder.subscribe(cx));
3879
3880 // Create the acp_thread Terminal wrapper
3881 thread.update(cx, |thread, cx| {
3882 thread.on_terminal_provider_event(
3883 TerminalProviderEvent::Created {
3884 terminal_id: terminal_id.clone(),
3885 label: "printf output_before_kill && sleep 60".to_string(),
3886 cwd: None,
3887 output_byte_limit: None,
3888 terminal: lower_terminal.clone(),
3889 },
3890 cx,
3891 );
3892 });
3893
3894 // Wait for the printf command to execute and produce output
3895 // Use real time since parking is enabled
3896 cx.executor().timer(Duration::from_millis(500)).await;
3897
3898 // Get the acp_thread Terminal and kill it
3899 let wait_for_exit = thread.update(cx, |thread, cx| {
3900 let term = thread.terminals.get(&terminal_id).unwrap();
3901 let wait_for_exit = term.read(cx).wait_for_exit();
3902 term.update(cx, |term, cx| {
3903 term.kill(cx);
3904 });
3905 wait_for_exit
3906 });
3907
3908 // KEY ASSERTION: wait_for_exit should complete within a reasonable time (not hang).
3909 // Before the fix to kill_active_task, this would hang forever because
3910 // only the foreground process was killed, not the shell, so the PTY
3911 // child never exited and wait_for_completed_task never completed.
3912 let exit_result = futures::select! {
3913 result = futures::FutureExt::fuse(wait_for_exit) => Some(result),
3914 _ = futures::FutureExt::fuse(cx.background_executor.timer(Duration::from_secs(5))) => None,
3915 };
3916
3917 assert!(
3918 exit_result.is_some(),
3919 "wait_for_exit should complete after kill, but it timed out. \
3920 This indicates kill_active_task is not properly killing the shell process."
3921 );
3922
3923 // Give the system a chance to process any pending updates
3924 cx.run_until_parked();
3925
3926 // Verify that the underlying terminal still has the output that was
3927 // written before the kill. This verifies that killing doesn't lose output.
3928 let inner_content = thread.read_with(cx, |thread, cx| {
3929 let term = thread.terminals.get(&terminal_id).unwrap();
3930 term.read(cx).inner().read(cx).get_content()
3931 });
3932
3933 assert!(
3934 inner_content.contains("output_before_kill"),
3935 "Underlying terminal should contain output from before kill, got: {}",
3936 inner_content
3937 );
3938 }
3939
3940 #[gpui::test]
3941 async fn test_push_user_content_block(cx: &mut gpui::TestAppContext) {
3942 init_test(cx);
3943
3944 let fs = FakeFs::new(cx.executor());
3945 let project = Project::test(fs, [], cx).await;
3946 let connection = Rc::new(FakeAgentConnection::new());
3947 let thread = cx
3948 .update(|cx| {
3949 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
3950 })
3951 .await
3952 .unwrap();
3953
3954 // Test creating a new user message
3955 thread.update(cx, |thread, cx| {
3956 thread.push_user_content_block(None, "Hello, ".into(), cx);
3957 });
3958
3959 thread.update(cx, |thread, cx| {
3960 assert_eq!(thread.entries.len(), 1);
3961 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
3962 assert_eq!(user_msg.id, None);
3963 assert_eq!(user_msg.content.to_markdown(cx), "Hello, ");
3964 } else {
3965 panic!("Expected UserMessage");
3966 }
3967 });
3968
3969 // Test appending to existing user message
3970 let message_1_id = UserMessageId::new();
3971 thread.update(cx, |thread, cx| {
3972 thread.push_user_content_block(Some(message_1_id.clone()), "world!".into(), cx);
3973 });
3974
3975 thread.update(cx, |thread, cx| {
3976 assert_eq!(thread.entries.len(), 1);
3977 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
3978 assert_eq!(user_msg.id, Some(message_1_id));
3979 assert_eq!(user_msg.content.to_markdown(cx), "Hello, world!");
3980 } else {
3981 panic!("Expected UserMessage");
3982 }
3983 });
3984
3985 // Test creating new user message after assistant message
3986 thread.update(cx, |thread, cx| {
3987 thread.push_assistant_content_block("Assistant response".into(), false, cx);
3988 });
3989
3990 let message_2_id = UserMessageId::new();
3991 thread.update(cx, |thread, cx| {
3992 thread.push_user_content_block(
3993 Some(message_2_id.clone()),
3994 "New user message".into(),
3995 cx,
3996 );
3997 });
3998
3999 thread.update(cx, |thread, cx| {
4000 assert_eq!(thread.entries.len(), 3);
4001 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[2] {
4002 assert_eq!(user_msg.id, Some(message_2_id));
4003 assert_eq!(user_msg.content.to_markdown(cx), "New user message");
4004 } else {
4005 panic!("Expected UserMessage at index 2");
4006 }
4007 });
4008 }
4009
4010 #[gpui::test]
4011 async fn test_thinking_concatenation(cx: &mut gpui::TestAppContext) {
4012 init_test(cx);
4013
4014 let fs = FakeFs::new(cx.executor());
4015 let project = Project::test(fs, [], cx).await;
4016 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
4017 |_, thread, mut cx| {
4018 async move {
4019 thread.update(&mut cx, |thread, cx| {
4020 thread
4021 .handle_session_update(
4022 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(
4023 "Thinking ".into(),
4024 )),
4025 cx,
4026 )
4027 .unwrap();
4028 thread
4029 .handle_session_update(
4030 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(
4031 "hard!".into(),
4032 )),
4033 cx,
4034 )
4035 .unwrap();
4036 })?;
4037 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
4038 }
4039 .boxed_local()
4040 },
4041 ));
4042
4043 let thread = cx
4044 .update(|cx| {
4045 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
4046 })
4047 .await
4048 .unwrap();
4049
4050 thread
4051 .update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
4052 .await
4053 .unwrap();
4054
4055 let output = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
4056 assert_eq!(
4057 output,
4058 indoc! {r#"
4059 ## User
4060
4061 Hello from Zed!
4062
4063 ## Assistant
4064
4065 <thinking>
4066 Thinking hard!
4067 </thinking>
4068
4069 "#}
4070 );
4071 }
4072
4073 #[gpui::test]
4074 async fn test_edits_concurrently_to_user(cx: &mut TestAppContext) {
4075 init_test(cx);
4076
4077 let fs = FakeFs::new(cx.executor());
4078 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\n"}))
4079 .await;
4080 let project = Project::test(fs.clone(), [], cx).await;
4081 let (read_file_tx, read_file_rx) = oneshot::channel::<()>();
4082 let read_file_tx = Rc::new(RefCell::new(Some(read_file_tx)));
4083 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
4084 move |_, thread, mut cx| {
4085 let read_file_tx = read_file_tx.clone();
4086 async move {
4087 let content = thread
4088 .update(&mut cx, |thread, cx| {
4089 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
4090 })
4091 .unwrap()
4092 .await
4093 .unwrap();
4094 assert_eq!(content, "one\ntwo\nthree\n");
4095 read_file_tx.take().unwrap().send(()).unwrap();
4096 thread
4097 .update(&mut cx, |thread, cx| {
4098 thread.write_text_file(
4099 path!("/tmp/foo").into(),
4100 "one\ntwo\nthree\nfour\nfive\n".to_string(),
4101 cx,
4102 )
4103 })
4104 .unwrap()
4105 .await
4106 .unwrap();
4107 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
4108 }
4109 .boxed_local()
4110 },
4111 ));
4112
4113 let (worktree, pathbuf) = project
4114 .update(cx, |project, cx| {
4115 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
4116 })
4117 .await
4118 .unwrap();
4119 let buffer = project
4120 .update(cx, |project, cx| {
4121 project.open_buffer((worktree.read(cx).id(), pathbuf), cx)
4122 })
4123 .await
4124 .unwrap();
4125
4126 let thread = cx
4127 .update(|cx| {
4128 connection.new_session(project, PathList::new(&[Path::new(path!("/tmp"))]), cx)
4129 })
4130 .await
4131 .unwrap();
4132
4133 let request = thread.update(cx, |thread, cx| {
4134 thread.send_raw("Extend the count in /tmp/foo", cx)
4135 });
4136 read_file_rx.await.ok();
4137 buffer.update(cx, |buffer, cx| {
4138 buffer.edit([(0..0, "zero\n".to_string())], None, cx);
4139 });
4140 cx.run_until_parked();
4141 assert_eq!(
4142 buffer.read_with(cx, |buffer, _| buffer.text()),
4143 "zero\none\ntwo\nthree\nfour\nfive\n"
4144 );
4145 assert_eq!(
4146 String::from_utf8(fs.read_file_sync(path!("/tmp/foo")).unwrap()).unwrap(),
4147 "zero\none\ntwo\nthree\nfour\nfive\n"
4148 );
4149 request.await.unwrap();
4150 }
4151
4152 #[gpui::test]
4153 async fn test_reading_from_line(cx: &mut TestAppContext) {
4154 init_test(cx);
4155
4156 let fs = FakeFs::new(cx.executor());
4157 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\nfour\n"}))
4158 .await;
4159 let project = Project::test(fs.clone(), [], cx).await;
4160 project
4161 .update(cx, |project, cx| {
4162 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
4163 })
4164 .await
4165 .unwrap();
4166
4167 let connection = Rc::new(FakeAgentConnection::new());
4168
4169 let thread = cx
4170 .update(|cx| {
4171 connection.new_session(project, PathList::new(&[Path::new(path!("/tmp"))]), cx)
4172 })
4173 .await
4174 .unwrap();
4175
4176 // Whole file
4177 let content = thread
4178 .update(cx, |thread, cx| {
4179 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
4180 })
4181 .await
4182 .unwrap();
4183
4184 assert_eq!(content, "one\ntwo\nthree\nfour\n");
4185
4186 // Only start line
4187 let content = thread
4188 .update(cx, |thread, cx| {
4189 thread.read_text_file(path!("/tmp/foo").into(), Some(3), None, false, cx)
4190 })
4191 .await
4192 .unwrap();
4193
4194 assert_eq!(content, "three\nfour\n");
4195
4196 // Only limit
4197 let content = thread
4198 .update(cx, |thread, cx| {
4199 thread.read_text_file(path!("/tmp/foo").into(), None, Some(2), false, cx)
4200 })
4201 .await
4202 .unwrap();
4203
4204 assert_eq!(content, "one\ntwo\n");
4205
4206 // Range
4207 let content = thread
4208 .update(cx, |thread, cx| {
4209 thread.read_text_file(path!("/tmp/foo").into(), Some(2), Some(2), false, cx)
4210 })
4211 .await
4212 .unwrap();
4213
4214 assert_eq!(content, "two\nthree\n");
4215
4216 // Invalid
4217 let err = thread
4218 .update(cx, |thread, cx| {
4219 thread.read_text_file(path!("/tmp/foo").into(), Some(6), Some(2), false, cx)
4220 })
4221 .await
4222 .unwrap_err();
4223
4224 assert_eq!(
4225 err.to_string(),
4226 "Invalid params: \"Attempting to read beyond the end of the file, line 5:0\""
4227 );
4228 }
4229
4230 #[gpui::test]
4231 async fn test_reading_empty_file(cx: &mut TestAppContext) {
4232 init_test(cx);
4233
4234 let fs = FakeFs::new(cx.executor());
4235 fs.insert_tree(path!("/tmp"), json!({"foo": ""})).await;
4236 let project = Project::test(fs.clone(), [], cx).await;
4237 project
4238 .update(cx, |project, cx| {
4239 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
4240 })
4241 .await
4242 .unwrap();
4243
4244 let connection = Rc::new(FakeAgentConnection::new());
4245
4246 let thread = cx
4247 .update(|cx| {
4248 connection.new_session(project, PathList::new(&[Path::new(path!("/tmp"))]), cx)
4249 })
4250 .await
4251 .unwrap();
4252
4253 // Whole file
4254 let content = thread
4255 .update(cx, |thread, cx| {
4256 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
4257 })
4258 .await
4259 .unwrap();
4260
4261 assert_eq!(content, "");
4262
4263 // Only start line
4264 let content = thread
4265 .update(cx, |thread, cx| {
4266 thread.read_text_file(path!("/tmp/foo").into(), Some(1), None, false, cx)
4267 })
4268 .await
4269 .unwrap();
4270
4271 assert_eq!(content, "");
4272
4273 // Only limit
4274 let content = thread
4275 .update(cx, |thread, cx| {
4276 thread.read_text_file(path!("/tmp/foo").into(), None, Some(2), false, cx)
4277 })
4278 .await
4279 .unwrap();
4280
4281 assert_eq!(content, "");
4282
4283 // Range
4284 let content = thread
4285 .update(cx, |thread, cx| {
4286 thread.read_text_file(path!("/tmp/foo").into(), Some(1), Some(1), false, cx)
4287 })
4288 .await
4289 .unwrap();
4290
4291 assert_eq!(content, "");
4292
4293 // Invalid
4294 let err = thread
4295 .update(cx, |thread, cx| {
4296 thread.read_text_file(path!("/tmp/foo").into(), Some(5), Some(2), false, cx)
4297 })
4298 .await
4299 .unwrap_err();
4300
4301 assert_eq!(
4302 err.to_string(),
4303 "Invalid params: \"Attempting to read beyond the end of the file, line 1:0\""
4304 );
4305 }
4306 #[gpui::test]
4307 async fn test_reading_non_existing_file(cx: &mut TestAppContext) {
4308 init_test(cx);
4309
4310 let fs = FakeFs::new(cx.executor());
4311 fs.insert_tree(path!("/tmp"), json!({})).await;
4312 let project = Project::test(fs.clone(), [], cx).await;
4313 project
4314 .update(cx, |project, cx| {
4315 project.find_or_create_worktree(path!("/tmp"), true, cx)
4316 })
4317 .await
4318 .unwrap();
4319
4320 let connection = Rc::new(FakeAgentConnection::new());
4321
4322 let thread = cx
4323 .update(|cx| {
4324 connection.new_session(project, PathList::new(&[Path::new(path!("/tmp"))]), cx)
4325 })
4326 .await
4327 .unwrap();
4328
4329 // Out of project file
4330 let err = thread
4331 .update(cx, |thread, cx| {
4332 thread.read_text_file(path!("/foo").into(), None, None, false, cx)
4333 })
4334 .await
4335 .unwrap_err();
4336
4337 assert_eq!(err.code, acp::ErrorCode::ResourceNotFound);
4338 }
4339
4340 #[gpui::test]
4341 async fn test_succeeding_canceled_toolcall(cx: &mut TestAppContext) {
4342 init_test(cx);
4343
4344 let fs = FakeFs::new(cx.executor());
4345 let project = Project::test(fs, [], cx).await;
4346 let id = acp::ToolCallId::new("test");
4347
4348 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
4349 let id = id.clone();
4350 move |_, thread, mut cx| {
4351 let id = id.clone();
4352 async move {
4353 thread
4354 .update(&mut cx, |thread, cx| {
4355 thread.handle_session_update(
4356 acp::SessionUpdate::ToolCall(
4357 acp::ToolCall::new(id.clone(), "Label")
4358 .kind(acp::ToolKind::Fetch)
4359 .status(acp::ToolCallStatus::InProgress),
4360 ),
4361 cx,
4362 )
4363 })
4364 .unwrap()
4365 .unwrap();
4366 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
4367 }
4368 .boxed_local()
4369 }
4370 }));
4371
4372 let thread = cx
4373 .update(|cx| {
4374 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
4375 })
4376 .await
4377 .unwrap();
4378
4379 let request = thread.update(cx, |thread, cx| {
4380 thread.send_raw("Fetch https://example.com", cx)
4381 });
4382
4383 run_until_first_tool_call(&thread, cx).await;
4384
4385 thread.read_with(cx, |thread, _| {
4386 assert!(matches!(
4387 thread.entries[1],
4388 AgentThreadEntry::ToolCall(ToolCall {
4389 status: ToolCallStatus::InProgress,
4390 ..
4391 })
4392 ));
4393 });
4394
4395 thread.update(cx, |thread, cx| thread.cancel(cx)).await;
4396
4397 thread.read_with(cx, |thread, _| {
4398 assert!(matches!(
4399 &thread.entries[1],
4400 AgentThreadEntry::ToolCall(ToolCall {
4401 status: ToolCallStatus::Canceled,
4402 ..
4403 })
4404 ));
4405 });
4406
4407 thread
4408 .update(cx, |thread, cx| {
4409 thread.handle_session_update(
4410 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4411 id,
4412 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
4413 )),
4414 cx,
4415 )
4416 })
4417 .unwrap();
4418
4419 request.await.unwrap();
4420
4421 thread.read_with(cx, |thread, _| {
4422 assert!(matches!(
4423 thread.entries[1],
4424 AgentThreadEntry::ToolCall(ToolCall {
4425 status: ToolCallStatus::Completed,
4426 ..
4427 })
4428 ));
4429 });
4430 }
4431
4432 #[gpui::test]
4433 async fn test_no_pending_edits_if_tool_calls_are_completed(cx: &mut TestAppContext) {
4434 init_test(cx);
4435 let fs = FakeFs::new(cx.background_executor.clone());
4436 fs.insert_tree(path!("/test"), json!({})).await;
4437 let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
4438
4439 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
4440 move |_, thread, mut cx| {
4441 async move {
4442 thread
4443 .update(&mut cx, |thread, cx| {
4444 thread.handle_session_update(
4445 acp::SessionUpdate::ToolCall(
4446 acp::ToolCall::new("test", "Label")
4447 .kind(acp::ToolKind::Edit)
4448 .status(acp::ToolCallStatus::Completed)
4449 .content(vec![acp::ToolCallContent::Diff(acp::Diff::new(
4450 "/test/test.txt",
4451 "foo",
4452 ))]),
4453 ),
4454 cx,
4455 )
4456 })
4457 .unwrap()
4458 .unwrap();
4459 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
4460 }
4461 .boxed_local()
4462 }
4463 }));
4464
4465 let thread = cx
4466 .update(|cx| {
4467 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
4468 })
4469 .await
4470 .unwrap();
4471
4472 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Hi".into()], cx)))
4473 .await
4474 .unwrap();
4475
4476 assert!(cx.read(|cx| !thread.read(cx).has_pending_edit_tool_calls()));
4477 }
4478
4479 #[gpui::test]
4480 async fn test_pending_edits_for_edit_tool_calls_with_locations(cx: &mut TestAppContext) {
4481 init_test(cx);
4482 let fs = FakeFs::new(cx.background_executor.clone());
4483 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
4484 .await;
4485 let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
4486 let connection = Rc::new(FakeAgentConnection::new());
4487
4488 let thread = cx
4489 .update(|cx| {
4490 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
4491 })
4492 .await
4493 .unwrap();
4494
4495 thread
4496 .update(cx, |thread, cx| {
4497 thread.handle_session_update(
4498 acp::SessionUpdate::ToolCall(
4499 acp::ToolCall::new("test", "Label")
4500 .kind(acp::ToolKind::Edit)
4501 .status(acp::ToolCallStatus::InProgress),
4502 ),
4503 cx,
4504 )
4505 })
4506 .unwrap();
4507
4508 thread
4509 .update(cx, |thread, cx| {
4510 thread.handle_session_update(
4511 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4512 "test",
4513 acp::ToolCallUpdateFields::new()
4514 .locations(vec![acp::ToolCallLocation::new(path!("/test/file.txt"))]),
4515 )),
4516 cx,
4517 )
4518 })
4519 .unwrap();
4520
4521 cx.run_until_parked();
4522
4523 assert!(cx.read(|cx| thread.read(cx).has_pending_edit_tool_calls()));
4524 }
4525
4526 #[gpui::test]
4527 async fn test_infer_external_modified_file_edits_from_tool_call_locations(
4528 cx: &mut TestAppContext,
4529 ) {
4530 init_test(cx);
4531 let fs = FakeFs::new(cx.background_executor.clone());
4532 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
4533 .await;
4534 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
4535 let connection = Rc::new(FakeAgentConnection::new());
4536
4537 let thread = cx
4538 .update(|cx| {
4539 connection.new_session(
4540 project.clone(),
4541 PathList::new(&[Path::new(path!("/test"))]),
4542 cx,
4543 )
4544 })
4545 .await
4546 .unwrap();
4547
4548 thread
4549 .update(cx, |thread, cx| {
4550 thread.handle_session_update(
4551 acp::SessionUpdate::ToolCall(
4552 acp::ToolCall::new("test", "Label")
4553 .kind(acp::ToolKind::Edit)
4554 .status(acp::ToolCallStatus::InProgress),
4555 ),
4556 cx,
4557 )
4558 })
4559 .unwrap();
4560
4561 thread
4562 .update(cx, |thread, cx| {
4563 thread.handle_session_update(
4564 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4565 "test",
4566 acp::ToolCallUpdateFields::new()
4567 .locations(vec![acp::ToolCallLocation::new(path!("/test/file.txt"))]),
4568 )),
4569 cx,
4570 )
4571 })
4572 .unwrap();
4573
4574 cx.run_until_parked();
4575
4576 fs.save(
4577 path!("/test/file.txt").as_ref(),
4578 &"one\ntwo\nthree\n".into(),
4579 Default::default(),
4580 )
4581 .await
4582 .unwrap();
4583 cx.run_until_parked();
4584
4585 thread
4586 .update(cx, |thread, cx| {
4587 thread.handle_session_update(
4588 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4589 "test",
4590 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
4591 )),
4592 cx,
4593 )
4594 })
4595 .unwrap();
4596
4597 cx.executor().advance_clock(Duration::from_millis(200));
4598 cx.run_until_parked();
4599
4600 let action_log = thread.read_with(cx, |thread, _| thread.action_log().clone());
4601 assert_eq!(
4602 action_log.read_with(cx, |action_log, cx| action_log.changed_buffers(cx).len()),
4603 1
4604 );
4605
4606 action_log
4607 .update(cx, |action_log, cx| action_log.reject_all_edits(None, cx))
4608 .await;
4609 cx.run_until_parked();
4610
4611 assert_eq!(
4612 String::from_utf8(fs.read_file_sync(path!("/test/file.txt")).unwrap()).unwrap(),
4613 "one\ntwo\n"
4614 );
4615 }
4616
4617 #[gpui::test]
4618 async fn test_infer_external_created_file_edits_from_tool_call_locations(
4619 cx: &mut TestAppContext,
4620 ) {
4621 init_test(cx);
4622 let fs = FakeFs::new(cx.background_executor.clone());
4623 fs.insert_tree(path!("/test"), json!({})).await;
4624 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
4625 let connection = Rc::new(FakeAgentConnection::new());
4626
4627 let thread = cx
4628 .update(|cx| {
4629 connection.new_session(
4630 project.clone(),
4631 PathList::new(&[Path::new(path!("/test"))]),
4632 cx,
4633 )
4634 })
4635 .await
4636 .unwrap();
4637
4638 thread
4639 .update(cx, |thread, cx| {
4640 thread.handle_session_update(
4641 acp::SessionUpdate::ToolCall(
4642 acp::ToolCall::new("test", "Label")
4643 .kind(acp::ToolKind::Edit)
4644 .status(acp::ToolCallStatus::InProgress),
4645 ),
4646 cx,
4647 )
4648 })
4649 .unwrap();
4650
4651 thread
4652 .update(cx, |thread, cx| {
4653 thread.handle_session_update(
4654 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4655 "test",
4656 acp::ToolCallUpdateFields::new()
4657 .locations(vec![acp::ToolCallLocation::new(path!("/test/new.txt"))]),
4658 )),
4659 cx,
4660 )
4661 })
4662 .unwrap();
4663
4664 cx.run_until_parked();
4665
4666 fs.insert_file(path!("/test/new.txt"), "hello\n".as_bytes().to_vec())
4667 .await;
4668 cx.run_until_parked();
4669
4670 thread
4671 .update(cx, |thread, cx| {
4672 thread.handle_session_update(
4673 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4674 "test",
4675 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
4676 )),
4677 cx,
4678 )
4679 })
4680 .unwrap();
4681
4682 cx.executor().advance_clock(Duration::from_millis(200));
4683 cx.run_until_parked();
4684
4685 let action_log = thread.read_with(cx, |thread, _| thread.action_log().clone());
4686 assert_eq!(
4687 action_log.read_with(cx, |action_log, cx| action_log.changed_buffers(cx).len()),
4688 1
4689 );
4690
4691 action_log
4692 .update(cx, |action_log, cx| action_log.reject_all_edits(None, cx))
4693 .await;
4694 cx.run_until_parked();
4695
4696 assert!(fs.read_file_sync(path!("/test/new.txt")).is_err());
4697 }
4698
4699 fn start_external_edit_tool_call(
4700 thread: &Entity<AcpThread>,
4701 tool_call_id: &acp::ToolCallId,
4702 locations: Vec<acp::ToolCallLocation>,
4703 cx: &mut TestAppContext,
4704 ) {
4705 let tool_call_id = tool_call_id.clone();
4706 thread
4707 .update(cx, move |thread, cx| {
4708 thread.handle_session_update(
4709 acp::SessionUpdate::ToolCall(
4710 acp::ToolCall::new(tool_call_id, "Label")
4711 .kind(acp::ToolKind::Edit)
4712 .status(acp::ToolCallStatus::InProgress)
4713 .locations(locations),
4714 ),
4715 cx,
4716 )
4717 })
4718 .unwrap();
4719 }
4720
4721 fn update_external_edit_tool_call_locations(
4722 thread: &Entity<AcpThread>,
4723 tool_call_id: &acp::ToolCallId,
4724 locations: Vec<acp::ToolCallLocation>,
4725 cx: &mut TestAppContext,
4726 ) {
4727 let tool_call_id = tool_call_id.clone();
4728 thread
4729 .update(cx, move |thread, cx| {
4730 thread.handle_session_update(
4731 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4732 tool_call_id,
4733 acp::ToolCallUpdateFields::new().locations(locations),
4734 )),
4735 cx,
4736 )
4737 })
4738 .unwrap();
4739 }
4740
4741 fn complete_external_edit_tool_call(
4742 thread: &Entity<AcpThread>,
4743 tool_call_id: &acp::ToolCallId,
4744 cx: &mut TestAppContext,
4745 ) {
4746 let tool_call_id = tool_call_id.clone();
4747 thread
4748 .update(cx, move |thread, cx| {
4749 thread.handle_session_update(
4750 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4751 tool_call_id,
4752 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
4753 )),
4754 cx,
4755 )
4756 })
4757 .unwrap();
4758 }
4759
4760 fn changed_buffer_count(thread: &Entity<AcpThread>, cx: &TestAppContext) -> usize {
4761 let action_log = thread.read_with(cx, |thread, _| thread.action_log().clone());
4762 action_log.read_with(cx, |action_log, cx| action_log.changed_buffers(cx).len())
4763 }
4764
4765 fn inferred_edit_candidate_count(thread: &Entity<AcpThread>, cx: &TestAppContext) -> usize {
4766 thread.read_with(cx, |thread, _| {
4767 thread
4768 .inferred_edit_candidates
4769 .values()
4770 .map(|candidates| candidates.len())
4771 .sum()
4772 })
4773 }
4774
4775 fn inferred_edit_candidate_is_ready(
4776 thread: &Entity<AcpThread>,
4777 tool_call_id: &acp::ToolCallId,
4778 abs_path: &PathBuf,
4779 cx: &TestAppContext,
4780 ) -> bool {
4781 thread.read_with(cx, |thread, _| {
4782 thread
4783 .inferred_edit_candidates
4784 .get(tool_call_id)
4785 .and_then(|candidates| candidates.get(abs_path))
4786 .is_some_and(|candidate_state| {
4787 matches!(candidate_state, InferredEditCandidateState::Ready(_))
4788 })
4789 })
4790 }
4791
4792 async fn open_test_buffer(
4793 project: &Entity<Project>,
4794 abs_path: &Path,
4795 cx: &mut TestAppContext,
4796 ) -> Entity<Buffer> {
4797 project
4798 .update(cx, |project, cx| {
4799 let project_path = project
4800 .project_path_for_absolute_path(abs_path, cx)
4801 .unwrap();
4802 project.open_buffer(project_path, cx)
4803 })
4804 .await
4805 .unwrap()
4806 }
4807
4808 #[gpui::test]
4809 async fn test_cancel_clears_inferred_candidate_state_for_external_edit_calls_with_locations(
4810 cx: &mut TestAppContext,
4811 ) {
4812 init_test(cx);
4813 let fs = FakeFs::new(cx.background_executor.clone());
4814 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
4815 .await;
4816 let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
4817 let connection = Rc::new(FakeAgentConnection::new());
4818
4819 let thread = cx
4820 .update(|cx| {
4821 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
4822 })
4823 .await
4824 .unwrap();
4825
4826 let tool_call_id = acp::ToolCallId::new("test");
4827 start_external_edit_tool_call(
4828 &thread,
4829 &tool_call_id,
4830 vec![acp::ToolCallLocation::new(path!("/test/file.txt"))],
4831 cx,
4832 );
4833 cx.run_until_parked();
4834
4835 assert_eq!(inferred_edit_candidate_count(&thread, cx), 1);
4836
4837 let cancel = thread.update(cx, |thread, cx| {
4838 thread.running_turn = Some(RunningTurn {
4839 id: 1,
4840 send_task: Task::ready(()),
4841 });
4842 thread.cancel(cx)
4843 });
4844 cancel.await;
4845 cx.run_until_parked();
4846
4847 assert_eq!(inferred_edit_candidate_count(&thread, cx), 0);
4848 thread.read_with(cx, |thread, _| {
4849 let (_, tool_call) = thread.tool_call(&tool_call_id).unwrap();
4850 assert!(matches!(tool_call.status, ToolCallStatus::Canceled));
4851 });
4852 }
4853
4854 #[gpui::test]
4855 async fn test_completed_update_after_cancel_does_not_reuse_stale_inferred_candidate_baseline(
4856 cx: &mut TestAppContext,
4857 ) {
4858 init_test(cx);
4859 let fs = FakeFs::new(cx.background_executor.clone());
4860 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
4861 .await;
4862 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
4863 let connection = Rc::new(FakeAgentConnection::new());
4864
4865 let thread = cx
4866 .update(|cx| {
4867 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
4868 })
4869 .await
4870 .unwrap();
4871
4872 let tool_call_id = acp::ToolCallId::new("test");
4873 start_external_edit_tool_call(
4874 &thread,
4875 &tool_call_id,
4876 vec![acp::ToolCallLocation::new(path!("/test/file.txt"))],
4877 cx,
4878 );
4879 cx.run_until_parked();
4880
4881 let cancel = thread.update(cx, |thread, cx| {
4882 thread.running_turn = Some(RunningTurn {
4883 id: 1,
4884 send_task: Task::ready(()),
4885 });
4886 thread.cancel(cx)
4887 });
4888 cancel.await;
4889 cx.run_until_parked();
4890
4891 assert_eq!(inferred_edit_candidate_count(&thread, cx), 0);
4892
4893 fs.save(
4894 path!("/test/file.txt").as_ref(),
4895 &"one\ntwo\nthree\n".into(),
4896 Default::default(),
4897 )
4898 .await
4899 .unwrap();
4900 cx.run_until_parked();
4901
4902 complete_external_edit_tool_call(&thread, &tool_call_id, cx);
4903 cx.executor().advance_clock(Duration::from_millis(200));
4904 cx.run_until_parked();
4905
4906 assert_eq!(changed_buffer_count(&thread, cx), 0);
4907 assert_eq!(inferred_edit_candidate_count(&thread, cx), 0);
4908 }
4909
4910 #[gpui::test]
4911 async fn test_user_edit_and_save_during_external_tool_call_does_not_infer_edit(
4912 cx: &mut TestAppContext,
4913 ) {
4914 init_test(cx);
4915 let fs = FakeFs::new(cx.background_executor.clone());
4916 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
4917 .await;
4918 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
4919 let connection = Rc::new(FakeAgentConnection::new());
4920
4921 let thread = cx
4922 .update(|cx| {
4923 connection.new_session(
4924 project.clone(),
4925 PathList::new(&[Path::new(path!("/test"))]),
4926 cx,
4927 )
4928 })
4929 .await
4930 .unwrap();
4931
4932 let tool_call_id = acp::ToolCallId::new("test");
4933 start_external_edit_tool_call(
4934 &thread,
4935 &tool_call_id,
4936 vec![acp::ToolCallLocation::new(path!("/test/file.txt"))],
4937 cx,
4938 );
4939 cx.run_until_parked();
4940
4941 let buffer = open_test_buffer(&project, Path::new(path!("/test/file.txt")), cx).await;
4942 buffer.update(cx, |buffer, cx| {
4943 buffer.edit([(0..0, "zero\n")], None, cx);
4944 });
4945
4946 project
4947 .update(cx, |project, cx| project.save_buffer(buffer.clone(), cx))
4948 .await
4949 .unwrap();
4950 cx.run_until_parked();
4951
4952 complete_external_edit_tool_call(&thread, &tool_call_id, cx);
4953 cx.executor().advance_clock(Duration::from_millis(200));
4954 cx.run_until_parked();
4955
4956 assert_eq!(changed_buffer_count(&thread, cx), 0);
4957 assert_eq!(inferred_edit_candidate_count(&thread, cx), 0);
4958 }
4959
4960 #[gpui::test]
4961 async fn test_already_open_buffer_captures_inferred_edit_baseline_synchronously(
4962 cx: &mut TestAppContext,
4963 ) {
4964 init_test(cx);
4965 let fs = FakeFs::new(cx.background_executor.clone());
4966 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
4967 .await;
4968 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
4969 let connection = Rc::new(FakeAgentConnection::new());
4970
4971 let thread = cx
4972 .update(|cx| {
4973 connection.new_session(
4974 project.clone(),
4975 PathList::new(&[Path::new(path!("/test"))]),
4976 cx,
4977 )
4978 })
4979 .await
4980 .unwrap();
4981
4982 let abs_path = PathBuf::from(path!("/test/file.txt"));
4983 let _buffer = open_test_buffer(&project, Path::new(path!("/test/file.txt")), cx).await;
4984
4985 let tool_call_id = acp::ToolCallId::new("test");
4986 start_external_edit_tool_call(&thread, &tool_call_id, Vec::new(), cx);
4987 update_external_edit_tool_call_locations(
4988 &thread,
4989 &tool_call_id,
4990 vec![acp::ToolCallLocation::new(abs_path.clone())],
4991 cx,
4992 );
4993
4994 assert!(inferred_edit_candidate_is_ready(
4995 &thread,
4996 &tool_call_id,
4997 &abs_path,
4998 cx
4999 ));
5000
5001 fs.save(
5002 path!("/test/file.txt").as_ref(),
5003 &"one\ntwo\nthree\n".into(),
5004 Default::default(),
5005 )
5006 .await
5007 .unwrap();
5008 cx.run_until_parked();
5009
5010 complete_external_edit_tool_call(&thread, &tool_call_id, cx);
5011 cx.executor().advance_clock(Duration::from_millis(200));
5012 cx.run_until_parked();
5013
5014 assert_eq!(changed_buffer_count(&thread, cx), 1);
5015 }
5016
5017 #[gpui::test]
5018 async fn test_location_registration_after_external_write_does_not_infer_without_prior_baseline(
5019 cx: &mut TestAppContext,
5020 ) {
5021 init_test(cx);
5022 let fs = FakeFs::new(cx.background_executor.clone());
5023 fs.insert_tree(path!("/test"), json!({"file.txt": "one\ntwo\n"}))
5024 .await;
5025 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
5026 let connection = Rc::new(FakeAgentConnection::new());
5027
5028 let thread = cx
5029 .update(|cx| {
5030 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5031 })
5032 .await
5033 .unwrap();
5034
5035 let tool_call_id = acp::ToolCallId::new("test");
5036 start_external_edit_tool_call(&thread, &tool_call_id, Vec::new(), cx);
5037
5038 fs.save(
5039 path!("/test/file.txt").as_ref(),
5040 &"one\ntwo\nthree\n".into(),
5041 Default::default(),
5042 )
5043 .await
5044 .unwrap();
5045 cx.run_until_parked();
5046
5047 update_external_edit_tool_call_locations(
5048 &thread,
5049 &tool_call_id,
5050 vec![acp::ToolCallLocation::new(path!("/test/file.txt"))],
5051 cx,
5052 );
5053 cx.run_until_parked();
5054
5055 complete_external_edit_tool_call(&thread, &tool_call_id, cx);
5056 cx.executor().advance_clock(Duration::from_millis(200));
5057 cx.run_until_parked();
5058
5059 assert_eq!(changed_buffer_count(&thread, cx), 0);
5060 }
5061
5062 #[gpui::test(iterations = 10)]
5063 async fn test_checkpoints(cx: &mut TestAppContext) {
5064 init_test(cx);
5065 let fs = FakeFs::new(cx.background_executor.clone());
5066 fs.insert_tree(
5067 path!("/test"),
5068 json!({
5069 ".git": {}
5070 }),
5071 )
5072 .await;
5073 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
5074
5075 let simulate_changes = Arc::new(AtomicBool::new(true));
5076 let next_filename = Arc::new(AtomicUsize::new(0));
5077 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
5078 let simulate_changes = simulate_changes.clone();
5079 let next_filename = next_filename.clone();
5080 let fs = fs.clone();
5081 move |request, thread, mut cx| {
5082 let fs = fs.clone();
5083 let simulate_changes = simulate_changes.clone();
5084 let next_filename = next_filename.clone();
5085 async move {
5086 if simulate_changes.load(SeqCst) {
5087 let filename = format!("/test/file-{}", next_filename.fetch_add(1, SeqCst));
5088 fs.write(Path::new(&filename), b"").await?;
5089 }
5090
5091 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
5092 panic!("expected text content block");
5093 };
5094 thread.update(&mut cx, |thread, cx| {
5095 thread
5096 .handle_session_update(
5097 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(
5098 content.text.to_uppercase().into(),
5099 )),
5100 cx,
5101 )
5102 .unwrap();
5103 })?;
5104 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
5105 }
5106 .boxed_local()
5107 }
5108 }));
5109 let thread = cx
5110 .update(|cx| {
5111 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5112 })
5113 .await
5114 .unwrap();
5115
5116 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Lorem".into()], cx)))
5117 .await
5118 .unwrap();
5119 thread.read_with(cx, |thread, cx| {
5120 assert_eq!(
5121 thread.to_markdown(cx),
5122 indoc! {"
5123 ## User (checkpoint)
5124
5125 Lorem
5126
5127 ## Assistant
5128
5129 LOREM
5130
5131 "}
5132 );
5133 });
5134 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
5135
5136 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["ipsum".into()], cx)))
5137 .await
5138 .unwrap();
5139 thread.read_with(cx, |thread, cx| {
5140 assert_eq!(
5141 thread.to_markdown(cx),
5142 indoc! {"
5143 ## User (checkpoint)
5144
5145 Lorem
5146
5147 ## Assistant
5148
5149 LOREM
5150
5151 ## User (checkpoint)
5152
5153 ipsum
5154
5155 ## Assistant
5156
5157 IPSUM
5158
5159 "}
5160 );
5161 });
5162 assert_eq!(
5163 fs.files(),
5164 vec![
5165 Path::new(path!("/test/file-0")),
5166 Path::new(path!("/test/file-1"))
5167 ]
5168 );
5169
5170 // Checkpoint isn't stored when there are no changes.
5171 simulate_changes.store(false, SeqCst);
5172 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["dolor".into()], cx)))
5173 .await
5174 .unwrap();
5175 thread.read_with(cx, |thread, cx| {
5176 assert_eq!(
5177 thread.to_markdown(cx),
5178 indoc! {"
5179 ## User (checkpoint)
5180
5181 Lorem
5182
5183 ## Assistant
5184
5185 LOREM
5186
5187 ## User (checkpoint)
5188
5189 ipsum
5190
5191 ## Assistant
5192
5193 IPSUM
5194
5195 ## User
5196
5197 dolor
5198
5199 ## Assistant
5200
5201 DOLOR
5202
5203 "}
5204 );
5205 });
5206 assert_eq!(
5207 fs.files(),
5208 vec![
5209 Path::new(path!("/test/file-0")),
5210 Path::new(path!("/test/file-1"))
5211 ]
5212 );
5213
5214 // Rewinding the conversation truncates the history and restores the checkpoint.
5215 thread
5216 .update(cx, |thread, cx| {
5217 let AgentThreadEntry::UserMessage(message) = &thread.entries[2] else {
5218 panic!("unexpected entries {:?}", thread.entries)
5219 };
5220 thread.restore_checkpoint(message.id.clone().unwrap(), cx)
5221 })
5222 .await
5223 .unwrap();
5224 thread.read_with(cx, |thread, cx| {
5225 assert_eq!(
5226 thread.to_markdown(cx),
5227 indoc! {"
5228 ## User (checkpoint)
5229
5230 Lorem
5231
5232 ## Assistant
5233
5234 LOREM
5235
5236 "}
5237 );
5238 });
5239 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
5240 }
5241
5242 #[gpui::test]
5243 async fn test_tool_result_refusal(cx: &mut TestAppContext) {
5244 use std::sync::atomic::AtomicUsize;
5245 init_test(cx);
5246
5247 let fs = FakeFs::new(cx.executor());
5248 let project = Project::test(fs, None, cx).await;
5249
5250 // Create a connection that simulates refusal after tool result
5251 let prompt_count = Arc::new(AtomicUsize::new(0));
5252 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
5253 let prompt_count = prompt_count.clone();
5254 move |_request, thread, mut cx| {
5255 let count = prompt_count.fetch_add(1, SeqCst);
5256 async move {
5257 if count == 0 {
5258 // First prompt: Generate a tool call with result
5259 thread.update(&mut cx, |thread, cx| {
5260 thread
5261 .handle_session_update(
5262 acp::SessionUpdate::ToolCall(
5263 acp::ToolCall::new("tool1", "Test Tool")
5264 .kind(acp::ToolKind::Fetch)
5265 .status(acp::ToolCallStatus::Completed)
5266 .raw_input(serde_json::json!({"query": "test"}))
5267 .raw_output(serde_json::json!({"result": "inappropriate content"})),
5268 ),
5269 cx,
5270 )
5271 .unwrap();
5272 })?;
5273
5274 // Now return refusal because of the tool result
5275 Ok(acp::PromptResponse::new(acp::StopReason::Refusal))
5276 } else {
5277 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
5278 }
5279 }
5280 .boxed_local()
5281 }
5282 }));
5283
5284 let thread = cx
5285 .update(|cx| {
5286 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5287 })
5288 .await
5289 .unwrap();
5290
5291 // Track if we see a Refusal event
5292 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
5293 let saw_refusal_event_captured = saw_refusal_event.clone();
5294 thread.update(cx, |_thread, cx| {
5295 cx.subscribe(
5296 &thread,
5297 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
5298 if matches!(event, AcpThreadEvent::Refusal) {
5299 *saw_refusal_event_captured.lock().unwrap() = true;
5300 }
5301 },
5302 )
5303 .detach();
5304 });
5305
5306 // Send a user message - this will trigger tool call and then refusal
5307 let send_task = thread.update(cx, |thread, cx| thread.send(vec!["Hello".into()], cx));
5308 cx.background_executor.spawn(send_task).detach();
5309 cx.run_until_parked();
5310
5311 // Verify that:
5312 // 1. A Refusal event WAS emitted (because it's a tool result refusal, not user prompt)
5313 // 2. The user message was NOT truncated
5314 assert!(
5315 *saw_refusal_event.lock().unwrap(),
5316 "Refusal event should be emitted for tool result refusals"
5317 );
5318
5319 thread.read_with(cx, |thread, _| {
5320 let entries = thread.entries();
5321 assert!(entries.len() >= 2, "Should have user message and tool call");
5322
5323 // Verify user message is still there
5324 assert!(
5325 matches!(entries[0], AgentThreadEntry::UserMessage(_)),
5326 "User message should not be truncated"
5327 );
5328
5329 // Verify tool call is there with result
5330 if let AgentThreadEntry::ToolCall(tool_call) = &entries[1] {
5331 assert!(
5332 tool_call.raw_output.is_some(),
5333 "Tool call should have output"
5334 );
5335 } else {
5336 panic!("Expected tool call at index 1");
5337 }
5338 });
5339 }
5340
5341 #[gpui::test]
5342 async fn test_user_prompt_refusal_emits_event(cx: &mut TestAppContext) {
5343 init_test(cx);
5344
5345 let fs = FakeFs::new(cx.executor());
5346 let project = Project::test(fs, None, cx).await;
5347
5348 let refuse_next = Arc::new(AtomicBool::new(false));
5349 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
5350 let refuse_next = refuse_next.clone();
5351 move |_request, _thread, _cx| {
5352 if refuse_next.load(SeqCst) {
5353 async move { Ok(acp::PromptResponse::new(acp::StopReason::Refusal)) }
5354 .boxed_local()
5355 } else {
5356 async move { Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)) }
5357 .boxed_local()
5358 }
5359 }
5360 }));
5361
5362 let thread = cx
5363 .update(|cx| {
5364 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5365 })
5366 .await
5367 .unwrap();
5368
5369 // Track if we see a Refusal event
5370 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
5371 let saw_refusal_event_captured = saw_refusal_event.clone();
5372 thread.update(cx, |_thread, cx| {
5373 cx.subscribe(
5374 &thread,
5375 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
5376 if matches!(event, AcpThreadEvent::Refusal) {
5377 *saw_refusal_event_captured.lock().unwrap() = true;
5378 }
5379 },
5380 )
5381 .detach();
5382 });
5383
5384 // Send a message that will be refused
5385 refuse_next.store(true, SeqCst);
5386 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
5387 .await
5388 .unwrap();
5389
5390 // Verify that a Refusal event WAS emitted for user prompt refusal
5391 assert!(
5392 *saw_refusal_event.lock().unwrap(),
5393 "Refusal event should be emitted for user prompt refusals"
5394 );
5395
5396 // Verify the message was truncated (user prompt refusal)
5397 thread.read_with(cx, |thread, cx| {
5398 assert_eq!(thread.to_markdown(cx), "");
5399 });
5400 }
5401
5402 #[gpui::test]
5403 async fn test_refusal(cx: &mut TestAppContext) {
5404 init_test(cx);
5405 let fs = FakeFs::new(cx.background_executor.clone());
5406 fs.insert_tree(path!("/"), json!({})).await;
5407 let project = Project::test(fs.clone(), [path!("/").as_ref()], cx).await;
5408
5409 let refuse_next = Arc::new(AtomicBool::new(false));
5410 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
5411 let refuse_next = refuse_next.clone();
5412 move |request, thread, mut cx| {
5413 let refuse_next = refuse_next.clone();
5414 async move {
5415 if refuse_next.load(SeqCst) {
5416 return Ok(acp::PromptResponse::new(acp::StopReason::Refusal));
5417 }
5418
5419 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
5420 panic!("expected text content block");
5421 };
5422 thread.update(&mut cx, |thread, cx| {
5423 thread
5424 .handle_session_update(
5425 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(
5426 content.text.to_uppercase().into(),
5427 )),
5428 cx,
5429 )
5430 .unwrap();
5431 })?;
5432 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
5433 }
5434 .boxed_local()
5435 }
5436 }));
5437 let thread = cx
5438 .update(|cx| {
5439 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5440 })
5441 .await
5442 .unwrap();
5443
5444 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
5445 .await
5446 .unwrap();
5447 thread.read_with(cx, |thread, cx| {
5448 assert_eq!(
5449 thread.to_markdown(cx),
5450 indoc! {"
5451 ## User
5452
5453 hello
5454
5455 ## Assistant
5456
5457 HELLO
5458
5459 "}
5460 );
5461 });
5462
5463 // Simulate refusing the second message. The message should be truncated
5464 // when a user prompt is refused.
5465 refuse_next.store(true, SeqCst);
5466 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["world".into()], cx)))
5467 .await
5468 .unwrap();
5469 thread.read_with(cx, |thread, cx| {
5470 assert_eq!(
5471 thread.to_markdown(cx),
5472 indoc! {"
5473 ## User
5474
5475 hello
5476
5477 ## Assistant
5478
5479 HELLO
5480
5481 "}
5482 );
5483 });
5484 }
5485
5486 async fn run_until_first_tool_call(
5487 thread: &Entity<AcpThread>,
5488 cx: &mut TestAppContext,
5489 ) -> usize {
5490 let (mut tx, mut rx) = mpsc::channel::<usize>(1);
5491
5492 let subscription = cx.update(|cx| {
5493 cx.subscribe(thread, move |thread, _, cx| {
5494 for (ix, entry) in thread.read(cx).entries.iter().enumerate() {
5495 if matches!(entry, AgentThreadEntry::ToolCall(_)) {
5496 return tx.try_send(ix).unwrap();
5497 }
5498 }
5499 })
5500 });
5501
5502 select! {
5503 _ = futures::FutureExt::fuse(cx.background_executor.timer(Duration::from_secs(10))) => {
5504 panic!("Timeout waiting for tool call")
5505 }
5506 ix = rx.next().fuse() => {
5507 drop(subscription);
5508 ix.unwrap()
5509 }
5510 }
5511 }
5512
5513 #[derive(Clone, Default)]
5514 struct FakeAgentConnection {
5515 auth_methods: Vec<acp::AuthMethod>,
5516 sessions: Arc<parking_lot::Mutex<HashMap<acp::SessionId, WeakEntity<AcpThread>>>>,
5517 set_title_calls: Rc<RefCell<Vec<SharedString>>>,
5518 on_user_message: Option<
5519 Rc<
5520 dyn Fn(
5521 acp::PromptRequest,
5522 WeakEntity<AcpThread>,
5523 AsyncApp,
5524 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
5525 + 'static,
5526 >,
5527 >,
5528 }
5529
5530 impl FakeAgentConnection {
5531 fn new() -> Self {
5532 Self {
5533 auth_methods: Vec::new(),
5534 on_user_message: None,
5535 sessions: Arc::default(),
5536 set_title_calls: Default::default(),
5537 }
5538 }
5539
5540 #[expect(unused)]
5541 fn with_auth_methods(mut self, auth_methods: Vec<acp::AuthMethod>) -> Self {
5542 self.auth_methods = auth_methods;
5543 self
5544 }
5545
5546 fn on_user_message(
5547 mut self,
5548 handler: impl Fn(
5549 acp::PromptRequest,
5550 WeakEntity<AcpThread>,
5551 AsyncApp,
5552 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
5553 + 'static,
5554 ) -> Self {
5555 self.on_user_message.replace(Rc::new(handler));
5556 self
5557 }
5558 }
5559
5560 impl AgentConnection for FakeAgentConnection {
5561 fn agent_id(&self) -> AgentId {
5562 AgentId::new("fake")
5563 }
5564
5565 fn telemetry_id(&self) -> SharedString {
5566 "fake".into()
5567 }
5568
5569 fn auth_methods(&self) -> &[acp::AuthMethod] {
5570 &self.auth_methods
5571 }
5572
5573 fn new_session(
5574 self: Rc<Self>,
5575 project: Entity<Project>,
5576 work_dirs: PathList,
5577 cx: &mut App,
5578 ) -> Task<gpui::Result<Entity<AcpThread>>> {
5579 let session_id = acp::SessionId::new(
5580 rand::rng()
5581 .sample_iter(&distr::Alphanumeric)
5582 .take(7)
5583 .map(char::from)
5584 .collect::<String>(),
5585 );
5586 let action_log = cx.new(|_| ActionLog::new(project.clone()));
5587 let thread = cx.new(|cx| {
5588 AcpThread::new(
5589 None,
5590 "Test",
5591 Some(work_dirs),
5592 self.clone(),
5593 project,
5594 action_log,
5595 session_id.clone(),
5596 watch::Receiver::constant(
5597 acp::PromptCapabilities::new()
5598 .image(true)
5599 .audio(true)
5600 .embedded_context(true),
5601 ),
5602 cx,
5603 )
5604 });
5605 self.sessions.lock().insert(session_id, thread.downgrade());
5606 Task::ready(Ok(thread))
5607 }
5608
5609 fn authenticate(&self, method: acp::AuthMethodId, _cx: &mut App) -> Task<gpui::Result<()>> {
5610 if self.auth_methods().iter().any(|m| m.id() == &method) {
5611 Task::ready(Ok(()))
5612 } else {
5613 Task::ready(Err(anyhow!("Invalid Auth Method")))
5614 }
5615 }
5616
5617 fn prompt(
5618 &self,
5619 _id: Option<UserMessageId>,
5620 params: acp::PromptRequest,
5621 cx: &mut App,
5622 ) -> Task<gpui::Result<acp::PromptResponse>> {
5623 let sessions = self.sessions.lock();
5624 let thread = sessions.get(¶ms.session_id).unwrap();
5625 if let Some(handler) = &self.on_user_message {
5626 let handler = handler.clone();
5627 let thread = thread.clone();
5628 cx.spawn(async move |cx| handler(params, thread, cx.clone()).await)
5629 } else {
5630 Task::ready(Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)))
5631 }
5632 }
5633
5634 fn cancel(&self, _session_id: &acp::SessionId, _cx: &mut App) {}
5635
5636 fn truncate(
5637 &self,
5638 session_id: &acp::SessionId,
5639 _cx: &App,
5640 ) -> Option<Rc<dyn AgentSessionTruncate>> {
5641 Some(Rc::new(FakeAgentSessionEditor {
5642 _session_id: session_id.clone(),
5643 }))
5644 }
5645
5646 fn set_title(
5647 &self,
5648 _session_id: &acp::SessionId,
5649 _cx: &App,
5650 ) -> Option<Rc<dyn AgentSessionSetTitle>> {
5651 Some(Rc::new(FakeAgentSessionSetTitle {
5652 calls: self.set_title_calls.clone(),
5653 }))
5654 }
5655
5656 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
5657 self
5658 }
5659 }
5660
5661 struct FakeAgentSessionSetTitle {
5662 calls: Rc<RefCell<Vec<SharedString>>>,
5663 }
5664
5665 impl AgentSessionSetTitle for FakeAgentSessionSetTitle {
5666 fn run(&self, title: SharedString, _cx: &mut App) -> Task<Result<()>> {
5667 self.calls.borrow_mut().push(title);
5668 Task::ready(Ok(()))
5669 }
5670 }
5671
5672 struct FakeAgentSessionEditor {
5673 _session_id: acp::SessionId,
5674 }
5675
5676 impl AgentSessionTruncate for FakeAgentSessionEditor {
5677 fn run(&self, _message_id: UserMessageId, _cx: &mut App) -> Task<Result<()>> {
5678 Task::ready(Ok(()))
5679 }
5680 }
5681
5682 #[gpui::test]
5683 async fn test_tool_call_not_found_creates_failed_entry(cx: &mut TestAppContext) {
5684 init_test(cx);
5685
5686 let fs = FakeFs::new(cx.executor());
5687 let project = Project::test(fs, [], cx).await;
5688 let connection = Rc::new(FakeAgentConnection::new());
5689 let thread = cx
5690 .update(|cx| {
5691 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5692 })
5693 .await
5694 .unwrap();
5695
5696 // Try to update a tool call that doesn't exist
5697 let nonexistent_id = acp::ToolCallId::new("nonexistent-tool-call");
5698 thread.update(cx, |thread, cx| {
5699 let result = thread.handle_session_update(
5700 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
5701 nonexistent_id.clone(),
5702 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
5703 )),
5704 cx,
5705 );
5706
5707 // The update should succeed (not return an error)
5708 assert!(result.is_ok());
5709
5710 // There should now be exactly one entry in the thread
5711 assert_eq!(thread.entries.len(), 1);
5712
5713 // The entry should be a failed tool call
5714 if let AgentThreadEntry::ToolCall(tool_call) = &thread.entries[0] {
5715 assert_eq!(tool_call.id, nonexistent_id);
5716 assert!(matches!(tool_call.status, ToolCallStatus::Failed));
5717 assert_eq!(tool_call.kind, acp::ToolKind::Fetch);
5718
5719 // Check that the content contains the error message
5720 assert_eq!(tool_call.content.len(), 1);
5721 if let ToolCallContent::ContentBlock(content_block) = &tool_call.content[0] {
5722 match content_block {
5723 ContentBlock::Markdown { markdown } => {
5724 let markdown_text = markdown.read(cx).source();
5725 assert!(markdown_text.contains("Tool call not found"));
5726 }
5727 ContentBlock::Empty => panic!("Expected markdown content, got empty"),
5728 ContentBlock::ResourceLink { .. } => {
5729 panic!("Expected markdown content, got resource link")
5730 }
5731 ContentBlock::Image { .. } => {
5732 panic!("Expected markdown content, got image")
5733 }
5734 }
5735 } else {
5736 panic!("Expected ContentBlock, got: {:?}", tool_call.content[0]);
5737 }
5738 } else {
5739 panic!("Expected ToolCall entry, got: {:?}", thread.entries[0]);
5740 }
5741 });
5742 }
5743
5744 /// Tests that restoring a checkpoint properly cleans up terminals that were
5745 /// created after that checkpoint, and cancels any in-progress generation.
5746 ///
5747 /// Reproduces issue #35142: When a checkpoint is restored, any terminal processes
5748 /// that were started after that checkpoint should be terminated, and any in-progress
5749 /// AI generation should be canceled.
5750 #[gpui::test]
5751 async fn test_restore_checkpoint_kills_terminal(cx: &mut TestAppContext) {
5752 init_test(cx);
5753
5754 let fs = FakeFs::new(cx.executor());
5755 let project = Project::test(fs, [], cx).await;
5756 let connection = Rc::new(FakeAgentConnection::new());
5757 let thread = cx
5758 .update(|cx| {
5759 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
5760 })
5761 .await
5762 .unwrap();
5763
5764 // Send first user message to create a checkpoint
5765 cx.update(|cx| {
5766 thread.update(cx, |thread, cx| {
5767 thread.send(vec!["first message".into()], cx)
5768 })
5769 })
5770 .await
5771 .unwrap();
5772
5773 // Send second message (creates another checkpoint) - we'll restore to this one
5774 cx.update(|cx| {
5775 thread.update(cx, |thread, cx| {
5776 thread.send(vec!["second message".into()], cx)
5777 })
5778 })
5779 .await
5780 .unwrap();
5781
5782 // Create 2 terminals BEFORE the checkpoint that have completed running
5783 let terminal_id_1 = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
5784 let mock_terminal_1 = cx.new(|cx| {
5785 let builder = ::terminal::TerminalBuilder::new_display_only(
5786 ::terminal::terminal_settings::CursorShape::default(),
5787 ::terminal::terminal_settings::AlternateScroll::On,
5788 None,
5789 0,
5790 cx.background_executor(),
5791 PathStyle::local(),
5792 )
5793 .unwrap();
5794 builder.subscribe(cx)
5795 });
5796
5797 thread.update(cx, |thread, cx| {
5798 thread.on_terminal_provider_event(
5799 TerminalProviderEvent::Created {
5800 terminal_id: terminal_id_1.clone(),
5801 label: "echo 'first'".to_string(),
5802 cwd: Some(PathBuf::from("/test")),
5803 output_byte_limit: None,
5804 terminal: mock_terminal_1.clone(),
5805 },
5806 cx,
5807 );
5808 });
5809
5810 thread.update(cx, |thread, cx| {
5811 thread.on_terminal_provider_event(
5812 TerminalProviderEvent::Output {
5813 terminal_id: terminal_id_1.clone(),
5814 data: b"first\n".to_vec(),
5815 },
5816 cx,
5817 );
5818 });
5819
5820 thread.update(cx, |thread, cx| {
5821 thread.on_terminal_provider_event(
5822 TerminalProviderEvent::Exit {
5823 terminal_id: terminal_id_1.clone(),
5824 status: acp::TerminalExitStatus::new().exit_code(0),
5825 },
5826 cx,
5827 );
5828 });
5829
5830 let terminal_id_2 = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
5831 let mock_terminal_2 = cx.new(|cx| {
5832 let builder = ::terminal::TerminalBuilder::new_display_only(
5833 ::terminal::terminal_settings::CursorShape::default(),
5834 ::terminal::terminal_settings::AlternateScroll::On,
5835 None,
5836 0,
5837 cx.background_executor(),
5838 PathStyle::local(),
5839 )
5840 .unwrap();
5841 builder.subscribe(cx)
5842 });
5843
5844 thread.update(cx, |thread, cx| {
5845 thread.on_terminal_provider_event(
5846 TerminalProviderEvent::Created {
5847 terminal_id: terminal_id_2.clone(),
5848 label: "echo 'second'".to_string(),
5849 cwd: Some(PathBuf::from("/test")),
5850 output_byte_limit: None,
5851 terminal: mock_terminal_2.clone(),
5852 },
5853 cx,
5854 );
5855 });
5856
5857 thread.update(cx, |thread, cx| {
5858 thread.on_terminal_provider_event(
5859 TerminalProviderEvent::Output {
5860 terminal_id: terminal_id_2.clone(),
5861 data: b"second\n".to_vec(),
5862 },
5863 cx,
5864 );
5865 });
5866
5867 thread.update(cx, |thread, cx| {
5868 thread.on_terminal_provider_event(
5869 TerminalProviderEvent::Exit {
5870 terminal_id: terminal_id_2.clone(),
5871 status: acp::TerminalExitStatus::new().exit_code(0),
5872 },
5873 cx,
5874 );
5875 });
5876
5877 // Get the second message ID to restore to
5878 let second_message_id = thread.read_with(cx, |thread, _| {
5879 // At this point we have:
5880 // - Index 0: First user message (with checkpoint)
5881 // - Index 1: Second user message (with checkpoint)
5882 // No assistant responses because FakeAgentConnection just returns EndTurn
5883 let AgentThreadEntry::UserMessage(message) = &thread.entries[1] else {
5884 panic!("expected user message at index 1");
5885 };
5886 message.id.clone().unwrap()
5887 });
5888
5889 // Create a terminal AFTER the checkpoint we'll restore to.
5890 // This simulates the AI agent starting a long-running terminal command.
5891 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
5892 let mock_terminal = cx.new(|cx| {
5893 let builder = ::terminal::TerminalBuilder::new_display_only(
5894 ::terminal::terminal_settings::CursorShape::default(),
5895 ::terminal::terminal_settings::AlternateScroll::On,
5896 None,
5897 0,
5898 cx.background_executor(),
5899 PathStyle::local(),
5900 )
5901 .unwrap();
5902 builder.subscribe(cx)
5903 });
5904
5905 // Register the terminal as created
5906 thread.update(cx, |thread, cx| {
5907 thread.on_terminal_provider_event(
5908 TerminalProviderEvent::Created {
5909 terminal_id: terminal_id.clone(),
5910 label: "sleep 1000".to_string(),
5911 cwd: Some(PathBuf::from("/test")),
5912 output_byte_limit: None,
5913 terminal: mock_terminal.clone(),
5914 },
5915 cx,
5916 );
5917 });
5918
5919 // Simulate the terminal producing output (still running)
5920 thread.update(cx, |thread, cx| {
5921 thread.on_terminal_provider_event(
5922 TerminalProviderEvent::Output {
5923 terminal_id: terminal_id.clone(),
5924 data: b"terminal is running...\n".to_vec(),
5925 },
5926 cx,
5927 );
5928 });
5929
5930 // Create a tool call entry that references this terminal
5931 // This represents the agent requesting a terminal command
5932 thread.update(cx, |thread, cx| {
5933 thread
5934 .handle_session_update(
5935 acp::SessionUpdate::ToolCall(
5936 acp::ToolCall::new("terminal-tool-1", "Running command")
5937 .kind(acp::ToolKind::Execute)
5938 .status(acp::ToolCallStatus::InProgress)
5939 .content(vec![acp::ToolCallContent::Terminal(acp::Terminal::new(
5940 terminal_id.clone(),
5941 ))])
5942 .raw_input(serde_json::json!({"command": "sleep 1000", "cd": "/test"})),
5943 ),
5944 cx,
5945 )
5946 .unwrap();
5947 });
5948
5949 // Verify terminal exists and is in the thread
5950 let terminal_exists_before =
5951 thread.read_with(cx, |thread, _| thread.terminals.contains_key(&terminal_id));
5952 assert!(
5953 terminal_exists_before,
5954 "Terminal should exist before checkpoint restore"
5955 );
5956
5957 // Verify the terminal's underlying task is still running (not completed)
5958 let terminal_running_before = thread.read_with(cx, |thread, _cx| {
5959 let terminal_entity = thread.terminals.get(&terminal_id).unwrap();
5960 terminal_entity.read_with(cx, |term, _cx| {
5961 term.output().is_none() // output is None means it's still running
5962 })
5963 });
5964 assert!(
5965 terminal_running_before,
5966 "Terminal should be running before checkpoint restore"
5967 );
5968
5969 // Verify we have the expected entries before restore
5970 let entry_count_before = thread.read_with(cx, |thread, _| thread.entries.len());
5971 assert!(
5972 entry_count_before > 1,
5973 "Should have multiple entries before restore"
5974 );
5975
5976 // Restore the checkpoint to the second message.
5977 // This should:
5978 // 1. Cancel any in-progress generation (via the cancel() call)
5979 // 2. Remove the terminal that was created after that point
5980 thread
5981 .update(cx, |thread, cx| {
5982 thread.restore_checkpoint(second_message_id, cx)
5983 })
5984 .await
5985 .unwrap();
5986
5987 // Verify that no send_task is in progress after restore
5988 // (cancel() clears the send_task)
5989 let has_send_task_after = thread.read_with(cx, |thread, _| thread.running_turn.is_some());
5990 assert!(
5991 !has_send_task_after,
5992 "Should not have a send_task after restore (cancel should have cleared it)"
5993 );
5994
5995 // Verify the entries were truncated (restoring to index 1 truncates at 1, keeping only index 0)
5996 let entry_count = thread.read_with(cx, |thread, _| thread.entries.len());
5997 assert_eq!(
5998 entry_count, 1,
5999 "Should have 1 entry after restore (only the first user message)"
6000 );
6001
6002 // Verify the 2 completed terminals from before the checkpoint still exist
6003 let terminal_1_exists = thread.read_with(cx, |thread, _| {
6004 thread.terminals.contains_key(&terminal_id_1)
6005 });
6006 assert!(
6007 terminal_1_exists,
6008 "Terminal 1 (from before checkpoint) should still exist"
6009 );
6010
6011 let terminal_2_exists = thread.read_with(cx, |thread, _| {
6012 thread.terminals.contains_key(&terminal_id_2)
6013 });
6014 assert!(
6015 terminal_2_exists,
6016 "Terminal 2 (from before checkpoint) should still exist"
6017 );
6018
6019 // Verify they're still in completed state
6020 let terminal_1_completed = thread.read_with(cx, |thread, _cx| {
6021 let terminal_entity = thread.terminals.get(&terminal_id_1).unwrap();
6022 terminal_entity.read_with(cx, |term, _cx| term.output().is_some())
6023 });
6024 assert!(terminal_1_completed, "Terminal 1 should still be completed");
6025
6026 let terminal_2_completed = thread.read_with(cx, |thread, _cx| {
6027 let terminal_entity = thread.terminals.get(&terminal_id_2).unwrap();
6028 terminal_entity.read_with(cx, |term, _cx| term.output().is_some())
6029 });
6030 assert!(terminal_2_completed, "Terminal 2 should still be completed");
6031
6032 // Verify the running terminal (created after checkpoint) was removed
6033 let terminal_3_exists =
6034 thread.read_with(cx, |thread, _| thread.terminals.contains_key(&terminal_id));
6035 assert!(
6036 !terminal_3_exists,
6037 "Terminal 3 (created after checkpoint) should have been removed"
6038 );
6039
6040 // Verify total count is 2 (the two from before the checkpoint)
6041 let terminal_count = thread.read_with(cx, |thread, _| thread.terminals.len());
6042 assert_eq!(
6043 terminal_count, 2,
6044 "Should have exactly 2 terminals (the completed ones from before checkpoint)"
6045 );
6046 }
6047
6048 /// Tests that update_last_checkpoint correctly updates the original message's checkpoint
6049 /// even when a new user message is added while the async checkpoint comparison is in progress.
6050 ///
6051 /// This is a regression test for a bug where update_last_checkpoint would fail with
6052 /// "no checkpoint" if a new user message (without a checkpoint) was added between when
6053 /// update_last_checkpoint started and when its async closure ran.
6054 #[gpui::test]
6055 async fn test_update_last_checkpoint_with_new_message_added(cx: &mut TestAppContext) {
6056 init_test(cx);
6057
6058 let fs = FakeFs::new(cx.executor());
6059 fs.insert_tree(path!("/test"), json!({".git": {}, "file.txt": "content"}))
6060 .await;
6061 let project = Project::test(fs.clone(), [Path::new(path!("/test"))], cx).await;
6062
6063 let handler_done = Arc::new(AtomicBool::new(false));
6064 let handler_done_clone = handler_done.clone();
6065 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
6066 move |_, _thread, _cx| {
6067 handler_done_clone.store(true, SeqCst);
6068 async move { Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)) }.boxed_local()
6069 },
6070 ));
6071
6072 let thread = cx
6073 .update(|cx| {
6074 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
6075 })
6076 .await
6077 .unwrap();
6078
6079 let send_future = thread.update(cx, |thread, cx| thread.send_raw("First message", cx));
6080 let send_task = cx.background_executor.spawn(send_future);
6081
6082 // Tick until handler completes, then a few more to let update_last_checkpoint start
6083 while !handler_done.load(SeqCst) {
6084 cx.executor().tick();
6085 }
6086 for _ in 0..5 {
6087 cx.executor().tick();
6088 }
6089
6090 thread.update(cx, |thread, cx| {
6091 thread.push_entry(
6092 AgentThreadEntry::UserMessage(UserMessage {
6093 id: Some(UserMessageId::new()),
6094 content: ContentBlock::Empty,
6095 chunks: vec!["Injected message (no checkpoint)".into()],
6096 checkpoint: None,
6097 indented: false,
6098 }),
6099 cx,
6100 );
6101 });
6102
6103 cx.run_until_parked();
6104 let result = send_task.await;
6105
6106 assert!(
6107 result.is_ok(),
6108 "send should succeed even when new message added during update_last_checkpoint: {:?}",
6109 result.err()
6110 );
6111 }
6112
6113 /// Tests that when a follow-up message is sent during generation,
6114 /// the first turn completing does NOT clear `running_turn` because
6115 /// it now belongs to the second turn.
6116 #[gpui::test]
6117 async fn test_follow_up_message_during_generation_does_not_clear_turn(cx: &mut TestAppContext) {
6118 init_test(cx);
6119
6120 let fs = FakeFs::new(cx.executor());
6121 let project = Project::test(fs, [], cx).await;
6122
6123 // First handler waits for this signal before completing
6124 let (first_complete_tx, first_complete_rx) = futures::channel::oneshot::channel::<()>();
6125 let first_complete_rx = RefCell::new(Some(first_complete_rx));
6126
6127 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
6128 move |params, _thread, _cx| {
6129 let first_complete_rx = first_complete_rx.borrow_mut().take();
6130 let is_first = params
6131 .prompt
6132 .iter()
6133 .any(|c| matches!(c, acp::ContentBlock::Text(t) if t.text.contains("first")));
6134
6135 async move {
6136 if is_first {
6137 // First handler waits until signaled
6138 if let Some(rx) = first_complete_rx {
6139 rx.await.ok();
6140 }
6141 }
6142 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
6143 }
6144 .boxed_local()
6145 }
6146 }));
6147
6148 let thread = cx
6149 .update(|cx| {
6150 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
6151 })
6152 .await
6153 .unwrap();
6154
6155 // Send first message (turn_id=1) - handler will block
6156 let first_request = thread.update(cx, |thread, cx| thread.send_raw("first", cx));
6157 assert_eq!(thread.read_with(cx, |t, _| t.turn_id), 1);
6158
6159 // Send second message (turn_id=2) while first is still blocked
6160 // This calls cancel() which takes turn 1's running_turn and sets turn 2's
6161 let second_request = thread.update(cx, |thread, cx| thread.send_raw("second", cx));
6162 assert_eq!(thread.read_with(cx, |t, _| t.turn_id), 2);
6163
6164 let running_turn_after_second_send =
6165 thread.read_with(cx, |thread, _| thread.running_turn.as_ref().map(|t| t.id));
6166 assert_eq!(
6167 running_turn_after_second_send,
6168 Some(2),
6169 "running_turn should be set to turn 2 after sending second message"
6170 );
6171
6172 // Now signal first handler to complete
6173 first_complete_tx.send(()).ok();
6174
6175 // First request completes - should NOT clear running_turn
6176 // because running_turn now belongs to turn 2
6177 first_request.await.unwrap();
6178
6179 let running_turn_after_first =
6180 thread.read_with(cx, |thread, _| thread.running_turn.as_ref().map(|t| t.id));
6181 assert_eq!(
6182 running_turn_after_first,
6183 Some(2),
6184 "first turn completing should not clear running_turn (belongs to turn 2)"
6185 );
6186
6187 // Second request completes - SHOULD clear running_turn
6188 second_request.await.unwrap();
6189
6190 let running_turn_after_second =
6191 thread.read_with(cx, |thread, _| thread.running_turn.is_some());
6192 assert!(
6193 !running_turn_after_second,
6194 "second turn completing should clear running_turn"
6195 );
6196 }
6197
6198 #[gpui::test]
6199 async fn test_send_returns_cancelled_response_and_marks_tools_as_cancelled(
6200 cx: &mut TestAppContext,
6201 ) {
6202 init_test(cx);
6203
6204 let fs = FakeFs::new(cx.executor());
6205 let project = Project::test(fs, [], cx).await;
6206
6207 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
6208 move |_params, thread, mut cx| {
6209 async move {
6210 thread
6211 .update(&mut cx, |thread, cx| {
6212 thread.handle_session_update(
6213 acp::SessionUpdate::ToolCall(
6214 acp::ToolCall::new(
6215 acp::ToolCallId::new("test-tool"),
6216 "Test Tool",
6217 )
6218 .kind(acp::ToolKind::Fetch)
6219 .status(acp::ToolCallStatus::InProgress),
6220 ),
6221 cx,
6222 )
6223 })
6224 .unwrap()
6225 .unwrap();
6226
6227 Ok(acp::PromptResponse::new(acp::StopReason::Cancelled))
6228 }
6229 .boxed_local()
6230 },
6231 ));
6232
6233 let thread = cx
6234 .update(|cx| {
6235 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
6236 })
6237 .await
6238 .unwrap();
6239
6240 let response = thread
6241 .update(cx, |thread, cx| thread.send_raw("test message", cx))
6242 .await;
6243
6244 let response = response
6245 .expect("send should succeed")
6246 .expect("should have response");
6247 assert_eq!(
6248 response.stop_reason,
6249 acp::StopReason::Cancelled,
6250 "response should have Cancelled stop_reason"
6251 );
6252
6253 thread.read_with(cx, |thread, _| {
6254 let tool_entry = thread
6255 .entries
6256 .iter()
6257 .find_map(|e| {
6258 if let AgentThreadEntry::ToolCall(call) = e {
6259 Some(call)
6260 } else {
6261 None
6262 }
6263 })
6264 .expect("should have tool call entry");
6265
6266 assert!(
6267 matches!(tool_entry.status, ToolCallStatus::Canceled),
6268 "tool should be marked as Canceled when response is Cancelled, got {:?}",
6269 tool_entry.status
6270 );
6271 });
6272 }
6273
6274 #[gpui::test]
6275 async fn test_provisional_title_replaced_by_real_title(cx: &mut TestAppContext) {
6276 init_test(cx);
6277
6278 let fs = FakeFs::new(cx.executor());
6279 let project = Project::test(fs, [], cx).await;
6280 let connection = Rc::new(FakeAgentConnection::new());
6281 let set_title_calls = connection.set_title_calls.clone();
6282
6283 let thread = cx
6284 .update(|cx| {
6285 connection.new_session(project, PathList::new(&[Path::new(path!("/test"))]), cx)
6286 })
6287 .await
6288 .unwrap();
6289
6290 // Initial title is the default.
6291 thread.read_with(cx, |thread, _| {
6292 assert_eq!(thread.title().as_ref(), "Test");
6293 });
6294
6295 // Setting a provisional title updates the display title.
6296 thread.update(cx, |thread, cx| {
6297 thread.set_provisional_title("Hello, can you help…".into(), cx);
6298 });
6299 thread.read_with(cx, |thread, _| {
6300 assert_eq!(thread.title().as_ref(), "Hello, can you help…");
6301 });
6302
6303 // The provisional title should NOT have propagated to the connection.
6304 assert_eq!(
6305 set_title_calls.borrow().len(),
6306 0,
6307 "provisional title should not propagate to the connection"
6308 );
6309
6310 // When the real title arrives via set_title, it replaces the
6311 // provisional title and propagates to the connection.
6312 let task = thread.update(cx, |thread, cx| {
6313 thread.set_title("Helping with Rust question".into(), cx)
6314 });
6315 task.await.expect("set_title should succeed");
6316 thread.read_with(cx, |thread, _| {
6317 assert_eq!(thread.title().as_ref(), "Helping with Rust question");
6318 });
6319 assert_eq!(
6320 set_title_calls.borrow().as_slice(),
6321 &[SharedString::from("Helping with Rust question")],
6322 "real title should propagate to the connection"
6323 );
6324 }
6325
6326 #[gpui::test]
6327 async fn test_session_info_update_replaces_provisional_title_and_emits_event(
6328 cx: &mut TestAppContext,
6329 ) {
6330 init_test(cx);
6331
6332 let fs = FakeFs::new(cx.executor());
6333 let project = Project::test(fs, [], cx).await;
6334 let connection = Rc::new(FakeAgentConnection::new());
6335
6336 let thread = cx
6337 .update(|cx| {
6338 connection.clone().new_session(
6339 project,
6340 PathList::new(&[Path::new(path!("/test"))]),
6341 cx,
6342 )
6343 })
6344 .await
6345 .unwrap();
6346
6347 let title_updated_events = Rc::new(RefCell::new(0usize));
6348 let title_updated_events_for_subscription = title_updated_events.clone();
6349 thread.update(cx, |_thread, cx| {
6350 cx.subscribe(
6351 &thread,
6352 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
6353 if matches!(event, AcpThreadEvent::TitleUpdated) {
6354 *title_updated_events_for_subscription.borrow_mut() += 1;
6355 }
6356 },
6357 )
6358 .detach();
6359 });
6360
6361 thread.update(cx, |thread, cx| {
6362 thread.set_provisional_title("Hello, can you help…".into(), cx);
6363 });
6364 assert_eq!(
6365 *title_updated_events.borrow(),
6366 1,
6367 "setting a provisional title should emit TitleUpdated"
6368 );
6369
6370 let result = thread.update(cx, |thread, cx| {
6371 thread.handle_session_update(
6372 acp::SessionUpdate::SessionInfoUpdate(
6373 acp::SessionInfoUpdate::new().title("Helping with Rust question"),
6374 ),
6375 cx,
6376 )
6377 });
6378 result.expect("session info update should succeed");
6379
6380 thread.read_with(cx, |thread, _| {
6381 assert_eq!(thread.title().as_ref(), "Helping with Rust question");
6382 assert!(
6383 !thread.has_provisional_title(),
6384 "session info title update should clear provisional title"
6385 );
6386 });
6387
6388 assert_eq!(
6389 *title_updated_events.borrow(),
6390 2,
6391 "session info title update should emit TitleUpdated"
6392 );
6393 assert!(
6394 connection.set_title_calls.borrow().is_empty(),
6395 "session info title update should not propagate back to the connection"
6396 );
6397 }
6398}