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