//! stdio transport (MCP local subprocess).
//!
//! Why a dedicated module: the stdio wire shape is the simplest MCP
//! transport and the one most local developers reach for first. Spawn
//! the configured server as a child process, frame newline-delimited
//! JSON-RPC over its stdin/stdout, and keep stderr separate so the
//! reporter can show server diagnostics without corrupting the message
//! stream.
//!
//! Responsibilities:
//!
//! - Spawn the configured command + args + env via
//!   [`tokio::process::Command`] with piped stdin, stdout, and stderr.
//! - Reader task: pull lines off the child's stdout, parse each as a
//!   [`JsonRpcMessage`], push onto the inbound channel. Malformed lines
//!   are skipped and surfaced to parent stderr for diagnosis.
//! - Writer task: drain the outbound channel, serialize each envelope as
//!   one line of JSON plus a `\n`, write to the child's stdin.
//! - Stderr pump: forward the child's stderr to the parent's stderr line
//!   by line so a chatty server never blocks the protocol stream.
//! - Graceful shutdown: drop the outbound channel (the writer task then
//!   drops stdin so the child sees EOF), wait up to
//!   [`StdioConfig::shutdown_timeout`] for the child to exit, escalate
//!   to a forced kill if it stalls.
//! - Surface typed errors for spawn failures, premature exits, and
//!   shutdown timeouts.
//!
//! The transport returns the same channel pair shape every other
//! transport in this crate produces (see ADR-0022 / ), so the
//! protocol [`Client`](crate::protocol::Client) consumes it through the
//! [`Transport`](crate::protocol::Transport) struct without knowing it
//! is talking to a subprocess.

use std::collections::{HashMap, VecDeque};
use std::process::{ExitStatus, Stdio as ProcessStdio};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use thiserror::Error;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::protocol::{JsonRpcMessage, Transport};

/// Default wait between the polite shutdown signal and the hard kill.
///
/// Five seconds gives a well-behaved MCP server time to flush its
/// outbound queue and exit cleanly. Servers that ignore the polite
/// signal get killed at the end of this window so the runner is never
/// stuck waiting on a wedged child.
pub const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);

/// Capacity of the inbound and outbound envelope queues (WOR-1313).
///
/// Bounded so a flooding or wedged peer fills a fixed-size queue
/// instead of growing process memory without limit. 1024 envelopes is
/// far beyond what a healthy request/response session keeps in flight,
/// so a full queue is a reliable misbehavior signal, not a tuning knob
/// users should ever hit.
pub const CHANNEL_CAPACITY: usize = 1024;

/// How long the reader waits on a full inbound queue before declaring
/// the transport flooded (WOR-1313).
///
/// Generous on purpose: a briefly slow consumer (a paused test, a busy
/// CI box) must not kill the connection. Only a consumer that makes no
/// progress for this entire window while the queue holds
/// [`CHANNEL_CAPACITY`] envelopes counts as flooded.
const INBOUND_FLOOD_TIMEOUT: Duration = Duration::from_secs(2);

/// Total budget for reaping the child after the forced kill (WOR-1314).
///
/// SIGKILL is not ignorable, so a child that fails to reap inside this
/// window is stuck in the kernel (uninterruptible I/O, a broken init as
/// subreaper). We stop waiting and report [`StdioError::ZombieChild`]
/// rather than hanging the runner on `wait()` forever.
const KILL_REAP_DEADLINE: Duration = Duration::from_secs(2);

/// Polling step for the post-kill reap loop (WOR-1314). Short enough
/// that a normally dying child adds at most ~50ms to shutdown.
const KILL_REAP_POLL_INTERVAL: Duration = Duration::from_millis(50);

/// How many trailing child stderr lines the transport retains for
/// diagnostics (WOR-1326).
///
/// When the child dies before answering `initialize`, the actionable
/// cause is on stderr. Sized so the canonical Node `MODULE_NOT_FOUND`
/// failure (an 18-line stack trace whose headline comes first) keeps
/// its `Error: Cannot find module ...` line inside the tail.
pub const STDERR_TAIL_LINES: usize = 20;

/// How long [`StdioTransport::exit_status_within`] waits for the
/// stderr pump to finish after the child exits, so the tail buffer
/// holds the child's final lines before the caller snapshots it.
const STDERR_DRAIN_TIMEOUT: Duration = Duration::from_millis(200);

