perf: send scrolling pane output as row shifts and skip blank cells (#4711)

* perf: send scrolling pane output as row shifts and skip blank cells

* fix: keep expanded scroll patches disjoint for the fast presentation path

4 files changed, 109 insertions, 3 deletions

diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 2dd082d6..27f3bfff 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -10,6 +10,8 @@ mod retained_graphics_tests;
 mod surface_delta_tests;
 #[path = "surface_interest.rs"]
 mod surface_interest_tests;
+#[path = "surface_scroll.rs"]
+mod surface_scroll_tests;
 
 fn client_shell_projection(
     receiver: &std::sync::mpsc::Receiver<Vec<u8>>,
@@ -719,6 +721,7 @@ async fn client_shell_attach_seeds_workspace() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 6,
             surface_cols: 80,
             surface_rows: 23,
@@ -761,6 +764,7 @@ async fn completion_guard_endpoint_pairs_runtime_completions_with_snapshots() {
         surface_active: false,
         surface_reuse: false,
         surface_delta: false,
+        surface_scroll: false,
         writer,
     });
     let (_, initial) = client_shell_projection(&control_rx);
@@ -809,6 +813,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols: 80,
             surface_rows: 23,
@@ -927,6 +932,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 77,
             surface_cols: 80,
             surface_rows: 23,
@@ -1029,6 +1035,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 7,
             surface_cols: 80,
             surface_rows: 23,
@@ -1196,6 +1203,7 @@ fn connect_test_shell(
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols,
             surface_rows,
@@ -1800,6 +1808,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 13,
             surface_cols: 80,
             surface_rows: 23,
@@ -1824,6 +1833,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 14,
             surface_cols: 80,
             surface_rows: 23,
@@ -2743,6 +2753,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 9,
             surface_cols: 80,
             surface_rows: 23,
@@ -2989,6 +3000,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: 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
index f010c1e3..996b5b1b 100644
--- a/src/server/headless/tests/surface_delta.rs
+++ b/src/server/headless/tests/surface_delta.rs
@@ -51,7 +51,7 @@ async fn surface_delta_reconstructs_metadata_text_and_hyperlinks() {
     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 mut decoder = protocol::surface_reuse::Decoder::new(true, false);
     let initial = decode_surface_message(&mut decoder, initial_message);
     assert_eq!(
         without_asset_payload(initial.clone()),
@@ -119,7 +119,7 @@ async fn surface_delta_updates_global_popup_without_affecting_legacy_peer() {
     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 mut decoder = protocol::surface_reuse::Decoder::new(true, false);
     let _ = decode_surface_message(&mut decoder, initial_message);
 
     server
@@ -160,7 +160,7 @@ async fn surface_delta_preserves_graphics_baseline_through_queue_recovery() {
     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 mut decoder = protocol::surface_reuse::Decoder::new(true, false);
     let _ = decode_surface_message(&mut decoder, initial_message);
 
     write_shared_test_pane(
diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs
index f3acc27b..a7da046e 100644
--- a/src/server/headless/tests/surface_interest.rs
+++ b/src/server/headless/tests/surface_interest.rs
@@ -66,6 +66,7 @@ async fn metadata_only_shell_is_isolated_until_surface_activation() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols: 101,
             surface_rows: 37,
@@ -281,6 +282,7 @@ async fn background_surface_activation_preserves_focused_viewer_geometry() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: 8,
             surface_cols: 100,
             surface_rows: 35,
@@ -394,6 +396,7 @@ async fn presentation_sync_epoch_replays_modes_and_title() {
         server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -510,6 +513,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
         source_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: source_client_id,
             surface_cols: 80,
             surface_rows: 24,
@@ -533,6 +537,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
         target_server.handle_server_event(ServerEvent::ClientShellConnected {
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: false,
             client_id: target_client_id,
             surface_cols: 80,
             surface_rows: 24,
diff --git a/src/server/headless/tests/surface_scroll.rs b/src/server/headless/tests/surface_scroll.rs
new file mode 100644
index 00000000..8975650b
--- /dev/null
+++ b/src/server/headless/tests/surface_scroll.rs
@@ -0,0 +1,89 @@
+use super::*;
+
+fn scrolling_lines(range: std::ops::Range<usize>) -> Vec<u8> {
+    range
+        .map(|n| {
+            format!(
+                "build step {n:>3}: compiled module_{n} in {}ms\r\n",
+                n * 7 % 97
+            )
+        })
+        .collect::<String>()
+        .into_bytes()
+}
+
+fn apply_rows(frame: &mut FrameData, rows: &[crate::protocol::PaneSurfacePatchRow]) {
+    for row in rows {
+        let start = usize::from(row.y) * usize::from(frame.width) + usize::from(row.x);
+        frame.cells[start..start + row.cells.len()].clone_from_slice(&row.cells);
+    }
+}
+
+#[tokio::test]
+async fn surface_scroll_sends_scrolling_output_as_a_shift_and_new_rows() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(&scrolling_lines(0..40));
+    server
+        .clients
+        .get_mut(&1)
+        .expect("scroll client")
+        .render_state
+        .enable_surface_scroll(true);
+    server.render_and_stream();
+    let mut decoder = protocol::surface_reuse::Decoder::new(false, true);
+    let ServerMessage::PaneSurface(mut shell) = decoder
+        .decode(read_server_message(
+            render_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
+        ))
+        .expect("initial surface")
+    else {
+        panic!("expected the initial pane surface");
+    };
+
+    write_shared_test_pane(&mut server, pane_id, &scrolling_lines(40..42));
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let bytes = render_rx.recv_timeout(Duration::from_secs(1)).unwrap();
+    let message = read_server_message(bytes.clone());
+    assert!(matches!(
+        &message,
+        ServerMessage::EndpointControl { kind, .. } if kind == protocol::surface_scroll::MESSAGE_KIND
+    ));
+
+    // The decoder hands the client shell an ordinary patch that reaches the
+    // server's committed surface exactly.
+    let ServerMessage::PaneSurfacePatch(patch) = decoder.decode(message).expect("scroll decode")
+    else {
+        panic!("expected an expanded pane patch");
+    };
+    apply_rows(&mut shell.frame, &patch.rows);
+    let committed = server.clients[&1]
+        .render_state
+        .last_pane_surface()
+        .expect("committed surface");
+    assert_eq!(shell.frame.cells, committed.frame.cells);
+    assert!(frame_text(&shell.frame).contains("module_41"));
+
+    let expanded = HeadlessServer::frame_server_message(&ServerMessage::PaneSurfacePatch(patch))
+        .expect("expanded frame");
+    assert!(
+        bytes.len() * 4 < expanded.len(),
+        "scroll frame {} should be far smaller than the rows it replaces {}",
+        bytes.len(),
+        expanded.len()
+    );
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn surface_scroll_is_not_sent_to_a_peer_that_did_not_negotiate_it() {
+    let (mut server, _control_rx, render_rx, pane_id) =
+        retained_test_server_with_control(&scrolling_lines(0..40));
+    server.render_and_stream();
+    let _ = render_rx.recv_timeout(Duration::from_secs(1)).unwrap();
+
+    write_shared_test_pane(&mut server, pane_id, &scrolling_lines(40..42));
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    let message = read_server_message(render_rx.recv_timeout(Duration::from_secs(1)).unwrap());
+    assert!(matches!(message, ServerMessage::PaneSurfacePatch(_)));
+    shutdown_test_runtimes(&mut server);
+}