6 files changed, 1752 insertions, 416 deletions

diff --git a/src/server/alt_screen_read.rs b/src/server/alt_screen_read.rs
index 240e96de..425d4a41 100644
--- a/src/server/alt_screen_read.rs
+++ b/src/server/alt_screen_read.rs
@@ -373,7 +373,11 @@ fn send_wheel(
     let column = snapshot.cols.saturating_sub(1) / 2;
     let row = u16::try_from(snapshot.rows.len().saturating_sub(1) / 2).unwrap_or(0);
     let event = runtime
-        .encode_mouse_wheel(kind, column, row, KeyModifiers::empty())
+        .encode_mouse_wheel(
+            kind,
+            crate::input::mouse::Position::Cell { column, row },
+            KeyModifiers::empty(),
+        )
         .ok_or(())?;
     let mut bytes = Vec::with_capacity(event.len().saturating_mul(events));
     for _ in 0..events {
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 2675b4c3..b5023b97 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -41,6 +41,8 @@ const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4);
 const MAX_INPUT_PAYLOAD: usize = 1024 * 1024; // 1 MB
 /// Maximum structured input events accepted in one client message.
 const MAX_INPUT_EVENT_BATCH: usize = 4096;
+/// Maximum encoded mouse report accepted with pixel geometry.
+const MAX_PIXEL_MOUSE_PAYLOAD: usize = 128;
 
 /// Channels owned by the server side of a client writer thread.
 #[derive(Clone, Debug)]
@@ -51,115 +53,99 @@ pub(crate) struct ClientWriter {
     pub(crate) render: ClientRenderWriter,
 }
 
-#[cfg(test)]
 impl ClientWriter {
+    pub(crate) fn replace_with_cleanup(&self, data: Vec<u8>) {
+        self.render.queue.replace_with_cleanup(data);
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_fill_render(&self, data: Vec<u8>) {
+        self.render.try_send(data).unwrap();
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_close(&self) {
+        self.render.queue.close_writer();
+    }
+
+    #[cfg(test)]
     pub(crate) fn test_channel(
         control: std::sync::mpsc::Sender<Vec<u8>>,
         render: std::sync::mpsc::SyncSender<Vec<u8>>,
     ) -> Self {
-        Self {
-            control: ClientControlWriter {
-                target: ClientControlTarget::Channel(control),
-            },
-            render: ClientRenderWriter {
-                target: ClientRenderTarget::Channel(render),
-            },
-        }
+        let queue = ClientWriterQueue::new();
+        let drain = queue.clone();
+        let control_writer = ClientControlWriter::queue(queue.clone());
+        let mut render_writer = ClientRenderWriter::queue(queue);
+        render_writer.test_render = Some(render.clone());
+        let writer = Self {
+            control: control_writer,
+            render: render_writer,
+        };
+        std::thread::spawn(move || {
+            while let Some(item) = drain.recv() {
+                let sent = match item {
+                    ClientWriteItem::Control(data) => control.send(data).is_ok(),
+                    ClientWriteItem::Render(data) => render.send(data).is_ok(),
+                };
+                if !sent {
+                    break;
+                }
+            }
+            drain.close_writer();
+        });
+        writer
     }
 }
 
 #[derive(Debug)]
 pub(crate) struct ClientControlWriter {
-    target: ClientControlTarget,
-}
-
-#[derive(Debug)]
-enum ClientControlTarget {
-    Queue(Arc<ClientWriterQueue>),
+    queue: Arc<ClientWriterQueue>,
     #[cfg(test)]
-    Channel(std::sync::mpsc::Sender<Vec<u8>>),
+    test_render: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
 }
 
 #[derive(Debug)]
 pub(crate) struct ClientRenderWriter {
-    target: ClientRenderTarget,
-}
-
-#[derive(Debug)]
-enum ClientRenderTarget {
-    Queue(Arc<ClientWriterQueue>),
+    queue: Arc<ClientWriterQueue>,
     #[cfg(test)]
-    Channel(std::sync::mpsc::SyncSender<Vec<u8>>),
+    test_render: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
 }
 
-impl Clone for ClientControlWriter {
-    fn clone(&self) -> Self {
-        match &self.target {
-            ClientControlTarget::Queue(queue) => {
-                queue.add_sender();
+macro_rules! writer_handle {
+    ($type:ty) => {
+        impl Clone for $type {
+            fn clone(&self) -> Self {
+                self.queue.add_sender();
                 Self {
-                    target: ClientControlTarget::Queue(queue.clone()),
+                    queue: self.queue.clone(),
+                    #[cfg(test)]
+                    test_render: self.test_render.clone(),
                 }
             }
-            #[cfg(test)]
-            ClientControlTarget::Channel(sender) => Self {
-                target: ClientControlTarget::Channel(sender.clone()),
-            },
         }
-    }
-}
-
-impl Drop for ClientControlWriter {
-    fn drop(&mut self) {
-        match &self.target {
-            ClientControlTarget::Queue(queue) => queue.remove_sender(),
-            #[cfg(test)]
-            ClientControlTarget::Channel(_) => {}
+        impl Drop for $type {
+            fn drop(&mut self) {
+                self.queue.remove_sender();
+            }
         }
-    }
+    };
 }
+writer_handle!(ClientControlWriter);
+writer_handle!(ClientRenderWriter);
 
 impl ClientControlWriter {
     fn queue(queue: Arc<ClientWriterQueue>) -> Self {
         queue.add_sender();
         Self {
-            target: ClientControlTarget::Queue(queue),
-        }
-    }
-
-    pub(crate) fn send(&self, data: Vec<u8>) -> Result<(), SendError<Vec<u8>>> {
-        match &self.target {
-            ClientControlTarget::Queue(queue) => queue.send_control(data),
-            #[cfg(test)]
-            ClientControlTarget::Channel(sender) => sender.send(data),
-        }
-    }
-}
-
-impl Clone for ClientRenderWriter {
-    fn clone(&self) -> Self {
-        match &self.target {
-            ClientRenderTarget::Queue(queue) => {
-                queue.add_sender();
-                Self {
-                    target: ClientRenderTarget::Queue(queue.clone()),
-                }
-            }
+            queue,
             #[cfg(test)]
-            ClientRenderTarget::Channel(sender) => Self {
-                target: ClientRenderTarget::Channel(sender.clone()),
-            },
+            test_render: None,
         }
     }
-}
 
-impl Drop for ClientRenderWriter {
-    fn drop(&mut self) {
-        match &self.target {
-            ClientRenderTarget::Queue(queue) => queue.remove_sender(),
-            #[cfg(test)]
-            ClientRenderTarget::Channel(_) => {}
-        }
+    pub(crate) fn send(&self, data: Vec<u8>) -> Result<(), SendError<Vec<u8>>> {
+        self.queue.send_control(data)
     }
 }
 
@@ -167,16 +153,22 @@ impl ClientRenderWriter {
     fn queue(queue: Arc<ClientWriterQueue>) -> Self {
         queue.add_sender();
         Self {
-            target: ClientRenderTarget::Queue(queue),
+            queue,
+            #[cfg(test)]
+            test_render: None,
         }
     }
 
     pub(crate) fn try_send(&self, data: Vec<u8>) -> Result<(), TrySendError<Vec<u8>>> {
-        match &self.target {
-            ClientRenderTarget::Queue(queue) => queue.try_send_render(data),
-            #[cfg(test)]
-            ClientRenderTarget::Channel(sender) => sender.try_send(data),
+        #[cfg(test)]
+        if let Some(sender) = &self.test_render {
+            return sender.try_send(data);
         }
+        self.queue.try_send_render(data)
+    }
+
+    pub(crate) fn send_ordered(&self, data: Vec<u8>) -> Result<(), TrySendError<Vec<u8>>> {
+        self.queue.send_ordered(data)
     }
 }
 
@@ -189,6 +181,7 @@ struct ClientWriterQueue {
 #[derive(Debug, Default)]
 struct ClientWriterQueueState {
     control: VecDeque<Vec<u8>>,
+    ordered: VecDeque<Vec<u8>>,
     render: Option<Vec<u8>>,
     senders: usize,
     writer_alive: bool,
@@ -245,16 +238,45 @@ impl ClientWriterQueue {
         Ok(())
     }
 
+    fn send_ordered(&self, data: Vec<u8>) -> Result<(), TrySendError<Vec<u8>>> {
+        let mut state = self.lock_state();
+        if !state.writer_alive {
+            return Err(TrySendError::Disconnected(data));
+        }
+        if !state.ordered.is_empty() {
+            return Err(TrySendError::Full(data));
+        }
+        if let Some(older) = state.render.take() {
+            state.ordered.push_back(older);
+        }
+        state.ordered.push_back(data);
+        self.ready.notify_one();
+        Ok(())
+    }
+
+    fn replace_with_cleanup(&self, data: Vec<u8>) {
+        let mut state = self.lock_state();
+        state.render = None;
+        state.ordered.clear();
+        if state.writer_alive {
+            state.control.push_back(data);
+            self.ready.notify_one();
+        }
+    }
+
     fn recv(&self) -> Option<ClientWriteItem> {
         let mut state = self.lock_state();
         loop {
             if let Some(data) = state.control.pop_front() {
                 return Some(ClientWriteItem::Control(data));
             }
-            if let Some(data) = state.render.take() {
+            if let Some(data) = state.ordered.pop_front() {
                 self.ready.notify_one();
                 return Some(ClientWriteItem::Render(data));
             }
+            if let Some(data) = state.render.take() {
+                return Some(ClientWriteItem::Render(data));
+            }
             if state.senders == 0 {
                 return None;
             }
@@ -268,6 +290,8 @@ impl ClientWriterQueue {
     fn close_writer(&self) {
         let mut state = self.lock_state();
         state.writer_alive = false;
+        state.render = None;
+        state.ordered.clear();
         self.ready.notify_all();
     }
 
@@ -291,10 +315,29 @@ pub(crate) enum ServerEvent {
         render_encoding: RenderEncoding,
         keybindings: Option<Box<crate::config::LiveKeybindConfig>>,
         direct_attach_requested: bool,
+        direct_graphics: bool,
         writer: ClientWriter,
     },
     /// A client sent an input message.
     ClientInput { client_id: u64, data: Vec<u8> },
+    /// A client reported the one armed Kitty regular-file response.
+    GraphicsTransmissionResult {
+        client_id: u64,
+        transfer_id: u64,
+        image_id: u32,
+        success: bool,
+    },
+    GraphicsTransmissionStarted {
+        client_id: u64,
+        transfer_id: u64,
+        image_id: u32,
+    },
+    /// One confirmed SGR pixel report with client read-time geometry.
+    ClientInputPixels {
+        client_id: u64,
+        data: Vec<u8>,
+        geometry: crate::input::mouse::HostGeometry,
+    },
     /// A client sent structured input events.
     ClientInputEvents {
         client_id: u64,
@@ -515,6 +558,7 @@ pub(crate) fn handle_client_handshake(
         render_encoding,
         keybindings,
         direct_attach_requested,
+        direct_graphics,
     ) = match hello {
         ClientMessage::Hello {
             version,
@@ -564,6 +608,7 @@ pub(crate) fn handle_client_handshake(
                 requested_encoding,
                 keybindings,
                 launch_mode == ClientLaunchMode::TerminalAttach,
+                launch_mode == ClientLaunchMode::AppDirectGraphics,
             )
         }
         _ => {
@@ -618,6 +663,7 @@ pub(crate) fn handle_client_handshake(
         render_encoding,
         keybindings,
         direct_attach_requested,
+        direct_graphics,
         writer,
     });
 
@@ -633,19 +679,17 @@ fn client_writer_loop(
     server_event_tx: mpsc::Sender<ServerEvent>,
 ) {
     while let Some(item) = writer_queue.recv() {
-        match item {
-            ClientWriteItem::Control(data) => {
-                if !write_framed_bytes(&mut stream, &data) {
-                    break;
-                }
-            }
+        let written = match item {
+            ClientWriteItem::Control(data) => write_framed_bytes(&mut stream, &data),
             ClientWriteItem::Render(data) => {
                 let _ =
                     server_event_tx.blocking_send(ServerEvent::ClientWriterDrained { client_id });
-                if !write_framed_bytes(&mut stream, &data) {
-                    break;
-                }
+                write_framed_bytes(&mut stream, &data)
             }
+        };
+        if !written {
+            let _ = server_event_tx.blocking_send(ServerEvent::ClientDisconnected { client_id });
+            break;
         }
     }
     writer_queue.close_writer();
@@ -728,6 +772,47 @@ fn client_read_loop(
                     ServerEvent::ClientInput { client_id, data }
                 }
             }
+            ClientMessage::InputPixels {
+                data,
+                cols,
+                rows,
+                width_px,
+                height_px,
+            } => {
+                let Some(geometry) =
+                    crate::input::mouse::HostGeometry::new(cols, rows, width_px, height_px)
+                else {
+                    warn!(
+                        client_id,
+                        cols,
+                        rows,
+                        width_px,
+                        height_px,
+                        "invalid pixel mouse geometry from client, closing"
+                    );
+                    let _ = server_event_tx
+                        .blocking_send(ServerEvent::ClientDisconnected { client_id });
+                    break;
+                };
+                if data.len() > MAX_PIXEL_MOUSE_PAYLOAD
+                    || crate::input::mouse::parse_report(&data).is_none()
+                {
+                    warn!(
+                        client_id,
+                        size = data.len(),
+                        max = MAX_PIXEL_MOUSE_PAYLOAD,
+                        "invalid pixel mouse report from client, closing"
+                    );
+                    let _ = server_event_tx
+                        .blocking_send(ServerEvent::ClientDisconnected { client_id });
+                    break;
+                }
+                ServerEvent::ClientInputPixels {
+                    client_id,
+                    data,
+                    geometry,
+                }
+            }
             ClientMessage::InputEvents { events } => match input_event_limit(&events) {
                 InputEventLimit::WithinLimits => {
                     ServerEvent::ClientInputEvents { client_id, events }
@@ -777,6 +862,24 @@ fn client_read_loop(
                     takeover,
                 }
             }
+            ClientMessage::GraphicsTransmissionResult {
+                transfer_id,
+                image_id,
+                success,
+            } => ServerEvent::GraphicsTransmissionResult {
+                client_id,
+                transfer_id,
+                image_id,
+                success,
+            },
+            ClientMessage::GraphicsTransmissionStarted {
+                transfer_id,
+                image_id,
+            } => ServerEvent::GraphicsTransmissionStarted {
+                client_id,
+                transfer_id,
+                image_id,
+            },
             ClientMessage::ClipboardImage { extension, data } => {
                 if data.len() > MAX_CLIPBOARD_IMAGE_PAYLOAD {
                     warn!(
@@ -948,6 +1051,30 @@ mod tests {
     }
 
     #[test]
+    fn ordered_direct_follows_older_render_and_stays_bounded() {
+        let (writer, queue) = test_queue_writer();
+        writer.render.try_send(b"old".to_vec()).unwrap();
+        writer.render.send_ordered(b"direct".to_vec()).unwrap();
+        assert!(matches!(
+            writer.render.send_ordered(b"second".to_vec()),
+            Err(TrySendError::Full(_))
+        ));
+        writer.render.try_send(b"new".to_vec()).unwrap();
+
+        for expected in [b"old".as_slice(), b"direct", b"new"] {
+            assert_eq!(
+                queue.recv(),
+                Some(ClientWriteItem::Render(expected.to_vec()))
+            );
+        }
+        queue.close_writer();
+        assert!(matches!(
+            writer.render.send_ordered(b"closed".to_vec()),
+            Err(TrySendError::Disconnected(_))
+        ));
+    }
+
+    #[test]
     fn client_writer_prioritizes_control_and_reports_render_drain() {
         let (mut client_stream, server_stream, _path) = local_stream_pair("client-writer-priority");
         let (writer, queue) = test_queue_writer();
@@ -1203,6 +1330,7 @@ new_tab = "ctrl+notakey"
                 render_encoding,
                 keybindings,
                 direct_attach_requested,
+                direct_graphics,
                 writer,
             } => {
                 assert_eq!(client_id, 42);
@@ -1211,6 +1339,7 @@ new_tab = "ctrl+notakey"
                 assert_eq!(render_encoding, RenderEncoding::TerminalAnsi);
                 assert!(keybindings.is_none());
                 assert!(!direct_attach_requested);
+                assert!(!direct_graphics);
                 drop(writer);
             }
             other => panic!("expected ClientConnected, got {other:?}"),
@@ -1395,6 +1524,76 @@ new_tab = "ctrl+notakey"
     }
 
     #[test]
+    fn client_read_loop_disconnects_invalid_pixel_mouse_geometry() {
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-read-invalid-pixel-geometry");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::InputPixels {
+                data: b"\x1b[<35;1;1M".to_vec(),
+                cols: 0,
+                rows: 24,
+                width_px: 800,
+                height_px: 480,
+            },
+        )
+        .expect("write invalid pixel geometry");
+
+        assert!(matches!(
+            recv_server_event(&mut server_event_rx, "invalid pixel geometry disconnect"),
+            ServerEvent::ClientDisconnected { client_id: 7 }
+        ));
+        drop(client_stream);
+        should_quit.store(true, Ordering::Release);
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
+    fn client_read_loop_disconnects_invalid_pixel_mouse_report() {
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-read-invalid-pixel-report");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::InputPixels {
+                data: vec![b'x'; MAX_PIXEL_MOUSE_PAYLOAD + 1],
+                cols: 80,
+                rows: 24,
+                width_px: 800,
+                height_px: 480,
+            },
+        )
+        .expect("write invalid pixel report");
+
+        assert!(matches!(
+            recv_server_event(&mut server_event_rx, "invalid pixel report disconnect"),
+            ServerEvent::ClientDisconnected { client_id: 7 }
+        ));
+        drop(client_stream);
+        should_quit.store(true, Ordering::Release);
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
     fn client_read_loop_disconnects_marker_wrapped_invalid_utf8() {
         let (mut client_stream, server_stream, _path) =
             local_stream_pair("client-read-invalid-utf8-paste");
diff --git a/src/server/clients.rs b/src/server/clients.rs
index bb0f31c6..0d3d9ae0 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -24,7 +24,6 @@ pub(crate) type RenderTarget = (
 pub(crate) enum DeferredRender {
     #[default]
     None,
-    Graphics,
     Full,
 }
 
@@ -56,14 +55,18 @@ pub(crate) struct ClientConnection {
     pub(crate) render_state: ClientRenderState,
     /// Client-local host Kitty graphics cache.
     pub(crate) graphics_cache: crate::kitty_graphics::HostGraphicsCache,
+    /// Passive eligibility for audited local Kitty regular-file graphics.
+    pub(crate) direct_graphics: bool,
+    /// Whether this frontend preserves exact SGR pixel reports.
+    pub(crate) pixel_mouse: bool,
     /// Whether the next graphics frame must clear and rebuild host-side Kitty state.
     pub(crate) graphics_surface_reset_pending: bool,
     /// 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>,
+    /// Last SGR pixel provenance mode sent to this client.
+    pub(crate) host_sgr_pixels_active: Option<bool>,
     /// Last Kitty report-all mode sent to this client's host terminal.
     pub(crate) host_keyboard_report_all_active: Option<bool>,
     /// Temporary files staged from this client's local clipboard image pastes.
@@ -125,10 +128,12 @@ impl ClientConnection {
             last_activity,
             render_state: ClientRenderState::new(render_encoding),
             graphics_cache: crate::kitty_graphics::HostGraphicsCache::default(),
+            direct_graphics: false,
+            pixel_mouse: false,
             graphics_surface_reset_pending: false,
             render_pending: false,
-            pane_graphics_render_pending: false,
             host_mouse_capture_active: None,
+            host_sgr_pixels_active: None,
             host_keyboard_report_all_active: None,
             staged_clipboard_files: Vec::new(),
             writer,
@@ -137,14 +142,11 @@ impl ClientConnection {
 
     pub(crate) fn request_repaint(&mut self) {
         self.render_state.request_repaint();
-        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
         }
@@ -152,18 +154,10 @@ impl ClientConnection {
 
     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 {
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 5a90177b..61c825b8 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -44,7 +44,6 @@ use crate::ipc::{
 };
 use crate::protocol::{
     self, AttachScrollDirection, AttachScrollSource, FrameData, ServerMessage, MAX_FRAME_SIZE,
-    MAX_GRAPHICS_FRAME_SIZE,
 };
 #[cfg(unix)]
 use crate::server::client_accept::{
@@ -66,8 +65,7 @@ use crate::server::terminal_attach::paste_payload_for_runtime;
 
 mod pane_graphics;
 
-#[cfg(test)]
-use pane_graphics::frame_pane_graphics_for_client;
+use crate::protocol::MAX_GRAPHICS_FRAME_SIZE;
 use pane_graphics::RetainedGraphicsOutcome;
 
 #[cfg(test)]
@@ -367,12 +365,13 @@ fn apply_terminal_attach_scroll(
     match runtime.wheel_routing() {
         Some(crate::pane::WheelRouting::MouseReport) => {
             runtime.scroll_reset();
-            let column = column.unwrap_or(0);
-            let row = row.unwrap_or(0);
+            let position = crate::input::mouse::Position::Cell {
+                column: column.unwrap_or(0),
+                row: row.unwrap_or(0),
+            };
             let Some(bytes) = runtime.encode_mouse_wheel(
                 wheel_kind,
-                column,
-                row,
+                position,
                 KeyModifiers::from_bits_truncate(modifiers),
             ) else {
                 return Err(format!(
@@ -671,7 +670,14 @@ impl HeadlessServer {
                 crate::render_prof::event("full_render_cause.default_workspace");
             }
 
-            self.cancel_inactive_pane_graphics_streams();
+            if self.app.pane_graphics.retain_live_panes(&self.app.state) {
+                needs_render = true;
+                needs_graphics_render = true;
+            }
+            if self.expire_direct_graphics(now) {
+                needs_render = true;
+                needs_graphics_render = true;
+            }
 
             self.drain_client_config_reload_request();
             self.stream_host_mouse_capture_mode();
@@ -1002,6 +1008,7 @@ impl HeadlessServer {
                 },
                 respond_to,
                 response_write_complete: None,
+                stream_active: None,
             },
             true,
         );
@@ -1083,6 +1090,15 @@ impl HeadlessServer {
     }
 
     fn sync_foreground_client_state(&mut self) {
+        self.app.direct_graphics_available = self.direct_graphics_available();
+        self.app.pixel_mouse_available = self.foreground_client_id.is_some_and(|id| {
+            self.clients
+                .get(&id)
+                .is_some_and(|client| client.pixel_mouse)
+        });
+        if !self.app.direct_graphics_available {
+            self.retire_all_direct_graphics();
+        }
         let Some(client_id) = self.foreground_client_id else {
             self.effective_size = (MIN_COLS, MIN_ROWS);
             self.app.state.outer_terminal_focus = None;
@@ -1485,11 +1501,21 @@ impl HeadlessServer {
             .count()
     }
 
+    fn direct_graphics_available(&self) -> bool {
+        self.app_client_count() == 1
+            && self.foreground_client_id.is_some_and(|id| {
+                self.clients.get(&id).is_some_and(|client| {
+                    client.is_full_app_client() && client.writer.is_some() && client.direct_graphics
+                })
+            })
+    }
+
     fn has_app_client(&self) -> bool {
         self.app_client_count() > 0
     }
 
     fn remove_client(&mut self, client_id: u64) -> bool {
+        self.retire_direct_graphics_for_client(client_id);
         let was_foreground = self.foreground_client_id == Some(client_id);
         self.app.clear_input_source(client_id);
         self.send_client_graphics_cleanup(client_id);
@@ -1551,7 +1577,7 @@ impl HeadlessServer {
         let Ok(serialized) = Self::frame_server_message(&ServerMessage::Graphics { bytes }) else {
             return;
         };
-        let _ = writer.control.send(serialized);
+        writer.replace_with_cleanup(serialized);
     }
 
     fn send_all_clients_graphics_cleanup(&mut self) {
@@ -2781,6 +2807,7 @@ impl HeadlessServer {
                 writer,
                 render_encoding,
                 direct_attach_requested,
+                direct_graphics,
             } => {
                 if self.handoff_in_progress {
                     if let Ok(message) =
@@ -2806,24 +2833,24 @@ impl HeadlessServer {
                     "client connected"
                 );
                 let last_activity = self.allocate_activity_stamp();
-                self.clients.insert(
-                    client_id,
-                    ClientConnection::new_with_mode(
-                        ClientConnectionMode::App,
-                        keybindings,
-                        (cols, rows),
-                        crate::kitty_graphics::HostCellSize {
-                            width_px: cell_width_px,
-                            height_px: cell_height_px,
-                        },
-                        crate::terminal_theme::TerminalTheme::default(),
-                        None,
-                        last_activity,
-                        render_encoding,
-                        direct_attach_requested,
-                        Some(writer),
-                    ),
+                let mut connection = ClientConnection::new_with_mode(
+                    ClientConnectionMode::App,
+                    keybindings,
+                    (cols, rows),
+                    crate::kitty_graphics::HostCellSize {
+                        width_px: cell_width_px,
+                        height_px: cell_height_px,
+                    },
+                    crate::terminal_theme::TerminalTheme::default(),
+                    None,
+                    last_activity,
+                    render_encoding,
+                    direct_attach_requested,
+                    Some(writer),
                 );
+                connection.direct_graphics = direct_graphics;
+                connection.pixel_mouse = direct_graphics;
+                self.clients.insert(client_id, connection);
                 if !direct_attach_requested {
                     self.foreground_client_id = Some(client_id);
                 }
@@ -2835,6 +2862,17 @@ impl HeadlessServer {
                 self.nudge_handoff_panes_on_first_client_attach();
                 true
             }
+            ServerEvent::GraphicsTransmissionResult {
+                client_id,
+                transfer_id,
+                image_id,
+                success,
+            } => self.complete_direct_graphics(client_id, transfer_id, image_id, success),
+            ServerEvent::GraphicsTransmissionStarted {
+                client_id,
+                transfer_id,
+                image_id,
+            } => self.start_direct_graphics_response(client_id, transfer_id, image_id),
             ServerEvent::ClientAttachTerminal {
                 client_id,
                 terminal_id,
@@ -2859,6 +2897,35 @@ impl HeadlessServer {
             } => self.handle_terminal_attach_scroll(
                 client_id, source, direction, lines, column, row, modifiers,
             ),
+            ServerEvent::ClientInputPixels {
+                client_id,
+                data,
+                geometry,
+            } => {
+                let coordinates_valid = crate::input::mouse::parse_report(&data)
+                    .and_then(|(x, y)| geometry.cell(x, y))
+                    .is_some();
+                let valid = coordinates_valid
+                    && self.clients.get(&client_id).is_some_and(|client| {
+                        let cell = client.cell_size;
+                        client.is_full_app_client()
+                            && client.host_sgr_pixels_active == Some(true)
+                            && client.terminal_size == (geometry.cols, geometry.rows)
+                            && cell.is_known()
+                            && cell.width_px == geometry.width_px / u32::from(geometry.cols)
+                            && cell.height_px == geometry.height_px / u32::from(geometry.rows)
+                    });
+                if !valid || self.handoff_in_progress || !self.focused_pane_graphics_demand() {
+                    return false;
+                }
+                let foreground_changed = self.promote_client_to_foreground(client_id);
+                if foreground_changed {
+                    self.resize_shared_runtime_to_effective_size_before_input();
+                }
+                self.app
+                    .route_client_pixel_mouse(client_id, &data, geometry)
+                    || foreground_changed
+            }
             ServerEvent::ClientInput { client_id, data } => {
                 if self.handoff_in_progress {
                     debug!(
@@ -2987,18 +3054,21 @@ impl HeadlessServer {
                 }) = self.clients.get_mut(&client_id)
                 {
                     *terminal_size = (cols, rows);
-                    *cell_size = crate::kitty_graphics::HostCellSize {
+                    let observed = crate::kitty_graphics::HostCellSize {
                         width_px: cell_width_px,
                         height_px: cell_height_px,
                     };
+                    if observed.is_known() {
+                        *cell_size = observed;
+                    }
                     render_state.request_repaint();
-                    Some(terminal_id.clone())
+                    Some((terminal_id.clone(), *cell_size))
                 } else {
                     None
                 };
-                if let Some(terminal_id) = direct_terminal_id {
+                if let Some((terminal_id, cell_size)) = direct_terminal_id {
                     if let Some(runtime) = self.runtime_for_terminal_id_string(&terminal_id) {
-                        runtime.resize(rows, cols, cell_width_px, cell_height_px);
+                        runtime.resize(rows, cols, cell_size.width_px, cell_size.height_px);
                     }
                     return true;
                 }
@@ -3011,19 +3081,25 @@ impl HeadlessServer {
                 }) = self.clients.get_mut(&client_id)
                 {
                     *terminal_size = (cols, rows);
-                    *cell_size = crate::kitty_graphics::HostCellSize {
+                    let observed = crate::kitty_graphics::HostCellSize {
                         width_px: cell_width_px,
                         height_px: cell_height_px,
                     };
+                    if observed.is_known() {
+                        *cell_size = observed;
+                    }
                     render_state.request_repaint();
                     return true;
                 }
                 if let Some(client) = self.clients.get_mut(&client_id) {
                     client.terminal_size = (cols, rows);
-                    client.cell_size = crate::kitty_graphics::HostCellSize {
+                    let observed = crate::kitty_graphics::HostCellSize {
                         width_px: cell_width_px,
                         height_px: cell_height_px,
                     };
+                    if observed.is_known() {
+                        client.cell_size = observed;
+                    }
                 }
                 self.promote_client_to_foreground(client_id);
                 self.resize_shared_runtime_to_effective_size();
@@ -3055,19 +3131,10 @@ 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,
+        if self.handle_server_event(ev) {
+            RenderImpact::Full
+        } else {
+            RenderImpact::None
         }
     }
 
@@ -3299,6 +3366,7 @@ impl HeadlessServer {
         if matches!(
             &msg.request.method,
             api::schema::Method::PaneGraphicsStreamSet(_)
+                | api::schema::Method::PaneGraphicsStreamDirect(_)
         ) {
             return self.handle_pane_graphics_stream_frame(msg);
         }
@@ -3341,6 +3409,11 @@ impl HeadlessServer {
         };
 
         let metadata_expired = self.app.expire_due_metadata(Instant::now());
+        let stream_open = match &msg.request.method {
+            api::schema::Method::PaneGraphicsStreamOpen(params) => Some(params.clone()),
+            _ => None,
+        };
+        let stream_active = msg.stream_active.clone();
 
         if let api::schema::Method::ServerLiveHandoff(params) = &msg.request.method {
             let handoff_result = self.perform_live_handoff(params.clone());
@@ -3398,7 +3471,7 @@ impl HeadlessServer {
                 | api::schema::Method::PaneGraphicsStreamOpen(_)
                 | api::schema::Method::PaneGraphicsStreamClose(_)
         )
-        .then_some(self.app.state.pane_graphics_revision);
+        .then_some(self.app.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
@@ -3501,6 +3574,10 @@ impl HeadlessServer {
                 }
             }
         }
+        if let (Some(params), Some(active)) = (stream_open.as_ref(), stream_active) {
+            self.app
+                .attach_pane_graphics_stream_active(params, active, &response);
+        }
         if let Some(spec) = alt_screen_read_spec {
             if let Ok(success) = serde_json::from_str::<api::schema::SuccessResponse>(&response) {
                 if let api::schema::ResponseResult::PaneRead { read } = success.result {
@@ -3523,7 +3600,7 @@ 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;
+            changed |= revision_before != self.app.pane_graphics.revision();
         }
 
         // Forward new toast state only when a client-local delivery mode is selected.
@@ -3671,32 +3748,73 @@ impl HeadlessServer {
         changed
     }
 
+    fn focused_pane_graphics_demand(&self) -> bool {
+        self.app
+            .state
+            .active
+            .and_then(|ws_idx| self.app.state.workspaces.get(ws_idx))
+            .and_then(crate::workspace::Workspace::focused_pane_id)
+            .is_some_and(|pane_id| self.app.pane_graphics.active_for_pane(pane_id))
+    }
+
     fn stream_host_mouse_capture_mode(&mut self) {
         let enabled = self
             .app
             .state
             .should_capture_host_mouse_from(&self.app.terminal_runtimes);
-        let serialized = match Self::frame_server_message(&ServerMessage::MouseCapture { enabled })
-        {
-            Ok(framed) => framed,
-            Err(err) => {
-                warn!(err = %err, "failed to serialize mouse capture mode for clients");
-                return;
-            }
-        };
-
+        let pixel_mouse_requested = self
+            .clients
+            .values()
+            .any(|client| client.is_full_app_client() && client.pixel_mouse);
+        let sgr_pixels = pixel_mouse_requested
+            && self.focused_pane_graphics_demand()
+            && self
+                .app
+                .state
+                .active
+                .and_then(|ws_idx| {
+                    self.app
+                        .state
+                        .workspaces
+                        .get(ws_idx)
+                        .and_then(crate::workspace::Workspace::focused_pane_id)
+                        .and_then(|pane_id| {
+                            self.app.state.runtime_for_pane_in_workspace(
+                                &self.app.terminal_runtimes,
+                                ws_idx,
+                                pane_id,
+                            )
+                        })
+                })
+                .and_then(crate::terminal::TerminalRuntime::input_state)
+                .is_some_and(|state| {
+                    state.mouse_protocol_encoding == crate::input::MouseProtocolEncoding::SgrPixels
+                });
         let mut broken_clients: Vec<u64> = Vec::new();
         for (&client_id, client) in &mut self.clients {
             if !client.is_full_app_client() {
                 continue;
             }
-            if client.host_mouse_capture_active == Some(enabled) {
+            let client_sgr_pixels = sgr_pixels && client.pixel_mouse;
+            if client.host_mouse_capture_active == Some(enabled)
+                && client.host_sgr_pixels_active == Some(client_sgr_pixels)
+            {
                 continue;
             }
             let Some(writer) = &client.writer else {
                 continue;
             };
-            if writer.control.send(serialized.clone()).is_err() {
+            let serialized = match Self::frame_server_message(&ServerMessage::MouseCapture {
+                enabled,
+                sgr_pixels: client_sgr_pixels,
+            }) {
+                Ok(framed) => framed,
+                Err(err) => {
+                    warn!(err = %err, "failed to serialize mouse capture mode for client");
+                    continue;
+                }
+            };
+            if writer.control.send(serialized).is_err() {
                 debug!(
                     client_id,
                     "client writer channel closed during mouse capture update"
@@ -3705,6 +3823,7 @@ impl HeadlessServer {
                 continue;
             }
             client.host_mouse_capture_active = Some(enabled);
+            client.host_sgr_pixels_active = Some(client_sgr_pixels);
         }
 
         for client_id in broken_clients {
@@ -3877,6 +3996,7 @@ impl HeadlessServer {
             && cell_size.is_known()
             && crate::kitty_graphics::has_visible_pane_graphics(
                 &self.app.state,
+                &self.app.pane_graphics,
                 &self.app.terminal_runtimes,
                 self.app.state.view.tab_surface(),
                 *cell_size,
@@ -4166,31 +4286,48 @@ impl HeadlessServer {
                 continue;
             };
             let mut next_graphics_cache = client.graphics_cache.clone();
-            let graphics_surface_reset_pending = client.graphics_surface_reset_pending;
-            if is_app_client && self.app.state.kitty_graphics_enabled && cell_size.is_known() {
-                if graphics_surface_reset_pending {
-                    frame.graphics = next_graphics_cache.clear_bytes();
+            let mut reset_graphics = Vec::new();
+            let mut encoded = if is_app_client
+                && self.app.state.kitty_graphics_enabled
+                && cell_size.is_known()
+            {
+                if client.graphics_surface_reset_pending {
+                    if self.app.pane_graphics.slots.is_empty() {
+                        reset_graphics = next_graphics_cache.clear_bytes();
+                    } else {
+                        next_graphics_cache = crate::kitty_graphics::HostGraphicsCache::default();
+                    }
                 }
                 let graphics_started = crate::render_prof::timer();
-                frame
-                    .graphics
-                    .extend(crate::kitty_graphics::encode_local_pane_graphics(
-                        &self.app.state,
-                        &self.app.terminal_runtimes,
-                        self.app.state.view.tab_surface(),
-                        cell_size,
-                        &mut next_graphics_cache,
-                    ));
+                let encoded = crate::kitty_graphics::encode_local_pane_graphics(
+                    &self.app.state,
+                    &self.app.pane_graphics,
+                    &self.app.terminal_runtimes,
+                    self.app.state.view.tab_surface(),
+                    cell_size,
+                    Some(crate::kitty_graphics::HEADLESS_GRAPHICS_TRANSACTION_BUDGET),
+                    &mut next_graphics_cache,
+                );
                 crate::render_prof::duration_since("full_render.graphics_encode", graphics_started);
+                encoded
+            } else if self.app.pane_graphics.slots.is_empty() {
+                crate::kitty_graphics::EncodedGraphics {
+                    bytes: next_graphics_cache.clear_bytes(),
+                    incomplete: false,
+                }
             } else {
-                frame.graphics = next_graphics_cache.clear_bytes();
+                next_graphics_cache.clear_next()
+            };
+            if !reset_graphics.is_empty() {
+                reset_graphics.extend(encoded.bytes);
+                encoded.bytes = reset_graphics;
             }
+            frame.graphics = encoded.bytes;
 
             let Some(writer) = client.writer.as_ref().cloned() else {
                 crate::render_prof::event("full_render.writer_missing");
                 continue;
             };
-
             let mut commit_graphics_cache = true;
             if frame.graphics.len() > MAX_GRAPHICS_FRAME_SIZE {
                 warn!(
@@ -4201,32 +4338,30 @@ impl HeadlessServer {
                 );
                 frame.graphics.clear();
                 commit_graphics_cache = false;
+                encoded.incomplete = false;
             }
-
-            let max_frame_size = if frame.graphics.is_empty() {
-                MAX_FRAME_SIZE
-            } else {
-                MAX_GRAPHICS_FRAME_SIZE
-            };
             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.clear_deferred_render();
+                if commit_graphics_cache {
+                    client.graphics_cache = next_graphics_cache;
+                    client.graphics_surface_reset_pending = false;
+                }
+                if encoded.incomplete {
+                    client.defer_full_render();
+                    deferred_frame = true;
+                } else {
+                    client.clear_deferred_render();
+                }
                 crate::render_prof::event("full_render.skip_identical");
-                crate::render_prof::duration_since("full_render.prepare_frame", prepare_started);
                 continue;
             };
-            crate::render_prof::duration_since("full_render.prepare_frame", prepare_started);
-
-            let serialize_started = crate::render_prof::timer();
-            let serialized = match Self::frame_server_message_with_max(
-                prepared.message(),
-                max_frame_size,
-            ) {
-                Ok(framed) => {
-                    crate::render_prof::duration_since("full_render.serialize", serialize_started);
-                    framed
-                }
+            let max = if has_graphics {
+                MAX_GRAPHICS_FRAME_SIZE
+            } else {
+                crate::protocol::MAX_FRAME_SIZE
+            };
+            let serialized = match Self::frame_server_message_with_max(prepared.message(), max) {
+                Ok(frame) => frame,
                 Err(protocol::FramingError::Oversized { claimed, max }) if has_graphics => {
                     warn!(
                         client_id,
@@ -4234,10 +4369,6 @@ impl HeadlessServer {
                     );
                     let Some(mut text_only_frame) = prepared.into_frame() else {
                         crate::render_prof::event("full_render.serialize_error");
-                        crate::render_prof::duration_since(
-                            "full_render.serialize",
-                            serialize_started,
-                        );
                         continue;
                     };
                     text_only_frame.graphics.clear();
@@ -4246,10 +4377,6 @@ impl HeadlessServer {
                     else {
                         client.clear_deferred_render();
                         crate::render_prof::event("full_render.skip_identical_text_only");
-                        crate::render_prof::duration_since(
-                            "full_render.serialize",
-                            serialize_started,
-                        );
                         continue;
                     };
                     let framed = match Self::frame_server_message(text_only_prepared.message()) {
@@ -4258,16 +4385,12 @@ impl HeadlessServer {
                             warn!(client_id, err = %err, "failed to serialize text-only frame for client");
                             broken_clients.push(client_id);
                             crate::render_prof::event("full_render.serialize_error");
-                            crate::render_prof::duration_since(
-                                "full_render.serialize",
-                                serialize_started,
-                            );
                             continue;
                         }
                     };
                     prepared = text_only_prepared;
                     commit_graphics_cache = false;
-                    crate::render_prof::duration_since("full_render.serialize", serialize_started);
+                    encoded.incomplete = false;
                     framed
                 }
                 Err(protocol::FramingError::Oversized { claimed, max }) => {
@@ -4276,45 +4399,36 @@ impl HeadlessServer {
                         claimed, max, "skipping oversized frame for client"
                     );
                     crate::render_prof::event("full_render.serialize_oversized");
-                    crate::render_prof::duration_since("full_render.serialize", serialize_started);
                     continue;
                 }
                 Err(err) => {
-                    warn!(client_id, err = %err, "failed to serialize frame for client");
+                    warn!(client_id, err = %err, "failed to serialize frame");
                     broken_clients.push(client_id);
                     crate::render_prof::event("full_render.serialize_error");
-                    crate::render_prof::duration_since("full_render.serialize", serialize_started);
                     continue;
                 }
             };
-            crate::render_prof::counter("full_render.bytes", serialized.len() as u64);
-
-            let send_started = crate::render_prof::timer();
             match writer.render.try_send(serialized) {
                 Ok(()) => {
-                    client.clear_deferred_render();
                     if commit_graphics_cache {
                         client.graphics_cache = next_graphics_cache;
                         client.graphics_surface_reset_pending = false;
                     }
                     client.render_state.commit_sent_frame(prepared);
+                    if encoded.incomplete {
+                        client.defer_full_render();
+                        deferred_frame = true;
+                    } else {
+                        client.clear_deferred_render();
+                    }
                     crate::render_prof::event("full_render.sent");
-                    crate::render_prof::duration_since("full_render.try_send", send_started);
                 }
                 Err(std::sync::mpsc::TrySendError::Full(_)) => {
                     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);
-                    debug!(client_id, "render queue full, deferring latest frame");
-                    continue;
                 }
                 Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
-                    debug!(client_id, "client writer channel closed, marking as broken");
                     broken_clients.push(client_id);
-                    crate::render_prof::event("full_render.writer_disconnected");
-                    crate::render_prof::duration_since("full_render.try_send", send_started);
-                    continue;
                 }
             }
         }
@@ -5044,7 +5158,9 @@ mod tests {
     }
 
     fn read_server_frame(bytes: Vec<u8>) -> FrameData {
-        match read_server_message(bytes) {
+        match protocol::read_message(&mut std::io::Cursor::new(bytes), MAX_GRAPHICS_FRAME_SIZE)
+            .expect("decode server frame")
+        {
             ServerMessage::Frame(frame) => frame,
             other => panic!("expected frame, got {other:?}"),
         }
@@ -5139,6 +5255,7 @@ mod tests {
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         });
         let response: api::schema::SuccessResponse =
             serde_json::from_str(&response_rx.recv().unwrap()).unwrap();
@@ -5179,6 +5296,7 @@ mod tests {
                 },
                 respond_to,
                 response_write_complete: None,
+                stream_active: None,
             })
         );
         let response = response_rx
@@ -5440,6 +5558,42 @@ mod tests {
     }
 
     #[test]
+    fn direct_graphics_requires_one_negotiated_app_client() {
+        let mut server = test_headless_server();
+        let (writer_a, _control_a, _render_a) = test_client_writer();
+        assert!(server.handle_server_event(ServerEvent::ClientConnected {
+            client_id: 1,
+            cols: 80,
+            rows: 24,
+            cell_width_px: 10,
+            cell_height_px: 20,
+            render_encoding: RenderEncoding::SemanticFrame,
+            keybindings: None,
+            direct_attach_requested: false,
+            direct_graphics: true,
+            writer: writer_a,
+        }));
+        assert!(server.clients[&1].direct_graphics);
+        assert!(server.clients[&1].pixel_mouse);
+        assert!(server.direct_graphics_available());
+
+        let (writer_b, _control_b, _render_b) = test_client_writer();
+        assert!(server.handle_server_event(ServerEvent::ClientConnected {
+            client_id: 2,
+            cols: 80,
+            rows: 24,
+            cell_width_px: 10,
+            cell_height_px: 20,
+            render_encoding: RenderEncoding::SemanticFrame,
+            keybindings: None,
+            direct_attach_requested: false,
+            direct_graphics: false,
+            writer: writer_b,
+        }));
+        assert!(!server.direct_graphics_available());
+    }
+
+    #[test]
     fn foreground_client_applies_client_keybindings() {
         let mut server = test_headless_server();
         let local_config: crate::config::Config = toml::from_str(
@@ -5463,6 +5617,7 @@ new_tab = "prefix+t"
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: Some(Box::new(local_keybindings)),
             direct_attach_requested: false,
+            direct_graphics: false,
             writer: writer_a,
         }));
         assert_eq!(
@@ -5487,6 +5642,7 @@ new_tab = "prefix+t"
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: None,
             direct_attach_requested: false,
+            direct_graphics: false,
             writer: writer_b,
         }));
         assert_eq!(
@@ -5527,6 +5683,7 @@ new_tab = "prefix+t"
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: Some(Box::new(local_keybindings)),
             direct_attach_requested: false,
+            direct_graphics: false,
             writer: writer_a,
         }));
         assert_eq!(server.app.state.config_diagnostic, without_keybindings);
@@ -5540,6 +5697,7 @@ new_tab = "prefix+t"
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: None,
             direct_attach_requested: false,
+            direct_graphics: false,
             writer: writer_b,
         }));
         assert_eq!(
@@ -5583,6 +5741,7 @@ next_tab = ""
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: Some(Box::new(local_keybindings)),
             direct_attach_requested: false,
+            direct_graphics: false,
             writer,
         }));
         server.app.state.mode = crate::app::Mode::Settings;
@@ -5658,6 +5817,7 @@ next_tab = ""
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: Some(Box::new(local_config.live_keybinds().unwrap())),
             direct_attach_requested: false,
+            direct_graphics: false,
             writer: writer_a,
         }));
         server.app.state.mode = crate::app::Mode::Settings;
@@ -5678,6 +5838,7 @@ next_tab = ""
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: None,
             direct_attach_requested: false,
+            direct_graphics: false,
             writer: writer_b,
         }));
         assert_eq!(
@@ -5712,6 +5873,7 @@ next_tab = ""
             render_encoding: RenderEncoding::TerminalAnsi,
             keybindings: None,
             direct_attach_requested: true,
+            direct_graphics: false,
             writer,
         }));
         assert!(server.clients.contains_key(&7));
@@ -5777,6 +5939,7 @@ next_tab = ""
             render_encoding: RenderEncoding::TerminalAnsi,
             keybindings: None,
             direct_attach_requested: true,
+            direct_graphics: false,
             writer,
         }));
         control_rx
@@ -6189,6 +6352,7 @@ next_tab = ""
             render_encoding,
             keybindings: None,
             direct_attach_requested: false,
+            direct_graphics: false,
             writer,
         }));
 
@@ -6223,6 +6387,7 @@ next_tab = ""
             render_encoding: RenderEncoding::TerminalAnsi,
             keybindings: None,
             direct_attach_requested: true,
+            direct_graphics: false,
             writer,
         }));
 
@@ -6256,6 +6421,7 @@ next_tab = ""
             render_encoding: RenderEncoding::SemanticFrame,
             keybindings: None,
             direct_attach_requested: false,
+            direct_graphics: false,
             writer,
         }));
         assert!(server.has_app_client());
@@ -6356,6 +6522,7 @@ next_tab = ""
             render_encoding: RenderEncoding::TerminalAnsi,
             keybindings: None,
             direct_attach_requested: true,
+            direct_graphics: false,
             writer,
         }));
         assert!(
@@ -7778,6 +7945,51 @@ next_tab = ""
         assert_eq!(server.foreground_client_id, Some(3));
     }
 
+    #[test]
+    fn terminal_attach_resize_preserves_known_cell_size_when_pixels_are_omitted() {
+        with_terminal_session_test_server(|server, _terminal_id, terminal_id, _pane_id| {
+            let mut client = ClientConnection::new(
+                (80, 24),
+                crate::kitty_graphics::HostCellSize {
+                    width_px: 10,
+                    height_px: 20,
+                },
+                crate::terminal_theme::TerminalTheme::default(),
+                Some(true),
+                1,
+                RenderEncoding::SemanticFrame,
+                None,
+            );
+            client.mode = ClientConnectionMode::TerminalAttach {
+                terminal_id: terminal_id.clone(),
+            };
+            server.clients.insert(1, client);
+
+            assert!(server.handle_server_event(ServerEvent::ClientResize {
+                client_id: 1,
+                cols: 100,
+                rows: 30,
+                cell_width_px: 0,
+                cell_height_px: 0,
+            }));
+
+            assert_eq!(
+                server
+                    .runtime_for_terminal_id_string(&terminal_id)
+                    .unwrap()
+                    .pixel_size(),
+                Some((1_000, 600))
+            );
+            assert_eq!(
+                server.clients[&1].cell_size,
+                crate::kitty_graphics::HostCellSize {
+                    width_px: 10,
+                    height_px: 20,
+                }
+            );
+        });
+    }
+
     #[tokio::test]
     async fn passive_mouse_motion_forwards_without_requesting_render() {
         let mut server = test_headless_server();
@@ -8335,6 +8547,7 @@ next_tab = ""
             render_encoding: RenderEncoding::TerminalAnsi,
             keybindings: None,
             direct_attach_requested: true,
+            direct_graphics: false,
             writer,
         }));
         assert!(
@@ -8531,16 +8744,11 @@ next_tab = ""
 
         let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
             .expect("serialize dummy message");
-        server
-            .clients
-            .get(&1)
-            .unwrap()
+        server.clients[&1]
             .writer
             .as_ref()
             .unwrap()
-            .render
-            .try_send(queued)
-            .expect("pre-fill render queue");
+            .test_fill_render(queued);
 
         assert!(server.handle_server_event(ServerEvent::ClientInput {
             client_id: 1,
@@ -8671,10 +8879,7 @@ next_tab = ""
         let (client_tx, _client_control_rx, client_rx) = test_client_writer();
         let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
             .expect("serialize dummy message");
-        client_tx
-            .render
-            .try_send(queued)
-            .expect("pre-fill render queue");
+        client_tx.test_fill_render(queued);
 
         server.clients.insert(
             1,
@@ -8715,10 +8920,7 @@ next_tab = ""
         let (client_tx, _client_control_rx, client_rx) = test_client_writer();
         let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
             .expect("serialize dummy message");
-        client_tx
-            .render
-            .try_send(queued)
-            .expect("pre-fill render queue");
+        client_tx.test_fill_render(queued);
 
         server.clients.insert(
             1,
@@ -9003,7 +9205,7 @@ next_tab = ""
                     .recv_timeout(Duration::from_millis(100))
                     .expect("mouse capture message")
             ),
-            ServerMessage::MouseCapture { enabled: true }
+            ServerMessage::MouseCapture { enabled: true, .. }
         ));
     }
 
@@ -9525,16 +9727,11 @@ next_tab = ""
         let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
         let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
             .expect("serialize dummy message");
-        server
-            .clients
-            .get(&1)
-            .unwrap()
+        server.clients[&1]
             .writer
             .as_ref()
             .unwrap()
-            .render
-            .try_send(queued)
-            .expect("pre-fill render queue");
+            .test_fill_render(queued);
         server.app.full_redraw_pending = true;
 
         server.render_and_stream();
@@ -9742,6 +9939,7 @@ next_tab = ""
         let mut server = test_headless_server();
         let (foreground_tx, foreground_control_rx, _foreground_rx) = test_client_writer();
         drop(foreground_control_rx);
+        foreground_tx.test_close();
 
         server.clients.insert(
             1,
@@ -10125,6 +10323,7 @@ next_tab = ""
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         });
 
         assert!(changed);
@@ -10211,6 +10410,7 @@ next_tab = ""
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         });
 
         assert!(changed);
@@ -10263,6 +10463,7 @@ next_tab = ""
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         });
 
         assert!(changed);
@@ -10295,6 +10496,7 @@ next_tab = ""
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         });
 
         assert!(changed);
@@ -10332,6 +10534,7 @@ next_tab = ""
                 },
                 respond_to,
                 response_write_complete: None,
+                stream_active: None,
             })
         );
 
@@ -10388,6 +10591,7 @@ next_tab = ""
                 },
                 respond_to,
                 response_write_complete: None,
+                stream_active: None,
             })
         );
 
@@ -10688,6 +10892,7 @@ next_tab = ""
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         });
 
         assert!(changed);
diff --git a/src/server/headless/pane_graphics.rs b/src/server/headless/pane_graphics.rs
index ed28eab0..d027f79b 100644
--- a/src/server/headless/pane_graphics.rs
+++ b/src/server/headless/pane_graphics.rs
@@ -1,5 +1,3 @@
-use tracing::warn;
-
 use super::{HeadlessServer, RenderImpact};
 use crate::api;
 use crate::protocol::{ServerMessage, MAX_GRAPHICS_FRAME_SIZE};
@@ -12,31 +10,16 @@ pub(super) enum RetainedGraphicsOutcome {
     Fallback,
 }
 
-pub(super) fn frame_pane_graphics_for_client(bytes: Vec<u8>) -> Vec<u8> {
+pub(super) fn frame_pane_graphics(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
+    [b"\x1b7".as_slice(), &bytes, b"\x1b8"].concat()
 }
 
 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)
-        });
+        !self.app.pane_graphics.slots.is_empty()
     }
 
     pub(super) fn handle_pane_graphics_stream_frame(
@@ -49,12 +32,164 @@ impl HeadlessServer {
         }
 
         let internal_changed = self.drain_all_internal_events_with_forwarding();
+        let direct = match &msg.request.method {
+            api::schema::Method::PaneGraphicsStreamDirect(params) => Some(params.clone()),
+            _ => None,
+        };
+        let direct_key = direct.as_ref().and_then(|params| {
+            self.app.parse_pane_id(&params.pane_id).map(|(_, pane_id)| {
+                (
+                    pane_id,
+                    params
+                        .layer_id
+                        .clone()
+                        .unwrap_or_else(|| api::schema::PANE_GRAPHICS_PRIMARY_LAYER_ID.into()),
+                )
+            })
+        });
+        let direct_client = self
+            .direct_graphics_available()
+            .then_some(self.foreground_client_id)
+            .flatten();
+        let gate_busy = self
+            .app
+            .pane_graphics
+            .slots
+            .values()
+            .any(|slot| slot.direct_gate.is_some());
+        self.app.direct_graphics_available = direct_client.is_some() && !gate_busy;
         let response = self
             .app
             .handle_api_request_after_internal_events_drained(msg.request);
+        self.app.direct_graphics_available = self.direct_graphics_available();
         let succeeded = serde_json::from_str::<api::schema::SuccessResponse>(&response).is_ok();
-        let _ = msg.respond_to.send(response);
 
+        if succeeded {
+            if let (Some(params), Some(client_id)) = (direct, direct_client) {
+                let direct_frame = direct_key.as_ref().and_then(|key| {
+                    self.app.pane_graphics.slots.get(key).and_then(|slot| {
+                        let active_owner = slot.stream_owner.as_deref() == Some(&params.owner)
+                            && slot.stream_is_active();
+                        active_owner.then_some(())?;
+                        slot.layer
+                            .as_ref()
+                            .and_then(crate::app::pane_graphics::Layer::direct_lease)
+                            .map(|lease| {
+                                (
+                                    slot.host_image_id,
+                                    lease.path().to_string_lossy().into_owned(),
+                                    lease.len() as u64,
+                                    lease.fingerprint(),
+                                )
+                            })
+                    })
+                });
+                if let (Some(key), Some((image_id, path, expected_len, transfer_id))) =
+                    (direct_key.clone(), direct_frame)
+                {
+                    let mut next_cache = self
+                        .clients
+                        .get(&client_id)
+                        .map(|client| client.graphics_cache.clone())
+                        .unwrap_or_default();
+                    let command = (!internal_changed)
+                        .then(|| {
+                            crate::kitty_graphics::prepare_direct_file(
+                                &self.app.state,
+                                &self.app.pane_graphics,
+                                self.app.state.view.tab_surface(),
+                                self.clients
+                                    .get(&client_id)
+                                    .map(|client| client.cell_size)
+                                    .unwrap_or_default(),
+                                &mut next_cache,
+                                &key,
+                            )
+                        })
+                        .flatten();
+                    let Some(command) = command else {
+                        if self.install_inline_fallback(&key) {
+                            if msg.respond_to.send(response).is_err() {
+                                self.retire_direct_gate(&key);
+                            }
+                            return if internal_changed {
+                                RenderImpact::Full
+                            } else {
+                                RenderImpact::Graphics
+                            };
+                        }
+                        self.retire_direct_gate(&key);
+                        return if internal_changed {
+                            RenderImpact::Full
+                        } else {
+                            RenderImpact::Graphics
+                        };
+                    };
+                    let message = ServerMessage::GraphicsFile {
+                        path,
+                        expected_len,
+                        image_id,
+                        transfer_id,
+                        leading: command.leading,
+                        control: command.control,
+                    };
+                    let send =
+                        Self::frame_server_message_with_max(&message, MAX_GRAPHICS_FRAME_SIZE)
+                            .map_err(|_| std::sync::mpsc::TrySendError::Full(Vec::new()))
+                            .and_then(|framed| {
+                                let Some(writer) = self
+                                    .clients
+                                    .get(&client_id)
+                                    .and_then(|client| client.writer.as_ref())
+                                else {
+                                    return Err(std::sync::mpsc::TrySendError::Disconnected(
+                                        framed,
+                                    ));
+                                };
+                                writer.render.send_ordered(framed)
+                            });
+                    match send {
+                        Ok(()) => {
+                            if let Some(slot) = self.app.pane_graphics.slots.get_mut(&key) {
+                                slot.direct_gate = Some(crate::app::pane_graphics::DirectGate {
+                                    transfer_id,
+                                    client_id,
+                                    deadline: std::time::Instant::now()
+                                        + crate::app::pane_graphics::DIRECT_DELIVERY_TIMEOUT,
+                                    written: false,
+                                    success_response: response,
+                                    respond_to: msg.respond_to,
+                                });
+                            }
+                            if let Some(client) = self.clients.get_mut(&client_id) {
+                                client.graphics_cache = next_cache;
+                            }
+                            return if internal_changed {
+                                RenderImpact::Full
+                            } else {
+                                RenderImpact::None
+                            };
+                        }
+                        Err(error) => {
+                            self.handle_unwritten_direct_failure(
+                                &key,
+                                response,
+                                msg.respond_to,
+                                error,
+                            );
+                            return RenderImpact::Graphics;
+                        }
+                    }
+                }
+            }
+        }
+
+        let response_failed = msg.respond_to.send(response).is_err();
+        if succeeded && response_failed {
+            if let Some(key) = direct_key {
+                self.retire_direct_gate(&key);
+            }
+        }
         if internal_changed {
             RenderImpact::Full
         } else if succeeded {
@@ -64,6 +199,276 @@ impl HeadlessServer {
         }
     }
 
+    fn install_inline_fallback(&mut self, key: &crate::app::pane_graphics::Key) -> bool {
+        let len = self
+            .app
+            .pane_graphics
+            .slots
+            .get(key)
+            .and_then(|slot| slot.layer.as_ref())
+            .and_then(crate::app::pane_graphics::Layer::direct_lease)
+            .map(crate::pane_graphics_files::Lease::len);
+        let Some(len) = len else {
+            return false;
+        };
+        if !self.app.pane_graphics.can_store_inline(key, len) {
+            return false;
+        }
+        let data = self
+            .app
+            .pane_graphics
+            .slots
+            .get(key)
+            .and_then(|slot| slot.layer.as_ref())
+            .and_then(crate::app::pane_graphics::Layer::direct_lease)
+            .and_then(|lease| lease.copy_rgba().ok());
+        let Some(data) = data else {
+            return false;
+        };
+        let slot = self
+            .app
+            .pane_graphics
+            .slots
+            .get_mut(key)
+            .expect("matched slot");
+        if !slot.stream_is_active() {
+            return false;
+        }
+        let layer = slot.layer.take().expect("direct layer");
+        slot.layer = Some(crate::app::pane_graphics::Layer::inline(
+            layer.format,
+            layer.image_width,
+            layer.image_height,
+            data,
+            layer.render,
+            layer.z_index,
+        ));
+        self.app.pane_graphics.mark_changed();
+        true
+    }
+
+    pub(super) fn handle_unwritten_direct_failure(
+        &mut self,
+        key: &crate::app::pane_graphics::Key,
+        response: String,
+        respond_to: std::sync::mpsc::Sender<String>,
+        error: std::sync::mpsc::TrySendError<Vec<u8>>,
+    ) -> bool {
+        if matches!(error, std::sync::mpsc::TrySendError::Full(_))
+            && self.install_inline_fallback(key)
+            && respond_to.send(response).is_ok()
+        {
+            return true;
+        }
+        self.retire_direct_gate(key);
+        false
+    }
+
+    pub(super) fn retire_direct_gate(&mut self, key: &crate::app::pane_graphics::Key) {
+        if self.app.pane_graphics.slots.remove(key).is_some() {
+            self.app.pane_graphics.mark_changed();
+        }
+    }
+
+    pub(super) fn start_direct_graphics_response(
+        &mut self,
+        client_id: u64,
+        transfer_id: u64,
+        image_id: u32,
+    ) -> bool {
+        if let Some(gate) = self.app.pane_graphics.slots.values_mut().find_map(|slot| {
+            (slot.host_image_id == image_id && slot.stream_is_active())
+                .then_some(slot.direct_gate.as_mut())
+                .flatten()
+                .filter(|gate| gate.client_id == client_id && gate.transfer_id == transfer_id)
+        }) {
+            gate.written = true;
+            gate.deadline =
+                std::time::Instant::now() + crate::app::pane_graphics::DIRECT_RESPONSE_TIMEOUT;
+        }
+        false
+    }
+
+    pub(super) fn complete_direct_graphics(
+        &mut self,
+        client_id: u64,
+        transfer_id: u64,
+        image_id: u32,
+        success: bool,
+    ) -> bool {
+        if self.shutting_down {
+            self.retire_all_direct_graphics();
+            return false;
+        }
+        let key = self.app.pane_graphics.slots.iter().find_map(|(key, slot)| {
+            slot.direct_gate
+                .as_ref()
+                .is_some_and(|gate| {
+                    slot.stream_is_active()
+                        && gate.client_id == client_id
+                        && gate.transfer_id == transfer_id
+                        && slot.host_image_id == image_id
+                        && (!success || gate.written)
+                })
+                .then(|| key.clone())
+        });
+        let Some(key) = key else {
+            return false;
+        };
+        let pane_is_live = self
+            .app
+            .state
+            .workspaces
+            .iter()
+            .any(|workspace| workspace.pane_state(key.0).is_some());
+        if !pane_is_live {
+            self.retire_direct_gate(&key);
+            return false;
+        }
+        if success {
+            let (gate, host_image_id) = {
+                let slot = self
+                    .app
+                    .pane_graphics
+                    .slots
+                    .get_mut(&key)
+                    .expect("matched slot");
+                if !slot.stream_is_active() {
+                    return false;
+                }
+                if let Some(layer) = slot.layer.as_mut() {
+                    layer.mark_resident(client_id);
+                }
+                (
+                    slot.direct_gate.take().expect("matched gate"),
+                    slot.host_image_id,
+                )
+            };
+            if let (Some(client), Some(layer)) = (
+                self.clients.get_mut(&client_id),
+                self.app
+                    .pane_graphics
+                    .slots
+                    .get(&key)
+                    .and_then(|slot| slot.layer.as_ref()),
+            ) {
+                client
+                    .graphics_cache
+                    .trust_pane_layer(&key, host_image_id, layer);
+            }
+            if gate.respond_to.send(gate.success_response).is_err() {
+                self.retire_direct_gate(&key);
+                return true;
+            }
+            return true;
+        }
+
+        if let Some(client) = self.clients.get_mut(&client_id) {
+            client.direct_graphics = false;
+            client.graphics_cache.forget_pane_layer(&key, image_id);
+        }
+        self.app.direct_graphics_available = false;
+        if !self.install_inline_fallback(&key) {
+            self.retire_direct_gate(&key);
+            self.retire_all_direct_graphics();
+            return true;
+        }
+        let gate = self
+            .app
+            .pane_graphics
+            .slots
+            .get_mut(&key)
+            .filter(|slot| slot.stream_is_active())
+            .and_then(|slot| slot.direct_gate.take());
+        let Some(gate) = gate else {
+            self.retire_direct_gate(&key);
+            return true;
+        };
+        if gate.respond_to.send(gate.success_response).is_err() {
+            self.retire_direct_gate(&key);
+            return true;
+        }
+        self.retire_all_direct_graphics();
+        true
+    }
+
+    pub(super) fn expire_direct_graphics(&mut self, now: std::time::Instant) -> bool {
+        let expired = self
+            .app
+            .pane_graphics
+            .slots
+            .iter()
+            .filter_map(|(key, slot)| {
+                slot.direct_gate
+                    .as_ref()
+                    .filter(|gate| gate.deadline <= now)
+                    .map(|gate| (key.clone(), gate.client_id))
+            })
+            .collect::<Vec<_>>();
+        for (key, client_id) in &expired {
+            if let Some(slot) = self.app.pane_graphics.slots.get(key) {
+                if let Some(gate) = &slot.direct_gate {
+                    self.send_to_client(
+                        *client_id,
+                        ServerMessage::GraphicsTransmissionRetired {
+                            transfer_id: gate.transfer_id,
+                            image_id: slot.host_image_id,
+                        },
+                    );
+                }
+            }
+            if let Some(client) = self.clients.get_mut(client_id) {
+                client.direct_graphics = false;
+            }
+            self.retire_direct_gate(key);
+        }
+        if !expired.is_empty() {
+            self.app.direct_graphics_available = false;
+            self.retire_all_direct_graphics();
+        }
+        !expired.is_empty()
+    }
+
+    pub(super) fn retire_all_direct_graphics(&mut self) {
+        let keys = self
+            .app
+            .pane_graphics
+            .slots
+            .iter()
+            .filter(|(_, slot)| {
+                slot.layer
+                    .as_ref()
+                    .is_some_and(crate::app::pane_graphics::Layer::terminal_only)
+            })
+            .map(|(key, _)| key.clone())
+            .collect::<Vec<_>>();
+        for key in keys {
+            self.retire_direct_gate(&key);
+        }
+    }
+
+    pub(super) fn retire_direct_graphics_for_client(&mut self, client_id: u64) {
+        let keys = self
+            .app
+            .pane_graphics
+            .slots
+            .iter()
+            .filter_map(|(key, slot)| {
+                let layer = slot.layer.as_ref()?;
+                let owned = layer.resident_client() == Some(client_id)
+                    || (layer.direct_lease().is_some()
+                        && slot
+                            .direct_gate
+                            .as_ref()
+                            .is_some_and(|gate| gate.client_id == client_id));
+                (layer.terminal_only() && owned).then(|| key.clone())
+            })
+            .collect::<Vec<_>>();
+        for key in keys {
+            self.retire_direct_gate(&key);
+        }
+    }
+
     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 {
@@ -117,74 +522,62 @@ impl HeadlessServer {
 
             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,
-                    self.app.state.view.tab_surface(),
-                    cell_size,
-                    &mut next_graphics_cache,
-                ));
+            let encoded = crate::kitty_graphics::encode_local_pane_graphics(
+                &self.app.state,
+                &self.app.pane_graphics,
+                &self.app.terminal_runtimes,
+                self.app.state.view.tab_surface(),
+                cell_size,
+                Some(crate::kitty_graphics::HEADLESS_GRAPHICS_TRANSACTION_BUDGET),
+                &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"
-                );
+            prepared.push((client_id, encoded, next_graphics_cache));
+        }
+
+        let mut broken_clients = Vec::new();
+        for (client_id, encoded, next_graphics_cache) in prepared {
+            let Some(client) = self.clients.get_mut(&client_id) else {
                 continue;
-            }
-            let serialized = if bytes.is_empty() {
+            };
+            let serialized = if encoded.bytes.is_empty() {
                 None
             } else {
-                let serialize_started = crate::render_prof::timer();
-                let framed = match Self::frame_server_message_with_max(
-                    &ServerMessage::Graphics { bytes },
+                match Self::frame_server_message_with_max(
+                    &ServerMessage::Graphics {
+                        bytes: frame_pane_graphics(encoded.bytes),
+                    },
                     MAX_GRAPHICS_FRAME_SIZE,
                 ) {
-                    Ok(framed) => framed,
-                    Err(err) => {
-                        warn!(client_id, err = %err, "failed to serialize retained graphics");
+                    Ok(serialized) => Some(serialized),
+                    Err(_) => {
+                        crate::render_prof::event("retained_graphics_fallback.oversized");
                         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 result = match (serialized, client.writer.as_ref()) {
+                (None, _) => Ok(()),
+                (Some(bytes), Some(writer)) => writer.render.try_send(bytes),
+                (Some(bytes), None) => Err(std::sync::mpsc::TrySendError::Disconnected(bytes)),
             };
-            let Some(writer) = client.writer.as_ref().cloned() else {
-                broken_clients.push(client_id);
-                continue;
-            };
-            match writer.render.try_send(serialized) {
+            match result {
                 Ok(()) => {
                     client.graphics_cache = next_graphics_cache;
-                    client.clear_deferred_render();
+                    if encoded.incomplete {
+                        client.defer_full_render();
+                        deferred = true;
+                    } else {
+                        client.clear_deferred_render();
+                    }
                     crate::render_prof::event("retained_graphics.sent");
                 }
                 Err(std::sync::mpsc::TrySendError::Full(_)) => {
-                    client.defer_pane_graphics_render();
+                    client.defer_full_render();
                     deferred = true;
-                    crate::render_prof::event("retained_graphics.deferred");
                 }
                 Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
-                    broken_clients.push(client_id);
+                    broken_clients.push(client_id)
                 }
             }
         }
diff --git a/src/server/headless/tests/pane_graphics.rs b/src/server/headless/tests/pane_graphics.rs
index 1eba9816..f9bd0a01 100644
--- a/src/server/headless/tests/pane_graphics.rs
+++ b/src/server/headless/tests/pane_graphics.rs
@@ -1,12 +1,52 @@
 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 receive_render(receiver: &std::sync::mpsc::Receiver<Vec<u8>>, timeout: Duration) -> Vec<u8> {
+    receiver.recv_timeout(timeout).unwrap()
+}
+
+#[tokio::test]
+async fn cold_redraw_advances_one_bounded_layer_after_each_send() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"cold redraw");
+    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,
+    };
+    const LAYERS: usize = 8;
+    for index in 0..LAYERS {
+        set_named_graphics_layer(
+            &mut server,
+            pane_id,
+            &format!("layer-{index:02}"),
+            vec![index as u8; 1024 * 1024],
+            index as i32,
+        );
+    }
+
+    fill_render_lane(&server);
+    server.render_and_stream();
+    assert!(server.clients[&1].graphics_cache.is_empty());
+    let _older = client_rx.recv_timeout(Duration::from_secs(1)).unwrap();
+
+    for expected in 1..=LAYERS {
+        server.render_and_stream();
+        let bytes = client_rx.recv_timeout(Duration::from_secs(5)).unwrap();
+        assert!(bytes.len() <= MAX_GRAPHICS_FRAME_SIZE + 4);
+        let frame = read_server_frame(bytes);
+        assert_eq!(
+            frame
+                .graphics
+                .windows(4)
+                .filter(|part| *part == b"a=t,")
+                .count(),
+            1
+        );
+        assert_eq!(
+            server.clients[&1].graphics_cache.test_image_count(),
+            expected
+        );
+    }
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
 }
 
 fn enable_graphics_and_render(
@@ -19,39 +59,72 @@ fn enable_graphics_and_render(
         height_px: 20,
     };
     server.render_and_stream();
-    read_server_frame(
-        client_rx
-            .recv_timeout(Duration::from_millis(100))
-            .expect("initial frame"),
-    )
+    read_server_frame(receive_render(client_rx, Duration::from_millis(100)))
+}
+
+fn graphics_key(pane_id: crate::layout::PaneId) -> crate::app::pane_graphics::Key {
+    (pane_id, api::schema::PANE_GRAPHICS_PRIMARY_LAYER_ID.into())
+}
+
+fn active_gate() -> std::sync::Arc<std::sync::atomic::AtomicBool> {
+    std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true))
 }
 
 fn set_graphics_layer(server: &mut HeadlessServer, pane_id: crate::layout::PaneId, data: Vec<u8>) {
-    server.app.state.pane_graphics_layers.insert(
+    set_named_graphics_layer(
+        server,
         pane_id,
-        crate::app::state::PaneGraphicsLayer::new(
-            api::schema::PaneGraphicsFormat::Png,
-            1,
-            1,
-            data,
-            api::schema::PaneGraphicsPlacementParams::default(),
-        ),
+        api::schema::PANE_GRAPHICS_PRIMARY_LAYER_ID,
+        data,
+        0,
     );
 }
 
+fn set_named_graphics_layer(
+    server: &mut HeadlessServer,
+    pane_id: crate::layout::PaneId,
+    layer_id: &str,
+    data: Vec<u8>,
+    z_index: i32,
+) {
+    let key = (pane_id, layer_id.into());
+    let host_image_id = server.app.pane_graphics.reserve_image_id(&key).unwrap();
+    let layer = crate::app::pane_graphics::Layer::inline(
+        api::schema::PaneGraphicsFormat::Png,
+        1,
+        1,
+        data,
+        Default::default(),
+        z_index,
+    );
+    server.app.pane_graphics.slots.insert(
+        key,
+        crate::app::pane_graphics::Slot::test(host_image_id, Some(layer)),
+    );
+}
+
+fn set_stream_owner(server: &mut HeadlessServer, pane_id: crate::layout::PaneId, owner: &str) {
+    let key = graphics_key(pane_id);
+    if let Some(slot) = server.app.pane_graphics.slots.get_mut(&key) {
+        slot.stream_owner = Some(owner.into());
+        slot.stream_active = Some(active_gate());
+    } else {
+        let host_image_id = server.app.pane_graphics.reserve_image_id(&key).unwrap();
+        let mut slot = crate::app::pane_graphics::Slot::test(host_image_id, None);
+        slot.stream_owner = Some(owner.into());
+        slot.stream_active = Some(active_gate());
+        server.app.pane_graphics.slots.insert(key, slot);
+    }
+}
+
 fn fill_render_lane(server: &HeadlessServer) {
     let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
         .expect("dummy frame");
-    server
-        .clients
-        .get(&1)
-        .unwrap()
+    server.clients[&1]
         .writer
         .as_ref()
         .unwrap()
-        .render
-        .try_send(queued)
-        .expect("pre-fill render lane");
+        .test_fill_render(queued);
 }
 
 fn stream_set_message(
@@ -68,6 +141,8 @@ fn stream_set_message(
                 method: api::schema::Method::PaneGraphicsStreamSet(
                     api::schema::PaneGraphicsSetParams {
                         pane_id: pane_id.into(),
+                        layer_id: None,
+                        z_index: 0,
                         owner: owner.into(),
                         format: api::schema::PaneGraphicsFormat::Png,
                         image_width: 1,
@@ -80,12 +155,115 @@ fn stream_set_message(
             },
             respond_to,
             response_write_complete: None,
+            stream_active: None,
         },
         response_rx,
     )
 }
 
 #[tokio::test]
+async fn pixel_mouse_activation_requires_graphics_demand_not_direct_transport() {
+    let (mut server, _client_rx, pane_id) =
+        retained_test_server(b"\x1b[?1003h\x1b[?1006h\x1b[?1016h");
+    let (writer, control_rx, _render_rx) = test_client_writer();
+    let client = server.clients.get_mut(&1).unwrap();
+    client.writer = Some(writer);
+    client.direct_graphics = false;
+    client.pixel_mouse = true;
+    client.host_mouse_capture_active = None;
+    client.host_sgr_pixels_active = None;
+    server.app.direct_graphics_available = false;
+
+    server.stream_host_mouse_capture_mode();
+    assert!(matches!(
+        read_server_message(control_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
+        ServerMessage::MouseCapture {
+            enabled: true,
+            sgr_pixels: false
+        }
+    ));
+
+    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3]);
+    server.stream_host_mouse_capture_mode();
+    assert!(matches!(
+        read_server_message(control_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
+        ServerMessage::MouseCapture {
+            enabled: true,
+            sgr_pixels: true
+        }
+    ));
+}
+
+#[tokio::test]
+async fn pixel_input_metadata_cannot_resize_authoritative_client_state() {
+    let (mut server, _client_rx, pane_id) =
+        retained_test_server(b"\x1b[?1003h\x1b[?1006h\x1b[?1016h");
+    set_graphics_layer(&mut server, pane_id, vec![1]);
+    let client = server.clients.get_mut(&1).unwrap();
+    client.pixel_mouse = true;
+    client.host_sgr_pixels_active = Some(true);
+    server.foreground_client_id = None;
+    assert!(!server.handle_server_event(ServerEvent::ClientInputPixels {
+        client_id: 1,
+        data: b"\x1b[<0;500;300M".to_vec(),
+        geometry: crate::input::mouse::HostGeometry::new(80, 24, 800, 480).unwrap(),
+    }));
+    server.clients.get_mut(&1).unwrap().cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    for (geometry, data) in [
+        (
+            crate::input::mouse::HostGeometry::new(100, 30, 1_000, 600).unwrap(),
+            b"\x1b[<0;500;300M".as_slice(),
+        ),
+        (
+            crate::input::mouse::HostGeometry::new(80, 24, 960, 480).unwrap(),
+            b"\x1b[<0;500;300M",
+        ),
+        (
+            crate::input::mouse::HostGeometry::new(80, 24, 800, 480).unwrap(),
+            b"\x1b[<0;0;1M",
+        ),
+    ] {
+        assert!(!server.handle_server_event(ServerEvent::ClientInputPixels {
+            client_id: 1,
+            data: data.to_vec(),
+            geometry,
+        }));
+    }
+    assert_eq!(server.clients[&1].terminal_size, (80, 24));
+    assert_eq!(
+        (server.effective_size, server.foreground_client_id),
+        ((80, 24), None)
+    );
+}
+
+#[test]
+fn direct_eligibility_is_installed_with_the_client_connection() {
+    let mut server = test_headless_server();
+    let (writer, _control_rx, _render_rx) = test_client_writer();
+
+    assert!(server.handle_server_event(ServerEvent::ClientConnected {
+        client_id: 7,
+        cols: 80,
+        rows: 24,
+        cell_width_px: 10,
+        cell_height_px: 20,
+        render_encoding: RenderEncoding::SemanticFrame,
+        keybindings: None,
+        direct_attach_requested: false,
+        direct_graphics: true,
+        writer,
+    }));
+
+    let client = server.clients.get(&7).expect("connected client");
+    assert!(client.direct_graphics);
+    assert_eq!(server.foreground_client_id, Some(7));
+    assert!(server.app.direct_graphics_available);
+}
+
+#[tokio::test]
 async fn focus_repaint_preserves_uploaded_graphics() {
     let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
     let (client_2_writer, _client_2_control_rx, client_2_rx) = test_client_writer();
@@ -109,11 +287,8 @@ async fn focus_repaint_preserves_uploaded_graphics() {
     let initial_graphics = String::from_utf8_lossy(&initial.graphics);
     assert!(initial_graphics.contains("a=t"));
     assert!(initial_graphics.contains("a=p"));
-    let client_2_initial = read_server_frame(
-        client_2_rx
-            .recv_timeout(Duration::from_millis(100))
-            .expect("second client initial frame"),
-    );
+    let client_2_initial =
+        read_server_frame(receive_render(&client_2_rx, Duration::from_millis(100)));
     assert!(String::from_utf8_lossy(&client_2_initial.graphics).contains("a=t"));
 
     assert!(server.handle_server_event(ServerEvent::ClientInput {
@@ -123,12 +298,68 @@ async fn focus_repaint_preserves_uploaded_graphics() {
     assert_eq!(server.foreground_client_id, Some(2));
     server.render_and_stream();
 
-    let focused = read_server_frame(
-        client_2_rx
-            .recv_timeout(Duration::from_millis(100))
-            .expect("focus redraw"),
+    let focused = read_server_frame(receive_render(&client_2_rx, Duration::from_millis(100)));
+    let focused_graphics = String::from_utf8_lossy(&focused.graphics);
+    assert!(focused_graphics.contains("a=p"));
+    assert!(!focused_graphics.contains("a=t"));
+}
+
+#[tokio::test]
+async fn resize_replays_placement_without_retransmitting_or_closing_stream() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3]);
+    set_stream_owner(&mut server, pane_id, "owner-resize");
+    let initial = enable_graphics_and_render(&mut server, &client_rx);
+    assert!(String::from_utf8_lossy(&initial.graphics).contains("a=t"));
+
+    for (cols, rows, cell_width_px, cell_height_px) in
+        [(100, 30, 10, 20), (100, 30, 12, 24), (100, 30, 12, 24)]
+    {
+        assert!(server.handle_server_event(ServerEvent::ClientResize {
+            client_id: 1,
+            cols,
+            rows,
+            cell_width_px,
+            cell_height_px,
+        }));
+        server.render_and_stream();
+        let frame = read_server_frame(receive_render(&client_rx, Duration::from_millis(100)));
+        let graphics = String::from_utf8_lossy(&frame.graphics);
+        assert!(!graphics.contains("a=t"));
+        assert!(graphics.contains("a=p"));
+    }
+    assert_eq!(
+        server
+            .app
+            .pane_graphics
+            .slots
+            .get(&graphics_key(pane_id))
+            .and_then(|slot| slot.stream_owner.as_deref()),
+        Some("owner-resize")
     );
-    assert!(focused.graphics.is_empty());
+}
+
+#[tokio::test]
+async fn graphics_pruning_preserves_live_panes_and_removes_closed_panes() {
+    let (mut server, _client_rx, pane_id) = retained_test_server(b"aaaa");
+    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3]);
+
+    assert!(!server
+        .app
+        .pane_graphics
+        .retain_live_panes(&server.app.state));
+    assert!(server
+        .app
+        .pane_graphics
+        .slots
+        .contains_key(&graphics_key(pane_id)));
+
+    server.app.state.workspaces.clear();
+    assert!(server
+        .app
+        .pane_graphics
+        .retain_live_panes(&server.app.state));
+    assert!(server.app.pane_graphics.slots.is_empty());
 }
 
 #[tokio::test]
@@ -164,58 +395,25 @@ async fn retained_update_sends_only_graphics_message() {
 }
 
 #[tokio::test]
-async fn retained_update_defers_on_full_render_lane() {
+async fn retained_graphics_stays_ordered_after_an_older_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);
     set_graphics_layer(&mut server, pane_id, vec![4, 5, 6]);
-
+    let older = client_rx.recv_timeout(Duration::from_secs(1)).unwrap();
     assert_eq!(
         server.render_retained_graphics_update_and_stream(),
-        RetainedGraphicsOutcome::Deferred
+        RetainedGraphicsOutcome::Sent
     );
-    let client = server.clients.get(&1).unwrap();
-    assert_eq!(client.deferred_render(), DeferredRender::Graphics);
+    let graphics = client_rx.recv_timeout(Duration::from_secs(1)).unwrap();
     assert!(matches!(
-        read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
+        read_server_message(older),
         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_repaint();
-    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
-    );
+    assert!(matches!(
+        read_server_message(graphics),
+        ServerMessage::Graphics { .. }
+    ));
 }
 
 #[tokio::test]
@@ -247,6 +445,64 @@ async fn retained_update_falls_back_for_mixed_app_geometry() {
 }
 
 #[test]
+fn stream_open_gate_is_owned_by_the_layer_and_cancels_on_removal() {
+    let mut server = test_headless_server();
+    server.app.state.kitty_graphics_enabled = true;
+    let workspace = crate::workspace::Workspace::test_new("gated");
+    let pane_id = workspace.tabs[0].root_pane;
+    let public = format!("{}:p1", workspace.id);
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    let active = active_gate();
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+
+    server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "open-gated".into(),
+            method: api::schema::Method::PaneGraphicsStreamOpen(
+                api::schema::PaneGraphicsStreamParams {
+                    pane_id: public.clone(),
+                    layer_id: None,
+                    z_index: 0,
+                    owner: "worker-1".into(),
+                },
+            ),
+        },
+        respond_to,
+        response_write_complete: None,
+        stream_active: Some(active.clone()),
+    });
+    assert!(
+        serde_json::from_str::<api::schema::SuccessResponse>(&response_rx.recv().unwrap()).is_ok()
+    );
+    let (frame, frame_response) =
+        stream_set_message("gated-frame", &public, "worker-1", vec![1, 2, 3]);
+    assert_eq!(
+        server.handle_api_request_with_render_impact(frame),
+        RenderImpact::Graphics
+    );
+    assert!(frame_response.recv().is_ok());
+    assert!(active.load(std::sync::atomic::Ordering::Acquire));
+    active.store(false, std::sync::atomic::Ordering::Release);
+    let (delayed, delayed_response) =
+        stream_set_message("delayed-frame", &public, "worker-1", vec![4, 5, 6]);
+    assert_eq!(
+        server.handle_api_request_with_render_impact(delayed),
+        RenderImpact::None
+    );
+    let error: api::schema::ErrorResponse =
+        serde_json::from_str(&delayed_response.recv().unwrap()).unwrap();
+    assert_eq!(error.error.code, "stream_closed");
+    assert!(server
+        .app
+        .pane_graphics
+        .slots
+        .remove(&graphics_key(pane_id))
+        .is_some());
+    assert!(!active.load(std::sync::atomic::Ordering::Acquire));
+}
+
+#[test]
 fn stream_set_has_graphics_only_render_impact() {
     let mut server = test_headless_server();
     let workspace = crate::workspace::Workspace::test_new("graphics");
@@ -256,11 +512,7 @@ fn stream_set_has_graphics_only_render_impact() {
     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());
+    set_stream_owner(&mut server, pane_id, "owner-a");
 
     let (request, response_rx) =
         stream_set_message("wrong-owner", &public_pane_id, "owner-b", vec![1, 2, 3]);
@@ -307,13 +559,15 @@ fn stream_set_has_graphics_only_render_impact() {
         RenderImpact::Full
     );
 
-    server.app.state.pane_graphics_streams.clear();
+    server.app.pane_graphics.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,
+                layer_id: None,
+                z_index: 0,
                 owner: String::new(),
                 format: api::schema::PaneGraphicsFormat::Png,
                 image_width: 1,
@@ -325,6 +579,7 @@ fn stream_set_has_graphics_only_render_impact() {
         },
         respond_to,
         response_write_complete: None,
+        stream_active: None,
     });
     assert_eq!(impact, RenderImpact::Full);
 }
