use crate::pty::{build_shell_command, resolve_shell, spawn_reader_thread};
use crate::state::{OUTPUT_RING_BUFFER_CAPACITY, VT_LOG_BUFFER_CAPACITY, VtLogBuffer};
use crate::{AppState, MAX_CONCURRENT_SESSIONS, OutputRingBuffer, PtySession};
use axum::Json;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use futures_util::stream::StreamExt;
use parking_lot::Mutex;
use portable_pty::{PtySize, native_pty_system};
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
#[cfg(feature = "desktop")]
use tauri::Emitter;
use uuid::Uuid;

use super::types::*;

/// Standard 404 response for missing sessions.
fn session_not_found() -> (StatusCode, Json<serde_json::Value>) {
    (
        StatusCode::NOT_FOUND,
        Json(serde_json::json!({"error": "Session not found"})),
    )
}

pub(super) async fn health(State(state): State<Arc<AppState>>) -> Json<HealthResponse> {
    let uptime = state.server_start_time.elapsed().as_secs();
    let session_count = state.sessions.len();
    #[cfg(unix)]
    let socket_path = {
        let p = state.bound_socket_path.read();
        if p.as_os_str().is_empty() {
            None
        } else {
            Some(p.display().to_string())
        }
    };
    #[cfg(not(unix))]
    let socket_path = None;
    Json(HealthResponse {
        ok: true,
        uptime_secs: uptime,
        session_count,
        protocol_version: 1,
        socket_path,
    })
}

pub(super) async fn app_version() -> Json<super::types::VersionResponse> {
    Json(super::types::VersionResponse {
        version: env!("CARGO_PKG_VERSION"),
        git_hash: env!("BUILD_GIT_HASH"),
    })
}

pub(super) async fn list_sessions(State(state): State<Arc<AppState>>) -> Json<Vec<SessionInfo>> {
    let sessions: Vec<SessionInfo> = state
        .sessions
        .iter()
        .map(|entry| {
            let session_id = entry.key().clone();
            let session = entry.value().lock();
            let session_state = state.session_state_with_shell(&session_id);
            SessionInfo {
                session_id,
                cwd: session.cwd.clone(),
                worktree_path: session
                    .worktree
                    .as_ref()
                    .map(|w| w.path.to_string_lossy().to_string()),
                worktree_branch: session.worktree.as_ref().and_then(|w| w.branch.clone()),
                display_name: session.display_name.clone(),
                state: session_state,
            }
        })
        .collect();
    Json(sessions)
}

pub(super) async fn write_to_session(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<WriteRequest>,
) -> impl IntoResponse {
    let entry = match state.sessions.get(&session_id) {
        Some(e) => e,
        None => return session_not_found(),
    };
    let mut session = entry.lock();
    if let Err(e) = session.writer.write_all(body.data.as_bytes()) {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": format!("Write failed: {}", e)})),
        );
    }
    if let Err(e) = session.writer.flush() {
        tracing::warn!(session_id = %session_id, "PTY flush failed: {e}");
    }

    // Feed input through InputLineBuffer FSM to track slash_mode accurately.
    // The old substring heuristic false-positived on pastes starting with '/'.
    let input_entry = state
        .input_buffers
        .entry(session_id.clone())
        .or_insert_with(|| {
            parking_lot::Mutex::new(crate::input_line_buffer::InputLineBuffer::new())
        });
    let mut buf = input_entry.lock();
    let actions = buf.feed(&body.data);
    let line_submitted = actions.iter().any(|a| {
        matches!(
            a,
            crate::input_line_buffer::InputAction::Line(_)
                | crate::input_line_buffer::InputAction::Interrupt
        )
    });
    // Determine slash mode. The InputLineBuffer may accumulate junk from
    // terminal responses (e.g. DA reply "1;2c"), so buf.content() alone is
    // unreliable. Use multiple signals:
    let in_slash = if line_submitted {
        false
    } else if buf.content().starts_with('/') {
        true
    } else if body.data == "/" {
        // Fresh slash keystroke from PWA — always enters slash mode
        true
    } else {
        // Maintain current slash_mode for subsequent chars (delta sync sends
        // one char at a time after the initial "/"), unless dismissed
        let is_bare_esc =
            body.data == "\x1b" || (body.data.contains('\x1b') && !body.data.contains("\x1b["));
        let dismissed = is_bare_esc || body.data.contains('\x03');
        !dismissed
            && state
                .slash_mode
                .get(&session_id)
                .is_some_and(|v| v.load(std::sync::atomic::Ordering::Relaxed))
    };
    tracing::trace!(
        "write_pty slash_mode: in_slash={in_slash} buf='{}' data='{}'",
        buf.content(),
        body.data
    );
    state
        .slash_mode
        .entry(session_id.clone())
        .or_insert_with(|| std::sync::atomic::AtomicBool::new(false))
        .store(in_slash, std::sync::atomic::Ordering::Relaxed);

    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn set_session_name(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<SetNameRequest>,
) -> impl IntoResponse {
    let entry = match state.sessions.get(&session_id) {
        Some(e) => e,
        None => return session_not_found(),
    };
    entry.lock().display_name = body.name;
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn resize_session(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<ResizeRequest>,
) -> impl IntoResponse {
    if let Err(msg) = super::validate_terminal_size(body.rows, body.cols) {
        return (
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({"error": msg})),
        );
    }
    let entry = match state.sessions.get(&session_id) {
        Some(e) => e,
        None => return session_not_found(),
    };
    let session = entry.lock();
    if let Err(e) = session.master.resize(PtySize {
        rows: body.rows,
        cols: body.cols,
        pixel_width: 0,
        pixel_height: 0,
    }) {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": format!("Resize failed: {}", e)})),
        );
    }
    drop(session);
    // Resize VT log buffer to match new terminal dimensions.
    if let Some(vt_log) = state.vt_log_buffers.get(&session_id) {
        vt_log.lock().resize(body.rows, body.cols);
    }
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn get_output(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Query(query): Query<OutputQuery>,
) -> impl IntoResponse {
    let format = query.format.as_deref().unwrap_or("raw");

    // format=log: return VT100-extracted clean log lines (best for mobile/REST consumers)
    if format == "log" {
        let vt_log = match state.vt_log_buffers.get(&session_id) {
            Some(b) => b,
            None => return session_not_found(),
        };
        let buf = vt_log.lock();
        let limit = query.limit.unwrap_or(usize::MAX);
        let total = buf.total_lines();
        let offset = match query.offset {
            Some(o) => o.min(total),
            None => total.saturating_sub(limit),
        };
        let (lines, _) = buf.lines_since_owned(offset, limit);
        let trim = trim_screen_chrome(buf.screen_rows());
        // Get styled screen rows, trimmed to same cutoff
        let styled = buf.screen_log_lines();
        let screen: Vec<_> = styled.into_iter().take(trim.cutoff).collect();
        let input_line = buf.prompt_input_text();
        let mut resp = serde_json::json!({
            "lines": lines,
            "total_lines": total,
            "screen": screen,
        });
        if let Some(il) = &input_line {
            resp["input_line"] = serde_json::json!(il);
        }
        return (StatusCode::OK, Json(resp));
    }

    // format=text: serve clean rows from VtLogBuffer (no strip_ansi needed)
    if format == "text" {
        let vt_log = match state.vt_log_buffers.get(&session_id) {
            Some(b) => b,
            None => return session_not_found(),
        };
        let buf = vt_log.lock();
        let limit = query.limit.unwrap_or(usize::MAX);
        let total = buf.total_lines();
        let offset = total.saturating_sub(limit);
        let (log_lines, _) = buf.lines_since_owned(offset, limit);
        // Append current visible screen rows (non-empty) after the log
        let screen: Vec<String> = buf
            .screen_rows()
            .into_iter()
            .filter(|r| !r.is_empty())
            .collect();
        let mut all_lines: Vec<String> = log_lines.iter().map(|ll| ll.text()).collect();
        all_lines.extend(screen);
        let data = all_lines.join("\n");
        return (
            StatusCode::OK,
            Json(serde_json::json!({
                "data": data,
                "data_length": data.len(),
                "total_written": total,
            })),
        );
    }

    let ring = match state.output_buffers.get(&session_id) {
        Some(r) => r,
        None => return session_not_found(),
    };
    let limit = query.limit.unwrap_or(8192);
    let (bytes, total_written) = ring.lock().read_last(limit);
    let raw = String::from_utf8_lossy(&bytes).to_string();
    let data = raw;
    (
        StatusCode::OK,
        Json(serde_json::json!({
            "data": data,
            "data_length": data.len(),
            "total_written": total_written
        })),
    )
}

