blit_server/

lib.rs

use blit_alacritty::{SearchResult as AlacrittySearchResult, TerminalDriver as AlacrittyDriver};
use blit_compositor::{CompositorCommand, CompositorEvent, CompositorHandle};
use blit_remote::{
    C2S_ACK, C2S_AUDIO_SUBSCRIBE, C2S_AUDIO_UNSUBSCRIBE, C2S_CLIENT_FEATURES, C2S_CLIENT_METRICS,
    C2S_CLIPBOARD_GET, C2S_CLIPBOARD_LIST, C2S_CLIPBOARD_SET, C2S_CLOSE, C2S_COPY_RANGE,
    C2S_CREATE, C2S_CREATE_AT, C2S_CREATE_N, C2S_CREATE2, C2S_DISPLAY_RATE, C2S_FOCUS, C2S_INPUT,
    C2S_KILL, C2S_MOUSE, C2S_PING, C2S_QUIT, C2S_READ, C2S_RESIZE, C2S_RESTART, C2S_SCROLL,
    C2S_SEARCH, C2S_SUBSCRIBE, C2S_SURFACE_ACK, C2S_SURFACE_CAPTURE, C2S_SURFACE_CLOSE,
    C2S_SURFACE_FOCUS, C2S_SURFACE_INPUT, C2S_SURFACE_LIST, C2S_SURFACE_POINTER,
    C2S_SURFACE_POINTER_AXIS, C2S_SURFACE_RESIZE, C2S_SURFACE_SUBSCRIBE, C2S_SURFACE_TEXT,
    C2S_SURFACE_UNSUBSCRIBE, C2S_UNSUBSCRIBE, CAPTURE_FORMAT_AVIF, CAPTURE_FORMAT_PNG,
    CREATE2_HAS_COMMAND, CREATE2_HAS_SRC_PTY, FEATURE_AUDIO, FEATURE_COMPOSITOR,
    FEATURE_COPY_RANGE, FEATURE_CREATE_NONCE, FEATURE_RESIZE_BATCH, FEATURE_RESTART, FrameState,
    READ_ANSI, READ_TAIL, S2C_CLOSED, S2C_CREATED, S2C_CREATED_N, S2C_LIST, S2C_PING, S2C_QUIT,
    S2C_READY, S2C_SEARCH_RESULTS, S2C_SURFACE_CAPTURE, S2C_SURFACE_LIST, S2C_TEXT, S2C_TITLE,
    SURFACE_FRAME_FLAG_KEYFRAME, build_update_msg, msg_hello, msg_s2c_clipboard_content,
    msg_s2c_clipboard_list, msg_surface_app_id, msg_surface_created, msg_surface_destroyed,
    msg_surface_encoder, msg_surface_frame, msg_surface_resized, msg_surface_title,
};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex, Notify, mpsc};

#[cfg(unix)]
mod audio;
mod gpu_libs;
mod ipc;
mod nvenc_encode;
mod pty;
mod surface_encoder;
#[cfg(target_os = "linux")]
mod vaapi_encode;

pub use ipc::{IpcListener, default_ipc_path};
use pty::{PtyHandle, PtyWriteTarget};
use surface_encoder::SurfaceEncoder;
pub use surface_encoder::SurfaceEncoderPreference;
pub use surface_encoder::SurfaceH264EncoderPreference;
pub use surface_encoder::SurfaceQuality;

type PtyFds = Arc<std::sync::RwLock<HashMap<u16, PtyWriteTarget>>>;
pub struct Config {
    pub shell: String,
    pub shell_flags: String,
    pub scrollback: usize,
    pub ipc_path: String,
    pub surface_encoders: Vec<SurfaceEncoderPreference>,
    pub surface_quality: SurfaceQuality,
    pub vaapi_device: String,
    #[cfg(unix)]
    pub fd_channel: Option<std::os::unix::io::RawFd>,
    pub verbose: bool,
    /// Maximum number of concurrent client connections (0 = unlimited).
    pub max_connections: usize,
    /// Maximum number of PTYs across all clients (0 = unlimited).
    pub max_ptys: usize,
    /// Application-level ping interval.  The server sends S2C_PING to every
    /// client at this cadence so that transports without native keepalive
    /// (WebRTC data channels) can detect dead connections.  0 = disabled.
    pub ping_interval: Duration,
    /// Skip compositor initialization (e.g. for share-only mode).
    pub skip_compositor: bool,
}

trait PtyDriver: Send {
    fn size(&self) -> (u16, u16);
    fn resize(&mut self, rows: u16, cols: u16);
    fn process(&mut self, data: &[u8]);
    fn title(&self) -> &str;
    fn search_result(&self, query: &str) -> Option<PtySearchResult>;
    fn take_title_dirty(&mut self) -> bool;
    fn cursor_position(&self) -> (u16, u16);
    fn synced_output(&self) -> bool;
    fn snapshot(&mut self, echo: bool, icanon: bool) -> FrameState;
    fn scrollback_frame(&mut self, offset: usize) -> FrameState;
    fn reset_modes(&mut self);
    fn mouse_event(
        &self,
        type_: u8,
        button: u8,
        col: u16,
        row: u16,
        echo: bool,
        icanon: bool,
    ) -> Option<Vec<u8>>;
    fn get_text_range(
        &self,
        start_tail: u32,
        start_col: u16,
        end_tail: u32,
        end_col: u16,
    ) -> String;
    fn total_lines(&self) -> u32;
}

struct PtySearchResult {
    score: u32,
    primary_source: u8,
    matched_sources: u8,
    context: String,
    scroll_offset: Option<usize>,
}

impl PtyDriver for AlacrittyDriver {
    fn size(&self) -> (u16, u16) {
        AlacrittyDriver::size(self)
    }

    fn resize(&mut self, rows: u16, cols: u16) {
        AlacrittyDriver::resize(self, rows, cols);
    }

    fn process(&mut self, data: &[u8]) {
        AlacrittyDriver::process(self, data);
    }

    fn title(&self) -> &str {
        AlacrittyDriver::title(self)
    }

    fn search_result(&self, query: &str) -> Option<PtySearchResult> {
        AlacrittyDriver::search_result(self, query).map(|result: AlacrittySearchResult| {
            PtySearchResult {
                score: result.score,
                primary_source: result.primary_source as u8,
                matched_sources: result.matched_sources,
                context: result.context,
                scroll_offset: result.scroll_offset,
            }
        })
    }

    fn take_title_dirty(&mut self) -> bool {
        AlacrittyDriver::take_title_dirty(self)
    }

    fn cursor_position(&self) -> (u16, u16) {
        AlacrittyDriver::cursor_position(self)
    }

    fn synced_output(&self) -> bool {
        AlacrittyDriver::synced_output(self)
    }

    fn snapshot(&mut self, echo: bool, icanon: bool) -> FrameState {
        AlacrittyDriver::snapshot(self, echo, icanon)
    }

    fn scrollback_frame(&mut self, offset: usize) -> FrameState {
        AlacrittyDriver::scrollback_frame(self, offset)
    }

    fn reset_modes(&mut self) {
        AlacrittyDriver::reset_modes(self);
    }

    fn mouse_event(
        &self,
        type_: u8,
        button: u8,
        col: u16,
        row: u16,
        echo: bool,
        icanon: bool,
    ) -> Option<Vec<u8>> {
        AlacrittyDriver::mouse_event(self, type_, button, col, row, echo, icanon)
    }

    fn get_text_range(
        &self,
        start_tail: u32,
        start_col: u16,
        end_tail: u32,
        end_col: u16,
    ) -> String {
        AlacrittyDriver::get_text_range(self, start_tail, start_col, end_tail, end_col)
    }

    fn total_lines(&self) -> u32 {
        AlacrittyDriver::total_lines(self)
    }
}

// Keep small to limit bufferbloat on slow connections. The soft queue budget
// is enforced by both message count and queued bytes so a couple of tiny
// terminal diffs do not block a large surface frame.
const OUTBOX_CAPACITY: usize = 8;
const OUTBOX_SOFT_QUEUE_LIMIT_FRAMES: usize = 4;
const OUTBOX_SOFT_QUEUE_LIMIT_BYTES: usize = 128 * 1024;
const PREVIEW_FRAME_RESERVE: usize = 1;
const READY_FRAME_QUEUE_CAP: usize = 4;
const PTY_CHANNEL_CAPACITY: usize = 64;
const SYNC_OUTPUT_END: &[u8] = b"\x1b[?2026l";

/// Number of surface frames to send at wire speed after a keyframe request
/// (subscribe, resubscribe, or error recovery).  During this burst window
/// only outbox backpressure gates delivery — the time-based pacing interval
/// is skipped.  This lets bandwidth estimates ramp up quickly on high-latency
/// links instead of starving the pipeline with conservative initial rates.
const SURFACE_BURST_FRAMES: u8 = 4;

/// A chunk of data from the PTY reader, sent through a lock-free channel
/// so the reader never contends with the delivery tick for the Session mutex.
enum PtyInput {
    /// Raw bytes from the PTY, with the reader's sync-scan tail for boundary
    /// detection. The tick task calls `process()` + `respond_to_queries()`.
    Data(Vec<u8>),
    /// Data up to a sync-output-close boundary. `before` should be processed
    /// and then a snapshot taken. `after` is remainder for the next chunk.
    SyncBoundary { before: Vec<u8>, after: Vec<u8> },
    /// The PTY fd hit EOF or an error — the child likely exited.
    Eof,
}

const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024;

async fn read_frame(reader: &mut (impl AsyncRead + Unpin)) -> Option<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    reader.read_exact(&mut len_buf).await.ok()?;
    let len = u32::from_le_bytes(len_buf) as usize;
    if len == 0 {
        return Some(vec![]);
    }
    if len > MAX_FRAME_SIZE {
        return None;
    }
    let mut buf = vec![0u8; len];
    reader.read_exact(&mut buf).await.ok()?;
    Some(buf)
}

async fn write_frame(writer: &mut (impl AsyncWrite + Unpin), payload: &[u8]) -> bool {
    if payload.len() > u32::MAX as usize {
        return false;
    }
    let len = payload.len() as u32;
    let mut buf = Vec::with_capacity(4 + payload.len());
    buf.extend_from_slice(&len.to_le_bytes());
    buf.extend_from_slice(payload);
    writer.write_all(&buf).await.is_ok()
}

/// Write a length-prefixed frame, draining any pending audio frames
/// before the bulk write.  Audio frames are complete length-prefixed
/// messages on the same stream, so they MUST be written between bulk
/// frames — never interleaved within one.
///
/// Previous code inserted audio frames between write-syscall chunks of
/// a single bulk frame.  This corrupted the stream: the reader does
/// `read_exact(len, 4)` then `read_exact(payload, len)` and expects a
/// contiguous `4+len` bytes.  Inserting audio bytes inside that span
/// corrupts the payload and desynchronises every subsequent frame.
async fn write_frame_interleaved(
    writer: &mut (impl AsyncWrite + Unpin),
    payload: &[u8],
    audio_rx: &mut mpsc::Receiver<Vec<u8>>,
) -> bool {
    // Drain pending audio *before* the bulk frame so audio is never
    // starved for more than one bulk-frame write time.  Audio frames
    // are tiny (~160 B) so this is near-instant.
    while let Ok(audio_msg) = audio_rx.try_recv() {
        if !write_frame(writer, &audio_msg).await {
            return false;
        }
    }
    write_frame(writer, payload).await
}

struct Pty {
    handle: PtyHandle,
    driver: Box<dyn PtyDriver>,
    /// Client-chosen tag set at creation time.
    tag: String,
    dirty: bool,
    ready_frames: VecDeque<FrameState>,
    /// Receives raw byte chunks from the PTY reader task without mutex contention.
    byte_rx: mpsc::Receiver<PtyInput>,
    reader_handle: std::thread::JoinHandle<()>,
    /// Cached (echo, icanon) from tcgetattr; refreshed every ~250ms.
    lflag_cache: (bool, bool),
    lflag_last: Instant,
    /// When we last broadcast a title update for this PTY.
    last_title_send: Instant,
    /// Title changed but not yet sent (debounced).
    title_pending: bool,
    /// The subprocess has exited but the terminal state is retained for reading.
    exited: bool,
    /// Exit status: WEXITSTATUS if normal exit, negative signal number if signalled,
    /// EXIT_STATUS_UNKNOWN if not yet collected.
    exit_status: i32,
    /// Command used to create this PTY (None = default shell).
    command: Option<String>,
}

impl Pty {
    fn mark_dirty(&mut self) {
        self.dirty = true;
    }

    fn clear_dirty(&mut self) {
        self.dirty = false;
    }
}

struct CachedSurfaceInfo {
    surface_id: u16,
    parent_id: u16,
    width: u16,
    height: u16,
    title: String,
    app_id: String,
}

/// Last committed pixel buffer for a surface, kept so we can re-encode a
/// keyframe for late-joining clients without going back to the compositor.
struct LastPixels {
    width: u32,
    height: u32,
    pixels: blit_compositor::PixelData,
    /// Monotonically increasing counter bumped on every SurfaceCommit.
    /// Used to skip re-encoding when the pixel data hasn't changed.
    generation: u64,
}

struct SharedCompositor {
    handle: CompositorHandle,
    surfaces: HashMap<u16, CachedSurfaceInfo>,
    last_pixels: HashMap<u16, LastPixels>,
    /// Per-surface timestamp of the last RequestFrame sent.  Used to
    /// throttle requests to at most one per 1 ms so frame callbacks
    /// carry distinct `elapsed_ms` timestamps — video players (mpv)
    /// use these to pace their presentation clock.  Supports up to 1 kHz.
    last_frame_request: HashMap<u16, Instant>,
    created_at: Instant,
    /// Monotonically increasing counter for pixel generations.
    pixel_generation: u64,
    /// Last time we sent blanket RequestFrame for all surfaces (including
    /// those without subscribers).  Throttled to prevent hot-looping when
    /// apps commit at high rates without any client consuming frames.
    last_blanket_frame_request: Instant,
    /// Last dimensions sent to the compositor via `CompositorCommand::SurfaceResize`.
    /// Used to dedup resize commands — the composited output size
    /// (`info.width`/`info.height`) may differ from the requested size
    /// when the Wayland client sets `xdg_geometry` (e.g. excluding a
    /// title bar), so we compare against the actually-requested values.
    last_configured_size: HashMap<u16, (u16, u16, u16)>,
    /// Audio capture pipeline (PipeWire → pw-cat → Opus encode).
    /// `None` when PipeWire is not available or `BLIT_AUDIO=0`.
    #[cfg(unix)]
    audio_pipeline: Option<audio::AudioPipeline>,
    /// Compositor instance ID passed to `AudioPipeline::spawn()` so restarts
    /// reuse the same audio runtime directory.
    #[cfg(unix)]
    audio_session_id: u16,
    /// When the last audio pipeline restart was attempted.  Used to enforce a
    /// cooldown so we don't spin on persistent failures.
    #[cfg(unix)]
    last_audio_restart: Option<Instant>,
}

fn encode_rgba_to_png(pixels: &[u8], width: u32, height: u32) -> Vec<u8> {
    let mut buf = Vec::new();
    {
        let expected = (width * height * 4) as usize;
        let actual = pixels.len();
        if actual != expected {
            // Size mismatch — return a 1×1 red pixel PNG rather than panicking.
            let mut encoder = png::Encoder::new(&mut buf, 1, 1);
            encoder.set_color(png::ColorType::Rgba);
            encoder.set_depth(png::BitDepth::Eight);
            let mut writer = encoder.write_header().unwrap();
            writer.write_image_data(&[255, 0, 0, 255]).unwrap();
            eprintln!(
                "[capture] pixel buffer size mismatch: {width}x{height} expected {expected} got {actual}"
            );
        } else {
            let mut encoder = png::Encoder::new(&mut buf, width, height);
            encoder.set_color(png::ColorType::Rgba);
            encoder.set_depth(png::BitDepth::Eight);
            let mut writer = encoder.write_header().unwrap();
            writer.write_image_data(pixels).unwrap();
        }
    }
    buf
}

/// Encode RGBA pixels to AVIF.  `quality` 0 = lossless, 1–100 = lossy.
fn encode_rgba_to_avif(pixels: &[u8], width: u32, height: u32, quality: u8) -> Vec<u8> {
    let rgba: Vec<rgb::RGBA8> = pixels
        .chunks_exact(4)
        .map(|c| rgb::RGBA8::new(c[0], c[1], c[2], c[3]))
        .collect();
    let img = ravif::Img::new(&rgba[..], width as usize, height as usize);
    let q = if quality == 0 { 100.0 } else { quality as f32 };
    let encoder = ravif::Encoder::new()
        .with_quality(q)
        .with_alpha_quality(q)
        .with_speed(6)
        .with_alpha_color_mode(ravif::AlphaColorMode::UnassociatedClean)
        .with_num_threads(None);
    let result = encoder.encode_rgba(img).expect("AVIF encoding failed");
    result.avif_file
}

/// Encode RGBA pixels to the requested capture format.
fn encode_capture(pixels: &[u8], width: u32, height: u32, format: u8, quality: u8) -> Vec<u8> {
    match format {
        CAPTURE_FORMAT_AVIF => encode_rgba_to_avif(pixels, width, height, quality),
        _ => encode_rgba_to_png(pixels, width, height),
    }
}

async fn request_surface_capture_with_timeout(
    command_tx: std::sync::mpsc::Sender<CompositorCommand>,
    surface_id: u16,
    scale_120: u16,
    timeout: Duration,
) -> Option<(u32, u32, Vec<u8>)> {
    let (tx, rx) = std::sync::mpsc::sync_channel(1);
    command_tx
        .send(CompositorCommand::Capture {
            surface_id,
            scale_120,
            reply: tx,
        })
        .ok()?;

    // The compositor replies through a blocking std::sync::mpsc channel.
    // Wait for it off the async runtime so this request never stalls the
    // tokio worker thread or holds the Session mutex while blocked.
    tokio::task::spawn_blocking(move || rx.recv_timeout(timeout))
        .await
        .ok()?
        .ok()
        .flatten()
}

/// Capacity for the dedicated audio outbox channel.  Audio frames are tiny
/// (~160 B / 20 ms ≈ 8 KB/s) so a small channel is fine — but it must be
/// deep enough to absorb a short write stall without dropping frames.
const AUDIO_OUTBOX_CAPACITY: usize = 10; // 200 ms of audio

struct ClientState {
    tx: mpsc::Sender<Vec<u8>>,
    outbox_queued_frames: Arc<AtomicUsize>,
    outbox_queued_bytes: Arc<AtomicUsize>,
    /// Dedicated channel for audio frames.  The writer task selects on this
    /// with higher priority than the main outbox so audio is never starved
    /// by large video/terminal messages.
    audio_tx: mpsc::Sender<Vec<u8>>,
    lead: Option<u16>,
    subscriptions: HashSet<u16>,
    surface_subscriptions: HashSet<u16>,
    /// Whether this client is subscribed to audio frames.
    audio_subscribed: bool,
    /// Per-client audio bitrate preference in kbps from C2S_AUDIO_SUBSCRIBE.
    /// 0 means use the server/env default.
    #[cfg(unix)]
    audio_bitrate_kbps: u16,
    view_sizes: HashMap<u16, (u16, u16)>,
    scroll_offsets: HashMap<u16, usize>,
    scroll_caches: HashMap<u16, FrameState>,
    last_sent: HashMap<u16, FrameState>,
    preview_next_send_at: HashMap<u16, Instant>,
    /// EWMA RTT estimate in milliseconds.
    rtt_ms: f32,
    /// Minimum-path RTT estimate in milliseconds, excluding queue growth.
    min_rtt_ms: f32,
    /// Client's measured display refresh rate (fps), reported via C2S_DISPLAY_RATE.
    display_fps: f32,
    /// EWMA of delivered payload rate in bytes/sec.
    delivery_bps: f32,
    /// EWMA of actual ACKed goodput in bytes/sec, based on ACK cadence rather than RTT.
    goodput_bps: f32,
    /// EWMA of absolute goodput sample-to-sample jitter in bytes/sec.
    goodput_jitter_bps: f32,
    /// Decaying peak goodput jitter in bytes/sec.
    max_goodput_jitter_bps: f32,
    /// Last sampled ACK goodput for jitter estimation.
    last_goodput_sample_bps: f32,
    /// EWMA of acknowledged frame payload size in bytes.
    avg_frame_bytes: f32,
    /// EWMA of acknowledged lead/paced frame payload size in bytes.
    avg_paced_frame_bytes: f32,
    /// EWMA of acknowledged preview/unpaced frame payload size in bytes.
    avg_preview_frame_bytes: f32,
    /// EWMA of surface (video) frame payload size in bytes.  Tracked
    /// separately from terminal frame sizes so surface pacing uses
    /// `goodput_bps / avg_surface_frame_bytes` without polluting
    /// terminal congestion control estimates.
    avg_surface_frame_bytes: f32,
    /// Payload bytes currently in flight (sent, not yet ACKed).
    inflight_bytes: usize,
    /// Oldest in-flight frame first; ACKs arrive in order.
    inflight_frames: VecDeque<InFlightFrame>,
    /// Earliest time the next visual update should be sent for smooth pacing.
    next_send_at: Instant,
    /// Temporary additive window growth used to probe for more throughput after
    /// a conservative backoff. Decays when queue delay grows.
    probe_frames: f32,
    /// Diagnostics.
    frames_sent: u32,
    acks_recv: u32,
    acked_bytes_since_log: usize,
    browser_backlog_frames: u16,
    browser_ack_ahead_frames: u16,
    browser_apply_ms: f32,
    last_metrics_update: Instant,
    last_log: Instant,
    goodput_window_bytes: usize,
    goodput_window_start: Instant,
    surface_next_send_at: Instant,
    surface_needs_keyframe: bool,
    /// Number of surface frames delivered since the last keyframe request
    /// (subscribe, resubscribe, or error recovery).  Used for burst-start:
    /// the first few frames bypass time-based pacing so they flow at wire
    /// speed, letting bandwidth estimates ramp up quickly on high-latency
    /// links instead of starving the pipeline with conservative initial rates.
    surface_burst_remaining: u8,
    /// Per-client video encoders, one per subscribed surface.
    surface_encoders: HashMap<u16, SurfaceEncoder>,
    /// Surface frames in flight — separate from terminal inflight so surface
    /// ACKs feed shared RTT / goodput without corrupting terminal frame-size
    /// averages or probe_frames.
    surface_inflight_frames: VecDeque<InFlightFrame>,
    /// Surfaces whose encoder is currently in a spawn_blocking encode task.
    /// Prevents spawning parallel encode jobs for the same (client, surface)
    /// pair — which would create throwaway encoders and risk concurrent
    /// access to the underlying C codec library.
    surface_encodes_in_flight: HashSet<u16>,
    /// Per-surface pixel generation that was last encoded for this client.
    /// Used to skip re-encoding when pixel data hasn't changed.
    surface_last_encoded_gen: HashMap<u16, u64>,
    /// Per-client desired surface sizes (surface_id → (width, height, scale_120, codec_support)).
    /// Mirrors `view_sizes` for PTYs: the server mediates across all clients
    /// and picks min(width), min(height), max(scale).
    /// `scale_120` is the DPR in 1/120th units (Wayland convention): 240 = 2×.
    surface_view_sizes: HashMap<u16, (u16, u16, u16)>,
    /// Intersection of codec support across all surfaces for this client.
    /// Used to pick an encoder the client can decode.  0 = accept anything.
    surface_codec_support: u8,
    /// Per-surface codec support override from C2S_SURFACE_SUBSCRIBE.
    /// When non-zero, overrides `surface_codec_support` for encoder creation.
    surface_codec_overrides: HashMap<u16, u8>,
    /// Per-surface quality override from C2S_SURFACE_SUBSCRIBE.
    /// `None` means use the server-global default.
    surface_quality_overrides: HashMap<u16, SurfaceQuality>,
    /// Evdev keycodes currently held down by this client on compositor
    /// surfaces.  On disconnect we send synthetic key-up events for each
    /// so modifiers don't stay stuck and keys don't auto-repeat forever.
    pressed_surface_keys: HashSet<u32>,
}

struct InFlightFrame {
    sent_at: Instant,
    bytes: usize,
    paced: bool,
}

/// Frames to keep in flight: enough to cover one RTT at the client's reported
/// display rate. High-latency links need many frames in flight to avoid
/// devolving into stop-and-wait.
fn frame_window(rtt_ms: f32, display_fps: f32) -> usize {
    let frame_ms = 1_000.0 / display_fps.max(1.0);
    let base_frames = (rtt_ms / frame_ms).ceil().max(0.0) as usize;
    let slack_frames = ((base_frames as f32) * 0.125).ceil() as usize + 2;
    base_frames.saturating_add(slack_frames).max(2)
}

fn path_rtt_ms(client: &ClientState) -> f32 {
    if client.min_rtt_ms > 0.0 {
        client.min_rtt_ms
    } else {
        client.rtt_ms
    }
}

fn display_need_bps(client: &ClientState) -> f32 {
    client.avg_paced_frame_bytes.max(256.0) * client.display_fps.max(1.0)
}

fn effective_rtt_ms(client: &ClientState) -> f32 {
    let path_rtt = path_rtt_ms(client);
    let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
    let queue_allowance = frame_ms
        * if throughput_limited(client) {
            4.0
        } else {
            12.0
        };
    client.rtt_ms.clamp(path_rtt, path_rtt + queue_allowance)
}

fn window_rtt_ms(client: &ClientState) -> f32 {
    let effective = effective_rtt_ms(client);
    if !throughput_limited(client) {
        effective
    } else {
        client.rtt_ms.clamp(effective, effective * 2.0)
    }
}

fn target_frame_window(client: &ClientState) -> usize {
    let window_fps = if throughput_limited(client) {
        pacing_fps(client)
    } else {
        browser_pacing_fps(client)
    };
    frame_window(window_rtt_ms(client), window_fps)
        .saturating_add(client.probe_frames.round().max(0.0) as usize)
}

fn base_queue_ms(client: &ClientState) -> f32 {
    let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
    frame_ms * if throughput_limited(client) { 2.0 } else { 8.0 }
}

fn target_queue_ms(client: &ClientState) -> f32 {
    let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
    let probe_scale = if throughput_limited(client) {
        0.25
    } else {
        1.0
    };
    base_queue_ms(client) + client.probe_frames.max(0.0) * frame_ms * probe_scale
}

fn browser_ready(client: &ClientState) -> bool {
    client.browser_ack_ahead_frames <= 1
        && client.browser_apply_ms <= 1.0
        && !outbox_backpressured(client)
}

fn bandwidth_floor_bps(client: &ClientState) -> f32 {
    let browser_ready = browser_ready(client);
    let backlog_scale = match client.browser_backlog_frames {
        0..=2 => 0.9,
        3..=8 => 0.8,
        _ => 0.65,
    };
    let penalty = client
        .goodput_jitter_bps
        .max(client.max_goodput_jitter_bps * 0.5)
        .min(client.goodput_bps * if browser_ready { 0.75 } else { 0.9 });
    let goodput_floor = (client.goodput_bps - penalty)
        .max(client.goodput_bps * if browser_ready { 0.35 } else { 0.2 });
    // On a browser-ready path, the per-frame delivery estimate is already
    // end-to-end and reacts much faster than ACK-window goodput. Halving it
    // leaves large-frame local links chronically underpaced.
    let delivery_floor = client.delivery_bps * if browser_ready { 1.0 } else { 0.5 };
    let recent_sample_floor = if browser_ready && client.last_goodput_sample_bps > 0.0 {
        client.last_goodput_sample_bps * backlog_scale
    } else {
        0.0
    };
    goodput_floor.max(recent_sample_floor).max(delivery_floor)
}

fn pacing_fps(client: &ClientState) -> f32 {
    let frame_bytes = client.avg_paced_frame_bytes.max(256.0);
    let sustainable = bandwidth_floor_bps(client) / frame_bytes;
    sustainable.min(browser_pacing_fps(client))
}

fn throughput_limited(client: &ClientState) -> bool {
    let floor = bandwidth_floor_bps(client);
    // Consider total demand: lead at cadence rate plus previews at their cap.
    // The old check (pacing_fps < cadence * 0.9) only saw lead bandwidth,
    // which is often tiny, so previews could starve the lead undetected.
    let lead_bps = client.avg_paced_frame_bytes.max(256.0) * browser_pacing_fps(client);
    let preview_bps = client.avg_preview_frame_bytes.max(256.0) * client.display_fps.max(1.0);
    (lead_bps + preview_bps) > floor * 0.9
}

fn browser_pacing_fps(client: &ClientState) -> f32 {
    let mut fps = client.display_fps.max(1.0);

    // Backlog and ack-ahead are direct signals from the browser about
    // whether it's keeping up.  No predictive apply-time bound — it
    // consistently underestimates capacity and causes 30fps death spirals.
    let backlog = client.browser_backlog_frames as f32;
    if backlog > 4.0 {
        fps = fps.min(fps * (4.0 / backlog));
    }

    if client.browser_ack_ahead_frames > 4 {
        fps = fps.min(client.display_fps.max(1.0) * 0.5);
    }

    fps.max(1.0)
}

fn browser_backlog_blocked(client: &ClientState) -> bool {
    client.browser_backlog_frames > 8
}

fn byte_budget_for(client: &ClientState, budget_ms: f32) -> usize {
    let budget_bps = if throughput_limited(client) {
        bandwidth_floor_bps(client)
    } else {
        client.goodput_bps.max(bandwidth_floor_bps(client))
    };
    let bytes = budget_bps * budget_ms.max(1.0) / 1_000.0;
    bytes.ceil().max(client.avg_frame_bytes.max(256.0)) as usize
}

