feat: add isolated pane graphics streaming

6 files changed, 1324 insertions, 2 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index dad4cfd8..b0d7caed 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -7,6 +7,7 @@ mod subscriptions;
 mod wait;
 
 pub use event_hub::EventHub;
+pub(crate) use server::cancel_inactive_pane_graphics_streams;
 pub use server::{start_server, start_server_with_capabilities, ServerHandle};
 pub use status::{read_runtime_status_at, RuntimeStatus};
 
@@ -51,6 +52,12 @@ pub(crate) fn request_changes_ui(request: &Request) -> bool {
             | Method::PaneResize(_)
             | Method::PaneFocus(_)
             | Method::PaneRename(_)
+            | Method::PaneGraphicsSet(_)
+            | Method::PaneGraphicsClear(_)
+            | Method::PaneGraphicsStream(_)
+            | Method::PaneGraphicsStreamSet(_)
+            | Method::PaneGraphicsStreamOpen(_)
+            | Method::PaneGraphicsStreamClose(_)
             | Method::PaneReportAgent(_)
             | Method::PaneReportAgentSession(_)
             | Method::PaneReportMetadata(_)
diff --git a/src/api/schema.rs b/src/api/schema.rs
index 4fb3d841..4449b2d9 100644
--- a/src/api/schema.rs
+++ b/src/api/schema.rs
@@ -161,6 +161,24 @@ pub enum Method {
     PaneSendInput(PaneSendInputParams),
     #[serde(rename = "pane.read")]
     PaneRead(PaneReadParams),
+    #[serde(rename = "pane.graphics.set")]
+    PaneGraphicsSet(PaneGraphicsSetParams),
+    #[serde(rename = "pane.graphics.clear")]
+    PaneGraphicsClear(PaneGraphicsClearParams),
+    #[serde(rename = "pane.graphics.info")]
+    PaneGraphicsInfo(PaneTarget),
+    #[serde(rename = "pane.graphics.stream")]
+    #[schemars(skip)]
+    PaneGraphicsStream(PaneGraphicsStreamParams),
+    #[serde(skip)]
+    #[schemars(skip)]
+    PaneGraphicsStreamSet(PaneGraphicsSetParams),
+    #[serde(skip)]
+    #[schemars(skip)]
+    PaneGraphicsStreamOpen(PaneGraphicsStreamParams),
+    #[serde(skip)]
+    #[schemars(skip)]
+    PaneGraphicsStreamClose(PaneGraphicsStreamParams),
     #[serde(rename = "pane.report_agent")]
     PaneReportAgent(PaneReportAgentParams),
     #[serde(rename = "pane.report_agent_session")]
diff --git a/src/api/schema/panes.rs b/src/api/schema/panes.rs
index 2e37e734..097b6366 100644
--- a/src/api/schema/panes.rs
+++ b/src/api/schema/panes.rs
@@ -2,6 +2,9 @@ use std::collections::HashMap;
 
 use serde::{Deserialize, Serialize};
 
+pub(crate) const PANE_GRAPHICS_SET_MAX_BYTES: usize = 512 * 1024;
+pub(crate) const PANE_GRAPHICS_STREAM_MAX_BYTES: usize = 16 * 1024 * 1024;
+
 use super::agents::AgentSessionInfo;
 use super::common::{AgentStatus, PaneAgentState, ReadFormat, ReadSource, SplitDirection};
 
@@ -257,6 +260,58 @@ pub struct PaneReadParams {
     pub strip_ansi: bool,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum PaneGraphicsFormat {
+    Png,
+    Rgb,
+    Rgba,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
+pub struct PaneGraphicsSetParams {
+    pub pane_id: String,
+    #[serde(skip)]
+    #[schemars(skip)]
+    pub owner: String,
+    pub format: PaneGraphicsFormat,
+    pub image_width: u32,
+    pub image_height: u32,
+    #[serde(skip)]
+    pub data: Option<Vec<u8>>,
+    #[serde(default)]
+    pub data_base64: String,
+    #[serde(default)]
+    pub placement: PaneGraphicsPlacementParams,
+}
+
+#[derive(
+    Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, Default,
+)]
+pub struct PaneGraphicsPlacementParams {
+    #[serde(default)]
+    pub viewport_col: i32,
+    #[serde(default)]
+    pub viewport_row: i32,
+    #[serde(default)]
+    pub grid_cols: u32,
+    #[serde(default)]
+    pub grid_rows: u32,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
+pub struct PaneGraphicsClearParams {
+    pub pane_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
+pub struct PaneGraphicsStreamParams {
+    pub pane_id: String,
+    #[serde(skip)]
+    #[schemars(skip)]
+    pub owner: String,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
 pub struct PaneReportAgentParams {
     pub pane_id: String,
diff --git a/src/api/schema/response.rs b/src/api/schema/response.rs
index 18a2635d..129c8f91 100644
--- a/src/api/schema/response.rs
+++ b/src/api/schema/response.rs
@@ -152,6 +152,10 @@ pub enum ResponseResult {
     PaneRead {
         read: PaneReadResult,
     },
+    PaneGraphicsInfo {
+        cell_width_px: u32,
+        cell_height_px: u32,
+    },
     AgentExplain {
         explain: serde_json::Value,
     },
diff --git a/src/api/server.rs b/src/api/server.rs
index 752893c6..aec7274d 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -22,6 +22,9 @@ use crate::ipc::{
     socket_file_identity, LocalStream, LocalStreamRead, SocketFileIdentity,
 };
 
+mod pane_graphics_stream;
+pub(crate) use pane_graphics_stream::cancel_inactive_streams as cancel_inactive_pane_graphics_streams;
+
 const SOCKET_PERMISSION_MODE: u32 = 0o600;
 pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
 pub(super) const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
@@ -155,14 +158,14 @@ fn handle_connection(
 
     let request = match serde_json::from_str::<Request>(line) {
         Ok(request) => request,
-        Err(err) => {
+        Err(request_error) => {
             write_json_line_allow_disconnect(
                 &mut stream,
                 &ErrorResponse {
                     id: String::new(),
                     error: ErrorBody {
                         code: "invalid_request".into(),
-                        message: format!("invalid request: {err}"),
+                        message: format!("invalid request: {request_error}"),
                     },
                 },
             )?;
@@ -176,6 +179,22 @@ fn handle_connection(
     crate::logging::api_request_started(&request_id, method, changes_ui);
 
     match request.method {
+        Method::PaneGraphicsStream(params) => {
+            let result =
+                pane_graphics_stream::serve(stream, request_id.clone(), params, api_tx, running);
+            match &result {
+                Ok(()) => crate::logging::api_request_completed(
+                    &request_id,
+                    method,
+                    "stream_closed",
+                    changes_ui,
+                ),
+                Err(err) => {
+                    crate::logging::api_request_failed(&request_id, method, &err.to_string())
+                }
+            }
+            result
+        }
         Method::EventsSubscribe(params) => {
             let result = stream_subscriptions(
                 stream,
@@ -374,6 +393,13 @@ fn api_method_name(method: &Method) -> &'static str {
         Method::PaneSendKeys(_) => "pane.send_keys",
         Method::PaneSendInput(_) => "pane.send_input",
         Method::PaneRead(_) => "pane.read",
+        Method::PaneGraphicsSet(_) => "pane.graphics.set",
+        Method::PaneGraphicsClear(_) => "pane.graphics.clear",
+        Method::PaneGraphicsInfo(_) => "pane.graphics.info",
+        Method::PaneGraphicsStream(_) => "pane.graphics.stream",
+        Method::PaneGraphicsStreamSet(_) => "pane.graphics.stream.set",
+        Method::PaneGraphicsStreamOpen(_) => "pane.graphics.stream.open",
+        Method::PaneGraphicsStreamClose(_) => "pane.graphics.stream.close",
         Method::PaneReportAgent(_) => "pane.report_agent",
         Method::PaneReportAgentSession(_) => "pane.report_agent_session",
         Method::PaneReportMetadata(_) => "pane.report_metadata",
@@ -1223,3 +1249,37 @@ mod tests {
         server_thread.join().unwrap();
     }
 }
+
+#[cfg(test)]
+mod pane_graphics_request_tests {
+    use super::*;
+    use base64::Engine as _;
+
+    #[test]
+    fn maximum_public_graphics_request_fits_initial_json_line() {
+        let request = Request {
+            id: "graphics-max".into(),
+            method: Method::PaneGraphicsSet(crate::api::schema::PaneGraphicsSetParams {
+                pane_id: "pane_1".into(),
+                owner: String::new(),
+                format: crate::api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data_base64: base64::engine::general_purpose::STANDARD
+                    .encode(vec![1_u8; crate::api::schema::PANE_GRAPHICS_SET_MAX_BYTES]),
+                data: None,
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        };
+        let encoded = serde_json::to_vec(&request).unwrap();
+
+        assert!(encoded.len() < MAX_INITIAL_REQUEST_BYTES);
+    }
+
+    #[test]
+    fn duplicate_method_cannot_be_reinterpreted_as_graphics_stream() {
+        let encoded = r#"{"id":"duplicate","method":"ping","method":"pane.graphics.stream","params":{"pane_id":"pane_1"}}"#;
+
+        assert!(serde_json::from_str::<Request>(encoded).is_err());
+    }
+}
diff --git a/src/api/server/pane_graphics_stream.rs b/src/api/server/pane_graphics_stream.rs
new file mode 100644
index 00000000..c6c58052
--- /dev/null
+++ b/src/api/server/pane_graphics_stream.rs
@@ -0,0 +1,1178 @@
+use std::collections::HashMap;
+use std::io::{self, Read};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex, OnceLock, Weak};
+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_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);
+static REGISTERED_STREAM_COUNT: AtomicUsize = AtomicUsize::new(0);
+static REGISTERED_STREAMS: OnceLock<Mutex<HashMap<String, Weak<AtomicBool>>>> = OnceLock::new();
+
+#[derive(serde::Deserialize)]
+struct FrameHeader {
+    format: crate::api::schema::PaneGraphicsFormat,
+    image_width: u32,
+    image_height: u32,
+    data_length: usize,
+    #[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 owner = next_owner();
+    params.owner = owner.clone();
+    let open_response = dispatch_to_app_with_timeout(
+        Request {
+            id: request_id.clone(),
+            method: Method::PaneGraphicsStreamOpen(params),
+        },
+        api_tx,
+        Some(open_timeout),
+    );
+    if api_response_outcome(&open_response) != "ok" {
+        let write_result = write_text_line_allow_disconnect(&mut stream, &open_response);
+        clear_layer(&pane_id, &owner, api_tx);
+        write_result?;
+        return Ok(());
+    }
+
+    if let Err(err) = write_json_line(
+        &mut stream,
+        &SuccessResponse {
+            id: request_id.clone(),
+            result: ResponseResult::Ok {},
+        },
+    ) {
+        clear_layer(&pane_id, &owner, api_tx);
+        if is_connection_closed_error(&err) {
+            return Ok(());
+        }
+        return Err(err);
+    }
+
+    let stream_active = Arc::new(AtomicBool::new(true));
+    register_stream(&owner, &stream_active);
+    let result = serve_frames(
+        &mut stream,
+        &request_id,
+        &owner,
+        &pane_id,
+        api_tx,
+        running,
+        &stream_active,
+        read_timeouts,
+    );
+    unregister_stream(&owner);
+    clear_layer(&pane_id, &owner, api_tx);
+    result
+}
+
+fn serve_frames(
+    stream: &mut LocalStream,
+    request_id: &str,
+    owner: &str,
+    pane_id: &str,
+    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 header.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 header.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,
+            header.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(),
+                    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_registry() -> &'static Mutex<HashMap<String, Weak<AtomicBool>>> {
+    REGISTERED_STREAMS.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+fn register_stream(owner: &str, active: &Arc<AtomicBool>) {
+    let Ok(mut streams) = stream_registry().lock() else {
+        active.store(false, Ordering::Release);
+        return;
+    };
+    if streams
+        .insert(owner.to_string(), Arc::downgrade(active))
+        .is_none()
+    {
+        REGISTERED_STREAM_COUNT.fetch_add(1, Ordering::Release);
+    }
+}
+
+fn unregister_stream(owner: &str) {
+    let Ok(mut streams) = stream_registry().lock() else {
+        return;
+    };
+    if streams.remove(owner).is_some() {
+        REGISTERED_STREAM_COUNT.fetch_sub(1, Ordering::Release);
+    }
+}
+
+pub(crate) fn cancel_inactive_streams(mut is_active: impl FnMut(&str) -> bool) {
+    if REGISTERED_STREAM_COUNT.load(Ordering::Acquire) == 0 {
+        return;
+    }
+    let Ok(mut streams) = stream_registry().lock() else {
+        return;
+    };
+    let before = streams.len();
+    streams.retain(|owner, active| {
+        let keep = is_active(owner);
+        if !keep {
+            if let Some(active) = active.upgrade() {
+                active.store(false, Ordering::Release);
+            }
+        }
+        keep
+    });
+    REGISTERED_STREAM_COUNT.fetch_sub(before.saturating_sub(streams.len()), Ordering::Release);
+}
+
+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, 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(),
+                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 stream.read(&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 stream.read(&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 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<T>,
+) -> std::io::Result<T> {
+    match stream.set_recv_timeout(Some(CONNECTION_POLL_INTERVAL)) {
+        Ok(()) => {
+            let result = read(stream, ReadWait::SocketTimeout);
+            merge_read_reset_result(result, stream.set_recv_timeout(None))
+        }
+        Err(err) if err.kind() == io::ErrorKind::Unsupported => {
+            stream.set_nonblocking(true)?;
+            let result = read(stream, ReadWait::Poll(PollBackoff::new()));
+            merge_read_reset_result(result, stream.set_nonblocking(false))
+        }
+        Err(err) => Err(err),
+    }
+}
+
+fn merge_read_reset_result<T>(
+    result: std::io::Result<T>,
+    reset_result: std::io::Result<()>,
+) -> std::io::Result<T> {
+    match (result, reset_result) {
+        (Ok(value), Ok(())) => Ok(value),
+        (Ok(_), Err(err)) => Err(err),
+        (Err(err), _) => 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, 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, Ordering};
+    use std::sync::Arc;
+    use std::time::{Duration, Instant};
+    use tokio::sync::mpsc;
+
+    fn local_stream_pair(name: &str) -> (LocalStream, LocalStream, PathBuf) {
+        let unique = format!(
+            "herdr-{name}-{}-{}",
+            std::process::id(),
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        );
+        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
+    }
+
+    fn assert_server_stream_owner(owner: &str) {
+        assert!(owner.starts_with("pane.graphics.stream:"));
+    }
+
+    #[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:?}"),
+        };
+        open.respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: open.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        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:?}"),
+        }
+        msg.respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: msg.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        drop(client);
+        running.store(false, Ordering::Relaxed);
+        let clear = api_rx.blocking_recv().unwrap();
+        match &clear.request.method {
+            Method::PaneGraphicsStreamClose(params) => {
+                assert_eq!(params.pane_id, "pane_1");
+                assert_eq!(params.owner, stream_owner);
+            }
+            other => panic!("unexpected clear request: {other:?}"),
+        }
+        clear
+            .respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: clear.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+        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 require experimental.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");
+
+        let close = api_rx.blocking_recv().unwrap();
+        match &close.request.method {
+            Method::PaneGraphicsStreamClose(params) => {
+                assert_eq!(params.pane_id, "pane_1");
+                assert_eq!(params.owner, stream_owner);
+            }
+            other => panic!("unexpected close request: {other:?}"),
+        }
+        close
+            .respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: close.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        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(),
+                    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"));
+
+        let close = api_rx.blocking_recv().unwrap();
+        match &close.request.method {
+            Method::PaneGraphicsStreamClose(params) => {
+                assert_eq!(params.pane_id, "pane_1");
+                assert_eq!(params.owner, stream_owner);
+            }
+            other => panic!("unexpected close request: {other:?}"),
+        }
+        close
+            .respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: close.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        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);
+        open.respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: open.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .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(2))
+            .unwrap()
+            .unwrap();
+        match &close.request.method {
+            Method::PaneGraphicsStreamClose(params) => {
+                assert_eq!(params.pane_id, "pane_1");
+                assert_eq!(params.owner, stream_owner);
+            }
+            other => panic!("unexpected close request: {other:?}"),
+        }
+        close
+            .respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: close.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        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 inactive_owner_cancels_idle_stream_and_dispatches_close() {
+        let (mut client, server, _path) = local_stream_pair("graphics-owner-cancel");
+        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-cancel".into(),
+                PaneGraphicsStreamParams {
+                    pane_id: "pane_1".into(),
+                    owner: String::new(),
+                },
+                &api_tx,
+                &server_running,
+                Duration::from_secs(1),
+                READ_TIMEOUTS,
+            )
+        });
+
+        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:?}"),
+        };
+        open.respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: open.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+        let ack: SuccessResponse = serde_json::from_str(&read_response_line(&mut client)).unwrap();
+        assert_eq!(ack.id, "stream-cancel");
+
+        cancel_inactive_streams(|registered| registered != owner);
+
+        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))
+            .expect("canceled idle stream should dispatch a close")
+            .expect("API request channel should remain open");
+        match &close.request.method {
+            Method::PaneGraphicsStreamClose(params) => {
+                assert_eq!(params.pane_id, "pane_1");
+                assert_eq!(params.owner, owner);
+            }
+            other => panic!("unexpected close request: {other:?}"),
+        }
+        close
+            .respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: close.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        drop(client);
+        running.store(false, Ordering::Relaxed);
+        assert!(server_thread.join().unwrap().is_ok());
+    }
+
+    #[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));
+        assert!(started.elapsed() < Duration::from_millis(500));
+    }
+
+    #[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(),
+                    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:?}"),
+        };
+        open.respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: open.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+        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();
+        match &close.request.method {
+            Method::PaneGraphicsStreamClose(params) => {
+                assert_eq!(params.pane_id, "pane_1");
+                assert_eq!(params.owner, owner);
+            }
+            other => panic!("unexpected close request: {other:?}"),
+        }
+        close
+            .respond_to
+            .send(
+                serde_json::to_string(&SuccessResponse {
+                    id: close.request.id,
+                    result: ResponseResult::Ok {},
+                })
+                .unwrap(),
+            )
+            .unwrap();
+
+        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",
+                &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());
+    }
+}