xbp_cli/commands/

worktree_watch.rs

use crate::codetime::GithubRepoRecord;
use crate::commands::code_snapshot::{snapshots_for_file, CodeSnapshot};
use crate::commands::cli_session::{cli_request_client, resolve_cli_access_token};
use crate::commands::terminal_table::{render_table, TableStyle};
use crate::config::{
    ensure_global_xbp_paths, map_windows_path_to_wsl_mnt, map_wsl_mnt_path_to_windows,
    resolve_device_identity, resolve_global_xbp_root_dir, resolve_worktree_watch_config, ApiConfig,
    SshConfig, WorktreeWatchConfig, WorktreeWatchRepoRank,
};
use chrono::{DateTime, Utc};
use colored::Colorize;
use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode};
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use reqwest::StatusCode;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::{mpsc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind};
use uuid::Uuid;

const BACKGROUND_CHILD_ENV: &str = "XBP_WORKTREE_WATCH_BACKGROUND_CHILD";
const DEFAULT_REMOTE: &str = "origin";
const EVENT_DEDUPE_WINDOW_SECONDS: i64 = 5;
const WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT: usize = 250;
const WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES: usize = 768 * 1024;
const WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS: usize = 3;
const WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS: u64 = 2;
/// Bound in-process fingerprint cache; durable dedupe prefers journal lines
/// (and legacy `*.seen` markers for older spools).
const MAX_RECENT_EVENT_DEDUPE_IN_MEMORY: usize = 4096;
/// Rotate `event-dedupe.jsonl` when it grows past this size (bytes).
const MAX_EVENT_DEDUPE_JSONL_BYTES: u64 = 2 * 1024 * 1024;
/// How often parent/single watch loops prune spool artifacts.
const SPOOL_HOUSEKEEPING_INTERVAL_SECONDS: u64 = 600;
/// Coalesce appends to spool jsonl files (fewer open/fsync storms).
const JSONL_FLUSH_MAX_LINES: usize = 32;
const JSONL_FLUSH_MAX_WAIT_MS: u64 = 250;
/// Max notify events drained per watch-loop tick (burst coalescing).
const NOTIFY_EVENT_DRAIN_CAP: usize = 64;

/// Client-side ignore rules: built-in VCS/generated dirs + global config filters
/// + `~/.xbp/.xbp-worktree-ignore` + project `.xbp/.xbpignore` / `watch_ignore_paths`.
///
/// Project `ignore_paths` (service discovery) intentionally do **not** apply here so
/// ignored package roots still count for worktree-watch activity.
/// Never uploaded; applied before events are spooled and again before sync.
#[derive(Debug, Clone, Default)]
struct WatchIgnoreRules {
    /// Normalized lowercase prefixes without leading `./` (e.g. `secrets`, `apps/web/.env`).
    forbidden_path_prefixes: Vec<String>,
    /// Lowercase path segment names blocked anywhere (includes built-ins + config).
    forbidden_folders: BTreeSet<String>,
    /// Lowercase substrings matched against the full relative path.
    banned_words: Vec<String>,
    /// Machine-wide gitignore-style rules (`~/.xbp/.xbp-worktree-ignore`).
    global_ignore: crate::utils::XbpIgnoreSet,
    /// Project-local gitignore-style rules (`.xbp/.xbpignore`, yaml `watch_ignore_paths`).
    project_ignore: crate::utils::XbpIgnoreSet,
}

impl WatchIgnoreRules {
    fn load() -> Self {
        Self::from_config(&resolve_worktree_watch_config())
    }

    fn from_config(config: &WorktreeWatchConfig) -> Self {
        Self::from_config_and_project(config, None)
    }

    fn from_config_and_project(config: &WorktreeWatchConfig, project_root: Option<&Path>) -> Self {
        let mut forbidden_folders = built_in_forbidden_folders();
        for folder in &config.forbidden_folders {
            let normalized = normalize_folder_token(folder);
            if !normalized.is_empty() {
                forbidden_folders.insert(normalized);
            }
        }

        let forbidden_path_prefixes = config
            .forbidden_paths
            .iter()
            .map(|path| normalize_path_prefix(path))
            .filter(|path| !path.is_empty())
            .collect::<Vec<_>>();

        let banned_words = config
            .banned_words
            .iter()
            .map(|word| word.trim().to_ascii_lowercase())
            .filter(|word| !word.is_empty())
            .collect::<Vec<_>>();

        let project_ignore = project_root
            .map(|root| load_project_watch_ignore(root))
            .unwrap_or_default();

        Self {
            forbidden_path_prefixes,
            forbidden_folders,
            banned_words,
            global_ignore: crate::config::load_global_worktree_ignore(),
            project_ignore,
        }
    }

    fn with_project_root(mut self, project_root: &Path) -> Self {
        self.project_ignore = load_project_watch_ignore(project_root);
        self
    }

    fn ignores_relative(&self, relative: &str) -> bool {
        let normalized = normalize_relative_watch_path(relative);
        if normalized.is_empty() {
            return false;
        }

        if self.global_ignore.is_ignored_relative(&normalized, false)
            || self.global_ignore.is_ignored_relative(&normalized, true)
        {
            return true;
        }

        if self.project_ignore.is_ignored_relative(&normalized, false)
            || self.project_ignore.is_ignored_relative(&normalized, true)
        {
            return true;
        }

        let lower = normalized.to_ascii_lowercase();
        let components = lower
            .split('/')
            .filter(|component| !component.is_empty())
            .collect::<Vec<_>>();

        if components
            .iter()
            .any(|component| self.forbidden_folders.contains(*component))
        {
            return true;
        }

        for prefix in &self.forbidden_path_prefixes {
            if lower == *prefix || lower.starts_with(&format!("{prefix}/")) {
                return true;
            }
            // Also treat a single-segment prefix as a folder ban anywhere.
            if !prefix.contains('/') && components.iter().any(|component| *component == prefix) {
                return true;
            }
        }

        for word in &self.banned_words {
            if lower.contains(word) {
                return true;
            }
        }

        false
    }

    fn ignores_path(&self, root: &Path, path: &Path) -> bool {
        if let Some(relative) = relative_watch_path(root, path) {
            return self.ignores_relative(&relative);
        }
        false
    }

    fn skips_discovery_dir(&self, path: &Path) -> bool {
        let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
            return false;
        };
        let trimmed = name.trim();
        if self
            .forbidden_folders
            .contains(&trimmed.to_ascii_lowercase())
        {
            return true;
        }
        if self.global_ignore.skips_dir_name(trimmed) {
            return true;
        }
        self.project_ignore.skips_dir_name(trimmed)
    }
}

fn load_project_watch_ignore(project_root: &Path) -> crate::utils::XbpIgnoreSet {
    use crate::strategies::XbpConfig;
    use crate::utils::{load_project_xbp_ignore, parse_config_with_auto_heal};

    let mut extra = Vec::new();
    // Watch-only patterns from xbp.yaml (`watch_ignore_paths` / `watch_ignore`).
    // Service-discovery `ignore_paths` are intentionally excluded so worktree-watch
    // still records activity under packages that are not registered as services.
    if let Some(found) = crate::utils::find_xbp_config_upwards(project_root) {
        if let Ok(content) = fs::read_to_string(&found.config_path) {
            if let Ok((config, _)) = parse_config_with_auto_heal::<XbpConfig>(&content, found.kind)
            {
                extra = config.watch_ignore_paths;
            }
        }
        return load_project_xbp_ignore(&found.project_root, &extra);
    }
    load_project_xbp_ignore(project_root, &extra)
}

fn built_in_forbidden_folders() -> BTreeSet<String> {
    use crate::utils::DEFAULT_NOISE_DIR_NAMES;

    DEFAULT_NOISE_DIR_NAMES
        .iter()
        .map(|name| name.to_ascii_lowercase())
        .collect()
}

fn normalize_folder_token(raw: &str) -> String {
    raw.trim()
        .trim_matches(|ch| ch == '/' || ch == '\\')
        .to_ascii_lowercase()
}

fn normalize_path_prefix(raw: &str) -> String {
    let trimmed = raw.trim().replace('\\', "/");
    let without_dot = trimmed
        .trim_start_matches("./")
        .trim_matches(|ch| ch == '/' || ch == '\\');
    without_dot.to_ascii_lowercase()
}

fn normalize_relative_watch_path(raw: &str) -> String {
    raw.trim()
        .replace('\\', "/")
        .trim_start_matches("./")
        .trim_matches('/')
        .to_string()
}

fn watch_ignore_rules() -> &'static WatchIgnoreRules {
    static RULES: OnceLock<WatchIgnoreRules> = OnceLock::new();
    RULES.get_or_init(WatchIgnoreRules::load)
}

/// Global config rules merged with per-repo `.xbp/.xbpignore` / `ignore_paths`.
fn watch_ignore_rules_for_root(root: &Path) -> WatchIgnoreRules {
    watch_ignore_rules().clone().with_project_root(root)
}

#[cfg(windows)]
const CREATE_NO_WINDOW: u32 = 0x08000000;

#[derive(Debug, Clone, Default)]
pub struct WorktreeWatchTargetOptions {
    pub repo: Option<PathBuf>,
    pub parent: Option<PathBuf>,
    /// Explicit multi-repo list (tray / advanced targeting). Ignored when `parent` is set.
    pub repos: Vec<PathBuf>,
}

/// Prefer a running parent watcher when `status` is invoked with no target.
///
/// Lets `xbp worktree-watch status` from any cwd show the active parent cover
/// (e.g. Documents/GitHub) instead of failing outside a git repo or only
/// reporting the current checkout as stopped.
pub fn resolve_default_worktree_watch_status_target() -> Result<WorktreeWatchTargetOptions, String> {
    if let Some(parent) = prefer_running_parent_root() {
        return Ok(WorktreeWatchTargetOptions {
            repo: None,
            parent: Some(parent),
            repos: Vec::new(),
        });
    }
    // Fall back to the most recent parent state even if the process just died,
    // so bare `status` still shows the cover matrix instead of a git cwd error.
    if let Some(parent) = prefer_any_parent_root() {
        return Ok(WorktreeWatchTargetOptions {
            repo: None,
            parent: Some(parent),
            repos: Vec::new(),
        });
    }
    normalize_worktree_watch_target(WorktreeWatchTargetOptions {
        repo: None,
        parent: None,
        repos: Vec::new(),
    })
}

/// Resolve inventory names and pin absolute paths so detached tray/watcher children
/// do not depend on the caller's cwd (or a bare short name like `athena`).
pub fn normalize_worktree_watch_target(
    target: WorktreeWatchTargetOptions,
) -> Result<WorktreeWatchTargetOptions, String> {
    if target.parent.is_some() && target.repo.is_some() {
        return Err("Pass either `--repo` or `--parent`, not both.".to_string());
    }
    if target.parent.is_some() && !target.repos.is_empty() {
        return Err("Pass either `--repos` or `--parent`, not both.".to_string());
    }
    if target.repo.is_some() && !target.repos.is_empty() {
        return Err("Pass either `--repo` or `--repos`, not both.".to_string());
    }

    if let Some(parent) = target.parent.as_ref() {
        let resolved = if parent.is_dir() {
            normalize_windows_verbatim_path(
                fs::canonicalize(parent).unwrap_or_else(|_| parent.clone()),
            )
        } else {
            // Allow bare folder names under common clone roots.
            if let Some(found) = resolve_parent_folder_hint(parent)? {
                found
            } else {
                return Err(format!(
                    "Parent folder not found: {}. Pass an absolute path or a folder under Documents/GitHub.",
                    parent.display()
                ));
            }
        };
        return Ok(WorktreeWatchTargetOptions {
            repo: None,
            parent: Some(resolved),
            repos: Vec::new(),
        });
    }

    if !target.repos.is_empty() {
        let mut repos = Vec::new();
        let mut seen = BTreeSet::new();
        for repo in &target.repos {
            let path = resolve_repo_path_input(Some(repo.as_path()))?;
            let absolute = normalize_windows_verbatim_path(
                fs::canonicalize(&path).unwrap_or(path),
            );
            let key = path_identity_key(&absolute);
            if seen.insert(key) {
                repos.push(absolute);
            }
        }
        return Ok(WorktreeWatchTargetOptions {
            repo: None,
            parent: None,
            repos,
        });
    }

    // Single repo (explicit or default to cwd) — always pin absolute root.
    let identity = resolve_repo_identity(target.repo.as_deref())?;
    Ok(WorktreeWatchTargetOptions {
        repo: Some(identity.root),
        parent: None,
        repos: Vec::new(),
    })
}

fn resolve_parent_folder_hint(parent: &Path) -> Result<Option<PathBuf>, String> {
    if parent.is_absolute() {
        return Ok(None);
    }
    let name = parent
        .file_name()
        .and_then(|value| value.to_str())
        .unwrap_or_default();
    if name.is_empty() {
        return Ok(None);
    }

    let mut candidates: Vec<PathBuf> = Vec::new();
    if let Some(home) = dirs::home_dir() {
        candidates.push(home.join("Documents").join("GitHub").join(name));
        candidates.push(home.join("Documents").join("github").join(name));
        candidates.push(home.join("Documents").join(name));
        candidates.push(home.join("src").join(name));
        candidates.push(home.join(name));
    }
    // If the hint itself is "GitHub", resolve the common clone root.
    if name.eq_ignore_ascii_case("github") {
        if let Some(home) = dirs::home_dir() {
            candidates.insert(0, home.join("Documents").join("GitHub"));
            candidates.insert(1, home.join("Documents").join("github"));
        }
    }

    for candidate in candidates {
        if candidate.is_dir() {
            return Ok(Some(normalize_windows_verbatim_path(
                fs::canonicalize(&candidate).unwrap_or(candidate),
            )));
        }
    }
    Ok(None)
}

