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