feat: add isolated pane graphics streaming

4 files changed, 852 insertions, 49 deletions

diff --git a/src/server/clients.rs b/src/server/clients.rs
index 705152eb..243a131d 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -20,6 +20,14 @@ pub(crate) type RenderTarget = (
     ClientConnectionMode,
 );
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) enum DeferredRender {
+    #[default]
+    None,
+    Graphics,
+    Full,
+}
+
 /// A connected client tracked by the server.
 pub(crate) struct ClientConnection {
     /// Whether this connection is the full app client or a direct terminal attach.
@@ -50,8 +58,10 @@ pub(crate) struct ClientConnection {
     pub(crate) graphics_cache: crate::kitty_graphics::HostGraphicsCache,
     /// Whether the next graphics frame must clear and rebuild host-side Kitty state.
     pub(crate) graphics_surface_reset_pending: bool,
-    /// Whether a render was skipped because the render channel was full.
+    /// Whether an ordinary render was skipped because the render channel was full.
     pub(crate) render_pending: bool,
+    /// Whether a pane-graphics-only render was skipped because the channel was full.
+    pane_graphics_render_pending: bool,
     /// Last host mouse capture mode sent to this client.
     pub(crate) host_mouse_capture_active: Option<bool>,
     /// Temporary files staged from this client's local clipboard image pastes.
@@ -115,6 +125,7 @@ impl ClientConnection {
             graphics_cache: crate::kitty_graphics::HostGraphicsCache::default(),
             graphics_surface_reset_pending: false,
             render_pending: false,
+            pane_graphics_render_pending: false,
             host_mouse_capture_active: None,
             staged_clipboard_files: Vec::new(),
             writer,
@@ -124,6 +135,39 @@ impl ClientConnection {
     pub(crate) fn request_full_redraw(&mut self) {
         self.render_state.reset_baseline();
         self.graphics_surface_reset_pending = true;
+        self.pane_graphics_render_pending = false;
+    }
+
+    pub(crate) fn deferred_render(&self) -> DeferredRender {
+        if self.render_pending {
+            DeferredRender::Full
+        } else if self.pane_graphics_render_pending {
+            DeferredRender::Graphics
+        } else {
+            DeferredRender::None
+        }
+    }
+
+    pub(crate) fn clear_deferred_render(&mut self) {
+        self.render_pending = false;
+        self.pane_graphics_render_pending = false;
+    }
+
+    pub(crate) fn defer_full_render(&mut self) {
+        self.render_pending = true;
+        self.pane_graphics_render_pending = false;
+    }
+
+    pub(crate) fn defer_pane_graphics_render(&mut self) {
+        if !self.render_pending {
+            self.pane_graphics_render_pending = true;
+        }
+    }
+
+    pub(crate) fn take_deferred_render(&mut self) -> DeferredRender {
+        let deferred = self.deferred_render();
+        self.clear_deferred_render();
+        deferred
     }
 
     pub(crate) fn is_full_app_client(&self) -> bool {
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 7725954c..720b43f2 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -53,7 +53,7 @@ use crate::server::client_accept::{
 use crate::server::client_transport::ServerEvent;
 use crate::server::clients::{
     events_include_interaction, latest_app_client, render_targets, terminal_stream_client_ids,
-    ClientConnection, ClientConnectionMode,
+    ClientConnection, ClientConnectionMode, DeferredRender,
 };
 use crate::server::keybindings::{app_keybindings, apply_keybindings};
 use crate::server::notifications::{
@@ -64,6 +64,12 @@ use crate::server::socket_paths::{
 };
 use crate::server::terminal_attach::paste_payload_for_runtime;
 
+mod pane_graphics;
+
+#[cfg(test)]
+use pane_graphics::frame_pane_graphics_for_client;
+use pane_graphics::RetainedGraphicsOutcome;
+
 #[cfg(test)]
 use crate::protocol::RenderEncoding;
 #[cfg(test)]
@@ -125,6 +131,31 @@ enum LoopEvent {
     RenderRequested,
 }
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+enum RenderImpact {
+    #[default]
+    None,
+    Graphics,
+    Full,
+}
+
+impl RenderImpact {
+    fn merge(&mut self, other: Self) {
+        *self = (*self).max(other);
+    }
+}
+
+fn record_render_impact(source: &'static str, impact: RenderImpact) {
+    let event = match (source, impact) {
+        ("api_requests", RenderImpact::Graphics) => "graphics_render_cause.api_requests",
+        ("api_requests", RenderImpact::Full) => "full_render_cause.api_requests",
+        ("server_events", RenderImpact::Graphics) => "graphics_render_cause.server_events",
+        ("server_events", RenderImpact::Full) => "full_render_cause.server_events",
+        _ => return,
+    };
+    crate::render_prof::event(event);
+}
+
 fn rect_fits_frame(rect: Rect, frame: &FrameData) -> bool {
     rect.x.saturating_add(rect.width) <= frame.width
         && rect.y.saturating_add(rect.height) <= frame.height
@@ -461,6 +492,7 @@ impl HeadlessServer {
 
         let mut needs_render = true;
         let mut needs_full_render = true;
+        let mut needs_graphics_render = false;
 
         loop {
             crate::render_prof::event("loop.tick");
@@ -496,6 +528,7 @@ impl HeadlessServer {
             if self.drain_internal_events_with_forwarding() {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
                 crate::render_prof::event("full_render_cause.internal_events");
             }
             if self.app.expire_due_metadata(Instant::now()) {
@@ -505,7 +538,22 @@ impl HeadlessServer {
             }
 
             // 3. Drain API requests.
-            if self.drain_api_requests_with_shutdown_check() {
+            if self.pane_graphics_runtime_active() {
+                let api_impact = self.drain_api_requests_with_render_impact();
+                record_render_impact("api_requests", api_impact);
+                match api_impact {
+                    RenderImpact::None => {}
+                    RenderImpact::Graphics => {
+                        needs_render = true;
+                        needs_graphics_render = true;
+                    }
+                    RenderImpact::Full => {
+                        needs_render = true;
+                        needs_full_render = true;
+                        needs_graphics_render = false;
+                    }
+                }
+            } else if self.drain_api_requests_with_shutdown_check() {
                 needs_render = true;
                 needs_full_render = true;
                 crate::render_prof::event("full_render_cause.api_requests");
@@ -518,7 +566,22 @@ impl HeadlessServer {
             self.accept_client_connections()?;
 
             // 5. Drain server events from client threads.
-            if self.drain_server_events() {
+            if self.pane_graphics_runtime_active() {
+                let server_impact = self.drain_server_events_with_render_impact();
+                record_render_impact("server_events", server_impact);
+                match server_impact {
+                    RenderImpact::None => {}
+                    RenderImpact::Graphics => {
+                        needs_render = true;
+                        needs_graphics_render = true;
+                    }
+                    RenderImpact::Full => {
+                        needs_render = true;
+                        needs_full_render = true;
+                        needs_graphics_render = false;
+                    }
+                }
+            } else if self.drain_server_events() {
                 needs_render = true;
                 needs_full_render = true;
                 crate::render_prof::event("full_render_cause.server_events");
@@ -529,20 +592,25 @@ impl HeadlessServer {
             if self.handle_scheduled_tasks_headless(now, needs_render) {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
                 crate::render_prof::event("full_render_cause.scheduled_tasks");
             }
 
             if self.handle_deferred_requests_headless() {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
             }
 
             if latest_app_client(&self.clients).is_some() && self.app.ensure_default_workspace() {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
                 crate::render_prof::event("full_render_cause.default_workspace");
             }
 
+            self.cancel_inactive_pane_graphics_streams();
+
             self.drain_client_config_reload_request();
             self.stream_host_mouse_capture_mode();
 
@@ -560,8 +628,24 @@ impl HeadlessServer {
                 } else if !pty_dirty {
                     crate::render_prof::event("retained_gate.not_pty_dirty");
                 }
-                let rendered_retained =
-                    pty_dirty && !needs_full_render && self.render_retained_pty_update_and_stream();
+                let mut deferred_graphics = false;
+                let rendered_retained = if needs_graphics_render && !needs_full_render && !pty_dirty
+                {
+                    match self.render_retained_graphics_update_and_stream() {
+                        RetainedGraphicsOutcome::Sent => true,
+                        RetainedGraphicsOutcome::Deferred => {
+                            deferred_graphics = true;
+                            false
+                        }
+                        RetainedGraphicsOutcome::Fallback => false,
+                    }
+                } else {
+                    pty_dirty && !needs_full_render && self.render_retained_pty_update_and_stream()
+                };
+                if deferred_graphics {
+                    needs_render = false;
+                    continue;
+                }
                 if !rendered_retained {
                     crate::render_prof::event("full_render.invoke");
                     self.render_and_stream();
@@ -569,6 +653,7 @@ impl HeadlessServer {
                 self.app.last_render_at = Some(now);
                 needs_render = false;
                 needs_full_render = false;
+                needs_graphics_render = false;
                 continue;
             }
 
@@ -607,16 +692,47 @@ impl HeadlessServer {
                     if self.handle_internal_event_with_forwarding(ev) {
                         needs_render = true;
                         needs_full_render = true;
+                        needs_graphics_render = false;
                     }
                 }
                 LoopEvent::Api(msg) => {
-                    if self.handle_api_request_with_shutdown_check(*msg) {
+                    if self.pane_graphics_runtime_active() {
+                        let impact = self.handle_api_request_with_render_impact(*msg);
+                        record_render_impact("api_requests", impact);
+                        match impact {
+                            RenderImpact::None => {}
+                            RenderImpact::Graphics => {
+                                needs_render = true;
+                                needs_graphics_render = true;
+                            }
+                            RenderImpact::Full => {
+                                needs_render = true;
+                                needs_full_render = true;
+                                needs_graphics_render = false;
+                            }
+                        }
+                    } else if self.handle_api_request_with_shutdown_check(*msg) {
                         needs_render = true;
                         needs_full_render = true;
                     }
                 }
                 LoopEvent::ServerEvent(ev) => {
-                    if self.handle_server_event(ev) {
+                    if self.pane_graphics_runtime_active() {
+                        let impact = self.handle_server_event_with_render_impact(ev);
+                        record_render_impact("server_events", impact);
+                        match impact {
+                            RenderImpact::None => {}
+                            RenderImpact::Graphics => {
+                                needs_render = true;
+                                needs_graphics_render = true;
+                            }
+                            RenderImpact::Full => {
+                                needs_render = true;
+                                needs_full_render = true;
+                                needs_graphics_render = false;
+                            }
+                        }
+                    } else if self.handle_server_event(ev) {
                         needs_render = true;
                         needs_full_render = true;
                     }
@@ -876,6 +992,7 @@ impl HeadlessServer {
         let Some(client_id) = self.foreground_client_id else {
             self.effective_size = (MIN_COLS, MIN_ROWS);
             self.app.state.outer_terminal_focus = None;
+            self.app.state.host_cell_size = crate::kitty_graphics::HostCellSize::default();
             let server_keybindings = self.server_keybindings.clone();
             apply_keybindings(&mut self.app, &server_keybindings);
             self.sync_visible_server_config_diagnostic(false);
@@ -885,6 +1002,7 @@ impl HeadlessServer {
             self.foreground_client_id = None;
             self.effective_size = (MIN_COLS, MIN_ROWS);
             self.app.state.outer_terminal_focus = None;
+            self.app.state.host_cell_size = crate::kitty_graphics::HostCellSize::default();
             let server_keybindings = self.server_keybindings.clone();
             apply_keybindings(&mut self.app, &server_keybindings);
             self.sync_visible_server_config_diagnostic(false);
@@ -893,6 +1011,12 @@ impl HeadlessServer {
 
         let terminal_size = client.terminal_size;
         let outer_terminal_focus = client.outer_terminal_focus;
+        let host_cell_size = if self.app.state.kitty_graphics_enabled && client.cell_size.is_known()
+        {
+            client.cell_size
+        } else {
+            crate::kitty_graphics::HostCellSize::default()
+        };
         let host_terminal_theme = client.host_terminal_theme;
         let host_terminal_appearance = client.host_terminal_appearance;
         let host_terminal_appearance_explicit = client.host_terminal_appearance_explicit;
@@ -905,6 +1029,7 @@ impl HeadlessServer {
 
         self.effective_size = terminal_size;
         self.app.state.outer_terminal_focus = outer_terminal_focus;
+        self.app.state.host_cell_size = host_cell_size;
         apply_keybindings(&mut self.app, &keybindings);
         self.sync_visible_server_config_diagnostic(uses_local_keybindings);
         if outer_terminal_focus == Some(true) {
@@ -1408,7 +1533,7 @@ impl HeadlessServer {
 
     /// Drains server events from the dedicated channel.
     ///
-    /// Returns true if any input was processed (requiring a re-render).
+    /// Uses the original full-render semantics when pane graphics are dormant.
     fn drain_server_events(&mut self) -> bool {
         let mut changed = false;
         while let Ok(ev) = self.server_event_rx.try_recv() {
@@ -1417,6 +1542,15 @@ impl HeadlessServer {
         changed
     }
 
+    /// Returns the strongest render impact from the drained event batch.
+    fn drain_server_events_with_render_impact(&mut self) -> RenderImpact {
+        let mut impact = RenderImpact::None;
+        while let Ok(ev) = self.server_event_rx.try_recv() {
+            impact.merge(self.handle_server_event_with_render_impact(ev));
+        }
+        impact
+    }
+
     fn terminal_id_by_string(&self, terminal_id: &str) -> Option<crate::terminal::TerminalId> {
         self.app
             .state
@@ -2450,7 +2584,7 @@ impl HeadlessServer {
         if let Some(client) = self.clients.get_mut(&client_id) {
             if host_surface_redraw {
                 client.request_full_redraw();
-                client.render_pending = true;
+                client.defer_full_render();
             } else {
                 // Ensure semantic clients receive one post-input frame even if the
                 // semantic buffer compares equal. Terminal-ANSI clients must keep their
@@ -2765,12 +2899,7 @@ impl HeadlessServer {
                 let Some(client) = self.clients.get_mut(&client_id) else {
                     return false;
                 };
-                if client.render_pending {
-                    client.render_pending = false;
-                    true
-                } else {
-                    false
-                }
+                client.take_deferred_render() != DeferredRender::None
             }
             ServerEvent::QuitSignal => {
                 // The quit check at the top of the loop handles this.
@@ -2780,6 +2909,23 @@ impl HeadlessServer {
         }
     }
 
+    fn handle_server_event_with_render_impact(&mut self, ev: ServerEvent) -> RenderImpact {
+        let deferred_render = match &ev {
+            ServerEvent::ClientWriterDrained { client_id } => self
+                .clients
+                .get(client_id)
+                .map_or(DeferredRender::None, ClientConnection::deferred_render),
+            _ => DeferredRender::None,
+        };
+        if !self.handle_server_event(ev) {
+            return RenderImpact::None;
+        }
+        match deferred_render {
+            DeferredRender::Graphics => RenderImpact::Graphics,
+            DeferredRender::None | DeferredRender::Full => RenderImpact::Full,
+        }
+    }
+
     fn ignore_client_event_during_handoff(ev: &ServerEvent) -> bool {
         !matches!(
             ev,
@@ -2801,6 +2947,14 @@ impl HeadlessServer {
         changed
     }
 
+    fn drain_api_requests_with_render_impact(&mut self) -> RenderImpact {
+        let mut impact = RenderImpact::None;
+        while let Ok(msg) = self.app.api_rx.try_recv() {
+            impact.merge(self.handle_api_request_with_render_impact(msg));
+        }
+        impact
+    }
+
     /// Handles a single API request with shutdown awareness.
     ///
     /// Also forwards any toast/sound notifications that result from the API
@@ -2811,6 +2965,23 @@ impl HeadlessServer {
         self.handle_api_request_with_shutdown_check_inner(msg, false)
     }
 
+    fn handle_api_request_with_render_impact(
+        &mut self,
+        msg: api::ApiRequestMessage,
+    ) -> RenderImpact {
+        if matches!(
+            &msg.request.method,
+            api::schema::Method::PaneGraphicsStreamSet(_)
+        ) {
+            return self.handle_pane_graphics_stream_frame(msg);
+        }
+        if self.handle_api_request_with_shutdown_check_inner(msg, false) {
+            RenderImpact::Full
+        } else {
+            RenderImpact::None
+        }
+    }
+
     fn handle_api_request_with_shutdown_check_inner(
         &mut self,
         msg: api::ApiRequestMessage,
@@ -2884,7 +3055,16 @@ impl HeadlessServer {
             _ => {}
         }
 
-        let mut changed = metadata_expired | api::request_changes_ui(&msg.request);
+        let pane_graphics_revision_before = matches!(
+            &msg.request.method,
+            api::schema::Method::PaneGraphicsSet(_)
+                | api::schema::Method::PaneGraphicsClear(_)
+                | api::schema::Method::PaneGraphicsStreamOpen(_)
+                | api::schema::Method::PaneGraphicsStreamClose(_)
+        )
+        .then_some(self.app.state.pane_graphics_revision);
+        let mut changed = metadata_expired
+            | (pane_graphics_revision_before.is_none() && api::request_changes_ui(&msg.request));
         let skip_default_workspace = skip_default_workspace_for_request
             || matches!(
                 &msg.request.method,
@@ -2965,6 +3145,10 @@ impl HeadlessServer {
         };
         let _ = msg.respond_to.send(response);
 
+        if let Some(revision_before) = pane_graphics_revision_before {
+            changed |= revision_before != self.app.state.pane_graphics_revision;
+        }
+
         // Forward new toast state only when a client-local delivery mode is selected.
         // Herdr delivery renders the toast in-frame and must not ask clients to
         // show a terminal or system notification.
@@ -3151,8 +3335,6 @@ impl HeadlessServer {
         }
     }
 
-    /// Renders the current state to client-sized virtual buffers and streams
-    /// frames to all connected clients.
     fn render_retained_pty_update_and_stream(&mut self) -> bool {
         crate::render_prof::event("retained.attempt");
         let retained_started = crate::render_prof::timer();
@@ -3188,7 +3370,7 @@ impl HeadlessServer {
         let Some(client) = self.clients.get(client_id) else {
             retained_fallback!("client_missing");
         };
-        if client.render_pending {
+        if client.deferred_render() != DeferredRender::None {
             retained_fallback!("render_pending");
         }
         if self.app.state.kitty_graphics_enabled && !client.graphics_cache.is_empty() {
@@ -3304,7 +3486,7 @@ impl HeadlessServer {
         };
         let prepare_started = crate::render_prof::timer();
         let Some(prepared) = client.render_state.prepare_frame(frame) else {
-            client.render_pending = false;
+            client.clear_deferred_render();
             crate::render_prof::event("retained_send.skip_identical");
             crate::render_prof::duration_since("retained_send.prepare_frame", prepare_started);
             return true;
@@ -3338,14 +3520,14 @@ impl HeadlessServer {
         let send_started = crate::render_prof::timer();
         match writer.render.try_send(serialized) {
             Ok(()) => {
-                client.render_pending = false;
+                client.clear_deferred_render();
                 client.render_state.commit_sent_frame(prepared);
                 crate::render_prof::event("retained_send.sent");
                 crate::render_prof::duration_since("retained_send.try_send", send_started);
                 true
             }
             Err(std::sync::mpsc::TrySendError::Full(_)) => {
-                client.render_pending = true;
+                client.defer_full_render();
                 crate::render_prof::event("retained_send_fallback.queue_full");
                 crate::render_prof::duration_since("retained_send.try_send", send_started);
                 debug!(
@@ -3398,24 +3580,20 @@ impl HeadlessServer {
             let mut frame = match mode {
                 ClientConnectionMode::App => {
                     let render_started = crate::render_prof::timer();
-                    let (buffer, cursor) =
+                    let render_cell_size =
                         if self.app.state.kitty_graphics_enabled && cell_size.is_known() {
-                            crate::server::render_stream::render_virtual_with_runtime_registry(
-                                &mut self.app.state,
-                                &self.app.terminal_runtimes,
-                                area,
-                                is_foreground,
-                                cell_size,
-                            )
+                            cell_size
                         } else {
-                            crate::server::render_stream::render_virtual_with_runtime_registry(
-                                &mut self.app.state,
-                                &self.app.terminal_runtimes,
-                                area,
-                                is_foreground,
-                                crate::kitty_graphics::HostCellSize::default(),
-                            )
+                            crate::kitty_graphics::HostCellSize::default()
                         };
+                    let (buffer, cursor) =
+                        crate::server::render_stream::render_virtual_with_runtime_registry(
+                            &mut self.app.state,
+                            &self.app.terminal_runtimes,
+                            area,
+                            is_foreground,
+                            render_cell_size,
+                        );
                     crate::render_prof::duration_since(
                         "full_render.render_virtual",
                         render_started,
@@ -3524,7 +3702,7 @@ impl HeadlessServer {
             let has_graphics = !frame.graphics.is_empty();
             let prepare_started = crate::render_prof::timer();
             let Some(mut prepared) = client.render_state.prepare_frame(frame) else {
-                client.render_pending = false;
+                client.clear_deferred_render();
                 crate::render_prof::event("full_render.skip_identical");
                 crate::render_prof::duration_since("full_render.prepare_frame", prepare_started);
                 continue;
@@ -3557,7 +3735,7 @@ impl HeadlessServer {
                     let Some(text_only_prepared) =
                         client.render_state.prepare_frame(text_only_frame)
                     else {
-                        client.render_pending = false;
+                        client.clear_deferred_render();
                         crate::render_prof::event("full_render.skip_identical_text_only");
                         crate::render_prof::duration_since(
                             "full_render.serialize",
@@ -3605,7 +3783,7 @@ impl HeadlessServer {
             let send_started = crate::render_prof::timer();
             match writer.render.try_send(serialized) {
                 Ok(()) => {
-                    client.render_pending = false;
+                    client.clear_deferred_render();
                     if commit_graphics_cache {
                         client.graphics_cache = next_graphics_cache;
                         client.graphics_surface_reset_pending = false;
@@ -3615,7 +3793,7 @@ impl HeadlessServer {
                     crate::render_prof::duration_since("full_render.try_send", send_started);
                 }
                 Err(std::sync::mpsc::TrySendError::Full(_)) => {
-                    client.render_pending = true;
+                    client.defer_full_render();
                     deferred_frame = true;
                     crate::render_prof::event("full_render.queue_full");
                     crate::render_prof::duration_since("full_render.try_send", send_started);
@@ -4229,6 +4407,9 @@ mod tests {
     use crate::app::AppState;
     use crate::protocol::CursorState;
 
+    #[path = "pane_graphics.rs"]
+    mod pane_graphics_tests;
+
     fn test_headless_server() -> HeadlessServer {
         test_headless_server_with_event_hub(api::EventHub::default())
     }
@@ -7333,11 +7514,17 @@ next_tab = ""
             client_id: 1,
             data: b"\x1b[I".to_vec(),
         }));
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
 
         server.render_and_stream();
 
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::ReloadSoundConfig
@@ -7356,7 +7543,10 @@ next_tab = ""
         assert!(server.handle_server_event(ServerEvent::ClientWriterDrained { client_id: 1 }));
         server.render_and_stream();
 
-        assert!(!server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::None
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::Frame(_)
@@ -7434,7 +7624,10 @@ next_tab = ""
             data: b"\x1b[I".to_vec(),
         }));
 
-        assert!(!server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::None
+        );
         assert!(!server.app.full_redraw_pending);
         assert_eq!(server.clients[&1].outer_terminal_focus, Some(true));
         assert_eq!(server.app.state.outer_terminal_focus, Some(true));
@@ -7510,7 +7703,10 @@ next_tab = ""
         server.foreground_client_id = Some(1);
 
         server.render_and_stream();
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::ReloadSoundConfig
@@ -7533,7 +7729,10 @@ next_tab = ""
                 .unwrap(),
             1
         );
-        assert!(!server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::None
+        );
     }
 
     #[test]
@@ -7984,7 +8183,10 @@ next_tab = ""
         server.render_and_stream();
 
         assert!(server.app.full_redraw_pending);
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::ReloadSoundConfig
diff --git a/src/server/headless/pane_graphics.rs b/src/server/headless/pane_graphics.rs
new file mode 100644
index 00000000..18174f45
--- /dev/null
+++ b/src/server/headless/pane_graphics.rs
@@ -0,0 +1,200 @@
+use tracing::warn;
+
+use super::{HeadlessServer, RenderImpact};
+use crate::api;
+use crate::protocol::{ServerMessage, MAX_GRAPHICS_FRAME_SIZE};
+use crate::server::clients::{render_targets, ClientConnectionMode, DeferredRender};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum RetainedGraphicsOutcome {
+    Sent,
+    Deferred,
+    Fallback,
+}
+
+pub(super) fn frame_pane_graphics_for_client(bytes: Vec<u8>) -> Vec<u8> {
+    if bytes.is_empty() {
+        return bytes;
+    }
+    let mut framed = Vec::with_capacity(bytes.len() + 4);
+    framed.extend_from_slice(b"\x1b7");
+    framed.extend_from_slice(&bytes);
+    framed.extend_from_slice(b"\x1b8");
+    framed
+}
+
+impl HeadlessServer {
+    pub(super) fn pane_graphics_runtime_active(&self) -> bool {
+        !self.app.state.pane_graphics_layers.is_empty()
+            || !self.app.state.pane_graphics_streams.is_empty()
+    }
+
+    pub(super) fn cancel_inactive_pane_graphics_streams(&self) {
+        api::cancel_inactive_pane_graphics_streams(|owner| {
+            self.app
+                .state
+                .pane_graphics_streams
+                .values()
+                .any(|active_owner| active_owner == owner)
+        });
+    }
+
+    pub(super) fn handle_pane_graphics_stream_frame(
+        &mut self,
+        msg: api::ApiRequestMessage,
+    ) -> RenderImpact {
+        if self.shutting_down {
+            let _ = self.handle_api_request_with_shutdown_check_inner(msg, false);
+            return RenderImpact::None;
+        }
+
+        let internal_changed = self.drain_all_internal_events_with_forwarding();
+        let response = self
+            .app
+            .handle_api_request_after_internal_events_drained(msg.request);
+        let succeeded = serde_json::from_str::<api::schema::SuccessResponse>(&response).is_ok();
+        let _ = msg.respond_to.send(response);
+
+        if internal_changed {
+            RenderImpact::Full
+        } else if succeeded {
+            RenderImpact::Graphics
+        } else {
+            RenderImpact::None
+        }
+    }
+
+    pub(super) fn render_retained_graphics_update_and_stream(&mut self) -> RetainedGraphicsOutcome {
+        crate::render_prof::event("retained_graphics.attempt");
+        if self.app.full_redraw_pending {
+            crate::render_prof::event("retained_graphics_fallback.full_redraw_pending");
+            return RetainedGraphicsOutcome::Fallback;
+        }
+
+        let render_targets = render_targets(&self.clients, self.foreground_client_id);
+        let mut app_view_size = None;
+        for (_, terminal_size, _, _, mode) in &render_targets {
+            if !matches!(mode, ClientConnectionMode::App) {
+                continue;
+            }
+            if app_view_size.is_some_and(|size| size != *terminal_size) {
+                crate::render_prof::event("retained_graphics_fallback.mixed_app_geometry");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+            app_view_size = Some(*terminal_size);
+        }
+        let mut deferred = false;
+        let mut prepared = Vec::new();
+
+        for (client_id, (cols, rows), cell_size, _is_foreground, mode) in render_targets {
+            if !matches!(mode, ClientConnectionMode::App) {
+                continue;
+            }
+            let Some(client) = self.clients.get_mut(&client_id) else {
+                crate::render_prof::event("retained_graphics_fallback.client_missing");
+                return RetainedGraphicsOutcome::Fallback;
+            };
+            if client.deferred_render() != DeferredRender::None {
+                deferred = true;
+                continue;
+            }
+            if client.graphics_surface_reset_pending || !cell_size.is_known() {
+                crate::render_prof::event("retained_graphics_fallback.client_state");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+            let Some(last_frame) = client.render_state.last_frame() else {
+                crate::render_prof::event("retained_graphics_fallback.no_last_frame");
+                return RetainedGraphicsOutcome::Fallback;
+            };
+            if last_frame.width != cols || last_frame.height != rows {
+                crate::render_prof::event("retained_graphics_fallback.frame_size_mismatch");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+            if client.writer.is_none() {
+                crate::render_prof::event("retained_graphics_fallback.writer_missing");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+
+            let mut next_graphics_cache = client.graphics_cache.clone();
+            let encode_started = crate::render_prof::timer();
+            let bytes =
+                frame_pane_graphics_for_client(crate::kitty_graphics::encode_local_pane_graphics(
+                    &self.app.state,
+                    &self.app.terminal_runtimes,
+                    cell_size,
+                    &mut next_graphics_cache,
+                ));
+            crate::render_prof::duration_since("retained_graphics.graphics_encode", encode_started);
+            if bytes.len() > MAX_GRAPHICS_FRAME_SIZE {
+                warn!(
+                    client_id,
+                    graphics_bytes = bytes.len(),
+                    max = MAX_GRAPHICS_FRAME_SIZE,
+                    "dropping oversized retained graphics payload"
+                );
+                continue;
+            }
+            let serialized = if bytes.is_empty() {
+                None
+            } else {
+                let serialize_started = crate::render_prof::timer();
+                let framed = match Self::frame_server_message_with_max(
+                    &ServerMessage::Graphics { bytes },
+                    MAX_GRAPHICS_FRAME_SIZE,
+                ) {
+                    Ok(framed) => framed,
+                    Err(err) => {
+                        warn!(client_id, err = %err, "failed to serialize retained graphics");
+                        return RetainedGraphicsOutcome::Fallback;
+                    }
+                };
+                crate::render_prof::duration_since(
+                    "retained_graphics.serialize",
+                    serialize_started,
+                );
+                Some(framed)
+            };
+            prepared.push((client_id, serialized, next_graphics_cache));
+        }
+
+        let mut broken_clients = Vec::new();
+        for (client_id, serialized, next_graphics_cache) in prepared {
+            let Some(client) = self.clients.get_mut(&client_id) else {
+                continue;
+            };
+            let Some(serialized) = serialized else {
+                client.graphics_cache = next_graphics_cache;
+                client.clear_deferred_render();
+                continue;
+            };
+            let Some(writer) = client.writer.as_ref().cloned() else {
+                broken_clients.push(client_id);
+                continue;
+            };
+            match writer.render.try_send(serialized) {
+                Ok(()) => {
+                    client.graphics_cache = next_graphics_cache;
+                    client.clear_deferred_render();
+                    crate::render_prof::event("retained_graphics.sent");
+                }
+                Err(std::sync::mpsc::TrySendError::Full(_)) => {
+                    client.defer_pane_graphics_render();
+                    deferred = true;
+                    crate::render_prof::event("retained_graphics.deferred");
+                }
+                Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
+                    broken_clients.push(client_id);
+                }
+            }
+        }
+        for client_id in broken_clients {
+            self.remove_client_and_resize_if_needed(client_id);
+        }
+
+        if deferred {
+            RetainedGraphicsOutcome::Deferred
+        } else {
+            RetainedGraphicsOutcome::Sent
+        }
+    }
+}
diff --git a/src/server/headless/tests/pane_graphics.rs b/src/server/headless/tests/pane_graphics.rs
new file mode 100644
index 00000000..627ffc78
--- /dev/null
+++ b/src/server/headless/tests/pane_graphics.rs
@@ -0,0 +1,357 @@
+use super::*;
+
+#[test]
+fn frames_preserve_cursor_without_changing_ordinary_messages() {
+    assert_eq!(frame_pane_graphics_for_client(Vec::new()), Vec::<u8>::new());
+    assert_eq!(
+        frame_pane_graphics_for_client(b"graphics".to_vec()),
+        b"\x1b7graphics\x1b8"
+    );
+}
+
+fn enable_graphics_and_render(
+    server: &mut HeadlessServer,
+    client_rx: &std::sync::mpsc::Receiver<Vec<u8>>,
+) -> FrameData {
+    server.app.state.kitty_graphics_enabled = true;
+    server.clients.get_mut(&1).unwrap().cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    server.render_and_stream();
+    read_server_frame(
+        client_rx
+            .recv_timeout(Duration::from_millis(100))
+            .expect("initial frame"),
+    )
+}
+
+fn set_graphics_layer(server: &mut HeadlessServer, pane_id: crate::layout::PaneId, data: Vec<u8>) {
+    server.app.state.pane_graphics_layers.insert(
+        pane_id,
+        crate::app::state::PaneGraphicsLayer::new(
+            api::schema::PaneGraphicsFormat::Png,
+            1,
+            1,
+            data,
+            api::schema::PaneGraphicsPlacementParams::default(),
+        ),
+    );
+}
+
+fn fill_render_lane(server: &HeadlessServer) {
+    let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
+        .expect("dummy frame");
+    server
+        .clients
+        .get(&1)
+        .unwrap()
+        .writer
+        .as_ref()
+        .unwrap()
+        .render
+        .try_send(queued)
+        .expect("pre-fill render lane");
+}
+
+fn stream_set_message(
+    id: &str,
+    pane_id: &str,
+    owner: &str,
+    data: Vec<u8>,
+) -> (api::ApiRequestMessage, std::sync::mpsc::Receiver<String>) {
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    (
+        api::ApiRequestMessage {
+            request: api::schema::Request {
+                id: id.into(),
+                method: api::schema::Method::PaneGraphicsStreamSet(
+                    api::schema::PaneGraphicsSetParams {
+                        pane_id: pane_id.into(),
+                        owner: owner.into(),
+                        format: api::schema::PaneGraphicsFormat::Png,
+                        image_width: 1,
+                        image_height: 1,
+                        data: Some(data),
+                        data_base64: String::new(),
+                        placement: api::schema::PaneGraphicsPlacementParams::default(),
+                    },
+                ),
+            },
+            respond_to,
+            response_write_complete: None,
+        },
+        response_rx,
+    )
+}
+
+#[tokio::test]
+async fn retained_update_sends_only_graphics_message() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    let baseline = enable_graphics_and_render(&mut server, &client_rx);
+    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3]);
+
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Sent
+    );
+    match read_server_message(
+        client_rx
+            .recv_timeout(Duration::from_millis(100))
+            .expect("graphics-only update"),
+    ) {
+        ServerMessage::Graphics { bytes } => {
+            assert!(bytes.windows(3).any(|window| window == b"\x1b_G"));
+        }
+        other => panic!("expected graphics-only message, got {other:?}"),
+    }
+    assert_frame_data_eq(
+        server
+            .clients
+            .get(&1)
+            .unwrap()
+            .render_state
+            .last_frame()
+            .expect("semantic baseline"),
+        &baseline,
+    );
+}
+
+#[tokio::test]
+async fn retained_update_defers_on_full_render_lane() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    let _ = enable_graphics_and_render(&mut server, &client_rx);
+    fill_render_lane(&server);
+    set_graphics_layer(&mut server, pane_id, vec![4, 5, 6]);
+
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Deferred
+    );
+    let client = server.clients.get(&1).unwrap();
+    assert_eq!(client.deferred_render(), DeferredRender::Graphics);
+    assert!(matches!(
+        read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
+        ServerMessage::ReloadSoundConfig
+    ));
+    assert_eq!(
+        server.handle_server_event_with_render_impact(ServerEvent::ClientWriterDrained {
+            client_id: 1
+        }),
+        RenderImpact::Graphics
+    );
+}
+
+#[tokio::test]
+async fn retained_update_does_not_downgrade_pending_full_render() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    let _ = enable_graphics_and_render(&mut server, &client_rx);
+    fill_render_lane(&server);
+    let client = server.clients.get_mut(&1).unwrap();
+    client.request_full_redraw();
+    server.render_and_stream();
+    assert_eq!(
+        server.clients.get(&1).unwrap().deferred_render(),
+        DeferredRender::Full
+    );
+
+    set_graphics_layer(&mut server, pane_id, vec![7, 8, 9]);
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Deferred
+    );
+    assert_eq!(
+        server.clients.get(&1).unwrap().deferred_render(),
+        DeferredRender::Full
+    );
+    assert_eq!(
+        server.handle_server_event_with_render_impact(ServerEvent::ClientWriterDrained {
+            client_id: 1
+        }),
+        RenderImpact::Full
+    );
+}
+
+#[tokio::test]
+async fn retained_update_falls_back_for_mixed_app_geometry() {
+    let (mut server, client_rx, _pane_id) = retained_test_server(b"aaaa");
+    let _ = enable_graphics_and_render(&mut server, &client_rx);
+
+    let (writer, _control_rx, _render_rx) = test_client_writer();
+    server.clients.insert(
+        2,
+        ClientConnection::new(
+            (60, 20),
+            crate::kitty_graphics::HostCellSize {
+                width_px: 10,
+                height_px: 20,
+            },
+            crate::terminal_theme::TerminalTheme::default(),
+            None,
+            2,
+            RenderEncoding::SemanticFrame,
+            Some(writer),
+        ),
+    );
+
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Fallback
+    );
+}
+
+#[test]
+fn stream_set_has_graphics_only_render_impact() {
+    let mut server = test_headless_server();
+    let workspace = crate::workspace::Workspace::test_new("graphics");
+    let pane_id = workspace.tabs[0].root_pane;
+    let public_pane_id = format!("{}:p1", workspace.id);
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+    server.app.state.kitty_graphics_enabled = true;
+    server
+        .app
+        .state
+        .pane_graphics_streams
+        .insert(pane_id, "owner-a".into());
+
+    let (request, response_rx) =
+        stream_set_message("wrong-owner", &public_pane_id, "owner-b", vec![1, 2, 3]);
+    assert_eq!(
+        server.handle_api_request_with_render_impact(request),
+        RenderImpact::None
+    );
+    assert!(serde_json::from_str::<api::schema::ErrorResponse>(
+        &response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap()
+    )
+    .is_ok());
+
+    let (request, response_rx) =
+        stream_set_message("stream-frame", &public_pane_id, "owner-a", vec![1, 2, 3]);
+    assert_eq!(
+        server.handle_api_request_with_render_impact(request),
+        RenderImpact::Graphics
+    );
+    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
+        &response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap()
+    )
+    .is_ok());
+
+    server
+        .app
+        .event_tx
+        .try_send(AppEvent::UpdateReady {
+            version: "9.9.9".into(),
+            install_command: "herdr update".into(),
+        })
+        .unwrap();
+    let (request, _response_rx) = stream_set_message(
+        "stream-frame-with-internal-event",
+        &public_pane_id,
+        "owner-a",
+        vec![4, 5, 6],
+    );
+    assert_eq!(
+        server.handle_api_request_with_render_impact(request),
+        RenderImpact::Full
+    );
+
+    server.app.state.pane_graphics_streams.clear();
+    let (respond_to, _response_rx) = std::sync::mpsc::channel();
+    let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "direct-frame".into(),
+            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
+                pane_id: public_pane_id,
+                owner: String::new(),
+                format: api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data: Some(vec![1, 2, 3]),
+                data_base64: String::new(),
+                placement: api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        },
+        respond_to,
+        response_write_complete: None,
+    });
+    assert_eq!(impact, RenderImpact::Full);
+}
+
+#[test]
+fn rejected_or_stale_requests_do_not_schedule_rendering() {
+    let mut server = test_headless_server();
+    let workspace = crate::workspace::Workspace::test_new("graphics");
+    let pane_id = workspace.tabs[0].root_pane;
+    let public_pane_id = format!("{}:p1", workspace.id);
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    let changed = server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "disabled-set".into(),
+            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: String::new(),
+                format: api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data: Some(vec![1, 2, 3]),
+                data_base64: String::new(),
+                placement: api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        },
+        respond_to,
+        response_write_complete: None,
+    });
+    assert!(!changed);
+    let response = response_rx
+        .recv_timeout(Duration::from_millis(100))
+        .unwrap();
+    assert_eq!(
+        serde_json::from_str::<api::schema::ErrorResponse>(&response)
+            .unwrap()
+            .error
+            .code,
+        "feature_disabled"
+    );
+
+    server.app.state.kitty_graphics_enabled = true;
+    server
+        .app
+        .state
+        .pane_graphics_streams
+        .insert(pane_id, "current-owner".into());
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "stale-close".into(),
+            method: api::schema::Method::PaneGraphicsStreamClose(
+                api::schema::PaneGraphicsStreamParams {
+                    pane_id: public_pane_id,
+                    owner: "stale-owner".into(),
+                },
+            ),
+        },
+        respond_to,
+        response_write_complete: None,
+    });
+    assert_eq!(impact, RenderImpact::None);
+    assert_eq!(
+        server.app.state.pane_graphics_streams.get(&pane_id),
+        Some(&"current-owner".to_string())
+    );
+    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
+        &response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap()
+    )
+    .is_ok());
+}