feat!: replace the custom graphics api with optimized native kitty rendering (#4561)

* feat!: replace the custom graphics api with optimized native kitty rendering

* fix: enable safe kitty optimizations automatically

---------

Co-authored-by: JJ Liebig <jonathan.liebig@gmail.com>

15 files changed, 2332 insertions, 1943 deletions

diff --git a/src/server/client_shell.rs b/src/server/client_shell.rs
index d542896a..d79c89cf 100644
--- a/src/server/client_shell.rs
+++ b/src/server/client_shell.rs
@@ -274,6 +274,7 @@ pub(super) struct RenderedPaneSurface {
     pub(super) popup: Option<Box<protocol::ClientShellPopupSurface>>,
     pub(super) graphics: protocol::SurfaceGraphicsScene,
     pub(super) graphics_delivery: crate::kitty_graphics::surface::DeliveryCache,
+    pub(super) graphics_sources: crate::kitty_graphics::surface::SourceFiles,
 }
 
 #[derive(Debug)]
@@ -438,16 +439,17 @@ pub(super) fn render_pane_surface(
     let popup = show_popup
         .then(|| render_popup_surface(app, area, resize_panes, cell_size))
         .flatten();
-    let (graphics, next_graphics_delivery) = crate::server::client_shell_graphics::collect(
-        app,
-        &layout.pane_infos,
-        &layout.split_borders,
-        popup.as_deref(),
-        target,
-        cell_size,
-        graphics_delivery,
-        client_id,
-    );
+    let (graphics, next_graphics_delivery, graphics_sources) =
+        crate::server::client_shell_graphics::collect(
+            app,
+            &layout.pane_infos,
+            &layout.split_borders,
+            popup.as_deref(),
+            target,
+            cell_size,
+            graphics_delivery,
+            client_id,
+        );
     if let Some(target) = target {
         for (&pane_id, &(epoch, _)) in &content_revisions_before {
             if let Some(runtime) = app.state.runtime_for_pane_in_workspace(
@@ -488,6 +490,7 @@ pub(super) fn render_pane_surface(
         popup,
         graphics,
         graphics_delivery: next_graphics_delivery,
+        graphics_sources,
     })
 }
 
diff --git a/src/server/client_shell_graphics.rs b/src/server/client_shell_graphics.rs
index 00355230..d0af03a9 100644
--- a/src/server/client_shell_graphics.rs
+++ b/src/server/client_shell_graphics.rs
@@ -1,4 +1,4 @@
-use crate::kitty_graphics::surface::DeliveryCache;
+use crate::kitty_graphics::surface::{DeliveryCache, SourceFiles};
 use crate::protocol::{ClientShellPopupSurface, SurfaceGraphicsScene};
 
 pub(crate) fn collect_retained(
@@ -8,7 +8,7 @@ pub(crate) fn collect_retained(
     cell_size: crate::kitty_graphics::HostCellSize,
     delivered: &DeliveryCache,
     client_id: u64,
-) -> Option<(SurfaceGraphicsScene, DeliveryCache)> {
+) -> Option<(SurfaceGraphicsScene, DeliveryCache, SourceFiles)> {
     let rect = |rect: crate::protocol::SurfaceRect| {
         ratatui::layout::Rect::new(rect.x, rect.y, rect.width, rect.height)
     };
@@ -52,7 +52,7 @@ pub(crate) fn collect(
     cell_size: crate::kitty_graphics::HostCellSize,
     delivered: &DeliveryCache,
     client_id: u64,
-) -> (SurfaceGraphicsScene, DeliveryCache) {
+) -> (SurfaceGraphicsScene, DeliveryCache, SourceFiles) {
     let popup_content_size = popup.map(|popup| (popup.frame.width, popup.frame.height));
     crate::kitty_graphics::surface::collect_scene(
         app,
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 0d297fb9..465408f5 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -132,6 +132,24 @@ impl ClientWriter {
         self.render.queue.discard_pending_render();
     }
 
+    #[cfg(all(test, unix))]
+    pub(crate) fn test_paused() -> Self {
+        let queue = ClientWriterQueue::new();
+        Self {
+            control: ClientControlWriter::queue(queue.clone()),
+            render: ClientRenderWriter::queue(queue),
+        }
+    }
+
+    #[cfg(all(test, unix))]
+    pub(crate) fn test_drain(&self) -> Vec<Vec<u8>> {
+        let mut state = self.render.queue.lock_state();
+        let mut frames = state.control.drain(..).collect::<Vec<_>>();
+        frames.extend(state.ordered.drain(..));
+        frames.extend(state.render.take());
+        frames
+    }
+
     #[cfg(test)]
     pub(crate) fn test_fill_render(&self, data: Vec<u8>) {
         self.render.try_send(data).unwrap();
diff --git a/src/server/headless.rs b/src/server/headless.rs
index bba59498..fe8a67e8 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -76,8 +76,8 @@ mod bootstrap;
 mod client_views;
 mod endpoint_requests;
 mod lifecycle;
+mod native_graphics;
 mod notifications;
-mod pane_graphics;
 mod render;
 mod retained_surface;
 mod surface_interest;
@@ -132,31 +132,16 @@ enum LoopEvent {
     RenderRequested,
 }
 
-#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+/// Presentation work caused by a server event.
+///
+/// Keeping this classification separate from event delivery makes the input
+/// contract explicit: events may reach a PTY without necessarily repainting.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum RenderImpact {
-    #[default]
     None,
-    Graphics,
     Full,
 }
 
-impl RenderImpact {
-    fn merge(&mut self, other: Self) {
-        *self = (*self).max(other);
-    }
-}
-
-fn record_render_impact(source: &'static str, impact: RenderImpact) {
-    let event = match (source, impact) {
-        ("api_requests", RenderImpact::Graphics) => "graphics_render_cause.api_requests",
-        ("api_requests", RenderImpact::Full) => "full_render_cause.api_requests",
-        ("server_events", RenderImpact::Graphics) => "graphics_render_cause.server_events",
-        ("server_events", RenderImpact::Full) => "full_render_cause.server_events",
-        _ => return,
-    };
-    crate::render_prof::event(event);
-}
-
 // ---------------------------------------------------------------------------
 // Constants
 // ---------------------------------------------------------------------------
@@ -205,6 +190,7 @@ pub struct HeadlessServer {
     client_socket_path: PathBuf,
     client_socket_identity: SocketFileIdentity,
     clients: HashMap<u64, ClientConnection>,
+    native_graphics: native_graphics::NativeGraphics,
     #[cfg(unix)]
     next_client_id: u64,
     /// The client currently driving session-wide host presentation and side effects.
@@ -350,6 +336,7 @@ impl HeadlessServer {
             client_socket_path: client_path,
             client_socket_identity,
             clients: HashMap::new(),
+            native_graphics: Default::default(),
             #[cfg(unix)]
             next_client_id: 1,
             foreground_client_id: None,
@@ -463,22 +450,7 @@ impl HeadlessServer {
             }
 
             // 3. Drain API requests.
-            if self.pane_graphics_runtime_active() {
-                let api_impact = self.drain_api_requests_with_render_impact();
-                record_render_impact("api_requests", api_impact);
-                match api_impact {
-                    RenderImpact::None => {}
-                    RenderImpact::Graphics => {
-                        needs_render = true;
-                        needs_graphics_render = true;
-                    }
-                    RenderImpact::Full => {
-                        needs_render = true;
-                        needs_full_render = true;
-                        needs_graphics_render = false;
-                    }
-                }
-            } else if self.drain_api_requests_with_shutdown_check() {
+            if self.drain_api_requests_with_shutdown_check() {
                 needs_render = true;
                 needs_full_render = true;
                 crate::render_prof::event("full_render_cause.api_requests");
@@ -494,22 +466,7 @@ impl HeadlessServer {
             self.accept_client_connections()?;
 
             // 5. Drain server events from client threads.
-            if self.pane_graphics_runtime_active() {
-                let server_impact = self.drain_server_events_with_render_impact();
-                record_render_impact("server_events", server_impact);
-                match server_impact {
-                    RenderImpact::None => {}
-                    RenderImpact::Graphics => {
-                        needs_render = true;
-                        needs_graphics_render = true;
-                    }
-                    RenderImpact::Full => {
-                        needs_render = true;
-                        needs_full_render = true;
-                        needs_graphics_render = false;
-                    }
-                }
-            } else if self.drain_server_events() {
+            if self.drain_server_events() {
                 needs_render = true;
                 needs_full_render = true;
                 crate::render_prof::event("full_render_cause.server_events");
@@ -541,13 +498,10 @@ impl HeadlessServer {
                 crate::render_prof::event("full_render_cause.default_workspace");
             }
 
-            if self.retain_live_pane_graphics() {
-                needs_render = true;
-                needs_graphics_render = true;
-            }
-            if self.expire_direct_graphics(now) {
+            if self.expire_native_graphics(now) {
                 needs_render = true;
-                needs_graphics_render = true;
+                needs_full_render = true;
+                needs_graphics_render = false;
             }
 
             self.drain_client_config_reload_request();
@@ -689,43 +643,13 @@ impl HeadlessServer {
                     }
                 }
                 LoopEvent::Api(msg) => {
-                    if self.pane_graphics_runtime_active() {
-                        let impact = self.handle_api_request_with_render_impact(*msg);
-                        record_render_impact("api_requests", impact);
-                        match impact {
-                            RenderImpact::None => {}
-                            RenderImpact::Graphics => {
-                                needs_render = true;
-                                needs_graphics_render = true;
-                            }
-                            RenderImpact::Full => {
-                                needs_render = true;
-                                needs_full_render = true;
-                                needs_graphics_render = false;
-                            }
-                        }
-                    } else if self.handle_api_request_with_shutdown_check(*msg) {
+                    if self.handle_api_request_with_shutdown_check(*msg) {
                         needs_render = true;
                         needs_full_render = true;
                     }
                 }
                 LoopEvent::ServerEvent(ev) => {
-                    if self.pane_graphics_runtime_active() {
-                        let impact = self.handle_server_event_with_render_impact(ev);
-                        record_render_impact("server_events", impact);
-                        match impact {
-                            RenderImpact::None => {}
-                            RenderImpact::Graphics => {
-                                needs_render = true;
-                                needs_graphics_render = true;
-                            }
-                            RenderImpact::Full => {
-                                needs_render = true;
-                                needs_full_render = true;
-                                needs_graphics_render = false;
-                            }
-                        }
-                    } else if self.handle_server_event(ev) {
+                    if self.handle_server_event_with_render_impact(ev) == RenderImpact::Full {
                         needs_render = true;
                         needs_full_render = true;
                     }
@@ -825,15 +749,11 @@ 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 = self.headless_size;
             self.app.state.outer_terminal_focus = None;
@@ -966,53 +886,12 @@ impl HeadlessServer {
         })
     }
 
-    fn existing_direct_graphics_client(&self) -> Option<u64> {
-        self.app
-            .pane_graphics
-            .slots
-            .values()
-            .filter_map(crate::app::pane_graphics::Slot::direct_client)
-            .filter_map(|client_id| {
-                let client = self.clients.get(&client_id)?;
-                self.client_supports_direct_graphics(client_id)
-                    .then_some((client.last_activity, client_id))
-            })
-            .max()
-            .map(|(_, client_id)| client_id)
-    }
-
-    fn direct_graphics_client(&self) -> Option<u64> {
-        self.existing_direct_graphics_client().or_else(|| {
-            self.foreground_client_id
-                .filter(|client_id| self.client_supports_direct_graphics(*client_id))
-        })
-    }
-
-    fn direct_graphics_client_for_key(&self, key: &crate::app::pane_graphics::Key) -> Option<u64> {
-        match self
-            .app
-            .pane_graphics
-            .slots
-            .get(key)
-            .and_then(crate::app::pane_graphics::Slot::direct_client)
-        {
-            Some(client_id) => self
-                .client_supports_direct_graphics(client_id)
-                .then_some(client_id),
-            None => self.direct_graphics_client(),
-        }
-    }
-
-    fn direct_graphics_available(&self) -> bool {
-        self.direct_graphics_client().is_some()
-    }
-
     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);
+        self.disconnect_native_graphics(client_id);
         let disconnected_focus = self
             .clients
             .get(&client_id)
@@ -1052,7 +931,6 @@ impl HeadlessServer {
                 self.send_shell_focus_target(target, crate::ghostty::FocusEvent::Lost);
             }
         }
-        self.app.direct_graphics_available = self.direct_graphics_available();
         if was_foreground {
             self.promote_latest_remaining_client()
         } else {
@@ -1131,31 +1009,17 @@ impl HeadlessServer {
     }
 
     /// Drains server events from the dedicated channel.
-    ///
-    /// Uses the original full-render semantics when pane graphics are dormant.
     fn drain_server_events(&mut self) -> bool {
         let mut changed = false;
         while !self.should_quit.load(Ordering::Acquire) {
             let Ok(ev) = self.server_event_rx.try_recv() else {
                 break;
             };
-            changed |= self.handle_server_event(ev);
+            changed |= self.handle_server_event_with_render_impact(ev) == RenderImpact::Full;
         }
         changed
     }
 
-    /// Returns the strongest render impact from the drained event batch.
-    fn drain_server_events_with_render_impact(&mut self) -> RenderImpact {
-        let mut impact = RenderImpact::None;
-        while !self.should_quit.load(Ordering::Acquire) {
-            let Ok(ev) = self.server_event_rx.try_recv() else {
-                break;
-            };
-            impact.merge(self.handle_server_event_with_render_impact(ev));
-        }
-        impact
-    }
-
     async fn reject_late_client_connections(&mut self) {
         self.server_event_rx.close();
         while let Some(event) = self.server_event_rx.recv().await {
@@ -2112,12 +1976,17 @@ impl HeadlessServer {
                 transfer_id,
                 image_id,
                 success,
-            } => self.complete_direct_graphics(client_id, transfer_id, image_id, success),
+            } => self
+                .native_result(client_id, transfer_id, image_id, success)
+                .unwrap_or(false),
             ServerEvent::GraphicsTransmissionStarted {
                 client_id,
                 transfer_id,
                 image_id,
-            } => self.start_direct_graphics_response(client_id, transfer_id, image_id),
+            } => {
+                self.native_started(client_id, transfer_id, image_id);
+                false
+            }
             ServerEvent::ClientAttachTerminal {
                 client_id,
                 terminal_id,
@@ -2937,35 +2806,6 @@ impl HeadlessServer {
         }
     }
 
-    fn drain_api_requests_with_render_impact(&mut self) -> RenderImpact {
-        let mut impact = RenderImpact::None;
-        while !self.should_quit.load(Ordering::Acquire) {
-            let Ok(msg) = self.app.api_rx.try_recv() else {
-                break;
-            };
-            impact.merge(self.handle_api_request_with_render_impact(msg));
-        }
-        impact
-    }
-
-    fn handle_api_request_with_render_impact(
-        &mut self,
-        msg: api::ApiRequestMessage,
-    ) -> RenderImpact {
-        if matches!(
-            &msg.request.method,
-            api::schema::Method::PaneGraphicsStreamSet(_)
-                | api::schema::Method::PaneGraphicsStreamDirect(_)
-        ) {
-            return self.handle_pane_graphics_stream_frame(msg);
-        }
-        if self.handle_api_request_with_shutdown_check(msg) {
-            RenderImpact::Full
-        } else {
-            RenderImpact::None
-        }
-    }
-
     fn handle_api_request_with_shutdown_check_inner(
         &mut self,
         msg: api::ApiRequestMessage,
@@ -2999,12 +2839,6 @@ 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());
             let handoff_succeeded = handoff_result.is_ok();
@@ -3054,16 +2888,7 @@ impl HeadlessServer {
             _ => {}
         }
 
-        let pane_graphics_revision_before = matches!(
-            &msg.request.method,
-            api::schema::Method::PaneGraphicsSet(_)
-                | api::schema::Method::PaneGraphicsClear(_)
-                | api::schema::Method::PaneGraphicsStreamOpen(_)
-                | api::schema::Method::PaneGraphicsStreamClose(_)
-        )
-        .then_some(self.app.pane_graphics.revision());
-        let mut changed = metadata_expired
-            | (pane_graphics_revision_before.is_none() && api::request_changes_ui(&msg.request));
+        let mut changed = metadata_expired | api::request_changes_ui(&msg.request);
         let skip_default_workspace = skip_default_workspace_for_request
             || matches!(
                 &msg.request.method,
@@ -3184,10 +3009,6 @@ 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 {
@@ -3210,10 +3031,6 @@ impl HeadlessServer {
         }
         let _ = msg.respond_to.send(response);
 
-        if let Some(revision_before) = pane_graphics_revision_before {
-            changed |= revision_before != self.app.pane_graphics.revision();
-        }
-
         // Forward new toast state only when a client-local delivery mode is selected.
         // Herdr delivery renders the toast in-frame and must not ask clients to
         // show a terminal or system notification.
diff --git a/src/server/headless/endpoint_requests.rs b/src/server/headless/endpoint_requests.rs
index afdcc80a..6d2dd6ba 100644
--- a/src/server/headless/endpoint_requests.rs
+++ b/src/server/headless/endpoint_requests.rs
@@ -127,7 +127,6 @@ impl HeadlessServer {
                     request: *request,
                     respond_to,
                     response_write_complete: None,
-                    stream_active: None,
                 },
             )
     }
diff --git a/src/server/headless/native_graphics.rs b/src/server/headless/native_graphics.rs
new file mode 100644
index 00000000..67417d38
--- /dev/null
+++ b/src/server/headless/native_graphics.rs
@@ -0,0 +1,599 @@
+//! Decoded native exports. Deliberately independent of API slots/leases.
+use super::HeadlessServer;
+use crate::kitty_graphics::surface::{DeliveryCache, SourceFiles, NATIVE_SLOT_BIT};
+use crate::pane_graphics_files::{FileStore, OwnedExport};
+use crate::protocol::{
+    ServerMessage, SurfaceGraphicsAsset, SurfaceGraphicsAssetKey, SurfaceGraphicsFormat,
+    SurfaceGraphicsScene, SurfaceGraphicsSource,
+};
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT as NATIVE_BIT;
+const MAX_FILE: usize = 16 * 1024 * 1024;
+const MAX_BYTES: usize = 64 * 1024 * 1024;
+const DELIVERY_TIMEOUT: Duration = Duration::from_secs(5);
+const RESPONSE_TIMEOUT: Duration = Duration::from_secs(3);
+
+pub(super) struct Pending {
+    export: Arc<OwnedExport>,
+    transfer_id: u64,
+    image_id: u32,
+    source: SurfaceGraphicsSource,
+    asset: SurfaceGraphicsAssetKey,
+    slot: bool,
+    deadline: Instant,
+    written: bool,
+    refresh_needed: bool,
+    scene: SurfaceGraphicsScene,
+    delivery: DeliveryCache,
+}
+
+struct AcknowledgedSlot {
+    asset: SurfaceGraphicsAssetKey,
+    slot: bool,
+}
+
+impl Pending {
+    pub(super) fn defer_inline_delivery(&mut self, delivery: &DeliveryCache) {
+        self.delivery = delivery.clone();
+        self.refresh_needed = true;
+    }
+}
+
+pub(super) struct NativeGraphics {
+    store: FileStore,
+    pending: HashMap<u64, Pending>,
+    disabled: HashSet<u64>,
+    next_transfer: u64,
+    source_retries: HashMap<u64, Instant>,
+    // The key and bank currently addressed for each logical source. Native state
+    // advances on ACK; inline state and source pruning advance on queued scenes.
+    acknowledged_slots: HashMap<u64, HashMap<SurfaceGraphicsSource, AcknowledgedSlot>>,
+}
+impl Default for NativeGraphics {
+    fn default() -> Self {
+        Self {
+            store: FileStore::default(),
+            pending: HashMap::new(),
+            disabled: HashSet::new(),
+            next_transfer: 1,
+            source_retries: HashMap::new(),
+            acknowledged_slots: HashMap::new(),
+        }
+    }
+}
+impl NativeGraphics {
+    #[cfg(all(test, unix))]
+    pub(super) fn busy(&self) -> bool {
+        !self.pending.is_empty()
+    }
+    pub(super) fn is_pending(&self, client: u64) -> bool {
+        self.pending.contains_key(&client)
+    }
+    pub(super) fn can_hold(&self, client: u64, scene: &SurfaceGraphicsScene) -> bool {
+        let Some(pending) = self.pending.get(&client) else {
+            return true;
+        };
+        let same_image = |a: &crate::protocol::SurfaceGraphicsAssetKey,
+                          b: &crate::protocol::SurfaceGraphicsAssetKey| {
+            a.source == b.source
+                && a.image_width == b.image_width
+                && a.image_height == b.image_height
+                && a.format == b.format
+                && a.data_len == b.data_len
+        };
+        scene.placements.len() == pending.scene.placements.len()
+            && scene
+                .placements
+                .iter()
+                .zip(&pending.scene.placements)
+                .all(|(current, held)| {
+                    if !same_image(&current.asset, &held.asset) {
+                        return false;
+                    }
+                    let mut geometry = current.clone();
+                    geometry.asset = held.asset.clone();
+                    geometry == *held
+                })
+            && scene.retained_assets.len() == pending.scene.retained_assets.len()
+            && scene
+                .retained_assets
+                .iter()
+                .zip(&pending.scene.retained_assets)
+                .all(|(current, held)| same_image(current, held))
+    }
+
+    pub(super) fn hold(&self, client: u64) -> Option<(SurfaceGraphicsScene, DeliveryCache)> {
+        self.pending
+            .get(&client)
+            .map(|p| (p.scene.clone(), p.delivery.clone()))
+    }
+    /// Records bank state only after the scene metadata has been queued.
+    ///
+    /// `inline_assets` contains only payloads actually queued on the wire; a
+    /// selected native upload has already been removed and changes bank only
+    /// after its ACK. A new inline key addresses bank 0, while replaying the
+    /// exact native key keeps the client's existing key-to-bank mapping.
+    /// Pruning is committed here as well: a scene rejected by the writer must
+    /// not mutate bank history for pixels the client is still displaying.
+    pub(super) fn commit_scene(
+        &mut self,
+        client: u64,
+        scene: &SurfaceGraphicsScene,
+        inline_assets: &[SurfaceGraphicsAssetKey],
+    ) {
+        let present_sources: HashSet<_> = scene
+            .placements
+            .iter()
+            .map(|placement| &placement.asset.source)
+            .chain(scene.retained_assets.iter().map(|asset| &asset.source))
+            .cloned()
+            .collect();
+        let slots = self.acknowledged_slots.entry(client).or_default();
+        slots.retain(|source, _| present_sources.contains(source));
+        for asset in inline_assets
+            .iter()
+            .filter(|asset| matches!(asset.source, SurfaceGraphicsSource::Terminal { .. }))
+        {
+            let same_resident_key = slots
+                .get(&asset.source)
+                .is_some_and(|resident| resident.asset == *asset);
+            if !same_resident_key {
+                slots.insert(
+                    asset.source.clone(),
+                    AcknowledgedSlot {
+                        asset: asset.clone(),
+                        slot: false,
+                    },
+                );
+            }
+        }
+        if slots.is_empty() {
+            self.acknowledged_slots.remove(&client);
+        }
+    }
+    #[cfg(all(test, unix))]
+    pub(super) fn prepare(
+        &mut self,
+        client: u64,
+        scope: &str,
+        scene: &mut SurfaceGraphicsScene,
+        delivery: &DeliveryCache,
+    ) -> Option<(Pending, ServerMessage)> {
+        self.prepare_with_sources(client, scope, scene, delivery, &mut SourceFiles::default())
+    }
+
+    pub(super) fn prepare_with_sources(
+        &mut self,
+        client: u64,
+        scope: &str,
+        scene: &mut SurfaceGraphicsScene,
+        delivery: &DeliveryCache,
+        sources: &mut SourceFiles,
+    ) -> Option<(Pending, ServerMessage)> {
+        if self.disabled.contains(&client)
+            || self.pending.contains_key(&client)
+            || self.pending.len() >= 8
+        {
+            return None;
+        }
+        let index = scene
+            .assets
+            .iter()
+            .position(|asset| eligible(asset, sources))?;
+        let asset = &scene.assets[index];
+        if self.pending.values().map(|p| p.export.len()).sum::<usize>()
+            + asset.key.data_len as usize
+            > MAX_BYTES
+            || self.next_transfer >= NATIVE_BIT
+        {
+            return None;
+        }
+        let export = if let Some(source) = sources.get(&asset.key) {
+            Arc::clone(source)
+        } else {
+            Arc::new(self.store.export(&asset.data).ok()?)
+        };
+        let transfer_id = NATIVE_BIT | self.next_transfer;
+        self.next_transfer += 1;
+        let asset_key = asset.key.clone();
+        let source = asset_key.source.clone();
+        // Start in bank 1: an older client may still be displaying an inline image
+        // under the unchanged logical base ID. Thereafter always stage into the
+        // bank opposite the one this client currently addresses.
+        let slot = !self
+            .acknowledged_slots
+            .get(&client)
+            .and_then(|slots| slots.get(&source))
+            .map(|resident| resident.slot)
+            .unwrap_or(false);
+        let base = crate::kitty_graphics::surface::native_host_image_id(scope, &asset.key);
+        let image_id = base ^ if slot { NATIVE_SLOT_BIT } else { 0 };
+        let message = ServerMessage::GraphicsFile {
+            path: export.path().to_string_lossy().into_owned(),
+            expected_len: export.len() as u64,
+            image_id,
+            transfer_id,
+            leading: Vec::new(),
+            control: format!(
+                "a=t,f=32,s={},v={},i={image_id},q=0",
+                asset.key.image_width, asset.key.image_height
+            ),
+            surface_asset: Some(asset_key.clone()),
+        };
+        sources.remove(&asset_key);
+        scene.assets.remove(index);
+        let mut held = scene.clone();
+        held.assets.clear();
+        Some((
+            Pending {
+                export,
+                transfer_id,
+                image_id,
+                source,
+                asset: asset_key,
+                slot,
+                deadline: Instant::now() + DELIVERY_TIMEOUT,
+                written: false,
+                refresh_needed: delivery.has_pending(),
+                scene: held,
+                delivery: delivery.clone(),
+            },
+            message,
+        ))
+    }
+    pub(super) fn commit(&mut self, client: u64, pending: Pending) {
+        self.pending.insert(client, pending);
+    }
+    fn matches(&self, client: u64, transfer: u64, image: u32) -> bool {
+        self.pending
+            .get(&client)
+            .is_some_and(|p| p.transfer_id == transfer && p.image_id == image)
+    }
+}
+fn materialize_sources(
+    scene: &mut SurfaceGraphicsScene,
+    delivery: &mut DeliveryCache,
+    sources: &mut SourceFiles,
+) -> bool {
+    let mut failed = false;
+    scene.assets.retain_mut(|asset| {
+        let Some(source) = sources.remove(&asset.key) else {
+            return true;
+        };
+        match source.copy_rgba() {
+            Ok(data) => {
+                asset.data = data;
+                true
+            }
+            Err(error) => {
+                tracing::debug!(%error, "native source fallback could not be read");
+                failed = true;
+                delivery.forget_asset(&asset.key);
+                false
+            }
+        }
+    });
+    sources.clear();
+    failed
+}
+
+fn eligible(asset: &SurfaceGraphicsAsset, sources: &SourceFiles) -> bool {
+    matches!(asset.key.source, SurfaceGraphicsSource::Terminal { .. })
+        && asset.key.format == SurfaceGraphicsFormat::Rgba
+        && asset.key.data_len > 0
+        && asset.key.data_len <= MAX_FILE as u64
+        && sources.get(&asset.key).map_or_else(
+            || asset.key.data_len == asset.data.len() as u64,
+            |source| asset.key.data_len == source.len() as u64,
+        )
+        && u64::from(asset.key.image_width)
+            .checked_mul(u64::from(asset.key.image_height))
+            .and_then(|pixels| pixels.checked_mul(4))
+            == Some(asset.key.data_len)
+}
+impl HeadlessServer {
+    pub(super) fn defer_changed_native_geometry(
+        &mut self,
+        client: u64,
+        scene: &SurfaceGraphicsScene,
+    ) -> bool {
+        if self.native_graphics.can_hold(client, scene) {
+            return false;
+        }
+        self.retire_native_graphics_for_client(client);
+        self.app.render_dirty.request_generic();
+        if let Some(client) = self.clients.get_mut(&client) {
+            client.defer_full_render();
+        }
+        true
+    }
+
+    fn materialize_native_sources(
+        &mut self,
+        client: u64,
+        scene: &mut SurfaceGraphicsScene,
+        delivery: &mut DeliveryCache,
+        sources: &mut SourceFiles,
+    ) {
+        if materialize_sources(scene, delivery, sources) {
+            self.native_graphics
+                .source_retries
+                .entry(client)
+                .or_insert_with(|| Instant::now() + std::time::Duration::from_secs(1));
+        } else {
+            self.native_graphics.source_retries.remove(&client);
+        }
+    }
+
+    pub(super) fn prepare_native_scene(
+        &mut self,
+        client: u64,
+        scene: &mut SurfaceGraphicsScene,
+        delivery: &mut DeliveryCache,
+        sources: &mut SourceFiles,
+    ) -> Option<(Pending, ServerMessage)> {
+        if let Some(pending) = self.native_graphics.pending.get_mut(&client) {
+            pending.refresh_needed |= !scene.assets.is_empty()
+                || scene.placements != pending.scene.placements
+                || scene.retained_assets != pending.scene.retained_assets;
+        }
+        if let Some((held, held_delivery)) = self.native_graphics.hold(client) {
+            *scene = held;
+            *delivery = held_delivery;
+            sources.clear();
+            return None;
+        }
+        if !self.client_supports_direct_graphics(client) {
+            self.materialize_native_sources(client, scene, delivery, sources);
+            return None;
+        }
+        let mut prepared = self.native_graphics.prepare_with_sources(
+            client,
+            &self.client_shell_boot_id,
+            scene,
+            delivery,
+            sources,
+        );
+        self.materialize_native_sources(client, scene, delivery, sources);
+        if let Some((pending, _)) = &mut prepared {
+            pending.delivery = delivery.clone();
+            pending.refresh_needed |= delivery.has_pending();
+        }
+        prepared
+    }
+
+    pub(super) fn native_started(&mut self, client: u64, transfer: u64, image: u32) -> bool {
+        if transfer & NATIVE_BIT == 0 {
+            return false;
+        }
+        if !self.native_graphics.disabled.contains(&client)
+            && self.native_graphics.matches(client, transfer, image)
+        {
+            let p = self
+                .native_graphics
+                .pending
+                .get_mut(&client)
+                .expect("matched");
+            p.written = true;
+            p.deadline = Instant::now() + RESPONSE_TIMEOUT;
+        }
+        true
+    }
+    pub(super) fn native_result(
+        &mut self,
+        client: u64,
+        transfer: u64,
+        image: u32,
+        success: bool,
+    ) -> Option<bool> {
+        if transfer & NATIVE_BIT == 0 {
+            return None;
+        }
+        if !self.native_graphics.matches(client, transfer, image) {
+            return Some(false);
+        }
+        if self.native_graphics.disabled.contains(&client) {
+            return Some(self.expire_native_graphics(Instant::now()));
+        }
+        if success {
+            if !self.native_graphics.pending[&client].written {
+                return Some(false);
+            }
+            let pending = self
+                .native_graphics
+                .pending
+                .remove(&client)
+                .expect("matched");
+            self.native_graphics
+                .acknowledged_slots
+                .entry(client)
+                .or_default()
+                .insert(
+                    pending.source,
+                    AcknowledgedSlot {
+                        asset: pending.asset,
+                        slot: pending.slot,
+                    },
+                );
+            let refresh = pending.refresh_needed;
+            if refresh {
+                if let Some(c) = self.clients.get_mut(&client) {
+                    c.defer_full_render();
+                }
+            }
+            return Some(refresh);
+        }
+        self.native_graphics.disabled.insert(client);
+        self.native_graphics
+            .pending
+            .get_mut(&client)
+            .expect("matched")
+            .deadline = Instant::now();
+        Some(self.expire_native_graphics(Instant::now()))
+    }
+    pub(super) fn expire_native_graphics(&mut self, now: Instant) -> bool {
+        let expired: Vec<_> = self
+            .native_graphics
+            .pending
+            .iter()
+            .filter(|(_, p)| p.deadline <= now)
+            .map(|(id, p)| (*id, p.transfer_id, p.image_id))
+            .collect();
+        let mut changed = false;
+        self.native_graphics.source_retries.retain(|id, deadline| {
+            if *deadline > now {
+                return true;
+            }
+            if let Some(client) = self.clients.get_mut(id) {
+                client.defer_full_render();
+                changed = true;
+            }
+            false
+        });
+        for (id, transfer_id, image_id) in expired {
+            self.native_graphics.disabled.insert(id);
+            let message = ServerMessage::GraphicsTransmissionRetired {
+                transfer_id,
+                image_id,
+            };
+            // Control messages precede queued renders. The client permanently rejects
+            // retired native transfers, including a file still in the render queue.
+            let sent = self.send_to_client(id, message);
+            if !sent {
+                continue;
+            } // Keep the export alive until retirement is queued or disconnect cleans it.
+            self.native_graphics.pending.remove(&id);
+            self.native_graphics.disabled.insert(id);
+            if let Some(c) = self.clients.get_mut(&id) {
+                c.shell_graphics_delivery = DeliveryCache::default();
+                c.defer_full_render();
+            }
+            changed = true;
+        }
+        changed
+    }
+    pub(super) fn retire_native_graphics_for_client(&mut self, id: u64) {
+        if let Some(p) = self.native_graphics.pending.get_mut(&id) {
+            p.deadline = Instant::now();
+            self.native_graphics.disabled.insert(id);
+            if self.expire_native_graphics(Instant::now()) {
+                self.app.render_dirty.request_generic();
+            }
+        }
+    }
+    pub(super) fn disconnect_native_graphics(&mut self, id: u64) {
+        self.native_graphics.pending.remove(&id);
+        self.native_graphics.disabled.remove(&id);
+        self.native_graphics.source_retries.remove(&id);
+        self.native_graphics.acknowledged_slots.remove(&id);
+    }
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use crate::protocol::{SurfaceGraphicsAssetKey, SurfaceGraphicsTarget};
+
+    fn asset() -> SurfaceGraphicsAsset {
+        SurfaceGraphicsAsset {
+            key: SurfaceGraphicsAssetKey {
+                source: SurfaceGraphicsSource::Terminal {
+                    target: SurfaceGraphicsTarget::Pane {
+                        pane_id: "native-test".into(),
+                    },
+                    image_id: 1,
+                },
+                image_width: 1,
+                image_height: 1,
+                format: SurfaceGraphicsFormat::Rgba,
+                data_len: 4,
+                data_fingerprint: 123,
+            },
+            data: vec![1, 2, 3, 255],
+        }
+    }
+
+    #[test]
+    fn eligibility_is_decoded_rgba_terminal_only() {
+        let mut a = asset();
+        assert!(eligible(&a, &SourceFiles::default()));
+        a.key.format = SurfaceGraphicsFormat::Png;
+        assert!(!eligible(&a, &SourceFiles::default()));
+        a.key.format = SurfaceGraphicsFormat::Rgba;
+        a.key.data_len = 5;
+        assert!(!eligible(&a, &SourceFiles::default()));
+        a.key.data_len = 4;
+        a.key.source = SurfaceGraphicsSource::PaneLayer {
+            pane_id: "p".into(),
+            layer_id: "l".into(),
+        };
+        assert!(!eligible(&a, &SourceFiles::default()));
+        let mut a = asset();
+        a.data.resize(MAX_FILE + 1, 0);
+        a.key.data_len = a.data.len() as u64;
+        assert!(!eligible(&a, &SourceFiles::default()));
+    }
+
+    #[test]
+    fn pending_guard_and_client_identity_are_independent() {
+        let mut state = NativeGraphics::default();
+        let mut scene = SurfaceGraphicsScene {
+            assets: vec![asset()],
+            ..Default::default()
+        };
+        let (pending, _) = state
+            .prepare(7, "scope", &mut scene, &DeliveryCache::default())
+            .unwrap();
+        let path = pending.export.path().to_owned();
+        let transfer = pending.transfer_id;
+        let image = pending.image_id;
+        assert_ne!(transfer & NATIVE_BIT, 0);
+        assert!(scene.assets.is_empty());
+        assert!(!state.busy()); // prepare is not a delivery commit
+        state.commit(7, pending);
+        assert!(state.matches(7, transfer, image));
+        assert!(!state.matches(8, transfer, image));
+        assert!(!state.matches(7, transfer + 1, image));
+        assert!(!state.matches(7, transfer, image + 1));
+        assert!(path.exists());
+        assert!(state.hold(7).unwrap().0.assets.is_empty());
+        state.pending.remove(&7);
+        assert!(!path.exists());
+        state.disabled.insert(7);
+        scene.assets.push(asset());
+        assert!(state
+            .prepare(7, "scope", &mut scene, &DeliveryCache::default())
+            .is_none());
+        assert_eq!(scene.assets.len(), 1); // native fallback keeps authoritative inline bytes
+    }
+
+    #[test]
+    fn one_pending_per_client_and_global_count_limit() {
+        let mut state = NativeGraphics::default();
+        for id in 0..8 {
+            let mut scene = SurfaceGraphicsScene {
+                assets: vec![asset()],
+                ..Default::default()
+            };
+            let (pending, _) = state
+                .prepare(id, "scope", &mut scene, &DeliveryCache::default())
+                .unwrap();
+            state.commit(id, pending);
+            scene.assets.push(asset());
+            assert!(state
+                .prepare(id, "scope", &mut scene, &DeliveryCache::default())
+                .is_none());
+        }
+        let mut scene = SurfaceGraphicsScene {
+            assets: vec![asset()],
+            ..Default::default()
+        };
+        assert!(state
+            .prepare(9, "scope", &mut scene, &DeliveryCache::default())
+            .is_none());
+        assert_eq!(scene.assets.len(), 1);
+    }
+}
diff --git a/src/server/headless/pane_graphics.rs b/src/server/headless/pane_graphics.rs
deleted file mode 100644
index 947d2cb8..00000000
--- a/src/server/headless/pane_graphics.rs
+++ /dev/null
@@ -1,513 +0,0 @@
-use super::{HeadlessServer, RenderImpact};
-use crate::api;
-use crate::protocol::{ServerMessage, MAX_GRAPHICS_FRAME_SIZE};
-use crate::server::clients::ClientConnectionMode;
-
-impl HeadlessServer {
-    pub(super) fn pane_graphics_runtime_active(&self) -> bool {
-        !self.app.pane_graphics.slots.is_empty()
-    }
-
-    pub(super) fn handle_pane_graphics_stream_frame(
-        &mut self,
-        msg: api::ApiRequestMessage,
-    ) -> RenderImpact {
-        if self.shutting_down {
-            let _ = self.handle_api_request_with_shutdown_check_inner(msg, false, false);
-            return RenderImpact::None;
-        }
-
-        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 = direct_key
-            .as_ref()
-            .and_then(|key| self.direct_graphics_client_for_key(key));
-        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();
-
-        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| {
-                                (
-                                    lease.path().to_string_lossy().into_owned(),
-                                    lease.len() as u64,
-                                    lease.fingerprint(),
-                                )
-                            })
-                    })
-                });
-                if let (Some(key), Some((path, expected_len, transfer_id))) =
-                    (direct_key.clone(), direct_frame)
-                {
-                    let prepared = self.clients.get(&client_id).and_then(|client| {
-                        matches!(client.mode, ClientConnectionMode::ClientShell).then_some(())?;
-                        let layer = self
-                            .app
-                            .pane_graphics
-                            .slots
-                            .get(&key)
-                            .and_then(|slot| slot.layer.as_ref())?;
-                        let asset = crate::kitty_graphics::surface::pane_layer_asset_key(
-                            &self.app, &key, layer,
-                        )?;
-                        let (image_id, control) =
-                            crate::kitty_graphics::surface::direct_upload_control(
-                                &self.client_shell_boot_id,
-                                &asset,
-                            );
-                        Some((
-                            crate::kitty_graphics::DirectFileCommand {
-                                leading: Vec::new(),
-                                control,
-                            },
-                            Some(asset),
-                            image_id,
-                        ))
-                    });
-                    let Some((command, surface_asset, image_id)) = prepared 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,
-                        surface_asset,
-                    };
-                    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,
-                                    image_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,
-                                });
-                            }
-                            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 {
-            RenderImpact::Graphics
-        } else {
-            RenderImpact::None
-        }
-    }
-
-    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 len > crate::api::schema::PANE_GRAPHICS_STREAM_MAX_BYTES
-            || !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();
-        }
-        self.app.direct_graphics_available = self.direct_graphics_available();
-    }
-
-    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.stream_is_active()
-                .then_some(slot.direct_gate.as_mut())
-                .flatten()
-                .filter(|gate| {
-                    gate.client_id == client_id
-                        && gate.transfer_id == transfer_id
-                        && gate.image_id == image_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
-                        && gate.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_with_client_notice(&key);
-            return false;
-        }
-        if success {
-            let gate = {
-                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")
-            };
-            if gate.respond_to.send(gate.success_response).is_err() {
-                self.retire_direct_gate(&key);
-                return true;
-            }
-            self.app.direct_graphics_available = self.direct_graphics_available();
-            return true;
-        }
-
-        if let Some(client) = self.clients.get_mut(&client_id) {
-            client.direct_graphics = false;
-        }
-        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: gate.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()
-    }
-
-    fn retire_direct_gate_with_client_notice(&mut self, key: &crate::app::pane_graphics::Key) {
-        if let Some((client_id, transfer_id, image_id)) = self
-            .app
-            .pane_graphics
-            .slots
-            .get(key)
-            .and_then(|slot| slot.direct_gate.as_ref())
-            .map(|gate| (gate.client_id, gate.transfer_id, gate.image_id))
-        {
-            self.send_to_client(
-                client_id,
-                ServerMessage::GraphicsTransmissionRetired {
-                    transfer_id,
-                    image_id,
-                },
-            );
-        }
-        self.retire_direct_gate(key);
-    }
-
-    pub(super) fn retain_live_pane_graphics(&mut self) -> bool {
-        let dead_direct = self
-            .app
-            .pane_graphics
-            .slots
-            .iter()
-            .filter(|((pane_id, _), slot)| {
-                slot.direct_gate.is_some()
-                    && !self
-                        .app
-                        .state
-                        .workspaces
-                        .iter()
-                        .any(|workspace| workspace.pane_state(*pane_id).is_some())
-            })
-            .map(|(key, _)| key.clone())
-            .collect::<Vec<_>>();
-        let changed = !dead_direct.is_empty();
-        for key in dead_direct {
-            self.retire_direct_gate_with_client_notice(&key);
-        }
-        let retained = self.app.pane_graphics.retain_live_panes(&self.app.state);
-        changed || retained
-    }
-
-    pub(super) fn retire_all_direct_graphics(&mut self) {
-        let retired = self
-            .app
-            .pane_graphics
-            .slots
-            .iter()
-            .filter(|(_, slot)| {
-                slot.layer
-                    .as_ref()
-                    .is_some_and(crate::app::pane_graphics::Layer::terminal_only)
-            })
-            .map(|(key, slot)| {
-                let notification = slot
-                    .direct_gate
-                    .as_ref()
-                    .map(|gate| (gate.client_id, gate.transfer_id, gate.image_id));
-                (key.clone(), notification)
-            })
-            .collect::<Vec<_>>();
-        for (_, notification) in &retired {
-            if let Some((client_id, transfer_id, image_id)) = notification {
-                self.send_to_client(
-                    *client_id,
-                    ServerMessage::GraphicsTransmissionRetired {
-                        transfer_id: *transfer_id,
-                        image_id: *image_id,
-                    },
-                );
-            }
-        }
-        for (key, _) in retired {
-            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);
-        }
-    }
-}
diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs
index a3bbedfa..5bc17c10 100644
--- a/src/server/headless/render.rs
+++ b/src/server/headless/render.rs
@@ -373,6 +373,15 @@ impl HeadlessServer {
     }
 
     pub(super) fn render_and_stream(&mut self) {
+        self.render_and_stream_with_graphics_limit(MAX_GRAPHICS_FRAME_SIZE);
+    }
+
+    #[cfg(all(test, unix))]
+    pub(super) fn render_and_stream_with_test_graphics_limit(&mut self, max: usize) {
+        self.render_and_stream_with_graphics_limit(max);
+    }
+
+    fn render_and_stream_with_graphics_limit(&mut self, graphics_frame_limit: usize) {
         let full_started = crate::render_prof::timer();
         let render_targets = render_targets(&self.clients, self.foreground_client_id);
 
@@ -665,8 +674,16 @@ impl HeadlessServer {
                         popup,
                         graphics,
                         graphics_delivery: next_graphics_delivery,
+                        graphics_sources,
                     } = shell_render.expect("active shell surface");
-                    surface_parts = Some((panes, splits, popup, graphics, next_graphics_delivery));
+                    surface_parts = Some((
+                        panes,
+                        splits,
+                        popup,
+                        graphics,
+                        next_graphics_delivery,
+                        graphics_sources,
+                    ));
                     frame
                 }
                 ClientConnectionMode::TerminalPending => continue,
@@ -725,6 +742,20 @@ impl HeadlessServer {
                 }
             };
 
+            if surface_parts
+                .as_ref()
+                .is_some_and(|(_, _, _, graphics, _, _)| {
+                    self.defer_changed_native_geometry(client_id, graphics)
+                })
+            {
+                continue;
+            }
+            let mut native_upload =
+                surface_parts
+                    .as_mut()
+                    .and_then(|(_, _, _, graphics, delivery, sources)| {
+                        self.prepare_native_scene(client_id, graphics, delivery, sources)
+                    });
             let Some(client) = self.clients.get_mut(&client_id) else {
                 continue;
             };
@@ -734,62 +765,102 @@ impl HeadlessServer {
             };
             let has_graphics = surface_parts
                 .as_ref()
-                .is_some_and(|(_, _, _, graphics, _)| {
+                .is_some_and(|(_, _, _, graphics, _, _)| {
                     !graphics.assets.is_empty()
                         || !graphics.placements.is_empty()
                         || !graphics.retained_assets.is_empty()
                 });
             let mut next_shell_graphics_delivery = None;
-            let prepared = if let Some((panes, splits, popup, graphics, delivery)) = surface_parts {
-                next_shell_graphics_delivery = Some(delivery);
-                client
-                    .render_state
-                    .prepare_pane_surface(protocol::PaneSurfaceFrame {
-                        boot_id: self.client_shell_boot_id.clone(),
-                        projection_revision: shell_projection_revision,
-                        surface_revision: 0,
-                        frame,
-                        panes,
-                        splits,
-                        popup,
-                        graphics,
-                    })
-            } else {
-                client.render_state.prepare_frame(frame)
-            };
+            let prepared =
+                if let Some((panes, splits, popup, graphics, delivery, _)) = surface_parts {
+                    next_shell_graphics_delivery = Some(delivery);
+                    client.render_state.prepare_pane_surface_with_file(
+                        protocol::PaneSurfaceFrame {
+                            boot_id: self.client_shell_boot_id.clone(),
+                            projection_revision: shell_projection_revision,
+                            surface_revision: 0,
+                            frame,
+                            panes,
+                            splits,
+                            popup,
+                            graphics,
+                        },
+                        native_upload.is_some(),
+                    )
+                } else {
+                    client.render_state.prepare_frame(frame)
+                };
             let Some(mut prepared) = prepared else {
                 client.clear_deferred_render();
                 crate::render_prof::event("full_render.skip_identical");
                 continue;
             };
             let max = if has_graphics {
-                MAX_GRAPHICS_FRAME_SIZE
+                graphics_frame_limit
             } else {
                 crate::protocol::MAX_FRAME_SIZE
             };
             let mut shell_assets_deferred = false;
-            let serialized = match Self::frame_server_message_with_max(prepared.message(), max) {
+            let mut suppress_impossible_asset_retry = false;
+            let mut stripped_assets = Vec::new();
+            let mut 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,
-                        claimed, max, "dropping graphics assets from oversized pane surface"
+                        claimed, max, "trimming inline graphics assets from oversized pane surface"
                     );
-                    if !prepared.strip_pane_surface_assets() {
-                        crate::render_prof::event("full_render.serialize_oversized");
+                    let framed = loop {
+                        let Some(key) = prepared.pop_pane_surface_asset() else {
+                            break None;
+                        };
+                        stripped_assets.push(key);
+                        match Self::frame_server_message_with_max(prepared.message(), max) {
+                            Ok(framed) => break Some(framed),
+                            Err(protocol::FramingError::Oversized { .. }) => {}
+                            Err(err) => {
+                                warn!(client_id, err = %err, "failed to serialize trimmed pane surface");
+                                broken_clients.push(client_id);
+                                break None;
+                            }
+                        }
+                    };
+                    let Some(framed) = framed else {
+                        if stripped_assets.is_empty() && prepared.has_queued_surface_assets() {
+                            // Delta/reuse owns an encoded payload that cannot be trimmed in
+                            // place. Drop its baseline so the bounded full-surface path runs next.
+                            client.render_state.request_repaint();
+                            client.defer_full_render();
+                        } else {
+                            crate::render_prof::event("full_render.serialize_oversized");
+                        }
                         continue;
-                    }
-                    next_shell_graphics_delivery = None;
-                    shell_assets_deferred = true;
-                    match Self::frame_server_message(prepared.message()) {
-                        Ok(framed) => framed,
-                        Err(err) => {
-                            warn!(client_id, err = %err, "failed to serialize pane surface without assets");
-                            broken_clients.push(client_id);
-                            crate::render_prof::event("full_render.serialize_error");
-                            continue;
+                    };
+                    let made_progress =
+                        native_upload.is_some() || prepared.has_queued_surface_assets();
+                    if let Some(delivery) = next_shell_graphics_delivery.as_mut() {
+                        for key in &stripped_assets {
+                            delivery.forget_asset(key);
                         }
+                        shell_assets_deferred = made_progress && !stripped_assets.is_empty();
+                    }
+                    if let (Some((pending, _)), Some(delivery)) = (
+                        native_upload.as_mut(),
+                        next_shell_graphics_delivery.as_ref(),
+                    ) {
+                        pending.defer_inline_delivery(delivery);
                     }
+                    if !made_progress {
+                        // Keep the asset absent from the delivery cache, but do not spin on
+                        // an identical frame when no payload can fit beside its metadata.
+                        suppress_impossible_asset_retry = true;
+                        warn!(
+                            client_id,
+                            "inline graphics asset cannot fit in a graphics frame; waiting for a later scene change"
+                        );
+                    }
+                    framed
                 }
                 Err(protocol::FramingError::Oversized { claimed, max }) => {
                     warn!(
@@ -806,11 +877,33 @@ impl HeadlessServer {
                     continue;
                 }
             };
-            let shell_graphics_pending = next_shell_graphics_delivery
-                .as_ref()
-                .is_some_and(crate::kitty_graphics::surface::DeliveryCache::has_pending);
-            match writer.render.try_send(serialized) {
+            let shell_graphics_pending = !suppress_impossible_asset_retry
+                && next_shell_graphics_delivery
+                    .as_ref()
+                    .is_some_and(crate::kitty_graphics::surface::DeliveryCache::has_pending);
+            if let Some((_, message)) = &native_upload {
+                let Ok(file_frame) =
+                    Self::frame_server_message_with_max(message, MAX_GRAPHICS_FRAME_SIZE)
+                else {
+                    client.defer_full_render();
+                    continue;
+                };
+                serialized.extend_from_slice(&file_frame);
+            }
+            let send = if native_upload.is_some() || self.native_graphics.is_pending(client_id) {
+                writer.render.send_ordered(serialized)
+            } else {
+                writer.render.try_send(serialized)
+            };
+            match send {
                 Ok(()) => {
+                    if let Some((graphics, inline_assets)) = prepared.queued_surface_graphics() {
+                        self.native_graphics
+                            .commit_scene(client_id, graphics, inline_assets);
+                    }
+                    if let Some((pending, _)) = native_upload {
+                        self.native_graphics.commit(client_id, pending);
+                    }
                     if let Some(delivery) = next_shell_graphics_delivery {
                         client.shell_graphics_delivery = delivery;
                     }
diff --git a/src/server/headless/retained_surface.rs b/src/server/headless/retained_surface.rs
index 07d33ef5..bf98bc3e 100644
--- a/src/server/headless/retained_surface.rs
+++ b/src/server/headless/retained_surface.rs
@@ -203,6 +203,7 @@ struct RetainedRecipientUpdate {
     graphics: Option<(
         protocol::PaneSurfaceFrame,
         crate::kitty_graphics::surface::DeliveryCache,
+        crate::kitty_graphics::surface::SourceFiles,
     )>,
 }
 
@@ -446,7 +447,7 @@ impl HeadlessServer {
                 let client = &self.clients[&client_id];
                 let mut next_surface = surface.clone();
                 crate::server::render_stream::apply_pane_surface_patch(&mut next_surface, &patch);
-                let Some((graphics, delivery)) =
+                let Some((graphics, delivery, sources)) =
                     crate::server::client_shell_graphics::collect_retained(
                         &self.app,
                         &next_surface,
@@ -460,7 +461,7 @@ impl HeadlessServer {
                 };
                 graphics_changed = graphics != surface.graphics;
                 next_surface.graphics = graphics;
-                Some((next_surface, delivery))
+                Some((next_surface, delivery, sources))
             } else {
                 None
             };
@@ -490,8 +491,17 @@ impl HeadlessServer {
             let RetainedRecipientUpdate {
                 client_id,
                 patch,
-                graphics,
+                mut graphics,
             } = update;
+            if graphics.as_ref().is_some_and(|(surface, _, _)| {
+                self.defer_changed_native_geometry(client_id, &surface.graphics)
+            }) {
+                deferred += 1;
+                continue;
+            }
+            let native_upload = graphics.as_mut().and_then(|(surface, delivery, sources)| {
+                self.prepare_native_scene(client_id, &mut surface.graphics, delivery, sources)
+            });
             let Some(client) = self.clients.get_mut(&client_id) else {
                 continue;
             };
@@ -502,9 +512,11 @@ impl HeadlessServer {
             };
             // The published row patch cannot carry images. Reuse the retained text/layout
             // in a graphics-capable surface message rather than invoking the full renderer.
-            let (prepared, graphics_delivery) = if let Some((surface, delivery)) = graphics {
+            let (prepared, graphics_delivery) = if let Some((surface, delivery, _)) = graphics {
                 (
-                    client.render_state.prepare_pane_surface(surface),
+                    client
+                        .render_state
+                        .prepare_pane_surface_with_file(surface, native_upload.is_some()),
                     Some(delivery),
                 )
             } else {
@@ -520,7 +532,7 @@ impl HeadlessServer {
             } else {
                 protocol::MAX_FRAME_SIZE
             };
-            let serialized =
+            let mut serialized =
                 match Self::frame_server_message_with_max(prepared.message(), max_frame_size) {
                     Ok(serialized) => serialized,
                     Err(error) => {
@@ -529,14 +541,39 @@ impl HeadlessServer {
                             %error,
                             "failed to serialize retained pane surface patch"
                         );
+                        // A delta may own an encoded graphics payload that cannot be
+                        // trimmed in place. Force the bounded full-surface recovery path.
+                        client.render_state.request_repaint();
                         client.defer_full_render();
                         deferred += 1;
                         continue;
                     }
                 };
+            if let Some((_, message)) = &native_upload {
+                let Ok(file_frame) =
+                    Self::frame_server_message_with_max(message, MAX_GRAPHICS_FRAME_SIZE)
+                else {
+                    client.defer_full_render();
+                    deferred += 1;
+                    continue;
+                };
+                serialized.extend_from_slice(&file_frame);
+            }
             crate::render_prof::counter("retained_surface.bytes", serialized.len() as u64);
-            match writer.render.try_send(serialized) {
+            let send = if native_upload.is_some() || self.native_graphics.is_pending(client_id) {
+                writer.render.send_ordered(serialized)
+            } else {
+                writer.render.try_send(serialized)
+            };
+            match send {
                 Ok(()) => {
+                    if let Some((graphics, inline_assets)) = prepared.queued_surface_graphics() {
+                        self.native_graphics
+                            .commit_scene(client_id, graphics, inline_assets);
+                    }
+                    if let Some((pending, _)) = native_upload {
+                        self.native_graphics.commit(client_id, pending);
+                    }
                     let graphics_pending = graphics_delivery
                         .as_ref()
                         .is_some_and(crate::kitty_graphics::surface::DeliveryCache::has_pending);
diff --git a/src/server/headless/surface_interest.rs b/src/server/headless/surface_interest.rs
index 9772fc03..44a68b04 100644
--- a/src/server/headless/surface_interest.rs
+++ b/src/server/headless/surface_interest.rs
@@ -57,7 +57,7 @@ impl HeadlessServer {
 
         if let Some(held_inputs) = held_inputs {
             self.release_client_shell_inputs(client_id, held_inputs);
-            self.retire_direct_graphics_for_client(client_id);
+            self.retire_native_graphics_for_client(client_id);
         }
         if changed {
             self.finish_shell_location_reconciliation(focus_before, &focused_tabs_before);
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index 84031ce4..34258dd3 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -1,9 +1,10 @@
 use super::*;
 
-#[path = "pane_graphics.rs"]
-mod pane_graphics_tests;
+mod native_graphics;
 #[path = "pane_move.rs"]
 mod pane_move_tests;
+#[path = "pane_graphics.rs"]
+mod retained_graphics_tests;
 #[path = "surface_delta.rs"]
 mod surface_delta_tests;
 #[path = "surface_interest.rs"]
@@ -102,6 +103,7 @@ fn test_headless_server_with_event_hub(event_hub: api::EventHub) -> HeadlessServ
         client_socket_path: socket_path,
         client_socket_identity,
         clients: HashMap::new(),
+        native_graphics: Default::default(),
         #[cfg(unix)]
         next_client_id: 1,
         foreground_client_id: None,
@@ -245,7 +247,6 @@ fn headless_pane_list(server: &mut HeadlessServer) -> Vec<api::schema::PaneInfo>
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
     let response: api::schema::SuccessResponse =
         serde_json::from_str(&response_rx.recv().unwrap()).unwrap();
@@ -303,7 +304,6 @@ fn headless_api_request_drains_all_pending_internal_events_before_reading_state(
             },
             respond_to,
             response_write_complete: None,
-            stream_active: None,
         })
     );
     let response = response_rx
@@ -2100,7 +2100,6 @@ async fn client_local_navigation_does_not_emit_global_focus_transitions() {
             },
             respond_to,
             response_write_complete: None,
-            stream_active: None,
         },
     );
     server.app.sync_focus_events();
@@ -2191,7 +2190,6 @@ async fn client_local_navigation_emits_pane_focused_only_when_that_client_moves(
                 },
                 respond_to,
                 response_write_complete: None,
-                stream_active: None,
             },
         );
         let response = response_rx.recv().expect("navigation response");
@@ -2323,7 +2321,6 @@ async fn repeated_layout_action_reapplies_controller_geometry() {
             },
             respond_to,
             response_write_complete: None,
-            stream_active: None,
         },
     ));
 
@@ -2368,7 +2365,6 @@ async fn public_close_reapplies_controller_geometry() {
             },
             respond_to,
             response_write_complete: None,
-            stream_active: None,
         })
     );
 
@@ -2572,7 +2568,6 @@ async fn public_background_tab_create_preserves_client_locations() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     assert_eq!(
@@ -2620,7 +2615,6 @@ async fn public_workspace_focus_preserves_each_clients_remembered_tabs() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     let first_location = server.clients[&41].shell_location.as_ref().unwrap();
@@ -2701,7 +2695,6 @@ async fn public_agent_focus_replaces_a_diverged_client_shell_projection() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
     let response: crate::api::schema::SuccessResponse =
         serde_json::from_str(&response_rx.recv().expect("agent focus response")).unwrap();
@@ -2776,7 +2769,6 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
     assert_eq!(server.app.state.active, Some(1));
     server.render_and_stream();
@@ -3393,7 +3385,6 @@ async fn worktree_discovery_does_not_block_client_typing() {
             },
             respond_to,
             response_write_complete: None,
-            stream_active: None,
         });
         entered
             .recv_timeout(Duration::from_secs(5))
@@ -6997,7 +6988,6 @@ fn notification_show_api_forwards_one_semantic_client_notification() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     assert!(changed);
@@ -7060,7 +7050,6 @@ fn notification_show_api_preserves_colon_in_forwarded_title() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     assert!(changed);
@@ -7106,7 +7095,6 @@ fn notification_show_api_validates_empty_title_before_disabled_delivery() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     assert!(changed);
@@ -7137,7 +7125,6 @@ fn notification_show_api_reports_no_foreground_client() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     assert!(changed);
@@ -7188,7 +7175,6 @@ fn notification_show_api_includes_sound_in_semantic_event() {
             },
             respond_to,
             response_write_complete: None,
-            stream_active: None,
         })
     );
 
@@ -7250,7 +7236,6 @@ fn completion_guard_api_report(server: &mut HeadlessServer, method: api::schema:
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
     let response = response_rx
         .recv_timeout(Duration::from_millis(100))
@@ -7585,7 +7570,6 @@ fn stale_api_agent_report_does_not_forward_done_sound() {
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
 
     assert!(changed);
diff --git a/src/server/headless/tests/native_graphics.rs b/src/server/headless/tests/native_graphics.rs
new file mode 100644
index 00000000..bde7cf08
--- /dev/null
+++ b/src/server/headless/tests/native_graphics.rs
@@ -0,0 +1,1411 @@
+//! Native export integration tests for file delivery, retirement, and fallback.
+#![cfg(unix)]
+
+use super::*;
+use crate::kitty_graphics::surface::{DeliveryCache, NATIVE_SLOT_BIT, NATIVE_TRANSFER_BIT};
+use crate::protocol::{
+    SurfaceGraphicsAsset, SurfaceGraphicsAssetKey, SurfaceGraphicsFormat, SurfaceGraphicsPlacement,
+    SurfaceGraphicsScene, SurfaceGraphicsSource, SurfaceGraphicsTarget,
+};
+use std::sync::mpsc::Receiver;
+use std::time::{Duration, Instant};
+
+fn native_started(server: &mut HeadlessServer, client: u64, transfer: u64, image: u32) -> bool {
+    server.native_started(client, transfer, image)
+}
+
+fn native_result(
+    server: &mut HeadlessServer,
+    client: u64,
+    transfer: u64,
+    image: u32,
+    success: bool,
+) -> bool {
+    server
+        .native_result(client, transfer, image, success)
+        .unwrap_or(false)
+}
+
+fn commit_scene(server: &mut HeadlessServer, client: u64, scene: &SurfaceGraphicsScene) {
+    let inline_assets = scene
+        .assets
+        .iter()
+        .map(|asset| asset.key.clone())
+        .collect::<Vec<_>>();
+    server
+        .native_graphics
+        .commit_scene(client, scene, &inline_assets);
+}
+
+fn scene() -> SurfaceGraphicsScene {
+    scene_for(1, 123)
+}
+
+fn scene_for(image_id: u32, data_fingerprint: u64) -> SurfaceGraphicsScene {
+    let key = SurfaceGraphicsAssetKey {
+        source: SurfaceGraphicsSource::Terminal {
+            target: SurfaceGraphicsTarget::Pane {
+                pane_id: "native-test".into(),
+            },
+            image_id,
+        },
+        image_width: 1,
+        image_height: 1,
+        format: SurfaceGraphicsFormat::Rgba,
+        data_len: 4,
+        data_fingerprint,
+    };
+    SurfaceGraphicsScene {
+        assets: vec![SurfaceGraphicsAsset {
+            key: key.clone(),
+            data: vec![1, 2, 3, 255],
+        }],
+        placements: vec![SurfaceGraphicsPlacement {
+            asset: key,
+            logical_placement_id: 1,
+            x: 2,
+            y: 3,
+            cols: 1,
+            rows: 1,
+            source_x: 0,
+            source_y: 0,
+            source_width: 1,
+            source_height: 1,
+            x_offset: 0,
+            y_offset: 0,
+            z: 0,
+            scrollback_offset: 0,
+        }],
+        retained_assets: Vec::new(),
+    }
+}
+
+fn add_client(server: &mut HeadlessServer, id: u64) -> (Receiver<Vec<u8>>, Receiver<Vec<u8>>) {
+    let (writer, control, render) = test_client_writer();
+    let mut client = ClientConnection::new(
+        (80, 24),
+        crate::kitty_graphics::HostCellSize {
+            width_px: 10,
+            height_px: 20,
+        },
+        id,
+        RenderEncoding::SemanticFrame,
+        Some(writer),
+    );
+    client.mode = ClientConnectionMode::ClientShell;
+    client.direct_graphics = true;
+    client.pixel_mouse = true;
+    server.clients.insert(id, client);
+    server.foreground_client_id = Some(id);
+    (control, render)
+}
+
+fn prepare_and_commit(server: &mut HeadlessServer, id: u64) -> (PathBuf, u64, u32) {
+    let mut desired = scene();
+    let (pending, message) = server
+        .native_graphics
+        .prepare(
+            id,
+            &server.client_shell_boot_id,
+            &mut desired,
+            &DeliveryCache::default(),
+        )
+        .expect("eligible native export");
+    let ServerMessage::GraphicsFile {
+        path,
+        transfer_id,
+        image_id,
+        expected_len,
+        leading,
+        control,
+        surface_asset,
+    } = message
+    else {
+        panic!("expected file upload");
+    };
+    assert_ne!(transfer_id & NATIVE_TRANSFER_BIT, 0);
+    assert_eq!(expected_len, 4);
+    assert!(leading.is_empty());
+    assert_eq!(control, format!("a=t,f=32,s=1,v=1,i={image_id},q=0"));
+    assert_eq!(surface_asset.as_ref(), Some(&desired.placements[0].asset));
+    assert!(desired.assets.is_empty());
+    assert_eq!(std::fs::read(&path).unwrap(), [1, 2, 3, 255]);
+    commit_scene(server, id, &desired);
+    server.native_graphics.commit(id, pending);
+    (PathBuf::from(path), transfer_id, image_id)
+}
+
+fn prepare_scene_and_commit(
+    server: &mut HeadlessServer,
+    id: u64,
+    mut desired: SurfaceGraphicsScene,
+) -> (u64, u32) {
+    let (pending, message) = server
+        .native_graphics
+        .prepare(
+            id,
+            &server.client_shell_boot_id,
+            &mut desired,
+            &DeliveryCache::default(),
+        )
+        .expect("eligible native export");
+    let ServerMessage::GraphicsFile {
+        transfer_id,
+        image_id,
+        ..
+    } = message
+    else {
+        panic!("expected file upload");
+    };
+    commit_scene(server, id, &desired);
+    server.native_graphics.commit(id, pending);
+    (transfer_id, image_id)
+}
+
+fn ack_native(server: &mut HeadlessServer, id: u64, transfer: u64, image: u32) {
+    assert!(native_started(server, id, transfer, image));
+    native_result(server, id, transfer, image, true);
+    assert!(!server.native_graphics.is_pending(id));
+}
+
+fn assert_retirement(control: &Receiver<Vec<u8>>, transfer: u64, image: u32) {
+    assert!(matches!(
+        read_server_message(control.recv_timeout(Duration::from_millis(100)).expect("control retirement")),
+        ServerMessage::GraphicsTransmissionRetired { transfer_id, image_id }
+            if transfer_id == transfer && image_id == image
+    ));
+}
+
+fn assert_native_disabled(server: &mut HeadlessServer, id: u64) {
+    let mut desired = scene();
+    assert!(server
+        .native_graphics
+        .prepare(
+            id,
+            &server.client_shell_boot_id,
+            &mut desired,
+            &DeliveryCache::default(),
+        )
+        .is_none());
+    assert_eq!(desired.assets[0].data, [1, 2, 3, 255]);
+    assert!(server.clients[&id].direct_graphics);
+}
+
+#[test]
+fn native_tokens_require_started_and_old_api_tokens_are_ignored() {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let (path, token, image) = prepare_and_commit(&mut server, 7);
+
+    assert!(!native_result(&mut server, 7, token, image, true));
+    assert!(server.native_graphics.is_pending(7));
+    assert!(path.exists());
+    assert!(!native_started(
+        &mut server,
+        7,
+        token & !NATIVE_TRANSFER_BIT,
+        image
+    ));
+    assert!(!native_result(
+        &mut server,
+        7,
+        token & !NATIVE_TRANSFER_BIT,
+        image,
+        true,
+    ));
+    assert!(native_started(&mut server, 7, token, image));
+    assert!(!native_result(&mut server, 7, token, image, true));
+    assert!(!server.native_graphics.is_pending(7));
+    assert!(!path.exists());
+}
+
+#[test]
+fn native_image_banks_advance_only_on_started_ack_and_are_scoped_and_pruned() {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let (_other_control, _other_render) = add_client(&mut server, 8);
+    let base = crate::kitty_graphics::surface::native_host_image_id(
+        &server.client_shell_boot_id,
+        &scene().placements[0].asset,
+    );
+
+    // Unknown sources start in bank 1, avoiding an inline image at the base ID.
+    let (first_transfer, first_image) = prepare_scene_and_commit(&mut server, 7, scene());
+    assert_eq!(first_image, base ^ NATIVE_SLOT_BIT);
+    // A success before Started is not an ACK and neither completes nor advances.
+    assert!(!native_result(
+        &mut server,
+        7,
+        first_transfer,
+        first_image,
+        true
+    ));
+    assert!(server.native_graphics.is_pending(7));
+    ack_native(&mut server, 7, first_transfer, first_image);
+
+    let (second_transfer, second_image) =
+        prepare_scene_and_commit(&mut server, 7, scene_for(1, 124));
+    assert_eq!(second_image, base);
+    // A late ACK for the preceding transfer cannot advance the pending revision.
+    assert!(!native_result(
+        &mut server,
+        7,
+        first_transfer,
+        first_image,
+        true
+    ));
+    assert!(server.native_graphics.is_pending(7));
+    ack_native(&mut server, 7, second_transfer, second_image);
+    let (third_transfer, third_image) = prepare_scene_and_commit(&mut server, 7, scene_for(1, 125));
+    assert_eq!(third_image, base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 7, third_transfer, third_image);
+
+    // Selection is independent per client.
+    let (other_transfer, other_image) = prepare_scene_and_commit(&mut server, 8, scene_for(1, 126));
+    assert_eq!(other_image, base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 8, other_transfer, other_image);
+
+    // A different logical source also starts in bank 1. Retention keeps source
+    // 1's bank history even though it has no placement in this scene.
+    let mut source_two = scene_for(2, 127);
+    source_two
+        .retained_assets
+        .push(scene_for(1, 127).placements[0].asset.clone());
+    let source_two_base = crate::kitty_graphics::surface::native_host_image_id(
+        &server.client_shell_boot_id,
+        &source_two.placements[0].asset,
+    );
+    let (source_two_transfer, source_two_image) =
+        prepare_scene_and_commit(&mut server, 7, source_two);
+    assert_eq!(source_two_image, source_two_base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 7, source_two_transfer, source_two_image);
+    let (retained_transfer, retained_image) =
+        prepare_scene_and_commit(&mut server, 7, scene_for(1, 128));
+    assert_eq!(retained_image, base);
+    ack_native(&mut server, 7, retained_transfer, retained_image);
+
+    // Move source 1 back to bank 1, then omit it entirely. Returning after an
+    // unrelated upload starts at bank 1 rather than retaining stale history.
+    let (before_prune_transfer, before_prune_image) =
+        prepare_scene_and_commit(&mut server, 7, scene_for(1, 129));
+    assert_eq!(before_prune_image, base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 7, before_prune_transfer, before_prune_image);
+    let (pruning_transfer, pruning_image) =
+        prepare_scene_and_commit(&mut server, 7, scene_for(2, 130));
+    assert_eq!(pruning_image, source_two_base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 7, pruning_transfer, pruning_image);
+    let (restored_transfer, restored_image) =
+        prepare_scene_and_commit(&mut server, 7, scene_for(1, 131));
+    assert_eq!(restored_image, base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 7, restored_transfer, restored_image);
+
+    // Reusing a disconnected client ID must not inherit its source-bank history.
+    server.remove_client(7);
+    let (_control, _render) = add_client(&mut server, 7);
+    let (_reconnected_transfer, reconnected_image) =
+        prepare_scene_and_commit(&mut server, 7, scene_for(1, 132));
+    assert_eq!(reconnected_image, base ^ NATIVE_SLOT_BIT);
+}
+
+#[test]
+fn inline_commit_updates_bank_only_for_a_new_asset_revision() {
+    // Replaying inline bytes for the exact ACKed key still addresses the native
+    // bank retained by the client, so it must not be recorded as bank 0.
+    let mut exact = test_headless_server();
+    let (_control, _render) = add_client(&mut exact, 7);
+    let base = crate::kitty_graphics::surface::native_host_image_id(
+        &exact.client_shell_boot_id,
+        &scene().placements[0].asset,
+    );
+    let (first_transfer, first_image) = prepare_scene_and_commit(&mut exact, 7, scene());
+    assert_eq!(first_image, base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut exact, 7, first_transfer, first_image);
+    commit_scene(&mut exact, 7, &scene());
+    let (_next_transfer, next_image) = prepare_scene_and_commit(&mut exact, 7, scene_for(1, 124));
+    assert_eq!(next_image, base);
+
+    // A new inline revision has no native key mapping on the client and is
+    // uploaded at the logical base ID. The next native revision must therefore
+    // stage into bank 1 rather than collide with that visible base image.
+    let mut revised = test_headless_server();
+    let (_control, _render) = add_client(&mut revised, 7);
+    let (first_transfer, first_image) = prepare_scene_and_commit(&mut revised, 7, scene());
+    ack_native(&mut revised, 7, first_transfer, first_image);
+    commit_scene(&mut revised, 7, &scene_for(1, 124));
+    let (_next_transfer, next_image) = prepare_scene_and_commit(&mut revised, 7, scene_for(1, 125));
+    assert_eq!(next_image, base ^ NATIVE_SLOT_BIT);
+}
+
+#[test]
+fn rejected_omitted_scene_does_not_prune_visible_bank_history() {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let base = crate::kitty_graphics::surface::native_host_image_id(
+        &server.client_shell_boot_id,
+        &scene().placements[0].asset,
+    );
+    let (first_transfer, first_image) = prepare_scene_and_commit(&mut server, 7, scene());
+    assert_eq!(first_image, base ^ NATIVE_SLOT_BIT);
+    ack_native(&mut server, 7, first_transfer, first_image);
+
+    // Preparing an omitted-source scene is speculative. Model the full render
+    // path rejecting it because its render lane is already occupied.
+    let writer = ClientWriter::test_paused();
+    server.clients.get_mut(&7).unwrap().writer = Some(writer.clone());
+    writer.render.try_send(vec![1]).unwrap();
+    let mut omitted = SurfaceGraphicsScene::default();
+    assert!(server
+        .prepare_native_scene(
+            7,
+            &mut omitted,
+            &mut DeliveryCache::default(),
+            &mut Default::default(),
+        )
+        .is_none());
+    assert!(matches!(
+        writer.render.try_send(vec![2]),
+        Err(std::sync::mpsc::TrySendError::Full(_))
+    ));
+
+    // The rejected omission did not change what is visible. Stage opposite the
+    // still-visible bank 1 when the source returns.
+    let (_next_transfer, next_image) = prepare_scene_and_commit(&mut server, 7, scene_for(1, 124));
+    assert_eq!(next_image, base);
+}
+
+#[test]
+fn unqueued_native_prepare_does_not_advance_image_bank() {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let base = crate::kitty_graphics::surface::native_host_image_id(
+        &server.client_shell_boot_id,
+        &scene_for(3, 200).placements[0].asset,
+    );
+    let mut dropped_scene = scene_for(3, 200);
+    let (dropped, dropped_message) = server
+        .native_graphics
+        .prepare(
+            7,
+            &server.client_shell_boot_id,
+            &mut dropped_scene,
+            &DeliveryCache::default(),
+        )
+        .unwrap();
+    let ServerMessage::GraphicsFile {
+        transfer_id: dropped_transfer,
+        image_id: dropped_image,
+        ..
+    } = dropped_message
+    else {
+        panic!("expected file upload");
+    };
+    assert_eq!(dropped_image, base ^ NATIVE_SLOT_BIT);
+    drop(dropped); // Model serialization/send failure before commit.
+
+    let (transfer, image) = prepare_scene_and_commit(&mut server, 7, scene_for(3, 201));
+    assert_eq!(image, dropped_image);
+    assert!(!native_result(
+        &mut server,
+        7,
+        dropped_transfer,
+        dropped_image,
+        true
+    ));
+    assert!(server.native_graphics.is_pending(7));
+    ack_native(&mut server, 7, transfer, image);
+    let (_next_transfer, next_image) = prepare_scene_and_commit(&mut server, 7, scene_for(3, 202));
+    assert_eq!(next_image, base);
+}
+
+#[test]
+fn native_timeout_uses_control_lane_and_falls_back_inline() {
+    let mut server = test_headless_server();
+    let (control, render) = add_client(&mut server, 7);
+    let (path, token, image) = prepare_and_commit(&mut server, 7);
+    // Occupy the render lane: retirement must not wait for file/render backpressure.
+    server.clients[&7]
+        .writer
+        .as_ref()
+        .unwrap()
+        .render
+        .send_ordered(vec![99])
+        .unwrap();
+    assert!(server.expire_native_graphics(Instant::now() + Duration::from_secs(60)));
+    assert_retirement(&control, token, image);
+    assert_eq!(
+        render.recv_timeout(Duration::from_millis(100)).unwrap(),
+        vec![99]
+    );
+    assert!(render.try_recv().is_err());
+    assert!(!path.exists());
+    assert!(!server.native_graphics.is_pending(7));
+    assert!(!server.clients[&7].shell_graphics_delivery.has_pending());
+    assert_eq!(server.clients[&7].deferred_render(), DeferredRender::Full);
+    assert_native_disabled(&mut server, 7);
+    assert!(!native_result(&mut server, 7, token, image, true));
+}
+
+#[test]
+fn native_rejection_falls_back_without_disabling_other_client() {
+    let mut server = test_headless_server();
+    let (control, render) = add_client(&mut server, 7);
+    let (_other_control, _other_render) = add_client(&mut server, 8);
+    let (path, token, image) = prepare_and_commit(&mut server, 7);
+    let (other_path, other_token, other_image) = prepare_and_commit(&mut server, 8);
+    assert!(native_result(&mut server, 7, token, image, false));
+    assert_retirement(&control, token, image);
+    assert!(render.try_recv().is_err());
+    assert!(!path.exists());
+    assert_native_disabled(&mut server, 7);
+    assert!(other_path.exists());
+    assert!(server.native_graphics.is_pending(8));
+    native_started(&mut server, 8, other_token, other_image);
+    native_result(&mut server, 8, other_token, other_image, true);
+    assert!(!other_path.exists());
+    assert!(!server.native_graphics.is_pending(8));
+    assert!(server.clients[&8].direct_graphics);
+}
+
+#[test]
+fn disconnect_unlinks_only_that_clients_native_export() {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let (_other_control, _other_render) = add_client(&mut server, 8);
+    let (path, token, image) = prepare_and_commit(&mut server, 7);
+    let (other_path, other_token, other_image) = prepare_and_commit(&mut server, 8);
+    server.remove_client(7);
+    assert!(!path.exists());
+    assert!(other_path.exists());
+    assert!(!server.native_graphics.is_pending(7));
+    assert!(server.native_graphics.is_pending(8));
+    assert!(!native_result(&mut server, 7, token, image, true));
+    native_started(&mut server, 8, other_token, other_image);
+    native_result(&mut server, 8, other_token, other_image, true);
+    assert!(!other_path.exists());
+    assert!(!server.native_graphics.is_pending(8));
+    assert!(server.clients[&8].direct_graphics);
+}
+
+#[test]
+fn pending_hold_accepts_pixel_revision_but_not_geometry_or_source_change() {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let (_path, _token, _image) = prepare_and_commit(&mut server, 7);
+    let original = scene();
+    assert!(server.native_graphics.can_hold(7, &original));
+    let mut next = original.clone();
+    next.placements[0].asset.data_fingerprint += 1;
+    next.assets[0].key.data_fingerprint += 1;
+    next.assets[0].data[0] += 1;
+    assert!(server.native_graphics.can_hold(7, &next));
+    assert_eq!(
+        server.native_graphics.hold(7).unwrap().0.placements,
+        original.placements
+    );
+    next.placements[0].cols += 1;
+    assert!(!server.native_graphics.can_hold(7, &next));
+    next = original.clone();
+    next.placements[0].x += 1;
+    assert!(!server.native_graphics.can_hold(7, &next));
+    next = original.clone();
+    next.placements[0].asset.image_width += 1;
+    assert!(!server.native_graphics.can_hold(7, &next));
+    next = original.clone();
+    next.placements[0].asset.source = SurfaceGraphicsSource::Terminal {
+        target: SurfaceGraphicsTarget::Pane {
+            pane_id: "another-pane".into(),
+        },
+        image_id: 1,
+    };
+    assert!(!server.native_graphics.can_hold(7, &next));
+    assert!(!server
+        .native_graphics
+        .can_hold(7, &SurfaceGraphicsScene::default()));
+    next = original;
+    next.retained_assets.push(next.placements[0].asset.clone());
+    assert!(!server.native_graphics.can_hold(7, &next));
+}
+
+fn drain_native_render_messages(writer: &ClientWriter) -> Vec<ServerMessage> {
+    let mut messages = Vec::new();
+    for batch in writer.test_drain() {
+        let len = batch.len() as u64;
+        let mut cursor = std::io::Cursor::new(batch);
+        while cursor.position() < len {
+            messages.push(
+                protocol::read_message(&mut cursor, protocol::MAX_GRAPHICS_FRAME_SIZE)
+                    .expect("decode ordered scene/file transaction"),
+            );
+        }
+    }
+    messages
+}
+
+fn write_chunked_test_image(
+    server: &mut HeadlessServer,
+    pane: crate::layout::PaneId,
+    image_id: u32,
+    side: u32,
+    pixels: &[u8],
+) {
+    use base64::Engine as _;
+
+    let encoded = base64::engine::general_purpose::STANDARD.encode(pixels);
+    let chunks = encoded.as_bytes().chunks(4096);
+    let count = chunks.len();
+    for (index, chunk) in chunks.enumerate() {
+        let more = u8::from(index + 1 < count);
+        let mut command = if index == 0 {
+            format!(
+                "\x1b_Ga=T,f=32,t=d,i={image_id},p={image_id},s={side},v={side},c=1,r=1,q=2,m={more};"
+            )
+            .into_bytes()
+        } else {
+            format!("\x1b_Gm={more};").into_bytes()
+        };
+        command.extend_from_slice(chunk);
+        command.extend_from_slice(b"\x1b\\");
+        write_shared_test_pane(server, pane, &command);
+    }
+}
+
+#[tokio::test]
+async fn delta_encoded_inline_commit_updates_next_native_bank() {
+    let (mut server, _control, _render, pane) = retained_test_server_with_control(
+        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
+    );
+    let writer = ClientWriter::test_paused();
+    let client = server.clients.get_mut(&1).unwrap();
+    client.writer = Some(writer.clone());
+    client.mode = ClientConnectionMode::ClientShell;
+    client.render_state =
+        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+    client.render_state.enable_surface_delta(true);
+    client.cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    client.direct_graphics = true;
+    client.pixel_mouse = true;
+    server.app.state.kitty_graphics_enabled = true;
+
+    server.render_and_stream();
+    let first = drain_native_render_messages(&writer);
+    let (first_transfer, first_image) = first
+        .iter()
+        .find_map(|message| match message {
+            ServerMessage::GraphicsFile {
+                transfer_id,
+                image_id,
+                ..
+            } => Some((*transfer_id, *image_id)),
+            _ => None,
+        })
+        .expect("first native upload");
+    assert!(native_started(&mut server, 1, first_transfer, first_image));
+    native_result(&mut server, 1, first_transfer, first_image, true);
+
+    // Model a client without direct-file capability for this replacement.
+    // Delta encoding
+    // wraps its surface in EndpointControl, but the queued asset still commits
+    // bank 0 bookkeeping.
+    server.clients.get_mut(&1).unwrap().direct_graphics = false;
+    write_shared_test_pane(
+        &mut server,
+        pane,
+        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;AP8A/w==\x1b\\",
+    );
+    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane])));
+    let inline = drain_native_render_messages(&writer);
+    assert!(inline.iter().any(|message| matches!(
+        message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    )));
+    assert!(!inline
+        .iter()
+        .any(|message| matches!(message, ServerMessage::GraphicsFile { .. })));
+
+    server.clients.get_mut(&1).unwrap().direct_graphics = true;
+    write_shared_test_pane(
+        &mut server,
+        pane,
+        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;AAD//w==\x1b\\",
+    );
+    server.render_and_stream();
+    let next = drain_native_render_messages(&writer);
+    assert!(next.iter().any(|message| matches!(
+        message,
+        ServerMessage::EndpointControl { kind, .. }
+            if kind == protocol::surface_delta::MESSAGE_KIND
+    )));
+    let next_image = next
+        .iter()
+        .find_map(|message| match message {
+            ServerMessage::GraphicsFile { image_id, .. } => Some(*image_id),
+            _ => None,
+        })
+        .expect("native upload after inline delta");
+    assert_eq!(next_image, first_image, "must stage opposite inline bank 0");
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn oversized_native_scene_queues_file_and_continues_stripped_assets_after_ack() {
+    let (mut server, _control, _render, pane) = retained_test_server_with_control(b"");
+    let writer = ClientWriter::test_paused();
+    let client = server.clients.get_mut(&1).unwrap();
+    client.writer = Some(writer.clone());
+    client.mode = ClientConnectionMode::ClientShell;
+    client.render_state =
+        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+    client.cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    client.direct_graphics = true;
+    client.pixel_mouse = true;
+    server.app.state.kitty_graphics_enabled = true;
+
+    const SIDE: u32 = 128;
+    let pixels = vec![0x7f; SIDE as usize * SIDE as usize * 4];
+    for image_id in [7, 8, 9] {
+        write_chunked_test_image(&mut server, pane, image_id, SIDE, &pixels);
+    }
+
+    // The limit is deliberately above one 64 KiB asset plus metadata but below
+    // the two inline assets left after selecting one native upload.
+    server.render_and_stream_with_test_graphics_limit(100_000);
+    let first = drain_native_render_messages(&writer);
+    let first_surface = first
+        .iter()
+        .find_map(|message| match message {
+            ServerMessage::PaneSurface(surface) => Some(surface),
+            _ => None,
+        })
+        .expect("trimmed metadata scene");
+    assert_eq!(first_surface.graphics.placements.len(), 3);
+    assert_eq!(first_surface.graphics.assets.len(), 1);
+    let first_inline = first_surface.graphics.assets[0].key.clone();
+    let (transfer, image, first_native) = first
+        .iter()
+        .find_map(|message| match message {
+            ServerMessage::GraphicsFile {
+                transfer_id,
+                image_id,
+                surface_asset: Some(asset),
+                ..
+            } => Some((*transfer_id, *image_id, asset.clone())),
+            _ => None,
+        })
+        .expect("selected native asset survives oversized recovery");
+    assert_ne!(first_inline.source, first_native.source);
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::Full);
+
+    assert!(native_started(&mut server, 1, transfer, image));
+    native_result(&mut server, 1, transfer, image, true);
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::Full);
+
+    server.render_and_stream_with_test_graphics_limit(100_000);
+    let continuation = drain_native_render_messages(&writer);
+    let continued_asset = continuation.iter().find_map(|message| match message {
+        ServerMessage::GraphicsFile {
+            surface_asset: Some(asset),
+            ..
+        } => Some(asset),
+        ServerMessage::PaneSurface(surface) => {
+            surface.graphics.assets.first().map(|asset| &asset.key)
+        }
+        _ => None,
+    });
+    let continued_asset = continued_asset.expect("stripped asset delivered on continuation");
+    assert_ne!(continued_asset.source, first_inline.source);
+    assert_ne!(continued_asset.source, first_native.source);
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn oversized_inline_trims_largest_and_sends_fitting_asset_without_retry_spin() {
+    let (mut server, _control, _render, pane) = retained_test_server_with_control(b"");
+    let writer = ClientWriter::test_paused();
+    let client = server.clients.get_mut(&1).unwrap();
+    client.writer = Some(writer.clone());
+    client.mode = ClientConnectionMode::ClientShell;
+    client.render_state =
+        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+    client.cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    client.direct_graphics = false;
+    client.pixel_mouse = true;
+    server.app.state.kitty_graphics_enabled = true;
+
+    for (image_id, side) in [(7, 128u32), (8, 16u32)] {
+        let pixels = vec![image_id as u8; side as usize * side as usize * 4];
+        write_chunked_test_image(&mut server, pane, image_id, side, &pixels);
+    }
+
+    // Probe the exact payload size of this surface with only the small asset.
+    // Resetting the server-side delivery/baseline makes the measured render
+    // observational only; the next call exercises the production recovery path.
+    server.render_and_stream();
+    let mut measured_surface = drain_native_render_messages(&writer)
+        .into_iter()
+        .find_map(|message| match message {
+            ServerMessage::PaneSurface(surface) => Some(surface),
+            _ => None,
+        })
+        .expect("measurement surface");
+    assert_eq!(measured_surface.graphics.assets.len(), 2);
+    measured_surface
+        .graphics
+        .assets
+        .retain(|asset| asset.key.image_width == 16);
+    let exact_small_limit = HeadlessServer::frame_server_message_with_max(
+        &ServerMessage::PaneSurface(measured_surface),
+        crate::protocol::MAX_GRAPHICS_FRAME_SIZE,
+    )
+    .unwrap()
+    .len()
+    .saturating_sub(4);
+    let client = server.clients.get_mut(&1).unwrap();
+    client.render_state =
+        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+    client.shell_graphics_delivery = DeliveryCache::default();
+    client.clear_deferred_render();
+
+    // Exactly metadata plus the 1 KiB image fits; adding the 64 KiB image does not.
+    server.render_and_stream_with_test_graphics_limit(exact_small_limit);
+    let first = drain_native_render_messages(&writer);
+    let first_surface = first
+        .iter()
+        .find_map(|message| match message {
+            ServerMessage::PaneSurface(surface) => Some(surface),
+            _ => None,
+        })
+        .expect("surface with fitting inline asset");
+    assert_eq!(first_surface.graphics.assets.len(), 1);
+    assert_eq!(first_surface.graphics.assets[0].key.image_width, 16);
+    assert!(!first
+        .iter()
+        .any(|message| matches!(message, ServerMessage::GraphicsFile { .. })));
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::Full);
+
+    // The remaining large asset cannot fit even alone. It remains absent from
+    // delivery state, but the identical impossible frame is not rescheduled forever.
+    server.render_and_stream_with_test_graphics_limit(exact_small_limit);
+    let second = drain_native_render_messages(&writer);
+    let second_surface = second
+        .iter()
+        .find_map(|message| match message {
+            ServerMessage::PaneSurface(surface) => Some(surface),
+            _ => None,
+        })
+        .expect("metadata-only bounded fallback");
+    assert!(second_surface.graphics.assets.is_empty());
+    assert!(server.clients[&1].shell_graphics_delivery.has_pending());
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+    shutdown_test_runtimes(&mut server);
+}
+
+#[tokio::test]
+async fn quiet_native_producer_geometry_retirement_schedules_full_inline_recovery() {
+    for retained in [false, true] {
+        let (mut server, _control, _render, pane) = retained_test_server_with_control(
+            b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
+        );
+        let writer = ClientWriter::test_paused();
+        let client = server.clients.get_mut(&1).unwrap();
+        client.writer = Some(writer.clone());
+        client.mode = ClientConnectionMode::ClientShell;
+        client.render_state =
+            crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+        client.cell_size = crate::kitty_graphics::HostCellSize {
+            width_px: 10,
+            height_px: 20,
+        };
+        client.direct_graphics = true;
+        client.pixel_mouse = true;
+        server.app.state.kitty_graphics_enabled = true;
+        server.app.render_dirty.take();
+        server.render_and_stream();
+
+        // The writer has already drained, so no future writer progress can wake
+        // the server after geometry cancellation. The native ACK is still pending.
+        let initial = drain_native_render_messages(&writer);
+        let metadata_index = initial
+            .iter()
+            .position(|message| matches!(message, ServerMessage::PaneSurface(_)))
+            .expect("native metadata scene");
+        let file_index = initial
+            .iter()
+            .position(|message| matches!(message, ServerMessage::GraphicsFile { .. }))
+            .expect("native file upload");
+        assert!(metadata_index < file_index);
+        let ServerMessage::PaneSurface(metadata) = &initial[metadata_index] else {
+            unreachable!()
+        };
+        assert!(metadata.graphics.assets.is_empty());
+        assert_eq!(metadata.graphics.placements.len(), 1);
+        assert_eq!(metadata.graphics.placements[0].cols, 1);
+        let ServerMessage::GraphicsFile {
+            transfer_id,
+            image_id,
+            path,
+            ..
+        } = &initial[file_index]
+        else {
+            unreachable!()
+        };
+        let (token, image, native_path) = (*transfer_id, *image_id, PathBuf::from(path));
+        assert!(native_path.exists());
+        assert!(server.native_graphics.is_pending(1));
+        assert!(writer.test_drain().is_empty());
+        native_started(&mut server, 1, token, image);
+
+        // A single producer update moves/resizes the placement, then goes quiet.
+        // Re-upload the same pixels to replace placement p=3 with two columns.
+        write_shared_test_pane(
+            &mut server,
+            pane,
+            b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=2,r=1,q=2;/wAA/w==\x1b\\",
+        );
+        server.app.render_dirty.take(); // model the main loop consuming this PTY wake
+        assert!(!server.app.render_dirty.is_pending());
+        if retained {
+            assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane])));
+        } else {
+            server.render_and_stream();
+        }
+        assert!(!server.native_graphics.is_pending(1), "retained={retained}");
+        assert!(!native_path.exists());
+        assert!(
+            server.app.render_dirty.is_pending(),
+            "geometry retirement must independently wake the quiet server; retained={retained}"
+        );
+        assert!(server.app.render_dirty.has_immediate_work());
+        assert_eq!(server.clients[&1].deferred_render(), DeferredRender::Full);
+        let retired = drain_native_render_messages(&writer);
+        assert!(retired.iter().any(|message| matches!(message,
+            ServerMessage::GraphicsTransmissionRetired { transfer_id, image_id }
+                if *transfer_id == token && *image_id == image
+        )));
+        assert!(!retired.iter().any(|message| matches!(
+            message,
+            ServerMessage::PaneSurface(_) | ServerMessage::GraphicsFile { .. }
+        )));
+
+        // A stale success must not restore native residency or consume the full
+        // recovery wake. No producer bytes or writer events occur after this ACK.
+        assert!(!native_result(&mut server, 1, token, image, true));
+        assert!(!server.native_graphics.is_pending(1));
+        let request = server.app.render_dirty.take();
+        assert!(
+            request.generic,
+            "generic full-render request must survive late ACK; retained={retained}"
+        );
+        server.render_and_stream();
+        let recovered = drain_native_render_messages(&writer);
+        assert!(!recovered
+            .iter()
+            .any(|message| matches!(message, ServerMessage::GraphicsFile { .. })));
+        let surface = recovered
+            .iter()
+            .find_map(|message| match message {
+                ServerMessage::PaneSurface(surface) => Some(surface),
+                _ => None,
+            })
+            .expect(
+                "next scheduled full render delivers inline fallback without new producer activity",
+            );
+        assert_eq!(surface.graphics.placements.len(), 1);
+        assert_eq!(surface.graphics.placements[0].cols, 2);
+        assert_eq!(surface.graphics.assets.len(), 1);
+        assert_eq!(surface.graphics.assets[0].data, [255, 0, 0, 255]);
+        assert!(server.clients[&1].direct_graphics);
+        assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+        shutdown_test_runtimes(&mut server);
+    }
+}
+
+#[tokio::test]
+#[ignore = "manual native-file retained-render scaling profile; one image, 1/15 populated panes"]
+async fn native_file_render_scale_profile() {
+    use base64::Engine as _;
+    use ratatui::layout::Direction;
+
+    const WARMUP: usize = 5;
+    const SAMPLES: usize = 35;
+    const IMAGE_WIDTH: u32 = 800;
+    const IMAGE_HEIGHT: u32 = 480;
+    for count in [1, 15] {
+        for (native, source_retention) in [(false, false), (true, false), (true, true)] {
+            let (mut server, _control, _render, root) =
+                retained_test_server_with_control(b"populated root terminal\r\n");
+            if source_retention {
+                // Test runtimes normally leave file transmission disabled. Enable
+                // it and the snapshot callback only for this source-producing pane.
+                server
+                    .app
+                    .state
+                    .runtime_for_pane_in_workspace(&server.app.terminal_runtimes, 0, root)
+                    .unwrap()
+                    .test_enable_kitty_source_forwarding();
+            }
+            let mut panes = vec![root];
+            for index in 1..count {
+                let workspace = &mut server.app.state.workspaces[0];
+                workspace.tabs[0].layout.focus_pane(panes[(index - 1) / 2]);
+                let pane = workspace.test_split(if index % 2 == 0 {
+                    Direction::Vertical
+                } else {
+                    Direction::Horizontal
+                });
+                workspace.insert_test_runtime(
+                    pane,
+                    crate::terminal::TerminalRuntime::test_with_screen_bytes(
+                        160,
+                        48,
+                        b"populated terminal\r\nsecond row\r\nthird row\r\n",
+                    ),
+                );
+                panes.push(pane);
+            }
+            let writer = ClientWriter::test_paused();
+            let client = server.clients.get_mut(&1).unwrap();
+            client.writer = Some(writer.clone());
+            client.mode = ClientConnectionMode::ClientShell;
+            client.terminal_size = (160, 48);
+            client.render_state =
+                crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+            client.cell_size = crate::kitty_graphics::HostCellSize {
+                width_px: 10,
+                height_px: 20,
+            };
+            // Compare actual capability fallback with automatic native delivery.
+            client.direct_graphics = native;
+            client.pixel_mouse = true;
+            server.app.state.kitty_graphics_enabled = true;
+            server.render_and_stream();
+            let baseline = drain_native_render_messages(&writer);
+            assert!(baseline
+                .iter()
+                .any(|message| matches!(message, ServerMessage::PaneSurface(_))));
+            assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+            let sources = panes.iter().copied().collect::<HashSet<_>>();
+
+            // Deterministic noisy RGB, opaque alpha. Only the first pixel changes
+            // each submission; no timing includes producer/base64 construction.
+            let mut rgba = vec![0u8; IMAGE_WIDTH as usize * IMAGE_HEIGHT as usize * 4];
+            let mut noise = 0x1234_5678u32;
+            for pixel in rgba.chunks_exact_mut(4) {
+                noise = noise.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
+                pixel[..3].copy_from_slice(&noise.to_le_bytes()[..3]);
+                pixel[3] = 255;
+            }
+            // Private /var/tmp native-store generation owns the producer path:
+            // bounded to one image and removed even if an assertion unwinds.
+            let producer_store = crate::pane_graphics_files::FileStore::native_sources();
+            let producer = source_retention
+                .then(|| producer_store.export(&rgba).expect("private producer file"));
+            let mut previous_source_path: Option<PathBuf> = None;
+            let mut times = Vec::with_capacity(SAMPLES);
+            for sample in 0..WARMUP + SAMPLES {
+                assert!(!server.native_graphics.is_pending(1));
+                assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+                assert!(writer.test_drain().is_empty());
+                rgba[..3].copy_from_slice(&(sample as u32 + 1).to_le_bytes()[..3]);
+                let mut payload = Vec::new();
+                payload.extend_from_slice(b"\x1b[H");
+                if let Some(producer) = &producer {
+                    std::fs::write(producer.path(), &rgba).unwrap();
+                    let encoded_path = base64::engine::general_purpose::STANDARD
+                        .encode(producer.path().as_os_str().as_encoded_bytes());
+                    payload.extend_from_slice(format!(
+                        "\x1b_Ga=T,f=32,t=f,i=7,p=3,s={IMAGE_WIDTH},v={IMAGE_HEIGHT},c=1,r=1,q=2;{encoded_path}\x1b\\"
+                    ).as_bytes());
+                } else {
+                    let encoded = base64::engine::general_purpose::STANDARD.encode(&rgba);
+                    let chunks = encoded.as_bytes().chunks(4096);
+                    let chunk_count = chunks.len();
+                    for (index, chunk) in chunks.enumerate() {
+                        let more = u8::from(index + 1 < chunk_count);
+                        if index == 0 {
+                            payload.extend_from_slice(format!(
+                            "\x1b_Ga=T,f=32,t=d,i=7,p=3,s={IMAGE_WIDTH},v={IMAGE_HEIGHT},c=1,r=1,q=2,m={more};"
+                        ).as_bytes());
+                        } else {
+                            payload.extend_from_slice(format!("\x1b_Gm={more};").as_bytes());
+                        }
+                        payload.extend_from_slice(chunk);
+                        payload.extend_from_slice(b"\x1b\\");
+                    }
+                }
+                for pane in &panes {
+                    write_shared_test_pane(
+                        &mut server,
+                        *pane,
+                        format!("\x1b[Hframe {sample:03}").as_bytes(),
+                    );
+                }
+                write_shared_test_pane(&mut server, root, &payload);
+                let retained_source_path = if source_retention {
+                    let runtime = server
+                        .app
+                        .state
+                        .runtime_for_pane_in_workspace(&server.app.terminal_runtimes, 0, root)
+                        .unwrap();
+                    let placements =
+                        runtime.kitty_image_placements_with_data_filter(|descriptor| {
+                            assert!(descriptor.source_file,
+                            "source profile requires working CoW retention, not decoded fallback");
+                            false
+                        });
+                    assert_eq!(placements.len(), 1);
+                    assert!(placements[0].data.is_empty());
+                    let source = placements[0].source_file.as_ref().expect("retained source");
+                    let path = source.path().to_owned();
+                    if let Some(previous) = previous_source_path.take() {
+                        assert_ne!(previous, path);
+                        assert!(!previous.exists(), "replacement releases previous source");
+                    }
+                    Some(path)
+                } else {
+                    None
+                };
+                server.app.render_dirty.take();
+
+                let started = Instant::now();
+                let retained = server.render_retained_pane_surface_and_stream(&sources);
+                let elapsed = started.elapsed();
+                assert!(
+                    retained,
+                    "profile must not silently fall back to full rendering"
+                );
+                if sample >= WARMUP {
+                    times.push(elapsed);
+                }
+
+                // Validate outside timing. Decoded exports close on ACK; source
+                // snapshots remain owned by the image until its replacement.
+                let messages = drain_native_render_messages(&writer);
+                let surfaces = messages
+                    .iter()
+                    .filter_map(|message| match message {
+                        ServerMessage::PaneSurface(surface) => Some(surface),
+                        _ => None,
+                    })
+                    .collect::<Vec<_>>();
+                assert_eq!(surfaces.len(), 1);
+                let graphics = &surfaces[0].graphics;
+                assert_eq!(graphics.placements.len(), 1);
+                assert_eq!(graphics.placements[0].asset.image_width, IMAGE_WIDTH);
+                assert_eq!(graphics.placements[0].asset.image_height, IMAGE_HEIGHT);
+                let files = messages
+                    .iter()
+                    .filter_map(|message| match message {
+                        ServerMessage::GraphicsFile {
+                            path,
+                            expected_len,
+                            transfer_id,
+                            image_id,
+                            ..
+                        } => Some((path, *expected_len, *transfer_id, *image_id)),
+                        _ => None,
+                    })
+                    .collect::<Vec<_>>();
+                if native {
+                    assert!(graphics.assets.is_empty());
+                    assert_eq!(files.len(), 1);
+                    let (path, expected_len, token, image) = files[0];
+                    assert_eq!(expected_len, rgba.len() as u64);
+                    assert_eq!(std::fs::read(path).unwrap(), rgba);
+                    assert_ne!(token & NATIVE_TRANSFER_BIT, 0);
+                    native_started(&mut server, 1, token, image);
+                    native_result(&mut server, 1, token, image, true);
+                    if source_retention {
+                        assert_eq!(
+                            retained_source_path.as_deref(),
+                            Some(std::path::Path::new(path))
+                        );
+                        assert!(
+                            std::path::Path::new(path).exists(),
+                            "ACK must preserve image backing"
+                        );
+                        previous_source_path = Some(PathBuf::from(path));
+                    } else {
+                        assert!(!std::path::Path::new(path).exists());
+                    }
+                } else {
+                    assert!(files.is_empty());
+                    assert_eq!(graphics.assets.len(), 1);
+                    assert_eq!(graphics.assets[0].data, rgba);
+                }
+                assert!(!server.native_graphics.is_pending(1));
+                assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+            }
+            times.sort_unstable();
+            println!(
+                "native_file_render_scale retained=true native_files={native} source_retention={source_retention} viewport=160x48 populated_panes={count} changing_images=1 image=800x480 warmup={WARMUP} samples={SAMPLES} median_us={} p95_us={}",
+                times[SAMPLES / 2].as_micros(), times[SAMPLES * 95 / 100].as_micros(),
+            );
+            shutdown_test_runtimes(&mut server);
+            drop(server);
+            if let Some(path) = previous_source_path {
+                // Aborted compression tasks release their terminal reference on poll.
+                for _ in 0..100 {
+                    if !path.exists() {
+                        break;
+                    }
+                    tokio::task::yield_now().await;
+                }
+                assert!(!path.exists(), "terminal shutdown releases final source");
+            }
+        }
+    }
+}
+
+fn source_backed_scene(
+    source: &Arc<crate::pane_graphics_files::OwnedExport>,
+) -> (
+    SurfaceGraphicsScene,
+    crate::kitty_graphics::surface::SourceFiles,
+) {
+    let mut desired = scene();
+    let key = &mut desired.assets[0].key;
+    key.image_width = (source.len() / 4) as u32;
+    key.data_len = source.len() as u64;
+    key.data_fingerprint = source.fingerprint();
+    desired.placements[0].asset = key.clone();
+    desired.placements[0].source_width = key.image_width;
+    let sources = std::collections::HashMap::from([(key.clone(), Arc::clone(source))]);
+    desired.assets[0].data.clear();
+    (desired, sources)
+}
+
+fn assert_source_upload_and_ack(
+    source: &Arc<crate::pane_graphics_files::OwnedExport>,
+    pixels: &[u8],
+) {
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    let (mut desired, mut sources) = source_backed_scene(source);
+    let mut delivery = DeliveryCache::default();
+    let (pending, message) = server
+        .prepare_native_scene(7, &mut desired, &mut delivery, &mut sources)
+        .expect("source-backed native upload");
+    assert!(desired.assets.is_empty(), "metadata must not carry pixels");
+    assert_eq!(desired.placements.len(), 1);
+    assert!(sources.is_empty());
+    let ServerMessage::GraphicsFile {
+        path,
+        transfer_id,
+        image_id,
+        expected_len,
+        surface_asset,
+        leading,
+        control,
+    } = message
+    else {
+        panic!("expected path-only upload");
+    };
+    assert_eq!(
+        PathBuf::from(&path),
+        source.path(),
+        "reuse source, not decoded export"
+    );
+    assert_eq!(expected_len, pixels.len() as u64);
+    assert_eq!(surface_asset.as_ref(), Some(&desired.placements[0].asset));
+    assert!(leading.is_empty());
+    assert_eq!(
+        control,
+        format!("a=t,f=32,s={},v=1,i={image_id},q=0", pixels.len() / 4)
+    );
+    assert_ne!(transfer_id & NATIVE_TRANSFER_BIT, 0);
+    assert_eq!(std::fs::read(&path).unwrap(), pixels);
+    commit_scene(&mut server, 7, &desired);
+    server.native_graphics.commit(7, pending);
+    native_started(&mut server, 7, transfer_id, image_id);
+    native_result(&mut server, 7, transfer_id, image_id, true);
+    assert!(!server.native_graphics.is_pending(7));
+    // The producer image's Arc still owns the backing after the transfer ACK.
+    assert_eq!(std::fs::read(&path).unwrap(), pixels);
+    assert_eq!(source.copy_rgba().unwrap(), pixels);
+}
+
+#[test]
+fn source_backed_native_upload_reuses_path_and_ack_preserves_image_backing() {
+    let store = crate::pane_graphics_files::FileStore::native_sources();
+    let pixels = [1, 2, 3, 255];
+    let source = Arc::new(store.export(&pixels).unwrap());
+    let path = source.path().to_owned();
+    assert_source_upload_and_ack(&source, &pixels);
+    drop(source);
+    assert!(!path.exists(), "final image owner releases the backing");
+}
+
+#[test]
+fn source_backed_client_without_direct_support_materializes_exact_pixels() {
+    let store = crate::pane_graphics_files::FileStore::native_sources();
+    let pixels = [1, 2, 3, 255, 200, 100, 50, 0];
+    let source = Arc::new(store.export(&pixels).unwrap());
+    let (mut desired, mut sources) = source_backed_scene(&source);
+    let expected_placements = desired.placements.clone();
+    let mut server = test_headless_server();
+    let (_control, _render) = add_client(&mut server, 7);
+    server.clients.get_mut(&7).unwrap().direct_graphics = false;
+    assert!(server
+        .prepare_native_scene(7, &mut desired, &mut DeliveryCache::default(), &mut sources)
+        .is_none());
+    assert!(sources.is_empty());
+    assert_eq!(desired.assets.len(), 1);
+    assert_eq!(desired.assets[0].data, pixels);
+    assert_eq!(desired.placements, expected_placements);
+    assert!(!server.native_graphics.is_pending(7));
+}
+
+#[cfg(target_os = "linux")]
+#[test]
+fn snapshotted_source_survives_producer_unlink_and_native_ack() {
+    use std::os::fd::AsRawFd;
+    use std::os::unix::fs::OpenOptionsExt as _;
+
+    let store = crate::pane_graphics_files::FileStore::native_sources();
+    let producer_path = store.source_directory().unwrap().join("producer.rgba");
+    let pixels = vec![73; 4096];
+    let mut producer = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(&producer_path)
+        .unwrap();
+    std::io::Write::write_all(&mut producer, &pixels).unwrap();
+    let snapshot = store.snapshot(i64::from(producer.as_raw_fd()), pixels.len());
+    std::fs::remove_file(&producer_path).unwrap();
+    drop(producer);
+    let source = match snapshot {
+        Ok(source) => Arc::new(source),
+        Err(error) => {
+            assert!(
+                error.kind() == std::io::ErrorKind::Unsupported
+                    || matches!(
+                        error.raw_os_error(),
+                        Some(
+                            libc::EOPNOTSUPP
+                                | libc::ENOTTY
+                                | libc::EXDEV
+                                | libc::EINVAL
+                                | libc::ENOSYS
+                        )
+                    ),
+                "unexpected snapshot failure: {error}"
+            );
+            eprintln!("native snapshot unsupported on test filesystem: {error}");
+            return;
+        }
+    };
+    assert!(!producer_path.exists());
+    let path = source.path().to_owned();
+    assert_source_upload_and_ack(&source, &pixels);
+    drop(source);
+    assert!(!path.exists());
+}
+
+#[tokio::test]
+async fn failed_source_read_retries_after_identical_suppression_without_producer_activity() {
+    let (mut server, _control, _render, _pane) =
+        retained_test_server_with_control(b"quiet source producer");
+    let writer = ClientWriter::test_paused();
+    let client = server.clients.get_mut(&1).unwrap();
+    client.writer = Some(writer.clone());
+    client.mode = ClientConnectionMode::ClientShell;
+    client.render_state =
+        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
+    server.clients.get_mut(&1).unwrap().direct_graphics = false;
+    server.render_and_stream();
+    let mut surface = drain_native_render_messages(&writer)
+        .into_iter()
+        .find_map(|message| match message {
+            ServerMessage::PaneSurface(surface) => Some(surface),
+            _ => None,
+        })
+        .expect("initial quiet surface");
+
+    // Inject the collected sidecar at the scheduler boundary, avoiding global
+    // environment changes and a filesystem-dependent CoW requirement.
+    let store = crate::pane_graphics_files::FileStore::native_sources();
+    let pixels = [1, 2, 3, 255];
+    let source = Arc::new(store.export(&pixels).unwrap());
+    std::fs::write(source.path(), []).unwrap();
+    let mut delivery = DeliveryCache::default();
+    for attempt in 0..2 {
+        let (mut graphics, mut sources) = source_backed_scene(&source);
+        assert!(server
+            .prepare_native_scene(1, &mut graphics, &mut delivery, &mut sources)
+            .is_none());
+        assert!(graphics.assets.is_empty());
+        surface.graphics = graphics;
+        let client = server.clients.get_mut(&1).unwrap();
+        let prepared = client.render_state.prepare_pane_surface(surface.clone());
+        if attempt == 0 {
+            client
+                .render_state
+                .commit_sent_frame(prepared.expect("first failed scene"));
+        } else {
+            assert!(
+                prepared.is_none(),
+                "repeated read failure produces identical metadata"
+            );
+            // Exactly the full-render identical-frame suppression branch.
+            client.clear_deferred_render();
+        }
+    }
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+    server.app.render_dirty.take();
+
+    // The timer is one-shot: an inactive client does not receive repeated wakes
+    // unless a real materialization attempt fails and explicitly re-arms it.
+    let tick = Instant::now() + Duration::from_secs(2);
+    assert!(server.expire_native_graphics(tick));
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::Full);
+    server.clients.get_mut(&1).unwrap().clear_deferred_render();
+    assert!(!server.expire_native_graphics(tick + Duration::from_secs(3600)));
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+
+    let (mut failed_graphics, mut failed_sources) = source_backed_scene(&source);
+    assert!(server
+        .prepare_native_scene(1, &mut failed_graphics, &mut delivery, &mut failed_sources)
+        .is_none());
+    assert!(failed_graphics.assets.is_empty());
+    let rearmed_tick = tick + Duration::from_secs(7200);
+    assert!(server.expire_native_graphics(rearmed_tick));
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::Full);
+
+    // No PTY bytes, writer progress, or native ACK occurs after recovery.
+    std::fs::write(source.path(), pixels).unwrap();
+    let (mut graphics, mut sources) = source_backed_scene(&source);
+    assert!(server
+        .prepare_native_scene(1, &mut graphics, &mut delivery, &mut sources)
+        .is_none());
+    assert_eq!(graphics.assets[0].data, pixels);
+    surface.graphics = graphics;
+    let client = server.clients.get_mut(&1).unwrap();
+    let prepared = client
+        .render_state
+        .prepare_pane_surface(surface)
+        .expect("recovered pixels must not be suppressed as identical");
+    let ServerMessage::PaneSurface(recovered) = prepared.message() else {
+        panic!("expected recovered surface");
+    };
+    assert_eq!(recovered.graphics.assets[0].data, pixels);
+    client.render_state.commit_sent_frame(prepared);
+    client.clear_deferred_render();
+    assert!(
+        !server.expire_native_graphics(rearmed_tick + Duration::from_secs(2)),
+        "successful materialization leaves the one-shot retry timer disarmed"
+    );
+    assert_eq!(server.clients[&1].deferred_render(), DeferredRender::None);
+    shutdown_test_runtimes(&mut server);
+}
diff --git a/src/server/headless/tests/pane_graphics.rs b/src/server/headless/tests/pane_graphics.rs
index 3e36a8fc..190eaa80 100644
--- a/src/server/headless/tests/pane_graphics.rs
+++ b/src/server/headless/tests/pane_graphics.rs
@@ -5,93 +5,6 @@ fn receive_render(receiver: &std::sync::mpsc::Receiver<Vec<u8>>, timeout: Durati
 }
 
 #[tokio::test]
