forge_agent/env/

git_diff.rs

//! Git diff scanner - on-demand `git` subprocess invocations that
//! produce a [`GitDiffSnapshot`] describing the project's current
//! changeset.
//!
//! Polled (not watched): the workspace reaches [`scan`] as
//! `forge_workspace::env::git_diff::scan`, a wildcard re-export of this
//! module through the workspace facade, and runs it on the
//! "snapshot is `None` OR age ≥ 10 s → fetch" rule while a seat is
//! held. The snapshot carries branch info alongside diff stats, so this
//! module replaces both the previous file-watcher `GitContextWatcher`
//! and the per-turn refresh hooks that fed it.
//!
//! The snapshot exposes two independent diff layers so callers can
//! render both simultaneously when both apply (worker on a topic
//! branch with uncommitted edits, for instance). Each layer is a
//! [`LayerState`] with three variants:
//! - `worktree`: uncommitted edits vs HEAD (the dirty tree).
//!   `Clean` when the tree is clean.
//! - `branch_ahead`: the branch's commits ahead of the default
//!   branch. `Clean` on the default branch, on detached HEAD, or
//!   when there's no resolvable default. `ScanFailed` carries the
//!   per-layer subprocess-error signal.
//!
//! `scan` always returns a value. Subprocess failures, missing
//! repos, oversize output, and timeouts all collapse to a non-InRepo
//! `repo_gate` (`NotARepo` / `ScannerFailed`); the failure surfaces in
//! the trace log so a real issue can be diagnosed without breaking the
//! rendering path.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::{Duration, Instant};

use parking_lot::Mutex;

use super::git_command;
use forge_primitives::git::{FileStatus, GitBranch, GitIssueRef, GitPrInfo};
use forge_primitives::git_diff::{
    BaseState, GitBranchAhead, GitDiffFile, GitDiffSnapshot, GitDiffStats, LayerState, PrRefresh,
    RepoGate,
};
use serde::Deserialize;
use tokio::time::timeout;

pub mod content;
pub mod file;
pub mod hunks;
pub mod resolver;

/// Max bytes accepted from any single `git` subprocess invocation.
/// `--numstat` is small in practice; the per-file `git diff <target>
/// --no-ext-diff -- <path>` calls produced by [`hunks::scan`] are
/// each bounded by one file's worth of changes. 8 MiB is generous
/// headroom for genuinely huge single-file diffs (lockfiles,
/// generated source, vendored snapshots) without exposing the scan
/// to unbounded memory growth on a corrupted-git-output edge case.
const STDOUT_SIZE_CAP: usize = 8 * 1024 * 1024;

/// Per-subprocess timeout. A scan's worst case is one budget per child, and a
/// repository with no remote and no local default branch issues ten of them
/// (`SCAN_CHILDREN`). In practice every command returns in <100 ms on a local
/// repo, and timeouts only fire on hung network mounts or genuinely huge
/// repositories.
const COMMAND_TIMEOUT: Duration = Duration::from_secs(10);

/// The children one scan issues for a repository with no remote and no local
/// default branch: `rev-parse`, the `origin/HEAD` probe plus its four
/// candidates (all missed), `status`, the worktree layer's `numstat` and
/// `name-status`, and `for-each-ref` for the pushed sha. The PR lookup adds a
/// `gh` child only once a pushed sha exists, which such a repository has none
/// of.
///
/// **That shape is the count.** One whose default does resolve leaves the
/// probe loop early and runs the ahead layer instead, so it needs a count of
/// its own; this constant is not it.
#[cfg(any(test, feature = "testing"))]
const SCAN_CHILDREN: u32 = 10;

/// One scan's worst case for the shape `SCAN_CHILDREN` counts, which is what a
/// test waiting on a scan derives its own bound from. Test-only: production
/// reads the per-child budget, never the count.
#[cfg(any(test, feature = "testing"))]
pub const SCAN_WORST_CASE: Duration =
    Duration::from_secs(COMMAND_TIMEOUT.as_secs() * SCAN_CHILDREN as u64);

/// The budget the tests give a scan's `git` children: **not the shell's
/// own.** A scan's commands are bounded by it while the machine starts each
/// child, and a loaded machine can hold a child's first output past the
/// shell's 10 s - the same stall the PR lookup hit (measured 2026-10-14, the
/// family's reds at 10.5 s and 11.9 s). A stalled child must be a slow case
/// in the tests, not a red.
#[cfg(test)]
pub(super) const TEST_GIT_BUDGET: Duration = Duration::from_secs(60);

/// Point the repo at `dir` at a `git` of our own that sleeps before it
/// answers: a child that cannot finish inside a zero budget's first poll,
/// where the machine's warm `git` can - `timeout` polls its value before
/// its delay, so an instant child answers inside a zero budget, which is
/// how the witness's refusal arms redded CI (#2249). With this the
/// refusals are a property of the seam, not a race.
///
/// The wrapper lives in `bin`, **outside the repo**: a file inside would
/// read back as an untracked change and skew a scan's own file list.
#[cfg(test)]
pub(super) fn stall_git_at(dir: &std::path::Path, bin: &std::path::Path) {
    stall_git_for(dir, bin, 5);
}

/// [`stall_git_at`] with the stall spelled, for a case that must show its own
/// budget cutting a child short of a stall longer than it.
#[cfg(test)]
pub(super) fn stall_git_for(dir: &std::path::Path, bin: &std::path::Path, secs: u64) {
    use std::os::unix::fs::PermissionsExt as _;

    let wrapper = bin.join("stalled-git");
    std::fs::write(&wrapper, format!("#!/bin/sh\n/bin/sleep {secs}\nexec git \"$@\"\n"))
        .expect("the wrapper");
    std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).expect("+x");
    crate::env::git_command::point_git_at(dir, &wrapper);
}

/// Point the repo at `dir` at a `git` whose `fetch` refuses and whose every
/// other subcommand is the machine's own: the shape a refresh that cannot run
/// leaves behind - the base stays where it was, and the answer is unchecked
/// with git's own words.
#[cfg(test)]
pub(super) fn fail_fetch_at(dir: &std::path::Path, bin: &std::path::Path) {
    use std::os::unix::fs::PermissionsExt as _;

    let wrapper = bin.join("fetchless-git");
    std::fs::write(
        &wrapper,
        "#!/bin/sh\ncase \"$*\" in\n  *\" fetch \"*) echo 'fatal: forced fetch failure' >&2; \
         exit 128 ;;\nesac\nexec git \"$@\"\n",
    )
    .expect("the wrapper");
    std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).expect("+x");
    crate::env::git_command::point_git_at(dir, &wrapper);
}

/// Point the repo at `dir` at a `git` whose `fetch` refuses the way git does
/// when another process holds a ref lock, so the refresh's lock
/// classification can be driven end to end.
#[cfg(test)]
pub(super) fn lock_fetch_at(dir: &std::path::Path, bin: &std::path::Path) {
    use std::os::unix::fs::PermissionsExt as _;

    let wrapper = bin.join("locked-git");
    std::fs::write(
        &wrapper,
        "#!/bin/sh\ncase \"$*\" in\n  *\" fetch \"*) echo \"fatal: cannot lock ref \
         'refs/remotes/origin/main': File exists.\" >&2; exit 1 ;;\nesac\nexec git \"$@\"\n",
    )
    .expect("the wrapper");
    std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).expect("+x");
    crate::env::git_command::point_git_at(dir, &wrapper);
}

/// Point the repo at `dir` at a `git` that cannot answer the `origin`
/// question at all: both `remote get-url` and the repo-presence probe refuse,
/// while every other subcommand is the machine's own.
#[cfg(test)]
pub(super) fn unanswerable_origin_at(dir: &std::path::Path, bin: &std::path::Path) {
    use std::os::unix::fs::PermissionsExt as _;

    let wrapper = bin.join("unanswerable-origin-git");
    std::fs::write(
        &wrapper,
        "#!/bin/sh\ncase \"$*\" in\n  *\" remote\"*) echo 'fatal: forced' >&2; exit 128 \
         ;;\n  *\" rev-parse --git-dir\"*) echo 'fatal: forced' >&2; exit 128 ;;\n\
         esac\nexec git \"$@\"\n",
    )
    .expect("the wrapper");
    std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).expect("+x");
    crate::env::git_command::point_git_at(dir, &wrapper);
}

/// The environment a test's `git` wrapper gives its child, so a case about a
/// checkout that names no ssh command is not answered by the developer's own
/// global config.
#[cfg(test)]
const ISOLATED_GIT_CONFIG: &str = "GIT_CONFIG_GLOBAL=/dev/null GIT_CONFIG_SYSTEM=/dev/null";

/// Write a `git` wrapper with one refusal on `fetch` and an optional stall on
/// another subcommand, delegating everything else to the machine's own git,
/// with the machine's own config kept out of reach.
#[cfg(test)]
fn write_wrapper(
    dir: &std::path::Path,
    bin: &std::path::Path,
    name: &str,
    stalls: &[(&str, &str)],
    fetch_body: &str,
) {
    use std::fmt::Write as _;
    use std::os::unix::fs::PermissionsExt as _;

    let mut script = String::from("#!/bin/sh\ncase \"$*\" in\n");
    writeln!(script, "  *\" fetch \"*) {fetch_body} ;;").expect("the wrapper's fetch arm");
    for (subcommand, body) in stalls {
        writeln!(script, "  *\" {subcommand} \"*) {body} ;;").expect("the wrapper's stall arm");
    }
    writeln!(script, "esac\n{ISOLATED_GIT_CONFIG} exec git \"$@\"")
        .expect("the wrapper's fallback");
    let wrapper = bin.join(name);
    std::fs::write(&wrapper, script).expect("the wrapper");
    std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).expect("+x");
    crate::env::git_command::point_git_at(dir, &wrapper);
}

/// Point `dir`'s `git` at a wrapper that cannot see the machine's own config:
/// the ssh classifier reads `core.sshCommand` from it, and a developer's global
/// value would answer for the checkout under test.
#[cfg(test)]
fn without_machine_config_at(dir: &std::path::Path, bin: &std::path::Path) {
    use std::os::unix::fs::PermissionsExt as _;

    let script = format!("#!/bin/sh\n{ISOLATED_GIT_CONFIG} exec git \"$@\"\n");
    let wrapper = bin.join("isolated-git");
    std::fs::write(&wrapper, script).expect("the wrapper");
    std::fs::set_permissions(&wrapper, std::fs::Permissions::from_mode(0o755)).expect("+x");
    crate::env::git_command::point_git_at(dir, &wrapper);
}

/// A fetch a test can hold open, and the marks that prove what happened while
/// it was held.
#[cfg(test)]
struct HeldFetch {
    /// Exists while a fetch is blocked in the wrapper.
    open: std::path::PathBuf,
    /// Dropped by the test to let the blocked fetch run.
    release: std::path::PathBuf,
    /// One line per fetch started.
    starts: std::path::PathBuf,
    /// One line per `rev-parse --git-common-dir`, the registry key's own
    /// probe: a second line is a second caller reaching the registry.
    probes: std::path::PathBuf,
}

#[cfg(test)]
impl HeldFetch {
    fn starts(&self) -> usize {
        std::fs::read_to_string(&self.starts).map(|text| text.lines().count()).unwrap_or_default()
    }

    fn probes(&self) -> usize {
        std::fs::read_to_string(&self.probes).map(|text| text.lines().count()).unwrap_or_default()
    }

    fn open(&self) -> bool {
        self.open.exists()
    }

    fn release(&self) {
        std::fs::write(&self.release, b"go").expect("release the fetch");
    }
}

/// Point `dir` at a `git` whose `fetch` marks itself, blocks until
/// [`HeldFetch::release`], and then runs the real one; every
/// `rev-parse --git-common-dir` marks itself too.
#[cfg(test)]
fn held_fetch_at(dir: &std::path::Path, bin: &std::path::Path) -> HeldFetch {
    let open = bin.join("fetch-open");
    let release = bin.join("fetch-release");
    let starts = bin.join("fetch-starts");
    let probes = bin.join("common-dir-probes");
    let body = format!(
        "echo start >> '{starts}'; : > '{open}'; \
         while [ ! -e '{release}' ]; do /bin/sleep 0.02; done; exec git \"$@\"",
        starts = starts.display(),
        open = open.display(),
        release = release.display(),
    );
    let probe = format!("echo probe >> '{}'", probes.display());
    write_wrapper(dir, bin, "held-git", &[("rev-parse", &probe)], &body);
    HeldFetch { open, release, starts, probes }
}

/// Poll `ready` until it answers, with a bound that fails the test rather than
/// hanging it: every wait here is for a child to reach a step, and a loaded
/// machine schedules it late.
#[cfg(test)]
async fn wait_for(ready: impl FnMut() -> bool, what: &str) {
    wait_for_within(ready, what, Duration::from_secs(30)).await;
}

/// [`wait_for`] with its bound spelled: a wait that spans a kick's own children
/// has to allow each the machine's budget, which a loaded machine can spend in
/// full before the fetch it is waiting on even starts.
#[cfg(test)]
async fn wait_for_within(mut ready: impl FnMut() -> bool, what: &str, bound: Duration) {
    let mut waited = Duration::ZERO;
    while !ready() {
        assert!(waited < bound, "waited for {what}");
        tokio::time::sleep(Duration::from_millis(10)).await;
        waited += Duration::from_millis(10);
    }
}

/// Number of file rows surfaced in the rendered diff list. The
/// renderer collapses the rest into a "+N more" line; the count above
/// the list tells the truth either way.
const TOP_FILE_COUNT: usize = 7;

/// Whether a git repository is reachable from a scan cwd, and whether
/// git could answer the question at all. The third variant matters:
/// "git told me there is no repo" and "git would not run" must not
/// collapse together, or a missing git binary reads as a non-repo and
/// silently suppresses the GIT section instead of reporting itself.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RepoPresence {
    /// A repo exists here, even if it has no commits yet.
    Present,
    /// git ran, reported no repository, and there is no `.git` on disk
    /// to contradict it.
    Absent,
    /// git could not be run, or it refused a checkout that does have a
    /// `.git`. Either way the question went unanswered, so the caller
    /// must report rather than suppress.
    Unusable,
}

/// Ask whether `cwd` sits inside a git repo with `rev-parse --git-dir`,
/// whose exit status answers repo-existence alone: zero inside any work
/// tree including one with no commits, non-zero outside one and for a
/// path that does not exist. Matching git's `fatal:` prose would not
/// work here even setting version drift aside - a deleted `.git/HEAD`
/// and a plain non-repo emit the same line.
///
/// A non-zero exit is then refined by whether `.git` is on disk: a
/// pruned worktree entry, a deleted `HEAD` and a corrupt gitfile all
/// exit 128 with `.git` still present, and each is a fault to report
/// rather than an absent repo. Undeterminable biases to `Unusable`, so
/// a broken checkout is never silently suppressed.
///
/// Not routed through [`run_git`]: a non-zero exit is the answer here
/// rather than a fault, so it must neither be logged as a failure nor
/// collapsed with "git would not run", which the caller has already
/// logged with git's stderr.
async fn repo_presence(cwd: &Path) -> RepoPresence {
    repo_presence_within(cwd, COMMAND_TIMEOUT).await
}

/// [`repo_presence`] with its command budget spelled; see [`scan_within`].
async fn repo_presence_within(cwd: &Path, budget: Duration) -> RepoPresence {
    let mut command = git_command::tokio_git_command(cwd);
    command.arg("-C").arg(cwd).args(["rev-parse", "--git-dir"]).kill_on_drop(true);
    match timeout(budget, command.output()).await {
        Ok(Ok(output)) if output.status.success() => RepoPresence::Present,
        Ok(Ok(_)) if dot_git_present(cwd) => RepoPresence::Unusable,
        Ok(Ok(_)) => RepoPresence::Absent,
        Ok(Err(_)) | Err(_) => RepoPresence::Unusable,
    }
}

/// Whether `.git` is on disk, without following symlinks - a dangling
/// `.git` symlink is a broken checkout rather than an absent repo, and
/// `Path::try_exists` reports it as missing because it stats the target.
/// Anything but a definite "not found" counts as present, so an
/// unreadable parent biases to [`RepoPresence::Unusable`] as well.
fn dot_git_present(cwd: &Path) -> bool {
    !matches!(
        cwd.join(".git").symlink_metadata(),
        Err(err) if err.kind() == std::io::ErrorKind::NotFound
    )
}

/// Classify `git rev-parse --abbrev-ref HEAD` output into either the
/// branch string to proceed with, or the terminal `repo_gate`. Empty
/// stdout is a clean "not a repo" signal (git ran and reported it);
/// Oversize means the scan subprocess itself misbehaved.
///
/// `presence` resolves the non-zero-exit case, which is ambiguous on
/// its own: git exits 128 both outside a repo and when it fails inside
/// one, so only the narrower repo-existence question separates a
/// legitimate non-repo from a sick scanner.
fn classify_rev_parse(output: GitOutput, presence: RepoPresence) -> Result<String, RepoGate> {
    match output {
        GitOutput::Ok(s) => Ok(s.trim().to_owned()),
        GitOutput::Empty => Err(RepoGate::NotARepo),
        GitOutput::Failed if presence == RepoPresence::Absent => Err(RepoGate::NotARepo),
        GitOutput::Failed | GitOutput::Oversize => Err(RepoGate::ScannerFailed),
    }
}

/// Resolve the rev-parse gate for `cwd`: the raw branch string, or the
/// terminal `repo_gate`. The repo-existence probe runs only on a
/// non-zero exit, so the healthy path still costs a single subprocess.
async fn rev_parse_gate(cwd: &Path, budget: Duration) -> Result<String, RepoGate> {
    let output = run_git_within(cwd, &["rev-parse", "--abbrev-ref", "HEAD"], budget).await;
    let presence = if matches!(output, GitOutput::Failed) {
        repo_presence_within(cwd, budget).await
    } else {
        RepoPresence::Present
    };
    classify_rev_parse(output, presence)
}

/// Minimum gap between PR-lookup attempts for one branch+sha, even
/// when the attempt found nothing or failed - without this a cached
/// `None` would stick for the branch's whole life (a PR opened after
/// the session's first scan would never show) and a persistent `gh`
/// failure would hammer every 10 s scan.
const PR_REFETCH_INTERVAL: Duration = Duration::from_secs(300);

/// Max commits the pushed-sha walk examines. Bounds the pathological
/// case (a deep unpushed history) so one scan can't walk the whole
/// repo; 500 first-parent commits is generous headroom over any real
/// worktree's unpushed depth.
const PUSHED_SHA_WALK_CAP: usize = 500;

/// The kick's own cadence. The chain stays honest while no review page is
/// open because a scan freshens the ref for the next one; throttled so a burst
/// of scans fetches at most once a window.
const WATCH_REFRESH_INTERVAL: Duration = Duration::from_secs(240);

/// The budget a drawing read's background refresh gets.
///
/// **Longer than a scan's child budget on purpose**: forge's own ssh command
/// gives up at `ConnectTimeout=15`, so a 10 s bound would cut the fetch before
/// git could name its failure and the page would draw "could not be checked"
/// with no reason. A fetch that outlives even this measures nothing, and a
/// cut run is not stored as a check.
const DISPLAY_FETCH_BUDGET: Duration = Duration::from_secs(30);

/// Per-checkout last refresh, keyed by the scan cwd.
///
/// **The cadence is per checkout rather than per repository by choice**: a
/// worktree's own scan is its own read, and two checkouts of one repo
/// refreshing their shared ref twice a window is cheap. The run registry below
/// is the one that must be per repository.
static LAST_WATCH_REFRESH: OnceLock<Mutex<HashMap<PathBuf, Instant>>> = OnceLock::new();

/// One base refresh in flight per repository and base branch, shared by every
/// kick that arrives while it runs: two checkouts of one repository fetch the
/// same ref once. Weak, so a repository nobody is refreshing for leaves
/// nothing behind; dead entries are pruned on the next look.
static BASE_RUNS: OnceLock<Runs> = OnceLock::new();

/// A run's address: the repository's common dir and the base branch.
type RunKey = (PathBuf, String);

/// Every run in flight, weak so a repository nobody is refreshing for leaves
/// nothing behind.
type Runs = Mutex<HashMap<RunKey, std::sync::Weak<BaseRun>>>;

/// One base refresh in flight, and the answer it will publish.
struct BaseRun {
    answer: tokio::sync::watch::Sender<Option<BaseRefresh>>,
}

impl BaseRun {
    fn new() -> std::sync::Arc<Self> {
        let (answer, _) = tokio::sync::watch::channel(None);
        std::sync::Arc::new(Self { answer })
    }

    fn publish(&self, refresh: BaseRefresh) {
        // **`send_replace`, not `send`.** tokio's `send` throws the value away
        // when no receiver is subscribed, which is the ordinary case for a
        // leader nobody joined; a caller that upgrades this run after that
        // would then wait on a channel whose value is still `None` and never
        // wake, wedging every later refresh for the repository.
        self.answer.send_replace(Some(refresh));
    }

    async fn wait(&self) -> BaseRefresh {
        let mut answer = self.answer.subscribe();
        if let Some(held) = answer.borrow().clone() {
            return held;
        }
        // The value is stored rather than only handed to a waiter, so a
        // publish between the borrow above and this await is still read here.
        let _ = answer.changed().await;
        answer.borrow().clone().unwrap_or_else(|| BaseRefresh {
            state: BaseState::Unverified,
            base_ref: None,
            error: Some("the base refresh ended without an answer".to_owned()),
            cut: false,
            unanswered: true,
            at: Instant::now(),
        })
    }
}

/// The run for `key`: the one already in flight (or its answer, when it has
/// finished and a caller still holds it), or a new one the caller leads.
fn join_base_run(key: RunKey) -> (std::sync::Arc<BaseRun>, bool) {
    let mut runs = BASE_RUNS.get_or_init(|| Mutex::new(HashMap::new())).lock();
    runs.retain(|_, weak| weak.strong_count() > 0);
    if let Some(run) = runs.get(&key).and_then(std::sync::Weak::upgrade) {
        return (run, false);
    }
    lead_base_run(&mut runs, key)
}

fn lead_base_run(
    runs: &mut HashMap<RunKey, std::sync::Weak<BaseRun>>,
    key: RunKey,
) -> (std::sync::Arc<BaseRun>, bool) {
    let run = BaseRun::new();
    runs.insert(key, std::sync::Arc::downgrade(&run));
    (run, true)
}

/// Publish the leader's answer, or a failure if the leader ended without one:
/// a caller waiting on a run whose leader panicked or was cancelled must not
/// wait for the process's life.
struct LeaderGuard {
    run: std::sync::Arc<BaseRun>,
    published: bool,
}

impl LeaderGuard {
    fn new(run: std::sync::Arc<BaseRun>) -> Self {
        Self { run, published: false }
    }

    fn publish(&mut self, refresh: BaseRefresh) {
        self.run.publish(refresh);
        self.published = true;
    }
}

impl Drop for LeaderGuard {
    fn drop(&mut self) {
        if self.published {
            return;
        }
        self.run.publish(BaseRefresh {
            state: BaseState::Unverified,
            base_ref: None,
            error: Some("the base refresh ended without an answer".to_owned()),
            cut: false,
            unanswered: true,
            at: Instant::now(),
        });
    }
}

/// One completed base refresh: when it measured, what it found, and git's own
/// words when it could not run.
///
/// **The words ride the measurement** because a page that drew the base
/// unchecked is owed the reason the background fetch gave, and the only path
/// to it is the measurement this store holds.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BaseMeasurement {
    pub at: Instant,
    pub state: BaseState,
    pub error: Option<String>,
}

/// What each checkout's last completed refresh measured, so a scan can carry
/// it onto the seat's row without waiting for the fetch itself. Pruned on the
/// same window as the kick, and the row keeps its stored base when an entry
/// goes.
///
/// **Keyed by the base branch as well as the checkout**: one worktree can draw
/// two bases (a release branch against `origin/main`, then a hotfix against
/// `origin/develop`), and a measurement of one must never answer the other.
static LAST_MEASURED: OnceLock<Mutex<HashMap<(PathBuf, String), BaseMeasurement>>> =
    OnceLock::new();

fn remember_measurement(cwd: &Path, default_branch: &str, refresh: &BaseRefresh) {
    // **A run that ended without answering measured nothing**, and neither did
    // one a bound cut: the first is a placeholder for a fetch that never
    // reported, the second an unchecked marker for one that outlived its
    // budget. Storing either would draw "could not be checked" on the next
    // scan's row and answer a later reading with a check nobody made.
    if refresh.unanswered || refresh.cut {
        return;
    }
    let mut held = LAST_MEASURED.get_or_init(|| Mutex::new(HashMap::new())).lock();
    let now = Instant::now();
    held.retain(|_, measured| measurement_in_window(measured, now));
    held.insert(
        (cwd.to_path_buf(), default_branch.to_owned()),
        BaseMeasurement {
            // **The instant the answer's state was established**, not the
            // instant it was recorded: a read that joined a run stores the
            // same instant its own row carries, so the watch does not rescan
            // for a measurement it has already adopted.
            at: refresh.at,
            state: refresh.state,
            error: refresh.error.clone(),
        },
    );
}

/// This checkout's last completed refresh of `default_branch`. `None` until
/// one has completed; a reader applies `measurement_in_window` itself, since
/// the store only prunes on write.
pub fn last_base_measurement(cwd: &Path, default_branch: &str) -> Option<BaseMeasurement> {
    LAST_MEASURED
        .get_or_init(|| Mutex::new(HashMap::new()))
        .lock()
        .get(&(cwd.to_path_buf(), default_branch.to_owned()))
        .cloned()
}

/// Point `cwd`'s `git` spawns at `program`: the `testing` feature's seam, so a
/// crate whose fixtures need a held child drives the same wrapper this crate's
/// own tests do. Pair with [`clear_git_point_for_tests`].
#[cfg(any(test, feature = "testing"))]
pub fn point_git_at_for_tests(cwd: &Path, program: &Path) {
    crate::env::git_command::point_git_at(cwd, program);
}

/// Drop `cwd`'s pointer, back to the machine's `git`.
#[cfg(any(test, feature = "testing"))]
pub fn clear_git_point_for_tests(cwd: &Path) {
    crate::env::git_command::clear_git_point(cwd);
}