/// Live snapshot for the worktree-watch system tray UI.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct WorktreeWatchTraySnapshot {
    pub target_label: String,
    pub mode: String,
    pub parent: Option<String>,
    pub repository_count: usize,
    pub running_watchers: usize,
    pub any_running: bool,
    pub all_running: bool,
    pub parent_watcher_running: bool,
    pub parent_watcher_pid: Option<u32>,
    pub total_unsynced_files: u64,
    pub total_unsynced_records: u64,
    pub repositories: Vec<WorktreeWatchTrayRepoStatus>,
    pub stats_line: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct WorktreeWatchTrayRepoStatus {
    pub root: String,
    pub owner: String,
    pub name: String,
    pub branch: String,
    pub running: bool,
    pub pid: Option<u32>,
    pub unsynced_files: u64,
    pub unsynced_records: u64,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchStartOptions {
    pub target: WorktreeWatchTargetOptions,
    pub detach: bool,
    pub sync_interval_seconds: u64,
    pub once: bool,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchSyncOptions {
    pub target: WorktreeWatchTargetOptions,
    pub dry_run: bool,
    pub resync: bool,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchStopOptions {
    pub target: WorktreeWatchTargetOptions,
    pub force: bool,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchStatusOptions {
    pub target: WorktreeWatchTargetOptions,
    pub json: bool,
    pub records: bool,
    pub record_limit: usize,
    pub stats: bool,
    pub repo_activity: bool,
    pub stats_gap_minutes: u64,
    /// Include adaptive tier / rank / process resource insight (default true).
    pub resource: bool,
}

#[derive(Debug, Clone)]
struct RepoIdentity {
    owner: String,
    name: String,
    branch: String,
    root: PathBuf,
    head_sha: Option<String>,
}

#[derive(Debug, Clone)]
struct SpoolLayout {
    root: PathBuf,
    events_file: PathBuf,
    commits_file: PathBuf,
    stats_file: PathBuf,
    watcher_state_file: PathBuf,
    sync_log_file: PathBuf,
    event_dedupe_file: PathBuf,
    event_dedupe_dir: PathBuf,
    commit_dedupe_dir: PathBuf,
}

#[derive(Debug, Clone)]
struct ParentSpoolLayout {
    watcher_state_file: PathBuf,
}

/// Adaptive watch intensity for a repository.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RepoWatchTier {
    /// Recent activity: preferred for recursive FS watch + frequent HEAD checks.
    Hot,
    /// Quiet: slower HEAD checks; FS watch only if under the concurrent watch budget.
    Warm,
    /// Idle: no FS watch (releases OS directory locks so the folder can be deleted).
    Cold,
}

impl RepoWatchTier {
    fn as_str(self) -> &'static str {
        match self {
            Self::Hot => "hot",
            Self::Warm => "warm",
            Self::Cold => "cold",
        }
    }
}

#[derive(Debug)]
struct WatchedRepo {
    identity: RepoIdentity,
    layout: SpoolLayout,
    last_head: Option<String>,
    event_dedupe: RecentEventDedupe,
    /// Last FS mutation or meaningful commit change.
    last_activity: Instant,
    /// Decayed activity score (higher = busier).
    activity_score: f64,
    tier: RepoWatchTier,
    /// Whether the native/poll watcher currently holds this repo root.
    fs_watched: bool,
    /// Next time we may run `git rev-parse HEAD` for this repo.
    next_commit_check: Instant,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationEvent {
    id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    fingerprint: Option<String>,
    repo_owner: String,
    repo_name: String,
    branch_name: String,
    repo_root: String,
    head_sha: Option<String>,
    event_kind: String,
    paths: Vec<String>,
    primary_path: Option<String>,
    old_path: Option<String>,
    new_path: Option<String>,
    added_lines: Option<u64>,
    removed_lines: Option<u64>,
    total_lines: Option<u64>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    code_snapshots: Vec<CodeSnapshot>,
    file_created: bool,
    file_removed: bool,
    folder_created: bool,
    folder_removed: bool,
    renamed_or_moved: bool,
    raw_kind: String,
    occurred_at: DateTime<Utc>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeCommitEvent {
    id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    fingerprint: Option<String>,
    repo_owner: String,
    repo_name: String,
    branch_name: String,
    repo_root: String,
    previous_head_sha: Option<String>,
    head_sha: String,
    subject: Option<String>,
    author_name: Option<String>,
    author_email: Option<String>,
    committed_at: Option<String>,
    occurred_at: DateTime<Utc>,
    /// Paths touched by previous_head..head (capped).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    files_changed: Option<Vec<String>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    files_changed_count: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    insertions: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    deletions: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    is_merge: Option<bool>,
    /// When HEAD jumped multiple commits, SHAs in the range (newest first, capped).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    included_commits: Option<Vec<String>>,
}

/// Intent written before install so watchers can relaunch on the new binary.
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchRestartIntent {
    #[serde(default)]
    parents: Vec<String>,
    #[serde(default)]
    repos: Vec<String>,
    #[serde(default)]
    tray: bool,
    #[serde(default)]
    stopped_at: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationIngestPayload {
    device: WorktreeMutationDevicePayload,
    repository: WorktreeMutationRepositoryPayload,
    events: Vec<WorktreeMutationEvent>,
    commits: Vec<WorktreeCommitEvent>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationDevicePayload {
    hardware_id: String,
    device_name: Option<String>,
    hostname: Option<String>,
    platform: String,
    /// Stable runtime isolation key (e.g. `windows`, `wsl-ubuntu`, `linux`).
    /// Lets Windows + WSL watchers coexist as distinct upload sources.
    #[serde(skip_serializing_if = "Option::is_none")]
    runtime_key: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationRepositoryPayload {
    owner: String,
    name: String,
    branch_name: String,
    repo_root: String,
}

#[derive(Debug)]
struct SyncCandidate {
    path: PathBuf,
    records: SyncRecords,
}

#[derive(Debug)]
struct WorktreeMutationUploadBatch {
    events: Vec<WorktreeMutationEvent>,
    commits: Vec<WorktreeCommitEvent>,
    estimated_json_bytes: usize,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WorktreeMutationSyncErrorKind {
    Retryable,
    NonRetryable,
}

#[derive(Debug)]
struct WorktreeMutationSyncError {
    kind: WorktreeMutationSyncErrorKind,
    message: String,
}

struct SyncLogAppend<'a> {
    identity: &'a RepoIdentity,
    layout: &'a SpoolLayout,
    endpoint: &'a str,
    status_code: u16,
    event_count: usize,
    commit_count: usize,
    candidates: &'a [SyncCandidate],
    resync: bool,
}

#[derive(Debug, Clone, Copy)]
enum SyncFileKind {
    Events,
    Commits,
}

#[derive(Debug)]
enum SyncRecords {
    Events(Vec<WorktreeMutationEvent>),
    Commits(Vec<WorktreeCommitEvent>),
}

impl SyncRecords {
    fn len(&self) -> usize {
        match self {
            Self::Events(records) => records.len(),
            Self::Commits(records) => records.len(),
        }
    }

    fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchStatsSummary {
    generated_at: DateTime<Utc>,
    repository_owner: String,
    repository_name: String,
    branch_name: String,
    repo_root: String,
    spool_root: String,
    session_gap_minutes: u64,
    total_events: u64,
    total_commits: u64,
    first_event_at: Option<DateTime<Utc>>,
    last_event_at: Option<DateTime<Utc>>,
    session_count: u64,
    observed_span_seconds: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
    by_file: Vec<WorktreeWatchFileStats>,
    by_filetype: Vec<WorktreeWatchBucketStats>,
    by_event_kind: Vec<WorktreeWatchBucketStats>,
}

#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchRepoActivitySummary {
    generated_at: DateTime<Utc>,
    repository_owner: String,
    repository_name: String,
    repo_root: String,
    repository_spool_root: String,
    stats_file: String,
    session_gap_minutes: u64,
    branch_count: u64,
    total_events: u64,
    total_commits: u64,
    first_event_at: Option<DateTime<Utc>>,
    last_event_at: Option<DateTime<Utc>>,
    session_count: u64,
    observed_span_seconds: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
    branches: Vec<WorktreeWatchBranchActivityStats>,
}

#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchBranchActivityStats {
    branch_name: String,
    spool_root: String,
    total_events: u64,
    total_commits: u64,
    first_event_at: Option<DateTime<Utc>>,
    last_event_at: Option<DateTime<Utc>>,
    session_count: u64,
    observed_span_seconds: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
}

#[derive(Debug, Serialize, Deserialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchFileStats {
    path: String,
    filetype: String,
    event_count: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
}

#[derive(Debug, Serialize, Deserialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchBucketStats {
    name: String,
    event_count: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatcherState {
    pid: u32,
    repo_root: String,
    repository_owner: String,
    repository_name: String,
    branch_name: String,
    executable: String,
    started_at: DateTime<Utc>,
    /// Present on modern state files so Windows/WSL/Linux watchers never clobber each other.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    runtime_key: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    platform: Option<String>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct ParentWorktreeWatcherState {
    pid: u32,
    parent_root: String,
    executable: String,
    started_at: DateTime<Utc>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    runtime_key: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    platform: Option<String>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchSyncLogEntry {
    id: String,
    synced_at: DateTime<Utc>,
    endpoint: String,
    repository_owner: String,
    repository_name: String,
    branch_name: String,
    repo_root: String,
    resync: bool,
    status_code: u16,
    event_count: usize,
    commit_count: usize,
    spool_file_count: usize,
    spool_files: Vec<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StopWatcherOutcome {
    Stopped(u32),
    RemovedStaleState,
    NoState,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchEventDedupeRecord {
    fingerprint: String,
    first_seen_at: DateTime<Utc>,
    event_kind: String,
    primary_path: Option<String>,
}

#[derive(Debug, Default)]
struct RecentEventDedupe {
    fingerprints: BTreeSet<String>,
}

pub async fn run_worktree_watch_start(options: WorktreeWatchStartOptions) -> Result<(), String> {
    if options.detach {
        // Keep detached start free of the foreground watch future so the local
        // HTTP API can call this without creating an async type cycle.
        return run_worktree_watch_start_detached(options);
    }

    if let Some(parent) = options.target.parent.as_deref() {
        let identities = resolve_target_identities(&options.target)?;
        if identities.is_empty() {
            println!("No git repositories found under parent folder.");
            return Ok(());
        }

        if options.once {
            for identity in &identities {
                let layout = prepare_spool_layout(identity)?;
                record_commit_snapshot(identity, &layout, None)?;
                println!("Recorded current git state for {}", identity.root.display());
            }
            return Ok(());
        }

        let parent = canonical_parent_path(parent)?;
        println!(
            "Watching {} recursively and routing mutations for {} repo(s).",
            parent.display(),
            identities.len()
        );
        return watch_parent_foreground(parent, identities, options.sync_interval_seconds).await;
    }

    let identity = resolve_repo_identity(options.target.repo.as_deref())?;
    let layout = prepare_spool_layout(&identity)?;
    println!(
        "Watching {} and spooling mutations to {} [runtime={}]",
        identity.root.display(),
        layout.root.display(),
        worktree_runtime_key()
    );

    if options.once {
        record_commit_snapshot(&identity, &layout, None)?;
        return Ok(());
    }

    watch_foreground(identity, layout, options.sync_interval_seconds).await
}

/// Spawn detached watcher process(es) only (no foreground watch loop).
///
/// Used by the local HTTP API and CLI `--detach` so handlers never type-depend
/// on the infinite `watch_foreground` future.
pub fn run_worktree_watch_start_detached(
    options: WorktreeWatchStartOptions,
) -> Result<(), String> {
    if let Some(parent) = options.target.parent.as_deref() {
        // Fast .git root count only — full identity resolve is deferred to the child.
        let parent_path = canonical_parent_path(parent)?;
        let repo_count = count_git_roots_under(&parent_path)?;
        if repo_count == 0 {
            println!("No git repositories found under parent folder.");
            return Ok(());
        }
        let path = spawn_detached_parent_worktree_watch(&parent_path)?;
        println!(
            "Started one background worktree watcher for {} covering {} repo(s).",
            path.display(),
            repo_count
        );
        if let Some(port) =
            crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
        {
            println!(
                "Local API (when free): http://127.0.0.1:{port}  — dashboard: apps/worktree-watch-app/"
            );
        }
        return Ok(());
    }

    if !options.target.repos.is_empty() {
        for repo in &options.target.repos {
            let path = spawn_detached_worktree_watch(Some(repo.as_path()))?;
            println!("Started background worktree watcher for {}", path.display());
        }
        if let Some(port) =
            crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
        {
            println!(
                "Local API (when free): http://127.0.0.1:{port}  — dashboard: apps/worktree-watch-app/"
            );
        }
        return Ok(());
    }

    spawn_detached_worktree_watch(options.target.repo.as_deref()).map(|path| {
        println!("Started background worktree watcher for {}", path.display());
        if let Some(port) =
            crate::commands::worktree_watch_api::resolve_worktree_watch_api_port(None)
        {
            println!(
                "Local API (when free): http://127.0.0.1:{port}  — dashboard: apps/worktree-watch-app/"
            );
        }
    })
}

pub async fn run_worktree_watch_sync(options: WorktreeWatchSyncOptions) -> Result<(), String> {
    let identities = resolve_target_identities(&options.target)?;
    let repo_count = identities.len();
    let mut plans = Vec::new();
    let mut total_files = 0usize;
    let mut total_records = 0usize;

    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        let candidates = collect_spool_candidates(&layout, options.resync)?;
        total_files += candidates.len();
        total_records += candidates
            .iter()
            .map(|candidate| candidate.records.len())
            .sum::<usize>();
        if !candidates.is_empty() {
            plans.push((identity, candidates));
        }
    }

    if options.dry_run {
        println!(
            "Would sync {} record(s) from {} spool file(s) across {} repo(s).",
            total_records, total_files, repo_count
        );
        if options.resync {
            println!("Resync mode includes files already marked `.synced.*.jsonl`.");
        }
        return Ok(());
    }

    if plans.is_empty() {
        print_no_sync_candidates_message(&options.target, options.resync)?;
        return Ok(());
    }

    let upload_repo_count = plans.len();
    for (identity, candidates) in plans {
        sync_candidates(&identity, candidates, options.resync).await?;
    }
    println!(
        "Synced {} worktree mutation record(s) across {} repo(s).",
        total_records, upload_repo_count
    );
    Ok(())
}

pub fn run_worktree_watch_stop(options: WorktreeWatchStopOptions) -> Result<(), String> {
    if let Some(parent) = options.target.parent.as_deref() {
        let parent = canonical_parent_path(parent)?;
        match stop_existing_parent_watcher(&parent, options.force)? {
            StopWatcherOutcome::Stopped(pid) => {
                println!(
                    "Stopped background parent worktree watcher {} for {}",
                    pid,
                    parent.display()
                );
            }
            StopWatcherOutcome::RemovedStaleState => {
                println!(
                    "Removed stale parent worktree watcher state for {}",
                    parent.display()
                );
            }
            StopWatcherOutcome::NoState => {
                println!("No parent worktree watcher state for {}", parent.display());
            }
        }
    }

    let identities = resolve_target_identities(&options.target)?;
    if identities.is_empty() {
        println!("No git repositories found for worktree watcher stop.");
        return Ok(());
    }

    let mut stopped = 0usize;
    let mut stale = 0usize;
    let mut missing = 0usize;
    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        match stop_existing_watcher(&identity, &layout, options.force)? {
            StopWatcherOutcome::Stopped(pid) => {
                stopped += 1;
                println!(
                    "Stopped background worktree watcher {} for {}",
                    pid,
                    identity.root.display()
                );
            }
            StopWatcherOutcome::RemovedStaleState => {
                stale += 1;
                println!(
                    "Removed stale worktree watcher state for {}",
                    identity.root.display()
                );
            }
            StopWatcherOutcome::NoState => {
                missing += 1;
                println!(
                    "No background worktree watcher state for {}",
                    identity.root.display()
                );
            }
        }
    }

    println!(
        "Worktree watcher stop complete: {stopped} stopped, {stale} stale state file(s) removed, {missing} without state."
    );
    Ok(())
}

/// Best-effort stop of whatever worktree-watch cover is active (parent or repo(s))
/// so `cargo install` / `xbp update --install` can replace the XBP binary without
/// Windows file locks / file-claim failures.
///
/// Never fails the caller: prints what it stopped (or a soft warning) and returns.
/// Records a restart intent so [`restart_worktree_watch_after_update`] can relaunch.
pub fn stop_active_worktree_watch_before_install() {
    let mut intent = WorktreeWatchRestartIntent {
        stopped_at: Some(Utc::now().to_rfc3339()),
        ..Default::default()
    };
    let mut stopped_parent = false;

    // Prefer live parent cover(s) — one process holds watches for the whole tree.
    // Avoid discovering every git repo under the parent (noisy + slow on large folders).
    if let Ok(mut candidates) = list_parent_watcher_state_files() {
        candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
        for (_path, state) in candidates {
            let system = process_system_for_pids(&[state.pid]);
            if !is_live_parent_watcher_process_with_system(&system, &state) {
                continue;
            }
            let parent = parent_root_as_local_path(&state.parent_root);
            match stop_existing_parent_watcher(&parent, true) {
                Ok(StopWatcherOutcome::Stopped(pid)) => {
                    stopped_parent = true;
                    intent.parents.push(parent.display().to_string());
                    println!(
                        "Stopped worktree-watch parent watcher {} for {} before install.",
                        pid,
                        parent.display()
                    );
                }
                Ok(StopWatcherOutcome::RemovedStaleState) => {
                    println!(
                        "Removed stale parent worktree-watch state for {} before install.",
                        parent.display()
                    );
                }
                Ok(StopWatcherOutcome::NoState) => {}
                Err(error) => {
                    println!(
                        "Warning: could not stop parent worktree-watch for {}: {error}",
                        parent.display()
                    );
                }
            }
        }
    }

    if stopped_parent {
        let _ = write_worktree_watch_restart_intent(&intent);
        return;
    }

    // No live parent: stop an active per-repo / multi-repo cover when present.
    let Ok(target) = resolve_default_worktree_watch_status_target() else {
        return;
    };
    // Default status target may fall back to a stale parent path; we already
    // handled live parents above, so skip parent-only targets here.
    if target.parent.is_some() {
        return;
    }

    let running = collect_worktree_watch_tray_snapshot(&target)
        .map(|snapshot| snapshot.any_running)
        .unwrap_or(false);
    if !running {
        return;
    }

    if let Some(repo) = target.repo.as_ref() {
        intent.repos.push(repo.display().to_string());
    }
    for repo in &target.repos {
        intent.repos.push(repo.display().to_string());
    }

    match run_worktree_watch_stop(WorktreeWatchStopOptions {
        target,
        force: true,
    }) {
        Ok(()) => {
            let _ = write_worktree_watch_restart_intent(&intent);
            println!("Stopped worktree-watch before install.");
        }
        Err(error) => {
            println!("Warning: could not stop worktree-watch before install: {error}");
        }
    }
}

/// Relaunch worktree watchers recorded by [`stop_active_worktree_watch_before_install`].
/// Best-effort: never fails the caller.
pub fn restart_worktree_watch_after_update() {
    let Some(intent) = read_worktree_watch_restart_intent() else {
        return;
    };
    let _ = clear_worktree_watch_restart_intent();

    let mut restarted = 0usize;
    for parent in &intent.parents {
        let path = PathBuf::from(parent);
        if !path.is_dir() {
            continue;
        }
        match spawn_detached_parent_worktree_watch(&path) {
            Ok(_) => {
                restarted += 1;
                println!(
                    "Restarted worktree-watch parent cover for {} after install.",
                    path.display()
                );
            }
            Err(error) => {
                println!(
                    "Warning: could not restart parent worktree-watch for {}: {error}",
                    path.display()
                );
            }
        }
    }

    for repo in &intent.repos {
        let path = PathBuf::from(repo);
        if !path.is_dir() {
            continue;
        }
        match resolve_repo_identity(Some(&path)).and_then(|id| {
            spawn_detached_worktree_watch_for_identity(&id).map(|_| id.root)
        }) {
            Ok(root) => {
                restarted += 1;
                println!(
                    "Restarted worktree-watch for {} after install.",
                    root.display()
                );
            }
            Err(error) => {
                println!(
                    "Warning: could not restart worktree-watch for {}: {error}",
                    path.display()
                );
            }
        }
    }

    if intent.tray {
        match crate::commands::worktree_watch_tray::spawn_detached_tray_after_update() {
            Ok(()) => {
                restarted += 1;
                println!("Restarted worktree-watch tray after install.");
            }
            Err(error) => {
                println!("Warning: could not restart worktree-watch tray: {error}");
            }
        }
    }

    if restarted == 0
        && intent.parents.is_empty()
        && intent.repos.is_empty()
        && !intent.tray
    {
        // nothing recorded
    }
}

fn worktree_watch_restart_intent_path() -> Result<PathBuf, String> {
    let paths = ensure_global_xbp_paths()?;
    Ok(paths
        .root_dir
        .join("worktree-watch")
        .join("restart-after-update.json"))
}

fn write_worktree_watch_restart_intent(intent: &WorktreeWatchRestartIntent) -> Result<(), String> {
    let path = worktree_watch_restart_intent_path()?;
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(|e| {
            format!(
                "Failed to create {}: {e}",
                parent.display()
            )
        })?;
    }
    let rendered = serde_json::to_string_pretty(intent)
        .map_err(|e| format!("Failed to serialize restart intent: {e}"))?;
    fs::write(&path, format!("{rendered}\n"))
        .map_err(|e| format!("Failed to write {}: {e}", path.display()))
}

fn read_worktree_watch_restart_intent() -> Option<WorktreeWatchRestartIntent> {
    let path = worktree_watch_restart_intent_path().ok()?;
    let content = fs::read_to_string(path).ok()?;
    serde_json::from_str(&content).ok()
}

fn clear_worktree_watch_restart_intent() -> Result<(), String> {
    let path = worktree_watch_restart_intent_path()?;
    if path.exists() {
        fs::remove_file(&path).map_err(|e| format!("Failed to remove {}: {e}", path.display()))?;
    }
    Ok(())
}

/// Mark that the tray should be relaunched after the next binary replace.
pub fn record_tray_restart_intent() {
    let mut intent = read_worktree_watch_restart_intent().unwrap_or_default();
    intent.tray = true;
    if intent.stopped_at.is_none() {
        intent.stopped_at = Some(Utc::now().to_rfc3339());
    }
    let _ = write_worktree_watch_restart_intent(&intent);
}

pub async fn run_worktree_watch_status(options: WorktreeWatchStatusOptions) -> Result<(), String> {
    // Lightweight overview (running detection + spool counts + last sync age).
    // Full JSONL decode only when --records / --stats / --repo-activity is set.
    let need_heavy = options.records || options.stats || options.repo_activity;
    let snapshot = collect_worktree_watch_tray_snapshot(&options.target)?;
    let watch_cfg = resolve_worktree_watch_config();
    let rank_index = build_repo_rank_index(&watch_cfg);

    let mut payloads = Vec::new();
    let mut total_files = 0usize;
    let mut total_records = 0usize;

    for tray_repo in &snapshot.repositories {
        let identity = identity_from_tray_repo(tray_repo);
        let mut status = if need_heavy {
            worktree_watch_status_payload(&identity, options.records, options.record_limit)?
        } else {
            // Reuse tray counts; only open sync-log for age.
            worktree_watch_status_from_tray_repo(tray_repo)?
        };

        if let Some(object) = status.as_object_mut() {
            object.insert("running".to_string(), json!(tray_repo.running));
            object.insert("pid".to_string(), json!(tray_repo.pid));
            if let Some(last_sync) = object.get("lastSync") {
                if let Some(synced_at) = last_sync.get("syncedAt").and_then(Value::as_str) {
                    if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
                        let age = (Utc::now() - dt.with_timezone(&Utc))
                            .num_seconds()
                            .max(0) as u64;
                        object.insert("lastSyncSecondsAgo".to_string(), json!(age));
                    }
                }
            }
            if options.resource {
                attach_repo_resource_fields(object, &identity.root, &rank_index, &watch_cfg);
            }
        }

        let stats = if options.stats {
            Some(generate_and_store_stats(
                &identity,
                options.stats_gap_minutes,
            )?)
        } else {
            None
        };
        let repo_activity = if options.repo_activity {
            Some(generate_and_store_repo_activity(
                &identity,
                options.stats_gap_minutes,
            )?)
        } else {
            None
        };
        // Refresh TODO→issue coding effort from this repo's mutation spool when present.
        if options.stats || options.repo_activity {
            if let Err(err) = crate::commands::todos::effort::recompute_effort(
                &identity.root,
                options.stats_gap_minutes,
                true,
            ) {
                eprintln!("note: todo effort recompute: {err}");
            }
        }
        let status = attach_optional_stats(status, stats, repo_activity)?;
        total_files += status
            .get("unsyncedFiles")
            .and_then(Value::as_u64)
            .unwrap_or(0) as usize;
        total_records += status
            .get("unsyncedRecords")
            .and_then(Value::as_u64)
            .unwrap_or(0) as usize;
        payloads.push(status);
    }

    if options.resource {
        // Hot repos first so the matrix / JSON surface busiest cover.
        payloads.sort_by(|a, b| {
            let sa = a.get("activityScore").and_then(Value::as_f64).unwrap_or(0.0);
            let sb = b.get("activityScore").and_then(Value::as_f64).unwrap_or(0.0);
            sb.partial_cmp(&sa)
                .unwrap_or(std::cmp::Ordering::Equal)
                .then_with(|| {
                    let ra = a.get("activityRank").and_then(Value::as_u64).unwrap_or(u64::MAX);
                    let rb = b.get("activityRank").and_then(Value::as_u64).unwrap_or(u64::MAX);
                    ra.cmp(&rb)
                })
        });
    }

    let resource = if options.resource {
        Some(build_resource_insight(&snapshot, &payloads, &watch_cfg))
    } else {
        None
    };

    if options.json {
        let mut payload = if options.target.parent.is_some() || payloads.len() > 1 {
            json!({
                "targetLabel": snapshot.target_label,
                "mode": snapshot.mode,
                "parent": snapshot.parent,
                "parentWatcherRunning": snapshot.parent_watcher_running,
                "parentWatcherPid": snapshot.parent_watcher_pid,
                "anyRunning": snapshot.any_running,
                "allRunning": snapshot.all_running,
                "runningWatchers": snapshot.running_watchers,
                "repositories": payloads,
                "repositoryCount": payloads.len(),
                "unsyncedFiles": total_files,
                "unsyncedRecords": total_records,
            })
        } else {
            payloads.into_iter().next().unwrap_or_else(|| json!({}))
        };
        if let Some(resource) = resource {
            if let Some(object) = payload.as_object_mut() {
                object.insert("resource".to_string(), resource);
            }
        }
        println!(
            "{}",
            serde_json::to_string_pretty(&payload)
                .map_err(|error| format!("Failed to render status JSON: {error}"))?
        );
    } else {
        print_status_overview(&snapshot, total_files, total_records);
        if let Some(resource) = resource.as_ref() {
            print_status_resource(resource);
        }
        print_status_matrix(&payloads, &snapshot, options.resource);
        if need_heavy {
            for (index, payload) in payloads.iter().enumerate() {
                if index > 0 {
                    println!();
                }
                if options.records {
                    println!(
                        "{} {}/{}",
                        "records".bright_black(),
                        payload
                            .get("repositoryOwner")
                            .and_then(Value::as_str)
                            .unwrap_or("unknown"),
                        payload
                            .get("repositoryName")
                            .and_then(Value::as_str)
                            .unwrap_or("repository")
                    );
                    print_status_records(payload);
                }
                if options.stats {
                    print_status_stats(payload);
                }
                if options.repo_activity {
                    print_repo_activity(payload);
                }
            }
        }
    }

    Ok(())
}

fn print_status_overview(
    snapshot: &WorktreeWatchTraySnapshot,
    total_files: usize,
    total_records: usize,
) {
    let running_label = if snapshot.any_running {
        "RUNNING".green().bold()
    } else {
        "STOPPED".red().bold()
    };
    println!(
        "{} {}  {}",
        "worktree-watch".bright_magenta().bold(),
        running_label,
        snapshot.target_label.bright_white()
    );

    if snapshot.mode == "parent" {
        let parent_label = if snapshot.parent_watcher_running {
            format!(
                "parent watcher {}",
                snapshot
                    .parent_watcher_pid
                    .map(|pid| format!("pid {pid}"))
                    .unwrap_or_else(|| "active".to_string())
            )
            .green()
            .to_string()
        } else {
            "parent watcher not detected".red().to_string()
        };
        println!(
            "  {}  {}  {} repo(s) covered",
            parent_label,
            "·".bright_black(),
            snapshot.repository_count
        );
    } else {
        println!(
            "  {} watching · {} repo(s)",
            snapshot.running_watchers, snapshot.repository_count
        );
    }

    println!(
        "  {} unsynced record(s) in {} file(s)",
        total_records.to_string().yellow(),
        total_files
    );
    if let Some(line) = snapshot.stats_line.as_ref() {
        println!("  {}", line.bright_black());
    }
    println!();
}

fn print_status_resource(resource: &Value) {
    println!("{}", "resource".bright_cyan().bold());

    let tiers = resource.get("tiers");
    let hot = tiers
        .and_then(|t| t.get("hot"))
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let warm = tiers
        .and_then(|t| t.get("warm"))
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let cold = tiers
        .and_then(|t| t.get("cold"))
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let unknown = tiers
        .and_then(|t| t.get("unknown"))
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let fs_watches = resource
        .get("estimatedFsWatches")
        .and_then(Value::as_u64)
        .unwrap_or(hot + warm);

    println!(
        "  tiers  {} hot · {} warm · {} cold{}",
        hot.to_string().red().bold(),
        warm.to_string().yellow(),
        cold.to_string().bright_black(),
        if unknown > 0 {
            format!(" · {unknown} unranked")
        } else {
            String::new()
        }
    );
    println!(
        "  watches  ~{} recursive FS watch(es)  {} cold unwatched (deletable)",
        fs_watches.to_string().cyan(),
        cold.to_string().bright_black()
    );

    if let Some(proc) = resource.get("process") {
        let pid = proc.get("pid").and_then(Value::as_u64);
        let cpu = proc.get("cpuPercent").and_then(Value::as_f64);
        let rss_mb = proc.get("rssMb").and_then(Value::as_f64);
        let mut parts = Vec::new();
        if let Some(pid) = pid {
            parts.push(format!("pid {pid}"));
        }
        if let Some(cpu) = cpu {
            parts.push(format!("cpu {cpu:.1}%"));
        }
        if let Some(rss) = rss_mb {
            parts.push(format!("rss {rss:.1} MiB"));
        }
        if !parts.is_empty() {
            let cpu_style = match cpu {
                Some(c) if c >= 5.0 => parts.join(" · ").red().to_string(),
                Some(c) if c >= 2.0 => parts.join(" · ").yellow().to_string(),
                _ => parts.join(" · ").green().to_string(),
            };
            println!("  process  {cpu_style}");
        }
    } else {
        println!(
            "  process  {}",
            "no live watcher process stats".bright_black()
        );
    }

    if let Some(sched) = resource.get("schedule") {
        let idle = sched
            .get("idleAfterSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let cold_after = sched
            .get("coldAfterSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let hot_c = sched
            .get("hotCommitCheckSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let warm_c = sched
            .get("warmCommitCheckSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let cold_c = sched
            .get("coldCommitCheckSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let max_hot = sched
            .get("maxHotRepos")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let max_fs = sched
            .get("maxFsWatches")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        println!(
            "  schedule  idle→warm {idle}s · cold {cold_after}s · HEAD {hot_c}/{warm_c}/{cold_c}s · max_hot {max_hot} · max_fs {max_fs}"
        );
    }

    if let Some(top) = resource.get("topRepos").and_then(Value::as_array) {
        if !top.is_empty() {
            println!("  top by activity score");
            for (i, entry) in top.iter().take(8).enumerate() {
                let name = entry
                    .get("repository")
                    .and_then(Value::as_str)
                    .unwrap_or("?");
                let tier = entry.get("tier").and_then(Value::as_str).unwrap_or("?");
                let score = entry.get("score").and_then(Value::as_f64).unwrap_or(0.0);
                let last = entry
                    .get("lastEventAt")
                    .and_then(Value::as_str)
                    .unwrap_or("—");
                let last_short = if last.len() >= 19 {
                    &last[11..19]
                } else {
                    last
                };
                let tier_col = match tier {
                    "hot" => tier.red().bold().to_string(),
                    "warm" => tier.yellow().to_string(),
                    "cold" => tier.bright_black().to_string(),
                    _ => tier.bright_black().to_string(),
                };
                println!(
                    "    {:>2}. {:<36}  {}  score {:>7.2}  last {}",
                    i + 1,
                    truncate_middle(name, 36),
                    tier_col,
                    score,
                    last_short
                );
            }
        }
    }

    if let Some(note) = resource.get("note").and_then(Value::as_str) {
        println!("  {}", note.bright_black());
    }
    println!();
}

fn print_status_matrix(
    payloads: &[Value],
    snapshot: &WorktreeWatchTraySnapshot,
    show_resource: bool,
) {
    if payloads.is_empty() {
        println!("{}", "No repositories found for this target.".yellow());
        return;
    }

    let headers: Vec<&str> = if show_resource {
        vec![
            "REPO",
            "BRANCH",
            "TIER",
            "RANK",
            "SCORE",
            "STATE",
            "PID",
            "UNSYNCED",
            "AGE",
        ]
    } else {
        vec![
            "REPO", "BRANCH", "STATE", "PID", "UNSYNCED", "LAST SYNC", "AGE",
        ]
    };
    let mut rows = Vec::with_capacity(payloads.len());

    for payload in payloads {
        let owner = payload
            .get("repositoryOwner")
            .or_else(|| payload.get("owner"))
            .and_then(Value::as_str)
            .unwrap_or("unknown");
        let name = payload
            .get("repositoryName")
            .or_else(|| payload.get("name"))
            .and_then(Value::as_str)
            .unwrap_or("repository");
        let branch = payload
            .get("branchName")
            .or_else(|| payload.get("branch"))
            .and_then(Value::as_str)
            .unwrap_or("unknown");
        let running = payload
            .get("running")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        let pid = payload
            .get("pid")
            .and_then(Value::as_u64)
            .map(|value| value.to_string())
            .unwrap_or_else(|| {
                if running {
                    snapshot
                        .parent_watcher_pid
                        .map(|value| value.to_string())
                        .unwrap_or_else(|| "—".to_string())
                } else {
                    "—".to_string()
                }
            });
        let unsynced = payload
            .get("unsyncedRecords")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let (last_sync_at, age) = last_sync_display(payload);
        let state = if running {
            "RUN".green().bold().to_string()
        } else {
            "OFF".red().to_string()
        };
        let unsynced_cell = if unsynced > 0 {
            unsynced.to_string().yellow().to_string()
        } else {
            "0".bright_black().to_string()
        };
        let age_cell = if age == "never" {
            age.bright_black().to_string()
        } else {
            age.cyan().to_string()
        };

        if show_resource {
            let tier = payload
                .get("watchTier")
                .and_then(Value::as_str)
                .unwrap_or("—");
            let tier_cell = match tier {
                "hot" => tier.red().bold().to_string(),
                "warm" => tier.yellow().to_string(),
                "cold" => tier.bright_black().to_string(),
                _ => tier.bright_black().to_string(),
            };
            let rank = payload
                .get("activityRank")
                .and_then(Value::as_u64)
                .map(|r| r.to_string())
                .unwrap_or_else(|| "—".to_string());
            let score = payload
                .get("activityScore")
                .and_then(Value::as_f64)
                .map(|s| format!("{s:.1}"))
                .unwrap_or_else(|| "—".to_string());
            rows.push(vec![
                format!("{owner}/{name}"),
                branch.to_string(),
                tier_cell,
                rank,
                score,
                state,
                pid,
                unsynced_cell,
                age_cell,
            ]);
        } else {
            rows.push(vec![
                format!("{owner}/{name}"),
                branch.to_string(),
                state,
                pid,
                unsynced_cell,
                last_sync_at,
                age_cell,
            ]);
        }
    }

    print!("{}", render_table(&headers, &rows, TableStyle::Box, ""));

    let running = snapshot.running_watchers;
    let stopped = snapshot
        .repository_count
        .saturating_sub(snapshot.running_watchers);
    println!();
    println!(
        "{} {} running · {} stopped · {} total",
        "summary".bright_black(),
        running.to_string().green().bold(),
        stopped.to_string().red(),
        snapshot.repository_count
    );
    if show_resource {
        println!(
            "{} tiers from ~/.xbp config ranks (updated while watcher runs)",
            "note".bright_black()
        );
    }
}

fn build_repo_rank_index(cfg: &WorktreeWatchConfig) -> BTreeMap<String, WorktreeWatchRepoRank> {
    let mut map = BTreeMap::new();
    for rank in &cfg.repo_ranks {
        let key = path_identity_key(Path::new(&rank.root));
        map.insert(key, rank.clone());
    }
    map
}

fn attach_repo_resource_fields(
    object: &mut serde_json::Map<String, Value>,
    root: &Path,
    ranks: &BTreeMap<String, WorktreeWatchRepoRank>,
    cfg: &WorktreeWatchConfig,
) {
    let key = path_identity_key(root);
    if let Some(rank) = ranks.get(&key) {
        let tier = rank
            .tier
            .clone()
            .unwrap_or_else(|| "unknown".to_string());
        object.insert("watchTier".to_string(), json!(tier));
        object.insert("activityScore".to_string(), json!(rank.score));
        if let Some(r) = rank.rank {
            object.insert("activityRank".to_string(), json!(r));
        }
        if let Some(at) = rank.last_event_at.as_ref() {
            object.insert("lastActivityAt".to_string(), json!(at));
        }
        // Hot is preferred for FS watches; warm only when under max_fs_watches budget.
        let fs_watched = matches!(tier.as_str(), "hot");
        object.insert("fsWatched".to_string(), json!(fs_watched));
        let commit_secs = match tier.as_str() {
            "hot" => cfg.hot_commit_check_seconds(),
            "warm" => cfg.warm_commit_check_seconds(),
            "cold" => cfg.cold_commit_check_seconds(),
            _ => cfg.warm_commit_check_seconds(),
        };
        object.insert("commitCheckSeconds".to_string(), json!(commit_secs));
    } else {
        object.insert("watchTier".to_string(), json!("unknown"));
        object.insert("activityScore".to_string(), json!(0.0));
        object.insert("fsWatched".to_string(), json!(false));
        object.insert(
            "commitCheckSeconds".to_string(),
            json!(cfg.cold_commit_check_seconds()),
        );
    }
}

fn build_resource_insight(
    snapshot: &WorktreeWatchTraySnapshot,
    payloads: &[Value],
    cfg: &WorktreeWatchConfig,
) -> Value {
    let mut hot = 0u64;
    let mut warm = 0u64;
    let mut cold = 0u64;
    let mut unknown = 0u64;
    for payload in payloads {
        match payload
            .get("watchTier")
            .and_then(Value::as_str)
            .unwrap_or("unknown")
        {
            "hot" => hot += 1,
            "warm" => warm += 1,
            "cold" => cold += 1,
            _ => unknown += 1,
        }
    }
    // Hot is preferred; warm only fills remaining budget (see max_fs_watches).
    let estimated_fs_watches = hot
        .saturating_add(warm)
        .min(cfg.max_fs_watches() as u64);

    let mut pids: Vec<u32> = Vec::new();
    if let Some(pid) = snapshot.parent_watcher_pid {
        pids.push(pid);
    }
    for repo in &snapshot.repositories {
        if let Some(pid) = repo.pid {
            if !pids.contains(&pid) {
                pids.push(pid);
            }
        }
    }
    let process = sample_watcher_process_stats(&pids);

    let mut top: Vec<Value> = payloads
        .iter()
        .filter_map(|p| {
            let owner = p
                .get("repositoryOwner")
                .or_else(|| p.get("owner"))
                .and_then(Value::as_str)?;
            let name = p
                .get("repositoryName")
                .or_else(|| p.get("name"))
                .and_then(Value::as_str)?;
            let score = p.get("activityScore").and_then(Value::as_f64).unwrap_or(0.0);
            if score <= 0.0
                && p.get("watchTier").and_then(Value::as_str) == Some("unknown")
            {
                return None;
            }
            Some(json!({
                "repository": format!("{owner}/{name}"),
                "tier": p.get("watchTier").and_then(Value::as_str).unwrap_or("unknown"),
                "score": score,
                "rank": p.get("activityRank"),
                "lastEventAt": p.get("lastActivityAt"),
                "fsWatched": p.get("fsWatched"),
                "commitCheckSeconds": p.get("commitCheckSeconds"),
            }))
        })
        .collect();
    top.sort_by(|a, b| {
        let sa = a.get("score").and_then(Value::as_f64).unwrap_or(0.0);
        let sb = b.get("score").and_then(Value::as_f64).unwrap_or(0.0);
        sb.partial_cmp(&sa)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    top.truncate(12);

    let note = if snapshot.any_running && hot + warm + cold == 0 {
        Some(
            "No repo_ranks yet — start/restart the watcher so it persists activity tiers to ~/.xbp/config.yaml"
                .to_string(),
        )
    } else if !snapshot.any_running {
        Some("Watcher not running — tier data is last known from config ranks".to_string())
    } else {
        None
    };

    json!({
        "tiers": {
            "hot": hot,
            "warm": warm,
            "cold": cold,
            "unknown": unknown,
        },
        "estimatedFsWatches": estimated_fs_watches,
        "repositoryCount": payloads.len(),
        "process": process,
        "schedule": {
            "idleAfterSeconds": cfg.idle_after_seconds(),
            "coldAfterSeconds": cfg.cold_after_seconds(),
            "hotCommitCheckSeconds": cfg.hot_commit_check_seconds(),
            "warmCommitCheckSeconds": cfg.warm_commit_check_seconds(),
            "coldCommitCheckSeconds": cfg.cold_commit_check_seconds(),
            "parentRescanSeconds": cfg.parent_rescan_seconds(),
            "rankPersistSeconds": cfg.rank_persist_seconds(),
            "maxHotRepos": cfg.max_hot_repos(),
            "maxFsWatches": cfg.max_fs_watches(),
            "spoolRetentionDays": cfg.spool_retention_days(),
        },
        "topRepos": top,
        "note": note,
    })
}

fn sample_watcher_process_stats(pids: &[u32]) -> Value {
    let filtered: Vec<u32> = pids
        .iter()
        .copied()
        .filter(|p| *p != 0)
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();
    if filtered.is_empty() {
        return Value::Null;
    }

    let pid_list: Vec<Pid> = filtered.iter().copied().map(Pid::from_u32).collect();
    let mut system = System::new();
    let refresh = ProcessRefreshKind::everything()
        .with_cmd(UpdateKind::Always)
        .with_exe(UpdateKind::Always)
        .with_memory();
    // Two samples so cpu_usage is meaningful.
    system.refresh_processes_specifics(ProcessesToUpdate::Some(&pid_list), true, refresh);
    std::thread::sleep(Duration::from_millis(200));
    system.refresh_processes_specifics(ProcessesToUpdate::Some(&pid_list), true, refresh);

    let mut total_cpu = 0.0_f64;
    let mut total_rss: u64 = 0;
    let mut live = Vec::new();
    for pid in &filtered {
        let Some(proc) = system.process(Pid::from_u32(*pid)) else {
            continue;
        };
        if !process_looks_like_worktree_watch(proc)
            && *pid != std::process::id()
            && !process_cmdline_contains_worktree(proc)
        {
            // Still report raw stats if it matches our tracked watcher pid from state.
        }
        let cpu = proc.cpu_usage() as f64;
        let rss = proc.memory(); // bytes in recent sysinfo
        total_cpu += f64::from(cpu);
        total_rss = total_rss.saturating_add(rss);
        live.push(json!({
            "pid": pid,
            "cpuPercent": (cpu * 10.0).round() / 10.0,
            "rssMb": (rss as f64 / (1024.0 * 1024.0) * 10.0).round() / 10.0,
            "name": proc.name().to_string_lossy(),
        }));
    }

    if live.is_empty() {
        return Value::Null;
    }

    json!({
        "pid": filtered.first().copied(),
        "pids": filtered,
        "cpuPercent": (total_cpu * 10.0).round() / 10.0,
        "rssMb": (total_rss as f64 / (1024.0 * 1024.0) * 10.0).round() / 10.0,
        "processes": live,
    })
}

fn process_cmdline_contains_worktree(process: &sysinfo::Process) -> bool {
    process
        .cmd()
        .iter()
        .any(|part| part.to_string_lossy().to_ascii_lowercase().contains("worktree-watch"))
}

fn truncate_middle(s: &str, max: usize) -> String {
    if s.chars().count() <= max {
        return s.to_string();
    }
    if max < 5 {
        return s.chars().take(max).collect();
    }
    let keep = max.saturating_sub(1) / 2;
    let front: String = s.chars().take(keep).collect();
    let back: String = s
        .chars()
        .rev()
        .take(max - keep - 1)
        .collect::<String>()
        .chars()
        .rev()
        .collect();
    format!("{front}…{back}")
}

fn last_sync_display(payload: &Value) -> (String, String) {
    let Some(last_sync) = payload.get("lastSync") else {
        return ("—".to_string(), "never".to_string());
    };
    let synced_at = last_sync
        .get("syncedAt")
        .and_then(Value::as_str)
        .unwrap_or("unknown");
    let age = if let Some(seconds) = payload
        .get("lastSyncSecondsAgo")
        .and_then(Value::as_u64)
    {
        format_age_seconds(seconds)
    } else if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
        let seconds = (Utc::now() - dt.with_timezone(&Utc)).num_seconds().max(0) as u64;
        format_age_seconds(seconds)
    } else {
        "unknown".to_string()
    };
    // Compact timestamp for the matrix column.
    let compact = if synced_at.len() >= 19 {
        synced_at[11..19].to_string()
    } else {
        synced_at.to_string()
    };
    (compact, age)
}

fn format_age_seconds(seconds: u64) -> String {
    if seconds < 60 {
        format!("{seconds}s ago")
    } else if seconds < 3600 {
        format!("{}m ago", seconds / 60)
    } else if seconds < 86400 {
        format!("{}h ago", seconds / 3600)
    } else {
        format!("{}d ago", seconds / 86400)
    }
}

fn attach_optional_stats(
    mut payload: Value,
    stats: Option<WorktreeWatchStatsSummary>,
    repo_activity: Option<WorktreeWatchRepoActivitySummary>,
) -> Result<Value, String> {
    if let Some(object) = payload.as_object_mut() {
        if let Some(stats) = stats {
            object.insert(
                "stats".to_string(),
                serde_json::to_value(stats)
                    .map_err(|error| format!("Failed to render stats payload: {error}"))?,
            );
        }
        if let Some(repo_activity) = repo_activity {
            object.insert(
                "repositoryActivity".to_string(),
                serde_json::to_value(repo_activity)
                    .map_err(|error| format!("Failed to render repo activity payload: {error}"))?,
            );
        }
    }
    Ok(payload)
}

pub fn spawn_detached_worktree_watch_for_current_repo() -> Result<Option<PathBuf>, String> {
    if is_background_child() {
        return Ok(None);
    }

    let identity = match resolve_repo_identity(None) {
        Ok(identity) => identity,
        Err(_) => return Ok(None),
    };
    spawn_detached_worktree_watch_for_identity(&identity).map(Some)
}

/// Snapshot of worktree-watch storage and watcher state for `xbp diag`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct WorktreeWatchDiagnostic {
    pub global_xbp_root: String,
    pub mutations_root: String,
    pub runtime_key: String,
    pub platform: String,
    pub is_wsl: bool,
    pub repo_owner: Option<String>,
    pub repo_name: Option<String>,
    pub branch: Option<String>,
    pub branch_spool: Option<String>,
    pub branch_spool_exists: bool,
    pub watcher_state_path: Option<String>,
    pub watcher_state_present: bool,
    pub watcher_running: bool,
    pub event_jsonl_files: usize,
    pub commit_jsonl_files: usize,
    pub alternate_spool_roots: Vec<String>,
    pub notes: Vec<String>,
}

/// Lightweight mutation activity for TODO→issue effort attribution.
#[derive(Debug, Clone)]
pub struct MutationActivityEvent {
    pub paths: Vec<String>,
    pub primary_path: Option<String>,
    pub occurred_at: DateTime<Utc>,
    pub added_lines: Option<u64>,
    pub removed_lines: Option<u64>,
    pub head_sha: Option<String>,
    pub branch_name: String,
}

/// Lightweight commit activity for TODO→issue effort attribution.
#[derive(Debug, Clone)]
pub struct CommitActivityEvent {
    pub occurred_at: DateTime<Utc>,
    pub head_sha: String,
    pub subject: Option<String>,
    pub branch_name: String,
    pub repo_root: String,
}

/// Load all spooled mutation events for a GitHub `owner/name` across branch folders.
pub fn load_repo_mutation_activity(
    owner: &str,
    name: &str,
) -> Result<Vec<MutationActivityEvent>, String> {
    let identity = RepoIdentity {
        owner: owner.to_string(),
        name: name.to_string(),
        branch: "main".to_string(),
        root: PathBuf::from("."),
        head_sha: None,
    };
    let mut events = Vec::new();
    for repo_root in repository_spool_search_roots(&identity)? {
        if !repo_root.is_dir() {
            continue;
        }
        for entry in fs::read_dir(&repo_root).map_err(|e| {
            format!(
                "Failed to read mutation spool {}: {e}",
                repo_root.display()
            )
        })? {
            let entry = entry.map_err(|e| format!("Failed to read spool entry: {e}"))?;
            if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
                continue;
            }
            let branch_layout = spool_layout_for_existing_root(&entry.path());
            let candidates = collect_spool_candidates(&branch_layout, true)?;
            for candidate in candidates {
                if let SyncRecords::Events(records) = candidate.records {
                    for record in records {
                        events.push(MutationActivityEvent {
                            paths: record.paths,
                            primary_path: record.primary_path,
                            occurred_at: record.occurred_at,
                            added_lines: record.added_lines,
                            removed_lines: record.removed_lines,
                            head_sha: record.head_sha,
                            branch_name: record.branch_name,
                        });
                    }
                }
            }
        }
    }
    events.sort_by_key(|e| e.occurred_at);
    Ok(events)
}

/// Load all spooled commit events for a GitHub `owner/name`.
pub fn load_repo_commit_activity(
    owner: &str,
    name: &str,
) -> Result<Vec<CommitActivityEvent>, String> {
    let identity = RepoIdentity {
        owner: owner.to_string(),
        name: name.to_string(),
        branch: "main".to_string(),
        root: PathBuf::from("."),
        head_sha: None,
    };
    let mut commits = Vec::new();
    for repo_root in repository_spool_search_roots(&identity)? {
        if !repo_root.is_dir() {
            continue;
        }
        for entry in fs::read_dir(&repo_root).map_err(|e| {
            format!(
                "Failed to read mutation spool {}: {e}",
                repo_root.display()
            )
        })? {
            let entry = entry.map_err(|e| format!("Failed to read spool entry: {e}"))?;
            if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
                continue;
            }
            let branch_layout = spool_layout_for_existing_root(&entry.path());
            let candidates = collect_spool_candidates(&branch_layout, true)?;
            for candidate in candidates {
                if let SyncRecords::Commits(records) = candidate.records {
                    for record in records {
                        commits.push(CommitActivityEvent {
                            occurred_at: record.occurred_at,
                            head_sha: record.head_sha,
                            subject: record.subject,
                            branch_name: record.branch_name,
                            repo_root: record.repo_root,
                        });
                    }
                }
            }
        }
    }
    commits.sort_by_key(|c| c.occurred_at);
    Ok(commits)
}

/// Collect home-level worktree-watch paths and optional current-repo spool status.
pub fn collect_worktree_watch_diagnostic() -> WorktreeWatchDiagnostic {
    let runtime_key = worktree_runtime_key();
    let platform = std::env::consts::OS.to_string();
    let is_wsl = is_wsl_runtime();
    let mut notes = Vec::new();

    let global_xbp_root = match ensure_global_xbp_paths() {
        Ok(paths) => paths.root_dir,
        Err(error) => {
            notes.push(format!("Failed to resolve global XBP home: {error}"));
            resolve_global_xbp_root_dir()
        }
    };
    let mutations_root = global_xbp_root.join("mutations");

    let identity = resolve_repo_identity(None).ok();
    let (
        repo_owner,
        repo_name,
        branch,
        branch_spool,
        branch_spool_exists,
        watcher_state_path,
        watcher_state_present,
        watcher_running,
        event_jsonl_files,
        commit_jsonl_files,
        alternate_spool_roots,
    ) = if let Some(identity) = identity.as_ref() {
        let layout = prepare_spool_layout(identity).ok();
        let branch_spool = layout
            .as_ref()
            .map(|value| value.root.display().to_string());
        let branch_spool_exists = layout
            .as_ref()
            .map(|value| value.root.exists())
            .unwrap_or(false);
        let watcher_state_path = layout
            .as_ref()
            .map(|value| value.watcher_state_file.display().to_string());
        let watcher_state_present = layout
            .as_ref()
            .map(|value| {
                watcher_state_read_candidates(&value.watcher_state_file)
                    .iter()
                    .any(|candidate| candidate.exists())
            })
            .unwrap_or(false);
        let watcher_running = layout
            .as_ref()
            .and_then(|value| read_watcher_state(&value.watcher_state_file).ok().flatten())
            .map(|state| is_matching_watcher_process(&state))
            .unwrap_or(false);

        let mut event_jsonl_files = 0usize;
        let mut commit_jsonl_files = 0usize;
        if let Some(layout) = layout.as_ref() {
            if let Ok(entries) = fs::read_dir(&layout.root) {
                for entry in entries.flatten() {
                    let name = entry.file_name().to_string_lossy().to_string();
                    if name.starts_with("events-") && name.ends_with(".jsonl") {
                        event_jsonl_files += 1;
                    } else if name.starts_with("commits-") && name.ends_with(".jsonl") {
                        commit_jsonl_files += 1;
                    }
                }
            }
        }

        let alternate_spool_roots = repository_spool_search_roots(identity)
            .unwrap_or_default()
            .into_iter()
            .map(|path| {
                path.join(sanitize_path_component(&identity.branch))
                    .display()
                    .to_string()
            })
            .filter(|path| {
                layout
                    .as_ref()
                    .map(|value| path_identity_key(Path::new(path)) != path_identity_key(&value.root))
                    .unwrap_or(true)
            })
            .filter(|path| Path::new(path).exists())
            .collect::<Vec<_>>();

        let root_display = global_xbp_root
            .display()
            .to_string()
            .replace('\\', "/")
            .to_ascii_lowercase();
        if is_wsl && !root_display.starts_with("/mnt/") {
            notes.push(
                "WSL global XBP home is not on a Windows drive mount. If Windows has mutations under C:\\Users\\…\\.xbp, set XBP_HOME=/mnt/c/Users/<you>/.xbp so status/sync share that spool."
                    .to_string(),
            );
        }
        if is_wsl && root_display.starts_with("/mnt/") {
            notes.push(
                "Using a Windows-profile-backed XBP home from WSL; spoils are reconcilable with native Windows."
                    .to_string(),
            );
        }
        if !alternate_spool_roots.is_empty() {
            notes.push(format!(
                "Found {} alternate spool location(s) with historical data (legacy homes / runtime layouts).",
                alternate_spool_roots.len()
            ));
        }

        (
            Some(identity.owner.clone()),
            Some(identity.name.clone()),
            Some(identity.branch.clone()),
            branch_spool,
            branch_spool_exists,
            watcher_state_path,
            watcher_state_present,
            watcher_running,
            event_jsonl_files,
            commit_jsonl_files,
            alternate_spool_roots,
        )
    } else {
        notes.push(
            "Not inside a git repository with a resolvable remote; showing global XBP home only."
                .to_string(),
        );
        (
            None,
            None,
            None,
            None,
            false,
            None,
            false,
            false,
            0,
            0,
            Vec::new(),
        )
    };

    WorktreeWatchDiagnostic {
        global_xbp_root: global_xbp_root.display().to_string(),
        mutations_root: mutations_root.display().to_string(),
        runtime_key,
        platform,
        is_wsl,
        repo_owner,
        repo_name,
        branch,
        branch_spool,
        branch_spool_exists,
        watcher_state_path,
        watcher_state_present,
        watcher_running,
        event_jsonl_files,
        commit_jsonl_files,
        alternate_spool_roots,
        notes,
    }
}

fn resolve_target_identities(
    target: &WorktreeWatchTargetOptions,
) -> Result<Vec<RepoIdentity>, String> {
    if target.repo.is_some() && target.parent.is_some() {
        return Err("Pass either `--repo` or `--parent`, not both.".to_string());
    }
    if !target.repos.is_empty() && target.parent.is_some() {
        return Err("Pass either `--repos` or `--parent`, not both.".to_string());
    }
    if target.repo.is_some() && !target.repos.is_empty() {
        return Err("Pass either `--repo` or `--repos`, not both.".to_string());
    }

    if let Some(parent) = target.parent.as_deref() {
        return discover_repo_identities_under(parent);
    }

    if !target.repos.is_empty() {
        let mut identities = Vec::new();
        let mut seen = BTreeSet::new();
        for repo in &target.repos {
            let identity = resolve_repo_identity(Some(repo.as_path()))?;
            let key = path_identity_key(&identity.root);
            if seen.insert(key) {
                identities.push(identity);
            }
        }
        return Ok(identities);
    }

    resolve_repo_identity(target.repo.as_deref()).map(|identity| vec![identity])
}

/// Collect a tray-friendly snapshot of watcher state + unsynced backlog.
pub fn collect_worktree_watch_tray_snapshot(
    target: &WorktreeWatchTargetOptions,
) -> Result<WorktreeWatchTraySnapshot, String> {
    let identities = resolve_target_identities(target)?;
    let mode = if target.parent.is_some() {
        "parent"
    } else if !target.repos.is_empty() {
        "repos"
    } else {
        "repo"
    }
    .to_string();

    let parent_label = target
        .parent
        .as_ref()
        .map(|path| path.display().to_string());
    let target_label = if let Some(parent) = parent_label.as_ref() {
        format!("parent:{parent}")
    } else if !target.repos.is_empty() {
        format!("{} repo(s)", target.repos.len())
    } else if let Some(repo) = target.repo.as_ref() {
        repo.display().to_string()
    } else {
        identities
            .first()
            .map(|identity| identity.root.display().to_string())
            .unwrap_or_else(|| "current repo".to_string())
    };

    // Collect PIDs first so we only refresh those processes once (System::new_all
    // per repo was freezing status for large parent folders on Windows).
    let mut parent_state: Option<ParentWorktreeWatcherState> = None;
    if let Some(parent) = target.parent.as_deref() {
        if let Ok(parent) = canonical_parent_path(parent) {
            if let Ok(layout) = prepare_parent_spool_layout(&parent) {
                parent_state = read_parent_watcher_state(&layout.watcher_state_file)?;
            }
        }
    }

    let mut repo_states: Vec<(usize, WorktreeWatcherState)> = Vec::new();
    let mut layouts: Vec<SpoolLayout> = Vec::with_capacity(identities.len());
    for (index, identity) in identities.iter().enumerate() {
        let layout = prepare_spool_layout(identity)?;
        if let Some(state) = read_watcher_state(&layout.watcher_state_file)? {
            if same_repo_watcher_state(identity, &state) {
                repo_states.push((index, state));
            }
        }
        layouts.push(layout);
    }

    let mut pids: Vec<u32> = repo_states.iter().map(|(_, state)| state.pid).collect();
    if let Some(state) = parent_state.as_ref() {
        pids.push(state.pid);
    }
    let process_system = process_system_for_pids(&pids);

    let mut parent_watcher_running = false;
    let mut parent_watcher_pid = None;
    if let Some(state) = parent_state.as_ref() {
        // Include self: the loopback API runs inside the watcher process, so
        // `state.pid == std::process::id()` must count as running (the
        // "other process" matcher intentionally excludes self for stop/replace).
        if is_live_parent_watcher_process_with_system(&process_system, state) {
            parent_watcher_running = true;
            parent_watcher_pid = Some(state.pid);
        }
    }

    let mut repositories = Vec::new();
    let mut running_watchers = 0usize;
    let mut total_unsynced_files = 0u64;
    let mut total_unsynced_records = 0u64;

    for (index, identity) in identities.iter().enumerate() {
        let layout = &layouts[index];
        // Line counts only — never fully parse/sanitize JSONL on the hot path.
        let counts = count_spool_summary(layout)?;
        let unsynced_files = counts.unsynced_files;
        let unsynced_records = counts.unsynced_records;
        total_unsynced_files += unsynced_files;
        total_unsynced_records += unsynced_records;

        let (running, pid) = match repo_states
            .iter()
            .find(|(repo_index, _)| *repo_index == index)
            .map(|(_, state)| state)
        {
            Some(state) if is_live_watcher_process_with_system(&process_system, state) => {
                (true, Some(state.pid))
            }
            _ => (false, None),
        };
        // Parent watcher covers repos under parent even without per-repo state.
        let running = running || parent_watcher_running;
        let pid = pid.or(if running {
            parent_watcher_pid
        } else {
            None
        });
        if running {
            running_watchers += 1;
        }

        repositories.push(WorktreeWatchTrayRepoStatus {
            root: normalize_repo_root_for_payload(&identity.root),
            owner: identity.owner.clone(),
            name: identity.name.clone(),
            branch: identity.branch.clone(),
            running,
            pid,
            unsynced_files,
            unsynced_records,
        });
    }

    let repository_count = repositories.len();
    let any_running = parent_watcher_running || running_watchers > 0;
    let all_running = repository_count > 0 && running_watchers >= repository_count;

    // Lightweight summary for tooltips; full stats are computed on demand via the tray menu.
    let stats_line = Some(format!(
        "{repository_count} repo(s) · {running_watchers} watching · {total_unsynced_records} unsynced · {total_unsynced_files} file(s)"
    ));

    Ok(WorktreeWatchTraySnapshot {
        target_label,
        mode,
        parent: parent_label,
        repository_count,
        running_watchers,
        any_running,
        all_running,
        parent_watcher_running,
        parent_watcher_pid,
        total_unsynced_files,
        total_unsynced_records,
        repositories,
        stats_line,
    })
}

/// Default session gap used when the local HTTP API recomputes coding stats.
pub const WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES: u64 = 30;

/// Build the JSON body for `GET /api/worktree-watch/status` (tray snapshot + per-repo status).
pub fn collect_worktree_watch_api_status(
    target: &WorktreeWatchTargetOptions,
    last_error: Option<&str>,
) -> Result<Value, String> {
    let snapshot = collect_worktree_watch_tray_snapshot(target)?;
    let watch_cfg = resolve_worktree_watch_config();
    let rank_index = build_repo_rank_index(&watch_cfg);
    let mut repositories = Vec::new();

    for tray_repo in &snapshot.repositories {
        // Reuse tray counts; attach lastSync without re-scanning spool JSONL bodies.
        let mut status = worktree_watch_status_from_tray_repo(tray_repo)?;
        if let Some(object) = status.as_object_mut() {
            object.insert("running".to_string(), json!(tray_repo.running));
            object.insert("pid".to_string(), json!(tray_repo.pid));
            object.insert("owner".to_string(), json!(tray_repo.owner));
            object.insert("name".to_string(), json!(tray_repo.name));
            object.insert("branch".to_string(), json!(tray_repo.branch));
            object.insert("root".to_string(), json!(tray_repo.root));
            if let Some(spool) = object.get("spoolRoot").cloned() {
                object.insert("spoolPath".to_string(), spool);
            }
            object.insert("lastError".to_string(), Value::Null);
            if let Some(last_sync) = object.get("lastSync") {
                if let Some(synced_at) = last_sync.get("syncedAt").and_then(Value::as_str) {
                    if let Ok(dt) = DateTime::parse_from_rfc3339(synced_at) {
                        let age = (Utc::now() - dt.with_timezone(&Utc))
                            .num_seconds()
                            .max(0) as u64;
                        object.insert("lastSyncSecondsAgo".to_string(), json!(age));
                    }
                }
            }
            let identity = identity_from_tray_repo(tray_repo);
            attach_repo_resource_fields(object, &identity.root, &rank_index, &watch_cfg);
        }
        repositories.push(status);
    }

    repositories.sort_by(|a, b| {
        let sa = a.get("activityScore").and_then(Value::as_f64).unwrap_or(0.0);
        let sb = b.get("activityScore").and_then(Value::as_f64).unwrap_or(0.0);
        sb.partial_cmp(&sa)
            .unwrap_or(std::cmp::Ordering::Equal)
    });

    let resource = build_resource_insight(&snapshot, &repositories, &watch_cfg);

    Ok(json!({
        "targetLabel": snapshot.target_label,
        "mode": snapshot.mode,
        "parent": snapshot.parent,
        "repositoryCount": snapshot.repository_count,
        "runningWatchers": snapshot.running_watchers,
        "anyRunning": snapshot.any_running,
        "allRunning": snapshot.all_running,
        "parentWatcherRunning": snapshot.parent_watcher_running,
        "parentWatcherPid": snapshot.parent_watcher_pid,
        "totalUnsyncedFiles": snapshot.total_unsynced_files,
        "totalUnsyncedRecords": snapshot.total_unsynced_records,
        "unsyncedFiles": snapshot.total_unsynced_files,
        "unsyncedRecords": snapshot.total_unsynced_records,
        "statsLine": snapshot.stats_line,
        "lastError": last_error,
        "resource": resource,
        "repositories": repositories,
    }))
}

fn identity_from_tray_repo(repo: &WorktreeWatchTrayRepoStatus) -> RepoIdentity {
    RepoIdentity {
        owner: repo.owner.clone(),
        name: repo.name.clone(),
        branch: repo.branch.clone(),
        root: parent_root_as_local_path(&repo.root),
        head_sha: None,
    }
}

fn worktree_watch_status_from_tray_repo(
    repo: &WorktreeWatchTrayRepoStatus,
) -> Result<Value, String> {
    let identity = identity_from_tray_repo(repo);
    let layout = prepare_spool_layout(&identity)?;
    let mut payload = json!({
        "repoRoot": repo.root,
        "repositoryOwner": repo.owner,
        "repositoryName": repo.name,
        "branchName": repo.branch,
        "platform": std::env::consts::OS,
        "runtimeKey": worktree_runtime_key(),
        "spoolRoot": layout.root.display().to_string(),
        "unsyncedFiles": repo.unsynced_files,
        "unsyncedRecords": repo.unsynced_records,
        "syncedFiles": 0,
        "syncedRecords": 0,
        "localSpoolFiles": repo.unsynced_files,
        "localSpoolRecords": repo.unsynced_records,
        "running": repo.running,
        "pid": repo.pid,
    });

    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "lastSync".to_string(),
                serde_json::to_value(last_sync)
                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
            );
        }
    }

    Ok(payload)
}

/// Build the JSON body for `GET /api/worktree-watch/repositories`.
pub fn collect_worktree_watch_api_repositories(
    target: &WorktreeWatchTargetOptions,
) -> Result<Value, String> {
    let status = collect_worktree_watch_api_status(target, None)?;
    let repositories = status
        .get("repositories")
        .cloned()
        .unwrap_or_else(|| json!([]));
    Ok(json!({ "repositories": repositories }))
}

/// Build the JSON body for `GET /api/worktree-watch/stats`.
///
/// Prefers cached `stats.json` when fresh so the dashboard poll stays cheap;
/// recomputes on miss/stale (same shape as `xbp worktree-watch status --stats`).
pub fn collect_worktree_watch_api_stats(
    target: &WorktreeWatchTargetOptions,
    session_gap_minutes: u64,
) -> Result<Value, String> {
    let identities = resolve_target_identities(target)?;
    if identities.is_empty() {
        return Ok(json!({
            "generatedAt": null,
            "message": "no stats yet",
            "repositories": [],
        }));
    }

    let gap = if session_gap_minutes == 0 {
        WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES
    } else {
        session_gap_minutes
    };

    let mut repositories = Vec::new();
    for identity in &identities {
        let stats = read_or_generate_stats(identity, gap, Duration::from_secs(120))?;
        repositories.push(json!({
            "owner": identity.owner,
            "name": identity.name,
            "branch": identity.branch,
            "repositoryOwner": identity.owner,
            "repositoryName": identity.name,
            "branchName": identity.branch,
            "repoRoot": normalize_repo_root_for_payload(&identity.root),
            "root": normalize_repo_root_for_payload(&identity.root),
            "stats": stats,
        }));
    }

    if repositories.len() == 1 {
        let first = repositories.remove(0);
        let stats = first
            .get("stats")
            .cloned()
            .unwrap_or(Value::Null);
        // Flat WorktreeWatchStatsSummary for single-repo targets (dashboard accepts both).
        if let Some(mut object) = stats.as_object().cloned() {
            object.insert(
                "repositories".to_string(),
                json!([{
                    "owner": first.get("owner"),
                    "name": first.get("name"),
                    "branch": first.get("branch"),
                    "stats": first.get("stats"),
                }]),
            );
            return Ok(Value::Object(object));
        }
        return Ok(first);
    }

    Ok(json!({
        "repositories": repositories,
        "repositoryCount": repositories.len(),
    }))
}

/// Fast status payload: spool file/line counts + lastSync, no full JSONL decode.
fn worktree_watch_status_payload_light(identity: &RepoIdentity) -> Result<Value, String> {
    let layout = prepare_spool_layout(identity)?;
    let counts = count_spool_summary(&layout)?;

    let mut payload = json!({
        "repoRoot": normalize_repo_root_for_payload(&identity.root),
        "repositoryOwner": identity.owner.clone(),
        "repositoryName": identity.name.clone(),
        "branchName": identity.branch.clone(),
        "platform": std::env::consts::OS,
        "runtimeKey": worktree_runtime_key(),
        "spoolRoot": layout.root.display().to_string(),
        "unsyncedFiles": counts.unsynced_files,
        "unsyncedRecords": counts.unsynced_records,
        "syncedFiles": counts.synced_files,
        "syncedRecords": counts.synced_records,
        "localSpoolFiles": counts.unsynced_files + counts.synced_files,
        "localSpoolRecords": counts.unsynced_records + counts.synced_records,
    });

    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "lastSync".to_string(),
                serde_json::to_value(last_sync)
                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
            );
        }
    }

    Ok(payload)
}

fn worktree_watch_status_payload(
    identity: &RepoIdentity,
    include_records: bool,
    record_limit: usize,
) -> Result<Value, String> {
    // Counts are always available via the light path; only decode JSONL when
    // the caller needs individual records.
    if !include_records {
        return worktree_watch_status_payload_light(identity);
    }

    let layout = prepare_spool_layout(identity)?;
    let candidates = collect_sync_candidates(&layout)?;
    let all_candidates = collect_spool_candidates(&layout, true)?;
    let record_count: usize = candidates
        .iter()
        .map(|candidate| candidate.records.len())
        .sum();
    let all_record_count: usize = all_candidates
        .iter()
        .map(|candidate| candidate.records.len())
        .sum();
    let synced_file_count = all_candidates.len().saturating_sub(candidates.len());
    let synced_record_count = all_record_count.saturating_sub(record_count);

    let mut payload = json!({
        "repoRoot": normalize_repo_root_for_payload(&identity.root),
        "repositoryOwner": identity.owner.clone(),
        "repositoryName": identity.name.clone(),
        "branchName": identity.branch.clone(),
        "platform": std::env::consts::OS,
        "runtimeKey": worktree_runtime_key(),
        "spoolRoot": layout.root.display().to_string(),
        "unsyncedFiles": candidates.len(),
        "unsyncedRecords": record_count,
        "syncedFiles": synced_file_count,
        "syncedRecords": synced_record_count,
        "localSpoolFiles": all_candidates.len(),
        "localSpoolRecords": all_record_count,
    });

    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "lastSync".to_string(),
                serde_json::to_value(last_sync)
                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
            );
        }
    }

    if let Some(object) = payload.as_object_mut() {
        object.insert(
            "records".to_string(),
            collect_status_records(&candidates, record_limit)?,
        );
    }

    Ok(payload)
}

