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