fix: speed up alternate-screen history reads (#2426)

refs #2387

4 files changed, 773 insertions, 115 deletions

diff --git a/src/pane.rs b/src/pane.rs
index 44f3fea9..a47e62eb 100644
--- a/src/pane.rs
+++ b/src/pane.rs
@@ -1039,6 +1039,7 @@ pub struct PaneRuntime {
     reported_cwd: Arc<Mutex<Option<std::path::PathBuf>>>,
     child_wait_completed: Option<Arc<AtomicBool>>,
     kitty_keyboard_flags: Arc<AtomicU16>,
+    content_seq: Arc<AtomicU64>,
     detection_content_seq: Arc<AtomicU64>,
     full_lifecycle_authority_active: Arc<AtomicBool>,
     detect_reset_notify: Arc<Notify>,
@@ -1915,6 +1916,7 @@ impl PaneRuntime {
         let child_pid = Arc::new(AtomicU32::new(child_pid));
         let reported_cwd = Arc::new(Mutex::new(None));
         let kitty_keyboard_flags = Arc::new(AtomicU16::new(keyboard_protocol_flags));
+        let content_seq = Arc::new(AtomicU64::new(0));
         let detection_content_seq = Arc::new(AtomicU64::new(0));
 
         let io = {
@@ -1922,6 +1924,7 @@ impl PaneRuntime {
             let response_writer = response_tx.clone();
             let render_notify = render_notify.clone();
             let render_dirty = render_dirty.clone();
+            let content_seq = content_seq.clone();
             let detection_content_seq = detection_content_seq.clone();
             let child_pid = child_pid.clone();
             let read_events = events.clone();
@@ -1929,9 +1932,11 @@ impl PaneRuntime {
             let rt = tokio::runtime::Handle::current();
             let delay_rt = rt.clone();
             let on_read = Box::new(move |bytes: &[u8]| {
+                content_seq.fetch_add(1, Ordering::AcqRel);
                 let shell_pid = child_pid.load(Ordering::Acquire);
                 let result =
                     terminal.process_pty_bytes(pane_id, shell_pid, bytes, &response_writer);
+                content_seq.fetch_add(1, Ordering::Release);
                 publish_terminal_bells(pane_id, result.terminal_bells, &read_events);
                 observe_detection_content_change(bytes, &detection_content_seq);
                 let title_requested =
@@ -1999,6 +2004,7 @@ impl PaneRuntime {
             reported_cwd,
             child_wait_completed: None,
             kitty_keyboard_flags,
+            content_seq,
             detection_content_seq,
             full_lifecycle_authority_active,
             detect_reset_notify,
@@ -2054,6 +2060,7 @@ impl PaneRuntime {
         let child_pid = Arc::new(AtomicU32::new(0));
         let reported_cwd = Arc::new(Mutex::new(None));
         let child_wait_completed = Arc::new(AtomicBool::new(false));
+        let content_seq = Arc::new(AtomicU64::new(0));
         let detection_content_seq = Arc::new(AtomicU64::new(0));
         let full_lifecycle_authority_active = Arc::new(AtomicBool::new(false));
         {
@@ -2087,15 +2094,18 @@ impl PaneRuntime {
             let response_writer = response_tx.clone();
             let render_notify = render_notify.clone();
             let render_dirty = render_dirty.clone();
+            let content_seq = content_seq.clone();
             let detection_content_seq = detection_content_seq.clone();
             let child_pid = child_pid.clone();
             let events = events.clone();
             let reported_cwd = reported_cwd.clone();
             let rt = tokio::runtime::Handle::current();
             let on_read = Box::new(move |bytes: &[u8]| {
+                content_seq.fetch_add(1, Ordering::AcqRel);
                 let shell_pid = child_pid.load(Ordering::Acquire);
                 let result =
                     terminal.process_pty_bytes(pane_id, shell_pid, bytes, &response_writer);
+                content_seq.fetch_add(1, Ordering::Release);
                 publish_terminal_bells(pane_id, result.terminal_bells, &events);
                 if agent_detection == AgentDetection::Enabled {
                     observe_detection_content_change(bytes, &detection_content_seq);
@@ -2545,6 +2555,7 @@ impl PaneRuntime {
             reported_cwd,
             child_wait_completed: Some(child_wait_completed),
             kitty_keyboard_flags,
+            content_seq,
             detection_content_seq,
             full_lifecycle_authority_active,
             detect_reset_notify,
@@ -2592,6 +2603,10 @@ impl PaneRuntime {
         (rows, cols)
     }
 
+    pub(crate) fn content_seq(&self) -> u64 {
+        self.content_seq.load(Ordering::Acquire)
+    }
+
     /// Resize if the dimensions actually changed.
     pub fn resize(&self, rows: u16, cols: u16, cell_width_px: u32, cell_height_px: u32) {
         let rows = rows.max(2);
@@ -3001,8 +3016,10 @@ impl PaneRuntime {
     }
 
     pub(crate) fn test_process_pty_bytes(&self, bytes: &[u8]) {
+        self.content_seq.fetch_add(1, Ordering::AcqRel);
         let (tx, _rx) = mpsc::channel(1);
         let _ = self.terminal.process_pty_bytes(self.pane_id, 0, bytes, &tx);
+        self.content_seq.fetch_add(1, Ordering::Release);
     }
 
     pub(crate) fn test_with_scrollback_bytes(
@@ -3042,6 +3059,7 @@ impl PaneRuntime {
                 reported_cwd: Arc::new(Mutex::new(None)),
                 child_wait_completed: None,
                 kitty_keyboard_flags: Arc::new(AtomicU16::new(0)),
+                content_seq: Arc::new(AtomicU64::new(0)),
                 detection_content_seq: Arc::new(AtomicU64::new(0)),
                 full_lifecycle_authority_active: Arc::new(AtomicBool::new(false)),
                 detect_reset_notify: Arc::new(Notify::new()),
@@ -3596,6 +3614,7 @@ mod tests {
             reported_cwd: Arc::new(Mutex::new(None)),
             child_wait_completed: None,
             kitty_keyboard_flags: Arc::new(AtomicU16::new(0)),
+            content_seq: Arc::new(AtomicU64::new(0)),
             detection_content_seq: Arc::new(AtomicU64::new(0)),
             full_lifecycle_authority_active: Arc::new(AtomicBool::new(false)),
             detect_reset_notify: Arc::new(Notify::new()),
@@ -3627,6 +3646,7 @@ mod tests {
             reported_cwd: Arc::new(Mutex::new(None)),
             child_wait_completed: None,
             kitty_keyboard_flags: Arc::new(AtomicU16::new(0)),
+            content_seq: Arc::new(AtomicU64::new(0)),
             detection_content_seq: Arc::new(AtomicU64::new(0)),
             full_lifecycle_authority_active: Arc::new(AtomicBool::new(false)),
             detect_reset_notify: Arc::new(Notify::new()),
diff --git a/src/server/alt_screen_read.rs b/src/server/alt_screen_read.rs
index 425d4a41..28ea789f 100644
--- a/src/server/alt_screen_read.rs
+++ b/src/server/alt_screen_read.rs
@@ -8,19 +8,20 @@ use tracing::debug;
 use crate::api::schema::{PaneReadResult, ResponseResult, SuccessResponse};
 use crate::terminal::{ScreenSnapshot, TerminalId, TerminalRuntime, UpwardMerge};
 
-const STEP_SETTLE: Duration = Duration::from_millis(120);
+const INITIAL_QUIET: Duration = Duration::from_millis(10);
+const OUTPUT_QUIET: Duration = Duration::from_millis(10);
+const STEP_TIMEOUT: Duration = Duration::from_millis(120);
 const MAX_DURATION: Duration = Duration::from_secs(15);
 const MAX_RESTORE_DURATION: Duration = Duration::from_secs(5);
-const MAX_UNALIGNED_CHECKS: u8 = 4;
 const WHEEL_STEP_EVENTS: usize = 3;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum Phase {
-    SettleInitial { checks: u8 },
+    SettleInitial,
     ProbeBottom,
     RestoreProbe,
-    Harvest { unaligned_checks: u8 },
-    Restore { stable_checks: u8 },
+    Harvest,
+    Restore,
 }
 
 pub(crate) struct PendingAltScreenRead {
@@ -36,8 +37,13 @@ pub(crate) struct PendingAltScreenRead {
     history: Vec<crate::ghostty::ScreenTextRow>,
     phase: Phase,
     next_poll_at: Instant,
+    step_deadline: Instant,
+    output_quiet_until: Option<Instant>,
+    step_observed_output: bool,
+    synchronized_redraw_pending: bool,
     started_at: Instant,
     restore_started_at: Option<Instant>,
+    observed_content_seq: u64,
     upward_events: usize,
     reached_top: bool,
     valid: bool,
@@ -53,6 +59,7 @@ impl PendingAltScreenRead {
         lines: usize,
         unwrap: bool,
         initial: ScreenSnapshot,
+        content_seq: u64,
         now: Instant,
     ) -> Self {
         Self {
@@ -66,10 +73,15 @@ impl PendingAltScreenRead {
             previous: initial.clone(),
             history: initial.rows.clone(),
             initial,
-            phase: Phase::SettleInitial { checks: 0 },
-            next_poll_at: now + STEP_SETTLE,
+            phase: Phase::SettleInitial,
+            next_poll_at: now + INITIAL_QUIET,
+            step_deadline: now + INITIAL_QUIET,
+            output_quiet_until: None,
+            step_observed_output: false,
+            synchronized_redraw_pending: false,
             started_at: now,
             restore_started_at: None,
+            observed_content_seq: content_seq,
             upward_events: 0,
             reached_top: false,
             valid: true,
@@ -77,7 +89,7 @@ impl PendingAltScreenRead {
     }
 
     pub(crate) fn next_deadline(&self) -> Instant {
-        self.next_poll_at
+        self.output_quiet_until.unwrap_or(self.next_poll_at)
     }
 
     pub(crate) fn frozen_snapshot(
@@ -112,197 +124,256 @@ impl PendingAltScreenRead {
     pub(crate) fn abort(mut self, runtime: Option<&TerminalRuntime>, now: Instant) -> PollOutcome {
         self.valid = false;
         match self.phase {
-            Phase::SettleInitial { .. } => self.complete_fallback(),
-            Phase::Harvest { .. } => match runtime {
-                Some(runtime) => self.start_restore(runtime, now),
+            Phase::SettleInitial => self.complete_fallback(),
+            Phase::Harvest => match runtime {
+                Some(runtime) => self.start_restore(runtime, now, None),
                 None => self.complete_fallback(),
             },
-            Phase::ProbeBottom | Phase::RestoreProbe | Phase::Restore { .. } => {
-                self.poll(runtime, now)
-            }
+            Phase::ProbeBottom | Phase::RestoreProbe | Phase::Restore => self.poll(runtime, now),
         }
     }
 
     pub(crate) fn poll(mut self, runtime: Option<&TerminalRuntime>, now: Instant) -> PollOutcome {
-        if now < self.next_poll_at {
-            return Some(self);
-        }
         let Some(runtime) = runtime else {
             return self.complete_fallback();
         };
-        let Some((screen, snapshot)) = runtime.screen_text_snapshot() else {
+        let restore_expired = self
+            .restore_started_at
+            .is_some_and(|started| now.duration_since(started) >= MAX_RESTORE_DURATION);
+        if restore_expired {
             return self.complete_fallback();
+        }
+        let traversal_expired = now.duration_since(self.started_at) >= MAX_DURATION
+            && matches!(
+                self.phase,
+                Phase::SettleInitial | Phase::ProbeBottom | Phase::Harvest
+            );
+        if traversal_expired {
+            self.valid = false;
+            match self.phase {
+                Phase::SettleInitial => return self.complete_fallback(),
+                Phase::Harvest => return self.start_restore(runtime, now, None),
+                Phase::ProbeBottom => {}
+                Phase::RestoreProbe | Phase::Restore => unreachable!(),
+            }
+        }
+
+        let content_seq = runtime.content_seq();
+        if content_seq != self.observed_content_seq {
+            self.observed_content_seq = content_seq;
+            self.step_observed_output = true;
+            let synchronized_frame_complete =
+                self.synchronized_redraw_pending && !runtime.synchronized_output_active();
+            if synchronized_frame_complete {
+                self.synchronized_redraw_pending = false;
+                self.output_quiet_until = None;
+            } else {
+                self.output_quiet_until = Some(now + OUTPUT_QUIET);
+                if !traversal_expired {
+                    return Some(self);
+                }
+            }
+        }
+        if !traversal_expired && self.output_quiet_until.is_some_and(|quiet| now < quiet) {
+            return Some(self);
+        }
+        self.output_quiet_until = None;
+        if !traversal_expired && runtime.synchronized_output_active() {
+            self.synchronized_redraw_pending = true;
+            self.next_poll_at = now + OUTPUT_QUIET;
+            return Some(self);
+        }
+        let step_expired = now >= self.step_deadline;
+        let output_observed = self.step_observed_output;
+        if !step_expired && !output_observed {
+            return Some(self);
+        }
+        let Some((screen, snapshot, snapshot_seq)) = runtime.screen_text_snapshot_with_seq() else {
+            if traversal_expired {
+                return self.complete_fallback();
+            }
+            self.observed_content_seq = runtime.content_seq();
+            self.step_observed_output = false;
+            self.output_quiet_until = None;
+            self.next_poll_at = now + OUTPUT_QUIET;
+            return Some(self);
         };
+        if runtime.content_seq() != snapshot_seq {
+            self.observed_content_seq = snapshot_seq;
+            self.step_observed_output = false;
+            self.next_poll_at = now + OUTPUT_QUIET;
+            return Some(self);
+        }
         if screen != crate::ghostty::ActiveScreen::Alternate
             || snapshot.cols != self.initial.cols
             || snapshot.rows.len() != self.initial.rows.len()
         {
             return self.complete_fallback();
         }
-        if self
-            .restore_started_at
-            .is_some_and(|started| now.duration_since(started) >= MAX_RESTORE_DURATION)
-        {
-            return self.complete_fallback();
-        }
-        if now.duration_since(self.started_at) >= MAX_DURATION
-            && !matches!(
-                self.phase,
-                Phase::ProbeBottom | Phase::Restore { .. } | Phase::RestoreProbe
-            )
-        {
-            self.valid = false;
-            return self.start_restore(runtime, now);
-        }
-
         match self.phase {
-            Phase::SettleInitial { checks } => {
-                if snapshot.similar_text(&self.initial) {
-                    if checks >= 1 {
-                        if send_wheel(
-                            runtime,
-                            MouseEventKind::ScrollDown,
-                            WHEEL_STEP_EVENTS,
-                            &snapshot,
-                        )
-                        .is_err()
-                        {
-                            return self.complete_fallback();
-                        }
-                        self.phase = Phase::ProbeBottom;
-                    } else {
-                        self.phase = Phase::SettleInitial { checks: checks + 1 };
-                    }
-                } else {
+            Phase::SettleInitial => {
+                if output_observed || !snapshot.similar_text(&self.initial) {
                     self.initial = snapshot.clone();
                     self.previous = snapshot.clone();
                     self.history = snapshot.rows;
-                    self.phase = Phase::SettleInitial { checks: 0 };
+                    self.observed_content_seq = snapshot_seq;
+                    self.step_observed_output = false;
+                    self.next_poll_at = now + INITIAL_QUIET;
+                    self.step_deadline = self.next_poll_at;
+                    return Some(self);
                 }
-                self.next_poll_at = now + STEP_SETTLE;
+                if send_wheel(
+                    runtime,
+                    MouseEventKind::ScrollDown,
+                    WHEEL_STEP_EVENTS,
+                    &snapshot,
+                )
+                .is_err()
+                {
+                    return self.complete_fallback();
+                }
+                self.phase = Phase::ProbeBottom;
+                self.arm_step(snapshot_seq, now);
                 Some(self)
             }
             Phase::ProbeBottom => {
+                let at_bottom = snapshot.similar_text(&self.initial);
+                if output_observed && at_bottom && !step_expired && !traversal_expired {
+                    self.step_observed_output = false;
+                    return Some(self);
+                }
                 debug!(
                     terminal_id = %self.terminal_id,
-                    at_bottom = snapshot.similar_text(&self.initial),
+                    at_bottom,
                     "alternate-screen read bottom probe settled"
                 );
-                if snapshot.similar_text(&self.initial) {
+                if at_bottom {
                     if self.valid {
-                        self.start_harvest(runtime, now)
+                        self.start_harvest(runtime, now, snapshot_seq)
                     } else {
                         self.complete_fallback()
                     }
-                } else if send_wheel(
-                    runtime,
-                    MouseEventKind::ScrollUp,
-                    WHEEL_STEP_EVENTS,
-                    &snapshot,
-                )
-                .is_ok()
-                {
+                } else {
+                    if send_wheel(
+                        runtime,
+                        MouseEventKind::ScrollUp,
+                        WHEEL_STEP_EVENTS,
+                        &snapshot,
+                    )
+                    .is_err()
+                    {
+                        return self.complete_fallback();
+                    }
                     self.phase = Phase::RestoreProbe;
                     self.restore_started_at = Some(now);
-                    self.next_poll_at = now + STEP_SETTLE;
+                    self.arm_step(snapshot_seq, now);
                     Some(self)
-                } else {
-                    self.complete_fallback()
                 }
             }
             Phase::RestoreProbe => {
                 if snapshot.similar_text(&self.initial) {
                     self.complete_fallback()
                 } else {
-                    self.next_poll_at = now + STEP_SETTLE;
+                    self.step_observed_output = false;
+                    if step_expired {
+                        self.next_poll_at = now + STEP_TIMEOUT;
+                        self.step_deadline = self.next_poll_at;
+                    }
                     Some(self)
                 }
             }
-            Phase::Harvest { unaligned_checks } => {
+            Phase::Harvest => {
                 let merge = crate::terminal::merge_scrolled_up(
                     &mut self.history,
                     &self.previous,
                     &snapshot,
                 );
+                debug!(
+                    terminal_id = %self.terminal_id,
+                    ?merge,
+                    retained_rows = self.history.len(),
+                    batch_events = WHEEL_STEP_EVENTS,
+                    step_expired,
+                    "alternate-screen harvest snapshot"
+                );
                 match merge {
                     UpwardMerge::Advanced { .. } => {
                         self.previous = snapshot;
                         if self.history.len() >= self.lines {
-                            self.start_restore(runtime, now)
+                            self.start_restore(runtime, now, Some(snapshot_seq))
                         } else {
-                            self.start_harvest(runtime, now)
+                            self.start_harvest(runtime, now, snapshot_seq)
                         }
                     }
-                    UpwardMerge::Unchanged => {
+                    UpwardMerge::Unchanged if step_expired => {
                         self.reached_top = true;
-                        self.start_restore(runtime, now)
+                        self.start_restore(runtime, now, Some(snapshot_seq))
                     }
-                    UpwardMerge::Unaligned if unaligned_checks + 1 < MAX_UNALIGNED_CHECKS => {
-                        self.phase = Phase::Harvest {
-                            unaligned_checks: unaligned_checks + 1,
-                        };
-                        self.next_poll_at = now + STEP_SETTLE;
-                        Some(self)
-                    }
-                    UpwardMerge::Unaligned => {
+                    UpwardMerge::Unaligned if step_expired => {
                         self.valid = false;
-                        self.start_restore(runtime, now)
+                        self.start_restore(runtime, now, Some(snapshot_seq))
+                    }
+                    UpwardMerge::Unchanged | UpwardMerge::Unaligned => {
+                        self.step_observed_output = false;
+                        Some(self)
                     }
                 }
             }
-            Phase::Restore { stable_checks } => {
-                if snapshot.similar_text(&self.previous) {
-                    if stable_checks >= 1 {
-                        if self.valid {
-                            self.complete_success()
-                        } else {
-                            self.complete_fallback()
-                        }
+            Phase::Restore => {
+                if snapshot.similar_text(&self.initial) {
+                    if self.valid {
+                        self.complete_success()
                     } else {
-                        self.phase = Phase::Restore {
-                            stable_checks: stable_checks + 1,
-                        };
-                        self.next_poll_at = now + STEP_SETTLE;
-                        Some(self)
+                        self.complete_fallback()
                     }
-                } else {
-                    self.previous = snapshot;
+                } else if step_expired || !snapshot.similar_text(&self.previous) {
                     if send_wheel(
                         runtime,
                         MouseEventKind::ScrollDown,
-                        restore_batch_size(&self.previous),
-                        &self.previous,
+                        restore_batch_size(&snapshot),
+                        &snapshot,
                     )
-                    .is_ok()
+                    .is_err()
                     {
-                        self.phase = Phase::Restore { stable_checks: 0 };
-                        self.next_poll_at = now + STEP_SETTLE;
-                        Some(self)
-                    } else {
-                        self.complete_fallback()
+                        return self.complete_fallback();
                     }
+                    self.previous = snapshot;
+                    self.arm_step(snapshot_seq, now);
+                    Some(self)
+                } else {
+                    self.step_observed_output = false;
+                    Some(self)
                 }
             }
         }
     }
 
-    fn start_harvest(mut self, runtime: &TerminalRuntime, now: Instant) -> PollOutcome {
+    fn start_harvest(
+        mut self,
+        runtime: &TerminalRuntime,
+        now: Instant,
+        baseline_seq: u64,
+    ) -> PollOutcome {
         let events = WHEEL_STEP_EVENTS;
         if send_wheel(runtime, MouseEventKind::ScrollUp, events, &self.previous).is_err() {
             return self.complete_fallback();
         }
         self.upward_events = self.upward_events.saturating_add(events);
-        self.phase = Phase::Harvest {
-            unaligned_checks: 0,
-        };
-        self.next_poll_at = now + STEP_SETTLE;
+        self.phase = Phase::Harvest;
+        self.arm_step(baseline_seq, now);
         Some(self)
     }
 
-    fn start_restore(mut self, runtime: &TerminalRuntime, now: Instant) -> PollOutcome {
+    fn start_restore(
+        mut self,
+        runtime: &TerminalRuntime,
+        now: Instant,
+        baseline_seq: Option<u64>,
+    ) -> PollOutcome {
         if self.upward_events == 0 {
             return self.complete_fallback();
         }
+        let baseline_seq = baseline_seq.unwrap_or_else(|| runtime.content_seq());
         if send_wheel(
             runtime,
             MouseEventKind::ScrollDown,
@@ -313,12 +384,21 @@ impl PendingAltScreenRead {
         {
             return self.complete_fallback();
         }
-        self.phase = Phase::Restore { stable_checks: 0 };
+        self.phase = Phase::Restore;
         self.restore_started_at = Some(now);
-        self.next_poll_at = now + STEP_SETTLE;
+        self.arm_step(baseline_seq, now);
         Some(self)
     }
 
+    fn arm_step(&mut self, baseline_seq: u64, now: Instant) {
+        self.observed_content_seq = baseline_seq;
+        self.output_quiet_until = None;
+        self.step_observed_output = false;
+        self.synchronized_redraw_pending = false;
+        self.next_poll_at = now + STEP_TIMEOUT;
+        self.step_deadline = self.next_poll_at;
+    }
+
     fn complete_success(mut self) -> PollOutcome {
         debug!(
             terminal_id = %self.terminal_id,
@@ -385,3 +465,532 @@ fn send_wheel(
     }
     runtime.try_send_bytes(Bytes::from(bytes)).map_err(|_| ())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::api::schema::{ReadFormat, ReadSource};
+
+    fn draw(lines: &[&str], enter_alt_screen: bool) -> Vec<u8> {
+        let mut bytes = Vec::new();
+        if enter_alt_screen {
+            bytes.extend_from_slice(b"\x1b[?1049h\x1b[?1000h\x1b[?1006h");
+        }
+        bytes.extend_from_slice(b"\x1b[2J\x1b[H");
+        bytes.extend_from_slice(lines.join("\r\n").as_bytes());
+        bytes
+    }
+
+    fn pending_read(
+        runtime: &TerminalRuntime,
+        now: Instant,
+        lines: usize,
+    ) -> (PendingAltScreenRead, mpsc::Receiver<String>) {
+        let (_, initial) = runtime.screen_text_snapshot().expect("initial snapshot");
+        let (respond_to, response_rx) = mpsc::channel();
+        let pending = PendingAltScreenRead::start(
+            TerminalId::alloc(),
+            "read".into(),
+            respond_to,
+            "fallback".into(),
+            PaneReadResult {
+                pane_id: "w1:p1".into(),
+                workspace_id: "w1".into(),
+                tab_id: "w1:t1".into(),
+                source: ReadSource::Recent,
+                format: ReadFormat::Text,
+                text: String::new(),
+                revision: 0,
+                truncated: false,
+            },
+            lines,
+            false,
+            initial,
+            runtime.content_seq(),
+            now,
+        );
+        (pending, response_rx)
+    }
+
+    fn response_text(response_rx: &mpsc::Receiver<String>) -> String {
+        let response: SuccessResponse = serde_json::from_str(
+            &response_rx
+                .recv_timeout(Duration::from_millis(50))
+                .expect("read response"),
+        )
+        .expect("valid response");
+        let ResponseResult::PaneRead { read } = response.result else {
+            panic!("expected pane read response");
+        };
+        read.text
+    }
+
+    #[test]
+    fn hard_deadline_wins_over_continuous_output_coalescing() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let (runtime, _input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+
+        runtime.test_process_pty_bytes(b"\x1b]0;still changing\x07");
+        assert!(pending
+            .poll(Some(&runtime), started + MAX_DURATION)
+            .is_none());
+        assert_eq!(
+            response_rx
+                .recv_timeout(Duration::from_millis(50))
+                .expect("fallback response"),
+            "fallback"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn probe_bottom_hard_deadline_wins_over_continuous_output() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+
+        runtime.test_process_pty_bytes(b"\x1b]0;still changing\x07");
+        assert!(pending
+            .poll(Some(&runtime), started + MAX_DURATION)
+            .is_none());
+        assert_eq!(
+            response_rx
+                .recv_timeout(Duration::from_millis(50))
+                .expect("fallback response"),
+            "fallback"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn probe_bottom_hard_deadline_wins_over_synchronized_output() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+
+        runtime.test_process_pty_bytes(b"\x1b[?2026h");
+        assert!(pending
+            .poll(Some(&runtime), started + MAX_DURATION)
+            .is_none());
+        assert_eq!(
+            response_rx
+                .recv_timeout(Duration::from_millis(50))
+                .expect("fallback response"),
+            "fallback"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn redraw_events_advance_harvest_and_restore_without_settle_delays() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let initial_bytes = draw(&["16", "17", "18", "19", "20"], true);
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&initial_bytes);
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET + STEP_TIMEOUT)
+            .expect("history harvest");
+        input_rx.try_recv().expect("upward wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&["13", "14", "15", "16", "17"], false));
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(1),
+            )
+            .expect("redraw coalescing");
+        assert!(input_rx.try_recv().is_err());
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(11),
+            )
+            .expect("viewport restore");
+        input_rx.try_recv().expect("restore wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], false));
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(12),
+            )
+            .expect("restore redraw coalescing");
+        assert!(
+            pending
+                .poll(
+                    Some(&runtime),
+                    started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(22),
+                )
+                .is_none(),
+            "restored redraw should complete after coalescing"
+        );
+        assert_eq!(
+            response_text(&response_rx),
+            "13\n14\n15\n16\n17\n18\n19\n20\n"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn restore_keeps_trying_after_a_slow_redraw_exceeds_the_step_timeout() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+        let harvest_started = started + INITIAL_QUIET + STEP_TIMEOUT;
+
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+        let pending = pending
+            .poll(Some(&runtime), harvest_started)
+            .expect("history harvest");
+        input_rx.try_recv().expect("upward wheel batch");
+        runtime.test_process_pty_bytes(&draw(&["13", "14", "15", "16", "17"], false));
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(1))
+            .expect("redraw coalescing");
+        let restore_started = harvest_started + Duration::from_millis(11);
+        let pending = pending
+            .poll(Some(&runtime), restore_started)
+            .expect("viewport restore");
+        input_rx.try_recv().expect("restore wheel batch");
+
+        let retry_at = restore_started + STEP_TIMEOUT;
+        let pending = pending
+            .poll(Some(&runtime), retry_at)
+            .expect("slow restore must remain pending");
+        input_rx.try_recv().expect("retry restore wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], false));
+        let pending = pending
+            .poll(Some(&runtime), retry_at + Duration::from_millis(1))
+            .expect("restore redraw coalescing");
+        assert!(pending
+            .poll(Some(&runtime), retry_at + Duration::from_millis(11))
+            .is_none());
+        assert_eq!(
+            response_text(&response_rx),
+            "13\n14\n15\n16\n17\n18\n19\n20\n"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn unrelated_output_does_not_retry_restore_before_the_step_timeout() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let initial = ["16", "17", "18", "19", "20"];
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&draw(&initial, true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+        let harvest_started = started + INITIAL_QUIET + STEP_TIMEOUT;
+
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+        let pending = pending
+            .poll(Some(&runtime), harvest_started)
+            .expect("history harvest");
+        input_rx.try_recv().expect("upward wheel batch");
+        runtime.test_process_pty_bytes(&draw(&["13", "14", "15", "16", "17"], false));
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(1))
+            .expect("redraw coalescing");
+        let restore_started = harvest_started + Duration::from_millis(11);
+        let pending = pending
+            .poll(Some(&runtime), restore_started)
+            .expect("viewport restore");
+        input_rx.try_recv().expect("restore wheel batch");
+
+        runtime.test_process_pty_bytes(b"\x1b]0;unrelated title\x07");
+        let pending = pending
+            .poll(Some(&runtime), restore_started + Duration::from_millis(1))
+            .expect("unrelated output coalescing");
+        let pending = pending
+            .poll(Some(&runtime), restore_started + Duration::from_millis(11))
+            .expect("restore remains pending");
+        assert!(input_rx.try_recv().is_err());
+
+        runtime.test_process_pty_bytes(&draw(&initial, false));
+        let pending = pending
+            .poll(Some(&runtime), restore_started + Duration::from_millis(12))
+            .expect("restore redraw coalescing");
+        assert!(pending
+            .poll(Some(&runtime), restore_started + Duration::from_millis(22))
+            .is_none());
+        assert_eq!(
+            response_text(&response_rx),
+            "13\n14\n15\n16\n17\n18\n19\n20\n"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn synchronized_redraw_is_not_consumed_after_the_step_timeout() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let initial = ["16", "17", "18", "19", "20"];
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&draw(&initial, true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+        let harvest_started = started + INITIAL_QUIET + STEP_TIMEOUT;
+
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+        let pending = pending
+            .poll(Some(&runtime), harvest_started)
+            .expect("history harvest");
+        input_rx.try_recv().expect("upward wheel batch");
+
+        let mut synchronized = b"\x1b[?2026h".to_vec();
+        synchronized.extend(draw(&["13", "14", "15", "16", "17"], false));
+        runtime.test_process_pty_bytes(&synchronized);
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(1))
+            .expect("synchronized redraw observed");
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                harvest_started + STEP_TIMEOUT + Duration::from_millis(1),
+            )
+            .expect("synchronized redraw must remain pending");
+        assert!(input_rx.try_recv().is_err());
+
+        runtime.test_process_pty_bytes(b"\x1b[?2026l");
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                harvest_started + STEP_TIMEOUT + Duration::from_millis(2),
+            )
+            .expect("viewport restore after synchronized redraw");
+        input_rx.try_recv().expect("restore wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&initial, false));
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                harvest_started + STEP_TIMEOUT + Duration::from_millis(3),
+            )
+            .expect("restore redraw coalescing");
+        assert!(pending
+            .poll(
+                Some(&runtime),
+                harvest_started + STEP_TIMEOUT + Duration::from_millis(13)
+            )
+            .is_none());
+        assert_eq!(
+            response_text(&response_rx),
+            "13\n14\n15\n16\n17\n18\n19\n20\n"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn aligned_intermediate_redraw_is_coalesced_before_scrolling_again() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let initial = ["16", "17", "18", "19", "20", "ready"];
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 6, 8);
+        runtime.test_process_pty_bytes(&draw(&initial, true));
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 9);
+        let harvest_started = started + INITIAL_QUIET + STEP_TIMEOUT;
+
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+        let pending = pending
+            .poll(Some(&runtime), harvest_started)
+            .expect("history harvest");
+        input_rx.try_recv().expect("upward wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&["13", "14", "15", "16", "17", "loading"], false));
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(1))
+            .expect("intermediate redraw coalescing");
+        assert!(input_rx.try_recv().is_err());
+        runtime.test_process_pty_bytes(&draw(&["13", "14", "15", "16", "17", "ready"], false));
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(5))
+            .expect("completed redraw coalescing");
+        assert!(input_rx.try_recv().is_err());
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(15))
+            .expect("viewport restore after completed redraw");
+        input_rx.try_recv().expect("restore wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&initial, false));
+        let pending = pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(16))
+            .expect("restore redraw coalescing");
+        assert!(pending
+            .poll(Some(&runtime), harvest_started + Duration::from_millis(26))
+            .is_none());
+        assert_eq!(
+            response_text(&response_rx),
+            "13\n14\n15\n16\n17\n18\n19\n20\nready\n"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    #[test]
+    fn incomplete_redraw_waits_for_an_aligned_screen_without_resetting_timeout() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _guard = rt.enter();
+        let initial_bytes = draw(&["16", "17", "18", "19", "20"], true);
+        let (runtime, mut input_rx) = TerminalRuntime::test_with_channel_capacity(20, 5, 8);
+        runtime.test_process_pty_bytes(&initial_bytes);
+        let started = Instant::now();
+        let (pending, response_rx) = pending_read(&runtime, started, 8);
+
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET)
+            .expect("bottom probe");
+        input_rx.try_recv().expect("bottom wheel probe");
+        let pending = pending
+            .poll(Some(&runtime), started + INITIAL_QUIET + STEP_TIMEOUT)
+            .expect("history harvest");
+        input_rx.try_recv().expect("upward wheel batch");
+
+        runtime.test_process_pty_bytes(&draw(&["13"], false));
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(1),
+            )
+            .expect("partial redraw coalescing");
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(11),
+            )
+            .expect("partial redraw must not complete");
+        assert!(
+            input_rx.try_recv().is_err(),
+            "partial redraw must not scroll again"
+        );
+
+        runtime.test_process_pty_bytes(&draw(&["13", "14", "15", "16", "17"], false));
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(12),
+            )
+            .expect("aligned redraw coalescing");
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(22),
+            )
+            .expect("aligned redraw should start restore");
+        input_rx.try_recv().expect("restore wheel batch");
+        runtime.test_process_pty_bytes(&draw(&["16", "17", "18", "19", "20"], false));
+        let pending = pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(23),
+            )
+            .expect("restore redraw coalescing");
+        assert!(pending
+            .poll(
+                Some(&runtime),
+                started + INITIAL_QUIET + STEP_TIMEOUT + Duration::from_millis(33),
+            )
+            .is_none());
+        assert_eq!(
+            response_text(&response_rx),
+            "13\n14\n15\n16\n17\n18\n19\n20\n"
+        );
+
+        drop(runtime);
+        drop(_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+}
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 7393a7c5..ac1e7db2 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -278,6 +278,7 @@ struct AltScreenReadSpec {
     lines: usize,
     unwrap: bool,
     initial: crate::terminal::ScreenSnapshot,
+    content_seq: u64,
 }
 
 enum AltScreenReadConflict {
@@ -3416,7 +3417,7 @@ impl HeadlessServer {
         if runtime.wheel_routing() != Some(crate::pane::WheelRouting::MouseReport) {
             return None;
         }
-        let (screen, initial) = runtime.screen_text_snapshot()?;
+        let (screen, initial, content_seq) = runtime.screen_text_snapshot_with_seq()?;
         if screen != crate::ghostty::ActiveScreen::Alternate || initial.rows.len() >= lines {
             return None;
         }
@@ -3425,6 +3426,7 @@ impl HeadlessServer {
             lines,
             unwrap: source == ReadSource::RecentUnwrapped,
             initial,
+            content_seq,
         })
     }
 
@@ -3776,6 +3778,7 @@ impl HeadlessServer {
                         spec.lines,
                         spec.unwrap,
                         spec.initial,
+                        spec.content_seq,
                         Instant::now(),
                     );
                     self.pending_alt_screen_reads.push(pending);
@@ -6636,6 +6639,7 @@ next_tab = ""
                         cols: 80,
                         rows: Vec::new(),
                     },
+                    0,
                     Instant::now(),
                 ),
             );
diff --git a/src/terminal/runtime.rs b/src/terminal/runtime.rs
index 156d18fd..619c98a4 100644
--- a/src/terminal/runtime.rs
+++ b/src/terminal/runtime.rs
@@ -468,6 +468,27 @@ impl TerminalRuntime {
         Some((screen, crate::terminal::ScreenSnapshot { cols, rows }))
     }
 
+    pub(crate) fn screen_text_snapshot_with_seq(
+        &self,
+    ) -> Option<(
+        crate::ghostty::ActiveScreen,
+        crate::terminal::ScreenSnapshot,
+        u64,
+    )> {
+        for _ in 0..3 {
+            let before = self.content_seq();
+            if !before.is_multiple_of(2) {
+                continue;
+            }
+            let (screen, snapshot) = self.screen_text_snapshot()?;
+            let after = self.content_seq();
+            if before == after {
+                return Some((screen, snapshot, after));
+            }
+        }
+        None
+    }
+
     pub fn encode_mouse_button(
         &self,
         kind: crossterm::event::MouseEventKind,
@@ -525,6 +546,10 @@ impl TerminalRuntime {
     pub(crate) fn current_size(&self) -> (u16, u16) {
         self.0.current_size()
     }
+
+    pub(crate) fn content_seq(&self) -> u64 {
+        self.0.content_seq()
+    }
 }
 
 #[cfg(test)]