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