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, 347 insertions, 0 deletions

diff --git a/src/app/pane_graphics.rs b/src/app/pane_graphics.rs
new file mode 100644
index 00000000..c1d4841a
--- /dev/null
+++ b/src/app/pane_graphics.rs
@@ -0,0 +1,347 @@
+use std::collections::HashMap;
+use std::hash::{Hash, Hasher};
+
+use crate::api::schema::{PaneGraphicsFormat, PaneGraphicsPlacementParams};
+use crate::layout::PaneId;
+
+pub(crate) type Key = (PaneId, String);
+
+pub(crate) const DIRECT_DELIVERY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+pub(crate) const DIRECT_RESPONSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+pub(crate) const DIRECT_OUTER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(9);
+
+#[derive(Debug)]
+pub(crate) enum Backing {
+    Inline(Vec<u8>),
+    Direct(crate::pane_graphics_files::Lease),
+    Resident { len: usize, client_id: u64 },
+}
+
+#[derive(Debug)]
+pub(crate) struct Layer {
+    pub(crate) format: PaneGraphicsFormat,
+    pub(crate) image_width: u32,
+    pub(crate) image_height: u32,
+    pub(crate) backing: Backing,
+    pub(crate) data_fingerprint: u64,
+    pub(crate) render: PaneGraphicsPlacementParams,
+    pub(crate) z_index: i32,
+}
+
+#[derive(Debug)]
+pub(crate) struct DirectGate {
+    pub(crate) transfer_id: u64,
+    pub(crate) client_id: u64,
+    pub(crate) deadline: std::time::Instant,
+    pub(crate) written: bool,
+    pub(crate) success_response: String,
+    pub(crate) respond_to: std::sync::mpsc::Sender<String>,
+}
+
+#[derive(Debug)]
+pub(crate) struct Slot {
+    pub(crate) host_image_id: u32,
+    pub(crate) layer: Option<Layer>,
+    pub(crate) stream_owner: Option<String>,
+    pub(crate) stream_active: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
+    pub(crate) direct_gate: Option<DirectGate>,
+}
+
+impl Slot {
+    #[cfg(test)]
+    pub(crate) fn test(host_image_id: u32, layer: Option<Layer>) -> Self {
+        Self {
+            host_image_id,
+            layer,
+            stream_owner: None,
+            stream_active: None,
+            direct_gate: None,
+        }
+    }
+
+    pub(crate) fn stream_is_active(&self) -> bool {
+        self.stream_active
+            .as_ref()
+            .is_some_and(|active| active.load(std::sync::atomic::Ordering::Acquire))
+    }
+}
+
+impl Drop for Slot {
+    fn drop(&mut self) {
+        if let Some(active) = &self.stream_active {
+            active.store(false, std::sync::atomic::Ordering::Release);
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct Runtime {
+    pub(crate) slots: HashMap<Key, Slot>,
+    revision: u64,
+    next_host_image_id: u32,
+}
+
+impl Default for Runtime {
+    fn default() -> Self {
+        Self {
+            slots: HashMap::new(),
+            revision: 0,
+            next_host_image_id: 1 << 31,
+        }
+    }
+}
+
+impl Runtime {
+    pub(crate) fn revision(&self) -> u64 {
+        self.revision
+    }
+
+    pub(crate) fn mark_changed(&mut self) {
+        self.revision = self.revision.wrapping_add(1);
+    }
+
+    pub(crate) fn clear(&mut self) {
+        self.slots.clear();
+        self.mark_changed();
+    }
+
+    pub(crate) fn reserve_image_id(&mut self, key: &Key) -> Option<u32> {
+        if let Some(id) = self.slots.get(key).map(|slot| slot.host_image_id) {
+            return Some(id);
+        }
+        for _ in 0..=self.slots.len() {
+            let id = self.next_host_image_id | (1 << 31);
+            self.next_host_image_id = id.wrapping_add(1) | (1 << 31);
+            if self.slots.values().all(|slot| slot.host_image_id != id) {
+                return Some(id);
+            }
+        }
+        None
+    }
+
+    pub(crate) fn layer_count(&self, pane_id: PaneId) -> usize {
+        self.slots.keys().filter(|(id, _)| *id == pane_id).count()
+    }
+
+    pub(crate) fn can_add_slot(&self, key: &Key) -> bool {
+        self.slots.contains_key(key)
+            || self.slots.len() < crate::api::schema::panes::PANE_GRAPHICS_MAX_LAYERS_TOTAL
+    }
+
+    pub(crate) fn can_store_inline(&self, key: &Key, len: usize) -> bool {
+        let replaced = self
+            .slots
+            .get(key)
+            .and_then(|slot| slot.layer.as_ref())
+            .and_then(Layer::inline_data)
+            .map_or(0, <[u8]>::len);
+        self.inline_bytes()
+            .saturating_sub(replaced)
+            .saturating_add(len)
+            <= crate::api::schema::panes::PANE_GRAPHICS_MAX_INLINE_BYTES_TOTAL
+    }
+
+    fn inline_bytes(&self) -> usize {
+        self.slots
+            .values()
+            .filter_map(|slot| slot.layer.as_ref())
+            .filter_map(Layer::inline_data)
+            .map(<[u8]>::len)
+            .sum()
+    }
+
+    pub(crate) fn active_for_pane(&self, pane_id: PaneId) -> bool {
+        self.slots
+            .iter()
+            .any(|((id, _), slot)| *id == pane_id && slot.layer.is_some())
+    }
+
+    pub(crate) fn attach_stream_active(
+        &mut self,
+        key: &Key,
+        owner: &str,
+        active: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    ) {
+        if let Some(slot) = self
+            .slots
+            .get_mut(key)
+            .filter(|slot| slot.stream_owner.as_deref() == Some(owner))
+        {
+            slot.stream_active = Some(active);
+        }
+    }
+
+    pub(crate) fn retain_live_panes(&mut self, state: &crate::app::state::AppState) -> bool {
+        if self.slots.is_empty() {
+            return false;
+        }
+        let before = self.slots.len();
+        self.slots.retain(|(pane_id, _), _| {
+            state
+                .workspaces
+                .iter()
+                .any(|workspace| workspace.pane_state(*pane_id).is_some())
+        });
+        if self.slots.len() == before {
+            return false;
+        }
+        self.mark_changed();
+        true
+    }
+}
+
+impl Layer {
+    pub(crate) fn inline(
+        format: PaneGraphicsFormat,
+        image_width: u32,
+        image_height: u32,
+        data: Vec<u8>,
+        render: PaneGraphicsPlacementParams,
+        z_index: i32,
+    ) -> Self {
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        data.hash(&mut hasher);
+        Self {
+            format,
+            image_width,
+            image_height,
+            backing: Backing::Inline(data),
+            data_fingerprint: hasher.finish(),
+            render,
+            z_index,
+        }
+    }
+
+    pub(crate) fn direct(
+        image_width: u32,
+        image_height: u32,
+        lease: crate::pane_graphics_files::Lease,
+        render: PaneGraphicsPlacementParams,
+        z_index: i32,
+    ) -> Self {
+        let data_fingerprint = lease.fingerprint();
+        Self {
+            format: PaneGraphicsFormat::Rgba,
+            image_width,
+            image_height,
+            backing: Backing::Direct(lease),
+            data_fingerprint,
+            render,
+            z_index,
+        }
+    }
+
+    pub(crate) fn data_len(&self) -> usize {
+        match &self.backing {
+            Backing::Inline(data) => data.len(),
+            Backing::Direct(lease) => lease.len(),
+            Backing::Resident { len, .. } => *len,
+        }
+    }
+
+    pub(crate) fn inline_data(&self) -> Option<&[u8]> {
+        match &self.backing {
+            Backing::Inline(data) => Some(data),
+            Backing::Direct(_) | Backing::Resident { .. } => None,
+        }
+    }
+
+    pub(crate) fn direct_lease(&self) -> Option<&crate::pane_graphics_files::Lease> {
+        match &self.backing {
+            Backing::Direct(lease) => Some(lease),
+            Backing::Inline(_) | Backing::Resident { .. } => None,
+        }
+    }
+
+    pub(crate) fn mark_resident(&mut self, client_id: u64) -> bool {
+        let Backing::Direct(lease) = &self.backing else {
+            return false;
+        };
+        self.backing = Backing::Resident {
+            len: lease.len(),
+            client_id,
+        };
+        true
+    }
+
+    pub(crate) fn resident_client(&self) -> Option<u64> {
+        match self.backing {
+            Backing::Resident { client_id, .. } => Some(client_id),
+            Backing::Inline(_) | Backing::Direct(_) => None,
+        }
+    }
+
+    pub(crate) fn terminal_only(&self) -> bool {
+        matches!(self.backing, Backing::Direct(_) | Backing::Resident { .. })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn stream_liveness_attaches_only_to_the_exact_owned_layer() {
+        let mut runtime = Runtime::default();
+        let first = (PaneId::from_raw(1), "first".into());
+        let second = (PaneId::from_raw(2), "second".into());
+        for (key, id) in [(&first, 1), (&second, 2)] {
+            runtime.slots.insert(
+                key.clone(),
+                Slot {
+                    host_image_id: (1 << 31) | id,
+                    layer: None,
+                    stream_owner: Some("shared-owner".into()),
+                    stream_active: None,
+                    direct_gate: None,
+                },
+            );
+        }
+        let active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
+
+        runtime.attach_stream_active(&second, "shared-owner", active.clone());
+
+        assert!(runtime.slots[&first].stream_active.is_none());
+        assert!(std::sync::Arc::ptr_eq(
+            runtime.slots[&second].stream_active.as_ref().unwrap(),
+            &active
+        ));
+    }
+
+    #[test]
+    fn runtime_bounds_total_slots_and_inline_bytes() {
+        let mut runtime = Runtime::default();
+        for index in 0..crate::api::schema::panes::PANE_GRAPHICS_MAX_LAYERS_TOTAL {
+            let key = (PaneId::from_raw(1), format!("layer-{index}"));
+            runtime
+                .slots
+                .insert(key, Slot::test((1 << 31) | index as u32, None));
+        }
+        assert!(!runtime.can_add_slot(&(PaneId::from_raw(2), "extra".into())));
+
+        let mut runtime = Runtime::default();
+        let key = (PaneId::from_raw(1), "primary".into());
+        runtime.slots.insert(
+            key.clone(),
+            Slot::test(
+                1 << 31,
+                Some(Layer::inline(
+                    PaneGraphicsFormat::Rgba,
+                    1,
+                    1,
+                    vec![0; 4],
+                    PaneGraphicsPlacementParams::default(),
+                    0,
+                )),
+            ),
+        );
+        assert!(runtime.can_store_inline(
+            &key,
+            crate::api::schema::panes::PANE_GRAPHICS_MAX_INLINE_BYTES_TOTAL,
+        ));
+        assert!(!runtime.can_store_inline(
+            &(PaneId::from_raw(2), "primary".into()),
+            crate::api::schema::panes::PANE_GRAPHICS_MAX_INLINE_BYTES_TOTAL,
+        ));
+    }
+}