1mod connection;
   2mod diff;
   3mod mention;
   4mod terminal;
   5
   6use agent_settings::AgentSettings;
   7use collections::HashSet;
   8pub use connection::*;
   9pub use diff::*;
  10use language::language_settings::FormatOnSave;
  11pub use mention::*;
  12use project::lsp_store::{FormatTrigger, LspFormatTarget};
  13use serde::{Deserialize, Serialize};
  14use settings::Settings as _;
  15use task::{Shell, ShellBuilder};
  16pub use terminal::*;
  17
  18use action_log::ActionLog;
  19use agent_client_protocol::{self as acp};
  20use anyhow::{Context as _, Result, anyhow};
  21use editor::Bias;
  22use futures::{FutureExt, channel::oneshot, future::BoxFuture};
  23use gpui::{AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Task, WeakEntity};
  24use itertools::Itertools;
  25use language::{Anchor, Buffer, BufferSnapshot, LanguageRegistry, Point, ToPoint, text_diff};
  26use markdown::Markdown;
  27use project::{AgentLocation, Project, git_store::GitStoreCheckpoint};
  28use std::collections::HashMap;
  29use std::error::Error;
  30use std::fmt::{Formatter, Write};
  31use std::ops::Range;
  32use std::process::ExitStatus;
  33use std::rc::Rc;
  34use std::time::{Duration, Instant};
  35use std::{fmt::Display, mem, path::PathBuf, sync::Arc};
  36use ui::App;
  37use util::{ResultExt, get_default_system_shell};
  38use uuid::Uuid;
  39
  40#[derive(Debug)]
  41pub struct UserMessage {
  42    pub id: Option<UserMessageId>,
  43    pub content: ContentBlock,
  44    pub chunks: Vec<acp::ContentBlock>,
  45    pub checkpoint: Option<Checkpoint>,
  46}
  47
  48#[derive(Debug)]
  49pub struct Checkpoint {
  50    git_checkpoint: GitStoreCheckpoint,
  51    pub show: bool,
  52}
  53
  54impl UserMessage {
  55    fn to_markdown(&self, cx: &App) -> String {
  56        let mut markdown = String::new();
  57        if self
  58            .checkpoint
  59            .as_ref()
  60            .is_some_and(|checkpoint| checkpoint.show)
  61        {
  62            writeln!(markdown, "## User (checkpoint)").unwrap();
  63        } else {
  64            writeln!(markdown, "## User").unwrap();
  65        }
  66        writeln!(markdown).unwrap();
  67        writeln!(markdown, "{}", self.content.to_markdown(cx)).unwrap();
  68        writeln!(markdown).unwrap();
  69        markdown
  70    }
  71}
  72
  73#[derive(Debug, PartialEq)]
  74pub struct AssistantMessage {
  75    pub chunks: Vec<AssistantMessageChunk>,
  76}
  77
  78impl AssistantMessage {
  79    pub fn to_markdown(&self, cx: &App) -> String {
  80        format!(
  81            "## Assistant\n\n{}\n\n",
  82            self.chunks
  83                .iter()
  84                .map(|chunk| chunk.to_markdown(cx))
  85                .join("\n\n")
  86        )
  87    }
  88}
  89
  90#[derive(Debug, PartialEq)]
  91pub enum AssistantMessageChunk {
  92    Message { block: ContentBlock },
  93    Thought { block: ContentBlock },
  94}
  95
  96impl AssistantMessageChunk {
  97    pub fn from_str(chunk: &str, language_registry: &Arc<LanguageRegistry>, cx: &mut App) -> Self {
  98        Self::Message {
  99            block: ContentBlock::new(chunk.into(), language_registry, cx),
 100        }
 101    }
 102
 103    fn to_markdown(&self, cx: &App) -> String {
 104        match self {
 105            Self::Message { block } => block.to_markdown(cx).to_string(),
 106            Self::Thought { block } => {
 107                format!("<thinking>\n{}\n</thinking>", block.to_markdown(cx))
 108            }
 109        }
 110    }
 111}
 112
 113#[derive(Debug)]
 114pub enum AgentThreadEntry {
 115    UserMessage(UserMessage),
 116    AssistantMessage(AssistantMessage),
 117    ToolCall(ToolCall),
 118}
 119
 120impl AgentThreadEntry {
 121    pub fn to_markdown(&self, cx: &App) -> String {
 122        match self {
 123            Self::UserMessage(message) => message.to_markdown(cx),
 124            Self::AssistantMessage(message) => message.to_markdown(cx),
 125            Self::ToolCall(tool_call) => tool_call.to_markdown(cx),
 126        }
 127    }
 128
 129    pub fn user_message(&self) -> Option<&UserMessage> {
 130        if let AgentThreadEntry::UserMessage(message) = self {
 131            Some(message)
 132        } else {
 133            None
 134        }
 135    }
 136
 137    pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
 138        if let AgentThreadEntry::ToolCall(call) = self {
 139            itertools::Either::Left(call.diffs())
 140        } else {
 141            itertools::Either::Right(std::iter::empty())
 142        }
 143    }
 144
 145    pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
 146        if let AgentThreadEntry::ToolCall(call) = self {
 147            itertools::Either::Left(call.terminals())
 148        } else {
 149            itertools::Either::Right(std::iter::empty())
 150        }
 151    }
 152
 153    pub fn location(&self, ix: usize) -> Option<(acp::ToolCallLocation, AgentLocation)> {
 154        if let AgentThreadEntry::ToolCall(ToolCall {
 155            locations,
 156            resolved_locations,
 157            ..
 158        }) = self
 159        {
 160            Some((
 161                locations.get(ix)?.clone(),
 162                resolved_locations.get(ix)?.clone()?,
 163            ))
 164        } else {
 165            None
 166        }
 167    }
 168}
 169
 170#[derive(Debug)]
 171pub struct ToolCall {
 172    pub id: acp::ToolCallId,
 173    pub label: Entity<Markdown>,
 174    pub kind: acp::ToolKind,
 175    pub content: Vec<ToolCallContent>,
 176    pub status: ToolCallStatus,
 177    pub locations: Vec<acp::ToolCallLocation>,
 178    pub resolved_locations: Vec<Option<AgentLocation>>,
 179    pub raw_input: Option<serde_json::Value>,
 180    pub raw_output: Option<serde_json::Value>,
 181}
 182
 183impl ToolCall {
 184    fn from_acp(
 185        tool_call: acp::ToolCall,
 186        status: ToolCallStatus,
 187        language_registry: Arc<LanguageRegistry>,
 188        terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
 189        cx: &mut App,
 190    ) -> Result<Self> {
 191        let title = if let Some((first_line, _)) = tool_call.title.split_once("\n") {
 192            first_line.to_owned() + "…"
 193        } else {
 194            tool_call.title
 195        };
 196        let mut content = Vec::with_capacity(tool_call.content.len());
 197        for item in tool_call.content {
 198            content.push(ToolCallContent::from_acp(
 199                item,
 200                language_registry.clone(),
 201                terminals,
 202                cx,
 203            )?);
 204        }
 205
 206        let result = Self {
 207            id: tool_call.id,
 208            label: cx
 209                .new(|cx| Markdown::new(title.into(), Some(language_registry.clone()), None, cx)),
 210            kind: tool_call.kind,
 211            content,
 212            locations: tool_call.locations,
 213            resolved_locations: Vec::default(),
 214            status,
 215            raw_input: tool_call.raw_input,
 216            raw_output: tool_call.raw_output,
 217        };
 218        Ok(result)
 219    }
 220
 221    fn update_fields(
 222        &mut self,
 223        fields: acp::ToolCallUpdateFields,
 224        language_registry: Arc<LanguageRegistry>,
 225        terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
 226        cx: &mut App,
 227    ) -> Result<()> {
 228        let acp::ToolCallUpdateFields {
 229            kind,
 230            status,
 231            title,
 232            content,
 233            locations,
 234            raw_input,
 235            raw_output,
 236        } = fields;
 237
 238        if let Some(kind) = kind {
 239            self.kind = kind;
 240        }
 241
 242        if let Some(status) = status {
 243            self.status = status.into();
 244        }
 245
 246        if let Some(title) = title {
 247            self.label.update(cx, |label, cx| {
 248                if let Some((first_line, _)) = title.split_once("\n") {
 249                    label.replace(first_line.to_owned() + "…", cx)
 250                } else {
 251                    label.replace(title, cx);
 252                }
 253            });
 254        }
 255
 256        if let Some(content) = content {
 257            let new_content_len = content.len();
 258            let mut content = content.into_iter();
 259
 260            // Reuse existing content if we can
 261            for (old, new) in self.content.iter_mut().zip(content.by_ref()) {
 262                old.update_from_acp(new, language_registry.clone(), terminals, cx)?;
 263            }
 264            for new in content {
 265                self.content.push(ToolCallContent::from_acp(
 266                    new,
 267                    language_registry.clone(),
 268                    terminals,
 269                    cx,
 270                )?)
 271            }
 272            self.content.truncate(new_content_len);
 273        }
 274
 275        if let Some(locations) = locations {
 276            self.locations = locations;
 277        }
 278
 279        if let Some(raw_input) = raw_input {
 280            self.raw_input = Some(raw_input);
 281        }
 282
 283        if let Some(raw_output) = raw_output {
 284            if self.content.is_empty()
 285                && let Some(markdown) = markdown_for_raw_output(&raw_output, &language_registry, cx)
 286            {
 287                self.content
 288                    .push(ToolCallContent::ContentBlock(ContentBlock::Markdown {
 289                        markdown,
 290                    }));
 291            }
 292            self.raw_output = Some(raw_output);
 293        }
 294        Ok(())
 295    }
 296
 297    pub fn diffs(&self) -> impl Iterator<Item = &Entity<Diff>> {
 298        self.content.iter().filter_map(|content| match content {
 299            ToolCallContent::Diff(diff) => Some(diff),
 300            ToolCallContent::ContentBlock(_) => None,
 301            ToolCallContent::Terminal(_) => None,
 302        })
 303    }
 304
 305    pub fn terminals(&self) -> impl Iterator<Item = &Entity<Terminal>> {
 306        self.content.iter().filter_map(|content| match content {
 307            ToolCallContent::Terminal(terminal) => Some(terminal),
 308            ToolCallContent::ContentBlock(_) => None,
 309            ToolCallContent::Diff(_) => None,
 310        })
 311    }
 312
 313    fn to_markdown(&self, cx: &App) -> String {
 314        let mut markdown = format!(
 315            "**Tool Call: {}**\nStatus: {}\n\n",
 316            self.label.read(cx).source(),
 317            self.status
 318        );
 319        for content in &self.content {
 320            markdown.push_str(content.to_markdown(cx).as_str());
 321            markdown.push_str("\n\n");
 322        }
 323        markdown
 324    }
 325
 326    async fn resolve_location(
 327        location: acp::ToolCallLocation,
 328        project: WeakEntity<Project>,
 329        cx: &mut AsyncApp,
 330    ) -> Option<AgentLocation> {
 331        let buffer = project
 332            .update(cx, |project, cx| {
 333                project
 334                    .project_path_for_absolute_path(&location.path, cx)
 335                    .map(|path| project.open_buffer(path, cx))
 336            })
 337            .ok()??;
 338        let buffer = buffer.await.log_err()?;
 339        let position = buffer
 340            .update(cx, |buffer, _| {
 341                if let Some(row) = location.line {
 342                    let snapshot = buffer.snapshot();
 343                    let column = snapshot.indent_size_for_line(row).len;
 344                    let point = snapshot.clip_point(Point::new(row, column), Bias::Left);
 345                    snapshot.anchor_before(point)
 346                } else {
 347                    Anchor::MIN
 348                }
 349            })
 350            .ok()?;
 351
 352        Some(AgentLocation {
 353            buffer: buffer.downgrade(),
 354            position,
 355        })
 356    }
 357
 358    fn resolve_locations(
 359        &self,
 360        project: Entity<Project>,
 361        cx: &mut App,
 362    ) -> Task<Vec<Option<AgentLocation>>> {
 363        let locations = self.locations.clone();
 364        project.update(cx, |_, cx| {
 365            cx.spawn(async move |project, cx| {
 366                let mut new_locations = Vec::new();
 367                for location in locations {
 368                    new_locations.push(Self::resolve_location(location, project.clone(), cx).await);
 369                }
 370                new_locations
 371            })
 372        })
 373    }
 374}
 375
 376#[derive(Debug)]
 377pub enum ToolCallStatus {
 378    /// The tool call hasn't started running yet, but we start showing it to
 379    /// the user.
 380    Pending,
 381    /// The tool call is waiting for confirmation from the user.
 382    WaitingForConfirmation {
 383        options: Vec<acp::PermissionOption>,
 384        respond_tx: oneshot::Sender<acp::PermissionOptionId>,
 385    },
 386    /// The tool call is currently running.
 387    InProgress,
 388    /// The tool call completed successfully.
 389    Completed,
 390    /// The tool call failed.
 391    Failed,
 392    /// The user rejected the tool call.
 393    Rejected,
 394    /// The user canceled generation so the tool call was canceled.
 395    Canceled,
 396}
 397
 398impl From<acp::ToolCallStatus> for ToolCallStatus {
 399    fn from(status: acp::ToolCallStatus) -> Self {
 400        match status {
 401            acp::ToolCallStatus::Pending => Self::Pending,
 402            acp::ToolCallStatus::InProgress => Self::InProgress,
 403            acp::ToolCallStatus::Completed => Self::Completed,
 404            acp::ToolCallStatus::Failed => Self::Failed,
 405        }
 406    }
 407}
 408
 409impl Display for ToolCallStatus {
 410    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
 411        write!(
 412            f,
 413            "{}",
 414            match self {
 415                ToolCallStatus::Pending => "Pending",
 416                ToolCallStatus::WaitingForConfirmation { .. } => "Waiting for confirmation",
 417                ToolCallStatus::InProgress => "In Progress",
 418                ToolCallStatus::Completed => "Completed",
 419                ToolCallStatus::Failed => "Failed",
 420                ToolCallStatus::Rejected => "Rejected",
 421                ToolCallStatus::Canceled => "Canceled",
 422            }
 423        )
 424    }
 425}
 426
 427#[derive(Debug, PartialEq, Clone)]
 428pub enum ContentBlock {
 429    Empty,
 430    Markdown { markdown: Entity<Markdown> },
 431    ResourceLink { resource_link: acp::ResourceLink },
 432}
 433
 434impl ContentBlock {
 435    pub fn new(
 436        block: acp::ContentBlock,
 437        language_registry: &Arc<LanguageRegistry>,
 438        cx: &mut App,
 439    ) -> Self {
 440        let mut this = Self::Empty;
 441        this.append(block, language_registry, cx);
 442        this
 443    }
 444
 445    pub fn new_combined(
 446        blocks: impl IntoIterator<Item = acp::ContentBlock>,
 447        language_registry: Arc<LanguageRegistry>,
 448        cx: &mut App,
 449    ) -> Self {
 450        let mut this = Self::Empty;
 451        for block in blocks {
 452            this.append(block, &language_registry, cx);
 453        }
 454        this
 455    }
 456
 457    pub fn append(
 458        &mut self,
 459        block: acp::ContentBlock,
 460        language_registry: &Arc<LanguageRegistry>,
 461        cx: &mut App,
 462    ) {
 463        if matches!(self, ContentBlock::Empty)
 464            && let acp::ContentBlock::ResourceLink(resource_link) = block
 465        {
 466            *self = ContentBlock::ResourceLink { resource_link };
 467            return;
 468        }
 469
 470        let new_content = self.block_string_contents(block);
 471
 472        match self {
 473            ContentBlock::Empty => {
 474                *self = Self::create_markdown_block(new_content, language_registry, cx);
 475            }
 476            ContentBlock::Markdown { markdown } => {
 477                markdown.update(cx, |markdown, cx| markdown.append(&new_content, cx));
 478            }
 479            ContentBlock::ResourceLink { resource_link } => {
 480                let existing_content = Self::resource_link_md(&resource_link.uri);
 481                let combined = format!("{}\n{}", existing_content, new_content);
 482
 483                *self = Self::create_markdown_block(combined, language_registry, cx);
 484            }
 485        }
 486    }
 487
 488    fn create_markdown_block(
 489        content: String,
 490        language_registry: &Arc<LanguageRegistry>,
 491        cx: &mut App,
 492    ) -> ContentBlock {
 493        ContentBlock::Markdown {
 494            markdown: cx
 495                .new(|cx| Markdown::new(content.into(), Some(language_registry.clone()), None, cx)),
 496        }
 497    }
 498
 499    fn block_string_contents(&self, block: acp::ContentBlock) -> String {
 500        match block {
 501            acp::ContentBlock::Text(text_content) => text_content.text,
 502            acp::ContentBlock::ResourceLink(resource_link) => {
 503                Self::resource_link_md(&resource_link.uri)
 504            }
 505            acp::ContentBlock::Resource(acp::EmbeddedResource {
 506                resource:
 507                    acp::EmbeddedResourceResource::TextResourceContents(acp::TextResourceContents {
 508                        uri,
 509                        ..
 510                    }),
 511                ..
 512            }) => Self::resource_link_md(&uri),
 513            acp::ContentBlock::Image(image) => Self::image_md(&image),
 514            acp::ContentBlock::Audio(_) | acp::ContentBlock::Resource(_) => String::new(),
 515        }
 516    }
 517
 518    fn resource_link_md(uri: &str) -> String {
 519        if let Some(uri) = MentionUri::parse(uri).log_err() {
 520            uri.as_link().to_string()
 521        } else {
 522            uri.to_string()
 523        }
 524    }
 525
 526    fn image_md(_image: &acp::ImageContent) -> String {
 527        "`Image`".into()
 528    }
 529
 530    pub fn to_markdown<'a>(&'a self, cx: &'a App) -> &'a str {
 531        match self {
 532            ContentBlock::Empty => "",
 533            ContentBlock::Markdown { markdown } => markdown.read(cx).source(),
 534            ContentBlock::ResourceLink { resource_link } => &resource_link.uri,
 535        }
 536    }
 537
 538    pub fn markdown(&self) -> Option<&Entity<Markdown>> {
 539        match self {
 540            ContentBlock::Empty => None,
 541            ContentBlock::Markdown { markdown } => Some(markdown),
 542            ContentBlock::ResourceLink { .. } => None,
 543        }
 544    }
 545
 546    pub fn resource_link(&self) -> Option<&acp::ResourceLink> {
 547        match self {
 548            ContentBlock::ResourceLink { resource_link } => Some(resource_link),
 549            _ => None,
 550        }
 551    }
 552}
 553
 554#[derive(Debug)]
 555pub enum ToolCallContent {
 556    ContentBlock(ContentBlock),
 557    Diff(Entity<Diff>),
 558    Terminal(Entity<Terminal>),
 559}
 560
 561impl ToolCallContent {
 562    pub fn from_acp(
 563        content: acp::ToolCallContent,
 564        language_registry: Arc<LanguageRegistry>,
 565        terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
 566        cx: &mut App,
 567    ) -> Result<Self> {
 568        match content {
 569            acp::ToolCallContent::Content { content } => Ok(Self::ContentBlock(ContentBlock::new(
 570                content,
 571                &language_registry,
 572                cx,
 573            ))),
 574            acp::ToolCallContent::Diff { diff } => Ok(Self::Diff(cx.new(|cx| {
 575                Diff::finalized(
 576                    diff.path,
 577                    diff.old_text,
 578                    diff.new_text,
 579                    language_registry,
 580                    cx,
 581                )
 582            }))),
 583            acp::ToolCallContent::Terminal { terminal_id } => terminals
 584                .get(&terminal_id)
 585                .cloned()
 586                .map(Self::Terminal)
 587                .ok_or_else(|| anyhow::anyhow!("Terminal with id `{}` not found", terminal_id)),
 588        }
 589    }
 590
 591    pub fn update_from_acp(
 592        &mut self,
 593        new: acp::ToolCallContent,
 594        language_registry: Arc<LanguageRegistry>,
 595        terminals: &HashMap<acp::TerminalId, Entity<Terminal>>,
 596        cx: &mut App,
 597    ) -> Result<()> {
 598        let needs_update = match (&self, &new) {
 599            (Self::Diff(old_diff), acp::ToolCallContent::Diff { diff: new_diff }) => {
 600                old_diff.read(cx).needs_update(
 601                    new_diff.old_text.as_deref().unwrap_or(""),
 602                    &new_diff.new_text,
 603                    cx,
 604                )
 605            }
 606            _ => true,
 607        };
 608
 609        if needs_update {
 610            *self = Self::from_acp(new, language_registry, terminals, cx)?;
 611        }
 612        Ok(())
 613    }
 614
 615    pub fn to_markdown(&self, cx: &App) -> String {
 616        match self {
 617            Self::ContentBlock(content) => content.to_markdown(cx).to_string(),
 618            Self::Diff(diff) => diff.read(cx).to_markdown(cx),
 619            Self::Terminal(terminal) => terminal.read(cx).to_markdown(cx),
 620        }
 621    }
 622}
 623
 624#[derive(Debug, PartialEq)]
 625pub enum ToolCallUpdate {
 626    UpdateFields(acp::ToolCallUpdate),
 627    UpdateDiff(ToolCallUpdateDiff),
 628    UpdateTerminal(ToolCallUpdateTerminal),
 629}
 630
 631impl ToolCallUpdate {
 632    fn id(&self) -> &acp::ToolCallId {
 633        match self {
 634            Self::UpdateFields(update) => &update.id,
 635            Self::UpdateDiff(diff) => &diff.id,
 636            Self::UpdateTerminal(terminal) => &terminal.id,
 637        }
 638    }
 639}
 640
 641impl From<acp::ToolCallUpdate> for ToolCallUpdate {
 642    fn from(update: acp::ToolCallUpdate) -> Self {
 643        Self::UpdateFields(update)
 644    }
 645}
 646
 647impl From<ToolCallUpdateDiff> for ToolCallUpdate {
 648    fn from(diff: ToolCallUpdateDiff) -> Self {
 649        Self::UpdateDiff(diff)
 650    }
 651}
 652
 653#[derive(Debug, PartialEq)]
 654pub struct ToolCallUpdateDiff {
 655    pub id: acp::ToolCallId,
 656    pub diff: Entity<Diff>,
 657}
 658
 659impl From<ToolCallUpdateTerminal> for ToolCallUpdate {
 660    fn from(terminal: ToolCallUpdateTerminal) -> Self {
 661        Self::UpdateTerminal(terminal)
 662    }
 663}
 664
 665#[derive(Debug, PartialEq)]
 666pub struct ToolCallUpdateTerminal {
 667    pub id: acp::ToolCallId,
 668    pub terminal: Entity<Terminal>,
 669}
 670
 671#[derive(Debug, Default)]
 672pub struct Plan {
 673    pub entries: Vec<PlanEntry>,
 674}
 675
 676#[derive(Debug)]
 677pub struct PlanStats<'a> {
 678    pub in_progress_entry: Option<&'a PlanEntry>,
 679    pub pending: u32,
 680    pub completed: u32,
 681}
 682
 683impl Plan {
 684    pub fn is_empty(&self) -> bool {
 685        self.entries.is_empty()
 686    }
 687
 688    pub fn stats(&self) -> PlanStats<'_> {
 689        let mut stats = PlanStats {
 690            in_progress_entry: None,
 691            pending: 0,
 692            completed: 0,
 693        };
 694
 695        for entry in &self.entries {
 696            match &entry.status {
 697                acp::PlanEntryStatus::Pending => {
 698                    stats.pending += 1;
 699                }
 700                acp::PlanEntryStatus::InProgress => {
 701                    stats.in_progress_entry = stats.in_progress_entry.or(Some(entry));
 702                }
 703                acp::PlanEntryStatus::Completed => {
 704                    stats.completed += 1;
 705                }
 706            }
 707        }
 708
 709        stats
 710    }
 711}
 712
 713#[derive(Debug)]
 714pub struct PlanEntry {
 715    pub content: Entity<Markdown>,
 716    pub priority: acp::PlanEntryPriority,
 717    pub status: acp::PlanEntryStatus,
 718}
 719
 720impl PlanEntry {
 721    pub fn from_acp(entry: acp::PlanEntry, cx: &mut App) -> Self {
 722        Self {
 723            content: cx.new(|cx| Markdown::new(entry.content.into(), None, None, cx)),
 724            priority: entry.priority,
 725            status: entry.status,
 726        }
 727    }
 728}
 729
 730#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 731pub struct TokenUsage {
 732    pub max_tokens: u64,
 733    pub used_tokens: u64,
 734}
 735
 736impl TokenUsage {
 737    pub fn ratio(&self) -> TokenUsageRatio {
 738        #[cfg(debug_assertions)]
 739        let warning_threshold: f32 = std::env::var("ZED_THREAD_WARNING_THRESHOLD")
 740            .unwrap_or("0.8".to_string())
 741            .parse()
 742            .unwrap();
 743        #[cfg(not(debug_assertions))]
 744        let warning_threshold: f32 = 0.8;
 745
 746        // When the maximum is unknown because there is no selected model,
 747        // avoid showing the token limit warning.
 748        if self.max_tokens == 0 {
 749            TokenUsageRatio::Normal
 750        } else if self.used_tokens >= self.max_tokens {
 751            TokenUsageRatio::Exceeded
 752        } else if self.used_tokens as f32 / self.max_tokens as f32 >= warning_threshold {
 753            TokenUsageRatio::Warning
 754        } else {
 755            TokenUsageRatio::Normal
 756        }
 757    }
 758}
 759
 760#[derive(Debug, Clone, PartialEq, Eq)]
 761pub enum TokenUsageRatio {
 762    Normal,
 763    Warning,
 764    Exceeded,
 765}
 766
 767#[derive(Debug, Clone)]
 768pub struct RetryStatus {
 769    pub last_error: SharedString,
 770    pub attempt: usize,
 771    pub max_attempts: usize,
 772    pub started_at: Instant,
 773    pub duration: Duration,
 774}
 775
 776pub struct AcpThread {
 777    title: SharedString,
 778    entries: Vec<AgentThreadEntry>,
 779    plan: Plan,
 780    project: Entity<Project>,
 781    action_log: Entity<ActionLog>,
 782    shared_buffers: HashMap<Entity<Buffer>, BufferSnapshot>,
 783    send_task: Option<Task<()>>,
 784    connection: Rc<dyn AgentConnection>,
 785    session_id: acp::SessionId,
 786    token_usage: Option<TokenUsage>,
 787    prompt_capabilities: acp::PromptCapabilities,
 788    _observe_prompt_capabilities: Task<anyhow::Result<()>>,
 789    terminals: HashMap<acp::TerminalId, Entity<Terminal>>,
 790}
 791
 792#[derive(Debug)]
 793pub enum AcpThreadEvent {
 794    NewEntry,
 795    TitleUpdated,
 796    TokenUsageUpdated,
 797    EntryUpdated(usize),
 798    EntriesRemoved(Range<usize>),
 799    ToolAuthorizationRequired,
 800    Retry(RetryStatus),
 801    Stopped,
 802    Error,
 803    LoadError(LoadError),
 804    PromptCapabilitiesUpdated,
 805    Refusal,
 806    AvailableCommandsUpdated(Vec<acp::AvailableCommand>),
 807    ModeUpdated(acp::SessionModeId),
 808}
 809
 810impl EventEmitter<AcpThreadEvent> for AcpThread {}
 811
 812#[derive(PartialEq, Eq, Debug)]
 813pub enum ThreadStatus {
 814    Idle,
 815    Generating,
 816}
 817
 818#[derive(Debug, Clone)]
 819pub enum LoadError {
 820    Unsupported {
 821        command: SharedString,
 822        current_version: SharedString,
 823        minimum_version: SharedString,
 824    },
 825    FailedToInstall(SharedString),
 826    Exited {
 827        status: ExitStatus,
 828    },
 829    Other(SharedString),
 830}
 831
 832impl Display for LoadError {
 833    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
 834        match self {
 835            LoadError::Unsupported {
 836                command: path,
 837                current_version,
 838                minimum_version,
 839            } => {
 840                write!(
 841                    f,
 842                    "version {current_version} from {path} is not supported (need at least {minimum_version})"
 843                )
 844            }
 845            LoadError::FailedToInstall(msg) => write!(f, "Failed to install: {msg}"),
 846            LoadError::Exited { status } => write!(f, "Server exited with status {status}"),
 847            LoadError::Other(msg) => write!(f, "{msg}"),
 848        }
 849    }
 850}
 851
 852impl Error for LoadError {}
 853
 854impl AcpThread {
 855    pub fn new(
 856        title: impl Into<SharedString>,
 857        connection: Rc<dyn AgentConnection>,
 858        project: Entity<Project>,
 859        action_log: Entity<ActionLog>,
 860        session_id: acp::SessionId,
 861        mut prompt_capabilities_rx: watch::Receiver<acp::PromptCapabilities>,
 862        cx: &mut Context<Self>,
 863    ) -> Self {
 864        let prompt_capabilities = prompt_capabilities_rx.borrow().clone();
 865        let task = cx.spawn::<_, anyhow::Result<()>>(async move |this, cx| {
 866            loop {
 867                let caps = prompt_capabilities_rx.recv().await?;
 868                this.update(cx, |this, cx| {
 869                    this.prompt_capabilities = caps;
 870                    cx.emit(AcpThreadEvent::PromptCapabilitiesUpdated);
 871                })?;
 872            }
 873        });
 874
 875        Self {
 876            action_log,
 877            shared_buffers: Default::default(),
 878            entries: Default::default(),
 879            plan: Default::default(),
 880            title: title.into(),
 881            project,
 882            send_task: None,
 883            connection,
 884            session_id,
 885            token_usage: None,
 886            prompt_capabilities,
 887            _observe_prompt_capabilities: task,
 888            terminals: HashMap::default(),
 889        }
 890    }
 891
 892    pub fn prompt_capabilities(&self) -> acp::PromptCapabilities {
 893        self.prompt_capabilities.clone()
 894    }
 895
 896    pub fn connection(&self) -> &Rc<dyn AgentConnection> {
 897        &self.connection
 898    }
 899
 900    pub fn action_log(&self) -> &Entity<ActionLog> {
 901        &self.action_log
 902    }
 903
 904    pub fn project(&self) -> &Entity<Project> {
 905        &self.project
 906    }
 907
 908    pub fn title(&self) -> SharedString {
 909        self.title.clone()
 910    }
 911
 912    pub fn entries(&self) -> &[AgentThreadEntry] {
 913        &self.entries
 914    }
 915
 916    pub fn session_id(&self) -> &acp::SessionId {
 917        &self.session_id
 918    }
 919
 920    pub fn status(&self) -> ThreadStatus {
 921        if self.send_task.is_some() {
 922            ThreadStatus::Generating
 923        } else {
 924            ThreadStatus::Idle
 925        }
 926    }
 927
 928    pub fn token_usage(&self) -> Option<&TokenUsage> {
 929        self.token_usage.as_ref()
 930    }
 931
 932    pub fn has_pending_edit_tool_calls(&self) -> bool {
 933        for entry in self.entries.iter().rev() {
 934            match entry {
 935                AgentThreadEntry::UserMessage(_) => return false,
 936                AgentThreadEntry::ToolCall(
 937                    call @ ToolCall {
 938                        status: ToolCallStatus::InProgress | ToolCallStatus::Pending,
 939                        ..
 940                    },
 941                ) if call.diffs().next().is_some() => {
 942                    return true;
 943                }
 944                AgentThreadEntry::ToolCall(_) | AgentThreadEntry::AssistantMessage(_) => {}
 945            }
 946        }
 947
 948        false
 949    }
 950
 951    pub fn used_tools_since_last_user_message(&self) -> bool {
 952        for entry in self.entries.iter().rev() {
 953            match entry {
 954                AgentThreadEntry::UserMessage(..) => return false,
 955                AgentThreadEntry::AssistantMessage(..) => continue,
 956                AgentThreadEntry::ToolCall(..) => return true,
 957            }
 958        }
 959
 960        false
 961    }
 962
 963    pub fn handle_session_update(
 964        &mut self,
 965        update: acp::SessionUpdate,
 966        cx: &mut Context<Self>,
 967    ) -> Result<(), acp::Error> {
 968        match update {
 969            acp::SessionUpdate::UserMessageChunk { content } => {
 970                self.push_user_content_block(None, content, cx);
 971            }
 972            acp::SessionUpdate::AgentMessageChunk { content } => {
 973                self.push_assistant_content_block(content, false, cx);
 974            }
 975            acp::SessionUpdate::AgentThoughtChunk { content } => {
 976                self.push_assistant_content_block(content, true, cx);
 977            }
 978            acp::SessionUpdate::ToolCall(tool_call) => {
 979                self.upsert_tool_call(tool_call, cx)?;
 980            }
 981            acp::SessionUpdate::ToolCallUpdate(tool_call_update) => {
 982                self.update_tool_call(tool_call_update, cx)?;
 983            }
 984            acp::SessionUpdate::Plan(plan) => {
 985                self.update_plan(plan, cx);
 986            }
 987            acp::SessionUpdate::AvailableCommandsUpdate { available_commands } => {
 988                cx.emit(AcpThreadEvent::AvailableCommandsUpdated(available_commands))
 989            }
 990            acp::SessionUpdate::CurrentModeUpdate { current_mode_id } => {
 991                cx.emit(AcpThreadEvent::ModeUpdated(current_mode_id))
 992            }
 993        }
 994        Ok(())
 995    }
 996
 997    pub fn push_user_content_block(
 998        &mut self,
 999        message_id: Option<UserMessageId>,
1000        chunk: acp::ContentBlock,
1001        cx: &mut Context<Self>,
1002    ) {
1003        let language_registry = self.project.read(cx).languages().clone();
1004        let entries_len = self.entries.len();
1005
1006        if let Some(last_entry) = self.entries.last_mut()
1007            && let AgentThreadEntry::UserMessage(UserMessage {
1008                id,
1009                content,
1010                chunks,
1011                ..
1012            }) = last_entry
1013        {
1014            *id = message_id.or(id.take());
1015            content.append(chunk.clone(), &language_registry, cx);
1016            chunks.push(chunk);
1017            let idx = entries_len - 1;
1018            cx.emit(AcpThreadEvent::EntryUpdated(idx));
1019        } else {
1020            let content = ContentBlock::new(chunk.clone(), &language_registry, cx);
1021            self.push_entry(
1022                AgentThreadEntry::UserMessage(UserMessage {
1023                    id: message_id,
1024                    content,
1025                    chunks: vec![chunk],
1026                    checkpoint: None,
1027                }),
1028                cx,
1029            );
1030        }
1031    }
1032
1033    pub fn push_assistant_content_block(
1034        &mut self,
1035        chunk: acp::ContentBlock,
1036        is_thought: bool,
1037        cx: &mut Context<Self>,
1038    ) {
1039        let language_registry = self.project.read(cx).languages().clone();
1040        let entries_len = self.entries.len();
1041        if let Some(last_entry) = self.entries.last_mut()
1042            && let AgentThreadEntry::AssistantMessage(AssistantMessage { chunks }) = last_entry
1043        {
1044            let idx = entries_len - 1;
1045            cx.emit(AcpThreadEvent::EntryUpdated(idx));
1046            match (chunks.last_mut(), is_thought) {
1047                (Some(AssistantMessageChunk::Message { block }), false)
1048                | (Some(AssistantMessageChunk::Thought { block }), true) => {
1049                    block.append(chunk, &language_registry, cx)
1050                }
1051                _ => {
1052                    let block = ContentBlock::new(chunk, &language_registry, cx);
1053                    if is_thought {
1054                        chunks.push(AssistantMessageChunk::Thought { block })
1055                    } else {
1056                        chunks.push(AssistantMessageChunk::Message { block })
1057                    }
1058                }
1059            }
1060        } else {
1061            let block = ContentBlock::new(chunk, &language_registry, cx);
1062            let chunk = if is_thought {
1063                AssistantMessageChunk::Thought { block }
1064            } else {
1065                AssistantMessageChunk::Message { block }
1066            };
1067
1068            self.push_entry(
1069                AgentThreadEntry::AssistantMessage(AssistantMessage {
1070                    chunks: vec![chunk],
1071                }),
1072                cx,
1073            );
1074        }
1075    }
1076
1077    fn push_entry(&mut self, entry: AgentThreadEntry, cx: &mut Context<Self>) {
1078        self.entries.push(entry);
1079        cx.emit(AcpThreadEvent::NewEntry);
1080    }
1081
1082    pub fn can_set_title(&mut self, cx: &mut Context<Self>) -> bool {
1083        self.connection.set_title(&self.session_id, cx).is_some()
1084    }
1085
1086    pub fn set_title(&mut self, title: SharedString, cx: &mut Context<Self>) -> Task<Result<()>> {
1087        if title != self.title {
1088            self.title = title.clone();
1089            cx.emit(AcpThreadEvent::TitleUpdated);
1090            if let Some(set_title) = self.connection.set_title(&self.session_id, cx) {
1091                return set_title.run(title, cx);
1092            }
1093        }
1094        Task::ready(Ok(()))
1095    }
1096
1097    pub fn update_token_usage(&mut self, usage: Option<TokenUsage>, cx: &mut Context<Self>) {
1098        self.token_usage = usage;
1099        cx.emit(AcpThreadEvent::TokenUsageUpdated);
1100    }
1101
1102    pub fn update_retry_status(&mut self, status: RetryStatus, cx: &mut Context<Self>) {
1103        cx.emit(AcpThreadEvent::Retry(status));
1104    }
1105
1106    pub fn update_tool_call(
1107        &mut self,
1108        update: impl Into<ToolCallUpdate>,
1109        cx: &mut Context<Self>,
1110    ) -> Result<()> {
1111        let update = update.into();
1112        let languages = self.project.read(cx).languages().clone();
1113
1114        let ix = self
1115            .index_for_tool_call(update.id())
1116            .context("Tool call not found")?;
1117        let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1118            unreachable!()
1119        };
1120
1121        match update {
1122            ToolCallUpdate::UpdateFields(update) => {
1123                let location_updated = update.fields.locations.is_some();
1124                call.update_fields(update.fields, languages, &self.terminals, cx)?;
1125                if location_updated {
1126                    self.resolve_locations(update.id, cx);
1127                }
1128            }
1129            ToolCallUpdate::UpdateDiff(update) => {
1130                call.content.clear();
1131                call.content.push(ToolCallContent::Diff(update.diff));
1132            }
1133            ToolCallUpdate::UpdateTerminal(update) => {
1134                call.content.clear();
1135                call.content
1136                    .push(ToolCallContent::Terminal(update.terminal));
1137            }
1138        }
1139
1140        cx.emit(AcpThreadEvent::EntryUpdated(ix));
1141
1142        Ok(())
1143    }
1144
1145    /// Updates a tool call if id matches an existing entry, otherwise inserts a new one.
1146    pub fn upsert_tool_call(
1147        &mut self,
1148        tool_call: acp::ToolCall,
1149        cx: &mut Context<Self>,
1150    ) -> Result<(), acp::Error> {
1151        let status = tool_call.status.into();
1152        self.upsert_tool_call_inner(tool_call.into(), status, cx)
1153    }
1154
1155    /// Fails if id does not match an existing entry.
1156    pub fn upsert_tool_call_inner(
1157        &mut self,
1158        update: acp::ToolCallUpdate,
1159        status: ToolCallStatus,
1160        cx: &mut Context<Self>,
1161    ) -> Result<(), acp::Error> {
1162        let language_registry = self.project.read(cx).languages().clone();
1163        let id = update.id.clone();
1164
1165        if let Some(ix) = self.index_for_tool_call(&id) {
1166            let AgentThreadEntry::ToolCall(call) = &mut self.entries[ix] else {
1167                unreachable!()
1168            };
1169
1170            call.update_fields(update.fields, language_registry, &self.terminals, cx)?;
1171            call.status = status;
1172
1173            cx.emit(AcpThreadEvent::EntryUpdated(ix));
1174        } else {
1175            let call = ToolCall::from_acp(
1176                update.try_into()?,
1177                status,
1178                language_registry,
1179                &self.terminals,
1180                cx,
1181            )?;
1182            self.push_entry(AgentThreadEntry::ToolCall(call), cx);
1183        };
1184
1185        self.resolve_locations(id, cx);
1186        Ok(())
1187    }
1188
1189    fn index_for_tool_call(&self, id: &acp::ToolCallId) -> Option<usize> {
1190        self.entries
1191            .iter()
1192            .enumerate()
1193            .rev()
1194            .find_map(|(index, entry)| {
1195                if let AgentThreadEntry::ToolCall(tool_call) = entry
1196                    && &tool_call.id == id
1197                {
1198                    Some(index)
1199                } else {
1200                    None
1201                }
1202            })
1203    }
1204
1205    fn tool_call_mut(&mut self, id: &acp::ToolCallId) -> Option<(usize, &mut ToolCall)> {
1206        // The tool call we are looking for is typically the last one, or very close to the end.
1207        // At the moment, it doesn't seem like a hashmap would be a good fit for this use case.
1208        self.entries
1209            .iter_mut()
1210            .enumerate()
1211            .rev()
1212            .find_map(|(index, tool_call)| {
1213                if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1214                    && &tool_call.id == id
1215                {
1216                    Some((index, tool_call))
1217                } else {
1218                    None
1219                }
1220            })
1221    }
1222
1223    pub fn tool_call(&mut self, id: &acp::ToolCallId) -> Option<(usize, &ToolCall)> {
1224        self.entries
1225            .iter()
1226            .enumerate()
1227            .rev()
1228            .find_map(|(index, tool_call)| {
1229                if let AgentThreadEntry::ToolCall(tool_call) = tool_call
1230                    && &tool_call.id == id
1231                {
1232                    Some((index, tool_call))
1233                } else {
1234                    None
1235                }
1236            })
1237    }
1238
1239    pub fn resolve_locations(&mut self, id: acp::ToolCallId, cx: &mut Context<Self>) {
1240        let project = self.project.clone();
1241        let Some((_, tool_call)) = self.tool_call_mut(&id) else {
1242            return;
1243        };
1244        let task = tool_call.resolve_locations(project, cx);
1245        cx.spawn(async move |this, cx| {
1246            let resolved_locations = task.await;
1247            this.update(cx, |this, cx| {
1248                let project = this.project.clone();
1249                let Some((ix, tool_call)) = this.tool_call_mut(&id) else {
1250                    return;
1251                };
1252                if let Some(Some(location)) = resolved_locations.last() {
1253                    project.update(cx, |project, cx| {
1254                        if let Some(agent_location) = project.agent_location() {
1255                            let should_ignore = agent_location.buffer == location.buffer
1256                                && location
1257                                    .buffer
1258                                    .update(cx, |buffer, _| {
1259                                        let snapshot = buffer.snapshot();
1260                                        let old_position =
1261                                            agent_location.position.to_point(&snapshot);
1262                                        let new_position = location.position.to_point(&snapshot);
1263                                        // ignore this so that when we get updates from the edit tool
1264                                        // the position doesn't reset to the startof line
1265                                        old_position.row == new_position.row
1266                                            && old_position.column > new_position.column
1267                                    })
1268                                    .ok()
1269                                    .unwrap_or_default();
1270                            if !should_ignore {
1271                                project.set_agent_location(Some(location.clone()), cx);
1272                            }
1273                        }
1274                    });
1275                }
1276                if tool_call.resolved_locations != resolved_locations {
1277                    tool_call.resolved_locations = resolved_locations;
1278                    cx.emit(AcpThreadEvent::EntryUpdated(ix));
1279                }
1280            })
1281        })
1282        .detach();
1283    }
1284
1285    pub fn request_tool_call_authorization(
1286        &mut self,
1287        tool_call: acp::ToolCallUpdate,
1288        options: Vec<acp::PermissionOption>,
1289        respect_always_allow_setting: bool,
1290        cx: &mut Context<Self>,
1291    ) -> Result<BoxFuture<'static, acp::RequestPermissionOutcome>> {
1292        let (tx, rx) = oneshot::channel();
1293
1294        if respect_always_allow_setting && AgentSettings::get_global(cx).always_allow_tool_actions {
1295            // Don't use AllowAlways, because then if you were to turn off always_allow_tool_actions,
1296            // some tools would (incorrectly) continue to auto-accept.
1297            if let Some(allow_once_option) = options.iter().find_map(|option| {
1298                if matches!(option.kind, acp::PermissionOptionKind::AllowOnce) {
1299                    Some(option.id.clone())
1300                } else {
1301                    None
1302                }
1303            }) {
1304                self.upsert_tool_call_inner(tool_call, ToolCallStatus::Pending, cx)?;
1305                return Ok(async {
1306                    acp::RequestPermissionOutcome::Selected {
1307                        option_id: allow_once_option,
1308                    }
1309                }
1310                .boxed());
1311            }
1312        }
1313
1314        let status = ToolCallStatus::WaitingForConfirmation {
1315            options,
1316            respond_tx: tx,
1317        };
1318
1319        self.upsert_tool_call_inner(tool_call, status, cx)?;
1320        cx.emit(AcpThreadEvent::ToolAuthorizationRequired);
1321
1322        let fut = async {
1323            match rx.await {
1324                Ok(option) => acp::RequestPermissionOutcome::Selected { option_id: option },
1325                Err(oneshot::Canceled) => acp::RequestPermissionOutcome::Cancelled,
1326            }
1327        }
1328        .boxed();
1329
1330        Ok(fut)
1331    }
1332
1333    pub fn authorize_tool_call(
1334        &mut self,
1335        id: acp::ToolCallId,
1336        option_id: acp::PermissionOptionId,
1337        option_kind: acp::PermissionOptionKind,
1338        cx: &mut Context<Self>,
1339    ) {
1340        let Some((ix, call)) = self.tool_call_mut(&id) else {
1341            return;
1342        };
1343
1344        let new_status = match option_kind {
1345            acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways => {
1346                ToolCallStatus::Rejected
1347            }
1348            acp::PermissionOptionKind::AllowOnce | acp::PermissionOptionKind::AllowAlways => {
1349                ToolCallStatus::InProgress
1350            }
1351        };
1352
1353        let curr_status = mem::replace(&mut call.status, new_status);
1354
1355        if let ToolCallStatus::WaitingForConfirmation { respond_tx, .. } = curr_status {
1356            respond_tx.send(option_id).log_err();
1357        } else if cfg!(debug_assertions) {
1358            panic!("tried to authorize an already authorized tool call");
1359        }
1360
1361        cx.emit(AcpThreadEvent::EntryUpdated(ix));
1362    }
1363
1364    pub fn first_tool_awaiting_confirmation(&self) -> Option<&ToolCall> {
1365        let mut first_tool_call = None;
1366
1367        for entry in self.entries.iter().rev() {
1368            match &entry {
1369                AgentThreadEntry::ToolCall(call) => {
1370                    if let ToolCallStatus::WaitingForConfirmation { .. } = call.status {
1371                        first_tool_call = Some(call);
1372                    } else {
1373                        continue;
1374                    }
1375                }
1376                AgentThreadEntry::UserMessage(_) | AgentThreadEntry::AssistantMessage(_) => {
1377                    // Reached the beginning of the turn.
1378                    // If we had pending permission requests in the previous turn, they have been cancelled.
1379                    break;
1380                }
1381            }
1382        }
1383
1384        first_tool_call
1385    }
1386
1387    pub fn plan(&self) -> &Plan {
1388        &self.plan
1389    }
1390
1391    pub fn update_plan(&mut self, request: acp::Plan, cx: &mut Context<Self>) {
1392        let new_entries_len = request.entries.len();
1393        let mut new_entries = request.entries.into_iter();
1394
1395        // Reuse existing markdown to prevent flickering
1396        for (old, new) in self.plan.entries.iter_mut().zip(new_entries.by_ref()) {
1397            let PlanEntry {
1398                content,
1399                priority,
1400                status,
1401            } = old;
1402            content.update(cx, |old, cx| {
1403                old.replace(new.content, cx);
1404            });
1405            *priority = new.priority;
1406            *status = new.status;
1407        }
1408        for new in new_entries {
1409            self.plan.entries.push(PlanEntry::from_acp(new, cx))
1410        }
1411        self.plan.entries.truncate(new_entries_len);
1412
1413        cx.notify();
1414    }
1415
1416    fn clear_completed_plan_entries(&mut self, cx: &mut Context<Self>) {
1417        self.plan
1418            .entries
1419            .retain(|entry| !matches!(entry.status, acp::PlanEntryStatus::Completed));
1420        cx.notify();
1421    }
1422
1423    #[cfg(any(test, feature = "test-support"))]
1424    pub fn send_raw(
1425        &mut self,
1426        message: &str,
1427        cx: &mut Context<Self>,
1428    ) -> BoxFuture<'static, Result<()>> {
1429        self.send(
1430            vec![acp::ContentBlock::Text(acp::TextContent {
1431                text: message.to_string(),
1432                annotations: None,
1433                meta: None,
1434            })],
1435            cx,
1436        )
1437    }
1438
1439    pub fn send(
1440        &mut self,
1441        message: Vec<acp::ContentBlock>,
1442        cx: &mut Context<Self>,
1443    ) -> BoxFuture<'static, Result<()>> {
1444        let block = ContentBlock::new_combined(
1445            message.clone(),
1446            self.project.read(cx).languages().clone(),
1447            cx,
1448        );
1449        let request = acp::PromptRequest {
1450            prompt: message.clone(),
1451            session_id: self.session_id.clone(),
1452            meta: None,
1453        };
1454        let git_store = self.project.read(cx).git_store().clone();
1455
1456        let message_id = if self.connection.truncate(&self.session_id, cx).is_some() {
1457            Some(UserMessageId::new())
1458        } else {
1459            None
1460        };
1461
1462        self.run_turn(cx, async move |this, cx| {
1463            this.update(cx, |this, cx| {
1464                this.push_entry(
1465                    AgentThreadEntry::UserMessage(UserMessage {
1466                        id: message_id.clone(),
1467                        content: block,
1468                        chunks: message,
1469                        checkpoint: None,
1470                    }),
1471                    cx,
1472                );
1473            })
1474            .ok();
1475
1476            let old_checkpoint = git_store
1477                .update(cx, |git, cx| git.checkpoint(cx))?
1478                .await
1479                .context("failed to get old checkpoint")
1480                .log_err();
1481            this.update(cx, |this, cx| {
1482                if let Some((_ix, message)) = this.last_user_message() {
1483                    message.checkpoint = old_checkpoint.map(|git_checkpoint| Checkpoint {
1484                        git_checkpoint,
1485                        show: false,
1486                    });
1487                }
1488                this.connection.prompt(message_id, request, cx)
1489            })?
1490            .await
1491        })
1492    }
1493
1494    pub fn can_resume(&self, cx: &App) -> bool {
1495        self.connection.resume(&self.session_id, cx).is_some()
1496    }
1497
1498    pub fn resume(&mut self, cx: &mut Context<Self>) -> BoxFuture<'static, Result<()>> {
1499        self.run_turn(cx, async move |this, cx| {
1500            this.update(cx, |this, cx| {
1501                this.connection
1502                    .resume(&this.session_id, cx)
1503                    .map(|resume| resume.run(cx))
1504            })?
1505            .context("resuming a session is not supported")?
1506            .await
1507        })
1508    }
1509
1510    fn run_turn(
1511        &mut self,
1512        cx: &mut Context<Self>,
1513        f: impl 'static + AsyncFnOnce(WeakEntity<Self>, &mut AsyncApp) -> Result<acp::PromptResponse>,
1514    ) -> BoxFuture<'static, Result<()>> {
1515        self.clear_completed_plan_entries(cx);
1516
1517        let (tx, rx) = oneshot::channel();
1518        let cancel_task = self.cancel(cx);
1519
1520        self.send_task = Some(cx.spawn(async move |this, cx| {
1521            cancel_task.await;
1522            tx.send(f(this, cx).await).ok();
1523        }));
1524
1525        cx.spawn(async move |this, cx| {
1526            let response = rx.await;
1527
1528            this.update(cx, |this, cx| this.update_last_checkpoint(cx))?
1529                .await?;
1530
1531            this.update(cx, |this, cx| {
1532                this.project
1533                    .update(cx, |project, cx| project.set_agent_location(None, cx));
1534                match response {
1535                    Ok(Err(e)) => {
1536                        this.send_task.take();
1537                        cx.emit(AcpThreadEvent::Error);
1538                        Err(e)
1539                    }
1540                    result => {
1541                        let canceled = matches!(
1542                            result,
1543                            Ok(Ok(acp::PromptResponse {
1544                                stop_reason: acp::StopReason::Cancelled,
1545                                meta: None,
1546                            }))
1547                        );
1548
1549                        // We only take the task if the current prompt wasn't canceled.
1550                        //
1551                        // This prompt may have been canceled because another one was sent
1552                        // while it was still generating. In these cases, dropping `send_task`
1553                        // would cause the next generation to be canceled.
1554                        if !canceled {
1555                            this.send_task.take();
1556                        }
1557
1558                        // Handle refusal - distinguish between user prompt and tool call refusals
1559                        if let Ok(Ok(acp::PromptResponse {
1560                            stop_reason: acp::StopReason::Refusal,
1561                            meta: _,
1562                        })) = result
1563                        {
1564                            if let Some((user_msg_ix, _)) = this.last_user_message() {
1565                                // Check if there's a completed tool call with results after the last user message
1566                                // This indicates the refusal is in response to tool output, not the user's prompt
1567                                let has_completed_tool_call_after_user_msg =
1568                                    this.entries.iter().skip(user_msg_ix + 1).any(|entry| {
1569                                        if let AgentThreadEntry::ToolCall(tool_call) = entry {
1570                                            // Check if the tool call has completed and has output
1571                                            matches!(tool_call.status, ToolCallStatus::Completed)
1572                                                && tool_call.raw_output.is_some()
1573                                        } else {
1574                                            false
1575                                        }
1576                                    });
1577
1578                                if has_completed_tool_call_after_user_msg {
1579                                    // Refusal is due to tool output - don't truncate, just notify
1580                                    // The model refused based on what the tool returned
1581                                    cx.emit(AcpThreadEvent::Refusal);
1582                                } else {
1583                                    // User prompt was refused - truncate back to before the user message
1584                                    let range = user_msg_ix..this.entries.len();
1585                                    if range.start < range.end {
1586                                        this.entries.truncate(user_msg_ix);
1587                                        cx.emit(AcpThreadEvent::EntriesRemoved(range));
1588                                    }
1589                                    cx.emit(AcpThreadEvent::Refusal);
1590                                }
1591                            } else {
1592                                // No user message found, treat as general refusal
1593                                cx.emit(AcpThreadEvent::Refusal);
1594                            }
1595                        }
1596
1597                        cx.emit(AcpThreadEvent::Stopped);
1598                        Ok(())
1599                    }
1600                }
1601            })?
1602        })
1603        .boxed()
1604    }
1605
1606    pub fn cancel(&mut self, cx: &mut Context<Self>) -> Task<()> {
1607        let Some(send_task) = self.send_task.take() else {
1608            return Task::ready(());
1609        };
1610
1611        for entry in self.entries.iter_mut() {
1612            if let AgentThreadEntry::ToolCall(call) = entry {
1613                let cancel = matches!(
1614                    call.status,
1615                    ToolCallStatus::Pending
1616                        | ToolCallStatus::WaitingForConfirmation { .. }
1617                        | ToolCallStatus::InProgress
1618                );
1619
1620                if cancel {
1621                    call.status = ToolCallStatus::Canceled;
1622                }
1623            }
1624        }
1625
1626        self.connection.cancel(&self.session_id, cx);
1627
1628        // Wait for the send task to complete
1629        cx.foreground_executor().spawn(send_task)
1630    }
1631
1632    /// Restores the git working tree to the state at the given checkpoint (if one exists)
1633    pub fn restore_checkpoint(
1634        &mut self,
1635        id: UserMessageId,
1636        cx: &mut Context<Self>,
1637    ) -> Task<Result<()>> {
1638        let Some((_, message)) = self.user_message_mut(&id) else {
1639            return Task::ready(Err(anyhow!("message not found")));
1640        };
1641
1642        let checkpoint = message
1643            .checkpoint
1644            .as_ref()
1645            .map(|c| c.git_checkpoint.clone());
1646        let rewind = self.rewind(id.clone(), cx);
1647        let git_store = self.project.read(cx).git_store().clone();
1648
1649        cx.spawn(async move |_, cx| {
1650            rewind.await?;
1651            if let Some(checkpoint) = checkpoint {
1652                git_store
1653                    .update(cx, |git, cx| git.restore_checkpoint(checkpoint, cx))?
1654                    .await?;
1655            }
1656
1657            Ok(())
1658        })
1659    }
1660
1661    /// Rewinds this thread to before the entry at `index`, removing it and all
1662    /// subsequent entries while rejecting any action_log changes made from that point.
1663    /// Unlike `restore_checkpoint`, this method does not restore from git.
1664    pub fn rewind(&mut self, id: UserMessageId, cx: &mut Context<Self>) -> Task<Result<()>> {
1665        let Some(truncate) = self.connection.truncate(&self.session_id, cx) else {
1666            return Task::ready(Err(anyhow!("not supported")));
1667        };
1668
1669        cx.spawn(async move |this, cx| {
1670            cx.update(|cx| truncate.run(id.clone(), cx))?.await?;
1671            this.update(cx, |this, cx| {
1672                if let Some((ix, _)) = this.user_message_mut(&id) {
1673                    let range = ix..this.entries.len();
1674                    this.entries.truncate(ix);
1675                    cx.emit(AcpThreadEvent::EntriesRemoved(range));
1676                }
1677                this.action_log()
1678                    .update(cx, |action_log, cx| action_log.reject_all_edits(cx))
1679            })?
1680            .await;
1681            Ok(())
1682        })
1683    }
1684
1685    fn update_last_checkpoint(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1686        let git_store = self.project.read(cx).git_store().clone();
1687
1688        let old_checkpoint = if let Some((_, message)) = self.last_user_message() {
1689            if let Some(checkpoint) = message.checkpoint.as_ref() {
1690                checkpoint.git_checkpoint.clone()
1691            } else {
1692                return Task::ready(Ok(()));
1693            }
1694        } else {
1695            return Task::ready(Ok(()));
1696        };
1697
1698        let new_checkpoint = git_store.update(cx, |git, cx| git.checkpoint(cx));
1699        cx.spawn(async move |this, cx| {
1700            let new_checkpoint = new_checkpoint
1701                .await
1702                .context("failed to get new checkpoint")
1703                .log_err();
1704            if let Some(new_checkpoint) = new_checkpoint {
1705                let equal = git_store
1706                    .update(cx, |git, cx| {
1707                        git.compare_checkpoints(old_checkpoint.clone(), new_checkpoint, cx)
1708                    })?
1709                    .await
1710                    .unwrap_or(true);
1711                this.update(cx, |this, cx| {
1712                    let (ix, message) = this.last_user_message().context("no user message")?;
1713                    let checkpoint = message.checkpoint.as_mut().context("no checkpoint")?;
1714                    checkpoint.show = !equal;
1715                    cx.emit(AcpThreadEvent::EntryUpdated(ix));
1716                    anyhow::Ok(())
1717                })??;
1718            }
1719
1720            Ok(())
1721        })
1722    }
1723
1724    fn last_user_message(&mut self) -> Option<(usize, &mut UserMessage)> {
1725        self.entries
1726            .iter_mut()
1727            .enumerate()
1728            .rev()
1729            .find_map(|(ix, entry)| {
1730                if let AgentThreadEntry::UserMessage(message) = entry {
1731                    Some((ix, message))
1732                } else {
1733                    None
1734                }
1735            })
1736    }
1737
1738    fn user_message_mut(&mut self, id: &UserMessageId) -> Option<(usize, &mut UserMessage)> {
1739        self.entries.iter_mut().enumerate().find_map(|(ix, entry)| {
1740            if let AgentThreadEntry::UserMessage(message) = entry {
1741                if message.id.as_ref() == Some(id) {
1742                    Some((ix, message))
1743                } else {
1744                    None
1745                }
1746            } else {
1747                None
1748            }
1749        })
1750    }
1751
1752    pub fn read_text_file(
1753        &self,
1754        path: PathBuf,
1755        line: Option<u32>,
1756        limit: Option<u32>,
1757        reuse_shared_snapshot: bool,
1758        cx: &mut Context<Self>,
1759    ) -> Task<Result<String>> {
1760        let project = self.project.clone();
1761        let action_log = self.action_log.clone();
1762        cx.spawn(async move |this, cx| {
1763            let load = project.update(cx, |project, cx| {
1764                let path = project
1765                    .project_path_for_absolute_path(&path, cx)
1766                    .context("invalid path")?;
1767                anyhow::Ok(project.open_buffer(path, cx))
1768            });
1769            let buffer = load??.await?;
1770
1771            let snapshot = if reuse_shared_snapshot {
1772                this.read_with(cx, |this, _| {
1773                    this.shared_buffers.get(&buffer.clone()).cloned()
1774                })
1775                .log_err()
1776                .flatten()
1777            } else {
1778                None
1779            };
1780
1781            let snapshot = if let Some(snapshot) = snapshot {
1782                snapshot
1783            } else {
1784                action_log.update(cx, |action_log, cx| {
1785                    action_log.buffer_read(buffer.clone(), cx);
1786                })?;
1787                project.update(cx, |project, cx| {
1788                    let position = buffer
1789                        .read(cx)
1790                        .snapshot()
1791                        .anchor_before(Point::new(line.unwrap_or_default(), 0));
1792                    project.set_agent_location(
1793                        Some(AgentLocation {
1794                            buffer: buffer.downgrade(),
1795                            position,
1796                        }),
1797                        cx,
1798                    );
1799                })?;
1800
1801                buffer.update(cx, |buffer, _| buffer.snapshot())?
1802            };
1803
1804            this.update(cx, |this, _| {
1805                let text = snapshot.text();
1806                this.shared_buffers.insert(buffer.clone(), snapshot);
1807                if line.is_none() && limit.is_none() {
1808                    return Ok(text);
1809                }
1810                let limit = limit.unwrap_or(u32::MAX) as usize;
1811                let Some(line) = line else {
1812                    return Ok(text.lines().take(limit).collect::<String>());
1813                };
1814
1815                let count = text.lines().count();
1816                if count < line as usize {
1817                    anyhow::bail!("There are only {} lines", count);
1818                }
1819                Ok(text
1820                    .lines()
1821                    .skip(line as usize + 1)
1822                    .take(limit)
1823                    .collect::<String>())
1824            })?
1825        })
1826    }
1827
1828    pub fn write_text_file(
1829        &self,
1830        path: PathBuf,
1831        content: String,
1832        cx: &mut Context<Self>,
1833    ) -> Task<Result<()>> {
1834        let project = self.project.clone();
1835        let action_log = self.action_log.clone();
1836        cx.spawn(async move |this, cx| {
1837            let load = project.update(cx, |project, cx| {
1838                let path = project
1839                    .project_path_for_absolute_path(&path, cx)
1840                    .context("invalid path")?;
1841                anyhow::Ok(project.open_buffer(path, cx))
1842            });
1843            let buffer = load??.await?;
1844            let snapshot = this.update(cx, |this, cx| {
1845                this.shared_buffers
1846                    .get(&buffer)
1847                    .cloned()
1848                    .unwrap_or_else(|| buffer.read(cx).snapshot())
1849            })?;
1850            let edits = cx
1851                .background_executor()
1852                .spawn(async move {
1853                    let old_text = snapshot.text();
1854                    text_diff(old_text.as_str(), &content)
1855                        .into_iter()
1856                        .map(|(range, replacement)| {
1857                            (
1858                                snapshot.anchor_after(range.start)
1859                                    ..snapshot.anchor_before(range.end),
1860                                replacement,
1861                            )
1862                        })
1863                        .collect::<Vec<_>>()
1864                })
1865                .await;
1866
1867            project.update(cx, |project, cx| {
1868                project.set_agent_location(
1869                    Some(AgentLocation {
1870                        buffer: buffer.downgrade(),
1871                        position: edits
1872                            .last()
1873                            .map(|(range, _)| range.end)
1874                            .unwrap_or(Anchor::MIN),
1875                    }),
1876                    cx,
1877                );
1878            })?;
1879
1880            let format_on_save = cx.update(|cx| {
1881                action_log.update(cx, |action_log, cx| {
1882                    action_log.buffer_read(buffer.clone(), cx);
1883                });
1884
1885                let format_on_save = buffer.update(cx, |buffer, cx| {
1886                    buffer.edit(edits, None, cx);
1887
1888                    let settings = language::language_settings::language_settings(
1889                        buffer.language().map(|l| l.name()),
1890                        buffer.file(),
1891                        cx,
1892                    );
1893
1894                    settings.format_on_save != FormatOnSave::Off
1895                });
1896                action_log.update(cx, |action_log, cx| {
1897                    action_log.buffer_edited(buffer.clone(), cx);
1898                });
1899                format_on_save
1900            })?;
1901
1902            if format_on_save {
1903                let format_task = project.update(cx, |project, cx| {
1904                    project.format(
1905                        HashSet::from_iter([buffer.clone()]),
1906                        LspFormatTarget::Buffers,
1907                        false,
1908                        FormatTrigger::Save,
1909                        cx,
1910                    )
1911                })?;
1912                format_task.await.log_err();
1913
1914                action_log.update(cx, |action_log, cx| {
1915                    action_log.buffer_edited(buffer.clone(), cx);
1916                })?;
1917            }
1918
1919            project
1920                .update(cx, |project, cx| project.save_buffer(buffer, cx))?
1921                .await
1922        })
1923    }
1924
1925    pub fn create_terminal(
1926        &self,
1927        command: String,
1928        args: Vec<String>,
1929        extra_env: Vec<acp::EnvVariable>,
1930        cwd: Option<PathBuf>,
1931        output_byte_limit: Option<u64>,
1932        cx: &mut Context<Self>,
1933    ) -> Task<Result<Entity<Terminal>>> {
1934        let env = match &cwd {
1935            Some(dir) => self.project.update(cx, |project, cx| {
1936                project.directory_environment(dir.as_path().into(), cx)
1937            }),
1938            None => Task::ready(None).shared(),
1939        };
1940
1941        let env = cx.spawn(async move |_, _| {
1942            let mut env = env.await.unwrap_or_default();
1943            if cfg!(unix) {
1944                env.insert("PAGER".into(), "cat".into());
1945            }
1946            for var in extra_env {
1947                env.insert(var.name, var.value);
1948            }
1949            env
1950        });
1951
1952        let project = self.project.clone();
1953        let language_registry = project.read(cx).languages().clone();
1954
1955        let terminal_id = acp::TerminalId(Uuid::new_v4().to_string().into());
1956        let terminal_task = cx.spawn({
1957            let terminal_id = terminal_id.clone();
1958            async move |_this, cx| {
1959                let env = env.await;
1960                let (command, args) = ShellBuilder::new(
1961                    project
1962                        .update(cx, |project, cx| {
1963                            project
1964                                .remote_client()
1965                                .and_then(|r| r.read(cx).default_system_shell())
1966                        })?
1967                        .as_deref(),
1968                    &Shell::Program(get_default_system_shell()),
1969                )
1970                .redirect_stdin_to_dev_null()
1971                .build(Some(command), &args);
1972                let terminal = project
1973                    .update(cx, |project, cx| {
1974                        project.create_terminal_task(
1975                            task::SpawnInTerminal {
1976                                command: Some(command.clone()),
1977                                args: args.clone(),
1978                                cwd: cwd.clone(),
1979                                env,
1980                                ..Default::default()
1981                            },
1982                            cx,
1983                        )
1984                    })?
1985                    .await?;
1986
1987                cx.new(|cx| {
1988                    Terminal::new(
1989                        terminal_id,
1990                        &format!("{} {}", command, args.join(" ")),
1991                        cwd,
1992                        output_byte_limit.map(|l| l as usize),
1993                        terminal,
1994                        language_registry,
1995                        cx,
1996                    )
1997                })
1998            }
1999        });
2000
2001        cx.spawn(async move |this, cx| {
2002            let terminal = terminal_task.await?;
2003            this.update(cx, |this, _cx| {
2004                this.terminals.insert(terminal_id, terminal.clone());
2005                terminal
2006            })
2007        })
2008    }
2009
2010    pub fn kill_terminal(
2011        &mut self,
2012        terminal_id: acp::TerminalId,
2013        cx: &mut Context<Self>,
2014    ) -> Result<()> {
2015        self.terminals
2016            .get(&terminal_id)
2017            .context("Terminal not found")?
2018            .update(cx, |terminal, cx| {
2019                terminal.kill(cx);
2020            });
2021
2022        Ok(())
2023    }
2024
2025    pub fn release_terminal(
2026        &mut self,
2027        terminal_id: acp::TerminalId,
2028        cx: &mut Context<Self>,
2029    ) -> Result<()> {
2030        self.terminals
2031            .remove(&terminal_id)
2032            .context("Terminal not found")?
2033            .update(cx, |terminal, cx| {
2034                terminal.kill(cx);
2035            });
2036
2037        Ok(())
2038    }
2039
2040    pub fn terminal(&self, terminal_id: acp::TerminalId) -> Result<Entity<Terminal>> {
2041        self.terminals
2042            .get(&terminal_id)
2043            .context("Terminal not found")
2044            .cloned()
2045    }
2046
2047    pub fn to_markdown(&self, cx: &App) -> String {
2048        self.entries.iter().map(|e| e.to_markdown(cx)).collect()
2049    }
2050
2051    pub fn emit_load_error(&mut self, error: LoadError, cx: &mut Context<Self>) {
2052        cx.emit(AcpThreadEvent::LoadError(error));
2053    }
2054}
2055
2056fn markdown_for_raw_output(
2057    raw_output: &serde_json::Value,
2058    language_registry: &Arc<LanguageRegistry>,
2059    cx: &mut App,
2060) -> Option<Entity<Markdown>> {
2061    match raw_output {
2062        serde_json::Value::Null => None,
2063        serde_json::Value::Bool(value) => Some(cx.new(|cx| {
2064            Markdown::new(
2065                value.to_string().into(),
2066                Some(language_registry.clone()),
2067                None,
2068                cx,
2069            )
2070        })),
2071        serde_json::Value::Number(value) => Some(cx.new(|cx| {
2072            Markdown::new(
2073                value.to_string().into(),
2074                Some(language_registry.clone()),
2075                None,
2076                cx,
2077            )
2078        })),
2079        serde_json::Value::String(value) => Some(cx.new(|cx| {
2080            Markdown::new(
2081                value.clone().into(),
2082                Some(language_registry.clone()),
2083                None,
2084                cx,
2085            )
2086        })),
2087        value => Some(cx.new(|cx| {
2088            Markdown::new(
2089                format!("```json\n{}\n```", value).into(),
2090                Some(language_registry.clone()),
2091                None,
2092                cx,
2093            )
2094        })),
2095    }
2096}
2097
2098#[cfg(test)]
2099mod tests {
2100    use super::*;
2101    use anyhow::anyhow;
2102    use futures::{channel::mpsc, future::LocalBoxFuture, select};
2103    use gpui::{App, AsyncApp, TestAppContext, WeakEntity};
2104    use indoc::indoc;
2105    use project::{FakeFs, Fs};
2106    use rand::{distr, prelude::*};
2107    use serde_json::json;
2108    use settings::SettingsStore;
2109    use smol::stream::StreamExt as _;
2110    use std::{
2111        any::Any,
2112        cell::RefCell,
2113        path::Path,
2114        rc::Rc,
2115        sync::atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
2116        time::Duration,
2117    };
2118    use util::path;
2119
2120    fn init_test(cx: &mut TestAppContext) {
2121        env_logger::try_init().ok();
2122        cx.update(|cx| {
2123            let settings_store = SettingsStore::test(cx);
2124            cx.set_global(settings_store);
2125            Project::init_settings(cx);
2126            language::init(cx);
2127        });
2128    }
2129
2130    #[gpui::test]
2131    async fn test_push_user_content_block(cx: &mut gpui::TestAppContext) {
2132        init_test(cx);
2133
2134        let fs = FakeFs::new(cx.executor());
2135        let project = Project::test(fs, [], cx).await;
2136        let connection = Rc::new(FakeAgentConnection::new());
2137        let thread = cx
2138            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2139            .await
2140            .unwrap();
2141
2142        // Test creating a new user message
2143        thread.update(cx, |thread, cx| {
2144            thread.push_user_content_block(
2145                None,
2146                acp::ContentBlock::Text(acp::TextContent {
2147                    annotations: None,
2148                    text: "Hello, ".to_string(),
2149                    meta: None,
2150                }),
2151                cx,
2152            );
2153        });
2154
2155        thread.update(cx, |thread, cx| {
2156            assert_eq!(thread.entries.len(), 1);
2157            if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
2158                assert_eq!(user_msg.id, None);
2159                assert_eq!(user_msg.content.to_markdown(cx), "Hello, ");
2160            } else {
2161                panic!("Expected UserMessage");
2162            }
2163        });
2164
2165        // Test appending to existing user message
2166        let message_1_id = UserMessageId::new();
2167        thread.update(cx, |thread, cx| {
2168            thread.push_user_content_block(
2169                Some(message_1_id.clone()),
2170                acp::ContentBlock::Text(acp::TextContent {
2171                    annotations: None,
2172                    text: "world!".to_string(),
2173                    meta: None,
2174                }),
2175                cx,
2176            );
2177        });
2178
2179        thread.update(cx, |thread, cx| {
2180            assert_eq!(thread.entries.len(), 1);
2181            if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[0] {
2182                assert_eq!(user_msg.id, Some(message_1_id));
2183                assert_eq!(user_msg.content.to_markdown(cx), "Hello, world!");
2184            } else {
2185                panic!("Expected UserMessage");
2186            }
2187        });
2188
2189        // Test creating new user message after assistant message
2190        thread.update(cx, |thread, cx| {
2191            thread.push_assistant_content_block(
2192                acp::ContentBlock::Text(acp::TextContent {
2193                    annotations: None,
2194                    text: "Assistant response".to_string(),
2195                    meta: None,
2196                }),
2197                false,
2198                cx,
2199            );
2200        });
2201
2202        let message_2_id = UserMessageId::new();
2203        thread.update(cx, |thread, cx| {
2204            thread.push_user_content_block(
2205                Some(message_2_id.clone()),
2206                acp::ContentBlock::Text(acp::TextContent {
2207                    annotations: None,
2208                    text: "New user message".to_string(),
2209                    meta: None,
2210                }),
2211                cx,
2212            );
2213        });
2214
2215        thread.update(cx, |thread, cx| {
2216            assert_eq!(thread.entries.len(), 3);
2217            if let AgentThreadEntry::UserMessage(user_msg) = &thread.entries[2] {
2218                assert_eq!(user_msg.id, Some(message_2_id));
2219                assert_eq!(user_msg.content.to_markdown(cx), "New user message");
2220            } else {
2221                panic!("Expected UserMessage at index 2");
2222            }
2223        });
2224    }
2225
2226    #[gpui::test]
2227    async fn test_thinking_concatenation(cx: &mut gpui::TestAppContext) {
2228        init_test(cx);
2229
2230        let fs = FakeFs::new(cx.executor());
2231        let project = Project::test(fs, [], cx).await;
2232        let connection = Rc::new(FakeAgentConnection::new().on_user_message(
2233            |_, thread, mut cx| {
2234                async move {
2235                    thread.update(&mut cx, |thread, cx| {
2236                        thread
2237                            .handle_session_update(
2238                                acp::SessionUpdate::AgentThoughtChunk {
2239                                    content: "Thinking ".into(),
2240                                },
2241                                cx,
2242                            )
2243                            .unwrap();
2244                        thread
2245                            .handle_session_update(
2246                                acp::SessionUpdate::AgentThoughtChunk {
2247                                    content: "hard!".into(),
2248                                },
2249                                cx,
2250                            )
2251                            .unwrap();
2252                    })?;
2253                    Ok(acp::PromptResponse {
2254                        stop_reason: acp::StopReason::EndTurn,
2255                        meta: None,
2256                    })
2257                }
2258                .boxed_local()
2259            },
2260        ));
2261
2262        let thread = cx
2263            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2264            .await
2265            .unwrap();
2266
2267        thread
2268            .update(cx, |thread, cx| thread.send_raw("Hello from Zed!", cx))
2269            .await
2270            .unwrap();
2271
2272        let output = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
2273        assert_eq!(
2274            output,
2275            indoc! {r#"
2276            ## User
2277
2278            Hello from Zed!
2279
2280            ## Assistant
2281
2282            <thinking>
2283            Thinking hard!
2284            </thinking>
2285
2286            "#}
2287        );
2288    }
2289
2290    #[gpui::test]
2291    async fn test_edits_concurrently_to_user(cx: &mut TestAppContext) {
2292        init_test(cx);
2293
2294        let fs = FakeFs::new(cx.executor());
2295        fs.insert_tree(path!("/tmp"), json!({"foo": "one\ntwo\nthree\n"}))
2296            .await;
2297        let project = Project::test(fs.clone(), [], cx).await;
2298        let (read_file_tx, read_file_rx) = oneshot::channel::<()>();
2299        let read_file_tx = Rc::new(RefCell::new(Some(read_file_tx)));
2300        let connection = Rc::new(FakeAgentConnection::new().on_user_message(
2301            move |_, thread, mut cx| {
2302                let read_file_tx = read_file_tx.clone();
2303                async move {
2304                    let content = thread
2305                        .update(&mut cx, |thread, cx| {
2306                            thread.read_text_file(path!("/tmp/foo").into(), None, None, false, cx)
2307                        })
2308                        .unwrap()
2309                        .await
2310                        .unwrap();
2311                    assert_eq!(content, "one\ntwo\nthree\n");
2312                    read_file_tx.take().unwrap().send(()).unwrap();
2313                    thread
2314                        .update(&mut cx, |thread, cx| {
2315                            thread.write_text_file(
2316                                path!("/tmp/foo").into(),
2317                                "one\ntwo\nthree\nfour\nfive\n".to_string(),
2318                                cx,
2319                            )
2320                        })
2321                        .unwrap()
2322                        .await
2323                        .unwrap();
2324                    Ok(acp::PromptResponse {
2325                        stop_reason: acp::StopReason::EndTurn,
2326                        meta: None,
2327                    })
2328                }
2329                .boxed_local()
2330            },
2331        ));
2332
2333        let (worktree, pathbuf) = project
2334            .update(cx, |project, cx| {
2335                project.find_or_create_worktree(path!("/tmp/foo"), true, cx)
2336            })
2337            .await
2338            .unwrap();
2339        let buffer = project
2340            .update(cx, |project, cx| {
2341                project.open_buffer((worktree.read(cx).id(), pathbuf), cx)
2342            })
2343            .await
2344            .unwrap();
2345
2346        let thread = cx
2347            .update(|cx| connection.new_thread(project, Path::new(path!("/tmp")), cx))
2348            .await
2349            .unwrap();
2350
2351        let request = thread.update(cx, |thread, cx| {
2352            thread.send_raw("Extend the count in /tmp/foo", cx)
2353        });
2354        read_file_rx.await.ok();
2355        buffer.update(cx, |buffer, cx| {
2356            buffer.edit([(0..0, "zero\n".to_string())], None, cx);
2357        });
2358        cx.run_until_parked();
2359        assert_eq!(
2360            buffer.read_with(cx, |buffer, _| buffer.text()),
2361            "zero\none\ntwo\nthree\nfour\nfive\n"
2362        );
2363        assert_eq!(
2364            String::from_utf8(fs.read_file_sync(path!("/tmp/foo")).unwrap()).unwrap(),
2365            "zero\none\ntwo\nthree\nfour\nfive\n"
2366        );
2367        request.await.unwrap();
2368    }
2369
2370    #[gpui::test]
2371    async fn test_succeeding_canceled_toolcall(cx: &mut TestAppContext) {
2372        init_test(cx);
2373
2374        let fs = FakeFs::new(cx.executor());
2375        let project = Project::test(fs, [], cx).await;
2376        let id = acp::ToolCallId("test".into());
2377
2378        let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2379            let id = id.clone();
2380            move |_, thread, mut cx| {
2381                let id = id.clone();
2382                async move {
2383                    thread
2384                        .update(&mut cx, |thread, cx| {
2385                            thread.handle_session_update(
2386                                acp::SessionUpdate::ToolCall(acp::ToolCall {
2387                                    id: id.clone(),
2388                                    title: "Label".into(),
2389                                    kind: acp::ToolKind::Fetch,
2390                                    status: acp::ToolCallStatus::InProgress,
2391                                    content: vec![],
2392                                    locations: vec![],
2393                                    raw_input: None,
2394                                    raw_output: None,
2395                                    meta: None,
2396                                }),
2397                                cx,
2398                            )
2399                        })
2400                        .unwrap()
2401                        .unwrap();
2402                    Ok(acp::PromptResponse {
2403                        stop_reason: acp::StopReason::EndTurn,
2404                        meta: None,
2405                    })
2406                }
2407                .boxed_local()
2408            }
2409        }));
2410
2411        let thread = cx
2412            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2413            .await
2414            .unwrap();
2415
2416        let request = thread.update(cx, |thread, cx| {
2417            thread.send_raw("Fetch https://example.com", cx)
2418        });
2419
2420        run_until_first_tool_call(&thread, cx).await;
2421
2422        thread.read_with(cx, |thread, _| {
2423            assert!(matches!(
2424                thread.entries[1],
2425                AgentThreadEntry::ToolCall(ToolCall {
2426                    status: ToolCallStatus::InProgress,
2427                    ..
2428                })
2429            ));
2430        });
2431
2432        thread.update(cx, |thread, cx| thread.cancel(cx)).await;
2433
2434        thread.read_with(cx, |thread, _| {
2435            assert!(matches!(
2436                &thread.entries[1],
2437                AgentThreadEntry::ToolCall(ToolCall {
2438                    status: ToolCallStatus::Canceled,
2439                    ..
2440                })
2441            ));
2442        });
2443
2444        thread
2445            .update(cx, |thread, cx| {
2446                thread.handle_session_update(
2447                    acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate {
2448                        id,
2449                        fields: acp::ToolCallUpdateFields {
2450                            status: Some(acp::ToolCallStatus::Completed),
2451                            ..Default::default()
2452                        },
2453                        meta: None,
2454                    }),
2455                    cx,
2456                )
2457            })
2458            .unwrap();
2459
2460        request.await.unwrap();
2461
2462        thread.read_with(cx, |thread, _| {
2463            assert!(matches!(
2464                thread.entries[1],
2465                AgentThreadEntry::ToolCall(ToolCall {
2466                    status: ToolCallStatus::Completed,
2467                    ..
2468                })
2469            ));
2470        });
2471    }
2472
2473    #[gpui::test]
2474    async fn test_no_pending_edits_if_tool_calls_are_completed(cx: &mut TestAppContext) {
2475        init_test(cx);
2476        let fs = FakeFs::new(cx.background_executor.clone());
2477        fs.insert_tree(path!("/test"), json!({})).await;
2478        let project = Project::test(fs, [path!("/test").as_ref()], cx).await;
2479
2480        let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2481            move |_, thread, mut cx| {
2482                async move {
2483                    thread
2484                        .update(&mut cx, |thread, cx| {
2485                            thread.handle_session_update(
2486                                acp::SessionUpdate::ToolCall(acp::ToolCall {
2487                                    id: acp::ToolCallId("test".into()),
2488                                    title: "Label".into(),
2489                                    kind: acp::ToolKind::Edit,
2490                                    status: acp::ToolCallStatus::Completed,
2491                                    content: vec![acp::ToolCallContent::Diff {
2492                                        diff: acp::Diff {
2493                                            path: "/test/test.txt".into(),
2494                                            old_text: None,
2495                                            new_text: "foo".into(),
2496                                            meta: None,
2497                                        },
2498                                    }],
2499                                    locations: vec![],
2500                                    raw_input: None,
2501                                    raw_output: None,
2502                                    meta: None,
2503                                }),
2504                                cx,
2505                            )
2506                        })
2507                        .unwrap()
2508                        .unwrap();
2509                    Ok(acp::PromptResponse {
2510                        stop_reason: acp::StopReason::EndTurn,
2511                        meta: None,
2512                    })
2513                }
2514                .boxed_local()
2515            }
2516        }));
2517
2518        let thread = cx
2519            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2520            .await
2521            .unwrap();
2522
2523        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Hi".into()], cx)))
2524            .await
2525            .unwrap();
2526
2527        assert!(cx.read(|cx| !thread.read(cx).has_pending_edit_tool_calls()));
2528    }
2529
2530    #[gpui::test(iterations = 10)]
2531    async fn test_checkpoints(cx: &mut TestAppContext) {
2532        init_test(cx);
2533        let fs = FakeFs::new(cx.background_executor.clone());
2534        fs.insert_tree(
2535            path!("/test"),
2536            json!({
2537                ".git": {}
2538            }),
2539        )
2540        .await;
2541        let project = Project::test(fs.clone(), [path!("/test").as_ref()], cx).await;
2542
2543        let simulate_changes = Arc::new(AtomicBool::new(true));
2544        let next_filename = Arc::new(AtomicUsize::new(0));
2545        let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2546            let simulate_changes = simulate_changes.clone();
2547            let next_filename = next_filename.clone();
2548            let fs = fs.clone();
2549            move |request, thread, mut cx| {
2550                let fs = fs.clone();
2551                let simulate_changes = simulate_changes.clone();
2552                let next_filename = next_filename.clone();
2553                async move {
2554                    if simulate_changes.load(SeqCst) {
2555                        let filename = format!("/test/file-{}", next_filename.fetch_add(1, SeqCst));
2556                        fs.write(Path::new(&filename), b"").await?;
2557                    }
2558
2559                    let acp::ContentBlock::Text(content) = &request.prompt[0] else {
2560                        panic!("expected text content block");
2561                    };
2562                    thread.update(&mut cx, |thread, cx| {
2563                        thread
2564                            .handle_session_update(
2565                                acp::SessionUpdate::AgentMessageChunk {
2566                                    content: content.text.to_uppercase().into(),
2567                                },
2568                                cx,
2569                            )
2570                            .unwrap();
2571                    })?;
2572                    Ok(acp::PromptResponse {
2573                        stop_reason: acp::StopReason::EndTurn,
2574                        meta: None,
2575                    })
2576                }
2577                .boxed_local()
2578            }
2579        }));
2580        let thread = cx
2581            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2582            .await
2583            .unwrap();
2584
2585        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["Lorem".into()], cx)))
2586            .await
2587            .unwrap();
2588        thread.read_with(cx, |thread, cx| {
2589            assert_eq!(
2590                thread.to_markdown(cx),
2591                indoc! {"
2592                    ## User (checkpoint)
2593
2594                    Lorem
2595
2596                    ## Assistant
2597
2598                    LOREM
2599
2600                "}
2601            );
2602        });
2603        assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
2604
2605        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["ipsum".into()], cx)))
2606            .await
2607            .unwrap();
2608        thread.read_with(cx, |thread, cx| {
2609            assert_eq!(
2610                thread.to_markdown(cx),
2611                indoc! {"
2612                    ## User (checkpoint)
2613
2614                    Lorem
2615
2616                    ## Assistant
2617
2618                    LOREM
2619
2620                    ## User (checkpoint)
2621
2622                    ipsum
2623
2624                    ## Assistant
2625
2626                    IPSUM
2627
2628                "}
2629            );
2630        });
2631        assert_eq!(
2632            fs.files(),
2633            vec![
2634                Path::new(path!("/test/file-0")),
2635                Path::new(path!("/test/file-1"))
2636            ]
2637        );
2638
2639        // Checkpoint isn't stored when there are no changes.
2640        simulate_changes.store(false, SeqCst);
2641        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["dolor".into()], cx)))
2642            .await
2643            .unwrap();
2644        thread.read_with(cx, |thread, cx| {
2645            assert_eq!(
2646                thread.to_markdown(cx),
2647                indoc! {"
2648                    ## User (checkpoint)
2649
2650                    Lorem
2651
2652                    ## Assistant
2653
2654                    LOREM
2655
2656                    ## User (checkpoint)
2657
2658                    ipsum
2659
2660                    ## Assistant
2661
2662                    IPSUM
2663
2664                    ## User
2665
2666                    dolor
2667
2668                    ## Assistant
2669
2670                    DOLOR
2671
2672                "}
2673            );
2674        });
2675        assert_eq!(
2676            fs.files(),
2677            vec![
2678                Path::new(path!("/test/file-0")),
2679                Path::new(path!("/test/file-1"))
2680            ]
2681        );
2682
2683        // Rewinding the conversation truncates the history and restores the checkpoint.
2684        thread
2685            .update(cx, |thread, cx| {
2686                let AgentThreadEntry::UserMessage(message) = &thread.entries[2] else {
2687                    panic!("unexpected entries {:?}", thread.entries)
2688                };
2689                thread.restore_checkpoint(message.id.clone().unwrap(), cx)
2690            })
2691            .await
2692            .unwrap();
2693        thread.read_with(cx, |thread, cx| {
2694            assert_eq!(
2695                thread.to_markdown(cx),
2696                indoc! {"
2697                    ## User (checkpoint)
2698
2699                    Lorem
2700
2701                    ## Assistant
2702
2703                    LOREM
2704
2705                "}
2706            );
2707        });
2708        assert_eq!(fs.files(), vec![Path::new(path!("/test/file-0"))]);
2709    }
2710
2711    #[gpui::test]
2712    async fn test_tool_result_refusal(cx: &mut TestAppContext) {
2713        use std::sync::atomic::AtomicUsize;
2714        init_test(cx);
2715
2716        let fs = FakeFs::new(cx.executor());
2717        let project = Project::test(fs, None, cx).await;
2718
2719        // Create a connection that simulates refusal after tool result
2720        let prompt_count = Arc::new(AtomicUsize::new(0));
2721        let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2722            let prompt_count = prompt_count.clone();
2723            move |_request, thread, mut cx| {
2724                let count = prompt_count.fetch_add(1, SeqCst);
2725                async move {
2726                    if count == 0 {
2727                        // First prompt: Generate a tool call with result
2728                        thread.update(&mut cx, |thread, cx| {
2729                            thread
2730                                .handle_session_update(
2731                                    acp::SessionUpdate::ToolCall(acp::ToolCall {
2732                                        id: acp::ToolCallId("tool1".into()),
2733                                        title: "Test Tool".into(),
2734                                        kind: acp::ToolKind::Fetch,
2735                                        status: acp::ToolCallStatus::Completed,
2736                                        content: vec![],
2737                                        locations: vec![],
2738                                        raw_input: Some(serde_json::json!({"query": "test"})),
2739                                        raw_output: Some(
2740                                            serde_json::json!({"result": "inappropriate content"}),
2741                                        ),
2742                                        meta: None,
2743                                    }),
2744                                    cx,
2745                                )
2746                                .unwrap();
2747                        })?;
2748
2749                        // Now return refusal because of the tool result
2750                        Ok(acp::PromptResponse {
2751                            stop_reason: acp::StopReason::Refusal,
2752                            meta: None,
2753                        })
2754                    } else {
2755                        Ok(acp::PromptResponse {
2756                            stop_reason: acp::StopReason::EndTurn,
2757                            meta: None,
2758                        })
2759                    }
2760                }
2761                .boxed_local()
2762            }
2763        }));
2764
2765        let thread = cx
2766            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2767            .await
2768            .unwrap();
2769
2770        // Track if we see a Refusal event
2771        let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
2772        let saw_refusal_event_captured = saw_refusal_event.clone();
2773        thread.update(cx, |_thread, cx| {
2774            cx.subscribe(
2775                &thread,
2776                move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
2777                    if matches!(event, AcpThreadEvent::Refusal) {
2778                        *saw_refusal_event_captured.lock().unwrap() = true;
2779                    }
2780                },
2781            )
2782            .detach();
2783        });
2784
2785        // Send a user message - this will trigger tool call and then refusal
2786        let send_task = thread.update(cx, |thread, cx| {
2787            thread.send(
2788                vec![acp::ContentBlock::Text(acp::TextContent {
2789                    text: "Hello".into(),
2790                    annotations: None,
2791                    meta: None,
2792                })],
2793                cx,
2794            )
2795        });
2796        cx.background_executor.spawn(send_task).detach();
2797        cx.run_until_parked();
2798
2799        // Verify that:
2800        // 1. A Refusal event WAS emitted (because it's a tool result refusal, not user prompt)
2801        // 2. The user message was NOT truncated
2802        assert!(
2803            *saw_refusal_event.lock().unwrap(),
2804            "Refusal event should be emitted for tool result refusals"
2805        );
2806
2807        thread.read_with(cx, |thread, _| {
2808            let entries = thread.entries();
2809            assert!(entries.len() >= 2, "Should have user message and tool call");
2810
2811            // Verify user message is still there
2812            assert!(
2813                matches!(entries[0], AgentThreadEntry::UserMessage(_)),
2814                "User message should not be truncated"
2815            );
2816
2817            // Verify tool call is there with result
2818            if let AgentThreadEntry::ToolCall(tool_call) = &entries[1] {
2819                assert!(
2820                    tool_call.raw_output.is_some(),
2821                    "Tool call should have output"
2822                );
2823            } else {
2824                panic!("Expected tool call at index 1");
2825            }
2826        });
2827    }
2828
2829    #[gpui::test]
2830    async fn test_user_prompt_refusal_emits_event(cx: &mut TestAppContext) {
2831        init_test(cx);
2832
2833        let fs = FakeFs::new(cx.executor());
2834        let project = Project::test(fs, None, cx).await;
2835
2836        let refuse_next = Arc::new(AtomicBool::new(false));
2837        let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2838            let refuse_next = refuse_next.clone();
2839            move |_request, _thread, _cx| {
2840                if refuse_next.load(SeqCst) {
2841                    async move {
2842                        Ok(acp::PromptResponse {
2843                            stop_reason: acp::StopReason::Refusal,
2844                            meta: None,
2845                        })
2846                    }
2847                    .boxed_local()
2848                } else {
2849                    async move {
2850                        Ok(acp::PromptResponse {
2851                            stop_reason: acp::StopReason::EndTurn,
2852                            meta: None,
2853                        })
2854                    }
2855                    .boxed_local()
2856                }
2857            }
2858        }));
2859
2860        let thread = cx
2861            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2862            .await
2863            .unwrap();
2864
2865        // Track if we see a Refusal event
2866        let saw_refusal_event = Arc::new(std::sync::Mutex::new(false));
2867        let saw_refusal_event_captured = saw_refusal_event.clone();
2868        thread.update(cx, |_thread, cx| {
2869            cx.subscribe(
2870                &thread,
2871                move |_thread, _event_thread, event: &AcpThreadEvent, _cx| {
2872                    if matches!(event, AcpThreadEvent::Refusal) {
2873                        *saw_refusal_event_captured.lock().unwrap() = true;
2874                    }
2875                },
2876            )
2877            .detach();
2878        });
2879
2880        // Send a message that will be refused
2881        refuse_next.store(true, SeqCst);
2882        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
2883            .await
2884            .unwrap();
2885
2886        // Verify that a Refusal event WAS emitted for user prompt refusal
2887        assert!(
2888            *saw_refusal_event.lock().unwrap(),
2889            "Refusal event should be emitted for user prompt refusals"
2890        );
2891
2892        // Verify the message was truncated (user prompt refusal)
2893        thread.read_with(cx, |thread, cx| {
2894            assert_eq!(thread.to_markdown(cx), "");
2895        });
2896    }
2897
2898    #[gpui::test]
2899    async fn test_refusal(cx: &mut TestAppContext) {
2900        init_test(cx);
2901        let fs = FakeFs::new(cx.background_executor.clone());
2902        fs.insert_tree(path!("/"), json!({})).await;
2903        let project = Project::test(fs.clone(), [path!("/").as_ref()], cx).await;
2904
2905        let refuse_next = Arc::new(AtomicBool::new(false));
2906        let connection = Rc::new(FakeAgentConnection::new().on_user_message({
2907            let refuse_next = refuse_next.clone();
2908            move |request, thread, mut cx| {
2909                let refuse_next = refuse_next.clone();
2910                async move {
2911                    if refuse_next.load(SeqCst) {
2912                        return Ok(acp::PromptResponse {
2913                            stop_reason: acp::StopReason::Refusal,
2914                            meta: None,
2915                        });
2916                    }
2917
2918                    let acp::ContentBlock::Text(content) = &request.prompt[0] else {
2919                        panic!("expected text content block");
2920                    };
2921                    thread.update(&mut cx, |thread, cx| {
2922                        thread
2923                            .handle_session_update(
2924                                acp::SessionUpdate::AgentMessageChunk {
2925                                    content: content.text.to_uppercase().into(),
2926                                },
2927                                cx,
2928                            )
2929                            .unwrap();
2930                    })?;
2931                    Ok(acp::PromptResponse {
2932                        stop_reason: acp::StopReason::EndTurn,
2933                        meta: None,
2934                    })
2935                }
2936                .boxed_local()
2937            }
2938        }));
2939        let thread = cx
2940            .update(|cx| connection.new_thread(project, Path::new(path!("/test")), cx))
2941            .await
2942            .unwrap();
2943
2944        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["hello".into()], cx)))
2945            .await
2946            .unwrap();
2947        thread.read_with(cx, |thread, cx| {
2948            assert_eq!(
2949                thread.to_markdown(cx),
2950                indoc! {"
2951                    ## User
2952
2953                    hello
2954
2955                    ## Assistant
2956
2957                    HELLO
2958
2959                "}
2960            );
2961        });
2962
2963        // Simulate refusing the second message. The message should be truncated
2964        // when a user prompt is refused.
2965        refuse_next.store(true, SeqCst);
2966        cx.update(|cx| thread.update(cx, |thread, cx| thread.send(vec!["world".into()], cx)))
2967            .await
2968            .unwrap();
2969        thread.read_with(cx, |thread, cx| {
2970            assert_eq!(
2971                thread.to_markdown(cx),
2972                indoc! {"
2973                    ## User
2974
2975                    hello
2976
2977                    ## Assistant
2978
2979                    HELLO
2980
2981                "}
2982            );
2983        });
2984    }
2985
2986    async fn run_until_first_tool_call(
2987        thread: &Entity<AcpThread>,
2988        cx: &mut TestAppContext,
2989    ) -> usize {
2990        let (mut tx, mut rx) = mpsc::channel::<usize>(1);
2991
2992        let subscription = cx.update(|cx| {
2993            cx.subscribe(thread, move |thread, _, cx| {
2994                for (ix, entry) in thread.read(cx).entries.iter().enumerate() {
2995                    if matches!(entry, AgentThreadEntry::ToolCall(_)) {
2996                        return tx.try_send(ix).unwrap();
2997                    }
2998                }
2999            })
3000        });
3001
3002        select! {
3003            _ = futures::FutureExt::fuse(smol::Timer::after(Duration::from_secs(10))) => {
3004                panic!("Timeout waiting for tool call")
3005            }
3006            ix = rx.next().fuse() => {
3007                drop(subscription);
3008                ix.unwrap()
3009            }
3010        }
3011    }
3012
3013    #[derive(Clone, Default)]
3014    struct FakeAgentConnection {
3015        auth_methods: Vec<acp::AuthMethod>,
3016        sessions: Arc<parking_lot::Mutex<HashMap<acp::SessionId, WeakEntity<AcpThread>>>>,
3017        on_user_message: Option<
3018            Rc<
3019                dyn Fn(
3020                        acp::PromptRequest,
3021                        WeakEntity<AcpThread>,
3022                        AsyncApp,
3023                    ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
3024                    + 'static,
3025            >,
3026        >,
3027    }
3028
3029    impl FakeAgentConnection {
3030        fn new() -> Self {
3031            Self {
3032                auth_methods: Vec::new(),
3033                on_user_message: None,
3034                sessions: Arc::default(),
3035            }
3036        }
3037
3038        #[expect(unused)]
3039        fn with_auth_methods(mut self, auth_methods: Vec<acp::AuthMethod>) -> Self {
3040            self.auth_methods = auth_methods;
3041            self
3042        }
3043
3044        fn on_user_message(
3045            mut self,
3046            handler: impl Fn(
3047                acp::PromptRequest,
3048                WeakEntity<AcpThread>,
3049                AsyncApp,
3050            ) -> LocalBoxFuture<'static, Result<acp::PromptResponse>>
3051            + 'static,
3052        ) -> Self {
3053            self.on_user_message.replace(Rc::new(handler));
3054            self
3055        }
3056    }
3057
3058    impl AgentConnection for FakeAgentConnection {
3059        fn auth_methods(&self) -> &[acp::AuthMethod] {
3060            &self.auth_methods
3061        }
3062
3063        fn new_thread(
3064            self: Rc<Self>,
3065            project: Entity<Project>,
3066            _cwd: &Path,
3067            cx: &mut App,
3068        ) -> Task<gpui::Result<Entity<AcpThread>>> {
3069            let session_id = acp::SessionId(
3070                rand::rng()
3071                    .sample_iter(&distr::Alphanumeric)
3072                    .take(7)
3073                    .map(char::from)
3074                    .collect::<String>()
3075                    .into(),
3076            );
3077            let action_log = cx.new(|_| ActionLog::new(project.clone()));
3078            let thread = cx.new(|cx| {
3079                AcpThread::new(
3080                    "Test",
3081                    self.clone(),
3082                    project,
3083                    action_log,
3084                    session_id.clone(),
3085                    watch::Receiver::constant(acp::PromptCapabilities {
3086                        image: true,
3087                        audio: true,
3088                        embedded_context: true,
3089                        meta: None,
3090                    }),
3091                    cx,
3092                )
3093            });
3094            self.sessions.lock().insert(session_id, thread.downgrade());
3095            Task::ready(Ok(thread))
3096        }
3097
3098        fn authenticate(&self, method: acp::AuthMethodId, _cx: &mut App) -> Task<gpui::Result<()>> {
3099            if self.auth_methods().iter().any(|m| m.id == method) {
3100                Task::ready(Ok(()))
3101            } else {
3102                Task::ready(Err(anyhow!("Invalid Auth Method")))
3103            }
3104        }
3105
3106        fn prompt(
3107            &self,
3108            _id: Option<UserMessageId>,
3109            params: acp::PromptRequest,
3110            cx: &mut App,
3111        ) -> Task<gpui::Result<acp::PromptResponse>> {
3112            let sessions = self.sessions.lock();
3113            let thread = sessions.get(&params.session_id).unwrap();
3114            if let Some(handler) = &self.on_user_message {
3115                let handler = handler.clone();
3116                let thread = thread.clone();
3117                cx.spawn(async move |cx| handler(params, thread, cx.clone()).await)
3118            } else {
3119                Task::ready(Ok(acp::PromptResponse {
3120                    stop_reason: acp::StopReason::EndTurn,
3121                    meta: None,
3122                }))
3123            }
3124        }
3125
3126        fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
3127            let sessions = self.sessions.lock();
3128            let thread = sessions.get(session_id).unwrap().clone();
3129
3130            cx.spawn(async move |cx| {
3131                thread
3132                    .update(cx, |thread, cx| thread.cancel(cx))
3133                    .unwrap()
3134                    .await
3135            })
3136            .detach();
3137        }
3138
3139        fn truncate(
3140            &self,
3141            session_id: &acp::SessionId,
3142            _cx: &App,
3143        ) -> Option<Rc<dyn AgentSessionTruncate>> {
3144            Some(Rc::new(FakeAgentSessionEditor {
3145                _session_id: session_id.clone(),
3146            }))
3147        }
3148
3149        fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
3150            self
3151        }
3152    }
3153
3154    struct FakeAgentSessionEditor {
3155        _session_id: acp::SessionId,
3156    }
3157
3158    impl AgentSessionTruncate for FakeAgentSessionEditor {
3159        fn run(&self, _message_id: UserMessageId, _cx: &mut App) -> Task<Result<()>> {
3160            Task::ready(Ok(()))
3161        }
3162    }
3163}