feat: add isolated pane graphics streaming

17 files changed, 3455 insertions, 100 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());
+    }
+}
diff --git a/src/app/actions.rs b/src/app/actions.rs
index 7cf43e79..68286952 100644
--- a/src/app/actions.rs
+++ b/src/app/actions.rs
@@ -1490,6 +1490,8 @@ impl AppState {
         }
         for pane_id in pane_ids {
             self.plugin_panes.remove(&pane_id);
+            self.pane_graphics_layers.remove(&pane_id);
+            self.pane_graphics_streams.remove(&pane_id);
         }
     }
 
@@ -3113,6 +3115,26 @@ mod tests {
         state
     }
 
+    fn insert_test_pane_graphics_layer(state: &mut AppState, pane_id: PaneId) {
+        state.pane_graphics_layers.insert(
+            pane_id,
+            crate::app::state::PaneGraphicsLayer::new(
+                crate::api::schema::PaneGraphicsFormat::Rgba,
+                1,
+                1,
+                vec![1, 2, 3, 4],
+                crate::api::schema::PaneGraphicsPlacementParams::default(),
+            ),
+        );
+    }
+
+    fn insert_test_pane_graphics_state(state: &mut AppState, pane_id: PaneId) {
+        insert_test_pane_graphics_layer(state, pane_id);
+        state
+            .pane_graphics_streams
+            .insert(pane_id, "test-stream".into());
+    }
+
     fn mark_linked_worktree(state: &mut AppState, ws_idx: usize) {
         state.workspaces[ws_idx].worktree_space = Some(crate::workspace::WorktreeSpaceMembership {
             key: "repo-key".into(),
@@ -5196,10 +5218,13 @@ mod tests {
                 entrypoint: "board".into(),
             },
         );
+        insert_test_pane_graphics_state(&mut state, closed);
 
         state.close_pane();
         assert_eq!(state.workspaces[0].panes.len(), 1);
         assert!(!state.plugin_panes.contains_key(&closed));
+        assert!(!state.pane_graphics_layers.contains_key(&closed));
+        assert!(!state.pane_graphics_streams.contains_key(&closed));
         state.assert_invariants_for_test();
     }
 
@@ -5264,11 +5289,14 @@ mod tests {
                 entrypoint: "board".into(),
             },
         );
+        insert_test_pane_graphics_state(&mut state, pane_id);
 
         state.close_tab();
 
         assert!(!state.terminals.contains_key(&terminal_id));
         assert!(!state.plugin_panes.contains_key(&pane_id));
+        assert!(!state.pane_graphics_layers.contains_key(&pane_id));
+        assert!(!state.pane_graphics_streams.contains_key(&pane_id));
         state.assert_invariants_for_test();
     }
 
@@ -5284,11 +5312,14 @@ mod tests {
                 entrypoint: "board".into(),
             },
         );
+        insert_test_pane_graphics_state(&mut state, pane_id);
 
         state.close_selected_workspace();
 
         assert!(!state.terminals.contains_key(&terminal_id));
         assert!(!state.plugin_panes.contains_key(&pane_id));
+        assert!(!state.pane_graphics_layers.contains_key(&pane_id));
+        assert!(!state.pane_graphics_streams.contains_key(&pane_id));
         state.assert_invariants_for_test();
     }
 
diff --git a/src/app/api.rs b/src/app/api.rs
index d6d657b7..d2c2858f 100644
--- a/src/app/api.rs
+++ b/src/app/api.rs
@@ -5,6 +5,7 @@ mod agents;
 mod env;
 mod integrations;
 mod layouts;
+mod pane_graphics;
 mod panes;
 pub(crate) mod plugins;
 mod responses;
