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