fn target_byte_window(client: &ClientState) -> usize {
    let budget = byte_budget_for(client, path_rtt_ms(client) + target_queue_ms(client));
    let frame_bytes = client.avg_paced_frame_bytes.max(256.0).ceil() as usize;
    let target_frames = target_frame_window(client);
    let pipeline_bytes = frame_bytes.saturating_mul(target_frames);
    // For small pipelines (e.g. idle terminals with 1KB frames), allow the
    // full frame window worth of bytes so we pipeline across the RTT instead
    // of stop-and-wait.  For large pipelines (e.g. 50KB frames × 5 frames =
    // 250KB), the budget (BDP-based) is the binding constraint; fall back to
    // a one-frame floor so we don't pile up many RTTs worth of large frames.
    const PIPELINE_FLOOR_LIMIT: usize = 32_768; // 32 KB
    let floor = if pipeline_bytes <= PIPELINE_FLOOR_LIMIT {
        pipeline_bytes
    } else {
        frame_bytes // one-frame floor for large pipelines
    };
    budget.max(floor)
}

fn send_interval(client: &ClientState) -> Duration {
    Duration::from_secs_f64(1.0 / browser_pacing_fps(client).max(1.0) as f64)
}

fn preview_fps(client: &ClientState) -> f32 {
    let mut fps = client.display_fps.max(1.0);
    if client.lead.is_some() && throughput_limited(client) {
        // Only budget preview bandwidth when the link is actually saturated.
        // Without this, large preview frames (e.g. 12 KB) at 30 fps consume
        // 360 KB/s, starving the lead even when lead frames are tiny.
        // On fast links (localhost, LAN), previews run at display_fps.
        let avail = bandwidth_floor_bps(client);
        let lead_bps = client.avg_paced_frame_bytes.max(256.0) * browser_pacing_fps(client);
        let preview_budget = (avail - lead_bps).max(avail * 0.25).max(0.0);
        let bw_cap = preview_budget / client.avg_preview_frame_bytes.max(256.0);
        fps = fps.min(bw_cap.max(1.0));
    }
    fps.max(1.0)
}

fn preview_send_interval(client: &ClientState) -> Duration {
    Duration::from_secs_f64(1.0 / preview_fps(client) as f64)
}

/// Surface frame rate: display rate capped by what the pipe can carry.
///
/// `goodput_bps` is fed by terminal ACKs, surface ACKs, and audio send
/// accounting — so it reflects total pipe utilisation.  Dividing by
/// `avg_surface_frame_bytes` gives the sustainable surface fps.
///
/// Additional backpressure layers below this:
/// - `surface_encodes_in_flight` limits to 1 encode per (client, surface)
/// - `outbox_backpressured` triggers on either queued bytes or queued messages
/// - TCP / SSH flow control on the wire
fn surface_pacing_fps(client: &ClientState) -> f32 {
    let frame_bytes = client.avg_surface_frame_bytes.max(256.0);
    let bw = client.goodput_bps.max(client.delivery_bps);
    let sustainable = bw / frame_bytes;
    sustainable.min(client.display_fps).max(1.0)
}

fn surface_send_interval(client: &ClientState) -> Duration {
    Duration::from_secs_f64(1.0 / surface_pacing_fps(client).max(1.0) as f64)
}

fn advance_deadline(deadline: &mut Instant, now: Instant, interval: Duration) {
    let scheduled = deadline.checked_add(interval).unwrap_or(now + interval);
    *deadline = if scheduled + interval < now {
        now + interval
    } else {
        scheduled
    };
}

fn should_snapshot_pty(dirty: bool, needful: bool, synced_output: bool) -> bool {
    dirty && needful && !synced_output
}

fn enqueue_ready_frame(queue: &mut VecDeque<FrameState>, frame: FrameState) -> bool {
    if queue.len() >= READY_FRAME_QUEUE_CAP {
        return false;
    }
    queue.push_back(frame);
    true
}

fn pty_has_visual_update(pty: &Pty) -> bool {
    pty.dirty || !pty.ready_frames.is_empty() || !pty.byte_rx.is_empty()
}

/// Find the first `\x1b[?2026l` in `bytes`, handling sequences that span
/// the `prefix`/`bytes` boundary. Uses SIMD-accelerated memchr for the
/// initial ESC scan.
fn find_sync_output_end(prefix: &[u8], bytes: &[u8]) -> Option<usize> {
    if bytes.is_empty() {
        return None;
    }
    let needle = SYNC_OUTPUT_END;
    let nlen = needle.len();

    // Check for a match straddling the prefix/bytes boundary.
    if !prefix.is_empty() {
        let tail = if prefix.len() >= nlen - 1 {
            &prefix[prefix.len() - (nlen - 1)..]
        } else {
            prefix
        };
        let combined_len = tail.len() + bytes.len().min(nlen);
        if combined_len >= nlen {
            // Small stack buffer to check the boundary region.
            let mut buf = [0u8; 32]; // SYNC_OUTPUT_END is 8 bytes, so 32 is plenty
            let blen = combined_len.min(buf.len());
            let tlen = tail.len().min(blen);
            buf[..tlen].copy_from_slice(&tail[..tlen]);
            let rest = (blen - tlen).min(bytes.len());
            buf[tlen..tlen + rest].copy_from_slice(&bytes[..rest]);
            for i in 0..=(blen.saturating_sub(nlen)) {
                if &buf[i..i + nlen] == needle {
                    let end_in_bytes = (i + nlen).saturating_sub(tail.len());
                    if end_in_bytes > 0 && end_in_bytes <= bytes.len() {
                        return Some(end_in_bytes);
                    }
                }
            }
        }
    }

    // SIMD-scan for ESC (0x1b) then verify the full sequence.
    let mut offset = 0;
    while let Some(pos) = memchr::memchr(0x1b, &bytes[offset..]) {
        let abs = offset + pos;
        if abs + nlen <= bytes.len() && &bytes[abs..abs + nlen] == needle {
            return Some(abs + nlen);
        }
        offset = abs + 1;
    }
    None
}

fn update_sync_scan_tail(tail: &mut Vec<u8>, bytes: &[u8]) {
    if bytes.is_empty() {
        return;
    }
    tail.extend_from_slice(bytes);
    let keep = SYNC_OUTPUT_END.len().saturating_sub(1);
    if tail.len() > keep {
        let drop = tail.len() - keep;
        tail.drain(..drop);
    }
}

fn preview_deadline(client: &ClientState, pid: u16, now: Instant) -> Instant {
    client
        .preview_next_send_at
        .get(&pid)
        .copied()
        .unwrap_or(now)
}

fn client_has_due_preview(sess: &Session, client: &ClientState, now: Instant) -> bool {
    if client.lead.is_none() {
        return false;
    }
    client.subscriptions.iter().copied().any(|pid| {
        Some(pid) != client.lead
            && preview_deadline(client, pid, now) <= now
            && sess
                .ptys
                .get(&pid)
                .map(pty_has_visual_update)
                .unwrap_or(false)
    })
}

fn outbox_queued_frames(client: &ClientState) -> usize {
    client.outbox_queued_frames.load(Ordering::Relaxed)
}

fn outbox_queued_bytes(client: &ClientState) -> usize {
    client.outbox_queued_bytes.load(Ordering::Relaxed)
}

fn outbox_backpressured(client: &ClientState) -> bool {
    outbox_queued_frames(client) >= OUTBOX_SOFT_QUEUE_LIMIT_FRAMES
        || outbox_queued_bytes(client) >= OUTBOX_SOFT_QUEUE_LIMIT_BYTES
}

fn mark_outbox_drained(
    queued_frames: &Arc<AtomicUsize>,
    queued_bytes: &Arc<AtomicUsize>,
    bytes: usize,
) {
    let _ = queued_frames.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
        Some(value.saturating_sub(1))
    });
    let _ = queued_bytes.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
        Some(value.saturating_sub(bytes))
    });
}

fn try_send_outbox_tracked(
    tx: &mpsc::Sender<Vec<u8>>,
    queued_frames: &Arc<AtomicUsize>,
    queued_bytes: &Arc<AtomicUsize>,
    msg: Vec<u8>,
) -> Result<(), mpsc::error::TrySendError<Vec<u8>>> {
    let bytes = msg.len();
    match tx.try_send(msg) {
        Ok(()) => {
            queued_frames.fetch_add(1, Ordering::Relaxed);
            queued_bytes.fetch_add(bytes, Ordering::Relaxed);
            Ok(())
        }
        Err(err) => Err(err),
    }
}

fn try_send_outbox(
    client: &ClientState,
    msg: Vec<u8>,
) -> Result<(), mpsc::error::TrySendError<Vec<u8>>> {
    try_send_outbox_tracked(
        &client.tx,
        &client.outbox_queued_frames,
        &client.outbox_queued_bytes,
        msg,
    )
}

async fn send_outbox_tracked(
    tx: &mpsc::Sender<Vec<u8>>,
    queued_frames: &Arc<AtomicUsize>,
    queued_bytes: &Arc<AtomicUsize>,
    msg: Vec<u8>,
) -> Result<(), mpsc::error::SendError<Vec<u8>>> {
    let bytes = msg.len();
    tx.send(msg).await?;
    queued_frames.fetch_add(1, Ordering::Relaxed);
    queued_bytes.fetch_add(bytes, Ordering::Relaxed);
    Ok(())
}

fn can_send_preview(client: &ClientState, pid: u16, now: Instant) -> bool {
    window_open(client) && now >= preview_deadline(client, pid, now)
}

fn record_preview_send(client: &mut ClientState, pid: u16, now: Instant) {
    let mut deadline = client
        .preview_next_send_at
        .get(&pid)
        .copied()
        .unwrap_or(now);
    advance_deadline(&mut deadline, now, preview_send_interval(client));
    client.preview_next_send_at.insert(pid, deadline);
}

fn window_open(client: &ClientState) -> bool {
    !browser_backlog_blocked(client)
        && !outbox_backpressured(client)
        && client.inflight_frames.len() < target_frame_window(client)
        && client.inflight_bytes < target_byte_window(client)
}

/// Surface send window: only outbox backpressure.
///
/// Surface pacing is decoupled from terminal congestion control.  Surfaces
/// have their own time-based pacing (`surface_send_interval`) derived from
/// `display_fps` and `goodput_bps / avg_surface_frame_bytes`.  The shared
/// inflight window (`target_frame_window`, `target_byte_window`) is sized
/// for tiny terminal diffs and must not gate large video frames — doing so
/// causes the two streams to drag each other down, especially on SSH links
/// where the conservative bandwidth estimates are already fragile.
fn surface_window_open(client: &ClientState) -> bool {
    !outbox_backpressured(client)
}

fn lead_window_open(client: &ClientState, reserve_preview_slot: bool) -> bool {
    if !reserve_preview_slot || client.lead.is_none() {
        return window_open(client);
    }
    if browser_backlog_blocked(client) || outbox_backpressured(client) {
        return false;
    }
    let target_frames = target_frame_window(client);
    let reserve_frames = PREVIEW_FRAME_RESERVE.min(target_frames.saturating_sub(1));
    let frame_limit = target_frames.saturating_sub(reserve_frames).max(1);
    let reserve_bytes = client.avg_preview_frame_bytes.max(256.0).ceil() as usize;
    let byte_limit = target_byte_window(client)
        .saturating_sub(reserve_bytes)
        .max(client.avg_paced_frame_bytes.max(256.0).ceil() as usize);
    client.inflight_frames.len() < frame_limit && client.inflight_bytes < byte_limit
}

fn can_send_frame(client: &ClientState, now: Instant, reserve_preview_slot: bool) -> bool {
    lead_window_open(client, reserve_preview_slot) && now >= client.next_send_at
}

fn record_send(client: &mut ClientState, bytes: usize, now: Instant, paced: bool) {
    client.inflight_bytes += bytes;
    client.inflight_frames.push_back(InFlightFrame {
        sent_at: now,
        bytes,
        paced,
    });
    if paced {
        let interval = send_interval(client);
        advance_deadline(&mut client.next_send_at, now, interval);
    }
}

fn ewma_with_direction(old: f32, sample: f32, rise_alpha: f32, fall_alpha: f32) -> f32 {
    let alpha = if sample > old { rise_alpha } else { fall_alpha };
    old * (1.0 - alpha) + sample * alpha
}

fn window_saturated(client: &ClientState, inflight_frames: usize, inflight_bytes: usize) -> bool {
    let target_frames = target_frame_window(client);
    let target_bytes = target_byte_window(client);
    inflight_frames.saturating_mul(10) >= target_frames.saturating_mul(9)
        || inflight_bytes.saturating_mul(10) >= target_bytes.saturating_mul(9)
}

fn record_ack(client: &mut ClientState) {
    if let Some(frame) = client.inflight_frames.pop_front() {
        let prev_inflight_frames = client.inflight_frames.len() + 1;
        let prev_inflight_bytes = client.inflight_bytes;
        client.inflight_bytes = client.inflight_bytes.saturating_sub(frame.bytes);
        client.acked_bytes_since_log = client.acked_bytes_since_log.saturating_add(frame.bytes);
        let sample_ms = frame.sent_at.elapsed().as_secs_f32() * 1_000.0;
        client.rtt_ms = ewma_with_direction(client.rtt_ms, sample_ms, 0.125, 0.25);
        if client.min_rtt_ms > 0.0 {
            // Only update downward: min_rtt tracks the unloaded path RTT and
            // must not drift upward during congestion (queued RTT ≠ path RTT).
            client.min_rtt_ms = client.min_rtt_ms.min(sample_ms);
        } else {
            client.min_rtt_ms = sample_ms;
        }
        client.min_rtt_ms = client.min_rtt_ms.max(0.5);
        let sample_bps = frame.bytes as f32 / sample_ms.max(1.0e-3) * 1_000.0;
        client.delivery_bps = ewma_with_direction(client.delivery_bps, sample_bps, 0.5, 0.125);
        client.avg_frame_bytes =
            ewma_with_direction(client.avg_frame_bytes, frame.bytes as f32, 0.5, 0.125);
        if frame.paced {
            client.avg_paced_frame_bytes =
                ewma_with_direction(client.avg_paced_frame_bytes, frame.bytes as f32, 0.5, 0.125);
        } else {
            client.avg_preview_frame_bytes = ewma_with_direction(
                client.avg_preview_frame_bytes,
                frame.bytes as f32,
                0.5,
                0.125,
            );
        }
        let frame_ms = 1_000.0 / browser_pacing_fps(client).max(1.0);
        let path_rtt = path_rtt_ms(client);
        let likely_window_limited =
            window_saturated(client, prev_inflight_frames, prev_inflight_bytes);
        client.goodput_window_bytes = client.goodput_window_bytes.saturating_add(frame.bytes);
        let now = Instant::now();
        let goodput_elapsed = now
            .duration_since(client.goodput_window_start)
            .as_secs_f32();
        if goodput_elapsed >= 0.02 {
            let sample_goodput = client.goodput_window_bytes as f32 / goodput_elapsed.max(1.0e-3);
            if likely_window_limited || client.browser_backlog_frames > 0 {
                let prev_goodput_sample = if client.last_goodput_sample_bps > 0.0 {
                    client.last_goodput_sample_bps
                } else {
                    sample_goodput
                };
                let jitter_sample = (sample_goodput - prev_goodput_sample).abs();
                client.goodput_bps =
                    ewma_with_direction(client.goodput_bps, sample_goodput, 0.5, 0.125);
                // Only update jitter from windows with at least 2 frames.
                // Single-frame windows are pure measurement noise (0 or 1
                // frame per 25 ms is a Bernoulli trial, not a congestion
                // signal) and inflate jitter_bps, which in turn depresses
                // bandwidth_floor_bps and causes pacing to stall.
                let min_reliable = (client.avg_paced_frame_bytes.max(256.0) * 2.0) as usize;
                if client.goodput_window_bytes >= min_reliable {
                    client.goodput_jitter_bps =
                        ewma_with_direction(client.goodput_jitter_bps, jitter_sample, 0.5, 0.125);
                    let jitter_decay = if browser_ready(client) && sample_ms < path_rtt * 3.0 {
                        0.90
                    } else {
                        0.98
                    };
                    client.max_goodput_jitter_bps =
                        (client.max_goodput_jitter_bps * jitter_decay).max(jitter_sample);
                    // Cap jitter at 45% of goodput so jitter_ratio can never
                    // exceed 0.45 from measurement noise alone.  Real congestion
                    // will still drive goodput_bps down and widen the window.
                    client.max_goodput_jitter_bps =
                        client.max_goodput_jitter_bps.min(client.goodput_bps * 0.45);
                } else {
                    // Thin sample: gently decay jitter rather than updating it.
                    client.goodput_jitter_bps *= 0.9;
                    client.max_goodput_jitter_bps *= 0.95;
                }
                // Sticky-high: never let last_goodput_sample_bps drop abruptly.
                // A sudden drop (e.g. 1-frame window following a 2-frame window)
                // inflates jitter_sample on the next cycle, collapsing probe_frames.
                client.last_goodput_sample_bps =
                    (client.last_goodput_sample_bps * 0.99).max(sample_goodput);
            } else {
                // When the path is underfilled, ACK cadence mostly measures our
                // own pacing rather than network capacity.  Use a fall alpha
                // proportional to estimation error: when the estimate is 10x+
                // the sample, converge aggressively; when close, stay gentle.
                let ratio = client.goodput_bps / sample_goodput.max(1.0);
                let fall_alpha = if ratio > 10.0 {
                    0.5
                } else if ratio > 3.0 {
                    0.25
                } else {
                    0.03
                };
                client.goodput_bps =
                    ewma_with_direction(client.goodput_bps, sample_goodput, 0.5, fall_alpha);
                client.goodput_jitter_bps *= 0.5;
                client.max_goodput_jitter_bps *= 0.9;
                client.last_goodput_sample_bps =
                    (client.last_goodput_sample_bps * 0.99).max(sample_goodput);
            }
            client.goodput_window_bytes = 0;
            client.goodput_window_start = now;
        }
        let queue_baseline_ms = if throughput_limited(client) {
            window_rtt_ms(client)
        } else {
            path_rtt
        };
        let queue_delay_ms = (sample_ms - queue_baseline_ms).max(0.0);
        let max_probe_frames = (browser_pacing_fps(client) * 0.125).max(4.0);
        let jitter_ratio = client.max_goodput_jitter_bps / client.goodput_bps.max(1.0);
        let low_delay_frames = if throughput_limited(client) { 2.0 } else { 8.0 };
        let high_delay_frames = if throughput_limited(client) {
            4.0
        } else {
            12.0
        };
        if likely_window_limited
            && queue_delay_ms <= frame_ms * low_delay_frames
            && jitter_ratio < 0.25
        {
            client.probe_frames = (client.probe_frames + 1.0).min(max_probe_frames);
        } else if !likely_window_limited
            && browser_ready(client)
            && queue_delay_ms <= frame_ms * 2.0
            && jitter_ratio < 0.25
        {
            client.probe_frames = (client.probe_frames + 0.25).min(max_probe_frames * 0.5);
        } else if queue_delay_ms > frame_ms * high_delay_frames || jitter_ratio > 0.5 {
            client.probe_frames = (client.probe_frames * 0.5).max(1.0);
        } else if queue_delay_ms > frame_ms * 2.0 || !browser_ready(client) {
            client.probe_frames = (client.probe_frames - 0.5).max(0.0);
        }
    } else {
        client.inflight_bytes = 0;
    }
}

/// Process a surface ACK.  Feeds delivery_bps and goodput estimates (same
/// pipe) from the surface inflight queue.  Does NOT update rtt_ms / min_rtt_ms
/// — surface frames are large and their wall-clock delivery time is dominated
/// by serialization and wire transfer, not network latency.  Feeding those
/// samples into the shared RTT inflates it by orders of magnitude and
/// destabilises terminal pacing and congestion control.
fn record_surface_ack(client: &mut ClientState) {
    if let Some(frame) = client.surface_inflight_frames.pop_front() {
        client.acked_bytes_since_log = client.acked_bytes_since_log.saturating_add(frame.bytes);

        let sample_ms = frame.sent_at.elapsed().as_secs_f32() * 1_000.0;

        // Shared delivery rate (bandwidth, not latency — safe to update).
        let sample_bps = frame.bytes as f32 / sample_ms.max(1.0e-3) * 1_000.0;
        client.delivery_bps = ewma_with_direction(client.delivery_bps, sample_bps, 0.5, 0.125);

        // Shared goodput window — accumulate bytes, flush periodically.
        // Surface traffic at display_fps is sustained, so always use the
        // window-limited EWMA parameters (rise 0.5, fall 0.125).  No
        // jitter tracking — jitter is a terminal congestion-control signal
        // and large keyframe/P-frame variance would poison it.
        client.goodput_window_bytes = client.goodput_window_bytes.saturating_add(frame.bytes);
        let now = Instant::now();
        let goodput_elapsed = now
            .duration_since(client.goodput_window_start)
            .as_secs_f32();
        if goodput_elapsed >= 0.02 {
            let sample_goodput = client.goodput_window_bytes as f32 / goodput_elapsed.max(1.0e-3);
            client.goodput_bps =
                ewma_with_direction(client.goodput_bps, sample_goodput, 0.5, 0.125);
            client.last_goodput_sample_bps =
                (client.last_goodput_sample_bps * 0.99).max(sample_goodput);
            client.goodput_window_bytes = 0;
            client.goodput_window_start = now;
        }
    }
}

fn reset_inflight(client: &mut ClientState) {
    client.inflight_bytes = 0;
    client.inflight_frames.clear();
    client.next_send_at = Instant::now();
    client.browser_backlog_frames = 0;
    client.browser_ack_ahead_frames = 0;
}

fn is_unset_view_size(rows: u16, cols: u16) -> bool {
    rows == 0 && cols == 0
}

fn subscribe_client_to(client: &mut ClientState, pty_id: u16) {
    if client.subscriptions.insert(pty_id) {
        client.last_sent.remove(&pty_id);
        client.preview_next_send_at.remove(&pty_id);
    }
}

fn unsubscribe_client_from(client: &mut ClientState, pty_id: u16) -> bool {
    let removed_sub = client.subscriptions.remove(&pty_id);
    client.last_sent.remove(&pty_id);
    client.preview_next_send_at.remove(&pty_id);
    client.scroll_offsets.remove(&pty_id);
    client.scroll_caches.remove(&pty_id);
    let removed_view = client.view_sizes.remove(&pty_id).is_some();
    if client.lead == Some(pty_id) {
        client.lead = None;
    }
    removed_sub || removed_view
}

fn update_client_scroll_state(client: &mut ClientState, pty_id: u16, next_offset: usize) -> bool {
    let prev_offset = client.scroll_offsets.get(&pty_id).copied().unwrap_or(0);
    if prev_offset == next_offset {
        return false;
    }

    if prev_offset == 0 && next_offset > 0 {
        client.scroll_caches.insert(
            pty_id,
            client.last_sent.get(&pty_id).cloned().unwrap_or_default(),
        );
    } else if prev_offset > 0
        && next_offset == 0
        && let Some(cache) = client.scroll_caches.remove(&pty_id)
    {
        if cache.rows() > 0 && cache.cols() > 0 {
            client.last_sent.insert(pty_id, cache);
        } else {
            client.last_sent.remove(&pty_id);
        }
    }

    if next_offset > 0 {
        client.scroll_offsets.insert(pty_id, next_offset);
    } else {
        client.scroll_offsets.remove(&pty_id);
    }
    reset_inflight(client);
    true
}

struct Session {
    ptys: HashMap<u16, Pty>,
    compositor: Option<SharedCompositor>,
    next_client_id: u64,
    next_compositor_id: u16,
    next_pty_id: u16,
    tick_fires: u32,
    tick_snaps: u32,
    surface_commits: u32,
    surface_encodes: u32,
    surface_encode_bytes: u64,
    surface_frames_sent: u32,
    last_ping: Instant,
    clients: HashMap<u64, ClientState>,
}

struct SearchResultRow {
    pty_id: u16,
    score: u32,
    primary_source: u8,
    matched_sources: u8,
    context: String,
    scroll_offset: Option<usize>,
}

struct TickOutcome {
    next_deadline: Option<Instant>,
}

impl Session {
    fn new() -> Self {
        Self {
            ptys: HashMap::new(),
            compositor: None,
            next_client_id: 1,
            next_compositor_id: 1,
            next_pty_id: 1,
            clients: HashMap::new(),
            tick_fires: 0,
            tick_snaps: 0,
            surface_commits: 0,
            surface_encodes: 0,
            surface_encode_bytes: 0,
            last_ping: Instant::now(),
            surface_frames_sent: 0,
        }
    }

    fn ensure_compositor(
        &mut self,
        verbose: bool,
        event_notify: Arc<dyn Fn() + Send + Sync>,
        gpu_device: &str,
    ) -> &str {
        if self.compositor.is_none() {
            let session_id = self.next_compositor_id;
            self.next_compositor_id = self.next_compositor_id.wrapping_add(1);
            // Create the epoch before spawning anything so audio and video
            // share the same time origin for A/V sync.
            let created_at = Instant::now();
            let handle = blit_compositor::spawn_compositor(verbose, event_notify, gpu_device);
            #[cfg(unix)]
            let audio_pipeline = {
                let audio_disabled = std::env::var("BLIT_AUDIO")
                    .map(|v| v == "0")
                    .unwrap_or(false);
                if !audio_disabled && audio::pipewire_available() {
                    let runtime_dir = std::path::Path::new(&handle.socket_name)
                        .parent()
                        .unwrap_or(std::path::Path::new("/tmp"));
                    let bitrate = std::env::var("BLIT_AUDIO_BITRATE")
                        .ok()
                        .and_then(|v| v.parse::<i32>().ok())
                        .unwrap_or(0);
                    // Wrap in block_in_place so the thread::sleep calls
                    // inside spawn() don't stall the tokio runtime.
                    tokio::task::block_in_place(|| {
                        match audio::AudioPipeline::spawn(
                            runtime_dir,
                            session_id,
                            bitrate,
                            verbose,
                            created_at,
                        ) {
                            Ok(pipeline) => {
                                if verbose {
                                    eprintln!(
                                        "[audio] pipeline started, PULSE_SERVER={}",
                                        pipeline.pulse_server_path(),
                                    );
                                }
                                Some(pipeline)
                            }
                            Err(e) => {
                                eprintln!("[audio] failed to start pipeline: {e}");
                                None
                            }
                        }
                    })
                } else {
                    if verbose && !audio_disabled {
                        eprintln!("[audio] PipeWire not available, audio disabled");
                    }
                    None
                }
            };

            self.compositor = Some(SharedCompositor {
                handle,
                surfaces: HashMap::new(),
                last_pixels: HashMap::new(),
                last_frame_request: HashMap::new(),
                created_at,
                pixel_generation: 0,
                last_blanket_frame_request: Instant::now(),
                last_configured_size: HashMap::new(),
                #[cfg(unix)]
                audio_pipeline,
                #[cfg(unix)]
                audio_session_id: session_id,
                #[cfg(unix)]
                last_audio_restart: None,
            });
        }
        &self.compositor.as_ref().unwrap().handle.socket_name
    }

    /// Returns the `PULSE_SERVER` path if the audio pipeline is active.
    #[cfg(unix)]
    fn pulse_server_path(&self) -> Option<String> {
        self.compositor
            .as_ref()
            .and_then(|cs| cs.audio_pipeline.as_ref())
            .map(|ap| ap.pulse_server_path())
    }

    /// Returns the `PIPEWIRE_REMOTE` path if the audio pipeline is active.
    #[cfg(unix)]
    fn pipewire_remote_path(&self) -> Option<String> {
        self.compositor
            .as_ref()
            .and_then(|cs| cs.audio_pipeline.as_ref())
            .map(|ap| ap.pipewire_remote_path())
    }

    fn allocate_pty_id(&mut self, max_ptys: usize) -> Option<u16> {
        if max_ptys > 0 && self.ptys.len() >= max_ptys {
            return None;
        }
        let start = self.next_pty_id;
        let mut id = start;
        loop {
            if !self.ptys.contains_key(&id) {
                self.next_pty_id = if id == u16::MAX { 1 } else { id + 1 };
                return Some(id);
            }
            id = if id == u16::MAX { 1 } else { id + 1 };
            if id == start {
                return None;
            }
        }
    }

    fn send_to_all(&self, msg: &[u8]) {
        for c in self.clients.values() {
            let _ = try_send_outbox(c, msg.to_vec());
        }
    }

    fn mediated_size_for_pty(&self, pty_id: u16) -> Option<(u16, u16)> {
        let mut min_rows: Option<u16> = None;
        let mut min_cols: Option<u16> = None;
        for c in self.clients.values() {
            if let Some((r, cols)) = c.view_sizes.get(&pty_id).copied() {
                min_rows = Some(min_rows.map_or(r, |m: u16| m.min(r)));
                min_cols = Some(min_cols.map_or(cols, |m: u16| m.min(cols)));
            }
        }
        match (min_rows, min_cols) {
            (Some(r), Some(c)) => Some((r.max(1), c.max(1))),
            _ => None,
        }
    }

    fn resize_pty(&mut self, pty_id: u16, rows: u16, cols: u16) -> bool {
        let pty = match self.ptys.get_mut(&pty_id) {
            Some(p) => p,
            None => return false,
        };
        let (cur_rows, cur_cols) = pty.driver.size();
        if cur_rows == rows && cur_cols == cols {
            return false;
        }
        pty.ready_frames.clear();
        pty.driver.resize(rows, cols);
        pty.mark_dirty();
        for c in self.clients.values_mut() {
            if c.subscriptions.contains(&pty_id) {
                c.last_sent.remove(&pty_id);
            }
            if c.scroll_caches.remove(&pty_id).is_some() {
                reset_inflight(c);
            }
        }
        if !pty.exited {
            pty::resize_pty_os(&pty.handle, rows, cols);
        }
        true
    }

