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