fix(ui): keep blocking host work off the UI thread and off gpui's pool

Five findings from review, all about where blocking work runs and what
a stale handle is still pointing at.

- `live_pane_count` ran a routed `List` — an SSH handshake, and on a WSL
  route as far as installing the server — straight from the Stop/Delete
  action handler. That is `guard_off_ui`'s debug abort in a dev build
  and a frozen window in a release one. It is now split into a UI-thread
  read and a background count, with the prompt raised through the window
  handle afterwards.
- `teardown_workspace_forwards` blocked the UI thread on a daemon reply
  that waits for the SSH server to acknowledge `cancel_tcpip_forward`.
  On a machine that has gone unreachable — exactly when someone reaches
  for Stop Workspace — it never came. Backgrounded, and `on_workspace`
  now sets a read timeout so the thread is not parked forever either.
- The file tree's and editor's watch subscriptions had no record of
  which host opened them. A reconnect inserts a fresh `RemoteHost` under
  the same `HostId`, so `set_dirs` failed on a dead `ControlClient`,
  was warned and dropped, and nothing opened a new one: after the first
  reconnect the tree stopped seeing remote changes for the life of the
  window, and the editor's external-change detection — what stops a save
  clobbering someone else's edit — was silently off. Both now compare
  the host by pointer and reopen when it differs.
- Closing a remote window that was empty *because its machine could not
  be reached* deleted the workspace: its `RemoteRef`, cached layout and
  geometry, while its panes were still running over there. Only a
  machine that answered licenses dropping the entry.
- `HostOps` ran blocking calls on gpui's background executor, which on
  Linux is a fixed pool with no blocking tier. Four stalled host calls
  on a four-core client took every worker, including the one the
  reconnect needed to clear the stall. They now run on their own elastic
  pool.

src/terminal/remote.rs

