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