Stop routing session events via the DAP store (#29588)

This cleans up a bunch of indirection and will make it easier to
show the session building state in the debugger terminal

Closes #ISSUE

Release Notes:

- N/A

crates/project/src/debugger/session.rs

@@ -14,14 +14,18 @@ use anyhow::{Context as _, Result, anyhow};
 use collections::{HashMap, HashSet, IndexMap, IndexSet};
 use dap::adapters::{DebugAdapterBinary, DebugTaskDefinition};
 use dap::messages::Response;
+use dap::requests::{Request, RunInTerminal, StartDebugging};
 use dap::{
     Capabilities, ContinueArguments, EvaluateArgumentsContext, Module, Source, StackFrameId,
     SteppingGranularity, StoppedEvent, VariableReference,
     client::{DebugAdapterClient, SessionId},
     messages::{Events, Message},
 };
-use dap::{ExceptionBreakpointsFilter, ExceptionFilterOptions, OutputEventCategory};
-use futures::channel::oneshot;
+use dap::{
+    ExceptionBreakpointsFilter, ExceptionFilterOptions, OutputEventCategory,
+    RunInTerminalRequestArguments, StartDebuggingRequestArguments,
+};
+use futures::channel::{mpsc, oneshot};
 use futures::{FutureExt, future::Shared};
 use gpui::{
     App, AppContext, AsyncApp, BackgroundExecutor, Context, Entity, EventEmitter, SharedString,
@@ -522,7 +526,6 @@ pub struct Session {
     is_session_terminated: bool,
     requests: HashMap<TypeId, HashMap<RequestSlot, Shared<Task<Option<()>>>>>,
     exception_breakpoints: BTreeMap<String, (ExceptionBreakpointsFilter, IsEnabled)>,
-    start_debugging_requests_tx: futures::channel::mpsc::UnboundedSender<(SessionId, Message)>,
     background_tasks: Vec<Task<()>>,
 }
 
@@ -608,13 +611,20 @@ pub enum SessionEvent {
     Threads,
     InvalidateInlineValue,
     CapabilitiesLoaded,
+    RunInTerminal {
+        request: RunInTerminalRequestArguments,
+        sender: mpsc::Sender<Result<u32>>,
+    },
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub enum SessionStateEvent {
     Running,
     Shutdown,
     Restart,
+    SpawnChildSession {
+        request: StartDebuggingRequestArguments,
+    },
 }
 
 impl EventEmitter<SessionEvent> for Session {}
@@ -629,7 +639,6 @@ impl Session {
         session_id: SessionId,
         parent_session: Option<Entity<Session>>,
         template: DebugTaskDefinition,
-        start_debugging_requests_tx: futures::channel::mpsc::UnboundedSender<(SessionId, Message)>,
         cx: &mut App,
     ) -> Entity<Self> {
         cx.new::<Self>(|cx| {
@@ -678,7 +687,6 @@ impl Session {
                 is_session_terminated: false,
                 exception_breakpoints: Default::default(),
                 definition: template,
-                start_debugging_requests_tx,
             };
 
             this
@@ -702,7 +710,6 @@ impl Session {
     ) -> Task<Result<()>> {
         let (message_tx, mut message_rx) = futures::channel::mpsc::unbounded();
         let (initialized_tx, initialized_rx) = futures::channel::oneshot::channel();
-        let session_id = self.session_id();
 
         let background_tasks = vec![cx.spawn(async move |this: WeakEntity<Session>, cx| {
             let mut initialized_tx = Some(initialized_tx);
@@ -719,10 +726,15 @@ impl Session {
                             break;
                         };
                     }
-                } else {
-                    let Ok(Ok(_)) = this.update(cx, |this, _| {
-                        this.start_debugging_requests_tx
-                            .unbounded_send((session_id, message))
+                } else if let Message::Request(request) = message {
+                    let Ok(_) = this.update(cx, |this, cx| {
+                        if request.command == StartDebugging::COMMAND {
+                            this.handle_start_debugging_request(request, cx)
+                                .detach_and_log_err(cx);
+                        } else if request.command == RunInTerminal::COMMAND {
+                            this.handle_run_in_terminal_request(request, cx)
+                                .detach_and_log_err(cx);
+                        }
                     }) else {
                         break;
                     };
@@ -830,6 +842,109 @@ impl Session {
         }
     }
 
+    fn handle_start_debugging_request(
+        &mut self,
+        request: dap::messages::Request,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        let request_seq = request.seq;
+
+        let launch_request: Option<Result<StartDebuggingRequestArguments, _>> = request
+            .arguments
+            .as_ref()
+            .map(|value| serde_json::from_value(value.clone()));
+
+        let mut success = true;
+        if let Some(Ok(request)) = launch_request {
+            cx.emit(SessionStateEvent::SpawnChildSession { request });
+        } else {
+            log::error!(
+                "Failed to parse launch request arguments: {:?}",
+                request.arguments
+            );
+            success = false;
+        }
+
+        cx.spawn(async move |this, cx| {
+            this.update(cx, |this, cx| {
+                this.respond_to_client(
+                    request_seq,
+                    success,
+                    StartDebugging::COMMAND.to_string(),
+                    None,
+                    cx,
+                )
+            })?
+            .await
+        })
+    }
+
+    fn handle_run_in_terminal_request(
+        &mut self,
+        request: dap::messages::Request,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<()>> {
+        let request_args = serde_json::from_value::<RunInTerminalRequestArguments>(
+            request.arguments.unwrap_or_default(),
+        )
+        .expect("To parse StartDebuggingRequestArguments");
+
+        let seq = request.seq;
+
+        let (tx, mut rx) = mpsc::channel::<Result<u32>>(1);
+        cx.emit(SessionEvent::RunInTerminal {
+            request: request_args,
+            sender: tx,
+        });
+        cx.notify();
+
+        cx.spawn(async move |session, cx| {
+            let result = util::maybe!(async move {
+                rx.next().await.ok_or_else(|| {
+                    anyhow!("failed to receive response from spawn terminal".to_string())
+                })?
+            })
+            .await;
+            let (success, body) = match result {
+                Ok(pid) => (
+                    true,
+                    serde_json::to_value(dap::RunInTerminalResponse {
+                        process_id: None,
+                        shell_process_id: Some(pid as u64),
+                    })
+                    .ok(),
+                ),
+                Err(error) => (
+                    false,
+                    serde_json::to_value(dap::ErrorResponse {
+                        error: Some(dap::Message {
+                            id: seq,
+                            format: error.to_string(),
+                            variables: None,
+                            send_telemetry: None,
+                            show_user: None,
+                            url: None,
+                            url_label: None,
+                        }),
+                    })
+                    .ok(),
+                ),
+            };
+
+            session
+                .update(cx, |session, cx| {
+                    session.respond_to_client(
+                        seq,
+                        success,
+                        RunInTerminal::COMMAND.to_string(),
+                        body,
+                        cx,
+                    )
+                })?
+                .await
+        })
+    }
+
     pub(super) fn request_initialize(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
         let adapter_id = String::from(self.definition.adapter.clone());
         let request = Initialize { adapter_id };