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