@@ -1802,6 +1802,13 @@ impl RemoteTerminal {
 
     /// Send one workspace-scoped request and return the daemon's reply.
     ///
+    /// How long a workspace-addressed request waits for the daemon.
+    ///
+    /// Generous, because behind it is an SSH round trip to the workspace's own
+    /// machine and possibly a connection being established — but finite, which
+    /// is the point.
+    const WORKSPACE_OP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+
     /// The counterpart of the `pane_id`-addressed helpers above for a pane that
     /// lives on a *remote workspace*: there is no pane on the local daemon to
     /// name, so the request carries the workspace and a secret-free spec naming
@@ -1813,6 +1820,17 @@ impl RemoteTerminal {
     /// empty list.
     pub fn on_workspace(req: WorkspaceRequest) -> anyhow::Result<DaemonMsg> {
         let mut stream = connect()?;
+        // Bounded, because the daemon's answer is not just its own work: it
+        // resolves the workspace's SSH connection and, for the forward ops,
+        // waits for the *server* to acknowledge a `cancel_tcpip_forward`. On a
+        // box that has gone unreachable — lid closed, VPN dropped, which is
+        // exactly when someone reaches for Stop Workspace — that acknowledgement
+        // never comes. Without a deadline this read parks forever, and the
+        // thread with it.
+        //
+        // Best effort: a transport that will not take a timeout degrades to the
+        // old unbounded read rather than failing the request outright.
+        let _ = stream.set_read_timeout(Some(Self::WORKSPACE_OP_TIMEOUT));
         ClientMsg::OnWorkspace(Box::new(req)).encode(&mut stream)?;
         match DaemonMsg::read(&mut stream)? {
             DaemonMsg::Error(msg) => Err(anyhow::anyhow!(msg)),

src/ui/app.rs

@@ -1263,7 +1263,17 @@ impl Tty7App {
         // An empty workspace has nothing to come back to, so it is dropped
         // outright instead of accumulating as a blank row in the picker —
         // every `New Workspace` the user closes without using would leave one.
-        if self.tabs.is_empty() {
+        //
+        // Unless the emptiness is *this client's* ignorance rather than the
+        // machine's answer. `claimable_session` deliberately opens a remote
+        // workspace empty when its machine cannot be reached, so a window
+        // opened while the box was asleep and then closed — there was nothing
+        // in it to work on — would take the entry with it: its `RemoteRef`, its
+        // cached layout and its geometry, while its panes are still running
+        // over there. Nothing would reconnect it and nothing would offer it
+        // again; the only way back is re-adding the machine by hand.
+        let answered = WorkspaceStore::machine_is_connected(cx, self.workspace);
+        if self.tabs.is_empty() && answered {
             WorkspaceStore::remove(cx, self.workspace);
         } else {
             WorkspaceStore::close_window(cx, self.workspace);
@@ -1297,10 +1307,21 @@ impl Tty7App {
         else {
             return;
         };
-        let left = route.teardown();
-        if !left.is_empty() {
-            log::warn!("{} forwards survived a workspace teardown", left.len());
-        }
+        // Off the UI thread. `teardown` dials the daemon, which resolves the
+        // workspace's SSH connection and waits for the server to acknowledge a
+        // `cancel_tcpip_forward` — on a machine that has gone unreachable, which
+        // is exactly when someone reaches for Stop Workspace, that never comes
+        // back inside the request timeout. `ForwardRoute::list` is already
+        // backgrounded for the same reason; this was the one that was not, and
+        // it ran while the window was being torn down.
+        cx.background_executor()
+            .spawn(async move {
+                let left = route.teardown();
+                if !left.is_empty() {
+                    log::warn!("{} forwards survived a workspace teardown", left.len());
+                }
+            })
+            .detach();
     }
 
     /// Stop a workspace — kill its sessions and close its window — confirming

src/ui/code_editor.rs

@@ -41,7 +41,7 @@ use gpui_component::{
 };
 
 use crate::ui::app::Tty7App;
-use crate::ui::host_ops::{HostOps, MTime, WatchSub};
+use crate::ui::host_ops::{HostOps, MTime, SharedHost, WatchSub};
 
 /// Refuse to open files larger than this: the component's code editor is rated
 /// to ~50K lines, and a multi-megabyte blob is almost never what a terminal
@@ -165,6 +165,15 @@ pub(crate) struct EditorPanelState {
     /// and a server-side watcher recreated every time a file is opened or
     /// closed. `Arc` because `set_dirs` is itself a host call.
     watch: Option<Arc<WatchSub>>,
+    /// The host `watch` was opened against, kept so a subscription is never
+    /// reused across a different one.
+    ///
+    /// A `HostId` is not enough to tell them apart: reconnecting removes the
+    /// dead `RemoteHost` and inserts a fresh one under the *same* id, so the id
+    /// matches while the `ControlClient` behind the old subscription is gone.
+    /// Compared by pointer, which distinguishes both that and an outright
+    /// switch to another machine.
+    watch_host: Option<SharedHost>,
     /// A subscription is being opened; keeps a burst of opens from asking for
     /// one each.
     watch_opening: bool,
@@ -217,6 +226,7 @@ impl EditorPanelState {
         .detach();
         Self {
             watch: None,
+            watch_host: None,
             watch_opening: false,
             watch_busy: false,
             watch_dirty: false,
@@ -434,6 +444,25 @@ impl Tty7App {
             return;
         };
 
+        // Same rule as the file tree's: a subscription belongs to the host that
+        // opened it. A reconnect inserts a fresh `RemoteHost` under the same
+        // `HostId`, so the id matches while the `ControlClient` behind this
+        // subscription is gone — `set_dirs` then fails, is warned and dropped,
+        // and nothing opens a new one. The cost here is quieter and worse than
+        // a stale tree: external-change detection is what stops a save
+        // clobbering an edit made on the other side.
+        if !self
+            .editor
+            .watch_host
+            .as_ref()
+            .is_some_and(|opened_with| Arc::ptr_eq(opened_with, &host))
+        {
+            self.editor.watch = None;
+            self.editor.watch_host = None;
+            self.editor.watch_busy = false;
+            self.editor.watch_dirty = false;
+        }
+
         if let Some(sub) = self.editor.watch.clone() {
             if self.editor.watch_busy {
                 self.editor.watch_dirty = true;
@@ -462,6 +491,7 @@ impl Tty7App {
             return;
         }
         self.editor.watch_opening = true;
+        let opened_host = Arc::clone(&host);
         let opened_with = self.editor.watched_dirs.clone();
         HostOps::run(
             host,
@@ -481,6 +511,7 @@ impl Tty7App {
                 };
                 let events = sub.events().clone();
                 app.editor.watch = Some(sub);
+                app.editor.watch_host = Some(opened_host);
                 cx.spawn(async move |app, cx| {
                     while let Ok(batch) = events.recv().await {
                         let ok = app.update(cx, |app, _cx| {

src/ui/file_tree.rs

@@ -211,6 +211,15 @@ pub(crate) struct FileTreeState {
     /// triangle. `Arc` because `set_dirs` is itself a host call and has to be
     /// handed to the background executor.
     watch: Option<Arc<WatchSub>>,
+    /// The host `watch` was opened against, kept so a subscription is never
+    /// reused across a different one.
+    ///
+    /// A `HostId` is not enough to tell them apart: reconnecting removes the
+    /// dead `RemoteHost` and inserts a fresh one under the *same* id, so the id
+    /// matches while the `ControlClient` behind the old subscription is gone.
+    /// Compared by pointer, which distinguishes both that and an outright
+    /// switch to another machine.
+    watch_host: Option<SharedHost>,
     /// A subscription is being opened. Without this, render would ask for one
     /// per frame until the first answer lands.
     watch_opening: bool,
@@ -255,6 +264,7 @@ impl FileTreeState {
         })
         .detach();
         Self {
+            watch_host: None,
             children: ByHost::default(),
             loads: InFlight::default(),
             stale: HashSet::new(),
@@ -283,6 +293,25 @@ impl FileTreeState {
     fn sync_watch(&mut self, host: SharedHost, dirs: HashSet<PathBuf>, cx: &mut Context<Tty7App>) {
         self.watched = dirs;
         let want: Vec<PathBuf> = self.watched.iter().cloned().collect();
+        // A subscription belongs to the host that opened it. Reconnecting drops
+        // the dead `RemoteHost` and inserts a fresh one under the same
+        // `HostId`, and adopting another workspace can change the host outright
+        // — in both cases the subscription here is over a `ControlClient` that
+        // is gone. `set_dirs` on it then fails with `ConnectionReset`, which is
+        // warned and dropped, and nothing ever opens a new one: after the first
+        // reconnect of a remote workspace the tree stops seeing changes made on
+        // the far side for the rest of the window's life.
+        if !self
+            .watch_host
+            .as_ref()
+            .is_some_and(|opened_with| Arc::ptr_eq(opened_with, &host))
+        {
+            // Dropping the subscription is what unsubscribes, on both sides.
+            self.watch = None;
+            self.watch_host = None;
+            self.watch_busy = false;
+            self.watch_dirty = false;
+        }
         if let Some(sub) = self.watch.clone() {
             if self.watch_busy {
                 self.watch_dirty = true;
@@ -316,6 +345,7 @@ impl FileTreeState {
         }
         self.watch_opening = true;
         let host_id = host.id();
+        let opened_host = Arc::clone(&host);
         let opened_with = self.watched.clone();
         HostOps::run(
             host,
@@ -337,6 +367,7 @@ impl FileTreeState {
                 // independent of the subscription the state holds.
                 let events = sub.events().clone();
                 app.file_tree.watch = Some(sub);
+                app.file_tree.watch_host = Some(opened_host);
                 cx.spawn(async move |app, cx| {
                     while let Ok(batch) = events.recv().await {
                         let ok = app.update(cx, |app, _cx| {

src/ui/host_ops.rs

@@ -45,7 +45,7 @@ use std::borrow::Borrow;
 use std::collections::{HashMap, HashSet};
 use std::hash::Hash;
 
-use gpui::{App, AppContext as _, Context, Window};
+use gpui::{App, Context, Window};
 use gpui_component::WindowExt as _;
 
 // The host vocabulary, re-exported so a view imports everything it needs from
@@ -56,6 +56,156 @@ pub use tty7_core::host::{
     Entry, Host, HostId, MTime, Meta, Output, SearchHit, SharedHost, WatchSub,
 };
 
+/// Where blocking [`Host`] calls actually run.
+///
+/// **Not gpui's background executor.** On Linux that is a fixed pool of
+/// `available_parallelism().max(2)` worker threads with no separate blocking
+/// tier, so N stalled host calls on an N-core client occupy every worker there
+/// is. Everything else that uses `background_executor` then queues behind them
+/// — including the reconnect in `remote_workspace::launch_attempt`, which is
+/// the one thing that would clear the stall. Expanding a subtree on a link that
+/// has gone silent is enough: one call per directory, each parked for its
+/// deadline (5s for a `ReadDir`, 30s for a `ReadFile`). macOS is far less
+/// exposed, since libdispatch grows its global queues when their threads block,
+/// which is why this does not show up in development.
+///
+/// Elastic and its own: a thread per concurrent call, reused while warm and
+/// retired after [`LINGER`], capped at [`MAX_THREADS`]. A `Host` call is
+/// user-driven — a directory expanded, a file opened — not per-frame, so the
+/// steady state is one or two threads.
+mod blocking {
+    use std::collections::VecDeque;
+    use std::sync::{Arc, Condvar, Mutex, OnceLock};
+    use std::time::Duration;
+
+    type Job = Box<dyn FnOnce() + Send + 'static>;
+
+    /// Ceiling on threads. Deliberately well above any core count: these are
+    /// parked on a socket rather than competing for CPU, and what has to fit is
+    /// the number of host calls in flight — one per expanded directory in a
+    /// burst, plus whatever the editor and the git probes are doing.
+    const MAX_THREADS: usize = 64;
+
+    /// How long an idle worker waits for more work before retiring.
+    const LINGER: Duration = Duration::from_secs(30);
+
+    struct Inner {
+        state: Mutex<State>,
+        wake: Condvar,
+    }
+
+    struct State {
+        jobs: VecDeque<Job>,
+        threads: usize,
+        /// Workers parked in `wait_timeout`, counted from before they park
+        /// until after they have re-acquired the lock on the way out.
+        idle: usize,
+    }
+
+    impl State {
+        /// Whether a job just queued needs a thread spawned for it.
+        ///
+        /// Compares the backlog against the parked workers rather than asking
+        /// whether *any* worker is parked: `idle` still counts a worker that
+        /// has been handed a job but has not yet woken, and the job meant for
+        /// it is still in `jobs`, so counting both sides cancels the window out.
+        fn wants_another_thread(&self) -> bool {
+            self.jobs.len() > self.idle && self.threads < MAX_THREADS
+        }
+    }
+
+    fn pool() -> &'static Arc<Inner> {
+        static POOL: OnceLock<Arc<Inner>> = OnceLock::new();
+        POOL.get_or_init(|| {
+            Arc::new(Inner {
+                state: Mutex::new(State {
+                    jobs: VecDeque::new(),
+                    threads: 0,
+                    idle: 0,
+                }),
+                wake: Condvar::new(),
+            })
+        })
+    }
+
+    /// Queue `job`. Never refuses: a dropped job is a `Host` call whose caller
+    /// waits forever, and at this cap the backlog is a better failure than that.
+    pub(super) fn submit(job: impl FnOnce() + Send + 'static) {
+        let inner = pool();
+        let mut st = inner.state.lock().unwrap_or_else(|e| e.into_inner());
+        st.jobs.push_back(Box::new(job));
+        if st.wants_another_thread() {
+            st.threads += 1;
+            let spawned = Arc::clone(inner);
+            match std::thread::Builder::new()
+                .name("tty7-host-op".into())
+                .spawn(move || worker(spawned))
+            {
+                Ok(_) => return,
+                Err(e) => {
+                    // Out of threads: leave the job for whoever is already
+                    // running. With nobody at all there is no one to run it, so
+                    // run it here — blocking this caller, which is the lesser
+                    // harm against never answering.
+                    st.threads -= 1;
+                    log::warn!("could not start a host-op thread: {e}");
+                    if st.threads == 0
+                        && let Some(job) = st.jobs.pop_back()
+                    {
+                        drop(st);
+                        job();
+                        return;
+                    }
+                }
+            }
+        }
+        drop(st);
+        inner.wake.notify_one();
+    }
+
+    fn worker(inner: Arc<Inner>) {
+        loop {
+            let job = {
+                let mut st = inner.state.lock().unwrap_or_else(|e| e.into_inner());
+                loop {
+                    if let Some(job) = st.jobs.pop_front() {
+                        break job;
+                    }
+                    st.idle += 1;
+                    let (guard, timeout) = inner
+                        .wake
+                        .wait_timeout(st, LINGER)
+                        .unwrap_or_else(|e| e.into_inner());
+                    st = guard;
+                    st.idle -= 1;
+                    if timeout.timed_out() && st.jobs.is_empty() {
+                        st.threads -= 1;
+                        return;
+                    }
+                }
+            };
+            job();
+        }
+    }
+}
+
+/// Run `f` on the blocking pool and await its result.
+///
+/// `None` means the job was dropped without running, which happens only when
+/// the process is going down — the caller lands nothing rather than inventing
+/// an answer.
+async fn off_thread<T, F>(f: F) -> Option<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> T + Send + 'static,
+{
+    let (tx, rx) = smol::channel::bounded(1);
+    blocking::submit(move || {
+        let _ = tx.send_blocking(f());
+    });
+    rx.recv().await.ok()
+}
+
 /// The GPUI-facing facade over [`Host`].
 ///
 /// A unit struct rather than a value: there is no per-instance state, and
@@ -84,7 +234,9 @@ impl HostOps {
         // optimizer sees through after the first.
         tty7_core::host::register_ui_thread();
         cx.spawn(async move |this, cx| {
-            let out = cx.background_spawn(async move { f(&*host) }).await;
+            let Some(out) = off_thread(move || f(&*host)).await else {
+                return;
+            };
             let _ = this.update(cx, |view, cx| land(view, out, cx));
         })
         .detach();
@@ -114,7 +266,9 @@ impl HostOps {
     {
         tty7_core::host::register_ui_thread();
         cx.spawn(async move |_this, cx| {
-            let out = cx.background_spawn(async move { f(&*host) }).await;
+            let Some(out) = off_thread(move || f(&*host)).await else {
+                return;
+            };
             cx.update(|cx| land(cx, out));
         })
         .detach();
@@ -131,7 +285,9 @@ impl HostOps {
     {
         tty7_core::host::register_ui_thread();
         cx.spawn_in(window, async move |this, cx| {
-            let out = cx.background_spawn(async move { f(&*host) }).await;
+            let Some(out) = off_thread(move || f(&*host)).await else {
+                return;
+            };
             let _ = this.update_in(cx, |view, window, cx| land(view, out, window, cx));
         })
         .detach();

src/ui/windows.rs

@@ -280,20 +280,41 @@ pub fn menu_order(cx: &App) -> Vec<(WorkspaceId, bool)> {
 /// [`pane_liveness`](crate::terminal::pane_liveness): the prompt states an exact
 /// number about an irreversible action, so it wants a fresh count, not one that
 /// may be ten seconds old. This runs on a click, not on a frame.
-pub fn live_pane_count(cx: &App, workspace: WorkspaceId) -> Option<usize> {
+/// What [`live_pane_count`] needs from the app, gathered on the UI thread so the
+/// count itself does not have to run there.
+pub struct PaneCountQuery {
+    route: crate::terminal::PaneRoute,
+    claimed: Vec<u64>,
+}
+
+/// Read the inputs for [`live_pane_count`]. Cheap; UI thread only.
+pub fn pane_count_query(cx: &App, workspace: WorkspaceId) -> Option<PaneCountQuery> {
     let ws = WorkspaceStore::all(cx).get(workspace)?;
-    let claimed = ws.pane_ids();
+    Some(PaneCountQuery {
+        // Routed to the workspace's own machine: a remote workspace's pane ids
+        // mean nothing to this computer's daemon, so asking it would count
+        // whichever *local* panes happen to hold those numbers and put a "3
+        // running sessions will be ended" warning on a workspace that has none.
+        route: crate::ui::remote_workspace::pane_route_for(cx, workspace),
+        claimed: ws.pane_ids(),
+    })
+}
+
+/// **Blocking. Never call this on the UI thread.**
+///
+/// For a remote route this dials the workspace's machine — an SSH handshake if
+/// nothing is pooled — and a WSL one can go as far as installing the server
+/// binary. `guard_off_ui` makes a UI-thread call a debug-build abort rather
+/// than a dropped frame, which is what it did when this was reached straight
+/// from the Stop/Delete action handler.
+pub fn live_pane_count(q: &PaneCountQuery) -> Option<usize> {
+    let PaneCountQuery { route, claimed } = q;
     if claimed.is_empty() {
         return Some(0);
     }
     // One short-lived connection, only when there is something to ask about —
-    // the picker renders far more often than a workspace is closed. Routed to
-    // the workspace's own machine: a remote workspace's pane ids mean nothing
-    // to this computer's daemon, so asking it would count whichever *local*
-    // panes happen to hold those numbers and put a "3 running sessions will be
-    // ended" warning on a workspace that has none.
-    let route = crate::ui::remote_workspace::pane_route_for(cx, workspace);
-    match crate::terminal::RemoteTerminal::try_list_panes_on(&route) {
+    // the picker renders far more often than a workspace is closed.
+    match crate::terminal::RemoteTerminal::try_list_panes_on(route) {
         Ok(panes) => {
             let alive: std::collections::HashSet<u64> = panes
                 .into_iter()
@@ -369,33 +390,57 @@ fn confirm_destructive(
     verb: &'static str,
     act: fn(&mut App, WorkspaceId),
 ) {
-    let live = live_pane_count(cx, workspace);
     let name = WorkspaceStore::all(cx)
         .get(workspace)
         .map(|w| w.display_name())
         .unwrap_or_else(|| "this workspace".to_string());
-    // Only a machine that *answered* zero licenses skipping the prompt. An
-    // unreachable one is the case most likely to still have work in it.
-    if live == Some(0) && verb == "Stop" {
-        act(cx, workspace);
-        return;
-    }
-    let detail = destructive_detail(live, verb);
-    // Title Case, like every other prompt title in the app — this one used to
-    // lowercase "workspace" while its siblings read "Close Window?" /
-    // "Quit and Stop Daemon?".
-    let answer = window.prompt(
-        gpui::PromptLevel::Warning,
-        &format!("{verb} Workspace \u{201c}{name}\u{201d}?"),
-        Some(&detail),
-        &["Cancel", verb],
-        cx,
-    );
+    let query = pane_count_query(cx, workspace);
+    let handle = window.window_handle();
+
     cx.spawn(async move |cx| {
+        // The count dials the workspace's machine, so it does not belong on the
+        // UI thread — on a remote route that is an SSH handshake, and on a WSL
+        // one it can go as far as installing the server. Reached straight from
+        // the action handler, it was a `guard_off_ui` abort in a debug build
+        // and a window frozen for the length of a connect in a release one.
+        let live = match query {
+            Some(q) => {
+                cx.background_spawn(async move { live_pane_count(&q) })
+                    .await
+            }
+            None => None,
+        };
+
+        // Only a machine that *answered* zero licenses skipping the prompt. An
+        // unreachable one is the case most likely to still have work in it.
+        if live == Some(0) && verb == "Stop" {
+            let _ = cx.update(|cx| act(cx, workspace));
+            return;
+        }
+
+        let detail = destructive_detail(live, verb);
+        // Title Case, like every other prompt title in the app — this one used
+        // to lowercase "workspace" while its siblings read "Close Window?" /
+        // "Quit and Stop Daemon?".
+        let Ok(answer) = handle.update(cx, |_, window, cx| {
+            window.prompt(
+                gpui::PromptLevel::Warning,
+                &format!("{verb} Workspace \u{201c}{name}\u{201d}?"),
+                Some(&detail),
+                &["Cancel", verb],
+                cx,
+            )
+        }) else {
+            // The window went away while we were asking its machine. Nothing to
+            // confirm against, and acting unprompted is exactly what this path
+            // exists to prevent.
+            return;
+        };
+
         // Index 1 == the verb button; Cancel and a dismissed prompt both leave
         // the workspace alone.
         if let Ok(1) = answer.await {
-            cx.update(|cx| act(cx, workspace));
+            let _ = cx.update(|cx| act(cx, workspace));
         }
     })
     .detach();