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