processkit 1.0.1

//! Record/replay cassettes over the [`ProcessRunner`] seam (`record` feature).
//!
//! [`RecordReplayRunner`] closes the gap between the hand-written
//! [`ScriptedRunner`](crate::testing::ScriptedRunner) and the input-asserting
//! [`RecordingRunner`](crate::testing::RecordingRunner): run the real tool **once** with
//! the runner in *record* mode and every `Invocation → ProcessResult` pair is
//! captured to a human-diffable JSON cassette; switch to *replay* mode and the
//! cassette serves results that compare equal to the recorded ones — fast,
//! hermetic, no subprocess in CI.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, Ordering};

use serde::{Deserialize, Serialize};

use crate::command::Command;
use crate::doubles::Invocation;
use crate::error::{Error, Result};
use crate::result::{Outcome, ProcessResult};
use crate::runner::{JobRunner, ProcessRunner};

/// The on-disk format revision. Bumped if the cassette schema ever changes
/// incompatibly; loading a cassette with an unknown version fails loudly
/// instead of misreading it.
const CASSETTE_VERSION: u32 = 1;

/// The whole fixture file: a format version plus the entries in capture order.
#[derive(Debug, Serialize, Deserialize)]
struct Cassette {
    version: u32,
    entries: Vec<Entry>,
}

/// One captured `Invocation → ProcessResult` pair.
///
/// Strings are lossy UTF-8 (the cassette is a text fixture). **Only env
/// *values* are redacted** — overrides are stored as variable *names* only.
/// Everything else (`program`, `args`, `cwd`, `stdout`, `stderr`) is stored
/// **verbatim** and can carry secrets — a `--password=…` argv, a token echoed
/// to stdout — so review a cassette before committing it. `timeout` is
/// deliberately absent: it is the *command's* configuration, re-read at replay
/// time, exactly like the live runner.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Entry {
    // --- the match key ---
    program: String,
    args: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    cwd: Option<String>,
    /// FNV-1a digest of the stdin *source identity* — keyed so two invocations
    /// differing only in stdin don't collide on replay. In-memory bytes hash
    /// their content; a `from_file` source hashes its **path** (the file is not
    /// read at key time, so changing the file's bytes does not change the key).
    /// One-shot streaming sources (`from_reader`/`from_lines`) are rejected by
    /// record/replay — their bytes can't be keyed — so this digest only ever
    /// describes a replayable source.
    /// `None` for empty/absent stdin. An older cassette recorded *with* stdin
    /// but no digest loads this as `None` and must be re-recorded to match a
    /// stdin invocation again. See `Stdin::content_digest` for the hashing.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    stdin_digest: Option<u64>,
    // --- stored for visibility, not matched on ---
    /// Whether stdin was supplied (human-readable; matching uses `stdin_digest`).
    #[serde(default, skip_serializing_if = "is_false")]
    has_stdin: bool,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    env_names: Vec<String>,
    // --- the captured output ---
    stdout: String,
    stderr: String,
    code: Option<i32>,
    #[serde(default, skip_serializing_if = "is_false")]
    timed_out: bool,
    // Signal number for Signalled outcomes; absent for Exited/TimedOut and in
    // cassettes written before this field was added (loaded as None).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    signal: Option<i32>,
}

#[allow(clippy::trivially_copy_pass_by_ref)] // signature dictated by serde
fn is_false(b: &bool) -> bool {
    !*b
}

/// Write `json` to `path`, restricting the file to owner-only (`0600`) on Unix.
///
/// A cassette redacts env *values* (it stores names only), but argv, cwd,
/// stdout, and stderr are stored **verbatim** — any of which can carry a secret.
/// So the file is created owner-only rather than inheriting a world-readable
/// umask.
///
/// On Unix the open also refuses to follow a symlink at `path` (`O_NOFOLLOW`),
/// so a planted `cassette.json` symlink can't redirect the secret-bearing write
/// (and the `0600`) onto the link's target — it fails loud (`ELOOP`) instead. On
/// Windows the file inherits the directory ACL (the unit of access control
/// there); **restrict the containing directory** (or use a per-user temp dir,
/// not a world-writable shared one) if the fixture can carry secrets.
fn write_cassette(path: &Path, json: &str) -> std::io::Result<()> {
    #[cfg(unix)]
    {
        use std::io::Write;
        use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
        // `mode(0o600)` applies only at *creation*. A pre-existing (possibly
        // world-readable) cassette being rewritten keeps its old perms through
        // the truncating `open`, so tighten the fd *before* writing the content,
        // never holding it at loose perms. `O_NOFOLLOW`: never write through a
        // symlink planted at `path`.
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .mode(0o600)
            .custom_flags(libc::O_NOFOLLOW)
            .open(path)?;
        file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
        file.write_all(json.as_bytes())?;
        Ok(())
    }
    #[cfg(not(unix))]
    {
        std::fs::write(path, json)
    }
}

