debugger: Open debugger panel on session startup (#29186)

Now all debug sessions are routed through the debug panel and are
started synchronously instead of by a task that returns a session once
the initialization process is finished. A session is `Mode::Booting`
while it's starting the debug adapter process and then transitions to
`Mode::Running` once this is completed.

This PR also added new tests for the dap logger, reverse start debugging
request, and debugging over SSH.

Release Notes:

- N/A

---------

Co-authored-by: Anthony Eid <hello@anthonyeid.me>
Co-authored-by: Anthony <anthony@zed.dev>
Co-authored-by: Cole Miller <m@cole-miller.net>
Co-authored-by: Cole Miller <cole@zed.dev>
Co-authored-by: Zed AI <ai@zed.dev>
Co-authored-by: Remco Smits <djsmits12@gmail.com>

crates/project/src/debugger/session.rs

@@ -28,7 +28,6 @@ use gpui::{
     Task, WeakEntity,
 };
 
-use rpc::AnyProtoClient;
 use serde_json::{Value, json};
 use smol::stream::StreamExt;
 use std::any::TypeId;
@@ -115,54 +114,14 @@ impl From<dap::Thread> for Thread {
     }
 }
 
-type UpstreamProjectId = u64;
-
-struct RemoteConnection {
-    _client: AnyProtoClient,
-    _upstream_project_id: UpstreamProjectId,
-    _adapter_name: SharedString,
-}
-
-impl RemoteConnection {
-    fn send_proto_client_request<R: DapCommand>(
-        &self,
-        _request: R,
-        _session_id: SessionId,
-        cx: &mut App,
-    ) -> Task<Result<R::Response>> {
-        // let message = request.to_proto(session_id, self.upstream_project_id);
-        // let upstream_client = self.client.clone();
-        cx.background_executor().spawn(async move {
-            // debugger(todo): Properly send messages when we wrap dap_commands in envelopes again
-            // let response = upstream_client.request(message).await?;
-            // request.response_from_proto(response)
-            Err(anyhow!("Sending dap commands over RPC isn't supported yet"))
-        })
-    }
-
-    fn request<R: DapCommand>(
-        &self,
-        request: R,
-        session_id: SessionId,
-        cx: &mut App,
-    ) -> Task<Result<R::Response>>
-    where
-        <R::DapRequest as dap::requests::Request>::Response: 'static,
-        <R::DapRequest as dap::requests::Request>::Arguments: 'static + Send,
-    {
-        return self.send_proto_client_request::<R>(request, session_id, cx);
-    }
-}
-
 enum Mode {
-    Local(LocalMode),
-    Remote(RemoteConnection),
+    Building,
+    Running(LocalMode),
 }
 
 #[derive(Clone)]
 pub struct LocalMode {
     client: Arc<DebugAdapterClient>,
-    definition: DebugTaskDefinition,
     binary: DebugAdapterBinary,
     root_binary: Option<Arc<DebugAdapterBinary>>,
     pub(crate) breakpoint_store: Entity<BreakpointStore>,
@@ -186,56 +145,47 @@ fn client_source(abs_path: &Path) -> dap::Source {
 }
 
 impl LocalMode {
-    fn new(
+    async fn new(
         session_id: SessionId,
         parent_session: Option<Entity<Session>>,
         worktree: WeakEntity<Worktree>,
         breakpoint_store: Entity<BreakpointStore>,
-        config: DebugTaskDefinition,
         binary: DebugAdapterBinary,
         messages_tx: futures::channel::mpsc::UnboundedSender<Message>,
         cx: AsyncApp,
-    ) -> Task<Result<Self>> {
-        cx.spawn(async move |cx| {
-            let message_handler = Box::new(move |message| {
-                messages_tx.unbounded_send(message).ok();
-            });
+    ) -> Result<Self> {
+        let message_handler = Box::new(move |message| {
+            messages_tx.unbounded_send(message).ok();
+        });
 
-            let root_binary = if let Some(parent_session) = parent_session.as_ref() {
-                Some(parent_session.read_with(cx, |session, _| session.root_binary().clone())?)
+        let root_binary = if let Some(parent_session) = parent_session.as_ref() {
+            Some(parent_session.read_with(&cx, |session, _| session.root_binary().clone())?)
+        } else {
+            None
+        };
+
+        let client = Arc::new(
+            if let Some(client) = parent_session
+                .and_then(|session| cx.update(|cx| session.read(cx).adapter_client()).ok())
+                .flatten()
+            {
+                client
+                    .reconnect(session_id, binary.clone(), message_handler, cx.clone())
+                    .await?
             } else {
-                None
-            };
-
-            let client = Arc::new(
-                if let Some(client) = parent_session
-                    .and_then(|session| cx.update(|cx| session.read(cx).adapter_client()).ok())
-                    .flatten()
-                {
-                    client
-                        .reconnect(session_id, binary.clone(), message_handler, cx.clone())
-                        .await?
-                } else {
-                    DebugAdapterClient::start(
-                        session_id,
-                        binary.clone(),
-                        message_handler,
-                        cx.clone(),
-                    )
+                DebugAdapterClient::start(session_id, binary.clone(), message_handler, cx.clone())
                     .await
                     .with_context(|| "Failed to start communication with debug adapter")?
-                },
-            );
+            },
+        );
 
-            Ok(Self {
-                client,
-                breakpoint_store,
-                worktree,
-                tmp_breakpoint: None,
-                definition: config,
-                root_binary,
-                binary,
-            })
+        Ok(Self {
+            client,
+            breakpoint_store,
+            worktree,
+            tmp_breakpoint: None,
+            root_binary,
+            binary,
         })
     }
 
@@ -371,19 +321,10 @@ impl LocalMode {
         })
     }
 
-    pub fn label(&self) -> String {
-        self.definition.label.clone()
-    }
-
-    fn request_initialization(&self, cx: &App) -> Task<Result<Capabilities>> {
-        let adapter_id = self.definition.adapter.clone();
-
-        self.request(Initialize { adapter_id }, cx.background_executor().clone())
-    }
-
     fn initialize_sequence(
         &self,
         capabilities: &Capabilities,
+        definition: &DebugTaskDefinition,
         initialized_rx: oneshot::Receiver<()>,
         dap_store: WeakEntity<DapStore>,
         cx: &App,
@@ -391,7 +332,7 @@ impl LocalMode {
         let mut raw = self.binary.request_args.clone();
 
         merge_json_value_into(
-            self.definition.initialize_args.clone().unwrap_or(json!({})),
+            definition.initialize_args.clone().unwrap_or(json!({})),
             &mut raw.configuration,
         );
 
@@ -426,9 +367,9 @@ impl LocalMode {
         let supports_exception_filters = capabilities
             .supports_exception_filter_options
             .unwrap_or_default();
+        let this = self.clone();
+        let worktree = self.worktree().clone();
         let configuration_sequence = cx.spawn({
-            let this = self.clone();
-            let worktree = self.worktree().clone();
             async move |cx| {
                 initialized_rx.await?;
                 let errors_by_path = cx
@@ -511,16 +452,10 @@ impl LocalMode {
         })
     }
 }
-impl From<RemoteConnection> for Mode {
-    fn from(value: RemoteConnection) -> Self {
-        Self::Remote(value)
-    }
-}
 
 impl Mode {
     fn request_dap<R: DapCommand>(
         &self,
-        session_id: SessionId,
         request: R,
         cx: &mut Context<Session>,
     ) -> Task<Result<R::Response>>
@@ -529,10 +464,13 @@ impl Mode {
         <R::DapRequest as dap::requests::Request>::Arguments: 'static + Send,
     {
         match self {
-            Mode::Local(debug_adapter_client) => {
+            Mode::Running(debug_adapter_client) => {
                 debug_adapter_client.request(request, cx.background_executor().clone())
             }
-            Mode::Remote(remote_connection) => remote_connection.request(request, session_id, cx),
+            Mode::Building => Task::ready(Err(anyhow!(
+                "no adapter running to send request: {:?}",
+                request
+            ))),
         }
     }
 }
@@ -609,10 +547,11 @@ pub struct OutputToken(pub usize);
 /// Represents a current state of a single debug adapter and provides ways to mutate it.
 pub struct Session {
     mode: Mode,
+    definition: DebugTaskDefinition,
     pub(super) capabilities: Capabilities,
     id: SessionId,
     child_session_ids: HashSet<SessionId>,
-    parent_id: Option<SessionId>,
+    parent_session: Option<Entity<Session>>,
     ignore_breakpoints: bool,
     modules: Vec<dap::Module>,
     loaded_sources: Vec<dap::Source>,
@@ -626,7 +565,8 @@ pub struct Session {
     is_session_terminated: bool,
     requests: HashMap<TypeId, HashMap<RequestSlot, Shared<Task<Option<()>>>>>,
     exception_breakpoints: BTreeMap<String, (ExceptionBreakpointsFilter, IsEnabled)>,
-    _background_tasks: Vec<Task<()>>,
+    start_debugging_requests_tx: futures::channel::mpsc::UnboundedSender<(SessionId, Message)>,
+    background_tasks: Vec<Task<()>>,
 }
 
 trait CacheableCommand: Any + Send + Sync {
@@ -708,9 +648,12 @@ pub enum SessionEvent {
     StackTrace,
     Variables,
     Threads,
+    CapabilitiesLoaded,
 }
 
-pub(super) enum SessionStateEvent {
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum SessionStateEvent {
+    Running,
     Shutdown,
     Restart,
 }
@@ -722,80 +665,140 @@ impl EventEmitter<SessionStateEvent> for Session {}
 // remote side will only send breakpoint updates when it is a breakpoint created by that peer
 // BreakpointStore notifies session on breakpoint changes
 impl Session {
-    pub(crate) fn local(
+    pub(crate) fn new(
         breakpoint_store: Entity<BreakpointStore>,
-        worktree: WeakEntity<Worktree>,
         session_id: SessionId,
         parent_session: Option<Entity<Session>>,
-        binary: DebugAdapterBinary,
-        config: DebugTaskDefinition,
+        template: DebugTaskDefinition,
         start_debugging_requests_tx: futures::channel::mpsc::UnboundedSender<(SessionId, Message)>,
-        initialized_tx: oneshot::Sender<()>,
         cx: &mut App,
-    ) -> Task<Result<Entity<Self>>> {
-        let (message_tx, message_rx) = futures::channel::mpsc::unbounded();
+    ) -> Entity<Self> {
+        cx.new::<Self>(|cx| {
+            cx.subscribe(&breakpoint_store, |this, _, event, cx| match event {
+                BreakpointStoreEvent::BreakpointsUpdated(path, reason) => {
+                    if let Some(local) = (!this.ignore_breakpoints)
+                        .then(|| this.as_local_mut())
+                        .flatten()
+                    {
+                        local
+                            .send_breakpoints_from_path(path.clone(), *reason, cx)
+                            .detach();
+                    };
+                }
+                BreakpointStoreEvent::BreakpointsCleared(paths) => {
+                    if let Some(local) = (!this.ignore_breakpoints)
+                        .then(|| this.as_local_mut())
+                        .flatten()
+                    {
+                        local.unset_breakpoints_from_paths(paths, cx).detach();
+                    }
+                }
+                BreakpointStoreEvent::ActiveDebugLineChanged => {}
+            })
+            .detach();
 
-        cx.spawn(async move |cx| {
+            let this = Self {
+                mode: Mode::Building,
+                id: session_id,
+                child_session_ids: HashSet::default(),
+                parent_session,
+                capabilities: Capabilities::default(),
+                ignore_breakpoints: false,
+                variables: Default::default(),
+                stack_frames: Default::default(),
+                thread_states: ThreadStates::default(),
+                output_token: OutputToken(0),
+                output: circular_buffer::CircularBuffer::boxed(),
+                requests: HashMap::default(),
+                modules: Vec::default(),
+                loaded_sources: Vec::default(),
+                threads: IndexMap::default(),
+                background_tasks: Vec::default(),
+                locations: Default::default(),
+                is_session_terminated: false,
+                exception_breakpoints: Default::default(),
+                definition: template,
+                start_debugging_requests_tx,
+            };
+
+            this
+        })
+    }
+
+    pub fn worktree(&self) -> Option<Entity<Worktree>> {
+        match &self.mode {
+            Mode::Building => None,
+            Mode::Running(local_mode) => local_mode.worktree.upgrade(),
+        }
+    }
+
+    pub fn boot(
+        &mut self,
+        binary: DebugAdapterBinary,
+        worktree: Entity<Worktree>,
+        breakpoint_store: Entity<BreakpointStore>,
+        dap_store: WeakEntity<DapStore>,
+        cx: &mut Context<Self>,
+    ) -> 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);
+            while let Some(message) = message_rx.next().await {
+                if let Message::Event(event) = message {
+                    if let Events::Initialized(_) = *event {
+                        if let Some(tx) = initialized_tx.take() {
+                            tx.send(()).ok();
+                        }
+                    } else {
+                        let Ok(_) = this.update(cx, |session, cx| {
+                            session.handle_dap_event(event, cx);
+                        }) else {
+                            break;
+                        };
+                    }
+                } else {
+                    let Ok(Ok(_)) = this.update(cx, |this, _| {
+                        this.start_debugging_requests_tx
+                            .unbounded_send((session_id, message))
+                    }) else {
+                        break;
+                    };
+                }
+            }
+        })];
+        self.background_tasks = background_tasks;
+        let id = self.id;
+        let parent_session = self.parent_session.clone();
+
+        cx.spawn(async move |this, cx| {
             let mode = LocalMode::new(
-                session_id,
-                parent_session.clone(),
-                worktree,
+                id,
+                parent_session,
+                worktree.downgrade(),
                 breakpoint_store.clone(),
-                config.clone(),
                 binary,
                 message_tx,
                 cx.clone(),
             )
             .await?;
+            this.update(cx, |this, cx| {
+                this.mode = Mode::Running(mode);
+                cx.emit(SessionStateEvent::Running);
+            })?;
 
-            cx.new(|cx| {
-                create_local_session(
-                    breakpoint_store,
-                    session_id,
-                    parent_session,
-                    start_debugging_requests_tx,
-                    initialized_tx,
-                    message_rx,
-                    mode,
-                    cx,
-                )
-            })
+            this.update(cx, |session, cx| session.request_initialize(cx))?
+                .await?;
+
+            this.update(cx, |session, cx| {
+                session.initialize_sequence(initialized_rx, dap_store.clone(), cx)
+            })?
+            .await
         })
     }
 
-    pub(crate) fn remote(
-        session_id: SessionId,
-        client: AnyProtoClient,
-        upstream_project_id: u64,
-        ignore_breakpoints: bool,
-    ) -> Self {
-        Self {
-            mode: Mode::Remote(RemoteConnection {
-                _adapter_name: SharedString::new(""), // todo(debugger) we need to pipe in the right values to deserialize the debugger pane layout
-                _client: client,
-                _upstream_project_id: upstream_project_id,
-            }),
-            id: session_id,
-            child_session_ids: HashSet::default(),
-            parent_id: None,
-            capabilities: Capabilities::default(),
-            ignore_breakpoints,
-            variables: Default::default(),
-            stack_frames: Default::default(),
-            thread_states: ThreadStates::default(),
-            output_token: OutputToken(0),
-            output: circular_buffer::CircularBuffer::boxed(),
-            requests: HashMap::default(),
-            modules: Vec::default(),
-            loaded_sources: Vec::default(),
-            threads: IndexMap::default(),
-            _background_tasks: Vec::default(),
-            locations: Default::default(),
-            is_session_terminated: false,
-            exception_breakpoints: Default::default(),
-        }
-    }
-
     pub fn session_id(&self) -> SessionId {
         self.id
     }
@@ -812,8 +815,14 @@ impl Session {
         self.child_session_ids.remove(&session_id);
     }
 
-    pub fn parent_id(&self) -> Option<SessionId> {
-        self.parent_id
+    pub fn parent_id(&self, cx: &App) -> Option<SessionId> {
+        self.parent_session
+            .as_ref()
+            .map(|session| session.read(cx).id)
+    }
+
+    pub fn parent_session(&self) -> Option<&Entity<Self>> {
+        self.parent_session.as_ref()
     }
 
     pub fn capabilities(&self) -> &Capabilities {
@@ -821,35 +830,35 @@ impl Session {
     }
 
     pub(crate) fn root_binary(&self) -> Arc<DebugAdapterBinary> {
-        let Mode::Local(local_mode) = &self.mode else {
-            panic!("Session is not local");
-        };
-        local_mode
-            .root_binary
-            .clone()
-            .unwrap_or_else(|| Arc::new(local_mode.binary.clone()))
+        match &self.mode {
+            Mode::Building => {
+                // todo(debugger): Implement root_binary for building mode
+                unimplemented!()
+            }
+            Mode::Running(running) => running
+                .root_binary
+                .clone()
+                .unwrap_or_else(|| Arc::new(running.binary.clone())),
+        }
     }
 
     pub fn binary(&self) -> &DebugAdapterBinary {
-        let Mode::Local(local_mode) = &self.mode else {
+        let Mode::Running(local_mode) = &self.mode else {
             panic!("Session is not local");
         };
         &local_mode.binary
     }
 
     pub fn adapter_name(&self) -> SharedString {
-        match &self.mode {
-            Mode::Local(local_mode) => local_mode.definition.adapter.clone().into(),
-            Mode::Remote(remote_mode) => remote_mode._adapter_name.clone(),
-        }
+        self.definition.adapter.clone().into()
     }
 
-    pub fn configuration(&self) -> Option<DebugTaskDefinition> {
-        if let Mode::Local(local_mode) = &self.mode {
-            Some(local_mode.definition.clone())
-        } else {
-            None
-        }
+    pub fn label(&self) -> String {
+        self.definition.label.clone()
+    }
+
+    pub fn definition(&self) -> DebugTaskDefinition {
+        self.definition.clone()
     }
 
     pub fn is_terminated(&self) -> bool {
@@ -857,31 +866,33 @@ impl Session {
     }
 
     pub fn is_local(&self) -> bool {
-        matches!(self.mode, Mode::Local(_))
+        matches!(self.mode, Mode::Running(_))
     }
 
     pub fn as_local_mut(&mut self) -> Option<&mut LocalMode> {
         match &mut self.mode {
-            Mode::Local(local_mode) => Some(local_mode),
-            Mode::Remote(_) => None,
+            Mode::Running(local_mode) => Some(local_mode),
+            Mode::Building => None,
         }
     }
 
     pub fn as_local(&self) -> Option<&LocalMode> {
         match &self.mode {
-            Mode::Local(local_mode) => Some(local_mode),
-            Mode::Remote(_) => None,
+            Mode::Running(local_mode) => Some(local_mode),
+            Mode::Building => None,
         }
     }
 
     pub(super) fn request_initialize(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
+        let adapter_id = self.definition.adapter.clone();
+        let request = Initialize { adapter_id };
         match &self.mode {
-            Mode::Local(local_mode) => {
-                let capabilities = local_mode.clone().request_initialization(cx);
+            Mode::Running(local_mode) => {
+                let capabilities = local_mode.request(request, cx.background_executor().clone());
 
                 cx.spawn(async move |this, cx| {
                     let capabilities = capabilities.await?;
-                    this.update(cx, |session, _| {
+                    this.update(cx, |session, cx| {
                         session.capabilities = capabilities;
                         let filters = session
                             .capabilities
@@ -895,12 +906,13 @@ impl Session {
                                 .entry(filter.filter.clone())
                                 .or_insert_with(|| (filter, default));
                         }
+                        cx.emit(SessionEvent::CapabilitiesLoaded);
                     })?;
                     Ok(())
                 })
             }
-            Mode::Remote(_) => Task::ready(Err(anyhow!(
-                "Cannot send initialize request from remote session"
+            Mode::Building => Task::ready(Err(anyhow!(
+                "Cannot send initialize request, task still building"
             ))),
         }
     }
@@ -912,10 +924,14 @@ impl Session {
         cx: &mut Context<Self>,
     ) -> Task<Result<()>> {
         match &self.mode {
-            Mode::Local(local_mode) => {
-                local_mode.initialize_sequence(&self.capabilities, initialize_rx, dap_store, cx)
-            }
-            Mode::Remote(_) => Task::ready(Err(anyhow!("cannot initialize remote session"))),
+            Mode::Running(local_mode) => local_mode.initialize_sequence(
+                &self.capabilities,
+                &self.definition,
+                initialize_rx,
+                dap_store,
+                cx,
+            ),
+            Mode::Building => Task::ready(Err(anyhow!("cannot initialize, still building"))),
         }
     }
 
@@ -926,7 +942,7 @@ impl Session {
         cx: &mut Context<Self>,
     ) {
         match &mut self.mode {
-            Mode::Local(local_mode) => {
+            Mode::Running(local_mode) => {
                 if !matches!(
                     self.thread_states.thread_state(active_thread_id),
                     Some(ThreadStatus::Stopped)
@@ -949,7 +965,7 @@ impl Session {
                 })
                 .detach();
             }
-            Mode::Remote(_) => {}
+            Mode::Building => {}
         }
     }
 
@@ -983,13 +999,13 @@ impl Session {
         body: Option<serde_json::Value>,
         cx: &mut Context<Self>,
     ) -> Task<Result<()>> {
-        let Some(local_session) = self.as_local().cloned() else {
+        let Some(local_session) = self.as_local() else {
             unreachable!("Cannot respond to remote client");
         };
+        let client = local_session.client.clone();
 
         cx.background_spawn(async move {
-            local_session
-                .client
+            client
                 .send_message(Message::Response(Response {
                     body,
                     success,
@@ -1178,7 +1194,6 @@ impl Session {
 
             let task = Self::request_inner::<Arc<T>>(
                 &self.capabilities,
-                self.id,
                 &self.mode,
                 command,
                 process_result,
@@ -1199,7 +1214,6 @@ impl Session {
 
     fn request_inner<T: DapCommand + PartialEq + Eq + Hash>(
         capabilities: &Capabilities,
-        session_id: SessionId,
         mode: &Mode,
         request: T,
         process_result: impl FnOnce(
@@ -1225,7 +1239,7 @@ impl Session {
             });
         }
 
-        let request = mode.request_dap(session_id, request, cx);
+        let request = mode.request_dap(request, cx);
         cx.spawn(async move |this, cx| {
             let result = request.await;
             this.update(cx, |this, cx| process_result(this, result, cx))
@@ -1245,14 +1259,7 @@ impl Session {
         + 'static,
         cx: &mut Context<Self>,
     ) -> Task<Option<T::Response>> {
-        Self::request_inner(
-            &self.capabilities,
-            self.id,
-            &self.mode,
-            request,
-            process_result,
-            cx,
-        )
+        Self::request_inner(&self.capabilities, &self.mode, request, process_result, cx)
     }
 
     fn invalidate_command_type<Command: DapCommand>(&mut self) {
@@ -1569,8 +1576,8 @@ impl Session {
 
     pub fn adapter_client(&self) -> Option<Arc<DebugAdapterClient>> {
         match self.mode {
-            Mode::Local(ref local) => Some(local.client.clone()),
-            Mode::Remote(_) => None,
+            Mode::Running(ref local) => Some(local.client.clone()),
+            Mode::Building => None,
         }
     }
 
@@ -1936,83 +1943,3 @@ impl Session {
         }
     }
 }
-
-fn create_local_session(
-    breakpoint_store: Entity<BreakpointStore>,
-    session_id: SessionId,
-    parent_session: Option<Entity<Session>>,
-    start_debugging_requests_tx: futures::channel::mpsc::UnboundedSender<(SessionId, Message)>,
-    initialized_tx: oneshot::Sender<()>,
-    mut message_rx: futures::channel::mpsc::UnboundedReceiver<Message>,
-    mode: LocalMode,
-    cx: &mut Context<Session>,
-) -> Session {
-    let _background_tasks = vec![cx.spawn(async move |this: WeakEntity<Session>, cx| {
-        let mut initialized_tx = Some(initialized_tx);
-        while let Some(message) = message_rx.next().await {
-            if let Message::Event(event) = message {
-                if let Events::Initialized(_) = *event {
-                    if let Some(tx) = initialized_tx.take() {
-                        tx.send(()).ok();
-                    }
-                } else {
-                    let Ok(_) = this.update(cx, |session, cx| {
-                        session.handle_dap_event(event, cx);
-                    }) else {
-                        break;
-                    };
-                }
-            } else {
-                let Ok(_) = start_debugging_requests_tx.unbounded_send((session_id, message))
-                else {
-                    break;
-                };
-            }
-        }
-    })];
-
-    cx.subscribe(&breakpoint_store, |this, _, event, cx| match event {
-        BreakpointStoreEvent::BreakpointsUpdated(path, reason) => {
-            if let Some(local) = (!this.ignore_breakpoints)
-                .then(|| this.as_local_mut())
-                .flatten()
-            {
-                local
-                    .send_breakpoints_from_path(path.clone(), *reason, cx)
-                    .detach();
-            };
-        }
-        BreakpointStoreEvent::BreakpointsCleared(paths) => {
-            if let Some(local) = (!this.ignore_breakpoints)
-                .then(|| this.as_local_mut())
-                .flatten()
-            {
-                local.unset_breakpoints_from_paths(paths, cx).detach();
-            }
-        }
-        BreakpointStoreEvent::ActiveDebugLineChanged => {}
-    })
-    .detach();
-
-    Session {
-        mode: Mode::Local(mode),
-        id: session_id,
-        child_session_ids: HashSet::default(),
-        parent_id: parent_session.map(|session| session.read(cx).id),
-        variables: Default::default(),
-        capabilities: Capabilities::default(),
-        thread_states: ThreadStates::default(),
-        output_token: OutputToken(0),
-        ignore_breakpoints: false,
-        output: circular_buffer::CircularBuffer::boxed(),
-        requests: HashMap::default(),
-        modules: Vec::default(),
-        loaded_sources: Vec::default(),
-        threads: IndexMap::default(),
-        stack_frames: IndexMap::default(),
-        locations: Default::default(),
-        exception_breakpoints: Default::default(),
-        _background_tasks,
-        is_session_terminated: false,
-    }
-}