    fn resize_ptys_to_mediated_sizes<I>(&mut self, pty_ids: I) -> bool
    where
        I: IntoIterator<Item = u16>,
    {
        let mut changed = false;
        let mut seen = HashSet::new();
        for pty_id in pty_ids {
            if !seen.insert(pty_id) {
                continue;
            }
            if let Some((rows, cols)) = self.mediated_size_for_pty(pty_id) {
                changed |= self.resize_pty(pty_id, rows, cols);
            }
        }
        changed
    }

    // ------------------------------------------------------------------
    // Surface sizing — same consumer-tracking model as PTY sizing.
    // Each client reports how large it can display a surface; the server
    // picks min(width), min(height) across all clients and configures the
    // compositor accordingly.
    // ------------------------------------------------------------------

    /// Returns (width, height, scale_120) mediated across all clients.
    /// Resolution: min across clients.  DPI: max across clients.
    fn mediated_size_for_surface(
        &self,
        surface_id: u16,
        max: Option<(u16, u16)>,
    ) -> Option<(u16, u16, u16)> {
        let mut min_w: Option<u16> = None;
        let mut min_h: Option<u16> = None;
        let mut max_scale: u16 = 0;
        for c in self.clients.values() {
            if let Some(&(w, h, s)) = c.surface_view_sizes.get(&surface_id) {
                min_w = Some(min_w.map_or(w, |m: u16| m.min(w)));
                min_h = Some(min_h.map_or(h, |m: u16| m.min(h)));
                max_scale = max_scale.max(s);
            }
        }
        match (min_w, min_h) {
            (Some(w), Some(h)) => {
                let (w, h) = (w.max(1), h.max(1));
                let (w, h) = if let Some((mw, mh)) = max {
                    (w.min(mw), h.min(mh))
                } else {
                    (w, h)
                };
                Some((w, h, max_scale))
            }
            _ => None,
        }
    }

    fn resize_surface(&mut self, surface_id: u16, width: u16, height: u16, scale_120: u16) -> bool {
        let cs = match self.compositor.as_mut() {
            Some(cs) => cs,
            None => return false,
        };
        // Dedup against the last *requested* dimensions, not the composited
        // output dimensions (`info.width`/`info.height`).  The composited
        // output may be smaller when the Wayland client sets xdg_geometry
        // (e.g. Chromium excludes the title bar), so comparing against it
        // would cause every resize to look like a change, flooding the
        // compositor with redundant configures and re-creating the encoder
        // (keyframe) on every tick during a drag-resize.
        if let Some(&(lw, lh, ls)) = cs.last_configured_size.get(&surface_id)
            && lw == width
            && lh == height
            && ls == scale_120
        {
            return false;
        }
        cs.last_configured_size
            .insert(surface_id, (width, height, scale_120));
        let _ = cs.handle.command_tx.send(CompositorCommand::SurfaceResize {
            surface_id,
            width,
            height,
            scale_120,
        });
        true
    }

    fn resize_surfaces_to_mediated_sizes<I>(
        &mut self,
        surface_ids: I,
        encoder_preferences: &[SurfaceEncoderPreference],
    ) where
        I: IntoIterator<Item = u16>,
    {
        let max = SurfaceEncoderPreference::max_dimensions_for_list(encoder_preferences);
        let mut seen = HashSet::new();
        for sid in surface_ids {
            if !seen.insert(sid) {
                continue;
            }
            if let Some((w, h, scale_120)) = self.mediated_size_for_surface(sid, max) {
                self.resize_surface(sid, w, h, scale_120);
            }
        }
    }

    fn pty_list_msg(&self) -> Vec<u8> {
        let mut msg = vec![S2C_LIST];
        let count = self.ptys.len() as u16;
        msg.extend_from_slice(&count.to_le_bytes());
        let mut ids: Vec<u16> = self.ptys.keys().copied().collect();
        ids.sort();
        for id in ids {
            let pty = &self.ptys[&id];
            let tag = pty.tag.as_bytes();
            msg.extend_from_slice(&id.to_le_bytes());
            msg.extend_from_slice(&(tag.len() as u16).to_le_bytes());
            msg.extend_from_slice(tag);
            let cmd = pty.command.as_deref().unwrap_or("").as_bytes();
            msg.extend_from_slice(&(cmd.len() as u16).to_le_bytes());
            msg.extend_from_slice(cmd);
        }
        msg
    }

    fn surface_list_msg(&self) -> Vec<u8> {
        let cs = match self.compositor.as_ref() {
            Some(cs) => cs,
            None => {
                let mut msg = vec![S2C_SURFACE_LIST];
                msg.extend_from_slice(&0u16.to_le_bytes());
                return msg;
            }
        };
        let mut msg = vec![S2C_SURFACE_LIST];
        let count = cs.surfaces.len() as u16;
        msg.extend_from_slice(&count.to_le_bytes());
        let mut ids: Vec<u16> = cs.surfaces.keys().copied().collect();
        ids.sort();
        for id in ids {
            let info = &cs.surfaces[&id];
            let title = info.title.as_bytes();
            let app_id = info.app_id.as_bytes();
            msg.extend_from_slice(&info.surface_id.to_le_bytes());
            msg.extend_from_slice(&info.parent_id.to_le_bytes());
            msg.extend_from_slice(&info.width.to_le_bytes());
            msg.extend_from_slice(&info.height.to_le_bytes());
            msg.extend_from_slice(&(title.len() as u16).to_le_bytes());
            msg.extend_from_slice(title);
            msg.extend_from_slice(&(app_id.len() as u16).to_le_bytes());
            msg.extend_from_slice(app_id);
        }
        msg
    }
}

struct AppStateInner {
    config: Config,
    session: Mutex<Session>,
    pty_fds: PtyFds,
    delivery_notify: Arc<Notify>,
    /// Signalled when a client sends C2S_QUIT to initiate server shutdown.
    shutdown_notify: Arc<Notify>,
    /// Tracks the number of currently connected clients for enforcing
    /// `config.max_connections`.
    active_connections: std::sync::atomic::AtomicUsize,
}

type AppState = Arc<AppStateInner>;

fn nudge_delivery(state: &AppState) {
    state.delivery_notify.notify_one();
}

#[cfg(unix)]
#[allow(dead_code)]
fn spawn_compositor_child(
    command: &str,
    argv: Option<&[&str]>,
    wayland_socket: &str,
    dir: Option<&str>,
) -> libc::pid_t {
    use std::ffi::CString;
    let pid = unsafe { libc::fork() };
    if pid == 0 {
        if let Some(d) = dir {
            let c_dir = CString::new(d).unwrap();
            unsafe {
                libc::chdir(c_dir.as_ptr());
            }
        }
        unsafe {
            let wd_path = std::path::Path::new(wayland_socket);
            if let Some(dir) = wd_path.parent() {
                let xdg = std::env::var_os("XDG_RUNTIME_DIR");
                let needs_update = match &xdg {
                    Some(x) => std::path::Path::new(x) != dir,
                    None => true,
                };
                if needs_update {
                    std::env::set_var("XDG_RUNTIME_DIR", dir);
                }
            }
            std::env::set_var("WAYLAND_DISPLAY", wayland_socket);
            std::env::remove_var("DISPLAY");
            std::env::remove_var("DBUS_SESSION_BUS_ADDRESS");
            std::env::remove_var("DBUS_SYSTEM_BUS_ADDRESS");
        }
        if let Some(args) = argv {
            let prog = CString::new(args[0]).unwrap();
            let c_args: Vec<CString> = args.iter().map(|a| CString::new(*a).unwrap()).collect();
            let c_ptrs: Vec<*const libc::c_char> = c_args
                .iter()
                .map(|a| a.as_ptr())
                .chain(std::iter::once(std::ptr::null()))
                .collect();
            unsafe {
                libc::execvp(prog.as_ptr(), c_ptrs.as_ptr());
            }
        } else {
            let prog = CString::new(command).unwrap();
            let c_ptrs = [prog.as_ptr(), std::ptr::null()];
            unsafe {
                libc::execvp(prog.as_ptr(), c_ptrs.as_ptr());
                libc::_exit(1);
            }
        }
    }
    pid
}

/// Map xterm-256 color index to (r, g, b) in 16-bit per channel.
fn xterm256_color(idx: u8) -> (u16, u16, u16) {
    // Standard 16 colors (0-15)
    const BASE16: [(u8, u8, u8); 16] = [
        (0, 0, 0),
        (128, 0, 0),
        (0, 128, 0),
        (128, 128, 0),
        (0, 0, 128),
        (128, 0, 128),
        (0, 128, 128),
        (192, 192, 192),
        (128, 128, 128),
        (255, 0, 0),
        (0, 255, 0),
        (255, 255, 0),
        (0, 0, 255),
        (255, 0, 255),
        (0, 255, 255),
        (255, 255, 255),
    ];
    let (r8, g8, b8) = if idx < 16 {
        BASE16[idx as usize]
    } else if idx < 232 {
        // 6x6x6 color cube (indices 16-231)
        let n = idx - 16;
        let ri = n / 36;
        let gi = (n % 36) / 6;
        let bi = n % 6;
        let to_val = |v: u8| if v == 0 { 0u8 } else { 55 + 40 * v };
        (to_val(ri), to_val(gi), to_val(bi))
    } else {
        // Grayscale ramp (indices 232-255)
        let v = 8 + 10 * (idx - 232);
        (v, v, v)
    };
    // Scale 8-bit to 16-bit (0xFF -> 0xFFFF)
    let scale = |v: u8| (v as u16) << 8 | v as u16;
    (scale(r8), scale(g8), scale(b8))
}
fn parse_terminal_queries(data: &[u8], size: (u16, u16), cursor: (u16, u16)) -> Vec<String> {
    const DA1_RESPONSE: &[u8] = b"\x1b[?64;1;2;6;9;15;18;21;22c";

    let mut results = Vec::new();
    let mut i = 0;
    while i < data.len() {
        if data[i] != 0x1b || i + 1 >= data.len() {
            i += 1;
            continue;
        }

        // Handle OSC sequences: \x1b] ... (ST or BEL)
        if data[i + 1] == b']' {
            let osc_start = i + 2;
            // Find the terminator: BEL (\x07) or ST (\x1b\\)
            let mut end = osc_start;
            while end < data.len() {
                if data[end] == 0x07 {
                    break;
                }
                if data[end] == 0x1b && end + 1 < data.len() && data[end + 1] == b'\\' {
                    break;
                }
                end += 1;
            }
            if end < data.len() {
                let payload = &data[osc_start..end];
                // OSC 11 ; ? — query background color
                if payload == b"11;?" {
                    // Respond with dark background (rgb:0000/0000/0000)
                    results.push("\x1b]11;rgb:0000/0000/0000\x1b\\".into());
                }
                // OSC 10 ; ? — query foreground color
                else if payload == b"10;?" {
                    results.push("\x1b]10;rgb:ffff/ffff/ffff\x1b\\".into());
                }
                // OSC 4 ; N ; ? — query palette color N
                else if payload.starts_with(b"4;") && payload.ends_with(b";?") {
                    let idx_bytes = &payload[2..payload.len() - 2];
                    if let Ok(idx_str) = std::str::from_utf8(idx_bytes)
                        && let Ok(idx) = idx_str.parse::<u8>()
                    {
                        let (r, g, b) = xterm256_color(idx);
                        results.push(format!("\x1b]4;{idx};rgb:{r:04x}/{g:04x}/{b:04x}\x1b\\"));
                    }
                }
                i = end + if data[end] == 0x07 { 1 } else { 2 };
                continue;
            }
            i = end;
            continue;
        }

        // Handle CSI sequences: \x1b[ ...
        if i + 2 >= data.len() || data[i + 1] != b'[' {
            i += 1;
            continue;
        }
        i += 2;
        let has_q = i < data.len() && data[i] == b'?';
        if has_q {
            i += 1;
        }
        let param_start = i;
        while i < data.len() && (data[i].is_ascii_digit() || data[i] == b';') {
            i += 1;
        }
        if i >= data.len() {
            break;
        }
        let final_byte = data[i];
        let params = &data[param_start..i];
        i += 1;
        if has_q {
            continue;
        }
        let resp: Option<String> = match final_byte {
            b'c' if params.is_empty() || params == b"0" => {
                Some(String::from_utf8_lossy(DA1_RESPONSE).into_owned())
            }
            b'n' if params == b"6" => Some(format!("\x1b[{};{}R", cursor.0 + 1, cursor.1 + 1)),
            b'n' if params == b"5" => Some("\x1b[0n".into()),
            b't' if params == b"18" => {
                let (rows, cols) = size;
                Some(format!("\x1b[8;{rows};{cols}t"))
            }
            b't' if params == b"14" => {
                let (rows, cols) = size;
                Some(format!("\x1b[4;{};{}t", rows * 16, cols * 8))
            }
            _ => None,
        };
        if let Some(r) = resp {
            results.push(r);
        }
    }
    results
}

async fn cleanup_pty_internal(pty_id: u16, state: &AppState) {
    state.pty_fds.write().unwrap().remove(&pty_id);
    let mut sess = state.session.lock().await;
    if let Some(pty) = sess.ptys.get_mut(&pty_id) {
        if pty.exited {
            return;
        }
        pty.exited = true;
        pty::close_pty(&pty.handle);
        pty.exit_status = pty::collect_exit_status(&pty.handle);
        pty.mark_dirty();
        let msg = blit_remote::msg_exited(pty_id, pty.exit_status);
        sess.send_to_all(&msg);
    }
}

fn take_snapshot(pty: &mut Pty) -> FrameState {
    if pty.lflag_last.elapsed() >= Duration::from_millis(250) {
        pty.lflag_cache = pty::pty_lflag(&pty.handle);
        pty.lflag_last = Instant::now();
    }
    let (echo, icanon) = pty.lflag_cache;
    pty.driver.snapshot(echo, icanon)
}

fn build_scrollback_update(
    pty: &mut Pty,
    id: u16,
    offset: usize,
    prev_frame: &FrameState,
) -> Option<(Vec<u8>, FrameState)> {
    let frame = pty.driver.scrollback_frame(offset);
    let msg = build_update_msg(id, &frame, prev_frame);
    msg.map(|m| (m, frame))
}

fn build_search_results_msg(request_id: u16, results: &[SearchResultRow]) -> Vec<u8> {
    let count = results.len().min(u16::MAX as usize);
    let payload_bytes: usize = results[..count]
        .iter()
        .map(|result| 14 + result.context.len().min(u16::MAX as usize))
        .sum();
    let mut msg = Vec::with_capacity(5 + payload_bytes);
    msg.push(S2C_SEARCH_RESULTS);
    msg.extend_from_slice(&request_id.to_le_bytes());
    msg.extend_from_slice(&(count as u16).to_le_bytes());
    for result in &results[..count] {
        msg.extend_from_slice(&result.pty_id.to_le_bytes());
        msg.extend_from_slice(&result.score.to_le_bytes());
        msg.push(result.primary_source);
        msg.push(result.matched_sources);
        let scroll_offset = result
            .scroll_offset
            .map(|offset| offset.min(u32::MAX as usize - 1) as u32)
            .unwrap_or(u32::MAX);
        msg.extend_from_slice(&scroll_offset.to_le_bytes());
        let context = result.context.as_bytes();
        let context_len = context.len().min(u16::MAX as usize);
        msg.extend_from_slice(&(context_len as u16).to_le_bytes());
        msg.extend_from_slice(&context[..context_len]);
    }
    msg
}

enum SendOutcome {
    NoChange,
    Sent,
    Backpressured,
}

fn try_send_update(
    client: &mut ClientState,
    pid: u16,
    current: FrameState,
    msg: Option<Vec<u8>>,
    now: Instant,
    paced: bool,
) -> SendOutcome {
    let Some(msg) = msg else {
        return SendOutcome::NoChange;
    };
    let bytes = msg.len();
    if try_send_outbox(client, msg).is_ok() {
        client.last_sent.insert(pid, current);
        record_send(client, bytes, now, paced);
        client.frames_sent = client.frames_sent.wrapping_add(1);
        SendOutcome::Sent
    } else {
        // Outbox full — the sender can't keep up.  Advance last_sent to
        // the current frame so the NEXT diff is small (only changes since
        // now), effectively dropping this intermediate state.  Without
        // this, backpressure causes the tick to re-dirty the PTY, building
        // ever-larger diffs that make the backlog worse.
        client.last_sent.insert(pid, current);
        SendOutcome::Backpressured
    }
}

pub async fn run(config: Config) {
    let state: AppState = Arc::new(AppStateInner {
        config,
        session: Mutex::new(Session::new()),
        pty_fds: Arc::new(std::sync::RwLock::new(HashMap::new())),
        delivery_notify: Arc::new(Notify::new()),
        shutdown_notify: Arc::new(Notify::new()),
        active_connections: std::sync::atomic::AtomicUsize::new(0),
    });

    // Start the compositor eagerly so it is ready before any client
    // connects or any terminal is created.
    if !state.config.skip_compositor {
        let notify = state.delivery_notify.clone();
        let event_notify = Arc::new(move || notify.notify_one()) as Arc<dyn Fn() + Send + Sync>;
        let mut sess = state.session.lock().await;
        sess.ensure_compositor(
            state.config.verbose,
            event_notify,
            &state.config.vaapi_device,
        );
    }

    let delivery_state = state.clone();
    tokio::spawn(async move {
        let mut next_deadline: Option<Instant> = None;
        loop {
            if let Some(deadline) = next_deadline {
                tokio::select! {
                    _ = delivery_state.delivery_notify.notified() => {}
                    _ = tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)) => {}
                }
            } else {
                delivery_state.delivery_notify.notified().await;
            }
            let outcome = tick(&delivery_state).await;
            next_deadline = outcome.next_deadline;
        }
    });

    tokio::spawn(async {
        loop {
            tokio::time::sleep(Duration::from_secs(5)).await;
            pty::reap_zombies();
        }
    });

    #[cfg(unix)]
    if let Some(channel_fd) = state.config.fd_channel {
        ipc::run_fd_channel(channel_fd, state).await;
        return;
    }

    #[cfg(unix)]
    let listener = {
        if let Some(l) = IpcListener::from_systemd_fd(state.config.verbose) {
            l
        } else {
            IpcListener::bind(&state.config.ipc_path, state.config.verbose)
        }
    };
    #[cfg(not(unix))]
    let mut listener = IpcListener::bind(&state.config.ipc_path, state.config.verbose);

    // Broadcast S2C_QUIT on SIGTERM / SIGINT so clients can reconnect promptly
    // instead of waiting for a transport-level timeout.
    {
        let state = state.clone();
        tokio::spawn(async move {
            #[cfg(unix)]
            {
                use tokio::signal::unix::{SignalKind, signal};
                let mut sigterm = signal(SignalKind::terminate()).expect("signal handler");
                let mut sigint = signal(SignalKind::interrupt()).expect("signal handler");
                tokio::select! {
                    _ = sigterm.recv() => {}
                    _ = sigint.recv() => {}
                }
            }
            #[cfg(not(unix))]
            {
                let _ = tokio::signal::ctrl_c().await;
            }
            let sess = state.session.lock().await;
            sess.send_to_all(&[S2C_QUIT]);
            drop(sess);
            state.shutdown_notify.notify_waiters();
        });
    }

    let shutdown = state.shutdown_notify.clone();
    loop {
        let stream = tokio::select! {
            result = listener.accept() => match result {
                Ok(s) => s,
                Err(e) => {
                    eprintln!("accept error: {e}");
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    continue;
                }
            },
            _ = shutdown.notified() => break,
        };
        let max = state.config.max_connections;
        if max > 0 {
            let current = state
                .active_connections
                .load(std::sync::atomic::Ordering::Relaxed);
            if current >= max {
                eprintln!("max connections ({max}) reached, rejecting");
                drop(stream);
                continue;
            }
        }
        state
            .active_connections
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let state = state.clone();
        tokio::spawn(async move {
            handle_client(stream, state.clone()).await;
            state
                .active_connections
                .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
        });
    }
    // Brief grace period for S2C_QUIT to reach clients before the process exits.
    tokio::time::sleep(Duration::from_millis(100)).await;
}

/// Minimum interval between blanket RequestFrame rounds.  Keeps video
/// players (mpv) and browsers ticking even when no client is consuming
/// frames.  Also used as the maximum tick-loop sleep so the loop never
/// blocks longer than this.  33 ms ≈ 30 Hz.
const BLANKET_FRAME_INTERVAL: Duration = Duration::from_millis(33);

