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 has_provisional_title(&self) -> bool {
1211 self.provisional_title.is_some()
1212 }
1213
1214 pub fn entries(&self) -> &[AgentThreadEntry] {
1215 &self.entries
1216 }
1217
1218 pub fn session_id(&self) -> &acp::SessionId {
1219 &self.session_id
1220 }
1221
1222 pub fn cwd(&self) -> Option<&PathBuf> {
1223 self.cwd.as_ref()
1224 }
1225
1226 pub fn status(&self) -> ThreadStatus {
1227 if self.running_turn.is_some() {
1228 ThreadStatus::Generating
1229 } else {
1230 ThreadStatus::Idle
1231 }
1232 }
1233
1234 pub fn had_error(&self) -> bool {
1235 self.had_error
1236 }
1237
1238 pub fn is_waiting_for_confirmation(&self) -> bool {
1239 for entry in self.entries.iter().rev() {
1240 match entry {
1241 AgentThreadEntry::UserMessage(_) => return false,
1242 AgentThreadEntry::ToolCall(ToolCall {
1243 status: ToolCallStatus::WaitingForConfirmation { .. },
1244 ..
1245 }) => return true,
1246 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1247 }
1248 }
1249 false
1250 }
1251
1252 pub fn token_usage(&self) -> Option<&TokenUsage> {
1253 self.token_usage.as_ref()
1254 }
1255
1256 pub fn has_pending_edit_tool_calls(&self) -> bool {
1257 for entry in self.entries.iter().rev() {
1258 match entry {
1259 AgentThreadEntry::UserMessage(_) => return false,
1260 AgentThreadEntry::ToolCall(
1261 call @ ToolCall {
1262 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
1263 ..
1264 },
1265 ) if call.diffs().next().is_some() => {
1266 return true;
1267 }
1268 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1269 }
1270 }
1271
1272 false
1273 }
1274
1275 pub fn has_in_progress_tool_calls(&self) -> bool {
1276 for entry in self.entries.iter().rev() {
1277 match entry {
1278 AgentThreadEntry::UserMessage(_) => return false,
1279 AgentThreadEntry::ToolCall(ToolCall {
1280 status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
1281 ..
1282 }) => {
1283 return true;
1284 }
1285 AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
1286 }
1287 }
1288
1289 false
1290 }
1291
1292 pub fn used_tools_since_last_user_message(&self) -> bool {
1293 for entry in self.entries.iter().rev() {
1294 match entry {
1295 AgentThreadEntry::UserMessage(..) => return false,
1296 AgentThreadEntry::AssistantMessage(..) => continue,
1297 AgentThreadEntry::ToolCall(..) => return true,
1298 }
1299 }
1300
1301 false
1302 }
1303
1304 pub fn handle_session_update(
1305 &mut self,
1306 update: acp::SessionUpdate,
1307 cx: &mut Context<Self>,
1308 ) -> Result<(), acp::Error> {
1309 match update {
1310 acp::SessionUpdate::UserMessageChunk(acp::ContentChunk { content, .. }) => {
1311 self.push_user_content_block(None, content, cx);
1312 }
1313 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk { content, .. }) => {
1314 self.push_assistant_content_block(content, false, cx);
1315 }
1316 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk { content, .. }) => {
1317 self.push_assistant_content_block(content, true, cx);
1318 }
1319 acp::SessionUpdate::ToolCall(tool_call) => {
1320 self.upsert_tool_call(tool_call, cx)?;
1321 }
1322 acp::SessionUpdate::ToolCallUpdate(tool_call_update) => {
1323 self.update_tool_call(tool_call_update, cx)?;
1324 }
1325 acp::SessionUpdate::Plan(plan) => {
1326 self.update_plan(plan, cx);
1327 }
1328 acp::SessionUpdate::AvailableCommandsUpdate(acp::AvailableCommandsUpdate {
1329 available_commands,
1330 ..
1331 }) => cx.emit(AcpThreadEvent::AvailableCommandsUpdated(available_commands)),
1332 acp::SessionUpdate::CurrentModeUpdate(acp::CurrentModeUpdate {
1333 current_mode_id,
1334 ..
1335 }) => cx.emit(AcpThreadEvent::ModeUpdated(current_mode_id)),
1336 acp::SessionUpdate::ConfigOptionUpdate(acp::ConfigOptionUpdate {
1337 config_options,
1338 ..
1339 }) => cx.emit(AcpThreadEvent::ConfigOptionsUpdated(config_options)),
1340 _ => {}
1341 }
1342 Ok(())
1343 }
1344
1345 pub fn push_user_content_block(
1346 &mut self,
1347 message_id: Option<UserMessageId>,
1348 chunk: acp::ContentBlock,
1349 cx: &mut Context<Self>,
1350 ) {
1351 self.push_user_content_block_with_indent(message_id, chunk, false, cx)
1352 }
1353
1354 pub fn push_user_content_block_with_indent(
1355 &mut self,
1356 message_id: Option<UserMessageId>,
1357 chunk: acp::ContentBlock,
1358 indented: bool,
1359 cx: &mut Context<Self>,
1360 ) {
1361 let language_registry = self.project.read(cx).languages().clone();
1362 let path_style = self.project.read(cx).path_style(cx);
1363 let entries_len = self.entries.len();
1364
1365 if let Some(last_entry) = self.entries.last_mut()
1366 && let AgentThreadEntry::UserMessage(UserMessage {
1367 id,
1368 content,
1369 chunks,
1370 indented: existing_indented,
1371 ..
1372 }) = last_entry
1373 && *existing_indented == indented
1374 {
1375 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1376 *id = message_id.or(id.take());
1377 content.append(chunk.clone(), &language_registry, path_style, cx);
1378 chunks.push(chunk);
1379 let idx = entries_len - 1;
1380 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1381 } else {
1382 let content = ContentBlock::new(chunk.clone(), &language_registry, path_style, cx);
1383 self.push_entry(
1384 AgentThreadEntry::UserMessage(UserMessage {
1385 id: message_id,
1386 content,
1387 chunks: vec![chunk],
1388 checkpoint: None,
1389 indented,
1390 }),
1391 cx,
1392 );
1393 }
1394 }
1395
1396 pub fn push_assistant_content_block(
1397 &mut self,
1398 chunk: acp::ContentBlock,
1399 is_thought: bool,
1400 cx: &mut Context<Self>,
1401 ) {
1402 self.push_assistant_content_block_with_indent(chunk, is_thought, false, cx)
1403 }
1404
1405 pub fn push_assistant_content_block_with_indent(
1406 &mut self,
1407 chunk: acp::ContentBlock,
1408 is_thought: bool,
1409 indented: bool,
1410 cx: &mut Context<Self>,
1411 ) {
1412 let path_style = self.project.read(cx).path_style(cx);
1413
1414 // For text chunks going to an existing Markdown block, buffer for smooth
1415 // streaming instead of appending all at once which may feel more choppy.
1416 if let acp::ContentBlock::Text(text_content) = &chunk {
1417 if let Some(markdown) = self.streaming_markdown_target(is_thought, indented) {
1418 let entries_len = self.entries.len();
1419 cx.emit(AcpThreadEvent::EntryUpdated(entries_len - 1));
1420 self.buffer_streaming_text(&markdown, text_content.text.clone(), cx);
1421 return;
1422 }
1423 }
1424
1425 let language_registry = self.project.read(cx).languages().clone();
1426 let entries_len = self.entries.len();
1427 if let Some(last_entry) = self.entries.last_mut()
1428 && let AgentThreadEntry::AssistantMessage(AssistantMessage {
1429 chunks,
1430 indented: existing_indented,
1431 is_subagent_output: _,
1432 }) = last_entry
1433 && *existing_indented == indented
1434 {
1435 let idx = entries_len - 1;
1436 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1437 cx.emit(AcpThreadEvent::EntryUpdated(idx));
1438 match (chunks.last_mut(), is_thought) {
1439 (Some(AssistantMessageChunk::Message { block }), false)
1440 | (Some(AssistantMessageChunk::Thought { block }), true) => {
1441 block.append(chunk, &language_registry, path_style, cx)
1442 }
1443 _ => {
1444 let block = ContentBlock::new(chunk, &language_registry, path_style, cx);
1445 if is_thought {
1446 chunks.push(AssistantMessageChunk::Thought { block })
1447 } else {
1448 chunks.push(AssistantMessageChunk::Message { block })
1449 }
1450 }
1451 }
1452 } else {
1453 let block = ContentBlock::new(chunk, &language_registry, path_style, cx);
1454 let chunk = if is_thought {
1455 AssistantMessageChunk::Thought { block }
1456 } else {
1457 AssistantMessageChunk::Message { block }
1458 };
1459
1460 self.push_entry(
1461 AgentThreadEntry::AssistantMessage(AssistantMessage {
1462 chunks: vec![chunk],
1463 indented,
1464 is_subagent_output: false,
1465 }),
1466 cx,
1467 );
1468 }
1469 }
1470
1471 fn streaming_markdown_target(
1472 &self,
1473 is_thought: bool,
1474 indented: bool,
1475 ) -> Option<Entity<Markdown>> {
1476 let last_entry = self.entries.last()?;
1477 if let AgentThreadEntry::AssistantMessage(AssistantMessage {
1478 chunks,
1479 indented: existing_indented,
1480 ..
1481 }) = last_entry
1482 && *existing_indented == indented
1483 && let [.., chunk] = chunks.as_slice()
1484 {
1485 match (chunk, is_thought) {
1486 (
1487 AssistantMessageChunk::Message {
1488 block: ContentBlock::Markdown { markdown },
1489 },
1490 false,
1491 )
1492 | (
1493 AssistantMessageChunk::Thought {
1494 block: ContentBlock::Markdown { markdown },
1495 },
1496 true,
1497 ) => Some(markdown.clone()),
1498 _ => None,
1499 }
1500 } else {
1501 None
1502 }
1503 }
1504
1505 /// Add text to the streaming buffer. If the target changed (e.g. switching
1506 /// from thoughts to message text), flush the old buffer first.
1507 fn buffer_streaming_text(
1508 &mut self,
1509 markdown: &Entity<Markdown>,
1510 text: String,
1511 cx: &mut Context<Self>,
1512 ) {
1513 if let Some(buffer) = &mut self.streaming_text_buffer {
1514 if buffer.target.entity_id() == markdown.entity_id() {
1515 buffer.pending.push_str(&text);
1516
1517 buffer.bytes_to_reveal_per_tick = (buffer.pending.len() as f32
1518 / StreamingTextBuffer::REVEAL_TARGET
1519 * StreamingTextBuffer::TASK_UPDATE_MS as f32)
1520 .ceil() as usize;
1521 return;
1522 }
1523 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1524 }
1525
1526 let target = markdown.clone();
1527 let _reveal_task = self.start_streaming_reveal(cx);
1528 let pending_len = text.len();
1529 let bytes_to_reveal = (pending_len as f32 / StreamingTextBuffer::REVEAL_TARGET
1530 * StreamingTextBuffer::TASK_UPDATE_MS as f32)
1531 .ceil() as usize;
1532 self.streaming_text_buffer = Some(StreamingTextBuffer {
1533 pending: text,
1534 bytes_to_reveal_per_tick: bytes_to_reveal,
1535 target,
1536 _reveal_task,
1537 });
1538 }
1539
1540 /// Flush all buffered streaming text into the Markdown entity immediately.
1541 fn flush_streaming_text(
1542 streaming_text_buffer: &mut Option<StreamingTextBuffer>,
1543 cx: &mut Context<Self>,
1544 ) {
1545 if let Some(buffer) = streaming_text_buffer.take() {
1546 if !buffer.pending.is_empty() {
1547 buffer
1548 .target
1549 .update(cx, |markdown, cx| markdown.append(&buffer.pending, cx));
1550 }
1551 }
1552 }
1553
1554 /// Spawns a foreground task that periodically drains
1555 /// `streaming_text_buffer.pending` into the target `Markdown` entity,
1556 /// producing smooth, continuous text output.
1557 fn start_streaming_reveal(&self, cx: &mut Context<Self>) -> Task<()> {
1558 cx.spawn(async move |this, cx| {
1559 loop {
1560 cx.background_executor()
1561 .timer(Duration::from_millis(StreamingTextBuffer::TASK_UPDATE_MS))
1562 .await;
1563
1564 let should_continue = this
1565 .update(cx, |this, cx| {
1566 let Some(buffer) = &mut this.streaming_text_buffer else {
1567 return false;
1568 };
1569
1570 if buffer.pending.is_empty() {
1571 return true;
1572 }
1573
1574 let pending_len = buffer.pending.len();
1575
1576 let byte_boundary = buffer
1577 .pending
1578 .ceil_char_boundary(buffer.bytes_to_reveal_per_tick)
1579 .min(pending_len);
1580
1581 buffer.target.update(cx, |markdown: &mut Markdown, cx| {
1582 markdown.append(&buffer.pending[..byte_boundary], cx);
1583 buffer.pending.drain(..byte_boundary);
1584 });
1585
1586 true
1587 })
1588 .unwrap_or(false);
1589
1590 if !should_continue {
1591 break;
1592 }
1593 }
1594 })
1595 }
1596
1597 fn push_entry(&mut self, entry: AgentThreadEntry, cx: &mut Context<Self>) {
1598 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
1599 self.entries.push(entry);
1600 cx.emit(AcpThreadEvent::NewEntry);
1601 }
1602
1603 pub fn can_set_title(&mut self, cx: &mut Context<Self>) -> bool {
1604 self.connection.set_title(&self.session_id, cx).is_some()
1605 }
1606
1607 pub fn set_title(&mut self, title: SharedString, cx: &mut Context<Self>) -> Task<Result<()>> {
1608 let had_provisional = self.provisional_title.take().is_some();
1609 if title != self.title {
1610 self.title = title.clone();
1611 cx.emit(AcpThreadEvent::TitleUpdated);
1612 if let Some(set_title) = self.connection.set_title(&self.session_id, cx) {
1613 return set_title.run(title, cx);
1614 }
1615 } else if had_provisional {
1616 cx.emit(AcpThreadEvent::TitleUpdated);
1617 }
1618 Task::ready(Ok(()))
1619 }
1620
1621 /// Sets a provisional display title without propagating back to the
1622 /// underlying agent connection. This is used for quick preview titles
1623 /// (e.g. first 20 chars of the user message) that should be shown
1624 /// immediately but replaced once the LLM generates a proper title via
1625 /// `set_title`.
1626 pub fn set_provisional_title(&mut self, title: SharedString, cx: &mut Context<Self>) {
1627 self.provisional_title = Some(title);
1628 cx.emit(AcpThreadEvent::TitleUpdated);
1629 }
1630
1631 pub fn subagent_spawned(&mut self, session_id: acp::SessionId, cx: &mut Context<Self>) {
1632 cx.emit(AcpThreadEvent::SubagentSpawned(session_id));
1633 }
1634
1635 pub fn update_token_usage(&mut self, usage: Option<TokenUsage>, cx: &mut Context<Self>) {
1636 self.token_usage = usage;
1637 cx.emit(AcpThreadEvent::TokenUsageUpdated);
1638 }
1639
1640 pub fn update_retry_status(&mut self, status: RetryStatus, cx: &mut Context<Self>) {
1641 cx.emit(AcpThreadEvent::Retry(status));
1642 }
1643
1644 pub fn update_tool_call(
1645 &mut self,
1646 update: impl Into<ToolCallUpdate>,
1647 cx: &mut Context<Self>,
1648 ) -> Result<()> {
1649 let update = update.into();
1650 let languages = self.project.read(cx).languages().clone();
1651 let path_style = self.project.read(cx).path_style(cx);
1652
1653 let ix = match self.index_for_tool_call(update.id()) {
1654 Some(ix) => ix,
1655 None => {
1656 // Tool call not found - create a failed tool call entry
1657 let failed_tool_call = ToolCall {
1658 id: update.id().clone(),
1659 label: cx.new(|cx| Markdown::new("Tool call not found".into(), None, None, cx)),
1660 kind: acp::ToolKind::Fetch,
1661 content: vec![ToolCallContent::ContentBlock(ContentBlock::new(
1662 "Tool call not found".into(),
1663 &languages,
1664 path_style,
1665 cx,
1666 ))],
1667 status: ToolCallStatus::Failed,
1668 locations: Vec::new(),
1669 resolved_locations: Vec::new(),
1670 raw_input: None,
1671 raw_input_markdown: None,
1672 raw_output: None,
1673 tool_name: None,
1674 subagent_session_info: None,
1675 };
1676 self.push_entry(AgentThreadEntry::ToolCall(failed_tool_call), cx);
1677 return Ok(());
1678 }
1679 };
1680 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1681 unreachable!()
1682 };
1683
1684 match update {
1685 ToolCallUpdate::UpdateFields(update) => {
1686 let location_updated = update.fields.locations.is_some();
1687 call.update_fields(
1688 update.fields,
1689 update.meta,
1690 languages,
1691 path_style,
1692 &self.terminals,
1693 cx,
1694 )?;
1695 if location_updated {
1696 self.resolve_locations(update.tool_call_id, cx);
1697 }
1698 }
1699 ToolCallUpdate::UpdateDiff(update) => {
1700 call.content.clear();
1701 call.content.push(ToolCallContent::Diff(update.diff));
1702 }
1703 ToolCallUpdate::UpdateTerminal(update) => {
1704 call.content.clear();
1705 call.content
1706 .push(ToolCallContent::Terminal(update.terminal));
1707 }
1708 }
1709
1710 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1711
1712 Ok(())
1713 }
1714
1715 /// Updates a tool call if id matches an existing entry, otherwise inserts a new one.
1716 pub fn upsert_tool_call(
1717 &mut self,
1718 tool_call: acp::ToolCall,
1719 cx: &mut Context<Self>,
1720 ) -> Result<(), acp::Error> {
1721 let status = tool_call.status.into();
1722 self.upsert_tool_call_inner(tool_call.into(), status, cx)
1723 }
1724
1725 /// Fails if id does not match an existing entry.
1726 pub fn upsert_tool_call_inner(
1727 &mut self,
1728 update: acp::ToolCallUpdate,
1729 status: ToolCallStatus,
1730 cx: &mut Context<Self>,
1731 ) -> Result<(), acp::Error> {
1732 let language_registry = self.project.read(cx).languages().clone();
1733 let path_style = self.project.read(cx).path_style(cx);
1734 let id = update.tool_call_id.clone();
1735
1736 let agent_telemetry_id = self.connection().telemetry_id();
1737 let session = self.session_id();
1738 let parent_session_id = self.parent_session_id();
1739 if let ToolCallStatus::Completed | ToolCallStatus::Failed = status {
1740 let status = if matches!(status, ToolCallStatus::Completed) {
1741 "completed"
1742 } else {
1743 "failed"
1744 };
1745 telemetry::event!(
1746 "Agent Tool Call Completed",
1747 agent_telemetry_id,
1748 session,
1749 parent_session_id,
1750 status
1751 );
1752 }
1753
1754 if let Some(ix) = self.index_for_tool_call(&id) {
1755 let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1756 unreachable!()
1757 };
1758
1759 call.update_fields(
1760 update.fields,
1761 update.meta,
1762 language_registry,
1763 path_style,
1764 &self.terminals,
1765 cx,
1766 )?;
1767 call.status = status;
1768
1769 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1770 } else {
1771 let call = ToolCall::from_acp(
1772 update.try_into()?,
1773 status,
1774 language_registry,
1775 self.project.read(cx).path_style(cx),
1776 &self.terminals,
1777 cx,
1778 )?;
1779 self.push_entry(AgentThreadEntry::ToolCall(call), cx);
1780 };
1781
1782 self.resolve_locations(id, cx);
1783 Ok(())
1784 }
1785
1786 fn index_for_tool_call(&self, id: &acp::ToolCallId) -> Option<usize> {
1787 self.entries
1788 .iter()
1789 .enumerate()
1790 .rev()
1791 .find_map(|(index, entry)| {
1792 if let AgentThreadEntry::ToolCall(tool_call) = entry
1793 && &tool_call.id == id
1794 {
1795 Some(index)
1796 } else {
1797 None
1798 }
1799 })
1800 }
1801
1802 fn tool_call_mut(&mut self, id: &acp::ToolCallId) -> Option<(usize, &mut ToolCall)> {
1803 // The tool call we are looking for is typically the last one, or very close to the end.
1804 // At the moment, it doesn't seem like a hashmap would be a good fit for this use case.
1805 self.entries
1806 .iter_mut()
1807 .enumerate()
1808 .rev()
1809 .find_map(|(index, tool_call)| {
1810 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1811 && &tool_call.id == id
1812 {
1813 Some((index, tool_call))
1814 } else {
1815 None
1816 }
1817 })
1818 }
1819
1820 pub fn tool_call(&self, id: &acp::ToolCallId) -> Option<(usize, &ToolCall)> {
1821 self.entries
1822 .iter()
1823 .enumerate()
1824 .rev()
1825 .find_map(|(index, tool_call)| {
1826 if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1827 && &tool_call.id == id
1828 {
1829 Some((index, tool_call))
1830 } else {
1831 None
1832 }
1833 })
1834 }
1835
1836 pub fn tool_call_for_subagent(&self, session_id: &acp::SessionId) -> Option<&ToolCall> {
1837 self.entries.iter().find_map(|entry| match entry {
1838 AgentThreadEntry::ToolCall(tool_call) => {
1839 if let Some(subagent_session_info) = &tool_call.subagent_session_info
1840 && &subagent_session_info.session_id == session_id
1841 {
1842 Some(tool_call)
1843 } else {
1844 None
1845 }
1846 }
1847 _ => None,
1848 })
1849 }
1850
1851 pub fn resolve_locations(&mut self, id: acp::ToolCallId, cx: &mut Context<Self>) {
1852 let project = self.project.clone();
1853 let should_update_agent_location = self.parent_session_id.is_none();
1854 let Some((_, tool_call)) = self.tool_call_mut(&id) else {
1855 return;
1856 };
1857 let task = tool_call.resolve_locations(project, cx);
1858 cx.spawn(async move |this, cx| {
1859 let resolved_locations = task.await;
1860
1861 this.update(cx, |this, cx| {
1862 let project = this.project.clone();
1863
1864 for location in resolved_locations.iter().flatten() {
1865 this.shared_buffers
1866 .insert(location.buffer.clone(), location.buffer.read(cx).snapshot());
1867 }
1868 let Some((ix, tool_call)) = this.tool_call_mut(&id) else {
1869 return;
1870 };
1871
1872 if let Some(Some(location)) = resolved_locations.last() {
1873 project.update(cx, |project, cx| {
1874 let should_ignore = if let Some(agent_location) = project
1875 .agent_location()
1876 .filter(|agent_location| agent_location.buffer == location.buffer)
1877 {
1878 let snapshot = location.buffer.read(cx).snapshot();
1879 let old_position = agent_location.position.to_point(&snapshot);
1880 let new_position = location.position.to_point(&snapshot);
1881
1882 // ignore this so that when we get updates from the edit tool
1883 // the position doesn't reset to the startof line
1884 old_position.row == new_position.row
1885 && old_position.column > new_position.column
1886 } else {
1887 false
1888 };
1889 if !should_ignore && should_update_agent_location {
1890 project.set_agent_location(Some(location.into()), cx);
1891 }
1892 });
1893 }
1894
1895 let resolved_locations = resolved_locations
1896 .iter()
1897 .map(|l| l.as_ref().map(|l| AgentLocation::from(l)))
1898 .collect::<Vec<_>>();
1899
1900 if tool_call.resolved_locations != resolved_locations {
1901 tool_call.resolved_locations = resolved_locations;
1902 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1903 }
1904 })
1905 })
1906 .detach();
1907 }
1908
1909 pub fn request_tool_call_authorization(
1910 &mut self,
1911 tool_call: acp::ToolCallUpdate,
1912 options: PermissionOptions,
1913 cx: &mut Context<Self>,
1914 ) -> Result<Task<acp::RequestPermissionOutcome>> {
1915 let (tx, rx) = oneshot::channel();
1916
1917 let status = ToolCallStatus::WaitingForConfirmation {
1918 options,
1919 respond_tx: tx,
1920 };
1921
1922 let tool_call_id = tool_call.tool_call_id.clone();
1923 self.upsert_tool_call_inner(tool_call, status, cx)?;
1924 cx.emit(AcpThreadEvent::ToolAuthorizationRequested(
1925 tool_call_id.clone(),
1926 ));
1927
1928 Ok(cx.spawn(async move |this, cx| {
1929 let outcome = match rx.await {
1930 Ok(option) => acp::RequestPermissionOutcome::Selected(
1931 acp::SelectedPermissionOutcome::new(option),
1932 ),
1933 Err(oneshot::Canceled) => acp::RequestPermissionOutcome::Cancelled,
1934 };
1935 this.update(cx, |_this, cx| {
1936 cx.emit(AcpThreadEvent::ToolAuthorizationReceived(tool_call_id))
1937 })
1938 .ok();
1939 outcome
1940 }))
1941 }
1942
1943 pub fn authorize_tool_call(
1944 &mut self,
1945 id: acp::ToolCallId,
1946 option_id: acp::PermissionOptionId,
1947 option_kind: acp::PermissionOptionKind,
1948 cx: &mut Context<Self>,
1949 ) {
1950 let Some((ix, call)) = self.tool_call_mut(&id) else {
1951 return;
1952 };
1953
1954 let new_status = match option_kind {
1955 acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways => {
1956 ToolCallStatus::Rejected
1957 }
1958 acp::PermissionOptionKind::AllowOnce | acp::PermissionOptionKind::AllowAlways => {
1959 ToolCallStatus::InProgress
1960 }
1961 _ => ToolCallStatus::InProgress,
1962 };
1963
1964 let curr_status = mem::replace(&mut call.status, new_status);
1965
1966 if let ToolCallStatus::WaitingForConfirmation { respond_tx, .. } = curr_status {
1967 respond_tx.send(option_id).log_err();
1968 } else if cfg!(debug_assertions) {
1969 panic!("tried to authorize an already authorized tool call");
1970 }
1971
1972 cx.emit(AcpThreadEvent::EntryUpdated(ix));
1973 }
1974
1975 pub fn plan(&self) -> &Plan {
1976 &self.plan
1977 }
1978
1979 pub fn update_plan(&mut self, request: acp::Plan, cx: &mut Context<Self>) {
1980 let new_entries_len = request.entries.len();
1981 let mut new_entries = request.entries.into_iter();
1982
1983 // Reuse existing markdown to prevent flickering
1984 for (old, new) in self.plan.entries.iter_mut().zip(new_entries.by_ref()) {
1985 let PlanEntry {
1986 content,
1987 priority,
1988 status,
1989 } = old;
1990 content.update(cx, |old, cx| {
1991 old.replace(new.content, cx);
1992 });
1993 *priority = new.priority;
1994 *status = new.status;
1995 }
1996 for new in new_entries {
1997 self.plan.entries.push(PlanEntry::from_acp(new, cx))
1998 }
1999 self.plan.entries.truncate(new_entries_len);
2000
2001 cx.notify();
2002 }
2003
2004 fn clear_completed_plan_entries(&mut self, cx: &mut Context<Self>) {
2005 self.plan
2006 .entries
2007 .retain(|entry| !matches!(entry.status, acp::PlanEntryStatus::Completed));
2008 cx.notify();
2009 }
2010
2011 #[cfg(any(test, feature = "test-support"))]
2012 pub fn send_raw(
2013 &mut self,
2014 message: &str,
2015 cx: &mut Context<Self>,
2016 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2017 self.send(vec![message.into()], cx)
2018 }
2019
2020 pub fn send(
2021 &mut self,
2022 message: Vec<acp::ContentBlock>,
2023 cx: &mut Context<Self>,
2024 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2025 let block = ContentBlock::new_combined(
2026 message.clone(),
2027 self.project.read(cx).languages().clone(),
2028 self.project.read(cx).path_style(cx),
2029 cx,
2030 );
2031 let request = acp::PromptRequest::new(self.session_id.clone(), message.clone());
2032 let git_store = self.project.read(cx).git_store().clone();
2033
2034 let message_id = if self.connection.truncate(&self.session_id, cx).is_some() {
2035 Some(UserMessageId::new())
2036 } else {
2037 None
2038 };
2039
2040 self.run_turn(cx, async move |this, cx| {
2041 this.update(cx, |this, cx| {
2042 this.push_entry(
2043 AgentThreadEntry::UserMessage(UserMessage {
2044 id: message_id.clone(),
2045 content: block,
2046 chunks: message,
2047 checkpoint: None,
2048 indented: false,
2049 }),
2050 cx,
2051 );
2052 })
2053 .ok();
2054
2055 let old_checkpoint = git_store
2056 .update(cx, |git, cx| git.checkpoint(cx))
2057 .await
2058 .context("failed to get old checkpoint")
2059 .log_err();
2060 this.update(cx, |this, cx| {
2061 if let Some((_ix, message)) = this.last_user_message() {
2062 message.checkpoint = old_checkpoint.map(|git_checkpoint| Checkpoint {
2063 git_checkpoint,
2064 show: false,
2065 });
2066 }
2067 this.connection.prompt(message_id, request, cx)
2068 })?
2069 .await
2070 })
2071 }
2072
2073 pub fn can_retry(&self, cx: &App) -> bool {
2074 self.connection.retry(&self.session_id, cx).is_some()
2075 }
2076
2077 pub fn retry(
2078 &mut self,
2079 cx: &mut Context<Self>,
2080 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2081 self.run_turn(cx, async move |this, cx| {
2082 this.update(cx, |this, cx| {
2083 this.connection
2084 .retry(&this.session_id, cx)
2085 .map(|retry| retry.run(cx))
2086 })?
2087 .context("retrying a session is not supported")?
2088 .await
2089 })
2090 }
2091
2092 fn run_turn(
2093 &mut self,
2094 cx: &mut Context<Self>,
2095 f: impl 'static + AsyncFnOnce(WeakEntity<Self>, &mut AsyncApp) -> Result<acp::PromptResponse>,
2096 ) -> BoxFuture<'static, Result<Option<acp::PromptResponse>>> {
2097 self.clear_completed_plan_entries(cx);
2098 self.had_error = false;
2099
2100 let (tx, rx) = oneshot::channel();
2101 let cancel_task = self.cancel(cx);
2102
2103 self.turn_id += 1;
2104 let turn_id = self.turn_id;
2105 self.running_turn = Some(RunningTurn {
2106 id: turn_id,
2107 send_task: cx.spawn(async move |this, cx| {
2108 cancel_task.await;
2109 tx.send(f(this, cx).await).ok();
2110 }),
2111 });
2112
2113 cx.spawn(async move |this, cx| {
2114 let response = rx.await;
2115
2116 this.update(cx, |this, cx| this.update_last_checkpoint(cx))?
2117 .await?;
2118
2119 this.update(cx, |this, cx| {
2120 if this.parent_session_id.is_none() {
2121 this.project
2122 .update(cx, |project, cx| project.set_agent_location(None, cx));
2123 }
2124 let Ok(response) = response else {
2125 // tx dropped, just return
2126 return Ok(None);
2127 };
2128
2129 let is_same_turn = this
2130 .running_turn
2131 .as_ref()
2132 .is_some_and(|turn| turn_id == turn.id);
2133
2134 // If the user submitted a follow up message, running_turn might
2135 // already point to a different turn. Therefore we only want to
2136 // take the task if it's the same turn.
2137 if is_same_turn {
2138 this.running_turn.take();
2139 }
2140
2141 match response {
2142 Ok(r) => {
2143 Self::flush_streaming_text(&mut this.streaming_text_buffer, cx);
2144
2145 if r.stop_reason == acp::StopReason::MaxTokens {
2146 this.had_error = true;
2147 cx.emit(AcpThreadEvent::Error);
2148 log::error!("Max tokens reached. Usage: {:?}", this.token_usage);
2149 return Err(anyhow!("Max tokens reached"));
2150 }
2151
2152 let canceled = matches!(r.stop_reason, acp::StopReason::Cancelled);
2153 if canceled {
2154 this.mark_pending_tools_as_canceled();
2155 }
2156
2157 // Handle refusal - distinguish between user prompt and tool call refusals
2158 if let acp::StopReason::Refusal = r.stop_reason {
2159 this.had_error = true;
2160 if let Some((user_msg_ix, _)) = this.last_user_message() {
2161 // Check if there's a completed tool call with results after the last user message
2162 // This indicates the refusal is in response to tool output, not the user's prompt
2163 let has_completed_tool_call_after_user_msg =
2164 this.entries.iter().skip(user_msg_ix + 1).any(|entry| {
2165 if let AgentThreadEntry::ToolCall(tool_call) = entry {
2166 // Check if the tool call has completed and has output
2167 matches!(tool_call.status, ToolCallStatus::Completed)
2168 && tool_call.raw_output.is_some()
2169 } else {
2170 false
2171 }
2172 });
2173
2174 if has_completed_tool_call_after_user_msg {
2175 // Refusal is due to tool output - don't truncate, just notify
2176 // The model refused based on what the tool returned
2177 cx.emit(AcpThreadEvent::Refusal);
2178 } else {
2179 // User prompt was refused - truncate back to before the user message
2180 let range = user_msg_ix..this.entries.len();
2181 if range.start < range.end {
2182 this.entries.truncate(user_msg_ix);
2183 cx.emit(AcpThreadEvent::EntriesRemoved(range));
2184 }
2185 cx.emit(AcpThreadEvent::Refusal);
2186 }
2187 } else {
2188 // No user message found, treat as general refusal
2189 cx.emit(AcpThreadEvent::Refusal);
2190 }
2191 }
2192
2193 cx.emit(AcpThreadEvent::Stopped(r.stop_reason));
2194 Ok(Some(r))
2195 }
2196 Err(e) => {
2197 Self::flush_streaming_text(&mut this.streaming_text_buffer, cx);
2198
2199 this.had_error = true;
2200 cx.emit(AcpThreadEvent::Error);
2201 log::error!("Error in run turn: {:?}", e);
2202 Err(e)
2203 }
2204 }
2205 })?
2206 })
2207 .boxed()
2208 }
2209
2210 pub fn cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
2211 let Some(turn) = self.running_turn.take() else {
2212 return Task::ready(());
2213 };
2214 self.connection.cancel(&self.session_id, cx);
2215
2216 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
2217 self.mark_pending_tools_as_canceled();
2218
2219 // Wait for the send task to complete
2220 cx.background_spawn(turn.send_task)
2221 }
2222
2223 fn mark_pending_tools_as_canceled(&mut self) {
2224 for entry in self.entries.iter_mut() {
2225 if let AgentThreadEntry::ToolCall(call) = entry {
2226 let cancel = matches!(
2227 call.status,
2228 ToolCallStatus::Pending
2229 | ToolCallStatus::WaitingForConfirmation { .. }
2230 | ToolCallStatus::InProgress
2231 );
2232
2233 if cancel {
2234 call.status = ToolCallStatus::Canceled;
2235 }
2236 }
2237 }
2238 }
2239
2240 /// Restores the git working tree to the state at the given checkpoint (if one exists)
2241 pub fn restore_checkpoint(
2242 &mut self,
2243 id: UserMessageId,
2244 cx: &mut Context<Self>,
2245 ) -> Task<Result<()>> {
2246 let Some((_, message)) = self.user_message_mut(&id) else {
2247 return Task::ready(Err(anyhow!("message not found")));
2248 };
2249
2250 let checkpoint = message
2251 .checkpoint
2252 .as_ref()
2253 .map(|c| c.git_checkpoint.clone());
2254
2255 // Cancel any in-progress generation before restoring
2256 let cancel_task = self.cancel(cx);
2257 let rewind = self.rewind(id.clone(), cx);
2258 let git_store = self.project.read(cx).git_store().clone();
2259
2260 cx.spawn(async move |_, cx| {
2261 cancel_task.await;
2262 rewind.await?;
2263 if let Some(checkpoint) = checkpoint {
2264 git_store
2265 .update(cx, |git, cx| git.restore_checkpoint(checkpoint, cx))
2266 .await?;
2267 }
2268
2269 Ok(())
2270 })
2271 }
2272
2273 /// Rewinds this thread to before the entry at `index`, removing it and all
2274 /// subsequent entries while rejecting any action_log changes made from that point.
2275 /// Unlike `restore_checkpoint`, this method does not restore from git.
2276 pub fn rewind(&mut self, id: UserMessageId, cx: &mut Context<Self>) -> Task<Result<()>> {
2277 let Some(truncate) = self.connection.truncate(&self.session_id, cx) else {
2278 return Task::ready(Err(anyhow!("not supported")));
2279 };
2280
2281 Self::flush_streaming_text(&mut self.streaming_text_buffer, cx);
2282 let telemetry = ActionLogTelemetry::from(&*self);
2283 cx.spawn(async move |this, cx| {
2284 cx.update(|cx| truncate.run(id.clone(), cx)).await?;
2285 this.update(cx, |this, cx| {
2286 if let Some((ix, _)) = this.user_message_mut(&id) {
2287 // Collect all terminals from entries that will be removed
2288 let terminals_to_remove: Vec<acp::TerminalId> = this.entries[ix..]
2289 .iter()
2290 .flat_map(|entry| entry.terminals())
2291 .filter_map(|terminal| terminal.read(cx).id().clone().into())
2292 .collect();
2293
2294 let range = ix..this.entries.len();
2295 this.entries.truncate(ix);
2296 cx.emit(AcpThreadEvent::EntriesRemoved(range));
2297
2298 // Kill and remove the terminals
2299 for terminal_id in terminals_to_remove {
2300 if let Some(terminal) = this.terminals.remove(&terminal_id) {
2301 terminal.update(cx, |terminal, cx| {
2302 terminal.kill(cx);
2303 });
2304 }
2305 }
2306 }
2307 this.action_log().update(cx, |action_log, cx| {
2308 action_log.reject_all_edits(Some(telemetry), cx)
2309 })
2310 })?
2311 .await;
2312 Ok(())
2313 })
2314 }
2315
2316 fn update_last_checkpoint(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
2317 let git_store = self.project.read(cx).git_store().clone();
2318
2319 let Some((_, message)) = self.last_user_message() else {
2320 return Task::ready(Ok(()));
2321 };
2322 let Some(user_message_id) = message.id.clone() else {
2323 return Task::ready(Ok(()));
2324 };
2325 let Some(checkpoint) = message.checkpoint.as_ref() else {
2326 return Task::ready(Ok(()));
2327 };
2328 let old_checkpoint = checkpoint.git_checkpoint.clone();
2329
2330 let new_checkpoint = git_store.update(cx, |git, cx| git.checkpoint(cx));
2331 cx.spawn(async move |this, cx| {
2332 let Some(new_checkpoint) = new_checkpoint
2333 .await
2334 .context("failed to get new checkpoint")
2335 .log_err()
2336 else {
2337 return Ok(());
2338 };
2339
2340 let equal = git_store
2341 .update(cx, |git, cx| {
2342 git.compare_checkpoints(old_checkpoint.clone(), new_checkpoint, cx)
2343 })
2344 .await
2345 .unwrap_or(true);
2346
2347 this.update(cx, |this, cx| {
2348 if let Some((ix, message)) = this.user_message_mut(&user_message_id) {
2349 if let Some(checkpoint) = message.checkpoint.as_mut() {
2350 checkpoint.show = !equal;
2351 cx.emit(AcpThreadEvent::EntryUpdated(ix));
2352 }
2353 }
2354 })?;
2355
2356 Ok(())
2357 })
2358 }
2359
2360 fn last_user_message(&mut self) -> Option<(usize, &mut UserMessage)> {
2361 self.entries
2362 .iter_mut()
2363 .enumerate()
2364 .rev()
2365 .find_map(|(ix, entry)| {
2366 if let AgentThreadEntry::UserMessage(message) = entry {
2367 Some((ix, message))
2368 } else {
2369 None
2370 }
2371 })
2372 }
2373
2374 fn user_message_mut(&mut self, id: &UserMessageId) -> Option<(usize, &mut UserMessage)> {
2375 self.entries.iter_mut().enumerate().find_map(|(ix, entry)| {
2376 if let AgentThreadEntry::UserMessage(message) = entry {
2377 if message.id.as_ref() == Some(id) {
2378 Some((ix, message))
2379 } else {
2380 None
2381 }
2382 } else {
2383 None
2384 }
2385 })
2386 }
2387
2388 pub fn read_text_file(
2389 &self,
2390 path: PathBuf,
2391 line: Option<u32>,
2392 limit: Option<u32>,
2393 reuse_shared_snapshot: bool,
2394 cx: &mut Context<Self>,
2395 ) -> Task<Result<String, acp::Error>> {
2396 // Args are 1-based, move to 0-based
2397 let line = line.unwrap_or_default().saturating_sub(1);
2398 let limit = limit.unwrap_or(u32::MAX);
2399 let project = self.project.clone();
2400 let action_log = self.action_log.clone();
2401 let should_update_agent_location = self.parent_session_id.is_none();
2402 cx.spawn(async move |this, cx| {
2403 let load = project.update(cx, |project, cx| {
2404 let path = project
2405 .project_path_for_absolute_path(&path, cx)
2406 .ok_or_else(|| {
2407 acp::Error::resource_not_found(Some(path.display().to_string()))
2408 })?;
2409 Ok::<_, acp::Error>(project.open_buffer(path, cx))
2410 })?;
2411
2412 let buffer = load.await?;
2413
2414 let snapshot = if reuse_shared_snapshot {
2415 this.read_with(cx, |this, _| {
2416 this.shared_buffers.get(&buffer.clone()).cloned()
2417 })
2418 .log_err()
2419 .flatten()
2420 } else {
2421 None
2422 };
2423
2424 let snapshot = if let Some(snapshot) = snapshot {
2425 snapshot
2426 } else {
2427 action_log.update(cx, |action_log, cx| {
2428 action_log.buffer_read(buffer.clone(), cx);
2429 });
2430
2431 let snapshot = buffer.update(cx, |buffer, _| buffer.snapshot());
2432 this.update(cx, |this, _| {
2433 this.shared_buffers.insert(buffer.clone(), snapshot.clone());
2434 })?;
2435 snapshot
2436 };
2437
2438 let max_point = snapshot.max_point();
2439 let start_position = Point::new(line, 0);
2440
2441 if start_position > max_point {
2442 return Err(acp::Error::invalid_params().data(format!(
2443 "Attempting to read beyond the end of the file, line {}:{}",
2444 max_point.row + 1,
2445 max_point.column
2446 )));
2447 }
2448
2449 let start = snapshot.anchor_before(start_position);
2450 let end = snapshot.anchor_before(Point::new(line.saturating_add(limit), 0));
2451
2452 if should_update_agent_location {
2453 project.update(cx, |project, cx| {
2454 project.set_agent_location(
2455 Some(AgentLocation {
2456 buffer: buffer.downgrade(),
2457 position: start,
2458 }),
2459 cx,
2460 );
2461 });
2462 }
2463
2464 Ok(snapshot.text_for_range(start..end).collect::<String>())
2465 })
2466 }
2467
2468 pub fn write_text_file(
2469 &self,
2470 path: PathBuf,
2471 content: String,
2472 cx: &mut Context<Self>,
2473 ) -> Task<Result<()>> {
2474 let project = self.project.clone();
2475 let action_log = self.action_log.clone();
2476 let should_update_agent_location = self.parent_session_id.is_none();
2477 cx.spawn(async move |this, cx| {
2478 let load = project.update(cx, |project, cx| {
2479 let path = project
2480 .project_path_for_absolute_path(&path, cx)
2481 .context("invalid path")?;
2482 anyhow::Ok(project.open_buffer(path, cx))
2483 });
2484 let buffer = load?.await?;
2485 let snapshot = this.update(cx, |this, cx| {
2486 this.shared_buffers
2487 .get(&buffer)
2488 .cloned()
2489 .unwrap_or_else(|| buffer.read(cx).snapshot())
2490 })?;
2491 let edits = cx
2492 .background_executor()
2493 .spawn(async move {
2494 let old_text = snapshot.text();
2495 text_diff(old_text.as_str(), &content)
2496 .into_iter()
2497 .map(|(range, replacement)| {
2498 (snapshot.anchor_range_around(range), replacement)
2499 })
2500 .collect::<Vec<_>>()
2501 })
2502 .await;
2503
2504 if should_update_agent_location {
2505 project.update(cx, |project, cx| {
2506 project.set_agent_location(
2507 Some(AgentLocation {
2508 buffer: buffer.downgrade(),
2509 position: edits
2510 .last()
2511 .map(|(range, _)| range.end)
2512 .unwrap_or(Anchor::min_for_buffer(buffer.read(cx).remote_id())),
2513 }),
2514 cx,
2515 );
2516 });
2517 }
2518
2519 let format_on_save = cx.update(|cx| {
2520 action_log.update(cx, |action_log, cx| {
2521 action_log.buffer_read(buffer.clone(), cx);
2522 });
2523
2524 let format_on_save = buffer.update(cx, |buffer, cx| {
2525 buffer.edit(edits, None, cx);
2526
2527 let settings = language::language_settings::language_settings(
2528 buffer.language().map(|l| l.name()),
2529 buffer.file(),
2530 cx,
2531 );
2532
2533 settings.format_on_save != FormatOnSave::Off
2534 });
2535 action_log.update(cx, |action_log, cx| {
2536 action_log.buffer_edited(buffer.clone(), cx);
2537 });
2538 format_on_save
2539 });
2540
2541 if format_on_save {
2542 let format_task = project.update(cx, |project, cx| {
2543 project.format(
2544 HashSet::from_iter([buffer.clone()]),
2545 LspFormatTarget::Buffers,
2546 false,
2547 FormatTrigger::Save,
2548 cx,
2549 )
2550 });
2551 format_task.await.log_err();
2552
2553 action_log.update(cx, |action_log, cx| {
2554 action_log.buffer_edited(buffer.clone(), cx);
2555 });
2556 }
2557
2558 project
2559 .update(cx, |project, cx| project.save_buffer(buffer, cx))
2560 .await
2561 })
2562 }
2563
2564 pub fn create_terminal(
2565 &self,
2566 command: String,
2567 args: Vec<String>,
2568 extra_env: Vec<acp::EnvVariable>,
2569 cwd: Option<PathBuf>,
2570 output_byte_limit: Option<u64>,
2571 cx: &mut Context<Self>,
2572 ) -> Task<Result<Entity<Terminal>>> {
2573 let env = match &cwd {
2574 Some(dir) => self.project.update(cx, |project, cx| {
2575 project.environment().update(cx, |env, cx| {
2576 env.directory_environment(dir.as_path().into(), cx)
2577 })
2578 }),
2579 None => Task::ready(None).shared(),
2580 };
2581 let env = cx.spawn(async move |_, _| {
2582 let mut env = env.await.unwrap_or_default();
2583 // Disables paging for `git` and hopefully other commands
2584 env.insert("PAGER".into(), "".into());
2585 for var in extra_env {
2586 env.insert(var.name, var.value);
2587 }
2588 env
2589 });
2590
2591 let project = self.project.clone();
2592 let language_registry = project.read(cx).languages().clone();
2593 let is_windows = project.read(cx).path_style(cx).is_windows();
2594
2595 let terminal_id = acp::TerminalId::new(Uuid::new_v4().to_string());
2596 let terminal_task = cx.spawn({
2597 let terminal_id = terminal_id.clone();
2598 async move |_this, cx| {
2599 let env = env.await;
2600 let shell = project
2601 .update(cx, |project, cx| {
2602 project
2603 .remote_client()
2604 .and_then(|r| r.read(cx).default_system_shell())
2605 })
2606 .unwrap_or_else(|| get_default_system_shell_preferring_bash());
2607 let (task_command, task_args) =
2608 ShellBuilder::new(&Shell::Program(shell), is_windows)
2609 .redirect_stdin_to_dev_null()
2610 .build(Some(command.clone()), &args);
2611 let terminal = project
2612 .update(cx, |project, cx| {
2613 project.create_terminal_task(
2614 task::SpawnInTerminal {
2615 command: Some(task_command),
2616 args: task_args,
2617 cwd: cwd.clone(),
2618 env,
2619 ..Default::default()
2620 },
2621 cx,
2622 )
2623 })
2624 .await?;
2625
2626 anyhow::Ok(cx.new(|cx| {
2627 Terminal::new(
2628 terminal_id,
2629 &format!("{} {}", command, args.join(" ")),
2630 cwd,
2631 output_byte_limit.map(|l| l as usize),
2632 terminal,
2633 language_registry,
2634 cx,
2635 )
2636 }))
2637 }
2638 });
2639
2640 cx.spawn(async move |this, cx| {
2641 let terminal = terminal_task.await?;
2642 this.update(cx, |this, _cx| {
2643 this.terminals.insert(terminal_id, terminal.clone());
2644 terminal
2645 })
2646 })
2647 }
2648
2649 pub fn kill_terminal(
2650 &mut self,
2651 terminal_id: acp::TerminalId,
2652 cx: &mut Context<Self>,
2653 ) -> Result<()> {
2654 self.terminals
2655 .get(&terminal_id)
2656 .context("Terminal not found")?
2657 .update(cx, |terminal, cx| {
2658 terminal.kill(cx);
2659 });
2660
2661 Ok(())
2662 }
2663
2664 pub fn release_terminal(
2665 &mut self,
2666 terminal_id: acp::TerminalId,
2667 cx: &mut Context<Self>,
2668 ) -> Result<()> {
2669 self.terminals
2670 .remove(&terminal_id)
2671 .context("Terminal not found")?
2672 .update(cx, |terminal, cx| {
2673 terminal.kill(cx);
2674 });
2675
2676 Ok(())
2677 }
2678
2679 pub fn terminal(&self, terminal_id: acp::TerminalId) -> Result<Entity<Terminal>> {
2680 self.terminals
2681 .get(&terminal_id)
2682 .context("Terminal not found")
2683 .cloned()
2684 }
2685
2686 pub fn to_markdown(&self, cx: &App) -> String {
2687 self.entries.iter().map(|e| e.to_markdown(cx)).collect()
2688 }
2689
2690 pub fn emit_load_error(&mut self, error: LoadError, cx: &mut Context<Self>) {
2691 cx.emit(AcpThreadEvent::LoadError(error));
2692 }
2693
2694 pub fn register_terminal_created(
2695 &mut self,
2696 terminal_id: acp::TerminalId,
2697 command_label: String,
2698 working_dir: Option<PathBuf>,
2699 output_byte_limit: Option<u64>,
2700 terminal: Entity<::terminal::Terminal>,
2701 cx: &mut Context<Self>,
2702 ) -> Entity<Terminal> {
2703 let language_registry = self.project.read(cx).languages().clone();
2704
2705 let entity = cx.new(|cx| {
2706 Terminal::new(
2707 terminal_id.clone(),
2708 &command_label,
2709 working_dir.clone(),
2710 output_byte_limit.map(|l| l as usize),
2711 terminal,
2712 language_registry,
2713 cx,
2714 )
2715 });
2716 self.terminals.insert(terminal_id.clone(), entity.clone());
2717 entity
2718 }
2719
2720 pub fn mark_as_subagent_output(&mut self, cx: &mut Context<Self>) {
2721 for entry in self.entries.iter_mut().rev() {
2722 if let AgentThreadEntry::AssistantMessage(assistant_message) = entry {
2723 assistant_message.is_subagent_output = true;
2724 cx.notify();
2725 return;
2726 }
2727 }
2728 }
2729
2730 pub fn on_terminal_provider_event(
2731 &mut self,
2732 event: TerminalProviderEvent,
2733 cx: &mut Context<Self>,
2734 ) {
2735 match event {
2736 TerminalProviderEvent::Created {
2737 terminal_id,
2738 label,
2739 cwd,
2740 output_byte_limit,
2741 terminal,
2742 } => {
2743 let entity = self.register_terminal_created(
2744 terminal_id.clone(),
2745 label,
2746 cwd,
2747 output_byte_limit,
2748 terminal,
2749 cx,
2750 );
2751
2752 if let Some(mut chunks) = self.pending_terminal_output.remove(&terminal_id) {
2753 for data in chunks.drain(..) {
2754 entity.update(cx, |term, cx| {
2755 term.inner().update(cx, |inner, cx| {
2756 inner.write_output(&data, cx);
2757 })
2758 });
2759 }
2760 }
2761
2762 if let Some(_status) = self.pending_terminal_exit.remove(&terminal_id) {
2763 entity.update(cx, |_term, cx| {
2764 cx.notify();
2765 });
2766 }
2767
2768 cx.notify();
2769 }
2770 TerminalProviderEvent::Output { terminal_id, data } => {
2771 if let Some(entity) = self.terminals.get(&terminal_id) {
2772 entity.update(cx, |term, cx| {
2773 term.inner().update(cx, |inner, cx| {
2774 inner.write_output(&data, cx);
2775 })
2776 });
2777 } else {
2778 self.pending_terminal_output
2779 .entry(terminal_id)
2780 .or_default()
2781 .push(data);
2782 }
2783 }
2784 TerminalProviderEvent::TitleChanged { terminal_id, title } => {
2785 if let Some(entity) = self.terminals.get(&terminal_id) {
2786 entity.update(cx, |term, cx| {
2787 term.inner().update(cx, |inner, cx| {
2788 inner.breadcrumb_text = title;
2789 cx.emit(::terminal::Event::BreadcrumbsChanged);
2790 })
2791 });
2792 }
2793 }
2794 TerminalProviderEvent::Exit {
2795 terminal_id,
2796 status,
2797 } => {
2798 if let Some(entity) = self.terminals.get(&terminal_id) {
2799 entity.update(cx, |_term, cx| {
2800 cx.notify();
2801 });
2802 } else {
2803 self.pending_terminal_exit.insert(terminal_id, status);
2804 }
2805 }
2806 }
2807 }
2808}
2809
2810fn markdown_for_raw_output(
2811 raw_output: &serde_json::Value,
2812 language_registry: &Arc<LanguageRegistry>,
2813 cx: &mut App,
2814) -> Option<Entity<Markdown>> {
2815 match raw_output {
2816 serde_json::Value::Null => None,
2817 serde_json::Value::Bool(value) => Some(cx.new(|cx| {
2818 Markdown::new(
2819 value.to_string().into(),
2820 Some(language_registry.clone()),
2821 None,
2822 cx,
2823 )
2824 })),
2825 serde_json::Value::Number(value) => Some(cx.new(|cx| {
2826 Markdown::new(
2827 value.to_string().into(),
2828 Some(language_registry.clone()),
2829 None,
2830 cx,
2831 )
2832 })),
2833 serde_json::Value::String(value) => Some(cx.new(|cx| {
2834 Markdown::new(
2835 value.clone().into(),
2836 Some(language_registry.clone()),
2837 None,
2838 cx,
2839 )
2840 })),
2841 value => Some(cx.new(|cx| {
2842 let pretty_json = to_string_pretty(value).unwrap_or_else(|_| value.to_string());
2843
2844 Markdown::new(
2845 format!("```json\n{}\n```", pretty_json).into(),
2846 Some(language_registry.clone()),
2847 None,
2848 cx,
2849 )
2850 })),
2851 }
2852}
2853
2854#[cfg(test)]
2855mod tests {
2856 use super::*;
2857 use anyhow::anyhow;
2858 use futures::{channel::mpsc, future::LocalBoxFuture, select};
2859 use gpui::{App, AsyncApp, TestAppContext, WeakEntity};
2860 use indoc::indoc;
2861 use project::{FakeFs, Fs};
2862 use rand::{distr, prelude::*};
2863 use serde_json::json;
2864 use settings::SettingsStore;
2865 use smol::stream::StreamExt as _;
2866 use std::{
2867 any::Any,
2868 cell::RefCell,
2869 path::Path,
2870 rc::Rc,
2871 sync::atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
2872 time::Duration,
2873 };
2874 use util::path;
2875
2876 fn init_test(cx: &mut TestAppContext) {
2877 env_logger::try_init().ok();
2878 cx.update(|cx| {
2879 let settings_store = SettingsStore::test(cx);
2880 cx.set_global(settings_store);
2881 });
2882 }
2883
2884 #[gpui::test]
2885 async fn test_terminal_output_buffered_before_created_renders(cx: &mut gpui::TestAppContext) {
2886 init_test(cx);
2887
2888 let fs = FakeFs::new(cx.executor());
2889 let project = Project::test(fs, [], cx).await;
2890 let connection = Rc::new(FakeAgentConnection::new());
2891 let thread = cx
2892 .update(|cx| connection.new_session(project, std::path::Path::new(path!("/test")), cx))
2893 .await
2894 .unwrap();
2895
2896 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
2897
2898 // Send Output BEFORE Created - should be buffered by acp_thread
2899 thread.update(cx, |thread, cx| {
2900 thread.on_terminal_provider_event(
2901 TerminalProviderEvent::Output {
2902 terminal_id: terminal_id.clone(),
2903 data: b"hello buffered".to_vec(),
2904 },
2905 cx,
2906 );
2907 });
2908
2909 // Create a display-only terminal and then send Created
2910 let lower = cx.new(|cx| {
2911 let builder = ::terminal::TerminalBuilder::new_display_only(
2912 ::terminal::terminal_settings::CursorShape::default(),
2913 ::terminal::terminal_settings::AlternateScroll::On,
2914 None,
2915 0,
2916 cx.background_executor(),
2917 PathStyle::local(),
2918 )
2919 .unwrap();
2920 builder.subscribe(cx)
2921 });
2922
2923 thread.update(cx, |thread, cx| {
2924 thread.on_terminal_provider_event(
2925 TerminalProviderEvent::Created {
2926 terminal_id: terminal_id.clone(),
2927 label: "Buffered Test".to_string(),
2928 cwd: None,
2929 output_byte_limit: None,
2930 terminal: lower.clone(),
2931 },
2932 cx,
2933 );
2934 });
2935
2936 // After Created, buffered Output should have been flushed into the renderer
2937 let content = thread.read_with(cx, |thread, cx| {
2938 let term = thread.terminal(terminal_id.clone()).unwrap();
2939 term.read_with(cx, |t, cx| t.inner().read(cx).get_content())
2940 });
2941
2942 assert!(
2943 content.contains("hello buffered"),
2944 "expected buffered output to render, got: {content}"
2945 );
2946 }
2947
2948 #[gpui::test]
2949 async fn test_terminal_output_and_exit_buffered_before_created(cx: &mut gpui::TestAppContext) {
2950 init_test(cx);
2951
2952 let fs = FakeFs::new(cx.executor());
2953 let project = Project::test(fs, [], cx).await;
2954 let connection = Rc::new(FakeAgentConnection::new());
2955 let thread = cx
2956 .update(|cx| connection.new_session(project, std::path::Path::new(path!("/test")), cx))
2957 .await
2958 .unwrap();
2959
2960 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
2961
2962 // Send Output BEFORE Created
2963 thread.update(cx, |thread, cx| {
2964 thread.on_terminal_provider_event(
2965 TerminalProviderEvent::Output {
2966 terminal_id: terminal_id.clone(),
2967 data: b"pre-exit data".to_vec(),
2968 },
2969 cx,
2970 );
2971 });
2972
2973 // Send Exit BEFORE Created
2974 thread.update(cx, |thread, cx| {
2975 thread.on_terminal_provider_event(
2976 TerminalProviderEvent::Exit {
2977 terminal_id: terminal_id.clone(),
2978 status: acp::TerminalExitStatus::new().exit_code(0),
2979 },
2980 cx,
2981 );
2982 });
2983
2984 // Now create a display-only lower-level terminal and send Created
2985 let lower = cx.new(|cx| {
2986 let builder = ::terminal::TerminalBuilder::new_display_only(
2987 ::terminal::terminal_settings::CursorShape::default(),
2988 ::terminal::terminal_settings::AlternateScroll::On,
2989 None,
2990 0,
2991 cx.background_executor(),
2992 PathStyle::local(),
2993 )
2994 .unwrap();
2995 builder.subscribe(cx)
2996 });
2997
2998 thread.update(cx, |thread, cx| {
2999 thread.on_terminal_provider_event(
3000 TerminalProviderEvent::Created {
3001 terminal_id: terminal_id.clone(),
3002 label: "Buffered Exit Test".to_string(),
3003 cwd: None,
3004 output_byte_limit: None,
3005 terminal: lower.clone(),
3006 },
3007 cx,
3008 );
3009 });
3010
3011 // Output should be present after Created (flushed from buffer)
3012 let content = thread.read_with(cx, |thread, cx| {
3013 let term = thread.terminal(terminal_id.clone()).unwrap();
3014 term.read_with(cx, |t, cx| t.inner().read(cx).get_content())
3015 });
3016
3017 assert!(
3018 content.contains("pre-exit data"),
3019 "expected pre-exit data to render, got: {content}"
3020 );
3021 }
3022
3023 /// Test that killing a terminal via Terminal::kill properly:
3024 /// 1. Causes wait_for_exit to complete (doesn't hang forever)
3025 /// 2. The underlying terminal still has the output that was written before the kill
3026 ///
3027 /// This test verifies that the fix to kill_active_task (which now also kills
3028 /// the shell process in addition to the foreground process) properly allows
3029 /// wait_for_exit to complete instead of hanging indefinitely.
3030 #[cfg(unix)]
3031 #[gpui::test]
3032 async fn test_terminal_kill_allows_wait_for_exit_to_complete(cx: &mut gpui::TestAppContext) {
3033 use std::collections::HashMap;
3034 use task::Shell;
3035 use util::shell_builder::ShellBuilder;
3036
3037 init_test(cx);
3038 cx.executor().allow_parking();
3039
3040 let fs = FakeFs::new(cx.executor());
3041 let project = Project::test(fs, [], cx).await;
3042 let connection = Rc::new(FakeAgentConnection::new());
3043 let thread = cx
3044 .update(|cx| connection.new_session(project.clone(), Path::new(path!("/test")), cx))
3045 .await
3046 .unwrap();
3047
3048 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
3049
3050 // Create a real PTY terminal that runs a command which prints output then sleeps
3051 // We use printf instead of echo and chain with && sleep to ensure proper execution
3052 let (completion_tx, _completion_rx) = smol::channel::unbounded();
3053 let (program, args) = ShellBuilder::new(&Shell::System, false).build(
3054 Some("printf 'output_before_kill\\n' && sleep 60".to_owned()),
3055 &[],
3056 );
3057
3058 let builder = cx
3059 .update(|cx| {
3060 ::terminal::TerminalBuilder::new(
3061 None,
3062 None,
3063 task::Shell::WithArguments {
3064 program,
3065 args,
3066 title_override: None,
3067 },
3068 HashMap::default(),
3069 ::terminal::terminal_settings::CursorShape::default(),
3070 ::terminal::terminal_settings::AlternateScroll::On,
3071 None,
3072 vec![],
3073 0,
3074 false,
3075 0,
3076 Some(completion_tx),
3077 cx,
3078 vec![],
3079 PathStyle::local(),
3080 )
3081 })
3082 .await
3083 .unwrap();
3084
3085 let lower_terminal = cx.new(|cx| builder.subscribe(cx));
3086
3087 // Create the acp_thread Terminal wrapper
3088 thread.update(cx, |thread, cx| {
3089 thread.on_terminal_provider_event(
3090 TerminalProviderEvent::Created {
3091 terminal_id: terminal_id.clone(),
3092 label: "printf output_before_kill && sleep 60".to_string(),
3093 cwd: None,
3094 output_byte_limit: None,
3095 terminal: lower_terminal.clone(),
3096 },
3097 cx,
3098 );
3099 });
3100
3101 // Wait for the printf command to execute and produce output
3102 // Use real time since parking is enabled
3103 cx.executor().timer(Duration::from_millis(500)).await;
3104
3105 // Get the acp_thread Terminal and kill it
3106 let wait_for_exit = thread.update(cx, |thread, cx| {
3107 let term = thread.terminals.get(&terminal_id).unwrap();
3108 let wait_for_exit = term.read(cx).wait_for_exit();
3109 term.update(cx, |term, cx| {
3110 term.kill(cx);
3111 });
3112 wait_for_exit
3113 });
3114
3115 // KEY ASSERTION: wait_for_exit should complete within a reasonable time (not hang).
3116 // Before the fix to kill_active_task, this would hang forever because
3117 // only the foreground process was killed, not the shell, so the PTY
3118 // child never exited and wait_for_completed_task never completed.
3119 let exit_result = futures::select! {
3120 result = futures::FutureExt::fuse(wait_for_exit) => Some(result),
3121 _ = futures::FutureExt::fuse(cx.background_executor.timer(Duration::from_secs(5))) => None,
3122 };
3123
3124 assert!(
3125 exit_result.is_some(),
3126 "wait_for_exit should complete after kill, but it timed out. \
3127 This indicates kill_active_task is not properly killing the shell process."
3128 );
3129
3130 // Give the system a chance to process any pending updates
3131 cx.run_until_parked();
3132
3133 // Verify that the underlying terminal still has the output that was
3134 // written before the kill. This verifies that killing doesn't lose output.
3135 let inner_content = thread.read_with(cx, |thread, cx| {
3136 let term = thread.terminals.get(&terminal_id).unwrap();
3137 term.read(cx).inner().read(cx).get_content()
3138 });
3139
3140 assert!(
3141 inner_content.contains("output_before_kill"),
3142 "Underlying terminal should contain output from before kill, got: {}",
3143 inner_content
3144 );
3145 }
3146
3147 #[gpui::test]
3148 async fn test_push_user_content_block(cx: &mut gpui::TestAppContext) {
3149 init_test(cx);
3150
3151 let fs = FakeFs::new(cx.executor());
3152 let project = Project::test(fs, [], cx).await;
3153 let connection = Rc::new(FakeAgentConnection::new());
3154 let thread = cx
3155 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3156 .await
3157 .unwrap();
3158
3159 // Test creating a new user message
3160 thread.update(cx, |thread, cx| {
3161 thread.push_user_content_block(None, "Hello, ".into(), cx);
3162 });
3163
3164 thread.update(cx, |thread, cx| {
3165 assert_eq!(thread.entries.len(), 1);
3166 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
3167 assert_eq!(user_msg.id, None);
3168 assert_eq!(user_msg.content.to_markdown(cx), "Hello, ");
3169 } else {
3170 panic!("Expected UserMessage");
3171 }
3172 });
3173
3174 // Test appending to existing user message
3175 let message_1_id = UserMessageId::new();
3176 thread.update(cx, |thread, cx| {
3177 thread.push_user_content_block(Some(message_1_id.clone()), "world!".into(), cx);
3178 });
3179
3180 thread.update(cx, |thread, cx| {
3181 assert_eq!(thread.entries.len(), 1);
3182 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
3183 assert_eq!(user_msg.id, Some(message_1_id));
3184 assert_eq!(user_msg.content.to_markdown(cx), "Hello, world!");
3185 } else {
3186 panic!("Expected UserMessage");
3187 }
3188 });
3189
3190 // Test creating new user message after assistant message
3191 thread.update(cx, |thread, cx| {
3192 thread.push_assistant_content_block("Assistant response".into(), false, cx);
3193 });
3194
3195 let message_2_id = UserMessageId::new();
3196 thread.update(cx, |thread, cx| {
3197 thread.push_user_content_block(
3198 Some(message_2_id.clone()),
3199 "New user message".into(),
3200 cx,
3201 );
3202 });
3203
3204 thread.update(cx, |thread, cx| {
3205 assert_eq!(thread.entries.len(), 3);
3206 if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[2] {
3207 assert_eq!(user_msg.id, Some(message_2_id));
3208 assert_eq!(user_msg.content.to_markdown(cx), "New user message");
3209 } else {
3210 panic!("Expected UserMessage at index 2");
3211 }
3212 });
3213 }
3214
3215 #[gpui::test]
3216 async fn test_thinking_concatenation(cx: &mut gpui::TestAppContext) {
3217 init_test(cx);
3218
3219 let fs = FakeFs::new(cx.executor());
3220 let project = Project::test(fs, [], cx).await;
3221 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
3222 |_, thread, mut cx| {
3223 async move {
3224 thread.update(&mut cx, |thread, cx| {
3225 thread
3226 .handle_session_update(
3227 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(
3228 "Thinking ".into(),
3229 )),
3230 cx,
3231 )
3232 .unwrap();
3233 thread
3234 .handle_session_update(
3235 acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new(
3236 "hard!".into(),
3237 )),
3238 cx,
3239 )
3240 .unwrap();
3241 })?;
3242 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3243 }
3244 .boxed_local()
3245 },
3246 ));
3247
3248 let thread = cx
3249 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3250 .await
3251 .unwrap();
3252
3253 thread
3254 .update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
3255 .await
3256 .unwrap();
3257
3258 let output = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
3259 assert_eq!(
3260 output,
3261 indoc! {r#"
3262 ## User
3263
3264 Hello from Zed!
3265
3266 ## Assistant
3267
3268 <thinking>
3269 Thinking hard!
3270 </thinking>
3271
3272 "#}
3273 );
3274 }
3275
3276 #[gpui::test]
3277 async fn test_edits_concurrently_to_user(cx: &mut TestAppContext) {
3278 init_test(cx);
3279
3280 let fs = FakeFs::new(cx.executor());
3281 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\n"}))
3282 .await;
3283 let project = Project::test(fs.clone(), [], cx).await;
3284 let (read_file_tx, read_file_rx) = oneshot::channel::<()>();
3285 let read_file_tx = Rc::new(RefCell::new(Some(read_file_tx)));
3286 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
3287 move |_, thread, mut cx| {
3288 let read_file_tx = read_file_tx.clone();
3289 async move {
3290 let content = thread
3291 .update(&mut cx, |thread, cx| {
3292 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
3293 })
3294 .unwrap()
3295 .await
3296 .unwrap();
3297 assert_eq!(content, "one\ntwo\nthree\n");
3298 read_file_tx.take().unwrap().send(()).unwrap();
3299 thread
3300 .update(&mut cx, |thread, cx| {
3301 thread.write_text_file(
3302 path!("/tmp/foo").into(),
3303 "one\ntwo\nthree\nfour\nfive\n".to_string(),
3304 cx,
3305 )
3306 })
3307 .unwrap()
3308 .await
3309 .unwrap();
3310 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3311 }
3312 .boxed_local()
3313 },
3314 ));
3315
3316 let (worktree, pathbuf) = project
3317 .update(cx, |project, cx| {
3318 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
3319 })
3320 .await
3321 .unwrap();
3322 let buffer = project
3323 .update(cx, |project, cx| {
3324 project.open_buffer((worktree.read(cx).id(), pathbuf), cx)
3325 })
3326 .await
3327 .unwrap();
3328
3329 let thread = cx
3330 .update(|cx| connection.new_session(project, Path::new(path!("/tmp")), cx))
3331 .await
3332 .unwrap();
3333
3334 let request = thread.update(cx, |thread, cx| {
3335 thread.send_raw("Extend the count in /tmp/foo", cx)
3336 });
3337 read_file_rx.await.ok();
3338 buffer.update(cx, |buffer, cx| {
3339 buffer.edit([(0..0, "zero\n".to_string())], None, cx);
3340 });
3341 cx.run_until_parked();
3342 assert_eq!(
3343 buffer.read_with(cx, |buffer, _| buffer.text()),
3344 "zero\none\ntwo\nthree\nfour\nfive\n"
3345 );
3346 assert_eq!(
3347 String::from_utf8(fs.read_file_sync(path!("/tmp/foo")).unwrap()).unwrap(),
3348 "zero\none\ntwo\nthree\nfour\nfive\n"
3349 );
3350 request.await.unwrap();
3351 }
3352
3353 #[gpui::test]
3354 async fn test_reading_from_line(cx: &mut TestAppContext) {
3355 init_test(cx);
3356
3357 let fs = FakeFs::new(cx.executor());
3358 fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\nfour\n"}))
3359 .await;
3360 let project = Project::test(fs.clone(), [], cx).await;
3361 project
3362 .update(cx, |project, cx| {
3363 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
3364 })
3365 .await
3366 .unwrap();
3367
3368 let connection = Rc::new(FakeAgentConnection::new());
3369
3370 let thread = cx
3371 .update(|cx| connection.new_session(project, Path::new(path!("/tmp")), cx))
3372 .await
3373 .unwrap();
3374
3375 // Whole file
3376 let content = thread
3377 .update(cx, |thread, cx| {
3378 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
3379 })
3380 .await
3381 .unwrap();
3382
3383 assert_eq!(content, "one\ntwo\nthree\nfour\n");
3384
3385 // Only start line
3386 let content = thread
3387 .update(cx, |thread, cx| {
3388 thread.read_text_file(path!("/tmp/foo").into(), Some(3), None, false, cx)
3389 })
3390 .await
3391 .unwrap();
3392
3393 assert_eq!(content, "three\nfour\n");
3394
3395 // Only limit
3396 let content = thread
3397 .update(cx, |thread, cx| {
3398 thread.read_text_file(path!("/tmp/foo").into(), None, Some(2), false, cx)
3399 })
3400 .await
3401 .unwrap();
3402
3403 assert_eq!(content, "one\ntwo\n");
3404
3405 // Range
3406 let content = thread
3407 .update(cx, |thread, cx| {
3408 thread.read_text_file(path!("/tmp/foo").into(), Some(2), Some(2), false, cx)
3409 })
3410 .await
3411 .unwrap();
3412
3413 assert_eq!(content, "two\nthree\n");
3414
3415 // Invalid
3416 let err = thread
3417 .update(cx, |thread, cx| {
3418 thread.read_text_file(path!("/tmp/foo").into(), Some(6), Some(2), false, cx)
3419 })
3420 .await
3421 .unwrap_err();
3422
3423 assert_eq!(
3424 err.to_string(),
3425 "Invalid params: \"Attempting to read beyond the end of the file, line 5:0\""
3426 );
3427 }
3428
3429 #[gpui::test]
3430 async fn test_reading_empty_file(cx: &mut TestAppContext) {
3431 init_test(cx);
3432
3433 let fs = FakeFs::new(cx.executor());
3434 fs.insert_tree(path!("/tmp"), json!({"foo": ""})).await;
3435 let project = Project::test(fs.clone(), [], cx).await;
3436 project
3437 .update(cx, |project, cx| {
3438 project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
3439 })
3440 .await
3441 .unwrap();
3442
3443 let connection = Rc::new(FakeAgentConnection::new());
3444
3445 let thread = cx
3446 .update(|cx| connection.new_session(project, Path::new(path!("/tmp")), cx))
3447 .await
3448 .unwrap();
3449
3450 // Whole file
3451 let content = thread
3452 .update(cx, |thread, cx| {
3453 thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
3454 })
3455 .await
3456 .unwrap();
3457
3458 assert_eq!(content, "");
3459
3460 // Only start line
3461 let content = thread
3462 .update(cx, |thread, cx| {
3463 thread.read_text_file(path!("/tmp/foo").into(), Some(1), None, false, cx)
3464 })
3465 .await
3466 .unwrap();
3467
3468 assert_eq!(content, "");
3469
3470 // Only limit
3471 let content = thread
3472 .update(cx, |thread, cx| {
3473 thread.read_text_file(path!("/tmp/foo").into(), None, Some(2), false, cx)
3474 })
3475 .await
3476 .unwrap();
3477
3478 assert_eq!(content, "");
3479
3480 // Range
3481 let content = thread
3482 .update(cx, |thread, cx| {
3483 thread.read_text_file(path!("/tmp/foo").into(), Some(1), Some(1), false, cx)
3484 })
3485 .await
3486 .unwrap();
3487
3488 assert_eq!(content, "");
3489
3490 // Invalid
3491 let err = thread
3492 .update(cx, |thread, cx| {
3493 thread.read_text_file(path!("/tmp/foo").into(), Some(5), Some(2), false, cx)
3494 })
3495 .await
3496 .unwrap_err();
3497
3498 assert_eq!(
3499 err.to_string(),
3500 "Invalid params: \"Attempting to read beyond the end of the file, line 1:0\""
3501 );
3502 }
3503 #[gpui::test]
3504 async fn test_reading_non_existing_file(cx: &mut TestAppContext) {
3505 init_test(cx);
3506
3507 let fs = FakeFs::new(cx.executor());
3508 fs.insert_tree(path!("/tmp"), json!({})).await;
3509 let project = Project::test(fs.clone(), [], cx).await;
3510 project
3511 .update(cx, |project, cx| {
3512 project.find_or_create_worktree(path!("/tmp"), true, cx)
3513 })
3514 .await
3515 .unwrap();
3516
3517 let connection = Rc::new(FakeAgentConnection::new());
3518
3519 let thread = cx
3520 .update(|cx| connection.new_session(project, Path::new(path!("/tmp")), cx))
3521 .await
3522 .unwrap();
3523
3524 // Out of project file
3525 let err = thread
3526 .update(cx, |thread, cx| {
3527 thread.read_text_file(path!("/foo").into(), None, None, false, cx)
3528 })
3529 .await
3530 .unwrap_err();
3531
3532 assert_eq!(err.code, acp::ErrorCode::ResourceNotFound);
3533 }
3534
3535 #[gpui::test]
3536 async fn test_succeeding_canceled_toolcall(cx: &mut TestAppContext) {
3537 init_test(cx);
3538
3539 let fs = FakeFs::new(cx.executor());
3540 let project = Project::test(fs, [], cx).await;
3541 let id = acp::ToolCallId::new("test");
3542
3543 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3544 let id = id.clone();
3545 move |_, thread, mut cx| {
3546 let id = id.clone();
3547 async move {
3548 thread
3549 .update(&mut cx, |thread, cx| {
3550 thread.handle_session_update(
3551 acp::SessionUpdate::ToolCall(
3552 acp::ToolCall::new(id.clone(), "Label")
3553 .kind(acp::ToolKind::Fetch)
3554 .status(acp::ToolCallStatus::InProgress),
3555 ),
3556 cx,
3557 )
3558 })
3559 .unwrap()
3560 .unwrap();
3561 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3562 }
3563 .boxed_local()
3564 }
3565 }));
3566
3567 let thread = cx
3568 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3569 .await
3570 .unwrap();
3571
3572 let request = thread.update(cx, |thread, cx| {
3573 thread.send_raw("Fetch https://example.com", cx)
3574 });
3575
3576 run_until_first_tool_call(&thread, cx).await;
3577
3578 thread.read_with(cx, |thread, _| {
3579 assert!(matches!(
3580 thread.entries[1],
3581 AgentThreadEntry::ToolCall(ToolCall {
3582 status: ToolCallStatus::InProgress,
3583 ..
3584 })
3585 ));
3586 });
3587
3588 thread.update(cx, |thread, cx| thread.cancel(cx)).await;
3589
3590 thread.read_with(cx, |thread, _| {
3591 assert!(matches!(
3592 &thread.entries[1],
3593 AgentThreadEntry::ToolCall(ToolCall {
3594 status: ToolCallStatus::Canceled,
3595 ..
3596 })
3597 ));
3598 });
3599
3600 thread
3601 .update(cx, |thread, cx| {
3602 thread.handle_session_update(
3603 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
3604 id,
3605 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
3606 )),
3607 cx,
3608 )
3609 })
3610 .unwrap();
3611
3612 request.await.unwrap();
3613
3614 thread.read_with(cx, |thread, _| {
3615 assert!(matches!(
3616 thread.entries[1],
3617 AgentThreadEntry::ToolCall(ToolCall {
3618 status: ToolCallStatus::Completed,
3619 ..
3620 })
3621 ));
3622 });
3623 }
3624
3625 #[gpui::test]
3626 async fn test_no_pending_edits_if_tool_calls_are_completed(cx: &mut TestAppContext) {
3627 init_test(cx);
3628 let fs = FakeFs::new(cx.background_executor.clone());
3629 fs.insert_tree(path!("/test"), json!({})).await;
3630 let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
3631
3632 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3633 move |_, thread, mut cx| {
3634 async move {
3635 thread
3636 .update(&mut cx, |thread, cx| {
3637 thread.handle_session_update(
3638 acp::SessionUpdate::ToolCall(
3639 acp::ToolCall::new("test", "Label")
3640 .kind(acp::ToolKind::Edit)
3641 .status(acp::ToolCallStatus::Completed)
3642 .content(vec![acp::ToolCallContent::Diff(acp::Diff::new(
3643 "/test/test.txt",
3644 "foo",
3645 ))]),
3646 ),
3647 cx,
3648 )
3649 })
3650 .unwrap()
3651 .unwrap();
3652 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3653 }
3654 .boxed_local()
3655 }
3656 }));
3657
3658 let thread = cx
3659 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3660 .await
3661 .unwrap();
3662
3663 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Hi".into()], cx)))
3664 .await
3665 .unwrap();
3666
3667 assert!(cx.read(|cx| !thread.read(cx).has_pending_edit_tool_calls()));
3668 }
3669
3670 #[gpui::test(iterations = 10)]
3671 async fn test_checkpoints(cx: &mut TestAppContext) {
3672 init_test(cx);
3673 let fs = FakeFs::new(cx.background_executor.clone());
3674 fs.insert_tree(
3675 path!("/test"),
3676 json!({
3677 ".git": {}
3678 }),
3679 )
3680 .await;
3681 let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
3682
3683 let simulate_changes = Arc::new(AtomicBool::new(true));
3684 let next_filename = Arc::new(AtomicUsize::new(0));
3685 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3686 let simulate_changes = simulate_changes.clone();
3687 let next_filename = next_filename.clone();
3688 let fs = fs.clone();
3689 move |request, thread, mut cx| {
3690 let fs = fs.clone();
3691 let simulate_changes = simulate_changes.clone();
3692 let next_filename = next_filename.clone();
3693 async move {
3694 if simulate_changes.load(SeqCst) {
3695 let filename = format!("/test/file-{}", next_filename.fetch_add(1, SeqCst));
3696 fs.write(Path::new(&filename), b"").await?;
3697 }
3698
3699 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
3700 panic!("expected text content block");
3701 };
3702 thread.update(&mut cx, |thread, cx| {
3703 thread
3704 .handle_session_update(
3705 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(
3706 content.text.to_uppercase().into(),
3707 )),
3708 cx,
3709 )
3710 .unwrap();
3711 })?;
3712 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3713 }
3714 .boxed_local()
3715 }
3716 }));
3717 let thread = cx
3718 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3719 .await
3720 .unwrap();
3721
3722 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Lorem".into()], cx)))
3723 .await
3724 .unwrap();
3725 thread.read_with(cx, |thread, cx| {
3726 assert_eq!(
3727 thread.to_markdown(cx),
3728 indoc! {"
3729 ## User (checkpoint)
3730
3731 Lorem
3732
3733 ## Assistant
3734
3735 LOREM
3736
3737 "}
3738 );
3739 });
3740 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
3741
3742 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["ipsum".into()], cx)))
3743 .await
3744 .unwrap();
3745 thread.read_with(cx, |thread, cx| {
3746 assert_eq!(
3747 thread.to_markdown(cx),
3748 indoc! {"
3749 ## User (checkpoint)
3750
3751 Lorem
3752
3753 ## Assistant
3754
3755 LOREM
3756
3757 ## User (checkpoint)
3758
3759 ipsum
3760
3761 ## Assistant
3762
3763 IPSUM
3764
3765 "}
3766 );
3767 });
3768 assert_eq!(
3769 fs.files(),
3770 vec![
3771 Path::new(path!("/test/file-0")),
3772 Path::new(path!("/test/file-1"))
3773 ]
3774 );
3775
3776 // Checkpoint isn't stored when there are no changes.
3777 simulate_changes.store(false, SeqCst);
3778 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["dolor".into()], cx)))
3779 .await
3780 .unwrap();
3781 thread.read_with(cx, |thread, cx| {
3782 assert_eq!(
3783 thread.to_markdown(cx),
3784 indoc! {"
3785 ## User (checkpoint)
3786
3787 Lorem
3788
3789 ## Assistant
3790
3791 LOREM
3792
3793 ## User (checkpoint)
3794
3795 ipsum
3796
3797 ## Assistant
3798
3799 IPSUM
3800
3801 ## User
3802
3803 dolor
3804
3805 ## Assistant
3806
3807 DOLOR
3808
3809 "}
3810 );
3811 });
3812 assert_eq!(
3813 fs.files(),
3814 vec![
3815 Path::new(path!("/test/file-0")),
3816 Path::new(path!("/test/file-1"))
3817 ]
3818 );
3819
3820 // Rewinding the conversation truncates the history and restores the checkpoint.
3821 thread
3822 .update(cx, |thread, cx| {
3823 let AgentThreadEntry::UserMessage(message) = &thread.entries[2] else {
3824 panic!("unexpected entries {:?}", thread.entries)
3825 };
3826 thread.restore_checkpoint(message.id.clone().unwrap(), cx)
3827 })
3828 .await
3829 .unwrap();
3830 thread.read_with(cx, |thread, cx| {
3831 assert_eq!(
3832 thread.to_markdown(cx),
3833 indoc! {"
3834 ## User (checkpoint)
3835
3836 Lorem
3837
3838 ## Assistant
3839
3840 LOREM
3841
3842 "}
3843 );
3844 });
3845 assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
3846 }
3847
3848 #[gpui::test]
3849 async fn test_tool_result_refusal(cx: &mut TestAppContext) {
3850 use std::sync::atomic::AtomicUsize;
3851 init_test(cx);
3852
3853 let fs = FakeFs::new(cx.executor());
3854 let project = Project::test(fs, None, cx).await;
3855
3856 // Create a connection that simulates refusal after tool result
3857 let prompt_count = Arc::new(AtomicUsize::new(0));
3858 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3859 let prompt_count = prompt_count.clone();
3860 move |_request, thread, mut cx| {
3861 let count = prompt_count.fetch_add(1, SeqCst);
3862 async move {
3863 if count == 0 {
3864 // First prompt: Generate a tool call with result
3865 thread.update(&mut cx, |thread, cx| {
3866 thread
3867 .handle_session_update(
3868 acp::SessionUpdate::ToolCall(
3869 acp::ToolCall::new("tool1", "Test Tool")
3870 .kind(acp::ToolKind::Fetch)
3871 .status(acp::ToolCallStatus::Completed)
3872 .raw_input(serde_json::json!({"query": "test"}))
3873 .raw_output(serde_json::json!({"result": "inappropriate content"})),
3874 ),
3875 cx,
3876 )
3877 .unwrap();
3878 })?;
3879
3880 // Now return refusal because of the tool result
3881 Ok(acp::PromptResponse::new(acp::StopReason::Refusal))
3882 } else {
3883 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
3884 }
3885 }
3886 .boxed_local()
3887 }
3888 }));
3889
3890 let thread = cx
3891 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3892 .await
3893 .unwrap();
3894
3895 // Track if we see a Refusal event
3896 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
3897 let saw_refusal_event_captured = saw_refusal_event.clone();
3898 thread.update(cx, |_thread, cx| {
3899 cx.subscribe(
3900 &thread,
3901 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
3902 if matches!(event, AcpThreadEvent::Refusal) {
3903 *saw_refusal_event_captured.lock().unwrap() = true;
3904 }
3905 },
3906 )
3907 .detach();
3908 });
3909
3910 // Send a user message - this will trigger tool call and then refusal
3911 let send_task = thread.update(cx, |thread, cx| thread.send(vec!["Hello".into()], cx));
3912 cx.background_executor.spawn(send_task).detach();
3913 cx.run_until_parked();
3914
3915 // Verify that:
3916 // 1. A Refusal event WAS emitted (because it's a tool result refusal, not user prompt)
3917 // 2. The user message was NOT truncated
3918 assert!(
3919 *saw_refusal_event.lock().unwrap(),
3920 "Refusal event should be emitted for tool result refusals"
3921 );
3922
3923 thread.read_with(cx, |thread, _| {
3924 let entries = thread.entries();
3925 assert!(entries.len() >= 2, "Should have user message and tool call");
3926
3927 // Verify user message is still there
3928 assert!(
3929 matches!(entries[0], AgentThreadEntry::UserMessage(_)),
3930 "User message should not be truncated"
3931 );
3932
3933 // Verify tool call is there with result
3934 if let AgentThreadEntry::ToolCall(tool_call) = &entries[1] {
3935 assert!(
3936 tool_call.raw_output.is_some(),
3937 "Tool call should have output"
3938 );
3939 } else {
3940 panic!("Expected tool call at index 1");
3941 }
3942 });
3943 }
3944
3945 #[gpui::test]
3946 async fn test_user_prompt_refusal_emits_event(cx: &mut TestAppContext) {
3947 init_test(cx);
3948
3949 let fs = FakeFs::new(cx.executor());
3950 let project = Project::test(fs, None, cx).await;
3951
3952 let refuse_next = Arc::new(AtomicBool::new(false));
3953 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
3954 let refuse_next = refuse_next.clone();
3955 move |_request, _thread, _cx| {
3956 if refuse_next.load(SeqCst) {
3957 async move { Ok(acp::PromptResponse::new(acp::StopReason::Refusal)) }
3958 .boxed_local()
3959 } else {
3960 async move { Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)) }
3961 .boxed_local()
3962 }
3963 }
3964 }));
3965
3966 let thread = cx
3967 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
3968 .await
3969 .unwrap();
3970
3971 // Track if we see a Refusal event
3972 let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
3973 let saw_refusal_event_captured = saw_refusal_event.clone();
3974 thread.update(cx, |_thread, cx| {
3975 cx.subscribe(
3976 &thread,
3977 move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
3978 if matches!(event, AcpThreadEvent::Refusal) {
3979 *saw_refusal_event_captured.lock().unwrap() = true;
3980 }
3981 },
3982 )
3983 .detach();
3984 });
3985
3986 // Send a message that will be refused
3987 refuse_next.store(true, SeqCst);
3988 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
3989 .await
3990 .unwrap();
3991
3992 // Verify that a Refusal event WAS emitted for user prompt refusal
3993 assert!(
3994 *saw_refusal_event.lock().unwrap(),
3995 "Refusal event should be emitted for user prompt refusals"
3996 );
3997
3998 // Verify the message was truncated (user prompt refusal)
3999 thread.read_with(cx, |thread, cx| {
4000 assert_eq!(thread.to_markdown(cx), "");
4001 });
4002 }
4003
4004 #[gpui::test]
4005 async fn test_refusal(cx: &mut TestAppContext) {
4006 init_test(cx);
4007 let fs = FakeFs::new(cx.background_executor.clone());
4008 fs.insert_tree(path!("/"), json!({})).await;
4009 let project = Project::test(fs.clone(), [path!("/").as_ref()], cx).await;
4010
4011 let refuse_next = Arc::new(AtomicBool::new(false));
4012 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
4013 let refuse_next = refuse_next.clone();
4014 move |request, thread, mut cx| {
4015 let refuse_next = refuse_next.clone();
4016 async move {
4017 if refuse_next.load(SeqCst) {
4018 return Ok(acp::PromptResponse::new(acp::StopReason::Refusal));
4019 }
4020
4021 let acp::ContentBlock::Text(content) = &request.prompt[0] else {
4022 panic!("expected text content block");
4023 };
4024 thread.update(&mut cx, |thread, cx| {
4025 thread
4026 .handle_session_update(
4027 acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(
4028 content.text.to_uppercase().into(),
4029 )),
4030 cx,
4031 )
4032 .unwrap();
4033 })?;
4034 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
4035 }
4036 .boxed_local()
4037 }
4038 }));
4039 let thread = cx
4040 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4041 .await
4042 .unwrap();
4043
4044 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
4045 .await
4046 .unwrap();
4047 thread.read_with(cx, |thread, cx| {
4048 assert_eq!(
4049 thread.to_markdown(cx),
4050 indoc! {"
4051 ## User
4052
4053 hello
4054
4055 ## Assistant
4056
4057 HELLO
4058
4059 "}
4060 );
4061 });
4062
4063 // Simulate refusing the second message. The message should be truncated
4064 // when a user prompt is refused.
4065 refuse_next.store(true, SeqCst);
4066 cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["world".into()], cx)))
4067 .await
4068 .unwrap();
4069 thread.read_with(cx, |thread, cx| {
4070 assert_eq!(
4071 thread.to_markdown(cx),
4072 indoc! {"
4073 ## User
4074
4075 hello
4076
4077 ## Assistant
4078
4079 HELLO
4080
4081 "}
4082 );
4083 });
4084 }
4085
4086 async fn run_until_first_tool_call(
4087 thread: &Entity<AcpThread>,
4088 cx: &mut TestAppContext,
4089 ) -> usize {
4090 let (mut tx, mut rx) = mpsc::channel::<usize>(1);
4091
4092 let subscription = cx.update(|cx| {
4093 cx.subscribe(thread, move |thread, _, cx| {
4094 for (ix, entry) in thread.read(cx).entries.iter().enumerate() {
4095 if matches!(entry, AgentThreadEntry::ToolCall(_)) {
4096 return tx.try_send(ix).unwrap();
4097 }
4098 }
4099 })
4100 });
4101
4102 select! {
4103 _ = futures::FutureExt::fuse(cx.background_executor.timer(Duration::from_secs(10))) => {
4104 panic!("Timeout waiting for tool call")
4105 }
4106 ix = rx.next().fuse() => {
4107 drop(subscription);
4108 ix.unwrap()
4109 }
4110 }
4111 }
4112
4113 #[derive(Clone, Default)]
4114 struct FakeAgentConnection {
4115 auth_methods: Vec<acp::AuthMethod>,
4116 sessions: Arc<parking_lot::Mutex<HashMap<acp::SessionId, WeakEntity<AcpThread>>>>,
4117 set_title_calls: Rc<RefCell<Vec<SharedString>>>,
4118 on_user_message: Option<
4119 Rc<
4120 dyn Fn(
4121 acp::PromptRequest,
4122 WeakEntity<AcpThread>,
4123 AsyncApp,
4124 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
4125 + 'static,
4126 >,
4127 >,
4128 }
4129
4130 impl FakeAgentConnection {
4131 fn new() -> Self {
4132 Self {
4133 auth_methods: Vec::new(),
4134 on_user_message: None,
4135 sessions: Arc::default(),
4136 set_title_calls: Default::default(),
4137 }
4138 }
4139
4140 #[expect(unused)]
4141 fn with_auth_methods(mut self, auth_methods: Vec<acp::AuthMethod>) -> Self {
4142 self.auth_methods = auth_methods;
4143 self
4144 }
4145
4146 fn on_user_message(
4147 mut self,
4148 handler: impl Fn(
4149 acp::PromptRequest,
4150 WeakEntity<AcpThread>,
4151 AsyncApp,
4152 ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
4153 + 'static,
4154 ) -> Self {
4155 self.on_user_message.replace(Rc::new(handler));
4156 self
4157 }
4158 }
4159
4160 impl AgentConnection for FakeAgentConnection {
4161 fn telemetry_id(&self) -> SharedString {
4162 "fake".into()
4163 }
4164
4165 fn auth_methods(&self) -> &[acp::AuthMethod] {
4166 &self.auth_methods
4167 }
4168
4169 fn new_session(
4170 self: Rc<Self>,
4171 project: Entity<Project>,
4172 cwd: &Path,
4173 cx: &mut App,
4174 ) -> Task<gpui::Result<Entity<AcpThread>>> {
4175 let session_id = acp::SessionId::new(
4176 rand::rng()
4177 .sample_iter(&distr::Alphanumeric)
4178 .take(7)
4179 .map(char::from)
4180 .collect::<String>(),
4181 );
4182 let action_log = cx.new(|_| ActionLog::new(project.clone()));
4183 let thread = cx.new(|cx| {
4184 AcpThread::new(
4185 None,
4186 "Test",
4187 Some(cwd.to_path_buf()),
4188 self.clone(),
4189 project,
4190 action_log,
4191 session_id.clone(),
4192 watch::Receiver::constant(
4193 acp::PromptCapabilities::new()
4194 .image(true)
4195 .audio(true)
4196 .embedded_context(true),
4197 ),
4198 cx,
4199 )
4200 });
4201 self.sessions.lock().insert(session_id, thread.downgrade());
4202 Task::ready(Ok(thread))
4203 }
4204
4205 fn authenticate(&self, method: acp::AuthMethodId, _cx: &mut App) -> Task<gpui::Result<()>> {
4206 if self.auth_methods().iter().any(|m| m.id() == &method) {
4207 Task::ready(Ok(()))
4208 } else {
4209 Task::ready(Err(anyhow!("Invalid Auth Method")))
4210 }
4211 }
4212
4213 fn prompt(
4214 &self,
4215 _id: Option<UserMessageId>,
4216 params: acp::PromptRequest,
4217 cx: &mut App,
4218 ) -> Task<gpui::Result<acp::PromptResponse>> {
4219 let sessions = self.sessions.lock();
4220 let thread = sessions.get(¶ms.session_id).unwrap();
4221 if let Some(handler) = &self.on_user_message {
4222 let handler = handler.clone();
4223 let thread = thread.clone();
4224 cx.spawn(async move |cx| handler(params, thread, cx.clone()).await)
4225 } else {
4226 Task::ready(Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)))
4227 }
4228 }
4229
4230 fn cancel(&self, _session_id: &acp::SessionId, _cx: &mut App) {}
4231
4232 fn truncate(
4233 &self,
4234 session_id: &acp::SessionId,
4235 _cx: &App,
4236 ) -> Option<Rc<dyn AgentSessionTruncate>> {
4237 Some(Rc::new(FakeAgentSessionEditor {
4238 _session_id: session_id.clone(),
4239 }))
4240 }
4241
4242 fn set_title(
4243 &self,
4244 _session_id: &acp::SessionId,
4245 _cx: &App,
4246 ) -> Option<Rc<dyn AgentSessionSetTitle>> {
4247 Some(Rc::new(FakeAgentSessionSetTitle {
4248 calls: self.set_title_calls.clone(),
4249 }))
4250 }
4251
4252 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
4253 self
4254 }
4255 }
4256
4257 struct FakeAgentSessionSetTitle {
4258 calls: Rc<RefCell<Vec<SharedString>>>,
4259 }
4260
4261 impl AgentSessionSetTitle for FakeAgentSessionSetTitle {
4262 fn run(&self, title: SharedString, _cx: &mut App) -> Task<Result<()>> {
4263 self.calls.borrow_mut().push(title);
4264 Task::ready(Ok(()))
4265 }
4266 }
4267
4268 struct FakeAgentSessionEditor {
4269 _session_id: acp::SessionId,
4270 }
4271
4272 impl AgentSessionTruncate for FakeAgentSessionEditor {
4273 fn run(&self, _message_id: UserMessageId, _cx: &mut App) -> Task<Result<()>> {
4274 Task::ready(Ok(()))
4275 }
4276 }
4277
4278 #[gpui::test]
4279 async fn test_tool_call_not_found_creates_failed_entry(cx: &mut TestAppContext) {
4280 init_test(cx);
4281
4282 let fs = FakeFs::new(cx.executor());
4283 let project = Project::test(fs, [], cx).await;
4284 let connection = Rc::new(FakeAgentConnection::new());
4285 let thread = cx
4286 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4287 .await
4288 .unwrap();
4289
4290 // Try to update a tool call that doesn't exist
4291 let nonexistent_id = acp::ToolCallId::new("nonexistent-tool-call");
4292 thread.update(cx, |thread, cx| {
4293 let result = thread.handle_session_update(
4294 acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new(
4295 nonexistent_id.clone(),
4296 acp::ToolCallUpdateFields::new().status(acp::ToolCallStatus::Completed),
4297 )),
4298 cx,
4299 );
4300
4301 // The update should succeed (not return an error)
4302 assert!(result.is_ok());
4303
4304 // There should now be exactly one entry in the thread
4305 assert_eq!(thread.entries.len(), 1);
4306
4307 // The entry should be a failed tool call
4308 if let AgentThreadEntry::ToolCall(tool_call) = &thread.entries[0] {
4309 assert_eq!(tool_call.id, nonexistent_id);
4310 assert!(matches!(tool_call.status, ToolCallStatus::Failed));
4311 assert_eq!(tool_call.kind, acp::ToolKind::Fetch);
4312
4313 // Check that the content contains the error message
4314 assert_eq!(tool_call.content.len(), 1);
4315 if let ToolCallContent::ContentBlock(content_block) = &tool_call.content[0] {
4316 match content_block {
4317 ContentBlock::Markdown { markdown } => {
4318 let markdown_text = markdown.read(cx).source();
4319 assert!(markdown_text.contains("Tool call not found"));
4320 }
4321 ContentBlock::Empty => panic!("Expected markdown content, got empty"),
4322 ContentBlock::ResourceLink { .. } => {
4323 panic!("Expected markdown content, got resource link")
4324 }
4325 ContentBlock::Image { .. } => {
4326 panic!("Expected markdown content, got image")
4327 }
4328 }
4329 } else {
4330 panic!("Expected ContentBlock, got: {:?}", tool_call.content[0]);
4331 }
4332 } else {
4333 panic!("Expected ToolCall entry, got: {:?}", thread.entries[0]);
4334 }
4335 });
4336 }
4337
4338 /// Tests that restoring a checkpoint properly cleans up terminals that were
4339 /// created after that checkpoint, and cancels any in-progress generation.
4340 ///
4341 /// Reproduces issue #35142: When a checkpoint is restored, any terminal processes
4342 /// that were started after that checkpoint should be terminated, and any in-progress
4343 /// AI generation should be canceled.
4344 #[gpui::test]
4345 async fn test_restore_checkpoint_kills_terminal(cx: &mut TestAppContext) {
4346 init_test(cx);
4347
4348 let fs = FakeFs::new(cx.executor());
4349 let project = Project::test(fs, [], cx).await;
4350 let connection = Rc::new(FakeAgentConnection::new());
4351 let thread = cx
4352 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4353 .await
4354 .unwrap();
4355
4356 // Send first user message to create a checkpoint
4357 cx.update(|cx| {
4358 thread.update(cx, |thread, cx| {
4359 thread.send(vec!["first message".into()], cx)
4360 })
4361 })
4362 .await
4363 .unwrap();
4364
4365 // Send second message (creates another checkpoint) - we'll restore to this one
4366 cx.update(|cx| {
4367 thread.update(cx, |thread, cx| {
4368 thread.send(vec!["second message".into()], cx)
4369 })
4370 })
4371 .await
4372 .unwrap();
4373
4374 // Create 2 terminals BEFORE the checkpoint that have completed running
4375 let terminal_id_1 = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
4376 let mock_terminal_1 = cx.new(|cx| {
4377 let builder = ::terminal::TerminalBuilder::new_display_only(
4378 ::terminal::terminal_settings::CursorShape::default(),
4379 ::terminal::terminal_settings::AlternateScroll::On,
4380 None,
4381 0,
4382 cx.background_executor(),
4383 PathStyle::local(),
4384 )
4385 .unwrap();
4386 builder.subscribe(cx)
4387 });
4388
4389 thread.update(cx, |thread, cx| {
4390 thread.on_terminal_provider_event(
4391 TerminalProviderEvent::Created {
4392 terminal_id: terminal_id_1.clone(),
4393 label: "echo 'first'".to_string(),
4394 cwd: Some(PathBuf::from("/test")),
4395 output_byte_limit: None,
4396 terminal: mock_terminal_1.clone(),
4397 },
4398 cx,
4399 );
4400 });
4401
4402 thread.update(cx, |thread, cx| {
4403 thread.on_terminal_provider_event(
4404 TerminalProviderEvent::Output {
4405 terminal_id: terminal_id_1.clone(),
4406 data: b"first\n".to_vec(),
4407 },
4408 cx,
4409 );
4410 });
4411
4412 thread.update(cx, |thread, cx| {
4413 thread.on_terminal_provider_event(
4414 TerminalProviderEvent::Exit {
4415 terminal_id: terminal_id_1.clone(),
4416 status: acp::TerminalExitStatus::new().exit_code(0),
4417 },
4418 cx,
4419 );
4420 });
4421
4422 let terminal_id_2 = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
4423 let mock_terminal_2 = cx.new(|cx| {
4424 let builder = ::terminal::TerminalBuilder::new_display_only(
4425 ::terminal::terminal_settings::CursorShape::default(),
4426 ::terminal::terminal_settings::AlternateScroll::On,
4427 None,
4428 0,
4429 cx.background_executor(),
4430 PathStyle::local(),
4431 )
4432 .unwrap();
4433 builder.subscribe(cx)
4434 });
4435
4436 thread.update(cx, |thread, cx| {
4437 thread.on_terminal_provider_event(
4438 TerminalProviderEvent::Created {
4439 terminal_id: terminal_id_2.clone(),
4440 label: "echo 'second'".to_string(),
4441 cwd: Some(PathBuf::from("/test")),
4442 output_byte_limit: None,
4443 terminal: mock_terminal_2.clone(),
4444 },
4445 cx,
4446 );
4447 });
4448
4449 thread.update(cx, |thread, cx| {
4450 thread.on_terminal_provider_event(
4451 TerminalProviderEvent::Output {
4452 terminal_id: terminal_id_2.clone(),
4453 data: b"second\n".to_vec(),
4454 },
4455 cx,
4456 );
4457 });
4458
4459 thread.update(cx, |thread, cx| {
4460 thread.on_terminal_provider_event(
4461 TerminalProviderEvent::Exit {
4462 terminal_id: terminal_id_2.clone(),
4463 status: acp::TerminalExitStatus::new().exit_code(0),
4464 },
4465 cx,
4466 );
4467 });
4468
4469 // Get the second message ID to restore to
4470 let second_message_id = thread.read_with(cx, |thread, _| {
4471 // At this point we have:
4472 // - Index 0: First user message (with checkpoint)
4473 // - Index 1: Second user message (with checkpoint)
4474 // No assistant responses because FakeAgentConnection just returns EndTurn
4475 let AgentThreadEntry::UserMessage(message) = &thread.entries[1] else {
4476 panic!("expected user message at index 1");
4477 };
4478 message.id.clone().unwrap()
4479 });
4480
4481 // Create a terminal AFTER the checkpoint we'll restore to.
4482 // This simulates the AI agent starting a long-running terminal command.
4483 let terminal_id = acp::TerminalId::new(uuid::Uuid::new_v4().to_string());
4484 let mock_terminal = cx.new(|cx| {
4485 let builder = ::terminal::TerminalBuilder::new_display_only(
4486 ::terminal::terminal_settings::CursorShape::default(),
4487 ::terminal::terminal_settings::AlternateScroll::On,
4488 None,
4489 0,
4490 cx.background_executor(),
4491 PathStyle::local(),
4492 )
4493 .unwrap();
4494 builder.subscribe(cx)
4495 });
4496
4497 // Register the terminal as created
4498 thread.update(cx, |thread, cx| {
4499 thread.on_terminal_provider_event(
4500 TerminalProviderEvent::Created {
4501 terminal_id: terminal_id.clone(),
4502 label: "sleep 1000".to_string(),
4503 cwd: Some(PathBuf::from("/test")),
4504 output_byte_limit: None,
4505 terminal: mock_terminal.clone(),
4506 },
4507 cx,
4508 );
4509 });
4510
4511 // Simulate the terminal producing output (still running)
4512 thread.update(cx, |thread, cx| {
4513 thread.on_terminal_provider_event(
4514 TerminalProviderEvent::Output {
4515 terminal_id: terminal_id.clone(),
4516 data: b"terminal is running...\n".to_vec(),
4517 },
4518 cx,
4519 );
4520 });
4521
4522 // Create a tool call entry that references this terminal
4523 // This represents the agent requesting a terminal command
4524 thread.update(cx, |thread, cx| {
4525 thread
4526 .handle_session_update(
4527 acp::SessionUpdate::ToolCall(
4528 acp::ToolCall::new("terminal-tool-1", "Running command")
4529 .kind(acp::ToolKind::Execute)
4530 .status(acp::ToolCallStatus::InProgress)
4531 .content(vec![acp::ToolCallContent::Terminal(acp::Terminal::new(
4532 terminal_id.clone(),
4533 ))])
4534 .raw_input(serde_json::json!({"command": "sleep 1000", "cd": "/test"})),
4535 ),
4536 cx,
4537 )
4538 .unwrap();
4539 });
4540
4541 // Verify terminal exists and is in the thread
4542 let terminal_exists_before =
4543 thread.read_with(cx, |thread, _| thread.terminals.contains_key(&terminal_id));
4544 assert!(
4545 terminal_exists_before,
4546 "Terminal should exist before checkpoint restore"
4547 );
4548
4549 // Verify the terminal's underlying task is still running (not completed)
4550 let terminal_running_before = thread.read_with(cx, |thread, _cx| {
4551 let terminal_entity = thread.terminals.get(&terminal_id).unwrap();
4552 terminal_entity.read_with(cx, |term, _cx| {
4553 term.output().is_none() // output is None means it's still running
4554 })
4555 });
4556 assert!(
4557 terminal_running_before,
4558 "Terminal should be running before checkpoint restore"
4559 );
4560
4561 // Verify we have the expected entries before restore
4562 let entry_count_before = thread.read_with(cx, |thread, _| thread.entries.len());
4563 assert!(
4564 entry_count_before > 1,
4565 "Should have multiple entries before restore"
4566 );
4567
4568 // Restore the checkpoint to the second message.
4569 // This should:
4570 // 1. Cancel any in-progress generation (via the cancel() call)
4571 // 2. Remove the terminal that was created after that point
4572 thread
4573 .update(cx, |thread, cx| {
4574 thread.restore_checkpoint(second_message_id, cx)
4575 })
4576 .await
4577 .unwrap();
4578
4579 // Verify that no send_task is in progress after restore
4580 // (cancel() clears the send_task)
4581 let has_send_task_after = thread.read_with(cx, |thread, _| thread.running_turn.is_some());
4582 assert!(
4583 !has_send_task_after,
4584 "Should not have a send_task after restore (cancel should have cleared it)"
4585 );
4586
4587 // Verify the entries were truncated (restoring to index 1 truncates at 1, keeping only index 0)
4588 let entry_count = thread.read_with(cx, |thread, _| thread.entries.len());
4589 assert_eq!(
4590 entry_count, 1,
4591 "Should have 1 entry after restore (only the first user message)"
4592 );
4593
4594 // Verify the 2 completed terminals from before the checkpoint still exist
4595 let terminal_1_exists = thread.read_with(cx, |thread, _| {
4596 thread.terminals.contains_key(&terminal_id_1)
4597 });
4598 assert!(
4599 terminal_1_exists,
4600 "Terminal 1 (from before checkpoint) should still exist"
4601 );
4602
4603 let terminal_2_exists = thread.read_with(cx, |thread, _| {
4604 thread.terminals.contains_key(&terminal_id_2)
4605 });
4606 assert!(
4607 terminal_2_exists,
4608 "Terminal 2 (from before checkpoint) should still exist"
4609 );
4610
4611 // Verify they're still in completed state
4612 let terminal_1_completed = thread.read_with(cx, |thread, _cx| {
4613 let terminal_entity = thread.terminals.get(&terminal_id_1).unwrap();
4614 terminal_entity.read_with(cx, |term, _cx| term.output().is_some())
4615 });
4616 assert!(terminal_1_completed, "Terminal 1 should still be completed");
4617
4618 let terminal_2_completed = thread.read_with(cx, |thread, _cx| {
4619 let terminal_entity = thread.terminals.get(&terminal_id_2).unwrap();
4620 terminal_entity.read_with(cx, |term, _cx| term.output().is_some())
4621 });
4622 assert!(terminal_2_completed, "Terminal 2 should still be completed");
4623
4624 // Verify the running terminal (created after checkpoint) was removed
4625 let terminal_3_exists =
4626 thread.read_with(cx, |thread, _| thread.terminals.contains_key(&terminal_id));
4627 assert!(
4628 !terminal_3_exists,
4629 "Terminal 3 (created after checkpoint) should have been removed"
4630 );
4631
4632 // Verify total count is 2 (the two from before the checkpoint)
4633 let terminal_count = thread.read_with(cx, |thread, _| thread.terminals.len());
4634 assert_eq!(
4635 terminal_count, 2,
4636 "Should have exactly 2 terminals (the completed ones from before checkpoint)"
4637 );
4638 }
4639
4640 /// Tests that update_last_checkpoint correctly updates the original message's checkpoint
4641 /// even when a new user message is added while the async checkpoint comparison is in progress.
4642 ///
4643 /// This is a regression test for a bug where update_last_checkpoint would fail with
4644 /// "no checkpoint" if a new user message (without a checkpoint) was added between when
4645 /// update_last_checkpoint started and when its async closure ran.
4646 #[gpui::test]
4647 async fn test_update_last_checkpoint_with_new_message_added(cx: &mut TestAppContext) {
4648 init_test(cx);
4649
4650 let fs = FakeFs::new(cx.executor());
4651 fs.insert_tree(path!("/test"), json!({".git": {}, "file.txt": "content"}))
4652 .await;
4653 let project = Project::test(fs.clone(), [Path::new(path!("/test"))], cx).await;
4654
4655 let handler_done = Arc::new(AtomicBool::new(false));
4656 let handler_done_clone = handler_done.clone();
4657 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
4658 move |_, _thread, _cx| {
4659 handler_done_clone.store(true, SeqCst);
4660 async move { Ok(acp::PromptResponse::new(acp::StopReason::EndTurn)) }.boxed_local()
4661 },
4662 ));
4663
4664 let thread = cx
4665 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4666 .await
4667 .unwrap();
4668
4669 let send_future = thread.update(cx, |thread, cx| thread.send_raw("First message", cx));
4670 let send_task = cx.background_executor.spawn(send_future);
4671
4672 // Tick until handler completes, then a few more to let update_last_checkpoint start
4673 while !handler_done.load(SeqCst) {
4674 cx.executor().tick();
4675 }
4676 for _ in 0..5 {
4677 cx.executor().tick();
4678 }
4679
4680 thread.update(cx, |thread, cx| {
4681 thread.push_entry(
4682 AgentThreadEntry::UserMessage(UserMessage {
4683 id: Some(UserMessageId::new()),
4684 content: ContentBlock::Empty,
4685 chunks: vec!["Injected message (no checkpoint)".into()],
4686 checkpoint: None,
4687 indented: false,
4688 }),
4689 cx,
4690 );
4691 });
4692
4693 cx.run_until_parked();
4694 let result = send_task.await;
4695
4696 assert!(
4697 result.is_ok(),
4698 "send should succeed even when new message added during update_last_checkpoint: {:?}",
4699 result.err()
4700 );
4701 }
4702
4703 /// Tests that when a follow-up message is sent during generation,
4704 /// the first turn completing does NOT clear `running_turn` because
4705 /// it now belongs to the second turn.
4706 #[gpui::test]
4707 async fn test_follow_up_message_during_generation_does_not_clear_turn(cx: &mut TestAppContext) {
4708 init_test(cx);
4709
4710 let fs = FakeFs::new(cx.executor());
4711 let project = Project::test(fs, [], cx).await;
4712
4713 // First handler waits for this signal before completing
4714 let (first_complete_tx, first_complete_rx) = futures::channel::oneshot::channel::<()>();
4715 let first_complete_rx = RefCell::new(Some(first_complete_rx));
4716
4717 let connection = Rc::new(FakeAgentConnection::new().on_user_message({
4718 move |params, _thread, _cx| {
4719 let first_complete_rx = first_complete_rx.borrow_mut().take();
4720 let is_first = params
4721 .prompt
4722 .iter()
4723 .any(|c| matches!(c, acp::ContentBlock::Text(t) if t.text.contains("first")));
4724
4725 async move {
4726 if is_first {
4727 // First handler waits until signaled
4728 if let Some(rx) = first_complete_rx {
4729 rx.await.ok();
4730 }
4731 }
4732 Ok(acp::PromptResponse::new(acp::StopReason::EndTurn))
4733 }
4734 .boxed_local()
4735 }
4736 }));
4737
4738 let thread = cx
4739 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4740 .await
4741 .unwrap();
4742
4743 // Send first message (turn_id=1) - handler will block
4744 let first_request = thread.update(cx, |thread, cx| thread.send_raw("first", cx));
4745 assert_eq!(thread.read_with(cx, |t, _| t.turn_id), 1);
4746
4747 // Send second message (turn_id=2) while first is still blocked
4748 // This calls cancel() which takes turn 1's running_turn and sets turn 2's
4749 let second_request = thread.update(cx, |thread, cx| thread.send_raw("second", cx));
4750 assert_eq!(thread.read_with(cx, |t, _| t.turn_id), 2);
4751
4752 let running_turn_after_second_send =
4753 thread.read_with(cx, |thread, _| thread.running_turn.as_ref().map(|t| t.id));
4754 assert_eq!(
4755 running_turn_after_second_send,
4756 Some(2),
4757 "running_turn should be set to turn 2 after sending second message"
4758 );
4759
4760 // Now signal first handler to complete
4761 first_complete_tx.send(()).ok();
4762
4763 // First request completes - should NOT clear running_turn
4764 // because running_turn now belongs to turn 2
4765 first_request.await.unwrap();
4766
4767 let running_turn_after_first =
4768 thread.read_with(cx, |thread, _| thread.running_turn.as_ref().map(|t| t.id));
4769 assert_eq!(
4770 running_turn_after_first,
4771 Some(2),
4772 "first turn completing should not clear running_turn (belongs to turn 2)"
4773 );
4774
4775 // Second request completes - SHOULD clear running_turn
4776 second_request.await.unwrap();
4777
4778 let running_turn_after_second =
4779 thread.read_with(cx, |thread, _| thread.running_turn.is_some());
4780 assert!(
4781 !running_turn_after_second,
4782 "second turn completing should clear running_turn"
4783 );
4784 }
4785
4786 #[gpui::test]
4787 async fn test_send_returns_cancelled_response_and_marks_tools_as_cancelled(
4788 cx: &mut TestAppContext,
4789 ) {
4790 init_test(cx);
4791
4792 let fs = FakeFs::new(cx.executor());
4793 let project = Project::test(fs, [], cx).await;
4794
4795 let connection = Rc::new(FakeAgentConnection::new().on_user_message(
4796 move |_params, thread, mut cx| {
4797 async move {
4798 thread
4799 .update(&mut cx, |thread, cx| {
4800 thread.handle_session_update(
4801 acp::SessionUpdate::ToolCall(
4802 acp::ToolCall::new(
4803 acp::ToolCallId::new("test-tool"),
4804 "Test Tool",
4805 )
4806 .kind(acp::ToolKind::Fetch)
4807 .status(acp::ToolCallStatus::InProgress),
4808 ),
4809 cx,
4810 )
4811 })
4812 .unwrap()
4813 .unwrap();
4814
4815 Ok(acp::PromptResponse::new(acp::StopReason::Cancelled))
4816 }
4817 .boxed_local()
4818 },
4819 ));
4820
4821 let thread = cx
4822 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4823 .await
4824 .unwrap();
4825
4826 let response = thread
4827 .update(cx, |thread, cx| thread.send_raw("test message", cx))
4828 .await;
4829
4830 let response = response
4831 .expect("send should succeed")
4832 .expect("should have response");
4833 assert_eq!(
4834 response.stop_reason,
4835 acp::StopReason::Cancelled,
4836 "response should have Cancelled stop_reason"
4837 );
4838
4839 thread.read_with(cx, |thread, _| {
4840 let tool_entry = thread
4841 .entries
4842 .iter()
4843 .find_map(|e| {
4844 if let AgentThreadEntry::ToolCall(call) = e {
4845 Some(call)
4846 } else {
4847 None
4848 }
4849 })
4850 .expect("should have tool call entry");
4851
4852 assert!(
4853 matches!(tool_entry.status, ToolCallStatus::Canceled),
4854 "tool should be marked as Canceled when response is Cancelled, got {:?}",
4855 tool_entry.status
4856 );
4857 });
4858 }
4859
4860 #[gpui::test]
4861 async fn test_provisional_title_replaced_by_real_title(cx: &mut TestAppContext) {
4862 init_test(cx);
4863
4864 let fs = FakeFs::new(cx.executor());
4865 let project = Project::test(fs, [], cx).await;
4866 let connection = Rc::new(FakeAgentConnection::new());
4867 let set_title_calls = connection.set_title_calls.clone();
4868
4869 let thread = cx
4870 .update(|cx| connection.new_session(project, Path::new(path!("/test")), cx))
4871 .await
4872 .unwrap();
4873
4874 // Initial title is the default.
4875 thread.read_with(cx, |thread, _| {
4876 assert_eq!(thread.title().as_ref(), "Test");
4877 });
4878
4879 // Setting a provisional title updates the display title.
4880 thread.update(cx, |thread, cx| {
4881 thread.set_provisional_title("Hello, can you help…".into(), cx);
4882 });
4883 thread.read_with(cx, |thread, _| {
4884 assert_eq!(thread.title().as_ref(), "Hello, can you help…");
4885 });
4886
4887 // The provisional title should NOT have propagated to the connection.
4888 assert_eq!(
4889 set_title_calls.borrow().len(),
4890 0,
4891 "provisional title should not propagate to the connection"
4892 );
4893
4894 // When the real title arrives via set_title, it replaces the
4895 // provisional title and propagates to the connection.
4896 let task = thread.update(cx, |thread, cx| {
4897 thread.set_title("Helping with Rust question".into(), cx)
4898 });
4899 task.await.expect("set_title should succeed");
4900 thread.read_with(cx, |thread, _| {
4901 assert_eq!(thread.title().as_ref(), "Helping with Rust question");
4902 });
4903 assert_eq!(
4904 set_title_calls.borrow().as_slice(),
4905 &[SharedString::from("Helping with Rust question")],
4906 "real title should propagate to the connection"
4907 );
4908 }
4909}