fix: avoid resending terminal cells for metadata updates (#4137)

13 files changed, 439 insertions, 7 deletions

diff --git a/src/client/handshake.rs b/src/client/handshake.rs
index 937f92ac..ef55fe22 100644
--- a/src/client/handshake.rs
+++ b/src/client/handshake.rs
@@ -182,6 +182,7 @@ pub(super) fn do_handshake(
             endpoint_keybindings,
             mouse_capture,
             surface_active,
+            surface_reuse: 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 f78150ac..13125c2d 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -535,6 +535,10 @@ async fn run_client_loop(
         } else {
             crate::protocol::MAX_FRAME_SIZE
         };
+        let negotiation = endpoint::EndpointNegotiation::new(
+            handshake.endpoint_methods.unwrap_or_default(),
+            handshake.endpoint_capabilities.unwrap_or_default(),
+        );
         let transport = start_endpoint_transport(
             stream,
             (),
@@ -542,11 +546,8 @@ async fn run_client_loop(
             endpoint::ClientEndpointId::Local,
             1,
             max_frame_size,
+            negotiation.supports_capability(protocol::surface_reuse::CAPABILITY),
         )?;
-        let negotiation = endpoint::EndpointNegotiation::new(
-            handshake.endpoint_methods.unwrap_or_default(),
-            handshake.endpoint_capabilities.unwrap_or_default(),
-        );
         let mut registry = endpoint::EndpointRegistry::new(transport, 1, negotiation);
         if state.shell.is_some() {
             registry.send(&ClientMessage::ClientShellFocus { focused: true });
@@ -1195,6 +1196,8 @@ async fn run_client_loop(
                     ) {
                         continue;
                     }
+                    let surface_reuse =
+                        negotiation.supports_capability(protocol::surface_reuse::CAPABILITY);
                     let agent_view_projection_supported = negotiation.supports_capability(
                         crate::protocol::endpoint::AGENT_VIEW_PROJECTION_CAPABILITY,
                     );
@@ -1226,6 +1229,7 @@ async fn run_client_loop(
                             MAX_GRAPHICS_FRAME_SIZE,
                             endpoint_id,
                             generation,
+                            surface_reuse,
                         );
                     });
                 }
diff --git a/src/client/transport.rs b/src/client/transport.rs
index 2320598b..4396690a 100644
--- a/src/client/transport.rs
+++ b/src/client/transport.rs
@@ -7,6 +7,7 @@ pub(super) fn start_endpoint_transport(
     endpoint_id: endpoint::ClientEndpointId,
     generation: u64,
     max_frame_size: usize,
+    surface_reuse: bool,
 ) -> Result<endpoint::NativeEndpointTransport, ClientError> {
     let reader = stream.try_clone().map_err(ClientError::ConnectionFailed)?;
     let transport = endpoint::NativeEndpointTransport::with_lifetime(stream, lifetime)
@@ -23,6 +24,7 @@ pub(super) fn start_endpoint_transport(
                 max_frame_size,
                 endpoint_id,
                 generation,
+                surface_reuse,
             );
         })
         .map_err(ClientError::ConnectionFailed)?;
