use acp_thread::{    AgentConnection, AgentSessionInfo, AgentSessionList, AgentSessionListRequest,    AgentSessionListResponse, ElicitationStore,};use action_log::ActionLog;use agent_client_protocol::schema::{    ProtocolVersion,    v1::{self as acp, ErrorCode},};use agent_client_protocol::{Agent, Client, ConnectionTo, JsonRpcResponse, Lines, Responder};use anyhow::anyhow;use async_channel;use collections::{HashMap, HashSet};use feature_flags::{AcpBetaFeatureFlag, FeatureFlagAppExt as _};use futures::channel::mpsc;use futures::future::Shared;use futures::io::BufReader;use futures::{AsyncBufReadExt as _, Future, FutureExt as _, StreamExt as _};use project::agent_server_store::{    AgentServerCommand, AgentServerStore, AllAgentServersSettings, CustomAgentServerSettings,};use project::{AgentId, Project};use remote::remote_client::Interactive;use serde::Deserialize;use settings::{AgentConfigOptionValue, SettingsStore};use std::path::PathBuf;use std::process::{ExitStatus, Stdio};use std::rc::Rc;use std::sync::{Arc, Mutex};use std::{any::Any, cell::RefCell, collections::VecDeque};use task::{Shell, ShellBuilder, SpawnInTerminal};use thiserror::Error;use util::ResultExt as _;use util::path_list::PathList;use util::process::Child; use anyhow::{Context as _, Result};use gpui::{App, AppContext as _, AsyncApp, Entity, SharedString, Subscription, Task, WeakEntity}; use acp_thread::{AcpThread, AuthRequired, LoadError, TerminalProviderEvent};use terminal::TerminalBuilder;use terminal::terminal_settings::{AlternateScroll, CursorShape}; use crate::{CURSOR_ID, GEMINI_ID}; pub const GEMINI_TERMINAL_AUTH_METHOD_ID: &str = "spawn-gemini-cli";const PARAMETERIZED_MODEL_PICKER_META_KEY: &str = "parameterizedModelPicker";const MAX_DEBUG_BACKLOG_MESSAGES: usize = 2000; #[derive(Clone, Copy, Debug, PartialEq, Eq)]pub enum AcpDebugMessageDirection {    Incoming,    Outgoing,    Stderr,} #[derive(Clone)]pub enum AcpDebugMessageContent {    Request {        id: acp::RequestId,        method: Arc<str>,        params: Option<serde_json::Value>,    },    Response {        id: acp::RequestId,        result: Result<Option<serde_json::Value>, acp::Error>,    },    Notification {        method: Arc<str>,        params: Option<serde_json::Value>,    },    Stderr {        line: Arc<str>,    },} #[derive(Clone)]pub struct AcpDebugMessage {    pub direction: AcpDebugMessageDirection,    pub message: AcpDebugMessageContent,} impl AcpDebugMessage {    fn parse_line(direction: AcpDebugMessageDirection, line: &str) -> Vec<Self> {        if direction == AcpDebugMessageDirection::Stderr {            return vec![Self {                direction,                message: AcpDebugMessageContent::Stderr {                    line: Arc::from(line),                },            }];        }         let Ok(value) = serde_json::from_str(line) else {            return Vec::new();        };         match value {            serde_json::Value::Array(entries) => entries                .into_iter()                .filter_map(|entry| Self::parse_value(direction, entry))                .collect(),            value => Self::parse_value(direction, value).into_iter().collect(),        }    }     fn parse_value(direction: AcpDebugMessageDirection, value: serde_json::Value) -> Option<Self> {        let object = value.as_object()?;         let parsed_id = object            .get("id")            .map(|raw| serde_json::from_value::<acp::RequestId>(raw.clone()));         let message = if let Some(method) = object.get("method").and_then(|method| method.as_str())        {            match parsed_id {                Some(Ok(id)) => AcpDebugMessageContent::Request {                    id,                    method: method.into(),                    params: object.get("params").cloned(),                },                Some(Err(err)) => {                    log::warn!("Skipping JSON-RPC message with unparsable id: {err}");                    return None;                }                None => AcpDebugMessageContent::Notification {                    method: method.into(),                    params: object.get("params").cloned(),                },            }        } else if let Some(parsed_id) = parsed_id {            let id = match parsed_id {                Ok(id) => id,                Err(err) => {                    log::warn!("Skipping JSON-RPC response with unparsable id: {err}");                    return None;                }            };             if let Some(error) = object.get("error") {                let acp_error =                    serde_json::from_value::<acp::Error>(error.clone()).unwrap_or_else(|err| {                        log::warn!("Failed to deserialize ACP error: {err}");                        acp::Error::internal_error().data(error.to_string())                    });                 AcpDebugMessageContent::Response {                    id,                    result: Err(acp_error),                }            } else {                AcpDebugMessageContent::Response {                    id,                    result: Ok(object.get("result").cloned()),                }            }        } else {            return None;        };         Some(Self { direction, message })    }} #[derive(Default)]struct AcpDebugLogState {    messages: VecDeque<AcpDebugMessage>,    subscribers: Vec<async_channel::Sender<AcpDebugMessage>>,} #[derive(Clone, Default)]struct AcpDebugLog {    state: Arc<Mutex<AcpDebugLogState>>,} impl AcpDebugLog {    fn subscribe(        &self,    ) -> (        Vec<AcpDebugMessage>,        async_channel::Receiver<AcpDebugMessage>,    ) {        let mut state = self            .state            .lock()            .unwrap_or_else(|poisoned| poisoned.into_inner());        let backlog = state.messages.iter().cloned().collect();        let (sender, receiver) = async_channel::unbounded();        state.subscribers.push(sender);        (backlog, receiver)    }     fn record_line(&self, direction: AcpDebugMessageDirection, line: &str) {        let messages = AcpDebugMessage::parse_line(direction, line);        if messages.is_empty() {            return;        }        self.record_messages(messages);    }     fn record_messages(&self, messages: Vec<AcpDebugMessage>) {        let mut state = self            .state            .lock()            .unwrap_or_else(|poisoned| poisoned.into_inner());         state.subscribers.retain(|sender| !sender.is_closed());        for message in messages {            if state.messages.len() == MAX_DEBUG_BACKLOG_MESSAGES {                state.messages.pop_front();            }            state.messages.push_back(message.clone());             for sender in &state.subscribers {                sender.try_send(message.clone()).log_err();            }        }    }     fn trailing_stderr(&self) -> Option<String> {        let state = self.state.lock().ok()?;        let mut lines = state            .messages            .iter()            .rev()            .take_while(|message| matches!(&message.message, AcpDebugMessageContent::Stderr { .. }))            .filter_map(|message| match &message.message {                AcpDebugMessageContent::Stderr { line } if !line.is_empty() => Some(line.as_ref()),                _ => None,            })            .collect::<Vec<_>>();         if lines.is_empty() {            return None;        }         lines.reverse();        Some(lines.join("\n"))    }} fn exited_load_error_with_stderr(status: ExitStatus, debug_log: &AcpDebugLog) -> LoadError {    LoadError::Exited {        status,        stderr: debug_log.trailing_stderr().map(SharedString::from),    }} #[derive(Debug, Error)]#[error("Unsupported version")]pub struct UnsupportedVersion; /// Helper for flattening the nested `Result` shapes that come out of/// `entity.update(cx, |_, cx| fallible_op(cx))` into a single `Result<T,/// acp::Error>`.////// `anyhow::Error` values get converted via `acp::Error::from`, which/// downcasts an `acp::Error` back out of `anyhow` when present, so typed/// errors like auth-required survive the trip.trait FlattenAcpResult<T> {    fn flatten_acp(self) -> Result<T, acp::Error>;} impl<T> FlattenAcpResult<T> for Result<Result<T, anyhow::Error>, anyhow::Error> {    fn flatten_acp(self) -> Result<T, acp::Error> {        match self {            Ok(Ok(value)) => Ok(value),            Ok(Err(err)) => Err(err.into()),            Err(err) => Err(err.into()),        }    }} impl<T> FlattenAcpResult<T> for Result<Result<T, acp::Error>, anyhow::Error> {    fn flatten_acp(self) -> Result<T, acp::Error> {        match self {            Ok(Ok(value)) => Ok(value),            Ok(Err(err)) => Err(err),            Err(err) => Err(err.into()),        }    }} /// Holds state needed by foreground work dispatched from background handler closures.struct ClientContext {    sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,    session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>>,    request_elicitations: Entity<ElicitationStore>,} fn dispatch_queue_closed_error() -> acp::Error {    acp::Error::internal_error().data("ACP foreground dispatch queue closed")} /// Work items sent from `Send` handler closures to the `!Send` foreground thread.trait ForegroundWorkItem: Send {    fn run(self: Box<Self>, cx: &mut AsyncApp, ctx: &ClientContext);    fn reject(self: Box<Self>);} type ForegroundWork = Box<dyn ForegroundWorkItem>; struct RequestForegroundWork<Req, Res>where    Req: Send + 'static,    Res: JsonRpcResponse + Send + 'static,{    request: Req,    responder: Responder<Res>,    handler: fn(Req, Responder<Res>, &mut AsyncApp, &ClientContext),} impl<Req, Res> ForegroundWorkItem for RequestForegroundWork<Req, Res>where    Req: Send + 'static,    Res: JsonRpcResponse + Send + 'static,{    fn run(self: Box<Self>, cx: &mut AsyncApp, ctx: &ClientContext) {        let Self {            request,            responder,            handler,        } = *self;        handler(request, responder, cx, ctx);    }     fn reject(self: Box<Self>) {        let Self { responder, .. } = *self;        log::error!("ACP foreground dispatch queue closed while handling inbound request");        responder            .respond_with_error(dispatch_queue_closed_error())            .log_err();    }} struct NotificationForegroundWork<Notif>where    Notif: Send + 'static,{    notification: Notif,    handler: fn(Notif, &mut AsyncApp, &ClientContext),} impl<Notif> ForegroundWorkItem for NotificationForegroundWork<Notif>where    Notif: Send + 'static,{    fn run(self: Box<Self>, cx: &mut AsyncApp, ctx: &ClientContext) {        let Self {            notification,            handler,        } = *self;        handler(notification, cx, ctx);    }     fn reject(self: Box<Self>) {        log::error!("ACP foreground dispatch queue closed while handling inbound notification");    }} fn enqueue_request<Req, Res>(    dispatch_tx: &mpsc::UnboundedSender<ForegroundWork>,    request: Req,    responder: Responder<Res>,    handler: fn(Req, Responder<Res>, &mut AsyncApp, &ClientContext),) where    Req: Send + 'static,    Res: JsonRpcResponse + Send + 'static,{    let work: ForegroundWork = Box::new(RequestForegroundWork {        request,        responder,        handler,    });    if let Err(err) = dispatch_tx.unbounded_send(work) {        err.into_inner().reject();    }} fn enqueue_notification<Notif>(    dispatch_tx: &mpsc::UnboundedSender<ForegroundWork>,    notification: Notif,    handler: fn(Notif, &mut AsyncApp, &ClientContext),) where    Notif: Send + 'static,{    let work: ForegroundWork = Box::new(NotificationForegroundWork {        notification,        handler,    });    if let Err(err) = dispatch_tx.unbounded_send(work) {        err.into_inner().reject();    }} pub struct AcpConnection {    id: AgentId,    telemetry_id: SharedString,    agent_version: Option<SharedString>,    connection: ConnectionTo<Agent>,    sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,    pending_sessions: Rc<RefCell<HashMap<acp::SessionId, PendingAcpSession>>>,    auth_methods: Vec<acp::AuthMethod>,    agent_server_store: WeakEntity<AgentServerStore>,    agent_capabilities: acp::AgentCapabilities,    request_elicitations: Entity<ElicitationStore>,    defaults: AcpConnectionDefaults,    child: Option<Child>,    session_list: Option<Rc<AcpSessionList>>,    debug_log: AcpDebugLog,    _settings_subscription: Subscription,    _io_task: Task<()>,    _dispatch_task: Task<()>,    _wait_task: Task<Result<()>>,    _stderr_task: Task<Result<()>>,} #[derive(Clone, Default)]struct AcpConnectionDefaults {    mode: Rc<RefCell<Option<acp::SessionModeId>>>,    config_options: Rc<RefCell<HashMap<String, AgentConfigOptionValue>>>,} impl AcpConnectionDefaults {    fn new(        mode: Option<acp::SessionModeId>,        config_options: HashMap<String, AgentConfigOptionValue>,    ) -> Self {        Self {            mode: Rc::new(RefCell::new(mode)),            config_options: Rc::new(RefCell::new(config_options)),        }    }     fn mode(&self) -> Option<acp::SessionModeId> {        self.mode.borrow().clone()    }     fn config_option(&self, config_id: &str) -> Option<AgentConfigOptionValue> {        self.config_options.borrow().get(config_id).cloned()    }     fn set(        &self,        mode: Option<acp::SessionModeId>,        config_options: HashMap<String, AgentConfigOptionValue>,    ) {        *self.mode.borrow_mut() = mode;        *self.config_options.borrow_mut() = config_options;    }     fn refresh_from_settings(&self, agent_id: &AgentId, cx: &App) {        let Some(settings_store) = cx.try_global::<SettingsStore>() else {            self.set(None, HashMap::default());            return;        };        let settings = settings_store.get::<AllAgentServersSettings>(None);        let Some(agent_settings) = settings.get(agent_id.as_ref()) else {            self.set(None, HashMap::default());            return;        };         let default_config_options = match agent_settings {            CustomAgentServerSettings::Custom {                default_config_options,                ..            }            | CustomAgentServerSettings::Registry {                default_config_options,                ..            } => default_config_options.clone(),        };        self.set(            agent_settings.default_mode().map(acp::SessionModeId::new),            default_config_options,        );    }     fn observe_settings(&self, agent_id: AgentId, cx: &mut App) -> Subscription {        if cx.try_global::<SettingsStore>().is_none() {            return Subscription::new(|| {});        }         self.refresh_from_settings(&agent_id, cx);        let defaults = self.clone();        cx.observe_global::<SettingsStore>(move |cx| {            defaults.refresh_from_settings(&agent_id, cx);        })    }} struct PendingAcpSession {    task: Shared<Task<Result<Entity<AcpThread>, Arc<anyhow::Error>>>>,    ref_count: usize,} struct SessionConfigResponse {    modes: Option<acp::SessionModeState>,    config_options: Option<Vec<acp::SessionConfigOption>>,}

Lines 1–500 of 5127. Load more