feat(display): wire Phase 2 IPC — present, subscribe_input, poll_events

Implements the three unimplemented display IPC methods from PG-42
(primalSpring Display Phase 2 handoff):

- display.present: accepts pixel data via base64 inline or shared
  memory path, writes to window's DRM dumb buffer via with_mapping
- display.subscribe_input: sets server-side input focus to a window
- display.poll_events: drains buffered evdev events non-blockingly

Supporting changes:
- WindowManager::present_window() writes raw pixels to a window's
  framebuffer via DRM mapping
- InputManager::poll_events() now drains the mpsc channel with
  try_recv instead of returning empty Vec
- InputManager::empty() constructor for fallback when discovery fails
- DisplayServer integrates InputManager, passes to dispatch
- DisplayClient gains present(), present_shm(), subscribe_input(),
  poll_events() matching the server-side methods
- DisplayMethod::Present expanded with data/shm_path fields
- 8 new tests for Phase 2 dispatch + InputManager drain behavior

petalTongue can now render through toadStool's framebuffer and
receive input events over JSON-RPC IPC.

Made-with: Cursor

Author: westgate

crates/runtime/display/Cargo.toml

@@ -31,6 +31,7 @@ futures = { workspace = true }
 tarpc = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+base64 = { workspace = true }
 
 # Error handling
 thiserror = { workspace = true }

crates/runtime/display/src/input/mod.rs

@@ -66,6 +66,18 @@ pub struct InputManager {
 }
 
 impl InputManager {
+    /// Create an empty input manager with no devices (fallback when discovery fails).
+    #[must_use]
+    pub fn empty() -> Self {
+        let (event_tx, event_rx) = mpsc::channel(1000);
+        Self {
+            devices: vec![],
+            shared_focus: Arc::new(RwLock::new(None)),
+            event_tx,
+            event_rx: Some(event_rx),
+        }
+    }
+
     /// Discover and initialize input devices
     ///
     /// **Self-knowledge**: Discovers own hardware at runtime!
@@ -214,19 +226,25 @@ impl InputManager {
         )
     }
 
-    /// Poll for input events
+    /// Poll for input events (non-blocking).
     ///
-    /// Non-blocking check for pending events.
+    /// Drains all pending events from the internal channel and returns them.
+    /// Returns an empty `Vec` when no events are queued. Safe to call in a
+    /// tight loop — uses `try_recv` internally.
     ///
-    /// Note: This is a simplified API. For streaming, use `subscribe_events()`.
+    /// For streaming, use `subscribe_events()` instead.
     ///
     /// # Errors
     ///
     /// Currently always returns `Ok`; reserved for future error cases.