pub(super) async fn close_session(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    if state.sessions.contains_key(&session_id) {
        // Send Ctrl+C then cleanup
        if let Some(entry) = state.sessions.get(&session_id) {
            let mut session = entry.lock();
            let _ = session.writer.write_all(&[0x03]);
            let _ = session.writer.flush();
        }
        crate::pty::cleanup_session(&session_id, &state);

        // Broadcast to SSE/WebSocket consumers
        tracing::info!(source = "session", session_id = %session_id, "Session closed: explicit close");
        let _ = state.event_bus.send(crate::state::AppEvent::SessionClosed {
            session_id: session_id.clone(),
            reason: "explicit_close".to_string(),
        });
        #[cfg(feature = "desktop")]
        if let Some(app) = state.app_handle.read().as_ref() {
            let _ = app.emit(
                "session-closed",
                serde_json::json!({
                    "session_id": session_id,
                    "reason": "explicit_close",
                }),
            );
        }

        (StatusCode::OK, Json(serde_json::json!({"ok": true})))
    } else {
        session_not_found()
    }
}

/// Shared PTY setup: opens a PTY, spawns the shell, registers buffers and reader thread.
///
/// Returns `(session_id, cwd_string)` on success. Both `create_session` and
/// `create_session_with_worktree` delegate here after deriving the cwd and worktree.
pub(super) fn spawn_pty_session(
    state: Arc<AppState>,
    shell: String,
    cwd: Option<String>,
    rows: u16,
    cols: u16,
    worktree: Option<crate::state::WorktreeInfo>,
) -> Result<String, (StatusCode, Json<serde_json::Value>)> {
    let session_id = Uuid::new_v4().to_string();
    let pty_system = native_pty_system();

    let pair = pty_system
        .openpty(PtySize {
            rows,
            cols,
            pixel_width: 0,
            pixel_height: 0,
        })
        .map_err(|e| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({"error": format!("Failed to open PTY: {}", e)})),
            )
        })?;

    let mut cmd = build_shell_command(&shell);
    if let Some(ref dir) = cwd {
        let dir = crate::cli::expand_tilde(dir);
        cmd.cwd(dir);
    }

    let child = pair.slave.spawn_command(cmd).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": format!("Failed to spawn shell: {}", e)})),
        )
    })?;

    let writer = pair.master.take_writer().map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": format!("Failed to get PTY writer: {}", e)})),
        )
    })?;

    let reader = pair.master.try_clone_reader().map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": format!("Failed to get PTY reader: {}", e)})),
        )
    })?;

    let paused = Arc::new(AtomicBool::new(false));
    state.sessions.insert(
        session_id.clone(),
        Mutex::new(PtySession {
            writer,
            master: pair.master,
            _child: child,
            paused: paused.clone(),
            worktree,
            cwd: cwd.clone(),
            display_name: None,
            shell: shell.clone(),
        }),
    );
    state.assign_term_alias(&session_id);
    state.metrics.total_spawned.fetch_add(1, Ordering::Relaxed);
    state
        .metrics
        .active_sessions
        .fetch_add(1, Ordering::Relaxed);

    state.output_buffers.insert(
        session_id.clone(),
        Mutex::new(OutputRingBuffer::new(OUTPUT_RING_BUFFER_CAPACITY)),
    );
    state.vt_log_buffers.insert(
        session_id.clone(),
        Mutex::new(VtLogBuffer::new(
            rows,
            cols.max(220),
            VT_LOG_BUFFER_CAPACITY,
        )),
    );
    state
        .last_output_ms
        .insert(session_id.clone(), std::sync::atomic::AtomicU64::new(0));
    let (grid_watch_tx, _) = tokio::sync::watch::channel(Vec::new());
    state.grid_watch.insert(session_id.clone(), grid_watch_tx);

    // Broadcast to SSE/WebSocket consumers (before state is moved to reader thread)
    let _ = state
        .event_bus
        .send(crate::state::AppEvent::SessionCreated {
            session_id: session_id.clone(),
            cwd: cwd.clone(),
            agent_type: None,
        });

    #[cfg(feature = "desktop")]
    let state_ref = state.clone();
    spawn_reader_thread(reader, paused, session_id.clone(), state, None);

    #[cfg(feature = "desktop")]
    if let Some(app) = state_ref.app_handle.read().as_ref() {
        let _ = app.emit(
            "session-created",
            serde_json::json!({
                "session_id": session_id,
                "cwd": cwd,
            }),
        );
    }

    Ok(session_id)
}

