truth-mirror 0.13.3

//! Integration tests for the queue-tied watcher lifecycle: `watch --until-empty`,
//! `ensure-watcher` (spawn / no-op / stale recovery / race), and the end-to-end
//! enqueue → watcher appears → queue drains → watcher exits after grace path.
//!
//! Every test uses a temporary state dir; none touch any real `.truth` state.
//! The spawn/liveness tests are unix-gated (they rely on process-group detach and
//! the pid liveness probe); the Windows spawn shim is compile-checked via cfg but
//! not exercised here.

mod common;

use std::{
    fs,
    path::Path,
    process::Command as StdCommand,
    time::{Duration, Instant},
};

use common::{git, git_stdout, make_executable, truth_mirror};
use predicates::prelude::*;
use truth_mirror::watcher::{self, LockClaim};

/// Write an executable mock `codex` reviewer that always REJECTs into `bin_dir`
/// and return a `PATH` value with that dir prepended.
fn mock_reviewer_path(bin_dir: &Path) -> String {
    let fake_codex = bin_dir.join("codex");
    fs::write(
        &fake_codex,
        r#"#!/usr/bin/env python3
import json
print(json.dumps({
    "verdict": "REJECT",
    "summary": "The claim is not substantiated.",
    "findings": [{
        "severity": "high",
        "title": "claim not substantiated",
        "body": "The cited evidence does not prove the claim.",
        "file": "file.txt",
        "line_start": 1,
        "line_end": 1,
        "confidence": 95,
        "recommendation": "Add evidence that directly proves the claim."
    }],
    "next_steps": ["Re-run the verification command."],
    "memory_skill": {
        "kind": "anti_pattern_skill",
        "learning_source": "claim not substantiated",
        "reasoning": "The rejection identifies a reusable unsupported-claim pattern."
    }
}))
"#,
    )
    .unwrap();
    make_executable(&fake_codex);
    format!(
        "{}:{}",
        bin_dir.display(),
        std::env::var("PATH").unwrap_or_default()
    )
}

/// A pid guaranteed to be dead: spawn a trivial process, reap it, return its
/// pid. Copilot (PR #13, watcher_cli.rs ~66): spawning the external `true`
/// binary isn't portable (notably absent on Windows). Spawning this test
/// binary itself with `--help` (which the standard libtest harness
/// understands and exits 0 for) is a guaranteed-dead-pid source with no
/// external command dependency.
fn reaped_pid() -> u32 {
    let mut child = StdCommand::new(std::env::current_exe().unwrap())
        .arg("--help")
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
        .unwrap();
    let pid = child.id();
    child.wait().unwrap();
    pid
}

/// Create a git repo with one claim-bearing commit and enqueue it for review the
/// same way the post-commit hook does. Returns the committed SHA.
fn repo_with_one_queued_commit(repo: &Path, state_dir_name: &str) -> String {
    git(repo, &["init"]);
    git(repo, &["config", "user.email", "truth@example.invalid"]);
    git(repo, &["config", "user.name", "Truth Mirror Test"]);
    fs::write(repo.join("file.txt"), "hello\n").unwrap();
    git(repo, &["add", "file.txt"]);
    git(
        repo,
        &[
            "commit",
            "-m",
            "feat: add file",
            "-m",
            "CLAIM: add file | verified: cargo test | evidence: tests:watcher-e2e",
        ],
    );
    let sha = git_stdout(repo, &["rev-parse", "HEAD"]);

    let state = repo.join(state_dir_name);
    fs::create_dir_all(&state).unwrap();
    // Pin the reviewer pair so the detached watcher — which is spawned with no
    // reviewer flags and resolves its pair from config — deterministically calls
    // the mock `codex` reviewer instead of the built-in default (`claude`).
    fs::write(
        state.join("config.toml"),
        r#"default_writer = "claude"

[pairs.claude]
reviewer = { harness = "codex", model = "gpt-5.5" }
arbiter  = { harness = "gemini", model = "gemini-2.5-pro" }
"#,
    )
    .unwrap();
    fs::write(
        state.join("review-queue.jsonl"),
        format!("{{\"commit_sha\":\"{sha}\",\"enqueued_at_unix\":100}}\n"),
    )
    .unwrap();
    sha
}

fn queue_path(state: &Path) -> std::path::PathBuf {
    state.join("review-queue.jsonl")
}

fn queue_is_empty(state: &Path) -> bool {
    fs::read_to_string(queue_path(state))
        .unwrap_or_default()
        .trim()
        .is_empty()
}

