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