/// The repository a checkout belongs to, as the fetch gate's key: git's
/// common dir, canonicalised.
///
/// **Keyed on the common dir rather than the scan cwd because linked
/// worktrees share `refs/remotes/*`.** Two seats in `<repo>` and
/// `<repo>/.claude/worktrees/x` - this project's normal layout - contend for
/// one ref lock, so a gate keyed per cwd would let both fetch at once and
/// misattribute the loser's refusal. A path git cannot answer for falls back
/// to the cwd: a separate gate cannot deadlock, it only fails to serialise.
async fn repo_key(cwd: &Path, budget: Duration) -> PathBuf {
    let common = match run_git_within(cwd, &["rev-parse", "--git-common-dir"], budget).await {
        GitOutput::Ok(path) => path.trim().to_owned(),
        GitOutput::Empty | GitOutput::Failed | GitOutput::Oversize => return cwd.to_path_buf(),
    };
    let joined = cwd.join(common);
    joined.canonicalize().unwrap_or(joined)
}

/// Whether a watch refresh is due: none yet, or the last one is `window` old.
fn refresh_due(last: Option<Instant>, now: Instant, window: Duration) -> bool {
    match last {
        None => true,
        Some(last) => now.saturating_duration_since(last) >= window,
    }
}

/// Claim a kick's slot for `key`, stamping `now` when it is due, so two scans
/// cannot both fetch.
fn claim_refresh_slot(key: &Path, now: Instant, window: Duration) -> bool {
    let mut map = LAST_WATCH_REFRESH.get_or_init(|| Mutex::new(HashMap::new())).lock();
    // A checkout nobody has scanned for a window is not a slot to keep: the
    // entry would outlive the project it names for the life of the process.
    map.retain(|_, last| now.saturating_duration_since(*last) < window);
    let due = refresh_due(map.get(key).copied(), now, window);
    if due {
        map.insert(key.to_path_buf(), now);
    }
    due
}

/// Stamp `key`'s slot whatever it held: a refresh that just ran left the ref
/// fresh, so a kick inside the window has nothing to fetch.
fn stamp_refresh_slot(key: &Path, now: Instant) {
    LAST_WATCH_REFRESH
        .get_or_init(|| Mutex::new(HashMap::new()))
        .lock()
        .insert(key.to_path_buf(), now);
}

/// Handles for the background fetches kicked so far, so a test can settle an
/// in-flight fetch before its fixture directory is removed. A test that drops
/// its guard while `git fetch` is still writing into the tree aborts the
/// removal part-way and leaves the rest of the directory behind.
///
/// Per process and never pruned, so under libtest's threaded runner one test's
/// settle drains another's handle. nextest, which this repo runs, gives each
/// test a process of its own.
#[cfg(test)]
static FETCH_TASKS: OnceLock<Mutex<Vec<tokio::task::JoinHandle<()>>>> = OnceLock::new();

/// Hand a kicked fetch's handle to the test-only registry above. A production
/// build discards it: the fetch is fire-and-forget, which is what the scan
/// promises its caller.
fn track_background_fetch(fetch: tokio::task::JoinHandle<()>) {
    #[cfg(test)]
    FETCH_TASKS.get_or_init(|| Mutex::new(Vec::new())).lock().push(fetch);
    #[cfg(not(test))]
    drop(fetch);
}

/// Await every background fetch kicked so far, and report how many that was.
/// The count is the caller's guard against a silent no-op: a fixture that
/// stops resolving a remote-tracking default, or a window that swallows the
/// kick, would leave nothing to settle and the teardown racing an unseen
/// child again.
#[cfg(test)]
async fn settle_background_fetches() -> usize {
    let tasks = std::mem::take(&mut *FETCH_TASKS.get_or_init(|| Mutex::new(Vec::new())).lock());
    let settled = tasks.len();
    for task in tasks {
        let _ = task.await;
    }
    settled
}

/// Kick a throttled, non-blocking base refresh so the remote-tracking ref the
/// branch-ahead layer compares against is fresher for the NEXT scan. Only fires
/// for a remote-tracking default (`origin/...`); a purely-local default has no
/// remote to fetch. Failures log from [`fetch_base`], which owns the one fetch
/// path with its flags and lock line.
///
/// **Nobody waits on this refresh**, and it may not spend a reader's wait: a
/// drawing read that arrives while it runs answers what it already holds and
/// lets this run's answer serve the next read (see
/// [`refresh_base_for_display`]) rather than joining it. A second kick for the
/// same repository joins the run rather than fetching again, and `budget`
/// bounds the run, so a hung remote cannot leave a child behind.
///
/// **`scan_within` is the one scan that can serve a seat with no loop**: its
/// record read would otherwise re-read the tree for ever against a ref
/// nothing ever refreshes.
fn kick_background_fetch(cwd: &Path, default_branch: &str, budget: Duration) {
    if !default_branch.starts_with("origin/") {
        return;
    }
    if !claim_refresh_slot(cwd, Instant::now(), WATCH_REFRESH_INTERVAL) {
        return;
    }
    let cwd = cwd.to_path_buf();
    let branch = default_branch.to_owned();
    track_background_fetch(tokio::spawn(async move {
        let _ = refresh_base(&cwd, &branch, budget).await;
    }));
}

/// What one base refresh answered.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BaseRefresh {
    /// How current the base is.
    pub state: BaseState,
    /// The sha the base resolved to after the refresh. Both the stats scan and
    /// the content read are taken against this one sha, so a count and a file
    /// list can never describe two different refs.
    pub base_ref: Option<String>,
    /// Why the fetch did not run, in git's own first line (or forge's words
    /// for a child it could not start or had to bound). `None` when it ran.
    /// The record carries this so a page can say what failed rather than only
    /// that the base could not be checked.
    pub error: Option<String>,
    /// The run's own bound ended the fetch. A cut run measured nothing, so its
    /// answer is never stored as the checkout's check.
    pub cut: bool,
    /// The run ended without answering - its leader panicked or was cancelled.
    /// The state and error are a placeholder rather than a measurement, and no
    /// scan's row carries them.
    pub unanswered: bool,
    /// **When this answer's state was established**: the instant a completed
    /// refresh measured it, the instant a drawing read decided to answer the
    /// base unchecked, or the instant the stored measurement being reused was
    /// taken. A scan stamps the row's `base_at` from this rather than from its
    /// own write, so a measurement that lands while the row is being stored is
    /// still newer than the place it must land in.
    pub at: Instant,
}

/// Answer the base a drawn branch diff is measured against, and keep it fresh
/// off the read.
///
/// **Called by the read that draws the diff, never by a record read.** A state
/// carried on a stored scan would be as old as that scan and could call a base
/// current after the remote had moved on, which is the stale-diff defect
/// itself.
///
/// **A remote-tracking base is never waited on.** The read answers the last
/// completed check the store holds, or the base unchecked when it holds none,
/// pinned to the local tip the diff is read at, and kicks a throttled
/// background refresh either way; that run's measurement answers the next read
/// (#2818). A local base has no fetch to run: the probes that say whether it
/// can lag behind anything answer here, as they did before the refresh became
/// a kick.
pub async fn refresh_base_for_display(cwd: &Path, default_branch: &str) -> BaseRefresh {
    let decided_at = Instant::now();
    if !default_branch.starts_with("origin/") {
        let deadline = Some(Instant::now() + COMMAND_TIMEOUT);
        let (state, base_ref) = local_base(cwd, default_branch, deadline).await;
        return BaseRefresh {
            state,
            base_ref,
            error: None,
            cut: false,
            unanswered: false,
            at: decided_at,
        };
    }
    // **Kicked before the local reads below**, so the fetch runs beside them
    // rather than after. The slot is claimed synchronously: a run another
    // checkout of this repository already kicked is joined, not repeated, and
    // one this checkout fetched for inside the window adds nothing.
    kick_background_fetch(cwd, default_branch, DISPLAY_FETCH_BUDGET);
    stored_base(cwd, default_branch, decided_at).await
}

/// Answer the last completed check the store holds, or the base unchecked,
/// pinned to the local tip read beside it.
///
/// **A run that published while the tip was being read still counts**, as long
/// as it completed before that tip was read: its fetch wrote the ref the tip
/// names, so the pair is one base. The store is re-read after the tip for that
/// reason, and a measurement stamped after the tip is left for the next read
/// rather than paired with a ref it does not describe.
async fn stored_base(cwd: &Path, default_branch: &str, decided_at: Instant) -> BaseRefresh {
    let measured = last_base_measurement(cwd, default_branch)
        .filter(|measured| measurement_in_window(measured, decided_at));
    let base_ref = local_tip(cwd, default_branch, COMMAND_TIMEOUT).await;
    let tip_at = Instant::now();
    let measured = measured.or_else(|| {
        last_base_measurement(cwd, default_branch)
            .filter(|measured| measured.at <= tip_at && measurement_in_window(measured, tip_at))
    });
    match measured {
        // **The instant the check was made**, not this read's: a run
        // publishing while the read is answered must still land newer than
        // the place this answer is stored.
        Some(measured) => answer_measurement(measured, base_ref),
        None => BaseRefresh {
            state: BaseState::Unverified,
            base_ref,
            error: None,
            cut: false,
            unanswered: false,
            at: decided_at,
        },
    }
}

/// Answer the base by running the refresh on this call: **the one path that
/// waits**, for a seat with no session.
///
/// A sessionless seat has no watch, so nothing would deliver a kicked check to
/// its page: the read is the only thing that can carry it. The slot is claimed
/// first, so a second open inside the window answers the stored check rather
/// than fetching again (#2818).
pub async fn refresh_base_now(cwd: &Path, default_branch: &str) -> BaseRefresh {
    let decided_at = Instant::now();
    if !default_branch.starts_with("origin/") {
        let deadline = Some(Instant::now() + COMMAND_TIMEOUT);
        let (state, base_ref) = local_base(cwd, default_branch, deadline).await;
        return BaseRefresh {
            state,
            base_ref,
            error: None,
            cut: false,
            unanswered: false,
            at: decided_at,
        };
    }
    if claim_refresh_slot(cwd, decided_at, WATCH_REFRESH_INTERVAL) {
        // **The display budget, not the scan's**: forge's own ssh gives up at
        // `ConnectTimeout=15` and http at `lowSpeedTime=20`, so a ten-second
        // bound would cut the fetch before git could name its failure and the
        // page would draw "could not be checked" with no reason.
        let refresh = refresh_base(cwd, default_branch, DISPLAY_FETCH_BUDGET).await;
        give_back_slot(cwd, &refresh);
        return refresh;
    }
    // **The slot is another run's.** Answer the check it left when there is
    // one; otherwise wait on a run - a sessionless page has no watch to heal
    // an answer that names no check.
    if let Some(measured) = last_base_measurement(cwd, default_branch)
        .filter(|measured| measurement_in_window(measured, decided_at))
    {
        let base_ref = local_tip(cwd, default_branch, COMMAND_TIMEOUT).await;
        answer_measurement(measured, base_ref)
    } else {
        let refresh = refresh_base(cwd, default_branch, DISPLAY_FETCH_BUDGET).await;
        give_back_slot(cwd, &refresh);
        refresh
    }
}

/// The answer a stored check gives a drawing read: its state, words and
/// instant, pinned to the local tip the caller read beside it.
fn answer_measurement(measured: BaseMeasurement, base_ref: Option<String>) -> BaseRefresh {
    BaseRefresh {
        state: if base_ref.is_some() { measured.state } else { BaseState::Unverified },
        base_ref,
        error: measured.error,
        cut: false,
        unanswered: false,
        at: measured.at,
    }
}

/// **A run that answered no check gives the slot back.** A cut or aborted run
/// measured nothing; leaving its claim stamped would answer silence for the
/// whole window on the one seat whose read is what heals it.
fn give_back_slot(cwd: &Path, refresh: &BaseRefresh) {
    if !refresh.cut && !refresh.unanswered {
        return;
    }
    if let Some(slots) = LAST_WATCH_REFRESH.get() {
        slots.lock().remove(cwd);
    }
}

/// Whether a measurement still answers a drawing read: the store's own window,
/// which is how long the store keeps one.
fn measurement_in_window(measured: &BaseMeasurement, now: Instant) -> bool {
    now.saturating_duration_since(measured.at) < WATCH_REFRESH_INTERVAL
}

/// [`refresh_base_for_display`] with a bound: a test gives the refresh one the
/// machine's load cannot eat, since a budget spent starting a child is the
/// machine's bound rather than the test's subject. Production's bounds are the
/// kick's (the scan's child budget) and a sessionless read's (the display
/// budget), both reaching [`refresh_base`].
#[cfg(test)]
async fn refresh_base_within(cwd: &Path, default_branch: &str, budget: Duration) -> BaseRefresh {
    refresh_base(cwd, default_branch, budget).await
}

/// The one refresh path: one run per repository and base branch, joined by a
/// kick that arrives while it works, and its answer stored as the checkout's
/// measurement for the reads that follow.
async fn refresh_base(cwd: &Path, default_branch: &str, budget: Duration) -> BaseRefresh {
    let key = (repo_key(cwd, COMMAND_TIMEOUT).await, default_branch.to_owned());
    let refresh = refresh_once(cwd, default_branch, budget, key).await;
    remember_measurement(cwd, default_branch, &refresh);
    refresh
}

/// Join (or lead) one run for `key` and answer it. A kick that arrives while
/// another works joins it rather than fetching the same base twice.
async fn refresh_once(
    cwd: &Path,
    default_branch: &str,
    budget: Duration,
    key: RunKey,
) -> BaseRefresh {
    let (run, leader) = join_base_run(key);
    if leader {
        let mut guard = LeaderGuard::new(std::sync::Arc::clone(&run));
        let refresh = refresh_base_owned(cwd, default_branch, budget).await;
        guard.publish(refresh.clone());
        refresh
    } else {
        run.wait().await
    }
}

/// One refresh's own work: the fetch, then the state read off the ref that
/// fetch wrote, every step sharing `budget`.
async fn refresh_base_owned(cwd: &Path, default_branch: &str, budget: Duration) -> BaseRefresh {
    let deadline = Some(Instant::now() + budget);
    // **The user's own ssh command wins**, whichever spelling holds it: forge
    // sets `GIT_SSH_COMMAND` only when nothing else names one.
    let ssh = if ssh_is_forges(cwd).await {
        git_command::SshCommand::Forge
    } else {
        git_command::SshCommand::User
    };
    // **A local base never reaches here in production**: both callers branch
    // on a remote-tracking default first and answer a local base from
    // [`local_base`]. This crate's bounded tests drive the refresh directly
    // with one, and they get the same probes rather than a fetch.
    let Some(remote_branch) = default_branch.strip_prefix("origin/") else {
        let (state, base_ref) = local_base(cwd, default_branch, deadline).await;
        return BaseRefresh {
            state,
            base_ref,
            error: None,
            cut: false,
            unanswered: false,
            at: Instant::now(),
        };
    };
    let fetched = fetch_base(cwd, remote_branch, deadline, budget, ssh).await;
    if fetched.error.is_none() {
        // **The ref is fresh now.** Stamping the kick's window here is what
        // keeps a scan seconds later from fetching the same answer again.
        stamp_refresh_slot(cwd, Instant::now());
    }
    // **A successful fetch is the answer, so the remote is not asked
    // again.** `fetch_base` carries an explicit
    // `+refs/heads/<b>:refs/remotes/origin/<b>` refspec, which writes that
    // destination whatever the repository's own `remote.origin.fetch`
    // says, and a fetch that cannot find the ref exits non-zero. So exit 0
    // leaves the ref at the remote's tip and the base is current by
    // construction. This replaces the post-fetch `ls-remote` that guarded
    // against a fetch writing nothing: the explicit refspec is what makes
    // that guard unnecessary. A fetch that failed wrote nothing, and no
    // remote tip is read to count a lag against, so its base is answered
    // unchecked with git's own words rather than with a count.
    let base_ref = local_tip(cwd, default_branch, remaining(deadline)).await;
    let state = if fetched.error.is_none() && base_ref.is_some() {
        BaseState::Current
    } else {
        BaseState::Unverified
    };
    BaseRefresh {
        state,
        base_ref,
        error: fetched.error,
        cut: fetched.cut,
        unanswered: false,
        at: Instant::now(),
    }
}

/// The state of a base that names no remote-tracking ref, and the sha it
/// resolves to: the probes a local base owes.
///
/// A local base has no ref to fetch. With an `origin` remote it can still be
/// arbitrarily behind one nobody ever fetched, and there is no
/// remote-tracking ref to name that lag against, so the base is not claimed
/// rather than silently called local. **A base that tracks a remote of its
/// own can lag too**: a local `main` whose upstream is `upstream/main` is
/// stale exactly as an `origin/main` would be, and calling it local would be
/// the same silence.
async fn local_base(
    cwd: &Path,
    default_branch: &str,
    deadline: Option<Instant>,
) -> (BaseState, Option<String>) {
    let local = local_tip(cwd, default_branch, remaining(deadline)).await;
    let state = match origin_remote(cwd, deadline).await {
        OriginRemote::Present => BaseState::Unverified,
        OriginRemote::Absent => {
            match local_base_tracks_upstream(cwd, default_branch, deadline).await {
                Some(false) => BaseState::Local,
                Some(true) | None => BaseState::Unverified,
            }
        }
        // **A probe that never answered must not answer "no remote".** That is
        // the silent stale claim by another name: the base is drawn as purely
        // local while an `origin` it may lag behind sits right there.
        OriginRemote::Unanswered(reason) => {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "git_base_origin_probe_unanswered",
                message = "the origin probe could not answer, so a local base is not claimed",
                outcome = "unverified",
                reason = %reason,
                base = %default_branch,
            );
            BaseState::Unverified
        }
    };
    (state, local)
}

/// Whether forge builds the ssh command for this checkout's network children:
/// true when the environment names none and the repository's own
/// `core.sshCommand` is unset, because `GIT_SSH_COMMAND` outranks both.
async fn ssh_is_forges(cwd: &Path) -> bool {
    if std::env::var_os("GIT_SSH_COMMAND").is_some() || std::env::var_os("GIT_SSH").is_some() {
        return false;
    }
    // **`git config --get` on a missing key EXITS 1**, which the classified
    // reader reports as a failure, not as an empty answer: the exit code is
    // what tells "this checkout names no command" from "git could not
    // answer", and only the first may install forge's.
    let mut command = git_command::tokio_git_command(cwd);
    command.arg("-C").arg(cwd).args(["config", "--get", "core.sshCommand"]).kill_on_drop(true);
    match child_output(command.output(), COMMAND_TIMEOUT).await {
        // The key is set. An empty value is git's own "unset", so it is not.
        Some(Ok(out)) if out.status.success() => out.stdout.iter().all(u8::is_ascii_whitespace),
        // Exit 1 with nothing on stdout: no such key.
        Some(Ok(out)) if out.status.code() == Some(1) && out.stdout.is_empty() => true,
        // A probe that did not answer leaves the question open, and forge does
        // not overwrite a command it could not read.
        _ => false,
    }
}

/// Fetch one remote-tracking branch without touching the rest of the user's
/// checkout: an explicit refspec with its destination, so a remote whose own
/// fetch refspec differs still updates the ref this layer reads;
/// `--no-auto-gc` so a killed child leaves no maintenance daemon behind;
/// `--no-write-fetch-head` so their `FETCH_HEAD` is not clobbered on every
/// open; and `--quiet` so progress never rides a captured stream.
///
/// Answers what the child did: `error` is `None` when git fetched and git's own
/// words when it did not, and `cut` says the run's bound ended it.
async fn fetch_base(
    cwd: &Path,
    remote_branch: &str,
    deadline: Option<Instant>,
    budget: Duration,
    ssh: git_command::SshCommand,
) -> Fetched {
    let refspec = format!("+refs/heads/{remote_branch}:refs/remotes/origin/{remote_branch}");
    let mut command = git_command::network_tokio_git_command(cwd, ssh);
    command
        .arg("-C")
        .arg(cwd)
        .args(["fetch", "--no-auto-gc", "--no-write-fetch-head", "--quiet", "origin", &refspec])
        .kill_on_drop(true);
    // **A bounded run kills the child, and a fetch killed while it holds a ref
    // lock leaves `<ref>.lock` behind.** The window is between the lock being
    // taken and the ref renamed - microseconds - and the alternative is a hung
    // remote keeping a child per open. A stranded lock is not cleaned up here:
    // removing a lock another git may hold is worse than naming it, which the
    // lock WARN below does.
    let left = remaining(deadline);
    match child_output(command.output(), left).await {
        Some(Ok(out)) if out.status.success() => Fetched { error: None, cut: false },
        Some(Ok(out)) => {
            let stderr = String::from_utf8_lossy(&out.stderr);
            let first = stderr.lines().next().unwrap_or_default();
            if is_lock_failure(&stderr) {
                // **A ref lock is the environment's mark in the user's tree**,
                // and nothing here can say whose: the user's own git, another
                // forge instance on the same repo, and a killed child's
                // stranded lock all read the same. DEBUG for that reason
                // (rule 20): the line names the repo, the branch and git's own
                // words, which carry the lock file's path - everything acting
                // on it needs, without claiming it is forge's health.
                tracing::debug!(
                    target: crate::logging::targets::ENV_GIT,
                    cwd = %cwd.display(),
                    event_name = "git_base_fetch_lock_refused",
                    message = "the base fetch could not take a ref lock in the checkout",
                    outcome = "failure",
                    branch = %remote_branch,
                    stderr = %first,
                );
            } else {
                tracing::debug!(
                    target: crate::logging::targets::ENV_GIT,
                    cwd = %cwd.display(),
                    event_name = "git_base_fetch_nonzero_exit",
                    message = "the base fetch exited non-zero",
                    outcome = "failure",
                    exit_code = out.status.code().unwrap_or(-1),
                    branch = %remote_branch,
                    stderr = %first,
                );
            }
            // **Redact first, then the cap.** A cap that lands inside a
            // credential's userinfo removes the `@` the redactor needs, and
            // the credential would reach the record whole.
            Fetched {
                error: Some(clipped(&redact(&first_line(&out.stderr)), FAILURE_WORDS_CAP)),
                cut: false,
            }
        }
        Some(Err(err)) => {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "git_base_fetch_failed",
                message = "the base fetch spawn / wait failed",
                outcome = "failure",
                error = %err,
                branch = %remote_branch,
            );
            Fetched { error: Some(format!("git could not be run: {err}")), cut: false }
        }
        None => {
            // **The run's budget, not what was left of it**: a run that spent
            // its bound before the child started would otherwise say "0s".
            let given = budget.as_secs_f64();
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "git_base_fetch_timeout",
                message = "the base fetch did not finish inside the budget it was given",
                outcome = "timeout",
                branch = %remote_branch,
                budget_s = given,
            );
            let words = format!("the fetch did not finish inside the {given}s it was given");
            Fetched { error: Some(words), cut: true }
        }
    }
}

/// What one fetch's child answered.
struct Fetched {
    /// Why it did not run, in git's own words; `None` when it ran.
    error: Option<String>,
    /// The run's own bound ended it.
    cut: bool,
}

/// Git's own words for a failed child: the first non-empty stderr line,
/// uncapped so the redactor can see the whole URL before anything is cut.
fn first_line(stderr: &[u8]) -> String {
    let text = String::from_utf8_lossy(stderr);
    text.lines()
        .map(str::trim)
        .find(|line| !line.is_empty())
        .unwrap_or("git exited non-zero without saying why")
        .to_owned()
}

/// Carried characters of git's failure line. A page draws it in one sentence
/// beside the base note; git's own `fatal:` lines are far shorter than this,
/// and the cap applies to the redacted line.
const FAILURE_WORDS_CAP: usize = 200;

/// Git's words with any credential a remote URL carries taken out.
///
/// git prints the URL it could not reach, userinfo included, and the record
/// this feeds is re-sent on every read and drawn on a page. Everything before
/// the `@` is the part a URL keeps its credential in, so it goes whole.
fn redact(words: &str) -> String {
    let mut out = String::with_capacity(words.len());
    let mut rest = words;
    while let Some(at) = rest.find("://") {
        let (before, after) = rest.split_at(at + 3);
        out.push_str(before);
        // The authority runs to the next path separator, quote or space.
        let end = after
            .find(|c: char| c == '/' || c == '\'' || c == '"' || c.is_whitespace())
            .unwrap_or(after.len());
        let (authority, tail) = after.split_at(end);
        match authority.rfind('@') {
            Some(userinfo) => {
                out.push_str("<redacted>");
                out.push_str(&authority[userinfo..]);
            }
            None => out.push_str(authority),
        }
        rest = tail;
    }
    out.push_str(rest);
    out
}

/// Whether git refused the fetch because a ref lock it needed is held - by
/// another git in the same checkout, or by a killed child that left one.
///
/// **Forge runs one fetch per repository and base branch** (a caller joins the
/// run already in flight), so no second fetch of this ref is this process's:
/// two branches of one repository take different ref locks and may run
/// together. What reaches this check is the user's git, another forge
/// instance on the same repo, or a stranded lock.
fn is_lock_failure(stderr: &str) -> bool {
    stderr.contains("cannot lock ref")
        || (stderr.contains("Unable to create") && stderr.contains(".lock"))
}

/// The budget a command handed no deadline reads as: no cap at all. Every
/// refresh runs under a budget of its own (a kick's, or a sessionless read's),
/// so this is what "no deadline" means to the helpers rather than a mode any
/// read takes.
const UNBOUNDED: Duration = Duration::MAX;

/// The budget left before `deadline`, zero once spent and [`UNBOUNDED`] when
/// there is no deadline at all.
fn remaining(deadline: Option<Instant>) -> Duration {
    deadline.map_or(UNBOUNDED, |deadline| deadline.saturating_duration_since(Instant::now()))
}

/// One child's output under `budget`; `None` when the budget ran out first and
/// the child was killed.
async fn child_output(
    child: impl std::future::Future<Output = std::io::Result<std::process::Output>>,
    budget: Duration,
) -> Option<std::io::Result<std::process::Output>> {
    if budget == UNBOUNDED { Some(child.await) } else { timeout(budget, child).await.ok() }
}

/// The sha `reference` resolves to, or `None` when git cannot read it.
async fn local_tip(cwd: &Path, reference: &str, budget: Duration) -> Option<String> {
    if budget.is_zero() {
        return None;
    }
    match run_git_within(cwd, &["rev-parse", reference], budget).await {
        GitOutput::Ok(sha) => Some(sha.trim().to_owned()),
        GitOutput::Empty | GitOutput::Failed | GitOutput::Oversize => None,
    }
}

