perf: compress idle terminal scrollback (#3556)

6 files changed, 771 insertions, 33 deletions

diff --git a/src/ghostty/mod.rs b/src/ghostty/mod.rs
index 261638af..691160e6 100644
--- a/src/ghostty/mod.rs
+++ b/src/ghostty/mod.rs
@@ -74,6 +74,13 @@ pub enum Dirty {
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum TerminalCompressionResult {
+    Unsupported,
+    Pending,
+    Complete,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct RowSelection {
     pub start_x: u16,
     pub end_x: u16,
@@ -871,6 +878,41 @@ impl Terminal {
         }
     }
 
+    pub(crate) fn compression_activity(&self) -> Result<u64, Error> {
+        let mut activity = 0;
+        // SAFETY: self.raw is a live terminal handle and activity is a valid out pointer.
+        unsafe {
+            ffi::ghostty_terminal_compression_activity(self.raw, &mut activity).into_result()?;
+        }
+        Ok(activity)
+    }
+
+    pub(crate) fn compress_incremental(&mut self) -> Result<TerminalCompressionResult, Error> {
+        let mut result =
+            ffi::GhosttyTerminalCompressionResult_GHOSTTY_TERMINAL_COMPRESSION_RESULT_UNSUPPORTED;
+        // SAFETY: self.raw is a live terminal handle and result is a valid out pointer.
+        unsafe {
+            ffi::ghostty_terminal_compress(
+                self.raw,
+                ffi::GhosttyTerminalCompressionMode_GHOSTTY_TERMINAL_COMPRESSION_MODE_INCREMENTAL,
+                &mut result,
+            )
+            .into_result()?;
+        }
+        match result {
+            ffi::GhosttyTerminalCompressionResult_GHOSTTY_TERMINAL_COMPRESSION_RESULT_UNSUPPORTED => {
+                Ok(TerminalCompressionResult::Unsupported)
+            }
+            ffi::GhosttyTerminalCompressionResult_GHOSTTY_TERMINAL_COMPRESSION_RESULT_PENDING => {
+                Ok(TerminalCompressionResult::Pending)
+            }
+            ffi::GhosttyTerminalCompressionResult_GHOSTTY_TERMINAL_COMPRESSION_RESULT_COMPLETE => {
+                Ok(TerminalCompressionResult::Complete)
+            }
+            _ => Err(Error(ffi::GhosttyResult_GHOSTTY_INVALID_VALUE)),
+        }
+    }
+
     pub fn set_default_palette(&mut self, palette: &[RgbColor; 256]) -> Result<(), Error> {
         let palette = palette.map(|color| ffi::GhosttyColorRgb {
             r: color.r,
@@ -3304,6 +3346,38 @@ mod tests {
     }
 
     #[test]
+    fn incremental_compression_preserves_cold_scrollback() {
+        let mut terminal = Terminal::new(80, 24, 20_000_000).unwrap();
+        let initial_activity = terminal.compression_activity().unwrap();
+        let suffix = "x".repeat(66);
+        for line in 1..=10_000 {
+            terminal.write(format!("{line:05} {suffix}\r\n").as_bytes());
+        }
+        assert_ne!(terminal.compression_activity().unwrap(), initial_activity);
+
+        let mut complete = false;
+        for _ in 0..10_000 {
+            match terminal.compress_incremental().unwrap() {
+                TerminalCompressionResult::Unsupported => return,
+                TerminalCompressionResult::Pending => {}
+                TerminalCompressionResult::Complete => {
+                    complete = true;
+                    break;
+                }
+            }
+        }
+        assert!(complete, "incremental compression did not converge");
+
+        let oldest = terminal.read_text_screen((0, 0), (79, 0), false).unwrap();
+        assert!(oldest.starts_with("00001 "));
+        let last_row = terminal.total_rows().unwrap() as u32 - 1;
+        let newest = terminal
+            .read_text_screen((0, last_row - 1), (79, last_row), false)
+            .unwrap();
+        assert!(newest.contains("10000"));
+    }
+
+    #[test]
     fn kitty_image_fingerprint_covers_full_payload() {
         let mut data = vec![1u8; 4096 * 4];
         let original =
diff --git a/src/pane.rs b/src/pane.rs
index c71665f5..262fbe51 100644
--- a/src/pane.rs
+++ b/src/pane.rs
@@ -3,7 +3,7 @@ use std::io;
 use std::path::Path;
 use std::sync::{
     atomic::{AtomicBool, AtomicU16, AtomicU32, AtomicU64, Ordering},
-    Arc, Mutex,
+    Arc, Mutex, OnceLock,
 };
 
 use bytes::Bytes;
@@ -44,8 +44,8 @@ use self::agent_detection::{
 pub use self::terminal::InputState;
 use self::terminal::{GhosttyPaneTerminal, PaneTerminal};
 pub(crate) use self::terminal::{
-    TerminalDirtyPatch, TerminalDirtyPatchOutcome, TerminalReadSnapshot, TerminalSearchDirection,
-    TerminalSearchWindow, TerminalTextPoint, TerminalWordMotion,
+    TerminalCompressionStep, TerminalDirtyPatch, TerminalDirtyPatchOutcome, TerminalReadSnapshot,
+    TerminalSearchDirection, TerminalSearchWindow, TerminalTextPoint, TerminalWordMotion,
 };
 pub use self::{
     state::PaneState,
@@ -53,9 +53,32 @@ pub use self::{
 };
 
 const RELEASE_REACQUIRE_SUPPRESSION: std::time::Duration = std::time::Duration::from_secs(1);
+const TERMINAL_COMPRESSION_IDLE: std::time::Duration = std::time::Duration::from_millis(250);
+const TERMINAL_COMPRESSION_STEP: std::time::Duration = std::time::Duration::from_millis(1);
 const PANE_TERM: &str = "xterm-256color";
 const PANE_COLORTERM: &str = "truecolor";
 
+fn terminal_compression_permits() -> Arc<tokio::sync::Semaphore> {
+    static PERMITS: OnceLock<Arc<tokio::sync::Semaphore>> = OnceLock::new();
+    PERMITS
+        .get_or_init(|| Arc::new(tokio::sync::Semaphore::new(4)))
+        .clone()
+}
+
+fn spawn_blocking_with_compression_permit<T, F>(
+    permit: tokio::sync::OwnedSemaphorePermit,
+    operation: F,
+) -> tokio::task::JoinHandle<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> T + Send + 'static,
+{
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        operation()
+    })
+}
+
 fn apply_pane_terminal_env(cmd: &mut CommandBuilder) {
     // Each pane is rendered by herdr's own terminal layer, not the outer terminal
     // that launched the app. Advertising the inherited TERM leaks the host terminal
@@ -1026,8 +1049,190 @@ impl AgentDetectionPresence {
 // PaneRuntime — PTY, parser, channels, background tasks
 // ---------------------------------------------------------------------------
 
+#[derive(Clone)]
+struct TerminalCompressionWake {
+    notify: Arc<Notify>,
+    generation: Arc<AtomicU64>,
+}
+
+impl TerminalCompressionWake {
+    fn wake(&self) {
+        self.generation.fetch_add(1, Ordering::Release);
+        self.notify.notify_one();
+    }
+}
+
+/// Drives libghostty-vt's caller-owned compression after terminal activity settles.
+struct TerminalCompressionTask {
+    wake: TerminalCompressionWake,
+    #[cfg(test)]
+    completed_passes: Arc<AtomicU64>,
+    handle: tokio::task::AbortHandle,
+}
+
+impl Drop for TerminalCompressionTask {
+    fn drop(&mut self) {
+        self.handle.abort();
+    }
+}
+
+impl TerminalCompressionTask {
+    fn spawn(pane_id: PaneId, terminal: Arc<PaneTerminal>) -> Self {
+        let wake = TerminalCompressionWake {
+            notify: Arc::new(Notify::new()),
+            generation: Arc::new(AtomicU64::new(0)),
+        };
+        let task_notify = wake.notify.clone();
+        let task_generation = wake.generation.clone();
+        #[cfg(test)]
+        let completed_passes = Arc::new(AtomicU64::new(0));
+        #[cfg(test)]
+        let task_completed_passes = completed_passes.clone();
+        let handle = tokio::spawn(async move {
+            run_terminal_compression_task(
+                pane_id,
+                terminal,
+                task_notify,
+                task_generation,
+                #[cfg(test)]
+                task_completed_passes,
+            )
+            .await;
+        })
+        .abort_handle();
+        Self {
+            wake,
+            #[cfg(test)]
+            completed_passes,
+            handle,
+        }
+    }
+
+    fn wake(&self) {
+        self.wake.wake();
+    }
+
+    fn notifier(&self) -> TerminalCompressionWake {
+        self.wake.clone()
+    }
+
+    fn abort(&self) {
+        self.handle.abort();
+    }
+
+    #[cfg(test)]
+    fn completed_passes(&self) -> u64 {
+        self.completed_passes.load(Ordering::Acquire)
+    }
+}
+
+async fn run_terminal_compression_task(
+    pane_id: PaneId,
+    terminal: Arc<PaneTerminal>,
+    notify: Arc<Notify>,
+    generation: Arc<AtomicU64>,
+    #[cfg(test)] completed_passes: Arc<AtomicU64>,
+) {
+    let mut observed_generation = generation.load(Ordering::Acquire);
+    let mut activity = loop {
+        match terminal.try_compression_activity() {
+            Ok(Some(activity)) => break activity,
+            Ok(None) => tokio::time::sleep(TERMINAL_COMPRESSION_IDLE).await,
+            Err(err) => {
+                warn!(pane = pane_id.raw(), err = %err, "failed to read terminal compression activity");
+                return;
+            }
+        }
+    };
+
+    'schedule: loop {
+        loop {
+            tokio::time::sleep(TERMINAL_COMPRESSION_IDLE).await;
+            let current = match terminal.try_compression_activity() {
+                Ok(Some(current)) => current,
+                Ok(None) => continue,
+                Err(err) => {
+                    warn!(pane = pane_id.raw(), err = %err, "failed to read terminal compression activity");
+                    return;
+                }
+            };
+            let current_generation = generation.load(Ordering::Acquire);
+            if activity == current && observed_generation == current_generation {
+                break;
+            }
+            activity = current;
+            observed_generation = current_generation;
+        }
+
+        loop {
+            let current_generation = generation.load(Ordering::Acquire);
+            if observed_generation != current_generation {
+                observed_generation = current_generation;
+                continue 'schedule;
+            }
+
+            let permit = match terminal_compression_permits().acquire_owned().await {
+                Ok(permit) => permit,
+                Err(_) => return,
+            };
+            let current_generation = generation.load(Ordering::Acquire);
+            if observed_generation != current_generation {
+                observed_generation = current_generation;
+                continue 'schedule;
+            }
+
+            let terminal_for_step = terminal.clone();
+            let step = spawn_blocking_with_compression_permit(permit, move || {
+                terminal_for_step.try_compress_incremental_if_activity(activity)
+            })
+            .await;
+            let step = match step {
+                Ok(Ok(step)) => step,
+                Ok(Err(err)) => {
+                    warn!(pane = pane_id.raw(), err = %err, "failed to compress terminal scrollback");
+                    return;
+                }
+                Err(err) => {
+                    warn!(pane = pane_id.raw(), err = %err, "terminal compression worker failed");
+                    return;
+                }
+            };
+
+            match step {
+                TerminalCompressionStep::Busy => continue 'schedule,
+                TerminalCompressionStep::ActivityChanged(current) => {
+                    activity = current;
+                    observed_generation = generation.load(Ordering::Acquire);
+                    continue 'schedule;
+                }
+                TerminalCompressionStep::Compressed(
+                    crate::ghostty::TerminalCompressionResult::Unsupported,
+                ) => return,
+                TerminalCompressionStep::Compressed(
+                    crate::ghostty::TerminalCompressionResult::Pending,
+                ) => tokio::time::sleep(TERMINAL_COMPRESSION_STEP).await,
+                TerminalCompressionStep::Compressed(
+                    crate::ghostty::TerminalCompressionResult::Complete,
+                ) => {
+                    #[cfg(test)]
+                    completed_passes.fetch_add(1, Ordering::Release);
+                    loop {
+                        notify.notified().await;
+                        let current_generation = generation.load(Ordering::Acquire);
+                        if observed_generation != current_generation {
+                            observed_generation = current_generation;
+                            continue 'schedule;
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
 /// PTY runtime for a pane. Owns the terminal, I/O channels, and background tasks.
-/// Dropping this shuts down all background tasks and closes the PTY.
+/// Dropping this aborts async tasks and closes the PTY. An already-running bounded
+/// compression step may finish before releasing its terminal reference.
 pub struct PaneRuntime {
     pane_id: PaneId,
     terminal: Arc<PaneTerminal>,
@@ -1045,6 +1250,7 @@ pub struct PaneRuntime {
     pending_release: Arc<Mutex<Option<PendingAgentRelease>>>,
     preserve_processes_on_drop: bool,
     // Task handles for deterministic shutdown
+    compression: TerminalCompressionTask,
     detect_handle: Option<tokio::task::AbortHandle>,
 }
 
@@ -1223,6 +1429,7 @@ impl Drop for PaneRuntime {
         if let Some(handle) = &self.detect_handle {
             handle.abort();
         }
+        self.compression.abort();
         self.io.shutdown();
         if !self.preserve_processes_on_drop {
             shutdown_pane_processes(
@@ -1589,6 +1796,7 @@ impl PaneRuntime {
         if let Some(handle) = self.detect_handle.take() {
             handle.abort();
         }
+        self.compression.abort();
         self.io.shutdown();
         shutdown_pane_processes(
             self.pane_id,
@@ -1615,6 +1823,7 @@ impl PaneRuntime {
         if let Some(handle) = self.detect_handle.take() {
             handle.abort();
         }
+        self.compression.abort();
         self.preserve_processes_on_drop = true;
     }
 
@@ -1905,6 +2114,7 @@ impl PaneRuntime {
             pane_terminal.seed_history_ansi(ansi);
         }
         let terminal = Arc::new(PaneTerminal::new(pane_terminal));
+        let compression = TerminalCompressionTask::spawn(pane_id, terminal.clone());
         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));
@@ -1923,6 +2133,7 @@ impl PaneRuntime {
             let child_pid = child_pid.clone();
             let read_events = events.clone();
             let reported_cwd = reported_cwd.clone();
+            let compression_wake = compression.notifier();
             let rt = tokio::runtime::Handle::current();
             let delay_rt = rt.clone();
             let on_read = Box::new(move |bytes: &[u8]| {
@@ -1936,6 +2147,7 @@ impl PaneRuntime {
                     terminal.process_pty_bytes(pane_id, shell_pid, bytes, &response_writer);
                 content_seq.fetch_add(1, Ordering::Release);
                 drop(_content_write_guard);
+                compression_wake.wake();
                 publish_terminal_bells(pane_id, result.terminal_bells, &read_events);
                 observe_detection_content_change(bytes, &detection_content_seq);
                 let title_requested =
@@ -2010,6 +2222,7 @@ impl PaneRuntime {
             detect_reset_notify,
             pending_release,
             preserve_processes_on_drop: true,
+            compression,
             detect_handle: Some(detect_handle),
         })
     }
@@ -2051,6 +2264,7 @@ impl PaneRuntime {
             pane_terminal.seed_history_ansi(ansi);
         }
         let terminal = Arc::new(PaneTerminal::new(pane_terminal));
+        let compression = TerminalCompressionTask::spawn(pane_id, terminal.clone());
         let kitty_keyboard_flags = Arc::new(AtomicU16::new(0));
         let content_write_lock = Arc::new(Mutex::new(()));
 
@@ -2101,6 +2315,7 @@ impl PaneRuntime {
             let child_pid = child_pid.clone();
             let events = events.clone();
             let reported_cwd = reported_cwd.clone();
+            let compression_wake = compression.notifier();
             let rt = tokio::runtime::Handle::current();
             let on_read = Box::new(move |bytes: &[u8]| {
                 let _content_write_guard = match content_write_lock.lock() {
@@ -2113,6 +2328,7 @@ impl PaneRuntime {
                     terminal.process_pty_bytes(pane_id, shell_pid, bytes, &response_writer);
                 content_seq.fetch_add(1, Ordering::Release);
                 drop(_content_write_guard);
+                compression_wake.wake();
                 publish_terminal_bells(pane_id, result.terminal_bells, &events);
                 if agent_detection == AgentDetection::Enabled {
                     observe_detection_content_change(bytes, &detection_content_seq);
@@ -2574,6 +2790,7 @@ impl PaneRuntime {
             detect_reset_notify,
             pending_release,
             preserve_processes_on_drop: false,
+            compression,
             detect_handle,
         })
     }
@@ -2634,6 +2851,7 @@ impl PaneRuntime {
             .resize(rows, cols, cell_width_px, cell_height_px);
         self.content_seq.fetch_add(1, Ordering::Release);
         drop(_content_write_guard);
+        self.compression.wake();
         mark_detection_content_changed(&self.detection_content_seq);
         self.io.resize(
             rows,
@@ -2654,21 +2872,25 @@ impl PaneRuntime {
     /// Scroll up by N lines (into scrollback history).
     pub fn scroll_up(&self, lines: usize) {
         self.terminal.scroll_up(lines);
+        self.compression.wake();
     }
 
     /// Scroll down by N lines (toward live output).
     pub fn scroll_down(&self, lines: usize) {
         self.terminal.scroll_down(lines);
+        self.compression.wake();
     }
 
     /// Reset scroll to live view (offset = 0).
     pub fn scroll_reset(&self) {
         self.terminal.scroll_reset();
+        self.compression.wake();
     }
 
     /// Set scrollback offset measured from the live bottom of the terminal.
     pub fn set_scroll_offset_from_bottom(&self, lines: usize) {
         self.terminal.set_scroll_offset_from_bottom(lines);
+        self.compression.wake();
     }
 
     pub fn scroll_metrics(&self) -> Option<ScrollMetrics> {
@@ -2687,8 +2909,16 @@ impl PaneRuntime {
         )>,
         limit: usize,
     ) -> crate::pane::TerminalSearchWindow {
-        self.terminal
-            .search_text_window(query, case_sensitive, direction, cursor, previous, limit)
+        let result = self.terminal.search_text_window(
+            query,
+            case_sensitive,
+            direction,
+            cursor,
+            previous,
+            limit,
+        );
+        self.compression.wake();
+        result
     }
 
     pub(crate) fn word_motion_target(
@@ -2697,7 +2927,9 @@ impl PaneRuntime {
         col: u16,
         motion: crate::pane::TerminalWordMotion,
     ) -> Option<crate::pane::TerminalTextPoint> {
-        self.terminal.word_motion_target(row, col, motion)
+        let result = self.terminal.word_motion_target(row, col, motion);
+        self.compression.wake();
+        result
     }
 
     pub(crate) fn terminal_dimensions(&self) -> Option<(u16, u16)> {
@@ -2709,7 +2941,9 @@ impl PaneRuntime {
         row: u32,
         direction: i8,
     ) -> Option<crate::pane::TerminalTextPoint> {
-        self.terminal.paragraph_motion_target(row, direction)
+        let result = self.terminal.paragraph_motion_target(row, direction);
+        self.compression.wake();
+        result
     }
 
     #[cfg(any(unix, test))]
@@ -2786,23 +3020,33 @@ impl PaneRuntime {
     }
 
     pub(crate) fn recent_text_snapshot(&self, lines: usize) -> TerminalReadSnapshot {
-        self.terminal.recent_text_snapshot(lines)
+        let result = self.terminal.recent_text_snapshot(lines);
+        self.compression.wake();
+        result
     }
 
     pub(crate) fn recent_ansi_snapshot(&self, lines: usize) -> TerminalReadSnapshot {
-        self.terminal.recent_ansi_snapshot(lines)
+        let result = self.terminal.recent_ansi_snapshot(lines);
+        self.compression.wake();
+        result
     }
 
     pub(crate) fn recent_unwrapped_text_snapshot(&self, lines: usize) -> TerminalReadSnapshot {
-        self.terminal.recent_unwrapped_text_snapshot(lines)
+        let result = self.terminal.recent_unwrapped_text_snapshot(lines);
+        self.compression.wake();
+        result
     }
 
     pub fn recent_unwrapped_ansi(&self, lines: usize) -> String {
-        self.terminal.recent_unwrapped_ansi(lines)
+        let result = self.terminal.recent_unwrapped_ansi(lines);
+        self.compression.wake();
+        result
     }
 
     pub(crate) fn recent_unwrapped_ansi_snapshot(&self, lines: usize) -> TerminalReadSnapshot {
-        self.terminal.recent_unwrapped_ansi_snapshot(lines)
+        let result = self.terminal.recent_unwrapped_ansi_snapshot(lines);
+        self.compression.wake();
+        result
     }
 
     pub fn snapshot_history(&self) -> Option<String> {
@@ -2811,7 +3055,9 @@ impl PaneRuntime {
     }
 
     pub fn extract_selection(&self, selection: &crate::selection::Selection) -> Option<String> {
-        self.terminal.extract_selection(selection)
+        let result = self.terminal.extract_selection(selection);
+        self.compression.wake();
+        result
     }
 
     pub fn render(&self, frame: &mut Frame, area: Rect, show_cursor: bool) {
@@ -2909,7 +3155,9 @@ impl PaneRuntime {
         u16,
         Vec<crate::ghostty::ScreenTextRow>,
     )> {
-        self.terminal.screen_text_snapshot()
+        let result = self.terminal.screen_text_snapshot();
+        self.compression.wake();
+        result
     }
 
     pub fn encode_mouse_button(
@@ -3060,6 +3308,7 @@ impl PaneRuntime {
         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);
+        self.compression.wake();
     }
 
     pub(crate) fn test_with_scrollback_bytes(
@@ -3083,13 +3332,16 @@ impl PaneRuntime {
         let mut terminal =
             crate::ghostty::Terminal::new(cols, rows, scrollback_limit_bytes).unwrap();
         terminal.write(bytes);
+        let pane_id = PaneId::from_raw(0);
+        let terminal = Arc::new(PaneTerminal::new(
+            GhosttyPaneTerminal::new(terminal, tx.clone()).unwrap(),
+        ));
+        let compression = TerminalCompressionTask::spawn(pane_id, terminal.clone());
 
         (
             Self {
-                pane_id: PaneId::from_raw(0),
-                terminal: Arc::new(PaneTerminal::new(
-                    GhosttyPaneTerminal::new(terminal, tx.clone()).unwrap(),
-                )),
+                pane_id,
+                terminal,
                 io: PaneRuntimeIo::TestChannel {
                     sender: tx,
                     resize_tx,
@@ -3106,6 +3358,7 @@ impl PaneRuntime {
                 detect_reset_notify: Arc::new(Notify::new()),
                 pending_release: Arc::new(Mutex::new(None)),
                 preserve_processes_on_drop: true,
+                compression,
                 detect_handle: Some(tokio::spawn(async {}).abort_handle()),
             },
             rx,
@@ -3648,6 +3901,96 @@ mod tests {
     }
 
     #[tokio::test]
+    async fn compression_permit_survives_an_aborted_async_waiter() {
+        let semaphore = Arc::new(tokio::sync::Semaphore::new(1));
+        let permit = semaphore.clone().acquire_owned().await.unwrap();
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        let handle = spawn_blocking_with_compression_permit(permit, move || {
+            let _ = started_tx.send(());
+            let _ = release_rx.recv();
+        });
+        started_rx.await.unwrap();
+
+        handle.abort();
+        assert!(semaphore.clone().try_acquire_owned().is_err());
+
+        release_tx.send(()).unwrap();
+        handle.await.unwrap();
+        assert!(semaphore.try_acquire_owned().is_ok());
+    }
+
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    #[tokio::test]
+    async fn compression_task_rechecks_history_after_a_read() {
+        let suffix = "x".repeat(66);
+        let history = (1..=2_000)
+            .map(|line| format!("{line:05} {suffix}\r\n"))
+            .collect::<String>();
+        let runtime =
+            PaneRuntime::test_with_scrollback_bytes(80, 24, 20_000_000, history.as_bytes());
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while runtime.compression.completed_passes() == 0 {
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap();
+        let completed_before_read = runtime.compression.completed_passes();
+
+        let snapshot = runtime.recent_unwrapped_text_snapshot(usize::MAX);
+        assert!(snapshot.text.contains("00001 "));
+        assert!(snapshot.text.contains("02000 "));
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while runtime.compression.completed_passes() == completed_before_read {
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap();
+    }
+
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    #[tokio::test]
+    async fn compressed_scrollback_survives_shrink_and_grow_resize() {
+        let suffix = "x".repeat(66);
+        let history = (1..=2_000)
+            .map(|line| format!("{line:05} {suffix}\r\n"))
+            .collect::<String>();
+        let runtime =
+            PaneRuntime::test_with_scrollback_bytes(80, 45, 20_000_000, history.as_bytes());
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while runtime.compression.completed_passes() == 0 {
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap();
+
+        runtime.resize(21, 80, 0, 0);
+        let completed_after_shrink = runtime.compression.completed_passes();
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while runtime.compression.completed_passes() == completed_after_shrink {
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap();
+
+        runtime.resize(45, 80, 0, 0);
+
+        assert_eq!(runtime.current_size(), (45, 80));
+        assert_eq!(runtime.terminal_dimensions(), Some((80, 45)));
+        assert_eq!(runtime.scroll_metrics().unwrap().viewport_rows, 45);
+        let snapshot = runtime.recent_unwrapped_text_snapshot(usize::MAX);
+        assert!(snapshot.text.contains("00001 "));
+        assert!(snapshot.text.contains("02000 "));
+    }
+
+    #[tokio::test]
     async fn focus_events_are_forwarded_when_enabled() {
         let (tx, mut rx) = mpsc::channel(4);
         let (resize_tx, _resize_rx) = watch::channel((80, 24, 0, 0));
@@ -3655,11 +3998,14 @@ mod tests {
         terminal
             .mode_set(crate::ghostty::MODE_FOCUS_EVENT, true)
             .unwrap();
+        let pane_id = PaneId::from_raw(0);
+        let terminal = Arc::new(PaneTerminal::new(
+            GhosttyPaneTerminal::new(terminal, tx.clone()).unwrap(),
+        ));
+        let compression = TerminalCompressionTask::spawn(pane_id, terminal.clone());
         let runtime = PaneRuntime {
-            pane_id: PaneId::from_raw(0),
-            terminal: Arc::new(PaneTerminal::new(
-                GhosttyPaneTerminal::new(terminal, tx.clone()).unwrap(),
-            )),
+            pane_id,
+            terminal,
             io: PaneRuntimeIo::TestChannel {
                 sender: tx,
                 resize_tx,
@@ -3676,6 +4022,7 @@ mod tests {
             detect_reset_notify: Arc::new(Notify::new()),
             pending_release: Arc::new(Mutex::new(None)),
             preserve_processes_on_drop: true,
+            compression,
             detect_handle: Some(tokio::spawn(async {}).abort_handle()),
         };
 
@@ -3688,11 +4035,14 @@ mod tests {
         let (tx, mut rx) = mpsc::channel(4);
         let (resize_tx, _resize_rx) = watch::channel((80, 24, 0, 0));
         let terminal = crate::ghostty::Terminal::new(80, 24, 0).unwrap();
+        let pane_id = PaneId::from_raw(0);
+        let terminal = Arc::new(PaneTerminal::new(
+            GhosttyPaneTerminal::new(terminal, tx.clone()).unwrap(),
+        ));
+        let compression = TerminalCompressionTask::spawn(pane_id, terminal.clone());
         let runtime = PaneRuntime {
-            pane_id: PaneId::from_raw(0),
-            terminal: Arc::new(PaneTerminal::new(
-                GhosttyPaneTerminal::new(terminal, tx.clone()).unwrap(),
-            )),
+            pane_id,
+            terminal,
             io: PaneRuntimeIo::TestChannel {
                 sender: tx,
                 resize_tx,
@@ -3709,6 +4059,7 @@ mod tests {
             detect_reset_notify: Arc::new(Notify::new()),
             pending_release: Arc::new(Mutex::new(None)),
             preserve_processes_on_drop: true,
+            compression,
             detect_handle: Some(tokio::spawn(async {}).abort_handle()),
         };
 
diff --git a/src/pane/terminal.rs b/src/pane/terminal.rs
index 154a9539..ecc12ed5 100644
--- a/src/pane/terminal.rs
+++ b/src/pane/terminal.rs
@@ -176,6 +176,13 @@ pub(crate) struct TerminalReadSnapshot {
     pub truncated: bool,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum TerminalCompressionStep {
+    Busy,
+    ActivityChanged(u64),
+    Compressed(crate::ghostty::TerminalCompressionResult),
+}
+
 pub(crate) struct GhosttyPaneTerminal {
     pub core: Mutex<GhosttyPaneCore>,
     key_encoder: Mutex<crate::ghostty::KeyEncoder>,
@@ -479,6 +486,18 @@ impl PaneTerminal {
         self.ghostty.detection_text()
     }
 
+    pub(crate) fn try_compression_activity(&self) -> Result<Option<u64>, crate::ghostty::Error> {
+        self.ghostty.try_compression_activity()
+    }
+
+    pub(crate) fn try_compress_incremental_if_activity(
+        &self,
+        expected_activity: u64,
+    ) -> Result<TerminalCompressionStep, crate::ghostty::Error> {
+        self.ghostty
+            .try_compress_incremental_if_activity(expected_activity)
+    }
+
     pub(crate) fn recent_text_snapshot(&self, lines: usize) -> TerminalReadSnapshot {
         self.ghostty.recent_text_snapshot(lines)
     }
@@ -2074,6 +2093,37 @@ impl GhosttyPaneTerminal {
             .unwrap_or_default()
     }
 
+    fn try_lock_core(&self) -> Option<std::sync::MutexGuard<'_, GhosttyPaneCore>> {
+        match self.core.try_lock() {
+            Ok(core) => Some(core),
+            Err(std::sync::TryLockError::WouldBlock) => None,
+            Err(std::sync::TryLockError::Poisoned(poisoned)) => Some(poisoned.into_inner()),
+        }
+    }
+
+    pub(crate) fn try_compression_activity(&self) -> Result<Option<u64>, crate::ghostty::Error> {
+        let Some(core) = self.try_lock_core() else {
+            return Ok(None);
+        };
+        core.terminal.compression_activity().map(Some)
+    }
+
+    pub(crate) fn try_compress_incremental_if_activity(
+        &self,
+        expected_activity: u64,
+    ) -> Result<TerminalCompressionStep, crate::ghostty::Error> {
+        let Some(mut core) = self.try_lock_core() else {
+            return Ok(TerminalCompressionStep::Busy);
+        };
+        let activity = core.terminal.compression_activity()?;
+        if activity != expected_activity {
+            return Ok(TerminalCompressionStep::ActivityChanged(activity));
+        }
+        core.terminal
+            .compress_incremental()
+            .map(TerminalCompressionStep::Compressed)
+    }
+
     #[cfg(test)]
     pub fn recent_text(&self, lines: usize) -> String {
         self.recent_text_snapshot(lines).text
diff --git a/src/server/headless/client_views.rs b/src/server/headless/client_views.rs
index 3a0f895c..2d49a50a 100644
--- a/src/server/headless/client_views.rs
+++ b/src/server/headless/client_views.rs
@@ -281,6 +281,33 @@ impl HeadlessServer {
         )
     }
 
+    fn public_request_may_change_geometry(method: &api::schema::Method) -> bool {
+        use api::schema::Method;
+
+        matches!(
+            method,
+            Method::CommandInvoke(_)
+                | Method::LayoutSetSplitRatio(_)
+                | Method::PaneClose(_)
+                | Method::PaneEditScrollback(_)
+                | Method::PaneFocus(_)
+                | Method::PaneFocusDirection(_)
+                | Method::PaneResize(_)
+                | Method::PaneSplit(_)
+                | Method::PaneSwap(_)
+                | Method::PaneZoom(_)
+                | Method::TabClose(_)
+                | Method::TabCreate(_)
+                | Method::TabFocus(_)
+                | Method::WorkspaceClose(_)
+                | Method::WorkspaceCreate(_)
+                | Method::WorkspaceFocus(_)
+                | Method::WorktreeCreate(_)
+                | Method::WorktreeOpen(_)
+                | Method::WorktreeRemove(_)
+        )
+    }
+
     pub(super) fn deferred_endpoint_navigation_tab_id(response: &[u8]) -> Option<String> {
         let response = serde_json::from_slice::<serde_json::Value>(response).ok()?;
         if response.pointer("/result/type")?.as_str()? != "worktree_created" {
@@ -522,6 +549,18 @@ impl HeadlessServer {
         target: crate::ui::TabSurfaceTarget,
         start_pending_agent_resumes: bool,
     ) -> bool {
+        if !self.resize_shell_tab_geometry_to_target(client_id, target) {
+            return false;
+        }
+        self.finish_shell_tab_geometry_change(start_pending_agent_resumes);
+        true
+    }
+
+    fn resize_shell_tab_geometry_to_target(
+        &mut self,
+        client_id: u64,
+        target: crate::ui::TabSurfaceTarget,
+    ) -> bool {
         let Some(client) = self.clients.get(&client_id) else {
             return false;
         };
@@ -562,7 +601,6 @@ impl HeadlessServer {
         {
             let _ = resize_popup_runtime(&self.app, Rect::new(0, 0, cols, rows), cell_size);
         }
-        self.finish_shell_tab_geometry_change(start_pending_agent_resumes);
         true
     }
 
@@ -581,6 +619,66 @@ impl HeadlessServer {
         self.apply_shell_tab_geometry(client_id, start_pending_agent_resumes)
     }
 
+    pub(super) fn reapply_controlled_shell_tab_geometry(
+        &mut self,
+        start_pending_agent_resumes: bool,
+    ) -> bool {
+        let mut viewed_tabs = HashMap::<String, Vec<u64>>::new();
+        for (&client_id, client) in &self.clients {
+            if !client.is_shell_client() || client.writer.is_none() {
+                continue;
+            }
+            let Some(tab_id) = self.shell_tab_id_for_client(client_id) else {
+                continue;
+            };
+            viewed_tabs.entry(tab_id).or_default().push(client_id);
+        }
+        for viewers in viewed_tabs.values_mut() {
+            viewers.sort_unstable();
+        }
+        for (tab_id, viewers) in viewed_tabs {
+            let controller_is_viewing = self
+                .tab_geometry_controllers
+                .get(&tab_id)
+                .is_some_and(|controller| viewers.contains(controller));
+            if !controller_is_viewing {
+                self.tab_geometry_controllers.insert(tab_id, viewers[0]);
+            }
+        }
+
+        if self.resize_tabs_for_only_shell_client(start_pending_agent_resumes) {
+            return true;
+        }
+
+        let mut controlled_tabs = self
+            .tab_geometry_controllers
+            .iter()
+            .filter_map(|(tab_id, &client_id)| {
+                self.app
+                    .parse_tab_id(tab_id)
+                    .map(|(workspace_index, tab_index)| {
+                        (
+                            tab_id.clone(),
+                            client_id,
+                            crate::ui::TabSurfaceTarget {
+                                workspace_index,
+                                tab_index,
+                            },
+                        )
+                    })
+            })
+            .collect::<Vec<_>>();
+        controlled_tabs.sort_unstable_by(|left, right| left.0.cmp(&right.0));
+        let mut reapplied = false;
+        for (_, client_id, target) in controlled_tabs {
+            reapplied |= self.resize_shell_tab_geometry_to_target(client_id, target);
+        }
+        if reapplied {
+            self.finish_shell_tab_geometry_change(start_pending_agent_resumes);
+        }
+        reapplied
+    }
+
     pub(super) fn claim_shell_tab_geometry(
         &mut self,
         client_id: u64,
@@ -683,6 +781,7 @@ impl HeadlessServer {
     ) -> bool {
         let target_before = self.default_shell_target();
         let popup_before = self.app.state.popup_pane.is_some();
+        let method_claims_geometry = Self::public_request_may_change_geometry(&msg.request.method);
         let explicit_public_focus_target = match &msg.request.method {
             api::schema::Method::WorkspaceFocus(params) => self
                 .app
@@ -742,7 +841,9 @@ impl HeadlessServer {
         if reconcile || target_changed || self.app.state.popup_pane.is_some() != popup_before {
             self.reconcile_client_shell_locations();
         }
-        changed
+        let geometry_changed =
+            method_claims_geometry && self.reapply_controlled_shell_tab_geometry(false);
+        changed | geometry_changed
     }
 
     pub(super) fn handle_client_shell_api_request(
@@ -782,8 +883,12 @@ impl HeadlessServer {
             );
         }
         let geometry_changed = method_claims_geometry
-            && (self.claim_shell_tab_geometry(client_id, false)
-                || self.resize_shell_tab_if_controller(client_id, false));
+            && if reconcile {
+                self.reapply_controlled_shell_tab_geometry(false)
+            } else {
+                self.claim_shell_tab_geometry(client_id, false)
+                    || self.resize_shell_tab_if_controller(client_id, false)
+            };
         changed | navigation_changed | geometry_changed
     }
 }
diff --git a/src/server/headless/notifications.rs b/src/server/headless/notifications.rs
index 6764dd2d..dd661809 100644
--- a/src/server/headless/notifications.rs
+++ b/src/server/headless/notifications.rs
@@ -652,6 +652,7 @@ impl HeadlessServer {
                 }
                 self.reconcile_client_shell_locations();
                 self.finish_shell_location_reconciliation(focus_before, &focused_tabs_before);
+                self.reapply_controlled_shell_tab_geometry(false);
                 for (pane_id, terminal_id) in shutdown_terminals {
                     if self.app.find_pane(pane_id).is_none() {
                         self.shutdown_terminal_stream_clients(
@@ -698,6 +699,7 @@ impl HeadlessServer {
                 }
                 self.reconcile_client_shell_locations();
                 self.finish_shell_location_reconciliation(focus_before, &focused_tabs_before);
+                self.reapply_controlled_shell_tab_geometry(false);
 
                 if self.app.find_pane(pane_id_val).is_none() {
                     if let Some(terminal_id) = terminal_id {
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 29a4cfbf..77a16fc7 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -1578,6 +1578,108 @@ async fn repeated_layout_action_reapplies_controller_geometry() {
 }
 
 #[tokio::test]
+async fn public_close_reapplies_controller_geometry() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("public-close-geometry");
+    let first_pane = workspace.tabs[0].root_pane;
+    let second_pane = workspace.test_split(ratatui::layout::Direction::Vertical);
+    workspace.insert_test_runtime(
+        first_pane,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b""),
+    );
+    workspace.insert_test_runtime(
+        second_pane,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b""),
+    );
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+    server.app.state.mode = crate::app::Mode::Terminal;
+    let second_pane_id = server.app.public_pane_id(0, second_pane).unwrap();
+
+    let (control, _) = connect_test_shell(&mut server, 66, 100, 30);
+    let _ = control.recv().expect("snapshot");
+    let shrunk = server.app.state.workspaces[0].test_runtimes[&first_pane].current_size();
+    assert!(shrunk.0 < 30);
+
+    let (respond_to, _response_rx) = std::sync::mpsc::channel();
+    assert!(
+        server.handle_api_request_with_shutdown_check(crate::api::ApiRequestMessage {
+            request: crate::api::schema::Request {
+                id: "public-close-geometry".into(),
+                method: crate::api::schema::Method::PaneClose(crate::api::schema::PaneTarget {
+                    pane_id: second_pane_id,
+                }),
+            },
+            respond_to,
+            response_write_complete: None,
+            stream_active: None,
+        })
+    );
+
+    let runtime = &server.app.state.workspaces[0].test_runtimes[&first_pane];
+    let grown = runtime.current_size();
+    assert!(grown.0 > shrunk.0);
+    assert_eq!(runtime.terminal_dimensions(), Some((grown.1, grown.0)));
+    assert_eq!(
+        runtime.scroll_metrics().unwrap().viewport_rows,
+        grown.0 as usize
+    );
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn geometry_reapply_replaces_a_controller_that_left_the_tab() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("geometry-controller-viewer");
+    let first_pane = workspace.tabs[0].root_pane;
+    let second_tab = workspace.test_add_tab(Some("second"));
+    let second_pane = workspace.tabs[second_tab].root_pane;
+    let third_tab = workspace.test_add_tab(Some("third"));
+    let third_pane = workspace.tabs[third_tab].root_pane;
+    for pane_id in [first_pane, second_pane, third_pane] {
+        workspace.insert_test_runtime(
+            pane_id,
+            crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b""),
+        );
+    }
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+    server.app.state.mode = crate::app::Mode::Terminal;
+    let second_tab_id = server.app.public_tab_id(0, second_tab).unwrap();
+    let third_tab_id = server.app.public_tab_id(0, third_tab).unwrap();
+
+    let (first_control, _) = connect_test_shell(&mut server, 67, 100, 30);
+    let (second_control, _) = connect_test_shell(&mut server, 68, 70, 20);
+    let _ = first_control.recv().expect("first snapshot");
+    let _ = second_control.recv().expect("second snapshot");
+
+    assert!(server.focus_shell_client_on_tab(67, &second_tab_id));
+    assert!(server.claim_shell_tab_geometry(67, false));
+    assert!(server.focus_shell_client_on_tab(67, &third_tab_id));
+    assert!(server.claim_shell_tab_geometry(67, false));
+    assert!(server.focus_shell_client_on_tab(68, &second_tab_id));
+    assert_eq!(
+        server.tab_geometry_controllers.get(&second_tab_id),
+        Some(&67)
+    );
+    let stale_size = server.app.state.workspaces[0].test_runtimes[&second_pane].current_size();
+
+    assert!(server.reapply_controlled_shell_tab_geometry(false));
+
+    assert_eq!(
+        server.tab_geometry_controllers.get(&second_tab_id),
+        Some(&68)
+    );
+    assert_ne!(
+        server.app.state.workspaces[0].test_runtimes[&second_pane].current_size(),
+        stale_size
+    );
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
 async fn client_shell_tabs_render_accept_input_and_resize_independently() {
     let mut server = test_headless_server();
     let mut workspace = crate::workspace::Workspace::test_new("independent-geometry");
@@ -3436,8 +3538,8 @@ async fn pane_death_reconciles_each_client_view_and_focus() {
         .public_tab_id(0, second_tab)
         .expect("second tab id");
 
-    let (first_control, _) = connect_matching_test_shell(&mut server, 71);
-    let (second_control, _) = connect_matching_test_shell(&mut server, 72);
+    let (first_control, _) = connect_test_shell(&mut server, 71, 100, 30);
+    let (second_control, _) = connect_test_shell(&mut server, 72, 70, 20);
     let _ = first_control.recv().expect("first snapshot");
     let _ = second_control.recv().expect("second snapshot");
     assert!(server.focus_shell_client_on_tab(72, &second_tab_id));
@@ -3464,6 +3566,60 @@ async fn pane_death_reconciles_each_client_view_and_focus() {
         second_input.try_recv().is_err(),
         "focus gain was duplicated"
     );
+    assert_eq!(
+        server.tab_geometry_controllers.get(&second_tab_id),
+        Some(&71)
+    );
+    let before_resize = server.app.state.workspaces[0].test_runtimes[&second_pane].current_size();
+    assert!(server.handle_server_event(ServerEvent::ClientShellResize {
+        client_id: 71,
+        surface_cols: 90,
+        surface_rows: 25,
+        cell_width_px: 0,
+        cell_height_px: 0,
+        pixel_mouse: false,
+    }));
+    assert_ne!(
+        server.app.state.workspaces[0].test_runtimes[&second_pane].current_size(),
+        before_resize
+    );
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn pane_death_reapplies_controller_geometry() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("pane-death-geometry");
+    let first_pane = workspace.tabs[0].root_pane;
+    let dead_pane = workspace.test_split(ratatui::layout::Direction::Vertical);
+    workspace.insert_test_runtime(
+        first_pane,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b""),
+    );
+    workspace.insert_test_runtime(
+        dead_pane,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b""),
+    );
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+    server.app.state.mode = crate::app::Mode::Terminal;
+
+    let (control, _) = connect_test_shell(&mut server, 73, 185, 46);
+    let _ = control.recv().expect("snapshot");
+    let shrunk = server.app.state.workspaces[0].test_runtimes[&first_pane].current_size();
+    assert!(shrunk.0 < 46);
+
+    assert!(server.handle_internal_event_with_forwarding(AppEvent::PaneDied { pane_id: dead_pane }));
+
+    let runtime = &server.app.state.workspaces[0].test_runtimes[&first_pane];
+    let grown = runtime.current_size();
+    assert!(grown.0 > shrunk.0);
+    assert_eq!(runtime.terminal_dimensions(), Some((grown.1, grown.0)));
+    assert_eq!(
+        runtime.scroll_metrics().unwrap().viewport_rows,
+        grown.0 as usize
+    );
     shutdown_test_runtimes(&mut server);
 }