xbp_cli/commands/

worktree_watch.rs

use crate::commands::cli_session::{cli_request_client, resolve_cli_access_token};
use crate::config::{
    resolve_device_identity, resolve_worktree_watch_config, ApiConfig, WorktreeWatchConfig,
};
use chrono::{DateTime, Utc};
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};
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, OnceLock};
use std::time::{Duration, Instant};
use sysinfo::{Pid, System};
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;

/// Client-side ignore rules: built-in VCS/generated dirs + global config filters.
/// 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>,
}

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

    fn from_config(config: &WorktreeWatchConfig) -> 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<_>>();

        Self {
            forbidden_path_prefixes,
            forbidden_folders,
            banned_words,
        }
    }

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

        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 {
        let Ok(relative) = path.strip_prefix(root) else {
            return false;
        };
        self.ignores_relative(&relative.to_string_lossy())
    }

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

fn built_in_forbidden_folders() -> BTreeSet<String> {
    [
        ".git",
        ".hg",
        ".svn",
        ".next",
        ".turbo",
        ".vercel",
        "node_modules",
        "target",
        "dist",
        "build",
    ]
    .into_iter()
    .map(str::to_string)
    .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)
}

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

#[derive(Debug, Clone)]
pub struct WorktreeWatchTargetOptions {
    pub repo: Option<PathBuf>,
    pub parent: Option<PathBuf>,
}

#[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,
}

#[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,
}

#[derive(Debug)]
struct WatchedRepo {
    identity: RepoIdentity,
    layout: SpoolLayout,
    last_head: Option<String>,
    event_dedupe: RecentEventDedupe,
}

#[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>,
    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>,
}

#[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,
}

#[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, 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, 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, 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>,
}

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

#[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 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.detach {
            let path = spawn_detached_parent_worktree_watch(parent)?;
            println!(
                "Started one background worktree watcher for {} covering {} repo(s).",
                path.display(),
                identities.len()
            );
            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;
    }

    if options.detach {
        return spawn_detached_worktree_watch(options.target.repo.as_deref()).map(|path| {
            println!("Started background worktree watcher for {}", path.display());
        });
    }

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

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

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

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(())
}

