use std::io::{self, IsTerminal};
use std::sync::atomic::{AtomicBool, Ordering};

use crossterm::event::{
    KeyboardEnhancementFlags, PopKeyboardEnhancementFlags, PushKeyboardEnhancementFlags,
};
use crossterm::execute;
use running_process::pty::NativePtyProcess;
use running_process::pty::PtySize;

use crate::console_title::OscTitleStripper;
use crate::dnd::{looks_like_dropped_path, normalize_dropped_path};
use crate::graphics::GraphicsConfig;
use crate::verbose_log;

#[path = "session_output.rs"]
mod session_output;
use session_output::{redraw_graphics_header_for_resize, run_output_writer};
#[path = "session_stdin.rs"]
mod session_stdin;
use session_stdin::{
    normalize_interactive_console_stdin_chunk, should_normalize_interactive_console_stdin,
    should_spawn_byte_stream_stdin_reader, stdin_chunk_requests_interrupt,
    stdin_source_is_real_stdin,
};

/// Resize the PTY. On Windows, `running_process::pty::NativePtyProcess::resize_impl`
/// is a deliberate no-op (see that crate's `pty/mod.rs:730-737`), so reaching
/// the underlying master's `resize()` directly is the only way to honor a
/// `SIGWINCH`/`Event::Resize`. On POSIX the library's implementation does the
/// right thing, so delegate. Issue #31, theory T2.
pub fn resize_pty(process: &NativePtyProcess, rows: u16, cols: u16) -> io::Result<()> {
    #[cfg(windows)]
    {
        let guard = process
            .handles
            .lock()
            .map_err(|e| io::Error::other(e.to_string()))?;
        if let Some(handles) = guard.as_ref() {
            handles
                .master
                .resize(PtySize {
                    rows,
                    cols,
                    pixel_width: 0,
                    pixel_height: 0,
                })
                .map_err(|e| io::Error::other(e.to_string()))?;
        }
        Ok(())
    }
    #[cfg(not(windows))]
    {
        // PtySize is used on Windows only; silence unused-import warnings.
        let _ = PtySize {
            rows,
            cols,
            pixel_width: 0,
            pixel_height: 0,
        };
        process
            .resize_impl(rows, cols)
            .map_err(|e| io::Error::other(e.to_string()))
    }
}

/// Counts of F3 events observed in a stream chunk.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct F3Events {
    /// Number of F3 press events seen. Repeats (autorepeat) are intentionally
    /// not counted as new presses — they indicate the key is still held.
    pub presses: u32,
    /// Number of F3 release events seen. Only fires on terminals that
    /// implement the kitty keyboard protocol with REPORT_EVENT_TYPES.
    pub releases: u32,
}

/// Byte-level observer that reports F3 press / release events seen in a
/// stream, without modifying the bytes. The raw pump forwards every byte
/// to the child verbatim and asks this observer how many F3 events flowed
/// past so it can call `InteractiveHooks::on_f3_press` / `on_f3_release`
/// once per event.
///
/// Three encodings are matched, covering the cross-platform terminal
/// matrix:
///
/// * Legacy SS3 form `\x1bOR` — emitted by Windows ConPTY and most POSIX
///   terminals without kitty keyboard protocol. Press-only.
/// * CSI tilde form `\x1b[13~` — emitted by xterm and most Linux consoles.
///   Press-only by default; with kitty REPORT_EVENT_TYPES enabled the
///   terminal extends it to `\x1b[13;1:3~` for release, `\x1b[13;1:2~`
///   for repeat.
/// * Kitty CSI u form `\x1b[13u` or `\x1b[57346u` (functional encoding) —
///   press-only by default; the `;mod:event-type` suffix carries
///   release/repeat the same way.
///
/// Issue #13 hold-to-record relies on the release branch. Terminals that
/// don't emit release events (notably ConPTY) fall back to the
/// VAD-silence auto-stop inside the voice module — see `voice.rs`.
///
/// The state machine survives across `observe` calls, so any of these
/// sequences split across reads (even one byte at a time) still fires
/// exactly once.
pub struct F3Observer {
    state: F3State,
    /// Parameter bytes accumulated between `\x1b[` and a CSI terminator.
    /// Capped at MAX_CSI_LEN to keep a runaway terminal from growing this
    /// unboundedly.
    csi_buf: Vec<u8>,
}

#[derive(Debug, Clone, Copy)]
enum F3State {
    Idle,
    Esc,
    /// Saw `\x1bO`; one more byte and we know if this is SS3-R (F3 press).
    Ss3,
    /// Saw `\x1b[`; accumulating parameter bytes until a CSI terminator.
    Csi,
}

/// Max parameter-byte payload a CSI sequence can have before we abandon
/// the match. A real F3 event tops out at ~16 bytes (`\x1b[57346;1:3u`),
/// 64 is generous and bounds memory if the terminal is misbehaving.
const MAX_CSI_LEN: usize = 64;