#[derive(Debug, Default, Clone, Copy)]
struct SpoolCountSummary {
    unsynced_files: u64,
    unsynced_records: u64,
    synced_files: u64,
    synced_records: u64,
}

/// Count spool JSONL files and non-empty lines without parsing or rewriting.
fn count_spool_summary(layout: &SpoolLayout) -> Result<SpoolCountSummary, String> {
    let mut roots = vec![layout.root.clone()];
    if let Some(branch) = layout.root.file_name().and_then(|name| name.to_str()) {
        if let Some(repo_root) = layout.root.parent() {
            if let (Some(name), Some(owner)) = (
                repo_root.file_name().and_then(|n| n.to_str()),
                repo_root
                    .parent()
                    .and_then(|p| p.file_name())
                    .and_then(|n| n.to_str()),
            ) {
                let identity = RepoIdentity {
                    owner: owner.to_string(),
                    name: name.to_string(),
                    branch: branch.to_string(),
                    root: PathBuf::new(),
                    head_sha: None,
                };
                if let Ok(search_roots) = repository_spool_search_roots(&identity) {
                    for search_root in search_roots {
                        let branch_root = search_root.join(branch);
                        if !roots.iter().any(|existing| {
                            path_identity_key(existing) == path_identity_key(&branch_root)
                        }) {
                            roots.push(branch_root);
                        }
                    }
                }
            }
        }
    }

    let mut summary = SpoolCountSummary::default();
    let mut seen_files = BTreeSet::new();
    for root in roots {
        if !root.exists() {
            continue;
        }
        for entry in fs::read_dir(&root).map_err(|error| {
            format!(
                "Failed to read spool directory {}: {error}",
                root.display()
            )
        })? {
            let path = entry
                .map_err(|error| format!("Failed to read spool entry: {error}"))?
                .path();
            let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
                continue;
            };
            if !file_name.ends_with(".jsonl") {
                continue;
            }
            if !(file_name.starts_with("events-") || file_name.starts_with("commits-")) {
                continue;
            }
            let identity_key = path_identity_key(&path);
            if !seen_files.insert(identity_key) {
                continue;
            }
            let lines = count_nonempty_lines(&path)?;
            if file_name.contains(".synced.") {
                summary.synced_files += 1;
                summary.synced_records += lines;
            } else {
                summary.unsynced_files += 1;
                summary.unsynced_records += lines;
            }
        }
    }
    Ok(summary)
}

fn count_nonempty_lines(path: &Path) -> Result<u64, String> {
    let file = File::open(path)
        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
    let reader = BufReader::new(file);
    let mut count = 0u64;
    for line in reader.lines() {
        let line =
            line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
        if !line.trim().is_empty() {
            count += 1;
        }
    }
    Ok(count)
}

fn read_or_generate_stats(
    identity: &RepoIdentity,
    session_gap_minutes: u64,
    max_age: Duration,
) -> Result<WorktreeWatchStatsSummary, String> {
    let layout = prepare_spool_layout(identity)?;
    if layout.stats_file.exists() {
        if let Ok(raw) = fs::read_to_string(&layout.stats_file) {
            if let Ok(stats) = serde_json::from_str::<WorktreeWatchStatsSummary>(&raw) {
                let age = Utc::now()
                    .signed_duration_since(stats.generated_at)
                    .to_std()
                    .unwrap_or(Duration::from_secs(u64::MAX));
                if age <= max_age {
                    return Ok(stats);
                }
            }
        }
    }
    generate_and_store_stats(identity, session_gap_minutes)
}

fn collect_status_records(candidates: &[SyncCandidate], limit: usize) -> Result<Value, String> {
    let available: usize = candidates
        .iter()
        .map(|candidate| candidate.records.len())
        .sum();
    let mut shown = 0usize;
    let mut events = Vec::new();
    let mut commits = Vec::new();

    'candidates: for candidate in candidates {
        match &candidate.records {
            SyncRecords::Events(records) => {
                for event in records {
                    if shown >= limit {
                        break 'candidates;
                    }
                    events.push(json!({
                        "sourceFile": candidate.path.display().to_string(),
                        "occurredAt": event.occurred_at,
                        "eventKind": event.event_kind.clone(),
                        "primaryPath": event.primary_path.clone(),
                        "oldPath": event.old_path.clone(),
                        "newPath": event.new_path.clone(),
                        "paths": event.paths.clone(),
                        "addedLines": event.added_lines,
                        "removedLines": event.removed_lines,
                        "totalLines": event.total_lines,
                        "headSha": event.head_sha.clone(),
                        "rawKind": event.raw_kind.clone(),
                    }));
                    shown += 1;
                }
            }
            SyncRecords::Commits(records) => {
                for commit in records {
                    if shown >= limit {
                        break 'candidates;
                    }
                    commits.push(json!({
                        "sourceFile": candidate.path.display().to_string(),
                        "occurredAt": commit.occurred_at,
                        "headSha": commit.head_sha.clone(),
                        "previousHeadSha": commit.previous_head_sha.clone(),
                        "subject": commit.subject.clone(),
                        "authorName": commit.author_name.clone(),
                        "authorEmail": commit.author_email.clone(),
                        "committedAt": commit.committed_at.clone(),
                        "filesChanged": commit.files_changed.clone(),
                        "filesChangedCount": commit.files_changed_count,
                        "insertions": commit.insertions,
                        "deletions": commit.deletions,
                        "isMerge": commit.is_merge,
                        "includedCommits": commit.included_commits.clone(),
                    }));
                    shown += 1;
                }
            }
        }
    }

    Ok(json!({
        "available": available,
        "shown": shown,
        "truncated": shown < available,
        "events": events,
        "commits": commits,
    }))
}

fn print_status_records(payload: &Value) {
    let Some(records) = payload.get("records") else {
        return;
    };
    let shown = records.get("shown").and_then(Value::as_u64).unwrap_or(0);
    if shown == 0 {
        println!("records: none");
        return;
    }

    println!("records:");
    if let Some(events) = records.get("events").and_then(Value::as_array) {
        if !events.is_empty() {
            println!("  mutations:");
            for event in events {
                print_status_event_record(event);
            }
        }
    }
    if let Some(commits) = records.get("commits").and_then(Value::as_array) {
        if !commits.is_empty() {
            println!("  commits:");
            for commit in commits {
                print_status_commit_record(commit);
            }
        }
    }
    if records
        .get("truncated")
        .and_then(Value::as_bool)
        .unwrap_or(false)
    {
        let available = records
            .get("available")
            .and_then(Value::as_u64)
            .unwrap_or(shown);
        println!("  showing {shown} of {available} record(s)");
    }
}