pub async fn run_worktree_watch_status(options: WorktreeWatchStatusOptions) -> Result<(), String> {
    let identities = resolve_target_identities(&options.target)?;
    let mut payloads = Vec::new();
    let mut total_files = 0usize;
    let mut total_records = 0usize;

    for identity in identities {
        let status =
            worktree_watch_status_payload(&identity, options.records, options.record_limit)?;
        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
        };
        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.json {
        let payload = if options.target.parent.is_some() {
            json!({
                "repositories": payloads,
                "repositoryCount": payloads.len(),
                "unsyncedFiles": total_files,
                "unsyncedRecords": total_records,
            })
        } else {
            payloads.into_iter().next().unwrap_or_else(|| json!({}))
        };
        println!(
            "{}",
            serde_json::to_string_pretty(&payload)
                .map_err(|error| format!("Failed to render status JSON: {error}"))?
        );
    } else {
        for (index, payload) in payloads.iter().enumerate() {
            if index > 0 {
                println!();
            }
            println!(
                "repo: {}/{}",
                payload
                    .get("repositoryOwner")
                    .and_then(Value::as_str)
                    .unwrap_or("unknown"),
                payload
                    .get("repositoryName")
                    .and_then(Value::as_str)
                    .unwrap_or("repository")
            );
            println!(
                "branch: {}",
                payload
                    .get("branchName")
                    .and_then(Value::as_str)
                    .unwrap_or("unknown")
            );
            println!(
                "root: {}",
                payload
                    .get("repoRoot")
                    .and_then(Value::as_str)
                    .unwrap_or("")
            );
            println!(
                "spool: {}",
                payload
                    .get("spoolRoot")
                    .and_then(Value::as_str)
                    .unwrap_or("")
            );
            println!(
                "unsynced files: {}",
                payload
                    .get("unsyncedFiles")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            println!(
                "unsynced records: {}",
                payload
                    .get("unsyncedRecords")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            println!(
                "synced files: {}",
                payload
                    .get("syncedFiles")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            println!(
                "synced records: {}",
                payload
                    .get("syncedRecords")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            print_last_sync(payload);
            if options.records {
                print_status_records(payload);
            }
            if options.stats {
                print_status_stats(payload);
            }
            if options.repo_activity {
                print_repo_activity(payload);
            }
        }
        if options.target.parent.is_some() {
            println!();
            println!(
                "total: {} repo(s), {} unsynced file(s), {} unsynced record(s)",
                payloads.len(),
                total_files,
                total_records
            );
        }
    }

    Ok(())
}

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)
}

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 let Some(parent) = target.parent.as_deref() {
        return discover_repo_identities_under(parent);
    }

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

fn worktree_watch_status_payload(
    identity: &RepoIdentity,
    include_records: bool,
    record_limit: usize,
) -> Result<Value, String> {
    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": identity.root.display().to_string(),
        "repositoryOwner": identity.owner.clone(),
        "repositoryName": identity.name.clone(),
        "branchName": identity.branch.clone(),
        "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 include_records {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "records".to_string(),
                collect_status_records(&candidates, record_limit)?,
            );
        }
    }

    Ok(payload)
}

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(),
                    }));
                    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
        );
    }
}

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}");
    }
}

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> {
    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> {
    let (tx, rx) = mpsc::channel();
    let mut watcher = RecommendedWatcher::new(
        move |result| {
            let _ = tx.send(result);
        },
        Config::default(),
    )
    .map_err(|error| format!("Failed to create filesystem watcher: {error}"))?;

    watcher
        .watch(&identity.root, RecursiveMode::Recursive)
        .map_err(|error| {
            format!(
                "Failed to watch repository root {}: {error}",
                identity.root.display()
            )
        })?;

    let mut last_head = identity.head_sha.clone();
    let mut last_commit_check = Instant::now();
    let mut last_sync = Instant::now();
    let mut event_dedupe = RecentEventDedupe {
        fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
    };

    loop {
        match rx.recv_timeout(Duration::from_millis(750)) {
            Ok(Ok(event)) => {
                if let Some(mutation) = mutation_event_from_notify(&identity, event) {
                    append_unique_mutation_event(&layout, &mutation, &mut event_dedupe)?;
                }
            }
            Ok(Err(error)) => {
                eprintln!("worktree watcher error: {error}");
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                return Err("Filesystem watcher disconnected.".to_string());
            }
        }

        if last_commit_check.elapsed() >= Duration::from_secs(3) {
            last_head = record_commit_snapshot(&identity, &layout, last_head.as_deref())?;
            last_commit_check = Instant::now();
        }

        if sync_interval_seconds > 0
            && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
        {
            if let Err(error) = sync_spool_for_identity(&identity, &layout).await {
                eprintln!("worktree mutation sync failed: {error}");
            }
            last_sync = Instant::now();
        }
    }
}

async fn watch_parent_foreground(
    parent_root: PathBuf,
    identities: Vec<RepoIdentity>,
    sync_interval_seconds: u64,
) -> Result<(), String> {
    let (tx, rx) = mpsc::channel();
    let mut watcher = RecommendedWatcher::new(
        move |result| {
            let _ = tx.send(result);
        },
        Config::default(),
    )
    .map_err(|error| format!("Failed to create parent filesystem watcher: {error}"))?;

    watcher
        .watch(&parent_root, RecursiveMode::Recursive)
        .map_err(|error| {
            format!(
                "Failed to watch parent folder {}: {error}",
                parent_root.display()
            )
        })?;

    let mut repos = prepare_watched_repos(identities)?;
    let mut last_commit_check = Instant::now();
    let mut last_sync = Instant::now();

    loop {
        match rx.recv_timeout(Duration::from_millis(750)) {
            Ok(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)
                    {
                        append_unique_mutation_event(
                            &repo.layout,
                            &mutation,
                            &mut repo.event_dedupe,
                        )?;
                    }
                }
            }
            Ok(Err(error)) => {
                eprintln!("parent worktree watcher error: {error}");
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                return Err("Parent filesystem watcher disconnected.".to_string());
            }
        }

        if last_commit_check.elapsed() >= Duration::from_secs(3) {
            for repo in &mut repos {
                repo.last_head = record_commit_snapshot(
                    &repo.identity,
                    &repo.layout,
                    repo.last_head.as_deref(),
                )?;
            }
            last_commit_check = Instant::now();
        }

        if sync_interval_seconds > 0
            && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
        {
            for repo in &repos {
                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();
        }
    }
}