impl F3Observer {
    pub fn new() -> Self {
        Self {
            state: F3State::Idle,
            csi_buf: Vec::new(),
        }
    }

    /// Scan `chunk` and return the F3 events it contains. Updates internal
    /// state so subsequent calls see continuing matches.
    pub fn observe(&mut self, chunk: &[u8]) -> F3Events {
        let mut events = F3Events::default();
        for &b in chunk {
            match self.state {
                F3State::Idle => {
                    if b == 0x1b {
                        self.state = F3State::Esc;
                    }
                }
                F3State::Esc => match b {
                    b'O' => self.state = F3State::Ss3,
                    b'[' => {
                        self.state = F3State::Csi;
                        self.csi_buf.clear();
                    }
                    0x1b => {} // stay in Esc, a new sequence is starting
                    _ => self.state = F3State::Idle,
                },
                F3State::Ss3 => match b {
                    b'R' => {
                        // \x1bOR — F3 press in SS3 encoding.
                        events.presses += 1;
                        self.state = F3State::Idle;
                    }
                    0x1b => self.state = F3State::Esc,
                    _ => self.state = F3State::Idle,
                },
                F3State::Csi => {
                    if is_csi_terminator(b) {
                        if let Some(kind) = parse_f3_csi(&self.csi_buf, b) {
                            match kind {
                                F3Kind::Press => events.presses += 1,
                                F3Kind::Release => events.releases += 1,
                                // Repeat = key still held; deliberately silent.
                                F3Kind::Repeat => {}
                            }
                        }
                        self.state = F3State::Idle;
                        self.csi_buf.clear();
                    } else if b == 0x1b {
                        // Nested escape — abandon this CSI, start a new sequence.
                        self.state = F3State::Esc;
                        self.csi_buf.clear();
                    } else if self.csi_buf.len() < MAX_CSI_LEN {
                        self.csi_buf.push(b);
                    } else {
                        // Overrun — give up on this sequence.
                        self.state = F3State::Idle;
                        self.csi_buf.clear();
                    }
                }
            }
        }
        events
    }
}

impl Default for F3Observer {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum F3Kind {
    Press,
    Repeat,
    Release,
}

/// CSI terminator bytes per ECMA-48 (`0x40..=0x7E`, "Final Byte"). We
/// only care about a couple in practice (`u`, `~`) but accepting the
/// full range keeps misbehaving terminals from getting us stuck inside
/// `F3State::Csi`.
fn is_csi_terminator(b: u8) -> bool {
    matches!(b, 0x40..=0x7E)
}

/// Decide whether a parameter-bytes payload (e.g. `13;1:3`) plus a
/// terminator (e.g. `~` or `u`) is an F3 event, and which kind.
/// Returns `None` for anything that isn't F3 — different keycodes,
/// non-keyboard CSI sequences, malformed payloads.
fn parse_f3_csi(params: &[u8], terminator: u8) -> Option<F3Kind> {
    if terminator != b'u' && terminator != b'~' {
        return None;
    }
    let payload = std::str::from_utf8(params).ok()?;
    let mut parts = payload.split(';');
    let keycode_str = parts.next()?;

    // First param is the keycode. F3 keycodes:
    //   - `\x1b[13~` (CSI tilde, legacy F3 — see Linux/xterm function-key map)
    //   - `\x1b[13u` (CSI u, F3 with disambiguation but no functional encoding)
    //   - `\x1b[57346u` (CSI u, F3 with kitty functional encoding)
    let is_f3 = match terminator {
        b'~' => keycode_str == "13",
        b'u' => keycode_str == "13" || keycode_str == "57346",
        _ => false,
    };
    if !is_f3 {
        return None;
    }

    // Second param is `modifier[:event-type[:text]]`. Event-type defaults
    // to 1 (press) when omitted.
    let event_type = parts
        .next()
        .and_then(|modifier_field| modifier_field.split(':').nth(1))
        .and_then(|et| et.parse::<u32>().ok())
        .unwrap_or(1);

    match event_type {
        2 => Some(F3Kind::Repeat),
        3 => Some(F3Kind::Release),
        _ => Some(F3Kind::Press),
    }
}

/// Where to send synthetic input bytes generated by an `InteractiveHooks`
/// implementation (voice transcript, drag-drop paths, etc.).
///
/// Two impls ship in-tree:
/// * `NativePtyProcessSink` — wraps a `&NativePtyProcess` and forwards
///   to `write_impl`. Used by the direct local-PTY pump in `runner.rs`.
/// * A TCP-backed sink in `daemon::attach` — sends `WorkerClientMessage::Input`
///   frames to the daemon worker. Used by centralized-mode foreground
///   attach so voice + DnD reach the daemon-owned PTY just like keystrokes.
///
/// The `submit` flag mirrors `NativePtyProcess::write_impl`'s second
/// argument: `true` ends the input with a synthetic Enter so the agent
/// processes the buffer, `false` leaves the cursor mid-line so a human
/// can edit before submitting.
pub trait PtyInputSink {
    fn write_input(&mut self, bytes: &[u8], submit: bool) -> io::Result<()>;
}

/// Adapter: a `PtyInputSink` over a local `NativePtyProcess`.
pub struct NativePtyProcessSink<'a> {
    process: &'a NativePtyProcess,
}