fn ledger_contains_reject(state: &Path) -> bool {
    fs::read_to_string(state.join("ledger.jsonl"))
        .unwrap_or_default()
        .contains("\"verdict\":\"REJECT\"")
}

fn wait_until(mut predicate: impl FnMut() -> bool, timeout: Duration) -> bool {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        if predicate() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    predicate()
}

#[test]
fn watch_until_empty_drains_queued_commit_and_exits_after_grace() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let bin = temp.path().join("bin");
    fs::create_dir_all(&repo).unwrap();
    fs::create_dir_all(&bin).unwrap();

    let sha = repo_with_one_queued_commit(&repo, ".truth-mirror");
    let state = repo.join(".truth-mirror");
    let path = mock_reviewer_path(&bin);

    // `--until-empty` with a zero grace window drains the one queued commit and
    // exits 0 immediately (no daemon left running).
    let started = Instant::now();
    truth_mirror(&repo)
        .env("PATH", path)
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--until-empty",
            "--grace",
            "1",
            "--poll-secs",
            "1",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .assert()
        .success()
        .stdout(predicate::str::contains("reviewed 1 commit"));
    assert!(
        started.elapsed() >= Duration::from_millis(900),
        "watcher should linger through the configured grace window"
    );

    assert!(ledger_contains_reject(&state), "verdict should be recorded");
    assert!(
        fs::read_to_string(state.join("ledger.jsonl"))
            .unwrap()
            .contains(&sha)
    );
    assert!(queue_is_empty(&state), "queue should be drained on exit");
    // The watcher released its single-flight lock on exit.
    assert!(
        !state.join("watcher.lock").exists(),
        "lock should be released after until-empty exit"
    );
}

#[test]
fn watch_once_waits_for_live_lock_then_releases_it() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let state = repo.join(".truth-mirror");
    fs::create_dir_all(&repo).unwrap();
    git(&repo, &["init"]);
    fs::create_dir_all(&state).unwrap();
    fs::write(state.join("review-queue.jsonl"), "").unwrap();

    assert_eq!(
        watcher::try_acquire_lock(&state).unwrap(),
        LockClaim::Acquired
    );
    let lock_token = watcher::lock_acquisition_token(&state).unwrap();

    let mut child = StdCommand::new(env!("CARGO_BIN_EXE_truth-mirror"))
        .current_dir(&repo)
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--once",
            "--wait-for-lock",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .spawn()
        .unwrap();
    std::thread::sleep(Duration::from_millis(200));
    assert!(
        child.try_wait().unwrap().is_none(),
        "waited watch command should remain blocked while the lock is held"
    );
    watcher::release_lock_if_owned(&state, &lock_token).unwrap();

    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if let Some(status) = child.try_wait().unwrap() {
            assert!(status.success(), "waited watch command failed: {status}");
            break;
        }
        assert!(
            Instant::now() < deadline,
            "waited watch command did not finish after lock release"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
    assert!(!state.join(watcher::WATCHER_LOCK_FILE).exists());
}

#[test]
fn wait_for_lock_rejects_unbounded_daemon_mode() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    fs::create_dir_all(&repo).unwrap();
    git(&repo, &["init"]);

    truth_mirror(&repo)
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--wait-for-lock",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .assert()
        .failure()
        .stderr(predicate::str::contains(
            "--wait-for-lock is only supported with --once or --until-empty",
        ));
}

#[test]
fn watch_until_empty_waits_for_live_lock_then_releases_it() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let state = repo.join(".truth-mirror");
    fs::create_dir_all(&repo).unwrap();
    git(&repo, &["init"]);
    fs::create_dir_all(&state).unwrap();
    fs::write(state.join("review-queue.jsonl"), "").unwrap();

    assert_eq!(
        watcher::try_acquire_lock(&state).unwrap(),
        LockClaim::Acquired
    );
    let lock_token = watcher::lock_acquisition_token(&state).unwrap();

    let mut child = StdCommand::new(env!("CARGO_BIN_EXE_truth-mirror"))
        .current_dir(&repo)
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--until-empty",
            "--wait-for-lock",
            "--grace",
            "0",
            "--poll-secs",
            "1",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .spawn()
        .unwrap();
    std::thread::sleep(Duration::from_millis(200));
    assert!(child.try_wait().unwrap().is_none());
    watcher::release_lock_if_owned(&state, &lock_token).unwrap();

    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if let Some(status) = child.try_wait().unwrap() {
            assert!(
                status.success(),
                "waited until-empty command failed: {status}"
            );
            break;
        }
        assert!(
            Instant::now() < deadline,
            "waited until-empty command did not finish after lock release"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
    assert!(!state.join(watcher::WATCHER_LOCK_FILE).exists());
}