fn prepare_watched_repos(identities: Vec<RepoIdentity>) -> Result<Vec<WatchedRepo>, String> {
    let mut repos = Vec::with_capacity(identities.len());
    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        let event_dedupe = RecentEventDedupe {
            fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
        };
        repos.push(WatchedRepo {
            last_head: identity.head_sha.clone(),
            identity,
            layout,
            event_dedupe,
        });
    }
    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);
    let line_counts = primary_path
        .as_deref()
        .and_then(|path| git_numstat_for_path(&identity.root, path).ok());
    let total_lines = primary_path
        .as_deref()
        .and_then(|path| file_line_count_for_path(&identity.root, path).ok());
    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);

    Some(WorktreeMutationEvent {
        id: Uuid::new_v4().to_string(),
        fingerprint: None,
        repo_owner: identity.owner.clone(),
        repo_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        head_sha: current_head(&identity.root).ok().flatten(),
        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,
        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: Utc::now(),
    })
}

fn append_unique_mutation_event(
    layout: &SpoolLayout,
    mutation: &WorktreeMutationEvent,
    dedupe: &mut RecentEventDedupe,
) -> Result<bool, String> {
    let fingerprint = mutation_event_fingerprint(mutation);
    if dedupe.fingerprints.contains(&fingerprint)
        || event_dedupe_file_contains(&layout.event_dedupe_file, &fingerprint)?
    {
        dedupe.fingerprints.insert(fingerprint);
        return Ok(false);
    }

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

    let mut mutation = mutation.clone();
    mutation.fingerprint = Some(fingerprint.clone());
    if let Err(error) = append_json_line(&layout.events_file, &mutation) {
        release_event_fingerprint_claim(layout, &fingerprint);
        return Err(error);
    }
    append_json_line(
        &layout.event_dedupe_file,
        &WorktreeWatchEventDedupeRecord {
            fingerprint: fingerprint.clone(),
            first_seen_at: Utc::now(),
            event_kind: mutation.event_kind.clone(),
            primary_path: mutation.primary_path.clone(),
        },
    )?;
    dedupe.fingerprints.insert(fingerprint);
    Ok(true)
}

fn claim_event_fingerprint(layout: &SpoolLayout, fingerprint: &str) -> Result<bool, String> {
    fs::create_dir_all(&layout.event_dedupe_dir).map_err(|error| {
        format!(
            "Failed to create event dedupe directory {}: {error}",
            layout.event_dedupe_dir.display()
        )
    })?;
    let marker = event_dedupe_marker_path(layout, fingerprint);
    match OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&marker)
    {
        Ok(mut file) => {
            writeln!(file, "{}", Utc::now().to_rfc3339()).map_err(|error| {
                format!(
                    "Failed to write event dedupe marker {}: {error}",
                    marker.display()
                )
            })?;
            Ok(true)
        }
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
        Err(error) => Err(format!(
            "Failed to create event dedupe marker {}: {error}",
            marker.display()
        )),
    }
}

fn release_event_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
    let _ = fs::remove_file(event_dedupe_marker_path(layout, fingerprint));
}

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;
    let payload = json!({
        "repoOwner": mutation.repo_owner,
        "repoName": mutation.repo_name,
        "branchName": mutation.branch_name,
        "repoRoot": mutation.repo_root,
        "headSha": mutation.head_sha,
        "eventKind": mutation.event_kind,
        "paths": mutation.paths,
        "primaryPath": mutation.primary_path,
        "oldPath": mutation.old_path,
        "newPath": mutation.new_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 load_event_dedupe_fingerprints(path: &Path) -> Result<BTreeSet<String>, String> {
    let mut fingerprints = BTreeSet::new();
    if !path.exists() {
        return Ok(fingerprints);
    }

    for value in read_jsonl_values(path)? {
        if let Some(fingerprint) = value.get("fingerprint").and_then(Value::as_str) {
            fingerprints.insert(fingerprint.to_string());
        }
    }
    Ok(fingerprints)
}

fn event_dedupe_file_contains(path: &Path, fingerprint: &str) -> Result<bool, String> {
    if !path.exists() {
        return Ok(false);
    }
    for value in read_jsonl_values(path)? {
        if value
            .get("fingerprint")
            .and_then(Value::as_str)
            .map(|value| value == fingerprint)
            .unwrap_or(false)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

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));
    }

    let commit = WorktreeCommitEvent {
        id: Uuid::new_v4().to_string(),
        fingerprint: None,
        repo_owner: identity.owner.clone(),
        repo_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        previous_head_sha: previous_head.map(str::to_string),
        head_sha: head_sha.clone(),
        subject: git_output(&identity.root, &["log", "-1", "--pretty=%s"]).ok(),
        author_name: git_output(&identity.root, &["log", "-1", "--pretty=%an"]).ok(),
        author_email: git_output(&identity.root, &["log", "-1", "--pretty=%ae"]).ok(),
        committed_at: git_output(&identity.root, &["log", "-1", "--pretty=%cI"]).ok(),
        occurred_at: Utc::now(),
    };
    append_unique_commit_event(layout, &commit)?;
    Ok(Some(head_sha))
}

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

    let mut commit = commit.clone();
    commit.fingerprint = Some(fingerprint.clone());
    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!({
        "repoOwner": commit.repo_owner,
        "repoName": commit.repo_name,
        "branchName": commit.branch_name,
        "repoRoot": commit.repo_root,
        "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> {
    fs::create_dir_all(&layout.commit_dedupe_dir).map_err(|error| {
        format!(
            "Failed to create commit dedupe directory {}: {error}",
            layout.commit_dedupe_dir.display()
        )
    })?;
    let marker = commit_dedupe_marker_path(layout, fingerprint);
    match OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&marker)
    {
        Ok(mut file) => {
            writeln!(file, "{}", Utc::now().to_rfc3339()).map_err(|error| {
                format!(
                    "Failed to write commit dedupe marker {}: {error}",
                    marker.display()
                )
            })?;
            Ok(true)
        }
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
        Err(error) => Err(format!(
            "Failed to create commit dedupe marker {}: {error}",
            marker.display()
        )),
    }
}