impl Entry {
    /// Capture one record-mode call. Lossy UTF-8 throughout — see the type doc.
    fn from_parts(
        invocation: &Invocation,
        result: &ProcessResult<String>,
        stdin_digest: Option<u64>,
    ) -> Self {
        let mut env_names: Vec<String> = invocation
            .envs
            .iter()
            .map(|(name, _value)| name.to_string_lossy().into_owned())
            .collect();
        // Sorted + deduped: stable diffs, and repeated overrides of one var
        // are one fact ("this var shaped the run"), not a sequence.
        env_names.sort();
        env_names.dedup();
        Self {
            program: invocation.program.to_string_lossy().into_owned(),
            args: invocation
                .args
                .iter()
                .map(|a| a.to_string_lossy().into_owned())
                .collect(),
            cwd: invocation
                .cwd
                .as_ref()
                .map(|c| c.to_string_lossy().into_owned()),
            stdin_digest,
            has_stdin: invocation.has_stdin,
            env_names,
            stdout: result.stdout().clone(),
            stderr: result.stderr().to_owned(),
            code: result.code(),
            timed_out: result.timed_out(),
            signal: match result.outcome() {
                Outcome::Signalled(s) => s,
                _ => None,
            },
        }
    }
}

/// What an invocation is matched on: program + args + cwd + the stdin source
/// digest (content for in-memory bytes, path for a `from_file` source).
/// Env overrides are excluded so an irrelevant env difference between the
/// record and replay environments can't cause a spurious miss.
///
/// The string components are *lossy* UTF-8 decodes, so two distinct non-UTF-8
/// invocations that differ only in their invalid bytes produce the same key and
/// collide on replay (the first recorded one answers for both). Accepted: keying
/// on raw bytes would defeat the human-diffable text fixture, and valid-UTF-8
/// invocations (the common case) never collide.
type Key = (String, Vec<String>, Option<String>, bool, Option<u64>);

/// The stdin source digest keyed into a cassette match — `None` for an
/// empty/absent stdin. The digest never persists the stdin payload: in-memory
/// bytes hash their content, a `from_file` source hashes its path.
fn stdin_digest_of(command: &Command) -> Option<u64> {
    command
        .stdin_source()
        .filter(|s| !s.is_empty())
        .map(|s| s.content_digest())
}

/// Reject a one-shot streaming stdin source (`from_reader`/`from_lines`) in
/// record/replay. Such a source's bytes are consumed lazily and never captured
/// into the match key — `content_digest` can only hash a constant discriminant
/// for them — so two invocations differing *only* in streamed stdin would
/// collide on one cassette key and silently replay each other's recording.
/// Failing loud (consistent with the streaming `start` half already returning
/// [`Error::Unsupported`]) is safer than a silent wrong answer; use a replayable
/// source (`from_bytes`/`from_string`/`from_file`) for a recordable invocation.
fn reject_unrecordable_stdin(command: &Command) -> Result<()> {
    if command.stdin_source().is_some_and(|s| s.is_one_shot()) {
        return Err(Error::Unsupported {
            operation: "cassette record/replay with one-shot streaming stdin \
                        (from_reader/from_lines); use from_bytes/from_string/from_file"
                .to_string(),
        });
    }
    Ok(())
}

/// The key of a live invocation — must decode exactly like
/// [`key_of_entry`] (both sides go through the same lossy conversion). The
/// `stdin_digest` is computed from the command, not carried on the
/// [`Invocation`] (which records only *whether* stdin was supplied). The
/// `has_stdin` bool is keyed alongside the digest so an older entry that loads
/// `stdin_digest: None` regardless of its stored `has_stdin` cannot match a
/// no-stdin replay — only miss.
fn key_of(invocation: &Invocation, stdin_digest: Option<u64>) -> Key {
    (
        invocation.program.to_string_lossy().into_owned(),
        invocation
            .args
            .iter()
            .map(|a| a.to_string_lossy().into_owned())
            .collect(),
        invocation
            .cwd
            .as_ref()
            .map(|c| c.to_string_lossy().into_owned()),
        invocation.has_stdin,
        stdin_digest,
    )
}