-async fn client_shell_surface_sends_complete_placements_and_each_live_asset_once() {
-    let (mut server, _control_rx, client_rx, pane_id) =
-        retained_test_server_with_control(b"client shell graphics");
-    let client = server.clients.get_mut(&1).unwrap();
-    client.mode = ClientConnectionMode::ClientShell;
-    client.render_state =
-        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
-    client.cell_size = crate::kitty_graphics::HostCellSize {
-        width_px: 10,
-        height_px: 20,
-    };
-    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3, 4]);
-
-    server.render_and_stream();
-    let first = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
-    let ServerMessage::PaneSurface(first) = first else {
-        panic!("expected client shell pane surface");
-    };
-    assert_eq!(first.graphics.placements.len(), 1);
-    assert_eq!(first.graphics.assets.len(), 1);
-    assert_eq!(first.graphics.assets[0].data, vec![1, 2, 3, 4]);
-    assert!(matches!(
-        first.graphics.placements[0].asset.source,
-        crate::protocol::SurfaceGraphicsSource::PaneLayer { .. }
-    ));
-
-    server.clients.get_mut(&1).unwrap().request_repaint();
-    server.render_and_stream();
-    let second = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
-    let ServerMessage::PaneSurface(second) = second else {
-        panic!("expected replacement client shell pane surface");
-    };
-    assert_eq!(second.graphics.placements, first.graphics.placements);
-    assert!(second.graphics.assets.is_empty());
-
-    // A post-commit typed surface.set(true) resets only this viewer's delivery cache, so the
-    // already selected target receives the asset again without relying on graphics that may have
-    // arrived during the frozen source frame.
-    assert!(server.set_client_shell_surface_active(1, true).is_some());
-    server.render_and_stream();
-    let replay = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
-    let ServerMessage::PaneSurface(replay) = replay else {
-        panic!("expected post-commit graphics replay");
-    };
-    assert_eq!(replay.graphics.placements, first.graphics.placements);
-    assert_eq!(replay.graphics.assets.len(), 1);
-    assert_eq!(replay.graphics.assets[0].data, vec![1, 2, 3, 4]);
-}
-
-#[tokio::test]
-async fn client_shell_asset_delivery_is_bounded_to_the_current_live_scene() {
-    let (mut server, _control_rx, client_rx, pane_id) =
-        retained_test_server_with_control(b"client shell graphics");
-    let client = server.clients.get_mut(&1).unwrap();
-    client.mode = ClientConnectionMode::ClientShell;
-    client.render_state =
-        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
-    client.cell_size = crate::kitty_graphics::HostCellSize {
-        width_px: 10,
-        height_px: 20,
-    };
-    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3, 4]);
-
-    server.render_and_stream();
-    let _first = receive_render(&client_rx, Duration::from_millis(100));
-    server.app.pane_graphics.slots.clear();
-    server.app.pane_graphics.mark_changed();
-    server.clients.get_mut(&1).unwrap().request_repaint();
-    server.render_and_stream();
-    let removed = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
-    let ServerMessage::PaneSurface(removed) = removed else {
-        panic!("expected removed client shell scene");
-    };
-    assert!(removed.graphics.placements.is_empty());
-
-    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3, 4]);
-    server.clients.get_mut(&1).unwrap().request_repaint();
-    server.render_and_stream();
-    let restored = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
-    let ServerMessage::PaneSurface(restored) = restored else {
-        panic!("expected restored client shell scene");
-    };
-    assert_eq!(restored.graphics.assets.len(), 1);
-    assert_eq!(restored.graphics.assets[0].data, vec![1, 2, 3, 4]);
-}
-
-#[tokio::test]
 async fn first_kitty_image_updates_retained_surface_without_full_redraw() {
     let (mut server, _control_rx, client_rx, pane_id) =
         retained_test_server_with_control(b"text before image");
@@ -409,87 +322,6 @@ async fn client_shell_delivers_equal_pixels_for_distinct_terminal_image_ids() {
     assert!(surface.graphics.assets.is_empty());
 }
 
-#[tokio::test]
-async fn full_client_shell_render_lane_does_not_commit_graphics_delivery() {
-    let (mut server, _control_rx, client_rx, pane_id) =
-        retained_test_server_with_control(b"client shell graphics");
-    let client = server.clients.get_mut(&1).unwrap();
-    client.mode = ClientConnectionMode::ClientShell;
-    client.render_state =
-        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
-    client.cell_size = crate::kitty_graphics::HostCellSize {
-        width_px: 10,
-        height_px: 20,
-    };
-    set_graphics_layer(&mut server, pane_id, vec![5, 6, 7, 8]);
-
-    fill_render_lane(&server);
-    server.render_and_stream();
-    let _older = receive_render(&client_rx, Duration::from_millis(100));
-    server.render_and_stream();
-    let message = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
-    let ServerMessage::PaneSurface(surface) = message else {
-        panic!("expected client shell pane surface");
-    };
-    assert_eq!(surface.graphics.assets.len(), 1);
-    assert_eq!(surface.graphics.assets[0].data, vec![5, 6, 7, 8]);
-}
-
-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>) {
-    set_named_graphics_layer(
-        server,
-        pane_id,
-        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");
@@ -500,105 +332,6 @@ fn fill_render_lane(server: &HeadlessServer) {
         .test_fill_render(queued);
 }
 
-fn stream_set_message(
-    id: &str,
-    pane_id: &str,
-    owner: &str,
-    data: Vec<u8>,
-) -> (api::ApiRequestMessage, std::sync::mpsc::Receiver<String>) {
-    let (respond_to, response_rx) = std::sync::mpsc::channel();
-    (
-        api::ApiRequestMessage {
-            request: api::schema::Request {
-                id: id.into(),
-                method: api::schema::Method::PaneGraphicsStreamSet(
-                    api::schema::PaneGraphicsSetParams {
-                        pane_id: pane_id.into(),
-                        layer_id: None,
-                        z_index: 0,
-                        owner: owner.into(),
-                        format: api::schema::PaneGraphicsFormat::Png,
-                        image_width: 1,
-                        image_height: 1,
-                        data: Some(data),
-                        data_base64: String::new(),
-                        placement: api::schema::PaneGraphicsPlacementParams::default(),
-                    },
-                ),
-            },
-            respond_to,
-            response_write_complete: None,
-            stream_active: None,
-        },
-        response_rx,
-    )
-}
-
-#[cfg(unix)]
-fn sparse_direct_frame(
-    server: &HeadlessServer,
-    name: &str,
-    image_width: u32,
-    image_height: u32,
-) -> String {
-    use std::os::unix::fs::OpenOptionsExt as _;
-
-    let path = server
-        .app
-        .pane_graphics_files
-        .source_directory()
-        .unwrap()
-        .join(name);
-    let file = std::fs::OpenOptions::new()
-        .write(true)
-        .create_new(true)
-        .mode(0o600)
-        .open(&path)
-        .unwrap();
-    file.set_len(u64::from(image_width) * u64::from(image_height) * 4)
-        .unwrap();
-    path.to_string_lossy().into_owned()
-}
-
-#[cfg(unix)]
-fn direct_stream_message(
-    id: &str,
-    pane_id: &str,
-    owner: &str,
-    path: String,
-    image_width: u32,
-    image_height: u32,
-    sequence: u64,
-) -> (api::ApiRequestMessage, std::sync::mpsc::Receiver<String>) {
-    let (respond_to, response_rx) = std::sync::mpsc::channel();
-    (
-        api::ApiRequestMessage {
-            request: api::schema::Request {
-                id: id.into(),
-                method: api::schema::Method::PaneGraphicsStreamDirect(
-                    api::schema::PaneGraphicsDirectParams {
-                        pane_id: pane_id.into(),
-                        layer_id: None,
-                        z_index: 0,
-                        owner: owner.into(),
-                        image_width,
-                        image_height,
-                        format: api::schema::PaneGraphicsFormat::Rgba,
-                        path,
-                        sequence,
-                        revision: 1,
-                        placement: Default::default(),
-                    },
-                ),
-            },
-            respond_to,
-            response_write_complete: None,
-            stream_active: None,
-        },
-        response_rx,
-    )
-}
-
 #[tokio::test]
 async fn pixel_mouse_activation_follows_child_1016_without_graphics_demand() {
     let (mut server, _client_rx, _pane_id) =
@@ -614,7 +347,6 @@ async fn pixel_mouse_activation_follows_child_1016_without_graphics_demand() {
     };
     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!(
@@ -625,868 +357,3 @@ async fn pixel_mouse_activation_follows_child_1016_without_graphics_demand() {
         }
     ));
 }
-
-#[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());
-}
-
-#[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");
-    let pane_id = workspace.tabs[0].root_pane;
-    let public_pane_id = format!("{}:p1", workspace.id);
-    server.app.state.workspaces = vec![workspace];
-    server.app.state.active = Some(0);
-    server.app.state.selected = 0;
-    server.app.state.kitty_graphics_enabled = true;
-    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]);
-    assert_eq!(
-        server.handle_api_request_with_render_impact(request),
-        RenderImpact::None
-    );
-    assert!(serde_json::from_str::<api::schema::ErrorResponse>(
-        &response_rx
-            .recv_timeout(Duration::from_millis(100))
-            .unwrap()
-    )
-    .is_ok());
-
-    let (request, response_rx) =
-        stream_set_message("stream-frame", &public_pane_id, "owner-a", vec![1, 2, 3]);
-    assert_eq!(
-        server.handle_api_request_with_render_impact(request),
-        RenderImpact::Graphics
-    );
-    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
-        &response_rx
-            .recv_timeout(Duration::from_millis(100))
-            .unwrap()
-    )
-    .is_ok());
-
-    server
-        .app
-        .event_tx
-        .try_send(AppEvent::UpdateReady {
-            version: "9.9.9".into(),
-            install_command: "herdr update".into(),
-        })
-        .unwrap();
-    let (request, _response_rx) = stream_set_message(
-        "stream-frame-with-internal-event",
-        &public_pane_id,
-        "owner-a",
-        vec![4, 5, 6],
-    );
-    assert_eq!(
-        server.handle_api_request_with_render_impact(request),
-        RenderImpact::Full
-    );
-
-    server.app.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,
-                image_height: 1,
-                data: Some(vec![1, 2, 3]),
-                data_base64: String::new(),
-                placement: api::schema::PaneGraphicsPlacementParams::default(),
-            }),
-        },
-        respond_to,
-        response_write_complete: None,
-        stream_active: None,
-    });
-    assert_eq!(impact, RenderImpact::Full);
-}
-
-#[test]
-fn rejected_or_stale_requests_do_not_schedule_rendering() {
-    let mut server = test_headless_server();
-    let workspace = crate::workspace::Workspace::test_new("graphics");
-    let pane_id = workspace.tabs[0].root_pane;
-    let public_pane_id = format!("{}:p1", workspace.id);
-    server.app.state.workspaces = vec![workspace];
-    server.app.state.active = Some(0);
-    server.app.state.selected = 0;
-    server.app.state.kitty_graphics_enabled = false;
-
-    let (respond_to, response_rx) = std::sync::mpsc::channel();
-    let changed = server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
-        request: api::schema::Request {
-            id: "disabled-set".into(),
-            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
-                pane_id: public_pane_id.clone(),
-                layer_id: None,
-                z_index: 0,
-                owner: String::new(),
-                format: api::schema::PaneGraphicsFormat::Png,
-                image_width: 1,
-                image_height: 1,
-                data: Some(vec![1, 2, 3]),
-                data_base64: String::new(),
-                placement: api::schema::PaneGraphicsPlacementParams::default(),
-            }),
-        },
-        respond_to,
-        response_write_complete: None,
-        stream_active: None,
-    });
-    assert!(!changed);
-    let response = response_rx
-        .recv_timeout(Duration::from_millis(100))
-        .unwrap();
-    assert_eq!(
-        serde_json::from_str::<api::schema::ErrorResponse>(&response)
-            .unwrap()
-            .error
-            .code,
-        "feature_disabled"
-    );
-
-    server.app.state.kitty_graphics_enabled = true;
-    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 {
-            id: "stale-close".into(),
-            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
-            .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
-            .recv_timeout(Duration::from_millis(100))
-            .unwrap()
-    )
-    .is_ok());
-}
-
-#[cfg(unix)]
-#[tokio::test]
-async fn direct_graphics_availability_follows_foreground_client_with_background_clients() {
-    let (mut server, _client_rx, _pane_id) = retained_test_server(b"direct eligibility");
-    let foreground = server.clients.get_mut(&1).unwrap();
-    foreground.direct_graphics = true;
-    foreground.pixel_mouse = true;
-
-    let (background_writer, _background_control_rx, _background_render_rx) = test_client_writer();
-    server.clients.insert(
-        2,
-        ClientConnection::new(
-            (80, 24),
-            crate::kitty_graphics::HostCellSize::default(),
-            2,
-            RenderEncoding::SemanticFrame,
-            Some(background_writer),
-        ),
-    );
-
-    server.sync_foreground_client_state();
-    assert!(server.direct_graphics_available());
-    assert!(server.app.direct_graphics_available);
-
-    server.foreground_client_id = Some(2);
-    server.sync_foreground_client_state();
-    assert!(!server.direct_graphics_available());
-    assert!(!server.app.direct_graphics_available);
-}
-
-#[cfg(unix)]
-#[test]
-fn direct_graphics_routing_prefers_the_target_stream_owner() {
-    let (mut server, first_key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
-    add_direct_client(&mut server, 7);
-    add_direct_client(&mut server, 8);
-    server.clients.get_mut(&8).unwrap().last_activity = 2;
-    let (transfer_id, image_id) = direct_ids(&server, &first_key);
-    assert!(server.complete_direct_graphics(7, transfer_id, image_id, true));
-    assert_eq!(response_rx.recv().unwrap(), "ack");
-
-    let second_key = (first_key.0, "second".into());
-    let second_layer = crate::app::pane_graphics::Layer {
-        format: crate::api::schema::PaneGraphicsFormat::Rgba,
-        image_width: 1,
-        image_height: 1,
-        backing: crate::app::pane_graphics::Backing::Resident {
-            len: 4,
-            client_id: 8,
-        },
-        data_fingerprint: 2,
-        render: Default::default(),
-        z_index: 0,
-    };
-    server.app.pane_graphics.slots.insert(
-        second_key.clone(),
-        crate::app::pane_graphics::Slot::test((1 << 31) | 901, Some(second_layer)),
-    );
-
-    assert_eq!(server.direct_graphics_client_for_key(&first_key), Some(7));
-    assert_eq!(server.direct_graphics_client_for_key(&second_key), Some(8));
-    assert_eq!(
-        server.direct_graphics_client_for_key(&(first_key.0, "new".into())),
-        Some(8)
-    );
-}
-
-#[cfg(unix)]
-#[test]
-fn resident_direct_stream_survives_non_direct_client_becoming_foreground() {
-    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
-    add_direct_client(&mut server, 7);
-    server.foreground_client_id = Some(7);
-    server.sync_foreground_client_state();
-    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 (writer, _control_rx, _render_rx) = test_client_writer();
-    server.clients.insert(
-        8,
-        ClientConnection::new(
-            (80, 24),
-            crate::kitty_graphics::HostCellSize::default(),
-            2,
-            RenderEncoding::SemanticFrame,
-            Some(writer),
-        ),
-    );
-    server.foreground_client_id = Some(8);
-    server.sync_foreground_client_state();
-
-    assert!(server.direct_graphics_available());
-    assert!(server.app.direct_graphics_available);
-    assert_eq!(
-        server.app.pane_graphics.slots[&key]
-            .layer
-            .as_ref()
-            .and_then(crate::app::pane_graphics::Layer::resident_client),
-        Some(7)
-    );
-
-    server.remove_client_and_resize_if_needed(7);
-    assert!(!server.app.pane_graphics.slots.contains_key(&key));
-    assert!(!server.direct_graphics_available());
-    assert!(!server.app.direct_graphics_available);
-}
-
-#[cfg(unix)]
-#[tokio::test]
-async fn client_shell_direct_graphics_uploads_without_server_authored_coordinates() {
-    let (mut server, _control_rx, client_rx, pane_id) =
-        retained_test_server_with_control(b"client shell direct");
-    server.app.state.kitty_graphics_enabled = true;
-    let client = server.clients.get_mut(&1).unwrap();
-    client.mode = ClientConnectionMode::ClientShell;
-    client.render_state =
-        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
-    client.cell_size = crate::kitty_graphics::HostCellSize {
-        width_px: 10,
-        height_px: 20,
-    };
-    client.direct_graphics = true;
-    client.pixel_mouse = true;
-    server.app.direct_graphics_available = true;
-    let (background_writer, _background_control_rx, background_render_rx) = test_client_writer();
-    server.clients.insert(
-        2,
-        ClientConnection::new(
-            (80, 24),
-            crate::kitty_graphics::HostCellSize {
-                width_px: 10,
-                height_px: 20,
-            },
-            2,
-            RenderEncoding::SemanticFrame,
-            Some(background_writer),
-        ),
-    );
-    set_stream_owner(&mut server, pane_id, "browser");
-    let public_pane_id = server.app.public_pane_id(0, pane_id).unwrap();
-    let path = sparse_direct_frame(&server, "client-shell-direct.rgba", 1, 1);
-    let (message, response_rx) =
-        direct_stream_message("shell-direct", &public_pane_id, "browser", path, 1, 1, 1);
-
-    assert_eq!(
-        server.handle_pane_graphics_stream_frame(message),
-        RenderImpact::None
-    );
-    let (transfer_id, image_id, asset) = match read_server_message(
-        client_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("client shell direct upload"),
-    ) {
-        ServerMessage::GraphicsFile {
-            transfer_id,
-            image_id,
-            leading,
-            control,
-            surface_asset: Some(asset),
-            ..
-        } => {
-            assert!(leading.is_empty());
-            assert!(control.starts_with("a=t,"), "{control}");
-            assert!(!control.contains("\u{1b}["), "{control}");
-            (transfer_id, image_id, asset)
-        }
-        other => panic!("expected client shell graphics file, got {other:?}"),
-    };
-    assert!(background_render_rx.try_recv().is_err());
-    assert_eq!(
-        image_id,
-        crate::kitty_graphics::surface::host_image_id(&server.client_shell_boot_id, &asset)
-    );
-    let (pending, _) = crate::server::client_shell_graphics::collect(
-        &server.app,
-        &[],
-        &[],
-        None,
-        Some(crate::ui::TabSurfaceTarget {
-            workspace_index: 0,
-            tab_index: 0,
-        }),
-        crate::kitty_graphics::HostCellSize {
-            width_px: 10,
-            height_px: 20,
-        },
-        &crate::kitty_graphics::surface::DeliveryCache::default(),
-        1,
-    );
-    assert_eq!(pending.retained_assets, vec![asset.clone()]);
-
-    server.start_direct_graphics_response(1, transfer_id, image_id);
-    assert!(server.complete_direct_graphics(1, transfer_id, image_id, true));
-    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
-        &response_rx.recv_timeout(Duration::from_secs(1)).unwrap()
-    )
-    .is_ok());
-
-    server.foreground_client_id = Some(2);
-    server.sync_foreground_client_state();
-    let next_path = sparse_direct_frame(&server, "client-shell-direct-next.rgba", 1, 1);
-    let (next_message, next_response_rx) = direct_stream_message(
-        "shell-direct-next",
-        &public_pane_id,
-        "browser",
-        next_path,
-        1,
-        1,
-        2,
-    );
-    assert_eq!(
-        server.handle_pane_graphics_stream_frame(next_message),
-        RenderImpact::None
-    );
-    let (next_transfer_id, next_image_id, next_asset) = match read_server_message(
-        client_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("sticky owner direct upload"),
-    ) {
-        ServerMessage::GraphicsFile {
-            transfer_id,
-            image_id,
-            surface_asset: Some(asset),
-            ..
-        } => (transfer_id, image_id, asset),
-        other => panic!("expected sticky owner graphics file, got {other:?}"),
-    };
-    assert!(background_render_rx.try_recv().is_err());
-    server.start_direct_graphics_response(1, next_transfer_id, next_image_id);
-    assert!(server.complete_direct_graphics(1, next_transfer_id, next_image_id, true));
-    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
-        &next_response_rx
-            .recv_timeout(Duration::from_secs(1))
-            .unwrap()
-    )
-    .is_ok());
-    assert!(server.app.direct_graphics_available);
-
-    server.clients.get_mut(&1).unwrap().request_repaint();
-    server.render_and_stream();
-    let surface = read_server_message(
-        client_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("resident client shell scene"),
-    );
-    let ServerMessage::PaneSurface(surface) = surface else {
-        panic!("expected client shell pane surface");
-    };
-    assert_eq!(surface.graphics.placements.len(), 1);
-    assert!(surface.graphics.assets.is_empty());
-    assert_eq!(surface.graphics.placements[0].asset, next_asset);
-    assert_eq!(surface.graphics.retained_assets, vec![next_asset.clone()]);
-
-    let background_surface = read_server_message(
-        background_render_rx
-            .recv_timeout(Duration::from_secs(1))
-            .expect("non-owning client shell scene"),
-    );
-    let ServerMessage::PaneSurface(background_surface) = background_surface else {
-        panic!("expected non-owning client shell pane surface");
-    };
-    assert!(background_surface.graphics.assets.is_empty());
-    assert!(background_surface.graphics.placements.is_empty());
-    assert!(background_surface.graphics.retained_assets.is_empty());
-
-    let (hidden, _) = crate::server::client_shell_graphics::collect(
-        &server.app,
-        &[],
-        &[],
-        None,
-        Some(crate::ui::TabSurfaceTarget {
-            workspace_index: 0,
-            tab_index: 0,
-        }),
-        crate::kitty_graphics::HostCellSize {
-            width_px: 10,
-            height_px: 20,
-        },
-        &crate::kitty_graphics::surface::DeliveryCache::default(),
-        1,
-    );
-    assert!(hidden.placements.is_empty());
-    assert_eq!(hidden.retained_assets, vec![next_asset]);
-}
-
-#[cfg(unix)]
-fn direct_gate_server(
-    data: &[u8],
-) -> (
-    HeadlessServer,
-    crate::app::pane_graphics::Key,
-    std::sync::mpsc::Receiver<String>,
-) {
-    direct_gate_server_with_file(data.len(), Some(data))
-}
-
-#[cfg(unix)]
-fn direct_gate_server_with_file(
-    len: usize,
-    data: Option<&[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();
-    if let Some(data) = data {
-        file.write_all(data).unwrap();
-    } else {
-        file.set_len(len as u64).unwrap();
-    }
-    drop(file);
-    let lease = server.app.pane_graphics_files.lease(&path, 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(),
-        image_id: (1 << 31) | 900,
-        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];
-    let gate = slot.direct_gate.as_ref().unwrap();
-    (gate.transfer_id, gate.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,
-        },
-        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 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 eligibility_loss_cancels_the_queued_direct_upload() {
-    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
-    let (writer, control_rx, _render_rx) = test_client_writer();
-    let mut client = ClientConnection::new(
-        (80, 24),
-        crate::kitty_graphics::HostCellSize {
-            width_px: 10,
-            height_px: 20,
-        },
-        1,
-        RenderEncoding::SemanticFrame,
-        Some(writer),
-    );
-    client.direct_graphics = true;
-    client.pixel_mouse = true;
-    server.clients.insert(7, client);
-    let (transfer_id, image_id) = direct_ids(&server, &key);
-
-    server.retire_all_direct_graphics();
-
-    assert!(matches!(
-        read_server_message(
-            control_rx
-                .recv_timeout(Duration::from_millis(100))
-                .expect("direct retirement")
-        ),
-        ServerMessage::GraphicsTransmissionRetired {
-            transfer_id: retired_transfer,
-            image_id: retired_image,
-        } if retired_transfer == transfer_id && retired_image == image_id
-    ));
-    assert!(!server.app.pane_graphics.slots.contains_key(&key));
-    assert!(response_rx.recv().is_err());
-}
-
-#[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_cancels_the_pending_client_upload() {
-    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
-    let (writer, control_rx, _render_rx) = test_client_writer();
-    let mut client = ClientConnection::new(
-        (80, 24),
-        crate::kitty_graphics::HostCellSize {
-            width_px: 10,
-            height_px: 20,
-        },
-        1,
-        RenderEncoding::SemanticFrame,
-        Some(writer),
-    );
-    client.direct_graphics = true;
-    client.pixel_mouse = true;
-    server.clients.insert(7, client);
-    let (transfer_id, image_id) = direct_ids(&server, &key);
-    server.app.state.workspaces.clear();
-
-    assert!(server.retain_live_pane_graphics());
-
-    assert!(matches!(
-        read_server_message(
-            control_rx
-                .recv_timeout(Duration::from_millis(100))
-                .expect("pane removal retirement")
-        ),
-        ServerMessage::GraphicsTransmissionRetired {
-            transfer_id: retired_transfer,
-            image_id: retired_image,
-        } if retired_transfer == transfer_id && retired_image == image_id
-    ));
-    assert!(!server.app.pane_graphics.slots.contains_key(&key));
-    assert!(response_rx.recv().is_err());
-}
-
-#[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);
-}
diff --git a/src/server/headless/tests/pane_move.rs b/src/server/headless/tests/pane_move.rs
index 789b144f..99b39c8b 100644
--- a/src/server/headless/tests/pane_move.rs
+++ b/src/server/headless/tests/pane_move.rs
@@ -29,7 +29,6 @@ fn public_move(
         },
         respond_to,
         response_write_complete: None,