fn release_commit_fingerprint_claim(layout: &SpoolLayout, fingerprint: &str) {
    let _ = fs::remove_file(commit_dedupe_marker_path(layout, fingerprint));
}

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 = device.hardware_id;
    let hostname = current_hostname();
    let platform = std::env::consts::OS.to_string();
    let repo_root = identity.root.display().to_string();
    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
        );
    }

    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,
            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,
    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(),
            },
            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> {
    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> {
    let mut candidates = Vec::new();
    if !layout.root.exists() {
        return Ok(candidates);
    }

    for entry in fs::read_dir(&layout.root).map_err(|error| {
        format!(
            "Failed to read spool directory {}: {error}",
            layout.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 records = sanitize_spool_records(&path, kind, read_spool_records(&path, kind)?)?;
        if records.is_empty() {
            continue;
        }
        candidates.push(SyncCandidate { path, records });
    }

    Ok(candidates)
}

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> {
    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())),
    }
}

fn read_jsonl_records<T>(path: &Path) -> Result<Vec<T>, String>
where
    T: for<'de> Deserialize<'de>,
{
    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 in reader.lines() {
        let line =
            line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
        if line.trim().is_empty() || line.trim_matches('\0').trim().is_empty() {
            continue;
        }
        records.push(serde_json::from_str(&line).map_err(|error| {
            format!(
                "Failed to parse JSONL record in {}: {error}",
                path.display()
            )
        })?);
    }
    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());

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

    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());

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

    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 replace_existing_parent_watcher(
    parent: &Path,
    layout: &ParentSpoolLayout,
) -> Result<(), String> {
    let Some(state) = read_parent_watcher_state(&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()
        );
        let _ = fs::remove_file(&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()
        ));
    }

    let _ = fs::remove_file(&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()
        );
        let _ = fs::remove_file(&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()
        ));
    }

    let _ = fs::remove_file(&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 {
        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)?;
        let _ = fs::remove_file(&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
        ));
    }

    let _ = fs::remove_file(&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(&layout.watcher_state_file)? else {
        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)?;
        let _ = fs::remove_file(&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
        ));
    }

    let _ = fs::remove_file(&layout.watcher_state_file);
    Ok(StopWatcherOutcome::RemovedStaleState)
}

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

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()
        )
    })
}

fn write_watcher_state(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    pid: u32,
    executable: &Path,
) -> Result<(), String> {
    let state = WorktreeWatcherState {
        pid,
        repo_root: identity.root.display().to_string(),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        executable: executable.display().to_string(),
        started_at: Utc::now(),
    };
    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: parent.display().to_string(),
        executable: executable.display().to_string(),
        started_at: Utc::now(),
    };
    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 {
    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 is_matching_watcher_process(state: &WorktreeWatcherState) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let system = System::new_all();
    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(" ");
    command_line.contains("worktree-watch")
        && command_line.contains("start")
        && command_line.contains(&state.repo_root)
        || watcher_state_process_identity_matches(process, state)
}

fn is_matching_parent_watcher_process(state: &ParentWorktreeWatcherState) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let system = System::new_all();
    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(" ");
    (command_line.contains("worktree-watch")
        && command_line.contains("start")
        && command_line.contains("--parent")
        && command_line.contains(&state.parent_root))
        || parent_watcher_state_process_identity_matches(process, state)
}

