fix(ssh): stop a failed shell probe disabling integration for good

`remote_bootstrap` cached whatever `probe_remote_shell` returned, and
that function answers `None` for two unrelated reasons: the remote runs a
shell there is no bootstrap for, and the probe never got asked — the
channel would not open, the exec failed, or the five-second read ended
before the answer arrived.

The cache is a `HashMap` on the process-wide `SshManager`, behind a
`OnceLock`, with no TTL and nothing that evicts. So one slow or busy
moment on the first pane to a host turned into "this host has no shell
integration" for the rest of the daemon's life: no prompt marks, no cwd
tracking, no command status, no error, and reconnecting does not clear it
because the entry outlives the connection.

Told the two apart at the only place that can tell them apart — the
output. `probe_answer` reads exactly as far as `parse_probe` does before
it can form an opinion, and returns `None` when the marker and the line
it introduces did not both arrive. Only an answer is remembered; a
silence is asked again on the next pane.

The cost lands on the case that deserves it: a host that genuinely cannot
be probed pays one probe per pane instead of one per daemon, bounded by
the same timeout as before. A host that answers pays nothing extra.

The decision is a pure function over the probe's text, so it is tested
rather than argued about; the async path around it is unchanged bar the
return type.

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

@@ -1296,6 +1296,26 @@ pub mod remote {
         }
     }
 
+    /// What the probe *said*, kept apart from its having said nothing.
+    ///
+    /// `None` means the far side never answered: the channel or the command
+    /// failed, or the timeout cut the read off before the marker and the line
+    /// it introduces both arrived. That is a different fact from "answered,
+    /// and logs in with a shell there is no bootstrap for", which is
+    /// `Some(None)`. Only the second is worth remembering — the first is a
+    /// slow link or a busy server, and the next pane deserves a fresh ask.
+    pub(crate) fn probe_answer(output: &str) -> Option<Option<(RemoteShell, String)>> {
+        // Read exactly as far as `parse_probe` does before it can form an
+        // opinion, so the two cannot disagree about whether there was one.
+        let mut lines = output
+            .lines()
+            .map(|l| l.trim_end_matches('\r').trim())
+            .skip_while(|l| *l != PROBE_MARKER);
+        lines.next()?;
+        lines.find(|l| !l.is_empty())?;
+        Some(parse_probe(output))
+    }
+
     pub(crate) fn parse_probe(output: &str) -> Option<(RemoteShell, String)> {
         let mut lines = output
             .lines()
@@ -1436,6 +1456,39 @@ fi
             assert_eq!(parse_probe("/bin/zsh\n"), None);
         }
 
+        #[test]
+        fn a_probe_that_never_answered_is_told_apart_from_one_that_said_no() {
+            // Answered, with a shell there is a bootstrap for.
+            assert_eq!(
+                probe_answer("__tty7_shell\n/bin/zsh\n"),
+                Some(Some((RemoteShell::Zsh, "/bin/zsh".to_string())))
+            );
+            // Answered, and the answer is one there is no bootstrap for. Worth
+            // remembering: asking again gets the same `/bin/ksh` every time.
+            assert_eq!(probe_answer("__tty7_shell\n/bin/ksh\n"), Some(None));
+            assert_eq!(probe_answer("__tty7_shell\n$SHELL\n"), Some(None));
+
+            // Never answered. The channel or the exec failed, or the read timed
+            // out — none of which says anything about the remote's shell, and
+            // all of which used to be cached as if it did.
+            assert_eq!(probe_answer(""), None, "nothing came back at all");
+            assert_eq!(
+                probe_answer("Welcome to prod!\n"),
+                None,
+                "a banner, and then the read was cut off"
+            );
+            assert_eq!(
+                probe_answer("__tty7_shell\n"),
+                None,
+                "the marker arrived and the line it introduces did not"
+            );
+            assert_eq!(
+                probe_answer("__tty7_shell\n\n"),
+                None,
+                "and a blank line is not that line"
+            );
+        }
+
         #[test]
         fn zsh_bootstrap_gates_zdotdir_on_every_redirector_landing() {
             let script = bootstrap_command(RemoteShell::Zsh, "/bin/zsh");

crates/tty7-core/src/daemon/ssh/mod.rs

@@ -449,17 +449,28 @@ impl SshManager {
         let cached = { self.probes.lock().unwrap().get(&key).cloned() };
         let probed = match cached {
             Some(hit) => hit,
-            None => {
-                let probed = probe_remote_shell(conn).await;
-                match &probed {
-                    Some((shell, path)) => {
-                        log::debug!("ssh {key:?}: remote shell {shell:?} at {path}")
+            // Only an answer is remembered. This map is on the process-wide
+            // `SshManager` and nothing ever evicts from it, so caching a probe
+            // that failed would spend the rest of the daemon's life claiming a
+            // host has no shell integration because one channel open, one
+            // exec, or one five-second read went badly. Reconnecting would not
+            // clear it either.
+            None => match probe_remote_shell(conn).await {
+                Some(answer) => {
+                    match &answer {
+                        Some((shell, path)) => {
+                            log::debug!("ssh {key:?}: remote shell {shell:?} at {path}")
+                        }
+                        None => log::debug!("ssh {key:?}: no remote shell integration"),
                     }
-                    None => log::debug!("ssh {key:?}: no remote shell integration"),
+                    self.probes.lock().unwrap().insert(key, answer.clone());
+                    answer
                 }
-                self.probes.lock().unwrap().insert(key, probed.clone());
-                probed
-            }
+                None => {
+                    log::debug!("ssh {key:?}: shell probe did not answer; will ask again");
+                    None
+                }
+            },
         };
         probed.map(|(shell, path)| remote::bootstrap_command(shell, &path))
     }
@@ -612,7 +623,9 @@ const PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 
 const PROBE_OUTPUT_LIMIT: usize = 8 * 1024;
 
-async fn probe_remote_shell(conn: &SshConnection) -> Option<(remote::RemoteShell, String)> {
+/// `None` when the remote never answered, so the caller knows not to remember
+/// it. See [`remote::probe_answer`].
+async fn probe_remote_shell(conn: &SshConnection) -> Option<Option<(remote::RemoteShell, String)>> {
     let mut channel = conn.open_session_channel().await.ok()?;
     channel.exec(true, remote::PROBE_COMMAND).await.ok()?;
 
@@ -633,7 +646,7 @@ async fn probe_remote_shell(conn: &SshConnection) -> Option<(remote::RemoteShell
     };
     let _ = tokio::time::timeout(PROBE_TIMEOUT, collect).await;
 
-    remote::parse_probe(&String::from_utf8_lossy(&out))
+    remote::probe_answer(&String::from_utf8_lossy(&out))
 }
 
 async fn probe_remote_env(conn: &SshConnection) -> Option<remote_link::RemoteEnv> {