#[test]
fn watch_once_fails_fast_with_live_lock_diagnostic() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let state = repo.join(".truth-mirror");
    fs::create_dir_all(&repo).unwrap();
    git(&repo, &["init"]);
    fs::create_dir_all(&state).unwrap();
    fs::write(
        state.join("review-queue.jsonl"),
        "{\"commit_sha\":\"abc123\",\"enqueued_at_unix\":100}\n",
    )
    .unwrap();

    assert_eq!(
        watcher::try_acquire_lock(&state).unwrap(),
        LockClaim::Acquired
    );
    let lock_token = watcher::lock_acquisition_token(&state).unwrap();

    truth_mirror(&repo)
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--once",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .assert()
        .failure()
        .stderr(
            predicate::str::contains("watcher lock is already held")
                .and(predicate::str::contains("owner=pid="))
                .and(predicate::str::contains("refusing to touch the queue")),
        );

    assert!(
        !fs::read_to_string(state.join("review-queue.jsonl"))
            .unwrap()
            .trim()
            .is_empty(),
        "fail-fast lock conflict must not drain the queue"
    );
    watcher::release_lock_if_owned(&state, &lock_token).unwrap();
}

#[test]
fn watch_drains_claims_with_configured_custom_evidence_patterns() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let bin = temp.path().join("bin");
    fs::create_dir_all(&repo).unwrap();
    fs::create_dir_all(&bin).unwrap();
    git(&repo, &["init"]);
    git(&repo, &["config", "user.email", "truth@example.invalid"]);
    git(&repo, &["config", "user.name", "Truth Mirror Test"]);
    fs::write(repo.join("file.txt"), "custom evidence\n").unwrap();
    git(&repo, &["add", "file.txt"]);
    git(
        &repo,
        &[
            "commit",
            "-m",
            "feat: configured evidence",
            "-m",
            "CLAIM: review custom evidence | verified: cargo test | evidence: ticket-091",
        ],
    );
    let sha = git_stdout(&repo, &["rev-parse", "HEAD"]);
    let state = repo.join(".truth-mirror");
    fs::create_dir_all(&state).unwrap();
    fs::write(
        state.join("config.toml"),
        r#"default_writer = "claude"

[pairs.claude]
reviewer = { harness = "codex", model = "gpt-5.5" }

[gates]
evidence_patterns = ["ticket-"]
"#,
    )
    .unwrap();
    fs::write(
        state.join("review-queue.jsonl"),
        format!("{{\"commit_sha\":\"{sha}\",\"enqueued_at_unix\":100}}\n"),
    )
    .unwrap();

    truth_mirror(&repo)
        .env("PATH", mock_reviewer_path(&bin))
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--until-empty",
            "--grace",
            "0",
            "--poll-secs",
            "1",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .assert()
        .success()
        .stdout(predicate::str::contains("reviewed 1 commit"));

    assert!(ledger_contains_reject(&state));
    assert!(queue_is_empty(&state));
    let run = fs::read_dir(state.join("runs"))
        .unwrap()
        .next()
        .unwrap()
        .unwrap();
    let run_contents = fs::read_to_string(run.path()).unwrap();
    assert!(run_contents.contains("\"status\": \"completed\""));
}