fn watcher_state_pid_exists_with_same_executable(state: &WorktreeWatcherState) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let system = System::new_all();
    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 = System::new_all();
    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_state(process, state)
}

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

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_state(
    process: &sysinfo::Process,
    state: &WorktreeWatcherState,
) -> bool {
    let process_started = process.start_time() as i64;
    let state_started = state.started_at.timestamp();
    (process_started - state_started).abs() <= 10
}

fn parent_process_start_time_matches_state(
    process: &sysinfo::Process,
    state: &ParentWorktreeWatcherState,
) -> bool {
    let process_started = process.start_time() as i64;
    let state_started = state.started_at.timestamp();
    (process_started - state_started).abs() <= 10
}

fn stop_process(state: &WorktreeWatcherState, force: bool) -> Result<(), String> {
    let system = System::new_all();
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return Ok(());
    };
    if !force && !is_matching_watcher_process(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 = System::new_all();
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return Ok(());
    };
    if !force && !is_matching_parent_watcher_process(state) {
        return Ok(());
    }
    if process.kill() {
        Ok(())
    } else {
        Err(format!(
            "Failed to stop existing parent watcher process {}",
            state.pid
        ))
    }
}

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

    let mut identities = Vec::new();
    let mut seen = BTreeSet::new();
    discover_repo_identities_in_dir(&parent, &parent, &mut seen, &mut identities)?;
    identities.sort_by(|left, right| left.root.cmp(&right.root));
    Ok(identities)
}

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

    if dir.join(".git").exists() {
        if let Ok(identity) = resolve_repo_identity(Some(dir)) {
            let key = path_identity_key(&identity.root);
            if seen.insert(key) {
                identities.push(identity);
            }
            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() {
            discover_repo_identities_in_dir(parent, &entry.path(), seen, identities)?;
        }
    }

    Ok(())
}

fn resolve_repo_identity(repo: Option<&Path>) -> Result<RepoIdentity, String> {
    let start = match repo {
        Some(path) => path.to_path_buf(),
        None => std::env::current_dir()
            .map_err(|error| format!("Failed to resolve current directory: {error}"))?,
    };
    let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"])?;
    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,
    })
}

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()),
    ))
}

fn repository_spool_root(identity: &RepoIdentity) -> Result<PathBuf, String> {
    let home = dirs::home_dir().ok_or_else(|| "Failed to resolve home directory.".to_string())?;
    Ok(home
        .join(".xbp")
        .join("mutations")
        .join(sanitize_path_component(&identity.owner))
        .join(sanitize_path_component(&identity.name)))
}

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");
    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: root.join("watcher-state.json"),
        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 home = dirs::home_dir().ok_or_else(|| "Failed to resolve home directory.".to_string())?;
    let parent = canonical_parent_path(parent)?;
    let parent_key = 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 = home
        .join(".xbp")
        .join("mutations")
        .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("watcher-state.json"),
    })
}

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)
}

fn append_json_line<T: Serialize>(path: &Path, value: &T) -> Result<(), String> {
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
    let line = serde_json::to_string(value)
        .map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
    writeln!(file, "{line}")
        .map_err(|error| format!("Failed to append spool file {}: {error}", path.display()))
}

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 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 {
    let value = path.to_string_lossy().to_string();
    if cfg!(windows) {
        value.to_ascii_lowercase()
    } else {
        value
    }
}

fn path_string_has_ignored_component(path: &str) -> bool {
    watch_ignore_rules().ignores_relative(path)
}

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

fn relative_slash_path(root: &Path, path: &Path) -> Option<String> {
    let relative = path.strip_prefix(root).ok().unwrap_or(path);
    Some(relative.to_string_lossy().replace('\\', "/"))
}

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

fn current_hostname() -> Option<String> {
    std::env::var("COMPUTERNAME")
        .or_else(|_| std::env::var("HOSTNAME"))
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

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::*;

    #[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 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(".next")));
        assert!(!is_skipped_discovery_dir(Path::new("mollie-api-rust")));
    }

    #[test]
    fn ignores_nested_generated_mutation_paths() {
        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")
        ));
        assert!(!is_ignored_path(
            root,
            Path::new(r"C:\Users\floris\Documents\GitHub\athena\apps\web\src\main.ts")
        ));
    }

    #[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()],
        });

        // 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 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)
        );
    }

    #[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 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(),
        };

        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
        ));
    }

    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(),
        }
    }
}