xianyun_executor/exec/

streamed.rs

//! Streaming command execution (TTY scrolling area / prefixed lines /
//! logger fallback).
//!
//! Ported from `internal/executor/exec_streamed.go`. Uses the same
//! `tokio::process::Command` + background-runtime bridge as `exec::run`
//! (see `exec::runtime`). `chooseStreamer`'s three-way fan-out is ported as
//! `choose_streamer`/`Streamer`: a scrolling area (TTY +
//! `use_scrolling_area`), prefixed lines (TTY, no scrolling area), or the
//! `tracing`-based logger fallback (no TTY) used by every existing test in
//! this module (stderr isn't a terminal under `cargo test`).
//!
//! Go renders with `pterm` (ANSI styling) directly. This port uses
//! `anstyle` for the SGR codes and writes through `anstream`'s
//! auto-detecting stream wrapper, which strips styling when stdout/stderr
//! isn't a terminal or `NO_COLOR` is set -- matching this project's
//! terminal-accessibility requirements without each call site needing its
//! own `NO_COLOR` check.

use std::process::Stdio;
use std::time::Duration;

use anstyle::{AnsiColor, Style};
use tokio::io::AsyncBufReadExt;
use tokio::process::Command as TokioCommand;
use tokio::sync::mpsc;

use super::{ExecError, apply_env, runtime};

/// Ported from `ExecStreamedOpts`.
///
/// `timeout` has no Go counterpart on the struct itself -- see
/// `Options::timeout`'s doc comment in `exec::mod` for why this port
/// threads a deadline explicitly instead of via an ambient context.
#[derive(Debug, Clone, Default)]
pub struct ExecStreamedOpts {
    pub working_dir: Option<String>,
    pub use_scrolling_area: bool,
    pub process_descriptive_name: Option<String>,
    pub env: Option<Vec<String>>,
    pub timeout: Option<Duration>,
}

/// Which output stream a [`StreamLine`] came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StreamSource {
    Stdout,
    Stderr,
}

/// One line read from the child's stdout or stderr, forwarded to whichever
/// `Streamer` is rendering this command's output.
struct StreamLine {
    source: StreamSource,
    text: String,
}

/// Reads `reader` line by line, forwarding each to `tx` tagged with
/// `source`. Mirrors Go's `command.Status()` polling loop collecting new
/// lines, but delivers each line as it arrives instead of rescanning a
/// growing buffered slice every 100ms.
async fn stream_lines<R>(reader: R, source: StreamSource, tx: mpsc::UnboundedSender<StreamLine>)
where
    R: tokio::io::AsyncRead + Unpin,
{
    let mut lines = tokio::io::BufReader::new(reader).lines();
    while let Ok(Some(text)) = lines.next_line().await {
        if tx.send(StreamLine { source, text }).is_err() {
            return;
        }
    }
}

/// Ported from `chooseStreamer`'s three cases.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Streamer {
    /// Ported from `limitedAreaStreamer`.
    LimitedArea,
    /// Ported from `prefixedStreamer`.
    Prefixed,
    /// Ported from `logStreamer`.
    Log,
}

/// Ported from `chooseStreamer`. Takes `is_tty` explicitly (rather than
/// calling [`is_stderr_tty`] itself) so the TTY-dependent branching logic
/// is unit-testable without a real terminal attached to the test process.
fn choose_streamer(is_tty: bool, opts: &ExecStreamedOpts) -> Streamer {
    if is_tty {
        if opts.use_scrolling_area {
            Streamer::LimitedArea
        } else {
            Streamer::Prefixed
        }
    } else {
        Streamer::Log
    }
}

/// Ported from the package-level `shellescape.QuoteCommand` call sites in
/// `exec_streamed.go`: a minimal, display-only shell quoter (this never
/// feeds back into an actual shell invocation, only a human-readable
/// header/prefix line). Wraps any argument containing whitespace or shell
/// metacharacters in single quotes, escaping embedded single quotes.
fn quote_command(parts: &[&str]) -> String {
    parts
        .iter()
        .map(|part| quote_arg(part))
        .collect::<Vec<_>>()
        .join(" ")
}