/// Ring buffer of the child's most recent stderr lines, shared between
/// the stderr pump task and the transport handle (WOR-1326).
type StderrTail = Arc<Mutex<VecDeque<String>>>;

/// Construction parameters for [`StdioTransport::spawn`].
///
/// Held as a struct (not a long positional argument list) so future
/// knobs (current dir, stderr redirection target) can land without
/// breaking the builder shape.
#[derive(Debug, Clone)]
pub struct StdioConfig {
    /// Program to execute. Looked up via `$PATH` if it is not absolute.
    pub command: String,
    /// Argv after the program name.
    pub args: Vec<String>,
    /// Environment variables overlaid on the parent's environment.
    pub env: HashMap<String, String>,
    /// Optional working directory. Defaults to the parent's CWD.
    pub working_dir: Option<std::path::PathBuf>,
    /// How long to wait between the polite shutdown signal (EOF on the
    /// child's stdin) and the forced kill during
    /// [`StdioTransport::shutdown`].
    pub shutdown_timeout: Duration,
}

impl Default for StdioConfig {
    fn default() -> Self {
        Self {
            command: String::new(),
            args: Vec::new(),
            env: HashMap::new(),
            working_dir: None,
            shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
        }
    }
}

/// Errors raised while spawning, running, or shutting down a stdio
/// transport.
///
/// Spawn-time errors surface to the caller of
/// [`StdioTransport::spawn`]. Runtime errors (malformed JSON from the
/// server, premature exit) close the inbound channel; the protocol
/// Client observes them as `TransportClosed`. Shutdown errors surface
/// to the caller of [`StdioTransport::shutdown`].
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum StdioError {
    /// The command was empty at spawn time.
    #[error("stdio transport requires a non-empty command")]
    EmptyCommand,
    /// `tokio::process::Command::spawn` failed (binary missing,
    /// permission denied, etc.).
    #[error("failed to spawn `{command}`: {source}")]
    SpawnFailed {
        /// Command we tried to run.
        command: String,
        /// Underlying OS error.
        #[source]
        source: std::io::Error,
    },
    /// Spawn succeeded but one of stdin or stdout was not piped, so the
    /// transport cannot frame messages. This should not happen in
    /// practice but the typed variant keeps the error path total.
    #[error("child process did not expose piped stdio")]
    MissingPipes,
    /// The server emitted a line that was not valid JSON-RPC. The
    /// reader task skips the line and keeps reading; the variant exists
    /// so callers that want to surface the raw line can downcast.
    #[error("malformed JSON line from server: {source}: {line:?}")]
    MalformedJson {
        /// The raw line as received (newline stripped).
        line: String,
        /// Underlying serde error.
        #[source]
        source: serde_json::Error,
    },
    /// The child exited before [`StdioTransport::shutdown`] was called.
    /// Carries the exit status so callers can distinguish clean exit
    /// from a crash.
    #[error("child process exited before shutdown: {status}")]
    PrematureExit {
        /// Exit status of the child.
        status: ExitStatus,
    },
    /// The child exited while the MCP handshake was still pending
    /// (WOR-1326). Carries the exit status and the tail of the child's
    /// stderr so the failure names its actual cause (a missing module,
    /// a crash on startup) instead of surfacing as a bare timeout.
    #[error(
        "server process exited during startup ({status}){}",
        render_stderr_tail(stderr_tail)
    )]
    StartupExit {
        /// Exit status of the dead child.
        status: ExitStatus,
        /// Last few stderr lines the child wrote before exiting, oldest
        /// first.
        stderr_tail: Vec<String>,
    },
    /// The child did not exit within
    /// [`StdioConfig::shutdown_timeout`] of being asked to. We escalated
    /// to a forced kill and the kill was reaped, so no process leaked;
    /// the error exists so callers can flag the server as misbehaved.
    #[error("child did not exit within {0:?}; escalated to forced kill")]
    ShutdownTimeout(Duration),
    /// The child survived the forced kill: it could not be reaped
    /// within the post-kill deadline (WOR-1314). SIGKILL cannot be
    /// caught, so this means the process is wedged in the kernel and
    /// the pid may linger as a zombie until the OS clears it. Named so
    /// callers can tell "killed late but cleaned up"
    /// ([`StdioError::ShutdownTimeout`]) from "leaked a process".
    #[error("child process {pid} could not be reaped after forced kill; leaving it to the OS")]
    ZombieChild {
        /// OS process id of the unreapable child, for the operator.
        pid: u32,
    },
    /// The peer kept the inbound queue full past the flood window
    /// (WOR-1313): it produced messages faster than the consumer could
    /// drain for the entire timeout. The reader shuts the connection
    /// down instead of buffering without bound; the variant documents
    /// the incident in logs the same way `MalformedJson` does.
    #[error(
        "inbound queue stayed full ({capacity} envelopes) for {timeout:?}; \
         treating peer as flooding and closing the transport"
    )]
    InboundFlooded {
        /// Queue capacity that was saturated.
        capacity: usize,
        /// How long the reader waited before giving up.
        timeout: Duration,
    },
}