impl<'a> NativePtyProcessSink<'a> {
    pub fn new(process: &'a NativePtyProcess) -> Self {
        Self { process }
    }
}

impl<'a> PtyInputSink for NativePtyProcessSink<'a> {
    fn write_input(&mut self, bytes: &[u8], submit: bool) -> io::Result<()> {
        self.process
            .write_impl(bytes, submit)
            .map_err(|err| io::Error::other(err.to_string()))
    }
}

pub trait InteractiveHooks {
    fn intercept_f3(&self) -> bool {
        false
    }

    fn on_f3_press(&mut self, _sink: &mut dyn PtyInputSink) -> io::Result<()> {
        Ok(())
    }

    fn on_f3_release(&mut self, _sink: &mut dyn PtyInputSink) -> io::Result<()> {
        Ok(())
    }

    fn on_tick(&mut self, _sink: &mut dyn PtyInputSink) -> io::Result<()> {
        Ok(())
    }
}

/// Enable raw mode and keyboard-enhancement flags on the current
/// terminal, returning a guard that restores the original state on drop.
/// Only useful when stdin is an actual TTY; see `enter_raw_mode_if_tty`.
#[derive(Debug)]
pub struct RawTerminalGuard {
    enhancement_flags_pushed: bool,
}

/// Public factory for a `RawTerminalGuard` returning `None` when stdin
/// is piped (no point putting a pipe into raw mode). `run_plan_pty` in
/// main.rs owns the guard for the duration of the pump call so the
/// terminal is restored even if the pump panics.
pub fn enter_raw_mode_if_tty() -> Option<RawTerminalGuard> {
    if terminals_are_interactive() {
        RawTerminalGuard::enter().ok()
    } else {
        None
    }
}

impl RawTerminalGuard {
    pub fn enter() -> io::Result<Self> {
        crossterm::terminal::enable_raw_mode()?;

        let mut stdout = io::stdout();
        let flags = KeyboardEnhancementFlags::REPORT_EVENT_TYPES
            | KeyboardEnhancementFlags::DISAMBIGUATE_ESCAPE_CODES;
        let enhancement_flags_pushed =
            execute!(stdout, PushKeyboardEnhancementFlags(flags)).is_ok();

        Ok(Self {
            enhancement_flags_pushed,
        })
    }
}

impl Drop for RawTerminalGuard {
    fn drop(&mut self) {
        let _ = if self.enhancement_flags_pushed {
            execute!(io::stdout(), PopKeyboardEnhancementFlags)
        } else {
            Ok(())
        };
        let _ = crossterm::terminal::disable_raw_mode();
    }
}

pub fn terminals_are_interactive() -> bool {
    io::stdin().is_terminal() && io::stdout().is_terminal()
}

/// Raw-byte pump replacing the crossterm event loop on the PTY path.
///
/// Bytes from `stdin_source` flow into the child's PTY via
/// `write_impl`. An `F3Observer` watches the byte stream and — when
/// `hooks.intercept_f3()` is true — counts `\x1bOR` sequences, firing
/// `on_f3_press` once per observed press. The bytes are NOT consumed:
/// the child still receives `\x1bOR` and can handle F3 itself if it
/// wants to. `on_tick` runs every loop iteration regardless of stdin
/// activity so hooks that poll background state (e.g. voice transcripts)
/// still make progress during idle.
///
/// For interactive Windows console input only, the reader normalizes BS
/// (`0x08`) to DEL (`0x7f`) before forwarding. That keeps Backspace aligned
/// with xterm-style TUI expectations even when the console does not emit VT
/// input bytes despite PTY mode requesting them. Non-interactive sources used
/// by tests and pipes are still forwarded unchanged.
///
/// This replaces the old `run_interactive_pty_session` /
/// `run_pty_output_loop` split. The event-loop approach parsed stdin
/// through crossterm's `event::read` — a lossy demultiplexer that dropped
/// every escape sequence it didn't recognize (DSR replies, DA, XTWINOPS,
/// OSC color queries, etc.), which hung child TUIs like codex Ink that
/// write those queries on startup and wait for a reply. See **PR #47**,
/// which made this replacement — not issue #46, which despite being named in
/// that PR's title is "CI: macos-15-intel integration test can't locate
/// mock-agent" and concluded it was not a PTY regression.
///
/// Current scope: stdin forwarding + F3 observation + hook ticks +
/// Ctrl+C + child-exit detection. Resize handling (SIGWINCH on Unix,
/// polled `terminal_size::terminal_size()` on Windows).
///
/// Ctrl-C flow: raw mode turns the keyboard chord into byte `0x03` instead
/// of a terminal signal. For interactive stdin, the pump forwards that byte
/// once, then escalates via `interrupt_pty_process`. On POSIX this sends
/// SIGINT to the child's pgroup and waits up to 2s for exit. On Windows the
/// escalation closes the PTY directly (no extra 0x03 write), because the
/// underlying `send_interrupt_impl` duplicates the byte that stdin already
/// forwarded and makes Ink-based TUIs see a single press as
/// "Ctrl-C twice = exit". The external `interrupted` flag is still honored
/// for non-keyboard interrupts such as OS signals or tests.
pub fn run_raw_pty_pump<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    run_raw_pty_pump_with_extra_rx(process, interrupted, hooks, stdin_source, None)
}

