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