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 futures::AsyncBufReadExt as _;
8use futures::io::BufReader;
9use project::Project;
10use project::agent_server_store::AgentServerCommand;
11use serde::Deserialize;
12use util::ResultExt as _;
13
14use std::path::PathBuf;
15use std::{any::Any, cell::RefCell};
16use std::{path::Path, rc::Rc};
17use thiserror::Error;
18
19use anyhow::{Context as _, Result};
20use gpui::{App, AppContext as _, AsyncApp, Entity, SharedString, Task, WeakEntity};
21
22use acp_thread::{AcpThread, AuthRequired, LoadError, TerminalProviderEvent};
23use terminal::TerminalBuilder;
24use terminal::terminal_settings::{AlternateScroll, CursorShape};
25
26#[derive(Debug, Error)]
27#[error("Unsupported version")]
28pub struct UnsupportedVersion;
29
30pub struct AcpConnection {
31 server_name: SharedString,
32 connection: Rc<acp::ClientSideConnection>,
33 sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
34 auth_methods: Vec<acp::AuthMethod>,
35 agent_capabilities: acp::AgentCapabilities,
36 default_mode: Option<acp::SessionModeId>,
37 root_dir: PathBuf,
38 // NB: Don't move this into the wait_task, since we need to ensure the process is
39 // killed on drop (setting kill_on_drop on the command seems to not always work).
40 child: smol::process::Child,
41 _io_task: Task<Result<()>>,
42 _wait_task: Task<Result<()>>,
43 _stderr_task: Task<Result<()>>,
44}
45
46pub struct AcpSession {
47 thread: WeakEntity<AcpThread>,
48 suppress_abort_err: bool,
49 models: Option<Rc<RefCell<acp::SessionModelState>>>,
50 session_modes: Option<Rc<RefCell<acp::SessionModeState>>>,
51}
52
53pub async fn connect(
54 server_name: SharedString,
55 command: AgentServerCommand,
56 root_dir: &Path,
57 default_mode: Option<acp::SessionModeId>,
58 is_remote: bool,
59 cx: &mut AsyncApp,
60) -> Result<Rc<dyn AgentConnection>> {
61 let conn = AcpConnection::stdio(
62 server_name,
63 command.clone(),
64 root_dir,
65 default_mode,
66 is_remote,
67 cx,
68 )
69 .await?;
70 Ok(Rc::new(conn) as _)
71}
72
73const MINIMUM_SUPPORTED_VERSION: acp::ProtocolVersion = acp::V1;
74
75impl AcpConnection {
76 pub async fn stdio(
77 server_name: SharedString,
78 command: AgentServerCommand,
79 root_dir: &Path,
80 default_mode: Option<acp::SessionModeId>,
81 is_remote: bool,
82 cx: &mut AsyncApp,
83 ) -> Result<Self> {
84 let mut child = util::command::new_smol_command(command.path);
85 child
86 .args(command.args.iter().map(|arg| arg.as_str()))
87 .envs(command.env.iter().flatten())
88 .stdin(std::process::Stdio::piped())
89 .stdout(std::process::Stdio::piped())
90 .stderr(std::process::Stdio::piped());
91 if !is_remote {
92 child.current_dir(root_dir);
93 }
94 let mut child = child.spawn()?;
95
96 let stdout = child.stdout.take().context("Failed to take stdout")?;
97 let stdin = child.stdin.take().context("Failed to take stdin")?;
98 let stderr = child.stderr.take().context("Failed to take stderr")?;
99 log::trace!("Spawned (pid: {})", child.id());
100
101 let sessions = Rc::new(RefCell::new(HashMap::default()));
102
103 let client = ClientDelegate {
104 sessions: sessions.clone(),
105 cx: cx.clone(),
106 };
107 let (connection, io_task) = acp::ClientSideConnection::new(client, stdin, stdout, {
108 let foreground_executor = cx.foreground_executor().clone();
109 move |fut| {
110 foreground_executor.spawn(fut).detach();
111 }
112 });
113
114 let io_task = cx.background_spawn(io_task);
115
116 let stderr_task = cx.background_spawn(async move {
117 let mut stderr = BufReader::new(stderr);
118 let mut line = String::new();
119 while let Ok(n) = stderr.read_line(&mut line).await
120 && n > 0
121 {
122 log::warn!("agent stderr: {}", &line);
123 line.clear();
124 }
125 Ok(())
126 });
127
128 let wait_task = cx.spawn({
129 let sessions = sessions.clone();
130 let status_fut = child.status();
131 async move |cx| {
132 let status = status_fut.await?;
133
134 for session in sessions.borrow().values() {
135 session
136 .thread
137 .update(cx, |thread, cx| {
138 thread.emit_load_error(LoadError::Exited { status }, cx)
139 })
140 .ok();
141 }
142
143 anyhow::Ok(())
144 }
145 });
146
147 let connection = Rc::new(connection);
148
149 cx.update(|cx| {
150 AcpConnectionRegistry::default_global(cx).update(cx, |registry, cx| {
151 registry.set_active_connection(server_name.clone(), &connection, cx)
152 });
153 })?;
154
155 let response = connection
156 .initialize(acp::InitializeRequest {
157 protocol_version: acp::VERSION,
158 client_capabilities: acp::ClientCapabilities {
159 fs: acp::FileSystemCapability {
160 read_text_file: true,
161 write_text_file: true,
162 meta: None,
163 },
164 terminal: true,
165 meta: None,
166 },
167 meta: None,
168 })
169 .await?;
170
171 if response.protocol_version < MINIMUM_SUPPORTED_VERSION {
172 return Err(UnsupportedVersion.into());
173 }
174
175 Ok(Self {
176 auth_methods: response.auth_methods,
177 root_dir: root_dir.to_owned(),
178 connection,
179 server_name,
180 sessions,
181 agent_capabilities: response.agent_capabilities,
182 default_mode,
183 _io_task: io_task,
184 _wait_task: wait_task,
185 _stderr_task: stderr_task,
186 child,
187 })
188 }
189
190 pub fn prompt_capabilities(&self) -> &acp::PromptCapabilities {
191 &self.agent_capabilities.prompt_capabilities
192 }
193
194 pub fn root_dir(&self) -> &Path {
195 &self.root_dir
196 }
197}
198
199impl Drop for AcpConnection {
200 fn drop(&mut self) {
201 // See the comment on the child field.
202 self.child.kill().log_err();
203 }
204}
205
206impl AgentConnection for AcpConnection {
207 fn new_thread(
208 self: Rc<Self>,
209 project: Entity<Project>,
210 cwd: &Path,
211 cx: &mut App,
212 ) -> Task<Result<Entity<AcpThread>>> {
213 let name = self.server_name.clone();
214 let conn = self.connection.clone();
215 let sessions = self.sessions.clone();
216 let default_mode = self.default_mode.clone();
217 let cwd = cwd.to_path_buf();
218 let context_server_store = project.read(cx).context_server_store().read(cx);
219 let mcp_servers = if project.read(cx).is_local() {
220 context_server_store
221 .configured_server_ids()
222 .iter()
223 .filter_map(|id| {
224 let configuration = context_server_store.configuration_for_server(id)?;
225 let command = configuration.command();
226 Some(acp::McpServer::Stdio {
227 name: id.0.to_string(),
228 command: command.path.clone(),
229 args: command.args.clone(),
230 env: if let Some(env) = command.env.as_ref() {
231 env.iter()
232 .map(|(name, value)| acp::EnvVariable {
233 name: name.clone(),
234 value: value.clone(),
235 meta: None,
236 })
237 .collect()
238 } else {
239 vec![]
240 },
241 })
242 })
243 .collect()
244 } else {
245 // In SSH projects, the external agent is running on the remote
246 // machine, and currently we only run MCP servers on the local
247 // machine. So don't pass any MCP servers to the agent in that case.
248 Vec::new()
249 };
250
251 cx.spawn(async move |cx| {
252 let response = conn
253 .new_session(acp::NewSessionRequest { mcp_servers, cwd, meta: None })
254 .await
255 .map_err(|err| {
256 if err.code == acp::ErrorCode::AUTH_REQUIRED.code {
257 let mut error = AuthRequired::new();
258
259 if err.message != acp::ErrorCode::AUTH_REQUIRED.message {
260 error = error.with_description(err.message);
261 }
262
263 anyhow!(error)
264 } else {
265 anyhow!(err)
266 }
267 })?;
268
269 let modes = response.modes.map(|modes| Rc::new(RefCell::new(modes)));
270 let models = response.models.map(|models| Rc::new(RefCell::new(models)));
271
272 if let Some(default_mode) = default_mode {
273 if let Some(modes) = modes.as_ref() {
274 let mut modes_ref = modes.borrow_mut();
275 let has_mode = modes_ref.available_modes.iter().any(|mode| mode.id == default_mode);
276
277 if has_mode {
278 let initial_mode_id = modes_ref.current_mode_id.clone();
279
280 cx.spawn({
281 let default_mode = default_mode.clone();
282 let session_id = response.session_id.clone();
283 let modes = modes.clone();
284 async move |_| {
285 let result = conn.set_session_mode(acp::SetSessionModeRequest {
286 session_id,
287 mode_id: default_mode,
288 meta: None,
289 })
290 .await.log_err();
291
292 if result.is_none() {
293 modes.borrow_mut().current_mode_id = initial_mode_id;
294 }
295 }
296 }).detach();
297
298 modes_ref.current_mode_id = default_mode;
299 } else {
300 let available_modes = modes_ref
301 .available_modes
302 .iter()
303 .map(|mode| format!("- `{}`: {}", mode.id, mode.name))
304 .collect::<Vec<_>>()
305 .join("\n");
306
307 log::warn!(
308 "`{default_mode}` is not valid {name} mode. Available options:\n{available_modes}",
309 );
310 }
311 } else {
312 log::warn!(
313 "`{name}` does not support modes, but `default_mode` was set in settings.",
314 );
315 }
316 }
317
318 let session_id = response.session_id;
319 let action_log = cx.new(|_| ActionLog::new(project.clone()))?;
320 let thread = cx.new(|cx| {
321 AcpThread::new(
322 self.server_name.clone(),
323 self.clone(),
324 project,
325 action_log,
326 session_id.clone(),
327 // ACP doesn't currently support per-session prompt capabilities or changing capabilities dynamically.
328 watch::Receiver::constant(self.agent_capabilities.prompt_capabilities.clone()),
329 cx,
330 )
331 })?;
332
333
334 let session = AcpSession {
335 thread: thread.downgrade(),
336 suppress_abort_err: false,
337 session_modes: modes,
338 models,
339 };
340 sessions.borrow_mut().insert(session_id, session);
341
342 Ok(thread)
343 })
344 }
345
346 fn auth_methods(&self) -> &[acp::AuthMethod] {
347 &self.auth_methods
348 }
349
350 fn authenticate(&self, method_id: acp::AuthMethodId, cx: &mut App) -> Task<Result<()>> {
351 let conn = self.connection.clone();
352 cx.foreground_executor().spawn(async move {
353 conn.authenticate(acp::AuthenticateRequest {
354 method_id: method_id.clone(),
355 meta: None,
356 })
357 .await?;
358
359 Ok(())
360 })
361 }
362
363 fn prompt(
364 &self,
365 _id: Option<acp_thread::UserMessageId>,
366 params: acp::PromptRequest,
367 cx: &mut App,
368 ) -> Task<Result<acp::PromptResponse>> {
369 let conn = self.connection.clone();
370 let sessions = self.sessions.clone();
371 let session_id = params.session_id.clone();
372 cx.foreground_executor().spawn(async move {
373 let result = conn.prompt(params).await;
374
375 let mut suppress_abort_err = false;
376
377 if let Some(session) = sessions.borrow_mut().get_mut(&session_id) {
378 suppress_abort_err = session.suppress_abort_err;
379 session.suppress_abort_err = false;
380 }
381
382 match result {
383 Ok(response) => Ok(response),
384 Err(err) => {
385 if err.code == acp::ErrorCode::AUTH_REQUIRED.code {
386 return Err(anyhow!(acp::Error::auth_required()));
387 }
388
389 if err.code != ErrorCode::INTERNAL_ERROR.code {
390 anyhow::bail!(err)
391 }
392
393 let Some(data) = &err.data else {
394 anyhow::bail!(err)
395 };
396
397 // Temporary workaround until the following PR is generally available:
398 // https://github.com/google-gemini/gemini-cli/pull/6656
399
400 #[derive(Deserialize)]
401 #[serde(deny_unknown_fields)]
402 struct ErrorDetails {
403 details: Box<str>,
404 }
405
406 match serde_json::from_value(data.clone()) {
407 Ok(ErrorDetails { details }) => {
408 if suppress_abort_err
409 && (details.contains("This operation was aborted")
410 || details.contains("The user aborted a request"))
411 {
412 Ok(acp::PromptResponse {
413 stop_reason: acp::StopReason::Cancelled,
414 meta: None,
415 })
416 } else {
417 Err(anyhow!(details))
418 }
419 }
420 Err(_) => Err(anyhow!(err)),
421 }
422 }
423 }
424 })
425 }
426
427 fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
428 if let Some(session) = self.sessions.borrow_mut().get_mut(session_id) {
429 session.suppress_abort_err = true;
430 }
431 let conn = self.connection.clone();
432 let params = acp::CancelNotification {
433 session_id: session_id.clone(),
434 meta: None,
435 };
436 cx.foreground_executor()
437 .spawn(async move { conn.cancel(params).await })
438 .detach();
439 }
440
441 fn session_modes(
442 &self,
443 session_id: &acp::SessionId,
444 _cx: &App,
445 ) -> Option<Rc<dyn acp_thread::AgentSessionModes>> {
446 let sessions = self.sessions.clone();
447 let sessions_ref = sessions.borrow();
448 let Some(session) = sessions_ref.get(session_id) else {
449 return None;
450 };
451
452 if let Some(modes) = session.session_modes.as_ref() {
453 Some(Rc::new(AcpSessionModes {
454 connection: self.connection.clone(),
455 session_id: session_id.clone(),
456 state: modes.clone(),
457 }) as _)
458 } else {
459 None
460 }
461 }
462
463 fn model_selector(
464 &self,
465 session_id: &acp::SessionId,
466 ) -> Option<Rc<dyn acp_thread::AgentModelSelector>> {
467 let sessions = self.sessions.clone();
468 let sessions_ref = sessions.borrow();
469 let Some(session) = sessions_ref.get(session_id) else {
470 return None;
471 };
472
473 if let Some(models) = session.models.as_ref() {
474 Some(Rc::new(AcpModelSelector::new(
475 session_id.clone(),
476 self.connection.clone(),
477 models.clone(),
478 )) as _)
479 } else {
480 None
481 }
482 }
483
484 fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
485 self
486 }
487}
488
489struct AcpSessionModes {
490 session_id: acp::SessionId,
491 connection: Rc<acp::ClientSideConnection>,
492 state: Rc<RefCell<acp::SessionModeState>>,
493}
494
495impl acp_thread::AgentSessionModes for AcpSessionModes {
496 fn current_mode(&self) -> acp::SessionModeId {
497 self.state.borrow().current_mode_id.clone()
498 }
499
500 fn all_modes(&self) -> Vec<acp::SessionMode> {
501 self.state.borrow().available_modes.clone()
502 }
503
504 fn set_mode(&self, mode_id: acp::SessionModeId, cx: &mut App) -> Task<Result<()>> {
505 let connection = self.connection.clone();
506 let session_id = self.session_id.clone();
507 let old_mode_id;
508 {
509 let mut state = self.state.borrow_mut();
510 old_mode_id = state.current_mode_id.clone();
511 state.current_mode_id = mode_id.clone();
512 };
513 let state = self.state.clone();
514 cx.foreground_executor().spawn(async move {
515 let result = connection
516 .set_session_mode(acp::SetSessionModeRequest {
517 session_id,
518 mode_id,
519 meta: None,
520 })
521 .await;
522
523 if result.is_err() {
524 state.borrow_mut().current_mode_id = old_mode_id;
525 }
526
527 result?;
528
529 Ok(())
530 })
531 }
532}
533
534struct AcpModelSelector {
535 session_id: acp::SessionId,
536 connection: Rc<acp::ClientSideConnection>,
537 state: Rc<RefCell<acp::SessionModelState>>,
538}
539
540impl AcpModelSelector {
541 fn new(
542 session_id: acp::SessionId,
543 connection: Rc<acp::ClientSideConnection>,
544 state: Rc<RefCell<acp::SessionModelState>>,
545 ) -> Self {
546 Self {
547 session_id,
548 connection,
549 state,
550 }
551 }
552}
553
554impl acp_thread::AgentModelSelector for AcpModelSelector {
555 fn list_models(&self, _cx: &mut App) -> Task<Result<acp_thread::AgentModelList>> {
556 Task::ready(Ok(acp_thread::AgentModelList::Flat(
557 self.state
558 .borrow()
559 .available_models
560 .clone()
561 .into_iter()
562 .map(acp_thread::AgentModelInfo::from)
563 .collect(),
564 )))
565 }
566
567 fn select_model(&self, model_id: acp::ModelId, cx: &mut App) -> Task<Result<()>> {
568 let connection = self.connection.clone();
569 let session_id = self.session_id.clone();
570 let old_model_id;
571 {
572 let mut state = self.state.borrow_mut();
573 old_model_id = state.current_model_id.clone();
574 state.current_model_id = model_id.clone();
575 };
576 let state = self.state.clone();
577 cx.foreground_executor().spawn(async move {
578 let result = connection
579 .set_session_model(acp::SetSessionModelRequest {
580 session_id,
581 model_id,
582 meta: None,
583 })
584 .await;
585
586 if result.is_err() {
587 state.borrow_mut().current_model_id = old_model_id;
588 }
589
590 result?;
591
592 Ok(())
593 })
594 }
595
596 fn selected_model(&self, _cx: &mut App) -> Task<Result<acp_thread::AgentModelInfo>> {
597 let state = self.state.borrow();
598 Task::ready(
599 state
600 .available_models
601 .iter()
602 .find(|m| m.model_id == state.current_model_id)
603 .cloned()
604 .map(acp_thread::AgentModelInfo::from)
605 .ok_or_else(|| anyhow::anyhow!("Model not found")),
606 )
607 }
608}
609
610struct ClientDelegate {
611 sessions: Rc<RefCell<HashMap<acp::SessionId, AcpSession>>>,
612 cx: AsyncApp,
613}
614
615#[async_trait::async_trait(?Send)]
616impl acp::Client for ClientDelegate {
617 async fn request_permission(
618 &self,
619 arguments: acp::RequestPermissionRequest,
620 ) -> Result<acp::RequestPermissionResponse, acp::Error> {
621 let respect_always_allow_setting;
622 let thread;
623 {
624 let sessions_ref = self.sessions.borrow();
625 let session = sessions_ref
626 .get(&arguments.session_id)
627 .context("Failed to get session")?;
628 respect_always_allow_setting = session.session_modes.is_none();
629 thread = session.thread.clone();
630 }
631
632 let cx = &mut self.cx.clone();
633
634 let task = thread.update(cx, |thread, cx| {
635 thread.request_tool_call_authorization(
636 arguments.tool_call,
637 arguments.options,
638 respect_always_allow_setting,
639 cx,
640 )
641 })??;
642
643 let outcome = task.await;
644
645 Ok(acp::RequestPermissionResponse {
646 outcome,
647 meta: None,
648 })
649 }
650
651 async fn write_text_file(
652 &self,
653 arguments: acp::WriteTextFileRequest,
654 ) -> Result<acp::WriteTextFileResponse, acp::Error> {
655 let cx = &mut self.cx.clone();
656 let task = self
657 .session_thread(&arguments.session_id)?
658 .update(cx, |thread, cx| {
659 thread.write_text_file(arguments.path, arguments.content, cx)
660 })?;
661
662 task.await?;
663
664 Ok(Default::default())
665 }
666
667 async fn read_text_file(
668 &self,
669 arguments: acp::ReadTextFileRequest,
670 ) -> Result<acp::ReadTextFileResponse, acp::Error> {
671 let task = self.session_thread(&arguments.session_id)?.update(
672 &mut self.cx.clone(),
673 |thread, cx| {
674 thread.read_text_file(arguments.path, arguments.line, arguments.limit, false, cx)
675 },
676 )?;
677
678 let content = task.await?;
679
680 Ok(acp::ReadTextFileResponse {
681 content,
682 meta: None,
683 })
684 }
685
686 async fn session_notification(
687 &self,
688 notification: acp::SessionNotification,
689 ) -> Result<(), acp::Error> {
690 let sessions = self.sessions.borrow();
691 let session = sessions
692 .get(¬ification.session_id)
693 .context("Failed to get session")?;
694
695 if let acp::SessionUpdate::CurrentModeUpdate { current_mode_id } = ¬ification.update {
696 if let Some(session_modes) = &session.session_modes {
697 session_modes.borrow_mut().current_mode_id = current_mode_id.clone();
698 } else {
699 log::error!(
700 "Got a `CurrentModeUpdate` notification, but they agent didn't specify `modes` during setting setup."
701 );
702 }
703 }
704
705 // Clone so we can inspect meta both before and after handing off to the thread
706 let update_clone = notification.update.clone();
707
708 // Pre-handle: if a ToolCall carries terminal_info, create/register a display-only terminal.
709 if let acp::SessionUpdate::ToolCall(tc) = &update_clone {
710 if let Some(meta) = &tc.meta {
711 if let Some(terminal_info) = meta.get("terminal_info") {
712 if let Some(id_str) = terminal_info.get("terminal_id").and_then(|v| v.as_str())
713 {
714 let terminal_id = acp::TerminalId(id_str.into());
715
716 // Create a minimal display-only lower-level terminal and register it.
717 let _ = session.thread.update(&mut self.cx.clone(), |thread, cx| {
718 let builder = TerminalBuilder::new_display_only(
719 None,
720 CursorShape::default(),
721 AlternateScroll::On,
722 None,
723 0,
724 cx,
725 )?;
726 let lower = cx.new(|cx| builder.subscribe(cx));
727 thread.on_terminal_provider_event(
728 TerminalProviderEvent::Created {
729 terminal_id: terminal_id.clone(),
730 label: tc.title.clone(),
731 cwd: None,
732 output_byte_limit: None,
733 terminal: lower,
734 },
735 cx,
736 );
737 anyhow::Ok(())
738 });
739 }
740 }
741 }
742 }
743
744 // Forward the update to the acp_thread as usual.
745 session.thread.update(&mut self.cx.clone(), |thread, cx| {
746 thread.handle_session_update(notification.update.clone(), cx)
747 })??;
748
749 // Post-handle: stream terminal output/exit if present on ToolCallUpdate meta.
750 if let acp::SessionUpdate::ToolCallUpdate(tcu) = &update_clone {
751 if let Some(meta) = &tcu.meta {
752 if let Some(term_out) = meta.get("terminal_output") {
753 if let Some(id_str) = term_out.get("terminal_id").and_then(|v| v.as_str()) {
754 let terminal_id = acp::TerminalId(id_str.into());
755 if let Some(s) = term_out.get("data").and_then(|v| v.as_str()) {
756 let data = s.as_bytes().to_vec();
757 let _ = session.thread.update(&mut self.cx.clone(), |thread, cx| {
758 thread.on_terminal_provider_event(
759 TerminalProviderEvent::Output {
760 terminal_id: terminal_id.clone(),
761 data,
762 },
763 cx,
764 );
765 });
766 }
767 }
768 }
769
770 // terminal_exit
771 if let Some(term_exit) = meta.get("terminal_exit") {
772 if let Some(id_str) = term_exit.get("terminal_id").and_then(|v| v.as_str()) {
773 let terminal_id = acp::TerminalId(id_str.into());
774 let status = acp::TerminalExitStatus {
775 exit_code: term_exit
776 .get("exit_code")
777 .and_then(|v| v.as_u64())
778 .map(|i| i as u32),
779 signal: term_exit
780 .get("signal")
781 .and_then(|v| v.as_str().map(|s| s.to_string())),
782 meta: None,
783 };
784 let _ = session.thread.update(&mut self.cx.clone(), |thread, cx| {
785 thread.on_terminal_provider_event(
786 TerminalProviderEvent::Exit {
787 terminal_id: terminal_id.clone(),
788 status,
789 },
790 cx,
791 );
792 });
793 }
794 }
795 }
796 }
797
798 Ok(())
799 }
800
801 async fn create_terminal(
802 &self,
803 args: acp::CreateTerminalRequest,
804 ) -> Result<acp::CreateTerminalResponse, acp::Error> {
805 let thread = self.session_thread(&args.session_id)?;
806 let project = thread.read_with(&self.cx, |thread, _cx| thread.project().clone())?;
807
808 let mut env = if let Some(dir) = &args.cwd {
809 project
810 .update(&mut self.cx.clone(), |project, cx| {
811 project.directory_environment(&task::Shell::System, dir.clone().into(), cx)
812 })?
813 .await
814 .unwrap_or_default()
815 } else {
816 Default::default()
817 };
818 for var in args.env {
819 env.insert(var.name, var.value);
820 }
821
822 // Use remote shell or default system shell, as appropriate
823 let remote_shell = project.update(&mut self.cx.clone(), |project, cx| {
824 project
825 .remote_client()
826 .and_then(|r| r.read(cx).default_system_shell())
827 })?;
828 let (task_command, task_args) =
829 task::ShellBuilder::new(remote_shell.as_deref(), &task::Shell::System)
830 .redirect_stdin_to_dev_null()
831 .build(Some(args.command.clone()), &args.args);
832
833 let terminal_entity = project
834 .update(&mut self.cx.clone(), |project, cx| {
835 project.create_terminal_task(
836 task::SpawnInTerminal {
837 command: Some(task_command),
838 args: task_args,
839 cwd: args.cwd.clone(),
840 env,
841 ..Default::default()
842 },
843 cx,
844 )
845 })?
846 .await?;
847
848 // Register with renderer
849 let terminal_entity = thread.update(&mut self.cx.clone(), |thread, cx| {
850 thread.register_terminal_created(
851 acp::TerminalId(uuid::Uuid::new_v4().to_string().into()),
852 format!("{} {}", args.command, args.args.join(" ")),
853 args.cwd.clone(),
854 args.output_byte_limit,
855 terminal_entity,
856 cx,
857 )
858 })?;
859 let terminal_id =
860 terminal_entity.read_with(&self.cx, |terminal, _| terminal.id().clone())?;
861 Ok(acp::CreateTerminalResponse {
862 terminal_id,
863 meta: None,
864 })
865 }
866
867 async fn kill_terminal_command(
868 &self,
869 args: acp::KillTerminalCommandRequest,
870 ) -> Result<acp::KillTerminalCommandResponse, acp::Error> {
871 self.session_thread(&args.session_id)?
872 .update(&mut self.cx.clone(), |thread, cx| {
873 thread.kill_terminal(args.terminal_id, cx)
874 })??;
875
876 Ok(Default::default())
877 }
878
879 async fn ext_method(&self, _args: acp::ExtRequest) -> Result<acp::ExtResponse, acp::Error> {
880 Err(acp::Error::method_not_found())
881 }
882
883 async fn ext_notification(&self, _args: acp::ExtNotification) -> Result<(), acp::Error> {
884 Err(acp::Error::method_not_found())
885 }
886
887 async fn release_terminal(
888 &self,
889 args: acp::ReleaseTerminalRequest,
890 ) -> Result<acp::ReleaseTerminalResponse, acp::Error> {
891 self.session_thread(&args.session_id)?
892 .update(&mut self.cx.clone(), |thread, cx| {
893 thread.release_terminal(args.terminal_id, cx)
894 })??;
895
896 Ok(Default::default())
897 }
898
899 async fn terminal_output(
900 &self,
901 args: acp::TerminalOutputRequest,
902 ) -> Result<acp::TerminalOutputResponse, acp::Error> {
903 self.session_thread(&args.session_id)?
904 .read_with(&mut self.cx.clone(), |thread, cx| {
905 let out = thread
906 .terminal(args.terminal_id)?
907 .read(cx)
908 .current_output(cx);
909
910 Ok(out)
911 })?
912 }
913
914 async fn wait_for_terminal_exit(
915 &self,
916 args: acp::WaitForTerminalExitRequest,
917 ) -> Result<acp::WaitForTerminalExitResponse, acp::Error> {
918 let exit_status = self
919 .session_thread(&args.session_id)?
920 .update(&mut self.cx.clone(), |thread, cx| {
921 anyhow::Ok(thread.terminal(args.terminal_id)?.read(cx).wait_for_exit())
922 })??
923 .await;
924
925 Ok(acp::WaitForTerminalExitResponse {
926 exit_status,
927 meta: None,
928 })
929 }
930}
931
932impl ClientDelegate {
933 fn session_thread(&self, session_id: &acp::SessionId) -> Result<WeakEntity<AcpThread>> {
934 let sessions = self.sessions.borrow();
935 sessions
936 .get(session_id)
937 .context("Failed to get session")
938 .map(|session| session.thread.clone())
939 }
940}