@@ -1013,6 +1014,31 @@ impl App {
             Method::PaneFocus(target) => return self.handle_pane_focus(request.id, target),
             Method::PaneRename(params) => return self.handle_pane_rename(request.id, params),
             Method::PaneRead(params) => return self.handle_pane_read(request.id, params),
+            Method::PaneGraphicsSet(params) => {
+                return self.handle_pane_graphics_set(request.id, params);
+            }
+            Method::PaneGraphicsClear(params) => {
+                return self.handle_pane_graphics_clear(request.id, params);
+            }
+            Method::PaneGraphicsInfo(params) => {
+                return self.handle_pane_graphics_info(request.id, params);
+            }
+            Method::PaneGraphicsStream(_) => {
+                return responses::encode_error(
+                    request.id,
+                    "stream_transport_required",
+                    "pane.graphics.stream requires the streaming socket transport",
+                );
+            }
+            Method::PaneGraphicsStreamSet(params) => {
+                return self.handle_pane_graphics_stream_set(request.id, params);
+            }
+            Method::PaneGraphicsStreamOpen(params) => {
+                return self.handle_pane_graphics_stream_open(request.id, params);
+            }
+            Method::PaneGraphicsStreamClose(params) => {
+                return self.handle_pane_graphics_stream_close(request.id, params);
+            }
             Method::PaneReportAgent(params) => {
                 return self.handle_pane_report_agent(request.id, params);
             }
diff --git a/src/app/api/pane_graphics.rs b/src/app/api/pane_graphics.rs
new file mode 100644
index 00000000..7851f69a
--- /dev/null
+++ b/src/app/api/pane_graphics.rs
@@ -0,0 +1,702 @@
+use base64::Engine;
+
+use crate::api::schema::{
+    PaneGraphicsClearParams, PaneGraphicsSetParams, PaneGraphicsStreamParams, ResponseResult,
+    PANE_GRAPHICS_SET_MAX_BYTES, PANE_GRAPHICS_STREAM_MAX_BYTES,
+};
+use crate::app::state::PaneGraphicsLayer;
+use crate::app::App;
+use crate::layout::PaneId;
+
+use super::responses::{encode_error, encode_success};
+
+impl App {
+    pub(super) fn handle_pane_graphics_info(
+        &mut self,
+        id: String,
+        target: crate::api::schema::PaneTarget,
+    ) -> String {
+        if let Err(response) = require_pane_graphics_enabled(self, &id) {
+            return response;
+        }
+        if self.parse_pane_id(&target.pane_id).is_none() {
+            return pane_not_found(id, &target.pane_id);
+        }
+        if !self.state.host_cell_size.is_known() {
+            return encode_error(id, "cell_size_unavailable", "host cell size is unavailable");
+        }
+        encode_success(
+            id,
+            ResponseResult::PaneGraphicsInfo {
+                cell_width_px: self.state.host_cell_size.width_px,
+                cell_height_px: self.state.host_cell_size.height_px,
+            },
+        )
+    }
+
+    pub(super) fn handle_pane_graphics_set(
+        &mut self,
+        id: String,
+        params: PaneGraphicsSetParams,
+    ) -> String {
+        if let Err(response) = require_pane_graphics_enabled(self, &id) {
+            return response;
+        }
+        let Some((_ws_idx, pane_id)) = self.parse_pane_id(&params.pane_id) else {
+            return pane_not_found(id, &params.pane_id);
+        };
+        if self.state.pane_graphics_streams.contains_key(&pane_id) {
+            return encode_error(
+                id,
+                "stream_conflict",
+                "pane already has an active graphics stream",
+            );
+        }
+        self.set_pane_graphics_layer(id, pane_id, params)
+    }
+
+    fn set_pane_graphics_layer(
+        &mut self,
+        id: String,
+        pane_id: PaneId,
+        params: PaneGraphicsSetParams,
+    ) -> String {
+        if params.image_width == 0 || params.image_height == 0 {
+            return encode_error(
+                id,
+                "invalid_image",
+                "image_width and image_height must be greater than zero",
+            );
+        }
+        let max_bytes = if params.data.is_some() {
+            PANE_GRAPHICS_STREAM_MAX_BYTES
+        } else {
+            PANE_GRAPHICS_SET_MAX_BYTES
+        };
+        let data = match params.data {
+            Some(data) => data,
+            None => match base64::engine::general_purpose::STANDARD.decode(params.data_base64) {
+                Ok(data) => data,
+                Err(_) => {
+                    return encode_error(id, "invalid_image", "data_base64 is not valid base64");
+                }
+            },
+        };
+        if data.is_empty() {
+            return encode_error(id, "invalid_image", "image data must not be empty");
+        }
+        if data.len() > max_bytes {
+            return encode_error(id, "image_too_large", "image data is too large");
+        }
+        match pane_graphics_expected_data_len(
+            params.format,
+            params.image_width,
+            params.image_height,
+        ) {
+            Ok(Some(expected_len)) if data.len() != expected_len => {
+                return encode_error(
+                    id,
+                    "invalid_image",
+                    "image data length does not match format and dimensions",
+                );
+            }
+            Ok(_) => {}
+            Err(()) => {
+                return encode_error(id, "invalid_image", "image dimensions are too large");
+            }
+        }
+
+        let layer = PaneGraphicsLayer::new(
+            params.format,
+            params.image_width,
+            params.image_height,
+            data,
+            params.placement,
+        );
+        self.state.pane_graphics_layers.insert(pane_id, layer);
+        self.state.pane_graphics_revision = self.state.pane_graphics_revision.wrapping_add(1);
+
+        encode_success(id, ResponseResult::Ok {})
+    }
+
+    pub(super) fn handle_pane_graphics_clear(
+        &mut self,
+        id: String,
+        params: PaneGraphicsClearParams,
+    ) -> String {
+        if let Err(response) = require_pane_graphics_enabled(self, &id) {
+            return response;
+        }
+        let Some((_ws_idx, pane_id)) = self.parse_pane_id(&params.pane_id) else {
+            return pane_not_found(id, &params.pane_id);
+        };
+        if self.state.pane_graphics_streams.contains_key(&pane_id) {
+            return encode_error(
+                id,
+                "stream_conflict",
+                "pane already has an active graphics stream",
+            );
+        }
+        if self.state.pane_graphics_layers.remove(&pane_id).is_some() {
+            self.state.pane_graphics_revision = self.state.pane_graphics_revision.wrapping_add(1);
+        }
+
+        encode_success(id, ResponseResult::Ok {})
+    }
+
+    pub(super) fn handle_pane_graphics_stream_set(
+        &mut self,
+        id: String,
+        params: PaneGraphicsSetParams,
+    ) -> String {
+        if let Err(response) = require_pane_graphics_enabled(self, &id) {
+            return response;
+        }
+        let Some((_ws_idx, pane_id)) = self.parse_pane_id(&params.pane_id) else {
+            return pane_not_found(id, &params.pane_id);
+        };
+        match self.state.pane_graphics_streams.get(&pane_id) {
+            Some(owner) if owner == &params.owner => {}
+            Some(_) => {
+                return encode_error(
+                    id,
+                    "stream_conflict",
+                    "pane graphics stream owner does not match active stream",
+                );
+            }
+            None => return encode_error(id, "stream_closed", "pane graphics stream is not active"),
+        }
+        self.set_pane_graphics_layer(id, pane_id, params)
+    }
+
+    pub(super) fn handle_pane_graphics_stream_open(
+        &mut self,
+        id: String,
+        params: PaneGraphicsStreamParams,
+    ) -> String {
+        if let Err(response) = require_pane_graphics_enabled(self, &id) {
+            return response;
+        }
+        if params.owner.is_empty() {
+            return encode_error(
+                id,
+                "invalid_stream",
+                "pane graphics stream owner is required",
+            );
+        }
+        let Some((_ws_idx, pane_id)) = self.parse_pane_id(&params.pane_id) else {
+            return pane_not_found(id, &params.pane_id);
+        };
+        if self.state.pane_graphics_streams.contains_key(&pane_id) {
+            return encode_error(
+                id,
+                "stream_conflict",
+                "pane already has an active graphics stream",
+            );
+        }
+        self.state
+            .pane_graphics_streams
+            .insert(pane_id, params.owner);
+        self.state.pane_graphics_layers.remove(&pane_id);
+        self.state.pane_graphics_revision = self.state.pane_graphics_revision.wrapping_add(1);
+
+        encode_success(id, ResponseResult::Ok {})
+    }
+
+    pub(super) fn handle_pane_graphics_stream_close(
+        &mut self,
+        id: String,
+        params: PaneGraphicsStreamParams,
+    ) -> String {
+        let Some((_ws_idx, pane_id)) = self.parse_pane_id(&params.pane_id) else {
+            return pane_not_found(id, &params.pane_id);
+        };
+        if self
+            .state
+            .pane_graphics_streams
+            .get(&pane_id)
+            .is_some_and(|owner| owner == &params.owner)
+        {
+            self.state.pane_graphics_streams.remove(&pane_id);
+            self.state.pane_graphics_layers.remove(&pane_id);
+            self.state.pane_graphics_revision = self.state.pane_graphics_revision.wrapping_add(1);
+        }
+
+        encode_success(id, ResponseResult::Ok {})
+    }
+}
+
+fn require_pane_graphics_enabled(app: &App, id: &str) -> Result<(), String> {
+    if app.state.kitty_graphics_enabled {
+        return Ok(());
+    }
+    Err(encode_error(
+        id.to_owned(),
+        "feature_disabled",
+        "pane graphics require experimental.kitty_graphics",
+    ))
+}
+
+fn pane_graphics_expected_data_len(
+    format: crate::api::schema::PaneGraphicsFormat,
+    image_width: u32,
+    image_height: u32,
+) -> Result<Option<usize>, ()> {
+    let bytes_per_pixel = match format {
+        crate::api::schema::PaneGraphicsFormat::Png => return Ok(None),
+        crate::api::schema::PaneGraphicsFormat::Rgb => 3_u64,
+        crate::api::schema::PaneGraphicsFormat::Rgba => 4_u64,
+    };
+    let pixels = u64::from(image_width)
+        .checked_mul(u64::from(image_height))
+        .ok_or(())?;
+    pixels
+        .checked_mul(bytes_per_pixel)
+        .and_then(|bytes| usize::try_from(bytes).ok())
+        .map(Some)
+        .ok_or(())
+}
+
+fn pane_not_found(id: String, pane_id: &str) -> String {
+    encode_error(id, "pane_not_found", format!("pane {pane_id} not found"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::api::schema::{ErrorResponse, SuccessResponse};
+    use crate::config::Config;
+    use crate::workspace::Workspace;
+
+    fn app_with_test_workspace() -> (App, String) {
+        let (_api_tx, api_rx) = tokio::sync::mpsc::unbounded_channel();
+        let mut app = App::new(
+            &Config::default(),
+            true,
+            None,
+            api_rx,
+            crate::api::EventHub::default(),
+        );
+        app.state.workspaces = vec![Workspace::test_new("pane-graphics")];
+        app.state.ensure_test_terminals();
+        let pane_id = app.state.workspaces[0].tabs[0].root_pane;
+        let public_pane_id = app.public_pane_id(0, pane_id).unwrap();
+        (app, public_pane_id)
+    }
+
+    #[test]
+    fn api_pane_graphics_info_is_explicit_and_side_effect_free() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        let request = |pane_id: String| crate::api::schema::Request {
+            id: "graphics-info".into(),
+            method: crate::api::schema::Method::PaneGraphicsInfo(crate::api::schema::PaneTarget {
+                pane_id,
+            }),
+        };
+
+        let response = app.handle_api_request(request(public_pane_id.clone()));
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "feature_disabled");
+
+        app.state.kitty_graphics_enabled = true;
+        app.state.host_cell_size = crate::kitty_graphics::HostCellSize {
+            width_px: 11,
+            height_px: 22,
+        };
+        let response = app.handle_api_request(request(public_pane_id));
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(
+            success.result,
+            ResponseResult::PaneGraphicsInfo {
+                cell_width_px: 11,
+                cell_height_px: 22,
+            }
+        ));
+        assert!(app.state.pane_graphics_layers.is_empty());
+        assert!(app.state.pane_graphics_streams.is_empty());
+    }
+
+    #[test]
+    fn api_pane_graphics_set_and_clear_updates_runtime_layers() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        app.state.kitty_graphics_enabled = true;
+        let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+        let data = base64::engine::general_purpose::STANDARD.encode([1_u8, 2, 3, 4]);
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-set".into(),
+            method: crate::api::schema::Method::PaneGraphicsSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: String::new(),
+                format: crate::api::schema::PaneGraphicsFormat::Rgba,
+                image_width: 1,
+                image_height: 1,
+                data: None,
+                data_base64: data,
+                placement: crate::api::schema::PaneGraphicsPlacementParams {
+                    grid_cols: 10,
+                    grid_rows: 4,
+                    ..Default::default()
+                },
+            }),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+
+        let layer = app
+            .state
+            .pane_graphics_layers
+            .get(&pane_id)
+            .expect("graphics layer stored");
+        assert_eq!(layer.image_width, 1);
+        assert_eq!(layer.image_height, 1);
+        assert_eq!(layer.render.grid_cols, 10);
+        assert_eq!(layer.render.grid_rows, 4);
+        assert!(!app.state.session_dirty);
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-clear".into(),
+            method: crate::api::schema::Method::PaneGraphicsClear(PaneGraphicsClearParams {
+                pane_id: public_pane_id,
+            }),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn api_pane_graphics_rejects_when_kitty_graphics_disabled() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+        let data = base64::engine::general_purpose::STANDARD.encode([1_u8, 2, 3, 4]);
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-set".into(),
+            method: crate::api::schema::Method::PaneGraphicsSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: String::new(),
+                format: crate::api::schema::PaneGraphicsFormat::Rgba,
+                image_width: 1,
+                image_height: 1,
+                data: None,
+                data_base64: data,
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        });
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "feature_disabled");
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-clear".into(),
+            method: crate::api::schema::Method::PaneGraphicsClear(PaneGraphicsClearParams {
+                pane_id: public_pane_id,
+            }),
+        });
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "feature_disabled");
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn api_pane_graphics_stream_claim_uses_resolved_pane_identity() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        app.state.kitty_graphics_enabled = true;
+        let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+        let alias = format!("{public_pane_id}:alias");
+        app.state
+            .public_pane_id_aliases
+            .insert(alias.clone(), pane_id);
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-open".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamOpen(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: public_pane_id,
+                    owner: "stream-1".into(),
+                },
+            ),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+        assert_eq!(
+            app.state.pane_graphics_streams.get(&pane_id),
+            Some(&"stream-1".to_string())
+        );
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-open-alias".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamOpen(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: alias.clone(),
+                    owner: "stream-2".into(),
+                },
+            ),
+        });
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "stream_conflict");
+        assert_eq!(
+            app.state.pane_graphics_streams.get(&pane_id),
+            Some(&"stream-1".to_string())
+        );
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-close-wrong-owner".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamClose(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: alias.clone(),
+                    owner: "stream-2".into(),
+                },
+            ),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+        assert_eq!(
+            app.state.pane_graphics_streams.get(&pane_id),
+            Some(&"stream-1".to_string())
+        );
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-close-alias".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamClose(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: alias,
+                    owner: "stream-1".into(),
+                },
+            ),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+        assert!(!app.state.pane_graphics_streams.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn api_pane_graphics_stream_owns_layer_while_active() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        app.state.kitty_graphics_enabled = true;
+        let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+        let public_data = base64::engine::general_purpose::STANDARD.encode([1_u8, 2, 3, 4]);
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-open".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamOpen(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: public_pane_id.clone(),
+                    owner: "stream-1".into(),
+                },
+            ),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-set-public".into(),
+            method: crate::api::schema::Method::PaneGraphicsSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: String::new(),
+                format: crate::api::schema::PaneGraphicsFormat::Rgba,
+                image_width: 1,
+                image_height: 1,
+                data: None,
+                data_base64: public_data.clone(),
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        });
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "stream_conflict");
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-set-stream".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: "stream-1".into(),
+                format: crate::api::schema::PaneGraphicsFormat::Rgba,
+                image_width: 1,
+                image_height: 1,
+                data: None,
+                data_base64: public_data,
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+        assert!(app.state.pane_graphics_layers.contains_key(&pane_id));
+
+        let wrong_owner_data = base64::engine::general_purpose::STANDARD.encode([5_u8, 6, 7, 8]);
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-set-stream-wrong-owner".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: "stream-2".into(),
+                format: crate::api::schema::PaneGraphicsFormat::Rgba,
+                image_width: 1,
+                image_height: 1,
+                data: None,
+                data_base64: wrong_owner_data,
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        });
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "stream_conflict");
+        assert_eq!(
+            app.state.pane_graphics_layers.get(&pane_id).unwrap().data,
+            vec![1_u8, 2, 3, 4]
+        );
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-clear-public".into(),
+            method: crate::api::schema::Method::PaneGraphicsClear(PaneGraphicsClearParams {
+                pane_id: public_pane_id.clone(),
+            }),
+        });
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "stream_conflict");
+        assert!(app.state.pane_graphics_layers.contains_key(&pane_id));
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-close".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamClose(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: public_pane_id,
+                    owner: "stream-1".into(),
+                },
+            ),
+        });
+        let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+        assert!(matches!(success.result, ResponseResult::Ok {}));
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn api_pane_graphics_rejects_overflowing_raw_dimensions() {
+        for format in [
+            crate::api::schema::PaneGraphicsFormat::Rgb,
+            crate::api::schema::PaneGraphicsFormat::Rgba,
+        ] {
+            let (mut app, public_pane_id) = app_with_test_workspace();
+            app.state.kitty_graphics_enabled = true;
+            let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+
+            let response = app.handle_api_request(crate::api::schema::Request {
+                id: "graphics-overflow".into(),
+                method: crate::api::schema::Method::PaneGraphicsSet(PaneGraphicsSetParams {
+                    pane_id: public_pane_id,
+                    owner: String::new(),
+                    format,
+                    image_width: u32::MAX,
+                    image_height: u32::MAX,
+                    data: None,
+                    data_base64: base64::engine::general_purpose::STANDARD.encode([1_u8]),
+                    placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+                }),
+            });
+
+            let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+            assert_eq!(error.error.code, "invalid_image");
+            assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+        }
+    }
+
+    #[test]
+    fn api_pane_graphics_validates_exact_raw_lengths() {
+        for (format, exact_len) in [
+            (crate::api::schema::PaneGraphicsFormat::Rgb, 6_usize),
+            (crate::api::schema::PaneGraphicsFormat::Rgba, 8_usize),
+        ] {
+            for data_len in [exact_len, exact_len - 1] {
+                let (mut app, public_pane_id) = app_with_test_workspace();
+                app.state.kitty_graphics_enabled = true;
+                let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+                let response = app.handle_api_request(crate::api::schema::Request {
+                    id: "graphics-raw-length".into(),
+                    method: crate::api::schema::Method::PaneGraphicsSet(PaneGraphicsSetParams {
+                        pane_id: public_pane_id,
+                        owner: String::new(),
+                        format,
+                        image_width: 2,
+                        image_height: 1,
+                        data: None,
+                        data_base64: base64::engine::general_purpose::STANDARD
+                            .encode(vec![1_u8; data_len]),
+                        placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+                    }),
+                });
+
+                if data_len == exact_len {
+                    let success: SuccessResponse = serde_json::from_str(&response).unwrap();
+                    assert!(matches!(success.result, ResponseResult::Ok {}));
+                    assert!(app.state.pane_graphics_layers.contains_key(&pane_id));
+                } else {
+                    let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+                    assert_eq!(error.error.code, "invalid_image");
+                    assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn api_pane_graphics_uses_transport_specific_limits() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        app.state.kitty_graphics_enabled = true;
+        let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-public-too-large".into(),
+            method: crate::api::schema::Method::PaneGraphicsSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id,
+                owner: String::new(),
+                format: crate::api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data: None,
+                data_base64: base64::engine::general_purpose::STANDARD.encode(vec![
+                    1_u8;
+                    PANE_GRAPHICS_SET_MAX_BYTES
+                        + 1
+                ]),
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        });
+
+        let error: ErrorResponse = serde_json::from_str(&response).unwrap();
+        assert_eq!(error.error.code, "image_too_large");
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn api_pane_graphics_stream_methods_require_kitty_graphics() {
+        let (mut app, public_pane_id) = app_with_test_workspace();
+        let (_, pane_id) = app.parse_pane_id(&public_pane_id).unwrap();
+        let open = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-open".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamOpen(
+                crate::api::schema::PaneGraphicsStreamParams {
+                    pane_id: public_pane_id.clone(),
+                    owner: "stream-1".into(),
+                },
+            ),
+        });
+        let error: ErrorResponse = serde_json::from_str(&open).unwrap();
+        assert_eq!(error.error.code, "feature_disabled");
+
+        let set = app.handle_api_request(crate::api::schema::Request {
+            id: "graphics-stream-set".into(),
+            method: crate::api::schema::Method::PaneGraphicsStreamSet(PaneGraphicsSetParams {
+                pane_id: public_pane_id,
+                owner: "stream-1".into(),
+                format: crate::api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data: Some(vec![1_u8]),
+                data_base64: String::new(),
+                placement: crate::api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        });
+        let error: ErrorResponse = serde_json::from_str(&set).unwrap();
+        assert_eq!(error.error.code, "feature_disabled");
+        assert!(!app.state.pane_graphics_streams.contains_key(&pane_id));
+        assert!(!app.state.pane_graphics_layers.contains_key(&pane_id));
+    }
+}
diff --git a/src/app/api/plugins/mod.rs b/src/app/api/plugins/mod.rs
index 1e0ec7d8..b7f90df9 100644
--- a/src/app/api/plugins/mod.rs
+++ b/src/app/api/plugins/mod.rs
@@ -1164,6 +1164,11 @@ command = ["echo", "b"]
         app.state.active = Some(0);
         app.state.selected = 0;
         app.state.mode = crate::app::Mode::Terminal;
