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