fix: keep remote surface updates incremental (#4209)

refs #3769

refs #3745

19 files changed, 2031 insertions, 52 deletions

diff --git a/src/client/handshake.rs b/src/client/handshake.rs
index ef55fe22..69e2f52f 100644
--- a/src/client/handshake.rs
+++ b/src/client/handshake.rs
@@ -183,6 +183,7 @@ pub(super) fn do_handshake(
             mouse_capture,
             surface_active,
             surface_reuse: true,
+            surface_delta: 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 13125c2d..9fff9719 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -539,6 +539,10 @@ 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 transport = start_endpoint_transport(
             stream,
             (),
@@ -546,7 +550,7 @@ async fn run_client_loop(
             endpoint::ClientEndpointId::Local,
             1,
             max_frame_size,
-            negotiation.supports_capability(protocol::surface_reuse::CAPABILITY),
+            surface_decoder,
         )?;
         let mut registry = endpoint::EndpointRegistry::new(transport, 1, negotiation);
         if state.shell.is_some() {
@@ -1198,6 +1202,8 @@ async fn run_client_loop(
                     }
                     let surface_reuse =
                         negotiation.supports_capability(protocol::surface_reuse::CAPABILITY);
+                    let surface_delta =
+                        negotiation.supports_capability(protocol::surface_delta::CAPABILITY);
                     let agent_view_projection_supported = negotiation.supports_capability(
                         crate::protocol::endpoint::AGENT_VIEW_PROJECTION_CAPABILITY,
                     );
@@ -1220,6 +1226,8 @@ 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(
@@ -1229,7 +1237,7 @@ async fn run_client_loop(
                             MAX_GRAPHICS_FRAME_SIZE,
                             endpoint_id,
                             generation,
-                            surface_reuse,
+                            surface_decoder,
                         );
                     });
                 }
diff --git a/src/client/transport.rs b/src/client/transport.rs
index 4396690a..7f083bcf 100644
--- a/src/client/transport.rs
+++ b/src/client/transport.rs
@@ -7,7 +7,7 @@ pub(super) fn start_endpoint_transport(
     endpoint_id: endpoint::ClientEndpointId,
     generation: u64,
     max_frame_size: usize,
-    surface_reuse: bool,
+    surface_decoder: Option<protocol::surface_reuse::Decoder>,
 ) -> Result<endpoint::NativeEndpointTransport, ClientError> {
     let reader = stream.try_clone().map_err(ClientError::ConnectionFailed)?;
     let transport = endpoint::NativeEndpointTransport::with_lifetime(stream, lifetime)
@@ -24,7 +24,7 @@ pub(super) fn start_endpoint_transport(
                 max_frame_size,
                 endpoint_id,
                 generation,
-                surface_reuse,
+                surface_decoder,
             );
         })
         .map_err(ClientError::ConnectionFailed)?;
@@ -39,7 +39,7 @@ pub(super) fn server_reader_thread(
     max_frame_size: usize,
     endpoint_id: endpoint::ClientEndpointId,
     generation: u64,
-    surface_reuse: bool,
+    mut surface_decoder: Option<protocol::surface_reuse::Decoder>,
 ) {
     if stream.set_nonblocking(true).is_err() {
         let _ = event_tx.blocking_send(ClientLoopEvent::ServerDisconnected {
@@ -53,7 +53,6 @@ pub(super) fn server_reader_thread(
         stream: &mut stream,
         stopped: should_quit,
     };
-    let mut surface_decoder = surface_reuse.then(protocol::surface_reuse::Decoder::default);
     loop {
         if should_quit.load(Ordering::Acquire) {
             break;
diff --git a/src/protocol/endpoint.rs b/src/protocol/endpoint.rs
index 2f7c8338..df83b242 100644
--- a/src/protocol/endpoint.rs
+++ b/src/protocol/endpoint.rs
@@ -49,6 +49,9 @@ pub struct EndpointClientHello {
     /// Accept the optional cell-retaining surface encoding on this connection.
     #[serde(default)]
     pub surface_reuse: bool,
+    /// Accept the optional surface-delta encoding on this connection.
+    #[serde(default)]
+    pub surface_delta: bool,
     #[serde(default)]
     pub snapshot_codecs: Vec<String>,
     #[serde(default)]
@@ -143,6 +146,7 @@ impl EndpointServerWelcome {
             methods,
             capabilities: vec![
                 super::surface_reuse::CAPABILITY.into(),
+                super::surface_delta::CAPABILITY.into(),
                 SURFACE_INTEREST_CAPABILITY.into(),
                 PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(),
                 HEALTH_CHECK_CAPABILITY.into(),
@@ -186,6 +190,7 @@ mod tests {
             mouse_capture: true,
             surface_active: true,
             surface_reuse: false,
+            surface_delta: false,
             snapshot_codecs: vec![SNAPSHOT_CODEC_V1.into()],
             surface_codecs: vec![SURFACE_CODEC_V1.into()],
             input_codecs: vec![INPUT_CODEC_V1.into()],
@@ -333,9 +338,11 @@ mod tests {
         let mut value = serde_json::to_value(hello()).unwrap();
         value.as_object_mut().unwrap().remove("surface_active");
         value.as_object_mut().unwrap().remove("surface_reuse");
+        value.as_object_mut().unwrap().remove("surface_delta");
         let decoded: EndpointClientHello = serde_json::from_value(value).unwrap();
         assert!(decoded.surface_active);
         assert!(!decoded.surface_reuse);
+        assert!(!decoded.surface_delta);
     }
 
     #[test]
@@ -345,6 +352,7 @@ mod tests {
             welcome.capabilities,
             vec![
                 super::super::surface_reuse::CAPABILITY.to_string(),
+                super::super::surface_delta::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 4a8766b6..24e2ebd2 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -2,6 +2,7 @@
 
 pub mod endpoint;
 pub(crate) mod render_ansi;
+pub(crate) mod surface_delta;
 pub(crate) mod surface_reuse;
 mod wire;
 
diff --git a/src/protocol/surface_delta.rs b/src/protocol/surface_delta.rs
new file mode 100644
index 00000000..8f94633b
--- /dev/null
+++ b/src/protocol/surface_delta.rs
@@ -0,0 +1,546 @@
+//! Optional sparse delivery of a completely recomputed pane surface.
+
+use base64::{engine::general_purpose::STANDARD_NO_PAD, Engine as _};
+use serde::{Deserialize, Serialize};
+
+use super::{CellData, PaneSurfaceFrame, PaneSurfacePatchRow, ServerMessage};
+
+mod decode;
+
+pub(crate) const CAPABILITY: &str = "surface_delta";
+pub(crate) const MESSAGE_KIND: &str = "endpoint.surface-delta.v1";
+pub(crate) const MAX_SPANS: usize = 4096;
+
+// This binary layout is frozen with MESSAGE_KIND, independently of the v1 core.
+#[derive(Serialize, Deserialize)]
+pub(crate) struct SurfaceDelta<S, R = Vec<PaneSurfacePatchRow>, P = GridUpdate> {
+    pub(crate) base_projection_revision: u64,
+    pub(crate) base_surface_revision: u64,
+    pub(crate) surface: S,
+    pub(crate) rows: R,
+    pub(crate) popup_cells: Option<P>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub(crate) enum GridUpdate {
+    Patch(Vec<PaneSurfacePatchRow>),
+    Replace(Vec<CellData>),
+}
+
+#[cfg(test)]
+pub(crate) fn decode(data: &str) -> Result<SurfaceDelta<PaneSurfaceFrame>, String> {
+    decode::decode(data, None)
+}
+
+pub(crate) fn decode_for(
+    data: &str,
+    expected: (u16, u16),
+) -> Result<SurfaceDelta<PaneSurfaceFrame>, String> {
+    decode::decode(data, Some(expected))
+}
+
+pub(crate) fn apply_rows(cells: &mut [CellData], width: u16, rows: &[PaneSurfacePatchRow]) {
+    for row in 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);
+    }
+}
+
+#[derive(Serialize)]
+struct CellSpan<'a> {
+    x: u16,
+    y: u16,
+    cells: &'a [CellData],
+}
+
+#[derive(Serialize)]
+enum GridUpdateRef<'a> {
+    Patch(Vec<CellSpan<'a>>),
+    Replace(&'a [CellData]),
+}
+
+fn changed_rows<'a>(
+    last: &[CellData],
+    next: &'a [CellData],
+    width: u16,
+    full_size: usize,
+) -> Result<Option<Vec<CellSpan<'a>>>, String> {
+    let mut rows = Vec::new();
+    if width == 0 {
+        return Ok(Some(rows));
+    }
+    let mut size = 0;
+    for (y, (old_row, new_row)) in last
+        .chunks(usize::from(width))
+        .zip(next.chunks(usize::from(width)))
+        .enumerate()
+    {
+        let mut x = 0;
+        while x < new_row.len() {
+            if old_row[x] == new_row[x] {
+                x += 1;
+                continue;
+            }
+            let start = x;
+            while x < new_row.len() && old_row[x] != new_row[x] {
+                x += 1;
+            }
+            let span = CellSpan {
+                x: start as u16,
+                y: y as u16,
+                cells: &new_row[start..x],
+            };
+            size += encoded_size(&span)?;
+            // This lower bound excludes metadata, so aborting cannot discard a smaller delta.
+            if rows.len() == MAX_SPANS
+                || base64::encoded_len(size, false).is_none_or(|size| size >= full_size)
+            {
+                return Ok(None);
+            }
+            rows.push(span);
+        }
+    }
+    Ok(Some(rows))
+}
+
+fn encoded_size(value: &impl Serialize) -> Result<usize, String> {
+    let mut writer = bincode::enc::write::SizeWriter::default();
+    bincode::serde::encode_into_writer(value, &mut writer, bincode::config::standard())
+        .map_err(|error| error.to_string())?;
+    Ok(writer.bytes_written)
+}
+
+pub(crate) fn message(
+    last: &PaneSurfaceFrame,
+    full: &mut ServerMessage,
+) -> Result<Option<ServerMessage>, String> {
+    let ServerMessage::PaneSurface(surface) = &*full else {
+        return Ok(None);
+    };
+    let expected_cells = usize::from(surface.frame.width) * usize::from(surface.frame.height);
+    if last.boot_id != surface.boot_id
+        || last.frame.width != surface.frame.width
+        || last.frame.height != surface.frame.height
+        || last.frame.cells.len() != expected_cells
+        || !decode::metadata_fits(surface)
+    {
+        return Ok(None);
+    }
+    let full_size = encoded_size(full)?;
+    let ServerMessage::PaneSurface(surface) = full else {
+        return Ok(None);
+    };
+    let max = if surface.graphics.assets.is_empty()
+        && surface.graphics.placements.is_empty()
+        && surface.graphics.retained_assets.is_empty()
+    {
+        super::MAX_FRAME_SIZE
+    } else {
+        super::MAX_GRAPHICS_FRAME_SIZE
+    };
+    let cells = std::mem::take(&mut surface.frame.cells);
+    let popup_grid = surface
+        .popup
+        .as_mut()
+        .map(|popup| std::mem::take(&mut popup.frame.cells));
+    let encoded = (|| {
+        let Some(rows) = changed_rows(&last.frame.cells, &cells, surface.frame.width, full_size)?
+        else {
+            return Ok(None);
+        };
+        let popup_cells = if let (Some(popup), Some(grid)) = (&surface.popup, &popup_grid) {
+            if let Some(previous) = last.popup.as_ref().filter(|previous| {
+                previous.terminal_id == popup.terminal_id
+                    && previous.frame.width == popup.frame.width
+                    && previous.frame.height == popup.frame.height
+                    && previous.frame.cells.len() == grid.len()
+            }) {
+                let Some(rows) =
+                    changed_rows(&previous.frame.cells, grid, popup.frame.width, full_size)?
+                else {
+                    return Ok(None);
+                };
+                Some(GridUpdateRef::Patch(rows))
+            } else {
+                Some(GridUpdateRef::Replace(grid))
+            }
+        } else {
+            None
+        };
+        let delta = SurfaceDelta {
+            base_projection_revision: last.projection_revision,
+            base_surface_revision: last.surface_revision,
+            surface: &*surface,
+            rows,
+            popup_cells,
+        };
+        let size = encoded_size(&delta)?;
+        let encoded_len = base64::encoded_len(size, false).ok_or("surface delta size overflow")?;
+        if encoded_len > max || encoded_len >= full_size {
+            return Ok(None);
+        }
+        bincode::serde::encode_to_vec(&delta, bincode::config::standard())
+            .map(Some)
+            .map_err(|error| error.to_string())
+    })();
+    surface.frame.cells = cells;
+    if let (Some(popup), Some(cells)) = (surface.popup.as_mut(), popup_grid) {
+        popup.frame.cells = cells;
+    }
+    let Some(bytes) = encoded? else {
+        return Ok(None);
+    };
+    let message = ServerMessage::EndpointControl {
+        kind: MESSAGE_KIND.into(),
+        data: STANDARD_NO_PAD.encode(bytes),
+    };
+    let size = encoded_size(&message)?;
+    Ok((size <= max && size < full_size).then_some(message))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::protocol::{ClientShellPopupSurface, FrameData};
+    use ratatui::{buffer::Buffer, layout::Rect};
+
+    fn surface() -> PaneSurfaceFrame {
+        PaneSurfaceFrame {
+            boot_id: "boot".into(),
+            projection_revision: 1,
+            surface_revision: 1,
+            frame: FrameData::from_ratatui_buffer(&Buffer::empty(Rect::new(0, 0, 120, 40)), None),
+            panes: Vec::new(),
+            splits: Vec::new(),
+            popup: None,
+            graphics: Default::default(),
+        }
+    }
+
+    fn framed_len(message: &ServerMessage) -> usize {
+        let mut bytes = Vec::new();
+        crate::protocol::write_message(&mut bytes, message).unwrap();
+        bytes.len()
+    }
+
+    fn reconstruct(last: &PaneSurfaceFrame, next: &PaneSurfaceFrame) -> ServerMessage {
+        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+        decoder
+            .decode(ServerMessage::PaneSurface(last.clone()))
+            .unwrap();
+        let update = message(last, &mut ServerMessage::PaneSurface(next.clone()))
+            .unwrap()
+            .expect("smaller delta");
+        let mut bytes = Vec::new();
+        crate::protocol::write_message(&mut bytes, &update).unwrap();
+        let wire = crate::protocol::read_message(
+            &mut bytes.as_slice(),
+            crate::protocol::MAX_GRAPHICS_FRAME_SIZE,
+        )
+        .unwrap();
+        let ServerMessage::PaneSurface(decoded) = decoder.decode(wire).unwrap() else {
+            panic!("full surface");
+        };
+        assert_eq!(&decoded, next);
+        update
+    }
+
+    #[test]
+    fn surface_delta_keeps_a_cell_plus_projection_update_small() {
+        let last = surface();
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        next.projection_revision += 1;
+        next.frame.cells[0].symbol = "x".into();
+        reconstruct(&last, &next);
+        let expected = next.clone();
+        let mut full = ServerMessage::PaneSurface(next);
+        let update = message(&last, &mut full).unwrap().expect("sparse delta");
+        assert!(framed_len(&update) < 1000);
+        let ServerMessage::PaneSurface(next) = full else {
+            panic!("full fallback");
+        };
+        assert_eq!(next, expected, "encoding must not consume the target grid");
+    }
+
+    #[test]
+    fn surface_delta_keeps_popup_typing_small() {
+        let mut last = surface();
+        last.popup = Some(Box::new(ClientShellPopupSurface {
+            terminal_id: "popup".into(),
+            title: "popup".into(),
+            width: None,
+            height: None,
+            frame: FrameData::from_ratatui_buffer(&Buffer::empty(Rect::new(0, 0, 100, 30)), None),
+            mouse_reporting: false,
+            sgr_pixel_mouse: false,
+            pixel_width: 0,
+            pixel_height: 0,
+        }));
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        next.popup.as_mut().unwrap().frame.cells[0].symbol = "x".into();
+        let update = reconstruct(&last, &next);
+        assert!(framed_len(&update) < 1000);
+        let mut replaced = next.clone();
+        replaced.surface_revision += 1;
+        replaced.popup.as_mut().unwrap().terminal_id = "replacement".into();
+        reconstruct(&next, &replaced);
+        let mut closed = replaced.clone();
+        closed.surface_revision += 1;
+        closed.popup = None;
+        reconstruct(&replaced, &closed);
+        let mut opened = closed.clone();
+        opened.surface_revision += 1;
+        opened.popup = last.popup;
+        reconstruct(&closed, &opened);
+    }
+
+    #[test]
+    fn surface_delta_reconstructs_cursor_hyperlinks_and_new_assets() {
+        use crate::protocol::{
+            CursorState, SurfaceGraphicsAsset, SurfaceGraphicsAssetKey, SurfaceGraphicsFormat,
+            SurfaceGraphicsSource,
+        };
+        let last = surface();
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        next.projection_revision += 1;
+        next.frame.cursor = Some(CursorState {
+            x: 3,
+            y: 2,
+            visible: true,
+            shape: 1,
+        });
+        next.frame.hyperlinks = vec!["https://example.com/\"quoted\"".into()];
+        next.frame.cells[0].hyperlink = Some(0);
+        next.graphics.assets.push(SurfaceGraphicsAsset {
+            key: SurfaceGraphicsAssetKey {
+                source: SurfaceGraphicsSource::PaneLayer {
+                    pane_id: "w1:p1".into(),
+                    layer_id: "image".into(),
+                },
+                image_width: 1,
+                image_height: 1,
+                format: SurfaceGraphicsFormat::Rgba,
+                data_len: 4,
+                data_fingerprint: 1,
+            },
+            data: vec![0, 128, 255, 255],
+        });
+        assert!(framed_len(&reconstruct(&last, &next)) < 1000);
+        let mut changed = next.clone();
+        changed.surface_revision += 1;
+        changed.graphics.assets[0].data[0] = 255;
+        changed.graphics.assets[0].key.data_fingerprint = 2;
+        changed.frame.hyperlinks[0] = "https://other.example".into();
+        reconstruct(&next, &changed);
+    }
+
+    #[test]
+    fn surface_delta_uses_full_for_resize_and_when_full_is_smaller() {
+        let last = surface();
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        for cell in &mut next.frame.cells {
+            cell.symbol = "x".into();
+        }
+        assert!(
+            message(&last, &mut ServerMessage::PaneSurface(next.clone()))
+                .unwrap()
+                .is_none()
+        );
+        next.frame.width += 1;
+        assert!(message(&last, &mut ServerMessage::PaneSurface(next))
+            .unwrap()
+            .is_none());
+    }
+
+    #[test]
+    fn surface_delta_rejects_wrong_baselines_and_invalid_metadata_without_advancing() {
+        let last = surface();
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        next.frame.cells[0].symbol = "x".into();
+        let update = reconstruct(&last, &next);
+        let ServerMessage::EndpointControl { data, .. } = &update else {
+            panic!("delta");
+        };
+        for case in 0..9 {
+            let mut corrupt = decode(data).unwrap();
+            match case {
+                0 => corrupt.base_surface_revision += 1,
+                1 => corrupt.base_projection_revision += 1,
+                2 => corrupt.surface.surface_revision += 1,
+                3 => corrupt.surface.boot_id = "different boot".into(),
+                4 => corrupt.surface.frame.width += 1,
+                5 => corrupt
+                    .surface
+                    .frame
+                    .cells
+                    .push(next.frame.cells[0].clone()),
+                6 => corrupt.rows[0].cells[0].hyperlink = Some(0),
+                7 => corrupt.rows.push(corrupt.rows[0].clone()),
+                8 => corrupt.surface.projection_revision = 0,
+                _ => unreachable!(),
+            }
+            let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+            decoder
+                .decode(ServerMessage::PaneSurface(last.clone()))
+                .unwrap();
+            let bad = ServerMessage::EndpointControl {
+                kind: MESSAGE_KIND.into(),
+                data: STANDARD_NO_PAD.encode(
+                    bincode::serde::encode_to_vec(&corrupt, bincode::config::standard()).unwrap(),
+                ),
+            };
+            assert!(decoder.decode(bad).is_err(), "case {case}");
+            let ServerMessage::PaneSurface(decoded) = decoder.decode(update.clone()).unwrap()
+            else {
+                panic!("full reconstruction");
+            };
+            assert_eq!(decoded, next);
+        }
+    }
+
+    #[test]
+    fn surface_delta_shares_baselines_with_legacy_patches_and_reuse() {
+        let last = surface();
+        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);
+        decoder.decode(ServerMessage::PaneSurface(last)).unwrap();
+        decoder
+            .decode(ServerMessage::PaneSurfacePatch(
+                crate::protocol::PaneSurfacePatch {
+                    boot_id: next.boot_id.clone(),
+                    projection_revision: next.projection_revision,
+                    base_surface_revision: 1,
+                    surface_revision: 2,
+                    rows: vec![PaneSurfacePatchRow {
+                        x: 0,
+                        y: 0,
+                        cells: vec![next.frame.cells[0].clone()],
+                    }],
+                    panes: Vec::new(),
+                    cursor: None,
+                },
+            ))
+            .unwrap();
+        let base_revision = next.surface_revision;
+        next.surface_revision += 1;
+        next.projection_revision += 1;
+        let reused = crate::protocol::surface_reuse::message(base_revision, &mut next)
+            .unwrap()
+            .unwrap();
+        decoder.decode(reused).unwrap();
+        let mut latest = next.clone();
+        latest.surface_revision += 1;
+        latest.projection_revision += 1;
+        latest.frame.cells[1].symbol = "b".into();
+        let update = message(&next, &mut ServerMessage::PaneSurface(latest.clone()))
+            .unwrap()
+            .unwrap();
+        let ServerMessage::PaneSurface(decoded) = decoder.decode(update).unwrap() else {
+            panic!("full reconstruction");
+        };
+        assert_eq!(decoded, latest);
+    }
+
+    #[test]
+    fn surface_delta_checkerboard_borrows_cells_and_bounds_run_planning() {
+        let last = surface();
+        let mut next = last.clone();
+        for (index, cell) in next.frame.cells.iter_mut().enumerate() {
+            if index % 2 == 0 {
+                cell.symbol = "x".into();
+            }
+        }
+        let spans = changed_rows(&last.frame.cells, &next.frame.cells, 120, usize::MAX)
+            .unwrap()
+            .unwrap();
+        assert_eq!(spans.len(), next.frame.cells.len() / 2);
+        for span in spans {
+            let start = usize::from(span.y) * 120 + usize::from(span.x);
+            assert_eq!(span.cells.as_ptr(), next.frame.cells[start..].as_ptr());
+        }
+        assert!(changed_rows(&last.frame.cells, &next.frame.cells, 120, 1)
+            .unwrap()
+            .is_none());
+        let mut large =
+            FrameData::from_ratatui_buffer(&Buffer::empty(Rect::new(0, 0, 120, 100)), None);
+        let baseline = large.cells.clone();
+        for (index, cell) in large.cells.iter_mut().enumerate() {
+            if index % 2 == 0 {
+                cell.symbol = "x".into();
+            }
+        }
+        assert!(changed_rows(&baseline, &large.cells, 120, usize::MAX)
+            .unwrap()
+            .is_none());
+    }
+
+    #[test]
+    fn surface_delta_v1_binary_fixture_is_frozen() {
+        use sha2::{Digest, Sha256};
+        let last = surface();
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        next.projection_revision += 1;
+        next.frame.cells[0].symbol = "x".into();
+        let update = reconstruct(&last, &next);
+        let mut bytes = Vec::new();
+        crate::protocol::write_message(&mut bytes, &update).unwrap();
+        assert_eq!(
+            format!("{:x}", Sha256::digest(bytes)),
+            "1effe2cbf998ff334bda9151995b87d54e646a1b90c10d853723d5bc1c8a84bf"
+        );
+    }
+
+    #[test]
+    fn surface_delta_rejects_invalid_spans_atomically_and_requires_negotiation() {
+        let last = surface();
+        let mut next = last.clone();
+        next.surface_revision += 1;
+        next.frame.cells[0].symbol = "x".into();
+        let update = message(&last, &mut ServerMessage::PaneSurface(next.clone()))
+            .unwrap()
+            .unwrap();
+        let mut legacy = crate::protocol::surface_reuse::Decoder::default();
+        legacy
+            .decode(ServerMessage::PaneSurface(last.clone()))
+            .unwrap();
+        assert!(legacy.decode(update.clone()).is_err());
+        let mut decoder = crate::protocol::surface_reuse::Decoder::new(true);
+        decoder.decode(ServerMessage::PaneSurface(last)).unwrap();
+        let ServerMessage::EndpointControl { data, .. } = &update else {
+            panic!("delta");
+        };
+        let mut corrupt = decode(data).unwrap();
+        corrupt.rows.push(PaneSurfacePatchRow {
+            x: 0,
+            y: 40,
+            cells: vec![next.frame.cells[0].clone()],
+        });
+        let bad = ServerMessage::EndpointControl {
+            kind: MESSAGE_KIND.into(),
+            data: STANDARD_NO_PAD.encode(
+                bincode::serde::encode_to_vec(&corrupt, bincode::config::standard()).unwrap(),
+            ),
+        };
+        assert!(decoder.decode(bad).is_err());
+        let ServerMessage::PaneSurface(decoded) = decoder.decode(update.clone()).unwrap() else {
+            panic!("recovered");
+        };
+        assert_eq!(decoded, next);
+        assert!(
+            decoder.decode(update.clone()).is_err(),
+            "duplicate revision"
+        );
+        assert!(decode("not-base64!").is_err());
+        let mut trailing = STANDARD_NO_PAD.decode(data).unwrap();
+        trailing.push(0);
+        assert!(decode(&STANDARD_NO_PAD.encode(trailing)).is_err());
+    }
+}
diff --git a/src/protocol/surface_delta/decode.rs b/src/protocol/surface_delta/decode.rs
new file mode 100644
index 00000000..82fd52f9
--- /dev/null
+++ b/src/protocol/surface_delta/decode.rs
@@ -0,0 +1,806 @@
+//! Allocation-bounded decoder for the surface-delta codec.
+//!
+//! Bincode's byte limit is necessary for strings and pixel payloads, but it is
+//! not a logical item limit: nested `Vec`s can otherwise reserve far more
+//! memory than their encoded representation occupies. Every collection owned
+//! by this codec is therefore decoded only after its count and grid budget have
+//! been checked.
+
+use base64::{engine::general_purpose::STANDARD_NO_PAD, Engine as _};
+use bincode::{
+    de::{read::Reader as _, Decode, Decoder},
+    error::DecodeError,
+};
+
+use super::{GridUpdate, SurfaceDelta};
+use crate::protocol::{
+    CellData, ClientShellPopupSurface, CursorState, FrameData, PaneSurfaceFrame, PaneSurfacePane,
+    PaneSurfacePatchRow, PaneSurfaceSplit, PaneSurfaceSplitDirection, SurfaceGraphicsAsset,
+    SurfaceGraphicsAssetKey, SurfaceGraphicsPlacement, SurfaceGraphicsScene, SurfaceRect,
+    MAX_FRAME_SIZE, MAX_GRAPHICS_FRAME_SIZE,
+};
+
+// These are intentionally practical protocol limits rather than bincode byte
+// limits. A sender that exceeds one uses the unchanged full-frame codec.
+const MAX_GRID_DIMENSION: u16 = 4096;
+const MAX_GRID_CELLS: usize = 1_000_000;
+const MAX_PANES: usize = 4096;
+const MAX_SPLITS: usize = 4096;
+const MAX_SPLIT_PATH: usize = 4096;
+const MAX_HYPERLINKS: usize = 65_536;
+const MAX_ASSETS: usize = 4096;
+const MAX_PLACEMENTS: usize = 65_536;
+const MAX_RETAINED_ASSETS: usize = 65_536;
+
+#[derive(Clone, Copy)]
+struct DecodeContext {
+    expected: Option<(u16, u16)>,
+    data_len: usize,
+}
+
+struct Decoded(SurfaceDelta<PaneSurfaceFrame>);
+
+pub(super) fn decode(
+    data: &str,
+    expected: Option<(u16, u16)>,
+) -> Result<SurfaceDelta<PaneSurfaceFrame>, String> {
+    // EndpointControl's data string itself is subject to the outer graphics
+    // frame cap. Checking before base64 decoding also bounds that allocation.
+    if data.len() > MAX_GRAPHICS_FRAME_SIZE {
+        return Err("surface delta exceeds the frame limit".into());
+    }
+    let bytes = STANDARD_NO_PAD
+        .decode(data)
+        .map_err(|error| error.to_string())?;
+    if bytes.len() > MAX_GRAPHICS_FRAME_SIZE {
+        return Err("decoded surface delta exceeds the frame limit".into());
+    }
+    decode_bytes_with_data_len(&bytes, expected, data.len())
+}
+
+#[cfg(test)]
+fn decode_bytes(
+    bytes: &[u8],
+    expected: Option<(u16, u16)>,
+) -> Result<SurfaceDelta<PaneSurfaceFrame>, String> {
+    let data_len = base64::encoded_len(bytes.len(), false).unwrap_or(usize::MAX);
+    decode_bytes_with_data_len(bytes, expected, data_len)
+}
+
+fn decode_bytes_with_data_len(
+    bytes: &[u8],
+    expected: Option<(u16, u16)>,
+    data_len: usize,
+) -> Result<SurfaceDelta<PaneSurfaceFrame>, String> {
+    let context = DecodeContext { expected, data_len };
+    let (Decoded(delta), consumed) = bincode::decode_from_slice_with_context(
+        bytes,
+        bincode::config::standard().with_limit::<MAX_GRAPHICS_FRAME_SIZE>(),
+        context,
+    )
+    .map_err(|error| error.to_string())?;
+    if consumed != bytes.len() {
+        return Err("trailing surface delta bytes".into());
+    }
+    Ok(delta)
+}
+
+impl Decode<DecodeContext> for Decoded {
+    fn decode<D: Decoder<Context = DecodeContext>>(decoder: &mut D) -> Result<Self, DecodeError> {
+        let base_projection_revision = u64::decode(decoder)?;
+        let base_surface_revision = u64::decode(decoder)?;
+        let (surface, has_graphics) = decode_surface(decoder)?;
+
+        // The outer transport permits the graphics cap so that it can carry
+        // EndpointControl. Once metadata tells us this is an ordinary surface,
+        // reject it at the ordinary frame cap before decoding either cell grid.
+        if !has_graphics && decoder.context().data_len > MAX_FRAME_SIZE {
+            return Err(DecodeError::Other(
+                "ordinary surface delta exceeds the frame limit",
+            ));
+        }
+
+        let metadata_dimensions = (surface.frame.width, surface.frame.height);
+        let (width, height) = match decoder.context().expected {
+            Some(expected) if expected != metadata_dimensions => {
+                return Err(DecodeError::Other(
+                    "surface dimensions do not match the baseline",
+                ));
+            }
+            Some(expected) => expected,
+            None => metadata_dimensions,
+        };
+        checked_grid_size(width, height)?;
+        let rows = decode_rows(decoder, width, height)?;
+        let popup_cells = decode_popup_update(decoder, surface.popup.as_deref())?;
+
+        Ok(Self(SurfaceDelta {
+            base_projection_revision,
+            base_surface_revision,
+            surface,
+            rows,
+            popup_cells,
+        }))
+    }
+}
+
+fn decode_surface<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+) -> Result<(PaneSurfaceFrame, bool), DecodeError> {
+    let boot_id = String::decode(decoder)?;
+    let projection_revision = u64::decode(decoder)?;
+    let surface_revision = u64::decode(decoder)?;
+    let frame = decode_frame(decoder)?;
+    let panes = decode_bounded_vec::<D, PaneSurfacePane>(decoder, MAX_PANES, "too many panes")?;
+    let splits = decode_splits(decoder)?;
+    let popup = decode_popup(decoder)?;
+    let graphics = decode_graphics_scene(decoder)?;
+    let has_graphics = !graphics.assets.is_empty()
+        || !graphics.placements.is_empty()
+        || !graphics.retained_assets.is_empty();
+    Ok((
+        PaneSurfaceFrame {
+            boot_id,
+            projection_revision,
+            surface_revision,
+            frame,
+            panes,
+            splits,
+            popup,
+            graphics,
+        },
+        has_graphics,
+    ))
+}
+
+fn decode_frame<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+) -> Result<FrameData, DecodeError> {
+    // Delta metadata is encoded after the sender has taken both full cell
+    // grids. Reading only the count prevents a forged Vec prefix from causing
+    // any cell allocation.
+    require_empty_sequence(decoder, "surface metadata contains main cells")?;
+    let width = u16::decode(decoder)?;
+    let height = u16::decode(decoder)?;
+    checked_grid_size(width, height)?;
+    let cursor = Option::<CursorState>::decode(decoder)?;
+    let hyperlinks =
+        decode_bounded_vec::<D, String>(decoder, MAX_HYPERLINKS, "too many hyperlinks")?;
+    let graphics = decode_bytes_vec(
+        decoder,
+        MAX_GRAPHICS_FRAME_SIZE,
+        "graphics payload too large",
+    )?;
+    Ok(FrameData {
+        cells: Vec::new(),
+        width,
+        height,
+        cursor,
+        hyperlinks,
+        graphics,
+    })
+}
+
+fn decode_popup<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+) -> Result<Option<Box<ClientShellPopupSurface>>, DecodeError> {
+    match u8::decode(decoder)? {
+        0 => Ok(None),
+        1 => {
+            let terminal_id = String::decode(decoder)?;
+            let title = String::decode(decoder)?;
+            let width = Option::decode(decoder)?;
+            let height = Option::decode(decoder)?;
+            let frame = decode_frame(decoder)?;
+            let mouse_reporting = bool::decode(decoder)?;
+            let sgr_pixel_mouse = bool::decode(decoder)?;
+            let pixel_width = u32::decode(decoder)?;
+            let pixel_height = u32::decode(decoder)?;
+            Ok(Some(Box::new(ClientShellPopupSurface {
+                terminal_id,
+                title,
+                width,
+                height,
+                frame,
+                mouse_reporting,
+                sgr_pixel_mouse,
+                pixel_width,
+                pixel_height,
+            })))
+        }
+        _ => Err(DecodeError::Other("invalid popup option")),
+    }
+}
+
+fn decode_splits<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+) -> Result<Vec<PaneSurfaceSplit>, DecodeError> {
+    let count = decode_count(decoder, MAX_SPLITS, "too many splits")?;
+    let mut splits = Vec::with_capacity(count);
+    for _ in 0..count {
+        splits.push(PaneSurfaceSplit {
+            direction: PaneSurfaceSplitDirection::decode(decoder)?,
+            pos: u16::decode(decoder)?,
+            area: SurfaceRect::decode(decoder)?,
+            hit_rect: SurfaceRect::decode(decoder)?,
+            path: decode_bounded_vec::<D, bool>(decoder, MAX_SPLIT_PATH, "split path too long")?,
+        });
+    }
+    Ok(splits)
+}
+
+fn decode_graphics_scene<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+) -> Result<SurfaceGraphicsScene, DecodeError> {
+    let asset_count = decode_count(decoder, MAX_ASSETS, "too many graphics assets")?;
+    let mut assets = Vec::with_capacity(asset_count);
+    for _ in 0..asset_count {
+        assets.push(SurfaceGraphicsAsset {
+            key: SurfaceGraphicsAssetKey::decode(decoder)?,
+            data: decode_bytes_vec(
+                decoder,
+                MAX_GRAPHICS_FRAME_SIZE,
+                "graphics asset payload too large",
+            )?,
+        });
+    }
+    let placements = decode_bounded_vec::<D, SurfaceGraphicsPlacement>(
+        decoder,
+        MAX_PLACEMENTS,
+        "too many graphics placements",
+    )?;
+    let retained_assets = decode_bounded_vec::<D, SurfaceGraphicsAssetKey>(
+        decoder,
+        MAX_RETAINED_ASSETS,
+        "too many retained graphics assets",
+    )?;
+    Ok(SurfaceGraphicsScene {
+        assets,
+        placements,
+        retained_assets,
+    })
+}
+
+fn decode_popup_update<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+    popup: Option<&ClientShellPopupSurface>,
+) -> Result<Option<GridUpdate>, DecodeError> {
+    match u8::decode(decoder)? {
+        0 => Ok(None),
+        1 => {
+            let popup = popup.ok_or(DecodeError::Other(
+                "popup cell update has no popup metadata",
+            ))?;
+            let width = popup.frame.width;
+            let height = popup.frame.height;
+            let update = match u32::decode(decoder)? {
+                0 => GridUpdate::Patch(decode_rows(decoder, width, height)?),
+                1 => {
+                    let expected = checked_grid_size(width, height)?;
+                    let count = decode_sequence_len(decoder)?;
+                    if count != expected {
+                        return Err(DecodeError::Other(
+                            "popup replacement does not match popup dimensions",
+                        ));
+                    }
+                    let mut cells = Vec::with_capacity(count);
+                    for _ in 0..count {
+                        cells.push(CellData::decode(decoder)?);
+                    }
+                    GridUpdate::Replace(cells)
+                }
+                _ => return Err(DecodeError::Other("invalid popup grid update")),
+            };
+            Ok(Some(update))
+        }
+        _ => Err(DecodeError::Other("invalid popup update option")),
+    }
+}
+
+fn decode_rows<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+    width: u16,
+    height: u16,
+) -> Result<Vec<PaneSurfacePatchRow>, DecodeError> {
+    let cell_budget = checked_grid_size(width, height)?;
+    let count = decode_count(
+        decoder,
+        super::MAX_SPANS.min(cell_budget),
+        "too many surface delta spans",
+    )?;
+    let row_width = usize::from(width);
+    let mut total_cells = 0usize;
+    let mut previous_end = 0usize;
+    let mut rows = Vec::with_capacity(count);
+    for _ in 0..count {
+        let x = u16::decode(decoder)?;
+        let y = u16::decode(decoder)?;
+        let cells_len = decode_sequence_len(decoder)?;
+        let x_usize = usize::from(x);
+        let y_usize = usize::from(y);
+        if cells_len == 0 || y >= height || x_usize >= row_width || cells_len > row_width - x_usize
+        {
+            return Err(DecodeError::Other("surface delta span is outside its row"));
+        }
+        let start = y_usize
+            .checked_mul(row_width)
+            .and_then(|value| value.checked_add(x_usize))
+            .ok_or(DecodeError::Other("surface delta span overflow"))?;
+        if start < previous_end {
+            return Err(DecodeError::Other(
+                "surface delta spans overlap or are not sorted",
+            ));
+        }
+        total_cells = total_cells
+            .checked_add(cells_len)
+            .ok_or(DecodeError::Other("surface delta cell budget overflow"))?;
+        if total_cells > cell_budget {
+            return Err(DecodeError::Other("surface delta cell budget exceeded"));
+        }
+        let mut cells = Vec::with_capacity(cells_len);
+        for _ in 0..cells_len {
+            cells.push(CellData::decode(decoder)?);
+        }
+        previous_end = start + cells_len;
+        rows.push(PaneSurfacePatchRow { x, y, cells });
+    }
+    Ok(rows)
+}
+
+fn require_empty_sequence<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+    message: &'static str,
+) -> Result<(), DecodeError> {
+    if decode_sequence_len(decoder)? == 0 {
+        Ok(())
+    } else {
+        Err(DecodeError::Other(message))
+    }
+}
+
+fn decode_bounded_vec<D, T>(
+    decoder: &mut D,
+    max: usize,
+    message: &'static str,
+) -> Result<Vec<T>, DecodeError>
+where
+    D: Decoder<Context = DecodeContext>,
+    T: Decode<DecodeContext>,
+{
+    let count = decode_count(decoder, max, message)?;
+    let mut values = Vec::with_capacity(count);
+    for _ in 0..count {
+        values.push(T::decode(decoder)?);
+    }
+    Ok(values)
+}
+
+fn decode_bytes_vec<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+    max: usize,
+    message: &'static str,
+) -> Result<Vec<u8>, DecodeError> {
+    let count = decode_count(decoder, max, message)?;
+    decoder.claim_container_read::<u8>(count)?;
+    let mut bytes = vec![0; count];
+    decoder.reader().read(&mut bytes)?;
+    Ok(bytes)
+}
+
+fn decode_count<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+    max: usize,
+    message: &'static str,
+) -> Result<usize, DecodeError> {
+    let count = decode_sequence_len(decoder)?;
+    if count > max {
+        Err(DecodeError::Other(message))
+    } else {
+        Ok(count)
+    }
+}
+
+fn decode_sequence_len<D: Decoder<Context = DecodeContext>>(
+    decoder: &mut D,
+) -> Result<usize, DecodeError> {
+    let count = u64::decode(decoder)?;
+    usize::try_from(count).map_err(|_| DecodeError::OutsideUsizeRange(count))
+}
+
+fn checked_grid_size(width: u16, height: u16) -> Result<usize, DecodeError> {
+    if width > MAX_GRID_DIMENSION || height > MAX_GRID_DIMENSION {
+        return Err(DecodeError::Other("surface dimensions exceed the limit"));
+    }
+    let cells = usize::from(width) * usize::from(height);
+    if cells > MAX_GRID_CELLS {
+        Err(DecodeError::Other("surface cell count exceeds the limit"))
+    } else {
+        Ok(cells)
+    }
+}
+
+/// Mirrors all decoder-side metadata limits. The sparse sender must fall back
+/// to the full-frame codec when this returns false.
+pub(super) fn metadata_fits(surface: &PaneSurfaceFrame) -> bool {
+    frame_metadata_fits(&surface.frame)
+        && surface.panes.len() <= MAX_PANES
+        && surface.splits.len() <= MAX_SPLITS
+        && surface
+            .splits
+            .iter()
+            .all(|split| split.path.len() <= MAX_SPLIT_PATH)
+        && surface
+            .popup
+            .as_deref()
+            .is_none_or(|popup| frame_metadata_fits(&popup.frame))
+        && surface.graphics.assets.len() <= MAX_ASSETS
+        && surface
+            .graphics
+            .assets
+            .iter()
+            .all(|asset| asset.data.len() <= MAX_GRAPHICS_FRAME_SIZE)
+        && surface.graphics.placements.len() <= MAX_PLACEMENTS
+        && surface.graphics.retained_assets.len() <= MAX_RETAINED_ASSETS
+}
+
+fn frame_metadata_fits(frame: &FrameData) -> bool {
+    checked_grid_size(frame.width, frame.height).is_ok_and(|cells| frame.cells.len() == cells)
+        && frame.hyperlinks.len() <= MAX_HYPERLINKS
+        && frame.graphics.len() <= MAX_GRAPHICS_FRAME_SIZE
+}
+
+#[cfg(test)]
+mod tests {
+    use serde::{ser::SerializeSeq as _, Serialize, Serializer};
+
+    use super::*;
+    use crate::protocol::{
+        ClientShellPopupSize, SurfaceGraphicsFormat, SurfaceGraphicsSource, SurfaceGraphicsTarget,
+    };
+
+    fn cell() -> CellData {
+        CellData {
+            symbol: "x".into(),
+            fg: 1,
+            bg: 2,
+            modifier: 0,
+            skip: false,
+            hyperlink: None,
+        }
+    }
+
+    fn surface(width: u16, height: u16) -> PaneSurfaceFrame {
+        PaneSurfaceFrame {
+            boot_id: "boot".into(),
+            projection_revision: 2,
+            surface_revision: 3,
+            frame: FrameData {
+                cells: Vec::new(),
+                width,
+                height,
+                cursor: None,
+                hyperlinks: Vec::new(),
+                graphics: Vec::new(),
+            },
+            panes: Vec::new(),
+            splits: Vec::new(),
+            popup: None,
+            graphics: SurfaceGraphicsScene::default(),
+        }
+    }
+
+    fn encoded<T: Serialize>(value: &T) -> Vec<u8> {
+        bincode::serde::encode_to_vec(value, bincode::config::standard())
+            .expect("encode test value")
+    }
+
+    fn graphics_key() -> SurfaceGraphicsAssetKey {
+        SurfaceGraphicsAssetKey {
+            source: SurfaceGraphicsSource::Terminal {
+                target: SurfaceGraphicsTarget::Pane {
+                    pane_id: "pane".into(),
+                },
+                image_id: 1,
+            },
+            image_width: 1,
+            image_height: 1,
+            format: SurfaceGraphicsFormat::Rgba,
+            data_len: 4,
+            data_fingerprint: 1,
+        }
+    }
+
+    #[test]
+    fn native_decoder_matches_the_serde_wire_layout() {
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 2,
+            surface: surface(4, 2),
+            rows: vec![PaneSurfacePatchRow {
+                x: 1,
+                y: 1,
+                cells: vec![cell()],
+            }],
+            popup_cells: None::<GridUpdate>,
+        };
+        let decoded = decode_bytes(&encoded(&delta), Some((4, 2))).expect("decode delta");
+        assert!(decoded.surface.frame.cells.is_empty());
+        assert_eq!(decoded.rows.len(), 1);
+        assert_eq!(decoded.rows[0].cells, vec![cell()]);
+    }
+
+    #[test]
+    fn ordinary_cap_uses_base64_length_and_the_sender_graphics_predicate() {
+        let ordinary = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: surface(1, 1),
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: None::<GridUpdate>,
+        };
+        let bytes = encoded(&ordinary);
+        assert!(decode_bytes_with_data_len(&bytes, Some((1, 1)), MAX_FRAME_SIZE + 1).is_err());
+
+        let mut graphics = surface(1, 1);
+        graphics.graphics.retained_assets.push(graphics_key());
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: graphics,
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: None::<GridUpdate>,
+        };
+        let bytes = encoded(&delta);
+        assert!(decode_bytes_with_data_len(&bytes, Some((1, 1)), MAX_FRAME_SIZE + 1).is_ok());
+    }
+
+    #[test]
+    fn rejects_nonempty_metadata_grids_before_cell_allocation() {
+        let mut main = surface(1, 1);
+        main.frame.cells.push(cell());
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: main,
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: None::<GridUpdate>,
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((1, 1))).is_err());
+
+        let mut popup_surface = surface(1, 1);
+        popup_surface.popup = Some(Box::new(ClientShellPopupSurface {
+            terminal_id: "popup".into(),
+            title: "popup".into(),
+            width: Some(ClientShellPopupSize::Cells(1)),
+            height: Some(ClientShellPopupSize::Cells(1)),
+            frame: FrameData {
+                cells: vec![cell()],
+                width: 1,
+                height: 1,
+                cursor: None,
+                hyperlinks: Vec::new(),
+                graphics: Vec::new(),
+            },
+            mouse_reporting: false,
+            sgr_pixel_mouse: false,
+            pixel_width: 1,
+            pixel_height: 1,
+        }));
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: popup_surface,
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: Some(GridUpdate::Replace(vec![cell()])),
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((1, 1))).is_err());
+    }
+
+    struct CountOnly(usize);
+
+    impl Serialize for CountOnly {
+        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            serializer.serialize_seq(Some(self.0))?.end()
+        }
+    }
+
+    #[test]
+    fn rejects_excessive_span_count_without_allocating_the_claimed_rows() {
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: surface(2, 2),
+            rows: CountOnly(super::super::MAX_SPANS + 1),
+            popup_cells: None::<GridUpdate>,
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((2, 2))).is_err());
+    }
+
+    #[test]
+    fn accepts_max_spans_within_the_grid_cell_budget() {
+        let rows = (0..super::super::MAX_SPANS)
+            .map(|index| PaneSurfacePatchRow {
+                x: (index % 64) as u16,
+                y: (index / 64) as u16,
+                cells: vec![cell()],
+            })
+            .collect::<Vec<_>>();
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: surface(64, 64),
+            rows,
+            popup_cells: None::<GridUpdate>,
+        };
+        let decoded = decode_bytes(&encoded(&delta), Some((64, 64))).expect("bounded delta");
+        assert_eq!(decoded.rows.len(), super::super::MAX_SPANS);
+    }
+
+    #[test]
+    fn row_bounds_use_the_actual_baseline_and_popup_replace_is_exact() {
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: surface(100, 100),
+            rows: vec![PaneSurfacePatchRow {
+                x: 2,
+                y: 0,
+                cells: vec![cell()],
+            }],
+            popup_cells: None::<GridUpdate>,
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((2, 2))).is_err());
+
+        let mut popup_surface = surface(1, 1);
+        popup_surface.popup = Some(Box::new(ClientShellPopupSurface {
+            terminal_id: "popup".into(),
+            title: "popup".into(),
+            width: None,
+            height: None,
+            frame: FrameData {
+                cells: Vec::new(),
+                width: 2,
+                height: 1,
+                cursor: None,
+                hyperlinks: Vec::new(),
+                graphics: Vec::new(),
+            },
+            mouse_reporting: false,
+            sgr_pixel_mouse: false,
+            pixel_width: 1,
+            pixel_height: 1,
+        }));
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: popup_surface,
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: Some(GridUpdate::Replace(vec![cell()])),
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((1, 1))).is_err());
+    }
+
+    struct SurfaceWithPanes<'a> {
+        source: &'a PaneSurfaceFrame,
+        pane_count: usize,
+    }
+
+    impl Serialize for SurfaceWithPanes<'_> {
+        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            use serde::ser::SerializeTuple as _;
+            let mut tuple = serializer.serialize_tuple(8)?;
+            tuple.serialize_element(&self.source.boot_id)?;
+            tuple.serialize_element(&self.source.projection_revision)?;
+            tuple.serialize_element(&self.source.surface_revision)?;
+            tuple.serialize_element(&self.source.frame)?;
+            tuple.serialize_element(&CountOnly(self.pane_count))?;
+            tuple.serialize_element(&self.source.splits)?;
+            tuple.serialize_element(&self.source.popup)?;
+            tuple.serialize_element(&self.source.graphics)?;
+            tuple.end()
+        }
+    }
+
+    #[test]
+    fn rejects_excessive_metadata_count_without_allocating_the_claimed_items() {
+        let source = surface(2, 2);
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: SurfaceWithPanes {
+                source: &source,
+                pane_count: MAX_PANES + 1,
+            },
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: None::<GridUpdate>,
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((2, 2))).is_err());
+    }
+
+    struct OversizedAsset<'a>(&'a SurfaceGraphicsAssetKey);
+
+    impl Serialize for OversizedAsset<'_> {
+        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            use serde::ser::SerializeTuple as _;
+            let mut tuple = serializer.serialize_tuple(2)?;
+            tuple.serialize_element(self.0)?;
+            tuple.serialize_element(&CountOnly(MAX_GRAPHICS_FRAME_SIZE + 1))?;
+            tuple.end()
+        }
+    }
+
+    struct SceneWithOversizedAsset<'a>(&'a SurfaceGraphicsAssetKey);
+
+    impl Serialize for SceneWithOversizedAsset<'_> {
+        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            use serde::ser::{SerializeSeq as _, SerializeTuple as _};
+            let mut tuple = serializer.serialize_tuple(3)?;
+            struct Assets<'a>(&'a SurfaceGraphicsAssetKey);
+            impl Serialize for Assets<'_> {
+                fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+                    let mut sequence = serializer.serialize_seq(Some(1))?;
+                    sequence.serialize_element(&OversizedAsset(self.0))?;
+                    sequence.end()
+                }
+            }
+            tuple.serialize_element(&Assets(self.0))?;
+            tuple.serialize_element(&Vec::<SurfaceGraphicsPlacement>::new())?;
+            tuple.serialize_element(&Vec::<SurfaceGraphicsAssetKey>::new())?;
+            tuple.end()
+        }
+    }
+
+    struct SurfaceWithGraphics<'a> {
+        source: &'a PaneSurfaceFrame,
+        key: &'a SurfaceGraphicsAssetKey,
+    }
+
+    impl Serialize for SurfaceWithGraphics<'_> {
+        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+            use serde::ser::SerializeTuple as _;
+            let mut tuple = serializer.serialize_tuple(8)?;
+            tuple.serialize_element(&self.source.boot_id)?;
+            tuple.serialize_element(&self.source.projection_revision)?;
+            tuple.serialize_element(&self.source.surface_revision)?;
+            tuple.serialize_element(&self.source.frame)?;
+            tuple.serialize_element(&self.source.panes)?;
+            tuple.serialize_element(&self.source.splits)?;
+            tuple.serialize_element(&self.source.popup)?;
+            tuple.serialize_element(&SceneWithOversizedAsset(self.key))?;
+            tuple.end()
+        }
+    }
+
+    #[test]
+    fn rejects_oversized_pixel_payload_prefix_before_allocating_it() {
+        let source = surface(2, 2);
+        let key = graphics_key();
+        let delta = SurfaceDelta {
+            base_projection_revision: 1,
+            base_surface_revision: 1,
+            surface: SurfaceWithGraphics {
+                source: &source,
+                key: &key,
+            },
+            rows: Vec::<PaneSurfacePatchRow>::new(),
+            popup_cells: None::<GridUpdate>,
+        };
+        assert!(decode_bytes(&encoded(&delta), Some((2, 2))).is_err());
+    }
+
+    #[test]
+    fn sender_eligibility_matches_grid_and_metadata_limits() {
+        let mut exact = surface(2, 2);
+        exact.frame.cells = vec![cell(); 4];
+        assert!(metadata_fits(&exact));
+        exact.frame.cells.pop();
+        assert!(!metadata_fits(&exact));
+
+        let mut candidate = surface(MAX_GRID_DIMENSION + 1, 1);
+        assert!(!metadata_fits(&candidate));
+        candidate.frame.width = 1;
+        candidate.frame.cells = vec![cell()];
+        candidate.frame.hyperlinks = vec![String::new(); MAX_HYPERLINKS + 1];
+        assert!(!metadata_fits(&candidate));
+    }
+}
diff --git a/src/protocol/surface_reuse.rs b/src/protocol/surface_reuse.rs
index bba85824..aa2187d4 100644
--- a/src/protocol/surface_reuse.rs
+++ b/src/protocol/surface_reuse.rs
@@ -50,6 +50,23 @@ struct CellBaseline {
     width: u16,
     height: u16,
     cells: Vec<CellData>,
+    popup: Option<PopupBaseline>,
+}
+
+struct PopupBaseline {
+    terminal_id: String,
+    width: u16,
+    height: u16,
+    cells: Vec<CellData>,
+}
+
+fn popup_baseline(surface: &PaneSurfaceFrame) -> Option<PopupBaseline> {
+    surface.popup.as_ref().map(|popup| PopupBaseline {
+        terminal_id: popup.terminal_id.clone(),
+        width: popup.frame.width,
+        height: popup.frame.height,
+        cells: popup.frame.cells.clone(),
+    })
 }
 
 /// Connection-local decoding happens before activation and presentation filtering, so
