tenant.openagents/omega

1use acp_thread::{
2    AgentConnection, AgentSessionInfo, AgentSessionList, AgentSessionListRequest,
3    AgentSessionListResponse, ElicitationStore,
4};
5use action_log::ActionLog;
6use agent_client_protocol::schema::{
7    ProtocolVersion,
8    v1::{self as acp, ErrorCode},
9};
10use agent_client_protocol::{Agent, Client, ConnectionTo, JsonRpcResponse, Lines, Responder};
11use anyhow::anyhow;
12use async_channel;
13use collections::{HashMap, HashSet};
14use feature_flags::{AcpBetaFeatureFlag, FeatureFlagAppExt as _};
15use futures::channel::mpsc;
16use futures::future::Shared;
17use futures::io::BufReader;
18use futures::{AsyncBufReadExt as _, Future, FutureExt as _, StreamExt as _};
19use project::agent_server_store::{
20    AgentServerCommand, AgentServerStore, AllAgentServersSettings, CustomAgentServerSettings,
21};
22use project::{AgentId, Project};
23use remote::remote_client::Interactive;
24use serde::Deserialize;
25use settings::{AgentConfigOptionValue, SettingsStore};
26use std::path::PathBuf;
27use std::process::{ExitStatus, Stdio};
28use std::rc::Rc;
29use std::sync::{Arc, Mutex};
30use std::{any::Any, cell::RefCell, collections::VecDeque};
31use task::{Shell, ShellBuilder, SpawnInTerminal};
32use thiserror::Error;
33use util::ResultExt as _;
34use util::path_list::PathList;
35use util::process::Child;
36
37use anyhow::{Context as _, Result};
38use gpui::{App, AppContext as _, AsyncApp, Entity, SharedString, Subscription, Task, WeakEntity};
39
40use acp_thread::{AcpThread, AuthRequired, LoadError, TerminalProviderEvent};
41use terminal::TerminalBuilder;
42use terminal::terminal_settings::{AlternateScroll, CursorShape};
43
44use crate::{CURSOR_ID, GEMINI_ID};
45
46pub const GEMINI_TERMINAL_AUTH_METHOD_ID: &str = "spawn-gemini-cli";
47const PARAMETERIZED_MODEL_PICKER_META_KEY: &str = "parameterizedModelPicker";
48const MAX_DEBUG_BACKLOG_MESSAGES: usize = 2000;
49
50#[derive(Clone, Copy, Debug, PartialEq, Eq)]
51pub enum AcpDebugMessageDirection {
52    Incoming,
53    Outgoing,
54    Stderr,
55}
56
57#[derive(Clone)]
58pub enum AcpDebugMessageContent {
59    Request {
60        id: acp::RequestId,
61        method: Arc<str>,
62        params: Option<serde_json::Value>,
63    },
64    Response {
65        id: acp::RequestId,
66        result: Result<Option<serde_json::Value>, acp::Error>,
67    },
68    Notification {
69        method: Arc<str>,
70        params: Option<serde_json::Value>,
71    },
72    Stderr {
73        line: Arc<str>,
74    },
75}
76
77#[derive(Clone)]
78pub struct AcpDebugMessage {
79    pub direction: AcpDebugMessageDirection,
80    pub message: AcpDebugMessageContent,
81}
82
83impl AcpDebugMessage {
84    fn parse_line(direction: AcpDebugMessageDirection, line: &str) -> Vec<Self> {
85        if direction == AcpDebugMessageDirection::Stderr {
86            return vec![Self {
87                direction,
88                message: AcpDebugMessageContent::Stderr {
89                    line: Arc::from(line),
90                },
91            }];
92        }
93
94        let Ok(value) = serde_json::from_str(line) else {
95            return Vec::new();
96        };
97
98        match value {
99            serde_json::Value::Array(entries) => entries
100                .into_iter()
101                .filter_map(|entry| Self::parse_value(direction, entry))
102                .collect(),
103            value => Self::parse_value(direction, value).into_iter().collect(),
104        }
105    }
106
107    fn parse_value(direction: AcpDebugMessageDirection, value: serde_json::Value) -> Option<Self> {
108        let object = value.as_object()?;
109
110        let parsed_id = object
111            .get("id")
112            .map(|raw| serde_json::from_value::<acp::RequestId>(raw.clone()));
113
114        let message = if let Some(method) = object.get("method").and_then(|method| method.as_str())
115        {
116            match parsed_id {
117                Some(Ok(id)) => AcpDebugMessageContent::Request {
118                    id,
119                    method: method.into(),
120                    params: object.get("params").cloned(),
121                },
122                Some(Err(err)) => {
123                    log::warn!("Skipping JSON-RPC message with unparsable id: {err}");
124                    return None;
125                }
126                None => AcpDebugMessageContent::Notification {
127                    method: method.into(),
128                    params: object.get("params").cloned(),
129                },
130            }
131        } else if let Some(parsed_id) = parsed_id {
132            let id = match parsed_id {
133                Ok(id) => id,
134                Err(err) => {
135                    log::warn!("Skipping JSON-RPC response with unparsable id: {err}");
136                    return None;
137                }
138            };
139
140            if let Some(error) = object.get("error") {
141                let acp_error =
142                    serde_json::from_value::<acp::Error>(error.clone()).unwrap_or_else(|err| {
143                        log::warn!("Failed to deserialize ACP error: {err}");
144                        acp::Error::internal_error().data(error.to_string())
145                    });
146
147                AcpDebugMessageContent::Response {
148                    id,
149                    result: Err(acp_error),
150                }
151            } else {
152                AcpDebugMessageContent::Response {
153                    id,
154                    result: Ok(object.get("result").cloned()),
155                }
156            }
157        } else {
158            return None;
159        };
160
161        Some(Self { direction, message })
162    }
163}
164
165#[derive(Default)]
166struct AcpDebugLogState {
167    messages: VecDeque<AcpDebugMessage>,
168    subscribers: Vec<async_channel::Sender<AcpDebugMessage>>,
169}
170
171#[derive(Clone, Default)]
172struct AcpDebugLog {
173    state: Arc<Mutex<AcpDebugLogState>>,
174}
175
176impl AcpDebugLog {
177    fn subscribe(
178        &self,
179    ) -> (
180        Vec<AcpDebugMessage>,
181        async_channel::Receiver<AcpDebugMessage>,
182    ) {
183        let mut state = self
184            .state
185            .lock()
186            .unwrap_or_else(|poisoned| poisoned.into_inner());
187        let backlog = state.messages.iter().cloned().collect();
188        let (sender, receiver) = async_channel::unbounded();
189        state.subscribers.push(sender);
190        (backlog, receiver)
191    }
192
193    fn record_line(&self, direction: AcpDebugMessageDirection, line: &str) {
194        let messages = AcpDebugMessage::parse_line(direction, line);
195        if messages.is_empty() {
196            return;
197        }
198        self.record_messages(messages);
199    }
200
201    fn record_messages(&self, messages: Vec<AcpDebugMessage>) {
202        let mut state = self
203            .state
204            .lock()
205            .unwrap_or_else(|poisoned| poisoned.into_inner());
206
207        state.subscribers.retain(|sender| !sender.is_closed());
208        for message in messages {
209            if state.messages.len() == MAX_DEBUG_BACKLOG_MESSAGES {
210                state.messages.pop_front();
211            }
212            state.messages.push_back(message.clone());
213
214            for sender in &state.subscribers {
215                sender.try_send(message.clone()).log_err();
216            }
217        }
218    }
219
220    fn trailing_stderr(&self) -> Option<String> {
221        let state = self.state.lock().ok()?;
222        let mut lines = state
223            .messages
224            .iter()
225            .rev()
226            .take_while(|message| matches!(&message.message, AcpDebugMessageContent::Stderr { .. }))
227            .filter_map(|message| match &message.message {
228                AcpDebugMessageContent::Stderr { line } if !line.is_empty() => Some(line.as_ref()),
229                _ => None,
230            })
231            .collect::<Vec<_>>();
232
233        if lines.is_empty() {
234            return None;
235        }
236
237        lines.reverse();
238        Some(lines.join("\n"))
239    }
240}
241
242fn exited_load_error_with_stderr(status: ExitStatus, debug_log: &AcpDebugLog) -> LoadError {
243    LoadError::Exited {
244        status,
245        stderr: debug_log.trailing_stderr().map(SharedString::from),
246    }
247}
248
249#[derive(Debug, Error)]
250#[error("Unsupported version")]
251pub struct UnsupportedVersion;
252
253/// Helper for flattening the nested `Result` shapes that come out of
254/// `entity.update(cx, |_, cx| fallible_op(cx))` into a single `Result<T,
255/// acp::Error>`.
256///
257/// `anyhow::Error` values get converted via `acp::Error::from`, which
258/// downcasts an `acp::Error` back out of `anyhow` when present, so typed
259/// errors like auth-required survive the trip.
260trait FlattenAcpResult<T> {
261    fn flatten_acp(self) -> Result<T, acp::Error>;
262}
263
264impl<T> FlattenAcpResult<T> for Result<Result<T, anyhow::Error>, anyhow::Error> {
265    fn flatten_acp(self) -> Result<T, acp::Error> {
266        match self {
267            Ok(Ok(value)) => Ok(value),
268            Ok(Err(err)) => Err(err.into()),
269            Err(err) => Err(err.into()),
270        }
271    }
272}
273
274impl<T> FlattenAcpResult<T> for Result<Result<T, acp::Error>, anyhow::Error> {
275    fn flatten_acp(self) -> Result<T, acp::Error> {
276        match self {
277            Ok(Ok(value)) => Ok(value),
278            Ok(Err(err)) => Err(err),
279            Err(err) => Err(err.into()),
280        }
281    }
282}
283
284/// Holds state needed by foreground work dispatched from background handler closures.
285struct ClientContext {
286    sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
287    session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>>,
288    request_elicitations: Entity<ElicitationStore>,
289}
290
291fn dispatch_queue_closed_error() -> acp::Error {
292    acp::Error::internal_error().data("ACP foreground dispatch queue closed")
293}
294
295/// Work items sent from `Send` handler closures to the `!Send` foreground thread.
296trait ForegroundWorkItem: Send {
297    fn run(self: Box<Self>, cx: &mut AsyncApp, ctx: &ClientContext);
298    fn reject(self: Box<Self>);
299}
300
301type ForegroundWork = Box<dyn ForegroundWorkItem>;
302
303struct RequestForegroundWork<Req, Res>
304where
305    Req: Send + 'static,
306    Res: JsonRpcResponse + Send + 'static,
307{
308    request: Req,
309    responder: Responder<Res>,
310    handler: fn(Req, Responder<Res>, &mut AsyncApp, &ClientContext),
311}
312
313impl<Req, Res> ForegroundWorkItem for RequestForegroundWork<Req, Res>
314where
315    Req: Send + 'static,
316    Res: JsonRpcResponse + Send + 'static,
317{
318    fn run(self: Box<Self>, cx: &mut AsyncApp, ctx: &ClientContext) {
319        let Self {
320            request,
321            responder,
322            handler,
323        } = *self;
324        handler(request, responder, cx, ctx);
325    }
326
327    fn reject(self: Box<Self>) {
328        let Self { responder, .. } = *self;
329        log::error!("ACP foreground dispatch queue closed while handling inbound request");
330        responder
331            .respond_with_error(dispatch_queue_closed_error())
332            .log_err();
333    }
334}
335
336struct NotificationForegroundWork<Notif>
337where
338    Notif: Send + 'static,
339{
340    notification: Notif,
341    handler: fn(Notif, &mut AsyncApp, &ClientContext),
342}
343
344impl<Notif> ForegroundWorkItem for NotificationForegroundWork<Notif>
345where
346    Notif: Send + 'static,
347{
348    fn run(self: Box<Self>, cx: &mut AsyncApp, ctx: &ClientContext) {
349        let Self {
350            notification,
351            handler,
352        } = *self;
353        handler(notification, cx, ctx);
354    }
355
356    fn reject(self: Box<Self>) {
357        log::error!("ACP foreground dispatch queue closed while handling inbound notification");
358    }
359}
360
361fn enqueue_request<Req, Res>(
362    dispatch_tx: &mpsc::UnboundedSender<ForegroundWork>,
363    request: Req,
364    responder: Responder<Res>,
365    handler: fn(Req, Responder<Res>, &mut AsyncApp, &ClientContext),
366) where
367    Req: Send + 'static,
368    Res: JsonRpcResponse + Send + 'static,
369{
370    let work: ForegroundWork = Box::new(RequestForegroundWork {
371        request,
372        responder,
373        handler,
374    });
375    if let Err(err) = dispatch_tx.unbounded_send(work) {
376        err.into_inner().reject();
377    }
378}
379
380fn enqueue_notification<Notif>(
381    dispatch_tx: &mpsc::UnboundedSender<ForegroundWork>,
382    notification: Notif,
383    handler: fn(Notif, &mut AsyncApp, &ClientContext),
384) where
385    Notif: Send + 'static,
386{
387    let work: ForegroundWork = Box::new(NotificationForegroundWork {
388        notification,
389        handler,
390    });
391    if let Err(err) = dispatch_tx.unbounded_send(work) {
392        err.into_inner().reject();
393    }
394}
395
396pub struct AcpConnection {
397    id: AgentId,
398    telemetry_id: SharedString,
399    agent_version: Option<SharedString>,
400    connection: ConnectionTo<Agent>,
401    sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
402    pending_sessions: Rc<RefCell<HashMap<acp::SessionId, PendingAcpSession>>>,
403    auth_methods: Vec<acp::AuthMethod>,
404    agent_server_store: WeakEntity<AgentServerStore>,
405    agent_capabilities: acp::AgentCapabilities,
406    request_elicitations: Entity<ElicitationStore>,
407    defaults: AcpConnectionDefaults,
408    child: Option<Child>,
409    session_list: Option<Rc<AcpSessionList>>,
410    debug_log: AcpDebugLog,
411    _settings_subscription: Subscription,
412    _io_task: Task<()>,
413    _dispatch_task: Task<()>,
414    _wait_task: Task<Result<()>>,
415    _stderr_task: Task<Result<()>>,
416}
417
418#[derive(Clone, Default)]
419struct AcpConnectionDefaults {
420    mode: Rc<RefCell<Option<acp::SessionModeId>>>,
421    config_options: Rc<RefCell<HashMap<String, AgentConfigOptionValue>>>,
422}
423
424impl AcpConnectionDefaults {
425    fn new(
426        mode: Option<acp::SessionModeId>,
427        config_options: HashMap<String, AgentConfigOptionValue>,
428    ) -> Self {
429        Self {
430            mode: Rc::new(RefCell::new(mode)),
431            config_options: Rc::new(RefCell::new(config_options)),
432        }
433    }
434
435    fn mode(&self) -> Option<acp::SessionModeId> {
436        self.mode.borrow().clone()
437    }
438
439    fn config_option(&self, config_id: &str) -> Option<AgentConfigOptionValue> {
440        self.config_options.borrow().get(config_id).cloned()
441    }
442
443    fn set(
444        &self,
445        mode: Option<acp::SessionModeId>,
446        config_options: HashMap<String, AgentConfigOptionValue>,
447    ) {
448        *self.mode.borrow_mut() = mode;
449        *self.config_options.borrow_mut() = config_options;
450    }
451
452    fn refresh_from_settings(&self, agent_id: &AgentId, cx: &App) {
453        let Some(settings_store) = cx.try_global::<SettingsStore>() else {
454            self.set(None, HashMap::default());
455            return;
456        };
457        let settings = settings_store.get::<AllAgentServersSettings>(None);
458        let Some(agent_settings) = settings.get(agent_id.as_ref()) else {
459            self.set(None, HashMap::default());
460            return;
461        };
462
463        let default_config_options = match agent_settings {
464            CustomAgentServerSettings::Custom {
465                default_config_options,
466                ..
467            }
468            | CustomAgentServerSettings::Registry {
469                default_config_options,
470                ..
471            } => default_config_options.clone(),
472        };
473        self.set(
474            agent_settings.default_mode().map(acp::SessionModeId::new),
475            default_config_options,
476        );
477    }
478
479    fn observe_settings(&self, agent_id: AgentId, cx: &mut App) -> Subscription {
480        if cx.try_global::<SettingsStore>().is_none() {
481            return Subscription::new(|| {});
482        }
483
484        self.refresh_from_settings(&agent_id, cx);
485        let defaults = self.clone();
486        cx.observe_global::<SettingsStore>(move |cx| {
487            defaults.refresh_from_settings(&agent_id, cx);
488        })
489    }
490}
491
492struct PendingAcpSession {
493    task: Shared<Task<Result<Entity<AcpThread>, Arc<anyhow::Error>>>>,
494    ref_count: usize,
495}
496
497struct SessionConfigResponse {
498    modes: Option<acp::SessionModeState>,
499    config_options: Option<Vec<acp::SessionConfigOption>>,
500}
501
502#[derive(Clone)]
503struct ConfigOptions {
504    config_options: Rc<RefCell<Vec<acp::SessionConfigOption>>>,
505    tx: Rc<RefCell<watch::Sender<()>>>,
506    rx: watch::Receiver<()>,
507}
508
509impl ConfigOptions {
510    fn new(config_options: Rc<RefCell<Vec<acp::SessionConfigOption>>>) -> Self {
511        let (tx, rx) = watch::channel(());
512        Self {
513            config_options,
514            tx: Rc::new(RefCell::new(tx)),
515            rx,
516        }
517    }
518}
519
520pub struct AcpSession {
521    thread: WeakEntity<AcpThread>,
522    suppress_abort_err: bool,
523    session_modes: Option<Rc<RefCell<acp::SessionModeState>>>,
524    config_options: Option<ConfigOptions>,
525    ref_count: usize,
526}
527
528pub struct AcpSessionList {
529    connection: ConnectionTo<Agent>,
530    supports_delete: bool,
531    updates_tx: async_channel::Sender<acp_thread::SessionListUpdate>,
532    updates_rx: async_channel::Receiver<acp_thread::SessionListUpdate>,
533}
534
535impl AcpSessionList {
536    fn new(connection: ConnectionTo<Agent>, supports_delete: bool) -> Self {
537        let (tx, rx) = async_channel::unbounded();
538        Self {
539            connection,
540            supports_delete,
541            updates_tx: tx,
542            updates_rx: rx,
543        }
544    }
545
546    fn notify_update(&self) {
547        self.updates_tx
548            .try_send(acp_thread::SessionListUpdate::Refresh)
549            .log_err();
550    }
551
552    fn send_info_update(&self, session_id: acp::SessionId, update: acp::SessionInfoUpdate) {
553        self.updates_tx
554            .try_send(acp_thread::SessionListUpdate::SessionInfo { session_id, update })
555            .log_err();
556    }
557}
558
559impl AgentSessionList for AcpSessionList {
560    fn list_sessions(
561        &self,
562        request: AgentSessionListRequest,
563        cx: &mut App,
564    ) -> Task<Result<AgentSessionListResponse>> {
565        let conn = self.connection.clone();
566        cx.foreground_executor().spawn(async move {
567            let acp_request = acp::ListSessionsRequest::new()
568                .cwd(request.cwd)
569                .cursor(request.cursor);
570            let response = conn
571                .send_request(acp_request)
572                .block_task()
573                .await
574                .map_err(map_acp_error)?;
575            Ok(AgentSessionListResponse {
576                sessions: response
577                    .sessions
578                    .into_iter()
579                    .map(|s| AgentSessionInfo {
580                        session_id: s.session_id,
581                        work_dirs: Some(work_dirs_from_session_info(
582                            s.cwd,
583                            s.additional_directories,
584                        )),
585                        title: s.title.map(Into::into),
586                        updated_at: s.updated_at.and_then(|date_str| {
587                            chrono::DateTime::parse_from_rfc3339(&date_str)
588                                .ok()
589                                .map(|dt| dt.with_timezone(&chrono::Utc))
590                        }),
591                        created_at: None,
592                        meta: s.meta,
593                    })
594                    .collect(),
595                next_cursor: response.next_cursor,
596                meta: response.meta,
597            })
598        })
599    }
600
601    fn supports_delete(&self) -> bool {
602        self.supports_delete
603    }
604
605    fn delete_session(&self, session_id: &acp::SessionId, cx: &mut App) -> Task<Result<()>> {
606        if !self.supports_delete() {
607            return Task::ready(Err(anyhow::anyhow!("delete_session not supported")));
608        }
609
610        let conn = self.connection.clone();
611        let updates_tx = self.updates_tx.clone();
612        let session_id = session_id.clone();
613        cx.foreground_executor().spawn(async move {
614            conn.send_request(acp::DeleteSessionRequest::new(session_id))
615                .block_task()
616                .await
617                .map_err(map_acp_error)?;
618            updates_tx
619                .try_send(acp_thread::SessionListUpdate::Refresh)
620                .log_err();
621            Ok(())
622        })
623    }
624
625    fn watch(
626        &self,
627        _cx: &mut App,
628    ) -> Option<async_channel::Receiver<acp_thread::SessionListUpdate>> {
629        Some(self.updates_rx.clone())
630    }
631
632    fn notify_refresh(&self) {
633        self.notify_update();
634    }
635
636    fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
637        self
638    }
639}
640
641pub async fn connect(
642    agent_id: AgentId,
643    project: Entity<Project>,
644    command: AgentServerCommand,
645    agent_server_store: WeakEntity<AgentServerStore>,
646    default_mode: Option<acp::SessionModeId>,
647    default_config_options: HashMap<String, AgentConfigOptionValue>,
648    cx: &mut AsyncApp,
649) -> Result<Rc<dyn AgentConnection>> {
650    let conn = AcpConnection::stdio(
651        agent_id,
652        project,
653        command.clone(),
654        agent_server_store,
655        default_mode,
656        default_config_options,
657        cx,
658    )
659    .await?;
660    Ok(Rc::new(conn) as _)
661}
662
663const MINIMUM_SUPPORTED_VERSION: ProtocolVersion = ProtocolVersion::V1;
664
665/// Build a `Client` connection over `transport` with Zed's full
666/// agent→client handler set wired up.
667///
668/// All incoming requests and notifications are forwarded to the foreground
669/// dispatch queue via `dispatch_tx`, where they are handled by the
670/// `handle_*` functions on a GPUI context. The returned future drives the
671/// connection and completes when the transport closes; callers are expected
672/// to spawn it on a background executor and hold the task for the lifetime
673/// of the connection. The `connection_tx` oneshot receives the
674/// `ConnectionTo<Agent>` handle as soon as the builder runs its `main_fn`.
675fn connect_client_future(
676    name: &'static str,
677    transport: impl agent_client_protocol::ConnectTo<Client> + 'static,
678    dispatch_tx: mpsc::UnboundedSender<ForegroundWork>,
679    connection_tx: futures::channel::oneshot::Sender<ConnectionTo<Agent>>,
680) -> impl Future<Output = Result<(), acp::Error>> {
681    // Each handler forwards its inputs onto the foreground dispatch queue.
682    // The SDK requires the closure to be `Send`, so we move a clone of
683    // `dispatch_tx` into each one.
684    macro_rules! on_request {
685        ($handler:ident) => {{
686            let dispatch_tx = dispatch_tx.clone();
687            async move |req, responder, _connection| {
688                enqueue_request(&dispatch_tx, req, responder, $handler);
689                Ok(())
690            }
691        }};
692    }
693    macro_rules! on_notification {
694        ($handler:ident) => {{
695            let dispatch_tx = dispatch_tx.clone();
696            async move |notif, _connection| {
697                enqueue_notification(&dispatch_tx, notif, $handler);
698                Ok(())
699            }
700        }};
701    }
702
703    Client
704        .builder()
705        .name(name)
706        // --- Request handlers (agent→client) ---
707        .on_receive_request(
708            on_request!(handle_request_permission),
709            agent_client_protocol::on_receive_request!(),
710        )
711        .on_receive_request(
712            on_request!(handle_write_text_file),
713            agent_client_protocol::on_receive_request!(),
714        )
715        .on_receive_request(
716            on_request!(handle_read_text_file),
717            agent_client_protocol::on_receive_request!(),
718        )
719        .on_receive_request(
720            on_request!(handle_create_terminal),
721            agent_client_protocol::on_receive_request!(),
722        )
723        .on_receive_request(
724            on_request!(handle_kill_terminal),
725            agent_client_protocol::on_receive_request!(),
726        )
727        .on_receive_request(
728            on_request!(handle_release_terminal),
729            agent_client_protocol::on_receive_request!(),
730        )
731        .on_receive_request(
732            on_request!(handle_terminal_output),
733            agent_client_protocol::on_receive_request!(),
734        )
735        .on_receive_request(
736            on_request!(handle_wait_for_terminal_exit),
737            agent_client_protocol::on_receive_request!(),
738        )
739        .on_receive_request(
740            on_request!(handle_create_elicitation),
741            agent_client_protocol::on_receive_request!(),
742        )
743        // --- Notification handlers (agent→client) ---
744        .on_receive_notification(
745            on_notification!(handle_session_notification),
746            agent_client_protocol::on_receive_notification!(),
747        )
748        .on_receive_notification(
749            on_notification!(handle_complete_elicitation),
750            agent_client_protocol::on_receive_notification!(),
751        )
752        .connect_with(
753            transport,
754            move |connection: ConnectionTo<Agent>| async move {
755                if connection_tx.send(connection).is_err() {
756                    log::error!("failed to send ACP connection handle — receiver was dropped");
757                }
758                // Keep the connection alive until the transport closes.
759                futures::future::pending::<Result<(), acp::Error>>().await
760            },
761        )
762}
763
764fn client_capabilities_for_agent(agent_id: &AgentId) -> acp::ClientCapabilities {
765    let mut meta = acp::Meta::from_iter([
766        ("terminal_output".into(), true.into()),
767        ("terminal-auth".into(), true.into()),
768    ]);
769
770    if agent_id.as_ref() == CURSOR_ID {
771        meta.insert(PARAMETERIZED_MODEL_PICKER_META_KEY.into(), true.into());
772    }
773
774    acp::ClientCapabilities::new()
775        .fs(acp::FileSystemCapabilities::new()
776            .read_text_file(true)
777            .write_text_file(true))
778        .terminal(true)
779        .auth(acp::AuthCapabilities::new().terminal(true))
780        .session(
781            acp::ClientSessionCapabilities::new().config_options(
782                acp::SessionConfigOptionsCapabilities::new()
783                    .boolean(acp::BooleanConfigOptionCapabilities::new()),
784            ),
785        )
786        .elicitation(
787            acp::ElicitationCapabilities::new()
788                .form(acp::ElicitationFormCapabilities::new())
789                .url(acp::ElicitationUrlCapabilities::new()),
790        )
791        .meta(meta)
792}
793
794impl AcpConnection {
795    pub fn subscribe_debug_messages(
796        &self,
797    ) -> (
798        Vec<AcpDebugMessage>,
799        async_channel::Receiver<AcpDebugMessage>,
800    ) {
801        self.debug_log.subscribe()
802    }
803
804    pub async fn stdio(
805        agent_id: AgentId,
806        project: Entity<Project>,
807        command: AgentServerCommand,
808        agent_server_store: WeakEntity<AgentServerStore>,
809        default_mode: Option<acp::SessionModeId>,
810        default_config_options: HashMap<String, AgentConfigOptionValue>,
811        cx: &mut AsyncApp,
812    ) -> Result<Self> {
813        let root_dir = project.read_with(cx, |project, cx| {
814            project
815                .default_path_list(cx)
816                .ordered_paths()
817                .next()
818                .cloned()
819        });
820        let original_command = command.clone();
821        let (path, args, env) = project
822            .read_with(cx, |project, cx| {
823                project.remote_client().and_then(|client| {
824                    let template = client
825                        .read(cx)
826                        .build_command(
827                            Some(command.path.display().to_string()),
828                            &command.args,
829                            &command.env.clone().into_iter().flatten().collect(),
830                            root_dir.as_ref().map(|path| path.display().to_string()),
831                            None,
832                            Interactive::No,
833                        )
834                        .log_err()?;
835                    Some((template.program, template.args, template.env))
836                })
837            })
838            .unwrap_or_else(|| {
839                (
840                    command.path.display().to_string(),
841                    command.args,
842                    command.env.unwrap_or_default(),
843                )
844            });
845
846        let builder = ShellBuilder::new(&Shell::System, cfg!(windows)).non_interactive();
847        let mut child = builder.build_std_command(Some(path.clone()), &args);
848        child.envs(env.clone());
849        if let Some(cwd) = project.read_with(cx, |project, _cx| {
850            if project.is_local() {
851                root_dir.as_ref()
852            } else {
853                None
854            }
855        }) {
856            child.current_dir(cwd);
857        }
858        let mut child = Child::spawn(child, Stdio::piped(), Stdio::piped(), Stdio::piped())?;
859
860        let stdout = child.stdout.take().context("Failed to take stdout")?;
861        let stdin = child.stdin.take().context("Failed to take stdin")?;
862        let stderr = child.stderr.take().context("Failed to take stderr")?;
863        log::debug!("Spawning external agent server: {:?}, {:?}", path, args);
864        log::trace!("Spawned (pid: {})", child.id());
865
866        let sessions = Rc::new(RefCell::new(HashMap::default()));
867        let debug_log = AcpDebugLog::default();
868
869        let (release_channel, version): (Option<&str>, String) = cx.update(|cx| {
870            (
871                release_channel::ReleaseChannel::try_global(cx)
872                    .map(|release_channel| release_channel.display_name()),
873                release_channel::AppVersion::global(cx).to_string(),
874            )
875        });
876
877        let client_session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>> =
878            Rc::new(RefCell::new(None));
879        let request_elicitations = cx.new(|_| ElicitationStore::default());
880
881        // Set up the foreground dispatch channel for bridging Send handler
882        // closures to the !Send foreground thread.
883        let (dispatch_tx, dispatch_rx) = mpsc::unbounded::<ForegroundWork>();
884
885        let incoming_lines = futures::io::BufReader::new(stdout).lines();
886        let tapped_incoming = incoming_lines.inspect({
887            let debug_log = debug_log.clone();
888            move |result| match result {
889                Ok(line) => debug_log.record_line(AcpDebugMessageDirection::Incoming, line),
890                Err(err) => {
891                    log::warn!("ACP transport read error: {err}");
892                }
893            }
894        });
895
896        let tapped_outgoing = futures::sink::unfold(
897            (Box::pin(stdin), debug_log.clone()),
898            async move |(mut writer, debug_log), line: String| {
899                use futures::AsyncWriteExt;
900                debug_log.record_line(AcpDebugMessageDirection::Outgoing, &line);
901                let mut bytes = line.into_bytes();
902                bytes.push(b'\n');
903                writer.write_all(&bytes).await?;
904                Ok::<_, std::io::Error>((writer, debug_log))
905            },
906        );
907
908        let transport = Lines::new(tapped_outgoing, tapped_incoming);
909
910        let stderr_task = cx.background_spawn({
911            let debug_log = debug_log.clone();
912            async move {
913                let mut stderr = BufReader::new(stderr);
914                let mut line = String::new();
915                while let Ok(n) = stderr.read_line(&mut line).await
916                    && n > 0
917                {
918                    let trimmed = line.trim_end_matches(['\n', '\r']);
919                    log::warn!("agent stderr: {trimmed}");
920                    debug_log.record_line(AcpDebugMessageDirection::Stderr, trimmed);
921                    line.clear();
922                }
923                Ok(())
924            }
925        });
926
927        // `connect_client_future` installs the production handler set and
928        // hands us back both the connection-future (to run on a background
929        // executor) and a oneshot receiver that produces the
930        // `ConnectionTo<Agent>` once the transport handshake is ready.
931        let (connection_tx, connection_rx) = futures::channel::oneshot::channel();
932        let connection_future =
933            connect_client_future("zed", transport, dispatch_tx.clone(), connection_tx);
934        let io_task = cx.background_spawn(async move {
935            if let Err(err) = connection_future.await {
936                log::error!("ACP connection error: {err}");
937            }
938        });
939
940        let connection_rx = async move {
941            connection_rx
942                .await
943                .context("Failed to receive ACP connection handle")
944        }
945        .boxed_local();
946        let status_fut = child
947            .status()
948            .map({
949                let debug_log = debug_log.clone();
950                move |status| match status {
951                    Ok(status) => Ok(exited_load_error_with_stderr(status, &debug_log)),
952                    Err(err) => Err(anyhow!("failed to wait for agent server exit: {err}")),
953                }
954            })
955            .boxed_local();
956        let (connection, status_fut) = match futures::future::select(connection_rx, status_fut)
957            .await
958        {
959            futures::future::Either::Left((connection, status_fut)) => (connection?, status_fut),
960            futures::future::Either::Right((load_error, _connection_rx)) => {
961                return Err(load_error?.into());
962            }
963        };
964
965        // Set up the foreground dispatch loop to process work items from handlers.
966        let dispatch_context = ClientContext {
967            sessions: sessions.clone(),
968            session_list: client_session_list.clone(),
969            request_elicitations: request_elicitations.clone(),
970        };
971        let dispatch_task = cx.spawn({
972            let mut dispatch_rx = dispatch_rx;
973            async move |cx| {
974                while let Some(work) = dispatch_rx.next().await {
975                    work.run(cx, &dispatch_context);
976                }
977            }
978        });
979
980        let initialize_response = connection
981            .send_request(
982                acp::InitializeRequest::new(ProtocolVersion::V1)
983                    .client_capabilities(client_capabilities_for_agent(&agent_id))
984                    .client_info(
985                        acp::Implementation::new("zed", version)
986                            .title(release_channel.map(ToOwned::to_owned)),
987                    ),
988            )
989            .block_task()
990            .boxed_local();
991        let (response, status_fut) =
992            match futures::future::select(initialize_response, status_fut).await {
993                futures::future::Either::Left((Ok(response), status_fut)) => (response, status_fut),
994                futures::future::Either::Left((Err(error), status_fut)) => {
995                    let timer = cx
996                        .background_executor()
997                        .timer(std::time::Duration::from_millis(250))
998                        .boxed_local();
999                    if let futures::future::Either::Left((load_error, _timer)) =
1000                        futures::future::select(status_fut, timer).await
1001                    {
1002                        return Err(load_error?.into());
1003                    }
1004
1005                    return Err(error.into());
1006                }
1007                futures::future::Either::Right((load_error, _initialize_response)) => {
1008                    return Err(load_error?.into());
1009                }
1010            };
1011
1012        if response.protocol_version < MINIMUM_SUPPORTED_VERSION {
1013            return Err(UnsupportedVersion.into());
1014        }
1015
1016        let wait_task = cx.spawn({
1017            let sessions = sessions.clone();
1018            async move |cx| {
1019                let load_error = status_fut.await?;
1020                emit_load_error_to_all_sessions(&sessions, load_error, cx);
1021                anyhow::Ok(())
1022            }
1023        });
1024
1025        let agent_info = response.agent_info;
1026        let telemetry_id = agent_info
1027            .as_ref()
1028            // Use the one the agent provides if we have one
1029            .map(|info| SharedString::from(info.name.clone()))
1030            // Otherwise, just use the name
1031            .unwrap_or_else(|| agent_id.0.clone());
1032        let agent_version = agent_info
1033            .and_then(|info| (!info.version.is_empty()).then(|| SharedString::from(info.version)));
1034        let agent_supports_delete = response
1035            .agent_capabilities
1036            .session_capabilities
1037            .delete
1038            .is_some();
1039
1040        let session_list = if response
1041            .agent_capabilities
1042            .session_capabilities
1043            .list
1044            .is_some()
1045        {
1046            let list = Rc::new(AcpSessionList::new(
1047                connection.clone(),
1048                agent_supports_delete,
1049            ));
1050            *client_session_list.borrow_mut() = Some(list.clone());
1051            Some(list)
1052        } else {
1053            None
1054        };
1055
1056        // TODO: Remove this override once Google team releases their official auth methods
1057        let auth_methods = if agent_id.0.as_ref() == GEMINI_ID {
1058            let mut gemini_args = original_command.args.clone();
1059            gemini_args.retain(|a| a != "--experimental-acp" && a != "--acp");
1060            let value = serde_json::json!({
1061                "label": "gemini /auth",
1062                "command": original_command.path.to_string_lossy(),
1063                "args": gemini_args,
1064                "env": original_command.env.unwrap_or_default(),
1065            });
1066            let meta = acp::Meta::from_iter([("terminal-auth".to_string(), value)]);
1067            vec![acp::AuthMethod::Agent(
1068                acp::AuthMethodAgent::new(GEMINI_TERMINAL_AUTH_METHOD_ID, "Login")
1069                    .description("Login with your Google or Vertex AI account")
1070                    .meta(meta),
1071            )]
1072        } else {
1073            response.auth_methods
1074        };
1075        let defaults = AcpConnectionDefaults::new(default_mode, default_config_options);
1076        let settings_subscription = cx.update({
1077            let agent_id = agent_id.clone();
1078            let defaults = defaults.clone();
1079            move |cx| defaults.observe_settings(agent_id, cx)
1080        });
1081
1082        Ok(Self {
1083            id: agent_id,
1084            auth_methods,
1085            agent_server_store,
1086            connection,
1087            telemetry_id,
1088            agent_version,
1089            sessions,
1090            pending_sessions: Rc::new(RefCell::new(HashMap::default())),
1091            agent_capabilities: response.agent_capabilities,
1092            request_elicitations,
1093            defaults,
1094            session_list,
1095            debug_log,
1096            _settings_subscription: settings_subscription,
1097            _io_task: io_task,
1098            _dispatch_task: dispatch_task,
1099            _wait_task: wait_task,
1100            _stderr_task: stderr_task,
1101            child: Some(child),
1102        })
1103    }
1104
1105    pub fn prompt_capabilities(&self) -> &acp::PromptCapabilities {
1106        &self.agent_capabilities.prompt_capabilities
1107    }
1108
1109    #[cfg(any(test, feature = "test-support"))]
1110    fn new_for_test(
1111        connection: ConnectionTo<Agent>,
1112        sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
1113        agent_capabilities: acp::AgentCapabilities,
1114        request_elicitations: Entity<ElicitationStore>,
1115        agent_server_store: WeakEntity<AgentServerStore>,
1116        io_task: Task<()>,
1117        dispatch_task: Task<()>,
1118        cx: &mut App,
1119    ) -> Self {
1120        let agent_id = AgentId::new("test");
1121        let defaults = AcpConnectionDefaults::default();
1122        let settings_subscription = defaults.observe_settings(agent_id.clone(), cx);
1123
1124        Self {
1125            id: agent_id,
1126            telemetry_id: "test".into(),
1127            agent_version: None,
1128            connection,
1129            sessions,
1130            pending_sessions: Rc::new(RefCell::new(HashMap::default())),
1131            auth_methods: vec![],
1132            agent_server_store,
1133            agent_capabilities,
1134            request_elicitations,
1135            defaults,
1136            child: None,
1137            session_list: None,
1138            debug_log: AcpDebugLog::default(),
1139            _settings_subscription: settings_subscription,
1140            _io_task: io_task,
1141            _dispatch_task: dispatch_task,
1142            _wait_task: Task::ready(Ok(())),
1143            _stderr_task: Task::ready(Ok(())),
1144        }
1145    }
1146
1147    fn session_directories_from_work_dirs(
1148        &self,
1149        work_dirs: &PathList,
1150    ) -> Result<SessionDirectories> {
1151        let supports_additional_directories = self.supports_session_additional_directories();
1152        session_directories_from_work_dirs(work_dirs, supports_additional_directories)
1153    }
1154
1155    fn open_or_create_session(
1156        self: Rc<Self>,
1157        session_id: acp::SessionId,
1158        project: Entity<Project>,
1159        work_dirs: PathList,
1160        title: Option<SharedString>,
1161        rpc_call: impl FnOnce(
1162            ConnectionTo<Agent>,
1163            acp::SessionId,
1164            SessionDirectories,
1165        )
1166            -> futures::future::LocalBoxFuture<'static, Result<SessionConfigResponse>>
1167        + 'static,
1168        cx: &mut App,
1169    ) -> Task<Result<Entity<AcpThread>>> {
1170        // Check `pending_sessions` before `sessions` because the session is now
1171        // inserted into `sessions` before the load RPC completes (so that
1172        // notifications dispatched during history replay can find the thread).
1173        // Concurrent loads should still wait for the in-flight task so that
1174        // ref-counting happens in one place and the caller sees a fully loaded
1175        // session.
1176        if let Some(pending) = self.pending_sessions.borrow_mut().get_mut(&session_id) {
1177            pending.ref_count += 1;
1178            let task = pending.task.clone();
1179            return cx
1180                .foreground_executor()
1181                .spawn(async move { task.await.map_err(|err| anyhow!(err)) });
1182        }
1183
1184        if let Some(session) = self.sessions.borrow_mut().get_mut(&session_id) {
1185            session.ref_count += 1;
1186            if let Some(thread) = session.thread.upgrade() {
1187                return Task::ready(Ok(thread));
1188            }
1189        }
1190
1191        let directories = match self.session_directories_from_work_dirs(&work_dirs) {
1192            Ok(directories) => directories,
1193            Err(error) => return Task::ready(Err(error)),
1194        };
1195
1196        let shared_task = cx
1197            .spawn({
1198                let session_id = session_id.clone();
1199                let this = self.clone();
1200                async move |cx| {
1201                    let action_log = cx.new(|_| ActionLog::new(project.clone()));
1202                    let thread: Entity<AcpThread> = cx.new(|cx| {
1203                        AcpThread::new(
1204                            None,
1205                            title,
1206                            Some(work_dirs),
1207                            this.clone(),
1208                            project,
1209                            action_log,
1210                            session_id.clone(),
1211                            watch::Receiver::constant(
1212                                this.agent_capabilities.prompt_capabilities.clone(),
1213                            ),
1214                            cx,
1215                        )
1216                    });
1217
1218                    // Register the session before awaiting the RPC so that any
1219                    // `session/update` notifications that arrive during the call
1220                    // (e.g. history replay during `session/load`) can find the thread.
1221                    // Modes/config are filled in once the response arrives.
1222                    this.sessions.borrow_mut().insert(
1223                        session_id.clone(),
1224                        AcpSession {
1225                            thread: thread.downgrade(),
1226                            suppress_abort_err: false,
1227                            session_modes: None,
1228                            config_options: None,
1229                            ref_count: 1,
1230                        },
1231                    );
1232
1233                    let response =
1234                        match rpc_call(this.connection.clone(), session_id.clone(), directories)
1235                            .await
1236                        {
1237                            Ok(response) => response,
1238                            Err(err) => {
1239                                this.sessions.borrow_mut().remove(&session_id);
1240                                this.pending_sessions.borrow_mut().remove(&session_id);
1241                                return Err(Arc::new(err));
1242                            }
1243                        };
1244
1245                    let (modes, config_options) =
1246                        config_state(response.modes, response.config_options);
1247
1248                    if let Some(config_opts) = config_options.as_ref() {
1249                        this.apply_default_config_options(&session_id, config_opts, cx);
1250                    }
1251
1252                    let ref_count = this
1253                        .pending_sessions
1254                        .borrow_mut()
1255                        .remove(&session_id)
1256                        .map_or(1, |pending| pending.ref_count);
1257
1258                    // If `close_session` ran to completion while the load RPC was in
1259                    // flight, it will have removed both the pending entry and the
1260                    // sessions entry (and dispatched the ACP close RPC). In that case
1261                    // the thread has no live session to attach to, so fail the load
1262                    // instead of handing back an orphaned thread.
1263                    {
1264                        let mut sessions = this.sessions.borrow_mut();
1265                        let Some(session) = sessions.get_mut(&session_id) else {
1266                            return Err(Arc::new(anyhow!(
1267                                "session was closed before load completed"
1268                            )));
1269                        };
1270                        session.session_modes = modes;
1271                        session.config_options = config_options.map(ConfigOptions::new);
1272                        session.ref_count = ref_count;
1273                    }
1274
1275                    Ok(thread)
1276                }
1277            })
1278            .shared();
1279
1280        self.pending_sessions.borrow_mut().insert(
1281            session_id,
1282            PendingAcpSession {
1283                task: shared_task.clone(),
1284                ref_count: 1,
1285            },
1286        );
1287
1288        cx.foreground_executor()
1289            .spawn(async move { shared_task.await.map_err(|err| anyhow!(err)) })
1290    }
1291
1292    fn apply_default_config_options(
1293        &self,
1294        session_id: &acp::SessionId,
1295        config_options: &Rc<RefCell<Vec<acp::SessionConfigOption>>>,
1296        cx: &mut AsyncApp,
1297    ) {
1298        let id = self.id.clone();
1299        let defaults_to_apply: Vec<_> = {
1300            let config_opts_ref = config_options.borrow();
1301            config_opts_ref
1302                .iter()
1303                .filter_map(|config_option| {
1304                    let default_value = self.defaults.config_option(config_option.id.0.as_ref())?;
1305
1306                    let value_to_apply = match &config_option.kind {
1307                        acp::SessionConfigKind::Select(select) => {
1308                            let value_id = default_value.as_value_id()?;
1309                            match &select.options {
1310                                acp::SessionConfigSelectOptions::Ungrouped(options) => options
1311                                    .iter()
1312                                    .any(|opt| &*opt.value.0 == value_id)
1313                                    .then(|| {
1314                                        acp::SessionConfigOptionValue::value_id(
1315                                            value_id.to_string(),
1316                                        )
1317                                    }),
1318                                acp::SessionConfigSelectOptions::Grouped(groups) => groups
1319                                    .iter()
1320                                    .any(|group| {
1321                                        group.options.iter().any(|opt| &*opt.value.0 == value_id)
1322                                    })
1323                                    .then(|| {
1324                                        acp::SessionConfigOptionValue::value_id(
1325                                            value_id.to_string(),
1326                                        )
1327                                    }),
1328                                _ => None,
1329                            }
1330                        }
1331                        acp::SessionConfigKind::Boolean(_) => default_value
1332                            .as_bool()
1333                            .map(acp::SessionConfigOptionValue::boolean),
1334                        _ => None,
1335                    };
1336
1337                    if let Some(value_to_apply) = value_to_apply {
1338                        let initial_value = match &config_option.kind {
1339                            acp::SessionConfigKind::Select(select) => {
1340                                acp::SessionConfigOptionValue::value_id(
1341                                    select.current_value.clone(),
1342                                )
1343                            }
1344                            acp::SessionConfigKind::Boolean(boolean) => {
1345                                acp::SessionConfigOptionValue::boolean(boolean.current_value)
1346                            }
1347                            _ => return None,
1348                        };
1349
1350                        Some((config_option.id.clone(), value_to_apply, initial_value))
1351                    } else {
1352                        log::warn!(
1353                            "`{}` is not a valid value for config option `{}` in {}",
1354                            default_value,
1355                            config_option.id.0,
1356                            id
1357                        );
1358                        None
1359                    }
1360                })
1361                .collect()
1362        };
1363
1364        for (config_id, default_value, initial_value) in defaults_to_apply {
1365            cx.spawn({
1366                let default_value_for_request = default_value.clone();
1367                let session_id = session_id.clone();
1368                let config_id_clone = config_id.clone();
1369                let config_opts = config_options.clone();
1370                let conn = self.connection.clone();
1371                async move |_| {
1372                    let result = conn
1373                        .send_request(acp::SetSessionConfigOptionRequest::new(
1374                            session_id,
1375                            config_id_clone.clone(),
1376                            default_value_for_request,
1377                        ))
1378                        .block_task()
1379                        .await
1380                        .log_err();
1381
1382                    if result.is_none() {
1383                        let mut opts = config_opts.borrow_mut();
1384                        if let Some(opt) = opts.iter_mut().find(|o| o.id == config_id_clone) {
1385                            match (&mut opt.kind, &initial_value) {
1386                                (
1387                                    acp::SessionConfigKind::Select(select),
1388                                    acp::SessionConfigOptionValue::ValueId { value },
1389                                ) => {
1390                                    select.current_value = value.clone();
1391                                }
1392                                (
1393                                    acp::SessionConfigKind::Boolean(boolean),
1394                                    acp::SessionConfigOptionValue::Boolean { value },
1395                                ) => {
1396                                    boolean.current_value = *value;
1397                                }
1398                                _ => {}
1399                            }
1400                        }
1401                    }
1402                }
1403            })
1404            .detach();
1405
1406            let mut opts = config_options.borrow_mut();
1407            if let Some(opt) = opts.iter_mut().find(|o| o.id == config_id) {
1408                match (&mut opt.kind, &default_value) {
1409                    (
1410                        acp::SessionConfigKind::Select(select),
1411                        acp::SessionConfigOptionValue::ValueId { value },
1412                    ) => {
1413                        select.current_value = value.clone();
1414                    }
1415                    (
1416                        acp::SessionConfigKind::Boolean(boolean),
1417                        acp::SessionConfigOptionValue::Boolean { value },
1418                    ) => {
1419                        boolean.current_value = *value;
1420                    }
1421                    _ => {}
1422                }
1423            }
1424        }
1425    }
1426}
1427
1428#[derive(Clone, Debug, PartialEq, Eq)]
1429struct SessionDirectories {
1430    cwd: PathBuf,
1431    additional_directories: Vec<PathBuf>,
1432}
1433
1434impl SessionDirectories {
1435    fn into_new_session_request(self, mcp_servers: Vec<acp::McpServer>) -> acp::NewSessionRequest {
1436        acp::NewSessionRequest::new(self.cwd)
1437            .additional_directories(self.additional_directories)
1438            .mcp_servers(mcp_servers)
1439    }
1440
1441    fn into_load_session_request(
1442        self,
1443        session_id: acp::SessionId,
1444        mcp_servers: Vec<acp::McpServer>,
1445    ) -> acp::LoadSessionRequest {
1446        acp::LoadSessionRequest::new(session_id, self.cwd)
1447            .additional_directories(self.additional_directories)
1448            .mcp_servers(mcp_servers)
1449    }
1450
1451    fn into_resume_session_request(
1452        self,
1453        session_id: acp::SessionId,
1454        mcp_servers: Vec<acp::McpServer>,
1455    ) -> acp::ResumeSessionRequest {
1456        acp::ResumeSessionRequest::new(session_id, self.cwd)
1457            .additional_directories(self.additional_directories)
1458            .mcp_servers(mcp_servers)
1459    }
1460}
1461
1462fn session_directories_from_work_dirs(
1463    work_dirs: &PathList,
1464    supports_additional_directories: bool,
1465) -> Result<SessionDirectories> {
1466    let mut ordered_paths = work_dirs.ordered_paths();
1467    let cwd = ordered_paths
1468        .next()
1469        .cloned()
1470        .ok_or_else(|| anyhow!("Working directory cannot be empty"))?;
1471    let additional_directories = if supports_additional_directories {
1472        ordered_paths.cloned().collect()
1473    } else {
1474        Vec::new()
1475    };
1476
1477    Ok(SessionDirectories {
1478        cwd,
1479        additional_directories,
1480    })
1481}
1482
1483fn work_dirs_from_session_info(cwd: PathBuf, additional_directories: Vec<PathBuf>) -> PathList {
1484    let mut seen_paths = HashSet::default();
1485    let mut paths = Vec::with_capacity(1 + additional_directories.len());
1486
1487    seen_paths.insert(cwd.clone());
1488    paths.push(cwd);
1489
1490    for path in additional_directories {
1491        if seen_paths.insert(path.clone()) {
1492            paths.push(path);
1493        }
1494    }
1495
1496    PathList::new(&paths)
1497}
1498
1499fn emit_load_error_to_all_sessions(
1500    sessions: &Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
1501    error: LoadError,
1502    cx: &mut AsyncApp,
1503) {
1504    let threads: Vec<_> = sessions
1505        .borrow()
1506        .values()
1507        .map(|session| session.thread.clone())
1508        .collect();
1509
1510    for thread in threads {
1511        thread
1512            .update(cx, |thread, cx| thread.emit_load_error(error.clone(), cx))
1513            .ok();
1514    }
1515}
1516
1517impl Drop for AcpConnection {
1518    fn drop(&mut self) {
1519        if let Some(ref mut child) = self.child {
1520            child.kill().log_err();
1521        }
1522    }
1523}
1524
1525fn terminal_auth_task_id(agent_id: &AgentId, method_id: &acp::AuthMethodId) -> String {
1526    format!("external-agent-{}-{}-login", agent_id.0, method_id.0)
1527}
1528
1529fn terminal_auth_task(
1530    command: &AgentServerCommand,
1531    agent_id: &AgentId,
1532    method: &acp::AuthMethodTerminal,
1533) -> SpawnInTerminal {
1534    acp_thread::build_terminal_auth_task(
1535        terminal_auth_task_id(agent_id, &method.id),
1536        method.name.clone(),
1537        command.path.to_string_lossy().into_owned(),
1538        command.args.clone(),
1539        command.env.clone().unwrap_or_default(),
1540    )
1541}
1542
1543/// Used to support the _meta method prior to stabilization
1544fn meta_terminal_auth_task(
1545    agent_id: &AgentId,
1546    method_id: &acp::AuthMethodId,
1547    method: &acp::AuthMethod,
1548) -> Option<SpawnInTerminal> {
1549    #[derive(Deserialize)]
1550    struct MetaTerminalAuth {
1551        label: String,
1552        command: String,
1553        #[serde(default)]
1554        args: Vec<String>,
1555        #[serde(default)]
1556        env: HashMap<String, String>,
1557    }
1558
1559    let meta = match method {
1560        acp::AuthMethod::EnvVar(env_var) => env_var.meta.as_ref(),
1561        acp::AuthMethod::Terminal(terminal) => terminal.meta.as_ref(),
1562        acp::AuthMethod::Agent(agent) => agent.meta.as_ref(),
1563        _ => None,
1564    }?;
1565    let terminal_auth =
1566        serde_json::from_value::<MetaTerminalAuth>(meta.get("terminal-auth")?.clone()).ok()?;
1567
1568    Some(acp_thread::build_terminal_auth_task(
1569        terminal_auth_task_id(agent_id, method_id),
1570        terminal_auth.label.clone(),
1571        terminal_auth.command,
1572        terminal_auth.args,
1573        terminal_auth.env,
1574    ))
1575}
1576
1577impl AgentConnection for AcpConnection {
1578    fn agent_id(&self) -> AgentId {
1579        self.id.clone()
1580    }
1581
1582    fn telemetry_id(&self) -> SharedString {
1583        self.telemetry_id.clone()
1584    }
1585
1586    fn agent_version(&self) -> Option<SharedString> {
1587        self.agent_version.clone()
1588    }
1589
1590    fn new_session(
1591        self: Rc<Self>,
1592        project: Entity<Project>,
1593        work_dirs: PathList,
1594        cx: &mut App,
1595    ) -> Task<Result<Entity<AcpThread>>> {
1596        let directories = match self.session_directories_from_work_dirs(&work_dirs) {
1597            Ok(directories) => directories,
1598            Err(error) => return Task::ready(Err(error)),
1599        };
1600        let name = self.id.0.clone();
1601        let mcp_servers = mcp_servers_for_project(&project, cx);
1602
1603        cx.spawn(async move |cx| {
1604            let response = self
1605                .connection
1606                .send_request(directories.into_new_session_request(mcp_servers))
1607                .block_task()
1608            .await
1609            .map_err(map_acp_error)?;
1610
1611            let (modes, config_options) = config_state(response.modes, response.config_options);
1612
1613            let default_mode = self.defaults.mode();
1614            if let Some(default_mode) = default_mode {
1615                if let Some(modes) = modes.as_ref() {
1616                    let mut modes_ref = modes.borrow_mut();
1617                    let has_mode = modes_ref
1618                        .available_modes
1619                        .iter()
1620                        .any(|mode| mode.id == default_mode);
1621
1622                    if has_mode {
1623                        let initial_mode_id = modes_ref.current_mode_id.clone();
1624
1625                        cx.spawn({
1626                            let default_mode = default_mode.clone();
1627                            let session_id = response.session_id.clone();
1628                            let modes = modes.clone();
1629                            let conn = self.connection.clone();
1630                            async move |_| {
1631                                let result = conn
1632                                    .send_request(acp::SetSessionModeRequest::new(
1633                                        session_id,
1634                                        default_mode,
1635                                    ))
1636                                    .block_task()
1637                                .await
1638                                .log_err();
1639
1640                                if result.is_none() {
1641                                    modes.borrow_mut().current_mode_id = initial_mode_id;
1642                                }
1643                            }
1644                        })
1645                        .detach();
1646
1647                        modes_ref.current_mode_id = default_mode;
1648                    } else {
1649                        let available_modes = modes_ref
1650                            .available_modes
1651                            .iter()
1652                            .map(|mode| format!("- `{}`: {}", mode.id, mode.name))
1653                            .collect::<Vec<_>>()
1654                            .join("\n");
1655
1656                        log::warn!(
1657                            "`{default_mode}` is not valid {name} mode. Available options:\n{available_modes}",
1658                        );
1659                    }
1660                }
1661            }
1662
1663            if let Some(config_opts) = config_options.as_ref() {
1664                self.apply_default_config_options(&response.session_id, config_opts, cx);
1665            }
1666
1667            let action_log = cx.new(|_| ActionLog::new(project.clone()));
1668            let thread: Entity<AcpThread> = cx.new(|cx| {
1669                AcpThread::new(
1670                    None,
1671                    None,
1672                    Some(work_dirs),
1673                    self.clone(),
1674                    project,
1675                    action_log,
1676                    response.session_id.clone(),
1677                    // ACP doesn't currently support per-session prompt capabilities or changing capabilities dynamically.
1678                    watch::Receiver::constant(
1679                        self.agent_capabilities.prompt_capabilities.clone(),
1680                    ),
1681                    cx,
1682                )
1683            });
1684
1685            self.sessions.borrow_mut().insert(
1686                response.session_id,
1687                AcpSession {
1688                    thread: thread.downgrade(),
1689                    suppress_abort_err: false,
1690                    session_modes: modes,
1691                    config_options: config_options.map(ConfigOptions::new),
1692                    ref_count: 1,
1693                },
1694            );
1695
1696            Ok(thread)
1697        })
1698    }
1699
1700    fn supports_load_session(&self) -> bool {
1701        self.agent_capabilities.load_session
1702    }
1703
1704    fn supports_resume_session(&self) -> bool {
1705        self.agent_capabilities
1706            .session_capabilities
1707            .resume
1708            .is_some()
1709    }
1710
1711    fn supports_session_additional_directories(&self) -> bool {
1712        self.agent_capabilities
1713            .session_capabilities
1714            .additional_directories
1715            .is_some()
1716    }
1717
1718    fn load_session(
1719        self: Rc<Self>,
1720        session_id: acp::SessionId,
1721        project: Entity<Project>,
1722        work_dirs: PathList,
1723        title: Option<SharedString>,
1724        cx: &mut App,
1725    ) -> Task<Result<Entity<AcpThread>>> {
1726        if !self.agent_capabilities.load_session {
1727            return Task::ready(Err(anyhow!(LoadError::Other(
1728                "Loading sessions is not supported by this agent.".into()
1729            ))));
1730        }
1731
1732        let mcp_servers = mcp_servers_for_project(&project, cx);
1733        self.open_or_create_session(
1734            session_id,
1735            project,
1736            work_dirs,
1737            title,
1738            move |connection, session_id, directories| {
1739                Box::pin(async move {
1740                    let response = connection
1741                        .send_request(
1742                            directories.into_load_session_request(session_id.clone(), mcp_servers),
1743                        )
1744                        .block_task()
1745                        .await
1746                        .map_err(map_session_open_error)?;
1747                    Ok(SessionConfigResponse {
1748                        modes: response.modes,
1749                        config_options: response.config_options,
1750                    })
1751                })
1752            },
1753            cx,
1754        )
1755    }
1756
1757    fn resume_session(
1758        self: Rc<Self>,
1759        session_id: acp::SessionId,
1760        project: Entity<Project>,
1761        work_dirs: PathList,
1762        title: Option<SharedString>,
1763        cx: &mut App,
1764    ) -> Task<Result<Entity<AcpThread>>> {
1765        if self
1766            .agent_capabilities
1767            .session_capabilities
1768            .resume
1769            .is_none()
1770        {
1771            return Task::ready(Err(anyhow!(LoadError::Other(
1772                "Resuming sessions is not supported by this agent.".into()
1773            ))));
1774        }
1775
1776        let mcp_servers = mcp_servers_for_project(&project, cx);
1777        self.open_or_create_session(
1778            session_id,
1779            project,
1780            work_dirs,
1781            title,
1782            move |connection, session_id, directories| {
1783                Box::pin(async move {
1784                    let response = connection
1785                        .send_request(
1786                            directories
1787                                .into_resume_session_request(session_id.clone(), mcp_servers),
1788                        )
1789                        .block_task()
1790                        .await
1791                        .map_err(map_session_open_error)?;
1792                    Ok(SessionConfigResponse {
1793                        modes: response.modes,
1794                        config_options: response.config_options,
1795                    })
1796                })
1797            },
1798            cx,
1799        )
1800    }
1801
1802    fn supports_close_session(&self) -> bool {
1803        self.agent_capabilities.session_capabilities.close.is_some()
1804    }
1805
1806    fn close_session(
1807        self: Rc<Self>,
1808        session_id: &acp::SessionId,
1809        cx: &mut App,
1810    ) -> Task<Result<()>> {
1811        if !self.supports_close_session() {
1812            return Task::ready(Err(anyhow!(LoadError::Other(
1813                "Closing sessions is not supported by this agent.".into()
1814            ))));
1815        }
1816
1817        // If a load is still in flight, decrement its ref count. The pending
1818        // entry is the source of truth for how many handles exist during a
1819        // load, so we must tick it down here as well as the `sessions` entry
1820        // that was pre-registered to receive history-replay notifications.
1821        // Only once the pending ref count hits zero do we actually close the
1822        // session; the load task will observe the missing sessions entry and
1823        // fail with "session was closed before load completed".
1824        let pending_ref_count = {
1825            let mut pending_sessions = self.pending_sessions.borrow_mut();
1826            pending_sessions.get_mut(session_id).map(|pending| {
1827                pending.ref_count = pending.ref_count.saturating_sub(1);
1828                pending.ref_count
1829            })
1830        };
1831        match pending_ref_count {
1832            Some(0) => {
1833                self.pending_sessions.borrow_mut().remove(session_id);
1834                self.sessions.borrow_mut().remove(session_id);
1835
1836                let conn = self.connection.clone();
1837                let session_id = session_id.clone();
1838                return cx.foreground_executor().spawn(async move {
1839                    conn.send_request(acp::CloseSessionRequest::new(session_id))
1840                        .block_task()
1841                        .await?;
1842                    Ok(())
1843                });
1844            }
1845            Some(_) => return Task::ready(Ok(())),
1846            None => {}
1847        }
1848
1849        let mut sessions = self.sessions.borrow_mut();
1850        let Some(session) = sessions.get_mut(session_id) else {
1851            return Task::ready(Ok(()));
1852        };
1853
1854        session.ref_count = session.ref_count.saturating_sub(1);
1855        if session.ref_count > 0 {
1856            return Task::ready(Ok(()));
1857        }
1858
1859        sessions.remove(session_id);
1860        drop(sessions);
1861
1862        let conn = self.connection.clone();
1863        let session_id = session_id.clone();
1864        cx.foreground_executor().spawn(async move {
1865            conn.send_request(acp::CloseSessionRequest::new(session_id.clone()))
1866                .block_task()
1867                .await?;
1868            Ok(())
1869        })
1870    }
1871
1872    fn auth_methods(&self) -> &[acp::AuthMethod] {
1873        &self.auth_methods
1874    }
1875
1876    fn terminal_auth_task(
1877        &self,
1878        method_id: &acp::AuthMethodId,
1879        cx: &App,
1880    ) -> Option<Task<Result<SpawnInTerminal>>> {
1881        let method = self
1882            .auth_methods
1883            .iter()
1884            .find(|method| method.id() == method_id)?;
1885
1886        match method {
1887            acp::AuthMethod::Terminal(terminal) if cx.has_flag::<AcpBetaFeatureFlag>() => {
1888                let agent_id = self.id.clone();
1889                let terminal = terminal.clone();
1890                let store = self.agent_server_store.clone();
1891                Some(cx.spawn(async move |cx| {
1892                    let command = store
1893                        .update(cx, |store, cx| {
1894                            let agent = store
1895                                .get_external_agent(&agent_id)
1896                                .context("Agent server not found")?;
1897                            anyhow::Ok(agent.get_command(
1898                                terminal.args.clone(),
1899                                HashMap::from_iter(terminal.env.clone()),
1900                                &mut cx.to_async(),
1901                            ))
1902                        })?
1903                        .context("Failed to get agent command")?
1904                        .await?;
1905                    Ok(terminal_auth_task(&command, &agent_id, &terminal))
1906                }))
1907            }
1908            _ => meta_terminal_auth_task(&self.id, method_id, method)
1909                .map(|task| Task::ready(Ok(task))),
1910        }
1911    }
1912
1913    fn authenticate(&self, method_id: acp::AuthMethodId, cx: &mut App) -> Task<Result<()>> {
1914        let conn = self.connection.clone();
1915        cx.foreground_executor().spawn(async move {
1916            conn.send_request(acp::AuthenticateRequest::new(method_id))
1917                .block_task()
1918                .await?;
1919            Ok(())
1920        })
1921    }
1922
1923    fn supports_logout(&self) -> bool {
1924        self.agent_capabilities.auth.logout.is_some()
1925    }
1926
1927    fn logout(&self, cx: &mut App) -> Task<Result<()>> {
1928        if !self.supports_logout() {
1929            return Task::ready(Err(anyhow!("Logout is not supported by this agent.")));
1930        }
1931
1932        let conn = self.connection.clone();
1933        cx.foreground_executor().spawn(async move {
1934            conn.send_request(acp::LogoutRequest::new())
1935                .block_task()
1936                .await?;
1937            Ok(())
1938        })
1939    }
1940
1941    fn prompt(
1942        &self,
1943        params: acp::PromptRequest,
1944        cx: &mut App,
1945    ) -> Task<Result<acp::PromptResponse>> {
1946        let conn = self.connection.clone();
1947        let sessions = self.sessions.clone();
1948        let session_id = params.session_id.clone();
1949        cx.foreground_executor().spawn(async move {
1950            let result = conn.send_request(params).block_task().await;
1951
1952            let mut suppress_abort_err = false;
1953
1954            if let Some(session) = sessions.borrow_mut().get_mut(&session_id) {
1955                suppress_abort_err = session.suppress_abort_err;
1956                session.suppress_abort_err = false;
1957            }
1958
1959            match result {
1960                Ok(response) => Ok(response),
1961                Err(err) => {
1962                    if err.code == acp::ErrorCode::AuthRequired {
1963                        return Err(anyhow!(acp::Error::auth_required()));
1964                    }
1965
1966                    if err.code != ErrorCode::InternalError {
1967                        anyhow::bail!(err)
1968                    }
1969
1970                    let Some(data) = &err.data else {
1971                        anyhow::bail!(err)
1972                    };
1973
1974                    // Temporary workaround until the following PR is generally available:
1975                    // https://github.com/google-gemini/gemini-cli/pull/6656
1976
1977                    #[derive(Deserialize)]
1978                    #[serde(deny_unknown_fields)]
1979                    struct ErrorDetails {
1980                        details: Box<str>,
1981                    }
1982
1983                    match serde_json::from_value(data.clone()) {
1984                        Ok(ErrorDetails { details }) => {
1985                            if suppress_abort_err
1986                                && (details.contains("This operation was aborted")
1987                                    || details.contains("The user aborted a request"))
1988                            {
1989                                Ok(acp::PromptResponse::new(acp::StopReason::Cancelled))
1990                            } else {
1991                                Err(anyhow!(details))
1992                            }
1993                        }
1994                        Err(_) => Err(anyhow!(err)),
1995                    }
1996                }
1997            }
1998        })
1999    }
2000
2001    fn cancel(&self, session_id: &acp::SessionId, _cx: &mut App) {
2002        if let Some(session) = self.sessions.borrow_mut().get_mut(session_id) {
2003            session.suppress_abort_err = true;
2004        }
2005        let params = acp::CancelNotification::new(session_id.clone());
2006        self.connection.send_notification(params).log_err();
2007    }
2008
2009    fn request_elicitations(&self) -> Option<Entity<ElicitationStore>> {
2010        Some(self.request_elicitations.clone())
2011    }
2012
2013    fn session_modes(
2014        &self,
2015        session_id: &acp::SessionId,
2016        _cx: &App,
2017    ) -> Option<Rc<dyn acp_thread::AgentSessionModes>> {
2018        let sessions = self.sessions.clone();
2019        let sessions_ref = sessions.borrow();
2020        let Some(session) = sessions_ref.get(session_id) else {
2021            return None;
2022        };
2023
2024        if let Some(modes) = session.session_modes.as_ref() {
2025            Some(Rc::new(AcpSessionModes {
2026                connection: self.connection.clone(),
2027                session_id: session_id.clone(),
2028                state: modes.clone(),
2029            }) as _)
2030        } else {
2031            None
2032        }
2033    }
2034
2035    fn session_config_options(
2036        &self,
2037        session_id: &acp::SessionId,
2038        _cx: &App,
2039    ) -> Option<Rc<dyn acp_thread::AgentSessionConfigOptions>> {
2040        let sessions = self.sessions.borrow();
2041        let session = sessions.get(session_id)?;
2042
2043        let config_opts = session.config_options.as_ref()?;
2044
2045        Some(Rc::new(AcpSessionConfigOptions {
2046            session_id: session_id.clone(),
2047            connection: self.connection.clone(),
2048            state: config_opts.config_options.clone(),
2049            watch_tx: config_opts.tx.clone(),
2050            watch_rx: config_opts.rx.clone(),
2051        }) as _)
2052    }
2053
2054    fn session_list(&self, _cx: &mut App) -> Option<Rc<dyn AgentSessionList>> {
2055        self.session_list.clone().map(|s| s as _)
2056    }
2057
2058    fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
2059        self
2060    }
2061}
2062
2063fn map_acp_error(err: acp::Error) -> anyhow::Error {
2064    if err.code == acp::ErrorCode::AuthRequired {
2065        let mut error = AuthRequired::new();
2066
2067        if err.message != acp::ErrorCode::AuthRequired.to_string() {
2068            error = error.with_description(err.message);
2069        }
2070
2071        anyhow!(error)
2072    } else {
2073        anyhow!(err)
2074    }
2075}
2076
2077/// Map an error from a `session/load` or `session/resume` call.
2078///
2079/// `ResourceNotFound` from these two calls means one specific thing: the agent
2080/// does not have the session we named. Omega keeps its own thread record, and
2081/// that record outlives the agent's session store, so this is an ordinary end
2082/// of life rather than a failure to start. Left as a generic error it rendered
2083/// as "Failed to Launch: Resource not found: <uuid>" — a title that blames
2084/// startup and a body that names an identifier the reader cannot act on.
2085///
2086/// Only the session-scoped calls get this treatment. `ResourceNotFound` from a
2087/// file read means a missing file and must keep saying so.
2088fn map_session_open_error(err: acp::Error) -> anyhow::Error {
2089    if err.code == acp::ErrorCode::ResourceNotFound {
2090        return anyhow!(LoadError::SessionGone);
2091    }
2092    map_acp_error(err)
2093}
2094
2095#[cfg(any(test, feature = "test-support"))]
2096pub mod test_support {
2097    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
2098
2099    use acp_thread::{
2100        AgentSessionClientUserMessageIds, AgentSessionConfigOptions, AgentSessionModes,
2101        AgentSessionRetry, AgentSessionSetTitle, AgentSessionTruncate, AgentTelemetry,
2102    };
2103
2104    use super::*;
2105
2106    #[derive(Clone, Default)]
2107    pub struct FakeAcpAgentServer {
2108        load_session_count: Arc<AtomicUsize>,
2109        close_session_count: Arc<AtomicUsize>,
2110        fail_next_prompt: Arc<AtomicBool>,
2111        auth_elicitation_request: Arc<Mutex<Option<acp::CreateElicitationRequest>>>,
2112        auth_elicitation_response:
2113            Arc<Mutex<Option<async_channel::Sender<acp::CreateElicitationResponse>>>>,
2114        auth_elicitation_completion: Arc<Mutex<Option<acp::CompleteElicitationNotification>>>,
2115        exit_status_sender:
2116            Arc<std::sync::Mutex<Option<async_channel::Sender<std::process::ExitStatus>>>>,
2117    }
2118
2119    impl FakeAcpAgentServer {
2120        pub fn new() -> Self {
2121            Self::default()
2122        }
2123
2124        pub fn load_session_count(&self) -> Arc<AtomicUsize> {
2125            self.load_session_count.clone()
2126        }
2127
2128        pub fn close_session_count(&self) -> Arc<AtomicUsize> {
2129            self.close_session_count.clone()
2130        }
2131
2132        pub fn simulate_server_exit(&self) {
2133            let sender = self
2134                .exit_status_sender
2135                .lock()
2136                .expect("exit status sender lock should not be poisoned")
2137                .clone()
2138                .expect("fake ACP server must be connected before simulating exit");
2139            sender
2140                .try_send(std::process::ExitStatus::default())
2141                .expect("fake ACP server exit receiver should still be alive");
2142        }
2143
2144        pub fn fail_next_prompt(&self) {
2145            self.fail_next_prompt.store(true, Ordering::SeqCst);
2146        }
2147
2148        pub fn request_elicitation_during_auth(
2149            &self,
2150            request: acp::CreateElicitationRequest,
2151        ) -> async_channel::Receiver<acp::CreateElicitationResponse> {
2152            let (response_tx, response_rx) = async_channel::bounded(1);
2153            *self
2154                .auth_elicitation_request
2155                .lock()
2156                .expect("auth elicitation request lock should not be poisoned") = Some(request);
2157            *self
2158                .auth_elicitation_response
2159                .lock()
2160                .expect("auth elicitation response lock should not be poisoned") =
2161                Some(response_tx);
2162            response_rx
2163        }
2164    }
2165
2166    impl crate::AgentServer for FakeAcpAgentServer {
2167        fn logo(&self) -> ui::IconName {
2168            ui::IconName::OmegaAgent
2169        }
2170
2171        fn agent_id(&self) -> AgentId {
2172            AgentId::new("Test")
2173        }
2174
2175        fn connect(
2176            &self,
2177            _delegate: crate::AgentServerDelegate,
2178            project: Entity<Project>,
2179            cx: &mut App,
2180        ) -> Task<anyhow::Result<Rc<dyn AgentConnection>>> {
2181            let load_session_count = self.load_session_count.clone();
2182            let close_session_count = self.close_session_count.clone();
2183            let fail_next_prompt = self.fail_next_prompt.clone();
2184            let auth_elicitation_request = self.auth_elicitation_request.clone();
2185            let auth_elicitation_response = self.auth_elicitation_response.clone();
2186            let auth_elicitation_completion = self.auth_elicitation_completion.clone();
2187            let exit_status_sender = self.exit_status_sender.clone();
2188            cx.spawn(async move |cx| {
2189                let harness = build_fake_acp_connection(
2190                    project,
2191                    load_session_count,
2192                    close_session_count,
2193                    fail_next_prompt,
2194                    auth_elicitation_request,
2195                    auth_elicitation_response,
2196                    auth_elicitation_completion,
2197                    cx,
2198                )
2199                .await?;
2200                let (exit_tx, exit_rx) = async_channel::bounded(1);
2201                *exit_status_sender
2202                    .lock()
2203                    .expect("exit status sender lock should not be poisoned") = Some(exit_tx);
2204                let connection = harness.connection.clone();
2205                let simulate_exit_task = cx.spawn(async move |cx| {
2206                    while let Ok(status) = exit_rx.recv().await {
2207                        emit_load_error_to_all_sessions(
2208                            &connection.sessions,
2209                            LoadError::Exited {
2210                                status,
2211                                stderr: None,
2212                            },
2213                            cx,
2214                        );
2215                    }
2216                    Ok(())
2217                });
2218                Ok(Rc::new(FakeAcpAgentConnection {
2219                    inner: harness.connection,
2220                    _keep_agent_alive: harness.keep_agent_alive,
2221                    _simulate_exit_task: simulate_exit_task,
2222                }) as Rc<dyn AgentConnection>)
2223            })
2224        }
2225
2226        fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
2227            self
2228        }
2229    }
2230
2231    pub struct FakeAcpConnectionHarness {
2232        pub connection: Rc<AcpConnection>,
2233        pub load_session_count: Arc<AtomicUsize>,
2234        pub close_session_count: Arc<AtomicUsize>,
2235        pub logout_count: Arc<AtomicUsize>,
2236        pub keep_agent_alive: Task<anyhow::Result<()>>,
2237    }
2238
2239    struct FakeAcpAgentConnection {
2240        inner: Rc<AcpConnection>,
2241        _keep_agent_alive: Task<anyhow::Result<()>>,
2242        _simulate_exit_task: Task<anyhow::Result<()>>,
2243    }
2244
2245    impl AgentConnection for FakeAcpAgentConnection {
2246        fn agent_id(&self) -> AgentId {
2247            self.inner.agent_id()
2248        }
2249
2250        fn telemetry_id(&self) -> SharedString {
2251            self.inner.telemetry_id()
2252        }
2253
2254        fn agent_version(&self) -> Option<SharedString> {
2255            self.inner.agent_version()
2256        }
2257
2258        fn new_session(
2259            self: Rc<Self>,
2260            project: Entity<Project>,
2261            work_dirs: PathList,
2262            cx: &mut App,
2263        ) -> Task<Result<Entity<AcpThread>>> {
2264            self.inner.clone().new_session(project, work_dirs, cx)
2265        }
2266
2267        fn supports_load_session(&self) -> bool {
2268            self.inner.supports_load_session()
2269        }
2270
2271        fn load_session(
2272            self: Rc<Self>,
2273            session_id: acp::SessionId,
2274            project: Entity<Project>,
2275            work_dirs: PathList,
2276            title: Option<SharedString>,
2277            cx: &mut App,
2278        ) -> Task<Result<Entity<AcpThread>>> {
2279            self.inner
2280                .clone()
2281                .load_session(session_id, project, work_dirs, title, cx)
2282        }
2283
2284        fn supports_close_session(&self) -> bool {
2285            self.inner.supports_close_session()
2286        }
2287
2288        fn close_session(
2289            self: Rc<Self>,
2290            session_id: &acp::SessionId,
2291            cx: &mut App,
2292        ) -> Task<Result<()>> {
2293            self.inner.clone().close_session(session_id, cx)
2294        }
2295
2296        fn supports_resume_session(&self) -> bool {
2297            self.inner.supports_resume_session()
2298        }
2299
2300        fn supports_session_additional_directories(&self) -> bool {
2301            self.inner.supports_session_additional_directories()
2302        }
2303
2304        fn resume_session(
2305            self: Rc<Self>,
2306            session_id: acp::SessionId,
2307            project: Entity<Project>,
2308            work_dirs: PathList,
2309            title: Option<SharedString>,
2310            cx: &mut App,
2311        ) -> Task<Result<Entity<AcpThread>>> {
2312            self.inner
2313                .clone()
2314                .resume_session(session_id, project, work_dirs, title, cx)
2315        }
2316
2317        fn auth_methods(&self) -> &[acp::AuthMethod] {
2318            self.inner.auth_methods()
2319        }
2320
2321        fn terminal_auth_task(
2322            &self,
2323            method: &acp::AuthMethodId,
2324            cx: &App,
2325        ) -> Option<Task<Result<SpawnInTerminal>>> {
2326            self.inner.terminal_auth_task(method, cx)
2327        }
2328
2329        fn authenticate(&self, method: acp::AuthMethodId, cx: &mut App) -> Task<Result<()>> {
2330            self.inner.authenticate(method, cx)
2331        }
2332
2333        fn supports_logout(&self) -> bool {
2334            self.inner.supports_logout()
2335        }
2336
2337        fn logout(&self, cx: &mut App) -> Task<Result<()>> {
2338            self.inner.logout(cx)
2339        }
2340
2341        fn client_user_message_ids(
2342            &self,
2343            cx: &App,
2344        ) -> Option<Rc<dyn AgentSessionClientUserMessageIds>> {
2345            self.inner.client_user_message_ids(cx)
2346        }
2347
2348        fn prompt(
2349            &self,
2350            params: acp::PromptRequest,
2351            cx: &mut App,
2352        ) -> Task<Result<acp::PromptResponse>> {
2353            self.inner.prompt(params, cx)
2354        }
2355
2356        fn retry(
2357            &self,
2358            session_id: &acp::SessionId,
2359            cx: &App,
2360        ) -> Option<Rc<dyn AgentSessionRetry>> {
2361            self.inner.retry(session_id, cx)
2362        }
2363
2364        fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
2365            self.inner.cancel(session_id, cx)
2366        }
2367
2368        fn request_elicitations(&self) -> Option<Entity<ElicitationStore>> {
2369            self.inner.request_elicitations()
2370        }
2371
2372        fn truncate(
2373            &self,
2374            session_id: &acp::SessionId,
2375            cx: &App,
2376        ) -> Option<Rc<dyn AgentSessionTruncate>> {
2377            self.inner.truncate(session_id, cx)
2378        }
2379
2380        fn set_title(
2381            &self,
2382            session_id: &acp::SessionId,
2383            cx: &App,
2384        ) -> Option<Rc<dyn AgentSessionSetTitle>> {
2385            self.inner.set_title(session_id, cx)
2386        }
2387
2388        fn telemetry(&self) -> Option<Rc<dyn AgentTelemetry>> {
2389            self.inner.telemetry()
2390        }
2391
2392        fn session_modes(
2393            &self,
2394            session_id: &acp::SessionId,
2395            cx: &App,
2396        ) -> Option<Rc<dyn AgentSessionModes>> {
2397            self.inner.session_modes(session_id, cx)
2398        }
2399
2400        fn session_config_options(
2401            &self,
2402            session_id: &acp::SessionId,
2403            cx: &App,
2404        ) -> Option<Rc<dyn AgentSessionConfigOptions>> {
2405            self.inner.session_config_options(session_id, cx)
2406        }
2407
2408        fn session_list(&self, cx: &mut App) -> Option<Rc<dyn AgentSessionList>> {
2409            self.inner.session_list(cx)
2410        }
2411
2412        fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
2413            self
2414        }
2415    }
2416
2417    async fn build_fake_acp_connection(
2418        project: Entity<Project>,
2419        load_session_count: Arc<AtomicUsize>,
2420        close_session_count: Arc<AtomicUsize>,
2421        fail_next_prompt: Arc<AtomicBool>,
2422        auth_elicitation_request: Arc<Mutex<Option<acp::CreateElicitationRequest>>>,
2423        auth_elicitation_response: Arc<
2424            Mutex<Option<async_channel::Sender<acp::CreateElicitationResponse>>>,
2425        >,
2426        auth_elicitation_completion: Arc<Mutex<Option<acp::CompleteElicitationNotification>>>,
2427        cx: &mut AsyncApp,
2428    ) -> Result<FakeAcpConnectionHarness> {
2429        let (client_transport, agent_transport) = agent_client_protocol::Channel::duplex();
2430
2431        let logout_count = Arc::new(AtomicUsize::new(0));
2432        let sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>> =
2433            Rc::new(RefCell::new(HashMap::default()));
2434        let client_session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>> =
2435            Rc::new(RefCell::new(None));
2436
2437        let agent_future = Agent
2438            .builder()
2439            .name("fake-agent")
2440            .on_receive_request(
2441                async move |req: acp::InitializeRequest, responder, _cx| {
2442                    responder.respond(
2443                        acp::InitializeResponse::new(req.protocol_version).agent_capabilities(
2444                            acp::AgentCapabilities::default()
2445                                .load_session(true)
2446                                .session_capabilities(
2447                                    acp::SessionCapabilities::default()
2448                                        .close(acp::SessionCloseCapabilities::new()),
2449                                ),
2450                        ),
2451                    )
2452                },
2453                agent_client_protocol::on_receive_request!(),
2454            )
2455            .on_receive_request(
2456                {
2457                    let auth_elicitation_request = auth_elicitation_request.clone();
2458                    let auth_elicitation_response = auth_elicitation_response.clone();
2459                    let auth_elicitation_completion = auth_elicitation_completion.clone();
2460                    async move |_req: acp::AuthenticateRequest, responder, cx| {
2461                        let request = auth_elicitation_request
2462                            .lock()
2463                            .expect("auth elicitation request lock should not be poisoned")
2464                            .take();
2465                        let response_tx = auth_elicitation_response
2466                            .lock()
2467                            .expect("auth elicitation response lock should not be poisoned")
2468                            .take();
2469                        let completion = auth_elicitation_completion
2470                            .lock()
2471                            .expect("auth elicitation completion lock should not be poisoned")
2472                            .take();
2473
2474                        if let Some(request) = request {
2475                            cx.send_request(request)
2476                                .on_receiving_result(async move |result| {
2477                                    if let (Ok(response), Some(response_tx)) = (result, response_tx)
2478                                    {
2479                                        response_tx.send(response).await.ok();
2480                                    }
2481                                    responder.respond(Default::default())
2482                                })?;
2483                            if let Some(completion) = completion {
2484                                cx.send_notification(completion)?;
2485                            }
2486                            Ok(())
2487                        } else {
2488                            responder.respond(Default::default())
2489                        }
2490                    }
2491                },
2492                agent_client_protocol::on_receive_request!(),
2493            )
2494            .on_receive_request(
2495                async move |_req: acp::NewSessionRequest, responder, _cx| {
2496                    responder.respond(acp::NewSessionResponse::new(acp::SessionId::new("unused")))
2497                },
2498                agent_client_protocol::on_receive_request!(),
2499            )
2500            .on_receive_request(
2501                {
2502                    let fail_next_prompt = fail_next_prompt.clone();
2503                    async move |_req: acp::PromptRequest, responder, _cx| {
2504                        if fail_next_prompt.swap(false, Ordering::SeqCst) {
2505                            responder.respond_with_error(acp::ErrorCode::InternalError.into())
2506                        } else {
2507                            responder.respond(acp::PromptResponse::new(acp::StopReason::EndTurn))
2508                        }
2509                    }
2510                },
2511                agent_client_protocol::on_receive_request!(),
2512            )
2513            .on_receive_request(
2514                {
2515                    let load_session_count = load_session_count.clone();
2516                    async move |_req: acp::LoadSessionRequest, responder, _cx| {
2517                        load_session_count.fetch_add(1, Ordering::SeqCst);
2518                        responder.respond(acp::LoadSessionResponse::new())
2519                    }
2520                },
2521                agent_client_protocol::on_receive_request!(),
2522            )
2523            .on_receive_request(
2524                {
2525                    let close_session_count = close_session_count.clone();
2526                    async move |_req: acp::CloseSessionRequest, responder, _cx| {
2527                        close_session_count.fetch_add(1, Ordering::SeqCst);
2528                        responder.respond(acp::CloseSessionResponse::new())
2529                    }
2530                },
2531                agent_client_protocol::on_receive_request!(),
2532            )
2533            .on_receive_request(
2534                {
2535                    let logout_count = logout_count.clone();
2536                    async move |_req: acp::LogoutRequest, responder, _cx| {
2537                        logout_count.fetch_add(1, Ordering::SeqCst);
2538                        responder.respond(acp::LogoutResponse::new())
2539                    }
2540                },
2541                agent_client_protocol::on_receive_request!(),
2542            )
2543            .on_receive_notification(
2544                async move |_notif: acp::CancelNotification, _cx| Ok(()),
2545                agent_client_protocol::on_receive_notification!(),
2546            )
2547            .connect_to(agent_transport);
2548
2549        let agent_io_task = cx.background_spawn(agent_future);
2550
2551        // Wire the production handler set into the fake client so inbound
2552        // requests/notifications from the fake agent are dispatched the
2553        // same way the real `stdio` path does.
2554        let (dispatch_tx, dispatch_rx) = mpsc::unbounded::<ForegroundWork>();
2555
2556        let (connection_tx, connection_rx) = futures::channel::oneshot::channel();
2557        let client_future = connect_client_future(
2558            "zed-test",
2559            client_transport,
2560            dispatch_tx.clone(),
2561            connection_tx,
2562        );
2563        let client_io_task = cx.background_spawn(async move {
2564            client_future.await.ok();
2565        });
2566
2567        let client_conn: ConnectionTo<Agent> = connection_rx
2568            .await
2569            .context("failed to receive fake ACP connection handle")?;
2570
2571        let response = client_conn
2572            .send_request(acp::InitializeRequest::new(ProtocolVersion::V1))
2573            .block_task()
2574            .await?;
2575
2576        let agent_capabilities = response.agent_capabilities;
2577
2578        let request_elicitations = cx.new(|_| ElicitationStore::default());
2579        let dispatch_context = ClientContext {
2580            sessions: sessions.clone(),
2581            session_list: client_session_list.clone(),
2582            request_elicitations: request_elicitations.clone(),
2583        };
2584        let dispatch_task = cx.spawn({
2585            let mut dispatch_rx = dispatch_rx;
2586            async move |cx| {
2587                while let Some(work) = dispatch_rx.next().await {
2588                    work.run(cx, &dispatch_context);
2589                }
2590            }
2591        });
2592
2593        let agent_server_store =
2594            project.read_with(cx, |project, _| project.agent_server_store().downgrade());
2595
2596        let connection = cx.update(|cx| {
2597            AcpConnection::new_for_test(
2598                client_conn,
2599                sessions,
2600                agent_capabilities,
2601                request_elicitations,
2602                agent_server_store,
2603                client_io_task,
2604                dispatch_task,
2605                cx,
2606            )
2607        });
2608
2609        let keep_agent_alive = cx.background_spawn(async move {
2610            agent_io_task.await.ok();
2611            anyhow::Ok(())
2612        });
2613
2614        Ok(FakeAcpConnectionHarness {
2615            connection: Rc::new(connection),
2616            load_session_count,
2617            close_session_count,
2618            logout_count,
2619            keep_agent_alive,
2620        })
2621    }
2622
2623    pub async fn connect_fake_acp_connection(
2624        project: Entity<Project>,
2625        cx: &mut gpui::TestAppContext,
2626    ) -> FakeAcpConnectionHarness {
2627        cx.update(|cx| {
2628            let store = settings::SettingsStore::test(cx);
2629            cx.set_global(store);
2630        });
2631
2632        build_fake_acp_connection(
2633            project,
2634            Arc::new(AtomicUsize::new(0)),
2635            Arc::new(AtomicUsize::new(0)),
2636            Arc::new(AtomicBool::new(false)),
2637            Arc::new(Mutex::new(None)),
2638            Arc::new(Mutex::new(None)),
2639            Arc::new(Mutex::new(None)),
2640            &mut cx.to_async(),
2641        )
2642        .await
2643        .expect("failed to initialize ACP connection")
2644    }
2645
2646    #[cfg(test)]
2647    pub async fn connect_fake_acp_connection_with_auth_elicitation(
2648        project: Entity<Project>,
2649        request: acp::CreateElicitationRequest,
2650        cx: &mut gpui::TestAppContext,
2651    ) -> (
2652        FakeAcpConnectionHarness,
2653        async_channel::Receiver<acp::CreateElicitationResponse>,
2654    ) {
2655        cx.update(|cx| {
2656            let store = settings::SettingsStore::test(cx);
2657            cx.set_global(store);
2658        });
2659
2660        let (response_tx, response_rx) = async_channel::bounded(1);
2661        let harness = build_fake_acp_connection(
2662            project,
2663            Arc::new(AtomicUsize::new(0)),
2664            Arc::new(AtomicUsize::new(0)),
2665            Arc::new(AtomicBool::new(false)),
2666            Arc::new(Mutex::new(Some(request))),
2667            Arc::new(Mutex::new(Some(response_tx))),
2668            Arc::new(Mutex::new(None)),
2669            &mut cx.to_async(),
2670        )
2671        .await
2672        .expect("failed to initialize ACP connection");
2673
2674        (harness, response_rx)
2675    }
2676
2677    #[cfg(test)]
2678    pub async fn connect_fake_acp_connection_with_auth_elicitation_completion(
2679        project: Entity<Project>,
2680        request: acp::CreateElicitationRequest,
2681        completion: acp::CompleteElicitationNotification,
2682        cx: &mut gpui::TestAppContext,
2683    ) -> (
2684        FakeAcpConnectionHarness,
2685        async_channel::Receiver<acp::CreateElicitationResponse>,
2686    ) {
2687        cx.update(|cx| {
2688            let store = settings::SettingsStore::test(cx);
2689            cx.set_global(store);
2690        });
2691
2692        let (response_tx, response_rx) = async_channel::bounded(1);
2693        let harness = build_fake_acp_connection(
2694            project,
2695            Arc::new(AtomicUsize::new(0)),
2696            Arc::new(AtomicUsize::new(0)),
2697            Arc::new(AtomicBool::new(false)),
2698            Arc::new(Mutex::new(Some(request))),
2699            Arc::new(Mutex::new(Some(response_tx))),
2700            Arc::new(Mutex::new(Some(completion))),
2701            &mut cx.to_async(),
2702        )
2703        .await
2704        .expect("failed to initialize ACP connection");
2705
2706        (harness, response_rx)
2707    }
2708}
2709
2710#[cfg(test)]
2711mod tests {
2712    use std::sync::atomic::{AtomicUsize, Ordering};
2713
2714    use super::*;
2715    use feature_flags::FeatureFlag as _;
2716    use settings::Settings as _;
2717
2718    fn init_feature_flags_test(cx: &mut gpui::TestAppContext) {
2719        cx.update(|cx| {
2720            let mut settings_store = SettingsStore::test(cx);
2721            settings_store.register_setting::<feature_flags::FeatureFlagsSettings>();
2722            cx.set_global(settings_store);
2723            cx.update_flags(false, vec![]);
2724        });
2725    }
2726
2727    #[gpui::test]
2728    async fn client_capabilities_include_elicitation_without_acp_beta(
2729        cx: &mut gpui::TestAppContext,
2730    ) {
2731        init_feature_flags_test(cx);
2732        let capabilities = client_capabilities_for_agent(&AgentId::new("codex-acp"));
2733        let elicitation = capabilities
2734            .elicitation
2735            .expect("elicitation should always be advertised");
2736
2737        assert!(elicitation.form.is_some());
2738        assert!(elicitation.url.is_some());
2739    }
2740
2741    #[gpui::test]
2742    async fn request_scoped_elicitation_during_auth_uses_connection_store(
2743        cx: &mut gpui::TestAppContext,
2744    ) {
2745        init_feature_flags_test(cx);
2746        cx.update(|cx| {
2747            cx.update_flags(false, vec![AcpBetaFeatureFlag::NAME.to_string()]);
2748        });
2749
2750        let fs = fs::FakeFs::new(cx.executor());
2751        fs.insert_tree("/", serde_json::json!({ "a": {} })).await;
2752        let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await;
2753
2754        let request_id = acp::RequestId::Number(1);
2755        let (harness, response_rx) =
2756            test_support::connect_fake_acp_connection_with_auth_elicitation(
2757                project,
2758                acp::CreateElicitationRequest::new(
2759                    acp::ElicitationFormMode::new(
2760                        acp::ElicitationRequestScope::new(request_id.clone()),
2761                        acp::ElicitationSchema::new().string("name", true),
2762                    ),
2763                    "Provide a name",
2764                ),
2765                cx,
2766            )
2767            .await;
2768        let connection = harness.connection.clone();
2769        let auth_task =
2770            cx.update(|cx| connection.authenticate(acp::AuthMethodId::new("login"), cx));
2771        cx.run_until_parked();
2772
2773        let store = connection
2774            .request_elicitations()
2775            .expect("ACP connections expose request-scoped elicitations");
2776        let elicitation_id = store.read_with(cx, |store, _| {
2777            let [elicitation] = store.elicitations() else {
2778                panic!(
2779                    "expected one request-scoped elicitation, got {:?}",
2780                    store.elicitations()
2781                );
2782            };
2783            let acp::ElicitationScope::Request(scope) = elicitation.request.scope() else {
2784                panic!("expected request-scoped elicitation");
2785            };
2786            assert_eq!(scope.request_id, request_id);
2787            elicitation.id.clone()
2788        });
2789        assert!(
2790            connection.sessions.borrow().is_empty(),
2791            "auth-time request-scoped elicitations must not require a session"
2792        );
2793
2794        let expected_content = std::collections::BTreeMap::from([(
2795            "name".to_string(),
2796            acp::ElicitationContentValue::from("Ada"),
2797        )]);
2798        store.update(cx, |store, cx| {
2799            store.respond_to_elicitation(
2800                &elicitation_id,
2801                acp::CreateElicitationResponse::new(acp::ElicitationAction::Accept(
2802                    acp::ElicitationAcceptAction::new().content(expected_content.clone()),
2803                )),
2804                cx,
2805            );
2806        });
2807
2808        let response = response_rx
2809            .recv()
2810            .await
2811            .expect("fake auth flow should receive elicitation response");
2812        assert_eq!(
2813            response.action,
2814            acp::ElicitationAction::Accept(
2815                acp::ElicitationAcceptAction::new().content(expected_content)
2816            )
2817        );
2818        auth_task.await.expect("auth should complete");
2819    }
2820
2821    #[gpui::test]
2822    async fn request_scoped_url_elicitation_completion_after_create_is_observed(
2823        cx: &mut gpui::TestAppContext,
2824    ) {
2825        init_feature_flags_test(cx);
2826        cx.update(|cx| {
2827            cx.update_flags(false, vec![AcpBetaFeatureFlag::NAME.to_string()]);
2828        });
2829
2830        let fs = fs::FakeFs::new(cx.executor());
2831        fs.insert_tree("/", serde_json::json!({ "a": {} })).await;
2832        let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await;
2833
2834        let request_id = acp::RequestId::Number(1);
2835        let url_elicitation_id = acp::ElicitationId::new("auth-url");
2836        let (harness, response_rx) =
2837            test_support::connect_fake_acp_connection_with_auth_elicitation_completion(
2838                project,
2839                acp::CreateElicitationRequest::new(
2840                    acp::ElicitationUrlMode::new(
2841                        acp::ElicitationRequestScope::new(request_id.clone()),
2842                        url_elicitation_id.clone(),
2843                        "https://auth.example.com/device",
2844                    ),
2845                    "Authorize Zed in your browser",
2846                ),
2847                acp::CompleteElicitationNotification::new(url_elicitation_id),
2848                cx,
2849            )
2850            .await;
2851        let connection = harness.connection.clone();
2852        let auth_task =
2853            cx.update(|cx| connection.authenticate(acp::AuthMethodId::new("login"), cx));
2854        cx.run_until_parked();
2855
2856        let response = response_rx
2857            .recv()
2858            .await
2859            .expect("fake auth flow should receive elicitation response");
2860        assert_eq!(
2861            response.action,
2862            acp::ElicitationAction::Accept(acp::ElicitationAcceptAction::new())
2863        );
2864
2865        let store = connection
2866            .request_elicitations()
2867            .expect("ACP connections expose request-scoped elicitations");
2868        store.read_with(cx, |store, _| {
2869            let [elicitation] = store.elicitations() else {
2870                panic!(
2871                    "expected one request-scoped elicitation, got {:?}",
2872                    store.elicitations()
2873                );
2874            };
2875            let acp::ElicitationScope::Request(scope) = elicitation.request.scope() else {
2876                panic!("expected request-scoped elicitation");
2877            };
2878            assert_eq!(scope.request_id, request_id);
2879            assert!(matches!(
2880                elicitation.status,
2881                acp_thread::ElicitationStatus::Completed
2882            ));
2883        });
2884
2885        auth_task.await.expect("auth should complete");
2886    }
2887
2888    #[gpui::test]
2889    async fn request_scoped_elicitation_ignores_open_sessions(cx: &mut gpui::TestAppContext) {
2890        init_feature_flags_test(cx);
2891        cx.update(|cx| {
2892            cx.update_flags(false, vec![AcpBetaFeatureFlag::NAME.to_string()]);
2893        });
2894
2895        let fs = fs::FakeFs::new(cx.executor());
2896        fs.insert_tree("/", serde_json::json!({ "a": {} })).await;
2897        let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await;
2898
2899        let request_id = acp::RequestId::Number(1);
2900        let (harness, response_rx) =
2901            test_support::connect_fake_acp_connection_with_auth_elicitation(
2902                project.clone(),
2903                acp::CreateElicitationRequest::new(
2904                    acp::ElicitationFormMode::new(
2905                        acp::ElicitationRequestScope::new(request_id.clone()),
2906                        acp::ElicitationSchema::new().string("name", true),
2907                    ),
2908                    "Provide a name",
2909                ),
2910                cx,
2911            )
2912            .await;
2913        let connection = harness.connection.clone();
2914        let work_dirs = util::path_list::PathList::new(&[std::path::Path::new("/a")]);
2915
2916        let first_thread = cx
2917            .update(|cx| {
2918                connection.clone().load_session(
2919                    acp::SessionId::new("session-1"),
2920                    project.clone(),
2921                    work_dirs.clone(),
2922                    None,
2923                    cx,
2924                )
2925            })
2926            .await
2927            .expect("first load_session should succeed");
2928        let second_thread = cx
2929            .update(|cx| {
2930                connection.clone().load_session(
2931                    acp::SessionId::new("session-2"),
2932                    project,
2933                    work_dirs,
2934                    None,
2935                    cx,
2936                )
2937            })
2938            .await
2939            .expect("second load_session should succeed");
2940        cx.run_until_parked();
2941        assert_eq!(
2942            connection.sessions.borrow().len(),
2943            2,
2944            "test setup should have multiple open sessions"
2945        );
2946
2947        let auth_task =
2948            cx.update(|cx| connection.authenticate(acp::AuthMethodId::new("login"), cx));
2949        cx.run_until_parked();
2950
2951        let store = connection
2952            .request_elicitations()
2953            .expect("ACP connections expose request-scoped elicitations");
2954        let elicitation_id = store.read_with(cx, |store, _| {
2955            let [elicitation] = store.elicitations() else {
2956                panic!(
2957                    "expected one request-scoped elicitation, got {:?}",
2958                    store.elicitations()
2959                );
2960            };
2961            let acp::ElicitationScope::Request(scope) = elicitation.request.scope() else {
2962                panic!("expected request-scoped elicitation");
2963            };
2964            assert_eq!(scope.request_id, request_id);
2965            elicitation.id.clone()
2966        });
2967
2968        for thread in [first_thread, second_thread] {
2969            thread.read_with(cx, |thread, _| {
2970                assert!(
2971                    thread.entries().iter().all(|entry| !matches!(
2972                        entry,
2973                        acp_thread::AgentThreadEntry::Elicitation(_)
2974                    )),
2975                    "request-scoped elicitation should not be inserted into a session thread"
2976                );
2977            });
2978        }
2979
2980        store.update(cx, |store, cx| {
2981            store.respond_to_elicitation(
2982                &elicitation_id,
2983                acp::CreateElicitationResponse::new(acp::ElicitationAction::Decline),
2984                cx,
2985            );
2986        });
2987
2988        let response = response_rx
2989            .recv()
2990            .await
2991            .expect("fake auth flow should receive elicitation response");
2992        assert_eq!(response.action, acp::ElicitationAction::Decline);
2993        auth_task.await.expect("auth should complete");
2994    }
2995
2996    #[test]
2997    fn cursor_client_capabilities_include_parameterized_model_picker_meta() {
2998        let capabilities = client_capabilities_for_agent(&AgentId::new(CURSOR_ID));
2999        let meta = capabilities
3000            .meta
3001            .expect("expected client capabilities meta");
3002
3003        assert_eq!(
3004            meta.get(PARAMETERIZED_MODEL_PICKER_META_KEY),
3005            Some(&serde_json::json!(true))
3006        );
3007        assert_eq!(meta.get("terminal_output"), Some(&serde_json::json!(true)));
3008        assert_eq!(meta.get("terminal-auth"), Some(&serde_json::json!(true)));
3009    }
3010
3011    #[test]
3012    fn non_cursor_client_capabilities_do_not_include_parameterized_model_picker_meta() {
3013        let capabilities = client_capabilities_for_agent(&AgentId::new("codex-acp"));
3014        let meta = capabilities
3015            .meta
3016            .expect("expected client capabilities meta");
3017
3018        assert!(!meta.contains_key(PARAMETERIZED_MODEL_PICKER_META_KEY));
3019    }
3020
3021    #[test]
3022    fn client_capabilities_include_boolean_config_options() {
3023        let capabilities = client_capabilities_for_agent(&AgentId::new("codex-acp"));
3024
3025        assert!(
3026            capabilities
3027                .session
3028                .and_then(|session| session.config_options)
3029                .and_then(|config_options| config_options.boolean)
3030                .is_some()
3031        );
3032    }
3033
3034    #[test]
3035    fn terminal_auth_task_builds_spawn_from_prebuilt_command() {
3036        let command = AgentServerCommand {
3037            path: "/path/to/agent".into(),
3038            args: vec!["--acp".into(), "--verbose".into(), "/auth".into()],
3039            env: Some(HashMap::from_iter([
3040                ("BASE".into(), "1".into()),
3041                ("SHARED".into(), "override".into()),
3042                ("EXTRA".into(), "2".into()),
3043            ])),
3044        };
3045        let method = acp::AuthMethodTerminal::new("login", "Login");
3046
3047        let task = terminal_auth_task(&command, &AgentId::new("test-agent"), &method);
3048
3049        assert_eq!(task.command.as_deref(), Some("/path/to/agent"));
3050        assert_eq!(task.args, vec!["--acp", "--verbose", "/auth"]);
3051        assert_eq!(
3052            task.env,
3053            HashMap::from_iter([
3054                ("BASE".into(), "1".into()),
3055                ("SHARED".into(), "override".into()),
3056                ("EXTRA".into(), "2".into()),
3057            ])
3058        );
3059        assert_eq!(task.label, "Login");
3060        assert_eq!(task.command_label, "Login");
3061    }
3062
3063    #[test]
3064    fn legacy_terminal_auth_task_parses_meta_and_retries_session() {
3065        let method_id = acp::AuthMethodId::new("legacy-login");
3066        let method = acp::AuthMethod::Agent(
3067            acp::AuthMethodAgent::new(method_id.clone(), "Login").meta(acp::Meta::from_iter([(
3068                "terminal-auth".to_string(),
3069                serde_json::json!({
3070                    "label": "legacy /auth",
3071                    "command": "legacy-agent",
3072                    "args": ["auth", "--interactive"],
3073                    "env": {
3074                        "AUTH_MODE": "interactive",
3075                    },
3076                }),
3077            )])),
3078        );
3079
3080        let task = meta_terminal_auth_task(&AgentId::new("test-agent"), &method_id, &method)
3081            .expect("expected legacy terminal auth task");
3082
3083        assert_eq!(task.id.0, "external-agent-test-agent-legacy-login-login");
3084        assert_eq!(task.command.as_deref(), Some("legacy-agent"));
3085        assert_eq!(task.args, vec!["auth", "--interactive"]);
3086        assert_eq!(
3087            task.env,
3088            HashMap::from_iter([("AUTH_MODE".into(), "interactive".into())])
3089        );
3090        assert_eq!(task.label, "legacy /auth");
3091    }
3092
3093    #[test]
3094    fn legacy_terminal_auth_task_returns_none_for_invalid_meta() {
3095        let method_id = acp::AuthMethodId::new("legacy-login");
3096        let method = acp::AuthMethod::Agent(
3097            acp::AuthMethodAgent::new(method_id.clone(), "Login").meta(acp::Meta::from_iter([(
3098                "terminal-auth".to_string(),
3099                serde_json::json!({
3100                    "label": "legacy /auth",
3101                }),
3102            )])),
3103        );
3104
3105        assert!(
3106            meta_terminal_auth_task(&AgentId::new("test-agent"), &method_id, &method).is_none()
3107        );
3108    }
3109
3110    #[test]
3111    fn first_class_terminal_auth_takes_precedence_over_legacy_meta() {
3112        let method_id = acp::AuthMethodId::new("login");
3113        let method = acp::AuthMethod::Terminal(
3114            acp::AuthMethodTerminal::new(method_id, "Login")
3115                .args(vec!["/auth".into()])
3116                .env(std::collections::HashMap::from_iter([(
3117                    "AUTH_MODE".into(),
3118                    "first-class".into(),
3119                )]))
3120                .meta(acp::Meta::from_iter([(
3121                    "terminal-auth".to_string(),
3122                    serde_json::json!({
3123                        "label": "legacy /auth",
3124                        "command": "legacy-agent",
3125                        "args": ["legacy-auth"],
3126                        "env": {
3127                            "AUTH_MODE": "legacy",
3128                        },
3129                    }),
3130                )])),
3131        );
3132
3133        let command = AgentServerCommand {
3134            path: "/path/to/agent".into(),
3135            args: vec!["--acp".into(), "/auth".into()],
3136            env: Some(HashMap::from_iter([
3137                ("BASE".into(), "1".into()),
3138                ("AUTH_MODE".into(), "first-class".into()),
3139            ])),
3140        };
3141
3142        let task = match &method {
3143            acp::AuthMethod::Terminal(terminal) => {
3144                terminal_auth_task(&command, &AgentId::new("test-agent"), terminal)
3145            }
3146            _ => unreachable!(),
3147        };
3148
3149        assert_eq!(task.command.as_deref(), Some("/path/to/agent"));
3150        assert_eq!(task.args, vec!["--acp", "/auth"]);
3151        assert_eq!(
3152            task.env,
3153            HashMap::from_iter([
3154                ("BASE".into(), "1".into()),
3155                ("AUTH_MODE".into(), "first-class".into()),
3156            ])
3157        );
3158        assert_eq!(task.label, "Login");
3159    }
3160
3161    #[test]
3162    fn trailing_stderr_only_uses_final_stderr_block() {
3163        let debug_log = AcpDebugLog::default();
3164        debug_log.record_line(AcpDebugMessageDirection::Stderr, "stale stderr");
3165        debug_log.record_line(
3166            AcpDebugMessageDirection::Incoming,
3167            r#"{"method":"initialized"}"#,
3168        );
3169
3170        assert_eq!(debug_log.trailing_stderr(), None);
3171
3172        debug_log.record_line(AcpDebugMessageDirection::Stderr, "recent stderr");
3173        assert_eq!(
3174            debug_log.trailing_stderr().as_deref(),
3175            Some("recent stderr")
3176        );
3177    }
3178
3179    #[test]
3180    fn debug_log_records_each_json_rpc_batch_entry() {
3181        let debug_log = AcpDebugLog::default();
3182        debug_log.record_line(
3183            AcpDebugMessageDirection::Incoming,
3184            r#"{"jsonrpc":"2.0","method":"legacy/update"}"#,
3185        );
3186        debug_log.record_line(
3187            AcpDebugMessageDirection::Incoming,
3188            r#"[
3189                {"jsonrpc":"2.0","method":"session/update","params":{"value":1}},
3190                null,
3191                [{"jsonrpc":"2.0","method":"nested/update"}],
3192                {"jsonrpc":"2.0","id":1,"method":"session/one","params":{"value":2}},
3193                {"jsonrpc":"2.0","id":{"invalid":true},"method":"invalid/id"}
3194            ]"#,
3195        );
3196        debug_log.record_line(
3197            AcpDebugMessageDirection::Outgoing,
3198            r#"[
3199                {"jsonrpc":"2.0","id":1,"result":{"accepted":true}},
3200                {"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"Invalid Request"}}
3201            ]"#,
3202        );
3203
3204        let (messages, _receiver) = debug_log.subscribe();
3205        let mut messages = messages.iter();
3206
3207        assert!(matches!(
3208            messages.next(),
3209            Some(AcpDebugMessage {
3210                direction: AcpDebugMessageDirection::Incoming,
3211                message: AcpDebugMessageContent::Notification { method, .. },
3212            }) if method.as_ref() == "legacy/update"
3213        ));
3214        assert!(matches!(
3215            messages.next(),
3216            Some(AcpDebugMessage {
3217                direction: AcpDebugMessageDirection::Incoming,
3218                message: AcpDebugMessageContent::Notification { method, .. },
3219            }) if method.as_ref() == "session/update"
3220        ));
3221        assert!(matches!(
3222            messages.next(),
3223            Some(AcpDebugMessage {
3224                direction: AcpDebugMessageDirection::Incoming,
3225                message: AcpDebugMessageContent::Request { id, method, .. },
3226            }) if id == &acp::RequestId::Number(1) && method.as_ref() == "session/one"
3227        ));
3228        assert!(matches!(
3229            messages.next(),
3230            Some(AcpDebugMessage {
3231                direction: AcpDebugMessageDirection::Outgoing,
3232                message: AcpDebugMessageContent::Response {
3233                    id,
3234                    result: Ok(Some(_)),
3235                },
3236            }) if id == &acp::RequestId::Number(1)
3237        ));
3238        assert!(matches!(
3239            messages.next(),
3240            Some(AcpDebugMessage {
3241                direction: AcpDebugMessageDirection::Outgoing,
3242                message: AcpDebugMessageContent::Response {
3243                    id,
3244                    result: Err(_),
3245                },
3246            }) if id == &acp::RequestId::Null
3247        ));
3248        assert!(messages.next().is_none());
3249    }
3250
3251    #[test]
3252    fn session_directories_use_ordered_paths_when_supported() {
3253        let work_dirs = PathList::new(&[
3254            std::path::PathBuf::from("/workspace-b"),
3255            std::path::PathBuf::from("/workspace-a"),
3256            std::path::PathBuf::from("/workspace-c"),
3257        ]);
3258
3259        let directories =
3260            session_directories_from_work_dirs(&work_dirs, true).expect("work dirs should convert");
3261
3262        assert_eq!(
3263            directories,
3264            SessionDirectories {
3265                cwd: std::path::PathBuf::from("/workspace-b"),
3266                additional_directories: vec![
3267                    std::path::PathBuf::from("/workspace-a"),
3268                    std::path::PathBuf::from("/workspace-c")
3269                ],
3270            }
3271        );
3272
3273        let session_id = acp::SessionId::new("session-1");
3274        let new_session_request = directories.clone().into_new_session_request(Vec::new());
3275        let load_session_request = directories
3276            .clone()
3277            .into_load_session_request(session_id.clone(), Vec::new());
3278        let resume_session_request =
3279            directories.into_resume_session_request(session_id, Vec::new());
3280
3281        assert_eq!(
3282            new_session_request.cwd,
3283            std::path::PathBuf::from("/workspace-b")
3284        );
3285        assert_eq!(
3286            new_session_request.additional_directories,
3287            vec![
3288                std::path::PathBuf::from("/workspace-a"),
3289                std::path::PathBuf::from("/workspace-c")
3290            ]
3291        );
3292        assert_eq!(
3293            load_session_request.additional_directories,
3294            new_session_request.additional_directories
3295        );
3296        assert_eq!(
3297            resume_session_request.additional_directories,
3298            new_session_request.additional_directories
3299        );
3300    }
3301
3302    #[test]
3303    fn session_directories_drop_additional_paths_when_unsupported() {
3304        let work_dirs = PathList::new(&[
3305            std::path::PathBuf::from("/workspace-b"),
3306            std::path::PathBuf::from("/workspace-a"),
3307        ]);
3308
3309        let directories = session_directories_from_work_dirs(&work_dirs, false)
3310            .expect("work dirs should convert");
3311
3312        assert_eq!(
3313            directories,
3314            SessionDirectories {
3315                cwd: std::path::PathBuf::from("/workspace-b"),
3316                additional_directories: Vec::new(),
3317            }
3318        );
3319    }
3320
3321    #[test]
3322    fn session_info_work_dirs_preserve_cwd_then_additional_directories() {
3323        let work_dirs = work_dirs_from_session_info(
3324            std::path::PathBuf::from("/workspace-b"),
3325            vec![
3326                std::path::PathBuf::from("/workspace-a"),
3327                std::path::PathBuf::from("/workspace-c"),
3328            ],
3329        );
3330
3331        assert_eq!(
3332            work_dirs.ordered_paths().cloned().collect::<Vec<_>>(),
3333            vec![
3334                std::path::PathBuf::from("/workspace-b"),
3335                std::path::PathBuf::from("/workspace-a"),
3336                std::path::PathBuf::from("/workspace-c"),
3337            ]
3338        );
3339    }
3340
3341    #[test]
3342    fn session_info_work_dirs_deduplicate_cwd_and_additional_directories() {
3343        let work_dirs = work_dirs_from_session_info(
3344            std::path::PathBuf::from("/workspace-b"),
3345            vec![
3346                std::path::PathBuf::from("/workspace-a"),
3347                std::path::PathBuf::from("/workspace-b"),
3348                std::path::PathBuf::from("/workspace-a"),
3349                std::path::PathBuf::from("/workspace-c"),
3350            ],
3351        );
3352
3353        assert_eq!(
3354            work_dirs.ordered_paths().cloned().collect::<Vec<_>>(),
3355            vec![
3356                std::path::PathBuf::from("/workspace-b"),
3357                std::path::PathBuf::from("/workspace-a"),
3358                std::path::PathBuf::from("/workspace-c"),
3359            ]
3360        );
3361    }
3362
3363    #[gpui::test]
3364    async fn session_list_includes_additional_directories_in_work_dirs(
3365        cx: &mut gpui::TestAppContext,
3366    ) {
3367        let connection = connect_session_list_test_agent(
3368            vec![
3369                acp::SessionInfo::new("session-1", "/workspace-b").additional_directories(vec![
3370                    std::path::PathBuf::from("/workspace-a"),
3371                    std::path::PathBuf::from("/workspace-b"),
3372                    std::path::PathBuf::from("/workspace-a"),
3373                    std::path::PathBuf::from("/workspace-c"),
3374                ]),
3375            ],
3376            cx,
3377        )
3378        .await;
3379        let session_list = AcpSessionList::new(connection, false);
3380
3381        let response = cx
3382            .update(|cx| session_list.list_sessions(AgentSessionListRequest::default(), cx))
3383            .await
3384            .expect("session list should load");
3385        let session = response
3386            .sessions
3387            .first()
3388            .expect("session list should include the returned session");
3389        let work_dirs = session
3390            .work_dirs
3391            .as_ref()
3392            .expect("session should include work dirs");
3393
3394        assert_eq!(
3395            work_dirs.ordered_paths().cloned().collect::<Vec<_>>(),
3396            vec![
3397                std::path::PathBuf::from("/workspace-b"),
3398                std::path::PathBuf::from("/workspace-a"),
3399                std::path::PathBuf::from("/workspace-c"),
3400            ]
3401        );
3402    }
3403
3404    async fn connect_session_list_test_agent(
3405        sessions: Vec<acp::SessionInfo>,
3406        cx: &mut gpui::TestAppContext,
3407    ) -> ConnectionTo<Agent> {
3408        let (client_transport, agent_transport) = agent_client_protocol::Channel::duplex();
3409        let sessions = Arc::new(sessions);
3410
3411        cx.background_spawn(
3412            Agent
3413                .builder()
3414                .name("list-test-agent")
3415                .on_receive_request(
3416                    {
3417                        let sessions = sessions.clone();
3418                        async move |_request: acp::ListSessionsRequest, responder, _cx| {
3419                            responder.respond(acp::ListSessionsResponse::new((*sessions).clone()))
3420                        }
3421                    },
3422                    agent_client_protocol::on_receive_request!(),
3423                )
3424                .connect_to(agent_transport),
3425        )
3426        .detach();
3427
3428        let (connection_tx, connection_rx) = futures::channel::oneshot::channel();
3429        cx.background_spawn(Client.builder().name("list-test-client").connect_with(
3430            client_transport,
3431            move |connection: ConnectionTo<Agent>| async move {
3432                connection_tx.send(connection).ok();
3433                futures::future::pending::<Result<(), acp::Error>>().await
3434            },
3435        ))
3436        .detach();
3437
3438        connection_rx
3439            .await
3440            .expect("failed to receive ACP connection")
3441    }
3442
3443    #[gpui::test]
3444    async fn additional_directories_support_respects_agent_capability(
3445        cx: &mut gpui::TestAppContext,
3446    ) {
3447        cx.update(|cx| {
3448            let store = settings::SettingsStore::test(cx);
3449            cx.set_global(store);
3450        });
3451
3452        let fs = fs::FakeFs::new(cx.executor());
3453        fs.insert_tree("/", serde_json::json!({ "a": {}, "b": {} }))
3454            .await;
3455        let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await;
3456        let mut harness = test_support::connect_fake_acp_connection(project, cx).await;
3457
3458        let work_dirs = PathList::new(&[
3459            std::path::PathBuf::from("/workspace-b"),
3460            std::path::PathBuf::from("/workspace-a"),
3461        ]);
3462
3463        let missing_capability = harness
3464            .connection
3465            .session_directories_from_work_dirs(&work_dirs)
3466            .expect("work dirs should convert");
3467        assert!(missing_capability.additional_directories.is_empty());
3468
3469        Rc::get_mut(&mut harness.connection)
3470            .expect("test harness should own the only ACP connection handle")
3471            .agent_capabilities
3472            .session_capabilities
3473            .additional_directories = Some(acp::SessionAdditionalDirectoriesCapabilities::new());
3474
3475        let supported = harness
3476            .connection
3477            .session_directories_from_work_dirs(&work_dirs)
3478            .expect("work dirs should convert");
3479        assert_eq!(
3480            supported,
3481            SessionDirectories {
3482                cwd: std::path::PathBuf::from("/workspace-b"),
3483                additional_directories: vec![std::path::PathBuf::from("/workspace-a")],
3484            }
3485        );
3486    }
3487
3488    async fn connect_session_delete_test_agent(
3489        deleted_sessions: Arc<std::sync::Mutex<Vec<acp::SessionId>>>,
3490        cx: &mut gpui::TestAppContext,
3491    ) -> ConnectionTo<Agent> {
3492        let (client_transport, agent_transport) = agent_client_protocol::Channel::duplex();
3493
3494        cx.background_spawn(
3495            Agent
3496                .builder()
3497                .name("delete-test-agent")
3498                .on_receive_request(
3499                    {
3500                        let deleted_sessions = deleted_sessions.clone();
3501                        async move |request: acp::DeleteSessionRequest, responder, _cx| {
3502                            deleted_sessions
3503                                .lock()
3504                                .expect("deleted sessions lock should not be poisoned")
3505                                .push(request.session_id);
3506                            responder.respond(acp::DeleteSessionResponse::default())
3507                        }
3508                    },
3509                    agent_client_protocol::on_receive_request!(),
3510                )
3511                .connect_to(agent_transport),
3512        )
3513        .detach();
3514
3515        let (connection_tx, connection_rx) = futures::channel::oneshot::channel();
3516        cx.background_spawn(Client.builder().name("delete-test-client").connect_with(
3517            client_transport,
3518            move |connection: ConnectionTo<Agent>| async move {
3519                connection_tx.send(connection).ok();
3520                futures::future::pending::<Result<(), acp::Error>>().await
3521            },
3522        ))
3523        .detach();
3524
3525        connection_rx
3526            .await
3527            .expect("failed to receive ACP connection")
3528    }
3529
3530    #[gpui::test]
3531    async fn settings_changes_refresh_active_connection_defaults(cx: &mut gpui::TestAppContext) {
3532        cx.update(|cx| {
3533            let store = settings::SettingsStore::test(cx);
3534            cx.set_global(store);
3535        });
3536
3537        let fs = fs::FakeFs::new(cx.executor());
3538        fs.insert_tree("/", serde_json::json!({ "a": {} })).await;
3539        let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await;
3540        let harness = test_support::connect_fake_acp_connection(project, cx).await;
3541
3542        cx.update(|cx| {
3543            AllAgentServersSettings::override_global(
3544                AllAgentServersSettings(HashMap::from_iter([(
3545                    "test".to_string(),
3546                    settings::CustomAgentServerSettings::Custom {
3547                        path: PathBuf::from("test-agent"),
3548                        args: Vec::new(),
3549                        env: HashMap::default(),
3550                        default_mode: Some("manual".to_string()),
3551                        default_config_options: HashMap::from_iter([(
3552                            "mode".to_string(),
3553                            AgentConfigOptionValue::from("manual"),
3554                        )]),
3555                        favorite_config_option_values: HashMap::default(),
3556                    }
3557                    .into(),
3558                )])),
3559                cx,
3560            );
3561        });
3562        cx.run_until_parked();
3563
3564        assert_eq!(
3565            harness.connection.defaults.mode(),
3566            Some(acp::SessionModeId::new("manual"))
3567        );
3568        assert_eq!(
3569            harness
3570                .connection
3571                .defaults
3572                .config_option("mode")
3573                .as_ref()
3574                .and_then(AgentConfigOptionValue::as_value_id),
3575            Some("manual"),
3576        );
3577
3578        cx.update(|cx| {
3579            AllAgentServersSettings::override_global(
3580                AllAgentServersSettings(HashMap::default()),
3581                cx,
3582            );
3583        });
3584        cx.run_until_parked();
3585
3586        assert_eq!(harness.connection.defaults.mode(), None);
3587        assert_eq!(harness.connection.defaults.config_option("mode"), None);
3588    }
3589
3590    #[gpui::test]
3591    async fn default_config_options_apply_boolean_defaults(cx: &mut gpui::TestAppContext) {
3592        let (connection, set_config_requests) = connect_config_defaults_test_agent(cx).await;
3593        connection.defaults.set(
3594            None,
3595            HashMap::from_iter([(
3596                "web_search".to_string(),
3597                AgentConfigOptionValue::Boolean(true),
3598            )]),
3599        );
3600        let config_options = Rc::new(RefCell::new(vec![acp::SessionConfigOption::boolean(
3601            "web_search",
3602            "Web Search",
3603            false,
3604        )]));
3605
3606        let mut async_cx = cx.to_async();
3607        connection.apply_default_config_options(
3608            &acp::SessionId::new("session-config-defaults"),
3609            &config_options,
3610            &mut async_cx,
3611        );
3612        drop(async_cx);
3613        cx.run_until_parked();
3614
3615        let requests = set_config_requests
3616            .lock()
3617            .expect("set config requests mutex poisoned");
3618        assert_eq!(requests.len(), 1);
3619        assert_eq!(
3620            requests[0].config_id,
3621            acp::SessionConfigId::new("web_search")
3622        );
3623        assert_eq!(
3624            requests[0].value,
3625            acp::SessionConfigOptionValue::boolean(true)
3626        );
3627
3628        let options = config_options.borrow();
3629        assert!(
3630            matches!(&options[0].kind, acp::SessionConfigKind::Boolean(boolean) if boolean.current_value)
3631        );
3632    }
3633
3634    async fn connect_config_defaults_test_agent(
3635        cx: &mut gpui::TestAppContext,
3636    ) -> (
3637        AcpConnection,
3638        Arc<Mutex<Vec<acp::SetSessionConfigOptionRequest>>>,
3639    ) {
3640        let set_config_requests = Arc::new(Mutex::new(Vec::new()));
3641        let (client_transport, agent_transport) = agent_client_protocol::Channel::duplex();
3642
3643        cx.background_spawn(
3644            Agent
3645                .builder()
3646                .name("config-defaults-test-agent")
3647                .on_receive_request(
3648                    {
3649                        let set_config_requests = set_config_requests.clone();
3650                        async move |req: acp::SetSessionConfigOptionRequest, responder, _cx| {
3651                            set_config_requests
3652                                .lock()
3653                                .expect("set config requests mutex poisoned")
3654                                .push(req);
3655
3656                            responder.respond(acp::SetSessionConfigOptionResponse::new(Vec::new()))
3657                        }
3658                    },
3659                    agent_client_protocol::on_receive_request!(),
3660                )
3661                .connect_to(agent_transport),
3662        )
3663        .detach();
3664
3665        let (connection_tx, connection_rx) = futures::channel::oneshot::channel();
3666        let client_io_task = cx.background_spawn(async move {
3667            Client
3668                .builder()
3669                .name("config-defaults-test-client")
3670                .connect_with(
3671                    client_transport,
3672                    move |connection: ConnectionTo<Agent>| async move {
3673                        connection_tx.send(connection).ok();
3674                        futures::future::pending::<Result<(), acp::Error>>().await
3675                    },
3676                )
3677                .await
3678                .ok();
3679        });
3680
3681        let client_conn = connection_rx
3682            .await
3683            .expect("failed to receive ACP connection");
3684        let sessions = Rc::new(RefCell::new(HashMap::default()));
3685
3686        let connection = cx.update(|cx| {
3687            let request_elicitations = cx.new(|_| ElicitationStore::default());
3688            AcpConnection::new_for_test(
3689                client_conn,
3690                sessions,
3691                acp::AgentCapabilities::default(),
3692                request_elicitations,
3693                WeakEntity::new_invalid(),
3694                client_io_task,
3695                Task::ready(()),
3696                cx,
3697            )
3698        });
3699
3700        (connection, set_config_requests)
3701    }
3702
3703    #[gpui::test]
3704    async fn session_list_delete_sends_session_delete_when_supported(
3705        cx: &mut gpui::TestAppContext,
3706    ) {
3707        let deleted_sessions = Arc::new(std::sync::Mutex::new(Vec::new()));
3708        let connection = connect_session_delete_test_agent(deleted_sessions.clone(), cx).await;
3709        let session_list = AcpSessionList::new(connection, true);
3710        let session_id = acp::SessionId::new("session-to-delete");
3711
3712        cx.update(|cx| session_list.delete_session(&session_id, cx))
3713            .await
3714            .expect("delete_session failed");
3715
3716        assert_eq!(
3717            *deleted_sessions
3718                .lock()
3719                .expect("deleted sessions lock should not be poisoned"),
3720            vec![session_id]
3721        );
3722    }
3723
3724    #[gpui::test]
3725    async fn session_list_delete_does_not_send_when_unsupported(cx: &mut gpui::TestAppContext) {
3726        let deleted_sessions = Arc::new(std::sync::Mutex::new(Vec::new()));
3727        let connection = connect_session_delete_test_agent(deleted_sessions.clone(), cx).await;
3728        let session_list = AcpSessionList::new(connection, false);
3729        let session_id = acp::SessionId::new("session-to-delete");
3730
3731        let error = cx
3732            .update(|cx| session_list.delete_session(&session_id, cx))
3733            .await
3734            .expect_err("delete_session should fail when unsupported");
3735
3736        assert!(
3737            error.to_string().contains("delete_session not supported"),
3738            "unexpected error: {error}"
3739        );
3740        assert!(
3741            deleted_sessions
3742                .lock()
3743                .expect("deleted sessions lock should not be poisoned")
3744                .is_empty()
3745        );
3746    }
3747
3748    #[cfg(not(windows))]
3749    #[gpui::test]
3750    async fn startup_returns_error_when_agent_exits_before_initialization(
3751        cx: &mut gpui::TestAppContext,
3752    ) {
3753        cx.update(|cx| {
3754            let store = settings::SettingsStore::test(cx);
3755            cx.set_global(store);
3756        });
3757        cx.executor().allow_parking();
3758
3759        let temp_dir = tempfile::tempdir().unwrap();
3760        let project = project::Project::example([temp_dir.path()], &mut cx.to_async()).await;
3761        let agent_server_store =
3762            project.read_with(cx, |project, _| project.agent_server_store().downgrade());
3763        let command = AgentServerCommand {
3764            path: "/bin/sh".into(),
3765            args: vec![
3766                "-c".into(),
3767                r#"printf '%s\n' 'npm error code ETARGET' 'npm error notarget No matching version found for @agentclientprotocol/claude-agent-acp@0.32.0 with a date before 4/28/2026, 12:11:38 PM.' >&2; exit 1"#.into(),
3768            ],
3769            env: None,
3770        };
3771
3772        let mut async_cx = cx.to_async();
3773        let startup = AcpConnection::stdio(
3774            AgentId::new("test-agent"),
3775            project,
3776            command,
3777            agent_server_store,
3778            None,
3779            HashMap::default(),
3780            &mut async_cx,
3781        )
3782        .fuse();
3783        let timeout = cx
3784            .background_executor
3785            .timer(std::time::Duration::from_secs(5))
3786            .fuse();
3787        futures::pin_mut!(startup, timeout);
3788
3789        let result = futures::select! {
3790            result = startup => result,
3791            _ = timeout => panic!("timed out waiting for failed ACP startup"),
3792        };
3793
3794        let Err(error) = result else {
3795            panic!("expected ACP startup to fail");
3796        };
3797        let load_error = error
3798            .downcast::<LoadError>()
3799            .expect("startup failure should preserve the typed load error");
3800        match load_error {
3801            LoadError::Exited { status, .. } => {
3802                assert!(!status.success(), "expected non-zero exit status");
3803            }
3804            error => panic!("expected exited load error, got: {error:?}"),
3805        };
3806    }
3807
3808    async fn connect_fake_agent(
3809        cx: &mut gpui::TestAppContext,
3810    ) -> (
3811        Rc<AcpConnection>,
3812        Entity<project::Project>,
3813        Arc<AtomicUsize>,
3814        Arc<AtomicUsize>,
3815        Arc<std::sync::Mutex<Vec<acp::SessionUpdate>>>,
3816        Arc<std::sync::Mutex<Option<async_channel::Receiver<()>>>>,
3817        Task<anyhow::Result<()>>,
3818    ) {
3819        cx.update(|cx| {
3820            let store = settings::SettingsStore::test(cx);
3821            cx.set_global(store);
3822        });
3823
3824        let fs = fs::FakeFs::new(cx.executor());
3825        fs.insert_tree("/", serde_json::json!({ "a": {} })).await;
3826        let project = project::Project::test(fs, [std::path::Path::new("/a")], cx).await;
3827
3828        let load_count = Arc::new(AtomicUsize::new(0));
3829        let close_count = Arc::new(AtomicUsize::new(0));
3830        let load_session_updates: Arc<std::sync::Mutex<Vec<acp::SessionUpdate>>> =
3831            Arc::new(std::sync::Mutex::new(Vec::new()));
3832        let load_session_gate: Arc<std::sync::Mutex<Option<async_channel::Receiver<()>>>> =
3833            Arc::new(std::sync::Mutex::new(None));
3834
3835        let (client_transport, agent_transport) = agent_client_protocol::Channel::duplex();
3836
3837        let sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>> =
3838            Rc::new(RefCell::new(HashMap::default()));
3839        let client_session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>> =
3840            Rc::new(RefCell::new(None));
3841
3842        // Build the fake agent side. It handles the requests issued by
3843        // `AcpConnection` during the test and tracks load/close counts.
3844        let agent_future = Agent
3845            .builder()
3846            .name("fake-agent")
3847            .on_receive_request(
3848                async move |req: acp::InitializeRequest, responder, _cx| {
3849                    responder.respond(
3850                        acp::InitializeResponse::new(req.protocol_version).agent_capabilities(
3851                            acp::AgentCapabilities::default()
3852                                .load_session(true)
3853                                .session_capabilities(
3854                                    acp::SessionCapabilities::default()
3855                                        .close(acp::SessionCloseCapabilities::new()),
3856                                ),
3857                        ),
3858                    )
3859                },
3860                agent_client_protocol::on_receive_request!(),
3861            )
3862            .on_receive_request(
3863                async move |_req: acp::AuthenticateRequest, responder, _cx| {
3864                    responder.respond(Default::default())
3865                },
3866                agent_client_protocol::on_receive_request!(),
3867            )
3868            .on_receive_request(
3869                async move |_req: acp::NewSessionRequest, responder, _cx| {
3870                    responder.respond(acp::NewSessionResponse::new(acp::SessionId::new("unused")))
3871                },
3872                agent_client_protocol::on_receive_request!(),
3873            )
3874            .on_receive_request(
3875                async move |_req: acp::PromptRequest, responder, _cx| {
3876                    responder.respond(acp::PromptResponse::new(acp::StopReason::EndTurn))
3877                },
3878                agent_client_protocol::on_receive_request!(),
3879            )
3880            .on_receive_request(
3881                {
3882                    let load_count = load_count.clone();
3883                    let load_session_updates = load_session_updates.clone();
3884                    let load_session_gate = load_session_gate.clone();
3885                    async move |req: acp::LoadSessionRequest, responder, cx| {
3886                        load_count.fetch_add(1, Ordering::SeqCst);
3887
3888                        // Simulate spec-compliant history replay: send
3889                        // notifications to the client before responding to the
3890                        // load request.
3891                        let updates = std::mem::take(
3892                            &mut *load_session_updates
3893                                .lock()
3894                                .expect("load_session_updates mutex poisoned"),
3895                        );
3896                        for update in updates {
3897                            cx.send_notification(acp::SessionNotification::new(
3898                                req.session_id.clone(),
3899                                update,
3900                            ))?;
3901                        }
3902
3903                        // If a gate was installed, park on it before responding
3904                        // so tests can interleave other work (e.g.
3905                        // `close_session`) with an in-flight load.
3906                        let gate = load_session_gate
3907                            .lock()
3908                            .expect("load_session_gate mutex poisoned")
3909                            .take();
3910                        if let Some(gate) = gate {
3911                            gate.recv().await.ok();
3912                        }
3913
3914                        responder.respond(acp::LoadSessionResponse::new())
3915                    }
3916                },
3917                agent_client_protocol::on_receive_request!(),
3918            )
3919            .on_receive_request(
3920                {
3921                    let close_count = close_count.clone();
3922                    async move |_req: acp::CloseSessionRequest, responder, _cx| {
3923                        close_count.fetch_add(1, Ordering::SeqCst);
3924                        responder.respond(acp::CloseSessionResponse::new())
3925                    }
3926                },
3927                agent_client_protocol::on_receive_request!(),
3928            )
3929            .on_receive_notification(
3930                async move |_notif: acp::CancelNotification, _cx| Ok(()),
3931                agent_client_protocol::on_receive_notification!(),
3932            )
3933            .connect_to(agent_transport);
3934
3935        let agent_io_task = cx.background_spawn(agent_future);
3936
3937        // Wire the production handler set into the fake client so inbound
3938        // requests/notifications from the fake agent reach the same
3939        // dispatcher that the real `stdio` path uses.
3940        let (dispatch_tx, dispatch_rx) = mpsc::unbounded::<ForegroundWork>();
3941
3942        let (connection_tx, connection_rx) = futures::channel::oneshot::channel();
3943        let client_future = connect_client_future(
3944            "zed-test",
3945            client_transport,
3946            dispatch_tx.clone(),
3947            connection_tx,
3948        );
3949        let client_io_task = cx.background_spawn(async move {
3950            client_future.await.ok();
3951        });
3952
3953        let client_conn: ConnectionTo<Agent> = connection_rx
3954            .await
3955            .expect("failed to receive ACP connection handle");
3956
3957        let response = client_conn
3958            .send_request(acp::InitializeRequest::new(ProtocolVersion::V1))
3959            .block_task()
3960            .await
3961            .expect("failed to initialize ACP connection");
3962
3963        let agent_capabilities = response.agent_capabilities;
3964
3965        let request_elicitations = cx.new(|_| ElicitationStore::default());
3966        let dispatch_context = ClientContext {
3967            sessions: sessions.clone(),
3968            session_list: client_session_list.clone(),
3969            request_elicitations: request_elicitations.clone(),
3970        };
3971        // `TestAppContext::spawn` hands out an `AsyncApp` by value, whereas the
3972        // production path uses `Context::spawn` which hands out `&mut AsyncApp`.
3973        // Bind the value-form to a local and take `&mut` of it to reuse the
3974        // same dispatch loop shape.
3975        let dispatch_task = cx.spawn({
3976            let mut dispatch_rx = dispatch_rx;
3977            move |cx| async move {
3978                let mut cx = cx;
3979                while let Some(work) = dispatch_rx.next().await {
3980                    work.run(&mut cx, &dispatch_context);
3981                }
3982            }
3983        });
3984
3985        let agent_server_store =
3986            project.read_with(cx, |project, _| project.agent_server_store().downgrade());
3987
3988        let connection = cx.update(|cx| {
3989            AcpConnection::new_for_test(
3990                client_conn,
3991                sessions,
3992                agent_capabilities,
3993                request_elicitations,
3994                agent_server_store,
3995                client_io_task,
3996                dispatch_task,
3997                cx,
3998            )
3999        });
4000
4001        let keep_agent_alive = cx.background_spawn(async move {
4002            agent_io_task.await.ok();
4003            anyhow::Ok(())
4004        });
4005
4006        (
4007            Rc::new(connection),
4008            project,
4009            load_count,
4010            close_count,
4011            load_session_updates,
4012            load_session_gate,
4013            keep_agent_alive,
4014        )
4015    }
4016
4017    #[gpui::test]
4018    async fn test_loaded_sessions_keep_state_until_last_close(cx: &mut gpui::TestAppContext) {
4019        let (
4020            connection,
4021            project,
4022            load_count,
4023            close_count,
4024            _load_session_updates,
4025            _load_session_gate,
4026            _keep_agent_alive,
4027        ) = connect_fake_agent(cx).await;
4028
4029        let session_id = acp::SessionId::new("session-1");
4030        let work_dirs = util::path_list::PathList::new(&[std::path::Path::new("/a")]);
4031
4032        // Load the same session twice concurrently — the second call should join
4033        // the pending task rather than issuing a second ACP load_session RPC.
4034        let first_load = cx.update(|cx| {
4035            connection.clone().load_session(
4036                session_id.clone(),
4037                project.clone(),
4038                work_dirs.clone(),
4039                None,
4040                cx,
4041            )
4042        });
4043        let second_load = cx.update(|cx| {
4044            connection.clone().load_session(
4045                session_id.clone(),
4046                project.clone(),
4047                work_dirs.clone(),
4048                None,
4049                cx,
4050            )
4051        });
4052
4053        let first_thread = first_load.await.expect("first load failed");
4054        let second_thread = second_load.await.expect("second load failed");
4055        cx.run_until_parked();
4056
4057        assert_eq!(
4058            first_thread.entity_id(),
4059            second_thread.entity_id(),
4060            "concurrent loads for the same session should share one AcpThread"
4061        );
4062        assert_eq!(
4063            load_count.load(Ordering::SeqCst),
4064            1,
4065            "underlying ACP load_session should be called exactly once for concurrent loads"
4066        );
4067
4068        // The session has ref_count 2. The first close should not send the ACP
4069        // close_session RPC — the session is still referenced.
4070        cx.update(|cx| connection.clone().close_session(&session_id, cx))
4071            .await
4072            .expect("first close failed");
4073
4074        assert_eq!(
4075            close_count.load(Ordering::SeqCst),
4076            0,
4077            "ACP close_session should not be sent while ref_count > 0"
4078        );
4079        assert!(
4080            connection.sessions.borrow().contains_key(&session_id),
4081            "session should still be tracked after first close"
4082        );
4083
4084        // The second close drops ref_count to 0 — now the ACP RPC must be sent.
4085        cx.update(|cx| connection.clone().close_session(&session_id, cx))
4086            .await
4087            .expect("second close failed");
4088        cx.run_until_parked();
4089
4090        assert_eq!(
4091            close_count.load(Ordering::SeqCst),
4092            1,
4093            "ACP close_session should be sent exactly once when ref_count reaches 0"
4094        );
4095        assert!(
4096            !connection.sessions.borrow().contains_key(&session_id),
4097            "session should be removed after final close"
4098        );
4099    }
4100
4101    // Regression test: per the ACP spec, an agent replays the entire conversation
4102    // history as `session/update` notifications *before* responding to the
4103    // `session/load` request. These notifications must be applied to the
4104    // reconstructed thread, not dropped because the session hasn't been
4105    // registered yet.
4106    #[gpui::test]
4107    async fn test_load_session_replays_notifications_sent_before_response(
4108        cx: &mut gpui::TestAppContext,
4109    ) {
4110        let (
4111            connection,
4112            project,
4113            _load_count,
4114            _close_count,
4115            load_session_updates,
4116            _load_session_gate,
4117            _keep_agent_alive,
4118        ) = connect_fake_agent(cx).await;
4119
4120        // Queue up some history updates that the fake agent will stream to
4121        // the client during the `load_session` call, before responding.
4122        *load_session_updates
4123            .lock()
4124            .expect("load_session_updates mutex poisoned") = vec![
4125            acp::SessionUpdate::UserMessageChunk(acp::ContentChunk::new(acp::ContentBlock::Text(
4126                acp::TextContent::new(String::from("hello agent")),
4127            ))),
4128            acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new(acp::ContentBlock::Text(
4129                acp::TextContent::new(String::from("hi user")),
4130            ))),
4131        ];
4132
4133        let session_id = acp::SessionId::new("session-replay");
4134        let work_dirs = util::path_list::PathList::new(&[std::path::Path::new("/a")]);
4135
4136        let thread = cx
4137            .update(|cx| {
4138                connection.clone().load_session(
4139                    session_id.clone(),
4140                    project.clone(),
4141                    work_dirs,
4142                    None,
4143                    cx,
4144                )
4145            })
4146            .await
4147            .expect("load_session failed");
4148        cx.run_until_parked();
4149
4150        let entries = thread.read_with(cx, |thread, _| {
4151            thread
4152                .entries()
4153                .iter()
4154                .map(|entry| match entry {
4155                    acp_thread::AgentThreadEntry::UserMessage(_) => "user",
4156                    acp_thread::AgentThreadEntry::AssistantMessage(_) => "assistant",
4157                    acp_thread::AgentThreadEntry::ToolCall(_) => "tool_call",
4158                    acp_thread::AgentThreadEntry::Elicitation(_) => "elicitation",
4159                    acp_thread::AgentThreadEntry::CompletedPlan(_) => "plan",
4160                    acp_thread::AgentThreadEntry::ContextCompaction(_) => "compaction",
4161                    acp_thread::AgentThreadEntry::SystemNote(_) => "system_note",
4162                })
4163                .collect::<Vec<_>>()
4164        });
4165
4166        assert_eq!(
4167            entries,
4168            vec!["user", "assistant"],
4169            "replayed notifications should be applied to the thread"
4170        );
4171    }
4172
4173    // Regression test: if `close_session` is issued while a `load_session`
4174    // RPC is still in flight, the close must take effect cleanly — the load
4175    // must fail with a recognizable error (not return an orphaned thread),
4176    // no entry must remain in `sessions` or `pending_sessions`, and the ACP
4177    // `close_session` RPC must be dispatched.
4178    #[gpui::test]
4179    async fn test_close_session_during_in_flight_load(cx: &mut gpui::TestAppContext) {
4180        let (
4181            connection,
4182            project,
4183            load_count,
4184            close_count,
4185            _load_session_updates,
4186            load_session_gate,
4187            _keep_agent_alive,
4188        ) = connect_fake_agent(cx).await;
4189
4190        // Install a gate so the fake agent's `load_session` handler parks
4191        // before sending its response. We'll close the session while the
4192        // load is parked.
4193        let (gate_tx, gate_rx) = async_channel::bounded::<()>(1);
4194        *load_session_gate
4195            .lock()
4196            .expect("load_session_gate mutex poisoned") = Some(gate_rx);
4197
4198        let session_id = acp::SessionId::new("session-close-during-load");
4199        let work_dirs = util::path_list::PathList::new(&[std::path::Path::new("/a")]);
4200
4201        let load_task = cx.update(|cx| {
4202            connection.clone().load_session(
4203                session_id.clone(),
4204                project.clone(),
4205                work_dirs,
4206                None,
4207                cx,
4208            )
4209        });
4210
4211        // Let the load RPC reach the agent and park on the gate.
4212        cx.run_until_parked();
4213        assert_eq!(
4214            load_count.load(Ordering::SeqCst),
4215            1,
4216            "load_session RPC should have been dispatched"
4217        );
4218        assert!(
4219            connection
4220                .pending_sessions
4221                .borrow()
4222                .contains_key(&session_id),
4223            "pending_sessions entry should exist while load is in flight"
4224        );
4225        assert!(
4226            connection.sessions.borrow().contains_key(&session_id),
4227            "sessions entry should be pre-registered to receive replay notifications"
4228        );
4229
4230        // Close the session while the load is still parked. This should take
4231        // the pending path and dispatch the ACP close RPC.
4232        let close_task = cx.update(|cx| connection.clone().close_session(&session_id, cx));
4233
4234        // Release the gate so the load RPC can finally respond.
4235        gate_tx.send(()).await.expect("gate send failed");
4236        drop(gate_tx);
4237
4238        let load_result = load_task.await;
4239        close_task.await.expect("close failed");
4240        cx.run_until_parked();
4241
4242        let err = load_result.expect_err("load should fail after close-during-load");
4243        assert!(
4244            err.to_string()
4245                .contains("session was closed before load completed"),
4246            "expected close-during-load error, got: {err}"
4247        );
4248
4249        assert_eq!(
4250            close_count.load(Ordering::SeqCst),
4251            1,
4252            "ACP close_session should be sent exactly once"
4253        );
4254        assert!(
4255            !connection.sessions.borrow().contains_key(&session_id),
4256            "sessions entry should be removed after close-during-load"
4257        );
4258        assert!(
4259            !connection
4260                .pending_sessions
4261                .borrow()
4262                .contains_key(&session_id),
4263            "pending_sessions entry should be removed after close-during-load"
4264        );
4265    }
4266
4267    // Regression test: when two concurrent `load_session` calls share a pending
4268    // task and one of them issues `close_session` before the load RPC
4269    // resolves, the remaining load must still succeed and the session must
4270    // stay live. If `close_session` incorrectly short-circuits via the
4271    // `sessions` path (removing the entry while a load is still in flight),
4272    // the pending task will fail and both concurrent loaders will lose
4273    // their handle.
4274    #[gpui::test]
4275    async fn test_close_during_load_preserves_other_concurrent_loader(
4276        cx: &mut gpui::TestAppContext,
4277    ) {
4278        let (
4279            connection,
4280            project,
4281            load_count,
4282            close_count,
4283            _load_session_updates,
4284            load_session_gate,
4285            _keep_agent_alive,
4286        ) = connect_fake_agent(cx).await;
4287
4288        let (gate_tx, gate_rx) = async_channel::bounded::<()>(1);
4289        *load_session_gate
4290            .lock()
4291            .expect("load_session_gate mutex poisoned") = Some(gate_rx);
4292
4293        let session_id = acp::SessionId::new("session-concurrent-close");
4294        let work_dirs = util::path_list::PathList::new(&[std::path::Path::new("/a")]);
4295
4296        // Kick off two concurrent loads; the second must join the first's pending
4297        // task rather than issuing a second RPC.
4298        let first_load = cx.update(|cx| {
4299            connection.clone().load_session(
4300                session_id.clone(),
4301                project.clone(),
4302                work_dirs.clone(),
4303                None,
4304                cx,
4305            )
4306        });
4307        let second_load = cx.update(|cx| {
4308            connection.clone().load_session(
4309                session_id.clone(),
4310                project.clone(),
4311                work_dirs.clone(),
4312                None,
4313                cx,
4314            )
4315        });
4316
4317        cx.run_until_parked();
4318        assert_eq!(
4319            load_count.load(Ordering::SeqCst),
4320            1,
4321            "load_session RPC should only be dispatched once for concurrent loads"
4322        );
4323
4324        // Close one of the two handles while the shared load is still parked.
4325        // Because a second loader still holds a pending ref, this should be a
4326        // no-op on the wire.
4327        cx.update(|cx| connection.clone().close_session(&session_id, cx))
4328            .await
4329            .expect("close during load failed");
4330        assert_eq!(
4331            close_count.load(Ordering::SeqCst),
4332            0,
4333            "close_session RPC must not be dispatched while another load handle remains"
4334        );
4335
4336        // Release the gate so the load RPC can finally respond.
4337        gate_tx.send(()).await.expect("gate send failed");
4338        drop(gate_tx);
4339
4340        let first_thread = first_load.await.expect("first load should still succeed");
4341        let second_thread = second_load.await.expect("second load should still succeed");
4342        cx.run_until_parked();
4343
4344        assert_eq!(
4345            first_thread.entity_id(),
4346            second_thread.entity_id(),
4347            "concurrent loads should share one AcpThread"
4348        );
4349        assert!(
4350            connection.sessions.borrow().contains_key(&session_id),
4351            "session must remain tracked while a load handle is still outstanding"
4352        );
4353        assert!(
4354            !connection
4355                .pending_sessions
4356                .borrow()
4357                .contains_key(&session_id),
4358            "pending_sessions entry should be cleared once the load resolves"
4359        );
4360
4361        // Final close drops ref_count to 0 and dispatches the ACP close RPC.
4362        cx.update(|cx| connection.clone().close_session(&session_id, cx))
4363            .await
4364            .expect("final close failed");
4365        cx.run_until_parked();
4366        assert_eq!(
4367            close_count.load(Ordering::SeqCst),
4368            1,
4369            "close_session RPC should fire exactly once when the last handle is released"
4370        );
4371        assert!(
4372            !connection.sessions.borrow().contains_key(&session_id),
4373            "session should be removed after final close"
4374        );
4375    }
4376}
4377
4378fn mcp_servers_for_project(project: &Entity<Project>, cx: &App) -> Vec<acp::McpServer> {
4379    let context_server_store = project.read(cx).context_server_store().read(cx);
4380    let is_local = project.read(cx).is_local();
4381    context_server_store
4382        .configured_server_ids()
4383        .iter()
4384        .filter_map(|id| {
4385            let configuration = context_server_store.configuration_for_server(id)?;
4386            match &*configuration {
4387                project::context_server_store::ContextServerConfiguration::Custom {
4388                    command,
4389                    remote,
4390                    ..
4391                }
4392                | project::context_server_store::ContextServerConfiguration::Extension {
4393                    command,
4394                    remote,
4395                    ..
4396                } if is_local || *remote => Some(acp::McpServer::Stdio(
4397                    acp::McpServerStdio::new(id.0.to_string(), &command.path)
4398                        .args(command.args.clone())
4399                        .env(if let Some(env) = command.env.as_ref() {
4400                            env.iter()
4401                                .map(|(name, value)| acp::EnvVariable::new(name, value))
4402                                .collect()
4403                        } else {
4404                            vec![]
4405                        }),
4406                )),
4407                project::context_server_store::ContextServerConfiguration::Http {
4408                    url,
4409                    headers,
4410                    timeout: _,
4411                    oauth: _,
4412                } => Some(acp::McpServer::Http(
4413                    acp::McpServerHttp::new(id.0.to_string(), url.to_string()).headers(
4414                        headers
4415                            .iter()
4416                            .map(|(name, value)| acp::HttpHeader::new(name, value))
4417                            .collect(),
4418                    ),
4419                )),
4420                _ => None,
4421            }
4422        })
4423        .collect()
4424}
4425
4426fn config_state(
4427    modes: Option<acp::SessionModeState>,
4428    config_options: Option<Vec<acp::SessionConfigOption>>,
4429) -> (
4430    Option<Rc<RefCell<acp::SessionModeState>>>,
4431    Option<Rc<RefCell<Vec<acp::SessionConfigOption>>>>,
4432) {
4433    if let Some(opts) = config_options {
4434        return (None, Some(Rc::new(RefCell::new(opts))));
4435    }
4436
4437    let modes = modes.map(|modes| Rc::new(RefCell::new(modes)));
4438    (modes, None)
4439}
4440
4441struct AcpSessionModes {
4442    session_id: acp::SessionId,
4443    connection: ConnectionTo<Agent>,
4444    state: Rc<RefCell<acp::SessionModeState>>,
4445}
4446
4447impl acp_thread::AgentSessionModes for AcpSessionModes {
4448    fn current_mode(&self) -> acp::SessionModeId {
4449        self.state.borrow().current_mode_id.clone()
4450    }
4451
4452    fn all_modes(&self) -> Vec<acp::SessionMode> {
4453        self.state.borrow().available_modes.clone()
4454    }
4455
4456    fn set_mode(&self, mode_id: acp::SessionModeId, cx: &mut App) -> Task<Result<()>> {
4457        let connection = self.connection.clone();
4458        let session_id = self.session_id.clone();
4459        let old_mode_id;
4460        {
4461            let mut state = self.state.borrow_mut();
4462            old_mode_id = state.current_mode_id.clone();
4463            state.current_mode_id = mode_id.clone();
4464        };
4465        let state = self.state.clone();
4466        cx.foreground_executor().spawn(async move {
4467            let result = connection
4468                .send_request(acp::SetSessionModeRequest::new(session_id, mode_id))
4469                .block_task()
4470                .await;
4471
4472            if result.is_err() {
4473                state.borrow_mut().current_mode_id = old_mode_id;
4474            }
4475
4476            result?;
4477
4478            Ok(())
4479        })
4480    }
4481}
4482
4483struct AcpSessionConfigOptions {
4484    session_id: acp::SessionId,
4485    connection: ConnectionTo<Agent>,
4486    state: Rc<RefCell<Vec<acp::SessionConfigOption>>>,
4487    watch_tx: Rc<RefCell<watch::Sender<()>>>,
4488    watch_rx: watch::Receiver<()>,
4489}
4490
4491impl acp_thread::AgentSessionConfigOptions for AcpSessionConfigOptions {
4492    fn config_options(&self) -> Vec<acp::SessionConfigOption> {
4493        self.state.borrow().clone()
4494    }
4495
4496    fn set_config_option(
4497        &self,
4498        config_id: acp::SessionConfigId,
4499        value: acp::SessionConfigOptionValue,
4500        cx: &mut App,
4501    ) -> Task<Result<Vec<acp::SessionConfigOption>>> {
4502        let connection = self.connection.clone();
4503        let session_id = self.session_id.clone();
4504        let state = self.state.clone();
4505
4506        let watch_tx = self.watch_tx.clone();
4507
4508        cx.foreground_executor().spawn(async move {
4509            let response = connection
4510                .send_request(acp::SetSessionConfigOptionRequest::new(
4511                    session_id, config_id, value,
4512                ))
4513                .block_task()
4514                .await?;
4515
4516            *state.borrow_mut() = response.config_options.clone();
4517            watch_tx.borrow_mut().send(()).ok();
4518            Ok(response.config_options)
4519        })
4520    }
4521
4522    fn watch(&self, _cx: &mut App) -> Option<watch::Receiver<()>> {
4523        Some(self.watch_rx.clone())
4524    }
4525}
4526
4527// ---------------------------------------------------------------------------
4528// Handler functions dispatched from background handler closures to the
4529// foreground thread via the ForegroundWork channel.
4530// ---------------------------------------------------------------------------
4531
4532fn session_thread(
4533    ctx: &ClientContext,
4534    session_id: &acp::SessionId,
4535) -> Result<WeakEntity<AcpThread>, acp::Error> {
4536    let sessions = ctx.sessions.borrow();
4537    sessions
4538        .get(session_id)
4539        .map(|session| session.thread.clone())
4540        .ok_or_else(|| acp::Error::internal_error().data(format!("unknown session: {session_id}")))
4541}
4542
4543fn respond_err<T: JsonRpcResponse>(responder: Responder<T>, err: acp::Error) {
4544    // Log the actual error we're returning — otherwise agents that hit an
4545    // error path (e.g. unknown session) would see only the generic internal
4546    // error returned over the wire with no trace of why on the client side.
4547    log::warn!(
4548        "Responding to ACP request `{method}` with error: {err:?}",
4549        method = responder.method()
4550    );
4551    responder.respond_with_error(err).log_err();
4552}
4553
4554fn respond_result<T: JsonRpcResponse>(responder: Responder<T>, result: Result<T, acp::Error>) {
4555    match result {
4556        Ok(response) => {
4557            responder.respond(response).log_err();
4558        }
4559        Err(err) => respond_err(responder, err),
4560    }
4561}
4562
4563fn handle_request_permission(
4564    args: acp::RequestPermissionRequest,
4565    responder: Responder<acp::RequestPermissionResponse>,
4566    cx: &mut AsyncApp,
4567    ctx: &ClientContext,
4568) {
4569    let thread = match session_thread(ctx, &args.session_id) {
4570        Ok(t) => t,
4571        Err(e) => return respond_err(responder, e),
4572    };
4573
4574    let cancellation = responder.cancellation();
4575    let tool_call_id = args.tool_call.tool_call_id.clone();
4576    cx.spawn(async move |cx| {
4577        let result: Result<_, acp::Error> = async {
4578            let task = thread
4579                .update(cx, |thread, cx| {
4580                    thread.request_tool_call_authorization(
4581                        args.tool_call,
4582                        acp_thread::PermissionOptions::Flat(args.options),
4583                        acp_thread::AuthorizationKind::PermissionGrant,
4584                        cx,
4585                    )
4586                })
4587                .flatten_acp()?;
4588            cancellation
4589                .run_until_cancelled(async { Ok(task.await) })
4590                .await
4591        }
4592        .await;
4593
4594        match result {
4595            Ok(outcome) => {
4596                responder
4597                    .respond(acp::RequestPermissionResponse::new(outcome.into()))
4598                    .log_err();
4599            }
4600            Err(e) => {
4601                if e.code == ErrorCode::RequestCancelled {
4602                    thread
4603                        .update(cx, |thread, cx| {
4604                            thread.cancel_tool_call_authorization(&tool_call_id, cx)
4605                        })
4606                        .log_err();
4607                }
4608                respond_err(responder, e)
4609            }
4610        }
4611    })
4612    .detach();
4613}
4614
4615fn handle_create_elicitation(
4616    args: acp::CreateElicitationRequest,
4617    responder: Responder<acp::CreateElicitationResponse>,
4618    cx: &mut AsyncApp,
4619    ctx: &ClientContext,
4620) {
4621    match args.scope() {
4622        acp::ElicitationScope::Session(scope) => {
4623            let thread = match session_thread(ctx, &scope.session_id) {
4624                Ok(t) => t,
4625                Err(e) => return respond_err(responder, e),
4626            };
4627
4628            let (elicitation_id, task) = match thread
4629                .update(cx, |thread, cx| {
4630                    thread.request_elicitation_with_id(args, cx)
4631                })
4632                .flatten_acp()
4633            {
4634                Ok(task) => task,
4635                Err(e) => return respond_err(responder, e),
4636            };
4637
4638            let cancellation = responder.cancellation();
4639            cx.spawn(async move |cx| {
4640                let result: Result<_, acp::Error> = cancellation
4641                    .run_until_cancelled(async { Ok(task.await) })
4642                    .await;
4643
4644                match result {
4645                    Ok(response) => {
4646                        responder.respond(response).log_err();
4647                    }
4648                    Err(e) => {
4649                        if e.code == ErrorCode::RequestCancelled {
4650                            thread
4651                                .update(cx, |thread, cx| {
4652                                    thread.cancel_elicitation(&elicitation_id, cx)
4653                                })
4654                                .log_err();
4655                        }
4656                        respond_err(responder, e);
4657                    }
4658                }
4659            })
4660            .detach();
4661        }
4662        acp::ElicitationScope::Request(_) => {
4663            let store = ctx.request_elicitations.clone();
4664            let (elicitation_id, task) =
4665                match store.update(cx, |store, cx| store.request_elicitation_with_id(args, cx)) {
4666                    Ok(task) => task,
4667                    Err(e) => return respond_err(responder, e),
4668                };
4669            let store = store.downgrade();
4670
4671            let cancellation = responder.cancellation();
4672            cx.spawn(async move |cx| {
4673                let result: Result<_, acp::Error> = cancellation
4674                    .run_until_cancelled(async { Ok(task.await) })
4675                    .await;
4676
4677                match result {
4678                    Ok(response) => {
4679                        responder.respond(response).log_err();
4680                    }
4681                    Err(e) => {
4682                        if e.code == ErrorCode::RequestCancelled {
4683                            store
4684                                .update(cx, |store, cx| {
4685                                    store.cancel_elicitation(&elicitation_id, cx)
4686                                })
4687                                .log_err();
4688                        }
4689                        respond_err(responder, e);
4690                    }
4691                }
4692            })
4693            .detach();
4694        }
4695        _ => {
4696            respond_err(
4697                responder,
4698                acp::Error::invalid_params().data("unknown elicitation scope"),
4699            );
4700        }
4701    }
4702}
4703
4704fn handle_complete_elicitation(
4705    args: acp::CompleteElicitationNotification,
4706    cx: &mut AsyncApp,
4707    ctx: &ClientContext,
4708) {
4709    let threads = ctx
4710        .sessions
4711        .borrow()
4712        .values()
4713        .map(|session| session.thread.clone())
4714        .collect::<Vec<_>>();
4715    let request_elicitations = ctx.request_elicitations.clone();
4716    let elicitation_id = args.elicitation_id;
4717
4718    cx.spawn(async move |cx| {
4719        for thread in threads {
4720            thread
4721                .update(cx, |thread, cx| {
4722                    thread.complete_url_elicitation(&elicitation_id, cx);
4723                })
4724                .ok();
4725        }
4726        request_elicitations.update(cx, |store, cx| {
4727            store.complete_url_elicitation(&elicitation_id, cx);
4728        });
4729    })
4730    .detach();
4731}
4732
4733fn handle_write_text_file(
4734    args: acp::WriteTextFileRequest,
4735    responder: Responder<acp::WriteTextFileResponse>,
4736    cx: &mut AsyncApp,
4737    ctx: &ClientContext,
4738) {
4739    let thread = match session_thread(ctx, &args.session_id) {
4740        Ok(t) => t,
4741        Err(e) => return respond_err(responder, e),
4742    };
4743
4744    cx.spawn(async move |cx| {
4745        let result: Result<_, acp::Error> = async {
4746            thread
4747                .update(cx, |thread, cx| {
4748                    thread.write_text_file(args.path, args.content, cx)
4749                })
4750                .map_err(acp::Error::from)?
4751                .await?;
4752            Ok(())
4753        }
4754        .await;
4755
4756        match result {
4757            Ok(()) => {
4758                responder
4759                    .respond(acp::WriteTextFileResponse::default())
4760                    .log_err();
4761            }
4762            Err(e) => respond_err(responder, e),
4763        }
4764    })
4765    .detach();
4766}
4767
4768fn handle_read_text_file(
4769    args: acp::ReadTextFileRequest,
4770    responder: Responder<acp::ReadTextFileResponse>,
4771    cx: &mut AsyncApp,
4772    ctx: &ClientContext,
4773) {
4774    let thread = match session_thread(ctx, &args.session_id) {
4775        Ok(t) => t,
4776        Err(e) => return respond_err(responder, e),
4777    };
4778
4779    cx.spawn(async move |cx| {
4780        let cancellation = responder.cancellation();
4781        let result = cancellation
4782            .run_until_cancelled(async {
4783                thread
4784                    .update(cx, |thread, cx| {
4785                        thread.read_text_file(args.path, args.line, args.limit, false, cx)
4786                    })
4787                    .map_err(acp::Error::from)?
4788                    .await
4789            })
4790            .await;
4791
4792        respond_result(responder, result.map(acp::ReadTextFileResponse::new));
4793    })
4794    .detach();
4795}
4796
4797fn handle_session_notification(
4798    notification: acp::SessionNotification,
4799    cx: &mut AsyncApp,
4800    ctx: &ClientContext,
4801) {
4802    // Extract everything we need from the session while briefly borrowing.
4803    let (thread, session_modes, config_opts_data) = {
4804        let sessions = ctx.sessions.borrow();
4805        let Some(session) = sessions.get(&notification.session_id) else {
4806            log::warn!(
4807                "Received session notification for unknown session: {:?}",
4808                notification.session_id
4809            );
4810            return;
4811        };
4812        (
4813            session.thread.clone(),
4814            session.session_modes.clone(),
4815            session
4816                .config_options
4817                .as_ref()
4818                .map(|opts| (opts.config_options.clone(), opts.tx.clone())),
4819        )
4820    };
4821    // Borrow is dropped here.
4822
4823    // Apply mode/config/session_list updates without holding the borrow.
4824    if let acp::SessionUpdate::CurrentModeUpdate(acp::CurrentModeUpdate {
4825        current_mode_id, ..
4826    }) = &notification.update
4827    {
4828        if let Some(session_modes) = &session_modes {
4829            session_modes.borrow_mut().current_mode_id = current_mode_id.clone();
4830        }
4831    }
4832
4833    if let acp::SessionUpdate::ConfigOptionUpdate(acp::ConfigOptionUpdate {
4834        config_options, ..
4835    }) = &notification.update
4836    {
4837        if let Some((config_opts_cell, tx_cell)) = &config_opts_data {
4838            *config_opts_cell.borrow_mut() = config_options.clone();
4839            tx_cell.borrow_mut().send(()).ok();
4840        }
4841    }
4842
4843    if let acp::SessionUpdate::SessionInfoUpdate(info_update) = &notification.update
4844        && let Some(session_list) = ctx.session_list.borrow().as_ref()
4845    {
4846        session_list.send_info_update(notification.session_id.clone(), info_update.clone());
4847    }
4848
4849    // Pre-handle: if a ToolCall carries terminal_info, create/register a display-only terminal.
4850    if let acp::SessionUpdate::ToolCall(tc) = &notification.update {
4851        if let Some(meta) = &tc.meta {
4852            if let Some(terminal_info) = meta.get("terminal_info") {
4853                if let Some(id_str) = terminal_info.get("terminal_id").and_then(|v| v.as_str()) {
4854                    let terminal_id = acp::TerminalId::new(id_str);
4855                    let cwd = terminal_info
4856                        .get("cwd")
4857                        .and_then(|v| v.as_str().map(PathBuf::from));
4858
4859                    thread
4860                        .update(cx, |thread, cx| {
4861                            let builder = TerminalBuilder::new_display_only(
4862                                CursorShape::default(),
4863                                AlternateScroll::On,
4864                                None,
4865                                0,
4866                                cx.background_executor(),
4867                                thread.project().read(cx).path_style(cx),
4868                            );
4869                            let lower = cx.new(|cx| builder.subscribe(cx));
4870                            thread.on_terminal_provider_event(
4871                                TerminalProviderEvent::Created {
4872                                    terminal_id,
4873                                    label: tc.title.clone(),
4874                                    cwd,
4875                                    output_byte_limit: None,
4876                                    terminal: lower,
4877                                },
4878                                cx,
4879                            );
4880                        })
4881                        .log_err();
4882                }
4883            }
4884        }
4885    }
4886
4887    // Forward the update to the acp_thread as usual.
4888    if let Err(err) = thread
4889        .update(cx, |thread, cx| {
4890            thread.handle_session_update(notification.update.clone(), cx)
4891        })
4892        .flatten_acp()
4893    {
4894        log::error!(
4895            "Failed to handle session update for {:?}: {err:?}",
4896            notification.session_id
4897        );
4898    }
4899
4900    // Post-handle: stream terminal output/exit if present on ToolCallUpdate meta.
4901    if let acp::SessionUpdate::ToolCallUpdate(tcu) = &notification.update {
4902        if let Some(meta) = &tcu.meta {
4903            if let Some(term_out) = meta.get("terminal_output") {
4904                if let Some(id_str) = term_out.get("terminal_id").and_then(|v| v.as_str()) {
4905                    let terminal_id = acp::TerminalId::new(id_str);
4906                    if let Some(s) = term_out.get("data").and_then(|v| v.as_str()) {
4907                        let data = s.as_bytes().to_vec();
4908                        thread
4909                            .update(cx, |thread, cx| {
4910                                thread.on_terminal_provider_event(
4911                                    TerminalProviderEvent::Output { terminal_id, data },
4912                                    cx,
4913                                );
4914                            })
4915                            .log_err();
4916                    }
4917                }
4918            }
4919
4920            if let Some(term_exit) = meta.get("terminal_exit") {
4921                if let Some(id_str) = term_exit.get("terminal_id").and_then(|v| v.as_str()) {
4922                    let terminal_id = acp::TerminalId::new(id_str);
4923                    let status = acp::TerminalExitStatus::new()
4924                        .exit_code(
4925                            term_exit
4926                                .get("exit_code")
4927                                .and_then(|v| v.as_u64())
4928                                .map(|i| i as u32),
4929                        )
4930                        .signal(
4931                            term_exit
4932                                .get("signal")
4933                                .and_then(|v| v.as_str().map(|s| s.to_string())),
4934                        );
4935
4936                    thread
4937                        .update(cx, |thread, cx| {
4938                            thread.on_terminal_provider_event(
4939                                TerminalProviderEvent::Exit {
4940                                    terminal_id,
4941                                    status,
4942                                },
4943                                cx,
4944                            );
4945                        })
4946                        .log_err();
4947                }
4948            }
4949        }
4950    }
4951}
4952
4953fn handle_create_terminal(
4954    args: acp::CreateTerminalRequest,
4955    responder: Responder<acp::CreateTerminalResponse>,
4956    cx: &mut AsyncApp,
4957    ctx: &ClientContext,
4958) {
4959    let thread = match session_thread(ctx, &args.session_id) {
4960        Ok(t) => t,
4961        Err(e) => return respond_err(responder, e),
4962    };
4963    let project = match thread
4964        .read_with(cx, |thread, _cx| thread.project().clone())
4965        .map_err(acp::Error::from)
4966    {
4967        Ok(p) => p,
4968        Err(e) => return respond_err(responder, e),
4969    };
4970
4971    cx.spawn(async move |cx| {
4972        let result: Result<_, acp::Error> = async {
4973            let terminal_entity = acp_thread::create_terminal_entity(
4974                args.command.clone(),
4975                &args.args,
4976                args.env
4977                    .into_iter()
4978                    .map(|env| (env.name, env.value))
4979                    .collect(),
4980                args.cwd.clone(),
4981                &project,
4982                cx,
4983            )
4984            .await?;
4985
4986            let terminal_entity = thread.update(cx, |thread, cx| {
4987                thread.register_terminal_created(
4988                    acp::TerminalId::new(uuid::Uuid::new_v4().to_string()),
4989                    format!("{} {}", args.command, args.args.join(" ")),
4990                    args.cwd.clone(),
4991                    args.output_byte_limit,
4992                    terminal_entity,
4993                    cx,
4994                )
4995            })?;
4996            let terminal_id = terminal_entity.read_with(cx, |terminal, _| terminal.id().clone());
4997            Ok(terminal_id)
4998        }
4999        .await;
5000
5001        match result {
5002            Ok(terminal_id) => {
5003                responder
5004                    .respond(acp::CreateTerminalResponse::new(terminal_id))
5005                    .log_err();
5006            }
5007            Err(e) => respond_err(responder, e),
5008        }
5009    })
5010    .detach();
5011}
5012
5013fn handle_kill_terminal(
5014    args: acp::KillTerminalRequest,
5015    responder: Responder<acp::KillTerminalResponse>,
5016    cx: &mut AsyncApp,
5017    ctx: &ClientContext,
5018) {
5019    let thread = match session_thread(ctx, &args.session_id) {
5020        Ok(t) => t,
5021        Err(e) => return respond_err(responder, e),
5022    };
5023
5024    match thread
5025        .update(cx, |thread, cx| thread.kill_terminal(args.terminal_id, cx))
5026        .flatten_acp()
5027    {
5028        Ok(()) => {
5029            responder
5030                .respond(acp::KillTerminalResponse::default())
5031                .log_err();
5032        }
5033        Err(e) => respond_err(responder, e),
5034    }
5035}
5036
5037fn handle_release_terminal(
5038    args: acp::ReleaseTerminalRequest,
5039    responder: Responder<acp::ReleaseTerminalResponse>,
5040    cx: &mut AsyncApp,
5041    ctx: &ClientContext,
5042) {
5043    let thread = match session_thread(ctx, &args.session_id) {
5044        Ok(t) => t,
5045        Err(e) => return respond_err(responder, e),
5046    };
5047
5048    match thread
5049        .update(cx, |thread, cx| {
5050            thread.release_terminal(args.terminal_id, cx)
5051        })
5052        .flatten_acp()
5053    {
5054        Ok(()) => {
5055            responder
5056                .respond(acp::ReleaseTerminalResponse::default())
5057                .log_err();
5058        }
5059        Err(e) => respond_err(responder, e),
5060    }
5061}
5062
5063fn handle_terminal_output(
5064    args: acp::TerminalOutputRequest,
5065    responder: Responder<acp::TerminalOutputResponse>,
5066    cx: &mut AsyncApp,
5067    ctx: &ClientContext,
5068) {
5069    let thread = match session_thread(ctx, &args.session_id) {
5070        Ok(t) => t,
5071        Err(e) => return respond_err(responder, e),
5072    };
5073
5074    match thread
5075        .read_with(cx, |thread, cx| -> anyhow::Result<_> {
5076            let out = thread
5077                .terminal(args.terminal_id)?
5078                .read(cx)
5079                .current_output(cx);
5080            Ok(out)
5081        })
5082        .flatten_acp()
5083    {
5084        Ok(output) => {
5085            responder.respond(output).log_err();
5086        }
5087        Err(e) => respond_err(responder, e),
5088    }
5089}
5090
5091fn handle_wait_for_terminal_exit(
5092    args: acp::WaitForTerminalExitRequest,
5093    responder: Responder<acp::WaitForTerminalExitResponse>,
5094    cx: &mut AsyncApp,
5095    ctx: &ClientContext,
5096) {
5097    let thread = match session_thread(ctx, &args.session_id) {
5098        Ok(t) => t,
5099        Err(e) => return respond_err(responder, e),
5100    };
5101
5102    cx.spawn(async move |cx| {
5103        let cancellation = responder.cancellation();
5104        let result = cancellation
5105            .run_until_cancelled(async {
5106                let exit_status = thread
5107                    .update(cx, |thread, cx| {
5108                        anyhow::Ok(thread.terminal(args.terminal_id)?.read(cx).wait_for_exit())
5109                    })
5110                    .flatten_acp()?
5111                    .await;
5112                Ok(exit_status)
5113            })
5114            .await;
5115
5116        respond_result(responder, result.map(acp::WaitForTerminalExitResponse::new));
5117    })
5118    .detach();
5119}
5120
5121#[cfg(test)]
5122mod session_open_error_tests {
5123    use super::*;
5124
5125    /// A vanished session must not read as a failure to start.
5126    ///
5127    /// This shipped as "Failed to Launch: Resource not found: <uuid>" — a
5128    /// title that blames startup for something startup did not do, and a body
5129    /// naming an identifier the reader cannot act on.
5130    #[test]
5131    fn a_missing_session_is_reported_as_gone_not_as_a_launch_failure() {
5132        let mapped = map_session_open_error(acp::Error::resource_not_found(Some(
5133            "1297dcfa-027c-422c-9b56-acb526144c93".into(),
5134        )));
5135        let load_error = mapped
5136            .downcast_ref::<LoadError>()
5137            .expect("a missing session must map to a LoadError");
5138        assert!(
5139            matches!(load_error, LoadError::SessionGone),
5140            "expected SessionGone, got {load_error:?}"
5141        );
5142        let rendered = load_error.to_string();
5143        assert!(
5144            !rendered.contains("1297dcfa"),
5145            "the session id is not actionable and must not be shown: {rendered}"
5146        );
5147        assert!(
5148            rendered.contains("new thread"),
5149            "the reader needs the way forward: {rendered}"
5150        );
5151    }
5152
5153    /// Only the session-scoped calls get this treatment. `ResourceNotFound`
5154    /// from a file read still means a missing file.
5155    #[test]
5156    fn other_errors_are_left_alone() {
5157        let mapped = map_session_open_error(acp::Error::internal_error());
5158        assert!(
5159            mapped.downcast_ref::<LoadError>().is_none(),
5160            "an unrelated error must not be rewritten into SessionGone"
5161        );
5162    }
5163}
5164