@@ -345,6 +600,8 @@ fn rejected_or_stale_requests_do_not_schedule_rendering() {
             id: "disabled-set".into(),
             method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
                 pane_id: public_pane_id.clone(),
+                layer_id: None,
+                z_index: 0,
                 owner: String::new(),
                 format: api::schema::PaneGraphicsFormat::Png,
                 image_width: 1,
@@ -356,6 +613,7 @@ fn rejected_or_stale_requests_do_not_schedule_rendering() {
         },
         respond_to,
         response_write_complete: None,
+        stream_active: None,
     });
     assert!(!changed);
     let response = response_rx
@@ -370,11 +628,7 @@ fn rejected_or_stale_requests_do_not_schedule_rendering() {
     );
 
     server.app.state.kitty_graphics_enabled = true;
-    server
-        .app
-        .state
-        .pane_graphics_streams
-        .insert(pane_id, "current-owner".into());
+    set_stream_owner(&mut server, pane_id, "current-owner");
     let (respond_to, response_rx) = std::sync::mpsc::channel();
     let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
         request: api::schema::Request {
@@ -382,17 +636,25 @@ fn rejected_or_stale_requests_do_not_schedule_rendering() {
             method: api::schema::Method::PaneGraphicsStreamClose(
                 api::schema::PaneGraphicsStreamParams {
                     pane_id: public_pane_id,
+                    layer_id: None,
+                    z_index: 0,
                     owner: "stale-owner".into(),
                 },
             ),
         },
         respond_to,
         response_write_complete: None,
+        stream_active: None,
     });
     assert_eq!(impact, RenderImpact::None);
     assert_eq!(
-        server.app.state.pane_graphics_streams.get(&pane_id),
-        Some(&"current-owner".to_string())
+        server
+            .app
+            .pane_graphics
+            .slots
+            .get(&graphics_key(pane_id))
+            .and_then(|slot| slot.stream_owner.as_deref()),
+        Some("current-owner")
     );
     assert!(serde_json::from_str::<api::schema::SuccessResponse>(
         &response_rx
@@ -401,3 +663,282 @@ fn rejected_or_stale_requests_do_not_schedule_rendering() {
     )
     .is_ok());
 }
