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

8 files changed, 189 insertions, 10 deletions

diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 465408f5..88daf250 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -418,6 +418,7 @@ pub(crate) enum ServerEvent {
         surface_active: bool,
         surface_reuse: bool,
         surface_delta: bool,
+        surface_scroll: bool,
         writer: ClientWriter,
     },
     /// A client sent an input message.
@@ -785,6 +786,7 @@ pub(crate) fn handle_client_handshake(
                     hello.surface_active,
                     hello.surface_reuse,
                     hello.surface_delta,
+                    hello.surface_scroll,
                 )),
             )
         }
@@ -881,6 +883,7 @@ pub(crate) fn handle_client_handshake(
         surface_active,
         surface_reuse,
         surface_delta,
+        surface_scroll,
     )) = shell_options
     {
         ServerEvent::ClientShellConnected {
@@ -896,6 +899,7 @@ pub(crate) fn handle_client_handshake(
             surface_active,
             surface_reuse,
             surface_delta,
+            surface_scroll,
             writer,
         }
     } else {
@@ -1472,6 +1476,7 @@ mod tests {
             surface_active: true,
             surface_reuse: false,
             surface_delta: false,
+            surface_scroll: 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()],
@@ -1989,10 +1994,12 @@ mod tests {
                 surface_active,
                 surface_reuse,
                 surface_delta,
+                surface_scroll,
                 writer,
             } => {
                 assert!(!surface_reuse);
                 assert!(!surface_delta);
+                assert!(!surface_scroll);
                 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 da3baf30..18869e09 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -1861,6 +1861,7 @@ impl HeadlessServer {
                 surface_active,
                 surface_reuse,
                 surface_delta,
+                surface_scroll,
                 writer,
             } => {
                 if self.handoff_in_progress {
@@ -1908,6 +1909,9 @@ impl HeadlessServer {
                 connection.shell_surface_active = surface_active;
                 connection.render_state.enable_surface_reuse(surface_reuse);
                 connection.render_state.enable_surface_delta(surface_delta);
+                connection
+                    .render_state
+                    .enable_surface_scroll(surface_scroll);
                 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 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);
+}
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index f4bb310b..754c9dbc 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -376,7 +376,8 @@ fn surface_encoding_client(mode: &str, surface: &PaneSurfaceFrame) -> SurfaceEnc
         state: super::render_stream::ClientRenderState::new(
             crate::protocol::RenderEncoding::SemanticFrame,
         ),
-        decoder: (reuse || delta).then(|| crate::protocol::surface_reuse::Decoder::new(delta)),
+        decoder: (reuse || delta)
+            .then(|| crate::protocol::surface_reuse::Decoder::new(delta, false)),
     };
     client.state.enable_surface_reuse(reuse);
     client.state.enable_surface_delta(delta);
@@ -588,3 +589,39 @@ async fn render_scale_profile() {
     print_surface_reuse_profiles();
     print_surface_damage_profiles();
 }
+
+/// Dirty-row collection while output scrolls a large pane, the host's hot path
+/// for streaming agents and build logs.
+#[tokio::test(flavor = "current_thread")]
+#[ignore = "manual dirty collection profile"]
+async fn dirty_scroll_collection_profile() {
+    const WIDTH: u16 = 295;
+    const HEIGHT: u16 = 84;
+    let runtime = TerminalRuntime::test_with_scrollback_bytes(WIDTH, HEIGHT, 1024 * 1024, b"");
+    let mut samples = Vec::new();
+    let mut rows = 0;
+    for line in 0..2_000u32 {
+        runtime.test_process_pty_bytes(
+            format!("{:010}{:010}\r\n", line.wrapping_mul(2_654_435_761), line).as_bytes(),
+        );
+        let started = Instant::now();
+        let snapshot = black_box(runtime.collect_dirty_patch_snapshot(WIDTH, HEIGHT));
+        if line as usize >= WARMUP_COUNT {
+            samples.push(started.elapsed());
+        }
+        if let Some(crate::pane::TerminalDirtyPatchOutcome::Patch(patch)) =
+            snapshot.map(|snapshot| snapshot.patch)
+        {
+            rows += patch.rows.len();
+        }
+    }
+    samples.sort_unstable();
+    let at = |fraction: f64| samples[((samples.len() - 1) as f64 * fraction) as usize].as_micros();
+    println!(
+        "dirty scroll collection {WIDTH}x{HEIGHT}: median {}us p95 {}us max {}us, {} rows per update",
+        at(0.5),
+        at(0.95),
+        at(1.0),
+        rows / 2_000
+    );
+}
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index ee874e52..aa1bbabe 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -19,6 +19,7 @@ pub(crate) enum ClientRenderState {
         surface_revision: u64,
         surface_reuse: bool,
         surface_delta: bool,
+        surface_scroll: bool,
         recompute_pending: bool,
     },
     /// Terminal-ANSI clients keep a terminal diff encoder and sequence number.
@@ -37,6 +38,7 @@ impl ClientRenderState {
                 surface_revision: 0,
                 surface_reuse: false,
                 surface_delta: false,
+                surface_scroll: false,
                 recompute_pending: false,
             },
             RenderEncoding::TerminalAnsi => Self::TerminalAnsi {
@@ -59,6 +61,12 @@ impl ClientRenderState {
         }
     }
 
+    pub(crate) fn enable_surface_scroll(&mut self, enabled: bool) {
+        if let Self::Semantic { surface_scroll, .. } = self {
+            *surface_scroll = enabled;
+        }
+    }
+
     pub(crate) fn request_recompute(&mut self) {
         if let Self::Semantic {
             surface_delta: true,
@@ -171,6 +179,7 @@ impl ClientRenderState {
             surface_reuse,
             surface_delta,
             recompute_pending,
+            ..
         } = self
         else {
             return None;
@@ -239,6 +248,7 @@ impl ClientRenderState {
         let Self::Semantic {
             last_surface,
             surface_revision,
+            surface_scroll,
             ..
         } = self
         else {
@@ -256,8 +266,18 @@ impl ClientRenderState {
         }
         let next_revision = surface_revision.saturating_add(1);
         patch.surface_revision = next_revision;
-        Some(PreparedRender::SemanticPatch {
-            message: ServerMessage::PaneSurfacePatch(patch),
+        let scrolled = (*surface_scroll)
+            .then(|| crate::protocol::surface_scroll::message(last, &patch))
+            .flatten();
+        Some(match scrolled {
+            Some(message) => PreparedRender::SemanticPatch {
+                message,
+                encoded: Some(Box::new(patch)),
+            },
+            None => PreparedRender::SemanticPatch {
+                message: ServerMessage::PaneSurfacePatch(patch),
+                encoded: None,
+            },
         })
     }
 
@@ -284,10 +304,13 @@ impl ClientRenderState {
                     surface_revision,
                     ..
                 },
-                PreparedRender::SemanticPatch {
-                    message: ServerMessage::PaneSurfacePatch(patch),
-                },
+                PreparedRender::SemanticPatch { message, encoded },
             ) => {
+                let patch = match (encoded, message) {
+                    (Some(patch), _) => *patch,
+                    (None, ServerMessage::PaneSurfacePatch(patch)) => patch,
+                    (None, _) => unreachable!("a plain semantic patch carries its pane patch"),
+                };
                 let surface = last_surface
                     .as_deref_mut()
                     .expect("prepared patch baseline");
@@ -358,6 +381,8 @@ pub(crate) enum PreparedRender {
     },
     SemanticPatch {
         message: ServerMessage,
+        /// The pane patch a compact `message` encodes; `None` when `message` is that patch.
+        encoded: Option<Box<PaneSurfacePatch>>,
     },
     TerminalAnsi {
         message: ServerMessage,
@@ -370,7 +395,7 @@ impl PreparedRender {
     pub(crate) fn message(&self) -> &ServerMessage {
         match self {
             Self::Semantic { message, .. }
-            | Self::SemanticPatch { message }
+            | Self::SemanticPatch { message, .. }
             | Self::TerminalAnsi { message, .. } => message,
         }
     }