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