-        stream_active: None,
     });
     let response = response_rx.recv().expect("pane move response");
     match serde_json::from_str::<SuccessResponse>(&response) {
diff --git a/src/server/render_stream.rs b/src/server/render_stream.rs
index db613e96..ee874e52 100644
--- a/src/server/render_stream.rs
+++ b/src/server/render_stream.rs
@@ -7,7 +7,7 @@ use crate::app::state::AppState;
 use crate::protocol::render_ansi::{BlitEncoder, EncodedBlit};
 use crate::protocol::{
     CursorState, FrameData, PaneSurfaceFrame, PaneSurfacePatch, RenderEncoding, ServerMessage,
-    TerminalFrame,
+    SurfaceGraphicsAssetKey, SurfaceGraphicsScene, TerminalFrame,
 };
 use crate::terminal::TerminalRuntimeRegistry;
 
@@ -152,9 +152,18 @@ impl ClientRenderState {
         }
     }
 
+    #[cfg(test)]
     pub(crate) fn prepare_pane_surface(
         &mut self,
+        surface: PaneSurfaceFrame,
+    ) -> Option<PreparedRender> {
+        self.prepare_pane_surface_with_file(surface, false)
+    }
+
+    pub(crate) fn prepare_pane_surface_with_file(
+        &mut self,
         mut surface: PaneSurfaceFrame,
+        has_file_upload: bool,
     ) -> Option<PreparedRender> {
         let Self::Semantic {
             last_surface,
@@ -166,7 +175,8 @@ impl ClientRenderState {
         else {
             return None;
         };
-        if !*recompute_pending
+        if !has_file_upload
+            && !*recompute_pending
             && surface.graphics.assets.is_empty()
             && last_surface.as_deref().is_some_and(|last| {
                 last.projection_revision == surface.projection_revision
@@ -182,6 +192,7 @@ impl ClientRenderState {
         }
         surface.surface_revision = surface_revision.saturating_add(1);
         let assets = std::mem::take(&mut surface.graphics.assets);
+        let queued_graphics_assets = assets.iter().map(|asset| asset.key.clone()).collect();
         let committed_surface = surface.clone();
         surface.graphics.assets = assets;
         let mut message = ServerMessage::PaneSurface(surface);
@@ -217,6 +228,7 @@ impl ClientRenderState {
         Some(PreparedRender::Semantic {
             message: delta.or(reused).unwrap_or(message),
             committed_surface: Box::new(committed_surface),
+            queued_graphics_assets,
         })
     }
 
@@ -342,6 +354,7 @@ pub(crate) enum PreparedRender {
     Semantic {
         message: ServerMessage,
         committed_surface: Box<PaneSurfaceFrame>,
+        queued_graphics_assets: Vec<SurfaceGraphicsAssetKey>,
     },
     SemanticPatch {
         message: ServerMessage,
@@ -362,19 +375,56 @@ impl PreparedRender {
         }
     }
 
-    pub(crate) fn strip_pane_surface_assets(&mut self) -> bool {
+    /// Graphics metadata represented by this semantic update plus only the
+    /// asset keys whose pixel payloads were queued. This is independent of the
+    /// selected wire codec and avoids cloning asset byte vectors.
+    pub(crate) fn queued_surface_graphics(
+        &self,
+    ) -> Option<(&SurfaceGraphicsScene, &[SurfaceGraphicsAssetKey])> {
+        match self {
+            Self::Semantic {
+                committed_surface,
+                queued_graphics_assets,
+                ..
+            } => Some((&committed_surface.graphics, queued_graphics_assets)),
+            Self::SemanticPatch { .. } | Self::TerminalAnsi { .. } => None,
+        }
+    }
+
+    pub(crate) fn has_queued_surface_assets(&self) -> bool {
+        matches!(self, Self::Semantic { queued_graphics_assets, .. } if !queued_graphics_assets.is_empty())
+    }
+
+    /// Removes the largest inline payload from a full semantic surface while
+    /// preserving placement metadata. Largest-first guarantees that a fitting
+    /// smaller asset is not discarded behind an oversized one. Equal sizes use
+    /// deterministic scene order. Encoded delta/reuse messages return `None`; callers
+    /// can invalidate that baseline and retry as a full surface.
+    pub(crate) fn pop_pane_surface_asset(&mut self) -> Option<SurfaceGraphicsAssetKey> {
         let Self::Semantic {
             message: ServerMessage::PaneSurface(surface),
+            queued_graphics_assets,
             ..
         } = self
         else {
-            return false;
+            return None;
         };
-        if surface.graphics.assets.is_empty() {
-            return false;
+        let index = surface
+            .graphics
+            .assets
+            .iter()
+            .enumerate()
+            .max_by_key(|(index, asset)| (asset.data.len(), *index))?
+            .0;
+        let asset = surface.graphics.assets.remove(index);
+        let key = asset.key;
+        if let Some(index) = queued_graphics_assets
+            .iter()
+            .position(|queued| *queued == key)
+        {
+            queued_graphics_assets.remove(index);
         }
-        surface.graphics.assets.clear();
-        true
+        Some(key)
     }
 }
 
@@ -731,6 +781,31 @@ mod tests {
     }
 
     #[test]
+    fn deferred_file_upload_keeps_identical_metadata_and_retries_without_committing() {
+        for reuse in [false, true] {
+            let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
+            state.enable_surface_reuse(reuse);
+            let surface = popup_surface("native");
+            let first = state.prepare_pane_surface(surface.clone()).unwrap();
+            state.commit_sent_frame(first);
+            assert!(state.prepare_pane_surface(surface.clone()).is_none());
+            let file = state
+                .prepare_pane_surface_with_file(surface.clone(), true)
+                .unwrap();
+            let retry = state
+                .prepare_pane_surface_with_file(surface.clone(), true)
+                .unwrap();
+            let config = bincode::config::standard();
+            assert_eq!(
+                bincode::serde::encode_to_vec(file.message(), config).unwrap(),
+                bincode::serde::encode_to_vec(retry.message(), config).unwrap()
+            );
+            state.commit_sent_frame(retry);
+            assert!(state.prepare_pane_surface(surface).is_none());
+        }
+    }
+
+    #[test]
     fn popup_only_surface_changes_are_not_deduplicated() {
         let mut state = ClientRenderState::new(RenderEncoding::SemanticFrame);
         let prepared = state