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