//! Streaming command execution — local or over SSH.
//!
//! The single most valuable primitive both tools share. Sando's `deploy.rs`
//! shelled out to `ssh`/`rsync` with buffered `.output()`; an app-build
//! orchestrator instead needs the merged stdout+stderr streamed live (a
//! `cargo tauri build` writes progress to stderr for minutes), so this is a
//! proper streaming runner rather than a buffered one.
//!
//! A [`RemoteHost`] is either the local machine (`ssh_target == "local"` or
//! empty) or a tailnet host reached over SSH. [`RemoteHost::run_streaming`]
//! spawns the command, drains both pipes concurrently into a shared
//! [`LogSink`], and returns the exit status plus the full captured bytes so a
//! caller can still post-process the whole output (classify, grep a success
//! banner, etc.).
//!
//! This is the low-level transport primitive. The capability-gated
//! [`crate::Executor`] trait ([`crate::LocalExec`] / [`crate::SshExec`]) is the
//! layer built on top of it.

use anyhow::{Context, Result};
use async_trait::async_trait;
use std::process::{ExitStatus, Stdio};
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::process::Command;
use tokio::sync::Mutex;

/// SSH options used everywhere we shell out to ssh — fail fast, no prompts.
/// Matches Sando's original `deploy.rs` so behavior is identical across tools.
pub const SSH_FLAGS: &[&str] = &[
    "-o",
    "BatchMode=yes",
    "-o",
    "ConnectTimeout=10",
    "-o",
    "StrictHostKeyChecking=accept-new",
];

/// A sink for streamed command output. Each chunk reflects a tokio read
/// boundary — chunks are NOT line-aligned; consumers that want lines must
/// reassemble. `ops_core::live_log::LiveLog` is the canonical implementation
/// (append-to-disk + broadcast).
///
/// `#[async_trait]` keeps the trait object-safe so it can be passed as
/// `&mut dyn LogSink` into the [`crate::Executor`] trait while still being
/// usable behind an `Arc<Mutex<_>>` by [`RemoteHost::run_streaming`].
#[async_trait]
pub trait LogSink: Send {
    async fn write_chunk(&mut self, bytes: &[u8]);
}

/// A build host: the local machine or an SSH target (a tailnet alias such as
/// `mbp`, or `user@host`), optionally on a non-default SSH port.
#[derive(Debug, Clone)]
pub struct RemoteHost {
    ssh_target: String,
    /// `None` = ssh's default (22, or whatever `~/.ssh/config` says for this
    /// target). The port lives here rather than on one transport so the exec
    /// path (`ssh -p`) and the sync path (`rsync -e "ssh -p"`) can never
    /// disagree about which port a host is on.
    port: Option<u16>,
}

/// Result of a streamed run: the exit status plus a bounded recent tail of the
/// captured stdout and stderr (each already forwarded to the sink — and, for
/// the engine, the on-disk log — byte-exact as it arrived). The returned buffer
/// is only for callers that branch on output, so it is capped at
/// [`OUTPUT_TAIL_CAP`] to avoid holding a multi-GB build log in RAM.
#[derive(Debug)]
pub struct RunOutput {
    pub status: ExitStatus,
    pub stdout: Vec<u8>,
    pub stderr: Vec<u8>,
}

impl RunOutput {
    pub fn success(&self) -> bool {
        self.status.success()
    }
}

/// Cap for the in-memory capture returned in [`RunOutput`]. The live sink and
/// the on-disk log carry the byte-exact stream; this buffer is a recent tail.
pub(crate) const OUTPUT_TAIL_CAP: usize = 256 * 1024;

/// Append `chunk`, then trim the front so `buf` retains at most its last `cap`
/// bytes — a rolling tail that bounds memory for verbose, long-running commands.
pub(crate) fn push_bounded(buf: &mut Vec<u8>, chunk: &[u8], cap: usize) {
    buf.extend_from_slice(chunk);
    if buf.len() > cap {
        buf.drain(..buf.len() - cap);
    }
}

