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