@@ -57,11 +74,27 @@ struct CellBaseline {
 #[derive(Default)]
 pub(crate) struct Decoder {
     baseline: Option<CellBaseline>,
+    surface_delta: bool,
 }
 
 impl Decoder {
+    pub(crate) fn new(surface_delta: bool) -> Self {
+        Self {
+            baseline: None,
+            surface_delta,
+        }
+    }
+
     pub(crate) fn decode(&mut self, message: ServerMessage) -> Result<ServerMessage, String> {
         let message = match message {
+            ServerMessage::EndpointControl { kind, data }
+                if kind == super::surface_delta::MESSAGE_KIND =>
+            {
+                if !self.surface_delta {
+                    return Err("surface delta was not negotiated".into());
+                }
+                return self.decode_delta(&data).map(ServerMessage::PaneSurface);
+            }
             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}"))?;
@@ -81,6 +114,9 @@ impl Decoder {
                 surface.frame.cells.clone_from(&base.cells);
                 base.projection_revision = surface.projection_revision;
                 base.surface_revision = surface.surface_revision;
+                if self.surface_delta {
+                    base.popup = popup_baseline(&surface);
+                }
                 return Ok(ServerMessage::PaneSurface(surface));
             }
             message => message,
@@ -94,6 +130,9 @@ impl Decoder {
                 base.width = surface.frame.width;
                 base.height = surface.frame.height;
                 base.cells.clone_from(&surface.frame.cells);
+                if self.surface_delta {
+                    base.popup = popup_baseline(surface);
+                }
             }
             ServerMessage::PaneSurfacePatch(patch) => {
                 if let Some(base) = &mut self.baseline {
@@ -124,4 +163,64 @@ impl Decoder {
         }
         Ok(message)
     }
+
+    fn decode_delta(&mut self, data: &str) -> Result<PaneSurfaceFrame, String> {
+        use super::surface_delta::{self, GridUpdate};
+        let Some(base) = &mut self.baseline else {
+            return Err("surface delta without a baseline".into());
+        };
+        let delta = surface_delta::decode_for(data, (base.width, base.height))?;
+        let mut surface = delta.surface;
+        if surface.boot_id != base.boot_id
+            || delta.base_projection_revision != base.projection_revision
+            || delta.base_surface_revision != base.surface_revision
+            || surface.surface_revision != base.surface_revision.saturating_add(1)
+            || surface.projection_revision < base.projection_revision
+            || base.cells.len() != usize::from(base.width) * usize::from(base.height)
+        {
+            return Err("surface delta does not match its baseline".into());
+        }
+        surface.frame.cells.clone_from(&base.cells);
+        surface_delta::apply_rows(&mut surface.frame.cells, base.width, &delta.rows);
+        match (&mut surface.popup, delta.popup_cells) {
+            (None, None) => {}
+            (Some(popup), Some(update)) => {
+                let count = usize::from(popup.frame.width) * usize::from(popup.frame.height);
+                match update {
+                    GridUpdate::Replace(cells) => popup.frame.cells = cells,
+                    GridUpdate::Patch(rows) => {
+                        let previous = base
+                            .popup
+                            .as_ref()
+                            .filter(|previous| {
+                                previous.terminal_id == popup.terminal_id
+                                    && previous.width == popup.frame.width
+                                    && previous.height == popup.frame.height
+                                    && previous.cells.len() == count
+                            })
+                            .ok_or("popup delta does not match its baseline")?;
+                        popup.frame.cells.clone_from(&previous.cells);
+                        surface_delta::apply_rows(&mut popup.frame.cells, previous.width, &rows);
+                    }
+                }
+            }
+            _ => return Err("popup delta is missing or unexpected".into()),
+        }
+        for frame in
+            std::iter::once(&surface.frame).chain(surface.popup.as_ref().map(|popup| &popup.frame))
+        {
+            if frame.cells.iter().any(|cell| {
+                cell.hyperlink
+                    .is_some_and(|index| index as usize >= frame.hyperlinks.len())
+            }) {
+                return Err("surface delta has an invalid hyperlink index".into());
+            }
+        }
+        // Validate the entire update before advancing either grid or revision.
+        surface_delta::apply_rows(&mut base.cells, base.width, &delta.rows);
+        base.popup = popup_baseline(&surface);
+        base.projection_revision = surface.projection_revision;
+        base.surface_revision = surface.surface_revision;
+        Ok(surface)
+    }
 }
diff --git a/src/protocol/wire.rs b/src/protocol/wire.rs
index a29c6ca6..aba166d6 100644
--- a/src/protocol/wire.rs
+++ b/src/protocol/wire.rs
@@ -699,7 +699,7 @@ pub enum AttachScrollSource {
 
 /// A single cell in a rendered frame, serialized independently from ratatui's
 /// `Cell` type to keep the wire protocol stable.
-#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct CellData {
     /// Grapheme cluster displayed in this cell (usually 1–2 chars).
     pub symbol: String,
@@ -751,7 +751,7 @@ impl CellData {
 pub type CursorShapeParam = u8;
 
 /// Cursor position within a rendered frame.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct CursorState {
     /// Column offset (0-based) of the cursor.
     pub x: u16,
@@ -1092,7 +1092,7 @@ pub struct ClientShellAgent {
 }
 
 /// Origin-relative geometry for one pane in a rendered pane surface.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct PaneSurfacePane {
     pub pane_id: String,
     pub content_revision: u64,
@@ -1108,7 +1108,7 @@ pub struct PaneSurfacePane {
     pub pixel_height: u32,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct PaneSurfaceScrollMetrics {
     pub offset_from_bottom: u64,
     pub max_offset_from_bottom: u64,
@@ -1125,14 +1125,14 @@ pub struct PaneSurfaceSplit {
     pub path: Vec<bool>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub enum PaneSurfaceSplitDirection {
     Horizontal,
     Vertical,
 }
 
 /// Wire-safe rectangle relative to a pane surface.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct SurfaceRect {
     pub x: u16,
     pub y: u16,
@@ -1151,13 +1151,13 @@ impl From<ratatui::layout::Rect> for SurfaceRect {
     }
 }
 
-#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub enum SurfaceGraphicsTarget {
     Pane { pane_id: String },
     Popup { terminal_id: String },
 }
 
-#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub enum SurfaceGraphicsSource {
     Terminal {
         target: SurfaceGraphicsTarget,
@@ -1169,14 +1169,14 @@ pub enum SurfaceGraphicsSource {
     },
 }
 
-#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub enum SurfaceGraphicsFormat {
     Rgb,
     Rgba,
     Png,
 }
 
-#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct SurfaceGraphicsAssetKey {
     pub source: SurfaceGraphicsSource,
     pub image_width: u32,
@@ -1194,7 +1194,7 @@ pub struct SurfaceGraphicsAsset {
 }
 
 /// One already-clipped desired placement relative to its target surface.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub struct SurfaceGraphicsPlacement {
     pub asset: SurfaceGraphicsAssetKey,
     pub logical_placement_id: u32,
@@ -1239,7 +1239,7 @@ pub struct PaneSurfaceFrame {
     pub graphics: SurfaceGraphicsScene,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, bincode::Decode)]
 pub enum ClientShellPopupSize {
     Cells(u16),
     Percent(u8),
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index aba4ee76..0d297fb9 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -399,6 +399,7 @@ pub(crate) enum ServerEvent {
         mouse_capture: bool,
         surface_active: bool,
         surface_reuse: bool,
+        surface_delta: bool,
         writer: ClientWriter,
     },
     /// A client sent an input message.
@@ -765,6 +766,7 @@ pub(crate) fn handle_client_handshake(
                     hello.mouse_capture,
                     hello.surface_active,
                     hello.surface_reuse,
+                    hello.surface_delta,
                 )),
             )
         }
@@ -860,6 +862,7 @@ pub(crate) fn handle_client_handshake(
         mouse_capture,
         surface_active,
         surface_reuse,
+        surface_delta,
     )) = shell_options
     {
         ServerEvent::ClientShellConnected {
@@ -874,6 +877,7 @@ pub(crate) fn handle_client_handshake(
             mouse_capture,
             surface_active,
             surface_reuse,
+            surface_delta,
             writer,
         }
     } else {
@@ -1449,6 +1453,7 @@ mod tests {
             mouse_capture: true,
             surface_active: true,
             surface_reuse: false,
+            surface_delta: 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()],
@@ -1965,9 +1970,11 @@ mod tests {
                 mouse_capture,
                 surface_active,
                 surface_reuse,
+                surface_delta,
                 writer,
             } => {
                 assert!(!surface_reuse);
+                assert!(!surface_delta);
                 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/clients.rs b/src/server/clients.rs
index c2b419f4..aeda3822 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -261,6 +261,10 @@ impl ClientConnection {
         self.render_state.request_repaint();
     }
 
+    pub(crate) fn request_recompute(&mut self) {
+        self.render_state.request_recompute();
+    }
+
     pub(crate) fn track_shell_input(
         &mut self,
         target: ClientShellInputTarget,
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 01d91bbd..f07c0c37 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -779,7 +779,7 @@ impl HeadlessServer {
         // rendering semantics. Force one fresh frame to every remaining client
         // even if the next rendered buffer compares equal to its cached frame.
         for client in self.clients.values_mut() {
-            client.request_repaint();
+            client.request_recompute();
         }
         if !start_pending_agent_resumes {
             self.app.pending_agent_resume_deadline = None;
@@ -792,7 +792,7 @@ impl HeadlessServer {
             .start_pending_agent_resumes(self.app.pending_agent_resume_due(now))
         {
             for client in self.clients.values_mut() {
-                client.request_repaint();
+                client.request_recompute();
             }
         }
     }
@@ -1969,6 +1969,7 @@ impl HeadlessServer {
                 mouse_capture,
                 surface_active,
                 surface_reuse,
+                surface_delta,
                 writer,
             } => {
                 if self.handoff_in_progress {
@@ -2015,6 +2016,7 @@ impl HeadlessServer {
                 connection.shell_mouse_capture = mouse_capture;
                 connection.shell_surface_active = surface_active;
                 connection.render_state.enable_surface_reuse(surface_reuse);
+                connection.render_state.enable_surface_delta(surface_delta);
                 connection.shell_projection_revision = 1;
                 let config_diagnostic = if endpoint_keybindings {
                     self.server_config_diagnostic.as_deref()
diff --git a/src/server/headless/client_views.rs b/src/server/headless/client_views.rs
index d6030f74..4e0bd2e2 100644
--- a/src/server/headless/client_views.rs
+++ b/src/server/headless/client_views.rs
@@ -516,7 +516,7 @@ impl HeadlessServer {
 
     fn finish_shell_tab_geometry_change(&mut self, start_pending_agent_resumes: bool) {
         for client in self.clients.values_mut() {
-            client.request_repaint();
+            client.request_recompute();
         }
         if !start_pending_agent_resumes {
             self.app.pending_agent_resume_deadline = None;
@@ -529,7 +529,7 @@ impl HeadlessServer {
             .start_pending_agent_resumes(self.app.pending_agent_resume_due(now))
         {
             for client in self.clients.values_mut() {
-                client.request_repaint();
+                client.request_recompute();
             }
         }
     }
diff --git a/src/server/headless/retained_surface.rs b/src/server/headless/retained_surface.rs
index d0fb9bba..2a637eac 100644
--- a/src/server/headless/retained_surface.rs
+++ b/src/server/headless/retained_surface.rs
@@ -265,6 +265,9 @@ impl HeadlessServer {
                 crate::render_prof::event("retained_surface.recipient_deferred");
                 continue;
             }
+            if client.render_state.requires_recompute() {
+                fallback!("recompute_pending");
+            }
             let Some(surface) = client.render_state.last_pane_surface() else {
                 fallback!("no_baseline");
             };
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 093b5625..60f89d63 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -4,6 +4,8 @@ use super::*;
 mod pane_graphics_tests;
 #[path = "pane_move.rs"]
 mod pane_move_tests;
+#[path = "surface_delta.rs"]
+mod surface_delta_tests;
 #[path = "surface_interest.rs"]
 mod surface_interest_tests;
 
@@ -690,6 +692,7 @@ async fn client_shell_attach_seeds_workspace() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 6,
             surface_cols: 80,
             surface_rows: 23,
@@ -721,6 +724,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols: 80,
             surface_rows: 23,
@@ -838,6 +842,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 77,
             surface_cols: 80,
             surface_rows: 23,
@@ -947,6 +952,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 7,
             surface_cols: 80,
             surface_rows: 23,
@@ -1115,6 +1121,7 @@ fn connect_test_shell(
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols,
             surface_rows,
@@ -1577,6 +1584,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 13,
             surface_cols: 80,
             surface_rows: 23,
@@ -1602,6 +1610,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 14,
             surface_cols: 80,
             surface_rows: 23,
@@ -2505,6 +2514,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 9,
             surface_cols: 80,
             surface_rows: 23,
@@ -2756,6 +2766,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: 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
new file mode 100644
index 00000000..f010c1e3
--- /dev/null
+++ b/src/server/headless/tests/surface_delta.rs
@@ -0,0 +1,228 @@
+use super::*;
+
+fn receive_message(receiver: &std::sync::mpsc::Receiver<Vec<u8>>) -> (Vec<u8>, ServerMessage) {
+    let bytes = receiver
+        .recv_timeout(Duration::from_secs(1))
+        .expect("render message");
+    let message = read_server_message(bytes.clone());
+    (bytes, message)
+}
+
+fn decode_surface_message(
+    decoder: &mut protocol::surface_reuse::Decoder,
+    message: ServerMessage,
+) -> crate::protocol::PaneSurfaceFrame {
+    match decoder.decode(message).expect("decode surface message") {
+        ServerMessage::PaneSurface(surface) => surface,
+        other => panic!("expected decoded pane surface, got {other:?}"),
+    }
+}
+
+fn without_asset_payload(
+    mut surface: crate::protocol::PaneSurfaceFrame,
+) -> crate::protocol::PaneSurfaceFrame {
+    surface.graphics.assets.clear();
+    surface
+}
+
+fn enable_delta(server: &mut HeadlessServer, client_id: u64) {
+    server
+        .clients
+        .get_mut(&client_id)
+        .expect("delta client")
+        .render_state
+        .enable_surface_delta(true);
+}
+
+fn fill_render_lane(server: &HeadlessServer, client_id: u64) {
+    let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
+        .expect("dummy frame");
+    server.clients[&client_id]
+        .writer
+        .as_ref()
+        .expect("client writer")
+        .test_fill_render(queued);
+}
+
+#[tokio::test]
+async fn surface_delta_reconstructs_metadata_text_and_hyperlinks() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(b"initial text");
+    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 initial = decode_surface_message(&mut decoder, initial_message);
+    assert_eq!(
+        without_asset_payload(initial.clone()),
+        without_asset_payload(
+            server.clients[&1]
+                .render_state
+                .last_pane_surface()
+                .expect("initial baseline")
+                .clone()
+        )
+    );
+
+    let initial_projection_revision = initial.projection_revision;
+    server.app.state.workspaces[0].custom_name = Some("renamed workspace".into());
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\rupdated text \x1b]8;;https://example.test/path\x1b\\linked\x1b]8;;\x1b\\",
+    );
+    server.clients.get_mut(&1).unwrap().request_recompute();
+    assert!(!server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    server.render_and_stream();
+    let (delta_bytes, delta_message) = receive_message(&render_rx);
+    assert!(delta_bytes.len() < initial_bytes.len());
+    assert!(matches!(
+        &delta_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let decoded = decode_surface_message(&mut decoder, delta_message);
+    let committed = server.clients[&1]
+        .render_state
+        .last_pane_surface()
+        .expect("updated baseline")
+        .clone();
+    assert_eq!(
+        without_asset_payload(decoded.clone()),
+        without_asset_payload(committed)
+    );
+    assert!(decoded.projection_revision > initial_projection_revision);
+    assert!(frame_text(&decoded.frame).contains("updated text"));
+    assert_eq!(decoded.frame.hyperlinks, vec!["https://example.test/path"]);
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn surface_delta_updates_global_popup_without_affecting_legacy_peer() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("popup-delta");
+    let pane_id = workspace.focused_pane_id().expect("focused pane");
+    workspace.insert_test_runtime(
+        pane_id,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 23, b"base"),
+    );
+    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 popup_runtime = crate::terminal::TerminalRuntime::test_with_screen_bytes(40, 12, b"POPUP");
+    let (_, popup_terminal_id) = server.app.install_test_popup_runtime(popup_runtime);
+
+    let (_delta_control, delta_render) = connect_matching_test_shell(&mut server, 1);
+    let (_legacy_control, legacy_render) = connect_matching_test_shell(&mut server, 2);
+    enable_delta(&mut server, 1);
+    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 _ = decode_surface_message(&mut decoder, initial_message);
+
+    server
+        .app
+        .terminal_runtimes
+        .get(&popup_terminal_id)
+        .expect("popup runtime")
+        .test_process_pty_bytes(b"\rPOPUP updated");
+    server.clients.get_mut(&1).unwrap().request_recompute();
+    server.render_and_stream();
+
+    let (delta_bytes, delta_message) = receive_message(&delta_render);
+    assert!(delta_bytes.len() * 4 < initial_bytes.len());
+    assert!(matches!(
+        &delta_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let decoded = decode_surface_message(&mut decoder, delta_message);
+    assert!(decoded
+        .popup
+        .as_ref()
+        .is_some_and(|popup| frame_text(&popup.frame).contains("POPUP updated")));
+
+    let (_, legacy_message) = receive_message(&legacy_render);
+    assert!(matches!(legacy_message, ServerMessage::PaneSurface(_)));
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn surface_delta_preserves_graphics_baseline_through_queue_recovery() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(b"text before image");
+    server.clients.get_mut(&1).unwrap().cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    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 _ = decode_surface_message(&mut decoder, initial_message);
+
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
+    );
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let (image_bytes, image_message) = receive_message(&render_rx);
+    assert!(image_bytes.len() * 4 < initial_bytes.len());
+    assert!(matches!(
+        &image_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let image = decode_surface_message(&mut decoder, image_message);
+    assert!(!image.graphics.placements.is_empty());
+
+    write_shared_test_pane(&mut server, pane_id, b"\rupdated text");
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let (text_bytes, text_message) = receive_message(&render_rx);
+    assert!(text_bytes.len() * 4 < initial_bytes.len());
+    assert!(matches!(
+        &text_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let text_update = decode_surface_message(&mut decoder, text_message);
+    assert_eq!(text_update.graphics.placements, image.graphics.placements);
+    assert!(text_update.graphics.assets.is_empty());
+
+    let baseline_before_queue = server.clients[&1]
+        .render_state
+        .last_pane_surface()
+        .expect("graphics baseline")
+        .clone();
+    fill_render_lane(&server, 1);
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\x1b_Ga=T,f=32,t=d,i=8,p=3,s=1,v=1,c=1,r=1,q=2;AP8A/w==\x1b\\",
+    );
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    assert_eq!(
+        without_asset_payload(
+            server.clients[&1]
+                .render_state
+                .last_pane_surface()
+                .expect("baseline after full queue")
+                .clone()
+        ),
+        without_asset_payload(baseline_before_queue)
+    );
+    let _ = render_rx
+        .recv_timeout(Duration::from_secs(1))
+        .expect("drain full render lane");
+
+    server.render_and_stream();
+    let (_, recovered_message) = receive_message(&render_rx);
+    let recovered = decode_surface_message(&mut decoder, recovered_message);
+    assert!(!recovered.graphics.placements.is_empty());
+    assert_eq!(recovered.graphics.assets.len(), 1);
+    assert_eq!(recovered.graphics.assets[0].data, [0, 255, 0, 255]);
+    shutdown_test_runtimes(&mut server);
+}
diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs
index 554b216e..7c8b33f8 100644
--- a/src/server/headless/tests/surface_interest.rs
+++ b/src/server/headless/tests/surface_interest.rs
@@ -65,6 +65,7 @@ async fn metadata_only_shell_is_isolated_until_surface_activation() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols: 101,
             surface_rows: 37,
@@ -282,6 +283,7 @@ async fn background_surface_activation_preserves_focused_viewer_geometry() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 8,
             surface_cols: 100,
             surface_rows: 35,
@@ -394,6 +396,7 @@ async fn presentation_sync_epoch_replays_modes_and_title() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -509,6 +512,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     assert!(
         source_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: source_client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -535,6 +539,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     assert!(
         target_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: target_client_id,
             surface_cols: 80,
             surface_rows: 24,
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index 82f7ba91..5c3935de 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -312,9 +312,172 @@ fn print_token_rule_profiles() {
     }
 }
 
+#[derive(Clone, Copy)]
+enum SurfaceDamagePattern {
+    Dense,
+    Checkerboard,
+}
+
+impl SurfaceDamagePattern {
+    fn name(self) -> &'static str {
+        match self {
+            Self::Dense => "dense",
+            Self::Checkerboard => "checkerboard",
+        }
+    }
+
+    fn apply(self, surface: &mut PaneSurfaceFrame, iteration: usize) {
+        for (index, cell) in surface.frame.cells.iter_mut().enumerate() {
+            cell.symbol = if matches!(self, Self::Checkerboard) && !index.is_multiple_of(2) {
+                "c"
+            } else if iteration.is_multiple_of(2) {
+                "a"
+            } else {
+                "b"
+            }
+            .into();
+        }
+    }
+}
+
+struct SurfaceEncodingClient {
+    state: super::render_stream::ClientRenderState,
+    decoder: Option<crate::protocol::surface_reuse::Decoder>,
+}
+
+fn surface_encoding_client(mode: &str, surface: &PaneSurfaceFrame) -> SurfaceEncodingClient {
+    let reuse = mode == "reuse";
+    let delta = mode == "delta";
+    let mut client = SurfaceEncodingClient {
+        state: super::render_stream::ClientRenderState::new(
+            crate::protocol::RenderEncoding::SemanticFrame,
+        ),
+        decoder: (reuse || delta).then(|| crate::protocol::surface_reuse::Decoder::new(delta)),
+    };
+    client.state.enable_surface_reuse(reuse);
+    client.state.enable_surface_delta(delta);
+    let initial = client
+        .state
+        .prepare_pane_surface(surface.clone())
+        .expect("initial surface should be prepared");
+    if let Some(decoder) = client.decoder.as_mut() {
+        decoder.decode(initial.message().clone()).unwrap();
+    }
+    client.state.commit_sent_frame(initial);
+    client
+}
+
+fn profile_surface_damage(
+    surface: &PaneSurfaceFrame,
+    pattern: SurfaceDamagePattern,
+    mode: &str,
+    client_count: usize,
+) {
+    let mut clients = (0..client_count)
+        .map(|_| surface_encoding_client(mode, surface))
+        .collect::<Vec<_>>();
+    let mut samples = Vec::with_capacity(SAMPLE_COUNT);
+    let mut bytes_per_update = 0usize;
+    for iteration in 0..WARMUP_COUNT + SAMPLE_COUNT {
+        let mut candidate = surface.clone();
+        candidate.projection_revision = iteration as u64 + 2;
+        pattern.apply(&mut candidate, iteration);
+        if iteration > 0 && matches!(pattern, SurfaceDamagePattern::Checkerboard) {
+            let previous = clients[0].state.last_pane_surface().unwrap();
+            let changed = previous
+                .frame
+                .cells
+                .iter()
+                .zip(&candidate.frame.cells)
+                .filter(|(old, new)| old != new)
+                .count();
+            assert_eq!(changed, candidate.frame.cells.len().div_ceil(2));
+        }
+        let started = Instant::now();
+        bytes_per_update = 0;
+        for client in &mut clients {
+            let prepared = client
+                .state
+                .prepare_pane_surface(candidate.clone())
+                .expect("damage should produce a surface update");
+            let mut bytes = Vec::new();
+            crate::protocol::write_message(&mut bytes, prepared.message()).unwrap();
+            bytes_per_update += bytes.len();
+            let decoded = crate::protocol::read_message(
+                &mut bytes.as_slice(),
+                crate::protocol::MAX_GRAPHICS_FRAME_SIZE,
+            )
+            .unwrap();
+            black_box(if let Some(decoder) = client.decoder.as_mut() {
+                decoder.decode(decoded).unwrap()
+            } else {
+                decoded
+            });
+            client.state.commit_sent_frame(prepared);
+        }
+        if iteration >= WARMUP_COUNT {
+            samples.push(started.elapsed());
+        }
+    }
+    let stats = summarize(samples);
+    println!(
+        "  {:<11}  {client_count:>7}  {mode:>7}  {:>9}  {:>12}",
+        pattern.name(),
+        stats.median_us,
+        bytes_per_update
+    );
+}
+
+fn print_surface_damage_profiles() {
+    println!("surface dense/checkerboard encode + decode at {COLS}x{ROWS}");
+    println!("  pattern       clients     mode  median_us  bytes/update");
+    for pane_count in [1, 15] {
+        let mut pipeline = RenderPipeline::new(active_panes(pane_count));
+        pipeline.render_once();
+        let rendered = super::client_shell::render_pane_surface(
+            &mut pipeline.app,
+            Some(crate::ui::TabSurfaceTarget {
+                workspace_index: 0,
+                tab_index: 0,
+            }),
+            Rect::new(0, 0, COLS, ROWS),
+            true,
+            false,
+            HostCellSize::default(),
+            &pipeline.graphics_delivery,
+            1,
+        );
+        let mut surface = PaneSurfaceFrame {
+            boot_id: "bench-boot".into(),
+            projection_revision: 1,
+            surface_revision: 0,
+            frame: rendered.frame,
+            panes: rendered.panes,
+            splits: rendered.splits,
+            popup: rendered.popup,
+            graphics: rendered.graphics,
+        };
+        // Normalize the baseline so dense and checkerboard iterations have stable damage.
+        for cell in &mut surface.frame.cells {
+            cell.symbol = "a".into();
+        }
+        println!("  active panes={pane_count}");
+        for pattern in [
+            SurfaceDamagePattern::Dense,
+            SurfaceDamagePattern::Checkerboard,
+        ] {
+            for client_count in [1, 15] {
+                for mode in ["legacy", "reuse", "delta"] {
+                    profile_surface_damage(&surface, pattern, mode, client_count);
+                }
+            }
+        }
+    }
+}
+
 fn print_surface_reuse_profiles() {
     println!("surface encode + decode at {COLS}x{ROWS}");
-    println!("  layout       panes  cells_changed  reuse  median_us  bytes_per_update");
+    println!("  layout       panes  cells_changed  mode     median_us  bytes_per_update");
     for (label, build) in [
         ("background", workspaces as fn(usize) -> Vec<Workspace>),
         ("active", active_panes),
@@ -346,17 +509,8 @@ fn print_surface_reuse_profiles() {
                 graphics: rendered.graphics,
             };
             for cells_changed in [false, true] {
-                for enabled in [false, true] {
-                    let mut state = super::render_stream::ClientRenderState::new(
-                        crate::protocol::RenderEncoding::SemanticFrame,
-                    );
-                    state.enable_surface_reuse(enabled);
-                    let mut decoder = crate::protocol::surface_reuse::Decoder::default();
-                    let initial = state.prepare_pane_surface(surface.clone()).unwrap();
-                    if enabled {
-                        decoder.decode(initial.message().clone()).unwrap();
-                    }
-                    state.commit_sent_frame(initial);
+                for mode in ["legacy", "reuse", "delta"] {
+                    let mut client = surface_encoding_client(mode, &surface);
                     let mut samples = Vec::new();
                     let mut bytes_per_update = 0;
                     for index in 0..WARMUP_COUNT + SAMPLE_COUNT {
@@ -367,7 +521,7 @@ fn print_surface_reuse_profiles() {
                                 if index % 2 == 0 { "a" } else { "b" }.into();
                         }
                         let started = Instant::now();
-                        let prepared = state.prepare_pane_surface(candidate).unwrap();
+                        let prepared = client.state.prepare_pane_surface(candidate).unwrap();
                         let mut bytes = Vec::new();
                         crate::protocol::write_message(&mut bytes, prepared.message()).unwrap();
                         bytes_per_update = bytes.len();
@@ -376,19 +530,19 @@ fn print_surface_reuse_profiles() {
                             crate::protocol::MAX_FRAME_SIZE,
                         )
                         .unwrap();
-                        black_box(if enabled {
+                        black_box(if let Some(decoder) = client.decoder.as_mut() {
                             decoder.decode(decoded).unwrap()
                         } else {
                             decoded
                         });
-                        state.commit_sent_frame(prepared);
+                        client.state.commit_sent_frame(prepared);
                         if index >= WARMUP_COUNT {
                             samples.push(started.elapsed());
                         }
                     }
                     let stats = summarize(samples);
                     println!(
-                    "  {label:<10}  {count:>5}  {cells_changed:>13}  {enabled:>5}  {:>9}  {bytes_per_update:>16}",
+                    "  {label:<10}  {count:>5}  {cells_changed:>13}  {mode:>7}  {:>9}  {bytes_per_update:>16}",
                     stats.median_us
                 );
                 }
@@ -406,4 +560,5 @@ async fn render_scale_profile() {
     print_snapshot_encoding_profiles("active panes", active_panes);
     print_token_rule_profiles();
     print_surface_reuse_profiles();
+    print_surface_damage_profiles();
 }
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index 669c0097..b0af26d7 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -18,6 +18,8 @@ pub(crate) enum ClientRenderState {
         last_surface: Option<Box<PaneSurfaceFrame>>,
         surface_revision: u64,
         surface_reuse: bool,
+        surface_delta: bool,
+        recompute_pending: bool,
     },
     /// Terminal-ANSI clients keep a terminal diff encoder and sequence number.
     TerminalAnsi {
@@ -34,6 +36,8 @@ impl ClientRenderState {
                 last_surface: None,
                 surface_revision: 0,
                 surface_reuse: false,
+                surface_delta: false,
+                recompute_pending: false,
             },
             RenderEncoding::TerminalAnsi => Self::TerminalAnsi {
                 blit_encoder: BlitEncoder::new(),
@@ -49,6 +53,35 @@ impl ClientRenderState {
         }
     }
 
+    pub(crate) fn enable_surface_delta(&mut self, enabled: bool) {
+        if let Self::Semantic { surface_delta, .. } = self {
+            *surface_delta = enabled;
+        }
+    }
+
+    pub(crate) fn request_recompute(&mut self) {
+        if let Self::Semantic {
+            surface_delta: true,
+            recompute_pending,
+            ..
+        } = self
+        {
+            *recompute_pending = true;
+        } else {
+            self.request_repaint();
+        }
+    }
+
+    pub(crate) fn requires_recompute(&self) -> bool {
+        matches!(
+            self,
+            Self::Semantic {
+                recompute_pending: true,
+                ..
+            }
+        )
+    }
+
     pub(crate) fn reset_baseline(&mut self) {
         match self {
             Self::Semantic { last_surface, .. } => *last_surface = None,
@@ -127,11 +160,14 @@ impl ClientRenderState {
             last_surface,
             surface_revision,
             surface_reuse,
+            surface_delta,
+            recompute_pending,
         } = self
         else {
             return None;
         };
-        if surface.graphics.assets.is_empty()
+        if !*recompute_pending
+            && surface.graphics.assets.is_empty()
             && last_surface.as_deref().is_some_and(|last| {
                 last.projection_revision == surface.projection_revision
                     && last.frame == surface.frame
@@ -145,26 +181,41 @@ impl ClientRenderState {
             return None;
         }
         surface.surface_revision = surface_revision.saturating_add(1);
-        let mut committed_surface = surface.clone();
-        committed_surface.graphics.assets.clear();
-        let reused = (*surface_reuse)
+        let assets = std::mem::take(&mut surface.graphics.assets);
+        let committed_surface = surface.clone();
+        surface.graphics.assets = assets;
+        let mut message = ServerMessage::PaneSurface(surface);
+        let delta = (*surface_delta)
             .then_some(last_surface.as_deref())
             .flatten()
-            .filter(|last| {
-                last.boot_id == surface.boot_id
-                    && last.frame == surface.frame
-                    // Popup cells are not part of the reusable grid; keep their compact codec.
-                    && surface.popup.is_none()
-                    && surface.graphics.assets.is_empty()
-            })
             .and_then(|last| {
-                crate::protocol::surface_reuse::message(last.surface_revision, &mut surface)
-                    .map_err(|error| tracing::warn!(%error, "failed to encode surface reuse"))
+                crate::protocol::surface_delta::message(last, &mut message)
+                    .map_err(|error| tracing::warn!(%error, "failed to encode surface delta"))
                     .ok()
                     .flatten()
             });
+        let reused = if let ServerMessage::PaneSurface(surface) = &mut message {
+            (delta.is_none() && *surface_reuse)
+                .then_some(last_surface.as_deref())
+                .flatten()
+                .filter(|last| {
+                    last.boot_id == surface.boot_id
+                        && last.frame == surface.frame
+                        // Popup cells are not part of the reusable grid; keep their compact codec.
+                        && surface.popup.is_none()
+                        && surface.graphics.assets.is_empty()
+                })
+                .and_then(|last| {
+                    crate::protocol::surface_reuse::message(last.surface_revision, surface)
+                        .map_err(|error| tracing::warn!(%error, "failed to encode surface reuse"))
+                        .ok()
+                        .flatten()
+                })
+        } else {
+            None
+        };
         Some(PreparedRender::Semantic {
-            message: reused.unwrap_or(ServerMessage::PaneSurface(surface)),
+            message: delta.or(reused).unwrap_or(message),
             committed_surface: Box::new(committed_surface),
         })
     }
@@ -181,6 +232,9 @@ impl ClientRenderState {
         else {
             return None;
         };
+        if self.requires_recompute() {
+            return None;
+        }
         let last = last_surface.as_deref()?;
         if last.boot_id != patch.boot_id
             || last.projection_revision != patch.projection_revision
@@ -201,6 +255,7 @@ impl ClientRenderState {
                 Self::Semantic {
                     last_surface,
                     surface_revision,
+                    recompute_pending,
                     ..
                 },
                 PreparedRender::Semantic {
@@ -209,6 +264,7 @@ impl ClientRenderState {
             ) => {
                 *surface_revision = committed_surface.surface_revision;
                 *last_surface = Some(committed_surface);
+                *recompute_pending = false;
             }
             (
                 Self::Semantic {
@@ -518,6 +574,46 @@ mod tests {
     }
 
     #[test]
+    fn surface_delta_recompute_preserves_wire_baseline_but_epoch_reset_drops_it() {
+        for enabled in [false, true] {
+            let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
+            state.enable_surface_delta(enabled);
+            let mut surface = popup_surface("popup");
+            surface.popup = None;
+            surface.frame = FrameData::from_ratatui_buffer(
+                &ratatui::buffer::Buffer::empty(Rect::new(0, 0, 120, 40)),
+                None,
+            );
+            let initial = state.prepare_pane_surface(surface.clone()).unwrap();
+            state.commit_sent_frame(initial);
+            state.request_recompute();
+            assert_eq!(state.last_pane_surface().is_some(), enabled);
+            assert_eq!(state.requires_recompute(), enabled);
+            // A freshness request still emits a new revision when every cell is equal.
+            let fresh = state.prepare_pane_surface(surface.clone()).unwrap();
+            assert_eq!(
+                matches!(fresh.message(), ServerMessage::EndpointControl { kind, .. }
+                if kind == crate::protocol::surface_delta::MESSAGE_KIND),
+                enabled
+            );
+            assert_eq!(
+                state.requires_recompute(),
+                enabled,
+                "prepare must not commit"
+            );
+            state.commit_sent_frame(fresh);
+            assert!(!state.requires_recompute());
+            assert_eq!(state.last_pane_surface().unwrap().surface_revision, 2);
+            state.request_repaint();
+            assert!(state.last_pane_surface().is_none());
+            let recovery = state.prepare_pane_surface(surface).unwrap();
+            assert!(
+                matches!(recovery.message(), ServerMessage::PaneSurface(frame) if frame.surface_revision == 3)
+            );
+        }
+    }
+
+    #[test]
     fn surface_reuse_preserves_projection_and_patch_baselines_without_resending_cells() {
         for enabled in [false, true] {
             let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);