+        app.state.kitty_graphics_enabled = true;
+        app.state.host_cell_size = crate::kitty_graphics::HostCellSize {
+            width_px: 11,
+            height_px: 22,
+        };
         app.state.terminals.get_mut(&root_terminal).unwrap().cwd = "/tmp".into();
         let target_public_pane_id = app.public_pane_id(0, root_pane).unwrap();
 
@@ -1182,7 +1187,7 @@ platforms = ["linux", "macos"]
 [[panes]]
 id = "board"
 title = "Plugin Board"
-command = ["sh", "-c", "printf '%s\n%s\n%s\n%s\n%s\n%s\n%s\n' \"$PWD\" \"$HERDR_PLUGIN_ID\" \"$HERDR_PLUGIN_ENTRYPOINT_ID\" \"$HERDR_WORKSPACE_ID\" \"$HERDR_PANE_ID\" \"$HERDR_BIN_PATH\" \"$HERDR_PLUGIN_CONTEXT_JSON\" > {}"]
+command = ["sh", "-c", "printf '%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n' \"$PWD\" \"$HERDR_PLUGIN_ID\" \"$HERDR_PLUGIN_ENTRYPOINT_ID\" \"$HERDR_WORKSPACE_ID\" \"$HERDR_PANE_ID\" \"$HERDR_BIN_PATH\" \"$HERDR_PLUGIN_CONTEXT_JSON\" \"${{HERDR_CELL_WIDTH_PX-unset}}\" \"${{HERDR_CELL_HEIGHT_PX-unset}}\" > {}"]
 "#,
                 capture.display()
             ),
@@ -1254,6 +1259,8 @@ command = ["sh", "-c", "printf '%s\n%s\n%s\n%s\n%s\n%s\n%s\n' \"$PWD\" \"$HERDR_
             context.focused_pane_id.as_deref(),
             Some(target_public_pane_id.as_str())
         );