/// Live stdio transport handle.
///
/// Owns the child process and the reader, writer, and stderr-pump
/// tasks. Drop the handle to abort the tasks and let the child go (best
/// effort, via `kill_on_drop`). The usual shutdown route is
/// [`StdioTransport::shutdown`], which waits for the child to exit and
/// escalates to a hard kill if needed.
#[derive(Debug)]
pub struct StdioTransport {
    child: Child,
    writer: JoinHandle<()>,
    reader: JoinHandle<()>,
    stderr_pump: Option<JoinHandle<()>>,
    stderr_tail: StderrTail,
    shutdown_timeout: Duration,
}

impl StdioTransport {
    /// Spawn the configured server and wire its stdio to a [`Transport`].
    ///
    /// The reader and writer tasks run on the current tokio runtime.
    /// Both tasks exit when either end of the channels closes, which is
    /// the same shutdown contract every other transport in this crate
    /// uses.
    pub fn spawn(config: StdioConfig) -> Result<(Transport, Self), StdioError> {
        if config.command.trim().is_empty() {
            return Err(StdioError::EmptyCommand);
        }

        let mut cmd = Command::new(&config.command);
        cmd.args(&config.args);
        for (k, v) in &config.env {
            cmd.env(k, v);
        }
        if let Some(dir) = &config.working_dir {
            cmd.current_dir(dir);
        }
        cmd.stdin(ProcessStdio::piped())
            .stdout(ProcessStdio::piped())
            .stderr(ProcessStdio::piped())
            .kill_on_drop(true);

        // Spawn, retrying briefly on ETXTBSY. Executing a file that was written
        // moments ago (a freshly installed server binary, or a script a test
        // just wrote) can fail with "text file busy" while the kernel still
        // sees a writer; a short bounded retry clears it without masking a
        // genuinely un-executable command. ETXTBSY is 26 on Linux and macOS;
        // Windows has no such error, so the guard simply never matches there.
        let mut child = {
            const ETXTBSY: i32 = 26;
            let mut attempt: u64 = 0;
            loop {
                match cmd.spawn() {
                    Ok(child) => break child,
                    Err(e) if e.raw_os_error() == Some(ETXTBSY) && attempt < 5 => {
                        attempt += 1;
                        std::thread::sleep(std::time::Duration::from_millis(20 * attempt));
                    }
                    Err(e) => {
                        return Err(StdioError::SpawnFailed {
                            command: config.command.clone(),
                            source: e,
                        });
                    }
                }
            }
        };

        tracing::debug!(
            target: "mcptest_core::transport::stdio",
            command = %config.command,
            args = ?config.args,
            pid = child.id(),
            "spawned stdio child"
        );

        let child_stdin = child.stdin.take().ok_or(StdioError::MissingPipes)?;
        let child_stdout = child.stdout.take().ok_or(StdioError::MissingPipes)?;
        let child_stderr = child.stderr.take();

        // Bounded both ways (WOR-1313): the inbound queue caps how much
        // a flooding server can buffer in this process, and the
        // outbound queue pushes back on the Client (which sends with
        // `send_wait`) instead of accumulating frames a wedged child
        // will never read.
        let (outbound_tx, outbound_rx) = mpsc::channel::<JsonRpcMessage>(CHANNEL_CAPACITY);
        let (inbound_tx, inbound_rx) = mpsc::channel::<JsonRpcMessage>(CHANNEL_CAPACITY);

        let writer = spawn_writer(child_stdin, outbound_rx);
        let reader = spawn_reader(child_stdout, inbound_tx);
        let stderr_tail: StderrTail = Arc::new(Mutex::new(VecDeque::new()));
        let stderr_pump =
            child_stderr.map(|stderr| spawn_stderr_pump(stderr, Arc::clone(&stderr_tail)));

        let transport = Transport::new_bounded(outbound_tx, inbound_rx);
        Ok((
            transport,
            Self {
                child,
                writer,
                reader,
                stderr_pump,
                stderr_tail,
                shutdown_timeout: config.shutdown_timeout,
            },
        ))
    }

