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