/// Production pump entry that wires a side channel for IDropTarget
/// callbacks (issue #79). Constructs the platform-native resize watcher
/// internally and passes through to `run_raw_pty_pump_full`.
///
/// `extra_rx` chunks are interleaved with stdin chunks and forwarded to
/// the PTY exactly like real stdin, EXCEPT they bypass the
/// bracketed-paste normalizer (the OLE/IDropTarget callback already
/// hands us a normalized, newline-joined path).
pub fn run_raw_pty_pump_with_extra_rx<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    run_raw_pty_pump_with_extra_rx_verbose(
        process,
        interrupted,
        hooks,
        stdin_source,
        extra_rx,
        false,
    )
}

/// Production pump entry with optional clud-level diagnostics.
pub fn run_raw_pty_pump_with_extra_rx_verbose<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
    verbose: bool,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    run_raw_pty_pump_with_extra_rx_verbose_and_graphics(
        process,
        interrupted,
        hooks,
        stdin_source,
        extra_rx,
        verbose,
        None,
    )
}

/// Production pump entry with optional clud-level diagnostics and a PTY
/// header renderer that can redraw/reserve rows after terminal resizes.
pub fn run_raw_pty_pump_with_extra_rx_verbose_and_graphics<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
    verbose: bool,
    graphics: Option<GraphicsConfig>,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    let (resize_tx, resize_rx) = std::sync::mpsc::channel::<(u16, u16)>();
    spawn_os_resize_watcher(resize_tx);
    run_raw_pty_pump_full_verbose(
        process,
        interrupted,
        hooks,
        stdin_source,
        resize_rx,
        extra_rx,
        PumpOptions { verbose, graphics },
    )
}

/// Spawn the platform-native resize-watcher thread that feeds
/// `resize_tx` with `(rows, cols)` whenever the user resizes their
/// terminal window.
///
/// Unix: a SIGWINCH signal handler via `signal-hook`. Zero-latency;
/// the kernel delivers the signal the moment the terminal resizes.
///
/// Windows: 150 ms polling of `crossterm::terminal::size()`. The
/// zero-latency option would be `ReadConsoleInputW` filtering for
/// `WINDOW_BUFFER_SIZE_EVENT`, but that consumes events from the
/// shared console input buffer and races with our stdin reader for
/// keystrokes. Polling avoids the race at the cost of up to 150 ms
/// redraw lag — imperceptible in practice. See the plan file for the
/// deferred zero-latency variant.
fn spawn_os_resize_watcher(resize_tx: std::sync::mpsc::Sender<(u16, u16)>) {
    #[cfg(unix)]
    {
        use signal_hook::consts::signal::SIGWINCH;
        use signal_hook::iterator::Signals;
        std::thread::spawn(move || {
            let Ok(mut signals) = Signals::new([SIGWINCH]) else {
                return;
            };
            for _ in signals.forever() {
                if let Ok((cols, rows)) = crossterm::terminal::size() {
                    if resize_tx.send((rows, cols)).is_err() {
                        break; // pump exited
                    }
                }
            }
        });
    }
    #[cfg(windows)]
    {
        std::thread::spawn(move || {
            let mut last: Option<(u16, u16)> = None;
            loop {
                let Ok((cols, rows)) = crossterm::terminal::size() else {
                    break;
                };
                let now = (rows, cols);
                if Some(now) != last {
                    if resize_tx.send(now).is_err() {
                        break; // pump exited
                    }
                    last = Some(now);
                }
                std::thread::sleep(std::time::Duration::from_millis(150));
            }
        });
    }
}

/// Inner pump entry that takes an explicit resize receiver. Tests use this
/// to inject synthetic resize events without involving platform signal
/// machinery; production wrappers (`run_raw_pty_pump`) construct the
/// channel and spawn a platform-native resize producer thread.
pub fn run_raw_pty_pump_with_resize_rx<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    resize_rx: std::sync::mpsc::Receiver<(u16, u16)>,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    run_raw_pty_pump_full(process, interrupted, hooks, stdin_source, resize_rx, None)
}