    /// Report the child's exit status if it terminates within `window`,
    /// `None` if it is still running when the window closes.
    ///
    /// Why: when the `initialize` handshake fails, the connector needs
    /// to know whether the failure is "the child died" (an actionable
    /// startup error, WOR-1326) or something else. On exit this also
    /// waits briefly for the stderr pump so [`Self::stderr_tail`] holds
    /// the child's final lines.
    pub async fn exit_status_within(&mut self, window: Duration) -> Option<ExitStatus> {
        let status = tokio::time::timeout(window, self.child.wait())
            .await
            .ok()?
            .ok()?;
        if let Some(pump) = self.stderr_pump.take() {
            // The child is gone, so its stderr is at EOF and the pump is
            // about to finish on its own; the timeout only guards against
            // a pathologically stalled pipe. A timed-out pump is simply
            // dropped (it holds no resources beyond the closed pipe).
            let _ = tokio::time::timeout(STDERR_DRAIN_TIMEOUT, pump).await;
        }
        Some(status)
    }

    /// Snapshot the last few stderr lines the child wrote, oldest first
    /// (WOR-1326). Capped at [`STDERR_TAIL_LINES`].
    pub fn stderr_tail(&self) -> Vec<String> {
        match self.stderr_tail.lock() {
            Ok(tail) => tail.iter().cloned().collect(),
            // A panicked stderr pump poisons the lock. The tail is
            // best-effort diagnostics, so degrade to an empty snapshot.
            Err(_) => Vec::new(),
        }
    }

    /// Wait for the child to exit, escalating to a hard kill on timeout.
    ///
    /// Stdin is closed first by aborting the writer task before waiting.
    /// The writer owns the child's `ChildStdin`, so aborting it drops
    /// stdin immediately and delivers EOF to the child regardless of
    /// whether the caller has already dropped its outbound `Transport`.
    /// Without this ordering the wait window starts with stdin still open,
    /// an EOF-respecting child never sees the signal, and every graceful
    /// shutdown burns the full `shutdown_timeout` before being force-killed.
    ///
    /// If the child ignores EOF for longer than `shutdown_timeout`, this
    /// method falls back to
    /// [`Child::start_kill`](tokio::process::Child::start_kill) (which
    /// sends SIGKILL on Unix and `TerminateProcess` on Windows). Three
    /// distinguishable outcomes (WOR-1314): `Ok(())` for a clean exit,
    /// [`StdioError::ShutdownTimeout`] when the kill landed and the
    /// child was reaped, and [`StdioError::ZombieChild`] when even the
    /// kill could not reap the process within a short deadline.
    pub async fn shutdown(mut self) -> Result<(), StdioError> {
        // Close stdin first so the child receives EOF before we start the
        // wait timer. The writer task owns the `ChildStdin` handle; aborting
        // it drops that handle, which is the EOF signal an MCP server
        // depends on to know it should flush and exit.
        self.writer.abort();
        let _ = self.writer.await;

        let outcome = tokio::time::timeout(self.shutdown_timeout, self.child.wait()).await;
        let result = match outcome {
            Ok(_status) => Ok(()),
            Err(_) => {
                // Polite EOF was ignored. Escalate to a forced kill so
                // the caller is never blocked on a wedged child. Snapshot
                // the pid before the kill: a reaped child reports None.
                let pid = self.child.id().unwrap_or(0);
                let _ = self.child.start_kill();
                if reap_with_deadline(&mut self.child, KILL_REAP_DEADLINE, KILL_REAP_POLL_INTERVAL)
                    .await
                {
                    Err(StdioError::ShutdownTimeout(self.shutdown_timeout))
                } else {
                    tracing::error!(
                        target: "mcptest_core::transport::stdio",
                        pid,
                        deadline = ?KILL_REAP_DEADLINE,
                        "child survived forced kill and could not be reaped"
                    );
                    Err(StdioError::ZombieChild { pid })
                }
            }
        };

        // Abort the reader and stderr pump so their pipe halves drop. We
        // ignore the join results because aborted tasks return
        // `JoinError::Cancelled`, which is the expected outcome here.
        self.reader.abort();
        if let Some(p) = self.stderr_pump.take() {
            p.abort();
            let _ = p.await;
        }
        let _ = self.reader.await;

        result
    }
}