#[test]
fn watch_until_empty_picks_up_late_enqueue_during_grace_window() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let bin = temp.path().join("bin");
    fs::create_dir_all(&repo).unwrap();
    fs::create_dir_all(&bin).unwrap();

    let _first = repo_with_one_queued_commit(&repo, ".truth-mirror");
    let state = repo.join(".truth-mirror");
    let path = mock_reviewer_path(&bin);

    // Second claim-bearing commit that we will enqueue LATE, after the first has
    // drained but while the watcher is still inside its grace window.
    fs::write(repo.join("file.txt"), "second\n").unwrap();
    git(&repo, &["add", "file.txt"]);
    git(
        &repo,
        &[
            "commit",
            "-m",
            "feat: second change",
            "-m",
            "CLAIM: second change | verified: cargo test | evidence: tests:watcher-late",
        ],
    );
    let second_sha = git_stdout(&repo, &["rev-parse", "HEAD"]);

    // A grace window long enough to reliably win the enqueue race, with fast
    // polling so the late arrival is drained quickly once it lands.
    let exe = assert_cmd::cargo::cargo_bin("truth-mirror");
    let mut child = StdCommand::new(exe)
        .current_dir(&repo)
        .env("PATH", path)
        .args([
            "--state-dir",
            ".truth-mirror",
            "watch",
            "--until-empty",
            "--grace",
            "2",
            "--poll-secs",
            "1",
            "--watched-agent",
            "claude",
            "--watched-model",
            "model-a",
            "--reviewer-harness",
            "codex",
            "--reviewer-model",
            "model-b",
        ])
        .spawn()
        .unwrap();

    // Wait until the first commit has drained (queue empties), proving the watcher
    // is now inside its grace window rather than still working.
    assert!(
        wait_until(|| queue_is_empty(&state), Duration::from_secs(10)),
        "first commit should drain and empty the queue"
    );

    // Enqueue the late arrival while the watcher lingers in the grace window.
    fs::write(
        queue_path(&state),
        format!("{{\"commit_sha\":\"{second_sha}\",\"enqueued_at_unix\":200}}\n"),
    )
    .unwrap();

    // The watcher must pick it up, drain it, and only then exit 0 after a fresh
    // grace window with the queue staying empty.
    let status = wait_child(&mut child, Duration::from_secs(30));
    assert!(
        status.success(),
        "watcher should exit 0 after draining late arrival"
    );

    let ledger = fs::read_to_string(state.join("ledger.jsonl")).unwrap();
    assert!(
        ledger.contains(&second_sha),
        "late-enqueued commit should have been reviewed; ledger:\n{ledger}"
    );
    assert!(queue_is_empty(&state), "queue should be empty on exit");
}

#[test]
fn ensure_watcher_spawns_when_none_then_is_a_noop_when_alive() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let bin = temp.path().join("bin");
    fs::create_dir_all(&repo).unwrap();
    fs::create_dir_all(&bin).unwrap();

    let _sha = repo_with_one_queued_commit(&repo, ".truth-mirror");
    let state = repo.join(".truth-mirror");
    let path = mock_reviewer_path(&bin);

    // First ensure-watcher: nothing running, so it spawns a detached watcher and
    // announces it started.
    truth_mirror(&repo)
        .env("PATH", &path)
        .args(["--state-dir", ".truth-mirror", "ensure-watcher"])
        .assert()
        .success()
        .stdout(predicate::str::contains("watcher started"));

    // A lock file appears, owned by the live watcher.
    assert!(
        wait_until(
            || state.join("watcher.lock").exists(),
            Duration::from_secs(5)
        ),
        "a watcher lock should appear"
    );

    // The detached watcher drains the queued commit and eventually exits after its
    // (default 60s) grace window — we only assert the queue drains, then move on.
    assert!(
        wait_until(|| queue_is_empty(&state), Duration::from_secs(20)),
        "detached watcher should drain the queued commit"
    );
    assert!(ledger_contains_reject(&state));

    // While the watcher is still alive (inside its grace window), a second
    // ensure-watcher is a quiet no-op: it does NOT announce a new start.
    truth_mirror(&repo)
        .env("PATH", &path)
        .args(["--state-dir", ".truth-mirror", "ensure-watcher"])
        .assert()
        .success()
        .stdout(predicate::str::is_empty());
}

#[test]
fn ensure_watcher_recovers_from_a_stale_lock() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let bin = temp.path().join("bin");
    fs::create_dir_all(&repo).unwrap();
    fs::create_dir_all(&bin).unwrap();

    let _sha = repo_with_one_queued_commit(&repo, ".truth-mirror");
    let state = repo.join(".truth-mirror");
    let path = mock_reviewer_path(&bin);

    // Plant a stale lock owned by a reaped child. A made-up PID can be
    // unprobeable on some platforms and must remain Unknown, not be reclaimed.
    let dead_pid = reaped_pid();
    fs::write(
        state.join("watcher.lock"),
        format!(
            r#"{{"identity":{{"pid":{dead_pid},"start_token":"definitely-dead"}},"acquisition_token":"stale-token","created_at_unix":1}}"#
        ),
    )
    .unwrap();

    // ensure-watcher must reclaim the stale lock and spawn a fresh watcher rather
    // than wedging on the dead owner.
    truth_mirror(&repo)
        .env("PATH", &path)
        .args(["--state-dir", ".truth-mirror", "ensure-watcher"])
        .assert()
        .success()
        .stdout(predicate::str::contains("watcher started"));

    assert!(
        wait_until(|| queue_is_empty(&state), Duration::from_secs(20)),
        "recovered watcher should drain the queued commit"
    );
    assert!(ledger_contains_reject(&state));
}