/// The key of a stored entry (already lossy strings).
fn key_of_entry(entry: &Entry) -> Key {
    (
        entry.program.clone(),
        entry.args.clone(),
        entry.cwd.clone(),
        entry.has_stdin,
        entry.stdin_digest,
    )
}

/// The replay-side state for one key: its entries in capture order plus a
/// cursor implementing the order-then-repeat-last consumption.
#[derive(Debug)]
struct ReplaySlot {
    entries: Vec<Entry>,
    next: usize,
}

impl ReplaySlot {
    /// The entry for this call: in capture order while they last, then the
    /// last one forever — so a sequence of differing outputs replays
    /// faithfully, and a retry/probe loop that re-runs the command after the
    /// sequence is exhausted still gets a stable answer.
    fn play(&mut self) -> &Entry {
        let index = self.next.min(self.entries.len() - 1);
        self.next = self.next.saturating_add(1);
        &self.entries[index]
    }
}

enum Mode<R> {
    Record {
        inner: R,
        path: PathBuf,
        recorded: Mutex<Vec<Entry>>,
        /// Runs recorded since the last successful save — the drop-time flush
        /// fires only when there is something unwritten, so a save-then-record
        /// sequence can't silently lose the late runs.
        dirty: AtomicBool,
    },
    Replay {
        slots: Mutex<HashMap<Key, ReplaySlot>>,
    },
}

/// A [`ProcessRunner`] that records real runs to a JSON cassette, or replays a
/// cassette hermetically (`record` feature).
///
/// **Record** mode wraps a real inner runner, captures each completed call's
/// invocation and result, and writes the cassette on [`save`](Self::save) (or
/// best-effort on drop). Errors (spawn failure, …) record nothing; non-zero
/// exits and captured timeouts are results and are recorded.
///
/// **Replay** mode loads the cassette and serves results without spawning:
///
/// - **Matching**: program + args + cwd + stdin source digest. Env override
///   *values* are never written — only sorted variable names. Everything else
///   (argv, cwd, stdout, stderr) is stored verbatim, so review fixtures
///   before committing. File is written owner-only (`0600`) on Unix.
/// - **Duplicates** replay in capture order, then the last entry repeats.
/// - **A miss is [`Error::CassetteMiss`]** (not `is_not_found()`): never a
///   surprise subprocess.
/// - The replayed result carries the *replaying* command's
///   [`timeout`](Command::timeout), so a recorded timed-out run surfaces as
///   [`Error::Timeout`](crate::Error::Timeout) with the real deadline.
/// - Covers **`output_string`** only; `start` returns
///   [`Error::Unsupported`](crate::Error::Unsupported).
///
/// Non-UTF-8 programs/args/paths are stored lossily; both sides apply the same
/// conversion, so matching still works. Two distinct non-UTF-8 invocations that
/// differ only in invalid bytes share the same key and collide on replay.
///
/// [`save`](Self::save) is the explicit write; drop flushes best-effort except
/// while unwinding (a panic never silently persists a cassette).
pub struct RecordReplayRunner<R: ProcessRunner = JobRunner> {
    mode: Mode<R>,
}

impl<R: ProcessRunner> RecordReplayRunner<R> {
    /// Record every run through `inner`, to be written to `path` as a JSON
    /// cassette by [`save`](Self::save) (or best-effort when the runner
    /// drops). Nothing touches the filesystem until then.
    pub fn record(path: impl Into<PathBuf>, inner: R) -> Self {
        Self {
            mode: Mode::Record {
                inner,
                path: path.into(),
                recorded: Mutex::new(Vec::new()),
                dirty: AtomicBool::new(false),
            },
        }
    }

    /// Write the cassette now (record mode). This is the error-surfacing path
    /// — the drop-time flush swallows failures. Idempotent (rewrites the full
    /// cassette each time); a no-op `Ok` in replay mode. Runs recorded *after*
    /// a save are still covered: the drop-time flush fires whenever anything
    /// was recorded since the last successful save.
    pub fn save(&self) -> Result<()> {
        let Mode::Record {
            path,
            recorded,
            dirty,
            ..
        } = &self.mode
        else {
            return Ok(());
        };
        // Hold the entries lock until `dirty` is cleared, so a run recorded
        // concurrently with the save can't be marked clean without being in
        // the written file (it blocks, then lands as dirty again).
        // `expect`, not poison-recovery: no user code ever runs under the
        // cassette locks, so poisoning is a logic bug worth failing loudly on.
        let entries = recorded.lock().expect("cassette mutex poisoned");
        let cassette = Cassette {
            version: CASSETTE_VERSION,
            entries: entries.clone(),
        };
        let json = serde_json::to_string_pretty(&cassette)
            .map_err(|e| Error::Io(std::io::Error::from(e)))?;
        write_cassette(path, &json).map_err(Error::Io)?;
        dirty.store(false, Ordering::SeqCst);
        Ok(())
    }
}