fn print_status_stats(payload: &Value) {
    let Some(stats) = payload.get("stats") else {
        return;
    };
    println!("stats:");
    println!(
        "  coding time: {}",
        format_duration(
            stats
                .get("estimatedCodingSeconds")
                .and_then(Value::as_u64)
                .unwrap_or(0)
        )
    );
    println!(
        "  sessions: {}",
        stats
            .get("sessionCount")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    println!(
        "  observed span: {}",
        format_duration(
            stats
                .get("observedSpanSeconds")
                .and_then(Value::as_u64)
                .unwrap_or(0)
        )
    );
    println!(
        "  events: {}",
        stats
            .get("totalEvents")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    println!(
        "  commits: {}",
        stats
            .get("totalCommits")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    println!(
        "  lines: +{} -{}",
        stats.get("addedLines").and_then(Value::as_u64).unwrap_or(0),
        stats
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    if let Some(spool_root) = value_str(stats, "spoolRoot") {
        println!(
            "  stored: {}",
            Path::new(spool_root).join("stats.json").display()
        );
    }
    print_stats_bucket_section(stats, "byFiletype", "  by filetype:", 10);
    print_stats_bucket_section(stats, "byEventKind", "  by event kind:", 10);
    print_stats_file_section(stats, 10);
}

fn print_repo_activity(payload: &Value) {
    let Some(activity) = payload.get("repositoryActivity") else {
        return;
    };
    println!("repo activity:");
    println!(
        "  coding time: {}",
        format_duration(
            activity
                .get("estimatedCodingSeconds")
                .and_then(Value::as_u64)
                .unwrap_or(0)
        )
    );
    println!(
        "  observed span: {}",
        format_duration(
            activity
                .get("observedSpanSeconds")
                .and_then(Value::as_u64)
                .unwrap_or(0)
        )
    );
    println!(
        "  branches: {}",
        activity
            .get("branchCount")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    println!(
        "  sessions: {}, events: {}, commits: {}, lines: +{} -{}",
        activity
            .get("sessionCount")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        activity
            .get("totalEvents")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        activity
            .get("totalCommits")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        activity
            .get("addedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        activity
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    if let Some(stats_file) = value_str(activity, "statsFile") {
        println!("  stored: {stats_file}");
    }
    let Some(branches) = activity.get("branches").and_then(Value::as_array) else {
        return;
    };
    if branches.is_empty() {
        return;
    }
    println!("  by branch:");
    for branch in branches.iter().take(20) {
        let name = value_str(branch, "branchName").unwrap_or("unknown");
        let coding_seconds = branch
            .get("estimatedCodingSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let observed_seconds = branch
            .get("observedSpanSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let sessions = branch
            .get("sessionCount")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let events = branch
            .get("totalEvents")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let commits = branch
            .get("totalCommits")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        println!(
            "    {name}: {} coding, {} span, {} session(s), {} event(s), {} commit(s)",
            format_duration(coding_seconds),
            format_duration(observed_seconds),
            sessions,
            events,
            commits
        );
    }
}

#[allow(dead_code)]
fn print_last_sync(payload: &Value) {
    let Some(last_sync) = payload.get("lastSync") else {
        return;
    };
    let synced_at = value_str(last_sync, "syncedAt").unwrap_or("unknown-time");
    let status = last_sync
        .get("statusCode")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let events = last_sync
        .get("eventCount")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let commits = last_sync
        .get("commitCount")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let files = last_sync
        .get("spoolFileCount")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let resync = last_sync
        .get("resync")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    println!(
        "last sync: {synced_at}, HTTP {status}, {events} event(s), {commits} commit(s), {files} file(s), resync={resync}"
    );
}

fn print_stats_bucket_section(stats: &Value, key: &str, title: &str, limit: usize) {
    let Some(items) = stats.get(key).and_then(Value::as_array) else {
        return;
    };
    if items.is_empty() {
        return;
    }
    println!("{title}");
    for item in items.iter().take(limit) {
        let name = value_str(item, "name").unwrap_or("unknown");
        let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
        let seconds = item
            .get("estimatedCodingSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
        let removed = item
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        println!(
            "    {name}: {}, {} event(s), +{} -{}",
            format_duration(seconds),
            events,
            added,
            removed
        );
    }
}

fn print_stats_file_section(stats: &Value, limit: usize) {
    let Some(items) = stats.get("byFile").and_then(Value::as_array) else {
        return;
    };
    if items.is_empty() {
        return;
    }
    println!("  by file:");
    for item in items.iter().take(limit) {
        let path = value_str(item, "path").unwrap_or("(no path)");
        let filetype = value_str(item, "filetype").unwrap_or("(none)");
        let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
        let seconds = item
            .get("estimatedCodingSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
        let removed = item
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        println!(
            "    {path} [{filetype}]: {}, {} event(s), +{} -{}",
            format_duration(seconds),
            events,
            added,
            removed
        );
    }
}

fn print_status_event_record(event: &Value) {
    let occurred_at = value_str(event, "occurredAt").unwrap_or("unknown-time");
    let kind = value_str(event, "eventKind").unwrap_or("event");
    let primary_path = event_display_path(event);
    let added = event.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
    let removed = event
        .get("removedLines")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let total_lines = event.get("totalLines").and_then(Value::as_u64);
    if let Some(total_lines) = total_lines {
        println!(
            "    {occurred_at} {kind} {primary_path} (+{added} -{removed}, {total_lines} lines)"
        );
    } else {
        println!("    {occurred_at} {kind} {primary_path} (+{added} -{removed})");
    }

    if let Some(head_sha) = value_str(event, "headSha") {
        println!("      head: {}", short_sha(head_sha));
    }
    if let Some(paths) = event.get("paths").and_then(Value::as_array) {
        if paths.len() > 1 {
            let rendered = paths
                .iter()
                .filter_map(Value::as_str)
                .collect::<Vec<_>>()
                .join(", ");
            println!("      paths: {rendered}");
        }
    }
}

fn print_status_commit_record(commit: &Value) {
    let occurred_at = value_str(commit, "occurredAt").unwrap_or("unknown-time");
    let head_sha = value_str(commit, "headSha")
        .map(short_sha)
        .unwrap_or_else(|| "unknown".to_string());
    let subject = value_str(commit, "subject").unwrap_or("(no subject)");
    println!("    {occurred_at} {head_sha} {subject}");

    if let Some(author) = value_str(commit, "authorName") {
        println!("      author: {author}");
    }
    let files = commit
        .get("filesChangedCount")
        .and_then(Value::as_u64)
        .or_else(|| {
            commit
                .get("filesChanged")
                .and_then(Value::as_array)
                .map(|a| a.len() as u64)
        });
    let ins = commit.get("insertions").and_then(Value::as_u64);
    let del = commit.get("deletions").and_then(Value::as_u64);
    match (files, ins, del) {
        (Some(f), Some(i), Some(d)) => {
            println!("      changes: {f} file(s), +{i} -{d}");
        }
        (Some(f), _, _) => println!("      changes: {f} file(s)"),
        _ => {}
    }
    if commit.get("isMerge").and_then(Value::as_bool) == Some(true) {
        println!("      merge: true");
    }
}

fn event_display_path(event: &Value) -> String {
    let old_path = value_str(event, "oldPath");
    let new_path = value_str(event, "newPath");
    if let (Some(old_path), Some(new_path)) = (old_path, new_path) {
        return format!("{old_path} -> {new_path}");
    }
    value_str(event, "primaryPath")
        .map(str::to_string)
        .or_else(|| {
            event
                .get("paths")
                .and_then(Value::as_array)
                .and_then(|paths| paths.first())
                .and_then(Value::as_str)
                .map(str::to_string)
        })
        .unwrap_or_else(|| "(no path)".to_string())
}

fn value_str<'a>(value: &'a Value, key: &str) -> Option<&'a str> {
    value.get(key).and_then(Value::as_str)
}

fn short_sha(value: &str) -> String {
    value.chars().take(12).collect()
}

fn generate_and_store_stats(
    identity: &RepoIdentity,
    session_gap_minutes: u64,
) -> Result<WorktreeWatchStatsSummary, String> {
    let layout = prepare_spool_layout(identity)?;
    let candidates = collect_spool_candidates(&layout, true)?;
    let summary = build_stats_summary(identity, &layout, &candidates, session_gap_minutes)?;
    let rendered = serde_json::to_string_pretty(&summary)
        .map_err(|error| format!("Failed to serialize stats: {error}"))?;
    fs::write(&layout.stats_file, format!("{rendered}\n")).map_err(|error| {
        format!(
            "Failed to write stats file {}: {error}",
            layout.stats_file.display()
        )
    })?;
    Ok(summary)
}

fn generate_and_store_repo_activity(
    identity: &RepoIdentity,
    session_gap_minutes: u64,
) -> Result<WorktreeWatchRepoActivitySummary, String> {
    let repository_spool_root = repository_spool_root(identity)?;
    let summary =
        build_repo_activity_summary(identity, &repository_spool_root, session_gap_minutes)?;
    let rendered = serde_json::to_string_pretty(&summary)
        .map_err(|error| format!("Failed to serialize repo activity: {error}"))?;
    fs::create_dir_all(&repository_spool_root).map_err(|error| {
        format!(
            "Failed to create repository spool root {}: {error}",
            repository_spool_root.display()
        )
    })?;
    let stats_file = repository_spool_root.join("repo-activity.json");
    fs::write(&stats_file, format!("{rendered}\n")).map_err(|error| {
        format!(
            "Failed to write repo activity file {}: {error}",
            stats_file.display()
        )
    })?;
    Ok(summary)
}

fn build_repo_activity_summary(
    identity: &RepoIdentity,
    repository_spool_root: &Path,
    session_gap_minutes: u64,
) -> Result<WorktreeWatchRepoActivitySummary, String> {
    flush_jsonl_write_buffers();
    let mut branches = Vec::new();
    if repository_spool_root.exists() {
        for entry in fs::read_dir(repository_spool_root).map_err(|error| {
            format!(
                "Failed to read repository spool root {}: {error}",
                repository_spool_root.display()
            )
        })? {
            let entry =
                entry.map_err(|error| format!("Failed to read repository spool entry: {error}"))?;
            let file_type = entry.file_type().map_err(|error| {
                format!("Failed to inspect {}: {error}", entry.path().display())
            })?;
            if !file_type.is_dir() {
                continue;
            }
            let branch_name = entry.file_name().to_string_lossy().to_string();
            let branch_root = entry.path();
            let branch_layout = spool_layout_for_existing_root(&branch_root);
            let branch_identity = RepoIdentity {
                branch: branch_name.clone(),
                ..identity.clone()
            };
            let candidates = collect_spool_candidates(&branch_layout, true)?;
            let stats = build_stats_summary(
                &branch_identity,
                &branch_layout,
                &candidates,
                session_gap_minutes,
            )?;
            if stats.total_events == 0 && stats.total_commits == 0 {
                continue;
            }
            branches.push(WorktreeWatchBranchActivityStats {
                branch_name,
                spool_root: branch_root.display().to_string(),
                total_events: stats.total_events,
                total_commits: stats.total_commits,
                first_event_at: stats.first_event_at,
                last_event_at: stats.last_event_at,
                session_count: stats.session_count,
                observed_span_seconds: stats.observed_span_seconds,
                estimated_coding_seconds: stats.estimated_coding_seconds,
                added_lines: stats.added_lines,
                removed_lines: stats.removed_lines,
            });
        }
    }

    branches.sort_by(|left, right| {
        right
            .estimated_coding_seconds
            .cmp(&left.estimated_coding_seconds)
            .then_with(|| right.total_events.cmp(&left.total_events))
            .then_with(|| left.branch_name.cmp(&right.branch_name))
    });

    let first_event_at = branches
        .iter()
        .filter_map(|branch| branch.first_event_at)
        .min();
    let last_event_at = branches
        .iter()
        .filter_map(|branch| branch.last_event_at)
        .max();
    let observed_span_seconds = observed_span_seconds(first_event_at, last_event_at);
    let stats_file = repository_spool_root.join("repo-activity.json");

    Ok(WorktreeWatchRepoActivitySummary {
        generated_at: Utc::now(),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        repo_root: identity.root.display().to_string(),
        repository_spool_root: repository_spool_root.display().to_string(),
        stats_file: stats_file.display().to_string(),
        session_gap_minutes,
        branch_count: branches.len() as u64,
        total_events: branches.iter().map(|branch| branch.total_events).sum(),
        total_commits: branches.iter().map(|branch| branch.total_commits).sum(),
        first_event_at,
        last_event_at,
        session_count: branches.iter().map(|branch| branch.session_count).sum(),
        observed_span_seconds,
        estimated_coding_seconds: branches
            .iter()
            .map(|branch| branch.estimated_coding_seconds)
            .sum(),
        added_lines: branches.iter().map(|branch| branch.added_lines).sum(),
        removed_lines: branches.iter().map(|branch| branch.removed_lines).sum(),
        branches,
    })
}

fn build_stats_summary(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    candidates: &[SyncCandidate],
    session_gap_minutes: u64,
) -> Result<WorktreeWatchStatsSummary, String> {
    let mut events = Vec::new();
    let mut total_commits = 0u64;
    for candidate in candidates {
        match &candidate.records {
            SyncRecords::Events(records) => events.extend(records.iter().cloned()),
            SyncRecords::Commits(records) => total_commits += records.len() as u64,
        }
    }
    events.sort_by_key(|event| event.occurred_at);

    let gap_seconds = session_gap_minutes.saturating_mul(60).max(60);
    let mut previous_at: Option<DateTime<Utc>> = None;
    let mut session_count = 0u64;
    let mut total_seconds = 0u64;
    let mut total_added = 0u64;
    let mut total_removed = 0u64;
    let mut by_file: BTreeMap<String, WorktreeWatchFileStats> = BTreeMap::new();
    let mut by_filetype: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();
    let mut by_event_kind: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();

    for event in &events {
        if starts_new_coding_session(previous_at, event.occurred_at, gap_seconds) {
            session_count += 1;
        }
        let seconds = coding_seconds_for_event(previous_at, event.occurred_at, gap_seconds);
        previous_at = Some(event.occurred_at);
        let added = event.added_lines.unwrap_or(0);
        let removed = event.removed_lines.unwrap_or(0);
        let path = stats_event_path(event);
        let filetype = filetype_for_path(&path);

        total_seconds += seconds;
        total_added += added;
        total_removed += removed;
        update_file_stats(&mut by_file, &path, &filetype, seconds, added, removed);
        update_bucket_stats(&mut by_filetype, &filetype, seconds, added, removed);
        update_bucket_stats(
            &mut by_event_kind,
            &event.event_kind,
            seconds,
            added,
            removed,
        );
    }

    Ok(WorktreeWatchStatsSummary {
        generated_at: Utc::now(),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        spool_root: layout.root.display().to_string(),
        session_gap_minutes,
        total_events: events.len() as u64,
        total_commits,
        first_event_at: events.first().map(|event| event.occurred_at),
        last_event_at: events.last().map(|event| event.occurred_at),
        session_count,
        observed_span_seconds: observed_span_seconds(
            events.first().map(|event| event.occurred_at),
            events.last().map(|event| event.occurred_at),
        ),
        estimated_coding_seconds: total_seconds,
        added_lines: total_added,
        removed_lines: total_removed,
        by_file: sorted_file_stats(by_file),
        by_filetype: sorted_bucket_stats(by_filetype),
        by_event_kind: sorted_bucket_stats(by_event_kind),
    })
}

fn coding_seconds_for_event(
    previous_at: Option<DateTime<Utc>>,
    occurred_at: DateTime<Utc>,
    gap_seconds: u64,
) -> u64 {
    let Some(previous_at) = previous_at else {
        return 60;
    };
    let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
    if delta <= 0 {
        0
    } else {
        (delta as u64).min(gap_seconds)
    }
}

fn starts_new_coding_session(
    previous_at: Option<DateTime<Utc>>,
    occurred_at: DateTime<Utc>,
    gap_seconds: u64,
) -> bool {
    let Some(previous_at) = previous_at else {
        return true;
    };
    let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
    delta > gap_seconds as i64
}

fn observed_span_seconds(
    first_event_at: Option<DateTime<Utc>>,
    last_event_at: Option<DateTime<Utc>>,
) -> u64 {
    let (Some(first_event_at), Some(last_event_at)) = (first_event_at, last_event_at) else {
        return 0;
    };
    last_event_at
        .signed_duration_since(first_event_at)
        .num_seconds()
        .max(0) as u64
}

fn update_file_stats(
    by_file: &mut BTreeMap<String, WorktreeWatchFileStats>,
    path: &str,
    filetype: &str,
    seconds: u64,
    added: u64,
    removed: u64,
) {
    let stats = by_file
        .entry(path.to_string())
        .or_insert_with(|| WorktreeWatchFileStats {
            path: path.to_string(),
            filetype: filetype.to_string(),
            ..WorktreeWatchFileStats::default()
        });
    stats.event_count += 1;
    stats.estimated_coding_seconds += seconds;
    stats.added_lines += added;
    stats.removed_lines += removed;
}

fn update_bucket_stats(
    buckets: &mut BTreeMap<String, WorktreeWatchBucketStats>,
    name: &str,
    seconds: u64,
    added: u64,
    removed: u64,
) {
    let stats = buckets
        .entry(name.to_string())
        .or_insert_with(|| WorktreeWatchBucketStats {
            name: name.to_string(),
            ..WorktreeWatchBucketStats::default()
        });
    stats.event_count += 1;
    stats.estimated_coding_seconds += seconds;
    stats.added_lines += added;
    stats.removed_lines += removed;
}

fn sorted_file_stats(
    by_file: BTreeMap<String, WorktreeWatchFileStats>,
) -> Vec<WorktreeWatchFileStats> {
    let mut stats = by_file.into_values().collect::<Vec<_>>();
    stats.sort_by(|left, right| {
        right
            .estimated_coding_seconds
            .cmp(&left.estimated_coding_seconds)
            .then_with(|| right.event_count.cmp(&left.event_count))
            .then_with(|| left.path.cmp(&right.path))
    });
    stats
}

fn sorted_bucket_stats(
    buckets: BTreeMap<String, WorktreeWatchBucketStats>,
) -> Vec<WorktreeWatchBucketStats> {
    let mut stats = buckets.into_values().collect::<Vec<_>>();
    stats.sort_by(|left, right| {
        right
            .estimated_coding_seconds
            .cmp(&left.estimated_coding_seconds)
            .then_with(|| right.event_count.cmp(&left.event_count))
            .then_with(|| left.name.cmp(&right.name))
    });
    stats
}

fn stats_event_path(event: &WorktreeMutationEvent) -> String {
    event
        .new_path
        .as_deref()
        .or(event.primary_path.as_deref())
        .or_else(|| event.paths.first().map(String::as_str))
        .unwrap_or("(no path)")
        .to_string()
}

fn filetype_for_path(path: &str) -> String {
    Path::new(path)
        .extension()
        .and_then(|value| value.to_str())
        .map(|value| value.to_ascii_lowercase())
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "(none)".to_string())
}

fn format_duration(seconds: u64) -> String {
    let hours = seconds / 3600;
    let minutes = (seconds % 3600) / 60;
    let remaining_seconds = seconds % 60;
    if hours > 0 {
        format!("{hours}h {minutes}m")
    } else if minutes > 0 {
        format!("{minutes}m {remaining_seconds}s")
    } else {
        format!("{remaining_seconds}s")
    }
}

async fn watch_foreground(
    identity: RepoIdentity,
    layout: SpoolLayout,
    sync_interval_seconds: u64,
) -> Result<(), String> {
    // Refresh watcher state so status sees this process (foreground + detach child).
    if let Ok(executable) = std::env::current_exe() {
        let _ = write_watcher_state(&identity, &layout, std::process::id(), &executable);
    }

    // Local dashboard API (loopback). Background thread avoids an async type
    // cycle with POST /start handlers that call detached start.
    let api_target = WorktreeWatchTargetOptions {
        repo: Some(identity.root.clone()),
        parent: None,
        repos: Vec::new(),
    };
    if let Some(port) = crate::commands::worktree_watch_api::spawn_worktree_watch_api_in_background(
        api_target,
        sync_interval_seconds,
        None,
    ) {
        if !is_background_child() {
            crate::commands::worktree_watch_api::print_api_listen_hint(port);
        } else {
            tracing::info!(
                "worktree-watch local API listening on http://127.0.0.1:{port}"
            );
        }
    }

    let watch_cfg = resolve_worktree_watch_config();
    let (tx, rx) = mpsc::channel();
    let mut watcher = create_path_watcher(&identity.root, tx)?;
    watcher
        .watch(&identity.root, RecursiveMode::Recursive)
        .map_err(|error| {
            format!(
                "Failed to watch repository root {}: {error}",
                identity.root.display()
            )
        })?;
    log_watcher_backend(&identity.root, "repository");

    let mut last_head = identity.head_sha.clone();
    let mut last_activity = Instant::now();
    let mut last_commit_check = Instant::now();
    let mut last_sync = Instant::now();
    let mut last_rank_persist = Instant::now();
    let mut last_housekeeping = Instant::now();
    let mut activity_score = seed_activity_score(&watch_cfg, &identity.root);
    let mut fs_watched = true;
    let mut event_dedupe = RecentEventDedupe {
        fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
    };
    let retention =
        Duration::from_secs(watch_cfg.spool_retention_days().saturating_mul(86_400));
    let _ = prune_spool_layout(&layout, retention);

    loop {
        let idle_secs = last_activity.elapsed().as_secs();
        let tier = tier_from_idle(idle_secs, &watch_cfg);
        let timeout = event_loop_timeout(tier);

        // Cold single-repo: release FS watch so the directory can be deleted/moved.
        if tier == RepoWatchTier::Cold && fs_watched {
            let _ = watcher.unwatch(&identity.root);
            fs_watched = false;
            tracing::info!(
                "worktree-watch: unwatched idle repo {} (releases file locks)",
                identity.root.display()
            );
        } else if tier != RepoWatchTier::Cold && !fs_watched {
            if identity.root.is_dir() {
                if watcher
                    .watch(&identity.root, RecursiveMode::Recursive)
                    .is_ok()
                {
                    fs_watched = true;
                    tracing::info!(
                        "worktree-watch: re-watched active repo {}",
                        identity.root.display()
                    );
                }
            }
        }

        match rx.recv_timeout(timeout) {
            Ok(first) => {
                for item in drain_notify_batch(&rx, first) {
                    match item {
                        Ok(event) => {
                            if let Some(mutation) = mutation_event_from_notify(&identity, event) {
                                if append_unique_mutation_event(
                                    &layout,
                                    &mutation,
                                    &mut event_dedupe,
                                )? {
                                    last_activity = Instant::now();
                                    activity_score = bump_activity_score(activity_score);
                                }
                            }
                        }
                        Err(error) => {
                            eprintln!("worktree watcher error: {error}");
                        }
                    }
                }
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                return Err("Filesystem watcher disconnected.".to_string());
            }
        }

        // Coalesce buffered events/claims to disk once per tick (not per FS event).
        flush_jsonl_write_buffers();

        let commit_every = commit_check_interval(tier, &watch_cfg);
        if last_commit_check.elapsed() >= commit_every {
            let prev = last_head.clone();
            match record_commit_snapshot(&identity, &layout, last_head.as_deref()) {
                Ok(head) => {
                    if head.as_deref() != prev.as_deref() {
                        last_activity = Instant::now();
                        activity_score = bump_activity_score(activity_score);
                    }
                    last_head = head;
                }
                Err(error) => {
                    // Repo may have been deleted while cold/unwatched.
                    if !identity.root.is_dir() {
                        eprintln!(
                            "worktree-watch: repo {} vanished; exiting watch loop",
                            identity.root.display()
                        );
                        flush_jsonl_write_buffers();
                        return Ok(());
                    }
                    eprintln!("worktree commit check failed: {error}");
                }
            }
            last_commit_check = Instant::now();
        }

        if sync_interval_seconds > 0
            && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
        {
            // Ensure spool lines hit disk before upload reads them.
            flush_jsonl_write_buffers();
            if let Err(error) = sync_spool_for_identity(&identity, &layout).await {
                eprintln!("worktree mutation sync failed: {error}");
            }
            last_sync = Instant::now();
        }

        if last_rank_persist.elapsed() >= Duration::from_secs(watch_cfg.rank_persist_seconds()) {
            persist_single_repo_rank(&identity, activity_score, last_activity, tier);
            last_rank_persist = Instant::now();
        }

        if last_housekeeping.elapsed()
            >= Duration::from_secs(SPOOL_HOUSEKEEPING_INTERVAL_SECONDS)
        {
            let _ = prune_spool_layout(&layout, retention);
            last_housekeeping = Instant::now();
        }
    }
}

async fn watch_parent_foreground(
    parent_root: PathBuf,
    identities: Vec<RepoIdentity>,
    sync_interval_seconds: u64,
) -> Result<(), String> {
    // Ensure status/CLI can detect this process even for non-detach starts
    // (and refresh state if the detached parent wrote a different PID).
    if let Ok(layout) = prepare_parent_spool_layout(&parent_root) {
        if let Ok(executable) = std::env::current_exe() {
            let _ = write_parent_watcher_state(
                &parent_root,
                &layout,
                std::process::id(),
                &executable,
            );
        }
    }

    let api_target = WorktreeWatchTargetOptions {
        repo: None,
        parent: Some(parent_root.clone()),
        repos: Vec::new(),
    };
    if let Some(port) = crate::commands::worktree_watch_api::spawn_worktree_watch_api_in_background(
        api_target,
        sync_interval_seconds,
        None,
    ) {
        if !is_background_child() {
            crate::commands::worktree_watch_api::print_api_listen_hint(port);
        } else {
            tracing::info!(
                "worktree-watch local API listening on http://127.0.0.1:{port}"
            );
        }
    }

    let watch_cfg = resolve_worktree_watch_config();
    // Watch each repo root individually — NOT the entire parent tree. Recursive
    // parent watches pin every nested directory handle on Windows (file-claim /
    // delete blocked) and burn CPU on unrelated folder noise.
    let (tx, rx) = mpsc::channel();
    let mut watcher = create_path_watcher(&parent_root, tx)?;
    log_watcher_backend(&parent_root, "parent-per-repo");

    let mut repos = prepare_watched_repos(identities, &watch_cfg)?;
    apply_initial_repo_ranks(&mut repos, &watch_cfg);
    enforce_hot_cap(&mut repos, &watch_cfg);
    apply_fs_watch_budget(&mut watcher, &mut repos, &watch_cfg);
    let initial_fs = repos.iter().filter(|r| r.fs_watched).count();
    let retention = Duration::from_secs(watch_cfg.spool_retention_days().saturating_mul(86_400));
    for repo in &repos {
        let _ = prune_spool_layout(&repo.layout, retention);
    }

    let mut last_sync = Instant::now();
    let mut last_rescan = Instant::now();
    let mut last_rank_persist = Instant::now();
    let mut last_tier_pass = Instant::now();
    let mut last_housekeeping = Instant::now();

    eprintln!(
        "worktree-watch parent: {} repo(s) under {} ({} FS watch(es), max {}; idle→cold unwatch)",
        repos.len(),
        parent_root.display(),
        initial_fs,
        watch_cfg.max_fs_watches(),
    );

    loop {
        let any_hot = repos.iter().any(|r| r.tier == RepoWatchTier::Hot);
        let timeout = if any_hot {
            Duration::from_millis(1000)
        } else if repos.iter().any(|r| r.tier == RepoWatchTier::Warm) {
            Duration::from_millis(2000)
        } else {
            Duration::from_millis(5000)
        };

        match rx.recv_timeout(timeout) {
            Ok(first) => {
                for item in drain_notify_batch(&rx, first) {
                    match item {
                        Ok(event) => {
                            for (repo_index, routed_event) in route_parent_event(&repos, &event) {
                                let repo = &mut repos[repo_index];
                                if let Some(mutation) =
                                    mutation_event_from_notify(&repo.identity, routed_event)
                                {
                                    if append_unique_mutation_event(
                                        &repo.layout,
                                        &mutation,
                                        &mut repo.event_dedupe,
                                    )? {
                                        note_repo_activity(repo);
                                    }
                                }
                            }
                        }
                        Err(error) => {
                            eprintln!("parent worktree watcher error: {error}");
                        }
                    }
                }
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                return Err("Parent filesystem watcher disconnected.".to_string());
            }
        }

        // Per-repo adaptive commit checks — budgeted so 60+ warm repos never all
        // spawn `git` on the same tick (that was ~continuous single-digit % CPU).
        let now = Instant::now();
        let mut due: Vec<usize> = repos
            .iter()
            .enumerate()
            .filter(|(_, r)| now >= r.next_commit_check && r.identity.root.is_dir())
            .map(|(i, _)| i)
            .collect();
        due.sort_by(|&a, &b| {
            let tier_rank = |t: RepoWatchTier| match t {
                RepoWatchTier::Hot => 0u8,
                RepoWatchTier::Warm => 1,
                RepoWatchTier::Cold => 2,
            };
            tier_rank(repos[a].tier)
                .cmp(&tier_rank(repos[b].tier))
                .then_with(|| {
                    repos[b]
                        .activity_score
                        .partial_cmp(&repos[a].activity_score)
                        .unwrap_or(std::cmp::Ordering::Equal)
                })
        });
        let budget = watch_cfg.max_commit_checks_per_tick();
        for idx in due.into_iter().take(budget) {
            let repo = &mut repos[idx];
            // Warm repos without an FS watch only need rare HEAD probes (cold cadence).
            let interval = if repo.tier == RepoWatchTier::Warm && !repo.fs_watched {
                Duration::from_secs(watch_cfg.cold_commit_check_seconds())
            } else {
                commit_check_interval(repo.tier, &watch_cfg)
            };
            let prev = repo.last_head.clone();
            match record_commit_snapshot(
                &repo.identity,
                &repo.layout,
                repo.last_head.as_deref(),
            ) {
                Ok(head) => {
                    if head.as_deref() != prev.as_deref() {
                        note_repo_activity(repo);
                    }
                    repo.last_head = head;
                }
                Err(error) => {
                    tracing::debug!(
                        "commit check {}: {error}",
                        repo.identity.root.display()
                    );
                }
            }
            repo.next_commit_check = Instant::now() + interval;
        }

        // Coalesce buffered events/claims to disk once per tick (not per FS event).
        flush_jsonl_write_buffers();

        // Tier transitions + unwatch cold (frees Windows file locks).
        if last_tier_pass.elapsed() >= Duration::from_secs(15) {
            refresh_repo_tiers(&mut watcher, &mut repos, &watch_cfg);
            last_tier_pass = Instant::now();
        }

        if last_rescan.elapsed() >= Duration::from_secs(watch_cfg.parent_rescan_seconds()) {
            if let Err(error) =
                rescan_parent_repos(&parent_root, &mut watcher, &mut repos, &watch_cfg)
            {
                eprintln!("parent worktree rescan failed: {error}");
            }
            // Cold repos have no FS watch — probe mtime/HEAD so activity can reheat them.
            probe_cold_repo_activity(&mut watcher, &mut repos, &watch_cfg);
            last_rescan = Instant::now();
        }

        if sync_interval_seconds > 0
            && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
        {
            // Ensure spool lines hit disk before upload reads them.
            flush_jsonl_write_buffers();
            // Sync hot repos first, then warm; skip cold unless they still have spool backlog.
            let mut order: Vec<usize> = (0..repos.len()).collect();
            order.sort_by(|&a, &b| {
                repos[b]
                    .activity_score
                    .partial_cmp(&repos[a].activity_score)
                    .unwrap_or(std::cmp::Ordering::Equal)
            });
            for idx in order {
                let repo = &repos[idx];
                if repo.tier == RepoWatchTier::Cold {
                    // Still drain spool occasionally so idle data is not stuck forever.
                    // Cheap when empty.
                }
                if let Err(error) = sync_spool_for_identity(&repo.identity, &repo.layout).await {
                    eprintln!(
                        "worktree mutation sync failed for {}: {error}",
                        repo.identity.root.display()
                    );
                }
            }
            last_sync = Instant::now();
        }

        if last_rank_persist.elapsed() >= Duration::from_secs(watch_cfg.rank_persist_seconds()) {
            persist_repo_ranks(&repos);
            last_rank_persist = Instant::now();
        }

        if last_housekeeping.elapsed()
            >= Duration::from_secs(SPOOL_HOUSEKEEPING_INTERVAL_SECONDS)
        {
            let retention =
                Duration::from_secs(watch_cfg.spool_retention_days().saturating_mul(86_400));
            // Spread work: hot first, then a slice of the rest each pass.
            let mut order: Vec<usize> = (0..repos.len()).collect();
            order.sort_by_key(|&i| match repos[i].tier {
                RepoWatchTier::Hot => 0u8,
                RepoWatchTier::Warm => 1,
                RepoWatchTier::Cold => 2,
            });
            for &idx in order.iter().take(16) {
                let _ = prune_spool_layout(&repos[idx].layout, retention);
            }
            last_housekeeping = Instant::now();
        }
    }
}

/// Prune durable spool clutter that otherwise grows forever on busy machines:
/// - `.synced.*.jsonl` older than retention
/// - `event-dedupe/*.seen` / `commit-dedupe/*.seen` older than retention
/// - rotate oversized `event-dedupe.jsonl` (markers remain the source of truth)
fn prune_spool_layout(layout: &SpoolLayout, retention: Duration) -> Result<u64, String> {
    let mut removed = 0u64;
    removed += prune_old_files_in_dir(&layout.event_dedupe_dir, retention, |name| {
        name.ends_with(".seen")
    })?;
    removed += prune_old_files_in_dir(&layout.commit_dedupe_dir, retention, |name| {
        name.ends_with(".seen")
    })?;
    removed += prune_old_files_in_dir(&layout.root, retention, |name| {
        name.ends_with(".jsonl") && name.contains(".synced.")
    })?;
    removed += rotate_oversized_jsonl(&layout.event_dedupe_file, MAX_EVENT_DEDUPE_JSONL_BYTES)?;
    Ok(removed)
}

fn prune_old_files_in_dir(
    dir: &Path,
    retention: Duration,
    mut name_ok: impl FnMut(&str) -> bool,
) -> Result<u64, String> {
    if !dir.is_dir() {
        return Ok(0);
    }
    let cutoff = std::time::SystemTime::now()
        .checked_sub(retention)
        .unwrap_or(std::time::SystemTime::UNIX_EPOCH);
    let mut removed = 0u64;
    let entries = match fs::read_dir(dir) {
        Ok(entries) => entries,
        Err(_) => return Ok(0),
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_file() {
            continue;
        }
        let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
            continue;
        };
        if !name_ok(name) {
            continue;
        }
        let Ok(meta) = entry.metadata() else {
            continue;
        };
        let Ok(modified) = meta.modified() else {
            continue;
        };
        if modified < cutoff {
            if fs::remove_file(&path).is_ok() {
                removed += 1;
            }
        }
    }
    Ok(removed)
}

fn rotate_oversized_jsonl(path: &Path, max_bytes: u64) -> Result<u64, String> {
    let Ok(meta) = fs::metadata(path) else {
        return Ok(0);
    };
    if meta.len() <= max_bytes {
        return Ok(0);
    }
    // Drop the bloated telemetry log; `.seen` markers keep idempotency.
    let backup = path.with_extension("jsonl.pruned");
    let _ = fs::remove_file(&backup);
    match fs::rename(path, &backup) {
        Ok(()) => {
            let _ = fs::remove_file(&backup);
            Ok(1)
        }
        Err(_) => {
            // Fallback: truncate in place.
            match OpenOptions::new().write(true).truncate(true).open(path) {
                Ok(_) => Ok(1),
                Err(error) => Err(format!(
                    "Failed to rotate oversized dedupe log {}: {error}",
                    path.display()
                )),
            }
        }
    }
}

fn note_repo_activity(repo: &mut WatchedRepo) {
    repo.last_activity = Instant::now();
    repo.activity_score = bump_activity_score(repo.activity_score);
    if repo.tier != RepoWatchTier::Hot {
        repo.tier = RepoWatchTier::Hot;
        repo.next_commit_check = Instant::now();
    }
}

fn tier_from_idle(idle_secs: u64, cfg: &WorktreeWatchConfig) -> RepoWatchTier {
    if idle_secs >= cfg.cold_after_seconds() {
        RepoWatchTier::Cold
    } else if idle_secs >= cfg.idle_after_seconds() {
        RepoWatchTier::Warm
    } else {
        RepoWatchTier::Hot
    }
}

fn commit_check_interval(tier: RepoWatchTier, cfg: &WorktreeWatchConfig) -> Duration {
    let secs = match tier {
        RepoWatchTier::Hot => cfg.hot_commit_check_seconds(),
        RepoWatchTier::Warm => cfg.warm_commit_check_seconds(),
        RepoWatchTier::Cold => cfg.cold_commit_check_seconds(),
    };
    Duration::from_secs(secs)
}

fn event_loop_timeout(tier: RepoWatchTier) -> Duration {
    match tier {
        RepoWatchTier::Hot => Duration::from_millis(1000),
        RepoWatchTier::Warm => Duration::from_millis(2500),
        RepoWatchTier::Cold => Duration::from_millis(5000),
    }
}

fn bump_activity_score(score: f64) -> f64 {
    // Simple growth with soft cap so ranks stay comparable.
    (score * 0.98 + 1.0).min(10_000.0)
}

fn seed_activity_score(cfg: &WorktreeWatchConfig, root: &Path) -> f64 {
    let key = path_identity_key(root);
    cfg.repo_ranks
        .iter()
        .find(|r| path_identity_key(Path::new(&r.root)) == key)
        .map(|r| r.score)
        .unwrap_or(0.0)
}

fn apply_initial_repo_ranks(repos: &mut [WatchedRepo], cfg: &WorktreeWatchConfig) {
    for repo in repos.iter_mut() {
        repo.activity_score = seed_activity_score(cfg, &repo.identity.root);
        // Seed last_activity so cold ranks don't all start hot.
        if let Some(rank) = cfg.repo_ranks.iter().find(|r| {
            path_identity_key(Path::new(&r.root)) == path_identity_key(&repo.identity.root)
        }) {
            if let Some(ts) = rank.last_event_at.as_deref() {
                if let Ok(dt) = DateTime::parse_from_rfc3339(ts) {
                    let age = (Utc::now() - dt.with_timezone(&Utc))
                        .num_seconds()
                        .max(0) as u64;
                    repo.tier = tier_from_idle(age, cfg);
                    // Approximate Instant: use now - age when age is small enough for Instant.
                    if age < 24 * 3600 {
                        repo.last_activity =
                            Instant::now() - Duration::from_secs(age.min(23 * 3600));
                    } else {
                        repo.last_activity =
                            Instant::now() - Duration::from_secs(cfg.cold_after_seconds());
                    }
                }
            }
        }
    }
}

fn enforce_hot_cap(repos: &mut [WatchedRepo], cfg: &WorktreeWatchConfig) {
    let max_hot = cfg.max_hot_repos();
    let mut hot_idx: Vec<usize> = repos
        .iter()
        .enumerate()
        .filter(|(_, r)| r.tier == RepoWatchTier::Hot)
        .map(|(i, _)| i)
        .collect();
    if hot_idx.len() <= max_hot {
        return;
    }
    hot_idx.sort_by(|&a, &b| {
        repos[b]
            .activity_score
            .partial_cmp(&repos[a].activity_score)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    for &idx in hot_idx.iter().skip(max_hot) {
        repos[idx].tier = RepoWatchTier::Warm;
    }
}

fn refresh_repo_tiers(
    watcher: &mut PathWatcher,
    repos: &mut [WatchedRepo],
    cfg: &WorktreeWatchConfig,
) {
    for repo in repos.iter_mut() {
        // Decay score slowly while idle so ranks fall over time.
        let idle = repo.last_activity.elapsed().as_secs();
        if idle > 60 {
            repo.activity_score *= 0.995;
        }
        let desired = tier_from_idle(idle, cfg);
        if desired != repo.tier {
            tracing::debug!(
                "worktree-watch tier {} → {} for {}",
                repo.tier.as_str(),
                desired.as_str(),
                repo.identity.root.display()
            );
            repo.tier = desired;
            repo.next_commit_check = Instant::now() + commit_check_interval(desired, cfg);
        }
    }
    enforce_hot_cap(repos, cfg);
    apply_fs_watch_budget(watcher, repos, cfg);
}

/// Cap concurrent recursive FS watches. Prefer hot, then highest activity score.
fn apply_fs_watch_budget(
    watcher: &mut PathWatcher,
    repos: &mut [WatchedRepo],
    cfg: &WorktreeWatchConfig,
) {
    let max = cfg.max_fs_watches();
    let want = fs_watch_want_mask(repos, max);
    for (index, repo) in repos.iter_mut().enumerate() {
        let _ = set_repo_fs_watch(watcher, repo, want[index]);
    }
}

/// Which repos should hold a recursive FS watch under the budget.
///
/// Prefer hot over warm, then higher activity score. Cold never watches.
/// Non-existent roots are skipped (production always has real dirs).
fn fs_watch_want_mask(repos: &[WatchedRepo], max: usize) -> Vec<bool> {
    let mut eligible: Vec<usize> = repos
        .iter()
        .enumerate()
        .filter(|(_, repo)| repo.tier != RepoWatchTier::Cold && repo.identity.root.is_dir())
        .map(|(index, _)| index)
        .collect();
    eligible.sort_by(|&a, &b| {
        let hot_a = repos[a].tier == RepoWatchTier::Hot;
        let hot_b = repos[b].tier == RepoWatchTier::Hot;
        hot_b
            .cmp(&hot_a)
            .then_with(|| {
                repos[b]
                    .activity_score
                    .partial_cmp(&repos[a].activity_score)
                    .unwrap_or(std::cmp::Ordering::Equal)
            })
            .then_with(|| {
                path_identity_key(&repos[a].identity.root)
                    .cmp(&path_identity_key(&repos[b].identity.root))
            })
    });

    let mut want = vec![false; repos.len()];
    for &index in eligible.iter().take(max) {
        want[index] = true;
    }
    want
}

fn set_repo_fs_watch(
    watcher: &mut PathWatcher,
    repo: &mut WatchedRepo,
    want_watch: bool,
) -> Result<(), String> {
    if want_watch && !repo.fs_watched {
        match watcher.watch(&repo.identity.root, RecursiveMode::Recursive) {
            Ok(()) => {
                repo.fs_watched = true;
                tracing::info!(
                    "worktree-watch: watching {} ({})",
                    repo.identity.root.display(),
                    repo.tier.as_str()
                );
            }
            Err(error) => {
                // Path may be mid-delete; leave cold.
                tracing::warn!(
                    "worktree-watch: watch {} failed: {error}",
                    repo.identity.root.display()
                );
                repo.tier = RepoWatchTier::Cold;
                repo.fs_watched = false;
            }
        }
    } else if !want_watch && repo.fs_watched {
        let _ = watcher.unwatch(&repo.identity.root);
        repo.fs_watched = false;
        tracing::info!(
            "worktree-watch: unwatched {} (budget/cold — directory locks released)",
            repo.identity.root.display()
        );
    }
    Ok(())
}

/// Cheap reheating for cold (unwatched) repos: if `.git/index` / HEAD moved, go hot.
fn probe_cold_repo_activity(
    watcher: &mut PathWatcher,
    repos: &mut [WatchedRepo],
    cfg: &WorktreeWatchConfig,
) {
    for repo in repos.iter_mut() {
        if repo.tier != RepoWatchTier::Cold {
            continue;
        }
        if !repo.identity.root.is_dir() {
            continue;
        }
        if cold_repo_looks_active(repo) {
            note_repo_activity(repo);
            tracing::info!(
                "worktree-watch: reheated cold repo {}",
                repo.identity.root.display()
            );
        }
    }
    enforce_hot_cap(repos, cfg);
    apply_fs_watch_budget(watcher, repos, cfg);
}

fn cold_repo_looks_active(repo: &WatchedRepo) -> bool {
    use std::time::SystemTime;
    let root = &repo.identity.root;
    // Approximate wall-clock of last_activity for mtime comparison.
    let last_activity_wall = SystemTime::now()
        .checked_sub(repo.last_activity.elapsed())
        .unwrap_or(SystemTime::UNIX_EPOCH);
    // Prefer index/HEAD mtimes — cheap; updates on commits and many git operations.
    for rel in [".git/index", ".git/HEAD", ".git/logs/HEAD"] {
        let path = root.join(rel);
        if let Ok(meta) = fs::metadata(&path) {
            if let Ok(modified) = meta.modified() {
                if modified > last_activity_wall {
                    return true;
                }
            }
        }
    }
    // Fallback: HEAD sha changed since last snapshot.
    match current_head(root) {
        Ok(Some(head)) if Some(head.as_str()) != repo.last_head.as_deref() => true,
        _ => false,
    }
}

fn rescan_parent_repos(
    parent_root: &Path,
    watcher: &mut PathWatcher,
    repos: &mut Vec<WatchedRepo>,
    cfg: &WorktreeWatchConfig,
) -> Result<(), String> {
    let discovered = discover_repo_identities_under(parent_root)?;
    let live_keys: BTreeSet<String> = discovered
        .iter()
        .map(|id| path_identity_key(&id.root))
        .collect();
    let existing_keys: BTreeSet<String> = repos
        .iter()
        .map(|r| path_identity_key(&r.identity.root))
        .collect();

    for identity in discovered {
        let key = path_identity_key(&identity.root);
        if existing_keys.contains(&key) {
            continue;
        }
        let prepared = prepare_watched_repos(vec![identity], cfg)?;
        for mut repo in prepared {
            tracing::info!(
                "worktree-watch parent: added repo {} ({})",
                repo.identity.root.display(),
                repo.tier.as_str()
            );
            // Newly discovered roots stay cold until edited (or ranked hot later).
            // Avoid pinning FS watches / Windows file locks on brand-new clones.
            repo.tier = RepoWatchTier::Cold;
            repo.fs_watched = false;
            repo.last_activity = Instant::now() - Duration::from_secs(cfg.cold_after_seconds());
            repos.push(repo);
        }
    }

    // Drop vanished roots (unwatch first so Windows can finish deletes).
    let mut keep = Vec::with_capacity(repos.len());
    for mut repo in repos.drain(..) {
        let key = path_identity_key(&repo.identity.root);
        if live_keys.contains(&key) && repo.identity.root.is_dir() {
            keep.push(repo);
        } else {
            if repo.fs_watched {
                let _ = watcher.unwatch(&repo.identity.root);
                repo.fs_watched = false;
            }
            tracing::info!(
                "worktree-watch parent: removed vanished repo {}",
                repo.identity.root.display()
            );
        }
    }
    *repos = keep;
    repos.sort_by_key(|repo| std::cmp::Reverse(repo.identity.root.components().count()));
    enforce_hot_cap(repos, cfg);
    apply_fs_watch_budget(watcher, repos, cfg);
    Ok(())
}

fn persist_repo_ranks(repos: &[WatchedRepo]) {
    let mut ranks: Vec<WorktreeWatchRepoRank> = repos
        .iter()
        .map(|repo| WorktreeWatchRepoRank {
            root: repo.identity.root.display().to_string(),
            owner: Some(repo.identity.owner.clone()),
            name: Some(repo.identity.name.clone()),
            score: (repo.activity_score * 1000.0).round() / 1000.0,
            last_event_at: Some(
                (Utc::now()
                    - chrono::Duration::from_std(repo.last_activity.elapsed())
                        .unwrap_or_else(|_| chrono::Duration::seconds(0)))
                .to_rfc3339(),
            ),
            rank: None,
            tier: Some(repo.tier.as_str().to_string()),
        })
        .collect();
    ranks.sort_by(|a, b| {
        b.score
            .partial_cmp(&a.score)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    for (i, rank) in ranks.iter_mut().enumerate() {
        rank.rank = Some((i + 1) as u32);
    }
    // Keep config bounded.
    ranks.truncate(200);

    let mut config = match SshConfig::load() {
        Ok(c) => c,
        Err(error) => {
            tracing::warn!("worktree-watch: could not load config to persist ranks: {error}");
            return;
        }
    };
    let mut ww = config.worktree_watch.unwrap_or_default();
    ww.repo_ranks = ranks;
    config.worktree_watch = Some(ww);
    if let Err(error) = config.save() {
        tracing::warn!("worktree-watch: could not persist repo ranks: {error}");
    }
}

fn persist_single_repo_rank(
    identity: &RepoIdentity,
    score: f64,
    last_activity: Instant,
    tier: RepoWatchTier,
) {
    let mut config = match SshConfig::load() {
        Ok(c) => c,
        Err(_) => return,
    };
    let mut ww = config.worktree_watch.unwrap_or_default();
    let key = path_identity_key(&identity.root);
    let entry = WorktreeWatchRepoRank {
        root: identity.root.display().to_string(),
        owner: Some(identity.owner.clone()),
        name: Some(identity.name.clone()),
        score: (score * 1000.0).round() / 1000.0,
        last_event_at: Some(
            (Utc::now()
                - chrono::Duration::from_std(last_activity.elapsed())
                    .unwrap_or_else(|_| chrono::Duration::seconds(0)))
            .to_rfc3339(),
        ),
        rank: None,
        tier: Some(tier.as_str().to_string()),
    };
    if let Some(existing) = ww
        .repo_ranks
        .iter_mut()
        .find(|r| path_identity_key(Path::new(&r.root)) == key)
    {
        *existing = entry;
    } else {
        ww.repo_ranks.push(entry);
    }
    ww.repo_ranks.sort_by(|a, b| {
        b.score
            .partial_cmp(&a.score)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    for (i, rank) in ww.repo_ranks.iter_mut().enumerate() {
        rank.rank = Some((i + 1) as u32);
    }
    ww.repo_ranks.truncate(200);
    config.worktree_watch = Some(ww);
    let _ = config.save();
}

/// Filesystem watcher handle that can be either native (inotify/FSEvents/ReadDirectoryChanges)
/// or polling (needed for WSL mounts of Windows drives where inotify is unreliable).
enum PathWatcher {
    Recommended(RecommendedWatcher),
    Poll(notify::PollWatcher),
}

impl PathWatcher {
    fn watch(&mut self, path: &Path, mode: RecursiveMode) -> notify::Result<()> {
        match self {
            Self::Recommended(watcher) => watcher.watch(path, mode),
            Self::Poll(watcher) => watcher.watch(path, mode),
        }
    }

    fn unwatch(&mut self, path: &Path) -> notify::Result<()> {
        match self {
            Self::Recommended(watcher) => watcher.unwatch(path),
            Self::Poll(watcher) => watcher.unwatch(path),
        }
    }
}

fn create_path_watcher(
    path: &Path,
    tx: mpsc::Sender<notify::Result<Event>>,
) -> Result<PathWatcher, String> {
    let config = Config::default();
    if should_use_poll_watcher(path) {
        // Slower poll: adaptive cold/hot layers handle responsiveness.
        let poll_config = config.with_poll_interval(Duration::from_secs(5));
        notify::PollWatcher::new(
            move |result| {
                let _ = tx.send(result);
            },
            poll_config,
        )
        .map(PathWatcher::Poll)
        .map_err(|error| format!("Failed to create poll filesystem watcher: {error}"))
    } else {
        RecommendedWatcher::new(
            move |result| {
                let _ = tx.send(result);
            },
            config,
        )
        .map(PathWatcher::Recommended)
        .map_err(|error| format!("Failed to create filesystem watcher: {error}"))
    }
}

fn should_use_poll_watcher(path: &Path) -> bool {
    // WSL inotify does not reliably cover 9p/drvfs mounts under /mnt/*.
    if !is_wsl_runtime() {
        return false;
    }
    let key = path_identity_key(path).replace('\\', "/");
    key.starts_with("/mnt/") || key.contains("/mnt/")
}

fn log_watcher_backend(path: &Path, label: &str) {
    if should_use_poll_watcher(path) {
        eprintln!(
            "worktree-watch: using poll watcher for {label} {} (WSL mount; inotify is unreliable)",
            path.display()
        );
    } else {
        tracing::info!(
            "worktree-watch: native watcher for {label} {}",
            path.display()
        );
    }
}

fn prepare_watched_repos(
    identities: Vec<RepoIdentity>,
    cfg: &WorktreeWatchConfig,
) -> Result<Vec<WatchedRepo>, String> {
    let mut repos = Vec::with_capacity(identities.len());
    let now = Instant::now();
    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        let event_dedupe = RecentEventDedupe {
            fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
        };
        let score = seed_activity_score(cfg, &identity.root);
        // Prefer cold for unknown/idle repos so parent covers do not pin dozens of
        // directory handles. Ranked / previously active repos start warmer.
        // Low residual scores (1–2 from old ranks) stay cold — they were flooding
        // warm-tier git HEAD polls across 60+ clones.
        let tier = if score >= 5.0 {
            RepoWatchTier::Hot
        } else if score >= 3.0 {
            RepoWatchTier::Warm
        } else {
            RepoWatchTier::Cold
        };
        // Cold start: pretend idle long enough that we do not immediately re-hot.
        let last_activity = if matches!(tier, RepoWatchTier::Cold) {
            now - Duration::from_secs(cfg.cold_after_seconds())
        } else if matches!(tier, RepoWatchTier::Warm) {
            now - Duration::from_secs(cfg.idle_after_seconds())
        } else {
            now
        };
        // Stagger first HEAD probe so a 70-repo parent does not spawn 70 gits at t=0.
        let stagger_ms = (repos.len() as u64).saturating_mul(400) % 45_000;
        repos.push(WatchedRepo {
            last_head: identity.head_sha.clone(),
            identity,
            layout,
            event_dedupe,
            last_activity,
            activity_score: score,
            tier,
            fs_watched: false,
            next_commit_check: now + Duration::from_millis(stagger_ms),
        });
    }
    repos.sort_by_key(|repo| std::cmp::Reverse(repo.identity.root.components().count()));
    Ok(repos)
}

fn route_parent_event(repos: &[WatchedRepo], event: &Event) -> Vec<(usize, Event)> {
    let mut paths_by_repo: BTreeMap<usize, Vec<PathBuf>> = BTreeMap::new();
    for path in &event.paths {
        for (index, repo) in repos.iter().enumerate() {
            if path_belongs_to_repo(path, &repo.identity.root) {
                let paths = paths_by_repo.entry(index).or_default();
                if !paths.iter().any(|existing| existing == path) {
                    paths.push(path.clone());
                }
                break;
            }
        }
    }

    let mut routed = Vec::new();
    for (index, paths) in paths_by_repo {
        let mut repo_event = event.clone();
        repo_event.paths = paths;
        routed.push((index, repo_event));
    }
    routed
}

fn path_belongs_to_repo(path: &Path, repo_root: &Path) -> bool {
    let path_key = path_identity_key(path).replace('\\', "/");
    let root_key = path_identity_key(repo_root).replace('\\', "/");
    path_key == root_key || path_key.starts_with(&format!("{root_key}/"))
}

fn mutation_event_from_notify(
    identity: &RepoIdentity,
    event: Event,
) -> Option<WorktreeMutationEvent> {
    if event.kind.is_access() || event.kind.is_other() {
        return None;
    }

    let paths: Vec<String> = event
        .paths
        .iter()
        .filter(|path| !is_ignored_path(&identity.root, path))
        .filter_map(|path| relative_slash_path(&identity.root, path))
        .fold(Vec::new(), |mut paths, path| {
            if !paths.iter().any(|existing| existing == &path) {
                paths.push(path);
            }
            paths
        });

    if paths.is_empty() {
        return None;
    }

    let primary_path = paths.first().cloned();
    let (old_path, new_path) = rename_paths(&event.kind, &paths);
    // Expensive git/file scans only for content modifies — not every create/remove/noise event.
    let want_line_stats = matches!(
        event.kind,
        EventKind::Modify(ModifyKind::Data(_)) | EventKind::Modify(ModifyKind::Any)
    );
    let line_counts = if want_line_stats {
        primary_path
            .as_deref()
            .and_then(|path| git_numstat_for_path(&identity.root, path).ok())
    } else {
        None
    };
    let total_lines = if want_line_stats {
        primary_path
            .as_deref()
            .and_then(|path| file_line_count_for_path(&identity.root, path).ok())
    } else {
        None
    };
    let code_snapshots = primary_path
        .as_deref()
        .and_then(|path| fs::read_to_string(identity.root.join(path)).ok())
        .map(|source| {
            snapshots_for_file(primary_path.as_deref().unwrap_or_default(), &source)
                .into_iter()
                .take(32)
                .collect()
        })
        .unwrap_or_default();
    let event_kind = classify_event_kind(&event.kind);
    let is_dir = event
        .paths
        .first()
        .and_then(|path| fs::metadata(path).ok())
        .map(|metadata| metadata.is_dir())
        .unwrap_or(false);

    let occurred_at = Utc::now();
    let mut mutation = WorktreeMutationEvent {
        id: String::new(),
        fingerprint: None,
        repo_owner: identity.owner.clone(),
        repo_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: normalize_repo_root_for_payload(&identity.root),
        // Avoid git spawn on every FS event (HEAD is refreshed on the adaptive commit timer).
        head_sha: identity.head_sha.clone(),
        event_kind: event_kind.to_string(),
        paths,
        primary_path,
        old_path,
        new_path,
        added_lines: line_counts.map(|counts| counts.0),
        removed_lines: line_counts.map(|counts| counts.1),
        total_lines,
        code_snapshots,
        file_created: matches!(
            event.kind,
            EventKind::Create(CreateKind::File | CreateKind::Any)
        ) && !is_dir,
        file_removed: matches!(
            event.kind,
            EventKind::Remove(RemoveKind::File | RemoveKind::Any)
        ) && !is_dir,
        folder_created: matches!(
            event.kind,
            EventKind::Create(CreateKind::Folder | CreateKind::Any)
        ) && is_dir,
        folder_removed: matches!(
            event.kind,
            EventKind::Remove(RemoveKind::Folder | RemoveKind::Any)
        ) && is_dir,
        renamed_or_moved: is_rename_or_move(&event.kind),
        raw_kind: format!("{:?}", event.kind),
        occurred_at,
    };
    // Stable id/fingerprint for idempotent spool + upload (same logical mutation
    // within the dedupe window always maps to the same identity).
    let fingerprint = mutation_event_fingerprint(&mutation);
    mutation.fingerprint = Some(fingerprint.clone());
    mutation.id = stable_event_id(&fingerprint);
    Some(mutation)
}

fn append_unique_mutation_event(
    layout: &SpoolLayout,
    mutation: &WorktreeMutationEvent,
    dedupe: &mut RecentEventDedupe,
) -> Result<bool, String> {
    let fingerprint = mutation
        .fingerprint
        .clone()
        .unwrap_or_else(|| mutation_event_fingerprint(mutation));
    // In-memory cache + `.seen` markers only. Scanning event-dedupe.jsonl was
    // O(file size) per event and loaded multi‑10MB sets at process start.
    if dedupe.fingerprints.contains(&fingerprint) {
        return Ok(false);
    }

    if !claim_event_fingerprint(layout, &fingerprint)? {
        dedupe_remember(dedupe, fingerprint);
        return Ok(false);
    }

    let mut mutation = mutation.clone();
    mutation.fingerprint = Some(fingerprint.clone());
    if mutation.id.is_empty() {
        mutation.id = stable_event_id(&fingerprint);
    }
    if let Err(error) = append_json_line(&layout.events_file, &mutation) {
        release_event_fingerprint_claim(layout, &fingerprint);
        return Err(error);
    }
    // Claims journal already records the fingerprint (see claim_event_fingerprint).
    // Skip a second per-event write to event-dedupe.jsonl — that doubled IO for
    // every mutation and grew multi‑10MB logs that were never loaded at start.
    dedupe_remember(dedupe, fingerprint);
    Ok(true)
}

fn dedupe_remember(dedupe: &mut RecentEventDedupe, fingerprint: String) {
    if dedupe.fingerprints.len() >= MAX_RECENT_EVENT_DEDUPE_IN_MEMORY {
        dedupe.fingerprints.clear();
    }
    dedupe.fingerprints.insert(fingerprint);
}

fn claim_event_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
    // Legacy per-fingerprint markers (one file each → 10k–100k files under
    // ~/.xbp/mutations). Honor them for older spools, but do not create new ones.
    let marker = event_dedupe_marker_path(layout, fingerprint);
    if marker.is_file() {
        return Ok(false);
    }
    // Process-wide claim set: covers multi-dedupe-instance races in one OS process
    // without creating per-fingerprint files. Journal is durable telemetry only.
    let claim_key = format!("{}|{fingerprint}", layout.root.display());
    {
        let mut claimed = event_claims_in_process()
            .lock()
            .map_err(|_| "event claim lock poisoned".to_string())?;
        if !claimed.insert(claim_key) {
            return Ok(false);
        }
    }
    let claims = layout.root.join("event-claims.jsonl");
    append_json_line(
        &claims,
        &json!({
            "fingerprint": fingerprint,
            "firstSeenAt": Utc::now().to_rfc3339(),
        }),
    )?;
    Ok(true)
}

fn release_event_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
    // Best-effort: remove legacy marker if present. Journal lines are append-only
    // (failed event writes are rare after claim).
    let _ = fs::remove_file(event_dedupe_marker_path(layout, fingerprint));
    let claim_key = format!("{}|{fingerprint}", layout.root.display());
    if let Ok(mut claimed) = event_claims_in_process().lock() {
        claimed.remove(&claim_key);
    }
}

fn event_claims_in_process() -> &'static Mutex<BTreeSet<String>> {
    static CLAIMS: OnceLock<Mutex<BTreeSet<String>>> = OnceLock::new();
    CLAIMS.get_or_init(|| Mutex::new(BTreeSet::new()))
}

fn event_dedupe_marker_path(layout: &SpoolLayout, fingerprint: &str) -> PathBuf {
    layout.event_dedupe_dir.join(format!("{fingerprint}.seen"))
}

fn mutation_event_fingerprint(mutation: &WorktreeMutationEvent) -> String {
    let bucket = mutation.occurred_at.timestamp() / EVENT_DEDUPE_WINDOW_SECONDS;
    // Omit absolute `repoRoot` so Windows `C:\...` and WSL `/mnt/c/...` spellings
    // of the same tree share one fingerprint. Platform is kept only as telemetry
    // on the upload device payload — not in the local idempotency key — so both
    // environments can share one home-level spool safely.
    let payload = json!({
        "schema": 3,
        "repoOwner": mutation.repo_owner,
        "repoName": mutation.repo_name,
        "branchName": mutation.branch_name,
        "headSha": mutation.head_sha,
        "eventKind": mutation.event_kind,
        "paths": normalize_path_list_for_fingerprint(&mutation.paths),
        "primaryPath": mutation.primary_path.as_deref().map(normalize_relative_watch_path),
        "oldPath": mutation.old_path.as_deref().map(normalize_relative_watch_path),
        "newPath": mutation.new_path.as_deref().map(normalize_relative_watch_path),
        "addedLines": mutation.added_lines,
        "removedLines": mutation.removed_lines,
        "fileCreated": mutation.file_created,
        "fileRemoved": mutation.file_removed,
        "folderCreated": mutation.folder_created,
        "folderRemoved": mutation.folder_removed,
        "renamedOrMoved": mutation.renamed_or_moved,
        "timeBucket": bucket,
    });
    let encoded = serde_json::to_vec(&payload).unwrap_or_default();
    let mut hasher = Sha256::new();
    hasher.update(encoded);
    format!("{:x}", hasher.finalize())
}

fn normalize_path_list_for_fingerprint(paths: &[String]) -> Vec<String> {
    let mut normalized = paths
        .iter()
        .map(|path| normalize_relative_watch_path(path))
        .filter(|path| !path.is_empty())
        .collect::<Vec<_>>();
    normalized.sort();
    normalized.dedup();
    normalized
}

fn stable_event_id(fingerprint: &str) -> String {
    // Deterministic id derived from fingerprint so resync / multi-process claims
    // never invent a second identity for the same logical mutation.
    format!("wtmut_{fingerprint}")
}

fn normalize_repo_root_for_payload(root: &Path) -> String {
    path_identity_key(root).replace('\\', "/")
}

fn load_event_dedupe_fingerprints(_path: &Path) -> Result<BTreeSet<String>, String> {
    // Intentionally empty. Durable cross-process dedupe uses `event-dedupe/*.seen`
    // markers. Loading multi‑10MB `event-dedupe.jsonl` files for every branch under
    // a parent cover was a primary cause of ~1 GiB RSS.
    Ok(BTreeSet::new())
}

fn classify_event_kind(kind: &EventKind) -> &'static str {
    match kind {
        EventKind::Create(CreateKind::File) => "file_create",
        EventKind::Create(CreateKind::Folder) => "folder_create",
        EventKind::Create(_) => "create",
        EventKind::Remove(RemoveKind::File) => "file_remove",
        EventKind::Remove(RemoveKind::Folder) => "folder_remove",
        EventKind::Remove(_) => "remove",
        EventKind::Modify(ModifyKind::Name(
            RenameMode::From | RenameMode::To | RenameMode::Both,
        )) => "rename_or_move",
        EventKind::Modify(ModifyKind::Data(_)) => "file_modify",
        EventKind::Modify(ModifyKind::Metadata(_)) => "metadata_modify",
        EventKind::Modify(_) => "modify",
        _ => "other",
    }
}

fn is_rename_or_move(kind: &EventKind) -> bool {
    matches!(
        kind,
        EventKind::Modify(ModifyKind::Name(
            RenameMode::From | RenameMode::To | RenameMode::Both
        ))
    )
}

fn rename_paths(kind: &EventKind, paths: &[String]) -> (Option<String>, Option<String>) {
    if !is_rename_or_move(kind) {
        return (None, None);
    }

    match paths {
        [old_path, new_path, ..] => (Some(old_path.clone()), Some(new_path.clone())),
        [path] => match kind {
            EventKind::Modify(ModifyKind::Name(RenameMode::From)) => (Some(path.clone()), None),
            EventKind::Modify(ModifyKind::Name(RenameMode::To)) => (None, Some(path.clone())),
            _ => (None, Some(path.clone())),
        },
        _ => (None, None),
    }
}

fn record_commit_snapshot(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    previous_head: Option<&str>,
) -> Result<Option<String>, String> {
    let head = current_head(&identity.root)?;
    let Some(head_sha) = head else {
        return Ok(None);
    };

    if previous_head == Some(head_sha.as_str()) {
        return Ok(Some(head_sha));
    }

    // One git log for parents + subject/author/email/time (was rev-list + separate log).
    let enrich = enrich_commit_range(&identity.root, previous_head, &head_sha);

    let mut commit = WorktreeCommitEvent {
        id: String::new(),
        fingerprint: None,
        repo_owner: identity.owner.clone(),
        repo_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: normalize_repo_root_for_payload(&identity.root),
        previous_head_sha: previous_head.map(str::to_string),
        head_sha: head_sha.clone(),
        subject: enrich.subject,
        author_name: enrich.author_name,
        author_email: enrich.author_email,
        committed_at: enrich.committed_at,
        occurred_at: Utc::now(),
        files_changed: enrich.files_changed,
        files_changed_count: enrich.files_changed_count,
        insertions: enrich.insertions,
        deletions: enrich.deletions,
        is_merge: enrich.is_merge,
        included_commits: enrich.included_commits,
    };
    let fingerprint = commit_event_fingerprint(&commit);
    commit.fingerprint = Some(fingerprint.clone());
    commit.id = stable_event_id(&fingerprint);
    append_unique_commit_event(layout, &commit)?;
    Ok(Some(head_sha))
}

const COMMIT_FILES_CAP: usize = 200;
const INCLUDED_COMMITS_CAP: usize = 32;

#[derive(Default)]
struct CommitRangeEnrichment {
    subject: Option<String>,
    author_name: Option<String>,
    author_email: Option<String>,
    committed_at: Option<String>,
    files_changed: Option<Vec<String>>,
    files_changed_count: Option<u64>,
    insertions: Option<u64>,
    deletions: Option<u64>,
    is_merge: Option<bool>,
    included_commits: Option<Vec<String>>,
}

fn enrich_commit_range(
    repo_root: &Path,
    previous_head: Option<&str>,
    head_sha: &str,
) -> CommitRangeEnrichment {
    let mut out = CommitRangeEnrichment::default();

    let range = match previous_head {
        Some(prev) if !prev.is_empty() && prev != head_sha => format!("{prev}..{head_sha}"),
        _ => format!("{head_sha}^!"),
    };

    // Bundle merge detection + commit meta into one spawn:
    // %P = parent SHAs, then subject/author/email/time.
    // Remaining: numstat (+ optional multi-commit log).
    if let Ok(meta) = git_output(
        repo_root,
        &[
            "log",
            "-1",
            "--pretty=%P%x00%s%x00%an%x00%ae%x00%cI",
            head_sha,
        ],
    ) {
        let mut parts = meta.split('\0').map(str::trim);
        let parents = parts.next().unwrap_or("");
        let parent_count = if parents.is_empty() {
            0
        } else {
            parents.split_whitespace().count()
        };
        out.is_merge = Some(parent_count > 1);
        out.subject = non_empty_owned(parts.next());
        out.author_name = non_empty_owned(parts.next());
        out.author_email = non_empty_owned(parts.next());
        out.committed_at = non_empty_owned(parts.next());
    }

    if let Ok(numstat) = git_output(repo_root, &["diff", "--numstat", &range]) {
        let mut files = Vec::new();
        let mut insertions = 0u64;
        let mut deletions = 0u64;
        let mut full_count = 0u64;
        for line in numstat.lines() {
            let mut parts = line.split('\t');
            let add = parts.next().unwrap_or("0");
            let del = parts.next().unwrap_or("0");
            let path = parts.next().unwrap_or("").trim();
            if path.is_empty() {
                continue;
            }
            full_count += 1;
            if add != "-" {
                insertions += add.parse::<u64>().unwrap_or(0);
            }
            if del != "-" {
                deletions += del.parse::<u64>().unwrap_or(0);
            }
            if files.len() < COMMIT_FILES_CAP {
                files.push(path.to_string());
            }
        }
        // Dropped separate `diff --name-only` spawn; numstat already lists paths.
        out.files_changed_count = Some(full_count);
        if !files.is_empty() {
            out.files_changed = Some(files);
        }
        out.insertions = Some(insertions);
        out.deletions = Some(deletions);
    }

    if let Some(prev) = previous_head.filter(|p| !p.is_empty() && *p != head_sha) {
        if let Ok(log) = git_output(
            repo_root,
            &["log", "--format=%H", &format!("{prev}..{head_sha}")],
        ) {
            let shas: Vec<String> = log
                .lines()
                .map(str::trim)
                .filter(|l| !l.is_empty())
                .take(INCLUDED_COMMITS_CAP)
                .map(str::to_string)
                .collect();
            if shas.len() > 1 {
                out.included_commits = Some(shas);
            }
        }
    }

    out
}

fn non_empty_owned(value: Option<&str>) -> Option<String> {
    value
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(str::to_string)
}

/// Drain a burst of notify events so one tick batches many FS writes into one
/// jsonl flush instead of open/append per event under heavy editors.
fn drain_notify_batch(
    rx: &mpsc::Receiver<Result<Event, notify::Error>>,
    first: Result<Event, notify::Error>,
) -> Vec<Result<Event, notify::Error>> {
    let mut batch = Vec::with_capacity(8);
    batch.push(first);
    while batch.len() < NOTIFY_EVENT_DRAIN_CAP {
        match rx.try_recv() {
            Ok(next) => batch.push(next),
            Err(_) => break,
        }
    }
    batch
}

fn append_unique_commit_event(
    layout: &SpoolLayout,
    commit: &WorktreeCommitEvent,
) -> Result<bool, String> {
    let fingerprint = commit
        .fingerprint
        .clone()
        .unwrap_or_else(|| commit_event_fingerprint(commit));
    if !claim_commit_fingerprint(layout, &fingerprint)? {
        return Ok(false);
    }

    let mut commit = commit.clone();
    commit.fingerprint = Some(fingerprint.clone());
    if commit.id.is_empty() {
        commit.id = stable_event_id(&fingerprint);
    }
    if let Err(error) = append_json_line(&layout.commits_file, &commit) {
        release_commit_fingerprint_claim(layout, &fingerprint);
        return Err(error);
    }
    Ok(true)
}

fn commit_event_fingerprint(commit: &WorktreeCommitEvent) -> String {
    let payload = json!({
        "schema": 3,
        "repoOwner": commit.repo_owner,
        "repoName": commit.repo_name,
        "branchName": commit.branch_name,
        "previousHeadSha": commit.previous_head_sha,
        "headSha": commit.head_sha,
    });
    let encoded = serde_json::to_vec(&payload).unwrap_or_default();
    let mut hasher = Sha256::new();
    hasher.update(encoded);
    format!("{:x}", hasher.finalize())
}

fn claim_commit_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
    let marker = commit_dedupe_marker_path(layout, fingerprint);
    if marker.is_file() {
        return Ok(false);
    }
    let claim_key = format!("{}|{fingerprint}", layout.root.display());
    {
        let mut claimed = commit_claims_in_process()
            .lock()
            .map_err(|_| "commit claim lock poisoned".to_string())?;
        if !claimed.insert(claim_key) {
            return Ok(false);
        }
    }
    let claims = layout.root.join("commit-claims.jsonl");
    append_json_line(
        &claims,
        &json!({
            "fingerprint": fingerprint,
            "firstSeenAt": Utc::now().to_rfc3339(),
        }),
    )?;
    Ok(true)
}

fn release_commit_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
    let _ = fs::remove_file(commit_dedupe_marker_path(layout, fingerprint));
    let claim_key = format!("{}|{fingerprint}", layout.root.display());
    if let Ok(mut claimed) = commit_claims_in_process().lock() {
        claimed.remove(&claim_key);
    }
}

fn commit_claims_in_process() -> &'static Mutex<BTreeSet<String>> {
    static CLAIMS: OnceLock<Mutex<BTreeSet<String>>> = OnceLock::new();
    CLAIMS.get_or_init(|| Mutex::new(BTreeSet::new()))
}

fn commit_dedupe_marker_path(layout: &SpoolLayout, fingerprint: &str) -> PathBuf {
    layout.commit_dedupe_dir.join(format!("{fingerprint}.seen"))
}

async fn sync_spool_for_identity(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
) -> Result<(), String> {
    let candidates = collect_sync_candidates(layout)?;
    if candidates.is_empty() {
        return Ok(());
    }

    sync_candidates(identity, candidates, false).await
}

async fn sync_candidates(
    identity: &RepoIdentity,
    candidates: Vec<SyncCandidate>,
    resync: bool,
) -> Result<(), String> {
    let mut events = Vec::new();
    let mut commits = Vec::new();
    for candidate in &candidates {
        match &candidate.records {
            SyncRecords::Events(records) => events.extend(records.iter().cloned()),
            SyncRecords::Commits(records) => commits.extend(records.iter().cloned()),
        }
    }

    let token = resolve_cli_access_token()?;
    let device = resolve_device_identity()?;
    let client = cli_request_client()?;
    let api = ApiConfig::from_env();
    let endpoint = api.cli_worktree_mutations_endpoint();
    let event_count = events.len();
    let commit_count = commits.len();
    let hardware_id = worktree_device_hardware_id(&device.hardware_id);
    let hostname = current_hostname();
    let platform = std::env::consts::OS.to_string();
    let runtime_key = worktree_runtime_key();
    let repo_root = normalize_repo_root_for_payload(&identity.root);
    // Ensure every record has a stable fingerprint/id before upload (legacy spool rows).
    let events = events
        .into_iter()
        .map(ensure_mutation_event_identity)
        .collect::<Vec<_>>();
    let commits = commits
        .into_iter()
        .map(ensure_commit_event_identity)
        .collect::<Vec<_>>();
    let batches = split_worktree_mutation_upload_batches(events.clone(), commits.clone());
    let batch_count = batches.len();

    if batch_count > 1 {
        println!(
            "Uploading {} worktree mutation record(s) in {} batch(es) for {}/{} on {} [{}].",
            event_count + commit_count,
            batch_count,
            identity.owner,
            identity.name,
            identity.branch,
            runtime_key
        );
    }

    let mut status_code = 202;
    for attempt in 1..=WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS {
        let attempt_batches =
            split_worktree_mutation_upload_batches(events.clone(), commits.clone());
        match upload_worktree_mutation_batches(
            &client,
            &endpoint,
            &token,
            &hardware_id,
            hostname.as_deref(),
            &platform,
            &runtime_key,
            identity,
            &repo_root,
            attempt_batches,
        )
        .await
        {
            Ok(code) => {
                status_code = code;
                break;
            }
            Err(error)
                if error.kind == WorktreeMutationSyncErrorKind::Retryable
                    && attempt < WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS =>
            {
                eprintln!(
                    "Worktree mutation sync attempt {attempt}/{} failed for {}/{} on {}: {} Restarting from the first batch in {} second(s).",
                    WORKTREE_MUTATION_SYNC_MAX_ATTEMPTS,
                    identity.owner,
                    identity.name,
                    identity.branch,
                    error.message,
                    WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS
                );
                tokio::time::sleep(Duration::from_secs(
                    WORKTREE_MUTATION_SYNC_RETRY_DELAY_SECONDS,
                ))
                .await;
            }
            Err(error) => return Err(error.message),
        }
    }

    let layout = prepare_spool_layout(identity)?;
    append_sync_log(SyncLogAppend {
        identity,
        layout: &layout,
        endpoint: &endpoint,
        status_code,
        event_count,
        commit_count,
        candidates: &candidates,
        resync,
    })?;

    for candidate in candidates {
        if !is_synced_spool_path(&candidate.path) {
            mark_synced(&candidate.path)?;
        }
    }

    Ok(())
}

async fn upload_worktree_mutation_batches(
    client: &reqwest::Client,
    endpoint: &str,
    token: &str,
    hardware_id: &str,
    hostname: Option<&str>,
    platform: &str,
    runtime_key: &str,
    identity: &RepoIdentity,
    repo_root: &str,
    batches: Vec<WorktreeMutationUploadBatch>,
) -> Result<u16, WorktreeMutationSyncError> {
    let batch_count = batches.len();
    let mut status_code = 202;

    for (batch_index, batch) in batches.into_iter().enumerate() {
        let batch_event_count = batch.events.len();
        let batch_commit_count = batch.commits.len();
        let payload = WorktreeMutationIngestPayload {
            device: WorktreeMutationDevicePayload {
                hardware_id: hardware_id.to_string(),
                device_name: hostname.map(str::to_string),
                hostname: hostname.map(str::to_string),
                platform: platform.to_string(),
                runtime_key: Some(runtime_key.to_string()),
            },
            repository: WorktreeMutationRepositoryPayload {
                owner: identity.owner.clone(),
                name: identity.name.clone(),
                branch_name: identity.branch.clone(),
                repo_root: repo_root.to_string(),
            },
            events: batch.events,
            commits: batch.commits,
        };

        let response = client
            .post(endpoint)
            .bearer_auth(token)
            .json(&payload)
            .send()
            .await
            .map_err(|error| WorktreeMutationSyncError {
                kind: WorktreeMutationSyncErrorKind::Retryable,
                message: format!(
                    "Failed to upload worktree mutation batch {}/{}: {error}",
                    batch_index + 1,
                    batch_count
                ),
            })?;

        if response.status() == StatusCode::UNAUTHORIZED {
            return Err(WorktreeMutationSyncError {
                kind: WorktreeMutationSyncErrorKind::NonRetryable,
                message: "Your stored CLI session is no longer valid. Run `xbp login` again."
                    .to_string(),
            });
        }

        if !response.status().is_success() {
            let status = response.status();
            let body = response.text().await.unwrap_or_default();
            return Err(WorktreeMutationSyncError {
                kind: if should_retry_worktree_mutation_sync_status(status) {
                    WorktreeMutationSyncErrorKind::Retryable
                } else {
                    WorktreeMutationSyncErrorKind::NonRetryable
                },
                message: format!(
                    "Worktree mutation upload batch {}/{} failed with {status}: {body}",
                    batch_index + 1,
                    batch_count
                ),
            });
        }

        status_code = response.status().as_u16();
        if batch_count > 1 {
            println!(
                "Uploaded worktree mutation batch {}/{} ({} event(s), {} commit(s)).",
                batch_index + 1,
                batch_count,
                batch_event_count,
                batch_commit_count
            );
        }
    }

    Ok(status_code)
}

fn should_retry_worktree_mutation_sync_status(status: StatusCode) -> bool {
    status.is_server_error()
        || status == StatusCode::REQUEST_TIMEOUT
        || status == StatusCode::TOO_MANY_REQUESTS
}

fn split_worktree_mutation_upload_batches(
    events: Vec<WorktreeMutationEvent>,
    commits: Vec<WorktreeCommitEvent>,
) -> Vec<WorktreeMutationUploadBatch> {
    let mut batches = Vec::new();
    let mut current = empty_worktree_mutation_upload_batch();

    for event in events {
        let estimated_json_bytes = serialized_json_len(&event);
        if should_start_next_worktree_mutation_upload_batch(&current, estimated_json_bytes) {
            batches.push(current);
            current = empty_worktree_mutation_upload_batch();
        }
        current.estimated_json_bytes += estimated_json_bytes;
        current.events.push(event);
    }

    for commit in commits {
        let estimated_json_bytes = serialized_json_len(&commit);
        if should_start_next_worktree_mutation_upload_batch(&current, estimated_json_bytes) {
            batches.push(current);
            current = empty_worktree_mutation_upload_batch();
        }
        current.estimated_json_bytes += estimated_json_bytes;
        current.commits.push(commit);
    }

    if current_record_count(&current) > 0 {
        batches.push(current);
    }

    batches
}

fn empty_worktree_mutation_upload_batch() -> WorktreeMutationUploadBatch {
    WorktreeMutationUploadBatch {
        events: Vec::new(),
        commits: Vec::new(),
        estimated_json_bytes: 0,
    }
}

fn should_start_next_worktree_mutation_upload_batch(
    batch: &WorktreeMutationUploadBatch,
    next_record_json_bytes: usize,
) -> bool {
    if current_record_count(batch) == 0 {
        return false;
    }

    current_record_count(batch) >= WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT
        || batch.estimated_json_bytes + next_record_json_bytes
            > WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES
}

fn current_record_count(batch: &WorktreeMutationUploadBatch) -> usize {
    batch.events.len() + batch.commits.len()
}

fn serialized_json_len<T: Serialize>(value: &T) -> usize {
    serde_json::to_vec(value)
        .map(|encoded| encoded.len())
        .unwrap_or(0)
}

fn collect_sync_candidates(layout: &SpoolLayout) -> Result<Vec<SyncCandidate>, String> {
    // Readers must see buffered appends (batch flush is tick-based).
    flush_jsonl_write_buffers();
    collect_spool_candidates(layout, false)
}

fn print_no_sync_candidates_message(
    target: &WorktreeWatchTargetOptions,
    resync: bool,
) -> Result<(), String> {
    if resync {
        println!("No local worktree mutation JSONL files found.");
        return Ok(());
    }

    let identities = resolve_target_identities(target)?;
    let mut synced_files = 0usize;
    let mut synced_records = 0usize;
    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        let all_candidates = collect_spool_candidates(&layout, true)?;
        let unsynced_candidates = collect_sync_candidates(&layout)?;
        synced_files += all_candidates
            .len()
            .saturating_sub(unsynced_candidates.len());
        let all_records = all_candidates
            .iter()
            .map(|candidate| candidate.records.len())
            .sum::<usize>();
        let unsynced_records = unsynced_candidates
            .iter()
            .map(|candidate| candidate.records.len())
            .sum::<usize>();
        synced_records += all_records.saturating_sub(unsynced_records);
    }

    if synced_files > 0 {
        println!(
            "No unsynced worktree mutation files found. Found {synced_files} already-synced local spool file(s) with {synced_records} record(s). Use `xbp worktree-watch sync --resync` to upload them again."
        );
    } else {
        println!("No unsynced worktree mutation files found.");
    }
    Ok(())
}

fn collect_spool_candidates(
    layout: &SpoolLayout,
    include_synced: bool,
) -> Result<Vec<SyncCandidate>, String> {
    // Directory scans only see paths that already exist on disk.
    flush_jsonl_write_buffers();
    // When the layout root is a branch spool, also scan sibling legacy roots for
    // the same owner/repo/branch so WSL can see Windows-written files and vice versa.
    let mut roots = vec![layout.root.clone()];
    if let Some(branch) = layout.root.file_name().and_then(|name| name.to_str()) {
        if let Some(repo_root) = layout.root.parent() {
            if let (Some(name), Some(owner)) = (
                repo_root.file_name().and_then(|n| n.to_str()),
                repo_root.parent().and_then(|p| p.file_name()).and_then(|n| n.to_str()),
            ) {
                let identity = RepoIdentity {
                    owner: owner.to_string(),
                    name: name.to_string(),
                    branch: branch.to_string(),
                    root: PathBuf::new(),
                    head_sha: None,
                };
                if let Ok(search_roots) = repository_spool_search_roots(&identity) {
                    for search_root in search_roots {
                        let branch_root = search_root.join(branch);
                        if !roots.iter().any(|existing| {
                            path_identity_key(existing) == path_identity_key(&branch_root)
                        }) {
                            roots.push(branch_root);
                        }
                    }
                }
            }
        }
    }

    let mut candidates = Vec::new();
    let mut seen_files = BTreeSet::new();
    for root in roots {
        collect_spool_candidates_from_root(&root, include_synced, &mut candidates, &mut seen_files)?;
    }
    Ok(candidates)
}

fn collect_spool_candidates_from_root(
    root: &Path,
    include_synced: bool,
    candidates: &mut Vec<SyncCandidate>,
    seen_files: &mut BTreeSet<String>,
) -> Result<(), String> {
    if !root.exists() {
        return Ok(());
    }

    for entry in fs::read_dir(root).map_err(|error| {
        format!(
            "Failed to read spool directory {}: {error}",
            root.display()
        )
    })? {
        let path = entry
            .map_err(|error| format!("Failed to read spool entry: {error}"))?
            .path();
        let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
            continue;
        };
        if !file_name.ends_with(".jsonl") || (!include_synced && file_name.contains(".synced.")) {
            continue;
        }

        let kind = if file_name.starts_with("events-") {
            SyncFileKind::Events
        } else if file_name.starts_with("commits-") {
            SyncFileKind::Commits
        } else {
            continue;
        };

        let identity_key = path_identity_key(&path);
        if !seen_files.insert(identity_key) {
            continue;
        }

        let records = sanitize_spool_records(&path, kind, read_spool_records(&path, kind)?)?;
        if records.is_empty() {
            continue;
        }
        candidates.push(SyncCandidate { path, records });
    }

    Ok(())
}

fn append_sync_log(request: SyncLogAppend<'_>) -> Result<(), String> {
    let entry = WorktreeWatchSyncLogEntry {
        id: Uuid::new_v4().to_string(),
        synced_at: Utc::now(),
        endpoint: request.endpoint.to_string(),
        repository_owner: request.identity.owner.clone(),
        repository_name: request.identity.name.clone(),
        branch_name: request.identity.branch.clone(),
        repo_root: request.identity.root.display().to_string(),
        resync: request.resync,
        status_code: request.status_code,
        event_count: request.event_count,
        commit_count: request.commit_count,
        spool_file_count: request.candidates.len(),
        spool_files: request
            .candidates
            .iter()
            .map(|candidate| candidate.path.display().to_string())
            .collect(),
    };
    append_json_line(&request.layout.sync_log_file, &entry)
}

fn sanitize_spool_records(
    path: &Path,
    kind: SyncFileKind,
    records: SyncRecords,
) -> Result<SyncRecords, String> {
    match (kind, records) {
        (SyncFileKind::Events, SyncRecords::Events(records)) => {
            let mut sanitized = Vec::with_capacity(records.len());
            let mut changed = false;

            for record in records {
                let original = serde_json::to_value(&record).ok();
                let Some(record) = sanitize_spooled_event(record) else {
                    changed = true;
                    continue;
                };
                if !changed {
                    changed = serde_json::to_value(&record).ok() != original;
                }
                sanitized.push(record);
            }

            if changed {
                rewrite_jsonl_records(path, &sanitized)?;
            }

            Ok(SyncRecords::Events(sanitized))
        }
        (_, records) => Ok(records),
    }
}

fn sanitize_spooled_event(mut event: WorktreeMutationEvent) -> Option<WorktreeMutationEvent> {
    event.paths = dedupe_filtered_spool_paths(event.paths);
    event.primary_path = sanitize_optional_spool_path(event.primary_path);
    event.old_path = sanitize_optional_spool_path(event.old_path);
    event.new_path = sanitize_optional_spool_path(event.new_path);

    if event.paths.is_empty() {
        if let Some(path) = event.primary_path.clone() {
            event.paths.push(path);
        }
        if let Some(path) = event.old_path.clone() {
            if !event.paths.iter().any(|existing| existing == &path) {
                event.paths.push(path);
            }
        }
        if let Some(path) = event.new_path.clone() {
            if !event.paths.iter().any(|existing| existing == &path) {
                event.paths.push(path);
            }
        }
    }

    if event.primary_path.is_none() {
        event.primary_path = event.paths.first().cloned();
    }

    if event.paths.is_empty() {
        return None;
    }

    Some(event)
}

fn sanitize_optional_spool_path(path: Option<String>) -> Option<String> {
    path.and_then(sanitize_spool_path)
}

fn sanitize_spool_path(path: String) -> Option<String> {
    let trimmed = path.trim();
    if trimmed.is_empty() || watch_ignore_rules().ignores_relative(trimmed) {
        return None;
    }
    Some(trimmed.replace('\\', "/"))
}

fn dedupe_filtered_spool_paths(paths: Vec<String>) -> Vec<String> {
    let mut sanitized = Vec::with_capacity(paths.len());
    for path in paths {
        let Some(path) = sanitize_spool_path(path) else {
            continue;
        };
        if !sanitized.iter().any(|existing| existing == &path) {
            sanitized.push(path);
        }
    }
    sanitized
}

fn rewrite_jsonl_records<T: Serialize>(path: &Path, records: &[T]) -> Result<(), String> {
    if records.is_empty() {
        if path.exists() {
            fs::remove_file(path).map_err(|error| {
                format!(
                    "Failed to remove emptied spool file {}: {error}",
                    path.display()
                )
            })?;
        }
        return Ok(());
    }

    let temp_path = path.with_extension(format!(
        "{}.{}.tmp",
        path.extension()
            .and_then(|value| value.to_str())
            .unwrap_or("jsonl"),
        Uuid::new_v4()
    ));
    {
        let mut file = File::create(&temp_path).map_err(|error| {
            format!(
                "Failed to create temporary spool file {}: {error}",
                temp_path.display()
            )
        })?;
        for record in records {
            let line = serde_json::to_string(record)
                .map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
            writeln!(file, "{line}").map_err(|error| {
                format!(
                    "Failed to write temporary spool file {}: {error}",
                    temp_path.display()
                )
            })?;
        }
    }

    if path.exists() {
        fs::remove_file(path).map_err(|error| {
            format!(
                "Failed to replace rewritten spool file {}: {error}",
                path.display()
            )
        })?;
    }

    fs::rename(&temp_path, path)
        .map_err(|error| format!("Failed to rewrite spool file {}: {error}", path.display()))
}

fn read_last_sync_log_entry(path: &Path) -> Result<Option<WorktreeWatchSyncLogEntry>, String> {
    if !path.exists() {
        return Ok(None);
    }
    let values = read_jsonl_values(path)?;
    let Some(value) = values.into_iter().last() else {
        return Ok(None);
    };
    serde_json::from_value(value)
        .map(Some)
        .map_err(|error| format!("Failed to decode sync log {}: {error}", path.display()))
}

fn is_synced_spool_path(path: &Path) -> bool {
    path.file_name()
        .and_then(|value| value.to_str())
        .map(|value| value.contains(".synced."))
        .unwrap_or(false)
}

fn mark_synced(path: &Path) -> Result<(), String> {
    let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
        return Ok(());
    };
    let synced_name = file_name.replace(
        ".jsonl",
        &format!(".synced.{}.jsonl", Utc::now().timestamp()),
    );
    let synced_path = path.with_file_name(synced_name);
    fs::rename(path, &synced_path).map_err(|error| {
        format!(
            "Failed to mark spool file {} as synced: {error}",
            path.display()
        )
    })
}

fn read_spool_records(path: &Path, kind: SyncFileKind) -> Result<SyncRecords, String> {
    flush_jsonl_write_buffers();
    match kind {
        SyncFileKind::Events => read_jsonl_records::<WorktreeMutationEvent>(path)
            .map(SyncRecords::Events)
            .map_err(|error| format!("Failed to decode event from {}: {error}", path.display())),
        SyncFileKind::Commits => read_jsonl_records::<WorktreeCommitEvent>(path)
            .map(SyncRecords::Commits)
            .map_err(|error| format!("Failed to decode commit from {}: {error}", path.display())),
    }
}

/// Strip whitespace and sparse-file / torn-write NUL padding before JSON parse.
/// Windows concurrent appends have produced lines that start with long `\0` runs
/// then valid JSON; parsing the raw line yields "expected value at line 1 column 1".
fn sanitize_jsonl_line(line: &str) -> Option<&str> {
    let trimmed = line.trim().trim_matches('\0').trim();
    if trimmed.is_empty() {
        None
    } else {
        Some(trimmed)
    }
}

fn read_jsonl_records<T>(path: &Path) -> Result<Vec<T>, String>
where
    T: for<'de> Deserialize<'de>,
{
    flush_jsonl_write_buffers();
    let file = File::open(path)
        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
    let reader = BufReader::new(file);
    let mut records = Vec::new();
    for (line_no, line) in reader.lines().enumerate() {
        let line =
            line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
        let Some(payload) = sanitize_jsonl_line(&line) else {
            continue;
        };
        match serde_json::from_str(payload) {
            Ok(record) => records.push(record),
            Err(error) => {
                // Skip unparseable lines (partial writes, encoding glitches) rather than
                // killing worktree-watch / dedupe load for the whole repo.
                eprintln!(
                    "warning: skipping unparseable JSONL record in {} at line {}: {error}",
                    path.display(),
                    line_no + 1
                );
            }
        }
    }
    Ok(records)
}

fn read_jsonl_values(path: &Path) -> Result<Vec<Value>, String> {
    read_jsonl_records(path)
}

fn spawn_detached_worktree_watch(repo: Option<&Path>) -> Result<PathBuf, String> {
    let identity = resolve_repo_identity(repo)?;
    spawn_detached_worktree_watch_for_identity(&identity)
}

fn spawn_detached_parent_worktree_watch(parent: &Path) -> Result<PathBuf, String> {
    let parent = canonical_parent_path(parent)?;
    let layout = prepare_parent_spool_layout(&parent)?;
    replace_existing_parent_watcher(&parent, &layout)?;
    let executable = std::env::current_exe()
        .map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
    let mut command = Command::new(&executable);
    command
        .arg("worktree-watch")
        .arg("start")
        .arg("--parent")
        .arg(&parent)
        .env(BACKGROUND_CHILD_ENV, "1")
        .current_dir(&parent)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    configure_detached_child_process(&mut command);

    let child = command
        .spawn()
        .map_err(|error| format!("Failed to spawn background parent worktree watcher: {error}"))?;
    write_parent_watcher_state(&parent, &layout, child.id(), &executable)?;
    Ok(parent)
}

fn spawn_detached_worktree_watch_for_identity(identity: &RepoIdentity) -> Result<PathBuf, String> {
    let layout = prepare_spool_layout(identity)?;
    replace_existing_watcher(identity, &layout)?;
    let executable = std::env::current_exe()
        .map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
    let mut command = Command::new(&executable);
    command
        .arg("worktree-watch")
        .arg("start")
        .arg("--repo")
        .arg(&identity.root)
        .env(BACKGROUND_CHILD_ENV, "1")
        .current_dir(&identity.root)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    configure_detached_child_process(&mut command);

    let child: std::process::Child = command
        .spawn()
        .map_err(|error| format!("Failed to spawn background worktree watcher: {error}"))?;
    write_watcher_state(identity, &layout, child.id(), &executable)?;
    Ok(identity.root.clone())
}

fn configure_detached_child_process(command: &mut Command) {
    #[cfg(windows)]
    {
        use std::os::windows::process::CommandExt;
        command.creation_flags(CREATE_NO_WINDOW);
    }

    // On Unix (including WSL), start a new session so the watcher survives the
    // launching shell and does not receive SIGHUP when the parent exits.
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        // SAFETY: setsid is called only in the child just after fork, before exec.
        unsafe {
            command.pre_exec(|| {
                extern "C" {
                    fn setsid() -> i32;
                }
                let _ = setsid();
                Ok(())
            });
        }
    }
}

fn replace_existing_parent_watcher(
    parent: &Path,
    layout: &ParentSpoolLayout,
) -> Result<(), String> {
    let Some(state) = read_parent_watcher_state_any(&layout.watcher_state_file)? else {
        return Ok(());
    };

    if path_identity_key(parent) != path_identity_key(Path::new(&state.parent_root)) {
        return Ok(());
    }

    if is_matching_parent_watcher_process(&state) {
        stop_parent_process(&state, false)?;
        println!(
            "Replaced existing background parent worktree watcher {} for {}",
            state.pid,
            parent.display()
        );
        remove_all_watcher_state_files(&layout.watcher_state_file);
        return Ok(());
    }

    if parent_watcher_state_pid_exists_with_same_executable(&state) {
        return Err(format!(
            "Existing parent watcher state for {} points to running XBP process {}, but its command line could not be verified. Run `xbp worktree-watch stop --parent \"{}\" --force` before starting another detached parent watcher.",
            parent.display(),
            state.pid,
            parent.display()
        ));
    }

    remove_all_watcher_state_files(&layout.watcher_state_file);
    Ok(())
}

fn replace_existing_watcher(identity: &RepoIdentity, layout: &SpoolLayout) -> Result<(), String> {
    let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
        return Ok(());
    };

    if !same_repo_watcher_state(identity, &state) {
        return Ok(());
    }

    if is_matching_watcher_process(&state) {
        stop_process(&state, false)?;
        println!(
            "Replaced existing background worktree watcher {} for {}",
            state.pid,
            identity.root.display()
        );
        remove_all_watcher_state_files(&layout.watcher_state_file);
        return Ok(());
    }

    if watcher_state_pid_exists_with_same_executable(&state) {
        return Err(format!(
            "Existing watcher state for {} points to running XBP process {}, but its command line could not be verified. Run `xbp worktree-watch stop --repo \"{}\" --force` before starting another detached watcher.",
            identity.root.display(),
            state.pid,
            identity.root.display()
        ));
    }

    remove_all_watcher_state_files(&layout.watcher_state_file);
    Ok(())
}

