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