/// Reject a cassette entry whose outcome fields *contradict* each other.
/// The decode model is: `timed_out` → `TimedOut`; else `code` present →
/// `Exited`; else → `Signalled(signal)` (with `signal` optionally absent, i.e.
/// "killed, signal unknown"). So at most one of `code` / `timed_out` / `signal`
/// may be set — an entry that sets two or more (e.g. both `code` and `signal`)
/// is malformed: the decoder would silently pick one and drop the rest. Fail
/// loud on load, like an unknown `version` does. (An entry that sets *none* is
/// the legitimate `Signalled(None)` and is allowed.)
fn validate_entry_outcome(entry: &Entry) -> Result<()> {
    let indicators = usize::from(entry.code.is_some())
        + usize::from(entry.timed_out)
        + usize::from(entry.signal.is_some());
    if indicators > 1 {
        return Err(Error::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!(
                "cassette entry for `{}` has a contradictory outcome: at most one of \
                 `code` (exited), `timed_out`, or `signal` (signalled) may be set — found {indicators}",
                entry.program
            ),
        )));
    }
    Ok(())
}

impl RecordReplayRunner<JobRunner> {
    /// Load the cassette at `path` and serve its entries hermetically — no
    /// subprocess is ever spawned in replay mode.
    ///
    /// Errors are [`Error::Io`]: a missing file keeps its `NotFound` kind; a
    /// corrupt file or an unknown format `version` is `InvalidData`.
    pub fn replay(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();
        const MAX_CASSETTE_BYTES: u64 = 64 << 20; // 64 MiB
        if let Ok(meta) = std::fs::metadata(path)
            && meta.len() > MAX_CASSETTE_BYTES
        {
            return Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!(
                    "cassette is {} bytes, over the {MAX_CASSETTE_BYTES}-byte limit",
                    meta.len()
                ),
            )));
        }
        let text = std::fs::read_to_string(path).map_err(Error::Io)?;
        let cassette: Cassette =
            serde_json::from_str(&text).map_err(|e| Error::Io(std::io::Error::from(e)))?;
        if cassette.version != CASSETTE_VERSION {
            return Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!(
                    "cassette version {} is not supported (this build reads version {CASSETTE_VERSION})",
                    cassette.version
                ),
            )));
        }
        let mut slots: HashMap<Key, ReplaySlot> = HashMap::new();
        for entry in cassette.entries {
            validate_entry_outcome(&entry)?;
            slots
                .entry(key_of_entry(&entry))
                .or_insert_with(|| ReplaySlot {
                    entries: Vec::new(),
                    next: 0,
                })
                .entries
                .push(entry);
        }
        Ok(Self {
            mode: Mode::Replay {
                slots: Mutex::new(slots),
            },
        })
    }
}

#[async_trait::async_trait]
impl<R: ProcessRunner> ProcessRunner for RecordReplayRunner<R> {
    async fn output_string(&self, command: &Command) -> Result<ProcessResult<String>> {
        reject_unrecordable_stdin(command)?;
        match &self.mode {
            Mode::Record {
                inner,
                recorded,
                dirty,
                ..
            } => {
                let result = inner.output_string(command).await?;
                let invocation = Invocation::from_command(command);
                let stdin_digest = stdin_digest_of(command);
                let mut entries = recorded.lock().expect("cassette mutex poisoned");
                entries.push(Entry::from_parts(&invocation, &result, stdin_digest));
                dirty.store(true, Ordering::SeqCst);
                Ok(result)
            }
            Mode::Replay { slots } => {
                let invocation = Invocation::from_command(command);
                let stdin_digest = stdin_digest_of(command);
                // Release the lock before invoking line handlers — a handler that
                // re-enters this replayer would otherwise deadlock.
                let entry = {
                    let mut slots = slots.lock().expect("cassette mutex poisoned");
                    let Some(slot) = slots.get_mut(&key_of(&invocation, stdin_digest)) else {
                        return Err(Error::CassetteMiss {
                            program: command.program_name(),
                        });
                    };
                    slot.play().clone()
                };
                crate::doubles::replay_line_handlers(command, &entry.stdout, &entry.stderr);
                let outcome = match (entry.code, entry.timed_out) {
                    (_, true) => Outcome::TimedOut,
                    (Some(code), false) => Outcome::Exited(code),
                    (None, false) => Outcome::Signalled(entry.signal),
                };
                Ok(ProcessResult::new(
                    entry.program,
                    entry.stdout,
                    entry.stderr,
                    outcome,
                    command.configured_timeout(),
                )
                .with_ok_codes(command.ok_codes_vec()))
            }
        }
    }
}