#[test]
fn two_racing_ensure_watchers_yield_exactly_one_watcher() {
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    let bin = temp.path().join("bin");
    fs::create_dir_all(&repo).unwrap();
    fs::create_dir_all(&bin).unwrap();

    let _sha = repo_with_one_queued_commit(&repo, ".truth-mirror");
    let state = repo.join(".truth-mirror");
    let path = mock_reviewer_path(&bin);
    let exe = assert_cmd::cargo::cargo_bin("truth-mirror");

    // Fire two ensure-watcher invocations concurrently, as two racing pushes would.
    let spawn_one = || {
        let exe = exe.clone();
        let repo = repo.clone();
        let path = path.clone();
        std::thread::spawn(move || {
            let output = StdCommand::new(exe)
                .current_dir(&repo)
                .env("PATH", path)
                .args(["--state-dir", ".truth-mirror", "ensure-watcher"])
                .output()
                .unwrap();
            assert!(output.status.success());
            String::from_utf8_lossy(&output.stdout).contains("watcher started")
        })
    };
    let a = spawn_one();
    let b = spawn_one();
    let a_started = a.join().unwrap();
    let b_started = b.join().unwrap();

    // Exactly one of the two racing calls may claim to have started a watcher.
    assert!(
        a_started ^ b_started,
        "exactly one ensure-watcher should start a watcher (a={a_started}, b={b_started})"
    );

    // And that single watcher drains the queue.
    assert!(
        wait_until(|| queue_is_empty(&state), Duration::from_secs(20)),
        "the single watcher should drain the queued commit"
    );
}

/// Wait up to `timeout` for `child` to exit, returning its status. Panics (killing
/// the child) if it outlives the timeout so a hung watcher fails loudly.
fn wait_child(child: &mut std::process::Child, timeout: Duration) -> std::process::ExitStatus {
    let deadline = Instant::now() + timeout;
    loop {
        match child.try_wait().unwrap() {
            Some(status) => return status,
            None if Instant::now() >= deadline => {
                let _ = child.kill();
                let _ = child.wait();
                panic!("watcher did not exit within {timeout:?}");
            }
            None => std::thread::sleep(Duration::from_millis(100)),
        }
    }
}

#[cfg(unix)]
#[test]
fn ensure_watcher_reconciles_orphaned_running_rows_without_spawning_when_held() {
    // 0.9.2 field failure: harness-killed watchers left orphaned `running`
    // rows; ensure-watcher trusted them and silently no-opped. Now it
    // liveness-checks recorded pids and reconciles dead rows every call.
    let temp = tempfile::tempdir().unwrap();
    let repo = temp.path().join("repo");
    fs::create_dir_all(&repo).unwrap();
    git(&repo, &["init"]);

    let state = repo.join(".truth-mirror");
    let runs = state.join("runs");
    fs::create_dir_all(&runs).unwrap();

    // An orphaned running row whose recorded worker pid is provably dead (a
    // real reaped child — a made-up pid number probes UNPROVABLE on some
    // platforms and would correctly be left alone).
    let dead_pid = reaped_pid();
    fs::write(
        runs.join("orphan.json"),
        format!(
            "{{\"id\":\"orphan\",\"commit_sha\":\"abc123\",\"target\":\"commit\",\"status\":\"running\",\"phase\":\"reviewing\",\"ledger_entries\":0,\"error\":null,\"worker_pid\":{dead_pid},\"created_at_unix\":1,\"updated_at_unix\":1,\"started_at_unix\":1,\"completed_at_unix\":null}}"
        ),
    )
    .unwrap();

    // A live foreign process holds the watcher lock, so ensure-watcher must
    // NOT spawn — but it must still reconcile the orphan row first.
    let mut sleeper = std::process::Command::new("sleep")
        .arg("30")
        .spawn()
        .unwrap();
    fs::write(
        state.join("watcher.lock"),
        format!(
            "{{\"identity\":{{\"pid\":{},\"start_token\":\"\"}},\"created_at_unix\":1}}",
            sleeper.id()
        ),
    )
    .unwrap();

    truth_mirror(&repo)
        .args(["--state-dir", ".truth-mirror", "ensure-watcher"])
        .assert()
        .success()
        // Held lock: quiet no-op, no spawn announcement.
        .stdout(predicate::str::is_empty());

    let _ = sleeper.kill();
    let _ = sleeper.wait();

    let reaped = fs::read_to_string(runs.join("orphan.json")).unwrap();
    assert!(
        reaped.contains("\"status\": \"failed\""),
        "orphan row should be reconciled to failed, got: {reaped}"
    );
    assert!(
        reaped.contains("stale run"),
        "reaped row should carry the stale reason, got: {reaped}"
    );
}