+        assert_eq!(lines.next(), Some("unset"));
+        assert_eq!(lines.next(), Some("unset"));
 
         for (_, runtime) in app.terminal_runtimes.drain() {
             runtime.shutdown();
diff --git a/src/app/mod.rs b/src/app/mod.rs
index f32d7785..7391d7d3 100644
--- a/src/app/mod.rs
+++ b/src/app/mod.rs
@@ -656,11 +656,15 @@ impl App {
             integration_install_messages: Vec::new(),
             installed_plugins: load_plugin_registry(no_session),
             plugin_panes: std::collections::HashMap::new(),
+            pane_graphics_layers: std::collections::HashMap::new(),
+            pane_graphics_streams: std::collections::HashMap::new(),
+            pane_graphics_revision: 0,
             plugin_command_logs: Vec::new(),
             next_plugin_command_log_id: 1,
             plugin_commands_in_flight: 0,
             global_menu: state::MenuListState::new(0),
             host_terminal_theme: crate::terminal_theme::TerminalTheme::default(),
+            host_cell_size: crate::kitty_graphics::HostCellSize::default(),
             session_dirty: false,
             terminal_runtime_shutdowns: Vec::new(),
         };
@@ -1026,7 +1030,14 @@ impl App {
                 terminal.draw(|frame| {
                     let area = frame.area();
                     if kitty_graphics_enabled {
-                        cell_size = crate::kitty_graphics::HostCellSize::from_terminal(area);
+                        let observed_cell_size =
+                            crate::kitty_graphics::HostCellSize::try_from_terminal(area);
+                        if let Some(observed_cell_size) = observed_cell_size {
+                            self.state.host_cell_size = observed_cell_size;
+                        }
+                        cell_size = observed_cell_size.unwrap_or_else(|| {
+                            crate::kitty_graphics::HostCellSize::fallback_for_area(area)
+                        });
                         crate::ui::compute_view_with_cell_size(
                             &mut self.state,
                             &self.terminal_runtimes,
@@ -1426,6 +1437,9 @@ impl App {
             crate::kitty_graphics::set_enabled(config.experimental.kitty_graphics);
             if was_kitty_graphics_enabled && !config.experimental.kitty_graphics {
                 let _ = crate::kitty_graphics::clear_all_host_graphics();
+                self.state.pane_graphics_layers.clear();
+                self.state.pane_graphics_streams.clear();
+                self.state.host_cell_size = crate::kitty_graphics::HostCellSize::default();
             }
             self.state.reveal_hidden_cursor_for_cjk_ime =
                 config.experimental.reveal_hidden_cursor_for_cjk_ime;
diff --git a/src/app/state.rs b/src/app/state.rs
index d2315422..c0f2c8bb 100644
--- a/src/app/state.rs
+++ b/src/app/state.rs
@@ -2,6 +2,7 @@ use crate::config::{Keybinds, NewTerminalCwdConfig, SoundConfig, ToastConfig, To
 use crossterm::event::{KeyCode, KeyModifiers};
 use ratatui::layout::{Direction, Rect};
 use ratatui::style::Color;
+use std::hash::{Hash, Hasher};
 
 use crate::detect::AgentState;
 use crate::layout::{PaneId, PaneInfo, SplitBorder};
@@ -16,6 +17,42 @@ pub(crate) struct PluginPaneRecord {
     pub entrypoint: String,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct PaneGraphicsLayer {
+    pub format: crate::api::schema::PaneGraphicsFormat,
+    pub image_width: u32,
+    pub image_height: u32,
+    pub data: Vec<u8>,
+    pub data_fingerprint: u64,
+    pub render: crate::api::schema::PaneGraphicsPlacementParams,
+}
+
+impl PaneGraphicsLayer {
+    pub(crate) fn new(
+        format: crate::api::schema::PaneGraphicsFormat,
+        image_width: u32,
+        image_height: u32,
+        data: Vec<u8>,
+        render: crate::api::schema::PaneGraphicsPlacementParams,
+    ) -> Self {
+        let data_fingerprint = pane_graphics_data_fingerprint(&data);
+        Self {
+            format,
+            image_width,
+            image_height,
+            data,
+            data_fingerprint,
+            render,
+        }
+    }
+}
+
+fn pane_graphics_data_fingerprint(data: &[u8]) -> u64 {
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    data.hash(&mut hasher);
+    hasher.finish()
+}
+
 // ---------------------------------------------------------------------------
 // Selection autoscroll types
 // ---------------------------------------------------------------------------
@@ -1472,6 +1509,12 @@ pub struct AppState {
     pub(crate) installed_plugins: InstalledPluginRegistry,
     /// Pane ids opened through the plugin pane API.
     pub(crate) plugin_panes: std::collections::HashMap<PaneId, PluginPaneRecord>,
+    /// Runtime image layers owned by API clients and composited over panes.
+    pub(crate) pane_graphics_layers: std::collections::HashMap<PaneId, PaneGraphicsLayer>,
+    /// Active streaming graphics owner token by pane id.
+    pub(crate) pane_graphics_streams: std::collections::HashMap<PaneId, String>,
+    /// Monotonic marker for accepted pane graphics mutations.
+    pub(crate) pane_graphics_revision: u64,
     /// Recent plugin action/event command executions.
     pub(crate) plugin_command_logs: Vec<crate::api::schema::PluginCommandLogInfo>,
     pub(crate) next_plugin_command_log_id: u64,
@@ -1480,6 +1523,8 @@ pub struct AppState {
     pub global_menu: MenuListState,
     /// Resolved host terminal default colors for theming embedded panes.
     pub host_terminal_theme: TerminalTheme,
+    /// Last known foreground host terminal cell size in pixels.
+    pub(crate) host_cell_size: crate::kitty_graphics::HostCellSize,
     /// Set when a persisted session snapshot would change.
     pub session_dirty: bool,
     /// Terminal runtimes that should be shut down by the app/runtime layer
@@ -1826,11 +1871,15 @@ impl AppState {
             integration_install_messages: Vec::new(),
             installed_plugins: std::collections::HashMap::new(),
             plugin_panes: std::collections::HashMap::new(),
+            pane_graphics_layers: std::collections::HashMap::new(),
+            pane_graphics_streams: std::collections::HashMap::new(),
+            pane_graphics_revision: 0,
             plugin_command_logs: Vec::new(),
             next_plugin_command_log_id: 1,
             plugin_commands_in_flight: 0,
             global_menu: MenuListState::new(0),
             host_terminal_theme: TerminalTheme::default(),
+            host_cell_size: crate::kitty_graphics::HostCellSize::default(),
             session_dirty: false,
             terminal_runtime_shutdowns: Vec::new(),
         }
diff --git a/src/kitty_graphics.rs b/src/kitty_graphics.rs
index 2e83841c..9cbe4c2f 100644
--- a/src/kitty_graphics.rs
+++ b/src/kitty_graphics.rs
@@ -11,12 +11,15 @@ use ratatui::layout::Rect;
 
 use crate::app::state::AppState;
 use crate::app::Mode;
-use crate::ghostty::{KittyImageDescriptor, KittyImageFormat, KittyImagePlacement};
-use crate::layout::PaneId;
+use crate::ghostty::{
+    KittyImageDescriptor, KittyImageFormat, KittyImagePlacement, KittyPlacementRenderInfo,
+};
+use crate::layout::{PaneId, PaneInfo};
 use crate::terminal::TerminalRuntimeRegistry;
 
 const KITTY_CHUNK_BYTES: usize = 3072;
 const HOST_IMAGE_ID_BASE: u32 = 10_000;
+const PANE_GRAPHICS_IMAGE_ID_BIT: u32 = 1 << 31;
 
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub(crate) struct HostCellSize {
@@ -25,28 +28,27 @@ pub(crate) struct HostCellSize {
 }
 
 impl HostCellSize {
-    pub(crate) fn from_terminal(area: Rect) -> Self {
+    pub(crate) fn try_from_terminal(area: Rect) -> Option<Self> {
         let Ok(size) = crossterm::terminal::window_size() else {
-            return Self::fallback_for_area(area);
+            return None;
         };
-        if size.columns == 0 || size.rows == 0 {
-            return Self::fallback_for_area(area);
+        if size.columns == 0 || size.rows == 0 || size.width == 0 || size.height == 0 {
+            return None;
         }
-        if size.width == 0 || size.height == 0 {
-            return Self::fallback_for_area(area);
-        }
-        Self {
-            width_px: (size.width as u32 / size.columns as u32).max(1),
-            height_px: (size.height as u32 / size.rows as u32).max(1),
-        }
-        .for_area(area)
+        Some(
+            Self {
+                width_px: (size.width as u32 / size.columns as u32).max(1),
+                height_px: (size.height as u32 / size.rows as u32).max(1),
+            }
+            .for_area(area),
+        )
     }
 
     pub(crate) fn is_known(self) -> bool {
         self.width_px > 0 && self.height_px > 0
     }
 
-    fn fallback_for_area(area: Rect) -> Self {
+    pub(crate) fn fallback_for_area(area: Rect) -> Self {
         Self {
             width_px: 8,
             height_px: 16,
@@ -73,11 +75,18 @@ struct HostPlacement {
     pane_id: PaneId,
     area: Rect,
     cell_size: HostCellSize,
+    source_key: HostSourceKey,
     placement: KittyImagePlacement,
     scrollback_offset: u32,
 }
 
 #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
+enum HostSourceKey {
+    Terminal { pane_id: PaneId, image_id: u32 },
+    PaneLayer { pane_id: PaneId },
+}
+
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 struct ImageSignature {
     image_width: u32,
     image_height: u32,
@@ -121,7 +130,7 @@ pub(crate) struct HostGraphicsCache {
     images: HashMap<u32, ImageSignature>,
     placements: HashMap<(u32, u32), PlacementSignature>,
     /// Host image currently backing each (pane, source image id) pair.
-    sources: HashMap<(PaneId, u32), u32>,
+    sources: HashMap<HostSourceKey, u32>,
     view: Option<HostViewKey>,
 }
 
@@ -190,9 +199,7 @@ pub(crate) fn encode_local_pane_graphics(
     }
 
     let view_key = active_view_key(app);
-    let uploaded_images = cache.images.clone();
-    let placements =
-        collect_visible_placements(app, terminal_runtimes, cell_size, &uploaded_images);
+    let placements = collect_visible_placements(app, terminal_runtimes, cell_size, &cache.images);
     tracing::debug!(
         placements_collected = placements.len(),
         "collect_visible_placements result"
@@ -240,24 +247,36 @@ pub(crate) fn has_visible_pane_graphics(
     }
 
     for info in &app.view.pane_infos {
-        let Some(runtime) = app.runtime_for_pane_in_workspace(terminal_runtimes, ws_idx, info.id)
-        else {
-            continue;
-        };
-        let scrollback_offset = runtime
-            .scroll_metrics()
-            .map(|m| m.offset_from_bottom as u32)
-            .unwrap_or(0);
-        for placement in runtime.kitty_image_placements_with_data_filter(|_| false) {
-            let host_placement = HostPlacement {
-                pane_id: info.id,
-                area: info.inner_rect,
-                cell_size,
-                placement,
-                scrollback_offset,
-            };
-            if clipped_placement(&host_placement).is_some() {
-                return true;
+        let empty_uploaded = HashMap::new();
+        if app.pane_graphics_layers.get(&info.id).is_some_and(|layer| {
+            let host_placement =
+                pane_graphics_host_placement(info, cell_size, layer, &empty_uploaded, false);
+            clipped_placement(&host_placement).is_some()
+        }) {
+            return true;
+        }
+
+        if let Some(runtime) = app.runtime_for_pane_in_workspace(terminal_runtimes, ws_idx, info.id)
+        {
+            let scrollback_offset = runtime
+                .scroll_metrics()
+                .map(|m| m.offset_from_bottom as u32)
+                .unwrap_or(0);
+            for placement in runtime.kitty_image_placements_with_data_filter(|_| false) {
+                let host_placement = HostPlacement {
+                    pane_id: info.id,
+                    area: info.inner_rect,
+                    cell_size,
+                    source_key: HostSourceKey::Terminal {
+                        pane_id: info.id,
+                        image_id: placement.image_id,
+                    },
+                    placement,
+                    scrollback_offset,
+                };
+                if clipped_placement(&host_placement).is_some() {
+                    return true;
+                }
             }
         }
     }
@@ -270,15 +289,34 @@ fn encode_graphics_update(
     view_changed: bool,
     host_images: &mut HashMap<u32, ImageSignature>,
     host_placements: &mut HashMap<(u32, u32), PlacementSignature>,
-    sources: &mut HashMap<(PaneId, u32), u32>,
+    sources: &mut HashMap<HostSourceKey, u32>,
 ) {
     // Prune sources that are no longer visible: a stale entry would keep its
     // old host image referenced and block the superseded-image delete.
-    let current_sources: HashSet<(PaneId, u32)> = placements
+    let current_sources: HashSet<HostSourceKey> = placements
         .iter()
-        .map(|placement| (placement.pane_id, placement.placement.image_id))
+        .filter(|placement| {
+            matches!(placement.source_key, HostSourceKey::Terminal { .. })
+                || clipped_placement(placement).is_some()
+        })
+        .map(|placement| placement.source_key)
         .collect();
-    sources.retain(|source, _| current_sources.contains(source));
+    let mut removed_pane_layer_images = HashSet::new();
+    sources.retain(|source, host_id| {
+        let retain = current_sources.contains(source);
+        if !retain && matches!(source, HostSourceKey::PaneLayer { .. }) {
+            removed_pane_layer_images.insert(*host_id);
+        }
+        retain
+    });
+    for host_id in removed_pane_layer_images {
+        if sources.values().any(|id| *id == host_id) {
+            continue;
+        }
+        encode_delete_image(bytes, host_id);
+        host_images.remove(&host_id);
+        host_placements.retain(|(image_id, _), _| *image_id != host_id);
+    }
 
     let mut current_placements = HashSet::new();
     for placement in placements {
@@ -298,7 +336,7 @@ fn encode_graphics_update(
             continue;
         };
         let host_id = host_image_id(placement.pane_id, &placement.placement);
-        let host_placement_id = host_placement_id(placement.pane_id, &placement.placement);
+        let host_placement_id = host_placement_id(placement.source_key, &placement.placement);
         let image_signature = image_signature(placement, format_code);
         let placement_signature =
             placement_signature(clipped, placement.placement.z, placement.scrollback_offset);
@@ -336,7 +374,7 @@ fn encode_graphics_update(
             host_images,
             host_placements,
             &mut current_placements,
-            (placement.pane_id, placement.placement.image_id),
+            placement.source_key,
             host_id,
         );
 
@@ -385,11 +423,11 @@ fn encode_graphics_update(
 /// image it previously pointed at once no other source references it.
 fn release_superseded_source_image(
     bytes: &mut Vec<u8>,
-    sources: &mut HashMap<(PaneId, u32), u32>,
+    sources: &mut HashMap<HostSourceKey, u32>,
     host_images: &mut HashMap<u32, ImageSignature>,
     host_placements: &mut HashMap<(u32, u32), PlacementSignature>,
     current_placements: &mut HashSet<(u32, u32)>,
-    source: (PaneId, u32),
+    source: HostSourceKey,
     host_id: u32,
 ) {
     let Some(previous) = sources.insert(source, host_id) else {
@@ -505,6 +543,16 @@ fn collect_visible_placements(
     );
     let mut placements = Vec::new();
     for info in &app.view.pane_infos {
+        if let Some(layer) = app.pane_graphics_layers.get(&info.id) {
+            placements.push(pane_graphics_host_placement(
+                info,
+                cell_size,
+                layer,
+                uploaded_images,
+                true,
+            ));
+        }
+
         let runtime = match app.runtime_for_pane_in_workspace(terminal_runtimes, ws_idx, info.id) {
             Some(rt) => rt,
             None => {
@@ -526,6 +574,10 @@ fn collect_visible_placements(
                 pane_id: info.id,
                 area: info.inner_rect,
                 cell_size,
+                source_key: HostSourceKey::Terminal {
+                    pane_id: info.id,
+                    image_id: placement.image_id,
+                },
                 placement,
                 scrollback_offset,
             });
@@ -538,6 +590,82 @@ fn collect_visible_placements(
     placements
 }
 
+fn pane_graphics_host_placement(
+    info: &PaneInfo,
+    cell_size: HostCellSize,
+    layer: &crate::app::state::PaneGraphicsLayer,
+    uploaded_images: &HashMap<u32, ImageSignature>,
+    include_data: bool,
+) -> HostPlacement {
+    let format = pane_graphics_kitty_format(layer.format);
+    let format_code = kitty_format_code(format);
+    let signature = ImageSignature {
+        image_width: layer.image_width,
+        image_height: layer.image_height,
+        format_code,
+        data_len: layer.data.len(),
+        data_fingerprint: layer.data_fingerprint,
+    };
+    let host_id = pane_graphics_host_image_id(info.id, signature);
+    let data = if !include_data || uploaded_images.get(&host_id).copied() == Some(signature) {
+        Vec::new()
+    } else {
+        layer.data.clone()
+    };
+    let render = layer.render;
+    let grid_cols = if render.grid_cols == 0 {
+        u32::from(info.inner_rect.width)
+    } else {
+        render.grid_cols
+    };
+    let grid_rows = if render.grid_rows == 0 {
+        u32::from(info.inner_rect.height)
+    } else {
+        render.grid_rows
+    };
+
+    HostPlacement {
+        pane_id: info.id,
+        area: info.inner_rect,
+        cell_size,
+        source_key: HostSourceKey::PaneLayer { pane_id: info.id },
+        scrollback_offset: 0,
+        placement: KittyImagePlacement {
+            image_id: 1,
+            placement_id: 1,
+            z: 0,
+            x_offset: 0,
+            y_offset: 0,
+            image_width: layer.image_width,
+            image_height: layer.image_height,
+            format,
+            data_len: layer.data.len(),
+            data_fingerprint: layer.data_fingerprint,
+            data,
+            render: KittyPlacementRenderInfo {
+                pixel_width: layer.image_width,
+                pixel_height: layer.image_height,
+                grid_cols,
+                grid_rows,
+                viewport_col: render.viewport_col,
+                viewport_row: render.viewport_row,
+                source_x: 0,
+                source_y: 0,
+                source_width: 0,
+                source_height: 0,
+            },
+        },
+    }
+}
+
+fn pane_graphics_kitty_format(format: crate::api::schema::PaneGraphicsFormat) -> KittyImageFormat {
+    match format {
+        crate::api::schema::PaneGraphicsFormat::Png => KittyImageFormat::Png,
+        crate::api::schema::PaneGraphicsFormat::Rgb => KittyImageFormat::Rgb,
+        crate::api::schema::PaneGraphicsFormat::Rgba => KittyImageFormat::Rgba,
+    }
+}
+
 fn host_image_id(pane_id: PaneId, placement: &KittyImagePlacement) -> u32 {
     let format_code = kitty_format_code(placement.format);
     host_image_id_for_signature(
@@ -559,9 +687,22 @@ fn host_image_id_for_signature(pane_id: PaneId, signature: ImageSignature) -> u3
     HOST_IMAGE_ID_BASE + ((hasher.finish() as u32) % 900_000)
 }
 
-fn host_placement_id(pane_id: PaneId, placement: &KittyImagePlacement) -> u32 {
+fn pane_graphics_host_image_id(pane_id: PaneId, signature: ImageSignature) -> u32 {
     let mut hasher = DefaultHasher::new();
     pane_id.raw().hash(&mut hasher);
+    signature.hash(&mut hasher);
+    PANE_GRAPHICS_IMAGE_ID_BIT | ((hasher.finish() as u32) & !PANE_GRAPHICS_IMAGE_ID_BIT)
+}
+
+fn host_placement_id(source_key: HostSourceKey, placement: &KittyImagePlacement) -> u32 {
+    let mut hasher = DefaultHasher::new();
+    match source_key {
+        HostSourceKey::Terminal { pane_id, .. } => pane_id.raw().hash(&mut hasher),
+        HostSourceKey::PaneLayer { pane_id } => {
+            "pane.graphics".hash(&mut hasher);
+            pane_id.raw().hash(&mut hasher);
+        }
+    }
     placement.image_id.hash(&mut hasher);
     placement.placement_id.hash(&mut hasher);
     1 + ((hasher.finish() as u32) % 900_000)
@@ -849,7 +990,6 @@ fn encode_kitty_data(out: &mut Vec<u8>, control: &str, data: &[u8]) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::ghostty::KittyPlacementRenderInfo;
 
     fn test_placement(viewport_col: i32, viewport_row: i32) -> HostPlacement {
         HostPlacement {
@@ -859,6 +999,10 @@ mod tests {
                 width_px: 10,
                 height_px: 10,
             },
+            source_key: HostSourceKey::Terminal {
+                pane_id: PaneId::from_raw(1),
+                image_id: 7,
+            },
             scrollback_offset: 0,
             placement: KittyImagePlacement {
                 image_id: 7,
@@ -888,6 +1032,49 @@ mod tests {
         }
     }
 
+    fn pane_layer_placement(viewport_col: i32, viewport_row: i32) -> HostPlacement {
+        let mut placement = test_placement(viewport_col, viewport_row);
+        placement.source_key = HostSourceKey::PaneLayer {
+            pane_id: placement.pane_id,
+        };
+        placement
+    }
+
+    #[test]
+    fn terminal_placement_id_preserves_legacy_identity() {
+        let placement = test_placement(0, 0);
+        let mut legacy = DefaultHasher::new();
+        placement.pane_id.raw().hash(&mut legacy);
+        placement.placement.image_id.hash(&mut legacy);
+        placement.placement.placement_id.hash(&mut legacy);
+        let expected = 1 + ((legacy.finish() as u32) % 900_000);
+
+        assert_eq!(
+            host_placement_id(placement.source_key, &placement.placement),
+            expected
+        );
+        assert_ne!(
+            host_placement_id(
+                HostSourceKey::PaneLayer {
+                    pane_id: placement.pane_id,
+                },
+                &placement.placement,
+            ),
+            expected
+        );
+    }
+
+    #[test]
+    fn pane_graphics_image_ids_are_disjoint_from_terminal_image_ids() {
+        let placement = test_placement(0, 0);
+        let signature = image_signature(&placement, kitty_format_code(placement.placement.format));
+        let terminal_id = host_image_id_for_signature(placement.pane_id, signature);
+        let pane_graphics_id = pane_graphics_host_image_id(placement.pane_id, signature);
+
+        assert_eq!(terminal_id & PANE_GRAPHICS_IMAGE_ID_BIT, 0);
+        assert_ne!(pane_graphics_id & PANE_GRAPHICS_IMAGE_ID_BIT, 0);
+    }
+
     #[test]
     fn clipped_placement_handles_positive_viewport_without_wrapping() {
         let placement = test_placement(2, 2);
@@ -915,6 +1102,44 @@ mod tests {
     }
 
     #[test]
+    fn pane_graphics_layer_defaults_to_full_pane_grid() {
+        let info = PaneInfo {
+            id: PaneId::from_raw(9),
+            rect: Rect::new(0, 0, 12, 5),
+            inner_rect: Rect::new(2, 1, 8, 3),
+            scrollbar_rect: None,
+            borders: ratatui::widgets::Borders::NONE,
+            is_focused: true,
+        };
+        let layer = crate::app::state::PaneGraphicsLayer::new(
+            crate::api::schema::PaneGraphicsFormat::Rgba,
+            80,
+            30,
+            vec![255; 80 * 30 * 4],
+            crate::api::schema::PaneGraphicsPlacementParams::default(),
+        );
+
+        let placement = pane_graphics_host_placement(
+            &info,
+            HostCellSize {
+                width_px: 10,
+                height_px: 10,
+            },
+            &layer,
+            &HashMap::new(),
+            true,
+        );
+        let (clipped, format_code) = clipped_placement(&placement).expect("visible layer");
+
+        assert_eq!(format_code, 32);
+        assert_eq!(clipped.x, 2);
+        assert_eq!(clipped.y, 1);
+        assert_eq!(clipped.cols, 8);
+        assert_eq!(clipped.rows, 3);
+        assert_eq!(placement.placement.data.len(), 80 * 30 * 4);
+    }
+
+    #[test]
     fn graphics_update_uploads_once_then_repositions_only() {
         let mut images = HashMap::new();
         let mut placements = HashMap::new();
@@ -1190,6 +1415,10 @@ mod tests {
             let mut twin = test_placement(5, 5);
             twin.placement.image_id = 8;
             twin.placement.placement_id = 4;
+            twin.source_key = HostSourceKey::Terminal {
+                pane_id: twin.pane_id,
+                image_id: twin.placement.image_id,
+            };
             twin
         }
 
@@ -1233,6 +1462,10 @@ mod tests {
             let mut twin = test_placement(5, 5);
             twin.placement.image_id = 8;
             twin.placement.placement_id = 4;
+            twin.source_key = HostSourceKey::Terminal {
+                pane_id: twin.pane_id,
+                image_id: twin.placement.image_id,
+            };
             twin
         }
 
@@ -1312,6 +1545,174 @@ mod tests {
     }
 
     #[test]
+    fn removed_pane_layer_deletes_unreferenced_host_image() {
+        let mut images = HashMap::new();
+        let mut placements = HashMap::new();
+        let mut sources = HashMap::new();
+        let mut bytes = Vec::new();
+        encode_graphics_update(
+            &mut bytes,
+            &[pane_layer_placement(0, 0)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+        let host_id = *images.keys().next().expect("uploaded pane layer");
+
+        bytes.clear();
+        encode_graphics_update(
+            &mut bytes,
+            &[],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+
+        let delete = String::from_utf8_lossy(&bytes);
+        assert!(delete.contains(&format!("a=d,d=I,i={host_id}")));
+        assert!(images.is_empty());
+        assert!(placements.is_empty());
+        assert!(sources.is_empty());
+    }
+
+    #[test]
+    fn clipped_pane_layer_deletes_unreferenced_host_image() {
+        let mut images = HashMap::new();
+        let mut placements = HashMap::new();
+        let mut sources = HashMap::new();
+        let mut bytes = Vec::new();
+        encode_graphics_update(
+            &mut bytes,
+            &[pane_layer_placement(0, 0)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+        let host_id = *images.keys().next().expect("uploaded pane layer");
+
+        bytes.clear();
+        encode_graphics_update(
+            &mut bytes,
+            &[pane_layer_placement(100, 100)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+
+        let delete = String::from_utf8_lossy(&bytes);
+        assert!(delete.contains(&format!("a=d,d=I,i={host_id}")));
+        assert!(images.is_empty());
+        assert!(placements.is_empty());
+        assert!(sources.is_empty());
+    }
+
+    #[test]
+    fn clipped_terminal_source_retains_identity_for_later_content_replacement() {
+        let mut images = HashMap::new();
+        let mut placements = HashMap::new();
+        let mut sources = HashMap::new();
+        let mut bytes = Vec::new();
+        encode_graphics_update(
+            &mut bytes,
+            &[test_placement(0, 0)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+        let original_host_id = *images.keys().next().expect("uploaded terminal image");
+
+        bytes.clear();
+        encode_graphics_update(
+            &mut bytes,
+            &[test_placement(100, 100)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+        assert_eq!(images.len(), 1);
+        assert_eq!(sources.len(), 1);
+
+        bytes.clear();
+        let mut changed = test_placement(0, 0);
+        changed.placement.data_fingerprint = 43;
+        encode_graphics_update(
+            &mut bytes,
+            &[changed],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+
+        let update = String::from_utf8_lossy(&bytes);
+        assert!(update.contains(&format!("a=d,d=I,i={original_host_id}")));
+        assert_eq!(images.len(), 1);
+        assert_eq!(sources.len(), 1);
+    }
+
+    #[test]
+    fn removed_pane_layer_preserves_image_shared_with_terminal_source() {
+        let mut images = HashMap::new();
+        let mut placements = HashMap::new();
+        let mut sources = HashMap::new();
+        let mut bytes = Vec::new();
+        encode_graphics_update(
+            &mut bytes,
+            &[pane_layer_placement(0, 0), test_placement(4, 0)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+        assert_eq!(images.len(), 1);
+
+        bytes.clear();
+        encode_graphics_update(
+            &mut bytes,
+            &[test_placement(4, 0)],
+            false,
+            &mut images,
+            &mut placements,
+            &mut sources,
+        );
+
+        let update = String::from_utf8_lossy(&bytes);
+        assert!(!update.contains("a=d,d=I"));
+        assert_eq!(images.len(), 1);
+        assert_eq!(placements.len(), 1);
+        assert_eq!(sources.len(), 1);
+    }
+
+    #[test]
+    fn maximum_pane_graphics_stream_payload_fits_client_graphics_frame() {
+        let mut placement = pane_layer_placement(0, 0);
+        placement.placement.format = KittyImageFormat::Png;
+        placement.placement.image_width = 1;
+        placement.placement.image_height = 1;
+        placement.placement.data = vec![1_u8; crate::api::schema::PANE_GRAPHICS_STREAM_MAX_BYTES];
+        placement.placement.data_len = placement.placement.data.len();
+        let (clipped, format_code) = clipped_placement(&placement).expect("visible placement");
+        let host_id = host_image_id(placement.pane_id, &placement.placement);
+        let mut encoded = Vec::new();
+
+        assert!(encode_upload_image(
+            &mut encoded,
+            &placement,
+            format_code,
+            host_id,
+        ));
+        encode_display_placement(&mut encoded, clipped, host_id, 1, 0);
+
+        assert!(encoded.len() < crate::protocol::MAX_GRAPHICS_FRAME_SIZE);
+    }
+
+    #[test]
     fn view_change_deletes_stale_placement_immediately() {
         let mut images = HashMap::new();
         let mut placements = HashMap::new();
diff --git a/src/server/clients.rs b/src/server/clients.rs
index 705152eb..243a131d 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -20,6 +20,14 @@ pub(crate) type RenderTarget = (
     ClientConnectionMode,
 );
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) enum DeferredRender {
+    #[default]
+    None,
+    Graphics,
+    Full,
+}
+
 /// A connected client tracked by the server.
 pub(crate) struct ClientConnection {
     /// Whether this connection is the full app client or a direct terminal attach.
@@ -50,8 +58,10 @@ pub(crate) struct ClientConnection {
     pub(crate) graphics_cache: crate::kitty_graphics::HostGraphicsCache,
     /// Whether the next graphics frame must clear and rebuild host-side Kitty state.
     pub(crate) graphics_surface_reset_pending: bool,
-    /// Whether a render was skipped because the render channel was full.
+    /// Whether an ordinary render was skipped because the render channel was full.
     pub(crate) render_pending: bool,
+    /// Whether a pane-graphics-only render was skipped because the channel was full.
+    pane_graphics_render_pending: bool,
     /// Last host mouse capture mode sent to this client.
     pub(crate) host_mouse_capture_active: Option<bool>,
     /// Temporary files staged from this client's local clipboard image pastes.
@@ -115,6 +125,7 @@ impl ClientConnection {
             graphics_cache: crate::kitty_graphics::HostGraphicsCache::default(),
             graphics_surface_reset_pending: false,
             render_pending: false,
+            pane_graphics_render_pending: false,
             host_mouse_capture_active: None,
             staged_clipboard_files: Vec::new(),
             writer,
@@ -124,6 +135,39 @@ impl ClientConnection {
     pub(crate) fn request_full_redraw(&mut self) {
         self.render_state.reset_baseline();
         self.graphics_surface_reset_pending = true;
+        self.pane_graphics_render_pending = false;
+    }
+
+    pub(crate) fn deferred_render(&self) -> DeferredRender {
+        if self.render_pending {
+            DeferredRender::Full
+        } else if self.pane_graphics_render_pending {
+            DeferredRender::Graphics
+        } else {
+            DeferredRender::None
+        }
+    }
+
+    pub(crate) fn clear_deferred_render(&mut self) {
+        self.render_pending = false;
+        self.pane_graphics_render_pending = false;
+    }
+
+    pub(crate) fn defer_full_render(&mut self) {
+        self.render_pending = true;
+        self.pane_graphics_render_pending = false;
+    }
+
+    pub(crate) fn defer_pane_graphics_render(&mut self) {
+        if !self.render_pending {
+            self.pane_graphics_render_pending = true;
+        }
+    }
+
+    pub(crate) fn take_deferred_render(&mut self) -> DeferredRender {
+        let deferred = self.deferred_render();
+        self.clear_deferred_render();
+        deferred
     }
 
     pub(crate) fn is_full_app_client(&self) -> bool {
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 7725954c..720b43f2 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -53,7 +53,7 @@ use crate::server::client_accept::{
 use crate::server::client_transport::ServerEvent;
 use crate::server::clients::{
     events_include_interaction, latest_app_client, render_targets, terminal_stream_client_ids,
-    ClientConnection, ClientConnectionMode,
+    ClientConnection, ClientConnectionMode, DeferredRender,
 };
 use crate::server::keybindings::{app_keybindings, apply_keybindings};
 use crate::server::notifications::{
@@ -64,6 +64,12 @@ use crate::server::socket_paths::{
 };
 use crate::server::terminal_attach::paste_payload_for_runtime;
 
+mod pane_graphics;
+
+#[cfg(test)]
+use pane_graphics::frame_pane_graphics_for_client;
+use pane_graphics::RetainedGraphicsOutcome;
+
 #[cfg(test)]
 use crate::protocol::RenderEncoding;
 #[cfg(test)]
@@ -125,6 +131,31 @@ enum LoopEvent {
     RenderRequested,
 }
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+enum RenderImpact {
+    #[default]
+    None,
+    Graphics,
+    Full,
+}
+
+impl RenderImpact {
+    fn merge(&mut self, other: Self) {
+        *self = (*self).max(other);
+    }
+}
+
+fn record_render_impact(source: &'static str, impact: RenderImpact) {
+    let event = match (source, impact) {
+        ("api_requests", RenderImpact::Graphics) => "graphics_render_cause.api_requests",
+        ("api_requests", RenderImpact::Full) => "full_render_cause.api_requests",
+        ("server_events", RenderImpact::Graphics) => "graphics_render_cause.server_events",
+        ("server_events", RenderImpact::Full) => "full_render_cause.server_events",
+        _ => return,
+    };
+    crate::render_prof::event(event);
+}
+
 fn rect_fits_frame(rect: Rect, frame: &FrameData) -> bool {
     rect.x.saturating_add(rect.width) <= frame.width
         && rect.y.saturating_add(rect.height) <= frame.height
@@ -461,6 +492,7 @@ impl HeadlessServer {
 
         let mut needs_render = true;
         let mut needs_full_render = true;
+        let mut needs_graphics_render = false;
 
         loop {
             crate::render_prof::event("loop.tick");
@@ -496,6 +528,7 @@ impl HeadlessServer {
             if self.drain_internal_events_with_forwarding() {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
                 crate::render_prof::event("full_render_cause.internal_events");
             }
             if self.app.expire_due_metadata(Instant::now()) {
@@ -505,7 +538,22 @@ impl HeadlessServer {
             }
 
             // 3. Drain API requests.
-            if self.drain_api_requests_with_shutdown_check() {
+            if self.pane_graphics_runtime_active() {
+                let api_impact = self.drain_api_requests_with_render_impact();
+                record_render_impact("api_requests", api_impact);
+                match api_impact {
+                    RenderImpact::None => {}
+                    RenderImpact::Graphics => {
+                        needs_render = true;
+                        needs_graphics_render = true;
+                    }
+                    RenderImpact::Full => {
+                        needs_render = true;
+                        needs_full_render = true;
+                        needs_graphics_render = false;
+                    }
+                }
+            } else if self.drain_api_requests_with_shutdown_check() {
                 needs_render = true;
                 needs_full_render = true;
                 crate::render_prof::event("full_render_cause.api_requests");
@@ -518,7 +566,22 @@ impl HeadlessServer {
             self.accept_client_connections()?;
 
             // 5. Drain server events from client threads.
-            if self.drain_server_events() {
+            if self.pane_graphics_runtime_active() {
+                let server_impact = self.drain_server_events_with_render_impact();
+                record_render_impact("server_events", server_impact);
+                match server_impact {
+                    RenderImpact::None => {}
+                    RenderImpact::Graphics => {
+                        needs_render = true;
+                        needs_graphics_render = true;
+                    }
+                    RenderImpact::Full => {
+                        needs_render = true;
+                        needs_full_render = true;
+                        needs_graphics_render = false;
+                    }
+                }
+            } else if self.drain_server_events() {
                 needs_render = true;
                 needs_full_render = true;
                 crate::render_prof::event("full_render_cause.server_events");
@@ -529,20 +592,25 @@ impl HeadlessServer {
             if self.handle_scheduled_tasks_headless(now, needs_render) {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
                 crate::render_prof::event("full_render_cause.scheduled_tasks");
             }
 
             if self.handle_deferred_requests_headless() {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
             }
 
             if latest_app_client(&self.clients).is_some() && self.app.ensure_default_workspace() {
                 needs_render = true;
                 needs_full_render = true;
+                needs_graphics_render = false;
                 crate::render_prof::event("full_render_cause.default_workspace");
             }
 
+            self.cancel_inactive_pane_graphics_streams();
+
             self.drain_client_config_reload_request();
             self.stream_host_mouse_capture_mode();
 
@@ -560,8 +628,24 @@ impl HeadlessServer {
                 } else if !pty_dirty {
                     crate::render_prof::event("retained_gate.not_pty_dirty");
                 }
-                let rendered_retained =
-                    pty_dirty && !needs_full_render && self.render_retained_pty_update_and_stream();
+                let mut deferred_graphics = false;
+                let rendered_retained = if needs_graphics_render && !needs_full_render && !pty_dirty
+                {
+                    match self.render_retained_graphics_update_and_stream() {
+                        RetainedGraphicsOutcome::Sent => true,
+                        RetainedGraphicsOutcome::Deferred => {
+                            deferred_graphics = true;
+                            false
+                        }
+                        RetainedGraphicsOutcome::Fallback => false,
+                    }
+                } else {
+                    pty_dirty && !needs_full_render && self.render_retained_pty_update_and_stream()
+                };
+                if deferred_graphics {
+                    needs_render = false;
+                    continue;
+                }
                 if !rendered_retained {
                     crate::render_prof::event("full_render.invoke");
                     self.render_and_stream();
@@ -569,6 +653,7 @@ impl HeadlessServer {
                 self.app.last_render_at = Some(now);
                 needs_render = false;
                 needs_full_render = false;
+                needs_graphics_render = false;
                 continue;
             }
 
@@ -607,16 +692,47 @@ impl HeadlessServer {
                     if self.handle_internal_event_with_forwarding(ev) {
                         needs_render = true;
                         needs_full_render = true;
+                        needs_graphics_render = false;
                     }
                 }
                 LoopEvent::Api(msg) => {
-                    if self.handle_api_request_with_shutdown_check(*msg) {
+                    if self.pane_graphics_runtime_active() {
+                        let impact = self.handle_api_request_with_render_impact(*msg);
+                        record_render_impact("api_requests", impact);
+                        match impact {
+                            RenderImpact::None => {}
+                            RenderImpact::Graphics => {
+                                needs_render = true;
+                                needs_graphics_render = true;
+                            }
+                            RenderImpact::Full => {
+                                needs_render = true;
+                                needs_full_render = true;
+                                needs_graphics_render = false;
+                            }
+                        }
+                    } else if self.handle_api_request_with_shutdown_check(*msg) {
                         needs_render = true;
                         needs_full_render = true;
                     }
                 }
                 LoopEvent::ServerEvent(ev) => {
-                    if self.handle_server_event(ev) {
+                    if self.pane_graphics_runtime_active() {
+                        let impact = self.handle_server_event_with_render_impact(ev);
+                        record_render_impact("server_events", impact);
+                        match impact {
+                            RenderImpact::None => {}
+                            RenderImpact::Graphics => {
+                                needs_render = true;
+                                needs_graphics_render = true;
+                            }
+                            RenderImpact::Full => {
+                                needs_render = true;
+                                needs_full_render = true;
+                                needs_graphics_render = false;
+                            }
+                        }
+                    } else if self.handle_server_event(ev) {
                         needs_render = true;
                         needs_full_render = true;
                     }
@@ -876,6 +992,7 @@ impl HeadlessServer {
         let Some(client_id) = self.foreground_client_id else {
             self.effective_size = (MIN_COLS, MIN_ROWS);
             self.app.state.outer_terminal_focus = None;
+            self.app.state.host_cell_size = crate::kitty_graphics::HostCellSize::default();
             let server_keybindings = self.server_keybindings.clone();
             apply_keybindings(&mut self.app, &server_keybindings);
             self.sync_visible_server_config_diagnostic(false);
@@ -885,6 +1002,7 @@ impl HeadlessServer {
             self.foreground_client_id = None;
             self.effective_size = (MIN_COLS, MIN_ROWS);
             self.app.state.outer_terminal_focus = None;
+            self.app.state.host_cell_size = crate::kitty_graphics::HostCellSize::default();
             let server_keybindings = self.server_keybindings.clone();
             apply_keybindings(&mut self.app, &server_keybindings);
             self.sync_visible_server_config_diagnostic(false);
@@ -893,6 +1011,12 @@ impl HeadlessServer {
 
         let terminal_size = client.terminal_size;
         let outer_terminal_focus = client.outer_terminal_focus;
+        let host_cell_size = if self.app.state.kitty_graphics_enabled && client.cell_size.is_known()
+        {
+            client.cell_size
+        } else {
+            crate::kitty_graphics::HostCellSize::default()
+        };
         let host_terminal_theme = client.host_terminal_theme;
         let host_terminal_appearance = client.host_terminal_appearance;
         let host_terminal_appearance_explicit = client.host_terminal_appearance_explicit;
@@ -905,6 +1029,7 @@ impl HeadlessServer {
 
         self.effective_size = terminal_size;
         self.app.state.outer_terminal_focus = outer_terminal_focus;
+        self.app.state.host_cell_size = host_cell_size;
         apply_keybindings(&mut self.app, &keybindings);
         self.sync_visible_server_config_diagnostic(uses_local_keybindings);
         if outer_terminal_focus == Some(true) {
@@ -1408,7 +1533,7 @@ impl HeadlessServer {
 
     /// Drains server events from the dedicated channel.
     ///
-    /// Returns true if any input was processed (requiring a re-render).
+    /// Uses the original full-render semantics when pane graphics are dormant.
     fn drain_server_events(&mut self) -> bool {
         let mut changed = false;
         while let Ok(ev) = self.server_event_rx.try_recv() {
@@ -1417,6 +1542,15 @@ impl HeadlessServer {
         changed
     }
 
+    /// Returns the strongest render impact from the drained event batch.
+    fn drain_server_events_with_render_impact(&mut self) -> RenderImpact {
+        let mut impact = RenderImpact::None;
+        while let Ok(ev) = self.server_event_rx.try_recv() {
+            impact.merge(self.handle_server_event_with_render_impact(ev));
+        }
+        impact
+    }
+
     fn terminal_id_by_string(&self, terminal_id: &str) -> Option<crate::terminal::TerminalId> {
         self.app
             .state
@@ -2450,7 +2584,7 @@ impl HeadlessServer {
         if let Some(client) = self.clients.get_mut(&client_id) {
             if host_surface_redraw {
                 client.request_full_redraw();
-                client.render_pending = true;
+                client.defer_full_render();
             } else {
                 // Ensure semantic clients receive one post-input frame even if the
                 // semantic buffer compares equal. Terminal-ANSI clients must keep their
@@ -2765,12 +2899,7 @@ impl HeadlessServer {
                 let Some(client) = self.clients.get_mut(&client_id) else {
                     return false;
                 };
-                if client.render_pending {
-                    client.render_pending = false;
-                    true
-                } else {
-                    false
-                }
+                client.take_deferred_render() != DeferredRender::None
             }
             ServerEvent::QuitSignal => {
                 // The quit check at the top of the loop handles this.
@@ -2780,6 +2909,23 @@ impl HeadlessServer {
         }
     }
 
+    fn handle_server_event_with_render_impact(&mut self, ev: ServerEvent) -> RenderImpact {
+        let deferred_render = match &ev {
+            ServerEvent::ClientWriterDrained { client_id } => self
+                .clients
+                .get(client_id)
+                .map_or(DeferredRender::None, ClientConnection::deferred_render),
+            _ => DeferredRender::None,
+        };
+        if !self.handle_server_event(ev) {
+            return RenderImpact::None;
+        }
+        match deferred_render {
+            DeferredRender::Graphics => RenderImpact::Graphics,
+            DeferredRender::None | DeferredRender::Full => RenderImpact::Full,
+        }
+    }
+
     fn ignore_client_event_during_handoff(ev: &ServerEvent) -> bool {
         !matches!(
             ev,
@@ -2801,6 +2947,14 @@ impl HeadlessServer {
         changed
     }
 
+    fn drain_api_requests_with_render_impact(&mut self) -> RenderImpact {
+        let mut impact = RenderImpact::None;
+        while let Ok(msg) = self.app.api_rx.try_recv() {
+            impact.merge(self.handle_api_request_with_render_impact(msg));
+        }
+        impact
+    }
+
     /// Handles a single API request with shutdown awareness.
     ///
     /// Also forwards any toast/sound notifications that result from the API
@@ -2811,6 +2965,23 @@ impl HeadlessServer {
         self.handle_api_request_with_shutdown_check_inner(msg, false)
     }
 
+    fn handle_api_request_with_render_impact(
+        &mut self,
+        msg: api::ApiRequestMessage,
+    ) -> RenderImpact {
+        if matches!(
+            &msg.request.method,
+            api::schema::Method::PaneGraphicsStreamSet(_)
+        ) {
+            return self.handle_pane_graphics_stream_frame(msg);
+        }
+        if self.handle_api_request_with_shutdown_check_inner(msg, false) {
+            RenderImpact::Full
+        } else {
+            RenderImpact::None
+        }
+    }
+
     fn handle_api_request_with_shutdown_check_inner(
         &mut self,
         msg: api::ApiRequestMessage,
@@ -2884,7 +3055,16 @@ impl HeadlessServer {
             _ => {}
         }
 
-        let mut changed = metadata_expired | api::request_changes_ui(&msg.request);
+        let pane_graphics_revision_before = matches!(
+            &msg.request.method,
+            api::schema::Method::PaneGraphicsSet(_)
+                | api::schema::Method::PaneGraphicsClear(_)
+                | api::schema::Method::PaneGraphicsStreamOpen(_)
+                | api::schema::Method::PaneGraphicsStreamClose(_)
+        )
+        .then_some(self.app.state.pane_graphics_revision);
+        let mut changed = metadata_expired
+            | (pane_graphics_revision_before.is_none() && api::request_changes_ui(&msg.request));
         let skip_default_workspace = skip_default_workspace_for_request
             || matches!(
                 &msg.request.method,
@@ -2965,6 +3145,10 @@ impl HeadlessServer {
         };
         let _ = msg.respond_to.send(response);
 
+        if let Some(revision_before) = pane_graphics_revision_before {
+            changed |= revision_before != self.app.state.pane_graphics_revision;
+        }
+
         // Forward new toast state only when a client-local delivery mode is selected.
         // Herdr delivery renders the toast in-frame and must not ask clients to
         // show a terminal or system notification.
@@ -3151,8 +3335,6 @@ impl HeadlessServer {
         }
     }
 
-    /// Renders the current state to client-sized virtual buffers and streams
-    /// frames to all connected clients.
     fn render_retained_pty_update_and_stream(&mut self) -> bool {
         crate::render_prof::event("retained.attempt");
         let retained_started = crate::render_prof::timer();
@@ -3188,7 +3370,7 @@ impl HeadlessServer {
         let Some(client) = self.clients.get(client_id) else {
             retained_fallback!("client_missing");
         };
-        if client.render_pending {
+        if client.deferred_render() != DeferredRender::None {
             retained_fallback!("render_pending");
         }
         if self.app.state.kitty_graphics_enabled && !client.graphics_cache.is_empty() {
@@ -3304,7 +3486,7 @@ impl HeadlessServer {
         };
         let prepare_started = crate::render_prof::timer();
         let Some(prepared) = client.render_state.prepare_frame(frame) else {
-            client.render_pending = false;
+            client.clear_deferred_render();
             crate::render_prof::event("retained_send.skip_identical");
             crate::render_prof::duration_since("retained_send.prepare_frame", prepare_started);
             return true;
@@ -3338,14 +3520,14 @@ impl HeadlessServer {
         let send_started = crate::render_prof::timer();
         match writer.render.try_send(serialized) {
             Ok(()) => {
-                client.render_pending = false;
+                client.clear_deferred_render();
                 client.render_state.commit_sent_frame(prepared);
                 crate::render_prof::event("retained_send.sent");
                 crate::render_prof::duration_since("retained_send.try_send", send_started);
                 true
             }
             Err(std::sync::mpsc::TrySendError::Full(_)) => {
-                client.render_pending = true;
+                client.defer_full_render();
                 crate::render_prof::event("retained_send_fallback.queue_full");
                 crate::render_prof::duration_since("retained_send.try_send", send_started);
                 debug!(
@@ -3398,24 +3580,20 @@ impl HeadlessServer {
             let mut frame = match mode {
                 ClientConnectionMode::App => {
                     let render_started = crate::render_prof::timer();
-                    let (buffer, cursor) =
+                    let render_cell_size =
                         if self.app.state.kitty_graphics_enabled && cell_size.is_known() {
-                            crate::server::render_stream::render_virtual_with_runtime_registry(
-                                &mut self.app.state,
-                                &self.app.terminal_runtimes,
-                                area,
-                                is_foreground,
-                                cell_size,
-                            )
+                            cell_size
                         } else {
-                            crate::server::render_stream::render_virtual_with_runtime_registry(
-                                &mut self.app.state,
-                                &self.app.terminal_runtimes,
-                                area,
-                                is_foreground,
-                                crate::kitty_graphics::HostCellSize::default(),
-                            )
+                            crate::kitty_graphics::HostCellSize::default()
                         };
+                    let (buffer, cursor) =
+                        crate::server::render_stream::render_virtual_with_runtime_registry(
+                            &mut self.app.state,
+                            &self.app.terminal_runtimes,
+                            area,
+                            is_foreground,
+                            render_cell_size,
+                        );
                     crate::render_prof::duration_since(
                         "full_render.render_virtual",
                         render_started,
@@ -3524,7 +3702,7 @@ impl HeadlessServer {
             let has_graphics = !frame.graphics.is_empty();
             let prepare_started = crate::render_prof::timer();
             let Some(mut prepared) = client.render_state.prepare_frame(frame) else {
-                client.render_pending = false;
+                client.clear_deferred_render();
                 crate::render_prof::event("full_render.skip_identical");
                 crate::render_prof::duration_since("full_render.prepare_frame", prepare_started);
                 continue;
@@ -3557,7 +3735,7 @@ impl HeadlessServer {
                     let Some(text_only_prepared) =
                         client.render_state.prepare_frame(text_only_frame)
                     else {
-                        client.render_pending = false;
+                        client.clear_deferred_render();
                         crate::render_prof::event("full_render.skip_identical_text_only");
                         crate::render_prof::duration_since(
                             "full_render.serialize",
@@ -3605,7 +3783,7 @@ impl HeadlessServer {
             let send_started = crate::render_prof::timer();
             match writer.render.try_send(serialized) {
                 Ok(()) => {
-                    client.render_pending = false;
+                    client.clear_deferred_render();
                     if commit_graphics_cache {
                         client.graphics_cache = next_graphics_cache;
                         client.graphics_surface_reset_pending = false;
@@ -3615,7 +3793,7 @@ impl HeadlessServer {
                     crate::render_prof::duration_since("full_render.try_send", send_started);
                 }
                 Err(std::sync::mpsc::TrySendError::Full(_)) => {
-                    client.render_pending = true;
+                    client.defer_full_render();
                     deferred_frame = true;
                     crate::render_prof::event("full_render.queue_full");
                     crate::render_prof::duration_since("full_render.try_send", send_started);
@@ -4229,6 +4407,9 @@ mod tests {
     use crate::app::AppState;
     use crate::protocol::CursorState;
 
+    #[path = "pane_graphics.rs"]
+    mod pane_graphics_tests;
+
     fn test_headless_server() -> HeadlessServer {
         test_headless_server_with_event_hub(api::EventHub::default())
     }
@@ -7333,11 +7514,17 @@ next_tab = ""
             client_id: 1,
             data: b"\x1b[I".to_vec(),
         }));
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
 
         server.render_and_stream();
 
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::ReloadSoundConfig
@@ -7356,7 +7543,10 @@ next_tab = ""
         assert!(server.handle_server_event(ServerEvent::ClientWriterDrained { client_id: 1 }));
         server.render_and_stream();
 
-        assert!(!server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::None
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::Frame(_)
@@ -7434,7 +7624,10 @@ next_tab = ""
             data: b"\x1b[I".to_vec(),
         }));
 
-        assert!(!server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::None
+        );
         assert!(!server.app.full_redraw_pending);
         assert_eq!(server.clients[&1].outer_terminal_focus, Some(true));
         assert_eq!(server.app.state.outer_terminal_focus, Some(true));
@@ -7510,7 +7703,10 @@ next_tab = ""
         server.foreground_client_id = Some(1);
 
         server.render_and_stream();
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::ReloadSoundConfig
@@ -7533,7 +7729,10 @@ next_tab = ""
                 .unwrap(),
             1
         );
-        assert!(!server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::None
+        );
     }
 
     #[test]
@@ -7984,7 +8183,10 @@ next_tab = ""
         server.render_and_stream();
 
         assert!(server.app.full_redraw_pending);
-        assert!(server.clients.get(&1).unwrap().render_pending);
+        assert_eq!(
+            server.clients.get(&1).unwrap().deferred_render(),
+            DeferredRender::Full
+        );
         assert!(matches!(
             read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
             ServerMessage::ReloadSoundConfig
diff --git a/src/server/headless/pane_graphics.rs b/src/server/headless/pane_graphics.rs
new file mode 100644
index 00000000..18174f45
--- /dev/null
+++ b/src/server/headless/pane_graphics.rs
@@ -0,0 +1,200 @@
+use tracing::warn;
+
+use super::{HeadlessServer, RenderImpact};
+use crate::api;
+use crate::protocol::{ServerMessage, MAX_GRAPHICS_FRAME_SIZE};
+use crate::server::clients::{render_targets, ClientConnectionMode, DeferredRender};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum RetainedGraphicsOutcome {
+    Sent,
+    Deferred,
+    Fallback,
+}
+
+pub(super) fn frame_pane_graphics_for_client(bytes: Vec<u8>) -> Vec<u8> {
+    if bytes.is_empty() {
+        return bytes;
+    }
+    let mut framed = Vec::with_capacity(bytes.len() + 4);
+    framed.extend_from_slice(b"\x1b7");
+    framed.extend_from_slice(&bytes);
+    framed.extend_from_slice(b"\x1b8");
+    framed
+}
+
+impl HeadlessServer {
+    pub(super) fn pane_graphics_runtime_active(&self) -> bool {
+        !self.app.state.pane_graphics_layers.is_empty()
+            || !self.app.state.pane_graphics_streams.is_empty()
+    }
+
+    pub(super) fn cancel_inactive_pane_graphics_streams(&self) {
+        api::cancel_inactive_pane_graphics_streams(|owner| {
+            self.app
+                .state
+                .pane_graphics_streams
+                .values()
+                .any(|active_owner| active_owner == owner)
+        });
+    }
+
+    pub(super) fn handle_pane_graphics_stream_frame(
+        &mut self,
+        msg: api::ApiRequestMessage,
+    ) -> RenderImpact {
+        if self.shutting_down {
+            let _ = self.handle_api_request_with_shutdown_check_inner(msg, false);
+            return RenderImpact::None;
+        }
+
+        let internal_changed = self.drain_all_internal_events_with_forwarding();
+        let response = self
+            .app
+            .handle_api_request_after_internal_events_drained(msg.request);
+        let succeeded = serde_json::from_str::<api::schema::SuccessResponse>(&response).is_ok();
+        let _ = msg.respond_to.send(response);
+
+        if internal_changed {
+            RenderImpact::Full
+        } else if succeeded {
+            RenderImpact::Graphics
+        } else {
+            RenderImpact::None
+        }
+    }
+
+    pub(super) fn render_retained_graphics_update_and_stream(&mut self) -> RetainedGraphicsOutcome {
+        crate::render_prof::event("retained_graphics.attempt");
+        if self.app.full_redraw_pending {
+            crate::render_prof::event("retained_graphics_fallback.full_redraw_pending");
+            return RetainedGraphicsOutcome::Fallback;
+        }
+
+        let render_targets = render_targets(&self.clients, self.foreground_client_id);
+        let mut app_view_size = None;
+        for (_, terminal_size, _, _, mode) in &render_targets {
+            if !matches!(mode, ClientConnectionMode::App) {
+                continue;
+            }
+            if app_view_size.is_some_and(|size| size != *terminal_size) {
+                crate::render_prof::event("retained_graphics_fallback.mixed_app_geometry");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+            app_view_size = Some(*terminal_size);
+        }
+        let mut deferred = false;
+        let mut prepared = Vec::new();
+
+        for (client_id, (cols, rows), cell_size, _is_foreground, mode) in render_targets {
+            if !matches!(mode, ClientConnectionMode::App) {
+                continue;
+            }
+            let Some(client) = self.clients.get_mut(&client_id) else {
+                crate::render_prof::event("retained_graphics_fallback.client_missing");
+                return RetainedGraphicsOutcome::Fallback;
+            };
+            if client.deferred_render() != DeferredRender::None {
+                deferred = true;
+                continue;
+            }
+            if client.graphics_surface_reset_pending || !cell_size.is_known() {
+                crate::render_prof::event("retained_graphics_fallback.client_state");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+            let Some(last_frame) = client.render_state.last_frame() else {
+                crate::render_prof::event("retained_graphics_fallback.no_last_frame");
+                return RetainedGraphicsOutcome::Fallback;
+            };
+            if last_frame.width != cols || last_frame.height != rows {
+                crate::render_prof::event("retained_graphics_fallback.frame_size_mismatch");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+            if client.writer.is_none() {
+                crate::render_prof::event("retained_graphics_fallback.writer_missing");
+                return RetainedGraphicsOutcome::Fallback;
+            }
+
+            let mut next_graphics_cache = client.graphics_cache.clone();
+            let encode_started = crate::render_prof::timer();
+            let bytes =
+                frame_pane_graphics_for_client(crate::kitty_graphics::encode_local_pane_graphics(
+                    &self.app.state,
+                    &self.app.terminal_runtimes,
+                    cell_size,
+                    &mut next_graphics_cache,
+                ));
+            crate::render_prof::duration_since("retained_graphics.graphics_encode", encode_started);
+            if bytes.len() > MAX_GRAPHICS_FRAME_SIZE {
+                warn!(
+                    client_id,
+                    graphics_bytes = bytes.len(),
+                    max = MAX_GRAPHICS_FRAME_SIZE,
+                    "dropping oversized retained graphics payload"
+                );
+                continue;
+            }
+            let serialized = if bytes.is_empty() {
+                None
+            } else {
+                let serialize_started = crate::render_prof::timer();
+                let framed = match Self::frame_server_message_with_max(
+                    &ServerMessage::Graphics { bytes },
+                    MAX_GRAPHICS_FRAME_SIZE,
+                ) {
+                    Ok(framed) => framed,
+                    Err(err) => {
+                        warn!(client_id, err = %err, "failed to serialize retained graphics");
+                        return RetainedGraphicsOutcome::Fallback;
+                    }
+                };
+                crate::render_prof::duration_since(
+                    "retained_graphics.serialize",
+                    serialize_started,
+                );
+                Some(framed)
+            };
+            prepared.push((client_id, serialized, next_graphics_cache));
+        }
+
+        let mut broken_clients = Vec::new();
+        for (client_id, serialized, next_graphics_cache) in prepared {
+            let Some(client) = self.clients.get_mut(&client_id) else {
+                continue;
+            };
+            let Some(serialized) = serialized else {
+                client.graphics_cache = next_graphics_cache;
+                client.clear_deferred_render();
+                continue;
+            };
+            let Some(writer) = client.writer.as_ref().cloned() else {
+                broken_clients.push(client_id);
+                continue;
+            };
+            match writer.render.try_send(serialized) {
+                Ok(()) => {
+                    client.graphics_cache = next_graphics_cache;
+                    client.clear_deferred_render();
+                    crate::render_prof::event("retained_graphics.sent");
+                }
+                Err(std::sync::mpsc::TrySendError::Full(_)) => {
+                    client.defer_pane_graphics_render();
+                    deferred = true;
+                    crate::render_prof::event("retained_graphics.deferred");
+                }
+                Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
+                    broken_clients.push(client_id);
+                }
+            }
+        }
+        for client_id in broken_clients {
+            self.remove_client_and_resize_if_needed(client_id);
+        }
+
+        if deferred {
+            RetainedGraphicsOutcome::Deferred
+        } else {
+            RetainedGraphicsOutcome::Sent
+        }
+    }
+}
diff --git a/src/server/headless/tests/pane_graphics.rs b/src/server/headless/tests/pane_graphics.rs
new file mode 100644
index 00000000..627ffc78
--- /dev/null
+++ b/src/server/headless/tests/pane_graphics.rs
@@ -0,0 +1,357 @@
+use super::*;
+
+#[test]
+fn frames_preserve_cursor_without_changing_ordinary_messages() {
+    assert_eq!(frame_pane_graphics_for_client(Vec::new()), Vec::<u8>::new());
+    assert_eq!(
+        frame_pane_graphics_for_client(b"graphics".to_vec()),
+        b"\x1b7graphics\x1b8"
+    );
+}
+
+fn enable_graphics_and_render(
+    server: &mut HeadlessServer,
+    client_rx: &std::sync::mpsc::Receiver<Vec<u8>>,
+) -> FrameData {
+    server.app.state.kitty_graphics_enabled = true;
+    server.clients.get_mut(&1).unwrap().cell_size = crate::kitty_graphics::HostCellSize {
+        width_px: 10,
+        height_px: 20,
+    };
+    server.render_and_stream();
+    read_server_frame(
+        client_rx
+            .recv_timeout(Duration::from_millis(100))
+            .expect("initial frame"),
+    )
+}
+
+fn set_graphics_layer(server: &mut HeadlessServer, pane_id: crate::layout::PaneId, data: Vec<u8>) {
+    server.app.state.pane_graphics_layers.insert(
+        pane_id,
+        crate::app::state::PaneGraphicsLayer::new(
+            api::schema::PaneGraphicsFormat::Png,
+            1,
+            1,
+            data,
+            api::schema::PaneGraphicsPlacementParams::default(),
+        ),
+    );
+}
+
+fn fill_render_lane(server: &HeadlessServer) {
+    let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
+        .expect("dummy frame");
+    server
+        .clients
+        .get(&1)
+        .unwrap()
+        .writer
+        .as_ref()
+        .unwrap()
+        .render
+        .try_send(queued)
+        .expect("pre-fill render lane");
+}
+
+fn stream_set_message(
+    id: &str,
+    pane_id: &str,
+    owner: &str,
+    data: Vec<u8>,
+) -> (api::ApiRequestMessage, std::sync::mpsc::Receiver<String>) {
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    (
+        api::ApiRequestMessage {
+            request: api::schema::Request {
+                id: id.into(),
+                method: api::schema::Method::PaneGraphicsStreamSet(
+                    api::schema::PaneGraphicsSetParams {
+                        pane_id: pane_id.into(),
+                        owner: owner.into(),
+                        format: api::schema::PaneGraphicsFormat::Png,
+                        image_width: 1,
+                        image_height: 1,
+                        data: Some(data),
+                        data_base64: String::new(),
+                        placement: api::schema::PaneGraphicsPlacementParams::default(),
+                    },
+                ),
+            },
+            respond_to,
+            response_write_complete: None,
+        },
+        response_rx,
+    )
+}
+
+#[tokio::test]
+async fn retained_update_sends_only_graphics_message() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    let baseline = enable_graphics_and_render(&mut server, &client_rx);
+    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3]);
+
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Sent
+    );
+    match read_server_message(
+        client_rx
+            .recv_timeout(Duration::from_millis(100))
+            .expect("graphics-only update"),
+    ) {
+        ServerMessage::Graphics { bytes } => {
+            assert!(bytes.windows(3).any(|window| window == b"\x1b_G"));
+        }
+        other => panic!("expected graphics-only message, got {other:?}"),
+    }
+    assert_frame_data_eq(
+        server
+            .clients
+            .get(&1)
+            .unwrap()
+            .render_state
+            .last_frame()
+            .expect("semantic baseline"),
+        &baseline,
+    );
+}
+
+#[tokio::test]
+async fn retained_update_defers_on_full_render_lane() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    let _ = enable_graphics_and_render(&mut server, &client_rx);
+    fill_render_lane(&server);
+    set_graphics_layer(&mut server, pane_id, vec![4, 5, 6]);
+
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Deferred
+    );
+    let client = server.clients.get(&1).unwrap();
+    assert_eq!(client.deferred_render(), DeferredRender::Graphics);
+    assert!(matches!(
+        read_server_message(client_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
+        ServerMessage::ReloadSoundConfig
+    ));
+    assert_eq!(
+        server.handle_server_event_with_render_impact(ServerEvent::ClientWriterDrained {
+            client_id: 1
+        }),
+        RenderImpact::Graphics
+    );
+}
+
+#[tokio::test]
+async fn retained_update_does_not_downgrade_pending_full_render() {
+    let (mut server, client_rx, pane_id) = retained_test_server(b"aaaa");
+    let _ = enable_graphics_and_render(&mut server, &client_rx);
+    fill_render_lane(&server);
+    let client = server.clients.get_mut(&1).unwrap();
+    client.request_full_redraw();
+    server.render_and_stream();
+    assert_eq!(
+        server.clients.get(&1).unwrap().deferred_render(),
+        DeferredRender::Full
+    );
+
+    set_graphics_layer(&mut server, pane_id, vec![7, 8, 9]);
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Deferred
+    );
+    assert_eq!(
+        server.clients.get(&1).unwrap().deferred_render(),
+        DeferredRender::Full
+    );
+    assert_eq!(
+        server.handle_server_event_with_render_impact(ServerEvent::ClientWriterDrained {
+            client_id: 1
+        }),
+        RenderImpact::Full
+    );
+}
+
+#[tokio::test]
+async fn retained_update_falls_back_for_mixed_app_geometry() {
+    let (mut server, client_rx, _pane_id) = retained_test_server(b"aaaa");
+    let _ = enable_graphics_and_render(&mut server, &client_rx);
+
+    let (writer, _control_rx, _render_rx) = test_client_writer();
+    server.clients.insert(
+        2,
+        ClientConnection::new(
+            (60, 20),
+            crate::kitty_graphics::HostCellSize {
+                width_px: 10,
+                height_px: 20,
+            },
+            crate::terminal_theme::TerminalTheme::default(),
+            None,
+            2,
+            RenderEncoding::SemanticFrame,
+            Some(writer),
+        ),
+    );
+
+    assert_eq!(
+        server.render_retained_graphics_update_and_stream(),
+        RetainedGraphicsOutcome::Fallback
+    );
+}
+
+#[test]
+fn stream_set_has_graphics_only_render_impact() {
+    let mut server = test_headless_server();
+    let workspace = crate::workspace::Workspace::test_new("graphics");
+    let pane_id = workspace.tabs[0].root_pane;
+    let public_pane_id = format!("{}:p1", workspace.id);
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+    server.app.state.kitty_graphics_enabled = true;
+    server
+        .app
+        .state
+        .pane_graphics_streams
+        .insert(pane_id, "owner-a".into());
+
+    let (request, response_rx) =
+        stream_set_message("wrong-owner", &public_pane_id, "owner-b", vec![1, 2, 3]);
+    assert_eq!(
+        server.handle_api_request_with_render_impact(request),
+        RenderImpact::None
+    );
+    assert!(serde_json::from_str::<api::schema::ErrorResponse>(
+        &response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap()
+    )
+    .is_ok());
+
+    let (request, response_rx) =
+        stream_set_message("stream-frame", &public_pane_id, "owner-a", vec![1, 2, 3]);
+    assert_eq!(
+        server.handle_api_request_with_render_impact(request),
+        RenderImpact::Graphics
+    );
+    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
+        &response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap()
+    )
+    .is_ok());
+
+    server
+        .app
+        .event_tx
+        .try_send(AppEvent::UpdateReady {
+            version: "9.9.9".into(),
+            install_command: "herdr update".into(),
+        })
+        .unwrap();
+    let (request, _response_rx) = stream_set_message(
+        "stream-frame-with-internal-event",
+        &public_pane_id,
+        "owner-a",
+        vec![4, 5, 6],
+    );
+    assert_eq!(
+        server.handle_api_request_with_render_impact(request),
+        RenderImpact::Full
+    );
+
+    server.app.state.pane_graphics_streams.clear();
+    let (respond_to, _response_rx) = std::sync::mpsc::channel();
+    let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "direct-frame".into(),
+            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
+                pane_id: public_pane_id,
+                owner: String::new(),
+                format: api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data: Some(vec![1, 2, 3]),
+                data_base64: String::new(),
+                placement: api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        },
+        respond_to,
+        response_write_complete: None,
+    });
+    assert_eq!(impact, RenderImpact::Full);
+}
+
+#[test]
+fn rejected_or_stale_requests_do_not_schedule_rendering() {
+    let mut server = test_headless_server();
+    let workspace = crate::workspace::Workspace::test_new("graphics");
+    let pane_id = workspace.tabs[0].root_pane;
+    let public_pane_id = format!("{}:p1", workspace.id);
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.active = Some(0);
+    server.app.state.selected = 0;
+
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    let changed = server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "disabled-set".into(),
+            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
+                pane_id: public_pane_id.clone(),
+                owner: String::new(),
+                format: api::schema::PaneGraphicsFormat::Png,
+                image_width: 1,
+                image_height: 1,
+                data: Some(vec![1, 2, 3]),
+                data_base64: String::new(),
+                placement: api::schema::PaneGraphicsPlacementParams::default(),
+            }),
+        },
+        respond_to,
+        response_write_complete: None,
+    });
+    assert!(!changed);
+    let response = response_rx
+        .recv_timeout(Duration::from_millis(100))
+        .unwrap();
+    assert_eq!(
+        serde_json::from_str::<api::schema::ErrorResponse>(&response)
+            .unwrap()
+            .error
+            .code,
+        "feature_disabled"
+    );
+
+    server.app.state.kitty_graphics_enabled = true;
+    server
+        .app
+        .state
+        .pane_graphics_streams
+        .insert(pane_id, "current-owner".into());
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "stale-close".into(),
+            method: api::schema::Method::PaneGraphicsStreamClose(
+                api::schema::PaneGraphicsStreamParams {
+                    pane_id: public_pane_id,
+                    owner: "stale-owner".into(),
+                },
+            ),
+        },
+        respond_to,
+        response_write_complete: None,
+    });
+    assert_eq!(impact, RenderImpact::None);
+    assert_eq!(
+        server.app.state.pane_graphics_streams.get(&pane_id),
+        Some(&"current-owner".to_string())
+    );
+    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
+        &response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap()
+    )
+    .is_ok());
+}