diff options
Diffstat (limited to 'src/server')
| -rw-r--r-- | src/server/autodetect.rs | 296 | ||||
| -rw-r--r-- | src/server/headless.rs | 2272 | ||||
| -rw-r--r-- | src/server/mod.rs | 3 | ||||
| -rw-r--r-- | src/server/protocol.rs | 1212 |
4 files changed, 3783 insertions, 0 deletions
diff --git a/src/server/autodetect.rs b/src/server/autodetect.rs new file mode 100644 index 00000000..b0ca2a2e --- /dev/null +++ b/src/server/autodetect.rs @@ -0,0 +1,296 @@ +//! Auto-detect launch behavior for the `herdr` command. +//! +//! When the user runs `herdr` with no subcommand: +//! 1. Check if a server is already listening on the client socket +//! 2. If no server → spawn one as a background daemon → wait for socket readiness (up to 5s) +//! 3. Attach as a thin client to the server +//! +//! The `--no-session` flag bypasses server/client entirely and runs monolithically +//! (escape hatch for users who want the traditional single-process behavior). + +use std::io; +use std::os::unix::net::UnixStream; +use std::os::unix::process::CommandExt; +use std::path::Path; +use std::process::Command; +use std::time::Duration; + +use tracing::{info, warn}; + +use super::headless::client_socket_path; + +/// Maximum time to wait for the server's client socket to become ready +/// after spawning the server process. +const SERVER_READY_TIMEOUT: Duration = Duration::from_secs(5); + +/// Poll interval when waiting for the server socket to appear. +const SOCKET_POLL_INTERVAL: Duration = Duration::from_millis(50); + +// --------------------------------------------------------------------------- +// Server detection +// --------------------------------------------------------------------------- + +/// Checks whether a herdr server is currently listening on the client socket. +/// +/// This works by attempting to connect to the client socket. If the connection +/// succeeds, a server is running. If the socket file doesn't exist or the +/// connection is refused, no server is running. Stale sockets (from a crashed +/// server) are detected because `UnixStream::connect` returns `ConnectionRefused` +/// when nobody is listening. +#[allow(dead_code)] // Public API for external use and testing +pub fn is_server_listening() -> bool { + is_server_listening_at(&client_socket_path()) +} + +/// Checks whether a herdr server is listening at a specific socket path. +fn is_server_listening_at(socket_path: &Path) -> bool { + if !socket_path.exists() { + return false; + } + + match UnixStream::connect(socket_path) { + Ok(_) => { + // Server is listening. Close the test connection immediately. + // The server's handshake handler will time out on this connection + // since we don't send Hello, which is fine. + true + } + Err(err) + if matches!( + err.kind(), + io::ErrorKind::ConnectionRefused | io::ErrorKind::TimedOut + ) => + { + // Socket file exists but nobody is listening — stale socket. + false + } + Err(err) if err.kind() == io::ErrorKind::NotFound => { + // Socket file disappeared between exists() and connect(). + false + } + Err(err) => { + // Other errors (permission denied, etc.) — assume not listening. + warn!(err = %err, "unexpected error checking server socket"); + false + } + } +} + +// --------------------------------------------------------------------------- +// Server spawning +// --------------------------------------------------------------------------- + +/// Spawns the herdr server as a background daemon process. +/// +/// The server process is fully detached: +/// - Runs in its own session (setsid) so it survives the client exiting +/// - Stdin/stdout/stderr are redirected to /dev/null +/// - Inherits all relevant environment variables +/// (`HERDR_SOCKET_PATH`, `HERDR_CLIENT_SOCKET_PATH`, `XDG_CONFIG_HOME`, etc.) +/// +/// Returns the PID of the spawned server process. +pub fn spawn_server_daemon() -> io::Result<u32> { + let exe = std::env::current_exe().map_err(|err| { + io::Error::new( + err.kind(), + format!("failed to determine herdr executable path: {err}"), + ) + })?; + + info!(exe = %exe.display(), "spawning server daemon"); + + let child = Command::new(&exe) + .arg("server") + // Create a new process group so the server survives the parent's exit + // and doesn't receive SIGHUP when the client's terminal closes. + .process_group(0) + // Redirect stdio to /dev/null + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .map_err(|err: io::Error| { + io::Error::new(err.kind(), format!("failed to spawn herdr server: {err}")) + })?; + + let pid = child.id(); + info!(pid, "server daemon spawned"); + + Ok(pid) +} + +// --------------------------------------------------------------------------- +// Socket readiness +// --------------------------------------------------------------------------- + +/// Waits for the server's client socket to become ready for connections. +/// +/// Polls the socket path at regular intervals until a connection succeeds +/// or the timeout elapses. Returns an error if the server doesn't become +/// ready within the timeout. +pub fn wait_for_server_socket(socket_path: &Path, timeout: Duration) -> io::Result<()> { + let deadline = std::time::Instant::now() + timeout; + + while std::time::Instant::now() < deadline { + if is_server_listening_at(socket_path) { + info!(path = %socket_path.display(), "server socket ready"); + return Ok(()); + } + std::thread::sleep(SOCKET_POLL_INTERVAL); + } + + Err(io::Error::new( + io::ErrorKind::TimedOut, + format!( + "server did not become ready within {}s (socket: {})", + timeout.as_secs(), + socket_path.display() + ), + )) +} + +// --------------------------------------------------------------------------- +// Auto-detect launch +// --------------------------------------------------------------------------- + +/// Performs auto-detect launch: check for server, spawn if needed, then +/// attach as a thin client. +/// +/// This is the entry point called from `main.rs` when the user runs `herdr` +/// without `--no-session` and without a subcommand. +/// +/// Flow: +/// 1. Check if a server is listening on the client socket +/// 2. If no server → spawn server daemon → wait for socket readiness +/// 3. Run the thin client (which connects to the server) +pub fn auto_detect_launch() -> io::Result<()> { + let socket_path = client_socket_path(); + info!(path = %socket_path.display(), "auto-detect launch starting"); + + if is_server_listening_at(&socket_path) { + info!("server already running, attaching as client"); + } else { + info!("no server running, spawning server daemon"); + spawn_server_daemon()?; + wait_for_server_socket(&socket_path, SERVER_READY_TIMEOUT)?; + info!("server ready, attaching as client"); + } + + // Now attach as a thin client. + crate::client::run_client() +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::net::UnixListener; + + fn temp_socket_path(name: &str) -> std::path::PathBuf { + let unique = format!( + "herdr-autodetect-test-{}-{}-{}", + name, + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + std::env::temp_dir().join(unique).join("test.sock") + } + + #[test] + fn is_server_listening_returns_false_for_nonexistent_path() { + let path = temp_socket_path("nonexistent"); + assert!(!is_server_listening_at(&path)); + } + + #[test] + fn is_server_listening_returns_true_for_live_socket() { + let dir = temp_socket_path("live"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("test.sock"); + + let _listener = UnixListener::bind(&path).unwrap(); + assert!(is_server_listening_at(&path)); + } + + #[test] + fn is_server_listening_returns_false_for_stale_socket() { + let dir = temp_socket_path("stale"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("test.sock"); + + // Create a socket and immediately drop the listener. + // This leaves a stale socket file with nobody listening. + { + let _listener = UnixListener::bind(&path).unwrap(); + } + + // The socket file exists but nobody is listening. + assert!(!is_server_listening_at(&path)); + } + + #[test] + fn is_server_listening_returns_false_when_listener_dropped() { + let dir = temp_socket_path("dropped"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("test.sock"); + + // Bind and immediately drop the listener. + drop(UnixListener::bind(&path).unwrap()); + + // Socket is stale — should return false. + assert!(!is_server_listening_at(&path)); + + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn wait_for_server_socket_succeeds_immediately() { + let dir = temp_socket_path("wait-ok"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("test.sock"); + + let _listener = UnixListener::bind(&path).unwrap(); + + // Should succeed immediately (socket is already ready). + let result = wait_for_server_socket(&path, Duration::from_millis(100)); + assert!(result.is_ok()); + } + + #[test] + fn wait_for_server_socket_times_out() { + let dir = temp_socket_path("wait-timeout"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("test.sock"); + + // No listener — should time out. + let result = wait_for_server_socket(&path, Duration::from_millis(50)); + assert!(result.is_err()); + assert_eq!(result.unwrap_err().kind(), io::ErrorKind::TimedOut); + } + + #[test] + fn wait_for_server_socket_succeeds_after_delay() { + let dir = temp_socket_path("wait-delay"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("test.sock"); + + // Spawn a thread that will create the listener after a short delay. + let path_clone = path.clone(); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(50)); + let _listener = UnixListener::bind(&path_clone).unwrap(); + // Keep the listener alive for a bit. + std::thread::sleep(Duration::from_secs(1)); + }); + + // Wait with a generous timeout — should succeed. + let result = wait_for_server_socket(&path, Duration::from_secs(2)); + assert!(result.is_ok()); + } +} diff --git a/src/server/headless.rs b/src/server/headless.rs new file mode 100644 index 00000000..ba6098df --- /dev/null +++ b/src/server/headless.rs @@ -0,0 +1,2272 @@ +//! Headless server mode — runs the herdr event loop without a real terminal. +//! +//! The server: +//! - Does not enter raw mode or read stdin +//! - Creates and listens on both `herdr.sock` (existing JSON API) and +//! `herdr-client.sock` (new binary protocol) +//! - Initializes AppState and all PTYs from session restore or fresh state +//! - Runs the main event loop (drain events, drain API requests, scheduled tasks) +//! - Renders to a virtual ratatui Buffer in memory +//! - Accepts client connections on the client socket +//! - Streams frames to connected clients after each render +//! - Routes client input events through the existing input pipeline +//! - Continues running after client disconnect +//! - Handles stale socket cleanup, explicit server stop, minimum terminal size, +//! and pane spawn failure during restore + +use std::collections::HashMap; +use std::fs; +use std::io::{self, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use ratatui::backend::{Backend, ClearType, TestBackend, WindowSize}; +use ratatui::layout::{Position, Rect, Size}; +use tokio::sync::mpsc; +use tracing::{debug, error, info, warn}; + +use base64::Engine; + +use crate::api; +use crate::app; +use crate::app::state::AppState; +use crate::config; +use crate::events::AppEvent; +use crate::server::protocol::{ + self, ClientMessage, CursorState, FrameData, ServerMessage, MAX_FRAME_SIZE, PROTOCOL_VERSION, +}; + +// --------------------------------------------------------------------------- +// Loop event enum for the headless server event loop +// --------------------------------------------------------------------------- + +/// Events that the headless server event loop can process. +enum LoopEvent { + Timer, + Internal(AppEvent), + Api(api::ApiRequestMessage), + ServerEvent(ServerEvent), + RenderRequested, +} + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +/// Default shared runtime size (columns, rows) when no clients are attached. +const MIN_COLS: u16 = 80; +const MIN_ROWS: u16 = 24; + +/// Minimum accepted attached client size. +/// +/// Narrow observers must be allowed to drive narrow renders, otherwise the +/// server wraps pane content against a wider width and the client sees the +/// right edge clipped. +const MIN_CLIENT_COLS: u16 = 1; +const MIN_CLIENT_ROWS: u16 = 1; + +/// Legacy environment variable for overriding the client socket path. +/// +/// Contractual override behavior for auto-detect uses `HERDR_SOCKET_PATH`. +/// This variable is kept as a fallback for callers that explicitly need a +/// client-only override when `HERDR_SOCKET_PATH` is not set. +const CLIENT_SOCKET_PATH_ENV_VAR: &str = "HERDR_CLIENT_SOCKET_PATH"; + +/// Socket permission mode (owner read/write only). +const SOCKET_PERMISSION_MODE: u32 = 0o600; + +/// Timeout for in-flight API requests during shutdown. +#[allow(dead_code)] +const SHUTDOWN_API_TIMEOUT: Duration = Duration::from_secs(5); + +/// How long to wait for a client handshake before closing the connection. +/// Set to 4 seconds (rather than 5) to guarantee the connection is closed +/// within the 5-second deadline, even with +/// OS timer slack, thread scheduling, and cleanup overhead. +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 + +// --------------------------------------------------------------------------- +// Socket path helpers +// --------------------------------------------------------------------------- + +/// Returns the path for the client protocol socket. +/// +/// Contract-aligned override behavior: +/// 1. If `HERDR_SOCKET_PATH` is set, derive the client socket path from it by +/// inserting `-client` before `.sock` (e.g. `herdr.sock` -> `herdr-client.sock`). +/// This keeps JSON API and client socket overrides consistent. +/// 2. Otherwise, honor `HERDR_CLIENT_SOCKET_PATH` (legacy/testing fallback). +/// 3. Otherwise, use the app config directory. +pub fn client_socket_path() -> PathBuf { + client_socket_path_from_overrides( + std::env::var(api::SOCKET_PATH_ENV_VAR).ok().as_deref(), + std::env::var(CLIENT_SOCKET_PATH_ENV_VAR).ok().as_deref(), + ) +} + +fn client_socket_path_from_overrides( + api_socket_override: Option<&str>, + client_socket_override: Option<&str>, +) -> PathBuf { + if let Some(api_socket_override) = api_socket_override { + return derive_client_socket_from_api_socket(Path::new(api_socket_override)); + } + + if let Some(client_socket_override) = client_socket_override { + return PathBuf::from(client_socket_override); + } + + config::config_dir().join("herdr-client.sock") +} + +fn derive_client_socket_from_api_socket(api_socket_path: &Path) -> PathBuf { + let stem = api_socket_path + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or("herdr"); + let parent = api_socket_path.parent().unwrap_or_else(|| Path::new("")); + + if api_socket_path + .extension() + .and_then(|ext| ext.to_str()) + .is_some_and(|ext| ext == "sock") + { + return parent.join(format!("{stem}-client.sock")); + } + + parent.join(format!("{stem}-client.sock")) +} + +/// Clamp client-reported terminal dimensions to a minimum viable size. +pub fn clamp_terminal_size(cols: u16, rows: u16) -> (u16, u16) { + let clamped_cols = cols.max(MIN_CLIENT_COLS); + let clamped_rows = rows.max(MIN_CLIENT_ROWS); + (clamped_cols, clamped_rows) +} + +// --------------------------------------------------------------------------- +// Connected client state +// --------------------------------------------------------------------------- + +/// A connected client tracked by the server. +struct ClientConnection { + /// The client's terminal size (after clamping). + terminal_size: (u16, u16), + /// Last known host terminal default colors for this client. + host_terminal_theme: crate::terminal_theme::TerminalTheme, + /// Monotonic activity stamp used to choose the fallback foreground client. + last_activity: u64, + /// Channel for sending framed ServerMessage data to the client writer thread. + writer: Option<std::sync::mpsc::Sender<Vec<u8>>>, +} + +// --------------------------------------------------------------------------- +// Stale socket cleanup +// --------------------------------------------------------------------------- + +/// Prepares a socket path for binding: creates parent directories, +/// removes stale socket files (where no server is listening), and +/// returns an error if a live server is already bound. +fn prepare_socket_path(path: &Path) -> io::Result<()> { + crate::ipc::prepare_socket_path(path, |path| { + format!( + "herdr server is already running (socket busy at {})", + path.display() + ) + }) +} + +/// Restricts socket file permissions to owner-only (0o600). +fn restrict_socket_permissions(path: &Path) -> io::Result<()> { + crate::ipc::restrict_socket_permissions(path, SOCKET_PERMISSION_MODE) +} + +// --------------------------------------------------------------------------- +// Virtual rendering +// --------------------------------------------------------------------------- + +struct CursorTrackingBackend { + inner: TestBackend, + rendered_cursor: Option<Position>, +} + +impl CursorTrackingBackend { + fn new(width: u16, height: u16) -> Self { + Self { + inner: TestBackend::new(width, height), + rendered_cursor: None, + } + } + + fn buffer(&self) -> &ratatui::buffer::Buffer { + self.inner.buffer() + } + + fn rendered_cursor(&self) -> Option<CursorState> { + self.rendered_cursor.map(|pos| CursorState { + x: pos.x, + y: pos.y, + visible: true, + }) + } +} + +impl Backend for CursorTrackingBackend { + type Error = std::convert::Infallible; + + fn draw<'a, I>(&mut self, content: I) -> Result<(), Self::Error> + where + I: Iterator<Item = (u16, u16, &'a ratatui::buffer::Cell)>, + { + self.inner.draw(content) + } + + fn append_lines(&mut self, n: u16) -> Result<(), Self::Error> { + self.inner.append_lines(n) + } + + fn hide_cursor(&mut self) -> Result<(), Self::Error> { + self.inner.hide_cursor()?; + self.rendered_cursor = None; + Ok(()) + } + + fn show_cursor(&mut self) -> Result<(), Self::Error> { + self.inner.show_cursor() + } + + fn get_cursor_position(&mut self) -> Result<Position, Self::Error> { + self.inner.get_cursor_position() + } + + fn set_cursor_position<P: Into<Position>>(&mut self, position: P) -> Result<(), Self::Error> { + let position = position.into(); + self.inner.set_cursor_position(position)?; + self.rendered_cursor = Some(position); + Ok(()) + } + + fn clear(&mut self) -> Result<(), Self::Error> { + self.inner.clear() + } + + fn clear_region(&mut self, clear_type: ClearType) -> Result<(), Self::Error> { + self.inner.clear_region(clear_type) + } + + fn size(&self) -> Result<Size, Self::Error> { + self.inner.size() + } + + fn window_size(&mut self) -> Result<WindowSize, Self::Error> { + self.inner.window_size() + } + + fn flush(&mut self) -> Result<(), Self::Error> { + self.inner.flush() + } +} + +/// Renders the AppState to an in-memory ratatui Buffer. +/// +/// This produces the same output as the monolithic binary's terminal draw, +/// but writes to a `Buffer` instead of stdout. Cursor visibility is captured +/// from explicit frame cursor intent rather than incidental backend state. +fn render_virtual( + app_state: &mut AppState, + area: Rect, + resize_panes: bool, +) -> (ratatui::buffer::Buffer, Option<CursorState>) { + if resize_panes { + crate::ui::compute_view(app_state, area); + } else { + crate::ui::compute_view_without_resizing_panes(app_state, area); + } + + let backend = CursorTrackingBackend::new(area.width, area.height); + let mut terminal = ratatui::Terminal::new(backend).expect("TestBackend::new should never fail"); + + terminal + .draw(|frame| { + crate::ui::render(app_state, frame); + }) + .expect("render to TestBackend should never fail"); + + let buffer = terminal.backend().buffer().clone(); + let cursor = terminal.backend().rendered_cursor(); + + (buffer, cursor) +} + +// --------------------------------------------------------------------------- +// Headless server +// --------------------------------------------------------------------------- + +/// The headless server — runs the herdr event loop without a real terminal. +pub struct HeadlessServer { + app: app::App, + client_listener: UnixListener, + client_socket_path: PathBuf, + clients: HashMap<u64, ClientConnection>, + next_client_id: u64, + /// The client currently driving the shared pane runtime size and theme. + foreground_client_id: Option<u64>, + /// Monotonic activity counter used to pick the most recently active client. + next_activity_stamp: u64, + /// Shared pane runtime size derived from the foreground client, + /// or MIN_COLS × MIN_ROWS when no clients are connected. + effective_size: (u16, u16), + /// Flag set when shutdown is initiated. + shutting_down: bool, + /// Flag set by Ctrl+C or `server stop` signal. + should_quit: Arc<AtomicBool>, + /// Channel for receiving server events from client connection threads. + server_event_rx: mpsc::Receiver<ServerEvent>, + /// Sender for server events (cloned for each client thread). + server_event_tx: mpsc::Sender<ServerEvent>, +} + +impl HeadlessServer { + /// Creates and starts the headless server. + /// + /// This: + /// 1. Prepares the client socket path (cleans up stale sockets) + /// 2. Binds the client socket listener + /// 3. Returns the server ready to run + pub fn new(app: app::App) -> io::Result<Self> { + let client_path = client_socket_path(); + prepare_socket_path(&client_path)?; + + let listener = UnixListener::bind(&client_path)?; + restrict_socket_permissions(&client_path)?; + info!(path = %client_path.display(), "client protocol socket listening"); + + // Set non-blocking on the listener so we can poll it from the event loop. + listener.set_nonblocking(true)?; + + let should_quit = Arc::new(AtomicBool::new(false)); + + // Channel for server events from client threads. + let (server_event_tx, server_event_rx) = mpsc::channel(64); + + Ok(Self { + app, + client_listener: listener, + client_socket_path: client_path, + clients: HashMap::new(), + next_client_id: 1, + foreground_client_id: None, + next_activity_stamp: 1, + effective_size: (MIN_COLS, MIN_ROWS), + shutting_down: false, + should_quit, + server_event_rx, + server_event_tx, + }) + } + + /// Runs the headless server event loop until shutdown. + /// + /// This is the server's main loop — analogous to `App::run()` but without + /// a real terminal. It: + /// - Drains internal events (pane death, state changes) + /// - Drains API requests (from the JSON socket) + /// - Accepts new client connections + /// - Reads client messages and routes input + /// - Handles scheduled tasks (resize poll, animation, session save, etc.) + /// - Renders virtually and streams frames to clients + pub async fn run(&mut self) -> io::Result<()> { + info!("headless server starting"); + + // Register SIGINT handler for graceful shutdown. + let should_quit = self.should_quit.clone(); + let quit_notify = self.server_event_tx.clone(); + ctrlc_handler(should_quit, quit_notify); + + // No input_rx needed — server doesn't read stdin. + // We use None for input_rx so the event loop doesn't try to read from stdin. + self.app.input_rx = None; + + let mut needs_render = true; + + loop { + // If shutdown has been initiated, complete it and exit. + if self.shutting_down { + self.complete_shutdown()?; + break; + } + + // Check if we should start shutting down. + if self.app.state.should_quit || self.should_quit.load(Ordering::Acquire) { + self.initiate_shutdown(); + continue; + } + + // 1. Check render_dirty flag from PTY reader tasks. + if self.app.render_dirty.load(Ordering::Acquire) { + needs_render = true; + } + + // 2. Drain internal events. + if self.drain_internal_events_with_forwarding() { + needs_render = true; + } + + // 3. Drain API requests. + if self.drain_api_requests_with_shutdown_check() { + needs_render = true; + } + + self.app.sync_focus_events(); + self.app.sync_session_save_schedule(); + + // 4. Accept new client connections. + self.accept_client_connections()?; + + // 5. Drain server events from client threads. + if self.drain_server_events() { + needs_render = true; + } + + // 6. Handle scheduled tasks. + let now = Instant::now(); + if self.handle_scheduled_tasks_headless(now) { + needs_render = true; + } + + // Handle deferred requests. + if self.app.state.request_complete_onboarding { + self.app.state.request_complete_onboarding = false; + self.app.complete_onboarding(); + needs_render = true; + } + + if self.app.state.request_new_workspace { + self.app.state.request_new_workspace = false; + self.app.create_workspace(); + needs_render = true; + } + + if self.app.state.request_new_tab { + self.app.state.request_new_tab = false; + self.app.create_tab(); + needs_render = true; + } + + if self.app.state.request_reload_keybinds { + self.app.state.request_reload_keybinds = false; + self.app.reload_keybinds(); + needs_render = true; + } + + self.app.sync_animation_timer(now); + + // 7. Render virtually and stream frames. + if needs_render && self.app.can_render_now(now) { + self.app.render_dirty.swap(false, Ordering::AcqRel); + self.render_and_stream(); + self.app.last_render_at = Some(now); + needs_render = false; + continue; + } + + // 8. Wait for next event. + let next_deadline = self.app.next_loop_deadline(now, needs_render); + let event = { + tokio::select! { + maybe_api = self.app.api_rx.recv() => match maybe_api { + Some(msg) => LoopEvent::Api(msg), + None => LoopEvent::Timer, + }, + maybe_ev = self.app.event_rx.recv() => match maybe_ev { + Some(ev) => LoopEvent::Internal(ev), + None => LoopEvent::Timer, + }, + maybe_server_ev = self.server_event_rx.recv() => match maybe_server_ev { + Some(ev) => LoopEvent::ServerEvent(ev), + None => LoopEvent::Timer, + }, + _ = sleep_until_or_pending(next_deadline) => LoopEvent::Timer, + _ = self.app.render_notify.notified() => LoopEvent::RenderRequested, + } + }; + + match event { + LoopEvent::Timer => {} + LoopEvent::Internal(ev) => { + if self.handle_internal_event_with_forwarding(ev) { + needs_render = true; + } + } + LoopEvent::Api(msg) => { + if self.handle_api_request_with_shutdown_check(msg) { + needs_render = true; + } + } + LoopEvent::ServerEvent(ev) => { + if self.handle_server_event(ev) { + needs_render = true; + } + } + LoopEvent::RenderRequested => { + if self.app.render_dirty.load(Ordering::Acquire) { + needs_render = true; + } + } + } + } + + // Save session on exit. + if !self.app.no_session { + self.app.save_session_now(); + } + + info!("headless server exiting"); + Ok(()) + } + + fn allocate_activity_stamp(&mut self) -> u64 { + let stamp = self.next_activity_stamp; + self.next_activity_stamp = self.next_activity_stamp.saturating_add(1); + stamp + } + + fn resize_shared_runtime_to_effective_size(&mut self) { + if self.foreground_client_id.is_none() { + return; + } + let (cols, rows) = self.effective_size; + let area = Rect::new(0, 0, cols, rows); + crate::ui::compute_view(&mut self.app.state, area); + } + + fn sync_foreground_client_state(&mut self) { + let Some(client_id) = self.foreground_client_id else { + self.effective_size = (MIN_COLS, MIN_ROWS); + return; + }; + let Some(client) = self.clients.get(&client_id) else { + self.foreground_client_id = None; + self.effective_size = (MIN_COLS, MIN_ROWS); + return; + }; + + self.effective_size = client.terminal_size; + if !client.host_terminal_theme.is_empty() { + self.app.set_host_terminal_theme(client.host_terminal_theme); + } + } + + fn promote_client_to_foreground(&mut self, client_id: u64) -> bool { + let stamp = self.allocate_activity_stamp(); + let Some(client) = self.clients.get_mut(&client_id) else { + return false; + }; + client.last_activity = stamp; + + let changed = self.foreground_client_id != Some(client_id); + self.foreground_client_id = Some(client_id); + self.sync_foreground_client_state(); + changed + } + + fn promote_latest_remaining_client(&mut self) -> bool { + let next_foreground = self + .clients + .iter() + .max_by_key(|(_, client)| client.last_activity) + .map(|(&client_id, _)| client_id); + let changed = next_foreground != self.foreground_client_id; + self.foreground_client_id = next_foreground; + self.sync_foreground_client_state(); + changed + } + + fn remove_client(&mut self, client_id: u64) -> bool { + let was_foreground = self.foreground_client_id == Some(client_id); + self.clients.remove(&client_id); + if was_foreground { + self.promote_latest_remaining_client() + } else { + false + } + } + + fn update_client_host_theme_from_events( + &mut self, + client_id: u64, + events: &[crate::raw_input::RawInputEvent], + ) -> bool { + let Some(client) = self.clients.get_mut(&client_id) else { + return false; + }; + + let mut next_theme = client.host_terminal_theme; + for event in events { + if let crate::raw_input::RawInputEvent::HostDefaultColor { kind, color } = event { + next_theme = next_theme.with_color(*kind, *color); + } + } + + if next_theme == client.host_terminal_theme { + return false; + } + + client.host_terminal_theme = next_theme; + if self.foreground_client_id == Some(client_id) { + self.app.set_host_terminal_theme(next_theme) + } else { + false + } + } + + fn events_include_interaction(events: &[crate::raw_input::RawInputEvent]) -> bool { + events.iter().any(|event| { + matches!( + event, + crate::raw_input::RawInputEvent::Key(_) + | crate::raw_input::RawInputEvent::Mouse(_) + | crate::raw_input::RawInputEvent::Paste(_) + ) + }) + } + + /// Accepts pending client connections from the non-blocking listener. + fn accept_client_connections(&mut self) -> io::Result<()> { + loop { + match self.client_listener.accept() { + Ok((stream, _addr)) => { + let client_id = self.next_client_id; + self.next_client_id += 1; + + if let Err(err) = stream.set_nonblocking(true) { + warn!(err = %err, "failed to set client stream nonblocking"); + continue; + } + + // Spawn a thread for the handshake and read loop. + let should_quit = self.should_quit.clone(); + let server_event_tx = self.server_event_tx.clone(); + std::thread::spawn(move || { + if let Err(err) = handle_client_handshake( + stream, + client_id, + &server_event_tx, + &should_quit, + ) { + debug!(client_id, err = %err, "client handshake failed"); + } + }); + } + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => { + // No more pending connections. + break; + } + Err(err) => { + error!(err = %err, "client listener accept failed"); + break; + } + } + } + Ok(()) + } + + /// Drains server events from the dedicated channel. + /// + /// Returns true if any input was processed (requiring a re-render). + fn drain_server_events(&mut self) -> bool { + let mut changed = false; + while let Ok(ev) = self.server_event_rx.try_recv() { + changed |= self.handle_server_event(ev); + } + changed + } + + /// Handles a single internal event with forwarding logic for clipboard, + /// sound, and toast notifications to connected clients. + /// + /// ALL internal events MUST be routed through this method to ensure + /// clipboard/notify forwarding is never bypassed. Do not call + /// `self.app.handle_internal_event()` directly for any internal event + /// in the headless server — use this method instead. + /// + /// Returns true if the event changed visual state (requiring a re-render). + fn handle_internal_event_with_forwarding(&mut self, ev: AppEvent) -> bool { + match &ev { + AppEvent::ClipboardWrite { content } => { + // Clipboard writes are client-local side effects. Forward them only to + // the foreground client instead of broadcasting to every attached client. + if let Some(client_id) = self.foreground_client_id { + let data = base64::engine::general_purpose::STANDARD.encode(content.as_slice()); + self.send_to_client(client_id, ServerMessage::Clipboard { data }); + } + // ClipboardWrite doesn't change visual state — no render needed. + false + } + AppEvent::StateChanged { + pane_id, + agent, + state, + } => { + // Capture toast before handling. + let toast_before = self.app.state.toast.clone(); + let pane_id_val = *pane_id; + let agent_val = *agent; + let state_val = *state; + + // Find the previous state of this pane before the event + // is processed. We need this to determine if a sound + // notification would be triggered. + let prev_state = self + .app + .state + .workspaces + .iter() + .find_map(|ws| { + ws.tabs + .iter() + .find_map(|tab| tab.panes.get(&pane_id_val).map(|p| p.state)) + }) + .unwrap_or(crate::detect::AgentState::Unknown); + + // Handle the state change (updates pane state, sets toast on AppState). + // Note: apply_pane_state_change inside handle_internal_event will try + // to play sounds, but sound.enabled=false in the headless server, so + // sound::play is never called. Toast may still be set on AppState if + // toast_config.enabled is true. + self.app.handle_internal_event(ev); + + // Forward sound notification to clients. + // We check the agent-specific sound setting but NOT sound.enabled, + // because the server sets enabled=false to prevent local playback — + // clients should still receive sound notifications and decide + // locally whether to play them based on their own config. + let is_active_tab = self + .app + .state + .active + .and_then(|ws_idx| self.app.state.workspaces.get(ws_idx)) + .is_some_and(|ws| { + ws.find_tab_index_for_pane(pane_id_val) + .is_some_and(|tab_idx| ws.active_tab_index() == tab_idx) + }); + + if !matches!( + self.app.state.sound.agents.for_agent(agent_val), + crate::config::AgentSoundSetting::Off + ) { + if let Some(sound) = crate::app::actions::notification_sound_for_state_change( + is_active_tab, + prev_state, + state_val, + ) { + let msg = match sound { + crate::sound::Sound::Done => "agent done", + crate::sound::Sound::Request => "agent attention", + }; + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Sound, + message: msg.to_owned(), + }); + } + } + + // Forward any new toast as a notification. + if self.app.state.toast.is_some() && self.app.state.toast != toast_before { + if let Some(toast) = &self.app.state.toast { + let msg = format!("{}: {}", toast.title, toast.context); + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Toast, + message: msg, + }); + } + } + + true + } + AppEvent::HookStateReported { + pane_id, + agent_label, + state, + .. + } => { + // The hook authority may not change the effective pane state if + // the detected agent doesn't match. We forward based on the + // hook-reported state transition regardless. + let toast_before = self.app.state.toast.clone(); + let pane_id_val = *pane_id; + let agent_val = crate::detect::parse_agent_label(agent_label); + let hook_state_val = *state; + + // Capture the previous hook authority state for this pane. + // If no hook authority exists, use the effective state. + let prev_hook_state = self + .app + .state + .workspaces + .iter() + .find_map(|ws| { + ws.tabs.iter().find_map(|tab| { + tab.panes.get(&pane_id_val).map(|p| { + p.hook_authority + .as_ref() + .map(|h| h.state) + .unwrap_or(p.state) + }) + }) + }) + .unwrap_or(crate::detect::AgentState::Unknown); + + self.app.handle_internal_event(ev); + + // Forward sound notification based on hook state transition. + // This ensures API-reported state changes (pane.report_agent) + // produce notifications even before fallback detection confirms. + let is_active_tab = self + .app + .state + .active + .and_then(|ws_idx| self.app.state.workspaces.get(ws_idx)) + .is_some_and(|ws| { + ws.find_tab_index_for_pane(pane_id_val) + .is_some_and(|tab_idx| ws.active_tab_index() == tab_idx) + }); + + if !matches!( + self.app.state.sound.agents.for_agent(agent_val), + crate::config::AgentSoundSetting::Off + ) { + if let Some(sound) = crate::app::actions::notification_sound_for_state_change( + is_active_tab, + prev_hook_state, + hook_state_val, + ) { + let msg = match sound { + crate::sound::Sound::Done => "agent done", + crate::sound::Sound::Request => "agent attention", + }; + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Sound, + message: msg.to_owned(), + }); + } + } + + // Forward any new toast as a notification. + if self.app.state.toast.is_some() && self.app.state.toast != toast_before { + if let Some(toast) = &self.app.state.toast { + let msg = format!("{}: {}", toast.title, toast.context); + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Toast, + message: msg, + }); + } + } + + true + } + AppEvent::UpdateReady { version: _ } => { + let toast_before = self.app.state.toast.clone(); + + self.app.handle_internal_event(ev); + + // Forward the update toast notification. + if self.app.state.toast.is_some() && self.app.state.toast != toast_before { + if let Some(toast) = &self.app.state.toast { + let msg = format!("{}: {}", toast.title, toast.context); + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Toast, + message: msg, + }); + } + } + + true + } + _ => { + self.app.handle_internal_event(ev); + true + } + } + } + + /// Drains internal events, forwarding clipboard, sound, and toast + /// notifications to connected clients instead of processing them locally. + /// + /// In the monolithic mode: + /// - `ClipboardWrite` events are written to stdout via `write_osc52_bytes`. + /// - Sound notifications are played locally via `sound::play`. + /// - Toast notifications are set on AppState and rendered into the frame. + /// + /// In the headless server, there is no stdout terminal or audio subsystem, + /// so we: + /// - Forward `ClipboardWrite` as `ServerMessage::Clipboard` to the + /// foreground client only. + /// - Detect when a sound would be played and forward as + /// `ServerMessage::Notify { kind: Sound }` to all clients. + /// - Detect when a toast is set on AppState and forward as + /// `ServerMessage::Notify { kind: Toast }` to all clients. + fn drain_internal_events_with_forwarding(&mut self) -> bool { + let mut changed = false; + while let Ok(ev) = self.app.event_rx.try_recv() { + changed |= self.handle_internal_event_with_forwarding(ev); + } + changed + } + + /// Encodes a server message into a length-prefixed frame. + fn frame_server_message(msg: &ServerMessage) -> Result<Vec<u8>, protocol::FramingError> { + let mut framed = Vec::new(); + protocol::write_message(&mut framed, msg)?; + Ok(framed) + } + + /// Sends a message to all connected clients. + /// Broken connections are tracked and cleaned up. + fn send_to_all_clients(&mut self, msg: ServerMessage) { + let serialized = match Self::frame_server_message(&msg) { + Ok(framed) => framed, + Err(err) => { + warn!(err = %err, "failed to serialize message for clients"); + return; + } + }; + + let mut broken_clients: Vec<u64> = Vec::new(); + for (&client_id, client) in &mut self.clients { + if let Some(writer) = &client.writer { + if writer.send(serialized.clone()).is_err() { + debug!(client_id, "client writer channel closed during broadcast"); + broken_clients.push(client_id); + } + } + } + + // Remove broken clients. + for client_id in broken_clients { + let foreground_changed = self.remove_client(client_id); + if foreground_changed { + self.resize_shared_runtime_to_effective_size(); + } + } + } + + /// Sends a message to a specific client. Returns false if the client + /// was not found or the send failed (client removed). + fn send_to_client(&mut self, client_id: u64, msg: ServerMessage) -> bool { + let serialized = match Self::frame_server_message(&msg) { + Ok(framed) => framed, + Err(err) => { + warn!(client_id, err = %err, "failed to serialize message for client"); + return false; + } + }; + + if let Some(client) = self.clients.get(&client_id) { + if let Some(writer) = &client.writer { + if writer.send(serialized).is_err() { + debug!( + client_id, + "client writer channel closed during targeted send" + ); + let foreground_changed = self.remove_client(client_id); + if foreground_changed { + self.resize_shared_runtime_to_effective_size(); + } + return false; + } + } + true + } else { + false + } + } + + /// Handles a server event. Returns true if the event requires a re-render. + fn handle_server_event(&mut self, ev: ServerEvent) -> bool { + match ev { + ServerEvent::ClientConnected { + client_id, + cols, + rows, + writer, + } => { + info!(client_id, cols, rows, "client connected"); + let last_activity = self.allocate_activity_stamp(); + self.clients.insert( + client_id, + ClientConnection { + terminal_size: (cols, rows), + host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), + last_activity, + writer: Some(writer), + }, + ); + self.foreground_client_id = Some(client_id); + self.sync_foreground_client_state(); + self.resize_shared_runtime_to_effective_size(); + true + } + ServerEvent::ClientInput { client_id, data } => { + debug!(client_id, len = data.len(), "client input received"); + let events = crate::raw_input::parse_raw_input_bytes_sync(&data); + let interaction = Self::events_include_interaction(&events); + let foreground_changed = if interaction { + self.promote_client_to_foreground(client_id) + } else { + false + }; + if foreground_changed { + self.resize_shared_runtime_to_effective_size(); + } + let theme_changed = self.update_client_host_theme_from_events(client_id, &events); + self.app + .route_client_events(events, self.foreground_client_id == Some(client_id)); + + // Check if the detach keybind was triggered during input processing. + if self.app.state.detach_requested { + self.app.state.detach_requested = false; + info!(client_id, "client detach requested via keybind"); + + // Send a ServerShutdown with "detached" reason to this client + // so it exits cleanly (not with a connection-lost error). + // The client will close its connection after receiving this, + // which triggers a ClientDisconnected event that removes it. + self.send_to_client( + client_id, + ServerMessage::ServerShutdown { + reason: Some("detached".to_owned()), + }, + ); + + // Don't remove the client here — let the client disconnect + // naturally after receiving the ServerShutdown. The client's + // read loop will see EOF and the server will get a + // ClientDisconnected event which handles cleanup. + // + // However, we do need to stop sending frames to this client + // since it's detaching. Drop the writer channel so no more + // frames are queued for this client. + if let Some(client) = self.clients.get_mut(&client_id) { + client.writer = None; + } + + // No re-render needed for remaining clients. + false + } else { + foreground_changed || theme_changed || interaction + } + } + ServerEvent::ClientResize { + client_id, + cols, + rows, + } => { + info!(client_id, cols, rows, "client resize"); + if let Some(client) = self.clients.get_mut(&client_id) { + client.terminal_size = (cols, rows); + } + self.promote_client_to_foreground(client_id); + self.resize_shared_runtime_to_effective_size(); + true + } + ServerEvent::ClientDetach { client_id } => { + info!(client_id, "client detached"); + let foreground_changed = self.remove_client(client_id); + if foreground_changed { + self.resize_shared_runtime_to_effective_size(); + } + true + } + ServerEvent::ClientDisconnected { client_id } => { + info!(client_id, "client disconnected"); + let foreground_changed = self.remove_client(client_id); + if foreground_changed { + self.resize_shared_runtime_to_effective_size(); + } + true + } + ServerEvent::QuitSignal => { + // The quit check at the top of the loop handles this. + // No render needed — the next iteration will initiate shutdown. + false + } + } + } + + /// Drains API requests with shutdown awareness. + /// + /// During shutdown, remaining requests get a `server_unavailable` error. + fn drain_api_requests_with_shutdown_check(&mut self) -> bool { + let mut changed = false; + while let Ok(msg) = self.app.api_rx.try_recv() { + changed |= self.handle_api_request_with_shutdown_check(msg); + } + changed + } + + /// Handles a single API request with shutdown awareness. + /// + /// Also forwards any toast/sound notifications that result from the API + /// request to connected clients. API methods like `pane.report_agent` + /// trigger internal events that may set toast state or would normally + /// play sounds — in headless mode we forward these to clients instead. + fn handle_api_request_with_shutdown_check(&mut self, msg: api::ApiRequestMessage) -> bool { + if self.shutting_down { + // During shutdown, respond with server_unavailable. + let response = serde_json::to_string(&api::schema::ErrorResponse { + id: msg.request.id, + error: api::schema::ErrorBody { + code: "server_unavailable".into(), + message: "server is shutting down".into(), + }, + }) + .unwrap_or_else(|_| { + r#"{"id":"","error":{"code":"server_unavailable","message":"server is shutting down"}}"# + .to_string() + }); + let _ = msg.respond_to.send(response); + return false; + } + + let changed = api::request_changes_ui(&msg.request); + + // Capture toast and pane agent states before the API call so we can + // forward any resulting notifications to connected clients. + // API requests like pane.report_agent trigger handle_internal_event + // internally, which bypasses drain_internal_events_with_forwarding. + // Since sound.enabled=false in the headless server, sounds would be + // silently dropped; toasts may be set but not forwarded. + // + // Note: pane.report_agent sets hook_authority on the pane, but the + // effective state may not change until the fallback detector confirms + // the agent (detected_agent must match). So we capture both the + // effective state AND the hook authority state for comparison. + let toast_before = self.app.state.toast.clone(); + let pane_states_before: Vec<( + usize, + crate::layout::PaneId, + crate::detect::AgentState, + Option<crate::detect::AgentState>, + )> = self + .app + .state + .workspaces + .iter() + .enumerate() + .flat_map(|(ws_idx, ws)| { + ws.tabs.iter().flat_map(move |tab| { + tab.panes.iter().map(move |(&pane_id, pane)| { + ( + ws_idx, + pane_id, + pane.state, + pane.hook_authority.as_ref().map(|h| h.state), + ) + }) + }) + }) + .collect(); + + let response = self.app.handle_api_request(msg.request); + let _ = msg.respond_to.send(response); + + // Forward any new toast as a notification to clients. + let toast_after = self.app.state.toast.clone(); + if toast_after.is_some() && toast_after != toast_before { + if let Some(toast) = &toast_after { + let msg_text = format!("{}: {}", toast.title, toast.context); + debug!(msg = %msg_text, "forwarding toast notification from API request"); + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Toast, + message: msg_text, + }); + } + } + + // Forward sound notifications for any pane state changes that occurred + // during the API request. Compare before/after pane states (including + // hook authority state) to find transitions that would trigger a sound. + // + // pane.report_agent sets hook_authority on the pane, but the effective + // state may not change until the fallback detector confirms the agent + // (detected_agent must match hook_authority.agent). We check BOTH the + // effective state AND the hook authority state so that API-reported + // state changes trigger notifications even before fallback confirmation. + for (ws_idx, pane_id, prev_effective_state, prev_hook_state) in &pane_states_before { + let pane_after = self + .app + .state + .workspaces + .get(*ws_idx) + .and_then(|ws| ws.tabs.iter().find_map(|tab| tab.panes.get(pane_id))); + + let Some(pane_after) = pane_after else { + continue; + }; + + let new_effective_state = pane_after.state; + let new_hook_state = pane_after.hook_authority.as_ref().map(|h| h.state); + + // Check effective state change first. + let effective_changed = new_effective_state != *prev_effective_state; + // Check hook authority state change — this catches API-reported + // state changes that haven't been confirmed by fallback detection. + let hook_changed = new_hook_state != *prev_hook_state; + + if effective_changed || hook_changed { + // Use the hook state if available (it reflects the API-reported + // state), otherwise use the effective state. + let prev_state = prev_hook_state.unwrap_or(*prev_effective_state); + let new_state = new_hook_state.unwrap_or(new_effective_state); + + // Skip if the derived transition is a no-op. + if prev_state == new_state { + continue; + } + + let is_active_tab = self.app.state.pane_is_in_active_tab(*ws_idx, *pane_id); + + // Get the known agent for sound settings. Unknown custom labels + // fall back to None so clients use the generic sound behavior. + let agent = pane_after.effective_known_agent(); + + debug!( + ws_idx, + pane_id = pane_id.raw(), + prev_state = ?prev_state, + new_state = ?new_state, + agent = ?agent, + effective_changed, + hook_changed, + "pane state changed during API request, checking sound notification" + ); + + // Check agent-specific sound setting but NOT sound.enabled, + // because the server sets enabled=false to prevent local playback. + // Clients decide locally whether to play sounds. + if !matches!( + self.app.state.sound.agents.for_agent(agent), + crate::config::AgentSoundSetting::Off + ) { + if let Some(sound) = crate::app::actions::notification_sound_for_state_change( + is_active_tab, + prev_state, + new_state, + ) { + let msg_text = match sound { + crate::sound::Sound::Done => "agent done", + crate::sound::Sound::Request => "agent attention", + }; + debug!(sound = ?sound, "forwarding sound notification from API request"); + self.send_to_all_clients(ServerMessage::Notify { + kind: protocol::NotifyKind::Sound, + message: msg_text.to_owned(), + }); + } + } + } + } + + changed + } + + /// Renders the current state to client-sized virtual buffers and streams + /// frames to all connected clients. + fn render_and_stream(&mut self) { + let foreground_client_id = self.foreground_client_id; + let mut render_targets: Vec<(u64, (u16, u16), std::sync::mpsc::Sender<Vec<u8>>, bool)> = + self.clients + .iter() + .filter_map(|(&client_id, client)| { + client.writer.as_ref().map(|writer| { + ( + client_id, + client.terminal_size, + writer.clone(), + foreground_client_id == Some(client_id), + ) + }) + }) + .collect(); + + render_targets.sort_by_key(|(client_id, _, _, is_foreground)| (*is_foreground, *client_id)); + + if render_targets.is_empty() { + let (cols, rows) = self.effective_size; + let area = Rect::new(0, 0, cols, rows); + let _ = render_virtual(&mut self.app.state, area, true); + debug!( + cols, + rows, "rendered virtual frame with no attached clients" + ); + return; + } + + let mut broken_clients: Vec<u64> = Vec::new(); + for (client_id, (cols, rows), writer, is_foreground) in render_targets { + let area = Rect::new(0, 0, cols, rows); + let (buffer, cursor) = render_virtual(&mut self.app.state, area, is_foreground); + let message = ServerMessage::Frame(FrameData::from_ratatui_buffer(&buffer, cursor)); + let serialized = match Self::frame_server_message(&message) { + Ok(framed) => framed, + Err(err) => { + warn!(client_id, err = %err, "failed to serialize frame for client"); + broken_clients.push(client_id); + continue; + } + }; + + if writer.send(serialized).is_err() { + debug!(client_id, "client writer channel closed, marking as broken"); + broken_clients.push(client_id); + } + } + + if !broken_clients.is_empty() { + for client_id in broken_clients { + let foreground_changed = self.remove_client(client_id); + if foreground_changed { + self.resize_shared_runtime_to_effective_size(); + } + } + } + + let (cols, rows) = self.effective_size; + debug!(cols, rows, foreground_client_id = ?self.foreground_client_id, "rendered virtual frame(s)"); + } + + /// Handle scheduled tasks for the headless server. + /// + /// Similar to `App::handle_scheduled_tasks` but without resize polling + /// (the server doesn't have a terminal to resize). + fn handle_scheduled_tasks_headless(&mut self, now: Instant) -> bool { + let mut changed = false; + + self.app.sync_animation_timer(now); + + // No resize polling needed — server has no terminal. + // Client resize messages drive size changes instead. + + if self + .app + .config_diagnostic_deadline + .is_some_and(|deadline| now >= deadline) + { + self.app.config_diagnostic_deadline = None; + self.app.state.config_diagnostic = None; + changed = true; + } + + if self + .app + .toast_deadline + .is_some_and(|deadline| now >= deadline) + { + self.app.toast_deadline = None; + self.app.state.toast = None; + changed = true; + } + + if self + .app + .next_animation_tick + .is_some_and(|deadline| now >= deadline) + { + self.app.state.spinner_tick = self.app.state.spinner_tick.wrapping_add(1); + self.app.next_animation_tick = Some(now + app::ANIMATION_INTERVAL); + changed = true; + } + + if self + .app + .git_refresh_deadline() + .is_some_and(|deadline| now >= deadline) + { + for ws in &mut self.app.state.workspaces { + ws.refresh_git_ahead_behind(); + } + self.app.last_git_remote_status_refresh = now; + changed = true; + } + + if self + .app + .next_auto_update_check + .is_some_and(|deadline| now >= deadline) + { + self.app.run_auto_update_check(); + } + + if self + .app + .session_save_deadline + .is_some_and(|deadline| now >= deadline) + { + self.app.save_session_now(); + } + + self.app.sync_animation_timer(now); + changed + } + + /// Initiates graceful shutdown. + fn initiate_shutdown(&mut self) { + if self.shutting_down { + return; + } + info!("server shutdown initiated"); + self.shutting_down = true; + + // Send ServerShutdown to all connected clients. + let shutdown_msg = ServerMessage::ServerShutdown { + reason: Some("server is shutting down".to_owned()), + }; + self.send_to_all_clients(shutdown_msg); + + // Give client writer threads a moment to flush the shutdown message. + // A short sleep ensures the message is written to the socket before + // we close the connections. + std::thread::sleep(Duration::from_millis(50)); + + // Signal the main loop to exit. + self.should_quit.store(true, Ordering::Release); + self.app.state.should_quit = true; + } + + /// Completes the shutdown sequence: send ServerShutdown to clients, + /// close client connections, remove socket files, and clean up. + fn complete_shutdown(&mut self) -> io::Result<()> { + info!("completing server shutdown"); + + // Send ServerShutdown to all remaining clients. + if !self.clients.is_empty() { + let shutdown_msg = ServerMessage::ServerShutdown { + reason: Some("server is shutting down".to_owned()), + }; + self.send_to_all_clients(shutdown_msg); + + // Give writer threads a moment to flush before closing. + std::thread::sleep(Duration::from_millis(50)); + } + + // Drain remaining API requests with server_unavailable. + self.drain_api_requests_with_shutdown_check(); + + // Close all client connections. + self.clients.clear(); + + // Remove socket files. + self.cleanup_sockets()?; + + Ok(()) + } + + /// Removes socket files created by the server. + fn cleanup_sockets(&self) -> io::Result<()> { + if let Err(err) = fs::remove_file(&self.client_socket_path) { + if err.kind() != io::ErrorKind::NotFound { + warn!( + path = %self.client_socket_path.display(), + err = %err, + "failed to remove client socket on shutdown" + ); + } + } + Ok(()) + } +} + +impl Drop for HeadlessServer { + fn drop(&mut self) { + let _ = self.cleanup_sockets(); + } +} + +// --------------------------------------------------------------------------- +// Client handshake +// --------------------------------------------------------------------------- + +/// Internal event sent from the handshake thread to the main event loop. +#[derive(Debug)] +pub enum ServerEvent { + /// A new client completed the handshake. + ClientConnected { + client_id: u64, + cols: u16, + rows: u16, + writer: std::sync::mpsc::Sender<Vec<u8>>, + }, + /// A client sent an input message. + ClientInput { client_id: u64, data: Vec<u8> }, + /// A client sent a resize message. + ClientResize { + client_id: u64, + cols: u16, + rows: u16, + }, + /// A client detached gracefully. + ClientDetach { client_id: u64 }, + /// A client connection was lost. + ClientDisconnected { client_id: u64 }, + /// Ctrl+C or external shutdown signal received. + QuitSignal, +} + +/// 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. +fn handle_client_handshake( + mut stream: UnixStream, + client_id: u64, + server_event_tx: &mpsc::Sender<ServerEvent>, + should_quit: &Arc<AtomicBool>, +) -> io::Result<()> { + // Reset to blocking mode — the accept loop sets nonblocking but + // 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))?; + + // Read the Hello message. + let hello: ClientMessage = match protocol::read_message(&mut stream, MAX_FRAME_SIZE) { + Ok(msg) => msg, + Err(protocol::FramingError::UnexpectedEof) => { + debug!(client_id, "client disconnected before handshake"); + return Ok(()); + } + Err(protocol::FramingError::Oversized { claimed, max }) => { + warn!(client_id, claimed, max, "oversized handshake from client"); + return Ok(()); + } + Err(err) => { + debug!(client_id, err = %err, "failed to read client hello"); + return Ok(()); + } + }; + + let (client_cols, client_rows) = match hello { + ClientMessage::Hello { + version, + cols, + rows, + } => { + // Version check. + match protocol::check_client_version(version) { + protocol::VersionCheck::Compatible => {} + protocol::VersionCheck::Incompatible(reason) => { + // Send rejection Welcome. + let welcome = ServerMessage::Welcome { + version: PROTOCOL_VERSION, + error: Some(reason), + }; + let _ = protocol::write_message(&mut stream, &welcome); + return Ok(()); + } + } + + // Clamp size. + let (clamped_cols, clamped_rows) = clamp_terminal_size(cols, rows); + (clamped_cols, clamped_rows) + } + _ => { + // First message must be Hello. + debug!(client_id, "first message was not Hello, closing"); + let welcome = ServerMessage::Welcome { + version: PROTOCOL_VERSION, + error: Some("expected Hello as first message".to_owned()), + }; + let _ = protocol::write_message(&mut stream, &welcome); + return Ok(()); + } + }; + + // Send Welcome. + let welcome = ServerMessage::Welcome { + version: PROTOCOL_VERSION, + error: None, + }; + protocol::write_message(&mut stream, &welcome) + .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + + // Clear read timeout for normal operation. + stream.set_read_timeout(None)?; + + // Create a channel for writing frames back to the client. + let (frame_tx, frame_rx) = std::sync::mpsc::channel::<Vec<u8>>(); + + // Notify the main loop about the new client. + let _ = server_event_tx.blocking_send(ServerEvent::ClientConnected { + client_id, + cols: client_cols, + rows: client_rows, + writer: frame_tx, + }); + + // Spawn a writer thread that forwards frames from the channel to the stream. + let write_stream = stream.try_clone()?; + let write_quit = should_quit.clone(); + std::thread::spawn(move || { + client_writer_loop(write_stream, frame_rx, &write_quit); + }); + + // Enter read loop — read client messages and forward to main loop. + client_read_loop(stream, client_id, server_event_tx, should_quit) +} + +/// The client writer loop — sends frames from the channel to the client stream. +fn client_writer_loop( + mut stream: UnixStream, + frame_rx: std::sync::mpsc::Receiver<Vec<u8>>, + should_quit: &Arc<AtomicBool>, +) { + while !should_quit.load(Ordering::Acquire) { + match frame_rx.recv_timeout(Duration::from_millis(100)) { + Ok(data) => { + if let Err(err) = stream.write_all(&data) { + debug!(err = %err, "client write failed, closing writer"); + break; + } + if let Err(err) = stream.flush() { + debug!(err = %err, "client flush failed, closing writer"); + break; + } + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue, + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, + } + } + debug!("client writer thread exiting"); +} + +/// The client read loop — reads messages from the client and forwards to the server event channel. +fn client_read_loop( + mut stream: UnixStream, + client_id: u64, + server_event_tx: &mpsc::Sender<ServerEvent>, + should_quit: &Arc<AtomicBool>, +) -> io::Result<()> { + while !should_quit.load(Ordering::Acquire) { + let msg: ClientMessage = match protocol::read_message(&mut stream, MAX_FRAME_SIZE) { + Ok(msg) => msg, + Err(protocol::FramingError::UnexpectedEof) => { + // Client disconnected. + let _ = + server_event_tx.blocking_send(ServerEvent::ClientDisconnected { client_id }); + break; + } + Err(protocol::FramingError::Oversized { claimed, max }) => { + warn!( + client_id, + claimed, max, "oversized message from client, closing" + ); + let _ = + server_event_tx.blocking_send(ServerEvent::ClientDisconnected { client_id }); + break; + } + Err(err) => { + debug!(client_id, err = %err, "client read error, closing"); + let _ = + server_event_tx.blocking_send(ServerEvent::ClientDisconnected { client_id }); + break; + } + }; + + let event = match msg { + ClientMessage::Input { data } => { + // Validate input size. + if data.len() > MAX_INPUT_PAYLOAD { + warn!( + client_id, + size = data.len(), + "oversized input from client, closing" + ); + ServerEvent::ClientDisconnected { client_id } + } else { + ServerEvent::ClientInput { client_id, data } + } + } + ClientMessage::Resize { cols, rows } => { + let (clamped_cols, clamped_rows) = clamp_terminal_size(cols, rows); + ServerEvent::ClientResize { + client_id, + cols: clamped_cols, + rows: clamped_rows, + } + } + ClientMessage::Detach => ServerEvent::ClientDetach { client_id }, + ClientMessage::Hello { .. } => { + // Duplicate Hello — ignore. + continue; + } + }; + + if server_event_tx.blocking_send(event).is_err() { + break; // Main loop gone. + } + } + + debug!(client_id, "client read thread exiting"); + Ok(()) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Installs a Ctrl+C handler that sets the should_quit flag and wakes up +/// the event loop by sending a QuitSignal on the server event channel. +fn ctrlc_handler(should_quit: Arc<AtomicBool>, server_event_tx: mpsc::Sender<ServerEvent>) { + let _ = ctrlc::set_handler(move || { + should_quit.store(true, Ordering::Release); + // Wake up the event loop so the quit flag is checked promptly. + let _ = server_event_tx.try_send(ServerEvent::QuitSignal); + }); +} + +/// Sleep until a deadline, or return pending if none. +async fn sleep_until_or_pending(deadline: Option<Instant>) { + match deadline { + Some(deadline) => tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await, + None => std::future::pending().await, + } +} + +// --------------------------------------------------------------------------- +// Entry point +// --------------------------------------------------------------------------- + +/// Run the headless server. This is the entry point called from main.rs. +pub fn run_server() -> io::Result<()> { + init_logging(); + + let loaded_config = config::Config::load(); + let (api_tx, api_rx) = tokio::sync::mpsc::unbounded_channel(); + let event_hub = api::EventHub::default(); + + // Start the JSON API socket server. + let _api_server = match api::start_server(api_tx, event_hub.clone()) { + Ok(server) => server, + Err(err) if err.kind() == io::ErrorKind::AddrInUse => { + eprintln!("error: herdr server is already running"); + eprintln!("api socket: {}", api::socket_path().display()); + std::process::exit(1); + } + Err(err) => return Err(err), + }; + + let no_session = false; // Server always does session persistence. + let _show_changelog = false; + + let rt = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?; + + let result = rt.block_on(async { + // Create the App (with AppState, event channels, etc.). + let mut app = app::App::new( + &loaded_config.config, + no_session, + None, // config_diagnostic + None, // startup_release_notes + api_rx, + event_hub, + ); + + // The server runs headless — disable local sound playback. + // Sound notifications are forwarded to connected clients as + // ServerMessage::Notify instead of played locally. + app.state.sound.enabled = false; + + // Create the headless server. + let mut server = match HeadlessServer::new(app) { + Ok(server) => server, + Err(err) if err.kind() == io::ErrorKind::AddrInUse => { + eprintln!("error: herdr server is already running"); + eprintln!("client socket: {}", client_socket_path().display()); + std::process::exit(1); + } + Err(err) => return Err(err), + }; + + info!( + api_socket = %api::socket_path().display(), + client_socket = %client_socket_path().display(), + "herdr server started" + ); + + server.run().await + }); + + rt.shutdown_timeout(Duration::from_millis(100)); + result +} + +/// Initialize logging for the server process. +fn init_logging() { + crate::logging::init_file_logging("herdr-server.log"); +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + use crate::server::protocol::CursorState; + + fn test_headless_server() -> HeadlessServer { + let config = crate::config::Config::default(); + let (_api_tx, api_rx) = tokio::sync::mpsc::unbounded_channel(); + let app = crate::app::App::new(&config, true, None, None, api_rx, api::EventHub::default()); + + let dir = std::env::temp_dir().join(format!( + "herdr-headless-test-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + let _ = fs::create_dir_all(&dir); + let socket_path = dir.join("client.sock"); + let listener = UnixListener::bind(&socket_path).expect("bind test listener"); + listener + .set_nonblocking(true) + .expect("set listener nonblocking"); + let (server_event_tx, server_event_rx) = mpsc::channel(64); + + HeadlessServer { + app, + client_listener: listener, + client_socket_path: socket_path, + clients: HashMap::new(), + next_client_id: 1, + foreground_client_id: None, + next_activity_stamp: 1, + effective_size: (MIN_COLS, MIN_ROWS), + shutting_down: false, + should_quit: Arc::new(AtomicBool::new(false)), + server_event_rx, + server_event_tx, + } + } + + fn read_server_message(bytes: Vec<u8>) -> ServerMessage { + let mut cursor = std::io::Cursor::new(bytes); + protocol::read_message(&mut cursor, MAX_FRAME_SIZE).expect("decode server message") + } + + fn read_server_frame(bytes: Vec<u8>) -> FrameData { + match read_server_message(bytes) { + ServerMessage::Frame(frame) => frame, + other => panic!("expected frame, got {other:?}"), + } + } + + #[test] + fn clamp_terminal_size_zero_zero() { + assert_eq!( + clamp_terminal_size(0, 0), + (MIN_CLIENT_COLS, MIN_CLIENT_ROWS) + ); + } + + #[test] + fn clamp_terminal_size_one_one() { + assert_eq!(clamp_terminal_size(1, 1), (1, 1)); + } + + #[test] + fn clamp_terminal_size_preserves_narrow_client_size() { + assert_eq!(clamp_terminal_size(40, 12), (40, 12)); + } + + #[test] + fn clamp_terminal_size_valid() { + assert_eq!(clamp_terminal_size(120, 40), (120, 40)); + } + + #[test] + fn clamp_terminal_size_exact_minimum() { + assert_eq!( + clamp_terminal_size(MIN_CLIENT_COLS, MIN_CLIENT_ROWS), + (MIN_CLIENT_COLS, MIN_CLIENT_ROWS) + ); + } + + #[test] + fn client_socket_path_derived_from_api_socket_override() { + let path = client_socket_path_from_overrides(Some("/tmp/test-herdr.sock"), None); + assert_eq!(path, PathBuf::from("/tmp/test-herdr-client.sock")); + } + + #[test] + fn client_socket_path_api_override_takes_precedence_over_legacy_client_override() { + let path = client_socket_path_from_overrides( + Some("/tmp/test-herdr.sock"), + Some("/tmp/legacy-client.sock"), + ); + assert_eq!(path, PathBuf::from("/tmp/test-herdr-client.sock")); + } + + #[test] + fn client_socket_path_respects_legacy_client_override_without_api_override() { + let path = client_socket_path_from_overrides(None, Some("/tmp/test-herdr-client.sock")); + assert_eq!(path, PathBuf::from("/tmp/test-herdr-client.sock")); + } + + #[test] + fn client_socket_path_defaults_to_config_dir() { + let path = client_socket_path_from_overrides(None, None); + assert_eq!(path, config::config_dir().join("herdr-client.sock")); + } + + #[test] + fn derive_client_socket_from_api_socket_without_sock_extension() { + let derived = derive_client_socket_from_api_socket(Path::new("/tmp/custom-api")); + assert_eq!(derived, PathBuf::from("/tmp/custom-api-client.sock")); + } + + #[test] + fn prepare_socket_path_removes_stale_socket() { + let dir = std::env::temp_dir().join(format!( + "herdr-test-stale-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + let _ = fs::create_dir_all(&dir); + let socket_path = dir.join("stale.sock"); + + // Create a socket file that nobody is listening on. + let _ = UnixListener::bind(&socket_path); + // The listener goes out of scope, so the socket becomes stale. + + // prepare_socket_path should remove it without error. + let result = prepare_socket_path(&socket_path); + assert!(result.is_ok(), "should remove stale socket: {result:?}"); + + // Socket file should be gone. + assert!(!socket_path.exists()); + + let _ = fs::remove_dir_all(&dir); + } + + #[test] + fn prepare_socket_path_rejects_live_socket() { + let dir = std::env::temp_dir().join(format!( + "herdr-test-live-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + let _ = fs::create_dir_all(&dir); + let socket_path = dir.join("live.sock"); + + // Bind a live listener. + let _listener = UnixListener::bind(&socket_path).expect("bind"); + + let result = prepare_socket_path(&socket_path); + assert!(result.is_err()); + assert_eq!(result.unwrap_err().kind(), io::ErrorKind::AddrInUse); + + let _ = fs::remove_dir_all(&dir); + } + + #[test] + fn virtual_render_produces_nonempty_buffer() { + let mut state = AppState::test_new(); + let area = Rect::new(0, 0, 80, 24); + let (buffer, _cursor) = render_virtual(&mut state, area, true); + assert_eq!(buffer.area.width, 80); + assert_eq!(buffer.area.height, 24); + } + + #[test] + fn virtual_render_without_frame_cursor_keeps_cursor_hidden() { + let mut state = AppState::test_new(); + let area = Rect::new(0, 0, 80, 24); + let (_buffer, cursor) = render_virtual(&mut state, area, true); + + assert_eq!(cursor, None); + } + + #[tokio::test] + async fn virtual_render_preserves_explicit_frame_cursor_position() { + let mut state = AppState::test_new(); + let mut ws = crate::workspace::Workspace::test_new("test"); + let pane_id = ws.tabs[0].root_pane; + ws.tabs[0].runtimes.insert( + pane_id, + crate::pane::PaneRuntime::test_with_screen_bytes(20, 5, b"left"), + ); + + state.workspaces = vec![ws]; + state.active = Some(0); + state.selected = 0; + state.mode = crate::app::Mode::Terminal; + + let area = Rect::new(0, 0, 80, 24); + let (_buffer, cursor) = render_virtual(&mut state, area, true); + let pane = state + .view + .pane_infos + .iter() + .find(|info| info.id == pane_id) + .expect("focused pane info"); + + assert_eq!( + cursor, + Some(CursorState { + x: pane.inner_rect.x + 4, + y: pane.inner_rect.y, + visible: true, + }) + ); + } + + #[test] + fn latest_active_client_drives_shared_size_theme_and_fallback() { + let mut server = test_headless_server(); + + server.clients.insert( + 1, + ClientConnection { + terminal_size: (160, 45), + host_terminal_theme: crate::terminal_theme::TerminalTheme { + foreground: Some(crate::terminal_theme::RgbColor { + r: 0xaa, + g: 0xbb, + b: 0xcc, + }), + background: Some(crate::terminal_theme::RgbColor { + r: 0x11, + g: 0x22, + b: 0x33, + }), + }, + last_activity: 1, + writer: None, + }, + ); + server.clients.insert( + 2, + ClientConnection { + terminal_size: (80, 24), + host_terminal_theme: crate::terminal_theme::TerminalTheme { + foreground: Some(crate::terminal_theme::RgbColor { + r: 0x10, + g: 0x20, + b: 0x30, + }), + background: Some(crate::terminal_theme::RgbColor { + r: 0xdd, + g: 0xee, + b: 0xff, + }), + }, + last_activity: 2, + writer: None, + }, + ); + + assert!(server.promote_client_to_foreground(1)); + assert_eq!(server.foreground_client_id, Some(1)); + assert_eq!(server.effective_size, (160, 45)); + assert_eq!( + server.app.state.host_terminal_theme, + server.clients[&1].host_terminal_theme + ); + + assert!(server.promote_client_to_foreground(2)); + assert_eq!(server.foreground_client_id, Some(2)); + assert_eq!(server.effective_size, (80, 24)); + assert_eq!( + server.app.state.host_terminal_theme, + server.clients[&2].host_terminal_theme + ); + + assert!(server.remove_client(2)); + assert_eq!(server.foreground_client_id, Some(1)); + assert_eq!(server.effective_size, (160, 45)); + assert_eq!( + server.app.state.host_terminal_theme, + server.clients[&1].host_terminal_theme + ); + } + + #[test] + fn render_and_stream_uses_each_client_terminal_size() { + let mut server = test_headless_server(); + server.app.state.workspaces = vec![crate::workspace::Workspace::test_new("test")]; + server.app.state.active = Some(0); + server.app.state.selected = 0; + server.app.state.mode = crate::app::Mode::Terminal; + + let (desktop_tx, desktop_rx) = std::sync::mpsc::channel(); + let (phone_tx, phone_rx) = std::sync::mpsc::channel(); + + server.clients.insert( + 1, + ClientConnection { + terminal_size: (120, 40), + host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), + last_activity: 1, + writer: Some(desktop_tx), + }, + ); + server.clients.insert( + 2, + ClientConnection { + terminal_size: (80, 24), + host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), + last_activity: 2, + writer: Some(phone_tx), + }, + ); + server.foreground_client_id = Some(1); + server.sync_foreground_client_state(); + server.resize_shared_runtime_to_effective_size(); + + server.render_and_stream(); + + let desktop_frame = read_server_frame(desktop_rx.recv().expect("desktop frame")); + let phone_frame = read_server_frame(phone_rx.recv().expect("phone frame")); + + assert_eq!((desktop_frame.width, desktop_frame.height), (120, 40)); + assert_eq!((phone_frame.width, phone_frame.height), (80, 24)); + } + + #[test] + fn clipboard_write_targets_foreground_client_only() { + let mut server = test_headless_server(); + let (background_tx, background_rx) = std::sync::mpsc::channel(); + let (foreground_tx, foreground_rx) = std::sync::mpsc::channel(); + + server.clients.insert( + 1, + ClientConnection { + terminal_size: (120, 40), + host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), + last_activity: 1, + writer: Some(background_tx), + }, + ); + server.clients.insert( + 2, + ClientConnection { + terminal_size: (80, 24), + host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), + last_activity: 2, + writer: Some(foreground_tx), + }, + ); + server.foreground_client_id = Some(2); + server.sync_foreground_client_state(); + + let changed = server.handle_internal_event_with_forwarding(AppEvent::ClipboardWrite { + content: b"test".to_vec(), + }); + + assert!(!changed); + match read_server_message( + foreground_rx + .recv_timeout(Duration::from_millis(100)) + .expect("foreground clipboard message"), + ) { + ServerMessage::Clipboard { data } => assert_eq!(data, "dGVzdA=="), + other => panic!("expected clipboard message, got {other:?}"), + } + assert!( + background_rx + .recv_timeout(Duration::from_millis(50)) + .is_err(), + "background client should not receive clipboard writes" + ); + } + + #[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 + // OS overhead (thread scheduling, timer slack, cleanup). + assert!( + HANDSHAKE_TIMEOUT < Duration::from_secs(5), + "HANDSHAKE_TIMEOUT ({:?}) must be less than 5 seconds to guarantee \ + connection close within the 5-second deadline", + HANDSHAKE_TIMEOUT + ); + } + + /// Verify that no direct calls to `self.app.handle_internal_event` + /// exist outside of `handle_internal_event_with_forwarding` in this + /// module. This ensures the forwarding bypass cannot be reintroduced. + /// + /// The search pattern looks for `handle_internal_event` calls that + /// are NOT inside the `handle_internal_event_with_forwarding` method. + #[test] + fn no_handle_internal_event_bypass_in_module() { + let source = include_str!("headless.rs"); + + // Find all lines containing handle_internal_event + let mut bypass_lines: Vec<String> = Vec::new(); + let mut inside_forwarding_method = false; + let mut forwarding_method_brace_depth = 0u32; + + for (i, line) in source.lines().enumerate() { + let line_num = i + 1; + + // Track when we're inside handle_internal_event_with_forwarding + if line.contains("fn handle_internal_event_with_forwarding") { + inside_forwarding_method = true; + forwarding_method_brace_depth = 0; + } + + if inside_forwarding_method { + // Count braces to track when we exit the method + for ch in line.chars() { + match ch { + '{' => forwarding_method_brace_depth += 1, + '}' => { + forwarding_method_brace_depth = + forwarding_method_brace_depth.saturating_sub(1); + if forwarding_method_brace_depth == 0 { + inside_forwarding_method = false; + } + } + _ => {} + } + } + } else if line.contains("self.app.handle_internal_event(") + && !line.trim().starts_with("///") + && !line.contains("contains(") + { + // Direct call to handle_internal_event outside the forwarding method + bypass_lines.push(format!("line {}: {}", line_num, line.trim())); + } + } + + assert!( + bypass_lines.is_empty(), + "Found direct calls to self.app.handle_internal_event outside \ + handle_internal_event_with_forwarding (bypass risk):\n {}", + bypass_lines.join("\n ") + ); + } +} diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 00000000..5b5f1d4e --- /dev/null +++ b/src/server/mod.rs @@ -0,0 +1,3 @@ +pub mod autodetect; +pub mod headless; +pub mod protocol; diff --git a/src/server/protocol.rs b/src/server/protocol.rs new file mode 100644 index 00000000..f3784a5e --- /dev/null +++ b/src/server/protocol.rs @@ -0,0 +1,1212 @@ +//! Wire protocol for herdr server/client communication. +//! +//! Defines the message types, framing, version negotiation, and safety +//! constraints for the binary protocol over Unix domain sockets. + +use std::io::{self, Read, Write}; + +use serde::{Deserialize, Serialize}; + +// --------------------------------------------------------------------------- +// Protocol constants +// --------------------------------------------------------------------------- + +/// Current protocol version. Bumped when wire format changes incompatibly. +pub const PROTOCOL_VERSION: u32 = 1; + +/// Maximum allowed frame payload size (2 MB). Frames larger than this are +/// rejected to prevent denial-of-service via oversized length prefixes. +pub const MAX_FRAME_SIZE: usize = 2 * 1024 * 1024; + +/// Length of the u32 little-endian length prefix in bytes. +const LENGTH_PREFIX_BYTES: usize = 4; + +// --------------------------------------------------------------------------- +// Client → Server messages +// --------------------------------------------------------------------------- + +/// Messages sent from the client to the server over the client protocol socket. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum ClientMessage { + /// Handshake: client announces its protocol version and terminal dimensions. + Hello { + /// Protocol version the client speaks. + version: u32, + /// Terminal width in columns. + cols: u16, + /// Terminal height in rows. + rows: u16, + }, + + /// Raw input bytes read from the client's stdin. + Input { + /// Raw terminal input (possibly multi-byte escape sequences). + data: Vec<u8>, + }, + + /// Terminal resize notification from the client. + Resize { + /// New terminal width in columns. + cols: u16, + /// New terminal height in rows. + rows: u16, + }, + + /// Graceful disconnect request. + Detach, +} + +// --------------------------------------------------------------------------- +// Server → Client messages +// --------------------------------------------------------------------------- + +/// A single cell in a rendered frame, serialized independently from ratatui's +/// `Cell` type to keep the wire protocol stable. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CellData { + /// Grapheme cluster displayed in this cell (usually 1–2 chars). + pub symbol: String, + /// Foreground color as a packed u32 (0xAARRGGBB or ratatui Color index). + pub fg: u32, + /// Background color as a packed u32. + pub bg: u32, + /// Bitmask of style modifiers (bold, italic, etc.). + pub modifier: u16, + /// Whether this cell should be skipped during diff-based rendering. + pub skip: bool, +} + +/// Cursor position within a rendered frame. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CursorState { + /// Column offset (0-based) of the cursor. + pub x: u16, + /// Row offset (0-based) of the cursor. + pub y: u16, + /// Whether the cursor is visible. + pub visible: bool, +} + +/// A rendered frame to be displayed by the client. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FrameData { + /// Cells in row-major order. Length must equal `width * height`. + pub cells: Vec<CellData>, + /// Frame width in columns. + pub width: u16, + /// Frame height in rows. + pub height: u16, + /// Cursor state for this frame, if applicable. + pub cursor: Option<CursorState>, +} + +impl FrameData { + /// Creates a `FrameData` from a ratatui `Buffer` and optional cursor. + /// + /// This converts ratatui's internal cell representation into the + /// wire-protocol cell format. The conversion is lossless for all + /// commonly used cell attributes. + pub fn from_ratatui_buffer( + buffer: &ratatui::buffer::Buffer, + cursor: Option<CursorState>, + ) -> Self { + let area = buffer.area; + let width = area.width; + let height = area.height; + + let mut cells = Vec::with_capacity((width as usize) * (height as usize)); + for row in 0..height { + for col in 0..width { + let cell = buffer.cell((col, row)).expect("cell within bounds"); + cells.push(CellData { + symbol: cell.symbol().to_owned(), + fg: color_to_u32(cell.fg), + bg: color_to_u32(cell.bg), + modifier: modifier_to_u16(cell.modifier), + skip: cell.skip, + }); + } + } + + FrameData { + cells, + width, + height, + cursor, + } + } + + /// Reconstructs a ratatui `Buffer` from this frame data. + /// + /// Returns `None` if the cells vector length doesn't match `width * height`. + #[cfg(test)] + pub fn to_ratatui_buffer(&self) -> Option<ratatui::buffer::Buffer> { + let expected = (self.width as usize) * (self.height as usize); + if self.cells.len() != expected { + return None; + } + + let area = ratatui::layout::Rect::new(0, 0, self.width, self.height); + let mut buffer = ratatui::buffer::Buffer::filled(area, ratatui::buffer::Cell::new(" ")); + + for row in 0..self.height { + for col in 0..self.width { + let idx = (row as usize) * (self.width as usize) + (col as usize); + let cell_data = &self.cells[idx]; + let cell = buffer.cell_mut((col, row)).expect("cell within bounds"); + cell.set_symbol(&cell_data.symbol); + cell.fg = u32_to_color(cell_data.fg); + cell.bg = u32_to_color(cell_data.bg); + cell.modifier = u16_to_modifier(cell_data.modifier); + cell.skip = cell_data.skip; + } + } + + Some(buffer) + } +} + +/// Notification kind forwarded from server to client. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum NotifyKind { + /// Play a sound (bell/agent-done, etc.). + Sound, + /// Display a toast message. + Toast, +} + +/// Messages sent from the server to the client over the client protocol socket. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum ServerMessage { + /// Handshake response: server acknowledges (or rejects) the client. + Welcome { + /// Protocol version the server speaks. + version: u32, + /// If present, the handshake failed and this describes why. + /// The client should exit with a clear error message. + error: Option<String>, + }, + + /// A rendered frame to be displayed by the client. + Frame(FrameData), + + /// Server is shutting down. Clients should exit gracefully. + ServerShutdown { + /// Optional reason for the shutdown. + reason: Option<String>, + }, + + /// A notification event (sound/toast) to be rendered locally by the client. + Notify { + /// What kind of notification. + kind: NotifyKind, + /// Human-readable message. + message: String, + }, + + /// OSC 52 clipboard data forwarded from a PTY through the server. + Clipboard { + /// Base64-encoded clipboard data. + data: String, + }, +} + +// --------------------------------------------------------------------------- +// Color / Modifier conversion helpers +// --------------------------------------------------------------------------- + +/// Converts a ratatui `Color` to a packed u32 for wire transport. +/// +/// Encoding: +/// - Named colors (Reset, Black, …, White) → `0x00_00_00_XX` where XX is 0..=16 +/// - Indexed palette → `0x01_00_00_XX` where XX is the palette index +/// - RGB → `0x02_RR_GG_BB` with components in the lower 3 bytes +fn color_to_u32(color: ratatui::style::Color) -> u32 { + match color { + ratatui::style::Color::Reset => 0x00_00_00_00, + ratatui::style::Color::Black => 0x00_00_00_01, + ratatui::style::Color::Red => 0x00_00_00_02, + ratatui::style::Color::Green => 0x00_00_00_03, + ratatui::style::Color::Yellow => 0x00_00_00_04, + ratatui::style::Color::Blue => 0x00_00_00_05, + ratatui::style::Color::Magenta => 0x00_00_00_06, + ratatui::style::Color::Cyan => 0x00_00_00_07, + ratatui::style::Color::Gray => 0x00_00_00_08, + ratatui::style::Color::DarkGray => 0x00_00_00_09, + ratatui::style::Color::LightRed => 0x00_00_00_0A, + ratatui::style::Color::LightGreen => 0x00_00_00_0B, + ratatui::style::Color::LightYellow => 0x00_00_00_0C, + ratatui::style::Color::LightBlue => 0x00_00_00_0D, + ratatui::style::Color::LightMagenta => 0x00_00_00_0E, + ratatui::style::Color::LightCyan => 0x00_00_00_0F, + ratatui::style::Color::White => 0x00_00_00_10, + ratatui::style::Color::Indexed(i) => 0x01_00_00_00 | (i as u32), + ratatui::style::Color::Rgb(r, g, b) => { + 0x02_00_00_00 | ((r as u32) << 16) | ((g as u32) << 8) | (b as u32) + } + } +} + +/// Converts a packed u32 back to a ratatui `Color`. +#[cfg(test)] +fn u32_to_color(val: u32) -> ratatui::style::Color { + match val >> 24 { + 0x00 => match val & 0xFF { + 0x00 => ratatui::style::Color::Reset, + 0x01 => ratatui::style::Color::Black, + 0x02 => ratatui::style::Color::Red, + 0x03 => ratatui::style::Color::Green, + 0x04 => ratatui::style::Color::Yellow, + 0x05 => ratatui::style::Color::Blue, + 0x06 => ratatui::style::Color::Magenta, + 0x07 => ratatui::style::Color::Cyan, + 0x08 => ratatui::style::Color::Gray, + 0x09 => ratatui::style::Color::DarkGray, + 0x0A => ratatui::style::Color::LightRed, + 0x0B => ratatui::style::Color::LightGreen, + 0x0C => ratatui::style::Color::LightYellow, + 0x0D => ratatui::style::Color::LightBlue, + 0x0E => ratatui::style::Color::LightMagenta, + 0x0F => ratatui::style::Color::LightCyan, + 0x10 => ratatui::style::Color::White, + _ => ratatui::style::Color::Reset, // unknown named → Reset + }, + 0x01 => ratatui::style::Color::Indexed((val & 0xFF) as u8), + 0x02 => { + let r = ((val >> 16) & 0xFF) as u8; + let g = ((val >> 8) & 0xFF) as u8; + let b = (val & 0xFF) as u8; + ratatui::style::Color::Rgb(r, g, b) + } + _ => ratatui::style::Color::Reset, // unknown tag → Reset + } +} + +/// Converts a ratatui `Modifier` bitmask to a u16 for wire transport. +fn modifier_to_u16(modifier: ratatui::style::Modifier) -> u16 { + modifier.bits() +} + +/// Converts a u16 back to a ratatui `Modifier`. +#[cfg(test)] +fn u16_to_modifier(val: u16) -> ratatui::style::Modifier { + ratatui::style::Modifier::from_bits_truncate(val) +} + +// --------------------------------------------------------------------------- +// Framing: length-prefixed binary messages +// --------------------------------------------------------------------------- + +/// Errors that can occur during framing operations. +#[derive(Debug)] +pub enum FramingError { + /// The decoded payload length exceeds the configured maximum frame size. + Oversized { claimed: usize, max: usize }, + /// An I/O error occurred while reading or writing. + Io(io::Error), + /// Bincode serialization or deserialization failed. + Bincode(String), + /// The connection was closed before a complete frame could be read. + UnexpectedEof, +} + +impl std::fmt::Display for FramingError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + FramingError::Oversized { claimed, max } => { + write!(f, "frame size {claimed} exceeds maximum {max}") + } + FramingError::Io(e) => write!(f, "I/O error: {e}"), + FramingError::Bincode(e) => write!(f, "bincode error: {e}"), + FramingError::UnexpectedEof => write!(f, "unexpected end of stream"), + } + } +} + +impl std::error::Error for FramingError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + FramingError::Io(e) => Some(e), + _ => None, + } + } +} + +impl From<io::Error> for FramingError { + fn from(e: io::Error) -> Self { + FramingError::Io(e) + } +} + +/// Serializes a message and writes it as a length-prefixed frame: +/// `[u32LE length][bincode payload]`. +/// +/// This is a blocking/synchronous write suitable for use with `std::os::unix::net::UnixStream` +/// in blocking mode, or with any `Write` implementor. +/// +/// # Errors +/// +/// Returns `FramingError::Bincode` if the payload length exceeds `u32::MAX` +/// (would be truncated by the length prefix cast). +pub fn write_message<W: Write, M: Serialize>(writer: &mut W, msg: &M) -> Result<(), FramingError> { + let payload = bincode::serde::encode_to_vec(msg, bincode::config::standard()) + .map_err(|e| FramingError::Bincode(e.to_string()))?; + + let len = payload.len(); + if len > u32::MAX as usize { + return Err(FramingError::Bincode(format!( + "payload length {len} exceeds u32::MAX ({}), would be truncated by length prefix", + u32::MAX + ))); + } + + writer.write_all(&(len as u32).to_le_bytes())?; + writer.write_all(&payload)?; + writer.flush()?; + Ok(()) +} + +/// Reads and deserializes a length-prefixed frame from a reader. +/// +/// Reassembles partial reads correctly. Rejects frames whose declared +/// length exceeds `max_frame_size` without panicking or allocating +/// oversized buffers. +pub fn read_message<R: Read, M: for<'de> Deserialize<'de>>( + reader: &mut R, + max_frame_size: usize, +) -> Result<M, FramingError> { + // Read the 4-byte length prefix, reassembling partial reads. + let mut len_buf = [0u8; LENGTH_PREFIX_BYTES]; + read_exact_or_eof(reader, &mut len_buf)?; + let claimed_len = u32::from_le_bytes(len_buf) as usize; + + if claimed_len > max_frame_size { + return Err(FramingError::Oversized { + claimed: claimed_len, + max: max_frame_size, + }); + } + + // Read the payload, reassembling partial reads. + let mut payload = vec![0u8; claimed_len]; + read_exact_or_eof(reader, &mut payload)?; + + let (msg, consumed) = bincode::serde::decode_from_slice(&payload, bincode::config::standard()) + .map_err(|e| FramingError::Bincode(e.to_string()))?; + + // Enforce that the decoder consumed the full payload. + // Trailing bytes after the decoded message indicate a protocol violation + // (e.g., a corrupted length prefix or concatenated payloads). + if consumed != claimed_len { + return Err(FramingError::Bincode(format!( + "decoded {} bytes but payload length was {claimed_len}; trailing bytes are not allowed", + consumed + ))); + } + + Ok(msg) +} + +/// Like `Read::read_exact`, but returns `FramingError::UnexpectedEof` +/// when the reader hits end-of-stream before filling the buffer, instead +/// of the generic `io::ErrorKind::UnexpectedEof`. +fn read_exact_or_eof<R: Read>(reader: &mut R, buf: &mut [u8]) -> Result<(), FramingError> { + reader.read_exact(buf).map_err(|e| { + if e.kind() == io::ErrorKind::UnexpectedEof { + FramingError::UnexpectedEof + } else { + FramingError::Io(e) + } + }) +} + +// --------------------------------------------------------------------------- +// Version negotiation +// --------------------------------------------------------------------------- + +/// Result of checking a client's protocol version against the server's. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum VersionCheck { + /// Versions are compatible. The server should reply `Welcome { version, error: None }`. + Compatible, + /// Versions are incompatible. The server should reply + /// `Welcome { version, error: Some(reason) }` and close the connection. + Incompatible(String), +} + +/// Checks whether a client's protocol version is compatible with this server. +/// +/// Current rules: +/// - Version 0 (pre-persistence client) is always rejected. +/// - Matching major versions are accepted. +/// - A client with a newer version than the server is rejected. +/// - A client with an older version than the server is rejected +/// (backward compatibility is not yet supported). +pub fn check_client_version(client_version: u32) -> VersionCheck { + if client_version == 0 { + return VersionCheck::Incompatible( + "pre-persistence client (version 0) is not supported".to_owned(), + ); + } + + if client_version == PROTOCOL_VERSION { + VersionCheck::Compatible + } else if client_version < PROTOCOL_VERSION { + VersionCheck::Incompatible(format!( + "client version {client_version} is older than server version {PROTOCOL_VERSION}; please upgrade your herdr client" + )) + } else { + VersionCheck::Incompatible(format!( + "client version {client_version} is newer than server version {PROTOCOL_VERSION}; please upgrade the herdr server" + )) + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use ratatui::style::{Color, Modifier}; + + // ---- Round-trip: ClientMessage ---- + + #[test] + fn client_hello_roundtrip() { + let msg = ClientMessage::Hello { + version: 1, + cols: 80, + rows: 24, + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ClientMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn client_input_roundtrip() { + let msg = ClientMessage::Input { + data: vec![0x1b, 0x5b, 0x41], // ESC [ A (up arrow) + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ClientMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn client_resize_roundtrip() { + let msg = ClientMessage::Resize { cols: 80, rows: 24 }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ClientMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn client_detach_roundtrip() { + let msg = ClientMessage::Detach; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ClientMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + // ---- Round-trip: ServerMessage ---- + + #[test] + fn server_welcome_roundtrip() { + let msg = ServerMessage::Welcome { + version: 1, + error: None, + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ServerMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn server_welcome_with_error_roundtrip() { + let msg = ServerMessage::Welcome { + version: 1, + error: Some("incompatible version".to_owned()), + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ServerMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn server_frame_roundtrip_nontrivial() { + // Build a 3×2 frame with varied styles (≥2×2). + let frame = FrameData { + cells: vec![ + CellData { + symbol: "H".into(), + fg: color_to_u32(Color::Red), + bg: color_to_u32(Color::Black), + modifier: Modifier::BOLD.bits(), + skip: false, + }, + CellData { + symbol: "i".into(), + fg: color_to_u32(Color::Green), + bg: color_to_u32(Color::Reset), + modifier: Modifier::ITALIC.bits(), + skip: false, + }, + CellData { + symbol: "!".into(), + fg: color_to_u32(Color::Rgb(255, 128, 0)), + bg: color_to_u32(Color::Indexed(220)), + modifier: (Modifier::BOLD | Modifier::UNDERLINED).bits(), + skip: false, + }, + CellData { + symbol: " ".into(), + fg: color_to_u32(Color::Reset), + bg: color_to_u32(Color::Reset), + modifier: Modifier::empty().bits(), + skip: true, + }, + CellData { + symbol: "→".into(), // multi-byte grapheme + fg: color_to_u32(Color::Cyan), + bg: color_to_u32(Color::Blue), + modifier: Modifier::REVERSED.bits(), + skip: false, + }, + CellData { + symbol: "🦀".into(), // emoji, wide grapheme cluster + fg: color_to_u32(Color::Yellow), + bg: color_to_u32(Color::Magenta), + modifier: Modifier::empty().bits(), + skip: false, + }, + ], + width: 3, + height: 2, + cursor: Some(CursorState { + x: 0, + y: 0, + visible: true, + }), + }; + let msg = ServerMessage::Frame(frame.clone()); + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ServerMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn server_shutdown_roundtrip() { + let msg = ServerMessage::ServerShutdown { + reason: Some("updating".to_owned()), + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ServerMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn server_notify_roundtrip() { + let msg = ServerMessage::Notify { + kind: NotifyKind::Sound, + message: "agent done".to_owned(), + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ServerMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn server_clipboard_roundtrip() { + let msg = ServerMessage::Clipboard { + data: "dGVzdA==".to_owned(), // base64 "test" + }; + let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let (decoded, _): (ServerMessage, _) = + bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap(); + assert_eq!(msg, decoded); + } + + // ---- Framing ---- + + #[test] + fn framing_small_message_roundtrip() { + let msg = ClientMessage::Hello { + version: 1, + cols: 80, + rows: 24, + }; + let mut buf = Vec::new(); + write_message(&mut buf, &msg).unwrap(); + let decoded: ClientMessage = read_message(&mut buf.as_slice(), MAX_FRAME_SIZE).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn framing_large_payload_roundtrip() { + // Create a Frame message that is ≥128 KB. + // Use a large frame with verbose cell data to exceed 128 KB after bincode encoding. + // 200×50 = 10000 cells. With varied symbols and styles, this should easily exceed 128 KB. + let width: u16 = 200; + let height: u16 = 50; + let cells: Vec<CellData> = (0..(width as usize) * (height as usize)) + .map(|i| CellData { + symbol: if i % 256 < 32 { + " ".to_owned() + } else { + format!("{:03}", i % 1000) + }, + fg: color_to_u32(Color::Rgb((i % 256) as u8, ((i / 256) % 256) as u8, 128)), + bg: color_to_u32(Color::Indexed((i % 256) as u8)), + modifier: ((i % 16) as u16), + skip: i % 100 == 0, + }) + .collect(); + + let frame = FrameData { + cells, + width, + height, + cursor: Some(CursorState { + x: 10, + y: 5, + visible: true, + }), + }; + let msg = ServerMessage::Frame(frame); + + let mut buf = Vec::new(); + write_message(&mut buf, &msg).unwrap(); + // Verify the payload is at least 128 KB + assert!( + buf.len() >= 128 * 1024, + "framed payload should be >= 128 KB, got {} bytes", + buf.len() + ); + + let decoded: ServerMessage = read_message(&mut buf.as_slice(), MAX_FRAME_SIZE).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn framing_multiple_messages_sequential() { + // Write 100+ messages of varying types and read them back. + let mut buf = Vec::new(); + let mut expected = Vec::new(); + + for i in 0..150u32 { + let msg = match i % 4 { + 0 => ClientMessage::Hello { + version: 1, + cols: (80 + (i % 40) as u16), + rows: (24 + (i % 20) as u16), + }, + 1 => ClientMessage::Input { + data: vec![(i % 256) as u8; (i as usize % 50) + 1], + }, + 2 => ClientMessage::Resize { + cols: (100 + (i % 30) as u16), + rows: (30 + (i % 10) as u16), + }, + 3 => ClientMessage::Detach, + _ => unreachable!(), + }; + write_message(&mut buf, &msg).unwrap(); + expected.push(msg); + } + + let mut cursor = buf.as_slice(); + for expected_msg in &expected { + let decoded: ClientMessage = read_message(&mut cursor, MAX_FRAME_SIZE).unwrap(); + assert_eq!(*expected_msg, decoded); + } + } + + #[test] + fn framing_oversized_rejected_without_panic() { + // Craft a frame with a huge length prefix (4 GB claim). + let mut buf: Vec<u8> = (u32::MAX).to_le_bytes().to_vec(); + // Add a few garbage bytes after the length prefix. + buf.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF]); + + let result: Result<ClientMessage, FramingError> = + read_message(&mut buf.as_slice(), MAX_FRAME_SIZE); + match result { + Err(FramingError::Oversized { claimed, max }) => { + assert_eq!(claimed, u32::MAX as usize); + assert_eq!(max, MAX_FRAME_SIZE); + } + other => panic!("expected Oversized error, got: {other:?}"), + } + } + + #[test] + fn framing_malformed_payload_rejected_without_panic() { + // Valid length prefix pointing to garbage data. + let payload = vec![0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02]; + let mut buf = (payload.len() as u32).to_le_bytes().to_vec(); + buf.extend_from_slice(&payload); + + let result: Result<ClientMessage, FramingError> = + read_message(&mut buf.as_slice(), MAX_FRAME_SIZE); + assert!(result.is_err(), "malformed payload should be rejected"); + match result { + Err(FramingError::Bincode(_)) => {} // expected + other => panic!("expected Bincode error, got: {other:?}"), + } + } + + #[test] + fn framing_truncated_stream_returns_unexpected_eof() { + // Write a length prefix claiming 100 bytes, but only provide 4. + let mut buf: Vec<u8> = 100u32.to_le_bytes().to_vec(); + buf.extend_from_slice(&[0xAA, 0xBB, 0xCC, 0xDD]); + + let result: Result<ClientMessage, FramingError> = + read_message(&mut buf.as_slice(), MAX_FRAME_SIZE); + match result { + Err(FramingError::UnexpectedEof) => {} + other => panic!("expected UnexpectedEof, got: {other:?}"), + } + } + + #[test] + fn framing_zero_length_message() { + // A 1-byte message (smallest possible valid bincode payload). + // Actually, let's test with the smallest real message: Detach. + let msg = ClientMessage::Detach; + let mut buf = Vec::new(); + write_message(&mut buf, &msg).unwrap(); + + // Verify the length prefix is correct + let len = u32::from_le_bytes(buf[..4].try_into().unwrap()) as usize; + assert_eq!( + len, + buf.len() - 4, + "length prefix should match payload size" + ); + + let decoded: ClientMessage = read_message(&mut buf.as_slice(), MAX_FRAME_SIZE).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn framing_partial_read_reassembly() { + // Simulate partial reads by using a reader that yields small chunks. + let msg = ClientMessage::Input { + data: vec![42; 500], // 500-byte input payload + }; + let mut full_buf = Vec::new(); + write_message(&mut full_buf, &msg).unwrap(); + + // Wrap in a chunked reader that only yields 7 bytes at a time. + let mut chunked = ChunkedReader::new(full_buf, 7); + let decoded: ClientMessage = read_message(&mut chunked, MAX_FRAME_SIZE).unwrap(); + assert_eq!(msg, decoded); + } + + // ---- Version negotiation ---- + + #[test] + fn version_compatible() { + assert_eq!(check_client_version(1), VersionCheck::Compatible); + } + + #[test] + fn version_older_client_rejected() { + match check_client_version(1) { + VersionCheck::Compatible => {} // current version + _ => panic!("current version should be compatible"), + } + // If we imagine a future server version 2, a v1 client would be rejected. + // We test this by checking the logic directly. + let result = check_client_version(0); + assert!( + matches!(result, VersionCheck::Incompatible(_)), + "version 0 should be incompatible" + ); + } + + #[test] + fn version_newer_client_rejected() { + let result = check_client_version(PROTOCOL_VERSION + 1); + assert!(matches!(result, VersionCheck::Incompatible(_))); + if let VersionCheck::Incompatible(msg) = result { + assert!(msg.contains("newer"), "error should mention newer version"); + } + } + + // ---- Pre-persistence client rejection ---- + + #[test] + fn prepersistence_version_zero_rejected() { + let result = check_client_version(0); + match result { + VersionCheck::Incompatible(msg) => { + assert!( + msg.contains("pre-persistence"), + "error should mention pre-persistence: {msg}" + ); + } + _ => panic!("version 0 should be rejected as incompatible"), + } + } + + #[test] + fn prepersistence_version_zero_welcome_has_error() { + // Simulating what the server would send to a v0 client. + let check = check_client_version(0); + let response = match check { + VersionCheck::Compatible => ServerMessage::Welcome { + version: PROTOCOL_VERSION, + error: None, + }, + VersionCheck::Incompatible(reason) => ServerMessage::Welcome { + version: PROTOCOL_VERSION, + error: Some(reason), + }, + }; + + match response { + ServerMessage::Welcome { error: Some(_), .. } => {} + other => panic!("expected Welcome with error, got: {other:?}"), + } + } + + // ---- Malformed/oversized input ---- + + #[test] + fn oversized_frame_does_not_panic() { + // Claim 4GB payload — should return Oversized error, not panic. + let mut buf: Vec<u8> = 0xFFC00000u32.to_le_bytes().to_vec(); // ~4 GB claim + buf.extend_from_slice(&[0; 8]); + + let result: Result<ClientMessage, FramingError> = + read_message(&mut buf.as_slice(), MAX_FRAME_SIZE); + assert!(result.is_err()); + // Did not panic — test passing is proof. + } + + #[test] + fn malformed_frame_does_not_panic() { + // Random garbage bytes after a valid-ish length prefix. + let garbage: Vec<u8> = (0..200).map(|i| (i ^ 0xAA) as u8).collect(); + let mut buf = (garbage.len() as u32).to_le_bytes().to_vec(); + buf.extend_from_slice(&garbage); + + let result: Result<ClientMessage, FramingError> = + read_message(&mut buf.as_slice(), MAX_FRAME_SIZE); + assert!(result.is_err()); + // Did not panic. + } + + #[test] + fn oversized_input_rejected_custom_max() { + // Verify a custom (small) max_frame_size is enforced. + let msg = ClientMessage::Input { + data: vec![0x41; 1000], + }; + let mut buf = Vec::new(); + write_message(&mut buf, &msg).unwrap(); + + let result: Result<ClientMessage, FramingError> = read_message(&mut buf.as_slice(), 64); + // The actual bincode payload for 1000 bytes of input will be > 64 bytes. + assert!( + matches!(result, Err(FramingError::Oversized { .. })), + "expected Oversized with small max_frame_size" + ); + } + + // ---- FrameData ↔ ratatui Buffer conversion ---- + + #[test] + fn frame_data_roundtrip_through_ratatui_buffer() { + let area = ratatui::layout::Rect::new(0, 0, 5, 3); + let mut buffer = ratatui::buffer::Buffer::filled(area, ratatui::buffer::Cell::new(" ")); + + // Write some styled content. + buffer.cell_mut((0, 0)).unwrap().set_symbol("H"); + buffer.cell_mut((0, 0)).unwrap().fg = Color::Red; + buffer.cell_mut((0, 0)).unwrap().modifier = Modifier::BOLD; + + buffer.cell_mut((1, 0)).unwrap().set_symbol("i"); + buffer.cell_mut((1, 0)).unwrap().fg = Color::Green; + buffer.cell_mut((1, 0)).unwrap().modifier = Modifier::ITALIC; + + buffer.cell_mut((2, 0)).unwrap().set_symbol("!"); + buffer.cell_mut((2, 0)).unwrap().fg = Color::Rgb(255, 128, 0); + buffer.cell_mut((2, 0)).unwrap().bg = Color::Indexed(220); + + let cursor = CursorState { + x: 1, + y: 0, + visible: true, + }; + let frame = FrameData::from_ratatui_buffer(&buffer, Some(cursor.clone())); + + // Verify frame dimensions. + assert_eq!(frame.width, 5); + assert_eq!(frame.height, 3); + assert_eq!(frame.cells.len(), 15); + assert_eq!(frame.cursor, Some(cursor)); + + // Verify specific cells survived the conversion. + assert_eq!(frame.cells[0].symbol, "H"); + assert_eq!(frame.cells[0].fg, color_to_u32(Color::Red)); + assert_eq!(frame.cells[0].modifier, Modifier::BOLD.bits()); + + assert_eq!(frame.cells[1].symbol, "i"); + assert_eq!(frame.cells[1].fg, color_to_u32(Color::Green)); + assert_eq!(frame.cells[1].modifier, Modifier::ITALIC.bits()); + + assert_eq!(frame.cells[2].symbol, "!"); + assert_eq!(frame.cells[2].fg, color_to_u32(Color::Rgb(255, 128, 0))); + assert_eq!(frame.cells[2].bg, color_to_u32(Color::Indexed(220))); + + // Convert back to ratatui buffer and compare. + let restored = frame.to_ratatui_buffer().expect("should reconstruct"); + assert_eq!(restored.area, area); + assert_eq!(restored.cell((0, 0)).unwrap().symbol(), "H"); + assert_eq!(restored.cell((0, 0)).unwrap().fg, Color::Red); + assert_eq!(restored.cell((0, 0)).unwrap().modifier, Modifier::BOLD); + assert_eq!(restored.cell((1, 0)).unwrap().symbol(), "i"); + assert_eq!(restored.cell((2, 0)).unwrap().symbol(), "!"); + assert_eq!(restored.cell((2, 0)).unwrap().fg, Color::Rgb(255, 128, 0)); + } + + #[test] + fn frame_data_rejects_mismatched_cell_count() { + let frame = FrameData { + cells: vec![ + CellData { + symbol: "X".into(), + fg: 0, + bg: 0, + modifier: 0, + skip: false, + }; + 5 + ], // 5 cells but 3×2 = 6 expected + width: 3, + height: 2, + cursor: None, + }; + assert!(frame.to_ratatui_buffer().is_none()); + } + + // ---- Color conversion coverage ---- + + #[test] + fn color_roundtrip_all_named_colors() { + let named = [ + Color::Reset, + Color::Black, + Color::Red, + Color::Green, + Color::Yellow, + Color::Blue, + Color::Magenta, + Color::Cyan, + Color::Gray, + Color::DarkGray, + Color::LightRed, + Color::LightGreen, + Color::LightYellow, + Color::LightBlue, + Color::LightMagenta, + Color::LightCyan, + Color::White, + ]; + for c in named { + assert_eq!( + u32_to_color(color_to_u32(c)), + c, + "roundtrip failed for {c:?}" + ); + } + } + + #[test] + fn color_roundtrip_indexed() { + for i in 0..=255u8 { + let c = Color::Indexed(i); + assert_eq!( + u32_to_color(color_to_u32(c)), + c, + "roundtrip failed for Indexed({i})" + ); + } + } + + #[test] + fn color_roundtrip_rgb() { + let c = Color::Rgb(0xAB, 0xCD, 0xEF); + assert_eq!(u32_to_color(color_to_u32(c)), c); + + let c = Color::Rgb(0, 0, 0); + assert_eq!(u32_to_color(color_to_u32(c)), c); + + let c = Color::Rgb(255, 255, 255); + assert_eq!(u32_to_color(color_to_u32(c)), c); + } + + // ---- Modifier conversion ---- + + #[test] + fn modifier_roundtrip() { + let all_mods = [ + Modifier::BOLD, + Modifier::ITALIC, + Modifier::REVERSED, + Modifier::UNDERLINED, + Modifier::DIM, + Modifier::SLOW_BLINK, + Modifier::CROSSED_OUT, + Modifier::BOLD | Modifier::ITALIC, + Modifier::BOLD | Modifier::UNDERLINED | Modifier::REVERSED, + Modifier::empty(), + ]; + for m in all_mods { + assert_eq!( + u16_to_modifier(modifier_to_u16(m)), + m, + "roundtrip failed for {m:?}" + ); + } + } + + #[test] + fn read_message_rejects_trailing_bytes() { + // Encode a valid message, then append an extra byte after it. + let msg = ClientMessage::Detach; + let mut payload = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap(); + let original_len = payload.len(); + payload.push(0xDE); // trailing garbage + + // Frame it with the inflated length (original + 1). + let mut buf = (payload.len() as u32).to_le_bytes().to_vec(); + buf.extend_from_slice(&payload); + + let result: Result<ClientMessage, FramingError> = + read_message(&mut buf.as_slice(), MAX_FRAME_SIZE); + match result { + Err(FramingError::Bincode(msg)) => { + assert!( + msg.contains("trailing bytes"), + "error should mention trailing bytes: {msg}" + ); + assert!( + msg.contains(&format!("decoded {original_len}")), + "error should mention decoded byte count: {msg}" + ); + } + other => panic!("expected Bincode error about trailing bytes, got: {other:?}"), + } + } + + #[test] + fn read_message_accepts_exact_payload() { + // A normally-framed message should decode without error. + let msg = ClientMessage::Hello { + version: 1, + cols: 80, + rows: 24, + }; + let mut buf = Vec::new(); + write_message(&mut buf, &msg).unwrap(); + let decoded: ClientMessage = read_message(&mut buf.as_slice(), MAX_FRAME_SIZE).unwrap(); + assert_eq!(msg, decoded); + } + + #[test] + fn write_message_rejects_oversized_payload() { + // We can't easily create a message that exceeds u32::MAX in a test, + // but we can verify the check exists by testing that normal messages + // have lengths well within the limit and the function doesn't fail. + let msg = ClientMessage::Detach; + let mut buf = Vec::new(); + assert!(write_message(&mut buf, &msg).is_ok()); + } + + // ---- Unix socketpair integration test ---- + + #[test] + fn framing_over_unix_socketpair() { + use std::os::unix::net::UnixStream; + + let (mut a, mut b) = UnixStream::pair().expect("socketpair"); + + let messages = vec![ + ClientMessage::Hello { + version: 1, + cols: 200, + rows: 60, + }, + ClientMessage::Input { + data: b"hello world".to_vec(), + }, + ClientMessage::Resize { + cols: 100, + rows: 30, + }, + ClientMessage::Detach, + ]; + + // Set non-blocking so we can write and read in the same test. + a.set_nonblocking(false).unwrap(); + b.set_nonblocking(false).unwrap(); + + for msg in &messages { + write_message(&mut a, msg).unwrap(); + } + + for expected in &messages { + let decoded: ClientMessage = read_message(&mut b, MAX_FRAME_SIZE).unwrap(); + assert_eq!(*expected, decoded); + } + } + + // ---- Helper: chunked reader for simulating partial reads ---- + + /// A `Read` wrapper that yields at most `chunk_size` bytes per `read()` call, + /// simulating partial reads on a real socket. + struct ChunkedReader { + data: Vec<u8>, + pos: usize, + chunk_size: usize, + } + + impl ChunkedReader { + fn new(data: Vec<u8>, chunk_size: usize) -> Self { + Self { + data, + pos: 0, + chunk_size, + } + } + } + + impl Read for ChunkedReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { + if self.pos >= self.data.len() { + return Ok(0); + } + let remaining = self.data.len() - self.pos; + let to_read = buf.len().min(remaining).min(self.chunk_size); + buf[..to_read].copy_from_slice(&self.data[self.pos..self.pos + to_read]); + self.pos += to_read; + Ok(to_read) + } + } +} |