path: root/src/server/headless.rs

//! Headless server mode — runs the herdr event loop without a real terminal.
//!
//! The server:
//! - Does not enter raw mode or read stdin
//! - Creates and listens on both `herdr.sock` (existing JSON API) and
//!   `herdr-client.sock` (new binary protocol)
//! - Initializes AppState and all PTYs from session restore or fresh state
//! - Runs the main event loop (drain events, drain API requests, scheduled tasks)
//! - Renders to a virtual ratatui Buffer in memory
//! - Accepts client connections on the client socket
//! - Streams frames to connected clients after each render
//! - Routes client input events through the existing input pipeline
//! - Continues running after client disconnect
//! - Handles stale socket cleanup, explicit server stop, minimum terminal size,
//!   and pane spawn failure during restore

use std::collections::HashMap;
use std::fs;
use std::io::{self, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use ratatui::backend::{Backend, ClearType, TestBackend, WindowSize};
use ratatui::layout::{Position, Rect, Size};
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};

use base64::Engine;

use crate::api;
use crate::app;
use crate::app::state::AppState;
use crate::config;
use crate::events::AppEvent;
use crate::server::protocol::{
    self, ClientMessage, CursorState, FrameData, ServerMessage, MAX_FRAME_SIZE, PROTOCOL_VERSION,
};

// ---------------------------------------------------------------------------
// Loop event enum for the headless server event loop
// ---------------------------------------------------------------------------

/// Events that the headless server event loop can process.
enum LoopEvent {
    Timer,
    Internal(AppEvent),
    Api(api::ApiRequestMessage),
    ServerEvent(ServerEvent),
    RenderRequested,
}

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/// Default shared runtime size (columns, rows) when no clients are attached.
const MIN_COLS: u16 = 80;
const MIN_ROWS: u16 = 24;

/// Minimum accepted attached client size.
///
/// Narrow observers must be allowed to drive narrow renders, otherwise the
/// server wraps pane content against a wider width and the client sees the
/// right edge clipped.
const MIN_CLIENT_COLS: u16 = 1;
const MIN_CLIENT_ROWS: u16 = 1;

/// Legacy environment variable for overriding the client socket path.
///
/// Contractual override behavior for auto-detect uses `HERDR_SOCKET_PATH`.
/// This variable is kept as a fallback for callers that explicitly need a
/// client-only override when `HERDR_SOCKET_PATH` is not set.
const CLIENT_SOCKET_PATH_ENV_VAR: &str = "HERDR_CLIENT_SOCKET_PATH";

/// Socket permission mode (owner read/write only).
const SOCKET_PERMISSION_MODE: u32 = 0o600;

/// Timeout for in-flight API requests during shutdown.
#[allow(dead_code)]
const SHUTDOWN_API_TIMEOUT: Duration = Duration::from_secs(5);

/// How long to wait for a client handshake before closing the connection.
/// Set to 4 seconds (rather than 5) to guarantee the connection is closed
/// within the 5-second deadline, even with
/// OS timer slack, thread scheduling, and cleanup overhead.
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(4);

/// Maximum input payload size (bytes) for a single `ClientMessage::Input`.
const MAX_INPUT_PAYLOAD: usize = 1024 * 1024; // 1 MB

/// How often the idle headless loop wakes to poll the std UnixListener for new
/// client connections.
///
/// The listener is non-blocking and not integrated into `tokio::select!`, so
/// a low-frequency wake is required to notice new thin-client attaches while
/// otherwise idle. Keep this much slower than the old resize-poll cadence to
/// avoid reintroducing the idle CPU spin.
const CLIENT_ACCEPT_POLL_INTERVAL: Duration = Duration::from_millis(250);

// ---------------------------------------------------------------------------
// Socket path helpers
// ---------------------------------------------------------------------------

/// Returns the path for the client protocol socket.
///
/// Contract-aligned override behavior:
/// 1. If `HERDR_SOCKET_PATH` is set, derive the client socket path from it by
///    inserting `-client` before `.sock` (e.g. `herdr.sock` -> `herdr-client.sock`).
///    This keeps JSON API and client socket overrides consistent.
/// 2. Otherwise, honor `HERDR_CLIENT_SOCKET_PATH` (legacy/testing fallback).
/// 3. Otherwise, use the app config directory.
pub fn client_socket_path() -> PathBuf {
    client_socket_path_from_overrides(
        std::env::var(api::SOCKET_PATH_ENV_VAR).ok().as_deref(),
        std::env::var(CLIENT_SOCKET_PATH_ENV_VAR).ok().as_deref(),
    )
}

fn client_socket_path_from_overrides(
    api_socket_override: Option<&str>,
    client_socket_override: Option<&str>,
) -> PathBuf {
    if let Some(api_socket_override) = api_socket_override {
        return derive_client_socket_from_api_socket(Path::new(api_socket_override));
    }

    if let Some(client_socket_override) = client_socket_override {
        return PathBuf::from(client_socket_override);
    }

    config::config_dir().join("herdr-client.sock")
}

fn derive_client_socket_from_api_socket(api_socket_path: &Path) -> PathBuf {
    let stem = api_socket_path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("herdr");
    let parent = api_socket_path.parent().unwrap_or_else(|| Path::new(""));

    if api_socket_path
        .extension()
        .and_then(|ext| ext.to_str())
        .is_some_and(|ext| ext == "sock")
    {
        return parent.join(format!("{stem}-client.sock"));
    }

    parent.join(format!("{stem}-client.sock"))
}

/// Clamp client-reported terminal dimensions to a minimum viable size.
pub fn clamp_terminal_size(cols: u16, rows: u16) -> (u16, u16) {
    let clamped_cols = cols.max(MIN_CLIENT_COLS);
    let clamped_rows = rows.max(MIN_CLIENT_ROWS);
    (clamped_cols, clamped_rows)
}

// ---------------------------------------------------------------------------
// Connected client state
// ---------------------------------------------------------------------------

/// A connected client tracked by the server.
struct ClientConnection {
    /// The client's terminal size (after clamping).
    terminal_size: (u16, u16),
    /// Last known host terminal default colors for this client.
    host_terminal_theme: crate::terminal_theme::TerminalTheme,
    /// Monotonic activity stamp used to choose the fallback foreground client.
    last_activity: u64,
    /// Last frame sent to this client. Used to skip identical frame sends.
    last_frame: Option<FrameData>,
    /// 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<()> {
        crate::logging::startup("server");

        // Register SIGINT handler for graceful shutdown.
        let should_quit = self.should_quit.clone();
        let quit_notify = self.server_event_tx.clone();
        ctrlc_handler(should_quit, quit_notify);

        // No input_rx needed — server doesn't read stdin.
        // We use None for input_rx so the event loop doesn't try to read from stdin.
        self.app.input_rx = None;

        let mut needs_render = true;

