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