feat(graphics): add direct pane frame streaming (#2523)

* feat(graphics): add direct pane frame streaming

* feat(graphics): expose pane visibility

* test(graphics): preserve legacy terminal image output

* fix(graphics): harden streaming lifecycle

* fix(input): preserve mouse releases outside panes

3 files changed, 894 insertions, 46 deletions

diff --git a/src/client/direct_graphics.rs b/src/client/direct_graphics.rs
new file mode 100644
index 00000000..1bf10041
--- /dev/null
+++ b/src/client/direct_graphics.rs
@@ -0,0 +1,393 @@
+const MAX_RESPONSE_BYTES: usize = 4096;
+const KITTY_PREFIX: &[u8] = b"\x1b_G";
+const KITTY_SUFFIX: &[u8] = b"\x1b\\";
+const RESPONSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+const LATE_RESPONSE_DRAIN: std::time::Duration = std::time::Duration::from_secs(1);
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) struct Response {
+    pub(super) transfer_id: u64,
+    pub(super) image_id: u32,
+    pub(super) success: bool,
+}
+
+#[derive(Debug, Default)]
+pub(super) struct ResponseMatcher {
+    expected: Option<(u64, u32, Option<std::time::Instant>)>,
+    retired: Option<(u32, std::time::Instant)>,
+    active: std::sync::Arc<std::sync::atomic::AtomicBool>,
+}
+
+impl ResponseMatcher {
+    pub(super) fn active_handle(&self) -> std::sync::Arc<std::sync::atomic::AtomicBool> {
+        self.active.clone()
+    }
+
+    fn refresh_active(&self) {
+        self.active.store(
+            self.expected.is_some() || self.retired.is_some(),
+            std::sync::atomic::Ordering::Release,
+        );
+    }
+    pub(super) fn interested(&mut self) -> bool {
+        self.expire();
+        self.expected.is_some() || self.retired.is_some()
+    }
+    pub(super) fn arm(&mut self, transfer_id: u64, image_id: u32) -> bool {
+        self.expire();
+        if self.expected.is_some() {
+            return false;
+        }
+        self.expected = Some((transfer_id, image_id, None));
+        self.refresh_active();
+        true
+    }
+
+    pub(super) fn start(&mut self, transfer_id: u64) {
+        self.start_at(transfer_id, std::time::Instant::now());
+    }
+
+    fn start_at(&mut self, transfer_id: u64, now: std::time::Instant) {
+        if let Some((id, _, deadline)) = &mut self.expected {
+            if *id == transfer_id {
+                *deadline = Some(now + RESPONSE_TIMEOUT);
+            }
+        }
+    }
+
+    pub(super) fn cancel(&mut self, transfer_id: u64) {
+        if self.expected.is_some_and(|(id, _, _)| id == transfer_id) {
+            self.expected = None;
+            self.refresh_active();
+        }
+    }
+
+    pub(super) fn retire(&mut self, transfer_id: u64) {
+        if self.expected.is_some_and(|(id, _, _)| id == transfer_id) {
+            if let Some((_, image_id, _)) = self.expected.take() {
+                self.retired = Some((image_id, std::time::Instant::now() + LATE_RESPONSE_DRAIN));
+                self.refresh_active();
+            }
+        }
+    }
+
+    pub(super) fn expire(&mut self) {
+        self.expire_at(std::time::Instant::now());
+    }
+
+    fn expire_at(&mut self, now: std::time::Instant) {
+        if self
+            .expected
+            .is_some_and(|(_, _, deadline)| deadline.is_some_and(|deadline| deadline <= now))
+        {
+            if let Some((_, image_id, _)) = self.expected.take() {
+                self.retired = Some((image_id, now + LATE_RESPONSE_DRAIN));
+            }
+        }
+        if self.retired.is_some_and(|(_, deadline)| deadline <= now) {
+            self.retired = None;
+        }
+        self.refresh_active();
+    }
+
+    pub(super) fn consume(&mut self, bytes: &[u8]) -> Option<Option<Response>> {
+        self.expire();
+        if !bytes.starts_with(b"\x1b_G")
+            || bytes.len() > MAX_RESPONSE_BYTES
+            || !bytes.ends_with(b"\x1b\\")
+        {
+            return None;
+        }
+        let payload = &bytes[3..bytes.len() - 2];
+        let separator = payload.iter().position(|byte| *byte == b';')?;
+        if let Some((retired_id, _)) = self.retired {
+            if matching_response_controls(&payload[..separator], retired_id) {
+                self.retired = None;
+                self.refresh_active();
+                return Some(None);
+            }
+        }
+        let (transfer_id, image_id, _) = self.expected?;
+        if !matching_response_controls(&payload[..separator], image_id) {
+            return None;
+        }
+        self.expected = None;
+        self.refresh_active();
+        Some(Some(Response {
+            transfer_id,
+            image_id,
+            success: &payload[separator + 1..] == b"OK",
+        }))
+    }
+}
+
+#[derive(Default)]
+pub(super) struct InputFilter {
+    pending: Vec<u8>,
+}
+
+impl InputFilter {
+    pub(super) fn has_pending(&self) -> bool {
+        !self.pending.is_empty()
+    }
+
+    pub(super) fn flush_if_inactive(&mut self, matcher: &mut ResponseMatcher) -> Option<Vec<u8>> {
+        (self.has_pending() && !matcher.interested()).then(|| std::mem::take(&mut self.pending))
+    }
+
+    pub(super) fn push(
+        &mut self,
+        bytes: &[u8],
+        matcher: &mut ResponseMatcher,
+    ) -> (Vec<Vec<u8>>, Vec<Response>) {
+        self.pending.extend_from_slice(bytes);
+        let mut output = Vec::new();
+        let mut responses = Vec::new();
+        if !matcher.interested() {
+            output.push(std::mem::take(&mut self.pending));
+            return (output, responses);
+        }
+        loop {
+            let Some(start) = self
+                .pending
+                .windows(3)
+                .position(|window| window == KITTY_PREFIX)
+            else {
+                let keep = (1..KITTY_PREFIX.len())
+                    .rev()
+                    .find(|len| self.pending.ends_with(&KITTY_PREFIX[..*len]))
+                    .unwrap_or(0);
+                let emit = self.pending.len().saturating_sub(keep);
+                if emit > 0 {
+                    output.push(self.pending.drain(..emit).collect());
+                }
+                break;
+            };
+            if start > 0 {
+                output.push(self.pending.drain(..start).collect());
+            }
+            let end = self
+                .pending
+                .windows(2)
+                .position(|window| window == KITTY_SUFFIX);
+            let Some(end) = end else {
+                if self.pending.len() > MAX_RESPONSE_BYTES {
+                    output.push(std::mem::take(&mut self.pending));
+                }
+                break;
+            };
+            let command: Vec<u8> = self.pending.drain(..end + 2).collect();
+            match matcher.consume(&command) {
+                Some(Some(response)) => responses.push(response),
+                Some(None) => {}
+                None => output.push(command),
+            }
+        }
+        (output, responses)
+    }
+}
+
+fn matching_response_controls(bytes: &[u8], expected: u32) -> bool {
+    let mut matched = false;
+    for field in bytes.split(|byte| *byte == b',') {
+        let Some(separator) = field.iter().position(|byte| *byte == b'=') else {
+            return false;
+        };
+        let (key, value) = field.split_at(separator);
+        let value = &value[1..];
+        if value.is_empty() || !value.iter().all(u8::is_ascii_digit) {
+            return false;
+        }
+        match key {
+            b"i" if !matched
+                && std::str::from_utf8(value).ok().and_then(|v| v.parse().ok())
+                    == Some(expected) =>
+            {
+                matched = true
+            }
+            b"I" | b"p" => {}
+            _ => return false,
+        }
+    }
+    matched
+}
+
+pub(super) fn valid_control(control: &str, image_id: u32) -> bool {
+    if control.len() > 1024 || control.contains([';', '\x1b']) {
+        return false;
+    }
+    let mut action = false;
+    let mut format = false;
+    let mut image = false;
+    let mut quiet = false;
+    let mut cursor = false;
+    for field in control.split(',') {
+        let Some((key, value)) = field.split_once('=') else {
+            return false;
+        };
+        if key == "t"
+            || !matches!(
+                key,
+                "a" | "f"
+                    | "s"
+                    | "v"
+                    | "i"
+                    | "p"
+                    | "c"
+                    | "r"
+                    | "z"
+                    | "C"
+                    | "q"
+                    | "x"
+                    | "y"
+                    | "w"
+                    | "h"
+                    | "X"
+                    | "Y"
+            )
+        {
+            return false;
+        }
+        let numeric = value
+            .strip_prefix('-')
+            .unwrap_or(value)
+            .bytes()
+            .all(|byte| byte.is_ascii_digit())
+            && !value.is_empty();
+        if key != "a" && !numeric {
+            return false;
+        }
+        match key {
+            "a" => action = value == "T",
+            "f" => format = value == "32",
+            "i" => image = value.parse() == Ok(image_id),
+            "q" => quiet = value == "0",
+            "C" => cursor = value == "1",
+            _ => {}
+        }
+    }
+    action && format && image && quiet && cursor
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn matches_only_the_armed_image_and_preserves_unrelated_input() {
+        let mut matcher = ResponseMatcher::default();
+        let active = matcher.active_handle();
+        assert!(matcher.arm(7, 42));
+        assert!(active.load(std::sync::atomic::Ordering::Acquire));
+        assert!(!matcher.arm(8, 43));
+        assert_eq!(matcher.consume(b"typed"), None);
+        assert_eq!(matcher.consume(b"\x1b_Gi=41;OK\x1b\\"), None);
+        assert_eq!(
+            matcher.consume(b"\x1b_Gi=42;OK\x1b\\"),
+            Some(Some(Response {
+                transfer_id: 7,
+                image_id: 42,
+                success: true,
+            }))
+        );
+        assert!(!active.load(std::sync::atomic::Ordering::Acquire));
+    }
+
+    #[test]
+    fn explicit_error_is_reported_and_malformed_responses_are_ignored() {
+        let mut matcher = ResponseMatcher::default();
+        matcher.arm(9, 44);
+        assert_eq!(
+            matcher.consume(b"\x1b_Gi=44;ENOENT\x1b\\"),
+            Some(Some(Response {
+                transfer_id: 9,
+                image_id: 44,
+                success: false,
+            }))
+        );
+        matcher.arm(10, 45);
+        assert_eq!(matcher.consume(b"\x1b_Gi=45;broken"), None);
+        matcher.cancel(10);
+        matcher.arm(11, 46);
+        assert_eq!(matcher.consume(b"\x1b_Gi=46oops;OK\x1b\\"), None);
+    }
+
+    #[test]
+    fn validated_control_is_one_owned_rgba_transmit_and_display() {
+        assert!(valid_control(
+            "a=T,f=32,s=10,v=20,i=42,p=7,c=5,r=6,z=-1,C=1,q=0,x=2",
+            42
+        ));
+        for invalid in [
+            "a=T,f=24,i=42,C=1,q=0",
+            "a=T,f=32,i=41,C=1,q=0",
+            "a=T,t=f,f=32,i=42,C=1,q=0",
+            "a=p,f=32,i=42,C=1,q=0",
+        ] {
+            assert!(!valid_control(invalid, 42), "{invalid}");
+        }
+    }
+
+    #[test]
+    fn fragmented_response_is_held_before_generic_input_timeout() {
+        let mut matcher = ResponseMatcher::default();
+        let mut filter = InputFilter::default();
+        matcher.arm(19, 49);
+        for foreign in [
+            b"\x1b_Ga=p,i=49;OK\x1b\\".as_slice(),
+            b"\x1b_Gi=49oops;OK\x1b\\",
+        ] {
+            let (output, responses) = filter.push(foreign, &mut matcher);
+            assert_eq!(output, [foreign.to_vec()]);
+            assert!(responses.is_empty());
+        }
+        let (output, responses) = filter.push(b"typed\x1b_Gi=49;", &mut matcher);
+        assert_eq!(output, [b"typed".to_vec()]);
+        assert!(responses.is_empty());
+        let (output, responses) = filter.push(b"OK\x1b\\tail", &mut matcher);
+        assert_eq!(output, [b"tail".to_vec()]);
+        assert_eq!(responses[0].transfer_id, 19);
+        assert!(responses[0].success);
+    }
+
+    #[test]
+    fn attempted_output_failure_retires_response_without_forwarding_it() {
+        let mut matcher = ResponseMatcher::default();
+        let mut filter = InputFilter::default();
+        matcher.arm(20, 50);
+        matcher.retire(20);
+        for foreign in [
+            b"\x1b_Ga=p,i=50;OK\x1b\\".as_slice(),
+            b"\x1b_Gi=50oops;OK\x1b\\",
+        ] {
+            let (output, responses) = filter.push(foreign, &mut matcher);
+            assert_eq!(output, [foreign.to_vec()]);
+            assert!(responses.is_empty());
+        }
+        assert!(filter
+            .push(b"\x1b_Gi=50;EINVAL\x1b\\", &mut matcher)
+            .0
+            .is_empty());
+    }
+
+    #[test]
+    fn timed_out_expectation_requires_a_new_transfer_before_matching() {
+        let mut matcher = ResponseMatcher::default();
+        let started = std::time::Instant::now();
+        assert!(matcher.arm(12, 47));
+        matcher.start_at(12, started);
+        assert!(!matcher.arm(13, 48));
+        matcher.expire_at(started + RESPONSE_TIMEOUT);
+        assert!(matcher.arm(13, 48));
+        assert_eq!(matcher.consume(b"typed"), None);
+        assert_eq!(matcher.consume(b"\x1b_Gi=47;OK\x1b\\"), Some(None));
+        assert_eq!(
+            matcher.consume(b"\x1b_Gi=48;OK\x1b\\"),
+            Some(Some(Response {
+                transfer_id: 13,
+                image_id: 48,
+                success: true,
+            }))
+        );
+    }
+}
diff --git a/src/client/input.rs b/src/client/input.rs
index 4dc26eae..456f59b5 100644
--- a/src/client/input.rs
+++ b/src/client/input.rs
@@ -41,6 +41,9 @@ pub fn stdin_reader_loop(
     host_color_query_sent: bool,
     host_cell_size_query_sent: bool,
     host_mouse_capture_active: Arc<AtomicBool>,
+    host_sgr_pixels_active: Arc<AtomicBool>,
+    #[cfg(unix)] direct_response: Arc<std::sync::Mutex<super::direct_graphics::ResponseMatcher>>,
+    #[cfg(unix)] direct_response_active: Arc<AtomicBool>,
 ) {
     #[cfg(windows)]
     {
@@ -48,6 +51,7 @@ pub fn stdin_reader_loop(
             host_color_query_sent,
             host_cell_size_query_sent,
             host_mouse_capture_active,
+            host_sgr_pixels_active,
         );
         windows_stdin_reader_loop(event_tx, should_quit);
     }
@@ -59,6 +63,9 @@ pub fn stdin_reader_loop(
         host_color_query_sent,
         host_cell_size_query_sent,
         host_mouse_capture_active,
+        host_sgr_pixels_active,
+        direct_response,
+        direct_response_active,
     );
 }
 