        loop {
            // If shutdown has been initiated, complete it and exit.
            if self.shutting_down {
                self.complete_shutdown()?;
                break;
            }

            // Check if we should start shutting down.
            if self.app.state.should_quit || self.should_quit.load(Ordering::Acquire) {
                self.initiate_shutdown();
                continue;
            }

            // 1. Check render_dirty flag from PTY reader tasks.
            if self.app.render_dirty.load(Ordering::Acquire) {
                needs_render = true;
            }

            // 2. Drain internal events.
            if self.drain_internal_events_with_forwarding() {
                needs_render = true;
            }

            // 3. Drain API requests.
            if self.drain_api_requests_with_shutdown_check() {
                needs_render = true;
            }

            self.app.sync_focus_events();
            self.app.sync_session_save_schedule();

            // 4. Accept new client connections.
            self.accept_client_connections()?;

            // 5. Drain server events from client threads.
            if self.drain_server_events() {
                needs_render = true;
            }

            // 6. Handle scheduled tasks.
            let now = Instant::now();
            if self.handle_scheduled_tasks_headless(now) {
                needs_render = true;
            }

            // Handle deferred requests.
            if self.app.state.request_complete_onboarding {
                self.app.state.request_complete_onboarding = false;
                self.app.open_settings_from_onboarding();
                needs_render = true;
            }

            if self.app.state.request_new_workspace {
                self.app.state.request_new_workspace = false;
                self.app.create_workspace();
                needs_render = true;
            }

            if self.app.state.request_new_tab {
                self.app.state.request_new_tab = false;
                self.app.create_tab();
                needs_render = true;
            }

            if self.app.state.request_reload_keybinds {
                self.app.state.request_reload_keybinds = false;
                self.app.reload_keybinds();
                needs_render = true;
            }

            self.app.sync_headless_animation_timer(now);

            // 7. Render virtually and stream frames.
            if needs_render && self.app.can_render_now(now) {
                self.app.render_dirty.swap(false, Ordering::AcqRel);
                self.render_and_stream();
                self.app.last_render_at = Some(now);
                needs_render = false;
                continue;
            }

            // 8. Wait for next event.
            let next_deadline = self
                .app
                .next_headless_loop_deadline(now, needs_render)
                .map(|deadline| deadline.min(now + CLIENT_ACCEPT_POLL_INTERVAL))
                .or(Some(now + CLIENT_ACCEPT_POLL_INTERVAL));
            let event = {
                tokio::select! {
                    maybe_api = self.app.api_rx.recv() => match maybe_api {
                        Some(msg) => LoopEvent::Api(msg),
                        None => LoopEvent::Timer,
                    },
                    maybe_ev = self.app.event_rx.recv() => match maybe_ev {
                        Some(ev) => LoopEvent::Internal(ev),
                        None => LoopEvent::Timer,
                    },
                    maybe_server_ev = self.server_event_rx.recv() => match maybe_server_ev {
                        Some(ev) => LoopEvent::ServerEvent(ev),
                        None => LoopEvent::Timer,
                    },
                    _ = sleep_until_or_pending(next_deadline) => LoopEvent::Timer,
                    _ = self.app.render_notify.notified() => LoopEvent::RenderRequested,
                }
            };

            match event {
                LoopEvent::Timer => {}
                LoopEvent::Internal(ev) => {
                    if self.handle_internal_event_with_forwarding(ev) {
                        needs_render = true;
                    }
                }
                LoopEvent::Api(msg) => {
                    if self.handle_api_request_with_shutdown_check(msg) {
                        needs_render = true;
                    }
                }
                LoopEvent::ServerEvent(ev) => {
                    if self.handle_server_event(ev) {
                        needs_render = true;
                    }
                }
                LoopEvent::RenderRequested => {
                    if self.app.render_dirty.load(Ordering::Acquire) {
                        needs_render = true;
                    }
                }
            }
        }

        // Save session on exit.
        if !self.app.no_session {
            self.app.save_session_now();
        }