/// Most-general pump entry. See `run_raw_pty_pump_with_extra_rx` for
/// the public-facing version that constructs the resize receiver.
pub fn run_raw_pty_pump_full<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    resize_rx: std::sync::mpsc::Receiver<(u16, u16)>,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    run_raw_pty_pump_full_verbose(
        process,
        interrupted,
        hooks,
        stdin_source,
        resize_rx,
        extra_rx,
        PumpOptions::default(),
    )
}

/// Test-only seam: identical to [`run_raw_pty_pump_full`] but takes the
/// destination writer for filtered child output instead of hardcoding
/// `io::stdout()`. Lets integration tests inject a slow/counting sink
/// to prove the writer-thread decoupling from issue #538 (stdout flush
/// no longer gates stdin forwarding, output coalesces into O(1)
/// flushes). Not part of the stable public API — production code
/// always goes through the `io::stdout()`-bound `run_raw_pty_pump*`
/// family above.
#[doc(hidden)]
#[allow(clippy::too_many_arguments)]
pub fn run_raw_pty_pump_full_with_writer_for_test<H, R, W>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    resize_rx: std::sync::mpsc::Receiver<(u16, u16)>,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
    writer: W,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
    W: std::io::Write + Send,
{
    run_raw_pty_pump_full_verbose_with_writer(
        process,
        interrupted,
        hooks,
        stdin_source,
        resize_rx,
        extra_rx,
        PumpOptions::default(),
        writer,
    )
}

#[derive(Default)]
struct PumpOptions {
    verbose: bool,
    graphics: Option<GraphicsConfig>,
}

/// Idle poll cadence for the main loop's stdin wait when nothing is
/// pending. `stdin_rx.recv_timeout` wakes immediately once a chunk is
/// sent (mpsc's receiver is condvar-backed) — this bound only governs
/// how often we re-check resize / hooks / exit / interrupt during
/// genuine silence. It replaces the old design where stdin forwarding
/// shared the same 10ms wait as the (now separate-thread) output
/// reader — see issue #538.
const STDIN_IDLE_POLL: std::time::Duration = std::time::Duration::from_millis(5);

/// Blocking read timeout used by the dedicated output-reader thread
/// (issue #538). This thread never gates stdin forwarding — it feeds a
/// dedicated writer thread over an unbounded channel, so a stalled
/// terminal write can never delay it. The value only affects shutdown
/// responsiveness (how quickly the reader notices `stop_reader`) and
/// idle CPU usage.
const OUTPUT_READER_POLL_SECS: f64 = 0.05;

fn run_raw_pty_pump_full_verbose<H, R>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    resize_rx: std::sync::mpsc::Receiver<(u16, u16)>,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
    options: PumpOptions,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
{
    run_raw_pty_pump_full_verbose_with_writer(
        process,
        interrupted,
        hooks,
        stdin_source,
        resize_rx,
        extra_rx,
        options,
        io::stdout(),
    )
}

