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

6 files changed, 265 insertions, 1 deletions

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