use super::*;use acp_thread::{    AgentConnection, AgentModelGroupName, AgentModelList, PermissionOptions, UserMessageId,};use agent_client_protocol::{self as acp};use agent_settings::AgentProfileId;use anyhow::Result;use client::{Client, UserStore};use cloud_llm_client::CompletionIntent;use collections::IndexMap;use context_server::{ContextServer, ContextServerCommand, ContextServerId};use feature_flags::FeatureFlagAppExt as _;use fs::{FakeFs, Fs};use futures::{    FutureExt as _, StreamExt,    channel::{        mpsc::{self, UnboundedReceiver},        oneshot,    },    future::{Fuse, Shared},};use gpui::{    App, AppContext, AsyncApp, Entity, Task, TestAppContext, UpdateGlobal,    http_client::FakeHttpClient,};use indoc::indoc;use language_model::{    LanguageModel, LanguageModelCompletionError, LanguageModelCompletionEvent, LanguageModelId,    LanguageModelProviderName, LanguageModelRegistry, LanguageModelRequest,    LanguageModelRequestMessage, LanguageModelToolResult, LanguageModelToolSchemaFormat,    LanguageModelToolUse, MessageContent, Role, StopReason, fake_provider::FakeLanguageModel,};use pretty_assertions::assert_eq;use project::{    Project, context_server_store::ContextServerStore, project_settings::ProjectSettings,};use prompt_store::ProjectContext;use reqwest_client::ReqwestClient;use schemars::JsonSchema;use serde::{Deserialize, Serialize};use serde_json::json;use settings::{Settings, SettingsStore};use std::{    path::Path,    pin::Pin,    rc::Rc,    sync::{        Arc,        atomic::{AtomicBool, Ordering},    },    time::Duration,};use util::path; mod edit_file_thread_test;mod test_tools;use test_tools::*; fn init_test(cx: &mut TestAppContext) {    cx.update(|cx| {        let settings_store = SettingsStore::test(cx);        cx.set_global(settings_store);    });} struct FakeTerminalHandle {    killed: Arc<AtomicBool>,    stopped_by_user: Arc<AtomicBool>,    exit_sender: std::cell::RefCell<Option<futures::channel::oneshot::Sender<()>>>,    wait_for_exit: Shared<Task<acp::TerminalExitStatus>>,    output: acp::TerminalOutputResponse,    id: acp::TerminalId,} impl FakeTerminalHandle {    fn new_never_exits(cx: &mut App) -> Self {        let killed = Arc::new(AtomicBool::new(false));        let stopped_by_user = Arc::new(AtomicBool::new(false));         let (exit_sender, exit_receiver) = futures::channel::oneshot::channel();         let wait_for_exit = cx            .spawn(async move |_cx| {                // Wait for the exit signal (sent when kill() is called)                let _ = exit_receiver.await;                acp::TerminalExitStatus::new()            })            .shared();         Self {            killed,            stopped_by_user,            exit_sender: std::cell::RefCell::new(Some(exit_sender)),            wait_for_exit,            output: acp::TerminalOutputResponse::new("partial output".to_string(), false),            id: acp::TerminalId::new("fake_terminal".to_string()),        }    }     fn new_with_immediate_exit(cx: &mut App, exit_code: u32) -> Self {        let killed = Arc::new(AtomicBool::new(false));        let stopped_by_user = Arc::new(AtomicBool::new(false));        let (exit_sender, _exit_receiver) = futures::channel::oneshot::channel();         let wait_for_exit = cx            .spawn(async move |_cx| acp::TerminalExitStatus::new().exit_code(exit_code))            .shared();         Self {            killed,            stopped_by_user,            exit_sender: std::cell::RefCell::new(Some(exit_sender)),            wait_for_exit,            output: acp::TerminalOutputResponse::new("command output".to_string(), false),            id: acp::TerminalId::new("fake_terminal".to_string()),        }    }     fn was_killed(&self) -> bool {        self.killed.load(Ordering::SeqCst)    }     fn set_stopped_by_user(&self, stopped: bool) {        self.stopped_by_user.store(stopped, Ordering::SeqCst);    }     fn signal_exit(&self) {        if let Some(sender) = self.exit_sender.borrow_mut().take() {            let _ = sender.send(());        }    }} impl crate::TerminalHandle for FakeTerminalHandle {    fn id(&self, _cx: &AsyncApp) -> Result<acp::TerminalId> {        Ok(self.id.clone())    }     fn current_output(&self, _cx: &AsyncApp) -> Result<acp::TerminalOutputResponse> {        Ok(self.output.clone())    }     fn wait_for_exit(&self, _cx: &AsyncApp) -> Result<Shared<Task<acp::TerminalExitStatus>>> {        Ok(self.wait_for_exit.clone())    }     fn kill(&self, _cx: &AsyncApp) -> Result<()> {        self.killed.store(true, Ordering::SeqCst);        self.signal_exit();        Ok(())    }     fn was_stopped_by_user(&self, _cx: &AsyncApp) -> Result<bool> {        Ok(self.stopped_by_user.load(Ordering::SeqCst))    }} struct FakeThreadEnvironment {    handle: Rc<FakeTerminalHandle>,} impl crate::ThreadEnvironment for FakeThreadEnvironment {    fn create_terminal(        &self,        _command: String,        _cwd: Option<std::path::PathBuf>,        _output_byte_limit: Option<u64>,        _cx: &mut AsyncApp,    ) -> Task<Result<Rc<dyn crate::TerminalHandle>>> {        Task::ready(Ok(self.handle.clone() as Rc<dyn crate::TerminalHandle>))    }} /// Environment that creates multiple independent terminal handles for testing concurrent terminals.struct MultiTerminalEnvironment {    handles: std::cell::RefCell<Vec<Rc<FakeTerminalHandle>>>,} impl MultiTerminalEnvironment {    fn new() -> Self {        Self {            handles: std::cell::RefCell::new(Vec::new()),        }    }     fn handles(&self) -> Vec<Rc<FakeTerminalHandle>> {        self.handles.borrow().clone()    }} impl crate::ThreadEnvironment for MultiTerminalEnvironment {    fn create_terminal(        &self,        _command: String,        _cwd: Option<std::path::PathBuf>,        _output_byte_limit: Option<u64>,        cx: &mut AsyncApp,    ) -> Task<Result<Rc<dyn crate::TerminalHandle>>> {        let handle = Rc::new(cx.update(|cx| FakeTerminalHandle::new_never_exits(cx)));        self.handles.borrow_mut().push(handle.clone());        Task::ready(Ok(handle as Rc<dyn crate::TerminalHandle>))    }} fn always_allow_tools(cx: &mut TestAppContext) {    cx.update(|cx| {        let mut settings = agent_settings::AgentSettings::get_global(cx).clone();        settings.always_allow_tool_actions = true;        agent_settings::AgentSettings::override_global(settings, cx);    });} #[gpui::test]async fn test_echo(cx: &mut TestAppContext) {    let ThreadTest { model, thread, .. } = setup(cx, TestModel::Fake).await;    let fake_model = model.as_fake();     let events = thread        .update(cx, |thread, cx| {            thread.send(UserMessageId::new(), ["Testing: Reply with 'Hello'"], cx)        })        .unwrap();    cx.run_until_parked();    fake_model.send_last_completion_stream_text_chunk("Hello");    fake_model        .send_last_completion_stream_event(LanguageModelCompletionEvent::Stop(StopReason::EndTurn));    fake_model.end_last_completion_stream();     let events = events.collect().await;    thread.update(cx, |thread, _cx| {        assert_eq!(            thread.last_message().unwrap().to_markdown(),            indoc! {"                ## Assistant                 Hello            "}        )    });    assert_eq!(stop_events(events), vec![acp::StopReason::EndTurn]);} #[gpui::test]async fn test_terminal_tool_timeout_kills_handle(cx: &mut TestAppContext) {    init_test(cx);    always_allow_tools(cx);     let fs = FakeFs::new(cx.executor());    let project = Project::test(fs, [], cx).await;     let handle = Rc::new(cx.update(|cx| FakeTerminalHandle::new_never_exits(cx)));    let environment = Rc::new(FakeThreadEnvironment {        handle: handle.clone(),    });     #[allow(clippy::arc_with_non_send_sync)]    let tool = Arc::new(crate::TerminalTool::new(project, environment));    let (event_stream, mut rx) = crate::ToolCallEventStream::test();     let task = cx.update(|cx| {        tool.run(            crate::TerminalToolInput {                command: "sleep 1000".to_string(),                cd: ".".to_string(),                timeout_ms: Some(5),            },            event_stream,            cx,        )    });     let update = rx.expect_update_fields().await;    assert!(        update.content.iter().any(|blocks| {            blocks                .iter()                .any(|c| matches!(c, acp::ToolCallContent::Terminal(_)))        }),        "expected tool call update to include terminal content"    );     let mut task_future: Pin<Box<Fuse<Task<Result<String>>>>> = Box::pin(task.fuse());     let deadline = std::time::Instant::now() + Duration::from_millis(500);    loop {        if let Some(result) = task_future.as_mut().now_or_never() {            let result = result.expect("terminal tool task should complete");             assert!(                handle.was_killed(),                "expected terminal handle to be killed on timeout"            );            assert!(                result.contains("partial output"),                "expected result to include terminal output, got: {result}"            );            return;        }         if std::time::Instant::now() >= deadline {            panic!("timed out waiting for terminal tool task to complete");        }         cx.run_until_parked();        cx.background_executor.timer(Duration::from_millis(1)).await;    }} #[gpui::test]#[ignore]async fn test_terminal_tool_without_timeout_does_not_kill_handle(cx: &mut TestAppContext) {    init_test(cx);    always_allow_tools(cx);     let fs = FakeFs::new(cx.executor());    let project = Project::test(fs, [], cx).await;     let handle = Rc::new(cx.update(|cx| FakeTerminalHandle::new_never_exits(cx)));    let environment = Rc::new(FakeThreadEnvironment {        handle: handle.clone(),    });     #[allow(clippy::arc_with_non_send_sync)]    let tool = Arc::new(crate::TerminalTool::new(project, environment));    let (event_stream, mut rx) = crate::ToolCallEventStream::test();     let _task = cx.update(|cx| {        tool.run(            crate::TerminalToolInput {                command: "sleep 1000".to_string(),                cd: ".".to_string(),                timeout_ms: None,            },            event_stream,            cx,        )    });     let update = rx.expect_update_fields().await;    assert!(        update.content.iter().any(|blocks| {            blocks                .iter()                .any(|c| matches!(c, acp::ToolCallContent::Terminal(_)))        }),        "expected tool call update to include terminal content"    );     cx.background_executor        .timer(Duration::from_millis(25))        .await;     assert!(        !handle.was_killed(),        "did not expect terminal handle to be killed without a timeout"    );} #[gpui::test]async fn test_thinking(cx: &mut TestAppContext) {    let ThreadTest { model, thread, .. } = setup(cx, TestModel::Fake).await;    let fake_model = model.as_fake();     let events = thread        .update(cx, |thread, cx| {            thread.send(                UserMessageId::new(),                [indoc! {"                    Testing:                     Generate a thinking step where you just think the word 'Think',                    and have your final answer be 'Hello'                "}],                cx,            )        })        .unwrap();    cx.run_until_parked();    fake_model.send_last_completion_stream_event(LanguageModelCompletionEvent::Thinking {        text: "Think".to_string(),        signature: None,    });    fake_model.send_last_completion_stream_text_chunk("Hello");    fake_model        .send_last_completion_stream_event(LanguageModelCompletionEvent::Stop(StopReason::EndTurn));    fake_model.end_last_completion_stream();     let events = events.collect().await;    thread.update(cx, |thread, _cx| {        assert_eq!(            thread.last_message().unwrap().to_markdown(),            indoc! {"                ## Assistant                 <think>Think</think>                Hello            "}        )    });    assert_eq!(stop_events(events), vec![acp::StopReason::EndTurn]);} #[gpui::test]async fn test_system_prompt(cx: &mut TestAppContext) {    let ThreadTest {        model,        thread,        project_context,        ..    } = setup(cx, TestModel::Fake).await;    let fake_model = model.as_fake();     project_context.update(cx, |project_context, _cx| {        project_context.shell = "test-shell".into()    });    thread.update(cx, |thread, _| thread.add_tool(EchoTool));    thread        .update(cx, |thread, cx| {            thread.send(UserMessageId::new(), ["abc"], cx)        })        .unwrap();    cx.run_until_parked();    let mut pending_completions = fake_model.pending_completions();    assert_eq!(        pending_completions.len(),        1,        "unexpected pending completions: {:?}",        pending_completions    );     let pending_completion = pending_completions.pop().unwrap();    assert_eq!(pending_completion.messages[0].role, Role::System);     let system_message = &pending_completion.messages[0];    let system_prompt = system_message.content[0].to_str().unwrap();    assert!(        system_prompt.contains("test-shell"),        "unexpected system message: {:?}",        system_message    );    assert!(        system_prompt.contains("## Fixing Diagnostics"),        "unexpected system message: {:?}",        system_message    );} #[gpui::test]async fn test_system_prompt_without_tools(cx: &mut TestAppContext) {    let ThreadTest { model, thread, .. } = setup(cx, TestModel::Fake).await;    let fake_model = model.as_fake();     thread        .update(cx, |thread, cx| {            thread.send(UserMessageId::new(), ["abc"], cx)        })        .unwrap();    cx.run_until_parked();    let mut pending_completions = fake_model.pending_completions();    assert_eq!(        pending_completions.len(),        1,        "unexpected pending completions: {:?}",        pending_completions    );     let pending_completion = pending_completions.pop().unwrap();    assert_eq!(pending_completion.messages[0].role, Role::System);     let system_message = &pending_completion.messages[0];    let system_prompt = system_message.content[0].to_str().unwrap();    assert!(        !system_prompt.contains("## Tool Use"),        "unexpected system message: {:?}",        system_message    );    assert!(        !system_prompt.contains("## Fixing Diagnostics"),        "unexpected system message: {:?}",        system_message    );} #[gpui::test]async fn test_prompt_caching(cx: &mut TestAppContext) {    let ThreadTest { model, thread, .. } = setup(cx, TestModel::Fake).await;    let fake_model = model.as_fake();     // Send initial user message and verify it's cached    thread        .update(cx, |thread, cx| {            thread.send(UserMessageId::new(), ["Message 1"], cx)        })        .unwrap();    cx.run_until_parked();     let completion = fake_model.pending_completions().pop().unwrap();    assert_eq!(        completion.messages[1..],        vec![LanguageModelRequestMessage {

Lines 1–500 of 5693. Load more