1use crate::{
2 ContextServerRegistry, CopyPathTool, CreateDirectoryTool, DbLanguageModel, DbThread,
3 DeletePathTool, DiagnosticsTool, EditFileTool, FetchTool, FindPathTool, GrepTool,
4 ListDirectoryTool, MovePathTool, NowTool, OpenTool, ReadFileTool, SystemPromptTemplate,
5 Template, Templates, TerminalTool, ThinkingTool, WebSearchTool,
6};
7use acp_thread::{MentionUri, UserMessageId};
8use action_log::ActionLog;
9use agent::thread::{GitState, ProjectSnapshot, WorktreeSnapshot};
10use agent_client_protocol as acp;
11use agent_settings::{
12 AgentProfileId, AgentProfileSettings, AgentSettings, CompletionMode,
13 SUMMARIZE_THREAD_DETAILED_PROMPT, SUMMARIZE_THREAD_PROMPT,
14};
15use anyhow::{Context as _, Result, anyhow};
16use assistant_tool::adapt_schema_to_format;
17use chrono::{DateTime, Utc};
18use client::{ModelRequestUsage, RequestUsage};
19use cloud_llm_client::{CompletionIntent, CompletionRequestStatus, UsageLimit};
20use collections::{HashMap, HashSet, IndexMap};
21use fs::Fs;
22use futures::{
23 FutureExt,
24 channel::{mpsc, oneshot},
25 future::Shared,
26 stream::FuturesUnordered,
27};
28use git::repository::DiffType;
29use gpui::{
30 App, AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Task, WeakEntity,
31};
32use language_model::{
33 LanguageModel, LanguageModelCompletionError, LanguageModelCompletionEvent, LanguageModelExt,
34 LanguageModelImage, LanguageModelProviderId, LanguageModelRegistry, LanguageModelRequest,
35 LanguageModelRequestMessage, LanguageModelRequestTool, LanguageModelToolResult,
36 LanguageModelToolResultContent, LanguageModelToolSchemaFormat, LanguageModelToolUse,
37 LanguageModelToolUseId, Role, SelectedModel, StopReason, TokenUsage,
38};
39use project::{
40 Project,
41 git_store::{GitStore, RepositoryState},
42};
43use prompt_store::ProjectContext;
44use schemars::{JsonSchema, Schema};
45use serde::{Deserialize, Serialize};
46use settings::{Settings, update_settings_file};
47use smol::stream::StreamExt;
48use std::fmt::Write;
49use std::{
50 collections::BTreeMap,
51 ops::RangeInclusive,
52 path::Path,
53 sync::Arc,
54 time::{Duration, Instant},
55};
56use util::{ResultExt, debug_panic, markdown::MarkdownCodeBlock};
57use uuid::Uuid;
58
59const TOOL_CANCELED_MESSAGE: &str = "Tool canceled by user";
60pub const MAX_TOOL_NAME_LENGTH: usize = 64;
61
62/// The ID of the user prompt that initiated a request.
63///
64/// This equates to the user physically submitting a message to the model (e.g., by pressing the Enter key).
65#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
66pub struct PromptId(Arc<str>);
67
68impl PromptId {
69 pub fn new() -> Self {
70 Self(Uuid::new_v4().to_string().into())
71 }
72}
73
74impl std::fmt::Display for PromptId {
75 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76 write!(f, "{}", self.0)
77 }
78}
79
80pub(crate) const MAX_RETRY_ATTEMPTS: u8 = 4;
81pub(crate) const BASE_RETRY_DELAY: Duration = Duration::from_secs(5);
82
83#[derive(Debug, Clone)]
84enum RetryStrategy {
85 ExponentialBackoff {
86 initial_delay: Duration,
87 max_attempts: u8,
88 },
89 Fixed {
90 delay: Duration,
91 max_attempts: u8,
92 },
93}
94
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96pub enum Message {
97 User(UserMessage),
98 Agent(AgentMessage),
99 Resume,
100}
101
102impl Message {
103 pub fn as_agent_message(&self) -> Option<&AgentMessage> {
104 match self {
105 Message::Agent(agent_message) => Some(agent_message),
106 _ => None,
107 }
108 }
109
110 pub fn to_request(&self) -> Vec<LanguageModelRequestMessage> {
111 match self {
112 Message::User(message) => vec![message.to_request()],
113 Message::Agent(message) => message.to_request(),
114 Message::Resume => vec![LanguageModelRequestMessage {
115 role: Role::User,
116 content: vec!["Continue where you left off".into()],
117 cache: false,
118 }],
119 }
120 }
121
122 pub fn to_markdown(&self) -> String {
123 match self {
124 Message::User(message) => message.to_markdown(),
125 Message::Agent(message) => message.to_markdown(),
126 Message::Resume => "[resume]\n".into(),
127 }
128 }
129
130 pub fn role(&self) -> Role {
131 match self {
132 Message::User(_) | Message::Resume => Role::User,
133 Message::Agent(_) => Role::Assistant,
134 }
135 }
136}
137
138#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
139pub struct UserMessage {
140 pub id: UserMessageId,
141 pub content: Vec<UserMessageContent>,
142}
143
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
145pub enum UserMessageContent {
146 Text(String),
147 Mention { uri: MentionUri, content: String },
148 Image(LanguageModelImage),
149}
150
151impl UserMessage {
152 pub fn to_markdown(&self) -> String {
153 let mut markdown = String::from("## User\n\n");
154
155 for content in &self.content {
156 match content {
157 UserMessageContent::Text(text) => {
158 markdown.push_str(text);
159 markdown.push('\n');
160 }
161 UserMessageContent::Image(_) => {
162 markdown.push_str("<image />\n");
163 }
164 UserMessageContent::Mention { uri, content } => {
165 if !content.is_empty() {
166 let _ = writeln!(&mut markdown, "{}\n\n{}", uri.as_link(), content);
167 } else {
168 let _ = writeln!(&mut markdown, "{}", uri.as_link());
169 }
170 }
171 }
172 }
173
174 markdown
175 }
176
177 fn to_request(&self) -> LanguageModelRequestMessage {
178 let mut message = LanguageModelRequestMessage {
179 role: Role::User,
180 content: Vec::with_capacity(self.content.len()),
181 cache: false,
182 };
183
184 const OPEN_CONTEXT: &str = "<context>\n\
185 The following items were attached by the user. \
186 They are up-to-date and don't need to be re-read.\n\n";
187
188 const OPEN_FILES_TAG: &str = "<files>";
189 const OPEN_DIRECTORIES_TAG: &str = "<directories>";
190 const OPEN_SYMBOLS_TAG: &str = "<symbols>";
191 const OPEN_SELECTIONS_TAG: &str = "<selections>";
192 const OPEN_THREADS_TAG: &str = "<threads>";
193 const OPEN_FETCH_TAG: &str = "<fetched_urls>";
194 const OPEN_RULES_TAG: &str =
195 "<rules>\nThe user has specified the following rules that should be applied:\n";
196
197 let mut file_context = OPEN_FILES_TAG.to_string();
198 let mut directory_context = OPEN_DIRECTORIES_TAG.to_string();
199 let mut symbol_context = OPEN_SYMBOLS_TAG.to_string();
200 let mut selection_context = OPEN_SELECTIONS_TAG.to_string();
201 let mut thread_context = OPEN_THREADS_TAG.to_string();
202 let mut fetch_context = OPEN_FETCH_TAG.to_string();
203 let mut rules_context = OPEN_RULES_TAG.to_string();
204
205 for chunk in &self.content {
206 let chunk = match chunk {
207 UserMessageContent::Text(text) => {
208 language_model::MessageContent::Text(text.clone())
209 }
210 UserMessageContent::Image(value) => {
211 language_model::MessageContent::Image(value.clone())
212 }
213 UserMessageContent::Mention { uri, content } => {
214 match uri {
215 MentionUri::File { abs_path } => {
216 write!(
217 &mut file_context,
218 "\n{}",
219 MarkdownCodeBlock {
220 tag: &codeblock_tag(abs_path, None),
221 text: &content.to_string(),
222 }
223 )
224 .ok();
225 }
226 MentionUri::PastedImage => {
227 debug_panic!("pasted image URI should not be used in mention content")
228 }
229 MentionUri::Directory { .. } => {
230 write!(&mut directory_context, "\n{}\n", content).ok();
231 }
232 MentionUri::Symbol {
233 abs_path: path,
234 line_range,
235 ..
236 } => {
237 write!(
238 &mut symbol_context,
239 "\n{}",
240 MarkdownCodeBlock {
241 tag: &codeblock_tag(path, Some(line_range)),
242 text: content
243 }
244 )
245 .ok();
246 }
247 MentionUri::Selection {
248 abs_path: path,
249 line_range,
250 ..
251 } => {
252 write!(
253 &mut selection_context,
254 "\n{}",
255 MarkdownCodeBlock {
256 tag: &codeblock_tag(
257 path.as_deref().unwrap_or("Untitled".as_ref()),
258 Some(line_range)
259 ),
260 text: content
261 }
262 )
263 .ok();
264 }
265 MentionUri::Thread { .. } => {
266 write!(&mut thread_context, "\n{}\n", content).ok();
267 }
268 MentionUri::TextThread { .. } => {
269 write!(&mut thread_context, "\n{}\n", content).ok();
270 }
271 MentionUri::Rule { .. } => {
272 write!(
273 &mut rules_context,
274 "\n{}",
275 MarkdownCodeBlock {
276 tag: "",
277 text: content
278 }
279 )
280 .ok();
281 }
282 MentionUri::Fetch { url } => {
283 write!(&mut fetch_context, "\nFetch: {}\n\n{}", url, content).ok();
284 }
285 }
286
287 language_model::MessageContent::Text(uri.as_link().to_string())
288 }
289 };
290
291 message.content.push(chunk);
292 }
293
294 let len_before_context = message.content.len();
295
296 if file_context.len() > OPEN_FILES_TAG.len() {
297 file_context.push_str("</files>\n");
298 message
299 .content
300 .push(language_model::MessageContent::Text(file_context));
301 }
302
303 if directory_context.len() > OPEN_DIRECTORIES_TAG.len() {
304 directory_context.push_str("</directories>\n");
305 message
306 .content
307 .push(language_model::MessageContent::Text(directory_context));
308 }
309
310 if symbol_context.len() > OPEN_SYMBOLS_TAG.len() {
311 symbol_context.push_str("</symbols>\n");
312 message
313 .content
314 .push(language_model::MessageContent::Text(symbol_context));
315 }
316
317 if selection_context.len() > OPEN_SELECTIONS_TAG.len() {
318 selection_context.push_str("</selections>\n");
319 message
320 .content
321 .push(language_model::MessageContent::Text(selection_context));
322 }
323
324 if thread_context.len() > OPEN_THREADS_TAG.len() {
325 thread_context.push_str("</threads>\n");
326 message
327 .content
328 .push(language_model::MessageContent::Text(thread_context));
329 }
330
331 if fetch_context.len() > OPEN_FETCH_TAG.len() {
332 fetch_context.push_str("</fetched_urls>\n");
333 message
334 .content
335 .push(language_model::MessageContent::Text(fetch_context));
336 }
337
338 if rules_context.len() > OPEN_RULES_TAG.len() {
339 rules_context.push_str("</user_rules>\n");
340 message
341 .content
342 .push(language_model::MessageContent::Text(rules_context));
343 }
344
345 if message.content.len() > len_before_context {
346 message.content.insert(
347 len_before_context,
348 language_model::MessageContent::Text(OPEN_CONTEXT.into()),
349 );
350 message
351 .content
352 .push(language_model::MessageContent::Text("</context>".into()));
353 }
354
355 message
356 }
357}
358
359fn codeblock_tag(full_path: &Path, line_range: Option<&RangeInclusive<u32>>) -> String {
360 let mut result = String::new();
361
362 if let Some(extension) = full_path.extension().and_then(|ext| ext.to_str()) {
363 let _ = write!(result, "{} ", extension);
364 }
365
366 let _ = write!(result, "{}", full_path.display());
367
368 if let Some(range) = line_range {
369 if range.start() == range.end() {
370 let _ = write!(result, ":{}", range.start() + 1);
371 } else {
372 let _ = write!(result, ":{}-{}", range.start() + 1, range.end() + 1);
373 }
374 }
375
376 result
377}
378
379impl AgentMessage {
380 pub fn to_markdown(&self) -> String {
381 let mut markdown = String::from("## Assistant\n\n");
382
383 for content in &self.content {
384 match content {
385 AgentMessageContent::Text(text) => {
386 markdown.push_str(text);
387 markdown.push('\n');
388 }
389 AgentMessageContent::Thinking { text, .. } => {
390 markdown.push_str("<think>");
391 markdown.push_str(text);
392 markdown.push_str("</think>\n");
393 }
394 AgentMessageContent::RedactedThinking(_) => {
395 markdown.push_str("<redacted_thinking />\n")
396 }
397 AgentMessageContent::ToolUse(tool_use) => {
398 markdown.push_str(&format!(
399 "**Tool Use**: {} (ID: {})\n",
400 tool_use.name, tool_use.id
401 ));
402 markdown.push_str(&format!(
403 "{}\n",
404 MarkdownCodeBlock {
405 tag: "json",
406 text: &format!("{:#}", tool_use.input)
407 }
408 ));
409 }
410 }
411 }
412
413 for tool_result in self.tool_results.values() {
414 markdown.push_str(&format!(
415 "**Tool Result**: {} (ID: {})\n\n",
416 tool_result.tool_name, tool_result.tool_use_id
417 ));
418 if tool_result.is_error {
419 markdown.push_str("**ERROR:**\n");
420 }
421
422 match &tool_result.content {
423 LanguageModelToolResultContent::Text(text) => {
424 writeln!(markdown, "{text}\n").ok();
425 }
426 LanguageModelToolResultContent::Image(_) => {
427 writeln!(markdown, "<image />\n").ok();
428 }
429 }
430
431 if let Some(output) = tool_result.output.as_ref() {
432 writeln!(
433 markdown,
434 "**Debug Output**:\n\n```json\n{}\n```\n",
435 serde_json::to_string_pretty(output).unwrap()
436 )
437 .unwrap();
438 }
439 }
440
441 markdown
442 }
443
444 pub fn to_request(&self) -> Vec<LanguageModelRequestMessage> {
445 let mut assistant_message = LanguageModelRequestMessage {
446 role: Role::Assistant,
447 content: Vec::with_capacity(self.content.len()),
448 cache: false,
449 };
450 for chunk in &self.content {
451 match chunk {
452 AgentMessageContent::Text(text) => {
453 assistant_message
454 .content
455 .push(language_model::MessageContent::Text(text.clone()));
456 }
457 AgentMessageContent::Thinking { text, signature } => {
458 assistant_message
459 .content
460 .push(language_model::MessageContent::Thinking {
461 text: text.clone(),
462 signature: signature.clone(),
463 });
464 }
465 AgentMessageContent::RedactedThinking(value) => {
466 assistant_message.content.push(
467 language_model::MessageContent::RedactedThinking(value.clone()),
468 );
469 }
470 AgentMessageContent::ToolUse(tool_use) => {
471 if self.tool_results.contains_key(&tool_use.id) {
472 assistant_message
473 .content
474 .push(language_model::MessageContent::ToolUse(tool_use.clone()));
475 }
476 }
477 };
478 }
479
480 let mut user_message = LanguageModelRequestMessage {
481 role: Role::User,
482 content: Vec::new(),
483 cache: false,
484 };
485
486 for tool_result in self.tool_results.values() {
487 user_message
488 .content
489 .push(language_model::MessageContent::ToolResult(
490 tool_result.clone(),
491 ));
492 }
493
494 let mut messages = Vec::new();
495 if !assistant_message.content.is_empty() {
496 messages.push(assistant_message);
497 }
498 if !user_message.content.is_empty() {
499 messages.push(user_message);
500 }
501 messages
502 }
503}
504
505#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
506pub struct AgentMessage {
507 pub content: Vec<AgentMessageContent>,
508 pub tool_results: IndexMap<LanguageModelToolUseId, LanguageModelToolResult>,
509}
510
511#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
512pub enum AgentMessageContent {
513 Text(String),
514 Thinking {
515 text: String,
516 signature: Option<String>,
517 },
518 RedactedThinking(String),
519 ToolUse(LanguageModelToolUse),
520}
521
522#[derive(Debug)]
523pub enum ThreadEvent {
524 UserMessage(UserMessage),
525 AgentText(String),
526 AgentThinking(String),
527 ToolCall(acp::ToolCall),
528 ToolCallUpdate(acp_thread::ToolCallUpdate),
529 ToolCallAuthorization(ToolCallAuthorization),
530 Retry(acp_thread::RetryStatus),
531 Stop(acp::StopReason),
532}
533
534#[derive(Debug)]
535pub struct ToolCallAuthorization {
536 pub tool_call: acp::ToolCallUpdate,
537 pub options: Vec<acp::PermissionOption>,
538 pub response: oneshot::Sender<acp::PermissionOptionId>,
539}
540
541#[derive(Debug, thiserror::Error)]
542enum CompletionError {
543 #[error("max tokens")]
544 MaxTokens,
545 #[error("refusal")]
546 Refusal,
547 #[error(transparent)]
548 Other(#[from] anyhow::Error),
549}
550
551pub struct Thread {
552 id: acp::SessionId,
553 prompt_id: PromptId,
554 updated_at: DateTime<Utc>,
555 title: Option<SharedString>,
556 pending_title_generation: Option<Task<()>>,
557 summary: Option<SharedString>,
558 messages: Vec<Message>,
559 completion_mode: CompletionMode,
560 /// Holds the task that handles agent interaction until the end of the turn.
561 /// Survives across multiple requests as the model performs tool calls and
562 /// we run tools, report their results.
563 running_turn: Option<RunningTurn>,
564 pending_message: Option<AgentMessage>,
565 tools: BTreeMap<SharedString, Arc<dyn AnyAgentTool>>,
566 tool_use_limit_reached: bool,
567 request_token_usage: HashMap<UserMessageId, language_model::TokenUsage>,
568 #[allow(unused)]
569 cumulative_token_usage: TokenUsage,
570 #[allow(unused)]
571 initial_project_snapshot: Shared<Task<Option<Arc<ProjectSnapshot>>>>,
572 context_server_registry: Entity<ContextServerRegistry>,
573 profile_id: AgentProfileId,
574 project_context: Entity<ProjectContext>,
575 templates: Arc<Templates>,
576 model: Option<Arc<dyn LanguageModel>>,
577 summarization_model: Option<Arc<dyn LanguageModel>>,
578 pub(crate) project: Entity<Project>,
579 pub(crate) action_log: Entity<ActionLog>,
580}
581
582impl Thread {
583 pub fn new(
584 project: Entity<Project>,
585 project_context: Entity<ProjectContext>,
586 context_server_registry: Entity<ContextServerRegistry>,
587 templates: Arc<Templates>,
588 model: Option<Arc<dyn LanguageModel>>,
589 cx: &mut Context<Self>,
590 ) -> Self {
591 let profile_id = AgentSettings::get_global(cx).default_profile.clone();
592 let action_log = cx.new(|_cx| ActionLog::new(project.clone()));
593 Self {
594 id: acp::SessionId(uuid::Uuid::new_v4().to_string().into()),
595 prompt_id: PromptId::new(),
596 updated_at: Utc::now(),
597 title: None,
598 pending_title_generation: None,
599 summary: None,
600 messages: Vec::new(),
601 completion_mode: AgentSettings::get_global(cx).preferred_completion_mode,
602 running_turn: None,
603 pending_message: None,
604 tools: BTreeMap::default(),
605 tool_use_limit_reached: false,
606 request_token_usage: HashMap::default(),
607 cumulative_token_usage: TokenUsage::default(),
608 initial_project_snapshot: {
609 let project_snapshot = Self::project_snapshot(project.clone(), cx);
610 cx.foreground_executor()
611 .spawn(async move { Some(project_snapshot.await) })
612 .shared()
613 },
614 context_server_registry,
615 profile_id,
616 project_context,
617 templates,
618 model,
619 summarization_model: None,
620 project,
621 action_log,
622 }
623 }
624
625 pub fn id(&self) -> &acp::SessionId {
626 &self.id
627 }
628
629 pub fn replay(
630 &mut self,
631 cx: &mut Context<Self>,
632 ) -> mpsc::UnboundedReceiver<Result<ThreadEvent>> {
633 let (tx, rx) = mpsc::unbounded();
634 let stream = ThreadEventStream(tx);
635 for message in &self.messages {
636 match message {
637 Message::User(user_message) => stream.send_user_message(user_message),
638 Message::Agent(assistant_message) => {
639 for content in &assistant_message.content {
640 match content {
641 AgentMessageContent::Text(text) => stream.send_text(text),
642 AgentMessageContent::Thinking { text, .. } => {
643 stream.send_thinking(text)
644 }
645 AgentMessageContent::RedactedThinking(_) => {}
646 AgentMessageContent::ToolUse(tool_use) => {
647 self.replay_tool_call(
648 tool_use,
649 assistant_message.tool_results.get(&tool_use.id),
650 &stream,
651 cx,
652 );
653 }
654 }
655 }
656 }
657 Message::Resume => {}
658 }
659 }
660 rx
661 }
662
663 fn replay_tool_call(
664 &self,
665 tool_use: &LanguageModelToolUse,
666 tool_result: Option<&LanguageModelToolResult>,
667 stream: &ThreadEventStream,
668 cx: &mut Context<Self>,
669 ) {
670 let tool = self.tools.get(tool_use.name.as_ref()).cloned().or_else(|| {
671 self.context_server_registry
672 .read(cx)
673 .servers()
674 .find_map(|(_, tools)| {
675 if let Some(tool) = tools.get(tool_use.name.as_ref()) {
676 Some(tool.clone())
677 } else {
678 None
679 }
680 })
681 });
682
683 let Some(tool) = tool else {
684 stream
685 .0
686 .unbounded_send(Ok(ThreadEvent::ToolCall(acp::ToolCall {
687 id: acp::ToolCallId(tool_use.id.to_string().into()),
688 title: tool_use.name.to_string(),
689 kind: acp::ToolKind::Other,
690 status: acp::ToolCallStatus::Failed,
691 content: Vec::new(),
692 locations: Vec::new(),
693 raw_input: Some(tool_use.input.clone()),
694 raw_output: None,
695 })))
696 .ok();
697 return;
698 };
699
700 let title = tool.initial_title(tool_use.input.clone());
701 let kind = tool.kind();
702 stream.send_tool_call(&tool_use.id, title, kind, tool_use.input.clone());
703
704 let output = tool_result
705 .as_ref()
706 .and_then(|result| result.output.clone());
707 if let Some(output) = output.clone() {
708 let tool_event_stream = ToolCallEventStream::new(
709 tool_use.id.clone(),
710 stream.clone(),
711 Some(self.project.read(cx).fs().clone()),
712 );
713 tool.replay(tool_use.input.clone(), output, tool_event_stream, cx)
714 .log_err();
715 }
716
717 stream.update_tool_call_fields(
718 &tool_use.id,
719 acp::ToolCallUpdateFields {
720 status: Some(acp::ToolCallStatus::Completed),
721 raw_output: output,
722 ..Default::default()
723 },
724 );
725 }
726
727 pub fn from_db(
728 id: acp::SessionId,
729 db_thread: DbThread,
730 project: Entity<Project>,
731 project_context: Entity<ProjectContext>,
732 context_server_registry: Entity<ContextServerRegistry>,
733 action_log: Entity<ActionLog>,
734 templates: Arc<Templates>,
735 cx: &mut Context<Self>,
736 ) -> Self {
737 let profile_id = db_thread
738 .profile
739 .unwrap_or_else(|| AgentSettings::get_global(cx).default_profile.clone());
740 let model = LanguageModelRegistry::global(cx).update(cx, |registry, cx| {
741 db_thread
742 .model
743 .and_then(|model| {
744 let model = SelectedModel {
745 provider: model.provider.clone().into(),
746 model: model.model.into(),
747 };
748 registry.select_model(&model, cx)
749 })
750 .or_else(|| registry.default_model())
751 .map(|model| model.model)
752 });
753
754 Self {
755 id,
756 prompt_id: PromptId::new(),
757 title: if db_thread.title.is_empty() {
758 None
759 } else {
760 Some(db_thread.title.clone())
761 },
762 pending_title_generation: None,
763 summary: db_thread.detailed_summary,
764 messages: db_thread.messages,
765 completion_mode: db_thread.completion_mode.unwrap_or_default(),
766 running_turn: None,
767 pending_message: None,
768 tools: BTreeMap::default(),
769 tool_use_limit_reached: false,
770 request_token_usage: db_thread.request_token_usage.clone(),
771 cumulative_token_usage: db_thread.cumulative_token_usage,
772 initial_project_snapshot: Task::ready(db_thread.initial_project_snapshot).shared(),
773 context_server_registry,
774 profile_id,
775 project_context,
776 templates,
777 model,
778 summarization_model: None,
779 project,
780 action_log,
781 updated_at: db_thread.updated_at,
782 }
783 }
784
785 pub fn to_db(&self, cx: &App) -> Task<DbThread> {
786 let initial_project_snapshot = self.initial_project_snapshot.clone();
787 let mut thread = DbThread {
788 title: self.title(),
789 messages: self.messages.clone(),
790 updated_at: self.updated_at,
791 detailed_summary: self.summary.clone(),
792 initial_project_snapshot: None,
793 cumulative_token_usage: self.cumulative_token_usage,
794 request_token_usage: self.request_token_usage.clone(),
795 model: self.model.as_ref().map(|model| DbLanguageModel {
796 provider: model.provider_id().to_string(),
797 model: model.name().0.to_string(),
798 }),
799 completion_mode: Some(self.completion_mode),
800 profile: Some(self.profile_id.clone()),
801 };
802
803 cx.background_spawn(async move {
804 let initial_project_snapshot = initial_project_snapshot.await;
805 thread.initial_project_snapshot = initial_project_snapshot;
806 thread
807 })
808 }
809
810 /// Create a snapshot of the current project state including git information and unsaved buffers.
811 fn project_snapshot(
812 project: Entity<Project>,
813 cx: &mut Context<Self>,
814 ) -> Task<Arc<agent::thread::ProjectSnapshot>> {
815 let git_store = project.read(cx).git_store().clone();
816 let worktree_snapshots: Vec<_> = project
817 .read(cx)
818 .visible_worktrees(cx)
819 .map(|worktree| Self::worktree_snapshot(worktree, git_store.clone(), cx))
820 .collect();
821
822 cx.spawn(async move |_, cx| {
823 let worktree_snapshots = futures::future::join_all(worktree_snapshots).await;
824
825 let mut unsaved_buffers = Vec::new();
826 cx.update(|app_cx| {
827 let buffer_store = project.read(app_cx).buffer_store();
828 for buffer_handle in buffer_store.read(app_cx).buffers() {
829 let buffer = buffer_handle.read(app_cx);
830 if buffer.is_dirty()
831 && let Some(file) = buffer.file()
832 {
833 let path = file.path().to_string_lossy().to_string();
834 unsaved_buffers.push(path);
835 }
836 }
837 })
838 .ok();
839
840 Arc::new(ProjectSnapshot {
841 worktree_snapshots,
842 unsaved_buffer_paths: unsaved_buffers,
843 timestamp: Utc::now(),
844 })
845 })
846 }
847
848 fn worktree_snapshot(
849 worktree: Entity<project::Worktree>,
850 git_store: Entity<GitStore>,
851 cx: &App,
852 ) -> Task<agent::thread::WorktreeSnapshot> {
853 cx.spawn(async move |cx| {
854 // Get worktree path and snapshot
855 let worktree_info = cx.update(|app_cx| {
856 let worktree = worktree.read(app_cx);
857 let path = worktree.abs_path().to_string_lossy().to_string();
858 let snapshot = worktree.snapshot();
859 (path, snapshot)
860 });
861
862 let Ok((worktree_path, _snapshot)) = worktree_info else {
863 return WorktreeSnapshot {
864 worktree_path: String::new(),
865 git_state: None,
866 };
867 };
868
869 let git_state = git_store
870 .update(cx, |git_store, cx| {
871 git_store
872 .repositories()
873 .values()
874 .find(|repo| {
875 repo.read(cx)
876 .abs_path_to_repo_path(&worktree.read(cx).abs_path())
877 .is_some()
878 })
879 .cloned()
880 })
881 .ok()
882 .flatten()
883 .map(|repo| {
884 repo.update(cx, |repo, _| {
885 let current_branch =
886 repo.branch.as_ref().map(|branch| branch.name().to_owned());
887 repo.send_job(None, |state, _| async move {
888 let RepositoryState::Local { backend, .. } = state else {
889 return GitState {
890 remote_url: None,
891 head_sha: None,
892 current_branch,
893 diff: None,
894 };
895 };
896
897 let remote_url = backend.remote_url("origin");
898 let head_sha = backend.head_sha().await;
899 let diff = backend.diff(DiffType::HeadToWorktree).await.ok();
900
901 GitState {
902 remote_url,
903 head_sha,
904 current_branch,
905 diff,
906 }
907 })
908 })
909 });
910
911 let git_state = match git_state {
912 Some(git_state) => match git_state.ok() {
913 Some(git_state) => git_state.await.ok(),
914 None => None,
915 },
916 None => None,
917 };
918
919 WorktreeSnapshot {
920 worktree_path,
921 git_state,
922 }
923 })
924 }
925
926 pub fn project_context(&self) -> &Entity<ProjectContext> {
927 &self.project_context
928 }
929
930 pub fn project(&self) -> &Entity<Project> {
931 &self.project
932 }
933
934 pub fn action_log(&self) -> &Entity<ActionLog> {
935 &self.action_log
936 }
937
938 pub fn is_empty(&self) -> bool {
939 self.messages.is_empty() && self.title.is_none()
940 }
941
942 pub fn model(&self) -> Option<&Arc<dyn LanguageModel>> {
943 self.model.as_ref()
944 }
945
946 pub fn set_model(&mut self, model: Arc<dyn LanguageModel>, cx: &mut Context<Self>) {
947 let old_usage = self.latest_token_usage();
948 self.model = Some(model);
949 let new_usage = self.latest_token_usage();
950 if old_usage != new_usage {
951 cx.emit(TokenUsageUpdated(new_usage));
952 }
953 cx.notify()
954 }
955
956 pub fn summarization_model(&self) -> Option<&Arc<dyn LanguageModel>> {
957 self.summarization_model.as_ref()
958 }
959
960 pub fn set_summarization_model(
961 &mut self,
962 model: Option<Arc<dyn LanguageModel>>,
963 cx: &mut Context<Self>,
964 ) {
965 self.summarization_model = model;
966 cx.notify()
967 }
968
969 pub fn completion_mode(&self) -> CompletionMode {
970 self.completion_mode
971 }
972
973 pub fn set_completion_mode(&mut self, mode: CompletionMode, cx: &mut Context<Self>) {
974 let old_usage = self.latest_token_usage();
975 self.completion_mode = mode;
976 let new_usage = self.latest_token_usage();
977 if old_usage != new_usage {
978 cx.emit(TokenUsageUpdated(new_usage));
979 }
980 cx.notify()
981 }
982
983 #[cfg(any(test, feature = "test-support"))]
984 pub fn last_message(&self) -> Option<Message> {
985 if let Some(message) = self.pending_message.clone() {
986 Some(Message::Agent(message))
987 } else {
988 self.messages.last().cloned()
989 }
990 }
991
992 pub fn add_default_tools(&mut self, cx: &mut Context<Self>) {
993 let language_registry = self.project.read(cx).languages().clone();
994 self.add_tool(CopyPathTool::new(self.project.clone()));
995 self.add_tool(CreateDirectoryTool::new(self.project.clone()));
996 self.add_tool(DeletePathTool::new(
997 self.project.clone(),
998 self.action_log.clone(),
999 ));
1000 self.add_tool(DiagnosticsTool::new(self.project.clone()));
1001 self.add_tool(EditFileTool::new(cx.weak_entity(), language_registry));
1002 self.add_tool(FetchTool::new(self.project.read(cx).client().http_client()));
1003 self.add_tool(FindPathTool::new(self.project.clone()));
1004 self.add_tool(GrepTool::new(self.project.clone()));
1005 self.add_tool(ListDirectoryTool::new(self.project.clone()));
1006 self.add_tool(MovePathTool::new(self.project.clone()));
1007 self.add_tool(NowTool);
1008 self.add_tool(OpenTool::new(self.project.clone()));
1009 self.add_tool(ReadFileTool::new(
1010 self.project.clone(),
1011 self.action_log.clone(),
1012 ));
1013 self.add_tool(TerminalTool::new(self.project.clone(), cx));
1014 self.add_tool(ThinkingTool);
1015 self.add_tool(WebSearchTool);
1016 }
1017
1018 pub fn add_tool<T: AgentTool>(&mut self, tool: T) {
1019 self.tools.insert(T::name().into(), tool.erase());
1020 }
1021
1022 pub fn remove_tool(&mut self, name: &str) -> bool {
1023 self.tools.remove(name).is_some()
1024 }
1025
1026 pub fn profile(&self) -> &AgentProfileId {
1027 &self.profile_id
1028 }
1029
1030 pub fn set_profile(&mut self, profile_id: AgentProfileId) {
1031 self.profile_id = profile_id;
1032 }
1033
1034 pub fn cancel(&mut self, cx: &mut Context<Self>) {
1035 if let Some(running_turn) = self.running_turn.take() {
1036 running_turn.cancel();
1037 }
1038 self.flush_pending_message(cx);
1039 }
1040
1041 fn update_token_usage(&mut self, update: language_model::TokenUsage, cx: &mut Context<Self>) {
1042 let Some(last_user_message) = self.last_user_message() else {
1043 return;
1044 };
1045
1046 self.request_token_usage
1047 .insert(last_user_message.id.clone(), update);
1048 cx.emit(TokenUsageUpdated(self.latest_token_usage()));
1049 cx.notify();
1050 }
1051
1052 pub fn truncate(&mut self, message_id: UserMessageId, cx: &mut Context<Self>) -> Result<()> {
1053 self.cancel(cx);
1054 let Some(position) = self.messages.iter().position(
1055 |msg| matches!(msg, Message::User(UserMessage { id, .. }) if id == &message_id),
1056 ) else {
1057 return Err(anyhow!("Message not found"));
1058 };
1059
1060 for message in self.messages.drain(position..) {
1061 match message {
1062 Message::User(message) => {
1063 self.request_token_usage.remove(&message.id);
1064 }
1065 Message::Agent(_) | Message::Resume => {}
1066 }
1067 }
1068 self.summary = None;
1069 cx.notify();
1070 Ok(())
1071 }
1072
1073 pub fn latest_token_usage(&self) -> Option<acp_thread::TokenUsage> {
1074 let last_user_message = self.last_user_message()?;
1075 let tokens = self.request_token_usage.get(&last_user_message.id)?;
1076 let model = self.model.clone()?;
1077
1078 Some(acp_thread::TokenUsage {
1079 max_tokens: model.max_token_count_for_mode(self.completion_mode.into()),
1080 used_tokens: tokens.total_tokens(),
1081 })
1082 }
1083
1084 pub fn resume(
1085 &mut self,
1086 cx: &mut Context<Self>,
1087 ) -> Result<mpsc::UnboundedReceiver<Result<ThreadEvent>>> {
1088 self.messages.push(Message::Resume);
1089 cx.notify();
1090
1091 log::debug!("Total messages in thread: {}", self.messages.len());
1092 self.run_turn(cx)
1093 }
1094
1095 /// Sending a message results in the model streaming a response, which could include tool calls.
1096 /// After calling tools, the model will stops and waits for any outstanding tool calls to be completed and their results sent.
1097 /// The returned channel will report all the occurrences in which the model stops before erroring or ending its turn.
1098 pub fn send<T>(
1099 &mut self,
1100 id: UserMessageId,
1101 content: impl IntoIterator<Item = T>,
1102 cx: &mut Context<Self>,
1103 ) -> Result<mpsc::UnboundedReceiver<Result<ThreadEvent>>>
1104 where
1105 T: Into<UserMessageContent>,
1106 {
1107 let model = self.model().context("No language model configured")?;
1108
1109 log::info!("Thread::send called with model: {}", model.name().0);
1110 self.advance_prompt_id();
1111
1112 let content = content.into_iter().map(Into::into).collect::<Vec<_>>();
1113 log::debug!("Thread::send content: {:?}", content);
1114
1115 self.messages
1116 .push(Message::User(UserMessage { id, content }));
1117 cx.notify();
1118
1119 log::debug!("Total messages in thread: {}", self.messages.len());
1120 self.run_turn(cx)
1121 }
1122
1123 fn run_turn(
1124 &mut self,
1125 cx: &mut Context<Self>,
1126 ) -> Result<mpsc::UnboundedReceiver<Result<ThreadEvent>>> {
1127 self.cancel(cx);
1128
1129 let model = self.model.clone().context("No language model configured")?;
1130 let profile = AgentSettings::get_global(cx)
1131 .profiles
1132 .get(&self.profile_id)
1133 .context("Profile not found")?;
1134 let (events_tx, events_rx) = mpsc::unbounded::<Result<ThreadEvent>>();
1135 let event_stream = ThreadEventStream(events_tx);
1136 let message_ix = self.messages.len().saturating_sub(1);
1137 self.tool_use_limit_reached = false;
1138 self.summary = None;
1139 self.running_turn = Some(RunningTurn {
1140 event_stream: event_stream.clone(),
1141 tools: self.enabled_tools(profile, &model, cx),
1142 _task: cx.spawn(async move |this, cx| {
1143 log::debug!("Starting agent turn execution");
1144
1145 let turn_result: Result<()> = async {
1146 let mut intent = CompletionIntent::UserPrompt;
1147 loop {
1148 Self::stream_completion(&this, &model, intent, &event_stream, cx).await?;
1149
1150 let mut end_turn = true;
1151 this.update(cx, |this, cx| {
1152 // Generate title if needed.
1153 if this.title.is_none() && this.pending_title_generation.is_none() {
1154 this.generate_title(cx);
1155 }
1156
1157 // End the turn if the model didn't use tools.
1158 let message = this.pending_message.as_ref();
1159 end_turn =
1160 message.map_or(true, |message| message.tool_results.is_empty());
1161 this.flush_pending_message(cx);
1162 })?;
1163
1164 if this.read_with(cx, |this, _| this.tool_use_limit_reached)? {
1165 log::info!("Tool use limit reached, completing turn");
1166 return Err(language_model::ToolUseLimitReachedError.into());
1167 } else if end_turn {
1168 log::debug!("No tool uses found, completing turn");
1169 return Ok(());
1170 } else {
1171 intent = CompletionIntent::ToolResults;
1172 }
1173 }
1174 }
1175 .await;
1176 _ = this.update(cx, |this, cx| this.flush_pending_message(cx));
1177
1178 match turn_result {
1179 Ok(()) => {
1180 log::debug!("Turn execution completed");
1181 event_stream.send_stop(acp::StopReason::EndTurn);
1182 }
1183 Err(error) => {
1184 log::error!("Turn execution failed: {:?}", error);
1185 match error.downcast::<CompletionError>() {
1186 Ok(CompletionError::Refusal) => {
1187 event_stream.send_stop(acp::StopReason::Refusal);
1188 _ = this.update(cx, |this, _| this.messages.truncate(message_ix));
1189 }
1190 Ok(CompletionError::MaxTokens) => {
1191 event_stream.send_stop(acp::StopReason::MaxTokens);
1192 }
1193 Ok(CompletionError::Other(error)) | Err(error) => {
1194 event_stream.send_error(error);
1195 }
1196 }
1197 }
1198 }
1199
1200 _ = this.update(cx, |this, _| this.running_turn.take());
1201 }),
1202 });
1203 Ok(events_rx)
1204 }
1205
1206 async fn stream_completion(
1207 this: &WeakEntity<Self>,
1208 model: &Arc<dyn LanguageModel>,
1209 completion_intent: CompletionIntent,
1210 event_stream: &ThreadEventStream,
1211 cx: &mut AsyncApp,
1212 ) -> Result<()> {
1213 log::debug!("Stream completion started successfully");
1214
1215 let mut attempt = None;
1216 loop {
1217 let request = this.update(cx, |this, cx| {
1218 this.build_completion_request(completion_intent, cx)
1219 })??;
1220
1221 telemetry::event!(
1222 "Agent Thread Completion",
1223 thread_id = this.read_with(cx, |this, _| this.id.to_string())?,
1224 prompt_id = this.read_with(cx, |this, _| this.prompt_id.to_string())?,
1225 model = model.telemetry_id(),
1226 model_provider = model.provider_id().to_string(),
1227 attempt
1228 );
1229
1230 log::debug!(
1231 "Calling model.stream_completion, attempt {}",
1232 attempt.unwrap_or(0)
1233 );
1234 let mut events = model
1235 .stream_completion(request, cx)
1236 .await
1237 .map_err(|error| anyhow!(error))?;
1238 let mut tool_results = FuturesUnordered::new();
1239 let mut error = None;
1240
1241 while let Some(event) = events.next().await {
1242 match event {
1243 Ok(event) => {
1244 log::trace!("Received completion event: {:?}", event);
1245 tool_results.extend(this.update(cx, |this, cx| {
1246 this.handle_streamed_completion_event(event, event_stream, cx)
1247 })??);
1248 }
1249 Err(err) => {
1250 error = Some(err);
1251 break;
1252 }
1253 }
1254 }
1255
1256 while let Some(tool_result) = tool_results.next().await {
1257 log::debug!("Tool finished {:?}", tool_result);
1258
1259 event_stream.update_tool_call_fields(
1260 &tool_result.tool_use_id,
1261 acp::ToolCallUpdateFields {
1262 status: Some(if tool_result.is_error {
1263 acp::ToolCallStatus::Failed
1264 } else {
1265 acp::ToolCallStatus::Completed
1266 }),
1267 raw_output: tool_result.output.clone(),
1268 ..Default::default()
1269 },
1270 );
1271 this.update(cx, |this, _cx| {
1272 this.pending_message()
1273 .tool_results
1274 .insert(tool_result.tool_use_id.clone(), tool_result);
1275 })?;
1276 }
1277
1278 if let Some(error) = error {
1279 let completion_mode = this.read_with(cx, |thread, _cx| thread.completion_mode())?;
1280 if completion_mode == CompletionMode::Normal {
1281 return Err(anyhow!(error))?;
1282 }
1283
1284 let Some(strategy) = Self::retry_strategy_for(&error) else {
1285 return Err(anyhow!(error))?;
1286 };
1287
1288 let max_attempts = match &strategy {
1289 RetryStrategy::ExponentialBackoff { max_attempts, .. } => *max_attempts,
1290 RetryStrategy::Fixed { max_attempts, .. } => *max_attempts,
1291 };
1292
1293 let attempt = attempt.get_or_insert(0u8);
1294
1295 *attempt += 1;
1296
1297 let attempt = *attempt;
1298 if attempt > max_attempts {
1299 return Err(anyhow!(error))?;
1300 }
1301
1302 let delay = match &strategy {
1303 RetryStrategy::ExponentialBackoff { initial_delay, .. } => {
1304 let delay_secs = initial_delay.as_secs() * 2u64.pow((attempt - 1) as u32);
1305 Duration::from_secs(delay_secs)
1306 }
1307 RetryStrategy::Fixed { delay, .. } => *delay,
1308 };
1309 log::debug!("Retry attempt {attempt} with delay {delay:?}");
1310
1311 event_stream.send_retry(acp_thread::RetryStatus {
1312 last_error: error.to_string().into(),
1313 attempt: attempt as usize,
1314 max_attempts: max_attempts as usize,
1315 started_at: Instant::now(),
1316 duration: delay,
1317 });
1318 cx.background_executor().timer(delay).await;
1319 this.update(cx, |this, cx| {
1320 this.flush_pending_message(cx);
1321 if let Some(Message::Agent(message)) = this.messages.last() {
1322 if message.tool_results.is_empty() {
1323 this.messages.push(Message::Resume);
1324 }
1325 }
1326 })?;
1327 } else {
1328 return Ok(());
1329 }
1330 }
1331 }
1332
1333 /// A helper method that's called on every streamed completion event.
1334 /// Returns an optional tool result task, which the main agentic loop will
1335 /// send back to the model when it resolves.
1336 fn handle_streamed_completion_event(
1337 &mut self,
1338 event: LanguageModelCompletionEvent,
1339 event_stream: &ThreadEventStream,
1340 cx: &mut Context<Self>,
1341 ) -> Result<Option<Task<LanguageModelToolResult>>> {
1342 log::trace!("Handling streamed completion event: {:?}", event);
1343 use LanguageModelCompletionEvent::*;
1344
1345 match event {
1346 StartMessage { .. } => {
1347 self.flush_pending_message(cx);
1348 self.pending_message = Some(AgentMessage::default());
1349 }
1350 Text(new_text) => self.handle_text_event(new_text, event_stream, cx),
1351 Thinking { text, signature } => {
1352 self.handle_thinking_event(text, signature, event_stream, cx)
1353 }
1354 RedactedThinking { data } => self.handle_redacted_thinking_event(data, cx),
1355 ToolUse(tool_use) => {
1356 return Ok(self.handle_tool_use_event(tool_use, event_stream, cx));
1357 }
1358 ToolUseJsonParseError {
1359 id,
1360 tool_name,
1361 raw_input,
1362 json_parse_error,
1363 } => {
1364 return Ok(Some(Task::ready(
1365 self.handle_tool_use_json_parse_error_event(
1366 id,
1367 tool_name,
1368 raw_input,
1369 json_parse_error,
1370 ),
1371 )));
1372 }
1373 UsageUpdate(usage) => {
1374 telemetry::event!(
1375 "Agent Thread Completion Usage Updated",
1376 thread_id = self.id.to_string(),
1377 prompt_id = self.prompt_id.to_string(),
1378 model = self.model.as_ref().map(|m| m.telemetry_id()),
1379 model_provider = self.model.as_ref().map(|m| m.provider_id().to_string()),
1380 input_tokens = usage.input_tokens,
1381 output_tokens = usage.output_tokens,
1382 cache_creation_input_tokens = usage.cache_creation_input_tokens,
1383 cache_read_input_tokens = usage.cache_read_input_tokens,
1384 );
1385 self.update_token_usage(usage, cx);
1386 }
1387 StatusUpdate(CompletionRequestStatus::UsageUpdated { amount, limit }) => {
1388 self.update_model_request_usage(amount, limit, cx);
1389 }
1390 StatusUpdate(
1391 CompletionRequestStatus::Started
1392 | CompletionRequestStatus::Queued { .. }
1393 | CompletionRequestStatus::Failed { .. },
1394 ) => {}
1395 StatusUpdate(CompletionRequestStatus::ToolUseLimitReached) => {
1396 self.tool_use_limit_reached = true;
1397 }
1398 Stop(StopReason::Refusal) => return Err(CompletionError::Refusal.into()),
1399 Stop(StopReason::MaxTokens) => return Err(CompletionError::MaxTokens.into()),
1400 Stop(StopReason::ToolUse | StopReason::EndTurn) => {}
1401 }
1402
1403 Ok(None)
1404 }
1405
1406 fn handle_text_event(
1407 &mut self,
1408 new_text: String,
1409 event_stream: &ThreadEventStream,
1410 cx: &mut Context<Self>,
1411 ) {
1412 event_stream.send_text(&new_text);
1413
1414 let last_message = self.pending_message();
1415 if let Some(AgentMessageContent::Text(text)) = last_message.content.last_mut() {
1416 text.push_str(&new_text);
1417 } else {
1418 last_message
1419 .content
1420 .push(AgentMessageContent::Text(new_text));
1421 }
1422
1423 cx.notify();
1424 }
1425
1426 fn handle_thinking_event(
1427 &mut self,
1428 new_text: String,
1429 new_signature: Option<String>,
1430 event_stream: &ThreadEventStream,
1431 cx: &mut Context<Self>,
1432 ) {
1433 event_stream.send_thinking(&new_text);
1434
1435 let last_message = self.pending_message();
1436 if let Some(AgentMessageContent::Thinking { text, signature }) =
1437 last_message.content.last_mut()
1438 {
1439 text.push_str(&new_text);
1440 *signature = new_signature.or(signature.take());
1441 } else {
1442 last_message.content.push(AgentMessageContent::Thinking {
1443 text: new_text,
1444 signature: new_signature,
1445 });
1446 }
1447
1448 cx.notify();
1449 }
1450
1451 fn handle_redacted_thinking_event(&mut self, data: String, cx: &mut Context<Self>) {
1452 let last_message = self.pending_message();
1453 last_message
1454 .content
1455 .push(AgentMessageContent::RedactedThinking(data));
1456 cx.notify();
1457 }
1458
1459 fn handle_tool_use_event(
1460 &mut self,
1461 tool_use: LanguageModelToolUse,
1462 event_stream: &ThreadEventStream,
1463 cx: &mut Context<Self>,
1464 ) -> Option<Task<LanguageModelToolResult>> {
1465 cx.notify();
1466
1467 let tool = self.tool(tool_use.name.as_ref());
1468 let mut title = SharedString::from(&tool_use.name);
1469 let mut kind = acp::ToolKind::Other;
1470 if let Some(tool) = tool.as_ref() {
1471 title = tool.initial_title(tool_use.input.clone());
1472 kind = tool.kind();
1473 }
1474
1475 // Ensure the last message ends in the current tool use
1476 let last_message = self.pending_message();
1477 let push_new_tool_use = last_message.content.last_mut().is_none_or(|content| {
1478 if let AgentMessageContent::ToolUse(last_tool_use) = content {
1479 if last_tool_use.id == tool_use.id {
1480 *last_tool_use = tool_use.clone();
1481 false
1482 } else {
1483 true
1484 }
1485 } else {
1486 true
1487 }
1488 });
1489
1490 if push_new_tool_use {
1491 event_stream.send_tool_call(&tool_use.id, title, kind, tool_use.input.clone());
1492 last_message
1493 .content
1494 .push(AgentMessageContent::ToolUse(tool_use.clone()));
1495 } else {
1496 event_stream.update_tool_call_fields(
1497 &tool_use.id,
1498 acp::ToolCallUpdateFields {
1499 title: Some(title.into()),
1500 kind: Some(kind),
1501 raw_input: Some(tool_use.input.clone()),
1502 ..Default::default()
1503 },
1504 );
1505 }
1506
1507 if !tool_use.is_input_complete {
1508 return None;
1509 }
1510
1511 let Some(tool) = tool else {
1512 let content = format!("No tool named {} exists", tool_use.name);
1513 return Some(Task::ready(LanguageModelToolResult {
1514 content: LanguageModelToolResultContent::Text(Arc::from(content)),
1515 tool_use_id: tool_use.id,
1516 tool_name: tool_use.name,
1517 is_error: true,
1518 output: None,
1519 }));
1520 };
1521
1522 let fs = self.project.read(cx).fs().clone();
1523 let tool_event_stream =
1524 ToolCallEventStream::new(tool_use.id.clone(), event_stream.clone(), Some(fs));
1525 tool_event_stream.update_fields(acp::ToolCallUpdateFields {
1526 status: Some(acp::ToolCallStatus::InProgress),
1527 ..Default::default()
1528 });
1529 let supports_images = self.model().is_some_and(|model| model.supports_images());
1530 let tool_result = tool.run(tool_use.input, tool_event_stream, cx);
1531 log::debug!("Running tool {}", tool_use.name);
1532 Some(cx.foreground_executor().spawn(async move {
1533 let tool_result = tool_result.await.and_then(|output| {
1534 if let LanguageModelToolResultContent::Image(_) = &output.llm_output
1535 && !supports_images
1536 {
1537 return Err(anyhow!(
1538 "Attempted to read an image, but this model doesn't support it.",
1539 ));
1540 }
1541 Ok(output)
1542 });
1543
1544 match tool_result {
1545 Ok(output) => LanguageModelToolResult {
1546 tool_use_id: tool_use.id,
1547 tool_name: tool_use.name,
1548 is_error: false,
1549 content: output.llm_output,
1550 output: Some(output.raw_output),
1551 },
1552 Err(error) => LanguageModelToolResult {
1553 tool_use_id: tool_use.id,
1554 tool_name: tool_use.name,
1555 is_error: true,
1556 content: LanguageModelToolResultContent::Text(Arc::from(error.to_string())),
1557 output: None,
1558 },
1559 }
1560 }))
1561 }
1562
1563 fn handle_tool_use_json_parse_error_event(
1564 &mut self,
1565 tool_use_id: LanguageModelToolUseId,
1566 tool_name: Arc<str>,
1567 raw_input: Arc<str>,
1568 json_parse_error: String,
1569 ) -> LanguageModelToolResult {
1570 let tool_output = format!("Error parsing input JSON: {json_parse_error}");
1571 LanguageModelToolResult {
1572 tool_use_id,
1573 tool_name,
1574 is_error: true,
1575 content: LanguageModelToolResultContent::Text(tool_output.into()),
1576 output: Some(serde_json::Value::String(raw_input.to_string())),
1577 }
1578 }
1579
1580 fn update_model_request_usage(&self, amount: usize, limit: UsageLimit, cx: &mut Context<Self>) {
1581 self.project
1582 .read(cx)
1583 .user_store()
1584 .update(cx, |user_store, cx| {
1585 user_store.update_model_request_usage(
1586 ModelRequestUsage(RequestUsage {
1587 amount: amount as i32,
1588 limit,
1589 }),
1590 cx,
1591 )
1592 });
1593 }
1594
1595 pub fn title(&self) -> SharedString {
1596 self.title.clone().unwrap_or("New Thread".into())
1597 }
1598
1599 pub fn summary(&mut self, cx: &mut Context<Self>) -> Task<Result<SharedString>> {
1600 if let Some(summary) = self.summary.as_ref() {
1601 return Task::ready(Ok(summary.clone()));
1602 }
1603 let Some(model) = self.summarization_model.clone() else {
1604 return Task::ready(Err(anyhow!("No summarization model available")));
1605 };
1606 let mut request = LanguageModelRequest {
1607 intent: Some(CompletionIntent::ThreadContextSummarization),
1608 temperature: AgentSettings::temperature_for_model(&model, cx),
1609 ..Default::default()
1610 };
1611
1612 for message in &self.messages {
1613 request.messages.extend(message.to_request());
1614 }
1615
1616 request.messages.push(LanguageModelRequestMessage {
1617 role: Role::User,
1618 content: vec![SUMMARIZE_THREAD_DETAILED_PROMPT.into()],
1619 cache: false,
1620 });
1621 cx.spawn(async move |this, cx| {
1622 let mut summary = String::new();
1623 let mut messages = model.stream_completion(request, cx).await?;
1624 while let Some(event) = messages.next().await {
1625 let event = event?;
1626 let text = match event {
1627 LanguageModelCompletionEvent::Text(text) => text,
1628 LanguageModelCompletionEvent::StatusUpdate(
1629 CompletionRequestStatus::UsageUpdated { amount, limit },
1630 ) => {
1631 this.update(cx, |thread, cx| {
1632 thread.update_model_request_usage(amount, limit, cx);
1633 })?;
1634 continue;
1635 }
1636 _ => continue,
1637 };
1638
1639 let mut lines = text.lines();
1640 summary.extend(lines.next());
1641 }
1642
1643 log::debug!("Setting summary: {}", summary);
1644 let summary = SharedString::from(summary);
1645
1646 this.update(cx, |this, cx| {
1647 this.summary = Some(summary.clone());
1648 cx.notify()
1649 })?;
1650
1651 Ok(summary)
1652 })
1653 }
1654
1655 fn generate_title(&mut self, cx: &mut Context<Self>) {
1656 let Some(model) = self.summarization_model.clone() else {
1657 return;
1658 };
1659
1660 log::debug!(
1661 "Generating title with model: {:?}",
1662 self.summarization_model.as_ref().map(|model| model.name())
1663 );
1664 let mut request = LanguageModelRequest {
1665 intent: Some(CompletionIntent::ThreadSummarization),
1666 temperature: AgentSettings::temperature_for_model(&model, cx),
1667 ..Default::default()
1668 };
1669
1670 for message in &self.messages {
1671 request.messages.extend(message.to_request());
1672 }
1673
1674 request.messages.push(LanguageModelRequestMessage {
1675 role: Role::User,
1676 content: vec![SUMMARIZE_THREAD_PROMPT.into()],
1677 cache: false,
1678 });
1679 self.pending_title_generation = Some(cx.spawn(async move |this, cx| {
1680 let mut title = String::new();
1681
1682 let generate = async {
1683 let mut messages = model.stream_completion(request, cx).await?;
1684 while let Some(event) = messages.next().await {
1685 let event = event?;
1686 let text = match event {
1687 LanguageModelCompletionEvent::Text(text) => text,
1688 LanguageModelCompletionEvent::StatusUpdate(
1689 CompletionRequestStatus::UsageUpdated { amount, limit },
1690 ) => {
1691 this.update(cx, |thread, cx| {
1692 thread.update_model_request_usage(amount, limit, cx);
1693 })?;
1694 continue;
1695 }
1696 _ => continue,
1697 };
1698
1699 let mut lines = text.lines();
1700 title.extend(lines.next());
1701
1702 // Stop if the LLM generated multiple lines.
1703 if lines.next().is_some() {
1704 break;
1705 }
1706 }
1707 anyhow::Ok(())
1708 };
1709
1710 if generate.await.context("failed to generate title").is_ok() {
1711 _ = this.update(cx, |this, cx| this.set_title(title.into(), cx));
1712 }
1713 _ = this.update(cx, |this, _| this.pending_title_generation = None);
1714 }));
1715 }
1716
1717 pub fn set_title(&mut self, title: SharedString, cx: &mut Context<Self>) {
1718 self.pending_title_generation = None;
1719 if Some(&title) != self.title.as_ref() {
1720 self.title = Some(title);
1721 cx.emit(TitleUpdated);
1722 cx.notify();
1723 }
1724 }
1725
1726 fn last_user_message(&self) -> Option<&UserMessage> {
1727 self.messages
1728 .iter()
1729 .rev()
1730 .find_map(|message| match message {
1731 Message::User(user_message) => Some(user_message),
1732 Message::Agent(_) => None,
1733 Message::Resume => None,
1734 })
1735 }
1736
1737 fn pending_message(&mut self) -> &mut AgentMessage {
1738 self.pending_message.get_or_insert_default()
1739 }
1740
1741 fn flush_pending_message(&mut self, cx: &mut Context<Self>) {
1742 let Some(mut message) = self.pending_message.take() else {
1743 return;
1744 };
1745
1746 if message.content.is_empty() {
1747 return;
1748 }
1749
1750 for content in &message.content {
1751 let AgentMessageContent::ToolUse(tool_use) = content else {
1752 continue;
1753 };
1754
1755 if !message.tool_results.contains_key(&tool_use.id) {
1756 message.tool_results.insert(
1757 tool_use.id.clone(),
1758 LanguageModelToolResult {
1759 tool_use_id: tool_use.id.clone(),
1760 tool_name: tool_use.name.clone(),
1761 is_error: true,
1762 content: LanguageModelToolResultContent::Text(TOOL_CANCELED_MESSAGE.into()),
1763 output: None,
1764 },
1765 );
1766 }
1767 }
1768
1769 self.messages.push(Message::Agent(message));
1770 self.updated_at = Utc::now();
1771 self.summary = None;
1772 cx.notify()
1773 }
1774
1775 pub(crate) fn build_completion_request(
1776 &self,
1777 completion_intent: CompletionIntent,
1778 cx: &App,
1779 ) -> Result<LanguageModelRequest> {
1780 let model = self.model().context("No language model configured")?;
1781 let tools = if let Some(turn) = self.running_turn.as_ref() {
1782 turn.tools
1783 .iter()
1784 .filter_map(|(tool_name, tool)| {
1785 log::trace!("Including tool: {}", tool_name);
1786 Some(LanguageModelRequestTool {
1787 name: tool_name.to_string(),
1788 description: tool.description().to_string(),
1789 input_schema: tool.input_schema(model.tool_input_format()).log_err()?,
1790 })
1791 })
1792 .collect::<Vec<_>>()
1793 } else {
1794 Vec::new()
1795 };
1796
1797 log::debug!("Building completion request");
1798 log::debug!("Completion intent: {:?}", completion_intent);
1799 log::debug!("Completion mode: {:?}", self.completion_mode);
1800
1801 let messages = self.build_request_messages(cx);
1802 log::debug!("Request will include {} messages", messages.len());
1803 log::debug!("Request includes {} tools", tools.len());
1804
1805 let request = LanguageModelRequest {
1806 thread_id: Some(self.id.to_string()),
1807 prompt_id: Some(self.prompt_id.to_string()),
1808 intent: Some(completion_intent),
1809 mode: Some(self.completion_mode.into()),
1810 messages,
1811 tools,
1812 tool_choice: None,
1813 stop: Vec::new(),
1814 temperature: AgentSettings::temperature_for_model(model, cx),
1815 thinking_allowed: true,
1816 };
1817
1818 log::debug!("Completion request built successfully");
1819 Ok(request)
1820 }
1821
1822 fn enabled_tools(
1823 &self,
1824 profile: &AgentProfileSettings,
1825 model: &Arc<dyn LanguageModel>,
1826 cx: &App,
1827 ) -> BTreeMap<SharedString, Arc<dyn AnyAgentTool>> {
1828 fn truncate(tool_name: &SharedString) -> SharedString {
1829 if tool_name.len() > MAX_TOOL_NAME_LENGTH {
1830 let mut truncated = tool_name.to_string();
1831 truncated.truncate(MAX_TOOL_NAME_LENGTH);
1832 truncated.into()
1833 } else {
1834 tool_name.clone()
1835 }
1836 }
1837
1838 let mut tools = self
1839 .tools
1840 .iter()
1841 .filter_map(|(tool_name, tool)| {
1842 if tool.supported_provider(&model.provider_id())
1843 && profile.is_tool_enabled(tool_name)
1844 {
1845 Some((truncate(tool_name), tool.clone()))
1846 } else {
1847 None
1848 }
1849 })
1850 .collect::<BTreeMap<_, _>>();
1851
1852 let mut context_server_tools = Vec::new();
1853 let mut seen_tools = tools.keys().cloned().collect::<HashSet<_>>();
1854 let mut duplicate_tool_names = HashSet::default();
1855 for (server_id, server_tools) in self.context_server_registry.read(cx).servers() {
1856 for (tool_name, tool) in server_tools {
1857 if profile.is_context_server_tool_enabled(&server_id.0, &tool_name) {
1858 let tool_name = truncate(tool_name);
1859 if !seen_tools.insert(tool_name.clone()) {
1860 duplicate_tool_names.insert(tool_name.clone());
1861 }
1862 context_server_tools.push((server_id.clone(), tool_name, tool.clone()));
1863 }
1864 }
1865 }
1866
1867 // When there are duplicate tool names, disambiguate by prefixing them
1868 // with the server ID. In the rare case there isn't enough space for the
1869 // disambiguated tool name, keep only the last tool with this name.
1870 for (server_id, tool_name, tool) in context_server_tools {
1871 if duplicate_tool_names.contains(&tool_name) {
1872 let available = MAX_TOOL_NAME_LENGTH.saturating_sub(tool_name.len());
1873 if available >= 2 {
1874 let mut disambiguated = server_id.0.to_string();
1875 disambiguated.truncate(available - 1);
1876 disambiguated.push('_');
1877 disambiguated.push_str(&tool_name);
1878 tools.insert(disambiguated.into(), tool.clone());
1879 } else {
1880 tools.insert(tool_name, tool.clone());
1881 }
1882 } else {
1883 tools.insert(tool_name, tool.clone());
1884 }
1885 }
1886
1887 tools
1888 }
1889
1890 fn tool(&self, name: &str) -> Option<Arc<dyn AnyAgentTool>> {
1891 self.running_turn.as_ref()?.tools.get(name).cloned()
1892 }
1893
1894 fn build_request_messages(&self, cx: &App) -> Vec<LanguageModelRequestMessage> {
1895 log::trace!(
1896 "Building request messages from {} thread messages",
1897 self.messages.len()
1898 );
1899
1900 let system_prompt = SystemPromptTemplate {
1901 project: self.project_context.read(cx),
1902 available_tools: self.tools.keys().cloned().collect(),
1903 }
1904 .render(&self.templates)
1905 .context("failed to build system prompt")
1906 .expect("Invalid template");
1907 let mut messages = vec![LanguageModelRequestMessage {
1908 role: Role::System,
1909 content: vec![system_prompt.into()],
1910 cache: false,
1911 }];
1912 for message in &self.messages {
1913 messages.extend(message.to_request());
1914 }
1915
1916 if let Some(last_message) = messages.last_mut() {
1917 last_message.cache = true;
1918 }
1919
1920 if let Some(message) = self.pending_message.as_ref() {
1921 messages.extend(message.to_request());
1922 }
1923
1924 messages
1925 }
1926
1927 pub fn to_markdown(&self) -> String {
1928 let mut markdown = String::new();
1929 for (ix, message) in self.messages.iter().enumerate() {
1930 if ix > 0 {
1931 markdown.push('\n');
1932 }
1933 markdown.push_str(&message.to_markdown());
1934 }
1935
1936 if let Some(message) = self.pending_message.as_ref() {
1937 markdown.push('\n');
1938 markdown.push_str(&message.to_markdown());
1939 }
1940
1941 markdown
1942 }
1943
1944 fn advance_prompt_id(&mut self) {
1945 self.prompt_id = PromptId::new();
1946 }
1947
1948 fn retry_strategy_for(error: &LanguageModelCompletionError) -> Option<RetryStrategy> {
1949 use LanguageModelCompletionError::*;
1950 use http_client::StatusCode;
1951
1952 // General strategy here:
1953 // - If retrying won't help (e.g. invalid API key or payload too large), return None so we don't retry at all.
1954 // - If it's a time-based issue (e.g. server overloaded, rate limit exceeded), retry up to 4 times with exponential backoff.
1955 // - If it's an issue that *might* be fixed by retrying (e.g. internal server error), retry up to 3 times.
1956 match error {
1957 HttpResponseError {
1958 status_code: StatusCode::TOO_MANY_REQUESTS,
1959 ..
1960 } => Some(RetryStrategy::ExponentialBackoff {
1961 initial_delay: BASE_RETRY_DELAY,
1962 max_attempts: MAX_RETRY_ATTEMPTS,
1963 }),
1964 ServerOverloaded { retry_after, .. } | RateLimitExceeded { retry_after, .. } => {
1965 Some(RetryStrategy::Fixed {
1966 delay: retry_after.unwrap_or(BASE_RETRY_DELAY),
1967 max_attempts: MAX_RETRY_ATTEMPTS,
1968 })
1969 }
1970 UpstreamProviderError {
1971 status,
1972 retry_after,
1973 ..
1974 } => match *status {
1975 StatusCode::TOO_MANY_REQUESTS | StatusCode::SERVICE_UNAVAILABLE => {
1976 Some(RetryStrategy::Fixed {
1977 delay: retry_after.unwrap_or(BASE_RETRY_DELAY),
1978 max_attempts: MAX_RETRY_ATTEMPTS,
1979 })
1980 }
1981 StatusCode::INTERNAL_SERVER_ERROR => Some(RetryStrategy::Fixed {
1982 delay: retry_after.unwrap_or(BASE_RETRY_DELAY),
1983 // Internal Server Error could be anything, retry up to 3 times.
1984 max_attempts: 3,
1985 }),
1986 status => {
1987 // There is no StatusCode variant for the unofficial HTTP 529 ("The service is overloaded"),
1988 // but we frequently get them in practice. See https://http.dev/529
1989 if status.as_u16() == 529 {
1990 Some(RetryStrategy::Fixed {
1991 delay: retry_after.unwrap_or(BASE_RETRY_DELAY),
1992 max_attempts: MAX_RETRY_ATTEMPTS,
1993 })
1994 } else {
1995 Some(RetryStrategy::Fixed {
1996 delay: retry_after.unwrap_or(BASE_RETRY_DELAY),
1997 max_attempts: 2,
1998 })
1999 }
2000 }
2001 },
2002 ApiInternalServerError { .. } => Some(RetryStrategy::Fixed {
2003 delay: BASE_RETRY_DELAY,
2004 max_attempts: 3,
2005 }),
2006 ApiReadResponseError { .. }
2007 | HttpSend { .. }
2008 | DeserializeResponse { .. }
2009 | BadRequestFormat { .. } => Some(RetryStrategy::Fixed {
2010 delay: BASE_RETRY_DELAY,
2011 max_attempts: 3,
2012 }),
2013 // Retrying these errors definitely shouldn't help.
2014 HttpResponseError {
2015 status_code:
2016 StatusCode::PAYLOAD_TOO_LARGE | StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED,
2017 ..
2018 }
2019 | AuthenticationError { .. }
2020 | PermissionError { .. }
2021 | NoApiKey { .. }
2022 | ApiEndpointNotFound { .. }
2023 | PromptTooLarge { .. } => None,
2024 // These errors might be transient, so retry them
2025 SerializeRequest { .. } | BuildRequestBody { .. } => Some(RetryStrategy::Fixed {
2026 delay: BASE_RETRY_DELAY,
2027 max_attempts: 1,
2028 }),
2029 // Retry all other 4xx and 5xx errors once.
2030 HttpResponseError { status_code, .. }
2031 if status_code.is_client_error() || status_code.is_server_error() =>
2032 {
2033 Some(RetryStrategy::Fixed {
2034 delay: BASE_RETRY_DELAY,
2035 max_attempts: 3,
2036 })
2037 }
2038 Other(err)
2039 if err.is::<language_model::PaymentRequiredError>()
2040 || err.is::<language_model::ModelRequestLimitReachedError>() =>
2041 {
2042 // Retrying won't help for Payment Required or Model Request Limit errors (where
2043 // the user must upgrade to usage-based billing to get more requests, or else wait
2044 // for a significant amount of time for the request limit to reset).
2045 None
2046 }
2047 // Conservatively assume that any other errors are non-retryable
2048 HttpResponseError { .. } | Other(..) => Some(RetryStrategy::Fixed {
2049 delay: BASE_RETRY_DELAY,
2050 max_attempts: 2,
2051 }),
2052 }
2053 }
2054}
2055
2056struct RunningTurn {
2057 /// Holds the task that handles agent interaction until the end of the turn.
2058 /// Survives across multiple requests as the model performs tool calls and
2059 /// we run tools, report their results.
2060 _task: Task<()>,
2061 /// The current event stream for the running turn. Used to report a final
2062 /// cancellation event if we cancel the turn.
2063 event_stream: ThreadEventStream,
2064 /// The tools that were enabled for this turn.
2065 tools: BTreeMap<SharedString, Arc<dyn AnyAgentTool>>,
2066}
2067
2068impl RunningTurn {
2069 fn cancel(self) {
2070 log::debug!("Cancelling in progress turn");
2071 self.event_stream.send_canceled();
2072 }
2073}
2074
2075pub struct TokenUsageUpdated(pub Option<acp_thread::TokenUsage>);
2076
2077impl EventEmitter<TokenUsageUpdated> for Thread {}
2078
2079pub struct TitleUpdated;
2080
2081impl EventEmitter<TitleUpdated> for Thread {}
2082
2083pub trait AgentTool
2084where
2085 Self: 'static + Sized,
2086{
2087 type Input: for<'de> Deserialize<'de> + Serialize + JsonSchema;
2088 type Output: for<'de> Deserialize<'de> + Serialize + Into<LanguageModelToolResultContent>;
2089
2090 fn name() -> &'static str;
2091
2092 fn description(&self) -> SharedString {
2093 let schema = schemars::schema_for!(Self::Input);
2094 SharedString::new(
2095 schema
2096 .get("description")
2097 .and_then(|description| description.as_str())
2098 .unwrap_or_default(),
2099 )
2100 }
2101
2102 fn kind() -> acp::ToolKind;
2103
2104 /// The initial tool title to display. Can be updated during the tool run.
2105 fn initial_title(&self, input: Result<Self::Input, serde_json::Value>) -> SharedString;
2106
2107 /// Returns the JSON schema that describes the tool's input.
2108 fn input_schema(&self, format: LanguageModelToolSchemaFormat) -> Schema {
2109 crate::tool_schema::root_schema_for::<Self::Input>(format)
2110 }
2111
2112 /// Some tools rely on a provider for the underlying billing or other reasons.
2113 /// Allow the tool to check if they are compatible, or should be filtered out.
2114 fn supported_provider(&self, _provider: &LanguageModelProviderId) -> bool {
2115 true
2116 }
2117
2118 /// Runs the tool with the provided input.
2119 fn run(
2120 self: Arc<Self>,
2121 input: Self::Input,
2122 event_stream: ToolCallEventStream,
2123 cx: &mut App,
2124 ) -> Task<Result<Self::Output>>;
2125
2126 /// Emits events for a previous execution of the tool.
2127 fn replay(
2128 &self,
2129 _input: Self::Input,
2130 _output: Self::Output,
2131 _event_stream: ToolCallEventStream,
2132 _cx: &mut App,
2133 ) -> Result<()> {
2134 Ok(())
2135 }
2136
2137 fn erase(self) -> Arc<dyn AnyAgentTool> {
2138 Arc::new(Erased(Arc::new(self)))
2139 }
2140}
2141
2142pub struct Erased<T>(T);
2143
2144pub struct AgentToolOutput {
2145 pub llm_output: LanguageModelToolResultContent,
2146 pub raw_output: serde_json::Value,
2147}
2148
2149pub trait AnyAgentTool {
2150 fn name(&self) -> SharedString;
2151 fn description(&self) -> SharedString;
2152 fn kind(&self) -> acp::ToolKind;
2153 fn initial_title(&self, input: serde_json::Value) -> SharedString;
2154 fn input_schema(&self, format: LanguageModelToolSchemaFormat) -> Result<serde_json::Value>;
2155 fn supported_provider(&self, _provider: &LanguageModelProviderId) -> bool {
2156 true
2157 }
2158 fn run(
2159 self: Arc<Self>,
2160 input: serde_json::Value,
2161 event_stream: ToolCallEventStream,
2162 cx: &mut App,
2163 ) -> Task<Result<AgentToolOutput>>;
2164 fn replay(
2165 &self,
2166 input: serde_json::Value,
2167 output: serde_json::Value,
2168 event_stream: ToolCallEventStream,
2169 cx: &mut App,
2170 ) -> Result<()>;
2171}
2172
2173impl<T> AnyAgentTool for Erased<Arc<T>>
2174where
2175 T: AgentTool,
2176{
2177 fn name(&self) -> SharedString {
2178 T::name().into()
2179 }
2180
2181 fn description(&self) -> SharedString {
2182 self.0.description()
2183 }
2184
2185 fn kind(&self) -> agent_client_protocol::ToolKind {
2186 T::kind()
2187 }
2188
2189 fn initial_title(&self, input: serde_json::Value) -> SharedString {
2190 let parsed_input = serde_json::from_value(input.clone()).map_err(|_| input);
2191 self.0.initial_title(parsed_input)
2192 }
2193
2194 fn input_schema(&self, format: LanguageModelToolSchemaFormat) -> Result<serde_json::Value> {
2195 let mut json = serde_json::to_value(self.0.input_schema(format))?;
2196 adapt_schema_to_format(&mut json, format)?;
2197 Ok(json)
2198 }
2199
2200 fn supported_provider(&self, provider: &LanguageModelProviderId) -> bool {
2201 self.0.supported_provider(provider)
2202 }
2203
2204 fn run(
2205 self: Arc<Self>,
2206 input: serde_json::Value,
2207 event_stream: ToolCallEventStream,
2208 cx: &mut App,
2209 ) -> Task<Result<AgentToolOutput>> {
2210 cx.spawn(async move |cx| {
2211 let input = serde_json::from_value(input)?;
2212 let output = cx
2213 .update(|cx| self.0.clone().run(input, event_stream, cx))?
2214 .await?;
2215 let raw_output = serde_json::to_value(&output)?;
2216 Ok(AgentToolOutput {
2217 llm_output: output.into(),
2218 raw_output,
2219 })
2220 })
2221 }
2222
2223 fn replay(
2224 &self,
2225 input: serde_json::Value,
2226 output: serde_json::Value,
2227 event_stream: ToolCallEventStream,
2228 cx: &mut App,
2229 ) -> Result<()> {
2230 let input = serde_json::from_value(input)?;
2231 let output = serde_json::from_value(output)?;
2232 self.0.replay(input, output, event_stream, cx)
2233 }
2234}
2235
2236#[derive(Clone)]
2237struct ThreadEventStream(mpsc::UnboundedSender<Result<ThreadEvent>>);
2238
2239impl ThreadEventStream {
2240 fn send_user_message(&self, message: &UserMessage) {
2241 self.0
2242 .unbounded_send(Ok(ThreadEvent::UserMessage(message.clone())))
2243 .ok();
2244 }
2245
2246 fn send_text(&self, text: &str) {
2247 self.0
2248 .unbounded_send(Ok(ThreadEvent::AgentText(text.to_string())))
2249 .ok();
2250 }
2251
2252 fn send_thinking(&self, text: &str) {
2253 self.0
2254 .unbounded_send(Ok(ThreadEvent::AgentThinking(text.to_string())))
2255 .ok();
2256 }
2257
2258 fn send_tool_call(
2259 &self,
2260 id: &LanguageModelToolUseId,
2261 title: SharedString,
2262 kind: acp::ToolKind,
2263 input: serde_json::Value,
2264 ) {
2265 self.0
2266 .unbounded_send(Ok(ThreadEvent::ToolCall(Self::initial_tool_call(
2267 id,
2268 title.to_string(),
2269 kind,
2270 input,
2271 ))))
2272 .ok();
2273 }
2274
2275 fn initial_tool_call(
2276 id: &LanguageModelToolUseId,
2277 title: String,
2278 kind: acp::ToolKind,
2279 input: serde_json::Value,
2280 ) -> acp::ToolCall {
2281 acp::ToolCall {
2282 id: acp::ToolCallId(id.to_string().into()),
2283 title,
2284 kind,
2285 status: acp::ToolCallStatus::Pending,
2286 content: vec![],
2287 locations: vec![],
2288 raw_input: Some(input),
2289 raw_output: None,
2290 }
2291 }
2292
2293 fn update_tool_call_fields(
2294 &self,
2295 tool_use_id: &LanguageModelToolUseId,
2296 fields: acp::ToolCallUpdateFields,
2297 ) {
2298 self.0
2299 .unbounded_send(Ok(ThreadEvent::ToolCallUpdate(
2300 acp::ToolCallUpdate {
2301 id: acp::ToolCallId(tool_use_id.to_string().into()),
2302 fields,
2303 }
2304 .into(),
2305 )))
2306 .ok();
2307 }
2308
2309 fn send_retry(&self, status: acp_thread::RetryStatus) {
2310 self.0.unbounded_send(Ok(ThreadEvent::Retry(status))).ok();
2311 }
2312
2313 fn send_stop(&self, reason: acp::StopReason) {
2314 self.0.unbounded_send(Ok(ThreadEvent::Stop(reason))).ok();
2315 }
2316
2317 fn send_canceled(&self) {
2318 self.0
2319 .unbounded_send(Ok(ThreadEvent::Stop(acp::StopReason::Cancelled)))
2320 .ok();
2321 }
2322
2323 fn send_error(&self, error: impl Into<anyhow::Error>) {
2324 self.0.unbounded_send(Err(error.into())).ok();
2325 }
2326}
2327
2328#[derive(Clone)]
2329pub struct ToolCallEventStream {
2330 tool_use_id: LanguageModelToolUseId,
2331 stream: ThreadEventStream,
2332 fs: Option<Arc<dyn Fs>>,
2333}
2334
2335impl ToolCallEventStream {
2336 #[cfg(test)]
2337 pub fn test() -> (Self, ToolCallEventStreamReceiver) {
2338 let (events_tx, events_rx) = mpsc::unbounded::<Result<ThreadEvent>>();
2339
2340 let stream = ToolCallEventStream::new("test_id".into(), ThreadEventStream(events_tx), None);
2341
2342 (stream, ToolCallEventStreamReceiver(events_rx))
2343 }
2344
2345 fn new(
2346 tool_use_id: LanguageModelToolUseId,
2347 stream: ThreadEventStream,
2348 fs: Option<Arc<dyn Fs>>,
2349 ) -> Self {
2350 Self {
2351 tool_use_id,
2352 stream,
2353 fs,
2354 }
2355 }
2356
2357 pub fn update_fields(&self, fields: acp::ToolCallUpdateFields) {
2358 self.stream
2359 .update_tool_call_fields(&self.tool_use_id, fields);
2360 }
2361
2362 pub fn update_diff(&self, diff: Entity<acp_thread::Diff>) {
2363 self.stream
2364 .0
2365 .unbounded_send(Ok(ThreadEvent::ToolCallUpdate(
2366 acp_thread::ToolCallUpdateDiff {
2367 id: acp::ToolCallId(self.tool_use_id.to_string().into()),
2368 diff,
2369 }
2370 .into(),
2371 )))
2372 .ok();
2373 }
2374
2375 pub fn update_terminal(&self, terminal: Entity<acp_thread::Terminal>) {
2376 self.stream
2377 .0
2378 .unbounded_send(Ok(ThreadEvent::ToolCallUpdate(
2379 acp_thread::ToolCallUpdateTerminal {
2380 id: acp::ToolCallId(self.tool_use_id.to_string().into()),
2381 terminal,
2382 }
2383 .into(),
2384 )))
2385 .ok();
2386 }
2387
2388 pub fn authorize(&self, title: impl Into<String>, cx: &mut App) -> Task<Result<()>> {
2389 if agent_settings::AgentSettings::get_global(cx).always_allow_tool_actions {
2390 return Task::ready(Ok(()));
2391 }
2392
2393 let (response_tx, response_rx) = oneshot::channel();
2394 self.stream
2395 .0
2396 .unbounded_send(Ok(ThreadEvent::ToolCallAuthorization(
2397 ToolCallAuthorization {
2398 tool_call: acp::ToolCallUpdate {
2399 id: acp::ToolCallId(self.tool_use_id.to_string().into()),
2400 fields: acp::ToolCallUpdateFields {
2401 title: Some(title.into()),
2402 ..Default::default()
2403 },
2404 },
2405 options: vec![
2406 acp::PermissionOption {
2407 id: acp::PermissionOptionId("always_allow".into()),
2408 name: "Always Allow".into(),
2409 kind: acp::PermissionOptionKind::AllowAlways,
2410 },
2411 acp::PermissionOption {
2412 id: acp::PermissionOptionId("allow".into()),
2413 name: "Allow".into(),
2414 kind: acp::PermissionOptionKind::AllowOnce,
2415 },
2416 acp::PermissionOption {
2417 id: acp::PermissionOptionId("deny".into()),
2418 name: "Deny".into(),
2419 kind: acp::PermissionOptionKind::RejectOnce,
2420 },
2421 ],
2422 response: response_tx,
2423 },
2424 )))
2425 .ok();
2426 let fs = self.fs.clone();
2427 cx.spawn(async move |cx| match response_rx.await?.0.as_ref() {
2428 "always_allow" => {
2429 if let Some(fs) = fs.clone() {
2430 cx.update(|cx| {
2431 update_settings_file::<AgentSettings>(fs, cx, |settings, _| {
2432 settings.set_always_allow_tool_actions(true);
2433 });
2434 })?;
2435 }
2436
2437 Ok(())
2438 }
2439 "allow" => Ok(()),
2440 _ => Err(anyhow!("Permission to run tool denied by user")),
2441 })
2442 }
2443}
2444
2445#[cfg(test)]
2446pub struct ToolCallEventStreamReceiver(mpsc::UnboundedReceiver<Result<ThreadEvent>>);
2447
2448#[cfg(test)]
2449impl ToolCallEventStreamReceiver {
2450 pub async fn expect_authorization(&mut self) -> ToolCallAuthorization {
2451 let event = self.0.next().await;
2452 if let Some(Ok(ThreadEvent::ToolCallAuthorization(auth))) = event {
2453 auth
2454 } else {
2455 panic!("Expected ToolCallAuthorization but got: {:?}", event);
2456 }
2457 }
2458
2459 pub async fn expect_terminal(&mut self) -> Entity<acp_thread::Terminal> {
2460 let event = self.0.next().await;
2461 if let Some(Ok(ThreadEvent::ToolCallUpdate(acp_thread::ToolCallUpdate::UpdateTerminal(
2462 update,
2463 )))) = event
2464 {
2465 update.terminal
2466 } else {
2467 panic!("Expected terminal but got: {:?}", event);
2468 }
2469 }
2470}
2471
2472#[cfg(test)]
2473impl std::ops::Deref for ToolCallEventStreamReceiver {
2474 type Target = mpsc::UnboundedReceiver<Result<ThreadEvent>>;
2475
2476 fn deref(&self) -> &Self::Target {
2477 &self.0
2478 }
2479}
2480
2481#[cfg(test)]
2482impl std::ops::DerefMut for ToolCallEventStreamReceiver {
2483 fn deref_mut(&mut self) -> &mut Self::Target {
2484 &mut self.0
2485 }
2486}
2487
2488impl From<&str> for UserMessageContent {
2489 fn from(text: &str) -> Self {
2490 Self::Text(text.into())
2491 }
2492}
2493
2494impl From<acp::ContentBlock> for UserMessageContent {
2495 fn from(value: acp::ContentBlock) -> Self {
2496 match value {
2497 acp::ContentBlock::Text(text_content) => Self::Text(text_content.text),
2498 acp::ContentBlock::Image(image_content) => Self::Image(convert_image(image_content)),
2499 acp::ContentBlock::Audio(_) => {
2500 // TODO
2501 Self::Text("[audio]".to_string())
2502 }
2503 acp::ContentBlock::ResourceLink(resource_link) => {
2504 match MentionUri::parse(&resource_link.uri) {
2505 Ok(uri) => Self::Mention {
2506 uri,
2507 content: String::new(),
2508 },
2509 Err(err) => {
2510 log::error!("Failed to parse mention link: {}", err);
2511 Self::Text(format!("[{}]({})", resource_link.name, resource_link.uri))
2512 }
2513 }
2514 }
2515 acp::ContentBlock::Resource(resource) => match resource.resource {
2516 acp::EmbeddedResourceResource::TextResourceContents(resource) => {
2517 match MentionUri::parse(&resource.uri) {
2518 Ok(uri) => Self::Mention {
2519 uri,
2520 content: resource.text,
2521 },
2522 Err(err) => {
2523 log::error!("Failed to parse mention link: {}", err);
2524 Self::Text(
2525 MarkdownCodeBlock {
2526 tag: &resource.uri,
2527 text: &resource.text,
2528 }
2529 .to_string(),
2530 )
2531 }
2532 }
2533 }
2534 acp::EmbeddedResourceResource::BlobResourceContents(_) => {
2535 // TODO
2536 Self::Text("[blob]".to_string())
2537 }
2538 },
2539 }
2540 }
2541}
2542
2543impl From<UserMessageContent> for acp::ContentBlock {
2544 fn from(content: UserMessageContent) -> Self {
2545 match content {
2546 UserMessageContent::Text(text) => acp::ContentBlock::Text(acp::TextContent {
2547 text,
2548 annotations: None,
2549 }),
2550 UserMessageContent::Image(image) => acp::ContentBlock::Image(acp::ImageContent {
2551 data: image.source.to_string(),
2552 mime_type: "image/png".to_string(),
2553 annotations: None,
2554 uri: None,
2555 }),
2556 UserMessageContent::Mention { uri, content } => {
2557 acp::ContentBlock::Resource(acp::EmbeddedResource {
2558 resource: acp::EmbeddedResourceResource::TextResourceContents(
2559 acp::TextResourceContents {
2560 mime_type: None,
2561 text: content,
2562 uri: uri.to_uri().to_string(),
2563 },
2564 ),
2565 annotations: None,
2566 })
2567 }
2568 }
2569 }
2570}
2571
2572fn convert_image(image_content: acp::ImageContent) -> LanguageModelImage {
2573 LanguageModelImage {
2574 source: image_content.data.into(),
2575 // TODO: make this optional?
2576 size: gpui::Size::new(0.into(), 0.into()),
2577 }
2578}