debugger: Fix connections over SSH (#32834)

Before this change, we would see "connection reset" when sending the
initialize
request over SSH in the case that the debug adapter was slow to boot.

(Although we'd have successfully created a connection to the local SSH
port,
trying to read/write from it would not work until the remote end of the
connection had been established)

Fixes  #32575

Release Notes:

- debugger: Fix connecting to a Python debugger over SSH

crates/project/src/debugger/session.rs

@@ -29,6 +29,7 @@ use dap::{
     StartDebuggingRequestArgumentsRequest,
 };
 use futures::SinkExt;
+use futures::channel::mpsc::UnboundedSender;
 use futures::channel::{mpsc, oneshot};
 use futures::{FutureExt, future::Shared};
 use gpui::{
@@ -139,6 +140,7 @@ pub struct RunningMode {
     executor: BackgroundExecutor,
     is_started: bool,
     has_ever_stopped: bool,
+    messages_tx: UnboundedSender<Message>,
 }
 
 fn client_source(abs_path: &Path) -> dap::Source {
@@ -163,34 +165,35 @@ impl RunningMode {
         worktree: WeakEntity<Worktree>,
         binary: DebugAdapterBinary,
         messages_tx: futures::channel::mpsc::UnboundedSender<Message>,
-        cx: AsyncApp,
+        cx: &mut AsyncApp,
     ) -> Result<Self> {
-        let message_handler = Box::new(move |message| {
-            messages_tx.unbounded_send(message).ok();
+        let message_handler = Box::new({
+            let messages_tx = messages_tx.clone();
+            move |message| {
+                messages_tx.unbounded_send(message).ok();
+            }
         });
 
-        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())
-                    .await?
-            },
-        );
+        let client = if let Some(client) = parent_session
+            .and_then(|session| cx.update(|cx| session.read(cx).adapter_client()).ok())
+            .flatten()
+        {
+            client
+                .create_child_connection(session_id, binary.clone(), message_handler, cx)
+                .await?
+        } else {
+            DebugAdapterClient::start(session_id, binary.clone(), message_handler, cx).await?
+        };
 
         Ok(Self {
-            client,
+            client: Arc::new(client),
             worktree,
             tmp_breakpoint: None,
             binary,
             executor: cx.background_executor().clone(),
             is_started: false,
             has_ever_stopped: false,
+            messages_tx,
         })
     }
 
@@ -481,6 +484,22 @@ impl RunningMode {
         })
     }
 
+    fn reconnect_for_ssh(&self, cx: &mut AsyncApp) -> Option<Task<Result<()>>> {
+        let client = self.client.clone();
+        let messages_tx = self.messages_tx.clone();
+        let message_handler = Box::new(move |message| {
+            messages_tx.unbounded_send(message).ok();
+        });
+        if client.should_reconnect_for_ssh() {
+            Some(cx.spawn(async move |cx| {
+                client.connect(message_handler, cx).await?;
+                anyhow::Ok(())
+            }))
+        } else {
+            None
+        }
+    }
+
     fn request<R: LocalDapCommand>(&self, request: R) -> Task<Result<R::Response>>
     where
         <R::DapRequest as dap::requests::Request>::Response: 'static,
@@ -855,7 +874,7 @@ impl Session {
                 worktree.downgrade(),
                 binary.clone(),
                 message_tx,
-                cx.clone(),
+                cx,
             )
             .await?;
             this.update(cx, |this, cx| {
@@ -1131,35 +1150,58 @@ impl Session {
     pub(super) fn request_initialize(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
         let adapter_id = self.adapter().to_string();
         let request = Initialize { adapter_id };
-        match &self.mode {
-            Mode::Running(local_mode) => {
-                let capabilities = local_mode.request(request);
 
-                cx.spawn(async move |this, cx| {
-                    let capabilities = capabilities.await?;
-                    this.update(cx, |session, cx| {
-                        session.capabilities = capabilities;
-                        let filters = session
-                            .capabilities
-                            .exception_breakpoint_filters
-                            .clone()
-                            .unwrap_or_default();
-                        for filter in filters {
-                            let default = filter.default.unwrap_or_default();
-                            session
-                                .exception_breakpoints
-                                .entry(filter.filter.clone())
-                                .or_insert_with(|| (filter, default));
-                        }
-                        cx.emit(SessionEvent::CapabilitiesLoaded);
-                    })?;
-                    Ok(())
-                })
-            }
-            Mode::Building => Task::ready(Err(anyhow!(
+        let Mode::Running(running) = &self.mode else {
+            return Task::ready(Err(anyhow!(
                 "Cannot send initialize request, task still building"
-            ))),
-        }
+            )));
+        };
+        let mut response = running.request(request.clone());
+
+        cx.spawn(async move |this, cx| {
+            loop {
+                let capabilities = response.await;
+                match capabilities {
+                    Err(e) => {
+                        let Ok(Some(reconnect)) = this.update(cx, |this, cx| {
+                            this.as_running()
+                                .and_then(|running| running.reconnect_for_ssh(&mut cx.to_async()))
+                        }) else {
+                            return Err(e);
+                        };
+                        log::info!("Failed to connect to debug adapter: {}, retrying...", e);
+                        reconnect.await?;
+
+                        let Ok(Some(r)) = this.update(cx, |this, _| {
+                            this.as_running()
+                                .map(|running| running.request(request.clone()))
+                        }) else {
+                            return Err(e);
+                        };
+                        response = r
+                    }
+                    Ok(capabilities) => {
+                        this.update(cx, |session, cx| {
+                            session.capabilities = capabilities;
+                            let filters = session
+                                .capabilities
+                                .exception_breakpoint_filters
+                                .clone()
+                                .unwrap_or_default();
+                            for filter in filters {
+                                let default = filter.default.unwrap_or_default();
+                                session
+                                    .exception_breakpoints
+                                    .entry(filter.filter.clone())
+                                    .or_insert_with(|| (filter, default));
+                            }
+                            cx.emit(SessionEvent::CapabilitiesLoaded);
+                        })?;
+                        return Ok(());
+                    }
+                }
+            }
+        })
     }
 
     pub(super) fn initialize_sequence(