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