fn quote_arg(arg: &str) -> String {
    let needs_quoting = arg.is_empty()
        || arg
            .chars()
            .any(|c| c.is_whitespace() || "'\"\\$`!*?[]{}()<>|&;~#".contains(c));

    if !needs_quoting {
        return arg.to_string();
    }

    format!("'{}'", arg.replace('\'', r"'\''"))
}

/// Ported from `logStreamer`.
fn run_log_streamer(process_name: &str, line: &StreamLine) {
    match line.source {
        StreamSource::Stdout => tracing::debug!(command = %process_name, "stdout" = %line.text),
        StreamSource::Stderr => tracing::debug!(command = %process_name, "stderr" = %line.text),
    }
}

/// Ported from `prefixedStreamer`. Prints the command's header line once,
/// then each incoming line prefixed, wrapped, and colored (gray for
/// stdout, light white for stderr, matching Go's palette). Generic over
/// the writer so tests can assert on a `Vec<u8>` instead of real stderr;
/// production calls it with `anstream::stderr()`, which auto-strips the
/// styling below when not a terminal or `NO_COLOR` is set.
async fn run_prefixed_streamer(
    mut rx: mpsc::UnboundedReceiver<StreamLine>,
    process_name: &str,
    cmd_line: &str,
    mut out: impl std::io::Write,
) {
    let header_style = Style::new().bold().fg_color(Some(AnsiColor::White.into()));
    let _ = writeln!(out, "{header_style}{cmd_line}{header_style:#}");

    let prefix = format_prefix(process_name);
    let max_line_width = term_width_for_prefixed();

    let stdout_style = Style::new().fg_color(Some(AnsiColor::BrightBlack.into()));
    let stderr_style = Style::new().fg_color(Some(AnsiColor::BrightWhite.into()));

    while let Some(line) = rx.recv().await {
        let style = match line.source {
            StreamSource::Stdout => stdout_style,
            StreamSource::Stderr => stderr_style,
        };
        for wrapped in wrap_line(&line.text, max_line_width) {
            let _ = writeln!(out, "{prefix} {style}{wrapped}{style:#}");
        }
    }
}

/// Ported from the prefix-building half of `prefixedStreamer` (right-pad
/// to 8 chars, bold white).
fn format_prefix(process_name: &str) -> String {
    let truncated: String = process_name.chars().take(8).collect();
    let padded = format!("{truncated:<8}");
    let style = Style::new().bold().fg_color(Some(AnsiColor::White.into()));
    format!("{style}{padded}{style:#}")
}

/// Ported from `getTermWidth`.
fn term_width_for_prefixed() -> usize {
    let width = terminal_width().unwrap_or(80);
    width.saturating_sub(10).max(40)
}

/// Ported from `pterm.GetTerminalWidth`.
fn terminal_width() -> Option<usize> {
    terminal_size::terminal_size().map(|(terminal_size::Width(w), _)| w as usize)
}

/// Ported from `limitedAreaStreamer`. Generic over the writer for the same
/// testability reason as [`run_prefixed_streamer`].
async fn run_limited_area_streamer(
    mut rx: mpsc::UnboundedReceiver<StreamLine>,
    cmd_line: String,
    mut out: impl std::io::Write,
) {
    const MAX_LINES: usize = 5;

    let width = terminal_width().unwrap_or(80);
    let max_line_width = width.saturating_sub(1).max(40);

    let mut area = AreaPrinter::default();
    let mut display_lines: Vec<String> = Vec::new();
    let mut ticker = tokio::time::interval(Duration::from_millis(100));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        tokio::select! {
            received = rx.recv() => {
                match received {
                    Some(line) => {
                        display_lines.push(line.text);
                        if display_lines.len() > MAX_LINES {
                            let excess = display_lines.len() - MAX_LINES;
                            display_lines.drain(0..excess);
                        }
                    }
                    None => break,
                }
            }
            _ = ticker.tick() => {
                flush_lines_to_area(&cmd_line, &display_lines, max_line_width, &mut area);
                let _ = area.redraw(&mut out);
            }
        }
    }

    flush_lines_to_area(&cmd_line, &display_lines, max_line_width, &mut area);
    let _ = area.redraw(&mut out);
    let _ = area.finish(&mut out);
}