async fn tick(state: &AppState) -> TickOutcome {
    let mut sess = state.session.lock().await;
    sess.tick_fires += 1;
    let mut next_deadline: Option<Instant> = None;
    let now = Instant::now();

    // Application-level keepalive.
    let ping_interval = state.config.ping_interval;
    if !ping_interval.is_zero() && now.duration_since(sess.last_ping) >= ping_interval {
        sess.send_to_all(&[S2C_PING]);
        sess.last_ping = now;
    }
    if !ping_interval.is_zero() {
        let next_ping = sess.last_ping + ping_interval;
        next_deadline = Some(next_deadline.map_or(next_ping, |d: Instant| d.min(next_ping)));
    }

    let max_fps = sess
        .clients
        .values()
        .map(browser_pacing_fps)
        .fold(1.0_f32, f32::max);
    let title_interval = Duration::from_secs_f64(1.0 / max_fps as f64);
    let ids: Vec<u16> = sess.ptys.keys().copied().collect();
    for &id in &ids {
        let Some(pty) = sess.ptys.get_mut(&id) else {
            continue;
        };
        if pty.driver.take_title_dirty() {
            pty.mark_dirty();
            pty.title_pending = true;
        }
        if pty.title_pending && now.duration_since(pty.last_title_send) >= title_interval {
            let msg = {
                let title_bytes = pty.driver.title().as_bytes();
                let mut msg = Vec::with_capacity(3 + title_bytes.len());
                msg.push(S2C_TITLE);
                msg.extend_from_slice(&id.to_le_bytes());
                msg.extend_from_slice(title_bytes);
                msg
            };
            pty.last_title_send = now;
            pty.title_pending = false;
            sess.send_to_all(&msg);
        }
    }

    // Drain bytes from PTY reader channels. This is the only place
    // process() is called, so there is no contention with the readers.
    let mut eof_ptys: Vec<u16> = Vec::with_capacity(ids.len());
    for &id in &ids {
        let Some(pty) = sess.ptys.get_mut(&id) else {
            continue;
        };
        while let Ok(input) = pty.byte_rx.try_recv() {
            match input {
                PtyInput::Data(data) => {
                    pty::respond_to_queries(
                        &pty.handle,
                        &data,
                        pty.driver.size(),
                        pty.driver.cursor_position(),
                    );
                    pty.driver.process(&data);
                    pty.mark_dirty();
                }
                PtyInput::SyncBoundary { before, after } => {
                    if !before.is_empty() {
                        pty::respond_to_queries(
                            &pty.handle,
                            &before,
                            pty.driver.size(),
                            pty.driver.cursor_position(),
                        );
                        pty.driver.process(&before);
                        pty.mark_dirty();
                    }
                    if !pty.driver.synced_output() {
                        let frame = take_snapshot(pty);
                        enqueue_ready_frame(&mut pty.ready_frames, frame);
                        pty.clear_dirty();
                    }
                    if !after.is_empty() {
                        pty::respond_to_queries(
                            &pty.handle,
                            &after,
                            pty.driver.size(),
                            pty.driver.cursor_position(),
                        );
                        pty.driver.process(&after);
                        pty.mark_dirty();
                    }
                }
                PtyInput::Eof => {
                    eof_ptys.push(id);
                }
            }
        }
    }
    // Handle EOF outside the borrow loop.
    drop(sess);
    for id in eof_ptys {
        tokio::time::sleep(Duration::from_millis(50)).await;
        cleanup_pty_internal(id, state).await;
    }
    let mut sess = state.session.lock().await;

    // Only snapshot PTYs that have at least one client ready to consume a fresh
    // frame right now. This avoids burning CPU on snapshot+diff+compress work
    // while the lead is merely waiting for its next pacing deadline.
    let needful_ptys: HashSet<u16> = sess
        .clients
        .values()
        .flat_map(|c| {
            let reserve_preview_slot = client_has_due_preview(&sess, c, now);
            c.subscriptions.iter().copied().filter(move |pid| {
                let scrolled = c.scroll_offsets.get(pid).copied().unwrap_or(0) > 0;
                if Some(*pid) == c.lead {
                    !scrolled && can_send_frame(c, now, reserve_preview_slot)
                } else {
                    !scrolled && can_send_preview(c, *pid, now)
                }
            })
        })
        .collect();

    let mut snapshots: HashMap<u16, FrameState> = HashMap::new();
    for &id in &ids {
        let Some(pty) = sess.ptys.get_mut(&id) else {
            continue;
        };
        if needful_ptys.contains(&id)
            && let Some(frame) = pty.ready_frames.pop_front()
        {
            snapshots.insert(id, frame);
            sess.tick_snaps += 1;
            continue;
        }
        if !should_snapshot_pty(
            pty.dirty,
            needful_ptys.contains(&id),
            pty.driver.synced_output(),
        ) {
            continue;
        }
        // Applications that care about complete-frame boundaries should
        // use DEC synchronized output (?2026). Outside that bracket we
        // snapshot immediately instead of heuristically coalescing reads.
        snapshots.insert(id, take_snapshot(pty));
        pty.clear_dirty();
        sess.tick_snaps += 1;
    }

    let client_ids: Vec<u64> = sess.clients.keys().copied().collect();
    for cid in client_ids {
        // When the pipe is idle (nothing in flight), RTT cannot be measured
        // and the last observed value stales.  Decay it toward min_rtt so
        // a stale congested RTT doesn't permanently suppress the send window
        // after congestion clears or traffic patterns change (e.g. switching
        // from a large-frame burst to idle small-frame updates).
        if let Some(c) = sess.clients.get_mut(&cid) {
            if c.inflight_bytes == 0 && c.min_rtt_ms > 0.0 && c.rtt_ms > c.min_rtt_ms {
                c.rtt_ms = (c.rtt_ms * 0.99 + c.min_rtt_ms * 0.01).max(c.min_rtt_ms);
            }
            // Decay stale browser metrics so a missed/delayed metrics update
            // can't permanently block the delivery loop.
            if c.last_metrics_update.elapsed() > Duration::from_secs(1) {
                c.browser_backlog_frames = 0;
                c.browser_ack_ahead_frames = 0;
            }
        }
        let (
            lead,
            subscriptions,
            scrolled_ptys,
            can_send_lead,
            lead_has_window,
            any_send_window,
            lead_deadline,
        ) = {
            let Some(c) = sess.clients.get(&cid) else {
                continue;
            };
            let reserve_preview_slot = client_has_due_preview(&sess, c, now);
            (
                c.lead,
                c.subscriptions.iter().copied().collect::<Vec<_>>(),
                c.scroll_offsets
                    .iter()
                    .map(|(&k, &v)| (k, v))
                    .collect::<Vec<_>>(),
                can_send_frame(c, now, reserve_preview_slot),
                lead_window_open(c, reserve_preview_slot),
                lead_window_open(c, reserve_preview_slot) || window_open(c),
                c.next_send_at,
            )
        };

        if subscriptions.is_empty() {
            continue;
        }

        // Send scrollback frames for any scrolled PTY.
        for &(scroll_pid, scroll_offset) in &scrolled_ptys {
            if scroll_offset == 0 {
                continue;
            }
            let is_lead = lead == Some(scroll_pid);
            let can_send = if is_lead { can_send_lead } else { true };
            if can_send {
                let prev_frame = {
                    let Some(c) = sess.clients.get(&cid) else {
                        continue;
                    };
                    c.scroll_caches
                        .get(&scroll_pid)
                        .cloned()
                        .unwrap_or_default()
                };
                let outcome = if let Some(pty) = sess.ptys.get_mut(&scroll_pid) {
                    if let Some((msg, new_frame)) =
                        build_scrollback_update(pty, scroll_pid, scroll_offset, &prev_frame)
                    {
                        let Some(c) = sess.clients.get_mut(&cid) else {
                            break;
                        };
                        let bytes = msg.len();
                        if try_send_outbox(c, msg).is_ok() {
                            c.scroll_caches.insert(scroll_pid, new_frame);
                            record_send(c, bytes, now, is_lead);
                            c.frames_sent += 1;
                            SendOutcome::Sent
                        } else {
                            SendOutcome::Backpressured
                        }
                    } else {
                        SendOutcome::NoChange
                    }
                } else {
                    SendOutcome::NoChange
                };
                match outcome {
                    SendOutcome::Sent => {}
                    SendOutcome::Backpressured => {
                        if let Some(pty) = sess.ptys.get_mut(&scroll_pid) {
                            pty.mark_dirty();
                        }
                    }
                    SendOutcome::NoChange => {}
                }
            } else if is_lead && lead_has_window {
                next_deadline = Some(match next_deadline {
                    Some(existing) => existing.min(lead_deadline),
                    None => lead_deadline,
                });
            }
        }

        let lead_scroll_offset = lead
            .and_then(|pid| {
                scrolled_ptys
                    .iter()
                    .find(|&&(k, _)| k == pid)
                    .map(|&(_, v)| v)
            })
            .unwrap_or(0);

        if let Some(pid) = lead {
            if lead_scroll_offset == 0 && can_send_lead {
                if let Some(cur) = snapshots.get(&pid).cloned() {
                    let previous = sess
                        .clients
                        .get(&cid)
                        .and_then(|c| c.last_sent.get(&pid).cloned())
                        .unwrap_or_default();
                    drop(sess);
                    let msg = build_update_msg(pid, &cur, &previous);
                    sess = state.session.lock().await;
                    let Some(c) = sess.clients.get_mut(&cid) else {
                        continue;
                    };
                    match try_send_update(c, pid, cur, msg, now, true) {
                        SendOutcome::Sent => {}
                        SendOutcome::Backpressured => {
                            if let Some(pty) = sess.ptys.get_mut(&pid) {
                                pty.mark_dirty();
                            }
                        }
                        SendOutcome::NoChange => {}
                    }
                } else {
                    let has_pending = sess
                        .ptys
                        .get(&pid)
                        .map(pty_has_visual_update)
                        .unwrap_or(false);
                    let _ = has_pending;
                }
            } else {
                let has_pending = sess
                    .ptys
                    .get(&pid)
                    .map(pty_has_visual_update)
                    .unwrap_or(false);
                if has_pending && lead_has_window {
                    next_deadline = Some(match next_deadline {
                        Some(existing) => existing.min(lead_deadline),
                        None => lead_deadline,
                    });
                }
            }
        }

        if !any_send_window {
            continue;
        }

        let mut preview_ids = subscriptions;
        preview_ids.retain(|pid| Some(*pid) != lead);
        preview_ids.sort_unstable();

        for pid in preview_ids {
            let (preview_can_send, preview_due_at, preview_has_window) =
                match sess.clients.get(&cid) {
                    Some(c) => (
                        can_send_preview(c, pid, now),
                        preview_deadline(c, pid, now),
                        window_open(c),
                    ),
                    None => (false, now, false),
                };
            if !preview_has_window {
                break;
            }
            if !preview_can_send {
                let has_pending = sess
                    .ptys
                    .get(&pid)
                    .map(pty_has_visual_update)
                    .unwrap_or(false);
                // Only set a deadline when the reason is *timing* (deadline
                // in the future), not capacity (preview window closed).
                // A past deadline here spins the delivery loop because
                // sleep_until(past) returns immediately.
                if has_pending && preview_due_at > now {
                    next_deadline = Some(match next_deadline {
                        Some(existing) => existing.min(preview_due_at),
                        None => preview_due_at,
                    });
                }
                continue;
            }
            let Some(cur) = snapshots.get(&pid) else {
                let has_pending = sess
                    .ptys
                    .get(&pid)
                    .map(pty_has_visual_update)
                    .unwrap_or(false);
                let _ = has_pending;
                continue;
            };
            let cur = cur.clone();
            let previous = sess
                .clients
                .get(&cid)
                .and_then(|c| c.last_sent.get(&pid).cloned())
                .unwrap_or_default();
            drop(sess);
            let msg = build_update_msg(pid, &cur, &previous);
            sess = state.session.lock().await;
            let Some(c) = sess.clients.get_mut(&cid) else {
                break;
            };
            match try_send_update(c, pid, cur, msg, now, false) {
                SendOutcome::Sent => {
                    record_preview_send(c, pid, now);
                }
                SendOutcome::Backpressured => {
                    if let Some(pty) = sess.ptys.get_mut(&pid) {
                        pty.mark_dirty();
                    }
                    break;
                }
                SendOutcome::NoChange => {}
            }
        }
    }

    // Surface IDs whose per-client encoders need to be invalidated.
    let mut invalidate_client_encoders: Vec<u16> = Vec::new();

    let mut surface_commit_count = 0u32;
    if let Some(cs) = sess.compositor.as_mut() {
        let mut events = Vec::new();
        while let Ok(event) = cs.handle.event_rx.try_recv() {
            events.push(event);
        }
        let mut broadcast: Vec<Vec<u8>> = Vec::new();
        for event in events {
            match event {
                CompositorEvent::SurfaceCreated {
                    surface_id,
                    title,
                    app_id,
                    parent_id,
                    width,
                    height,
                } => {
                    broadcast.push(msg_surface_created(
                        surface_id, parent_id, width, height, &title, &app_id,
                    ));
                    cs.surfaces.insert(
                        surface_id,
                        CachedSurfaceInfo {
                            surface_id,
                            parent_id,
                            width,
                            height,
                            title,
                            app_id,
                        },
                    );
                    cs.last_pixels.remove(&surface_id);
                    invalidate_client_encoders.push(surface_id);
                }
                CompositorEvent::SurfaceDestroyed { surface_id } => {
                    cs.surfaces.remove(&surface_id);
                    cs.last_pixels.remove(&surface_id);
                    cs.last_configured_size.remove(&surface_id);
                    invalidate_client_encoders.push(surface_id);
                    broadcast.push(msg_surface_destroyed(surface_id));
                }
                CompositorEvent::SurfaceCommit {
                    surface_id,
                    width,
                    height,
                    pixels,
                } => {
                    surface_commit_count += 1;
                    if let Some(info) = cs.surfaces.get_mut(&surface_id) {
                        info.width = width as u16;
                        info.height = height as u16;
                    }
                    cs.pixel_generation += 1;
                    cs.last_pixels.insert(
                        surface_id,
                        LastPixels {
                            width,
                            height,
                            pixels,
                            generation: cs.pixel_generation,
                        },
                    );
                }
                CompositorEvent::SurfaceTitle { surface_id, title } => {
                    if let Some(info) = cs.surfaces.get_mut(&surface_id) {
                        info.title = title.clone();
                    }
                    broadcast.push(msg_surface_title(surface_id, &title));
                }
                CompositorEvent::SurfaceAppId { surface_id, app_id } => {
                    if let Some(info) = cs.surfaces.get_mut(&surface_id) {
                        info.app_id = app_id.clone();
                    }
                    broadcast.push(msg_surface_app_id(surface_id, &app_id));
                }
                CompositorEvent::SurfaceResized {
                    surface_id,
                    width,
                    height,
                } => {
                    if let Some(info) = cs.surfaces.get_mut(&surface_id) {
                        info.width = width;
                        info.height = height;
                    }
                    cs.last_pixels.remove(&surface_id);
                    invalidate_client_encoders.push(surface_id);
                    broadcast.push(msg_surface_resized(surface_id, width, height));
                }
                CompositorEvent::ClipboardContent {
                    mime_type, data, ..
                } => {
                    broadcast.push(msg_s2c_clipboard_content(&mime_type, &data));
                }
                CompositorEvent::SurfaceCursor { surface_id, cursor } => {
                    // Format: [0x29][surface_id:2][type:1][payload...]
                    // type 0 = named: [name_len:1][name:N]
                    // type 1 = hidden (no payload)
                    // type 2 = custom: [hotx:2][hoty:2][w:2][h:2][png:N]
                    let mut msg = Vec::new();
                    msg.push(blit_remote::S2C_SURFACE_CURSOR);
                    msg.extend_from_slice(&surface_id.to_le_bytes());
                    match &cursor {
                        blit_compositor::CursorImage::Named(name) => {
                            msg.push(0); // type = named
                            msg.push(name.len() as u8);
                            msg.extend_from_slice(name.as_bytes());
                        }
                        blit_compositor::CursorImage::Hidden => {
                            msg.push(1); // type = hidden
                        }
                        blit_compositor::CursorImage::Custom {
                            hotspot_x,
                            hotspot_y,
                            width,
                            height,
                            rgba,
                        } => {
                            // Encode as PNG to keep message small.
                            let mut png_buf = Vec::new();
                            {
                                let mut encoder =
                                    png::Encoder::new(&mut png_buf, *width as u32, *height as u32);
                                encoder.set_color(png::ColorType::Rgba);
                                encoder.set_depth(png::BitDepth::Eight);
                                if let Ok(mut writer) = encoder.write_header() {
                                    let _ = writer.write_image_data(rgba);
                                }
                            }
                            msg.push(2); // type = custom
                            msg.extend_from_slice(&hotspot_x.to_le_bytes());
                            msg.extend_from_slice(&hotspot_y.to_le_bytes());
                            msg.extend_from_slice(&width.to_le_bytes());
                            msg.extend_from_slice(&height.to_le_bytes());
                            msg.extend_from_slice(&png_buf);
                        }
                    }
                    broadcast.push(msg);
                }
            }
        }
        for msg in &broadcast {
            sess.send_to_all(msg);
        }
    }
    sess.surface_commits += surface_commit_count;

    // Apply deferred per-client encoder invalidation (couldn't mutate
    // sess.clients while sess.compositor was borrowed above).
    for sid in invalidate_client_encoders {
        for c in sess.clients.values_mut() {
            c.surface_encoders.remove(&sid);
            c.surface_last_encoded_gen.remove(&sid);
        }
    }

    // Per-client surface encode + deliver.
    // Each client has its own encoder per surface.  We encode from
    // shared last_pixels into each client's encoder and deliver.
    //
    // Snapshot pixel metadata from the compositor first to avoid
    // holding an immutable borrow on sess.compositor while mutating
    // sess.clients.
    let pixel_snapshot: Vec<(u16, u32, u32, u64)> = sess
        .compositor
        .as_ref()
        .map(|cs| {
            cs.last_pixels
                .iter()
                .map(|(&sid, lp)| (sid, lp.width, lp.height, lp.generation))
                .collect()
        })
        .unwrap_or_default();

    // ---- Surface encode (off main thread) + deliver ----
    //
    // Collect encode jobs, drop the session lock, run encodes in
    // spawn_blocking, re-acquire the lock, and deliver.

    struct EncodeJob {
        cid: u64,
        sid: u16,
        px_w: u32,
        px_h: u32,
        pixels: blit_compositor::PixelData,
        needs_keyframe: bool,
        encoder: SurfaceEncoder,
        generation: u64,
    }
    struct EncodeResult {
        cid: u64,
        sid: u16,
        px_w: u32,
        px_h: u32,
        generation: u64,
        encoder: SurfaceEncoder,
        nal_data: Option<(Vec<u8>, bool)>, // (data, is_keyframe)
        codec_flag: u8,
    }

    let mut encode_jobs: Vec<EncodeJob> = Vec::new();

    // Collect (cid, subs, needs_kf) for clients that are due, then build
    // encode jobs in a second pass to avoid overlapping borrows.
    struct ClientWork {
        cid: u64,
        subs: HashSet<u16>,
        needs_keyframe: bool,
    }
    let mut client_work: Vec<ClientWork> = Vec::new();

    if !pixel_snapshot.is_empty() {
        for (&cid, client) in sess.clients.iter_mut() {
            if !surface_window_open(client) {
                continue;
            }
            // During burst-start (first few frames after subscribe/keyframe
            // request), skip the time-based pacing gate — let outbox
            // backpressure (checked above) be the only flow control.
            // This lets frames flow at wire speed, establishing bandwidth
            // estimates quickly on high-latency links.
            if client.surface_burst_remaining == 0 && client.surface_next_send_at > now {
                let deadline = client.surface_next_send_at;
                next_deadline = Some(match next_deadline {
                    Some(existing) => existing.min(deadline),
                    None => deadline,
                });
                continue;
            }
            if client.surface_subscriptions.is_empty() {
                continue;
            }
            client_work.push(ClientWork {
                cid,
                subs: client.surface_subscriptions.clone(),
                needs_keyframe: client.surface_needs_keyframe,
            });
            // Don't advance the deadline here — wait until we know an
            // encode job was actually collected (see below).  Advancing
            // eagerly wastes time slots when the encode is skipped due
            // to in-flight limits or unchanged pixel data.
        }

        // Track which clients actually had encode jobs collected so we
        // can advance their deadlines after the job-collection pass.
        let mut clients_with_encodes: HashSet<u64> = HashSet::new();
        #[cfg(target_os = "linux")]
        let mut pending_external_bufs: Option<Vec<blit_compositor::ExternalOutputBuffer>> = None;

        for work in &client_work {
            for &(sid, px_w, px_h, px_gen) in &pixel_snapshot {
                if !work.subs.contains(&sid) {
                    continue;
                }
                let client = sess.clients.get_mut(&work.cid).unwrap();

                // Skip encoding if the pixel data hasn't changed since the
                // last encode for this client, unless a keyframe is needed
                // (e.g. late-joining client).
                if !work.needs_keyframe
                    && let Some(&last_gen) = client.surface_last_encoded_gen.get(&sid)
                    && last_gen == px_gen
                {
                    continue;
                }

                let pixels = {
                    let cs = sess.compositor.as_ref().unwrap();
                    match cs.last_pixels.get(&sid) {
                        Some(lp) if lp.width == px_w && lp.height == px_h => lp.pixels.clone(),
                        _ => continue,
                    }
                };
                let client = sess.clients.get_mut(&work.cid).unwrap();

                // Skip if an encode job is already in flight for this
                // (client, surface) pair.  Spawning a second job would
                // create a throwaway encoder whose output races with the
                // first, and concurrent C-library encode instances for the
                // same client can corrupt memory.
                if client.surface_encodes_in_flight.contains(&sid) {
                    continue;
                }

                let needs_new_encoder = client
                    .surface_encoders
                    .get(&sid)
                    .is_none_or(|e| e.source_dimensions() != (px_w, px_h));
                if needs_new_encoder {
                    client.surface_encoders.remove(&sid);
                    let codec_support = client
                        .surface_codec_overrides
                        .get(&sid)
                        .copied()
                        .unwrap_or(client.surface_codec_support);
                    let quality = client
                        .surface_quality_overrides
                        .get(&sid)
                        .copied()
                        .unwrap_or(state.config.surface_quality);
                    match SurfaceEncoder::new(
                        &state.config.surface_encoders,
                        px_w,
                        px_h,
                        &state.config.vaapi_device,
                        quality,
                        state.config.verbose,
                        codec_support,
                    ) {
                        Ok(encoder) => {
                            #[cfg(target_os = "linux")]
                            {
                                let exported = encoder.export_vpp_surfaces();
                                if !exported.is_empty() {
                                    let va_display = encoder.va_display_usize();
                                    let bufs = exported
                                        .into_iter()
                                        .map(|e| blit_compositor::ExternalOutputBuffer {
                                            fd: std::sync::Arc::new(e.fd),
                                            fourcc: e.fourcc,
                                            modifier: e.modifier,
                                            stride: e.stride,
                                            offset: e.offset,
                                            width: e.width,
                                            height: e.height,
                                            va_surface_id: e.surface_id,
                                            va_display,
                                        })
                                        .collect();
                                    pending_external_bufs = Some(bufs);
                                }
                            }
                            let enc_msg = msg_surface_encoder(sid, encoder.encoder_name());
                            client.surface_encoders.insert(sid, encoder);
                            let _ = try_send_outbox(client, enc_msg);
                        }
                        Err(err) => {
                            if state.config.verbose {
                                eprintln!(
                                    "[surface-encoder] cid={} sid={sid} {px_w}x{px_h}: {err}",
                                    work.cid
                                );
                            }
                            continue;
                        }
                    }
                }

                let encoder = client.surface_encoders.remove(&sid).unwrap();
                client.surface_encodes_in_flight.insert(sid);
                // A fresh encoder always needs a keyframe, regardless of
                // the client's global flag.
                let needs_kf = work.needs_keyframe || needs_new_encoder;
                clients_with_encodes.insert(work.cid);
                encode_jobs.push(EncodeJob {
                    cid: work.cid,
                    sid,
                    px_w,
                    px_h,
                    pixels,
                    needs_keyframe: needs_kf,
                    encoder,
                    generation: px_gen,
                });
            }
        }

        // Send VA-API exported surfaces to compositor if a new encoder was created.
        #[cfg(target_os = "linux")]
        if let Some(bufs) = pending_external_bufs
            && let Some(cs) = sess.compositor.as_ref()
        {
            let _ = cs.handle.command_tx.send(
                blit_compositor::CompositorCommand::SetExternalOutputBuffers { buffers: bufs },
            );
            cs.handle.wake();
        }

        // Advance the pacing deadline only for clients that actually had
        // at least one encode job collected.  Clients skipped due to
        // in-flight limits or unchanged pixels keep their current
        // deadline so the next tick retries without burning a time slot.
        for work in &client_work {
            if let Some(client) = sess.clients.get_mut(&work.cid)
                && clients_with_encodes.contains(&work.cid)
            {
                let interval = surface_send_interval(client);
                advance_deadline(&mut client.surface_next_send_at, now, interval);
            }
        }
    }

    if !encode_jobs.is_empty() {
        // Fire-and-forget: spawn the encode and deliver asynchronously
        // so the tick loop is never blocked by slow encoders.
        let state2 = state.clone();
        tokio::spawn(async move {
            // Track (cid, sid) for each job so we can clean up
            // surface_encodes_in_flight if a task panics or times out.
            let job_ids: Vec<(u64, u16)> = encode_jobs.iter().map(|j| (j.cid, j.sid)).collect();

            let handles: Vec<_> = encode_jobs
                .into_iter()
                .map(|mut job| {
                    tokio::task::spawn_blocking(move || {
                        if job.needs_keyframe {
                            job.encoder.request_keyframe();
                        }
                        let nal_data = job.encoder.encode_pixels(&job.pixels);
                        let codec_flag = job.encoder.codec_flag();
                        EncodeResult {
                            cid: job.cid,
                            sid: job.sid,
                            px_w: job.px_w,
                            px_h: job.px_h,
                            generation: job.generation,
                            encoder: job.encoder,
                            nal_data,
                            codec_flag,
                        }
                    })
                })
                .collect();

            // Timeout: if a hardware encoder hangs (e.g. vaSyncSurface on
            // AMD), don't block delivery of other surfaces' results forever.
            const ENCODE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

            let mut results = Vec::with_capacity(handles.len());
            let mut failed: Vec<(u64, u16)> = Vec::new();
            for (i, h) in handles.into_iter().enumerate() {
                match tokio::time::timeout(ENCODE_TIMEOUT, h).await {
                    Ok(Ok(r)) => results.push(r),
                    Ok(Err(_join_err)) => {
                        // spawn_blocking panicked — encoder is lost.
                        eprintln!(
                            "[surface-encoder] encode task panicked: cid={} sid={}",
                            job_ids[i].0, job_ids[i].1
                        );
                        failed.push(job_ids[i]);
                    }
                    Err(_timeout) => {
                        // Encoder hung (e.g. GPU hang in vaSyncSurface).
                        // The blocking thread is leaked but we must not
                        // let it stall all other surfaces forever.
                        eprintln!(
                            "[surface-encoder] encode timed out ({}s): cid={} sid={}",
                            ENCODE_TIMEOUT.as_secs(),
                            job_ids[i].0,
                            job_ids[i].1
                        );
                        failed.push(job_ids[i]);
                    }
                }
            }

            // Deliver encoded frames.
            let mut sess = state2.session.lock().await;
            let now = Instant::now();
            let mut local_encodes = 0u32;
            let mut local_encode_bytes = 0u64;
            let mut local_frames_sent = 0u32;

            // Clean up in-flight tracking for panicked/timed-out encodes.
            // Without this, the surface is permanently blocked from future
            // encode jobs and frame delivery stops for that surface.
            for (cid, sid) in failed {
                if let Some(client) = sess.clients.get_mut(&cid) {
                    client.surface_encodes_in_flight.remove(&sid);
                    // The encoder was moved into the spawn_blocking closure
                    // and is now lost.  A fresh encoder will be created on
                    // the next tick when surface_encoders doesn't contain
                    // this sid.  Force a keyframe so the new encoder starts
                    // with a clean reference chain.
                    client.surface_needs_keyframe = true;
                }
            }

            for result in results {
                // Return the encoder to the client, but only if its
                // dimensions still match the current surface.  A resize
                // that arrived while the encode was in flight will have
                // invalidated the old encoder; reinserting the stale one
                // would force the next tick to discard and recreate it,
                // wasting work and risking feeding a C encoder (openh264)
                // frames at the wrong resolution.
                let dims_match = sess
                    .compositor
                    .as_ref()
                    .and_then(|cs| cs.last_pixels.get(&result.sid))
                    .is_some_and(|lp| result.encoder.source_dimensions() == (lp.width, lp.height));
                if let Some(client) = sess.clients.get_mut(&result.cid) {
                    client.surface_encodes_in_flight.remove(&result.sid);
                    if dims_match {
                        client.surface_encoders.insert(result.sid, result.encoder);
                    }
                    // Record the generation we just encoded so we don't
                    // re-encode identical pixel data on subsequent ticks.
                    client
                        .surface_last_encoded_gen
                        .insert(result.sid, result.generation);
                }

                let Some((nal_data, is_keyframe)) = result.nal_data else {
                    continue;
                };

                local_encodes += 1;
                local_encode_bytes += nal_data.len() as u64;

                let created_at = sess
                    .compositor
                    .as_ref()
                    .map(|cs| cs.created_at)
                    .unwrap_or(now);

                let flags = result.codec_flag
                    | if is_keyframe {
                        SURFACE_FRAME_FLAG_KEYFRAME
                    } else {
                        0
                    };
                let timestamp = created_at.elapsed().as_millis() as u32;
                let msg = msg_surface_frame(
                    result.sid,
                    timestamp,
                    flags,
                    result.px_w as u16,
                    result.px_h as u16,
                    &nal_data,
                );
                let bytes = msg.len();

                let Some(client) = sess.clients.get_mut(&result.cid) else {
                    continue;
                };

                // Don't check window_open here — we already checked before
                // starting the encode job.  Dropping an encoded P-frame
                // breaks the decoder's reference chain and causes glitches.
                // With surface_encodes_in_flight limiting to 1 concurrent
                // encode per surface, at most 1 frame arrives after the
                // window closes, which is acceptable.
                match try_send_outbox(client, msg) {
                    Err(_e) => {
                        // Outbox full (rare — window_open checked at encode
                        // time but outbox filled during async encode).
                        // Request keyframe to avoid broken reference chain.
                        client.surface_needs_keyframe = true;
                    }
                    Ok(()) => {
                        // Track surface frames in their own inflight queue
                        // so surface ACKs feed shared goodput / RTT without
                        // polluting terminal frame-size averages or probing.
                        client.surface_inflight_frames.push_back(InFlightFrame {
                            sent_at: now,
                            bytes,
                            paced: true,
                        });
                        // Prefer updating avg_surface_frame_bytes from delta
                        // (non-keyframe) frames — keyframes are 5-10× larger
                        // than P-frames and would inflate the average, dragging
                        // surface_pacing_fps below the sustainable rate.
                        //
                        // However, we must still update from keyframes with a
                        // very slow alpha: all-intra encoders (e.g. AV1 VAAPI
                        // before P-frame support) only produce keyframes, so
                        // skipping them entirely leaves the average stuck at
                        // the 8 KB initial value, causing the pacer to wildly
                        // overshoot the send rate and saturate the transport.
                        if !is_keyframe {
                            client.avg_surface_frame_bytes = ewma_with_direction(
                                client.avg_surface_frame_bytes,
                                bytes as f32,
                                0.5,
                                0.125,
                            );
                        } else if client.avg_surface_frame_bytes <= 16_384.0 {
                            // First keyframe while the estimate is still at or
                            // near the initial 8 KB seed.  No P-frame data has
                            // been seen yet, so the seed is pure fiction.  Use a
                            // realistic P-frame estimate: keyframes are typically
                            // 3-8× larger than P-frames, so divide by 4.  This
                            // prevents surface_pacing_fps from being wildly
                            // optimistic (8 KB → 32 fps at 256 KB/s) when the
                            // actual frames are 50-200 KB keyframes.
                            client.avg_surface_frame_bytes = (bytes as f32 / 4.0).max(4_096.0);
                        } else {
                            // Slow convergence so one keyframe doesn't wreck
                            // the estimate for dozens of subsequent P-frames.
                            client.avg_surface_frame_bytes = ewma_with_direction(
                                client.avg_surface_frame_bytes,
                                bytes as f32,
                                0.05,
                                0.05,
                            );
                        }
                        client.frames_sent = client.frames_sent.wrapping_add(1);
                        local_frames_sent += 1;
                        if client.surface_needs_keyframe && is_keyframe {
                            client.surface_needs_keyframe = false;
                        }
                        client.surface_burst_remaining =
                            client.surface_burst_remaining.saturating_sub(1);
                    }
                }
            }
            sess.surface_encodes += local_encodes;
            sess.surface_encode_bytes += local_encode_bytes;
            sess.surface_frames_sent += local_frames_sent;
            drop(sess);
            // Wake the tick loop so it can request the next frame.
            state2.delivery_notify.notify_one();
        });
    }

    // Request frames from the compositor for surfaces that have at least
    // one subscriber whose pacing says it can accept a new frame.  This
    // fires the surface's pending wl_surface.frame callback so the
    // Wayland client will paint and commit its next frame.
    //
    // No timers — the pipeline is fully demand-driven:
    //   server sends RequestFrame + wake() → compositor fires callback →
    //   app paints & commits → compositor sends SurfaceCommit +
    //   event_notify → delivery_notify wakes tick loop → encode & deliver.
    // See ARCHITECTURE.md § "Compositor sessions".
    {
        // Only request frames for surfaces where at least one client is
        // ready to consume the result.  Without this check, apps that are
        // always ready to paint (video players like mpv) cause a hot loop:
        // RequestFrame → commit → SurfaceCommit wakes tick → no client
        // ready → RequestFrame again → 100% CPU.
        let mut wanted: HashSet<u16> = HashSet::new();
        let mut blanket_requested = false;
        // Request frames for all known surfaces so Wayland apps can make
        // rendering progress.  Video players (mpv) need frequent callbacks
        // to advance their presentation clock; browsers need them for
        // page loads and animations.
        if let Some(cs) = sess.compositor.as_ref()
            && now.duration_since(cs.last_blanket_frame_request) >= BLANKET_FRAME_INTERVAL
        {
            for &sid in cs.surfaces.keys() {
                wanted.insert(sid);
            }
            blanket_requested = true;
        }
        for client in sess.clients.values() {
            if !surface_window_open(client) {
                continue;
            }
            let surface_ready = client.surface_next_send_at <= now;
            if !surface_ready {
                // Record the client's surface deadline so the outer loop
                // sleeps until this client is ready.
                let deadline = client.surface_next_send_at;
                next_deadline = Some(match next_deadline {
                    Some(existing) => existing.min(deadline),
                    None => deadline,
                });
                continue;
            }
            for &sid in &client.surface_subscriptions {
                wanted.insert(sid);
            }
        }

        if let Some(cs) = sess.compositor.as_mut() {
            if blanket_requested {
                cs.last_blanket_frame_request = now;
            }

            // Gate: at most one RequestFrame per surface per millisecond.
            // This ensures each wl_callback.done carries a distinct
            // elapsed_ms timestamp (video players like mpv use these to
            // pace their presentation clock).  Supports up to 1 kHz.
            // The gate auto-expires: if the app doesn't commit, the next
            // tick ≥1 ms later will send a fresh request.
            const MIN_REQUEST_INTERVAL: Duration = Duration::from_millis(1);
            let mut sent_any = false;
            for sid in &wanted {
                let dominated = cs
                    .last_frame_request
                    .get(sid)
                    .is_some_and(|&t| now.duration_since(t) < MIN_REQUEST_INTERVAL);
                if !dominated {
                    cs.last_frame_request.insert(*sid, now);
                    let _ = cs
                        .handle
                        .command_tx
                        .send(CompositorCommand::RequestFrame { surface_id: *sid });
                    sent_any = true;
                }
            }
            if sent_any {
                cs.handle.wake();
            }
        }
    }

    // -- Audio frame delivery -----------------------------------------------
    #[cfg(unix)]
    if let Some(ref mut cs) = sess.compositor
        && let Some(ref mut ap) = cs.audio_pipeline
        && ap.is_alive()
    {
        let new_frames = ap.poll_frames();
        if !new_frames.is_empty() {
            for client in sess.clients.values_mut() {
                if client.audio_subscribed {
                    for frame in &new_frames {
                        let msg = audio::msg_audio_frame(frame);
                        let bytes = msg.len();
                        // Send via the dedicated audio channel so audio is
                        // never blocked behind large video/terminal messages
                        // in the shared outbox.  The writer task interleaves
                        // audio between write syscalls of bulk messages.
                        if client.audio_tx.try_send(msg).is_ok() {
                            // Count audio bytes into the goodput window so
                            // the shared bandwidth estimate reflects total
                            // pipe utilisation (terminals + video + audio).
                            client.goodput_window_bytes += bytes;
                        }
                    }
                }
            }
        }
        // Schedule the next tick soon so audio frames don't accumulate.
        // 20 ms matches the Opus frame interval.
        let next_audio = now + Duration::from_millis(20);
        next_deadline = Some(next_deadline.map_or(next_audio, |d: Instant| d.min(next_audio)));
    }

    // -- Audio pipeline auto-restart ----------------------------------------
    // If the pipeline died (pw-cat exited, encoder crashed, PipeWire gone),
    // drop it, wait for a cooldown, and respawn.  This avoids permanent
    // audio loss that previously required a full client reconnect.
    //
    // Bitrate is pre-computed here to avoid borrowing sess.clients inside
    // the sess.compositor mutable borrow (they're the same MutexGuard).
    #[cfg(unix)]
    let audio_restart_bitrate: i32 = sess
        .clients
        .values()
        .filter(|c| c.audio_subscribed)
        .map(|c| c.audio_bitrate_kbps)
        .max()
        .map(|kbps| kbps as i32 * 1000)
        .unwrap_or(0);
    #[cfg(unix)]
    if let Some(ref mut cs) = sess.compositor {
        let pipeline_dead = cs.audio_pipeline.as_mut().is_some_and(|ap| !ap.is_alive());
        if pipeline_dead {
            const RESTART_COOLDOWN: Duration = Duration::from_secs(5);
            let can_restart = cs
                .last_audio_restart
                .is_none_or(|t| now.duration_since(t) >= RESTART_COOLDOWN);
            if can_restart {
                cs.last_audio_restart = Some(now);
                // Drop the dead pipeline — triggers shutdown() which kills
                // orphaned child processes and cleans up the runtime dir.
                cs.audio_pipeline = None;
                let runtime_dir = std::path::Path::new(&cs.handle.socket_name)
                    .parent()
                    .unwrap_or(std::path::Path::new("/tmp"));
                let session_id = cs.audio_session_id;
                let epoch = cs.created_at;
                let verbose = state.config.verbose;
                eprintln!("[audio] pipeline died, restarting...");
                let pipeline = tokio::task::block_in_place(|| {
                    audio::AudioPipeline::spawn(
                        runtime_dir,
                        session_id,
                        audio_restart_bitrate,
                        verbose,
                        epoch,
                    )
                });
                match pipeline {
                    Ok(p) => {
                        eprintln!(
                            "[audio] pipeline restarted, PULSE_SERVER={}",
                            p.pulse_server_path(),
                        );
                        cs.audio_pipeline = Some(p);
                    }
                    Err(e) => {
                        eprintln!("[audio] failed to restart pipeline: {e}");
                    }
                }
            }
        }
    }

    // Guarantee the tick loop wakes up at least every BLANKET_FRAME_INTERVAL
    // even when no other time-based work is pending.  Without this, the
    // loop blocks forever on delivery_notify when there are no events,
    // and the blanket RequestFrame (which keeps video players like mpv
    // ticking) never fires.
    {
        let blanket_deadline = now + BLANKET_FRAME_INTERVAL;
        next_deadline = Some(next_deadline.map_or(blanket_deadline, |d| d.min(blanket_deadline)));
    }

    TickOutcome { next_deadline }
}

