Add non-destructive disconnect status line and tick-driven prediction policy

Item 1: mosh-style disconnect status line. When UDP packets stop arriving,
paint a single reverse-video cyan "[dosh] last contact Ns ago — reconnecting…"
on the terminal's bottom row using save/restore cursor (ESC7/ESC8) so the app
cursor never moves and full-screen TUIs are not corrupted. Cleared the instant a
packet resumes and refreshed on the status timer tick. Configurable via
client.toml `disconnect_status` (default on) and `DOSH_DISCONNECT_STATUS` env
override. State machine and rendered escape sequences are pure and unit-tested.

Item 2: re-evaluate the prediction display policy on the select-loop status tick
via `Predictor::refresh_policy`, so a latency spike flips speculation on (or off
on recovery) promptly instead of waiting for the next keystroke. Repaints only
when the on-screen result would change, so idle ticks are no-ops.

Tests: 8 new client unit tests (status state machine + overlay bytes, predictor
spike/recovery/idle). fmt + full test suite green. No wire-format change.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/bin/dosh-client.rs

@@ -2097,6 +2097,9 @@ async fn run_terminal(
         predict && cred.mode != "view-only" && !forward_only,
         predict_mode,
     );
+    // Non-destructive disconnect status line. Off in forward-only mode (no TTY to
+    // draw on) and respecting the client config (default on, env override).
+    let mut disconnect_status = DisconnectStatus::new(resolve_disconnect_status() && !forward_only);
     let (forward_tx, mut forward_rx) = mpsc::channel::<ForwardEvent>(1024);
     let _forward_keepalive = if local_forwards.is_empty() {
         Some(forward_tx.clone())
@@ -2194,6 +2197,8 @@ async fn run_terminal(
                 if packet.header.conn_id != [0u8; 16] && packet.header.conn_id != cred.client_id {
                     continue;
                 }
+                // Contact resumed: clear any disconnect status line immediately.
+                disconnect_status.apply(disconnect_status.on_contact())?;
                 match packet.header.kind {
                     PacketKind::Frame | PacketKind::ResumeOk => {
                         let decrypted = protocol::decrypt_body(&packet, &cred.session_key, SERVER_TO_CLIENT)
@@ -2542,9 +2547,24 @@ async fn run_terminal(
                     send_seq += 1;
                     socket.send_to(&packet, addr).await?;
                 }
+                // Re-evaluate the prediction display policy on every timer tick so
+                // a latency spike (or recovery) flips speculation on/off promptly
+                // without waiting for the next keystroke to drive `redraw`.
+                if !forward_only {
+                    predictor.refresh_policy()?;
+                }
+                // Refresh / clear the non-destructive disconnect status line based
+                // on how long the link has been silent (recomputed after any
+                // reconnect attempt above may have reset `last_packet_at`).
+                if !forward_only {
+                    let action = disconnect_status.on_tick(last_packet_at.elapsed());
+                    disconnect_status.apply(action)?;
+                }
             }
         }
     }
+    // Leave the terminal clean on exit: erase any lingering status line.
+    let _ = disconnect_status.apply(StatusAction::Clear);
     let _ = detach_once(&socket, &cred, send_seq).await;
     Ok(())
 }
@@ -3335,6 +3355,19 @@ impl Predictor {
         self.update_triggers();
     }
 