/// Whether the checkout has an `origin` remote: yes, no, or a probe that
/// could not answer.
enum OriginRemote {
    Present,
    Absent,
    /// git could not be run, or a sick checkout refused it. The question is
    /// open, and an open question must not answer "no remote". The reason is
    /// what the line about it carries: a spent budget, a git that refused,
    /// and an oversized answer are three different things to look at.
    Unanswered(&'static str),
}

/// Ask whether `cwd` has an `origin` remote at all: a local base beside one is
/// a base nobody fetched, which is not the same as a repository without a
/// remote to lag behind.
///
/// **The question is asked with `git remote`, not `get-url`.** A non-zero
/// `get-url` is two answers in one shape - "no such remote" and a git that
/// would not run - and reading both as "no remote" is how a base nobody
/// fetched got drawn as purely local. `git remote` answers a list on success;
/// a repository with none is an empty list, and anything else is a probe that
/// did not answer.
///
/// The match is the literal `origin`: it is the one remote a base refresh
/// knows how to fetch, so a repository whose only remote is named otherwise
/// still reads as having nothing this read could refresh.
async fn origin_remote(cwd: &Path, deadline: Option<Instant>) -> OriginRemote {
    if remaining(deadline).is_zero() {
        return OriginRemote::Unanswered("budget_spent");
    }
    match run_git_within(cwd, &["remote"], remaining(deadline)).await {
        GitOutput::Ok(listing) => {
            if listing.lines().any(|name| name.trim() == "origin") {
                OriginRemote::Present
            } else {
                OriginRemote::Absent
            }
        }
        // Ran in a repository with no remotes at all.
        GitOutput::Empty => OriginRemote::Absent,
        GitOutput::Failed => match repo_presence_within(cwd, remaining(deadline)).await {
            // Not a repository: there is no origin to lag behind either.
            RepoPresence::Absent => OriginRemote::Absent,
            // A repository git refused to list: the question is open.
            RepoPresence::Present | RepoPresence::Unusable => {
                OriginRemote::Unanswered("git_refused")
            }
        },
        GitOutput::Oversize => OriginRemote::Unanswered("oversize"),
    }
}

/// Whether the local base itself tracks a remote (`main@{upstream}`): a base
/// that does can be behind that remote exactly as an `origin/...` can, and
/// nothing was fetched for it here. `None` when git could not answer, which
/// the caller treats as unverified rather than as "nothing to lag behind".
async fn local_base_tracks_upstream(
    cwd: &Path,
    base: &str,
    deadline: Option<Instant>,
) -> Option<bool> {
    if remaining(deadline).is_zero() {
        return None;
    }
    let spec = format!("{base}@{{upstream}}");
    match run_git_within(cwd, &["rev-parse", "--abbrev-ref", &spec], remaining(deadline)).await {
        GitOutput::Ok(_) => Some(true),
        GitOutput::Empty | GitOutput::Failed => Some(false),
        GitOutput::Oversize => None,
    }
}

/// Run the full scan sequence against `cwd` and return a snapshot.
/// Always succeeds - every failure path collapses to a non-InRepo
/// `repo_gate` and a log line naming the step that failed. Callers
/// should treat the snapshot as authoritative for rendering
/// regardless of which variant came back.
///
/// `prev` is the most recent snapshot for this `cwd` (if any). It's
/// used to reuse cached PR info while the branch name, pushed sha
/// and fetch age (5 minutes) all say the cached answer still holds
/// (`pr_cache_fresh`). Pass `None` for cold starts.
/// The caller's current branch name via `git rev-parse --abbrev-ref HEAD`,
/// keeping a detached HEAD apart from a failed read: `Ok(Some(name))` on
/// a named branch, `Ok(None)` on a detached HEAD, `Err(gate)` when git
/// reported nothing usable. Mirrors the `GitBranch::Named` string
/// [`scan`] derives, so the review MCP can resolve a caller's `(project,
/// branch)` review scope to the same key the `/diff` overlay persists
/// under - and name the step that failed when it can't.
pub async fn current_branch(cwd: &Path) -> Result<Option<String>, RepoGate> {
    let name = rev_parse_gate(cwd, COMMAND_TIMEOUT).await?;
    Ok((name != "HEAD" && !name.is_empty()).then_some(name))
}

/// How many files `cwd`'s worktree has changed against HEAD: tracked
/// modifications plus untracked files, one `--porcelain` line each. `Err`
/// carries the repo gate, so a caller can tell "not a repo" from a git
/// that would not run rather than reading either as a clean tree.
pub async fn changed_file_count(cwd: &Path) -> Result<usize, RepoGate> {
    match run_git(cwd, &["status", "--porcelain"]).await {
        GitOutput::Ok(text) => Ok(text.lines().filter(|line| !line.trim().is_empty()).count()),
        GitOutput::Empty => Ok(0),
        GitOutput::Oversize => Err(RepoGate::ScannerFailed),
        // Same split `rev_parse_gate` draws: a non-zero exit with `.git`
        // still on disk is a broken checkout to report, not an absent repo.
        GitOutput::Failed => match repo_presence(cwd).await {
            RepoPresence::Absent => Err(RepoGate::NotARepo),
            RepoPresence::Present | RepoPresence::Unusable => Err(RepoGate::ScannerFailed),
        },
    }
}

/// `env` is `cwd`'s project's own declared environment, carried as the
/// identity of every `gh` call the scan makes, so the PR row reads as
/// the project rather than as the server's ambient login.
pub async fn scan<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    prev: Option<&GitDiffSnapshot>,
) -> GitDiffSnapshot {
    scan_within(cwd, env, prev, None, PrRefresh::Fetch, true, COMMAND_TIMEOUT).await
}

/// [`scan`] for a read that will fetch the base itself a moment later: the
/// same local scan with none of the background kick, so one read of a
/// sessionless seat pays one fetch - its own - and never queues that fetch
/// behind forge's on the shared per-repository gate.
pub async fn scan_without_base_fetch<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    prev: Option<&GitDiffSnapshot>,
) -> GitDiffSnapshot {
    scan_within(cwd, env, prev, None, PrRefresh::Fetch, false, COMMAND_TIMEOUT).await
}

/// [`scan`] for a read a client is waiting on: a PR cache past its window is
/// served for the branch and sha it belongs to rather than looked up, so the
/// read never waits on the network; a cache from another branch or push
/// answers nothing. The seat's own loop scans with [`scan`] and pushes the
/// refreshed row.
pub async fn scan_serving_cached_pr<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    prev: Option<&GitDiffSnapshot>,
) -> GitDiffSnapshot {
    scan_within(cwd, env, prev, None, PrRefresh::ServeCached, false, COMMAND_TIMEOUT).await
}

/// A read pinned to one base revision: the branch-ahead layer is computed
/// against `base` (a sha a refresh just resolved) rather than against the
/// ref's name, so this read and a content read taken beside it describe one
/// commit even if the ref moves between them. Never fetches and never looks
/// the PR up: the caller has already refreshed the base and is drawing it.
pub async fn scan_pinned<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    prev: Option<&GitDiffSnapshot>,
    base: &str,
) -> GitDiffSnapshot {
    scan_within(cwd, env, prev, Some(base), PrRefresh::ServeCached, false, COMMAND_TIMEOUT).await
}

/// [`scan`] with the scan's command budget spelled: the shell gives every
/// child a scan spawns the 10 s `COMMAND_TIMEOUT`, and a test gives them one
/// of its own.
///
/// **The budget is a bound on the machine, not on the scan's work**: it is
/// spent starting each child, so a machine can hold a child's first output
/// past the shell's 10 s and fail or blank a layer that is merely slow -
/// measured 2026-10-14: the PR lookup's `gh` child was killed at 10.5 s and
/// 10.9 s while a gate ran beside the suite, and a `gh` held back 11 s
/// reproduces it on demand, the same mechanism that reds a scan whose `git`
/// child stalls. A test that inherits the shell's budget goes red over a
/// stall that is no part of what it pins.
async fn scan_within<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    prev: Option<&GitDiffSnapshot>,
    base: Option<&str>,
    pr_refresh: PrRefresh,
    // Whether this scan kicks the throttled background fetch. A read that
    // will fetch the base itself passes `false`.
    kick: bool,
    budget: Duration,
) -> GitDiffSnapshot {
    let raw_branch = match rev_parse_gate(cwd, budget).await {
        Ok(branch) => branch,
        // Both the not-a-repo and scanner-crash cases collapse to the
        // same all-Clean snapshot, differing only by `repo_gate` so the
        // renderer can tell "not a git repository" from "scanner
        // unhealthy."
        Err(repo_gate) => {
            return GitDiffSnapshot {
                branch: GitBranch::NoRepo,
                default_branch: None,
                repo_gate,
                pushed_sha: None,
                worktree: LayerState::Clean,
                branch_ahead: LayerState::Clean,
                pr: None,
                closes: Vec::new(),
                pr_fetched_at: None,
            };
        }
    };
    let detached = raw_branch == "HEAD";
    let branch = if detached {
        GitBranch::Detached
    } else if raw_branch.is_empty() {
        GitBranch::Unknown
    } else {
        GitBranch::Named(raw_branch.clone())
    };

    let default_branch = resolve_default_branch(cwd, budget).await;
    let on_default = checked_out_on_default(&raw_branch, default_branch.as_deref());
    // **An ordinary scan keeps `origin/<default>` fresh for the NEXT read.**
    // It is kicked, never awaited: this scan reads the ref as it stands and
    // nobody's read waits on the fetch. The watch's own pass also refreshes
    // inline and claims the same window, so one repository fetches at most
    // once per window whichever path gets there first - and a sessionless
    // seat, which has no loop to refresh for it, is covered here.
    if kick
        && pr_refresh == PrRefresh::Fetch
        && !detached
        && !on_default
        && let Some(default) = default_branch.as_deref()
        && default.starts_with("origin/")
    {
        kick_background_fetch(cwd, default, budget);
    }
    let dirty_probe = is_worktree_dirty(cwd, budget).await;

    // Layer 1: uncommitted edits vs HEAD. The three legal states
    // (clean tree, populated diff, scan failed) are unrepresentable
    // as illegal combinations because `LayerState` encodes them
    // directly. Either upstream failure (dirty probe OR numstat)
    // collapses to `ScanFailed` so the renderer surfaces "(scan
    // failed)" instead of silently dropping to clean-tree.
    let worktree: LayerState<GitDiffStats> = match dirty_probe {
        None => LayerState::ScanFailed,
        Some(false) => LayerState::Clean,
        Some(true) => match numstat(cwd, &["diff", "--numstat", "HEAD"], budget).await {
            Ok(mut stats) => {
                // One more read of the same tree state for the glyphs the
                // counts cannot carry; a failed pass leaves every file
                // `Modified` rather than failing the layer.
                if let Some(marks) = name_statuses(cwd, "HEAD", budget).await {
                    for file in &mut stats.files {
                        if let Some(status) = marks.get(&file.path) {
                            file.status = *status;
                        }
                    }
                }
                LayerState::Populated(stats)
            }
            Err(_) => LayerState::ScanFailed,
        },
    };

    // Layer 2: commits the branch has ahead of the default branch.
    // Skipped on detached HEAD (no meaningful "branch name" to be
    // ahead from), on the default branch itself (the diff against
    // itself is empty by construction), and when default_branch is
    // unknown (no sensible base). A clean branch sitting at the
    // merge-base with no commits ahead collapses to `Clean` so the
    // renderer doesn't show an empty layer-2 row. `ScanFailed`
    // carries the subprocess-error signal mirroring layer 1.
    let branch_ahead: LayerState<GitBranchAhead> = if !detached && !on_default {
        // A pinned read uses the sha it was given; a free one uses the ref's
        // name. The snapshot still carries the NAME for display.
        if let Some(default) = base.or(default_branch.as_deref()) {
            let range = format!("{default}...HEAD");
            match numstat(cwd, &["diff", "--numstat", &range], budget).await {
                Ok(stats) => match commit_count_in_range(cwd, default, "HEAD", budget).await {
                    None => LayerState::ScanFailed,
                    Some(0) => LayerState::Clean,
                    Some(commit_count) => {
                        // `commit_count == 0` is the cleanest signal
                        // that the branch has no commits ahead of
                        // default; gate on that alone. The earlier
                        // `&& stats.total_files == 0` clause masked
                        // the racy merge-base / force-push corner
                        // case where commit_count == 0 but stats
                        // showed files, producing a "0 commits vs
                        // main" subtitle that read oddly to the user.
                        // One more read of the same range for the glyphs
                        // the counts cannot carry, exactly as the worktree
                        // layer corrects its own; a failed pass leaves
                        // every file `Modified` rather than failing the
                        // layer.
                        let mut stats = stats;
                        if let Some(marks) = name_statuses(cwd, &range, budget).await {
                            for file in &mut stats.files {
                                if let Some(status) = marks.get(&file.path) {
                                    file.status = *status;
                                }
                            }
                        }
                        // The chain itself, for the view that draws what
                        // the branch is working on rather than only how
                        // far it runs.
                        let commits = commits_in_range(cwd, default, "HEAD", budget)
                            .await
                            .unwrap_or_default();
                        LayerState::Populated(GitBranchAhead { commit_count, stats, commits })
                    }
                },
                Err(_) => LayerState::ScanFailed,
            }
        } else {
            LayerState::Clean
        }
    } else {
        LayerState::Clean
    };

    // PR / closes only make sense for named non-default branches -
    // default branch never has a PR open against itself, detached /
    // unknown branches can't be queried by name. For eligible
    // branches, resolve HEAD's newest pushed ancestor (the lookup
    // key - never the branch name, which diverges from the PR head
    // ref for `HEAD:<name>` pushes) and reuse the prior snapshot's
    // data while the sha and fetch age say the cache still holds.
    let (pr, closes, pr_fetched_at, pushed_sha) = match &branch {
        GitBranch::Named(name) if !on_default => {
            let pushed_sha = resolve_pushed_sha(cwd, PUSHED_SHA_WALK_CAP, budget).await;
            let (pr, closes, pr_fetched_at) =
                pr_for_head_within(cwd, env, name, pushed_sha.as_deref(), prev, pr_refresh, budget)
                    .await;
            (pr, closes, pr_fetched_at, pushed_sha)
        }
        _ => (None, Vec::new(), None, None),
    };

    // repo_gate=InRepo here - we've passed the rev-parse gate and
    // every downstream subprocess (default-branch resolution,
    // numstat, gh pr) is best-effort; failures collapse to safe
    // defaults plus per-layer `LayerState::ScanFailed` so the
    // renderer can surface them at the layer level rather than
    // poisoning the overall snapshot. The ScannerFailed gate is
    // only the rev-parse-failed return at the top of this function.
    GitDiffSnapshot {
        branch,
        default_branch,
        repo_gate: RepoGate::InRepo,
        pushed_sha,
        worktree,
        branch_ahead,
        pr,
        closes,
        pr_fetched_at,
    }
}

/// Count commits in `<base>..HEAD` via `git rev-list --count`.
/// Returns `None` on any failure path (subprocess error, parse
/// failure, anomalous empty output) so callers can route the
/// signal into `LayerState::ScanFailed` instead of masking the
/// failure as 0. Each non-Ok branch emits a WARN log mirroring the
/// `gh_pr_lookup_*` pattern.
async fn commit_count_in_range(
    cwd: &Path,
    base: &str,
    head: &str,
    budget: Duration,
) -> Option<u32> {
    let range = format!("{base}..{head}");
    match run_git_within(cwd, &["rev-list", "--count", &range], budget).await {
        GitOutput::Ok(s) => match s.trim().parse::<u32>() {
            Ok(count) => Some(count),
            Err(err) => {
                tracing::warn!(
                    target: crate::logging::targets::ENV_GIT,
                    cwd = %cwd.display(),
                    event_name = "git_commit_count_parse_failed",
                    message = "rev-list --count stdout did not parse as u32",
                    outcome = "failure",
                    error = %err,
                    range = %range,
                    raw = %s.trim(),
                );
                None
            }
        },
        GitOutput::Empty => {
            // rev-list --count always emits `0\n` on success even for
            // an empty range; Empty here is anomalous (binary stdout,
            // truncation, etc.). Log separately so operators can tell
            // it apart from Failed/Oversize when triaging a vanished
            // layer-2 row.
            tracing::warn!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "git_commit_count_empty",
                message = "rev-list --count returned empty stdout (expected at minimum '0\\n')",
                outcome = "anomalous",
                range = %range,
            );
            None
        }
        GitOutput::Failed | GitOutput::Oversize => None,
    }
}

/// Resolve the ref a feature branch's "ahead" diff compares against.
/// Prefers the remote-tracking `origin/<default>` (via `origin/HEAD`,
/// else `origin/main` / `origin/master`) so a locally-stale `main`
/// can't fold already-merged commits into the diff; falls back to a
/// purely-local `main` / `master` for a repo with no `origin` remote.
/// The returned value IS the compare-ref - the `origin/` prefix carries
/// the remote-vs-local decision, so callers wanting the plain branch
/// name strip it. `None` when nothing resolves.
async fn resolve_default_branch(cwd: &Path, budget: Duration) -> Option<String> {
    // `origin/HEAD` advertises the remote's default; `--short` yields
    // e.g. `origin/main`. Keep the prefix - it's the remote-tracking
    // ref we want to compare against (GitOutput::Ok is non-empty).
    if let GitOutput::Ok(s) =
        run_git_within(cwd, &["symbolic-ref", "--short", "refs/remotes/origin/HEAD"], budget).await
    {
        return Some(s.trim().to_owned());
    }
    // No `origin/HEAD`: try the conventional remote-tracking refs, then
    // a purely-local `main` / `master` (a repo with no origin remote).
    for candidate in ["origin/main", "origin/master", "main", "master"] {
        if let GitOutput::Ok(_) =
            run_git_within(cwd, &["rev-parse", "--verify", candidate], budget).await
        {
            return Some(candidate.to_owned());
        }
    }
    // Nothing resolved. A feature-branch diff has no meaningful base,
    // so layer 2 collapses to `LayerState::Clean`. WARN so it's grep-
    // able when the user reports "the GIT section never shows my
    // branch's diff".
    tracing::warn!(
        target: crate::logging::targets::ENV_GIT,
        cwd = %cwd.display(),
        event_name = "git_default_branch_unknown",
        message = "default branch fallback exhausted (no origin/HEAD, origin/main, origin/master, main, or master)",
        outcome = "skipped",
    );
    None
}

/// Returns `Some(dirty)` when `git status` ran cleanly (`Ok` /
/// `Empty`); `None` when the subprocess hit `Failed` / `Oversize`.
/// Callers thread the `None` signal into `LayerState::ScanFailed`
/// so the renderer can surface "(scan failed)" for layer 1 instead
/// of silently rendering a clean tree when the probe upstream of
/// numstat collapsed.
async fn is_worktree_dirty(cwd: &Path, budget: Duration) -> Option<bool> {
    match run_git_within(cwd, &["status", "--porcelain=v1", "--untracked-files=no"], budget).await {
        GitOutput::Ok(s) => Some(!s.trim().is_empty()),
        GitOutput::Empty => Some(false),
        GitOutput::Failed | GitOutput::Oversize => None,
    }
}

/// Underlying `git diff --numstat` subprocess didn't return a usable
/// result. Callers map this to `LayerState::ScanFailed` so the
/// renderer can surface the failure. The variants carry enough
/// classification to triage from the log emitted by `run_git`;
/// downstream callers don't differentiate them today but the type
/// keeps the option open.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NumstatError {
    /// Subprocess crashed, timed out, or exited non-zero. See the
    /// `git_subprocess_failed` / `git_subprocess_nonzero_exit` /
    /// `git_subprocess_timeout` events.
    Subprocess,
    /// Stdout exceeded the per-command size cap. See
    /// `git_subprocess_oversize` WARN event.
    Oversize,
}

/// Run `git <args>` and parse the `--numstat` output. Returns the
/// top-`TOP_FILE_COUNT` files (sorted by total changes desc, alpha
/// tie-break) plus the full file count and overall add/remove
/// totals.
/// Returns `Err(NumstatError)` when the underlying `git` subprocess
/// hit `Failed` / `Oversize`; callers map that signal to
/// `LayerState::ScanFailed` so the renderer can distinguish "scan
/// failed" from "clean tree, nothing to show". `Empty` (exit 0, no
/// stdout) collapses to a zero-row stats block since that's the
/// legitimate "no diff to report" outcome.
async fn numstat(
    cwd: &Path,
    args: &[&str],
    budget: Duration,
) -> Result<GitDiffStats, NumstatError> {
    let raw = match run_git_within(cwd, args, budget).await {
        GitOutput::Ok(s) => s,
        GitOutput::Empty => String::new(),
        GitOutput::Failed => return Err(NumstatError::Subprocess),
        GitOutput::Oversize => return Err(NumstatError::Oversize),
    };
    Ok(stats_of(parse_numstat(&raw)))
}

/// One file list as a stats block: the full totals, then the
/// top-`TOP_FILE_COUNT` files by churn (alpha tie-break) - the count above
/// the list tells the truth either way. The shape a layer's stats and each
/// commit's own stats both take.
fn stats_of(mut files: Vec<GitDiffFile>) -> GitDiffStats {
    let total_files = files.len();
    let total_added: u32 = files.iter().fold(0u32, |acc, f| acc.saturating_add(f.added));
    let total_removed: u32 = files.iter().fold(0u32, |acc, f| acc.saturating_add(f.removed));
    files.sort_by(|a, b| {
        let a_total = a.added.saturating_add(a.removed);
        let b_total = b.added.saturating_add(b.removed);
        b_total.cmp(&a_total).then_with(|| a.path.cmp(&b.path))
    });
    files.truncate(TOP_FILE_COUNT);
    GitDiffStats { files, total_files, total_added, total_removed }
}

/// Parse `<added>\t<removed>\t<path>` lines. Skips binary entries
/// (`added` or `removed` reported as `-`). Uses `splitn(3, '\t')`
/// so paths containing tabs survive intact.
///
/// `--numstat` carries no status code, so every file lands `Modified`
/// and [`name_statuses`] corrects what it can afterwards: a file the
/// merge never reaches keeps the least-alarming class.
fn parse_numstat(raw: &str) -> Vec<GitDiffFile> {
    raw.lines().filter_map(parse_numstat_line).collect()
}

/// One `<added>\t<removed>\t<path>` line, or `None` for anything else - a
/// binary entry (counts as `-`) and a format line among them.
fn parse_numstat_line(line: &str) -> Option<GitDiffFile> {
    let mut parts = line.splitn(3, '\t');
    let added = parts.next()?;
    let removed = parts.next()?;
    let path = parts.next()?;
    if added == "-" || removed == "-" {
        return None;
    }
    let added = added.parse::<u32>().ok()?;
    let removed = removed.parse::<u32>().ok()?;
    Some(GitDiffFile {
        path: rename_target(path),
        added,
        removed,
        status: forge_primitives::git::FileStatus::Modified,
    })
}

/// A `--numstat` path resolved to the file's new name: a rename crosses as
/// `dir/{old => new}.rs`, or `old => new` when the two share no affixes, while
/// `--name-status` and the diff body both speak the new path alone - so the
/// merge in [`scan`] needs the same key from every read.
fn rename_target(path: &str) -> String {
    let Some((before, after)) = path.split_once(" => ") else {
        return path.to_owned();
    };
    match (before.rfind('{'), after.find('}')) {
        (Some(open), Some(close)) => {
            format!("{}{}{}", &before[..open], &after[..close], &after[close + 1..])
        }
        _ => after.to_owned(),
    }
}

/// One changed path's status against `target`, from
/// `git diff --name-status` - the same classifier the hunks pass
/// reads, so the glyph beside a file matches the one its diff body
/// would draw. Best-effort: a failed read leaves the counts alone.
async fn name_statuses(
    cwd: &Path,
    target: &str,
    budget: Duration,
) -> Option<HashMap<String, FileStatus>> {
    match run_git_within(cwd, &["diff", "--name-status", target], budget).await {
        GitOutput::Ok(raw) => Some(
            self::hunks::parse_name_status_entries(&raw)
                .into_iter()
                .map(|entry| (entry.path, entry.status))
                .collect(),
        ),
        GitOutput::Empty => Some(HashMap::new()),
        GitOutput::Failed | GitOutput::Oversize => None,
    }
}

/// The branch's commit chain ahead of `base`, newest first, capped at
/// [`COMMITS_WALK_CAP`]: the count says how many there are, these say
/// what they are, and each carries the files it changed. Two walks of the
/// same range - `--numstat` for the counts, `--name-status` for the marks,
/// which git will not emit together. `%h` is the short sha, `%s` the
/// subject; the unit separator cannot occur in either, so one line splits
/// cleanly.
async fn commits_in_range(
    cwd: &Path,
    base: &str,
    head: &str,
    budget: Duration,
) -> Option<Vec<forge_primitives::git::GitCommit>> {
    let range = format!("{base}..{head}");
    let cap = format!("--max-count={COMMITS_WALK_CAP}");
    let listed = match run_git_within(
        cwd,
        &["log", &cap, "--numstat", "--format=%h%x1f%s%x1f%ct", &range],
        budget,
    )
    .await
    {
        GitOutput::Ok(raw) => raw,
        GitOutput::Empty => return Some(Vec::new()),
        GitOutput::Failed | GitOutput::Oversize => return None,
    };
    // Best-effort: a failed marks walk leaves every file `Modified`
    // rather than failing the chain.
    let marks = match run_git_within(
        cwd,
        &["log", &cap, "--name-status", "--format=%h%x1f", &range],
        budget,
    )
    .await
    {
        GitOutput::Ok(raw) => parse_commit_marks(&raw),
        GitOutput::Empty | GitOutput::Failed | GitOutput::Oversize => HashMap::new(),
    };

    let mut commits = Vec::new();
    let mut open: Option<(String, String, u64)> = None;
    let mut files: Vec<GitDiffFile> = Vec::new();
    for line in listed.lines() {
        let fields: Vec<&str> = line.split('\u{1f}').collect();
        if let [sha, subject, secs] = fields[..] {
            // The format line opens a commit; the numstat lines under it
            // are that commit's own files.
            if let Some((sha, subject, time)) = open.take() {
                commits.push(forge_primitives::git::GitCommit {
                    sha,
                    subject,
                    stats: stats_of(std::mem::take(&mut files)),
                    time,
                });
            }
            open = Some((sha.to_owned(), subject.to_owned(), secs.parse().unwrap_or(0)));
        } else if let Some(file) = parse_numstat_line(line) {
            files.push(file);
        }
    }
    if let Some((sha, subject, time)) = open.take() {
        commits.push(forge_primitives::git::GitCommit {
            sha,
            subject,
            stats: stats_of(files),
            time,
        });
    }
    for commit in &mut commits {
        if let Some(by_path) = marks.get(&commit.sha) {
            for file in &mut commit.stats.files {
                if let Some(status) = by_path.get(&file.path) {
                    file.status = *status;
                }
            }
        }
    }
    Some(commits)
}

