perf: send scrolling pane output as row shifts and skip blank cells (#4711)

* perf: send scrolling pane output as row shifts and skip blank cells

* fix: keep expanded scroll patches disjoint for the fast presentation path

18 files changed, 1001 insertions, 34 deletions

diff --git a/crates/ghostty-vt/src/lib.rs b/crates/ghostty-vt/src/lib.rs
index 056ff142..7ee1e751 100644
--- a/crates/ghostty-vt/src/lib.rs
+++ b/crates/ghostty-vt/src/lib.rs
@@ -3381,6 +3381,17 @@ impl<'a> RowCellIter<'a> {
         unsafe { ffi::ghostty_render_state_row_cells_select(self.cells.raw, x).into_result() }
     }
 
+    /// Whether the current cell was never written: no text, default style,
+    /// narrow, and no hyperlink. One read replaces the style and text lookups
+    /// for the blank cells that fill most wide panes.
+    pub fn is_default_blank(&self) -> Result<bool, Error> {
+        // Packed cell bits 0..44 hold content, style id, and width; bit 45 is
+        // the hyperlink flag. Protection (44) and OSC 133 semantic marks
+        // (46..48) do not change how a cell renders.
+        const PRESENTATION_BITS: u64 = ((1 << 44) - 1) | (1 << 45);
+        Ok(self.raw_cell()? & PRESENTATION_BITS == 0)
+    }
+
     fn raw_cell(&self) -> Result<ffi::GhosttyCell, Error> {
         let mut raw = ffi::GhosttyCell::default();
         unsafe {
diff --git a/src/client/handshake.rs b/src/client/handshake.rs
index f25cc1b0..2520286c 100644
--- a/src/client/handshake.rs
+++ b/src/client/handshake.rs
@@ -200,6 +200,7 @@ pub(super) fn do_handshake(
             surface_active,
             surface_reuse: true,
             surface_delta: true,
+            surface_scroll: true,
             snapshot_codecs: vec![SNAPSHOT_CODEC_V1.into()],
             surface_codecs: vec![SURFACE_CODEC_V1.into()],
             input_codecs: vec![INPUT_CODEC_V1.into()],
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 2cec6d2d..b78517ac 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -145,6 +145,16 @@ use crate::protocol::{self, ClientMessage, ServerMessage, MAX_GRAPHICS_FRAME_SIZ
 use crate::protocol::{AttachScrollDirection, AttachScrollSource, NotifyKind};
 use crate::server::socket_paths::client_socket_path;
 
+/// The decoder for whichever optional surface encodings this connection negotiated.
+fn negotiated_surface_decoder(
+    negotiation: &endpoint::EndpointNegotiation,
+) -> Option<protocol::surface_reuse::Decoder> {
+    let reuse = negotiation.supports_capability(protocol::surface_reuse::CAPABILITY);
+    let delta = negotiation.supports_capability(protocol::surface_delta::CAPABILITY);
+    let scroll = negotiation.supports_capability(protocol::surface_scroll::CAPABILITY);
+    (reuse || delta || scroll).then(|| protocol::surface_reuse::Decoder::new(delta, scroll))
+}
+
 fn run_client_with_mode(
     attach_request: Option<(String, bool)>,
     attach_escape: Option<AttachEscapeState>,
@@ -588,10 +598,7 @@ async fn run_client_loop(
             handshake.endpoint_methods.unwrap_or_default(),
             handshake.endpoint_capabilities.unwrap_or_default(),
         );
-        let surface_reuse = negotiation.supports_capability(protocol::surface_reuse::CAPABILITY);
-        let surface_delta = negotiation.supports_capability(protocol::surface_delta::CAPABILITY);
-        let surface_decoder = (surface_reuse || surface_delta)
-            .then(|| protocol::surface_reuse::Decoder::new(surface_delta));
+        let surface_decoder = negotiated_surface_decoder(&negotiation);
         let transport = start_endpoint_transport(
             stream,
             (),
@@ -1285,10 +1292,7 @@ async fn run_client_loop(
                     ) {
                         continue;
                     }
-                    let surface_reuse =
-                        negotiation.supports_capability(protocol::surface_reuse::CAPABILITY);
-                    let surface_delta =
-                        negotiation.supports_capability(protocol::surface_delta::CAPABILITY);
+                    let surface_decoder = negotiated_surface_decoder(&negotiation);
                     let agent_view_projection_supported = negotiation.supports_capability(
                         crate::protocol::endpoint::AGENT_VIEW_PROJECTION_CAPABILITY,
                     );
@@ -1313,8 +1317,6 @@ async fn run_client_loop(
                     if let Some(frame) = frame {
                         state.present_frame(frame);
                     }
-                    let surface_decoder = (surface_reuse || surface_delta)
-                        .then(|| protocol::surface_reuse::Decoder::new(surface_delta));
                     let reader_tx = event_tx.clone();
                     std::thread::spawn(move || {
                         server_reader_thread(
diff --git a/src/pane/terminal.rs b/src/pane/terminal.rs
index 8ac4bd1c..24379d22 100644
--- a/src/pane/terminal.rs
+++ b/src/pane/terminal.rs
@@ -2677,6 +2677,7 @@ fn ghostty_collect_dirty_patch(
     let mut grapheme_bytes = Vec::new();
     let mut symbol_scratch = String::new();
     let mut patch_rows = Vec::new();
+    let blank = blank_cell_data(default_fg, default_bg);
     while let Some(y) = rows.next_dirty() {
         if y >= area_height {
             break;
@@ -2692,6 +2693,15 @@ fn ghostty_collect_dirty_patch(
         let mut patch_cells = Vec::with_capacity(usize::from(area_width));
         let mut x = 0u16;
         while x < area_width && cells.next() {
+            match cells.is_default_blank() {
+                Ok(true) => {
+                    patch_cells.push(blank.clone());
+                    x += 1;
+                    continue;
+                }
+                Ok(false) => {}
+                Err(_) => fallback!("raw_cell_error"),
+            }
             let Ok(basic) = cells.basic_data() else {
                 fallback!("basic_data_error");
             };
@@ -2720,10 +2730,7 @@ fn ghostty_collect_dirty_patch(
             patch_cells.push(cell_data_from_style(symbol, style));
             x += 1;
         }
-        while x < area_width {
-            patch_cells.push(blank_cell_data(default_fg, default_bg));
-            x += 1;
-        }
+        patch_cells.resize(usize::from(area_width), blank.clone());
         patch_rows.push((y, patch_cells));
     }
 
diff --git a/src/pane/terminal/migration_tests.rs b/src/pane/terminal/migration_tests.rs
index 09e5df01..aaf91f27 100644
--- a/src/pane/terminal/migration_tests.rs
+++ b/src/pane/terminal/migration_tests.rs
@@ -311,6 +311,13 @@ fn incremental_rows_reconstruct_full_render() {
         b"\x1b[3;4H",
         b"\x1b[?1049hALT",
         b"\x1b[?1049l",
+        // Blank cells must match the full renderer under every color source.
+        b"\x1b[3;1H\x1b[44m\x1b[2K\x1b[0m",
+        b"\x1b[4;2H\x1b[7m \x1b[0m \x1b[8m \x1b[0m",
+        b"\x1b[?5h",
+        b"\x1b[?5l",
+        b"\x1b[5;1H\x1b]133;A\x07$ \x1b]133;B\x07  \x1b]133;C\x07",
+        b"\x1b[5;1H\r\n\r\nscrolled\r\n",
     ] {
         incremental.write(bytes);
         full.write(bytes);
diff --git a/src/protocol/endpoint.rs b/src/protocol/endpoint.rs
index 2704bf65..d1e577c4 100644
--- a/src/protocol/endpoint.rs
+++ b/src/protocol/endpoint.rs
@@ -61,6 +61,9 @@ pub struct EndpointClientHello {
     /// Accept the optional surface-delta encoding on this connection.
     #[serde(default)]
     pub surface_delta: bool,
+    /// Accept the optional scroll-aware patch encoding on this connection.
+    #[serde(default)]
+    pub surface_scroll: bool,
     #[serde(default)]
     pub snapshot_codecs: Vec<String>,
     #[serde(default)]
@@ -165,6 +168,7 @@ impl EndpointServerWelcome {
             capabilities: vec![
                 super::surface_reuse::CAPABILITY.into(),
                 super::surface_delta::CAPABILITY.into(),
+                super::surface_scroll::CAPABILITY.into(),
                 SURFACE_INTEREST_CAPABILITY.into(),
                 PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(),
                 HEALTH_CHECK_CAPABILITY.into(),
@@ -210,6 +214,7 @@ mod tests {
             surface_active: true,
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             snapshot_codecs: vec![SNAPSHOT_CODEC_V1.into()],
             surface_codecs: vec![SURFACE_CODEC_V1.into()],
             input_codecs: vec![INPUT_CODEC_V1.into()],
@@ -358,10 +363,12 @@ mod tests {
         value.as_object_mut().unwrap().remove("surface_active");
         value.as_object_mut().unwrap().remove("surface_reuse");
         value.as_object_mut().unwrap().remove("surface_delta");
+        value.as_object_mut().unwrap().remove("surface_scroll");
         let decoded: EndpointClientHello = serde_json::from_value(value).unwrap();
         assert!(decoded.surface_active);
         assert!(!decoded.surface_reuse);
         assert!(!decoded.surface_delta);
+        assert!(!decoded.surface_scroll);
     }
 
     #[test]
@@ -372,6 +379,7 @@ mod tests {
             vec![
                 super::super::surface_reuse::CAPABILITY.to_string(),
                 super::super::surface_delta::CAPABILITY.to_string(),
+                super::super::surface_scroll::CAPABILITY.to_string(),
                 SURFACE_INTEREST_CAPABILITY.to_string(),
                 PRESENTATION_EFFECTS_FENCE_CAPABILITY.to_string(),
                 HEALTH_CHECK_CAPABILITY.to_string(),
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index 24e2ebd2..6a3cac7a 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -4,6 +4,7 @@ pub mod endpoint;
 pub(crate) mod render_ansi;
 pub(crate) mod surface_delta;
 pub(crate) mod surface_reuse;
+pub(crate) mod surface_scroll;
 mod wire;
 
 pub use wire::*;
diff --git a/src/protocol/surface_delta.rs b/src/protocol/surface_delta.rs
index 8f94633b..b0abe4da 100644
--- a/src/protocol/surface_delta.rs
+++ b/src/protocol/surface_delta.rs
@@ -224,7 +224,7 @@ mod tests {
     }
 
     fn reconstruct(last: &PaneSurfaceFrame, next: &PaneSurfaceFrame) -> ServerMessage {
-        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true, false);
         decoder
             .decode(ServerMessage::PaneSurface(last.clone()))
             .unwrap();
@@ -384,7 +384,7 @@ mod tests {
                 8 => corrupt.surface.projection_revision = 0,
                 _ => unreachable!(),
             }
-            let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+            let mut decoder = crate::protocol::surface_reuse::Decoder::new(true, false);
             decoder
                 .decode(ServerMessage::PaneSurface(last.clone()))
                 .unwrap();
@@ -409,7 +409,7 @@ mod tests {
         let mut next = last.clone();
         next.surface_revision += 1;
         next.frame.cells[0].symbol = "a".into();
-        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true, false);
         decoder.decode(ServerMessage::PaneSurface(last)).unwrap();
         decoder
             .decode(ServerMessage::PaneSurfacePatch(
@@ -512,7 +512,7 @@ mod tests {
             .decode(ServerMessage::PaneSurface(last.clone()))
             .unwrap();
         assert!(legacy.decode(update.clone()).is_err());
-        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true, false);
         decoder.decode(ServerMessage::PaneSurface(last)).unwrap();
         let ServerMessage::EndpointControl { data, .. } = &update else {
             panic!("delta");
diff --git a/src/protocol/surface_reuse.rs b/src/protocol/surface_reuse.rs
index aa2187d4..ec430492 100644
--- a/src/protocol/surface_reuse.rs
+++ b/src/protocol/surface_reuse.rs
@@ -53,6 +53,16 @@ struct CellBaseline {
     popup: Option<PopupBaseline>,
 }
 
+impl CellBaseline {
+    /// Whether a patch applies to exactly this baseline as its next revision.
+    fn matches_patch(&self, patch: &super::PaneSurfacePatch) -> bool {
+        patch.boot_id == self.boot_id
+            && patch.projection_revision == self.projection_revision
+            && patch.base_surface_revision == self.surface_revision
+            && patch.surface_revision == self.surface_revision.saturating_add(1)
+    }
+}
+
 struct PopupBaseline {
     terminal_id: String,
     width: u16,
@@ -75,13 +85,15 @@ fn popup_baseline(surface: &PaneSurfaceFrame) -> Option<PopupBaseline> {
 pub(crate) struct Decoder {
     baseline: Option<CellBaseline>,
     surface_delta: bool,
+    surface_scroll: bool,
 }
 
 impl Decoder {
-    pub(crate) fn new(surface_delta: bool) -> Self {
+    pub(crate) fn new(surface_delta: bool, surface_scroll: bool) -> Self {
         Self {
             baseline: None,
             surface_delta,
+            surface_scroll,
         }
     }
 
@@ -95,6 +107,16 @@ impl Decoder {
                 }
                 return self.decode_delta(&data).map(ServerMessage::PaneSurface);
             }
+            ServerMessage::EndpointControl { kind, data }
+                if kind == super::surface_scroll::MESSAGE_KIND =>
+            {
+                if !self.surface_scroll {
+                    return Err("surface scroll was not negotiated".into());
+                }
+                return self
+                    .decode_scroll(&data)
+                    .map(ServerMessage::PaneSurfacePatch);
+            }
             ServerMessage::EndpointControl { kind, data } if kind == MESSAGE_KIND => {
                 let reuse: SurfaceReuse<PaneSurfaceFrame> = serde_json::from_str(&data)
                     .map_err(|error| format!("invalid surface reuse: {error}"))?;
@@ -136,11 +158,7 @@ impl Decoder {
             }
             ServerMessage::PaneSurfacePatch(patch) => {
                 if let Some(base) = &mut self.baseline {
-                    if patch.boot_id != base.boot_id
-                        || patch.projection_revision != base.projection_revision
-                        || patch.base_surface_revision != base.surface_revision
-                        || patch.surface_revision != base.surface_revision.saturating_add(1)
-                    {
+                    if !base.matches_patch(patch) {
                         self.baseline = None;
                     } else {
                         for row in &patch.rows {
@@ -164,6 +182,19 @@ impl Decoder {
         Ok(message)
     }
 
+    fn decode_scroll(&mut self, data: &str) -> Result<super::PaneSurfacePatch, String> {
+        let Some(base) = &mut self.baseline else {
+            return Err("surface scroll without a baseline".into());
+        };
+        let scroll = super::surface_scroll::decode(data)?;
+        if !base.matches_patch(&scroll.patch) {
+            return Err("surface scroll does not match its baseline".into());
+        }
+        let patch = super::surface_scroll::apply(&mut base.cells, base.width, base.height, scroll)?;
+        base.surface_revision = patch.surface_revision;
+        Ok(patch)
+    }
+
     fn decode_delta(&mut self, data: &str) -> Result<PaneSurfaceFrame, String> {
         use super::surface_delta::{self, GridUpdate};
         let Some(base) = &mut self.baseline else {
diff --git a/src/protocol/surface_scroll.rs b/src/protocol/surface_scroll.rs
new file mode 100644
index 00000000..4e9a85ba
--- /dev/null
+++ b/src/protocol/surface_scroll.rs
@@ -0,0 +1,720 @@
+//! Optional scroll-aware delivery of retained pane surface patches.
+//!
+//! Scrolling output moves every visible row, so a row diff resends the whole
+//! pane on each frame. This codec sends each pane's vertical shift once, plus
+//! only the rows that still differ after it. The receiver expands the result
+//! back into an ordinary [`PaneSurfacePatch`] against its retained grid.
+
+use base64::{engine::general_purpose::STANDARD_NO_PAD, Engine as _};
+
+use super::{
+    CellData, FrameData, PaneSurfaceFrame, PaneSurfacePatch, PaneSurfacePatchRow, ServerMessage,
+    SurfaceRect, MAX_FRAME_SIZE,
+};
+
+pub(crate) const CAPABILITY: &str = "surface_scroll";
+pub(crate) const MESSAGE_KIND: &str = "endpoint.surface-scroll.v1";
+
+// This byte layout is frozen with MESSAGE_KIND: one count byte, then per
+// scroll x, y, width, height (u16 LE) and shift (i16 LE), then one framed
+// `ServerMessage::PaneSurfacePatch` decoded by the shared bounded reader.
+const MAX_SCROLLS: usize = 64;
+const SCROLL_BYTES: usize = 10;
+
+/// Reorders the rows of one pane region before the patch rows apply.
+///
+/// A positive `shift` moves content up: row `y` shows the previous row
+/// `y + shift`. Rows that scroll out of the region rotate into the vacated
+/// rows, so applying a scroll never clones cell content.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct SurfaceScroll {
+    pub(crate) rect: SurfaceRect,
+    pub(crate) shift: i16,
+}
+
+#[derive(Debug)]
+pub(crate) struct ScrollPatch {
+    pub(crate) scrolls: Vec<SurfaceScroll>,
+    pub(crate) patch: PaneSurfacePatch,
+}
+
+/// The row swaps that define [`SurfaceScroll`]. Both peers must derive the
+/// same row order, so encoder and decoder share this one sequence.
+fn for_each_swap(height: usize, shift: i16, mut swap: impl FnMut(usize, usize)) {
+    let distance = usize::from(shift.unsigned_abs());
+    if shift > 0 {
+        for y in 0..height - distance {
+            swap(y, y + distance);
+        }
+    } else {
+        for y in (distance..height).rev() {
+            swap(y, y - distance);
+        }
+    }
+}
+
+fn scroll_fits(scroll: &SurfaceScroll, width: u16, height: u16) -> bool {
+    let rect = scroll.rect;
+    rect.width > 0
+        && rect.height >= 2
+        && scroll.shift != 0
+        && scroll.shift.unsigned_abs() < rect.height
+        && rect
+            .x
+            .checked_add(rect.width)
+            .is_some_and(|end| end <= width)
+        && rect
+            .y
+            .checked_add(rect.height)
+            .is_some_and(|end| end <= height)
+}
+
+fn rects_overlap(a: SurfaceRect, b: SurfaceRect) -> bool {
+    a.x < b.x + b.width && b.x < a.x + a.width && a.y < b.y + b.height && b.y < a.y + a.height
+}
+
+/// Pane regions never overlap. Requiring that bounds expansion to one copy of the grid.
+fn scrolls_disjoint(scrolls: &[SurfaceScroll]) -> bool {
+    scrolls.iter().enumerate().all(|(index, scroll)| {
+        scrolls[index + 1..]
+            .iter()
+            .all(|other| !rects_overlap(scroll.rect, other.rect))
+    })
+}
+
+fn row_fits(row: &PaneSurfacePatchRow, width: u16, height: u16) -> bool {
+    row.y < height && usize::from(row.x).saturating_add(row.cells.len()) <= usize::from(width)
+}
+
+/// Applies one validated scroll to a `width`-wide row-major grid.
+fn apply_scroll(cells: &mut [CellData], width: u16, scroll: &SurfaceScroll) {
+    let rect = scroll.rect;
+    let stride = usize::from(width);
+    let row_start = |y: usize| (usize::from(rect.y) + y) * stride + usize::from(rect.x);
+    for_each_swap(usize::from(rect.height), scroll.shift, |a, b| {
+        let (a, b) = (row_start(a), row_start(b));
+        for x in 0..usize::from(rect.width) {
+            cells.swap(a + x, b + x);
+        }
+    });
+}
+
+/// One pane row as the patch leaves it: committed cells overlaid by the patch
+/// spans that touch the row, read in place so detection copies nothing.
+struct RowView<'a> {
+    base: &'a [CellData],
+    left: usize,
+    spans: Vec<&'a PaneSurfacePatchRow>,
+}
+
+impl RowView<'_> {
+    fn cell(&self, x: usize) -> &CellData {
+        let column = self.left + x;
+        for span in self.spans.iter().rev() {
+            let start = usize::from(span.x);
+            if column >= start && column < start + span.cells.len() {
+                return &span.cells[column - start];
+            }
+        }
+        &self.base[x]
+    }
+
+    fn hash(&self) -> u64 {
+        (0..self.base.len()).fold(0, |hash, x| cell_hash(hash, self.cell(x)))
+    }
+}
+
+// Row fingerprints only nominate a shift; the residual diff compares real
+// cells, so a collision can cost bytes but never correctness.
+fn mix(hash: u64, value: u64) -> u64 {
+    (hash.rotate_left(5) ^ value).wrapping_mul(0x517c_c1b7_2722_0a95)
+}
+
+fn cell_hash(hash: u64, cell: &CellData) -> u64 {
+    let hash = cell
+        .symbol
+        .bytes()
+        .fold(hash, |hash, byte| mix(hash, u64::from(byte)));
+    let hash = mix(hash, u64::from(cell.fg) << 32 | u64::from(cell.bg));
+    mix(
+        hash,
+        u64::from(cell.modifier)
+            | u64::from(cell.skip) << 16
+            | cell.hyperlink.map_or(0, |index| u64::from(index) + 1) << 17,
+    )
+}
+
+fn fully_inside(row: &PaneSurfacePatchRow, rect: SurfaceRect) -> bool {
+    row.y >= rect.y
+        && row.y - rect.y < rect.height
+        && row.x >= rect.x
+        && usize::from(row.x) + row.cells.len() <= usize::from(rect.x) + usize::from(rect.width)
+}
+
+/// Finds the shift that best explains one pane's patch and returns it with the
+/// rows that still differ afterwards, or `None` when the shift would not at
+/// least halve the cells this pane's patch carries.
+fn pane_scroll(
+    frame: &FrameData,
+    rect: SurfaceRect,
+    rows: &[PaneSurfacePatchRow],
+) -> Option<(i16, Vec<PaneSurfacePatchRow>)> {
+    let (width, height) = (usize::from(rect.width), usize::from(rect.height));
+    let stride = usize::from(frame.width);
+    let previous = |y: usize| {
+        let start = (usize::from(rect.y) + y) * stride + usize::from(rect.x);
+        &frame.cells[start..start + width]
+    };
+
+    let mut next = (0..height)
+        .map(|y| RowView {
+            base: previous(y),
+            left: usize::from(rect.x),
+            spans: Vec::new(),
+        })
+        .collect::<Vec<_>>();
+    let mut patch_cells = 0;
+    for row in rows {
+        let overlaps = row.y >= rect.y
+            && row.y - rect.y < rect.height
+            && usize::from(row.x) < usize::from(rect.x) + width
+            && usize::from(row.x) + row.cells.len() > usize::from(rect.x);
+        if overlaps {
+            next[usize::from(row.y - rect.y)].spans.push(row);
+            if fully_inside(row, rect) {
+                patch_cells += row.cells.len();
+            }
+        }
+    }
+    // A shift can only beat the plain patch when at least two rows changed.
+    if next.iter().filter(|row| !row.spans.is_empty()).count() < 2 {
+        return None;
+    }
+
+    let old_hashes = next
+        .iter()
+        .map(|row| row.base.iter().fold(0, cell_hash))
+        .collect::<Vec<_>>();
+    let new_hashes = next
+        .iter()
+        .zip(&old_hashes)
+        .map(|(row, &old)| {
+            if row.spans.is_empty() {
+                old
+            } else {
+                row.hash()
+            }
+        })
+        .collect::<Vec<_>>();
+    let matches = |shift: isize| {
+        (0..height)
+            .filter(|&y| {
+                let source = y as isize + shift;
+                source >= 0
+                    && (source as usize) < height
+                    && new_hashes[y] == old_hashes[source as usize]
+            })
+            .count()
+    };
+    let unshifted = matches(0);
+    let (mut best_shift, mut best) = (0isize, unshifted);
+    for distance in 1..height {
+        if height - distance <= best {
+            break;
+        }
+        for shift in [distance as isize, -(distance as isize)] {
+            let count = matches(shift);
+            if count > best {
+                (best_shift, best) = (shift, count);
+            }
+        }
+    }
+    if best_shift == 0 {
+        return None;
+    }
+    let shift = i16::try_from(best_shift).ok()?;
+
+    let mut order = (0..height).collect::<Vec<_>>();
+    for_each_swap(height, shift, |a, b| order.swap(a, b));
+    let mut residual = Vec::new();
+    let mut residual_cells = 0;
+    for (y, &source) in order.iter().enumerate() {
+        let (base, target) = (previous(source), &next[y]);
+        let mut x = 0;
+        while x < width {
+            if base[x] == *target.cell(x) {
+                x += 1;
+                continue;
+            }
+            let start = x;
+            while x < width && base[x] != *target.cell(x) {
+                x += 1;
+            }
+            residual_cells += x - start;
+            if residual_cells * 2 >= patch_cells {
+                return None;
+            }
+            residual.push(PaneSurfacePatchRow {
+                x: rect.x + start as u16,
+                y: rect.y + y as u16,
+                cells: (start..x).map(|x| target.cell(x).clone()).collect(),
+            });
+        }
+    }
+    Some((shift, residual))
+}
+
+fn encoded_size(value: &impl serde::Serialize) -> Option<usize> {
+    let mut writer = bincode::enc::write::SizeWriter::default();
+    bincode::serde::encode_into_writer(value, &mut writer, bincode::config::standard()).ok()?;
+    Some(writer.bytes_written)
+}
+
+/// Encodes `patch` against the committed `last` surface as a scroll message
+/// when some pane's shift at least halves its patched cells. Returns `None`
+/// otherwise, so ordinary edits never pay for the wrapper.
+pub(crate) fn message(last: &PaneSurfaceFrame, patch: &PaneSurfacePatch) -> Option<ServerMessage> {
+    let frame = &last.frame;
+    if frame.cells.len() != usize::from(frame.width) * usize::from(frame.height) {
+        return None;
+    }
+    let mut scrolls = Vec::new();
+    let mut residual = Vec::new();
+    for pane in &patch.panes {
+        let rect = pane.inner_rect;
+        // Both peers hold the committed geometry; never scroll a region that moved.
+        let committed = last
+            .panes
+            .iter()
+            .any(|existing| existing.pane_id == pane.pane_id && existing.inner_rect == rect);
+        let scroll = SurfaceScroll { rect, shift: 1 };
+        if !committed || !scroll_fits(&scroll, frame.width, frame.height) {
+            continue;
+        }
+        if let Some((shift, rows)) = pane_scroll(frame, rect, &patch.rows) {
+            scrolls.push(SurfaceScroll { rect, shift });
+            residual.extend(rows);
+            if scrolls.len() == MAX_SCROLLS {
+                break;
+            }
+        }
+    }
+    if scrolls.is_empty() {
+        return None;
+    }
+
+    // Rows that straddle a scrolled region keep their final values, so they
+    // stay in the patch and apply after the scroll.
+    let rows = patch
+        .rows
+        .iter()
+        .filter(|row| !scrolls.iter().any(|scroll| fully_inside(row, scroll.rect)))
+        .cloned()
+        .chain(residual)
+        .collect();
+    let inner = ServerMessage::PaneSurfacePatch(PaneSurfacePatch {
+        boot_id: patch.boot_id.clone(),
+        projection_revision: patch.projection_revision,
+        base_surface_revision: patch.base_surface_revision,
+        surface_revision: patch.surface_revision,
+        rows,
+        panes: patch.panes.clone(),
+        cursor: patch.cursor.clone(),
+    });
+    let mut bytes = Vec::with_capacity(1 + scrolls.len() * SCROLL_BYTES);
+    bytes.push(scrolls.len() as u8);
+    for scroll in &scrolls {
+        let rect = scroll.rect;
+        for value in [rect.x, rect.y, rect.width, rect.height] {
+            bytes.extend_from_slice(&value.to_le_bytes());
+        }
+        bytes.extend_from_slice(&scroll.shift.to_le_bytes());
+    }
+    super::write_message(&mut bytes, &inner).ok()?;
+    let message = ServerMessage::EndpointControl {
+        kind: MESSAGE_KIND.into(),
+        data: STANDARD_NO_PAD.encode(bytes),
+    };
+    (encoded_size(&message)? <= MAX_FRAME_SIZE).then_some(message)
+}
+
+/// Decodes one scroll message. Geometry is validated against the receiver's
+/// baseline by [`apply`], never trusted here.
+pub(crate) fn decode(data: &str) -> Result<ScrollPatch, String> {
+    let limit = base64::encoded_len(MAX_FRAME_SIZE, false).unwrap_or(usize::MAX);
+    if data.len() > limit {
+        return Err("surface scroll exceeds the frame limit".into());
+    }
+    let bytes = STANDARD_NO_PAD
+        .decode(data)
+        .map_err(|error| format!("invalid surface scroll: {error}"))?;
+    let count = usize::from(*bytes.first().ok_or("empty surface scroll")?);
+    if count == 0 || count > MAX_SCROLLS {
+        return Err("surface scroll has an invalid scroll count".into());
+    }
+    let header = 1 + count * SCROLL_BYTES;
+    if bytes.len() < header {
+        return Err("surface scroll is truncated".into());
+    }
+    let scrolls = bytes[1..header]
+        .chunks_exact(SCROLL_BYTES)
+        .map(|chunk| {
+            let value = |at: usize| u16::from_le_bytes([chunk[at], chunk[at + 1]]);
+            SurfaceScroll {
+                rect: SurfaceRect {
+                    x: value(0),
+                    y: value(2),
+                    width: value(4),
+                    height: value(6),
+                },
+                shift: i16::from_le_bytes([chunk[8], chunk[9]]),
+            }
+        })
+        .collect();
+    let mut frame = &bytes[header..];
+    let decoded = super::read_message::<_, ServerMessage>(&mut frame, MAX_FRAME_SIZE);
+    if !frame.is_empty() {
+        return Err("surface scroll has trailing bytes".into());
+    }
+    match decoded {
+        Ok(ServerMessage::PaneSurfacePatch(patch)) => Ok(ScrollPatch { scrolls, patch }),
+        Ok(_) => Err("surface scroll does not carry a pane patch".into()),
+        Err(error) => Err(format!("invalid surface scroll patch: {error}")),
+    }
+}
+
+/// Applies a decoded scroll patch to a retained grid and returns the ordinary
+/// patch a receiver without scroll support would have needed. The grid is
+/// untouched when validation fails.
+pub(crate) fn apply(
+    cells: &mut [CellData],
+    width: u16,
+    height: u16,
+    scroll_patch: ScrollPatch,
+) -> Result<PaneSurfacePatch, String> {
+    let ScrollPatch { scrolls, mut patch } = scroll_patch;
+    if cells.len() != usize::from(width) * usize::from(height) {
+        return Err("surface scroll baseline has an invalid grid".into());
+    }
+    if !scrolls
+        .iter()
+        .all(|scroll| scroll_fits(scroll, width, height))
+        || !scrolls_disjoint(&scrolls)
+        || !patch.rows.iter().all(|row| row_fits(row, width, height))
+    {
+        return Err("surface scroll exceeds the cell baseline".into());
+    }
+    for scroll in &scrolls {
+        apply_scroll(cells, width, scroll);
+    }
+    for row in &patch.rows {
+        let start = usize::from(row.y) * usize::from(width) + usize::from(row.x);
+        cells[start..start + row.cells.len()].clone_from_slice(&row.cells);
+    }
+    // Each scrolled row is emitted once with its final cells, so the expanded patch keeps
+    // only the parts of other rows outside every region. Receivers require disjoint rows.
+    let mut rows = Vec::with_capacity(patch.rows.len());
+    for row in std::mem::take(&mut patch.rows) {
+        rows.extend(outside_scrolls(row, &scrolls));
+    }
+    for scroll in &scrolls {
+        let rect = scroll.rect;
+        for y in rect.y..rect.y + rect.height {
+            let start = usize::from(y) * usize::from(width) + usize::from(rect.x);
+            rows.push(PaneSurfacePatchRow {
+                x: rect.x,
+                y,
+                cells: cells[start..start + usize::from(rect.width)].to_vec(),
+            });
+        }
+    }
+    patch.rows = rows;
+    Ok(patch)
+}
+
+/// The spans of `row` not covered by any scrolled region.
+fn outside_scrolls(
+    row: PaneSurfacePatchRow,
+    scrolls: &[SurfaceScroll],
+) -> Vec<PaneSurfacePatchRow> {
+    let start = usize::from(row.x);
+    let mut covered = scrolls
+        .iter()
+        .map(|scroll| scroll.rect)
+        .filter(|rect| row.y >= rect.y && row.y - rect.y < rect.height)
+        .map(|rect| {
+            (
+                usize::from(rect.x),
+                usize::from(rect.x) + usize::from(rect.width),
+            )
+        })
+        .collect::<Vec<_>>();
+    if covered.is_empty() {
+        return vec![row];
+    }
+    covered.sort_unstable();
+    let end = start + row.cells.len();
+    let mut spans = Vec::new();
+    let mut cursor = start;
+    for (left, right) in covered.into_iter().chain([(end, end)]) {
+        let span_end = left.clamp(cursor, end);
+        if span_end > cursor {
+            spans.push(PaneSurfacePatchRow {
+                x: cursor as u16,
+                y: row.y,
+                cells: row.cells[cursor - start..span_end - start].to_vec(),
+            });
+        }
+        cursor = cursor.max(right.min(end));
+    }
+    spans
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::protocol::{FrameData, PaneSurfacePane};
+    use ratatui::{buffer::Buffer, layout::Rect};
+
+    const WIDTH: u16 = 30;
+    const HEIGHT: u16 = 12;
+    const PANE: SurfaceRect = SurfaceRect {
+        x: 2,
+        y: 1,
+        width: 20,
+        height: 10,
+    };
+
+    fn pane() -> PaneSurfacePane {
+        PaneSurfacePane {
+            pane_id: "w1:p1".into(),
+            content_revision: 1,
+            rect: SurfaceRect {
+                x: 1,
+                y: 0,
+                width: 22,
+                height: 12,
+            },
+            inner_rect: PANE,
+            scrollbar_rect: None,
+            scroll: None,
+            focused: true,
+            mouse_reporting: false,
+            sgr_pixel_mouse: false,
+            alternate_screen_active: false,
+            pixel_width: 0,
+            pixel_height: 0,
+        }
+    }
+
+    fn surface() -> PaneSurfaceFrame {
+        let mut frame =
+            FrameData::from_ratatui_buffer(&Buffer::empty(Rect::new(0, 0, WIDTH, HEIGHT)), None);
+        for y in 0..PANE.height {
+            write(&mut frame, y, &line(i32::from(y)));
+        }
+        PaneSurfaceFrame {
+            boot_id: "boot".into(),
+            projection_revision: 1,
+            surface_revision: 1,
+            frame,
+            panes: vec![pane()],
+            splits: Vec::new(),
+            popup: None,
+            graphics: Default::default(),
+        }
+    }
+
+    /// Distinct output lines, like a log or build scrolling past.
+    fn line(n: i32) -> String {
+        let word = ["alpha", "bravo", "charlie", "delta", "echo"][n.rem_euclid(5) as usize];
+        format!("{n:>3} {word} {}", word.to_uppercase())
+    }
+
+    fn write(frame: &mut FrameData, pane_row: u16, text: &str) {
+        let start = usize::from(PANE.y + pane_row) * usize::from(WIDTH) + usize::from(PANE.x);
+        for (x, cell) in frame.cells[start..start + usize::from(PANE.width)]
+            .iter_mut()
+            .enumerate()
+        {
+            cell.symbol = text.chars().nth(x).map_or(" ".into(), String::from);
+        }
+    }
+
+    /// The changed-cell spans the retained renderer would publish for `next`.
+    fn row_patch(last: &PaneSurfaceFrame, next: &FrameData) -> PaneSurfacePatch {
+        let mut rows = Vec::new();
+        for y in 0..HEIGHT {
+            let start = usize::from(y) * usize::from(WIDTH);
+            let (old, new) = (
+                &last.frame.cells[start..start + usize::from(WIDTH)],
+                &next.cells[start..start + usize::from(WIDTH)],
+            );
+            let mut x = 0;
+            while x < old.len() {
+                if old[x] == new[x] {
+                    x += 1;
+                    continue;
+                }
+                let first = x;
+                while x < old.len() && old[x] != new[x] {
+                    x += 1;
+                }
+                rows.push(PaneSurfacePatchRow {
+                    x: first as u16,
+                    y,
+                    cells: new[first..x].to_vec(),
+                });
+            }
+        }
+        PaneSurfacePatch {
+            boot_id: last.boot_id.clone(),
+            projection_revision: last.projection_revision,
+            base_surface_revision: last.surface_revision,
+            surface_revision: last.surface_revision + 1,
+            rows,
+            panes: vec![pane()],
+            cursor: None,
+        }
+    }
+
+    fn round_trip(last: &PaneSurfaceFrame, patch: &PaneSurfacePatch) -> (usize, FrameData) {
+        let message = message(last, patch).expect("scroll message");
+        let size = encoded_size(&message).unwrap();
+        let ServerMessage::EndpointControl { kind, data } = message else {
+            panic!("scroll must be an endpoint control frame");
+        };
+        assert_eq!(kind, MESSAGE_KIND);
+        let mut cells = last.frame.cells.clone();
+        let expanded = apply(&mut cells, WIDTH, HEIGHT, decode(&data).unwrap()).unwrap();
+        // A receiver without scroll support must reach the same grid.
+        let mut plain = last.frame.clone();
+        for row in &expanded.rows {
+            let start = usize::from(row.y) * usize::from(WIDTH) + usize::from(row.x);
+            plain.cells[start..start + row.cells.len()].clone_from_slice(&row.cells);
+        }
+        assert_eq!(plain.cells, cells);
+        assert_disjoint(&expanded.rows);
+        (size, plain)
+    }
+
+    /// Receivers only take the fast presentation path for non-overlapping rows.
+    fn assert_disjoint(rows: &[PaneSurfacePatchRow]) {
+        for (index, a) in rows.iter().enumerate() {
+            for b in &rows[index + 1..] {
+                let overlap = a.y == b.y
+                    && usize::from(a.x) < usize::from(b.x) + b.cells.len()
+                    && usize::from(b.x) < usize::from(a.x) + a.cells.len();
+                assert!(!overlap, "expanded rows overlap: {a:?} / {b:?}");
+            }
+        }
+    }
+
+    #[test]
+    fn rows_crossing_a_scrolled_region_keep_only_their_outside_cells() {
+        let last = surface();
+        let mut next = last.frame.clone();
+        for y in 0..PANE.height {
+            write(&mut next, y, &line(i32::from(y) + 1));
+        }
+        let mut patch = row_patch(&last, &next);
+        // One full-width row that runs across the pane, as a border repaint would.
+        let y = PANE.y + 4;
+        let start = usize::from(y) * usize::from(WIDTH);
+        patch.rows.push(PaneSurfacePatchRow {
+            x: 0,
+            y,
+            cells: next.cells[start..start + usize::from(WIDTH)].to_vec(),
+        });
+        let (_, result) = round_trip(&last, &patch);
+        assert_eq!(result.cells, next.cells);
+    }
+
+    #[test]
+    fn scrolling_output_sends_the_shift_and_only_new_rows() {
+        let last = surface();
+        let mut next = last.frame.clone();
+        for y in 0..PANE.height {
+            write(&mut next, y, &line(i32::from(y) + 1));
+        }
+        let patch = row_patch(&last, &next);
+
+        let (size, result) = round_trip(&last, &patch);
+        assert_eq!(result.cells, next.cells);
+        let plain = encoded_size(&patch).unwrap();
+        assert!(
+            size * 3 < plain,
+            "scroll must be much smaller: {size} vs {plain}"
+        );
+    }
+
+    #[test]
+    fn reverse_scroll_and_edits_reach_the_same_grid() {
+        let last = surface();
+        let mut next = last.frame.clone();
+        for y in 0..PANE.height {
+            write(&mut next, y, &line(i32::from(y) - 2));
+        }
+        write(&mut next, 5, "edited in place");
+        let (_, result) = round_trip(&last, &row_patch(&last, &next));
+        assert_eq!(result.cells, next.cells);
+    }
+
+    #[test]
+    fn a_single_row_change_stays_a_plain_patch() {
+        let last = surface();
+        let mut next = last.frame.clone();
+        write(&mut next, 9, "typed");
+        assert!(message(&last, &row_patch(&last, &next)).is_none());
+    }
+
+    #[test]
+    fn unrelated_rewrites_stay_a_plain_patch() {
+        let last = surface();
+        let mut next = last.frame.clone();
+        for y in 0..PANE.height {
+            write(&mut next, y, &format!("other {y}"));
+        }
+        assert!(message(&last, &row_patch(&last, &next)).is_none());
+    }
+
+    #[test]
+    fn decode_rejects_regions_outside_the_baseline() {
+        let last = surface();
+        let mut next = last.frame.clone();
+        for y in 0..PANE.height {
+            write(&mut next, y, &line(i32::from(y) + 1));
+        }
+        let Some(ServerMessage::EndpointControl { data, .. }) =
+            message(&last, &row_patch(&last, &next))
+        else {
+            panic!("scroll message");
+        };
+        let mut cells = last.frame.cells.clone();
+        let mut decoded = decode(&data).unwrap();
+        decoded.scrolls[0].rect.height = HEIGHT;
+        assert!(apply(&mut cells, WIDTH, HEIGHT, decoded).is_err());
+        assert_eq!(
+            cells, last.frame.cells,
+            "a rejected scroll must not touch the grid"
+        );
+
+        let mut decoded = decode(&data).unwrap();
+        decoded.scrolls.push(decoded.scrolls[0]);
+        assert!(
+            apply(&mut cells, WIDTH, HEIGHT, decoded).is_err(),
+            "overlapping regions"
+        );
+        assert_eq!(cells, last.frame.cells);
+
+        let mut decoded = decode(&data).unwrap();
+        decoded.scrolls[0].shift = PANE.height as i16;
+        assert!(apply(&mut cells, WIDTH, HEIGHT, decoded).is_err());
+        let mut trailing = STANDARD_NO_PAD.decode(&data).unwrap();
+        trailing.push(0);
+        assert!(decode(&STANDARD_NO_PAD.encode(trailing)).is_err());
+        assert!(decode("").is_err());
+        assert!(decode(&STANDARD_NO_PAD.encode([0u8])).is_err());
+        assert!(decode(&STANDARD_NO_PAD.encode([1u8, 0, 0])).is_err());
+    }
+}
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 465408f5..88daf250 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -418,6 +418,7 @@ pub(crate) enum ServerEvent {
         surface_active: bool,
         surface_reuse: bool,
         surface_delta: bool,
+        surface_scroll: bool,
         writer: ClientWriter,
     },
     /// A client sent an input message.
@@ -785,6 +786,7 @@ pub(crate) fn handle_client_handshake(
                     hello.surface_active,
                     hello.surface_reuse,
                     hello.surface_delta,
+                    hello.surface_scroll,
                 )),
             )
         }
@@ -881,6 +883,7 @@ pub(crate) fn handle_client_handshake(
         surface_active,
         surface_reuse,
         surface_delta,
+        surface_scroll,
     )) = shell_options
     {
         ServerEvent::ClientShellConnected {
@@ -896,6 +899,7 @@ pub(crate) fn handle_client_handshake(
             surface_active,
             surface_reuse,
             surface_delta,
+            surface_scroll,
             writer,
         }
     } else {
@@ -1472,6 +1476,7 @@ mod tests {
             surface_active: true,
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             snapshot_codecs: vec![crate::protocol::endpoint::SNAPSHOT_CODEC_V1.into()],
             surface_codecs: vec![crate::protocol::endpoint::SURFACE_CODEC_V1.into()],
             input_codecs: vec![crate::protocol::endpoint::INPUT_CODEC_V1.into()],
@@ -1989,10 +1994,12 @@ mod tests {
                 surface_active,
                 surface_reuse,
                 surface_delta,
+                surface_scroll,
                 writer,
             } => {
                 assert!(!surface_reuse);
                 assert!(!surface_delta);
+                assert!(!surface_scroll);
                 assert_eq!(client_id, 43);
                 assert_eq!((surface_cols, surface_rows), (80, 29));
                 assert_eq!((cell_width_px, cell_height_px), (8, 16));
diff --git a/src/server/headless.rs b/src/server/headless.rs
index da3baf30..18869e09 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -1861,6 +1861,7 @@ impl HeadlessServer {
                 surface_active,
                 surface_reuse,
                 surface_delta,
+                surface_scroll,
                 writer,
             } => {
                 if self.handoff_in_progress {
@@ -1908,6 +1909,9 @@ impl HeadlessServer {
                 connection.shell_surface_active = surface_active;
                 connection.render_state.enable_surface_reuse(surface_reuse);
                 connection.render_state.enable_surface_delta(surface_delta);
+                connection
+                    .render_state
+                    .enable_surface_scroll(surface_scroll);
                 connection.shell_projection_revision = 1;
                 let config_diagnostic = if endpoint_keybindings {
                     self.server_config_diagnostic.as_deref()
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 2dd082d6..27f3bfff 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -10,6 +10,8 @@ mod retained_graphics_tests;
 mod surface_delta_tests;
 #[path = "surface_interest.rs"]
 mod surface_interest_tests;
+#[path = "surface_scroll.rs"]
+mod surface_scroll_tests;
 
 fn client_shell_projection(
     receiver: &std::sync::mpsc::Receiver<Vec<u8>>,
@@ -719,6 +721,7 @@ async fn client_shell_attach_seeds_workspace() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 6,
             surface_cols: 80,
             surface_rows: 23,
@@ -761,6 +764,7 @@ async fn completion_guard_endpoint_pairs_runtime_completions_with_snapshots() {
         surface_active: false,
         surface_reuse: false,
         surface_delta: false,
+        surface_scroll: false,
         writer,
     });
     let (_, initial) = client_shell_projection(&control_rx);
@@ -809,6 +813,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols: 80,
             surface_rows: 23,
@@ -927,6 +932,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 77,
             surface_cols: 80,
             surface_rows: 23,
@@ -1029,6 +1035,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 7,
             surface_cols: 80,
             surface_rows: 23,
@@ -1196,6 +1203,7 @@ fn connect_test_shell(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols,
             surface_rows,
@@ -1800,6 +1808,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 13,
             surface_cols: 80,
             surface_rows: 23,
@@ -1824,6 +1833,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 14,
             surface_cols: 80,
             surface_rows: 23,
@@ -2743,6 +2753,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 9,
             surface_cols: 80,
             surface_rows: 23,
@@ -2989,6 +3000,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 12,
             surface_cols: 80,
             surface_rows: 23,
diff --git a/src/server/headless/tests/surface_delta.rs b/src/server/headless/tests/surface_delta.rs
index f010c1e3..996b5b1b 100644
--- a/src/server/headless/tests/surface_delta.rs
+++ b/src/server/headless/tests/surface_delta.rs
@@ -51,7 +51,7 @@ async fn surface_delta_reconstructs_metadata_text_and_hyperlinks() {
     enable_delta(&mut server, 1);
     server.render_and_stream();
     let (initial_bytes, initial_message) = receive_message(&render_rx);
-    let mut decoder = protocol::surface_reuse::Decoder::new(true);
+    let mut decoder = protocol::surface_reuse::Decoder::new(true, false);
     let initial = decode_surface_message(&mut decoder, initial_message);
     assert_eq!(
         without_asset_payload(initial.clone()),
@@ -119,7 +119,7 @@ async fn surface_delta_updates_global_popup_without_affecting_legacy_peer() {
     server.render_and_stream();
     let (initial_bytes, initial_message) = receive_message(&delta_render);
     let (_, _legacy_initial) = receive_message(&legacy_render);
-    let mut decoder = protocol::surface_reuse::Decoder::new(true);
+    let mut decoder = protocol::surface_reuse::Decoder::new(true, false);
     let _ = decode_surface_message(&mut decoder, initial_message);
 
     server
@@ -160,7 +160,7 @@ async fn surface_delta_preserves_graphics_baseline_through_queue_recovery() {
     enable_delta(&mut server, 1);
     server.render_and_stream();
     let (initial_bytes, initial_message) = receive_message(&render_rx);
-    let mut decoder = protocol::surface_reuse::Decoder::new(true);
+    let mut decoder = protocol::surface_reuse::Decoder::new(true, false);
     let _ = decode_surface_message(&mut decoder, initial_message);
 
     write_shared_test_pane(
diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs
index f3acc27b..a7da046e 100644
--- a/src/server/headless/tests/surface_interest.rs
+++ b/src/server/headless/tests/surface_interest.rs
@@ -66,6 +66,7 @@ async fn metadata_only_shell_is_isolated_until_surface_activation() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols: 101,
             surface_rows: 37,
@@ -281,6 +282,7 @@ async fn background_surface_activation_preserves_focused_viewer_geometry() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 8,
             surface_cols: 100,
             surface_rows: 35,
@@ -394,6 +396,7 @@ async fn presentation_sync_epoch_replays_modes_and_title() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -510,6 +513,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
         source_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: source_client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -533,6 +537,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
         target_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: target_client_id,
             surface_cols: 80,
             surface_rows: 24,
diff --git a/src/server/headless/tests/surface_scroll.rs b/src/server/headless/tests/surface_scroll.rs
new file mode 100644
index 00000000..8975650b
--- /dev/null
+++ b/src/server/headless/tests/surface_scroll.rs
@@ -0,0 +1,89 @@
+use super::*;
+
+fn scrolling_lines(range: std::ops::Range<usize>) -> Vec<u8> {
+    range
+        .map(|n| {
+            format!(
+                "build step {n:>3}: compiled module_{n} in {}ms\r\n",
+                n * 7 % 97
+            )
+        })
+        .collect::<String>()
+        .into_bytes()
+}
+
+fn apply_rows(frame: &mut FrameData, rows: &[crate::protocol::PaneSurfacePatchRow]) {
+    for row in rows {
+        let start = usize::from(row.y) * usize::from(frame.width) + usize::from(row.x);
+        frame.cells[start..start + row.cells.len()].clone_from_slice(&row.cells);
+    }
+}
+
+#[tokio::test]
+async fn surface_scroll_sends_scrolling_output_as_a_shift_and_new_rows() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(&scrolling_lines(0..40));
+    server
+        .clients
+        .get_mut(&1)
+        .expect("scroll client")
+        .render_state
+        .enable_surface_scroll(true);
+    server.render_and_stream();
+    let mut decoder = protocol::surface_reuse::Decoder::new(false, true);
+    let ServerMessage::PaneSurface(mut shell) = decoder
+        .decode(read_server_message(
+            render_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
+        ))
+        .expect("initial surface")
+    else {
+        panic!("expected the initial pane surface");
+    };
+
+    write_shared_test_pane(&mut server, pane_id, &scrolling_lines(40..42));
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let bytes = render_rx.recv_timeout(Duration::from_secs(1)).unwrap();
+    let message = read_server_message(bytes.clone());
+    assert!(matches!(
+        &message,
+        ServerMessage::EndpointControl { kind, .. } if kind == protocol::surface_scroll::MESSAGE_KIND
+    ));
+
+    // The decoder hands the client shell an ordinary patch that reaches the
+    // server's committed surface exactly.
+    let ServerMessage::PaneSurfacePatch(patch) = decoder.decode(message).expect("scroll decode")
+    else {
+        panic!("expected an expanded pane patch");
+    };
+    apply_rows(&mut shell.frame, &patch.rows);
+    let committed = server.clients[&1]
+        .render_state
+        .last_pane_surface()
+        .expect("committed surface");
+    assert_eq!(shell.frame.cells, committed.frame.cells);
+    assert!(frame_text(&shell.frame).contains("module_41"));
+
+    let expanded = HeadlessServer::frame_server_message(&ServerMessage::PaneSurfacePatch(patch))
+        .expect("expanded frame");
+    assert!(
+        bytes.len() * 4 < expanded.len(),
+        "scroll frame {} should be far smaller than the rows it replaces {}",
+        bytes.len(),
+        expanded.len()
+    );
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn surface_scroll_is_not_sent_to_a_peer_that_did_not_negotiate_it() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(&scrolling_lines(0..40));
+    server.render_and_stream();
+    let _ = render_rx.recv_timeout(Duration::from_secs(1)).unwrap();
+
+    write_shared_test_pane(&mut server, pane_id, &scrolling_lines(40..42));
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let message = read_server_message(render_rx.recv_timeout(Duration::from_secs(1)).unwrap());
+    assert!(matches!(message, ServerMessage::PaneSurfacePatch(_)));
+    shutdown_test_runtimes(&mut server);
+}
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index f4bb310b..754c9dbc 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -376,7 +376,8 @@ fn surface_encoding_client(mode: &str, surface: &PaneSurfaceFrame) -> SurfaceEnc
         state: super::render_stream::ClientRenderState::new(
             crate::protocol::RenderEncoding::SemanticFrame,
         ),
-        decoder: (reuse || delta).then(|| crate::protocol::surface_reuse::Decoder::new(delta)),
+        decoder: (reuse || delta)
+            .then(|| crate::protocol::surface_reuse::Decoder::new(delta, false)),
     };
     client.state.enable_surface_reuse(reuse);
     client.state.enable_surface_delta(delta);
@@ -588,3 +589,39 @@ async fn render_scale_profile() {
     print_surface_reuse_profiles();
     print_surface_damage_profiles();
 }
+
+/// Dirty-row collection while output scrolls a large pane, the host's hot path
+/// for streaming agents and build logs.
+#[tokio::test(flavor = "current_thread")]
+#[ignore = "manual dirty collection profile"]
+async fn dirty_scroll_collection_profile() {
+    const WIDTH: u16 = 295;
+    const HEIGHT: u16 = 84;
+    let runtime = TerminalRuntime::test_with_scrollback_bytes(WIDTH, HEIGHT, 1024 * 1024, b"");
+    let mut samples = Vec::new();
+    let mut rows = 0;
+    for line in 0..2_000u32 {
+        runtime.test_process_pty_bytes(
+            format!("{:010}{:010}\r\n", line.wrapping_mul(2_654_435_761), line).as_bytes(),
+        );
+        let started = Instant::now();
+        let snapshot = black_box(runtime.collect_dirty_patch_snapshot(WIDTH, HEIGHT));
+        if line as usize >= WARMUP_COUNT {
+            samples.push(started.elapsed());
+        }
+        if let Some(crate::pane::TerminalDirtyPatchOutcome::Patch(patch)) =
+            snapshot.map(|snapshot| snapshot.patch)
+        {
+            rows += patch.rows.len();
+        }
+    }
+    samples.sort_unstable();
+    let at = |fraction: f64| samples[((samples.len() - 1) as f64 * fraction) as usize].as_micros();
+    println!(
+        "dirty scroll collection {WIDTH}x{HEIGHT}: median {}us p95 {}us max {}us, {} rows per update",
+        at(0.5),
+        at(0.95),
+        at(1.0),
+        rows / 2_000
+    );
+}
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index ee874e52..aa1bbabe 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -19,6 +19,7 @@ pub(crate) enum ClientRenderState {
         surface_revision: u64,
         surface_reuse: bool,
         surface_delta: bool,
+        surface_scroll: bool,
         recompute_pending: bool,
     },
     /// Terminal-ANSI clients keep a terminal diff encoder and sequence number.
@@ -37,6 +38,7 @@ impl ClientRenderState {
                 surface_revision: 0,
                 surface_reuse: false,
                 surface_delta: false,
+                surface_scroll: false,
                 recompute_pending: false,
             },
             RenderEncoding::TerminalAnsi => Self::TerminalAnsi {
@@ -59,6 +61,12 @@ impl ClientRenderState {
         }
     }
 
+    pub(crate) fn enable_surface_scroll(&mut self, enabled: bool) {
+        if let Self::Semantic { surface_scroll, .. } = self {
+            *surface_scroll = enabled;
+        }
+    }
+
     pub(crate) fn request_recompute(&mut self) {
         if let Self::Semantic {
             surface_delta: true,
@@ -171,6 +179,7 @@ impl ClientRenderState {
             surface_reuse,
             surface_delta,
             recompute_pending,
+            ..
         } = self
         else {
             return None;
@@ -239,6 +248,7 @@ impl ClientRenderState {
         let Self::Semantic {
             last_surface,
             surface_revision,
+            surface_scroll,
             ..
         } = self
         else {
@@ -256,8 +266,18 @@ impl ClientRenderState {
         }
         let next_revision = surface_revision.saturating_add(1);
         patch.surface_revision = next_revision;
-        Some(PreparedRender::SemanticPatch {
-            message: ServerMessage::PaneSurfacePatch(patch),
+        let scrolled = (*surface_scroll)
+            .then(|| crate::protocol::surface_scroll::message(last, &patch))
+            .flatten();
+        Some(match scrolled {
+            Some(message) => PreparedRender::SemanticPatch {
+                message,
+                encoded: Some(Box::new(patch)),
+            },
+            None => PreparedRender::SemanticPatch {
+                message: ServerMessage::PaneSurfacePatch(patch),
+                encoded: None,
+            },
         })
     }
 
@@ -284,10 +304,13 @@ impl ClientRenderState {
                     surface_revision,
                     ..
                 },
-                PreparedRender::SemanticPatch {
-                    message: ServerMessage::PaneSurfacePatch(patch),
-                },
+                PreparedRender::SemanticPatch { message, encoded },
             ) => {
+                let patch = match (encoded, message) {
+                    (Some(patch), _) => *patch,
+                    (None, ServerMessage::PaneSurfacePatch(patch)) => patch,
+                    (None, _) => unreachable!("a plain semantic patch carries its pane patch"),
+                };
                 let surface = last_surface
                     .as_deref_mut()
                     .expect("prepared patch baseline");
@@ -358,6 +381,8 @@ pub(crate) enum PreparedRender {
     },
     SemanticPatch {
         message: ServerMessage,
+        /// The pane patch a compact `message` encodes; `None` when `message` is that patch.
+        encoded: Option<Box<PaneSurfacePatch>>,
     },
     TerminalAnsi {
         message: ServerMessage,
@@ -370,7 +395,7 @@ impl PreparedRender {
     pub(crate) fn message(&self) -> &ServerMessage {
         match self {
             Self::Semantic { message, .. }
-            | Self::SemanticPatch { message }
+            | Self::SemanticPatch { message, .. }
             | Self::TerminalAnsi { message, .. } => message,
         }
     }