fix: keep remote surface updates incremental (#4209)

refs #3769

refs #3745

10 files changed, 545 insertions, 34 deletions

diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index aba4ee76..0d297fb9 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -399,6 +399,7 @@ pub(crate) enum ServerEvent {
         mouse_capture: bool,
         surface_active: bool,
         surface_reuse: bool,
+        surface_delta: bool,
         writer: ClientWriter,
     },
     /// A client sent an input message.
@@ -765,6 +766,7 @@ pub(crate) fn handle_client_handshake(
                     hello.mouse_capture,
                     hello.surface_active,
                     hello.surface_reuse,
+                    hello.surface_delta,
                 )),
             )
         }
@@ -860,6 +862,7 @@ pub(crate) fn handle_client_handshake(
         mouse_capture,
         surface_active,
         surface_reuse,
+        surface_delta,
     )) = shell_options
     {
         ServerEvent::ClientShellConnected {
@@ -874,6 +877,7 @@ pub(crate) fn handle_client_handshake(
             mouse_capture,
             surface_active,
             surface_reuse,
+            surface_delta,
             writer,
         }
     } else {
@@ -1449,6 +1453,7 @@ mod tests {
             mouse_capture: true,
             surface_active: true,
             surface_reuse: false,
+            surface_delta: false,
             snapshot_codecs: vec![crate::protocol::endpoint::SNAPSHOT_CODEC_V1.into()],
             surface_codecs: vec![crate::protocol::endpoint::SURFACE_CODEC_V1.into()],
             input_codecs: vec![crate::protocol::endpoint::INPUT_CODEC_V1.into()],
@@ -1965,9 +1970,11 @@ mod tests {
                 mouse_capture,
                 surface_active,
                 surface_reuse,
+                surface_delta,
                 writer,
             } => {
                 assert!(!surface_reuse);
+                assert!(!surface_delta);
                 assert_eq!(client_id, 43);
                 assert_eq!((surface_cols, surface_rows), (80, 29));
                 assert_eq!((cell_width_px, cell_height_px), (8, 16));
diff --git a/src/server/clients.rs b/src/server/clients.rs
index c2b419f4..aeda3822 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -261,6 +261,10 @@ impl ClientConnection {
         self.render_state.request_repaint();
     }
 
+    pub(crate) fn request_recompute(&mut self) {
+        self.render_state.request_recompute();
+    }
+
     pub(crate) fn track_shell_input(
         &mut self,
         target: ClientShellInputTarget,
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 01d91bbd..f07c0c37 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -779,7 +779,7 @@ impl HeadlessServer {
         // rendering semantics. Force one fresh frame to every remaining client
         // even if the next rendered buffer compares equal to its cached frame.
         for client in self.clients.values_mut() {
-            client.request_repaint();
+            client.request_recompute();
         }
         if !start_pending_agent_resumes {
             self.app.pending_agent_resume_deadline = None;
@@ -792,7 +792,7 @@ impl HeadlessServer {
             .start_pending_agent_resumes(self.app.pending_agent_resume_due(now))
         {
             for client in self.clients.values_mut() {
-                client.request_repaint();
+                client.request_recompute();
             }
         }
     }
@@ -1969,6 +1969,7 @@ impl HeadlessServer {
                 mouse_capture,
                 surface_active,
                 surface_reuse,
+                surface_delta,
                 writer,
             } => {
                 if self.handoff_in_progress {
@@ -2015,6 +2016,7 @@ impl HeadlessServer {
                 connection.shell_mouse_capture = mouse_capture;
                 connection.shell_surface_active = surface_active;
                 connection.render_state.enable_surface_reuse(surface_reuse);
+                connection.render_state.enable_surface_delta(surface_delta);
                 connection.shell_projection_revision = 1;
                 let config_diagnostic = if endpoint_keybindings {
                     self.server_config_diagnostic.as_deref()
diff --git a/src/server/headless/client_views.rs b/src/server/headless/client_views.rs
index d6030f74..4e0bd2e2 100644
--- a/src/server/headless/client_views.rs
+++ b/src/server/headless/client_views.rs
@@ -516,7 +516,7 @@ impl HeadlessServer {
 
     fn finish_shell_tab_geometry_change(&mut self, start_pending_agent_resumes: bool) {
         for client in self.clients.values_mut() {
-            client.request_repaint();
+            client.request_recompute();
         }
         if !start_pending_agent_resumes {
             self.app.pending_agent_resume_deadline = None;
@@ -529,7 +529,7 @@ impl HeadlessServer {
             .start_pending_agent_resumes(self.app.pending_agent_resume_due(now))
         {
             for client in self.clients.values_mut() {
-                client.request_repaint();
+                client.request_recompute();
             }
         }
     }
diff --git a/src/server/headless/retained_surface.rs b/src/server/headless/retained_surface.rs
index d0fb9bba..2a637eac 100644
--- a/src/server/headless/retained_surface.rs
+++ b/src/server/headless/retained_surface.rs
@@ -265,6 +265,9 @@ impl HeadlessServer {
                 crate::render_prof::event("retained_surface.recipient_deferred");
                 continue;
             }
+            if client.render_state.requires_recompute() {
+                fallback!("recompute_pending");
+            }
             let Some(surface) = client.render_state.last_pane_surface() else {
                 fallback!("no_baseline");
             };
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 093b5625..60f89d63 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -4,6 +4,8 @@ use super::*;
 mod pane_graphics_tests;
 #[path = "pane_move.rs"]
 mod pane_move_tests;
+#[path = "surface_delta.rs"]
+mod surface_delta_tests;
 #[path = "surface_interest.rs"]
 mod surface_interest_tests;
 
@@ -690,6 +692,7 @@ async fn client_shell_attach_seeds_workspace() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 6,
             surface_cols: 80,
             surface_rows: 23,
@@ -721,6 +724,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols: 80,
             surface_rows: 23,
@@ -838,6 +842,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 77,
             surface_cols: 80,
             surface_rows: 23,
@@ -947,6 +952,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 7,
             surface_cols: 80,
             surface_rows: 23,
@@ -1115,6 +1121,7 @@ fn connect_test_shell(
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols,
             surface_rows,
@@ -1577,6 +1584,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 13,
             surface_cols: 80,
             surface_rows: 23,
@@ -1602,6 +1610,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 14,
             surface_cols: 80,
             surface_rows: 23,
@@ -2505,6 +2514,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 9,
             surface_cols: 80,
             surface_rows: 23,
@@ -2756,6 +2766,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 12,
             surface_cols: 80,
             surface_rows: 23,
diff --git a/src/server/headless/tests/surface_delta.rs b/src/server/headless/tests/surface_delta.rs
new file mode 100644
index 00000000..f010c1e3
--- /dev/null
+++ b/src/server/headless/tests/surface_delta.rs
@@ -0,0 +1,228 @@
+use super::*;
+
+fn receive_message(receiver: &std::sync::mpsc::Receiver<Vec<u8>>) -> (Vec<u8>, ServerMessage) {
+    let bytes = receiver
+        .recv_timeout(Duration::from_secs(1))
+        .expect("render message");
+    let message = read_server_message(bytes.clone());
+    (bytes, message)
+}
+
+fn decode_surface_message(
+    decoder: &mut protocol::surface_reuse::Decoder,
+    message: ServerMessage,
+) -> crate::protocol::PaneSurfaceFrame {
+    match decoder.decode(message).expect("decode surface message") {
+        ServerMessage::PaneSurface(surface) => surface,
+        other => panic!("expected decoded pane surface, got {other:?}"),
+    }
+}
+
+fn without_asset_payload(
+    mut surface: crate::protocol::PaneSurfaceFrame,
+) -> crate::protocol::PaneSurfaceFrame {
+    surface.graphics.assets.clear();
+    surface
+}
+
+fn enable_delta(server: &mut HeadlessServer, client_id: u64) {
+    server
+        .clients
+        .get_mut(&client_id)
+        .expect("delta client")
+        .render_state
+        .enable_surface_delta(true);
+}
+
+fn fill_render_lane(server: &HeadlessServer, client_id: u64) {
+    let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
+        .expect("dummy frame");
+    server.clients[&client_id]
+        .writer
+        .as_ref()
+        .expect("client writer")
+        .test_fill_render(queued);
+}
+
+#[tokio::test]
+async fn surface_delta_reconstructs_metadata_text_and_hyperlinks() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(b"initial text");
+    enable_delta(&mut server, 1);
+    server.render_and_stream();
+    let (initial_bytes, initial_message) = receive_message(&render_rx);
+    let mut decoder = protocol::surface_reuse::Decoder::new(true);
+    let initial = decode_surface_message(&mut decoder, initial_message);
+    assert_eq!(
+        without_asset_payload(initial.clone()),
+        without_asset_payload(
+            server.clients[&1]
+                .render_state
+                .last_pane_surface()
+                .expect("initial baseline")
+                .clone()
+        )
+    );
+
+    let initial_projection_revision = initial.projection_revision;
+    server.app.state.workspaces[0].custom_name = Some("renamed workspace".into());
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\rupdated text \x1b]8;;https://example.test/path\x1b\\linked\x1b]8;;\x1b\\",
+    );
+    server.clients.get_mut(&1).unwrap().request_recompute();
+    assert!(!server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    server.render_and_stream();
+    let (delta_bytes, delta_message) = receive_message(&render_rx);
+    assert!(delta_bytes.len() < initial_bytes.len());
+    assert!(matches!(
+        &delta_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let decoded = decode_surface_message(&mut decoder, delta_message);
+    let committed = server.clients[&1]
+        .render_state
+        .last_pane_surface()
+        .expect("updated baseline")
+        .clone();
+    assert_eq!(
+        without_asset_payload(decoded.clone()),
+        without_asset_payload(committed)
+    );
+    assert!(decoded.projection_revision > initial_projection_revision);
+    assert!(frame_text(&decoded.frame).contains("updated text"));
+    assert_eq!(decoded.frame.hyperlinks, vec!["https://example.test/path"]);
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn surface_delta_updates_global_popup_without_affecting_legacy_peer() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("popup-delta");
+    let pane_id = workspace.focused_pane_id().expect("focused pane");
+    workspace.insert_test_runtime(
+        pane_id,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 23, b"base"),
+    );
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+    server.app.state.mode = crate::app::Mode::Terminal;
+    let popup_runtime = crate::terminal::TerminalRuntime::test_with_screen_bytes(40, 12, b"POPUP");
+    let (_, popup_terminal_id) = server.app.install_test_popup_runtime(popup_runtime);
+
+    let (_delta_control, delta_render) = connect_matching_test_shell(&mut server, 1);
+    let (_legacy_control, legacy_render) = connect_matching_test_shell(&mut server, 2);
+    enable_delta(&mut server, 1);
+    server.render_and_stream();
+    let (initial_bytes, initial_message) = receive_message(&delta_render);
+    let (_, _legacy_initial) = receive_message(&legacy_render);
+    let mut decoder = protocol::surface_reuse::Decoder::new(true);
+    let _ = decode_surface_message(&mut decoder, initial_message);
+
+    server
+        .app
+        .terminal_runtimes
+        .get(&popup_terminal_id)
+        .expect("popup runtime")
+        .test_process_pty_bytes(b"\rPOPUP updated");
+    server.clients.get_mut(&1).unwrap().request_recompute();
+    server.render_and_stream();
+
+    let (delta_bytes, delta_message) = receive_message(&delta_render);
+    assert!(delta_bytes.len() * 4 < initial_bytes.len());
+    assert!(matches!(
+        &delta_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let decoded = decode_surface_message(&mut decoder, delta_message);
+    assert!(decoded
+        .popup
+        .as_ref()
+        .is_some_and(|popup| frame_text(&popup.frame).contains("POPUP updated")));
+
+    let (_, legacy_message) = receive_message(&legacy_render);
+    assert!(matches!(legacy_message, ServerMessage::PaneSurface(_)));
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn surface_delta_preserves_graphics_baseline_through_queue_recovery() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(b"text before image");
+    server.clients.get_mut(&1).unwrap().cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    enable_delta(&mut server, 1);
+    server.render_and_stream();
+    let (initial_bytes, initial_message) = receive_message(&render_rx);
+    let mut decoder = protocol::surface_reuse::Decoder::new(true);
+    let _ = decode_surface_message(&mut decoder, initial_message);
+
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
+    );
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let (image_bytes, image_message) = receive_message(&render_rx);
+    assert!(image_bytes.len() * 4 < initial_bytes.len());
+    assert!(matches!(
+        &image_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let image = decode_surface_message(&mut decoder, image_message);
+    assert!(!image.graphics.placements.is_empty());
+
+    write_shared_test_pane(&mut server, pane_id, b"\rupdated text");
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let (text_bytes, text_message) = receive_message(&render_rx);
+    assert!(text_bytes.len() * 4 < initial_bytes.len());
+    assert!(matches!(
+        &text_message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    ));
+    let text_update = decode_surface_message(&mut decoder, text_message);
+    assert_eq!(text_update.graphics.placements, image.graphics.placements);
+    assert!(text_update.graphics.assets.is_empty());
+
+    let baseline_before_queue = server.clients[&1]
+        .render_state
+        .last_pane_surface()
+        .expect("graphics baseline")
+        .clone();
+    fill_render_lane(&server, 1);
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\x1b_Ga=T,f=32,t=d,i=8,p=3,s=1,v=1,c=1,r=1,q=2;AP8A/w==\x1b\\",
+    );
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    assert_eq!(
+        without_asset_payload(
+            server.clients[&1]
+                .render_state
+                .last_pane_surface()
+                .expect("baseline after full queue")
+                .clone()
+        ),
+        without_asset_payload(baseline_before_queue)
+    );
+    let _ = render_rx
+        .recv_timeout(Duration::from_secs(1))
+        .expect("drain full render lane");
+
+    server.render_and_stream();
+    let (_, recovered_message) = receive_message(&render_rx);
+    let recovered = decode_surface_message(&mut decoder, recovered_message);
+    assert!(!recovered.graphics.placements.is_empty());
+    assert_eq!(recovered.graphics.assets.len(), 1);
+    assert_eq!(recovered.graphics.assets[0].data, [0, 255, 0, 255]);
+    shutdown_test_runtimes(&mut server);
+}
diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs
index 554b216e..7c8b33f8 100644
--- a/src/server/headless/tests/surface_interest.rs
+++ b/src/server/headless/tests/surface_interest.rs
@@ -65,6 +65,7 @@ async fn metadata_only_shell_is_isolated_until_surface_activation() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols: 101,
             surface_rows: 37,
@@ -282,6 +283,7 @@ async fn background_surface_activation_preserves_focused_viewer_geometry() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: 8,
             surface_cols: 100,
             surface_rows: 35,
@@ -394,6 +396,7 @@ async fn presentation_sync_epoch_replays_modes_and_title() {
     assert!(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -509,6 +512,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     assert!(
         source_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: source_client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -535,6 +539,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     assert!(
         target_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
+            surface_delta: false,
             client_id: target_client_id,
             surface_cols: 80,
             surface_rows: 24,
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index 82f7ba91..5c3935de 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -312,9 +312,172 @@ fn print_token_rule_profiles() {
     }
 }
 
+#[derive(Clone, Copy)]
+enum SurfaceDamagePattern {
+    Dense,
+    Checkerboard,
+}
+
+impl SurfaceDamagePattern {
+    fn name(self) -> &'static str {
+        match self {
+            Self::Dense => "dense",
+            Self::Checkerboard => "checkerboard",
+        }
+    }
+
+    fn apply(self, surface: &mut PaneSurfaceFrame, iteration: usize) {
+        for (index, cell) in surface.frame.cells.iter_mut().enumerate() {
+            cell.symbol = if matches!(self, Self::Checkerboard) && !index.is_multiple_of(2) {
+                "c"
+            } else if iteration.is_multiple_of(2) {
+                "a"
+            } else {
+                "b"
+            }
+            .into();
+        }
+    }
+}
+
+struct SurfaceEncodingClient {
+    state: super::render_stream::ClientRenderState,
+    decoder: Option<crate::protocol::surface_reuse::Decoder>,
+}
+
+fn surface_encoding_client(mode: &str, surface: &PaneSurfaceFrame) -> SurfaceEncodingClient {
+    let reuse = mode == "reuse";
+    let delta = mode == "delta";
+    let mut client = SurfaceEncodingClient {
+        state: super::render_stream::ClientRenderState::new(
+            crate::protocol::RenderEncoding::SemanticFrame,
+        ),
+        decoder: (reuse || delta).then(|| crate::protocol::surface_reuse::Decoder::new(delta)),
+    };
+    client.state.enable_surface_reuse(reuse);
+    client.state.enable_surface_delta(delta);
+    let initial = client
+        .state
+        .prepare_pane_surface(surface.clone())
+        .expect("initial surface should be prepared");
+    if let Some(decoder) = client.decoder.as_mut() {
+        decoder.decode(initial.message().clone()).unwrap();
+    }
+    client.state.commit_sent_frame(initial);
+    client
+}
+
+fn profile_surface_damage(
+    surface: &PaneSurfaceFrame,
+    pattern: SurfaceDamagePattern,
+    mode: &str,
+    client_count: usize,
+) {
+    let mut clients = (0..client_count)
+        .map(|_| surface_encoding_client(mode, surface))
+        .collect::<Vec<_>>();
+    let mut samples = Vec::with_capacity(SAMPLE_COUNT);
+    let mut bytes_per_update = 0usize;
+    for iteration in 0..WARMUP_COUNT + SAMPLE_COUNT {
+        let mut candidate = surface.clone();
+        candidate.projection_revision = iteration as u64 + 2;
+        pattern.apply(&mut candidate, iteration);
+        if iteration > 0 && matches!(pattern, SurfaceDamagePattern::Checkerboard) {
+            let previous = clients[0].state.last_pane_surface().unwrap();
+            let changed = previous
+                .frame
+                .cells
+                .iter()
+                .zip(&candidate.frame.cells)
+                .filter(|(old, new)| old != new)
+                .count();
+            assert_eq!(changed, candidate.frame.cells.len().div_ceil(2));
+        }
+        let started = Instant::now();
+        bytes_per_update = 0;
+        for client in &mut clients {
+            let prepared = client
+                .state
+                .prepare_pane_surface(candidate.clone())
+                .expect("damage should produce a surface update");
+            let mut bytes = Vec::new();
+            crate::protocol::write_message(&mut bytes, prepared.message()).unwrap();
+            bytes_per_update += bytes.len();
+            let decoded = crate::protocol::read_message(
+                &mut bytes.as_slice(),
+                crate::protocol::MAX_GRAPHICS_FRAME_SIZE,
+            )
+            .unwrap();
+            black_box(if let Some(decoder) = client.decoder.as_mut() {
+                decoder.decode(decoded).unwrap()
+            } else {
+                decoded
+            });
+            client.state.commit_sent_frame(prepared);
+        }
+        if iteration >= WARMUP_COUNT {
+            samples.push(started.elapsed());
+        }
+    }
+    let stats = summarize(samples);
+    println!(
+        "  {:<11}  {client_count:>7}  {mode:>7}  {:>9}  {:>12}",
+        pattern.name(),
+        stats.median_us,
+        bytes_per_update
+    );
+}
+
+fn print_surface_damage_profiles() {
+    println!("surface dense/checkerboard encode + decode at {COLS}x{ROWS}");
+    println!("  pattern       clients     mode  median_us  bytes/update");
+    for pane_count in [1, 15] {
+        let mut pipeline = RenderPipeline::new(active_panes(pane_count));
+        pipeline.render_once();
+        let rendered = super::client_shell::render_pane_surface(
+            &mut pipeline.app,
+            Some(crate::ui::TabSurfaceTarget {
+                workspace_index: 0,
+                tab_index: 0,
+            }),
+            Rect::new(0, 0, COLS, ROWS),
+            true,
+            false,
+            HostCellSize::default(),
+            &pipeline.graphics_delivery,
+            1,
+        );
+        let mut surface = PaneSurfaceFrame {
+            boot_id: "bench-boot".into(),
+            projection_revision: 1,
+            surface_revision: 0,
+            frame: rendered.frame,
+            panes: rendered.panes,
+            splits: rendered.splits,
+            popup: rendered.popup,
+            graphics: rendered.graphics,
+        };
+        // Normalize the baseline so dense and checkerboard iterations have stable damage.
+        for cell in &mut surface.frame.cells {
+            cell.symbol = "a".into();
+        }
+        println!("  active panes={pane_count}");
+        for pattern in [
+            SurfaceDamagePattern::Dense,
+            SurfaceDamagePattern::Checkerboard,
+        ] {
+            for client_count in [1, 15] {
+                for mode in ["legacy", "reuse", "delta"] {
+                    profile_surface_damage(&surface, pattern, mode, client_count);
+                }
+            }
+        }
+    }
+}
+
 fn print_surface_reuse_profiles() {
     println!("surface encode + decode at {COLS}x{ROWS}");
-    println!("  layout       panes  cells_changed  reuse  median_us  bytes_per_update");
+    println!("  layout       panes  cells_changed  mode     median_us  bytes_per_update");
     for (label, build) in [
         ("background", workspaces as fn(usize) -> Vec<Workspace>),
         ("active", active_panes),
@@ -346,17 +509,8 @@ fn print_surface_reuse_profiles() {
                 graphics: rendered.graphics,
             };
             for cells_changed in [false, true] {
-                for enabled in [false, true] {
-                    let mut state = super::render_stream::ClientRenderState::new(
-                        crate::protocol::RenderEncoding::SemanticFrame,
-                    );
-                    state.enable_surface_reuse(enabled);
-                    let mut decoder = crate::protocol::surface_reuse::Decoder::default();
-                    let initial = state.prepare_pane_surface(surface.clone()).unwrap();
-                    if enabled {
-                        decoder.decode(initial.message().clone()).unwrap();
-                    }
-                    state.commit_sent_frame(initial);
+                for mode in ["legacy", "reuse", "delta"] {
+                    let mut client = surface_encoding_client(mode, &surface);
                     let mut samples = Vec::new();
                     let mut bytes_per_update = 0;
                     for index in 0..WARMUP_COUNT + SAMPLE_COUNT {
@@ -367,7 +521,7 @@ fn print_surface_reuse_profiles() {
                                 if index % 2 == 0 { "a" } else { "b" }.into();
                         }
                         let started = Instant::now();
-                        let prepared = state.prepare_pane_surface(candidate).unwrap();
+                        let prepared = client.state.prepare_pane_surface(candidate).unwrap();
                         let mut bytes = Vec::new();
                         crate::protocol::write_message(&mut bytes, prepared.message()).unwrap();
                         bytes_per_update = bytes.len();
@@ -376,19 +530,19 @@ fn print_surface_reuse_profiles() {
                             crate::protocol::MAX_FRAME_SIZE,
                         )
                         .unwrap();
-                        black_box(if enabled {
+                        black_box(if let Some(decoder) = client.decoder.as_mut() {
                             decoder.decode(decoded).unwrap()
                         } else {
                             decoded
                         });
-                        state.commit_sent_frame(prepared);
+                        client.state.commit_sent_frame(prepared);
                         if index >= WARMUP_COUNT {
                             samples.push(started.elapsed());
                         }
                     }
                     let stats = summarize(samples);
                     println!(
-                    "  {label:<10}  {count:>5}  {cells_changed:>13}  {enabled:>5}  {:>9}  {bytes_per_update:>16}",
+                    "  {label:<10}  {count:>5}  {cells_changed:>13}  {mode:>7}  {:>9}  {bytes_per_update:>16}",
                     stats.median_us
                 );
                 }
@@ -406,4 +560,5 @@ async fn render_scale_profile() {
     print_snapshot_encoding_profiles("active panes", active_panes);
     print_token_rule_profiles();
     print_surface_reuse_profiles();
+    print_surface_damage_profiles();
 }
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index 669c0097..b0af26d7 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -18,6 +18,8 @@ pub(crate) enum ClientRenderState {
         last_surface: Option<Box<PaneSurfaceFrame>>,
         surface_revision: u64,
         surface_reuse: bool,
+        surface_delta: bool,
+        recompute_pending: bool,
     },
     /// Terminal-ANSI clients keep a terminal diff encoder and sequence number.
     TerminalAnsi {
@@ -34,6 +36,8 @@ impl ClientRenderState {
                 last_surface: None,
                 surface_revision: 0,
                 surface_reuse: false,
+                surface_delta: false,
+                recompute_pending: false,
             },
             RenderEncoding::TerminalAnsi => Self::TerminalAnsi {
                 blit_encoder: BlitEncoder::new(),
@@ -49,6 +53,35 @@ impl ClientRenderState {
         }
     }
 
+    pub(crate) fn enable_surface_delta(&mut self, enabled: bool) {
+        if let Self::Semantic { surface_delta, .. } = self {
+            *surface_delta = enabled;
+        }
+    }
+
+    pub(crate) fn request_recompute(&mut self) {
+        if let Self::Semantic {
+            surface_delta: true,
+            recompute_pending,
+            ..
+        } = self
+        {
+            *recompute_pending = true;
+        } else {
+            self.request_repaint();
+        }
+    }
+
+    pub(crate) fn requires_recompute(&self) -> bool {
+        matches!(
+            self,
+            Self::Semantic {
+                recompute_pending: true,
+                ..
+            }
+        )
+    }
+
     pub(crate) fn reset_baseline(&mut self) {
         match self {
             Self::Semantic { last_surface, .. } => *last_surface = None,
@@ -127,11 +160,14 @@ impl ClientRenderState {
             last_surface,
             surface_revision,
             surface_reuse,
+            surface_delta,
+            recompute_pending,
         } = self
         else {
             return None;
         };
-        if surface.graphics.assets.is_empty()
+        if !*recompute_pending
+            && surface.graphics.assets.is_empty()
             && last_surface.as_deref().is_some_and(|last| {
                 last.projection_revision == surface.projection_revision
                     && last.frame == surface.frame
@@ -145,26 +181,41 @@ impl ClientRenderState {
             return None;
         }
         surface.surface_revision = surface_revision.saturating_add(1);
-        let mut committed_surface = surface.clone();
-        committed_surface.graphics.assets.clear();
-        let reused = (*surface_reuse)
+        let assets = std::mem::take(&mut surface.graphics.assets);
+        let committed_surface = surface.clone();
+        surface.graphics.assets = assets;
+        let mut message = ServerMessage::PaneSurface(surface);
+        let delta = (*surface_delta)
             .then_some(last_surface.as_deref())
             .flatten()
-            .filter(|last| {
-                last.boot_id == surface.boot_id
-                    && last.frame == surface.frame
-                    // Popup cells are not part of the reusable grid; keep their compact codec.
-                    && surface.popup.is_none()
-                    && surface.graphics.assets.is_empty()
-            })
             .and_then(|last| {
-                crate::protocol::surface_reuse::message(last.surface_revision, &mut surface)
-                    .map_err(|error| tracing::warn!(%error, "failed to encode surface reuse"))
+                crate::protocol::surface_delta::message(last, &mut message)
+                    .map_err(|error| tracing::warn!(%error, "failed to encode surface delta"))
                     .ok()
                     .flatten()
             });
+        let reused = if let ServerMessage::PaneSurface(surface) = &mut message {
+            (delta.is_none() && *surface_reuse)
+                .then_some(last_surface.as_deref())
+                .flatten()
+                .filter(|last| {
+                    last.boot_id == surface.boot_id
+                        && last.frame == surface.frame
+                        // Popup cells are not part of the reusable grid; keep their compact codec.
+                        && surface.popup.is_none()
+                        && surface.graphics.assets.is_empty()
+                })
+                .and_then(|last| {
+                    crate::protocol::surface_reuse::message(last.surface_revision, surface)
+                        .map_err(|error| tracing::warn!(%error, "failed to encode surface reuse"))
+                        .ok()
+                        .flatten()
+                })
+        } else {
+            None
+        };
         Some(PreparedRender::Semantic {
-            message: reused.unwrap_or(ServerMessage::PaneSurface(surface)),
+            message: delta.or(reused).unwrap_or(message),
             committed_surface: Box::new(committed_surface),
         })
     }
@@ -181,6 +232,9 @@ impl ClientRenderState {
         else {
             return None;
         };
+        if self.requires_recompute() {
+            return None;
+        }
         let last = last_surface.as_deref()?;
         if last.boot_id != patch.boot_id
             || last.projection_revision != patch.projection_revision
@@ -201,6 +255,7 @@ impl ClientRenderState {
                 Self::Semantic {
                     last_surface,
                     surface_revision,
+                    recompute_pending,
                     ..
                 },
                 PreparedRender::Semantic {
@@ -209,6 +264,7 @@ impl ClientRenderState {
             ) => {
                 *surface_revision = committed_surface.surface_revision;
                 *last_surface = Some(committed_surface);
+                *recompute_pending = false;
             }
             (
                 Self::Semantic {
@@ -518,6 +574,46 @@ mod tests {
     }
 
     #[test]
+    fn surface_delta_recompute_preserves_wire_baseline_but_epoch_reset_drops_it() {
+        for enabled in [false, true] {
+            let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
+            state.enable_surface_delta(enabled);
+            let mut surface = popup_surface("popup");
+            surface.popup = None;
+            surface.frame = FrameData::from_ratatui_buffer(
+                &ratatui::buffer::Buffer::empty(Rect::new(0, 0, 120, 40)),
+                None,
+            );
+            let initial = state.prepare_pane_surface(surface.clone()).unwrap();
+            state.commit_sent_frame(initial);
+            state.request_recompute();
+            assert_eq!(state.last_pane_surface().is_some(), enabled);
+            assert_eq!(state.requires_recompute(), enabled);
+            // A freshness request still emits a new revision when every cell is equal.
+            let fresh = state.prepare_pane_surface(surface.clone()).unwrap();
+            assert_eq!(
+                matches!(fresh.message(), ServerMessage::EndpointControl { kind, .. }
+                if kind == crate::protocol::surface_delta::MESSAGE_KIND),
+                enabled
+            );
+            assert_eq!(
+                state.requires_recompute(),
+                enabled,
+                "prepare must not commit"
+            );
+            state.commit_sent_frame(fresh);
+            assert!(!state.requires_recompute());
+            assert_eq!(state.last_pane_surface().unwrap().surface_revision, 2);
+            state.request_repaint();
+            assert!(state.last_pane_surface().is_none());
+            let recovery = state.prepare_pane_surface(surface).unwrap();
+            assert!(
+                matches!(recovery.message(), ServerMessage::PaneSurface(frame) if frame.surface_revision == 3)
+            );
+        }
+    }
+
+    #[test]
     fn surface_reuse_preserves_projection_and_patch_baselines_without_resending_cells() {
         for enabled in [false, true] {
             let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);