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

1 files changed, 181 insertions, 12 deletions

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());