1use crate::debugger::breakpoint_store::BreakpointSessionState;
   2
   3use super::breakpoint_store::{
   4    BreakpointStore, BreakpointStoreEvent, BreakpointUpdatedReason, SourceBreakpoint,
   5};
   6use super::dap_command::{
   7    self, Attach, ConfigurationDone, ContinueCommand, DapCommand, DisconnectCommand,
   8    EvaluateCommand, Initialize, Launch, LoadedSourcesCommand, LocalDapCommand, LocationsCommand,
   9    ModulesCommand, NextCommand, PauseCommand, RestartCommand, RestartStackFrameCommand,
  10    ScopesCommand, SetExceptionBreakpoints, SetVariableValueCommand, StackTraceCommand,
  11    StepBackCommand, StepCommand, StepInCommand, StepOutCommand, TerminateCommand,
  12    TerminateThreadsCommand, ThreadsCommand, VariablesCommand,
  13};
  14use super::dap_store::DapStore;
  15use anyhow::{Context as _, Result, anyhow};
  16use collections::{HashMap, HashSet, IndexMap};
  17use dap::adapters::{DebugAdapterBinary, DebugAdapterName};
  18use dap::messages::Response;
  19use dap::requests::{Request, RunInTerminal, StartDebugging};
  20use dap::{
  21    Capabilities, ContinueArguments, EvaluateArgumentsContext, Module, Source, StackFrameId,
  22    SteppingGranularity, StoppedEvent, VariableReference,
  23    client::{DebugAdapterClient, SessionId},
  24    messages::{Events, Message},
  25};
  26use dap::{
  27    ExceptionBreakpointsFilter, ExceptionFilterOptions, OutputEvent, OutputEventCategory,
  28    RunInTerminalRequestArguments, StackFramePresentationHint, StartDebuggingRequestArguments,
  29    StartDebuggingRequestArgumentsRequest, VariablePresentationHint,
  30};
  31use futures::SinkExt;
  32use futures::channel::mpsc::UnboundedSender;
  33use futures::channel::{mpsc, oneshot};
  34use futures::{FutureExt, future::Shared};
  35use gpui::{
  36    App, AppContext, AsyncApp, BackgroundExecutor, Context, Entity, EventEmitter, SharedString,
  37    Task, WeakEntity,
  38};
  39
  40use rpc::ErrorExt;
  41use serde_json::Value;
  42use smol::stream::StreamExt;
  43use std::any::TypeId;
  44use std::collections::BTreeMap;
  45use std::u64;
  46use std::{
  47    any::Any,
  48    collections::hash_map::Entry,
  49    hash::{Hash, Hasher},
  50    path::Path,
  51    sync::Arc,
  52};
  53use task::TaskContext;
  54use text::{PointUtf16, ToPointUtf16};
  55use util::ResultExt;
  56use worktree::Worktree;
  57
  58#[derive(Debug, Copy, Clone, Hash, PartialEq, PartialOrd, Ord, Eq)]
  59#[repr(transparent)]
  60pub struct ThreadId(pub u64);
  61
  62impl ThreadId {
  63    pub const MIN: ThreadId = ThreadId(u64::MIN);
  64    pub const MAX: ThreadId = ThreadId(u64::MAX);
  65}
  66
  67impl From<u64> for ThreadId {
  68    fn from(id: u64) -> Self {
  69        Self(id)
  70    }
  71}
  72
  73#[derive(Clone, Debug)]
  74pub struct StackFrame {
  75    pub dap: dap::StackFrame,
  76    pub scopes: Vec<dap::Scope>,
  77}
  78
  79impl From<dap::StackFrame> for StackFrame {
  80    fn from(stack_frame: dap::StackFrame) -> Self {
  81        Self {
  82            scopes: vec![],
  83            dap: stack_frame,
  84        }
  85    }
  86}
  87
  88#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
  89pub enum ThreadStatus {
  90    #[default]
  91    Running,
  92    Stopped,
  93    Stepping,
  94    Exited,
  95    Ended,
  96}
  97
  98impl ThreadStatus {
  99    pub fn label(&self) -> &'static str {
 100        match self {
 101            ThreadStatus::Running => "Running",
 102            ThreadStatus::Stopped => "Stopped",
 103            ThreadStatus::Stepping => "Stepping",
 104            ThreadStatus::Exited => "Exited",
 105            ThreadStatus::Ended => "Ended",
 106        }
 107    }
 108}
 109
 110#[derive(Debug)]
 111pub struct Thread {
 112    dap: dap::Thread,
 113    stack_frames: Vec<StackFrame>,
 114    stack_frames_error: Option<anyhow::Error>,
 115    _has_stopped: bool,
 116}
 117
 118impl From<dap::Thread> for Thread {
 119    fn from(dap: dap::Thread) -> Self {
 120        Self {
 121            dap,
 122            stack_frames: Default::default(),
 123            stack_frames_error: None,
 124            _has_stopped: false,
 125        }
 126    }
 127}
 128
 129#[derive(Debug, Clone, PartialEq)]
 130pub struct Watcher {
 131    pub expression: SharedString,
 132    pub value: SharedString,
 133    pub variables_reference: u64,
 134    pub presentation_hint: Option<VariablePresentationHint>,
 135}
 136
 137pub enum Mode {
 138    Building,
 139    Running(RunningMode),
 140}
 141
 142#[derive(Clone)]
 143pub struct RunningMode {
 144    client: Arc<DebugAdapterClient>,
 145    binary: DebugAdapterBinary,
 146    tmp_breakpoint: Option<SourceBreakpoint>,
 147    worktree: WeakEntity<Worktree>,
 148    executor: BackgroundExecutor,
 149    is_started: bool,
 150    has_ever_stopped: bool,
 151    messages_tx: UnboundedSender<Message>,
 152}
 153
 154fn client_source(abs_path: &Path) -> dap::Source {
 155    dap::Source {
 156        name: abs_path
 157            .file_name()
 158            .map(|filename| filename.to_string_lossy().to_string()),
 159        path: Some(abs_path.to_string_lossy().to_string()),
 160        source_reference: None,
 161        presentation_hint: None,
 162        origin: None,
 163        sources: None,
 164        adapter_data: None,
 165        checksums: None,
 166    }
 167}
 168
 169impl RunningMode {
 170    async fn new(
 171        session_id: SessionId,
 172        parent_session: Option<Entity<Session>>,
 173        worktree: WeakEntity<Worktree>,
 174        binary: DebugAdapterBinary,
 175        messages_tx: futures::channel::mpsc::UnboundedSender<Message>,
 176        cx: &mut AsyncApp,
 177    ) -> Result<Self> {
 178        let message_handler = Box::new({
 179            let messages_tx = messages_tx.clone();
 180            move |message| {
 181                messages_tx.unbounded_send(message).ok();
 182            }
 183        });
 184
 185        let client = if let Some(client) = parent_session
 186            .and_then(|session| cx.update(|cx| session.read(cx).adapter_client()).ok())
 187            .flatten()
 188        {
 189            client
 190                .create_child_connection(session_id, binary.clone(), message_handler, cx)
 191                .await?
 192        } else {
 193            DebugAdapterClient::start(session_id, binary.clone(), message_handler, cx).await?
 194        };
 195
 196        Ok(Self {
 197            client: Arc::new(client),
 198            worktree,
 199            tmp_breakpoint: None,
 200            binary,
 201            executor: cx.background_executor().clone(),
 202            is_started: false,
 203            has_ever_stopped: false,
 204            messages_tx,
 205        })
 206    }
 207
 208    pub(crate) fn worktree(&self) -> &WeakEntity<Worktree> {
 209        &self.worktree
 210    }
 211
 212    fn unset_breakpoints_from_paths(&self, paths: &Vec<Arc<Path>>, cx: &mut App) -> Task<()> {
 213        let tasks: Vec<_> = paths
 214            .into_iter()
 215            .map(|path| {
 216                self.request(dap_command::SetBreakpoints {
 217                    source: client_source(path),
 218                    source_modified: None,
 219                    breakpoints: vec![],
 220                })
 221            })
 222            .collect();
 223
 224        cx.background_spawn(async move {
 225            futures::future::join_all(tasks)
 226                .await
 227                .iter()
 228                .for_each(|res| match res {
 229                    Ok(_) => {}
 230                    Err(err) => {
 231                        log::warn!("Set breakpoints request failed: {}", err);
 232                    }
 233                });
 234        })
 235    }
 236
 237    fn send_breakpoints_from_path(
 238        &self,
 239        abs_path: Arc<Path>,
 240        reason: BreakpointUpdatedReason,
 241        breakpoint_store: &Entity<BreakpointStore>,
 242        cx: &mut App,
 243    ) -> Task<()> {
 244        let breakpoints =
 245            breakpoint_store
 246                .read(cx)
 247                .source_breakpoints_from_path(&abs_path, cx)
 248                .into_iter()
 249                .filter(|bp| bp.state.is_enabled())
 250                .chain(self.tmp_breakpoint.iter().filter_map(|breakpoint| {
 251                    breakpoint.path.eq(&abs_path).then(|| breakpoint.clone())
 252                }))
 253                .map(Into::into)
 254                .collect();
 255
 256        let raw_breakpoints = breakpoint_store
 257            .read(cx)
 258            .breakpoints_from_path(&abs_path)
 259            .into_iter()
 260            .filter(|bp| bp.bp.state.is_enabled())
 261            .collect::<Vec<_>>();
 262
 263        let task = self.request(dap_command::SetBreakpoints {
 264            source: client_source(&abs_path),
 265            source_modified: Some(matches!(reason, BreakpointUpdatedReason::FileSaved)),
 266            breakpoints,
 267        });
 268        let session_id = self.client.id();
 269        let breakpoint_store = breakpoint_store.downgrade();
 270        cx.spawn(async move |cx| match cx.background_spawn(task).await {
 271            Ok(breakpoints) => {
 272                let breakpoints =
 273                    breakpoints
 274                        .into_iter()
 275                        .zip(raw_breakpoints)
 276                        .filter_map(|(dap_bp, zed_bp)| {
 277                            Some((
 278                                zed_bp,
 279                                BreakpointSessionState {
 280                                    id: dap_bp.id?,
 281                                    verified: dap_bp.verified,
 282                                },
 283                            ))
 284                        });
 285                breakpoint_store
 286                    .update(cx, |this, _| {
 287                        this.mark_breakpoints_verified(session_id, &abs_path, breakpoints);
 288                    })
 289                    .ok();
 290            }
 291            Err(err) => log::warn!("Set breakpoints request failed for path: {}", err),
 292        })
 293    }
 294
 295    fn send_exception_breakpoints(
 296        &self,
 297        filters: Vec<ExceptionBreakpointsFilter>,
 298        supports_filter_options: bool,
 299    ) -> Task<Result<Vec<dap::Breakpoint>>> {
 300        let arg = if supports_filter_options {
 301            SetExceptionBreakpoints::WithOptions {
 302                filters: filters
 303                    .into_iter()
 304                    .map(|filter| ExceptionFilterOptions {
 305                        filter_id: filter.filter,
 306                        condition: None,
 307                        mode: None,
 308                    })
 309                    .collect(),
 310            }
 311        } else {
 312            SetExceptionBreakpoints::Plain {
 313                filters: filters.into_iter().map(|filter| filter.filter).collect(),
 314            }
 315        };
 316        self.request(arg)
 317    }
 318
 319    fn send_source_breakpoints(
 320        &self,
 321        ignore_breakpoints: bool,
 322        breakpoint_store: &Entity<BreakpointStore>,
 323        cx: &App,
 324    ) -> Task<HashMap<Arc<Path>, anyhow::Error>> {
 325        let mut breakpoint_tasks = Vec::new();
 326        let breakpoints = breakpoint_store.read(cx).all_source_breakpoints(cx);
 327        let mut raw_breakpoints = breakpoint_store.read_with(cx, |this, _| this.all_breakpoints());
 328        debug_assert_eq!(raw_breakpoints.len(), breakpoints.len());
 329        let session_id = self.client.id();
 330        for (path, breakpoints) in breakpoints {
 331            let breakpoints = if ignore_breakpoints {
 332                vec![]
 333            } else {
 334                breakpoints
 335                    .into_iter()
 336                    .filter(|bp| bp.state.is_enabled())
 337                    .map(Into::into)
 338                    .collect()
 339            };
 340
 341            let raw_breakpoints = raw_breakpoints
 342                .remove(&path)
 343                .unwrap_or_default()
 344                .into_iter()
 345                .filter(|bp| bp.bp.state.is_enabled());
 346            let error_path = path.clone();
 347            let send_request = self
 348                .request(dap_command::SetBreakpoints {
 349                    source: client_source(&path),
 350                    source_modified: Some(false),
 351                    breakpoints,
 352                })
 353                .map(|result| result.map_err(move |e| (error_path, e)));
 354
 355            let task = cx.spawn({
 356                let breakpoint_store = breakpoint_store.downgrade();
 357                async move |cx| {
 358                    let breakpoints = cx.background_spawn(send_request).await?;
 359
 360                    let breakpoints = breakpoints.into_iter().zip(raw_breakpoints).filter_map(
 361                        |(dap_bp, zed_bp)| {
 362                            Some((
 363                                zed_bp,
 364                                BreakpointSessionState {
 365                                    id: dap_bp.id?,
 366                                    verified: dap_bp.verified,
 367                                },
 368                            ))
 369                        },
 370                    );
 371                    breakpoint_store
 372                        .update(cx, |this, _| {
 373                            this.mark_breakpoints_verified(session_id, &path, breakpoints);
 374                        })
 375                        .ok();
 376
 377                    Ok(())
 378                }
 379            });
 380            breakpoint_tasks.push(task);
 381        }
 382
 383        cx.background_spawn(async move {
 384            futures::future::join_all(breakpoint_tasks)
 385                .await
 386                .into_iter()
 387                .filter_map(Result::err)
 388                .collect::<HashMap<_, _>>()
 389        })
 390    }
 391
 392    fn initialize_sequence(
 393        &self,
 394        capabilities: &Capabilities,
 395        initialized_rx: oneshot::Receiver<()>,
 396        dap_store: WeakEntity<DapStore>,
 397        cx: &mut Context<Session>,
 398    ) -> Task<Result<()>> {
 399        let raw = self.binary.request_args.clone();
 400
 401        // Of relevance: https://github.com/microsoft/vscode/issues/4902#issuecomment-368583522
 402        let launch = match raw.request {
 403            dap::StartDebuggingRequestArgumentsRequest::Launch => self.request(Launch {
 404                raw: raw.configuration,
 405            }),
 406            dap::StartDebuggingRequestArgumentsRequest::Attach => self.request(Attach {
 407                raw: raw.configuration,
 408            }),
 409        };
 410
 411        let configuration_done_supported = ConfigurationDone::is_supported(capabilities);
 412        let exception_filters = capabilities
 413            .exception_breakpoint_filters
 414            .as_ref()
 415            .map(|exception_filters| {
 416                exception_filters
 417                    .iter()
 418                    .filter(|filter| filter.default == Some(true))
 419                    .cloned()
 420                    .collect::<Vec<_>>()
 421            })
 422            .unwrap_or_default();
 423        let supports_exception_filters = capabilities
 424            .supports_exception_filter_options
 425            .unwrap_or_default();
 426        let this = self.clone();
 427        let worktree = self.worktree().clone();
 428        let configuration_sequence = cx.spawn({
 429            async move |_, cx| {
 430                let breakpoint_store =
 431                    dap_store.read_with(cx, |dap_store, _| dap_store.breakpoint_store().clone())?;
 432                initialized_rx.await?;
 433                let errors_by_path = cx
 434                    .update(|cx| this.send_source_breakpoints(false, &breakpoint_store, cx))?
 435                    .await;
 436
 437                dap_store.update(cx, |_, cx| {
 438                    let Some(worktree) = worktree.upgrade() else {
 439                        return;
 440                    };
 441
 442                    for (path, error) in &errors_by_path {
 443                        log::error!("failed to set breakpoints for {path:?}: {error}");
 444                    }
 445
 446                    if let Some(failed_path) = errors_by_path.keys().next() {
 447                        let failed_path = failed_path
 448                            .strip_prefix(worktree.read(cx).abs_path())
 449                            .unwrap_or(failed_path)
 450                            .display();
 451                        let message = format!(
 452                            "Failed to set breakpoints for {failed_path}{}",
 453                            match errors_by_path.len() {
 454                                0 => unreachable!(),
 455                                1 => "".into(),
 456                                2 => " and 1 other path".into(),
 457                                n => format!(" and {} other paths", n - 1),
 458                            }
 459                        );
 460                        cx.emit(super::dap_store::DapStoreEvent::Notification(message));
 461                    }
 462                })?;
 463
 464                this.send_exception_breakpoints(exception_filters, supports_exception_filters)
 465                    .await
 466                    .ok();
 467                let ret = if configuration_done_supported {
 468                    this.request(ConfigurationDone {})
 469                } else {
 470                    Task::ready(Ok(()))
 471                }
 472                .await;
 473                ret
 474            }
 475        });
 476
 477        let task = cx.background_spawn(futures::future::try_join(launch, configuration_sequence));
 478
 479        cx.spawn(async move |this, cx| {
 480            let result = task.await;
 481
 482            this.update(cx, |this, cx| {
 483                if let Some(this) = this.as_running_mut() {
 484                    this.is_started = true;
 485                    cx.notify();
 486                }
 487            })
 488            .ok();
 489
 490            result?;
 491            anyhow::Ok(())
 492        })
 493    }
 494
 495    fn reconnect_for_ssh(&self, cx: &mut AsyncApp) -> Option<Task<Result<()>>> {
 496        let client = self.client.clone();
 497        let messages_tx = self.messages_tx.clone();
 498        let message_handler = Box::new(move |message| {
 499            messages_tx.unbounded_send(message).ok();
 500        });
 501        if client.should_reconnect_for_ssh() {
 502            Some(cx.spawn(async move |cx| {
 503                client.connect(message_handler, cx).await?;
 504                anyhow::Ok(())
 505            }))
 506        } else {
 507            None
 508        }
 509    }
 510
 511    fn request<R: LocalDapCommand>(&self, request: R) -> Task<Result<R::Response>>
 512    where
 513        <R::DapRequest as dap::requests::Request>::Response: 'static,
 514        <R::DapRequest as dap::requests::Request>::Arguments: 'static + Send,
 515    {
 516        let request = Arc::new(request);
 517
 518        let request_clone = request.clone();
 519        let connection = self.client.clone();
 520        self.executor.spawn(async move {
 521            let args = request_clone.to_dap();
 522            let response = connection.request::<R::DapRequest>(args).await?;
 523            request.response_from_dap(response)
 524        })
 525    }
 526}
 527
 528impl Mode {
 529    pub(super) fn request_dap<R: DapCommand>(&self, request: R) -> Task<Result<R::Response>>
 530    where
 531        <R::DapRequest as dap::requests::Request>::Response: 'static,
 532        <R::DapRequest as dap::requests::Request>::Arguments: 'static + Send,
 533    {
 534        match self {
 535            Mode::Running(debug_adapter_client) => debug_adapter_client.request(request),
 536            Mode::Building => Task::ready(Err(anyhow!(
 537                "no adapter running to send request: {request:?}"
 538            ))),
 539        }
 540    }
 541
 542    /// Did this debug session stop at least once?
 543    pub(crate) fn has_ever_stopped(&self) -> bool {
 544        match self {
 545            Mode::Building => false,
 546            Mode::Running(running_mode) => running_mode.has_ever_stopped,
 547        }
 548    }
 549
 550    fn stopped(&mut self) {
 551        if let Mode::Running(running) = self {
 552            running.has_ever_stopped = true;
 553        }
 554    }
 555}
 556
 557#[derive(Default)]
 558struct ThreadStates {
 559    global_state: Option<ThreadStatus>,
 560    known_thread_states: IndexMap<ThreadId, ThreadStatus>,
 561}
 562
 563impl ThreadStates {
 564    fn stop_all_threads(&mut self) {
 565        self.global_state = Some(ThreadStatus::Stopped);
 566        self.known_thread_states.clear();
 567    }
 568
 569    fn exit_all_threads(&mut self) {
 570        self.global_state = Some(ThreadStatus::Exited);
 571        self.known_thread_states.clear();
 572    }
 573
 574    fn continue_all_threads(&mut self) {
 575        self.global_state = Some(ThreadStatus::Running);
 576        self.known_thread_states.clear();
 577    }
 578
 579    fn stop_thread(&mut self, thread_id: ThreadId) {
 580        self.known_thread_states
 581            .insert(thread_id, ThreadStatus::Stopped);
 582    }
 583
 584    fn continue_thread(&mut self, thread_id: ThreadId) {
 585        self.known_thread_states
 586            .insert(thread_id, ThreadStatus::Running);
 587    }
 588
 589    fn process_step(&mut self, thread_id: ThreadId) {
 590        self.known_thread_states
 591            .insert(thread_id, ThreadStatus::Stepping);
 592    }
 593
 594    fn thread_status(&self, thread_id: ThreadId) -> ThreadStatus {
 595        self.thread_state(thread_id)
 596            .unwrap_or(ThreadStatus::Running)
 597    }
 598
 599    fn thread_state(&self, thread_id: ThreadId) -> Option<ThreadStatus> {
 600        self.known_thread_states
 601            .get(&thread_id)
 602            .copied()
 603            .or(self.global_state)
 604    }
 605
 606    fn exit_thread(&mut self, thread_id: ThreadId) {
 607        self.known_thread_states
 608            .insert(thread_id, ThreadStatus::Exited);
 609    }
 610
 611    fn any_stopped_thread(&self) -> bool {
 612        self.global_state
 613            .is_some_and(|state| state == ThreadStatus::Stopped)
 614            || self
 615                .known_thread_states
 616                .values()
 617                .any(|status| *status == ThreadStatus::Stopped)
 618    }
 619}
 620const MAX_TRACKED_OUTPUT_EVENTS: usize = 5000;
 621
 622type IsEnabled = bool;
 623
 624#[derive(Copy, Clone, Default, Debug, PartialEq, PartialOrd, Eq, Ord)]
 625pub struct OutputToken(pub usize);
 626/// Represents a current state of a single debug adapter and provides ways to mutate it.
 627pub struct Session {
 628    pub mode: Mode,
 629    id: SessionId,
 630    label: SharedString,
 631    adapter: DebugAdapterName,
 632    pub(super) capabilities: Capabilities,
 633    child_session_ids: HashSet<SessionId>,
 634    parent_session: Option<Entity<Session>>,
 635    modules: Vec<dap::Module>,
 636    loaded_sources: Vec<dap::Source>,
 637    output_token: OutputToken,
 638    output: Box<circular_buffer::CircularBuffer<MAX_TRACKED_OUTPUT_EVENTS, dap::OutputEvent>>,
 639    threads: IndexMap<ThreadId, Thread>,
 640    thread_states: ThreadStates,
 641    watchers: HashMap<SharedString, Watcher>,
 642    variables: HashMap<VariableReference, Vec<dap::Variable>>,
 643    stack_frames: IndexMap<StackFrameId, StackFrame>,
 644    locations: HashMap<u64, dap::LocationsResponse>,
 645    is_session_terminated: bool,
 646    requests: HashMap<TypeId, HashMap<RequestSlot, Shared<Task<Option<()>>>>>,
 647    pub(crate) breakpoint_store: Entity<BreakpointStore>,
 648    ignore_breakpoints: bool,
 649    exception_breakpoints: BTreeMap<String, (ExceptionBreakpointsFilter, IsEnabled)>,
 650    background_tasks: Vec<Task<()>>,
 651    task_context: TaskContext,
 652}
 653
 654trait CacheableCommand: Any + Send + Sync {
 655    fn dyn_eq(&self, rhs: &dyn CacheableCommand) -> bool;
 656    fn dyn_hash(&self, hasher: &mut dyn Hasher);
 657    fn as_any_arc(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
 658}
 659
 660impl<T> CacheableCommand for T
 661where
 662    T: DapCommand + PartialEq + Eq + Hash,
 663{
 664    fn dyn_eq(&self, rhs: &dyn CacheableCommand) -> bool {
 665        (rhs as &dyn Any)
 666            .downcast_ref::<Self>()
 667            .map_or(false, |rhs| self == rhs)
 668    }
 669
 670    fn dyn_hash(&self, mut hasher: &mut dyn Hasher) {
 671        T::hash(self, &mut hasher);
 672    }
 673
 674    fn as_any_arc(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
 675        self
 676    }
 677}
 678
 679pub(crate) struct RequestSlot(Arc<dyn CacheableCommand>);
 680
 681impl<T: DapCommand + PartialEq + Eq + Hash> From<T> for RequestSlot {
 682    fn from(request: T) -> Self {
 683        Self(Arc::new(request))
 684    }
 685}
 686
 687impl PartialEq for RequestSlot {
 688    fn eq(&self, other: &Self) -> bool {
 689        self.0.dyn_eq(other.0.as_ref())
 690    }
 691}
 692
 693impl Eq for RequestSlot {}
 694
 695impl Hash for RequestSlot {
 696    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
 697        self.0.dyn_hash(state);
 698        (&*self.0 as &dyn Any).type_id().hash(state)
 699    }
 700}
 701
 702#[derive(Debug, Clone, Hash, PartialEq, Eq)]
 703pub struct CompletionsQuery {
 704    pub query: String,
 705    pub column: u64,
 706    pub line: Option<u64>,
 707    pub frame_id: Option<u64>,
 708}
 709
 710impl CompletionsQuery {
 711    pub fn new(
 712        buffer: &language::Buffer,
 713        cursor_position: language::Anchor,
 714        frame_id: Option<u64>,
 715    ) -> Self {
 716        let PointUtf16 { row, column } = cursor_position.to_point_utf16(&buffer.snapshot());
 717        Self {
 718            query: buffer.text(),
 719            column: column as u64,
 720            frame_id,
 721            line: Some(row as u64),
 722        }
 723    }
 724}
 725
 726#[derive(Debug)]
 727pub enum SessionEvent {
 728    Modules,
 729    LoadedSources,
 730    Stopped(Option<ThreadId>),
 731    StackTrace,
 732    Variables,
 733    Watchers,
 734    Threads,
 735    InvalidateInlineValue,
 736    CapabilitiesLoaded,
 737    RunInTerminal {
 738        request: RunInTerminalRequestArguments,
 739        sender: mpsc::Sender<Result<u32>>,
 740    },
 741    ConsoleOutput,
 742}
 743
 744#[derive(Clone, Debug, PartialEq, Eq)]
 745pub enum SessionStateEvent {
 746    Running,
 747    Shutdown,
 748    Restart,
 749    SpawnChildSession {
 750        request: StartDebuggingRequestArguments,
 751    },
 752}
 753
 754impl EventEmitter<SessionEvent> for Session {}
 755impl EventEmitter<SessionStateEvent> for Session {}
 756
 757// local session will send breakpoint updates to DAP for all new breakpoints
 758// remote side will only send breakpoint updates when it is a breakpoint created by that peer
 759// BreakpointStore notifies session on breakpoint changes
 760impl Session {
 761    pub(crate) fn new(
 762        breakpoint_store: Entity<BreakpointStore>,
 763        session_id: SessionId,
 764        parent_session: Option<Entity<Session>>,
 765        label: SharedString,
 766        adapter: DebugAdapterName,
 767        task_context: TaskContext,
 768        cx: &mut App,
 769    ) -> Entity<Self> {
 770        cx.new::<Self>(|cx| {
 771            cx.subscribe(&breakpoint_store, |this, store, event, cx| match event {
 772                BreakpointStoreEvent::BreakpointsUpdated(path, reason) => {
 773                    if let Some(local) = (!this.ignore_breakpoints)
 774                        .then(|| this.as_running_mut())
 775                        .flatten()
 776                    {
 777                        local
 778                            .send_breakpoints_from_path(path.clone(), *reason, &store, cx)
 779                            .detach();
 780                    };
 781                }
 782                BreakpointStoreEvent::BreakpointsCleared(paths) => {
 783                    if let Some(local) = (!this.ignore_breakpoints)
 784                        .then(|| this.as_running_mut())
 785                        .flatten()
 786                    {
 787                        local.unset_breakpoints_from_paths(paths, cx).detach();
 788                    }
 789                }
 790                BreakpointStoreEvent::SetDebugLine | BreakpointStoreEvent::ClearDebugLines => {}
 791            })
 792            .detach();
 793            cx.on_app_quit(Self::on_app_quit).detach();
 794
 795            let this = Self {
 796                mode: Mode::Building,
 797                id: session_id,
 798                child_session_ids: HashSet::default(),
 799                parent_session,
 800                capabilities: Capabilities::default(),
 801                watchers: HashMap::default(),
 802                variables: Default::default(),
 803                stack_frames: Default::default(),
 804                thread_states: ThreadStates::default(),
 805                output_token: OutputToken(0),
 806                output: circular_buffer::CircularBuffer::boxed(),
 807                requests: HashMap::default(),
 808                modules: Vec::default(),
 809                loaded_sources: Vec::default(),
 810                threads: IndexMap::default(),
 811                background_tasks: Vec::default(),
 812                locations: Default::default(),
 813                is_session_terminated: false,
 814                ignore_breakpoints: false,
 815                breakpoint_store,
 816                exception_breakpoints: Default::default(),
 817                label,
 818                adapter,
 819                task_context,
 820            };
 821
 822            this
 823        })
 824    }
 825
 826    pub fn task_context(&self) -> &TaskContext {
 827        &self.task_context
 828    }
 829
 830    pub fn worktree(&self) -> Option<Entity<Worktree>> {
 831        match &self.mode {
 832            Mode::Building => None,
 833            Mode::Running(local_mode) => local_mode.worktree.upgrade(),
 834        }
 835    }
 836
 837    pub fn boot(
 838        &mut self,
 839        binary: DebugAdapterBinary,
 840        worktree: Entity<Worktree>,
 841        dap_store: WeakEntity<DapStore>,
 842        cx: &mut Context<Self>,
 843    ) -> Task<Result<()>> {
 844        let (message_tx, mut message_rx) = futures::channel::mpsc::unbounded();
 845        let (initialized_tx, initialized_rx) = futures::channel::oneshot::channel();
 846
 847        let background_tasks = vec![cx.spawn(async move |this: WeakEntity<Session>, cx| {
 848            let mut initialized_tx = Some(initialized_tx);
 849            while let Some(message) = message_rx.next().await {
 850                if let Message::Event(event) = message {
 851                    if let Events::Initialized(_) = *event {
 852                        if let Some(tx) = initialized_tx.take() {
 853                            tx.send(()).ok();
 854                        }
 855                    } else {
 856                        let Ok(_) = this.update(cx, |session, cx| {
 857                            session.handle_dap_event(event, cx);
 858                        }) else {
 859                            break;
 860                        };
 861                    }
 862                } else if let Message::Request(request) = message {
 863                    let Ok(_) = this.update(cx, |this, cx| {
 864                        if request.command == StartDebugging::COMMAND {
 865                            this.handle_start_debugging_request(request, cx)
 866                                .detach_and_log_err(cx);
 867                        } else if request.command == RunInTerminal::COMMAND {
 868                            this.handle_run_in_terminal_request(request, cx)
 869                                .detach_and_log_err(cx);
 870                        }
 871                    }) else {
 872                        break;
 873                    };
 874                }
 875            }
 876        })];
 877        self.background_tasks = background_tasks;
 878        let id = self.id;
 879        let parent_session = self.parent_session.clone();
 880
 881        cx.spawn(async move |this, cx| {
 882            let mode = RunningMode::new(
 883                id,
 884                parent_session,
 885                worktree.downgrade(),
 886                binary.clone(),
 887                message_tx,
 888                cx,
 889            )
 890            .await?;
 891            this.update(cx, |this, cx| {
 892                this.mode = Mode::Running(mode);
 893                cx.emit(SessionStateEvent::Running);
 894            })?;
 895
 896            this.update(cx, |session, cx| session.request_initialize(cx))?
 897                .await?;
 898
 899            let result = this
 900                .update(cx, |session, cx| {
 901                    session.initialize_sequence(initialized_rx, dap_store.clone(), cx)
 902                })?
 903                .await;
 904
 905            if result.is_err() {
 906                let mut console = this.update(cx, |session, cx| session.console_output(cx))?;
 907
 908                console
 909                    .send(format!(
 910                        "Tried to launch debugger with: {}",
 911                        serde_json::to_string_pretty(&binary.request_args.configuration)
 912                            .unwrap_or_default(),
 913                    ))
 914                    .await
 915                    .ok();
 916            }
 917
 918            result
 919        })
 920    }
 921
 922    pub fn session_id(&self) -> SessionId {
 923        self.id
 924    }
 925
 926    pub fn child_session_ids(&self) -> HashSet<SessionId> {
 927        self.child_session_ids.clone()
 928    }
 929
 930    pub fn add_child_session_id(&mut self, session_id: SessionId) {
 931        self.child_session_ids.insert(session_id);
 932    }
 933
 934    pub fn remove_child_session_id(&mut self, session_id: SessionId) {
 935        self.child_session_ids.remove(&session_id);
 936    }
 937
 938    pub fn parent_id(&self, cx: &App) -> Option<SessionId> {
 939        self.parent_session
 940            .as_ref()
 941            .map(|session| session.read(cx).id)
 942    }
 943
 944    pub fn parent_session(&self) -> Option<&Entity<Self>> {
 945        self.parent_session.as_ref()
 946    }
 947
 948    pub fn capabilities(&self) -> &Capabilities {
 949        &self.capabilities
 950    }
 951
 952    pub fn binary(&self) -> Option<&DebugAdapterBinary> {
 953        match &self.mode {
 954            Mode::Building => None,
 955            Mode::Running(running_mode) => Some(&running_mode.binary),
 956        }
 957    }
 958
 959    pub fn adapter(&self) -> DebugAdapterName {
 960        self.adapter.clone()
 961    }
 962
 963    pub fn label(&self) -> SharedString {
 964        self.label.clone()
 965    }
 966
 967    pub fn is_terminated(&self) -> bool {
 968        self.is_session_terminated
 969    }
 970
 971    pub fn console_output(&mut self, cx: &mut Context<Self>) -> mpsc::UnboundedSender<String> {
 972        let (tx, mut rx) = mpsc::unbounded();
 973
 974        cx.spawn(async move |this, cx| {
 975            while let Some(output) = rx.next().await {
 976                this.update(cx, |this, cx| {
 977                    let event = dap::OutputEvent {
 978                        category: None,
 979                        output,
 980                        group: None,
 981                        variables_reference: None,
 982                        source: None,
 983                        line: None,
 984                        column: None,
 985                        data: None,
 986                        location_reference: None,
 987                    };
 988                    this.push_output(event, cx);
 989                })?;
 990            }
 991            anyhow::Ok(())
 992        })
 993        .detach();
 994
 995        return tx;
 996    }
 997
 998    pub fn is_started(&self) -> bool {
 999        match &self.mode {
1000            Mode::Building => false,
1001            Mode::Running(running) => running.is_started,
1002        }
1003    }
1004
1005    pub fn is_building(&self) -> bool {
1006        matches!(self.mode, Mode::Building)
1007    }
1008
1009    pub fn is_running(&self) -> bool {
1010        matches!(self.mode, Mode::Running(_))
1011    }
1012
1013    pub fn as_running_mut(&mut self) -> Option<&mut RunningMode> {
1014        match &mut self.mode {
1015            Mode::Running(local_mode) => Some(local_mode),
1016            Mode::Building => None,
1017        }
1018    }
1019
1020    pub fn as_running(&self) -> Option<&RunningMode> {
1021        match &self.mode {
1022            Mode::Running(local_mode) => Some(local_mode),
1023            Mode::Building => None,
1024        }
1025    }
1026
1027    fn handle_start_debugging_request(
1028        &mut self,
1029        request: dap::messages::Request,
1030        cx: &mut Context<Self>,
1031    ) -> Task<Result<()>> {
1032        let request_seq = request.seq;
1033
1034        let launch_request: Option<Result<StartDebuggingRequestArguments, _>> = request
1035            .arguments
1036            .as_ref()
1037            .map(|value| serde_json::from_value(value.clone()));
1038
1039        let mut success = true;
1040        if let Some(Ok(request)) = launch_request {
1041            cx.emit(SessionStateEvent::SpawnChildSession { request });
1042        } else {
1043            log::error!(
1044                "Failed to parse launch request arguments: {:?}",
1045                request.arguments
1046            );
1047            success = false;
1048        }
1049
1050        cx.spawn(async move |this, cx| {
1051            this.update(cx, |this, cx| {
1052                this.respond_to_client(
1053                    request_seq,
1054                    success,
1055                    StartDebugging::COMMAND.to_string(),
1056                    None,
1057                    cx,
1058                )
1059            })?
1060            .await
1061        })
1062    }
1063
1064    fn handle_run_in_terminal_request(
1065        &mut self,
1066        request: dap::messages::Request,
1067        cx: &mut Context<Self>,
1068    ) -> Task<Result<()>> {
1069        let request_args = match serde_json::from_value::<RunInTerminalRequestArguments>(
1070            request.arguments.unwrap_or_default(),
1071        ) {
1072            Ok(args) => args,
1073            Err(error) => {
1074                return cx.spawn(async move |session, cx| {
1075                    let error = serde_json::to_value(dap::ErrorResponse {
1076                        error: Some(dap::Message {
1077                            id: request.seq,
1078                            format: error.to_string(),
1079                            variables: None,
1080                            send_telemetry: None,
1081                            show_user: None,
1082                            url: None,
1083                            url_label: None,
1084                        }),
1085                    })
1086                    .ok();
1087
1088                    session
1089                        .update(cx, |this, cx| {
1090                            this.respond_to_client(
1091                                request.seq,
1092                                false,
1093                                StartDebugging::COMMAND.to_string(),
1094                                error,
1095                                cx,
1096                            )
1097                        })?
1098                        .await?;
1099
1100                    Err(anyhow!("Failed to parse RunInTerminalRequestArguments"))
1101                });
1102            }
1103        };
1104
1105        let seq = request.seq;
1106
1107        let (tx, mut rx) = mpsc::channel::<Result<u32>>(1);
1108        cx.emit(SessionEvent::RunInTerminal {
1109            request: request_args,
1110            sender: tx,
1111        });
1112        cx.notify();
1113
1114        cx.spawn(async move |session, cx| {
1115            let result = util::maybe!(async move {
1116                rx.next().await.ok_or_else(|| {
1117                    anyhow!("failed to receive response from spawn terminal".to_string())
1118                })?
1119            })
1120            .await;
1121            let (success, body) = match result {
1122                Ok(pid) => (
1123                    true,
1124                    serde_json::to_value(dap::RunInTerminalResponse {
1125                        process_id: None,
1126                        shell_process_id: Some(pid as u64),
1127                    })
1128                    .ok(),
1129                ),
1130                Err(error) => (
1131                    false,
1132                    serde_json::to_value(dap::ErrorResponse {
1133                        error: Some(dap::Message {
1134                            id: seq,
1135                            format: error.to_string(),
1136                            variables: None,
1137                            send_telemetry: None,
1138                            show_user: None,
1139                            url: None,
1140                            url_label: None,
1141                        }),
1142                    })
1143                    .ok(),
1144                ),
1145            };
1146
1147            session
1148                .update(cx, |session, cx| {
1149                    session.respond_to_client(
1150                        seq,
1151                        success,
1152                        RunInTerminal::COMMAND.to_string(),
1153                        body,
1154                        cx,
1155                    )
1156                })?
1157                .await
1158        })
1159    }
1160
1161    pub(super) fn request_initialize(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1162        let adapter_id = self.adapter().to_string();
1163        let request = Initialize { adapter_id };
1164
1165        let Mode::Running(running) = &self.mode else {
1166            return Task::ready(Err(anyhow!(
1167                "Cannot send initialize request, task still building"
1168            )));
1169        };
1170        let mut response = running.request(request.clone());
1171
1172        cx.spawn(async move |this, cx| {
1173            loop {
1174                let capabilities = response.await;
1175                match capabilities {
1176                    Err(e) => {
1177                        let Ok(Some(reconnect)) = this.update(cx, |this, cx| {
1178                            this.as_running()
1179                                .and_then(|running| running.reconnect_for_ssh(&mut cx.to_async()))
1180                        }) else {
1181                            return Err(e);
1182                        };
1183                        log::info!("Failed to connect to debug adapter: {}, retrying...", e);
1184                        reconnect.await?;
1185
1186                        let Ok(Some(r)) = this.update(cx, |this, _| {
1187                            this.as_running()
1188                                .map(|running| running.request(request.clone()))
1189                        }) else {
1190                            return Err(e);
1191                        };
1192                        response = r
1193                    }
1194                    Ok(capabilities) => {
1195                        this.update(cx, |session, cx| {
1196                            session.capabilities = capabilities;
1197                            let filters = session
1198                                .capabilities
1199                                .exception_breakpoint_filters
1200                                .clone()
1201                                .unwrap_or_default();
1202                            for filter in filters {
1203                                let default = filter.default.unwrap_or_default();
1204                                session
1205                                    .exception_breakpoints
1206                                    .entry(filter.filter.clone())
1207                                    .or_insert_with(|| (filter, default));
1208                            }
1209                            cx.emit(SessionEvent::CapabilitiesLoaded);
1210                        })?;
1211                        return Ok(());
1212                    }
1213                }
1214            }
1215        })
1216    }
1217
1218    pub(super) fn initialize_sequence(
1219        &mut self,
1220        initialize_rx: oneshot::Receiver<()>,
1221        dap_store: WeakEntity<DapStore>,
1222        cx: &mut Context<Self>,
1223    ) -> Task<Result<()>> {
1224        match &self.mode {
1225            Mode::Running(local_mode) => {
1226                local_mode.initialize_sequence(&self.capabilities, initialize_rx, dap_store, cx)
1227            }
1228            Mode::Building => Task::ready(Err(anyhow!("cannot initialize, still building"))),
1229        }
1230    }
1231
1232    pub fn run_to_position(
1233        &mut self,
1234        breakpoint: SourceBreakpoint,
1235        active_thread_id: ThreadId,
1236        cx: &mut Context<Self>,
1237    ) {
1238        match &mut self.mode {
1239            Mode::Running(local_mode) => {
1240                if !matches!(
1241                    self.thread_states.thread_state(active_thread_id),
1242                    Some(ThreadStatus::Stopped)
1243                ) {
1244                    return;
1245                };
1246                let path = breakpoint.path.clone();
1247                local_mode.tmp_breakpoint = Some(breakpoint);
1248                let task = local_mode.send_breakpoints_from_path(
1249                    path,
1250                    BreakpointUpdatedReason::Toggled,
1251                    &self.breakpoint_store,
1252                    cx,
1253                );
1254
1255                cx.spawn(async move |this, cx| {
1256                    task.await;
1257                    this.update(cx, |this, cx| {
1258                        this.continue_thread(active_thread_id, cx);
1259                    })
1260                })
1261                .detach();
1262            }
1263            Mode::Building => {}
1264        }
1265    }
1266
1267    pub fn has_new_output(&self, last_update: OutputToken) -> bool {
1268        self.output_token.0.checked_sub(last_update.0).unwrap_or(0) != 0
1269    }
1270
1271    pub fn output(
1272        &self,
1273        since: OutputToken,
1274    ) -> (impl Iterator<Item = &dap::OutputEvent>, OutputToken) {
1275        if self.output_token.0 == 0 {
1276            return (self.output.range(0..0), OutputToken(0));
1277        };
1278
1279        let events_since = self.output_token.0.checked_sub(since.0).unwrap_or(0);
1280
1281        let clamped_events_since = events_since.clamp(0, self.output.len());
1282        (
1283            self.output
1284                .range(self.output.len() - clamped_events_since..),
1285            self.output_token,
1286        )
1287    }
1288
1289    pub fn respond_to_client(
1290        &self,
1291        request_seq: u64,
1292        success: bool,
1293        command: String,
1294        body: Option<serde_json::Value>,
1295        cx: &mut Context<Self>,
1296    ) -> Task<Result<()>> {
1297        let Some(local_session) = self.as_running() else {
1298            unreachable!("Cannot respond to remote client");
1299        };
1300        let client = local_session.client.clone();
1301
1302        cx.background_spawn(async move {
1303            client
1304                .send_message(Message::Response(Response {
1305                    body,
1306                    success,
1307                    command,
1308                    seq: request_seq + 1,
1309                    request_seq,
1310                    message: None,
1311                }))
1312                .await
1313        })
1314    }
1315
1316    fn handle_stopped_event(&mut self, event: StoppedEvent, cx: &mut Context<Self>) {
1317        self.mode.stopped();
1318        // todo(debugger): Find a clean way to get around the clone
1319        let breakpoint_store = self.breakpoint_store.clone();
1320        if let Some((local, path)) = self.as_running_mut().and_then(|local| {
1321            let breakpoint = local.tmp_breakpoint.take()?;
1322            let path = breakpoint.path.clone();
1323            Some((local, path))
1324        }) {
1325            local
1326                .send_breakpoints_from_path(
1327                    path,
1328                    BreakpointUpdatedReason::Toggled,
1329                    &breakpoint_store,
1330                    cx,
1331                )
1332                .detach();
1333        };
1334
1335        if event.all_threads_stopped.unwrap_or_default() || event.thread_id.is_none() {
1336            self.thread_states.stop_all_threads();
1337            self.invalidate_command_type::<StackTraceCommand>();
1338        }
1339
1340        // Event if we stopped all threads we still need to insert the thread_id
1341        // to our own data
1342        if let Some(thread_id) = event.thread_id {
1343            self.thread_states.stop_thread(ThreadId(thread_id));
1344
1345            self.invalidate_state(
1346                &StackTraceCommand {
1347                    thread_id,
1348                    start_frame: None,
1349                    levels: None,
1350                }
1351                .into(),
1352            );
1353        }
1354
1355        self.invalidate_generic();
1356        self.threads.clear();
1357        self.variables.clear();
1358        cx.emit(SessionEvent::Stopped(
1359            event
1360                .thread_id
1361                .map(Into::into)
1362                .filter(|_| !event.preserve_focus_hint.unwrap_or(false)),
1363        ));
1364        cx.emit(SessionEvent::InvalidateInlineValue);
1365        cx.notify();
1366    }
1367
1368    pub(crate) fn handle_dap_event(&mut self, event: Box<Events>, cx: &mut Context<Self>) {
1369        match *event {
1370            Events::Initialized(_) => {
1371                debug_assert!(
1372                    false,
1373                    "Initialized event should have been handled in LocalMode"
1374                );
1375            }
1376            Events::Stopped(event) => self.handle_stopped_event(event, cx),
1377            Events::Continued(event) => {
1378                if event.all_threads_continued.unwrap_or_default() {
1379                    self.thread_states.continue_all_threads();
1380                    self.breakpoint_store.update(cx, |store, cx| {
1381                        store.remove_active_position(Some(self.session_id()), cx)
1382                    });
1383                } else {
1384                    self.thread_states
1385                        .continue_thread(ThreadId(event.thread_id));
1386                }
1387                // todo(debugger): We should be able to get away with only invalidating generic if all threads were continued
1388                self.invalidate_generic();
1389            }
1390            Events::Exited(_event) => {
1391                self.clear_active_debug_line(cx);
1392            }
1393            Events::Terminated(_) => {
1394                self.shutdown(cx).detach();
1395            }
1396            Events::Thread(event) => {
1397                let thread_id = ThreadId(event.thread_id);
1398
1399                match event.reason {
1400                    dap::ThreadEventReason::Started => {
1401                        self.thread_states.continue_thread(thread_id);
1402                    }
1403                    dap::ThreadEventReason::Exited => {
1404                        self.thread_states.exit_thread(thread_id);
1405                    }
1406                    reason => {
1407                        log::error!("Unhandled thread event reason {:?}", reason);
1408                    }
1409                }
1410                self.invalidate_state(&ThreadsCommand.into());
1411                cx.notify();
1412            }
1413            Events::Output(event) => {
1414                if event
1415                    .category
1416                    .as_ref()
1417                    .is_some_and(|category| *category == OutputEventCategory::Telemetry)
1418                {
1419                    return;
1420                }
1421
1422                self.push_output(event, cx);
1423                cx.notify();
1424            }
1425            Events::Breakpoint(event) => self.breakpoint_store.update(cx, |store, _| {
1426                store.update_session_breakpoint(self.session_id(), event.reason, event.breakpoint);
1427            }),
1428            Events::Module(event) => {
1429                match event.reason {
1430                    dap::ModuleEventReason::New => {
1431                        self.modules.push(event.module);
1432                    }
1433                    dap::ModuleEventReason::Changed => {
1434                        if let Some(module) = self
1435                            .modules
1436                            .iter_mut()
1437                            .find(|other| event.module.id == other.id)
1438                        {
1439                            *module = event.module;
1440                        }
1441                    }
1442                    dap::ModuleEventReason::Removed => {
1443                        self.modules.retain(|other| event.module.id != other.id);
1444                    }
1445                }
1446
1447                // todo(debugger): We should only send the invalidate command to downstream clients.
1448                // self.invalidate_state(&ModulesCommand.into());
1449            }
1450            Events::LoadedSource(_) => {
1451                self.invalidate_state(&LoadedSourcesCommand.into());
1452            }
1453            Events::Capabilities(event) => {
1454                self.capabilities = self.capabilities.merge(event.capabilities);
1455                cx.notify();
1456            }
1457            Events::Memory(_) => {}
1458            Events::Process(_) => {}
1459            Events::ProgressEnd(_) => {}
1460            Events::ProgressStart(_) => {}
1461            Events::ProgressUpdate(_) => {}
1462            Events::Invalidated(_) => {}
1463            Events::Other(_) => {}
1464        }
1465    }
1466
1467    /// Ensure that there's a request in flight for the given command, and if not, send it. Use this to run requests that are idempotent.
1468    fn fetch<T: DapCommand + PartialEq + Eq + Hash>(
1469        &mut self,
1470        request: T,
1471        process_result: impl FnOnce(&mut Self, Result<T::Response>, &mut Context<Self>) + 'static,
1472        cx: &mut Context<Self>,
1473    ) {
1474        const {
1475            assert!(
1476                T::CACHEABLE,
1477                "Only requests marked as cacheable should invoke `fetch`"
1478            );
1479        }
1480
1481        if !self.thread_states.any_stopped_thread()
1482            && request.type_id() != TypeId::of::<ThreadsCommand>()
1483            || self.is_session_terminated
1484        {
1485            return;
1486        }
1487
1488        let request_map = self
1489            .requests
1490            .entry(std::any::TypeId::of::<T>())
1491            .or_default();
1492
1493        if let Entry::Vacant(vacant) = request_map.entry(request.into()) {
1494            let command = vacant.key().0.clone().as_any_arc().downcast::<T>().unwrap();
1495
1496            let task = Self::request_inner::<Arc<T>>(
1497                &self.capabilities,
1498                &self.mode,
1499                command,
1500                |this, result, cx| {
1501                    process_result(this, result, cx);
1502                    None
1503                },
1504                cx,
1505            );
1506            let task = cx
1507                .background_executor()
1508                .spawn(async move {
1509                    let _ = task.await?;
1510                    Some(())
1511                })
1512                .shared();
1513
1514            vacant.insert(task);
1515            cx.notify();
1516        }
1517    }
1518
1519    fn request_inner<T: DapCommand + PartialEq + Eq + Hash>(
1520        capabilities: &Capabilities,
1521        mode: &Mode,
1522        request: T,
1523        process_result: impl FnOnce(
1524            &mut Self,
1525            Result<T::Response>,
1526            &mut Context<Self>,
1527        ) -> Option<T::Response>
1528        + 'static,
1529        cx: &mut Context<Self>,
1530    ) -> Task<Option<T::Response>> {
1531        if !T::is_supported(&capabilities) {
1532            log::warn!(
1533                "Attempted to send a DAP request that isn't supported: {:?}",
1534                request
1535            );
1536            let error = Err(anyhow::Error::msg(
1537                "Couldn't complete request because it's not supported",
1538            ));
1539            return cx.spawn(async move |this, cx| {
1540                this.update(cx, |this, cx| process_result(this, error, cx))
1541                    .ok()
1542                    .flatten()
1543            });
1544        }
1545
1546        let request = mode.request_dap(request);
1547        cx.spawn(async move |this, cx| {
1548            let result = request.await;
1549            this.update(cx, |this, cx| process_result(this, result, cx))
1550                .ok()
1551                .flatten()
1552        })
1553    }
1554
1555    fn request<T: DapCommand + PartialEq + Eq + Hash>(
1556        &self,
1557        request: T,
1558        process_result: impl FnOnce(
1559            &mut Self,
1560            Result<T::Response>,
1561            &mut Context<Self>,
1562        ) -> Option<T::Response>
1563        + 'static,
1564        cx: &mut Context<Self>,
1565    ) -> Task<Option<T::Response>> {
1566        Self::request_inner(&self.capabilities, &self.mode, request, process_result, cx)
1567    }
1568
1569    fn invalidate_command_type<Command: DapCommand>(&mut self) {
1570        self.requests.remove(&std::any::TypeId::of::<Command>());
1571    }
1572
1573    fn invalidate_generic(&mut self) {
1574        self.invalidate_command_type::<ModulesCommand>();
1575        self.invalidate_command_type::<LoadedSourcesCommand>();
1576        self.invalidate_command_type::<ThreadsCommand>();
1577    }
1578
1579    fn invalidate_state(&mut self, key: &RequestSlot) {
1580        self.requests
1581            .entry((&*key.0 as &dyn Any).type_id())
1582            .and_modify(|request_map| {
1583                request_map.remove(&key);
1584            });
1585    }
1586
1587    fn push_output(&mut self, event: OutputEvent, cx: &mut Context<Self>) {
1588        self.output.push_back(event);
1589        self.output_token.0 += 1;
1590        cx.emit(SessionEvent::ConsoleOutput);
1591    }
1592
1593    pub fn any_stopped_thread(&self) -> bool {
1594        self.thread_states.any_stopped_thread()
1595    }
1596
1597    pub fn thread_status(&self, thread_id: ThreadId) -> ThreadStatus {
1598        self.thread_states.thread_status(thread_id)
1599    }
1600
1601    pub fn threads(&mut self, cx: &mut Context<Self>) -> Vec<(dap::Thread, ThreadStatus)> {
1602        self.fetch(
1603            dap_command::ThreadsCommand,
1604            |this, result, cx| {
1605                let Some(result) = result.log_err() else {
1606                    return;
1607                };
1608
1609                this.threads = result
1610                    .into_iter()
1611                    .map(|thread| (ThreadId(thread.id), Thread::from(thread.clone())))
1612                    .collect();
1613
1614                this.invalidate_command_type::<StackTraceCommand>();
1615                cx.emit(SessionEvent::Threads);
1616                cx.notify();
1617            },
1618            cx,
1619        );
1620
1621        self.threads
1622            .values()
1623            .map(|thread| {
1624                (
1625                    thread.dap.clone(),
1626                    self.thread_states.thread_status(ThreadId(thread.dap.id)),
1627                )
1628            })
1629            .collect()
1630    }
1631
1632    pub fn modules(&mut self, cx: &mut Context<Self>) -> &[Module] {
1633        self.fetch(
1634            dap_command::ModulesCommand,
1635            |this, result, cx| {
1636                let Some(result) = result.log_err() else {
1637                    return;
1638                };
1639
1640                this.modules = result;
1641                cx.emit(SessionEvent::Modules);
1642                cx.notify();
1643            },
1644            cx,
1645        );
1646
1647        &self.modules
1648    }
1649
1650    pub fn ignore_breakpoints(&self) -> bool {
1651        self.ignore_breakpoints
1652    }
1653
1654    pub fn toggle_ignore_breakpoints(
1655        &mut self,
1656        cx: &mut App,
1657    ) -> Task<HashMap<Arc<Path>, anyhow::Error>> {
1658        self.set_ignore_breakpoints(!self.ignore_breakpoints, cx)
1659    }
1660
1661    pub(crate) fn set_ignore_breakpoints(
1662        &mut self,
1663        ignore: bool,
1664        cx: &mut App,
1665    ) -> Task<HashMap<Arc<Path>, anyhow::Error>> {
1666        if self.ignore_breakpoints == ignore {
1667            return Task::ready(HashMap::default());
1668        }
1669
1670        self.ignore_breakpoints = ignore;
1671
1672        if let Some(local) = self.as_running() {
1673            local.send_source_breakpoints(ignore, &self.breakpoint_store, cx)
1674        } else {
1675            // todo(debugger): We need to propagate this change to downstream sessions and send a message to upstream sessions
1676            unimplemented!()
1677        }
1678    }
1679
1680    pub fn exception_breakpoints(
1681        &self,
1682    ) -> impl Iterator<Item = &(ExceptionBreakpointsFilter, IsEnabled)> {
1683        self.exception_breakpoints.values()
1684    }
1685
1686    pub fn toggle_exception_breakpoint(&mut self, id: &str, cx: &App) {
1687        if let Some((_, is_enabled)) = self.exception_breakpoints.get_mut(id) {
1688            *is_enabled = !*is_enabled;
1689            self.send_exception_breakpoints(cx);
1690        }
1691    }
1692
1693    fn send_exception_breakpoints(&mut self, cx: &App) {
1694        if let Some(local) = self.as_running() {
1695            let exception_filters = self
1696                .exception_breakpoints
1697                .values()
1698                .filter_map(|(filter, is_enabled)| is_enabled.then(|| filter.clone()))
1699                .collect();
1700
1701            let supports_exception_filters = self
1702                .capabilities
1703                .supports_exception_filter_options
1704                .unwrap_or_default();
1705            local
1706                .send_exception_breakpoints(exception_filters, supports_exception_filters)
1707                .detach_and_log_err(cx);
1708        } else {
1709            debug_assert!(false, "Not implemented");
1710        }
1711    }
1712
1713    pub fn breakpoints_enabled(&self) -> bool {
1714        self.ignore_breakpoints
1715    }
1716
1717    pub fn loaded_sources(&mut self, cx: &mut Context<Self>) -> &[Source] {
1718        self.fetch(
1719            dap_command::LoadedSourcesCommand,
1720            |this, result, cx| {
1721                let Some(result) = result.log_err() else {
1722                    return;
1723                };
1724                this.loaded_sources = result;
1725                cx.emit(SessionEvent::LoadedSources);
1726                cx.notify();
1727            },
1728            cx,
1729        );
1730
1731        &self.loaded_sources
1732    }
1733
1734    fn fallback_to_manual_restart(
1735        &mut self,
1736        res: Result<()>,
1737        cx: &mut Context<Self>,
1738    ) -> Option<()> {
1739        if res.log_err().is_none() {
1740            cx.emit(SessionStateEvent::Restart);
1741            return None;
1742        }
1743        Some(())
1744    }
1745
1746    fn empty_response(&mut self, res: Result<()>, _cx: &mut Context<Self>) -> Option<()> {
1747        res.log_err()?;
1748        Some(())
1749    }
1750
1751    fn on_step_response<T: DapCommand + PartialEq + Eq + Hash>(
1752        thread_id: ThreadId,
1753    ) -> impl FnOnce(&mut Self, Result<T::Response>, &mut Context<Self>) -> Option<T::Response> + 'static
1754    {
1755        move |this, response, cx| match response.log_err() {
1756            Some(response) => {
1757                this.breakpoint_store.update(cx, |store, cx| {
1758                    store.remove_active_position(Some(this.session_id()), cx)
1759                });
1760                Some(response)
1761            }
1762            None => {
1763                this.thread_states.stop_thread(thread_id);
1764                cx.notify();
1765                None
1766            }
1767        }
1768    }
1769
1770    fn clear_active_debug_line_response(
1771        &mut self,
1772        response: Result<()>,
1773        cx: &mut Context<Session>,
1774    ) -> Option<()> {
1775        response.log_err()?;
1776        self.clear_active_debug_line(cx);
1777        Some(())
1778    }
1779
1780    fn clear_active_debug_line(&mut self, cx: &mut Context<Session>) {
1781        self.breakpoint_store.update(cx, |store, cx| {
1782            store.remove_active_position(Some(self.id), cx)
1783        });
1784    }
1785
1786    pub fn pause_thread(&mut self, thread_id: ThreadId, cx: &mut Context<Self>) {
1787        self.request(
1788            PauseCommand {
1789                thread_id: thread_id.0,
1790            },
1791            Self::empty_response,
1792            cx,
1793        )
1794        .detach();
1795    }
1796
1797    pub fn restart_stack_frame(&mut self, stack_frame_id: u64, cx: &mut Context<Self>) {
1798        self.request(
1799            RestartStackFrameCommand { stack_frame_id },
1800            Self::empty_response,
1801            cx,
1802        )
1803        .detach();
1804    }
1805
1806    pub fn restart(&mut self, args: Option<Value>, cx: &mut Context<Self>) {
1807        if self.capabilities.supports_restart_request.unwrap_or(false) && !self.is_terminated() {
1808            self.request(
1809                RestartCommand {
1810                    raw: args.unwrap_or(Value::Null),
1811                },
1812                Self::fallback_to_manual_restart,
1813                cx,
1814            )
1815            .detach();
1816        } else {
1817            cx.emit(SessionStateEvent::Restart);
1818        }
1819    }
1820
1821    fn on_app_quit(&mut self, cx: &mut Context<Self>) -> Task<()> {
1822        let debug_adapter = self.adapter_client();
1823
1824        cx.background_spawn(async move {
1825            if let Some(client) = debug_adapter {
1826                client.shutdown().await.log_err();
1827            }
1828        })
1829    }
1830
1831    pub fn shutdown(&mut self, cx: &mut Context<Self>) -> Task<()> {
1832        self.is_session_terminated = true;
1833        self.thread_states.exit_all_threads();
1834        cx.notify();
1835
1836        let task = if self
1837            .capabilities
1838            .supports_terminate_request
1839            .unwrap_or_default()
1840        {
1841            self.request(
1842                TerminateCommand {
1843                    restart: Some(false),
1844                },
1845                Self::clear_active_debug_line_response,
1846                cx,
1847            )
1848        } else {
1849            self.request(
1850                DisconnectCommand {
1851                    restart: Some(false),
1852                    terminate_debuggee: Some(true),
1853                    suspend_debuggee: Some(false),
1854                },
1855                Self::clear_active_debug_line_response,
1856                cx,
1857            )
1858        };
1859
1860        cx.emit(SessionStateEvent::Shutdown);
1861
1862        let debug_client = self.adapter_client();
1863
1864        cx.background_spawn(async move {
1865            let _ = task.await;
1866
1867            if let Some(client) = debug_client {
1868                client.shutdown().await.log_err();
1869            }
1870        })
1871    }
1872
1873    pub fn completions(
1874        &mut self,
1875        query: CompletionsQuery,
1876        cx: &mut Context<Self>,
1877    ) -> Task<Result<Vec<dap::CompletionItem>>> {
1878        let task = self.request(query, |_, result, _| result.log_err(), cx);
1879
1880        cx.background_executor().spawn(async move {
1881            anyhow::Ok(
1882                task.await
1883                    .map(|response| response.targets)
1884                    .context("failed to fetch completions")?,
1885            )
1886        })
1887    }
1888
1889    pub fn continue_thread(&mut self, thread_id: ThreadId, cx: &mut Context<Self>) {
1890        self.thread_states.continue_thread(thread_id);
1891        self.request(
1892            ContinueCommand {
1893                args: ContinueArguments {
1894                    thread_id: thread_id.0,
1895                    single_thread: Some(true),
1896                },
1897            },
1898            Self::on_step_response::<ContinueCommand>(thread_id),
1899            cx,
1900        )
1901        .detach();
1902    }
1903
1904    pub fn adapter_client(&self) -> Option<Arc<DebugAdapterClient>> {
1905        match self.mode {
1906            Mode::Running(ref local) => Some(local.client.clone()),
1907            Mode::Building => None,
1908        }
1909    }
1910
1911    pub fn has_ever_stopped(&self) -> bool {
1912        self.mode.has_ever_stopped()
1913    }
1914    pub fn step_over(
1915        &mut self,
1916        thread_id: ThreadId,
1917        granularity: SteppingGranularity,
1918        cx: &mut Context<Self>,
1919    ) {
1920        let supports_single_thread_execution_requests =
1921            self.capabilities.supports_single_thread_execution_requests;
1922        let supports_stepping_granularity = self
1923            .capabilities
1924            .supports_stepping_granularity
1925            .unwrap_or_default();
1926
1927        let command = NextCommand {
1928            inner: StepCommand {
1929                thread_id: thread_id.0,
1930                granularity: supports_stepping_granularity.then(|| granularity),
1931                single_thread: supports_single_thread_execution_requests,
1932            },
1933        };
1934
1935        self.thread_states.process_step(thread_id);
1936        self.request(
1937            command,
1938            Self::on_step_response::<NextCommand>(thread_id),
1939            cx,
1940        )
1941        .detach();
1942    }
1943
1944    pub fn step_in(
1945        &mut self,
1946        thread_id: ThreadId,
1947        granularity: SteppingGranularity,
1948        cx: &mut Context<Self>,
1949    ) {
1950        let supports_single_thread_execution_requests =
1951            self.capabilities.supports_single_thread_execution_requests;
1952        let supports_stepping_granularity = self
1953            .capabilities
1954            .supports_stepping_granularity
1955            .unwrap_or_default();
1956
1957        let command = StepInCommand {
1958            inner: StepCommand {
1959                thread_id: thread_id.0,
1960                granularity: supports_stepping_granularity.then(|| granularity),
1961                single_thread: supports_single_thread_execution_requests,
1962            },
1963        };
1964
1965        self.thread_states.process_step(thread_id);
1966        self.request(
1967            command,
1968            Self::on_step_response::<StepInCommand>(thread_id),
1969            cx,
1970        )
1971        .detach();
1972    }
1973
1974    pub fn step_out(
1975        &mut self,
1976        thread_id: ThreadId,
1977        granularity: SteppingGranularity,
1978        cx: &mut Context<Self>,
1979    ) {
1980        let supports_single_thread_execution_requests =
1981            self.capabilities.supports_single_thread_execution_requests;
1982        let supports_stepping_granularity = self
1983            .capabilities
1984            .supports_stepping_granularity
1985            .unwrap_or_default();
1986
1987        let command = StepOutCommand {
1988            inner: StepCommand {
1989                thread_id: thread_id.0,
1990                granularity: supports_stepping_granularity.then(|| granularity),
1991                single_thread: supports_single_thread_execution_requests,
1992            },
1993        };
1994
1995        self.thread_states.process_step(thread_id);
1996        self.request(
1997            command,
1998            Self::on_step_response::<StepOutCommand>(thread_id),
1999            cx,
2000        )
2001        .detach();
2002    }
2003
2004    pub fn step_back(
2005        &mut self,
2006        thread_id: ThreadId,
2007        granularity: SteppingGranularity,
2008        cx: &mut Context<Self>,
2009    ) {
2010        let supports_single_thread_execution_requests =
2011            self.capabilities.supports_single_thread_execution_requests;
2012        let supports_stepping_granularity = self
2013            .capabilities
2014            .supports_stepping_granularity
2015            .unwrap_or_default();
2016
2017        let command = StepBackCommand {
2018            inner: StepCommand {
2019                thread_id: thread_id.0,
2020                granularity: supports_stepping_granularity.then(|| granularity),
2021                single_thread: supports_single_thread_execution_requests,
2022            },
2023        };
2024
2025        self.thread_states.process_step(thread_id);
2026
2027        self.request(
2028            command,
2029            Self::on_step_response::<StepBackCommand>(thread_id),
2030            cx,
2031        )
2032        .detach();
2033    }
2034
2035    pub fn stack_frames(
2036        &mut self,
2037        thread_id: ThreadId,
2038        cx: &mut Context<Self>,
2039    ) -> Result<Vec<StackFrame>> {
2040        if self.thread_states.thread_status(thread_id) == ThreadStatus::Stopped
2041            && self.requests.contains_key(&ThreadsCommand.type_id())
2042            && self.threads.contains_key(&thread_id)
2043        // ^ todo(debugger): We need a better way to check that we're not querying stale data
2044        // We could still be using an old thread id and have sent a new thread's request
2045        // This isn't the biggest concern right now because it hasn't caused any issues outside of tests
2046        // But it very well could cause a minor bug in the future that is hard to track down
2047        {
2048            self.fetch(
2049                super::dap_command::StackTraceCommand {
2050                    thread_id: thread_id.0,
2051                    start_frame: None,
2052                    levels: None,
2053                },
2054                move |this, stack_frames, cx| {
2055                    let entry =
2056                        this.threads
2057                            .entry(thread_id)
2058                            .and_modify(|thread| match &stack_frames {
2059                                Ok(stack_frames) => {
2060                                    thread.stack_frames = stack_frames
2061                                        .iter()
2062                                        .cloned()
2063                                        .map(StackFrame::from)
2064                                        .collect();
2065                                    thread.stack_frames_error = None;
2066                                }
2067                                Err(error) => {
2068                                    thread.stack_frames.clear();
2069                                    thread.stack_frames_error = Some(error.cloned());
2070                                }
2071                            });
2072                    debug_assert!(
2073                        matches!(entry, indexmap::map::Entry::Occupied(_)),
2074                        "Sent request for thread_id that doesn't exist"
2075                    );
2076                    if let Ok(stack_frames) = stack_frames {
2077                        this.stack_frames.extend(
2078                            stack_frames
2079                                .into_iter()
2080                                .filter(|frame| {
2081                                    // Workaround for JavaScript debug adapter sending out "fake" stack frames for delineating await points. This is fine,
2082                                    // except that they always use an id of 0 for it, which collides with other (valid) stack frames.
2083                                    !(frame.id == 0
2084                                        && frame.line == 0
2085                                        && frame.column == 0
2086                                        && frame.presentation_hint
2087                                            == Some(StackFramePresentationHint::Label))
2088                                })
2089                                .map(|frame| (frame.id, StackFrame::from(frame))),
2090                        );
2091                    }
2092
2093                    this.invalidate_command_type::<ScopesCommand>();
2094                    this.invalidate_command_type::<VariablesCommand>();
2095
2096                    cx.emit(SessionEvent::StackTrace);
2097                },
2098                cx,
2099            );
2100        }
2101
2102        match self.threads.get(&thread_id) {
2103            Some(thread) => {
2104                if let Some(error) = &thread.stack_frames_error {
2105                    Err(error.cloned())
2106                } else {
2107                    Ok(thread.stack_frames.clone())
2108                }
2109            }
2110            None => Ok(Vec::new()),
2111        }
2112    }
2113
2114    pub fn scopes(&mut self, stack_frame_id: u64, cx: &mut Context<Self>) -> &[dap::Scope] {
2115        if self.requests.contains_key(&TypeId::of::<ThreadsCommand>())
2116            && self
2117                .requests
2118                .contains_key(&TypeId::of::<StackTraceCommand>())
2119        {
2120            self.fetch(
2121                ScopesCommand { stack_frame_id },
2122                move |this, scopes, cx| {
2123                    let Some(scopes) = scopes.log_err() else {
2124                        return
2125                    };
2126
2127                    for scope in scopes.iter() {
2128                        this.variables(scope.variables_reference, cx);
2129                    }
2130
2131                    let entry = this
2132                        .stack_frames
2133                        .entry(stack_frame_id)
2134                        .and_modify(|stack_frame| {
2135                            stack_frame.scopes = scopes;
2136                        });
2137
2138                    cx.emit(SessionEvent::Variables);
2139
2140                    debug_assert!(
2141                        matches!(entry, indexmap::map::Entry::Occupied(_)),
2142                        "Sent scopes request for stack_frame_id that doesn't exist or hasn't been fetched"
2143                    );
2144                },
2145                cx,
2146            );
2147        }
2148
2149        self.stack_frames
2150            .get(&stack_frame_id)
2151            .map(|frame| frame.scopes.as_slice())
2152            .unwrap_or_default()
2153    }
2154
2155    pub fn variables_by_stack_frame_id(&self, stack_frame_id: StackFrameId) -> Vec<dap::Variable> {
2156        let Some(stack_frame) = self.stack_frames.get(&stack_frame_id) else {
2157            return Vec::new();
2158        };
2159
2160        stack_frame
2161            .scopes
2162            .iter()
2163            .filter_map(|scope| self.variables.get(&scope.variables_reference))
2164            .flatten()
2165            .cloned()
2166            .collect()
2167    }
2168
2169    pub fn watchers(&self) -> &HashMap<SharedString, Watcher> {
2170        &self.watchers
2171    }
2172
2173    pub fn add_watcher(
2174        &mut self,
2175        expression: SharedString,
2176        frame_id: u64,
2177        cx: &mut Context<Self>,
2178    ) -> Task<Result<()>> {
2179        let request = self.mode.request_dap(EvaluateCommand {
2180            expression: expression.to_string(),
2181            context: Some(EvaluateArgumentsContext::Watch),
2182            frame_id: Some(frame_id),
2183            source: None,
2184        });
2185
2186        cx.spawn(async move |this, cx| {
2187            let response = request.await?;
2188
2189            this.update(cx, |session, cx| {
2190                session.watchers.insert(
2191                    expression.clone(),
2192                    Watcher {
2193                        expression,
2194                        value: response.result.into(),
2195                        variables_reference: response.variables_reference,
2196                        presentation_hint: response.presentation_hint,
2197                    },
2198                );
2199                cx.emit(SessionEvent::Watchers);
2200            })
2201        })
2202    }
2203
2204    pub fn refresh_watchers(&mut self, frame_id: u64, cx: &mut Context<Self>) {
2205        let watches = self.watchers.clone();
2206        for (_, watch) in watches.into_iter() {
2207            self.add_watcher(watch.expression.clone(), frame_id, cx)
2208                .detach();
2209        }
2210    }
2211
2212    pub fn remove_watcher(&mut self, expression: SharedString) {
2213        self.watchers.remove(&expression);
2214    }
2215
2216    pub fn variables(
2217        &mut self,
2218        variables_reference: VariableReference,
2219        cx: &mut Context<Self>,
2220    ) -> Vec<dap::Variable> {
2221        let command = VariablesCommand {
2222            variables_reference,
2223            filter: None,
2224            start: None,
2225            count: None,
2226            format: None,
2227        };
2228
2229        self.fetch(
2230            command,
2231            move |this, variables, cx| {
2232                let Some(variables) = variables.log_err() else {
2233                    return;
2234                };
2235
2236                this.variables.insert(variables_reference, variables);
2237
2238                cx.emit(SessionEvent::Variables);
2239                cx.emit(SessionEvent::InvalidateInlineValue);
2240            },
2241            cx,
2242        );
2243
2244        self.variables
2245            .get(&variables_reference)
2246            .cloned()
2247            .unwrap_or_default()
2248    }
2249
2250    pub fn set_variable_value(
2251        &mut self,
2252        stack_frame_id: u64,
2253        variables_reference: u64,
2254        name: String,
2255        value: String,
2256        cx: &mut Context<Self>,
2257    ) {
2258        if self.capabilities.supports_set_variable.unwrap_or_default() {
2259            self.request(
2260                SetVariableValueCommand {
2261                    name,
2262                    value,
2263                    variables_reference,
2264                },
2265                move |this, response, cx| {
2266                    let response = response.log_err()?;
2267                    this.invalidate_command_type::<VariablesCommand>();
2268                    this.refresh_watchers(stack_frame_id, cx);
2269                    cx.emit(SessionEvent::Variables);
2270                    Some(response)
2271                },
2272                cx,
2273            )
2274            .detach();
2275        }
2276    }
2277
2278    pub fn evaluate(
2279        &mut self,
2280        expression: String,
2281        context: Option<EvaluateArgumentsContext>,
2282        frame_id: Option<u64>,
2283        source: Option<Source>,
2284        cx: &mut Context<Self>,
2285    ) -> Task<()> {
2286        let event = dap::OutputEvent {
2287            category: None,
2288            output: format!("> {expression}"),
2289            group: None,
2290            variables_reference: None,
2291            source: None,
2292            line: None,
2293            column: None,
2294            data: None,
2295            location_reference: None,
2296        };
2297        self.push_output(event, cx);
2298        let request = self.mode.request_dap(EvaluateCommand {
2299            expression,
2300            context,
2301            frame_id,
2302            source,
2303        });
2304        cx.spawn(async move |this, cx| {
2305            let response = request.await;
2306            this.update(cx, |this, cx| {
2307                match response {
2308                    Ok(response) => {
2309                        let event = dap::OutputEvent {
2310                            category: None,
2311                            output: format!("< {}", &response.result),
2312                            group: None,
2313                            variables_reference: Some(response.variables_reference),
2314                            source: None,
2315                            line: None,
2316                            column: None,
2317                            data: None,
2318                            location_reference: None,
2319                        };
2320                        this.push_output(event, cx);
2321                    }
2322                    Err(e) => {
2323                        let event = dap::OutputEvent {
2324                            category: None,
2325                            output: format!("{}", e),
2326                            group: None,
2327                            variables_reference: None,
2328                            source: None,
2329                            line: None,
2330                            column: None,
2331                            data: None,
2332                            location_reference: None,
2333                        };
2334                        this.push_output(event, cx);
2335                    }
2336                };
2337                cx.notify();
2338            })
2339            .ok();
2340        })
2341    }
2342
2343    pub fn location(
2344        &mut self,
2345        reference: u64,
2346        cx: &mut Context<Self>,
2347    ) -> Option<dap::LocationsResponse> {
2348        self.fetch(
2349            LocationsCommand { reference },
2350            move |this, response, _| {
2351                let Some(response) = response.log_err() else {
2352                    return;
2353                };
2354                this.locations.insert(reference, response);
2355            },
2356            cx,
2357        );
2358        self.locations.get(&reference).cloned()
2359    }
2360
2361    pub fn is_attached(&self) -> bool {
2362        let Mode::Running(local_mode) = &self.mode else {
2363            return false;
2364        };
2365        local_mode.binary.request_args.request == StartDebuggingRequestArgumentsRequest::Attach
2366    }
2367
2368    pub fn disconnect_client(&mut self, cx: &mut Context<Self>) {
2369        let command = DisconnectCommand {
2370            restart: Some(false),
2371            terminate_debuggee: Some(false),
2372            suspend_debuggee: Some(false),
2373        };
2374
2375        self.request(command, Self::empty_response, cx).detach()
2376    }
2377
2378    pub fn terminate_threads(&mut self, thread_ids: Option<Vec<ThreadId>>, cx: &mut Context<Self>) {
2379        if self
2380            .capabilities
2381            .supports_terminate_threads_request
2382            .unwrap_or_default()
2383        {
2384            self.request(
2385                TerminateThreadsCommand {
2386                    thread_ids: thread_ids.map(|ids| ids.into_iter().map(|id| id.0).collect()),
2387                },
2388                Self::clear_active_debug_line_response,
2389                cx,
2390            )
2391            .detach();
2392        } else {
2393            self.shutdown(cx).detach();
2394        }
2395    }
2396
2397    pub fn thread_state(&self, thread_id: ThreadId) -> Option<ThreadStatus> {
2398        self.thread_states.thread_state(thread_id)
2399    }
2400}