/// Poll `try_wait` until the child is reaped or `deadline` elapses.
///
/// Exists because a bare `wait().await` after the forced kill can hang
/// forever on a child stuck in the kernel (WOR-1314); the bounded poll
/// turns that hang into a reportable [`StdioError::ZombieChild`].
async fn reap_with_deadline(child: &mut Child, deadline: Duration, step: Duration) -> bool {
    let started = tokio::time::Instant::now();
    loop {
        match child.try_wait() {
            Ok(Some(_status)) => return true,
            Ok(None) => {}
            Err(err) => {
                // The OS refused to report on our own child; nothing
                // further we can do, so treat it as unreaped.
                tracing::error!(
                    target: "mcptest_core::transport::stdio",
                    error = %err,
                    "try_wait failed while reaping killed child"
                );
                return false;
            }
        }
        if started.elapsed() >= deadline {
            return false;
        }
        tokio::time::sleep(step).await;
    }
}

/// Drain the outbound channel and write newline-framed JSON to the
/// child's stdin. Exits when the channel closes (graceful shutdown) or
/// the child closes its stdin (broken pipe).
fn spawn_writer(
    mut stdin: ChildStdin,
    mut outbound_rx: mpsc::Receiver<JsonRpcMessage>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        while let Some(msg) = outbound_rx.recv().await {
            let mut buf = match serde_json::to_vec(&msg) {
                Ok(buf) => buf,
                Err(err) => {
                    // Envelope refused to serialize. Surface the loss
                    // loudly (WOR-1313) instead of dropping it in
                    // silence, then keep the session alive: one bad
                    // envelope must not tear down the transport.
                    tracing::error!(
                        target: "mcptest_core::transport::stdio",
                        error = %err,
                        method = ?msg.method,
                        id = ?msg.id,
                        "dropping outbound envelope that failed to serialize"
                    );
                    continue;
                }
            };
            buf.push(b'\n');
            if stdin.write_all(&buf).await.is_err() {
                // Child has closed its stdin. Nothing further to do;
                // the reader task will surface the exit through its own
                // channel close.
                return;
            }
            if stdin.flush().await.is_err() {
                return;
            }
        }
        // Channel closed by the Client. Drop stdin so the child sees
        // EOF and exits.
    })
}

