diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/ghostty/mod.rs | 74 | ||||
| -rw-r--r-- | src/pane.rs | 405 | ||||
| -rw-r--r-- | src/pane/terminal.rs | 50 | ||||
| -rw-r--r-- | src/server/headless/client_views.rs | 113 | ||||
| -rw-r--r-- | src/server/headless/notifications.rs | 2 | ||||
| -rw-r--r-- | src/server/headless/tests/mod.rs | 160 |
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); } |