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