fix: keep remote surface updates incremental (#4209)

refs #3769

refs #3745

3 files changed, 244 insertions, 0 deletions

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,