fallow_engine/

churn.rs

//! Git churn analysis for hotspot detection.
//!
//! Shells out to `git log` to collect per-file change history, then computes
//! recency-weighted churn scores and trend indicators.

use rustc_hash::FxHashMap;
use std::io::Read as _;
use std::path::{Component, Path, PathBuf};
use std::process::{Command, Output};
use std::sync::OnceLock;

use serde::Deserialize;

use crate::changed_files::git_path_from_bytes;

pub use fallow_types::churn::ChurnTrend;

/// Function pointer signature used by `set_spawn_hook` to intercept the
/// `git log --numstat` subprocess. Lets the CLI route long-running git
/// log calls through its `ScopedChild` registry so SIGINT / SIGTERM
/// reap the subprocess instead of leaving it running after the parent
/// exits. See `crates/cli/src/signal/` and issue #477.
pub type ChurnSpawnHook = fn(&mut Command) -> std::io::Result<Output>;

static SPAWN_HOOK: OnceLock<ChurnSpawnHook> = OnceLock::new();

fn git_command() -> Command {
    crate::git_env::git_command()
}

/// Install a spawn-hook that wraps the `git log` subprocess. Idempotent;
/// subsequent calls are no-ops. Called once from the CLI's `main()` to
/// route through the signal registry; defaults to `Command::output`
/// when not set so the function-pointer indirection stays free for tests
/// and embedders that don't care.
pub fn set_spawn_hook(hook: ChurnSpawnHook) {
    let _ = SPAWN_HOOK.set(hook);
}

fn spawn_output(command: &mut Command) -> std::io::Result<Output> {
    if let Some(hook) = SPAWN_HOOK.get() {
        hook(command)
    } else {
        command.output()
    }
}

/// Number of seconds in one day.
const SECS_PER_DAY: f64 = 86_400.0;

/// Recency weight half-life in days. A commit from 90 days ago counts half
/// as much as today's commit; 180 days ago counts 25%.
const HALF_LIFE_DAYS: f64 = 90.0;

/// Schema discriminator a `--churn-file` document must declare.
const CHURN_FILE_SCHEMA: &str = "fallow-churn/v1";

/// Upper bound on imported churn events. A file past this size is a sign of a
/// pathological export (whole-history dump of a giant monorepo) rather than a
/// useful hotspot window; parsing is rejected so we never allocate unbounded
/// state from a single untrusted file. Mirrors the diff parser's
/// `MAX_ADDED_LINES` guard in the CLI.
const MAX_CHURN_EVENTS: usize = 5_000_000;

/// Upper bound on the serialized churn import before JSON deserialization.
const MAX_CHURN_FILE_BYTES: usize = 256 * 1024 * 1024;

/// Reject an imported `timestamp` more than this many seconds in the future
/// (one year). A unix-seconds commit time is never legitimately this far ahead
/// even with clock skew, so a value past it is almost always a millisecond
/// timestamp (~52000 years out) or corruption. Caught loudly because the
/// recency decay uses `saturating_sub`, so a future timestamp would otherwise
/// clamp to age 0, give every commit full weight, and silently collapse the
/// recency signal that distinguishes recent from old churn.
const MAX_FUTURE_TIMESTAMP_SECS: u64 = 365 * 24 * 60 * 60;

/// Unit of a relative churn window.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChurnWindowUnit {
    /// Fixed 24-hour days.
    Days,
    /// Fixed 7-day weeks.
    Weeks,
    /// Calendar months in UTC.
    Months,
    /// Calendar years in UTC.
    Years,
}

impl ChurnWindowUnit {
    /// Single-letter token used in the churn cache key.
    const fn token(self) -> char {
        match self {
            Self::Days => 'd',
            Self::Weeks => 'w',
            Self::Months => 'm',
            Self::Years => 'y',
        }
    }
}

/// The span of history churn analysis covers.
///
/// A relative window is stored as its duration rather than as the wall-clock
/// string git resolves, so the same token always keys the same cache entry
/// while the cutoff it resolves to moves with the run clock.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChurnWindow {
    /// A span measured back from the run clock.
    Relative {
        /// How many `unit`s of history to cover.
        count: u64,
        /// The unit `count` is measured in.
        unit: ChurnWindowUnit,
    },
    /// An absolute ISO `YYYY-MM-DD` date, read as UTC midnight.
    Date(String),
    /// No window: imported churn covers whatever the exporter selected.
    Imported,
}

impl ChurnWindow {
    /// Stable cache key for this window. Absolute for a date, the duration
    /// token (`"1y"`, `"90d"`) for a relative span.
    #[must_use]
    pub fn cache_token(&self) -> String {
        match self {
            Self::Relative { count, unit } => format!("{count}{}", unit.token()),
            Self::Date(date) => date.clone(),
            Self::Imported => String::new(),
        }
    }

    /// The oldest commit timestamp this window includes, resolved against
    /// `clock`. `None` for imported churn, which has no cutoff to apply.
    #[must_use]
    pub fn cutoff_secs(&self, clock: &crate::clock::AnalysisClock) -> Option<u64> {
        match self {
            Self::Relative { count, unit } => Some(match unit {
                ChurnWindowUnit::Days => clock.minus_days(*count),
                ChurnWindowUnit::Weeks => clock.minus_days(count.saturating_mul(7)),
                ChurnWindowUnit::Months => clock.minus_months(*count),
                ChurnWindowUnit::Years => clock.minus_years(*count),
            }),
            Self::Date(date) => crate::clock::utc_midnight_epoch(date),
            Self::Imported => None,
        }
    }
}

/// Parsed `--since` window plus the label reports print for it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SinceDuration {
    /// History span to analyze.
    pub window: ChurnWindow,
    /// Human-readable display string (e.g., `"6 months"`).
    pub display: String,
}

impl SinceDuration {
    /// A relative window of `count` `unit`s, labelled for report headers.
    #[must_use]
    pub fn relative(count: u64, unit: ChurnWindowUnit, display: impl Into<String>) -> Self {
        Self {
            window: ChurnWindow::Relative { count, unit },
            display: display.into(),
        }
    }
}

/// Per-author commit aggregation for a single file.
///
/// Authors are interned via [`ChurnResult::author_pool`] indices to keep
/// per-file maps small and the bitcode cache compact.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct AuthorContribution {
    /// Total commits by this author touching this file in the analysis window.
    pub commits: u32,
    /// Recency-weighted commit sum (exponential decay, half-life 90 days).
    pub weighted_commits: f64,
    /// Earliest commit timestamp by this author (epoch seconds).
    pub first_commit_ts: u64,
    /// Latest commit timestamp by this author (epoch seconds).
    pub last_commit_ts: u64,
}

/// Per-file churn data collected from git history.
#[derive(Debug, Clone)]
pub struct FileChurn {
    /// Absolute file path.
    pub path: PathBuf,
    /// Total number of commits touching this file in the analysis window.
    pub commits: u32,
    /// Recency-weighted commit count (exponential decay, half-life 90 days).
    pub weighted_commits: f64,
    /// Total lines added across all commits.
    pub lines_added: u32,
    /// Total lines deleted across all commits.
    pub lines_deleted: u32,
    /// Churn trend: accelerating, stable, or cooling.
    pub trend: ChurnTrend,
    /// Per-author contributions keyed by interned author index.
    /// Indices reference [`ChurnResult::author_pool`].
    pub authors: FxHashMap<u32, AuthorContribution>,
}

/// Result of churn analysis.
#[derive(Debug, Clone)]
pub struct ChurnResult {
    /// Per-file churn data, keyed by absolute path.
    pub files: FxHashMap<PathBuf, FileChurn>,
    /// Whether the repository is a shallow clone.
    pub shallow_clone: bool,
    /// Author email pool. Per-file [`AuthorContribution`] entries reference
    /// authors by their index into this vector.
    pub author_pool: Vec<String>,
    /// The instant recency weighting and staleness were measured against.
    /// Ownership and routing reuse it so every churn-derived number in one run
    /// agrees on "now".
    pub clock: crate::clock::AnalysisClock,
    /// Bytes of `git log` output that this run read. Zero when the churn
    /// cache or a churn file supplied all the history.
    pub git_log_bytes: u64,
    /// Every commit in the history window with the files it changed, sorted
    /// by committer timestamp and then by id. The history window can be
    /// wider than the window of [`files`](Self::files), see
    /// [`analyze_churn_cached_with_history`]. Imported events without a
    /// `commit` id are not in this list.
    pub commits: Vec<ChurnCommit>,
}

/// One commit in the churn history window and the files it changed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChurnCommit {
    /// Commit id (`%H`), or the opaque `commit` value of an imported event.
    pub sha: String,
    /// Committer timestamp (epoch seconds).
    pub committed_at: u64,
    /// Changed files, sorted by path.
    pub files: Vec<ChurnCommitFile>,
}

/// One file that a [`ChurnCommit`] changed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChurnCommitFile {
    /// Absolute file path.
    pub path: PathBuf,
    /// Lines added plus lines deleted in this file in this commit.
    pub lines_changed: u32,
}

/// Parse a `--since` value into a git-compatible duration.
///
/// Accepts:
/// - Durations: `6m`, `6months`, `90d`, `90days`, `1y`, `1year`, `2w`, `2weeks`
/// - ISO dates: `2025-06-01`
///
/// # Errors
///
/// Returns an error if the input is not a recognized duration format or ISO date,
/// the numeric part is invalid, or the duration is zero.
pub fn parse_since(input: &str) -> Result<SinceDuration, String> {
    if is_iso_date(input) {
        return Ok(SinceDuration {
            window: ChurnWindow::Date(input.to_string()),
            display: input.to_string(),
        });
    }

    let (num_str, unit) = split_number_unit(input)?;
    let num: u64 = num_str
        .parse()
        .map_err(|_| format!("invalid number in --since: {input}"))?;

    if num == 0 {
        return Err("--since duration must be greater than 0".to_string());
    }

    let (unit, label) = match unit {
        "d" | "day" | "days" => (ChurnWindowUnit::Days, "day"),
        "w" | "week" | "weeks" => (ChurnWindowUnit::Weeks, "week"),
        "m" | "month" | "months" => (ChurnWindowUnit::Months, "month"),
        "y" | "year" | "years" => (ChurnWindowUnit::Years, "year"),
        _ => {
            return Err(format!(
                "unknown duration unit '{unit}' in --since. Use d/w/m/y (e.g., 6m, 90d, 1y)"
            ));
        }
    };
    let plural = if num == 1 { "" } else { "s" };
    Ok(SinceDuration::relative(
        num,
        unit,
        format!("{num} {label}{plural}"),
    ))
}

/// Analyze git churn for files in the given root directory.
///
/// Returns `None` if git is not available or the directory is not a git repository.
pub fn analyze_churn(root: &Path, since: &SinceDuration) -> Option<ChurnResult> {
    let clock = crate::clock::AnalysisClock::for_repo(root);
    let shallow = is_shallow_clone(root);
    let state = analyze_churn_events(root, since, None, &clock)?;
    Some(build_churn_result(state, shallow, clock))
}

/// A `fallow-churn/v1` import document: a normalized, VCS-agnostic stand-in for
/// `git log --numstat` output. Unknown fields are ignored (no
/// `deny_unknown_fields`) so wrappers may carry extra metadata.
#[derive(Debug, Deserialize)]
struct ChurnFileDoc {
    schema: String,
    #[serde(default)]
    events: Vec<ChurnFileEvent>,
}