fn stop_existing_watcher(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    force: bool,
) -> Result<StopWatcherOutcome, String> {
    let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
        // Still purge orphaned legacy/runtime state files so stop is idempotent.
        if remove_all_watcher_state_files(&layout.watcher_state_file) > 0 {
            return Ok(StopWatcherOutcome::RemovedStaleState);
        }
        return Ok(StopWatcherOutcome::NoState);
    };

    if !same_repo_watcher_state(identity, &state) {
        return Ok(StopWatcherOutcome::NoState);
    }

    if is_matching_watcher_process(&state) || force {
        let pid = state.pid;
        stop_process(&state, force)?;
        remove_all_watcher_state_files(&layout.watcher_state_file);
        return Ok(StopWatcherOutcome::Stopped(pid));
    }

    if watcher_state_pid_exists_with_same_executable(&state) {
        return Err(format!(
            "Watcher state for {} points to running XBP process {}, but its command line could not be verified. Re-run with `--force` to stop that PID.",
            identity.root.display(),
            state.pid
        ));
    }

    // Stale PID: remove legacy watcher-state.json AND runtime-scoped copies.
    remove_all_watcher_state_files(&layout.watcher_state_file);
    Ok(StopWatcherOutcome::RemovedStaleState)
}

fn stop_existing_parent_watcher(parent: &Path, force: bool) -> Result<StopWatcherOutcome, String> {
    let layout = prepare_parent_spool_layout(parent)?;
    let Some(state) = read_parent_watcher_state_any(&layout.watcher_state_file)? else {
        if remove_all_watcher_state_files(&layout.watcher_state_file) > 0 {
            return Ok(StopWatcherOutcome::RemovedStaleState);
        }
        return Ok(StopWatcherOutcome::NoState);
    };

    if path_identity_key(parent) != path_identity_key(Path::new(&state.parent_root)) {
        return Ok(StopWatcherOutcome::NoState);
    }

    if is_matching_parent_watcher_process(&state) || force {
        let pid = state.pid;
        stop_parent_process(&state, force)?;
        remove_all_watcher_state_files(&layout.watcher_state_file);
        return Ok(StopWatcherOutcome::Stopped(pid));
    }

    if parent_watcher_state_pid_exists_with_same_executable(&state) {
        return Err(format!(
            "Parent watcher state for {} points to running XBP process {}, but its command line could not be verified. Re-run with `--force` to stop that PID.",
            parent.display(),
            state.pid
        ));
    }

    remove_all_watcher_state_files(&layout.watcher_state_file);
    Ok(StopWatcherOutcome::RemovedStaleState)
}

