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