fix: keep remote surface updates incremental (#4209)

refs #3769

refs #3745

6 files changed, 1472 insertions, 12 deletions

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),