@@ -37,6 +39,7 @@ pub(super) fn server_reader_thread(
     max_frame_size: usize,
     endpoint_id: endpoint::ClientEndpointId,
     generation: u64,
+    surface_reuse: bool,
 ) {
     if stream.set_nonblocking(true).is_err() {
         let _ = event_tx.blocking_send(ClientLoopEvent::ServerDisconnected {
@@ -50,12 +53,21 @@ 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;
         }
 
-        match protocol::read_message(&mut stream, max_frame_size) {
+        let message = protocol::read_message(&mut stream, max_frame_size).and_then(|message| {
+            match &mut surface_decoder {
+                Some(decoder) => decoder.decode(message).map_err(|error| {
+                    protocol::FramingError::Io(io::Error::new(io::ErrorKind::InvalidData, error))
+                }),
+                None => Ok(message),
+            }
+        });
+        match message {
             Ok(msg) => {
                 if event_tx
                     .blocking_send(ClientLoopEvent::ServerMessage {
diff --git a/src/protocol/endpoint.rs b/src/protocol/endpoint.rs
index 640e594e..2f7c8338 100644
--- a/src/protocol/endpoint.rs
+++ b/src/protocol/endpoint.rs
@@ -46,6 +46,9 @@ pub struct EndpointClientHello {
     pub mouse_capture: bool,
     #[serde(default = "default_true")]
     pub surface_active: bool,
+    /// Accept the optional cell-retaining surface encoding on this connection.
+    #[serde(default)]
+    pub surface_reuse: bool,
     #[serde(default)]
     pub snapshot_codecs: Vec<String>,
     #[serde(default)]
@@ -139,6 +142,7 @@ impl EndpointServerWelcome {
             blob_codec: BLOB_CODEC_V1.into(),
             methods,
             capabilities: vec![
+                super::surface_reuse::CAPABILITY.into(),
                 SURFACE_INTEREST_CAPABILITY.into(),
                 PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(),
                 HEALTH_CHECK_CAPABILITY.into(),
@@ -181,6 +185,7 @@ mod tests {
             endpoint_keybindings: false,
             mouse_capture: true,
             surface_active: true,
+            surface_reuse: false,
             snapshot_codecs: vec![SNAPSHOT_CODEC_V1.into()],
             surface_codecs: vec![SURFACE_CODEC_V1.into()],
             input_codecs: vec![INPUT_CODEC_V1.into()],
@@ -327,8 +332,10 @@ mod tests {
     fn legacy_hello_defaults_to_an_active_surface() {
         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");
         let decoded: EndpointClientHello = serde_json::from_value(value).unwrap();
         assert!(decoded.surface_active);
+        assert!(!decoded.surface_reuse);
     }
 
     #[test]
@@ -337,6 +344,7 @@ mod tests {
         assert_eq!(
             welcome.capabilities,
             vec![
+                super::super::surface_reuse::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 7ee6e8de..4a8766b6 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_reuse;
 mod wire;
 
 pub use wire::*;
diff --git a/src/protocol/surface_reuse.rs b/src/protocol/surface_reuse.rs
new file mode 100644
index 00000000..bba85824
--- /dev/null
+++ b/src/protocol/surface_reuse.rs
@@ -0,0 +1,127 @@
+//! Optional endpoint encoding that retains unchanged terminal cells across projections.
+
+use super::{CellData, PaneSurfaceFrame, ServerMessage};
+use serde::{Deserialize, Serialize};
+
+pub(crate) const CAPABILITY: &str = "surface_reuse";
+pub(crate) const MESSAGE_KIND: &str = "endpoint.surface-reuse.v1";
+
+#[derive(Serialize, Deserialize)]
+struct SurfaceReuse<S> {
+    base_surface_revision: u64,
+    surface: S,
+}
+
+pub(crate) fn message(
+    base_surface_revision: u64,
+    surface: &mut PaneSurfaceFrame,
+) -> serde_json::Result<Option<ServerMessage>> {
+    let cells = std::mem::take(&mut surface.frame.cells);
+    let data = serde_json::to_string(&SurfaceReuse {
+        base_surface_revision,
+        surface: &*surface,
+    });
+    surface.frame.cells = cells;
+    let message = ServerMessage::EndpointControl {
+        kind: MESSAGE_KIND.into(),
+        data: data?,
+    };
+    // JSON can expand non-cell data (for example escaped hyperlink URLs). A
+    // failed compact encoding must fall back, not strand a newer snapshot.
+    let size = bincode::serde::encode_into_std_write(
+        &message,
+        &mut std::io::sink(),
+        bincode::config::standard(),
+    );
+    match size {
+        Ok(size) => Ok((size <= super::MAX_FRAME_SIZE).then_some(message)),
+        Err(error) => {
+            tracing::warn!(%error, "failed to size surface reuse");
+            Ok(None)
+        }
+    }
+}
+
+#[derive(Default)]
+struct CellBaseline {
+    boot_id: String,
+    projection_revision: u64,
+    surface_revision: u64,
+    width: u16,
+    height: u16,
+    cells: Vec<CellData>,
+}
+
+/// Connection-local decoding happens before activation and presentation filtering, so
+/// switching endpoints cannot discard a baseline needed by the next wire message.
+#[derive(Default)]
+pub(crate) struct Decoder {
+    baseline: Option<CellBaseline>,
+}
+
+impl Decoder {
+    pub(crate) fn decode(&mut self, message: ServerMessage) -> Result<ServerMessage, String> {
+        let message = match message {
+            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}"))?;
+                let Some(base) = &mut self.baseline else {
+                    return Err("surface reuse without a baseline".into());
+                };
+                let mut surface = reuse.surface;
+                if base.boot_id != surface.boot_id
+                    || base.surface_revision != reuse.base_surface_revision
+                    || surface.surface_revision != base.surface_revision.saturating_add(1)
+                    || base.width != surface.frame.width
+                    || base.height != surface.frame.height
+                    || !surface.frame.cells.is_empty()
+                {
+                    return Err("surface reuse does not match its baseline".into());
+                }
+                surface.frame.cells.clone_from(&base.cells);
+                base.projection_revision = surface.projection_revision;
+                base.surface_revision = surface.surface_revision;
+                return Ok(ServerMessage::PaneSurface(surface));
+            }
+            message => message,
+        };
+        match &message {
+            ServerMessage::PaneSurface(surface) => {
+                let base = self.baseline.get_or_insert_with(CellBaseline::default);
+                base.boot_id.clone_from(&surface.boot_id);
+                base.projection_revision = surface.projection_revision;
+                base.surface_revision = surface.surface_revision;
+                base.width = surface.frame.width;
+                base.height = surface.frame.height;
+                base.cells.clone_from(&surface.frame.cells);
+            }
+            ServerMessage::PaneSurfacePatch(patch) => {
+                if let Some(base) = &mut self.baseline {
+                    if patch.boot_id != base.boot_id
+                        || patch.projection_revision != base.projection_revision
+                        || patch.base_surface_revision != base.surface_revision
+                        || patch.surface_revision != base.surface_revision.saturating_add(1)
+                    {
+                        self.baseline = None;
+                    } else {
+                        for row in &patch.rows {
+                            let start =
+                                usize::from(row.y) * usize::from(base.width) + usize::from(row.x);
+                            let end = start.saturating_add(row.cells.len());
+                            if row.y >= base.height
+                                || usize::from(row.x) + row.cells.len() > usize::from(base.width)
+                                || end > base.cells.len()
+                            {
+                                return Err("surface patch exceeds the cell baseline".into());
+                            }
+                            base.cells[start..end].clone_from_slice(&row.cells);
+                        }
+                        base.surface_revision = patch.surface_revision;
+                    }
+                }
+            }
+            _ => {}
+        }
+        Ok(message)
+    }
+}
diff --git a/src/protocol/wire.rs b/src/protocol/wire.rs
index 052c73a3..a29c6ca6 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, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct CellData {
     /// Grapheme cluster displayed in this cell (usually 1–2 chars).
     pub symbol: String,
@@ -715,6 +715,21 @@ pub struct CellData {
     pub hyperlink: Option<u32>,
 }
 
+impl Clone for CellData {
+    fn clone(&self) -> Self {
+        Self {
+            symbol: self.symbol.clone(),
+            ..*self
+        }
+    }
+
+    fn clone_from(&mut self, source: &Self) {
+        let mut symbol = std::mem::take(&mut self.symbol);
+        symbol.clone_from(&source.symbol);
+        *self = Self { symbol, ..*source };
+    }
+}
+
 impl CellData {
     pub(crate) fn from_ratatui_cell(cell: &ratatui::buffer::Cell) -> Self {
         Self {
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 210a840b..aba4ee76 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -398,6 +398,7 @@ pub(crate) enum ServerEvent {
         endpoint_keybindings: bool,
         mouse_capture: bool,
         surface_active: bool,
+        surface_reuse: bool,
         writer: ClientWriter,
     },
     /// A client sent an input message.
@@ -763,6 +764,7 @@ pub(crate) fn handle_client_handshake(
                     hello.endpoint_keybindings,
                     hello.mouse_capture,
                     hello.surface_active,
+                    hello.surface_reuse,
                 )),
             )
         }
@@ -857,6 +859,7 @@ pub(crate) fn handle_client_handshake(
         endpoint_keybindings,
         mouse_capture,
         surface_active,
+        surface_reuse,
     )) = shell_options
     {
         ServerEvent::ClientShellConnected {
@@ -870,6 +873,7 @@ pub(crate) fn handle_client_handshake(
             endpoint_keybindings,
             mouse_capture,
             surface_active,
+            surface_reuse,
             writer,
         }
     } else {
@@ -1444,6 +1448,7 @@ mod tests {
             endpoint_keybindings: true,
             mouse_capture: true,
             surface_active: true,
+            surface_reuse: 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()],
@@ -1959,8 +1964,10 @@ mod tests {
                 endpoint_keybindings,
                 mouse_capture,
                 surface_active,
+                surface_reuse,
                 writer,
             } => {
+                assert!(!surface_reuse);
                 assert_eq!(client_id, 43);
                 assert_eq!((surface_cols, surface_rows), (80, 29));
                 assert_eq!((cell_width_px, cell_height_px), (8, 16));
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 5648ac0a..01d91bbd 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -1968,6 +1968,7 @@ impl HeadlessServer {
                 endpoint_keybindings,
                 mouse_capture,
                 surface_active,
+                surface_reuse,
                 writer,
             } => {
                 if self.handoff_in_progress {
@@ -2013,6 +2014,7 @@ impl HeadlessServer {
                 connection.shell_uses_endpoint_keybindings = endpoint_keybindings;
                 connection.shell_mouse_capture = mouse_capture;
                 connection.shell_surface_active = surface_active;
+                connection.render_state.enable_surface_reuse(surface_reuse);
                 connection.shell_projection_revision = 1;
                 let config_diagnostic = if endpoint_keybindings {
                     self.server_config_diagnostic.as_deref()
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 4850203f..8e731211 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -687,6 +687,7 @@ async fn client_shell_attach_seeds_workspace() {
 
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 6,
             surface_cols: 80,
             surface_rows: 23,
@@ -717,6 +718,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
     let client_id = 41;
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id,
             surface_cols: 80,
             surface_rows: 23,
@@ -833,6 +835,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
     let (writer, control_rx, _render_rx) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 77,
             surface_cols: 80,
             surface_rows: 23,
@@ -941,6 +944,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
     let (writer, control_rx, render_rx) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 7,
             surface_cols: 80,
             surface_rows: 23,
@@ -1108,6 +1112,7 @@ fn connect_test_shell(
     let (writer, control, render) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id,
             surface_cols,
             surface_rows,
@@ -1569,6 +1574,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
     let (local_writer, local_control, _local_render) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 13,
             surface_cols: 80,
             surface_rows: 23,
@@ -1593,6 +1599,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
     let (endpoint_writer, endpoint_control, _endpoint_render) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 14,
             surface_cols: 80,
             surface_rows: 23,
@@ -2495,6 +2502,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
     let (writer, control_rx, render_rx) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 9,
             surface_cols: 80,
             surface_rows: 23,
@@ -2745,6 +2753,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
     let (writer, control_rx, render_rx) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 12,
             surface_cols: 80,
             surface_rows: 23,
diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs
index df06dc03..554b216e 100644
--- a/src/server/headless/tests/surface_interest.rs
+++ b/src/server/headless/tests/surface_interest.rs
@@ -64,6 +64,7 @@ async fn metadata_only_shell_is_isolated_until_surface_activation() {
 
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id,
             surface_cols: 101,
             surface_rows: 37,
@@ -280,6 +281,7 @@ async fn background_surface_activation_preserves_focused_viewer_geometry() {
     let (writer, background_control, _) = test_client_writer();
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: 8,
             surface_cols: 100,
             surface_rows: 35,
@@ -391,6 +393,7 @@ async fn presentation_sync_epoch_replays_modes_and_title() {
     let client_id = 63;
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -505,6 +508,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     let source_client_id = 78;
     assert!(
         source_server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: false,
             client_id: source_client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -530,6 +534,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     let target_client_id = 79;
     assert!(
         target_server.handle_server_event(ServerEvent::ClientShellConnected {
+            surface_reuse: 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 7503f06d..82f7ba91 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -312,6 +312,91 @@ fn print_token_rule_profiles() {
     }
 }
 
+fn print_surface_reuse_profiles() {
+    println!("surface encode + decode at {COLS}x{ROWS}");
+    println!("  layout       panes  cells_changed  reuse  median_us  bytes_per_update");
+    for (label, build) in [
+        ("background", workspaces as fn(usize) -> Vec<Workspace>),
+        ("active", active_panes),
+    ] {
+        for count in [1, 15] {
+            let mut pipeline = RenderPipeline::new(build(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 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,
+            };
+            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);
+                    let mut samples = Vec::new();
+                    let mut bytes_per_update = 0;
+                    for index in 0..WARMUP_COUNT + SAMPLE_COUNT {
+                        let mut candidate = surface.clone();
+                        candidate.projection_revision = index as u64 + 2;
+                        if cells_changed {
+                            candidate.frame.cells[0].symbol =
+                                if index % 2 == 0 { "a" } else { "b" }.into();
+                        }
+                        let started = Instant::now();
+                        let prepared = 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();
+                        let decoded = crate::protocol::read_message(
+                            &mut bytes.as_slice(),
+                            crate::protocol::MAX_FRAME_SIZE,
+                        )
+                        .unwrap();
+                        black_box(if enabled {
+                            decoder.decode(decoded).unwrap()
+                        } else {
+                            decoded
+                        });
+                        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}",
+                    stats.median_us
+                );
+                }
+            }
+        }
+    }
+}
+
 #[tokio::test(flavor = "current_thread")]
 #[ignore = "manual client-rendered pipeline scaling profile"]
 async fn render_scale_profile() {
@@ -320,4 +405,5 @@ async fn render_scale_profile() {
     print_profiles("active panes (one workspace)", active_panes);
     print_snapshot_encoding_profiles("active panes", active_panes);
     print_token_rule_profiles();
+    print_surface_reuse_profiles();
 }
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index 349454f3..669c0097 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -17,6 +17,7 @@ pub(crate) enum ClientRenderState {
     Semantic {
         last_surface: Option<Box<PaneSurfaceFrame>>,
         surface_revision: u64,
+        surface_reuse: bool,
     },
     /// Terminal-ANSI clients keep a terminal diff encoder and sequence number.
     TerminalAnsi {
@@ -32,6 +33,7 @@ impl ClientRenderState {
             RenderEncoding::SemanticFrame => Self::Semantic {
                 last_surface: None,
                 surface_revision: 0,
+                surface_reuse: false,
             },
             RenderEncoding::TerminalAnsi => Self::TerminalAnsi {
                 blit_encoder: BlitEncoder::new(),
@@ -41,6 +43,12 @@ impl ClientRenderState {
         }
     }
 
+    pub(crate) fn enable_surface_reuse(&mut self, enabled: bool) {
+        if let Self::Semantic { surface_reuse, .. } = self {
+            *surface_reuse = enabled;
+        }
+    }
+
     pub(crate) fn reset_baseline(&mut self) {
         match self {
             Self::Semantic { last_surface, .. } => *last_surface = None,
@@ -118,6 +126,7 @@ impl ClientRenderState {
         let Self::Semantic {
             last_surface,
             surface_revision,
+            surface_reuse,
         } = self
         else {
             return None;
@@ -138,8 +147,24 @@ impl ClientRenderState {
         surface.surface_revision = surface_revision.saturating_add(1);
         let mut committed_surface = surface.clone();
         committed_surface.graphics.assets.clear();
+        let reused = (*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, &mut surface)
+                    .map_err(|error| tracing::warn!(%error, "failed to encode surface reuse"))
+                    .ok()
+                    .flatten()
+            });
         Some(PreparedRender::Semantic {
-            message: ServerMessage::PaneSurface(surface),
+            message: reused.unwrap_or(ServerMessage::PaneSurface(surface)),
             committed_surface: Box::new(committed_surface),
         })
     }
@@ -151,6 +176,7 @@ impl ClientRenderState {
         let Self::Semantic {
             last_surface,
             surface_revision,
+            ..
         } = self
         else {
             return None;
@@ -175,6 +201,7 @@ impl ClientRenderState {
                 Self::Semantic {
                     last_surface,
                     surface_revision,
+                    ..
                 },
                 PreparedRender::Semantic {
                     committed_surface, ..
@@ -187,6 +214,7 @@ impl ClientRenderState {
                 Self::Semantic {
                     last_surface,
                     surface_revision,
+                    ..
                 },
                 PreparedRender::SemanticPatch {
                     message: ServerMessage::PaneSurfacePatch(patch),
@@ -490,6 +518,133 @@ mod tests {
     }
 
     #[test]
+    fn surface_reuse_preserves_projection_and_patch_baselines_without_resending_cells() {
+        for enabled in [false, true] {
+            let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
+            state.enable_surface_reuse(enabled);
+            let mut decoder = crate::protocol::surface_reuse::Decoder::default();
+            let mut surface = popup_surface("popup");
+            surface.popup = None;
+            let buffer = ratatui::buffer::Buffer::empty(Rect::new(0, 0, 240, 100));
+            surface.frame = FrameData::from_ratatui_buffer(&buffer, None);
+            let initial = state.prepare_pane_surface(surface.clone()).unwrap();
+            decoder.decode(initial.message().clone()).unwrap();
+            state.commit_sent_frame(initial);
+
+            surface.projection_revision += 1;
+            let update = state.prepare_pane_surface(surface.clone()).unwrap();
+            let mut bytes = Vec::new();
+            crate::protocol::write_message(&mut bytes, update.message()).unwrap();
+            if enabled {
+                assert!(
+                    matches!(update.message(), ServerMessage::EndpointControl { kind, .. }
+                    if kind == crate::protocol::surface_reuse::MESSAGE_KIND)
+                );
+                assert!(
+                    bytes.len() < 2000,
+                    "metadata update was {} bytes",
+                    bytes.len()
+                );
+            } else {
+                assert!(matches!(update.message(), ServerMessage::PaneSurface(_)));
+                assert!(bytes.len() > 100_000);
+            }
+            let ServerMessage::PaneSurface(decoded) =
+                decoder.decode(update.message().clone()).unwrap()
+            else {
+                panic!("decoded full surface");
+            };
+            assert_eq!(decoded.frame, surface.frame);
+            assert_eq!(decoded.projection_revision, surface.projection_revision);
+            assert_eq!(decoded.surface_revision, 2);
+            state.commit_sent_frame(update);
+
+            let mut changed_cell = surface.frame.cells[0].clone();
+            changed_cell.symbol = "x".into();
+            let patch = state
+                .prepare_pane_surface_patch(PaneSurfacePatch {
+                    boot_id: surface.boot_id.clone(),
+                    projection_revision: surface.projection_revision,
+                    base_surface_revision: 2,
+                    surface_revision: 0,
+                    rows: vec![crate::protocol::PaneSurfacePatchRow {
+                        x: 0,
+                        y: 0,
+                        cells: vec![changed_cell.clone()],
+                    }],
+                    panes: Vec::new(),
+                    cursor: None,
+                })
+                .unwrap();
+            decoder.decode(patch.message().clone()).unwrap();
+            state.commit_sent_frame(patch);
+            surface.frame.cells[0] = changed_cell;
+            surface.projection_revision += 1;
+            let update = state.prepare_pane_surface(surface.clone()).unwrap();
+            let ServerMessage::PaneSurface(decoded) =
+                decoder.decode(update.message().clone()).unwrap()
+            else {
+                panic!("decoded surface after patch");
+            };
+            assert_eq!(decoded.frame, surface.frame);
+            assert_eq!(decoded.surface_revision, 4);
+            state.commit_sent_frame(update);
+
+            // A changed border or terminal cell must still reach the client.
+            surface.frame.cells[0].symbol = "y".into();
+            let changed = state.prepare_pane_surface(surface.clone()).unwrap();
+            assert!(matches!(changed.message(), ServerMessage::PaneSurface(_)));
+            let ServerMessage::PaneSurface(decoded) =
+                decoder.decode(changed.message().clone()).unwrap()
+            else {
+                panic!("changed full surface");
+            };
+            assert_eq!(decoded.frame, surface.frame);
+            state.commit_sent_frame(changed);
+
+            state.request_repaint();
+            assert!(matches!(
+                state.prepare_pane_surface(surface).unwrap().message(),
+                ServerMessage::PaneSurface(_)
+            ));
+        }
+    }
+
+    #[test]
+    fn surface_reuse_keeps_popup_cells_on_the_binary_codec() {
+        let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
+        state.enable_surface_reuse(true);
+        let mut surface = popup_surface("popup");
+        let buffer = ratatui::buffer::Buffer::empty(Rect::new(0, 0, 400, 100));
+        surface.popup.as_mut().unwrap().frame = FrameData::from_ratatui_buffer(&buffer, None);
+        let initial = state.prepare_pane_surface(surface.clone()).unwrap();
+        state.commit_sent_frame(initial);
+        surface.projection_revision += 1;
+        let update = state.prepare_pane_surface(surface).unwrap();
+        assert!(matches!(update.message(), ServerMessage::PaneSurface(_)));
+        let mut bytes = Vec::new();
+        crate::protocol::write_message(&mut bytes, update.message()).unwrap();
+        assert!(bytes.len() < crate::protocol::MAX_FRAME_SIZE);
+    }
+
+    #[test]
+    fn surface_reuse_falls_back_when_json_metadata_exceeds_the_frame_limit() {
+        let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
+        state.enable_surface_reuse(true);
+        let mut surface = popup_surface("popup");
+        surface.popup = None;
+        surface.frame.hyperlinks = vec!["\"".repeat(crate::protocol::MAX_FRAME_SIZE / 2)];
+        let initial = state.prepare_pane_surface(surface.clone()).unwrap();
+        state.commit_sent_frame(initial);
+        surface.projection_revision += 1;
+        let update = state.prepare_pane_surface(surface).unwrap();
+        assert!(matches!(update.message(), ServerMessage::PaneSurface(_)));
+        let mut bytes = Vec::new();
+        crate::protocol::write_message(&mut bytes, update.message()).unwrap();
+        assert!(bytes.len() < crate::protocol::MAX_FRAME_SIZE);
+    }
+
+    #[test]
     fn popup_only_surface_changes_are_not_deduplicated() {
         let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
         let prepared = state