/// Read newline-delimited JSON envelopes off the child's stdout and
/// push parsed messages onto the inbound channel. Exits on EOF or
/// channel close. Malformed lines are reported to parent stderr and
/// skipped so a single garbled message never tears down the session.
/// A queue that stays full past [`INBOUND_FLOOD_TIMEOUT`] means the
/// peer is flooding (WOR-1313): the reader logs a typed error and
/// closes the connection instead of buffering or blocking forever.
fn spawn_reader(stdout: ChildStdout, inbound_tx: mpsc::Sender<JsonRpcMessage>) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut reader = BufReader::new(stdout);
        let mut line = String::new();
        loop {
            line.clear();
            let read = reader.read_line(&mut line).await;
            let Ok(n) = read else {
                return;
            };
            if n == 0 {
                // EOF: child closed stdout. Drop the sender so the
                // Client observes `TransportClosed`.
                return;
            }
            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }
            let parsed: Result<JsonRpcMessage, _> = serde_json::from_str(trimmed);
            let msg = match parsed {
                Ok(m) => m,
                Err(err) => {
                    // Surface the structured error to stderr so a
                    // developer can spot a misbehaving server. The
                    // typed `StdioError::MalformedJson` variant
                    // documents the same wire incident.
                    let stamped = StdioError::MalformedJson {
                        line: trimmed.to_string(),
                        source: err,
                    };
                    tracing::warn!(
                        target: "mcptest_core::transport::stdio",
                        error = %stamped,
                        "malformed JSON line from child"
                    );
                    continue;
                }
            };
            match inbound_tx.send_timeout(msg, INBOUND_FLOOD_TIMEOUT).await {
                Ok(()) => {}
                Err(mpsc::error::SendTimeoutError::Closed(_)) => {
                    // Client dropped its end. Nothing more to do.
                    return;
                }
                Err(mpsc::error::SendTimeoutError::Timeout(_)) => {
                    // The queue held CHANNEL_CAPACITY envelopes for the
                    // whole window with no consumer progress: the peer
                    // is outrunning the session. Returning drops
                    // `inbound_tx`, which closes the channel; the
                    // Client observes `TransportClosed` after draining
                    // what is already buffered.
                    let stamped = StdioError::InboundFlooded {
                        capacity: CHANNEL_CAPACITY,
                        timeout: INBOUND_FLOOD_TIMEOUT,
                    };
                    tracing::error!(
                        target: "mcptest_core::transport::stdio",
                        error = %stamped,
                        "shutting down flooded stdio transport"
                    );
                    return;
                }
            }
        }
    })
}

/// Drain the child's stderr line by line.  reroutes the lines
/// through `tracing::warn!` so operators can filter them with the
/// global `--log-level` knob (default `warn`, so server stderr is
/// still visible without extra flags). Each line is also kept in the
/// shared tail ring buffer (WOR-1326) so a startup failure can quote
/// the child's final words instead of pointing at the log scroll.
fn spawn_stderr_pump(stderr: ChildStderr, tail: StderrTail) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut reader = BufReader::new(stderr);
        let mut line = String::new();
        loop {
            line.clear();
            let read = reader.read_line(&mut line).await;
            let Ok(n) = read else {
                return;
            };
            if n == 0 {
                return;
            }
            let trimmed = line.trim_end();
            tracing::warn!(
                target: "mcptest_core::transport::stdio",
                line = %trimmed,
                "child stderr"
            );
            // A poisoned lock means a previous pump panicked; the tail
            // is best-effort diagnostics, so skipping the line is fine.
            if let Ok(mut buf) = tail.lock() {
                if buf.len() == STDERR_TAIL_LINES {
                    buf.pop_front();
                }
                buf.push_back(trimmed.to_string());
            }
        }
    })
}

