1use acp_thread::{
2 AgentConnection, AgentSessionInfo, AgentSessionList, AgentSessionListRequest,
3 AgentSessionListResponse,
4};
5use acp_tools::AcpConnectionRegistry;
6use action_log::ActionLog;
7use agent_client_protocol::{self as acp, Agent as _, ErrorCode};
8use anyhow::anyhow;
9use collections::HashMap;
10use futures::AsyncBufReadExt as _;
11use futures::io::BufReader;
12use project::Project;
13use project::agent_server_store::{AgentServerCommand, GEMINI_NAME};
14use serde::Deserialize;
15use settings::Settings as _;
16use task::ShellBuilder;
17use util::ResultExt as _;
18use util::process::Child;
19
20use std::path::PathBuf;
21use std::process::Stdio;
22use std::{any::Any, cell::RefCell};
23use std::{path::Path, rc::Rc};
24use thiserror::Error;
25
26use anyhow::{Context as _, Result};
27use gpui::{App, AppContext as _, AsyncApp, Entity, SharedString, Task, WeakEntity};
28
29use acp_thread::{AcpThread, AuthRequired, LoadError, TerminalProviderEvent};
30use terminal::TerminalBuilder;
31use terminal::terminal_settings::{AlternateScroll, CursorShape, TerminalSettings};
32
33#[derive(Debug, Error)]
34#[error("Unsupported version")]
35pub struct UnsupportedVersion;
36
37pub struct AcpConnection {
38 server_name: SharedString,
39 display_name: SharedString,
40 telemetry_id: SharedString,
41 connection: Rc<acp::ClientSideConnection>,
42 sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
43 auth_methods: Vec<acp::AuthMethod>,
44 agent_capabilities: acp::AgentCapabilities,
45 default_mode: Option<acp::SessionModeId>,
46 default_model: Option<acp::ModelId>,
47 default_config_options: HashMap<String, String>,
48 child: Child,
49 session_list: Option<Rc<AcpSessionList>>,
50 _io_task: Task<Result<(), acp::Error>>,
51 _wait_task: Task<Result<()>>,
52 _stderr_task: Task<Result<()>>,
53}
54
55struct ConfigOptions {
56 config_options: Rc<RefCell<Vec<acp::SessionConfigOption>>>,
57 tx: Rc<RefCell<watch::Sender<()>>>,
58 rx: watch::Receiver<()>,
59}
60
61impl ConfigOptions {
62 fn new(config_options: Rc<RefCell<Vec<acp::SessionConfigOption>>>) -> Self {
63 let (tx, rx) = watch::channel(());
64 Self {
65 config_options,
66 tx: Rc::new(RefCell::new(tx)),
67 rx,
68 }
69 }
70}
71
72pub struct AcpSession {
73 thread: WeakEntity<AcpThread>,
74 suppress_abort_err: bool,
75 models: Option<Rc<RefCell<acp::SessionModelState>>>,
76 session_modes: Option<Rc<RefCell<acp::SessionModeState>>>,
77 config_options: Option<ConfigOptions>,
78}
79
80pub struct AcpSessionList {
81 connection: Rc<acp::ClientSideConnection>,
82 updates_tx: smol::channel::Sender<acp_thread::SessionListUpdate>,
83 updates_rx: smol::channel::Receiver<acp_thread::SessionListUpdate>,
84}
85
86impl AcpSessionList {
87 fn new(connection: Rc<acp::ClientSideConnection>) -> Self {
88 let (tx, rx) = smol::channel::unbounded();
89 Self {
90 connection,
91 updates_tx: tx,
92 updates_rx: rx,
93 }
94 }
95
96 fn notify_update(&self) {
97 self.updates_tx
98 .try_send(acp_thread::SessionListUpdate::Refresh)
99 .log_err();
100 }
101
102 fn send_info_update(&self, session_id: acp::SessionId, update: acp::SessionInfoUpdate) {
103 self.updates_tx
104 .try_send(acp_thread::SessionListUpdate::SessionInfo { session_id, update })
105 .log_err();
106 }
107}
108
109impl AgentSessionList for AcpSessionList {
110 fn list_sessions(
111 &self,
112 request: AgentSessionListRequest,
113 cx: &mut App,
114 ) -> Task<Result<AgentSessionListResponse>> {
115 let conn = self.connection.clone();
116 cx.foreground_executor().spawn(async move {
117 let acp_request = acp::ListSessionsRequest::new()
118 .cwd(request.cwd)
119 .cursor(request.cursor);
120 let response = conn.list_sessions(acp_request).await?;
121 Ok(AgentSessionListResponse {
122 sessions: response
123 .sessions
124 .into_iter()
125 .map(|s| AgentSessionInfo {
126 session_id: s.session_id,
127 cwd: Some(s.cwd),
128 title: s.title.map(Into::into),
129 updated_at: s.updated_at.and_then(|date_str| {
130 chrono::DateTime::parse_from_rfc3339(&date_str)
131 .ok()
132 .map(|dt| dt.with_timezone(&chrono::Utc))
133 }),
134 meta: s.meta,
135 })
136 .collect(),
137 next_cursor: response.next_cursor,
138 meta: response.meta,
139 })
140 })
141 }
142
143 fn watch(
144 &self,
145 _cx: &mut App,
146 ) -> Option<smol::channel::Receiver<acp_thread::SessionListUpdate>> {
147 Some(self.updates_rx.clone())
148 }
149
150 fn notify_refresh(&self) {
151 self.notify_update();
152 }
153
154 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
155 self
156 }
157}
158
159pub async fn connect(
160 server_name: SharedString,
161 display_name: SharedString,
162 command: AgentServerCommand,
163 default_mode: Option<acp::SessionModeId>,
164 default_model: Option<acp::ModelId>,
165 default_config_options: HashMap<String, String>,
166 cx: &mut AsyncApp,
167) -> Result<Rc<dyn AgentConnection>> {
168 let conn = AcpConnection::stdio(
169 server_name,
170 display_name,
171 command.clone(),
172 default_mode,
173 default_model,
174 default_config_options,
175 cx,
176 )
177 .await?;
178 Ok(Rc::new(conn) as _)
179}
180
181const MINIMUM_SUPPORTED_VERSION: acp::ProtocolVersion = acp::ProtocolVersion::V1;
182
183impl AcpConnection {
184 pub async fn stdio(
185 server_name: SharedString,
186 display_name: SharedString,
187 command: AgentServerCommand,
188 default_mode: Option<acp::SessionModeId>,
189 default_model: Option<acp::ModelId>,
190 default_config_options: HashMap<String, String>,
191 cx: &mut AsyncApp,
192 ) -> Result<Self> {
193 let shell = cx.update(|cx| TerminalSettings::get(None, cx).shell.clone());
194 let builder = ShellBuilder::new(&shell, cfg!(windows)).non_interactive();
195 let mut child =
196 builder.build_std_command(Some(command.path.display().to_string()), &command.args);
197 child.envs(command.env.iter().flatten());
198 let mut child = Child::spawn(child, Stdio::piped(), Stdio::piped(), Stdio::piped())?;
199
200 let stdout = child.stdout.take().context("Failed to take stdout")?;
201 let stdin = child.stdin.take().context("Failed to take stdin")?;
202 let stderr = child.stderr.take().context("Failed to take stderr")?;
203 log::debug!(
204 "Spawning external agent server: {:?}, {:?}",
205 command.path,
206 command.args
207 );
208 log::trace!("Spawned (pid: {})", child.id());
209
210 let sessions = Rc::new(RefCell::new(HashMap::default()));
211
212 let (release_channel, version): (Option<&str>, String) = cx.update(|cx| {
213 (
214 release_channel::ReleaseChannel::try_global(cx)
215 .map(|release_channel| release_channel.display_name()),
216 release_channel::AppVersion::global(cx).to_string(),
217 )
218 });
219
220 let client_session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>> =
221 Rc::new(RefCell::new(None));
222
223 let client = ClientDelegate {
224 sessions: sessions.clone(),
225 session_list: client_session_list.clone(),
226 cx: cx.clone(),
227 };
228 let (connection, io_task) = acp::ClientSideConnection::new(client, stdin, stdout, {
229 let foreground_executor = cx.foreground_executor().clone();
230 move |fut| {
231 foreground_executor.spawn(fut).detach();
232 }
233 });
234
235 let io_task = cx.background_spawn(io_task);
236
237 let stderr_task = cx.background_spawn(async move {
238 let mut stderr = BufReader::new(stderr);
239 let mut line = String::new();
240 while let Ok(n) = stderr.read_line(&mut line).await
241 && n > 0
242 {
243 log::warn!("agent stderr: {}", line.trim());
244 line.clear();
245 }
246 Ok(())
247 });
248
249 let wait_task = cx.spawn({
250 let sessions = sessions.clone();
251 let status_fut = child.status();
252 async move |cx| {
253 let status = status_fut.await?;
254
255 for session in sessions.borrow().values() {
256 session
257 .thread
258 .update(cx, |thread, cx| {
259 thread.emit_load_error(LoadError::Exited { status }, cx)
260 })
261 .ok();
262 }
263
264 anyhow::Ok(())
265 }
266 });
267
268 let connection = Rc::new(connection);
269
270 cx.update(|cx| {
271 AcpConnectionRegistry::default_global(cx).update(cx, |registry, cx| {
272 registry.set_active_connection(server_name.clone(), &connection, cx)
273 });
274 });
275
276 let response = connection
277 .initialize(
278 acp::InitializeRequest::new(acp::ProtocolVersion::V1)
279 .client_capabilities(
280 acp::ClientCapabilities::new()
281 .fs(acp::FileSystemCapability::new()
282 .read_text_file(true)
283 .write_text_file(true))
284 .terminal(true)
285 // Experimental: Allow for rendering terminal output from the agents
286 .meta(acp::Meta::from_iter([
287 ("terminal_output".into(), true.into()),
288 ("terminal-auth".into(), true.into()),
289 ])),
290 )
291 .client_info(
292 acp::Implementation::new("zed", version)
293 .title(release_channel.map(ToOwned::to_owned)),
294 ),
295 )
296 .await?;
297
298 if response.protocol_version < MINIMUM_SUPPORTED_VERSION {
299 return Err(UnsupportedVersion.into());
300 }
301
302 let telemetry_id = response
303 .agent_info
304 // Use the one the agent provides if we have one
305 .map(|info| info.name.into())
306 // Otherwise, just use the name
307 .unwrap_or_else(|| server_name.clone());
308
309 let session_list = if response
310 .agent_capabilities
311 .session_capabilities
312 .list
313 .is_some()
314 {
315 let list = Rc::new(AcpSessionList::new(connection.clone()));
316 *client_session_list.borrow_mut() = Some(list.clone());
317 Some(list)
318 } else {
319 None
320 };
321
322 // TODO: Remove this override once Google team releases their official auth methods
323 let auth_methods = if server_name == GEMINI_NAME {
324 let mut args = command.args.clone();
325 args.retain(|a| a != "--experimental-acp");
326 let value = serde_json::json!({
327 "label": "gemini /auth",
328 "command": command.path.to_string_lossy().into_owned(),
329 "args": args,
330 "env": command.env.clone().unwrap_or_default(),
331 });
332 let meta = acp::Meta::from_iter([("terminal-auth".to_string(), value)]);
333 vec![
334 acp::AuthMethod::new("spawn-gemini-cli", "Login")
335 .description("Login with your Google or Vertex AI account")
336 .meta(meta),
337 ]
338 } else {
339 response.auth_methods
340 };
341 Ok(Self {
342 auth_methods,
343 connection,
344 server_name,
345 display_name,
346 telemetry_id,
347 sessions,
348 agent_capabilities: response.agent_capabilities,
349 default_mode,
350 default_model,
351 default_config_options,
352 session_list,
353 _io_task: io_task,
354 _wait_task: wait_task,
355 _stderr_task: stderr_task,
356 child,
357 })
358 }
359
360 pub fn prompt_capabilities(&self) -> &acp::PromptCapabilities {
361 &self.agent_capabilities.prompt_capabilities
362 }
363}
364
365impl Drop for AcpConnection {
366 fn drop(&mut self) {
367 self.child.kill().log_err();
368 }
369}
370
371impl AgentConnection for AcpConnection {
372 fn telemetry_id(&self) -> SharedString {
373 self.telemetry_id.clone()
374 }
375
376 fn new_session(
377 self: Rc<Self>,
378 project: Entity<Project>,
379 cwd: &Path,
380 cx: &mut App,
381 ) -> Task<Result<Entity<AcpThread>>> {
382 let name = self.server_name.clone();
383 let cwd = cwd.to_path_buf();
384 let mcp_servers = mcp_servers_for_project(&project, cx);
385
386 cx.spawn(async move |cx| {
387 let response = self.connection
388 .new_session(acp::NewSessionRequest::new(cwd.clone()).mcp_servers(mcp_servers))
389 .await
390 .map_err(map_acp_error)?;
391
392 let (modes, models, config_options) = config_state(response.modes, response.models, response.config_options);
393
394 if let Some(default_mode) = self.default_mode.clone() {
395 if let Some(modes) = modes.as_ref() {
396 let mut modes_ref = modes.borrow_mut();
397 let has_mode = modes_ref.available_modes.iter().any(|mode| mode.id == default_mode);
398
399 if has_mode {
400 let initial_mode_id = modes_ref.current_mode_id.clone();
401
402 cx.spawn({
403 let default_mode = default_mode.clone();
404 let session_id = response.session_id.clone();
405 let modes = modes.clone();
406 let conn = self.connection.clone();
407 async move |_| {
408 let result = conn.set_session_mode(acp::SetSessionModeRequest::new(session_id, default_mode))
409 .await.log_err();
410
411 if result.is_none() {
412 modes.borrow_mut().current_mode_id = initial_mode_id;
413 }
414 }
415 }).detach();
416
417 modes_ref.current_mode_id = default_mode;
418 } else {
419 let available_modes = modes_ref
420 .available_modes
421 .iter()
422 .map(|mode| format!("- `{}`: {}", mode.id, mode.name))
423 .collect::<Vec<_>>()
424 .join("\n");
425
426 log::warn!(
427 "`{default_mode}` is not valid {name} mode. Available options:\n{available_modes}",
428 );
429 }
430 }
431 }
432
433 if let Some(default_model) = self.default_model.clone() {
434 if let Some(models) = models.as_ref() {
435 let mut models_ref = models.borrow_mut();
436 let has_model = models_ref.available_models.iter().any(|model| model.model_id == default_model);
437
438 if has_model {
439 let initial_model_id = models_ref.current_model_id.clone();
440
441 cx.spawn({
442 let default_model = default_model.clone();
443 let session_id = response.session_id.clone();
444 let models = models.clone();
445 let conn = self.connection.clone();
446 async move |_| {
447 let result = conn.set_session_model(acp::SetSessionModelRequest::new(session_id, default_model))
448 .await.log_err();
449
450 if result.is_none() {
451 models.borrow_mut().current_model_id = initial_model_id;
452 }
453 }
454 }).detach();
455
456 models_ref.current_model_id = default_model;
457 } else {
458 let available_models = models_ref
459 .available_models
460 .iter()
461 .map(|model| format!("- `{}`: {}", model.model_id, model.name))
462 .collect::<Vec<_>>()
463 .join("\n");
464
465 log::warn!(
466 "`{default_model}` is not a valid {name} model. Available options:\n{available_models}",
467 );
468 }
469 }
470 }
471
472 if let Some(config_opts) = config_options.as_ref() {
473 let defaults_to_apply: Vec<_> = {
474 let config_opts_ref = config_opts.borrow();
475 config_opts_ref
476 .iter()
477 .filter_map(|config_option| {
478 let default_value = self.default_config_options.get(&*config_option.id.0)?;
479
480 let is_valid = match &config_option.kind {
481 acp::SessionConfigKind::Select(select) => match &select.options {
482 acp::SessionConfigSelectOptions::Ungrouped(options) => {
483 options.iter().any(|opt| &*opt.value.0 == default_value.as_str())
484 }
485 acp::SessionConfigSelectOptions::Grouped(groups) => groups
486 .iter()
487 .any(|g| g.options.iter().any(|opt| &*opt.value.0 == default_value.as_str())),
488 _ => false,
489 },
490 _ => false,
491 };
492
493 if is_valid {
494 let initial_value = match &config_option.kind {
495 acp::SessionConfigKind::Select(select) => {
496 Some(select.current_value.clone())
497 }
498 _ => None,
499 };
500 Some((config_option.id.clone(), default_value.clone(), initial_value))
501 } else {
502 log::warn!(
503 "`{}` is not a valid value for config option `{}` in {}",
504 default_value,
505 config_option.id.0,
506 name
507 );
508 None
509 }
510 })
511 .collect()
512 };
513
514 for (config_id, default_value, initial_value) in defaults_to_apply {
515 cx.spawn({
516 let default_value_id = acp::SessionConfigValueId::new(default_value.clone());
517 let session_id = response.session_id.clone();
518 let config_id_clone = config_id.clone();
519 let config_opts = config_opts.clone();
520 let conn = self.connection.clone();
521 async move |_| {
522 let result = conn
523 .set_session_config_option(
524 acp::SetSessionConfigOptionRequest::new(
525 session_id,
526 config_id_clone.clone(),
527 default_value_id,
528 ),
529 )
530 .await
531 .log_err();
532
533 if result.is_none() {
534 if let Some(initial) = initial_value {
535 let mut opts = config_opts.borrow_mut();
536 if let Some(opt) = opts.iter_mut().find(|o| o.id == config_id_clone) {
537 if let acp::SessionConfigKind::Select(select) =
538 &mut opt.kind
539 {
540 select.current_value = initial;
541 }
542 }
543 }
544 }
545 }
546 })
547 .detach();
548
549 let mut opts = config_opts.borrow_mut();
550 if let Some(opt) = opts.iter_mut().find(|o| o.id == config_id) {
551 if let acp::SessionConfigKind::Select(select) = &mut opt.kind {
552 select.current_value = acp::SessionConfigValueId::new(default_value);
553 }
554 }
555 }
556 }
557
558 let action_log = cx.new(|_| ActionLog::new(project.clone()));
559 let thread: Entity<AcpThread> = cx.new(|cx| {
560 AcpThread::new(
561 None,
562 self.display_name.clone(),
563 Some(cwd),
564 self.clone(),
565 project,
566 action_log,
567 response.session_id.clone(),
568 // ACP doesn't currently support per-session prompt capabilities or changing capabilities dynamically.
569 watch::Receiver::constant(self.agent_capabilities.prompt_capabilities.clone()),
570 cx,
571 )
572 });
573
574 self.sessions.borrow_mut().insert(
575 response.session_id,
576 AcpSession {
577 thread: thread.downgrade(),
578 suppress_abort_err: false,
579 session_modes: modes,
580 models,
581 config_options: config_options.map(ConfigOptions::new),
582 },
583 );
584
585 Ok(thread)
586 })
587 }
588
589 fn supports_load_session(&self) -> bool {
590 self.agent_capabilities.load_session
591 }
592
593 fn supports_resume_session(&self) -> bool {
594 self.agent_capabilities
595 .session_capabilities
596 .resume
597 .is_some()
598 }
599
600 fn load_session(
601 self: Rc<Self>,
602 session_id: acp::SessionId,
603 project: Entity<Project>,
604 cwd: &Path,
605 title: Option<SharedString>,
606 cx: &mut App,
607 ) -> Task<Result<Entity<AcpThread>>> {
608 if !self.agent_capabilities.load_session {
609 return Task::ready(Err(anyhow!(LoadError::Other(
610 "Loading sessions is not supported by this agent.".into()
611 ))));
612 }
613
614 let cwd = cwd.to_path_buf();
615 let mcp_servers = mcp_servers_for_project(&project, cx);
616 let action_log = cx.new(|_| ActionLog::new(project.clone()));
617 let title = title.unwrap_or_else(|| self.display_name.clone());
618 let thread: Entity<AcpThread> = cx.new(|cx| {
619 AcpThread::new(
620 None,
621 title,
622 Some(cwd.clone()),
623 self.clone(),
624 project,
625 action_log,
626 session_id.clone(),
627 watch::Receiver::constant(self.agent_capabilities.prompt_capabilities.clone()),
628 cx,
629 )
630 });
631
632 self.sessions.borrow_mut().insert(
633 session_id.clone(),
634 AcpSession {
635 thread: thread.downgrade(),
636 suppress_abort_err: false,
637 session_modes: None,
638 models: None,
639 config_options: None,
640 },
641 );
642
643 cx.spawn(async move |_| {
644 let response = match self
645 .connection
646 .load_session(
647 acp::LoadSessionRequest::new(session_id.clone(), cwd).mcp_servers(mcp_servers),
648 )
649 .await
650 {
651 Ok(response) => response,
652 Err(err) => {
653 self.sessions.borrow_mut().remove(&session_id);
654 return Err(map_acp_error(err));
655 }
656 };
657
658 let (modes, models, config_options) =
659 config_state(response.modes, response.models, response.config_options);
660 if let Some(session) = self.sessions.borrow_mut().get_mut(&session_id) {
661 session.session_modes = modes;
662 session.models = models;
663 session.config_options = config_options.map(ConfigOptions::new);
664 }
665
666 Ok(thread)
667 })
668 }
669
670 fn resume_session(
671 self: Rc<Self>,
672 session_id: acp::SessionId,
673 project: Entity<Project>,
674 cwd: &Path,
675 title: Option<SharedString>,
676 cx: &mut App,
677 ) -> Task<Result<Entity<AcpThread>>> {
678 if self
679 .agent_capabilities
680 .session_capabilities
681 .resume
682 .is_none()
683 {
684 return Task::ready(Err(anyhow!(LoadError::Other(
685 "Resuming sessions is not supported by this agent.".into()
686 ))));
687 }
688
689 let cwd = cwd.to_path_buf();
690 let mcp_servers = mcp_servers_for_project(&project, cx);
691 let action_log = cx.new(|_| ActionLog::new(project.clone()));
692 let title = title.unwrap_or_else(|| self.display_name.clone());
693 let thread: Entity<AcpThread> = cx.new(|cx| {
694 AcpThread::new(
695 None,
696 title,
697 Some(cwd.clone()),
698 self.clone(),
699 project,
700 action_log,
701 session_id.clone(),
702 watch::Receiver::constant(self.agent_capabilities.prompt_capabilities.clone()),
703 cx,
704 )
705 });
706
707 self.sessions.borrow_mut().insert(
708 session_id.clone(),
709 AcpSession {
710 thread: thread.downgrade(),
711 suppress_abort_err: false,
712 session_modes: None,
713 models: None,
714 config_options: None,
715 },
716 );
717
718 cx.spawn(async move |_| {
719 let response = match self
720 .connection
721 .resume_session(
722 acp::ResumeSessionRequest::new(session_id.clone(), cwd)
723 .mcp_servers(mcp_servers),
724 )
725 .await
726 {
727 Ok(response) => response,
728 Err(err) => {
729 self.sessions.borrow_mut().remove(&session_id);
730 return Err(map_acp_error(err));
731 }
732 };
733
734 let (modes, models, config_options) =
735 config_state(response.modes, response.models, response.config_options);
736 if let Some(session) = self.sessions.borrow_mut().get_mut(&session_id) {
737 session.session_modes = modes;
738 session.models = models;
739 session.config_options = config_options.map(ConfigOptions::new);
740 }
741
742 Ok(thread)
743 })
744 }
745
746 fn auth_methods(&self) -> &[acp::AuthMethod] {
747 &self.auth_methods
748 }
749
750 fn authenticate(&self, method_id: acp::AuthMethodId, cx: &mut App) -> Task<Result<()>> {
751 let conn = self.connection.clone();
752 cx.foreground_executor().spawn(async move {
753 conn.authenticate(acp::AuthenticateRequest::new(method_id))
754 .await?;
755 Ok(())
756 })
757 }
758
759 fn prompt(
760 &self,
761 _id: Option<acp_thread::UserMessageId>,
762 params: acp::PromptRequest,
763 cx: &mut App,
764 ) -> Task<Result<acp::PromptResponse>> {
765 let conn = self.connection.clone();
766 let sessions = self.sessions.clone();
767 let session_id = params.session_id.clone();
768 cx.foreground_executor().spawn(async move {
769 let result = conn.prompt(params).await;
770
771 let mut suppress_abort_err = false;
772
773 if let Some(session) = sessions.borrow_mut().get_mut(&session_id) {
774 suppress_abort_err = session.suppress_abort_err;
775 session.suppress_abort_err = false;
776 }
777
778 match result {
779 Ok(response) => Ok(response),
780 Err(err) => {
781 if err.code == acp::ErrorCode::AuthRequired {
782 return Err(anyhow!(acp::Error::auth_required()));
783 }
784
785 if err.code != ErrorCode::InternalError {
786 anyhow::bail!(err)
787 }
788
789 let Some(data) = &err.data else {
790 anyhow::bail!(err)
791 };
792
793 // Temporary workaround until the following PR is generally available:
794 // https://github.com/google-gemini/gemini-cli/pull/6656
795
796 #[derive(Deserialize)]
797 #[serde(deny_unknown_fields)]
798 struct ErrorDetails {
799 details: Box<str>,
800 }
801
802 match serde_json::from_value(data.clone()) {
803 Ok(ErrorDetails { details }) => {
804 if suppress_abort_err
805 && (details.contains("This operation was aborted")
806 || details.contains("The user aborted a request"))
807 {
808 Ok(acp::PromptResponse::new(acp::StopReason::Cancelled))
809 } else {
810 Err(anyhow!(details))
811 }
812 }
813 Err(_) => Err(anyhow!(err)),
814 }
815 }
816 }
817 })
818 }
819
820 fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
821 if let Some(session) = self.sessions.borrow_mut().get_mut(session_id) {
822 session.suppress_abort_err = true;
823 }
824 let conn = self.connection.clone();
825 let params = acp::CancelNotification::new(session_id.clone());
826 cx.foreground_executor()
827 .spawn(async move { conn.cancel(params).await })
828 .detach();
829 }
830
831 fn session_modes(
832 &self,
833 session_id: &acp::SessionId,
834 _cx: &App,
835 ) -> Option<Rc<dyn acp_thread::AgentSessionModes>> {
836 let sessions = self.sessions.clone();
837 let sessions_ref = sessions.borrow();
838 let Some(session) = sessions_ref.get(session_id) else {
839 return None;
840 };
841
842 if let Some(modes) = session.session_modes.as_ref() {
843 Some(Rc::new(AcpSessionModes {
844 connection: self.connection.clone(),
845 session_id: session_id.clone(),
846 state: modes.clone(),
847 }) as _)
848 } else {
849 None
850 }
851 }
852
853 fn model_selector(
854 &self,
855 session_id: &acp::SessionId,
856 ) -> Option<Rc<dyn acp_thread::AgentModelSelector>> {
857 let sessions = self.sessions.clone();
858 let sessions_ref = sessions.borrow();
859 let Some(session) = sessions_ref.get(session_id) else {
860 return None;
861 };
862
863 if let Some(models) = session.models.as_ref() {
864 Some(Rc::new(AcpModelSelector::new(
865 session_id.clone(),
866 self.connection.clone(),
867 models.clone(),
868 )) as _)
869 } else {
870 None
871 }
872 }
873
874 fn session_config_options(
875 &self,
876 session_id: &acp::SessionId,
877 _cx: &App,
878 ) -> Option<Rc<dyn acp_thread::AgentSessionConfigOptions>> {
879 let sessions = self.sessions.borrow();
880 let session = sessions.get(session_id)?;
881
882 let config_opts = session.config_options.as_ref()?;
883
884 Some(Rc::new(AcpSessionConfigOptions {
885 session_id: session_id.clone(),
886 connection: self.connection.clone(),
887 state: config_opts.config_options.clone(),
888 watch_tx: config_opts.tx.clone(),
889 watch_rx: config_opts.rx.clone(),
890 }) as _)
891 }
892
893 fn session_list(&self, _cx: &mut App) -> Option<Rc<dyn AgentSessionList>> {
894 self.session_list.clone().map(|s| s as _)
895 }
896
897 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
898 self
899 }
900}
901
902fn map_acp_error(err: acp::Error) -> anyhow::Error {
903 if err.code == acp::ErrorCode::AuthRequired {
904 let mut error = AuthRequired::new();
905
906 if err.message != acp::ErrorCode::AuthRequired.to_string() {
907 error = error.with_description(err.message);
908 }
909
910 anyhow!(error)
911 } else {
912 anyhow!(err)
913 }
914}
915
916fn mcp_servers_for_project(project: &Entity<Project>, cx: &App) -> Vec<acp::McpServer> {
917 let context_server_store = project.read(cx).context_server_store().read(cx);
918 let is_local = project.read(cx).is_local();
919 context_server_store
920 .configured_server_ids()
921 .iter()
922 .filter_map(|id| {
923 let configuration = context_server_store.configuration_for_server(id)?;
924 match &*configuration {
925 project::context_server_store::ContextServerConfiguration::Custom {
926 command,
927 remote,
928 ..
929 }
930 | project::context_server_store::ContextServerConfiguration::Extension {
931 command,
932 remote,
933 ..
934 } if is_local || *remote => Some(acp::McpServer::Stdio(
935 acp::McpServerStdio::new(id.0.to_string(), &command.path)
936 .args(command.args.clone())
937 .env(if let Some(env) = command.env.as_ref() {
938 env.iter()
939 .map(|(name, value)| acp::EnvVariable::new(name, value))
940 .collect()
941 } else {
942 vec![]
943 }),
944 )),
945 project::context_server_store::ContextServerConfiguration::Http {
946 url,
947 headers,
948 timeout: _,
949 } => Some(acp::McpServer::Http(
950 acp::McpServerHttp::new(id.0.to_string(), url.to_string()).headers(
951 headers
952 .iter()
953 .map(|(name, value)| acp::HttpHeader::new(name, value))
954 .collect(),
955 ),
956 )),
957 _ => None,
958 }
959 })
960 .collect()
961}
962
963fn config_state(
964 modes: Option<acp::SessionModeState>,
965 models: Option<acp::SessionModelState>,
966 config_options: Option<Vec<acp::SessionConfigOption>>,
967) -> (
968 Option<Rc<RefCell<acp::SessionModeState>>>,
969 Option<Rc<RefCell<acp::SessionModelState>>>,
970 Option<Rc<RefCell<Vec<acp::SessionConfigOption>>>>,
971) {
972 if let Some(opts) = config_options {
973 return (None, None, Some(Rc::new(RefCell::new(opts))));
974 }
975
976 let modes = modes.map(|modes| Rc::new(RefCell::new(modes)));
977 let models = models.map(|models| Rc::new(RefCell::new(models)));
978 (modes, models, None)
979}
980
981struct AcpSessionModes {
982 session_id: acp::SessionId,
983 connection: Rc<acp::ClientSideConnection>,
984 state: Rc<RefCell<acp::SessionModeState>>,
985}
986
987impl acp_thread::AgentSessionModes for AcpSessionModes {
988 fn current_mode(&self) -> acp::SessionModeId {
989 self.state.borrow().current_mode_id.clone()
990 }
991
992 fn all_modes(&self) -> Vec<acp::SessionMode> {
993 self.state.borrow().available_modes.clone()
994 }
995
996 fn set_mode(&self, mode_id: acp::SessionModeId, cx: &mut App) -> Task<Result<()>> {
997 let connection = self.connection.clone();
998 let session_id = self.session_id.clone();
999 let old_mode_id;
1000 {
1001 let mut state = self.state.borrow_mut();
1002 old_mode_id = state.current_mode_id.clone();
1003 state.current_mode_id = mode_id.clone();
1004 };
1005 let state = self.state.clone();
1006 cx.foreground_executor().spawn(async move {
1007 let result = connection
1008 .set_session_mode(acp::SetSessionModeRequest::new(session_id, mode_id))
1009 .await;
1010
1011 if result.is_err() {
1012 state.borrow_mut().current_mode_id = old_mode_id;
1013 }
1014
1015 result?;
1016
1017 Ok(())
1018 })
1019 }
1020}
1021
1022struct AcpModelSelector {
1023 session_id: acp::SessionId,
1024 connection: Rc<acp::ClientSideConnection>,
1025 state: Rc<RefCell<acp::SessionModelState>>,
1026}
1027
1028impl AcpModelSelector {
1029 fn new(
1030 session_id: acp::SessionId,
1031 connection: Rc<acp::ClientSideConnection>,
1032 state: Rc<RefCell<acp::SessionModelState>>,
1033 ) -> Self {
1034 Self {
1035 session_id,
1036 connection,
1037 state,
1038 }
1039 }
1040}
1041
1042impl acp_thread::AgentModelSelector for AcpModelSelector {
1043 fn list_models(&self, _cx: &mut App) -> Task<Result<acp_thread::AgentModelList>> {
1044 Task::ready(Ok(acp_thread::AgentModelList::Flat(
1045 self.state
1046 .borrow()
1047 .available_models
1048 .clone()
1049 .into_iter()
1050 .map(acp_thread::AgentModelInfo::from)
1051 .collect(),
1052 )))
1053 }
1054
1055 fn select_model(&self, model_id: acp::ModelId, cx: &mut App) -> Task<Result<()>> {
1056 let connection = self.connection.clone();
1057 let session_id = self.session_id.clone();
1058 let old_model_id;
1059 {
1060 let mut state = self.state.borrow_mut();
1061 old_model_id = state.current_model_id.clone();
1062 state.current_model_id = model_id.clone();
1063 };
1064 let state = self.state.clone();
1065 cx.foreground_executor().spawn(async move {
1066 let result = connection
1067 .set_session_model(acp::SetSessionModelRequest::new(session_id, model_id))
1068 .await;
1069
1070 if result.is_err() {
1071 state.borrow_mut().current_model_id = old_model_id;
1072 }
1073
1074 result?;
1075
1076 Ok(())
1077 })
1078 }
1079
1080 fn selected_model(&self, _cx: &mut App) -> Task<Result<acp_thread::AgentModelInfo>> {
1081 let state = self.state.borrow();
1082 Task::ready(
1083 state
1084 .available_models
1085 .iter()
1086 .find(|m| m.model_id == state.current_model_id)
1087 .cloned()
1088 .map(acp_thread::AgentModelInfo::from)
1089 .ok_or_else(|| anyhow::anyhow!("Model not found")),
1090 )
1091 }
1092}
1093
1094struct AcpSessionConfigOptions {
1095 session_id: acp::SessionId,
1096 connection: Rc<acp::ClientSideConnection>,
1097 state: Rc<RefCell<Vec<acp::SessionConfigOption>>>,
1098 watch_tx: Rc<RefCell<watch::Sender<()>>>,
1099 watch_rx: watch::Receiver<()>,
1100}
1101
1102impl acp_thread::AgentSessionConfigOptions for AcpSessionConfigOptions {
1103 fn config_options(&self) -> Vec<acp::SessionConfigOption> {
1104 self.state.borrow().clone()
1105 }
1106
1107 fn set_config_option(
1108 &self,
1109 config_id: acp::SessionConfigId,
1110 value: acp::SessionConfigValueId,
1111 cx: &mut App,
1112 ) -> Task<Result<Vec<acp::SessionConfigOption>>> {
1113 let connection = self.connection.clone();
1114 let session_id = self.session_id.clone();
1115 let state = self.state.clone();
1116
1117 let watch_tx = self.watch_tx.clone();
1118
1119 cx.foreground_executor().spawn(async move {
1120 let response = connection
1121 .set_session_config_option(acp::SetSessionConfigOptionRequest::new(
1122 session_id, config_id, value,
1123 ))
1124 .await?;
1125
1126 *state.borrow_mut() = response.config_options.clone();
1127 watch_tx.borrow_mut().send(()).ok();
1128 Ok(response.config_options)
1129 })
1130 }
1131
1132 fn watch(&self, _cx: &mut App) -> Option<watch::Receiver<()>> {
1133 Some(self.watch_rx.clone())
1134 }
1135}
1136
1137struct ClientDelegate {
1138 sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
1139 session_list: Rc<RefCell<Option<Rc<AcpSessionList>>>>,
1140 cx: AsyncApp,
1141}
1142
1143#[async_trait::async_trait(?Send)]
1144impl acp::Client for ClientDelegate {
1145 async fn request_permission(
1146 &self,
1147 arguments: acp::RequestPermissionRequest,
1148 ) -> Result<acp::RequestPermissionResponse, acp::Error> {
1149 let thread;
1150 {
1151 let sessions_ref = self.sessions.borrow();
1152 let session = sessions_ref
1153 .get(&arguments.session_id)
1154 .context("Failed to get session")?;
1155 thread = session.thread.clone();
1156 }
1157
1158 let cx = &mut self.cx.clone();
1159
1160 let task = thread.update(cx, |thread, cx| {
1161 thread.request_tool_call_authorization(
1162 arguments.tool_call,
1163 acp_thread::PermissionOptions::Flat(arguments.options),
1164 cx,
1165 )
1166 })??;
1167
1168 let outcome = task.await;
1169
1170 Ok(acp::RequestPermissionResponse::new(outcome))
1171 }
1172
1173 async fn write_text_file(
1174 &self,
1175 arguments: acp::WriteTextFileRequest,
1176 ) -> Result<acp::WriteTextFileResponse, acp::Error> {
1177 let cx = &mut self.cx.clone();
1178 let task = self
1179 .session_thread(&arguments.session_id)?
1180 .update(cx, |thread, cx| {
1181 thread.write_text_file(arguments.path, arguments.content, cx)
1182 })?;
1183
1184 task.await?;
1185
1186 Ok(Default::default())
1187 }
1188
1189 async fn read_text_file(
1190 &self,
1191 arguments: acp::ReadTextFileRequest,
1192 ) -> Result<acp::ReadTextFileResponse, acp::Error> {
1193 let task = self.session_thread(&arguments.session_id)?.update(
1194 &mut self.cx.clone(),
1195 |thread, cx| {
1196 thread.read_text_file(arguments.path, arguments.line, arguments.limit, false, cx)
1197 },
1198 )?;
1199
1200 let content = task.await?;
1201
1202 Ok(acp::ReadTextFileResponse::new(content))
1203 }
1204
1205 async fn session_notification(
1206 &self,
1207 notification: acp::SessionNotification,
1208 ) -> Result<(), acp::Error> {
1209 let sessions = self.sessions.borrow();
1210 let session = sessions
1211 .get(¬ification.session_id)
1212 .context("Failed to get session")?;
1213
1214 if let acp::SessionUpdate::CurrentModeUpdate(acp::CurrentModeUpdate {
1215 current_mode_id,
1216 ..
1217 }) = ¬ification.update
1218 {
1219 if let Some(session_modes) = &session.session_modes {
1220 session_modes.borrow_mut().current_mode_id = current_mode_id.clone();
1221 }
1222 }
1223
1224 if let acp::SessionUpdate::ConfigOptionUpdate(acp::ConfigOptionUpdate {
1225 config_options,
1226 ..
1227 }) = ¬ification.update
1228 {
1229 if let Some(opts) = &session.config_options {
1230 *opts.config_options.borrow_mut() = config_options.clone();
1231 opts.tx.borrow_mut().send(()).ok();
1232 }
1233 }
1234
1235 if let acp::SessionUpdate::SessionInfoUpdate(info_update) = ¬ification.update
1236 && let Some(session_list) = self.session_list.borrow().as_ref()
1237 {
1238 session_list.send_info_update(notification.session_id.clone(), info_update.clone());
1239 }
1240
1241 // Clone so we can inspect meta both before and after handing off to the thread
1242 let update_clone = notification.update.clone();
1243
1244 // Pre-handle: if a ToolCall carries terminal_info, create/register a display-only terminal.
1245 if let acp::SessionUpdate::ToolCall(tc) = &update_clone {
1246 if let Some(meta) = &tc.meta {
1247 if let Some(terminal_info) = meta.get("terminal_info") {
1248 if let Some(id_str) = terminal_info.get("terminal_id").and_then(|v| v.as_str())
1249 {
1250 let terminal_id = acp::TerminalId::new(id_str);
1251 let cwd = terminal_info
1252 .get("cwd")
1253 .and_then(|v| v.as_str().map(PathBuf::from));
1254
1255 // Create a minimal display-only lower-level terminal and register it.
1256 let _ = session.thread.update(&mut self.cx.clone(), |thread, cx| {
1257 let builder = TerminalBuilder::new_display_only(
1258 CursorShape::default(),
1259 AlternateScroll::On,
1260 None,
1261 0,
1262 cx.background_executor(),
1263 thread.project().read(cx).path_style(cx),
1264 )?;
1265 let lower = cx.new(|cx| builder.subscribe(cx));
1266 thread.on_terminal_provider_event(
1267 TerminalProviderEvent::Created {
1268 terminal_id,
1269 label: tc.title.clone(),
1270 cwd,
1271 output_byte_limit: None,
1272 terminal: lower,
1273 },
1274 cx,
1275 );
1276 anyhow::Ok(())
1277 });
1278 }
1279 }
1280 }
1281 }
1282
1283 // Forward the update to the acp_thread as usual.
1284 session.thread.update(&mut self.cx.clone(), |thread, cx| {
1285 thread.handle_session_update(notification.update.clone(), cx)
1286 })??;
1287
1288 // Post-handle: stream terminal output/exit if present on ToolCallUpdate meta.
1289 if let acp::SessionUpdate::ToolCallUpdate(tcu) = &update_clone {
1290 if let Some(meta) = &tcu.meta {
1291 if let Some(term_out) = meta.get("terminal_output") {
1292 if let Some(id_str) = term_out.get("terminal_id").and_then(|v| v.as_str()) {
1293 let terminal_id = acp::TerminalId::new(id_str);
1294 if let Some(s) = term_out.get("data").and_then(|v| v.as_str()) {
1295 let data = s.as_bytes().to_vec();
1296 let _ = session.thread.update(&mut self.cx.clone(), |thread, cx| {
1297 thread.on_terminal_provider_event(
1298 TerminalProviderEvent::Output { terminal_id, data },
1299 cx,
1300 );
1301 });
1302 }
1303 }
1304 }
1305
1306 // terminal_exit
1307 if let Some(term_exit) = meta.get("terminal_exit") {
1308 if let Some(id_str) = term_exit.get("terminal_id").and_then(|v| v.as_str()) {
1309 let terminal_id = acp::TerminalId::new(id_str);
1310 let status = acp::TerminalExitStatus::new()
1311 .exit_code(
1312 term_exit
1313 .get("exit_code")
1314 .and_then(|v| v.as_u64())
1315 .map(|i| i as u32),
1316 )
1317 .signal(
1318 term_exit
1319 .get("signal")
1320 .and_then(|v| v.as_str().map(|s| s.to_string())),
1321 );
1322
1323 let _ = session.thread.update(&mut self.cx.clone(), |thread, cx| {
1324 thread.on_terminal_provider_event(
1325 TerminalProviderEvent::Exit {
1326 terminal_id,
1327 status,
1328 },
1329 cx,
1330 );
1331 });
1332 }
1333 }
1334 }
1335 }
1336
1337 Ok(())
1338 }
1339
1340 async fn create_terminal(
1341 &self,
1342 args: acp::CreateTerminalRequest,
1343 ) -> Result<acp::CreateTerminalResponse, acp::Error> {
1344 let thread = self.session_thread(&args.session_id)?;
1345 let project = thread.read_with(&self.cx, |thread, _cx| thread.project().clone())?;
1346
1347 let terminal_entity = acp_thread::create_terminal_entity(
1348 args.command.clone(),
1349 &args.args,
1350 args.env
1351 .into_iter()
1352 .map(|env| (env.name, env.value))
1353 .collect(),
1354 args.cwd.clone(),
1355 &project,
1356 &mut self.cx.clone(),
1357 )
1358 .await?;
1359
1360 // Register with renderer
1361 let terminal_entity = thread.update(&mut self.cx.clone(), |thread, cx| {
1362 thread.register_terminal_created(
1363 acp::TerminalId::new(uuid::Uuid::new_v4().to_string()),
1364 format!("{} {}", args.command, args.args.join(" ")),
1365 args.cwd.clone(),
1366 args.output_byte_limit,
1367 terminal_entity,
1368 cx,
1369 )
1370 })?;
1371 let terminal_id = terminal_entity.read_with(&self.cx, |terminal, _| terminal.id().clone());
1372 Ok(acp::CreateTerminalResponse::new(terminal_id))
1373 }
1374
1375 async fn kill_terminal_command(
1376 &self,
1377 args: acp::KillTerminalCommandRequest,
1378 ) -> Result<acp::KillTerminalCommandResponse, acp::Error> {
1379 self.session_thread(&args.session_id)?
1380 .update(&mut self.cx.clone(), |thread, cx| {
1381 thread.kill_terminal(args.terminal_id, cx)
1382 })??;
1383
1384 Ok(Default::default())
1385 }
1386
1387 async fn ext_method(&self, _args: acp::ExtRequest) -> Result<acp::ExtResponse, acp::Error> {
1388 Err(acp::Error::method_not_found())
1389 }
1390
1391 async fn ext_notification(&self, _args: acp::ExtNotification) -> Result<(), acp::Error> {
1392 Err(acp::Error::method_not_found())
1393 }
1394
1395 async fn release_terminal(
1396 &self,
1397 args: acp::ReleaseTerminalRequest,
1398 ) -> Result<acp::ReleaseTerminalResponse, acp::Error> {
1399 self.session_thread(&args.session_id)?
1400 .update(&mut self.cx.clone(), |thread, cx| {
1401 thread.release_terminal(args.terminal_id, cx)
1402 })??;
1403
1404 Ok(Default::default())
1405 }
1406
1407 async fn terminal_output(
1408 &self,
1409 args: acp::TerminalOutputRequest,
1410 ) -> Result<acp::TerminalOutputResponse, acp::Error> {
1411 self.session_thread(&args.session_id)?
1412 .read_with(&mut self.cx.clone(), |thread, cx| {
1413 let out = thread
1414 .terminal(args.terminal_id)?
1415 .read(cx)
1416 .current_output(cx);
1417
1418 Ok(out)
1419 })?
1420 }
1421
1422 async fn wait_for_terminal_exit(
1423 &self,
1424 args: acp::WaitForTerminalExitRequest,
1425 ) -> Result<acp::WaitForTerminalExitResponse, acp::Error> {
1426 let exit_status = self
1427 .session_thread(&args.session_id)?
1428 .update(&mut self.cx.clone(), |thread, cx| {
1429 anyhow::Ok(thread.terminal(args.terminal_id)?.read(cx).wait_for_exit())
1430 })??
1431 .await;
1432
1433 Ok(acp::WaitForTerminalExitResponse::new(exit_status))
1434 }
1435}
1436
1437impl ClientDelegate {
1438 fn session_thread(&self, session_id: &acp::SessionId) -> Result<WeakEntity<AcpThread>> {
1439 let sessions = self.sessions.borrow();
1440 sessions
1441 .get(session_id)
1442 .context("Failed to get session")
1443 .map(|session| session.thread.clone())
1444 }
1445}