/// Rebuild an `ExitStatus` from a plain process exit code (0..=255).
///
/// On Unix the code lives in the high byte of the wait-status encoding. All a
/// recipe branches on — `.success()` and the non-zero distinction — round-trips
/// exactly.
#[cfg(unix)]
pub(crate) fn exit_status_from_code(code: i32) -> ExitStatus {
    use std::os::unix::process::ExitStatusExt;
    ExitStatus::from_raw((code & 0xff) << 8)
}

#[cfg(not(unix))]
pub(crate) fn exit_status_from_code(code: i32) -> ExitStatus {
    use std::os::windows::process::ExitStatusExt;
    ExitStatus::from_raw(code as u32)
}

/// A per-run marker used to carry a remote command's real exit code back to us
/// in its own stdout.
///
/// **Why this exists.** `ssh` is supposed to relay the remote command's exit
/// status, and a regular sshd does. Tailscale SSH does not: it closes the
/// channel without an `exit-status` message, so the local `ssh` exits 0 no
/// matter how the remote command ended. Verified 2026-07-16 — `ssh mbp 'exit
/// 42'` reports 0 while `ssh astra 'exit 42'` (regular sshd) reports 42. Neither
/// `-t`, `-T`, nor a `bash -c` wrapper changes it. That turns every remote
/// failure into a silent success, which for a release pipeline means shipping a
/// green step that produced no artifact.
///
/// So we stop trusting the transport's status and have the remote shell tell us
/// the code in-band: [`RcSentinel::wrap`] appends a `printf` of `$?` to the
/// script, and [`RcFilter`] pulls that line back out of the stream before anyone
/// sees it. When the sentinel is missing we fail closed rather than guess.
///
/// The marker carries a per-run nonce so build output cannot forge it, and the
/// real sentinel is always the last thing on stdout, so parsing is anchored to
/// the end of the stream rather than searching the whole log.
#[derive(Debug, Clone)]
pub(crate) struct RcSentinel {
    /// `__ops_exec_rc_<nonce>=`, the shell-safe body of the marker.
    body: String,
    /// `\n__ops_exec_rc_<nonce>=` — what to scan the byte stream for.
    marker: Vec<u8>,
}

impl RcSentinel {
    pub(crate) fn new() -> Self {
        let body = format!("__ops_exec_rc_{}=", nonce());
        let marker = format!("\n{body}").into_bytes();
        Self { body, marker }
    }

    /// Wrap `script` so its exit code is printed as the final line of stdout.
    ///
    /// The script runs in a **subshell**, which is load-bearing: our callers
    /// send `set -e` scripts, and an `exit`/errexit at the top level of the
    /// remote shell would take the whole shell down before the `printf` ran,
    /// costing us the sentinel on exactly the failures we care about most. A
    /// subshell contains both, and the parent (which never sets `-e`) lives to
    /// report `$?`.
    pub(crate) fn wrap(&self, script: &str) -> String {
        format!(
            "(\n{script}\n)\n__ops_exec_status=$?\nprintf '\\n{}%s\\n' \"$__ops_exec_status\"\n",
            self.body
        )
    }

    /// How many trailing bytes must be held back to be sure a complete sentinel
    /// is never split across the boundary: the marker, the digits, and the
    /// closing newline.
    fn tail_len(&self) -> usize {
        self.marker.len() + 16
    }

    /// Split a stream tail into (bytes that are real output, the parsed code).
    /// Returns `None` for the code when no complete sentinel is present.
    fn split<'a>(&self, tail: &'a [u8]) -> (&'a [u8], Option<i32>) {
        let Some(idx) = last_index_of(tail, &self.marker) else {
            return (tail, None);
        };
        let rest = &tail[idx + self.marker.len()..];
        let Some(end) = rest.iter().position(|b| *b == b'\n') else {
            return (tail, None);
        };
        // The sentinel is the last thing written; anything after it means this
        // is not the line we appended.
        if !rest[end + 1..].is_empty() {
            return (tail, None);
        }
        match std::str::from_utf8(&rest[..end])
            .ok()
            .and_then(|s| s.trim().parse::<i32>().ok())
        {
            Some(code) => (&tail[..idx], Some(code)),
            None => (tail, None),
        }
    }
}

/// A short, shell-safe, per-run token. Time plus a counter is plenty: this only
/// has to be unpredictable to the *script*, not to an attacker with our source.
fn nonce() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    let t = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_nanos() as u64);
    format!("{t:016x}{:04x}", n & 0xffff)
}