// Manual: no `R: Debug` bound; entries/slots are summarized as counts.
impl<R: ProcessRunner> std::fmt::Debug for RecordReplayRunner<R> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.mode {
            Mode::Record {
                path,
                recorded,
                dirty,
                ..
            } => f
                .debug_struct("RecordReplayRunner::Record")
                .field("path", path)
                .field(
                    "recorded",
                    &recorded.lock().expect("cassette mutex poisoned").len(),
                )
                .field("dirty", &dirty.load(Ordering::SeqCst))
                .finish_non_exhaustive(),
            Mode::Replay { slots } => f
                .debug_struct("RecordReplayRunner::Replay")
                .field(
                    "keys",
                    &slots.lock().expect("cassette mutex poisoned").len(),
                )
                .finish_non_exhaustive(),
        }
    }
}

impl<R: ProcessRunner> Drop for RecordReplayRunner<R> {
    fn drop(&mut self) {
        // Best-effort flush; skip while unwinding so a panic never silently
        // persists a cassette that may carry secrets in argv/stdout.
        if let Mode::Record { dirty, .. } = &self.mode
            && dirty.load(Ordering::SeqCst)
            && !std::thread::panicking()
        {
            let _ = self.save();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::doubles::{Reply, ScriptedRunner};
    use crate::result::Outcome;
    use crate::runner::ProcessRunnerExt;
    use std::time::Duration;

    /// A scripted inner runner standing in for the real tool.
    fn scripted() -> ScriptedRunner {
        ScriptedRunner::new()
            .on(["tool", "--version"], Reply::ok("tool 1.2.3\n"))
            .on(["tool", "fail"], Reply::fail(7, "boom"))
    }

    fn temp_cassette() -> (tempfile::TempDir, PathBuf) {
        let dir = tempfile::tempdir().expect("create temp dir");
        let path = dir.path().join("cassette.json");
        (dir, path)
    }

    #[cfg(unix)]
    #[test]
    fn write_cassette_refuses_to_follow_a_symlink() {
        let dir = tempfile::tempdir().expect("temp dir");
        let target = dir.path().join("victim.txt");
        std::fs::write(&target, "original").expect("seed victim");
        let link = dir.path().join("cassette.json");
        std::os::unix::fs::symlink(&target, &link).expect("create symlink");

        let err = write_cassette(&link, "{\"secret\":true}")
            .expect_err("writing through a symlink must fail (O_NOFOLLOW)");
        assert_eq!(
            err.raw_os_error(),
            Some(libc::ELOOP),
            "O_NOFOLLOW on a symlink yields ELOOP, got {err:?}"
        );
        assert_eq!(
            std::fs::read_to_string(&target).expect("read victim"),
            "original",
            "the victim file must be untouched"
        );
    }

    #[tokio::test]
    async fn round_trip_is_identical() {
        let (_dir, path) = temp_cassette();

        let recorder = RecordReplayRunner::record(&path, scripted());
        let ok = recorder
            .output_string(&Command::new("tool").arg("--version"))
            .await
            .expect("record ok run");
        let fail = recorder
            .output_string(&Command::new("tool").arg("fail"))
            .await
            .expect("record failing run (non-zero exit is a result, not Err)");
        recorder.save().expect("save cassette");

        let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
        let ok2 = replayer
            .output_string(&Command::new("tool").arg("--version"))
            .await
            .expect("replay ok run");
        let fail2 = replayer
            .output_string(&Command::new("tool").arg("fail"))
            .await
            .expect("replay failing run");
        assert_eq!(ok, ok2, "replay must be identical to the recording");
        assert_eq!(fail, fail2);
        assert_eq!(fail2.code(), Some(7));
        assert_eq!(fail2.stderr(), "boom");
    }

    #[tokio::test]
    async fn duplicate_key_plays_in_order_then_repeats_last() {
        let (_dir, path) = temp_cassette();

        let json = serde_json::json!({
            "version": 1,
            "entries": [
                {
                    "program": "git", "args": ["head"],
                    "stdout": "aaa", "stderr": "", "code": 0
                },
                {
                    "program": "git", "args": ["head"],
                    "stdout": "bbb", "stderr": "", "code": 0
                }
            ]
        });
        std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap();

        let cmd = Command::new("git").arg("head");
        let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
        let first = replayer.run(&cmd).await.expect("first replay");
        let second = replayer.run(&cmd).await.expect("second replay");
        let third = replayer.run(&cmd).await.expect("third replay repeats last");
        assert_eq!(first, "aaa");
        assert_eq!(second, "bbb");
        assert_eq!(third, "bbb", "exhausted key must repeat the last entry");
    }

    #[tokio::test]
    async fn replay_rejects_an_entry_with_contradictory_outcome() {
        let (_dir, path) = temp_cassette();
        let json = serde_json::json!({
            "version": 1,
            "entries": [
                { "program": "x", "args": [], "stdout": "", "stderr": "", "code": 0, "signal": 9 }
            ]
        });
        std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap();
        let err = RecordReplayRunner::replay(&path)
            .expect_err("a contradictory outcome must be rejected");
        assert!(
            matches!(&err, Error::Io(e) if e.kind() == std::io::ErrorKind::InvalidData),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn replay_miss_is_a_distinct_cassette_miss_error() {
        let (_dir, path) = temp_cassette();
        let recorder = RecordReplayRunner::record(&path, scripted());
        let _ = recorder
            .output_string(&Command::new("tool").arg("--version"))
            .await
            .expect("record");
        recorder.save().expect("save");

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let err = replayer
            .output_string(&Command::new("tool").arg("--other"))
            .await
            .expect_err("an unrecorded invocation must not be served");
        match &err {
            Error::CassetteMiss { program } => assert_eq!(program, "tool"),
            other => panic!("expected Error::CassetteMiss, got {other:?}"),
        }
        // A stale cassette is NOT mistaken for a missing program.
        assert!(
            !err.is_not_found(),
            "a cassette miss must not read as not-found: {err:?}"
        );
    }

    #[tokio::test]
    async fn replay_invokes_line_handlers() {
        let (_dir, path) = temp_cassette();
        let recorder = RecordReplayRunner::record(&path, scripted());
        let _ = recorder
            .output_string(&Command::new("tool").arg("--version"))
            .await
            .expect("record");
        recorder.save().expect("save");

        let seen = std::sync::Arc::new(Mutex::new(Vec::new()));
        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let cmd = Command::new("tool").arg("--version").on_stdout_line({
            let seen = seen.clone();
            move |l| seen.lock().unwrap().push(l.to_owned())
        });
        let _ = replayer.output_string(&cmd).await.expect("replay");
        assert_eq!(
            *seen.lock().unwrap(),
            ["tool 1.2.3"],
            "replay must invoke the command's line handler"
        );
    }

    #[tokio::test]
    async fn stdin_content_is_part_of_the_match_key() {
        let (_dir, path) = temp_cassette();
        let inner = ScriptedRunner::new()
            .on_sequence(["tool"], [Reply::ok("out-A\n"), Reply::ok("out-B\n")]);
        let recorder = RecordReplayRunner::record(&path, inner);
        let _ = recorder
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_string("A")))
            .await
            .expect("record A");
        let _ = recorder
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_string("B")))
            .await
            .expect("record B");
        recorder.save().expect("save");

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        // Replay B FIRST: with stdin in the key it gets out-B. Keying on
        // `has_stdin` alone would collide and return out-A (the first entry).
        let b = replayer
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_string("B")))
            .await
            .expect("replay B");
        assert_eq!(
            b.stdout(),
            "out-B\n",
            "stdin B must replay its own recording"
        );
        let a = replayer
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_string("A")))
            .await
            .expect("replay A");
        assert_eq!(
            a.stdout(),
            "out-A\n",
            "stdin A must replay its own recording"
        );
    }

    #[tokio::test]
    async fn one_shot_streaming_stdin_is_rejected_in_both_modes() {
        let (_dir, path) = temp_cassette();
        let inner = ScriptedRunner::new().fallback(Reply::ok("out\n"));
        let recorder = RecordReplayRunner::record(&path, inner);
        let err = recorder
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_reader(&b"payload"[..])))
            .await
            .expect_err("record must reject a one-shot streaming stdin");
        assert!(matches!(err, Error::Unsupported { .. }), "got {err:?}");

        // Record a plain entry so the cassette loads, then prove replay rejects
        // a streaming stdin too.
        let _ = recorder
            .output_string(&Command::new("tool"))
            .await
            .expect("record a replayable entry");
        recorder.save().expect("save");
        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let err = replayer
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_reader(&b"payload"[..])))
            .await
            .expect_err("replay must reject a one-shot streaming stdin");
        assert!(matches!(err, Error::Unsupported { .. }), "got {err:?}");
    }

    #[tokio::test]
    async fn no_stdin_replay_does_not_match_a_stdin_recorded_entry() {
        let (_dir, path) = temp_cassette();
        let recorder =
            RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("out\n")));
        let _ = recorder
            .output_string(&Command::new("tool").stdin(crate::Stdin::from_string("input")))
            .await
            .expect("record with stdin");
        recorder.save().expect("save");

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let err = replayer
            .output_string(&Command::new("tool"))
            .await
            .expect_err("a no-stdin call must not match a stdin-recorded entry");
        assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
    }

    #[tokio::test]
    async fn replayed_timeout_carries_the_commands_deadline() {
        let (_dir, path) = temp_cassette();
        let recorder = RecordReplayRunner::record(
            &path,
            ScriptedRunner::new().on(["tool", "slow"], Reply::timeout()),
        );
        let _ = recorder
            .output_string(&Command::new("tool").arg("slow"))
            .await
            .expect("a captured timeout is a result, not an Err");
        recorder.save().expect("save");

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let err = replayer
            .run(
                &Command::new("tool")
                    .arg("slow")
                    .timeout(Duration::from_secs(7)),
            )
            .await
            .expect_err("run() raises the captured timeout");
        match err {
            Error::Timeout { timeout, .. } => assert_eq!(timeout, Duration::from_secs(7)),
            other => panic!("expected Error::Timeout, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn env_values_never_reach_the_file() {
        let (_dir, path) = temp_cassette();
        let recorder =
            RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("done")));
        let _ = recorder
            .output_string(
                &Command::new("tool")
                    .env("API_TOKEN", "hunter2-very-secret")
                    .env("MODE", "fast"),
            )
            .await
            .expect("record");
        recorder.save().expect("save");

        let json = std::fs::read_to_string(&path).expect("read cassette");
        assert!(json.contains("API_TOKEN"), "names are stored: {json}");
        assert!(json.contains("MODE"));
        assert!(
            !json.contains("hunter2-very-secret") && !json.contains("fast"),
            "values must never be written: {json}"
        );

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let out = replayer
            .run(&Command::new("tool"))
            .await
            .expect("env is not part of the match key");
        assert_eq!(out, "done");
    }

    #[tokio::test]
    async fn signal_number_survives_round_trip() {
        let (_dir, path) = temp_cassette();
        let json = r#"{"version":1,"entries":[{"program":"tool","args":[],"stdout":"","stderr":"","code":null,"signal":9}]}"#;
        std::fs::write(&path, json).expect("write cassette");

        let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
        let result = replayer
            .output_string(&Command::new("tool"))
            .await
            .expect("replay");
        assert_eq!(result.outcome(), Outcome::Signalled(Some(9)));
    }

    #[tokio::test]
    async fn cassette_without_signal_field_loads_as_signalled_none() {
        let (_dir, path) = temp_cassette();
        let json = r#"{"version":1,"entries":[{"program":"tool","args":[],"stdout":"","stderr":"","code":null}]}"#;
        std::fs::write(&path, json).expect("write cassette");

        let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
        let result = replayer
            .output_string(&Command::new("tool"))
            .await
            .expect("replay");
        assert_eq!(result.outcome(), Outcome::Signalled(None));
    }

    #[tokio::test]
    async fn load_errors_are_typed_io() {
        let (_dir, path) = temp_cassette();
        match RecordReplayRunner::replay(&path) {
            Err(Error::Io(e)) => assert_eq!(e.kind(), std::io::ErrorKind::NotFound),
            other => panic!("expected Io(NotFound), got {other:?}"),
        }

        std::fs::write(&path, "{ not json").unwrap();
        match RecordReplayRunner::replay(&path) {
            Err(Error::Io(e)) => assert_eq!(e.kind(), std::io::ErrorKind::InvalidData),
            other => panic!("expected Io(InvalidData), got {other:?}"),
        }

        std::fs::write(&path, r#"{ "version": 99, "entries": [] }"#).unwrap();
        match RecordReplayRunner::replay(&path) {
            Err(Error::Io(e)) => {
                assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
                assert!(e.to_string().contains("version 99"), "got: {e}");
            }
            other => panic!("expected Io(InvalidData), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn drop_without_save_flushes_best_effort() {
        let (_dir, path) = temp_cassette();
        {
            let recorder = RecordReplayRunner::record(&path, scripted());
            let _ = recorder
                .output_string(&Command::new("tool").arg("--version"))
                .await
                .expect("record");
        }
        let replayer = RecordReplayRunner::replay(&path).expect("dropped recorder left a cassette");
        let out = replayer
            .run(&Command::new("tool").arg("--version"))
            .await
            .expect("replay after drop-flush");
        assert_eq!(out, "tool 1.2.3");
    }

    #[tokio::test]
    async fn cwd_is_part_of_the_match_key() {
        let (_dir, path) = temp_cassette();
        let recorder =
            RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("from-a")));
        let _ = recorder
            .output_string(&Command::new("tool").current_dir("dir-a"))
            .await
            .expect("record in dir-a");
        recorder.save().expect("save");

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let err = replayer
            .output_string(&Command::new("tool").current_dir("dir-b"))
            .await
            .expect_err("a different cwd is a different invocation");
        assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
        let err = replayer
            .output_string(&Command::new("tool"))
            .await
            .expect_err("a missing cwd is a different invocation too");
        assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
        let out = replayer
            .run(&Command::new("tool").current_dir("dir-a"))
            .await
            .expect("the recorded cwd matches");
        assert_eq!(out, "from-a");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn cassette_file_is_written_owner_only() {
        use std::os::unix::fs::PermissionsExt;
        let (_dir, path) = temp_cassette();
        let recorder = RecordReplayRunner::record(&path, scripted());
        let _ = recorder
            .output_string(&Command::new("tool").arg("--version"))
            .await
            .expect("record");
        recorder.save().expect("save");

        let mode = std::fs::metadata(&path)
            .expect("stat cassette")
            .permissions()
            .mode();
        assert_eq!(
            mode & 0o777,
            0o600,
            "cassette must be owner-only, got {:o}",
            mode & 0o777
        );
    }

    #[tokio::test]
    async fn drop_while_unwinding_does_not_persist_a_surprise_cassette() {
        let (_dir, path) = temp_cassette();
        let recorder = RecordReplayRunner::record(&path, scripted());
        let _ = recorder
            .output_string(&Command::new("tool").arg("--version"))
            .await
            .expect("record (now dirty, unsaved)");

        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
            let _hold = recorder;
            panic!("boom mid-recording");
        }));
        assert!(outcome.is_err(), "the scope must have panicked");
        assert!(
            !path.exists(),
            "a recorder dropped during unwind must not persist a cassette: {path:?}"
        );
    }

    #[tokio::test]
    async fn save_then_record_more_then_drop_flushes_the_late_runs() {
        let (_dir, path) = temp_cassette();
        {
            let recorder = RecordReplayRunner::record(&path, scripted());
            let _ = recorder
                .output_string(&Command::new("tool").arg("--version"))
                .await
                .expect("record first");
            recorder.save().expect("first save");
            let _ = recorder
                .output_string(&Command::new("tool").arg("fail"))
                .await
                .expect("record second");
        }
        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let result = replayer
            .output_string(&Command::new("tool").arg("fail"))
            .await
            .expect("the post-save run was flushed by drop");
        assert_eq!(result.code(), Some(7));
    }

    #[tokio::test]
    async fn non_utf8_args_are_recorded_lossily_not_fatally() {
        // A program argument that is not valid Unicode, per platform.
        #[cfg(unix)]
        let bad = {
            use std::os::unix::ffi::OsStringExt;
            std::ffi::OsString::from_vec(vec![b'a', 0xFF, b'b'])
        };
        #[cfg(windows)]
        let bad = {
            use std::os::windows::ffi::OsStringExt;
            // A lone surrogate is valid UTF-16-ish for OsString but not Unicode.
            std::ffi::OsString::from_wide(&[0x61, 0xD800, 0x62])
        };

        let (_dir, path) = temp_cassette();
        let recorder =
            RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("ok")));
        let cmd = Command::new("tool").arg(&bad);
        let _ = recorder.output_string(&cmd).await.expect("record lossily");
        recorder.save().expect("save");

        let replayer = RecordReplayRunner::replay(&path).expect("load");
        let out = replayer.run(&cmd).await.expect("replay matches lossily");
        assert_eq!(out, "ok");
    }
}