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