feat: add persistent server-client sessions

   Make herdr persistent by default.

   Launching herdr now starts or reattaches to a background session
   server. Clients can detach and reattach while panes and agent
   processes keep running. Session mode now supports multi-client attach,
   auto-detect startup, and a thin-client/headless-server split.

   This also refactors the large app, ui, input, pane, config,
   workspace, and persistence modules into smaller focused submodules,
   while preserving behavior and colocating tests with the code they
   exercise.

   Upgrade notes:
   - persistence mode is now the default
   - in-app quit detaches the current client instead of stopping the server
   - use `herdr server stop` to stop the background session
   - use `--no-session` for the old single-process behavior
   - default socket paths now live under the config directory

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)
+        }
+    }
+}