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