/// Render the stderr tail as a Display suffix for
/// [`StdioError::StartupExit`]: a note when the child was silent,
/// otherwise the lines indented one per row so the actionable cause
/// reads directly out of the error message.
fn render_stderr_tail(lines: &[String]) -> String {
    if lines.is_empty() {
        return "; the server wrote nothing to stderr".to_string();
    }
    let mut out = String::from("; last stderr lines:");
    for line in lines {
        out.push_str("\n  ");
        out.push_str(line);
    }
    out
}

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

    #[test]
    fn config_default_uses_five_second_timeout() {
        let c = StdioConfig::default();
        assert_eq!(c.shutdown_timeout, Duration::from_secs(5));
    }

    #[test]
    fn empty_command_is_rejected_at_spawn() {
        let cfg = StdioConfig {
            command: "".into(),
            ..Default::default()
        };
        match StdioTransport::spawn(cfg) {
            Err(StdioError::EmptyCommand) => {}
            other => panic!("expected EmptyCommand, got {other:?}"),
        }
    }

    #[test]
    fn whitespace_only_command_is_rejected_at_spawn() {
        let cfg = StdioConfig {
            command: "   ".into(),
            ..Default::default()
        };
        assert!(matches!(
            StdioTransport::spawn(cfg),
            Err(StdioError::EmptyCommand)
        ));
    }

    #[tokio::test]
    async fn missing_binary_returns_spawn_error() {
        let cfg = StdioConfig {
            command: "/definitely/not/a/real/binary/mcptest-stdio".into(),
            ..Default::default()
        };
        match StdioTransport::spawn(cfg) {
            Err(StdioError::SpawnFailed { command, .. }) => {
                assert!(command.ends_with("mcptest-stdio"));
            }
            other => panic!("expected SpawnFailed, got {other:?}"),
        }
    }

    #[test]
    fn premature_exit_error_renders_status() {
        // We cannot fabricate an `ExitStatus` portably without
        // platform-specific extensions, so this test only validates the
        // Display impl by going through Debug.
        let err = StdioError::ShutdownTimeout(Duration::from_secs(2));
        let rendered = format!("{err}");
        assert!(rendered.contains("2s"), "rendered: {rendered}");
    }

    #[test]
    fn zombie_child_error_names_the_pid() {
        let err = StdioError::ZombieChild { pid: 4242 };
        let rendered = format!("{err}");
        assert!(rendered.contains("4242"), "rendered: {rendered}");
    }

    #[test]
    fn inbound_flooded_error_names_capacity_and_window() {
        let err = StdioError::InboundFlooded {
            capacity: CHANNEL_CAPACITY,
            timeout: INBOUND_FLOOD_TIMEOUT,
        };
        let rendered = format!("{err}");
        assert!(rendered.contains("1024"), "rendered: {rendered}");
        assert!(rendered.contains("2s"), "rendered: {rendered}");
    }

    #[test]
    fn empty_stderr_tail_renders_a_silent_server_note() {
        assert_eq!(
            render_stderr_tail(&[]),
            "; the server wrote nothing to stderr"
        );
    }

    #[test]
    fn stderr_tail_renders_one_indented_line_per_entry() {
        let lines = vec!["first".to_string(), "second".to_string()];
        assert_eq!(
            render_stderr_tail(&lines),
            "; last stderr lines:\n  first\n  second"
        );
    }

    #[cfg(unix)]
    #[test]
    fn startup_exit_error_names_status_and_quotes_stderr() {
        use std::os::unix::process::ExitStatusExt;
        let err = StdioError::StartupExit {
            status: ExitStatus::from_raw(7 << 8),
            stderr_tail: vec!["Error: Cannot find module '/nope.js'".to_string()],
        };
        let rendered = format!("{err}");
        assert!(
            rendered.contains("exited during startup"),
            "rendered: {rendered}"
        );
        assert!(rendered.contains("exit status: 7"), "rendered: {rendered}");
        assert!(
            rendered.contains("Cannot find module '/nope.js'"),
            "rendered: {rendered}"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn exit_status_within_reports_a_fast_exit_and_keeps_the_stderr_tail() {
        let cfg = StdioConfig {
            command: "sh".into(),
            args: vec![
                "-c".into(),
                "i=1; while [ $i -le 25 ]; do echo line$i >&2; i=$((i+1)); done; exit 3".into(),
            ],
            ..Default::default()
        };
        let (_transport, mut handle) = StdioTransport::spawn(cfg).expect("spawn sh");
        let status = handle
            .exit_status_within(Duration::from_secs(5))
            .await
            .expect("child exits well within the window");
        assert_eq!(status.code(), Some(3));
        let tail = handle.stderr_tail();
        assert_eq!(tail.len(), STDERR_TAIL_LINES, "tail: {tail:?}");
        assert_eq!(tail.first().map(String::as_str), Some("line6"));
        assert_eq!(tail.last().map(String::as_str), Some("line25"));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn exit_status_within_returns_none_while_the_child_lives() {
        let cfg = StdioConfig {
            command: "sh".into(),
            args: vec!["-c".into(), "sleep 5".into()],
            ..Default::default()
        };
        let (_transport, mut handle) = StdioTransport::spawn(cfg).expect("spawn sh");
        assert!(handle
            .exit_status_within(Duration::from_millis(100))
            .await
            .is_none());
        // kill_on_drop reaps the sleeping child when the handle drops.
    }

    #[test]
    fn malformed_json_error_includes_line() {
        let parse_err: serde_json::Error =
            serde_json::from_str::<JsonRpcMessage>("not json").expect_err("expected parse error");
        let err = StdioError::MalformedJson {
            line: "not json".into(),
            source: parse_err,
        };
        let rendered = format!("{err}");
        assert!(rendered.contains("not json"), "rendered: {rendered}");
    }
}
