dap_store.rs

  1use super::{
  2    breakpoint_store::BreakpointStore,
  3    dap_command::EvaluateCommand,
  4    locators,
  5    session::{self, Session, SessionStateEvent},
  6};
  7use crate::{
  8    InlayHint, InlayHintLabel, ProjectEnvironment, ResolveState,
  9    project_settings::ProjectSettings,
 10    terminals::{SshCommand, wrap_for_ssh},
 11    worktree_store::WorktreeStore,
 12};
 13use anyhow::{Context as _, Result, anyhow};
 14use async_trait::async_trait;
 15use collections::HashMap;
 16use dap::{
 17    Capabilities, CompletionItem, CompletionsArguments, DapRegistry, DebugRequest,
 18    EvaluateArguments, EvaluateArgumentsContext, EvaluateResponse, Source, StackFrameId,
 19    adapters::{
 20        DapDelegate, DebugAdapterBinary, DebugAdapterName, DebugTaskDefinition, TcpArguments,
 21    },
 22    client::SessionId,
 23    inline_value::VariableLookupKind,
 24    messages::Message,
 25    requests::{Completions, Evaluate},
 26};
 27use fs::Fs;
 28use futures::{
 29    StreamExt,
 30    channel::mpsc::{self, UnboundedSender},
 31    future::{Shared, join_all},
 32};
 33use gpui::{App, AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Task};
 34use http_client::HttpClient;
 35use language::{Buffer, LanguageToolchainStore, language_settings::InlayHintKind};
 36use node_runtime::NodeRuntime;
 37
 38use remote::SshRemoteClient;
 39use rpc::{
 40    AnyProtoClient, TypedEnvelope,
 41    proto::{self},
 42};
 43use settings::{Settings, WorktreeId};
 44use std::{
 45    borrow::Borrow,
 46    collections::BTreeMap,
 47    ffi::OsStr,
 48    net::Ipv4Addr,
 49    path::{Path, PathBuf},
 50    sync::{Arc, Once},
 51};
 52use task::{DebugScenario, SpawnInTerminal, TaskTemplate};
 53use util::ResultExt as _;
 54use worktree::Worktree;
 55
 56#[derive(Debug)]
 57pub enum DapStoreEvent {
 58    DebugClientStarted(SessionId),
 59    DebugSessionInitialized(SessionId),
 60    DebugClientShutdown(SessionId),
 61    DebugClientEvent {
 62        session_id: SessionId,
 63        message: Message,
 64    },
 65    Notification(String),
 66    RemoteHasInitialized,
 67}
 68
 69enum DapStoreMode {
 70    Local(LocalDapStore),
 71    Ssh(SshDapStore),
 72    Collab,
 73}
 74
 75pub struct LocalDapStore {
 76    fs: Arc<dyn Fs>,
 77    node_runtime: NodeRuntime,
 78    http_client: Arc<dyn HttpClient>,
 79    environment: Entity<ProjectEnvironment>,
 80    toolchain_store: Arc<dyn LanguageToolchainStore>,
 81}
 82
 83pub struct SshDapStore {
 84    ssh_client: Entity<SshRemoteClient>,
 85    upstream_client: AnyProtoClient,
 86    upstream_project_id: u64,
 87}
 88
 89pub struct DapStore {
 90    mode: DapStoreMode,
 91    downstream_client: Option<(AnyProtoClient, u64)>,
 92    breakpoint_store: Entity<BreakpointStore>,
 93    worktree_store: Entity<WorktreeStore>,
 94    sessions: BTreeMap<SessionId, Entity<Session>>,
 95    next_session_id: u32,
 96}
 97
 98impl EventEmitter<DapStoreEvent> for DapStore {}
 99