+    /// Test seam: backdate the oldest outstanding prediction so the glitch-force
+    /// display path (latency spike) can be exercised without real time passing.
+    #[cfg(test)]
+    fn backdate_pending_for_test(&mut self, ms: u64) {
+        self.oldest_pending_at = Some(Instant::now() - Duration::from_millis(ms));
+    }
+
+    /// Test seam: whether an overlay is currently painted on screen.
+    #[cfg(test)]
+    fn is_drawn(&self) -> bool {
+        self.drawn_width > 0
+    }
+
     /// Whether we should *display* predictions right now under the active policy.
     fn should_display(&self) -> bool {
         match self.mode {
@@ -3407,7 +3440,28 @@ impl Predictor {
         // right amount even if the model changes before the next repaint.
         self.drawn_width = bytes.len();
         self.drawn_flagged = flag;
-        let _ = self.drawn_flagged;
+        Ok(())
+    }
+
+    /// Re-evaluate the display policy on a select-loop timer tick (not just on a
+    /// keystroke). The experimental policy can decide to *start* showing purely
+    /// with the passage of time — `should_display` force-shows once a prediction
+    /// has been outstanding longer than `GLITCH_FORCE_MS` — so without this, a
+    /// sudden latency spike would only surface predictions on the *next* key,
+    /// one event late. Calling this from the status tick flips speculation on (or
+    /// off, or re-flags it) promptly. Repaints only when the on-screen result
+    /// would actually change, so an idle tick is a cheap no-op (no flicker).
+    fn refresh_policy(&mut self) -> Result<()> {
+        if !self.enabled || self.alternate_screen {
+            return Ok(());
+        }
+        self.update_triggers();
+        let want = self.should_display() && !self.visible_bytes().is_empty();
+        let have = self.drawn_width > 0;
+        let flag_changed = want && have && self.flagging != self.drawn_flagged;
+        if want != have || flag_changed {
+            self.redraw()?;
+        }
         Ok(())
     }
 
@@ -3493,6 +3547,21 @@ fn resolve_predict_mode() -> PredictMode {
     }
 }
 
+/// Resolve whether the non-destructive disconnect status line is enabled.
+/// `DOSH_DISCONNECT_STATUS` (0/false/off to disable) overrides for quick
+/// experiments; otherwise the client config's `disconnect_status` applies,
+/// defaulting to on.
+fn resolve_disconnect_status() -> bool {
+    if let Ok(env) = std::env::var("DOSH_DISCONNECT_STATUS") {
+        let v = env.trim().to_ascii_lowercase();
+        return !matches!(v.as_str(), "0" | "false" | "off" | "no");
+    }
+    match load_client_config(None) {
+        Ok(cfg) => cfg.disconnect_status,
+        Err(_) => true,
+    }
+}
+
 fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
     haystack
         .windows(needle.len())
@@ -3597,6 +3666,180 @@ fn render_frame(frame: &Frame) -> Result<()> {
     Ok(())
 }
 
+/// Seconds of silence from the server before the disconnect status line appears.
+/// Short enough to give quick feedback on a lost link, long enough that a normal
+/// idle period (the run loop only pings after 2s of quiet) never flashes it.
+const DISCONNECT_STATUS_THRESHOLD_SECS: u64 = 2;
+
+/// Non-destructive, mosh-style disconnect status line.
+///
+/// When server packets stop arriving we paint a single, unobtrusive line on the
+/// terminal's *bottom row* — e.g. reverse-video cyan `[dosh] last contact 3s ago
+/// — reconnecting…` — using save/restore cursor (ESC 7 / ESC 8) so the real
+/// application cursor never moves and full-screen TUIs are not corrupted. (The
+/// old in-band overlay wrote over the active line and corrupted TUIs; this draws
+/// only the last row, parks the cursor there, then restores, touching nothing
+/// the app owns.) The line is refreshed on the status timer tick and cleared the
+/// instant a packet resumes.
+///
+/// The decision logic and the rendered text are pure and unit-tested; only
+/// [`DisconnectStatus::emit`] / [`DisconnectStatus::emit_clear`] touch stdout.
+struct DisconnectStatus {
+    enabled: bool,
+    /// Whether a status line is currently painted on screen (so we know to clear
+    /// it and to avoid redundant repaints of identical text).
+    shown: bool,
+    /// The text currently painted, to skip no-op repaints (the elapsed seconds
+    /// only change once per second so most ticks are no-ops).
+    drawn_text: String,
+}
+
+/// What the run loop should do with the status line on a given tick or packet.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum StatusAction {
+    /// Leave the screen as-is (already correct).
+    None,
+    /// Paint/repaint the status line with this exact text on the bottom row.
+    Show(String),
+    /// Erase the status line (link recovered or feature disabled).
+    Clear,
+}
+
+impl DisconnectStatus {
+    fn new(enabled: bool) -> Self {
+        Self {
+            enabled,
+            shown: false,
+            drawn_text: String::new(),
+        }
+    }
+
+    /// Pure: render the status text for a given elapsed silence. Kept separate
+    /// from any I/O so it can be asserted in tests.
+    fn render_text(elapsed: Duration) -> String {
+        format!(
+            "[dosh] last contact {}s ago — reconnecting…",
+            elapsed.as_secs()
+        )
+    }
+
+    /// Pure decision for a status timer tick: given how long the link has been
+    /// silent, decide whether to show (and with what text), clear, or do nothing.
+    /// Does not mutate screen state; the caller applies the returned action and
+    /// then calls [`Self::applied`].
+    fn on_tick(&self, stale: Duration) -> StatusAction {
+        if !self.enabled {
+            return if self.shown {
+                StatusAction::Clear
+            } else {
+                StatusAction::None
+            };
+        }
+        if stale >= Duration::from_secs(DISCONNECT_STATUS_THRESHOLD_SECS) {
+            let text = Self::render_text(stale);
+            if self.shown && text == self.drawn_text {
+                StatusAction::None
+            } else {
+                StatusAction::Show(text)
+            }
+        } else if self.shown {
+            StatusAction::Clear
+        } else {
+            StatusAction::None
+        }
+    }
+
+    /// Pure decision for "a packet just arrived": clear any visible status line
+    /// the instant contact resumes, otherwise do nothing.
+    fn on_contact(&self) -> StatusAction {
+        if self.shown {
+            StatusAction::Clear
+        } else {
+            StatusAction::None
+        }
+    }
+
+    /// Record that an action was applied to the screen, updating internal state.
+    fn applied(&mut self, action: &StatusAction) {
+        match action {
+            StatusAction::Show(text) => {
+                self.shown = true;
+                self.drawn_text = text.clone();
+            }
+            StatusAction::Clear => {
+                self.shown = false;
+                self.drawn_text.clear();
+            }
+            StatusAction::None => {}
+        }
+    }
+
+    /// Apply an action to the real terminal (idempotent for `None`).
+    fn apply(&mut self, action: StatusAction) -> Result<()> {
+        match &action {
+            StatusAction::Show(text) => self.emit(text)?,
+            StatusAction::Clear => self.emit_clear()?,
+            StatusAction::None => {}
+        }
+        self.applied(&action);
+        Ok(())
+    }
+
+    /// Paint `text` on the terminal's bottom row using save/restore cursor so the
+    /// app's cursor is untouched. Sequence: save cursor (ESC 7), move to the last
+    /// row col 1, clear that row, set reverse-video cyan, write text, reset
+    /// attributes, restore cursor (ESC 8).
+    fn emit(&self, text: &str) -> Result<()> {
+        let (_, rows) = terminal_size();
+        let bytes = render_status_overlay(text, rows);
+        emit_status(&bytes)
+    }
+
+    /// Erase the bottom-row status line, again with save/restore cursor.
+    fn emit_clear(&self) -> Result<()> {
+        let (_, rows) = terminal_size();
+        emit_status(&render_status_clear(rows))
+    }
+}
+
+/// Pure: build the escape sequence that paints `text` on `rows`-tall terminal's
+/// bottom row with reverse-video cyan, leaving the cursor where it was.
+fn render_status_overlay(text: &str, rows: u16) -> Vec<u8> {
+    let mut out = Vec::with_capacity(text.len() + 24);
+    out.extend_from_slice(b"\x1b7"); // save cursor
+    // Move to bottom row, column 1.
+    out.extend_from_slice(format!("\x1b[{};1H", rows.max(1)).as_bytes());
+    out.extend_from_slice(b"\x1b[2K"); // clear entire row
+    out.extend_from_slice(b"\x1b[7;36m"); // reverse video + cyan
+    out.extend_from_slice(text.as_bytes());
+    out.extend_from_slice(b"\x1b[0m"); // reset attributes
+    out.extend_from_slice(b"\x1b8"); // restore cursor
+    out
+}
+
+/// Pure: build the escape sequence that clears the bottom-row status line.
+fn render_status_clear(rows: u16) -> Vec<u8> {
+    let mut out = Vec::with_capacity(12);
+    out.extend_from_slice(b"\x1b7");
+    out.extend_from_slice(format!("\x1b[{};1H", rows.max(1)).as_bytes());
+    out.extend_from_slice(b"\x1b[2K");
+    out.extend_from_slice(b"\x1b8");
+    out
+}
+
+#[cfg(not(test))]
+fn emit_status(bytes: &[u8]) -> Result<()> {
+    let mut stdout = std::io::stdout();
+    stdout.write_all(bytes)?;
+    stdout.flush()?;
+    Ok(())
+}
+
+#[cfg(test)]
+fn emit_status(_bytes: &[u8]) -> Result<()> {
+    Ok(())
+}
+
 fn cache_path(root: &str, server: &str, session: &str, mode: &str) -> std::path::PathBuf {
     let safe = format!("{server}_{session}_{mode}")
         .chars()
@@ -3655,16 +3898,18 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 #[cfg(test)]
 mod tests {
     use super::{
-        DynamicForward, FrameBuffer, LocalForward, PredictMode, Predictor, RemoteForward,
+        DisconnectStatus, DynamicForward, FrameBuffer, LocalForward, PredictMode, Predictor,
-        SshConfig, auth_allows, load_first_native_identity_with_prompt, parse_dynamic_forward,
+        RemoteForward, SshConfig, StatusAction, auth_allows,
-        parse_local_forward, parse_remote_forward, parse_ssh_config, raw_contains_host_table,
+        load_first_native_identity_with_prompt, parse_dynamic_forward, parse_local_forward,
-        recv_response_until, requested_env, ssh_destination_host, ssh_username, ssh_with_user,
+        parse_remote_forward, parse_ssh_config, raw_contains_host_table, recv_response_until,
-        startup_command, toml_bare_key_or_quoted,
+        render_status_clear, render_status_overlay, requested_env, ssh_destination_host,
+        ssh_username, ssh_with_user, startup_command, toml_bare_key_or_quoted,
     };
     use dosh::config::{ClientConfig, HostConfig};
     use dosh::native::EnvVar;
     use dosh::protocol::{self, Frame, PacketKind};
     use ed25519_dalek::{SigningKey, VerifyingKey};
+    use std::time::Duration;
 
     #[test]
     fn parses_ssh_config_hostname() {
@@ -4255,4 +4500,148 @@ mod tests {
             }
         );
     }
+
+    // --- Item 1: disconnect status line state machine ---
+
+    /// The status line stays hidden while the link is fresh and only appears once
+    /// silence crosses the threshold; the rendered text reports elapsed seconds.
+    #[test]
+    fn disconnect_status_shows_after_threshold() {
+        let mut status = DisconnectStatus::new(true);
+
+        // Fresh link: nothing to do.
+        assert_eq!(status.on_tick(Duration::from_secs(0)), StatusAction::None);
+        assert_eq!(status.on_tick(Duration::from_secs(1)), StatusAction::None);
+
+        // Crossed the threshold: show with the elapsed seconds in the text.
+        let action = status.on_tick(Duration::from_secs(3));
+        assert_eq!(
+            action,
+            StatusAction::Show("[dosh] last contact 3s ago — reconnecting…".to_string())
+        );
+        status.applied(&action);
+        assert!(status.shown);
+    }
+
+    /// Once shown, a same-second tick is a no-op but a new elapsed value repaints.
+    #[test]
+    fn disconnect_status_repaints_only_on_text_change() {
+        let mut status = DisconnectStatus::new(true);
+        let first = status.on_tick(Duration::from_secs(3));
+        status.applied(&first);
+
+        // Same elapsed second -> no redundant repaint.
+        assert_eq!(status.on_tick(Duration::from_secs(3)), StatusAction::None);
+
+        // Next second -> repaint with updated text.
+        assert_eq!(
+            status.on_tick(Duration::from_secs(4)),
+            StatusAction::Show("[dosh] last contact 4s ago — reconnecting…".to_string())
+        );
+    }
+
+    /// Contact resuming clears a shown line immediately, and a sub-threshold tick
+    /// also clears it; clearing when nothing is shown is a no-op.
+    #[test]
+    fn disconnect_status_clears_on_contact_and_recovery() {
+        let mut status = DisconnectStatus::new(true);
+
+        // Nothing shown yet: contact is a no-op.
+        assert_eq!(status.on_contact(), StatusAction::None);
+
+        let show = status.on_tick(Duration::from_secs(5));
+        status.applied(&show);
+        assert!(status.shown);
+
+        // A packet arrives: clear immediately.
+        let clear = status.on_contact();
+        assert_eq!(clear, StatusAction::Clear);
+        status.applied(&clear);
+        assert!(!status.shown);
+
+        // Show again, then a sub-threshold tick (link recovered) clears it too.
+        let show = status.on_tick(Duration::from_secs(5));
+        status.applied(&show);
+        assert_eq!(status.on_tick(Duration::from_secs(0)), StatusAction::Clear);
+    }
+
+    /// When the feature is disabled the machine never shows, and clears anything
+    /// already on screen (e.g. config toggled off at runtime).
+    #[test]
+    fn disconnect_status_disabled_never_shows() {
+        let mut status = DisconnectStatus::new(false);
+        assert_eq!(status.on_tick(Duration::from_secs(10)), StatusAction::None);
+        // Simulate a stale "shown" flag and confirm it gets cleared.
+        status.shown = true;
+        assert_eq!(status.on_tick(Duration::from_secs(10)), StatusAction::Clear);
+    }
+
+    /// The rendered overlay must be non-destructive: save cursor (ESC 7), park on
+    /// the bottom row, paint reverse-video cyan, then restore cursor (ESC 8) so
+    /// the application's cursor and full-screen TUIs are never disturbed.
+    #[test]
+    fn disconnect_status_overlay_is_save_restore_bottom_row() {
+        let bytes = render_status_overlay("[dosh] last contact 3s ago — reconnecting…", 24);
+        assert!(bytes.starts_with(b"\x1b7"), "must save cursor first");
+        assert!(bytes.ends_with(b"\x1b8"), "must restore cursor last");
+        // Positions to the last row (row 24), column 1.
+        assert!(bytes.windows(7).any(|w| w == b"\x1b[24;1H"));
+        // Clears the row and uses reverse-video (7) + cyan (36).
+        assert!(bytes.windows(4).any(|w| w == b"\x1b[2K"));
+        assert!(bytes.windows(7).any(|w| w == b"\x1b[7;36m"));
+        // Resets attributes before restoring the cursor.
+        assert!(bytes.windows(4).any(|w| w == b"\x1b[0m"));
+
+        let clear = render_status_clear(24);
+        assert!(clear.starts_with(b"\x1b7"));
+        assert!(clear.ends_with(b"\x1b8"));
+        assert!(clear.windows(4).any(|w| w == b"\x1b[2K"));
+    }
+
+    // --- Item 2: predictor policy re-evaluated on a timer tick ---
+
+    /// A latency spike must surface predictions on a *timer tick*, not only on
+    /// the next keystroke: `refresh_policy` force-shows once a prediction has been
+    /// outstanding past the glitch threshold, even though SRTT hasn't caught up.
+    #[test]
+    fn predictor_refresh_policy_force_shows_on_spike() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        // Fast SRTT so the SRTT trigger stays off; only the spike timer can show.
+        p.set_srtt_for_test(5.0);
+        p.observe_input(b"abc").unwrap();
+        assert!(p.pending_len() > 0);
+        // Just typed: not outstanding long enough yet -> still hidden.
+        assert!(!p.is_drawn());
+
+        // Simulate the spike: the prediction has now been outstanding well past
+        // the glitch threshold. A keystroke-free timer tick must reveal it.
+        p.backdate_pending_for_test(super::GLITCH_FORCE_MS as u64 + 50);
+        p.refresh_policy().unwrap();
+        assert!(p.is_drawn(), "tick should reveal prediction on a spike");
+    }
+
+    /// `refresh_policy` also turns the overlay *off* on a tick once SRTT recovers
+    /// below the trigger (no keystroke required).
+    #[test]
+    fn predictor_refresh_policy_hides_on_recovery() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        // Slow link: SRTT trigger latched on -> prediction is shown on input.
+        p.set_srtt_for_test(60.0);
+        p.observe_input(b"abc").unwrap();
+        assert!(p.is_drawn());
+
+        // Link recovers below the low trigger; a timer tick must hide it.
+        p.set_srtt_for_test(5.0);
+        p.refresh_policy().unwrap();
+        assert!(!p.is_drawn(), "tick should hide prediction after recovery");
+    }
+
+    /// An idle tick with nothing outstanding is a cheap no-op (no spurious draw).
+    #[test]
+    fn predictor_refresh_policy_idle_is_noop() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        p.set_srtt_for_test(60.0);
+        p.refresh_policy().unwrap();
+        assert!(!p.is_drawn());
+    }
 }

src/config.rs

@@ -114,6 +114,12 @@ pub struct ClientConfig {
     pub use_ssh_agent: bool,
     #[serde(default)]
     pub forward_agent: bool,
+    /// Show a non-destructive, mosh-style status line on the bottom screen row
+    /// when UDP packets stop arriving ("[dosh] last contact Ns ago …"). Drawn
+    /// with save/restore cursor so it never moves the app cursor or corrupts a
+    /// full-screen TUI, and cleared the instant packets resume. Default on.
+    #[serde(default = "default_true")]
+    pub disconnect_status: bool,
     #[serde(default = "default_send_env")]
     pub send_env: Vec<String>,
     #[serde(default)]
@@ -165,6 +171,7 @@ impl Default for ClientConfig {
             identity_files: default_identity_files(),
             use_ssh_agent: true,
             forward_agent: false,
+            disconnect_status: true,
             send_env: default_send_env(),
             set_env: HashMap::new(),
         }