/// Marks per commit from a `--name-status` walk of the same chain: the
/// status lines carry no sha of their own, so each block belongs to the
/// format line that opened it.
fn parse_commit_marks(raw: &str) -> HashMap<String, HashMap<String, FileStatus>> {
    let mut marks: HashMap<String, HashMap<String, FileStatus>> = HashMap::new();
    let mut open: Option<String> = None;
    for line in raw.lines() {
        if let Some((sha, _)) = line.split_once('\u{1f}') {
            open = Some(sha.to_owned());
            continue;
        }
        let Some(sha) = open.as_deref() else { continue };
        // "M\tpath" and "R100\told\tnew" - the new path is the last field.
        let mut parts = line.split('\t');
        let (Some(code), Some(path)) = (parts.next(), parts.next_back()) else {
            continue;
        };
        let Some(status) = code.chars().next().and_then(hunks::status_of) else {
            continue;
        };
        marks.entry(sha.to_owned()).or_default().insert(path.to_owned(), status);
    }
    marks
}

/// How many commits the chain walk reads. Past it the count above the
/// list still tells the truth; the list is what a reader scans, not
/// what they count.
const COMMITS_WALK_CAP: u32 = 20;

/// Whether a scan's PR block is past its refetch window, so a scan that may
/// reach the network would look it up again.
///
/// The seat's watch is the scan that refreshes it: the read path serves what
/// the store holds, and this is the leg that makes the watch's next pass scan
/// even when the tree itself is still - otherwise the correction would wait
/// for the tree's own staleness window (10 s) rather than its own (~1 s).
pub fn pr_refresh_due(diff: &GitDiffSnapshot, now: std::time::SystemTime) -> bool {
    let GitBranch::Named(name) = &diff.branch else {
        return false;
    };
    // The scan's own eligibility: a branch checked out ON the default has no
    // PR against itself, and its block is never looked up - so its absent
    // attempt time is "nothing to fetch", not "due".
    !checked_out_on_default(name, diff.default_branch.as_deref())
        && diff.pr_fetched_at.is_none_or(|at| {
            !now.duration_since(at).is_ok_and(|elapsed| elapsed < PR_REFETCH_INTERVAL)
        })
}

/// Whether `branch` is the default a scan measures against, compared as plain
/// names: `default_branch` may be a remote-tracking ref (`origin/main`), and a
/// checked-out `main` is the default rather than a feature branch ahead of it.
fn checked_out_on_default(branch: &str, default: Option<&str>) -> bool {
    default
        .map(|default| default.strip_prefix("origin/").unwrap_or(default))
        .is_some_and(|default| default == branch)
}

/// Whether the previous snapshot's PR block is still reusable: the
/// branch name and pushed sha both match the current scan AND the
/// last lookup attempt is less than [`PR_REFETCH_INTERVAL`] old.
/// The sha leg - not the branch name - is what makes a PR opened or
/// pushed after an earlier scan show up; the timer leg unsticks a
/// cached `None` (a scan that ran before the PR existed would
/// otherwise hold `pr: None` for the branch's whole life).
fn pr_cache_fresh(
    prev: Option<&GitDiffSnapshot>,
    branch: &str,
    pushed_sha: Option<&str>,
    now: std::time::SystemTime,
) -> bool {
    let Some(prev) = prev else {
        return false;
    };
    let GitBranch::Named(prev_name) = &prev.branch else {
        return false;
    };
    prev_name == branch
        && prev.pushed_sha.as_deref() == pushed_sha
        && prev.pr_fetched_at.is_some_and(|at| {
            now.duration_since(at).is_ok_and(|elapsed| elapsed < PR_REFETCH_INTERVAL)
        })
}

/// What one PR lookup produced. `Failed` is kept apart from `None`
/// so a transient `gh` blip cannot evict a known-good cached PR.
enum PrLookup {
    Found(GitPrInfo, Vec<GitIssueRef>),
    /// The query ran and answered "no open PR containing this sha".
    None,
    /// The query did not complete (gh missing / auth / rate limit /
    /// unparseable response).
    Failed,
}

/// Resolve PR info for the branch from `prev`'s cache when fresh,
/// else query fresh by the pushed sha. Stamps `pr_fetched_at` on the
/// returned attempt time so failed / empty lookups rate-limit too.
/// A `Failed` lookup keeps `prev`'s PR when branch and sha still
/// match - a gh blip must not flash the row off.
///
/// The `gh` children it spawns are bounded by `budget`; see
/// [`scan_within`] for why a test spells its own.
async fn pr_for_head_within<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    branch: &str,
    pushed_sha: Option<&str>,
    prev: Option<&GitDiffSnapshot>,
    pr_refresh: PrRefresh,
    budget: Duration,
) -> (Option<GitPrInfo>, Vec<GitIssueRef>, Option<std::time::SystemTime>) {
    let now = std::time::SystemTime::now();
    let cached = match prev {
        Some(prev) if pr_cache_fresh(Some(prev), branch, pushed_sha, now) => Some(prev),
        _ => None,
    };
    if let Some(prev) = cached {
        return (prev.pr.clone(), prev.closes.clone(), prev.pr_fetched_at);
    }
    // **A read somebody waits on does not go to the network.** The
    // PR is the one field a scan cannot answer locally, and its lookup is a
    // TLS + API round trip - measured at 0.9 s, which was the whole warm
    // open's remainder. The stored attempt time stands, so the seat's loop
    // (which nobody waits on) finds the cache stale and refreshes it on its
    // next pass.
    if pr_refresh == PrRefresh::ServeCached {
        return match prev {
            Some(prev) if same_branch_and_sha(prev, branch, pushed_sha) => {
                (prev.pr.clone(), prev.closes.clone(), prev.pr_fetched_at)
            }
            _ => (None, Vec::new(), None),
        };
    }
    match fetch_pr_for_pushed_sha(cwd, env, pushed_sha, budget).await {
        PrLookup::Found(pr, closes) => (Some(pr), closes, Some(now)),
        PrLookup::None => (None, Vec::new(), Some(now)),
        PrLookup::Failed => match prev {
            Some(prev) if same_branch_and_sha(prev, branch, pushed_sha) => {
                (prev.pr.clone(), prev.closes.clone(), Some(now))
            }
            _ => (None, Vec::new(), Some(now)),
        },
    }
}

/// Whether `prev` describes the same branch and pushed sha the
/// current scan resolved - the condition under which a failed
/// lookup may keep `prev`'s cached PR.
fn same_branch_and_sha(prev: &GitDiffSnapshot, branch: &str, pushed_sha: Option<&str>) -> bool {
    matches!(&prev.branch, GitBranch::Named(prev_name) if prev_name == branch)
        && prev.pushed_sha.as_deref() == pushed_sha
}

/// Resolve HEAD's newest pushed ancestor: walk HEAD's first-parent
/// history newest-first (bounded by `cap`) and stop at the first sha
/// the remote-tracking refs already contain. `None` when nothing of
/// HEAD's ancestry is pushed or no remote refs exist (a truthful
/// negative); a git failure also yields `None` but is an
/// UNRESOLVABLE lookup, not a negative - the row may flicker off for
/// one scan while git is broken and recovers on the next scan. The
/// walk happens inside git so a worktree ahead of the
/// PR tip by unpushed commits still resolves the commit the PR was
/// pushed from.
async fn resolve_pushed_sha(cwd: &Path, cap: usize, budget: Duration) -> Option<String> {
    let remote_shas = match run_git_within(
        cwd,
        &["for-each-ref", "refs/remotes", "--format=%(objectname)"],
        budget,
    )
    .await
    {
        GitOutput::Ok(raw) => raw
            .lines()
            .map(str::trim)
            .filter(|line| !line.is_empty())
            .map(str::to_owned)
            .collect::<std::collections::HashSet<_>>(),
        GitOutput::Empty | GitOutput::Failed | GitOutput::Oversize => {
            std::collections::HashSet::default()
        }
    };
    if remote_shas.is_empty() {
        return None;
    }
    let max_count = format!("--max-count={cap}");
    let walk =
        run_git_within(cwd, &["rev-list", "--first-parent", &max_count, "HEAD"], budget).await;
    let GitOutput::Ok(raw) = walk else {
        return None;
    };
    raw.lines().map(str::trim).find(|sha| remote_shas.contains(*sha)).map(str::to_owned)
}

/// Look up the open PR containing `pushed_sha`. Returns
/// [`PrLookup::None`] when the query ran and answered "no open PR"
/// (nothing pushed, or no open PR contains the sha) and
/// [`PrLookup::Failed`] on every failure path - `gh` missing,
/// unauthenticated, not a github repo, JSON parse error. Failures
/// carry a structured event so operators can grep for
/// `gh_pr_lookup_*` when triaging "PR row never shows"; the
/// subprocess shapes are at DEBUG and the parse failure at WARN. The
/// commit-not-on-remote shape logs at DEBUG instead because an
/// unpushed HEAD is a normal mid-work state, not an operator signal.
async fn fetch_pr_for_pushed_sha<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    pushed_sha: Option<&str>,
    budget: Duration,
) -> PrLookup {
    let Some(sha) = pushed_sha else {
        return PrLookup::None;
    };
    let endpoint = format!("repos/{{owner}}/{{repo}}/commits/{sha}/pulls");
    let raw = match run_gh(cwd, env, &["api", &endpoint], GhNotFound::Tolerate, budget).await {
        GitOutput::Ok(s) => s,
        GitOutput::Empty => {
            // A completed query with a definitive negative: exit 0
            // with `[]`, or the commit-not-on-remote shape run_gh
            // folds into Empty. No open PR, no cache-keep.
            return PrLookup::None;
        }
        GitOutput::Failed | GitOutput::Oversize => return PrLookup::Failed,
    };
    match pick_open_pr(&raw) {
        Err(err) => {
            tracing::warn!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "gh_pr_lookup_parse_failed",
                message = "gh api commits pulls returned unparseable json",
                outcome = "failure",
                error = %err,
                sha = %sha,
            );
            PrLookup::Failed
        }
        Ok(None) => PrLookup::None,
        Ok(Some((number, url, draft))) => {
            let closes = fetch_closing_issues(cwd, env, number, budget).await;
            PrLookup::Found(GitPrInfo { number, url, draft }, closes)
        }
    }
}

/// Select the open PR from the `commits/<sha>/pulls` REST response.
/// Multiple open PRs can contain the same commit (stacked PRs,
/// cherry-picks); the most recently `updated_at` one wins - the
/// actively-evolving PR is the right answer in stacked workflows -
/// with the higher number breaking ties. One row renders, never a
/// list.
///
/// `updated_at` compares lexicographically, which is chronological
/// for GitHub's uniform RFC3339 `Z`-suffixed format.
fn pick_open_pr(raw: &str) -> Result<Option<(u64, String, bool)>, serde_json::Error> {
    let entries: Vec<GhApiPull> = serde_json::from_str(raw)?;
    Ok(entries
        .into_iter()
        .filter(|pr| pr.state == "open")
        .max_by(|a, b| (&a.updated_at, a.number).cmp(&(&b.updated_at, b.number)))
        .map(|pr| (pr.number, pr.html_url, pr.draft)))
}

/// `commits/<sha>/pulls` REST entry shape. Only the fields the
/// selection rule reads are deserialised; `gh api` adds others
/// (`title`, `body`, …) that serde silently drops.
#[derive(Deserialize)]
struct GhApiPull {
    number: u64,
    state: String,
    html_url: String,
    updated_at: String,
    /// The row says draft or open; the two read differently to a
    /// reviewer, and GitHub's own word for it is here.
    #[serde(default)]
    draft: bool,
}

/// Fetch the PR's closing-issue list via
/// `gh pr view <n> --json closingIssuesReferences`. The REST
/// `commits/<sha>/pulls` response doesn't carry closing issues (a
/// GraphQL-only field), so this second call keeps the
/// `→ closes #M` tail rendering. Degrades to an empty list on any
/// failure - the PR row is the headline, the closes are the tail.
async fn fetch_closing_issues<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    number: u64,
    budget: Duration,
) -> Vec<GitIssueRef> {
    let number = number.to_string();
    let raw = match run_gh(
        cwd,
        env,
        &["pr", "view", &number, "--json", "closingIssuesReferences"],
        GhNotFound::Fail,
        budget,
    )
    .await
    {
        GitOutput::Ok(raw) => raw,
        GitOutput::Empty | GitOutput::Failed | GitOutput::Oversize => return Vec::new(),
    };
    let parsed: GhPrView = match serde_json::from_str(&raw) {
        Ok(parsed) => parsed,
        Err(err) => {
            tracing::warn!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "gh_pr_lookup_parse_failed",
                message = "gh pr view returned unparseable json",
                outcome = "failure",
                error = %err,
                pr = %number,
            );
            return Vec::new();
        }
    };
    parsed
        .closing_issues
        .into_iter()
        .map(|issue| GitIssueRef { number: issue.number, url: issue.url })
        .collect()
}

/// `gh pr view --json` wrapper shape.
#[derive(Deserialize)]
struct GhPrView {
    #[serde(default, rename = "closingIssuesReferences")]
    closing_issues: Vec<GhIssueEntry>,
}

/// `closingIssuesReferences` element shape. Same selective-field
/// pattern as `GhPrEntry`.
#[derive(Deserialize)]
struct GhIssueEntry {
    number: u64,
    url: String,
}

/// Result of one `git` subprocess invocation. Callers treat all
/// failure variants the same way (collapse to `NoRepo` / zero
/// stats), but the variants are split so the log captures the right
/// context - `Failed` means non-zero exit with stderr that an
/// operator might need; `Empty` is a legitimate "ran fine, no
/// output" signal (the common case for `status --porcelain` on a
/// clean tree).
pub(super) enum GitOutput {
    Ok(String),
    /// Exit 0, empty stdout (legitimate "no output to report" case).
    Empty,
    /// Timeout, spawn error, interrupted, or non-zero exit. Logged
    /// at DEBUG with stderr (truncated) inside `run_git`: a repo or a
    /// network that will not answer is the environment's state, every
    /// caller degrades to a rendered "(scan failed)", and the GIT
    /// section is re-probed on the next tick.
    Failed,
    /// Stdout exceeded [`STDOUT_SIZE_CAP`].
    Oversize,
}

/// Cap on captured stderr surfaced into the failure log. Far below the
/// stdout cap because stderr is conversational - a couple of
/// `fatal:` lines is more than enough context.
const STDERR_LOG_CAP: usize = 1024;

/// `text` cut to `cap` bytes without splitting a character: git's stderr is
/// arbitrary bytes, and a slice at a byte offset inside one panics.
fn clipped(text: &str, cap: usize) -> String {
    if text.len() <= cap {
        return text.to_owned();
    }
    let mut end = cap;
    while end > 0 && !text.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}…", &text[..end])
}

/// Spawn `git <args>` against `cwd`, await with a per-command
/// timeout, return classified output. Failures log DEBUG with the
/// captured stderr, so triage of "the GIT section shows nothing"
/// can still tell "clean tree" from "corrupt index / fatal: …"
/// without re-running, at a level that does not report a miss the
/// probe makes on every scan of a perfectly ordinary repo.
pub(super) async fn run_git(cwd: &Path, args: &[&str]) -> GitOutput {
    run_git_within(cwd, args, COMMAND_TIMEOUT).await
}

/// [`run_git`] with its command budget spelled: the shell gives a `git`
/// child [`COMMAND_TIMEOUT`], and a test gives it one of its own.
///
/// **The budget is a bound on the machine, not on git**: it is spent
/// starting the child, so a loaded machine can hold a `git`'s first
/// output past the shell's 10 s and fail a scan that is merely slow -
/// see `scan_within` for the measurement and the reproduction.
pub(super) async fn run_git_within(cwd: &Path, args: &[&str], budget: Duration) -> GitOutput {
    let mut command = git_command::tokio_git_command(cwd);
    command.arg("-C").arg(cwd).args(args).kill_on_drop(true);
    classify_git(cwd, args, command, budget).await
}

/// One git child's output, classified, with its failures logged.
async fn classify_git(
    cwd: &Path,
    args: &[&str],
    mut command: tokio::process::Command,
    budget: Duration,
) -> GitOutput {
    let fut = command.output();
    let output = match child_output(fut, budget).await {
        Some(Ok(out)) => out,
        Some(Err(err)) => {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "git_subprocess_failed",
                message = "git subprocess spawn / wait failed",
                outcome = "failure",
                error = %err,
                args = ?args,
            );
            return GitOutput::Failed;
        }
        None => {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "git_subprocess_timeout",
                message = "git subprocess timed out",
                outcome = "timeout",
                args = ?args,
            );
            return GitOutput::Failed;
        }
    };
    if !output.status.success() {
        // Non-zero exit. The renderer still collapses to NoRepo /
        // zero stats - keep the surface failure-tolerant - but record
        // the exit code + truncated stderr so triage can tell
        // "git: command not found" / "fatal: not a git repository" /
        // "fatal: index file corrupt" apart without reproducing.
        // Cwd-not-in-a-repo is one of the legitimate hits here, and
        // `origin/HEAD` is absent by design in a repo whose remote was
        // added rather than cloned, which is why the level is DEBUG.
        let stderr = String::from_utf8_lossy(&output.stderr);
        let stderr_truncated = clipped(&stderr, STDERR_LOG_CAP);
        tracing::debug!(
            target: crate::logging::targets::ENV_GIT,
            cwd = %cwd.display(),
            event_name = "git_subprocess_nonzero_exit",
            message = "git subprocess exited non-zero",
            outcome = "failure",
            exit_code = output.status.code().unwrap_or(-1),
            stderr = %stderr_truncated,
            args = ?args,
        );
        return GitOutput::Failed;
    }
    if output.stdout.len() > STDOUT_SIZE_CAP {
        tracing::warn!(
            target: crate::logging::targets::ENV_GIT,
            cwd = %cwd.display(),
            event_name = "git_subprocess_oversize",
            message = "git subprocess stdout exceeded size cap",
            outcome = "oversize",
            bytes = output.stdout.len(),
            args = ?args,
        );
        return GitOutput::Oversize;
    }
    let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
    if stdout.trim().is_empty() { GitOutput::Empty } else { GitOutput::Ok(stdout) }
}

/// How a `gh` call's "not found" failure should be classified.
/// `Tolerate` covers the shape the endpoint legitimately produces for a
/// mid-work state (an unpushed HEAD), which is a completed query with a
/// definitive negative rather than a failure.
///
/// `Fail` is a failure, not a level: every arm here records at DEBUG,
/// so the two differ in what the caller may conclude from the result
/// and not in what the log says.
enum GhNotFound {
    Fail,
    Tolerate,
}

