Add mosh-grade predictive local echo

Replace the basic printable-only predictor with a clean-room speculative
local-echo engine modeled on the Mosh paper's prediction overlay design
(no GPL code copied or translated; original Rust written from the
algorithm).

Model:
- Per-line cell model with a local cursor; printable keystrokes echo
  immediately, dimmed (and underlined on a very slow link) via
  save/restore cursor so full-screen TUIs are never corrupted.
- Backspace and in-line Left/Right arrow prediction (CSI and SS3).
- Epoch grouping per keystroke burst; anything unmodelable (CR/LF,
  escapes, control bytes, wide chars, right margin, paste-sized bursts)
  ends the epoch safely rather than mispredicting.
- Frame-sequenced confirmation: an authoritative server frame supersedes
  and erases the overlay (dosh has no client-side emulator, so it cannot
  do Mosh's per-cell content match; frame arrival is the confirmation).
- Glitch detection: contradicting output (e.g. alternate-screen entry)
  discards the epoch.
- SRTT-based display policy with hysteresis and a latency-spike override;
  modes off / experimental(adaptive, default) / always via new
  client config `predict_mode` and `DOSH_PREDICT_MODE` env override.

Improvements over the prior predictor: in-line arrow motion, dim+flag
styling, exact-width erase, paste guard, and faster (frame-arrival)
confirmation. Prediction is display-only and never swallows input.

Adds unit tests: printable run, confirmation-clears, glitch-discards,
backspace, line-wrap budget, paste-skip, newline, inline arrows, and the
experimental SRTT threshold gate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/bin/dosh-client.rs

@@ -2074,7 +2074,15 @@ async fn run_terminal(
     let mut resize_tick = tokio::time::interval(Duration::from_millis(250));
     let mut last_size = size().unwrap_or((80, 24));
     let mut frame_buffer = FrameBuffer::default();
-    let mut predictor = Predictor::new(predict && cred.mode != "view-only" && !forward_only);
+    // Resolve the prediction display policy (off / experimental / always). An
+    // env var wins for ad-hoc tuning; otherwise the client config's
+    // `predict_mode` provides the persistent default. Predictions only run in a
+    // read-write interactive session.
+    let predict_mode = resolve_predict_mode();
+    let mut predictor = Predictor::with_mode(
+        predict && cred.mode != "view-only" && !forward_only,
+        predict_mode,
+    );
     let (forward_tx, mut forward_rx) = mpsc::channel::<ForwardEvent>(1024);
     let _forward_keepalive = if local_forwards.is_empty() {
         Some(forward_tx.clone())
@@ -2894,85 +2902,522 @@ async fn send_stream_packet(
     Ok(())
 }
 
+// ---------------------------------------------------------------------------
+// Predictive local echo (speculative echo).
+//
+// This is a clean-room Rust implementation written from the design described in
+// the Mosh paper (Winstein & Balakrishnan, "Mosh: An Interactive Remote Shell
+// for Mobile Clients", USENIX ATC 2012) and a conceptual reading of how a
+// prediction overlay behaves. No Mosh source was copied or translated; the data
+// structures, byte handling, confirmation strategy, and drawing here are
+// original to dosh.
+//
+// Why dosh's approach differs from Mosh's: Mosh runs a full terminal emulator
+// on the client and keeps a cell-accurate framebuffer, so it can confirm each
+// predicted character by comparing the predicted cell to the actual cell the
+// server produced. dosh's client has no client-side emulator -- it blits opaque
+// server byte-frames straight to the real terminal. So dosh cannot do
+// content-level confirmation. Instead it confirms predictions by *frame
+// sequencing*: when the server emits a new authoritative frame (output_seq
+// advances) that reflects the keystrokes we predicted, that frame is rendered
+// and supersedes our overlay; we erase the speculative glyphs and let the
+// server's bytes stand. To draw and erase safely without an emulator, dosh
+// keeps a tiny model of just the current line near the cursor.
+//
+// Correctness over coverage: a visibly wrong/sticky prediction is worse than no
+// prediction, so anything we cannot model with confidence (escape sequences,
+// control chars other than backspace, multi-byte / wide chars, the right
+// margin) ends the current epoch and stops further speculation until the next
+// confirmation re-bases us against authoritative output.
+
+/// Display policy for predictions, mirroring Mosh's off / adaptive / always
+/// design (we call the adaptive mode "experimental" to match Mosh's naming for
+/// the SRTT-gated mode that only shows predictions on a laggy link).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum PredictMode {
+    /// Never show predictions.
+    Off,
+    /// Show predictions only when the link is laggy (SRTT above a trigger).
+    Experimental,
+    /// Always show predictions whenever any are outstanding.
+    Always,
+}
+
+impl PredictMode {
+    fn parse(s: &str) -> Self {
+        match s.trim().to_ascii_lowercase().as_str() {
+            "off" | "never" | "false" | "no" => PredictMode::Off,
+            "always" => PredictMode::Always,
+            // "experimental" / "adaptive" / "auto" / anything else -> adaptive.
+            _ => PredictMode::Experimental,
+        }
+    }
+}
+
+/// SRTT (in milliseconds) above which the experimental mode begins showing
+/// predictions, and below which it stops. Hysteresis avoids flapping on a link
+/// hovering around the threshold. On a faster-than-this link, native local echo
+/// already feels instant, so speculation only adds risk.
+const SRTT_TRIGGER_HIGH_MS: f64 = 30.0;
+const SRTT_TRIGGER_LOW_MS: f64 = 20.0;
+
+/// SRTT (ms) above which we additionally underline ("flag") predicted cells so
+/// the user can tell speculative glyphs from confirmed ones on a very slow link.
+const FLAG_TRIGGER_HIGH_MS: f64 = 80.0;
+const FLAG_TRIGGER_LOW_MS: f64 = 50.0;
+
+/// If a prediction is still outstanding after this long, treat the link as
+/// laggy enough to force display even if the SRTT estimate hasn't caught up yet
+/// (covers a sudden latency spike on the very first slow keystroke).
+const GLITCH_FORCE_MS: u128 = 250;
+
+/// One speculatively-echoed character on the current line.
+#[derive(Clone, Debug)]
+struct PredictedCell {
+    /// The byte we drew (always a single printable ASCII byte, 0x20..=0x7e).
+    byte: u8,
+    /// Epoch this prediction belongs to (a keystroke burst).
+    epoch: u64,
+}
+
+/// Predictive local-echo engine. See the module comment above for the model.
 struct Predictor {
+    mode: PredictMode,
     enabled: bool,
+    /// True while the server is in the alternate screen (a full-screen TUI such
+    /// as vim/htop); we never speculate there because we cannot model arbitrary
+    /// cursor addressing safely.
     alternate_screen: bool,
-    pending: Vec<u8>,
+
+    /// Predicted cells for the current line, left-to-right starting at the
+    /// column where the cursor sat when this run of predictions began.
+    cells: Vec<PredictedCell>,
+    /// Cursor offset, in cells, from the start of `cells`. Equals `cells.len()`
+    /// while typing at the end of the line; can be less after Left-arrow or
+    /// backspace within the predicted span.
+    cursor: usize,
+    /// Current keystroke-burst epoch. Bumped whenever we hit something we cannot
+    /// model (CR/LF, escape, control byte, wide char, right margin) so those
+    /// speculative cells become "tentative" and stop being drawn.
+    epoch: u64,
+    /// Highest epoch confirmed by authoritative server output. Cells whose epoch
+    /// is greater than this are tentative and are not displayed.
+    confirmed_epoch: u64,
+
+    /// How many columns the current overlay occupies on screen (0 = nothing
+    /// drawn). Used to erase exactly what was painted before re-rendering.
+    drawn_width: usize,
+    /// Whether the last draw underlined the cells (flagging on).
+    drawn_flagged: bool,
+
+    // --- SRTT estimation (display-only; never gates whether input is sent) ---
+    /// Smoothed round-trip time estimate in milliseconds, or None until we have
+    /// a sample.
+    srtt_ms: Option<f64>,
+    /// When the oldest still-outstanding prediction was made. Used both to
+    /// sample SRTT on the next frame and to force display on a latency spike.
+    oldest_pending_at: Option<Instant>,
+    /// Hysteresis latches.
+    srtt_trigger: bool,
+    flagging: bool,
 }
 
 impl Predictor {
+    /// Construct with the default adaptive (experimental) display policy.
+    #[cfg(test)]
     fn new(enabled: bool) -> Self {
+        Self::with_mode(enabled, PredictMode::Experimental)
+    }
+
+    fn with_mode(enabled: bool, mode: PredictMode) -> Self {
         Self {
-            enabled,
+            mode,
+            enabled: enabled && mode != PredictMode::Off,
             alternate_screen: false,
-            pending: Vec::new(),
+            cells: Vec::new(),
+            cursor: 0,
+            epoch: 1,
+            confirmed_epoch: 0,
+            drawn_width: 0,
+            drawn_flagged: false,
+            srtt_ms: None,
+            oldest_pending_at: None,
+            srtt_trigger: false,
+            flagging: false,
         }
     }
 
+    /// Drop all speculative state. Called on reconnect/resume and screen resize,
+    /// where any cached line model would be stale.
     fn reset(&mut self) {
-        self.pending.clear();
+        let _ = self.erase_drawn();
+        self.cells.clear();
+        self.cursor = 0;
+        self.epoch += 1;
+        self.confirmed_epoch = self.epoch - 1;
         self.alternate_screen = false;
+        self.oldest_pending_at = None;
     }
 
+    /// Number of speculative cells currently outstanding (active in any epoch).
+    #[cfg(test)]
+    fn pending_len(&self) -> usize {
+        self.cells.len()
+    }
+
+    /// End the current keystroke burst (Mosh's "become tentative"). In dosh's
+    /// frame-confirmation model we cannot content-verify lingering cells, so the
+    /// safe choice is to drop the current epoch's speculative cells outright and
+    /// start a fresh epoch. This is how we bail out on anything we cannot model
+    /// (escape sequences, CR/LF, control bytes, wide chars, the right margin)
+    /// without risking a stale/wrong glyph.
+    fn become_tentative(&mut self) {
+        let _ = self.erase_drawn();
+        self.cells.clear();
+        self.cursor = 0;
+        self.epoch += 1;
+        self.oldest_pending_at = None;
+    }
+
+    /// Observe raw input bytes that are *also* being sent to the server. This is
+    /// display-only and never consumes input. The caller still forwards `bytes`
+    /// unconditionally.
     fn observe_input(&mut self, bytes: &[u8]) -> Result<()> {
         if !self.enabled || self.alternate_screen {
             return Ok(());
         }
-        if !Self::predictable(bytes) {
+        // A large burst is almost certainly a paste, not interactive typing.
-            self.clear_pending()?;
+        // Predicting it would risk wrapping a long dim run across many lines, so
-            return Ok(());
+        // we bail: drop any overlay and let the authoritative frame draw it.
+        if bytes.len() > PREDICT_BURST_LIMIT {
+            self.become_tentative();
+            return self.redraw();
         }
-        self.pending.extend_from_slice(bytes);
+        let mut i = 0;
-        self.draw_pending()
+        while i < bytes.len() {
+            let b = bytes[i];
+            match b {
+                // Printable ASCII: append a predicted cell at the cursor.
+                0x20..=0x7e => {
+                    self.predict_char(b);
+                    i += 1;
+                }
+                // Backspace (^H) or DEL (^?): remove the previous predicted cell.
+                0x08 | 0x7f => {
+                    self.predict_backspace();
+                    i += 1;
+                }
+                // ESC: try to recognize a cursor-motion CSI/SS3 we can model
+                // (Left/Right within the line); otherwise become tentative.
+                0x1b => {
+                    let consumed = self.predict_escape(&bytes[i..]);
+                    if consumed == 0 {
+                        // Unrecognized / incomplete escape: bail safely.
+                        self.become_tentative();
+                        i = bytes.len();
+                    } else {
+                        i += consumed;
+                    }
+                }
+                // CR / LF and any other control byte: we cannot model where the
+                // cursor lands (shell prompt, scroll, etc.), so end the epoch.
+                _ => {
+                    self.become_tentative();
+                    i += 1;
+                }
+            }
+        }
+        if self.oldest_pending_at.is_none() && !self.cells.is_empty() {
+            self.oldest_pending_at = Some(Instant::now());
+        }
+        self.redraw()
     }
 
+    /// Predict one printable character at the current cursor position.
+    fn predict_char(&mut self, byte: u8) {
+        // Right margin is tricky (shells, editors, and wrap behavior differ), so
+        // we refuse to predict past a conservative column budget and bail out
+        // instead of risking a wrong glyph at a wrap point.
+        if self.cells.len() >= PREDICT_MAX_LINE {
+            self.become_tentative();
+            return;
+        }
+        let cell = PredictedCell {
+            byte,
+            epoch: self.epoch,
+        };
+        if self.cursor < self.cells.len() {
+            // Typing over an existing predicted cell (after a Left arrow):
+            // overwrite in place, as a terminal in insert-off mode would.
+            self.cells[self.cursor] = cell;
+        } else {
+            self.cells.push(cell);
+        }
+        self.cursor += 1;
+    }
+
+    /// Predict a backspace: erase the previous predicted cell if we have one.
+    fn predict_backspace(&mut self) {
+        if self.cursor == 0 {
+            // We'd be backspacing past the start of our predicted span into
+            // territory we don't model -> bail safely.
+            self.become_tentative();
+            return;
+        }
+        self.cursor -= 1;
+        // Only safe to shrink the line if we're at its end; otherwise a
+        // mid-line backspace shifts everything left, which we don't model, so
+        // bail.
+        if self.cursor == self.cells.len() - 1 {
+            self.cells.pop();
+        } else {
+            self.become_tentative();
+        }
+    }
+
+    /// Recognize Left/Right cursor motion escape sequences we can model and
+    /// apply them to the local cursor. Returns the number of bytes consumed, or
+    /// 0 if `data` does not start with a sequence we handle.
+    ///
+    /// Improvement over the previous dosh predictor (which bailed on any escape)
+    /// and parity with Mosh: predicting in-line arrow-key motion so cursor
+    /// repositioning feels instant too.
+    fn predict_escape(&mut self, data: &[u8]) -> usize {
+        // Both CSI (ESC [) and application-cursor SS3 (ESC O) are used for arrows.
+        if data.len() < 3 || data[0] != 0x1b {
+            return 0;
+        }
+        if data[1] != b'[' && data[1] != b'O' {
+            return 0;
+        }
+        match data[2] {
+            // Right arrow: only within already-predicted cells.
+            b'C' if self.cursor < self.cells.len() => {
+                self.cursor += 1;
+                3
+            }
+            // Left arrow.
+            b'D' if self.cursor > 0 => {
+                self.cursor -= 1;
+                3
+            }
+            // Recognized arrow but at the edge of our predicted span, or
+            // Up/Down/Home/End/etc. which cross lines or jump unpredictably:
+            // refuse so the caller becomes tentative rather than mispredicting.
+            _ => 0,
+        }
+    }
+
+    /// Observe authoritative server output. Detects alternate-screen transitions
+    /// and confirms outstanding predictions: a fresh frame means the server has
+    /// re-rendered the line, so our speculative glyphs are superseded and erased.
     fn observe_output(&mut self, bytes: &[u8]) {
         if contains_bytes(bytes, b"\x1b[?1049h")
             || contains_bytes(bytes, b"\x1b[?1047h")
             || contains_bytes(bytes, b"\x1b[?47h")
         {
             self.alternate_screen = true;
-            self.pending.clear();
+            let _ = self.discard_all();
         }
         if contains_bytes(bytes, b"\x1b[?1049l")
             || contains_bytes(bytes, b"\x1b[?1047l")
             || contains_bytes(bytes, b"\x1b[?47l")
         {
             self.alternate_screen = false;
-            self.pending.clear();
+            let _ = self.discard_all();
         }
     }
 
+    /// Sample SRTT from a newly arrived frame and confirm/clear outstanding
+    /// predictions. Called once per accepted authoritative frame, *before* the
+    /// frame's bytes are written to the terminal so the server render lands on a
+    /// clean line. Returns nothing; drawing is handled by the caller's render
+    /// followed by `redraw` on the next input.
+    fn confirm_with_frame(&mut self) -> Result<()> {
+        if let Some(sent_at) = self.oldest_pending_at.take() {
+            let sample = sent_at.elapsed().as_secs_f64() * 1000.0;
+            self.update_srtt(sample);
+        }
+        // The frame is authoritative for the line; clear our overlay so the
+        // server's bytes are the only thing on screen. This is dosh's
+        // confirmation: frame arrival == the predicted keystrokes are now echoed
+        // by the server. Faster than waiting for per-cell content match.
+        self.discard_all()
+    }
+
+    fn update_srtt(&mut self, sample_ms: f64) {
+        // Standard exponentially weighted moving average (RFC 6298 style alpha).
+        const ALPHA: f64 = 0.125;
+        self.srtt_ms = Some(match self.srtt_ms {
+            None => sample_ms,
+            Some(prev) => (1.0 - ALPHA) * prev + ALPHA * sample_ms,
+        });
+    }
+
+    /// Test seam: pin the SRTT estimate and refresh the display triggers so the
+    /// experimental-mode threshold behavior can be exercised deterministically.
+    #[cfg(test)]
+    fn set_srtt_for_test(&mut self, ms: f64) {
+        self.srtt_ms = Some(ms);
+        self.update_triggers();
+    }
+
+    /// Whether we should *display* predictions right now under the active policy.
+    fn should_display(&self) -> bool {
+        match self.mode {
+            PredictMode::Off => false,
+            PredictMode::Always => true,
+            PredictMode::Experimental => {
+                if self.srtt_trigger {
+                    return true;
+                }
+                // Force display on a latency spike even before SRTT catches up.
+                if let Some(at) = self.oldest_pending_at
+                    && at.elapsed().as_millis() >= GLITCH_FORCE_MS
+                {
+                    return true;
+                }
+                false
+            }
+        }
+    }
+
+    /// Update the SRTT/flag hysteresis latches from the current estimate.
+    fn update_triggers(&mut self) {
+        let srtt = self.srtt_ms.unwrap_or(0.0);
+        if srtt > SRTT_TRIGGER_HIGH_MS {
+            self.srtt_trigger = true;
+        } else if srtt <= SRTT_TRIGGER_LOW_MS {
+            self.srtt_trigger = false;
+        }
+        if srtt > FLAG_TRIGGER_HIGH_MS {
+            self.flagging = true;
+        } else if srtt <= FLAG_TRIGGER_LOW_MS {
+            self.flagging = false;
+        }
+    }
+
+    /// The active (displayable) predicted bytes: cells in confirmed-or-current
+    /// epochs only, drawn from the start of the line up to the cursor budget.
+    fn visible_bytes(&self) -> Vec<u8> {
+        self.cells
+            .iter()
+            .filter(|c| c.epoch > self.confirmed_epoch || c.epoch == self.epoch)
+            .map(|c| c.byte)
+            .collect()
+    }
+
+    /// Repaint the prediction overlay: erase the old one, draw the new one if
+    /// policy says so. Uses save/restore cursor so it never moves the real
+    /// cursor and, when nothing should show, leaves the terminal untouched.
+    fn redraw(&mut self) -> Result<()> {
+        self.update_triggers();
+        // Always erase whatever we drew before so we never leave a stale glyph.
+        self.erase_drawn()?;
+        if !self.should_display() {
+            return Ok(());
+        }
+        let bytes = self.visible_bytes();
+        if bytes.is_empty() {
+            return Ok(());
+        }
+        let flag = self.flagging;
+        // Save cursor, dim (and optionally underline) predicted glyphs, draw,
+        // reset attributes, restore cursor.
+        let mut out = Vec::with_capacity(bytes.len() + 12);
+        out.extend_from_slice(b"\x1b7");
+        out.extend_from_slice(if flag { b"\x1b[2;4m" } else { b"\x1b[2m" });
+        out.extend_from_slice(&bytes);
+        out.extend_from_slice(b"\x1b[0m\x1b8");
+        emit_overlay(&out)?;
+        // Record exactly how many columns we painted so erasure overwrites the
+        // 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(())
+    }
+
+    /// Erase the currently drawn overlay (if any) by overwriting the previously
+    /// painted columns with spaces, using save/restore cursor so the real cursor
+    /// and surrounding text are untouched.
+    fn erase_drawn(&mut self) -> Result<()> {
+        let width = self.drawn_width;
+        self.drawn_width = 0;
+        if width == 0 {
+            return Ok(());
+        }
+        let mut out = Vec::with_capacity(width + 4);
+        out.extend_from_slice(b"\x1b7");
+        out.resize(out.len() + width, b' ');
+        out.extend_from_slice(b"\x1b8");
+        emit_overlay(&out)
+    }
+
+    /// Erase the overlay and drop all speculative state, advancing the confirmed
+    /// epoch so nothing lingers.
+    fn discard_all(&mut self) -> Result<()> {
+        self.erase_drawn()?;
+        self.cells.clear();
+        self.cursor = 0;
+        self.epoch += 1;
+        self.confirmed_epoch = self.epoch - 1;
+        self.oldest_pending_at = None;
+        Ok(())
+    }
+
+    /// Erase the overlay without dropping the SRTT estimate or model; used right
+    /// before an authoritative frame is rendered so its bytes land cleanly.
+    /// (Kept as the public name the run loop calls.)
+    fn clear_pending(&mut self) -> Result<()> {
+        self.confirm_with_frame()
+    }
+
+    /// Whether a byte sequence is a plain printable run. Retained as a small
+    /// predicate exercised by the unit tests.
+    #[cfg(test)]
     fn predictable(bytes: &[u8]) -> bool {
         !bytes.is_empty() && bytes.iter().all(|byte| matches!(byte, 0x20..=0x7e))
     }
-
-    fn draw_pending(&self) -> Result<()> {
-        if self.pending.is_empty() {
-            return Ok(());
 }
+
+/// Conservative cap on how many predicted cells we keep on one line before we
+/// stop speculating (a backstop against runaway predictions; the right-margin /
+/// wrap point is genuinely ambiguous without a client-side emulator).
+const PREDICT_MAX_LINE: usize = 256;
+
+/// Largest single input burst we treat as interactive typing. Anything larger
+/// is assumed to be a paste and is not speculated (would risk a long dim run
+/// wrapping across lines before the server frame supersedes it).
+const PREDICT_BURST_LIMIT: usize = 32;
+
+/// Write a prepared overlay byte sequence to the real terminal. Suppressed
+/// under `cfg(test)` so unit tests exercise the prediction model without
+/// emitting escape sequences to the test runner's terminal.
+#[cfg(not(test))]
+fn emit_overlay(bytes: &[u8]) -> Result<()> {
     let mut stdout = std::io::stdout();
-        stdout.write_all(b"\x1b7\x1b[4m")?;
+    stdout.write_all(bytes)?;
-        stdout.write_all(&self.pending)?;
-        stdout.write_all(b"\x1b[24m\x1b8")?;
     stdout.flush()?;
     Ok(())
 }
 
-    fn clear_pending(&mut self) -> Result<()> {
+#[cfg(test)]
-        if self.pending.is_empty() {
+fn emit_overlay(_bytes: &[u8]) -> Result<()> {
-            return Ok(());
-        }
-        let mut stdout = std::io::stdout();
-        stdout.write_all(b"\x1b7")?;
-        for _ in 0..self.pending.len() {
-            stdout.write_all(b" ")?;
-        }
-        stdout.write_all(b"\x1b8")?;
-        stdout.flush()?;
-        self.pending.clear();
     Ok(())
 }
+
+/// Resolve the prediction display policy. `DOSH_PREDICT_MODE` (off / experimental
+/// / always) overrides for quick experiments; otherwise the client config's
+/// `predict_mode` is used, defaulting to the adaptive (experimental) policy.
+fn resolve_predict_mode() -> PredictMode {
+    if let Ok(env) = std::env::var("DOSH_PREDICT_MODE") {
+        return PredictMode::parse(&env);
+    }
+    match load_client_config(None) {
+        Ok(cfg) => PredictMode::parse(&cfg.predict_mode),
+        Err(_) => PredictMode::Experimental,
+    }
 }
 
 fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
@@ -3137,8 +3582,8 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 #[cfg(test)]
 mod tests {
     use super::{
-        DynamicForward, FrameBuffer, LocalForward, Predictor, RemoteForward, SshConfig,
+        DynamicForward, FrameBuffer, LocalForward, PredictMode, Predictor, RemoteForward,
-        auth_allows, load_first_native_identity_with_prompt, parse_dynamic_forward,
+        SshConfig, auth_allows, load_first_native_identity_with_prompt, parse_dynamic_forward,
         parse_local_forward, parse_remote_forward, parse_ssh_config, raw_contains_host_table,
         recv_response_until, requested_env, ssh_destination_host, ssh_username, ssh_with_user,
         startup_command, toml_bare_key_or_quoted,
@@ -3514,6 +3959,170 @@ mod tests {
         assert!(!predictor.alternate_screen);
     }
 
+    /// Typing printable characters builds an in-order predicted line and the
+    /// local cursor tracks the end of it.
+    #[test]
+    fn predictor_predicts_printable_run() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        p.observe_input(b"hi").unwrap();
+        assert_eq!(p.visible_bytes(), b"hi");
+        assert_eq!(p.cursor, 2);
+        assert_eq!(p.pending_len(), 2);
+    }
+
+    /// An authoritative server frame confirms (and clears) outstanding
+    /// predictions: after a frame the overlay is empty.
+    #[test]
+    fn predictor_confirmation_clears_prediction() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        p.observe_input(b"abc").unwrap();
+        assert_eq!(p.pending_len(), 3);
+
+        // Server echoes the keystrokes; a new frame arrives.
+        p.clear_pending().unwrap(); // run loop calls this before render_frame
+        assert_eq!(p.pending_len(), 0);
+        assert!(p.visible_bytes().is_empty());
+    }
+
+    /// A glitch -- server output that contradicts the prediction (here the
+    /// server enters the alternate screen mid-burst) -- discards the epoch's
+    /// predictions so nothing wrong is left on screen.
+    #[test]
+    fn predictor_glitch_discards_epoch() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        p.observe_input(b"vim").unwrap();
+        assert_eq!(p.pending_len(), 3);
+        let epoch_before = p.epoch;
+
+        // Server diverges: enters a full-screen TUI. Our line prediction is now
+        // meaningless and must be dropped.
+        p.observe_output(b"\x1b[?1049h");
+        assert!(p.alternate_screen);
+        assert_eq!(p.pending_len(), 0);
+        // The contradicted epoch is fully retired (confirmed past).
+        assert!(p.confirmed_epoch >= epoch_before);
+
+        // While in the alternate screen we refuse to speculate at all.
+        p.observe_input(b"x").unwrap();
+        assert_eq!(p.pending_len(), 0);
+    }
+
+    /// Backspace prediction removes the last predicted cell from the end of the
+    /// line and moves the local cursor back.
+    #[test]
+    fn predictor_predicts_backspace() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        p.observe_input(b"hello").unwrap();
+        assert_eq!(p.visible_bytes(), b"hello");
+
+        p.observe_input(b"\x7f").unwrap(); // DEL
+        assert_eq!(p.visible_bytes(), b"hell");
+        assert_eq!(p.cursor, 4);
+
+        p.observe_input(&[0x08]).unwrap(); // ^H
+        assert_eq!(p.visible_bytes(), b"hel");
+        assert_eq!(p.cursor, 3);
+
+        // Backspacing past the start of our predicted span bails safely (we
+        // don't model what's to the left), ending the epoch with no glyphs.
+        p.observe_input(b"\x7f\x7f\x7f\x7f").unwrap();
+        assert_eq!(p.pending_len(), 0);
+    }
+
+    /// Right-margin / line-wrap edge: we refuse to predict past a conservative
+    /// per-line budget instead of risking a wrong glyph at the wrap point.
+    /// Bytes are fed in small interactive bursts so the per-line budget (not the
+    /// paste guard) is what trips.
+    #[test]
+    fn predictor_line_wrap_edge_bails_at_budget() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        let chunk = vec![b'x'; super::PREDICT_BURST_LIMIT];
+        let mut typed = 0;
+        while typed + chunk.len() <= super::PREDICT_MAX_LINE {
+            p.observe_input(&chunk).unwrap();
+            typed += chunk.len();
+        }
+        assert_eq!(p.pending_len(), super::PREDICT_MAX_LINE);
+
+        // One more character crosses the budget -> bail, drop the epoch.
+        p.observe_input(b"y").unwrap();
+        assert_eq!(p.pending_len(), 0);
+    }
+
+    /// A paste-sized burst is not speculated at all (no overlay), but the input
+    /// is still forwarded by the caller -- prediction is display-only.
+    #[test]
+    fn predictor_skips_paste_sized_burst() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        let paste = vec![b'a'; super::PREDICT_BURST_LIMIT + 1];
+        p.observe_input(&paste).unwrap();
+        assert_eq!(p.pending_len(), 0);
+    }
+
+    /// Carriage return / newline cannot be modeled (prompt, scroll), so it ends
+    /// the epoch and clears the speculative line.
+    #[test]
+    fn predictor_newline_ends_epoch() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        p.observe_input(b"ls").unwrap();
+        assert_eq!(p.pending_len(), 2);
+        p.observe_input(b"\r").unwrap();
+        assert_eq!(p.pending_len(), 0);
+    }
+
+    /// In-line arrow-key motion (improvement over the old printable-only
+    /// predictor): Left/Right move the local cursor within the predicted span
+    /// and a following keystroke overwrites in place.
+    #[test]
+    fn predictor_predicts_inline_arrows() {
+        let mut p = Predictor::with_mode(true, PredictMode::Always);
+        p.observe_input(b"abc").unwrap();
+        assert_eq!(p.cursor, 3);
+        p.observe_input(b"\x1b[D").unwrap(); // left
+        assert_eq!(p.cursor, 2);
+        p.observe_input(b"\x1bOD").unwrap(); // left (application/SS3)
+        assert_eq!(p.cursor, 1);
+        p.observe_input(b"\x1b[C").unwrap(); // right
+        assert_eq!(p.cursor, 2);
+        // Overwrite the cell at the cursor (index 2, the 'c') in place.
+        p.observe_input(b"Z").unwrap();
+        assert_eq!(p.visible_bytes(), b"abZ");
+        assert_eq!(p.cursor, 3);
+    }
+
+    /// Experimental (adaptive) mode shows no prediction while SRTT is below the
+    /// trigger, and shows it once SRTT rises above the trigger.
+    #[test]
+    fn predictor_experimental_respects_srtt_threshold() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        // Fast link: no fresh prediction yet, no SRTT sample -> not displayed.
+        p.set_srtt_for_test(5.0);
+        p.observe_input(b"abc").unwrap();
+        assert!(p.pending_len() > 0); // model still tracks the prediction...
+        assert!(!p.should_display()); // ...but policy hides it on a fast link.
+
+        // Slow link: SRTT above the high trigger -> predictions are displayed.
+        p.set_srtt_for_test(60.0);
+        assert!(p.should_display());
+    }
+
+    /// Always mode displays regardless of SRTT; Off never does.
+    #[test]
+    fn predictor_mode_policies() {
+        let mut always = Predictor::with_mode(true, PredictMode::Always);
+        always.observe_input(b"x").unwrap();
+        assert!(always.should_display());
+
+        let off = Predictor::with_mode(true, PredictMode::Off);
+        assert!(!off.enabled);
+        assert!(!off.should_display());
+
+        assert_eq!(PredictMode::parse("off"), PredictMode::Off);
+        assert_eq!(PredictMode::parse("ALWAYS"), PredictMode::Always);
+        assert_eq!(PredictMode::parse("adaptive"), PredictMode::Experimental);
+        assert_eq!(PredictMode::parse("anything"), PredictMode::Experimental);
+    }
+
     #[test]
     fn startup_command_joins_args_for_remote_shell() {
         assert_eq!(startup_command(&[]), None);

src/config.rs

@@ -84,6 +84,10 @@ pub struct ClientConfig {
     pub view_only: bool,
     #[serde(default)]
     pub predict: bool,
+    /// Prediction display policy: "off", "experimental" (adaptive, the default),
+    /// or "always". Controls when speculative local echo is shown.
+    #[serde(default = "default_predict_mode")]
+    pub predict_mode: String,
     pub cache_attach_tickets: bool,
     pub credential_cache: String,
     #[serde(default = "default_auth_preference")]
@@ -141,6 +145,7 @@ impl Default for ClientConfig {
             reconnect_timeout_secs: 5,
             view_only: false,
             predict: false,
+            predict_mode: default_predict_mode(),
             cache_attach_tickets: true,
             credential_cache: "~/.local/share/dosh/credentials".to_string(),
             auth_preference: default_auth_preference(),
@@ -183,6 +188,10 @@ fn default_native_auth_timeout_ms() -> u64 {
     700
 }
 
+fn default_predict_mode() -> String {
+    "experimental".to_string()
+}
+
 fn default_known_hosts() -> String {
     "~/.config/dosh/known_hosts".to_string()
 }