feat: add native windows beta support

1 files changed, 196 insertions, 15 deletions

diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 931f1100..4f0f12bc 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -5,18 +5,20 @@
 //! `HeadlessServer`.
 
 use std::io::{self, Write};
-use std::os::unix::net::UnixStream;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
+use interprocess::local_socket::traits::Stream as _;
+use interprocess::TryClone as _;
 use tokio::sync::mpsc;
 use tracing::{debug, warn};
 
+use crate::ipc::LocalStream;
 use crate::protocol::{
-    self, AttachScrollDirection, AttachScrollSource, ClientKeybindings, ClientLaunchMode,
-    ClientMessage, RenderEncoding, ServerMessage, MAX_CLIPBOARD_IMAGE_PAYLOAD, MAX_FRAME_SIZE,
-    MAX_GRAPHICS_FRAME_SIZE, PROTOCOL_VERSION,
+    self, AttachScrollDirection, AttachScrollSource, ClientInputEvent, ClientKeybindings,
+    ClientLaunchMode, ClientMessage, RenderEncoding, ServerMessage, MAX_CLIPBOARD_IMAGE_PAYLOAD,
+    MAX_FRAME_SIZE, MAX_GRAPHICS_FRAME_SIZE, PROTOCOL_VERSION,
 };
 
 /// Minimum accepted attached client size.
@@ -35,6 +37,8 @@ const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4);
 
 /// Maximum input payload size (bytes) for a single `ClientMessage::Input`.
 const MAX_INPUT_PAYLOAD: usize = 1024 * 1024; // 1 MB
+/// Maximum structured input events accepted in one client message.
+const MAX_INPUT_EVENT_BATCH: usize = 4096;
 
 /// Channels owned by the server side of a client writer thread.
 #[derive(Clone, Debug)]
@@ -62,6 +66,11 @@ pub(crate) enum ServerEvent {
     },
     /// A client sent an input message.
     ClientInput { client_id: u64, data: Vec<u8> },
+    /// A client sent structured input events.
+    ClientInputEvents {
+        client_id: u64,
+        events: Vec<crate::protocol::ClientInputEvent>,
+    },
     /// A client sent local clipboard image bytes to paste into a remote pane.
     ClientClipboardImage {
         client_id: u64,
@@ -126,12 +135,30 @@ fn parse_client_keybindings(
     }
 }
 