-    pub const fn poll_events(&mut self) -> Result<Vec<InputEvent>> {
-        // For now, return empty - real streaming happens via subscribe_events()
-        // This is here for compatibility with the old API
-        Ok(Vec::new())
+    pub fn poll_events(&mut self) -> Result<Vec<InputEvent>> {
+        let mut events = Vec::new();
+        if let Some(rx) = &mut self.event_rx {
+            while let Ok(event) = rx.try_recv() {
+                events.push(event);
+            }
+        }
+        Ok(events)
     }
 
     /// Route events to a specific window
@@ -341,4 +359,40 @@ mod tests {
 
         assert!(received.is_some());
     }
+
+    #[tokio::test]
+    async fn test_poll_events_drains_channel() {
+        let mut manager = InputManager::empty();
+        let window = WindowId::new();
+
+        let event_a = InputEvent::KeyPress {
+            key: KeyCode::A,
+            modifiers: Modifiers::default(),
+            window,
+        };
+        let event_b = InputEvent::KeyRelease {
+            key: KeyCode::A,
+            modifiers: Modifiers::default(),
+            window,
+        };
+
+        manager.inject_event(event_a).await.unwrap();
+        manager.inject_event(event_b).await.unwrap();
+
+        let polled = manager.poll_events().unwrap();
+        assert_eq!(polled.len(), 2, "should drain both events");
+
+        let polled_again = manager.poll_events().unwrap();
+        assert!(
+            polled_again.is_empty(),
+            "channel should be empty after drain"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_empty_manager_poll_returns_empty() {
+        let mut manager = InputManager::empty();
+        let events = manager.poll_events().unwrap();
+        assert!(events.is_empty());
+    }
 }

crates/runtime/display/src/ipc/client/operations.rs

@@ -1,8 +1,11 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
-//! Display IPC operations (windows, capabilities, endpoint metadata).
+//! Display IPC operations (windows, capabilities, presentation, input, endpoint metadata).
+
+use base64::Engine;
 
 use super::DisplayClient;
 use crate::DisplayError;
+use crate::input::InputEvent;
 use crate::ipc::types::DisplayCapabilitiesInfo;
 use crate::window::{CreateWindowRequest, WindowId, WindowInfo};
 
@@ -136,6 +139,124 @@ impl DisplayClient {
         }
     }
 
+    /// Present raw RGBA pixel data to a window (inline base64 mode).
+    ///
+    /// The pixel data is base64-encoded and sent in the JSON-RPC request.
+    /// For large framebuffers, prefer [`present_shm`](Self::present_shm).
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the IPC request fails or the server returns an error.
+    pub async fn present(&mut self, window_id: WindowId, pixels: &[u8]) -> crate::Result<()> {
+        let data_b64 = base64::engine::general_purpose::STANDARD.encode(pixels);
+        let req = crate::ipc::types::JsonRpcRequest::new(
+            "display.present",
+            Some(serde_json::json!({
+                "window_id": window_id.as_string(),
+                "data": data_b64,
+            })),
+        );
+
+        let response = self.send_request(req).await?;
+
+        if let Some(error) = response.error {
+            Err(DisplayError::IpcError(format!(
+                "Server error: {}",
+                error.message
+            )))
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Present framebuffer via shared memory path (zero-copy mode).
+    ///
+    /// The client writes raw RGBA pixel data to the given path (e.g.
+    /// `/dev/shm/toadstool-fb-{window_id}`), then calls this method.
+    /// The server reads the file and copies it to the DRM framebuffer.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the IPC request fails or the server returns an error.
+    pub async fn present_shm(&mut self, window_id: WindowId, shm_path: &str) -> crate::Result<()> {
+        let req = crate::ipc::types::JsonRpcRequest::new(
+            "display.present",
+            Some(serde_json::json!({
+                "window_id": window_id.as_string(),
+                "shm_path": shm_path,
+            })),
+        );
+
+        let response = self.send_request(req).await?;
+
+        if let Some(error) = response.error {
+            Err(DisplayError::IpcError(format!(
+                "Server error: {}",
+                error.message
+            )))
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Subscribe to input events for a window.
+    ///
+    /// Sets the server-side input focus to the given window so subsequent
+    /// [`poll_events`](Self::poll_events) calls return events for it.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the IPC request fails or the server returns an error.
+    pub async fn subscribe_input(&mut self, window_id: WindowId) -> crate::Result<()> {
+        let req = crate::ipc::types::JsonRpcRequest::new(
+            "display.subscribe_input",
+            Some(serde_json::json!({"window_id": window_id.as_string()})),
+        );
+
+        let response = self.send_request(req).await?;
+
+        if let Some(error) = response.error {
+            Err(DisplayError::IpcError(format!(
+                "Server error: {}",
+                error.message
+            )))
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Poll for pending input events (non-blocking).
+    ///
+    /// Returns all events buffered since the last poll. Returns an empty
+    /// `Vec` when no events are queued.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the IPC request fails or the server returns an error.
+    pub async fn poll_events(&mut self) -> crate::Result<Vec<InputEvent>> {
+        let req = crate::ipc::types::JsonRpcRequest::new("display.poll_events", None);
+
+        let response = self.send_request(req).await?;
+
+        if let Some(result) = response.result {
+            let events: Vec<InputEvent> = serde_json::from_value(
+                result
+                    .get("events")
+                    .cloned()
+                    .unwrap_or(serde_json::json!([])),
+            )
+            .map_err(|e| DisplayError::IpcError(format!("Parse error: {e}")))?;
+            Ok(events)
+        } else if let Some(error) = response.error {
+            Err(DisplayError::IpcError(format!(
+                "Server error: {}",
+                error.message
+            )))
+        } else {
+            Err(DisplayError::IpcError("Invalid response".to_string()))
+        }
+    }
+
     /// Get endpoint string for display purposes
     ///
     /// **Helper for health checks and monitoring**