pub(super) async fn create_session(
    State(state): State<Arc<AppState>>,
    Json(body): Json<CreateSessionRequest>,
) -> impl IntoResponse {
    if state.sessions.len() >= MAX_CONCURRENT_SESSIONS {
        return (
            StatusCode::TOO_MANY_REQUESTS,
            Json(serde_json::json!({"error": "Max concurrent sessions reached"})),
        );
    }

    let rows = body.rows.unwrap_or(24);
    let cols = body.cols.unwrap_or(80);
    if let Err(msg) = super::validate_terminal_size(rows, cols) {
        return (
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({"error": msg})),
        );
    }
    let shell = resolve_shell(body.shell);

    match spawn_pty_session(state, shell, body.cwd, rows, cols, None) {
        Ok(session_id) => (
            StatusCode::CREATED,
            Json(serde_json::json!({"session_id": session_id})),
        ),
        Err(err) => err,
    }
}

pub(super) async fn pause_session(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let entry = match state.sessions.get(&session_id) {
        Some(e) => e,
        None => return session_not_found(),
    };
    entry.lock().paused.store(true, Ordering::Relaxed);
    state
        .metrics
        .pauses_triggered
        .fetch_add(1, Ordering::Relaxed);
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn resume_session(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let entry = match state.sessions.get(&session_id) {
        Some(e) => e,
        None => return session_not_found(),
    };
    entry.lock().paused.store(false, Ordering::Relaxed);
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn get_kitty_flags(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let flags = state
        .kitty_states
        .get(&session_id)
        .map(|entry| entry.lock().current_flags())
        .unwrap_or(0);
    (StatusCode::OK, Json(serde_json::json!(flags)))
}

pub(super) async fn get_foreground_process(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let agent = (|| -> Option<String> {
        let entry = state.sessions.get(&session_id)?;
        let session = entry.value().lock();
        #[cfg(not(windows))]
        {
            let pgid = session.master.process_group_leader()?;
            let name = crate::pty::process_name_from_pid(pgid as u32)?;
            crate::pty::classify_agent(&name).map(|s| s.to_string())
        }
        #[cfg(windows)]
        {
            drop(session);
            None
        }
    })();

    match agent {
        Some(name) => (StatusCode::OK, Json(serde_json::json!({"agent": name}))),
        None => (StatusCode::OK, Json(serde_json::json!({"agent": null}))),
    }
}

pub(super) async fn get_stats(State(state): State<Arc<AppState>>) -> impl IntoResponse {
    Json(state.orchestrator_stats())
}

pub(super) async fn get_metrics(State(state): State<Arc<AppState>>) -> impl IntoResponse {
    Json(state.session_metrics_json())
}

pub(super) async fn get_process_stats(State(state): State<Arc<AppState>>) -> impl IntoResponse {
    Json(crate::pty::collect_process_stats(&state))
}

pub(super) async fn process_monitor_panel() -> impl IntoResponse {
    axum::response::Html(include_str!("process_monitor.html"))
}

pub(super) async fn create_session_with_worktree(
    State(state): State<Arc<AppState>>,
    Json(body): Json<CreateSessionWithWorktreeRequest>,
) -> impl IntoResponse {
    if state.sessions.len() >= MAX_CONCURRENT_SESSIONS {
        return (
            StatusCode::TOO_MANY_REQUESTS,
            Json(serde_json::json!({"error": "Max concurrent sessions reached"})),
        );
    }

    // Create the worktree first
    let wt_config = crate::worktree::WorktreeConfig {
        task_name: body.branch_name.clone(),
        base_repo: body.base_repo,
        branch: Some(body.branch_name),
        create_branch: true,
    };
    let worktrees_dir = crate::worktree::resolve_worktree_dir_for_repo(
        std::path::Path::new(&wt_config.base_repo),
        &state.worktrees_dir,
    );
    let wt_config_bg = wt_config.clone();
    let worktree = match tokio::task::spawn_blocking(move || {
        crate::worktree::create_worktree_with_stale_recovery(&worktrees_dir, &wt_config_bg, None)
    })
    .await
    {
        Ok(Ok(wt)) => wt,
        Ok(Err(e)) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({"error": e})),
            );
        }
        Err(e) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({"error": format!("task panic: {e}")})),
            );
        }
    };

    let base_repo = wt_config.base_repo.clone();
    state.invalidate_repo_caches(&base_repo);
    let worktree_path_str = worktree.path.to_string_lossy().to_string();
    let worktree_branch = worktree.branch.clone();
    let branch_name = worktree_branch.clone().unwrap_or_default();
    let _ = state
        .event_bus
        .send(crate::state::AppEvent::WorktreeCreated {
            repo_path: base_repo.clone(),
            branch: branch_name.clone(),
            worktree_path: worktree_path_str.clone(),
        });
    #[cfg(feature = "desktop")]
    if let Some(handle) = state.app_handle.read().as_ref() {
        let _ = handle.emit(
            "worktree-created",
            serde_json::json!({
                "repo_path": &base_repo,
                "branch": &branch_name,
                "worktree_path": &worktree_path_str,
            }),
        );
    }

    let rows = body.config.rows.unwrap_or(24);
    let cols = body.config.cols.unwrap_or(80);
    if let Err(msg) = super::validate_terminal_size(rows, cols) {
        return (
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({"error": msg})),
        );
    }
    let shell = resolve_shell(body.config.shell);

    match spawn_pty_session(
        state,
        shell,
        Some(worktree_path_str.clone()),
        rows,
        cols,
        Some(worktree),
    ) {
        Ok(session_id) => {
            let mut response = serde_json::json!({
                "session_id": session_id,
                "worktree_path": worktree_path_str.clone(),
                "branch": worktree_branch,
            });
            let repo_for_script = base_repo.clone();
            let cwd_for_script = worktree_path_str;
            if let Some(script) = tokio::task::spawn_blocking(move || {
                crate::config::resolve_effective_setup_script(&repo_for_script)
            })
            .await
            .ok()
            .flatten()
            {
                match tokio::task::spawn_blocking(move || {
                    crate::worktree::run_setup_script(script, cwd_for_script)
                })
                .await
                {
                    Ok(Ok(result)) => {
                        response["setup_script"] = result;
                    }
                    Ok(Err(e)) => {
                        response["setup_script_error"] = serde_json::json!(e);
                    }
                    Err(e) => {
                        response["setup_script_error"] =
                            serde_json::json!(format!("task panic: {e}"));
                    }
                }
            }
            (StatusCode::CREATED, Json(response))
        }
        Err(err) => err,
    }
}

