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>

1 files changed, 0 insertions, 1114 deletions

diff --git a/src/api/server/pane_graphics_stream.rs b/src/api/server/pane_graphics_stream.rs
deleted file mode 100644
index 55816b1d..00000000
--- a/src/api/server/pane_graphics_stream.rs
+++ /dev/null
@@ -1,1114 +0,0 @@
-use std::io::{self, Read};
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::Arc;
-use std::time::{Duration, Instant};
-
-use interprocess::local_socket::traits::Stream as _;
-
-use crate::api::schema::{
-    ErrorBody, ErrorResponse, Method, PaneGraphicsSetParams, PaneGraphicsStreamParams, Request,
-    ResponseResult, SuccessResponse,
-};
-use crate::api::ApiRequestSender;
-use crate::ipc::{is_connection_closed_error, LocalStream};
-
-use super::{
-    api_response_outcome, dispatch_stream_frame, dispatch_stream_open,
-    dispatch_to_app_with_timeout, write_json_line, write_json_line_allow_disconnect,
-    write_text_line_allow_disconnect, APP_RESPONSE_TIMEOUT, CONNECTION_POLL_INTERVAL,
-};
-
-const MAX_STREAM_FRAME_HEADER_BYTES: usize = 64 * 1024;
-const STREAM_FRAME_BODY_CHUNK_BYTES: usize = 64 * 1024;
-const STREAM_FRAME_HEADER_IDLE_TIMEOUT: Duration = Duration::from_secs(5);
-const STREAM_FRAME_HEADER_TIMEOUT: Duration = Duration::from_secs(30);
-const STREAM_FRAME_BODY_IDLE_TIMEOUT: Duration = Duration::from_secs(5);
-const STREAM_FRAME_BODY_TIMEOUT: Duration = Duration::from_secs(30);
-const STREAM_FALLBACK_POLL_INTERVAL: Duration = Duration::from_millis(1);
-const STREAM_FALLBACK_FAST_POLLS: u8 = 32;
-static NEXT_PANE_GRAPHICS_STREAM_OWNER: AtomicU64 = AtomicU64::new(1);
-
-#[derive(serde::Deserialize)]
-struct FrameFile {
-    path: String,
-}
-
-#[derive(serde::Deserialize)]
-struct FrameHeader {
-    format: crate::api::schema::PaneGraphicsFormat,
-    image_width: u32,
-    image_height: u32,
-    #[serde(default)]
-    data_length: Option<usize>,
-    #[serde(default)]
-    file: Option<FrameFile>,
-    #[serde(default)]
-    sequence: u64,
-    #[serde(default)]
-    revision: u64,
-    #[serde(default)]
-    placement: crate::api::schema::PaneGraphicsPlacementParams,
-}
-
-#[derive(Clone, Copy)]
-struct ReadTimeouts {
-    header_idle: Duration,
-    header_total: Duration,
-    body_idle: Duration,
-    body_total: Duration,
-}
-
-const READ_TIMEOUTS: ReadTimeouts = ReadTimeouts {
-    header_idle: STREAM_FRAME_HEADER_IDLE_TIMEOUT,
-    header_total: STREAM_FRAME_HEADER_TIMEOUT,
-    body_idle: STREAM_FRAME_BODY_IDLE_TIMEOUT,
-    body_total: STREAM_FRAME_BODY_TIMEOUT,
-};
-
-pub(super) fn serve(
-    stream: LocalStream,
-    request_id: String,
-    params: PaneGraphicsStreamParams,
-    api_tx: &ApiRequestSender,
-    running: &Arc<AtomicBool>,
-) -> std::io::Result<()> {
-    serve_with_open_timeout(
-        stream,
-        request_id,
-        params,
-        api_tx,
-        running,
-        APP_RESPONSE_TIMEOUT,
-    )
-}
-
-fn serve_with_open_timeout(
-    stream: LocalStream,
-    request_id: String,
-    params: PaneGraphicsStreamParams,
-    api_tx: &ApiRequestSender,
-    running: &Arc<AtomicBool>,
-    open_timeout: Duration,
-) -> std::io::Result<()> {
-    serve_with_timeouts(
-        stream,
-        request_id,
-        params,
-        api_tx,
-        running,
-        open_timeout,
-        READ_TIMEOUTS,
-    )
-}
-
-fn serve_with_timeouts(
-    mut stream: LocalStream,
-    request_id: String,
-    mut params: PaneGraphicsStreamParams,
-    api_tx: &ApiRequestSender,
-    running: &Arc<AtomicBool>,
-    open_timeout: Duration,
-    read_timeouts: ReadTimeouts,
-) -> std::io::Result<()> {
-    let pane_id = params.pane_id.clone();
-    let layer_id = params.layer_id.clone();
-    let z_index = params.z_index;
-    let owner = next_owner();
-    params.owner = owner.clone();
-    let stream_active = Arc::new(AtomicBool::new(true));
-    let open_response = dispatch_stream_open(
-        Request {
-            id: request_id.clone(),
-            method: Method::PaneGraphicsStreamOpen(params),
-        },
-        api_tx,
-        open_timeout,
-        Arc::clone(&stream_active),
-    );
-    if api_response_outcome(&open_response) != "ok" {
-        stream_active.store(false, Ordering::Release);
-        let write_result = write_text_line_allow_disconnect(&mut stream, &open_response);
-        clear_layer(&pane_id, layer_id.as_deref(), z_index, &owner, api_tx);
-        write_result?;
-        return Ok(());
-    }
-
-    if let Err(err) = write_json_line(
-        &mut stream,
-        &SuccessResponse {
-            id: request_id.clone(),
-            result: ResponseResult::Ok {},
-        },
-    ) {
-        stream_active.store(false, Ordering::Release);
-        clear_layer(&pane_id, layer_id.as_deref(), z_index, &owner, api_tx);
-        if is_connection_closed_error(&err) {
-            return Ok(());
-        }
-        return Err(err);
-    }
-
-    let result = serve_frames(
-        &mut stream,
-        &request_id,
-        &owner,
-        &pane_id,
-        layer_id.as_deref(),
-        z_index,
-        api_tx,
-        running,
-        &stream_active,
-        read_timeouts,
-    );
-    stream_active.store(false, Ordering::Release);
-    clear_layer(&pane_id, layer_id.as_deref(), z_index, &owner, api_tx);
-    result
-}
-
-fn serve_frames(
-    stream: &mut LocalStream,
-    request_id: &str,
-    owner: &str,
-    pane_id: &str,
-    layer_id: Option<&str>,
-    z_index: i32,
-    api_tx: &ApiRequestSender,
-    running: &Arc<AtomicBool>,
-    stream_active: &Arc<AtomicBool>,
-    timeouts: ReadTimeouts,
-) -> std::io::Result<()> {
-    let mut frame_seq = 0_u64;
-    while stream_is_running(running, stream_active) {
-        let Some(header_line) = read_line(
-            stream,
-            running,
-            stream_active,
-            MAX_STREAM_FRAME_HEADER_BYTES,
-            timeouts.header_idle,
-            timeouts.header_total,
-        )?
-        else {
-            return Ok(());
-        };
-        let header_line = header_line.trim();
-        if header_line.is_empty() {
-            continue;
-        }
-        let header = match serde_json::from_str::<FrameHeader>(header_line) {
-            Ok(header) => header,
-            Err(err) => {
-                write_json_line_allow_disconnect(
-                    stream,
-                    &ErrorResponse {
-                        id: request_id.to_string(),
-                        error: ErrorBody {
-                            code: "invalid_frame".into(),
-                            message: format!("invalid frame header: {err}"),
-                        },
-                    },
-                )?;
-                return Ok(());
-            }
-        };
-        if let Some(file) = header.file {
-            if !matches!(
-                header.format,
-                crate::api::schema::PaneGraphicsFormat::Rgba
-                    | crate::api::schema::PaneGraphicsFormat::Bgra
-            ) {
-                write_json_line_allow_disconnect(
-                    stream,
-                    &ErrorResponse {
-                        id: request_id.to_string(),
-                        error: ErrorBody {
-                            code: "invalid_frame".into(),
-                            message: "file frames require rgba or bgra".into(),
-                        },
-                    },
-                )?;
-                return Ok(());
-            }
-            let response = dispatch_stream_frame(
-                Request {
-                    id: format!("{request_id}:file:{}", header.sequence),
-                    method: Method::PaneGraphicsStreamDirect(
-                        crate::api::schema::PaneGraphicsDirectParams {
-                            pane_id: pane_id.to_owned(),
-                            layer_id: layer_id.map(str::to_owned),
-                            z_index,
-                            owner: owner.to_owned(),
-                            image_width: header.image_width,
-                            image_height: header.image_height,
-                            format: header.format,
-                            path: file.path,
-                            sequence: header.sequence,
-                            revision: header.revision,
-                            placement: header.placement,
-                        },
-                    ),
-                },
-                api_tx,
-                Arc::clone(stream_active),
-            );
-            write_text_line_allow_disconnect(stream, &response)?;
-            if api_response_outcome(&response) != "ok" {
-                return Ok(());
-            }
-            continue;
-        }
-
-        let Some(data_length) = header.data_length else {
-            write_json_line_allow_disconnect(
-                stream,
-                &ErrorResponse {
-                    id: request_id.to_string(),
-                    error: ErrorBody {
-                        code: "invalid_frame".into(),
-                        message: "frame requires data_length or file".into(),
-                    },
-                },
-            )?;
-            return Ok(());
-        };
-        if data_length == 0 {
-            write_json_line_allow_disconnect(
-                stream,
-                &ErrorResponse {
-                    id: request_id.to_string(),
-                    error: ErrorBody {
-                        code: "invalid_frame".into(),
-                        message: "frame data_length must be greater than zero".into(),
-                    },
-                },
-            )?;
-            return Ok(());
-        }
-        if data_length > crate::api::schema::PANE_GRAPHICS_STREAM_MAX_BYTES {
-            write_json_line_allow_disconnect(
-                stream,
-                &ErrorResponse {
-                    id: request_id.to_string(),
-                    error: ErrorBody {
-                        code: "image_too_large".into(),
-                        message: "frame data is too large".into(),
-                    },
-                },
-            )?;
-            return Ok(());
-        }
-
-        let Some(data) = read_exact(
-            stream,
-            data_length,
-            running,
-            stream_active,
-            timeouts.body_idle,
-            timeouts.body_total,
-        )?
-        else {
-            return Ok(());
-        };
-
-        frame_seq = frame_seq.saturating_add(1);
-        let frame_id = format!("{request_id}:frame:{frame_seq}");
-        let response = dispatch_to_app_with_timeout(
-            Request {
-                id: frame_id,
-                method: Method::PaneGraphicsStreamSet(PaneGraphicsSetParams {
-                    pane_id: pane_id.to_string(),
-                    layer_id: layer_id.map(str::to_owned),
-                    z_index,
-                    owner: owner.to_string(),
-                    format: header.format,
-                    image_width: header.image_width,
-                    image_height: header.image_height,
-                    data: Some(data),
-                    data_base64: String::new(),
-                    placement: header.placement,
-                }),
-            },
-            api_tx,
-            Some(APP_RESPONSE_TIMEOUT),
-        );
-        if api_response_outcome(&response) != "ok" {
-            write_text_line_allow_disconnect(stream, &response)?;
-            return Ok(());
-        }
-    }
-
-    Ok(())
-}
-
-fn stream_is_running(running: &AtomicBool, stream_active: &AtomicBool) -> bool {
-    running.load(Ordering::Relaxed) && stream_active.load(Ordering::Acquire)
-}
-
-fn next_owner() -> String {
-    let id = NEXT_PANE_GRAPHICS_STREAM_OWNER.fetch_add(1, Ordering::Relaxed);
-    format!("pane.graphics.stream:{}:{id}", std::process::id())
-}
-
-fn clear_layer(
-    pane_id: &str,
-    layer_id: Option<&str>,
-    z_index: i32,
-    owner: &str,
-    api_tx: &ApiRequestSender,
-) {
-    let _response = dispatch_to_app_with_timeout(
-        Request {
-            id: format!("pane.graphics.stream.clear:{pane_id}"),
-            method: Method::PaneGraphicsStreamClose(PaneGraphicsStreamParams {
-                pane_id: pane_id.to_string(),
-                layer_id: layer_id.map(str::to_owned),
-                z_index,
-                owner: owner.to_string(),
-            }),
-        },
-        api_tx,
-        Some(APP_RESPONSE_TIMEOUT),
-    );
-}
-
-fn read_line(
-    stream: &mut LocalStream,
-    running: &Arc<AtomicBool>,
-    stream_active: &Arc<AtomicBool>,
-    max_bytes: usize,
-    idle_timeout: Duration,
-    total_timeout: Duration,
-) -> std::io::Result<Option<String>> {
-    with_timed_reads(stream, |stream, mut wait| {
-        let mut bytes = Vec::new();
-        let mut byte = [0_u8; 1];
-        let mut total_deadline = None;
-        let mut idle_deadline = None;
-
-        loop {
-            if !stream_is_running(running, stream_active) {
-                return Ok(None);
-            }
-            ensure_before_deadlines(
-                idle_deadline,
-                total_deadline,
-                "timed out reading stream frame header",
-            )?;
-            match wait.read(stream, &mut byte) {
-                Ok(0) => return Ok(None),
-                Ok(_) => {
-                    wait.on_progress();
-                    let now = Instant::now();
-                    let total_deadline_at =
-                        *total_deadline.get_or_insert_with(|| now + total_timeout);
-                    idle_deadline = Some(now + idle_timeout);
-                    if now >= total_deadline_at {
-                        return Err(io::Error::new(
-                            io::ErrorKind::TimedOut,
-                            "timed out reading stream frame header",
-                        ));
-                    }
-                    bytes.push(byte[0]);
-                    if byte[0] == b'\n' {
-                        return String::from_utf8(bytes)
-                            .map(Some)
-                            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err));
-                    }
-                    if bytes.len() > max_bytes {
-                        return Err(io::Error::new(
-                            io::ErrorKind::InvalidData,
-                            "stream frame header is too large",
-                        ));
-                    }
-                }
-                Err(err) if read_should_retry(&err) => {
-                    wait.after_retry(idle_deadline, total_deadline);
-                }
-                Err(err) if is_connection_closed_error(&err) => return Ok(None),
-                Err(err) => return Err(err),
-            }
-        }
-    })
-}
-
-fn read_exact(
-    stream: &mut LocalStream,
-    len: usize,
-    running: &Arc<AtomicBool>,
-    stream_active: &Arc<AtomicBool>,
-    idle_timeout: Duration,
-    total_timeout: Duration,
-) -> std::io::Result<Option<Vec<u8>>> {
-    with_timed_reads(stream, |stream, mut wait| {
-        let mut data = Vec::new();
-        let mut chunk = vec![0_u8; STREAM_FRAME_BODY_CHUNK_BYTES.min(len)];
-        let total_deadline = Instant::now() + total_timeout;
-        let mut idle_deadline = Instant::now() + idle_timeout;
-
-        while data.len() < len {
-            if !stream_is_running(running, stream_active) {
-                return Ok(None);
-            }
-            ensure_before_deadlines(
-                Some(idle_deadline),
-                Some(total_deadline),
-                "timed out reading stream frame body",
-            )?;
-            let remaining = len - data.len();
-            let read_len = remaining.min(chunk.len());
-            match wait.read(stream, &mut chunk[..read_len]) {
-                Ok(0) if data.is_empty() => return Ok(None),
-                Ok(0) => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::UnexpectedEof,
-                        "stream ended mid-frame",
-                    ));
-                }
-                Ok(n) => {
-                    wait.on_progress();
-                    let now = Instant::now();
-                    if now >= total_deadline {
-                        return Err(io::Error::new(
-                            io::ErrorKind::TimedOut,
-                            "timed out reading stream frame body",
-                        ));
-                    }
-                    data.extend_from_slice(&chunk[..n]);
-                    idle_deadline = now + idle_timeout;
-                }
-                Err(err) if read_should_retry(&err) => {
-                    wait.after_retry(Some(idle_deadline), Some(total_deadline));
-                }
-                Err(err) if is_connection_closed_error(&err) && data.is_empty() => return Ok(None),
-                Err(err) => return Err(err),
-            }
-        }
-
-        Ok(Some(data))
-    })
-}
-
-#[derive(Clone, Copy)]
-enum ReadWait {
-    SocketTimeout,
-    Poll(PollBackoff),
-}
-
-impl ReadWait {
-    fn read(&self, stream: &mut LocalStream, buffer: &mut [u8]) -> io::Result<usize> {
-        if matches!(self, Self::SocketTimeout) {
-            return stream.read(buffer);
-        }
-        match crate::ipc::poll_local_stream_read_count(stream, buffer)? {
-            crate::ipc::LocalStreamReadCount::Data(count) => Ok(count),
-            crate::ipc::LocalStreamReadCount::Pending => Err(io::ErrorKind::WouldBlock.into()),
-            crate::ipc::LocalStreamReadCount::Closed => Ok(0),
-        }
-    }
-
-    fn after_retry(&mut self, idle_deadline: Option<Instant>, total_deadline: Option<Instant>) {
-        if let Self::Poll(backoff) = self {
-            sleep_until_poll(idle_deadline, total_deadline, backoff.interval);
-            backoff.advance();
-        }
-    }
-
-    fn on_progress(&mut self) {
-        if let Self::Poll(backoff) = self {
-            backoff.reset();
-        }
-    }
-}
-
-#[derive(Clone, Copy)]
-struct PollBackoff {
-    interval: Duration,
-    fast_polls_remaining: u8,
-}
-
-impl PollBackoff {
-    fn new() -> Self {
-        Self {
-            interval: STREAM_FALLBACK_POLL_INTERVAL,
-            fast_polls_remaining: STREAM_FALLBACK_FAST_POLLS,
-        }
-    }
-
-    fn advance(&mut self) {
-        if self.fast_polls_remaining > 0 {
-            self.fast_polls_remaining -= 1;
-            return;
-        }
-        self.interval = (self.interval * 2).min(CONNECTION_POLL_INTERVAL);
-    }
-
-    fn reset(&mut self) {
-        *self = Self::new();
-    }
-}
-
-fn with_timed_reads<T>(
-    stream: &mut LocalStream,
-    read: impl FnOnce(&mut LocalStream, ReadWait) -> std::io::Result<Option<T>>,
-) -> std::io::Result<Option<T>> {
-    match stream.set_recv_timeout(Some(CONNECTION_POLL_INTERVAL)) {
-        Ok(()) => {
-            let result = read(stream, ReadWait::SocketTimeout);
-            finish_timed_read(result, || stream.set_recv_timeout(None))
-        }
-        Err(err) if err.kind() == io::ErrorKind::Unsupported => {
-            crate::ipc::set_local_stream_polling(stream, true)?;
-            let result = read(stream, ReadWait::Poll(PollBackoff::new()));
-            finish_timed_read(result, || {
-                crate::ipc::set_local_stream_polling(stream, false)
-            })
-        }
-        // A peer can disconnect after the caller's running check but before
-        // setsockopt. macOS reports that closed-socket race as EINVAL.
-        Err(err) if err.kind() == io::ErrorKind::InvalidInput => Ok(None),
-        Err(err) => Err(err),
-    }
-}
-
-fn finish_timed_read<T>(
-    result: std::io::Result<Option<T>>,
-    reset: impl FnOnce() -> std::io::Result<()>,
-) -> std::io::Result<Option<T>> {
-    match result {
-        // None is terminal for this dedicated stream. macOS returns EINVAL when
-        // socket options are restored after the peer has already disconnected.
-        Ok(None) => Ok(None),
-        Ok(value) => {
-            reset()?;
-            Ok(value)
-        }
-        Err(err) => {
-            let _ = reset();
-            Err(err)
-        }
-    }
-}
-
-fn ensure_before_deadlines(
-    idle_deadline: Option<Instant>,
-    total_deadline: Option<Instant>,
-    message: &str,
-) -> std::io::Result<()> {
-    let now = Instant::now();
-    if idle_deadline.is_some_and(|deadline| now >= deadline)
-        || total_deadline.is_some_and(|deadline| now >= deadline)
-    {
-        return Err(io::Error::new(io::ErrorKind::TimedOut, message));
-    }
-    Ok(())
-}
-
-fn sleep_until_poll(
-    idle_deadline: Option<Instant>,
-    total_deadline: Option<Instant>,
-    poll_interval: Duration,
-) {
-    let now = Instant::now();
-    let until_deadline = [idle_deadline, total_deadline]
-        .into_iter()
-        .flatten()
-        .filter_map(|deadline| deadline.checked_duration_since(now))
-        .min()
-        .unwrap_or(poll_interval);
-    std::thread::sleep(poll_interval.min(until_deadline));
-}
-
-fn read_should_retry(err: &io::Error) -> bool {
-    matches!(
-        err.kind(),
-        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut | io::ErrorKind::Interrupted
-    )
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::api::schema::{ErrorResponse, Method, ResponseResult, SuccessResponse};
-    use crate::api::ApiRequestMessage;
-    #[cfg(unix)]
-    use crate::api::EventHub;
-    use crate::ipc::LocalStream;
-    use interprocess::local_socket::traits::Listener as _;
-    use std::io::{BufRead, BufReader, Write};
-    use std::path::PathBuf;
-    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-    use std::sync::Arc;
-    use std::time::{Duration, Instant};
-    use tokio::sync::mpsc;
-
-    static NEXT_LOCAL_STREAM_ID: AtomicU64 = AtomicU64::new(1);
-
-    fn local_stream_pair(_name: &str) -> (LocalStream, LocalStream, PathBuf) {
-        let unique = format!(
-            "hpg-{}-{}.sock",
-            std::process::id(),
-            NEXT_LOCAL_STREAM_ID.fetch_add(1, Ordering::Relaxed)
-        );
-        let path = std::env::temp_dir().join(unique);
-        let listener = crate::ipc::bind_local_listener(&path).unwrap();
-        let client = crate::ipc::connect_local_stream(&path).unwrap();
-        let server = listener.accept().unwrap();
-        (client, server, path)
-    }
-
-    fn read_response_line(stream: &mut LocalStream) -> String {
-        let mut reader = BufReader::new(stream);
-        let mut line = String::new();
-        reader.read_line(&mut line).unwrap();
-        line
-    }
-
-    #[cfg(unix)]
-    fn assert_server_stream_owner(owner: &str) {
-        assert!(owner.starts_with("pane.graphics.stream:"));
-    }
-
-    fn respond_ok(message: ApiRequestMessage) {
-        let response = serde_json::to_string(&SuccessResponse {
-            id: message.request.id,
-            result: ResponseResult::Ok {},
-        })
-        .unwrap();
-        message.respond_to.send(response).unwrap();
-    }
-
-    fn assert_close_and_respond(message: ApiRequestMessage, pane_id: &str, owner: &str) {
-        match &message.request.method {
-            Method::PaneGraphicsStreamClose(params) => {
-                assert_eq!(params.pane_id, pane_id);
-                assert_eq!(params.owner, owner);
-            }
-            other => panic!("unexpected close request: {other:?}"),
-        }
-        respond_ok(message);
-    }
-
-    #[test]
-    fn browser_file_header_accepts_damage_but_keeps_full_canonical_frame() {
-        let header: FrameHeader = serde_json::from_str(
-            r#"{"format":"rgba","image_width":2,"image_height":3,"sequence":7,"revision":8,"file":{"path":"/private/frame"},"damage":{"x":1,"y":1,"width":1,"height":1},"transport":"direct-kitty","placement":{"grid_cols":2,"grid_rows":3}}"#,
-        )
-        .unwrap();
-        assert_eq!(header.data_length, None);
-        assert_eq!(header.file.unwrap().path, "/private/frame");
-        assert_eq!((header.sequence, header.revision), (7, 8));
-        assert_eq!((header.image_width, header.image_height), (2, 3));
-        assert_eq!(
-            (header.placement.grid_cols, header.placement.grid_rows),
-            (2, 3)
-        );
-    }
-
-    #[test]
-    fn timed_read_skips_reset_after_stream_ends() {
-        let mut reset_called = false;
-        let result = finish_timed_read::<()>(Ok(None), || {
-            reset_called = true;
-            Ok(())
-        });
-
-        assert!(result.unwrap().is_none());
-        assert!(!reset_called);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    fn pane_graphics_stream_dispatches_binary_frames() {
-        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let (mut client, server, _path) = local_stream_pair("api-pane-graphics-stream");
-        client
-            .write_all(br#"{"id":"stream_1","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#)
-            .unwrap();
-        client.write_all(b"\n").unwrap();
-        client.flush().unwrap();
-
-        let running = Arc::new(AtomicBool::new(true));
-        let server_running = Arc::clone(&running);
-        let event_hub = EventHub::default();
-        let server_thread = std::thread::spawn(move || {
-            super::super::handle_connection(server, &api_tx, &event_hub, &server_running, None)
-        });
-
-        let open = api_rx.blocking_recv().unwrap();
-        let stream_owner = match &open.request.method {
-            Method::PaneGraphicsStreamOpen(params) => {
-                assert_eq!(params.pane_id, "pane_1");
-                assert_server_stream_owner(&params.owner);
-                assert_ne!(params.owner, "stream_1");
-                params.owner.clone()
-            }
-            other => panic!("unexpected open request: {other:?}"),
-        };
-        respond_ok(open);
-
-        let ack: SuccessResponse = serde_json::from_str(&read_response_line(&mut client)).unwrap();
-        assert_eq!(ack.id, "stream_1");
-        assert_eq!(ack.result, ResponseResult::Ok {});
-
-        client
-            .write_all(
-                br#"{"format":"png","image_width":2,"image_height":1,"data_length":4,"placement":{"grid_cols":10,"grid_rows":5}}"#,
-            )
-            .unwrap();
-        client.write_all(b"\n").unwrap();
-        client.write_all(&[1_u8, 2, 3, 4]).unwrap();
-        client.flush().unwrap();
-
-        let msg = api_rx.blocking_recv().unwrap();
-        match &msg.request.method {
-            Method::PaneGraphicsStreamSet(params) => {
-                assert_eq!(params.pane_id, "pane_1");
-                assert_eq!(params.owner, stream_owner);
-                assert_eq!(params.image_width, 2);
-                assert_eq!(params.image_height, 1);
-                assert_eq!(params.data.as_deref(), Some(&[1_u8, 2, 3, 4][..]));
-                assert!(params.data_base64.is_empty());
-                assert_eq!(params.placement.grid_cols, 10);
-                assert_eq!(params.placement.grid_rows, 5);
-            }
-            other => panic!("unexpected request: {other:?}"),
-        }
-        respond_ok(msg);
-
-        drop(client);
-        running.store(false, Ordering::Relaxed);
-        assert_close_and_respond(api_rx.blocking_recv().unwrap(), "pane_1", &stream_owner);
-        assert!(server_thread.join().unwrap().is_ok());
-    }
-
-    #[cfg(unix)]
-    #[test]
-    fn pane_graphics_stream_reports_open_errors_before_ack() {
-        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let (mut client, server, _path) = local_stream_pair("api-pane-graphics-stream-error");
-        client
-            .write_all(br#"{"id":"stream_2","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#)
-            .unwrap();
-        client.write_all(b"\n").unwrap();
-        client.flush().unwrap();
-
-        let running = Arc::new(AtomicBool::new(true));
-        let server_running = Arc::clone(&running);
-        let event_hub = EventHub::default();
-        let server_thread = std::thread::spawn(move || {
-            super::super::handle_connection(server, &api_tx, &event_hub, &server_running, None)
-        });
-
-        let open = api_rx.blocking_recv().unwrap();
-        let stream_owner = match &open.request.method {
-            Method::PaneGraphicsStreamOpen(params) => {
-                assert_eq!(params.pane_id, "pane_1");
-                assert_server_stream_owner(&params.owner);
-                assert_ne!(params.owner, "stream_2");
-                params.owner.clone()
-            }
-            other => panic!("unexpected open request: {other:?}"),
-        };
-        open.respond_to
-            .send(super::super::error_response_json(
-                open.request.id,
-                "feature_disabled",
-                "pane graphics are disabled by terminal.kitty_graphics".into(),
-            ))
-            .unwrap();
-
-        let response: ErrorResponse =
-            serde_json::from_str(&read_response_line(&mut client)).unwrap();
-        assert_eq!(response.id, "stream_2");
-        assert_eq!(response.error.code, "feature_disabled");
-
-        assert_close_and_respond(api_rx.blocking_recv().unwrap(), "pane_1", &stream_owner);
-
-        drop(client);
-        running.store(false, Ordering::Relaxed);
-        assert!(server_thread.join().unwrap().is_ok());
-    }
-
-    #[cfg(unix)]
-    #[test]
-    fn pane_graphics_stream_closes_claim_after_open_timeout() {
-        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let (mut client, server, _path) = local_stream_pair("api-pane-graphics-stream-timeout");
-        let running = Arc::new(AtomicBool::new(true));
-        let server_running = Arc::clone(&running);
-        let server_thread = std::thread::spawn(move || {
-            serve_with_open_timeout(
-                server,
-                "stream_timeout".into(),
-                PaneGraphicsStreamParams {
-                    pane_id: "pane_1".into(),
-                    layer_id: None,
-                    z_index: 0,
-                    owner: String::new(),
-                },
-                &api_tx,
-                &server_running,
-                Duration::from_millis(10),
-            )
-        });
-
-        let open = api_rx.blocking_recv().unwrap();
-        let stream_owner = match &open.request.method {
-            Method::PaneGraphicsStreamOpen(params) => {
-                assert_eq!(params.pane_id, "pane_1");
-                assert_server_stream_owner(&params.owner);
-                assert_ne!(params.owner, "stream_timeout");
-                params.owner.clone()
-            }
-            other => panic!("unexpected open request: {other:?}"),
-        };
-
-        let response: ErrorResponse =
-            serde_json::from_str(&read_response_line(&mut client)).unwrap();
-        assert_eq!(response.id, "stream_timeout");
-        assert_eq!(response.error.code, "server_unavailable");
-        assert!(response.error.message.contains("timed out"));
-
-        assert_close_and_respond(api_rx.blocking_recv().unwrap(), "pane_1", &stream_owner);
-
-        drop(open);
-        drop(client);
-        running.store(false, Ordering::Relaxed);
-        assert!(server_thread.join().unwrap().is_ok());
-    }
-
-    #[cfg(unix)]
-    #[test]
-    fn pane_graphics_stream_closes_claim_when_client_disconnects_before_ack() {
-        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let (mut client, server, _path) =
-            local_stream_pair("api-pane-graphics-stream-ack-disconnect");
-        client
-            .write_all(br#"{"id":"stream_3","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#)
-            .unwrap();
-        client.write_all(b"\n").unwrap();
-        client.flush().unwrap();
-
-        let running = Arc::new(AtomicBool::new(true));
-        let server_running = Arc::clone(&running);
-        let event_hub = EventHub::default();
-        let server_thread = std::thread::spawn(move || {
-            super::super::handle_connection(server, &api_tx, &event_hub, &server_running, None)
-        });
-
-        let open = api_rx.blocking_recv().unwrap();
-        let stream_owner = match &open.request.method {
-            Method::PaneGraphicsStreamOpen(params) => {
-                assert_eq!(params.pane_id, "pane_1");
-                assert_server_stream_owner(&params.owner);
-                assert_ne!(params.owner, "stream_3");
-                params.owner.clone()
-            }
-            other => panic!("unexpected open request: {other:?}"),
-        };
-
-        drop(client);
-        respond_ok(open);
-
-        let (close_tx, close_rx) = std::sync::mpsc::channel();
-        std::thread::spawn(move || {
-            close_tx.send(api_rx.blocking_recv()).unwrap();
-        });
-        let close = close_rx
-            .recv_timeout(Duration::from_secs(2))
-            .unwrap()
-            .unwrap();
-        assert_close_and_respond(close, "pane_1", &stream_owner);
-
-        running.store(false, Ordering::Relaxed);
-        assert!(server_thread.join().unwrap().is_ok());
-    }
-
-    #[test]
-    fn idle_graphics_stream_waits_for_header_without_timing_out() {
-        let (_client, mut server, _path) = local_stream_pair("graphics-idle-header");
-        let running = Arc::new(AtomicBool::new(true));
-        let active = Arc::new(AtomicBool::new(true));
-        let stop = Arc::clone(&running);
-        let stopper = std::thread::spawn(move || {
-            std::thread::sleep(Duration::from_millis(30));
-            stop.store(false, Ordering::Relaxed);
-        });
-
-        let result = read_line(
-            &mut server,
-            &running,
-            &active,
-            MAX_STREAM_FRAME_HEADER_BYTES,
-            Duration::from_millis(10),
-            Duration::from_millis(20),
-        )
-        .unwrap();
-
-        assert!(result.is_none());
-        stopper.join().unwrap();
-    }
-
-    #[test]
-    fn fallback_poll_backoff_preserves_fast_window_then_reaches_poll_ceiling() {
-        let mut backoff = PollBackoff::new();
-        for _ in 0..STREAM_FALLBACK_FAST_POLLS {
-            backoff.advance();
-            assert_eq!(backoff.interval, STREAM_FALLBACK_POLL_INTERVAL);
-        }
-
-        backoff.advance();
-        assert_eq!(backoff.interval, Duration::from_millis(2));
-        for _ in 0..6 {
-            backoff.advance();
-        }
-        assert_eq!(backoff.interval, CONNECTION_POLL_INTERVAL);
-
-        backoff.reset();
-        assert_eq!(backoff.interval, STREAM_FALLBACK_POLL_INTERVAL);
-        assert_eq!(backoff.fast_polls_remaining, STREAM_FALLBACK_FAST_POLLS);
-    }
-
-    #[test]
-    fn partial_graphics_header_times_out_after_first_byte() {
-        let (mut client, mut server, _path) = local_stream_pair("graphics-partial-header");
-        client.write_all(b"{").unwrap();
-        client.flush().unwrap();
-        let running = Arc::new(AtomicBool::new(true));
-        let active = Arc::new(AtomicBool::new(true));
-
-        let error = read_line(
-            &mut server,
-            &running,
-            &active,
-            MAX_STREAM_FRAME_HEADER_BYTES,
-            Duration::from_millis(20),
-            Duration::from_millis(100),
-        )
-        .unwrap_err();
-
-        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
-    }
-
-    #[test]
-    fn trickled_graphics_body_obeys_absolute_deadline() {
-        let (mut client, mut server, _path) = local_stream_pair("graphics-trickle-body");
-        client.write_all(&[1_u8]).unwrap();
-        client.flush().unwrap();
-        let running = Arc::new(AtomicBool::new(true));
-        let active = Arc::new(AtomicBool::new(true));
-        let writer_running = Arc::clone(&running);
-        let writer = std::thread::spawn(move || {
-            while writer_running.load(Ordering::Relaxed) {
-                if client.write_all(&[1_u8]).is_err() {
-                    break;
-                }
-                let _ = client.flush();
-                std::thread::sleep(Duration::from_millis(5));
-            }
-        });
-
-        let started = Instant::now();
-        let error = read_exact(
-            &mut server,
-            1024,
-            &running,
-            &active,
-            Duration::from_millis(20),
-            Duration::from_millis(60),
-        )
-        .unwrap_err();
-
-        running.store(false, Ordering::Relaxed);
-        writer.join().unwrap();
-        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
-        assert!(started.elapsed() >= Duration::from_millis(50));
-    }
-
-    #[test]
-    fn timed_out_header_dispatches_owner_scoped_stream_close() {
-        let (mut client, server, _path) = local_stream_pair("graphics-timeout-close");
-        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let running = Arc::new(AtomicBool::new(true));
-        let server_running = Arc::clone(&running);
-        let server_thread = std::thread::spawn(move || {
-            serve_with_timeouts(
-                server,
-                "stream-timeout".into(),
-                PaneGraphicsStreamParams {
-                    pane_id: "pane_1".into(),
-                    layer_id: None,
-                    z_index: 0,
-                    owner: String::new(),
-                },
-                &api_tx,
-                &server_running,
-                Duration::from_secs(1),
-                ReadTimeouts {
-                    header_idle: Duration::from_millis(20),
-                    header_total: Duration::from_millis(100),
-                    body_idle: Duration::from_millis(20),
-                    body_total: Duration::from_millis(100),
-                },
-            )
-        });
-
-        let open = api_rx.blocking_recv().unwrap();
-        let owner = match &open.request.method {
-            Method::PaneGraphicsStreamOpen(params) => params.owner.clone(),
-            other => panic!("unexpected open request: {other:?}"),
-        };
-        respond_ok(open);
-        let ack: SuccessResponse = serde_json::from_str(&read_response_line(&mut client)).unwrap();
-        assert_eq!(ack.id, "stream-timeout");
-        client.write_all(b"{").unwrap();
-        client.flush().unwrap();
-
-        let (close_tx, close_rx) = std::sync::mpsc::channel();
-        std::thread::spawn(move || close_tx.send(api_rx.blocking_recv()).unwrap());
-        let close = close_rx
-            .recv_timeout(Duration::from_secs(1))
-            .unwrap()
-            .unwrap();
-        assert_close_and_respond(close, "pane_1", &owner);
-
-        let error = server_thread.join().unwrap().unwrap_err();
-        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
-    }
-
-    #[test]
-    fn oversized_stream_frame_is_rejected_before_body_or_app_dispatch() {
-        let (mut client, mut server, _path) = local_stream_pair("graphics-oversized-frame");
-        let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let running = Arc::new(AtomicBool::new(true));
-        let server_running = Arc::clone(&running);
-        let stream_active = Arc::new(AtomicBool::new(true));
-        let server_thread = std::thread::spawn(move || {
-            serve_frames(
-                &mut server,
-                "stream-oversized",
-                "owner-1",
-                "pane_1",
-                None,
-                0,
-                &api_tx,
-                &server_running,
-                &stream_active,
-                READ_TIMEOUTS,
-            )
-        });
-        let header = serde_json::json!({
-            "format": "png",
-            "image_width": 1,
-            "image_height": 1,
-            "data_length": crate::api::schema::PANE_GRAPHICS_STREAM_MAX_BYTES + 1,
-        });
-        client.write_all(format!("{header}\n").as_bytes()).unwrap();
-        client.flush().unwrap();
-
-        let response: ErrorResponse =
-            serde_json::from_str(&read_response_line(&mut client)).unwrap();
-        assert_eq!(response.error.code, "image_too_large");
-        assert!(server_thread.join().unwrap().is_ok());
-        assert!(api_rx.try_recv().is_err());
-    }
-}