+
+#[cfg(unix)]
+fn direct_gate_server(
+    data: &[u8],
+) -> (
+    HeadlessServer,
+    crate::app::pane_graphics::Key,
+    std::sync::mpsc::Receiver<String>,
+) {
+    use std::io::Write as _;
+    use std::os::unix::fs::OpenOptionsExt as _;
+    let mut server = test_headless_server();
+    let workspace = crate::workspace::Workspace::test_new("direct-gate");
+    let pane_id = workspace.tabs[0].root_pane;
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    let key = graphics_key(pane_id);
+    let path = server
+        .app
+        .pane_graphics_files
+        .source_directory()
+        .unwrap()
+        .join("gate-frame");
+    let mut file = std::fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(&path)
+        .unwrap();
+    file.write_all(data).unwrap();
+    drop(file);
+    let lease = server
+        .app
+        .pane_graphics_files
+        .lease(&path, data.len())
+        .unwrap();
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    let layer =
+        crate::app::pane_graphics::Layer::direct(1, 1, lease.clone(), Default::default(), 0);
+    let mut slot = crate::app::pane_graphics::Slot::test((1 << 31) | 900, Some(layer));
+    slot.stream_owner = Some("owner".into());
+    slot.stream_active = Some(active_gate());
+    slot.direct_gate = Some(crate::app::pane_graphics::DirectGate {
+        transfer_id: lease.fingerprint(),
+        client_id: 7,
+        deadline: std::time::Instant::now() + Duration::from_secs(1),
+        written: true,
+        success_response: "ack".into(),
+        respond_to,
+    });
+    server.app.pane_graphics.slots.insert(key.clone(), slot);
+    (server, key, response_rx)
+}
+
+#[cfg(unix)]
+fn direct_ids(server: &HeadlessServer, key: &crate::app::pane_graphics::Key) -> (u64, u32) {
+    let slot = &server.app.pane_graphics.slots[key];
+    (
+        slot.direct_gate.as_ref().unwrap().transfer_id,
+        slot.host_image_id,
+    )
+}
+
+#[cfg(unix)]
+fn add_direct_client(server: &mut HeadlessServer, client_id: u64) {
+    let (writer, control_rx, render_rx) = test_client_writer();
+    std::mem::forget((control_rx, render_rx));
+    let mut client = ClientConnection::new(
+        (80, 24),
+        crate::kitty_graphics::HostCellSize {
+            width_px: 10,
+            height_px: 20,
+        },
+        crate::terminal_theme::TerminalTheme::default(),
+        None,
+        1,
+        RenderEncoding::SemanticFrame,
+        Some(writer),
+    );
+    client.direct_graphics = true;
+    client.pixel_mouse = true;
+    server.clients.insert(client_id, client);
+}
+
+#[cfg(unix)]
+#[test]
+fn terminal_response_deadline_starts_only_after_client_flush() {
+    let (mut server, key, _response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    let slot = server.app.pane_graphics.slots.get_mut(&key).unwrap();
+    let gate = slot.direct_gate.as_mut().unwrap();
+    gate.written = false;
+    let (transfer_id, image_id) = (gate.transfer_id, slot.host_image_id);
+    assert!(!server.complete_direct_graphics(7, transfer_id, image_id, true));
+    assert!(!server.start_direct_graphics_response(7, transfer_id, image_id));
+    let gate = server.app.pane_graphics.slots[&key]
+        .direct_gate
+        .as_ref()
+        .unwrap();
+    assert!(gate.written && gate.deadline > std::time::Instant::now());
+}
+
+#[cfg(unix)]
+#[test]
+fn outer_timeout_covers_both_direct_phases_and_cancellation_blocks_late_results() {
+    assert!(
+        crate::app::pane_graphics::DIRECT_OUTER_TIMEOUT
+            > crate::app::pane_graphics::DIRECT_DELIVERY_TIMEOUT
+                + crate::app::pane_graphics::DIRECT_RESPONSE_TIMEOUT
+    );
+    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    let slot = server.app.pane_graphics.slots.get_mut(&key).unwrap();
+    slot.stream_active
+        .as_ref()
+        .unwrap()
+        .store(false, std::sync::atomic::Ordering::Release);
+    let (transfer_id, image_id) = (
+        slot.direct_gate.as_ref().unwrap().transfer_id,
+        slot.host_image_id,
+    );
+
+    assert!(!server.complete_direct_graphics(7, transfer_id, image_id, true));
+    assert!(response_rx.try_recv().is_err());
+    assert!(server.app.pane_graphics.slots[&key]
+        .layer
+        .as_ref()
+        .unwrap()
+        .direct_lease()
+        .is_some());
+}
+
+#[cfg(unix)]
+#[test]
+fn matching_terminal_ok_releases_producer_and_acknowledges() {
+    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    let (transfer_id, image_id) = direct_ids(&server, &key);
+
+    assert!(server.complete_direct_graphics(7, transfer_id, image_id, true));
+
+    assert_eq!(response_rx.recv().unwrap(), "ack");
+    let layer = server.app.pane_graphics.slots[&key].layer.as_ref().unwrap();
+    assert!(layer.terminal_only());
+    assert!(layer.direct_lease().is_none());
+}
+
+#[cfg(unix)]
+#[test]
+fn explicit_terminal_error_acks_only_after_owned_inline_fallback() {
+    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    add_direct_client(&mut server, 7);
+    let (transfer_id, image_id) = direct_ids(&server, &key);
+    let layer = server.app.pane_graphics.slots[&key].layer.as_ref().unwrap();
+    server
+        .clients
+        .get_mut(&7)
+        .unwrap()
+        .graphics_cache
+        .trust_pane_layer(&key, image_id, layer);
+    assert!(server.complete_direct_graphics(7, transfer_id, image_id, false));
+
+    let layer = server.app.pane_graphics.slots[&key].layer.as_ref().unwrap();
+    assert_eq!(
+        (
+            response_rx.recv().unwrap(),
+            layer.inline_data(),
+            server.clients[&7].direct_graphics,
+            server.clients[&7].pixel_mouse,
+        ),
+        ("ack".into(), Some([1, 2, 3, 4].as_slice()), false, true)
+    );
+    assert!(server.clients[&7].graphics_cache.is_empty());
+}
+
+#[cfg(unix)]
+#[test]
+fn unwritten_direct_full_falls_back_without_stickiness_but_disconnect_retires() {
+    for error in [
+        std::sync::mpsc::TrySendError::Full(Vec::new()),
+        std::sync::mpsc::TrySendError::Disconnected(Vec::new()),
+    ] {
+        let should_ack = matches!(error, std::sync::mpsc::TrySendError::Full(_));
+        let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+        add_direct_client(&mut server, 7);
+        let gate = server
+            .app
+            .pane_graphics
+            .slots
+            .get_mut(&key)
+            .and_then(|slot| slot.direct_gate.take())
+            .unwrap();
+        let result = server.handle_unwritten_direct_failure(
+            &key,
+            gate.success_response,
+            gate.respond_to,
+            error,
+        );
+        let inline = server
+            .app
+            .pane_graphics
+            .slots
+            .get(&key)
+            .and_then(|slot| slot.layer.as_ref()?.inline_data())
+            .is_some();
+        assert_eq!(
+            (
+                result,
+                response_rx.try_recv().ok().as_deref() == Some("ack"),
+                inline,
+                server.clients[&7].direct_graphics,
+            ),
+            (should_ack, should_ack, should_ack, true)
+        );
+    }
+}
+
+#[cfg(unix)]
+#[test]
+fn client_loss_retires_only_its_direct_stream() {
+    let (mut pending, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    pending.retire_direct_graphics_for_client(8);
+    assert!(pending.app.pane_graphics.slots.contains_key(&key));
+    pending.retire_direct_graphics_for_client(7);
+    assert!(!pending.app.pane_graphics.slots.contains_key(&key));
+    assert!(response_rx.recv().is_err());
+
+    let (mut resident, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    let slot = resident.app.pane_graphics.slots.get(&key).unwrap();
+    assert!(resident.complete_direct_graphics(
+        7,
+        slot.direct_gate.as_ref().unwrap().transfer_id,
+        slot.host_image_id,
+        true,
+    ));
+    assert_eq!(response_rx.recv().unwrap(), "ack");
+    resident.retire_direct_graphics_for_client(8);
+    assert!(resident.app.pane_graphics.slots.contains_key(&key));
+    resident.retire_direct_graphics_for_client(7);
+    assert!(!resident.app.pane_graphics.slots.contains_key(&key));
+}
+
+#[cfg(unix)]
+#[test]
+fn pane_removal_and_shutdown_drop_direct_without_ack() {
+    let setups: [fn(&mut HeadlessServer); 2] = [
+        |server| server.app.state.workspaces.clear(),
+        |server| server.shutting_down = true,
+    ];
+    for setup in setups {
+        let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+        let (transfer_id, image_id) = direct_ids(&server, &key);
+        setup(&mut server);
+        assert!(!server.complete_direct_graphics(7, transfer_id, image_id, true));
+        assert!(response_rx.recv().is_err());
+        assert!(!server.app.pane_graphics.slots.contains_key(&key));
+    }
+}
+
+#[cfg(unix)]
+#[test]
+fn timeout_retires_stream_without_producer_ack() {
+    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
+    add_direct_client(&mut server, 7);
+    server
+        .app
+        .pane_graphics
+        .slots
+        .get_mut(&key)
+        .unwrap()
+        .direct_gate
+        .as_mut()
+        .unwrap()
+        .deadline = std::time::Instant::now() - Duration::from_millis(1);
+
+    assert!(server.expire_direct_graphics(std::time::Instant::now()));
+
+    assert!(response_rx.recv().is_err());
+    assert!(!server.app.pane_graphics.slots.contains_key(&key));
+    assert!(!server.clients[&7].direct_graphics);
+    assert!(server.clients[&7].pixel_mouse);
+}