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>

4 files changed, 1415 insertions, 1154 deletions

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) {