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