fn last_index_of(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    if needle.is_empty() || haystack.len() < needle.len() {
        return None;
    }
    (0..=haystack.len() - needle.len())
        .rev()
        .find(|&i| &haystack[i..i + needle.len()] == needle)
}

/// Strips an [`RcSentinel`] off the tail of a stdout stream while it is still
/// being streamed.
///
/// Holds back the last [`RcSentinel::tail_len`] bytes so the sentinel is never
/// forwarded to the sink (the operator's live log) or into the captured buffer
/// that callers grep for banners. Only the tail is delayed, and only until the
/// stream ends, so live output is unaffected in practice.
///
/// `None` = a transport that reports exit status honestly (local `sh -c`), where
/// there is no sentinel and every byte passes straight through.
pub(crate) struct RcFilter {
    sentinel: Option<RcSentinel>,
    hold: Vec<u8>,
}

impl RcFilter {
    pub(crate) fn new(sentinel: Option<RcSentinel>) -> Self {
        Self {
            sentinel,
            hold: Vec::new(),
        }
    }

    /// Feed a freshly-read chunk; returns the bytes that are safe to forward now.
    pub(crate) fn feed(&mut self, chunk: &[u8]) -> Vec<u8> {
        let Some(s) = &self.sentinel else {
            return chunk.to_vec();
        };
        self.hold.extend_from_slice(chunk);
        let cap = s.tail_len();
        if self.hold.len() > cap {
            let cut = self.hold.len() - cap;
            self.hold.drain(..cut).collect()
        } else {
            Vec::new()
        }
    }

    /// End of stream: returns the last real output bytes plus the parsed code.
    pub(crate) fn finish(self) -> (Vec<u8>, Option<i32>) {
        match &self.sentinel {
            None => (self.hold, None),
            Some(s) => {
                let (rest, code) = s.split(&self.hold);
                (rest.to_vec(), code)
            }
        }
    }
}

/// Decide the status to report from what the transport claimed and what the
/// remote shell actually said.
///
/// The sentinel wins whenever we have it: it is the remote command's own `$?`,
/// while `status` is only the transport's opinion of it. With no sentinel, a
/// non-zero transport status is still a real failure worth surfacing (ssh exits
/// 255 when it cannot connect, and no sentinel is written). A *successful*
/// transport status with no sentinel is the dangerous case this whole mechanism
/// exists for, so it is an error rather than a green step.
pub(crate) fn resolve_remote_status(
    host: &str,
    status: ExitStatus,
    sentinel_code: Option<i32>,
) -> Result<ExitStatus> {
    if let Some(code) = sentinel_code {
        return Ok(exit_status_from_code(code));
    }
    if !status.success() {
        return Ok(status);
    }
    anyhow::bail!(
        "{host}: the remote command produced no exit-status sentinel, and ssh reported success — \
         so its real exit code is unknown and cannot be trusted. This is what Tailscale SSH does \
         (it closes the channel without an exit-status message, making every command look like it \
          passed); it also happens if the remote shell is not POSIX or the connection dropped \
         mid-stream. Check `ssh {host} 'exit 42'`: a healthy host reports 42."
    )
}

impl RemoteHost {
    /// `"local"` (or empty) runs commands directly via `sh -c`; anything else
    /// is an SSH target, on ssh's default port until [`RemoteHost::with_port`].
    pub fn new(ssh_target: impl Into<String>) -> Self {
        Self {
            ssh_target: ssh_target.into(),
            port: None,
        }
    }

    /// Reach this host on a non-default SSH port. `None` keeps ssh's default,
    /// so a caller with an `Option<u16>` can pass it straight through.
    #[must_use]
    pub fn with_port(mut self, port: Option<u16>) -> Self {
        self.port = port;
        self
    }

    pub fn is_local(&self) -> bool {
        self.ssh_target == "local" || self.ssh_target.is_empty()
    }

    pub fn ssh_target(&self) -> &str {
        &self.ssh_target
    }

    pub fn port(&self) -> Option<u16> {
        self.port
    }