/// WebSocket upgrade handler for streaming PTY output.
/// Bidirectional: server sends PTY output, client sends PTY input.
/// Supports `?format=text` to strip ANSI, `?format=log` for VT100 log lines.
pub(super) async fn ws_stream(
    ws: WebSocketUpgrade,
    Path(id): Path<String>,
    Query(query): Query<OutputQuery>,
    State(state): State<Arc<AppState>>,
) -> Response {
    if !state.sessions.contains_key(&id) {
        return StatusCode::NOT_FOUND.into_response();
    }
    let format = query.format.as_deref().unwrap_or("raw");

    if format == "grid" {
        return ws
            .write_buffer_size(64 * 1024)
            .max_write_buffer_size(256 * 1024)
            .on_upgrade(move |socket| handle_ws_grid_session(socket, id, state));
    }

    // format=text and format=log both serve clean VtLogBuffer rows (no strip_ansi).
    let log_mode = format == "log" || format == "text";
    let initial_offset = query.offset;
    ws.on_upgrade(move |socket| handle_ws_session(socket, id, state, log_mode, initial_offset))
}

/// Handle a WebSocket connection for a PTY session.
///
/// Multiplexes two streams to the client:
/// 1. Raw PTY output via mpsc channel → `{"type":"output","data":"..."}`
/// 2. Parsed events via broadcast channel → `{"type":"parsed","event":{...}}`
///
/// When `log_mode` is true (`?format=log` or `?format=text`), instead of raw PTY
/// output the client receives VT100-extracted log lines:
/// `{"type":"log","lines":[...],"offset":N}`
///
/// Client → server messages are written to the PTY as input.
async fn handle_ws_session(
    socket: WebSocket,
    session_id: String,
    state: Arc<AppState>,
    log_mode: bool,
    initial_offset: Option<usize>,
) {
    let (mut ws_sender, mut ws_receiver) = socket.split();

    if log_mode {
        // Log/text mode: stream clean VtLogBuffer rows, no raw PTY chunks
        handle_ws_log_session(
            ws_sender,
            ws_receiver,
            session_id,
            state,
            initial_offset.unwrap_or(0),
        )
        .await;
        return;
    }

    // Subscribe to broadcast channel for parsed events (filtered by session_id)
    let mut event_rx = state.event_bus.subscribe();

    // Snapshot the ring buffer and register the live mpsc subscription
    // atomically while holding ring.lock(). The PTY writer takes the same
    // lock when appending + broadcasting to ws_clients, so serializing the
    // two sides guarantees every byte is delivered either via catch-up or
    // via the live channel — never both (duplicate) nor neither (gap).
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<String>();
    let snapshot = state.output_buffers.get(&session_id).map(|ring| {
        let r = ring.lock();
        let snap = if let Some(off) = initial_offset {
            r.read_since(off as u64)
        } else {
            r.read_last(OUTPUT_RING_BUFFER_CAPACITY)
        };
        state
            .ws_clients
            .entry(session_id.clone())
            .or_default()
            .push(tx);
        drop(r);
        snap
    });

    // Send catch-up data in chunks (64 KB) so the client can render progressively.
    const CATCHUP_CHUNK_SIZE: usize = 64 * 1024;
    if let Some((data, total)) = snapshot
        && !data.is_empty()
    {
        for chunk in data.chunks(CATCHUP_CHUNK_SIZE) {
            let text = String::from_utf8_lossy(chunk);
            if !text.is_empty() {
                let frame =
                    serde_json::json!({"type": "output", "data": text, "total_written": total});
                if futures_util::SinkExt::send(
                    &mut ws_sender,
                    Message::Text(frame.to_string().into()),
                )
                .await
                .is_err()
                {
                    return; // Client disconnected during catch-up
                }
            }
        }
    }

    // Spawn a task to forward PTY output + parsed events to the WebSocket
    let sid_for_events = session_id.clone();
    let send_task = tokio::spawn(async move {
        loop {
            tokio::select! {
                // Raw PTY output from mpsc channel
                data = rx.recv() => {
                    let Some(data) = data else { break };
                    let frame = serde_json::json!({"type": "output", "data": data});
                    if futures_util::SinkExt::send(
                        &mut ws_sender,
                        Message::Text(frame.to_string().into()),
                    ).await.is_err() {
                        break;
                    }
                }
                // Parsed events from broadcast channel
                result = event_rx.recv() => {
                    match result {
                        Ok(event) => {
                            // Filter: only forward events for this session
                            let matches = match &event {
                                crate::state::AppEvent::PtyParsed { session_id: sid, .. } => sid == &sid_for_events,
                                crate::state::AppEvent::PtyExit { session_id: sid } => sid == &sid_for_events,
                                crate::state::AppEvent::SessionClosed { session_id: sid, .. } => sid == &sid_for_events,
                                _ => false,
                            };
                            if !matches { continue; }

                            // Extract the inner payload (without serde tag wrapping)
                            let payload = match &event {
                                crate::state::AppEvent::PtyParsed { parsed, .. } => {
                                    serde_json::json!({"type": "parsed", "event": parsed})
                                }
                                crate::state::AppEvent::PtyExit { session_id: sid } => {
                                    serde_json::json!({"type": "exit", "session_id": sid})
                                }
                                crate::state::AppEvent::SessionClosed { session_id: sid, reason } => {
                                    serde_json::json!({"type": "closed", "session_id": sid, "reason": reason})
                                }
                                _ => continue,
                            };
                            if futures_util::SinkExt::send(
                                &mut ws_sender,
                                Message::Text(payload.to_string().into()),
                            ).await.is_err() {
                                break;
                            }
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                            tracing::warn!(session_id = %sid_for_events, lagged = n, "WebSocket broadcast lagged");
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                    }
                }
            }
        }
    });

    // Read messages from the client and write to PTY
    let state_clone = state.clone();
    let sid = session_id.clone();
    while let Some(Ok(msg)) = ws_receiver.next().await {
        match msg {
            Message::Text(text) => {
                if let Some(session) = state_clone.sessions.get(&sid) {
                    let mut s = session.lock();
                    if let Err(e) = s.writer.write_all(text.as_bytes()) {
                        tracing::error!(session_id = %sid, "PTY write failed: {e}");
                        break;
                    }
                    if let Err(e) = s.writer.flush() {
                        tracing::warn!(session_id = %sid, "PTY flush failed: {e}");
                    }
                }
            }
            Message::Binary(data) => {
                if let Some(session) = state_clone.sessions.get(&sid) {
                    let mut s = session.lock();
                    if let Err(e) = s.writer.write_all(&data) {
                        tracing::error!(session_id = %sid, "PTY write failed: {e}");
                        break;
                    }
                    if let Err(e) = s.writer.flush() {
                        tracing::warn!(session_id = %sid, "PTY flush failed: {e}");
                    }
                }
            }
            Message::Close(_) => break,
            _ => {}
        }
    }

    // Client disconnected — abort the send task and purge the dead sender
    send_task.abort();
    crate::state::purge_dead_ws_clients(&state.ws_clients, &session_id);
}

/// Handle a WebSocket connection in log mode (`?format=log`).
///
/// Sends VT100-extracted log lines: catch-up on connect, then polls for new
/// lines every 200 ms and batches them as `{"type":"log","lines":[...],"offset":N}`.
/// The client can still send PTY input (written as-is to the PTY).
async fn handle_ws_log_session(
    mut ws_sender: futures_util::stream::SplitSink<WebSocket, Message>,
    mut ws_receiver: futures_util::stream::SplitStream<WebSocket>,
    session_id: String,
    state: Arc<AppState>,
    skip_offset: usize,
) {
    // Send catch-up: only lines accumulated AFTER skip_offset.
    // When the client already fetched lines via HTTP, skip_offset = total_lines
    // from that response, so the catch-up only sends the delta.
    let initial_offset = {
        if let Some(vt_log) = state.vt_log_buffers.get(&session_id) {
            let (total, catchup_frame) = {
                let buf = vt_log.lock();
                let total = buf.total_lines();
                let frame = if total > skip_offset {
                    let (lines, _) = buf.lines_since_owned(skip_offset, usize::MAX);
                    if !lines.is_empty() {
                        // total_lines is the post-read cursor (monotonic): the client
                        // stores it and passes it back as ?offset= on reconnect, so the
                        // next catch-up resumes from here instead of replaying from mount.
                        Some(serde_json::json!({"type": "log", "lines": lines, "offset": skip_offset, "total_lines": total}).to_string())
                    } else {
                        None
                    }
                } else {
                    None
                };
                (total, frame)
            }; // lock released here
            if let Some(frame_str) = catchup_frame {
                let _ =
                    futures_util::SinkExt::send(&mut ws_sender, Message::Text(frame_str.into()))
                        .await;
            }
            total
        } else {
            0
        }
    };

    // Spawn polling task: check for new lines every 200ms AND forward state changes.
    let sid_poll = session_id.clone();
    let state_poll = state.clone();
    let send_task = tokio::spawn(async move {
        let mut offset = initial_offset;
        let mut event_rx = state_poll.event_bus.subscribe();
        let mut prev_screen_hash: u64 = 0;
        // Dedup: only send state frames when SessionState actually changed
        let mut prev_state: Option<crate::state::SessionState> = None;

        // Send initial state snapshot so the client has the correct status immediately
        if let Some(current) = state_poll.session_state_with_shell(&sid_poll) {
            let frame = serde_json::json!({"type": "state", "state": &current});
            prev_state = Some(current);
            let _ = futures_util::SinkExt::send(
                &mut ws_sender,
                Message::Text(frame.to_string().into()),
            )
            .await;
        }

        loop {
            // Track whether we need to check state and/or send log frames
            enum LoopAction {
                Poll,        // sleep arm: check state + send log/screen
                Event,       // event arm: check state only (relevant event)
                Skip,        // event arm: irrelevant event, skip state check
                SessionGone, // vt_log_buffer missing, exit loop
            }

            let action = tokio::select! {
                _ = tokio::time::sleep(tokio::time::Duration::from_millis(200)) => {
                    if state_poll.vt_log_buffers.contains_key(&sid_poll) {
                        LoopAction::Poll
                    } else {
                        LoopAction::SessionGone
                    }
                }
                event = event_rx.recv() => {
                    let Ok(event) = event else { continue };
                    let is_relevant = match &event {
                        crate::state::AppEvent::PtyParsed { session_id: sid, .. }
                        | crate::state::AppEvent::PtyExit { session_id: sid }
                        | crate::state::AppEvent::SessionClosed { session_id: sid, .. } => sid == &sid_poll,
                        _ => false,
                    };
                    if is_relevant { LoopAction::Event } else { LoopAction::Skip }
                }
            };

            if matches!(action, LoopAction::SessionGone) {
                break;
            }
            if matches!(action, LoopAction::Skip) {
                continue;
            }

            // On Event: yield to let the session_state_accumulator task process
            // the same broadcast event before we read the state.  Without this,
            // we may read stale state (the accumulator hasn't applied the event
            // yet) → dedup sees no change → client misses the update.
            if matches!(action, LoopAction::Event) {
                tokio::task::yield_now().await;
            }

            // Single session_state_with_shell call per iteration, used by both arms
            if let Some(current) = state_poll.session_state_with_shell(&sid_poll)
                && prev_state.as_ref() != Some(&current)
            {
                let frame = serde_json::json!({"type": "state", "state": &current});
                prev_state = Some(current);
                if futures_util::SinkExt::send(
                    &mut ws_sender,
                    Message::Text(frame.to_string().into()),
                )
                .await
                .is_err()
                {
                    break;
                }
            }

            // Poll arm: also send log lines and screen content
            if matches!(action, LoopAction::Poll) {
                let Some(vt_log) = state_poll.vt_log_buffers.get(&sid_poll) else {
                    break;
                };
                let (lines, new_offset, screen_lines, input_line) = {
                    let buf = vt_log.lock();
                    let (l, o) = buf.lines_since_owned(offset, usize::MAX);
                    let trim = trim_screen_chrome(buf.screen_rows());
                    let styled = buf.screen_log_lines();
                    let trimmed_styled: Vec<_> = styled.into_iter().take(trim.cutoff).collect();
                    let il = buf.prompt_input_text();
                    (l, o, trimmed_styled, il)
                }; // lock released
                // Hash screen rows to detect changes (use plain text for hashing)
                use std::hash::{Hash, Hasher};
                let mut hasher = std::collections::hash_map::DefaultHasher::new();
                for sl in &screen_lines {
                    for span in &sl.spans {
                        span.text.hash(&mut hasher);
                    }
                }
                input_line.hash(&mut hasher);
                let screen_hash = hasher.finish();
                let screen_changed = screen_hash != prev_screen_hash && !screen_lines.is_empty();
                // Send frame if there are new log lines OR screen content changed
                if !lines.is_empty() || screen_changed {
                    // total_lines = post-read monotonic cursor (== offset when no new
                    // lines). The client tracks it for reconnect resume — see catch-up above.
                    let mut frame = serde_json::json!({"type": "log", "offset": offset, "total_lines": new_offset});
                    if !lines.is_empty() {
                        frame["lines"] = serde_json::json!(lines);
                    }
                    if screen_changed {
                        frame["screen"] = serde_json::json!(screen_lines);
                        if let Some(ref il) = input_line {
                            frame["input_line"] = serde_json::json!(il);
                        }
                        prev_screen_hash = screen_hash;
                    }
                    if futures_util::SinkExt::send(
                        &mut ws_sender,
                        Message::Text(frame.to_string().into()),
                    )
                    .await
                    .is_err()
                    {
                        break;
                    }
                    if !lines.is_empty() {
                        offset = new_offset;
                    }
                }
            }
        }
    });

    // Read messages from the client and write to PTY (input passthrough)
    while let Some(Ok(msg)) = ws_receiver.next().await {
        match msg {
            Message::Text(text) => {
                if let Some(session) = state.sessions.get(&session_id) {
                    let mut s = session.lock();
                    if let Err(e) = s.writer.write_all(text.as_bytes()) {
                        tracing::error!(session_id = %session_id, "PTY write failed: {e}");
                        break;
                    }
                    if let Err(e) = s.writer.flush() {
                        tracing::error!("PTY flush failed: {e}");
                        break;
                    }
                }
            }
            Message::Binary(data) => {
                if let Some(session) = state.sessions.get(&session_id) {
                    let mut s = session.lock();
                    if let Err(e) = s.writer.write_all(&data) {
                        tracing::error!(session_id = %session_id, "PTY write failed: {e}");
                        break;
                    }
                    if let Err(e) = s.writer.flush() {
                        tracing::error!("PTY flush failed: {e}");
                        break;
                    }
                }
            }
            Message::Close(_) => break,
            _ => {}
        }
    }

    send_task.abort();
}

/// Handle a WebSocket connection in grid mode (`?format=grid`).
///
/// Streams binary grid frames (same format as Tauri Channel) using the
/// `grid_watch` channel. Uses latest-frame-wins semantics: slow clients
/// automatically skip intermediate frames via `tokio::sync::watch`.
///
/// On connect, sends a full frame (all rows marked dirty). Subsequent frames
/// are delta-based (only changed rows). Client sends text messages for
/// commands (e.g. `{"type":"ack"}`) and binary messages for PTY input.
async fn handle_ws_grid_session(socket: WebSocket, session_id: String, state: Arc<AppState>) {
    let (mut ws_sender, mut ws_receiver) = socket.split();

    // Subscribe to the grid watch channel (latest-frame-wins for slow clients).
    let mut frame_rx = match state.grid_watch.get(&session_id) {
        Some(tx) => tx.subscribe(),
        None => {
            let _ = futures_util::SinkExt::send(&mut ws_sender, Message::Close(None)).await;
            return;
        }
    };

    // Send initial full frame so the client can render immediately.
    // Scope the MutexGuard so it's dropped before the .await.
    let initial_frame = state.vt_log_buffers.get(&session_id).and_then(|vt| {
        let mut vt = vt.lock();
        vt.grid_force_full_damage();
        let frame = vt.serialize_dirty_rows();
        if frame.is_empty() { None } else { Some(frame) }
    });
    if let Some(frame) = initial_frame
        && futures_util::SinkExt::send(&mut ws_sender, Message::Binary(frame.into()))
            .await
            .is_err()
    {
        return;
    }

    // Subscribe to event bus for parsed events (exit, closed, etc.)
    let mut event_rx = state.event_bus.subscribe();
    let sid_for_events = session_id.clone();

    let send_task = tokio::spawn(async move {
        loop {
            tokio::select! {
                result = frame_rx.changed() => {
                    if result.is_err() { break; } // sender dropped
                    let frame = frame_rx.borrow_and_update().clone();
                    if !frame.is_empty()
                        && futures_util::SinkExt::send(
                            &mut ws_sender,
                            Message::Binary(frame.into()),
                        )
                        .await
                        .is_err()
                    {
                        break;
                    }
                }
                result = event_rx.recv() => {
                    match result {
                        Ok(event) => {
                            let matches = match &event {
                                crate::state::AppEvent::PtyParsed { session_id: sid, .. } => sid == &sid_for_events,
                                crate::state::AppEvent::PtyExit { session_id: sid } => sid == &sid_for_events,
                                crate::state::AppEvent::SessionClosed { session_id: sid, .. } => sid == &sid_for_events,
                                _ => false,
                            };
                            if !matches { continue; }

                            let payload = match &event {
                                crate::state::AppEvent::PtyParsed { parsed, .. } => {
                                    serde_json::json!({"type": "parsed", "event": parsed})
                                }
                                crate::state::AppEvent::PtyExit { session_id: sid } => {
                                    serde_json::json!({"type": "exit", "session_id": sid})
                                }
                                crate::state::AppEvent::SessionClosed { session_id: sid, reason } => {
                                    serde_json::json!({"type": "closed", "session_id": sid, "reason": reason})
                                }
                                _ => continue,
                            };
                            if futures_util::SinkExt::send(
                                &mut ws_sender,
                                Message::Text(payload.to_string().into()),
                            ).await.is_err() {
                                break;
                            }
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                            tracing::warn!(session_id = %sid_for_events, lagged = n, "grid WS broadcast lagged");
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                    }
                }
            }
        }
    });

    // Read messages from the client
    let state_clone = state.clone();
    let sid = session_id.clone();
    while let Some(Ok(msg)) = ws_receiver.next().await {
        match msg {
            Message::Text(_text) => {
                // Reserved for client commands (e.g. resize, scroll).
                // No ACK needed — watch channel handles backpressure naturally.
            }
            Message::Binary(data) => {
                if let Some(session) = state_clone.sessions.get(&sid) {
                    let mut s = session.lock();
                    if let Err(e) = s.writer.write_all(&data) {
                        tracing::error!(session_id = %sid, "PTY write failed: {e}");
                        break;
                    }
                    if let Err(e) = s.writer.flush() {
                        tracing::error!("PTY flush failed: {e}");
                        break;
                    }
                }
            }
            Message::Close(_) => break,
            _ => {}
        }
    }

    send_task.abort();
}

/// Remove agent TUI chrome from screen rows (status bars, prompt lines,
/// separators) and trim trailing empty rows.
///
/// Scans from the bottom for two anchor patterns:
/// 1. Separator line (all box-drawing chars like `────`) — cuts from there
/// 2. Prompt line (`❯`, `>`) — cuts from there, extending up past separators
///
/// The scan window is 15 rows to accommodate Claude Code's full footer
/// (prompt + input area + separator + status bar = ~12 rows).
/// Result of trimming screen chrome: cleaned rows.
struct TrimResult {
    /// How many rows were kept (cutoff index). Allows applying the same trim to parallel data.
    cutoff: usize,
}

use crate::chrome::find_chrome_cutoff;

fn trim_screen_chrome(rows: Vec<String>) -> TrimResult {
    let refs: Vec<&str> = rows.iter().map(|s| s.as_str()).collect();
    let cutoff = find_chrome_cutoff(&refs).unwrap_or(rows.len());
    TrimResult { cutoff }
}

// --- Terminal grid HTTP endpoints ---
// These delegate to the same VtLogBuffer/TerminalGrid logic as the Tauri commands in pty.rs.

pub(super) async fn terminal_scroll(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<super::types::TerminalScrollRequest>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return session_not_found();
    };
    let frame = {
        let mut vt = vt.lock();
        vt.grid_scroll(body.delta);
        vt.serialize_dirty_rows()
    };
    crate::pty::send_grid_frame(&state, &session_id, frame);
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn terminal_scroll_to(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<super::types::TerminalScrollToRequest>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return session_not_found();
    };
    let frame = {
        let mut vt = vt.lock();
        vt.grid_scroll_to_line(body.line);
        vt.serialize_dirty_rows()
    };
    crate::pty::send_grid_frame(&state, &session_id, frame);
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

pub(super) async fn terminal_scroll_info(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let vt = vt.lock();
    let info = serde_json::json!({
        "display_offset": vt.grid_display_offset(),
        "total_lines": vt.grid_total_lines(),
        "screen_lines": vt.grid_screen_lines(),
    });
    Json(info).into_response()
}

pub(super) async fn terminal_search(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<super::types::TerminalSearchRequest>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let matches = vt.lock().grid_search(&body.query);
    Json(serde_json::json!({"matches": matches})).into_response()
}

pub(super) async fn terminal_search_buffer(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Json(body): Json<super::types::TerminalSearchRequest>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let matches = vt.lock().grid_search_buffer(&body.query);
    Json(serde_json::json!({"matches": matches})).into_response()
}

pub(super) async fn terminal_get_row_text(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Query(query): Query<super::types::TerminalRowQuery>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let text = vt.lock().grid_get_row_text(query.row);
    Json(serde_json::json!({"text": text})).into_response()
}

pub(super) async fn terminal_get_lines(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Query(query): Query<super::types::TerminalLinesQuery>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let lines = vt.lock().grid_get_lines(query.start, query.end);
    Json(serde_json::json!({"lines": lines})).into_response()
}

pub(super) async fn terminal_get_cursor_line(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let text = vt.lock().grid_get_cursor_line();
    Json(serde_json::json!({"text": text})).into_response()
}

pub(super) async fn terminal_hyperlink_at(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
    Query(query): Query<super::types::TerminalCellQuery>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "Session not found"})),
        )
            .into_response();
    };
    let url = vt.lock().grid_hyperlink_at(query.row, query.col);
    Json(serde_json::json!({"url": url})).into_response()
}

