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>

18 files changed, 1416 insertions, 167 deletions

diff --git a/src/client/catalog_reload.rs b/src/client/catalog_reload.rs
index a20b74dc..89715f7d 100644
--- a/src/client/catalog_reload.rs
+++ b/src/client/catalog_reload.rs
@@ -47,7 +47,7 @@ pub(super) fn apply_profiles(
         endpoints.disconnect(&endpoint_id);
         let cancelled = commands.disconnect(&endpoint_id);
         #[cfg(unix)]
-        state.retire_endpoint_graphics(&endpoint_id);
+        state.forget_endpoint_graphics(&endpoint_id);
         if let Some(shell) = state.shell.as_mut() {
             for request_id in cancelled {
                 shell.cancel_endpoint_request(&request_id);
diff --git a/src/client/endpoint/supervisor.rs b/src/client/endpoint/supervisor.rs
index ff86ac77..5b4e7ad1 100644
--- a/src/client/endpoint/supervisor.rs
+++ b/src/client/endpoint/supervisor.rs
@@ -303,6 +303,7 @@ fn connect_once(
         options.endpoint_keybindings,
         options.mouse_capture,
         false,
+        matches!(target, ConnectTarget::Local(_)),
     )
     .map_err(handshake_error)?;
     if handshake.encoding != RenderEncoding::SemanticFrame {
diff --git a/src/client/frame_output.rs b/src/client/frame_output.rs
index 5d12c78c..d9e03f14 100644
--- a/src/client/frame_output.rs
+++ b/src/client/frame_output.rs
@@ -1,11 +1,86 @@
 use std::collections::HashSet;
-use std::io;
+use std::io::{self, Write as _};
 use std::sync::{Mutex, OnceLock};
 
-use crate::protocol::render_ansi;
+use crate::kitty_graphics::{GraphicsOperation, GraphicsOutput};
+use crate::protocol::{render_ansi, FrameData};
+use base64::Engine as _;
+
+/// Local output only; raw uploads are never added to a published frame codec.
+#[derive(Debug)]
+pub(crate) struct ComposedFrame {
+    pub(crate) frame: FrameData,
+    pub(crate) graphics: GraphicsOutput,
+}
+
+impl From<FrameData> for ComposedFrame {
+    fn from(mut frame: FrameData) -> Self {
+        let graphics = GraphicsOutput::from_bytes(std::mem::take(&mut frame.graphics));
+        Self { frame, graphics }
+    }
+}
+
+impl std::ops::Deref for ComposedFrame {
+    type Target = FrameData;
+
+    fn deref(&self) -> &Self::Target {
+        &self.frame
+    }
+}
 
 static RECEIVED_KITTY_GRAPHICS_IDS: OnceLock<Mutex<HashSet<u32>>> = OnceLock::new();
 
+pub(super) fn write_composed_frame(
+    mut writer: impl io::Write,
+    encoded: &[u8],
+    graphics: &GraphicsOutput,
+    files: &mut super::image_files::FileTransport,
+) -> io::Result<()> {
+    if graphics.is_empty() {
+        return writer.write_all(encoded);
+    }
+    let mut writer = io::BufWriter::with_capacity(64 * 1024, writer);
+    let insertion = render_ansi::final_sync_output_end(encoded).unwrap_or(encoded.len());
+    writer.write_all(&encoded[..insertion])?;
+    writer.write_all(b"\x1b7")?;
+    for operation in &graphics.operations {
+        match operation {
+            GraphicsOperation::Bytes(bytes) => {
+                record_received_kitty_graphics(bytes);
+                writer.write_all(bytes)?;
+            }
+            GraphicsOperation::Upload { control, data } => {
+                let file_eligible = control
+                    .split(',')
+                    .any(|part| matches!(part, "f=24" | "f=32" | "f=100"));
+                if file_eligible {
+                    if let Some(path) = files
+                        .probe()
+                        .and_then(|path| path.to_str().map(str::to_owned))
+                    {
+                        let path =
+                            base64::engine::general_purpose::STANDARD.encode(path.as_bytes());
+                        write!(writer, "\x1b_Ga=q,t=t,f=32,s=1,v=1,i=1,q=2;{path}\x1b\\")?;
+                    }
+                }
+                let path = file_eligible.then(|| files.prepare(data)).flatten();
+                let header = format!("\x1b_G{control};\x1b\\");
+                record_received_kitty_graphics(header.as_bytes());
+                if let Some(path) = path.as_ref().and_then(|path| path.to_str()) {
+                    let control = control.replace(",t=d,", ",t=t,");
+                    let path = base64::engine::general_purpose::STANDARD.encode(path.as_bytes());
+                    write!(writer, "\x1b_G{control};{path}\x1b\\")?;
+                } else {
+                    crate::kitty_graphics::write_kitty_data(&mut writer, control, data)?;
+                }
+            }
+        }
+    }
+    writer.write_all(b"\x1b8")?;
+    writer.write_all(&encoded[insertion..])?;
+    io::Write::flush(&mut writer)
+}
+
 pub(super) fn write_encoded_frame_with_graphics(
     mut writer: impl io::Write,
     encoded: &[u8],
diff --git a/src/client/frame_output_tests.rs b/src/client/frame_output_tests.rs
new file mode 100644
index 00000000..19c3717b
--- /dev/null
+++ b/src/client/frame_output_tests.rs
@@ -0,0 +1,192 @@
+use super::frame_output::*;
+use super::image_files::FileTransport;
+use crate::kitty_graphics::{GraphicsOperation, GraphicsOutput};
+use std::sync::Arc;
+
+fn output(format: u32) -> GraphicsOutput {
+    GraphicsOutput {
+        operations: vec![
+            GraphicsOperation::Bytes(b"before".to_vec()),
+            GraphicsOperation::Upload {
+                control: format!("a=t,t=d,f={format},s=1,v=1,i=42,q=2"),
+                data: Arc::from([1, 2, 3, 4]),
+            },
+            GraphicsOperation::Bytes(b"after".to_vec()),
+        ],
+    }
+}
+
+#[test]
+fn deferred_inline_output_matches_existing_frame_envelope() {
+    let graphics = output(32);
+    let encoded = b"\x1b[?2026htext\x1b[?2026l";
+    let mut expected = Vec::new();
+    write_encoded_frame_with_graphics(
+        &mut expected,
+        encoded,
+        &graphics.clone().into_inline_bytes(),
+    )
+    .unwrap();
+    let mut actual = Vec::new();
+    write_composed_frame(
+        &mut actual,
+        encoded,
+        &graphics,
+        &mut FileTransport::default(),
+    )
+    .unwrap();
+    assert_eq!(actual, expected);
+}
+
+#[test]
+fn native_swap_cleanup_and_placements_share_the_frame_sync_envelope() {
+    let graphics = GraphicsOutput::from_bytes(
+        concat!(
+            "\x1b_Ga=d,d=I,i=41,q=2;\x1b\\",
+            "\x1b_Ga=p,i=42,p=1,q=2;\x1b\\",
+            "\x1b_Ga=p,i=42,p=2,q=2;\x1b\\",
+        )
+        .as_bytes()
+        .to_vec(),
+    );
+    let mut actual = Vec::new();
+    write_composed_frame(
+        &mut actual,
+        b"\x1b[?2026hframe\x1b[?2026l",
+        &graphics,
+        &mut FileTransport::default(),
+    )
+    .unwrap();
+    let text = std::str::from_utf8(&actual).unwrap();
+    let sync_begin = text.find("\x1b[?2026h").unwrap();
+    let sync_end = text.rfind("\x1b[?2026l").unwrap();
+    let deletion = text.find("a=d,d=I,i=41").unwrap();
+    let placements = text
+        .match_indices("a=p,i=42")
+        .map(|(at, _)| at)
+        .collect::<Vec<_>>();
+    assert!(deletion > sync_begin && deletion < sync_end);
+    assert!(placements.len() >= 2);
+    assert!(placements
+        .iter()
+        .all(|placement| *placement > sync_begin && *placement < sync_end));
+}
+
+#[test]
+fn deferred_output_propagates_write_failure() {
+    struct Broken;
+    impl std::io::Write for Broken {
+        fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
+            Err(std::io::ErrorKind::BrokenPipe.into())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+    assert_eq!(
+        write_composed_frame(Broken, b"text", &output(32), &mut FileTransport::default())
+            .unwrap_err()
+            .kind(),
+        std::io::ErrorKind::BrokenPipe
+    );
+}
+
+#[cfg(unix)]
+mod local_files {
+    use super::*;
+    use base64::Engine as _;
+    use std::path::PathBuf;
+    struct Scratch(PathBuf);
+    impl Scratch {
+        fn new() -> Self {
+            static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+            let id = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            let path = PathBuf::from(format!(
+                "/var/tmp/herdr-output-test-{}-{id}",
+                std::process::id()
+            ));
+            std::fs::create_dir(&path).unwrap();
+            Self(path)
+        }
+    }
+    impl Drop for Scratch {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_dir_all(&self.0);
+        }
+    }
+    fn command_path(bytes: &[u8], header: &str) -> PathBuf {
+        let text = std::str::from_utf8(bytes).unwrap();
+        let command = text
+            .split("\x1b_G")
+            .find(|part| part.starts_with(header))
+            .unwrap();
+        let payload = command
+            .split_once(';')
+            .unwrap()
+            .1
+            .split("\x1b\\")
+            .next()
+            .unwrap();
+        PathBuf::from(
+            String::from_utf8(
+                base64::engine::general_purpose::STANDARD
+                    .decode(payload)
+                    .unwrap(),
+            )
+            .unwrap(),
+        )
+    }
+    #[test]
+    fn files_begin_only_after_probe_consumption_and_keep_order() {
+        for format in [24, 32, 100] {
+            let root = Scratch::new();
+            let mut files = FileTransport::for_test(root.0.clone());
+            let graphics = output(format);
+            let mut first = Vec::new();
+            write_composed_frame(&mut first, b"text", &graphics, &mut files).unwrap();
+            let first_text = std::str::from_utf8(&first).unwrap();
+            assert!(first_text.contains("a=q,t=t"));
+            assert!(first_text.contains("a=t,t=d"));
+            let probe = command_path(&first, "a=q,");
+            assert_eq!(std::fs::read(&probe).unwrap().len(), 4);
+            let mut waiting = Vec::new();
+            write_composed_frame(&mut waiting, b"text", &graphics, &mut files).unwrap();
+            assert!(!std::str::from_utf8(&waiting).unwrap().contains("a=q,"));
+            assert!(std::str::from_utf8(&waiting).unwrap().contains("a=t,t=d"));
+            std::fs::remove_file(&probe).unwrap();
+            let mut ready = Vec::new();
+            write_composed_frame(&mut ready, b"text", &graphics, &mut files).unwrap();
+            let text = std::str::from_utf8(&ready).unwrap();
+            assert!(text.contains(&format!("before\x1b_Ga=t,t=t,f={format}")));
+            assert!(text.contains("\x1b\\after"));
+            assert!(!text.contains("AQIDBA=="));
+            let image = command_path(&ready, "a=t,");
+            assert_eq!(std::fs::read(&image).unwrap(), [1, 2, 3, 4]);
+            drop(files);
+            assert!(!image.exists());
+        }
+    }
+    #[test]
+    fn unsupported_format_and_file_creation_failure_remain_inline() {
+        let root = Scratch::new();
+        let mut files = FileTransport::for_test(root.0.clone());
+        let mut unsupported = Vec::new();
+        write_composed_frame(&mut unsupported, b"", &output(999), &mut files).unwrap();
+        assert!(std::str::from_utf8(&unsupported)
+            .unwrap()
+            .contains("a=t,t=d,f=999"));
+        assert_eq!(std::fs::read_dir(&root.0).unwrap().count(), 0);
+        let mut broken = FileTransport::for_test(root.0.join("absent"));
+        let mut actual = Vec::new();
+        write_composed_frame(&mut actual, b"", &output(32), &mut broken).unwrap();
+        let mut expected = Vec::new();
+        write_composed_frame(
+            &mut expected,
+            b"",
+            &output(32),
+            &mut FileTransport::default(),
+        )
+        .unwrap();
+        assert_eq!(actual, expected);
+    }
+}
diff --git a/src/client/handshake.rs b/src/client/handshake.rs
index 69e2f52f..f25cc1b0 100644
--- a/src/client/handshake.rs
+++ b/src/client/handshake.rs
@@ -69,6 +69,17 @@ pub(super) fn direct_graphics_profile_values(
     supported && !blocked_transport && terminals
 }
 
+/// Server-owned files require a shared filesystem; the host terminal profile alone
+/// cannot establish that for a saved SSH endpoint.
+fn direct_graphics_capability(
+    local_transport: bool,
+    exact_cell_size: bool,
+    cell_size: (u32, u32),
+    profile_allowed: bool,
+) -> bool {
+    local_transport && exact_cell_size && cell_size.0 > 0 && cell_size.1 > 0 && profile_allowed
+}
+
 #[cfg(unix)]
 fn direct_graphics_profile_allowed() -> bool {
     let term_program = std::env::var("TERM_PROGRAM").unwrap_or_default();
@@ -138,6 +149,7 @@ pub(crate) fn probe_endpoint_negotiation(
         false,
         false,
         false,
+        false, // This probe uses an SSH bridge, not a shared filesystem.
     )
     .map_err(io::Error::other)?;
     Ok(super::endpoint::EndpointNegotiation::new(
@@ -150,7 +162,8 @@ pub(crate) fn probe_endpoint_negotiation(
 ///
 /// Direct terminal clients retain the same-install private protocol. Client-owned
 /// shells use the stable endpoint generation and negotiate whole codecs without
-/// comparing Herdr build versions.
+/// comparing Herdr build versions. `local_transport` means the endpoint shares the
+/// client's filesystem, not merely that its bridge exposes a local socket.
 pub(super) fn do_handshake(
     stream: &mut LocalStream,
     cols: u16,
@@ -162,6 +175,7 @@ pub(super) fn do_handshake(
     endpoint_keybindings: bool,
     mouse_capture: bool,
     surface_active: bool,
+    local_transport: bool,
 ) -> Result<HandshakeResult, ClientError> {
     stream
         .set_nonblocking(false)
@@ -175,10 +189,12 @@ pub(super) fn do_handshake(
             cell_height_px,
             surface_size,
             pixel_mouse: exact_cell_size && cfg!(unix),
-            direct_graphics: exact_cell_size
-                && cell_width_px > 0
-                && cell_height_px > 0
-                && direct_graphics_profile_allowed(),
+            direct_graphics: direct_graphics_capability(
+                local_transport,
+                exact_cell_size,
+                (cell_width_px, cell_height_px),
+                direct_graphics_profile_allowed(),
+            ),
             endpoint_keybindings,
             mouse_capture,
             surface_active,
@@ -295,3 +311,20 @@ pub(super) fn do_handshake(
         ))),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn direct_graphics_requires_local_transport_even_on_supported_host_terminal() {
+        let supported = direct_graphics_profile_values("ghostty", "", false, false, true);
+        assert!(supported);
+        assert!(direct_graphics_capability(true, true, (8, 16), supported));
+        assert!(!direct_graphics_capability(false, true, (8, 16), supported));
+        assert!(!direct_graphics_capability(true, false, (8, 16), supported));
+        assert!(!direct_graphics_capability(true, true, (0, 16), supported));
+        assert!(!direct_graphics_capability(true, true, (8, 0), supported));
+        assert!(!direct_graphics_capability(true, true, (8, 16), false));
+    }
+}
diff --git a/src/client/image_files.rs b/src/client/image_files.rs
new file mode 100644
index 00000000..cc854c85
--- /dev/null
+++ b/src/client/image_files.rs
@@ -0,0 +1,138 @@
+//! Client-owned Kitty `t=t` files, prepared only at the final output boundary.
+//!
+//! A returned path is published: it must stay readable until the terminal unlinks
+//! it, or the client shuts down. Never use `cleanup` to cancel a frame. This ledger
+//! does not encode pixels or accept paths from a server. A silent raw-file query
+//! gates conversion for the local Ghostty allowlist; an explicit kill switch
+//! always wins. Local writes alone do not prove support. Non-consumption disables
+//! new files for this client, without deleting files the terminal may still open.
+
+use std::path::PathBuf;
+
+#[derive(Default)]
+pub(crate) struct FileTransport {
+    #[cfg(unix)]
+    inner: Option<crate::platform::unix_image_files::Ledger>,
+}
+
+impl FileTransport {
+    #[cfg(all(test, unix))]
+    pub(crate) fn for_test(root: PathBuf) -> Self {
+        Self {
+            inner: Some(crate::platform::unix_image_files::Ledger::for_test(root)),
+        }
+    }
+
+    pub(crate) fn from_environment() -> Self {
+        Self {
+            #[cfg(unix)]
+            inner: environment_allows().then(crate::platform::unix_image_files::Ledger::new),
+        }
+    }
+
+    /// Once per client, return a private 1x1 RGBA query payload. The output
+    /// boundary must emit this as `a=q,t=t,f=32,s=1,v=1,q=2` before inline image
+    /// output. No real upload is converted until Ghostty consumes this file.
+    pub(crate) fn probe(&mut self) -> Option<PathBuf> {
+        #[cfg(unix)]
+        {
+            self.inner.as_mut()?.probe()
+        }
+        #[cfg(not(unix))]
+        {
+            None
+        }
+    }
+
+    /// Returns a complete, closed file containing exactly `data`, or requests
+    /// inline fallback. The caller must preserve output ordering and encode the
+    /// returned local path as Kitty's base64 filename payload with `t=t`.
+    pub(crate) fn prepare(&mut self, data: &[u8]) -> Option<PathBuf> {
+        #[cfg(unix)]
+        {
+            self.inner.as_mut()?.prepare(data)
+        }
+        #[cfg(not(unix))]
+        {
+            let _ = data;
+            None
+        }
+    }
+}
+
+#[cfg(unix)]
+use std::ffi::OsString;
+#[cfg(unix)]
+use std::io::IsTerminal;
+
+#[cfg(unix)]
+fn environment_allows() -> bool {
+    allowed(
+        std::io::stdin().is_terminal(),
+        std::io::stdout().is_terminal(),
+        |key| std::env::var_os(key),
+    )
+}
+
+#[cfg(unix)]
+fn allowed(stdin_tty: bool, stdout_tty: bool, env: impl Fn(&str) -> Option<OsString>) -> bool {
+    stdin_tty
+        && stdout_tty
+        && env("TERM_PROGRAM").as_deref() == Some(std::ffi::OsStr::new("ghostty"))
+        && [
+            "SSH_CONNECTION",
+            "SSH_TTY",
+            "SSH_CLIENT",
+            "TMUX",
+            "STY",
+            "HERDR_PANE_ID",
+            "HERDR_PANE_RUNTIME_ID",
+        ]
+        .iter()
+        .all(|key| env(key).is_none())
+    // Named-session selection (HERDR_SESSION) and HERDR_REMOTE_KEYBINDINGS
+    // deliberately do not affect client locality.
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    #[test]
+    fn conservative_gate_without_process_environment_mutation() {
+        let base = |key: &str| match key {
+            "TERM_PROGRAM" => Some(OsString::from("ghostty")),
+            "HERDR_SESSION" => Some(OsString::from("named-session")),
+            "HERDR_REMOTE_KEYBINDINGS" => Some(OsString::from("1")),
+            _ => None,
+        };
+        assert!(allowed(true, true, base));
+        assert!(!allowed(false, true, base));
+        assert!(!allowed(true, false, base));
+        for key in [
+            "SSH_CONNECTION",
+            "SSH_TTY",
+            "SSH_CLIENT",
+            "TMUX",
+            "STY",
+            "HERDR_PANE_ID",
+            "HERDR_PANE_RUNTIME_ID",
+        ] {
+            assert!(!allowed(true, true, |name| if name == key {
+                Some(OsString::new())
+            } else {
+                base(name)
+            }));
+        }
+        assert!(!allowed(true, true, |key| if key == "TERM_PROGRAM" {
+            Some(OsString::from("kitty"))
+        } else {
+            base(key)
+        }));
+    }
+    #[test]
+    fn default_is_disabled() {
+        let mut transport = FileTransport::default();
+        assert!(transport.probe().is_none());
+        assert!(transport.prepare(b"inline").is_none());
+    }
+}
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 9ad51a36..1a0f8dd5 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -24,7 +24,10 @@ mod endpoint_commands;
 mod errors;
 mod events;
 mod frame_output;
+#[cfg(test)]
+mod frame_output_tests;
 mod handshake;
+mod image_files;
 mod input;
 mod loop_config;
 mod notifications;
@@ -48,6 +51,8 @@ use events::ClientLoopEvent;
 use loop_config::ClientLoopConfig;
 use shell_runtime::*;
 use state::ClientState;
+#[cfg(unix)]
+use state::RetiredDirectGraphicsMatch;
 use transport::*;
 
 #[cfg(test)]
@@ -135,7 +140,7 @@ use tracing::{debug, info, warn};
 
 use crate::ipc::LocalStream;
 use crate::protocol::render_ansi;
-use crate::protocol::{self, ClientMessage, FrameData, ServerMessage, MAX_GRAPHICS_FRAME_SIZE};
+use crate::protocol::{self, ClientMessage, ServerMessage, MAX_GRAPHICS_FRAME_SIZE};
 #[cfg(test)]
 use crate::protocol::{AttachScrollDirection, AttachScrollSource, NotifyKind};
 use crate::server::socket_paths::client_socket_path;
@@ -244,6 +249,7 @@ fn run_client_with_mode(
                 endpoint_keybindings,
                 loop_config.mouse_capture_active,
                 true,
+                !is_remote_client_process(),
             )
             .map_err(|error| io::Error::other(error.to_string()))?;
             if federated
@@ -364,6 +370,31 @@ fn run_client_with_mode(
     Ok(())
 }
 
+// This guards server-supplied paths only. Client-owned temporary files generated
+// from received graphics bytes remain usable for remote endpoints.
+fn server_graphics_files_allowed(
+    endpoint_id: &endpoint::ClientEndpointId,
+    remote_client_process: bool,
+) -> bool {
+    endpoint_id.is_local() && !remote_client_process
+}
+
+#[cfg(unix)]
+fn graphics_owner_is_active(
+    state: &ClientState,
+    endpoints: &endpoint::EndpointRegistry,
+    owner: &endpoint::ClientEndpointId,
+) -> bool {
+    endpoints.active_id() == owner
+        && endpoints
+            .connection(owner)
+            .is_some_and(|connection| connection.surface_active)
+        && state
+            .shell
+            .as_ref()
+            .is_some_and(|shell| shell.endpoint_is_active(owner))
+}
+
 /// The main client event loop.
 ///
 /// Uses a threaded architecture:
@@ -392,6 +423,7 @@ async fn run_client_loop(
 
     let mut state = ClientState {
         blit_encoder: render_ansi::BlitEncoder::new(),
+        image_files: image_files::FileTransport::from_environment(),
         mouse_capture_active: config.mouse_capture_active,
         endpoint_mouse_capture_requested: false,
         endpoint_sgr_pixels_requested: false,
@@ -410,7 +442,10 @@ async fn run_client_loop(
         #[cfg(unix)]
         direct_graphics_response: Arc::new(Mutex::new(direct_graphics::ResponseMatcher::default())),
         #[cfg(unix)]
-        retired_direct_graphics: None,
+        retired_direct_graphics: HashMap::new(),
+        #[cfg(unix)]
+        disabled_native_graphics: Default::default(),
+        pending_native_cleanup: Vec::new(),
         #[cfg(unix)]
         pending_surface_graphics: HashMap::new(),
         attach_escape,
@@ -690,10 +725,10 @@ async fn run_client_loop(
                     let frozen = state.presentation_frozen;
                     state.presentation_frozen = false;
                     state.present_graphics(&cleanup);
+                    state.presentation_frozen = frozen;
                     if let Some(frame) = frame {
-                        state.present_frame(frame);
+                        state.present_frozen_chrome(frame);
                     }
-                    state.presentation_frozen = frozen;
                 }
             }
         }
@@ -937,42 +972,70 @@ async fn run_client_loop(
                 let pending_key = state
                     .pending_surface_graphics
                     .keys()
-                    .find(|(_, transfer_id, image_id)| {
+                    .find(|(_, _, transfer_id, image_id)| {
                         *transfer_id == response.transfer_id && *image_id == response.image_id
                     })
                     .cloned();
-                let owner = pending_key
-                    .as_ref()
-                    .map(|(endpoint_id, _, _)| endpoint_id.clone());
-                let composed = pending_key
-                    .and_then(|key| {
-                        state
-                            .pending_surface_graphics
-                            .remove(&key)
-                            .map(|asset| (key, asset))
-                    })
-                    .filter(|_| response.success)
-                    .and_then(|((endpoint_id, _, _), asset)| {
-                        if write_stream.active_id() != &endpoint_id {
+                let pending = pending_key.and_then(|key| {
+                    state
+                        .pending_surface_graphics
+                        .remove(&key)
+                        .map(|asset| (key, asset))
+                });
+                let Some(((owner, _, _, _), asset)) = pending else {
+                    // Raw/non-surface transfers retain their existing response handling.
+                    continue;
+                };
+                let eligible = response.success
+                    && graphics_owner_is_active(&state, &write_stream, &owner)
+                    && !state.presentation_frozen
+                    && state.shell.as_ref().is_some_and(|shell| {
+                        shell.accepts_direct_graphics_asset(&asset, response.image_id)
+                    });
+                let size = state.reported_size;
+                let prepared = eligible
+                    .then(|| {
+                        let shell = state.shell.as_mut()?;
+                        let checkpoint = shell.direct_graphics_checkpoint();
+                        if !shell.trust_direct_graphics_asset(&asset, response.image_id) {
+                            shell.restore_direct_graphics_checkpoint(checkpoint);
                             return None;
                         }
-                        let shell = state.shell.as_mut()?;
-                        shell
-                            .trust_direct_graphics_asset(&asset, response.image_id)
-                            .then(|| shell.compose(state.reported_size.0, state.reported_size.1))
-                            .flatten()
-                    });
+                        match shell.compose(size.0, size.1) {
+                            Some(frame) => Some((frame, checkpoint)),
+                            None => {
+                                shell.restore_direct_graphics_checkpoint(checkpoint);
+                                None
+                            }
+                        }
+                    })
+                    .flatten();
+                let accepted = if let Some((frame, checkpoint)) = prepared {
+                    if state.try_present_frame(frame) {
+                        true
+                    } else {
+                        state
+                            .shell
+                            .as_mut()
+                            .expect("prepared direct graphics requires a shell")
+                            .restore_direct_graphics_checkpoint(checkpoint);
+                        false
+                    }
+                } else {
+                    false
+                };
+                if !accepted {
+                    // The upload may have reached the terminal even when its response, owner, or
+                    // composed swap cannot be accepted. Never leave that unowned image resident.
+                    // Restoring the checkpoint also restores old-bank stale-image bookkeeping.
+                    state.queue_native_image_cleanup(response.image_id);
+                }
                 let message = ClientMessage::GraphicsTransmissionResult {
                     transfer_id: response.transfer_id,
                     image_id: response.image_id,
-                    success: response.success,
+                    success: accepted,
                 };
-                if let Some(owner) = owner {
-                    write_stream.send_to(&owner, &message);
-                }
-                if let Some(frame) = composed {
-                    state.present_frame(frame);
-                }
+                write_stream.send_to(&owner, &message);
             }
             #[cfg(unix)]
             ClientLoopEvent::PixelMouse(data, geometry) => {
@@ -1228,6 +1291,8 @@ async fn run_client_loop(
                         shell.compose(state.reported_size.0, state.reported_size.1)
                     });
                     let reader_quit = writer.stop_handle();
+                    #[cfg(unix)]
+                    state.start_endpoint_graphics_generation(&endpoint_id, generation);
                     write_stream.insert(
                         endpoint_id.clone(),
                         writer,
@@ -1287,6 +1352,30 @@ async fn run_client_loop(
                     continue;
                 }
                 write_stream.received(&endpoint_id, generation, now);
+                // Retirements belong to the exact live connection, even after deactivation.
+                // They must not be dropped with frozen/inactive presentation effects.
+                if matches!(
+                    message.as_ref(),
+                    ServerMessage::GraphicsTransmissionRetired { .. }
+                ) {
+                    #[cfg(unix)]
+                    if let ServerMessage::GraphicsTransmissionRetired {
+                        transfer_id,
+                        image_id,
+                    } = message.as_ref()
+                    {
+                        let owner_active =
+                            graphics_owner_is_active(&state, &write_stream, &endpoint_id);
+                        state.receive_graphics_retirement(
+                            &endpoint_id,
+                            generation,
+                            *transfer_id,
+                            *image_id,
+                            owner_active,
+                        );
+                    }
+                    continue;
+                }
                 let endpoint_active = write_stream.active_id() == &endpoint_id
                     && write_stream
                         .connection(&endpoint_id)
@@ -1463,32 +1552,73 @@ async fn run_client_loop(
                         control,
                         surface_asset,
                     } => {
+                        // Never interpret an SSH server's filesystem path on this host,
+                        // including legacy peers that send files without negotiation.
+                        if !server_graphics_files_allowed(&endpoint_id, is_remote_client_process())
+                        {
+                            write_stream.send_to(
+                                &endpoint_id,
+                                &ClientMessage::GraphicsTransmissionResult {
+                                    transfer_id,
+                                    image_id,
+                                    success: false,
+                                },
+                            );
+                            continue;
+                        }
                         #[cfg(unix)]
                         {
-                            if state.retired_direct_graphics.take()
-                                == Some((endpoint_id.clone(), transfer_id, image_id))
-                            {
+                            let retirement = state.match_retired_direct_graphics(
+                                &endpoint_id,
+                                generation,
+                                transfer_id,
+                                image_id,
+                            );
+                            if retirement == RetiredDirectGraphicsMatch::Exact {
                                 continue;
                             }
                             let surface_asset_valid = match (state.shell.as_ref(), &surface_asset) {
                                 (Some(shell), Some(asset)) => {
-                                    crate::kitty_graphics::surface::host_image_id(
-                                        shell.graphics_scope(),
-                                        asset,
-                                    ) == image_id
+                                    shell.accepts_direct_graphics_asset(asset, image_id)
                                 }
                                 (None, None) => true,
                                 _ => false,
                             };
-                            let valid = state.kitty_graphics_enabled
+                            let native = surface_asset.as_ref().is_some_and(|asset| {
+                                matches!(
+                                    asset.source,
+                                    crate::protocol::SurfaceGraphicsSource::Terminal { .. }
+                                )
+                            });
+                            let native_valid = !native
+                                || surface_asset.as_ref().is_some_and(|asset| {
+                                    !state.presentation_frozen
+                                        && state.disabled_native_graphics.get(&endpoint_id)
+                                            != Some(&generation)
+                                        && graphics_owner_is_active(
+                                            &state,
+                                            &write_stream,
+                                            &endpoint_id,
+                                        )
+                                        && leading.is_empty()
+                                        && asset.format
+                                            == crate::protocol::SurfaceGraphicsFormat::Rgba
+                                        && expected_len == asset.data_len
+                                        && control
+                                            == format!(
+                                                "a=t,f=32,s={},v={},i={image_id},q=0",
+                                                asset.image_width, asset.image_height
+                                            )
+                                });
+                            let valid = retirement != RetiredDirectGraphicsMatch::Saturated
+                                && state.kitty_graphics_enabled
+                                && native_valid
                                 && surface_asset_valid
                                 && usize::try_from(expected_len).ok().is_some_and(|len| {
-                                    crate::pane_graphics_files::validate_direct_source(
-                                        std::path::Path::new(&path),
-                                        len,
-                                    )
-                                    .is_ok()
-                                        && direct_graphics::valid_control(&control, image_id, len)
+                                    let path = std::path::Path::new(&path);
+                                    let valid_source = crate::pane_graphics_files::validate_direct_source(path, len).is_ok()
+                                        || (native && crate::pane_graphics_files::validate_native_source(path, len).is_ok());
+                                    valid_source && direct_graphics::valid_control(&control, image_id, len)
                                 })
                                 && state
                                     .direct_graphics_response
@@ -1503,8 +1633,9 @@ async fn run_client_loop(
                                     &path,
                                 );
                                 let mut stdout = io::stdout();
-                                let written = stdout
-                                    .write_all(&command)
+                                let written = state
+                                    .flush_native_cleanup(&mut stdout)
+                                    .and_then(|()| stdout.write_all(&command))
                                     .and_then(|()| stdout.flush())
                                     .is_ok();
                                 if written {
@@ -1517,7 +1648,7 @@ async fn run_client_loop(
                             if sent {
                                 if let Some(asset) = surface_asset {
                                     state.pending_surface_graphics.insert(
-                                        (endpoint_id.clone(), transfer_id, image_id),
+                                        (endpoint_id.clone(), generation, transfer_id, image_id),
                                         asset,
                                     );
                                 }
@@ -1528,12 +1659,11 @@ async fn run_client_loop(
                                     transfer_id,
                                     image_id,
                                 };
-                                if let Err(err) = write_to_server(&mut write_stream, &started) {
-                                    return Err(ClientError::ConnectionLost(err));
-                                }
+                                write_stream.send_to(&endpoint_id, &started);
                             } else {
                                 state.pending_surface_graphics.remove(&(
                                     endpoint_id.clone(),
+                                    generation,
                                     transfer_id,
                                     image_id,
                                 ));
@@ -1549,9 +1679,7 @@ async fn run_client_loop(
                                     image_id,
                                     success: false,
                                 };
-                                if let Err(err) = write_to_server(&mut write_stream, &result) {
-                                    return Err(ClientError::ConnectionLost(err));
-                                }
+                                write_stream.send_to(&endpoint_id, &result);
                             }
                         }
                         #[cfg(not(unix))]
@@ -1565,33 +1693,8 @@ async fn run_client_loop(
                             surface_asset,
                         );
                     }
-                    ServerMessage::GraphicsTransmissionRetired {
-                        transfer_id,
-                        image_id,
-                    } => {
-                        #[cfg(unix)]
-                        {
-                            state.retired_direct_graphics =
-                                Some((endpoint_id.clone(), transfer_id, image_id));
-                            state.pending_surface_graphics.remove(&(
-                                endpoint_id.clone(),
-                                transfer_id,
-                                image_id,
-                            ));
-                            let cleanup = state.shell.as_mut().map_or_else(Vec::new, |shell| {
-                                shell.retire_direct_graphics_image(image_id);
-                                shell
-                                    .compose(state.reported_size.0, state.reported_size.1)
-                                    .map(|frame| frame.graphics)
-                                    .unwrap_or_else(|| shell.take_pending_graphics_cleanup())
-                            });
-                            state.present_graphics(&cleanup);
-                            if let Ok(mut matcher) = state.direct_graphics_response.lock() {
-                                matcher.retire(transfer_id);
-                            }
-                        }
-                        #[cfg(not(unix))]
-                        let _ = (transfer_id, image_id);
+                    ServerMessage::GraphicsTransmissionRetired { .. } => {
+                        unreachable!("retirements are handled before presentation gating")
                     }
                     ServerMessage::ServerShutdown { reason } => {
                         if !federated && endpoint_id.is_local() {
diff --git a/src/client/shell/composition.rs b/src/client/shell/composition.rs
index 9b710dd5..b7ab4a96 100644
--- a/src/client/shell/composition.rs
+++ b/src/client/shell/composition.rs
@@ -147,7 +147,11 @@ impl ClientShellState {
         FrameData::from_ratatui_buffer_with_hyperlinks(&buffer, None, &[])
     }
 
-    pub(crate) fn compose(&mut self, cols: u16, rows: u16) -> Option<FrameData> {
+    pub(crate) fn compose(
+        &mut self,
+        cols: u16,
+        rows: u16,
+    ) -> Option<crate::client::frame_output::ComposedFrame> {
         self.last_composed_at = Some(std::time::Instant::now());
         self.selection_repaint_deadline = None;
         if self.last_composed_size != Some((cols, rows)) && self.mode == ClientShellMode::Navigate {
@@ -167,7 +171,7 @@ impl ClientShellState {
                     && self.navigation_target_valid(&pending.target)
             });
         if self.snapshot.is_none() || self.pane_surface.is_none() {
-            return Some(self.compose_unavailable(cols, rows));
+            return Some(self.compose_unavailable(cols, rows).into());
         }
         let snapshot = self.snapshot.as_deref()?;
         // Do not compose a retained surface while waiting for its matching snapshot or
@@ -716,8 +720,8 @@ impl ClientShellState {
             self.hits.pane_splits.clear();
             self.hits.popup = None;
         }
-        self.compose_graphics(&mut frame, layout, &occlusion);
-        Some(frame)
+        let graphics = self.compose_graphics(layout, &occlusion);
+        Some(crate::client::frame_output::ComposedFrame { frame, graphics })
     }
 }
 
diff --git a/src/client/shell/graphics.rs b/src/client/shell/graphics.rs
index a7ae81e3..35e61dbb 100644
--- a/src/client/shell/graphics.rs
+++ b/src/client/shell/graphics.rs
@@ -1,12 +1,37 @@
 use super::*;
 
+#[cfg(unix)]
+pub(crate) struct ClientGraphicsCheckpoint(crate::kitty_graphics::surface::ClientState);
+
 impl ClientShellState {
-    #[cfg(unix)]
+    #[cfg(all(test, unix))]
     pub(crate) fn graphics_scope(&self) -> &str {
         self.graphics.scope()
     }
 
     #[cfg(unix)]
+    pub(crate) fn accepts_direct_graphics_asset(
+        &self,
+        key: &crate::protocol::SurfaceGraphicsAssetKey,
+        image_id: u32,
+    ) -> bool {
+        self.graphics.accepts_direct_asset(key, image_id)
+    }
+
+    #[cfg(unix)]
+    pub(crate) fn direct_graphics_checkpoint(&self) -> ClientGraphicsCheckpoint {
+        ClientGraphicsCheckpoint(self.graphics.clone())
+    }
+
+    #[cfg(unix)]
+    pub(crate) fn restore_direct_graphics_checkpoint(
+        &mut self,
+        checkpoint: ClientGraphicsCheckpoint,
+    ) {
+        self.graphics = checkpoint.0;
+    }
+
+    #[cfg(unix)]
     pub(crate) fn trust_direct_graphics_asset(
         &mut self,
         key: &crate::protocol::SurfaceGraphicsAssetKey,
@@ -33,10 +58,9 @@ impl ClientShellState {
 
     pub(super) fn compose_graphics(
         &mut self,
-        frame: &mut FrameData,
         layout: ClientShellLayout,
         occlusion: &crate::kitty_graphics::surface::Occlusion,
-    ) {
+    ) -> crate::kitty_graphics::GraphicsOutput {
         let visibility = if self.endpoint_error.is_some() {
             crate::kitty_graphics::surface::Visibility::Hidden
         } else if self.hits.popup.is_some() {
@@ -49,12 +73,12 @@ impl ClientShellState {
             .popup
             .as_ref()
             .map(|popup| (popup.inner_rect.x, popup.inner_rect.y));
-        frame.graphics = self.graphics.encode(
+        self.graphics.encode_output(
             visibility,
             (layout.pane_surface.x, layout.pane_surface.y),
             popup_origin,
             self.graphics_cell_size,
             occlusion,
-        );
+        )
     }
 }
diff --git a/src/client/shell/state.rs b/src/client/shell/state.rs
index 0d6937e0..c170c9b0 100644
--- a/src/client/shell/state.rs
+++ b/src/client/shell/state.rs
@@ -1015,7 +1015,7 @@ impl ClientShellState {
             pane_surface_generation: None,
             pane_surface: None,
             pending_pane_surface: None,
-            graphics: crate::kitty_graphics::surface::ClientState::default(),
+            graphics: crate::kitty_graphics::surface::ClientState::new(),
             graphics_cell_size: crate::kitty_graphics::HostCellSize {
                 width_px: 1,
                 height_px: 1,
diff --git a/src/client/shell/tests/copy.rs b/src/client/shell/tests/copy.rs
index 112417fc..67e36f23 100644
--- a/src/client/shell/tests/copy.rs
+++ b/src/client/shell/tests/copy.rs
@@ -1563,7 +1563,7 @@ fn navigator_narrow_layout_and_long_search_stay_inside_the_popup() {
         navigator.query = "界".repeat(100).as_str().into();
         let frame = state.compose(width, height).expect("navigator frame");
         let popup = state.hits.navigator_popup;
-        let cursor = frame.cursor.expect("search cursor");
+        let cursor = frame.cursor.as_ref().expect("search cursor");
         assert!(super::super::contains(popup, (cursor.x, cursor.y)));
         assert!(state.hits.navigator_rows.is_empty());
         assert!(popup.right() <= width && popup.bottom() <= height);
diff --git a/src/client/shell/tests/graphics.rs b/src/client/shell/tests/graphics.rs
index 240e031c..e54301a4 100644
--- a/src/client/shell/tests/graphics.rs
+++ b/src/client/shell/tests/graphics.rs
@@ -63,6 +63,48 @@ fn add_main_image(
     surface.graphics.placements.push(placement);
 }
 
+#[cfg(unix)]
+#[test]
+fn failed_direct_ack_composition_restores_graphics_for_inline_retry() {
+    use crate::kitty_graphics::surface::host_image_id;
+
+    let mut state = ClientShellState::new(ClientShellConfig::from_config(&Config::default()));
+    state.set_snapshot(Box::new(snapshot()));
+    let mut pane = surface();
+    let (mut asset, mut placement) = image(
+        SurfaceGraphicsTarget::Pane {
+            pane_id: "pane_1".into(),
+        },
+        0,
+        0,
+        71,
+    );
+    asset.key.source = SurfaceGraphicsSource::PaneLayer {
+        pane_id: "pane_1".into(),
+        layer_id: "direct-rollback".into(),
+    };
+    placement.asset = asset.key.clone();
+    let key = asset.key.clone();
+    pane.graphics.assets.push(asset);
+    pane.graphics.placements.push(placement);
+    state.set_pane_surface(pane);
+
+    // Model an ACK arriving while projection pairing makes composition unavailable.
+    state.pending_pane_surface = Some(surface());
+    let image_id = host_image_id(state.graphics_scope(), &key);
+    let graphics_before = format!("{:?}", state.graphics);
+    let checkpoint = state.direct_graphics_checkpoint();
+    assert!(state.trust_direct_graphics_asset(&key, image_id));
+    assert!(state.compose(106, 20).is_none());
+    state.restore_direct_graphics_checkpoint(checkpoint);
+    assert_eq!(format!("{:?}", state.graphics), graphics_before);
+
+    // Once pairing is available, fallback still owns the asset and uploads it normally.
+    state.pending_pane_surface = None;
+    let retry = state.compose(106, 20).expect("fallback frame");
+    assert!(String::from_utf8_lossy(&retry.graphics.into_inline_bytes()).contains("a=t"));
+}
+
 fn is_placed(bytes: &[u8], point: (u16, u16)) -> bool {
     String::from_utf8_lossy(bytes).contains(&format!("\x1b[{};{}H", point.1 + 1, point.0 + 1))
 }
@@ -85,11 +127,11 @@ fn assert_graphics_cover(state: &mut ClientShellState, covered: Rect, cols: u16,
     state.set_pane_surface(surface);
     let frame = state.compose(cols, rows).unwrap();
     assert!(
-        is_placed(&frame.graphics, outside),
+        is_placed(&frame.graphics.clone().into_inline_bytes(), outside),
         "outside={outside:?} cover={covered:?}"
     );
     assert!(
-        !is_placed(&frame.graphics, inside),
+        !is_placed(&frame.graphics.clone().into_inline_bytes(), inside),
         "inside={inside:?} cover={covered:?}"
     );
 }
@@ -327,18 +369,38 @@ fn every_dialog_and_menu_occludes_its_panel_not_the_whole_screen() {
         state.compose(106, 40).unwrap();
         state.overlay = Some(overlay);
         let frame = state.compose(106, 40).unwrap();
-        assert!(is_placed(&frame.graphics, outside), "{:?}", state.overlay);
+        assert!(
+            is_placed(&frame.graphics.clone().into_inline_bytes(), outside),
+            "{:?}",
+            state.overlay
+        );
         if let Some(border) = border {
-            assert!(!is_placed(&frame.graphics, border), "{:?}", state.overlay);
-            assert!(String::from_utf8_lossy(&frame.graphics).contains("a=d,d=i"));
+            assert!(
+                !is_placed(&frame.graphics.clone().into_inline_bytes(), border),
+                "{:?}",
+                state.overlay
+            );
+            assert!(
+                String::from_utf8_lossy(&frame.graphics.clone().into_inline_bytes())
+                    .contains("a=d,d=i")
+            );
         }
         state.overlay = None;
         let restored = state.compose(106, 40).unwrap();
-        assert!(is_placed(&restored.graphics, outside));
+        assert!(is_placed(
+            &restored.graphics.clone().into_inline_bytes(),
+            outside
+        ));
         if let Some(border) = border {
-            assert!(is_placed(&restored.graphics, border));
+            assert!(is_placed(
+                &restored.graphics.clone().into_inline_bytes(),
+                border
+            ));
         }
-        assert!(!String::from_utf8_lossy(&restored.graphics).contains("a=t"));
+        assert!(
+            !String::from_utf8_lossy(&restored.graphics.clone().into_inline_bytes())
+                .contains("a=t")
+        );
     }
 }
 
@@ -377,17 +439,26 @@ fn selection_copy_cursor_and_search_hide_only_the_highlighted_images() {
         });
     let frame = state.compose(106, 20).unwrap();
     for point in &points[..3] {
-        assert!(!is_placed(&frame.graphics, *point), "{point:?}");
+        assert!(
+            !is_placed(&frame.graphics.clone().into_inline_bytes(), *point),
+            "{point:?}"
+        );
     }
-    assert!(is_placed(&frame.graphics, points[3]));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        points[3]
+    ));
     state.mode = ClientShellMode::Terminal;
     state.copy_mode = None;
     state.selection = None;
     let frame = state.compose(106, 20).unwrap();
     for point in points {
-        assert!(is_placed(&frame.graphics, point));
+        assert!(is_placed(
+            &frame.graphics.clone().into_inline_bytes(),
+            point
+        ));
     }
-    assert!(!String::from_utf8_lossy(&frame.graphics).contains("a=t"));
+    assert!(!String::from_utf8_lossy(&frame.graphics.clone().into_inline_bytes()).contains("a=t"));
 }
 
 #[test]
@@ -400,15 +471,26 @@ fn mobile_switcher_still_hides_the_entire_underlying_surface() {
     add_main_image(&mut surface, layout, point, 1);
     state.set_pane_surface(surface);
     let frame = state.compose(40, 24).unwrap();
-    assert!(is_placed(&frame.graphics, point));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        point
+    ));
     state.mode = ClientShellMode::Navigate;
     let frame = state.compose(40, 24).unwrap();
-    assert!(!is_placed(&frame.graphics, point));
-    assert!(String::from_utf8_lossy(&frame.graphics).contains("a=d,d=i"));
+    assert!(!is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        point
+    ));
+    assert!(
+        String::from_utf8_lossy(&frame.graphics.clone().into_inline_bytes()).contains("a=d,d=i")
+    );
     state.mode = ClientShellMode::Terminal;
     let frame = state.compose(40, 24).unwrap();
-    assert!(is_placed(&frame.graphics, point));
-    assert!(!String::from_utf8_lossy(&frame.graphics).contains("a=t"));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        point
+    ));
+    assert!(!String::from_utf8_lossy(&frame.graphics.clone().into_inline_bytes()).contains("a=t"));
 }
 
 #[test]
@@ -437,15 +519,33 @@ fn popup_terminal_keeps_own_graphics_and_hides_only_background_overlap() {
     surface.graphics.placements.push(placement);
     state.set_pane_surface(surface);
     let frame = state.compose(106, 20).unwrap();
-    assert!(is_placed(&frame.graphics, outside));
-    assert!(!is_placed(&frame.graphics, border));
-    assert!(is_placed(&frame.graphics, inside));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        outside
+    ));
+    assert!(!is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        border
+    ));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        inside
+    ));
     state.overlay = Some(ClientShellOverlay::Onboarding);
     let frame = state.compose(106, 20).unwrap();
-    assert!(is_placed(&frame.graphics, outside));
-    assert!(!is_placed(&frame.graphics, inside));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        outside
+    ));
+    assert!(!is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        inside
+    ));
     state.overlay = None;
     let frame = state.compose(106, 20).unwrap();
-    assert!(is_placed(&frame.graphics, inside));
-    assert!(!String::from_utf8_lossy(&frame.graphics).contains("a=t"));
+    assert!(is_placed(
+        &frame.graphics.clone().into_inline_bytes(),
+        inside
+    ));
+    assert!(!String::from_utf8_lossy(&frame.graphics.clone().into_inline_bytes()).contains("a=t"));
 }
diff --git a/src/client/shell/tests/input.rs b/src/client/shell/tests/input.rs
index b55cf9a2..b54548f2 100644
--- a/src/client/shell/tests/input.rs
+++ b/src/client/shell/tests/input.rs
@@ -216,18 +216,24 @@ fn client_shell_graphics_follow_final_shell_origin_and_local_overlay_visibility(
     state.set_pane_surface(pane_surface);
 
     let visible = state.compose(106, 20).expect("visible graphics frame");
-    let visible = String::from_utf8_lossy(&visible.graphics);
+    let visible = visible.graphics.clone().into_inline_bytes();
+    let visible = String::from_utf8_lossy(&visible);
     assert!(visible.contains("a=t,t=d"));
     assert!(visible.contains("\u{1b}[2;27H"));
 
     state.overlay = Some(ClientShellOverlay::Onboarding);
     let uncovered = state.compose(106, 20).expect("overlay frame");
-    assert!(!String::from_utf8_lossy(&uncovered.graphics).contains("a=d"));
-    assert!(String::from_utf8_lossy(&uncovered.graphics).contains("a=p"));
+    assert!(
+        !String::from_utf8_lossy(&uncovered.graphics.clone().into_inline_bytes()).contains("a=d")
+    );
+    assert!(
+        String::from_utf8_lossy(&uncovered.graphics.clone().into_inline_bytes()).contains("a=p")
+    );
 
     state.overlay = None;
     let restored = state.compose(106, 20).expect("restored graphics frame");
-    let restored = String::from_utf8_lossy(&restored.graphics);
+    let restored = restored.graphics.clone().into_inline_bytes();
+    let restored = String::from_utf8_lossy(&restored);
     assert!(restored.contains("a=p"));
     assert!(!restored.contains("a=t,t=d"));
 }
diff --git a/src/client/shell/tests/text_editing.rs b/src/client/shell/tests/text_editing.rs
index b63ce9f6..118a606e 100644
--- a/src/client/shell/tests/text_editing.rs
+++ b/src/client/shell/tests/text_editing.rs
@@ -384,7 +384,7 @@ fn every_field_renders_long_unicode_across_resize_without_mutation() {
             for (width, height) in [(120, 40), (60, 20), (12, 6), (1, 1), (120, 40)] {
                 let before = editor(&mut state).clone();
                 if let Some(frame) = state.compose(width, height) {
-                    if let Some(cursor) = frame.cursor.filter(|cursor| cursor.visible) {
+                    if let Some(cursor) = frame.cursor.as_ref().filter(|cursor| cursor.visible) {
                         assert!(cursor.x < width && cursor.y < height, "field {field}");
                     }
                 }
diff --git a/src/client/shell_runtime.rs b/src/client/shell_runtime.rs
index 3843654f..d0eafdf6 100644
--- a/src/client/shell_runtime.rs
+++ b/src/client/shell_runtime.rs
@@ -385,14 +385,6 @@ pub(super) fn complete_endpoint_activation(
         completion,
         endpoint::ActivationCompletion::AwaitingPresentationSync { .. }
     ) {
-        #[cfg(unix)]
-        if let endpoint::ActivationCompletion::AwaitingPresentationSync { previous, endpoint } =
-            &completion
-        {
-            if previous != endpoint {
-                state.retire_endpoint_graphics(previous);
-            }
-        }
         // The coherent target frame can replace the frozen source now, but the registry keeps
         // pane input disabled until a second projection epoch has replayed host modes/effects.
         state.unfreeze_presentation();
@@ -511,7 +503,7 @@ pub(super) fn handle_endpoint_disconnect(
 ) -> bool {
     supervisors.disconnected(endpoint_id, generation, now);
     #[cfg(unix)]
-    state.retire_endpoint_graphics(endpoint_id);
+    state.retire_endpoint_graphics(endpoint_id, generation);
     if pending_activation
         .as_ref()
         .is_some_and(|pending| pending.involves_endpoint(endpoint_id))
@@ -572,7 +564,7 @@ pub(super) fn handle_endpoint_attention(
         now,
     );
     #[cfg(unix)]
-    state.retire_endpoint_graphics(endpoint_id);
+    state.retire_endpoint_graphics(endpoint_id, generation);
     if pending_activation
         .as_ref()
         .is_some_and(|pending| pending.involves_endpoint(endpoint_id))
@@ -681,7 +673,7 @@ pub(super) fn install_client_shell_snapshot(
 pub(super) fn finish_client_shell_input(
     state: &mut ClientState,
     outcome: shell::ClientShellInput,
-    frame: Option<FrameData>,
+    frame: Option<super::frame_output::ComposedFrame>,
     endpoints: &mut endpoint::EndpointRegistry,
     pending_activation: &mut Option<endpoint::PendingEndpointActivation>,
     endpoint_commands: &mut endpoint_commands::EndpointCommands,
diff --git a/src/client/state.rs b/src/client/state.rs
index 83b07429..e6b8abf3 100644
--- a/src/client/state.rs
+++ b/src/client/state.rs
@@ -1,9 +1,39 @@
 use super::*;
 
+#[cfg(unix)]
+const MAX_RETIRED_DIRECT_GRAPHICS: usize = 64;
+
+#[cfg(unix)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum RetiredDirectGraphicsMatch {
+    None,
+    Exact,
+    Saturated,
+}
+
+#[cfg(unix)]
+pub(super) struct RetiredDirectGraphics {
+    generation: u64,
+    transfers: Vec<(u64, u32)>,
+    saturated: bool,
+}
+
+#[cfg(unix)]
+impl RetiredDirectGraphics {
+    fn new(generation: u64) -> Self {
+        Self {
+            generation,
+            transfers: Vec::new(),
+            saturated: false,
+        }
+    }
+}
+
 /// State tracking for the thin client.
 pub(super) struct ClientState {
     /// Stateful semantic-frame encoder used when the server sends FrameData.
     pub(super) blit_encoder: render_ansi::BlitEncoder,
+    pub(super) image_files: image_files::FileTransport,
     pub(super) mouse_capture_active: bool,
     pub(super) endpoint_mouse_capture_requested: bool,
     pub(super) endpoint_sgr_pixels_requested: bool,
@@ -24,10 +54,15 @@ pub(super) struct ClientState {
     #[cfg(unix)]
     pub(super) direct_graphics_response: Arc<Mutex<direct_graphics::ResponseMatcher>>,
     #[cfg(unix)]
-    pub(super) retired_direct_graphics: Option<(endpoint::ClientEndpointId, u64, u32)>,
+    pub(super) retired_direct_graphics: HashMap<endpoint::ClientEndpointId, RetiredDirectGraphics>,
     #[cfg(unix)]
-    pub(super) pending_surface_graphics:
-        HashMap<(endpoint::ClientEndpointId, u64, u32), crate::protocol::SurfaceGraphicsAssetKey>,
+    pub(super) disabled_native_graphics: HashMap<endpoint::ClientEndpointId, u64>,
+    pub(super) pending_native_cleanup: Vec<u8>,
+    #[cfg(unix)]
+    pub(super) pending_surface_graphics: HashMap<
+        (endpoint::ClientEndpointId, u64, u64, u32),
+        crate::protocol::SurfaceGraphicsAssetKey,
+    >,
     pub(super) attach_escape: Option<AttachEscapeState>,
     #[cfg(unix)]
     pub(super) mouse_scroll_lines: usize,
@@ -63,6 +98,7 @@ impl ClientState {
     pub(super) fn test_new() -> Self {
         Self {
             blit_encoder: render_ansi::BlitEncoder::new(),
+            image_files: image_files::FileTransport::default(),
             mouse_capture_active: false,
             endpoint_mouse_capture_requested: false,
             endpoint_sgr_pixels_requested: false,
@@ -81,7 +117,10 @@ impl ClientState {
             #[cfg(unix)]
             direct_graphics_response: Default::default(),
             #[cfg(unix)]
-            retired_direct_graphics: None,
+            retired_direct_graphics: HashMap::new(),
+            #[cfg(unix)]
+            disabled_native_graphics: Default::default(),
+            pending_native_cleanup: Vec::new(),
             #[cfg(unix)]
             pending_surface_graphics: HashMap::new(),
             attach_escape: None,
@@ -162,11 +201,109 @@ impl ClientState {
 
     /// Present a composed error/chrome frame while retaining the handoff input freeze. The pane
     /// cells are still the last coherent surface; only client chrome (including the error) moves.
-    pub(super) fn present_frozen_chrome(&mut self, frame_data: FrameData) {
+    pub(super) fn present_frozen_chrome(
+        &mut self,
+        frame_data: impl Into<frame_output::ComposedFrame>,
+    ) {
         let frozen = self.presentation_frozen;
+        // Chrome can repaint the frozen source, but cannot retire its staged images.
+        let deferred_cleanup = frozen.then(|| std::mem::take(&mut self.pending_native_cleanup));
         self.presentation_frozen = false;
         self.present_frame(frame_data);
         self.presentation_frozen = frozen;
+        if let Some(cleanup) = deferred_cleanup {
+            self.pending_native_cleanup = cleanup;
+        }
+    }
+
+    #[cfg(unix)]
+    pub(super) fn queue_native_image_cleanup(&mut self, image_id: u32) {
+        crate::kitty_graphics::encode_delete_image(&mut self.pending_native_cleanup, image_id);
+    }
+
+    /// Retirement is lifecycle bookkeeping, not a presentation effect. The caller has
+    /// already checked the endpoint generation, even for inactive/frozen owners.
+    #[cfg(unix)]
+    pub(super) fn receive_graphics_retirement(
+        &mut self,
+        endpoint_id: &endpoint::ClientEndpointId,
+        generation: u64,
+        transfer_id: u64,
+        image_id: u32,
+        owner_active: bool,
+    ) {
+        if transfer_id & crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT != 0 {
+            self.disabled_native_graphics
+                .insert(endpoint_id.clone(), generation);
+        }
+        self.record_retired_direct_graphics(endpoint_id.clone(), generation, transfer_id, image_id);
+        let upload_pending = self
+            .pending_surface_graphics
+            .remove(&(endpoint_id.clone(), generation, transfer_id, image_id))
+            .is_some();
+        // Retirement cleanup is terminal-owned. A native retirement owns the
+        // ID only while its exact upload is pending: collision rejection and a
+        // late post-ACK retirement must not delete the currently visible bank.
+        // API/non-native retirement retains its prior ownership behavior.
+        let owns_image = self.queue_retired_graphics_cleanup(
+            transfer_id,
+            image_id,
+            upload_pending,
+            owner_active,
+        );
+        let frame = if owns_image && owner_active {
+            let frozen = self.presentation_frozen;
+            let size = self.reported_size;
+            self.shell.as_mut().and_then(|shell| {
+                shell.retire_direct_graphics_image(image_id);
+                (!frozen).then(|| shell.compose(size.0, size.1)).flatten()
+            })
+        } else {
+            None
+        };
+        // A retirement repaint is a normal synchronized frame. In particular,
+        // do not extract its graphics into a graphics-only swap.
+        if let Some(frame) = frame {
+            self.present_frame(frame);
+        }
+        if let Ok(mut matcher) = self.direct_graphics_response.lock() {
+            matcher.retire(transfer_id);
+        }
+    }
+
+    /// Returns whether a retirement owns an image that may be removed from the terminal/cache.
+    /// Native rejections can retire a proposed colliding ID without ever uploading it; only an
+    /// exact still-pending native upload gives that retirement ownership of the image.
+    #[cfg(unix)]
+    pub(super) fn queue_retired_graphics_cleanup(
+        &mut self,
+        transfer_id: u64,
+        image_id: u32,
+        upload_pending: bool,
+        owner_active: bool,
+    ) -> bool {
+        let native = transfer_id & crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT != 0;
+        if native && !upload_pending {
+            return false;
+        }
+        if native || !owner_active {
+            self.queue_native_image_cleanup(image_id);
+        }
+        true
+    }
+
+    #[cfg(unix)]
+    pub(super) fn flush_native_cleanup(
+        &mut self,
+        writer: &mut impl std::io::Write,
+    ) -> io::Result<()> {
+        if self.presentation_frozen || self.pending_native_cleanup.is_empty() {
+            return Ok(());
+        }
+        writer.write_all(&self.pending_native_cleanup)?;
+        writer.flush()?;
+        self.pending_native_cleanup.clear();
+        Ok(())
     }
 
     pub(super) fn present_graphics(&mut self, graphics: &[u8]) {
@@ -182,7 +319,10 @@ impl ClientState {
         &mut self,
         patch: shell::ClientComposedSurfacePatch,
     ) -> io::Result<bool> {
-        if self.presentation_frozen || self.repaint_pending {
+        if self.presentation_frozen
+            || self.repaint_pending
+            || (self.kitty_graphics_enabled && !self.pending_native_cleanup.is_empty())
+        {
             crate::render_prof::event("client_surface_patch.fallback.repaint");
             return Ok(false);
         }
@@ -222,21 +362,28 @@ impl ClientState {
     }
 
     #[cfg(unix)]
-    pub(super) fn retire_endpoint_graphics(&mut self, endpoint_id: &endpoint::ClientEndpointId) {
-        let transfer_ids = self
+    fn retire_pending_endpoint_graphics(
+        &mut self,
+        endpoint_id: &endpoint::ClientEndpointId,
+        generation: Option<u64>,
+    ) {
+        let retired = self
             .pending_surface_graphics
             .keys()
-            .filter(|(owner, _, _)| owner == endpoint_id)
-            .map(|(_, transfer_id, _)| *transfer_id)
+            .filter(|(owner, pending_generation, _, _)| {
+                owner == endpoint_id
+                    && generation.is_none_or(|generation| *pending_generation == generation)
+            })
+            .cloned()
             .collect::<Vec<_>>();
-        self.pending_surface_graphics
-            .retain(|(owner, _, _), _| owner != endpoint_id);
-        if self
-            .retired_direct_graphics
-            .as_ref()
-            .is_some_and(|(owner, _, _)| owner == endpoint_id)
-        {
-            self.retired_direct_graphics = None;
+        let mut transfer_ids = Vec::with_capacity(retired.len());
+        for key in retired {
+            let (_, _, transfer_id, image_id) = &key;
+            if transfer_id & crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT != 0 {
+                self.queue_native_image_cleanup(*image_id);
+            }
+            transfer_ids.push(*transfer_id);
+            self.pending_surface_graphics.remove(&key);
         }
         if let Ok(mut matcher) = self.direct_graphics_response.lock() {
             for transfer_id in transfer_ids {
@@ -245,10 +392,162 @@ impl ClientState {
         }
     }
 
-    pub(super) fn present_frame(&mut self, frame_data: FrameData) {
-        if self.presentation_frozen {
+    #[cfg(unix)]
+    pub(super) fn record_retired_direct_graphics(
+        &mut self,
+        endpoint_id: endpoint::ClientEndpointId,
+        generation: u64,
+        transfer_id: u64,
+        image_id: u32,
+    ) {
+        let retired = self
+            .retired_direct_graphics
+            .entry(endpoint_id)
+            .or_insert_with(|| RetiredDirectGraphics::new(generation));
+        if retired.generation != generation {
+            *retired = RetiredDirectGraphics::new(generation);
+        }
+        if retired.saturated || retired.transfers.contains(&(transfer_id, image_id)) {
+            return;
+        }
+        if retired.transfers.len() == MAX_RETIRED_DIRECT_GRAPHICS {
+            // Never evict an older tombstone: a delayed file for an evicted tuple could otherwise
+            // overwrite a visible image. Reject every unrecognized file until generation reset.
+            retired.saturated = true;
             return;
         }
+        retired.transfers.push((transfer_id, image_id));
+    }
+
+    #[cfg(unix)]
+    pub(super) fn match_retired_direct_graphics(
+        &mut self,
+        endpoint_id: &endpoint::ClientEndpointId,
+        generation: u64,
+        transfer_id: u64,
+        image_id: u32,
+    ) -> RetiredDirectGraphicsMatch {
+        let Some(retired) = self.retired_direct_graphics.get_mut(endpoint_id) else {
+            return RetiredDirectGraphicsMatch::None;
+        };
+        if retired.generation != generation {
+            return RetiredDirectGraphicsMatch::None;
+        }
+        if let Some(index) = retired
+            .transfers
+            .iter()
+            .position(|transfer| *transfer == (transfer_id, image_id))
+        {
+            retired.transfers.swap_remove(index);
+            if retired.transfers.is_empty() && !retired.saturated {
+                self.retired_direct_graphics.remove(endpoint_id);
+            }
+            return RetiredDirectGraphicsMatch::Exact;
+        }
+        if retired.saturated {
+            RetiredDirectGraphicsMatch::Saturated
+        } else {
+            RetiredDirectGraphicsMatch::None
+        }
+    }
+
+    /// Forget graphics guards only for the connection generation which actually disconnected.
+    /// A delayed disconnect from an older generation must not clear its replacement's guards.
+    #[cfg(unix)]
+    pub(super) fn retire_endpoint_graphics(
+        &mut self,
+        endpoint_id: &endpoint::ClientEndpointId,
+        generation: u64,
+    ) {
+        self.retire_pending_endpoint_graphics(endpoint_id, Some(generation));
+        if self.disabled_native_graphics.get(endpoint_id) == Some(&generation) {
+            self.disabled_native_graphics.remove(endpoint_id);
+        }
+        if self
+            .retired_direct_graphics
+            .get(endpoint_id)
+            .is_some_and(|retired| retired.generation == generation)
+        {
+            self.retired_direct_graphics.remove(endpoint_id);
+        }
+    }
+
+    /// Reset bounded per-endpoint state when a replacement connection becomes authoritative.
+    #[cfg(unix)]
+    pub(super) fn start_endpoint_graphics_generation(
+        &mut self,
+        endpoint_id: &endpoint::ClientEndpointId,
+        generation: u64,
+    ) {
+        self.retire_pending_endpoint_graphics(endpoint_id, None);
+        if self.disabled_native_graphics.get(endpoint_id) != Some(&generation) {
+            self.disabled_native_graphics.remove(endpoint_id);
+        }
+        if self
+            .retired_direct_graphics
+            .get(endpoint_id)
+            .is_some_and(|retired| retired.generation != generation)
+        {
+            self.retired_direct_graphics.remove(endpoint_id);
+        }
+    }
+
+    #[cfg(unix)]
+    pub(super) fn forget_endpoint_graphics(&mut self, endpoint_id: &endpoint::ClientEndpointId) {
+        self.retire_pending_endpoint_graphics(endpoint_id, None);
+        self.disabled_native_graphics.remove(endpoint_id);
+        self.retired_direct_graphics.remove(endpoint_id);
+    }
+
+    pub(super) fn present_frame(&mut self, frame_data: impl Into<frame_output::ComposedFrame>) {
+        let _ = self.try_present_frame(frame_data);
+    }
+
+    fn write_composed_output(
+        &mut self,
+        writer: &mut impl io::Write,
+        encoded: &[u8],
+        mut graphics: crate::kitty_graphics::GraphicsOutput,
+    ) -> io::Result<()> {
+        if self.kitty_graphics_enabled {
+            if !self.pending_native_cleanup.is_empty() {
+                graphics.operations.insert(
+                    0,
+                    crate::kitty_graphics::GraphicsOperation::Bytes(
+                        self.pending_native_cleanup.clone(),
+                    ),
+                );
+            }
+            frame_output::write_composed_frame(
+                writer.by_ref(),
+                encoded,
+                &graphics,
+                &mut self.image_files,
+            )?;
+        } else {
+            writer.write_all(encoded)?;
+        }
+        writer.flush()?;
+        if self.kitty_graphics_enabled {
+            self.pending_native_cleanup.clear();
+        }
+        Ok(())
+    }
+
+    /// Presents and commits a frame only after all terminal output has been written successfully.
+    /// Callers which acknowledge presentation-sensitive work use the return value rather than
+    /// treating composition as presentation.
+    pub(super) fn try_present_frame(
+        &mut self,
+        frame_data: impl Into<frame_output::ComposedFrame>,
+    ) -> bool {
+        if self.presentation_frozen {
+            return false;
+        }
+        let frame_output::ComposedFrame {
+            frame: frame_data,
+            graphics,
+        } = frame_data.into();
         let frame_data = if self.draw_host_cursor {
             render_ansi::frame_with_drawn_cursor(frame_data)
         } else {
@@ -261,14 +560,280 @@ impl ClientState {
             self.blit_encoder.encode(&frame_data, self.repaint_pending)
         };
         let mut stdout = io::stdout();
-        let graphics = if self.kitty_graphics_enabled {
-            frame_data.graphics.as_slice()
-        } else {
-            &[]
-        };
-        let _ = write_encoded_frame_with_graphics(&mut stdout, &encoded.bytes, graphics);
-        let _ = stdout.flush();
+        if let Err(error) = self.write_composed_output(&mut stdout, &encoded.bytes, graphics) {
+            tracing::warn!(%error, "failed to present client frame");
+            self.repaint_pending = true;
+            return false;
+        }
         self.blit_encoder.commit(frame_data, encoded);
         self.repaint_pending = false;
+        true
+    }
+}
+
+#[cfg(all(test, unix))]
+mod native_cleanup_tests {
+    use super::*;
+
+    #[test]
+    fn retired_graphics_tombstones_survive_handoff_and_only_exact_files_consume_them() {
+        let mut state = ClientState::test_new();
+        let endpoint = endpoint::ClientEndpointId::Local;
+        state.record_retired_direct_graphics(endpoint.clone(), 7, 11, 1234);
+        state.record_retired_direct_graphics(endpoint.clone(), 7, 12, 5678);
+        state.disabled_native_graphics.insert(endpoint.clone(), 7);
+
+        // Switching away and back does not reset guards, and an unrelated queued file cannot
+        // consume either of two sequential API retirement tombstones.
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 7, 13, 9999),
+            RetiredDirectGraphicsMatch::None
+        );
+        assert_eq!(
+            state
+                .retired_direct_graphics
+                .get(&endpoint)
+                .unwrap()
+                .transfers
+                .len(),
+            2
+        );
+        assert_eq!(state.disabled_native_graphics.get(&endpoint), Some(&7));
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 7, 11, 1234),
+            RetiredDirectGraphicsMatch::Exact
+        );
+        assert!(state
+            .retired_direct_graphics
+            .get(&endpoint)
+            .unwrap()
+            .transfers
+            .contains(&(12, 5678)));
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 7, 12, 5678),
+            RetiredDirectGraphicsMatch::Exact
+        );
+        assert!(!state.retired_direct_graphics.contains_key(&endpoint));
+    }
+
+    #[test]
+    fn replacement_generation_resets_bounded_graphics_guards() {
+        let mut state = ClientState::test_new();
+        let endpoint = endpoint::ClientEndpointId::Local;
+        let native = crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT | 11;
+        state.record_retired_direct_graphics(endpoint.clone(), 7, native, 1234);
+        state.disabled_native_graphics.insert(endpoint.clone(), 7);
+
+        state.start_endpoint_graphics_generation(&endpoint, 8);
+
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 7, native, 1234),
+            RetiredDirectGraphicsMatch::None
+        );
+        assert!(!state.disabled_native_graphics.contains_key(&endpoint));
+    }
+
+    #[test]
+    fn retired_graphics_tombstone_overflow_fails_closed_until_generation_reset() {
+        let mut state = ClientState::test_new();
+        let endpoint = endpoint::ClientEndpointId::Local;
+        for transfer_id in 0..=MAX_RETIRED_DIRECT_GRAPHICS as u64 {
+            state.record_retired_direct_graphics(
+                endpoint.clone(),
+                7,
+                transfer_id,
+                transfer_id as u32,
+            );
+        }
+        let retired = state.retired_direct_graphics.get(&endpoint).unwrap();
+        assert_eq!(retired.transfers.len(), MAX_RETIRED_DIRECT_GRAPHICS);
+        assert!(retired.saturated);
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 7, 1, 1),
+            RetiredDirectGraphicsMatch::Exact
+        );
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 7, 10_000, 10_000),
+            RetiredDirectGraphicsMatch::Saturated
+        );
+        assert!(
+            state
+                .retired_direct_graphics
+                .get(&endpoint)
+                .unwrap()
+                .saturated
+        );
+
+        state.start_endpoint_graphics_generation(&endpoint, 8);
+        assert!(!state.retired_direct_graphics.contains_key(&endpoint));
+        state.record_retired_direct_graphics(endpoint.clone(), 8, 1, 2);
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 8, 9, 9),
+            RetiredDirectGraphicsMatch::None
+        );
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 8, 1, 2),
+            RetiredDirectGraphicsMatch::Exact
+        );
+    }
+
+    #[test]
+    fn pending_none_native_retirement_does_not_touch_cleanup_or_shell_cache() {
+        let mut state = ClientState::test_new();
+        state.pending_native_cleanup.extend_from_slice(b"existing");
+        let transfer = crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT | 7;
+
+        assert!(!state.queue_retired_graphics_cleanup(transfer, 1234, false, true));
+        assert_eq!(state.pending_native_cleanup, b"existing");
+        assert!(state
+            .shell
+            .as_mut()
+            .expect("test shell")
+            .take_pending_graphics_cleanup()
+            .is_empty());
+    }
+
+    #[test]
+    fn queued_cleanup_is_synchronized_and_respects_graphics_capability() {
+        let mut state = ClientState::test_new();
+        state.kitty_graphics_enabled = true;
+        state.queue_native_image_cleanup(42);
+        let encoded = b"\x1b[?2026htext\x1b[?2026l";
+        let mut output = Vec::new();
+        state
+            .write_composed_output(&mut output, encoded, Default::default())
+            .unwrap();
+        let text = String::from_utf8(output).unwrap();
+        assert!(text.find("\x1b[?2026h").unwrap() < text.find("a=d,d=I,i=42").unwrap());
+        assert!(text.find("a=d,d=I,i=42").unwrap() < text.find("\x1b[?2026l").unwrap());
+        assert!(state.pending_native_cleanup.is_empty());
+        state.kitty_graphics_enabled = false;
+        state.queue_native_image_cleanup(43);
+        let mut output = Vec::new();
+        state
+            .write_composed_output(&mut output, encoded, Default::default())
+            .unwrap();
+        assert_eq!(output, encoded);
+        assert!(!state.pending_native_cleanup.is_empty());
+        state.kitty_graphics_enabled = true;
+        let mut no_capacity = &mut [][..];
+        assert!(state
+            .write_composed_output(&mut no_capacity, encoded, Default::default())
+            .is_err());
+        assert!(!state.pending_native_cleanup.is_empty());
+    }
+
+    #[test]
+    fn active_api_retirement_leaves_cleanup_to_composition() {
+        let mut state = ClientState::test_new();
+        assert!(state.queue_retired_graphics_cleanup(7, 1234, true, true));
+        assert!(state.pending_native_cleanup.is_empty());
+        assert!(state.queue_retired_graphics_cleanup(8, 5678, true, false));
+        assert!(!state.pending_native_cleanup.is_empty());
+    }
+
+    #[test]
+    fn unacknowledged_native_upload_cleanup_survives_frozen_disconnect() {
+        let mut state = ClientState::test_new();
+        let endpoint = endpoint::ClientEndpointId::Local;
+        let key = crate::protocol::SurfaceGraphicsAssetKey {
+            source: crate::protocol::SurfaceGraphicsSource::Terminal {
+                target: crate::protocol::SurfaceGraphicsTarget::Pane {
+                    pane_id: "p".into(),
+                },
+                image_id: 1,
+            },
+            image_width: 1,
+            image_height: 1,
+            format: crate::protocol::SurfaceGraphicsFormat::Rgba,
+            data_len: 4,
+            data_fingerprint: 1,
+        };
+        let transfer = crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT | 2;
+        state
+            .pending_surface_graphics
+            .insert((endpoint.clone(), 9, transfer, 1234), key);
+        state.presentation_frozen = true;
+        state.retire_endpoint_graphics(&endpoint, 9);
+        assert!(state.pending_surface_graphics.is_empty());
+        let mut output = Vec::new();
+        state.flush_native_cleanup(&mut output).unwrap();
+        assert!(output.is_empty());
+        assert!(!state.pending_native_cleanup.is_empty());
+        state.unfreeze_presentation();
+        state.flush_native_cleanup(&mut output).unwrap();
+        assert!(String::from_utf8(output.clone())
+            .unwrap()
+            .contains("a=d,d=I,i=1234"));
+        state.flush_native_cleanup(&mut output).unwrap();
+        assert_eq!(
+            String::from_utf8(output).unwrap().matches("a=d,").count(),
+            1
+        );
+    }
+
+    #[test]
+    fn unacknowledged_native_upload_retirement_survives_frozen_handoff() {
+        let mut state = ClientState::test_new();
+        let endpoint = endpoint::ClientEndpointId::Local;
+        let key = crate::protocol::SurfaceGraphicsAssetKey {
+            source: crate::protocol::SurfaceGraphicsSource::Terminal {
+                target: crate::protocol::SurfaceGraphicsTarget::Pane {
+                    pane_id: "p".into(),
+                },
+                image_id: 1,
+            },
+            image_width: 1,
+            image_height: 1,
+            format: crate::protocol::SurfaceGraphicsFormat::Rgba,
+            data_len: 4,
+            data_fingerprint: 1,
+        };
+        let transfer = crate::kitty_graphics::surface::NATIVE_TRANSFER_BIT | 2;
+        state
+            .pending_surface_graphics
+            .insert((endpoint.clone(), 9, transfer, 1234), key);
+        state.presentation_frozen = true;
+        // The source connection is still live, but no longer owns the active shell.
+        state.receive_graphics_retirement(&endpoint, 8, transfer, 1234, false);
+        assert_eq!(state.pending_surface_graphics.len(), 1);
+        assert!(state.pending_native_cleanup.is_empty());
+        state.receive_graphics_retirement(&endpoint, 9, transfer, 1234, false);
+        assert_eq!(
+            state.match_retired_direct_graphics(&endpoint, 9, transfer, 1234),
+            RetiredDirectGraphicsMatch::Exact
+        );
+        assert!(state
+            .shell
+            .as_mut()
+            .unwrap()
+            .take_pending_graphics_cleanup()
+            .is_empty());
+        assert!(state.pending_surface_graphics.is_empty());
+        let mut output = Vec::new();
+        state.flush_native_cleanup(&mut output).unwrap();
+        assert!(output.is_empty());
+        assert!(!state.pending_native_cleanup.is_empty());
+        let cleanup = state.pending_native_cleanup.clone();
+        let frame = crate::protocol::FrameData::from_ratatui_buffer(
+            &ratatui::buffer::Buffer::empty(ratatui::layout::Rect::new(0, 0, 1, 1)),
+            None,
+        );
+        state.kitty_graphics_enabled = true;
+        assert!(!state.try_present_frame(frame.clone()));
+        assert_eq!(state.pending_native_cleanup, cleanup);
+        state.present_frozen_chrome(frame);
+        assert!(state.presentation_frozen);
+        assert_eq!(state.pending_native_cleanup, cleanup);
+        state.unfreeze_presentation();
+        state.flush_native_cleanup(&mut output).unwrap();
+        assert!(String::from_utf8(output.clone())
+            .unwrap()
+            .contains("a=d,d=I,i=1234"));
+        state.flush_native_cleanup(&mut output).unwrap();
+        assert_eq!(
+            String::from_utf8(output).unwrap().matches("a=d,").count(),
+            1
+        );
     }
 }
diff --git a/src/client/terminal_sessions.rs b/src/client/terminal_sessions.rs
index c7328efd..6d3160c8 100644
--- a/src/client/terminal_sessions.rs
+++ b/src/client/terminal_sessions.rs
@@ -100,6 +100,7 @@ fn connect_terminal_session_stream(
         false,
         false,
         true,
+        true,
     ) {
         Ok(handshake) if handshake.encoding == RenderEncoding::TerminalAnsi => {}
         Ok(handshake) => {
diff --git a/src/client/tests/mod.rs b/src/client/tests/mod.rs
index 2665e7fd..8519ea07 100644
--- a/src/client/tests/mod.rs
+++ b/src/client/tests/mod.rs
@@ -69,6 +69,21 @@ fn direct_graphics_profile_is_narrow_and_transport_safe() {
     ));
 }
 
+#[test]
+fn server_graphics_files_require_local_filesystem_not_just_local_terminal() {
+    let local = endpoint::ClientEndpointId::Local;
+    let ssh = endpoint::ClientEndpointId::Ssh(
+        endpoint::ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap(),
+    );
+    // Keep old local peers working without requiring a negotiated capability.
+    assert!(server_graphics_files_allowed(&local, false));
+    // Saved SSH endpoints run in the normal local client process.
+    assert!(!server_graphics_files_allowed(&ssh, false));
+    // A standalone --remote bridge can retain the Local endpoint identity.
+    assert!(!server_graphics_files_allowed(&local, true));
+    assert!(!server_graphics_files_allowed(&ssh, true));
+}
+
 fn restore_env_var(key: &str, value: Option<OsString>) {
     if let Some(value) = value {
         std::env::set_var(key, value);