async fn exec_streamed_command_async(
    opts: ExecStreamedOpts,
    name: &str,
    args: &[String],
) -> Result<(), ExecError> {
    let mut cmd = TokioCommand::new(name);
    cmd.args(args);
    apply_env(&mut cmd, &opts.env);
    if let Some(dir) = &opts.working_dir
        && !dir.is_empty()
    {
        cmd.current_dir(dir);
    }
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let process_name = opts
        .process_descriptive_name
        .clone()
        .unwrap_or_else(|| name.to_string());

    let mut child = cmd
        .spawn()
        .map_err(|error| ExecError::RunFailed(error.to_string()))?;

    let stdout = child.stdout.take().expect("stdout was piped");
    let stderr = child.stderr.take().expect("stderr was piped");

    let (tx, rx) = mpsc::unbounded_channel::<StreamLine>();
    let stdout_tx = tx.clone();
    let stdout_task = tokio::spawn(stream_lines(stdout, StreamSource::Stdout, stdout_tx));
    let stderr_task = tokio::spawn(stream_lines(stderr, StreamSource::Stderr, tx.clone()));
    drop(tx);

    let streamer_kind = choose_streamer(is_stderr_tty(), &opts);
    let cmd_line_parts: Vec<&str> = std::iter::once(name)
        .chain(args.iter().map(String::as_str))
        .collect();
    let cmd_line = quote_command(&cmd_line_parts);

    let renderer_task = match streamer_kind {
        Streamer::Log => {
            let process_name = process_name.clone();
            tokio::spawn(async move {
                let mut rx = rx;
                while let Some(line) = rx.recv().await {
                    run_log_streamer(&process_name, &line);
                }
            })
        }
        Streamer::Prefixed => {
            let process_name = process_name.clone();
            tokio::spawn(async move {
                run_prefixed_streamer(rx, &process_name, &cmd_line, anstream::stderr()).await
            })
        }
        Streamer::LimitedArea => {
            tokio::spawn(run_limited_area_streamer(rx, cmd_line, anstream::stderr()))
        }
    };

    let wait_result = match opts.timeout {
        Some(timeout) => {
            tokio::select! {
                res = child.wait() => Some(res),
                () = tokio::time::sleep(timeout) => None,
            }
        }
        None => Some(child.wait().await),
    };

    let cancelled = wait_result.is_none();
    let final_status = match wait_result {
        Some(res) => res,
        None => {
            let _ = child.kill().await;
            child.wait().await
        }
    };

    let _ = stdout_task.await;
    let _ = stderr_task.await;
    let _ = renderer_task.await;

    if cancelled {
        return Err(ExecError::Cancelled);
    }

    match final_status {
        Ok(status) => {
            if status.success() {
                Ok(())
            } else {
                Err(ExecError::CommandFailed(status.code().unwrap_or(-1)))
            }
        }
        Err(error) => Err(ExecError::RunFailed(error.to_string())),
    }
}

/// Ported from the package-level `execStreamedCommand` function.
pub fn exec_streamed_command(
    opts: Option<ExecStreamedOpts>,
    name: &str,
    args: &[String],
) -> Result<(), ExecError> {
    runtime().block_on(exec_streamed_command_async(
        opts.unwrap_or_default(),
        name,
        args,
    ))
}

/// Ported from `truncateLine`.
pub fn truncate_line(line: &str, max_width: usize) -> String {
    line.chars().take(max_width).collect()
}