@@ -69,6 +76,9 @@ fn unix_stdin_reader_loop(
     host_color_query_sent: bool,
     host_cell_size_query_sent: bool,
     host_mouse_capture_active: Arc<AtomicBool>,
+    host_sgr_pixels_active: Arc<AtomicBool>,
+    direct_response: Arc<std::sync::Mutex<super::direct_graphics::ResponseMatcher>>,
+    direct_response_active: Arc<AtomicBool>,
 ) {
     let stdin = io::stdin();
     let mut reader = stdin.lock();
@@ -83,15 +93,71 @@ fn unix_stdin_reader_loop(
         framer.host_cell_size_query_sent();
     }
     let mut pending_palette = Vec::new();
+    let mut pending_mode = None;
+    let mut last_geometry = None;
+    let mut direct_filter = super::direct_graphics::InputFilter::default();
 
     while !should_quit.load(Ordering::Acquire) {
+        if direct_filter.has_pending()
+            && stdin_read_ready(&reader, crate::raw_input::RAW_INPUT_IDLE_FLUSH_TIMEOUT_MS)
+                == Some(false)
+        {
+            let released = direct_response
+                .lock()
+                .ok()
+                .and_then(|mut matcher| direct_filter.flush_if_inactive(&mut matcher));
+            if let Some(data) = released {
+                if event_tx
+                    .blocking_send(ClientLoopEvent::StdinInput(data))
+                    .is_err()
+                {
+                    return;
+                }
+            }
+            continue;
+        }
         match reader.read(&mut scratch) {
             Ok(0) => break,
             Ok(n) => {
+                let sgr_pixels = *pending_mode
+                    .get_or_insert_with(|| host_sgr_pixels_active.load(Ordering::Acquire));
+                if sgr_pixels {
+                    last_geometry = retain_geometry(
+                        last_geometry,
+                        crate::input::mouse::HostGeometry::current(),
+                    );
+                }
+                let filtered = filter_direct_input(
+                    &scratch[..n],
+                    &mut direct_filter,
+                    &direct_response,
+                    &direct_response_active,
+                );
+                let chunks = if let Some((raw_chunks, responses)) = filtered {
+                    for response in responses {
+                        if event_tx
+                            .blocking_send(ClientLoopEvent::DirectGraphicsResponse(response))
+                            .is_err()
+                        {
+                            return;
+                        }
+                    }
+                    raw_chunks
+                        .into_iter()
+                        .flat_map(|chunk| framer.push(&chunk))
+                        .collect()
+                } else {
+                    framer.push(&scratch[..n])
+                };
+                if !framer.has_pending_input() {
+                    pending_mode = None;
+                }
                 if !send_unix_input_chunks(
-                    framer.push(&scratch[..n]),
+                    chunks,
                     &event_tx,
                     &mut pending_palette,
+                    sgr_pixels,
+                    last_geometry,
                 ) {
                     return;
                 }
@@ -104,8 +170,18 @@ fn unix_stdin_reader_loop(
                     let had_pending = framer.has_pending_input();
                     let chunks = framer.flush_timeout();
                     let held_escape = had_pending && chunks.is_empty();
-                    if !send_unix_input_chunks(chunks, &event_tx, &mut pending_palette)
-                        || !flush_unix_palette_input(&event_tx, &mut pending_palette)
+                    let sgr_pixels = pending_mode
+                        .unwrap_or_else(|| host_sgr_pixels_active.load(Ordering::Acquire));
+                    if !framer.has_pending_input() {
+                        pending_mode = None;
+                    }
+                    if !send_unix_input_chunks(
+                        chunks,
+                        &event_tx,
+                        &mut pending_palette,
+                        sgr_pixels,
+                        last_geometry,
+                    ) || !flush_unix_palette_input(&event_tx, &mut pending_palette)
                     {
                         return;
                     }
@@ -114,13 +190,20 @@ fn unix_stdin_reader_loop(
                             &reader,
                             crate::raw_input::RAW_INPUT_IDLE_FLUSH_TIMEOUT_MS,
                         ) == Some(false)
-                        && !send_unix_input_chunks(
-                            framer.flush_timeout(),
+                    {
+                        let chunks = framer.flush_timeout();
+                        if !framer.has_pending_input() {
+                            pending_mode = None;
+                        }
+                        if !send_unix_input_chunks(
+                            chunks,
                             &event_tx,
                             &mut pending_palette,
-                        )
-                    {
-                        return;
+                            sgr_pixels,
+                            last_geometry,
+                        ) {
+                            return;
+                        }
                     }
                 }
             }
@@ -135,10 +218,30 @@ fn unix_stdin_reader_loop(
 }
 
 #[cfg(unix)]
+fn filter_direct_input(
+    bytes: &[u8],
+    filter: &mut super::direct_graphics::InputFilter,
+    response: &std::sync::Mutex<super::direct_graphics::ResponseMatcher>,
+    active: &AtomicBool,
+) -> Option<(Vec<Vec<u8>>, Vec<super::direct_graphics::Response>)> {
+    if !active.load(Ordering::Acquire) && !filter.has_pending() {
+        return None;
+    }
+    Some(
+        response
+            .lock()
+            .map(|mut matcher| filter.push(bytes, &mut matcher))
+            .unwrap_or_else(|_| (vec![bytes.to_vec()], Vec::new())),
+    )
+}
+
+#[cfg(unix)]
 fn send_unix_input_chunks(
     chunks: Vec<Vec<u8>>,
     event_tx: &mpsc::Sender<ClientLoopEvent>,
     pending_palette: &mut Vec<Vec<u8>>,
+    sgr_pixels: bool,
+    geometry: Option<crate::input::mouse::HostGeometry>,
 ) -> bool {
     for data in chunks {
         let palette_response = std::str::from_utf8(&data)
@@ -160,10 +263,10 @@ fn send_unix_input_chunks(
         if !default_color_response && !flush_unix_palette_input(event_tx, pending_palette) {
             return false;
         }
-        if event_tx
-            .blocking_send(ClientLoopEvent::StdinInput(data))
-            .is_err()
-        {
+        let Some(event) = classify_unix_input(data, sgr_pixels, geometry) else {
+            continue;
+        };
+        if event_tx.blocking_send(event).is_err() {
             return false;
         }
     }
@@ -171,6 +274,26 @@ fn send_unix_input_chunks(
 }
 
 #[cfg(unix)]
+fn retain_geometry(
+    last: Option<crate::input::mouse::HostGeometry>,
+    observed: Option<crate::input::mouse::HostGeometry>,
+) -> Option<crate::input::mouse::HostGeometry> {
+    observed.or(last)
+}
+
+#[cfg(unix)]
+fn classify_unix_input(
+    data: Vec<u8>,
+    sgr_pixels: bool,
+    geometry: Option<crate::input::mouse::HostGeometry>,
+) -> Option<ClientLoopEvent> {
+    if sgr_pixels && crate::input::mouse::parse_report(&data).is_some() {
+        return geometry.map(|geometry| ClientLoopEvent::PixelMouse(data, geometry));
+    }
+    Some(ClientLoopEvent::StdinInput(data))
+}
+
+#[cfg(unix)]
 fn flush_unix_palette_input(
     event_tx: &mpsc::Sender<ClientLoopEvent>,
     pending_palette: &mut Vec<Vec<u8>>,
@@ -496,6 +619,50 @@ mod tests {
     }
 
     #[test]
+    fn inactive_direct_input_bypasses_filter() {
+        let response =
+            std::sync::Mutex::new(super::super::direct_graphics::ResponseMatcher::default());
+        let active = response.lock().unwrap().active_handle();
+        let mut filter = super::super::direct_graphics::InputFilter::default();
+        assert!(filter_direct_input(b"typed", &mut filter, &response, &active).is_none());
+        assert!(!filter.has_pending());
+    }
+
+    #[test]
+    fn pixel_mouse_classification_is_narrow_and_uses_read_geometry() {
+        let geometry = crate::input::mouse::HostGeometry::new(80, 24, 800, 480).unwrap();
+        let report = b"\x1b[<35;321;241M".to_vec();
+        let Some(ClientLoopEvent::PixelMouse(data, captured)) =
+            classify_unix_input(report.clone(), true, Some(geometry))
+        else {
+            panic!("expected dedicated pixel mouse event");
+        };
+        assert_eq!(data, report);
+        assert_eq!(captured, geometry);
+        assert!(classify_unix_input(report, true, None).is_none());
+
+        for raw in [
+            b"key".as_slice(),
+            b"\x1b[200~paste\x1b[201~".as_slice(),
+            b"\x1b_Gi=7;unrelated\x1b\\".as_slice(),
+            b"\x1b[<35;2;3Mtail".as_slice(),
+        ] {
+            let Some(ClientLoopEvent::StdinInput(data)) =
+                classify_unix_input(raw.to_vec(), true, Some(geometry))
+            else {
+                panic!("unrelated input must remain raw");
+            };
+            assert_eq!(data, raw);
+        }
+    }
+
+    #[test]
+    fn transient_geometry_failure_keeps_last_real_value() {
+        let geometry = crate::input::mouse::HostGeometry::new(80, 24, 800, 480).unwrap();
+        assert_eq!(retain_geometry(Some(geometry), None), Some(geometry));
+    }
+
+    #[test]
     fn palette_replies_are_forwarded_as_one_input_batch() {
         let (tx, mut rx) = mpsc::channel(4);
         let mut pending = Vec::new();
@@ -506,6 +673,8 @@ mod tests {
             ],
             &tx,
             &mut pending,
+            false,
+            None,
         ));
         assert!(rx.try_recv().is_err());
 
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 7932cb79..301377c9 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -12,9 +12,13 @@
 //! - Forwards OSC 52 clipboard writes from server to its own stdout
 //! - Displays sound/toast notifications forwarded from server
 
+#[cfg(unix)]
+mod direct_graphics;
 mod input;
 
 use std::collections::HashSet;
+#[cfg(unix)]
+use std::io::IsTerminal as _;
 use std::io::{self, BufRead, Write as _};
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
@@ -75,6 +79,12 @@ struct ClientState {
     sound_config: crate::config::SoundConfig,
     /// Whether this client may write Kitty graphics bytes to its host terminal.
     kitty_graphics_enabled: bool,
+    /// One bounded matcher, inactive unless a direct transmission is armed.
+    #[cfg(unix)]
+    direct_graphics_response: Arc<Mutex<direct_graphics::ResponseMatcher>>,
+    /// One server-retired direct transfer to suppress if it was still queued.
+    #[cfg(unix)]
+    retired_direct_graphics: Option<(u64, u32)>,
     /// Direct attach prefix escape state. None for full-app clients.
     attach_escape: Option<AttachEscapeState>,
     /// Rows scrolled for one direct-attach wheel notch.
@@ -356,9 +366,9 @@ fn setup_terminal_with_capabilities(
 
     if enable_client_protocols {
         if mouse_capture {
-            set_mouse_capture(true)?;
+            set_mouse_capture(true, false)?;
         } else {
-            set_mouse_capture(false)?;
+            set_mouse_capture(false, false)?;
         }
         execute!(io::stdout(), EnableBracketedPaste, EnableFocusChange)?;
         if host_color_scheme_reports {
@@ -370,9 +380,9 @@ fn setup_terminal_with_capabilities(
             write_host_color_scheme_report_mode(&mut io::stdout(), false)?;
         }
         if mouse_capture {
-            set_mouse_capture(true)?;
+            set_mouse_capture(true, false)?;
         } else {
-            set_mouse_capture(false)?;
+            set_mouse_capture(false, false)?;
         }
     }
 
@@ -549,10 +559,15 @@ fn restore_windows_input_mode_value(mode: u32) {
     }
 }
 
-fn set_mouse_capture(enabled: bool) -> io::Result<()> {
+fn set_mouse_capture(enabled: bool, sgr_pixels: bool) -> io::Result<()> {
     crate::terminal_modes::clear_host_mouse_reporting(&mut io::stdout())?;
     if enabled {
-        execute!(io::stdout(), EnableMouseCapture)
+        execute!(io::stdout(), EnableMouseCapture)?;
+        if sgr_pixels {
+            io::stdout().write_all(b"\x1b[?1016h")?;
+            io::stdout().flush()?;
+        }
+        Ok(())
     } else {
         match execute!(io::stdout(), DisableMouseCapture) {
             Ok(()) => Ok(()),
@@ -685,6 +700,44 @@ fn handshake_read_timeout() -> Duration {
     LOCAL_HANDSHAKE_READ_TIMEOUT
 }
 
+#[cfg(any(unix, test))]
+fn direct_graphics_profile_values(
+    term_program: &str,
+    term: &str,
+    kitty_window: bool,
+    blocked_transport: bool,
+    terminals: bool,
+) -> bool {
+    let supported = term_program.eq_ignore_ascii_case("ghostty")
+        || term_program.eq_ignore_ascii_case("wezterm")
+        || matches!(term, "xterm-ghostty" | "xterm-kitty" | "xterm-wezterm")
+        || kitty_window;
+    supported && !blocked_transport && terminals
+}
+
+#[cfg(unix)]
+fn direct_graphics_profile_allowed(direct_attach: bool) -> bool {
+    let term_program = std::env::var("TERM_PROGRAM").unwrap_or_default();
+    let term = std::env::var("TERM").unwrap_or_default();
+    direct_graphics_profile_values(
+        &term_program,
+        &term,
+        std::env::var_os("KITTY_WINDOW_ID").is_some(),
+        direct_attach
+            || is_remote_client_process()
+            || std::env::var_os("SSH_CONNECTION").is_some()
+            || std::env::var_os("SSH_TTY").is_some()
+            || std::env::var_os("TMUX").is_some()
+            || std::env::var_os("STY").is_some(),
+        io::stdin().is_terminal() && io::stdout().is_terminal(),
+    )
+}
+
+#[cfg(not(unix))]
+fn direct_graphics_profile_allowed(_direct_attach: bool) -> bool {
+    false
+}
+
 fn requested_keybindings() -> ClientKeybindings {
     match std::env::var(crate::remote::REMOTE_KEYBINDINGS_ENV_VAR)
         .ok()
@@ -726,6 +779,25 @@ fn set_handshake_recv_timeout(
         .map_err(ClientError::ConnectionFailed)
 }
 
+fn client_launch_mode(
+    direct_attach_requested: bool,
+    exact_cell_size: bool,
+    cell_width_px: u32,
+    cell_height_px: u32,
+) -> ClientLaunchMode {
+    if direct_attach_requested {
+        ClientLaunchMode::TerminalAttach
+    } else if exact_cell_size
+        && cell_width_px > 0
+        && cell_height_px > 0
+        && direct_graphics_profile_allowed(false)
+    {
+        ClientLaunchMode::AppDirectGraphics
+    } else {
+        ClientLaunchMode::App
+    }
+}
+
 /// Performs the client→server handshake.
 ///
 /// Sends Hello with the terminal size and protocol version, reads the Welcome
@@ -736,6 +808,7 @@ fn do_handshake(
     rows: u16,
     cell_width_px: u32,
     cell_height_px: u32,
+    exact_cell_size: bool,
     requested_encoding: RenderEncoding,
     direct_attach_requested: bool,
 ) -> Result<RenderEncoding, ClientError> {
@@ -752,11 +825,12 @@ fn do_handshake(
         cell_height_px,
         requested_encoding,
         keybindings: requested_keybindings(),
-        launch_mode: if direct_attach_requested {
-            ClientLaunchMode::TerminalAttach
-        } else {
-            ClientLaunchMode::App
-        },
+        launch_mode: client_launch_mode(
+            direct_attach_requested,
+            exact_cell_size,
+            cell_width_px,
+            cell_height_px,
+        ),
     };
     protocol::write_message(stream, &hello)
         .map_err(|e| ClientError::ConnectionFailed(io::Error::other(e.to_string())))?;
@@ -801,6 +875,11 @@ enum ClientLoopEvent {
     /// Raw input bytes from stdin.
     #[cfg(unix)]
     StdinInput(Vec<u8>),
+    /// One confirmed SGR pixel report with geometry captured by the reader.
+    #[cfg(unix)]
+    PixelMouse(Vec<u8>, crate::input::mouse::HostGeometry),
+    #[cfg(unix)]
+    DirectGraphicsResponse(direct_graphics::Response),
     /// Structured input events from platforms without Unix-style stdin bytes.
     #[cfg(windows)]
     StdinEvents(Vec<crate::protocol::ClientInputEvent>),
@@ -929,6 +1008,7 @@ fn connect_terminal_session_stream(
         rows,
         0,
         0,
+        false,
         RenderEncoding::TerminalAnsi,
         true,
     ) {
@@ -1149,7 +1229,7 @@ fn run_client_with_mode(
     };
 
     // Get the terminal geometry before handshake (before raw mode).
-    let (cols, rows, cell_width_px, cell_height_px) =
+    let (cols, rows, cell_width_px, cell_height_px, exact_cell_size) =
         initial_terminal_geometry(kitty_graphics_enabled);
 
     // Perform handshake while the stream is still in blocking mode.
@@ -1159,6 +1239,7 @@ fn run_client_with_mode(
         rows,
         cell_width_px,
         cell_height_px,
+        exact_cell_size,
         requested_encoding,
         direct_attach_requested,
     ) {
@@ -1296,6 +1377,10 @@ async fn run_client_loop(
         reported_size: (cols, rows),
         sound_config: config.sound_config,
         kitty_graphics_enabled: config.kitty_graphics_enabled,
+        #[cfg(unix)]
+        direct_graphics_response: Arc::new(Mutex::new(direct_graphics::ResponseMatcher::default())),
+        #[cfg(unix)]
+        retired_direct_graphics: None,
         attach_escape,
         #[cfg(unix)]
         mouse_scroll_lines: config.mouse_scroll_lines,
@@ -1309,6 +1394,7 @@ async fn run_client_loop(
     // Cell size reported by the host terminal, packed as width<<32 | height.
     // Zero means the host has not reported one.
     let reported_cell_size = Arc::new(AtomicU64::new(0));
+    let host_sgr_pixels_active = Arc::new(AtomicBool::new(false));
 
     // Channel for events from the stdin, resize, and server reader threads.
     let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<ClientLoopEvent>(256);
@@ -1323,6 +1409,14 @@ async fn run_client_loop(
     let stdin_quit = should_quit.clone();
     let stdin_tx = event_tx.clone();
     let stdin_mouse_capture_active = host_mouse_capture_active.clone();
+    let stdin_sgr_pixels_active = host_sgr_pixels_active.clone();
+    #[cfg(unix)]
+    let stdin_direct_response = state.direct_graphics_response.clone();
+    #[cfg(unix)]
+    let stdin_direct_response_active = stdin_direct_response
+        .lock()
+        .map(|matcher| matcher.active_handle())
+        .unwrap_or_default();
     std::thread::spawn(move || {
         input::stdin_reader_loop(
             stdin_tx,
@@ -1330,6 +1424,11 @@ async fn run_client_loop(
             will_query_host_terminal_theme,
             will_query_host_cell_size,
             stdin_mouse_capture_active,
+            stdin_sgr_pixels_active,
+            #[cfg(unix)]
+            stdin_direct_response,
+            #[cfg(unix)]
+            stdin_direct_response_active,
         );
     });
 
@@ -1475,6 +1574,30 @@ async fn run_client_loop(
                     return Err(ClientError::ConnectionLost(e));
                 }
             }
+            #[cfg(unix)]
+            ClientLoopEvent::DirectGraphicsResponse(response) => {
+                let message = ClientMessage::GraphicsTransmissionResult {
+                    transfer_id: response.transfer_id,
+                    image_id: response.image_id,
+                    success: response.success,
+                };
+                if let Err(err) = write_to_server(&mut write_stream, &message) {
+                    return Err(ClientError::ConnectionLost(err));
+                }
+            }
+            #[cfg(unix)]
+            ClientLoopEvent::PixelMouse(data, geometry) => {
+                let message = ClientMessage::InputPixels {
+                    data,
+                    cols: geometry.cols,
+                    rows: geometry.rows,
+                    width_px: geometry.width_px,
+                    height_px: geometry.height_px,
+                };
+                if let Err(err) = write_to_server(&mut write_stream, &message) {
+                    return Err(ClientError::ConnectionLost(err));
+                }
+            }
             #[cfg(windows)]
             ClientLoopEvent::StdinEvents(events) => {
                 if state.attach_escape.is_some() {
@@ -1578,6 +1701,98 @@ async fn run_client_loop(
                         warn!(err = %err, "failed to emit terminal bell");
                     }
                 }
+                ServerMessage::GraphicsFile {
+                    path,
+                    expected_len,
+                    image_id,
+                    transfer_id,
+                    leading,
+                    control,
+                } => {
+                    #[cfg(unix)]
+                    {
+                        if state.retired_direct_graphics.take() == Some((transfer_id, image_id)) {
+                            continue;
+                        }
+                        let valid = state.kitty_graphics_enabled
+                            && usize::try_from(expected_len).ok().is_some_and(|len| {
+                                crate::pane_graphics_files::validate_direct_source(
+                                    std::path::Path::new(&path),
+                                    len,
+                                )
+                                .is_ok()
+                            })
+                            && direct_graphics::valid_control(&control, image_id)
+                            && state
+                                .direct_graphics_response
+                                .lock()
+                                .is_ok_and(|mut matcher| matcher.arm(transfer_id, image_id));
+                        let sent = if valid {
+                            let mut command = Vec::new();
+                            crate::kitty_graphics::encode_kitty_regular_file(
+                                &mut command,
+                                &leading,
+                                &control,
+                                &path,
+                            );
+                            let mut stdout = io::stdout();
+                            let written = stdout
+                                .write_all(&command)
+                                .and_then(|()| stdout.flush())
+                                .is_ok();
+                            if written {
+                                record_received_kitty_graphics(&command);
+                            }
+                            written
+                        } else {
+                            false
+                        };
+                        if sent {
+                            if let Ok(mut matcher) = state.direct_graphics_response.lock() {
+                                matcher.start(transfer_id);
+                            }
+                            let started = ClientMessage::GraphicsTransmissionStarted {
+                                transfer_id,
+                                image_id,
+                            };
+                            if let Err(err) = write_to_server(&mut write_stream, &started) {
+                                return Err(ClientError::ConnectionLost(err));
+                            }
+                        } else {
+                            if let Ok(mut matcher) = state.direct_graphics_response.lock() {
+                                if valid {
+                                    matcher.retire(transfer_id);
+                                } else {
+                                    matcher.cancel(transfer_id);
+                                }
+                            }
+                            let result = ClientMessage::GraphicsTransmissionResult {
+                                transfer_id,
+                                image_id,
+                                success: false,
+                            };
+                            if let Err(err) = write_to_server(&mut write_stream, &result) {
+                                return Err(ClientError::ConnectionLost(err));
+                            }
+                        }
+                    }
+                    #[cfg(not(unix))]
+                    let _ = (path, expected_len, image_id, transfer_id, leading, control);
+                }
+                ServerMessage::GraphicsTransmissionRetired {
+                    transfer_id,
+                    image_id,
+                } => {
+                    #[cfg(unix)]
+                    {
+                        state.retired_direct_graphics = Some((transfer_id, image_id));
+                        if let Ok(mut matcher) = state.direct_graphics_response.lock() {
+                            matcher.retire(transfer_id);
+                        }
+                    }
+                    #[cfg(not(unix))]
+                    let _ = (transfer_id, image_id);
+                }
                 ServerMessage::ServerShutdown { reason } => {
                     return Err(ClientError::ServerShutdown { reason });
                 }
@@ -1604,17 +1819,24 @@ async fn run_client_loop(
                         &mut state.remote_image_paste_key,
                     );
                 }
-                ServerMessage::MouseCapture { enabled } => {
-                    let desired = enabled;
-                    if desired != state.mouse_capture_active {
-                        set_mouse_capture(desired).map_err(ClientError::ConnectionFailed)?;
+                ServerMessage::MouseCapture {
+                    enabled,
+                    sgr_pixels,
+                } => {
+                    let next_sgr_pixels = enabled && sgr_pixels;
+                    let mouse_mode_changed = enabled != state.mouse_capture_active
+                        || next_sgr_pixels != host_sgr_pixels_active.load(Ordering::Acquire);
+                    if mouse_mode_changed {
+                        set_mouse_capture(enabled, next_sgr_pixels)
+                            .map_err(ClientError::ConnectionFailed)?;
                         #[cfg(windows)]
-                        if windows_vti_input_backend_enabled() {
+                        if enabled && windows_vti_input_backend_enabled() {
                             let _ = enable_windows_virtual_terminal_input();
                         }
-                        state.mouse_capture_active = desired;
-                        host_mouse_capture_active.store(desired, Ordering::Release);
                     }
+                    state.mouse_capture_active = enabled;
+                    host_mouse_capture_active.store(enabled, Ordering::Release);
+                    host_sgr_pixels_active.store(next_sgr_pixels, Ordering::Release);
                 }
                 ServerMessage::KittyKeyboardReportAll { enabled } => {
                     if enabled != state.keyboard_report_all_active {
@@ -1643,7 +1865,12 @@ async fn run_client_loop(
                     "server closed connection",
                 )));
             }
-            ClientLoopEvent::Timer => {}
+            ClientLoopEvent::Timer => {
+                #[cfg(unix)]
+                if let Ok(mut matcher) = state.direct_graphics_response.lock() {
+                    matcher.expire();
+                }
+            }
         }
     }
 
@@ -2204,8 +2431,10 @@ fn ioctl_cell_size() -> Option<(u32, u32)> {
 }
 
 /// Cell size used when the ioctl reports no pixels.
-fn cell_size_fallback(reported: u64) -> (u32, u32) {
-    unpack_cell_size(reported).unwrap_or((DEFAULT_CELL_WIDTH_PX, DEFAULT_CELL_HEIGHT_PX))
+fn cell_size_fallback(reported: u64, last: Option<(u32, u32)>) -> (u32, u32) {
+    unpack_cell_size(reported)
+        .or(last.filter(|(width, height)| *width > 0 && *height > 0))
+        .unwrap_or((DEFAULT_CELL_WIDTH_PX, DEFAULT_CELL_HEIGHT_PX))
 }
 
 #[cfg(any(unix, test))]
@@ -2222,20 +2451,35 @@ fn unpack_cell_size(packed: u64) -> Option<(u32, u32)> {
 fn current_terminal_geometry(
     kitty_graphics_enabled: bool,
     reported_cell_size: &AtomicU64,
+    last_cell_size: Option<(u32, u32)>,
 ) -> (u16, u16, u32, u32) {
     let (cols, rows) = crossterm::terminal::size().unwrap_or((80, 24));
     if !kitty_graphics_enabled {
         return (cols, rows, 0, 0);
     }
-    let (cell_width_px, cell_height_px) = ioctl_cell_size()
-        .unwrap_or_else(|| cell_size_fallback(reported_cell_size.load(Ordering::Acquire)));
+    let (cell_width_px, cell_height_px) = ioctl_cell_size().unwrap_or_else(|| {
+        cell_size_fallback(reported_cell_size.load(Ordering::Acquire), last_cell_size)
+    });
     (cols, rows, cell_width_px, cell_height_px)
 }
 
-/// Reads the terminal geometry before the handshake, before any host cell
-/// size report can exist.
-fn initial_terminal_geometry(kitty_graphics_enabled: bool) -> (u16, u16, u32, u32) {
-    current_terminal_geometry(kitty_graphics_enabled, &AtomicU64::new(0))
+/// Reads terminal geometry before the handshake. Direct graphics is eligible
+/// only when the host supplied exact pixel dimensions through the ioctl.
+fn initial_terminal_geometry(kitty_graphics_enabled: bool) -> (u16, u16, u32, u32, bool) {
+    let (cols, rows) = crossterm::terminal::size().unwrap_or((80, 24));
+    if !kitty_graphics_enabled {
+        return (cols, rows, 0, 0, false);
+    }
+    match ioctl_cell_size() {
+        Some((width, height)) => (cols, rows, width, height, true),
+        None => (
+            cols,
+            rows,
+            DEFAULT_CELL_WIDTH_PX,
+            DEFAULT_CELL_HEIGHT_PX,
+            false,
+        ),
+    }
 }
 
 /// Reports polled changes and signalled resizes that return to the same size.
@@ -2272,7 +2516,11 @@ fn resize_poll_loop(
     while !should_quit.load(Ordering::Acquire) {
         std::thread::sleep(Duration::from_millis(100));
         let signalled = crate::platform::take_terminal_resize_signal();
-        let new_size = current_terminal_geometry(kitty_graphics_enabled, reported_cell_size);
+        let new_size = current_terminal_geometry(
+            kitty_graphics_enabled,
+            reported_cell_size,
+            Some((last_size.2, last_size.3)),
+        );
         if resize_report_required(signalled, new_size, last_size) {
             last_size = new_size;
             if resize_tx
@@ -2388,6 +2636,40 @@ mod tests {
         assert!(resize_report_required(false, (120, 40, 9, 18), size));
     }
 
+    #[test]
+    fn approximate_cell_size_never_enables_direct_graphics() {
+        assert_eq!(
+            client_launch_mode(false, false, 8, 16),
+            ClientLaunchMode::App
+        );
+        assert_eq!(
+            client_launch_mode(true, false, 8, 16),
+            ClientLaunchMode::TerminalAttach
+        );
+    }
+
+    #[test]
+    fn direct_graphics_profile_is_narrow_and_transport_safe() {
+        for (program, term, kitty, expected) in [
+            ("ghostty", "", false, true),
+            ("WezTerm", "", false, true),
+            ("", "xterm-kitty", false, true),
+            ("", "xterm-256color", true, true),
+            ("", "xterm-256color", false, false),
+        ] {
+            assert_eq!(
+                direct_graphics_profile_values(program, term, kitty, false, true),
+                expected
+            );
+        }
+        assert!(!direct_graphics_profile_values(
+            "ghostty", "", false, true, true
+        ));
+        assert!(!direct_graphics_profile_values(
+            "ghostty", "", false, false, false
+        ));
+    }
+
     fn restore_env_var(key: &str, value: Option<OsString>) {
         if let Some(value) = value {
             std::env::set_var(key, value);
@@ -2758,11 +3040,15 @@ mod tests {
     }
 
     #[test]
-    fn cell_size_fallback_prefers_reported_size_over_default() {
-        assert_eq!(cell_size_fallback(0), (8, 16));
-        assert_eq!(cell_size_fallback(pack_cell_size(10, 21)), (10, 21));
-        assert_eq!(cell_size_fallback(pack_cell_size(10, 0)), (8, 16));
-        assert_eq!(cell_size_fallback(pack_cell_size(0, 21)), (8, 16));
+    fn cell_size_fallback_prefers_reported_then_previous_size() {
+        assert_eq!(cell_size_fallback(0, None), (8, 16));
+        assert_eq!(cell_size_fallback(0, Some((11, 22))), (11, 22));
+        assert_eq!(
+            cell_size_fallback(pack_cell_size(10, 21), Some((11, 22))),
+            (10, 21)
+        );
+        assert_eq!(cell_size_fallback(pack_cell_size(10, 0), None), (8, 16));
+        assert_eq!(cell_size_fallback(pack_cell_size(0, 21), None), (8, 16));
     }
 
     #[test]