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