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