/// One per-(commit, file) change event, the natural shape of a `<vcs> log
/// --numstat` row. The optional `commit` id groups events into commits, which
/// change coupling and change dispersion need. An event without it still
/// feeds per-file churn.
#[derive(Debug, Deserialize)]
struct ChurnFileEvent {
    /// Opaque commit id. Events with the same id belong to one commit.
    #[serde(default)]
    commit: Option<String>,
    /// Repo-root-relative, forward-slash path. Joined to `root`.
    path: String,
    /// Commit time, unix SECONDS UTC (not milliseconds).
    timestamp: u64,
    /// Opaque author identity (email recommended); absent contributes no
    /// ownership signal. fallow does NOT apply mailmap to imported authors.
    #[serde(default)]
    author: Option<String>,
    /// Lines added in this file in this commit.
    added: u32,
    /// Lines deleted in this file in this commit.
    deleted: u32,
}

/// Build churn data from a normalized `fallow-churn/v1` JSON import instead of
/// `git log`. Lets projects on a non-git VCS (Yandex Arc, Mercurial, Perforce)
/// feed change history into hotspot / ownership / bus-factor analysis: a small
/// wrapper translates the VCS log into the contract and fallow runs all the
/// usual recency-weighting, trend, and ownership logic on the imported events.
///
/// `root` is the project root that relative event paths are joined to (matching
/// how the git path joins numstat paths), so the churn keys line up with the
/// analyzed files. Returns a human-readable error (the CLI maps it to exit code
/// 2) on an oversized or missing file, malformed JSON, wrong `schema`, an
/// invalid repo-relative event path, a far-future timestamp, line totals above
/// `u32::MAX`, or an event count past `MAX_CHURN_EVENTS`. An empty `events`
/// array is valid (no hotspots), not an error. Never runs `git`.
pub(crate) fn analyze_churn_from_file(path: &Path, root: &Path) -> Result<ChurnResult, String> {
    let raw = read_churn_file_with_limit(path, MAX_CHURN_FILE_BYTES)?;
    let doc: ChurnFileDoc = serde_json::from_str(&raw)
        .map_err(|e| format!("failed to parse churn file {}: {e}", path.display()))?;
    if doc.schema != CHURN_FILE_SCHEMA {
        return Err(format!(
            "churn file {} declares schema \"{}\", expected \"{CHURN_FILE_SCHEMA}\"",
            path.display(),
            doc.schema
        ));
    }
    if doc.events.len() > MAX_CHURN_EVENTS {
        return Err(format!(
            "churn file {} has {} events, exceeding the {MAX_CHURN_EVENTS} limit",
            path.display(),
            doc.events.len()
        ));
    }

    let state = churn_event_state_from_doc(&doc, path, root)?;
    Ok(build_churn_result(
        state,
        false,
        crate::clock::AnalysisClock::for_repo(root),
    ))
}

fn read_churn_file_with_limit(path: &Path, limit: usize) -> Result<String, String> {
    let file = std::fs::File::open(path)
        .map_err(|e| format!("failed to read churn file {}: {e}", path.display()))?;
    let mut bytes = Vec::new();
    file.take(limit as u64 + 1)
        .read_to_end(&mut bytes)
        .map_err(|e| format!("failed to read churn file {}: {e}", path.display()))?;
    if bytes.len() > limit {
        return Err(format!(
            "churn file {} is at least {} bytes, exceeding the {limit} byte limit",
            path.display(),
            bytes.len()
        ));
    }
    String::from_utf8(bytes)
        .map_err(|e| format!("failed to read churn file {} as UTF-8: {e}", path.display()))
}

/// Validate and fold a parsed `fallow-churn/v1` document into event state.
///
/// Rejects invalid paths, far-future (likely millisecond) timestamps, and line
/// totals outside the public `u32` contract. Interns authors into the pool
/// exactly as the git-log path does.
fn churn_event_state_from_doc(
    doc: &ChurnFileDoc,
    path: &Path,
    root: &Path,
) -> Result<ChurnEventState, String> {
    let mut builder = ChurnFileImportBuilder::new(path, root, churn_file_future_limit());

    for event in &doc.events {
        builder.push_event(event)?;
    }

    Ok(builder.finish())
}

/// The ceiling above which an imported event timestamp is rejected as
/// implausible (almost always seconds-versus-milliseconds confusion).
///
/// This deliberately reads the wall clock rather than the run's
/// [`crate::clock::AnalysisClock`]. The limit gates *acceptance* of an import,
/// never a scored value, so it cannot move `weighted_commits` or `stale_days`
/// the way a wall-clock "now" in the scoring path would. Pinning it to the run
/// clock would instead make the gate reject real data: the run clock is HEAD's
/// committer timestamp, so analyzing an older checkout while importing churn
/// that covers today would drop every recent event. Wall-clock drift here is
/// also one-directional, since a timestamp accepted today stays accepted on
/// every later run.
fn churn_file_future_limit() -> u64 {
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();
    now_secs.saturating_add(MAX_FUTURE_TIMESTAMP_SECS)
}

struct ChurnFileImportBuilder<'a> {
    path: &'a Path,
    root: &'a Path,
    future_limit: u64,
    files: FxHashMap<PathBuf, FileEvents>,
    totals: FxHashMap<PathBuf, (u64, u64)>,
    author_pool: Vec<String>,
    author_index: FxHashMap<String, u32>,
    commit_pool: Vec<String>,
    commit_index: FxHashMap<String, u32>,
}

impl<'a> ChurnFileImportBuilder<'a> {
    fn new(path: &'a Path, root: &'a Path, future_limit: u64) -> Self {
        Self {
            path,
            root,
            future_limit,
            files: FxHashMap::default(),
            totals: FxHashMap::default(),
            author_pool: Vec::new(),
            author_index: FxHashMap::default(),
            commit_pool: Vec::new(),
            commit_index: FxHashMap::default(),
        }
    }

    fn push_event(&mut self, event: &ChurnFileEvent) -> Result<(), String> {
        let rel = normalize_churn_event_path(self.path, &event.path)?;
        validate_churn_event_timestamp(self.path, event.timestamp, self.future_limit, &rel)?;

        let abs_path = self.root.join(&rel);
        let totals = self.totals.entry(abs_path.clone()).or_default();
        totals.0 += u64::from(event.added);
        totals.1 += u64::from(event.deleted);
        if totals.0 > u64::from(u32::MAX) || totals.1 > u64::from(u32::MAX) {
            return Err(format!(
                "churn file {} has line totals for \"{rel}\" exceeding the u32 limit \
                 (added {}, deleted {})",
                self.path.display(),
                totals.0,
                totals.1
            ));
        }
        let author_idx = self.intern_author(event.author.as_deref());
        let commit_idx = event
            .commit
            .as_deref()
            .map(str::trim)
            .filter(|commit| !commit.is_empty())
            .map(|commit| intern_author(commit, &mut self.commit_pool, &mut self.commit_index));
        self.files
            .entry(abs_path)
            .or_insert_with(|| FileEvents { events: Vec::new() })
            .events
            .push(CachedCommitEvent {
                timestamp: event.timestamp,
                committed_at: event.timestamp,
                lines_added: event.added,
                lines_deleted: event.deleted,
                author_idx,
                commit_idx,
            });
        Ok(())
    }

    fn intern_author(&mut self, author: Option<&str>) -> Option<u32> {
        author
            .map(str::trim)
            .filter(|email| !email.is_empty())
            .map(|email| intern_author(email, &mut self.author_pool, &mut self.author_index))
    }

    fn finish(self) -> ChurnEventState {
        ChurnEventState {
            files: self.files,
            author_pool: self.author_pool,
            commit_pool: self.commit_pool,
            git_log_bytes: 0,
        }
    }
}

fn normalize_churn_event_path(path: &Path, event_path: &str) -> Result<String, String> {
    let normalized = event_path.replace('\\', "/");
    let rel = normalized.trim();
    if rel.is_empty() {
        return Err(format!(
            "churn file {} has an event with an empty path",
            path.display()
        ));
    }
    let bytes = rel.as_bytes();
    let has_drive_prefix = bytes.len() >= 2 && bytes[0].is_ascii_alphabetic() && bytes[1] == b':';
    let has_empty_component = rel.split('/').any(str::is_empty);
    let components_are_normal = Path::new(rel)
        .components()
        .all(|component| matches!(component, Component::Normal(_)));
    if rel.starts_with('/') || has_drive_prefix || has_empty_component || !components_are_normal {
        return Err(format!(
            "churn file {} has an invalid repo-relative event path \"{event_path}\"",
            path.display()
        ));
    }
    Ok(rel.to_string())
}

fn validate_churn_event_timestamp(
    path: &Path,
    timestamp: u64,
    future_limit: u64,
    rel: &str,
) -> Result<(), String> {
    if timestamp <= future_limit {
        return Ok(());
    }

    Err(format!(
        "churn file {} has event timestamp {} for \"{rel}\" more than a year in the \
         future; timestamps must be unix SECONDS (not milliseconds), UTC",
        path.display(),
        timestamp
    ))
}

/// Check if the repository is a shallow clone.
#[must_use]
fn is_shallow_clone(root: &Path) -> bool {
    let mut command = git_command();
    command
        .args(["rev-parse", "--is-shallow-repository"])
        .current_dir(root);
    command.output().is_ok_and(|o| {
        String::from_utf8_lossy(&o.stdout)
            .trim()
            .eq_ignore_ascii_case("true")
    })
}

/// Check if the directory is inside a git repository.
#[must_use]
pub fn is_git_repo(root: &Path) -> bool {
    let mut command = git_command();
    command
        .args(["rev-parse", "--git-dir"])
        .current_dir(root)
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null());
    command.status().is_ok_and(|s| s.success())
}

/// Maximum size of a churn cache file (64 MB). The incremental cache stores
/// per-commit events, so it needs more headroom than the old aggregate rows.
const MAX_CHURN_CACHE_SIZE: usize = 64 * 1024 * 1024;

/// Cache schema version. Bump when the on-disk shape of [`ChurnCache`]
/// changes so older payloads are rejected on load. Version 5 stores paths in
/// their platform-native byte representation instead of lossy UTF-8 strings.
/// Version 6 keys on the window token instead of a wall-clock-resolved git
/// date string and stores each event's committer timestamp so a warm load can
/// prune to the same cutoff `git log --after` applies on a cold run.
/// Version 7 scopes the log to the project root, so entries written from a
/// subdirectory root with toplevel-relative paths are rejected.
/// Version 8 stores the commit id of each event, so change coupling can
/// rebuild the file set of each commit from the cache.
const CHURN_CACHE_VERSION: u8 = 8;

/// Serializable per-commit event for the disk cache.
#[derive(Clone, bitcode::Encode, bitcode::Decode)]
struct CachedCommitEvent {
    timestamp: u64,
    /// Committer timestamp. `git log --after` filters on this, while recency
    /// weighting stays on the author timestamp, so a rebased commit lands in
    /// the same window warm and cold without its age changing.
    committed_at: u64,
    lines_added: u32,
    lines_deleted: u32,
    author_idx: Option<u32>,
    /// Index into the commit pool. `None` for an imported event without a
    /// commit id.
    commit_idx: Option<u32>,
}