+fn input_events_within_limits(events: &[ClientInputEvent]) -> bool {
+    if events.len() > MAX_INPUT_EVENT_BATCH {
+        return false;
+    }
+
+    let mut paste_bytes = 0usize;
+    for event in events {
+        if let ClientInputEvent::Paste { text } = event {
+            paste_bytes = paste_bytes.saturating_add(text.len());
+            if paste_bytes > MAX_INPUT_PAYLOAD {
+                return false;
+            }
+        }
+    }
+
+    true
+}
+
 /// Handles the client handshake on a blocking thread.
 ///
 /// Reads the `Hello` message, validates the version, sends `Welcome`,
 /// and then enters a read loop forwarding messages to the server event channel.
 pub(crate) fn handle_client_handshake(
-    mut stream: UnixStream,
+    mut stream: LocalStream,
     client_id: u64,
     server_event_tx: &mpsc::Sender<ServerEvent>,
     should_quit: &Arc<AtomicBool>,
@@ -140,8 +167,9 @@ pub(crate) fn handle_client_handshake(
     // the handshake thread needs blocking I/O for read_message/write_message.
     stream.set_nonblocking(false)?;
 
-    // Set a read timeout for the handshake.
-    stream.set_read_timeout(Some(HANDSHAKE_TIMEOUT))?;
+    if let Err(err) = stream.set_recv_timeout(Some(HANDSHAKE_TIMEOUT)) {
+        debug!(client_id, err = %err, "client handshake read timeout unavailable");
+    }
 
     // Read the Hello message.
     let hello: ClientMessage = match protocol::read_message(&mut stream, MAX_FRAME_SIZE) {
@@ -240,8 +268,9 @@ pub(crate) fn handle_client_handshake(
     };
     protocol::write_message(&mut stream, &welcome).map_err(|e| io::Error::other(e.to_string()))?;
 
-    // Clear read timeout for normal operation.
-    stream.set_read_timeout(None)?;
+    if let Err(err) = stream.set_recv_timeout(None) {
+        debug!(client_id, err = %err, "failed to clear client handshake read timeout");
+    }
 
     // Create separate channels for reliable control messages and droppable renders.
     let (control_tx, control_rx) = std::sync::mpsc::channel::<Vec<u8>>();
@@ -283,7 +312,7 @@ pub(crate) fn handle_client_handshake(
 
 /// The client writer loop — prioritizes control messages over render frames.
 fn client_writer_loop(
-    mut stream: UnixStream,
+    mut stream: LocalStream,
     client_id: u64,
     control_rx: std::sync::mpsc::Receiver<Vec<u8>>,
     render_rx: std::sync::mpsc::Receiver<Vec<u8>>,
@@ -349,7 +378,7 @@ fn client_writer_loop(
     debug!("client writer thread exiting");
 }
 
-fn write_framed_bytes(stream: &mut UnixStream, data: &[u8]) -> bool {
+fn write_framed_bytes(stream: &mut LocalStream, data: &[u8]) -> bool {
     if let Err(err) = stream.write_all(data) {
         debug!(err = %err, "client write failed, closing writer");
         return false;
@@ -363,7 +392,7 @@ fn write_framed_bytes(stream: &mut UnixStream, data: &[u8]) -> bool {
 
 /// The client read loop — reads messages from the client and forwards to the server event channel.
 fn client_read_loop(
-    mut stream: UnixStream,
+    mut stream: LocalStream,
     client_id: u64,
     server_event_tx: &mpsc::Sender<ServerEvent>,
     should_quit: &Arc<AtomicBool>,
@@ -411,6 +440,20 @@ fn client_read_loop(
                     ServerEvent::ClientInput { client_id, data }
                 }
             }
+            ClientMessage::InputEvents { events } => {
+                if !input_events_within_limits(&events) {
+                    warn!(
+                        client_id,
+                        count = events.len(),
+                        "oversized input events from client, closing"
+                    );
+                    let _ = server_event_tx
+                        .blocking_send(ServerEvent::ClientDisconnected { client_id });
+                    break;
+                } else {
+                    ServerEvent::ClientInputEvents { client_id, events }
+                }
+            }
             ClientMessage::ClipboardImage { extension, data } => {
                 if data.len() > MAX_CLIPBOARD_IMAGE_PAYLOAD {
                     warn!(
@@ -487,6 +530,23 @@ fn client_read_loop(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use interprocess::local_socket::traits::Listener as _;
+
+    fn unique_test_path(name: &str) -> std::path::PathBuf {
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        std::env::temp_dir().join(format!("herdr-{name}-{}-{nanos}", std::process::id()))
+    }
+
+    fn local_stream_pair(name: &str) -> (LocalStream, LocalStream, std::path::PathBuf) {
+        let path = unique_test_path(name);
+        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)
+    }
 
     #[test]
     fn clamp_terminal_size_zero_zero() {
@@ -569,7 +629,7 @@ new_tab = "ctrl+notakey"
 
     #[test]
     fn handshake_negotiates_terminal_ansi_encoding() {
-        let (mut client_stream, server_stream) = UnixStream::pair().expect("socket pair");
+        let (mut client_stream, server_stream, _path) = local_stream_pair("client-handshake-ansi");
         let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
         let should_quit = Arc::new(AtomicBool::new(false));
         let handshake_quit = should_quit.clone();
@@ -643,7 +703,8 @@ new_tab = "ctrl+notakey"
 
     #[test]
     fn handshake_marks_terminal_attach_launch_mode() {
-        let (mut client_stream, server_stream) = UnixStream::pair().expect("socket pair");
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-handshake-terminal-attach");
         let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
         let should_quit = Arc::new(AtomicBool::new(false));
         let handshake_quit = should_quit.clone();
@@ -706,7 +767,7 @@ new_tab = "ctrl+notakey"
 
     #[test]
     fn client_read_loop_rejects_oversized_input() {
-        let (mut client_stream, server_stream) = UnixStream::pair().expect("socket pair");
+        let (mut client_stream, server_stream, _path) = local_stream_pair("client-read-oversized");
         let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
         let should_quit = Arc::new(AtomicBool::new(false));
         let read_quit = should_quit.clone();
@@ -739,6 +800,126 @@ new_tab = "ctrl+notakey"
     }
 
     #[test]
+    fn client_read_loop_forwards_input_events() {
+        let (mut client_stream, server_stream, _path) = local_stream_pair("client-read-events");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+        let events = vec![
+            ClientInputEvent::Key {
+                code: crate::protocol::ClientKeyCode::Enter,
+                modifiers: 0,
+                kind: crate::protocol::ClientKeyKind::Press,
+            },
+            ClientInputEvent::FocusGained,
+        ];
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::InputEvents {
+                events: events.clone(),
+            },
+        )
+        .expect("write input events");
+
+        match server_event_rx
+            .blocking_recv()
+            .expect("client input events event")
+        {
+            ServerEvent::ClientInputEvents {
+                client_id,
+                events: actual,
+            } => {
+                assert_eq!(client_id, 7);
+                assert_eq!(actual, events);
+            }
+            other => panic!("expected ClientInputEvents, got {other:?}"),
+        }
+
+        drop(client_stream);
+        should_quit.store(true, Ordering::Release);
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
+    fn client_read_loop_rejects_oversized_input_event_batch() {
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-read-oversized-events");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::InputEvents {
+                events: vec![ClientInputEvent::FocusGained; MAX_INPUT_EVENT_BATCH + 1],
+            },
+        )
+        .expect("write oversized input events");
+
+        match server_event_rx
+            .blocking_recv()
+            .expect("client disconnected event")
+        {
+            ServerEvent::ClientDisconnected { client_id } => assert_eq!(client_id, 7),
+            other => panic!("expected ClientDisconnected, got {other:?}"),
+        }
+
+        drop(client_stream);
+        should_quit.store(true, Ordering::Release);
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
+    fn client_read_loop_rejects_oversized_input_event_paste() {
+        let (mut client_stream, server_stream, _path) =
+            local_stream_pair("client-read-oversized-paste");
+        let (server_event_tx, mut server_event_rx) = mpsc::channel(4);
+        let should_quit = Arc::new(AtomicBool::new(false));
+        let read_quit = should_quit.clone();
+        let handle = std::thread::spawn(move || {
+            client_read_loop(server_stream, 7, &server_event_tx, &read_quit)
+        });
+
+        protocol::write_message(
+            &mut client_stream,
+            &ClientMessage::InputEvents {
+                events: vec![ClientInputEvent::Paste {
+                    text: "x".repeat(MAX_INPUT_PAYLOAD + 1),
+                }],
+            },
+        )
+        .expect("write oversized paste event");
+
+        match server_event_rx
+            .blocking_recv()
+            .expect("client disconnected event")
+        {
+            ServerEvent::ClientDisconnected { client_id } => assert_eq!(client_id, 7),
+            other => panic!("expected ClientDisconnected, got {other:?}"),
+        }
+
+        drop(client_stream);
+        should_quit.store(true, Ordering::Release);
+        handle
+            .join()
+            .expect("read thread join")
+            .expect("read thread result");
+    }
+
+    #[test]
     fn handshake_timeout_is_within_five_second_deadline() {
         // The handshake timeout must be short enough that
         // the connection is guaranteed to close within 5 seconds even with