//! 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 /// How often the idle headless loop wakes to poll the std UnixListener for new /// client connections. /// /// The listener is non-blocking and not integrated into `tokio::select!`, so /// a low-frequency wake is required to notice new thin-client attaches while /// otherwise idle. Keep this much slower than the old resize-poll cadence to /// avoid reintroducing the idle CPU spin. const CLIENT_ACCEPT_POLL_INTERVAL: Duration = Duration::from_millis(250); // --------------------------------------------------------------------------- // 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, /// Last frame sent to this client. Used to skip identical frame sends. last_frame: Option, /// Channel for sending framed ServerMessage data to the client writer thread. writer: Option>>, } // --------------------------------------------------------------------------- // 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, } 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 { 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, { 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 { self.inner.get_cursor_position() } fn set_cursor_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 { self.inner.size() } fn window_size(&mut self) -> Result { 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) { 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, next_client_id: u64, /// The client currently driving the shared pane runtime size and theme. foreground_client_id: Option, /// 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, /// Channel for receiving server events from client connection threads. server_event_rx: mpsc::Receiver, /// Sender for server events (cloned for each client thread). server_event_tx: mpsc::Sender, } 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 { 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<()> { crate::logging::startup("server"); // 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.open_settings_from_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_headless_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_headless_loop_deadline(now, needs_render) .map(|deadline| deadline.min(now + CLIENT_ACCEPT_POLL_INTERVAL)) .or(Some(now + CLIENT_ACCEPT_POLL_INTERVAL)); 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); // Shared runtime size changes affect pane wrapping and foreground-driven // rendering semantics. Force one fresh frame to every remaining client // even if the next rendered buffer compares equal to its cached frame. for client in self.clients.values_mut() { client.last_frame = None; } } 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(), }); } } let toast_msg = if self.app.state.toast.is_some() && self.app.state.toast != toast_before { self.app .state .toast .as_ref() .map(|toast| format!("{}: {}", toast.title, toast.context)) } else if matches!( self.app.state.toast_config.delivery, crate::config::ToastDelivery::Terminal ) { self.app .state .workspaces .iter() .enumerate() .find_map(|(ws_idx, ws)| { ws.tabs.iter().find_map(|tab| { tab.panes.get(&pane_id_val).and_then(|pane| { pane.effective_agent_label().and_then(|agent_label| { crate::app::actions::notification_toast_for_state_change( is_active_tab, prev_state, state_val, ) .map(|kind| { let event_text = match kind { crate::app::state::ToastKind::NeedsAttention => { "needs attention" } crate::app::state::ToastKind::Finished => { "finished" } crate::app::state::ToastKind::UpdateInstalled => { "updated" } }; format!( "{} {}: {}", agent_label, event_text, crate::app::actions::notification_context( ws, ws_idx, pane_id_val, ) ) }) }) }) }) }) } else { None }; if let Some(msg) = toast_msg { 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(), }); } } let toast_msg = if self.app.state.toast.is_some() && self.app.state.toast != toast_before { self.app .state .toast .as_ref() .map(|toast| format!("{}: {}", toast.title, toast.context)) } else if matches!( self.app.state.toast_config.delivery, crate::config::ToastDelivery::Terminal ) { self.app .state .workspaces .iter() .enumerate() .find_map(|(ws_idx, ws)| { ws.tabs.iter().find_map(|tab| { tab.panes.get(&pane_id_val).and_then(|pane| { pane.effective_agent_label().and_then(|agent_label| { crate::app::actions::notification_toast_for_state_change( is_active_tab, prev_hook_state, hook_state_val, ) .map(|kind| { let event_text = match kind { crate::app::state::ToastKind::NeedsAttention => { "needs attention" } crate::app::state::ToastKind::Finished => { "finished" } crate::app::state::ToastKind::UpdateInstalled => { "updated" } }; format!( "{} {}: {}", agent_label, event_text, crate::app::actions::notification_context( ws, ws_idx, pane_id_val, ) ) }) }) }) }) }) } else { None }; if let Some(msg) = toast_msg { 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(); let version = version.clone(); self.app.handle_internal_event(ev); let toast_msg = if self.app.state.toast.is_some() && self.app.state.toast != toast_before { self.app .state .toast .as_ref() .map(|toast| format!("{}: {}", toast.title, toast.context)) } else if matches!( self.app.state.toast_config.delivery, crate::config::ToastDelivery::Terminal ) { Some(format!( "v{version} available: detach, then run `herdr update`" )) } else { None }; if let Some(msg) = toast_msg { 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, 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 = 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, last_frame: None, 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"); if let Some(client) = self.clients.get_mut(&client_id) { // Ensure the next render after client input is delivered even if the // current frame buffer still compares equal. Input can change cursor or // PTY state asynchronously, and thin clients should not stall waiting for // a post-input frame behind identical-frame dedupe. client.last_frame = None; } 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, )> = 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(); let forwarded_toast_from_state = 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, }); true } else { false } } else { false }; // 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" ); if !forwarded_toast_from_state && matches!( self.app.state.toast_config.delivery, crate::config::ToastDelivery::Terminal ) { if let Some(kind) = crate::app::actions::notification_toast_for_state_change( is_active_tab, prev_state, new_state, ) { if let Some(agent_label) = pane_after.effective_agent_label() { let event_text = match kind { crate::app::state::ToastKind::NeedsAttention => "needs attention", crate::app::state::ToastKind::Finished => "finished", crate::app::state::ToastKind::UpdateInstalled => "updated", }; let msg_text = format!( "{} {}: {}", agent_label, event_text, crate::app::actions::notification_context( &self.app.state.workspaces[*ws_idx], *ws_idx, *pane_id, ) ); self.send_to_all_clients(ServerMessage::Notify { kind: protocol::NotifyKind::Toast, message: msg_text, }); } } } // 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), bool)> = self .clients .iter() .filter(|(_, client)| client.writer.is_some()) .map(|(&client_id, client)| { ( client_id, client.terminal_size, 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 = Vec::new(); for (client_id, (cols, rows), 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 frame = FrameData::from_ratatui_buffer(&buffer, cursor); let Some(client) = self.clients.get_mut(&client_id) else { continue; }; let Some(writer) = client.writer.as_ref().cloned() else { continue; }; if client.last_frame.as_ref() == Some(&frame) { continue; } let message = ServerMessage::Frame(frame.clone()); 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); continue; } client.last_frame = Some(frame); } 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_headless_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(app::HEADLESS_ANIMATION_TICK_STEP); self.app.next_animation_tick = Some(now + app::HEADLESS_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_headless_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>, }, /// A client sent an input message. ClientInput { client_id: u64, data: Vec }, /// 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, should_quit: &Arc, ) -> 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::>(); // 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>, should_quit: &Arc, ) { 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, should_quit: &Arc, ) -> 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, server_event_tx: mpsc::Sender) { 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) { 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)); crate::logging::shutdown("server"); 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!( "hh-{}-{}", 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 _ = fs::remove_file(&socket_path); 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) -> 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) -> 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 = PathBuf::from(format!( "/tmp/hs-{}-{}", 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 _listener = UnixListener::bind(&socket_path).expect("bind stale socket"); } // The listener scope ended, so the socket is now stale. let deadline = std::time::Instant::now() + Duration::from_secs(1); while std::time::Instant::now() < deadline { if std::os::unix::net::UnixStream::connect(&socket_path).is_err() { break; } std::thread::sleep(Duration::from_millis(10)); } // 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 = PathBuf::from(format!( "/tmp/hl-{}-{}", 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, last_frame: None, 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, last_frame: None, 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, last_frame: None, writer: Some(desktop_tx), }, ); server.clients.insert( 2, ClientConnection { terminal_size: (80, 24), host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), last_activity: 2, last_frame: None, 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 render_and_stream_skips_identical_frame_sends() { 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 (client_tx, client_rx) = std::sync::mpsc::channel(); server.clients.insert( 1, ClientConnection { terminal_size: (80, 24), host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), last_activity: 1, last_frame: None, writer: Some(client_tx), }, ); server.foreground_client_id = Some(1); server.sync_foreground_client_state(); server.resize_shared_runtime_to_effective_size(); server.render_and_stream(); let first = client_rx.recv_timeout(Duration::from_millis(100)); assert!(first.is_ok(), "expected first frame to be sent"); server.render_and_stream(); assert!( client_rx.recv_timeout(Duration::from_millis(50)).is_err(), "identical frame should not be sent twice" ); } #[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, last_frame: None, writer: Some(background_tx), }, ); server.clients.insert( 2, ClientConnection { terminal_size: (80, 24), host_terminal_theme: crate::terminal_theme::TerminalTheme::default(), last_activity: 2, last_frame: None, 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 = 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 ") ); } }