    /// The `ssh` argv this host is reached with, minus the target: the shared
    /// [`SSH_FLAGS`] plus `-p <port>` when one is set. Shared by the exec path
    /// ([`RemoteHost::command`]) and the sync path (rsync's `-e`), so the two
    /// cannot drift.
    pub(crate) fn ssh_args(&self) -> Vec<String> {
        let mut args: Vec<String> = SSH_FLAGS
            .iter()
            .map(std::string::ToString::to_string)
            .collect();
        if let Some(p) = self.port {
            args.push("-p".into());
            args.push(p.to_string());
        }
        args
    }

    /// Build the `Command` that runs `script` as a single `/bin/sh` program,
    /// locally or over SSH. The remote side runs `script` as the argument to
    /// the login shell (ssh joins argv with spaces and hands it to the shell),
    /// so multi-statement scripts and pipes work the same as locally.
    pub(crate) fn command(&self, script: &str) -> Command {
        if self.is_local() {
            let mut cmd = Command::new("sh");
            cmd.arg("-c").arg(script);
            cmd
        } else {
            let mut cmd = Command::new("ssh");
            cmd.args(self.ssh_args()).arg(&self.ssh_target).arg(script);
            cmd
        }
    }

    /// Spawn `script`, stream its merged output into `sink`, and return the
    /// exit status plus captured bytes. `kill_on_drop` means cancelling the
    /// caller's task (e.g. a newer build superseding this one) SIGKILLs the
    /// child and — for SSH — drops the connection.
    /// An SSH host cannot be trusted to report exit status (see [`RcSentinel`]),
    /// so remote scripts carry their code back in-band. Local `sh -c` reports it
    /// honestly and needs no sentinel.
    pub(crate) fn rc_sentinel(&self) -> Option<RcSentinel> {
        (!self.is_local()).then(RcSentinel::new)
    }

    /// The `Command` to run `script` on this host, plus the sentinel its output
    /// will carry (if any). The one place a script is prepared for a host, so
    /// no execution path can forget to wrap it.
    pub(crate) fn command_for(&self, script: &str) -> (Command, Option<RcSentinel>) {
        let sentinel = self.rc_sentinel();
        let script = match &sentinel {
            Some(s) => s.wrap(script),
            None => script.to_string(),
        };
        (self.command(&script), sentinel)
    }

    pub async fn run_streaming<S>(&self, script: &str, sink: Arc<Mutex<S>>) -> Result<RunOutput>
    where
        S: LogSink + Send + 'static,
    {
        let (mut cmd, sentinel) = self.command_for(script);
        let mut child = cmd
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .with_context(|| format!("spawning command on {}", self.ssh_target))?;

        // Only stdout carries the sentinel; stderr streams untouched.
        let stdout_task = tokio::spawn(drain(
            child.stdout.take(),
            sink.clone(),
            RcFilter::new(sentinel.clone()),
        ));
        let stderr_task = tokio::spawn(drain(
            child.stderr.take(),
            sink.clone(),
            RcFilter::new(None),
        ));
        let status = child.wait().await.context("waiting on child")?;
        let (stdout, code) = stdout_task.await.unwrap_or_default();
        let (stderr, _) = stderr_task.await.unwrap_or_default();
        let status = match sentinel {
            Some(_) => resolve_remote_status(&self.ssh_target, status, code)?,
            None => status,
        };
        Ok(RunOutput {
            status,
            stdout,
            stderr,
        })
    }
}

/// Drain `stream` into the shared sink and return the concatenated bytes plus
/// any exit code the sentinel carried.
async fn drain<R, S>(
    stream: Option<R>,
    sink: Arc<Mutex<S>>,
    mut filter: RcFilter,
) -> (Vec<u8>, Option<i32>)
where
    R: AsyncRead + Unpin + Send + 'static,
    S: LogSink + Send + 'static,
{
    let mut total = Vec::new();
    let Some(mut s) = stream else {
        return (total, None);
    };
    let mut buf = [0u8; 4096];
    loop {
        match s.read(&mut buf).await {
            Ok(0) | Err(_) => break,
            Ok(n) => {
                let out = filter.feed(&buf[..n]);
                if !out.is_empty() {
                    push_bounded(&mut total, &out, OUTPUT_TAIL_CAP);
                    sink.lock().await.write_chunk(&out).await;
                }
            }
        }
    }
    let (rest, code) = filter.finish();
    if !rest.is_empty() {
        push_bounded(&mut total, &rest, OUTPUT_TAIL_CAP);
        sink.lock().await.write_chunk(&rest).await;
    }
    (total, code)
}

