fix: keep synchronized pane frames atomic (#4508)

refs #2968

6 files changed, 402 insertions, 69 deletions

diff --git a/src/server/client_shell.rs b/src/server/client_shell.rs
index 554e8167..d542896a 100644
--- a/src/server/client_shell.rs
+++ b/src/server/client_shell.rs
@@ -276,6 +276,12 @@ pub(super) struct RenderedPaneSurface {
     pub(super) graphics_delivery: crate::kitty_graphics::surface::DeliveryCache,
 }
 
+#[derive(Debug)]
+pub(super) enum SurfaceRenderDeferred {
+    Synchronized,
+    Changed,
+}
+
 pub(super) fn render_pane_surface(
     app: &mut app::App,
     target: Option<crate::ui::TabSurfaceTarget>,
@@ -285,36 +291,54 @@ pub(super) fn render_pane_surface(
     cell_size: crate::kitty_graphics::HostCellSize,
     graphics_delivery: &crate::kitty_graphics::surface::DeliveryCache,
     client_id: u64,
-) -> RenderedPaneSurface {
-    let content_revisions_before = target
-        .and_then(|target| {
-            let workspace = app.state.workspaces.get(target.workspace_index)?;
-            let tab = workspace.tabs.get(target.tab_index)?;
-            Some(
-                tab.layout
-                    .pane_ids()
-                    .into_iter()
-                    .filter_map(|pane_id| {
-                        app.state
-                            .runtime_for_pane_in_workspace(
-                                &app.terminal_runtimes,
-                                target.workspace_index,
-                                pane_id,
-                            )
-                            .map(|runtime| (pane_id, runtime.content_seq()))
-                    })
-                    .collect::<std::collections::HashMap<_, _>>(),
-            )
-        })
-        .unwrap_or_default();
+) -> Result<RenderedPaneSurface, SurfaceRenderDeferred> {
+    let layout = crate::ui::compute_tab_surface_for(
+        &app.state,
+        &app.terminal_runtimes,
+        target,
+        area,
+        resize_panes,
+        cell_size,
+    );
+    let mut content_revisions_before = std::collections::HashMap::new();
+    if let Some(target) = target {
+        for pane in &layout.pane_infos {
+            if let Some(runtime) = app.state.runtime_for_pane_in_workspace(
+                &app.terminal_runtimes,
+                target.workspace_index,
+                pane.id,
+            ) {
+                let (synchronized, epoch) = runtime.synchronized_output_state();
+                if synchronized {
+                    return Err(SurfaceRenderDeferred::Synchronized);
+                }
+                let revision = runtime.content_seq();
+                content_revisions_before.insert(pane.id, (epoch, revision));
+            }
+        }
+    }
+    let popup_revision_before = if show_popup {
+        app.state
+            .popup_pane
+            .as_ref()
+            .and_then(|popup| app.terminal_runtimes.get(&popup.terminal_id))
+            .map(|runtime| {
+                let (synchronized, epoch) = runtime.synchronized_output_state();
+                if synchronized {
+                    return Err(SurfaceRenderDeferred::Synchronized);
+                }
+                Ok(epoch)
+            })
+            .transpose()?
+    } else {
+        None
+    };
     let (buffer, cursor, hyperlinks, layout) =
         crate::server::render_stream::render_tab_surface_virtual(
             &app.state,
             &app.terminal_runtimes,
-            target,
+            layout,
             area,
-            resize_panes,
-            cell_size,
         );
     let panes = target
         .map(|target| {
@@ -343,7 +367,9 @@ pub(super) fn render_pane_surface(
                         };
                         let content_revision = runtime.map_or(0, |runtime| {
                             let after = runtime.content_seq();
-                            if content_revisions_before.get(&pane.id).copied() == Some(after)
+                            if content_revisions_before
+                                .get(&pane.id)
+                                .is_some_and(|&(_, before)| before == after)
                                 && after.is_multiple_of(2)
                             {
                                 after
@@ -422,14 +448,47 @@ pub(super) fn render_pane_surface(
         graphics_delivery,
         client_id,
     );
-    RenderedPaneSurface {
+    if let Some(target) = target {
+        for (&pane_id, &(epoch, _)) in &content_revisions_before {
+            if let Some(runtime) = app.state.runtime_for_pane_in_workspace(
+                &app.terminal_runtimes,
+                target.workspace_index,
+                pane_id,
+            ) {
+                let (synchronized, after_epoch) = runtime.synchronized_output_state();
+                if synchronized {
+                    return Err(SurfaceRenderDeferred::Synchronized);
+                }
+                if after_epoch != epoch {
+                    return Err(SurfaceRenderDeferred::Changed);
+                }
+            }
+        }
+    }
+    if let Some(before) = popup_revision_before {
+        if let Some(runtime) = app
+            .state
+            .popup_pane
+            .as_ref()
+            .and_then(|popup| app.terminal_runtimes.get(&popup.terminal_id))
+        {
+            let (synchronized, after_epoch) = runtime.synchronized_output_state();
+            if synchronized {
+                return Err(SurfaceRenderDeferred::Synchronized);
+            }
+            if after_epoch != before {
+                return Err(SurfaceRenderDeferred::Changed);
+            }
+        }
+    }
+    Ok(RenderedPaneSurface {
         frame: FrameData::from_ratatui_buffer_with_hyperlinks(&buffer, cursor, &hyperlinks),
         panes,
         splits,
         popup,
         graphics,
         graphics_delivery: next_graphics_delivery,
-    }
+    })
 }
 
 fn render_popup_surface(
diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs
index 82564ba3..a3bbedfa 100644
--- a/src/server/headless/render.rs
+++ b/src/server/headless/render.rs
@@ -441,12 +441,33 @@ impl HeadlessServer {
                 });
             if changed {
                 if let Some(target) = self.shell_target_for_client(*client_id) {
+                    let area = Rect::new(0, 0, *cols, *rows);
+                    let layout = crate::ui::compute_tab_surface_for(
+                        &self.app.state,
+                        &self.app.terminal_runtimes,
+                        Some(target),
+                        area,
+                        false,
+                        *cell_size,
+                    );
+                    if layout.pane_infos.iter().any(|pane| {
+                        self.app
+                            .state
+                            .runtime_for_pane_in_workspace(
+                                &self.app.terminal_runtimes,
+                                target.workspace_index,
+                                pane.id,
+                            )
+                            .is_some_and(|runtime| runtime.synchronized_output_active())
+                    }) {
+                        continue;
+                    }
                     crate::ui::resize_tab_surface(
                         &self.app.state,
                         &self.app.terminal_runtimes,
                         target.workspace_index,
                         target.tab_index,
-                        Rect::new(0, 0, *cols, *rows),
+                        area,
                         if cell_size.is_known() {
                             *cell_size
                         } else {
@@ -473,6 +494,55 @@ impl HeadlessServer {
             let shell_target = self.shell_target_for_client(client_id);
             let shell_tab_id = self.shell_tab_id_for_client(client_id);
             let shell_shows_popup = shell_tab_id.as_deref() == self.popup_owner_tab_id.as_deref();
+            let shell_graphics_delivery = self
+                .clients
+                .get(&client_id)
+                .map(|client| client.shell_graphics_delivery.clone())
+                .unwrap_or_default();
+            let shell_render = if matches!(mode, ClientConnectionMode::ClientShell)
+                && self
+                    .clients
+                    .get(&client_id)
+                    .is_some_and(|client| client.shell_surface_active)
+            {
+                let render_started = crate::render_prof::timer();
+                let render_cell_size = if cell_size.is_known() {
+                    cell_size
+                } else {
+                    crate::kitty_graphics::HostCellSize::default()
+                };
+                let result = render_client_shell_pane_surface(
+                    &mut self.app,
+                    shell_target,
+                    area,
+                    false,
+                    shell_shows_popup,
+                    render_cell_size,
+                    &shell_graphics_delivery,
+                    client_id,
+                );
+                crate::render_prof::duration_since(
+                    "full_render.render_tab_surface_virtual",
+                    render_started,
+                );
+                match result {
+                    Ok(surface) => Some(surface),
+                    Err(reason) => {
+                        if let Some(client) = self.clients.get_mut(&client_id) {
+                            client.render_state.request_recompute();
+                        }
+                        if matches!(
+                            reason,
+                            crate::server::client_shell::SurfaceRenderDeferred::Changed
+                        ) {
+                            self.app.render_dirty.request_generic();
+                        }
+                        continue;
+                    }
+                }
+            } else {
+                None
+            };
             let mut shell_projection_revision = 0;
             if matches!(mode, ClientConnectionMode::ClientShell) {
                 let location = self
@@ -585,20 +655,9 @@ impl HeadlessServer {
                     continue;
                 }
             }
-            let shell_graphics_delivery = self
-                .clients
-                .get(&client_id)
-                .map(|client| client.shell_graphics_delivery.clone())
-                .unwrap_or_default();
             let mut surface_parts = None;
             let frame = match mode {
                 ClientConnectionMode::ClientShell => {
-                    let render_started = crate::render_prof::timer();
-                    let render_cell_size = if cell_size.is_known() {
-                        cell_size
-                    } else {
-                        crate::kitty_graphics::HostCellSize::default()
-                    };
                     let crate::server::client_shell::RenderedPaneSurface {
                         frame,
                         panes,
@@ -606,20 +665,7 @@ impl HeadlessServer {
                         popup,
                         graphics,
                         graphics_delivery: next_graphics_delivery,
-                    } = render_client_shell_pane_surface(
-                        &mut self.app,
-                        shell_target,
-                        area,
-                        false,
-                        shell_shows_popup,
-                        render_cell_size,
-                        &shell_graphics_delivery,
-                        client_id,
-                    );
-                    crate::render_prof::duration_since(
-                        "full_render.render_tab_surface_virtual",
-                        render_started,
-                    );
+                    } = shell_render.expect("active shell surface");
                     surface_parts = Some((panes, splits, popup, graphics, next_graphics_delivery));
                     frame
                 }
@@ -638,6 +684,13 @@ impl HeadlessServer {
                         broken_clients.push(client_id);
                         continue;
                     };
+                    let (synchronized, epoch) = runtime.synchronized_output_state();
+                    if synchronized {
+                        if let Some(client) = self.clients.get_mut(&client_id) {
+                            client.render_state.request_recompute();
+                        }
+                        continue;
+                    }
                     let render_started = crate::render_prof::timer();
                     let (buffer, cursor) =
                         crate::server::render_stream::render_terminal_virtual(runtime, area);
@@ -651,6 +704,16 @@ impl HeadlessServer {
                         "full_render.visible_hyperlinks",
                         hyperlinks_started,
                     );
+                    let (synchronized, after_epoch) = runtime.synchronized_output_state();
+                    if synchronized || after_epoch != epoch {
+                        if let Some(client) = self.clients.get_mut(&client_id) {
+                            client.render_state.request_recompute();
+                        }
+                        if !synchronized {
+                            self.app.render_dirty.request_generic();
+                        }
+                        continue;
+                    }
                     let frame_started = crate::render_prof::timer();
                     let frame = FrameData::from_ratatui_buffer_with_hyperlinks(
                         &buffer,
diff --git a/src/server/headless/retained_surface.rs b/src/server/headless/retained_surface.rs
index 2a637eac..07d33ef5 100644
--- a/src/server/headless/retained_surface.rs
+++ b/src/server/headless/retained_surface.rs
@@ -206,6 +206,20 @@ struct RetainedRecipientUpdate {
     )>,
 }
 
+fn has_synchronized_pane(app: &app::App, surface: &protocol::PaneSurfaceFrame) -> bool {
+    surface.panes.iter().any(|pane| {
+        app.parse_pane_id(&pane.pane_id)
+            .and_then(|(workspace_index, pane_id)| {
+                app.state.runtime_for_pane_in_workspace(
+                    &app.terminal_runtimes,
+                    workspace_index,
+                    pane_id,
+                )
+            })
+            .is_some_and(|runtime| runtime.synchronized_output_active())
+    })
+}
+
 impl HeadlessServer {
     /// Applies terminal dirty rows to the committed origin-relative pane surface.
     /// Any presentation or geometry uncertainty falls back to the complete renderer.
@@ -281,6 +295,9 @@ impl HeadlessServer {
             {
                 fallback!("baseline_mismatch");
             }
+            if has_synchronized_pane(&self.app, surface) {
+                fallback!("synchronized_visible");
+            }
             recipients.push(RetainedRecipient {
                 client_id: *client_id,
                 surface,
@@ -346,7 +363,7 @@ impl HeadlessServer {
         }
 
         let mut updates = Vec::with_capacity(recipients.len());
-        for recipient in recipients {
+        for recipient in &recipients {
             let client_id = recipient.client_id;
             let surface = recipient.surface;
             let mut panes = surface.panes.clone();
@@ -459,6 +476,12 @@ impl HeadlessServer {
         if updates.is_empty() {
             success!("unchanged");
         }
+        if recipients
+            .iter()
+            .any(|recipient| has_synchronized_pane(&self.app, recipient.surface))
+        {
+            fallback!("synchronized_during_patch");
+        }
 
         let mut sent = 0u64;
         let mut deferred = 0u64;
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 5f54f4b6..84031ce4 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -1263,6 +1263,147 @@ fn recv_pane_surface_patch(
 }
 
 #[tokio::test]
+async fn unrelated_render_keeps_synchronized_pane_frame_committed() {
+    let mut server = test_headless_server();
+    let pane_id = install_shared_view_test_runtime(&mut server);
+    let (_control, render) = connect_matching_test_shell(&mut server, 7);
+    server.render_and_stream();
+    let before = recv_pane_surface(&render, "baseline");
+    assert!(frame_text(&before.frame).contains("BASE"));
+    let projection_before = server.clients[&7].shell_projection_revision;
+
+    write_shared_test_pane(
+        &mut server,
+        pane_id,
+        b"\x1b[?2026h\x1b[?1049h\x1b[2J\x1b[HPARTIAL",
+    );
+    server.app.state.workspaces[0].custom_name = Some("renamed during frame".into());
+    server.clients.get_mut(&7).unwrap().request_recompute();
+    server.render_and_stream();
+    assert!(render.try_recv().is_err(), "partial frame was published");
+    assert_eq!(
+        server.clients[&7].shell_projection_revision,
+        projection_before
+    );
+
+    write_shared_test_pane(&mut server, pane_id, b"\rCOMPLETE\x1b[?2026l");
+    assert!(!server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
+    server.render_and_stream();
+    let after = recv_pane_surface(&render, "completed frame");
+    assert!(frame_text(&after.frame).contains("COMPLETE"));
+    assert!(after.projection_revision > projection_before);
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn sibling_retained_output_waits_for_synchronized_pane_to_finish() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("synchronized-split");
+    let first = workspace.tabs[0].root_pane;
+    let second = workspace.test_split(ratatui::layout::Direction::Vertical);
+    workspace.insert_test_runtime(
+        first,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 23, b"FIRST"),
+    );
+    workspace.insert_test_runtime(
+        second,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 23, b"SECOND"),
+    );
+    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 (_control, render) = connect_matching_test_shell(&mut server, 7);
+    server.render_and_stream();
+    let _ = recv_pane_surface(&render, "split baseline");
+
+    write_shared_test_pane(&mut server, first, b"\x1b[?2026h\rPARTIAL");
+    write_shared_test_pane(&mut server, second, b"\rUPDATED");
+    assert!(!server.render_retained_pane_surface_and_stream(&HashSet::from([second])));
+    server.render_and_stream();
+    assert!(
+        render.try_recv().is_err(),
+        "sibling published partial frame"
+    );
+
+    write_shared_test_pane(&mut server, first, b"\rCOMPLETE\x1b[?2026l");
+    assert!(!server.render_retained_pane_surface_and_stream(&HashSet::from([first])));
+    server.render_and_stream();
+    let after = recv_pane_surface(&render, "completed split");
+    let text = frame_text(&after.frame);
+    assert!(
+        text.contains("COMPLETE") && text.contains("UPDATED"),
+        "{text}"
+    );
+    assert!(!text.contains("PARTIAL"));
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn popup_synchronized_output_waits_for_complete_frame() {
+    let mut server = test_headless_server();
+    install_shared_view_test_runtime(&mut server);
+    let popup_runtime =
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(40, 12, b"POPUP_BASE");
+    let (_, popup_id) = server.app.install_test_popup_runtime(popup_runtime);
+    server.popup_owner_tab_id = server.app.public_tab_id(0, 0);
+    let (_control, render) = connect_matching_test_shell(&mut server, 7);
+    server.render_and_stream();
+    assert!(recv_pane_surface(&render, "popup baseline").popup.is_some());
+
+    server
+        .app
+        .terminal_runtimes
+        .get(&popup_id)
+        .unwrap()
+        .test_process_pty_bytes(b"\x1b[?2026h\rPOPUP_PARTIAL");
+    server.render_and_stream();
+    assert!(render.try_recv().is_err(), "partial popup was published");
+
+    server
+        .app
+        .terminal_runtimes
+        .get(&popup_id)
+        .unwrap()
+        .test_process_pty_bytes(b"\rPOPUP_COMPLETE\x1b[?2026l");
+    server.render_and_stream();
+    let after = recv_pane_surface(&render, "complete popup");
+    assert!(after
+        .popup
+        .as_ref()
+        .is_some_and(|popup| frame_text(&popup.frame).contains("POPUP_COMPLETE")));
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn zoom_hidden_synchronized_pane_does_not_block_surface() {
+    let mut server = test_headless_server();
+    let mut workspace = crate::workspace::Workspace::test_new("zoomed-sync");
+    let hidden = workspace.tabs[0].root_pane;
+    let visible = workspace.test_split(ratatui::layout::Direction::Vertical);
+    workspace.tabs[0].zoomed = true;
+    workspace.insert_test_runtime(
+        hidden,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 23, b"HIDDEN"),
+    );
+    workspace.insert_test_runtime(
+        visible,
+        crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 23, b"VISIBLE"),
+    );
+    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 (_control, render) = connect_matching_test_shell(&mut server, 7);
+    write_shared_test_pane(&mut server, hidden, b"\x1b[?2026h\rPARTIAL");
+    server.render_and_stream();
+    let surface = recv_pane_surface(&render, "zoomed visible pane");
+    assert!(frame_text(&surface.frame).contains("VISIBLE"));
+    assert!(!frame_text(&surface.frame).contains("PARTIAL"));
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
 async fn retained_snapshot_survives_a_writer_waiting_for_the_terminal_core() {
     let mut server = test_headless_server();
     let pane_id = install_shared_view_test_runtime(&mut server);
@@ -3830,6 +3971,60 @@ fn with_terminal_session_test_server(
     rt.shutdown_timeout(Duration::from_millis(100));
 }
 
+#[test]
+fn terminal_observers_wait_for_synchronized_output_with_or_without_baseline() {
+    with_terminal_session_test_server(|server, terminal_id, target, _| {
+        let connect = |server: &mut HeadlessServer, client_id| {
+            let (writer, _control, render) = test_client_writer();
+            assert!(!server.handle_server_event(ServerEvent::ClientConnected {
+                client_id,
+                cols: 80,
+                rows: 24,
+                cell_width_px: 0,
+                cell_height_px: 0,
+                pixel_mouse: false,
+                writer,
+            }));
+            assert!(
+                server.handle_server_event(ServerEvent::ClientObserveTerminal {
+                    client_id,
+                    target: target.clone(),
+                })
+            );
+            render
+        };
+        let first = connect(server, 7);
+        server.render_and_stream();
+        let _ = first.recv().expect("observer baseline");
+        server
+            .app
+            .terminal_runtimes
+            .get(&terminal_id)
+            .unwrap()
+            .test_process_pty_bytes(b"\x1b[?2026h\rPARTIAL");
+        let second = connect(server, 8);
+        server.render_and_stream();
+        assert!(first.try_recv().is_err());
+        assert!(second.try_recv().is_err());
+
+        server
+            .app
+            .terminal_runtimes
+            .get(&terminal_id)
+            .unwrap()
+            .test_process_pty_bytes(b"\rCOMPLETE\x1b[?2026l");
+        server.render_and_stream();
+        for frames in [&first, &second] {
+            let ServerMessage::Terminal(frame) =
+                read_server_message(frames.recv().expect("complete observer frame"))
+            else {
+                panic!("expected terminal frame");
+            };
+            assert!(String::from_utf8_lossy(&frame.bytes).contains("COMPLETE"));
+        }
+    });
+}
+
 fn connect_pending_terminal_client(server: &mut HeadlessServer, client_id: u64) {
     let _control_rx = connect_pending_terminal_client_with_control_rx(server, client_id);
 }
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index dcafa1a9..f4bb310b 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -95,7 +95,8 @@ impl RenderPipeline {
             },
             &self.graphics_delivery,
             1,
-        );
+        )
+        .expect("benchmark surface");
         let server_elapsed = started.elapsed();
         self.graphics_delivery = rendered.graphics_delivery;
 
@@ -469,7 +470,8 @@ fn print_surface_damage_profiles() {
             HostCellSize::default(),
             &pipeline.graphics_delivery,
             1,
-        );
+        )
+        .expect("benchmark surface");
         let mut surface = PaneSurfaceFrame {
             boot_id: "bench-boot".into(),
             projection_revision: 1,
@@ -520,7 +522,8 @@ fn print_surface_reuse_profiles() {
                 HostCellSize::default(),
                 &pipeline.graphics_delivery,
                 1,
-            );
+            )
+            .expect("benchmark surface");
             let surface = PaneSurfaceFrame {
                 boot_id: "bench-boot".into(),
                 projection_revision: 1,
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index b0af26d7..db613e96 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -472,19 +472,9 @@ pub(crate) type RenderedTabSurface = (
 pub(crate) fn render_tab_surface_virtual(
     app_state: &AppState,
     terminal_runtimes: &TerminalRuntimeRegistry,
-    target: Option<crate::ui::TabSurfaceTarget>,
+    layout: crate::ui::TabSurfaceLayout,
     area: Rect,
-    resize_panes: bool,
-    cell_size: crate::kitty_graphics::HostCellSize,
 ) -> RenderedTabSurface {
-    let layout = crate::ui::compute_tab_surface_for(
-        app_state,
-        terminal_runtimes,
-        target,
-        area,
-        resize_panes,
-        cell_size,
-    );
     let surface = crate::ui::TabSurfaceView {
         target: layout.target,
         pane_infos: &layout.pane_infos,