/// Serializable per-file churn entry for the disk cache.
#[derive(Clone, bitcode::Encode, bitcode::Decode)]
struct CachedFileChurn {
    path: Vec<u8>,
    events: Vec<CachedCommitEvent>,
}

/// Cached churn data keyed by last indexed SHA and since string.
#[derive(Clone, bitcode::Encode, bitcode::Decode)]
struct ChurnCache {
    /// Schema version; must equal [`CHURN_CACHE_VERSION`] to be accepted.
    version: u8,
    last_indexed_sha: String,
    /// [`ChurnWindow::cache_token`] of the window this entry was built for.
    window_token: String,
    files: Vec<CachedFileChurn>,
    shallow_clone: bool,
    /// Author email pool referenced by [`CachedCommitEvent::author_idx`].
    author_pool: Vec<String>,
    /// Commit id pool referenced by [`CachedCommitEvent::commit_idx`].
    commit_pool: Vec<String>,
}

/// Per-file commit events retained in memory while building or updating churn.
struct FileEvents {
    events: Vec<CachedCommitEvent>,
}

/// Event-level churn state. Unlike [`ChurnResult`], this preserves commit
/// timestamps so a cache can merge new commits and recompute trend/recency.
struct ChurnEventState {
    files: FxHashMap<PathBuf, FileEvents>,
    author_pool: Vec<String>,
    /// Commit id pool referenced by [`CachedCommitEvent::commit_idx`].
    commit_pool: Vec<String>,
    /// Bytes of `git log` output that built this state in this run. Zero for
    /// state restored from the churn cache or read from a churn file.
    git_log_bytes: u64,
}

/// Get the full HEAD SHA for cache keying.
fn get_head_sha(root: &Path) -> Option<String> {
    crate::repo_refs::head_sha(root).ok().flatten()
}

/// Check whether `ancestor` is still reachable from `descendant`.
fn is_ancestor(root: &Path, ancestor: &str, descendant: &str) -> bool {
    let mut command = git_command();
    command
        .args(["merge-base", "--is-ancestor", ancestor, descendant])
        .current_dir(root);
    command.status().is_ok_and(|s| s.success())
}

/// Read the churn cache file. Returns `None` when the file is missing, too
/// large, unreadable, or written by another cache version.
fn read_churn_cache(cache_dir: &Path) -> Option<ChurnCache> {
    let cache_file = cache_dir.join("churn.bin");
    let data = std::fs::read(&cache_file).ok()?;
    if data.len() > MAX_CHURN_CACHE_SIZE {
        return None;
    }
    let cache: ChurnCache = bitcode::decode(&data).ok()?;
    (cache.version == CHURN_CACHE_VERSION).then_some(cache)
}

/// Try to load churn data from disk cache. Returns `None` on cache miss
/// or version mismatch.
#[cfg(test)]
fn load_churn_cache(cache_dir: &Path, window_token: &str) -> Option<ChurnCache> {
    read_churn_cache(cache_dir).filter(|cache| cache.window_token == window_token)
}

/// True when `outer` includes every commit that `inner` includes at `clock`.
fn window_covers(
    outer: &ChurnWindow,
    inner: &ChurnWindow,
    clock: &crate::clock::AnalysisClock,
) -> bool {
    if outer == inner {
        return true;
    }
    match (outer.cutoff_secs(clock), inner.cutoff_secs(clock)) {
        (Some(outer_cutoff), Some(inner_cutoff)) => outer_cutoff <= inner_cutoff,
        _ => false,
    }
}

/// Load a cache entry whose window covers `window`, plus the window that the
/// entry was built for.
///
/// One cache file serves several windows. Health reads a 12 month history
/// while target churn reads 6 months. A narrower request prunes the wider
/// entry to its own cutoff, which gives the same events as a cold
/// `git log --after`, so the commands do not overwrite each other's entry.
fn load_covering_churn_cache(
    cache_dir: &Path,
    window: &ChurnWindow,
    clock: &crate::clock::AnalysisClock,
) -> Option<(ChurnCache, SinceDuration)> {
    let cache = read_churn_cache(cache_dir)?;
    let cached = parse_since(&cache.window_token).ok()?;
    window_covers(&cached.window, window, clock).then_some((cache, cached))
}

/// Save churn data to disk cache.
fn save_churn_cache(
    cache_dir: &Path,
    last_indexed_sha: &str,
    window_token: &str,
    state: &ChurnEventState,
    shallow_clone: bool,
) {
    let files: Vec<CachedFileChurn> = state
        .files
        .iter()
        .map(|f| CachedFileChurn {
            path: path_to_cache_bytes(f.0),
            events: f.1.events.clone(),
        })
        .collect();
    let cache = ChurnCache {
        version: CHURN_CACHE_VERSION,
        last_indexed_sha: last_indexed_sha.to_string(),
        window_token: window_token.to_string(),
        files,
        shallow_clone,
        author_pool: state.author_pool.clone(),
        commit_pool: state.commit_pool.clone(),
    };
    let _ = std::fs::create_dir_all(cache_dir);
    let data = bitcode::encode(&cache);
    let tmp = cache_dir.join("churn.bin.tmp");
    if std::fs::write(&tmp, data).is_ok() {
        let _ = std::fs::rename(&tmp, cache_dir.join("churn.bin"));
    }
}

/// Analyze churn with disk caching. Uses cached result when HEAD SHA and
/// since duration match. If HEAD advanced from the cached SHA, runs an
/// incremental `git log <cached>..HEAD --numstat` scan and merges it.
///
/// Returns `(ChurnResult, bool)` where the bool indicates whether reusable
/// cache state was used.
/// Returns `None` if git analysis fails.
pub fn analyze_churn_cached(
    root: &Path,
    since: &SinceDuration,
    cache_dir: &Path,
    no_cache: bool,
) -> Option<(ChurnResult, bool)> {
    analyze_churn_cached_with_history(root, since, since, cache_dir, no_cache)
}

/// Analyze churn with disk caching over two windows.
///
/// [`ChurnResult::files`] covers `since`, which is the hotspot window.
/// [`ChurnResult::commits`] covers `history`, which change coupling and
/// change dispersion read. One `git log` pass over the wider window feeds
/// both. When `since` is the wider window, it is used for both.
///
/// Returns `None` if git analysis fails.
pub fn analyze_churn_cached_with_history(
    root: &Path,
    since: &SinceDuration,
    history: &SinceDuration,
    cache_dir: &Path,
    no_cache: bool,
) -> Option<(ChurnResult, bool)> {
    let head_sha = get_head_sha(root)?;
    let clock = crate::clock::AnalysisClock::for_repo(root);
    let history = if window_covers(&history.window, &since.window, &clock) {
        history
    } else {
        since
    };
    let files_cutoff = since.window.cutoff_secs(&clock);

    if !no_cache
        && let Some((state, shallow_clone)) =
            try_reuse_churn_cache(root, history, cache_dir, &head_sha, &clock)
    {
        return Some((
            build_windowed_churn_result(state, files_cutoff, shallow_clone, clock),
            true,
        ));
    }

    let (state, shallow_clone) =
        analyze_fresh_churn(root, history, cache_dir, no_cache, &head_sha, &clock)?;
    Some((
        build_windowed_churn_result(state, files_cutoff, shallow_clone, clock),
        false,
    ))
}

fn try_reuse_churn_cache(
    root: &Path,
    since: &SinceDuration,
    cache_dir: &Path,
    head_sha: &str,
    clock: &crate::clock::AnalysisClock,
) -> Option<(ChurnEventState, bool)> {
    let (cache, cached_since) = load_covering_churn_cache(cache_dir, &since.window, clock)?;
    let cutoff = since.window.cutoff_secs(clock);
    if cache.last_indexed_sha == head_sha {
        let shallow_clone = cache.shallow_clone;
        return Some((cache.into_event_state(cutoff), shallow_clone));
    }

    if !is_ancestor(root, &cache.last_indexed_sha, head_sha) {
        return None;
    }

    // Extend the entry over its own window, so a narrower request does not
    // shrink the entry that a wider request saved.
    let (mut state, shallow_clone) =
        extend_churn_cache(root, &cached_since, cache_dir, head_sha, cache, clock)?;
    state.retain_window(cutoff);
    Some((state, shallow_clone))
}

fn extend_churn_cache(
    root: &Path,
    since: &SinceDuration,
    cache_dir: &Path,
    head_sha: &str,
    cache: ChurnCache,
    clock: &crate::clock::AnalysisClock,
) -> Option<(ChurnEventState, bool)> {
    let shallow_clone = is_shallow_clone(root);
    let range = format!("{}..HEAD", cache.last_indexed_sha);
    let delta = analyze_churn_events(root, since, Some(&range), clock)?;
    let mut state = cache.into_event_state(since.window.cutoff_secs(clock));
    merge_churn_states(&mut state, delta);
    save_churn_cache(
        cache_dir,
        head_sha,
        &since.window.cache_token(),
        &state,
        shallow_clone,
    );
    Some((state, shallow_clone))
}

fn analyze_fresh_churn(
    root: &Path,
    since: &SinceDuration,
    cache_dir: &Path,
    no_cache: bool,
    head_sha: &str,
    clock: &crate::clock::AnalysisClock,
) -> Option<(ChurnEventState, bool)> {
    let shallow_clone = is_shallow_clone(root);
    let state = analyze_churn_events(root, since, None, clock)?;
    if !no_cache {
        save_churn_cache(
            cache_dir,
            head_sha,
            &since.window.cache_token(),
            &state,
            shallow_clone,
        );
    }

    Some((state, shallow_clone))
}

impl ChurnCache {
    /// Rehydrate the cached events, dropping everything the current window no
    /// longer covers.
    ///
    /// The cache only ever appends, so an entry minted months ago still holds
    /// commits a cold `git log --after` would exclude today. Without this prune
    /// a warm run reports more history than a cold run over the same commit,
    /// which is exactly the cache-transparency invariant fallow asserts
    /// elsewhere.
    fn into_event_state(self, cutoff_secs: Option<u64>) -> ChurnEventState {
        let files = self
            .files
            .into_iter()
            .filter_map(|entry| {
                let path = path_from_cache_bytes(&entry.path)?;
                Some((
                    path,
                    FileEvents {
                        events: entry.events,
                    },
                ))
            })
            .collect();
        let mut state = ChurnEventState {
            files,
            author_pool: self.author_pool,
            commit_pool: self.commit_pool,
            git_log_bytes: 0,
        };
        state.retain_window(cutoff_secs);
        state
    }
}

impl ChurnEventState {
    /// Drop every event older than `cutoff_secs` (committer time), and every
    /// file that has no event left. `None` keeps everything.
    fn retain_window(&mut self, cutoff_secs: Option<u64>) {
        let Some(cutoff) = cutoff_secs else {
            return;
        };
        self.files.retain(|_, file| {
            file.events.retain(|event| event.committed_at >= cutoff);
            !file.events.is_empty()
        });
    }
}

#[cfg(unix)]
fn path_to_cache_bytes(path: &Path) -> Vec<u8> {
    use std::os::unix::ffi::OsStrExt;

    path.as_os_str().as_bytes().to_vec()
}