        info!("headless server exiting");
        Ok(())
    }

    fn allocate_activity_stamp(&mut self) -> u64 {
        let stamp = self.next_activity_stamp;
        self.next_activity_stamp = self.next_activity_stamp.saturating_add(1);
        stamp
    }

    fn resize_shared_runtime_to_effective_size(&mut self) {
        if self.foreground_client_id.is_none() {
            return;
        }
        let (cols, rows) = self.effective_size;
        let area = Rect::new(0, 0, cols, rows);
        crate::ui::compute_view(&mut self.app.state, area);

        // Shared runtime size changes affect pane wrapping and foreground-driven
        // rendering semantics. Force one fresh frame to every remaining client
        // even if the next rendered buffer compares equal to its cached frame.
        for client in self.clients.values_mut() {
            client.last_frame = None;
        }
    }

    fn sync_foreground_client_state(&mut self) {
        let Some(client_id) = self.foreground_client_id else {
            self.effective_size = (MIN_COLS, MIN_ROWS);
            return;
        };
        let Some(client) = self.clients.get(&client_id) else {
            self.foreground_client_id = None;
            self.effective_size = (MIN_COLS, MIN_ROWS);
            return;
        };

        self.effective_size = client.terminal_size;
        if !client.host_terminal_theme.is_empty() {
            self.app.set_host_terminal_theme(client.host_terminal_theme);
        }
    }

    fn promote_client_to_foreground(&mut self, client_id: u64) -> bool {
        let stamp = self.allocate_activity_stamp();
        let Some(client) = self.clients.get_mut(&client_id) else {
            return false;
        };
        client.last_activity = stamp;

        let changed = self.foreground_client_id != Some(client_id);
        self.foreground_client_id = Some(client_id);
        self.sync_foreground_client_state();
        changed
    }

    fn promote_latest_remaining_client(&mut self) -> bool {
        let next_foreground = self
            .clients
            .iter()
            .max_by_key(|(_, client)| client.last_activity)
            .map(|(&client_id, _)| client_id);
        let changed = next_foreground != self.foreground_client_id;
        self.foreground_client_id = next_foreground;
        self.sync_foreground_client_state();
        changed
    }

    fn remove_client(&mut self, client_id: u64) -> bool {
        let was_foreground = self.foreground_client_id == Some(client_id);
        self.clients.remove(&client_id);
        if was_foreground {
            self.promote_latest_remaining_client()
        } else {
            false
        }
    }

    fn update_client_host_theme_from_events(
        &mut self,
        client_id: u64,
        events: &[crate::raw_input::RawInputEvent],
    ) -> bool {
        let Some(client) = self.clients.get_mut(&client_id) else {
            return false;
        };

        let mut next_theme = client.host_terminal_theme;
        for event in events {
            if let crate::raw_input::RawInputEvent::HostDefaultColor { kind, color } = event {
                next_theme = next_theme.with_color(*kind, *color);
            }
        }

        if next_theme == client.host_terminal_theme {
            return false;
        }

        client.host_terminal_theme = next_theme;
        if self.foreground_client_id == Some(client_id) {
            self.app.set_host_terminal_theme(next_theme)
        } else {
            false
        }
    }

    fn events_include_interaction(events: &[crate::raw_input::RawInputEvent]) -> bool {
        events.iter().any(|event| {
            matches!(
                event,
                crate::raw_input::RawInputEvent::Key(_)
                    | crate::raw_input::RawInputEvent::Mouse(_)
                    | crate::raw_input::RawInputEvent::Paste(_)
            )
        })
    }

    /// Accepts pending client connections from the non-blocking listener.
    fn accept_client_connections(&mut self) -> io::Result<()> {
        loop {
            match self.client_listener.accept() {
                Ok((stream, _addr)) => {
                    let client_id = self.next_client_id;
                    self.next_client_id += 1;

                    if let Err(err) = stream.set_nonblocking(true) {
                        warn!(err = %err, "failed to set client stream nonblocking");
                        continue;
                    }

                    // Spawn a thread for the handshake and read loop.
                    let should_quit = self.should_quit.clone();
                    let server_event_tx = self.server_event_tx.clone();
                    std::thread::spawn(move || {
                        if let Err(err) = handle_client_handshake(
                            stream,
                            client_id,
                            &server_event_tx,
                            &should_quit,
                        ) {
                            debug!(client_id, err = %err, "client handshake failed");
                        }
                    });
                }
                Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => {
                    // No more pending connections.
                    break;
                }
                Err(err) => {
                    error!(err = %err, "client listener accept failed");
                    break;
                }
            }
        }
        Ok(())
    }

    /// Drains server events from the dedicated channel.
    ///
    /// Returns true if any input was processed (requiring a re-render).
    fn drain_server_events(&mut self) -> bool {
        let mut changed = false;
        while let Ok(ev) = self.server_event_rx.try_recv() {
            changed |= self.handle_server_event(ev);
        }
        changed
    }

    /// Handles a single internal event with forwarding logic for clipboard,
    /// sound, and toast notifications to connected clients.
    ///
    /// ALL internal events MUST be routed through this method to ensure
    /// clipboard/notify forwarding is never bypassed. Do not call
    /// `self.app.handle_internal_event()` directly for any internal event
    /// in the headless server — use this method instead.
    ///
    /// Returns true if the event changed visual state (requiring a re-render).
    fn handle_internal_event_with_forwarding(&mut self, ev: AppEvent) -> bool {
        match &ev {
            AppEvent::ClipboardWrite { content } => {
                // Clipboard writes are client-local side effects. Forward them only to
                // the foreground client instead of broadcasting to every attached client.
                if let Some(client_id) = self.foreground_client_id {
                    let data = base64::engine::general_purpose::STANDARD.encode(content.as_slice());
                    self.send_to_client(client_id, ServerMessage::Clipboard { data });
                }
                // ClipboardWrite doesn't change visual state — no render needed.
                false
            }
            AppEvent::StateChanged {
                pane_id,
                agent,
                state,
            } => {
                // Capture toast before handling.
                let toast_before = self.app.state.toast.clone();
                let pane_id_val = *pane_id;
                let agent_val = *agent;
                let state_val = *state;

                // Find the previous state of this pane before the event
                // is processed. We need this to determine if a sound
                // notification would be triggered.
                let prev_state = self
                    .app
                    .state
                    .workspaces
                    .iter()
                    .find_map(|ws| {
                        ws.tabs
                            .iter()
                            .find_map(|tab| tab.panes.get(&pane_id_val).map(|p| p.state))
                    })
                    .unwrap_or(crate::detect::AgentState::Unknown);

                // Handle the state change (updates pane state, sets toast on AppState).
                // Note: apply_pane_state_change inside handle_internal_event will try
                // to play sounds, but sound.enabled=false in the headless server, so
                // sound::play is never called. Toast may still be set on AppState if
                // toast_config.enabled is true.
                self.app.handle_internal_event(ev);

                // Forward sound notification to clients.
                // We check the agent-specific sound setting but NOT sound.enabled,
                // because the server sets enabled=false to prevent local playback —
                // clients should still receive sound notifications and decide
                // locally whether to play them based on their own config.
                let is_active_tab = self
                    .app
                    .state
                    .active
                    .and_then(|ws_idx| self.app.state.workspaces.get(ws_idx))
                    .is_some_and(|ws| {
                        ws.find_tab_index_for_pane(pane_id_val)
                            .is_some_and(|tab_idx| ws.active_tab_index() == tab_idx)
                    });

                if !matches!(
                    self.app.state.sound.agents.for_agent(agent_val),
                    crate::config::AgentSoundSetting::Off
                ) {
                    if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
                        is_active_tab,
                        prev_state,
                        state_val,
                    ) {
                        let msg = match sound {
                            crate::sound::Sound::Done => "agent done",
                            crate::sound::Sound::Request => "agent attention",
                        };
                        self.send_to_all_clients(ServerMessage::Notify {
                            kind: protocol::NotifyKind::Sound,
                            message: msg.to_owned(),
                        });
                    }
                }

                let toast_msg =
                    if self.app.state.toast.is_some() && self.app.state.toast != toast_before {
                        self.app
                            .state
                            .toast
                            .as_ref()
                            .map(|toast| format!("{}: {}", toast.title, toast.context))
                    } else if matches!(
                        self.app.state.toast_config.delivery,
                        crate::config::ToastDelivery::Terminal
                    ) {
                        self.app
                            .state
                            .workspaces
                            .iter()
                            .enumerate()
                            .find_map(|(ws_idx, ws)| {
                                ws.tabs.iter().find_map(|tab| {
                                    tab.panes.get(&pane_id_val).and_then(|pane| {
                                        pane.effective_agent_label().and_then(|agent_label| {
                                        crate::app::actions::notification_toast_for_state_change(
                                            is_active_tab,
                                            prev_state,
                                            state_val,
                                        )
                                        .map(|kind| {
                                            let event_text = match kind {
                                                crate::app::state::ToastKind::NeedsAttention => {
                                                    "needs attention"
                                                }
                                                crate::app::state::ToastKind::Finished => {
                                                    "finished"
                                                }
                                                crate::app::state::ToastKind::UpdateInstalled => {
                                                    "updated"
                                                }
                                            };
                                            format!(
                                                "{} {}: {}",
                                                agent_label,
                                                event_text,
                                                crate::app::actions::notification_context(
                                                    ws,
                                                    ws_idx,
                                                    pane_id_val,
                                                )
                                            )
                                        })
                                    })
                                    })
                                })
                            })
                    } else {
                        None
                    };

                if let Some(msg) = toast_msg {
                    self.send_to_all_clients(ServerMessage::Notify {
                        kind: protocol::NotifyKind::Toast,
                        message: msg,
                    });
                }

                true
            }
            AppEvent::HookStateReported {
                pane_id,
                agent_label,
                state,
                ..
            } => {
                // The hook authority may not change the effective pane state if
                // the detected agent doesn't match. We forward based on the
                // hook-reported state transition regardless.
                let toast_before = self.app.state.toast.clone();
                let pane_id_val = *pane_id;
                let agent_val = crate::detect::parse_agent_label(agent_label);
                let hook_state_val = *state;

                // Capture the previous hook authority state for this pane.
                // If no hook authority exists, use the effective state.
                let prev_hook_state = self
                    .app
                    .state
                    .workspaces
                    .iter()
                    .find_map(|ws| {
                        ws.tabs.iter().find_map(|tab| {
                            tab.panes.get(&pane_id_val).map(|p| {
                                p.hook_authority
                                    .as_ref()
                                    .map(|h| h.state)
                                    .unwrap_or(p.state)
                            })
                        })
                    })
                    .unwrap_or(crate::detect::AgentState::Unknown);

                self.app.handle_internal_event(ev);

                // Forward sound notification based on hook state transition.
                // This ensures API-reported state changes (pane.report_agent)
                // produce notifications even before fallback detection confirms.
                let is_active_tab = self
                    .app
                    .state
                    .active
                    .and_then(|ws_idx| self.app.state.workspaces.get(ws_idx))
                    .is_some_and(|ws| {
                        ws.find_tab_index_for_pane(pane_id_val)
                            .is_some_and(|tab_idx| ws.active_tab_index() == tab_idx)
                    });

                if !matches!(
                    self.app.state.sound.agents.for_agent(agent_val),
                    crate::config::AgentSoundSetting::Off
                ) {
                    if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
                        is_active_tab,
                        prev_hook_state,
                        hook_state_val,
                    ) {
                        let msg = match sound {
                            crate::sound::Sound::Done => "agent done",
                            crate::sound::Sound::Request => "agent attention",
                        };
                        self.send_to_all_clients(ServerMessage::Notify {
                            kind: protocol::NotifyKind::Sound,
                            message: msg.to_owned(),
                        });
                    }
                }

                let toast_msg =
                    if self.app.state.toast.is_some() && self.app.state.toast != toast_before {
                        self.app
                            .state
                            .toast
                            .as_ref()
                            .map(|toast| format!("{}: {}", toast.title, toast.context))
                    } else if matches!(
                        self.app.state.toast_config.delivery,
                        crate::config::ToastDelivery::Terminal
                    ) {
                        self.app
                            .state
                            .workspaces
                            .iter()
                            .enumerate()
                            .find_map(|(ws_idx, ws)| {
                                ws.tabs.iter().find_map(|tab| {
                                    tab.panes.get(&pane_id_val).and_then(|pane| {
                                        pane.effective_agent_label().and_then(|agent_label| {
                                        crate::app::actions::notification_toast_for_state_change(
                                            is_active_tab,
                                            prev_hook_state,
                                            hook_state_val,
                                        )
                                        .map(|kind| {
                                            let event_text = match kind {
                                                crate::app::state::ToastKind::NeedsAttention => {
                                                    "needs attention"
                                                }
                                                crate::app::state::ToastKind::Finished => {
                                                    "finished"
                                                }
                                                crate::app::state::ToastKind::UpdateInstalled => {
                                                    "updated"
                                                }
                                            };
                                            format!(
                                                "{} {}: {}",
                                                agent_label,
                                                event_text,
                                                crate::app::actions::notification_context(
                                                    ws,
                                                    ws_idx,
                                                    pane_id_val,
                                                )
                                            )
                                        })
                                    })
                                    })
                                })
                            })
                    } else {
                        None
                    };

                if let Some(msg) = toast_msg {
                    self.send_to_all_clients(ServerMessage::Notify {
                        kind: protocol::NotifyKind::Toast,
                        message: msg,
                    });
                }

                true
            }
            AppEvent::UpdateReady { version } => {
                let toast_before = self.app.state.toast.clone();
                let version = version.clone();

                self.app.handle_internal_event(ev);

                let toast_msg =
                    if self.app.state.toast.is_some() && self.app.state.toast != toast_before {
                        self.app
                            .state
                            .toast
                            .as_ref()
                            .map(|toast| format!("{}: {}", toast.title, toast.context))
                    } else if matches!(
                        self.app.state.toast_config.delivery,
                        crate::config::ToastDelivery::Terminal
                    ) {
                        Some(format!(
                            "v{version} available: detach, then run `herdr update`"
                        ))
                    } else {
                        None
                    };

                if let Some(msg) = toast_msg {
                    self.send_to_all_clients(ServerMessage::Notify {
                        kind: protocol::NotifyKind::Toast,
                        message: msg,
                    });
                }

                true
            }
            _ => {
                self.app.handle_internal_event(ev);
                true
            }
        }
    }

    /// Drains internal events, forwarding clipboard, sound, and toast
    /// notifications to connected clients instead of processing them locally.
    ///
    /// In the monolithic mode:
    /// - `ClipboardWrite` events are written to stdout via `write_osc52_bytes`.
    /// - Sound notifications are played locally via `sound::play`.
    /// - Toast notifications are set on AppState and rendered into the frame.
    ///
    /// In the headless server, there is no stdout terminal or audio subsystem,
    /// so we:
    /// - Forward `ClipboardWrite` as `ServerMessage::Clipboard` to the
    ///   foreground client only.
    /// - Detect when a sound would be played and forward as
    ///   `ServerMessage::Notify { kind: Sound }` to all clients.
    /// - Detect when a toast is set on AppState and forward as
    ///   `ServerMessage::Notify { kind: Toast }` to all clients.
    fn drain_internal_events_with_forwarding(&mut self) -> bool {
        let mut changed = false;
        while let Ok(ev) = self.app.event_rx.try_recv() {
            changed |= self.handle_internal_event_with_forwarding(ev);
        }
        changed
    }

    /// Encodes a server message into a length-prefixed frame.
    fn frame_server_message(msg: &ServerMessage) -> Result<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,
                        last_frame: None,
                        writer: Some(writer),
                    },
                );
                self.foreground_client_id = Some(client_id);
                self.sync_foreground_client_state();
                self.resize_shared_runtime_to_effective_size();
                true
            }
            ServerEvent::ClientInput { client_id, data } => {
                debug!(client_id, len = data.len(), "client input received");
                if let Some(client) = self.clients.get_mut(&client_id) {
                    // Ensure the next render after client input is delivered even if the
                    // current frame buffer still compares equal. Input can change cursor or
                    // PTY state asynchronously, and thin clients should not stall waiting for
                    // a post-input frame behind identical-frame dedupe.
                    client.last_frame = None;
                }
                let events = crate::raw_input::parse_raw_input_bytes_sync(&data);
                let interaction = Self::events_include_interaction(&events);
                let foreground_changed = if interaction {
                    self.promote_client_to_foreground(client_id)
                } else {
                    false
                };
                if foreground_changed {
                    self.resize_shared_runtime_to_effective_size();
                }
                let theme_changed = self.update_client_host_theme_from_events(client_id, &events);
                self.app
                    .route_client_events(events, self.foreground_client_id == Some(client_id));

                // Check if the detach keybind was triggered during input processing.
                if self.app.state.detach_requested {
                    self.app.state.detach_requested = false;
                    info!(client_id, "client detach requested via keybind");

                    // Send a ServerShutdown with "detached" reason to this client
                    // so it exits cleanly (not with a connection-lost error).
                    // The client will close its connection after receiving this,
                    // which triggers a ClientDisconnected event that removes it.
                    self.send_to_client(
                        client_id,
                        ServerMessage::ServerShutdown {
                            reason: Some("detached".to_owned()),
                        },
                    );

                    // Don't remove the client here — let the client disconnect
                    // naturally after receiving the ServerShutdown. The client's
                    // read loop will see EOF and the server will get a
                    // ClientDisconnected event which handles cleanup.
                    //
                    // However, we do need to stop sending frames to this client
                    // since it's detaching. Drop the writer channel so no more
                    // frames are queued for this client.
                    if let Some(client) = self.clients.get_mut(&client_id) {
                        client.writer = None;
                    }

                    // No re-render needed for remaining clients.
                    false
                } else {
                    foreground_changed || theme_changed || interaction
                }
            }
            ServerEvent::ClientResize {
                client_id,
                cols,
                rows,
            } => {
                info!(client_id, cols, rows, "client resize");
                if let Some(client) = self.clients.get_mut(&client_id) {
                    client.terminal_size = (cols, rows);
                }
                self.promote_client_to_foreground(client_id);
                self.resize_shared_runtime_to_effective_size();
                true
            }
            ServerEvent::ClientDetach { client_id } => {
                info!(client_id, "client detached");
                let foreground_changed = self.remove_client(client_id);
                if foreground_changed {
                    self.resize_shared_runtime_to_effective_size();
                }
                true
            }
            ServerEvent::ClientDisconnected { client_id } => {
                info!(client_id, "client disconnected");
                let foreground_changed = self.remove_client(client_id);
                if foreground_changed {
                    self.resize_shared_runtime_to_effective_size();
                }
                true
            }
            ServerEvent::QuitSignal => {
                // The quit check at the top of the loop handles this.
                // No render needed — the next iteration will initiate shutdown.
                false
            }
        }
    }

    /// Drains API requests with shutdown awareness.
    ///
    /// During shutdown, remaining requests get a `server_unavailable` error.
    fn drain_api_requests_with_shutdown_check(&mut self) -> bool {
        let mut changed = false;
        while let Ok(msg) = self.app.api_rx.try_recv() {
            changed |= self.handle_api_request_with_shutdown_check(msg);
        }
        changed
    }

    /// Handles a single API request with shutdown awareness.
    ///
    /// Also forwards any toast/sound notifications that result from the API
    /// request to connected clients. API methods like `pane.report_agent`
    /// trigger internal events that may set toast state or would normally
    /// play sounds — in headless mode we forward these to clients instead.
    fn handle_api_request_with_shutdown_check(&mut self, msg: api::ApiRequestMessage) -> bool {
        if self.shutting_down {
            // During shutdown, respond with server_unavailable.
            let response = serde_json::to_string(&api::schema::ErrorResponse {
                id: msg.request.id,
                error: api::schema::ErrorBody {
                    code: "server_unavailable".into(),
                    message: "server is shutting down".into(),
                },
            })
            .unwrap_or_else(|_| {
                r#"{"id":"","error":{"code":"server_unavailable","message":"server is shutting down"}}"#
                    .to_string()
            });
            let _ = msg.respond_to.send(response);
            return false;
        }

        let changed = api::request_changes_ui(&msg.request);

        // Capture toast and pane agent states before the API call so we can
        // forward any resulting notifications to connected clients.
        // API requests like pane.report_agent trigger handle_internal_event
        // internally, which bypasses drain_internal_events_with_forwarding.
        // Since sound.enabled=false in the headless server, sounds would be
        // silently dropped; toasts may be set but not forwarded.
        //
        // Note: pane.report_agent sets hook_authority on the pane, but the
        // effective state may not change until the fallback detector confirms
        // the agent (detected_agent must match). So we capture both the
        // effective state AND the hook authority state for comparison.
        let toast_before = self.app.state.toast.clone();
        let pane_states_before: Vec<(
            usize,
            crate::layout::PaneId,
            crate::detect::AgentState,
            Option<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();
        let forwarded_toast_from_state = if toast_after.is_some() && toast_after != toast_before {
            if let Some(toast) = &toast_after {
                let msg_text = format!("{}: {}", toast.title, toast.context);
                debug!(msg = %msg_text, "forwarding toast notification from API request");
                self.send_to_all_clients(ServerMessage::Notify {
                    kind: protocol::NotifyKind::Toast,
                    message: msg_text,
                });
                true
            } else {
                false
            }
        } else {
            false
        };

        // Forward sound notifications for any pane state changes that occurred
        // during the API request. Compare before/after pane states (including
        // hook authority state) to find transitions that would trigger a sound.
        //
        // pane.report_agent sets hook_authority on the pane, but the effective
        // state may not change until the fallback detector confirms the agent
        // (detected_agent must match hook_authority.agent). We check BOTH the
        // effective state AND the hook authority state so that API-reported
        // state changes trigger notifications even before fallback confirmation.
        for (ws_idx, pane_id, prev_effective_state, prev_hook_state) in &pane_states_before {
            let pane_after = self
                .app
                .state
                .workspaces
                .get(*ws_idx)
                .and_then(|ws| ws.tabs.iter().find_map(|tab| tab.panes.get(pane_id)));

            let Some(pane_after) = pane_after else {
                continue;
            };

            let new_effective_state = pane_after.state;
            let new_hook_state = pane_after.hook_authority.as_ref().map(|h| h.state);

            // Check effective state change first.
            let effective_changed = new_effective_state != *prev_effective_state;
            // Check hook authority state change — this catches API-reported
            // state changes that haven't been confirmed by fallback detection.
            let hook_changed = new_hook_state != *prev_hook_state;

            if effective_changed || hook_changed {
                // Use the hook state if available (it reflects the API-reported
                // state), otherwise use the effective state.
                let prev_state = prev_hook_state.unwrap_or(*prev_effective_state);
                let new_state = new_hook_state.unwrap_or(new_effective_state);

                // Skip if the derived transition is a no-op.
                if prev_state == new_state {
                    continue;
                }

                let is_active_tab = self.app.state.pane_is_in_active_tab(*ws_idx, *pane_id);

                // Get the known agent for sound settings. Unknown custom labels
                // fall back to None so clients use the generic sound behavior.
                let agent = pane_after.effective_known_agent();

                debug!(
                    ws_idx,
                    pane_id = pane_id.raw(),
                    prev_state = ?prev_state,
                    new_state = ?new_state,
                    agent = ?agent,
                    effective_changed,
                    hook_changed,
                    "pane state changed during API request, checking sound notification"
                );

                if !forwarded_toast_from_state
                    && matches!(
                        self.app.state.toast_config.delivery,
                        crate::config::ToastDelivery::Terminal
                    )
                {
                    if let Some(kind) = crate::app::actions::notification_toast_for_state_change(
                        is_active_tab,
                        prev_state,
                        new_state,
                    ) {
                        if let Some(agent_label) = pane_after.effective_agent_label() {
                            let event_text = match kind {
                                crate::app::state::ToastKind::NeedsAttention => "needs attention",
                                crate::app::state::ToastKind::Finished => "finished",
                                crate::app::state::ToastKind::UpdateInstalled => "updated",
                            };
                            let msg_text = format!(
                                "{} {}: {}",
                                agent_label,
                                event_text,
                                crate::app::actions::notification_context(
                                    &self.app.state.workspaces[*ws_idx],
                                    *ws_idx,
                                    *pane_id,
                                )
                            );
                            self.send_to_all_clients(ServerMessage::Notify {
                                kind: protocol::NotifyKind::Toast,
                                message: msg_text,
                            });
                        }
                    }
                }

                // Check agent-specific sound setting but NOT sound.enabled,
                // because the server sets enabled=false to prevent local playback.
                // Clients decide locally whether to play sounds.
                if !matches!(
                    self.app.state.sound.agents.for_agent(agent),
                    crate::config::AgentSoundSetting::Off
                ) {
                    if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
                        is_active_tab,
                        prev_state,
                        new_state,
                    ) {
                        let msg_text = match sound {
                            crate::sound::Sound::Done => "agent done",
                            crate::sound::Sound::Request => "agent attention",
                        };
                        debug!(sound = ?sound, "forwarding sound notification from API request");
                        self.send_to_all_clients(ServerMessage::Notify {
                            kind: protocol::NotifyKind::Sound,
                            message: msg_text.to_owned(),
                        });
                    }
                }
            }
        }

        changed
    }

    /// Renders the current state to client-sized virtual buffers and streams
    /// frames to all connected clients.
    fn render_and_stream(&mut self) {
        let foreground_client_id = self.foreground_client_id;
        let mut render_targets: Vec<(u64, (u16, u16), bool)> = self
            .clients
            .iter()
            .filter(|(_, client)| client.writer.is_some())
            .map(|(&client_id, client)| {
                (
                    client_id,
                    client.terminal_size,
                    foreground_client_id == Some(client_id),
                )
            })
            .collect();

        render_targets.sort_by_key(|(client_id, _, is_foreground)| (*is_foreground, *client_id));

        if render_targets.is_empty() {
            let (cols, rows) = self.effective_size;
            let area = Rect::new(0, 0, cols, rows);
            let _ = render_virtual(&mut self.app.state, area, true);
            debug!(
                cols,
                rows, "rendered virtual frame with no attached clients"
            );
            return;
        }

        let mut broken_clients: Vec<u64> = Vec::new();
        for (client_id, (cols, rows), is_foreground) in render_targets {
            let area = Rect::new(0, 0, cols, rows);
            let (buffer, cursor) = render_virtual(&mut self.app.state, area, is_foreground);
            let frame = FrameData::from_ratatui_buffer(&buffer, cursor);

            let Some(client) = self.clients.get_mut(&client_id) else {
                continue;
            };
            let Some(writer) = client.writer.as_ref().cloned() else {
                continue;
            };

            if client.last_frame.as_ref() == Some(&frame) {
                continue;
            }

            let message = ServerMessage::Frame(frame.clone());
            let serialized = match Self::frame_server_message(&message) {
                Ok(framed) => framed,
                Err(err) => {
                    warn!(client_id, err = %err, "failed to serialize frame for client");
                    broken_clients.push(client_id);
                    continue;
                }
            };

            if writer.send(serialized).is_err() {
                debug!(client_id, "client writer channel closed, marking as broken");
                broken_clients.push(client_id);
                continue;
            }

            client.last_frame = Some(frame);
        }

        if !broken_clients.is_empty() {
            for client_id in broken_clients {
                let foreground_changed = self.remove_client(client_id);
                if foreground_changed {
                    self.resize_shared_runtime_to_effective_size();
                }
            }
        }

        let (cols, rows) = self.effective_size;
        debug!(cols, rows, foreground_client_id = ?self.foreground_client_id, "rendered virtual frame(s)");
    }

    /// Handle scheduled tasks for the headless server.
    ///
    /// Similar to `App::handle_scheduled_tasks` but without resize polling
    /// (the server doesn't have a terminal to resize).
    fn handle_scheduled_tasks_headless(&mut self, now: Instant) -> bool {
        let mut changed = false;

        self.app.sync_headless_animation_timer(now);

        // No resize polling needed — server has no terminal.
        // Client resize messages drive size changes instead.

        if self
            .app
            .config_diagnostic_deadline
            .is_some_and(|deadline| now >= deadline)
        {
            self.app.config_diagnostic_deadline = None;
            self.app.state.config_diagnostic = None;
            changed = true;
        }

        if self
            .app
            .toast_deadline
            .is_some_and(|deadline| now >= deadline)
        {
            self.app.toast_deadline = None;
            self.app.state.toast = None;
            changed = true;
        }

        if self
            .app
            .next_animation_tick
            .is_some_and(|deadline| now >= deadline)
        {
            self.app.state.spinner_tick = self
                .app
                .state
                .spinner_tick
                .wrapping_add(app::HEADLESS_ANIMATION_TICK_STEP);
            self.app.next_animation_tick = Some(now + app::HEADLESS_ANIMATION_INTERVAL);
            changed = true;
        }

        if self
            .app
            .git_refresh_deadline()
            .is_some_and(|deadline| now >= deadline)
        {
            for ws in &mut self.app.state.workspaces {
                ws.refresh_git_ahead_behind();
            }
            self.app.last_git_remote_status_refresh = now;
            changed = true;
        }

        if self
            .app
            .next_auto_update_check
            .is_some_and(|deadline| now >= deadline)
        {
            self.app.run_auto_update_check();
        }

        if self
            .app
            .session_save_deadline
            .is_some_and(|deadline| now >= deadline)
        {
            self.app.save_session_now();
        }

        self.app.sync_headless_animation_timer(now);
        changed
    }

    /// Initiates graceful shutdown.
    fn initiate_shutdown(&mut self) {
        if self.shutting_down {
            return;
        }
        info!("server shutdown initiated");
        self.shutting_down = true;

        // Send ServerShutdown to all connected clients.
        let shutdown_msg = ServerMessage::ServerShutdown {
            reason: Some("server is shutting down".to_owned()),
        };
        self.send_to_all_clients(shutdown_msg);

        // Give client writer threads a moment to flush the shutdown message.
        // A short sleep ensures the message is written to the socket before
        // we close the connections.
        std::thread::sleep(Duration::from_millis(50));

        // Signal the main loop to exit.
        self.should_quit.store(true, Ordering::Release);
        self.app.state.should_quit = true;
    }

    /// Completes the shutdown sequence: send ServerShutdown to clients,
    /// close client connections, remove socket files, and clean up.
    fn complete_shutdown(&mut self) -> io::Result<()> {
        info!("completing server shutdown");

        // Send ServerShutdown to all remaining clients.
        if !self.clients.is_empty() {
            let shutdown_msg = ServerMessage::ServerShutdown {
                reason: Some("server is shutting down".to_owned()),
            };
            self.send_to_all_clients(shutdown_msg);

            // Give writer threads a moment to flush before closing.
            std::thread::sleep(Duration::from_millis(50));
        }

        // Drain remaining API requests with server_unavailable.
        self.drain_api_requests_with_shutdown_check();

        // Close all client connections.
        self.clients.clear();

        // Remove socket files.
        self.cleanup_sockets()?;

        Ok(())
    }

    /// Removes socket files created by the server.
    fn cleanup_sockets(&self) -> io::Result<()> {
        if let Err(err) = fs::remove_file(&self.client_socket_path) {
            if err.kind() != io::ErrorKind::NotFound {
                warn!(
                    path = %self.client_socket_path.display(),
                    err = %err,
                    "failed to remove client socket on shutdown"
                );
            }
        }
        Ok(())
    }
}

