fix(daemon): notice a pane came home from ssh without waiting for output

The foreground probe that clears a pane's remote context only runs when
the reader thread has bytes in hand. The prompt a shell draws after a
command is the last output a pane produces until the user types again, so
an `ssh` that exited inside the poll interval left the pane reporting
itself as remote indefinitely — nothing came along to probe on.

Everything keyed off that context stayed on the far end. Most visibly the
history scope: ↑ read the remote list, which for a host with no history
of its own is empty, so ↑ appeared dead until some unrelated output
arrived. Pressing Enter looked like it unblocked the pane because an
empty command is the cheapest way to make output.

A prompt mark that survives the foreground suppression is the shell
saying the command it ran is over, so the foreground has just gone back
to being the shell itself. Probe right then instead of waiting out the
interval.

Switching history scopes also dropped the list it was leaving, and the
reload that refills it is a background task, so ↑ had a second window of
recalling nothing. Park each scope's list instead, capped at four, and
step back into one instantly.

Claude-Session: https://claude.ai/code/session_01Mnerr8RZ23Nd4cxyfeqxiu

crates/tty7-core/src/daemon/pane.rs

@@ -1989,7 +1989,23 @@ impl DaemonPane {
                                 signals.shell.dedup();
                             }
 
-                            let poll_now = std::time::Instant::now() >= next_remote_check;
+                            // A prompt mark that survived the suppression above
+                            // is the shell saying the command it ran is over, so
+                            // the foreground has just gone back to being the
+                            // shell itself. Probe right then rather than waiting
+                            // out the interval: the probe is what clears the
+                            // remote context an `ssh` left behind, and the
+                            // interval alone can miss it forever. Polling only
+                            // runs on output, and the prompt the shell just drew
+                            // is the last output a pane produces until the user
+                            // types again — so a pane whose `ssh` exited inside
+                            // the interval kept reporting itself as remote, and
+                            // everything keyed off that (the history scope ↑
+                            // reads, most visibly) stayed on the far end until
+                            // some unrelated output arrived (#817).
+                            let back_at_prompt = signals.shell.iter().any(|s| s.at_prompt);
+                            let poll_now =
+                                back_at_prompt || std::time::Instant::now() >= next_remote_check;
                             if poll_now {
                                 next_remote_check =
                                     std::time::Instant::now() + REMOTE_CONTEXT_POLL_INTERVAL;
@@ -5041,6 +5057,91 @@ mod tests {
         assert_eq!(snap.state.session_id.as_deref(), Some("sess-1"));
     }
 
+    /// The foreground probe only ever runs on output, and the prompt a shell
+    /// draws after a command is the last output a pane produces until the user
+    /// types again. So an `ssh` that exited inside the poll interval used to
+    /// leave the pane reporting itself as remote indefinitely: nothing came
+    /// along to probe on. A prompt mark now forces the probe (#817).
+    #[test]
+    fn a_prompt_mark_reprobes_the_foreground_inside_the_poll_interval() {
+        /// Hands the reader one chunk per `read`, so two prompt marks arrive as
+        /// two passes through the loop a few microseconds apart — well inside
+        /// `REMOTE_CONTEXT_POLL_INTERVAL`.
+        struct Chunks(std::collections::VecDeque<Vec<u8>>);
+
+        impl Read for Chunks {
+            fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+                match self.0.pop_front() {
+                    Some(chunk) => {
+                        buf[..chunk.len()].copy_from_slice(&chunk);
+                        Ok(chunk.len())
+                    }
+                    None => Ok(0),
+                }
+            }
+        }
+
+        let state = Arc::new(Mutex::new(test_state(true)));
+        let (sub_tx, sub_rx) = mpsc::channel();
+        state.lock().unwrap().subscriber = Some(sub_tx);
+
+        let probes_taken = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let taken = probes_taken.clone();
+        let remote = Box::new(move || {
+            // First probe: `ssh` holds the pty. Every one after: it is gone.
+            (taken.fetch_add(1, Ordering::SeqCst) == 0).then(|| RemoteContext {
+                kind: RemoteKind::Ssh,
+                argv: vec!["ssh".into(), "box".into()],
+                target: "box".into(),
+            })
+        });
+
+        let handle = DaemonPane::spawn_reader(
+            state.clone(),
+            Arc::new(AtomicBool::new(false)),
+            Arc::new(OutputGate::new()),
+            Box::new(Chunks(
+                [
+                    b"\x1b]133;C;ssh box\x07".to_vec(),
+                    b"\x1b]133;D;0\x07".to_vec(),
+                ]
+                .into_iter()
+                .collect(),
+            )),
+            null_writer(),
+            || false,
+            ForegroundProbes {
+                remote,
+                agent: Box::new(|| None),
+                cwd: Box::new(|| None),
+            },
+            Arc::new(DeathReporter::new(|| {})),
+        );
+        handle.join().unwrap();
+
+        assert_eq!(
+            probes_taken.load(Ordering::SeqCst),
+            2,
+            "the prompt mark did not force a second probe"
+        );
+        assert!(
+            state.lock().unwrap().remote.is_none(),
+            "the pane still reports the ssh session it has already left"
+        );
+        let reported: Vec<Option<String>> = sub_rx
+            .try_iter()
+            .filter_map(|msg| match msg {
+                DaemonMsg::RemoteContext(ctx) => Some(ctx.map(|c| c.target)),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(
+            reported,
+            vec![Some("box".to_string()), None],
+            "the client was never told the pane came home"
+        );
+    }
+
     #[test]
     fn reader_eof_with_subscriber_sends_exited_not_on_dead() {
         let state = Arc::new(Mutex::new(test_state(true)));

src/terminal/view.rs

@@ -406,6 +406,14 @@ pub struct TerminalView {
     history_ranked: Vec<String>,
     history_frecency: Vec<f64>,
     history_scope: super::history::Scope,
+    /// What each scope this pane has already loaded held when it was left, so
+    /// stepping back into one (`exit` out of an `ssh` session, most of all)
+    /// has a list to recall from right away instead of an empty one that only
+    /// refills once a background read lands (#817).
+    history_cache: Vec<(super::history::Scope, super::history::History)>,
+    /// Whether the current scope's list is a finished load rather than the
+    /// empty placeholder one starts as. Only a finished one is worth stashing.
+    history_ready: bool,
     ranked_cwd: Option<std::path::PathBuf>,
     history_nav: Option<usize>,
     history_stash: String,
@@ -1557,6 +1565,8 @@ impl TerminalView {
             history_ranked,
             history_frecency,
             history_scope: super::history::Scope::Local,
+            history_cache: Vec::new(),
+            history_ready: true,
             ranked_cwd: None,
             history_nav: None,
             history_stash: String::new(),
@@ -3658,21 +3668,70 @@ impl TerminalView {
         super::history::Scope::Local
     }
 
+    /// How many scopes' lists to keep around. A pane hops between a handful of
+    /// hosts at most; the cap is only here so a long-lived pane that reaches
+    /// many of them cannot grow without bound.
+    const HISTORY_CACHE_MAX: usize = 4;
+
+    /// Park the current scope's list so coming back to it is instant. A list
+    /// that never finished loading is not worth parking — the empty one it
+    /// would leave behind is exactly what the cache exists to avoid handing
+    /// back.
+    fn stash_history(&mut self) {
+        if !self.history_ready {
+            return;
+        }
+        let scope = self.history_scope.clone();
+        self.history_cache.retain(|(cached, _)| *cached != scope);
+        self.history_cache.push((
+            scope,
+            super::history::History {
+                entries: std::mem::take(&mut self.history),
+                counts: std::mem::take(&mut self.history_counts),
+                cwds: std::mem::take(&mut self.history_cwds),
+                meta: std::mem::take(&mut self.history_meta),
+            },
+        ));
+        if self.history_cache.len() > Self::HISTORY_CACHE_MAX {
+            self.history_cache.remove(0);
+        }
+    }
+
     fn follow_history_scope(&mut self, cx: &mut Context<Self>) {
         let scope = self.desired_history_scope();
         if scope == self.history_scope {
             return;
         }
         self.flush_pending_history();
+        self.stash_history();
         self.history_scope = scope.clone();
-        self.history.clear();
-        self.history_counts.clear();
-        self.history_cwds.clear();
-        self.history_meta.clear();
+        match self
+            .history_cache
+            .iter()
+            .position(|(cached, _)| *cached == scope)
+        {
+            Some(i) => {
+                let (_, cached) = self.history_cache.remove(i);
+                self.history = cached.entries;
+                self.history_counts = cached.counts;
+                self.history_cwds = cached.cwds;
+                self.history_meta = cached.meta;
+                self.history_ready = true;
+            }
+            None => {
+                self.history.clear();
+                self.history_counts.clear();
+                self.history_cwds.clear();
+                self.history_meta.clear();
+                self.history_ready = false;
+            }
+        }
         self.history_ranked.clear();
         self.history_frecency.clear();
         self.history_nav = None;
         self.reverse_search = None;
+        let ranked_cwd = self.ranked_cwd.clone();
+        self.rerank_history(ranked_cwd.as_deref());
         cx.notify();
 
         let shell_files = self.remote_shell_history_sources(cx);
@@ -3700,6 +3759,7 @@ impl TerminalView {
                 view.history_counts = loaded.counts;
                 view.history_cwds = loaded.cwds;
                 view.history_meta = loaded.meta;
+                view.history_ready = true;
                 let cwd = view.ranked_cwd.clone();
                 view.rerank_history(cwd.as_deref());
                 cx.notify();
@@ -10022,6 +10082,107 @@ mod gpui_tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    /// Coming back from an `ssh` session left the pane with no history at all:
+    /// the scope switch cleared the list, and the reload that refills it is a
+    /// background task, so ↑ recalled nothing until that landed (#817).
+    #[gpui::test]
+    fn a_pane_back_from_ssh_still_recalls_its_local_history(cx: &mut TestAppContext) {
+        crate::core::config::pin_test_config_dir();
+        let (window, mut daemon) = harness(cx);
+        window
+            .update(cx, |view, _, _| {
+                view.history = vec!["cargo build".to_string(), "ssh box".to_string()];
+            })
+            .unwrap();
+
+        let away = crate::daemon::protocol::RemoteContext {
+            kind: crate::daemon::protocol::RemoteKind::Ssh,
+            argv: vec!["ssh".into(), "box".into()],
+            target: "box".into(),
+        };
+        DaemonMsg::RemoteContext(Some(away))
+            .encode(&mut daemon)
+            .unwrap();
+        for _ in 0..200 {
+            if window
+                .update(cx, |view, _, _| view.remote_context().is_some())
+                .unwrap()
+            {
+                break;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+        window
+            .update(cx, |view, _, cx| view.follow_history_scope(cx))
+            .unwrap();
+
+        DaemonMsg::RemoteContext(None).encode(&mut daemon).unwrap();
+        for _ in 0..200 {
+            if window
+                .update(cx, |view, _, _| view.remote_context().is_none())
+                .unwrap()
+            {
+                break;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+
+        window
+            .update(cx, |view, _, cx| {
+                view.follow_history_scope(cx);
+                view.handle_editor_key(&key("up"), cx);
+                assert_eq!(
+                    view.cmd.text(),
+                    "ssh box",
+                    "↑ right after the ssh session ended recalled nothing"
+                );
+            })
+            .unwrap();
+    }
+
+    /// The cache above must not hand a scope someone else's list: stepping into
+    /// an `ssh` session still starts from nothing until the far end's own
+    /// history is read.
+    #[gpui::test]
+    fn a_pane_going_out_to_ssh_does_not_inherit_the_local_history(cx: &mut TestAppContext) {
+        crate::core::config::pin_test_config_dir();
+        let (window, mut daemon) = harness(cx);
+        window
+            .update(cx, |view, _, _| {
+                view.history = vec!["rm -rf ./build".to_string()];
+            })
+            .unwrap();
+
+        DaemonMsg::RemoteContext(Some(crate::daemon::protocol::RemoteContext {
+            kind: crate::daemon::protocol::RemoteKind::Ssh,
+            argv: vec!["ssh".into(), "box".into()],
+            target: "box".into(),
+        }))
+        .encode(&mut daemon)
+        .unwrap();
+        for _ in 0..200 {
+            if window
+                .update(cx, |view, _, _| view.remote_context().is_some())
+                .unwrap()
+            {
+                break;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+
+        window
+            .update(cx, |view, _, cx| {
+                view.follow_history_scope(cx);
+                view.handle_editor_key(&key("up"), cx);
+                assert_eq!(
+                    view.cmd.text(),
+                    "",
+                    "a local command was recalled onto a remote prompt"
+                );
+            })
+            .unwrap();
+    }
+
     /// Absolute paths in a pane that is `ssh`-ed somewhere used to be resolved
     /// against *this* machine's filesystem, so `/etc/hosts` on the far end
     /// opened the local copy without a word about it.