/// Remove every `watcher-state*.json` next to the primary state path.
///
/// Historical bug: stop only deleted the runtime-scoped file
/// (`watcher-state-windows.json`) while `read_watcher_state` still fell back to
/// legacy `watcher-state.json`, so every subsequent `stop` reported
/// "Removed stale …" forever.
fn remove_all_watcher_state_files(primary: &Path) -> usize {
    let mut removed = 0usize;
    let mut seen = BTreeSet::new();

    for candidate in watcher_state_read_candidates(primary) {
        if !seen.insert(candidate.clone()) {
            continue;
        }
        if candidate.is_file() && fs::remove_file(&candidate).is_ok() {
            removed += 1;
        }
    }

    if let Some(dir) = primary.parent() {
        if let Ok(entries) = fs::read_dir(dir) {
            for entry in entries.flatten() {
                let path = entry.path();
                let Some(name) = path.file_name().and_then(|v| v.to_str()) else {
                    continue;
                };
                if !(name.starts_with("watcher-state") && name.ends_with(".json")) {
                    continue;
                }
                if !seen.insert(path.clone()) {
                    continue;
                }
                if path.is_file() && fs::remove_file(&path).is_ok() {
                    removed += 1;
                }
            }
        }
    }

    removed
}

fn read_watcher_state(path: &Path) -> Result<Option<WorktreeWatcherState>, String> {
    // Prefer runtime-scoped state; fall back to legacy shared watcher-state.json.
    for candidate in watcher_state_read_candidates(path) {
        if !candidate.exists() {
            continue;
        }
        let raw = fs::read_to_string(&candidate).map_err(|error| {
            format!(
                "Failed to read watcher state {}: {error}",
                candidate.display()
            )
        })?;
        return serde_json::from_str(&raw).map(Some).map_err(|error| {
            format!(
                "Failed to parse watcher state {}: {error}",
                candidate.display()
            )
        });
    }
    Ok(None)
}

fn watcher_state_read_candidates(path: &Path) -> Vec<PathBuf> {
    let mut candidates = vec![path.to_path_buf()];
    if let Some(parent) = path.parent() {
        let legacy = parent.join("watcher-state.json");
        if legacy != path {
            candidates.push(legacy);
        }
    }
    candidates
}

fn read_parent_watcher_state(path: &Path) -> Result<Option<ParentWorktreeWatcherState>, String> {
    if !path.exists() {
        return Ok(None);
    }
    let raw = fs::read_to_string(path).map_err(|error| {
        format!(
            "Failed to read parent watcher state {}: {error}",
            path.display()
        )
    })?;
    serde_json::from_str(&raw).map(Some).map_err(|error| {
        format!(
            "Failed to parse parent watcher state {}: {error}",
            path.display()
        )
    })
}

/// Read parent state from runtime-scoped path or legacy fallback (same candidates as repo watchers).
fn read_parent_watcher_state_any(
    primary: &Path,
) -> Result<Option<ParentWorktreeWatcherState>, String> {
    for candidate in watcher_state_read_candidates(primary) {
        if let Some(state) = read_parent_watcher_state(&candidate)? {
            return Ok(Some(state));
        }
    }
    Ok(None)
}

fn write_watcher_state(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    pid: u32,
    executable: &Path,
) -> Result<(), String> {
    let state = WorktreeWatcherState {
        pid,
        repo_root: normalize_repo_root_for_payload(&identity.root),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        executable: executable.display().to_string(),
        started_at: Utc::now(),
        runtime_key: Some(worktree_runtime_key()),
        platform: Some(std::env::consts::OS.to_string()),
    };
    let rendered = serde_json::to_string_pretty(&state)
        .map_err(|error| format!("Failed to serialize watcher state: {error}"))?;
    fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
        format!(
            "Failed to write watcher state {}: {error}",
            layout.watcher_state_file.display()
        )
    })
}

fn write_parent_watcher_state(
    parent: &Path,
    layout: &ParentSpoolLayout,
    pid: u32,
    executable: &Path,
) -> Result<(), String> {
    let state = ParentWorktreeWatcherState {
        pid,
        parent_root: normalize_repo_root_for_payload(parent),
        executable: executable.display().to_string(),
        started_at: Utc::now(),
        runtime_key: Some(worktree_runtime_key()),
        platform: Some(std::env::consts::OS.to_string()),
    };
    let rendered = serde_json::to_string_pretty(&state)
        .map_err(|error| format!("Failed to serialize parent watcher state: {error}"))?;
    fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
        format!(
            "Failed to write parent watcher state {}: {error}",
            layout.watcher_state_file.display()
        )
    })
}

fn same_repo_watcher_state(identity: &RepoIdentity, state: &WorktreeWatcherState) -> bool {
    if !runtime_keys_compatible(state.runtime_key.as_deref()) {
        return false;
    }
    path_identity_key(&identity.root) == path_identity_key(Path::new(&state.repo_root))
        && identity.owner == state.repository_owner
        && identity.name == state.repository_name
        && identity.branch == state.branch_name
}

fn runtime_keys_compatible(state_runtime: Option<&str>) -> bool {
    let current = worktree_runtime_key();
    match state_runtime {
        Some(key) => key == current,
        // Legacy state files (pre-runtime isolation) only match native Windows/Linux
        // paths that still live outside the runtimes/ spool tree — never cross WSL.
        None => !is_wsl_runtime(),
    }
}

fn process_system_for_pids(pids: &[u32]) -> System {
    let mut system = System::new();
    if pids.is_empty() {
        return system;
    }
    let pid_list: Vec<Pid> = pids
        .iter()
        .copied()
        .filter(|pid| *pid != 0 && *pid != std::process::id())
        .map(Pid::from_u32)
        .collect();
    if pid_list.is_empty() {
        return system;
    }
    // Refresh only the candidate watcher PIDs (not the entire process table).
    system.refresh_processes_specifics(
        ProcessesToUpdate::Some(&pid_list),
        true,
        ProcessRefreshKind::everything()
            .with_cmd(UpdateKind::Always)
            .with_exe(UpdateKind::Always),
    );
    system
}

fn is_matching_watcher_process(state: &WorktreeWatcherState) -> bool {
    let system = process_system_for_pids(&[state.pid]);
    // "Matching other process" — used by stop/replace; never treats self as match.
    is_matching_watcher_process_with_system(&system, state)
}

/// True when the state PID is a live watcher, including this process (status/API).
fn is_live_watcher_process_with_system(system: &System, state: &WorktreeWatcherState) -> bool {
    if state.pid == std::process::id() {
        return runtime_keys_compatible(state.runtime_key.as_deref());
    }
    is_matching_watcher_process_with_system(system, state)
}

fn is_matching_watcher_process_with_system(
    system: &System,
    state: &WorktreeWatcherState,
) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return false;
    };
    let command_line = process
        .cmd()
        .iter()
        .map(|part| part.to_string_lossy())
        .collect::<Vec<_>>()
        .join(" ");
    let command_matches = command_line_contains_ignore_case(&command_line, "worktree-watch")
        && command_line_contains_ignore_case(&command_line, "start")
        && command_line_mentions_path(&command_line, &state.repo_root);
    command_matches || watcher_state_process_identity_matches(process, state)
}

fn is_matching_parent_watcher_process(state: &ParentWorktreeWatcherState) -> bool {
    let system = process_system_for_pids(&[state.pid]);
    is_matching_parent_watcher_process_with_system(&system, state)
}

/// True when the parent state PID is a live watcher, including this process.
fn is_live_parent_watcher_process_with_system(
    system: &System,
    state: &ParentWorktreeWatcherState,
) -> bool {
    if state.pid == std::process::id() {
        return runtime_keys_compatible(state.runtime_key.as_deref());
    }
    is_matching_parent_watcher_process_with_system(system, state)
}

fn is_matching_parent_watcher_process_with_system(
    system: &System,
    state: &ParentWorktreeWatcherState,
) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    if !runtime_keys_compatible(state.runtime_key.as_deref()) {
        return false;
    }
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return false;
    };
    let command_line = process
        .cmd()
        .iter()
        .map(|part| part.to_string_lossy())
        .collect::<Vec<_>>()
        .join(" ");
    let command_matches = command_line_contains_ignore_case(&command_line, "worktree-watch")
        && command_line_contains_ignore_case(&command_line, "start")
        && command_line_contains_ignore_case(&command_line, "--parent")
        && command_line_mentions_path(&command_line, &state.parent_root);
    // Accept: verified worktree-watch cmdline, or same executable at this PID
    // (Windows often omits args / start_time can be coarse after CREATE_NO_WINDOW).
    command_matches
        || parent_watcher_state_process_identity_matches(process, state)
        || (process_executable_matches_path(process, Path::new(&state.executable))
            && process_looks_like_worktree_watch(process))
}

fn process_looks_like_worktree_watch(process: &sysinfo::Process) -> bool {
    let command_line = process
        .cmd()
        .iter()
        .map(|part| part.to_string_lossy())
        .collect::<Vec<_>>()
        .join(" ");
    if command_line_contains_ignore_case(&command_line, "worktree-watch") {
        return true;
    }
    // Empty cmd() is common for some Windows query modes — exe match + state file is enough.
    command_line.trim().is_empty()
}

fn command_line_contains_ignore_case(haystack: &str, needle: &str) -> bool {
    haystack.to_ascii_lowercase().contains(&needle.to_ascii_lowercase())
}

fn command_line_mentions_path(command_line: &str, path: &str) -> bool {
    if path.is_empty() {
        return false;
    }
    if command_line.contains(path) {
        return true;
    }
    let command_lower = command_line.to_ascii_lowercase().replace('\\', "/");
    let path_lower = path.to_ascii_lowercase().replace('\\', "/");
    if command_lower.contains(&path_lower) {
        return true;
    }
    // Cross-platform: C:\Users\… vs /mnt/c/users/…
    let path_key = path_identity_key(Path::new(path));
    if !path_key.is_empty() && command_lower.contains(&path_key) {
        return true;
    }
    for token in command_line.split_whitespace() {
        let cleaned = token.trim_matches('"').trim_matches('\'');
        if cleaned.is_empty() {
            continue;
        }
        if path_identity_key(Path::new(cleaned)) == path_key {
            return true;
        }
    }
    // Final path segment (GitHub / github) case-insensitive.
    Path::new(path)
        .file_name()
        .and_then(|name| name.to_str())
        .map(|name| command_line_contains_ignore_case(command_line, name))
        .unwrap_or(false)
}

fn watcher_state_pid_exists_with_same_executable(state: &WorktreeWatcherState) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let system = process_system_for_pids(&[state.pid]);
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return false;
    };
    process_executable_matches_state(process, state)
}

fn parent_watcher_state_pid_exists_with_same_executable(
    state: &ParentWorktreeWatcherState,
) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let system = process_system_for_pids(&[state.pid]);
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return false;
    };
    process_executable_matches_path(process, Path::new(&state.executable))
}

fn watcher_state_process_identity_matches(
    process: &sysinfo::Process,
    state: &WorktreeWatcherState,
) -> bool {
    process_executable_matches_state(process, state)
        && process_start_time_matches(process.start_time(), state.started_at)
}

fn parent_watcher_state_process_identity_matches(
    process: &sysinfo::Process,
    state: &ParentWorktreeWatcherState,
) -> bool {
    process_executable_matches_path(process, Path::new(&state.executable))
        && process_start_time_matches(process.start_time(), state.started_at)
}

fn process_executable_matches_state(
    process: &sysinfo::Process,
    state: &WorktreeWatcherState,
) -> bool {
    process_executable_matches_path(process, Path::new(&state.executable))
}

fn process_executable_matches_path(process: &sysinfo::Process, executable: &Path) -> bool {
    let expected = path_identity_key(executable);
    process
        .exe()
        .map(|path| path_identity_key(path) == expected)
        .unwrap_or(false)
}

fn process_start_time_matches(process_started: u64, state_started: DateTime<Utc>) -> bool {
    // 0 means "unknown" on some platforms / query modes — don't reject the match.
    if process_started == 0 {
        return true;
    }
    let process_started = process_started as i64;
    let state_started = state_started.timestamp();
    // Wider window: spawn → write state can lag, and Windows clocks are coarse.
    (process_started - state_started).abs() <= 120
}

fn stop_process(state: &WorktreeWatcherState, force: bool) -> Result<(), String> {
    let system = process_system_for_pids(&[state.pid]);
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return Ok(());
    };
    if !force && !is_matching_watcher_process_with_system(&system, state) {
        return Ok(());
    }
    if process.kill() {
        Ok(())
    } else {
        Err(format!(
            "Failed to stop existing watcher process {}",
            state.pid
        ))
    }
}

fn stop_parent_process(state: &ParentWorktreeWatcherState, force: bool) -> Result<(), String> {
    let system = process_system_for_pids(&[state.pid]);
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return Ok(());
    };
    if !force && !is_matching_parent_watcher_process_with_system(&system, state) {
        return Ok(());
    }
    if process.kill() {
        Ok(())
    } else {
        Err(format!(
            "Failed to stop existing parent watcher process {}",
            state.pid
        ))
    }
}

/// Prefer a live parent watcher so bare `status` reports the active cover.
fn prefer_running_parent_root() -> Option<PathBuf> {
    let mut candidates = list_parent_watcher_state_files().ok()?;
    // Newest state first.
    candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
    for (_path, state) in candidates {
        let system = process_system_for_pids(&[state.pid]);
        if !is_live_parent_watcher_process_with_system(&system, &state) {
            continue;
        }
        let root = parent_root_as_local_path(&state.parent_root);
        if root.is_dir() {
            return Some(root);
        }
        return Some(root);
    }
    None
}

/// Most recent parent cover path (even if the watcher process is not live).
fn prefer_any_parent_root() -> Option<PathBuf> {
    let mut candidates = list_parent_watcher_state_files().ok()?;
    candidates.sort_by(|left, right| right.1.started_at.cmp(&left.1.started_at));
    for (_path, state) in candidates {
        let root = parent_root_as_local_path(&state.parent_root);
        if root.is_dir() {
            return Some(root);
        }
    }
    None
}

fn count_git_roots_under(parent: &Path) -> Result<usize, String> {
    let mut roots = Vec::new();
    let mut seen = BTreeSet::new();
    collect_git_roots_in_dir(parent, parent, &mut seen, &mut roots)?;
    Ok(roots.len())
}

fn list_parent_watcher_state_files() -> Result<Vec<(PathBuf, ParentWorktreeWatcherState)>, String> {
    let parents_root = global_mutations_root()?.join("parents");
    if !parents_root.is_dir() {
        return Ok(Vec::new());
    }
    let mut out = Vec::new();
    for entry in fs::read_dir(&parents_root)
        .map_err(|error| format!("Failed to read {}: {error}", parents_root.display()))?
    {
        let entry = entry.map_err(|error| format!("Failed to read parent spool entry: {error}"))?;
        if !entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
            continue;
        }
        for file in fs::read_dir(entry.path()).map_err(|error| {
            format!(
                "Failed to read parent spool {}: {error}",
                entry.path().display()
            )
        })? {
            let file = file.map_err(|error| format!("Failed to read parent state entry: {error}"))?;
            let path = file.path();
            let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
                continue;
            };
            if !(name.starts_with("watcher-state") && name.ends_with(".json")) {
                continue;
            }
            if let Ok(Some(state)) = read_parent_watcher_state(&path) {
                // Prefer runtime-scoped files for this runtime.
                if let Some(runtime) = state.runtime_key.as_deref() {
                    if runtime != worktree_runtime_key() {
                        // Still allow legacy files without runtime_key via runtime_keys_compatible.
                        if !runtime_keys_compatible(Some(runtime)) {
                            continue;
                        }
                    }
                } else if !runtime_keys_compatible(None) {
                    continue;
                }
                out.push((path, state));
            }
        }
    }
    Ok(out)
}

fn parent_root_as_local_path(parent_root: &str) -> PathBuf {
    let trimmed = parent_root.trim();
    if trimmed.is_empty() {
        return PathBuf::new();
    }
    if let Some(windows) = map_wsl_mnt_path_to_windows(trimmed) {
        return PathBuf::from(windows);
    }
    let as_path = PathBuf::from(trimmed);
    if as_path.is_dir() {
        return as_path;
    }
    // path_identity_key form is lowercase /mnt/c/... even on Windows storage.
    if let Some(windows) = map_wsl_mnt_path_to_windows(&trimmed.replace('\\', "/")) {
        return PathBuf::from(windows);
    }
    as_path
}

fn discover_repo_identities_under(parent: &Path) -> Result<Vec<RepoIdentity>, String> {
    let parent = canonical_parent_path(parent)?;

    let mut roots = Vec::new();
    let mut seen = BTreeSet::new();
    collect_git_roots_in_dir(&parent, &parent, &mut seen, &mut roots)?;

    // Resolve identities in parallel — sequential git spawns for 60+ repos
    // made status/API unusable on Windows.
    let identities = resolve_repo_identities_parallel(&roots);
    let mut identities = identities.into_iter().flatten().collect::<Vec<_>>();
    identities.sort_by(|left, right| left.root.cmp(&right.root));
    Ok(identities)
}