impl Drop for HeadlessServer {
    fn drop(&mut self) {
        let _ = self.cleanup_sockets();
    }
}

// ---------------------------------------------------------------------------
// Client handshake
// ---------------------------------------------------------------------------

/// Internal event sent from the handshake thread to the main event loop.
#[derive(Debug)]
pub enum ServerEvent {
    /// A new client completed the handshake.
    ClientConnected {
        client_id: u64,
        cols: u16,
        rows: u16,
        writer: std::sync::mpsc::Sender<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));
    crate::logging::shutdown("server");
    result
}

/// Initialize logging for the server process.
fn init_logging() {
    crate::logging::init_file_logging("herdr-server.log");
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    use crate::server::protocol::CursorState;

    fn test_headless_server() -> HeadlessServer {
        let config = crate::config::Config::default();
        let (_api_tx, api_rx) = tokio::sync::mpsc::unbounded_channel();
        let app = crate::app::App::new(&config, true, None, None, api_rx, api::EventHub::default());

        let dir = std::env::temp_dir().join(format!(
            "hh-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        let _ = fs::create_dir_all(&dir);
        let socket_path = dir.join("client.sock");
        let _ = fs::remove_file(&socket_path);
        let listener = UnixListener::bind(&socket_path).expect("bind test listener");
        listener
            .set_nonblocking(true)
            .expect("set listener nonblocking");
        let (server_event_tx, server_event_rx) = mpsc::channel(64);

        HeadlessServer {
            app,
            client_listener: listener,
            client_socket_path: socket_path,
            clients: HashMap::new(),
            next_client_id: 1,
            foreground_client_id: None,
            next_activity_stamp: 1,
            effective_size: (MIN_COLS, MIN_ROWS),
            shutting_down: false,
            should_quit: Arc::new(AtomicBool::new(false)),
            server_event_rx,
            server_event_tx,
        }
    }

    fn read_server_message(bytes: Vec<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 = PathBuf::from(format!(
            "/tmp/hs-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        let _ = fs::create_dir_all(&dir);
        let socket_path = dir.join("stale.sock");

        // Create a socket file that nobody is listening on.
        {
            let _listener = UnixListener::bind(&socket_path).expect("bind stale socket");
        }
        // The listener scope ended, so the socket is now stale.
        let deadline = std::time::Instant::now() + Duration::from_secs(1);
        while std::time::Instant::now() < deadline {
            if std::os::unix::net::UnixStream::connect(&socket_path).is_err() {
                break;
            }
            std::thread::sleep(Duration::from_millis(10));
        }

        // prepare_socket_path should remove it without error.
        let result = prepare_socket_path(&socket_path);
        assert!(result.is_ok(), "should remove stale socket: {result:?}");

        // Socket file should be gone.
        assert!(!socket_path.exists());

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn prepare_socket_path_rejects_live_socket() {
        let dir = PathBuf::from(format!(
            "/tmp/hl-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        let _ = fs::create_dir_all(&dir);
        let socket_path = dir.join("live.sock");

        // Bind a live listener.
        let _listener = UnixListener::bind(&socket_path).expect("bind");

        let result = prepare_socket_path(&socket_path);
        assert!(result.is_err());
        assert_eq!(result.unwrap_err().kind(), io::ErrorKind::AddrInUse);

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn virtual_render_produces_nonempty_buffer() {
        let mut state = AppState::test_new();
        let area = Rect::new(0, 0, 80, 24);
        let (buffer, _cursor) = render_virtual(&mut state, area, true);
        assert_eq!(buffer.area.width, 80);
        assert_eq!(buffer.area.height, 24);
    }

    #[test]
    fn virtual_render_without_frame_cursor_keeps_cursor_hidden() {
        let mut state = AppState::test_new();
        let area = Rect::new(0, 0, 80, 24);
        let (_buffer, cursor) = render_virtual(&mut state, area, true);

        assert_eq!(cursor, None);
    }

    #[tokio::test]
    async fn virtual_render_preserves_explicit_frame_cursor_position() {
        let mut state = AppState::test_new();
        let mut ws = crate::workspace::Workspace::test_new("test");
        let pane_id = ws.tabs[0].root_pane;
        ws.tabs[0].runtimes.insert(
            pane_id,
            crate::pane::PaneRuntime::test_with_screen_bytes(20, 5, b"left"),
        );

        state.workspaces = vec![ws];
        state.active = Some(0);
        state.selected = 0;
        state.mode = crate::app::Mode::Terminal;

        let area = Rect::new(0, 0, 80, 24);
        let (_buffer, cursor) = render_virtual(&mut state, area, true);
        let pane = state
            .view
            .pane_infos
            .iter()
            .find(|info| info.id == pane_id)
            .expect("focused pane info");

        assert_eq!(
            cursor,
            Some(CursorState {
                x: pane.inner_rect.x + 4,
                y: pane.inner_rect.y,
                visible: true,
            })
        );
    }

    #[test]
    fn latest_active_client_drives_shared_size_theme_and_fallback() {
        let mut server = test_headless_server();

        server.clients.insert(
            1,
            ClientConnection {
                terminal_size: (160, 45),
                host_terminal_theme: crate::terminal_theme::TerminalTheme {
                    foreground: Some(crate::terminal_theme::RgbColor {
                        r: 0xaa,
                        g: 0xbb,
                        b: 0xcc,
                    }),
                    background: Some(crate::terminal_theme::RgbColor {
                        r: 0x11,
                        g: 0x22,
                        b: 0x33,
                    }),
                },
                last_activity: 1,
                last_frame: None,
                writer: None,
            },
        );
        server.clients.insert(
            2,
            ClientConnection {
                terminal_size: (80, 24),
                host_terminal_theme: crate::terminal_theme::TerminalTheme {
                    foreground: Some(crate::terminal_theme::RgbColor {
                        r: 0x10,
                        g: 0x20,
                        b: 0x30,
                    }),
                    background: Some(crate::terminal_theme::RgbColor {
                        r: 0xdd,
                        g: 0xee,
                        b: 0xff,
                    }),
                },
                last_activity: 2,
                last_frame: None,
                writer: None,
            },
        );

        assert!(server.promote_client_to_foreground(1));
        assert_eq!(server.foreground_client_id, Some(1));
        assert_eq!(server.effective_size, (160, 45));
        assert_eq!(
            server.app.state.host_terminal_theme,
            server.clients[&1].host_terminal_theme
        );

        assert!(server.promote_client_to_foreground(2));
        assert_eq!(server.foreground_client_id, Some(2));
        assert_eq!(server.effective_size, (80, 24));
        assert_eq!(
            server.app.state.host_terminal_theme,
            server.clients[&2].host_terminal_theme
        );

        assert!(server.remove_client(2));
        assert_eq!(server.foreground_client_id, Some(1));
        assert_eq!(server.effective_size, (160, 45));
        assert_eq!(
            server.app.state.host_terminal_theme,
            server.clients[&1].host_terminal_theme
        );
    }

    #[test]
    fn render_and_stream_uses_each_client_terminal_size() {
        let mut server = test_headless_server();
        server.app.state.workspaces = vec![crate::workspace::Workspace::test_new("test")];
        server.app.state.active = Some(0);
        server.app.state.selected = 0;
        server.app.state.mode = crate::app::Mode::Terminal;

        let (desktop_tx, desktop_rx) = std::sync::mpsc::channel();
        let (phone_tx, phone_rx) = std::sync::mpsc::channel();

        server.clients.insert(
            1,
            ClientConnection {
                terminal_size: (120, 40),
                host_terminal_theme: crate::terminal_theme::TerminalTheme::default(),
                last_activity: 1,
                last_frame: None,
                writer: Some(desktop_tx),
            },
        );
        server.clients.insert(
            2,
            ClientConnection {
                terminal_size: (80, 24),
                host_terminal_theme: crate::terminal_theme::TerminalTheme::default(),
                last_activity: 2,
                last_frame: None,
                writer: Some(phone_tx),
            },
        );
        server.foreground_client_id = Some(1);
        server.sync_foreground_client_state();
        server.resize_shared_runtime_to_effective_size();

        server.render_and_stream();

        let desktop_frame = read_server_frame(desktop_rx.recv().expect("desktop frame"));
        let phone_frame = read_server_frame(phone_rx.recv().expect("phone frame"));

        assert_eq!((desktop_frame.width, desktop_frame.height), (120, 40));
        assert_eq!((phone_frame.width, phone_frame.height), (80, 24));
    }

    #[test]
    fn render_and_stream_skips_identical_frame_sends() {
        let mut server = test_headless_server();
        server.app.state.workspaces = vec![crate::workspace::Workspace::test_new("test")];
        server.app.state.active = Some(0);
        server.app.state.selected = 0;
        server.app.state.mode = crate::app::Mode::Terminal;

        let (client_tx, client_rx) = std::sync::mpsc::channel();

        server.clients.insert(
            1,
            ClientConnection {
                terminal_size: (80, 24),
                host_terminal_theme: crate::terminal_theme::TerminalTheme::default(),
                last_activity: 1,
                last_frame: None,
                writer: Some(client_tx),
            },
        );
        server.foreground_client_id = Some(1);
        server.sync_foreground_client_state();
        server.resize_shared_runtime_to_effective_size();

        server.render_and_stream();
        let first = client_rx.recv_timeout(Duration::from_millis(100));
        assert!(first.is_ok(), "expected first frame to be sent");

        server.render_and_stream();
        assert!(
            client_rx.recv_timeout(Duration::from_millis(50)).is_err(),
            "identical frame should not be sent twice"
        );
    }

    #[test]
    fn clipboard_write_targets_foreground_client_only() {
        let mut server = test_headless_server();
        let (background_tx, background_rx) = std::sync::mpsc::channel();
        let (foreground_tx, foreground_rx) = std::sync::mpsc::channel();

        server.clients.insert(
            1,
            ClientConnection {
                terminal_size: (120, 40),
                host_terminal_theme: crate::terminal_theme::TerminalTheme::default(),
                last_activity: 1,
                last_frame: None,
                writer: Some(background_tx),
            },
        );
        server.clients.insert(
            2,
            ClientConnection {
                terminal_size: (80, 24),
                host_terminal_theme: crate::terminal_theme::TerminalTheme::default(),
                last_activity: 2,
                last_frame: None,
                writer: Some(foreground_tx),
            },
        );
        server.foreground_client_id = Some(2);
        server.sync_foreground_client_state();

        let changed = server.handle_internal_event_with_forwarding(AppEvent::ClipboardWrite {
            content: b"test".to_vec(),
        });

        assert!(!changed);
        match read_server_message(
            foreground_rx
                .recv_timeout(Duration::from_millis(100))
                .expect("foreground clipboard message"),
        ) {
            ServerMessage::Clipboard { data } => assert_eq!(data, "dGVzdA=="),
            other => panic!("expected clipboard message, got {other:?}"),
        }
        assert!(
            background_rx
                .recv_timeout(Duration::from_millis(50))
                .is_err(),
            "background client should not receive clipboard writes"
        );
    }

    #[test]
    fn handshake_timeout_is_within_five_second_deadline() {
        // The handshake timeout must be short enough that
        // the connection is guaranteed to close within 5 seconds even with
        // OS overhead (thread scheduling, timer slack, cleanup).
        assert!(
            HANDSHAKE_TIMEOUT < Duration::from_secs(5),
            "HANDSHAKE_TIMEOUT ({:?}) must be less than 5 seconds to guarantee \
             connection close within the 5-second deadline",
            HANDSHAKE_TIMEOUT
        );
    }

    /// Verify that no direct calls to `self.app.handle_internal_event`
    /// exist outside of `handle_internal_event_with_forwarding` in this
    /// module. This ensures the forwarding bypass cannot be reintroduced.
    ///
    /// The search pattern looks for `handle_internal_event` calls that
    /// are NOT inside the `handle_internal_event_with_forwarding` method.
    #[test]
    fn no_handle_internal_event_bypass_in_module() {
        let source = include_str!("headless.rs");

        // Find all lines containing handle_internal_event
        let mut bypass_lines: Vec<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  ")
        );
    }
}