#[cfg(unix)]
#[allow(
    clippy::unnecessary_wraps,
    reason = "Windows rejects truncated UTF-16 bytes through this shared fallible contract"
)]
fn path_from_cache_bytes(path: &[u8]) -> Option<PathBuf> {
    use std::ffi::OsStr;
    use std::os::unix::ffi::OsStrExt;

    Some(PathBuf::from(OsStr::from_bytes(path)))
}

#[cfg(windows)]
fn path_to_cache_bytes(path: &Path) -> Vec<u8> {
    use std::os::windows::ffi::OsStrExt;

    path.as_os_str()
        .encode_wide()
        .flat_map(u16::to_le_bytes)
        .collect()
}

#[cfg(windows)]
fn path_from_cache_bytes(path: &[u8]) -> Option<PathBuf> {
    use std::ffi::OsString;
    use std::os::windows::ffi::OsStringExt;

    let (chunks, remainder) = path.as_chunks::<2>();
    if !remainder.is_empty() {
        return None;
    }
    let wide: Vec<u16> = chunks
        .iter()
        .map(|&pair| u16::from_le_bytes(pair))
        .collect();
    Some(PathBuf::from(OsString::from_wide(&wide)))
}

/// Run `git log --numstat` and return event-level churn state.
///
/// The window is passed as an absolute `--after=@<epoch>` resolved against the
/// run clock rather than as a phrase git re-resolves against the wall clock, so
/// two runs over one commit see the same window boundary.
fn analyze_churn_events(
    root: &Path,
    since: &SinceDuration,
    revision_range: Option<&str>,
    clock: &crate::clock::AnalysisClock,
) -> Option<ChurnEventState> {
    let mut command = git_command();
    command.arg("log");
    if let Some(range) = revision_range {
        command.arg(range);
    }
    command
        .args([
            "--numstat",
            "--no-merges",
            "--no-renames",
            "--use-mailmap",
            "-z",
            "--format=format:%at|%ct|%H|%ae%x00",
        ])
        .current_dir(root);
    if let Some(cutoff) = since.window.cutoff_secs(clock) {
        command.arg(format!("--after=@{cutoff}"));
    }
    // numstat paths are relative to the git toplevel. When the project root
    // is a subdirectory, `--relative` rewrites them relative to the root, and
    // the `.` pathspec skips commits and files outside it. Both are no-ops at
    // the toplevel.
    command.args(["--relative", "--", "."]);

    let output = match spawn_output(&mut command) {
        Ok(o) => o,
        Err(e) => {
            tracing::warn!("hotspot analysis skipped: failed to run git: {e}");
            return None;
        }
    };

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        tracing::warn!("hotspot analysis skipped: git log failed: {stderr}");
        return None;
    }

    let mut state = parse_git_log_events_z(&output.stdout, root, clock.epoch_secs());
    state.git_log_bytes = output.stdout.len() as u64;
    Some(state)
}

/// Intern every value of `delta` into `base` and map each old `delta` index to
/// its index in `base`.
fn remap_pool(base: &mut Vec<String>, delta: Vec<String>) -> FxHashMap<u32, u32> {
    let mut base_index: FxHashMap<String, u32> = base
        .iter()
        .enumerate()
        .filter_map(|(idx, value)| u32::try_from(idx).ok().map(|idx| (value.clone(), idx)))
        .collect();
    let mut mapping: FxHashMap<u32, u32> = FxHashMap::default();
    for (old_idx, value) in delta.into_iter().enumerate() {
        let Ok(old_idx) = u32::try_from(old_idx) else {
            continue;
        };
        mapping.insert(old_idx, intern_author(&value, base, &mut base_index));
    }
    mapping
}

/// Merge new churn events into cached event state.
fn merge_churn_states(base: &mut ChurnEventState, delta: ChurnEventState) {
    base.git_log_bytes += delta.git_log_bytes;
    let author_mapping = remap_pool(&mut base.author_pool, delta.author_pool);
    let commit_mapping = remap_pool(&mut base.commit_pool, delta.commit_pool);

    for (path, mut file) in delta.files {
        for event in &mut file.events {
            event.author_idx = event
                .author_idx
                .and_then(|idx| author_mapping.get(&idx).copied());
            event.commit_idx = event
                .commit_idx
                .and_then(|idx| commit_mapping.get(&idx).copied());
        }
        base.files
            .entry(path)
            .and_modify(|existing| existing.events.append(&mut file.events))
            .or_insert(file);
    }
}

/// `now_secs` is the run clock's epoch, not the wall clock: it is the fallback
/// timestamp for a numstat record that arrives before any commit header, so
/// truncated or malformed git output still scores against the pinned instant.
fn parse_git_log_events_z(stdout: &[u8], root: &Path, now_secs: u64) -> ChurnEventState {
    let mut parser = GitLogEventParser::new(root, now_secs);
    for record in stdout.split(|byte| *byte == 0) {
        let record = record.strip_prefix(b"\n").unwrap_or(record);
        if record.is_empty() {
            continue;
        }
        if record.contains(&b'\t') {
            parser.record_numstat_bytes(record);
        } else {
            parser.consume_header(&String::from_utf8_lossy(record));
        }
    }
    parser.finish()
}

struct GitLogEventParser<'a> {
    root: &'a Path,
    now_secs: u64,
    files: FxHashMap<PathBuf, FileEvents>,
    author_pool: Vec<String>,
    author_index: FxHashMap<String, u32>,
    commit_pool: Vec<String>,
    commit_index: FxHashMap<String, u32>,
    current_timestamp: Option<u64>,
    current_committed_at: Option<u64>,
    current_author_idx: Option<u32>,
    current_commit_idx: Option<u32>,
}

impl<'a> GitLogEventParser<'a> {
    fn new(root: &'a Path, now_secs: u64) -> Self {
        Self {
            root,
            now_secs,
            files: FxHashMap::default(),
            author_pool: Vec::new(),
            author_index: FxHashMap::default(),
            commit_pool: Vec::new(),
            commit_index: FxHashMap::default(),
            current_timestamp: None,
            current_committed_at: None,
            current_author_idx: None,
            current_commit_idx: None,
        }
    }

    /// Numstat rows always hold a tab, so a tab-free record is a commit header,
    /// a legacy bare timestamp, or noise.
    fn consume_header(&mut self, line: &str) {
        let line = line.trim();
        if line.is_empty() || self.record_commit_header(line) {
            return;
        }
        self.record_legacy_timestamp(line);
    }

    /// Parse a `%at|%ct|%H|%ae` commit header. The older `%at|%ct|%ae` and
    /// `%at|%ae` headers are still accepted so a fixture or an embedder pinned
    /// to an older format keeps parsing. A header without `%H` gives events
    /// without a commit id, and a missing committer timestamp falls back to the
    /// author one.
    fn record_commit_header(&mut self, line: &str) -> bool {
        let Some((ts_str, rest)) = line.split_once('|') else {
            return false;
        };
        let Ok(ts) = ts_str.parse::<u64>() else {
            return false;
        };
        let (committed_at, rest) = match rest.split_once('|') {
            Some((committer_str, rest)) => (committer_str.parse::<u64>().unwrap_or(ts), rest),
            None => (ts, rest),
        };
        let (sha, email) = match rest.split_once('|') {
            Some((sha, email)) if is_commit_id(sha) => (Some(sha), email),
            _ => (None, rest),
        };

        self.current_timestamp = Some(ts);
        self.current_committed_at = Some(committed_at);
        self.current_author_idx = Some(intern_author(
            email,
            &mut self.author_pool,
            &mut self.author_index,
        ));
        self.current_commit_idx =
            sha.map(|sha| intern_author(sha, &mut self.commit_pool, &mut self.commit_index));
        true
    }

    fn record_legacy_timestamp(&mut self, line: &str) -> bool {
        let Ok(ts) = line.parse::<u64>() else {
            return false;
        };

        self.current_timestamp = Some(ts);
        self.current_committed_at = Some(ts);
        self.current_author_idx = None;
        self.current_commit_idx = None;
        true
    }

    fn record_numstat_bytes(&mut self, record: &[u8]) {
        let Some(first_tab) = record.iter().position(|byte| *byte == b'\t') else {
            return;
        };
        let Some(second_tab_offset) = record[first_tab + 1..]
            .iter()
            .position(|byte| *byte == b'\t')
        else {
            return;
        };
        let second_tab = first_tab + 1 + second_tab_offset;
        let Some(added) = std::str::from_utf8(&record[..first_tab])
            .ok()
            .and_then(|value| value.parse().ok())
        else {
            return;
        };
        let Some(deleted) = std::str::from_utf8(&record[first_tab + 1..second_tab])
            .ok()
            .and_then(|value| value.parse().ok())
        else {
            return;
        };
        let path = git_path_from_bytes(&record[second_tab + 1..]);
        self.record_numstat_path(added, deleted, path);
    }

    fn record_numstat_path(&mut self, added: u32, deleted: u32, path: PathBuf) {
        let ts = self.current_timestamp.unwrap_or(self.now_secs);
        self.files
            .entry(self.root.join(path))
            .or_insert_with(|| FileEvents { events: Vec::new() })
            .events
            .push(CachedCommitEvent {
                timestamp: ts,
                committed_at: self.current_committed_at.unwrap_or(ts),
                lines_added: added,
                lines_deleted: deleted,
                author_idx: self.current_author_idx,
                commit_idx: self.current_commit_idx,
            });
    }

    fn finish(self) -> ChurnEventState {
        ChurnEventState {
            files: self.files,
            author_pool: self.author_pool,
            commit_pool: self.commit_pool,
            git_log_bytes: 0,
        }
    }
}

/// True for a full SHA-1 (40) or SHA-256 (64) hexadecimal commit id.
fn is_commit_id(value: &str) -> bool {
    matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}

/// Aggregate one file's raw commit events into a [`FileChurn`], applying
/// recency weighting, trend detection, and per-author accumulation.
#[expect(
    clippy::cast_possible_truncation,
    reason = "commit count per file is bounded by git history depth"
)]
fn aggregate_file_churn(path: PathBuf, file: FileEvents, now_secs: u64) -> FileChurn {
    let mut timestamps = Vec::with_capacity(file.events.len());
    let mut weighted_commits = 0.0;
    let mut lines_added = 0_u32;
    let mut lines_deleted = 0_u32;
    let mut authors: FxHashMap<u32, AuthorContribution> = FxHashMap::default();

    for event in file.events {
        timestamps.push(event.timestamp);
        let age_days = (now_secs.saturating_sub(event.timestamp)) as f64 / SECS_PER_DAY;
        let weight = 0.5_f64.powf(age_days / HALF_LIFE_DAYS);
        weighted_commits += weight;
        lines_added = lines_added.saturating_add(event.lines_added);
        lines_deleted = lines_deleted.saturating_add(event.lines_deleted);
        accumulate_author(&mut authors, event.author_idx, weight, event.timestamp);
    }

    let commits = timestamps.len() as u32;
    let trend = compute_trend(&timestamps);
    for c in authors.values_mut() {
        c.weighted_commits = (c.weighted_commits * 100.0).round() / 100.0;
    }
    FileChurn {
        path,
        commits,
        weighted_commits: (weighted_commits * 100.0).round() / 100.0,
        lines_added,
        lines_deleted,
        trend,
        authors,
    }
}