/// Ported from `wrapLine`.
pub fn wrap_line(line: &str, max_width: usize) -> Vec<String> {
    if max_width == 0 {
        return vec![line.to_string()];
    }

    let chars: Vec<char> = line.chars().collect();
    if chars.len() <= max_width {
        return vec![line.to_string()];
    }

    chars
        .chunks(max_width)
        .map(|chunk| chunk.iter().collect())
        .collect()
}

/// Ported from `isStderrTTY`.
pub fn is_stderr_tty() -> bool {
    use std::io::IsTerminal;
    std::io::stderr().is_terminal()
}

/// Ported from `writePrefixedLinesFiltered`. `tui` mirrors
/// `*logfilter.TUI`; see `crate::logfilter` for its port status.
///
/// The `Println`-based TUI output path (`tui.Println`) is not exercised
/// here since `logfilter::Tui::start`/`println` are themselves
/// unimplemented pending a terminal-rendering crate choice (see
/// `PORT_NOTES.md`); when a `Tui` is present, matching lines are still
/// logged via `tracing` rather than silently dropped.
pub fn write_prefixed_lines_filtered(
    tui: Option<&crate::logfilter::Tui>,
    stdout: &[String],
    stderr: &[String],
    styled_prefix: &str,
    max_line_width: usize,
) {
    let matches = |line: &str| tui.is_none_or(|tui| tui.matches(line));

    for line in stdout {
        if !matches(line) {
            continue;
        }
        for wrapped in wrap_line(line, max_line_width) {
            tracing::debug!("{styled_prefix} {wrapped}");
        }
    }

    for line in stderr {
        if !matches(line) {
            continue;
        }
        for wrapped in wrap_line(line, max_line_width) {
            tracing::debug!("{styled_prefix} {wrapped}");
        }
    }
}

/// A port of `pterm.AreaPrinter`: an in-place, self-redrawing block of
/// terminal lines (not a full alternate-screen TUI -- `pterm` itself is a
/// terminal-formatting library, not a TUI framework, so cursor-up-and-
/// overwrite via raw ANSI escapes is the faithful equivalent, matching
/// what `pterm.AreaPrinter` does internally).
///
/// `buffer` holds the plain-text content [`flush_lines_to_area`] builds
/// (kept styling-free so truncation-width assertions stay exact); colors
/// are applied separately in [`AreaPrinter::redraw`], after width
/// truncation, so SGR escape bytes never skew a line's displayed width.
#[derive(Default)]
pub struct AreaPrinter {
    pub buffer: String,
    /// How many terminal lines the last [`redraw`](Self::redraw) call
    /// wrote, so the next call knows how far to move the cursor up before
    /// overwriting.
    rendered_lines: usize,
}

impl AreaPrinter {
    /// Clears the previously rendered block (if any) and rewrites
    /// `self.buffer` in its place, coloring the header line bold white and
    /// the rest dark gray (matching `pterm.AreaPrinter.Update`'s styling in
    /// `flushLinesToArea`).
    pub fn redraw(&mut self, writer: &mut impl std::io::Write) -> std::io::Result<()> {
        let lines: Vec<&str> = self.buffer.lines().collect();

        if self.rendered_lines > 0 {
            write!(writer, "\x1b[{}A", self.rendered_lines)?;
        }

        let header_style = Style::new().bold().fg_color(Some(AnsiColor::White.into()));
        let content_style = Style::new().fg_color(Some(AnsiColor::BrightBlack.into()));

        for (i, line) in lines.iter().enumerate() {
            let style = if i == 0 { header_style } else { content_style };
            write!(writer, "\r\x1b[2K{style}{line}{style:#}\n")?;
        }

        self.rendered_lines = lines.len();
        writer.flush()
    }

    /// Moves the cursor below the area for good, mirroring
    /// `pterm.AreaPrinter.Stop` plus `limitedAreaStreamer`'s trailing
    /// `fmt.Fprintln(os.Stderr)`.
    pub fn finish(&mut self, writer: &mut impl std::io::Write) -> std::io::Result<()> {
        self.rendered_lines = 0;
        writeln!(writer)
    }
}