fn collect_git_roots_in_dir(
    parent: &Path,
    dir: &Path,
    seen: &mut BTreeSet<String>,
    roots: &mut Vec<PathBuf>,
) -> Result<(), String> {
    if dir != parent && is_skipped_discovery_dir(dir) {
        return Ok(());
    }

    if dir.join(".git").exists() {
        let key = path_identity_key(dir);
        if seen.insert(key) {
            roots.push(dir.to_path_buf());
        }
        return Ok(());
    }

    let entries = fs::read_dir(dir)
        .map_err(|error| format!("Failed to read folder {}: {error}", dir.display()))?;
    for entry in entries {
        let entry = entry.map_err(|error| {
            format!(
                "Failed to read folder entry under {}: {error}",
                dir.display()
            )
        })?;
        let file_type = entry
            .file_type()
            .map_err(|error| format!("Failed to inspect {}: {error}", entry.path().display()))?;
        if file_type.is_dir() {
            collect_git_roots_in_dir(parent, &entry.path(), seen, roots)?;
        }
    }

    Ok(())
}

fn resolve_repo_identities_parallel(roots: &[PathBuf]) -> Vec<Option<RepoIdentity>> {
    if roots.is_empty() {
        return Vec::new();
    }
    if roots.len() == 1 {
        return vec![resolve_repo_identity_light(Some(&roots[0])).ok()];
    }

    let thread_count = roots.len().min(8).max(1);
    let chunk_size = (roots.len() + thread_count - 1) / thread_count;
    let mut handles = Vec::new();
    for chunk in roots.chunks(chunk_size) {
        let chunk = chunk.to_vec();
        handles.push(std::thread::spawn(move || {
            chunk
                .iter()
                .map(|root| resolve_repo_identity_light(Some(root)).ok())
                .collect::<Vec<_>>()
        }));
    }

    let mut out = Vec::with_capacity(roots.len());
    for handle in handles {
        match handle.join() {
            Ok(part) => out.extend(part),
            Err(_) => {
                // Fallback sequential if a worker panics.
            }
        }
    }
    // If a worker panicked, fill remaining via sequential so we never under-report silently.
    if out.len() < roots.len() {
        for root in roots.iter().skip(out.len()) {
            out.push(resolve_repo_identity_light(Some(root)).ok());
        }
    }
    out
}

/// Status/tray identity resolve: skip HEAD SHA (one less git spawn per repo).
fn resolve_repo_identity_light(repo: Option<&Path>) -> Result<RepoIdentity, String> {
    let start = resolve_repo_path_input(repo)?;
    let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"]).map_err(|error| {
        if repo.is_some() {
            format!(
                "Failed to run git rev-parse --show-toplevel in {}: {error}",
                start.display()
            )
        } else {
            format!("Failed to run git rev-parse --show-toplevel: {error}")
        }
    })?;
    let root = normalize_windows_verbatim_path(
        fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
    );
    let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
        .unwrap_or_else(|_| "unknown".to_string());
    let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
    let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
        let name = root
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or("repository")
            .to_string();
        ("unknown".to_string(), name)
    });

    Ok(RepoIdentity {
        owner,
        name,
        branch: sanitize_path_component(branch.trim()),
        root,
        head_sha: None,
    })
}

fn resolve_repo_identity(repo: Option<&Path>) -> Result<RepoIdentity, String> {
    let start = resolve_repo_path_input(repo)?;
    let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"]).map_err(|error| {
        if repo.is_some() {
            format!(
                "Failed to run git rev-parse --show-toplevel in {}: {error}",
                start.display()
            )
        } else {
            format!("Failed to run git rev-parse --show-toplevel: {error}")
        }
    })?;
    let root = normalize_windows_verbatim_path(
        fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
    );
    let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
        .unwrap_or_else(|_| "unknown".to_string());
    let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
    let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
        let name = root
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or("repository")
            .to_string();
        ("unknown".to_string(), name)
    });
    let head_sha = current_head(&root)?;

    Ok(RepoIdentity {
        owner,
        name,
        branch: sanitize_path_component(branch.trim()),
        root,
        head_sha,
    })
}

/// Resolve `--repo` / `--repos` values to a filesystem path.
///
/// Accepts:
/// - absolute or relative git roots that exist on disk
/// - short repo names / `owner/repo` / full paths from
///   `~/.xbp/config.yaml` → `system_inventory.github_repos`
fn resolve_repo_path_input(repo: Option<&Path>) -> Result<PathBuf, String> {
    let Some(input) = repo else {
        return std::env::current_dir()
            .map_err(|error| format!("Failed to resolve current directory: {error}"));
    };

    if path_looks_like_existing_dir(input) {
        return Ok(input.to_path_buf());
    }

    let token = input.to_string_lossy();
    let token = token.trim();
    if token.is_empty() {
        return Err("--repo value is empty".to_string());
    }

    // Bare names (athena) and owner/repo (xylex-group/athena) never contain path
    // separators on disk until inventory expands them; try inventory first.
    if let Some(path) = lookup_repo_path_from_system_inventory(token)? {
        return Ok(path);
    }

    // Keep original so callers can surface a path-oriented git error for real paths.
    if input.components().count() > 1
        || token.contains('/')
        || token.contains('\\')
        || token.contains(':')
    {
        return Ok(input.to_path_buf());
    }

    Err(format!(
        "Could not resolve `--repo {token}` as a filesystem path or inventory name. \
Refresh with `xbp diag --refresh-system-inventory`, pass a full path, or use `--parent`."
    ))
}

fn path_looks_like_existing_dir(path: &Path) -> bool {
    path.is_dir()
        || fs::metadata(path)
            .map(|meta| meta.is_dir())
            .unwrap_or(false)
}

fn lookup_repo_path_from_system_inventory(token: &str) -> Result<Option<PathBuf>, String> {
    let config = match SshConfig::load() {
        Ok(config) => config,
        Err(_) => return Ok(None),
    };
    let Some(inventory) = config.system_inventory.as_ref() else {
        return Ok(None);
    };

    if let Some(path) = match_github_repo_inventory(&inventory.github_repos, token)? {
        return Ok(Some(path));
    }

    // Fall back to discovered XBP project names (folder / project name).
    let needle = normalize_repo_lookup_token(token);
    let mut project_hits: Vec<PathBuf> = Vec::new();
    for project in &inventory.xbp_projects {
        let name = normalize_repo_lookup_token(&project.name);
        let root_name = Path::new(&project.root)
            .file_name()
            .and_then(|value| value.to_str())
            .map(normalize_repo_lookup_token)
            .unwrap_or_default();
        if name == needle || root_name == needle {
            let path = PathBuf::from(&project.root);
            if path_looks_like_existing_dir(&path)
                && !project_hits.iter().any(|existing| {
                    path_identity_key(existing.as_path()) == path_identity_key(path.as_path())
                })
            {
                project_hits.push(path);
            }
        }
    }
    match project_hits.len() {
        0 => Ok(None),
        1 => Ok(Some(project_hits.remove(0))),
        _ => Err(format_ambiguous_repo_matches(token, &project_hits)),
    }
}

fn match_github_repo_inventory(
    repos: &[GithubRepoRecord],
    token: &str,
) -> Result<Option<PathBuf>, String> {
    let needle = normalize_repo_lookup_token(token);
    if needle.is_empty() {
        return Ok(None);
    }

    let mut exact_path: Vec<PathBuf> = Vec::new();
    let mut full_name_hits: Vec<PathBuf> = Vec::new();
    let mut short_name_hits: Vec<PathBuf> = Vec::new();

    for record in repos {
        let path = PathBuf::from(record.path.trim());
        if record.path.trim().is_empty() || !path_looks_like_existing_dir(&path) {
            continue;
        }

        let path_key = path_identity_key(&path);
        let full_name = normalize_repo_lookup_token(&record.full_name);
        let short = normalize_repo_lookup_token(&record.repo);
        let owner_repo = normalize_repo_lookup_token(&format!("{}/{}", record.owner, record.repo));
        let folder = path
            .file_name()
            .and_then(|value| value.to_str())
            .map(normalize_repo_lookup_token)
            .unwrap_or_default();

        let push_unique = |bucket: &mut Vec<PathBuf>| {
            if !bucket
                .iter()
                .any(|existing| path_identity_key(existing.as_path()) == path_key)
            {
                bucket.push(path.clone());
            }
        };

        if path_identity_key(Path::new(record.path.trim())) == path_identity_key(Path::new(token))
            || normalize_repo_lookup_token(&record.path) == needle
        {
            push_unique(&mut exact_path);
        }
        if full_name == needle || owner_repo == needle {
            push_unique(&mut full_name_hits);
        }
        if short == needle || folder == needle {
            push_unique(&mut short_name_hits);
        }
    }

    if let Some(path) = exact_path.into_iter().next() {
        return Ok(Some(path));
    }
    match full_name_hits.len() {
        0 => {}
        1 => return Ok(Some(full_name_hits.remove(0))),
        _ => return Err(format_ambiguous_repo_matches(token, &full_name_hits)),
    }
    match short_name_hits.len() {
        0 => Ok(None),
        1 => Ok(Some(short_name_hits.remove(0))),
        _ => Err(format_ambiguous_repo_matches(token, &short_name_hits)),
    }
}

fn normalize_repo_lookup_token(value: &str) -> String {
    value
        .trim()
        .trim_end_matches('/')
        .trim_end_matches('\\')
        .trim_end_matches(".git")
        .replace('\\', "/")
        .to_ascii_lowercase()
}

fn format_ambiguous_repo_matches(token: &str, paths: &[PathBuf]) -> String {
    let list = paths
        .iter()
        .map(|path| format!("  - {}", path.display()))
        .collect::<Vec<_>>()
        .join("\n");
    format!(
        "Ambiguous `--repo {token}` matches multiple inventory paths:\n{list}\n\
Use a full path or `owner/repo` (for example `xylex-group/{token}`)."
    )
}

fn prepare_spool_layout(identity: &RepoIdentity) -> Result<SpoolLayout, String> {
    let root = repository_spool_root(identity)?.join(sanitize_path_component(&identity.branch));
    fs::create_dir_all(&root).map_err(|error| {
        format!(
            "Failed to create worktree mutation spool {}: {error}",
            root.display()
        )
    })?;
    let event_dedupe_dir = root.join("event-dedupe");
    fs::create_dir_all(&event_dedupe_dir).map_err(|error| {
        format!(
            "Failed to create worktree mutation dedupe dir {}: {error}",
            event_dedupe_dir.display()
        )
    })?;
    let commit_dedupe_dir = root.join("commit-dedupe");
    fs::create_dir_all(&commit_dedupe_dir).map_err(|error| {
        format!(
            "Failed to create worktree commit dedupe dir {}: {error}",
            commit_dedupe_dir.display()
        )
    })?;
    Ok(spool_layout_for_root(
        root,
        Some(Uuid::new_v4().to_string()),
    ))
}

/// Canonical mutations root under the established global XBP home
/// (`XBP_HOME` / Windows profile `.xbp` / `~/.xbp`), not the repo.
fn global_mutations_root() -> Result<PathBuf, String> {
    let paths = ensure_global_xbp_paths()?;
    Ok(paths.root_dir.join("mutations"))
}

fn repository_spool_root(identity: &RepoIdentity) -> Result<PathBuf, String> {
    Ok(global_mutations_root()?
        .join(sanitize_path_component(&identity.owner))
        .join(sanitize_path_component(&identity.name)))
}

/// All mutation roots that may already hold data for this identity.
///
/// Primary root is the established global XBP home; legacy/alternate homes and
/// older `runtimes/*` layouts are included so WSL/Windows can still read each
/// other's historical spoils while new writes go to the primary path.
fn repository_spool_search_roots(identity: &RepoIdentity) -> Result<Vec<PathBuf>, String> {
    let owner = sanitize_path_component(&identity.owner);
    let name = sanitize_path_component(&identity.name);
    let mut roots: Vec<PathBuf> = Vec::new();
    let mut push = |path: PathBuf| {
        if roots.iter().any(|existing| {
            path_identity_key(existing.as_path()) == path_identity_key(path.as_path())
        }) {
            return;
        }
        roots.push(path);
    };

    push(repository_spool_root(identity)?);

    for base in legacy_mutations_base_candidates() {
        push(base.join(&owner).join(&name));
        // Previous runtime-isolated layout.
        let runtimes = base.join("runtimes");
        if runtimes.is_dir() {
            if let Ok(entries) = fs::read_dir(&runtimes) {
                for entry in entries.flatten() {
                    if entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false) {
                        push(entry.path().join(&owner).join(&name));
                    }
                }
            }
        }
    }

    Ok(roots)
}

fn legacy_mutations_base_candidates() -> Vec<PathBuf> {
    let mut bases: Vec<PathBuf> = Vec::new();
    let mut push = |path: PathBuf| {
        if bases
            .iter()
            .any(|existing| path_identity_key(existing.as_path()) == path_identity_key(path.as_path()))
        {
            return;
        }
        bases.push(path);
    };

    if let Ok(primary) = global_mutations_root() {
        push(primary);
    }
    // Also surface the raw resolved root without create-side effects for alternate spellings.
    push(resolve_global_xbp_root_dir().join("mutations"));

    if let Some(home) = dirs::home_dir() {
        push(home.join(".xbp").join("mutations"));
        push(home.join(".config").join("xbp").join("mutations"));
    }

    // Dual-map the primary mutations root across WSL/Windows path spellings.
    if let Ok(primary) = global_mutations_root() {
        let rendered = primary.to_string_lossy().replace('\\', "/");
        if let Some(mnt) = map_windows_path_to_wsl_mnt(&rendered) {
            push(PathBuf::from(mnt));
        }
        if let Some(windows) = map_wsl_mnt_path_to_windows(&rendered) {
            if let Some(mnt) = map_windows_path_to_wsl_mnt(&windows) {
                push(PathBuf::from(mnt));
            }
            push(PathBuf::from(windows));
        }
    }

    bases
}

fn spool_layout_for_existing_root(root: &Path) -> SpoolLayout {
    spool_layout_for_root(root.to_path_buf(), None)
}

fn spool_layout_for_root(root: PathBuf, run_id: Option<String>) -> SpoolLayout {
    let run_id = run_id.unwrap_or_else(|| Uuid::new_v4().to_string());
    let event_dedupe_dir = root.join("event-dedupe");
    let commit_dedupe_dir = root.join("commit-dedupe");
    // Events/stats/dedupe are shared across Windows+WSL on the established home.
    // Watcher PID state is runtime-scoped so both can run without clobbering stop/start.
    let watcher_state_file = root.join(format!(
        "watcher-state-{}.json",
        sanitize_path_component(&worktree_runtime_key())
    ));
    SpoolLayout {
        events_file: root.join(format!("events-{run_id}.jsonl")),
        commits_file: root.join(format!("commits-{run_id}.jsonl")),
        stats_file: root.join("stats.json"),
        watcher_state_file,
        sync_log_file: root.join("sync-log.jsonl"),
        event_dedupe_file: root.join("event-dedupe.jsonl"),
        event_dedupe_dir,
        commit_dedupe_dir,
        root,
    }
}

fn prepare_parent_spool_layout(parent: &Path) -> Result<ParentSpoolLayout, String> {
    let parent = canonical_parent_path(parent)?;
    // Prefer a path-identity that is stable across WSL `/mnt/c/...` and Windows `C:\...`.
    let parent_key = cross_platform_path_identity_key(&parent);
    let mut hasher = Sha256::new();
    hasher.update(parent_key.as_bytes());
    let digest = format!("{:x}", hasher.finalize());
    let label = parent
        .file_name()
        .and_then(|value| value.to_str())
        .map(sanitize_path_component)
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "parent".to_string());
    let root = global_mutations_root()?
        .join("parents")
        .join(format!("{}-{}", label, &digest[..16]));
    fs::create_dir_all(&root).map_err(|error| {
        format!(
            "Failed to create parent worktree watcher state dir {}: {error}",
            root.display()
        )
    })?;
    Ok(ParentSpoolLayout {
        watcher_state_file: root.join(format!(
            "watcher-state-{}.json",
            sanitize_path_component(&worktree_runtime_key())
        )),
    })
}

fn canonical_parent_path(parent: &Path) -> Result<PathBuf, String> {
    let normalized = normalize_windows_verbatim_path(
        fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()),
    );
    if !normalized.is_dir() {
        return Err(format!(
            "Parent folder {} does not exist or is not a directory.",
            normalized.display()
        ));
    }
    Ok(normalized)
}

/// Process-wide buffer so bursty FS events open each jsonl file once per batch
/// instead of once per event (major IO win under editors that write many times).
struct JsonlWriteBuffers {
    lines_by_path: HashMap<PathBuf, Vec<String>>,
    line_count: usize,
    last_flush: Instant,
}

impl JsonlWriteBuffers {
    fn new() -> Self {
        Self {
            lines_by_path: HashMap::new(),
            line_count: 0,
            last_flush: Instant::now(),
        }
    }

    fn push(&mut self, path: PathBuf, line: String) {
        self.lines_by_path.entry(path).or_default().push(line);
        self.line_count = self.line_count.saturating_add(1);
    }

    fn should_flush(&self) -> bool {
        self.line_count >= JSONL_FLUSH_MAX_LINES
            || self.last_flush.elapsed() >= Duration::from_millis(JSONL_FLUSH_MAX_WAIT_MS)
    }

    fn flush(&mut self) -> Result<(), String> {
        if self.lines_by_path.is_empty() {
            return Ok(());
        }
        let batch = std::mem::take(&mut self.lines_by_path);
        self.line_count = 0;
        self.last_flush = Instant::now();
        for (path, lines) in batch {
            if let Some(parent) = path.parent() {
                let _ = fs::create_dir_all(parent);
            }
            let mut file = OpenOptions::new()
                .create(true)
                .append(true)
                .open(&path)
                .map_err(|error| {
                    format!("Failed to open spool file {}: {error}", path.display())
                })?;
            for line in lines {
                writeln!(file, "{line}").map_err(|error| {
                    format!("Failed to append spool file {}: {error}", path.display())
                })?;
            }
            // One flush per file per batch (not per line).
            let _ = file.flush();
        }
        Ok(())
    }
}

fn jsonl_buffers() -> &'static Mutex<JsonlWriteBuffers> {
    static BUFFERS: OnceLock<Mutex<JsonlWriteBuffers>> = OnceLock::new();
    BUFFERS.get_or_init(|| Mutex::new(JsonlWriteBuffers::new()))
}

fn append_json_line<T: Serialize>(path: &Path, value: &T) -> Result<(), String> {
    let line = serde_json::to_string(value)
        .map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
    let mut guard = jsonl_buffers()
        .lock()
        .map_err(|_| "jsonl write buffer lock poisoned".to_string())?;
    guard.push(path.to_path_buf(), line);
    if guard.should_flush() {
        guard.flush()?;
    }
    Ok(())
}

/// Force pending spool lines to disk (call each watch-loop tick and on shutdown paths).
fn flush_jsonl_write_buffers() {
    if let Ok(mut guard) = jsonl_buffers().lock() {
        let _ = guard.flush();
    }
}

fn git_numstat_for_path(repo_root: &Path, relative_path: &str) -> Result<(u64, u64), String> {
    let output = Command::new("git")
        .args(["diff", "--numstat", "--"])
        .arg(relative_path)
        .current_dir(repo_root)
        .output()
        .map_err(|error| format!("Failed to run git diff --numstat: {error}"))?;
    if !output.status.success() {
        return Ok((0, 0));
    }
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut added = 0;
    let mut removed = 0;
    for line in stdout.lines() {
        let mut parts = line.split_whitespace();
        added += parse_numstat_count(parts.next());
        removed += parse_numstat_count(parts.next());
    }
    Ok((added, removed))
}

fn file_line_count_for_path(repo_root: &Path, relative_path: &str) -> Result<u64, String> {
    let path = repo_root.join(relative_path);
    let meta = fs::metadata(&path)
        .map_err(|error| format!("Failed to stat file {}: {error}", path.display()))?;
    // Skip huge files — full reads were a major CPU/IO cost on hot trees.
    if meta.len() > 512 * 1024 {
        return Ok(0);
    }
    let content = fs::read_to_string(&path)
        .map_err(|error| format!("Failed to read file {}: {error}", path.display()))?;
    Ok(content.lines().count() as u64)
}

fn parse_numstat_count(value: Option<&str>) -> u64 {
    value.and_then(|raw| raw.parse::<u64>().ok()).unwrap_or(0)
}

fn current_head(repo_root: &Path) -> Result<Option<String>, String> {
    match git_output(repo_root, &["rev-parse", "HEAD"]) {
        Ok(value) => Ok(Some(value)),
        Err(error)
            if error.contains("unknown revision") || error.contains("ambiguous argument") =>
        {
            Ok(None)
        }
        Err(error) => Err(error),
    }
}

fn git_output(repo_root: &Path, args: &[&str]) -> Result<String, String> {
    let output = Command::new("git")
        .args(args)
        .current_dir(repo_root)
        .output()
        .map_err(|error| format!("Failed to run git {}: {error}", args.join(" ")))?;
    if !output.status.success() {
        return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
    }
    Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
}

fn parse_remote_owner_repo(remote: &str) -> Option<(String, String)> {
    let trimmed = remote.trim().trim_end_matches(".git");
    if trimmed.is_empty() {
        return None;
    }

    let path_part = if !trimmed.contains("://") {
        if let Some((_, path)) = trimmed.rsplit_once(':') {
            path
        } else {
            trimmed
        }
    } else {
        trimmed
            .trim_start_matches("https://")
            .trim_start_matches("http://")
            .trim_start_matches("ssh://")
            .split_once('/')
            .map(|(_, path)| path)
            .unwrap_or(trimmed)
    };
    let mut parts = path_part.rsplitn(2, '/');
    let name = parts.next()?.trim();
    let owner = parts.next()?.trim();
    if owner.is_empty() || name.is_empty() {
        return None;
    }
    Some((
        sanitize_path_component(owner),
        sanitize_path_component(name.trim_end_matches(".git")),
    ))
}

fn sanitize_path_component(value: &str) -> String {
    let sanitized: String = value
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
                ch
            } else {
                '-'
            }
        })
        .collect();
    sanitized
        .trim_matches('-')
        .chars()
        .take(160)
        .collect::<String>()
}

fn normalize_windows_verbatim_path(path: PathBuf) -> PathBuf {
    PathBuf::from(strip_windows_verbatim_prefix(&path.to_string_lossy()))
}