/// Fold a single commit's author contribution into the per-author map.
fn accumulate_author(
    authors: &mut FxHashMap<u32, AuthorContribution>,
    author_idx: Option<u32>,
    weight: f64,
    timestamp: u64,
) {
    let Some(idx) = author_idx else {
        return;
    };
    authors
        .entry(idx)
        .and_modify(|c| {
            c.commits += 1;
            c.weighted_commits += weight;
            c.first_commit_ts = c.first_commit_ts.min(timestamp);
            c.last_commit_ts = c.last_commit_ts.max(timestamp);
        })
        .or_insert(AuthorContribution {
            commits: 1,
            weighted_commits: weight,
            first_commit_ts: timestamp,
            last_commit_ts: timestamp,
        });
}

/// Convert event-level churn state into the public aggregate result.
///
/// Recency weighting is measured against `clock`, not the system clock, so the
/// same commit yields the same `weighted_commits` on every run.
fn build_churn_result(
    state: ChurnEventState,
    shallow_clone: bool,
    clock: crate::clock::AnalysisClock,
) -> ChurnResult {
    build_windowed_churn_result(state, None, shallow_clone, clock)
}

/// Like [`build_churn_result`], but the per-file aggregates only count events
/// at or after `files_cutoff`. The commit list keeps the full state.
fn build_windowed_churn_result(
    mut state: ChurnEventState,
    files_cutoff: Option<u64>,
    shallow_clone: bool,
    clock: crate::clock::AnalysisClock,
) -> ChurnResult {
    let commits = collect_churn_commits(&state);
    state.retain_window(files_cutoff);
    let now_secs = clock.epoch_secs();

    let files = state
        .files
        .into_iter()
        .map(|(path, file)| {
            let churn = aggregate_file_churn(path.clone(), file, now_secs);
            (path, churn)
        })
        .collect();

    ChurnResult {
        files,
        shallow_clone,
        author_pool: state.author_pool,
        clock,
        git_log_bytes: state.git_log_bytes,
        commits,
    }
}

/// Group the events of `state` by commit id. Each commit lists its files
/// sorted by path, and the commits are sorted by committer time, then id.
fn collect_churn_commits(state: &ChurnEventState) -> Vec<ChurnCommit> {
    let mut by_idx: FxHashMap<u32, ChurnCommit> = FxHashMap::default();
    for (path, file) in &state.files {
        for event in &file.events {
            let Some(idx) = event.commit_idx else {
                continue;
            };
            let Some(sha) = state.commit_pool.get(idx as usize) else {
                continue;
            };
            by_idx
                .entry(idx)
                .or_insert_with(|| ChurnCommit {
                    sha: sha.clone(),
                    committed_at: event.committed_at,
                    files: Vec::new(),
                })
                .files
                .push(ChurnCommitFile {
                    path: path.clone(),
                    lines_changed: event.lines_added.saturating_add(event.lines_deleted),
                });
        }
    }
    let mut commits: Vec<ChurnCommit> = by_idx
        .into_values()
        .map(|mut commit| {
            commit.files.sort_by(|a, b| a.path.cmp(&b.path));
            commit.files.dedup_by(|later, kept| {
                if later.path != kept.path {
                    return false;
                }
                kept.lines_changed = kept.lines_changed.saturating_add(later.lines_changed);
                true
            });
            commit
        })
        .collect();
    commits.sort_by(|a, b| {
        a.committed_at
            .cmp(&b.committed_at)
            .then_with(|| a.sha.cmp(&b.sha))
    });
    commits
}

/// Intern an author email into the pool, returning its stable index.
fn intern_author(email: &str, pool: &mut Vec<String>, index: &mut FxHashMap<String, u32>) -> u32 {
    if let Some(&idx) = index.get(email) {
        return idx;
    }
    #[expect(
        clippy::cast_possible_truncation,
        reason = "author count is bounded by git history; u32 is far above any realistic ceiling"
    )]
    let idx = pool.len() as u32;
    let owned = email.to_string();
    index.insert(owned.clone(), idx);
    pool.push(owned);
    idx
}

/// Compute churn trend by splitting commits into two temporal halves.
///
/// Finds the midpoint between the oldest and newest commit timestamps,
/// then compares commit counts in each half:
/// - Recent > 1.5× older → Accelerating
/// - Recent < 0.67× older → Cooling
/// - Otherwise → Stable
fn compute_trend(timestamps: &[u64]) -> ChurnTrend {
    if timestamps.len() < 2 {
        return ChurnTrend::Stable;
    }

    let min_ts = timestamps.iter().copied().min().unwrap_or(0);
    let max_ts = timestamps.iter().copied().max().unwrap_or(0);

    if max_ts == min_ts {
        return ChurnTrend::Stable;
    }

    let midpoint = min_ts + (max_ts - min_ts) / 2;
    let recent = timestamps.iter().filter(|&&ts| ts > midpoint).count() as f64;
    let older = timestamps.iter().filter(|&&ts| ts <= midpoint).count() as f64;

    if older < 1.0 {
        return ChurnTrend::Stable;
    }

    let ratio = recent / older;
    if ratio > 1.5 {
        ChurnTrend::Accelerating
    } else if ratio < 0.67 {
        ChurnTrend::Cooling
    } else {
        ChurnTrend::Stable
    }
}

fn is_iso_date(input: &str) -> bool {
    input.len() == 10
        && input.as_bytes().get(4) == Some(&b'-')
        && input.as_bytes().get(7) == Some(&b'-')
        && input[..4].bytes().all(|b| b.is_ascii_digit())
        && input[5..7].bytes().all(|b| b.is_ascii_digit())
        && input[8..10].bytes().all(|b| b.is_ascii_digit())
}