async fn handle_client<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
    stream: S,
    state: AppState,
) {
    let config = &state.config;
    let notify_for_compositor = {
        let n = state.delivery_notify.clone();
        Arc::new(move || n.notify_one()) as Arc<dyn Fn() + Send + Sync>
    };
    let (mut reader, mut writer) = tokio::io::split(stream);

    let (out_tx, mut out_rx) = mpsc::channel::<Vec<u8>>(OUTBOX_CAPACITY);
    let (audio_tx, mut audio_rx) = mpsc::channel::<Vec<u8>>(AUDIO_OUTBOX_CAPACITY);
    let outbox_frame_counter = Arc::new(AtomicUsize::new(0));
    let outbox_byte_counter = Arc::new(AtomicUsize::new(0));
    let sender_outbox_queued_frames = outbox_frame_counter.clone();
    let sender_outbox_queued_bytes = outbox_byte_counter.clone();
    let sender = tokio::spawn(async move {
        loop {
            // Drain all pending audio before waiting for the next message.
            // Audio frames are tiny (~160 B) so this is near-instant.
            while let Ok(audio_msg) = audio_rx.try_recv() {
                if !write_frame(&mut writer, &audio_msg).await {
                    return;
                }
            }

            // Wait for the next message from either channel.  Prefer audio
            // so that audio frames queued while we were writing are sent
            // before the next bulk message.
            let msg = tokio::select! {
                biased;
                msg = audio_rx.recv() => {
                    // Pure audio message — write it directly (tiny).
                    match msg {
                        Some(m) => {
                            if !write_frame(&mut writer, &m).await {
                                break;
                            }
                            continue;
                        }
                        None => break,
                    }
                }
                msg = out_rx.recv() => msg,
            };

            // Non-audio message: may be large (video keyframe, terminal
            // snapshot).  Use interleaved write so audio frames that arrive
            // while the kernel TCP buffer drains are written between write
            // syscalls rather than piling up and being dropped.
            match msg {
                Some(m) => {
                    let bytes = m.len();
                    let wrote = write_frame_interleaved(&mut writer, &m, &mut audio_rx).await;
                    mark_outbox_drained(
                        &sender_outbox_queued_frames,
                        &sender_outbox_queued_bytes,
                        bytes,
                    );
                    if !wrote {
                        break;
                    }
                }
                None => break,
            }
        }
    });
    let client_id;

    {
        let mut sess = state.session.lock().await;
        client_id = sess.next_client_id;
        sess.next_client_id += 1;
        sess.clients.insert(
            client_id,
            ClientState {
                tx: out_tx,
                outbox_queued_frames: outbox_frame_counter,
                outbox_queued_bytes: outbox_byte_counter,
                audio_tx,
                lead: None,
                subscriptions: HashSet::new(),
                surface_subscriptions: HashSet::new(),
                audio_subscribed: false,
                #[cfg(unix)]
                audio_bitrate_kbps: 0,
                view_sizes: HashMap::new(),
                scroll_offsets: HashMap::new(),
                scroll_caches: HashMap::new(),
                last_sent: HashMap::new(),
                preview_next_send_at: HashMap::new(),
                rtt_ms: 50.0,
                min_rtt_ms: 0.0,
                display_fps: 60.0,
                // Conservative seed — the rise alpha (0.5) converges up to
                // multi-MB/s in a handful of samples on low-latency paths. Starting
                // high causes catastrophic bufferbloat on slow links because
                // target_byte_window scales with the goodput estimate.
                delivery_bps: 262_144.0,
                goodput_bps: 262_144.0,
                goodput_jitter_bps: 0.0,
                max_goodput_jitter_bps: 0.0,
                last_goodput_sample_bps: 0.0,
                avg_frame_bytes: 1_024.0,
                avg_paced_frame_bytes: 1_024.0,
                avg_preview_frame_bytes: 1_024.0,
                avg_surface_frame_bytes: 8_192.0,
                inflight_bytes: 0,
                inflight_frames: VecDeque::new(),
                next_send_at: Instant::now(),
                probe_frames: 0.0,
                frames_sent: 0,
                acks_recv: 0,
                acked_bytes_since_log: 0,
                browser_backlog_frames: 0,
                browser_ack_ahead_frames: 0,
                browser_apply_ms: 0.0,
                last_metrics_update: Instant::now(),
                last_log: Instant::now(),
                goodput_window_bytes: 0,
                goodput_window_start: Instant::now(),
                surface_next_send_at: Instant::now(),
                surface_needs_keyframe: true,
                surface_burst_remaining: SURFACE_BURST_FRAMES,
                surface_inflight_frames: VecDeque::new(),
                surface_encoders: HashMap::new(),
                surface_encodes_in_flight: HashSet::new(),
                surface_last_encoded_gen: HashMap::new(),
                surface_view_sizes: HashMap::new(),
                surface_codec_support: 0,
                surface_codec_overrides: HashMap::new(),
                surface_quality_overrides: HashMap::new(),
                pressed_surface_keys: HashSet::new(),
            },
        );
        // Wake the tick loop so the new client gets its first frame.
        state.delivery_notify.notify_one();
        if let Some(c) = sess.clients.get(&client_id) {
            let mut features = FEATURE_CREATE_NONCE
                | FEATURE_RESTART
                | FEATURE_RESIZE_BATCH
                | FEATURE_COPY_RANGE
                | FEATURE_COMPOSITOR;
            #[cfg(unix)]
            {
                let audio_disabled = std::env::var("BLIT_AUDIO")
                    .map(|v| v == "0")
                    .unwrap_or(false);
                if !audio_disabled && audio::pipewire_available() {
                    features |= FEATURE_AUDIO;
                }
            }
            let _ = try_send_outbox(c, msg_hello(1, features));
        }
        let mut initial_msgs = Vec::with_capacity(2 + sess.ptys.len() * 2);
        // Send surface-created messages BEFORE the PTY list so that
        // the client's surface store is populated before `ready` is
        // set — otherwise the BSP reconciliation runs with an empty
        // surface list and wipes restored surface assignments.
        if let Some(cs) = sess.compositor.as_ref() {
            for info in cs.surfaces.values() {
                // Use the latest known pixel dimensions if the stored
                // width/height is still 0 (surface created before first commit).
                let (w, h) = if info.width == 0 && info.height == 0 {
                    cs.last_pixels
                        .get(&info.surface_id)
                        .map(|lp| (lp.width as u16, lp.height as u16))
                        .unwrap_or((0, 0))
                } else {
                    (info.width, info.height)
                };
                initial_msgs.push(msg_surface_created(
                    info.surface_id,
                    info.parent_id,
                    w,
                    h,
                    &info.title,
                    &info.app_id,
                ));
                // Also send a resize message so the client gets the
                // correct dimensions even if surface_created carried 0x0.
                if w > 0 && h > 0 {
                    initial_msgs.push(msg_surface_resized(info.surface_id, w, h));
                }
            }
        }
        initial_msgs.push(sess.pty_list_msg());
        for (&id, pty) in &sess.ptys {
            let title = pty.driver.title();
            if !title.is_empty() {
                let title_bytes = title.as_bytes();
                let mut msg = Vec::with_capacity(3 + title_bytes.len());
                msg.push(S2C_TITLE);
                msg.extend_from_slice(&id.to_le_bytes());
                msg.extend_from_slice(title_bytes);
                initial_msgs.push(msg);
            }
            if pty.exited {
                initial_msgs.push(blit_remote::msg_exited(id, pty.exit_status));
            }
        }
        initial_msgs.push(vec![S2C_READY]);
        let tx = sess.clients.get(&client_id).map(|c| {
            (
                c.tx.clone(),
                c.outbox_queued_frames.clone(),
                c.outbox_queued_bytes.clone(),
            )
        });
        drop(sess);
        if let Some((tx, queued_frames, queued_bytes)) = tx {
            for msg in initial_msgs {
                if send_outbox_tracked(&tx, &queued_frames, &queued_bytes, msg)
                    .await
                    .is_err()
                {
                    break;
                }
            }
        }
    }

    if state.config.verbose {
        eprintln!("client connected");
    }

    while let Some(data) = read_frame(&mut reader).await {
        if data.is_empty() {
            continue;
        }

        if data[0] == C2S_ACK {
            let mut sess = state.session.lock().await;
            let (
                do_log,
                frames_sent,
                acks_recv,
                rtt_ms,
                min_rtt_ms,
                eff_rtt_ms,
                inflight_bytes,
                delivery_bps,
                goodput_ewma_bps,
                goodput_jitter_bps,
                max_goodput_jitter_bps,
                avg_frame_bytes,
                avg_paced_frame_bytes,
                avg_preview_frame_bytes,
                display_fps,
                paced_fps,
                display_need_bps,
                probe_frames,
                goodput_bps,
                window_frames,
                window_bytes,
                outbox_frames,
                browser_backlog_frames,
                browser_ack_ahead_frames,
                browser_apply_ms,
                surface_fps,
                avg_surface_frame_bytes,
            ) = {
                let Some(c) = sess.clients.get_mut(&client_id) else {
                    continue;
                };
                c.acks_recv += 1;
                record_ack(c);
                let do_log = c.last_log.elapsed().as_secs_f32() >= 10.0;
                let log_elapsed = c.last_log.elapsed().as_secs_f32().max(1.0e-3);
                let paced_fps = pacing_fps(c);
                let display_need_bps = display_need_bps(c);
                let surface_fps = surface_pacing_fps(c);
                let out = (
                    do_log,
                    c.frames_sent,
                    c.acks_recv,
                    c.rtt_ms,
                    path_rtt_ms(c),
                    window_rtt_ms(c),
                    c.inflight_bytes,
                    c.delivery_bps,
                    c.goodput_bps,
                    c.goodput_jitter_bps,
                    c.max_goodput_jitter_bps,
                    c.avg_frame_bytes,
                    c.avg_paced_frame_bytes,
                    c.avg_preview_frame_bytes,
                    c.display_fps,
                    paced_fps,
                    display_need_bps,
                    c.probe_frames,
                    c.acked_bytes_since_log as f32 / log_elapsed,
                    target_frame_window(c),
                    target_byte_window(c),
                    outbox_queued_frames(c),
                    c.browser_backlog_frames,
                    c.browser_ack_ahead_frames,
                    c.browser_apply_ms,
                    surface_fps,
                    c.avg_surface_frame_bytes,
                );
                if do_log {
                    c.frames_sent = 0;
                    c.acks_recv = 0;
                    c.acked_bytes_since_log = 0;
                    c.last_log = Instant::now();
                }
                out
            };
            if do_log && config.verbose {
                let surf_info = sess.compositor.as_ref().map(|cs| {
                    let surfaces = cs.surfaces.len();
                    let pending = 0usize;
                    let subs: usize = sess
                        .clients
                        .values()
                        .map(|c| c.surface_subscriptions.len())
                        .sum();
                    (surfaces, pending, subs)
                });
                let (surf_count, surf_pending, surf_subs) = surf_info.unwrap_or((0, 0, 0));
                eprintln!(
                    "client {client_id}: sent={frames_sent} acks={acks_recv} rtt={rtt_ms:.0}ms min_rtt={min_rtt_ms:.0}ms eff_rtt={eff_rtt_ms:.0}ms window={window_frames}f/{window_bytes}B probe={probe_frames:.0}f inflight={inflight_bytes}B outbox={outbox_frames}f goodput={goodput_bps:.0}B/s goodput_ewma={goodput_ewma_bps:.0}B/s jitter={goodput_jitter_bps:.0}/{max_goodput_jitter_bps:.0}B/s rate={delivery_bps:.0}B/s avg_frame={avg_frame_bytes:.0}B lead_frame={avg_paced_frame_bytes:.0}B preview_frame={avg_preview_frame_bytes:.0}B need={display_need_bps:.0}B/s display_fps={display_fps:.0} paced_fps={paced_fps:.0} surface_fps={surface_fps:.0} surface_frame={avg_surface_frame_bytes:.0}B backlog={browser_backlog_frames} ack_ahead={browser_ack_ahead_frames} apply={browser_apply_ms:.1}ms | tick_fires={} tick_snaps={} | surfaces={surf_count} subs={surf_subs} pending_req={surf_pending} commits={} encodes={} enc_bytes={} surf_sent={}",
                    sess.tick_fires,
                    sess.tick_snaps,
                    sess.surface_commits,
                    sess.surface_encodes,
                    sess.surface_encode_bytes,
                    sess.surface_frames_sent,
                );
            }
            if do_log {
                sess.tick_fires = 0;
                sess.tick_snaps = 0;
                sess.surface_commits = 0;
                sess.surface_encodes = 0;
                sess.surface_encode_bytes = 0;
                sess.surface_frames_sent = 0;
            }
            nudge_delivery(&state);
            continue;
        }

        if data[0] == C2S_PING {
            // Application-level keepalive — no-op.  Its arrival is enough
            // to keep the connection alive (any received data resets
            // transport-level timeouts).
            continue;
        }

        if data[0] == C2S_DISPLAY_RATE && data.len() >= 3 {
            let fps = u16::from_le_bytes([data[1], data[2]]) as f32;
            if fps > 0.0 {
                let mut sess = state.session.lock().await;
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    c.display_fps = fps;
                }
            }
            nudge_delivery(&state);
            continue;
        }

        if data[0] == C2S_CLIENT_METRICS && data.len() >= 7 {
            let backlog_frames = u16::from_le_bytes([data[1], data[2]]);
            let ack_ahead_frames = u16::from_le_bytes([data[3], data[4]]);
            let apply_ms = u16::from_le_bytes([data[5], data[6]]) as f32 * 0.1;
            let mut sess = state.session.lock().await;
            if let Some(c) = sess.clients.get_mut(&client_id) {
                c.browser_backlog_frames = backlog_frames;
                c.browser_ack_ahead_frames = ack_ahead_frames;
                c.browser_apply_ms = apply_ms;
                c.last_metrics_update = Instant::now();
            }
            nudge_delivery(&state);
            continue;
        }

        // Server-side mouse: client sends structured mouse data, server generates
        // the correct escape sequence using the terminal's current mouse mode/encoding.
        if data[0] == C2S_MOUSE && data.len() >= 9 {
            let pid = u16::from_le_bytes([data[1], data[2]]);
            let type_ = data[3];
            let button = data[4];
            let col = u16::from_le_bytes([data[5], data[6]]);
            let row = u16::from_le_bytes([data[7], data[8]]);
            let sess = state.session.lock().await;
            if let Some(pty) = sess.ptys.get(&pid) {
                let (echo, icanon) = pty.lflag_cache;
                if let Some(seq) = pty
                    .driver
                    .mouse_event(type_, button, col, row, echo, icanon)
                    && let Some(&fd) = state.pty_fds.read().unwrap().get(&pid)
                {
                    pty::pty_write_all(fd, &seq);
                }
            }
            continue;
        }

        if data[0] == C2S_INPUT && data.len() >= 3 {
            let pid = u16::from_le_bytes([data[1], data[2]]);
            let mut need_nudge = false;
            {
                let mut sess = state.session.lock().await;
                if let Some(c) = sess.clients.get_mut(&client_id)
                    && update_client_scroll_state(c, pid, 0)
                    && let Some(pty) = sess.ptys.get_mut(&pid)
                {
                    pty.mark_dirty();
                    need_nudge = true;
                }
            }
            if need_nudge {
                nudge_delivery(&state);
            }
            if let Some(&fd) = state.pty_fds.read().unwrap().get(&pid) {
                pty::pty_write_all(fd, &data[3..]);
            }
            continue;
        }

        if data[0] == C2S_SEARCH && data.len() >= 3 {
            let request_id = u16::from_le_bytes([data[1], data[2]]);
            let query = std::str::from_utf8(&data[3..]).unwrap_or("").trim();
            let mut sess = state.session.lock().await;
            let lead = sess.clients.get(&client_id).and_then(|c| c.lead);
            let mut ranked: Vec<SearchResultRow> = if query.is_empty() {
                Vec::new()
            } else {
                sess.ptys
                    .iter()
                    .filter_map(|(&pty_id, pty)| {
                        pty.driver
                            .search_result(query)
                            .map(|result| SearchResultRow {
                                pty_id,
                                score: result.score,
                                primary_source: result.primary_source,
                                matched_sources: result.matched_sources,
                                context: result.context,
                                scroll_offset: result.scroll_offset,
                            })
                    })
                    .collect()
            };
            ranked.sort_by(|a, b| {
                b.score
                    .cmp(&a.score)
                    .then_with(|| (Some(b.pty_id) == lead).cmp(&(Some(a.pty_id) == lead)))
                    .then_with(|| a.pty_id.cmp(&b.pty_id))
            });
            if let Some(client) = sess.clients.get_mut(&client_id) {
                let _ = try_send_outbox(client, build_search_results_msg(request_id, &ranked));
            }
            continue;
        }

        if data[0] == C2S_SURFACE_CAPTURE && data.len() >= 3 {
            let surface_id = u16::from_le_bytes([data[1], data[2]]);
            // Extended message includes format and quality bytes.
            let format = data.get(3).copied().unwrap_or(CAPTURE_FORMAT_PNG);
            let quality = data.get(4).copied().unwrap_or(0);
            let scale_120 = if data.len() >= 7 {
                u16::from_le_bytes([data[5], data[6]])
            } else {
                0
            };

            let mut reply_msg = vec![S2C_SURFACE_CAPTURE];
            reply_msg.extend_from_slice(&surface_id.to_le_bytes());

            eprintln!("[capture] acquiring lock for surface {surface_id}");
            let (snapshot, command_tx) = {
                let sess = state.session.lock().await;
                eprintln!("[capture] lock acquired");
                let snap = sess
                    .compositor
                    .as_ref()
                    .and_then(|cs| cs.last_pixels.get(&surface_id))
                    .map(|lp| (lp.width, lp.height, lp.pixels.clone()));
                let cmd_tx = sess
                    .compositor
                    .as_ref()
                    .map(|cs| cs.handle.command_tx.clone());
                (snap, cmd_tx)
            };

            // Compositor direct capture (CPU compositing from the per-surface
            // pixel cache).  This is the primary path — it produces correct
            // lossless results for clients that use CPU-mappable DMA-BUFs
            // (Chromium/Brave) or SHM buffers.
            let mut captured: Option<(u32, u32, Vec<u8>)> = None;
            if let Some(ctx) = command_tx {
                captured = request_surface_capture_with_timeout(
                    ctx,
                    surface_id,
                    scale_120,
                    Duration::from_secs(5),
                )
                .await;
            }

            // Fallback: last_pixels from the video pipeline.  Used when
            // the compositor capture returns nothing (no cached buffers).
            if captured.is_none() {
                captured = snapshot.and_then(|(w, h, pixels)| {
                    if pixels.is_dmabuf() {
                        return None;
                    }
                    let rgba = pixels.to_rgba(w, h);
                    if rgba.is_empty() {
                        None
                    } else {
                        Some((w, h, rgba))
                    }
                });
            }

            eprintln!("[capture] acquiring client_tx lock");
            let client_tx = {
                let sess = state.session.lock().await;
                eprintln!("[capture] client_tx lock acquired");
                sess.clients.get(&client_id).map(|c| {
                    (
                        c.tx.clone(),
                        c.outbox_queued_frames.clone(),
                        c.outbox_queued_bytes.clone(),
                    )
                })
            };

            if let Some((w, h, rgba_pixels)) = captured {
                let image_data = encode_capture(&rgba_pixels, w, h, format, quality);
                reply_msg.extend_from_slice(&w.to_le_bytes());
                reply_msg.extend_from_slice(&h.to_le_bytes());
                reply_msg.extend_from_slice(&image_data);
            } else {
                reply_msg.extend_from_slice(&0u32.to_le_bytes());
                reply_msg.extend_from_slice(&0u32.to_le_bytes());
            }

            if let Some((client_tx, queued_frames, queued_bytes)) = client_tx {
                eprintln!("[capture] sending reply: {} bytes", reply_msg.len());
                match try_send_outbox_tracked(&client_tx, &queued_frames, &queued_bytes, reply_msg)
                {
                    Ok(()) => eprintln!("[capture] sent OK"),
                    Err(e) => eprintln!("[capture] try_send failed: {e}"),
                }
            } else {
                eprintln!("[capture] no client_tx");
            }
            continue;
        }

        if data[0] == C2S_QUIT {
            let sess = state.session.lock().await;
            sess.send_to_all(&[S2C_QUIT]);
            drop(sess);
            state.shutdown_notify.notify_waiters();
            break;
        }

        let mut sess = state.session.lock().await;
        let mut need_nudge = false;
        match data[0] {
            C2S_SCROLL if data.len() >= 7 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                let offset = u32::from_le_bytes([data[3], data[4], data[5], data[6]]) as usize;
                if sess.ptys.contains_key(&pid) {
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        update_client_scroll_state(c, pid, offset);
                    }
                    if let Some(pty) = sess.ptys.get_mut(&pid) {
                        pty.mark_dirty();
                        need_nudge = true;
                    }
                }
            }
            C2S_RESIZE if data.len() >= 7 => {
                let entries = data[1..].chunks_exact(6);
                if !entries.remainder().is_empty() {
                    continue;
                }
                let mut touched = Vec::with_capacity((data.len() - 1) / 6);
                for entry in entries {
                    let pid = u16::from_le_bytes([entry[0], entry[1]]);
                    if !sess.ptys.contains_key(&pid) {
                        continue;
                    }
                    let rows = u16::from_le_bytes([entry[2], entry[3]]);
                    let cols = u16::from_le_bytes([entry[4], entry[5]]);
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        if is_unset_view_size(rows, cols) {
                            if c.view_sizes.remove(&pid).is_some() {
                                touched.push(pid);
                            }
                        } else if rows == 0 || cols == 0 {
                            continue;
                        } else {
                            c.view_sizes.insert(pid, (rows, cols));
                            touched.push(pid);
                        }
                    }
                }
                if sess.resize_ptys_to_mediated_sizes(touched) {
                    need_nudge = true;
                }
            }
            C2S_CREATE => {
                // Format: [opcode][rows:2][cols:2][tag_len:2][tag:N][command...]
                let (rows, cols) = if data.len() >= 5 {
                    (
                        u16::from_le_bytes([data[1], data[2]]),
                        u16::from_le_bytes([data[3], data[4]]),
                    )
                } else {
                    (24, 80)
                };
                let tag_len = if data.len() >= 7 {
                    u16::from_le_bytes([data[5], data[6]]) as usize
                } else {
                    0
                };
                let tag = if data.len() >= 7 + tag_len {
                    std::str::from_utf8(&data[7..7 + tag_len]).unwrap_or_default()
                } else {
                    ""
                };
                let cmd_start = 7 + tag_len;
                let dir: Option<String> = None;
                let create_payload = data
                    .get(cmd_start..)
                    .and_then(|bytes| std::str::from_utf8(bytes).ok());
                let command = create_payload
                    .filter(|payload| !payload.contains('\0'))
                    .map(str::trim)
                    .filter(|payload| !payload.is_empty());
                let argv: Option<Vec<&str>> = create_payload
                    .filter(|payload| payload.contains('\0'))
                    .map(|payload| {
                        payload
                            .split('\0')
                            .filter(|arg| !arg.is_empty())
                            .collect::<Vec<_>>()
                    })
                    .filter(|args| !args.is_empty());
                let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
                    continue;
                };
                let socket_name = sess
                    .ensure_compositor(
                        config.verbose,
                        notify_for_compositor.clone(),
                        &config.vaapi_device,
                    )
                    .to_string();
                #[cfg(unix)]
                let pulse_server = sess.pulse_server_path();
                #[cfg(not(unix))]
                let pulse_server: Option<String> = None;
                #[cfg(unix)]
                let pipewire_remote = sess.pipewire_remote_path();
                #[cfg(not(unix))]
                let pipewire_remote: Option<String> = None;
                if let Some(pty) = pty::spawn_pty(
                    &config.shell,
                    &config.shell_flags,
                    rows,
                    cols,
                    id,
                    tag,
                    command,
                    argv.as_deref(),
                    dir.as_deref(),
                    config.scrollback,
                    state.clone(),
                    Some(&socket_name),
                    pulse_server.as_deref(),
                    pipewire_remote.as_deref(),
                ) {
                    let mut msg = Vec::with_capacity(3 + pty.tag.len());
                    msg.push(S2C_CREATED);
                    msg.extend_from_slice(&id.to_le_bytes());
                    msg.extend_from_slice(pty.tag.as_bytes());
                    sess.ptys.insert(id, pty);
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        c.lead = Some(id);
                        c.view_sizes.insert(id, (rows, cols));
                        subscribe_client_to(c, id);
                        reset_inflight(c);
                    }
                    sess.send_to_all(&msg);
                    need_nudge = true;
                }
            }
            C2S_CREATE_N => {
                // Format: [opcode][nonce:2][rows:2][cols:2][tag_len:2][tag:N][command...]
                let nonce = if data.len() >= 3 {
                    u16::from_le_bytes([data[1], data[2]])
                } else {
                    0
                };
                let (rows, cols) = if data.len() >= 7 {
                    (
                        u16::from_le_bytes([data[3], data[4]]),
                        u16::from_le_bytes([data[5], data[6]]),
                    )
                } else {
                    (24, 80)
                };
                let tag_len = if data.len() >= 9 {
                    u16::from_le_bytes([data[7], data[8]]) as usize
                } else {
                    0
                };
                let tag = if data.len() >= 9 + tag_len {
                    std::str::from_utf8(&data[9..9 + tag_len]).unwrap_or_default()
                } else {
                    ""
                };
                let cmd_start = 9 + tag_len;
                let dir: Option<String> = None;
                let create_payload = data
                    .get(cmd_start..)
                    .and_then(|bytes| std::str::from_utf8(bytes).ok());
                let command = create_payload
                    .filter(|payload| !payload.contains('\0'))
                    .map(str::trim)
                    .filter(|payload| !payload.is_empty());
                let argv: Option<Vec<&str>> = create_payload
                    .filter(|payload| payload.contains('\0'))
                    .map(|payload| {
                        payload
                            .split('\0')
                            .filter(|arg| !arg.is_empty())
                            .collect::<Vec<_>>()
                    })
                    .filter(|args| !args.is_empty());
                let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
                    continue;
                };
                let socket_name = sess
                    .ensure_compositor(
                        config.verbose,
                        notify_for_compositor.clone(),
                        &config.vaapi_device,
                    )
                    .to_string();
                #[cfg(unix)]
                let pulse_server = sess.pulse_server_path();
                #[cfg(not(unix))]
                let pulse_server: Option<String> = None;
                #[cfg(unix)]
                let pipewire_remote = sess.pipewire_remote_path();
                #[cfg(not(unix))]
                let pipewire_remote: Option<String> = None;
                if let Some(pty) = pty::spawn_pty(
                    &config.shell,
                    &config.shell_flags,
                    rows,
                    cols,
                    id,
                    tag,
                    command,
                    argv.as_deref(),
                    dir.as_deref(),
                    config.scrollback,
                    state.clone(),
                    Some(&socket_name),
                    pulse_server.as_deref(),
                    pipewire_remote.as_deref(),
                ) {
                    let tag_bytes = pty.tag.as_bytes();
                    let mut nonce_msg = Vec::with_capacity(5 + tag_bytes.len());
                    nonce_msg.push(S2C_CREATED_N);
                    nonce_msg.extend_from_slice(&nonce.to_le_bytes());
                    nonce_msg.extend_from_slice(&id.to_le_bytes());
                    nonce_msg.extend_from_slice(tag_bytes);
                    let mut broadcast_msg = Vec::with_capacity(3 + tag_bytes.len());
                    broadcast_msg.push(S2C_CREATED);
                    broadcast_msg.extend_from_slice(&id.to_le_bytes());
                    broadcast_msg.extend_from_slice(tag_bytes);
                    sess.ptys.insert(id, pty);
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        c.lead = Some(id);
                        c.view_sizes.insert(id, (rows, cols));
                        subscribe_client_to(c, id);
                        reset_inflight(c);
                        let _ = try_send_outbox(c, nonce_msg);
                    }
                    for (&cid, c) in sess.clients.iter() {
                        if cid != client_id {
                            let _ = try_send_outbox(c, broadcast_msg.clone());
                        }
                    }
                    need_nudge = true;
                }
            }
            C2S_CREATE_AT => {
                // Format: [opcode][rows:2][cols:2][tag_len:2][tag:N][src_pty_id:2]
                let (rows, cols) = if data.len() >= 5 {
                    (
                        u16::from_le_bytes([data[1], data[2]]),
                        u16::from_le_bytes([data[3], data[4]]),
                    )
                } else {
                    (24, 80)
                };
                let tag_len = if data.len() >= 7 {
                    u16::from_le_bytes([data[5], data[6]]) as usize
                } else {
                    0
                };
                let tag = if data.len() >= 7 + tag_len {
                    std::str::from_utf8(&data[7..7 + tag_len]).unwrap_or_default()
                } else {
                    ""
                };
                let src_start = 7 + tag_len;
                let dir = if data.len() >= src_start + 2 {
                    let src_id = u16::from_le_bytes([data[src_start], data[src_start + 1]]);
                    sess.ptys.get(&src_id).and_then(|p| pty::pty_cwd(&p.handle))
                } else {
                    None
                };
                let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
                    continue;
                };
                let socket_name = sess
                    .ensure_compositor(
                        config.verbose,
                        notify_for_compositor.clone(),
                        &config.vaapi_device,
                    )
                    .to_string();
                #[cfg(unix)]
                let pulse_server = sess.pulse_server_path();
                #[cfg(not(unix))]
                let pulse_server: Option<String> = None;
                #[cfg(unix)]
                let pipewire_remote = sess.pipewire_remote_path();
                #[cfg(not(unix))]
                let pipewire_remote: Option<String> = None;
                if let Some(pty) = pty::spawn_pty(
                    &config.shell,
                    &config.shell_flags,
                    rows,
                    cols,
                    id,
                    tag,
                    None,
                    None,
                    dir.as_deref(),
                    config.scrollback,
                    state.clone(),
                    Some(&socket_name),
                    pulse_server.as_deref(),
                    pipewire_remote.as_deref(),
                ) {
                    let mut msg = Vec::with_capacity(3 + pty.tag.len());
                    msg.push(S2C_CREATED);
                    msg.extend_from_slice(&id.to_le_bytes());
                    msg.extend_from_slice(pty.tag.as_bytes());
                    sess.ptys.insert(id, pty);
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        c.lead = Some(id);
                        c.view_sizes.insert(id, (rows, cols));
                        subscribe_client_to(c, id);
                        reset_inflight(c);
                    }
                    sess.send_to_all(&msg);
                    need_nudge = true;
                }
            }
            C2S_CREATE2 => {
                if data.len() < 10 {
                    continue;
                }
                let nonce = u16::from_le_bytes([data[1], data[2]]);
                let rows = u16::from_le_bytes([data[3], data[4]]);
                let cols = u16::from_le_bytes([data[5], data[6]]);
                let features = data[7];
                let tag_len = u16::from_le_bytes([data[8], data[9]]) as usize;
                let tag = if data.len() >= 10 + tag_len {
                    std::str::from_utf8(&data[10..10 + tag_len]).unwrap_or_default()
                } else {
                    ""
                };
                let mut cursor = 10 + tag_len;
                let dir = if features & CREATE2_HAS_SRC_PTY != 0 && data.len() >= cursor + 2 {
                    let src_id = u16::from_le_bytes([data[cursor], data[cursor + 1]]);
                    cursor += 2;
                    sess.ptys.get(&src_id).and_then(|p| pty::pty_cwd(&p.handle))
                } else {
                    None
                };
                let create_payload = if features & CREATE2_HAS_COMMAND != 0 {
                    data.get(cursor..).and_then(|b| std::str::from_utf8(b).ok())
                } else {
                    None
                };
                let command = create_payload
                    .filter(|p| !p.contains('\0'))
                    .map(str::trim)
                    .filter(|p| !p.is_empty());
                let argv: Option<Vec<&str>> = create_payload
                    .filter(|p| p.contains('\0'))
                    .map(|p| p.split('\0').filter(|a| !a.is_empty()).collect::<Vec<_>>())
                    .filter(|a| !a.is_empty());
                let Some(id) = sess.allocate_pty_id(config.max_ptys) else {
                    continue;
                };
                let socket_name = sess
                    .ensure_compositor(
                        config.verbose,
                        notify_for_compositor.clone(),
                        &config.vaapi_device,
                    )
                    .to_string();
                #[cfg(unix)]
                let pulse_server = sess.pulse_server_path();
                #[cfg(not(unix))]
                let pulse_server: Option<String> = None;
                #[cfg(unix)]
                let pipewire_remote = sess.pipewire_remote_path();
                #[cfg(not(unix))]
                let pipewire_remote: Option<String> = None;
                if let Some(pty) = pty::spawn_pty(
                    &config.shell,
                    &config.shell_flags,
                    rows,
                    cols,
                    id,
                    tag,
                    command,
                    argv.as_deref(),
                    dir.as_deref(),
                    config.scrollback,
                    state.clone(),
                    Some(&socket_name),
                    pulse_server.as_deref(),
                    pipewire_remote.as_deref(),
                ) {
                    let tag_bytes = pty.tag.as_bytes();
                    let mut nonce_msg = Vec::with_capacity(5 + tag_bytes.len());
                    nonce_msg.push(S2C_CREATED_N);
                    nonce_msg.extend_from_slice(&nonce.to_le_bytes());
                    nonce_msg.extend_from_slice(&id.to_le_bytes());
                    nonce_msg.extend_from_slice(tag_bytes);
                    let mut broadcast_msg = Vec::with_capacity(3 + tag_bytes.len());
                    broadcast_msg.push(S2C_CREATED);
                    broadcast_msg.extend_from_slice(&id.to_le_bytes());
                    broadcast_msg.extend_from_slice(tag_bytes);
                    sess.ptys.insert(id, pty);
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        c.lead = Some(id);
                        c.view_sizes.insert(id, (rows, cols));
                        subscribe_client_to(c, id);
                        reset_inflight(c);
                        let _ = try_send_outbox(c, nonce_msg);
                    }
                    for (&cid, c) in sess.clients.iter() {
                        if cid != client_id {
                            let _ = try_send_outbox(c, broadcast_msg.clone());
                        }
                    }
                    need_nudge = true;
                }
            }
            C2S_SURFACE_INPUT if data.len() >= 8 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                let keycode = u32::from_le_bytes([data[3], data[4], data[5], data[6]]);
                let pressed = data[7] != 0;
                if let Some(client) = sess.clients.get_mut(&client_id) {
                    if pressed {
                        client.pressed_surface_keys.insert(keycode);
                    } else {
                        client.pressed_surface_keys.remove(&keycode);
                    }
                }
                if let Some(cs) = sess.compositor.as_mut() {
                    let _ = cs.handle.command_tx.send(CompositorCommand::KeyInput {
                        surface_id,
                        keycode,
                        pressed,
                    });
                    cs.handle.wake();
                    state.delivery_notify.notify_one();
                }
            }
            C2S_SURFACE_TEXT if data.len() >= 3 => {
                let _surface_id = u16::from_le_bytes([data[1], data[2]]);
                if let Ok(text) = std::str::from_utf8(&data[3..])
                    && let Some(cs) = sess.compositor.as_mut()
                {
                    let _ = cs.handle.command_tx.send(CompositorCommand::TextInput {
                        text: text.to_string(),
                    });
                    cs.handle.wake();
                    state.delivery_notify.notify_one();
                }
            }
            C2S_SURFACE_POINTER if data.len() >= 9 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                let ptype = data[3];
                let button = data[4];
                let x = u16::from_le_bytes([data[5], data[6]]) as f64;
                let y = u16::from_le_bytes([data[7], data[8]]) as f64;
                if let Some(cs) = sess.compositor.as_mut() {
                    match ptype {
                        0 | 1 => {
                            let _ = cs.handle.command_tx.send(CompositorCommand::PointerMotion {
                                surface_id,
                                x,
                                y,
                            });
                            let _ = cs.handle.command_tx.send(CompositorCommand::PointerButton {
                                surface_id,
                                button: match button {
                                    1 => 0x112,
                                    2 => 0x111,
                                    _ => 0x110,
                                },
                                pressed: ptype == 0,
                            });
                        }
                        2 => {
                            let _ = cs.handle.command_tx.send(CompositorCommand::PointerMotion {
                                surface_id,
                                x,
                                y,
                            });
                        }
                        _ => {}
                    }
                    cs.handle.wake();
                }
                state.delivery_notify.notify_one();
            }
            C2S_SURFACE_POINTER_AXIS if data.len() >= 8 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                let axis = data[3];
                let value_x100 = i32::from_le_bytes([data[4], data[5], data[6], data[7]]);
                let value = value_x100 as f64 / 100.0;
                if let Some(cs) = sess.compositor.as_mut() {
                    let _ = cs.handle.command_tx.send(CompositorCommand::PointerAxis {
                        surface_id,
                        axis,
                        value,
                    });
                    cs.handle.wake();
                }
                state.delivery_notify.notify_one();
            }
            C2S_SURFACE_RESIZE if data.len() >= 9 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                let width = u16::from_le_bytes([data[3], data[4]]);
                let height = u16::from_le_bytes([data[5], data[6]]);
                // Scale in 1/120th units (Wayland convention): 240 = 2×.
                let scale_120 = u16::from_le_bytes([data[7], data[8]]);
                if state.config.verbose {
                    eprintln!(
                        "C2S_SURFACE_RESIZE: cid={client_id} sid={surface_id} {width}x{height} scale={scale_120}"
                    );
                }
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    if is_unset_view_size(width, height) {
                        c.surface_view_sizes.remove(&surface_id);
                    } else if width > 0 && height > 0 {
                        c.surface_view_sizes
                            .insert(surface_id, (width, height, scale_120));
                    }
                }
                sess.resize_surfaces_to_mediated_sizes(
                    std::iter::once(surface_id),
                    &state.config.surface_encoders,
                );
            }
            C2S_SURFACE_FOCUS if data.len() >= 3 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                if state.config.verbose {
                    eprintln!("C2S_SURFACE_FOCUS: cid={client_id} sid={surface_id}");
                }
                if let Some(cs) = sess.compositor.as_ref() {
                    let _ = cs
                        .handle
                        .command_tx
                        .send(CompositorCommand::SurfaceFocus { surface_id });
                }
            }
            C2S_SURFACE_CLOSE if data.len() >= 3 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                if let Some(cs) = sess.compositor.as_ref() {
                    let _ = cs
                        .handle
                        .command_tx
                        .send(CompositorCommand::SurfaceClose { surface_id });
                    cs.handle.wake();
                }
            }
            C2S_SURFACE_SUBSCRIBE if data.len() >= 3 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                // Extended fields (backward-compatible: absent = 0 = defaults).
                let codec_support = if data.len() >= 4 { data[3] } else { 0 };
                let quality_wire = if data.len() >= 5 { data[4] } else { 0 };
                if state.config.verbose {
                    eprintln!(
                        "C2S_SURFACE_SUBSCRIBE: cid={client_id} surface={surface_id} codec={codec_support:#04x} quality={quality_wire}"
                    );
                }
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    let was_subscribed = !c.surface_subscriptions.insert(surface_id);
                    c.surface_needs_keyframe = true;
                    // Reset burst window so the first frames after (re)subscribe
                    // bypass time-based pacing and flow at wire speed.
                    c.surface_burst_remaining = SURFACE_BURST_FRAMES;

                    // Track per-surface codec/quality overrides.
                    let old_codec = c
                        .surface_codec_overrides
                        .get(&surface_id)
                        .copied()
                        .unwrap_or(0);
                    let old_quality = c.surface_quality_overrides.get(&surface_id).copied();
                    let new_quality = SurfaceQuality::from_wire(quality_wire);

                    if codec_support != 0 {
                        c.surface_codec_overrides.insert(surface_id, codec_support);
                    } else {
                        c.surface_codec_overrides.remove(&surface_id);
                    }
                    if let Some(q) = new_quality {
                        c.surface_quality_overrides.insert(surface_id, q);
                    } else {
                        c.surface_quality_overrides.remove(&surface_id);
                    }

                    // Force encoder recreation if preferences changed on resubscribe.
                    if was_subscribed && (codec_support != old_codec || new_quality != old_quality)
                    {
                        c.surface_encoders.remove(&surface_id);
                    }
                }
                state.delivery_notify.notify_one();
            }
            C2S_SURFACE_UNSUBSCRIBE if data.len() >= 3 => {
                let surface_id = u16::from_le_bytes([data[1], data[2]]);
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    c.surface_subscriptions.remove(&surface_id);
                    c.surface_view_sizes.remove(&surface_id);
                    c.surface_codec_overrides.remove(&surface_id);
                    c.surface_quality_overrides.remove(&surface_id);
                }
                sess.resize_surfaces_to_mediated_sizes(
                    std::iter::once(surface_id),
                    &state.config.surface_encoders,
                );
            }
            #[cfg(unix)]
            C2S_AUDIO_SUBSCRIBE if data.len() >= 3 => {
                let bitrate_kbps = u16::from_le_bytes([data[1], data[2]]);
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    c.audio_subscribed = true;
                    c.audio_bitrate_kbps = bitrate_kbps;
                    if state.config.verbose {
                        eprintln!(
                            "C2S_AUDIO_SUBSCRIBE: cid={client_id} bitrate_kbps={bitrate_kbps}"
                        );
                    }
                    // Send ring buffer catch-up frames via the audio channel.
                    if let Some(cs) = sess.compositor.as_mut()
                        && let Some(ref ap) = cs.audio_pipeline
                    {
                        let msgs: Vec<_> = ap.ring_frames().map(audio::msg_audio_frame).collect();
                        if let Some(c) = sess.clients.get(&client_id) {
                            for msg in msgs {
                                let _ = c.audio_tx.try_send(msg);
                            }
                        }
                    }
                }
                // Recompute the effective audio bitrate across all
                // subscribed clients (use the max requested bitrate).
                if let Some(cs) = sess.compositor.as_ref()
                    && let Some(ref ap) = cs.audio_pipeline
                {
                    let max_kbps = sess
                        .clients
                        .values()
                        .filter(|c| c.audio_subscribed)
                        .map(|c| c.audio_bitrate_kbps)
                        .max()
                        .unwrap_or(0);
                    let bitrate = if max_kbps > 0 {
                        max_kbps as i32 * 1000
                    } else {
                        audio::DEFAULT_BITRATE
                    };
                    ap.set_bitrate(bitrate);
                }
                state.delivery_notify.notify_one();
            }
            #[cfg(unix)]
            C2S_AUDIO_UNSUBSCRIBE if !data.is_empty() => {
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    c.audio_subscribed = false;
                    c.audio_bitrate_kbps = 0;
                    if state.config.verbose {
                        eprintln!("C2S_AUDIO_UNSUBSCRIBE: cid={client_id}");
                    }
                }
                // Recompute effective bitrate after unsubscribe.
                if let Some(cs) = sess.compositor.as_ref()
                    && let Some(ref ap) = cs.audio_pipeline
                {
                    let max_kbps = sess
                        .clients
                        .values()
                        .filter(|c| c.audio_subscribed)
                        .map(|c| c.audio_bitrate_kbps)
                        .max()
                        .unwrap_or(0);
                    let bitrate = if max_kbps > 0 {
                        max_kbps as i32 * 1000
                    } else {
                        audio::DEFAULT_BITRATE
                    };
                    ap.set_bitrate(bitrate);
                }
            }
            C2S_SURFACE_ACK if data.len() >= 3 => {
                // Surface ACKs feed shared RTT / delivery_bps / goodput_bps
                // from a separate inflight queue so they don't corrupt
                // terminal frame-size averages or probe_frames.
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    c.acks_recv += 1;
                    record_surface_ack(c);
                }
                state.delivery_notify.notify_one();
            }
            C2S_CLIENT_FEATURES if data.len() >= 2 => {
                // Byte 0: codec_support bitmask.  Future bytes are ignored
                // if unknown, defaulting to 0 when absent.
                let codec_support = data[1];
                if let Some(c) = sess.clients.get_mut(&client_id) {
                    c.surface_codec_support = codec_support;
                }
            }
            C2S_CLIPBOARD_SET if data.len() >= 5 => {
                let mime_len = u16::from_le_bytes([data[1], data[2]]) as usize;
                if data.len() >= 3 + mime_len + 4 {
                    let mime = std::str::from_utf8(&data[3..3 + mime_len])
                        .unwrap_or("text/plain")
                        .to_string();
                    let data_len = u32::from_le_bytes([
                        data[3 + mime_len],
                        data[4 + mime_len],
                        data[5 + mime_len],
                        data[6 + mime_len],
                    ]) as usize;
                    let payload_start = 7 + mime_len;
                    if data.len() >= payload_start + data_len {
                        let payload = data[payload_start..payload_start + data_len].to_vec();
                        if let Some(cs) = sess.compositor.as_ref() {
                            let _ = cs
                                .handle
                                .command_tx
                                .send(CompositorCommand::ClipboardOffer {
                                    mime_type: mime,
                                    data: payload,
                                });
                        }
                    }
                }
            }
            C2S_CLIPBOARD_LIST if !data.is_empty() => {
                if let Some(cs) = sess.compositor.as_ref() {
                    let command_tx = cs.handle.command_tx.clone();
                    let client_tx = sess.clients.get(&client_id).map(|c| {
                        (
                            c.tx.clone(),
                            c.outbox_queued_frames.clone(),
                            c.outbox_queued_bytes.clone(),
                        )
                    });
                    if let Some((client_tx, queued_frames, queued_bytes)) = client_tx {
                        tokio::task::spawn_blocking(move || {
                            let (tx, rx) = std::sync::mpsc::sync_channel(1);
                            if command_tx
                                .send(CompositorCommand::ClipboardListMimes { reply: tx })
                                .is_ok()
                                && let Ok(mimes) = rx.recv_timeout(Duration::from_secs(2))
                            {
                                let _ = try_send_outbox_tracked(
                                    &client_tx,
                                    &queued_frames,
                                    &queued_bytes,
                                    msg_s2c_clipboard_list(&mimes),
                                );
                            }
                        });
                    }
                } else {
                    // No compositor — respond with empty list.
                    if let Some(c) = sess.clients.get(&client_id) {
                        let _ = try_send_outbox(c, msg_s2c_clipboard_list(&[]));
                    }
                }
            }
            C2S_CLIPBOARD_GET if data.len() >= 3 => {
                let mime_len = u16::from_le_bytes([data[1], data[2]]) as usize;
                if data.len() >= 3 + mime_len {
                    let mime = std::str::from_utf8(&data[3..3 + mime_len])
                        .unwrap_or("text/plain")
                        .to_string();
                    if let Some(cs) = sess.compositor.as_ref() {
                        let command_tx = cs.handle.command_tx.clone();
                        let client_tx = sess.clients.get(&client_id).map(|c| {
                            (
                                c.tx.clone(),
                                c.outbox_queued_frames.clone(),
                                c.outbox_queued_bytes.clone(),
                            )
                        });
                        if let Some((client_tx, queued_frames, queued_bytes)) = client_tx {
                            tokio::task::spawn_blocking(move || {
                                let (tx, rx) = std::sync::mpsc::sync_channel(1);
                                if command_tx
                                    .send(CompositorCommand::ClipboardGet {
                                        mime_type: mime.clone(),
                                        reply: tx,
                                    })
                                    .is_ok()
                                    && let Ok(content) = rx.recv_timeout(Duration::from_secs(2))
                                {
                                    let data = content.unwrap_or_default();
                                    let _ = try_send_outbox_tracked(
                                        &client_tx,
                                        &queued_frames,
                                        &queued_bytes,
                                        msg_s2c_clipboard_content(&mime, &data),
                                    );
                                }
                            });
                        }
                    } else {
                        // No compositor — respond with empty clipboard.
                        if let Some(c) = sess.clients.get(&client_id) {
                            let _ = try_send_outbox(c, msg_s2c_clipboard_content(&mime, &[]));
                        }
                    }
                }
            }
            C2S_SURFACE_LIST if !data.is_empty() => {
                let msg = sess.surface_list_msg();
                if let Some(c) = sess.clients.get(&client_id) {
                    let _ = try_send_outbox(c, msg);
                }
            }
            C2S_FOCUS if data.len() >= 3 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                if sess.ptys.contains_key(&pid) {
                    let old_pid = sess.clients.get(&client_id).and_then(|c| c.lead);
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        c.lead = Some(pid);
                        subscribe_client_to(c, pid);
                        if old_pid == Some(pid) {
                            update_client_scroll_state(c, pid, 0);
                        } else {
                            reset_inflight(c);
                        }
                    }
                    if let Some(pty) = sess.ptys.get_mut(&pid) {
                        pty.mark_dirty();
                        need_nudge = true;
                    }
                }
            }
            C2S_SUBSCRIBE if data.len() >= 3 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                if sess.ptys.contains_key(&pid) {
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        subscribe_client_to(c, pid);
                    }
                    if let Some(pty) = sess.ptys.get_mut(&pid) {
                        pty.mark_dirty();
                    }
                    need_nudge = true;
                }
            }
            C2S_UNSUBSCRIBE if data.len() >= 3 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                if sess.ptys.contains_key(&pid) {
                    let mut touched = Vec::new();
                    if let Some(c) = sess.clients.get_mut(&client_id) {
                        if unsubscribe_client_from(c, pid) {
                            touched.push(pid);
                        }
                        reset_inflight(c);
                    }
                    if sess.resize_ptys_to_mediated_sizes(touched) {
                        need_nudge = true;
                    }
                }
            }
            C2S_RESTART if data.len() >= 3 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                let restart_info = sess
                    .ptys
                    .get(&pid)
                    .filter(|p| p.exited)
                    .map(|p| (p.driver.size(), p.command.clone(), p.tag.clone()));
                if let Some(((rows, cols), command, tag)) = restart_info {
                    let wayland_display = sess
                        .compositor
                        .as_ref()
                        .map(|cs| cs.handle.socket_name.clone());
                    #[cfg(unix)]
                    let pulse_server = sess.pulse_server_path();
                    #[cfg(not(unix))]
                    let pulse_server: Option<String> = None;
                    #[cfg(unix)]
                    let pipewire_remote = sess.pipewire_remote_path();
                    #[cfg(not(unix))]
                    let pipewire_remote: Option<String> = None;
                    if let Some((new_handle, reader, byte_rx)) = pty::respawn_child(
                        &state.config.shell,
                        &state.config.shell_flags,
                        rows,
                        cols,
                        pid,
                        command.as_deref(),
                        state.clone(),
                        wayland_display.as_deref(),
                        pulse_server.as_deref(),
                        pipewire_remote.as_deref(),
                    ) {
                        let Some(pty) = sess.ptys.get_mut(&pid) else {
                            break;
                        };
                        pty.handle = new_handle;
                        pty.reader_handle = reader;
                        pty.byte_rx = byte_rx;
                        pty.driver.reset_modes();
                        pty.exited = false;
                        pty.exit_status = blit_remote::EXIT_STATUS_UNKNOWN;
                        pty.lflag_cache = pty::pty_lflag(&pty.handle);
                        pty.lflag_last = Instant::now();
                        pty.mark_dirty();
                        if let Some(c) = sess.clients.get_mut(&client_id) {
                            c.lead = Some(pid);
                            subscribe_client_to(c, pid);
                            update_client_scroll_state(c, pid, 0);
                            reset_inflight(c);
                        }
                        let mut msg = Vec::with_capacity(3 + tag.len());
                        msg.push(S2C_CREATED);
                        msg.extend_from_slice(&pid.to_le_bytes());
                        msg.extend_from_slice(tag.as_bytes());
                        sess.send_to_all(&msg);
                        need_nudge = true;
                    }
                }
            }
            C2S_READ if data.len() >= 13 => {
                let nonce = u16::from_le_bytes([data[1], data[2]]);
                let pid = u16::from_le_bytes([data[3], data[4]]);
                let req_offset = u32::from_le_bytes([data[5], data[6], data[7], data[8]]) as usize;
                let req_limit =
                    u32::from_le_bytes([data[9], data[10], data[11], data[12]]) as usize;
                let flags = data.get(13).copied().unwrap_or(0);
                let ansi = flags & READ_ANSI != 0;
                let tail = flags & READ_TAIL != 0;

                if let Some(pty) = sess.ptys.get_mut(&pid) {
                    let (rows, _cols) = pty.driver.size();
                    let viewport = take_snapshot(pty);
                    let scrollback_lines = viewport.scrollback_lines() as usize;
                    let total_lines = scrollback_lines + rows as usize;

                    let extract = |f: &FrameState| -> String {
                        if ansi {
                            f.get_ansi_text()
                        } else {
                            f.get_all_text()
                        }
                    };

                    let mut all_lines: Vec<String> =
                        Vec::with_capacity(scrollback_lines + rows as usize);

                    let mut scroll_offset = scrollback_lines;
                    while scroll_offset > 0 {
                        let frame = pty.driver.scrollback_frame(scroll_offset);
                        let page = extract(&frame);
                        let page_lines: Vec<&str> = page.lines().collect();
                        let take = if scroll_offset < rows as usize {
                            scroll_offset.min(page_lines.len())
                        } else {
                            page_lines.len()
                        };
                        for line in &page_lines[..take] {
                            all_lines.push(line.to_string());
                        }
                        if scroll_offset <= rows as usize {
                            break;
                        }
                        scroll_offset = scroll_offset.saturating_sub(rows as usize);
                    }

                    for line in extract(&viewport).lines() {
                        all_lines.push(line.to_string());
                    }

                    let (start, end) = if tail {
                        let end = all_lines.len().saturating_sub(req_offset);
                        let start = if req_limit == 0 {
                            0
                        } else {
                            end.saturating_sub(req_limit)
                        };
                        (start, end)
                    } else {
                        let start = req_offset.min(all_lines.len());
                        let end = if req_limit == 0 {
                            all_lines.len()
                        } else {
                            (start + req_limit).min(all_lines.len())
                        };
                        (start, end)
                    };
                    let text = all_lines[start..end].join("\n");

                    let mut msg = Vec::with_capacity(13 + text.len());
                    msg.push(S2C_TEXT);
                    msg.extend_from_slice(&nonce.to_le_bytes());
                    msg.extend_from_slice(&pid.to_le_bytes());
                    msg.extend_from_slice(&(total_lines as u32).to_le_bytes());
                    msg.extend_from_slice(&(start as u32).to_le_bytes());
                    msg.extend_from_slice(text.as_bytes());
                    if let Some(client) = sess.clients.get(&client_id) {
                        let _ = try_send_outbox(client, msg);
                    }
                }
            }
            C2S_COPY_RANGE if data.len() >= 18 => {
                let nonce = u16::from_le_bytes([data[1], data[2]]);
                let pid = u16::from_le_bytes([data[3], data[4]]);
                let start_tail = u32::from_le_bytes([data[5], data[6], data[7], data[8]]);
                let start_col = u16::from_le_bytes([data[9], data[10]]);
                let end_tail = u32::from_le_bytes([data[11], data[12], data[13], data[14]]);
                let end_col = u16::from_le_bytes([data[15], data[16]]);

                if let Some(pty) = sess.ptys.get(&pid) {
                    let text = pty
                        .driver
                        .get_text_range(start_tail, start_col, end_tail, end_col);
                    let total_lines = pty.driver.total_lines();

                    let mut msg = Vec::with_capacity(13 + text.len());
                    msg.push(S2C_TEXT);
                    msg.extend_from_slice(&nonce.to_le_bytes());
                    msg.extend_from_slice(&pid.to_le_bytes());
                    msg.extend_from_slice(&total_lines.to_le_bytes());
                    msg.extend_from_slice(&start_tail.to_le_bytes());
                    msg.extend_from_slice(text.as_bytes());
                    if let Some(client) = sess.clients.get(&client_id) {
                        let _ = try_send_outbox(client, msg);
                    }
                }
            }
            C2S_KILL if data.len() >= 7 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                let signal = i32::from_le_bytes([data[3], data[4], data[5], data[6]]);
                if let Some(pty) = sess.ptys.get(&pid)
                    && !pty.exited
                {
                    pty::kill_pty(&pty.handle, signal);
                }
            }
            C2S_CLOSE if data.len() >= 3 => {
                let pid = u16::from_le_bytes([data[1], data[2]]);
                if let Some(pty) = sess.ptys.remove(&pid) {
                    if !pty.exited {
                        state.pty_fds.write().unwrap().remove(&pid);
                        drop(pty.reader_handle);
                        pty::close_pty(&pty.handle);
                    }
                    for client in sess.clients.values_mut() {
                        unsubscribe_client_from(client, pid);
                    }
                    let mut msg = vec![S2C_CLOSED];
                    msg.extend_from_slice(&pid.to_le_bytes());
                    sess.send_to_all(&msg);
                }
            }
            _ => {}
        }
        drop(sess);
        if need_nudge {
            nudge_delivery(&state);
        }
    }

    {
        let mut sess = state.session.lock().await;
        let mut need_nudge = false;
        let client = sess.clients.remove(&client_id);
        let affected_ptys = client
            .as_ref()
            .map(|c| c.view_sizes.keys().copied().collect::<Vec<_>>())
            .unwrap_or_default();
        let affected_surfaces = client
            .as_ref()
            .map(|c| c.surface_view_sizes.keys().copied().collect::<Vec<_>>())
            .unwrap_or_default();
        if sess.resize_ptys_to_mediated_sizes(affected_ptys) {
            need_nudge = true;
        }
        sess.resize_surfaces_to_mediated_sizes(affected_surfaces, &state.config.surface_encoders);
        // Release any keys this client was holding when it disconnected.
        // Without this, modifier keys (Shift, Ctrl, etc.) stay stuck and
        // regular keys auto-repeat forever in the compositor.
        if let Some(ref client) = client
            && !client.pressed_surface_keys.is_empty()
            && let Some(cs) = sess.compositor.as_mut()
        {
            let keycodes: Vec<u32> = client.pressed_surface_keys.iter().copied().collect();
            let _ = cs
                .handle
                .command_tx
                .send(CompositorCommand::ReleaseKeys { keycodes });
            cs.handle.wake();
        }
        drop(sess);
        if need_nudge {
            nudge_delivery(&state);
        }
    }
    sender.abort();
    if state.config.verbose {
        eprintln!("client disconnected");
    }
}

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

    fn test_client_with_capacity(capacity: usize) -> (ClientState, mpsc::Receiver<Vec<u8>>) {
        let (tx, rx) = mpsc::channel(capacity);
        let (audio_tx, _audio_rx) = mpsc::channel(AUDIO_OUTBOX_CAPACITY);
        let client = ClientState {
            tx,
            outbox_queued_frames: Arc::new(AtomicUsize::new(0)),
            outbox_queued_bytes: Arc::new(AtomicUsize::new(0)),
            audio_tx,
            lead: None,
            subscriptions: HashSet::new(),
            surface_subscriptions: HashSet::new(),
            audio_subscribed: false,
            #[cfg(unix)]
            audio_bitrate_kbps: 0,
            view_sizes: HashMap::new(),
            scroll_offsets: HashMap::new(),
            scroll_caches: HashMap::new(),
            last_sent: HashMap::new(),
            preview_next_send_at: HashMap::new(),
            rtt_ms: 50.0,
            min_rtt_ms: 50.0,
            display_fps: 60.0,
            delivery_bps: 262_144.0,
            goodput_bps: 262_144.0,
            goodput_jitter_bps: 0.0,
            max_goodput_jitter_bps: 0.0,
            last_goodput_sample_bps: 0.0,
            avg_frame_bytes: 1_024.0,
            avg_paced_frame_bytes: 1_024.0,
            avg_preview_frame_bytes: 1_024.0,
            avg_surface_frame_bytes: 8_192.0,
            inflight_bytes: 0,
            inflight_frames: VecDeque::new(),
            next_send_at: Instant::now(),
            probe_frames: 0.0,
            frames_sent: 0,
            acks_recv: 0,
            acked_bytes_since_log: 0,
            browser_backlog_frames: 0,
            browser_ack_ahead_frames: 0,
            browser_apply_ms: 0.0,
            last_metrics_update: Instant::now(),
            last_log: Instant::now(),
            goodput_window_bytes: 0,
            goodput_window_start: Instant::now(),
            surface_next_send_at: Instant::now(),
            surface_needs_keyframe: true,
            surface_burst_remaining: SURFACE_BURST_FRAMES,
            surface_inflight_frames: VecDeque::new(),
            surface_encoders: HashMap::new(),
            surface_encodes_in_flight: HashSet::new(),
            surface_last_encoded_gen: HashMap::new(),
            surface_view_sizes: HashMap::new(),
            surface_codec_support: 0,
            surface_codec_overrides: HashMap::new(),
            surface_quality_overrides: HashMap::new(),
            pressed_surface_keys: HashSet::new(),
        };
        (client, rx)
    }

    fn test_client() -> ClientState {
        let (client, _rx) = test_client_with_capacity(OUTBOX_CAPACITY);
        client
    }

    fn fill_inflight(client: &mut ClientState, frames: usize, bytes_per_frame: usize) {
        let now = Instant::now();
        client.inflight_bytes = frames.saturating_mul(bytes_per_frame);
        client.inflight_frames = (0..frames)
            .map(|_| InFlightFrame {
                sent_at: now,
                bytes: bytes_per_frame,
                paced: true,
            })
            .collect();
    }

    fn sample_frame(text: &str) -> FrameState {
        let mut frame = FrameState::new(2, 8);
        frame.write_text(0, 0, text, blit_remote::CellStyle::default());
        frame
    }

    #[test]
    fn unset_view_size_accepts_zero_pair_only() {
        assert!(is_unset_view_size(0, 0));
        assert!(!is_unset_view_size(0, 80));
        assert!(!is_unset_view_size(u16::MAX, u16::MAX));
    }

    #[test]
    fn unsubscribe_client_from_clears_view_size() {
        let mut client = test_client();
        client.subscriptions.insert(7);
        client.view_sizes.insert(7, (24, 80));
        assert!(unsubscribe_client_from(&mut client, 7));
        assert!(!client.subscriptions.contains(&7));
        assert!(!client.view_sizes.contains_key(&7));
    }

    #[test]
    fn mediated_size_uses_per_pty_view_sizes_without_lead() {
        let mut session = Session::new();
        let mut c1 = test_client();
        let mut c2 = test_client();
        c1.view_sizes.insert(7, (30, 120));
        c2.view_sizes.insert(7, (24, 100));
        session.clients.insert(1, c1);
        session.clients.insert(2, c2);
        assert_eq!(session.mediated_size_for_pty(7), Some((24, 100)));
    }

    #[test]
    fn mediated_surface_size_picks_min_dimensions_max_scale() {
        let mut session = Session::new();
        let mut c1 = test_client();
        let mut c2 = test_client();
        c1.surface_view_sizes.insert(1, (1920, 1080, 240)); // 2×
        c2.surface_view_sizes.insert(1, (1280, 720, 120)); // 1×
        session.clients.insert(1, c1);
        session.clients.insert(2, c2);
        assert_eq!(
            session.mediated_size_for_surface(1, None),
            Some((1280, 720, 240))
        );
    }

    #[test]
    fn mediated_surface_size_none_when_no_clients() {
        let session = Session::new();
        assert_eq!(session.mediated_size_for_surface(1, None), None);
    }

    #[test]
    fn mediated_surface_size_single_client() {
        let mut session = Session::new();
        let mut c1 = test_client();
        c1.surface_view_sizes.insert(3, (800, 600, 120));
        session.clients.insert(1, c1);
        assert_eq!(
            session.mediated_size_for_surface(3, None),
            Some((800, 600, 120))
        );
    }

    #[test]
    fn mediated_surface_size_ignores_other_surfaces() {
        let mut session = Session::new();
        let mut c1 = test_client();
        c1.surface_view_sizes.insert(1, (1920, 1080, 240));
        c1.surface_view_sizes.insert(2, (640, 480, 120));
        session.clients.insert(1, c1);
        assert_eq!(
            session.mediated_size_for_surface(1, None),
            Some((1920, 1080, 240))
        );
        assert_eq!(
            session.mediated_size_for_surface(2, None),
            Some((640, 480, 120))
        );
        assert_eq!(session.mediated_size_for_surface(3, None), None);
    }

    #[test]
    fn mediated_surface_size_clamped_to_encoder_max() {
        let mut session = Session::new();
        let mut c1 = test_client();
        c1.surface_view_sizes.insert(1, (5000, 3000, 240));
        session.clients.insert(1, c1);
        assert_eq!(
            session.mediated_size_for_surface(1, None),
            Some((5000, 3000, 240))
        );
        assert_eq!(
            session.mediated_size_for_surface(1, Some((3840, 2160))),
            Some((3840, 2160, 240))
        );
    }

    #[test]
    fn due_preview_reserves_the_last_lead_slot() {
        let mut client = test_client();
        client.lead = Some(1);
        client.subscriptions.insert(1);
        client.subscriptions.insert(2);

        let target_frames = target_frame_window(&client);
        let lead_limit = target_frames.saturating_sub(1).max(1);
        fill_inflight(&mut client, lead_limit, 512);

        assert!(window_open(&client));
        assert!(lead_window_open(&client, false));
        assert!(!lead_window_open(&client, true));
        assert!(can_send_preview(&client, 2, Instant::now()));
    }

    #[test]
    fn entering_scrollback_uses_current_visible_frame_as_baseline() {
        let mut client = test_client();
        let live = sample_frame("live");
        client.lead = Some(7);
        client.subscriptions.insert(7);
        client.last_sent.insert(7, live.clone());

        assert!(update_client_scroll_state(&mut client, 7, 12));
        assert_eq!(client.scroll_offsets.get(&7), Some(&12));
        assert_eq!(client.scroll_caches.get(&7), Some(&live));
    }

    #[test]
    fn leaving_scrollback_seeds_live_diff_from_scrollback_view() {
        let mut client = test_client();
        let history = sample_frame("hist");
        client.lead = Some(7);
        client.subscriptions.insert(7);
        client.scroll_offsets.insert(7, 12);
        client.scroll_caches.insert(7, history.clone());

        assert!(update_client_scroll_state(&mut client, 7, 0));
        assert_eq!(client.scroll_offsets.get(&7), None);
        assert_eq!(client.last_sent.get(&7), Some(&history));
        assert_eq!(client.scroll_caches.get(&7), None);
    }

    #[tokio::test]
    async fn request_surface_capture_returns_pixels_from_compositor() {
        let (command_tx, command_rx) = std::sync::mpsc::channel();
        std::thread::Builder::new()
            .name("test-capture-reply".into())
            .spawn(move || {
                let CompositorCommand::Capture {
                    surface_id,
                    scale_120: _,
                    reply,
                } = command_rx.recv().unwrap()
                else {
                    panic!("expected capture command");
                };
                assert_eq!(surface_id, 7);
                let _ = reply.send(Some((2, 3, vec![1, 2, 3, 4])));
            })
            .unwrap();

        let result =
            request_surface_capture_with_timeout(command_tx, 7, 0, Duration::from_millis(50)).await;

        assert_eq!(result, Some((2, 3, vec![1, 2, 3, 4])));
    }

    #[tokio::test]
    async fn request_surface_capture_returns_none_when_compositor_disconnects() {
        let (command_tx, command_rx) = std::sync::mpsc::channel();
        std::thread::Builder::new()
            .name("test-capture-drop".into())
            .spawn(move || {
                let _ = command_rx.recv().unwrap();
            })
            .unwrap();

        let result =
            request_surface_capture_with_timeout(command_tx, 7, 0, Duration::from_millis(50)).await;

        assert_eq!(result, None);
    }

    // ── frame_window ──

    #[test]
    fn frame_window_minimum_is_two() {
        assert!(frame_window(0.0, 60.0) >= 2);
    }

    #[test]
    fn frame_window_scales_with_rtt() {
        let low = frame_window(10.0, 60.0);
        let high = frame_window(200.0, 60.0);
        assert!(high > low, "higher RTT should need more frames in flight");
    }

    #[test]
    fn frame_window_scales_with_fps() {
        let slow = frame_window(100.0, 10.0);
        let fast = frame_window(100.0, 120.0);
        assert!(fast > slow, "higher fps should need more frames in flight");
    }

    #[test]
    fn frame_window_zero_rtt() {
        assert!(frame_window(0.0, 120.0) >= 2);
    }

    // ── path_rtt_ms ──

    #[test]
    fn path_rtt_ms_uses_min_when_positive() {
        let mut client = test_client();
        client.rtt_ms = 100.0;
        client.min_rtt_ms = 30.0;
        assert_eq!(path_rtt_ms(&client), 30.0);
    }

    #[test]
    fn path_rtt_ms_falls_back_to_rtt_when_min_zero() {
        let mut client = test_client();
        client.rtt_ms = 80.0;
        client.min_rtt_ms = 0.0;
        assert_eq!(path_rtt_ms(&client), 80.0);
    }

    // ── ewma_with_direction ──

    #[test]
    fn ewma_rising_uses_rise_alpha() {
        let result = ewma_with_direction(100.0, 200.0, 0.5, 0.1);
        // rise: 100 * 0.5 + 200 * 0.5 = 150
        assert!((result - 150.0).abs() < 0.01);
    }

    #[test]
    fn ewma_falling_uses_fall_alpha() {
        let result = ewma_with_direction(200.0, 100.0, 0.5, 0.1);
        // fall: 200 * 0.9 + 100 * 0.1 = 190
        assert!((result - 190.0).abs() < 0.01);
    }

    #[test]
    fn ewma_same_value_unchanged() {
        let result = ewma_with_direction(50.0, 50.0, 0.5, 0.5);
        assert!((result - 50.0).abs() < 0.01);
    }

    // ── advance_deadline ──

    #[test]
    fn advance_deadline_steps_forward() {
        let now = Instant::now();
        let mut deadline = now;
        let interval = Duration::from_millis(16);
        advance_deadline(&mut deadline, now, interval);
        assert!(deadline > now);
        assert!(deadline <= now + interval + Duration::from_micros(100));
    }

    #[test]
    fn advance_deadline_resets_when_far_behind() {
        let now = Instant::now();
        // deadline is way in the past (more than 2 intervals ago)
        let mut deadline = now - Duration::from_secs(10);
        let interval = Duration::from_millis(16);
        advance_deadline(&mut deadline, now, interval);
        // Should snap to now + interval since scheduled + interval < now
        assert!(deadline >= now);
    }

    #[test]
    fn should_snapshot_pty_requires_dirty_and_needful() {
        assert!(should_snapshot_pty(true, true, false));
        assert!(!should_snapshot_pty(false, true, false));
        assert!(!should_snapshot_pty(true, false, false));
    }

    #[test]
    fn should_snapshot_pty_defers_synced_output() {
        assert!(!should_snapshot_pty(true, true, true));
        assert!(should_snapshot_pty(true, true, false));
    }

    #[test]
    fn enqueue_ready_frame_refuses_new_frames_when_capped() {
        let mut queue = VecDeque::new();
        for cols in 1..=(READY_FRAME_QUEUE_CAP as u16) {
            assert!(enqueue_ready_frame(&mut queue, FrameState::new(1, cols)));
        }
        assert!(!enqueue_ready_frame(
            &mut queue,
            FrameState::new(1, READY_FRAME_QUEUE_CAP as u16 + 1),
        ));
        assert_eq!(queue.len(), READY_FRAME_QUEUE_CAP);
        assert_eq!(queue.front().map(FrameState::cols), Some(1));
        assert_eq!(
            queue.back().map(FrameState::cols),
            Some(READY_FRAME_QUEUE_CAP as u16),
        );
    }

    #[test]
    fn find_sync_output_end_returns_end_of_first_close_sequence() {
        let bytes = b"abc\x1b[?2026lrest\x1b[?2026l";
        assert_eq!(find_sync_output_end(&[], bytes), Some(11));
    }

    #[test]
    fn find_sync_output_end_returns_none_without_close_sequence() {
        assert_eq!(find_sync_output_end(&[], b"\x1b[?2026hpartial"), None);
    }

    #[test]
    fn find_sync_output_end_detects_boundary_split_across_reads() {
        assert_eq!(find_sync_output_end(b"abc\x1b[?20", b"26lrest"), Some(3));
    }

    #[test]
    fn update_sync_scan_tail_keeps_recent_suffix_only() {
        let mut tail = Vec::new();
        update_sync_scan_tail(&mut tail, b"123456789");
        assert_eq!(tail, b"3456789");
    }

    // ── window_saturated ──

    #[test]
    fn window_saturated_at_90_percent_frames() {
        let client = test_client();
        let target = target_frame_window(&client);
        let frames_90 = (target * 9).div_ceil(10); // ceil(target * 0.9)
        assert!(window_saturated(&client, frames_90, 0));
    }

    #[test]
    fn window_saturated_not_at_low_usage() {
        let client = test_client();
        assert!(!window_saturated(&client, 1, 0));
    }

    #[test]
    fn window_saturated_at_90_percent_bytes() {
        let client = test_client();
        let target_bytes = target_byte_window(&client);
        let bytes_90 = (target_bytes * 9).div_ceil(10);
        assert!(window_saturated(&client, 0, bytes_90));
    }

    // ── outbox_queued_frames / outbox_backpressured ──

    #[test]
    fn outbox_queued_frames_zero_when_empty() {
        let client = test_client();
        assert_eq!(outbox_queued_frames(&client), 0);
    }

    #[test]
    fn outbox_backpressured_when_queue_full() {
        let (client, _rx) = test_client_with_capacity(OUTBOX_CAPACITY);
        // Fill the channel to trigger backpressure
        for _ in 0..OUTBOX_SOFT_QUEUE_LIMIT_FRAMES {
            let _ = try_send_outbox(&client, vec![0u8]);
        }
        assert!(outbox_backpressured(&client));
    }

    #[test]
    fn outbox_not_backpressured_by_small_frames_under_byte_budget() {
        let (client, _rx) = test_client_with_capacity(OUTBOX_CAPACITY);
        for _ in 0..(OUTBOX_SOFT_QUEUE_LIMIT_FRAMES - 1) {
            let _ = try_send_outbox(&client, vec![0u8; 512]);
        }
        assert!(!outbox_backpressured(&client));
    }

    #[test]
    fn outbox_backpressured_by_large_queued_bytes() {
        let (client, _rx) = test_client_with_capacity(OUTBOX_CAPACITY);
        let _ = try_send_outbox(&client, vec![0u8; OUTBOX_SOFT_QUEUE_LIMIT_BYTES]);
        assert!(outbox_backpressured(&client));
    }

    #[test]
    fn outbox_not_backpressured_when_empty() {
        let client = test_client();
        assert!(!outbox_backpressured(&client));
    }

    // ── browser_pacing_fps baseline ──

    #[test]
    fn browser_pacing_fps_matches_display_fps_when_browser_ready() {
        let mut client = test_client();
        client.rtt_ms = 1.0;
        client.min_rtt_ms = 1.0;
        client.browser_backlog_frames = 0;
        client.browser_ack_ahead_frames = 0;
        client.browser_apply_ms = 0.0;
        client.goodput_bps = 1_000_000.0;
        client.delivery_bps = 1_000_000.0;
        client.display_fps = 144.0;
        assert!((browser_pacing_fps(&client) - 144.0).abs() < 0.01);
    }

    #[test]
    fn browser_pacing_fps_drops_below_display_fps_when_backlogged() {
        let mut client = test_client();
        client.browser_backlog_frames = 20;
        let fps = browser_pacing_fps(&client);
        assert!(fps >= 1.0);
        assert!(fps < client.display_fps);
    }

    // ── effective_rtt_ms ──

    #[test]
    fn effective_rtt_ms_equals_path_when_queue_is_empty() {
        let mut client = test_client();
        client.rtt_ms = 1.0;
        client.min_rtt_ms = 1.0;
        client.browser_backlog_frames = 0;
        client.browser_ack_ahead_frames = 0;
        client.browser_apply_ms = 0.0;
        client.goodput_bps = 1_000_000.0;
        client.delivery_bps = 1_000_000.0;
        assert!((effective_rtt_ms(&client) - 1.0).abs() < 0.01);
    }

    #[test]
    fn effective_rtt_ms_at_least_path_rtt() {
        let client = test_client();
        assert!(effective_rtt_ms(&client) >= path_rtt_ms(&client));
    }

    // ── target_frame_window ──

    #[test]
    fn target_frame_window_at_least_two() {
        let client = test_client();
        assert!(target_frame_window(&client) >= 2);
    }

    #[test]
    fn target_frame_window_grows_with_probe() {
        let mut client = test_client();
        let base = target_frame_window(&client);
        client.probe_frames = 10.0;
        let probed = target_frame_window(&client);
        assert!(probed > base, "probe_frames should grow the window");
    }

    // ── bandwidth_floor_bps ──

    #[test]
    fn bandwidth_floor_bps_at_least_16k() {
        let mut client = test_client();
        client.goodput_bps = 0.0;
        client.delivery_bps = 0.0;
        assert_eq!(bandwidth_floor_bps(&client), 0.0);
    }

    #[test]
    fn bandwidth_floor_bps_scales_with_goodput() {
        let mut client = test_client();
        client.goodput_bps = 1_000_000.0;
        client.delivery_bps = 1_000_000.0;
        let floor = bandwidth_floor_bps(&client);
        assert!(floor > 0.0);
    }

    #[test]
    fn browser_ready_delivery_floor_can_drive_large_frames_to_display_fps() {
        let mut client = test_client();
        client.display_fps = 60.0;
        client.browser_backlog_frames = 0;
        client.browser_ack_ahead_frames = 0;
        client.browser_apply_ms = 0.2;
        client.goodput_bps = 3_000_000.0;
        client.delivery_bps = 9_500_000.0;
        client.last_goodput_sample_bps = 3_000_000.0;
        client.avg_paced_frame_bytes = 150_000.0;
        client.avg_preview_frame_bytes = 1_024.0;
        client.avg_frame_bytes = 150_000.0;

        assert!(
            (pacing_fps(&client) - client.display_fps).abs() < 0.01,
            "browser-ready delivery floor should let large frames reach display_fps on a fast path",
        );
    }

    // ── pacing_fps ──

    #[test]
    fn pacing_fps_zero_when_no_bandwidth() {
        let mut client = test_client();
        client.goodput_bps = 0.0;
        client.delivery_bps = 0.0;
        client.last_goodput_sample_bps = 0.0;
        assert!(
            pacing_fps(&client) == 0.0,
            "pacing_fps should be 0 with zero bandwidth"
        );
    }

    #[test]
    fn pacing_fps_reaches_display_fps_when_not_bandwidth_limited() {
        let mut client = test_client();
        client.rtt_ms = 1.0;
        client.min_rtt_ms = 1.0;
        client.browser_backlog_frames = 0;
        client.browser_ack_ahead_frames = 0;
        client.browser_apply_ms = 0.0;
        client.goodput_bps = 1_000_000.0;
        client.delivery_bps = 1_000_000.0;
        client.display_fps = 60.0;
        assert!((pacing_fps(&client) - 60.0).abs() < 0.01);
    }

    // ── throughput_limited ──

    #[test]
    fn throughput_limited_when_low_bandwidth() {
        let mut client = test_client();
        client.goodput_bps = 1_000.0;
        client.delivery_bps = 1_000.0;
        client.last_goodput_sample_bps = 0.0;
        assert!(throughput_limited(&client));
    }

    #[test]
    fn throughput_not_limited_with_high_bandwidth() {
        let mut client = test_client();
        client.goodput_bps = 100_000_000.0;
        client.delivery_bps = 100_000_000.0;
        assert!(!throughput_limited(&client));
    }

    // ── browser_pacing_fps ──

    #[test]
    fn browser_pacing_fps_at_least_one() {
        let client = test_client();
        assert!(browser_pacing_fps(&client) >= 1.0);
    }

    #[test]
    fn browser_pacing_fps_reduced_by_high_backlog() {
        let mut client = test_client();
        let normal = browser_pacing_fps(&client);
        client.browser_backlog_frames = 20;
        let backlogged = browser_pacing_fps(&client);
        assert!(backlogged < normal, "high backlog should reduce pacing fps");
    }

    #[test]
    fn browser_pacing_fps_reduced_by_high_ack_ahead() {
        let mut client = test_client();
        let normal = browser_pacing_fps(&client);
        client.browser_ack_ahead_frames = 10;
        let ahead = browser_pacing_fps(&client);
        assert!(ahead < normal, "high ack_ahead should reduce pacing fps");
    }

    // ── browser_backlog_blocked ──

    #[test]
    fn browser_backlog_blocked_over_threshold() {
        let mut client = test_client();
        client.browser_backlog_frames = 9;
        assert!(browser_backlog_blocked(&client));
    }

    #[test]
    fn browser_backlog_not_blocked_under_threshold() {
        let mut client = test_client();
        client.browser_backlog_frames = 8;
        assert!(!browser_backlog_blocked(&client));
    }

    // ── byte_budget_for ──

    #[test]
    fn byte_budget_for_at_least_one_frame() {
        let client = test_client();
        let budget = byte_budget_for(&client, 10.0);
        assert!(budget >= client.avg_frame_bytes.max(256.0) as usize);
    }

    #[test]
    fn byte_budget_for_grows_with_time() {
        let client = test_client();
        let short = byte_budget_for(&client, 10.0);
        let long = byte_budget_for(&client, 1000.0);
        assert!(long >= short);
    }

    // ── target_byte_window ──

    #[test]
    fn target_byte_window_positive() {
        let client = test_client();
        assert!(target_byte_window(&client) > 0);
    }

    #[test]
    fn target_byte_window_covers_frame_window() {
        let client = test_client();
        let byte_win = target_byte_window(&client);
        let frame_win = target_frame_window(&client);
        let min_bytes =
            (client.avg_paced_frame_bytes.max(256.0) * frame_win.max(2) as f32).ceil() as usize;
        assert!(
            byte_win >= min_bytes,
            "byte window should cover at least frame_window worth of paced frames"
        );
    }

    // ── send_interval ──

    #[test]
    fn send_interval_matches_browser_pacing() {
        let client = test_client();
        let interval = send_interval(&client);
        let expected = Duration::from_secs_f64(1.0 / browser_pacing_fps(&client) as f64);
        let diff = interval.abs_diff(expected);
        assert!(diff < Duration::from_micros(10));
    }

    // ── preview_fps ──

    #[test]
    fn preview_fps_at_least_one() {
        let client = test_client();
        assert!(preview_fps(&client) >= 1.0);
    }

    // ── window_open ──

    #[test]
    fn window_open_initially() {
        let client = test_client();
        assert!(window_open(&client));
    }

    #[test]
    fn window_open_false_when_browser_blocked() {
        let mut client = test_client();
        client.browser_backlog_frames = 20;
        assert!(!window_open(&client));
    }

    #[test]
    fn window_open_false_when_inflight_full() {
        let mut client = test_client();
        let target = target_frame_window(&client);
        fill_inflight(&mut client, target + 10, 1024);
        assert!(!window_open(&client));
    }

    // ── lead_window_open ──

    #[test]
    fn lead_window_open_no_reserve_same_as_window_open() {
        let client = test_client();
        assert_eq!(lead_window_open(&client, false), window_open(&client));
    }

    #[test]
    fn lead_window_open_reserves_preview_slot() {
        let mut client = test_client();
        client.lead = Some(1);
        client.subscriptions.insert(1);
        let target = target_frame_window(&client);
        // Fill to just under target minus reserve
        fill_inflight(&mut client, target.saturating_sub(1), 512);
        // Without reserve: may still be open
        // With reserve: should be closed
        assert!(!lead_window_open(&client, true));
    }

    // ── can_send_frame ──

    #[test]
    fn can_send_frame_when_window_open_and_time_due() {
        let mut client = test_client();
        client.next_send_at = Instant::now() - Duration::from_millis(100);
        assert!(can_send_frame(&client, Instant::now(), false));
    }

    #[test]
    fn can_send_frame_false_when_not_due() {
        let mut client = test_client();
        client.next_send_at = Instant::now() + Duration::from_secs(10);
        assert!(!can_send_frame(&client, Instant::now(), false));
    }

    #[test]
    fn can_send_frame_false_when_window_closed() {
        let mut client = test_client();
        client.browser_backlog_frames = 20; // triggers browser_backlog_blocked
        client.next_send_at = Instant::now() - Duration::from_millis(100);
        assert!(!can_send_frame(&client, Instant::now(), false));
    }

    // ── record_send / record_ack state transitions ──

    #[test]
    fn record_send_increases_inflight() {
        let mut client = test_client();
        let now = Instant::now();
        assert_eq!(client.inflight_bytes, 0);
        assert_eq!(client.inflight_frames.len(), 0);

        record_send(&mut client, 1000, now, true);
        assert_eq!(client.inflight_bytes, 1000);
        assert_eq!(client.inflight_frames.len(), 1);

        record_send(&mut client, 500, now, false);
        assert_eq!(client.inflight_bytes, 1500);
        assert_eq!(client.inflight_frames.len(), 2);
    }

    #[test]
    fn record_send_paced_advances_deadline() {
        let mut client = test_client();
        let now = Instant::now();
        client.next_send_at = now;
        record_send(&mut client, 1000, now, true);
        assert!(client.next_send_at > now);
    }

    #[test]
    fn record_send_unpaced_does_not_advance_deadline() {
        let mut client = test_client();
        let now = Instant::now();
        let before = client.next_send_at;
        record_send(&mut client, 1000, now, false);
        assert_eq!(client.next_send_at, before);
    }

    #[test]
    fn record_ack_decreases_inflight() {
        let mut client = test_client();
        let now = Instant::now();
        record_send(&mut client, 1000, now, true);
        record_send(&mut client, 500, now, true);
        assert_eq!(client.inflight_frames.len(), 2);

        record_ack(&mut client);
        assert_eq!(client.inflight_frames.len(), 1);
        assert_eq!(client.inflight_bytes, 500);
    }

    #[test]
    fn record_ack_on_empty_clears_bytes() {
        let mut client = test_client();
        client.inflight_bytes = 999; // stale state
        record_ack(&mut client);
        assert_eq!(client.inflight_bytes, 0);
    }

    #[test]
    fn record_ack_updates_rtt_estimate() {
        let mut client = test_client();
        let now = Instant::now();
        client.inflight_frames.push_back(InFlightFrame {
            sent_at: now - Duration::from_millis(20),
            bytes: 512,
            paced: true,
        });
        client.inflight_bytes = 512;
        let old_rtt = client.rtt_ms;
        record_ack(&mut client);
        // RTT should have been updated (moved toward ~20ms from the default 50ms)
        assert!(
            (client.rtt_ms - old_rtt).abs() > 0.01,
            "rtt_ms should be updated after ack"
        );
    }

    #[test]
    fn record_ack_paced_updates_avg_paced_frame_bytes() {
        let mut client = test_client();
        let now = Instant::now();
        client.inflight_frames.push_back(InFlightFrame {
            sent_at: now - Duration::from_millis(10),
            bytes: 4096,
            paced: true,
        });
        client.inflight_bytes = 4096;
        let old_avg = client.avg_paced_frame_bytes;
        record_ack(&mut client);
        // Should move toward 4096 from 1024
        assert!(client.avg_paced_frame_bytes > old_avg);
    }

    #[test]
    fn record_ack_unpaced_updates_avg_preview_frame_bytes() {
        let mut client = test_client();
        let now = Instant::now();
        client.inflight_frames.push_back(InFlightFrame {
            sent_at: now - Duration::from_millis(10),
            bytes: 8192,
            paced: false,
        });
        client.inflight_bytes = 8192;
        let old_avg = client.avg_preview_frame_bytes;
        record_ack(&mut client);
        assert!(client.avg_preview_frame_bytes > old_avg);
    }

    // ── Session::pty_list_msg format ──

    #[test]
    fn pty_list_msg_empty_session() {
        let sess = Session::new();
        let msg = sess.pty_list_msg();
        assert_eq!(msg[0], S2C_LIST);
        assert_eq!(u16::from_le_bytes([msg[1], msg[2]]), 0);
        assert_eq!(msg.len(), 3);
    }

    #[test]
    fn pty_list_msg_includes_tags() {
        let _sess = Session::new();
        // Insert minimal Pty entries. We can't call spawn_pty, so build
        // a mock-like Pty with a stub driver. Instead, directly insert
        // into the HashMap using an unsafe-free approach: just build the
        // wire message by hand and verify against a known layout.
        //
        // The wire format is: [S2C_LIST] [count:u16le] [id:u16le tag_len:u16le tag_bytes]...
        //
        // Since we can't easily construct a Pty without forking, verify
        // the format by constructing the expected bytes and comparing.
        let tag1 = "shell";
        let tag2 = "build";

        // Expected wire for ptys {1 => "shell", 3 => "build"} sorted by id:
        let mut expected = vec![S2C_LIST];
        expected.extend_from_slice(&2u16.to_le_bytes());
        // id=1
        expected.extend_from_slice(&1u16.to_le_bytes());
        expected.extend_from_slice(&(tag1.len() as u16).to_le_bytes());
        expected.extend_from_slice(tag1.as_bytes());
        // id=3
        expected.extend_from_slice(&3u16.to_le_bytes());
        expected.extend_from_slice(&(tag2.len() as u16).to_le_bytes());
        expected.extend_from_slice(tag2.as_bytes());

        // Verify our expected format starts with S2C_LIST and has correct count
        assert_eq!(expected[0], S2C_LIST);
        assert_eq!(u16::from_le_bytes([expected[1], expected[2]]), 2);
        // Verify tags are embedded
        let msg_str = String::from_utf8_lossy(&expected);
        assert!(msg_str.contains("shell"));
        assert!(msg_str.contains("build"));
    }

    // ── can_send_preview / record_preview_send ──

    #[test]
    fn can_send_preview_true_when_due() {
        let mut client = test_client();
        let now = Instant::now();
        client
            .preview_next_send_at
            .insert(5, now - Duration::from_millis(100));
        assert!(can_send_preview(&client, 5, now));
    }

    #[test]
    fn can_send_preview_false_when_not_due() {
        let mut client = test_client();
        let now = Instant::now();
        client
            .preview_next_send_at
            .insert(5, now + Duration::from_secs(10));
        assert!(!can_send_preview(&client, 5, now));
    }

    #[test]
    fn can_send_preview_false_when_window_closed() {
        let mut client = test_client();
        client.browser_backlog_frames = 20;
        let now = Instant::now();
        assert!(!can_send_preview(&client, 5, now));
    }

    #[test]
    fn can_send_preview_true_for_unseen_pid() {
        let client = test_client();
        let now = Instant::now();
        // No entry in preview_next_send_at means deadline defaults to now
        assert!(can_send_preview(&client, 99, now));
    }

    #[test]
    fn record_preview_send_sets_future_deadline() {
        let mut client = test_client();
        let now = Instant::now();
        record_preview_send(&mut client, 5, now);
        let deadline = client.preview_next_send_at.get(&5).unwrap();
        assert!(*deadline > now);
    }

    #[test]
    fn record_preview_send_successive_calls_advance() {
        let mut client = test_client();
        let now = Instant::now();
        record_preview_send(&mut client, 5, now);
        let first = *client.preview_next_send_at.get(&5).unwrap();
        record_preview_send(&mut client, 5, first);
        let second = *client.preview_next_send_at.get(&5).unwrap();
        assert!(second > first, "successive sends should advance deadline");
    }

    // ── congestion control end-to-end properties ──
    //
    // These tests encode the two goals of the congestion controller:
    //   1. Browser-ready, well-provisioned path → full display FPS, minimal added latency
    //   2. Bottleneck                           → lowest sustainable FPS, fast recovery when pipe clears
    //
    // Some tests assert desired future behaviour and currently FAIL due to
    // known issues (min_rtt contamination, lead_floor dominating byte window).
    // They are marked with a comment so they are easy to find when fixing.

    /// Return a client in ideal low-latency, high-bandwidth conditions:
    /// browser ready, abundant bandwidth, and tiny RTT. The normal pacing path
    /// should still reach display_fps.
    fn browser_ready_high_bandwidth_client() -> ClientState {
        let mut c = test_client();
        c.display_fps = 120.0;
        c.rtt_ms = 1.0;
        c.min_rtt_ms = 1.0;
        c.goodput_bps = 50_000_000.0;
        c.delivery_bps = 50_000_000.0;
        c.last_goodput_sample_bps = 50_000_000.0;
        c.avg_paced_frame_bytes = 30_000.0;
        c.avg_preview_frame_bytes = 1_024.0;
        c.avg_frame_bytes = 30_000.0;
        c.browser_apply_ms = 0.3;
        c
    }

    /// Return a client that has converged to a clearly congested state:
    /// ~10× min_rtt inflation, low goodput.
    fn congested_client() -> ClientState {
        let mut c = test_client();
        c.display_fps = 120.0;
        c.rtt_ms = 500.0;
        c.min_rtt_ms = 40.0;
        c.goodput_bps = 200_000.0;
        c.delivery_bps = 150_000.0;
        c.last_goodput_sample_bps = 200_000.0;
        c.avg_paced_frame_bytes = 50_000.0;
        c.avg_preview_frame_bytes = 1_024.0;
        c.avg_frame_bytes = 50_000.0;
        c.goodput_jitter_bps = 50_000.0;
        c.max_goodput_jitter_bps = 200_000.0;
        c.browser_apply_ms = 1.0;
        c
    }

    /// Simulate one ACK: insert a frame with the given RTT into inflight and
    /// call record_ack.  Forces a goodput-window sample each call so that
    /// goodput estimates respond within a few calls.
    fn sim_ack(client: &mut ClientState, bytes: usize, rtt_ms: f32) {
        let sent_at = Instant::now() - Duration::from_millis(rtt_ms as u64);
        client.inflight_bytes += bytes;
        client.inflight_frames.push_back(InFlightFrame {
            sent_at,
            bytes,
            paced: true,
        });
        // Age the goodput window so record_ack always emits a sample.
        client.goodput_window_start = Instant::now() - Duration::from_millis(25);
        record_ack(client);
    }

    fn sim_acks(client: &mut ClientState, n: usize, bytes: usize, rtt_ms: f32) {
        for _ in 0..n {
            sim_ack(client, bytes, rtt_ms);
        }
    }

    // ── property: full FPS on a browser-ready path ──

    #[test]
    fn browser_ready_high_bandwidth_client_uses_full_display_fps() {
        let client = browser_ready_high_bandwidth_client();
        assert!(
            (pacing_fps(&client) - client.display_fps).abs() < 0.01,
            "pacing_fps {} should equal display_fps {} when browser is ready and bandwidth is abundant",
            pacing_fps(&client),
            client.display_fps,
        );
    }

    #[test]
    fn browser_ready_high_bandwidth_client_send_interval_within_one_frame() {
        let client = browser_ready_high_bandwidth_client();
        let interval_ms = send_interval(&client).as_secs_f32() * 1000.0;
        let frame_ms = 1000.0 / client.display_fps;
        assert!(
            interval_ms <= frame_ms + 0.1,
            "send_interval {interval_ms:.2}ms exceeds one frame ({frame_ms:.2}ms) when browser is ready"
        );
    }

    // ── property: degraded FPS when bottlenecked ──

    #[test]
    fn congested_pipe_reduces_pacing_fps_substantially() {
        let client = congested_client();
        let fps = pacing_fps(&client);
        assert!(
            fps < client.display_fps * 0.5,
            "pacing_fps {fps:.0} should be well below display_fps {} when congested",
            client.display_fps,
        );
    }

    #[test]
    fn congested_pipe_is_throughput_limited() {
        let client = congested_client();
        assert!(
            throughput_limited(&client),
            "congested client must be recognised as throughput-limited"
        );
    }

    // ── property: byte window should stay near BDP ──
    //
    // KNOWN FAILING: lead_floor in target_byte_window overrides the BDP
    // budget when avg_paced_frame_bytes is large.  Fix: cap lead_floor.

    #[test]
    fn byte_window_bounded_near_bdp_when_congested() {
        let client = congested_client();
        // BDP at the unloaded path RTT.
        let bdp = client.goodput_bps * (path_rtt_ms(&client) / 1_000.0);
        let window = target_byte_window(&client);
        assert!(
            window < bdp as usize * 8,
            "byte window {window}B is {:.1}× BDP ({bdp:.0}B); \
             expected ≤ 8× — lead_floor may be dominating",
            window as f32 / bdp.max(1.0),
        );
    }

    // ── property: min_rtt must not drift upward under congestion ──
    //
    // KNOWN FAILING: the `min_rtt_ms * 0.999 + rtt_ms * 0.001` update
    // bleeds queued RTT into min_rtt.

    #[test]
    fn min_rtt_not_contaminated_by_congested_rtts() {
        let mut client = test_client();
        client.display_fps = 120.0;
        client.rtt_ms = 40.0;
        client.min_rtt_ms = 40.0;
        client.goodput_bps = 2_000_000.0;
        client.delivery_bps = 2_000_000.0;
        client.avg_paced_frame_bytes = 30_000.0;
        client.avg_preview_frame_bytes = 1_024.0;
        let original_min = client.min_rtt_ms;

        // 200 ACKs arriving with 500ms RTT (severe congestion).
        sim_acks(&mut client, 200, 30_000, 500.0);

        assert!(
            client.min_rtt_ms < original_min * 2.0,
            "min_rtt drifted from {original_min}ms to {:.1}ms after 200 congested ACKs",
            client.min_rtt_ms,
        );
    }

    // ── property: fast recovery when congestion clears ──

    #[test]
    fn delivery_bps_rises_quickly_when_congestion_clears() {
        let mut client = congested_client();
        let before = client.delivery_bps;

        // 10 ACKs at low latency / high throughput.
        sim_acks(&mut client, 10, 30_000, 40.0);

        assert!(
            client.delivery_bps > before * 2.0,
            "delivery_bps {:.0} should more than double from {before:.0} after 10 fast ACKs",
            client.delivery_bps,
        );
    }

    #[test]
    fn pacing_fps_recovers_after_congestion_clears() {
        let mut client = congested_client();

        // Use window-saturated rounds: fill the window with frames, age the
        // goodput window once, then ACK all.  The first ACK each round emits
        // a sample; the remaining target-1 ACKs carry over into the next
        // window, so sample throughput grows as target grows — mimicking a
        // real link where the sender keeps the pipe full across one RTT.
        for _ in 0..40 {
            let target = target_frame_window(&client).max(2);
            for _ in 0..target {
                let sent_at = Instant::now() - Duration::from_millis(40);
                client.inflight_bytes += 30_000;
                client.inflight_frames.push_back(InFlightFrame {
                    sent_at,
                    bytes: 30_000,
                    paced: true,
                });
            }
            client.goodput_window_start = Instant::now() - Duration::from_millis(25);
            for _ in 0..target {
                record_ack(&mut client);
            }
        }

        let fps = pacing_fps(&client);
        assert!(
            fps > client.display_fps * 0.7,
            "pacing_fps {fps:.0} didn't recover toward display_fps {} \
             after window-saturated rounds at low RTT",
            client.display_fps,
        );
    }

    #[test]
    fn rtt_estimate_drops_quickly_when_congestion_clears() {
        let mut client = test_client();
        client.rtt_ms = 500.0;
        client.min_rtt_ms = 40.0;
        client.goodput_bps = 2_000_000.0;
        client.avg_paced_frame_bytes = 30_000.0;
        client.avg_preview_frame_bytes = 1_024.0;

        // The asymmetric EWMA uses rise=0.125, fall=0.25, so rtt_ms drops
        // at fall_alpha=0.25 per sample toward the new low.
        sim_acks(&mut client, 10, 30_000, 40.0);

        assert!(
            client.rtt_ms < 300.0,
            "rtt_ms {:.0}ms did not fall fast enough after congestion cleared",
            client.rtt_ms,
        );
    }

    // ── property: probing ──

    #[test]
    fn probe_collapses_immediately_on_queue_delay() {
        let mut client = test_client();
        client.display_fps = 120.0;
        client.rtt_ms = 40.0;
        client.min_rtt_ms = 40.0;
        client.goodput_bps = 5_000_000.0;
        client.delivery_bps = 5_000_000.0;
        client.last_goodput_sample_bps = 5_000_000.0;
        client.avg_paced_frame_bytes = 10_000.0;
        client.avg_preview_frame_bytes = 1_024.0;
        client.probe_frames = 10.0;

        // ACKs arriving with high RTT signal queue buildup.
        sim_acks(&mut client, 5, 10_000, 600.0);

        assert!(
            client.probe_frames < 5.0,
            "probe_frames {:.1} should have collapsed on queue delay signal",
            client.probe_frames,
        );
    }

    #[test]
    fn probe_grows_when_window_saturated_with_clean_rtt() {
        let mut client = test_client();
        client.display_fps = 120.0;
        client.rtt_ms = 40.0;
        client.min_rtt_ms = 40.0;
        client.goodput_bps = 5_000_000.0;
        client.delivery_bps = 5_000_000.0;
        client.last_goodput_sample_bps = 5_000_000.0;
        client.avg_paced_frame_bytes = 10_000.0;
        client.avg_preview_frame_bytes = 1_024.0;
        client.goodput_jitter_bps = 0.0;
        client.max_goodput_jitter_bps = 0.0;
        client.probe_frames = 0.0;

        // Saturate inflight so window_saturated returns true during acks.
        let target = target_frame_window(&client);
        for _ in 0..target {
            let sent_at = Instant::now() - Duration::from_millis(40);
            client.inflight_bytes += 10_000;
            client.inflight_frames.push_back(InFlightFrame {
                sent_at,
                bytes: 10_000,
                paced: true,
            });
        }

        // Ack one frame with clean RTT.  One saturated ACK is sufficient to
        // verify the property: as probe_frames increments, target_frame_window
        // grows, so the remaining (target-1) frames would fall below the 90%
        // threshold and trigger gentle decay.  The property under test is that
        // *receiving an ACK while window-saturated* increments probe_frames —
        // not that it stays incremented across subsequent unsaturated ACKs.
        // Also: do NOT age the goodput window — that would emit a per-frame
        // sample far below goodput_bps, spiking jitter and collapsing probe.
        record_ack(&mut client);

        assert!(
            client.probe_frames > 0.0,
            "probe_frames should grow when window-saturated with clean RTT"
        );
    }

    // ── property: frame window larger on high-latency links ──

    #[test]
    fn frame_window_larger_on_high_latency_link() {
        let mut lo = test_client();
        lo.display_fps = 120.0;
        lo.rtt_ms = 10.0;
        lo.min_rtt_ms = 10.0;
        lo.goodput_bps = 5_000_000.0;
        lo.delivery_bps = 5_000_000.0;
        lo.avg_paced_frame_bytes = 10_000.0;
        lo.avg_preview_frame_bytes = 1_024.0;

        let mut hi = test_client();
        hi.display_fps = 120.0;
        hi.rtt_ms = 200.0;
        hi.min_rtt_ms = 200.0;
        hi.goodput_bps = 5_000_000.0;
        hi.delivery_bps = 5_000_000.0;
        hi.avg_paced_frame_bytes = 10_000.0;
        hi.avg_preview_frame_bytes = 1_024.0;

        let lo_win = target_frame_window(&lo);
        let hi_win = target_frame_window(&hi);
        assert!(
            hi_win > lo_win,
            "high-latency link ({hi_win}f) should need more frames in flight \
             than low-latency ({lo_win}f)"
        );
    }

    // ── property: small-frame byte window allows pipelining ──

    #[test]
    fn small_frame_byte_window_enables_pipelining() {
        // Tiny terminal frames (~1KB) with a stale congested RTT and low
        // goodput estimate (stop-and-wait artifact): byte window must be at
        // least target_frame_window × frame_bytes so the sender can pipeline
        // rather than stay stuck in stop-and-wait.
        let mut client = test_client();
        client.display_fps = 120.0;
        client.rtt_ms = 165.0;
        client.min_rtt_ms = 8.0;
        client.goodput_bps = 11_000.0; // stop-and-wait artifact
        client.delivery_bps = 6_800.0;
        client.last_goodput_sample_bps = 11_000.0;
        client.avg_paced_frame_bytes = 1_120.0;
        client.avg_preview_frame_bytes = 1_024.0;
        client.goodput_jitter_bps = 4_300.0;
        client.max_goodput_jitter_bps = 6_500.0;

        let window = target_byte_window(&client);
        let frames = target_frame_window(&client);
        let pipeline = frames * 1_120;

        assert!(
            window >= pipeline,
            "byte window {window}B should be >= pipeline ({frames}f × 1120B = {pipeline}B) \
             so small frames can pipeline across the RTT"
        );
    }

    #[test]
    fn large_frame_byte_window_bounded_by_one_frame_floor() {
        // With large frames (50KB), pipelining the full frame window (5×50KB=250KB)
        // would be many multiples of BDP.  Byte window should fall back to
        // the one-frame floor so the BDP budget governs.
        let mut client = test_client();
        client.display_fps = 120.0;
        client.rtt_ms = 165.0;
        client.min_rtt_ms = 8.0;
        client.goodput_bps = 11_000.0;
        client.delivery_bps = 6_800.0;
        client.last_goodput_sample_bps = 11_000.0;
        client.avg_paced_frame_bytes = 50_000.0; // large frame
        client.avg_preview_frame_bytes = 1_024.0;
        client.goodput_jitter_bps = 0.0;
        client.max_goodput_jitter_bps = 0.0;

        let window = target_byte_window(&client);
        let frames = target_frame_window(&client);
        let pipeline = frames.saturating_mul(50_000);

        assert!(
            window < pipeline,
            "byte window {window}B should be < full pipeline {pipeline}B \
             ({frames}f × 50KB) — large frames must use one-frame floor"
        );
        assert!(
            window >= 50_000,
            "byte window {window}B must be at least one frame (50KB)"
        );
    }

    // ── property: preview reservation applies uniformly ──

    #[test]
    fn preview_reservation_applies_even_on_low_latency_high_bandwidth_links() {
        let mut client = browser_ready_high_bandwidth_client();
        client.lead = Some(1);
        client.subscriptions.insert(1);
        let target = target_frame_window(&client);
        fill_inflight(&mut client, target.saturating_sub(1), 512);
        assert!(
            !lead_window_open(&client, true),
            "preview reservation should apply uniformly for lead clients"
        );
    }

    // ── property: blip recovery on healthy paths ──

    #[test]
    fn probe_recovers_on_healthy_path_after_blip() {
        let mut client = browser_ready_high_bandwidth_client();
        client.probe_frames = 8.0;

        // Blip: 3 ACKs with inflated RTT crush probes.
        sim_acks(&mut client, 3, 30_000, 200.0);
        let post_blip = client.probe_frames;
        assert!(
            post_blip < 4.0,
            "probe_frames {post_blip:.1} should have dropped after blip"
        );

        // Reset browser metrics to healthy (browser cleared backlog).
        client.browser_backlog_frames = 0;
        client.browser_ack_ahead_frames = 0;
        client.browser_apply_ms = 0.3;

        // Recovery: 20 healthy ACKs at low RTT on an underfilled path.
        sim_acks(&mut client, 20, 30_000, 1.0);

        assert!(
            client.probe_frames > post_blip,
            "probe_frames {:.1} should have recovered from {post_blip:.1} after healthy ACKs",
            client.probe_frames,
        );
    }

    #[test]
    fn jitter_decays_fast_on_browser_ready_path() {
        let mut client = browser_ready_high_bandwidth_client();

        // Inject elevated jitter (simulating post-blip state).
        client.max_goodput_jitter_bps = client.goodput_bps * 0.4;
        client.goodput_jitter_bps = client.goodput_bps * 0.3;
        let initial_jitter = client.max_goodput_jitter_bps;

        // 10 healthy ACKs on a browser-ready path.
        sim_acks(&mut client, 10, 30_000, 1.0);

        assert!(
            client.max_goodput_jitter_bps < initial_jitter * 0.5,
            "max_goodput_jitter_bps {:.0} should have decayed below {:.0} \
             (50% of initial {initial_jitter:.0}) after 10 healthy ACKs on a ready path",
            client.max_goodput_jitter_bps,
            initial_jitter * 0.5,
        );
    }

    #[test]
    fn byte_budget_uses_floor_when_goodput_depressed() {
        let mut client = browser_ready_high_bandwidth_client();
        client.goodput_bps = 100_000.0;

        let budget = byte_budget_for(&client, 100.0);
        let floor_budget = (bandwidth_floor_bps(&client) * 100.0 / 1_000.0).ceil() as usize;

        assert!(
            budget >= floor_budget,
            "byte_budget {budget} should be at least bandwidth_floor-based {floor_budget} \
             when goodput_bps is depressed but delivery_bps is high"
        );
    }

    #[test]
    fn probe_floor_maintained_under_congestion_signal() {
        let mut client = test_client();
        client.display_fps = 120.0;
        client.rtt_ms = 40.0;
        client.min_rtt_ms = 40.0;
        client.goodput_bps = 5_000_000.0;
        client.delivery_bps = 5_000_000.0;
        client.last_goodput_sample_bps = 5_000_000.0;
        client.avg_paced_frame_bytes = 10_000.0;
        client.avg_preview_frame_bytes = 1_024.0;
        client.probe_frames = 10.0;

        // Many ACKs with high RTT: probes should not drop below the floor.
        sim_acks(&mut client, 20, 10_000, 600.0);

        assert!(
            client.probe_frames >= 1.0,
            "probe_frames {:.1} should not drop below the floor of 1.0",
            client.probe_frames,
        );
    }

    // ── parse_terminal_queries ──

    #[test]
    fn parse_tq_da1_bare() {
        let results = parse_terminal_queries(b"\x1b[c", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert!(results[0].starts_with("\x1b[?64;"));
    }

    #[test]
    fn parse_tq_da1_with_zero_param() {
        let results = parse_terminal_queries(b"\x1b[0c", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert!(results[0].starts_with("\x1b[?64;"));
    }

    #[test]
    fn parse_tq_dsr_cursor_position() {
        let results = parse_terminal_queries(b"\x1b[6n", (24, 80), (5, 10));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b[6;11R");
    }

    #[test]
    fn parse_tq_dsr_status() {
        let results = parse_terminal_queries(b"\x1b[5n", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b[0n");
    }

    #[test]
    fn parse_tq_window_size_cells() {
        let results = parse_terminal_queries(b"\x1b[18t", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b[8;24;80t");
    }

    #[test]
    fn parse_tq_window_size_pixels() {
        let results = parse_terminal_queries(b"\x1b[14t", (30, 100), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b[4;480;800t");
    }

    #[test]
    fn parse_tq_multiple_queries() {
        let data = b"\x1b[c\x1b[6n\x1b[5n";
        let results = parse_terminal_queries(data, (24, 80), (2, 3));
        assert_eq!(results.len(), 3);
        assert!(results[0].starts_with("\x1b[?64;"));
        assert_eq!(results[1], "\x1b[3;4R");
        assert_eq!(results[2], "\x1b[0n");
    }

    #[test]
    fn parse_tq_question_mark_sequences_skipped() {
        let results = parse_terminal_queries(b"\x1b[?1h", (24, 80), (0, 0));
        assert!(results.is_empty());
    }

    #[test]
    fn parse_tq_unknown_final_byte_ignored() {
        let results = parse_terminal_queries(b"\x1b[42z", (24, 80), (0, 0));
        assert!(results.is_empty());
    }

    #[test]
    fn parse_tq_empty_input() {
        let results = parse_terminal_queries(b"", (24, 80), (0, 0));
        assert!(results.is_empty());
    }

    #[test]
    fn parse_tq_plain_text_no_csi() {
        let results = parse_terminal_queries(b"hello world", (24, 80), (0, 0));
        assert!(results.is_empty());
    }

    #[test]
    fn parse_tq_interleaved_with_text() {
        let results = parse_terminal_queries(b"abc\x1b[cdef\x1b[6n", (24, 80), (1, 2));
        assert_eq!(results.len(), 2);
    }

    // ── parse_terminal_queries: OSC ──

    #[test]
    fn parse_tq_osc11_background_color_bel() {
        let results = parse_terminal_queries(b"\x1b]11;?\x07", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b]11;rgb:0000/0000/0000\x1b\\");
    }

    #[test]
    fn parse_tq_osc11_background_color_st() {
        let results = parse_terminal_queries(b"\x1b]11;?\x1b\\", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b]11;rgb:0000/0000/0000\x1b\\");
    }

    #[test]
    fn parse_tq_osc10_foreground_color() {
        let results = parse_terminal_queries(b"\x1b]10;?\x07", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b]10;rgb:ffff/ffff/ffff\x1b\\");
    }

    #[test]
    fn parse_tq_osc4_palette_color_0() {
        let results = parse_terminal_queries(b"\x1b]4;0;?\x07", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b]4;0;rgb:0000/0000/0000\x1b\\");
    }

    #[test]
    fn parse_tq_osc4_palette_color_1() {
        let results = parse_terminal_queries(b"\x1b]4;1;?\x07", (24, 80), (0, 0));
        assert_eq!(results.len(), 1);
        assert_eq!(results[0], "\x1b]4;1;rgb:8080/0000/0000\x1b\\");
    }

    #[test]
    fn parse_tq_osc_mixed_with_csi() {
        let results =
            parse_terminal_queries(b"\x1b]11;?\x07\x1b[c\x1b]4;0;?\x07", (24, 80), (0, 0));
        assert_eq!(results.len(), 3);
        assert!(results[0].starts_with("\x1b]11;"));
        assert!(results[1].starts_with("\x1b[?64;"));
        assert!(results[2].starts_with("\x1b]4;0;"));
    }

    // ── build_search_results_msg ──

    #[test]
    fn search_results_empty() {
        let msg = build_search_results_msg(42, &[]);
        assert_eq!(msg[0], S2C_SEARCH_RESULTS);
        assert_eq!(u16::from_le_bytes([msg[1], msg[2]]), 42);
        assert_eq!(u16::from_le_bytes([msg[3], msg[4]]), 0);
        assert_eq!(msg.len(), 5);
    }

    #[test]
    fn search_results_single() {
        let results = vec![SearchResultRow {
            pty_id: 7,
            score: 100,
            primary_source: 1,
            matched_sources: 3,
            context: "hello".into(),
            scroll_offset: Some(42),
        }];
        let msg = build_search_results_msg(1, &results);
        assert_eq!(msg[0], S2C_SEARCH_RESULTS);
        assert_eq!(u16::from_le_bytes([msg[3], msg[4]]), 1);
        let pty_id = u16::from_le_bytes([msg[5], msg[6]]);
        assert_eq!(pty_id, 7);
        let score = u32::from_le_bytes([msg[7], msg[8], msg[9], msg[10]]);
        assert_eq!(score, 100);
        assert_eq!(msg[11], 1);
        assert_eq!(msg[12], 3);
        let scroll = u32::from_le_bytes([msg[13], msg[14], msg[15], msg[16]]);
        assert_eq!(scroll, 42);
        let ctx_len = u16::from_le_bytes([msg[17], msg[18]]) as usize;
        assert_eq!(ctx_len, 5);
        assert_eq!(&msg[19..19 + ctx_len], b"hello");
    }

    #[test]
    fn search_results_none_scroll_offset() {
        let results = vec![SearchResultRow {
            pty_id: 1,
            score: 0,
            primary_source: 0,
            matched_sources: 0,
            context: String::new(),
            scroll_offset: None,
        }];
        let msg = build_search_results_msg(0, &results);
        let scroll = u32::from_le_bytes([msg[13], msg[14], msg[15], msg[16]]);
        assert_eq!(scroll, u32::MAX);
    }

    // ── allocate_pty_id ──

    #[test]
    fn allocate_pty_id_empty_session() {
        let mut sess = Session::new();
        assert_eq!(sess.allocate_pty_id(0), Some(1));
    }

    #[test]
    fn allocate_pty_id_rotates() {
        let mut sess = Session::new();
        // Sequential allocations return increasing IDs (not always 1).
        assert_eq!(sess.allocate_pty_id(0), Some(1));
        assert_eq!(sess.allocate_pty_id(0), Some(2));
        assert_eq!(sess.allocate_pty_id(0), Some(3));
    }

    #[test]
    fn allocate_pty_id_wraps_at_max() {
        let mut sess = Session::new();
        sess.next_pty_id = u16::MAX;
        assert_eq!(sess.allocate_pty_id(0), Some(u16::MAX));
        // Next allocation wraps to 1.
        assert_eq!(sess.allocate_pty_id(0), Some(1));
    }

    // ── try_send_update ──

    #[test]
    fn try_send_no_change() {
        let mut client = test_client();
        let frame = sample_frame("x");
        let now = Instant::now();
        let outcome = try_send_update(&mut client, 1, frame, None, now, false);
        assert!(matches!(outcome, SendOutcome::NoChange));
    }

    #[test]
    fn try_send_sent() {
        let (mut client, _rx) = test_client_with_capacity(8);
        let frame = sample_frame("x");
        let now = Instant::now();
        let outcome = try_send_update(
            &mut client,
            1,
            frame.clone(),
            Some(vec![1, 2, 3]),
            now,
            true,
        );
        assert!(matches!(outcome, SendOutcome::Sent));
        assert!(client.last_sent.contains_key(&1));
    }

    #[test]
    fn try_send_backpressured() {
        let (mut client, _rx) = test_client_with_capacity(1);
        let frame = sample_frame("x");
        let now = Instant::now();
        let _ = try_send_outbox(&client, vec![0]);
        let outcome = try_send_update(
            &mut client,
            1,
            frame.clone(),
            Some(vec![1, 2, 3]),
            now,
            true,
        );
        assert!(matches!(outcome, SendOutcome::Backpressured));
        assert!(
            client.last_sent.contains_key(&1),
            "last_sent should advance even on backpressure"
        );
    }
}