session.rs

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