/// Single-quote a string for safe inclusion in a `/bin/sh` command. This is
/// complete POSIX single-quoting: the result is a single-quoted literal with
/// every embedded `'` rewritten as `'\''` (close-quote, escaped quote,
/// reopen-quote). Inside single quotes the shell treats every other byte —
/// `$`, backtick, `;`, `\`, newline — literally, so no metacharacter can act
/// and there is no way to break out. Proven adversarially by
/// `sh_quote_neutralizes_injection_through_real_sh`.
pub fn sh_quote(s: &str) -> String {
    let escaped = s.replace('\'', r"'\''");
    format!("'{escaped}'")
}

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

    #[test]
    fn ssh_args_carry_the_port_and_the_shared_flags() {
        // The point of the port living on RemoteHost: the exec path and the
        // sync path build their ssh invocation from this one place, so they
        // cannot disagree about which port a host is on.
        let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
        let args = host.ssh_args();
        let p = args.iter().position(|a| a == "-p").expect("-p present");
        assert_eq!(args[p + 1], "2222");
        assert!(args.contains(&"BatchMode=yes".to_string()));

        // Default: no -p, so ssh/ssh_config picks the port.
        assert!(
            !RemoteHost::new("mbp")
                .ssh_args()
                .contains(&"-p".to_string())
        );
    }

    #[test]
    fn ported_host_puts_the_port_on_the_ssh_command() {
        let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
        let cmd = host.command("true");
        let args: Vec<String> = cmd
            .as_std()
            .get_args()
            .map(|a| a.to_string_lossy().to_string())
            .collect();
        let p = args
            .iter()
            .position(|a| a == "-p")
            .expect("-p on the ssh argv");
        assert_eq!(args[p + 1], "2222");
        // The script is still the last arg, after the target.
        assert_eq!(args.last().unwrap(), "true");
    }

    #[test]
    fn local_host_ignores_the_port() {
        // "local" runs via `sh -c`; a port is meaningless and must not leak in.
        let host = RemoteHost::new("local").with_port(Some(2222));
        let cmd = host.command("true");
        let args: Vec<String> = cmd
            .as_std()
            .get_args()
            .map(|a| a.to_string_lossy().to_string())
            .collect();
        assert_eq!(args, vec!["-c".to_string(), "true".to_string()]);
    }

    #[test]
    fn push_bounded_keeps_only_the_last_cap_bytes() {
        let mut buf = Vec::new();
        push_bounded(&mut buf, b"hello", 8);
        push_bounded(&mut buf, b"world", 8); // 10 bytes -> trim front to last 8
        assert_eq!(buf, b"lloworld");
        // A single chunk larger than the cap is itself trimmed to the tail.
        let mut buf2 = Vec::new();
        push_bounded(&mut buf2, b"0123456789", 4);
        assert_eq!(buf2, b"6789");
        // Under the cap: untouched.
        let mut buf3 = Vec::new();
        push_bounded(&mut buf3, b"ok", 8);
        assert_eq!(buf3, b"ok");
    }

    /// A simple sink that accumulates every chunk for assertions.
    #[derive(Default)]
    pub(crate) struct VecSink(pub Vec<u8>);
    #[async_trait]
    impl LogSink for VecSink {
        async fn write_chunk(&mut self, bytes: &[u8]) {
            self.0.extend_from_slice(bytes);
        }
    }

    #[test]
    fn sh_quote_escapes() {
        assert_eq!(sh_quote("hello"), "'hello'");
        assert_eq!(sh_quote("it's"), r"'it'\''s'");
    }

    /// Adversarial proof: run a crafted payload through a real `/bin/sh` and
    /// confirm it round-trips verbatim — i.e. the shell treated it as a literal
    /// and no expansion, substitution, or command injection occurred.
    #[tokio::test]
    async fn sh_quote_neutralizes_injection_through_real_sh() {
        use tokio::process::Command;
        let payloads = [
            "v'; touch /tmp/ops-exec-should-not-exist; echo '",
            "$(echo pwned)",
            "`echo pwned`",
            "a\\b\\c",
            "line1\nline2",
            "$'\\x41'",
            "'; rm -rf / #",
            "${HOME}",
            "* ? [a-z] | & ; ( ) < >",
        ];
        for payload in payloads {
            let cmd = format!("printf '%s' {}", sh_quote(payload));
            let out = Command::new("sh")
                .arg("-c")
                .arg(&cmd)
                .output()
                .await
                .unwrap();
            assert!(out.status.success(), "sh failed for {payload:?}");
            assert_eq!(
                String::from_utf8_lossy(&out.stdout),
                payload,
                "sh_quote did not round-trip {payload:?} verbatim (injection or expansion occurred)"
            );
        }
        // The injection side effect must never have happened.
        assert!(!std::path::Path::new("/tmp/ops-exec-should-not-exist").exists());
    }

    #[test]
    fn local_detection() {
        assert!(RemoteHost::new("local").is_local());
        assert!(RemoteHost::new("").is_local());
        assert!(!RemoteHost::new("mbp").is_local());
    }

    #[tokio::test]
    async fn local_run_streams_stdout_and_captures_status() {
        let host = RemoteHost::new("local");
        let sink = Arc::new(Mutex::new(VecSink::default()));
        let out = host
            .run_streaming("printf 'hello '; printf 'world'", sink.clone())
            .await
            .unwrap();
        assert!(out.success());
        assert_eq!(out.stdout, b"hello world");
        assert_eq!(sink.lock().await.0, b"hello world");
    }

    #[tokio::test]
    async fn local_run_streams_stderr_too() {
        let host = RemoteHost::new("local");
        let sink = Arc::new(Mutex::new(VecSink::default()));
        let out = host
            .run_streaming("echo out; echo err 1>&2", sink.clone())
            .await
            .unwrap();
        assert!(out.success());
        assert_eq!(out.stdout, b"out\n");
        assert_eq!(out.stderr, b"err\n");
        let seen = sink.lock().await.0.clone();
        assert!(seen.windows(4).any(|w| w == b"out\n"));
        assert!(seen.windows(4).any(|w| w == b"err\n"));
    }

    #[tokio::test]
    async fn local_run_reports_nonzero_exit() {
        let host = RemoteHost::new("local");
        let sink = Arc::new(Mutex::new(VecSink::default()));
        let out = host.run_streaming("exit 3", sink).await.unwrap();
        assert!(!out.success());
        assert_eq!(out.status.code(), Some(3));
    }

    #[test]
    fn local_gets_no_sentinel_and_ssh_does() {
        assert!(RemoteHost::new("local").rc_sentinel().is_none());
        assert!(RemoteHost::new("mbp").rc_sentinel().is_some());
    }

    #[test]
    fn each_sentinel_has_its_own_nonce() {
        // Build output cannot forge a marker it has never seen.
        assert_ne!(RcSentinel::new().body, RcSentinel::new().body);
    }

    /// Drive a wrapped script through a real `/bin/sh` (standing in for the
    /// remote login shell) and report what the sentinel carried back.
    async fn run_wrapped(script: &str) -> (Vec<u8>, Option<i32>) {
        let s = RcSentinel::new();
        let out = tokio::process::Command::new("sh")
            .arg("-c")
            .arg(s.wrap(script))
            .output()
            .await
            .unwrap();
        let mut filter = RcFilter::new(Some(s));
        let mut seen = filter.feed(&out.stdout);
        let (rest, code) = filter.finish();
        seen.extend_from_slice(&rest);
        (seen, code)
    }

    #[tokio::test]
    async fn sentinel_carries_the_real_code_through_a_shell() {
        for code in [0, 1, 42, 255] {
            let (_, got) = run_wrapped(&format!("exit {code}")).await;
            assert_eq!(
                got,
                Some(code),
                "wrapped `exit {code}` should report {code}"
            );
        }
    }

    /// The reason for the subshell: our callers send `set -e` scripts, and an
    /// errexit at the top level of the remote shell would kill it before the
    /// sentinel printed — losing the code on exactly the failures that matter.
    #[tokio::test]
    async fn set_e_failure_still_reports_its_code() {
        let (_, code) = run_wrapped("set -e; false; echo unreachable").await;
        assert_eq!(code, Some(1));
    }

    /// Same for an explicit `exit`, which the driver's `set -e; git ...` scripts
    /// reach via a failing command.
    #[tokio::test]
    async fn explicit_exit_still_reports_its_code() {
        let (out, code) = run_wrapped("echo before; exit 7").await;
        assert_eq!(code, Some(7));
        assert_eq!(out, b"before\n");
    }

    #[tokio::test]
    async fn the_sentinel_never_reaches_the_caller_or_the_log() {
        // Output with and without a trailing newline: the wrapper adds a leading
        // \n so the marker always starts a line, and stripping must put the
        // stream back exactly as the script wrote it.
        let (out, code) = run_wrapped("printf 'no trailing newline'").await;
        assert_eq!(code, Some(0));
        assert_eq!(String::from_utf8_lossy(&out), "no trailing newline");

        let (out, code) = run_wrapped("echo 'with trailing newline'").await;
        assert_eq!(code, Some(0));
        assert_eq!(String::from_utf8_lossy(&out), "with trailing newline\n");
    }

    /// A build log that prints something sentinel-shaped must not be able to
    /// pass itself off as the exit code: the nonce is unguessable and the real
    /// sentinel is always last.
    #[tokio::test]
    async fn script_output_cannot_forge_the_code() {
        let (out, code) = run_wrapped("echo '__ops_exec_rc_deadbeef=0'; exit 9").await;
        assert_eq!(
            code,
            Some(9),
            "the forged line must not be read as the code"
        );
        assert_eq!(String::from_utf8_lossy(&out), "__ops_exec_rc_deadbeef=0\n");
    }

    /// The filter holds back a tail; a sentinel split across read boundaries
    /// must still be recognized (and never leak a fragment into the log).
    #[test]
    fn sentinel_split_across_chunks_is_still_stripped() {
        let s = RcSentinel::new();
        let stream = format!("hello\n\n{}42\n", s.body).into_bytes();
        let mut filter = RcFilter::new(Some(s));
        let mut seen = Vec::new();
        for chunk in stream.chunks(3) {
            seen.extend_from_slice(&filter.feed(chunk));
        }
        let (rest, code) = filter.finish();
        seen.extend_from_slice(&rest);
        assert_eq!(code, Some(42));
        assert_eq!(String::from_utf8_lossy(&seen), "hello\n");
    }

    #[test]
    fn a_missing_sentinel_yields_no_code_and_keeps_every_byte() {
        let mut filter = RcFilter::new(Some(RcSentinel::new()));
        let mut seen = filter.feed(b"build output, no sentinel\n");
        let (rest, code) = filter.finish();
        seen.extend_from_slice(&rest);
        assert_eq!(code, None);
        assert_eq!(
            String::from_utf8_lossy(&seen),
            "build output, no sentinel\n"
        );
    }

    /// The whole point: a transport that claims success without a sentinel is
    /// reported as an error, never as a green step.
    #[test]
    fn success_without_a_sentinel_fails_closed() {
        let err = resolve_remote_status("mbp", exit_status_from_code(0), None).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("mbp"), "should name the host: {msg}");
        assert!(msg.contains("no exit-status sentinel"), "{msg}");
    }

    #[test]
    fn the_sentinel_outranks_what_the_transport_claimed() {
        // Tailscale SSH's lie: ssh says 0, the remote command really failed.
        let got = resolve_remote_status("mbp", exit_status_from_code(0), Some(42)).unwrap();
        assert_eq!(got.code(), Some(42));
        assert!(!got.success());

        // And a real success is still a success.
        let got = resolve_remote_status("mbp", exit_status_from_code(0), Some(0)).unwrap();
        assert!(got.success());
    }

    /// ssh exits 255 and writes no sentinel when it cannot connect. That is a
    /// real failure with a real code, not the ambiguous case — surface it.
    #[test]
    fn a_transport_failure_with_no_sentinel_is_reported_as_itself() {
        let got = resolve_remote_status("mbp", exit_status_from_code(255), None).unwrap();
        assert_eq!(got.code(), Some(255));
    }
}