fn strip_windows_verbatim_prefix(input: &str) -> &str {
    input.strip_prefix(r"\\?\").unwrap_or(input)
}

fn path_identity_key(path: &Path) -> String {
    cross_platform_path_identity_key(path)
}

/// Path identity that treats `C:\foo` and `/mnt/c/foo` as the same location.
fn cross_platform_path_identity_key(path: &Path) -> String {
    let mut value = path.to_string_lossy().replace('\\', "/");
    if let Some(mnt) = map_windows_path_to_wsl_mnt(&value) {
        value = mnt;
    } else if let Some(windows) = map_wsl_mnt_path_to_windows(&value) {
        if let Some(mnt) = map_windows_path_to_wsl_mnt(&windows) {
            value = mnt;
        } else {
            value = windows.replace('\\', "/");
        }
    }
    // Drive-backed paths are case-insensitive on Windows; always fold for stable keys.
    value.to_ascii_lowercase()
}

/// Runtime isolation key for spoils + device identity.
///
/// Examples: `windows`, `linux`, `macos`, `wsl-ubuntu`, `wsl-debian`.
fn worktree_runtime_key() -> String {
    static KEY: OnceLock<String> = OnceLock::new();
    KEY.get_or_init(|| {
        if is_wsl_runtime() {
            let distro = std::env::var("WSL_DISTRO_NAME")
                .ok()
                .map(|value| value.trim().to_string())
                .filter(|value| !value.is_empty())
                .unwrap_or_else(|| "wsl".to_string());
            format!("wsl-{}", sanitize_path_component(&distro).to_ascii_lowercase())
        } else {
            std::env::consts::OS.to_string()
        }
    })
    .clone()
}

fn is_wsl_runtime() -> bool {
    if !cfg!(target_os = "linux") {
        return false;
    }
    if std::env::var_os("WSL_DISTRO_NAME").is_some() || std::env::var_os("WSL_INTEROP").is_some() {
        return true;
    }
    // Fall back to /proc/version marker used by Microsoft WSL kernels.
    fs::read_to_string("/proc/version")
        .map(|content| {
            let lower = content.to_ascii_lowercase();
            lower.contains("microsoft") || lower.contains("wsl")
        })
        .unwrap_or(false)
}

/// Device id scoped by runtime so Windows and WSL can share one xbp config home
/// without looking like the same upload source.
fn worktree_device_hardware_id(base_hardware_id: &str) -> String {
    format!("{base_hardware_id}@{}", worktree_runtime_key())
}

fn ensure_mutation_event_identity(mut event: WorktreeMutationEvent) -> WorktreeMutationEvent {
    if event.fingerprint.as_ref().is_none_or(|value| value.is_empty()) {
        let fingerprint = mutation_event_fingerprint(&event);
        event.fingerprint = Some(fingerprint.clone());
        if event.id.is_empty() || !event.id.starts_with("wtmut_") {
            event.id = stable_event_id(&fingerprint);
        }
    } else if event.id.is_empty() {
        event.id = stable_event_id(event.fingerprint.as_deref().unwrap_or_default());
    }
    event
}

fn ensure_commit_event_identity(mut commit: WorktreeCommitEvent) -> WorktreeCommitEvent {
    if commit.fingerprint.as_ref().is_none_or(|value| value.is_empty()) {
        let fingerprint = commit_event_fingerprint(&commit);
        commit.fingerprint = Some(fingerprint.clone());
        if commit.id.is_empty() || !commit.id.starts_with("wtmut_") {
            commit.id = stable_event_id(&fingerprint);
        }
    } else if commit.id.is_empty() {
        commit.id = stable_event_id(commit.fingerprint.as_deref().unwrap_or_default());
    }
    commit
}

fn path_string_has_ignored_component(path: &str) -> bool {
    // No project root available; apply global + built-in rules only.
    watch_ignore_rules().ignores_relative(path)
}

fn is_skipped_discovery_dir(path: &Path) -> bool {
    watch_ignore_rules().skips_discovery_dir(path)
}

/// Relative path from `root` to `path` with `/` separators.
///
/// Uses `Path::strip_prefix` when possible, then falls back to slash-normalized
/// string prefix matching so Windows-style absolute paths still work when the
/// binary runs on Unix (WSL tests / mixed path sources from notify).
fn relative_watch_path(root: &Path, path: &Path) -> Option<String> {
    if let Ok(relative) = path.strip_prefix(root) {
        return Some(normalize_relative_watch_path(&relative.to_string_lossy()));
    }

    let root_norm = normalize_absolute_watch_path(root);
    let path_norm = normalize_absolute_watch_path(path);
    if path_norm == root_norm {
        return Some(String::new());
    }
    let prefix = format!("{root_norm}/");
    path_norm
        .strip_prefix(&prefix)
        .map(|relative| normalize_relative_watch_path(relative))
}

fn normalize_absolute_watch_path(path: &Path) -> String {
    let mut value = path.to_string_lossy().replace('\\', "/");
    // Drop Windows verbatim prefix if present after slash normalization.
    if let Some(stripped) = value.strip_prefix("//?/") {
        value = stripped.to_string();
    }
    // Case-fold drive letter for stable prefix matching (C: vs c:).
    if value.len() >= 2 {
        let bytes = value.as_bytes();
        if bytes[1] == b':' && bytes[0].is_ascii_alphabetic() {
            let mut chars = value.chars();
            let drive = chars.next().unwrap().to_ascii_lowercase();
            value = format!("{drive}{}", chars.as_str());
        }
    }
    value.trim_end_matches('/').to_string()
}

fn relative_slash_path(root: &Path, path: &Path) -> Option<String> {
    relative_watch_path(root, path).or_else(|| {
        Some(normalize_relative_watch_path(
            &path.to_string_lossy().replace('\\', "/"),
        ))
    })
}

fn is_ignored_path(root: &Path, path: &Path) -> bool {
    watch_ignore_rules_for_root(root).ignores_path(root, path)
}

fn current_hostname() -> Option<String> {
    for key in ["COMPUTERNAME", "HOSTNAME", "NAME"] {
        if let Ok(value) = std::env::var(key) {
            let trimmed = value.trim();
            if !trimmed.is_empty() {
                return Some(trimmed.to_string());
            }
        }
    }

    if let Ok(content) = fs::read_to_string("/etc/hostname") {
        let trimmed = content.trim();
        if !trimmed.is_empty() {
            return Some(trimmed.to_string());
        }
    }

    // WSL often has a Linux hostname distinct from the Windows host; prefer it when set.
    if let Ok(output) = Command::new("hostname").output() {
        if output.status.success() {
            let value = String::from_utf8_lossy(&output.stdout).trim().to_string();
            if !value.is_empty() {
                return Some(value);
            }
        }
    }

    None
}

fn is_background_child() -> bool {
    std::env::var(BACKGROUND_CHILD_ENV)
        .ok()
        .map(|value| value == "1")
        .unwrap_or(false)
}

#[allow(dead_code)]
fn content_sha256(path: &Path) -> Option<String> {
    let bytes = fs::read(path).ok()?;
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    Some(format!("{:x}", hasher.finalize()))
}

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

    #[test]
    fn remove_all_watcher_state_files_clears_legacy_and_runtime() {
        let dir = std::env::temp_dir().join(format!(
            "xbp-watcher-state-test-{}",
            std::process::id()
        ));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).expect("temp dir");
        let primary = dir.join("watcher-state-windows.json");
        let legacy = dir.join("watcher-state.json");
        let wsl = dir.join("watcher-state-wsl-ubuntu.json");
        for path in [&primary, &legacy, &wsl] {
            let mut f = fs::File::create(path).expect("create");
            writeln!(f, "{{}}").expect("write");
        }
        let removed = remove_all_watcher_state_files(&primary);
        assert!(removed >= 3, "removed={removed}");
        assert!(!primary.exists());
        assert!(!legacy.exists());
        assert!(!wsl.exists());
        assert_eq!(remove_all_watcher_state_files(&primary), 0);
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn parses_https_and_ssh_remote_urls() {
        assert_eq!(
            parse_remote_owner_repo("https://github.com/xylex-group/xbp.git"),
            Some(("xylex-group".to_string(), "xbp".to_string()))
        );
        assert_eq!(
            parse_remote_owner_repo("git@github.com:xylex-group/xbp.git"),
            Some(("xylex-group".to_string(), "xbp".to_string()))
        );
    }

    #[test]
    fn sanitizes_branch_for_storage_path() {
        assert_eq!(
            sanitize_path_component("feature/worktree watcher"),
            "feature-worktree-watcher"
        );
    }

    #[test]
    fn normalizes_windows_verbatim_repo_roots() {
        assert_eq!(
            normalize_windows_verbatim_path(PathBuf::from(
                r"\\?\C:\Users\floris\Documents\GitHub\mollie-api-rust"
            )),
            PathBuf::from(r"C:\Users\floris\Documents\GitHub\mollie-api-rust")
        );
    }

    #[test]
    fn parent_path_matching_uses_repo_boundaries() {
        let repo = Path::new(r"C:\Users\floris\Documents\GitHub\xbp");
        assert!(path_belongs_to_repo(
            Path::new(r"C:\Users\floris\Documents\GitHub\xbp\src\main.rs"),
            repo
        ));
        assert!(!path_belongs_to_repo(
            Path::new(r"C:\Users\floris\Documents\GitHub\xbp-other\src\main.rs"),
            repo
        ));
    }

    #[test]
    fn adaptive_tiers_follow_idle_thresholds() {
        let cfg = WorktreeWatchConfig {
            idle_after_seconds: Some(300),
            cold_after_seconds: Some(1800),
            ..Default::default()
        };
        assert_eq!(tier_from_idle(0, &cfg), RepoWatchTier::Hot);
        assert_eq!(tier_from_idle(299, &cfg), RepoWatchTier::Hot);
        assert_eq!(tier_from_idle(300, &cfg), RepoWatchTier::Warm);
        assert_eq!(tier_from_idle(1799, &cfg), RepoWatchTier::Warm);
        assert_eq!(tier_from_idle(1800, &cfg), RepoWatchTier::Cold);
        assert!(cfg.hot_commit_check_seconds() < cfg.warm_commit_check_seconds());
        assert!(cfg.warm_commit_check_seconds() < cfg.cold_commit_check_seconds());
    }

    #[test]
    fn default_resource_knobs_are_tighter() {
        let cfg = WorktreeWatchConfig::default();
        assert_eq!(cfg.idle_after_seconds(), 180);
        assert_eq!(cfg.cold_after_seconds(), 600);
        assert_eq!(cfg.max_hot_repos(), 4);
        assert_eq!(cfg.max_fs_watches(), 6);
        assert_eq!(cfg.max_commit_checks_per_tick(), 6);
        assert_eq!(cfg.warm_commit_check_seconds(), 120);
        assert_eq!(cfg.cold_commit_check_seconds(), 300);
        assert_eq!(cfg.spool_retention_days(), 7);
        assert_eq!(cfg.parent_rescan_seconds(), 300);
    }

    #[test]
    fn fs_watch_budget_prefers_hot_and_caps() {
        let base = std::env::temp_dir().join(format!("xbp-fs-budget-{}", Uuid::new_v4()));
        let dir_a = base.join("a");
        let dir_b = base.join("b");
        let dir_c = base.join("c");
        let dir_d = base.join("d");
        for dir in [&dir_a, &dir_b, &dir_c, &dir_d] {
            fs::create_dir_all(dir).unwrap();
        }
        let now = Instant::now();
        let mk = |root: &Path, tier: RepoWatchTier, score: f64| -> WatchedRepo {
            WatchedRepo {
                identity: RepoIdentity {
                    owner: "o".into(),
                    name: root
                        .file_name()
                        .and_then(|n| n.to_str())
                        .unwrap_or("r")
                        .to_string(),
                    branch: "main".into(),
                    root: root.to_path_buf(),
                    head_sha: None,
                },
                layout: spool_layout_for_root(root.join(".xbp-spool-test"), None),
                last_head: None,
                event_dedupe: RecentEventDedupe::default(),
                last_activity: now,
                activity_score: score,
                tier,
                fs_watched: false,
                next_commit_check: now,
            }
        };
        let repos = vec![
            mk(&dir_a, RepoWatchTier::Hot, 1.0),
            mk(&dir_b, RepoWatchTier::Warm, 50.0),
            mk(&dir_c, RepoWatchTier::Warm, 40.0),
            mk(&dir_d, RepoWatchTier::Cold, 99.0),
        ];
        let want = fs_watch_want_mask(&repos, 2);
        let _ = fs::remove_dir_all(&base);
        assert!(want[0], "hot always preferred");
        assert!(want[1], "highest warm fills remaining budget");
        assert!(!want[2], "third non-cold loses budget");
        assert!(!want[3], "cold never watches");
    }

    #[test]
    fn prune_spool_removes_old_seen_and_synced() {
        let temp = std::env::temp_dir().join(format!("xbp-prune-{}", Uuid::new_v4()));
        let _ = fs::remove_dir_all(&temp);
        fs::create_dir_all(&temp).unwrap();
        let layout = spool_layout_for_root(temp.clone(), None);
        fs::create_dir_all(&layout.event_dedupe_dir).unwrap();
        fs::create_dir_all(&layout.commit_dedupe_dir).unwrap();
        let old_seen = layout.event_dedupe_dir.join("abc.seen");
        let fresh_seen = layout.event_dedupe_dir.join("def.seen");
        fs::write(&old_seen, "old").unwrap();
        fs::write(&fresh_seen, "new").unwrap();
        let synced = layout.root.join("events-1.synced.1.jsonl");
        fs::write(&synced, "{}\n").unwrap();
        let fat_dedupe = layout.event_dedupe_file.clone();
        fs::write(
            &fat_dedupe,
            vec![b'x'; (MAX_EVENT_DEDUPE_JSONL_BYTES as usize) + 8],
        )
        .unwrap();

        // retention=0: any file with mtime <= now is eligible for removal.
        let removed = prune_spool_layout(&layout, Duration::from_secs(0)).unwrap();
        let fat_gone =
            !fat_dedupe.exists() || fs::metadata(&fat_dedupe).map(|m| m.len()).unwrap_or(0) == 0;
        let _ = fs::remove_dir_all(&temp);
        assert!(removed >= 3, "expected seen+synced+rotate, got {removed}");
        assert!(fat_gone, "oversized dedupe log should be rotated or truncated");
    }

    #[test]
    fn load_event_dedupe_does_not_read_file() {
        let temp = std::env::temp_dir().join(format!("xbp-dedupe-load-{}", Uuid::new_v4()));
        let _ = fs::remove_dir_all(&temp);
        fs::create_dir_all(&temp).unwrap();
        let path = temp.join("event-dedupe.jsonl");
        // Would be expensive if fully parsed into a set at start.
        let line = r#"{"fingerprint":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}"#;
        let mut body = String::new();
        for _ in 0..20_000 {
            body.push_str(line);
            body.push('\n');
        }
        fs::write(&path, body).unwrap();
        let set = load_event_dedupe_fingerprints(&path).unwrap();
        let _ = fs::remove_dir_all(&temp);
        assert!(set.is_empty());
    }

    #[test]
    fn activity_score_bumps_and_caps() {
        let mut score = 0.0;
        for _ in 0..20 {
            score = bump_activity_score(score);
        }
        assert!(score > 5.0);
        let capped = bump_activity_score(10_000.0);
        assert!(capped <= 10_000.0);
    }

    #[test]
    fn truncate_middle_keeps_ends() {
        let s = truncate_middle("owner/very-long-repository-name-here", 20);
        assert!(s.chars().count() <= 20);
        assert!(s.contains('…') || s.len() <= 20);
    }

    #[test]
    fn rank_index_keys_by_path_identity() {
        let cfg = WorktreeWatchConfig {
            repo_ranks: vec![WorktreeWatchRepoRank {
                root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
                owner: Some("x".into()),
                name: Some("xbp".into()),
                score: 3.5,
                last_event_at: None,
                rank: Some(1),
                tier: Some("hot".into()),
            }],
            ..Default::default()
        };
        let index = build_repo_rank_index(&cfg);
        let key = path_identity_key(Path::new(r"C:\Users\floris\Documents\GitHub\xbp"));
        assert!(index.contains_key(&key));
        assert_eq!(index.get(&key).map(|r| r.score), Some(3.5));
    }

    #[test]
    fn skips_heavy_discovery_directories() {
        assert!(is_skipped_discovery_dir(Path::new("node_modules")));
        assert!(is_skipped_discovery_dir(Path::new("target")));
        assert!(is_skipped_discovery_dir(Path::new("dist")));
        assert!(is_skipped_discovery_dir(Path::new(".next")));
        assert!(is_skipped_discovery_dir(Path::new(".nx")));
        assert!(is_skipped_discovery_dir(Path::new(".xbp")));
        assert!(is_skipped_discovery_dir(Path::new(".cargo")));
        assert!(is_skipped_discovery_dir(Path::new(".codex")));
        assert!(is_skipped_discovery_dir(Path::new(".cursor")));
        assert!(is_skipped_discovery_dir(Path::new(".agents")));
        assert!(is_skipped_discovery_dir(Path::new(".idea")));
        assert!(is_skipped_discovery_dir(Path::new(".github")));
        assert!(is_skipped_discovery_dir(Path::new(".postman")));
        assert!(is_skipped_discovery_dir(Path::new("mcps")));
        assert!(is_skipped_discovery_dir(Path::new("agent-tools")));
        assert!(is_skipped_discovery_dir(Path::new("target-publish")));
        assert!(is_skipped_discovery_dir(Path::new("__pycache__")));
        assert!(is_skipped_discovery_dir(Path::new("terminals")));
        assert!(is_skipped_discovery_dir(Path::new(".open-next")));
        assert!(!is_skipped_discovery_dir(Path::new("mollie-api-rust")));
    }

    #[test]
    fn ignores_nested_generated_mutation_paths() {
        // Windows-style absolute paths must work even when tests run on Unix/WSL
        // (Path::strip_prefix does not treat `\` as a separator on Unix).
        let root = Path::new(r"C:\Users\floris\Documents\GitHub\athena");
        assert!(is_ignored_path(
            root,
            Path::new(
                r"C:\Users\floris\Documents\GitHub\athena\apps\docs\node_modules\@aws-sdk\client-lambda\dist-types\schemas"
            )
        ));
        assert!(is_ignored_path(
            root,
            Path::new(r"C:\Users\floris\Documents\GitHub\athena\target\debug\build")
        ));
        // Nx cache and OpenNext build output anywhere under the repo.
        assert!(is_ignored_path(
            root,
            Path::new(r"C:\Users\floris\Documents\GitHub\athena\.nx\cache\run.json")
        ));
        assert!(is_ignored_path(
            root,
            Path::new(
                r"C:\Users\floris\Documents\GitHub\athena\apps\docs\.open-next\server-functions\default\handler.mjs"
            )
        ));
        assert!(!is_ignored_path(
            root,
            Path::new(r"C:\Users\floris\Documents\GitHub\athena\apps\web\src\main.ts")
        ));

        // Forward-slash form (notify / mixed hosts) and relative component checks.
        let root_slash = Path::new("C:/Users/floris/Documents/GitHub/athena");
        assert!(is_ignored_path(
            root_slash,
            Path::new(
                "C:/Users/floris/Documents/GitHub/athena/apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas"
            )
        ));
        assert!(path_string_has_ignored_component(
            "apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas"
        ));
    }

    #[test]
    fn global_config_forbidden_paths_folders_and_banned_words() {
        let rules = WatchIgnoreRules::from_config(&WorktreeWatchConfig {
            forbidden_paths: vec!["secrets/".to_string(), "apps/web/.env.local".to_string()],
            forbidden_folders: vec![".cache".to_string(), "coverage".to_string()],
            banned_words: vec!["private-key".to_string(), "CREDENTIAL".to_string()],
            ..Default::default()
        });

        // Nested under a watched tree still blocked by prefix.
        assert!(rules.ignores_relative("secrets/prod/api.key"));
        assert!(rules.ignores_relative("apps/web/.env.local"));
        // Exact file ban is not a string-prefix of other filenames.
        assert!(!rules.ignores_relative("apps/web/.env.local.bak"));

        // Folder name anywhere
        assert!(rules.ignores_relative("apps/web/.cache/tmp"));
        assert!(rules.ignores_relative("packages/foo/coverage/index.html"));

        // Banned word (case-insensitive substring)
        assert!(rules.ignores_relative("docs/private-key-notes.md"));
        assert!(rules.ignores_relative("config/my-credential-store.json"));

        // Still allow normal source
        assert!(!rules.ignores_relative("apps/web/src/main.ts"));
        assert!(!rules.skips_discovery_dir(Path::new("mollie-api-rust")));
        assert!(rules.skips_discovery_dir(Path::new(".cache")));
        assert!(rules.skips_discovery_dir(Path::new("node_modules"))); // built-in
    }

    #[test]
    fn global_worktree_ignore_file_honors_mcps_patterns() {
        let root = std::env::temp_dir().join(format!(
            "xbp-global-watch-ignore-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("time")
                .as_nanos()
        ));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(&root).expect("global root");

        let ignore_path = root.join(crate::utils::GLOBAL_WORKTREE_IGNORE_FILENAME);
        fs::write(&ignore_path, "mcps/\n/custom-global/\n").expect("write global ignore");

        let rules = WatchIgnoreRules {
            forbidden_path_prefixes: Vec::new(),
            forbidden_folders: BTreeSet::new(),
            banned_words: Vec::new(),
            global_ignore: crate::utils::load_global_worktree_ignore_from_path(&ignore_path),
            project_ignore: crate::utils::XbpIgnoreSet::empty(),
        };

        assert!(rules.ignores_relative("mcps/github/catalog.json"));
        assert!(rules.ignores_relative("apps/api/mcps/local"));
        assert!(rules.ignores_relative("custom-global/notes.md"));
        assert!(!rules.ignores_relative("apps/web/src/main.ts"));
        assert!(rules.skips_discovery_dir(Path::new("mcps")));

        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn project_xbpignore_is_honored_by_worktree_watch_rules() {
        let root = std::env::temp_dir().join(format!(
            "xbp-watch-ignore-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("time")
                .as_nanos()
        ));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join(".xbp")).expect("xbp dir");
        fs::write(
            root.join(".xbp").join(".xbpignore"),
            "packages/icons/\n*.generated.ts\n",
        )
        .expect("write ignore");
        fs::write(
            root.join(".xbp").join("xbp.yaml"),
            "project_name: demo\nversion: 0.1.0\nport: 3000\nbuild_dir: ./\nignore_paths:\n  - e2e/\nwatch_ignore_paths:\n  - tmp-watch/\n",
        )
        .expect("write config");

        let rules = WatchIgnoreRules::from_config_and_project(
            &WorktreeWatchConfig::default(),
            Some(&root),
        );
        // `.xbpignore` still applies to worktree-watch
        assert!(rules.ignores_relative("packages/icons/src/index.ts"));
        assert!(rules.ignores_relative("apps/web/foo.generated.ts"));
        // Discovery-only `ignore_paths` must NOT silence worktree-watch
        assert!(!rules.ignores_relative("e2e/login.spec.ts"));
        // Explicit `watch_ignore_paths` do apply
        assert!(rules.ignores_relative("tmp-watch/file.ts"));
        assert!(!rules.ignores_relative("apps/web/src/main.ts"));

        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn builds_coding_stats_by_file_and_filetype() {
        let identity = RepoIdentity {
            owner: "xylex-group".to_string(),
            name: "xbp".to_string(),
            branch: "main".to_string(),
            root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
            head_sha: None,
        };
        let layout = SpoolLayout {
            root: PathBuf::from(r"C:\Users\floris\.xbp\mutations\xylex-group\xbp\main"),
            events_file: PathBuf::from("events.jsonl"),
            commits_file: PathBuf::from("commits.jsonl"),
            stats_file: PathBuf::from("stats.json"),
            watcher_state_file: PathBuf::from("watcher-state.json"),
            sync_log_file: PathBuf::from("sync-log.jsonl"),
            event_dedupe_file: PathBuf::from("event-dedupe.jsonl"),
            event_dedupe_dir: PathBuf::from("event-dedupe"),
            commit_dedupe_dir: PathBuf::from("commit-dedupe"),
        };
        let candidates = vec![SyncCandidate {
            path: PathBuf::from("events.jsonl"),
            records: SyncRecords::Events(vec![
                stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
                stats_test_event_record("2026-07-08T10:05:00Z", "src/main.rs", 2, 0),
                stats_test_event_record("2026-07-08T11:00:00Z", "README.md", 1, 3),
            ]),
        }];

        let summary = build_stats_summary(&identity, &layout, &candidates, 15).unwrap();

        assert_eq!(summary.total_events, 3);
        assert_eq!(summary.estimated_coding_seconds, 60 + 300 + 900);
        assert_eq!(summary.session_count, 2);
        assert_eq!(summary.observed_span_seconds, 60 * 60);
        assert_eq!(summary.added_lines, 7);
        assert_eq!(summary.removed_lines, 4);
        assert_eq!(summary.by_file[0].path, "README.md");
        assert_eq!(summary.by_file[0].estimated_coding_seconds, 900);
        assert_eq!(summary.by_filetype[0].name, "md");
        assert_eq!(summary.by_filetype[0].estimated_coding_seconds, 900);
        assert_eq!(summary.by_filetype[1].name, "rs");
        assert_eq!(summary.by_filetype[1].estimated_coding_seconds, 360);
    }

    #[test]
    fn builds_repo_activity_across_branch_spools() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        let main_root = temp_root.join("main");
        let feature_root = temp_root.join("feature-login");
        fs::create_dir_all(&main_root).unwrap();
        fs::create_dir_all(&feature_root).unwrap();

        append_json_line(
            &main_root.join("events-main.jsonl"),
            &stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
        )
        .unwrap();
        append_json_line(
            &main_root.join("events-main.jsonl"),
            &stats_test_event_record("2026-07-08T10:10:00Z", "src/lib.rs", 2, 0),
        )
        .unwrap();
        append_json_line(
            &feature_root.join("events-feature.jsonl"),
            &stats_test_event_record("2026-07-08T11:00:00Z", "src/auth.rs", 10, 2),
        )
        .unwrap();
        append_json_line(
            &feature_root.join("events-feature.jsonl"),
            &stats_test_event_record("2026-07-08T11:30:00Z", "src/auth.rs", 3, 1),
        )
        .unwrap();
        append_json_line(
            &feature_root.join("commits-feature.jsonl"),
            &worktree_commit_test_event("previous", "current"),
        )
        .unwrap();

        let identity = RepoIdentity {
            owner: "xylex-group".to_string(),
            name: "xbp".to_string(),
            branch: "main".to_string(),
            root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
            head_sha: None,
        };

        let summary = build_repo_activity_summary(&identity, &temp_root, 15).unwrap();

        assert_eq!(summary.branch_count, 2);
        assert_eq!(summary.total_events, 4);
        assert_eq!(summary.total_commits, 1);
        assert_eq!(summary.session_count, 3);
        assert_eq!(summary.estimated_coding_seconds, 60 + 600 + 60 + 900);
        assert_eq!(summary.observed_span_seconds, 90 * 60);
        assert_eq!(summary.branches[0].branch_name, "feature-login");
        assert_eq!(summary.branches[0].session_count, 2);
        assert_eq!(summary.branches[0].observed_span_seconds, 30 * 60);
        assert_eq!(summary.branches[1].branch_name, "main");

        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn mutation_fingerprint_ignores_random_event_id_but_keeps_time_bucket() {
        let first: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:00Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();
        let same_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:00Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();
        let next_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:06Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();

        assert_eq!(
            mutation_event_fingerprint(&first),
            mutation_event_fingerprint(&same_bucket)
        );
        assert_ne!(
            mutation_event_fingerprint(&first),
            mutation_event_fingerprint(&next_bucket)
        );

        // Fingerprint is path-root independent: absolute Windows/WSL spellings must not diverge.
        let mut windows_root = first.clone();
        windows_root.repo_root = r"C:\Users\floris\Documents\GitHub\xbp".to_string();
        let mut wsl_root = first.clone();
        wsl_root.repo_root = "/mnt/c/Users/floris/Documents/GitHub/xbp".to_string();
        assert_eq!(
            mutation_event_fingerprint(&windows_root),
            mutation_event_fingerprint(&wsl_root)
        );

        let fingerprint = mutation_event_fingerprint(&first);
        assert_eq!(stable_event_id(&fingerprint), format!("wtmut_{fingerprint}"));
        assert_eq!(
            ensure_mutation_event_identity(first.clone()).id,
            stable_event_id(&fingerprint)
        );
    }

    #[test]
    fn runtime_key_and_device_id_are_stable_and_scoped() {
        let key = worktree_runtime_key();
        assert!(!key.is_empty());
        assert_eq!(key, worktree_runtime_key());
        // On this host: windows | linux | macos | wsl-*
        assert!(
            key == "windows"
                || key == "linux"
                || key == "macos"
                || key.starts_with("wsl-")
                || key == "wsl",
            "unexpected runtime key {key}"
        );
        let device = worktree_device_hardware_id("xbp_hw_test");
        assert_eq!(device, format!("xbp_hw_test@{key}"));
    }

    #[test]
    fn should_poll_only_for_wsl_mnt_paths() {
        // On non-WSL CI/dev hosts this is always false; the path heuristic itself is cheap.
        let mnt = Path::new("/mnt/c/Users/floris/Documents/GitHub/xbp");
        let home = Path::new("/home/floris/src/xbp");
        if is_wsl_runtime() {
            assert!(should_use_poll_watcher(mnt));
            assert!(!should_use_poll_watcher(home));
        } else {
            assert!(!should_use_poll_watcher(mnt));
            assert!(!should_use_poll_watcher(home));
        }
    }

    #[test]
    fn repository_spool_root_uses_global_xbp_home_mutations() {
        let identity = RepoIdentity {
            owner: "xylex-group".to_string(),
            name: "xbp".to_string(),
            branch: "main".to_string(),
            root: PathBuf::from("/tmp/xbp"),
            head_sha: None,
        };
        let root = repository_spool_root(&identity).expect("global xbp home");
        let rendered = root.to_string_lossy().replace('\\', "/");
        assert!(
            rendered.contains("/mutations/"),
            "expected home-level mutations path, got {rendered}"
        );
        assert!(
            !rendered.contains("/mutations/runtimes/"),
            "spool must not be runtime-split when sharing one established home: {rendered}"
        );
        assert!(
            rendered.ends_with("xylex-group/xbp") || rendered.contains("xylex-group/xbp"),
            "expected owner/name suffix, got {rendered}"
        );
        // Must live under the established global root, never the repo path.
        assert!(!rendered.contains("/tmp/xbp/"));
    }

    #[test]
    fn cross_platform_path_identity_reconciles_wsl_and_windows() {
        let windows = Path::new(r"C:\Users\szymon\.xbp\mutations\SuitsFinance\speedrun-formations\main");
        let wsl = Path::new(
            "/mnt/c/Users/szymon/.xbp/mutations/SuitsFinance/speedrun-formations/main",
        );
        assert_eq!(
            cross_platform_path_identity_key(windows),
            cross_platform_path_identity_key(wsl)
        );
    }

    #[test]
    fn command_line_mentions_path_is_case_and_slash_insensitive() {
        let cmdline = r#""C:\xbp.exe" worktree-watch start --parent C:\Users\floris\Documents\GitHub"#;
        assert!(command_line_mentions_path(
            cmdline,
            "/mnt/c/users/floris/documents/github"
        ));
        assert!(command_line_mentions_path(
            cmdline,
            r"C:\Users\floris\Documents\GitHub"
        ));
        assert!(command_line_mentions_path(cmdline, "GitHub"));
    }

    #[test]
    fn append_unique_mutation_event_is_idempotent_across_dedupe_instances() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let layout = SpoolLayout {
            root: temp_root.clone(),
            events_file: temp_root.join("events.jsonl"),
            commits_file: temp_root.join("commits.jsonl"),
            stats_file: temp_root.join("stats.json"),
            watcher_state_file: temp_root.join("watcher-state.json"),
            sync_log_file: temp_root.join("sync-log.jsonl"),
            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
            event_dedupe_dir: temp_root.join("event-dedupe"),
            commit_dedupe_dir: temp_root.join("commit-dedupe"),
        };
        let mutation: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:00Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();

        let first =
            append_unique_mutation_event(&layout, &mutation, &mut RecentEventDedupe::default())
                .unwrap();
        let second =
            append_unique_mutation_event(&layout, &mutation, &mut RecentEventDedupe::default())
                .unwrap();

        assert!(first);
        assert!(!second);
        let events = read_jsonl_values(&layout.events_file).unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(
            events[0].get("fingerprint").and_then(Value::as_str),
            Some(mutation_event_fingerprint(&mutation).as_str())
        );
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn reads_spool_records_as_typed_events() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let events_file = temp_root.join("events.jsonl");
        append_json_line(
            &events_file,
            &stats_test_event_record("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
        )
        .unwrap();

        let records = read_spool_records(&events_file, SyncFileKind::Events).unwrap();

        match records {
            SyncRecords::Events(events) => {
                assert_eq!(events.len(), 1);
                assert_eq!(events[0].primary_path.as_deref(), Some("src/main.rs"));
            }
            SyncRecords::Commits(_) => panic!("expected typed event records"),
        }
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn read_spool_records_skips_nul_only_padding() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let events_file = temp_root.join("events.jsonl");
        fs::write(&events_file, vec![0; 4096]).unwrap();

        let records = read_spool_records(&events_file, SyncFileKind::Events).unwrap();

        match records {
            SyncRecords::Events(events) => assert!(events.is_empty()),
            SyncRecords::Commits(_) => panic!("expected event records"),
        }
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn read_jsonl_strips_leading_nuls_before_json() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let path = temp_root.join("event-dedupe.jsonl");
        let mut bytes = vec![0u8; 64];
        bytes.extend_from_slice(
            br#"{"fingerprint":"abc","firstSeenAt":"2026-07-09T00:00:00Z","eventKind":"modify","primaryPath":".xbp"}"#,
        );
        bytes.push(b'\n');
        fs::write(&path, bytes).unwrap();

        let values = read_jsonl_values(&path).unwrap();
        assert_eq!(values.len(), 1);
        assert_eq!(
            values[0].get("fingerprint").and_then(Value::as_str),
            Some("abc")
        );
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn read_jsonl_skips_garbage_lines() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let path = temp_root.join("mixed.jsonl");
        fs::write(
            &path,
            "not-json\n{\"fingerprint\":\"ok\"}\n\n{\"broken\":\n",
        )
        .unwrap();

        let values = read_jsonl_values(&path).unwrap();
        assert_eq!(values.len(), 1);
        assert_eq!(
            values[0].get("fingerprint").and_then(Value::as_str),
            Some("ok")
        );
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn collect_spool_candidates_rewrites_legacy_generated_events() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let events_file = temp_root.join("events-legacy.jsonl");
        append_json_line(
            &events_file,
            &stats_test_event_record(
                "2026-07-08T10:00:00Z",
                "apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas",
                0,
                0,
            ),
        )
        .unwrap();
        append_json_line(
            &events_file,
            &stats_test_event_record("2026-07-08T10:01:00Z", "apps/web/src/main.ts", 4, 1),
        )
        .unwrap();
        let layout = SpoolLayout {
            root: temp_root.clone(),
            events_file: temp_root.join("events.jsonl"),
            commits_file: temp_root.join("commits.jsonl"),
            stats_file: temp_root.join("stats.json"),
            watcher_state_file: temp_root.join("watcher-state.json"),
            sync_log_file: temp_root.join("sync-log.jsonl"),
            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
            event_dedupe_dir: temp_root.join("event-dedupe"),
            commit_dedupe_dir: temp_root.join("commit-dedupe"),
        };

        let candidates = collect_spool_candidates(&layout, false).unwrap();

        assert_eq!(candidates.len(), 1);
        match &candidates[0].records {
            SyncRecords::Events(events) => {
                assert_eq!(events.len(), 1);
                assert_eq!(
                    events[0].primary_path.as_deref(),
                    Some("apps/web/src/main.ts")
                );
            }
            SyncRecords::Commits(_) => panic!("expected event spool candidate"),
        }
        let rewritten = read_jsonl_records::<WorktreeMutationEvent>(&events_file).unwrap();
        assert_eq!(rewritten.len(), 1);
        assert_eq!(
            rewritten[0].primary_path.as_deref(),
            Some("apps/web/src/main.ts")
        );
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn collect_spool_candidates_deletes_fully_ignored_event_files() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let events_file = temp_root.join("events-legacy.jsonl");
        append_json_line(
            &events_file,
            &stats_test_event_record(
                "2026-07-08T10:00:00Z",
                "apps/docs/node_modules/@aws-sdk/client-lambda/dist-types/schemas",
                0,
                0,
            ),
        )
        .unwrap();
        let layout = SpoolLayout {
            root: temp_root.clone(),
            events_file: temp_root.join("events.jsonl"),
            commits_file: temp_root.join("commits.jsonl"),
            stats_file: temp_root.join("stats.json"),
            watcher_state_file: temp_root.join("watcher-state.json"),
            sync_log_file: temp_root.join("sync-log.jsonl"),
            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
            event_dedupe_dir: temp_root.join("event-dedupe"),
            commit_dedupe_dir: temp_root.join("commit-dedupe"),
        };

        let candidates = collect_spool_candidates(&layout, false).unwrap();

        assert!(candidates.is_empty());
        assert!(!events_file.exists());
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn append_unique_commit_event_is_idempotent_across_watcher_instances() {
        let temp_root = std::env::temp_dir().join(format!("xbp-watch-test-{}", Uuid::new_v4()));
        fs::create_dir_all(&temp_root).unwrap();
        let layout = SpoolLayout {
            root: temp_root.clone(),
            events_file: temp_root.join("events.jsonl"),
            commits_file: temp_root.join("commits.jsonl"),
            stats_file: temp_root.join("stats.json"),
            watcher_state_file: temp_root.join("watcher-state.json"),
            sync_log_file: temp_root.join("sync-log.jsonl"),
            event_dedupe_file: temp_root.join("event-dedupe.jsonl"),
            event_dedupe_dir: temp_root.join("event-dedupe"),
            commit_dedupe_dir: temp_root.join("commit-dedupe"),
        };
        let commit = WorktreeCommitEvent {
            id: Uuid::new_v4().to_string(),
            fingerprint: None,
            repo_owner: "xylex-group".to_string(),
            repo_name: "xbp".to_string(),
            branch_name: "main".to_string(),
            repo_root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
            previous_head_sha: Some("previous".to_string()),
            head_sha: "current".to_string(),
            subject: Some("test commit".to_string()),
            author_name: Some("Floris".to_string()),
            author_email: Some("floris@xylex.group".to_string()),
            committed_at: Some("2026-07-08T10:00:00Z".to_string()),
            occurred_at: Utc::now(),
            files_changed: None,
            files_changed_count: None,
            insertions: None,
            deletions: None,
            is_merge: None,
            included_commits: None,
        };

        assert!(append_unique_commit_event(&layout, &commit).unwrap());
        assert!(!append_unique_commit_event(&layout, &commit).unwrap());
        let commits = read_jsonl_values(&layout.commits_file).unwrap();
        assert_eq!(commits.len(), 1);
        assert_eq!(
            commits[0].get("fingerprint").and_then(Value::as_str),
            Some(commit_event_fingerprint(&commit).as_str())
        );
        let _ = fs::remove_dir_all(temp_root);
    }

    #[test]
    fn splits_worktree_mutation_uploads_into_bounded_batches() {
        let events = (0..(WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT + 1))
            .map(|index| {
                serde_json::from_value::<WorktreeMutationEvent>(stats_test_event(
                    "2026-07-08T10:00:00Z",
                    &format!("src/file-{index}.rs"),
                    1,
                    0,
                ))
                .unwrap()
            })
            .collect::<Vec<_>>();
        let commits = vec![
            worktree_commit_test_event("previous-a", "current-a"),
            worktree_commit_test_event("previous-b", "current-b"),
        ];

        let batches = split_worktree_mutation_upload_batches(events, commits);

        assert_eq!(batches.len(), 2);
        assert_eq!(
            batches[0].events.len() + batches[0].commits.len(),
            WORKTREE_MUTATION_SYNC_BATCH_RECORD_LIMIT
        );
        assert_eq!(batches[1].events.len(), 1);
        assert_eq!(batches[1].commits.len(), 2);
    }

    #[test]
    fn splits_worktree_mutation_uploads_by_estimated_json_bytes() {
        let long_path = format!("src/{}.rs", "x".repeat(8_000));
        let events = (0..200)
            .map(|index| {
                serde_json::from_value::<WorktreeMutationEvent>(stats_test_event(
                    "2026-07-08T10:00:00Z",
                    &format!("{long_path}-{index}"),
                    1,
                    0,
                ))
                .unwrap()
            })
            .collect::<Vec<_>>();

        let batches = split_worktree_mutation_upload_batches(events, Vec::new());

        assert!(batches.len() > 1);
        for batch in batches {
            assert!(
                batch.estimated_json_bytes <= WORKTREE_MUTATION_SYNC_BATCH_JSON_BYTES,
                "batch estimated JSON bytes exceeded limit: {}",
                batch.estimated_json_bytes
            );
        }
    }

    #[test]
    fn classifies_create_remove_and_rename_events() {
        assert_eq!(
            classify_event_kind(&EventKind::Create(CreateKind::File)),
            "file_create"
        );
        assert_eq!(
            classify_event_kind(&EventKind::Remove(RemoveKind::Folder)),
            "folder_remove"
        );
        assert_eq!(
            classify_event_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both))),
            "rename_or_move"
        );
    }

    #[test]
    fn retries_only_retryable_worktree_sync_statuses() {
        assert!(should_retry_worktree_mutation_sync_status(
            StatusCode::INTERNAL_SERVER_ERROR
        ));
        assert!(should_retry_worktree_mutation_sync_status(
            StatusCode::REQUEST_TIMEOUT
        ));
        assert!(should_retry_worktree_mutation_sync_status(
            StatusCode::TOO_MANY_REQUESTS
        ));
        assert!(!should_retry_worktree_mutation_sync_status(
            StatusCode::BAD_REQUEST
        ));
        assert!(!should_retry_worktree_mutation_sync_status(
            StatusCode::UNAUTHORIZED
        ));
        assert!(!should_retry_worktree_mutation_sync_status(
            StatusCode::PAYLOAD_TOO_LARGE
        ));
    }

    #[test]
    fn match_github_repo_inventory_resolves_short_name_and_owner_repo() {
        let temp = std::env::temp_dir().join(format!("xbp-repo-lookup-{}", Uuid::new_v4()));
        let athena = temp.join("athena");
        let other = temp.join("other-athena");
        fs::create_dir_all(&athena).unwrap();
        fs::create_dir_all(&other).unwrap();

        let repos = vec![
            GithubRepoRecord {
                owner: "xylex-group".to_string(),
                repo: "athena".to_string(),
                full_name: "xylex-group/athena".to_string(),
                path: athena.to_string_lossy().to_string(),
                remote_url: "https://github.com/xylex-group/athena.git".to_string(),
            },
            GithubRepoRecord {
                owner: "someone".to_string(),
                repo: "other".to_string(),
                full_name: "someone/other".to_string(),
                path: other.to_string_lossy().to_string(),
                remote_url: "https://github.com/someone/other.git".to_string(),
            },
        ];

        let by_short = match_github_repo_inventory(&repos, "athena").unwrap().unwrap();
        assert_eq!(
            path_identity_key(&by_short),
            path_identity_key(athena.as_path())
        );

        let by_full = match_github_repo_inventory(&repos, "xylex-group/athena")
            .unwrap()
            .unwrap();
        assert_eq!(
            path_identity_key(&by_full),
            path_identity_key(athena.as_path())
        );

        assert!(match_github_repo_inventory(&repos, "missing")
            .unwrap()
            .is_none());

        let _ = fs::remove_dir_all(temp);
    }

    #[test]
    fn match_github_repo_inventory_errors_on_ambiguous_short_name() {
        let temp = std::env::temp_dir().join(format!("xbp-repo-lookup-ambig-{}", Uuid::new_v4()));
        let a = temp.join("a");
        let b = temp.join("b");
        fs::create_dir_all(&a).unwrap();
        fs::create_dir_all(&b).unwrap();

        let repos = vec![
            GithubRepoRecord {
                owner: "one".to_string(),
                repo: "dup".to_string(),
                full_name: "one/dup".to_string(),
                path: a.to_string_lossy().to_string(),
                remote_url: "https://github.com/one/dup.git".to_string(),
            },
            GithubRepoRecord {
                owner: "two".to_string(),
                repo: "dup".to_string(),
                full_name: "two/dup".to_string(),
                path: b.to_string_lossy().to_string(),
                remote_url: "https://github.com/two/dup.git".to_string(),
            },
        ];

        let err = match_github_repo_inventory(&repos, "dup").unwrap_err();
        assert!(err.contains("Ambiguous"));
        assert!(err.contains("one/dup") || err.contains(a.to_string_lossy().as_ref()));

        let unique = match_github_repo_inventory(&repos, "two/dup").unwrap().unwrap();
        assert_eq!(path_identity_key(&unique), path_identity_key(b.as_path()));

        let _ = fs::remove_dir_all(temp);
    }

    fn stats_test_event(
        occurred_at: &str,
        path: &str,
        added_lines: u64,
        removed_lines: u64,
    ) -> Value {
        json!({
            "id": Uuid::new_v4().to_string(),
            "repoOwner": "xylex-group",
            "repoName": "xbp",
            "branchName": "main",
            "repoRoot": r"C:\Users\floris\Documents\GitHub\xbp",
            "headSha": null,
            "eventKind": "file_modify",
            "paths": [path],
            "primaryPath": path,
            "oldPath": null,
            "newPath": null,
            "addedLines": added_lines,
            "removedLines": removed_lines,
            "totalLines": added_lines + 42,
            "fileCreated": false,
            "fileRemoved": false,
            "folderCreated": false,
            "folderRemoved": false,
            "renamedOrMoved": false,
            "rawKind": "Modify(Data(Content))",
            "occurredAt": occurred_at,
        })
    }

    fn stats_test_event_record(
        occurred_at: &str,
        path: &str,
        added_lines: u64,
        removed_lines: u64,
    ) -> WorktreeMutationEvent {
        serde_json::from_value(stats_test_event(
            occurred_at,
            path,
            added_lines,
            removed_lines,
        ))
        .unwrap()
    }

    fn worktree_commit_test_event(previous_head_sha: &str, head_sha: &str) -> WorktreeCommitEvent {
        WorktreeCommitEvent {
            id: Uuid::new_v4().to_string(),
            fingerprint: None,
            repo_owner: "xylex-group".to_string(),
            repo_name: "xbp".to_string(),
            branch_name: "main".to_string(),
            repo_root: r"C:\Users\floris\Documents\GitHub\xbp".to_string(),
            previous_head_sha: Some(previous_head_sha.to_string()),
            head_sha: head_sha.to_string(),
            subject: Some("test commit".to_string()),
            author_name: Some("Floris".to_string()),
            author_email: Some("floris@xylex.group".to_string()),
            committed_at: Some("2026-07-08T10:00:00Z".to_string()),
            occurred_at: Utc::now(),
            files_changed: None,
            files_changed_count: None,
            insertions: None,
            deletions: None,
            is_merge: None,
            included_commits: None,
        }
    }

    #[test]
    fn enrich_commit_range_parses_numstat_lines() {
        // Pure unit coverage of the numstat parser path via a synthetic range
        // is environment-dependent; ensure the enrichment struct defaults cleanly.
        let empty = CommitRangeEnrichment::default();
        assert!(empty.files_changed.is_none());
        assert!(empty.insertions.is_none());
    }
}