/// Ported from `flushLinesToArea`.
pub fn flush_lines_to_area(
    cmd_line: &str,
    display_lines: &[String],
    max_line_width: usize,
    area: &mut AreaPrinter,
) {
    if display_lines.is_empty() {
        return;
    }

    let header = truncate_line(cmd_line, max_line_width);
    let mut content = header;
    for line in display_lines {
        content.push('\n');
        content.push_str(&truncate_line(line, max_line_width));
    }
    area.buffer = content;
}

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

    #[test]
    fn exec_streamed_command_success() {
        let err = exec_streamed_command(None, "echo", &["hello world".into()]);
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_failure() {
        let err = exec_streamed_command(None, "sh", &["-c".into(), "exit 1".into()]).unwrap_err();
        assert!(matches!(err, ExecError::CommandFailed(1)));
    }

    #[test]
    fn exec_streamed_command_large_output() {
        let err = exec_streamed_command(None, "seq", &["1".into(), "100".into()]);
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_very_large_output() {
        let err = exec_streamed_command(None, "seq", &["1".into(), "1000".into()]);
        assert!(err.is_ok());
    }

    /// Ported from `TestExecStreamedCommand_ContextCancellation`.
    ///
    /// NOTE: Go's version spawns the command in a goroutine, cancels the
    /// context after 100ms, and blocks on the result channel. Rust's
    /// equivalent needs a `tokio::spawn`ed task plus a `CancellationToken`
    /// (or racing `tokio::select!` against a `tokio::process::Child`) --
    /// `std::thread::spawn` alone has no cancellation primitive. See
    /// PORT_NOTES.md.
    #[test]
    fn exec_streamed_command_context_cancellation() {
        // Go: context.WithCancel, cancelled after 100ms from a separate
        // goroutine. This port has no ambient context to cancel out of
        // band (see PORT_NOTES.md), so the equivalent deadline is passed
        // explicitly via `ExecStreamedOpts.timeout`.
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                timeout: Some(Duration::from_millis(100)),
                ..Default::default()
            }),
            "sleep",
            &["10".into()],
        );
        assert!(err.is_err(), "expected error after cancellation");
    }

    #[test]
    fn exec_streamed_command_context_timeout() {
        // Go: context.WithTimeout(200ms).
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                timeout: Some(Duration::from_millis(200)),
                ..Default::default()
            }),
            "sleep",
            &["10".into()],
        )
        .unwrap_err();
        assert!(matches!(err, ExecError::Cancelled));
    }

    #[test]
    fn exec_streamed_command_mixed_output() {
        let err = exec_streamed_command(
            None,
            "sh",
            &["-c".into(), "echo stdout; echo stderr >&2".into()],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_command_not_found() {
        let err = exec_streamed_command(None, "nonexistent-command-xyz", &[]);
        assert!(err.is_err(), "expected error for nonexistent command");
    }

    #[test]
    fn exec_streamed_command_empty_output() {
        let err = exec_streamed_command(None, "true", &[]);
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_long_running_command() {
        let err = exec_streamed_command(
            None,
            "sh",
            &[
                "-c".into(),
                "for i in $(seq 1 20); do echo $i; sleep 0.01; done".into(),
            ],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_long_lines() {
        let long_string = "a".repeat(1 << 10);
        let err = exec_streamed_command(None, "echo", &[long_string]);
        assert!(err.is_ok());
    }

    #[test]
    fn truncate_line_cases() {
        let cases: &[(&str, &str, usize, &str)] = &[
            ("line shorter than max", "hello", 10, "hello"),
            ("line equal to max", "hello", 5, "hello"),
            ("line longer than max", "hello world", 5, "hello"),
            ("empty line", "", 10, ""),
            (
                "very long line",
                "this is a very long line that should be truncated",
                20,
                "this is a very long ",
            ),
            ("emoji truncation", "hello 👋 world 🌍", 10, "hello 👋 wo"),
            (
                "multi-byte unicode characters",
                "日本語のテキスト",
                5,
                "日本語のテ",
            ),
            ("mixed ascii and unicode", "test 测试 тест", 8, "test 测试 "),
            ("truncate at emoji boundary", "abc👋def", 4, "abc👋"),
        ];

        for (name, line, max_width, want) in cases {
            assert_eq!(truncate_line(line, *max_width), *want, "{name}");
        }
    }

    #[test]
    fn flush_lines_to_area_long_cmd_line() {
        let mut area = AreaPrinter::default();

        let long_cmd = "a".repeat(200);
        let max_width = 80;

        flush_lines_to_area(
            &long_cmd,
            &["output line".to_string()],
            max_width,
            &mut area,
        );

        for line in area.buffer.lines() {
            assert!(
                line.chars().count() <= max_width,
                "area header line exceeds max_width: {line:?}"
            );
        }
    }

    /// New coverage (no direct Go equivalent -- `pterm.AreaPrinter`'s
    /// redraw is opaque library internals Go itself doesn't unit test):
    /// the first `redraw` writes the buffer's lines without a leading
    /// cursor-up escape (nothing to clear yet), and colors the header line
    /// differently from the content lines.
    #[test]
    fn area_printer_redraw_first_call_has_no_cursor_up() {
        let mut area = AreaPrinter {
            buffer: "header\nline one\nline two".to_string(),
            ..Default::default()
        };

        let mut out = Vec::new();
        area.redraw(&mut out).unwrap();
        let rendered = String::from_utf8(out).unwrap();

        assert!(
            !rendered.starts_with("\x1b["),
            "first redraw should not move the cursor: {rendered:?}"
        );
        assert!(rendered.contains("header"));
        assert!(rendered.contains("line one"));
        assert!(rendered.contains("line two"));
        assert_eq!(area.rendered_lines, 3);
    }

    /// A second `redraw` moves the cursor up by exactly the previous
    /// line count before rewriting, so the block overwrites in place
    /// instead of scrolling the terminal.
    #[test]
    fn area_printer_redraw_second_call_moves_cursor_up_by_previous_line_count() {
        let mut area = AreaPrinter {
            buffer: "header\nline one".to_string(),
            ..Default::default()
        };
        area.redraw(&mut Vec::new()).unwrap();
        assert_eq!(area.rendered_lines, 2);

        area.buffer = "header\nline one\nline two".to_string();
        let mut out = Vec::new();
        area.redraw(&mut out).unwrap();
        let rendered = String::from_utf8(out).unwrap();

        assert!(
            rendered.starts_with("\x1b[2A"),
            "should move up by the previous render's 2 lines: {rendered:?}"
        );
        assert_eq!(area.rendered_lines, 3);
    }

    /// `finish` resets the line count (so a later `redraw` starts fresh,
    /// matching a brand-new command's area) and leaves the cursor below
    /// the block with a trailing newline, mirroring
    /// `limitedAreaStreamer`'s `fmt.Fprintln(os.Stderr)` after
    /// `area.Stop()`.
    #[test]
    fn area_printer_finish_resets_and_writes_trailing_newline() {
        let mut area = AreaPrinter {
            buffer: "header\nline one".to_string(),
            ..Default::default()
        };
        area.redraw(&mut Vec::new()).unwrap();
        assert_eq!(area.rendered_lines, 2);

        let mut out = Vec::new();
        area.finish(&mut out).unwrap();

        assert_eq!(area.rendered_lines, 0);
        assert_eq!(out, b"\n");
    }

    /// New coverage: `quote_command` (this port's `shellescape.QuoteCommand`
    /// stand-in) leaves plain arguments bare and single-quotes anything
    /// shell-meaningful, escaping embedded single quotes.
    #[test]
    fn quote_command_cases() {
        let cases: &[(&str, &[&str], &str)] = &[
            (
                "plain args need no quoting",
                &["helm", "install"],
                "helm install",
            ),
            (
                "an argument with a space is quoted",
                &["echo", "hello world"],
                "echo 'hello world'",
            ),
            ("an empty argument is quoted", &["echo", ""], "echo ''"),
            (
                "an embedded single quote is escaped",
                &["echo", "it's"],
                "echo 'it'\\''s'",
            ),
            (
                "a shell metacharacter is quoted",
                &["sh", "-c", "a && b"],
                "sh -c 'a && b'",
            ),
        ];

        for (name, parts, want) in cases {
            assert_eq!(quote_command(parts), *want, "{name}");
        }
    }

    /// New coverage: `format_prefix` right-pads to 8 characters and
    /// truncates longer names, matching `prefixedStreamer`'s prefix
    /// construction.
    #[test]
    fn format_prefix_pads_and_truncates() {
        assert_eq!(strip_ansi(&format_prefix("gitaly")), "gitaly  ");
        assert_eq!(
            strip_ansi(&format_prefix("a-very-long-process-name")),
            "a-very-l"
        );
    }

    /// Strips SGR escape sequences, for asserting on a styled string's
    /// plain-text content without hard-coding the exact ANSI bytes.
    fn strip_ansi(s: &str) -> String {
        let mut result = String::new();
        let mut chars = s.chars();
        while let Some(c) = chars.next() {
            if c == '\x1b' {
                for esc_c in chars.by_ref() {
                    if esc_c == 'm' {
                        break;
                    }
                }
                continue;
            }
            result.push(c);
        }
        result
    }

    #[test]
    fn is_stderr_tty_does_not_panic() {
        let _ = is_stderr_tty();
    }

    /// Ported from `chooseStreamer`'s three branches. Takes `is_tty`
    /// explicitly (see `choose_streamer`'s doc comment) so this is
    /// testable without a real terminal attached to the test process.
    #[test]
    fn choose_streamer_cases() {
        let scrolling = ExecStreamedOpts {
            use_scrolling_area: true,
            ..Default::default()
        };
        let not_scrolling = ExecStreamedOpts {
            use_scrolling_area: false,
            ..Default::default()
        };

        assert_eq!(
            choose_streamer(true, &scrolling),
            Streamer::LimitedArea,
            "TTY + scrolling area"
        );
        assert_eq!(
            choose_streamer(true, &not_scrolling),
            Streamer::Prefixed,
            "TTY without scrolling area"
        );
        assert_eq!(
            choose_streamer(false, &scrolling),
            Streamer::Log,
            "no TTY always falls back to the logger, regardless of the option"
        );
        assert_eq!(
            choose_streamer(false, &not_scrolling),
            Streamer::Log,
            "no TTY, no scrolling area"
        );
    }

    /// New coverage: `run_prefixed_streamer` prints the header once, then
    /// each line prefixed and wrapped, stdout and stderr both reaching the
    /// output (colored differently, but [`strip_ansi`] hides that here).
    #[test]
    fn run_prefixed_streamer_prints_header_then_prefixed_lines() {
        let (tx, rx) = mpsc::unbounded_channel();
        tx.send(StreamLine {
            source: StreamSource::Stdout,
            text: "building...".to_string(),
        })
        .unwrap();
        tx.send(StreamLine {
            source: StreamSource::Stderr,
            text: "a warning".to_string(),
        })
        .unwrap();
        drop(tx);

        let mut out = Vec::new();
        runtime().block_on(run_prefixed_streamer(
            rx,
            "helm",
            "helm install app",
            &mut out,
        ));
        let rendered = strip_ansi(&String::from_utf8(out).unwrap());

        let lines: Vec<&str> = rendered.lines().collect();
        assert_eq!(lines[0], "helm install app", "header line");
        assert_eq!(lines[1], "helm     building...");
        assert_eq!(lines[2], "helm     a warning");
    }

    /// New coverage: `run_limited_area_streamer` keeps only the last 5
    /// lines and redraws the area at least once before the channel closes.
    #[test]
    fn run_limited_area_streamer_keeps_only_the_last_five_lines() {
        let (tx, rx) = mpsc::unbounded_channel();
        for i in 1..=8 {
            tx.send(StreamLine {
                source: StreamSource::Stdout,
                text: format!("line {i}"),
            })
            .unwrap();
        }
        drop(tx);

        let mut out = Vec::new();
        runtime().block_on(run_limited_area_streamer(
            rx,
            "a-long-running-command".to_string(),
            &mut out,
        ));
        let rendered = strip_ansi(&String::from_utf8(out).unwrap());

        // The final redraw (after the channel closes) reflects the last 5
        // lines only; earlier lines are gone.
        assert!(rendered.contains("line 8"));
        assert!(rendered.contains("line 4"));
        assert!(!rendered.contains("line 3"));
        assert!(!rendered.contains("line 1"));
    }

    #[test]
    fn exec_streamed_command_rapid_output() {
        let err = exec_streamed_command(None, "seq", &["1".into(), "100000".into()]);
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_binary_output() {
        let err = exec_streamed_command(
            None,
            "sh",
            &["-c".into(), r"printf '\x00\x01\x02\x03\xff\xfe\xfd'".into()],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_unicode_output() {
        let err = exec_streamed_command(
            None,
            "sh",
            &["-c".into(), "echo '✓ Success 👍 日本語 🌍'".into()],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_with_opts_success() {
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                process_descriptive_name: Some("test-process".into()),
                use_scrolling_area: false,
                ..Default::default()
            }),
            "echo",
            &["hello world".into()],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_with_opts_failure() {
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                process_descriptive_name: Some("test-process".into()),
                use_scrolling_area: false,
                ..Default::default()
            }),
            "sh",
            &["-c".into(), "exit 1".into()],
        )
        .unwrap_err();
        assert!(matches!(err, ExecError::CommandFailed(1)));
    }

    #[test]
    fn exec_streamed_command_with_opts_context_cancellation() {
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                process_descriptive_name: Some("long-process".into()),
                use_scrolling_area: false,
                timeout: Some(Duration::from_millis(100)),
                ..Default::default()
            }),
            "sleep",
            &["10".into()],
        );
        assert!(err.is_err(), "expected error after cancellation");
    }

    /// Ported from `TestExecStreamedCommand_MultipleProcesses`.
    #[test]
    fn exec_streamed_command_multiple_processes() {
        for i in 1..=3 {
            let err = exec_streamed_command(
                Some(ExecStreamedOpts {
                    process_descriptive_name: Some(format!("process-{i}")),
                    use_scrolling_area: false,
                    ..Default::default()
                }),
                "echo",
                &[format!("process {i} output")],
            );
            assert!(err.is_ok());
        }
    }

    #[test]
    fn exec_streamed_command_process_identification() {
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                process_descriptive_name: Some("gitaly-process".into()),
                use_scrolling_area: false,
                ..Default::default()
            }),
            "echo",
            &["gitaly output".into()],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_working_directory() {
        let tmp_dir = std::env::temp_dir();

        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                working_dir: Some(tmp_dir.to_string_lossy().into_owned()),
                process_descriptive_name: Some("test-process".into()),
                use_scrolling_area: false,
                ..Default::default()
            }),
            "pwd",
            &[],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_empty_dir() {
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                working_dir: Some(String::new()),
                process_descriptive_name: Some("test-process".into()),
                use_scrolling_area: false,
                ..Default::default()
            }),
            "echo",
            &["test".into()],
        );
        assert!(err.is_ok());
    }

    #[test]
    fn exec_streamed_command_nonexistent_dir() {
        let err = exec_streamed_command(
            Some(ExecStreamedOpts {
                working_dir: Some("/nonexistent/directory/path".into()),
                process_descriptive_name: Some("test-process".into()),
                use_scrolling_area: false,
                ..Default::default()
            }),
            "echo",
            &["test".into()],
        );
        assert!(err.is_err());
    }

    #[test]
    fn write_prefixed_lines_filtered_nil_tui_does_not_panic() {
        write_prefixed_lines_filtered(None, &["[vite] some log".to_string()], &[], "prefix", 80);
    }
}