100impl DapStore {
101    pub fn init(client: &AnyProtoClient, cx: &mut App) {
102        static ADD_LOCATORS: Once = Once::new();
103        ADD_LOCATORS.call_once(|| {
104            let registry = DapRegistry::global(cx);
105            registry.add_locator(Arc::new(locators::cargo::CargoLocator {}));
106            registry.add_locator(Arc::new(locators::go::GoLocator {}));
107            registry.add_locator(Arc::new(locators::node::NodeLocator));
108            registry.add_locator(Arc::new(locators::python::PythonLocator));
109        });
110        client.add_entity_request_handler(Self::handle_run_debug_locator);
111        client.add_entity_request_handler(Self::handle_get_debug_adapter_binary);
112        client.add_entity_message_handler(Self::handle_log_to_debug_console);
113    }
114
115    #[expect(clippy::too_many_arguments)]
116    pub fn new_local(
117        http_client: Arc<dyn HttpClient>,
118        node_runtime: NodeRuntime,
119        fs: Arc<dyn Fs>,
120        environment: Entity<ProjectEnvironment>,
121        toolchain_store: Arc<dyn LanguageToolchainStore>,
122        worktree_store: Entity<WorktreeStore>,
123        breakpoint_store: Entity<BreakpointStore>,
124        cx: &mut Context<Self>,
125    ) -> Self {
126        let mode = DapStoreMode::Local(LocalDapStore {
127            fs,
128            environment,
129            http_client,
130            node_runtime,
131            toolchain_store,
132        });
133
134        Self::new(mode, breakpoint_store, worktree_store, cx)
135    }
136
137    pub fn new_ssh(
138        project_id: u64,
139        ssh_client: Entity<SshRemoteClient>,
140        breakpoint_store: Entity<BreakpointStore>,
141        worktree_store: Entity<WorktreeStore>,
142        cx: &mut Context<Self>,
143    ) -> Self {
144        let mode = DapStoreMode::Ssh(SshDapStore {
145            upstream_client: ssh_client.read(cx).proto_client(),
146            ssh_client,
147            upstream_project_id: project_id,
148        });
149
150        Self::new(mode, breakpoint_store, worktree_store, cx)
151    }
152
153    pub fn new_collab(
154        _project_id: u64,
155        _upstream_client: AnyProtoClient,
156        breakpoint_store: Entity<BreakpointStore>,
157        worktree_store: Entity<WorktreeStore>,
158        cx: &mut Context<Self>,
159    ) -> Self {
160        Self::new(DapStoreMode::Collab, breakpoint_store, worktree_store, cx)
161    }
162
163    fn new(
164        mode: DapStoreMode,
165        breakpoint_store: Entity<BreakpointStore>,
166        worktree_store: Entity<WorktreeStore>,
167        _cx: &mut Context<Self>,
168    ) -> Self {
169        Self {
170            mode,
171            next_session_id: 0,
172            downstream_client: None,
173            breakpoint_store,
174            worktree_store,
175            sessions: Default::default(),
176        }
177    }
178
179    pub fn get_debug_adapter_binary(
180        &mut self,
181        definition: DebugTaskDefinition,
182        session_id: SessionId,
183        console: UnboundedSender<String>,
184        cx: &mut Context<Self>,
185    ) -> Task<Result<DebugAdapterBinary>> {
186        match &self.mode {
187            DapStoreMode::Local(_) => {
188                let Some(worktree) = self.worktree_store.read(cx).visible_worktrees(cx).next()
189                else {
190                    return Task::ready(Err(anyhow!("Failed to find a worktree")));
191                };
192                let Some(adapter) = DapRegistry::global(cx).adapter(&definition.adapter) else {
193                    return Task::ready(Err(anyhow!("Failed to find a debug adapter")));
194                };
195
196                let user_installed_path = ProjectSettings::get_global(cx)
197                    .dap
198                    .get(&adapter.name())
199                    .and_then(|s| s.binary.as_ref().map(PathBuf::from));
200
201                let delegate = self.delegate(&worktree, console, cx);
202                let cwd: Arc<Path> = worktree.read(cx).abs_path().as_ref().into();
203
204                cx.spawn(async move |this, cx| {
205                    let mut binary = adapter
206                        .get_binary(&delegate, &definition, user_installed_path, cx)
207                        .await?;
208
209                    let env = this
210                        .update(cx, |this, cx| {
211                            this.as_local()
212                                .unwrap()
213                                .environment
214                                .update(cx, |environment, cx| {
215                                    environment.get_directory_environment(cwd, cx)
216                                })
217                        })?
218                        .await;
219
220                    if let Some(mut env) = env {
221                        env.extend(std::mem::take(&mut binary.envs));
222                        binary.envs = env;
223                    }
224
225                    Ok(binary)
226                })
227            }
228            DapStoreMode::Ssh(ssh) => {
229                let request = ssh.upstream_client.request(proto::GetDebugAdapterBinary {
230                    session_id: session_id.to_proto(),
231                    project_id: ssh.upstream_project_id,
232                    definition: Some(definition.to_proto()),
233                });
234                let ssh_client = ssh.ssh_client.clone();
235
236                cx.spawn(async move |_, cx| {
237                    let response = request.await?;
238                    let binary = DebugAdapterBinary::from_proto(response)?;
239                    let mut ssh_command = ssh_client.read_with(cx, |ssh, _| {
240                        anyhow::Ok(SshCommand {
241                            arguments: ssh.ssh_args().context("SSH arguments not found")?,
242                        })
243                    })??;
244
245                    let mut connection = None;
246                    if let Some(c) = binary.connection {
247                        let local_bind_addr = Ipv4Addr::LOCALHOST;
248                        let port =
249                            dap::transport::TcpTransport::unused_port(local_bind_addr).await?;
250
251                        ssh_command.add_port_forwarding(port, c.host.to_string(), c.port);
252                        connection = Some(TcpArguments {
253                            port,
254                            host: local_bind_addr,
255                            timeout: c.timeout,
256                        })
257                    }
258
259                    let (program, args) = wrap_for_ssh(
260                        &ssh_command,
261                        Some((&binary.command, &binary.arguments)),
262                        binary.cwd.as_deref(),
263                        binary.envs,
264                        None,
265                    );
266
267                    Ok(DebugAdapterBinary {
268                        command: program,
269                        arguments: args,
270                        envs: HashMap::default(),
271                        cwd: None,
272                        connection,
273                        request_args: binary.request_args,
274                    })
275                })
276            }
277            DapStoreMode::Collab => {
278                Task::ready(Err(anyhow!("Debugging is not yet supported via collab")))
279            }
280        }
281    }
282
283    pub fn debug_scenario_for_build_task(
284        &self,
285        build: TaskTemplate,
286        adapter: DebugAdapterName,
287        label: SharedString,
288        cx: &mut App,
289    ) -> Option<DebugScenario> {
290        DapRegistry::global(cx)
291            .locators()
292            .values()
293            .find_map(|locator| locator.create_scenario(&build, &label, adapter.clone()))
294    }
295
296    pub fn run_debug_locator(
297        &mut self,
298        locator_name: &str,
299        build_command: SpawnInTerminal,
300        cx: &mut Context<Self>,
301    ) -> Task<Result<DebugRequest>> {
302        match &self.mode {
303            DapStoreMode::Local(_) => {
304                // Pre-resolve args with existing environment.
305                let locators = DapRegistry::global(cx).locators();
306                let locator = locators.get(locator_name);
307
308                if let Some(locator) = locator.cloned() {
309                    cx.background_spawn(async move {
310                        let result = locator
311                            .run(build_command.clone())
312                            .await
313                            .log_with_level(log::Level::Error);
314                        if let Some(result) = result {
315                            return Ok(result);
316                        }
317
318                        anyhow::bail!(
319                            "None of the locators for task `{}` completed successfully",
320                            build_command.label
321                        )
322                    })
323                } else {
324                    Task::ready(Err(anyhow!(
325                        "Couldn't find any locator for task `{}`. Specify the `attach` or `launch` arguments in your debug scenario definition",
326                        build_command.label
327                    )))
328                }
329            }
330            DapStoreMode::Ssh(ssh) => {
331                let request = ssh.upstream_client.request(proto::RunDebugLocators {
332                    project_id: ssh.upstream_project_id,
333                    build_command: Some(build_command.to_proto()),
334                    locator: locator_name.to_owned(),
335                });
336                cx.background_spawn(async move {
337                    let response = request.await?;
338                    DebugRequest::from_proto(response)
339                })
340            }
341            DapStoreMode::Collab => {
342                Task::ready(Err(anyhow!("Debugging is not yet supported via collab")))
343            }
344        }
345    }
346
347    fn as_local(&self) -> Option<&LocalDapStore> {
348        match &self.mode {
349            DapStoreMode::Local(local_dap_store) => Some(local_dap_store),
350            _ => None,
351        }
352    }
353
354    pub fn new_session(
355        &mut self,
356        label: SharedString,
357        adapter: DebugAdapterName,
358        parent_session: Option<Entity<Session>>,
359        cx: &mut Context<Self>,
360    ) -> Entity<Session> {
361        let session_id = SessionId(util::post_inc(&mut self.next_session_id));
362
363        if let Some(session) = &parent_session {
364            session.update(cx, |session, _| {
365                session.add_child_session_id(session_id);
366            });
367        }
368
369        let session = Session::new(
370            self.breakpoint_store.clone(),
371            session_id,
372            parent_session,
373            label,
374            adapter,
375            cx,
376        );
377
378        self.sessions.insert(session_id, session.clone());
379        cx.notify();
380
381        cx.subscribe(&session, {
382            move |this: &mut DapStore, _, event: &SessionStateEvent, cx| match event {
383                SessionStateEvent::Shutdown => {
384                    this.shutdown_session(session_id, cx).detach_and_log_err(cx);
385                }
386                SessionStateEvent::Restart | SessionStateEvent::SpawnChildSession { .. } => {}
387                SessionStateEvent::Running => {
388                    cx.emit(DapStoreEvent::DebugClientStarted(session_id));
389                }
390            }
391        })
392        .detach();
393
394        session
395    }
396
397    pub fn boot_session(
398        &self,
399        session: Entity<Session>,
400        definition: DebugTaskDefinition,
401        cx: &mut Context<Self>,
402    ) -> Task<Result<()>> {
403        let Some(worktree) = self.worktree_store.read(cx).visible_worktrees(cx).next() else {
404            return Task::ready(Err(anyhow!("Failed to find a worktree")));
405        };
406
407        let dap_store = cx.weak_entity();
408        let console = session.update(cx, |session, cx| session.console_output(cx));
409        let session_id = session.read(cx).session_id();
410
411        cx.spawn({
412            let session = session.clone();
413            async move |this, cx| {
414                let binary = this
415                    .update(cx, |this, cx| {
416                        this.get_debug_adapter_binary(definition.clone(), session_id, console, cx)
417                    })?
418                    .await?;
419                session
420                    .update(cx, |session, cx| {
421                        session.boot(binary, worktree, dap_store, cx)
422                    })?
423                    .await
424            }
425        })
426    }
427
428    pub fn session_by_id(
429        &self,
430        session_id: impl Borrow<SessionId>,
431    ) -> Option<Entity<session::Session>> {
432        let session_id = session_id.borrow();
433        let client = self.sessions.get(session_id).cloned();
434
435        client
436    }
437    pub fn sessions(&self) -> impl Iterator<Item = &Entity<Session>> {
438        self.sessions.values()
439    }
440
441    pub fn capabilities_by_id(
442        &self,
443        session_id: impl Borrow<SessionId>,
444        cx: &App,
445    ) -> Option<Capabilities> {
446        let session_id = session_id.borrow();
447        self.sessions
448            .get(session_id)
449            .map(|client| client.read(cx).capabilities.clone())
450    }
451
452    pub fn breakpoint_store(&self) -> &Entity<BreakpointStore> {
453        &self.breakpoint_store
454    }
455
456    pub fn worktree_store(&self) -> &Entity<WorktreeStore> {
457        &self.worktree_store
458    }
459
460    #[allow(dead_code)]
461    async fn handle_ignore_breakpoint_state(
462        this: Entity<Self>,
463        envelope: TypedEnvelope<proto::IgnoreBreakpointState>,
464        mut cx: AsyncApp,
465    ) -> Result<()> {
466        let session_id = SessionId::from_proto(envelope.payload.session_id);
467
468        this.update(&mut cx, |this, cx| {
469            if let Some(session) = this.session_by_id(&session_id) {
470                session.update(cx, |session, cx| {
471                    session.set_ignore_breakpoints(envelope.payload.ignore, cx)
472                })
473            } else {
474                Task::ready(HashMap::default())
475            }
476        })?
477        .await;
478
479        Ok(())
480    }
481
482    fn delegate(
483        &self,
484        worktree: &Entity<Worktree>,
485        console: UnboundedSender<String>,
486        cx: &mut App,
487    ) -> Arc<dyn DapDelegate> {
488        let Some(local_store) = self.as_local() else {
489            unimplemented!("Starting session on remote side");
490        };
491
492        Arc::new(DapAdapterDelegate::new(
493            local_store.fs.clone(),
494            worktree.read(cx).snapshot(),
495            console,
496            local_store.node_runtime.clone(),
497            local_store.http_client.clone(),
498            local_store.toolchain_store.clone(),
499            local_store.environment.update(cx, |env, cx| {
500                env.get_worktree_environment(worktree.clone(), cx)
501            }),
502        ))
503    }
504
505    pub fn evaluate(
506        &self,
507        session_id: &SessionId,
508        stack_frame_id: u64,
509        expression: String,
510        context: EvaluateArgumentsContext,
511        source: Option<Source>,
512        cx: &mut Context<Self>,
513    ) -> Task<Result<EvaluateResponse>> {
514        let Some(client) = self
515            .session_by_id(session_id)
516            .and_then(|client| client.read(cx).adapter_client())
517        else {
518            return Task::ready(Err(anyhow!("Could not find client: {:?}", session_id)));
519        };
520
521        cx.background_executor().spawn(async move {
522            client
523                .request::<Evaluate>(EvaluateArguments {
524                    expression: expression.clone(),
525                    frame_id: Some(stack_frame_id),
526                    context: Some(context),
527                    format: None,
528                    line: None,
529                    column: None,
530                    source,
531                })
532                .await
533        })
534    }
535
536    pub fn completions(
537        &self,
538        session_id: &SessionId,
539        stack_frame_id: u64,
540        text: String,
541        completion_column: u64,
542        cx: &mut Context<Self>,
543    ) -> Task<Result<Vec<CompletionItem>>> {
544        let Some(client) = self
545            .session_by_id(session_id)
546            .and_then(|client| client.read(cx).adapter_client())
547        else {
548            return Task::ready(Err(anyhow!("Could not find client: {:?}", session_id)));
549        };
550
551        cx.background_executor().spawn(async move {
552            Ok(client
553                .request::<Completions>(CompletionsArguments {
554                    frame_id: Some(stack_frame_id),
555                    line: None,
556                    text,
557                    column: completion_column,
558                })
559                .await?
560                .targets)
561        })
562    }
563
564    pub fn resolve_inline_value_locations(
565        &self,
566        session: Entity<Session>,
567        stack_frame_id: StackFrameId,
568        buffer_handle: Entity<Buffer>,
569        inline_value_locations: Vec<dap::inline_value::InlineValueLocation>,
570        cx: &mut Context<Self>,
571    ) -> Task<Result<Vec<InlayHint>>> {
572        let snapshot = buffer_handle.read(cx).snapshot();
573        let all_variables = session.read(cx).variables_by_stack_frame_id(stack_frame_id);
574
575        fn format_value(mut value: String) -> String {
576            const LIMIT: usize = 100;
577
578            if value.len() > LIMIT {
579                let mut index = LIMIT;
580                // If index isn't a char boundary truncate will cause a panic
581                while !value.is_char_boundary(index) {
582                    index -= 1;
583                }
584                value.truncate(index);
585                value.push_str("...");
586            }
587
588            format!(": {}", value)
589        }
590
591        cx.spawn(async move |_, cx| {
592            let mut inlay_hints = Vec::with_capacity(inline_value_locations.len());
593            for inline_value_location in inline_value_locations.iter() {
594                let point = snapshot.point_to_point_utf16(language::Point::new(
595                    inline_value_location.row as u32,
596                    inline_value_location.column as u32,
597                ));
598                let position = snapshot.anchor_after(point);
599
600                match inline_value_location.lookup {
601                    VariableLookupKind::Variable => {
602                        let Some(variable) = all_variables
603                            .iter()
604                            .find(|variable| variable.name == inline_value_location.variable_name)
605                        else {
606                            continue;
607                        };
608
609                        inlay_hints.push(InlayHint {
610                            position,
611                            label: InlayHintLabel::String(format_value(variable.value.clone())),
612                            kind: Some(InlayHintKind::Type),
613                            padding_left: false,
614                            padding_right: false,
615                            tooltip: None,
616                            resolve_state: ResolveState::Resolved,
617                        });
618                    }
619                    VariableLookupKind::Expression => {
620                        let Ok(eval_task) = session.read_with(cx, |session, _| {
621                            session.mode.request_dap(EvaluateCommand {
622                                expression: inline_value_location.variable_name.clone(),
623                                frame_id: Some(stack_frame_id),
624                                source: None,
625                                context: Some(EvaluateArgumentsContext::Variables),
626                            })
627                        }) else {
628                            continue;
629                        };
630
631                        if let Some(response) = eval_task.await.log_err() {
632                            inlay_hints.push(InlayHint {
633                                position,
634                                label: InlayHintLabel::String(format_value(response.result)),
635                                kind: Some(InlayHintKind::Type),
636                                padding_left: false,
637                                padding_right: false,
638                                tooltip: None,
639                                resolve_state: ResolveState::Resolved,
640                            });
641                        };
642                    }
643                };
644            }
645
646            Ok(inlay_hints)
647        })
648    }
649
650    pub fn shutdown_sessions(&mut self, cx: &mut Context<Self>) -> Task<()> {
651        let mut tasks = vec![];
652        for session_id in self.sessions.keys().cloned().collect::<Vec<_>>() {
653            tasks.push(self.shutdown_session(session_id, cx));
654        }
655
656        cx.background_executor().spawn(async move {
657            futures::future::join_all(tasks).await;
658        })
659    }
660
661    pub fn shutdown_session(
662        &mut self,
663        session_id: SessionId,
664        cx: &mut Context<Self>,
665    ) -> Task<Result<()>> {
666        let Some(session) = self.sessions.remove(&session_id) else {
667            return Task::ready(Err(anyhow!("Could not find session: {:?}", session_id)));
668        };
669
670        let shutdown_children = session
671            .read(cx)
672            .child_session_ids()
673            .iter()
674            .map(|session_id| self.shutdown_session(*session_id, cx))
675            .collect::<Vec<_>>();
676
677        let shutdown_parent_task = if let Some(parent_session) = session
678            .read(cx)
679            .parent_id(cx)
680            .and_then(|session_id| self.session_by_id(session_id))
681        {
682            let shutdown_id = parent_session.update(cx, |parent_session, _| {
683                parent_session.remove_child_session_id(session_id);
684
685                if parent_session.child_session_ids().len() == 0 {
686                    Some(parent_session.session_id())
687                } else {
688                    None
689                }
690            });
691
692            shutdown_id.map(|session_id| self.shutdown_session(session_id, cx))
693        } else {
694            None
695        };
696
697        let shutdown_task = session.update(cx, |this, cx| this.shutdown(cx));
698
699        cx.background_spawn(async move {
700            if shutdown_children.len() > 0 {
701                let _ = join_all(shutdown_children).await;
702            }
703
704            shutdown_task.await;
705
706            if let Some(parent_task) = shutdown_parent_task {
707                parent_task.await?;
708            }
709
710            Ok(())
711        })
712    }
713
714    pub fn shared(
715        &mut self,
716        project_id: u64,
717        downstream_client: AnyProtoClient,
718        _: &mut Context<Self>,
719    ) {
720        self.downstream_client = Some((downstream_client.clone(), project_id));
721    }
722
723    pub fn unshared(&mut self, cx: &mut Context<Self>) {
724        self.downstream_client.take();
725
726        cx.notify();
727    }
728
729    async fn handle_run_debug_locator(
730        this: Entity<Self>,
731        envelope: TypedEnvelope<proto::RunDebugLocators>,
732        mut cx: AsyncApp,
733    ) -> Result<proto::DebugRequest> {
734        let task = envelope
735            .payload
736            .build_command
737            .context("missing definition")?;
738        let build_task = SpawnInTerminal::from_proto(task);
739        let locator = envelope.payload.locator;
740        let request = this
741            .update(&mut cx, |this, cx| {
742                this.run_debug_locator(&locator, build_task, cx)
743            })?
744            .await?;
745
746        Ok(request.to_proto())
747    }
748
749    async fn handle_get_debug_adapter_binary(
750        this: Entity<Self>,
751        envelope: TypedEnvelope<proto::GetDebugAdapterBinary>,
752        mut cx: AsyncApp,
753    ) -> Result<proto::DebugAdapterBinary> {
754        let definition = DebugTaskDefinition::from_proto(
755            envelope.payload.definition.context("missing definition")?,
756        )?;
757        let (tx, mut rx) = mpsc::unbounded();
758        let session_id = envelope.payload.session_id;
759        cx.spawn({
760            let this = this.clone();
761            async move |cx| {
762                while let Some(message) = rx.next().await {
763                    this.read_with(cx, |this, _| {
764                        if let Some((downstream, project_id)) = this.downstream_client.clone() {
765                            downstream
766                                .send(proto::LogToDebugConsole {
767                                    project_id,
768                                    session_id,
769                                    message,
770                                })
771                                .ok();
772                        }
773                    })
774                    .ok();
775                }
776            }
777        })
778        .detach();
779
780        let binary = this
781            .update(&mut cx, |this, cx| {
782                this.get_debug_adapter_binary(definition, SessionId::from_proto(session_id), tx, cx)
783            })?
784            .await?;
785        Ok(binary.to_proto())
786    }
787
788    async fn handle_log_to_debug_console(
789        this: Entity<Self>,
790        envelope: TypedEnvelope<proto::LogToDebugConsole>,
791        mut cx: AsyncApp,
792    ) -> Result<()> {
793        let session_id = SessionId::from_proto(envelope.payload.session_id);
794        this.update(&mut cx, |this, cx| {
795            let Some(session) = this.sessions.get(&session_id) else {
796                return;
797            };
798            session.update(cx, |session, cx| {
799                session
800                    .console_output(cx)
801                    .unbounded_send(envelope.payload.message)
802                    .ok();
803            })
804        })
805    }
806}
807
808#[derive(Clone)]
809pub struct DapAdapterDelegate {
810    fs: Arc<dyn Fs>,
811    console: mpsc::UnboundedSender<String>,
812    worktree: worktree::Snapshot,
813    node_runtime: NodeRuntime,
814    http_client: Arc<dyn HttpClient>,
815    toolchain_store: Arc<dyn LanguageToolchainStore>,
816    load_shell_env_task: Shared<Task<Option<HashMap<String, String>>>>,
817}
818
819impl DapAdapterDelegate {
820    pub fn new(
821        fs: Arc<dyn Fs>,
822        worktree: worktree::Snapshot,
823        status: mpsc::UnboundedSender<String>,
824        node_runtime: NodeRuntime,
825        http_client: Arc<dyn HttpClient>,
826        toolchain_store: Arc<dyn LanguageToolchainStore>,
827        load_shell_env_task: Shared<Task<Option<HashMap<String, String>>>>,
828    ) -> Self {
829        Self {
830            fs,
831            console: status,
832            worktree,
833            http_client,
834            node_runtime,
835            toolchain_store,
836            load_shell_env_task,
837        }
838    }
839}
840
841#[async_trait]
842impl dap::adapters::DapDelegate for DapAdapterDelegate {
843    fn worktree_id(&self) -> WorktreeId {
844        self.worktree.id()
845    }
846
847    fn worktree_root_path(&self) -> &Path {
848        &self.worktree.abs_path()
849    }
850    fn http_client(&self) -> Arc<dyn HttpClient> {
851        self.http_client.clone()
852    }
853
854    fn node_runtime(&self) -> NodeRuntime {
855        self.node_runtime.clone()
856    }
857
858    fn fs(&self) -> Arc<dyn Fs> {
859        self.fs.clone()
860    }
861
862    fn output_to_console(&self, msg: String) {
863        self.console.unbounded_send(msg).ok();
864    }
865
866    async fn which(&self, command: &OsStr) -> Option<PathBuf> {
867        which::which(command).ok()
868    }
869
870    async fn shell_env(&self) -> HashMap<String, String> {
871        let task = self.load_shell_env_task.clone();
872        task.await.unwrap_or_default()
873    }
874
875    fn toolchain_store(&self) -> Arc<dyn LanguageToolchainStore> {
876        self.toolchain_store.clone()
877    }
878    async fn read_text_file(&self, path: PathBuf) -> Result<String> {
879        let entry = self
880            .worktree
881            .entry_for_path(&path)
882            .with_context(|| format!("no worktree entry for path {path:?}"))?;
883        let abs_path = self
884            .worktree
885            .absolutize(&entry.path)
886            .with_context(|| format!("cannot absolutize path {path:?}"))?;
887
888        self.fs.load(&abs_path).await
889    }
890}