fn split_number_unit(input: &str) -> Result<(&str, &str), String> {
    let pos = input.find(|c: char| !c.is_ascii_digit()).ok_or_else(|| {
        format!("--since requires a unit suffix (e.g., 6m, 90d, 1y), got: {input}")
    })?;
    if pos == 0 {
        return Err(format!(
            "--since must start with a number (e.g., 6m, 90d, 1y), got: {input}"
        ));
    }
    Ok((&input[..pos], &input[pos..]))
}

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

    #[test]
    fn parse_since_months_short() {
        let d = parse_since("6m").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 6,
                unit: ChurnWindowUnit::Months
            }
        );
        assert_eq!(d.display, "6 months");
    }

    #[test]
    fn parse_since_months_long() {
        let d = parse_since("6months").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 6,
                unit: ChurnWindowUnit::Months
            }
        );
        assert_eq!(d.display, "6 months");
    }

    #[test]
    fn parse_since_days() {
        let d = parse_since("90d").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 90,
                unit: ChurnWindowUnit::Days
            }
        );
        assert_eq!(d.display, "90 days");
    }

    #[test]
    fn parse_since_year_singular() {
        let d = parse_since("1y").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 1,
                unit: ChurnWindowUnit::Years
            }
        );
        assert_eq!(d.display, "1 year");
    }

    #[test]
    fn parse_since_years_plural() {
        let d = parse_since("2years").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 2,
                unit: ChurnWindowUnit::Years
            }
        );
        assert_eq!(d.display, "2 years");
    }

    #[test]
    fn parse_since_weeks() {
        let d = parse_since("2w").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 2,
                unit: ChurnWindowUnit::Weeks
            }
        );
        assert_eq!(d.display, "2 weeks");
    }

    #[test]
    fn parse_since_iso_date() {
        let d = parse_since("2025-06-01").unwrap();
        assert_eq!(d.window, ChurnWindow::Date("2025-06-01".to_string()));
        assert_eq!(d.display, "2025-06-01");
    }

    #[test]
    fn parse_since_month_singular() {
        let d = parse_since("1month").unwrap();
        assert_eq!(d.display, "1 month");
    }

    #[test]
    fn parse_since_day_singular() {
        let d = parse_since("1day").unwrap();
        assert_eq!(d.display, "1 day");
    }

    #[test]
    fn parse_since_zero_rejected() {
        assert!(parse_since("0m").is_err());
    }

    #[test]
    fn parse_since_no_unit_rejected() {
        assert!(parse_since("90").is_err());
    }

    #[test]
    fn parse_since_unknown_unit_rejected() {
        assert!(parse_since("6x").is_err());
    }

    #[test]
    fn parse_since_no_number_rejected() {
        assert!(parse_since("months").is_err());
    }

    #[test]
    fn trend_empty_is_stable() {
        assert_eq!(compute_trend(&[]), ChurnTrend::Stable);
    }

    #[test]
    fn trend_single_commit_is_stable() {
        assert_eq!(compute_trend(&[100]), ChurnTrend::Stable);
    }

    #[test]
    fn trend_accelerating() {
        let timestamps = vec![100, 200, 800, 850, 900, 950, 1000];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Accelerating);
    }

    #[test]
    fn trend_cooling() {
        let timestamps = vec![100, 150, 200, 250, 300, 900, 1000];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Cooling);
    }

    #[test]
    fn trend_stable_even_distribution() {
        let timestamps = vec![100, 200, 300, 700, 800, 900];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Stable);
    }

    #[test]
    fn trend_same_timestamp_is_stable() {
        let timestamps = vec![500, 500, 500];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Stable);
    }

    #[test]
    fn iso_date_valid() {
        assert!(is_iso_date("2025-06-01"));
        assert!(is_iso_date("2025-12-31"));
    }

    #[test]
    fn iso_date_with_time_rejected() {
        assert!(!is_iso_date("2025-06-01T00:00:00"));
    }

    #[test]
    fn iso_date_invalid() {
        assert!(!is_iso_date("6months"));
        assert!(!is_iso_date("2025"));
        assert!(!is_iso_date("not-a-date"));
        assert!(!is_iso_date("abcd-ef-gh"));
    }

    #[test]
    fn trend_display() {
        assert_eq!(ChurnTrend::Accelerating.to_string(), "accelerating");
        assert_eq!(ChurnTrend::Stable.to_string(), "stable");
        assert_eq!(ChurnTrend::Cooling.to_string(), "cooling");
    }

    #[test]
    fn trend_boundary_1_5x_ratio() {
        let timestamps = vec![100, 200, 600, 800, 1000];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Stable);
    }

    #[test]
    fn trend_just_above_1_5x() {
        let timestamps = vec![100, 600, 800, 1000];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Accelerating);
    }

    #[test]
    fn trend_boundary_0_67x_ratio() {
        let timestamps = vec![100, 200, 300, 600, 1000];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Cooling);
    }

    #[test]
    fn trend_two_timestamps_different() {
        let timestamps = vec![100, 200];
        assert_eq!(compute_trend(&timestamps), ChurnTrend::Stable);
    }

    #[test]
    fn parse_since_week_singular() {
        let d = parse_since("1week").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 1,
                unit: ChurnWindowUnit::Weeks
            }
        );
        assert_eq!(d.display, "1 week");
    }

    #[test]
    fn parse_since_weeks_long() {
        let d = parse_since("3weeks").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 3,
                unit: ChurnWindowUnit::Weeks
            }
        );
        assert_eq!(d.display, "3 weeks");
    }

    #[test]
    fn parse_since_days_long() {
        let d = parse_since("30days").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 30,
                unit: ChurnWindowUnit::Days
            }
        );
        assert_eq!(d.display, "30 days");
    }

    #[test]
    fn parse_since_year_long() {
        let d = parse_since("1year").unwrap();
        assert_eq!(
            d.window,
            ChurnWindow::Relative {
                count: 1,
                unit: ChurnWindowUnit::Years
            }
        );
        assert_eq!(d.display, "1 year");
    }

    #[test]
    fn parse_since_overflow_number_rejected() {
        let result = parse_since("99999999999999999999d");
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(err.contains("invalid number"));
    }

    #[test]
    fn parse_since_zero_days_rejected() {
        assert!(parse_since("0d").is_err());
    }

    #[test]
    fn parse_since_zero_weeks_rejected() {
        assert!(parse_since("0w").is_err());
    }

    #[test]
    fn parse_since_zero_years_rejected() {
        assert!(parse_since("0y").is_err());
    }

    #[test]
    fn iso_date_wrong_separator_positions() {
        assert!(!is_iso_date("20-25-0601"));
        assert!(!is_iso_date("202506-01-"));
    }

    #[test]
    fn iso_date_too_short() {
        assert!(!is_iso_date("2025-06-0"));
    }

    #[test]
    fn iso_date_letters_in_day() {
        assert!(!is_iso_date("2025-06-ab"));
    }

    #[test]
    fn iso_date_letters_in_month() {
        assert!(!is_iso_date("2025-ab-01"));
    }

    #[test]
    fn split_number_unit_valid() {
        let (num, unit) = split_number_unit("42days").unwrap();
        assert_eq!(num, "42");
        assert_eq!(unit, "days");
    }

    #[test]
    fn split_number_unit_single_digit() {
        let (num, unit) = split_number_unit("1m").unwrap();
        assert_eq!(num, "1");
        assert_eq!(unit, "m");
    }

    #[test]
    fn split_number_unit_no_digits() {
        let err = split_number_unit("abc").unwrap_err();
        assert!(err.contains("must start with a number"));
    }

    #[test]
    fn split_number_unit_no_unit() {
        let err = split_number_unit("123").unwrap_err();
        assert!(err.contains("requires a unit suffix"));
    }

    #[test]
    fn trend_serde_serialization() {
        assert_eq!(
            serde_json::to_string(&ChurnTrend::Accelerating).unwrap(),
            "\"accelerating\""
        );
        assert_eq!(
            serde_json::to_string(&ChurnTrend::Stable).unwrap(),
            "\"stable\""
        );
        assert_eq!(
            serde_json::to_string(&ChurnTrend::Cooling).unwrap(),
            "\"cooling\""
        );
    }

    #[test]
    fn intern_author_returns_existing_index() {
        let mut pool = Vec::new();
        let mut index = FxHashMap::default();
        let i1 = intern_author("alice@x", &mut pool, &mut index);
        let i2 = intern_author("alice@x", &mut pool, &mut index);
        assert_eq!(i1, i2);
        assert_eq!(pool.len(), 1);
    }

    #[test]
    fn intern_author_assigns_sequential_indices() {
        let mut pool = Vec::new();
        let mut index = FxHashMap::default();
        assert_eq!(intern_author("alice@x", &mut pool, &mut index), 0);
        assert_eq!(intern_author("bob@x", &mut pool, &mut index), 1);
        assert_eq!(intern_author("carol@x", &mut pool, &mut index), 2);
        assert_eq!(intern_author("alice@x", &mut pool, &mut index), 0);
    }

    fn git(root: &Path, args: &[&str]) {
        let status = super::git_command()
            .args(args)
            .current_dir(root)
            .status()
            .expect("run git");
        assert!(status.success(), "git {args:?} failed");
    }

    fn write(root: &Path, path: &str, contents: &str) {
        let path = root.join(path);
        std::fs::create_dir_all(path.parent().expect("test path has parent")).unwrap();
        std::fs::write(path, contents).unwrap();
    }

    #[cfg(unix)]
    #[test]
    fn churn_preserves_special_filenames() {
        let repo = tempfile::tempdir().expect("create repo");
        let root = repo.path();
        git(root, &["init", "--quiet"]);
        git(root, &["config", "user.email", "churn@example.test"]);
        git(root, &["config", "user.name", "Churn Test"]);
        git(root, &["config", "commit.gpgsign", "false"]);

        let special_files = [
            "src/line\nbreak.ts",
            "src/space name.ts",
            "src/quote\"name.ts",
            "src/back\\slash.ts",
            "src/unicode-λ.ts",
        ]
        .map(|path| root.join(path));
        std::fs::create_dir_all(root.join("src")).expect("source dir");
        for special in &special_files {
            std::fs::write(special, "export const value = 1;\n").expect("special fixture");
        }
        git(root, &["add", "."]);
        git(root, &["commit", "--quiet", "-m", "initial"]);

        let since = parse_since("1y").expect("valid duration");
        let churn = analyze_churn(root, &since).expect("churn result");
        for special in special_files {
            assert!(
                churn.files.contains_key(&special),
                "missing {special:?}: {:?}",
                churn.files.keys()
            );
        }
    }

    #[cfg(unix)]
    #[test]
    fn churn_cache_preserves_non_utf8_filenames() {
        use std::ffi::OsString;
        use std::os::unix::ffi::OsStringExt;

        let invalid_path = PathBuf::from(OsString::from_vec(b"src/non-utf8-\xff.ts".to_vec()));
        let mut files = FxHashMap::default();
        files.insert(
            invalid_path.clone(),
            FileEvents {
                events: vec![CachedCommitEvent {
                    timestamp: 1,
                    committed_at: 1,
                    lines_added: 2,
                    lines_deleted: 1,
                    author_idx: None,
                    commit_idx: None,
                }],
            },
        );
        let state = ChurnEventState {
            files,
            author_pool: Vec::new(),
            commit_pool: Vec::new(),
            git_log_bytes: 0,
        };
        let cache_dir = tempfile::tempdir().expect("cache directory");
        save_churn_cache(cache_dir.path(), "abc123", "1y", &state, false);
        let warm = load_churn_cache(cache_dir.path(), "1y")
            .expect("warm churn cache")
            .into_event_state(None);

        assert!(warm.files.contains_key(&invalid_path));
    }

    #[test]
    fn churn_cache_rejects_pre_lossless_path_encoding_version() {
        let cache_dir = tempfile::tempdir().expect("cache directory");
        let cache = ChurnCache {
            version: 4,
            last_indexed_sha: "abc123".to_string(),
            window_token: "1y".to_string(),
            files: vec![CachedFileChurn {
                path: br#"/project/\"src/line\\nbreak.ts\""#.to_vec(),
                events: Vec::new(),
            }],
            shallow_clone: false,
            author_pool: Vec::new(),
            commit_pool: Vec::new(),
        };
        std::fs::write(cache_dir.path().join("churn.bin"), bitcode::encode(&cache))
            .expect("cache fixture");

        assert!(load_churn_cache(cache_dir.path(), "1y").is_none());
    }

    #[test]
    fn churn_cache_rejects_entries_without_commit_ids() {
        let cache_dir = tempfile::tempdir().expect("cache directory");
        let cache = ChurnCache {
            version: 7,
            last_indexed_sha: "abc123".to_string(),
            window_token: "1y".to_string(),
            files: Vec::new(),
            shallow_clone: false,
            author_pool: Vec::new(),
            commit_pool: Vec::new(),
        };
        std::fs::write(cache_dir.path().join("churn.bin"), bitcode::encode(&cache))
            .expect("cache fixture");

        assert!(load_churn_cache(cache_dir.path(), "1y").is_none());
    }

    #[test]
    fn commit_header_with_sha_records_the_commit_id() {
        let sha = "0123456789abcdef0123456789abcdef01234567";
        let header = format!("1700000000|1700000001|{sha}|alice@corp");
        let state = parse_git_log_events_z(
            &git_log_z(&[(header.as_str(), &["1\t2\tsrc/a.ts", "3\t0\tsrc/b.ts"])]),
            Path::new("/project"),
            1_700_000_100,
        );
        let result = build_churn_result(
            state,
            false,
            crate::clock::AnalysisClock::pinned(1_700_000_100),
        );

        assert_eq!(result.author_pool, vec!["alice@corp".to_string()]);
        assert_eq!(result.commits.len(), 1);
        let commit = &result.commits[0];
        assert_eq!(commit.sha, sha);
        assert_eq!(commit.committed_at, 1_700_000_001);
        let files: Vec<(PathBuf, u32)> = commit
            .files
            .iter()
            .map(|file| (file.path.clone(), file.lines_changed))
            .collect();
        assert_eq!(
            files,
            vec![
                (PathBuf::from("/project/src/a.ts"), 3),
                (PathBuf::from("/project/src/b.ts"), 3),
            ]
        );
    }

    #[test]
    fn commit_header_without_sha_keeps_the_email_and_no_commit() {
        let state = parse_git_log_events_z(
            &git_log_z(&[("1700000000|1700000000|alice@corp", &["1\t2\tsrc/a.ts"])]),
            Path::new("/project"),
            1_700_000_100,
        );
        let result = build_churn_result(
            state,
            false,
            crate::clock::AnalysisClock::pinned(1_700_000_100),
        );

        assert_eq!(result.author_pool, vec!["alice@corp".to_string()]);
        assert!(result.commits.is_empty());
        assert_eq!(result.files.len(), 1);
    }

    #[test]
    fn churn_file_commit_field_groups_events_into_commits() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{
              "schema": "fallow-churn/v1",
              "events": [
                { "commit": "c1", "path": "src/a.ts", "timestamp": 1700000000, "added": 1, "deleted": 0 },
                { "commit": "c1", "path": "src/b.ts", "timestamp": 1700000000, "added": 2, "deleted": 1 },
                { "commit": "c2", "path": "src/a.ts", "timestamp": 1700100000, "added": 4, "deleted": 0 },
                { "path": "src/c.ts", "timestamp": 1700100000, "added": 1, "deleted": 0 }
              ]
            }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();

        let commits: Vec<(String, usize)> = result
            .commits
            .iter()
            .map(|commit| (commit.sha.clone(), commit.files.len()))
            .collect();
        assert_eq!(commits, vec![("c1".to_string(), 2), ("c2".to_string(), 1)]);
        assert_eq!(
            result.files.len(),
            3,
            "an event without commit still counts"
        );
    }

    #[test]
    fn cached_churn_merges_new_commits_after_head_advances() {
        let repo = tempfile::tempdir().expect("create repo");
        let root = repo.path();
        git(root, &["init"]);
        git(root, &["config", "user.email", "churn@example.test"]);
        git(root, &["config", "user.name", "Churn Test"]);
        git(root, &["config", "commit.gpgsign", "false"]);

        write(root, "src/a.ts", "export const a = 1;\n");
        git(root, &["add", "."]);
        git(root, &["commit", "-m", "initial"]);

        let since = parse_since("1y").unwrap();
        let cache = tempfile::tempdir().expect("create cache dir");
        let (cold, cold_hit) = analyze_churn_cached(root, &since, cache.path(), false).unwrap();
        assert!(!cold_hit);
        let file = root.join("src/a.ts");
        assert_eq!(cold.files[&file].commits, 1);

        let (_warm, warm_hit) = analyze_churn_cached(root, &since, cache.path(), false).unwrap();
        assert!(warm_hit);

        write(
            root,
            "src/a.ts",
            "export const a = 1;\nexport const b = 2;\n",
        );
        git(root, &["add", "."]);
        git(root, &["commit", "-m", "update a"]);
        let head = get_head_sha(root).unwrap();

        let (incremental, incremental_hit) =
            analyze_churn_cached(root, &since, cache.path(), false).unwrap();
        assert!(incremental_hit);
        assert_eq!(incremental.files[&file].commits, 2);

        let cache = load_churn_cache(cache.path(), &since.window.cache_token()).unwrap();
        assert_eq!(cache.last_indexed_sha, head);
    }

    fn write_churn_file(dir: &std::path::Path, contents: &str) -> PathBuf {
        let path = dir.join("churn.json");
        std::fs::write(&path, contents).unwrap();
        path
    }

    #[test]
    fn churn_file_happy_path() {
        let dir = tempfile::tempdir().unwrap();
        let root = Path::new("/project");
        let path = write_churn_file(
            dir.path(),
            r#"{
              "schema": "fallow-churn/v1",
              "events": [
                { "path": "src/a.ts", "timestamp": 1700000000, "author": "alice@corp", "added": 10, "deleted": 5 },
                { "path": "src/a.ts", "timestamp": 1700100000, "author": "bob@corp", "added": 3, "deleted": 2 }
              ]
            }"#,
        );
        let result = analyze_churn_from_file(&path, root).unwrap();
        let churn = &result.files[&PathBuf::from("/project/src/a.ts")];
        assert_eq!(churn.commits, 2);
        assert_eq!(churn.lines_added, 13);
        assert_eq!(churn.lines_deleted, 7);
        assert_eq!(churn.authors.len(), 2);
        assert!(result.author_pool.contains(&"alice@corp".to_string()));
        assert!(result.author_pool.contains(&"bob@corp".to_string()));
        assert!(!result.shallow_clone);
    }

    #[test]
    fn churn_file_empty_events_is_valid() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        assert!(result.files.is_empty());
        assert!(result.author_pool.is_empty());
    }

    #[test]
    fn churn_file_missing_events_key_is_valid() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(dir.path(), r#"{ "schema": "fallow-churn/v1" }"#);
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        assert!(result.files.is_empty());
    }

    #[test]
    fn churn_file_bad_schema_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v2", "events": [] }"#,
        );
        let err = analyze_churn_from_file(&path, Path::new("/project")).unwrap_err();
        assert!(err.contains("expected \"fallow-churn/v1\""), "{err}");
    }

    #[test]
    fn churn_file_malformed_json_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(dir.path(), "{ not json");
        assert!(analyze_churn_from_file(&path, Path::new("/project")).is_err());
    }

    #[test]
    fn churn_file_missing_file_rejected() {
        let err = analyze_churn_from_file(Path::new("/no/such/churn.json"), Path::new("/project"))
            .unwrap_err();
        assert!(err.contains("failed to read churn file"), "{err}");
    }

    #[test]
    fn churn_file_reader_accepts_exact_limit() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(dir.path(), "12345678");
        assert_eq!(read_churn_file_with_limit(&path, 8).unwrap(), "12345678");
    }

    #[test]
    fn churn_file_reader_rejects_limit_plus_one() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(dir.path(), "123456789");
        let err = read_churn_file_with_limit(&path, 8).unwrap_err();
        assert!(err.contains("at least 9 bytes"), "{err}");
        assert!(err.contains("8 byte limit"), "{err}");
    }

    #[test]
    fn churn_file_empty_path_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [ { "path": "  ", "timestamp": 1700000000, "added": 1, "deleted": 0 } ] }"#,
        );
        let err = analyze_churn_from_file(&path, Path::new("/project")).unwrap_err();
        assert!(err.contains("empty path"), "{err}");
    }

    #[test]
    fn churn_file_rejects_non_relative_paths() {
        let invalid = [
            "/tmp/a.ts",
            r"C:\tmp\a.ts",
            "../a.ts",
            "src/../../a.ts",
            "./src/a.ts",
            "//server/share/a.ts",
        ];
        for event_path in invalid {
            let dir = tempfile::tempdir().unwrap();
            let body = format!(
                r#"{{ "schema": "fallow-churn/v1", "events": [ {{ "path": {event_path:?}, "timestamp": 1700000000, "added": 1, "deleted": 0 }} ] }}"#
            );
            let path = write_churn_file(dir.path(), &body);
            let err = analyze_churn_from_file(&path, Path::new("/project")).unwrap_err();
            assert!(err.contains(event_path), "{event_path}: {err}");
            assert!(err.contains("repo-relative"), "{event_path}: {err}");
        }
    }

    #[test]
    fn churn_file_accepts_unicode_and_spaces_in_path_components() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [ { "path": "src/ruimte map/naïef.ts", "timestamp": 1700000000, "added": 1, "deleted": 0 } ] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        assert!(
            result
                .files
                .contains_key(&PathBuf::from("/project/src/ruimte map/naïef.ts"))
        );
    }

    #[test]
    fn churn_file_millisecond_timestamp_rejected() {
        let dir = tempfile::tempdir().unwrap();
        // 1700000000000 is milliseconds; ~52000 years in the future as seconds.
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [ { "path": "src/a.ts", "timestamp": 1700000000000, "added": 1, "deleted": 0 } ] }"#,
        );
        let err = analyze_churn_from_file(&path, Path::new("/project")).unwrap_err();
        assert!(err.contains("milliseconds"), "{err}");
    }

    #[test]
    fn churn_file_missing_author_contributes_no_signal() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [ { "path": "src/a.ts", "timestamp": 1700000000, "added": 1, "deleted": 0 } ] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        let churn = &result.files[&PathBuf::from("/project/src/a.ts")];
        assert_eq!(churn.commits, 1);
        assert!(churn.authors.is_empty());
        assert!(result.author_pool.is_empty());
    }

    #[test]
    fn churn_file_empty_author_string_treated_as_absent() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [ { "path": "src/a.ts", "timestamp": 1700000000, "author": "  ", "added": 1, "deleted": 0 } ] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        assert!(result.author_pool.is_empty());
    }

    #[test]
    fn churn_file_unknown_fields_ignored() {
        // Extra keys (including the reserved `commit`) are accepted and ignored,
        // so a wrapper carrying extra metadata stays forward-compatible.
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "extra": true, "events": [ { "path": "src/a.ts", "timestamp": 1700000000, "author": "alice@corp", "added": 1, "deleted": 0, "commit": "abc123", "tz": "+0200" } ] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        assert_eq!(result.files[&PathBuf::from("/project/src/a.ts")].commits, 1);
    }

    #[test]
    fn churn_file_backslash_paths_normalized() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [ { "path": "src\\a.ts", "timestamp": 1700000000, "added": 1, "deleted": 0 } ] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        assert!(
            result
                .files
                .contains_key(&PathBuf::from("/project/src/a.ts"))
        );
    }

    #[test]
    fn churn_file_rejects_line_totals_above_u32() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [
                { "path": "src/a.ts", "timestamp": 1700000000, "added": 4294967295, "deleted": 0 },
                { "path": "src/a.ts", "timestamp": 1700000001, "added": 1, "deleted": 0 }
            ] }"#,
        );
        let err = analyze_churn_from_file(&path, Path::new("/project")).unwrap_err();
        assert!(err.contains("exceeding the u32 limit"), "{err}");
        assert!(err.contains("src/a.ts"), "{err}");
    }

    #[test]
    fn churn_file_accepts_line_totals_at_u32_max() {
        let dir = tempfile::tempdir().unwrap();
        let path = write_churn_file(
            dir.path(),
            r#"{ "schema": "fallow-churn/v1", "events": [
                { "path": "src/a.ts", "timestamp": 1700000000, "added": 4294967294, "deleted": 4294967295 },
                { "path": "src/a.ts", "timestamp": 1700000001, "added": 1, "deleted": 0 }
            ] }"#,
        );
        let result = analyze_churn_from_file(&path, Path::new("/project")).unwrap();
        let churn = &result.files[&PathBuf::from("/project/src/a.ts")];
        assert_eq!(churn.lines_added, u32::MAX);
        assert_eq!(churn.lines_deleted, u32::MAX);
    }

    /// A numstat row that arrives before any commit header (truncated or
    /// malformed git output) must fall back to the run clock, so the recorded
    /// timestamp, and therefore the file's weighted commits and staleness, is
    /// the same on every run over one commit instead of moving with wall time.
    #[test]
    fn headerless_numstat_falls_back_to_the_run_clock() {
        let root = Path::new("/project");
        let pinned = 1_700_000_000;

        let state = parse_git_log_events_z(b"10\t5\tsrc/a.ts", root, pinned);

        let events = &state.files[&PathBuf::from("/project/src/a.ts")].events;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].timestamp, pinned);
        assert_eq!(events[0].committed_at, pinned);
    }

    /// The same truncated output parsed against two different run clocks must
    /// disagree only by those clocks: nothing in the fallback path may consult
    /// the system clock.
    #[test]
    fn headerless_numstat_tracks_only_the_supplied_clock() {
        let root = Path::new("/project");
        let record: &[u8] = b"1\t0\tsrc/a.ts";

        let early = parse_git_log_events_z(record, root, 1_600_000_000);
        let late = parse_git_log_events_z(record, root, 1_700_000_000);

        let key = PathBuf::from("/project/src/a.ts");
        assert_eq!(early.files[&key].events[0].timestamp, 1_600_000_000);
        assert_eq!(late.files[&key].events[0].timestamp, 1_700_000_000);
    }

    /// A commit header still wins over the fallback: the clock only fills a gap.
    #[test]
    fn commit_header_timestamp_beats_the_run_clock_fallback() {
        let root = Path::new("/project");
        let record = b"1700000000|1700000500|dev@example.com\x002\t1\tsrc/a.ts";

        let state = parse_git_log_events_z(record, root, 1_234_567_890);

        let events = &state.files[&PathBuf::from("/project/src/a.ts")].events;
        assert_eq!(events[0].timestamp, 1_700_000_000);
        assert_eq!(events[0].committed_at, 1_700_000_500);
    }

    /// The run clock for the ported `git log -z` scenarios below.
    const NOW: u64 = 1_750_000_000;
    const HEADER: &str = "1700000000|1700000000|dev@example.com";

    /// Build the bytes that `git log --numstat -z --format=format:%at|%ct|%H|%ae%x00`
    /// prints: each commit is `header NUL LF`, then one NUL-terminated numstat
    /// record per file, and consecutive commits are split by one more NUL.
    fn git_log_z(commits: &[(&str, &[&str])]) -> Vec<u8> {
        let mut out = Vec::new();
        for (index, (header, rows)) in commits.iter().enumerate() {
            if index > 0 {
                out.push(0);
            }
            out.extend_from_slice(header.as_bytes());
            out.extend_from_slice(b"\0\n");
            for row in *rows {
                out.extend_from_slice(row.as_bytes());
                out.push(0);
            }
        }
        out
    }

    fn churn_from_git_log_z(stdout: &[u8], root: &Path, now_secs: u64) -> ChurnResult {
        build_churn_result(
            parse_git_log_events_z(stdout, root, now_secs),
            false,
            crate::clock::AnalysisClock::pinned(now_secs),
        )
    }

    fn churn_from_commits(commits: &[(&str, &[&str])]) -> ChurnResult {
        churn_from_git_log_z(&git_log_z(commits), Path::new("/project"), NOW)
    }

    #[test]
    fn git_log_z_fixture_matches_real_git_output() {
        assert_eq!(
            git_log_z(&[
                ("1|2|a@b", &["1\t0\ta.ts"]),
                ("3|4|a@b", &["1\t0\ta.ts", "1\t0\tb c.ts"])
            ]),
            b"1|2|a@b\0\n1\t0\ta.ts\0\x003|4|a@b\0\n1\t0\ta.ts\x001\t0\tb c.ts\0".to_vec()
        );
    }

    #[test]
    fn numstat_records_parse_counts_and_skip_binary_or_malformed_rows() {
        let root = Path::new("/project");
        type Expected = Option<(u32, u32, &'static str)>;
        let cases: &[(&str, Expected)] = &[
            ("10\t5\tsrc/file.ts", Some((10, 5, "src/file.ts"))),
            ("0\t0\tsrc/empty.ts", Some((0, 0, "src/empty.ts"))),
            (
                "3\t1\tpath with spaces/file.ts",
                Some((3, 1, "path with spaces/file.ts")),
            ),
            ("1\t0\tweird\tname.ts", Some((1, 0, "weird\tname.ts"))),
            (
                "4294967295\t8888\tsrc/big.ts",
                Some((u32::MAX, 8888, "src/big.ts")),
            ),
            ("4294967296\t0\tsrc/overflow.ts", None),
            ("-\t-\tsrc/image.png", None),
            ("-\t5\tsrc/file.ts", None),
            ("10\t-\tsrc/file.ts", None),
            ("10\t5", None),
        ];
        for (row, expected) in cases {
            let state = parse_git_log_events_z(&git_log_z(&[(HEADER, &[row])]), root, NOW);
            let Some((added, deleted, path)) = expected else {
                assert!(state.files.is_empty(), "{row:?} must not record churn");
                continue;
            };
            assert_eq!(state.files.len(), 1, "{row:?}");
            let events = &state.files[&root.join(path)].events;
            assert_eq!(events.len(), 1, "{row:?}");
            assert_eq!(events[0].lines_added, *added, "{row:?}");
            assert_eq!(events[0].lines_deleted, *deleted, "{row:?}");
        }
    }

    #[test]
    fn git_log_z_aggregates_commits_and_lines_per_file() {
        let result = churn_from_commits(&[
            (
                HEADER,
                &["10\t5\tsrc/a.ts", "3\t1\tsrc/b.ts", "-\t-\timage.png"],
            ),
            ("1700100000|1700100000|dev@example.com", &["3\t2\tsrc/a.ts"]),
        ]);
        assert_eq!(result.files.len(), 2);
        let a = &result.files[&PathBuf::from("/project/src/a.ts")];
        assert_eq!((a.commits, a.lines_added, a.lines_deleted), (2, 13, 7));
        let b = &result.files[&PathBuf::from("/project/src/b.ts")];
        assert_eq!((b.commits, b.lines_added, b.lines_deleted), (1, 3, 1));
        assert!(
            !result
                .files
                .contains_key(&PathBuf::from("/project/image.png"))
        );
    }

    #[test]
    fn git_log_z_empty_or_blank_output_records_nothing() {
        assert!(
            churn_from_git_log_z(b"", Path::new("/project"), NOW)
                .files
                .is_empty()
        );
        assert!(
            churn_from_git_log_z(b"\0\0\n\0  \0", Path::new("/project"), NOW)
                .files
                .is_empty()
        );
    }

    #[test]
    fn git_log_z_blank_records_between_commits_are_ignored() {
        let stdout = b"  \0\n1700000000|1700000000|dev@example.com\0\n  \0\n10\t5\tsrc/a.ts\0\0";
        let result = churn_from_git_log_z(stdout, Path::new("/project"), NOW);
        assert_eq!(result.files.len(), 1);
        assert_eq!(result.files[&PathBuf::from("/project/src/a.ts")].commits, 1);
    }

    #[test]
    fn git_log_z_paths_are_joined_to_the_root() {
        let root = Path::new("/my/project");
        let result =
            churn_from_git_log_z(&git_log_z(&[(HEADER, &["1\t0\tlib/utils.ts"])]), root, NOW);
        let key = PathBuf::from("/my/project/lib/utils.ts");
        assert_eq!(result.files[&key].path, key);
    }

    #[test]
    fn git_log_z_weights_commits_by_age_against_the_run_clock() {
        let day = 86_400;
        let now = format!("{NOW}|{NOW}|dev@example.com");
        let half_life = format!("{0}|{0}|dev@example.com", NOW - 45 * day);
        let two_half_lives = format!("{0}|{0}|dev@example.com", NOW - 180 * day);
        let result = churn_from_commits(&[
            (&now, &["1\t0\tsrc/fresh.ts"]),
            (&half_life, &["1\t0\tsrc/half.ts"]),
            (&two_half_lives, &["1\t0\tsrc/old.ts"]),
        ]);
        let weight = |name: &str| {
            result.files[&PathBuf::from(format!("/project/src/{name}"))].weighted_commits
        };
        assert!((weight("fresh.ts") - 1.0).abs() < f64::EPSILON);
        // 0.5^(45/90) = 0.7071..., rounded to two decimals.
        assert!((weight("half.ts") - 0.71).abs() < f64::EPSILON);
        assert!((weight("old.ts") - 0.25).abs() < f64::EPSILON);
    }

    #[test]
    fn git_log_z_headerless_numstat_weighs_as_a_commit_at_the_run_clock() {
        let result = churn_from_git_log_z(b"10\t5\tsrc/no_ts.ts\0", Path::new("/project"), NOW);
        let churn = &result.files[&PathBuf::from("/project/src/no_ts.ts")];
        assert_eq!(
            (churn.commits, churn.lines_added, churn.lines_deleted),
            (1, 10, 5)
        );
        assert!((churn.weighted_commits - 1.0).abs() < f64::EPSILON);
    }

    #[test]
    fn git_log_z_trend_is_computed_per_file() {
        let result = churn_from_commits(&[
            (
                "1000|1000|dev@example.com",
                &["5\t1\tsrc/old.ts", "1\t0\tsrc/hot.ts"],
            ),
            ("1800|1800|dev@example.com", &["1\t0\tsrc/hot.ts"]),
            ("1900|1900|dev@example.com", &["1\t0\tsrc/hot.ts"]),
            ("1950|1950|dev@example.com", &["1\t0\tsrc/hot.ts"]),
            (
                "2000|2000|dev@example.com",
                &["3\t1\tsrc/old.ts", "1\t0\tsrc/hot.ts"],
            ),
        ]);
        let old = &result.files[&PathBuf::from("/project/src/old.ts")];
        let hot = &result.files[&PathBuf::from("/project/src/hot.ts")];
        assert_eq!(old.commits, 2);
        assert_eq!(old.trend, ChurnTrend::Stable);
        assert_eq!(hot.commits, 5);
        assert_eq!(hot.trend, ChurnTrend::Accelerating);
    }

    #[test]
    fn git_log_z_interns_authors_and_aggregates_per_author() {
        let result = churn_from_commits(&[
            (
                "1700000000|1700000000|alice@example.com",
                &["1\t0\tsrc/index.ts"],
            ),
            (
                "1700100000|1700100000|bob@example.com",
                &["2\t0\tsrc/index.ts"],
            ),
            (
                "1700200000|1700200000|alice@example.com",
                &["1\t1\tsrc/index.ts"],
            ),
        ]);
        assert_eq!(
            result.author_pool,
            vec![
                "alice@example.com".to_string(),
                "bob@example.com".to_string()
            ]
        );
        let churn = &result.files[&PathBuf::from("/project/src/index.ts")];
        assert_eq!(churn.commits, 3);
        assert_eq!(churn.authors.len(), 2);
        let alice = &churn.authors[&0];
        assert_eq!(alice.commits, 2);
        assert_eq!(alice.first_commit_ts, 1_700_000_000);
        assert_eq!(alice.last_commit_ts, 1_700_200_000);
        let bob = &churn.authors[&1];
        assert_eq!(bob.commits, 1);
        assert_eq!(bob.first_commit_ts, 1_700_100_000);
    }

    #[test]
    fn git_log_z_legacy_headers_still_parse() {
        let result = churn_from_commits(&[
            ("1700000000", &["10\t5\tsrc/bare.ts"]),
            ("1700100000|alice@example.com", &["1\t0\tsrc/two_field.ts"]),
        ]);
        assert_eq!(result.author_pool, vec!["alice@example.com".to_string()]);
        let bare = &result.files[&PathBuf::from("/project/src/bare.ts")];
        assert_eq!(bare.commits, 1);
        assert!(bare.authors.is_empty());
        let two_field = &result.files[&PathBuf::from("/project/src/two_field.ts")];
        assert_eq!(two_field.authors[&0].first_commit_ts, 1_700_100_000);
    }

    #[test]
    fn churn_file_matches_git_log_z_parse() {
        // The same events fed via git numstat and via the JSON import must
        // produce identical aggregate churn: the import reuses
        // build_churn_result, so only the SOURCE differs.
        let dir = tempfile::tempdir().unwrap();
        let root = Path::new("/project");
        let path = write_churn_file(
            dir.path(),
            r#"{
              "schema": "fallow-churn/v1",
              "events": [
                { "path": "src/a.ts", "timestamp": 1700000000, "author": "alice@corp", "added": 10, "deleted": 5 },
                { "path": "src/b.ts", "timestamp": 1700000000, "author": "alice@corp", "added": 3, "deleted": 1 },
                { "path": "src/a.ts", "timestamp": 1700100000, "author": "bob@corp", "added": 3, "deleted": 2 }
              ]
            }"#,
        );
        let imported = analyze_churn_from_file(&path, root).unwrap();
        let git = churn_from_git_log_z(
            &git_log_z(&[
                (
                    "1700000000|1700000000|alice@corp",
                    &["10\t5\tsrc/a.ts", "3\t1\tsrc/b.ts"],
                ),
                ("1700100000|1700100000|bob@corp", &["3\t2\tsrc/a.ts"]),
            ]),
            root,
            imported.clock.epoch_secs(),
        );

        assert_eq!(
            git.author_pool, imported.author_pool,
            "author pools diverge"
        );
        assert_eq!(git.files.len(), imported.files.len());
        for (file, git_churn) in &git.files {
            let imp = &imported.files[file];
            assert_eq!(git_churn.commits, imp.commits, "commits for {file:?}");
            assert_eq!(git_churn.lines_added, imp.lines_added, "added for {file:?}");
            assert_eq!(
                git_churn.lines_deleted, imp.lines_deleted,
                "deleted for {file:?}"
            );
            assert_eq!(git_churn.trend, imp.trend, "trend for {file:?}");
            assert_eq!(git_churn.authors, imp.authors, "authors for {file:?}");
            assert!(
                (git_churn.weighted_commits - imp.weighted_commits).abs() < f64::EPSILON,
                "weighted_commits for {file:?}: {} vs {}",
                git_churn.weighted_commits,
                imp.weighted_commits
            );
        }
    }
}