/// Drains `rx`, coalescing every chunk already queued into one
/// `write_all` + one `flush` per wakeup, until the channel disconnects
/// (draining and flushing anything left one last time before
/// returning). Shared by the pump's dedicated writer thread and by
/// unit tests that exercise the coalescing behavior in isolation,
/// without spinning up a full PTY (issue #538).
/// Same as `run_raw_pty_pump_full_verbose`, but takes the destination
/// writer for filtered child output as a parameter instead of
/// hardcoding `io::stdout()`. Production callers go through the
/// `io::stdout()`-bound wrapper above; tests inject a slow/counting
/// sink to exercise the writer-thread decoupling (issue #538).
///
/// Architecture (issue #538): output reading/filtering/writing used to
/// happen inline in the same loop turn that forwards stdin, so a slow
/// terminal `flush()` delayed keystroke delivery, each chunk got its
/// own write+flush (no coalescing), and the loop's cadence for
/// checking stdin was floored at the output read's 10ms timeout. Now:
///
/// * A dedicated **reader thread** calls `read_chunk_impl` in a loop,
///   OSC-strips each chunk, coalesces any additional chunks already
///   queued (non-blocking `Some(0.0)` drain) into one buffer, and
///   sends it over an *unbounded* channel — `send` never blocks, so a
///   slow writer can never stall this thread.
/// * A dedicated **writer thread** (`run_output_writer`) blocks on
///   that channel, drains everything else pending, and does exactly
///   one `write_all` + one `flush` per wakeup — turning a burst of N
///   chunks into O(1) syscalls.
/// * The **main thread** (this function body) only handles resize,
///   `extra_rx`, stdin forwarding, hook ticks, and exit/interrupt
///   detection. It blocks on `stdin_rx.recv_timeout(STDIN_IDLE_POLL)`
///   instead of the old output-read timeout, so keystroke forwarding
///   wakes as soon as a chunk is queued rather than waiting out a
///   fixed poll window.
///
/// All three "threads" share `process: &NativePtyProcess` (its
/// handles/reader state are internally `Mutex`-guarded and already
/// designed for concurrent access — see `daemon/worker.rs`'s reader +
/// writer thread pair for the established precedent). `thread::scope`
/// is used instead of `thread::spawn` because `process` is a borrow,
/// not an owned `Arc`; the scope block joins both worker threads
/// before returning, which is also how "flush remaining chunks on
/// exit" is guaranteed — the writer only stops once its channel
/// disconnects, i.e. after the reader has stopped and sent everything
/// it read.
#[allow(clippy::too_many_arguments)]
fn run_raw_pty_pump_full_verbose_with_writer<H, R, W>(
    process: &NativePtyProcess,
    interrupted: &AtomicBool,
    hooks: &mut H,
    stdin_source: R,
    resize_rx: std::sync::mpsc::Receiver<(u16, u16)>,
    extra_rx: Option<std::sync::mpsc::Receiver<Vec<u8>>>,
    options: PumpOptions,
    writer: W,
) -> i32
where
    H: InteractiveHooks,
    R: std::io::Read + Send + 'static,
    W: std::io::Write + Send,
{
    use std::sync::mpsc;

    let (stdin_tx, stdin_rx) = mpsc::channel::<Vec<u8>>();
    let interactive_real_stdin = stdin_source_is_real_stdin::<R>() && terminals_are_interactive();
    let normalize_console_stdin =
        should_normalize_interactive_console_stdin(interactive_real_stdin);
    let interrupt_on_ctrl_c_byte = interactive_real_stdin;
    // Issue #188: when an `extra_rx` is wired and we're on Windows with
    // an interactive real-stdin console, the `console_input`
    // `ReadConsoleInputW` reader is feeding that channel and is the
    // authoritative source of console bytes (including Shift+Enter →
    // `\n`). Spawning the byte-stream stdin reader below would race
    // with it on the same STDIN console queue — `ReadFile` strips
    // modifier state, so a stolen Shift+Enter surfaces as `\r`. Skip
    // it in that exact case so `console_input` is the sole consumer.
    let spawn_byte_stream_stdin_reader =
        should_spawn_byte_stream_stdin_reader(interactive_real_stdin, extra_rx.is_some());
    if options.verbose {
        verbose_log::log(format_args!(
            "[clud] pty pump: start interactive_stdin={} normalize_console_stdin={} \
             spawn_byte_stream_stdin_reader={}",
            interactive_real_stdin, normalize_console_stdin, spawn_byte_stream_stdin_reader
        ));
    }

    // Detached reader: pumps `stdin_source` → channel until EOF or error.
    // Detached (not joined) so a blocked `read()` on real stdin doesn't
    // wedge shutdown when the child exits — the process is terminating
    // anyway. See Step 12.
    if spawn_byte_stream_stdin_reader {
        std::thread::spawn(move || {
            let mut reader = stdin_source;
            let mut buf = [0u8; 4096];
            loop {
                match reader.read(&mut buf) {
                    Ok(0) => break, // EOF
                    Ok(n) => {
                        let mut chunk = buf[..n].to_vec();
                        if normalize_console_stdin {
                            normalize_interactive_console_stdin_chunk(&mut chunk);
                        }
                        if stdin_tx.send(chunk).is_err() {
                            break; // Main thread dropped the receiver → exit.
                        }
                    }
                    Err(_) => break,
                }
            }
        });
    } else {
        // `console_input` is the sole consumer via `extra_rx`. Drop the
        // unused stdin source and channel sender so the corresponding
        // `stdin_rx.try_recv()` in the main loop returns `Empty`
        // immediately (no thread will ever send on it).
        drop(stdin_source);
        drop(stdin_tx);
    }

    let mut observer = F3Observer::new();
    // Issue #63 / #79: bracketed-paste passes through the PTY pump as
    // raw bytes. When the user drags a file onto the terminal, the
    // terminal emits `\x1b[200~ <path-shaped string> \x1b[201~`. We
    // normalize that path BEFORE forwarding so all backends see a
    // canonical form, regardless of which terminal produced the drop.
    let mut paste = BracketedPasteNormalizer::new();

    // Issue #538: output reading/filtering/writing now happens on two
    // dedicated threads (reader + writer) instead of inline in this
    // loop, so a slow terminal `flush()` can never delay stdin
    // forwarding below. `thread::scope` borrows `process` safely (it's
    // `&NativePtyProcess`, not an owned `Arc`) and joins both threads
    // before returning — which is also what guarantees the writer
    // flushes any remaining chunks before the pump exits.
    let stop_reader = AtomicBool::new(false);
    let reader_closed = AtomicBool::new(false);

    std::thread::scope(|scope| {
        let (output_tx, output_rx) = mpsc::channel::<Vec<u8>>();
        let stop_reader = &stop_reader;
        let reader_closed = &reader_closed;

        // Writer thread: coalesces every chunk already queued into one
        // write_all + one flush per wakeup (see `run_output_writer`).
        // Exits once `output_tx` is dropped AND the queue is drained —
        // i.e. after the reader thread has stopped and sent everything
        // it had. That ordering is the "flush remaining chunks first"
        // shutdown guarantee.
        scope.spawn(move || {
            run_output_writer(output_rx, writer);
        });

        // Reader thread: never blocks the writer or the main loop.
        // `output_tx.send` on an unbounded channel never blocks, so a
        // stalled terminal write on the writer thread cannot delay
        // this thread from continuing to read, and cannot delay the
        // main thread's stdin forwarding (which no longer touches
        // output at all). Strip OSC 0/2 (window-title) sequences from
        // the child's output before they reach the terminal — the
        // library also keeps an un-stripped copy in its own `chunks`
        // queue (used by callers like capture/replay), but only the
        // bytes sent here actually reach the user's terminal. `main.rs`
        // set `process.set_echo(false)` so the library's built-in
        // stdout writer is silent — we own forwarding now.
        scope.spawn(move || {
            let mut osc_strip = OscTitleStripper::new();
            loop {
                if stop_reader.load(Ordering::Acquire) {
                    // Final non-blocking drain so a chunk that arrived
                    // right before shutdown isn't lost.
                    while let Ok(Some(chunk)) = process.read_chunk_impl(Some(0.0)) {
                        let filtered = osc_strip.process(&chunk);
                        if !filtered.is_empty() {
                            let _ = output_tx.send(filtered);
                        }
                    }
                    break;
                }
                match process.read_chunk_impl(Some(OUTPUT_READER_POLL_SECS)) {
                    Ok(Some(chunk)) => {
                        let mut filtered = osc_strip.process(&chunk);
                        // Coalesce clud-side: drain whatever else is
                        // already queued without blocking, so a
                        // chatty child's burst becomes one send (and,
                        // downstream, one write+flush) instead of one
                        // per chunk.
                        while let Ok(Some(more)) = process.read_chunk_impl(Some(0.0)) {
                            filtered.extend_from_slice(&osc_strip.process(&more));
                        }
                        if !filtered.is_empty() {
                            let _ = output_tx.send(filtered);
                        }
                    }
                    Ok(None) => {}
                    Err(_) => {
                        reader_closed.store(true, Ordering::Release);
                        break;
                    }
                }
            }
        });

        let exit_code = loop {
            // Drain resize events — always before stdin so a
            // late-arriving resize doesn't wait on a chunk of typing
            // to unblock the loop.
            while let Ok((rows, cols)) = resize_rx.try_recv() {
                let pty_rows = options
                    .graphics
                    .as_ref()
                    .map(|config| {
                        redraw_graphics_header_for_resize(config, rows, cols, options.verbose)
                    })
                    .unwrap_or(rows);
                if let Err(err) = resize_pty(process, pty_rows, cols) {
                    eprintln!("[clud] warning: failed to resize pty: {}", err);
                }
            }

            // Drain one chunk from the side channel (drag-drop OLE
            // callback) — these are pre-normalized path bytes that
            // should bypass the bracketed-paste detector and go
            // straight to the PTY.
            //
            // The 0x03 byte check is required on Windows: when the
            // `console_input` reader (issue #141 / PR #144) is active,
            // it turns off `ENABLE_PROCESSED_INPUT` so the OS no
            // longer fires a `CTRL_C_EVENT` for Ctrl-C. The press
            // arrives instead as a KEY_EVENT whose translated 0x03
            // byte is delivered via this channel — without the check,
            // clud forwards it to the child but never observes the
            // interrupt itself.
            if let Some(ref rx) = extra_rx {
                if let Ok(chunk) = rx.try_recv() {
                    // Unlike stdin_rx, extra_rx is by construction
                    // always user-driven (keyboard via
                    // console_input_rx on Windows, or OLE drag-drop
                    // callback) — never a piped test fixture — so we
                    // don't need the `interrupt_on_ctrl_c_byte` gate
                    // that skips 0x03 detection on piped stdin.
                    let requested_interrupt = stdin_chunk_requests_interrupt(&chunk);
                    if let Err(err) = process.write_impl(&chunk, false) {
                        eprintln!(
                            "[clud] warning: failed to forward dropped paths to pty: {}",
                            err
                        );
                    }
                    if requested_interrupt {
                        if options.verbose {
                            verbose_log::log("[clud] pty pump: interrupt via extra_rx Ctrl+C byte");
                        }
                        break interrupt_pty_process(process, options.verbose);
                    }
                }
            }

            // Block on stdin with a short idle timeout instead of the
            // old per-iteration output-read wait. `recv_timeout` wakes
            // as soon as a chunk is sent (mpsc's internal condvar), so
            // a keystroke arriving mid-wait is forwarded immediately
            // rather than after the wait window elapses — this is
            // what removes the ~10ms idle floor (issue #538).
            match stdin_rx.recv_timeout(STDIN_IDLE_POLL) {
                Ok(chunk) => {
                    let requested_interrupt =
                        interrupt_on_ctrl_c_byte && stdin_chunk_requests_interrupt(&chunk);
                    let chunk = if interactive_real_stdin {
                        crate::paste_image::expand_ctrl_v_bytes(&chunk, || {
                            crate::paste_image::handle_clipboard().ok().flatten()
                        })
                    } else {
                        std::borrow::Cow::Borrowed(chunk.as_slice())
                    };
                    // Run the bracketed-paste normalizer over the
                    // chunk BEFORE forwarding to the PTY. Non-paste
                    // bytes pass through with O(1) state cost (just a
                    // 6-byte prefix matcher); paste bodies are
                    // buffered and rewritten in place.
                    let outgoing = paste.process(chunk.as_ref());
                    if let Err(err) = process.write_impl(&outgoing, false) {
                        eprintln!("[clud] warning: failed to forward stdin to pty: {}", err);
                    } else if hooks.intercept_f3() {
                        // F3 detection runs over the outgoing user
                        // input after Ctrl+V image expansion but
                        // before any backend write side effects. A
                        // press inside paste text is unusual but
                        // should keep detection symmetry with raw
                        // forwarding.
                        let events = observer.observe(chunk.as_ref());
                        let mut sink = NativePtyProcessSink::new(process);
                        for _ in 0..events.presses {
                            if let Err(err) = hooks.on_f3_press(&mut sink) {
                                eprintln!("[clud] warning: voice F3 press hook failed: {}", err);
                            }
                        }
                        for _ in 0..events.releases {
                            if let Err(err) = hooks.on_f3_release(&mut sink) {
                                eprintln!("[clud] warning: voice F3 release hook failed: {}", err);
                            }
                        }
                    }

                    if requested_interrupt || interrupted.load(Ordering::SeqCst) {
                        if options.verbose {
                            let source = if requested_interrupt {
                                "stdin Ctrl+C byte"
                            } else {
                                "interrupt flag"
                            };
                            verbose_log::log(format_args!(
                                "[clud] pty pump: interrupt via {source}"
                            ));
                        }
                        break interrupt_pty_process(process, options.verbose);
                    }
                }
                Err(mpsc::RecvTimeoutError::Timeout) => {}
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    // No stdin producer (EOF already hit, or the
                    // byte-stream reader was never spawned because
                    // `console_input`/OLE `extra_rx` is the sole input
                    // source on this platform config — see
                    // `should_spawn_byte_stream_stdin_reader`).
                    // `recv_timeout` returns instantly on a
                    // disconnected channel instead of honoring the
                    // timeout, so without this sleep the loop would
                    // busy-spin a full CPU core instead of idling.
                    std::thread::sleep(STDIN_IDLE_POLL);
                }
            }

            {
                let mut sink = NativePtyProcessSink::new(process);
                if let Err(err) = hooks.on_tick(&mut sink) {
                    eprintln!("[clud] warning: interactive hook tick failed: {}", err);
                }
            }

            if reader_closed.load(Ordering::Acquire) {
                if options.verbose {
                    verbose_log::log("[clud] pty pump: output reader observed pty closed");
                }
                break reap_pty_exit(process);
            }

            if let Ok(Some(code)) =
                running_process::pty::poll_pty_process(&process.handles, &process.returncode)
            {
                if options.verbose {
                    verbose_log::log(format_args!("[clud] pty pump: child exited code {code}"));
                }
                break code;
            }

            if interrupted.load(Ordering::SeqCst) {
                if options.verbose {
                    verbose_log::log("[clud] pty pump: interrupt flag observed");
                }
                break interrupt_pty_process(process, options.verbose);
            }
        };

        stop_reader.store(true, Ordering::Release);
        exit_code
    })
}

/// `extra_rx` already wired, the `console_input::ReadConsoleInputW`
/// worker is the authoritative source of console bytes — including the
/// modifier-aware Shift+Enter → `\n` translation. Spawning the
/// byte-stream reader in that case would race with `ReadConsoleInputW`
/// on the same STDIN console queue. `ReadFile` strips modifier state
/// before producing bytes, so a stolen Shift+Enter surfaces as `\r`
/// instead of `\n`. Returning `false` keeps `console_input` as the
/// sole consumer.
///
/// Every other configuration — POSIX, piped stdin, no `extra_rx` —
/// keeps the byte-stream reader so existing behavior (including
/// `echo "prompt" | clud` and POSIX interactive use) is unchanged.
mod bracketed_paste;

pub use bracketed_paste::BracketedPasteNormalizer;
#[cfg(test)]
use bracketed_paste::{PASTE_END, PASTE_START};

mod interrupt;

use interrupt::{interrupt_pty_process, reap_pty_exit};

#[cfg(test)]
#[path = "session_tests.rs"]
mod tests;
