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