/// Spawn `gh <args>` from `cwd` (gh derives the github repo from
/// the current working directory - there's no `-C` equivalent),
/// carrying `env` as the project's own identity.
/// Mirrors [`run_git`]'s timeout / classification / DEBUG logging so
/// failures distinguish "gh: command not found" (binary missing)
/// from "gh: To use GitHub CLI in a Git repository, please run …"
/// (not a github remote) from "no pull requests found" (legitimate
/// empty result) when triaging "PR row never shows".
async fn run_gh<S: std::hash::BuildHasher>(
    cwd: &Path,
    env: &HashMap<String, String, S>,
    args: &[&str],
    not_found: GhNotFound,
    budget: Duration,
) -> GitOutput {
    let mut command = git_command::gh_tokio_command(env);
    command.current_dir(cwd).args(args).kill_on_drop(true);
    let fut = command.output();
    let output = match child_output(fut, budget).await {
        Some(Ok(out)) => out,
        Some(Err(err)) => {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "gh_subprocess_failed",
                message = "gh subprocess spawn / wait failed",
                outcome = "failure",
                error = %err,
                args = ?args,
            );
            return GitOutput::Failed;
        }
        None => {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "gh_subprocess_timeout",
                message = "gh subprocess timed out",
                outcome = "timeout",
                args = ?args,
            );
            return GitOutput::Failed;
        }
    };
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        // The commits/<sha>/pulls endpoint answers HTTP 422 with
        // this prose when the sha isn't on the remote - an unpushed
        // HEAD mid-work, not an operator fault. That is a completed
        // query with a definitive negative, not a failure, so it
        // returns Empty (PrLookup::None downstream) rather than
        // inheriting the keep-cached-PR semantics of Failed.
        if matches!(not_found, GhNotFound::Tolerate) && stderr.contains("No commit found for SHA") {
            tracing::debug!(
                target: crate::logging::targets::ENV_GIT,
                cwd = %cwd.display(),
                event_name = "gh_pr_lookup_unpushed_sha",
                message = "commit not on the remote; PR lookup skipped",
                outcome = "no_data",
                args = ?args,
            );
            return GitOutput::Empty;
        }
        // gh exits non-zero on: missing auth (4), not a github
        // remote (1), API error (1). All collapse to "no PR" for
        // the renderer, and the log records stderr at DEBUG so triage
        // can still tell which case fired.
        let stderr_truncated = clipped(&stderr, STDERR_LOG_CAP);
        tracing::debug!(
            target: crate::logging::targets::ENV_GIT,
            cwd = %cwd.display(),
            event_name = "gh_subprocess_nonzero_exit",
            message = "gh subprocess exited non-zero",
            outcome = "failure",
            exit_code = output.status.code().unwrap_or(-1),
            stderr = %stderr_truncated,
            args = ?args,
        );
        return GitOutput::Failed;
    }
    if output.stdout.len() > STDOUT_SIZE_CAP {
        tracing::warn!(
            target: crate::logging::targets::ENV_GIT,
            cwd = %cwd.display(),
            event_name = "gh_subprocess_oversize",
            message = "gh subprocess stdout exceeded size cap",
            outcome = "oversize",
            bytes = output.stdout.len(),
            args = ?args,
        );
        return GitOutput::Oversize;
    }
    let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
    if stdout.trim().is_empty() { GitOutput::Empty } else { GitOutput::Ok(stdout) }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;
    use std::process::Command as StdCommand;
    use tempfile::TempDir;

    /// Buffer tracing output so an emitted record can be read back.
    #[derive(Clone, Default)]
    struct LogCapture(std::sync::Arc<parking_lot::Mutex<Vec<u8>>>);

    impl std::io::Write for LogCapture {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for LogCapture {
        type Writer = Self;
        fn make_writer(&'a self) -> Self::Writer {
            self.clone()
        }
    }

    /// The budget the PR-lookup cases give their `gh` child: **not the
    /// shell's own.** The wait is spent starting that child, and a loaded
    /// machine can hold a child's first write past the shell's 10 s -
    /// measured 2026-10-14, `the_pr_lookup_spawn_carries_the_projects_env`
    /// failed twice at 10.5 s and 10.9 s while a gate ran beside the suite,
    /// and a `gh` held back 11 s reproduces it on demand. The stall is the
    /// machine's, and no case here pins how long a `gh` may take, so they
    /// give the child a budget load cannot eat.
    const TEST_LOOKUP_BUDGET: Duration = Duration::from_secs(60);

    /// The scan these cases drive: the production entry with the budget the
    /// tests give it ([`TEST_GIT_BUDGET`]), so a machine's stall of a `git`
    /// child cannot red a case that is not about timing.
    async fn scan_in_test(
        cwd: &std::path::Path,
        env: &HashMap<String, String>,
        prev: Option<&GitDiffSnapshot>,
    ) -> GitDiffSnapshot {
        scan_within(cwd, env, prev, None, PrRefresh::Fetch, true, TEST_GIT_BUDGET).await
    }

    fn init_repo(dir: &TempDir, initial_branch: &str) {
        let path = dir.path();
        // The `git init -b <branch>` flag was added in git 2.28; CI
        // runners can still be on 2.25 (Ubuntu 20.04's default). To
        // stay compatible we `init` without `-b` and then point HEAD
        // at the desired branch via `symbolic-ref` - works on every
        // git version, including ones where `init.defaultBranch`
        // isn't recognised. Each call asserts `status.success()` so
        // a silent setup failure surfaces here rather than confusing
        // the assertion downstream.
        let run = |args: &[&str]| {
            let out =
                StdCommand::new("git").arg("-C").arg(path).args(args).output().expect("git ok");
            assert!(
                out.status.success(),
                "git {:?} failed in {}: stderr={}",
                args,
                path.display(),
                String::from_utf8_lossy(&out.stderr),
            );
        };
        run(&["init", "-q"]);
        run(&["symbolic-ref", "HEAD", &format!("refs/heads/{initial_branch}")]);
        run(&["config", "user.email", "test@example.com"]);
        run(&["config", "user.name", "Test"]);
        run(&["config", "commit.gpgsign", "false"]);
    }

    fn write_file(dir: &TempDir, name: &str, contents: &str) {
        let path = dir.path().join(name);
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).expect("mkdir");
        }
        std::fs::write(&path, contents).expect("write");
    }

    fn commit_all(dir: &TempDir, message: &str) {
        let path = dir.path();
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(path).args(args).output().expect("git ok");
        };
        run(&["add", "-A"]);
        run(&["commit", "-q", "-m", message]);
    }

    #[test]
    fn parse_numstat_handles_added_removed() {
        let raw = "12\t3\tsrc/foo.rs\n5\t0\tsrc/bar.rs\n";
        let parsed = parse_numstat(raw);
        assert_eq!(parsed.len(), 2);
        assert_eq!(parsed[0].path, "src/foo.rs");
        assert_eq!(parsed[0].added, 12);
        assert_eq!(parsed[0].removed, 3);
        assert_eq!(parsed[1].added, 5);
        assert_eq!(parsed[1].removed, 0);
    }

    #[test]
    fn parse_numstat_resolves_rename_paths_to_the_new_name() {
        // The forms measured from git: braces when the sides share affixes,
        // plain `old => new` when they do not. The name-status pass and the
        // diff body both speak the new path, so the merge needs it here.
        let raw = "0\t0\tdir/{a.txt => b.txt}\n1\t1\tplain.txt => renamed.txt\n3\t0\tkept.txt\n";
        let parsed = parse_numstat(raw);
        assert_eq!(parsed[0].path, "dir/b.txt", "the brace form resolves to the new path");
        assert_eq!(parsed[1].path, "renamed.txt", "the bare form takes the right side");
        assert_eq!(parsed[2].path, "kept.txt", "a path without a rename is its own");
    }

    #[test]
    fn parse_numstat_skips_binary_files() {
        let raw = "-\t-\tbin/blob.png\n4\t1\tsrc/foo.rs\n";
        let parsed = parse_numstat(raw);
        assert_eq!(parsed.len(), 1);
        assert_eq!(parsed[0].path, "src/foo.rs");
    }

    #[test]
    fn parse_numstat_handles_malformed_lines() {
        let raw = "12 missing tabs\n\n5\tnope\n4\t1\tsrc/foo.rs\n";
        let parsed = parse_numstat(raw);
        assert_eq!(parsed.len(), 1);
        assert_eq!(parsed[0].path, "src/foo.rs");
    }

    #[test]
    fn parse_numstat_keeps_tabs_inside_path_via_splitn() {
        // `splitn(3, '\t')` should leave the rest-of-line intact as
        // the path, including any tabs inside it.
        let raw = "4\t1\tsrc/weird\tpath.rs\n";
        let parsed = parse_numstat(raw);
        assert_eq!(parsed.len(), 1);
        assert_eq!(parsed[0].path, "src/weird\tpath.rs");
    }

    /// `tempfile::tempdir()` produces a dir with no `.git/`, so the
    /// repo-existence probe finds nothing and `repo_gate` is NotARepo.
    /// The distinction matters for the renderer: a healthy non-repo
    /// gets a clean hidden GIT section, while a sick scanner surfaces
    /// the unhealthy banner.
    #[tokio::test(flavor = "current_thread")]
    async fn scan_no_repo_collapses_to_not_in_repo() {
        let dir = tempfile::tempdir().expect("tempdir");
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert_eq!(snap.repo_gate, RepoGate::NotARepo);
        assert!(matches!(snap.worktree, LayerState::Clean));
        assert!(matches!(snap.branch_ahead, LayerState::Clean));
        assert!(snap.default_branch.is_none());
    }

    /// **The scan's `git` children take the budget the caller gives.** A
    /// scan whose budget runs out before `git` answers reads as a failed
    /// scan, and the same repo scans cleanly under the budget these cases
    /// use - the seam `repo_presence`, the background fetch and every
    /// `run_git` spawn now carry, production keeping `COMMAND_TIMEOUT`.
    ///
    /// **The refusing budget is ZERO, not a small millisecond count**: a
    /// child cannot answer on the first poll, so the refusal is a property
    /// of the seam rather than a race with a fast machine (a 1 ms budget
    /// was - CI's runner answered a `git` command inside it).
    #[tokio::test(flavor = "current_thread")]
    async fn the_scans_git_commands_take_the_budget_they_are_given() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "feature");
        // A commit, so the branch is born: an unborn HEAD fails rev-parse on
        // its own, which would let the refusal arm pass for the wrong reason.
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "seed");

        // Both refusals run against a `git` that cannot answer inside the
        // zero budget: the machine's own can, and did, on CI (#2249).
        let bin = tempfile::tempdir().expect("tempdir");
        stall_git_at(dir.path(), bin.path());

        let refused = scan_within(
            dir.path(),
            &HashMap::new(),
            None,
            None,
            PrRefresh::Fetch,
            true,
            Duration::ZERO,
        )
        .await;
        assert_eq!(
            refused.repo_gate,
            RepoGate::ScannerFailed,
            "a budget that runs out before git answers reads as a failed scan",
        );

        // The seam is not only the scan's first spawn: a helper called
        // deeper in refuses a zero budget the same way, so the claim has a
        // witness beyond `rev_parse_gate`'s collapse.
        let later = numstat(dir.path(), &["diff", "--numstat", "HEAD"], Duration::ZERO).await;
        assert!(
            matches!(later, Err(NumstatError::Subprocess)),
            "a later helper refuses a zero budget too: {later:?}",
        );

        crate::env::git_command::clear_git_point(dir.path());
        let scanned = scan_within(
            dir.path(),
            &HashMap::new(),
            None,
            None,
            PrRefresh::Fetch,
            true,
            TEST_GIT_BUDGET,
        )
        .await;
        assert_eq!(scanned.repo_gate, RepoGate::InRepo, "and the tests' budget scans the repo");
        assert!(
            matches!(scanned.branch, GitBranch::Named(ref name) if name == "feature"),
            "with the branch git reported: {:?}",
            scanned.branch,
        );
        assert!(
            TEST_GIT_BUDGET >= Duration::from_secs(30),
            "the tests' budget must stay far past the shell's own 10 s: a loaded machine held a \
             lookup child back for 10.5 s (measured 2026-10-14), the stall this seam exists to \
             keep out of the suite",
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn current_branch_reports_named_branch() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "feat/x");
        write_file(&dir, "README.md", "hi\n");
        commit_all(&dir, "init");
        assert_eq!(current_branch(dir.path()).await, Ok(Some("feat/x".to_owned())));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn current_branch_is_ok_none_on_detached_head() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "hi\n");
        commit_all(&dir, "init");
        let out = StdCommand::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["checkout", "--detach"])
            .output()
            .expect("git ok");
        assert!(out.status.success(), "detach: {}", String::from_utf8_lossy(&out.stderr));
        assert_eq!(
            current_branch(dir.path()).await,
            Ok(None),
            "a detached HEAD is a repo with no branch name, not a failed read",
        );
    }

    /// `git rev-parse --abbrev-ref HEAD` exits 128 both outside a work
    /// tree and when it fails inside one, so the exit code alone can't
    /// separate them. The repo-existence probe does: a plain directory
    /// has no repo, so it is a clean NotARepo rather than a scanner
    /// failure, and the renderer suppresses the GIT section.
    #[tokio::test(flavor = "current_thread")]
    async fn current_branch_reports_not_a_repo_outside_a_repo() {
        let dir = tempfile::tempdir().expect("tempdir");
        assert_eq!(current_branch(dir.path()).await, Err(RepoGate::NotARepo));
    }

    /// A path that does not exist has no repo either - git can't even
    /// chdir to it - so it lands on the same gate rather than claiming
    /// the scanner is sick. `run_git`'s record still carries git's
    /// stderr for triage.
    #[tokio::test(flavor = "current_thread")]
    async fn current_branch_reports_not_a_repo_for_a_missing_directory() {
        let dir = tempfile::tempdir().expect("tempdir");
        assert_eq!(current_branch(&dir.path().join("gone")).await, Err(RepoGate::NotARepo));
    }

    /// An unborn HEAD (`git init` with no commits) is the case the exit
    /// code cannot be trusted for: `rev-parse --abbrev-ref HEAD` fails
    /// because HEAD resolves to nothing, while `--git-dir` succeeds.
    #[tokio::test(flavor = "current_thread")]
    async fn current_branch_on_an_unborn_head_is_not_reported_as_a_non_repo() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        assert_eq!(
            current_branch(dir.path()).await,
            Err(RepoGate::ScannerFailed),
            "a repo with no commits is still a repo, so it must not be suppressed as NotARepo",
        );
    }

    /// The foreign-repo hazard, end to end: with `GIT_DIR` /
    /// `GIT_WORK_TREE` / `GIT_COMMON_DIR` exported at a foreign repo -
    /// the shape a git hook or a `rebase --exec` shell hands down -
    /// every routed spawn must answer from its own cwd, not the
    /// exported repo. The layer cannot be injected in-process (the
    /// env is process-global and `unsafe` to set under this edition),
    /// so the probe re-execs this test binary as a child with the
    /// three variables in its ambient environment and asserts the
    /// child's answers. Unscrubbed, `git -C <plain dir> rev-parse
    /// --git-dir` exits 0 answering from the foreign repo, so every
    /// assertion below fails against the unscrubbed shape.
    #[tokio::test(flavor = "current_thread")]
    async fn git_layer_answers_from_cwd_not_an_exported_git_dir() {
        const PROBE: &str = "FORGE_SCRUB_PROBE";
        const PROBE_OK: &str = "FORGE_SCRUB_PROBE_OK";
        const PROBE_TEST: &str =
            "env::git_diff::tests::git_layer_answers_from_cwd_not_an_exported_git_dir";

        let foreign = tempfile::tempdir().expect("foreign tempdir");
        let host = tempfile::tempdir().expect("host tempdir");
        let plain = tempfile::tempdir().expect("plain tempdir");
        init_repo(&foreign, "exported-only-branch");
        write_file(&foreign, "README.md", "foreign\n");
        commit_all(&foreign, "foreign init");
        init_repo(&host, "cwd-repo-branch");
        write_file(&host, "README.md", "host\n");
        commit_all(&host, "host init");

        if std::env::var(PROBE).is_err() {
            // Parent mode: export the three repo-location variables at
            // the foreign repo and run this test again as a child.
            let out = StdCommand::new(std::env::current_exe().expect("current_exe"))
                .args(["--exact", PROBE_TEST, "--test-threads=1", "--nocapture"])
                .env(PROBE, "1")
                .env("GIT_DIR", foreign.path().join(".git"))
                .env("GIT_WORK_TREE", foreign.path())
                .env("GIT_COMMON_DIR", foreign.path().join(".git"))
                .env("FORGE_SCRUB_HOST", host.path())
                .env("FORGE_SCRUB_PLAIN", plain.path())
                .output()
                .expect("spawn probe");
            assert!(
                out.status.success(),
                "probe failed under exported GIT_DIR: {}",
                String::from_utf8_lossy(&out.stdout),
            );
            assert!(
                String::from_utf8_lossy(&out.stdout).contains(PROBE_OK),
                "the probe never ran (filter matched nothing?)",
            );
            return;
        }

        // Child mode: the three variables are ambient here. Fixtures
        // already exist (git fixture calls would otherwise init the
        // foreign repo); only production spawns run.
        let host = PathBuf::from(std::env::var("FORGE_SCRUB_HOST").expect("host path"));
        let plain = PathBuf::from(std::env::var("FORGE_SCRUB_PLAIN").expect("plain path"));

        assert_eq!(
            repo_presence(&plain).await,
            RepoPresence::Absent,
            "a plain dir with GIT_DIR exported must not answer as a repo",
        );
        assert!(
            !crate::env::worktree::is_git_repo(&plain),
            "is_git_repo must not inherit the exported repo",
        );
        let GitOutput::Ok(branch) = run_git(&host, &["rev-parse", "--abbrev-ref", "HEAD"]).await
        else {
            panic!("run_git failed against a live repo");
        };
        assert_eq!(
            branch.trim(),
            "cwd-repo-branch",
            "run_git must answer from -C, not the exported GIT_DIR",
        );
        println!("{PROBE_OK}");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn scan_clean_default_branch_has_no_layers() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "hello\n");
        commit_all(&dir, "init");
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert_eq!(snap.repo_gate, RepoGate::InRepo);
        assert!(matches!(snap.worktree, LayerState::Clean), "clean tree → no layer 1");
        assert!(matches!(snap.branch_ahead, LayerState::Clean), "on default → no layer 2");
        assert_eq!(snap.default_branch.as_deref(), Some("main"));
    }

    /// **A scan that answers "no PR" still stamps its attempt (#2536).** The
    /// watch refreshes on `pr_refresh_due`, which reads that stamp: a lookup
    /// that answered nothing (here, an unpushed branch) must not leave the
    /// cache looking never-attempted, or every held seat on such a branch
    /// would rescan and spawn `gh` on every poke.
    #[tokio::test(flavor = "current_thread")]
    async fn a_fetch_scan_stamps_an_attempt_when_there_is_no_pr() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "feature");
        write_file(&dir, "README.md", "hello\n");
        commit_all(&dir, "init");
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert!(
            matches!(snap.branch, GitBranch::Named(ref name) if name == "feature"),
            "the scan reads the feature branch",
        );
        assert_eq!(snap.pr, None, "an unpushed branch has no PR to show");
        assert!(snap.pr_fetched_at.is_some(), "but the attempt is stamped");
        assert!(
            !pr_refresh_due(&snap, std::time::SystemTime::now()),
            "so the cache is not due on every poke",
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn scan_dirty_default_branch_populates_worktree_only() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "first\n");
        commit_all(&dir, "init");
        // Dirty: modify the tracked file without committing.
        write_file(&dir, "README.md", "second\nthird\n");
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        let LayerState::Populated(stats) = &snap.worktree else {
            panic!("layer 1 populated on dirty tree");
        };
        assert!(
            matches!(snap.branch_ahead, LayerState::Clean),
            "on default branch → layer 2 stays empty"
        );
        assert_eq!(stats.total_files, 1);
        assert_eq!(stats.files[0].path, "README.md");
        assert!(stats.files[0].added >= 1);
        assert_eq!(stats.total_added, stats.files[0].added);
        assert_eq!(stats.total_removed, stats.files[0].removed);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn scan_clean_feature_branch_populates_branch_ahead_only() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "first\n");
        commit_all(&dir, "init");
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        run(&["checkout", "-q", "-b", "feat/x"]);
        write_file(&dir, "feat.rs", "fn x() {}\n");
        commit_all(&dir, "feat commit");

        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert!(matches!(snap.worktree, LayerState::Clean), "clean tree → no layer 1");
        let LayerState::Populated(ahead) = &snap.branch_ahead else {
            panic!("layer 2 populated on feature branch");
        };
        assert_eq!(ahead.commit_count, 1, "one commit on feature branch beyond main");
        assert_eq!(ahead.stats.total_files, 1);
        assert_eq!(ahead.stats.files[0].path, "feat.rs");
        assert_eq!(snap.default_branch.as_deref(), Some("main"));
        assert_eq!(ahead.stats.total_added, ahead.stats.files[0].added);
        assert_eq!(ahead.stats.total_removed, ahead.stats.files[0].removed);
    }

    /// Feature branch with uncommitted edits: both layers must
    /// populate independently. Layer 1 carries the dirty tree;
    /// layer 2 carries the commit(s) on the feature branch.
    #[tokio::test(flavor = "current_thread")]
    async fn scan_dirty_feature_branch_populates_both_layers() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "first\n");
        commit_all(&dir, "init");
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        run(&["checkout", "-q", "-b", "feat/x"]);
        write_file(&dir, "feat.rs", "fn x() {}\n");
        commit_all(&dir, "feat commit");
        // Dirty the worktree AFTER the feature commit.
        write_file(&dir, "feat.rs", "fn x() {}\nfn y() {}\n");

        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        let LayerState::Populated(worktree) = &snap.worktree else {
            panic!("layer 1 populated on dirty tree");
        };
        let LayerState::Populated(ahead) = &snap.branch_ahead else {
            panic!("layer 2 populated on feature branch");
        };
        // Layer 1: in-progress edits to feat.rs.
        assert_eq!(worktree.total_files, 1);
        assert_eq!(worktree.files[0].path, "feat.rs");
        // Layer 2: the committed-and-unmerged work (the original
        // `feat commit` against main).
        assert_eq!(ahead.commit_count, 1);
        assert_eq!(ahead.stats.total_files, 1);
        assert_eq!(ahead.stats.files[0].path, "feat.rs");
        assert_eq!(
            ahead.stats.files[0].status,
            forge_primitives::git::FileStatus::Added,
            "the range's added file wears Added, not the numstat default"
        );
        // And the chain carries each commit's own files, marked the same way.
        assert_eq!(ahead.commits.len(), 1, "the chain is the one feature commit");
        assert_eq!(ahead.commits[0].stats.files[0].path, "feat.rs", "the commit's own file");
        assert_eq!(
            ahead.commits[0].stats.files[0].status,
            forge_primitives::git::FileStatus::Added,
            "with the commit's own mark"
        );
        assert_eq!(ahead.commits[0].stats.files[0].added, 1, "and the commit's own count");
        assert!(ahead.commits[0].time > 0, "the commit carries its instant");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn scan_detached_dirty_populates_worktree_no_branch_ahead() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "first\n");
        commit_all(&dir, "init");
        write_file(&dir, "README.md", "second\n");
        commit_all(&dir, "second");
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        run(&["checkout", "-q", "HEAD~1"]);
        // Dirty in detached state.
        write_file(&dir, "README.md", "third\n");

        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert!(snap.worktree.is_populated(), "detached + dirty → layer 1 populated");
        assert!(
            matches!(snap.branch_ahead, LayerState::Clean),
            "detached HEAD has no branch name, layer 2 stays empty"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn scan_detached_clean_has_no_layers() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "first\n");
        commit_all(&dir, "init");
        write_file(&dir, "README.md", "second\n");
        commit_all(&dir, "second");
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        run(&["checkout", "-q", "HEAD~1"]);
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert_eq!(snap.repo_gate, RepoGate::InRepo);
        assert!(matches!(snap.worktree, LayerState::Clean));
        assert!(matches!(snap.branch_ahead, LayerState::Clean));
        assert_eq!(snap.pr, None, "detached HEAD resolves no PR row");
        assert_eq!(snap.pushed_sha, None, "the pushed-sha walk is skipped when detached");
    }

    /// The pushed-sha walk resolves a DIVERGED upstream: the remote
    /// has commits HEAD lacks (a sibling commit the remote gained),
    /// and the walk still stops at HEAD's newest pushed ancestor -
    /// never the sibling, which is outside HEAD's ancestry.
    #[tokio::test(flavor = "current_thread")]
    async fn pushed_sha_walk_resolves_pushed_ancestor_when_upstream_diverged() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "audit/x");
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "pushed commit");
        let pushed = rev_parse_head(&dir);
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        run(&["update-ref", "refs/remotes/origin/audit/x", &pushed]);
        // A remote-only sibling commit: same parent as the pushed
        // tip, so the remote history diverged - HEAD lacks it.
        let out = StdCommand::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["commit-tree", "HEAD^{tree}", "-p", &pushed, "-m", "remote-only"])
            .output()
            .expect("git ok");
        assert!(out.status.success(), "commit-tree failed");
        let remote_only = String::from_utf8_lossy(&out.stdout).trim().to_owned();
        run(&["update-ref", "refs/remotes/origin/main", &remote_only]);
        write_file(&dir, "more.rs", "y\n");
        commit_all(&dir, "unpushed commit");

        assert_eq!(
            resolve_pushed_sha(dir.path(), 500, TEST_GIT_BUDGET).await.as_deref(),
            Some(pushed.as_str()),
            "the sibling remote-only commit is outside HEAD's ancestry and must not resolve"
        );
    }

    /// The unstick trap pinned at the cache-decision level: a cached
    /// `None` (scan ran before the PR was opened) goes stale on the
    /// same timer and refetches even with branch and sha unchanged.
    #[test]
    fn pr_cache_fresh_refetches_a_cached_none_when_stale() {
        let old = std::time::SystemTime::now()
            .checked_sub(PR_REFETCH_INTERVAL + std::time::Duration::from_secs(1))
            .expect("wall clock is at least PR_REFETCH_INTERVAL + 1s past boot");
        let prev = GitDiffSnapshot {
            branch: GitBranch::Named("feat/x".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: Some("aaaa".into()),
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: None,
            closes: Vec::new(),
            pr_fetched_at: Some(old),
        };
        assert!(
            !pr_cache_fresh(Some(&prev), "feat/x", Some("aaaa"), std::time::SystemTime::now()),
            "stale None must refetch so a PR opened after the first scan shows up"
        );
    }

    /// An unborn HEAD (fresh init, no commits): the scan collapses to
    /// the scanner-failed gate and the PR side stays dark - no row,
    /// no panic.
    #[tokio::test(flavor = "current_thread")]
    async fn scan_on_an_unborn_head_has_no_pr_row() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert_eq!(snap.repo_gate, RepoGate::ScannerFailed);
        assert_eq!(snap.pr, None);
        assert_eq!(snap.pushed_sha, None);
        assert_eq!(snap.pr_fetched_at, None);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn scan_dirty_default_branch_aggregates_totals_across_files() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "a.txt", "1\n2\n3\n");
        write_file(&dir, "b.txt", "1\n");
        commit_all(&dir, "init");
        // Add a line to a.txt, replace b.txt with a longer body, add
        // a new tracked file. Total adds should sum across all
        // changed files.
        write_file(&dir, "a.txt", "1\n2\n3\n4\n");
        write_file(&dir, "b.txt", "x\ny\nz\n");
        write_file(&dir, "c.txt", "new\n");
        // Stage c.txt so it shows up in `git diff HEAD`.
        StdCommand::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["add", "c.txt"])
            .output()
            .expect("git ok");

        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        let LayerState::Populated(stats) = &snap.worktree else {
            panic!("dirty tree → layer 1 populated");
        };
        assert_eq!(stats.total_files, 3);
        // Sum of per-file added/removed equals the overall total.
        let per_file_added: u32 = stats.files.iter().map(|f| f.added).sum();
        let per_file_removed: u32 = stats.files.iter().map(|f| f.removed).sum();
        assert_eq!(stats.total_added, per_file_added);
        assert_eq!(stats.total_removed, per_file_removed);
        // At least one row in each direction (we added 1+3+1 lines
        // and removed 1).
        assert!(stats.total_added >= 5);
        assert!(stats.total_removed >= 1);
    }

    #[test]
    fn top_files_sort_by_total_changes_then_alpha() {
        let mut files = [
            GitDiffFile {
                path: "small.rs".into(),
                added: 1,
                removed: 0,
                status: FileStatus::Modified,
            },
            GitDiffFile {
                path: "big-b.rs".into(),
                added: 50,
                removed: 10,
                status: FileStatus::Modified,
            },
            GitDiffFile {
                path: "big-a.rs".into(),
                added: 50,
                removed: 10,
                status: FileStatus::Modified,
            },
            GitDiffFile {
                path: "medium.rs".into(),
                added: 10,
                removed: 5,
                status: FileStatus::Modified,
            },
        ];
        files.sort_by(|a, b| {
            let a_total = a.added.saturating_add(a.removed);
            let b_total = b.added.saturating_add(b.removed);
            b_total.cmp(&a_total).then_with(|| a.path.cmp(&b.path))
        });
        assert_eq!(files[0].path, "big-a.rs");
        assert_eq!(files[1].path, "big-b.rs");
        assert_eq!(files[2].path, "medium.rs");
        assert_eq!(files[3].path, "small.rs");
    }

    /// Cache hit: same branch, same pushed sha, fetched inside the
    /// 5-minute window. The reuse that keeps steady-state 10 s scans
    /// gh-free.
    #[test]
    fn pr_cache_fresh_reuses_when_branch_sha_and_age_hold() {
        let prev = snapshot_with_pr(42);
        assert!(pr_cache_fresh(Some(&prev), "feat/x", Some("aaaa"), std::time::SystemTime::now()));
    }

    /// Refetch trigger: the pushed sha moved (new push / rebase
    /// landed on the remote), so the cached PR may no longer be the
    /// one containing the branch's pushed tip.
    #[test]
    fn pr_cache_fresh_refetches_when_pushed_sha_changes() {
        let prev = snapshot_with_pr(42);
        assert!(
            !pr_cache_fresh(Some(&prev), "feat/x", Some("bbbb"), std::time::SystemTime::now()),
            "sha change must invalidate the cached PR"
        );
    }

    /// Refetch trigger: the fetch is older than the interval, even
    /// with branch and sha unchanged - unsticks a cached `None` from
    /// before the PR was opened.
    #[test]
    fn pr_cache_fresh_refetches_when_fetch_is_stale() {
        let prev = snapshot_with_pr(42);
        let old = std::time::SystemTime::now()
            .checked_sub(PR_REFETCH_INTERVAL + std::time::Duration::from_secs(1))
            .expect("monotonic wall clock is at least PR_REFETCH_INTERVAL + 1s past boot");
        assert!(
            !pr_cache_fresh(Some(&prev), "feat/x", Some("aaaa"), old),
            "stale fetch must refetch so a PR opened after the first scan shows up"
        );
    }

    /// **A serving lookup answers only what belongs to this branch and sha.**
    /// The store's PR is served when it is this row's; another branch's or
    /// another push's is not - the shape a branch switch or a push makes, and
    /// serving it would put one seat's PR on another's row.
    #[tokio::test(flavor = "current_thread")]
    async fn serve_cached_answers_nothing_for_another_branch_or_sha() {
        let dir = tempfile::tempdir().expect("tempdir");
        let prev = snapshot_with_pr_at(42, Some(600));
        let budget = TEST_GIT_BUDGET;

        let (pr, closes, at) = pr_for_head_within(
            dir.path(),
            &HashMap::new(),
            "feat/y",
            Some("aaaa"),
            Some(&prev),
            PrRefresh::ServeCached,
            budget,
        )
        .await;
        assert_eq!(pr, None, "another branch's PR is not this row's");
        assert!(closes.is_empty(), "nor are its closing issues");
        assert_eq!(at, None, "and no attempt time is claimed for it");

        let (pr, _, at) = pr_for_head_within(
            dir.path(),
            &HashMap::new(),
            "feat/x",
            Some("bbbb"),
            Some(&prev),
            PrRefresh::ServeCached,
            budget,
        )
        .await;
        assert_eq!(pr, None, "another push's PR is not this row's");
        assert_eq!(at, None, "and no attempt time is claimed for it");

        let (pr, _, at) = pr_for_head_within(
            dir.path(),
            &HashMap::new(),
            "feat/x",
            Some("aaaa"),
            Some(&prev),
            PrRefresh::ServeCached,
            budget,
        )
        .await;
        assert_eq!(pr.map(|found| found.number), Some(42), "its own PR is served");
        assert_eq!(at, prev.pr_fetched_at, "with the attempt time the store holds");
    }

    /// **The watch's PR-due leg follows the scan's own eligibility.**
    /// A feature branch past the window is due; a fresh attempt is not; and a
    /// checkout ON the default has no PR against itself, so its absent attempt
    /// time reads "nothing to fetch" rather than "due" - the shape that had
    /// every on-default fetch scan re-read the tree.
    #[test]
    fn pr_refresh_due_follows_the_scans_own_eligibility() {
        // The attempts are stamped before `now`, so the fresh one is inside
        // the window rather than a clock tick past it.
        let fresh = snapshot_with_pr(42);
        let stale = snapshot_with_pr_at(42, Some(600));
        let mut on_default = snapshot_with_pr_at(42, None);
        on_default.branch = GitBranch::Named("main".into());
        let now = std::time::SystemTime::now();

        assert!(pr_refresh_due(&stale, now), "a stale attempt on a feature branch is due");
        assert!(!pr_refresh_due(&fresh, now), "an attempt inside the window is not due");
        assert!(
            !pr_refresh_due(&on_default, now),
            "a branch checked out on the default has nothing to look up",
        );
        // A feature branch with no attempt on record is what a first open or a
        // branch switch writes; it is due, or the cache would never fill.
        assert!(
            pr_refresh_due(&snapshot_with_pr_at(42, None), now),
            "a feature branch never attempted is due",
        );
    }

    /// `pr_for_head_within` short-circuits the `gh` calls when the cache
    /// is fresh - even when the `cwd` would make a real `gh` invocation
    /// fail (tempdir has no github remote). The returned `pr` /
    /// `closes` MUST be clones of `prev`'s, proving the cache hit.
    #[tokio::test(flavor = "current_thread")]
    async fn pr_for_head_reuses_prev_when_cache_fresh() {
        let dir = tempfile::tempdir().expect("tempdir");
        let pr = GitPrInfo { number: 42, url: "https://example/pull/42".into(), draft: false };
        let closes = vec![GitIssueRef { number: 7, url: "https://example/issues/7".into() }];
        let prev = GitDiffSnapshot {
            branch: GitBranch::Named("feat/x".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: Some("aaaa".into()),
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: Some(pr.clone()),
            closes: closes.clone(),
            pr_fetched_at: Some(std::time::SystemTime::now()),
        };

        let (got_pr, got_closes, got_at) = pr_for_head_within(
            dir.path(),
            &HashMap::new(),
            "feat/x",
            Some("aaaa"),
            Some(&prev),
            PrRefresh::Fetch,
            COMMAND_TIMEOUT,
        )
        .await;
        assert_eq!(got_pr, Some(pr));
        assert_eq!(got_closes, closes);
        assert_eq!(got_at, prev.pr_fetched_at);
    }

    /// A stale cache refetches - and when the refetch FAILS (tempdir
    /// makes `gh` fail whether or not it is installed), the known-good
    /// cached PR survives instead of flashing off: a gh blip must not
    /// evict the row.
    #[tokio::test(flavor = "current_thread")]
    async fn pr_for_head_keeps_cached_pr_when_refetch_fails() {
        let dir = tempfile::tempdir().expect("tempdir");
        let stale_at = std::time::SystemTime::now()
            .checked_sub(PR_REFETCH_INTERVAL + std::time::Duration::from_secs(1))
            .expect("wall clock is at least PR_REFETCH_INTERVAL + 1s past boot");
        let prev = GitDiffSnapshot {
            branch: GitBranch::Named("feat/x".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: Some("aaaa".into()),
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: Some(GitPrInfo { number: 42, url: "https://example/pull/42".into(), draft: false }),
            closes: Vec::new(),
            pr_fetched_at: Some(stale_at),
        };

        let (got_pr, _got_closes, got_at) = pr_for_head_within(
            dir.path(),
            &HashMap::new(),
            "feat/x",
            Some("aaaa"),
            Some(&prev),
            PrRefresh::Fetch,
            COMMAND_TIMEOUT,
        )
        .await;
        assert_eq!(got_pr.as_ref().map(|p| p.number), Some(42), "failed refetch keeps the row");
        let now = std::time::SystemTime::now();
        assert!(
            got_at.is_some_and(|at| now.duration_since(at).is_ok_and(|d| d.as_secs() < 60)),
            "the refetch attempt must be stamped so the next retry waits for the interval"
        );
    }

    /// A failed refetch after a BRANCH change has no matching cached
    /// PR to keep - the row clears rather than showing the previous
    /// branch's PR.
    #[tokio::test(flavor = "current_thread")]
    async fn pr_for_head_clears_failed_refetch_when_branch_changed() {
        let dir = tempfile::tempdir().expect("tempdir");
        let stale_at = std::time::SystemTime::now()
            .checked_sub(PR_REFETCH_INTERVAL + std::time::Duration::from_secs(1))
            .expect("wall clock is at least PR_REFETCH_INTERVAL + 1s past boot");
        let prev = GitDiffSnapshot {
            branch: GitBranch::Named("feat/x".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: Some("aaaa".into()),
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: Some(GitPrInfo { number: 42, url: "https://example/pull/42".into(), draft: false }),
            closes: Vec::new(),
            pr_fetched_at: Some(stale_at),
        };

        let (got_pr, got_closes, got_at) = pr_for_head_within(
            dir.path(),
            &HashMap::new(),
            "feat/y",
            Some("aaaa"),
            Some(&prev),
            PrRefresh::Fetch,
            COMMAND_TIMEOUT,
        )
        .await;
        assert_eq!(got_pr, None, "a failed lookup must not show another branch's PR");
        assert!(got_closes.is_empty());
        assert!(got_at.is_some());
    }

    /// The spawn a project's PR read makes carries that project's own
    /// identity: the env handed to `scan` reaches the `gh` child. Which
    /// ambient variables come off is pinned in `git_command`; this pins
    /// that the env is threaded rather than dropped on the way down.
    /// Real git for the repo, and a `gh` of our own the child finds
    /// through the PATH the project env carries, so the identity it was
    /// handed is readable afterwards.
    #[tokio::test(flavor = "current_thread")]
    async fn the_pr_lookup_spawn_carries_the_projects_env() {
        let recorded = scan_with_own_gh(Duration::ZERO, TEST_LOOKUP_BUDGET)
            .await
            .expect("the scan's gh child ran and recorded the identity it was handed");
        assert_eq!(recorded, "project-token", "the PR lookup spawns gh as the project");
    }

    /// **The PR lookup's wait is the budget it is given.** A `gh` child held
    /// back past a small budget is refused - the lookup never lands - and
    /// the same child is waited out by the budget these cases use, which is
    /// the seam `the_pr_lookup_spawn_carries_the_projects_env` rests on: the
    /// wait belongs to the machine, not to the lookup.
    #[tokio::test(flavor = "current_thread")]
    async fn the_pr_lookup_is_bounded_by_the_budget_it_is_given() {
        let refused = scan_with_own_gh(Duration::from_secs(3), Duration::from_secs(1)).await;
        assert!(
            refused.is_none(),
            "a budget that runs out before the child answers leaves the lookup unrun: {refused:?}",
        );

        let waited = scan_with_own_gh(Duration::from_secs(3), TEST_LOOKUP_BUDGET).await;
        assert_eq!(
            waited.as_deref(),
            Some("project-token"),
            "and the tests' budget waits that child out",
        );
        assert!(
            TEST_LOOKUP_BUDGET >= Duration::from_secs(30),
            "the tests' budget must stay far past the shell's own 10 s: a loaded machine held a \
             lookup child back for 10.5 s (measured 2026-10-14), which is the stall this seam \
             exists to keep out of the suite",
        );
    }

    /// The gh-identity fixture: a repo whose tip reads as pushed, and a `gh`
    /// of our own on the project's PATH that records the identity it was
    /// handed after holding its answer back `stall`. Returns what the scan
    /// read from that record (`None` when the child never wrote it).
    ///
    /// **The stall spells `/bin/sleep`**: the child's PATH is the fixture's
    /// own bin directory alone, so a bare `sleep` is not found and the stall
    /// silently never happens.
    async fn scan_with_own_gh(stall: Duration, budget: Duration) -> Option<String> {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "feature");
        let run = |args: &[&str]| {
            let out =
                StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git");
            assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
        };
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "pushed commit");
        // Simulate the push: the tip is on a remote-tracking ref and the
        // checkout is not the default branch, so the scan runs its PR
        // lookup against it.
        run(&["update-ref", "refs/remotes/origin/main", "HEAD"]);
        run(&["symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main"]);

        let bin = dir.path().join("bin");
        std::fs::create_dir_all(&bin).expect("bin dir");
        let seen = dir.path().join("gh-seen-identity");
        let gh = bin.join("gh");
        let hold = if stall.is_zero() {
            String::new()
        } else {
            format!("/bin/sleep {}\n", stall.as_secs_f64())
        };
        std::fs::write(
            &gh,
            format!(
                "#!/bin/sh\n{hold}printf '%s' \"$GH_TOKEN\" > {}\nprintf '[]'\n",
                seen.display()
            ),
        )
        .expect("write gh");
        std::fs::set_permissions(&gh, std::fs::Permissions::from_mode(0o755)).expect("chmod gh");
        let env = HashMap::from([
            ("PATH".to_owned(), bin.display().to_string()),
            ("GH_TOKEN".to_owned(), "project-token".to_owned()),
        ]);

        scan_within(dir.path(), &env, None, None, PrRefresh::Fetch, true, budget).await;
        std::fs::read_to_string(&seen).ok()
    }

    /// The incident shape: a worktree whose local branch name differs
    /// from the PR's head ref still resolves its PR, because the
    /// lookup key is the pushed ancestor sha, not the branch name.
    /// Real git, no network: `origin`'s tracking ref is simulated
    /// with `update-ref`, exactly what a `HEAD:<name>` push + fetch
    /// leaves behind.
    #[tokio::test(flavor = "current_thread")]
    async fn pushed_sha_walk_resolves_pushed_ancestor_when_worktree_is_ahead() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "worktree-inspector-pr-row");
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "pushed commit");
        let pushed = rev_parse_head(&dir);
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        // Simulate the push as `HEAD:audit/2026-09-03` + fetch: the
        // local branch stays `worktree-inspector-pr-row` while the
        // remote tracking ref names the pushed target.
        run(&["update-ref", "refs/remotes/origin/audit/2026-09-03", &pushed]);
        write_file(&dir, "more.rs", "fn y() {}\n");
        commit_all(&dir, "unpushed commit");
        write_file(&dir, "more2.rs", "fn z() {}\n");
        commit_all(&dir, "unpushed commit 2");

        assert_eq!(
            resolve_pushed_sha(dir.path(), 500, TEST_GIT_BUDGET).await.as_deref(),
            Some(pushed.as_str()),
            "the walk must stop at the pushed ancestor, not the unpushed tip"
        );
    }

    /// A worktree sitting exactly on its pushed tip resolves the tip
    /// itself (the names-match happy path).
    #[tokio::test(flavor = "current_thread")]
    async fn pushed_sha_walk_resolves_tip_when_pushed() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "pushed commit");
        let pushed = rev_parse_head(&dir);
        StdCommand::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["update-ref", "refs/remotes/origin/main", &pushed])
            .output()
            .expect("git ok");

        assert_eq!(
            resolve_pushed_sha(dir.path(), 500, TEST_GIT_BUDGET).await.as_deref(),
            Some(pushed.as_str())
        );
    }

    /// Nothing pushed - no remote refs share any of HEAD's history.
    /// No row, and the walk terminates rather than scanning the repo.
    #[tokio::test(flavor = "current_thread")]
    async fn pushed_sha_walk_returns_none_when_nothing_pushed() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "local commit");

        assert_eq!(resolve_pushed_sha(dir.path(), 500, TEST_GIT_BUDGET).await, None);
    }

    /// A pathological unpushed history terminates at the cap: shas
    /// beyond `cap` first-parent commits are never examined, so a
    /// pushed ancestor outside the cap does not resolve (truthful
    /// "no row" rather than an unbounded walk).
    #[tokio::test(flavor = "current_thread")]
    async fn pushed_sha_walk_caps_at_max_count() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "one\n");
        commit_all(&dir, "pushed commit");
        let pushed = rev_parse_head(&dir);
        StdCommand::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["update-ref", "refs/remotes/origin/main", &pushed])
            .output()
            .expect("git ok");
        for i in 0..4 {
            write_file(&dir, &format!("f{i}.rs"), "x\n");
            commit_all(&dir, &format!("unpushed {i}"));
        }

        // Cap 2 reaches only the newest 2 unpushed commits; the
        // pushed ancestor at depth 5 stays out of reach.
        assert_eq!(resolve_pushed_sha(dir.path(), 2, TEST_GIT_BUDGET).await, None);
        // Full cap reaches it.
        assert_eq!(
            resolve_pushed_sha(dir.path(), 500, TEST_GIT_BUDGET).await.as_deref(),
            Some(pushed.as_str())
        );
    }

    /// The open-filter + newest-updatedAt selection against the
    /// `commits/<sha>/pulls` REST shape: a closed PR with a NEWER
    /// updated_at must lose to an open PR, and two open PRs pick the
    /// most recently updated one (the actively-evolving stacked PR).
    #[test]
    fn pick_open_pr_filters_to_open_and_picks_newest_updated_at() {
        let raw = concat!(
            r#"[{"number":900,"state":"closed","html_url":"https://example/pull/900","updated_at":"2026-09-03T15:00:00Z"},"#,
            r#"{"number":858,"state":"open","html_url":"https://example/pull/858","updated_at":"2026-09-03T10:00:00Z"},"#,
            r#"{"number":859,"state":"open","html_url":"https://example/pull/859","updated_at":"2026-09-03T12:00:00Z"}]"#
        );
        let picked = pick_open_pr(raw).expect("parses");
        assert_eq!(
            picked,
            Some((859, "https://example/pull/859".to_owned(), false)),
            "open filter first, then most recently updated wins"
        );
    }

    /// Equal `updated_at` breaks deterministically on the higher
    /// number, so the pick never depends on response ordering.
    #[test]
    fn pick_open_pr_ties_break_on_higher_number() {
        let raw = concat!(
            r#"[{"number":401,"state":"open","html_url":"https://example/pull/401","updated_at":"2026-09-03T12:00:00Z"},"#,
            r#"{"number":402,"state":"open","html_url":"https://example/pull/402","updated_at":"2026-09-03T12:00:00Z"}]"#
        );
        assert_eq!(
            pick_open_pr(raw).expect("parses"),
            Some((402, "https://example/pull/402".to_owned(), false))
        );
    }

    /// Only merged/closed PRs contain the sha: no row. The
    /// open-only invariant means a merged PR row can never render.
    #[test]
    fn pick_open_pr_returns_none_when_only_closed_prs() {
        let raw = r#"[{"number":857,"state":"closed","html_url":"https://example/pull/857","updated_at":"2026-09-03T13:16:33Z"}]"#;
        assert_eq!(pick_open_pr(raw).expect("parses"), None);
    }

    /// Garbage JSON is an Err (the caller logs
    /// `gh_pr_lookup_parse_failed`), never a silent None.
    #[test]
    fn pick_open_pr_reports_parse_error() {
        assert!(pick_open_pr("not json").is_err());
    }

    fn snapshot_with_pr(number: u64) -> GitDiffSnapshot {
        snapshot_with_pr_at(number, Some(0))
    }

    fn snapshot_with_pr_at(number: u64, fetched_secs_ago: Option<u64>) -> GitDiffSnapshot {
        let pr = GitPrInfo { number, url: format!("https://example/pull/{number}"), draft: false };
        let pr_fetched_at = fetched_secs_ago
            .map(|secs| std::time::SystemTime::now() - std::time::Duration::from_secs(secs));
        GitDiffSnapshot {
            branch: GitBranch::Named("feat/x".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: Some("aaaa".into()),
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: Some(pr),
            closes: Vec::new(),
            pr_fetched_at,
        }
    }

    /// Full 40-hex sha of HEAD in the test repo.
    fn rev_parse_head(dir: &TempDir) -> String {
        rev_parse(dir.path(), "HEAD")
    }

    /// The full sha `rev` names in `dir`.
    fn rev_parse(dir: &Path, rev: &str) -> String {
        let out = StdCommand::new("git")
            .arg("-C")
            .arg(dir)
            .args(["rev-parse", rev])
            .output()
            .expect("git ok");
        assert!(out.status.success(), "rev-parse {rev} failed in {}", dir.display());
        String::from_utf8_lossy(&out.stdout).trim().to_owned()
    }

    /// Run `git <args>` at `dir`, asserting it succeeded.
    fn git_at(dir: &Path, args: &[&str]) {
        let out = StdCommand::new("git").arg("-C").arg(dir).args(args).output().expect("git ok");
        assert!(
            out.status.success(),
            "git {args:?} failed in {}: {}",
            dir.display(),
            String::from_utf8_lossy(&out.stderr),
        );
    }

    /// `scan` carries cached PR data through the cache-hit path. Set
    /// up a real git repo on a feature branch, build a `prev`
    /// snapshot for the same branch with a synthetic PR, and assert
    /// the returned snapshot mirrors prev's PR fields. Without the
    /// cache, `scan` would call `gh` against a non-github tempdir
    /// and the PR fields would come back empty.
    #[tokio::test(flavor = "current_thread")]
    async fn scan_carries_prev_pr_through_when_branch_unchanged() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "first\n");
        commit_all(&dir, "init");
        let run = |args: &[&str]| {
            StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git ok");
        };
        run(&["checkout", "-q", "-b", "feat/cache"]);
        write_file(&dir, "feat.rs", "fn x() {}\n");
        commit_all(&dir, "feat commit");

        let synthetic_pr =
            GitPrInfo { number: 99, url: "https://example/pull/99".into(), draft: false };
        let synthetic_closes =
            vec![GitIssueRef { number: 1, url: "https://example/issues/1".into() }];
        let prev = GitDiffSnapshot {
            branch: GitBranch::Named("feat/cache".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: None,
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: Some(synthetic_pr.clone()),
            closes: synthetic_closes.clone(),
            pr_fetched_at: Some(std::time::SystemTime::now()),
        };

        let snap = scan_in_test(dir.path(), &HashMap::new(), Some(&prev)).await;
        assert_eq!(snap.pr, Some(synthetic_pr), "cached PR must carry through scan");
        assert_eq!(snap.closes, synthetic_closes);
        assert_eq!(snap.pushed_sha, None, "nothing pushed in this repo");
    }

    /// On the default branch, `scan` skips the PR fetch entirely -
    /// `pr` / `closes` stay empty. Mirrors the "no PR opens against
    /// main" assumption (true for personal repos; fork → upstream
    /// edge case is out of scope for v1).
    #[tokio::test(flavor = "current_thread")]
    async fn scan_skips_pr_fetch_on_default_branch() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "README.md", "hello\n");
        commit_all(&dir, "init");

        let prev = GitDiffSnapshot {
            branch: GitBranch::Named("main".into()),
            default_branch: Some("main".into()),
            repo_gate: RepoGate::InRepo,
            pushed_sha: None,
            worktree: LayerState::Clean,
            branch_ahead: LayerState::Clean,
            pr: Some(GitPrInfo { number: 1, url: "url".into(), draft: false }),
            closes: Vec::new(),
            pr_fetched_at: Some(std::time::SystemTime::now()),
        };
        // Even with a cache-hit-shaped prev, the default-branch gate
        // wins and the PR field clears.
        let snap = scan_in_test(dir.path(), &HashMap::new(), Some(&prev)).await;
        assert_eq!(snap.pr, None);
        assert!(snap.closes.is_empty());
        assert_eq!(snap.pushed_sha, None, "the walk must not run on the default branch");
        assert_eq!(snap.pr_fetched_at, None);
    }

    #[test]
    fn classify_rev_parse_maps_output_to_repo_gate() {
        // Ok(stdout) trims to the branch; Empty is a clean non-repo
        // signal; Oversize is always a scanner fault.
        assert_eq!(
            classify_rev_parse(GitOutput::Ok("  main\n".to_owned()), RepoPresence::Present),
            Ok("main".to_owned())
        );
        assert_eq!(
            classify_rev_parse(GitOutput::Empty, RepoPresence::Absent),
            Err(RepoGate::NotARepo)
        );
        assert_eq!(
            classify_rev_parse(GitOutput::Oversize, RepoPresence::Present),
            Err(RepoGate::ScannerFailed)
        );
    }

    /// A non-zero exit is the ambiguous case, and the probe is what
    /// resolves it. `Unusable` must stay on the scanner gate: a git
    /// that would not run is a fault to report, not a licence to
    /// suppress the section as though there were no repo.
    #[test]
    fn a_non_zero_exit_is_resolved_by_the_repo_presence_probe() {
        assert_eq!(
            classify_rev_parse(GitOutput::Failed, RepoPresence::Absent),
            Err(RepoGate::NotARepo)
        );
        assert_eq!(
            classify_rev_parse(GitOutput::Failed, RepoPresence::Present),
            Err(RepoGate::ScannerFailed)
        );
        assert_eq!(
            classify_rev_parse(GitOutput::Failed, RepoPresence::Unusable),
            Err(RepoGate::ScannerFailed)
        );
    }

    /// The probe's own contract against real git: a plain directory has
    /// no repo, a fresh `git init` with no commits does. That second
    /// case is the one the branch rev-parse gets wrong on its own.
    #[tokio::test(flavor = "current_thread")]
    async fn repo_presence_separates_a_bare_directory_from_an_unborn_repo() {
        let plain = tempfile::tempdir().expect("tempdir");
        assert_eq!(repo_presence(plain.path()).await, RepoPresence::Absent);
        assert_eq!(repo_presence(&plain.path().join("gone")).await, RepoPresence::Absent);

        let repo = tempfile::tempdir().expect("tempdir");
        init_repo(&repo, "main");
        assert_eq!(repo_presence(repo.path()).await, RepoPresence::Present);
    }

    /// A checkout git refuses is not an absent repo. Deleting `.git/HEAD`
    /// leaves `.git` on disk while `rev-parse` exits 128 with the *same*
    /// `fatal:` line a plain non-repo gives, so the on-disk check is the
    /// only thing separating them - and reading it as absent would
    /// suppress the GIT section for a broken checkout instead of
    /// reporting it. The pruned-worktree case forge produces routinely
    /// has this shape.
    #[tokio::test(flavor = "current_thread")]
    async fn a_checkout_git_refuses_is_unusable_not_absent() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        std::fs::remove_file(dir.path().join(".git").join("HEAD")).expect("remove HEAD");

        assert_eq!(repo_presence(dir.path()).await, RepoPresence::Unusable);
        assert_eq!(
            current_branch(dir.path()).await,
            Err(RepoGate::ScannerFailed),
            "a checkout git refuses must surface the scanner banner, not read as a non-repo",
        );
        assert_eq!(
            scan_in_test(dir.path(), &HashMap::new(), None).await.repo_gate,
            RepoGate::ScannerFailed,
        );
    }

    /// The one shape where following symlinks reads a broken checkout as
    /// an absent one: git gives the same "not a git repository" line a
    /// plain non-repo gives, and `try_exists` says missing because it
    /// stats the vanished target rather than the link.
    #[cfg(unix)]
    #[tokio::test(flavor = "current_thread")]
    async fn a_dangling_dot_git_symlink_is_unusable_not_absent() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::os::unix::fs::symlink(dir.path().join("gone-target"), dir.path().join(".git"))
            .expect("symlink");

        assert_eq!(repo_presence(dir.path()).await, RepoPresence::Unusable);
        assert_eq!(
            scan_in_test(dir.path(), &HashMap::new(), None).await.repo_gate,
            RepoGate::ScannerFailed,
        );
    }

    /// The stale-base shape: a bare remote, a checkout whose `origin/main`
    /// sits one commit behind the remote, and a feature branch cut from the
    /// remote's tip with one commit of its own. On the stale ref the range
    /// folds the remote's commit in; on a refreshed one it is the branch's
    /// own change alone.
    fn stale_base_fixture() -> (TempDir, TempDir) {
        let upstream = tempfile::tempdir().expect("bare remote");
        git_at(upstream.path(), &["init", "-q", "--bare"]);
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "X base");
        let base = rev_parse_head(&dir);
        git_at(dir.path(), &["remote", "add", "origin", &upstream.path().to_string_lossy()]);
        git_at(dir.path(), &["push", "-q", "-u", "origin", "main"]);
        // Upstream moves on: one commit main gained that the checkout has not
        // fetched.
        write_file(&dir, "upstream.txt", "upstream\n");
        commit_all(&dir, "Y upstream");
        git_at(dir.path(), &["push", "-q", "origin", "main"]);
        // The feature branch is cut from upstream's tip, with one commit of
        // its own.
        git_at(dir.path(), &["checkout", "-q", "-b", "feature"]);
        write_file(&dir, "feat.txt", "feat\n");
        commit_all(&dir, "Z feat");
        // The stale ref: the checkout's origin/main is still X even though the
        // remote's main is Y.
        git_at(dir.path(), &["update-ref", "refs/remotes/origin/main", &base]);
        git_at(
            dir.path(),
            &["symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main"],
        );
        (upstream, dir)
    }

    /// **The display read's refresh moves the base before the diff is read.**
    ///
    /// Ved's Granite case: the checkout's `origin/main` was stale while the
    /// remote had moved on, so the merge-base with HEAD moved back and every
    /// commit main had gained since the stale ref folded into the branch's
    /// diff - a 3-commit branch read as 19 changed files. The read that draws
    /// the diff refreshes the base first; the control proves the stale ref
    /// really does inflate the range, so this cannot pass by a wrong fixture.
    #[tokio::test(flavor = "current_thread")]
    async fn the_display_refresh_moves_a_stale_base_before_the_diff_is_read() {
        let (upstream, dir) = stale_base_fixture();

        let stale = StdCommand::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["diff", "--name-only", "origin/main...HEAD"])
            .output()
            .expect("git ok");
        assert!(stale.status.success(), "the control range reads");
        let stale_files = String::from_utf8_lossy(&stale.stdout);
        assert!(
            stale_files.contains("upstream.txt") && stale_files.contains("feat.txt"),
            "control: the stale base folds upstream's commit into the branch diff: {stale_files:?}",
        );

        let base = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        assert_eq!(base.state, BaseState::Current, "the fetch answered a current base");
        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            rev_parse(upstream.path(), "main"),
            "and it moved the remote-tracking ref the layer reads",
        );
        assert_eq!(
            base.base_ref.as_deref(),
            Some(rev_parse(upstream.path(), "main").as_str()),
            "and the sha it pins is the ref's own after the fetch",
        );

        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        // The scan kicks a throttled background fetch; settle it before the
        // fixture directory goes.
        let _ = settle_background_fetches().await;
        let LayerState::Populated(ahead) = &snap.branch_ahead else {
            panic!("the feature branch has one commit ahead of the refreshed base");
        };
        assert_eq!(ahead.commit_count, 1, "the branch's own commit alone");
        assert_eq!(
            ahead.stats.files.iter().map(|file| file.path.as_str()).collect::<Vec<_>>(),
            vec!["feat.txt"],
            "upstream's already-merged commit is not the branch's change",
        );
    }

    /// **A base the fetch could not refresh is answered unchecked, pinned to
    /// the ref the checkout still holds.** No remote tip is read any more, so
    /// no lag is claimed; the stale ref stays where it was, and the layer read
    /// after this is still the stale one - the unchecked state is what keeps
    /// it from reading as the branch's own change.
    #[tokio::test(flavor = "current_thread")]
    async fn a_base_the_fetch_could_not_refresh_is_unchecked_and_pinned_to_the_stale_ref() {
        let (_upstream, dir) = stale_base_fixture();
        let stale = rev_parse(dir.path(), "refs/remotes/origin/main");
        let bin = tempfile::tempdir().expect("wrapper dir");
        fail_fetch_at(dir.path(), bin.path());

        let base = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;

        assert_eq!(
            base.state,
            BaseState::Unverified,
            "the fetch did not run, so the base is not claimed",
        );
        assert!(base.error.is_some(), "and git's own words ride the answer: {base:?}");
        assert_eq!(
            base.base_ref.as_deref(),
            Some(stale.as_str()),
            "pinned to the ref the checkout still holds",
        );
        // The scan is local, so the stale ref this case is about is exactly
        // what it reads; its kicked background fetch is settled before the
        // fixture goes.
        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        crate::env::git_command::clear_git_point(dir.path());
        let _ = settle_background_fetches().await;
        let LayerState::Populated(ahead) = &snap.branch_ahead else {
            panic!("the branch-ahead layer is read from the ref it holds");
        };
        assert_eq!(ahead.commit_count, 2, "the unfetched range still folds upstream's commit in");
        assert!(
            ahead.stats.files.iter().any(|file| file.path == "upstream.txt"),
            "so the file list is the stale one, which is what the label must cover",
        );
    }

    /// A base whose remote cannot be reached is neither refreshed nor
    /// measured, and says exactly that rather than drawing silence.
    #[tokio::test(flavor = "current_thread")]
    async fn a_base_with_no_reachable_remote_is_unverified() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");
        git_at(dir.path(), &["remote", "add", "origin", "/nonexistent/forge-base.git"]);
        git_at(dir.path(), &["update-ref", "refs/remotes/origin/main", "HEAD"]);
        git_at(
            dir.path(),
            &["symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main"],
        );

        assert_eq!(
            refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await.state,
            BaseState::Unverified,
            "an unreachable remote leaves the base unchecked, and the refresh says so",
        );
    }

    /// The control for the two below: a checkout whose `origin` probe really
    /// answers "no such remote" has nothing upstream to lag behind, and its
    /// local base is `local`.
    #[tokio::test(flavor = "current_thread")]
    async fn a_local_base_with_no_origin_remote_is_local() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");

        assert_eq!(
            refresh_base_within(dir.path(), "main", TEST_GIT_BUDGET).await.state,
            BaseState::Local,
            "a repository with no origin has nothing this base can lag behind",
        );
    }

    /// **A probe that never answered must not answer "no remote".** The base
    /// is local, and the checkout's `origin` question could not be asked at
    /// all: `local` would draw the stale base silently, so the answer is
    /// `unverified` and the line that says so names the checkout.
    #[tokio::test(flavor = "current_thread")]
    async fn a_local_base_beside_an_unanswerable_origin_probe_is_unverified() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");
        let bin = tempfile::tempdir().expect("wrapper dir");
        unanswerable_origin_at(dir.path(), bin.path());
        let capture = LogCapture::default();
        // The line is DEBUG: a git that will not answer is the environment's
        // weather, and the default filter has to be lowered to read it.
        let subscriber = tracing_subscriber::fmt()
            .with_writer(capture.clone())
            // No ANSI: the assertions read the fields, and the writer colours
            // the separators, so `behind=1` arrives as three styled spans.
            .with_ansi(false)
            .with_max_level(tracing::Level::DEBUG)
            .finish();

        let guard = tracing::subscriber::set_default(subscriber);
        let refresh = refresh_base_within(dir.path(), "main", TEST_GIT_BUDGET).await;
        drop(guard);
        crate::env::git_command::clear_git_point(dir.path());

        assert_eq!(
            refresh.state,
            BaseState::Unverified,
            "an unanswered origin probe leaves the base unchecked, not local",
        );
        let log = String::from_utf8_lossy(&capture.0.lock().clone()).into_owned();
        assert!(
            log.contains("git_base_origin_probe_unanswered"),
            "the decision is its own event: {log}",
        );
        assert!(
            log.contains(&dir.path().display().to_string()),
            "and it names the checkout a reader has to look at: {log}",
        );
        assert!(log.contains("git_refused"), "and why the probe could not answer: {log}");
    }

    /// A local base whose own branch tracks a remote can lag behind it just
    /// as an `origin/...` can, so it is not `local`: nothing was fetched for
    /// that upstream here.
    #[tokio::test(flavor = "current_thread")]
    async fn a_local_base_that_tracks_an_upstream_is_unverified() {
        let upstream = tempfile::tempdir().expect("bare remote");
        git_at(upstream.path(), &["init", "-q", "--bare"]);
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");
        git_at(dir.path(), &["remote", "add", "upstream", &upstream.path().to_string_lossy()]);
        git_at(dir.path(), &["push", "-q", "-u", "upstream", "main"]);

        assert_eq!(
            refresh_base_within(dir.path(), "main", TEST_GIT_BUDGET).await.state,
            BaseState::Unverified,
            "a base tracking a remote nobody fetched is not a local base",
        );
    }

    /// **A pinned scan reads the sha it was given, not the ref's current
    /// value.** The by-name scan answers the post-fetch base; the pinned one
    /// the stale sha, with that base's count and file list.
    #[tokio::test(flavor = "current_thread")]
    async fn a_pinned_scan_reads_the_sha_it_was_given() {
        let (upstream, dir) = stale_base_fixture();
        let stale = rev_parse(dir.path(), "refs/remotes/origin/main");
        // The fetch a review read would run: the ref moves to the tip.
        git_at(
            dir.path(),
            &["update-ref", "refs/remotes/origin/main", &rev_parse(upstream.path(), "main")],
        );

        let by_name = scan_serving_cached_pr(dir.path(), &HashMap::new(), None).await;
        let LayerState::Populated(named) = &by_name.branch_ahead else {
            panic!("the branch layer is read");
        };
        assert_eq!(named.commit_count, 1, "the by-name scan reads the moved ref");

        let pinned = scan_pinned(dir.path(), &HashMap::new(), None, &stale).await;
        let LayerState::Populated(ahead) = &pinned.branch_ahead else {
            panic!("the pinned scan reads the stale base");
        };
        assert_eq!(ahead.commit_count, 2, "the pinned scan reads the sha it was given");
        let paths: Vec<&str> = ahead.stats.files.iter().map(|file| file.path.as_str()).collect();
        assert!(
            paths.contains(&"upstream.txt") && paths.contains(&"feat.txt"),
            "and that base's own files, not the moved ref's: {paths:?}",
        );
    }

    /// **The content read follows the pinned base too.** The stats and the
    /// files must describe one commit; a content read at the ref's name would
    /// carry the moved base's files beside the pinned count.
    #[tokio::test(flavor = "current_thread")]
    async fn the_content_read_follows_the_pinned_base() {
        let (upstream, dir) = stale_base_fixture();
        let stale = rev_parse(dir.path(), "refs/remotes/origin/main");
        git_at(
            dir.path(),
            &["update-ref", "refs/remotes/origin/main", &rev_parse(upstream.path(), "main")],
        );
        let snapshot = scan_serving_cached_pr(dir.path(), &HashMap::new(), None).await;

        let content = super::content::scan_within(
            dir.path(),
            &snapshot,
            Some(BaseState::Current),
            Some(&stale),
            None,
            TEST_GIT_BUDGET,
        )
        .await;

        let LayerState::Populated(files) = &content.branch_ahead else {
            panic!("the branch layer is read");
        };
        let paths: Vec<&str> = files.iter().map(|file| file.path.as_str()).collect();
        assert!(
            paths.contains(&"upstream.txt") && paths.contains(&"feat.txt"),
            "the content is the pinned base's range, not the moved ref's: {paths:?}",
        );
    }

    /// **A refresh that just ran leaves the kick nothing to do.** Stamping the
    /// window after a successful fetch is what keeps a scan seconds later from
    /// fetching the same answer again.
    #[tokio::test(flavor = "current_thread")]
    async fn a_refresh_stamps_the_kicks_window() {
        let (_upstream, dir) = stale_base_fixture();

        let refresh = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        assert_eq!(refresh.state, BaseState::Current, "the refresh fetched");
        assert!(
            !claim_refresh_slot(dir.path(), Instant::now(), WATCH_REFRESH_INTERVAL),
            "and a kick inside the window it stamped does not fetch again",
        );
    }

    #[test]
    fn the_refresh_window_gates_the_next_fetch() {
        let now = Instant::now();
        let window = Duration::from_secs(240);
        assert!(refresh_due(None, now, window), "no prior fetch is due");
        assert!(!refresh_due(Some(now), now, window), "a just-now fetch is throttled");
        let stale = now
            .checked_sub(window + Duration::from_secs(1))
            .expect("the monotonic clock is past a window");
        assert!(refresh_due(Some(stale), now, window), "a stale prior fetch is due again");
    }

    #[test]
    fn a_claimed_slot_throttles_and_prunes() {
        let key = tempfile::tempdir().expect("a key of its own");
        let expired = tempfile::tempdir().expect("another key");
        let now = Instant::now();
        let window = Duration::from_secs(240);

        assert!(claim_refresh_slot(key.path(), now, window), "the first claim is due");
        assert!(!claim_refresh_slot(key.path(), now, window), "an immediate re-claim is throttled");
        assert!(claim_refresh_slot(expired.path(), now, window), "another key has its own window");
        assert!(
            claim_refresh_slot(key.path(), now + window, window),
            "a claim once the window elapses is due again",
        );
        let map = LAST_WATCH_REFRESH.get().expect("the map exists").lock();
        assert!(
            !map.contains_key(expired.path()),
            "and the claim pruned the other key's expired slot",
        );
    }

    /// The tri-state's `Present` arm: a local base beside an `origin` remote
    /// can be arbitrarily behind one nobody fetched, so it is never `local`.
    #[tokio::test(flavor = "current_thread")]
    async fn a_local_base_beside_an_origin_remote_is_unverified() {
        let upstream = tempfile::tempdir().expect("bare remote");
        git_at(upstream.path(), &["init", "-q", "--bare"]);
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");
        git_at(dir.path(), &["remote", "add", "origin", &upstream.path().to_string_lossy()]);

        assert_eq!(
            refresh_base_within(dir.path(), "main", TEST_GIT_BUDGET).await.state,
            BaseState::Unverified,
            "an origin exists, so a local base is a base nobody fetched",
        );
    }

    /// **One run per repository, not per checkout.** Two linked worktrees share
    /// `refs/remotes/*`, so a worktree's caller joins its checkout's refresh
    /// rather than fetching the same ref again.
    #[tokio::test(flavor = "current_thread")]
    async fn linked_worktrees_join_one_repository_run() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");
        let beside = tempfile::tempdir().expect("a second directory");
        let linked = beside.path().join("wt");
        git_at(dir.path(), &["worktree", "add", "-q", "-b", "linked", &linked.to_string_lossy()]);

        let main_key = repo_key(dir.path(), TEST_GIT_BUDGET).await;
        let linked_key = repo_key(&linked, TEST_GIT_BUDGET).await;
        assert_eq!(main_key, linked_key, "a worktree and its checkout share one ref store");
        let (first, leader) = join_base_run((main_key, "origin/main".to_owned()));
        assert!(leader, "the first caller starts the run");
        let (joined, leader) = join_base_run((linked_key, "origin/main".to_owned()));
        assert!(!leader, "and the worktree's caller joins it rather than starting a second");
        assert!(std::sync::Arc::ptr_eq(&first, &joined), "the one run answers both");
    }

    /// A ref lock held in the user's own checkout keeps the base stale; with no
    /// remote tip read any more, the answer is unchecked rather than a count.
    /// The operator's side is the lock classification's own: forge's mark in
    /// their tree is logged apart from the machine's network weather.
    #[tokio::test(flavor = "current_thread")]
    async fn a_base_the_users_git_has_locked_is_unchecked() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        lock_fetch_at(dir.path(), bin.path());

        assert_eq!(
            refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await.state,
            BaseState::Unverified,
            "a lock the fetch could not take leaves the base unchecked",
        );
        crate::env::git_command::clear_git_point(dir.path());
    }

    /// **The lock refusal is logged as forge's mark, not as the machine's
    /// weather**, and the line names the repo a reader has to look at.
    #[tokio::test(flavor = "current_thread")]
    async fn a_lock_refusal_is_named_with_its_checkout_and_gits_words() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        lock_fetch_at(dir.path(), bin.path());
        let capture = LogCapture::default();
        let subscriber = tracing_subscriber::fmt()
            .with_writer(capture.clone())
            .with_ansi(false)
            // DEBUG: nothing can say whose git holds the lock, so the level
            // does not claim it is forge's (rule 20).
            .with_max_level(tracing::Level::DEBUG)
            .finish();
        let guard = tracing::subscriber::set_default(subscriber);

        let _ = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        drop(guard);
        crate::env::git_command::clear_git_point(dir.path());

        let log = String::from_utf8_lossy(&capture.0.lock().clone()).into_owned();
        assert!(log.contains("git_base_fetch_lock_refused"), "the lock event is named: {log}");
        assert!(
            log.contains(&dir.path().display().to_string()),
            "and the repo it happened in: {log}",
        );
        assert!(
            log.contains("cannot lock ref"),
            "and git's own words, which carry the lock file: {log}",
        );
    }

    /// **A successful fetch is the answer, whatever is on disk beforehand.**
    /// The explicit refspec writes the destination, so a checkout whose base
    /// ref was rewritten past the remote ends the refresh current at the
    /// remote's tip, not unverified.
    #[tokio::test(flavor = "current_thread")]
    async fn a_base_rewritten_past_is_current_after_its_fetch() {
        let (upstream, dir) = stale_base_fixture();
        let newer = rev_parse(upstream.path(), "main");
        let older = rev_parse(upstream.path(), "main~1");
        // The base holds the commit the remote no longer has.
        git_at(dir.path(), &["update-ref", "refs/remotes/origin/main", &newer]);
        git_at(upstream.path(), &["update-ref", "refs/heads/main", &older]);

        let refresh = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        crate::env::git_command::clear_git_point(dir.path());

        assert_eq!(
            refresh.state,
            BaseState::Current,
            "the fetch wrote the remote's tip over the rewritten base",
        );
        assert_eq!(
            refresh.base_ref.as_deref(),
            Some(older.as_str()),
            "and the sha it pins is the ref's own after the fetch",
        );
    }

    /// The lock prose git uses, against the prose a dead network produces:
    /// the first is forge's to report, the second is the environment's.
    #[test]
    fn a_ref_lock_is_told_apart_from_a_dead_network() {
        assert!(is_lock_failure(
            "fatal: cannot lock ref 'refs/remotes/origin/main': is at 1a2b3c but expected 4d5e6f",
        ));
        assert!(is_lock_failure(
            "error: Unable to create '/repo/.git/refs/remotes/origin/main.lock': File exists.",
        ));
        assert!(!is_lock_failure(
            "fatal: unable to access 'https://github.com/x/y.git/': Could not resolve host: github.com",
        ));
    }

    /// **A local base answers `local`, and the probes are what say so.** A
    /// repo with no `origin` and a branch tracking nothing has no remote to
    /// lag behind; the drawing read keeps the probes the refresh path always
    /// ran, so the page draws no caveat (#2818).
    #[tokio::test(flavor = "current_thread")]
    async fn a_local_base_answers_local_without_a_fetch() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "X base");
        git_at(dir.path(), &["checkout", "-q", "-b", "feature"]);
        write_file(&dir, "feat.txt", "feat\n");
        commit_all(&dir, "Z feat");

        let answer = refresh_base_for_display(dir.path(), "main").await;

        assert_eq!(
            answer.state,
            BaseState::Local,
            "a base with nothing to lag behind is local: {answer:?}",
        );
        assert!(answer.base_ref.is_some(), "resolved to the local tip");
        assert!(answer.error.is_none() && !answer.cut && !answer.unanswered);
        assert_eq!(settle_background_fetches().await, 0, "a local base fetches nothing");
    }

    /// **A drawing read with no measurement answers at once, and kicks the
    /// refresh that serves the next one.** The fetch is held open, so a read
    /// that awaited it could not answer at all: the answer arriving while the
    /// wrapper provably blocks, with the fetch started anyway, is the whole
    /// property (#2818).
    #[tokio::test(flavor = "current_thread")]
    async fn a_reading_with_no_measurement_answers_at_once_and_kicks_the_refresh() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        let reading = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_for_display(&dir, "origin/main").await }
        });
        let answer = timeout(TEST_GIT_BUDGET, reading)
            .await
            .expect("the reading did not wait on the fetch")
            .expect("the read task finished");

        wait_for_within(|| held.starts() == 1, "the reading's refresh to start", TEST_GIT_BUDGET)
            .await;
        assert!(held.open(), "the read's refresh runs off the read: the fetch is still in flight");
        assert_eq!(
            answer.state,
            BaseState::Unverified,
            "with no check in hand, the base is unchecked"
        );
        assert!(answer.base_ref.is_some(), "and pinned to a local tip it did read");
        assert!(answer.error.is_none() && !answer.cut && !answer.unanswered);

        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());
    }

    /// **A drawing read in the 30-240 s gap answers the last base at once, and
    /// the refresh it kicks is what the next read draws.** The held fetch is
    /// the proof this read did not await it: the answer arrives while the
    /// wrapper blocks, carrying the check already in hand. The run's
    /// measurement then answers the next read, with no second fetch (#2818: a
    /// review open in the watch's 30-240 s gap paid one fetch per read).
    #[tokio::test(flavor = "current_thread")]
    async fn a_reading_answers_a_stored_check_past_thirty_seconds_and_the_refresh_lands() {
        let (upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        let stale_tip = rev_parse(dir.path(), "refs/remotes/origin/main");
        let measured_at = Instant::now()
            .checked_sub(Duration::from_secs(45))
            .expect("the test clock is past boot");
        remember_measurement(
            dir.path(),
            "origin/main",
            &BaseRefresh {
                state: BaseState::Current,
                base_ref: Some(stale_tip.clone()),
                error: None,
                cut: false,
                unanswered: false,
                at: measured_at,
            },
        );

        let reading = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_for_display(&dir, "origin/main").await }
        });
        let answer = timeout(TEST_GIT_BUDGET, reading)
            .await
            .expect("the reading did not wait on the fetch")
            .expect("the read task finished");

        assert_eq!(answer.state, BaseState::Current, "the last completed check answers the read");
        assert_eq!(
            answer.base_ref.as_deref(),
            Some(stale_tip.as_str()),
            "pinned to the base in hand"
        );
        assert_eq!(answer.at, measured_at, "at that check's own instant");
        wait_for_within(
            || held.starts() == 1,
            "the reading's own refresh to start",
            TEST_GIT_BUDGET,
        )
        .await;
        assert!(held.open(), "the refresh runs off the read: the fetch is still in flight");

        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());

        let next = refresh_base_for_display(dir.path(), "origin/main").await;
        assert_eq!(next.state, BaseState::Current, "the refreshed base answers the next read");
        assert_eq!(
            next.base_ref.as_deref(),
            Some(rev_parse(upstream.path(), "main").as_str()),
            "pinned to the tip the fetch moved the ref to",
        );
        assert_eq!(held.starts(), 1, "and the next read paid no second fetch");
    }

    /// A worker branch is cut from origin/main while local `main` lags
    /// origin/main (the usual state when the user only pulls merged
    /// PRs). The branch-ahead diff must compare against the remote-
    /// tracking origin/main - showing only the branch's OWN commit -
    /// not against a stale local main that would fold in already-merged
    /// work. Exercised through a real linked worktree.
    #[tokio::test(flavor = "current_thread")]
    async fn scan_feature_branch_compares_against_origin_main_not_stale_local_main() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        let run = |args: &[&str]| {
            let out =
                StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git");
            assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
        };
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "X base");
        // Y = a commit that lives on origin/main (already merged).
        run(&["checkout", "-q", "-b", "upstream"]);
        write_file(&dir, "merged.txt", "merged\n");
        commit_all(&dir, "Y merged");
        run(&["update-ref", "refs/remotes/origin/main", "upstream"]);
        run(&["symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main"]);
        // Local main rewinds to X - it now lags origin/main by Y.
        run(&["checkout", "-q", "main"]);
        run(&["branch", "-D", "upstream"]);
        // Worker worktree cut from origin/main (Y), plus its own commit Z.
        std::fs::create_dir_all(dir.path().join(".claude/worktrees")).expect("mkdir");
        let wt = dir.path().join(".claude/worktrees/w");
        run(&["worktree", "add", "-q", wt.to_str().expect("utf8"), "-b", "feat/w", "origin/main"]);
        let wt_run = |args: &[&str]| {
            let out = StdCommand::new("git").arg("-C").arg(&wt).args(args).output().expect("git");
            assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
        };
        std::fs::write(wt.join("z.txt"), "z\n").expect("write z");
        wt_run(&["add", "-A"]);
        wt_run(&["commit", "-q", "-m", "Z own"]);

        let snap = scan_in_test(&wt, &HashMap::new(), None).await;
        assert_eq!(
            snap.default_branch.as_deref(),
            Some("origin/main"),
            "compares against the remote-tracking ref, not stripped local main",
        );
        let LayerState::Populated(ahead) = &snap.branch_ahead else {
            panic!("branch-ahead populated for the worker branch");
        };
        assert_eq!(
            ahead.commit_count, 1,
            "only the branch's own commit vs origin/main, not the already-merged one",
        );
        assert_eq!(ahead.stats.total_files, 1);
        assert_eq!(
            ahead.stats.files[0].path, "z.txt",
            "merged.txt (already on origin/main) is excluded from the branch-ahead diff",
        );
    }

    /// On `main` with the default resolved to the remote-tracking ref
    /// `origin/main`, the branch-ahead layer stays empty even when local
    /// main is AHEAD of origin/main (unpushed commits): the on-default
    /// check strips the remote prefix so a checked-out `main` reads as
    /// the default branch rather than a feature branch.
    #[tokio::test(flavor = "current_thread")]
    async fn scan_on_local_main_ahead_of_origin_skips_branch_ahead() {
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        let run = |args: &[&str]| {
            let out =
                StdCommand::new("git").arg("-C").arg(dir.path()).args(args).output().expect("git");
            assert!(out.status.success(), "git {args:?}: {}", String::from_utf8_lossy(&out.stderr));
        };
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "base");
        run(&["update-ref", "refs/remotes/origin/main", "HEAD"]);
        run(&["symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main"]);
        // Local main advances past origin/main (unpushed commit).
        write_file(&dir, "local.txt", "local\n");
        commit_all(&dir, "unpushed");

        let snap = scan_in_test(dir.path(), &HashMap::new(), None).await;
        assert_eq!(snap.default_branch.as_deref(), Some("origin/main"));
        assert!(
            matches!(snap.branch_ahead, LayerState::Clean),
            "on `main` the branch-ahead layer stays empty even when ahead of origin/main",
        );
    }

    /// **A stored read leaves the base alone.** The `ServeCached` scan a held
    /// seat's record read takes neither fetches nor kicks: it reads the ref as
    /// it stands, and the read that draws the diff is what refreshes it. (The
    /// fetching scan is the `Fetch` one above; a sessionless seat's read takes
    /// that, because it has no loop to refresh for it.)
    #[tokio::test(flavor = "current_thread")]
    async fn a_stored_read_leaves_the_base_alone() {
        let (upstream, dir) = stale_base_fixture();
        let before = rev_parse(dir.path(), "refs/remotes/origin/main");

        let snap = super::scan_serving_cached_pr(dir.path(), &HashMap::new(), None).await;

        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            before,
            "a record read leaves the ref alone",
        );
        assert_ne!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            rev_parse(upstream.path(), "main"),
            "so the stale ref is still stale after a record read",
        );
        let LayerState::Populated(ahead) = &snap.branch_ahead else {
            panic!("the layer is read from the ref the scan holds");
        };
        assert_eq!(ahead.commit_count, 2, "and the stale range is what the read answers");
    }

    /// **An ordinary scan kicks a throttled fetch for the NEXT read**, and
    /// never waits on it. This is the path a sessionless seat has: no loop
    /// exists to refresh for it, so without the kick its record re-reads the
    /// tree for ever against a ref nothing ever moves.
    #[tokio::test(flavor = "current_thread")]
    async fn an_ordinary_scan_kicks_a_throttled_fetch_for_the_next_read() {
        let (upstream, dir) = stale_base_fixture();

        let _ = scan_in_test(dir.path(), &HashMap::new(), None).await;
        let kicked =
            LAST_WATCH_REFRESH.get().is_some_and(|map| map.lock().contains_key(dir.path()));
        assert!(kicked, "an origin-tracking scan claims (kicks) a background fetch");

        assert!(
            settle_background_fetches().await >= 1,
            "and there is one to settle, so the kick was real",
        );
        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            rev_parse(upstream.path(), "main"),
            "which leaves the ref fresh for the next read",
        );
    }

    /// **A read that fetches the base itself takes no background kick.** A
    /// sessionless review read did both, and its own display fetch then
    /// queued behind forge's on the shared repository gate - so the page said
    /// "could not be refreshed" about a fetch forge itself was holding.
    #[tokio::test(flavor = "current_thread")]
    async fn a_scan_without_base_fetch_kicks_nothing() {
        let (upstream, dir) = stale_base_fixture();
        let before = rev_parse(dir.path(), "refs/remotes/origin/main");

        let _ = scan_without_base_fetch(dir.path(), &HashMap::new(), None).await;

        assert_eq!(
            settle_background_fetches().await,
            0,
            "the quiet scan kicks no background fetch",
        );
        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            before,
            "and the ref is where the caller's own fetch will find it",
        );
        assert_ne!(before, rev_parse(upstream.path(), "main"), "precondition: the base is stale");
    }

    /// **The scan kicks its refresh and does not wait on it.** The fetch is
    /// held open by the fixture, so a scan that awaited it could not answer:
    /// the bound below is the failure, not a performance claim.
    #[tokio::test(flavor = "current_thread")]
    async fn a_scan_kicks_its_refresh_and_the_measurement_lands_for_the_next_one() {
        let (upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        // The bound is a failure instead of a hang, not a performance claim:
        // the fetch is held, so a scan that awaited it could not answer at all.
        // It sits above one child's own budget, because a machine that holds a
        // child to its bound must not read as a scan that waited.
        let scanned =
            timeout(Duration::from_secs(90), scan_in_test(dir.path(), &HashMap::new(), None))
                .await
                .expect("the scan answered while its fetch was still held");
        let _ = scanned;
        wait_for(|| held.starts() == 1, "the kicked fetch to start").await;
        assert!(held.open(), "the fetch the scan kicked is provably still in flight");

        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());
        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            rev_parse(upstream.path(), "main"),
            "and the fetch it kicked moved the base to the remote's tip",
        );
        assert_eq!(
            last_base_measurement(dir.path(), "origin/main").map(|measured| measured.state),
            Some(BaseState::Current),
            "and the state it measured is what the row carries next",
        );
    }

    /// **A failed fetch says what git said, on the reading after it.** The read
    /// that kicks it answers unchecked without words it has not seen; the
    /// measurement the failed run leaves carries git's first line to the next
    /// read, which is what the page draws beside the base line.
    ///
    /// **The fetch is held, not failed on arrival**: a failure that lands while
    /// the first read is still finishing would be re-read from the store and
    /// answered with words that read did not wait for.
    #[tokio::test(flavor = "current_thread")]
    async fn a_failed_fetch_carries_gits_own_words() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let release = bin.path().join("fetch-release");
        let body = format!(
            "while [ ! -e '{release}' ]; do /bin/sleep 0.02; done; \
             echo 'fatal: forced fetch failure' >&2; exit 128",
            release = release.display(),
        );
        write_wrapper(dir.path(), bin.path(), "held-failing-git", &[], &body);

        let unchecked = refresh_base_for_display(dir.path(), "origin/main").await;
        assert_eq!(unchecked.state, BaseState::Unverified);
        assert!(
            unchecked.error.is_none(),
            "the read that kicked the fetch names no failure it has not seen: {unchecked:?}",
        );
        std::fs::write(&release, b"go").expect("release the fetch");
        let _ = settle_background_fetches().await;

        let refresh = refresh_base_for_display(dir.path(), "origin/main").await;
        crate::env::git_command::clear_git_point(dir.path());
        assert_eq!(
            refresh.error.as_deref(),
            Some("fatal: forced fetch failure"),
            "the record carries git's first line",
        );
        assert_eq!(
            refresh.state,
            BaseState::Unverified,
            "the fetch did not run and no tip is read, so the base is not claimed",
        );
        assert!(!refresh.cut, "a fetch that failed for its own reason was not cut by a bound");
    }

    /// **A run's answer is kept for a caller that arrives after it**, even
    /// with no receiver subscribed when it was published: `send` would throw
    /// it away and leave that caller waiting for the process's life.
    #[tokio::test(flavor = "current_thread")]
    async fn a_published_answer_waits_for_a_caller_that_arrives_after_it() {
        let key = (PathBuf::from("/an-answer-kept"), "origin/main".to_owned());
        let (run, leader) = join_base_run(key.clone());
        assert!(leader, "the test leads");
        run.publish(BaseRefresh {
            state: BaseState::Current,
            base_ref: Some("abc123".to_owned()),
            error: None,
            cut: false,
            unanswered: false,
            at: Instant::now(),
        });
        // The leader's own `Arc` is still here, exactly as a caller mid-flight
        // would hold it, so the registry entry outlives the publish.
        let (joined, leader) = join_base_run(key);
        assert!(!leader, "the late caller joins the run");
        let answer = timeout(Duration::from_secs(2), joined.wait())
            .await
            .expect("the published answer is not thrown away");
        assert_eq!(answer.base_ref.as_deref(), Some("abc123"));
    }

    /// **A leader that ends without answering wakes its waiters.** A dropped
    /// or panicked leader must not leave a joiner waiting for ever. The joiners
    /// are kicks: a drawing read answers at once and never waits on a run.
    #[tokio::test(flavor = "current_thread")]
    async fn a_cancelled_leader_answers_its_waiters() {
        let (upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        let leader = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_within(&dir, "origin/main", TEST_GIT_BUDGET).await }
        });
        wait_for(|| held.starts() == 1, "the leader's fetch to start").await;
        let follower = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_within(&dir, "origin/main", TEST_GIT_BUDGET).await }
        });
        wait_for_within(
            || held.probes() >= 2,
            "the follower to reach the registry",
            TEST_GIT_BUDGET,
        )
        .await;
        // Its key probe is followed by the join, a few microseconds later; this
        // settle only puts the abort after it, and the verdict never depends on
        // the length of the wait.
        tokio::time::sleep(Duration::from_millis(200)).await;
        leader.abort();

        let answer = timeout(Duration::from_secs(5), follower)
            .await
            .expect("the follower was answered")
            .expect("the follower's task finished");
        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());
        let _ = upstream;
        assert!(
            answer.error.as_deref().is_some_and(|words| words.contains("without an answer")),
            "a leader that ended unanswered is named: {answer:?}",
        );
        assert!(
            last_base_measurement(dir.path(), "origin/main").is_none(),
            "and a placeholder state is not stored as the checkout's measurement",
        );
    }

    /// **A reading does not wait on the fetch a scan's kick started.** The
    /// kick leads a run with the scan's own bound; a drawing read answers the
    /// base it has as unchecked, pinned to a local tip, and the kicked run's
    /// answer serves the next read (#2685).
    ///
    /// The bound is a failure rather than a performance claim: the held fetch
    /// cannot finish, so a read that joined it could not answer at all - and
    /// the answer arriving while the fetch is still open is the proof it did
    /// not. It sits far above what a loaded machine adds to the one local
    /// `git` child the answer reads.
    #[tokio::test(flavor = "current_thread")]
    async fn a_reading_answers_unchecked_rather_than_joining_a_kicked_fetch() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        kick_background_fetch(dir.path(), "origin/main", TEST_GIT_BUDGET);
        wait_for_within(|| held.starts() == 1, "the kicked fetch to start", TEST_GIT_BUDGET).await;

        let reading = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_for_display(&dir, "origin/main").await }
        });
        let answer = timeout(TEST_GIT_BUDGET, reading)
            .await
            .expect("the reading did not wait on the kicked fetch")
            .expect("the read task finished");

        assert_eq!(held.starts(), 1, "the reading started no fetch of its own");
        assert!(held.open(), "the kicked fetch is provably still in flight");
        assert_eq!(
            answer.state,
            BaseState::Unverified,
            "the reading says the base was not checked"
        );
        assert!(answer.base_ref.is_some(), "and pins to a local tip it did read");
        assert!(answer.error.is_none() && !answer.cut && !answer.unanswered);

        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());
    }

    /// **A run that has just finished answers the next read without a fetch.**
    /// The page is told the base moved and reads again; the answer is the
    /// measurement the run took, so the completion costs no second wait on the
    /// same remote.
    #[tokio::test(flavor = "current_thread")]
    async fn a_reading_after_a_finished_refresh_answers_without_fetching_again() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        kick_background_fetch(dir.path(), "origin/main", TEST_GIT_BUDGET);
        wait_for_within(|| held.starts() == 1, "the kicked fetch to start", TEST_GIT_BUDGET).await;
        held.release();
        let _ = settle_background_fetches().await;
        wait_for(
            || last_base_measurement(dir.path(), "origin/main").is_some(),
            "the measurement to land",
        )
        .await;
        let measured = last_base_measurement(dir.path(), "origin/main");

        let answer = timeout(TEST_GIT_BUDGET, refresh_base_for_display(dir.path(), "origin/main"))
            .await
            .expect("the reading answered from the measurement it holds");

        assert_eq!(held.starts(), 1, "the finished run answered the read; no second fetch");
        assert_eq!(
            Some(answer.state),
            measured.map(|measured| measured.state),
            "the state is the measurement the run took"
        );
        assert_eq!(answer.state, BaseState::Current, "the fixture's fetch makes the base current");
        assert!(answer.base_ref.is_some(), "pinned to the tip the fetch left");

        assert!(answer.error.is_none() && !answer.cut && !answer.unanswered);
        crate::env::git_command::clear_git_point(dir.path());
    }

    /// **A sessionless read waits on the kick that claimed its slot.** An
    /// ordinary seat read's scan kicks a refresh and claims the cwd slot; a
    /// review read inside that window must wait on the in-flight run rather
    /// than answer unchecked - no watch exists to heal a miss (#2818).
    #[tokio::test(flavor = "current_thread")]
    async fn a_sessionless_read_waits_on_the_kick_that_claimed_its_slot() {
        let (upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        kick_background_fetch(dir.path(), "origin/main", TEST_GIT_BUDGET);
        wait_for_within(|| held.starts() == 1, "the kick's fetch to start", TEST_GIT_BUDGET).await;
        let probes_before = held.probes();

        let reader = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_now(&dir, "origin/main").await }
        });
        // It reaches the registry, and the held fetch is what it waits on.
        wait_for_within(
            || held.probes() > probes_before,
            "the read to reach the registry",
            TEST_GIT_BUDGET,
        )
        .await;
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(!reader.is_finished(), "the read waits on the run its slot names");

        held.release();
        let answer = timeout(TEST_GIT_BUDGET, reader)
            .await
            .expect("the read answered")
            .expect("the read task finished");
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());

        assert_eq!(answer.state, BaseState::Current, "the joined run's check answers");
        assert_eq!(
            answer.base_ref.as_deref(),
            Some(rev_parse(upstream.path(), "main").as_str()),
            "with the base at the remote's tip",
        );
        assert_eq!(held.starts(), 1, "the read joined the kick rather than fetching again");
    }

    /// **A run that answered no check gives its slot back.** A cut or aborted
    /// run measured nothing; leaving its claim stamped would answer silence
    /// for the whole window on the one seat whose read is what heals it.
    #[test]
    fn a_run_without_a_check_gives_the_slot_back() {
        let dir = tempfile::tempdir().expect("tempdir");
        let now = Instant::now();
        assert!(
            claim_refresh_slot(dir.path(), now, WATCH_REFRESH_INTERVAL),
            "the fixture claims the slot",
        );
        let answered = BaseRefresh {
            state: BaseState::Current,
            base_ref: Some("abc123".to_owned()),
            error: None,
            cut: false,
            unanswered: false,
            at: Instant::now(),
        };
        give_back_slot(dir.path(), &answered);
        assert!(
            !claim_refresh_slot(dir.path(), Instant::now(), WATCH_REFRESH_INTERVAL),
            "a run that answered keeps its slot",
        );

        let cut = BaseRefresh {
            state: BaseState::Unverified,
            base_ref: None,
            error: Some("the fetch did not finish inside the 30s it was given".to_owned()),
            cut: true,
            unanswered: false,
            at: Instant::now(),
        };
        give_back_slot(dir.path(), &cut);
        assert!(
            claim_refresh_slot(dir.path(), Instant::now(), WATCH_REFRESH_INTERVAL),
            "a run without a check gives the slot back",
        );
    }

    /// **A fetch its budget cut is not a measurement, and the reading after it
    /// answers unchecked while its own kick runs.** The cut left nothing to
    /// reuse, so the reading kicks a refresh and answers at once; that run's
    /// measurement is what the next reading draws.
    #[tokio::test(flavor = "current_thread")]
    async fn a_cut_run_is_not_a_measurement_a_later_reading_reuses() {
        let (upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        let cut = refresh_base_within(dir.path(), "origin/main", Duration::from_millis(300)).await;
        assert!(cut.cut, "the held fetch outlived the bound");
        assert!(
            last_base_measurement(dir.path(), "origin/main").is_none(),
            "a cut answer is not stored as the checkout's measurement",
        );

        let reader = tokio::spawn({
            let dir = dir.path().to_path_buf();
            async move { refresh_base_for_display(&dir, "origin/main").await }
        });
        let answer = timeout(TEST_GIT_BUDGET, reader)
            .await
            .expect("the reading did not wait on its own fetch")
            .expect("the read task finished");
        assert_eq!(
            answer.state,
            BaseState::Unverified,
            "the cut answer is not a check the reading draws",
        );
        assert!(answer.error.is_none(), "and the cut's own words are not carried");
        wait_for_within(
            || held.starts() >= 2,
            "the reading kicks a refresh of its own",
            TEST_GIT_BUDGET,
        )
        .await;

        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());

        let next = refresh_base_for_display(dir.path(), "origin/main").await;
        assert_eq!(next.state, BaseState::Current, "the reading's own run measured the base");
        assert_eq!(
            next.base_ref.as_deref(),
            Some(rev_parse(upstream.path(), "main").as_str()),
            "with the base at the remote's tip",
        );
        assert_eq!(held.starts(), 2, "the cut run and the reading's own");
    }

    /// **The explicit refspec writes the base ref, not the repository's own.**
    /// A repository whose `remote.origin.fetch` points somewhere else entirely
    /// still gets `refs/remotes/origin/<b>` moved by the refspec `fetch_base`
    /// passes, which is the reason the post-fetch `ls-remote` could go.
    #[tokio::test(flavor = "current_thread")]
    async fn the_explicit_refspec_writes_the_base_ref_whatever_the_remotes_own_says() {
        let upstream = tempfile::tempdir().expect("a bare remote");
        git_at(upstream.path(), &["init", "-q", "--bare"]);
        let dir = tempfile::tempdir().expect("tempdir");
        init_repo(&dir, "main");
        write_file(&dir, "base.txt", "base\n");
        commit_all(&dir, "X base");
        git_at(dir.path(), &["remote", "add", "origin", &upstream.path().to_string_lossy()]);
        git_at(dir.path(), &["push", "-q", "-u", "origin", "main"]);
        let stale = rev_parse_head(&dir);
        // The repository's own fetch refspec writes a different namespace, so
        // only the explicit refspec can move the ref this diff reads.
        git_at(
            dir.path(),
            &["config", "remote.origin.fetch", "+refs/heads/*:refs/remotes/upstream/*"],
        );
        // Upstream moves on, and the base ref is still where it was.
        write_file(&dir, "upstream.txt", "upstream\n");
        commit_all(&dir, "Y upstream");
        git_at(dir.path(), &["push", "-q", "origin", "main"]);
        git_at(dir.path(), &["update-ref", "refs/remotes/origin/main", &stale]);

        let refresh = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;

        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            rev_parse(upstream.path(), "main"),
            "the explicit refspec moved the base ref the diff reads",
        );
        assert_eq!(refresh.state, BaseState::Current, "and the refreshed base is current");
    }

    /// **A measurement answers a read only inside the store's window.** The
    /// reuse is a freshness claim, so its expiry is a property with its own
    /// test: widening the window past the store's own lifetime has to fail
    /// here.
    #[test]
    fn a_measurement_answers_a_reading_only_inside_the_store_window() {
        let now = Instant::now();
        // **The figures are literal on purpose.** This read used to accept a
        // check thirty seconds old and now accepts the store's four minutes,
        // so the expiry is pinned at 240 s rather than derived from the
        // constant it guards: moving the window has to red here.
        let fresh = BaseMeasurement {
            at: now.checked_sub(Duration::from_secs(239)).expect("the test clock is past boot"),
            state: BaseState::Current,
            error: None,
        };
        let stale = BaseMeasurement {
            at: now.checked_sub(Duration::from_secs(241)).expect("the test clock is past boot"),
            state: BaseState::Current,
            error: None,
        };

        assert!(measurement_in_window(&fresh, now), "a check 239 s old answers the read");
        assert!(!measurement_in_window(&stale, now), "one 241 s old does not");
    }

    /// **An expired check is not drawn as current.** The store prunes only on
    /// write, so an aged entry survives between writes; the read applies the
    /// window itself, or a check hours old would draw as `current`. The held
    /// fetch keeps the kick from landing a new check under the assertions,
    /// and a fresh check still answers once it is in the window.
    #[tokio::test(flavor = "current_thread")]
    async fn a_reading_ignores_an_expired_measurement_and_answers_unchecked() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());
        remember_measurement(
            dir.path(),
            "origin/main",
            &BaseRefresh {
                state: BaseState::Current,
                base_ref: None,
                error: Some("a check from hours ago".to_owned()),
                cut: false,
                unanswered: false,
                at: Instant::now()
                    .checked_sub(WATCH_REFRESH_INTERVAL + Duration::from_secs(60))
                    .expect("the test clock is past boot"),
            },
        );

        let answer = refresh_base_for_display(dir.path(), "origin/main").await;
        held.release();
        let _ = settle_background_fetches().await;

        assert_eq!(answer.state, BaseState::Unverified, "an expired check is not drawn");
        assert!(answer.error.is_none(), "nor are its words carried: {answer:?}");
        assert!(answer.base_ref.is_some(), "the base is pinned to the local tip");

        // The same store answers once the check is inside the window.
        remember_measurement(
            dir.path(),
            "origin/main",
            &BaseRefresh {
                state: BaseState::Current,
                base_ref: None,
                error: None,
                cut: false,
                unanswered: false,
                at: Instant::now(),
            },
        );
        let fresh = refresh_base_for_display(dir.path(), "origin/main").await;
        crate::env::git_command::clear_git_point(dir.path());
        assert_eq!(fresh.state, BaseState::Current, "a check inside the window answers");
    }

    /// **A measurement answers only the base it measured.** One worktree can
    /// draw two bases, so a measurement of `origin/main` must not answer a read
    /// for `origin/develop`.
    #[tokio::test(flavor = "current_thread")]
    async fn a_measurement_answers_only_the_branch_it_measured() {
        let (_upstream, dir) = stale_base_fixture();

        let _ = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        crate::env::git_command::clear_git_point(dir.path());

        assert!(
            last_base_measurement(dir.path(), "origin/main").is_some(),
            "the refresh recorded the base it measured",
        );
        assert!(
            last_base_measurement(dir.path(), "origin/develop").is_none(),
            "and it must not answer a read for a different base",
        );
    }

    /// **A burst of kicks costs one fetch, not one each.** The first leads a
    /// run the wrapper holds; the rest join it and share its answer.
    #[tokio::test(flavor = "current_thread")]
    async fn readers_share_the_refresh_the_first_leads() {
        let (upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());

        let mut readers = Vec::new();
        for _ in 0..3 {
            let dir = dir.path().to_path_buf();
            readers.push(tokio::spawn(async move {
                refresh_base_within(&dir, "origin/main", TEST_GIT_BUDGET).await
            }));
        }
        // The first of them leads the fetch, which the wrapper holds; the rest
        // can only join it.
        wait_for(|| held.starts() == 1, "the first reader's fetch to start").await;
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(held.starts(), 1, "one fetch for the burst, not one per reader");

        held.release();
        for reader in readers {
            // A failure bound, not a performance claim: the fetch is released,
            // and a loaded gate must not read as a broken join.
            let read = timeout(TEST_GIT_BUDGET, reader)
                .await
                .expect("the read answered")
                .expect("the read task finished");
            assert_eq!(read.state, BaseState::Current, "every reader drew the refreshed base");
            assert!(!read.cut && !read.unanswered);
        }
        crate::env::git_command::clear_git_point(dir.path());
        assert_eq!(
            rev_parse(dir.path(), "refs/remotes/origin/main"),
            rev_parse(upstream.path(), "main"),
            "with the base at the remote's tip",
        );
    }

    /// **The timeout words carry the run's budget**, not what was left of it:
    /// a run that spent its bound before the child started would say "0s".
    #[tokio::test(flavor = "current_thread")]
    async fn a_cut_fetch_names_the_budget_it_was_given() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let held = held_fetch_at(dir.path(), bin.path());
        let bound = Duration::from_millis(300);

        let refresh = refresh_base_within(dir.path(), "origin/main", bound).await;
        held.release();
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());

        assert!(refresh.cut, "the run's bound ended the fetch");
        assert_eq!(
            refresh.error.as_deref(),
            Some("the fetch did not finish inside the 0.3s it was given"),
            "the run's own budget, not the nothing left of it",
        );
    }

    /// **Two scans inside one window cost one fetch.** The throttle the kick
    /// carries is what keeps a tree's every pass from reaching the network.
    #[tokio::test(flavor = "current_thread")]
    async fn two_scans_in_one_window_cost_one_fetch() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let counted = bin.path().join("fetches");
        let at = counted.display();
        let counting = format!("echo start >> '{at}'; exec git \"$@\"");
        write_wrapper(dir.path(), bin.path(), "counting-git", &[], &counting);
        let fetches = || {
            std::fs::read_to_string(&counted).map(|text| text.lines().count()).unwrap_or_default()
        };

        let _ = scan_in_test(dir.path(), &HashMap::new(), None).await;
        let _ = settle_background_fetches().await;
        let _ = scan_in_test(dir.path(), &HashMap::new(), None).await;
        let _ = settle_background_fetches().await;
        crate::env::git_command::clear_git_point(dir.path());

        assert_eq!(fetches(), 1, "the second scan's kick is throttled by the first");
    }

    /// **A checkout that names no ssh command is forge's to bound.** The
    /// missing key exits 1, and the classifier reads that as "unset" rather
    /// than as a probe that failed.
    #[tokio::test(flavor = "current_thread")]
    async fn a_checkout_that_names_no_ssh_command_gets_forges() {
        if std::env::var_os("GIT_SSH_COMMAND").is_some() || std::env::var_os("GIT_SSH").is_some() {
            return;
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let bin = tempfile::tempdir().expect("wrapper dir");
        init_repo(&dir, "main");
        without_machine_config_at(dir.path(), bin.path());

        assert!(ssh_is_forges(dir.path()).await, "an ordinary checkout is forge's to bound");
        crate::env::git_command::clear_git_point(dir.path());
    }

    /// **And one that names its own is kept**, whichever spelling: forge sets
    /// `GIT_SSH_COMMAND` only when nothing else names one.
    #[tokio::test(flavor = "current_thread")]
    async fn a_checkouts_own_ssh_command_is_kept() {
        if std::env::var_os("GIT_SSH_COMMAND").is_some() || std::env::var_os("GIT_SSH").is_some() {
            return;
        }
        let dir = tempfile::tempdir().expect("tempdir");
        let bin = tempfile::tempdir().expect("wrapper dir");
        init_repo(&dir, "main");
        git_at(dir.path(), &["config", "core.sshCommand", "/bin/true"]);
        without_machine_config_at(dir.path(), bin.path());

        assert!(!ssh_is_forges(dir.path()).await, "the checkout's command governs");
        crate::env::git_command::clear_git_point(dir.path());
    }

    /// **The classifier's answer reaches the child**: the fetch itself carries
    /// forge's bounded command on a checkout that names none, and nothing at
    /// all over one that names its own. The wrapper records the environment it
    /// was given, so this is the wiring and not the constructor alone.
    #[tokio::test(flavor = "current_thread")]
    async fn a_fetch_carries_forges_ssh_command_only_when_the_checkout_names_none() {
        if std::env::var_os("GIT_SSH_COMMAND").is_some() || std::env::var_os("GIT_SSH").is_some() {
            return;
        }
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let marker = bin.path().join("ssh-env");
        let at = marker.display();
        let capturing = format!(
            "echo \"ssh=${{GIT_SSH_COMMAND:-none}}\" >> '{at}'; \
             {ISOLATED_GIT_CONFIG} exec git \"$@\""
        );
        write_wrapper(dir.path(), bin.path(), "env-git", &[], &capturing);

        let first = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        git_at(dir.path(), &["config", "core.sshCommand", "/bin/true"]);
        let second = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        crate::env::git_command::clear_git_point(dir.path());

        assert_eq!(first.state, BaseState::Current);
        assert_eq!(second.state, BaseState::Current);
        let seen = std::fs::read_to_string(&marker).expect("the wrapper's record");
        let lines: Vec<&str> = seen.lines().collect();
        assert!(
            lines.first().is_some_and(|line| line.contains("ConnectTimeout=15")),
            "forge's bound is on the fetch of a checkout that names none: {seen:?}",
        );
        assert_eq!(
            lines.get(1).copied(),
            Some("ssh=none"),
            "and nothing is installed over the checkout's own command: {seen:?}",
        );
    }

    /// **Git's words lose any credential a remote URL carries**: the record is
    /// re-sent on every read and drawn on a page.
    #[test]
    fn a_remote_url_loses_its_credential_in_the_words_a_page_draws() {
        assert_eq!(
            redact(
                "fatal: unable to access 'https://user:s3cret@host.example/x.git/': Could not \
                 resolve host"
            ),
            "fatal: unable to access 'https://<redacted>@host.example/x.git/': Could not resolve \
             host",
        );
        assert_eq!(
            redact("fatal: could not read from 'https://ghp_token@host.example/x'"),
            "fatal: could not read from 'https://<redacted>@host.example/x'",
        );
        assert_eq!(redact("fatal: no url here"), "fatal: no url here");
        assert_eq!(
            redact("fatal: 'https://host.example:8443/x' refused"),
            "fatal: 'https://host.example:8443/x' refused",
            "a port is not a credential",
        );
    }

    /// **The credential goes before the cap does.** Applied first, the cap
    /// lands inside the userinfo and takes the `@` the redactor needs with it,
    /// so the head of the password reaches the record - and every page that
    /// draws it. A redaction case under 200 characters cannot tell the two
    /// orders apart; this one puts the cap inside the password.
    #[tokio::test(flavor = "current_thread")]
    async fn a_failed_fetchs_credential_goes_before_the_words_are_capped() {
        let (_upstream, dir) = stale_base_fixture();
        let bin = tempfile::tempdir().expect("wrapper dir");
        let refused = format!(
            "printf '%s\\n' \"fatal: {}https://user:SECRETSECRET@host.example/x: {}\" >&2; exit 1",
            "x".repeat(170),
            "y".repeat(300),
        );
        write_wrapper(dir.path(), bin.path(), "credential-git", &[], &refused);

        let refresh = refresh_base_within(dir.path(), "origin/main", TEST_GIT_BUDGET).await;
        crate::env::git_command::clear_git_point(dir.path());

        let words = refresh.error.expect("the fetch failed");
        assert!(
            !words.contains("SECRET"),
            "the credential reached the words a page draws: {words:?}",
        );
        assert!(
            words.contains("<redacted>"),
            "and the words still say the URL carried one: {words:?}",
        );
    }

    /// **The stderr clip cuts on a character boundary**: git's stderr is
    /// arbitrary bytes and a byte slice inside one panics mid-refresh.
    #[test]
    fn the_stderr_clip_cuts_on_a_character_boundary() {
        let text = format!("{}é", "a".repeat(STDERR_LOG_CAP - 1));
        let out = clipped(&text, STDERR_LOG_CAP);
        assert!(out.ends_with('…'), "the clip marks itself: {out:?}");
        assert!(!out.contains('é'), "the character the cap landed inside is dropped, not split");
    }
}