pub(super) async fn terminal_request_frame(
    State(state): State<Arc<AppState>>,
    Path(session_id): Path<String>,
) -> impl IntoResponse {
    let Some(vt) = state.vt_log_buffers.get(&session_id) else {
        return session_not_found();
    };
    let frame = {
        let mut vt = vt.lock();
        vt.grid_force_full_damage();
        vt.serialize_dirty_rows()
    };
    crate::pty::send_grid_frame(&state, &session_id, frame);
    (StatusCode::OK, Json(serde_json::json!({"ok": true})))
}

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

    // is_separator_line tests live in chrome.rs (canonical location)

    // --- trim_screen_chrome ---

    #[test]
    fn trim_removes_prompt_and_separator() {
        let rows: Vec<String> = vec![
            "content line 1".into(),
            "content line 2".into(),
            "────────────────────────────────────────".into(),
            "❯ ".into(),
            "────────────────────────────────────────".into(),
            "  [Opus 4.6 | Max] tuicommander git:(main)".into(),
            "  ⏵⏵ bypass permissions on".into(),
        ];
        let result = trim_screen_chrome(rows);
        assert_eq!(result.cutoff, 2);
    }

    #[test]
    fn trim_handles_decorated_separator_with_badge() {
        let rows: Vec<String> = vec![
            "some output".into(),
            "──────────────────────────────── pwa ──".into(),
            "❯ hello".into(),
            "──────────────────────────────── pwa ──".into(),
            "  status bar".into(),
        ];
        let result = trim_screen_chrome(rows);
        assert_eq!(result.cutoff, 1);
    }

    #[test]
    fn trim_no_chrome_keeps_all() {
        let rows: Vec<String> = vec!["line 1".into(), "line 2".into(), "line 3".into()];
        let result = trim_screen_chrome(rows.clone());
        assert_eq!(result.cutoff, 3);
    }

    #[test]
    fn trim_empty_input() {
        let result = trim_screen_chrome(vec![]);
        assert_eq!(result.cutoff, 0);
    }

    // --- WebSocket catch-up/subscribe race ---
    //
    // Regression guard for the ring-buffer catch-up race: the PTY writer and
    // the WS handler must serialize `ring.write` + `ws_clients.send` on both
    // sides via `ring.lock()`. If the two sides drift into separate critical
    // sections, bytes written during the window are delivered twice (once in
    // the catch-up snapshot, once through the live mpsc queue).
    //
    // This test reproduces the contract with tight concurrent loops and
    // asserts that every written byte is observed exactly once by a late
    // subscriber that reads its snapshot + drains its mpsc queue.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ws_catchup_no_duplicate_with_concurrent_writer() {
        use crate::state::OutputRingBuffer;
        use parking_lot::Mutex as PlMutex;
        use std::sync::Arc;
        use std::sync::atomic::{AtomicU64, Ordering};

        const CHUNK_COUNT: u64 = 5_000;
        const CHUNK_SIZE: usize = 8; // just the index, no filler
        const CAPACITY: usize = CHUNK_COUNT as usize * CHUNK_SIZE * 2;
        // Subscriber waits until the writer has committed this many chunks
        // before attaching — guarantees a non-empty snapshot without relying
        // on wall-clock sleep.
        const ATTACH_AFTER: u64 = 100;

        let ring: Arc<PlMutex<OutputRingBuffer>> =
            Arc::new(PlMutex::new(OutputRingBuffer::new(CAPACITY)));
        let clients: Arc<PlMutex<Vec<tokio::sync::mpsc::UnboundedSender<Vec<u8>>>>> =
            Arc::new(PlMutex::new(Vec::new()));
        let writer_progress: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
        let subscriber_attached: Arc<std::sync::atomic::AtomicBool> =
            Arc::new(std::sync::atomic::AtomicBool::new(false));

        // PTY writer: mirror the pty.rs critical section — ring.write + broadcast
        // under a single ring.lock().
        let writer_ring = ring.clone();
        let writer_clients = clients.clone();
        let writer_progress_w = writer_progress.clone();
        let subscriber_attached_w = subscriber_attached.clone();
        let writer = tokio::task::spawn_blocking(move || {
            for i in 0..CHUNK_COUNT {
                let payload = i.to_be_bytes().to_vec();

                // Mirror pty.rs: ring.write + broadcast under one ring.lock().
                let mut ring_guard = writer_ring.lock();
                ring_guard.write(&payload);
                {
                    let mut subs = writer_clients.lock();
                    subs.retain(|tx| tx.send(payload.clone()).is_ok());
                }
                drop(ring_guard);

                writer_progress_w.store(i + 1, Ordering::Release);

                // After ATTACH_AFTER chunks, wait for the subscriber to attach
                // before continuing — ensures the race window is exercised.
                if i + 1 == ATTACH_AFTER {
                    while !subscriber_attached_w.load(Ordering::Acquire) {
                        std::hint::spin_loop();
                    }
                }
            }
        });

        // Subscriber: attach mid-stream. Spins on the atomic counter instead
        // of sleeping, so the race window is exercised deterministically.
        let sub_ring = ring.clone();
        let sub_clients = clients.clone();
        let writer_progress_s = writer_progress.clone();
        let subscriber_attached_s = subscriber_attached.clone();
        let subscriber = tokio::task::spawn_blocking(move || {
            // Spin until the writer has made enough progress for a non-empty snapshot.
            while writer_progress_s.load(Ordering::Acquire) < ATTACH_AFTER {
                std::hint::spin_loop();
            }

            let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
            let (snapshot_bytes, snapshot_total) = {
                let r = sub_ring.lock();
                let snap = r.read_last(CAPACITY);
                sub_clients.lock().push(tx);
                snap
            };
            // Signal the writer that we're attached — it can resume writing.
            subscriber_attached_s.store(true, Ordering::Release);
            (snapshot_bytes, snapshot_total, rx)
        });

        let (snapshot_bytes, snapshot_total, mut rx) = subscriber.await.unwrap();
        writer.await.unwrap();

        // Drain remaining live frames.
        let mut live_bytes: Vec<u8> = Vec::new();
        while let Ok(chunk) = rx.try_recv() {
            live_bytes.extend_from_slice(&chunk);
        }

        // Reconstruct the chunk indices seen by the subscriber. Every chunk
        // is exactly CHUNK_SIZE bytes = one u64 BE index.
        let extract_indices = |bytes: &[u8]| -> Vec<u64> {
            bytes
                .chunks_exact(CHUNK_SIZE)
                .map(|c| u64::from_be_bytes(c.try_into().unwrap()))
                .collect()
        };

        let snapshot_indices = extract_indices(&snapshot_bytes);
        let live_indices = extract_indices(&live_bytes);

        // Precondition: both streams must be non-empty, otherwise the race
        // window was not exercised and the boundary invariants are vacuous.
        assert!(
            !snapshot_indices.is_empty(),
            "subscriber attached too late — no snapshot data"
        );
        assert!(
            !live_indices.is_empty(),
            "writer finished before subscriber — race not exercised"
        );

        // Invariants:
        // 1. Snapshot indices are strictly monotonically increasing by 1.
        for pair in snapshot_indices.windows(2) {
            assert_eq!(pair[1], pair[0] + 1, "snapshot not contiguous: {:?}", pair);
        }
        // 2. Live indices are strictly monotonically increasing by 1.
        for pair in live_indices.windows(2) {
            assert_eq!(pair[1], pair[0] + 1, "live not contiguous: {:?}", pair);
        }
        // 3. No overlap: the last snapshot index must be exactly one less
        //    than the first live index (no duplicate, no gap).
        let first_live = *live_indices
            .first()
            .expect("live_indices must be non-empty");
        let last_snap = *snapshot_indices
            .last()
            .expect("snapshot_indices must be non-empty");
        assert_eq!(
            first_live,
            last_snap + 1,
            "catch-up/live boundary wrong: last snapshot={last_snap}, first live={first_live}"
        );
        // 4. Every chunk written is accounted for exactly once.
        let total_seen = snapshot_indices.len() + live_indices.len();
        assert_eq!(
            total_seen as u64, CHUNK_COUNT,
            "expected all {CHUNK_COUNT} chunks, got {total_seen}"
        );
        // 5. total_written reported by the snapshot matches the number of
        //    bytes the writer had committed at snapshot time.
        assert!(snapshot_total <= CHUNK_COUNT * CHUNK_SIZE as u64);
        assert!(snapshot_total >= snapshot_indices.len() as u64 * CHUNK_SIZE as u64);
    }

    // --- Grid watch channel (format=grid WS endpoint) ---

    /// Verifies that the grid_watch channel delivers frames with latest-frame-wins
    /// semantics: a slow receiver that misses intermediate sends still gets the
    /// most recent frame on its next `changed().await`.
    #[tokio::test]
    async fn grid_watch_latest_frame_wins() {
        let (tx, mut rx) = tokio::sync::watch::channel(Vec::<u8>::new());

        // Send 3 frames without the receiver polling
        tx.send(vec![1, 2, 3]).unwrap();
        tx.send(vec![4, 5, 6]).unwrap();
        tx.send(vec![7, 8, 9]).unwrap();

        // Receiver sees only the latest
        rx.changed().await.unwrap();
        let frame = rx.borrow_and_update().clone();
        assert_eq!(frame, vec![7, 8, 9]);

        // No pending change after consuming latest
        let result = tokio::time::timeout(std::time::Duration::from_millis(10), rx.changed()).await;
        assert!(result.is_err(), "should timeout — no new frame");
    }

    /// Verifies that a newly subscribed receiver gets the current value
    /// immediately (supports initial full-frame delivery in handle_ws_grid_session).
    #[tokio::test]
    async fn grid_watch_subscriber_gets_current_value() {
        let (tx, _rx) = tokio::sync::watch::channel(Vec::<u8>::new());

        // Publish a frame
        tx.send(vec![10, 20, 30]).unwrap();

        // New subscriber sees current value via borrow()
        let rx2 = tx.subscribe();
        let current = rx2.borrow().clone();
        assert_eq!(current, vec![10, 20, 30]);
    }

    /// Verifies that spawn_pty_session registers a grid_watch channel for the session,
    /// so that handle_ws_grid_session can subscribe to it (regression for BUG-2).
    #[tokio::test]
    async fn spawn_pty_session_registers_grid_watch() {
        let state = super::super::tests::test_state();

        assert!(state.grid_watch.is_empty());

        let result = super::spawn_pty_session(
            state.clone(),
            std::env::var("SHELL").unwrap_or_else(|_| "/bin/sh".into()),
            None,
            24,
            80,
            None,
        );

        let session_id = match result {
            Ok(id) => id,
            Err(_) => return, // PTY unavailable in CI — skip gracefully
        };

        assert!(
            state.grid_watch.contains_key(&session_id),
            "spawn_pty_session must register a grid_watch channel"
        );

        // Verify the channel is functional
        let tx = state.grid_watch.get(&session_id).unwrap();
        let mut rx = tx.subscribe();
        tx.send(vec![1, 2, 3]).unwrap();
        rx.changed().await.unwrap();
        assert_eq!(*rx.borrow_and_update(), vec![1, 2, 3]);
    }
}
